The fix lands in three concentric layers, all backed by a single
INTERNAL_SERVICE_TOKEN shared secret so we have one mechanism
instead of three:
1. docker-compose: drop the backend.ports: 8891:8000 and
runtime.ports: 8892:8000 mappings. Nginx is the only host
ingress again (architecture §2.2).
2. /internal/v1/*: the storage control plane had six endpoints, five
of which were dead code (frontend already migrated to
/api/v1/data-resources/* with JWT; schedule only ever called
POST /internal/v1/objects). Delete the dead routes, mount the
one survivor with Depends(require_internal_service) that
compares the X-Internal-Service-Token header against
settings.internal_service_token with secrets.compare_digest.
3. POST /api/v1/jupyter on the runtime container: previously open
inside the Docker network. Same token mechanism — backend's
runtime_http_client now carries the header, runtime's
handle_jupyter_action requires the same header. /api/v1/health
stays open for the Nginx and compose healthchecks.
The schedule worker was already configured to call
POST /internal/v1/objects; build_storage_http_client now
sets the token header so its existing call site keeps working
without changes.
Files touched:
backend/src/backend/storage_api.py # 5 dead routes deleted + token guard
backend/src/backend/main.py # runtime_http_client header
runtime/src/runtime/main.py # require_internal_service Depends
common/src/common/config.py # internal_service_token setting
schedule/src/schedule/service.py # httpx client header
docker-compose.yml # ports dropped, INTERNAL_SERVICE_TOKEN env
.env.example # INTERNAL_SERVICE_TOKEN placeholder
API.md / README.md / DEVELOP.md # §9 trimmed to 1 endpoint
Verified:
compileall -> 0 errors
pytest backend/tests -> 37 passed
in-process ASGI smoke:
POST /internal/v1/objects no/wrong/correct token -> 401/401/200
POST /api/v1/jupyter no/wrong/correct token -> 401/401/200
5 deleted internal routes -> 404
docker compose config (with env) -> OK
P0-1 still has one open sub-item (rclone RC --rc-no-auth) that
the user has explicitly deferred; not touched here.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
18 KiB
DEVELOP.md — Developer Guide
This guide is for engineers working on the model platform codebase. For
high-level design see ARCHITECTURE.md; for the current state of in-flight
refactors see HANDOVER.md.
Code layout
common/ Pure-Python shared library
config.py Settings (pydantic-settings, lru_cache singleton)
db/ SQLAlchemy 2.0 async engine, session_scope, Base
db/models/ 26 tables in 9 domain files (zero FK, zero relationship)
scheduler/ build_sqlalchemy_jobstore (delayed import)
storage/ AsyncStorageBackend abstraction (s3 + local impls) + Pydantic schemas
eventing.py add_outbox_event / utcnow / event_time
service_app.py /health/ready TCP probe, /api/v1/health
schemas.py StrictModel base
utils.py get_free_port, start_process
backend/ Public FastAPI service + tiny /internal/v1/objects RPC
main.py lifespan + route registration
jupyter.py /api/v1/auth/jupyter — the ONLY auth entry
scripts.py CRUD for scripts/notebooks (object storage via AsyncStorageBackend)
schedules.py DAG CRUD: schedules, nodes, edges
schedule_runs.py Trigger / list / get runs
schedule_schemas.py Pydantic request/response models
admin.py Admin endpoints
resources.py Misc data resources
storage_api.py /internal/v1/objects — single token-guarded endpoint (P0-1)
storage_client.py Stub (HTTP client removed post-migration; rewrite pending)
schedule_client.py Placeholder module (was the HTTP-push executor client)
runtime_client.py Self-contained httpx wrapper for the runtime
jupyter.py auth_request handler
dependencies.py request_context, database_session
schedule/ Schedule Executor (DAG worker)
context.py Constants + naive_utc
scheduler.py CronScheduler (APScheduler + 5s sync loop)
orchestrator.py DispatchOrchestrator (Outbox poll + DAG advance)
worker.py NodeExecutor (notebook / python execution)
service.py SchedulerService facade (composes the three)
main.py Lifespan + FastAPI app
storage_client.py SchedulerStorageClient — talks to backend /internal/v1/objects
execution.py execute_artifact (notebook + python paths)
notebook_runner.py Subprocess entry point (nbclient)
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 4 services
default.conf Nginx template
scripts/nginx-entrypoint.sh
.env.example
Configuration system
All env vars go through one place: common/src/common/config.py.
from common.config import settings
settings.database_url # str
settings.storage_backend # str: "s3" (default) or "local"
settings.local_storage_base_dir # str: root dir for storage data (default "/data"); see "Storage" below for per-mode derivation
settings.s3_endpoint # str (full URL, e.g. "http://s3:9000"; 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)
settings.jwt_secret # HS256 secret for the auth_request handler
settings.backend_api_url # schedule → backend HTTP base
settings.runtime_api_url # backend → runtime HTTP base
settings.public_base_url # runtime public base URL
settings.service_name # surfaced in /health
settings.readiness_targets # CSV host:port list for /health/ready
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.
Adding a new env var
- Add the field to
Settings:new_var: str = Field(default="x", description="...") - Add the line to
.env.examplewith a comment. - Use
settings.new_varat 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 0anddeleted_at DATETIME(3) NULL; queries must filteris_deleted == 0(ordeleted_at.is_(None)) to avoid logical-deleted rows. - Domain files under
common/src/common/db/models/are split by bounded context:audit/events/experiments/identity/runtime/schedules/scripts/storage/workspaces. - 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 bycreate_storage(config)fromcommon.storage.factory. - Two backends are registered:
local(filesystem, local mode) ands3(S3-compatible service, s3 mode). Selection is per-deployment viasettings.storage_backend("s3"default,"local"for dev / single-node / air-gapped). - The factory helper
build_storage_config(bucket_name)returns the rightcreate_storagekwargs 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_typeis in one place (backend/storage_api.py:resolve_bucket); route handlers only know aboutapp.state.object_stores[bucket_name]. - The runtime's view of the workspace bucket on disk is exposed by
common.storage.workspaces_root():s3mode:${settings.local_storage_base_dir}/workspace(default/data/workspace, the rclone FUSE mount target).localmode:${settings.local_storage_base_dir}/workspace(default/data/workspace, a subdir of the shared local-storage volume).settings.local_storage_base_diris the only path setting; the helper handles the per-mode suffix. Don't readsettings.workspaces_rootor any other path setting directly in runtime code — use this helper.
- The pre-2026 abstraction (
RustFSObjectStore/common.storage.client/StorageClientHTTP wrapper) is gone. Don't reintroduce it.
Auth
- Browser →
/jupyter/{workspace_id}/...→ Nginxauth_request→GET /api/v1/auth/jupyter(Backend). - The handler:
- Parses cookie / Bearer JWT (HS256 +
settings.jwt_secret). - Verifies
WorkspaceMembersfor the workspace. - Verifies
Scripts.is_lockedfor the requested notebook path (owner / unlocked → allow; otherwise 403). - Calls
RuntimeClient.get_workspace/start_workspace. - Returns
x-upstream-addr+x-jupyter-internal-tokenresponse headers. Browser never holds the runtime token.
- Parses cookie / Bearer JWT (HS256 +
Service-to-service auth (P0-1 fix)
- Schedule → Backend single endpoint
POST /internal/v1/objectsis guarded byrequire_internal_serviceinbackend.storage_api. - The token header is
X-Internal-Service-Token(case-insensitive on the wire because FastAPIHeaderlowercase-matches the namex-internal-service-token); the secret value comes fromsettings.internal_service_token/ envINTERNAL_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.exampleships a placeholderchange-me-internal-service-tokenand 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_setand proxies to the upstream sub-process withAuthorization: token $jupyter_token.
Permission gates
- For write operations on a script/notebook, call
require_script_modify_access(script, user_id=..., is_admin=...)frombackend/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 checkis_locked— workspace members can see the script list.
Outbox events
- The platform's only async-messaging fabric is the MySQL
OutboxEventstable. Producers (Backend) write rows in the same transaction as the business state. Consumers (Schedule Executor) poll every 250 ms and updateevent_statustopublishedorfailed(with retry). - Use
common.eventing.add_outbox_eventto write. event_typevalues currently in use:schedule.run.requested— produced byschedule_runs.py/ cron post-backjob.node.execute— produced by orchestrator when a node is readyjob.node.finished— produced by worker after node execution
consumer_inboxprovides exactly-once delivery per(consumer_name, event_id)(withprocess_status: processing → succeededlifecycle).
Async / sync signatures
SchedulerService.start()andSchedulerService.close()areasync def(so the FastAPI lifespan canawaitthem).CronScheduler.start(),DispatchOrchestrator.start()aredef(sync) — they onlycreate_task(...)and return. Don'tawaitthem.cron.start(),orchestrator.start(),worker.handle_node_executeare wired together inSchedulerService.__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
# 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 --frozen --package backend uvicorn backend.main:app --host 0.0.0.0 --port 8000 --reload
# Schedule Executor (terminal 2)
uv run --frozen --package schedule uvicorn schedule.main:app --host 0.0.0.0 --port 8001 --reload
# Runtime (terminal 3 — needs SYS_ADMIN, FUSE, devmode)
uv run --frozen --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
- Add the route handler in
backend/schedules.py(DAG template) orbackend/schedule_runs.py(run lifecycle). - Validate request via
backend/schedule_schemas.py. - For mutations on nodes/edges/versions: route through
create_script_record/get_script_rowand applyrequire_script_modify_accessif it touches a script. - If it produces an outbox event, use
add_outbox_event(session, event_type="...", producer="...", ...).
Add a new env var
See "Adding a new env var" above.
Add a new MySQL table
- Add a model class in
common/src/common/db/models/<domain>.py. Includeis_deleted TINYINT(1) NOT NULL DEFAULT 0anddeleted_at DATETIME(3) NULL. - Export it from
common/src/common/db/models/__init__.py. - Generate the migration:
uv run --frozen --package backend alembic revision --autogenerate -m "add <feature>" - Review the generated
migrations/versions/*.py— Alembic may miss comments / server defaults. Manually fix the migration. - Apply locally:
uv run --frozen --package backend alembic upgrade head
Wire a new storage bucket
The current 4 buckets are wired in backend/storage_api.py:resolve_bucket:
BUCKET_FOR_USAGE: dict[str, str] = {
"working_copy": settings.s3_workspace_bucket,
"public_script": settings.s3_workspace_bucket,
"data_resource": settings.s3_workspace_bucket,
"snapshot": settings.s3_workspace_bucket,
"version_artifact": settings.s3_version_bucket,
"run_log": settings.s3_run_log_bucket,
"run_result": settings.s3_run_log_bucket,
}
The constant PURPOSE_BUCKETS = ("workspace", "version", "run_log", "trash")
in common.storage.factory enumerates the four backends built in the
backend lifespan. To add a fifth bucket:
- Add the env var to
Settings(s3 mode only):s3_<feature>_bucket: str = Field(default="<feature>", description="...") - Add to
.env.examplewith a one-line comment. - Append
"<feature>"to thePURPOSE_BUCKETStuple incommon/storage/factory.py.build_storage_config("<feature>")will then automatically readsettings.s3_<feature>_bucket(s3 mode) or use<local_storage_base_dir>/<feature>(local mode). - Extend the
Literalincommon/storage/schemas.py(inCreateUploadRequest.usage_type,ServerObjectRequest.usage_type) to include the new value. - Add an entry in
BUCKET_FOR_USAGEmapping the newusage_typeto the new bucket env var. - Pre-create the bucket (s3 mode) or subdirectory (local mode) in the deployment. The backend no longer auto-creates buckets.
A workspace's artifact_bucket column (when non-null) overrides the
default for that workspace, regardless of usage_type.
Add a new schedule node type
schedule/execution.py dispatches on script_type in
execute_artifact. Add a new branch + a new _<type> function.
worker.py does not need to change — the dispatch happens inside
execute_artifact.
Tests
There is no formal test suite yet (see HANDOVER §Pending Tasks P1). A reasonable first test surface:
require_script_modify_access(admin / owner / non-owner-unlock / non-owner-lock): pure-function unit test, no DB.validate_dag(cycle detection + orphan detection) inbackend/schedules.py.execute_artifactend-to-end with mockedcontent_hashand a realtempfile.TemporaryDirectory.
Test convention: pytest with pytest-asyncio for async def
handlers. Use SQLite in-memory (or a MySQL test container) for DB
integration. Use moto for S3.
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_eventsis empty (Backend'sadd_outbox_eventfailed — checkadd_outbox_eventinschedule_runs.py).- The orchestrator's polling loop is dead. Check
docker compose logs scheduleand 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 service.py — the order is load-bearing.
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— design diagramsHANDOVER.md— current refactor state and pending workCLAUDE.md— agent-facing conventions for the repomodels / __init__.py— exhaustive list of all 26 tablescommon/config.py— all env vars in one place