merge: preamble + heartbeat + ask verb

Co-authored-by: Orca <help@stably.ai>

src/cli/handlers/orchestration.ts

@@ -233,6 +233,48 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
     )
   },
 
+  'orchestration ask': async ({ flags, client, cwd, json }) => {
+    const from = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'from')
+    const timeoutMs = flags.has('timeout-ms') ? Number(flags.get('timeout-ms')) : 600_000
+    const result = await client.call<{
+      answer: string | null
+      messageId: string | null
+      threadId: string
+      timedOut: boolean
+    }>(
+      'orchestration.ask',
+      {
+        to: getRequiredStringFlag(flags, 'to'),
+        question: getRequiredStringFlag(flags, 'question'),
+        options: getOptionalStringFlag(flags, 'options'),
+        timeoutMs: flags.has('timeout-ms') ? Number(flags.get('timeout-ms')) : undefined,
+        from
+      },
+      // Why: the runtime's `waitForMessage` can block up to `timeoutMs`, but
+      // the RPC transport has its own 60s default timeout that would fire
+      // first. Extend the per-call timeout by a small grace window so the
+      // RPC doesn't abort before the runtime's internal timeout resolves.
+      { timeoutMs: timeoutMs + 5_000 }
+    )
+    // Why: deliberate bypass of `printResult`. `--json` on `ask` emits a
+    // single-line bare JSON object (no RPC envelope, no multi-line pretty-
+    // print) so workers can pipe `orca orchestration ask … --json | jq -r
+    // .answer` without reaching into a `result` envelope. This diverges from
+    // every other orchestration verb; called out in the commit message and
+    // guarded by a unit test in orchestration.test.ts.
+    if (json) {
+      console.log(JSON.stringify(result.result))
+    } else if (result.result.answer !== null) {
+      console.log(result.result.answer)
+    }
+    if (result.result.timedOut) {
+      if (!json) {
+        console.error(`ask timeout after ${timeoutMs}ms (thread ${result.result.threadId})`)
+      }
+      process.exitCode = 1
+    }
+  },
+
   'orchestration dispatch-show': async ({ flags, client, json }) => {
     const result = await client.call<{
       dispatch: { id: string; task_id: string; status: string } | null

src/cli/specs/orchestration.ts

@@ -86,6 +86,13 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
     usage: 'orca orchestration dispatch-show --task <task_id> [--json]',
     allowedFlags: [...GLOBAL_FLAGS, 'task']
   },
+  {
+    path: ['orchestration', 'ask'],
+    summary: 'Ask the coordinator a question and block until answered',
+    usage:
+      'orca orchestration ask --to <handle> --question <text> [--options <csv>] [--timeout-ms <n>] [--from <handle>] [--json]',
+    allowedFlags: [...GLOBAL_FLAGS, 'to', 'question', 'options', 'timeout-ms', 'from']
+  },
   {
     path: ['orchestration', 'run'],
     summary: 'Start the coordinator loop',

src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap

@@ -0,0 +1,83 @@
+// Vitest Snapshot v1, https://vitest.dev/guide/snapshot.html
+
+exports[`buildDispatchPreamble > renders a stable snapshot of the full preamble 1`] = `
+"You are working inside Orca, a multi-agent IDE. You are a dispatched worker.
+Your coordinator's terminal handle is: term_COORD
+Your task ID is: task_SNAP
+
+You talk to the coordinator only through the CLI commands below. Do not use
+Slack, GitHub comments, or any other channel to reach a human during the run.
+
+=== CLI COMMANDS ===
+
+  # Report task completion (REQUIRED when done — even on failure).
+  #
+  # RULE: --body must be a 3-sentence executive summary (what you did,
+  # what you found, what's left). Never send an empty body; the coordinator
+  # reads the body first and only opens artifacts if it needs more detail.
+  # If you produced a long-form artifact, include its path as
+  # payload.reportPath so the coordinator can find it without a file search.
+  #
+  # RULE: send worker_done exactly once. Failure is still a worker_done
+  # with subject like "Failed: <reason>" — never silently exit.
+  orca orchestration send --to term_COORD \\
+    --type worker_done --subject "<short status>" \\
+    --body "<3-sentence summary: what you did, what you found, what's left>" \\
+    --payload '{"taskId":"task_SNAP","filesModified":["path/a","path/b"],"reportPath":"<optional: path to the full artifact>"}'
+
+  # BEHAVIOR RULE: send a heartbeat every 5 minutes
+  # while actively working on the task. The coordinator uses this to
+  # distinguish "still thinking" from "hung / crashed." Skip heartbeats only
+  # while blocked inside \`check --wait\` or \`ask\` — those calls are
+  # themselves liveness signals.
+  #
+  # Include BOTH taskId and dispatchId in the payload: the coordinator
+  # attributes the heartbeat to the specific dispatch context, not just
+  # the task, so a straggler heartbeat from a previously-failed dispatch
+  # cannot mask a hung retry.
+  orca orchestration send --to term_COORD \\
+    --type heartbeat --subject "alive" \\
+    --payload '{"taskId":"task_SNAP","dispatchId":"ctx_SNAP","phase":"<short: investigating|implementing|reviewing|waiting>"}'
+
+  # Ask the coordinator a question and block until it answers.
+  #
+  # BEHAVIOR RULE #1 (MUST NOT VIOLATE):
+  # NEVER use AskUserQuestion; use \`orca orchestration ask\` or send
+  # --type decision_gate. AskUserQuestion opens a local TUI prompt that the
+  # coordinator cannot see and cannot answer — your session will hang forever
+  # waiting on a human. Every interactive question goes through \`ask\` below.
+  #
+  # The \`ask\` verb is a thin wrapper: it sends a decision_gate message and
+  # blocks on \`check --wait\` until the coordinator replies, then prints the
+  # reply body. Use it anywhere you would otherwise have reached for
+  # AskUserQuestion.
+  orca orchestration ask --to term_COORD \\
+    --question "<your question>" \\
+    --options "<optional,comma,separated>" \\
+    --timeout-ms 600000
+
+  # Escalate a blocker or failure (pre-completion, when you need the
+  # coordinator to do something before you can continue):
+  orca orchestration send --to term_COORD \\
+    --type escalation --subject "Blocked: <reason>" \\
+    --body "<details>" \\
+    --payload '{"taskId":"task_SNAP"}'
+
+  # Check for messages from the coordinator:
+  orca orchestration check
+
+=== AFTER YOU SEND worker_done ===
+
+Keep the shell session open for a grace period (10 minutes) in case the
+coordinator sends a follow-up or re-dispatches you. Poll with
+\`orca orchestration check\` every 2 minutes during that window. If no
+follow-up arrives, you may exit after the grace period — the coordinator
+will not expect further output from you.
+
+If the coordinator re-dispatches you (you will receive a fresh preamble +
+TASK block), reset your polling and start the new task. Do not respond
+to the previous task's follow-ups after a re-dispatch.
+
+=== TASK ===
+TASK_BODY"
+`;

src/main/runtime/orchestration/coordinator.test.ts

@@ -383,6 +383,88 @@ describe('Coordinator', () => {
     expect(result.status).toBe('completed')
   })
 
+  it('logs a stale warning for dispatched rows past the threshold and does not auto-fail', async () => {
+    db = new OrchestrationDb(':memory:')
+    const runtime = createMockRuntime()
+    // No terminals available so dispatchReadyTasks creates one and we can
+    // drive the stale-scan deterministically via SQL backdating.
+    const task = db.createTask({ spec: 'work' })
+    const ctx = db.createDispatchContext(task.id, 'term_stale')
+
+    // Backdate dispatched_at and last_heartbeat_at beyond the 10-min threshold
+    // so getStaleDispatches returns this row on the first tick.
+    const sqlite = (
+      db as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => void } } }
+    ).db
+    const iso = (ms: number) => new Date(Date.now() - ms).toISOString()
+    sqlite
+      .prepare('UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?')
+      .run(iso(60 * 60 * 1000), iso(30 * 60 * 1000), ctx.id)
+
+    const logs: string[] = []
+    const coordinator = new Coordinator(db, runtime, {
+      spec: 'go',
+      coordinatorHandle: 'coord',
+      pollIntervalMs: 20,
+      onLog: (m) => logs.push(m)
+    })
+
+    // Drive one tick then stop — we only need the stale warning to have fired.
+    const runPromise = coordinator.run()
+    await new Promise((r) => {
+      setTimeout(r, 80)
+    })
+    coordinator.stop()
+    await runPromise
+
+    expect(logs.some((l) => /has not sent a heartbeat/.test(l) && l.includes(task.id))).toBe(true)
+    // Task status must NOT have been auto-failed — logging only.
+    expect(db.getTask(task.id)?.status).toBe('dispatched')
+  })
+
+  it('records heartbeat by dispatchId on worker heartbeat messages', async () => {
+    db = new OrchestrationDb(':memory:')
+    const runtime = createMockRuntime()
+    runtime.terminals = [{ handle: 'term_a', worktreeId: 'wt1', connected: true, writable: true }]
+
+    const task = db.createTask({ spec: 'work' })
+    const ctx = db.createDispatchContext(task.id, 'term_a')
+
+    const coordinator = new Coordinator(db, runtime, {
+      spec: 'go',
+      coordinatorHandle: 'coord',
+      pollIntervalMs: 20
+    })
+
+    const runPromise = coordinator.run()
+
+    db.insertMessage({
+      from: 'term_a',
+      to: 'coord',
+      subject: 'alive',
+      type: 'heartbeat',
+      payload: JSON.stringify({ taskId: task.id, dispatchId: ctx.id, phase: 'implementing' })
+    })
+
+    await new Promise((r) => {
+      setTimeout(r, 80)
+    })
+
+    expect(db.getDispatchContext(task.id)?.last_heartbeat_at).toBeTruthy()
+
+    // Complete the task so the coordinator run finishes cleanly.
+    db.insertMessage({
+      from: 'term_a',
+      to: 'coord',
+      subject: 'Done',
+      type: 'worker_done',
+      payload: JSON.stringify({ taskId: task.id })
+    })
+
+    const result = await runPromise
+    expect(result.status).toBe('completed')
+  })
+
   it('can be stopped', async () => {
     db = new OrchestrationDb(':memory:')
     const runtime = createMockRuntime()

src/main/runtime/orchestration/coordinator.ts

@@ -80,6 +80,13 @@ type CoordinatorState = {
 const DEFAULT_POLL_MS = 2000
 const MAX_CONCURRENT_DEFAULT = 4
 
+// Why: 10 min matches the preamble's documented heartbeat cadence (5 min) ×
+// 2, so a single missed heartbeat is the earliest a dispatch can look stale.
+// Keeping this in one place (not a per-call arg) ensures the preamble copy
+// and the detector logic stay aligned; moving it to a config would multiply
+// the places this constant must be kept in sync.
+const HUNG_THRESHOLD_MS = 10 * 60 * 1000
+
 export class Coordinator {
   private db: OrchestrationDb
   private runtime: CoordinatorRuntime
@@ -212,10 +219,27 @@ export class Coordinator {
     this.processMessages()
     this.processEscalations()
     this.processDecisionGates()
+    this.warnStaleDispatches()
     await this.dispatchReadyTasks()
     return this.checkConvergence()
   }
 
+  // Why: emit a single warning per stale dispatch per tick. This intentionally
+  // does NOT auto-fail the dispatch — the false-positive cost (a slow worker
+  // producing correct output) is higher than the false-negative cost (a hung
+  // worker keeps its terminal slot until a human notices). Auto-fail policy
+  // is a separate decision documented in R6 of DESIGN_DOC_PREAMBLE_FIX.md.
+  private warnStaleDispatches(): void {
+    const thresholdIso = new Date(Date.now() - HUNG_THRESHOLD_MS).toISOString()
+    const stale = this.db.getStaleDispatches(thresholdIso)
+    for (const ctx of stale) {
+      const minutes = Math.round(HUNG_THRESHOLD_MS / 60000)
+      this.opts.onLog(
+        `Warning: worker ${ctx.assignee_handle ?? '<unknown>'} on task ${ctx.task_id} has not sent a heartbeat in ~${minutes} min (dispatch ${ctx.id})`
+      )
+    }
+  }
+
   private processMessages(): void {
     const messages = this.db.getUnreadMessages(this.opts.coordinatorHandle)
     if (messages.length === 0) {
@@ -233,6 +257,9 @@ export class Coordinator {
         case 'decision_gate':
           this.handleDecisionGateMessage(msg)
           break
+        case 'heartbeat':
+          this.handleHeartbeat(msg)
+          break
         case 'status':
           this.opts.onLog(`Status from ${msg.from_handle}: ${msg.subject}`)
           break
@@ -244,6 +271,35 @@ export class Coordinator {
     this.db.markAsRead(messages.map((m) => m.id))
   }
 
+  // Why: attribute heartbeats to the specific dispatchId, not a
+  // (taskId, from_handle) lookup. A task that gets retried after a failed
+  // dispatch has multiple rows in dispatch_contexts — a late heartbeat from
+  // the previous (failed) assignee arriving while the new dispatch is active
+  // would falsely bump the new row's last_heartbeat_at if we resolved by
+  // "latest dispatch for this task" (§5.3.4). If the worker drops dispatchId
+  // from the payload, log-and-skip is the preferred failure mode: the stale
+  // detector will correctly flag the dispatch as hung because nothing
+  // refreshed last_heartbeat_at.
+  private handleHeartbeat(msg: MessageRow): void {
+    if (!msg.payload) {
+      this.opts.onLog(`Heartbeat from ${msg.from_handle} missing payload; ignored`)
+      return
+    }
+    let payload: { dispatchId?: unknown } = {}
+    try {
+      payload = JSON.parse(msg.payload)
+    } catch {
+      this.opts.onLog(`Heartbeat from ${msg.from_handle} has invalid JSON payload; ignored`)
+      return
+    }
+    const dispatchId = payload.dispatchId
+    if (typeof dispatchId !== 'string' || dispatchId.length === 0) {
+      this.opts.onLog(`Heartbeat from ${msg.from_handle} missing dispatchId; ignored`)
+      return
+    }
+    this.db.recordHeartbeat(dispatchId, msg.created_at)
+  }
+
   private handleWorkerDone(msg: MessageRow): void {
     this.opts.onLog(`Worker done: ${msg.from_handle} — ${msg.subject}`)
 
@@ -473,6 +529,11 @@ export class Coordinator {
     // the worker would otherwise read as part of its instructions).
     const preamble = buildDispatchPreamble({
       taskId: task.id,
+      dispatchId: dispatch.id,
+      // Why (§3.4, stale-base PR): use `strippedSpec` not `task.spec` so the
+      // `allow-stale-base: true` line isn't rendered into the worker's
+      // --- TASK --- block (worker would otherwise treat the infra flag as
+      // part of its instructions).
       taskSpec: strippedSpec,
       coordinatorHandle: this.opts.coordinatorHandle,
       devMode: process.env.ORCA_USER_DATA_PATH?.includes('orca-dev'),

src/main/runtime/orchestration/db.test.ts

@@ -1,4 +1,8 @@
 /* eslint-disable max-lines -- Why: DB tests cover messages, tasks, dispatch contexts, decision gates, coordinator runs, and lifecycle in one suite to share the createDb() helper and afterEach cleanup. */
+import Database from 'better-sqlite3'
+import { mkdtempSync, rmSync } from 'fs'
+import { tmpdir } from 'os'
+import { join } from 'path'
 import { afterEach, describe, expect, it } from 'vitest'
 import { OrchestrationDb } from './db'
 import type { MessageType } from './db'
@@ -454,4 +458,278 @@ describe('OrchestrationDb', () => {
       expect(d.listTasks()).toHaveLength(0)
     })
   })
+
+  describe('heartbeat + thread helpers (fresh schema)', () => {
+    it('insertMessage accepts type = heartbeat', () => {
+      const d = createDb()
+      const msg = d.insertMessage({
+        from: 'worker',
+        to: 'coord',
+        subject: 'alive',
+        type: 'heartbeat',
+        payload: JSON.stringify({ taskId: 'task_x', dispatchId: 'ctx_x' })
+      })
+      expect(msg.type).toBe('heartbeat')
+    })
+
+    it('recordHeartbeat updates last_heartbeat_at on dispatched rows', () => {
+      const d = createDb()
+      const task = d.createTask({ spec: 'work' })
+      const ctx = d.createDispatchContext(task.id, 'term_a')
+
+      d.recordHeartbeat(ctx.id, '2026-05-04T00:00:00.000Z')
+      const after = d.getDispatchContext(task.id)
+      expect(after?.last_heartbeat_at).toBe('2026-05-04T00:00:00.000Z')
+    })
+
+    it('recordHeartbeat is a no-op for completed rows (straggler ignored)', () => {
+      const d = createDb()
+      const task = d.createTask({ spec: 'work' })
+      const ctx = d.createDispatchContext(task.id, 'term_a')
+      d.completeDispatch(ctx.id)
+
+      d.recordHeartbeat(ctx.id, '2026-05-04T00:00:00.000Z')
+      const after = d.getDispatchContext(task.id)
+      expect(after?.last_heartbeat_at).toBeNull()
+    })
+
+    it('getStaleDispatches returns only dispatched rows past the grace window', () => {
+      const d = createDb()
+      // Fixture: four rows, SQL-backdated timestamps (no fake clock):
+      //  (a) dispatched, heartbeated 5 min ago → not stale
+      //  (b) dispatched, heartbeated 12 min ago → STALE (expected result)
+      //  (c) dispatched, never heartbeated, dispatched 30s ago → not stale (grace)
+      //  (d) completed, heartbeated 30 min ago → not stale (status filter)
+      const taskA = d.createTask({ spec: 'a' })
+      const taskB = d.createTask({ spec: 'b' })
+      const taskC = d.createTask({ spec: 'c' })
+      const taskD = d.createTask({ spec: 'd' })
+      const ctxA = d.createDispatchContext(taskA.id, 'term_a')
+      const ctxB = d.createDispatchContext(taskB.id, 'term_b')
+      const ctxC = d.createDispatchContext(taskC.id, 'term_c')
+      const ctxD = d.createDispatchContext(taskD.id, 'term_d')
+      d.completeDispatch(ctxD.id)
+
+      const now = Date.now()
+      const iso = (ms: number) => new Date(now - ms).toISOString()
+
+      // Backdate dispatched_at for a, b, d to long ago so the grace doesn't
+      // shield them. c keeps its default (≈now).
+      const sqlite = (d as unknown as { db: Database.Database }).db
+      sqlite
+        .prepare(
+          'UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?'
+        )
+        .run(iso(60 * 60 * 1000), iso(5 * 60 * 1000), ctxA.id)
+      sqlite
+        .prepare(
+          'UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?'
+        )
+        .run(iso(60 * 60 * 1000), iso(12 * 60 * 1000), ctxB.id)
+      sqlite
+        .prepare('UPDATE dispatch_contexts SET dispatched_at = ? WHERE id = ?')
+        .run(iso(30_000), ctxC.id)
+      sqlite
+        .prepare(
+          'UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?'
+        )
+        .run(iso(60 * 60 * 1000), iso(30 * 60 * 1000), ctxD.id)
+
+      const stale = d.getStaleDispatches(iso(10 * 60 * 1000))
+      expect(stale.map((s) => s.id)).toEqual([ctxB.id])
+    })
+
+    it('getThreadMessagesFor returns only same-thread replies to a handle', () => {
+      const d = createDb()
+      const outbound = d.insertMessage({
+        from: 'worker',
+        to: 'coord',
+        subject: 'Question',
+        type: 'decision_gate',
+        body: 'yes or no?'
+      })
+      // Reply in the same thread addressed to the worker
+      const reply = d.insertMessage({
+        from: 'coord',
+        to: 'worker',
+        subject: 'Re: Question',
+        body: 'yes',
+        threadId: outbound.id
+      })
+      // Distractor: different thread, same recipient
+      d.insertMessage({
+        from: 'coord',
+        to: 'worker',
+        subject: 'other',
+        body: 'unrelated',
+        threadId: 'thread_other'
+      })
+      // Distractor: same thread but not addressed to worker
+      d.insertMessage({
+        from: 'coord',
+        to: 'someone_else',
+        subject: 'cc',
+        body: 'not yours',
+        threadId: outbound.id
+      })
+
+      const replies = d.getThreadMessagesFor(outbound.id, 'worker', outbound.sequence)
+      expect(replies).toHaveLength(1)
+      expect(replies[0].id).toBe(reply.id)
+    })
+  })
+
+  describe('schema migration from v1 → v2', () => {
+    let dbPath: string
+    let tempDir: string
+
+    afterEach(() => {
+      if (tempDir) {
+        rmSync(tempDir, { recursive: true, force: true })
+      }
+    })
+
+    function createV1Snapshot(): string {
+      tempDir = mkdtempSync(join(tmpdir(), 'orca-db-migrate-'))
+      dbPath = join(tempDir, 'test.db')
+      const raw = new Database(dbPath)
+      // v1 schema: pre-heartbeat CHECK, no last_heartbeat_at column.
+      raw.exec(`
+        CREATE TABLE messages (
+          id            TEXT NOT NULL,
+          from_handle   TEXT NOT NULL,
+          to_handle     TEXT NOT NULL,
+          subject       TEXT NOT NULL,
+          body          TEXT NOT NULL DEFAULT '',
+          type          TEXT NOT NULL DEFAULT 'status'
+            CHECK(type IN (
+              'status', 'dispatch', 'worker_done', 'merge_ready',
+              'escalation', 'handoff', 'decision_gate'
+            )),
+          priority      TEXT NOT NULL DEFAULT 'normal'
+            CHECK(priority IN ('normal', 'high', 'urgent')),
+          thread_id     TEXT,
+          payload       TEXT,
+          read          INTEGER NOT NULL DEFAULT 0,
+          sequence      INTEGER PRIMARY KEY AUTOINCREMENT,
+          created_at    TEXT NOT NULL DEFAULT (datetime('now'))
+        );
+        CREATE UNIQUE INDEX idx_messages_id ON messages(id);
+        CREATE INDEX idx_inbox ON messages(to_handle, read);
+        CREATE INDEX idx_thread ON messages(thread_id);
+
+        CREATE TABLE tasks (
+          id TEXT PRIMARY KEY, parent_id TEXT, spec TEXT NOT NULL,
+          status TEXT NOT NULL DEFAULT 'pending'
+            CHECK(status IN ('pending','ready','dispatched','completed','failed','blocked')),
+          deps TEXT NOT NULL DEFAULT '[]', result TEXT,
+          created_at TEXT NOT NULL DEFAULT (datetime('now')),
+          completed_at TEXT
+        );
+
+        CREATE TABLE dispatch_contexts (
+          id TEXT PRIMARY KEY, task_id TEXT NOT NULL, assignee_handle TEXT,
+          status TEXT NOT NULL DEFAULT 'pending'
+            CHECK(status IN ('pending','dispatched','completed','failed','circuit_broken')),
+          failure_count INTEGER NOT NULL DEFAULT 0, last_failure TEXT,
+          dispatched_at TEXT, completed_at TEXT,
+          created_at TEXT NOT NULL DEFAULT (datetime('now'))
+        );
+
+        CREATE TABLE decision_gates (
+          id TEXT PRIMARY KEY, task_id TEXT NOT NULL, question TEXT NOT NULL,
+          options TEXT NOT NULL DEFAULT '[]',
+          status TEXT NOT NULL DEFAULT 'pending'
+            CHECK(status IN ('pending','resolved','timeout')),
+          resolution TEXT,
+          created_at TEXT NOT NULL DEFAULT (datetime('now')),
+          resolved_at TEXT
+        );
+
+        CREATE TABLE coordinator_runs (
+          id TEXT PRIMARY KEY, spec TEXT NOT NULL,
+          status TEXT NOT NULL DEFAULT 'idle'
+            CHECK(status IN ('idle','running','completed','failed')),
+          coordinator_handle TEXT NOT NULL,
+          poll_interval_ms INTEGER NOT NULL DEFAULT 2000,
+          created_at TEXT NOT NULL DEFAULT (datetime('now')),
+          completed_at TEXT
+        );
+      `)
+      // Seed a pre-existing v1 message so migration must preserve data.
+      raw
+        .prepare(
+          `INSERT INTO messages (id, from_handle, to_handle, subject, type) VALUES ('msg_v1', 'a', 'b', 'pre-migration', 'status')`
+        )
+        .run()
+      raw.pragma('user_version = 0')
+      raw.close()
+      return dbPath
+    }
+
+    it('migrates a v1 snapshot to v2, accepts heartbeat, preserves indexes', () => {
+      const path = createV1Snapshot()
+      const d = new OrchestrationDb(path)
+      db = d
+
+      // (a) INSERT type='heartbeat' now succeeds
+      expect(() =>
+        d.insertMessage({
+          from: 'w',
+          to: 'c',
+          subject: 'alive',
+          type: 'heartbeat',
+          payload: '{"taskId":"t","dispatchId":"ctx"}'
+        })
+      ).not.toThrow()
+
+      // (b) last_heartbeat_at column exists on dispatch_contexts
+      const task = d.createTask({ spec: 'work' })
+      const ctx = d.createDispatchContext(task.id, 'term_a')
+      d.recordHeartbeat(ctx.id, '2026-05-04T00:00:00.000Z')
+      expect(d.getDispatchContext(task.id)?.last_heartbeat_at).toBe('2026-05-04T00:00:00.000Z')
+
+      // (c) Indexes still attached to messages post-rebuild.
+      const sqlite = (d as unknown as { db: Database.Database }).db
+      const indexes = sqlite
+        .prepare(
+          `SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'messages' AND name NOT LIKE 'sqlite_%'`
+        )
+        .all() as { name: string }[]
+      const names = new Set(indexes.map((r) => r.name))
+      expect(names.has('idx_messages_id')).toBe(true)
+      expect(names.has('idx_inbox')).toBe(true)
+      expect(names.has('idx_thread')).toBe(true)
+
+      // v1 data preserved
+      expect(d.getMessageById('msg_v1')?.subject).toBe('pre-migration')
+    })
+
+    it('is idempotent: opening an already-migrated DB is a no-op', () => {
+      const path = createV1Snapshot()
+      const first = new OrchestrationDb(path)
+      first.insertMessage({
+        from: 'w',
+        to: 'c',
+        subject: 'alive',
+        type: 'heartbeat',
+        payload: '{}'
+      })
+      first.close()
+
+      const second = new OrchestrationDb(path)
+      db = second
+      expect(() =>
+        second.insertMessage({
+          from: 'w',
+          to: 'c',
+          subject: 'again',
+          type: 'heartbeat',
+          payload: '{}'
+        })
+      ).not.toThrow()
+      const inbox = second.getInbox(10)
+      expect(inbox.length).toBeGreaterThanOrEqual(2)
+    })
+  })
 })

src/main/runtime/orchestration/db.ts

@@ -33,6 +33,14 @@ function generateId(prefix: string): string {
   return `${prefix}_${randomBytes(6).toString('hex')}`
 }
 
+// Why: v1 → v2 added `'heartbeat'` to messages.type CHECK + `last_heartbeat_at`
+// column (preamble-hardening PR). v2 → v3 adds `delivered_at` column so
+// push-on-idle can distinguish queued-but-undelivered from user-acknowledged
+// messages without touching the `read` bit (check-wait PR). Bumping together
+// since both PRs ship in the same release — existing on-disk DBs must clear
+// both gaps in one atomic migration.
+const SCHEMA_VERSION = 3
+
 export class OrchestrationDb {
   private db: Database.Database
 
@@ -41,25 +49,7 @@ export class OrchestrationDb {
     this.db.pragma('journal_mode = WAL')
     this.db.pragma('synchronous = NORMAL')
     this.db.pragma('busy_timeout = 5000')
-    this.createTables()
-    // Why: migrations must run synchronously before any method can be called
-    // on this instance. The migration is idempotent (PRAGMA check) and
-    // hard-fails on error — a DB missing `delivered_at` would cause push-on-idle
-    // to throw inside markAsDelivered, which combined with the long-poll
-    // counter risks leaking slots. Better to crash startup than serve traffic
-    // on a broken schema. See design doc §3.2 / §3.3 / §7 risk #3.
-    this.runMigrations()
-  }
-
-  private runMigrations(): void {
-    const columns = this.db.prepare('PRAGMA table_info(messages)').all() as { name: string }[]
-    const hasDeliveredAt = columns.some((c) => c.name === 'delivered_at')
-    if (!hasDeliveredAt) {
-      // ALTER TABLE throws synchronously; any error other than
-      // "duplicate column name" (which is ruled out by the PRAGMA check above)
-      // must propagate so startup fails loudly.
-      this.db.exec('ALTER TABLE messages ADD COLUMN delivered_at TEXT')
-    }
+    this.migrate()
   }
 
   private createTables(): void {
@@ -73,7 +63,7 @@ export class OrchestrationDb {
         type          TEXT NOT NULL DEFAULT 'status'
           CHECK(type IN (
             'status', 'dispatch', 'worker_done', 'merge_ready',
-            'escalation', 'handoff', 'decision_gate'
+            'escalation', 'handoff', 'decision_gate', 'heartbeat'
           )),
         priority      TEXT NOT NULL DEFAULT 'normal'
           CHECK(priority IN ('normal', 'high', 'urgent')),
@@ -81,7 +71,8 @@ export class OrchestrationDb {
         payload       TEXT,
         read          INTEGER NOT NULL DEFAULT 0,
         sequence      INTEGER PRIMARY KEY AUTOINCREMENT,
-        created_at    TEXT NOT NULL DEFAULT (datetime('now'))
+        created_at    TEXT NOT NULL DEFAULT (datetime('now')),
+        delivered_at  TEXT
       );
 
       CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_id ON messages(id);
@@ -107,16 +98,17 @@ export class OrchestrationDb {
       CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_id);
 
       CREATE TABLE IF NOT EXISTS dispatch_contexts (
-        id              TEXT PRIMARY KEY,
-        task_id         TEXT NOT NULL,
-        assignee_handle TEXT,
-        status          TEXT NOT NULL DEFAULT 'pending'
+        id                  TEXT PRIMARY KEY,
+        task_id             TEXT NOT NULL,
+        assignee_handle     TEXT,
+        status              TEXT NOT NULL DEFAULT 'pending'
           CHECK(status IN ('pending', 'dispatched', 'completed', 'failed', 'circuit_broken')),
-        failure_count   INTEGER NOT NULL DEFAULT 0,
-        last_failure    TEXT,
-        dispatched_at   TEXT,
-        completed_at    TEXT,
-        created_at      TEXT NOT NULL DEFAULT (datetime('now'))
+        failure_count       INTEGER NOT NULL DEFAULT 0,
+        last_failure        TEXT,
+        dispatched_at       TEXT,
+        completed_at        TEXT,
+        created_at          TEXT NOT NULL DEFAULT (datetime('now')),
+        last_heartbeat_at   TEXT
       );
 
       CREATE INDEX IF NOT EXISTS idx_dispatch_task ON dispatch_contexts(task_id);
@@ -150,6 +142,115 @@ export class OrchestrationDb {
     `)
   }
 
+  // Why: `CREATE TABLE IF NOT EXISTS` is a no-op against an existing on-disk
+  // DB, so new schema shapes (added columns, widened CHECK constraints) do
+  // not reach an upgraded user unless we migrate explicitly. The transaction
+  // guarantees atomicity — a mid-migration crash leaves the DB at the prior
+  // version because `user_version` is bumped only on success. Idempotent
+  // re-invocation is a no-op (current >= SCHEMA_VERSION short-circuit).
+  private migrate(): void {
+    const current = this.db.pragma('user_version', { simple: true }) as number
+    if (current >= SCHEMA_VERSION) {
+      return
+    }
+
+    this.db.exec('BEGIN')
+    try {
+      // v1 → v2: add last_heartbeat_at column; widen messages.type CHECK to
+      // include 'heartbeat'. SQLite cannot ALTER a CHECK constraint, so we
+      // rebuild the messages table. We also include `delivered_at` in the
+      // rebuilt schema so DBs migrating from v1 pick up the v3 column in a
+      // single table-rewrite pass (avoids a second messages-rebuild later).
+      if (current < 2) {
+        if (!this.hasColumn('dispatch_contexts', 'last_heartbeat_at')) {
+          this.db.exec(`ALTER TABLE dispatch_contexts ADD COLUMN last_heartbeat_at TEXT`)
+        }
+
+        if (!this.messagesTypeCheckAllowsHeartbeat()) {
+          // Why — index list is not optional. createTables() already attached
+          // idx_messages_id / idx_inbox / idx_thread to the old messages table;
+          // DROP TABLE removes those indexes with it. CREATE INDEX IF NOT
+          // EXISTS in createTables() only runs on the next process startup,
+          // so skipping explicit recreation here would leave every
+          // getUnreadMessages / getMessageById call full-scanning for the
+          // rest of this process's lifetime — a silent O(N) perf regression.
+          // The three CREATE INDEX statements below mirror createTables()
+          // verbatim so the two definitions cannot drift.
+          this.db.exec(`
+            CREATE TABLE messages_new (
+              id            TEXT NOT NULL,
+              from_handle   TEXT NOT NULL,
+              to_handle     TEXT NOT NULL,
+              subject       TEXT NOT NULL,
+              body          TEXT NOT NULL DEFAULT '',
+              type          TEXT NOT NULL DEFAULT 'status'
+                CHECK(type IN (
+                  'status', 'dispatch', 'worker_done', 'merge_ready',
+                  'escalation', 'handoff', 'decision_gate', 'heartbeat'
+                )),
+              priority      TEXT NOT NULL DEFAULT 'normal'
+                CHECK(priority IN ('normal', 'high', 'urgent')),
+              thread_id     TEXT,
+              payload       TEXT,
+              read          INTEGER NOT NULL DEFAULT 0,
+              sequence      INTEGER PRIMARY KEY AUTOINCREMENT,
+              created_at    TEXT NOT NULL DEFAULT (datetime('now')),
+              delivered_at  TEXT
+            );
+            INSERT INTO messages_new (
+              id, from_handle, to_handle, subject, body, type, priority,
+              thread_id, payload, read, sequence, created_at
+            )
+            SELECT
+              id, from_handle, to_handle, subject, body, type, priority,
+              thread_id, payload, read, sequence, created_at
+            FROM messages;
+            DROP TABLE messages;
+            ALTER TABLE messages_new RENAME TO messages;
+
+            CREATE UNIQUE INDEX idx_messages_id ON messages(id);
+            CREATE INDEX idx_inbox ON messages(to_handle, read);
+            CREATE INDEX idx_thread ON messages(thread_id);
+          `)
+        }
+      }
+
+      // v2 → v3: add `delivered_at` column to messages. A DB that reached v2
+      // via the v1 → v2 rebuild above already has the column (we included
+      // it in messages_new); this handles DBs that were at v2 before this
+      // release shipped (preamble PR deployed standalone, then check-wait
+      // merged). ALTER TABLE is idempotent via the hasColumn probe — a
+      // duplicate-column error would abort the whole transaction.
+      if (current < 3) {
+        if (!this.hasColumn('messages', 'delivered_at')) {
+          this.db.exec(`ALTER TABLE messages ADD COLUMN delivered_at TEXT`)
+        }
+      }
+
+      this.db.pragma(`user_version = ${SCHEMA_VERSION}`)
+      this.db.exec('COMMIT')
+    } catch (err) {
+      this.db.exec('ROLLBACK')
+      throw err
+    }
+  }
+
+  private hasColumn(table: string, column: string): boolean {
+    const rows = this.db.pragma(`table_info(${table})`) as { name: string }[]
+    return rows.some((r) => r.name === column)
+  }
+
+  // Why: sqlite_master stores the original CREATE TABLE SQL including the
+  // CHECK clause. Inspecting that text is the cheapest reliable way to tell
+  // whether the pre-rebuild schema already knows about 'heartbeat' without
+  // needing a dedicated schema_meta row.
+  private messagesTypeCheckAllowsHeartbeat(): boolean {
+    const row = this.db
+      .prepare("SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'messages'")
+      .get() as { sql: string } | undefined
+    return !!row && row.sql.includes("'heartbeat'")
+  }
+
   // ── Messages ──
 
   insertMessage(msg: {
@@ -244,6 +345,25 @@ export class OrchestrationDb {
       .all(toHandle, limit) as MessageRow[]
   }
 
+  // Why: thread-scoped read for the `orchestration.ask` wait loop. Filtered
+  // by `to_handle` so a worker only sees replies addressed to it (not
+  // messages it sent), and ordered by `sequence` so the first post-ask
+  // reply is returned first. `afterSequence` lets the caller resume past an
+  // already-seen marker without re-reading the outbound ask itself. Uses
+  // the existing idx_thread index (see createTables) — no new index.
+  getThreadMessagesFor(threadId: string, toHandle: string, afterSequence?: number): MessageRow[] {
+    if (afterSequence !== undefined) {
+      return this.db
+        .prepare(
+          'SELECT * FROM messages WHERE thread_id = ? AND to_handle = ? AND sequence > ? ORDER BY sequence ASC'
+        )
+        .all(threadId, toHandle, afterSequence) as MessageRow[]
+    }
+    return this.db
+      .prepare('SELECT * FROM messages WHERE thread_id = ? AND to_handle = ? ORDER BY sequence ASC')
+      .all(threadId, toHandle) as MessageRow[]
+  }
+
   // ── Tasks ──
 
   createTask(task: { spec: string; deps?: string[]; parentId?: string }): TaskRow {
@@ -405,6 +525,40 @@ export class OrchestrationDb {
     return active ? this.failDispatch(active.id, error) : undefined
   }
 
+  // Why: only touch rows that are currently dispatched. A straggler heartbeat
+  // from a dispatch that already transitioned to `completed` / `failed` /
+  // `circuit_broken` MUST NOT retroactively bump `last_heartbeat_at`, because
+  // the stale-dispatch detector is the signal the coordinator uses to know a
+  // newer dispatch for the same task has hung. Silently no-op'ing keeps the
+  // zombie-heartbeat race from masking a hung retry (§5.3.4).
+  recordHeartbeat(dispatchId: string, at: string): void {
+    this.db
+      .prepare(
+        "UPDATE dispatch_contexts SET last_heartbeat_at = ? WHERE id = ? AND status = 'dispatched'"
+      )
+      .run(at, dispatchId)
+  }
+
+  // Why: the query restricts to currently-dispatched contexts AND respects a
+  // dispatched-at grace. Without `status = 'dispatched'`, every completed /
+  // failed / circuit_broken row with an old-or-null last_heartbeat_at would
+  // warn every tick (warning storm). Without `dispatched_at < :threshold`,
+  // a freshly-dispatched worker would trip the warning during its first
+  // heartbeat interval (false positive). Callers supply the threshold as an
+  // ISO timestamp so the SQLite string-compare ordering works correctly
+  // (ISO-8601 compares lexicographically in time order).
+  getStaleDispatches(thresholdIso: string): DispatchContextRow[] {
+    return this.db
+      .prepare(
+        `SELECT * FROM dispatch_contexts
+         WHERE status = 'dispatched'
+           AND dispatched_at IS NOT NULL
+           AND dispatched_at < ?
+           AND (last_heartbeat_at IS NULL OR last_heartbeat_at < ?)`
+      )
+      .all(thresholdIso, thresholdIso) as DispatchContextRow[]
+  }
+
   failDispatch(ctxId: string, error: string): DispatchContextRow | undefined {
     const ctx = this.db.prepare('SELECT * FROM dispatch_contexts WHERE id = ?').get(ctxId) as
       | DispatchContextRow

src/main/runtime/orchestration/preamble.test.ts

@@ -1,67 +1,118 @@
+import { spawnSync } from 'child_process'
 import { describe, expect, it } from 'vitest'
 import { buildDispatchPreamble } from './preamble'
 
+function baseParams(overrides: Partial<Parameters<typeof buildDispatchPreamble>[0]> = {}) {
+  return {
+    taskId: 'task_abc123',
+    dispatchId: 'ctx_def456',
+    taskSpec: 'Implement the login form',
+    coordinatorHandle: 'term_coord',
+    ...overrides
+  }
+}
+
 describe('buildDispatchPreamble', () => {
   it('substitutes template variables', () => {
-    const result = buildDispatchPreamble({
-      taskId: 'task_abc123',
-      taskSpec: 'Implement the login form',
-      coordinatorHandle: 'term_coord'
-    })
+    const result = buildDispatchPreamble(baseParams())
 
     expect(result).toContain('task_abc123')
+    expect(result).toContain('ctx_def456')
     expect(result).toContain('term_coord')
     expect(result).toContain('Implement the login form')
     expect(result).not.toContain('{{')
   })
 
-  it('includes worker_done command', () => {
-    const result = buildDispatchPreamble({
-      taskId: 'task_x',
-      taskSpec: 'do stuff',
-      coordinatorHandle: 'term_c'
-    })
+  it('includes worker_done command with --body 3-sentence summary prompt and reportPath', () => {
+    const result = buildDispatchPreamble(baseParams())
 
     expect(result).toContain('worker_done')
     expect(result).toContain('orchestration send')
     expect(result).toContain('orchestration check')
+    expect(result).toContain('--body')
+    expect(result).toMatch(/3-sentence summary/)
+    expect(result).toContain('reportPath')
   })
 
-  it('includes the task spec after separator', () => {
-    const result = buildDispatchPreamble({
-      taskId: 'task_x',
-      taskSpec: 'refactor the auth module',
-      coordinatorHandle: 'term_c'
-    })
+  it('CLI examples parse as valid shell (bash -n on the extracted block)', () => {
+    const result = buildDispatchPreamble(baseParams())
+    // Why: feeding `bash -n` the full preamble falsely fails on apostrophes
+    // in the surrounding prose. Slice between the CLI markers and strip
+    // shell-style comment lines so we only syntax-check the commands.
+    const cliStart = result.indexOf('=== CLI COMMANDS ===')
+    const cliEnd = result.indexOf('=== AFTER YOU SEND worker_done ===')
+    expect(cliStart).toBeGreaterThan(-1)
+    expect(cliEnd).toBeGreaterThan(cliStart)
+    const block = result.slice(cliStart, cliEnd)
+    const stripped = block
+      .split('\n')
+      .filter((line) => !line.trim().startsWith('#'))
+      .filter((line) => !line.trim().startsWith('==='))
+      .join('\n')
 
-    expect(result).toContain('--- TASK ---')
+    const check = spawnSync('bash', ['-n'], { input: stripped, encoding: 'utf8' })
+    expect(check.status).toBe(0)
+  })
+
+  it('includes heartbeat CLI block with taskId and dispatchId and 5-minute cadence', () => {
+    const result = buildDispatchPreamble(baseParams())
+    expect(result).toContain('--type heartbeat')
+    expect(result).toContain('--subject "alive"')
+    expect(result).toMatch(/5 minutes/)
+    // Both taskId and dispatchId are rendered inside the payload template
+    // (regression guard for §5.3.4 attribution — dispatchId attribution
+    // prevents the zombie-heartbeat-masks-hung-retry race).
+    const heartbeatLine = result
+      .split('\n')
+      .find((line) => line.includes('"taskId"') && line.includes('dispatchId'))
+    expect(heartbeatLine).toBeTruthy()
+    expect(heartbeatLine).toContain('task_abc123')
+    expect(heartbeatLine).toContain('ctx_def456')
+  })
+
+  it('includes ask block with BEHAVIOR RULE #1 forbidding AskUserQuestion', () => {
+    const result = buildDispatchPreamble(baseParams())
+    expect(result).toContain('orchestration ask')
+    expect(result).toContain('--question')
+    expect(result).toContain('--timeout-ms 600000')
+    // Why: the exact phrase is asserted so the rule can't be trimmed away by
+    // accident. BEHAVIOR RULE #1 is the only place AskUserQuestion appears.
+    expect(result).toContain('BEHAVIOR RULE #1')
+    expect(result).toContain('NEVER use AskUserQuestion')
+    // AskUserQuestion must appear ONLY inside the rule text, not anywhere
+    // else (e.g., not in an example payload or header). Count occurrences
+    // of the exact token as a sanity check.
+    const occurrences = (result.match(/AskUserQuestion/g) ?? []).length
+    // Three mentions: the one-liner ban, the TUI-prompt rationale, and the
+    // "when tempted to reach for AskUserQuestion" closing line.
+    expect(occurrences).toBe(3)
+  })
+
+  it('includes AFTER YOU SEND block with 2-minute poll cadence and release signal', () => {
+    const result = buildDispatchPreamble(baseParams())
+    expect(result).toContain('=== AFTER YOU SEND worker_done ===')
+    expect(result).toMatch(/2 minutes/)
+    expect(result).toMatch(/may exit/)
+  })
+
+  it('uses === TASK === separator with the task spec appended', () => {
+    const result = buildDispatchPreamble(baseParams({ taskSpec: 'refactor the auth module' }))
+    expect(result).toContain('=== TASK ===')
     expect(result).toContain('refactor the auth module')
   })
 
   it('uses orca CLI by default when devMode is not set', () => {
-    const result = buildDispatchPreamble({
-      taskId: 'task_x',
-      taskSpec: 'do stuff',
-      coordinatorHandle: 'term_c'
-    })
-
+    const result = buildDispatchPreamble(baseParams())
     expect(result).toContain('orca orchestration send')
     expect(result).toContain('orca orchestration check')
+    expect(result).toContain('orca orchestration ask')
   })
 
   it('uses orca-dev CLI when devMode is true', () => {
-    const result = buildDispatchPreamble({
-      taskId: 'task_x',
-      taskSpec: 'do stuff',
-      coordinatorHandle: 'term_c',
-      devMode: true
-    })
-
+    const result = buildDispatchPreamble(baseParams({ devMode: true }))
     expect(result).toContain('orca-dev orchestration send')
     expect(result).toContain('orca-dev orchestration check')
-    // Ensure no bare "orca " (without -dev) appears as a CLI command.
-    // We split on "orca-dev" first so those occurrences don't produce
-    // false positives, then check the remaining fragments.
+    expect(result).toContain('orca-dev orchestration ask')
     const fragments = result.split('orca-dev')
     for (const fragment of fragments) {
       expect(fragment).not.toMatch(/orca orchestration/)
@@ -69,13 +120,7 @@ describe('buildDispatchPreamble', () => {
   })
 
   it('uses orca CLI when devMode is false', () => {
-    const result = buildDispatchPreamble({
-      taskId: 'task_x',
-      taskSpec: 'do stuff',
-      coordinatorHandle: 'term_c',
-      devMode: false
-    })
-
+    const result = buildDispatchPreamble(baseParams({ devMode: false }))
     expect(result).toContain('orca orchestration send')
     expect(result).toContain('orca orchestration check')
   })
@@ -83,6 +128,7 @@ describe('buildDispatchPreamble', () => {
   it('appends a BASE DRIFT section when baseDrift.behind > 0', () => {
     const result = buildDispatchPreamble({
       taskId: 'task_x',
+      dispatchId: 'ctx_x',
       taskSpec: 'do stuff',
       coordinatorHandle: 'term_c',
       baseDrift: {
@@ -98,12 +144,13 @@ describe('buildDispatchPreamble', () => {
     expect(result).toContain('  - feat: B')
     expect(result).toContain('  - chore: C')
     // drift section must appear before the task spec
-    expect(result.indexOf('--- BASE DRIFT ---')).toBeLessThan(result.indexOf('--- TASK ---'))
+    expect(result.indexOf('--- BASE DRIFT ---')).toBeLessThan(result.indexOf('=== TASK ==='))
   })
 
   it('omits the drift section when baseDrift.behind is 0', () => {
     const result = buildDispatchPreamble({
       taskId: 'task_x',
+      dispatchId: 'ctx_x',
       taskSpec: 'do stuff',
       coordinatorHandle: 'term_c',
       baseDrift: {
@@ -120,6 +167,7 @@ describe('buildDispatchPreamble', () => {
   it('omits the drift section when baseDrift is undefined', () => {
     const result = buildDispatchPreamble({
       taskId: 'task_x',
+      dispatchId: 'ctx_x',
       taskSpec: 'do stuff',
       coordinatorHandle: 'term_c'
     })
@@ -131,6 +179,7 @@ describe('buildDispatchPreamble', () => {
   it('lists drift subjects in the order provided, each prefixed with two spaces and dash', () => {
     const result = buildDispatchPreamble({
       taskId: 'task_x',
+      dispatchId: 'ctx_x',
       taskSpec: 'do stuff',
       coordinatorHandle: 'term_c',
       baseDrift: {
@@ -147,4 +196,16 @@ describe('buildDispatchPreamble', () => {
     expect(secondIdx).toBeGreaterThan(firstIdx)
     expect(thirdIdx).toBeGreaterThan(secondIdx)
   })
+
+  it('renders a stable snapshot of the full preamble', () => {
+    // Why: single strict snapshot catches any accidental regression in
+    // formatting or rule presence in one line.
+    const result = buildDispatchPreamble({
+      taskId: 'task_SNAP',
+      dispatchId: 'ctx_SNAP',
+      taskSpec: 'TASK_BODY',
+      coordinatorHandle: 'term_COORD'
+    })
+    expect(result).toMatchSnapshot()
+  })
 })

src/main/runtime/orchestration/preamble.ts

@@ -1,5 +1,12 @@
 export type PreambleParams = {
   taskId: string
+  // Why: the heartbeat payload attributes liveness to a specific dispatch
+  // context (not just a task). A retried task has multiple dispatch_contexts
+  // rows; keying heartbeats on dispatchId prevents a straggler heartbeat from
+  // a previously-failed dispatch from masking a hung retry. Both call sites
+  // already have this value in scope (coordinator.ts dispatch.id, rpc
+  // orchestration.ts ctx.id).
+  dispatchId: string
   taskSpec: string
   coordinatorHandle: string
   devMode?: boolean
@@ -17,36 +24,99 @@ export type PreambleParams = {
   }
 }
 
+// Why: 5 minutes is frequent enough that the coordinator's stale-heartbeat
+// check (threshold 10 min) catches a hung worker within one tick, and
+// infrequent enough to avoid inbox spam on long tasks. One constant so
+// cadence tuning is a single-line change (Q1 in DESIGN_DOC_PREAMBLE_FIX.md).
+const HEARTBEAT_INTERVAL_MIN = 5
+
 // Why: the dispatch preamble teaches agents about Orca's CLI commands for
-// structured communication. Agents don't need prior knowledge of Orca — they
-// treat these as shell tools the same way they use git or npm.
+// structured communication. Behavioral rules (body summary, heartbeat cadence,
+// no-AskUserQuestion) live as inline comments above the relevant CLI example,
+// not as a separate prose block — LLM readers anchor on examples and skim
+// trailing prose, so rules must land at the point of use.
 export function buildDispatchPreamble(params: PreambleParams): string {
   // Why: in dev mode, agents must use orca-dev to connect to the dev runtime's
   // socket. Without this, agents inside the dev Electron app would call the
   // production CLI and talk to the wrong Orca instance (Section 6.4).
   const cli = params.devMode ? 'orca-dev' : 'orca'
 
-  const header = `You are working inside Orca, a multi-agent IDE. You have access to these
-CLI commands for communicating with the coordinator:
+  const header = `You are working inside Orca, a multi-agent IDE. You are a dispatched worker.
+Your coordinator's terminal handle is: ${params.coordinatorHandle}
+Your task ID is: ${params.taskId}
 
-  # Report task completion (REQUIRED when done):
+You talk to the coordinator only through the CLI commands below. Do not use
+Slack, GitHub comments, or any other channel to reach a human during the run.
+
+=== CLI COMMANDS ===
+
+  # Report task completion (REQUIRED when done — even on failure).
+  #
+  # RULE: --body must be a 3-sentence executive summary (what you did,
+  # what you found, what's left). Never send an empty body; the coordinator
+  # reads the body first and only opens artifacts if it needs more detail.
+  # If you produced a long-form artifact, include its path as
+  # payload.reportPath so the coordinator can find it without a file search.
+  #
+  # RULE: send worker_done exactly once. Failure is still a worker_done
+  # with subject like "Failed: <reason>" — never silently exit.
   ${cli} orchestration send --to ${params.coordinatorHandle} \\
-    --type worker_done --subject "Done" \\
-    --payload '{"taskId":"${params.taskId}","filesModified":[...]}'
+    --type worker_done --subject "<short status>" \\
+    --body "<3-sentence summary: what you did, what you found, what's left>" \\
+    --payload '{"taskId":"${params.taskId}","filesModified":["path/a","path/b"],"reportPath":"<optional: path to the full artifact>"}'
 
-  # Report a blocker or failure:
+  # BEHAVIOR RULE: send a heartbeat every ${HEARTBEAT_INTERVAL_MIN} minutes
+  # while actively working on the task. The coordinator uses this to
+  # distinguish "still thinking" from "hung / crashed." Skip heartbeats only
+  # while blocked inside \`check --wait\` or \`ask\` — those calls are
+  # themselves liveness signals.
+  #
+  # Include BOTH taskId and dispatchId in the payload: the coordinator
+  # attributes the heartbeat to the specific dispatch context, not just
+  # the task, so a straggler heartbeat from a previously-failed dispatch
+  # cannot mask a hung retry.
+  ${cli} orchestration send --to ${params.coordinatorHandle} \\
+    --type heartbeat --subject "alive" \\
+    --payload '{"taskId":"${params.taskId}","dispatchId":"${params.dispatchId}","phase":"<short: investigating|implementing|reviewing|waiting>"}'
+
+  # Ask the coordinator a question and block until it answers.
+  #
+  # BEHAVIOR RULE #1 (MUST NOT VIOLATE):
+  # NEVER use AskUserQuestion; use \`${cli} orchestration ask\` or send
+  # --type decision_gate. AskUserQuestion opens a local TUI prompt that the
+  # coordinator cannot see and cannot answer — your session will hang forever
+  # waiting on a human. Every interactive question goes through \`ask\` below.
+  #
+  # The \`ask\` verb is a thin wrapper: it sends a decision_gate message and
+  # blocks on \`check --wait\` until the coordinator replies, then prints the
+  # reply body. Use it anywhere you would otherwise have reached for
+  # AskUserQuestion.
+  ${cli} orchestration ask --to ${params.coordinatorHandle} \\
+    --question "<your question>" \\
+    --options "<optional,comma,separated>" \\
+    --timeout-ms 600000
+
+  # Escalate a blocker or failure (pre-completion, when you need the
+  # coordinator to do something before you can continue):
   ${cli} orchestration send --to ${params.coordinatorHandle} \\
     --type escalation --subject "Blocked: <reason>" \\
-    --body "<details>"
+    --body "<details>" \\
+    --payload '{"taskId":"${params.taskId}"}'
 
-  # Check for messages from the coordinator or other agents:
+  # Check for messages from the coordinator:
   ${cli} orchestration check
 
-Your assigned task ID is: ${params.taskId}
+=== AFTER YOU SEND worker_done ===
 
-When you finish your task, run the worker_done command above with the
-list of files you modified. If you are blocked or need help, send an
-escalation. Do not exit the session.`
+Keep the shell session open for a grace period (10 minutes) in case the
+coordinator sends a follow-up or re-dispatches you. Poll with
+\`${cli} orchestration check\` every 2 minutes during that window. If no
+follow-up arrives, you may exit after the grace period — the coordinator
+will not expect further output from you.
+
+If the coordinator re-dispatches you (you will receive a fresh preamble +
+TASK block), reset your polling and start the new task. Do not respond
+to the previous task's follow-ups after a re-dispatch.`
 
   // Why: the drift section fires only when the coordinator allowed dispatch
   // against a stale worktree (via `allow-stale-base: true` in the task spec,
@@ -58,7 +128,7 @@ escalation. Do not exit the session.`
 
   return `${header}${drift}
 
---- TASK ---
+=== TASK ===
 ${params.taskSpec}`
 }
 

src/main/runtime/orchestration/types.ts

@@ -6,6 +6,7 @@ export type MessageType =
   | 'escalation'
   | 'handoff'
   | 'decision_gate'
+  | 'heartbeat'
 
 export type MessagePriority = 'normal' | 'high' | 'urgent'
 
@@ -54,6 +55,7 @@ export type DispatchContextRow = {
   dispatched_at: string | null
   completed_at: string | null
   created_at: string
+  last_heartbeat_at: string | null
 }
 
 export type DecisionGateRow = {

src/main/runtime/rpc/methods/orchestration.test.ts

@@ -38,7 +38,7 @@ describe('orchestration RPC methods', () => {
 
   it('registers all expected methods', () => {
     const registry = buildRegistry(ORCHESTRATION_METHODS)
-    expect(registry.size).toBe(15)
+    expect(registry.size).toBe(16)
     expect(registry.has('orchestration.send')).toBe(true)
     expect(registry.has('orchestration.check')).toBe(true)
     expect(registry.has('orchestration.reply')).toBe(true)
@@ -48,6 +48,7 @@ describe('orchestration RPC methods', () => {
     expect(registry.has('orchestration.taskUpdate')).toBe(true)
     expect(registry.has('orchestration.dispatch')).toBe(true)
     expect(registry.has('orchestration.dispatchShow')).toBe(true)
+    expect(registry.has('orchestration.ask')).toBe(true)
     expect(registry.has('orchestration.run')).toBe(true)
     expect(registry.has('orchestration.runStop')).toBe(true)
     expect(registry.has('orchestration.gateCreate')).toBe(true)
@@ -726,6 +727,145 @@ describe('orchestration RPC methods', () => {
     })
   })
 
+  describe('orchestration.ask', () => {
+    it('sends a decision_gate and returns the first thread reply', async () => {
+      setup()
+      vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
+      vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
+      vi.spyOn(runtime, 'waitForMessage').mockImplementation(async () => {
+        // Simulate coordinator replying in the thread during the wait
+        const outbound = db.getInbox(10).find((m) => m.type === 'decision_gate')
+        if (outbound) {
+          db.insertMessage({
+            from: 'term_coord',
+            to: 'term_worker',
+            subject: 'Re: Question',
+            body: 'go ahead',
+            threadId: outbound.id
+          })
+        }
+      })
+
+      const result = (await call('orchestration.ask', {
+        from: 'term_worker',
+        to: 'term_coord',
+        question: 'proceed?',
+        options: 'yes, no',
+        timeoutMs: 500
+      })) as {
+        answer: string
+        messageId: string
+        threadId: string
+        timedOut: boolean
+      }
+
+      expect(result.timedOut).toBe(false)
+      expect(result.answer).toBe('go ahead')
+      expect(result.messageId).toMatch(/^msg_/)
+
+      // Outbound decision_gate message was persisted with parsed options.
+      const outbound = db.getInbox(10).find((m) => m.type === 'decision_gate')
+      expect(outbound).toBeTruthy()
+      expect(outbound?.subject).toBe('Question')
+      expect(outbound?.body).toBe('proceed?')
+      const payload = JSON.parse(outbound!.payload ?? '{}')
+      expect(payload.question).toBe('proceed?')
+      expect(payload.options).toEqual(['yes', 'no'])
+    })
+
+    it('returns timedOut when no reply arrives in the window', async () => {
+      setup()
+      vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
+      vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
+      vi.spyOn(runtime, 'waitForMessage').mockResolvedValue()
+
+      const result = (await call('orchestration.ask', {
+        from: 'term_worker',
+        to: 'term_coord',
+        question: 'still there?',
+        timeoutMs: 1
+      })) as { answer: string | null; timedOut: boolean; messageId: string | null }
+
+      expect(result.timedOut).toBe(true)
+      expect(result.answer).toBeNull()
+      expect(result.messageId).toBeNull()
+      // Outbound message still persisted (coordinator can still see it).
+      const outbound = db.getInbox(10).find((m) => m.type === 'decision_gate')
+      expect(outbound).toBeTruthy()
+    })
+
+    it('rejects group addresses with a dedicated error (no message persisted)', async () => {
+      setup()
+      await expect(
+        call('orchestration.ask', {
+          from: 'term_worker',
+          to: '@reviewers',
+          question: 'ok?'
+        })
+      ).rejects.toThrow(/does not support group addresses/)
+      expect(db.getInbox(10)).toHaveLength(0)
+    })
+
+    it('does not return distractor messages on a different thread', async () => {
+      setup()
+      vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
+      vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
+      let wakeCount = 0
+      vi.spyOn(runtime, 'waitForMessage').mockImplementation(async () => {
+        wakeCount++
+        const outbound = db.getInbox(20).find((m) => m.type === 'decision_gate')
+        if (wakeCount === 1 && outbound) {
+          // First wake: distractor in a DIFFERENT thread — must be ignored.
+          db.insertMessage({
+            from: 'term_coord',
+            to: 'term_worker',
+            subject: 'unrelated',
+            body: 'other',
+            threadId: 'thread_other'
+          })
+        } else if (wakeCount === 2 && outbound) {
+          // Second wake: correct thread reply.
+          db.insertMessage({
+            from: 'term_coord',
+            to: 'term_worker',
+            subject: 'Re: Question',
+            body: 'correct answer',
+            threadId: outbound.id
+          })
+        }
+      })
+
+      const result = (await call('orchestration.ask', {
+        from: 'term_worker',
+        to: 'term_coord',
+        question: 'filter?',
+        timeoutMs: 2_000
+      })) as { answer: string; timedOut: boolean }
+
+      expect(result.timedOut).toBe(false)
+      expect(result.answer).toBe('correct answer')
+    })
+
+    it('parses options CSV with whitespace and empty entries', async () => {
+      setup()
+      vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
+      vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
+      vi.spyOn(runtime, 'waitForMessage').mockResolvedValue()
+
+      await call('orchestration.ask', {
+        from: 'w',
+        to: 'c',
+        question: 'q',
+        options: 'a, b ,,c',
+        timeoutMs: 1
+      })
+
+      const outbound = db.getInbox(10).find((m) => m.type === 'decision_gate')
+      const payload = JSON.parse(outbound!.payload ?? '{}')
+      expect(payload.options).toEqual(['a', 'b', 'c'])
+    })
+  })
+
   describe('orchestration.reset', () => {
     it('resets all state', async () => {
       setup()

src/main/runtime/rpc/methods/orchestration.ts

@@ -15,7 +15,8 @@ const MESSAGE_TYPES: MessageType[] = [
   'merge_ready',
   'escalation',
   'handoff',
-  'decision_gate'
+  'decision_gate',
+  'heartbeat'
 ]
 
 const TASK_STATUSES: TaskStatus[] = [
@@ -40,7 +41,8 @@ const SendParams = z.object({
       'merge_ready',
       'escalation',
       'handoff',
-      'decision_gate'
+      'decision_gate',
+      'heartbeat'
     ])
     .optional(),
   priority: z.enum(['normal', 'high', 'urgent']).optional(),
@@ -117,6 +119,14 @@ const DispatchShowParams = z.object({
   task: OptionalString
 })
 
+const AskParams = z.object({
+  to: requiredString('Missing --to'),
+  question: requiredString('Missing --question'),
+  options: OptionalString,
+  timeoutMs: OptionalFiniteNumber,
+  from: OptionalString
+})
+
 const ResetParams = z.object({
   all: OptionalBoolean,
   tasks: OptionalBoolean,
@@ -371,6 +381,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
         try {
           const preamble = buildDispatchPreamble({
             taskId: task.id,
+            dispatchId: ctx.id,
             taskSpec: task.spec,
             coordinatorHandle: params.from ?? 'coordinator',
             devMode: params.devMode
@@ -400,6 +411,73 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
     }
   }),
 
+  defineMethod({
+    name: 'orchestration.ask',
+    params: AskParams,
+    handler: async (params, { runtime }) => {
+      // Why: group addresses have no unambiguous answer semantics (whose
+      // reply wins? first? consensus?) and the ~60-LOC scope is not the
+      // place to design that. Rejecting here closes the silent-timeout
+      // footgun where a worker passing `--to @reviewers` would have the
+      // decision_gate inserted against a literal string no one subscribes
+      // to. Workers that need fan-out fall back to `send --type decision_gate`.
+      if (isGroupAddress(params.to)) {
+        throw new Error(
+          'ask does not support group addresses; use send --type decision_gate for fan-out questions'
+        )
+      }
+
+      const db = runtime.getOrchestrationDb()
+      const from = params.from ?? 'unknown'
+      const timeoutMs = params.timeoutMs ?? 600_000
+      const options =
+        params.options
+          ?.split(',')
+          .map((s) => s.trim())
+          .filter(Boolean) ?? []
+
+      const payload = JSON.stringify({ question: params.question, options })
+      const outbound = db.insertMessage({
+        from,
+        to: params.to,
+        subject: 'Question',
+        body: params.question,
+        type: 'decision_gate',
+        payload
+      })
+      runtime.deliverPendingMessagesForHandle(params.to)
+      runtime.notifyMessageArrived(params.to)
+
+      const threadId = outbound.id
+      const deadline = Date.now() + timeoutMs
+      const afterSequence = outbound.sequence
+
+      // Why: loop with a remaining-budget guard so an unrelated distractor
+      // message that wakes waitForMessage does not cause indefinite iteration.
+      // waitForMessage is handle-scoped, so we re-query by thread on every
+      // wake-up to separate "reply in my thread arrived" from "something
+      // else was delivered to this handle."
+      while (true) {
+        const replies = db.getThreadMessagesFor(threadId, from, afterSequence)
+        if (replies.length > 0) {
+          const reply = replies[0]
+          db.markAsRead([reply.id])
+          return {
+            answer: reply.body,
+            messageId: reply.id,
+            threadId,
+            timedOut: false
+          }
+        }
+        const remainingMs = deadline - Date.now()
+        if (remainingMs <= 0) {
+          return { answer: null, messageId: null, threadId, timedOut: true }
+        }
+        await runtime.waitForMessage(from, { timeoutMs: remainingMs })
+      }
+    }
+  }),
+
   ...ORCHESTRATION_GATE_METHODS,
 
   defineMethod({