fix(wsl-mover): hold the orchestration store while a worker in scope is live; skip federated rows (D1b-7e review)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-11-04 02:57:36 +00:00 · 2026-10-20 22:10:55 -07:00
parent fab9f339d3
commit 05e35530ca
4 changed files with 130 additions and 28 deletions
@@ -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)

@@ -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()
@@ -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')
@@ -165,7 +165,7 @@ export const SQLITE_TABLE_DISPOSITIONS: Record<string, WslMoveFieldDisposition>
    ...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',