Improve orchestration migration safety for live legacy workers (#11107)

* fix(orchestration): clarify legacy migration safety * fix(cli): sanitize legacy formatted messages * test(runtime): allow near-cap fuzz under shard load --------- Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
2026-09-21 16:02:20 +00:00 · 2026-07-28 00:36:25 -07:00
parent df9f55990b
commit 77d4c64f7a
13 changed files with 302 additions and 25 deletions
@@ -63,6 +63,18 @@ describe('orchestration skill guidance', () => {
    expect(migration).toContain('task-list --run run_legacy_local')
    expect(migration).toContain('Read-only inspection never consumes legacy mail')
    expect(migration).toContain('does not run a legacy scheduler, translate old writes, or drain')
+    expect(migration).toContain('does not cancel the prior assignment')
+    expect(migration).toContain('invalidate its worktree')
+    expect(migration).toContain('discard filesystem changes')
+    expect(migration).toContain('leave it as the only editor in that worktree')
+    expect(migration).toContain('observe it manually, read-only')
+    expect(migration).toContain('until it reaches a stable handoff point')
+    expect(migration).toContain('visible activity is a reason to keep observing')
+    expect(migration).toContain(
+      'Never launch a replacement editor in the same worktree while the legacy worker may still write there.'
+    )
+    expect(migration).toContain('if remaining work needs new lifecycle supervision')
+    expect(migration).not.toContain('restart the work using Run -> Task -> `worker-start`')
  })

  it('treats long-running worker waits as liveness checkpoints, not failures', () => {
@@ -60,22 +60,29 @@ If a command returns `orchestration_migration_required`, `run_required`, or a li
 1. Confirm `effectsApplied` is `false`.
 2. Using the same CLI executable that returned the error, run the returned arguments: `skills get orchestration --full`.
 3. Read the guide completely. Do not retry the rejected command unchanged.
-4. Create or bind a lightweight Run, then restart the work using Run -> Task -> `worker-start`.
-5. Inspect any pre-upgrade terminal before creating replacement work.
+4. Inspect the pre-upgrade Run, terminal, and assigned worktree before deciding whether any work needs replacement.
+5. If the legacy worker is still making valid progress, leave it as the only editor in that worktree and observe it manually, read-only, until it reaches a stable handoff point.
+6. Only then, if remaining work needs new lifecycle supervision, create or bind a lightweight Run, create a Task for the remaining work, and use `worker-start` in a conflict-free placement.

 The arguments intentionally omit an executable name so this works with `orca`, `orca-ide`, `orca-dev`, or another configured Orca CLI command.

-Pre-upgrade terminals and agents are not killed during upgrade, but they are no longer supervised: old heartbeat, question, completion, scheduler, and mutation calls are rejected before effects. Legacy database rows remain available only for explicit inspection:
+The cutover removes lifecycle authority; it does not cancel the prior assignment, invalidate its worktree, discard filesystem changes, or stop the worker process. Pre-upgrade terminals and agents can continue their valid assigned work, but they are no longer supervised by Orca: old heartbeat, question, completion, scheduler, reply, acknowledgment, and mutation calls are rejected before effects.
+
+Legacy database rows and terminal output remain available for explicit read-only inspection:

 ```bash
 orca orchestration run-list --json
 orca orchestration run-show --id run_legacy_local --json
 orca orchestration task-list --run run_legacy_local --json
 orca orchestration inbox --full --json
-orca orchestration check --terminal <legacy_handle> --peek --json
+orca orchestration check --terminal <legacy_handle> --peek --format --json
+orca terminal read --terminal <legacy_handle> --json
+orca terminal wait --terminal <legacy_handle> --for tui-idle --timeout-ms 60000 --json
 ```

-Read-only inspection never consumes legacy mail. Do not use actionable `check`, acknowledgment, send, retry, or task updates against the legacy Run.
+Read-only inspection never consumes legacy mail. A stable handoff point means the worker has become idle, stopped, or completed a coherent edit/test/commit checkpoint; visible activity is a reason to keep observing, not to replace it. Do not prompt the worker to use old lifecycle commands.
+
+Never launch a replacement editor in the same worktree while the legacy worker may still write there. Wait for a stable handoff and preserve its filesystem work; if overlap is truly required, use a separate conflict-free worktree with an explicit plan for transferring existing dirty changes. Do not use actionable `check`, acknowledgment, reply, send, retry, or task updates against the legacy Run.

 ## Ownership

@@ -0,0 +1,142 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const callMock = vi.fn()
+
+vi.mock('../format', () => ({ printResult: vi.fn() }))
+vi.mock('../selectors', () => ({ getTerminalHandle: vi.fn() }))
+
+import { printResult } from '../format'
+import { ORCHESTRATION_HANDLERS } from './orchestration'
+
+beforeEach(() => {
+  callMock.mockReset()
+  vi.mocked(printResult).mockReset()
+})
+
+describe('legacy orchestration CLI inspection', () => {
+  it('labels legacy rows in plain check output', async () => {
+    const result = {
+      messages: [
+        {
+          id: 'msg_legacy',
+          run_id: 'run_legacy_local',
+          from_handle: 'term_worker',
+          subject: 'progress',
+          type: 'status'
+        }
+      ],
+      count: 1
+    }
+    callMock.mockResolvedValue({ result })
+
+    await ORCHESTRATION_HANDLERS['orchestration check']({
+      flags: new Map<string, string | boolean>([
+        ['terminal', 'term_coord'],
+        ['peek', true]
+      ]),
+      client: { call: callMock },
+      cwd: '/repo',
+      json: false
+    } as never)
+
+    const formatter = vi.mocked(printResult).mock.calls[0]?.[2]
+    expect(formatter?.(result)).toContain('msg_legacy [legacy, read-only]')
+  })
+
+  it('rebuilds legacy formatted output without runtime-supplied actions', async () => {
+    const result = {
+      messages: [
+        {
+          id: 'msg_legacy',
+          run_id: 'run_legacy_local',
+          from_handle: 'term_worker',
+          subject: 'progress',
+          type: 'status',
+          body: 'Tests are running.',
+          payload: '{"phase":"testing"}'
+        }
+      ],
+      count: 1,
+      formatted: '[Reply: orca orchestration reply --id msg_legacy --from term_coord --body "..."]'
+    }
+    callMock.mockResolvedValue({ result })
+
+    await ORCHESTRATION_HANDLERS['orchestration check']({
+      flags: new Map<string, string | boolean>([
+        ['terminal', 'term_coord'],
+        ['peek', true],
+        ['format', true]
+      ]),
+      client: { call: callMock },
+      cwd: '/repo',
+      json: false
+    } as never)
+
+    const formatter = vi.mocked(printResult).mock.calls[0]?.[2]
+    const output = formatter?.(result)
+    expect(output).toContain('msg_legacy [legacy, read-only]')
+    expect(output).toContain('Tests are running.')
+    expect(output).toContain('[payload] {"phase":"testing"}')
+    expect(output).not.toContain('[Reply:')
+    expect(output).not.toContain('orchestration reply')
+  })
+
+  it('preserves runtime formatting when every message belongs to a current Run', async () => {
+    const result = {
+      messages: [
+        {
+          id: 'msg_current',
+          run_id: 'run_current',
+          from_handle: 'term_worker',
+          subject: 'question'
+        }
+      ],
+      count: 1,
+      formatted: '[Reply: current Run action]'
+    }
+    callMock.mockResolvedValue({ result })
+
+    await ORCHESTRATION_HANDLERS['orchestration check']({
+      flags: new Map<string, string | boolean>([
+        ['terminal', 'term_coord'],
+        ['peek', true],
+        ['format', true]
+      ]),
+      client: { call: callMock },
+      cwd: '/repo',
+      json: false
+    } as never)
+
+    const formatter = vi.mocked(printResult).mock.calls[0]?.[2]
+    expect(formatter?.(result)).toBe(result.formatted)
+  })
+
+  it('labels legacy rows in full inbox output without hiding their body', async () => {
+    const result = {
+      messages: [
+        {
+          id: 'msg_legacy',
+          run_id: 'run_legacy_local',
+          from_handle: 'term_worker',
+          to_handle: 'term_coord',
+          subject: 'progress',
+          body: 'Tests are running.'
+        }
+      ],
+      count: 1
+    }
+    callMock.mockResolvedValue({ result })
+
+    await ORCHESTRATION_HANDLERS['orchestration inbox']({
+      flags: new Map<string, string | boolean>([['full', true]]),
+      client: { call: callMock },
+      cwd: '/repo',
+      json: false
+    } as never)
+
+    const formatter = vi.mocked(printResult).mock.calls[0]?.[2]
+    const output = formatter?.(result)
+    expect(output).toContain('msg_legacy [legacy, read-only]')
+    expect(output).toContain('Tests are running.')
+  })
+})
@@ -22,6 +22,7 @@ import type {
 import type { NativeChatMessage } from '../../shared/native-chat-types'
 import type { RuntimeTerminalRead } from '../../shared/runtime-types'
 import {
+  ORCHESTRATION_LEGACY_RUN_ID,
  orchestrationMigrationData,
  orchestrationSkillRecoveryData
 } from '../../shared/orchestration-rpc-contract'
@@ -76,6 +77,7 @@ const TASK_STATUS_VALUES = [

 type MessageSummary = {
  id: string
+  run_id?: string
  from_handle: string
  to_handle?: string
  subject: string
@@ -85,6 +87,31 @@ type MessageSummary = {
  read?: number
 }

+function formatMessageReadOnlyTag(message: MessageSummary): string {
+  return message.run_id === ORCHESTRATION_LEGACY_RUN_ID ? ' [legacy, read-only]' : ''
+}
+
+function isLegacyReadOnlyMessage(message: MessageSummary): boolean {
+  return message.run_id === ORCHESTRATION_LEGACY_RUN_ID
+}
+
+function formatLegacyAwareCheckMessages(messages: MessageSummary[]): string {
+  return messages
+    .map((message) => {
+      const lines = [
+        `${message.id}${formatMessageReadOnlyTag(message)} [${message.type ?? 'status'}] from=${message.from_handle} "${message.subject}"`
+      ]
+      if (message.body) {
+        lines.push(message.body)
+      }
+      if (message.payload) {
+        lines.push(`[payload] ${message.payload}`)
+      }
+      return lines.join('\n')
+    })
+    .join('\n\n')
+}
+
 type LifecycleSendRejection = {
  action: 'rejected'
  code: string
@@ -617,7 +644,9 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
    }
    printResult(result, json, (r) => {
      if (r.formatted) {
-        return r.formatted
+        return r.messages.some(isLegacyReadOnlyMessage)
+          ? formatLegacyAwareCheckMessages(r.messages)
+          : r.formatted
      }
      if (r.count === 0) {
        if (r.timedOut) {
@@ -631,7 +660,10 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
        return 'No messages.'
      }
      const rendered = r.messages
-        .map((m) => `${m.id} [${m.type ?? 'status'}] from=${m.from_handle} "${m.subject}"`)
+        .map(
+          (m) =>
+            `${m.id}${formatMessageReadOnlyTag(m)} [${m.type ?? 'status'}] from=${m.from_handle} "${m.subject}"`
+        )
        .join('\n')
      return r.deliveryId ? `Delivery ${r.deliveryId}\n${rendered}` : rendered
    })
@@ -669,7 +701,7 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
      // Why: default output omits body/payload for at-a-glance sweeps; --full prints them for auditing.
      return r.messages
        .map((m) => {
-          const head = `${m.id} ${m.from_handle} -> ${m.to_handle ?? '?'}: "${m.subject}"`
+          const head = `${m.id}${formatMessageReadOnlyTag(m)} ${m.from_handle} -> ${m.to_handle ?? '?'}: "${m.subject}"`
          if (!full) {
            return head
          }
@@ -31,6 +31,7 @@ import type {
  FederationRelayItemRow
 } from './types'
 import { buildOrchestrationTaskDisplayMetadata } from '../../../shared/orchestration-task-display'
+import { ORCHESTRATION_LEGACY_RUN_ID } from '../../../shared/orchestration-rpc-contract'
 import { parsePaneKey } from '../../../shared/stable-pane-id'
 import { OrchestrationError } from './orchestration-error'

@@ -140,7 +141,7 @@ function exposeQuestionTimestamps(question: QuestionRow): QuestionRow {
  }
 }

-export const LEGACY_RUN_ID = 'run_legacy_local'
+export const LEGACY_RUN_ID = ORCHESTRATION_LEGACY_RUN_ID

 // Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state.
 const SCHEMA_VERSION = 17
@@ -77,6 +77,17 @@ describe('formatMessageBanner', () => {
    )
  })

+  it('marks legacy messages read-only without reply or acknowledgment affordances', () => {
+    const banner = formatMessageBanner(
+      makeMessage({ id: 'msg_legacy', run_id: 'run_legacy_local' })
+    )
+
+    expect(banner).toContain('[LEGACY READ-ONLY]')
+    expect(banner).toContain('[Inspection only: reply and acknowledgment are unavailable.]')
+    expect(banner).not.toContain('[Reply:')
+    expect(banner).not.toContain('orchestration reply')
+  })
+
  it('ends with a separator line', () => {
    const banner = formatMessageBanner(makeMessage())
    const lines = banner.split('\n')
@@ -1,21 +1,23 @@
 import type { MessageRow } from './types'
+import { ORCHESTRATION_LEGACY_RUN_ID } from '../../../shared/orchestration-rpc-contract'

 const BANNER_WIDTH = 60
 const SEPARATOR = '─'.repeat(BANNER_WIDTH)

-// Why: rich message banners help agents (and humans reading terminal output)
-// quickly parse message metadata. Priority indicators surface urgent messages
-// visually. The reply hint reduces friction for agent-to-agent responses
-// (Section 4.8).
 export function formatMessageBanner(msg: MessageRow): string {
  const priorityTag =
    msg.priority === 'urgent' ? ' [URGENT]' : msg.priority === 'high' ? ' [HIGH]' : ''
+  const legacyReadOnly = msg.run_id === ORCHESTRATION_LEGACY_RUN_ID
+  const authorityTag = legacyReadOnly ? ' [LEGACY READ-ONLY]' : ''
  const senderName = msg.from_handle.toUpperCase()

-  const header = `──── From: ${senderName} (${msg.from_handle})${priorityTag} (${msg.type}) ────`
+  const header = `──── From: ${senderName} (${msg.from_handle})${priorityTag}${authorityTag} (${msg.type}) ────`

  const lines: string[] = [header]
  lines.push(`Subject: ${msg.subject}`)
+  if (legacyReadOnly) {
+    lines.push('[Inspection only: reply and acknowledgment are unavailable.]')
+  }

  if (msg.body) {
    lines.push(msg.body)
@@ -25,11 +27,12 @@ export function formatMessageBanner(msg: MessageRow): string {
    lines.push(`[Payload: ${msg.payload}]`)
  }

-  // Why: injected reply commands must retain the receiving pane's identity
-  // even when an older shell lacks Orca's terminal environment variables.
-  lines.push(
-    `[Reply: orca orchestration reply --id ${msg.id} --from ${msg.to_handle} --body "..."]`
-  )
+  if (!legacyReadOnly) {
+    // Why: older shells can lack Orca's terminal identity environment.
+    lines.push(
+      `[Reply: orca orchestration reply --id ${msg.id} --from ${msg.to_handle} --body "..."]`
+    )
+  }
  lines.push(SEPARATOR)

  return lines.join('\n')
@@ -49,6 +49,54 @@ describe('orchestration migration behavior', () => {
    expect(db.getTask(task.id)?.status).toBe('ready')
  })

+  it('formats legacy terminal inspection as read-only without consuming mail', async () => {
+    const { db, runtime } = createRuntime()
+    const message = db.insertMessage({
+      from: 'term_worker',
+      to: 'term_coord',
+      subject: 'still working',
+      body: 'Tests are running.'
+    })
+    const check = ORCHESTRATION_METHODS.find((method) => method.name === 'orchestration.check')!
+
+    const inspected = (await check.handler(
+      check.params!.parse({ terminal: 'term_coord', peek: true, format: true }),
+      { runtime }
+    )) as { count: number; formatted: string }
+
+    expect(inspected.count).toBe(1)
+    expect(inspected.formatted).toContain('[LEGACY READ-ONLY]')
+    expect(inspected.formatted).toContain('Tests are running.')
+    expect(inspected.formatted).not.toContain('[Reply:')
+    expect(db.getMessageById(message.id)?.read).toBe(0)
+  })
+
+  it('rejects replies to legacy mail without marking or inserting rows', async () => {
+    const { db, runtime } = createRuntime()
+    const message = db.insertMessage({
+      from: 'term_worker',
+      to: 'term_coord',
+      subject: 'legacy question'
+    })
+    const reply = ORCHESTRATION_METHODS.find((method) => method.name === 'orchestration.reply')!
+
+    await expect(
+      reply.handler(
+        reply.params!.parse({
+          id: message.id,
+          body: 'replacement started',
+          from: 'term_coord'
+        }),
+        { runtime }
+      )
+    ).rejects.toMatchObject({
+      code: 'legacy_read_only',
+      data: { effectsApplied: false }
+    })
+    expect(db.getMessageById(message.id)?.read).toBe(0)
+    expect(db.getInbox(100)).toHaveLength(1)
+  })
+
  it('rejects a pre-contract worker_done before message or lifecycle mutation', async () => {
    const { db, runtime } = createRuntime()
    const run = db.createRun({
@@ -1529,17 +1529,25 @@ describe('orchestration RPC methods', () => {
  describe('orchestration.reply', () => {
    it('replies to a message', async () => {
      setup()
-      const original = db.insertMessage({ from: 'a', to: 'b', subject: 'question' })
+      const original = db.insertMessage({
+        from: 'a',
+        to: 'b',
+        subject: 'question',
+        runId: activeRunId
+      })

      const result = (await call('orchestration.reply', {
        id: original.id,
        body: 'answer',
        from: 'b'
-      })) as { message: { to_handle: string; subject: string; thread_id: string } }
+      })) as {
+        message: { to_handle: string; subject: string; thread_id: string; run_id: string }
+      }

      expect(result.message.to_handle).toBe('a')
      expect(result.message.subject).toBe('Re: question')
      expect(result.message.thread_id).toBe(original.id)
+      expect(result.message.run_id).toBe(activeRunId)
    })

    it('throws on nonexistent message', async () => {
@@ -9,7 +9,10 @@ import { formatMessageBanner } from '../../orchestration/formatter'
 import { isGroupAddress, resolveGroupAddress } from '../../orchestration/groups'
 import { reconcileLifecycleMessage } from '../../orchestration/lifecycle-reconciliation'
 import { abbreviateOrchestrationTasks } from '../../../../shared/orchestration-task-summary'
-import { orchestrationSkillRecoveryData } from '../../../../shared/orchestration-rpc-contract'
+import {
+  ORCHESTRATION_LEGACY_RUN_ID,
+  orchestrationSkillRecoveryData
+} from '../../../../shared/orchestration-rpc-contract'
 import { clampOrchestrationAskTimeoutMs } from '../../../../shared/orchestration-ask-timeout'
 import { ORCHESTRATION_GATE_METHODS } from './orchestration-gates'
 import { ORCHESTRATION_RUN_METHODS } from './orchestration-runs'
@@ -899,6 +902,13 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
      if (!original) {
        throw new Error(`Message not found: ${params.id}`)
      }
+      if (original.run_id === ORCHESTRATION_LEGACY_RUN_ID) {
+        throw new OrchestrationError(
+          'legacy_read_only',
+          'Legacy orchestration messages are inspect-only; no reply was applied.',
+          { effectsApplied: false }
+        )
+      }

      const question = db.getQuestion(params.id)
      if (question) {
@@ -943,7 +953,8 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
        to: original.from_handle,
        subject: `Re: ${original.subject}`,
        body: params.body,
-        threadId: original.thread_id ?? original.id
+        threadId: original.thread_id ?? original.id,
+        runId: original.run_id
      })

      runtime.notifyMessageArrived(original.from_handle, reply.type)
@@ -475,7 +475,7 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization
      expectEquivalent(data, { seq: 88_888, rawLength: data.length }, `fuzz-cap seq trial=${trial}`)
      expectEquivalent(data, { seq: 88_888 }, `fuzz-cap delayed trial=${trial}`)
    }
-  }, 15_000)
+  }, 30_000)

  it('keeps every emitted frame within the wire cap and reassembles to the input', () => {
    const data = `${'a'.repeat(200 * 1024)}${SURROGATE_PAIR.repeat(4096)}${LONE_HIGH}`
@@ -13,6 +13,8 @@ export const ORCHESTRATION_SKILL_COMMAND_ARGS = [
  '--full'
 ] as const

+export const ORCHESTRATION_LEGACY_RUN_ID = 'run_legacy_local'
+
 const ORCHESTRATION_MUTATION_METHODS = new Set([
  'orchestration.runCreate',
  'orchestration.runUse',