From 05e35530cac9e6185002f527dc1b28649bbb02e5 Mon Sep 17 00:00:00 2001 From: Neil Date: Tue, 20 Oct 2026 14:34:27 -0700 Subject: [PATCH] fix(wsl-mover): hold the orchestration store while a worker in scope is live; skip federated rows (D1b-7e review) Co-Authored-By: Claude --- .../automation-wsl-move-refusal.test.ts | 17 ++++- ...move-orchestration-history-rewrite.test.ts | 74 +++++++++++++++++-- .../wsl-move-orchestration-history-rewrite.ts | 65 +++++++++++----- .../wsl-move/wsl-move-storage-dispositions.ts | 2 +- 4 files changed, 130 insertions(+), 28 deletions(-) diff --git a/src/main/automations/automation-wsl-move-refusal.test.ts b/src/main/automations/automation-wsl-move-refusal.test.ts index a191a4e7e43..b2f7973542b 100644 --- a/src/main/automations/automation-wsl-move-refusal.test.ts +++ b/src/main/automations/automation-wsl-move-refusal.test.ts @@ -23,6 +23,8 @@ import { registerWslMoveMainRewriteStores } from '../persistence/moving-wsl-proj import { registerWslMoveOrchestrationHistoryStore } from '../persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite' import { orchestrationDbFile } from '../runtime/orchestration/db/orchestration-db-file' import { WSL_HOSTS_INTERNAL_GATE_ENV } from '../../shared/wsl-host-internal-gate' +import { OrchestrationDb } from '../runtime/orchestration/db' +import Database from '../sqlite/sync-database' import { AutomationService } from './service' import { WSL_MOVE_IN_PROGRESS, WSL_MOVE_OWNED_BY_HOST } from './dispatch-refusal' import type * as PrecheckRunner from './precheck-runner' @@ -166,10 +168,23 @@ describe('scheduled automations under a WSL move', () => { fenceDir, cutover('move-1', 'destination-committed', {}, { repoIds: ['r1'] }) ) + const databasePath = orchestrationDbFile(fenceDir) + // A dispatch in the moved repo, so the orchestration store has real work to do. + new OrchestrationDb(databasePath).close() + const seed = new Database(databasePath) + seed + .prepare(`INSERT INTO worker_dispatches (dispatch_id, state, worktree_id) VALUES (?, ?, ?)`) + .run('d-1', 'succeeded', 'r1::\\\\wsl.localhost\\Ubuntu\\home\\dev\\r1') + seed.close() registerWslMoveMainRewriteStores(store) - registerWslMoveOrchestrationHistoryStore(orchestrationDbFile(fenceDir)) + registerWslMoveOrchestrationHistoryStore(databasePath) try { await runWslMoveDesktopRewritePass({ userDataPath: fenceDir }) + const read = new Database(databasePath) + expect(read.prepare(`SELECT worktree_id FROM worker_dispatches`).get()).toEqual({ + worktree_id: 'r1::/home/dev/r1' + }) + read.close() expect(store.listAutomations()).toEqual([automation]) const service = attachedService(store) diff --git a/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.test.ts b/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.test.ts index a211ef7ccb7..f906573bff3 100644 --- a/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.test.ts +++ b/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.test.ts @@ -28,6 +28,7 @@ const MOVED_ON_HOST = 'repo-a::/home/dev/repo-a' const STAYING = 'repo-b::\\\\wsl.localhost\\Ubuntu\\home\\dev\\repo-b' const OTHER_DISTRO = 'repo-a::\\\\wsl.localhost\\Debian\\home\\dev\\repo-a' const SSH_PATH = 'repo-a::/srv/repo-a' +const MOVED_WSL_DOLLAR = 'repo-a::\\\\wsl$\\Ubuntu\\home\\dev\\repo-a' const CHAT_HANDLE = `${STRUCTURED_WORKER_HANDLE_PREFIX}chat-1` let userDataPath: string @@ -61,8 +62,9 @@ function seedOrchestrationHistory(): void { ) dispatch.run('d-moved', 'succeeded', MOVED, 'term_1') dispatch.run('d-failed', 'failed', MOVED, 'term_2') - dispatch.run('d-live', 'ready', MOVED, 'term_3') - dispatch.run('d-chat', 'succeeded', MOVED, CHAT_HANDLE) + dispatch.run('d-wsl-dollar', 'stopped', MOVED_WSL_DOLLAR, 'term_3') + dispatch.run('d-chat', 'ready', MOVED, CHAT_HANDLE) + dispatch.run('d-federated', 'succeeded', MOVED, 'term_8') dispatch.run('d-staying', 'succeeded', STAYING, 'term_4') dispatch.run('d-other-distro', 'succeeded', OTHER_DISTRO, 'term_5') dispatch.run('d-ssh', 'succeeded', SSH_PATH, 'term_6') @@ -73,7 +75,6 @@ function seedOrchestrationHistory(): void { VALUES (?, ?, ?, ?, ?)` ) resource.run('r-moved', 'd-moved', 'd-moved', MOVED, 'term_1') - resource.run('r-live', 'd-live', 'd-live', MOVED, 'term_3') resource.run('r-chat', 'd-chat', 'd-chat', MOVED, CHAT_HANDLE) resource.run('r-remote', 'd-remote', 'd-remote', MOVED, 'term_7') db.prepare( @@ -81,6 +82,11 @@ function seedOrchestrationHistory(): void { (dispatch_id, task_id, home_peer_fingerprint, runtime_epoch, state, worktree_id) VALUES (?, ?, ?, ?, ?, ?)` ).run('d-remote', 'task-1', 'peer', 'epoch', 'succeeded', MOVED) + db.prepare( + `INSERT INTO federated_dispatches + (dispatch_id, environment_id, environment_name, peer_fingerprint, remote_worktree_id) + VALUES (?, ?, ?, ?, ?)` + ).run('d-federated', 'env-1', 'Remote', 'peer', MOVED) }) } @@ -133,16 +139,16 @@ describe('WSL move orchestration history rewrite', () => { expect(dispatchIds()).toEqual({ 'd-chat': MOVED, 'd-failed': MOVED_ON_HOST, - 'd-live': MOVED, + 'd-federated': MOVED, 'd-moved': MOVED_ON_HOST, 'd-no-worktree': null, 'd-other-distro': OTHER_DISTRO, 'd-ssh': SSH_PATH, - 'd-staying': STAYING + 'd-staying': STAYING, + 'd-wsl-dollar': MOVED_ON_HOST }) expect(resourceIds()).toEqual({ 'r-chat': MOVED, - 'r-live': MOVED, 'r-moved': MOVED_ON_HOST, 'r-remote': MOVED }) @@ -222,7 +228,61 @@ describe('WSL move orchestration history rewrite', () => { ]) }) - it('reads an older schema as it stands', async () => { + it('waits, unrecorded, while a worker in scope is still live, then rewrites it once settled', async () => { + seedOrchestrationHistory() + withDb((db) => { + db.prepare( + `INSERT INTO worker_dispatches (dispatch_id, state, worktree_id, agent_terminal_handle) + VALUES ('d-live', 'start_unknown', ?, 'term_9')` + ).run(MOVED) + db.prepare( + `INSERT INTO worker_terminal_resources + (id, origin_dispatch_id, owner_dispatch_id, worktree_id, terminal_handle) + VALUES ('r-live', 'd-live', 'd-live', ?, 'term_9')` + ).run(MOVED) + }) + const dispatches = dispatchIds() + seedWslMoveCutover(userDataPath, committedMove()) + + await runWslMoveDesktopRewritePass({ userDataPath, warn: () => {} }) + + expect(listWslMoveCutovers(userDataPath)[0]?.rewrittenStores).toEqual([]) + expect(dispatchIds()).toEqual(dispatches) + withDb((db) => + db.prepare(`UPDATE worker_dispatches SET state = 'failed' WHERE dispatch_id = 'd-live'`).run() + ) + + await runWslMoveDesktopRewritePass({ userDataPath }) + + expect(listWslMoveCutovers(userDataPath)[0]?.rewrittenStores).toEqual([ + WSL_MOVE_ORCHESTRATION_HISTORY_STORE_ID + ]) + expect(dispatchIds()['d-live']).toBe(MOVED_ON_HOST) + expect(resourceIds()['r-live']).toBe(MOVED_ON_HOST) + }) + + it('rewrites a schema older than worker-terminal resources and remote attachments', async () => { + const db = new Database(databasePath) + db.exec( + `CREATE TABLE worker_dispatches ( + dispatch_id TEXT PRIMARY KEY, state TEXT NOT NULL, + worktree_id TEXT, agent_terminal_handle TEXT)` + ) + db.prepare(`INSERT INTO worker_dispatches VALUES ('d-old', 'succeeded', ?, 'term_1')`).run( + MOVED + ) + db.close() + seedWslMoveCutover(userDataPath, committedMove()) + + await runWslMoveDesktopRewritePass({ userDataPath }) + + expect(listWslMoveCutovers(userDataPath)[0]?.rewrittenStores).toEqual([ + WSL_MOVE_ORCHESTRATION_HISTORY_STORE_ID + ]) + expect(dispatchIds()).toEqual({ 'd-old': MOVED_ON_HOST }) + }) + + it('records a schema with no worktree column, changing nothing', async () => { const db = new Database(databasePath) db.exec(`CREATE TABLE worker_dispatches (dispatch_id TEXT PRIMARY KEY, state TEXT NOT NULL)`) db.close() diff --git a/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.ts b/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.ts index 6a04cff7d85..73f17d9b288 100644 --- a/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.ts +++ b/src/main/persistence/moving-wsl-projects/wsl-move-orchestration-history-rewrite.ts @@ -14,8 +14,19 @@ import { wslMoveRewriteScope, type WslMoveRewriteScope } from './wsl-move-rewrit export const WSL_MOVE_ORCHESTRATION_HISTORY_STORE_ID = 'orchestration-history' -// The recoverable worker states; the move's gate refuses them, so a row still in one is left alone. -const LIVE_WORKER_STATES_SQL = `('starting', 'ready', 'start_unknown', 'stopping', 'stop_unknown')` +/** The recoverable worker states. The move's dispatch gate must refuse this same set. */ +export const WSL_MOVE_LIVE_WORKER_STATES = [ + 'starting', + 'ready', + 'start_unknown', + 'stopping', + 'stop_unknown' +] as const + +const LIVE_WORKER_STATES_SQL = `(${WSL_MOVE_LIVE_WORKER_STATES.map((state) => `'${state}'`).join(', ')})` + +// The lock wait blocks the main thread (synchronous SQLite), so keep it short; a busy pass retries. +const BUSY_TIMEOUT_MS = 2_000 type Rewrite = { key: string; from: string; to: string } @@ -34,11 +45,17 @@ function hasColumns(db: Database, table: string, columns: readonly string[]): bo return columns.every((column) => names.has(column)) } -function hostWorktreeId(scope: WslMoveRewriteScope, worktreeId: string): string { - return scope.inScope(worktreeId) ? scope.worktreeId(worktreeId) : worktreeId +/** Excludes rows whose dispatch id appears in `table`, when that table exists. */ +function unlessListedIn(db: Database, table: string, dispatchIdColumn: string): string { + return hasColumns(db, table, ['dispatch_id']) + ? `AND NOT EXISTS (SELECT 1 FROM ${table} x WHERE x.dispatch_id = ${dispatchIdColumn})` + : '' } -/** Rows to rewrite; a structured (chat) worker stays local with its chat (owner decision C). */ +/** + * Rows to rewrite; a structured (chat) worker stays local with its chat (owner decision C). A live + * row in scope throws, so the store gets no ledger entry and a later pass finds it settled. + */ function plan( rows: SqliteRow[], scope: WslMoveRewriteScope, @@ -51,8 +68,14 @@ function plan( if (!key || !from || isStructuredWorkerHandle(text(row, handleColumn))) { return [] } - const to = hostWorktreeId(scope, from) - return to === from ? [] : [{ key, from, to }] + const to = scope.inScope(from) ? scope.worktreeId(from) : from + if (to === from) { + return [] + } + if (row.live === 1) { + throw new Error('wsl_move_orchestration_worker_live') + } + return [{ key, from, to }] }) } @@ -67,10 +90,14 @@ function dispatchRewrites(db: Database, scope: WslMoveRewriteScope): Rewrite[] { ) { return [] } + // A federated dispatch's row names the remote host's worktree. const rows = db .prepare( - `SELECT dispatch_id, worktree_id, agent_terminal_handle FROM worker_dispatches - WHERE worktree_id IS NOT NULL AND state NOT IN ${LIVE_WORKER_STATES_SQL}` + `SELECT w.dispatch_id, w.worktree_id, w.agent_terminal_handle, + w.state IN ${LIVE_WORKER_STATES_SQL} AS live + FROM worker_dispatches w + WHERE w.worktree_id IS NOT NULL + ${unlessListedIn(db, 'federated_dispatches', 'w.dispatch_id')}` ) .all() return plan(rows, scope, 'dispatch_id', 'agent_terminal_handle') @@ -89,16 +116,14 @@ function resourceRewrites(db: Database, scope: WslMoveRewriteScope): Rewrite[] { return [] } // A federated worker's row belongs with its attachment, which a remote home reads verbatim. - const remote = hasColumns(db, 'remote_dispatch_attachments', ['dispatch_id']) - ? `AND r.owner_dispatch_id NOT IN (SELECT dispatch_id FROM remote_dispatch_attachments)` - : '' const rows = db .prepare( - `SELECT r.id, r.worktree_id, r.terminal_handle FROM worker_terminal_resources r + `SELECT r.id, r.worktree_id, r.terminal_handle, + COALESCE(w.state IN ${LIVE_WORKER_STATES_SQL}, 0) AS live + FROM worker_terminal_resources r LEFT JOIN worker_dispatches w ON w.dispatch_id = r.owner_dispatch_id WHERE r.worktree_id IS NOT NULL - AND (w.state IS NULL OR w.state NOT IN ${LIVE_WORKER_STATES_SQL}) - ${remote}` + ${unlessListedIn(db, 'remote_dispatch_attachments', 'r.owner_dispatch_id')}` ) .all() return plan(rows, scope, 'id', 'terminal_handle') @@ -118,7 +143,7 @@ function applyRewrites(db: Database, sql: string, rewrites: readonly Rewrite[]): /** * Idempotent: a rewritten id is no longer a UNC path of the distro, so a second run finds nothing. - * One write transaction for both tables. Returns how many rows changed. + * One write transaction for both tables, planned before any write. Returns how many rows changed. */ export function rewriteWslMoveOrchestrationHistory( db: Database, @@ -126,16 +151,18 @@ export function rewriteWslMoveOrchestrationHistory( ): number { db.exec('BEGIN IMMEDIATE') try { + const dispatches = dispatchRewrites(db, scope) + const resources = resourceRewrites(db, scope) const changed = applyRewrites( db, `UPDATE worker_dispatches SET worktree_id = ? WHERE dispatch_id = ? AND worktree_id = ?`, - dispatchRewrites(db, scope) + dispatches ) + applyRewrites( db, `UPDATE worker_terminal_resources SET worktree_id = ? WHERE id = ? AND worktree_id = ?`, - resourceRewrites(db, scope) + resources ) db.exec('COMMIT') return changed @@ -158,7 +185,7 @@ export function registerWslMoveOrchestrationHistoryStore(databasePath: string): if (!existsSync(databasePath)) { return } - const db = new Database(databasePath, { fileMustExist: true, timeout: 5_000 }) + const db = new Database(databasePath, { fileMustExist: true, timeout: BUSY_TIMEOUT_MS }) try { // A FULL commit is on disk when it returns, which is what the ledger entry needs. db.pragma('synchronous = FULL') diff --git a/src/shared/wsl-move/wsl-move-storage-dispositions.ts b/src/shared/wsl-move/wsl-move-storage-dispositions.ts index 430dc1d72a7..91abf735fd7 100644 --- a/src/shared/wsl-move/wsl-move-storage-dispositions.ts +++ b/src/shared/wsl-move/wsl-move-storage-dispositions.ts @@ -165,7 +165,7 @@ export const SQLITE_TABLE_DISPOSITIONS: Record ...tables('orchestration', ['worker_dispatches', 'worker_terminal_resources'], { disposition: 'rewrite-desktop', item: 53, - note: "worktree_id of a settled local dispatch becomes the host's id (D1b-7e); chat workers, live rows and JSON receipts stay as written." + note: "worktree_id of a settled local dispatch becomes the host's id (D1b-7e); a live one holds the store until it settles. Chat and federated workers and JSON receipts stay as written." }), ...tables( 'orchestration',