diff --git a/daemon/ops/runbook.md b/daemon/ops/runbook.md index 043823b..8dcb79b 100644 --- a/daemon/ops/runbook.md +++ b/daemon/ops/runbook.md @@ -334,21 +334,26 @@ reacquires on expiry or an API 401. Rotate a client secret in Linear, update the host value, restart, and verify an ack; rotation invalidates that app's existing tokens. Revoke the app installation in Linear to cut off access immediately. -Deploy-gate confirmations before relying on startup reconciliation in production: - -```bash -# Confirm bulk agentSessions is scoped to the calling app actor and pages as expected. -# If it is not, keep PLANNER_APP_ACTOR_ID / IMPLEMENTER_APP_ACTOR_ID populated so the -# delegate issue -> issue.agentSessions fallback can resolve real session IDs. -# If either APP_ACTOR_ID is missing, startup reconciliation still re-enables webhooks but -# skips session-discovery synthesis for that app to avoid importing foreign sessions. - -# Confirm the added admin scope authorizes webhooks/updateWebhook under client_credentials -# and does not force token-invalidating re-consent beyond the planned app reinstall. - -# Capture one prompted webhook and the same activity through GraphQL; verify -# webhook agentActivity.id == GraphQL AgentActivity.id. -``` +### Post-provision checklist + +- [ ] Confirm `PLANNER_APP_ACTOR_ID` and `IMPLEMENTER_APP_ACTOR_ID` are both + populated in `/etc/linear-agent-daemon/env`. Confirm bulk `agentSessions` is scoped to + the calling app actor and pages as expected; otherwise retain both IDs so the delegated + issue → `issue.agentSessions` fallback can resolve real session IDs. +- [ ] Attempt the `admin`-scope grant for both apps and run the + `webhooks/updateWebhook` client-credentials confirmation. Linear currently rejects this + scope (observed live 2026-07). If it still rejects the scope, stop this gate and file + **`webhook-reconcile-fallback` — "Decide and implement the webhook re-enable path when + Linear rejects the `admin` scope on client_credentials tokens"**; record AC6 as blocked, + not failed. +- [ ] After restart, inspect one full reconcile interval and confirm zero + `reconcile_sessions_skipped_missing_app_actor_id` events: + `journalctl -u linear-agent-daemon --since -2min | grep -c reconcile_sessions_skipped_missing_app_actor_id` + must print `0`. +- [ ] Only if the admin-scope gate succeeded, inspect one full reconcile interval and + confirm `reconcile_webhook` for both apps and zero `reconcile_webhook_failed` events. +- [ ] Capture one prompted webhook and the same activity through GraphQL; verify + `webhook agentActivity.id == GraphQL AgentActivity.id`. ## Planner credentials and repository diff --git a/daemon/src/config.ts b/daemon/src/config.ts index 1aa8e5e..1cf4482 100644 --- a/daemon/src/config.ts +++ b/daemon/src/config.ts @@ -34,6 +34,8 @@ export interface Config { webhookBaseUrl: string; artifactToken?: string; artifactsDir: string; + dispatchQuarantineDir: string; + dispatchQuarantineAgeMs: number; browserEnabled: boolean; playwrightMcpBin: string; playwrightChromeBin: string; @@ -169,6 +171,14 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { webhookBaseUrl: webhookBaseUrl.replace(/\/+$/, ""), ...(artifactToken ? { artifactToken } : {}), artifactsDir: env.ARTIFACTS_DIR?.trim() || `${dirname(dbPath)}/artifacts`, + dispatchQuarantineDir: + env.DISPATCH_QUARANTINE_DIR?.trim() || + `${dirname(dbPath)}/dispatch-quarantine`, + dispatchQuarantineAgeMs: positiveInteger( + env, + "DISPATCH_QUARANTINE_AGE_MS", + 24 * 60 * 60 * 1000, + ), browserEnabled: enabled(env, "BROWSER_ENABLED", false), playwrightMcpBin: env.PLAYWRIGHT_MCP_BIN?.trim() || "/usr/local/bin/playwright-mcp", playwrightChromeBin: env.PLAYWRIGHT_CHROME_BIN?.trim() || "/usr/bin/google-chrome", diff --git a/daemon/src/eventlog.ts b/daemon/src/eventlog.ts index d789e88..134dab4 100644 --- a/daemon/src/eventlog.ts +++ b/daemon/src/eventlog.ts @@ -2108,6 +2108,9 @@ export class EventLog { ) .all(linearSessionId) as AgentInvocationRow[]; } + hasCodexInvocation(sourceKey: string): boolean { + return this.invocationBySourceKey(sourceKey) !== undefined; + } private invocationBySourceKey( sourceKey: string, ): AgentInvocationRow | undefined { diff --git a/daemon/src/sessions.ts b/daemon/src/sessions.ts index 631f426..21eb4c0 100644 --- a/daemon/src/sessions.ts +++ b/daemon/src/sessions.ts @@ -1,10 +1,15 @@ import { randomUUID } from "node:crypto"; +import { constants } from "node:fs"; import { + copyFile, + lstat, mkdir, open, readdir, readFile, realpath, + rename, + stat, unlink, } from "node:fs/promises"; import { resolve, sep } from "node:path"; @@ -331,6 +336,8 @@ export class SessionWorker { private readonly logger: Logger; private readonly worktrees: WorktreeManager; private readonly legacyProfileLogged = new Set(); + // A marker occurrence may produce at most one degraded log line per daemon process. + private readonly degradedLogged = new Set(); constructor( private readonly log: EventLog, @@ -1201,6 +1208,72 @@ export class SessionWorker { ingestDispatches(): Promise { return this.triggerDispatchScan(); } + private async availableQuarantinePath( + directory: string, + name: string, + reserved: ReadonlySet, + ): Promise { + for (let suffix = 0; ; suffix++) { + const candidate = resolve( + directory, + suffix === 0 ? name : `${name}.${suffix}`, + ); + if (reserved.has(candidate)) continue; + try { + await lstat(candidate); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return candidate; + throw error; + } + } + } + private async moveDispatchFile( + source: string, + destination: string, + ): Promise { + try { + await rename(source, destination); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EXDEV") throw error; + await copyFile(source, destination, constants.COPYFILE_EXCL); + await unlink(source); + } + } + private async quarantineDispatchBundle( + directory: string, + linearSessionId: string, + base: string, + doneFile: string, + ): Promise { + const destinationDirectory = resolve( + this.config.dispatchQuarantineDir, + linearSessionId, + ); + await mkdir(destinationDirectory, { recursive: true }); + const names = (await readdir(directory)) + .filter((name) => name.startsWith(`${base}.`)) + .sort((a, b) => { + if (a === doneFile) return 1; + if (b === doneFile) return -1; + return a.localeCompare(b); + }); + const reserved = new Set(); + const moves: Array<{ source: string; destination: string }> = []; + for (const name of names) { + const destination = await this.availableQuarantinePath( + destinationDirectory, + name, + reserved, + ); + reserved.add(destination); + moves.push({ source: resolve(directory, name), destination }); + } + for (const move of moves) { + if (this.log.hasOpenTurn(linearSessionId)) return undefined; + await this.moveDispatchFile(move.source, move.destination); + } + return destinationDirectory; + } private async scanDispatchMarkers(): Promise { if (this.stopped) return; try { @@ -1230,6 +1303,14 @@ export class SessionWorker { const files = (await readdir(directory)) .filter((file) => file.endsWith(".done")) .sort(); + const degradedPrefix = `${session.linearSessionId}:`; + const currentMarkers = new Set(files); + for (const key of this.degradedLogged) + if ( + key.startsWith(degradedPrefix) && + !currentMarkers.has(key.slice(degradedPrefix.length)) + ) + this.degradedLogged.delete(key); const markers: Array<{ file: string; base: string; @@ -1320,16 +1401,57 @@ export class SessionWorker { const valid = sidecarShapeValid && sidecarTurnId !== undefined; const turnId = sidecarTurnId ?? this.log.latestTurnId(session.linearSessionId); - if (!turnId) { + const degradedKey = `${session.linearSessionId}:${file}`; + const sourceKey = `dispatch:${session.linearSessionId}:${file}`; + try { + const age = + this.now() - (await stat(resolve(directory, file))).mtimeMs; + if ( + age > this.config.dispatchQuarantineAgeMs && + (this.log.hasCodexInvocation(sourceKey) || !turnId) + ) { + const destination = await this.quarantineDispatchBundle( + directory, + session.linearSessionId, + base, + file, + ); + if (destination === undefined) continue; + this.degradedLogged.delete(degradedKey); + this.logger.log( + jsonLog({ + event: "dispatch_marker_quarantined", + linearSessionId: session.linearSessionId, + marker, + destination, + }), + ); + continue; + } + } catch (error) { this.logger.error( jsonLog({ - event: "dispatch_marker_ingest_degraded", + event: "dispatch_marker_quarantine_failed", linearSessionId: session.linearSessionId, - reason: "no_parent_turn", + marker, + error: String(error), }), ); continue; } + if (!turnId) { + if (!this.degradedLogged.has(degradedKey)) { + this.degradedLogged.add(degradedKey); + this.logger.error( + jsonLog({ + event: "dispatch_marker_ingest_degraded", + linearSessionId: session.linearSessionId, + reason: "no_parent_turn", + }), + ); + } + continue; + } const exitCode = typeof sidecar?.exit_code === "number" && Number.isFinite(sidecar.exit_code) @@ -1343,7 +1465,7 @@ export class SessionWorker { const invocation: CodexInvocationInput = { linearSessionId: session.linearSessionId, turnId, - sourceKey: `dispatch:${session.linearSessionId}:${file}`, + sourceKey, role: valid && typeof sidecar.role === "string" ? sidecar.role @@ -1392,7 +1514,12 @@ export class SessionWorker { ? { cacheReadTokens: Number(sidecar.cache_read_tokens) } : {}), }; - if (!valid) + if ( + !valid && + !this.degradedLogged.has(degradedKey) && + !this.log.hasCodexInvocation(sourceKey) + ) { + this.degradedLogged.add(degradedKey); this.logger.error( jsonLog({ event: "dispatch_marker_ingest_degraded", @@ -1400,6 +1527,7 @@ export class SessionWorker { reason: "invalid_sidecar", }), ); + } const result = this.log.ingestCodexMarker( invocation, { diff --git a/daemon/test/config.test.ts b/daemon/test/config.test.ts index f798849..887d263 100644 --- a/daemon/test/config.test.ts +++ b/daemon/test/config.test.ts @@ -16,6 +16,10 @@ describe("loadConfig", () => { expect(config.webhookBaseUrl).toBe("http://127.0.0.1:8787"); expect(config.artifactToken).toBeUndefined(); expect(config.artifactsDir).toBe("/var/lib/linear-agent-daemon/artifacts"); + expect(config.dispatchQuarantineDir).toBe( + "/var/lib/linear-agent-daemon/dispatch-quarantine", + ); + expect(config.dispatchQuarantineAgeMs).toBe(86_400_000); expect(config).toMatchObject({ browserEnabled: false, playwrightMcpBin: "/usr/local/bin/playwright-mcp", playwrightChromeBin: "/usr/bin/google-chrome", browserAttemptTimeoutMs: 14_400_000 }); expect(config.artifactMaxBodyBytes).toBe(32 * 1024 * 1024); @@ -147,6 +151,21 @@ describe("loadConfig", () => { ARTIFACTS_DIR: "/srv/artifacts", ARTIFACT_MAX_BODY_BYTES: "4096" }); expect(config).toMatchObject({ artifactToken: "secret", artifactsDir: "/srv/artifacts", artifactMaxBodyBytes: 4096 }); }); + it("loads dispatch quarantine overrides", () => { + const config = loadConfig({ + ...base, + DB_PATH: "/state/events.db", + DISPATCH_QUARANTINE_DIR: " /srv/dispatch-quarantine ", + DISPATCH_QUARANTINE_AGE_MS: "172800000", + }); + expect(config).toMatchObject({ + dispatchQuarantineDir: "/srv/dispatch-quarantine", + dispatchQuarantineAgeMs: 172_800_000, + }); + expect(() => + loadConfig({ ...base, DISPATCH_QUARANTINE_AGE_MS: "0" }), + ).toThrow("DISPATCH_QUARANTINE_AGE_MS"); + }); it("loads strict browser capability overrides", () => { expect(loadConfig({ ...base, BROWSER_ENABLED: "1", PLAYWRIGHT_MCP_BIN: "/mcp", PLAYWRIGHT_CHROME_BIN: "/chrome", BROWSER_ATTEMPT_TIMEOUT_MS: "1234" })).toMatchObject({ browserEnabled: true, playwrightMcpBin: "/mcp", diff --git a/daemon/test/operations-ingress.test.ts b/daemon/test/operations-ingress.test.ts index 24c0f46..767ebb9 100644 --- a/daemon/test/operations-ingress.test.ts +++ b/daemon/test/operations-ingress.test.ts @@ -40,6 +40,8 @@ describe("operation drain through signed webhook ingress", () => { port: 0, bindAddr: "127.0.0.1", dbPath: f.db, + dispatchQuarantineDir: join(f.dir, "dispatch-quarantine"), + dispatchQuarantineAgeMs: 86_400_000, replayWindowMs: 60_000, linearGraphqlUrl: `http://127.0.0.1:${gatewayPort}/graphql`, linearTokenUrl: "http://unused", diff --git a/daemon/test/reconcile.test.ts b/daemon/test/reconcile.test.ts index ff02602..c39d975 100644 --- a/daemon/test/reconcile.test.ts +++ b/daemon/test/reconcile.test.ts @@ -15,6 +15,7 @@ afterEach(() => { for (const dir of dirs.splice(0)) rmSync(dir, { recursive: tru function config(overrides: Partial = {}): Config { return { port: 0, bindAddr: "127.0.0.1", dbPath: ":memory:", replayWindowMs: 60_000, + dispatchQuarantineDir: "/tmp/dispatch-quarantine", dispatchQuarantineAgeMs: 86_400_000, linearGraphqlUrl: "http://unused", linearTokenUrl: "http://unused", webhookBaseUrl: "https://agent.example.com", reconcileIntervalMs: 60_000, reconcileRequestTimeoutMs: 1_000, apps: { diff --git a/daemon/test/server.test.ts b/daemon/test/server.test.ts index f7e6148..4016202 100644 --- a/daemon/test/server.test.ts +++ b/daemon/test/server.test.ts @@ -24,6 +24,7 @@ function setup() { const log = new EventLog(join(dir, "events.db")); const config: Config = { port: 0, bindAddr: "127.0.0.1", dbPath: join(dir, "events.db"), replayWindowMs: 60_000, + dispatchQuarantineDir: join(dir, "dispatch-quarantine"), dispatchQuarantineAgeMs: 86_400_000, linearGraphqlUrl: "http://unused", linearTokenUrl: "http://unused", cliproxyEnvFile, apps: { planner: { name: "planner", webhookSecret: "planner-secret", staticToken: "p" }, diff --git a/daemon/test/sessions.test.ts b/daemon/test/sessions.test.ts index b115d7a..4b93f63 100644 --- a/daemon/test/sessions.test.ts +++ b/daemon/test/sessions.test.ts @@ -6,7 +6,9 @@ import { mkdirSync, readFileSync, readdirSync, + renameSync, rmSync, + utimesSync, writeFileSync, } from "node:fs"; import { createServer } from "node:http"; @@ -108,6 +110,8 @@ function setup() { port: 0, bindAddr: "127.0.0.1", dbPath: join(dir, "events.db"), + dispatchQuarantineDir: join(dir, "dispatch-quarantine"), + dispatchQuarantineAgeMs: 86_400_000, replayWindowMs: 60_000, linearGraphqlUrl: "http://unused", linearTokenUrl: "http://unused", @@ -267,6 +271,42 @@ function appendImplementer( ), }); } +function setTurnsStatus( + dbPath: string, + status: "done" | "deleted", +): void { + const db = new Database(dbPath); + try { + if (status === "done") + db.prepare( + "UPDATE turns SET status='done' WHERE status IN ('pending','running','awaiting_activity')", + ).run(); + else db.prepare("DELETE FROM turns").run(); + } finally { + db.close(); + } +} +function dispatchFixture( + worktree: string, + owner: string, + basename: string, +): { directory: string; files: string[] } { + const directory = join(worktree, ".codex-dispatches", owner); + const files = [ + `${basename}.done`, + `${basename}.prompt`, + `${basename}.md`, + `${basename}.log`, + `${basename}.sh`, + ]; + mkdirSync(directory, { recursive: true }); + for (const file of files) + writeFileSync( + join(directory, file), + file.endsWith(".done") ? "0\n" : `${file}\n`, + ); + return { directory, files }; +} function turnUsage(dbPath: string, id = 1): Record { const db = new Database(dbPath, { readonly: true }); try { @@ -2019,6 +2059,303 @@ describe("SessionWorker", () => { await worker.stop(); log.close(); }); + it("logs each degraded dispatch marker once across rescans and reason transitions", async () => { + const poster = new Poster(); + for (const scenario of ["invalid", "no-parent", "transition"] as const) { + const { dir, log, config } = setup(); + config.sessionConcurrency = 0; + const owner = + scenario === "invalid" + ? ownerOne + : scenario === "no-parent" + ? ownerTwo + : "a0000000-0000-0000-0000-000000000003"; + const worktree = join(dir, "dispatch-worktree"); + appendImplementer(log, `${scenario}-created`, owner); + log.updateSessionWorktree(owner, worktree, `${scenario}-branch`); + setTurnsStatus(config.dbPath, scenario === "invalid" ? "done" : "deleted"); + dispatchFixture( + worktree, + owner, + `backend-verifier-1700000000-1234-${scenario.length}`, + ); + const logger = new CapturingLogger(); + const worker = new SessionWorker( + log, + poster as unknown as LinearGateway, + config, + { logger }, + ); + await worker.ingestDispatches(); + if (scenario === "invalid") setTurnsStatus(config.dbPath, "done"); + if (scenario === "transition") { + append(log, `${scenario}-prompted`, owner, "prompted"); + setTurnsStatus(config.dbPath, "done"); + } + await worker.ingestDispatches(); + const degraded = logger + .entries() + .filter((entry) => entry.event === "dispatch_marker_ingest_degraded"); + expect(degraded, scenario).toHaveLength(1); + expect(degraded[0]?.reason).toBe( + scenario === "invalid" ? "invalid_sidecar" : "no_parent_turn", + ); + await worker.stop(); + log.close(); + } + }); + it("prunes a consumed marker from the degraded-log memo", async () => { + const { dir, log, config } = setup(); + const poster = new Poster(); + const logger = new CapturingLogger(); + config.sessionConcurrency = 0; + appendImplementer(log, "memo-prune-created", ownerOne); + setTurnsStatus(config.dbPath, "deleted"); + const worktree = join(dir, "dispatch-worktree"); + log.updateSessionWorktree(ownerOne, worktree, "memo-prune-branch"); + const basename = "backend-verifier-1700000000-1234-25"; + const fixture = dispatchFixture(worktree, ownerOne, basename); + const worker = new SessionWorker( + log, + poster as unknown as LinearGateway, + config, + { logger }, + ); + + await worker.ingestDispatches(); + for (const file of fixture.files) rmSync(join(fixture.directory, file)); + await worker.ingestDispatches(); + dispatchFixture(worktree, ownerOne, basename); + await worker.ingestDispatches(); + + expect( + logger + .entries() + .filter((entry) => entry.event === "dispatch_marker_ingest_degraded"), + ).toHaveLength(2); + await worker.stop(); + log.close(); + }); + it("quarantines every stale ingested sibling while retaining its invocation", async () => { + const { dir, log, config } = setup(); + const poster = new Poster(); + const now = Date.now(); + config.sessionConcurrency = 0; + config.dispatchQuarantineAgeMs = 1_000; + appendImplementer(log, "quarantine-created", ownerOne); + setTurnsStatus(config.dbPath, "done"); + const worktree = join(dir, "dispatch-worktree"); + log.updateSessionWorktree(ownerOne, worktree, "quarantine-branch"); + const basename = "backend-verifier-1700000000-1234-20"; + const fixture = dispatchFixture(worktree, ownerOne, basename); + const worker = new SessionWorker( + log, + poster as unknown as LinearGateway, + config, + { now: () => now }, + ); + await worker.ingestDispatches(); + setTurnsStatus(config.dbPath, "done"); + utimesSync( + join(fixture.directory, `${basename}.done`), + new Date(now - 2_000), + new Date(now - 2_000), + ); + await worker.ingestDispatches(); + const quarantine = join(config.dispatchQuarantineDir, ownerOne); + expect(readdirSync(quarantine).sort()).toEqual(fixture.files.sort()); + expect(readdirSync(fixture.directory)).toEqual([]); + expect(log.invocations(ownerOne)).toEqual([ + expect.objectContaining({ + sourceKey: `dispatch:${ownerOne}:${basename}.done`, + }), + ]); + await worker.stop(); + log.close(); + }); + it("leaves a young ingested dispatch bundle in the live consume path", async () => { + const { dir, log, config } = setup(); + const poster = new Poster(); + config.sessionConcurrency = 0; + config.dispatchQuarantineAgeMs = 10_000; + appendImplementer(log, "young-created", ownerOne); + setTurnsStatus(config.dbPath, "done"); + const worktree = join(dir, "dispatch-worktree"); + log.updateSessionWorktree(ownerOne, worktree, "young-branch"); + const fixture = dispatchFixture( + worktree, + ownerOne, + "backend-verifier-1700000000-1234-21", + ); + const worker = new SessionWorker( + log, + poster as unknown as LinearGateway, + config, + ); + await worker.ingestDispatches(); + expect(readdirSync(fixture.directory).sort()).toEqual( + fixture.files.sort(), + ); + expect(existsSync(config.dispatchQuarantineDir)).toBe(false); + await worker.stop(); + log.close(); + }); + it("leaves a stale bundle when its owner opens a turn during quarantine moves", async () => { + const { dir, log, config } = setup(); + const poster = new Poster(); + const now = Date.now(); + config.sessionConcurrency = 0; + config.dispatchQuarantineAgeMs = 1_000; + appendImplementer(log, "quarantine-race-created", ownerOne); + setTurnsStatus(config.dbPath, "done"); + const worktree = join(dir, "dispatch-worktree"); + log.updateSessionWorktree(ownerOne, worktree, "quarantine-race-branch"); + const basename = "backend-verifier-1700000000-1234-24"; + const fixture = dispatchFixture(worktree, ownerOne, basename); + const worker = new SessionWorker( + log, + poster as unknown as LinearGateway, + config, + { now: () => now }, + ); + await worker.ingestDispatches(); + setTurnsStatus(config.dbPath, "done"); + utimesSync( + join(fixture.directory, `${basename}.done`), + new Date(now - 2_000), + new Date(now - 2_000), + ); + const hasOpenTurn = vi + .spyOn(log, "hasOpenTurn") + .mockReturnValueOnce(false) + .mockReturnValueOnce(false) + .mockReturnValueOnce(false) + .mockReturnValueOnce(true); + + await worker.ingestDispatches(); + expect(hasOpenTurn).toHaveBeenCalledTimes(4); + expect( + readdirSync(fixture.directory).includes(`${basename}.done`), + ).toBe(true); + expect(readdirSync(fixture.directory)).toHaveLength( + fixture.files.length - 1, + ); + const quarantine = join(config.dispatchQuarantineDir, ownerOne); + expect(readdirSync(quarantine)).toHaveLength(1); + expect(log.invocations(ownerOne)).toHaveLength(1); + hasOpenTurn.mockRestore(); + + await worker.ingestDispatches(); + expect(readdirSync(fixture.directory)).toEqual([]); + expect(readdirSync(quarantine).sort()).toEqual(fixture.files.sort()); + expect(log.invocations(ownerOne)).toHaveLength(1); + await worker.stop(); + log.close(); + }); + it("finishes a partial quarantine without overwriting archived siblings", async () => { + const { dir, log, config } = setup(); + const poster = new Poster(); + const now = Date.now(); + config.sessionConcurrency = 0; + config.dispatchQuarantineAgeMs = 1_000; + appendImplementer(log, "partial-created", ownerOne); + setTurnsStatus(config.dbPath, "done"); + const worktree = join(dir, "dispatch-worktree"); + log.updateSessionWorktree(ownerOne, worktree, "partial-branch"); + const basename = "backend-verifier-1700000000-1234-22"; + const fixture = dispatchFixture(worktree, ownerOne, basename); + const worker = new SessionWorker( + log, + poster as unknown as LinearGateway, + config, + { now: () => now }, + ); + await worker.ingestDispatches(); + setTurnsStatus(config.dbPath, "done"); + const quarantine = join(config.dispatchQuarantineDir, ownerOne); + mkdirSync(quarantine, { recursive: true }); + renameSync( + join(fixture.directory, `${basename}.prompt`), + join(quarantine, `${basename}.prompt`), + ); + writeFileSync(join(quarantine, `${basename}.md`), "archived report\n"); + utimesSync( + join(fixture.directory, `${basename}.done`), + new Date(now - 2_000), + new Date(now - 2_000), + ); + await worker.ingestDispatches(); + expect( + readFileSync(join(quarantine, `${basename}.prompt`), "utf8"), + ).toBe(`${basename}.prompt\n`); + expect(readFileSync(join(quarantine, `${basename}.md`), "utf8")).toBe( + "archived report\n", + ); + expect(readFileSync(join(quarantine, `${basename}.md.1`), "utf8")).toBe( + `${basename}.md\n`, + ); + expect(readdirSync(quarantine).sort()).toEqual( + [...fixture.files, `${basename}.md.1`].sort(), + ); + expect(readdirSync(fixture.directory)).toEqual([]); + await worker.stop(); + log.close(); + }); + it("logs a quarantine failure and completes the bundle on the next scan", async () => { + const { dir, log, config } = setup(); + const poster = new Poster(); + const logger = new CapturingLogger(); + const now = Date.now(); + config.sessionConcurrency = 0; + config.dispatchQuarantineAgeMs = 1_000; + appendImplementer(log, "quarantine-failure-created", ownerOne); + setTurnsStatus(config.dbPath, "done"); + const worktree = join(dir, "dispatch-worktree"); + log.updateSessionWorktree(ownerOne, worktree, "quarantine-failure-branch"); + const basename = "backend-verifier-1700000000-1234-23"; + const fixture = dispatchFixture(worktree, ownerOne, basename); + const worker = new SessionWorker( + log, + poster as unknown as LinearGateway, + config, + { logger, now: () => now }, + ); + await worker.ingestDispatches(); + setTurnsStatus(config.dbPath, "done"); + utimesSync( + join(fixture.directory, `${basename}.done`), + new Date(now - 2_000), + new Date(now - 2_000), + ); + mkdirSync(config.dispatchQuarantineDir, { recursive: true }); + const obstruction = join(config.dispatchQuarantineDir, ownerOne); + writeFileSync(obstruction, "not a directory\n"); + + await worker.ingestDispatches(); + expect( + logger + .entries() + .filter( + (entry) => entry.event === "dispatch_marker_quarantine_failed", + ), + ).toHaveLength(1); + expect(readdirSync(fixture.directory).sort()).toEqual( + fixture.files.sort(), + ); + + rmSync(obstruction); + await worker.ingestDispatches(); + const quarantine = join(config.dispatchQuarantineDir, ownerOne); + expect(readdirSync(quarantine).sort()).toEqual(fixture.files.sort()); + expect(readdirSync(fixture.directory)).toEqual([]); + expect(log.invocations(ownerOne)).toEqual([ + expect.objectContaining({ + sourceKey: `dispatch:${ownerOne}:${basename}.done`, + }), + ]); + await worker.stop(); + log.close(); + }); it("does not enqueue a marker while its owner has a running turn", async () => { const { log, config } = setup(); const poster = new Poster();