[STA-5729] Prevent live coordinator takeover of orchestration Runs

config/scripts/orchestration-skill-guidance.test.mjs

@@ -51,6 +51,40 @@ describe('orchestration skill guidance', () => {
     )
   })
 
+  it('defines coordinator authority and ordinary takeover for agents', () => {
+    const skill = readSkill()
+    const authority = getSection(skill, 'Run Coordinator Authority')
+
+    expect(authority).toContain('sole task-graph writer and consuming mailbox reader')
+    expect(authority).toContain('per-Run authority')
+    expect(authority).toContain('not an operating-system or global Orca privilege')
+    expect(authority).toContain('`run-use` is an authority claim, not a read-only selection')
+    expect(authority).toContain('`coordinatorStatus` is `live` or `unverifiable`')
+    expect(authority).toContain('`claimantStatus` is `changed`')
+    expect(authority).toContain('rejects the claim with `effectsApplied: false`')
+    expect(authority).toContain('a proven `exited` incumbent permits the claim to succeed')
+    expect(authority).toContain('`live`')
+    expect(authority).toContain('`unverifiable`')
+    expect(authority).toContain('`exited`')
+    expect(authority).toContain('Loss of contact is never evidence of exit')
+    expect(authority).toContain('No live-to-live transfer command exists')
+    expect(authority).toContain('`--takeover-legacy` is not a force override for ordinary Runs')
+    expect(authority).toContain('task-create')
+    expect(authority).toContain('task-update')
+    expect(authority).toContain('worker-start')
+    expect(authority).toContain('gate-create')
+    expect(authority).toContain('reply')
+    expect(authority).toContain('check')
+    expect(authority).toContain('run-show --id <run_id> --json')
+    expect(authority).toContain('binding.currentConsumer')
+    expect(authority).toContain('never proves that the Run is unowned')
+    expect(authority).toContain('stop or exit the owning coordinator process')
+    expect(skill).not.toContain('unless impersonating another terminal')
+    expect(skill).toContain(
+      'A declared handle is routing input, never proof of coordinator authority'
+    )
+  })
+
   it('teaches attested adoption without reviving the retired scheduler', () => {
     const skill = readSkill()
     const migration = getSection(skill, 'Contract Migration')

skill-guides/orchestration.md

@@ -99,6 +99,32 @@ Takeover fences only the old coordinator, binds the current one, and moves pendi
 
 Do not launch a replacement editor merely because the desktop app or runtime was updated. If adoption cannot prove continuing authority, keep the original worker as the only editor until it reaches a stable handoff point, then use a new current Dispatch in a conflict-free placement for any remaining work.
 
+## Run Coordinator Authority
+
+Orca orchestration is agent-operated. A Run therefore has one current coordinator agent, not a human operator. That coordinator is the sole task-graph writer and consuming mailbox reader for coordinator-side operations; workers retain only their exact Dispatch capabilities for heartbeat, questions, escalation, and `worker_done`. This is per-Run authority, not an operating-system or global Orca privilege.
+
+The single-writer rule prevents two agents from changing Task, Dispatch, worker, or gate state concurrently and prevents two consumers from acknowledging different views of the same FIFO Delivery. The current coordinator may create and update Tasks (`task-create`, `task-update`), start or dispatch workers (`worker-start`, `dispatch`), create and resolve gates (`gate-create`, `gate-resolve`), answer worker questions (`reply`), and consume or acknowledge Run mail (`check`). Explicit `run-show`, `task-list --run`, `gate-list --run`, and inbox calls remain read-only for non-owners and headless callers. `check --peek` is non-consuming, but an ordinary current Run still requires its coordinator authority; do not use it as a non-owner inspection path.
+
+`run-use` is an authority claim, not a read-only selection. When a transfer is intended, make one claim from the stable replacement agent terminal with `orca orchestration run-use --id <run_id> --json`; do not guess whether its process is a remint or spoof the owner with `--from`. Orca either preserves same-process authority, grants replacement authority after proving exit, or rejects the claim with `effectsApplied: false`.
+
+On rejection, read `error.data.coordinatorStatus`, `claimantStatus`, `nextSteps`, and any exact command-argument fields. `coordinatorStatus` is `live` or `unverifiable` because a proven `exited` incumbent permits the claim to succeed. If `claimantStatus` is `changed`, the invoking agent changed during proof; follow the returned retry arguments once from one stable process instead of treating it as a network verdict or retrying blindly. Never infer authority from a terminal handle alone.
+
+- Same coordinator process, reminted handle: authority and any outstanding Delivery are preserved without advancing the consumer generation.
+- Different process and `live` incumbent: `consumer_fenced`; continue from the owning coordinator terminal. To transfer intentionally, stop or exit the owning coordinator process before retrying from its replacement.
+- Different process and `unverifiable` incumbent: `consumer_fenced`; restore connectivity to the owning host. Loss of contact is never evidence of exit, including SSH, relay, Windows, WSL, and federated runtimes.
+- Different process and `exited` incumbent: the replacement may run `orca orchestration run-use --id <run_id> --json`. Orca advances the consumer generation, fences the old outstanding Delivery, preserves pending Run mail and worker assignments, and grants the replacement coordinator authority.
+
+Inspect without claiming authority:
+
+```bash
+orca orchestration run-show --id <run_id> --json
+orca orchestration task-list --run <run_id> --json
+```
+
+`binding.currentConsumer` is `true` only for the current coordinator. `false` permits inspection, not coordinator mutations or consuming `check` calls, and never proves that the Run is unowned.
+
+No live-to-live transfer command exists. A seamless live handoff would require a separate owner-authorized protocol; do not simulate one by retrying, changing `--from`, or replacing a terminal handle. `--takeover-legacy` is not a force override for ordinary Runs: it is limited to the automatically adopted legacy Run described above.
+
 ## Ownership
 
 New orchestration messages and tasks belong to one explicitly bound Run. A Run is only a durable namespace and coordinator inbox; it never schedules or places workers. Lifecycle authority comes from the active Dispatch, and terminal handles remain routing metadata rather than durable identity. Send `worker_done` and `heartbeat` from the worker's own terminal; Orca routes them to that Dispatch's Run.
@@ -135,7 +161,7 @@ orca orchestration inbox [--limit <n>] [--json]
 
 Rules:
 
-- Omit `--from` unless impersonating another terminal; Orca auto-resolves it from the current terminal.
+- Omit `--from` unless an injected preamble or exact recovery command supplies it; Orca normally resolves the current terminal. A declared handle is routing input, never proof of coordinator authority or a transfer mechanism.
 - A coordinator `check` returns the bound Run's oldest FIFO Delivery (up to 50 messages) and replays that exact batch until `--ack <delivery_id>`. Process every message before acknowledging; `check --ack <id> --wait` acknowledges, checks, and waits in one operation.
 - Use `--peek` and `--all` only for read-only history/debugging. Type filters decide when a waiter wakes; the returned actionable Delivery is still the oldest full batch.
 - Use `dispatch:<id>` for coordinator guidance to one supervised worker. Orca routes that stable address locally or through the connected-server relay; do not substitute a remote terminal handle.
@@ -198,7 +224,9 @@ Two limits worth knowing:
 
 - **It is a guardrail, not a security boundary.** A caller that declares another terminal's
   handle while its own launch evidence is unverifiable (an ordinary restored terminal, for
-  example) can be counted as that terminal instead. Orca does not treat workers as hostile.
+  example) can be counted as that terminal instead. This affects only nesting-depth
+  classification; it never grants coordinator or Dispatch authority. Orca does not treat
+  workers as hostile.
 - **It applies while a Dispatch is active.** After `worker_done`, or after a coordinator
   settles the task, the terminal is no longer a worker and is counted as a root again. The
   process may still be alive; that is the documented boundary, not an accident.

src/cli/handlers/orchestration-gate-cli.test.ts

@@ -158,7 +158,7 @@ describe('orchestration gate commands carry caller identity', () => {
     )
   })
 
-  it('inspects a named Run without resolving a caller terminal', async () => {
+  it('inspects a named Run without requiring a caller terminal', async () => {
     // Why: read-only inspection must stay reachable from a pane with no bound Run.
     getTerminalHandleMock.mockRejectedValue(
       new RuntimeClientError('no_active_terminal', 'no active terminal')
@@ -167,17 +167,22 @@ describe('orchestration gate commands carry caller identity', () => {
       callMock,
       okFixture('req_list', {
         gates: [{ id: 'gate_1', task_id: 'task_1', question: 'ship?', status: 'pending' }],
-        count: 1
+        count: 1,
+        runId: 'run_adopted',
+        binding: { currentConsumer: false }
       })
     )
 
-    await main(['orchestration', 'gate-list', '--run', 'run_adopted', '--json'], '/tmp/repo')
+    await main(['orchestration', 'gate-list', '--run', 'run_adopted'], '/tmp/repo')
 
     expect(process.exitCode).toBe(0)
-    expect(getTerminalHandleMock).not.toHaveBeenCalled()
+    expect(getTerminalHandleMock).toHaveBeenCalled()
     expect(paramsFor('orchestration.gateList')).toEqual(
       expect.objectContaining({ run: 'run_adopted', from: undefined })
     )
+    expect(logSpy.mock.calls.map((call) => String(call[0])).join('\n')).toContain(
+      'listed read-only'
+    )
   })
 
   it('fails an unbound gate-create with an actionable error and no mutation', async () => {

src/cli/handlers/orchestration-run-cli.test.ts

@@ -119,6 +119,75 @@ describe('lightweight Run CLI handlers', () => {
       takeoverLegacy: true
     })
   })
+
+  it('discloses a non-owner run-show as read-only', async () => {
+    const response = {
+      result: {
+        run: {
+          id: 'run_1',
+          objective: 'Inspect work',
+          consumer_generation: 1,
+          legacy: 0,
+          created_at: '2026-08-27 20:00:00'
+        },
+        binding: { currentConsumer: false }
+      }
+    }
+    callMock.mockResolvedValue(response)
+    vi.mocked(printResult).mockClear()
+
+    await ORCHESTRATION_HANDLERS['orchestration run-show']({
+      flags: new Map([['id', 'run_1']]),
+      client: { call: callMock },
+      cwd: '/tmp/repo',
+      json: false
+    } as never)
+
+    expect(callMock).toHaveBeenCalledWith('orchestration.runShow', {
+      id: 'run_1',
+      from: 'term_coord'
+    })
+    const format = vi.mocked(printResult).mock.calls[0]?.[2] as (result: {
+      run: {
+        id: string
+        objective: string
+        consumer_generation: number
+        legacy: number
+        created_at: string
+      }
+      binding: { currentConsumer: boolean }
+    }) => string
+    expect(format(response.result)).toContain('shown read-only')
+  })
+
+  it('discloses headless run-show inspection as read-only', async () => {
+    delete process.env.ORCA_TERMINAL_HANDLE
+    getTerminalHandleMock.mockRejectedValue({ code: 'no_active_terminal' })
+    callMock.mockResolvedValue({
+      result: {
+        run: {
+          id: 'run_1',
+          objective: 'Inspect headlessly',
+          consumer_generation: 1,
+          legacy: 0,
+          created_at: '2026-08-27 20:00:00'
+        },
+        binding: { currentConsumer: false }
+      }
+    })
+
+    await ORCHESTRATION_HANDLERS['orchestration run-show']({
+      flags: new Map([['id', 'run_1']]),
+      client: { call: callMock },
+      cwd: '/tmp/repo',
+      json: true
+    } as never)
+
+    expect(callMock).toHaveBeenCalledWith('orchestration.runShow', {
+      id: 'run_1',
+      from: undefined
+    })
+  })
 })
 
 describe('orchestration reset CLI handler', () => {
@@ -174,6 +243,69 @@ describe('orchestration reset CLI handler', () => {
 })
 
 describe('orchestration task-list brief output', () => {
+  it('keeps explicit Run inspection available without an active terminal', async () => {
+    delete process.env.ORCA_TERMINAL_HANDLE
+    getTerminalHandleMock.mockRejectedValue({ code: 'no_active_terminal' })
+    callMock.mockReset().mockResolvedValue({
+      result: {
+        tasks: [],
+        count: 0,
+        runId: 'run_1',
+        binding: { currentConsumer: false }
+      }
+    })
+
+    await ORCHESTRATION_HANDLERS['orchestration task-list']({
+      flags: new Map([['run', 'run_1']]),
+      client: { call: callMock },
+      cwd: '/tmp/repo',
+      json: true
+    } as never)
+
+    expect(callMock).toHaveBeenCalledWith(
+      'orchestration.taskList',
+      expect.objectContaining({ run: 'run_1', callerTerminalHandle: undefined })
+    )
+  })
+
+  it('discloses non-owner inspection even when the Run has no tasks', async () => {
+    process.env.ORCA_TERMINAL_HANDLE = 'term_inspector'
+    callMock.mockReset().mockResolvedValue({
+      result: {
+        tasks: [],
+        count: 0,
+        runId: 'run_1',
+        binding: { currentConsumer: false }
+      }
+    })
+    vi.mocked(printResult).mockClear()
+
+    await ORCHESTRATION_HANDLERS['orchestration task-list']({
+      flags: new Map([['run', 'run_1']]),
+      client: { call: callMock },
+      json: false
+    } as never)
+
+    expect(callMock).toHaveBeenCalledWith(
+      'orchestration.taskList',
+      expect.objectContaining({ run: 'run_1', callerTerminalHandle: 'term_inspector' })
+    )
+    const format = vi.mocked(printResult).mock.calls[0]?.[2] as (result: {
+      count: number
+      runId: string
+      tasks: never[]
+      binding: { currentConsumer: boolean }
+    }) => string
+    expect(
+      format({
+        count: 0,
+        runId: 'run_1',
+        tasks: [],
+        binding: { currentConsumer: false }
+      })
+    ).toContain('not bound to this terminal')
+  })
+
   it('requests server-side brief and falls back client-side for older runtimes', async () => {
     callMock.mockReset().mockResolvedValue({
       result: {

src/cli/handlers/orchestration-terminal-identity.test.ts

@@ -0,0 +1,45 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { RuntimeClientError } from '../runtime-client'
+import { resolveOrchestrationTerminalHandle } from './orchestration/terminal-identity'
+
+const originalTerminalHandle = process.env.ORCA_TERMINAL_HANDLE
+const originalPaneKey = process.env.ORCA_PANE_KEY
+
+afterEach(() => {
+  restoreEnv('ORCA_TERMINAL_HANDLE', originalTerminalHandle)
+  restoreEnv('ORCA_PANE_KEY', originalPaneKey)
+})
+
+describe('orchestration terminal identity', () => {
+  it('refreshes caller evidence after resolving a stale handle by pane', async () => {
+    process.env.ORCA_TERMINAL_HANDLE = 'term_stale'
+    process.env.ORCA_PANE_KEY = 'tab_1:leaf_1'
+    const call = vi
+      .fn()
+      .mockRejectedValueOnce(new RuntimeClientError('terminal_handle_stale', 'stale'))
+      .mockResolvedValueOnce({ result: { terminal: { handle: 'term_reminted' } } })
+    const refresh = vi.fn()
+
+    const handle = await resolveOrchestrationTerminalHandle(
+      new Map(),
+      '/tmp/repo',
+      {
+        call,
+        refreshOrchestrationCallerHandleAfterPaneRemint: refresh
+      } as never,
+      'from',
+      { validateEnvHandle: true }
+    )
+
+    expect(handle).toBe('term_reminted')
+    expect(refresh).toHaveBeenCalledWith('term_stale', 'tab_1:leaf_1', 'term_reminted')
+  })
+})
+
+function restoreEnv(name: string, value: string | undefined): void {
+  if (value === undefined) {
+    delete process.env[name]
+  } else {
+    process.env[name] = value
+  }
+}

src/cli/handlers/orchestration/gate-handlers.ts

@@ -2,7 +2,10 @@ import type { CommandHandler } from '../../dispatch'
 import { printResult } from '../../format'
 import { getOptionalStringFlag, getRequiredStringFlag } from '../../flags'
 import { callOrchestrationMutation } from './mutation-request'
-import { resolveCoordinatorTerminalHandle } from './terminal-identity'
+import {
+  resolveCoordinatorTerminalHandle,
+  resolveOptionalCoordinatorTerminalHandle
+} from './terminal-identity'
 
 export const ORCHESTRATION_GATE_HANDLERS: Record<string, CommandHandler> = {
   'orchestration gate-create': async ({ flags, client, cwd, json }) => {
@@ -37,24 +40,32 @@ export const ORCHESTRATION_GATE_HANDLERS: Record<string, CommandHandler> = {
   'orchestration gate-list': async ({ flags, client, cwd, json }) => {
     const run = getOptionalStringFlag(flags, 'run')
     // Why: named runs remain inspectable without a pane; only implicit runs resolve identity.
-    const from = run ? undefined : await resolveCoordinatorTerminalHandle(flags, cwd, client)
+    const from = run
+      ? await resolveOptionalCoordinatorTerminalHandle(flags, cwd, client)
+      : await resolveCoordinatorTerminalHandle(flags, cwd, client)
     const result = await client.call<{
       gates: { id: string; task_id: string; question: string; status: string }[]
       count: number
       runId?: string
+      binding?: { currentConsumer: boolean }
     }>('orchestration.gateList', {
       task: getOptionalStringFlag(flags, 'task'),
       status: getOptionalStringFlag(flags, 'status'),
       run,
       from
     })
     printResult(result, json, (value) => {
+      const nonOwnerNotice =
+        value.binding?.currentConsumer === false
+          ? `Run ${value.runId} is not bound to this terminal; listed read-only. Mutations require the owning coordinator.`
+          : undefined
       if (value.gates.length === 0) {
-        return 'No gates found.'
+        return nonOwnerNotice ? `${nonOwnerNotice}\nNo gates found.` : 'No gates found.'
       }
-      return value.gates
+      const gates = value.gates
         .map((gate) => `${gate.id} task=${gate.task_id} [${gate.status}] "${gate.question}"`)
         .join('\n')
+      return nonOwnerNotice ? `${nonOwnerNotice}\n${gates}` : gates
     })
   }
 }