Human-in-the-Loop

Goal: follow a real interrupt → answer → resume cycle with today’s code.

Target shape

Mission
  → Execution
    → Thread
      → Run #1  → tool ask_human → interrupt
      → Checkpoint (persisted)
      → Interruption (PENDING)
      → human answer
      → resume (Celery transport)
      → Run #2  → SUCCESS (or interrupt again)

How a tool interrupts

The catalog tool ask_human (agency.tools.human.ask_human) calls LangGraph interrupt({...}):

# agency.tools.human
return interrupt({
    "type": "ask_human",
    "message": message,
    "data": data or {},
})

Attach that tool to an Agent (tool_ids), write a Mission prompt that requires asking the user, and dispatch as in Create and execute a Mission.

ThreadRunner.check_interrupt sees __interrupt__ in the graph output, marks the current Run interrupted, and creates an Interruption row linked to the latest Checkpoint for that Run (when the interrupt id is new).

Persistence

Graph state needed for continuation is stored via CheckpointSaver (Django-backed checkpoints).
Interruption holds question, answer, status, FK to checkpoint / agent / user.
The Celery worker that ran start_mission finishes after the interrupt. Nothing keeps a worker blocked waiting for the human.

Why Run #1 stays INTERRUPTED

A Run is one technical attempt. When it hits HITL, its terminal outcome for that attempt is interrupted. Resume does not rewrite Run #1 to success.

ThreadRunner.resume creates a new Run (status resuming) on the same Thread (same thread_id) and invokes Command(resume=interruption.answer).

Delivering the human answer

Any path that saves Interruption.answer (and moves status toward answered) can trigger resume via signals (resume_mission.delay).

Current entry points:

Mission Control (execution window) — WebSocket on /channel/: UserConsumer.handle_interruption_answer
REST — PATCH /api/interruption/<uuid>/ with an answer payload (admin UI uses this)
Admin — answer form; optional manual resume action

Wire detail: Connections.

After answer

Interruption marked answered (status answered).
Celery task agency.resume_mission runs ThreadRunner().resume(interruption).
New Run advances the same Thread from the persisted checkpoint.
Inspect history: GET /api/execution/<id>/ (nested threads/runs, interruptions, messages) or Django admin.

DB / Celery / WebSocket roles

Layer	Role
PostgreSQL	Authoritative lifecycle and checkpoints
Celery	Transport to start and resume work
WebSocket	Realtime projection to the UI (lossy OK)

If the socket drops, rehydrate from REST/DB. Do not treat WS frames as the system of record.

Current behaviour note (`resume_mission`)

After resume, resume_mission currently sets execution.status = SUCCESS and stores execution.result, even if the new Run interrupted again or errored. That is current code behaviour, not an ideal lifecycle rule. Documented here so operators are not surprised; improving it is a code change outside this tutorial.

ASCII

Execution
└── Thread
    ├── Run #1  INTERRUPTED
    │     └── Checkpoint ──► Interruption PENDING → ANSWERED
    └── Run #2  SUCCESS   (same thread_id, Command(resume=…))