fix(orchestration): let a worker start helpers on the Run it works for (#29205)

* fix(orchestration): let a worker start helpers on the Run it works for

A dispatched worker could only start helper workers by taking a coordinator
seat on a new Run, and taking a seat moved its inbox. worker-start now also
accepts a worker on an active Dispatch: its helpers run in the same Run, report
to it (worker_done, heartbeat, escalation, status, ask), and it answers their
questions. The worker's own mailbox never moves. When it settles, whatever its
helpers sent that it never handled goes to the Run's coordinator.

Authority comes from the worker's own Dispatch, re-checked inside the
reservation transaction. Helpers are fresh agents on this host; a worker may
retry only its own helper's latest attempt. Remote workers keep the old
own-Run instructions. No saved field or wire change.

* fix(orchestration): read helper rows without type assertions

* fix(orchestration): hand a settled helper creator's mail to the Run as live delivery

When a worker that started helpers settled, its helpers' unhandled mail was
re-addressed to the Run by settlement SQL alone. Nothing woke the Run's
coordinator, delivered rows kept their old delivery stamp so pointers and the
restart scan skipped them, payload-free status reports were left behind
because selection required payload.dispatchId, and a question asked while the
creator was being stopped stayed with the stopped creator.

- Select handed-back mail by its recipient mailbox and Run (unread reports
  plus pending questions, even ones already read), reset delivery and pointer
  state on the moved rows, and fence the old mailbox's Deliveries.
- After the settling transaction ends, the runtime re-reads exactly the moved
  rows and wakes the Run with their real types; a rollback wakes no one and a
  failure falls back to the existing repoint retry. Restart recovery is the
  existing unread-undelivered scan.
- Hand mail back again at the final stop settlement, which also closes the
  stopped Dispatch's own questions asked during the stop.
- A worker_done receipt replay re-offers the handed-back targets.
- Scope markAsDelivered to the mailbox it pointed, so a late pointer stamp
  from the old owner cannot hide mail now owned by the Run.

* test(orchestration): drop an unused inbox type

* fix(orchestration): read only the Run when handing a settled worker's mail back

Every settlement runs the hand-back, so a full dispatch row read there added a statement per settlement. Also expect the mailbox-scoped delivered stamp in two delivery tests from main.

* fix(orchestration): re-offer handed-back Run mail on every worker_done replay

A helper creator's worker_done hands its helpers' unread reports and pending
questions to the Run. The send stored the Run's extra wake targets in the
mutation receipt, but the executor rewrites that receipt on completion and on
each replay, so an identical retry only woke the original destination.

Derive the targets at replay time instead: for the settled Dispatch (when it
has same-Run helpers), notify run:<id> for each type of unread report or
pending question there. Notification only; nothing is re-settled or inserted.
The receipt augmentation is removed.

* style(orchestration): format with the formatter version main now pins

skill-guides/orchestration.md

@@ -116,6 +116,8 @@ Use `task-create` plus `worker-start --task <task_id>` for planned fan-out with
 dependencies or a retry of a known Task. Use dependencies only for real ordering
 and prefer parallel waves over chains deeper than three or four steps; nested
 workers obey the depth limit, and a new Run does not reset the caller's depth.
+A dispatched worker starts its own helpers with `worker-start --spec` on the Run
+it works for, without `run-create`; the helpers report to that worker.
 
 A consuming `check` names its caller with `--terminal <handle>`, never `--from`;
 omit it inside the coordinator's own Orca terminal. It returns the bound Run's

skill-guides/orchestration/references/coordinator-loop.md

@@ -18,6 +18,8 @@ ORCA orchestration task-list --ready --brief --json
 `spec_truncated` identifies shortened rows. Omit it when full specs are needed or
 when an older CLI rejects the flag. A nested worker must respect
 `nested_worker_depth_exceeded`; creating another Run does not reset depth.
+Helpers a worker starts appear in this Run's `worker-list` with their
+`creatorDispatchId`, and report to that worker rather than to you.
 
 ## Launch preferences
 

skill-guides/orchestration/references/messaging-and-gates.md

@@ -55,7 +55,9 @@ Dispatch lifecycle messages never target groups.
 ## Questions and gates
 
 A worker uses `ask`; its timeout leaves one durable question pending, which the
-worker resumes by message ID. The coordinator answers that message with `reply`.
+worker resumes by message ID. The coordinator answers that message with `reply`;
+a helper's question goes to the worker that started it, which answers it the
+same way.
 
 Use a gate only for a coordinator-owned Task-DAG decision:
 

skill-guides/orchestration/references/worker-contract.md

@@ -53,6 +53,21 @@ you were replaced; `consumer_fenced` is the only way you learn that.
 If `send` or `ask` returns `consumer_fenced`, the command ran from another
 party's terminal (a coordinator or another worker); run it from your own terminal.
 
+## Helper workers
+
+When the preamble has a SUB-DISPATCH section, start helpers on your own
+Dispatch's Run. Do not `run-create` or `run-use` for them.
+
+```text
+ORCA orchestration worker-start --from <worker_handle> --spec "<sub-task>" --worktree current --agent <agent> --json
+ORCA orchestration reply --from <worker_handle> --id <message_id> --body "<answer>"
+```
+
+Their questions, escalations, and `worker_done` arrive in your own `check`, next
+to your coordinator's mail. Answer their questions with `reply`, and wait for
+every helper's `worker_done` before sending yours. After you settle, a helper
+still running reports to your coordinator's Run.
+
 ## Escalation
 
 Escalate only before completion and only when the coordinator must intervene:

src/main/runtime/orca-runtime-test-orchestration-messages.spec.ts

@@ -267,11 +267,13 @@ export class InMemoryOrchestrationMessages {
     )
   }
 
-  markAsDelivered(ids: string[]): void {
-    const deliveredIds = new Set(ids)
+  markAsDelivered(ids: string[], mailboxHandle: string): void {
+    const requested = new Set(ids)
+    const deliveredIds = new Set<string>()
     for (const message of this.messages) {
-      if (deliveredIds.has(message.id)) {
+      if (requested.has(message.id) && message.to_handle === mailboxHandle) {
         message.delivered_at = '1970-01-01 00:00:00'
+        deliveredIds.add(message.id)
       }
     }
     this.clearMailboxPointerEnter(deliveredIds)

src/main/runtime/orchestration/db-message-timestamp.test.ts

@@ -16,7 +16,7 @@ describe('orchestration message timestamps', () => {
     })
 
     expect(message.created_at).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$/)
-    db.markAsDelivered([message.id])
+    db.markAsDelivered([message.id], 'b')
     expect(db.getMessageById(message.id)?.delivered_at).toMatch(
       /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$/
     )

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

@@ -76,7 +76,7 @@ describe('OrchestrationDb', () => {
       const m1 = d.insertMessage({ runId, from: 'a', to: 'b', subject: 'one' })
       const m2 = d.insertMessage({ runId, from: 'a', to: 'b', subject: 'two' })
 
-      d.markAsDelivered([m1.id])
+      d.markAsDelivered([m1.id], 'b')
 
       // Push delivery query: only undelivered, unread.
       const pending = d.getUndeliveredUnreadMessages('b')

src/main/runtime/orchestration/db-undelivered-mailboxes.test.ts

@@ -22,7 +22,7 @@ describe('undelivered orchestration mailboxes', () => {
     })
     db.insertMessage({ runId: 'run_legacy_local', from: 'a', to: 'pending', subject: 'first' })
     db.insertMessage({ runId: 'run_legacy_local', from: 'a', to: 'pending', subject: 'second' })
-    db.markAsDelivered([delivered.id])
+    db.markAsDelivered([delivered.id], 'delivered')
     db.markAsRead([read.id])
 
     expect(db.getUndeliveredUnreadMailboxHandles()).toEqual(['pending'])

src/main/runtime/orchestration/db/messages/helper-mailbox-hand-back.ts

@@ -0,0 +1,95 @@
+import type { OrchestrationDb } from '../orchestration-db'
+
+/** Mail a settling helper creator never handled, now owned by its Run's mailbox. */
+export type MailHandBack = { mailbox: string; messageIds: string[]; types: string[] }
+
+/**
+ * A settled worker no longer reads its mailbox, so its helpers' unread reports and unanswered
+ * questions (even ones it read) become the Run coordinator's, delivered as if they just arrived.
+ */
+export function handHelperMailToRun(db: OrchestrationDb, dispatchId: string): void {
+  // Why: every settlement runs this, so read only the Run, not the whole dispatch row.
+  const runId = db.db
+    .prepare('SELECT run_id FROM dispatch_contexts WHERE id = ?')
+    .get(dispatchId)?.run_id
+  if (typeof runId !== 'string') {
+    return
+  }
+  const mailbox = `dispatch:${dispatchId}`
+  // Why: the recipient mailbox is the ownership fact; report payloads and senders are optional.
+  const rows = db.db
+    .prepare(
+      `SELECT id, type FROM messages
+       WHERE run_id = ? AND to_handle = ? AND (
+         (read = 0 AND delivery_contract = 'current_delivery'
+           AND type IN ('worker_done', 'escalation', 'status'))
+         OR id IN (
+           SELECT message_id FROM question_threads WHERE run_id = ? AND status = 'pending'))`
+    )
+    .all(runId, mailbox, runId)
+  if (
+    rows.length === 0 ||
+    !db.db
+      .prepare(
+        'SELECT 1 FROM dispatch_contexts WHERE creator_dispatch_id = ? AND run_id = ? LIMIT 1'
+      )
+      .get(dispatchId, runId)
+  ) {
+    return
+  }
+  const handBack: MailHandBack = {
+    mailbox: `run:${runId}`,
+    messageIds: rows.map((row) => String(row.id)),
+    types: [...new Set(rows.map((row) => String(row.type)))]
+  }
+  // Why: the old mailbox's delivery stamp and pointer reservation would hide the rows from the Run.
+  db.db
+    .prepare(
+      `UPDATE messages
+       SET to_handle = ?, read = 0, delivered_at = NULL, pointer_enter_pending = 0,
+           pointer_pty_id = NULL, pointer_process_incarnation = NULL
+       WHERE id IN (SELECT value FROM json_each(?))`
+    )
+    .run(handBack.mailbox, JSON.stringify(handBack.messageIds))
+  db.fenceUnacknowledgedMailboxDeliveries(mailbox)
+  db.pendingMailHandBacks.push(handBack)
+  if (db.pendingMailHandBacks.length === 1) {
+    // Why: settlements are synchronous, so this runs after the outermost COMMIT or ROLLBACK.
+    queueMicrotask(() => {
+      for (const pending of db.pendingMailHandBacks.splice(0)) {
+        db.mailHandBackListener?.(pending)
+      }
+    })
+  }
+}
+
+/** The types of handed-back rows the Run still owes attention to, read after the settlement. */
+export function committedHandBackTypes(db: OrchestrationDb, handBack: MailHandBack): string[] {
+  return db.db
+    .prepare(
+      `SELECT DISTINCT type FROM messages
+       WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery'
+         AND id IN (SELECT value FROM json_each(?))
+         AND (type <> 'question'
+           OR id IN (SELECT message_id FROM question_threads WHERE status = 'pending'))`
+    )
+    .all(handBack.mailbox, JSON.stringify(handBack.messageIds))
+    .map((row) => String(row.type))
+}
+
+/** The Run mail a helper creator's settlement may have handed back that still owes attention. */
+export function handBackReplayTargets(db: OrchestrationDb, dispatchId: string) {
+  return db.db
+    .prepare(
+      `SELECT DISTINCT m.to_handle AS mailbox, m.type FROM dispatch_contexts AS d
+       JOIN messages AS m ON m.to_handle = 'run:' || d.run_id
+       WHERE d.id = ? AND m.read = 0 AND m.delivery_contract = 'current_delivery'
+         AND EXISTS (SELECT 1 FROM dispatch_contexts AS h
+           WHERE h.creator_dispatch_id = d.id AND h.run_id = d.run_id)
+         AND (m.type IN ('worker_done', 'escalation', 'status') OR (m.type = 'question'
+           AND m.id IN (SELECT message_id FROM question_threads
+             WHERE run_id = d.run_id AND status = 'pending')))`
+    )
+    .all(dispatchId)
+    .map((row) => ({ to: String(row.mailbox), type: String(row.type) }))
+}

src/main/runtime/orchestration/db/messages/message-inbox.ts

@@ -9,7 +9,8 @@ const MESSAGE_MUTATION_SAVEPOINT = 'message_id_mutation'
 function runBatchedMessageMutation(
   db: OrchestrationDb,
   ids: string[],
-  sqlForPlaceholders: (placeholders: string) => string
+  sqlForPlaceholders: (placeholders: string) => string,
+  leadingBindings: string[] = []
 ): void {
   if (ids.length === 0) {
     return
@@ -19,7 +20,7 @@ function runBatchedMessageMutation(
     for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) {
       const batch = ids.slice(offset, offset + MESSAGE_ID_UPDATE_BATCH_SIZE)
       const placeholders = batch.map(() => '?').join(',')
-      db.db.prepare(sqlForPlaceholders(placeholders)).run(...batch)
+      db.db.prepare(sqlForPlaceholders(placeholders)).run(...leadingBindings, ...batch)
     }
     db.db.exec(`RELEASE ${MESSAGE_MUTATION_SAVEPOINT}`)
   } catch (error) {
@@ -165,7 +166,8 @@ export function markAsRead(this: OrchestrationDb, ids: string[]): void {
 }
 
 // Why: use datetime('now') so delivered_at matches the space-format UTC shape of the table's other timestamps for correct ordering (§3.2).
-export function markAsDelivered(this: OrchestrationDb, ids: string[]): void {
+// Scoped to the pointed mailbox so a late stamp cannot mark mail since handed elsewhere.
+export function markAsDelivered(this: OrchestrationDb, ids: string[], mailboxHandle: string): void {
   runBatchedMessageMutation(
     this,
     ids,
@@ -173,7 +175,8 @@ export function markAsDelivered(this: OrchestrationDb, ids: string[]): void {
       `UPDATE messages
        SET delivered_at = datetime('now'), pointer_enter_pending = 0,
            pointer_pty_id = NULL, pointer_process_incarnation = NULL
-       WHERE id IN (${placeholders})`
+       WHERE to_handle = ? AND id IN (${placeholders})`,
+    [mailboxHandle]
   )
 }
 

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

@@ -1,5 +1,6 @@
 import Database from '../../../sqlite/sync-database'
 import { attachOrchestrationDbMethods } from './attach-orchestration-db-methods'
+import type { MailHandBack } from './messages/helper-mailbox-hand-back'
 import { hardenOrchestrationDatabaseFiles } from './database-file-permissions'
 import { backfillFederatedStubHomeRuns } from './federation/federated-stub-home-run-backfill'
 import type { OrchestrationDbMethods } from './orchestration-db-methods'
@@ -24,6 +25,10 @@ class OrchestrationDbCore {
   // per-terminal fan-out. Only createDispatchContext flips this false→true.
   hasAnyDispatchContextsCache: boolean | undefined
   localMutationCallerFingerprint: string | undefined
+  // Why: in memory only; the re-armed rows themselves are what restart recovery reads.
+  pendingMailHandBacks: MailHandBack[] = []
+  /** The runtime's wake for mail a settlement handed to a Run, called once the transaction ends. */
+  mailHandBackListener: ((handBack: MailHandBack) => void) | undefined
 
   constructor(dbPath: (string & {}) | ':memory:') {
     this.db = new Database(dbPath)

src/main/runtime/orchestration/db/questions/question-threads.ts

@@ -2,6 +2,7 @@ import type { MessageRow, QuestionRow } from '../../types'
 import { OrchestrationError } from '../../orchestration-error'
 import { exposeQuestionTimestamps } from '../utc-timestamp'
 import type { OrchestrationDb } from '../orchestration-db'
+import { handHelperMailToRun } from '../messages/helper-mailbox-hand-back'
 
 export function createQuestion(
   this: OrchestrationDb,
@@ -11,6 +12,8 @@ export function createQuestion(
     askerHandle: string
     question: string
     options?: string[]
+    /** The asker's report mailbox; the Run when omitted. */
+    to?: string
   }
 ): { question: QuestionRow; message: MessageRow } {
   this.db.exec('BEGIN IMMEDIATE')
@@ -29,7 +32,7 @@ export function createQuestion(
     }
     const message = this.insertMessage({
       from: `dispatch:${params.dispatchId}`,
-      to: `run:${params.runId}`,
+      to: params.to ?? `run:${params.runId}`,
       subject: 'Question',
       body: params.question,
       type: 'question',
@@ -75,13 +78,21 @@ export function answerQuestion(
   params: {
     messageId: string
     runId: string
-    consumerGeneration: number
     body: string
-  }
+  } & (
+    | { consumerGeneration: number }
+    /** A worker answering its own helper, fenced on its Dispatch instead of the Run's consumer. */
+    | { answeredBy: { dispatchId: string; consumerGeneration: number } }
+  )
 ): { question: QuestionRow; message: MessageRow; duplicate: boolean } {
+  const answeredBy = 'answeredBy' in params ? params.answeredBy : null
+  const byDispatch = answeredBy?.dispatchId ?? null
+  const generation = 'consumerGeneration' in params ? params.consumerGeneration : null
   this.db.exec('BEGIN IMMEDIATE')
   try {
-    this.requireCurrentConsumer(params.runId, params.consumerGeneration)
+    if (generation !== null) {
+      this.requireCurrentConsumer(params.runId, generation)
+    }
     const question = this.getQuestionRaw(params.messageId)
     if (!question || question.run_id !== params.runId) {
       throw new OrchestrationError(
@@ -89,6 +100,20 @@ export function answerQuestion(
         `Question ${params.messageId} was not found in Run ${params.runId}.`
       )
     }
+    if (answeredBy !== null) {
+      const answerer = this.getDispatchContextById(answeredBy.dispatchId)
+      if (
+        this.getMessageById(question.message_id)?.to_handle !==
+          `dispatch:${answeredBy.dispatchId}` ||
+        (answerer?.status !== 'pending' && answerer?.status !== 'dispatched') ||
+        answerer.consumer_generation !== answeredBy.consumerGeneration
+      ) {
+        throw new OrchestrationError(
+          'consumer_fenced',
+          `Question ${params.messageId} was not asked of active Dispatch ${answeredBy.dispatchId}.`
+        )
+      }
+    }
     if (question.status === 'closed') {
       throw new OrchestrationError(
         'dispatch_inactive',
@@ -111,7 +136,7 @@ export function answerQuestion(
     }
 
     const message = this.insertMessage({
-      from: `run:${params.runId}`,
+      from: byDispatch !== null ? `dispatch:${byDispatch}` : `run:${params.runId}`,
       to: `dispatch:${question.dispatch_id}`,
       subject: 'Re: Question',
       body: params.body,
@@ -127,7 +152,7 @@ export function answerQuestion(
              answered_by_generation = ?, answered_at = datetime('now')
          WHERE message_id = ? AND status = 'pending'`
       )
-      .run(message.id, params.body, params.consumerGeneration, question.message_id)
+      .run(message.id, params.body, generation, question.message_id)
     const answered = this.getQuestionRaw(question.message_id) as QuestionRow
     const storedMessage = this.getMessageById(message.id) as MessageRow
     this.db.exec('COMMIT')
@@ -143,6 +168,8 @@ export function answerQuestion(
 }
 
 export function closeQuestionsForDispatch(this: OrchestrationDb, dispatchId: string): string[] {
+  // Every settlement path calls this, so it is also where helper mail moves to the Run.
+  handHelperMailToRun(this, dispatchId)
   const rows = this.db
     .prepare("SELECT message_id FROM question_threads WHERE dispatch_id = ? AND status = 'pending'")
     .all(dispatchId) as { message_id: string }[]

src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-start.ts

@@ -41,6 +41,8 @@ export function createStartingWorkerDispatch(
     /** Who is dispatching, for nesting depth. Required so a new caller must decide. */
     creator: DispatchCreator
     maxDepth: number
+    /** A worker's own Dispatch that authorized this helper; it must still be the creator here. */
+    requiredCreatorDispatchId?: string
   }
 ): { dispatch: DispatchContextRow; worker: WorkerDispatchRow; task: TaskRow } {
   this.db.exec('BEGIN IMMEDIATE')
@@ -133,6 +135,16 @@ export function createStartingWorkerDispatch(
 
     const id = generateId('ctx')
     const creatorDispatchId = this.resolveCreatorDispatchId(params.creator)
+    if (
+      params.requiredCreatorDispatchId !== undefined &&
+      creatorDispatchId !== params.requiredCreatorDispatchId
+    ) {
+      throw new OrchestrationError(
+        'consumer_fenced',
+        `Dispatch ${params.requiredCreatorDispatchId} is no longer this caller's active work, so it cannot start a helper.`,
+        { effectsApplied: false }
+      )
+    }
     if (params.mutationReceipt) {
       this.db
         .prepare(

src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-stop.ts

@@ -145,6 +145,8 @@ export function settleWorkerStop(this: OrchestrationDb, dispatchId: string): Wor
       })
     }
     reconcileTaskAfterDispatchInterruption(this, dispatch.task_id, dispatchId)
+    // Why: helpers keep reporting to a stopping creator, so mail sent since begin-stop moves here.
+    this.closeQuestionsForDispatch(dispatchId)
     this.db.exec('COMMIT')
     return this.getWorkerDispatch(dispatchId) as WorkerDispatchRow
   } catch (error) {
@@ -201,6 +203,7 @@ export function reconcileFederatedWorkerStop(
       })
     }
     reconcileTaskAfterDispatchInterruption(this, dispatch.task_id, dispatchId)
+    this.closeQuestionsForDispatch(dispatchId)
     commitLifecycleWriteTransaction(this.db, transaction)
     return this.getWorkerDispatch(dispatchId) as WorkerDispatchRow
   } catch (error) {

src/main/runtime/orchestration/dispatch-report-mailbox.ts

@@ -0,0 +1,41 @@
+import type { OrchestrationDb } from './db'
+import type { DispatchContextRow, MessageType } from './types'
+
+/** Reports about the worker's own task; gates and handoffs stay with the Run's coordinator. */
+export function isDispatchReportType(type: MessageType | undefined): boolean {
+  return (
+    type === undefined ||
+    type === 'worker_done' ||
+    type === 'heartbeat' ||
+    type === 'escalation' ||
+    type === 'status'
+  )
+}
+
+/** The live Dispatch that started this one in the same Run: the worker it reports to. */
+export function activeCreatorDispatch(
+  db: OrchestrationDb,
+  dispatch: Pick<DispatchContextRow, 'creator_dispatch_id' | 'run_id'>
+): DispatchContextRow | undefined {
+  const creator = dispatch.creator_dispatch_id
+    ? db.getDispatchContextById(dispatch.creator_dispatch_id)
+    : undefined
+  return creator &&
+    creator.run_id === dispatch.run_id &&
+    (creator.status === 'pending' || creator.status === 'dispatched')
+    ? creator
+    : undefined
+}
+
+/**
+ * Where a Dispatch's reports go: the worker that started it while that worker is still on its
+ * own task, otherwise the Run. Derived on every send, so a helper whose creator settled reports
+ * to the Run without anything being rewritten.
+ */
+export function dispatchReportMailbox(
+  db: OrchestrationDb,
+  dispatch: Pick<DispatchContextRow, 'creator_dispatch_id' | 'run_id'>
+): string {
+  const creator = activeCreatorDispatch(db, dispatch)
+  return creator ? `dispatch:${creator.id}` : `run:${dispatch.run_id}`
+}

src/main/runtime/orchestration/mailbox-pointer-stage.ts

@@ -152,7 +152,7 @@ function finishPointerWriteAndStageEnter<TWaiter extends OrchestrationMessageWai
         isCursorAgentTitle
       )
     ) {
-      db.markAsDelivered(flight.stagedMessageIds)
+      db.markAsDelivered(flight.stagedMessageIds, args.mailboxHandle)
       args.state.clearWatermark(args.mailboxHandle, args.newestSequence, ptyId)
       args.redrive(args.mailboxHandle)
       return

src/main/runtime/orchestration/message-batch-atomicity.test.ts

@@ -68,7 +68,11 @@ describe('message batch atomicity', () => {
     }
     rejectLastMessageUpdate(sqlite)
 
-    expect(() => db?.[method](messageIds)).toThrow('blocked')
+    expect(() =>
+      method === 'markAsDelivered'
+        ? db?.markAsDelivered(messageIds, 'recipient')
+        : db?.[method](messageIds)
+    ).toThrow('blocked')
 
     const changed = sqlite.prepare(changedCountSql).get() as { count: number }
     expect(changed.count).toBe(0)

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

@@ -86,8 +86,8 @@ describe('buildDispatchPreamble', () => {
     expect(headings).toHaveLength(0)
     expect(codeBlocks).toHaveLength(2)
     expect(codeBlocks[1]).toMatchObject({ lang: 'sh' })
-    expect(codeBlocks[1].value).toContain('orchestration worker-start --task <task_id>')
-    expect(result).toContain('able to dispatch further.\n\n---')
+    expect(codeBlocks[1].value).toContain('orchestration worker-start --from')
+    expect(result).toContain('helpers of their own.\n\n---')
     expect(result).toContain('before starting.\n\n---')
   })
 
@@ -256,11 +256,11 @@ describe('sub-dispatch section', () => {
     )
   })
 
-  it('appears with the run-create sequence when budget remains', () => {
+  it('starts helpers on this Dispatch, without a Run of their own, when budget remains', () => {
     const preamble = buildDispatchPreamble({ ...base, canDispatchSubWorkers: true })
     expect(preamble).toContain('=== SUB-DISPATCH ===')
-    expect(preamble).toContain('orchestration run-create')
-    expect(preamble).toContain('orchestration worker-start')
+    expect(preamble).toContain('orchestration worker-start --from term_worker --spec')
+    expect(preamble).not.toContain('orchestration run-create')
   })
 
   it('keeps the task block last so the spec is not buried', () => {

src/main/runtime/orchestration/preamble.ts

@@ -36,6 +36,8 @@ export type PreambleParams = {
   workerKind?: 'prompt-returning-agent' | 'bare-shell'
   // Why gated: advertising a verb the depth cap will reject just burns a turn.
   canDispatchSubWorkers?: boolean
+  // Why: a remote worker has no Dispatch row on its own host to start helpers under.
+  subWorkersInOwnRun?: boolean
 }
 
 // Why: 5 minutes is frequent enough that the coordinator's stale-heartbeat
@@ -155,7 +157,11 @@ ${postDoneInstructions}`
   const drift =
     params.baseDrift && params.baseDrift.behind > 0 ? buildDriftSection(params.baseDrift) : ''
 
-  const subDispatch = params.canDispatchSubWorkers ? buildSubDispatchSection(cli) : ''
+  const subDispatch = params.canDispatchSubWorkers
+    ? params.subWorkersInOwnRun
+      ? buildOwnRunSubDispatchSection(cli)
+      : buildSubDispatchSection(cli, params.workerHandle)
+    : ''
 
   return `${header}${drift}${subDispatch}
 
@@ -235,7 +241,27 @@ ${chat ? CHAT_REDISPATCH_PARAGRAPH : TERMINAL_REDISPATCH_PARAGRAPH}`
 // as a blocker.
 // Why fenced + blank line before the closing `---`: unfenced `<placeholders>` are stripped as raw
 // HTML by the Chat UI, and a rule directly under a paragraph is a setext H2 (giant last sentence).
-function buildSubDispatchSection(cli: string): string {
+function buildSubDispatchSection(cli: string, workerHandle: string): string {
+  return `
+
+=== SUB-DISPATCH ===
+You may start helper workers for this task. They work in this Dispatch's Run and
+report to you: their questions, escalations, and worker_done arrive in your own
+\`check\`, and your coordinator's mail keeps arriving there too.
+
+\`\`\`sh
+  ${cli} orchestration worker-start --from ${workerHandle} --spec "<sub-task>" --worktree current --agent <agent> --json
+  # Answer a helper's question by its message ID:
+  ${cli} orchestration reply --from ${workerHandle} --id <message_id> --body "<answer>"
+\`\`\`
+
+Wait for every helper's worker_done before reporting your own. Nesting is
+capped; your helpers may not be able to start helpers of their own.
+
+---`
+}
+
+function buildOwnRunSubDispatchSection(cli: string): string {
   return `
 
 === SUB-DISPATCH ===

src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts

@@ -162,7 +162,7 @@ describe('structured mailbox pointer delivery', () => {
     await flush()
     expect(send).toHaveBeenCalledTimes(1)
     expect(send.mock.calls[0]![0].operationId).toMatch(/^\d{13}-[0-9a-f]{32}$/)
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], 'dispatch:d1')
   })
 
   it('nudges through the worker`s own handle for direct peer mail outside a dispatch', async () => {
@@ -180,7 +180,7 @@ describe('structured mailbox pointer delivery', () => {
     expect(send.mock.calls[0]![0].body.blocks[0]).toMatchObject({
       text: expect.not.stringContaining('--run')
     })
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], IDENTITY.handle)
   })
 
   it('releases its in-flight lane when reading the context fails', async () => {
@@ -194,7 +194,7 @@ describe('structured mailbox pointer delivery', () => {
     delivery.deliverForHandle('dispatch:d1')
     await flush()
     expect(send).toHaveBeenCalledOnce()
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], 'dispatch:d1')
   })
 
   it('retains mail when the dispatch settles unknown', async () => {
@@ -224,7 +224,7 @@ describe('structured mailbox pointer delivery', () => {
     delivery.onJournalActivity('session-1')
     await flush()
     expect(send).toHaveBeenCalledTimes(1)
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], 'dispatch:d1')
   })
 
   it('rechecks a held notice when the answer edge lands during its host call', async () => {
@@ -238,7 +238,7 @@ describe('structured mailbox pointer delivery', () => {
     delivery.deliverForHandle('dispatch:d1')
     await flush()
     expect(send).toHaveBeenCalledTimes(2)
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], 'dispatch:d1')
   })
 
   it('sends the pointer through the chat, with who it is from, and counts it pointed once queued', async () => {
@@ -259,7 +259,7 @@ describe('structured mailbox pointer delivery', () => {
       }
     })
     // The chat's queue holds it now, as it holds the person's: the same mail is not pointed again.
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], 'dispatch:d1')
     expect(stored.has('dispatch:d1')).toBe(false)
   })
 
@@ -305,7 +305,10 @@ describe('structured mailbox pointer delivery', () => {
     expect(send.mock.calls[1]![0].body.from).toMatchObject({
       orchestration: { messages: [{ messageId: 'm2' }] }
     })
-    expect(markAsDelivered.mock.calls).toEqual([[['m1']], [['m2']]])
+    expect(markAsDelivered.mock.calls).toEqual([
+      [['m1'], 'run:run_1'],
+      [['m2'], 'run:run_1']
+    ])
     delivery.onJournalActivity('session-1')
     delivery.deliverForHandle('run:run_1')
     await flush()
@@ -323,7 +326,7 @@ describe('structured mailbox pointer delivery', () => {
     delivery.deliverForHandle('dispatch:d1')
     await flush()
     expect(send).toHaveBeenCalledTimes(1)
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], 'dispatch:d1')
   })
 
   it('does not re-nudge a mailbox still holding its own unacked batch', async () => {
@@ -478,7 +481,7 @@ describe('structured mailbox pointer delivery', () => {
     delivery.onJournalActivity('session-1')
     await flush()
     expect(send).toHaveBeenCalledTimes(1)
-    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'], 'dispatch:d1')
     expect(stored.has('dispatch:d1')).toBe(false)
   })
 

src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts

@@ -332,7 +332,7 @@ export class OrchestrationStructuredMailboxPointerDelivery<
     })
     if (operation.kind === 'stamp') {
       // A send this lane gave up waiting on ran after all.
-      db.markAsDelivered(staged)
+      db.markAsDelivered(staged, mailboxHandle)
       db.deleteStructuredPointerOperation(mailboxHandle)
       this.sentOperationIds.delete(mailboxHandle)
       return true
@@ -362,7 +362,7 @@ export class OrchestrationStructuredMailboxPointerDelivery<
       this.retain(mailboxHandle, sessionId, retainReasonForDispatch(outcome.state), reservedTypes)
       return false
     }
-    db.markAsDelivered(staged)
+    db.markAsDelivered(staged, mailboxHandle)
     db.deleteStructuredPointerOperation(mailboxHandle)
     this.sentOperationIds.delete(mailboxHandle)
     return true

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

@@ -211,6 +211,7 @@ export const ORCHESTRATION_FEDERATION_ATTACH_METHODS = [
               // Why the worker host's own setting: enforcement runs here, with this
               // host's code, against this host's cap.
               canDispatchSubWorkers: (params.depth ?? 1) < runtime.getNestedWorkerMaxDepth(),
+              subWorkersInOwnRun: true,
               cliCommand: runtime.getTerminalOrchestrationCliCommand(terminalHandle)
             })
           }

src/main/runtime/rpc/methods/orchestration/messaging/ask-methods.ts

@@ -2,6 +2,7 @@ import { defineMethod } from '../../../core'
 import { OrchestrationError } from '../../../../orchestration/orchestration-error'
 import { clampOrchestrationAskTimeoutMs } from '../../../../../../shared/orchestration-ask-timeout'
 import { isGroupAddress } from '../../../../orchestration/groups'
+import { dispatchReportMailbox } from '../../../../orchestration/dispatch-report-mailbox'
 import { AskParams } from '../schemas'
 import { rejectFederatedExplicitTarget } from '../routing'
 import { askRemoteRunHome } from './ask-remote'
@@ -109,9 +110,11 @@ export const ORCHESTRATION_ASK_METHODS = [
             `Dispatch ${activeDispatch.id} belongs to Run ${run.id}, not ${params.run}.`
           )
         }
+        const reportTo = dispatchReportMailbox(db, activeDispatch)
         if (
           params.to &&
           params.to !== `run:${run.id}` &&
+          params.to !== reportTo &&
           resolveOrchestrationParty(params.to, db).address !==
             mailboxAddressOf(runCoordinatorKey(run))
         ) {
@@ -120,15 +123,18 @@ export const ORCHESTRATION_ASK_METHODS = [
             `ask from Dispatch ${activeDispatch.id} must target its owning Run ${run.id}.`
           )
         }
+        // An explicit Run or coordinator target still reaches the Run; otherwise the asker's creator.
+        const to = params.to && params.to !== reportTo ? `run:${run.id}` : reportTo
         const created = db.createQuestion({
           runId: run.id,
           dispatchId: activeDispatch.id,
           askerHandle: from,
           question: params.question as string,
-          options
+          options,
+          to
         })
         question = created.question
-        runtime.notifyMessageArrived(`run:${run.id}`, created.message.type)
+        runtime.notifyMessageArrived(to, created.message.type)
       }
 
       const questionId = question.message_id

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

@@ -709,7 +709,7 @@ describe('orchestration RPC methods', () => {
       setup()
       const msg = db.insertMessage({ from: 'a', to: 'b', subject: 'hi' })
       // Why: simulate push-on-idle stamping delivered_at without the runtime loop.
-      db.markAsDelivered([msg.id])
+      db.markAsDelivered([msg.id], 'b')
 
       const result = (await call('orchestration.check', {
         terminal: 'b',

src/main/runtime/rpc/methods/orchestration/messaging/helper-question-answerer.ts

@@ -0,0 +1,38 @@
+import type { OrchestrationCompatibilityEvidence } from '../../../../../../shared/orchestration-compatibility-evidence'
+import type { OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity'
+import type { MessageRow } from '../../../../orchestration/types'
+import type { OrcaRuntimeService } from '../../../../orca-runtime'
+import { assertCallerHandleMatchesEvidence, orchestrationCallerIdentity } from '../runs/run-scope'
+
+/** The caller's Dispatch when a helper asked this question of it; the Run answers otherwise. */
+export function helperQuestionAnswerer(
+  runtime: OrcaRuntimeService,
+  args: {
+    question: MessageRow
+    from: string | undefined
+    session: OrchestrationSessionCaller | undefined
+    evidence: OrchestrationCompatibilityEvidence | undefined
+  }
+): { dispatchId: string; consumerGeneration: number } | undefined {
+  const askedOf = args.question.to_handle.startsWith('dispatch:')
+    ? args.question.to_handle.slice('dispatch:'.length)
+    : undefined
+  if (!askedOf || (!args.from && !args.session)) {
+    return undefined
+  }
+  const handle = args.from ?? ''
+  if (args.from) {
+    assertCallerHandleMatchesEvidence(runtime, args.from, args.evidence)
+  }
+  const caller = orchestrationCallerIdentity(runtime, {
+    handle,
+    session: args.session,
+    paneKey: args.session ? undefined : runtime.getTerminalPaneKey(handle)
+  })
+  const own = runtime
+    .getOrchestrationDb()
+    .getActiveDispatchForIdentity(caller.address, caller.paneKey ?? undefined)
+  return own?.id === askedOf
+    ? { dispatchId: own.id, consumerGeneration: own.consumer_generation }
+    : undefined
+}

src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts

@@ -13,6 +13,7 @@ import { exposeMessage } from './mailbox-message-receipt'
 import { resolveOrchestrationParty } from '../../../../orchestration/orchestration-party'
 import { recordReceiptBeforeNudge, replayMutationNudge } from './mutation-replay-nudge'
 import { resolveReplyRecipient } from './recipient-routing'
+import { helperQuestionAnswerer } from './helper-question-answerer'
 import {
   ReplyParams,
   InboxParams,
@@ -71,6 +72,28 @@ export const ORCHESTRATION_MESSAGE_METHODS = [
       }
 
       const question = db.getQuestion(params.id)
+      const creatorAnswerer =
+        question && (params.run === undefined || params.run === question.run_id)
+          ? helperQuestionAnswerer(runtime, {
+              question: original,
+              from: params.from,
+              session: orchestrationCaller,
+              evidence: orchestrationCompatibilityEvidence
+            })
+          : undefined
+      if (question && creatorAnswerer) {
+        const answered = db.answerQuestion({
+          messageId: question.message_id,
+          runId: question.run_id,
+          answeredBy: creatorAnswerer,
+          body: params.body
+        })
+        return recordReceiptBeforeNudge(
+          recordMutationReceipt,
+          { ...answered, message: exposeMessage(answered.message) },
+          () => runtime.notifyMessageArrived(`dispatch:${question.dispatch_id}`, 'status')
+        )
+      }
       if (question) {
         const run = resolveRunScope(runtime, {
           runId: params.run ?? question.run_id,

src/main/runtime/rpc/methods/orchestration/messaging/send-methods.ts

@@ -1,6 +1,10 @@
 import { defineMethod } from '../../../core'
 import { OrchestrationError } from '../../../../orchestration/orchestration-error'
 import { isGroupAddress } from '../../../../orchestration/groups'
+import {
+  dispatchReportMailbox,
+  isDispatchReportType
+} from '../../../../orchestration/dispatch-report-mailbox'
 import { orchestrationSkillRecoveryData } from '../../../../../../shared/orchestration-rpc-contract'
 import {
   SendParams,
@@ -21,6 +25,7 @@ import {
   stripMutationReplayNudge
 } from '../../../orchestration-mutation-receipt'
 import { replayMutationNudge } from './mutation-replay-nudge'
+import { handBackReplayTargets } from '../../../../orchestration/db/messages/helper-mailbox-hand-back'
 import { sendRemoteMessage } from './send-remote'
 import { mayNameSession } from './session-recipient'
 import { sendPointToPointMessage } from './send-point-to-point'
@@ -62,6 +67,12 @@ export const ORCHESTRATION_SEND_METHODS = [
           : undefined)
       if (replayNudge) {
         replayMutationNudge(runtime, replayNudge)
+        const settled =
+          params.type === 'worker_done' ? settledDispatchId(replayedMutationReceipt) : undefined
+        // Why: derived, since the executor's receipt rewrites drop stored hand-back targets.
+        for (const target of settled ? handBackReplayTargets(db, settled) : []) {
+          runtime.notifyMessageArrived(target.to, target.type)
+        }
         return stripMutationReplayNudge(replayedMutationReceipt)
       }
       const from = params.from ?? 'unknown'
@@ -134,7 +145,11 @@ export const ORCHESTRATION_SEND_METHODS = [
         (!to ||
           ((params.type === 'worker_done' || params.type === 'heartbeat') && routing.dispatchId))
       ) {
-        to = `run:${routing.run.id}`
+        const reporting =
+          routing.dispatchId && isDispatchReportType(params.type)
+            ? db.getDispatchContextById(routing.dispatchId)
+            : undefined
+        to = reporting ? dispatchReportMailbox(db, reporting) : `run:${routing.run.id}`
       }
       if (!to) {
         throw new OrchestrationError(
@@ -247,3 +262,16 @@ export const ORCHESTRATION_SEND_METHODS = [
     }
   })
 ]
+
+function settledDispatchId(receipt: unknown): string | undefined {
+  if (!receipt || typeof receipt !== 'object' || !('lifecycle' in receipt)) {
+    return undefined
+  }
+  const { lifecycle } = receipt
+  return lifecycle &&
+    typeof lifecycle === 'object' &&
+    'dispatchId' in lifecycle &&
+    typeof lifecycle.dispatchId === 'string'
+    ? lifecycle.dispatchId
+    : undefined
+}

src/main/runtime/rpc/methods/orchestration/worker/local-worker-start.ts

@@ -4,7 +4,7 @@ import { waitForWorkerAgentReady } from '../../../../launched-agent-composer-rea
 import type { OrcaRuntimeService } from '../../../../orca-runtime'
 import { describeTerminalWaitBlockedReason } from '../../../../../../shared/terminal-wait-blocked-reason-legacy-alias'
 import type { OrchestrationDb } from '../../../../orchestration/db'
-import type { RunRow, TaskRow } from '../../../../orchestration/types'
+import type { DispatchContextRow, RunRow, TaskRow } from '../../../../orchestration/types'
 import type {
   OrchestrationCallerIdentity,
   OrchestrationSessionCaller
@@ -48,6 +48,8 @@ export async function startLocalWorker(args: {
   orchestrationMutation?: WorkerStartMutation
   /** Settings-driven; the executing host still gets to refuse below. */
   mode: WorkerStartModeReceipt
+  /** Set when a worker starts its own helper; the helper reports to this Dispatch. */
+  helperCreator?: DispatchContextRow
 }): Promise<unknown> {
   const { runtime, db, run, coordinator, callerSession, existingTask, orchestrationMutation } = args
   // `--terminal` names a party: a chat by its `/clear` root address, as its Dispatch does.
@@ -147,11 +149,13 @@ export async function startLocalWorker(args: {
   const started = db.createStartingWorkerDispatch({
     creator: resolveDispatchCreator(runtime, params.from, callerSession),
     maxDepth: runtime.getNestedWorkerMaxDepth(),
+    requiredCreatorDispatchId: args.helperCreator?.id,
     taskId: existingTask?.id,
     taskSpec: params.spec,
     taskTitle: params.taskTitle,
     taskDeps: parseTaskDeps(params.deps),
-    taskParentId: params.parent,
+    // A helper's Task sits under its creator's Task, so the Run's Task tree shows who started it.
+    taskParentId: params.parent ?? args.helperCreator?.task_id,
     taskRunId: run.id,
     // A handle-less session creates root Tasks: Task lineage is recorded by terminal only.
     taskCreatedByTerminalHandle: coordinator?.terminalHandle ?? undefined,

src/main/runtime/rpc/methods/orchestration/worker/worker-start-authority.ts

@@ -0,0 +1,82 @@
+import type { OrchestrationDb } from '../../../../orchestration/db'
+import { OrchestrationError } from '../../../../orchestration/orchestration-error'
+import type {
+  OrchestrationCallerIdentity,
+  OrchestrationSessionCaller
+} from '../../../../orchestration/orchestration-caller-identity'
+import type { DispatchContextRow, RunRow, TaskRow } from '../../../../orchestration/types'
+import type { OrcaRuntimeService } from '../../../../orca-runtime'
+import { resolveDispatchCreator } from '../runs/dispatch-creator'
+
+/**
+ * The Run a worker-start lands in, and who may start it. The Run's coordinator starts workers in
+ * it; a worker on an active Dispatch starts helpers in the Run it works for, as itself, so the
+ * helpers report to it and its own mailbox never moves.
+ */
+export function resolveWorkerStartRun(
+  runtime: OrcaRuntimeService,
+  args: {
+    coordinator: OrchestrationCallerIdentity | null
+    from: string
+    callerSession: OrchestrationSessionCaller | undefined
+    runId: string | undefined
+  }
+): { run: RunRow; helperCreator?: DispatchContextRow } {
+  const db = runtime.getOrchestrationDb()
+  const coordinated = args.coordinator
+    ? db.getCurrentRunForCoordinator(args.coordinator)
+    : undefined
+  if (coordinated) {
+    if (args.runId && args.runId !== coordinated.id) {
+      throw fenced(`This caller coordinates Run ${coordinated.id}, not ${args.runId}.`)
+    }
+    return { run: coordinated }
+  }
+  // Same resolution that stamps the helper's creator, so authority and reporting always agree.
+  // It returns only an active, delegated Dispatch; a remote one has no row on this host.
+  const creatorId = db.resolveCreatorDispatchId(
+    resolveDispatchCreator(runtime, args.from, args.callerSession)
+  )
+  const creator = creatorId ? db.getDispatchContextById(creatorId) : undefined
+  const run = creator ? db.getRun(creator.run_id) : undefined
+  if (!creator || !run || run.legacy === 1) {
+    throw fenced(
+      'worker-start needs the coordinator bound to the Run, or a worker on an active Dispatch in it.'
+    )
+  }
+  if (args.runId && args.runId !== run.id) {
+    throw fenced(
+      `This worker's Dispatch ${creator.id} is in Run ${run.id}; it starts helpers only there.`
+    )
+  }
+  return { run, helperCreator: creator }
+}
+
+/**
+ * What a worker may start: a fresh helper in a terminal Orca launches for it, on this host, or a
+ * retry of its own helper's latest attempt. Never the coordinator's Tasks or another party's pane.
+ */
+export function assertHelperStartScope(
+  db: OrchestrationDb,
+  creator: DispatchContextRow,
+  args: { task: TaskRow | undefined; params: { on?: string; terminal?: string } }
+): void {
+  if (args.params.on || args.params.terminal) {
+    // Why --on: remote reports land in the Run home's Run mailbox, so the helper could not report back.
+    throw new OrchestrationError(
+      'invalid_argument',
+      'A worker starts helpers in a new agent on this host; omit --on and --terminal. No effects were applied.',
+      { effectsApplied: false }
+    )
+  }
+  if (args.task && db.getDispatchContext(args.task.id)?.creator_dispatch_id !== creator.id) {
+    throw new OrchestrationError(
+      'task_not_found',
+      `Task ${args.task.id} was not last run by a helper of Dispatch ${creator.id}; start new helper work with --spec.`
+    )
+  }
+}
+
+function fenced(message: string): OrchestrationError {
+  return new OrchestrationError('consumer_fenced', message)
+}

src/main/runtime/rpc/methods/orchestration/worker/worker-topology.ts

@@ -4,6 +4,7 @@ import type { TuiAgent } from '../../../../../../shared/tui-agent'
 import type { OrcaRuntimeService } from '../../../../orca-runtime'
 import type { OrchestrationDb } from '../../../../orchestration/db'
 import { createStructuredWorkerSession } from '../../orchestration-structured-worker-session'
+import { dispatchReportMailbox } from '../../../../orchestration/dispatch-report-mailbox'
 
 export type WorkerEffect = {
   kind: 'worktree' | 'terminal' | 'setup' | 'dispatch_input'
@@ -175,10 +176,11 @@ export function monitorWorkerSetup(args: {
       if (!evidence.changed) {
         return
       }
+      const dispatch = args.db.getDispatchContextById(args.dispatchId)
       const message = args.db.insertMessage({
         runId: args.runId,
         from: `dispatch:${args.dispatchId}`,
-        to: `run:${args.runId}`,
+        to: dispatch ? dispatchReportMailbox(args.db, dispatch) : `run:${args.runId}`,
         subject: `Setup ${setupState} for worker ${args.dispatchId}`,
         type: 'status',
         priority: setupState === 'failed' ? 'high' : 'normal',

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

@@ -7,6 +7,7 @@ import {
   readWorkerStartModeSettings
 } from '../../orchestration-worker-start-mode'
 import { resolveOrchestrationCaller } from '../runs/run-scope'
+import { assertHelperStartScope, resolveWorkerStartRun } from './worker-start-authority'
 import { WorkerStartParams } from '../../../../../../shared/rpc-contract/orchestration-worker-start-params'
 import {
   isWorkerStartTimeoutWithinTimerLimit,
@@ -38,13 +39,12 @@ export const ORCHESTRATION_WORKER_START_METHODS = [
         callerEvidence: orchestrationCompatibilityEvidence,
         callerSession: orchestrationCaller
       })
-      const run = coordinator ? db.getCurrentRunForCoordinator(coordinator) : undefined
-      if (!run || (params.run && params.run !== run.id)) {
-        throw new OrchestrationError(
-          'consumer_fenced',
-          'worker-start requires the coordinator terminal currently bound to the Task Run.'
-        )
-      }
+      const { run, helperCreator } = resolveWorkerStartRun(runtime, {
+        coordinator,
+        from: params.from,
+        callerSession: orchestrationCaller,
+        runId: params.run
+      })
       const existingTask = params.task ? db.getTask(params.task) : undefined
       if (params.task && (!existingTask || existingTask.run_id !== run.id)) {
         throw new OrchestrationError(
@@ -52,6 +52,9 @@ export const ORCHESTRATION_WORKER_START_METHODS = [
           `Task ${params.task} was not found in Run ${run.id}.`
         )
       }
+      if (helperCreator) {
+        assertHelperStartScope(db, helperCreator, { task: existingTask, params })
+      }
       await assertWorkerStartTaskSpecWithinPromptBudget(params.spec ?? existingTask!.spec)
       const mode = decideWorkerStartMode({
         params,
@@ -88,7 +91,8 @@ export const ORCHESTRATION_WORKER_START_METHODS = [
         callerSession: orchestrationCaller,
         existingTask,
         orchestrationMutation,
-        mode
+        mode,
+        helperCreator
       })
     }
   })

src/main/runtime/structured-chat-worker-helpers.test.ts

@@ -0,0 +1,331 @@
+import './rpc/unused-default-rpc-methods.test-fixture'
+// A dispatched chat starting helper workers on the Run it works for, on the coordinator-mail rig:
+// the helpers report to it, its own mailbox never moves, and whatever it never handled goes back
+// to the Run's coordinator once it settles.
+
+import { describe, expect, it, vi } from 'vitest'
+import { formatOrcaSessionAddress } from '../../shared/orca-session-address'
+import { testOrcaSessionId } from '../../shared/orca-session-address-test-fixture'
+import { idOf, resultOf } from './rpc/orchestration-session-caller-test-fixture'
+import type { DispatchCreator } from './orchestration/db/dispatch-depth'
+import {
+  COORDINATOR,
+  PEER_CHAT,
+  WAIT,
+  WORKER_2_PANE,
+  WORKER_PANE,
+  call,
+  coordinatorRunAndTask,
+  db,
+  dispatcher,
+  openChat,
+  request,
+  restartRuntime,
+  runtime,
+  turnText
+} from './structured-chat-coordinator-mail-rig.test-fixture'
+
+const WORKER = formatOrcaSessionAddress(testOrcaSessionId(PEER_CHAT))
+const HELPER = 'term_worker'
+/** A helper's own helper, two levels below the Run's coordinator. */
+const LEAF = 'term_worker_2'
+
+async function refusal(method: string, params: Record<string, unknown>, sessionId: string) {
+  const response = await dispatcher.dispatch(request(method, params, { sessionId }))
+  if (response.ok) {
+    throw new Error(`expected a refusal, got ${JSON.stringify(response)}`)
+  }
+  return response.error
+}
+
+/** The helper row worker-start reserves once the worker's authority resolved, in a live terminal. */
+function reserveHelper(
+  creatorDispatchId: string,
+  creator: DispatchCreator = { kind: 'session', orcaSessionId: testOrcaSessionId(PEER_CHAT) },
+  handle = HELPER,
+  paneKey = WORKER_PANE
+): { dispatchId: string; taskId: string } {
+  const started = db.createStartingWorkerDispatch({
+    creator,
+    maxDepth: runtime.getNestedWorkerMaxDepth(),
+    requiredCreatorDispatchId: creatorDispatchId,
+    taskSpec: 'review the parser',
+    taskRunId: db.getDispatchContextById(creatorDispatchId)!.run_id,
+    taskParentId: db.getDispatchContextById(creatorDispatchId)!.task_id,
+    startOptions: {}
+  })
+  db.prepareStartingWorkerAuthority({
+    dispatchId: started.dispatch.id,
+    handle,
+    paneKey,
+    processIncarnation: `runtime_test:${handle}:1`,
+    worktreeId: 'repo::worker',
+    effects: [],
+    setupState: 'not_applicable'
+  })
+  db.markWorkerDispatchReady(started.dispatch.id)
+  return { dispatchId: started.dispatch.id, taskId: started.task.id }
+}
+
+function helperSend(type: string, subject: string, helper: { dispatchId: string; taskId: string }) {
+  return call('orchestration.send', {
+    from: HELPER,
+    type,
+    subject,
+    payload: JSON.stringify({ ...helper, outcome: 'succeeded' })
+  })
+}
+
+function inboxOf(inbox: Record<string, unknown>) {
+  const messages = Array.isArray(inbox.messages)
+    ? inbox.messages.map((message: Record<string, unknown>) => ({
+        id: String(message.id),
+        subject: String(message.subject),
+        type: String(message.type)
+      }))
+    : []
+  return { deliveryId: inbox.deliveryId, messages }
+}
+
+/** A chat checks by its session; a terminal helper by its handle. */
+async function check(caller: string, ack?: unknown) {
+  const byTerminal = caller.startsWith('term_')
+  return inboxOf(
+    await call(
+      'orchestration.check',
+      { ...(byTerminal ? { terminal: caller } : {}), ...(ack ? { ack } : {}) },
+      byTerminal ? {} : { sessionId: caller }
+    )
+  )
+}
+
+async function subjects(caller: string, ack?: unknown): Promise<string[]> {
+  return (await check(caller, ack)).messages.map((message) => message.subject)
+}
+
+describe('a worker starting helpers on the Run it works for', () => {
+  it('gets their reports and questions in its own check, and hands the unhandled rest to the Run when it settles', async () => {
+    await openChat(COORDINATOR)
+    await openChat(PEER_CHAT)
+    const { runId, taskId } = await coordinatorRunAndTask()
+    const parent = idOf(
+      (
+        await call(
+          'orchestration.dispatch',
+          { task: taskId, to: WORKER },
+          { sessionId: COORDINATOR }
+        )
+      ).dispatch
+    )
+    vi.spyOn(runtime, 'getNestedWorkerMaxDepth').mockReturnValue(2)
+
+    // The coordinator seat no longer gates it: the worker reaches the helper-scope rules instead.
+    expect(
+      await refusal(
+        'orchestration.workerStart',
+        { run: runId, spec: 'review', terminal: HELPER },
+        PEER_CHAT
+      )
+    ).toMatchObject({ code: 'invalid_argument', message: expect.stringContaining('--terminal') })
+
+    const helper = reserveHelper(parent)
+    expect(db.getDispatchContextById(helper.dispatchId)).toMatchObject({
+      run_id: runId,
+      creator_dispatch_id: parent,
+      depth: 2
+    })
+
+    await helperSend('escalation', 'Blocked on fixtures', helper)
+    const asked = call('orchestration.ask', { from: HELPER, question: 'Which parser?' })
+    await vi.waitFor(
+      () =>
+        expect(db.db.prepare('SELECT COUNT(*) AS n FROM question_threads').get()).toEqual({ n: 1 }),
+      WAIT
+    )
+    const inbox = await check(PEER_CHAT)
+    const messages = inbox.messages
+    expect(messages.map((message) => message.subject)).toEqual(['Blocked on fixtures', 'Question'])
+    await call(
+      'orchestration.reply',
+      { id: messages.find((message) => message.type === 'question')!.id, body: 'the TOML one' },
+      { sessionId: PEER_CHAT }
+    )
+    expect(await asked).toMatchObject({ answer: 'the TOML one' })
+
+    expect(await subjects(COORDINATOR)).toEqual([])
+    expect(await call('orchestration.runCurrent', {}, { sessionId: PEER_CHAT })).toEqual({
+      run: null
+    })
+
+    // The worker's last check acknowledges what it read; what arrives after goes to the Run.
+    expect(await subjects(PEER_CHAT, inbox.deliveryId)).toEqual([])
+    await helperSend('escalation', 'Blocked again', helper)
+    await call(
+      'orchestration.send',
+      {
+        type: 'worker_done',
+        subject: 'Parent done',
+        payload: JSON.stringify({ taskId, dispatchId: parent, outcome: 'succeeded' })
+      },
+      { sessionId: PEER_CHAT }
+    )
+    await helperSend('worker_done', 'Helper done', helper)
+    expect(await subjects(COORDINATOR)).toEqual(['Blocked again', 'Parent done', 'Helper done'])
+  })
+})
+
+/** The Run's coordinator, a dispatched chat, its terminal helper, and that helper's own helper. */
+async function threeLevels() {
+  const chat = await openChat(COORDINATOR)
+  await openChat(PEER_CHAT)
+  const { runId, taskId } = await coordinatorRunAndTask()
+  const parent = idOf(
+    (await call('orchestration.dispatch', { task: taskId, to: WORKER }, { sessionId: COORDINATOR }))
+      .dispatch
+  )
+  vi.spyOn(runtime, 'getNestedWorkerMaxDepth').mockReturnValue(3)
+  const middle = reserveHelper(parent)
+  const leaf = reserveHelper(
+    middle.dispatchId,
+    { kind: 'terminal', handle: HELPER, paneKey: WORKER_PANE },
+    LEAF,
+    WORKER_2_PANE
+  )
+  return { chat, runId, middle, leaf }
+}
+
+/** The coordinator waiting on questions only, as its loop does between steps. */
+async function parkRunQuestionCheck(runId: string) {
+  expect(await subjects(COORDINATOR)).toEqual([])
+  const parked = call(
+    'orchestration.check',
+    { wait: true, types: 'question', timeoutMs: 10_000 },
+    { sessionId: COORDINATOR }
+  )
+  await vi.waitFor(
+    () => expect(runtime.messageWaitersByHandle.has(`run:${runId}`)).toBe(true),
+    WAIT
+  )
+  return { parked }
+}
+
+async function answerFromRun(parked: Promise<Record<string, unknown>>, expected: string[]) {
+  const woke = inboxOf(await parked)
+  expect(woke.messages.map((message) => message.subject)).toEqual(expected)
+  const question = woke.messages.find((message) => message.type === 'question')!
+  await call(
+    'orchestration.reply',
+    { id: question.id, body: 'the TOML one' },
+    { sessionId: COORDINATOR }
+  )
+  return woke
+}
+
+describe('a helper whose creator settles between it and the Run', () => {
+  it('wakes the Run for its question, which the Run answers, and reports arrive once', async () => {
+    const { runId, middle } = await threeLevels()
+    await call('orchestration.send', { from: LEAF, type: 'status', subject: 'Read status' })
+    const asked = call('orchestration.ask', { from: LEAF, question: 'Which parser?' })
+    await vi.waitFor(
+      () => expect(db.getAllMessages(`dispatch:${middle.dispatchId}`)).toHaveLength(2),
+      WAIT
+    )
+    // The middle worker reads both, but never answers.
+    const read = await check(HELPER)
+    expect(read.messages.map((message) => message.subject)).toEqual(['Read status', 'Question'])
+    expect(await subjects(HELPER, read.deliveryId)).toEqual([])
+    await call('orchestration.send', { from: LEAF, type: 'status', subject: 'Unread status' })
+    const { parked } = await parkRunQuestionCheck(runId)
+
+    await helperSend('worker_done', 'Middle done', middle)
+    const woke = await answerFromRun(parked, ['Question', 'Unread status'])
+    expect(await asked).toMatchObject({ answer: 'the TOML one' })
+    expect(await subjects(COORDINATOR, woke.deliveryId)).toEqual([])
+    expect(await subjects(PEER_CHAT)).toEqual(['Middle done'])
+  })
+
+  it('also hands over a question asked while its creator was being stopped', async () => {
+    const { runId, middle } = await threeLevels()
+    const { parked } = await parkRunQuestionCheck(runId)
+    db.beginWorkerStop(middle.dispatchId, 'test-stop')
+    const asked = call('orchestration.ask', { from: LEAF, question: 'Which parser?' })
+    await vi.waitFor(
+      () => expect(db.getAllMessages(`dispatch:${middle.dispatchId}`)).toHaveLength(1),
+      WAIT
+    )
+
+    db.settleWorkerStop(middle.dispatchId)
+    await answerFromRun(parked, ['Question'])
+    expect(await asked).toMatchObject({ answer: 'the TOML one' })
+  })
+
+  it('re-offers the handed-back question to the Run on every replay of its settling report', async () => {
+    const { runId, middle, leaf } = await threeLevels()
+    db.createQuestion({
+      runId,
+      dispatchId: leaf.dispatchId,
+      askerHandle: LEAF,
+      question: 'Which parser?',
+      to: `dispatch:${middle.dispatchId}`
+    })
+    const done = request('orchestration.send', {
+      from: HELPER,
+      type: 'worker_done',
+      subject: 'Middle done',
+      payload: JSON.stringify({ ...middle, outcome: 'succeeded' })
+    })
+    resultOf(await dispatcher.dispatch(done))
+    await new Promise((resolve) => setTimeout(resolve, 0))
+    const countMessages = () => db.db.prepare('SELECT COUNT(*) AS n FROM messages').get()
+    const before = countMessages()
+    const notify = vi.spyOn(runtime, 'notifyMessageArrived')
+
+    // The second replay reads the receipt the first replay rewrote.
+    for (const replays of [1, 2]) {
+      expect(resultOf(await dispatcher.dispatch(done))).toMatchObject({
+        mutation: { replayed: true }
+      })
+      expect(
+        notify.mock.calls.filter(([to, type]) => to === `run:${runId}` && type === 'question')
+      ).toHaveLength(replays)
+    }
+    expect(countMessages()).toEqual(before)
+  })
+
+  it('moves nothing on rollback; a restart offers a committed hand-back nothing woke', async () => {
+    const { chat, runId, middle, leaf } = await threeLevels()
+    const mailbox = `dispatch:${middle.dispatchId}`
+    db.createQuestion({
+      runId,
+      dispatchId: leaf.dispatchId,
+      askerHandle: LEAF,
+      question: 'Which parser?',
+      to: mailbox
+    })
+    await call('orchestration.send', { from: LEAF, type: 'status', subject: 'Unread status' })
+    const ids = db.getAllMessages(mailbox).map((message) => message.id)
+    // The middle worker's pointer for both already landed.
+    db.markAsDelivered(ids, mailbox)
+    const notify = vi.spyOn(runtime, 'notifyMessageArrived')
+
+    db.db.exec('BEGIN IMMEDIATE')
+    db.completeDispatch(middle.dispatchId)
+    db.db.exec('ROLLBACK')
+    await new Promise((resolve) => setTimeout(resolve, 0))
+    expect(db.getAllMessages(mailbox)).toHaveLength(2)
+    expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([])
+    expect(notify).not.toHaveBeenCalledWith(`run:${runId}`, expect.anything())
+
+    // Committed, but the process ends before anything wakes the Run; the old pointer lands late.
+    db.mailHandBackListener = undefined
+    db.completeDispatch(middle.dispatchId)
+    db.markAsDelivered(ids, mailbox)
+    expect(db.getUndeliveredUnreadMessages(`run:${runId}`).map((message) => message.id)).toEqual(
+      expect.arrayContaining(ids)
+    )
+
+    restartRuntime()
+    await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT)
+    expect(turnText(chat.turns[0]!)).toMatch(/orchestration message/)
+  })
+})

src/main/runtime/terminal/layout/terminal-resize-listeners.ts

@@ -21,6 +21,7 @@ import { classifyWorkerTerminalProcessIncarnation } from '../../orchestration/wo
 import { parseWorkerTerminalHostScope } from '../../../../shared/worker-terminal-host-scope'
 import { getRepoIdFromWorktreeId } from '../../../../shared/worktree/id'
 import { buildOrchestrationTaskDisplayMetadata } from '../../../../shared/orchestration-task-display'
+import { dispatchReportMailbox } from '../../orchestration/dispatch-report-mailbox'
 import { LOCAL_EXECUTION_HOST_ID, toSshExecutionHostId } from '../../../../shared/execution-host'
 
 export abstract class OrcaRuntimeWithSubscribeToTerminalResize extends OrcaRuntimeWithApplyMobileDisplayMode {
@@ -98,7 +99,7 @@ export abstract class OrcaRuntimeWithSubscribeToTerminalResize extends OrcaRunti
       const active = this._orchestrationDb.getActiveCoordinatorRun?.()
       const recipient =
         owningRun && owningRun.legacy !== 1
-          ? { to: `run:${owningRun.id}`, runId: owningRun.id }
+          ? { to: dispatchReportMailbox(this._orchestrationDb, dispatch), runId: owningRun.id }
           : active
             ? { to: active.coordinator_handle, runId: undefined }
             : null

src/main/runtime/workspaces/persisted-surfaces/automation-operations.ts

@@ -16,6 +16,10 @@ import {
   type AutomationListResult
 } from '../../../../shared/automation-list-scope'
 import { OrchestrationDb } from '../../orchestration/db'
+import {
+  committedHandBackTypes,
+  type MailHandBack
+} from '../../orchestration/db/messages/helper-mailbox-hand-back'
 import { join } from 'node:path'
 import { existsSync } from 'node:fs'
 import { getAppEnvironment } from '../../../../shared/app-environment'
@@ -150,6 +154,7 @@ export abstract class OrcaRuntimeWithAutomationOperations extends OrcaRuntimeWit
   getOrchestrationDb(): OrchestrationDb {
     if (!this._orchestrationDb) {
       this._orchestrationDb = new OrchestrationDb(this.orchestrationDbPath())
+      this.bindMailHandBack(this._orchestrationDb)
       this.ensureOrchestrationFederationRelay()
       this.scheduleRestoredMessageRepoints()
     }
@@ -171,10 +176,36 @@ export abstract class OrcaRuntimeWithAutomationOperations extends OrcaRuntimeWit
     this.orchestrationFederation.resetForDatabaseChange()
     this.mailPointerRepointScheduler.clear()
     this._orchestrationDb = db
+    this.bindMailHandBack(db)
     this.ensureOrchestrationFederationRelay()
     this.scheduleRestoredMessageRepoints()
   }
 
+  private bindMailHandBack(db: OrchestrationDb): void {
+    db.mailHandBackListener = (handBack) => this.offerHandedBackMail(db, handBack)
+  }
+
+  /** Wakes the Run for helper mail its creator's settlement moved there, once that committed. */
+  private offerHandedBackMail(db: OrchestrationDb, handBack: MailHandBack): void {
+    // Why: a replaced database's consumers are not this runtime's to wake.
+    if (this._orchestrationDb !== db) {
+      return
+    }
+    try {
+      if (db.db.isTransaction) {
+        this.mailPointerRepointScheduler.schedule(handBack.mailbox)
+        return
+      }
+      // Why: re-reading the moved rows means a rolled-back settlement wakes no one.
+      for (const type of committedHandBackTypes(db, handBack)) {
+        this.notifyMessageArrived(handBack.mailbox, type)
+      }
+    } catch (error) {
+      console.warn('[orchestration] failed to offer handed-back helper mail', error)
+      this.mailPointerRepointScheduler.schedule(handBack.mailbox)
+    }
+  }
+
   protected async flushWorkspaceSessionOrThrowAsync(): Promise<void> {
     const store = this.store
     if (store?.flushPendingOrThrowAsync) {