Skip to content

Latest commit

 

History

History
297 lines (234 loc) · 14.3 KB

File metadata and controls

297 lines (234 loc) · 14.3 KB

Commander — the natural-language command layer

Commander is where you talk to your AI workforce. You type intent in plain English, @mention a specialist to dispatch, and watch that agent's LangGraph execute node-by-node — with live state, human approval gates, and execution replay. Each agent is its own StateGraph; Commander is the surface that dispatches them, streams their reasoning, and pauses them at your approval before anything consequential runs.

Mission Control — the agent team, the NLP dispatch box, and example commands Mission Control: the agent roster on the left, the natural-language dispatch box, and example commands. Type intent, @mention a specialist, and the run renders live.


Where Commander sits in the loop

CortexObserver runs on one operating principle:

Humans define policies & budgets → Farms enforce them → Agents work within the constraints → Results flow back as live truth → Humans observe and adjust → repeat.

Commander owns the first and last legs of that circle. It turns human intent into a dispatched agent run, and it renders the agent's work back as something you can see, inspect, and approve. The Farms (tools, models, memory, knowledge, budgets) bound what an agent can do; Commander is how you direct it and how you watch it happen.


Dispatch — from a sentence to a running graph

You address an agent by handle. @mention a specialist to route to it; with no mention, the request defaults to @chat, the general-purpose assistant.

@allen deploy this CDK repo to dev
@amy   build a churn model from this dataset
@charles analyze NVDA
is it safe to take ibuprofen with my prescription?   →  defaults to @chat

A dispatch is a single POST that returns immediately with a Command record, then streams the rest over WebSocket:

POST /api/commander/agents/{key}/dispatch  { command: "Deploy Hello World to dev" }
        │
        ▼
dispatch_agent_task()                          (cortex/api/routes/commander.py)
  ├─ AgentRegistry.get_by_nickname("allen")    resolve the handle → live agent
  ├─ load bound Skills (Skills Store) + tool_bindings (Identity Store)
  ├─ create Command record (status="running"); db.commit()
  ├─ asyncio.create_task(_run_agent_graph())   ← the work runs in the background
  └─ return Command (status="running") immediately   ← the UI never blocks

The background task builds an AgentExecutionContext, calls agent.execute(context), and the agent's compiled graph runs node → node → END. Every node persists ExecutionStep rows and publishes a WebSocket event as it goes.

Sync vs async dispatch

The dispatch is always non-blocking — the HTTP call returns a running Command the instant the background task is scheduled, so Mission Control never waits on a long-running graph. What differs is how the agent's graph runs internally:

Deterministic nodes LLM / tool nodes
What Waiting, polling, capturing, looping Analysis, planning, summarizing, tool calls
How Plain async functions, no model call call_llm() / MCPFarm executor (awaited)
Example (@allen) _wait_for_deployment polls CF status every 15s analyze_request, create_plan, summarize

The Commander principle: LLM where needed, deterministic where possible. Analysis and planning nodes reason with a model; waiting, polling, and capturing are deterministic. The graph structure itself enforces procedure — you can't deploy before checking dependencies, because the edges don't allow it.

Streaming back to Mission Control

Backend (graph)                          Frontend (Mission Control)
───────────────                          ──────────────────────────
                                         dispatch → command_id, "Running…" badge
                                         3s polling fallback armed

commander.workflow_step      ───►        poll GET /commands/{id}, update steps
commander.deploy_polling     ───►        (silently handled)
commander.workflow_completed ───►        final card state + refresh metrics
  / workflow_failed

Events flow Redis pub/sub → WebSocket hub → browser → React handler (cortex/realtime/event_bus.py, ws_hub.py). The 3-second poll is a fallback; the WebSocket is the live path.


The Graph Workspace — watch the agent think

Mission Control dispatches; the Graph Workspace (/dashboard/commander/workspace) lets you see inside a run. It renders any agent's compiled StateGraph as an interactive React Flow DAG (@xyflow/react + @dagrejs/dagre) and animates it live as the graph executes.

The Graph Workspace rendering @amy's 23-node ML pipeline graph The Graph Workspace rendering @amy's 23-node LangGraph — profile, frame, train, critics, deployment judge. Nodes pulse while running and glow green on completion.

Capability What it does
Interactive DAG Click, zoom, pan, minimap over the agent's graph
Live execution Nodes pulse while running, green glow on complete
State inspection Checkpoint state viewer — command result, execution steps (JSON tree)
Execution replay Step-by-step timeline with Play / Pause / Step / Reset
Per-node token usage Input/output tokens, cost, and model per node — for prompt tuning
Node config editing System prompt, model, temperature, tool bindings per node
HITL gates Amber approve/reject node for paused approval points
Subgraphs Collapsible nested-graph nodes for agent-to-agent delegation

It's generic — every agent gets it for free

graph_serializer.py (cortex/agents/base/) introspects any compiled StateGraph into React Flow JSON with no agent-specific logic. Register a new agent and it appears in the Workspace automatically. The serializer:

  1. enumerates graph.nodes.keys() (excluding __start__ / __end__)
  2. reads direct edges from graph.builder.edges
  3. reads conditional edges (with labels) from graph.builder.branches
  4. unwraps node functions from PregelNode wrappers (bound.afunc / bound.func) for docstrings
  5. computes layout via topological layer assignment

Per-node token accounting

After a run completes, the Workspace fetches token usage from LLMInvocationLog, filtered by agent_id and the command's time window. The per-node breakdown is possible because every call_llm() tags its metadata with {"agent_id": "...", "node": "..."} — so you can see exactly which node spent which tokens, and tune the expensive ones.


Missions — a dispatch with a record

Every dispatch becomes a Command row (cortex/models/commander.py) — the durable record of a mission. It carries the command text, the resolved agent, status (running → completed / failed, or the HITL states below), the ExecutionStep trail, and the final result. Chained runs (a HITL resume) link back via parent_command_id (added in migration 040), so a paused-then-resumed mission is two linked Command rows, not one mutated row. This is what lets Mission Control render history, the Workspace replay a past run, and audits trace what an agent actually did back to the intent that triggered it.


Human-in-the-loop — the approval gate

This is the heart of governance at the Commander layer. Consequential actions pause for a human before they run. A deploy, a delete, a safety-gated medical answer — the agent reasons up to the decision point, then stops and waits for you.

@allen's 7-node graph showing the amber human-approval gate @allen's 7-node graph in the Workspace. The amber node is the human-approval (HITL) gate — the run halts there until you approve or reject.

How a pause works — interrupt() + Command(resume=…)

CortexObserver uses LangGraph's native interrupt mechanism, persisted through a PostgresSaver checkpointer. The gate node calls interrupt(); the graph suspends; its full state is checkpointed under thread_id = Command.id. Nothing downstream runs until a human resumes it.

from langgraph.types import interrupt

async def clarification_gate_node(state):
    # ... LLM decides whether the action needs human sign-off ...

    # Emit a workspace step BEFORE interrupt() so the timeline shows the
    # gate fired — GraphInterrupt short-circuits on_chain_end, and the
    # workspace would otherwise end silently.
    await BaseAgent.emit_step_from_state(
        state,
        step_name=f"Clarification: {len(questions)} question(s)",
        action_type="hitl_waiting",
        status="running",
        output={"questions": questions, "reason": reason},
    )

    response = interrupt({
        "type": "myagent.clarification",
        "questions": questions,
        "reason": reason,
        "agent_key": AGENT_KEY,
    })
    # On resume, `response` is whatever was passed to Command(resume=...).

The pause is surfaced through _extract_result, which detects __interrupt__ in the final state and returns a result with paused=True (a pause is not a failure). BaseAgent._execute_with_step_tracking already probes the checkpointer snapshot for pending interrupts before this runs.

The approve / reject flow

User: "is it safe to…"
   │  POST /dispatch  ──►  Command #A  (status=running)
   ▼
agent.execute() → parser → triage → gate ── interrupt() ──► PAUSE
   │
   ▼  Command #A: status=awaiting_clarification, result.paused=True
   │  checkpoint persisted under thread_id = Command #A.id
   ▼
Chat: Turn 1 — amber "Awaiting reply" badge + the agent's questions
      (immutable snapshot — never mutated again)
══════════════════════════════════════════════════════════════════
User: "56, no conditions"
   │  POST /dispatch  ──►  auto-detect finds Command #A
   │     (status="awaiting_clarification", same user + agent, ≤30 min)
   ▼
_resume_paused_command()
   ├─ mark Command #A status="resumed"   (result UNTOUCHED — Turn 1 stays valid;
   │                                       auto-detect won't refind it)
   ├─ agent.resume(ctx, …)  →  Command(resume=…) on the SAME thread_id
   │     graph resumes from the gate → composer → critics → finalizer
   └─ INSERT Command #B  (parent_command_id = Command #A.id, status=completed)
   ▼
Chat: Turn 3 — green "Success" badge + the final answer

Two design rules make this correct and worth calling out:

  • Resume INSERTs a new Command row (Option A), it never mutates the paused one. Turn 1's chat snapshot and the original checkpoint thread stay intact; the resumed turn is Command #B, linked by parent_command_id.
  • The thread_id never changes — it's the checkpointer key (Command #A.id). Resume runs Command(resume=…) against that same thread, so the graph continues exactly where interrupt() suspended it.

For a deploy-style gate (@allen), approve/reject is explicit rather than auto-detected: the plan is presented, and a human resolves it via POST /api/commander/commands/{id}/approve (resume into execute_plan) or /reject (transition to summarize with the plan rejected).

The WebSocket events that drive the UI

The agent interface (cortex/agents/base/agent_interface.py) emits graph lifecycle events the Workspace listens for:

Event Fires when UI effect
commander.graph.node_start a node begins node pulses (running)
commander.graph.node_end a node completes node glows green
commander.graph.hitl_waiting a gate calls interrupt() node turns amber, approve/reject surfaced
commander.workflow_step each tool execution step appended to the timeline
commander.workflow_completed run finishes (carries resumed=true on a resume) final card; the resume flag tells the WS handler to skip Turn 1's snapshot refresh

commander.graph.hitl_waiting is the one that turns the amber gate live in the Workspace and the amber "Awaiting reply" badge in Mission Control's chat.

One non-negotiable: the checkpointer must be attached

If interrupt() raises but the graph "succeeds" silently, or aget_state() returns an empty snapshot, the compiled graph has no checkpointer attached. BaseAgent.initialize() must set self._graph.checkpointer = checkpointer (an AsyncPostgresSaver over an AsyncConnectionPool, from cortex/commander/checkpointer.py). Without it, there's nothing to pause into and nothing to resume from. @bishop is the reference HITL agent; new safety-gated agents should follow its shape and reuse BaseAgent.resume() and the agent-agnostic _resume_paused_command rather than re-implementing either.


File reference

Component Path
Commander dispatch + resume backend/src/cortex/api/routes/commander.py
Commander models (Command, ExecutionStep) backend/src/cortex/models/commander.py
Base agent interface (execute, resume, HITL probe) backend/src/cortex/agents/base/agent_interface.py
Checkpointer (AsyncPostgresSaver) backend/src/cortex/commander/checkpointer.py
Graph serializer (StateGraph → React Flow) backend/src/cortex/agents/base/graph_serializer.py
Event bus + WebSocket hub backend/src/cortex/realtime/event_bus.py, ws_hub.py
@bishop — reference HITL agent backend/src/cortex/agents/bishop/
Mission Control UI frontend/src/components/commander/mission-control-layout.tsx
Graph Workspace UI (14 components) frontend/src/components/commander/graph/
Workspace store (Zustand) frontend/src/lib/workspace-store.ts

Related

  • Architecture — the circular loop, Humans/Agents/Farms, A.T.O.M, time as the universal index
  • Agents — the repeatable agent pattern and the full roster, with the Trading Desk & ML Studio deep dives
  • Governance — Policies/Procedures/Standards, Skills Store, Identity Store, risk budgets & approval gates