# 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: ``` //{.} ``` 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//` layout; `uv` workspace glues them into one `.venv`. Always invoke via `uv run [--package ] ` (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= 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", "")`). - `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/.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 " ``` 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__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__bucket: str = Field(default="", description="...") ``` 2. Add to `.env.example` with a one-line comment. 3. Append `""` to the `PURPOSE_BUCKETS` tuple in `common/storage/factory.py`. `build_storage_config("")` will then automatically read `settings.s3__bucket` (s3 mode) or use `/` (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 `_` 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`.