diff --git a/src/cli/command-scoped-flag-help.ts b/src/cli/command-scoped-flag-help.ts index faa8999a188..c6e243c9395 100644 --- a/src/cli/command-scoped-flag-help.ts +++ b/src/cli/command-scoped-flag-help.ts @@ -55,6 +55,10 @@ const COMMAND_SCOPED_FLAG_HELP: Record> = { 'skills install': { agent: '--agent 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 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' diff --git a/src/cli/handlers/orchestration-run-cli.test.ts b/src/cli/handlers/orchestration-run-cli.test.ts index 610a7a8566a..6a87b49b26f 100644 --- a/src/cli/handlers/orchestration-run-cli.test.ts +++ b/src/cli/handlers/orchestration-run-cli.test.ts @@ -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' + ) + }) +}) diff --git a/src/cli/handlers/orchestration/removed-work-format.ts b/src/cli/handlers/orchestration/removed-work-format.ts new file mode 100644 index 00000000000..d73ba169bf9 --- /dev/null +++ b/src/cli/handlers/orchestration/removed-work-format.ts @@ -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` + : '' +} diff --git a/src/cli/handlers/orchestration/reset-handler.ts b/src/cli/handlers/orchestration/reset-handler.ts index 536cc9e8933..399c01d74a2 100644 --- a/src/cli/handlers/orchestration/reset-handler.ts +++ b/src/cli/handlers/orchestration/reset-handler.ts @@ -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 = { 'orchestration reset': async ({ flags, client, json }) => { @@ -14,7 +16,7 @@ export const ORCHESTRATION_RESET_HANDLER: Record = { '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 = { messages: flags.has('messages') ? true : undefined } ) - printResult(result, json, (value) => `Reset: ${value.reset}`) + printResult(result, json, formatResetResult) } } diff --git a/src/cli/handlers/orchestration/run-handlers.ts b/src/cli/handlers/orchestration/run-handlers.ts index d8abe3533c2..f98e26e5fb0 100644 --- a/src/cli/handlers/orchestration/run-handlers.ts +++ b/src/cli/handlers/orchestration/run-handlers.ts @@ -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 = { @@ -67,6 +69,8 @@ export const ORCHESTRATION_RUN_HANDLERS: Record = { }, '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 = { 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)}` ) } } diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts index c4b9b515066..20008e45749 100644 --- a/src/cli/specs/orchestration.ts +++ b/src/cli/specs/orchestration.ts @@ -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 [--json]', - allowedFlags: [...GLOBAL_FLAGS, 'id'] + usage: 'orca orchestration run-show --id [--removed] [--cursor ] [--json]', + allowedFlags: [...GLOBAL_FLAGS, 'id', 'removed', 'cursor'] }, { path: ['orchestration', 'send'], diff --git a/src/main/runtime/orchestration/db/orchestration-db.ts b/src/main/runtime/orchestration/db/orchestration-db.ts index 4a3c9e5adbe..7d4a247c704 100644 --- a/src/main/runtime/orchestration/db/orchestration-db.ts +++ b/src/main/runtime/orchestration/db/orchestration-db.ts @@ -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) diff --git a/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts b/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts index 3873f628500..354465a5cf5 100644 --- a/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts +++ b/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts @@ -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; diff --git a/src/main/runtime/orchestration/db/reset/orchestration-reset.ts b/src/main/runtime/orchestration/db/reset/orchestration-reset.ts index 742df8cb68d..9761105970a 100644 --- a/src/main/runtime/orchestration/db/reset/orchestration-reset.ts +++ b/src/main/runtime/orchestration/db/reset/orchestration-reset.ts @@ -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 { diff --git a/src/main/runtime/orchestration/db/reset/removed-orchestration-work-read.ts b/src/main/runtime/orchestration/db/reset/removed-orchestration-work-read.ts new file mode 100644 index 00000000000..d3c4b0f65ff --- /dev/null +++ b/src/main/runtime/orchestration/db/reset/removed-orchestration-work-read.ts @@ -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 => + typeof value === 'object' && value !== null && !Array.isArray(value) + +function parseRecord(value: unknown): Record | undefined { + try { + const parsed: unknown = JSON.parse(String(value)) + return isObject(parsed) ? parsed : undefined + } catch { + return undefined + } +} + +function decodeEntry(row: SqliteRow, activeIds: ReadonlySet): 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 } + } +} diff --git a/src/main/runtime/orchestration/db/reset/removed-orchestration-work-store.ts b/src/main/runtime/orchestration/db/reset/removed-orchestration-work-store.ts new file mode 100644 index 00000000000..9f0a2e530fe --- /dev/null +++ b/src/main/runtime/orchestration/db/reset/removed-orchestration-work-store.ts @@ -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 = { + table: string + entityIdSql: string + runIdSql: string + fields: Record +} + +// Lineage columns only: spec, result, hashes, start options, effects and errors stay out. +const CAPTURE_SPECS: { [K in RemovedWorkKind]: CaptureSpec } = { + 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 = REMOVED_WORK_LINEAGE_FIELDS[kind] + const fields = Object.entries(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 + } +} diff --git a/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts b/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts index d77ab67f64c..5c5fb5fa5ad 100644 --- a/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts +++ b/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts @@ -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}', diff --git a/src/main/runtime/orchestration/db/schema/create-tables.ts b/src/main/runtime/orchestration/db/schema/create-tables.ts index c0fa229eb3c..fbb9298b25e 100644 --- a/src/main/runtime/orchestration/db/schema/create-tables.ts +++ b/src/main/runtime/orchestration/db/schema/create-tables.ts @@ -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) } diff --git a/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts b/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts index 4fcd78634f3..fe30f9c4b61 100644 --- a/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts +++ b/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts @@ -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) diff --git a/src/main/runtime/orchestration/db/schema/migrate.ts b/src/main/runtime/orchestration/db/schema/migrate.ts index 5b57bb5d9d0..9bf4adeb0ea 100644 --- a/src/main/runtime/orchestration/db/schema/migrate.ts +++ b/src/main/runtime/orchestration/db/schema/migrate.ts @@ -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) diff --git a/src/main/runtime/orchestration/dispatch-creator-identity-migration.test.ts b/src/main/runtime/orchestration/dispatch-creator-identity-migration.test.ts index bfe47373457..006148f8b67 100644 --- a/src/main/runtime/orchestration/dispatch-creator-identity-migration.test.ts +++ b/src/main/runtime/orchestration/dispatch-creator-identity-migration.test.ts @@ -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) { diff --git a/src/main/runtime/orchestration/nested-worker-depth-migration.test.ts b/src/main/runtime/orchestration/nested-worker-depth-migration.test.ts index 8aeaac82d91..e4494416431 100644 --- a/src/main/runtime/orchestration/nested-worker-depth-migration.test.ts +++ b/src/main/runtime/orchestration/nested-worker-depth-migration.test.ts @@ -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') diff --git a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts index f9a985512d2..80cc4695819 100644 --- a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts +++ b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts @@ -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; diff --git a/src/main/runtime/orchestration/orchestration-orca-session-column-migration.test.ts b/src/main/runtime/orchestration/orchestration-orca-session-column-migration.test.ts index 22a513a979f..e312bd3e2eb 100644 --- a/src/main/runtime/orchestration/orchestration-orca-session-column-migration.test.ts +++ b/src/main/runtime/orchestration/orchestration-orca-session-column-migration.test.ts @@ -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; diff --git a/src/main/runtime/orchestration/orchestration-reset-db.test.ts b/src/main/runtime/orchestration/orchestration-reset-db.test.ts index 6a662d20f3d..f8c2c23d519 100644 --- a/src/main/runtime/orchestration/orchestration-reset-db.test.ts +++ b/src/main/runtime/orchestration/orchestration-reset-db.test.ts @@ -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) { + 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() diff --git a/src/main/runtime/orchestration/r1-identity-migration.test.ts b/src/main/runtime/orchestration/r1-identity-migration.test.ts index 673a72fd844..49b13c38020 100644 --- a/src/main/runtime/orchestration/r1-identity-migration.test.ts +++ b/src/main/runtime/orchestration/r1-identity-migration.test.ts @@ -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;' ) diff --git a/src/main/runtime/rpc/methods/orchestration/runs/reset-methods.ts b/src/main/runtime/rpc/methods/orchestration/runs/reset-methods.ts index 36141403707..2563c6b4855 100644 --- a/src/main/runtime/rpc/methods/orchestration/runs/reset-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/runs/reset-methods.ts @@ -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() diff --git a/src/main/runtime/rpc/methods/orchestration/runs/runs.test.ts b/src/main/runtime/rpc/methods/orchestration/runs/runs.test.ts index dc0331dae97..9c2b0bdde4f 100644 --- a/src/main/runtime/rpc/methods/orchestration/runs/runs.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/runs/runs.test.ts @@ -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( diff --git a/src/main/runtime/rpc/methods/orchestration/runs/runs.ts b/src/main/runtime/rpc/methods/orchestration/runs/runs.ts index 6af5ac6a2ed..1add74b2e44 100644 --- a/src/main/runtime/rpc/methods/orchestration/runs/runs.ts +++ b/src/main/runtime/rpc/methods/orchestration/runs/runs.ts @@ -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) } } }) ] diff --git a/src/shared/cli-argument-boundary.ts b/src/shared/cli-argument-boundary.ts index 871a42d178e..83a41d41d7c 100644 --- a/src/shared/cli-argument-boundary.ts +++ b/src/shared/cli-argument-boundary.ts @@ -43,6 +43,7 @@ export const CLI_BOOLEAN_FLAGS = new Set([ 'reclaim-stale-lock', 'references', 'relations', + 'removed', 'reinstall', 'restore-window', 'return-preamble', diff --git a/src/shared/orchestration-fleet-worker-projection.ts b/src/shared/orchestration-fleet-worker-projection.ts index 78a3c903911..79806e6bea5 100644 --- a/src/shared/orchestration-fleet-worker-projection.ts +++ b/src/shared/orchestration-fleet-worker-projection.ts @@ -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[] +} + /** 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. diff --git a/src/shared/orchestration-removed-work.ts b/src/shared/orchestration-removed-work.ts new file mode 100644 index 00000000000..cddd1f1f0a7 --- /dev/null +++ b/src/shared/orchestration-removed-work.ts @@ -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 + | (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 diff --git a/src/shared/rpc-contract/orchestration-runs-params.ts b/src/shared/rpc-contract/orchestration-runs-params.ts index 30aad1e9eae..66ff03879a8 100644 --- a/src/shared/rpc-contract/orchestration-runs-params.ts +++ b/src/shared/rpc-contract/orchestration-runs-params.ts @@ -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() +}) diff --git a/src/shared/wsl-move/wsl-move-storage-dispositions.ts b/src/shared/wsl-move/wsl-move-storage-dispositions.ts index 32c8bf90aea..ba843d55fd5 100644 --- a/src/shared/wsl-move/wsl-move-storage-dispositions.ts +++ b/src/shared/wsl-move/wsl-move-storage-dispositions.ts @@ -120,6 +120,7 @@ export const SQLITE_SCHEMA_FILE_DATABASES: Record = { '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 'legacy_operation_receipts', 'attempt_observation_facts', 'structured_pointer_operations', + 'removed_orchestration_work', + 'removed_orchestration_work_coverage', 'won' ], {