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.

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).

# .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:

   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:

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:

  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:

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:

   # 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

# 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.

# 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

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

Static checks

# 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

# 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:

   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:

   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):

   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:

# 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 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.