From d8b486998beac3f5d5a829b7611c46c9b236495a Mon Sep 17 00:00:00 2001
From: Jinwoo-H <jinwoo0825@gmail.com>
Date: Mon, 4 May 2026 04:04:40 -0400
Subject: [PATCH 1/3] feat(orchestration): preamble rules + heartbeat schema

- Preamble (#7, #15, #9): worker_done body ("3-sentence summary" + reportPath),
  BEHAVIOR RULE #1 forbidding AskUserQuestion, heartbeat every 5 minutes with
  taskId+dispatchId payload, AFTER YOU SEND grace window.
- Schema v2 migration: adds 'heartbeat' to messages.type CHECK, adds
  dispatch_contexts.last_heartbeat_at, gated by user_version PRAGMA with
  transactional rebuild + explicit CREATE INDEX to avoid silent perf regress.
- DB helpers: recordHeartbeat (dispatched-only), getStaleDispatches,
  getThreadMessagesFor (thread+handle scoped for ask).

Co-authored-by: Orca <help@stably.ai>
---
 .../__snapshots__/preamble.test.ts.snap       |  83 ++++++
 src/main/runtime/orchestration/db.test.ts     | 278 ++++++++++++++++++
 src/main/runtime/orchestration/db.ts          | 176 ++++++++++-
 .../runtime/orchestration/preamble.test.ts    | 139 ++++++---
 src/main/runtime/orchestration/preamble.ts    | 100 ++++++-
 src/main/runtime/orchestration/types.ts       |   2 +
 6 files changed, 712 insertions(+), 66 deletions(-)
 create mode 100644 src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap

diff --git a/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap b/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap
new file mode 100644
index 00000000000..8c8fd9f09f5
--- /dev/null
+++ b/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"
+`;
diff --git a/src/main/runtime/orchestration/db.test.ts b/src/main/runtime/orchestration/db.test.ts
index d412255fdae..c0cd774bec4 100644
--- a/src/main/runtime/orchestration/db.test.ts
+++ b/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)
+    })
+  })
 })
diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts
index b7c74bee72e..95573709fb7 100644
--- a/src/main/runtime/orchestration/db.ts
+++ b/src/main/runtime/orchestration/db.ts
@@ -33,6 +33,13 @@ function generateId(prefix: string): string {
   return `${prefix}_${randomBytes(6).toString('hex')}`
 }
 
+// Why: bumped from implicit 1 → 2 in the preamble-hardening PR. Existing
+// on-disk DBs were created before `'heartbeat'` was a valid MessageType
+// and before the `last_heartbeat_at` column existed; without this gate the
+// coordinator would throw `no such column` on first tick and silently drop
+// every heartbeat INSERT on the CHECK constraint.
+const SCHEMA_VERSION = 2
+
 export class OrchestrationDb {
   private db: Database.Database
 
@@ -42,6 +49,7 @@ export class OrchestrationDb {
     this.db.pragma('synchronous = NORMAL')
     this.db.pragma('busy_timeout = 5000')
     this.createTables()
+    this.migrate()
   }
 
   private createTables(): void {
@@ -55,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')),
@@ -89,16 +97,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);
@@ -132,6 +141,100 @@ 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
+    }
+
+    // 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.
+    if (current < 2) {
+      this.db.exec('BEGIN')
+      try {
+        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'))
+            );
+            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);
+          `)
+        }
+
+        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: {
@@ -201,6 +304,25 @@ export class OrchestrationDb {
       .all(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 {
@@ -362,6 +484,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
diff --git a/src/main/runtime/orchestration/preamble.test.ts b/src/main/runtime/orchestration/preamble.test.ts
index 71c27d50f66..b6b62a35b86 100644
--- a/src/main/runtime/orchestration/preamble.test.ts
+++ b/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,14 +120,20 @@ 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')
   })
+
+  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()
+  })
 })
diff --git a/src/main/runtime/orchestration/preamble.ts b/src/main/runtime/orchestration/preamble.ts
index 3c61f0be56a..857b79126d1 100644
--- a/src/main/runtime/orchestration/preamble.ts
+++ b/src/main/runtime/orchestration/preamble.ts
@@ -1,41 +1,111 @@
 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
 }
 
+// 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'
 
-  return `You are working inside Orca, a multi-agent IDE. You have access to these
-CLI commands for communicating with the coordinator:
+  return `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.
 
---- TASK ---
+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 ===
 ${params.taskSpec}`
 }
diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts
index 1d0ae9a3914..4bb425bca70 100644
--- a/src/main/runtime/orchestration/types.ts
+++ b/src/main/runtime/orchestration/types.ts
@@ -6,6 +6,7 @@ export type MessageType =
   | 'escalation'
   | 'handoff'
   | 'decision_gate'
+  | 'heartbeat'
 
 export type MessagePriority = 'normal' | 'high' | 'urgent'
 
@@ -53,6 +54,7 @@ export type DispatchContextRow = {
   dispatched_at: string | null
   completed_at: string | null
   created_at: string
+  last_heartbeat_at: string | null
 }
 
 export type DecisionGateRow = {

From 1e77c830e95e66c69e708990cc15b5d9fa4959fc Mon Sep 17 00:00:00 2001
From: Jinwoo-H <jinwoo0825@gmail.com>
Date: Mon, 4 May 2026 04:04:49 -0400
Subject: [PATCH 2/3] feat(orchestration): coordinator heartbeat + stale
 detector
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Handle incoming 'heartbeat' messages by calling recordHeartbeat keyed on
payload.dispatchId (strict — log-and-skip if missing, no taskId fallback so
a straggler heartbeat from a previously-failed dispatch cannot mask a hung
retry per §5.3.4). On every tick after the 10-minute threshold, emit one
log per stale dispatched row — no auto-fail.

Also threads dispatchId through buildDispatchPreamble so workers can
attribute their heartbeats back to the correct dispatch context.

Co-authored-by: Orca <help@stably.ai>
---
 .../runtime/orchestration/coordinator.test.ts | 82 +++++++++++++++++++
 src/main/runtime/orchestration/coordinator.ts | 57 +++++++++++++
 2 files changed, 139 insertions(+)

diff --git a/src/main/runtime/orchestration/coordinator.test.ts b/src/main/runtime/orchestration/coordinator.test.ts
index ac0ebbb4974..d1423e07a48 100644
--- a/src/main/runtime/orchestration/coordinator.test.ts
+++ b/src/main/runtime/orchestration/coordinator.test.ts
@@ -355,6 +355,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()
diff --git a/src/main/runtime/orchestration/coordinator.ts b/src/main/runtime/orchestration/coordinator.ts
index 541102f4ae5..e80f6f4e194 100644
--- a/src/main/runtime/orchestration/coordinator.ts
+++ b/src/main/runtime/orchestration/coordinator.ts
@@ -41,6 +41,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
@@ -173,10 +180,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) {
@@ -194,6 +218,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
@@ -205,6 +232,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}`)
 
@@ -389,6 +445,7 @@ export class Coordinator {
     // so they talk to the dev runtime's socket, not production (Section 6.4).
     const preamble = buildDispatchPreamble({
       taskId: task.id,
+      dispatchId: dispatch.id,
       taskSpec: task.spec,
       coordinatorHandle: this.opts.coordinatorHandle,
       devMode: process.env.ORCA_USER_DATA_PATH?.includes('orca-dev')

From 7e1c4a2bf2b317893826a6072f10e629e44aef1c Mon Sep 17 00:00:00 2001
From: Jinwoo-H <jinwoo0825@gmail.com>
Date: Mon, 4 May 2026 04:04:56 -0400
Subject: [PATCH 3/3] feat(orchestration): orca orchestration ask verb
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds a CLI verb that sends a decision_gate message and blocks on the
coordinator's reply, scoped to the outbound message's thread. Group
addresses (@all, @idle, …) are rejected — fan-out questions must use
send --type decision_gate explicitly.

--json emits bare single-line JSON (bypassing printResult) so workers can
pipe `orca orchestration ask … --json | jq -r .answer` without unwrapping
an RPC envelope; human mode prints just the answer. On timeout the verb
exits 1 and returns {answer: null, timedOut: true}.

This is the CLI surface BEHAVIOR RULE #1 in the dispatch preamble points
workers at instead of AskUserQuestion.

Co-authored-by: Orca <help@stably.ai>
---
 src/cli/handlers/orchestration.ts             |  42 ++++++
 src/cli/specs/orchestration.ts                |   7 +
 .../runtime/rpc/methods/orchestration.test.ts | 142 +++++++++++++++++-
 src/main/runtime/rpc/methods/orchestration.ts |  82 +++++++++-
 4 files changed, 270 insertions(+), 3 deletions(-)

diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts
index be807c28a2f..14691813629 100644
--- a/src/cli/handlers/orchestration.ts
+++ b/src/cli/handlers/orchestration.ts
@@ -172,6 +172,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
diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts
index e527fe8b953..bc0a535174f 100644
--- a/src/cli/specs/orchestration.ts
+++ b/src/cli/specs/orchestration.ts
@@ -71,6 +71,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',
diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts
index 64a2507d0f1..10b3be35bae 100644
--- a/src/main/runtime/rpc/methods/orchestration.test.ts
+++ b/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)
@@ -621,6 +622,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()
diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts
index 5c467e369b0..3fc02f74593 100644
--- a/src/main/runtime/rpc/methods/orchestration.ts
+++ b/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(),
@@ -109,6 +111,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,
@@ -347,6 +357,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
@@ -376,6 +387,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({