fix(orchestration): preserve task reset lineage per Run (#28823)

* fix(orchestration): preserve task reset lineage per Run

STA-10382

* fix(orchestration): record removed Tasks, Dispatches and workers per entity

Replace the per-Run capped JSON snapshot with one row per removed entity in
removed_orchestration_work, written by AFTER DELETE triggers on tasks,
dispatch_contexts, worker_dispatches, worker_terminal_resources,
remote_dispatch_attachments and federated_dispatches. The triggers are
rebuilt on every open after migrate and dropped at migrate start; a
recorded_since coverage marker moves forward whenever any trigger was
missing, so a Run created before recording reads partial, never
complete-and-empty.

reset --tasks keeps its delete list and order; its reply gains optional
removed counts computed after commit (a failed summary only drops the
field). run-show gains an optional constant-size removedWork summary and
--removed/--cursor paging. reset --all clears the record; deleting a Run
deletes its rows.

Removes the snapshot modules, the run_reset_lineage table, the in-transaction
receipt completion and alwaysRequestId from the first version of this PR.

* chore: drop accidental pnpm-lock.yaml churn

* refactor(orchestration): decode removed work entries through unknown before the checked cast

* fix(orchestration): count settled workers with open terminals and name affected Runs

- A worker may still be running when its state was possibly live or its
  Orca-owned terminal was never released; the reset reply, run-show line and
  listing share one derivation from the removed records.
- The reset reply names the affected Runs (optional runIds); the CLI prints
  the exact run-show command for one Run and the list for several.
- The watermark read is no longer swallowed inside the reset transaction.
- run-show's removed-work read is guarded; a failure omits the field.
- Coverage compares created_at strictly after recorded_since.
- --removed lists a federated worker once, shows a user-taken-over terminal
  as such, and says when a cursor is past the end.

* fix(orchestration): count removed workers by held process minus durable exit proof

- A removed worker may still be running when Orca held its process (live
  state, federated dispatch pending or dispatched, or an Orca-owned terminal)
  and no durable exit proof exists (stopped, released, released terminal, or
  a certified termination reason on its Dispatch).
- The exit proof is one DURABLE_EXIT_PROOF constant shared by the fleet
  liveness projection (no behavior change) and the removed-work reader.
- Dispatch records keep terminationReason.
- Coverage is complete when a Run was created at or after recorded_since; a
  moved marker is stamped one second past the newest existing non-legacy Run,
  and the capture installs after the stub home Run backfill.

* docs(orchestration): correct the terminationReason and coverage marker comments

* fix(orchestration): decode removed work entries through a checked schema

The stored record is parsed once with a discriminated zod schema per kind,
so the listed entry keeps its precise type without an assertion. The same
lineage field table drives the capture triggers' key set and integer casts.

* fix(orchestration): exhaustive removed-work listing switch; keep valid ids in id lists

List worker and remote worker entries with explicit cases so the switch is
exhaustive over the entry kinds. An id list drops non-string elements and
keeps the rest, as before the schema decoder.

* test(orchestration): build removed-work CLI fixtures without type assertions

Use a real RuntimeClient with a spied call and decode listing fixtures through
the entry schema, so the tests carry checked types instead of casts.

* fix(orchestration): name only the removed kinds in run-show and word the no-record message truthfully

The run-show line lists the kinds a Run actually lost, terminals included,
and the --removed message for a missing record no longer claims the host
keeps none, since a new host omits it when its read fails.

* fix(wsl-move): classify the removed-work tables with the other orchestration tables

removed_orchestration_work and its coverage marker live in the orchestration
database and go with its Runs, so the WSL mover rewrites them in place like
every other orchestration table.

* fix(orchestration): keep the removed-work hint when a reset removes only terminal records
This commit is contained in:
Brennan Benson
2026-10-26 14:05:33 -07:00
committed by GitHub
parent c2ceb5204f
commit ffc2b3fba1
29 changed files with 1210 additions and 26 deletions
+4
View File
@@ -55,6 +55,10 @@ const COMMAND_SCOPED_FLAG_HELP: Record<string, Record<string, string>> = {
'skills install': {
agent: '--agent <names> Comma-separated install targets; default is detected agents'
},
'orchestration run-show': {
removed: '--removed List the Tasks, Dispatches and workers removed from this Run',
cursor: '--cursor <cursor> Cursor printed by the previous page of removed work'
},
'worktree set': {
unread: '--unread Mark the workspace unread in the sidebar',
read: '--read Mark the workspace read, clearing the unread dot'
@@ -7,6 +7,16 @@ vi.mock('../format', () => ({ printResult: vi.fn() }))
vi.mock('../selectors', () => ({ getTerminalHandle: getTerminalHandleMock }))
import { ORCHESTRATION_HANDLERS } from './orchestration'
import { RuntimeClient } from '../runtime-client'
import {
RemovedWorkEntrySchema,
type RemovedWorkEntry
} from '../../shared/orchestration-removed-work'
import {
formatRemovedWorkListing,
formatResetResult,
formatRunRemovedWorkLine
} from './orchestration/removed-work-format'
import { okFixture, queueFixtures } from '../test-fixtures'
afterEach(() => {
@@ -185,3 +195,144 @@ describe('orchestration reset CLI handler', () => {
expect(callMock).not.toHaveBeenCalled()
})
})
describe('removed work CLI text', () => {
const summary = {
coverage: 'partial' as const,
recordedSince: '2026-10-19T00:00:00Z',
tasks: 2,
dispatches: 1,
workers: 1,
terminals: 1,
activeWorkersAtRemoval: 1,
lastRemovedAt: '2026-10-19T01:00:00Z'
}
const list = 'orca orchestration run-show --id run_1 --removed'
it('adds a run-show line only when the Run has removed work', async () => {
expect(formatRunRemovedWorkLine('run_1', undefined)).toBe('')
expect(
formatRunRemovedWorkLine('run_1', {
...summary,
tasks: 0,
dispatches: 0,
workers: 0,
terminals: 0
})
).toBe('')
expect(formatRunRemovedWorkLine('run_1', summary)).toBe(
`\nremoved work: 2 tasks, 1 dispatch, 1 worker (1 may still be running), 1 terminal; removals before 2026-10-19T00:00:00Z were not recorded. List it: ${list}`
)
const client = new RuntimeClient()
const call = vi.spyOn(client, 'call').mockResolvedValue({
id: 'request',
ok: true,
result: { run: { id: 'run_1' } },
_meta: { runtimeId: 'runtime-1' }
})
await ORCHESTRATION_HANDLERS['orchestration run-show']({
flags: new Map([
['id', 'run_1'],
['cursor', '200']
]),
client,
cwd: '/tmp',
json: true
})
expect(call).toHaveBeenCalledWith('orchestration.runShow', {
id: 'run_1',
removed: true,
cursor: '200'
})
})
it('lists removed entries with a next page, and says when the host keeps no record', () => {
expect(formatRemovedWorkListing('run_1', undefined)).toBe(
'No record of removed work is available from this Orca host.'
)
expect(formatRemovedWorkListing('run_1', { ...summary, entries: [] }, '99')).toBe(
'No more removed work in this Run after cursor 99.'
)
const entries: RemovedWorkEntry[] = [
RemovedWorkEntrySchema.parse({
seq: 1,
kind: 'task',
id: 'task_x',
removedAt: '',
title: 'Build',
parentTaskId: 'task_y',
createdByHandle: 'term_a'
}),
RemovedWorkEntrySchema.parse({
seq: 2,
kind: 'worker',
id: 'ctx_x',
removedAt: '',
terminalHandle: 'term_b',
worktreeId: 'w',
state: 'ready',
activeAtRemoval: true
}),
{ seq: 3, kind: 'dispatch', id: 'ctx_y', removedAt: '', unreadable: true },
RemovedWorkEntrySchema.parse({
seq: 4,
kind: 'terminal',
id: 'wtr_x',
removedAt: '',
terminalHandle: 'term_b',
ownershipState: 'user_owned'
})
]
expect(
formatRemovedWorkListing('run_1', {
...summary,
coverage: 'complete',
entries,
page: { hasMore: true, nextCursor: '3' }
})
).toBe(
[
'Removed work in run_1',
'task task_x "Build", parent task_y, created by term_a',
'worker ctx_x in term_b (worktree w): ready when removed, may still be running',
'dispatch ctx_y: record unreadable',
'terminal wtr_x term_b: taken over by the user',
`More: ${list} --cursor 3`
].join('\n')
)
})
it('summarizes what a task reset removed', () => {
expect(formatResetResult({ reset: 'tasks' })).toBe('Reset: tasks')
expect(
formatResetResult({
reset: 'tasks',
removed: {
runs: 1,
runIds: ['run_1'],
tasks: 3,
dispatches: 4,
workers: 2,
possiblyRunning: 1
}
})
).toBe(
'Reset: tasks. Removed 3 tasks, 4 dispatches and 2 workers from 1 Run. 1 worker may still be running: reset stops tracking workers but does not stop them. List what was removed: orca orchestration run-show --id run_1 --removed'
)
expect(
formatResetResult({
reset: 'tasks',
removed: {
runs: 2,
runIds: ['run_1', 'run_2'],
tasks: 1,
dispatches: 0,
workers: 0,
possiblyRunning: 0
}
})
).toBe(
'Reset: tasks. Removed 1 task, 0 dispatches and 0 workers from 2 Runs. Runs: run_1, run_2. List what was removed in one: orca orchestration run-show --removed --id RUN_ID'
)
})
})
@@ -0,0 +1,111 @@
import type {
RemovedWorkEntry,
ResetRemovedWork,
RunRemovedWork
} from '../../../shared/orchestration-removed-work'
const count = (n: number, noun: string, plural = `${noun}s`): string =>
`${n} ${n === 1 ? noun : plural}`
const listCommand = (runId: string): string => `orca orchestration run-show --id ${runId} --removed`
function partialNote(removed: RunRemovedWork): string {
return removed.coverage === 'partial'
? `; removals before ${removed.recordedSince} were not recorded`
: ''
}
/** The run-show line, or nothing when the Run has no removed work or the host sent no record. */
export function formatRunRemovedWorkLine(runId: string, removed?: RunRemovedWork): string {
if (!removed || removed.tasks + removed.dispatches + removed.workers + removed.terminals === 0) {
return ''
}
const active = removed.activeWorkersAtRemoval
const running = active > 0 ? ` (${active} may still be running)` : ''
// Only the kinds this Run actually lost, so an orphan terminal never reads as "0 tasks".
const kinds = [
removed.tasks > 0 ? count(removed.tasks, 'task') : '',
removed.dispatches > 0 ? count(removed.dispatches, 'dispatch', 'dispatches') : '',
removed.workers > 0 ? `${count(removed.workers, 'worker')}${running}` : '',
removed.terminals > 0 ? count(removed.terminals, 'terminal') : ''
].filter(Boolean)
return `\nremoved work: ${kinds.join(', ')}${partialNote(removed)}. List it: ${listCommand(runId)}`
}
function formatEntry(entry: RemovedWorkEntry): string {
const part = (label: string, value: string | null | undefined): string =>
value ? `${label}${value}` : ''
if ('unreadable' in entry) {
return `${entry.kind} ${entry.id}: record unreadable`
}
switch (entry.kind) {
case 'task':
return `task ${entry.id}${entry.title ? ` "${entry.title}"` : ''}${part(', parent ', entry.parentTaskId)}${part(', created by ', entry.createdByHandle)}`
case 'dispatch': {
const creator = [entry.creatorDispatchId, entry.creatorHandle].filter(Boolean).join(' / ')
return `dispatch ${entry.id}${part(' for ', entry.taskId)}${part(', created by ', creator)}${part(', assigned to ', entry.assigneeHandle)}${part(', status ', entry.status)}`
}
case 'terminal':
return entry.ownershipState === 'user_owned'
? `terminal ${entry.id} ${entry.terminalHandle ?? ''}: taken over by the user`
: `terminal ${entry.id} ${entry.terminalHandle ?? ''} owned by ${entry.ownerDispatchId ?? 'unknown'}: ${entry.releaseState === 'released' ? 'released' : 'not released'}`
case 'federated_worker':
return `federated worker ${entry.id}${part(' on ', entry.environmentName)}${part(' in ', entry.remoteTerminalHandle)}: ${entry.dispatchStatus ?? 'unknown'} when removed${entry.activeAtRemoval ? ', may still be running' : ''}`
case 'worker':
case 'remote_worker': {
const label = entry.kind === 'remote_worker' ? 'remote worker' : 'worker'
return `${label} ${entry.id}${part(' in ', entry.terminalHandle)}${entry.worktreeId ? ` (worktree ${entry.worktreeId})` : ''}: ${entry.state ?? 'unknown'} when removed${entry.activeAtRemoval ? ', may still be running' : ''}`
}
}
}
/** run-show --removed: one line per removed entity, then the next-page hint. */
export function formatRemovedWorkListing(
runId: string,
removed?: RunRemovedWork,
cursor?: string
): string {
if (!removed) {
// Also shown when a host could not read its record; run-show omits the field either way.
return 'No record of removed work is available from this Orca host.'
}
const entries = removed.entries ?? []
if (entries.length === 0) {
return cursor
? `No more removed work in this Run after cursor ${cursor}.`
: `No removed work recorded for this Run${partialNote(removed)}.`
}
const lines = [`Removed work in ${runId}${partialNote(removed)}`, ...entries.map(formatEntry)]
if (removed.page?.nextCursor) {
lines.push(`More: ${listCommand(runId)} --cursor ${removed.page.nextCursor}`)
}
return lines.join('\n')
}
export function formatResetResult(result: { reset: string; removed?: ResetRemovedWork }): string {
const removed = result.removed
if (
!removed ||
(removed.tasks + removed.dispatches + removed.workers === 0 && removed.runs === 0)
) {
return `Reset: ${result.reset}`
}
const running = removed.possiblyRunning
const runningNote =
running > 0
? ` ${count(running, 'worker')} may still be running: reset stops tracking workers but does not stop them.`
: ''
return (
`Reset: ${result.reset}. Removed ${count(removed.tasks, 'task')}, ` +
`${count(removed.dispatches, 'dispatch', 'dispatches')} and ${count(removed.workers, 'worker')} ` +
`from ${count(removed.runs, 'Run')}.${runningNote}${listHint(removed.runIds ?? [])}`
)
}
function listHint(runIds: readonly string[]): string {
if (runIds.length === 1) {
return ` List what was removed: ${listCommand(runIds[0])}`
}
return runIds.length > 1
? ` Runs: ${runIds.join(', ')}. List what was removed in one: orca orchestration run-show --removed --id RUN_ID`
: ''
}
@@ -1,7 +1,9 @@
import type { CommandHandler } from '../../dispatch'
import { printResult } from '../../format'
import { RuntimeClientError } from '../../runtime-client'
import type { ResetRemovedWork } from '../../../shared/orchestration-removed-work'
import { callOrchestrationMutation } from './mutation-request'
import { formatResetResult } from './removed-work-format'
export const ORCHESTRATION_RESET_HANDLER: Record<string, CommandHandler> = {
'orchestration reset': async ({ flags, client, json }) => {
@@ -14,7 +16,7 @@ export const ORCHESTRATION_RESET_HANDLER: Record<string, CommandHandler> = {
'Choose exactly one reset scope: --all, --tasks, or --messages.'
)
}
const result = await callOrchestrationMutation<{ reset: string }>(
const result = await callOrchestrationMutation<{ reset: string; removed?: ResetRemovedWork }>(
client,
flags,
'orchestration.reset',
@@ -24,6 +26,6 @@ export const ORCHESTRATION_RESET_HANDLER: Record<string, CommandHandler> = {
messages: flags.has('messages') ? true : undefined
}
)
printResult(result, json, (value) => `Reset: ${value.reset}`)
printResult(result, json, formatResetResult)
}
}
+16 -7
View File
@@ -6,7 +6,9 @@ import {
getRequiredStringFlag
} from '../../flags'
import { ORCHESTRATION_RUN_PAGE_LIMIT } from '../../../shared/orchestration-run-pagination'
import type { RunRemovedWork } from '../../../shared/orchestration-removed-work'
import { callOrchestrationMutation } from './mutation-request'
import { formatRemovedWorkListing, formatRunRemovedWorkLine } from './removed-work-format'
import { resolveCoordinatorTerminalHandle } from './terminal-identity'
export const ORCHESTRATION_RUN_HANDLERS: Record<string, CommandHandler> = {
@@ -67,6 +69,8 @@ export const ORCHESTRATION_RUN_HANDLERS: Record<string, CommandHandler> = {
},
'orchestration run-show': async ({ flags, client, json }) => {
const cursor = getOptionalStringFlag(flags, 'cursor')
const removed = flags.has('removed') || cursor !== undefined
const result = await client.call<{
run: {
id: string
@@ -75,13 +79,18 @@ export const ORCHESTRATION_RUN_HANDLERS: Record<string, CommandHandler> = {
legacy: number
created_at: string
}
}>('orchestration.runShow', { id: getRequiredStringFlag(flags, 'id') })
printResult(
result,
json,
(r) =>
`${r.run.id}${r.run.legacy ? ' [legacy, inspect only]' : ''} ${r.run.objective}\n` +
`consumer generation ${r.run.consumer_generation}; created ${r.run.created_at}`
removedWork?: RunRemovedWork
}>('orchestration.runShow', {
id: getRequiredStringFlag(flags, 'id'),
...(removed ? { removed: true } : {}),
...(cursor ? { cursor } : {})
})
printResult(result, json, (r) =>
removed
? formatRemovedWorkListing(r.run.id, r.removedWork, cursor)
: `${r.run.id}${r.run.legacy ? ' [legacy, inspect only]' : ''} ${r.run.objective}\n` +
`consumer generation ${r.run.consumer_generation}; created ${r.run.created_at}` +
`${formatRunRemovedWorkLine(r.run.id, r.removedWork)}`
)
}
}
+2 -2
View File
@@ -41,8 +41,8 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
{
path: ['orchestration', 'run-show'],
summary: 'Show one lightweight orchestration Run',
usage: 'orca orchestration run-show --id <run_id> [--json]',
allowedFlags: [...GLOBAL_FLAGS, 'id']
usage: 'orca orchestration run-show --id <run_id> [--removed] [--cursor <cursor>] [--json]',
allowedFlags: [...GLOBAL_FLAGS, 'id', 'removed', 'cursor']
},
{
path: ['orchestration', 'send'],
@@ -4,6 +4,10 @@ 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'
import {
installRemovedWorkCapture,
isRemovedWorkCaptureIntact
} from './reset/removed-orchestration-work-store'
import {
createCoordinatorMailRoutingTrigger,
createRunCoordinatorAddressTriggers,
@@ -31,9 +35,13 @@ class OrchestrationDbCore {
this.db.pragma('synchronous = NORMAL')
this.db.pragma('busy_timeout = 5000')
createTables.call(this as unknown as OrchestrationDb)
// Read before migrate, which drops the capture triggers it may need to rebuild around.
const removedWorkCaptureWasIntact = isRemovedWorkCaptureIntact(this.db)
migrate.call(this as unknown as OrchestrationDb)
createRunCoordinatorAddressTriggers(this.db)
backfillFederatedStubHomeRuns(this.db)
// After every open step that creates Runs, before any step that deletes work.
installRemovedWorkCapture(this.db, removedWorkCaptureWasIntact)
backfillStructuredWorkerOrcaSessionIds(this.db)
reconcileSettledWorkerDispatches(this.db)
createCoordinatorMailRoutingTrigger.call(this as unknown as OrchestrationDb)
@@ -8,6 +8,7 @@ import { OrchestrationDb } from './orchestration-db'
import { createRootDispatch } from './root-dispatch-test-fixture'
import { LEGACY_CONTRACT_VERSION } from './contract-constants'
import { DISPATCH_CONTEXT_COLUMNS } from './row-column-lists'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from './reset/removed-orchestration-work-store'
import { warnStaleDispatches } from '../coordinator-task-dispatch'
const databases: OrchestrationDb[] = []
@@ -58,6 +59,7 @@ function fixture(count = 1, version: 'fresh' | 'v41' = 'fresh') {
})
if (version === 'v41') {
writer.db.exec(`
${DROP_REMOVED_WORK_TRIGGERS_SQL}
DROP INDEX idx_runs_coordinator_orca_session_id;
DROP INDEX idx_dispatch_assignee_orca_session_id;
DROP TRIGGER trg_runs_remember_coordinator_insert;
@@ -1,11 +1,20 @@
import { LEGACY_RUN_ID } from '../contract-constants'
import type { OrchestrationDb } from '../orchestration-db'
import type { ResetRemovedWork } from '../../../../../shared/orchestration-removed-work'
import {
readRemovedWorkWatermark,
summarizeRemovedWorkSince
} from './removed-orchestration-work-read'
export function runResetTransaction(this: OrchestrationDb, statements: string): void {
/** Returns the removed-work watermark, read inside the transaction so the reply counts only this reset. */
export function runResetTransaction(this: OrchestrationDb, statements: string): number {
this.db.exec('BEGIN IMMEDIATE')
try {
// Why unguarded: an error that ends the transaction must roll the whole reset back, not let the deletes autocommit.
const watermark = readRemovedWorkWatermark(this.db)
this.db.exec(statements)
this.db.exec('COMMIT')
return watermark
} catch (error) {
try {
this.db.exec('ROLLBACK')
@@ -41,15 +50,16 @@ export function resetAll(this: OrchestrationDb): void {
DELETE FROM messages;
DELETE FROM run_coordinator_handles;
DELETE FROM runs;
DELETE FROM removed_orchestration_work;
INSERT INTO runs (id, objective, home_database, consumer_generation, legacy)
VALUES ('${LEGACY_RUN_ID}', 'Legacy orchestration state (inspect only)', 'this_database', 0, 1);
`)
this.hasAnyDispatchContextsCache = undefined
}
export function resetTasks(this: OrchestrationDb): void {
export function resetTasks(this: OrchestrationDb): ResetRemovedWork | undefined {
// Messages survive; close threads so orphaned questions cannot masquerade as generic replies.
this.runResetTransaction(`
const watermark = this.runResetTransaction(`
DELETE FROM coordinator_runs;
DELETE FROM decision_gates;
DELETE FROM remote_questions;
@@ -72,6 +82,7 @@ export function resetTasks(this: OrchestrationDb): void {
DELETE FROM tasks;
`)
this.hasAnyDispatchContextsCache = undefined
return summarizeRemovedWorkSince(this.db, watermark)
}
export function resetMessages(this: OrchestrationDb): void {
@@ -0,0 +1,222 @@
import type Database from '../../../../sqlite/sync-database'
import type { SqliteRow } from '../../../../sqlite/sqlite-statement'
import {
REMOVED_WORK_PAGE_SIZE,
RemovedWorkEntrySchema,
type RemovedWorkEntry,
type ResetRemovedWork,
type RunRemovedWork
} from '../../../../../shared/orchestration-removed-work'
import { POTENTIALLY_LIVE_REMOTE_ATTACHMENT_STATES } from '../federation/remote-attachment-liveness'
import { DURABLE_EXIT_PROOF } from '../../../../../shared/orchestration-fleet-worker-projection'
import { OrchestrationError } from '../../orchestration-error'
import { exposeUtcTimestamp } from '../utc-timestamp'
const WORKER_KINDS = "('worker', 'remote_worker', 'federated_worker')"
const sqlList = (values: readonly string[]): string =>
values.map((value) => `'${value}'`).join(', ')
const field = (key: string): string => `json_extract(record, '$.${key}')`
// Every json_extract sits behind json_valid: a bad record must never fail the read.
const whenValid = (predicate: string): string =>
`CASE WHEN json_valid(record) THEN ${predicate} ELSE 0 END`
// Held: Orca was still supervising the process when tracking stopped, not a verdict about now.
const HELD_WORKER_ROW_SQL = whenValid(`(kind IN ('worker', 'remote_worker')
AND ${field('state')} IN (${sqlList(POTENTIALLY_LIVE_REMOTE_ATTACHMENT_STATES)}))
OR (kind = 'federated_worker' AND ${field('dispatchStatus')} IN ('pending', 'dispatched'))`)
// A settled worker keeps its agent open in a terminal Orca still owns.
const HELD_TERMINAL_ROW_SQL = `kind = 'terminal' AND ${whenValid(
`COALESCE(${field('ownershipState')}, '') NOT IN ('user_owned', 'released')`
)}`
// Proven ended: the same durable exit proof the fleet liveness verdict accepts.
const ENDED_WORKER_ROW_SQL = `kind IN ('worker', 'remote_worker') AND ${whenValid(
`${field('state')} = '${DURABLE_EXIT_PROOF.workerState}' OR ${field('stage')} = '${DURABLE_EXIT_PROOF.workerStage}'`
)}`
const RELEASED_TERMINAL_ROW_SQL = `kind = 'terminal' AND ${whenValid(
`${field('releaseState')} = '${DURABLE_EXIT_PROOF.resourceReleaseState}'`
)}`
const ENDED_DISPATCH_ROW_SQL = `kind = 'dispatch' AND ${whenValid(
`${field('terminationReason')} IN (${sqlList(DURABLE_EXIT_PROOF.terminationReasons)})`
)}`
/** Removed rows matching `scopeSql`, their distinct workers, and the workers that may still run. */
function removedWorkScopeSql(scopeSql: string): string {
return `WITH scope AS (
SELECT run_id, kind, entity_id, record FROM removed_orchestration_work WHERE ${scopeSql}
),
workers AS (SELECT DISTINCT entity_id AS id FROM scope WHERE kind IN ${WORKER_KINDS}),
held AS (
SELECT entity_id AS id FROM scope WHERE ${HELD_WORKER_ROW_SQL}
UNION SELECT ${field('ownerDispatchId')} FROM scope WHERE ${HELD_TERMINAL_ROW_SQL}
),
ended AS (
SELECT entity_id AS id FROM scope WHERE ${ENDED_WORKER_ROW_SQL}
UNION SELECT ${field('ownerDispatchId')} FROM scope WHERE ${RELEASED_TERMINAL_ROW_SQL}
UNION SELECT entity_id FROM scope WHERE ${ENDED_DISPATCH_ROW_SQL}
),
active AS (SELECT id FROM held WHERE id IN (SELECT id FROM workers) EXCEPT SELECT id FROM ended)`
}
function readRemovedWorkCounts(db: Database.Database, scopeSql: string, param: string | number) {
const row = db
.prepare(
`${removedWorkScopeSql(scopeSql)}
SELECT json_group_array(DISTINCT run_id) AS run_ids,
COALESCE(SUM(kind = 'task'), 0) AS tasks,
COALESCE(SUM(kind = 'dispatch'), 0) AS dispatches,
COALESCE(SUM(kind = 'terminal'), 0) AS terminals,
(SELECT COUNT(*) FROM workers) AS workers,
(SELECT COUNT(*) FROM active) AS active
FROM scope`
)
.get(param)
const runIds: unknown = JSON.parse(String(row?.run_ids ?? '[]'))
return {
runIds: Array.isArray(runIds) ? runIds.filter((id) => typeof id === 'string') : [],
tasks: Number(row?.tasks ?? 0),
dispatches: Number(row?.dispatches ?? 0),
terminals: Number(row?.terminals ?? 0),
workers: Number(row?.workers ?? 0),
active: Number(row?.active ?? 0)
}
}
const text = (value: unknown): string | null => (typeof value === 'string' ? value : null)
/** Read inside the reset transaction, so the summary covers exactly that reset's removals. */
export function readRemovedWorkWatermark(db: Database.Database): number {
return Number(
db.prepare('SELECT COALESCE(MAX(seq), 0) AS seq FROM removed_orchestration_work').get()?.seq
)
}
/** Never throws: the reset already committed, and its reply only loses this optional field. */
export function summarizeRemovedWorkSince(
db: Database.Database,
watermark: number
): ResetRemovedWork | undefined {
try {
const counts = readRemovedWorkCounts(db, 'seq > ?', watermark)
return {
runs: counts.runIds.length,
runIds: counts.runIds,
tasks: counts.tasks,
dispatches: counts.dispatches,
workers: counts.workers,
possiblyRunning: counts.active
}
} catch (error) {
console.warn('[orchestration] Could not summarize removed work after reset', error)
return undefined
}
}
const isObject = (value: unknown): value is Record<string, unknown> =>
typeof value === 'object' && value !== null && !Array.isArray(value)
function parseRecord(value: unknown): Record<string, unknown> | undefined {
try {
const parsed: unknown = JSON.parse(String(value))
return isObject(parsed) ? parsed : undefined
} catch {
return undefined
}
}
function decodeEntry(row: SqliteRow, activeIds: ReadonlySet<string>): RemovedWorkEntry {
const base = {
seq: Number(row.seq),
id: String(row.entity_id),
removedAt: exposeUtcTimestamp(text(row.removed_at)) ?? ''
}
const kind = String(row.kind)
const record = parseRecord(row.record)
// Non-worker kinds strip activeAtRemoval; an unknown kind or bad record reads as unreadable.
const parsed = RemovedWorkEntrySchema.safeParse({
...record,
...base,
kind,
createdAt: exposeUtcTimestamp(text(record?.createdAt)),
activeAtRemoval: activeIds.has(base.id)
})
return record && parsed.success ? parsed.data : { ...base, kind, unreadable: true }
}
/** Never throws for bookkeeping reasons: on a failed read run-show simply omits the field. */
export function readRunRemovedWork(
db: Database.Database,
runId: string,
options: { removed?: boolean | undefined; cursor?: string | undefined }
): RunRemovedWork | undefined {
if (options.cursor !== undefined && !/^\d+$/.test(options.cursor)) {
throw new OrchestrationError('cursor_invalid', 'The removed work cursor is invalid.')
}
try {
return readRunRemovedWorkPage(db, runId, options)
} catch (error) {
console.warn('[orchestration] Could not read removed work for run-show', error)
return undefined
}
}
function readRunRemovedWorkPage(
db: Database.Database,
runId: string,
options: { removed?: boolean | undefined; cursor?: string | undefined }
): RunRemovedWork | undefined {
// A moved marker sorts after every non-legacy Run that existed then (the empty legacy Run is
// excluded on purpose; see the capture install), so >= is exact.
const coverage = db
.prepare(
`SELECT recorded_since,
(SELECT created_at FROM runs WHERE id = ?) >= recorded_since AS complete,
(SELECT MAX(removed_at) FROM removed_orchestration_work WHERE run_id = ?) AS last_removed_at
FROM removed_orchestration_work_coverage`
)
.get(runId, runId)
if (!coverage) {
return undefined
}
const counts = readRemovedWorkCounts(db, 'run_id = ?', runId)
const summary: RunRemovedWork = {
coverage: coverage.complete === 1 ? 'complete' : 'partial',
recordedSince: exposeUtcTimestamp(text(coverage.recorded_since)) ?? '',
tasks: counts.tasks,
dispatches: counts.dispatches,
workers: counts.workers,
terminals: counts.terminals,
activeWorkersAtRemoval: counts.active,
lastRemovedAt: exposeUtcTimestamp(text(coverage.last_removed_at))
}
if (!options.removed && options.cursor === undefined) {
return summary
}
const activeIds = new Set(
db
.prepare(
`${removedWorkScopeSql('run_id = ?')}
SELECT id FROM active`
)
.all(runId)
.map((row) => String(row.id))
)
// A federated worker also has a local worker row; list it once, as the counts do.
const rows = db
.prepare(
`SELECT seq, kind, entity_id, record, removed_at FROM removed_orchestration_work removed
WHERE run_id = ? AND seq > ?
AND NOT (kind = 'worker' AND EXISTS (
SELECT 1 FROM removed_orchestration_work federated
WHERE federated.entity_id = removed.entity_id AND federated.kind = 'federated_worker'
AND federated.run_id = removed.run_id))
ORDER BY seq LIMIT ?`
)
.all(runId, Number(options.cursor ?? 0), REMOVED_WORK_PAGE_SIZE + 1)
const entries = rows.slice(0, REMOVED_WORK_PAGE_SIZE).map((row) => decodeEntry(row, activeIds))
const hasMore = rows.length > REMOVED_WORK_PAGE_SIZE
return {
...summary,
entries,
page: { hasMore, nextCursor: hasMore ? String(entries.at(-1)?.seq) : null }
}
}
@@ -0,0 +1,222 @@
import type Database from '../../../../sqlite/sync-database'
import {
REMOVED_WORK_INT_FIELD,
REMOVED_WORK_LINEAGE_FIELDS,
type RemovedWorkKind
} from '../../../../../shared/orchestration-removed-work'
/**
* One row per removed Task, Dispatch, worker or terminal, so deleting live rows never erases who
* created what. Rowid key and no constraints: recording a removal must never fail the delete.
* Rows live as long as their Run; reset --all clears the table.
*/
export const REMOVED_WORK_SCHEMA_SQL = `
CREATE TABLE IF NOT EXISTS removed_orchestration_work (
seq INTEGER PRIMARY KEY,
run_id TEXT,
kind TEXT NOT NULL,
entity_id TEXT NOT NULL,
record TEXT NOT NULL,
removed_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_removed_orchestration_work_run
ON removed_orchestration_work(run_id, seq);
CREATE INDEX IF NOT EXISTS idx_removed_orchestration_work_entity
ON removed_orchestration_work(entity_id, kind);
-- Removals before recorded_since were not recorded; a Run created since then has a complete record.
CREATE TABLE IF NOT EXISTS removed_orchestration_work_coverage (
singleton INTEGER PRIMARY KEY CHECK(singleton = 1),
recorded_since TEXT NOT NULL
);
CREATE TRIGGER IF NOT EXISTS trg_runs_forget_removed_work
AFTER DELETE ON runs
BEGIN
DELETE FROM removed_orchestration_work WHERE run_id = OLD.id;
END;
`
/** The Run of a Dispatch id, whichever of its rows a delete in any order has left behind. */
function runOfDispatchSql(dispatchIdSql: string): string {
return `COALESCE(
(SELECT run_id FROM dispatch_contexts WHERE id = ${dispatchIdSql}),
(SELECT NULLIF(home_run_id, '') FROM remote_dispatch_attachments WHERE dispatch_id = ${dispatchIdSql}),
(SELECT run_id FROM removed_orchestration_work
WHERE entity_id = ${dispatchIdSql} AND kind IN ('dispatch', 'remote_worker')
ORDER BY seq DESC LIMIT 1)
)`
}
type CaptureSpec<K extends RemovedWorkKind> = {
table: string
entityIdSql: string
runIdSql: string
fields: Record<keyof (typeof REMOVED_WORK_LINEAGE_FIELDS)[K], string>
}
// Lineage columns only: spec, result, hashes, start options, effects and errors stay out.
const CAPTURE_SPECS: { [K in RemovedWorkKind]: CaptureSpec<K> } = {
task: {
table: 'tasks',
entityIdSql: 'OLD.id',
runIdSql: 'OLD.run_id',
fields: {
parentTaskId: 'OLD.parent_id',
title: 'COALESCE(OLD.task_title, OLD.display_name)',
createdByHandle: 'OLD.created_by_terminal_handle',
createdByPaneKey: 'OLD.created_by_pane_key',
deps: 'OLD.deps',
status: 'OLD.status',
createdAt: 'OLD.created_at'
}
},
dispatch: {
table: 'dispatch_contexts',
entityIdSql: 'OLD.id',
runIdSql: 'OLD.run_id',
fields: {
taskId: 'OLD.task_id',
retryOfDispatchId: 'OLD.retry_of_dispatch_id',
creatorDispatchId: 'OLD.creator_dispatch_id',
creatorHandle: 'OLD.creator_handle',
creatorPaneKey: 'OLD.creator_pane_key',
creatorOrcaSessionId: 'OLD.creator_orca_session_id',
assigneeHandle: 'OLD.assignee_handle',
assigneePaneKey: 'OLD.assignee_pane_key',
assigneeOrcaSessionId: 'OLD.assignee_orca_session_id',
hostScope: 'OLD.host_scope',
depth: 'OLD.depth',
status: 'OLD.status',
terminationReason: 'OLD.termination_reason',
createdAt: 'OLD.created_at'
}
},
worker: {
table: 'worker_dispatches',
entityIdSql: 'OLD.dispatch_id',
runIdSql: runOfDispatchSql('OLD.dispatch_id'),
fields: {
state: 'OLD.state',
stage: 'OLD.stage',
worktreeId: 'OLD.worktree_id',
terminalHandle: 'OLD.agent_terminal_handle',
createdAt: 'OLD.created_at'
}
},
terminal: {
table: 'worker_terminal_resources',
entityIdSql: 'OLD.id',
runIdSql: `COALESCE(${runOfDispatchSql('OLD.owner_dispatch_id')}, ${runOfDispatchSql('OLD.origin_dispatch_id')})`,
fields: {
originDispatchId: 'OLD.origin_dispatch_id',
ownerDispatchId: 'OLD.owner_dispatch_id',
priorOwnerDispatchIds: 'OLD.prior_owner_dispatch_ids',
terminalHandle: 'OLD.terminal_handle',
paneKey: 'OLD.pane_key',
worktreeId: 'OLD.worktree_id',
hostScope: 'OLD.host_scope',
ownershipState: 'OLD.ownership_state',
releaseState: 'OLD.release_state',
createdAt: 'OLD.created_at'
}
},
remote_worker: {
table: 'remote_dispatch_attachments',
entityIdSql: 'OLD.dispatch_id',
runIdSql: "NULLIF(OLD.home_run_id, '')",
fields: {
taskId: 'OLD.task_id',
homePeerFingerprint: 'OLD.home_peer_fingerprint',
state: 'OLD.state',
stage: 'OLD.stage',
worktreeId: 'OLD.worktree_id',
terminalHandle: 'OLD.terminal_handle',
paneKey: 'OLD.pane_key',
depth: 'OLD.depth',
createdAt: 'OLD.created_at'
}
},
federated_worker: {
table: 'federated_dispatches',
entityIdSql: 'OLD.dispatch_id',
runIdSql: runOfDispatchSql('OLD.dispatch_id'),
fields: {
environmentId: 'OLD.environment_id',
environmentName: 'OLD.environment_name',
remoteWorktreeId: 'OLD.remote_worktree_id',
remoteTerminalHandle: 'OLD.remote_terminal_handle',
dispatchStatus: '(SELECT status FROM dispatch_contexts WHERE id = OLD.dispatch_id)',
createdAt: 'OLD.created_at'
}
}
}
const CAPTURE_KINDS: readonly RemovedWorkKind[] = [
'task',
'dispatch',
'worker',
'terminal',
'remote_worker',
'federated_worker'
]
const triggerName = (kind: RemovedWorkKind): string =>
`trg_${CAPTURE_SPECS[kind].table}_record_removed_work`
export const DROP_REMOVED_WORK_TRIGGERS_SQL = CAPTURE_KINDS.map(
(kind) => `DROP TRIGGER IF EXISTS ${triggerName(kind)};`
).join('\n')
// CAST keeps a stray BLOB from throwing in json_object; JSON columns are copied as raw text.
function captureTriggerSql(kind: RemovedWorkKind): string {
const spec = CAPTURE_SPECS[kind]
const lineageFields: Record<string, unknown> = REMOVED_WORK_LINEAGE_FIELDS[kind]
const fields = Object.entries<string>(spec.fields).map(
([key, sql]) =>
`'${key}', CAST(${sql} AS ${lineageFields[key] === REMOVED_WORK_INT_FIELD ? 'INTEGER' : 'TEXT'})`
)
return `CREATE TRIGGER ${triggerName(kind)}
AFTER DELETE ON ${spec.table}
BEGIN
INSERT INTO removed_orchestration_work (run_id, kind, entity_id, record)
VALUES (CAST(${spec.runIdSql} AS TEXT), '${kind}', COALESCE(CAST(${spec.entityIdSql} AS TEXT), ''),
json_object('v', 1, ${fields.join(', ')}));
END;`
}
export function isRemovedWorkCaptureIntact(db: Database.Database): boolean {
const names = CAPTURE_KINDS.map((kind) => `'${triggerName(kind)}'`).join(', ')
const row = db
.prepare(
`SELECT COUNT(*) AS n FROM sqlite_master WHERE type = 'trigger' AND name IN (${names})`
)
.get()
return Number(row?.n) === CAPTURE_KINDS.length
}
// A moved marker sorts after every existing non-legacy Run, so none of them reads complete.
const MOVED_RECORDED_SINCE_SQL = `MAX(datetime('now'), COALESCE(
(SELECT datetime(MAX(created_at), '+1 second') FROM runs WHERE legacy = 0), ''))`
/**
* Recreated on every open, after every open step that creates Runs and before any that deletes work:
* the bodies name columns, so a body stamped once would keep an older build's columns. Coverage moves
* whenever any trigger was missing before this open, so a record never claims removals it did not see.
*/
export function installRemovedWorkCapture(db: Database.Database, captureWasIntact: boolean): void {
db.exec('BEGIN IMMEDIATE')
try {
db.exec(`
${DROP_REMOVED_WORK_TRIGGERS_SQL}
${CAPTURE_KINDS.map(captureTriggerSql).join('\n')}
INSERT OR ${captureWasIntact ? 'IGNORE' : 'REPLACE'} INTO removed_orchestration_work_coverage
(singleton, recorded_since)
VALUES (1, ${captureWasIntact ? "datetime('now')" : MOVED_RECORDED_SINCE_SQL});
`)
db.exec('COMMIT')
} catch (error) {
db.exec('ROLLBACK')
throw error
}
}
@@ -112,6 +112,8 @@ CREATE TABLE IF NOT EXISTS remote_questions (
CREATE INDEX IF NOT EXISTS idx_remote_questions_dispatch_status
ON remote_questions(dispatch_id, status);
-- Deletes from tasks, dispatch_contexts and worker tables are recorded in removed_orchestration_work
-- (removed-orchestration-work-store.ts).
CREATE TABLE IF NOT EXISTS tasks (
id TEXT PRIMARY KEY,
run_id TEXT NOT NULL DEFAULT '${LEGACY_RUN_ID}',
@@ -137,6 +139,7 @@ CREATE TABLE IF NOT EXISTS tasks (
CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_id);
-- Deletes are recorded in removed_orchestration_work (removed-orchestration-work-store.ts).
CREATE TABLE IF NOT EXISTS dispatch_contexts (
id TEXT PRIMARY KEY,
run_id TEXT NOT NULL DEFAULT '${LEGACY_RUN_ID}',
@@ -2,9 +2,10 @@ import type { OrchestrationDb } from '../orchestration-db'
import { createCoreTablesSql } from './create-core-tables-sql'
import { createGraphTablesSql } from './create-graph-tables-sql'
import { DERIVED_DELIVERY_SCHEMA_SQL } from './migrate-v41'
import { REMOVED_WORK_SCHEMA_SQL } from '../reset/removed-orchestration-work-store'
export function createTables(this: OrchestrationDb): void {
this.db.exec(`${createCoreTablesSql()}\n${createGraphTablesSql()}`)
this.db.exec(`${createCoreTablesSql()}\n${createGraphTablesSql()}\n${REMOVED_WORK_SCHEMA_SQL}`)
this.createMailboxDeliveryIndexesIfPossible()
this.db.exec(DERIVED_DELIVERY_SCHEMA_SQL)
}
@@ -6,6 +6,7 @@ import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../../shared/protocol-v
import { OrchestrationDb } from '../orchestration-db'
import { SCHEMA_VERSION, federatedStubHomeRunId } from '../contract-constants'
import { migrateV40 } from './migrate-v40'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from '../reset/removed-orchestration-work-store'
import { importFederatedControlMessage } from '../../federation-control-message'
describe('federated home Run migration', () => {
@@ -43,6 +44,7 @@ describe('federated home Run migration', () => {
}
it('backfills a pre-upgrade attachment with a stub home Run that keeps its mailbox', () => {
db.db.exec(DROP_REMOVED_WORK_TRIGGERS_SQL)
db.db.exec('ALTER TABLE remote_dispatch_attachments DROP COLUMN home_run_id')
db.db.exec(`INSERT INTO remote_dispatch_attachments
(dispatch_id, task_id, home_peer_fingerprint, runtime_epoch)
@@ -14,6 +14,7 @@ import { migrateV40 } from './migrate-v40'
import { DERIVED_DELIVERY_SCHEMA_SQL, migrateV41 } from './migrate-v41'
import { migrateV42 } from './migrate-v42'
import { migrateV43 } from './migrate-v43'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from '../reset/removed-orchestration-work-store'
// CREATE TABLE IF NOT EXISTS won't alter existing tables; migrate atomically before bumping user_version.
export function migrate(this: OrchestrationDb): void {
@@ -28,6 +29,8 @@ export function migrate(this: OrchestrationDb): void {
this.db.exec(
'DROP TRIGGER IF EXISTS trg_deliveries_one_outstanding; DROP VIEW IF EXISTS outstanding_deliveries;'
)
// Why: their bodies name columns a step may drop or rebuild; the open after migrate recreates them.
this.db.exec(DROP_REMOVED_WORK_TRIGGERS_SQL)
applySchemaMigrationsV2ToV12.call(this, current)
applySchemaMigrationsV13ToV30.call(this, current)
migrateMailboxPointerEnterV33.call(this, current)
@@ -6,6 +6,7 @@ import Database from '../../sqlite/sync-database'
import { OrchestrationDb } from './db'
import { SCHEMA_VERSION } from './db/contract-constants'
import { createRootDispatch } from './db/root-dispatch-test-fixture'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from './db/reset/removed-orchestration-work-store'
import { resolveOrchestrationMigrationStartVersion } from './orchestration-schema-version-skew'
const CREATOR_COLUMNS = ['creator_handle', 'creator_pane_key'] as const
@@ -39,6 +40,7 @@ describe('OrchestrationDb v36 to v37 migration', () => {
seed.close()
const raw = new Database(dbPath)
raw.exec(DROP_REMOVED_WORK_TRIGGERS_SQL)
// The current mail-routing trigger reads creator_handle; the next open recreates it.
raw.exec('DROP TRIGGER trg_messages_route_coordinator_mail')
for (const column of CREATOR_COLUMNS) {
@@ -5,6 +5,7 @@ import { afterEach, describe, expect, it } from 'vitest'
import Database from '../../sqlite/sync-database'
import { dropDerivedDeliverySchema } from './db/schema/derived-delivery-test-fixture'
import { OrchestrationDb } from './db'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from './db/reset/removed-orchestration-work-store'
import { resolveOrchestrationMigrationStartVersion } from './orchestration-schema-version-skew'
import { SCHEMA_VERSION } from './db/contract-constants'
@@ -30,6 +31,7 @@ describe('nested worker depth migration (v30)', () => {
const oldDb = new Database(dbPath)
dropDerivedDeliverySchema(oldDb)
oldDb.exec(DROP_REMOVED_WORK_TRIGGERS_SQL)
oldDb.exec('ALTER TABLE dispatch_contexts DROP COLUMN depth')
oldDb.exec('ALTER TABLE remote_dispatch_attachments DROP COLUMN depth')
oldDb.exec('ALTER TABLE remote_dispatch_attachments DROP COLUMN home_run_id')
@@ -6,6 +6,7 @@ import Database from '../../sqlite/sync-database'
import { OrchestrationDb } from './db'
import { SCHEMA_VERSION } from './db/contract-constants'
import { createRootDispatch } from './db/root-dispatch-test-fixture'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from './db/reset/removed-orchestration-work-store'
const MUTATION_RECEIPT_MAX_ROWS = 10_000
@@ -243,6 +244,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => {
const oldDb = new Database(dbPath)
oldDb.exec(`
${DROP_REMOVED_WORK_TRIGGERS_SQL}
DROP INDEX IF EXISTS idx_dispatch_active_assignee_handle;
DROP INDEX IF EXISTS idx_dispatch_assignee_pane_leaf;
ALTER TABLE tasks DROP COLUMN created_by_pane_key;
@@ -13,6 +13,7 @@ import { SCHEMA_VERSION } from './db/contract-constants'
import { formatOrcaSessionAddress } from '../../../shared/orca-session-address'
import { testOrcaSessionId } from '../../../shared/orca-session-address-test-fixture'
import { RUN_PANE_KEY_MATCH_SUFFIX_SQL } from './db/pane-key-match'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from './db/reset/removed-orchestration-work-store'
import {
currentRunCoordinatorOrcaSessionId,
currentRunCoordinatorOrcaSessionIdSql
@@ -133,6 +134,7 @@ function seedStructuredAndPtyRows(db: OrchestrationDb): SeededRows {
function stripOrcaSessionSchema(path: string, version: number): void {
const raw = new Database(path)
raw.exec(`
${DROP_REMOVED_WORK_TRIGGERS_SQL}
DROP INDEX idx_runs_coordinator_orca_session_id;
DROP INDEX idx_dispatch_assignee_orca_session_id;
DROP TRIGGER trg_runs_remember_coordinator_insert;
@@ -1,5 +1,10 @@
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import { LEGACY_RUN_ID, OrchestrationDb } from './db'
import { readRunRemovedWork } from './db/reset/removed-orchestration-work-read'
import { settleWorkerForCompletedDispatch } from './db/worker-dispatch/worker-dispatch-settlement'
describe('OrchestrationDb reset scopes', () => {
let db: OrchestrationDb | undefined
@@ -63,6 +68,52 @@ describe('OrchestrationDb reset scopes', () => {
return { run, task, started, message, localQuestion }
}
// A child Task and worker created by a coordinator, plus a remote worker whose terminal outlives it in reset order.
function addChildWork(state: ReturnType<typeof createState>) {
const child = db!.createTask({
runId: state.run.id,
parentId: state.task.id,
spec: 'child work',
createdByTerminalHandle: 'term_worker'
})
const childDispatch = db!.createStartingWorkerDispatch({
taskId: child.id,
creator: { kind: 'terminal', handle: 'term_coord' },
maxDepth: Number.MAX_SAFE_INTEGER,
startOptions: { worktree: 'current' }
})
const remote = db!.createRemoteDispatchAttachment({
dispatchId: 'ctx_remote_reset_fixture',
runId: state.run.id,
taskId: child.id,
homePeerFingerprint: 'peer_remote',
protocolVersion: 1,
runtimeEpoch: 'remote_epoch',
mutationReceipt: {
callerFingerprint: 'peer_remote',
requestId: 'remote_request',
method: 'orchestration.federationAttachStart',
payloadHash: 'remote_hash'
}
})
const terminal = (dispatchId: string, handle: string) =>
db!.createWorkerTerminalResourceStatement({
dispatchId,
worktreeId: null,
terminalHandle: handle,
paneKey: null,
processIncarnation: `${handle}_process`,
ownership: 'owned'
})
terminal(childDispatch.dispatch.id, 'term_child')
return {
child,
childDispatch,
remote,
remoteTerminal: terminal(remote.dispatch_id, 'term_remote')
}
}
it('resetAll clears Runs, worker/federation state, and messages', () => {
const state = createState()
@@ -75,6 +126,7 @@ describe('OrchestrationDb reset scopes', () => {
const sqlite = (db as unknown as { db: { prepare: (sql: string) => { all: () => unknown[] } } })
.db
expect(sqlite.prepare('SELECT * FROM run_coordinator_handles').all()).toEqual([])
expect(sqlite.prepare('SELECT * FROM removed_orchestration_work').all()).toEqual([])
// The ledger survives so a lost reset response cannot replay as a new mutation.
expect(db!.getMutationReceipt('caller_1', 'request_1')).toBeDefined()
expect(db!.getInbox()).toEqual([])
@@ -119,6 +171,198 @@ describe('OrchestrationDb reset scopes', () => {
).toBeUndefined()
})
it('resetTasks records who created each removed Task, Dispatch and worker', () => {
const state = createState()
const { child, childDispatch, remote, remoteTerminal } = addChildWork(state)
// A finished worker is settled without closing its agent: its owned terminal is never released.
const childId = childDispatch.dispatch.id
db!.db.exec(`UPDATE worker_dispatches SET state = 'ready' WHERE dispatch_id = '${childId}'`)
db!.db.exec(`UPDATE dispatch_contexts SET status = 'completed' WHERE id = '${childId}'`)
settleWorkerForCompletedDispatch(db!.db, childId)
expect(db!.resetTasks()).toEqual({
runs: 1,
runIds: [state.run.id],
tasks: 2,
dispatches: 2,
workers: 3,
possiblyRunning: 3
})
const shown = readRunRemovedWork(db!.db, state.run.id, { removed: true })
expect(shown).toMatchObject({
coverage: 'complete',
tasks: 2,
dispatches: 2,
workers: 3,
terminals: 2,
activeWorkersAtRemoval: 3,
page: { hasMore: false, nextCursor: null }
})
expect(shown?.entries).toEqual(
expect.arrayContaining([
expect.objectContaining({
kind: 'task',
id: child.id,
parentTaskId: state.task.id,
createdByHandle: 'term_worker'
}),
expect.objectContaining({
kind: 'dispatch',
id: childDispatch.dispatch.id,
taskId: child.id,
creatorHandle: 'term_coord'
}),
expect.objectContaining({
kind: 'worker',
id: childId,
state: 'abandoned',
activeAtRemoval: true
}),
expect.objectContaining({ kind: 'remote_worker', id: remote.dispatch_id }),
// Its live Dispatch row was already gone; the Run comes from the recorded remote worker.
expect.objectContaining({
kind: 'terminal',
id: remoteTerminal.id,
ownerDispatchId: remote.dispatch_id
}),
expect.objectContaining({
kind: 'federated_worker',
id: state.started.dispatch.id,
environmentName: 'Windows'
})
])
)
// The federated worker's local worker row is folded into its one line, matching the count.
expect(
shown?.entries?.filter(
(entry) => entry.id === state.started.dispatch.id && entry.kind !== 'dispatch'
)
).toHaveLength(1)
})
it('does not count a worker with durable exit proof as possibly running', () => {
db = new OrchestrationDb(':memory:')
const run = db.createRun({
objective: 'Exit proof',
coordinatorHandle: 'term_coord',
coordinatorPaneKey: 'tab_coord:11111111-1111-4111-8111-111111111111'
})
// Each worker still owns an unreleased terminal; only a certified exit clears it.
const ids = ['stopped', 'operator_close', 'unknown'].map((label) => {
const task = db!.createTask({ runId: run.id, spec: label })
const { dispatch } = db!.createStartingWorkerDispatch({
taskId: task.id,
creator: { kind: 'system' },
maxDepth: Number.MAX_SAFE_INTEGER,
startOptions: { worktree: 'current' }
})
db!.createWorkerTerminalResourceStatement({
dispatchId: dispatch.id,
worktreeId: null,
terminalHandle: `term_${label}`,
paneKey: null,
processIncarnation: `${label}_process`,
ownership: 'owned'
})
const state = label === 'stopped' ? 'stopped' : 'failed'
db!.db.exec(
`UPDATE worker_dispatches SET state = '${state}' WHERE dispatch_id = '${dispatch.id}'`
)
if (label !== 'stopped') {
db!.db.exec(
`UPDATE dispatch_contexts SET status = 'failed', termination_reason = '${label}' WHERE id = '${dispatch.id}'`
)
}
return dispatch.id
})
expect(db.resetTasks()).toMatchObject({ workers: 3, possiblyRunning: 1 })
const workers = readRunRemovedWork(db.db, run.id, { removed: true })?.entries?.filter(
(entry) => entry.kind === 'worker'
)
expect(workers?.map((entry) => 'activeAtRemoval' in entry && entry.activeAtRemoval)).toEqual(
ids.map((id) => id === ids[2])
)
})
it('records raw deletes from every lineage table, as an older binary would run them', () => {
const state = createState()
addChildWork(state)
// Neither a BLOB in a text column nor removing the same id twice may fail the delete.
db!.db.exec(`UPDATE tasks SET task_title = x'00ff' WHERE id = '${state.task.id}'`)
for (const table of [
'tasks',
'dispatch_contexts',
'worker_dispatches',
'worker_terminal_resources',
'remote_dispatch_attachments',
'federated_dispatches'
]) {
db!.db.exec(`DELETE FROM ${table}`)
}
db!.db.exec(
`INSERT INTO tasks (id, run_id, spec) VALUES ('${state.task.id}', '${state.run.id}', 'again')`
)
db!.db.exec('DELETE FROM tasks')
const rows = db!.db
.prepare(
'SELECT kind, run_id, COUNT(*) AS n FROM removed_orchestration_work GROUP BY kind, run_id'
)
.all()
expect(rows).toEqual(
expect.arrayContaining(
[
['task', 3],
['dispatch', 2],
['worker', 2],
['terminal', 2],
['remote_worker', 1],
['federated_worker', 1]
].map(([kind, n]) => ({ kind, run_id: state.run.id, n }))
)
)
expect(rows).toHaveLength(6)
})
it('never reads a Run older than the record as complete, and moves coverage when capture was missing', () => {
const tempDir = mkdtempSync(join(tmpdir(), 'orca-removed-work-coverage-'))
const path = join(tempDir, 'orchestration.db')
try {
db = new OrchestrationDb(path)
const run = db.createRun({
objective: 'Older Run',
coordinatorHandle: 'term_coord',
coordinatorPaneKey: 'tab_coord:11111111-1111-4111-8111-111111111111'
})
db.db.exec(`UPDATE runs SET created_at = '2000-01-02 00:00:00' WHERE id = '${run.id}'`)
expect(readRunRemovedWork(db.db, run.id, {})).toMatchObject({ coverage: 'partial', tasks: 0 })
db.db.exec(
`UPDATE removed_orchestration_work_coverage SET recorded_since = '2000-01-01 00:00:00'`
)
db.close()
db = new OrchestrationDb(path)
expect(readRunRemovedWork(db.db, run.id, {})?.coverage).toBe('complete')
// An older build that dropped one capture trigger leaves a gap, so coverage restarts now.
db.db.exec('DROP TRIGGER trg_tasks_record_removed_work')
db.close()
db = new OrchestrationDb(path)
expect(readRunRemovedWork(db.db, run.id, {})?.coverage).toBe('partial')
expect(
db.db
.prepare("SELECT name FROM sqlite_master WHERE name = 'trg_tasks_record_removed_work'")
.get()
).toBeDefined()
} finally {
db?.close()
db = undefined
rmSync(tempDir, { recursive: true, force: true })
}
})
it('resetMessages preserves active relay cursors while clearing the Run inbox', () => {
const state = createState()
@@ -4,6 +4,7 @@ import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import Database from '../../sqlite/sync-database'
import { OrchestrationDb } from './db'
import { DROP_REMOVED_WORK_TRIGGERS_SQL } from './db/reset/removed-orchestration-work-store'
import { SCHEMA_VERSION } from './db/contract-constants'
const DISPATCH_IDENTITY_COLUMNS = [
@@ -51,6 +52,7 @@ describe('R1 identity migration', () => {
db = undefined
const v30 = new Database(dbPath)
v30.exec(DROP_REMOVED_WORK_TRIGGERS_SQL)
v30.exec(
'DROP INDEX IF EXISTS idx_dispatch_retry_of; DROP INDEX IF EXISTS idx_dispatch_resource;'
)
@@ -15,8 +15,8 @@ export const ORCHESTRATION_RESET_METHODS = [
}
if (params.tasks) {
runtime.stopOrchestrationFederationRelay()
db.resetTasks()
return { reset: 'tasks' }
const removed = db.resetTasks()
return removed ? { reset: 'tasks', removed } : { reset: 'tasks' }
}
if (params.messages) {
db.resetMessages()
@@ -24,6 +24,27 @@ describe('orchestration RPC methods', () => {
}
describe('lightweight Runs', () => {
it('shows another caller the work a reset removed, listed only on request', async () => {
setup(false)
const run = db.createRun({
objective: 'Removed work read contract',
coordinatorHandle: 'term_coord',
coordinatorPaneKey
})
const task = db.createTask({ runId: run.id, spec: 'worker task' })
db.resetTasks()
const other = { ...ctx, authenticatedCallerFingerprint: 'different_workspace_caller' }
const summary = await h.call('orchestration.runShow', { id: run.id }, other)
expect(summary).toMatchObject({ removedWork: { coverage: 'complete', tasks: 1 } })
expect(summary).not.toHaveProperty('removedWork.entries')
await expect(
h.call('orchestration.runShow', { id: run.id, removed: true }, other)
).resolves.toMatchObject({
removedWork: { entries: [{ kind: 'task', id: task.id }], page: { hasMore: false } }
})
})
it('creates and binds a Run to the runtime-resolved caller pane', async () => {
setup(false)
vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue(
@@ -1,5 +1,6 @@
import { defineMethod } from '../../../core'
import { OrchestrationError } from '../../../../orchestration/orchestration-error'
import { readRunRemovedWork } from '../../../../orchestration/db/reset/removed-orchestration-work-read'
import { assertCallerHandleMatchesEvidence, resolveOrchestrationCaller } from './run-scope'
import { exposeRun } from './run-receipt'
import type { OrcaRuntimeService } from '../../../../orca-runtime'
@@ -141,11 +142,13 @@ export const ORCHESTRATION_RUN_METHODS = [
permission: 'workspace',
params: RunShowParams,
handler: (params, { runtime }) => {
const run = runtime.getOrchestrationDb().getRun(params.id)
const db = runtime.getOrchestrationDb()
const run = db.getRun(params.id)
if (!run) {
throw new OrchestrationError('run_not_found', `Run ${params.id} was not found.`)
}
return { run: exposeRun(run) }
const removedWork = readRunRemovedWork(db.db, run.id, params)
return removedWork ? { run: exposeRun(run), removedWork } : { run: exposeRun(run) }
}
})
]
+1
View File
@@ -43,6 +43,7 @@ export const CLI_BOOLEAN_FLAGS = new Set([
'reclaim-stale-lock',
'references',
'relations',
'removed',
'reinstall',
'restore-window',
'return-preamble',
@@ -25,6 +25,19 @@ type FleetLivenessSubject = {
resource: { releaseState?: string | null; hostScope: string | null } | null
}
/** Durable facts that prove a supervised process ended; the removed-work record reads the same proof. */
export const DURABLE_EXIT_PROOF = {
workerState: 'stopped',
workerStage: 'released',
resourceReleaseState: 'released',
terminationReasons: ['operator_close', 'signaled', 'exited']
} as const satisfies {
workerState: string
workerStage: string
resourceReleaseState: string
terminationReasons: readonly NonNullable<FleetDurableWorker['terminationReason']>[]
}
/** Worker states that carry an outcome; anything else is still supposed to be running. */
const SETTLED_WORKER_STATES = new Set(['succeeded', 'failed', 'stopped', 'abandoned'])
@@ -33,9 +46,7 @@ function hasCertifiedExit(worker: FleetLivenessSubject): boolean {
return (
// unknown means stop without observed exit; null is pre-v29, when the stage alone recorded exit.
(worker.workerStage === 'process_exited' && worker.terminationReason !== 'unknown') ||
worker.terminationReason === 'operator_close' ||
worker.terminationReason === 'signaled' ||
worker.terminationReason === 'exited'
DURABLE_EXIT_PROOF.terminationReasons.some((reason) => reason === worker.terminationReason)
)
}
@@ -45,13 +56,13 @@ export function projectLiveness(
now: number
): FleetLiveness {
// Federated release proves terminal exit, independently of the worker outcome.
if (worker.workerStage === 'released') {
if (worker.workerStage === DURABLE_EXIT_PROOF.workerStage) {
return { verdict: 'exited', source: 'execution_host' }
}
if (worker.resource?.releaseState === 'released') {
if (worker.resource?.releaseState === DURABLE_EXIT_PROOF.resourceReleaseState) {
return { verdict: 'exited', source: 'resource_release' }
}
if (worker.workerState === 'stopped') {
if (worker.workerState === DURABLE_EXIT_PROOF.workerState) {
return { verdict: 'exited', source: 'worker_stop' }
}
// Operator-close settlement also proves exit; it must not fall through to missing_status.
+134
View File
@@ -0,0 +1,134 @@
import { z } from 'zod'
// Tolerant fields: a missing or mistyped value reads as null instead of failing the entry.
const text = z.string().nullable().catch(null)
/** Stored as an integer; every other record field is stored as text. */
export const REMOVED_WORK_INT_FIELD = z.number().nullable().catch(null)
const int = REMOVED_WORK_INT_FIELD
// Drops non-string elements and keeps the rest; text that isn't JSON reads as null.
const idList = z
.preprocess(parseJsonText, z.array(z.unknown()).nullable())
.transform((items) => items?.filter((item): item is string => typeof item === 'string') ?? null)
.catch(null)
function parseJsonText(value: unknown): unknown {
if (typeof value !== 'string') {
return null
}
try {
return JSON.parse(value)
} catch {
return null
}
}
/** Lineage fields kept for each kind of removed orchestration work. */
export const REMOVED_WORK_LINEAGE_FIELDS = {
task: {
parentTaskId: text,
title: text,
createdByHandle: text,
createdByPaneKey: text,
deps: idList,
status: text,
createdAt: text
},
dispatch: {
taskId: text,
retryOfDispatchId: text,
creatorDispatchId: text,
creatorHandle: text,
creatorPaneKey: text,
creatorOrcaSessionId: text,
assigneeHandle: text,
assigneePaneKey: text,
assigneeOrcaSessionId: text,
hostScope: text,
depth: int,
status: text,
// Null when no exit was ever recorded; an exit Orca could not certify is saved as 'unknown'.
terminationReason: text,
createdAt: text
},
worker: { state: text, stage: text, worktreeId: text, terminalHandle: text, createdAt: text },
terminal: {
originDispatchId: text,
ownerDispatchId: text,
priorOwnerDispatchIds: idList,
terminalHandle: text,
paneKey: text,
worktreeId: text,
hostScope: text,
ownershipState: text,
releaseState: text,
createdAt: text
},
remote_worker: {
taskId: text,
homePeerFingerprint: text,
state: text,
stage: text,
worktreeId: text,
terminalHandle: text,
paneKey: text,
depth: int,
createdAt: text
},
federated_worker: {
environmentId: text,
environmentName: text,
remoteWorktreeId: text,
remoteTerminalHandle: text,
dispatchStatus: text,
createdAt: text
}
}
export type RemovedWorkKind = keyof typeof REMOVED_WORK_LINEAGE_FIELDS
const entryBase = { seq: z.number(), id: z.string(), removedAt: z.string() }
const workerBase = { ...entryBase, activeAtRemoval: z.boolean() }
const lineage = REMOVED_WORK_LINEAGE_FIELDS
/** One listed entry: the stored record checked against its kind's fields. */
export const RemovedWorkEntrySchema = z.discriminatedUnion('kind', [
z.object({ ...entryBase, kind: z.literal('task'), ...lineage.task }),
z.object({ ...entryBase, kind: z.literal('dispatch'), ...lineage.dispatch }),
z.object({ ...workerBase, kind: z.literal('worker'), ...lineage.worker }),
z.object({ ...entryBase, kind: z.literal('terminal'), ...lineage.terminal }),
z.object({ ...workerBase, kind: z.literal('remote_worker'), ...lineage.remote_worker }),
z.object({ ...workerBase, kind: z.literal('federated_worker'), ...lineage.federated_worker })
])
type RemovedWorkEntryBase = { seq: number; id: string; removedAt: string }
export type RemovedWorkEntry =
| z.infer<typeof RemovedWorkEntrySchema>
| (RemovedWorkEntryBase & { kind: string; unreadable: true })
/** `orchestration.runShow` result `removedWork`: constant size unless entries were requested. */
export type RunRemovedWork = {
// 'partial': the Run predates recordedSince, so earlier removals were not recorded.
coverage: 'complete' | 'partial'
recordedSince: string
tasks: number
dispatches: number
workers: number
terminals: number
activeWorkersAtRemoval: number
lastRemovedAt: string | null
entries?: RemovedWorkEntry[]
page?: { hasMore: boolean; nextCursor: string | null }
}
/** `orchestration.reset --tasks` result `removed`: what this reset removed, across every Run. */
export type ResetRemovedWork = {
runs: number
// Absent from hosts that predate it.
runIds?: string[]
tasks: number
dispatches: number
workers: number
possiblyRunning: number
}
export const REMOVED_WORK_PAGE_SIZE = 200
@@ -20,4 +20,10 @@ export const RunListParams = z.object({
cursor: z.string().min(1).optional()
})
export const RunShowParams = z.object({ id: requiredString('Missing --id'), from: OptionalString })
export const RunShowParams = z.object({
id: requiredString('Missing --id'),
from: OptionalString,
// Lists removed work 200 entries per page; older hosts strip both and return no removedWork.
removed: OptionalBoolean,
cursor: z.string().min(1).optional()
})
@@ -120,6 +120,7 @@ export const SQLITE_SCHEMA_FILE_DATABASES: Record<string, string> = {
'src/main/runtime/orchestration/db/schema/migrate-v35.ts': 'orchestration',
'src/main/runtime/orchestration/db/schema/migrate-v39.ts': 'orchestration',
'src/main/runtime/orchestration/db/schema/migrate.ts': 'orchestration',
'src/main/runtime/orchestration/db/reset/removed-orchestration-work-store.ts': 'orchestration',
'src/main/runtime/orchestration/mutation-receipt-capacity.ts': 'orchestration',
// Test fixtures that mirror third-party agent and browser databases.
'src/main/ai-vault/session-scanner-opencode-sqlite-fixture.ts': 'test-fixture',
@@ -193,6 +194,8 @@ export const SQLITE_TABLE_DISPOSITIONS: Record<string, WslMoveFieldDisposition>
'legacy_operation_receipts',
'attempt_observation_facts',
'structured_pointer_operations',
'removed_orchestration_work',
'removed_orchestration_work_coverage',
'won'
],
{