# CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

## Project

SIENTIA DataOps Orchestrator — a Temporal.io workflow application for pipeline orchestration, real-time alerting, and scheduled reporting on the SIENTIA platform. Python 3.11+, package name `orchestrator`.

## Commands

Install deps:
```bash
pip install -r requirements.txt
pip install -r requirements-dev.txt   # ruff, mypy, bandit, pytest, pytest-cov, pytest-asyncio
```

Run the worker locally (activates venv, loads `.env`, runs `python -m orchestrator.worker.worker`):
```bash
./run_local.sh
```

Tests:
```bash
pytest tests/ --cov=orchestrator --cov-report=html   # full suite with coverage
pytest tests/orchestrator/activities/test_mongo_db.py  # single file
pytest tests/orchestrator/activities/test_mongo_db.py::TestClass::test_name  # single test
./run_coverage.sh   # runs pytest w/ coverage, opens htmlcov/index.html
```

Lint/format/type/security (run individually — `validate.sh` referenced by README no longer exists in the repo, so don't rely on it):
```bash
ruff format --check orchestrator/ tests/
ruff check orchestrator/ tests/
mypy orchestrator/
bandit -r orchestrator/ -ll

# autofix
ruff format orchestrator/ tests/
ruff check --fix orchestrator/ tests/
```

CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable workflow `Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml` on PRs to `main` — it's the source of truth for what must pass, not a local script.

## Architecture

### Layering
- **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`).
- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` vs all levels).
- **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections.
- **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing.

### Temporal-specific conventions (all workflow/activity files follow this)
- Every workflow and activity-consuming module wraps non-deterministic/side-effecting imports in `with workflow.unsafe.imports_passed_through(): ...` right after `from temporalio import workflow`, because Temporal's sandbox restricts what workflow code can import directly. When adding a new workflow file, follow the existing pattern (see `orchestrator/workflows/orchestrator.py`, `orchestrator/activities/activities.py`) rather than importing at module top level.
- Workflows are registered with `@workflow.defn(name='...')`; the `run` method is `@workflow.run`. Activity calls from within a workflow go through `workflow.start_local_activity_method` / `workflow.execute_activity_method` referencing `Activities.<method>` (the class method, not an instance) with a `retry_policy` from `sientia_do.temporal.policies`.
- Task queue isolation is deliberate: don't move activities between the orchestrator/alerts/reports worker activity lists without checking which queue actually needs them — each list in `worker.py` is scoped to what that workflow (+ its subworkflows) calls.
- Namespaces: Temporal schedules span two namespaces, `scouter` and `laborious` (`TEMPORAL_SCOUTER_NAMESPACE` / `TEMPORAL_LABORIOUS_NAMESPACE`), handled inside `TemporalManager`.

### External dependency
`sientia-dataops-library` (pinned via `requirements.txt` as a git dependency, currently `@1.8.2`) supplies `sientia_do.observability.logger`, `sientia_do.observability.metrics_controller`, `sientia_do.notifications.handlers`, `sientia_do.temporal.policies`, and `sientia_do.temporal.activities.postgres.Postgres`. It is a separate Aignosi repo — its version bump is a deliberate, explicit change, not incidental.

### Tests
`tests/` mirrors `orchestrator/` (`tests/orchestrator/activities/`, `tests/orchestrator/utils/`, `tests/orchestrator/workflows/...`). Async tests use `pytest-asyncio`; coverage targets the `orchestrator` package. When testing Temporal workflows, follow the existing pattern in `tests/orchestrator/workflows/test_orchestrator.py` / `test_alerts.py` / `test_reports.py` (using Temporal's test env / activity mocking) rather than hitting real Temporal/Mongo/Redis/Postgres.

### Ruff config notes (`pyproject.toml`)
Single quotes enforced by formatter. Notably ignored lint rules with reasons already documented inline: `B023` (closures over loop vars needed for schedule-action assembly), `BLE001` (blind `except` needed so activities can always notify on any error), `N802`/`N806` (Temporal decorators/variables don't follow standard naming), `S105`/`S106` (false-positive hardcoded-password matches).