control-plane

Imported from al2m4n/TheYgent (apps/control-plane/AGENTS.md). Install upstream with npx skills add al2m4n/TheYgent --skill control-plane. Copyright stays with the author.

AGENTS.md — control plane

apps/control-plane (theygent_control_plane) is the orchestration service: one FastAPI app on :8080 that runs agent graphs, keeps the agent registry and its versions, chat sessions, triggers, connections and secrets, MCP hosting, retrieval, observability, platform settings, identity, and the AI gateway's records. It reaches models only over the inference plane's HTTP data plane. The repo-wide rules in the root AGENTS.md apply first; the design is in control-plane.md, graph-execution.md and durable-execution.md.

Where it sits

• Called by: the interface (every operator surface except raw model traffic), the MCP server (with a personal API key), the AI gateway (/gateway/* with a service key), and external callers (/agents/{id}/invoke with an API key or THEYGENT_INVOKE_TOKEN, /hooks/{trigger_id} signed with X-TheYgent-Signature).
• Calls: the inference plane's /v1/* through packages/gateway-client, by logical model id only; MCP servers; HTTP tools; the Apify API; OIDC providers; an OTLP collector when export is on.
• Shares: Postgres with the worker (app tables in public, DBOS checkpoints in dbos); the IR models with everyone through packages/ir.
• Owns on disk: the artifact store (THEYGENT_ARTIFACT_DIR) for audio and image bytes.
• Never: imports an engine, reads the inference plane's state files, or receives the HF token or the organization's Anthropic key.

Map

All paths under src/theygent_control_plane/:

Route groups in app.py

Sections are marked with # ── … ── banners in create_app, in this order:

Wire casing here is snake_case. A few surfaces whose payloads mirror node config (crawler catalog, MCP hub install) accept camelCase aliases on purpose.

Key flows

1. Interactive graph run (POST /graphs/runs, POST /agents/{id}/runs): parse and validate_graph (400 invalid_ir, nothing persisted) → reject engine names and unexecutable types → create the Run row → walker.walk() inside a RunTrace → stream SSE deltas → persist the output, spans and node I/O. A client disconnect cancels the run.
2. Durable run (triggers when THEYGENT_DURABLE=1, POST /agents/{id}/durable-runs): DurableRuntime.fire() enqueues theygent_run with the pinned agent_ref; the durable package walks it step by step.
3. Publish: POST /agents hashes the view-stripped IR (theygent_ir.content_hash) → 409 agent_exists → the interface adds a version through POST /agents/{id}/versions; the same hash function runs on every run, so registry and runtime never disagree.
4. Chat turn: POST /runs (a model) or POST /agents/{id}/runs (an agent) with a session_id replays the stored turns verbatim and appends the new pair after the stream ends.

Making changes

• New endpoint: a request model near the top of app.py, a closure in the matching section, require_auth / require_editor / require_admin, governance.authorize() for sensitive reads, the house error envelope, a test in tests/, and the user docs API reference (docs/user-docs/content/reference/api.md).
• New node type: see the execution rules below and durable/AGENTS.md.
• New table or column: a migration, models.py, the store, and the chain table in alembic/AGENTS.md.
• New platform setting: the catalog in settings.py, the interface's settings tab, and docs/user-docs/content/reference/settings.md.
• New env var: read it empty-safe, add it to .env.example, the deploy manifests that need it, and docs/user-docs/content/reference/environment.md.

Structure rules

• One FastAPI app, built by the create_app factory in app.py; every stateful backend (stores, gateway client, MCP manager, tool resolver, gates, telemetry, settings) is injected so tests can substitute fakes. Importing the module must have no side effects — asgi.py is the entrypoint.
• Domain vs persistence split: Pydantic entities in run.py are the wire/logic shape, SQLAlchemy rows in models.py are the storage shape, store.py maps between them. ORM rows never leak out of a store; the wire contract is never welded to the table schema.
• Alembic owns the schema. Migrations are hand-written, linear, and reviewed; never create_all, including in tests. The DBOS dbos schema is managed by DBOS itself and stays outside the Alembic chain.
• Stores never commit — the caller owns the transaction. Read handlers use the request-scoped session; writes use one transaction per logical operation. Ordering keys are explicit columns (message.position, agent_version.seq, span.seq), never timestamps.
• Every sensitive read goes through governance.authorize() (trace, I/O, settings, transfer, gateway usage permissions). It ranks the caller's role against a per-permission floor; ownership checks live at the endpoints that carry an owner column. Keep new sensitive surfaces routed through it rather than adding a second authorization path.
• Errors are the house envelope {"error": {"message", "code"}} with stable snake_case codes; add new codes, don't repurpose existing ones.

Execution rules

• The walker (walker.py) does no DB I/O — it is a pure async interpreter over an injected context. Keep new node executors runtime-agnostic so both the in-process walker and the durable runtime reuse them.
• A new node type is additive in three places: a handler in walker.py, a lowering in durable/compiler.py, and a NODE_TYPE_KIND entry in packages/ir — plus tests in both runtimes. Dispatch is by kind first, then type.
• Durable discipline: the dbos import stays under durable/. The registered workflows are generic, never per agent: theygent_run, plus theygent_scheduled_fire, which runs it as a child. DBOS recovers by registered name, so per-agent workflows would be unrecoverable. Orchestration nodes (router/transform/loop control) do no I/O — they must re-derive identically from journaled step results on replay. All I/O lives in step bodies. Only serializable data crosses the workflow→step boundary; bytes travel as artifact references, never journaled.
• Model ids: engine names (mlx/vllm/llamacpp) are rejected up front with 400 engine_name_not_allowed on every model-carrying surface — before a Run row exists.
• The tool ok/err contract: a failing tool/mcp/http/rag/audio/image step binds a structured error to the node's error handle and the run continues; only transport-level failures fail the run. An edge is live only if its source executed and activated that handle (router selects one, tool ok XOR err, guardrail pass XOR block).
• $in resolution is loud: unknown roots, undeclared ports, and missing fields raise errors naming the node and token — silent literal pass-through is forbidden.
• Secrets resolve at step time only (inside SecretStore.resolve()); plaintext never appears in the IR, logs, or journal. Connection auth is built server-side per call.
• MCP connections follow the actor pattern: each transport's async contexts live on one dedicated task; callers enqueue calls. Lazy connect, reuse for process lifetime, one reconnect retry — no eviction/arbitration (MCP processes are cheap; engines are heavy).
• RAG: a graph references a source by stable id, so re-ingesting never changes an agent's contentHash. The embedding model is pinned per source by logical id; ingest replaces documents atomically (chunk + embed before any row is touched).
• Observability: both runtimes run inside the same span-capture wrapper; span/node-IO rows are the source for the UI waterfall, and OTLP export is an opt-in second sink. Payload capture obeys the per-agent IO policy — check it before persisting new payload kinds.

Tests

uv run --package theygent-control-plane pytest          # fast suite (ephemeral Postgres, fake gateway/MCP)
uv run --package theygent-control-plane pytest -m integration   # opt-in; needs real engines

Fixtures live in tests/_*.py (ephemeral DB, fake inference gateway, fake MCP servers, IR builders); tests/AGENTS.md maps which file covers what. Durable tests run the real DBOS runtime against the ephemeral DB. Every new endpoint, node type, or migration lands with coverage here — migrations are exercised by tests/test_migrations.py.