model-platform/DEVELOP.md

# DEVELOP.md — Developer Guide

This guide is for engineers working on the model platform codebase. It is the
authoritative source for architecture, code layout, configuration, conventions,
and common tasks. `ARCHITECTURE.md` now only records the simplification
history; `HANDOVER.md` covers recent commits and pending work.

## Architecture

The platform is a self-hosted Jupyter model development environment: interactive
workspaces, DAG scheduling, object-store artifacts, and per-workspace runtimes
— all exposed through a single Nginx gateway.

### Capability map

| Capability | Where it lives |
|---|---|
| Workspace notebook editing, row-level lock | `backend/api/jupyter.py` + `scripts.is_locked` |
| Jupyter auth routing (browser never holds runtime token) | `nginx/default.conf` + `auth_request` + `backend/api/jupyter.py` |
| Object storage for notebook / script / version / run_log (s3 / local) | `common/storage/` + `backend/services/storage.py` |
| DAG scheduling: nodes, edges, cron, manual trigger, retry, snapshot | `backend/api/schedules/` + `backend/api/schedules/runs.py` + `schedule/` (5 layers) |
| DAG execution via MySQL Outbox (no Redis, no in-process queue) | `schedule/application/orchestrator.py` + `schedule/execution/worker.py` |
| Per-workspace Jupyter subprocess pool, asyncio lock | `runtime/process.py` |
| Runtime rclone FUSE mount of workspace bucket (s3 mode only) | `runtime/mount.py` |
| 18 MySQL tables, soft delete, zero FK, async SQLAlchemy 2.0 | `common/db/models/` |

### Component diagram

```
                        ┌────────────────────┐
                        │  Browser (SPA)     │
                        └─────────┬──────────┘
                                  │ HTTPS / WS
                        ┌─────────▼──────────┐
                        │  Nginx (only :80) │  ← templates/default.conf
                        │   /api/  /jupyter/ /storage/
                        └────┬───────┬──────┘
                             │       │
              ┌──────────────┘       └─────────────┐
              ▼                                     ▼
   ┌──────────────────┐               ┌──────────────────────┐
   │ FastAPI Backend  │               │ Runtime (Jupyter)     │
   │ + /internal/v1   │   control     │ - rclone FUSE mount  │
   │   /objects       │   token-Auth  │                       │
   │ (storage)        ├──────────────►│ - subprocess pool    │
   │ - DAG CRUD       │               │   (per workspace)    │
   │ - script CRUD    │               └──────────┬───────────┘
   │ - auth_request   │                          │ FUSE / shared vol
   │ - /api/v1/...    │                          ▼
   └────┬──────┬──────┘               ┌──────────────────────┐
        │      │                       │ Object storage      │
        │      └──────── HTTP ───────►│ (s3: S3 service /    │
        ▼                              │  local: shared vol) │
   ┌────────────┐                      │ 4 buckets per usage │
   │  MySQL     │◄───────── poll ─────│                      │
   │ - 18 tbls  │                      └──────────────────────┘
   │ - outbox   │
   │ - jobstore │
   └────┬───────┘
        ▲
        │ outbox poll
   ┌────┴──────────────────────────┐
   │ Schedule Executor             │
   │ - CronScheduler (APScheduler) │
   │ - DispatchOrchestrator        │
   │ - NodeExecutor (worker)       │
   │ - SchedulerService (facade)   │
   └───────────────────────────────┘
```

### Services (docker-compose)

The architecture intentionally exposes only one host port (the gateway); all
other services are on the Docker internal network.

| Service | Image | Exposed | Purpose |
|---|---|---|---|
| `web` | `nginx:alpine` | host `:8888` → `:80` | SPA, `/api/` reverse proxy, `/jupyter/{ws}/` `auth_request` proxy, `/storage/` S3 passthrough (s3 mode only) |
| `backend` | `Dockerfile` | internal only | DAG CRUD, script CRUD, schedule trigger, `/api/v1/auth/jupyter`, `/internal/v1/objects` inter-service RPC (shared `INTERNAL_SERVICE_TOKEN`, see `§Auth`) |
| `runtime` | `Dockerfile` | internal only | Per-workspace Jupyter subprocess pool, rclone FUSE mount of `workspace` bucket (s3 mode) |
| `schedule` | `Dockerfile` | internal only | Cron tick + DAG execution (polls MySQL Outbox) |

The pre-2026 host-port mappings for backend/runtime (`8891:8000` / `8892:8000`)
were removed; no service is reachable from the host except Nginx anymore.
Backend ↔ schedule now talk via the `X-Internal-Service-Token` header on
`/internal/v1/*`. See `API.md §9`.

### Storage layout

The object store is selected at deploy time by `settings.storage_backend`
(`"s3"` default, `"local"` for dev / single-node / air-gapped). Either way
there are 4 purpose-named buckets, resolved in a single place
(`common/storage/factory.py:actual_bucket_name` + `USAGE_TYPE_TO_PURPOSE`).

| `usage_type` | Bucket | Env var | Default name (s3) |
|---|---|---|---|
| `working_copy`, `public_script`, `data_resource`, `snapshot` | `workspace` | `S3_WORKSPACE_BUCKET` | `workspace` |
| `version_artifact` | `version` | `S3_VERSION_BUCKET` | `version` |
| `run_log`, `run_result` | `run_log` | `S3_RUN_LOG_BUCKET` | `run-log` |
| (soft-delete target) | `trash` | `S3_TRASH_BUCKET` | `trash` |

`STORAGE_BACKEND=s3` → 4 separate S3 buckets.
`STORAGE_BACKEND=local` → 4 subdirectories under `LOCAL_STORAGE_BASE_DIR`
(default `/data`):

```
/data/
├── workspace/   # S3_WORKSPACE_BUCKET
├── version/     # S3_VERSION_BUCKET
├── run_log/     # S3_RUN_LOG_BUCKET
└── trash/       # S3_TRASH_BUCKET
```

A workspace's `Workspaces.artifact_bucket` column (when non-null) overrides
the default for that workspace, regardless of `usage_type` — useful for
isolating paid customers onto a dedicated bucket.

Object keys are a flat two-level path — `workspace_id` plus a server-issued
ULID — preserving the original file extension so Jupyter can pick its editor
from the suffix:

```
<bucket>/<workspace_id>/<ulid>{.<ext>}
```

File name, extension, MIME, and logical path all live on `StorageObjects` /
`Scripts` rows; reorganizing the bucket does not require rewriting the
database. Backend code never writes to the container local filesystem except
in `STORAGE_BACKEND=local` mode (where the shared `local-storage` volume IS
the canonical store). Schedule Executor stages node artifacts in
`tempfile.TemporaryDirectory()` (auto-cleanup). Only the `runtime` container
keeps a host volume — s3 mode needs it for rclone FUSE; local mode is a no-op
passthrough.

## Code layout

All Python packages use the `src/<pkg>/` layout; `uv` workspace glues them into
one `.venv`. Always invoke via `uv run [--package <pkg>] <cmd>` (see "Local
development" for the gotcha).

```
common/src/common/                Pure-Python shared library
  config.py                       Settings (pydantic-settings, lru_cache singleton)
  db/                             SQLAlchemy 2.0 async engine, session_scope, Base
  db/models/                      18 tables in 7 domain files (zero FK, zero relationship)
  auth/                           JWT / bcrypt / workspace membership helpers
  scheduler/                      APScheduler trigger helpers (delayed import)
  storage/                        AsyncStorageBackend abstraction + Pydantic schemas
    base.py                       Abstract interface
    factory.py                    create_storage + build_storage_config + PURPOSE_BUCKETS
    schemas.py                    CreateUploadRequest / ServerObjectRequest
    backends/local.py            Local filesystem impl
    backends/s3.py                S3-compatible impl (boto3)
    registry.py                   Bucket registry
  eventing.py                     add_outbox_event / utcnow / event_time
  service_app.py                  /health/ready TCP probe, /api/v1/health
  logging.py                      loguru config (LOG_LEVEL)
  schemas.py                      StrictModel base
  ids.py                          ULID generation helpers
  utils.py                        get_free_port, start_process

backend/src/backend/              Public FastAPI service
  main.py                         lifespan + route registration
  audit.py                        HTTP access log middleware (loguru sink)
  api/                            HTTP route handlers (one module per bounded context)
    auth.py                       /api/v1/auth/* (login / me / jupyter)
    jupyter.py                    /api/v1/auth/jupyter — the ONLY auth entry
    dependencies.py               request_context, database_session
    platform.py                   /api/v1/platform/* (system admin)
    admin.py                      /api/v1/admin/* (workspace-internal admin)
    scripts.py                    /api/v1/scripts/* + /api/v1/workspace-directories
    resources.py                  /api/v1/data-resources/*
    schedules/schedules.py        DAG CRUD
    schedules/runs.py             Run lifecycle
    storage.py                    /internal/v1/objects — single token-guarded endpoint (P0-1)
  services/                       Pure-Python business logic (no HTTP / no DI)
    scripts.py                    create_workspace_directory, visibility-filtered queries
    resources.py                  owner-scoped resource listing helpers
    jupyter.py                    jupyter_path / lock helpers
    storage.py                    object store helpers
    schedules.py                  DAG validation (cycle / orphan detection)
  schemas/                        Pydantic request / response models
    auth.py / common.py / jupyter.py / platform.py / resources.py / schedules.py / scripts.py
  clients/                        Outbound HTTP / RPC clients
    runtime.py                    Self-contained httpx wrapper for the runtime
    scheduler.py                  Backend → Schedule HTTP client (callback / dispatch)
    rclone.py                     rclone RC API client (FUSE cache invalidation)

schedule/src/schedule/            Schedule Executor (DAG worker)
  main.py                         Lifespan + FastAPI app
  notebook_runner.py              Subprocess entry point (nbclient) — DO NOT RENAME
  domain/                         Pure-Python domain types
    execution.py                  ExecutionResult (frozen dataclass) + state enums
    context.py                    Constants + naive_utc
  scheduling/                     Time-based trigger
    scheduler.py                  CronScheduler (APScheduler + 5s sync loop)
  application/                    Facades / orchestrators
    service.py                    SchedulerService (composes the three)
    orchestrator.py               DispatchOrchestrator (Outbox poll + DAG advance)
  execution/                      DAG node execution
    executor.py                   Dispatch helpers (script-type routing)
    worker.py                     NodeExecutor + asyncio entry, schedule-spawned task boundary
    runners/notebook.py           nbclient subprocess path (6-line `notebook_runner` shim re-exports `main`)
  infrastructure/                 External-system adapters
    storage/client.py             SchedulerStorageClient — talks to backend /internal/v1/objects

runtime/src/runtime/              Jupyter Runtime
  main.py                         FastAPI entry: jupyter action endpoints
  process.py                      Per-workspace subprocess pool + asyncio locks
  mount.py                        rclone FUSE mount lifecycle

frontend/                         React Router SPA (vite build → nginx)
  app/                            features/ routes/ services/ components/

migrations/                       Alembic schema versions
docker-compose.yml                5 services (migrate / web / backend / runtime / schedule)
default.conf                      Nginx template
scripts/nginx-entrypoint.sh
.env.example                      All 26 config.py keys documented
```

## Configuration system

All env vars go through one place: `common/src/common/config.py`.

```python
from common.config import settings

# Auth / runtime
settings.database_url          # str — SQLAlchemy async URL (mysql+asyncmy, charset utf8mb4)
settings.jwt_secret             # HS256 secret for the auth_request handler
settings.cookie_force_secure    # bool — write Secure flag even on plain HTTP (TLS-terminating proxy)
settings.service_name          # surfaced in /health
settings.schedule_event_namespace  # Outbox event-type namespace prefix (NOT APScheduler JobStore)
settings.readiness_targets     # CSV host:port list for /health/ready

# HTTP clients (intra-cluster URLs)
settings.runtime_api_url       # backend → runtime HTTP base
settings.public_base_url       # runtime public base URL (browser-facing /jupyter/)
settings.backend_api_url       # schedule → backend HTTP base
settings.rclone_rc_url         # backend → rclone RC control API
settings.internal_service_token  # Backend ↔ Schedule shared secret (X-Internal-Service-Token)

# Logging / audit
settings.log_level             # DEBUG / INFO / WARNING / ERROR / CRITICAL (lowercase → fallback INFO)
settings.audit_log_dir         # dir for daily audit logs (relative to cwd; "" disables file sink)
settings.audit_log_retention_days  # 0 disables cleanup
settings.audit_excluded_paths  # list[str] — paths skipped from audit (health probes, etc.)

# Storage
settings.storage_backend       # "s3" (default) or "local"
settings.local_storage_base_dir  # root dir for storage data (default "/data")
settings.s3_endpoint            # str (s3 mode only)
settings.s3_access_key          # str (s3 mode only)
settings.s3_secret_key          # str (s3 mode only)
settings.s3_workspace_bucket    # str (s3 mode only)
settings.s3_version_bucket      # str (s3 mode only)
settings.s3_run_log_bucket      # str (s3 mode only)
settings.s3_trash_bucket        # str (s3 mode only)
settings.s3_trash_retention_days  # int (s3 mode only)

# Schedule
settings.schedule_execution_concurrency  # int — max concurrent notebook subprocesses
```

### Encrypted env values

`Settings` runs a `model_validator` (`_decrypt_encrypted_fields`,
`common/src/common/config.py:168-180`) that scans every string field for the
prefix `ENC(...)` and decrypts the inner value with `APP_CONFIG_SECRET_KEY`
using Fernet. Use this for secrets that should not be stored in plain `.env`
files (e.g. third-party API tokens shipped via deployment config).

```bash
# .env
APP_CONFIG_SECRET_KEY=<base64 Fernet key — generate with `python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"`>
SOME_TOKEN=ENC(gAAAAABm...)        # ciphertext produced by Fernet.encrypt(b"plaintext")
```

At process start, `SOME_TOKEN` resolves to the decrypted plaintext. If the
field is *not* encrypted (no `ENC(...)` prefix), it passes through unchanged,
so plain `.env` files keep working. The pre-2026 in-repo `encrypt_secret.py`
script was the CLI wrapper around the same Fernet key; if you have old
ciphertexts they round-trip with the new `APP_CONFIG_SECRET_KEY` value.

`Settings` reads from process env first, then from a `.env` file at CWD
if present. `pydantic-settings` auto-loads. `case_sensitive=False` so
`DATABASE_URL` / `database_url` both work. The full list of 26 fields
is in `common/src/common/config.py`.

### Adding a new env var

1. Add the field to `Settings`:
   ```python
   new_var: str = Field(default="x", description="...")
   ```
2. Add the line to `.env.example` with a comment (keep it synced — every
   field in config.py must have a matching `.env.example` entry).
3. Use `settings.new_var` at the call site.

Do **not** call `os.environ["NEW_VAR"]` or `os.getenv("NEW_VAR")` in
application code. The grep below should return zero hits:

```bash
grep -rnE 'os\.(environ\[?["\x27][A-Z_]+|getenv\(["\x27][A-Z_]+)' --include="*.py" \
  backend/src/ schedule/src/ runtime/src/ common/src/
```

## Conventions

### Database / SQLAlchemy

- **No foreign keys, no `relationship`** — every join is explicit.
- Every table has `is_deleted TINYINT(1) NOT NULL DEFAULT 0` and
  `deleted_at DATETIME(3) NULL`; queries must filter `is_deleted == 0`
  (or `deleted_at.is_(None)`) to avoid logical-deleted rows.
- Domain files under `common/src/common/db/models/` are split by
  bounded context: `events` (2 tables: `ConsumerInbox`,
  `OutboxEvents`) / `identity` (4: `Users`, `Roles`, `RolePermissions`,
  `Permissions`) / `schedules` (5: `Schedules`, `ScheduleRuns`,
  `ScheduleNodes`, `ScheduleEdges`, `ScheduleNodeRuns`) / `scripts`
  (2: `Scripts`, `Versions`) / `storage` (3: `StorageObjects`,
  `UploadSessions`, `DataResources`) / `workspaces` (2: `Workspaces`,
  `WorkspaceMembers`) — 18 tables across 6 model files.
- All models are `class X(Base)` SQLAlchemy 2.0 declarative-mapped.
- The full schema is in `migrations/versions/`. Apply with:
  ```bash
  uv run --frozen --package backend alembic upgrade head
  ```

### Storage

- All object bytes go through `common.storage.AsyncStorageBackend`,
  created by `create_storage(config)` from `common.storage.factory`.
- Two backends are registered: `local` (filesystem, local mode) and
  `s3` (S3-compatible service, s3 mode). Selection is per-deployment
  via `settings.storage_backend` (`"s3"` default, `"local"` for
  dev / single-node / air-gapped).
- The factory helper `build_storage_config(bucket_name)` returns the
  right `create_storage` kwargs for each of the 4 purpose buckets
  (`workspace`, `version`, `run_log`, `trash`). Use it in lifespan code;
  route handlers don't see the difference.
- Bucket resolution from `usage_type` lives in **one place**
  (`common/src/common/storage/factory.py:actual_bucket_name` +
  `USAGE_TYPE_TO_PURPOSE`). The route handlers in
  `backend/src/backend/api/storage.py` only know about
  `app.state.object_stores[bucket_name]` and never call
  `settings.s3_*_bucket` directly.
- The runtime's view of the workspace bucket on disk is exposed by
  `common.storage.workspaces_root()`:
  - Both modes resolve to `${settings.local_storage_base_dir}/workspace`
    (default `/data/workspace`); s3 mode uses it as the rclone FUSE
    mount target, local mode uses it as a subdir of the shared
    `local-storage` volume.
  `settings.local_storage_base_dir` is the **only** path setting; the
  helper handles the per-mode suffix. Don't read `settings.workspaces_root`
  or any other path setting directly in runtime code — use this helper.
- The pre-2026 abstraction (`RustFSObjectStore` / `common.storage.client`
  / `StorageClient` HTTP wrapper) is gone. Don't reintroduce it.

### Auth

- Browser → `/jupyter/{workspace_id}/...` → Nginx `auth_request` →
  `GET /api/v1/auth/jupyter` (Backend).
- The handler:
  1. Parses cookie / Bearer JWT (HS256 + `settings.jwt_secret`).
  2. Verifies `WorkspaceMembers` for the workspace.
  3. Verifies `Scripts.is_locked` for the requested notebook path
     (owner / unlocked → allow; otherwise 403).
  4. Calls `RuntimeClient.get_workspace` / `start_workspace`.
  5. Returns `x-upstream-addr` + `x-jupyter-internal-token` response
     headers. **Browser never holds the runtime token.**

### Service-to-service auth (P0-1 fix)

- Schedule → Backend single endpoint ``POST /internal/v1/objects`` is
  guarded by ``require_internal_service`` in ``backend.api.storage``.
- The token header is ``X-Internal-Service-Token`` (case-insensitive
  on the wire because FastAPI ``Header`` lowercase-matches the name
  ``x-internal-service-token``); the secret value comes from
  ``settings.internal_service_token`` / env ``INTERNAL_SERVICE_TOKEN``.
- Comparison uses ``secrets.compare_digest`` — never equality.
- Backend and schedule must be configured with the same value; a
  mismatch fails fast at the first notebook run (``401``) which is
  intentional. ``.env.example`` ships a placeholder
  ``change-me-internal-service-token`` and the docker-compose
  ``${INTERNAL_SERVICE_TOKEN:?...}`` reference forces production
  deployments to set a real value.
- Removing the legacy backend / runtime host-port mappings
  (``8891:8000`` / ``8892:8000``) is part of the same fix — no service
  is reachable from the host except Nginx anymore.
- Nginx captures the headers via `auth_request_set` and proxies to the
  upstream sub-process with `Authorization: token $jupyter_token`.

### Permission gates

- For write operations on a script/notebook, call
  `require_script_modify_access(script, user_id=..., is_admin=...)`
  from `backend/src/backend/services/scripts.py`. It enforces:
  - admin or owner → allow
  - non-owner, `is_locked == 0` → allow
  - non-owner, `is_locked == 1` → 403
- Read endpoints (`list_scripts`, `get_script`) intentionally do **not**
  check `is_locked` — workspace members can see the script list.

### Owner-scoping + visibility (cross-owner browsing)

`GET /api/v1/scripts`, `GET /api/v1/data-resources`, and
`GET /api/v1/workspace-directories` all accept an optional
`owner_user_id` query param and follow the same model:

- `owner_user_id` 缺省 = 当前请求者本人(scope to `workspace/{me}/...`).
- 传值时 scope 到 `workspace/{owner_user_id}/...`,用于前端"点开其他成员
  分组"的懒加载(见 §3.4 of `API.md`).
- `visibility` 过滤(非 admin):`owner_user_id == me OR visibility IN
  (workspace, public)` — 自己可见自己全部(含 private),他人只见其
  workspace/public,排除他人 private.
- 系统管理员跳过 visibility 过滤.
- 目录行默认 `visibility='public'`(由 `create_workspace_directory` 写入),
  不施加 visibility 过滤,使跨 owner 目录树可见.

实现 helper 在 `backend/src/backend/services/scripts.py`
(`_build_list_scripts_owner_descendant_prefix`) 和
`backend/src/backend/api/resources.py` 中按相同模式分别构造 owner-scoped
LIKE 前缀。

### Eventing — Outbox + Inbox + business tables

The platform has **no** message broker. The only async fabric is three MySQL
tables, each on a different side of the same boundary:

| Table | Side | Job |
|---|---|---|
| business tables (`schedule_runs`, `schedule_node_runs`, etc.) | producer's local commit | The state change that *triggered* the event |
| `outbox_events` | producer side | Durable handoff: "this business state change produced this event" |
| `consumer_inbox` | consumer side | Idempotency: "this consumer has already processed this event" |

Together they implement the **Transactional Outbox** + **Consumer Inbox**
patterns: atomic publish on the producer side, idempotent side effects on
the consumer side. There is no Redis Stream, no Kafka, no in-process queue.
If MySQL is down, both sides are down — and that's intentional.

#### Why two tables and not one

A naïve "publish event by inserting a row, then have the consumer mark it
done" design breaks under two failure modes:

1. **Producer crash after business commit, before publish.** The state
   change is real but the event was never sent. Consumers never see it.
2. **Consumer crashes mid-processing.** The event was received but the
   side effect may have happened or not. Re-delivery causes double
   execution.

`outbox_events` fixes (1): writing to the business table and the outbox
happens in **one transaction**, so either both are durable or neither is.
A separate dispatcher (`schedule/application/orchestrator.py`) polls
`outbox_events` and pushes to consumers. The dispatcher crashing is
harmless — the next poll picks up where it left off.

`consumer_inbox` fixes (2): before applying a side effect, the consumer
inserts a row keyed by `(consumer_name, event_id)`. Re-delivery of the same
`event_id` hits the existing row and short-circuits.

#### Producer side — `outbox_events`

Schema lives in `common/db/models/events.py` (model `OutboxEvents`) and the
baseline migration `migrations/versions/e1f2a3b4c5d6_rebuild_baseline.py`.

Key fields:

- `event_id` (ULID, PK) — globally unique; `consumer_inbox` references this
  as the dedup key.
- `aggregate_type` + `aggregate_id` — the producer's domain object
  (e.g. `("schedule_run", "<run_id>")`).
- `event_type` — namespaced via `schedule_event_type(...)` so multiple
  deployments sharing one MySQL don't cross-consume (see below).
- `schema_version` (default 1) — consumer can branch on this when the
  payload shape changes.
- `payload_json` — the event body; opaque to the outbox.
- `event_status` (`pending` → `published` | `failed`) + `available_at` +
  `retry_count` + `last_error` — dispatcher state machine. Indexed on
  `(event_status, available_at, created_at)` because that's the poll
  hot path.
- `idempotency_key` + index — upstream dedup at the producer (e.g. two
  requests from the user that should produce one event, not two).
- `trace_id` — links to the request that produced the event for log
  correlation.

Write path — the **only** entry point is `common.eventing.add_outbox_event`:

```python
await add_outbox_event(
    session,
    event_type=schedule_event_type("schedule.run.requested"),
    producer="backend.api.schedules.runs",
    trace_id=request.state.trace_id,
    aggregate_type="schedule_run",
    aggregate_id=run_id,
    idempotency_key=f"schedule.run.requested:{run_id}",
    payload={"run_id": run_id, "schedule_id": schedule_id},
)
```

`session` is the **same** SQLAlchemy session as the business-table write;
the outbox row goes in via `session.add(...)` and commits together with
the business row. Never call `session.commit()` between the business
write and the outbox write — that defeats the whole pattern.

**Event type namespacing.** Every event type MUST go through
`common.eventing.schedule_event_type(raw)` before being passed to
`add_outbox_event`. The helper prefixes `settings.schedule_event_namespace`
(default `model-platform-develop`) so that two deployments sharing one
MySQL — common during development — don't accidentally consume each
other's events. The constants `SCHEDULE_RUN_REQUESTED_EVENT`,
`NODE_EXECUTE_EVENT`, `NODE_FINISHED_EVENT` in
`schedule/application/orchestrator.py` and `schedule/execution/worker.py`
are already namespaced; never hard-code the raw string.

Current event types in use:

| Event | Produced by | Consumed by |
|---|---|---|
| `schedule.run.requested` | `backend/api/schedules/runs.py`, cron post-back | `schedule/application/orchestrator.py` (`schedule-orchestrator`) |
| `job.node.execute` | `schedule/application/orchestrator.py` when a node is ready | `schedule/execution/worker.py` (`schedule-results` writes back via the orchestrator) |
| `job.node.finished` | `schedule/execution/worker.py` after a node run finishes | `schedule/application/orchestrator.py` (`schedule-results`) |

The dispatcher polls every `0.25s` when there's pending work, dropping to
`1s` when idle — see the loop in
`schedule/application/orchestrator.py` (the `asyncio.sleep(0.25)` /
`asyncio.sleep(1)` branches).

#### Consumer side — `consumer_inbox`

Model `ConsumerInbox` in `common/db/models/events.py`. Composite PK
`(consumer_name, event_id)` so multiple consumers can independently
process the same `event_id`.

Lifecycle (lives in `schedule/application/orchestrator.py:_start_inbox` /
`_finish_inbox`):

```
        ┌──────────────┐
        │  processing  │  ← INSERT or re-claim on re-delivery
        └──────┬───────┘
               │
       success │  failure
               ▼
       ┌─────────────┐
       │  succeeded  │   (terminal — no re-process)
       └─────────────┘

       on failure, error_message is set and the row is left in
       `failed` for inspection; the dispatcher does NOT auto-retry
       consumer failures (only producer-side publish failures).
```

The consumer **must** call `_start_inbox` (or equivalent) at the top of
every event handler. The function returns `(inbox_row, should_process)`;
if `should_process` is `False`, the event was already handled and the
handler returns immediately. On success the handler calls `_finish_inbox`
to flip `process_status` to `succeeded`.

Currently registered `consumer_name` values (see `orchestrator.py:594`,
`:924`):

- `schedule-orchestrator` — consumes `schedule.run.requested`
- `schedule-results` — consumes `job.node.finished`

A new consumer = a new `consumer_name` string. Two consumers sharing a
name will collide on the PK — pick a stable, descriptive name and treat
it as a contract.

#### Failure modes the design covers

| Scenario | What happens |
|---|---|
| Backend crashes after business commit, before dispatcher polls | Outbox row exists; next poll picks it up. |
| Dispatcher crashes after poll, before HTTP push to executor | `event_status` still `pending`; next poll retries. |
| Executor crashes mid-handler | Inbox row stays `processing`; on redelivery the handler re-enters `_start_inbox`, sees `succeeded`? — no, sees `processing` and re-runs. **This is currently a known soft spot** — the executor's `_finish_inbox` must run, and a crash before that means a re-run. Don't perform side effects before `_finish_inbox` succeeds. |
| Same `event_id` delivered twice (e.g. HTTP retry after success) | Inbox short-circuits; the second delivery is a no-op. |
| Multiple development deployments share one MySQL | `schedule_event_namespace` prefixes keep them isolated; each deployment only sees its own events. |

#### Adding a new event

1. Pick an `aggregate_type` / `aggregate_id` pair that identifies the
   producing domain object.
2. Pick a `consumer_name` for each consumer. Stable, descriptive,
   never reused for a different purpose.
3. Define the event type constant:
   ```python
   # in the producing module
   MY_NEW_EVENT = schedule_event_type("schedule.my_new_event")
   ```
4. Write it via `add_outbox_event` in the same transaction as the
   business mutation.
5. In the consumer, start with `_start_inbox` (or follow the pattern
   in `orchestrator.py`) before doing any side effects, and call
   `_finish_inbox` on success.
6. Update the table above.

Do **not** invent a new messaging fabric (Redis Stream, Kafka, in-process
queue). The whole point of this design is that MySQL is the single
authority — adding a second one doubles the failure surface.

### Async / sync signatures

- `SchedulerService.start()` and `SchedulerService.close()` are
  `async def` (so the FastAPI lifespan can `await` them).
- `CronScheduler.start()`, `DispatchOrchestrator.start()` are
  **`def` (sync)** — they only `create_task(...)` and return. Don't
  `await` them.
- `cron.start()`, `orchestrator.start()`, `worker.handle_node_execute`
  are wired together in `SchedulerService.__init__`; their lifetime
  is owned by the facade.

## Local development

### One-time setup

```bash
# Python workspace (monorepo via uv workspaces)
uv sync --all-packages

# Frontend deps
cd frontend && pnpm install && cd ..
```

### Per-service dev

Always go through `uv run` so the workspace `.venv` is used — bare
`uvicorn` / `python` resolves to system Python and `from backend.X`
imports fail with ModuleNotFoundError.

```bash
# Backend (terminal 1)
export DATABASE_URL="mysql+asyncmy://model_platform:model_platform@127.0.0.1:3306/model_platform?charset=utf8mb4"
export STORAGE_BACKEND=s3
export S3_ACCESS_KEY=modelplatform
export S3_SECRET_KEY=modelplatformsecret
export S3_ENDPOINT=http://127.0.0.1:9000
# Or for local mode:
# export STORAGE_BACKEND=local
# export LOCAL_STORAGE_BASE_DIR=/data
uv run --package backend uvicorn backend.main:app --host 0.0.0.0 --port 8000 --reload

# Schedule Executor (terminal 2)
uv run --package schedule uvicorn schedule.main:app --host 0.0.0.0 --port 8001 --reload

# Runtime (terminal 3 — needs SYS_ADMIN, FUSE, devmode)
uv run --package runtime python -m runtime.main
```

### Frontend dev

```bash
cd frontend
pnpm dev          # http://localhost:5173, proxies /api to backend
pnpm typecheck
pnpm build
```

### Static checks

```bash
# Python compile
uv run --frozen --package backend python -m compileall -q backend/src common/src
uv run --frozen --package schedule python -m compileall -q schedule/src
uv run --frozen --package runtime python -m compileall -q runtime/src

# Type check (frontend)
cd frontend && pnpm typecheck && cd ..

# Docker compose config
docker compose config --quiet
```

### Smoke test

```bash
# Run the full ORM import + Settings smoke test
PYTHONPATH="backend/src:common/src" uv run --frozen --package backend python -c "
from backend.main import app
from common.config import settings
print('backend:', len(app.routes), 'routes')
print('settings ok:', settings.s3_endpoint)
"
```

## Common tasks

### Add a new DAG endpoint

1. Add the route handler in `backend/src/backend/api/schedules/schedules.py`
   (DAG template) or `backend/src/backend/api/schedules/runs.py` (run
   lifecycle).
2. Validate request via Pydantic schemas in
   `backend/src/backend/schemas/schedules.py`.
3. For mutations on nodes/edges/versions: route through
   `create_script_record` / `get_script_row` and apply
   `require_script_modify_access` if it touches a script.
4. If it produces an outbox event, use
   `add_outbox_event(session, event_type=schedule_event_type("..."), producer="...", ...)`
   in the same transaction as the business write.

### Add a new env var

See "Adding a new env var" above.

### Add a new MySQL table

1. Add a model class in `common/src/common/db/models/<domain>.py`.
   Include `is_deleted TINYINT(1) NOT NULL DEFAULT 0` and
   `deleted_at DATETIME(3) NULL`.
2. Export it from `common/src/common/db/models/__init__.py`.
3. Generate the migration:
   ```bash
   uv run --frozen --package backend alembic revision --autogenerate -m "add <feature>"
   ```
4. Review the generated `migrations/versions/*.py` — Alembic may
   miss comments / server defaults. Manually fix the migration.
5. Apply locally:
   ```bash
   uv run --frozen --package backend alembic upgrade head
   ```

### Wire a new storage bucket

The current 4 buckets are wired in `common/src/common/storage/factory.py`.
`USAGE_TYPE_TO_PURPOSE` maps each `usage_type` (`working_copy`,
`public_script`, `data_resource`, `snapshot`, `version_artifact`,
`run_log`, `run_result`) to one of the 4 purpose buckets (`workspace`,
`version`, `run_log`, `trash`); `actual_bucket_name(purpose)` then
returns the s3-style identifier from the corresponding
`settings.s3_<purpose>_bucket`. `BUCKET_FOR_USAGE` is a dict
comprehension built from these two.

The constant `PURPOSE_BUCKETS = ("workspace", "version", "run_log", "trash")`
in `common/storage/factory.py` enumerates the four backends built in the
backend lifespan. To add a fifth bucket:

1. Add the env var to `Settings` (s3 mode only):
   ```python
   s3_<feature>_bucket: str = Field(default="<feature>", description="...")
   ```
2. Add to `.env.example` with a one-line comment.
3. Append `"<feature>"` to the `PURPOSE_BUCKETS` tuple in
   `common/storage/factory.py`. `build_storage_config("<feature>")`
   will then automatically read `settings.s3_<feature>_bucket` (s3
   mode) or use `<local_storage_base_dir>/<feature>` (local mode).
4. Extend the `Literal` in `common/storage/schemas.py` (in
   `CreateUploadRequest.usage_type`, `ServerObjectRequest.usage_type`)
   to include the new value.
5. Add an entry in `USAGE_TYPE_TO_PURPOSE` mapping the new `usage_type`
   to the new purpose; `BUCKET_FOR_USAGE` is regenerated automatically.
6. Pre-create the bucket (s3 mode) or subdirectory (local mode) in the
   deployment. The backend no longer auto-creates buckets.

A workspace's `Workspaces.artifact_bucket` column (when non-null)
overrides the default for that workspace, regardless of `usage_type`.

### Add a new schedule node type

`schedule/src/schedule/execution/runners/notebook.py:execute_artifact`
dispatches on `script_type`. Add a new branch + a new `_<type>` function
in that file. The NodeExecutor in `worker.py` does not need to change —
the dispatch happens inside `execute_artifact`.

## Tests

There is an in-tree test suite, mostly covering scripts / resources /
DAG validation / upload state transitions. It is **not** the formal
release-gate suite the project still owes (see `HANDOVER.md` §8 for
known gaps: trash reaper, cross-backend migration, ENC round-trip).

Current coverage:

| Area | Tests | Where |
|---|---|---|
| Scripts (CRUD, soft delete, parent path, same-name siblings, visibility) | 10 functions across `test_scripts.py`, `test_count_scripts.py`, `test_list_scripts_parent_path.py`, `test_storage_upload_status.py` | `backend/tests/` |
| Resources (visibility, ownership, idempotency) | ~5 in `test_resources.py` | `backend/tests/` |
| DAG validation (cycle / orphan detection) | `test_validate_dag.py` | `backend/tests/` |
| Audit log middleware | `test_audit_logging.py` | `backend/tests/` |
| Jupyter auth cache | `test_jupyter_auth_cache.py` | `backend/tests/` |
| Runtime client (directory listing, error mapping) | `test_runtime_client_directories.py` | `backend/tests/` |
| Schedule layer (worker, janitor, layering invariants) | 11 functions in `test_janitor.py`, `test_layering.py`, `test_worker.py` | `schedule/tests/` |

Run:

```bash
# Backend
uv run --package backend pytest backend/tests -q

# Schedule
uv run --package schedule pytest schedule/tests -q
```

Conventions:

- pytest + `pytest-asyncio` for `async def` handlers.
- MySQL is required for the ORM tests (not SQLite — CHAR(26) ULIDs and
  `mysql.TINYINT(1)` quirks do not translate). Local docker-compose
  MySQL is the typical target.
- For storage, `common.storage.factory` selects between s3 and local
  via `settings.storage_backend`; tests that exercise both modes
  monkeypatch that setting (see `common/tests/storage/test_factory.py`).
- For HTTP boundaries (httpx to backend / runtime), tests use `respx`
  with `assert_all_called=False` so unused stubs don't fail the test
  — see the engineering notes in `CLAUDE.md`.

## Troubleshooting

### "the greenlet library is required"

SQLAlchemy 2.0 needs `greenlet` for `engine.dispose()` in async
contexts. Add `greenlet>=3.0.0` to `common/pyproject.toml` and
`uv sync --all-packages`. (Already present in this repo.)

### "Can't connect to MySQL server"

Either MySQL isn't running, or the network namespace doesn't allow
`mysql:3306` resolution. Inside the Docker network, services reach
each other by service name (`mysql`, `backend`, `runtime`,
`schedule`, `s3`).

### Jupyter routing 401s

Inspect `docker compose logs backend` — `jupyter.py:check_notebook_is_locked`
or `load_active_membership` will return an explicit reason. Then
check the JWT (use `JWT_SECRET` from `.env`).

### Schedule run never advances

`schedule_runs.run_status` is stuck at `queued`. Two likely causes:
- `outbox_events` is empty (Backend's `add_outbox_event` failed —
  check `add_outbox_event` in `schedule_runs.py`).
- The orchestrator's polling loop is dead. Check
  `docker compose logs schedule` and look for "database event loop
  failed" exceptions.

### "AttributeError: 'SchedulerService' object has no attribute 'worker'"

`worker` must be constructed before `orchestrator` in
`SchedulerService.__init__`, because orchestrator's dispatch table
captures `self.worker.handle_node_execute` at construction time.
See `schedule/src/schedule/application/service.py` — the order is
load-bearing. (This was the symptom during the flat → layered schedule
refactor; if you see it today, the most likely cause is a partial
rebase that left an old import path.)

## Style

- Type hints everywhere (this repo uses `from __future__ import
  annotations`).
- 4-space indent, double quotes, no trailing whitespace.
- Comments are technical (explain *why*, not *what*).
- Module docstrings document non-obvious invariants. Don't add
  docstrings to functions whose behavior is self-evident from the
  name.
- 4 levels of indentation = "this function is doing too much; split
  it". (Project convention; see e.g. `execute_artifact`.)

## See also

- `ARCHITECTURE.md` — kept as a thin redirect; the authoritative
  architecture diagrams and capability map now live in
  [§Architecture](#architecture) above.
- `HANDOVER.md` — current refactor state, recent commits, pending work.
- `CLAUDE.md` — agent-facing conventions for the repo.
- `common/src/common/db/models/__init__.py` — exhaustive list of all 18 tables.
- `common/src/common/config.py` — all env vars in one place.
- Per-package READMEs: `backend/README.md`, `common/README.md`,
  `runtime/README.md`, `frontend/README.md`.