From f26737d9b783c8e57b8d503309518d9f792a479a Mon Sep 17 00:00:00 2001 From: Tyler Brown Date: Thu, 30 Jul 2026 11:55:15 -0700 Subject: [PATCH 1/3] fix(daemon): add event-driven worktree reconcile check so cleanup self-heals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Issue-completed webhook was never subscribed on the live app, so the cleanup pipeline never ran and 32 worktrees (18 GB) accumulated. The new WorktreeReconciler sweeps at startup, on turn completion, and on Issue-completed events (no standing timer), confirms each worktree's issue state with Linear, and closes out resolved ones through the existing CleanupWorker path — preserving dirty or PR-less state to the agents/ branch (or a git bundle) before any force removal. Unlinked worktrees get a persisted 14-day grace clock (WORKTREE_UNLINKED_GRACE_DAYS); lookup errors always skip; foreign directories are never touched. Closes #128 Co-Authored-By: Claude Fable 5 --- daemon/README.md | 9 +- daemon/ops/runbook.md | 8 +- daemon/src/cleanup.ts | 68 ++++- daemon/src/config.ts | 11 + daemon/src/eventlog.ts | 121 ++++++++ daemon/src/index.ts | 32 ++- daemon/src/linear.ts | 47 ++++ daemon/src/server.ts | 7 + daemon/src/sessions.ts | 8 +- daemon/src/worktree-reconciler.ts | 205 ++++++++++++++ daemon/src/worktrees.ts | 127 +++++++++ daemon/test/cleanup.test.ts | 44 ++- daemon/test/config.test.ts | 21 ++ daemon/test/linear.test.ts | 53 ++++ daemon/test/server.test.ts | 28 +- daemon/test/sessions.test.ts | 4 +- daemon/test/worktree-reconciler.test.ts | 350 ++++++++++++++++++++++++ daemon/test/worktrees.test.ts | 34 +++ docs/linear-agent-daemon-setup.md | 8 +- 19 files changed, 1150 insertions(+), 35 deletions(-) create mode 100644 daemon/src/worktree-reconciler.ts create mode 100644 daemon/test/worktree-reconciler.test.ts diff --git a/daemon/README.md b/daemon/README.md index 5f54968..55f4603 100644 --- a/daemon/README.md +++ b/daemon/README.md @@ -6,10 +6,13 @@ to SQLite, acknowledges new sessions asynchronously, and runs bloom-planner disc per-issue git worktrees. Planner turns stream Claude progress to Linear, persist terminal activities for retry, and resume the stored Claude session on follow-up prompts. Implementer assignments run a fresh, unattended literal `/do ` turn in the same issue worktree, -durably attach an opened PR to the Linear session, and clean up clean worktrees after completed -Issue webhooks. Follow-up replies to an implementer session resume its stored Claude session +durably attach an opened PR to the Linear session, and clean up worktrees after Linear confirms +their issues completed or canceled. Issue webhooks remain the fast path; an event-driven +reconciliation sweep also runs at startup and after session completion, with no standing sweep +timer. Follow-up replies to an implementer session resume its stored Claude session the same way planner prompts do, so a human can answer an implementer's question mid-stream. -Dirty worktrees are retained and reported to the session. +Dirty or PR-less state is pushed to its `agents/` branch (or saved as a git bundle) before +force removal. ## Local checks diff --git a/daemon/ops/runbook.md b/daemon/ops/runbook.md index 9724592..6c183f3 100644 --- a/daemon/ops/runbook.md +++ b/daemon/ops/runbook.md @@ -167,6 +167,8 @@ LINEAR_MCP_MONITOR_TIMEOUT_MS=10000 SESSIONS_ENABLED=1 TARGET_REPO_PATH=/var/lib/linear-agent-daemon/repos/bloom-mono WORKTREES_ROOT=/var/lib/linear-agent-daemon/worktrees +WORKTREE_UNLINKED_GRACE_DAYS=14 +WORKTREE_BUNDLES_DIR=/var/lib/linear-agent-daemon/worktree-bundles LINEAR_API_KEY=... PLANNER_HARNESS=claude IMPLEMENTER_HARNESS=claude @@ -793,7 +795,11 @@ elicitation/human-input request, `/do` starts on `agents/`, a PR is the PR appears in session external URLs. Confirm the final `/do` text matches PR extraction. Move the issue to a workflow state whose stable type is `completed`; verify the Issues webhook arrives and both worktree and local branch disappear. Repeat with an uncommitted -file and verify the worktree remains and a thought names its path. Confirm the full flow +file and verify the worktree is removed only after a thought names the pushed +`agents/` branch or a bundle under `WORKTREE_BUNDLES_DIR`. Restore a pushed +branch with `git fetch origin agents/:agents/`; inspect or restore +a bundle with `git bundle verify ` followed by +`git fetch 'refs/heads/*:refs/remotes/bundle/*'`. Confirm the full flow under systemd hardening. ```bash diff --git a/daemon/src/cleanup.ts b/daemon/src/cleanup.ts index bc97ed1..87a448f 100644 --- a/daemon/src/cleanup.ts +++ b/daemon/src/cleanup.ts @@ -27,6 +27,9 @@ export interface CleanupWorkerOptions { logger?: Logger; relay?: OtlpRelay; ingestDispatches?: () => Promise; + worktrees?: WorktreeManager; + bundlesDir?: string; + onIssueFinalized?: () => void; } export class CleanupWorker { @@ -39,13 +42,14 @@ export class CleanupWorker { constructor( private readonly log: EventLog, private readonly gateway: LinearGateway, - worktreesRoot: string, + private readonly worktreesRoot: string, targetRepoPath: string, private readonly options: CleanupWorkerOptions = {}, ) { this.now = options.now ?? Date.now; this.logger = options.logger ?? console; - this.worktrees = new WorktreeManager(worktreesRoot, targetRepoPath); + this.worktrees = + options.worktrees ?? new WorktreeManager(worktreesRoot, targetRepoPath); } private readonly worktrees: WorktreeManager; start(): void { @@ -82,6 +86,7 @@ export class CleanupWorker { await this.postNotification(note); } private async process(job: CleanupJobRow): Promise { + let preservationAttempt = false; try { await this.options.ingestDispatches?.(); if (!(await this.finalizeIssue(job))) { @@ -97,29 +102,64 @@ export class CleanupWorker { !session?.worktreePath || !(await this.worktrees.isPresent(session.worktreePath)) ) { - await this.worktrees.remove(job.issueIdentifier); + preservationAttempt = true; + const result = await this.worktrees.preserveAndRemove( + job.issueIdentifier, + this.options.bundlesDir ?? `${this.worktreesRoot}/../worktree-bundles`, + { alwaysPreserve: true }, + ); + preservationAttempt = false; this.log.clearSessionWorktrees(job.issueIdentifier); + this.log.noteCleanupOutcome( + job.id, + `Worktree cleanup completed; state ${result.detail}.`, + this.now(), + ); this.log.markCleanupDone(job.id); + this.options.onIssueFinalized?.(); return; } if (await this.worktrees.isClean(session.worktreePath)) { if (!this.log.hasExternalUrl(job.linearSessionId)) { - this.log.retainCleanup( + preservationAttempt = true; + const result = await this.worktrees.preserveAndRemove( + job.issueIdentifier, + this.options.bundlesDir ?? + `${this.worktreesRoot}/../worktree-bundles`, + { alwaysPreserve: true }, + ); + preservationAttempt = false; + this.log.clearSessionWorktrees(job.issueIdentifier); + this.log.noteCleanupOutcome( job.id, - `Worktree retained because no pull request was recorded; possible unpushed work is preserved: ${session.worktreePath}`, + `Worktree cleanup completed; state ${result.detail}.`, this.now(), ); + this.log.markCleanupDone(job.id); + this.options.onIssueFinalized?.(); return; } await this.worktrees.remove(job.issueIdentifier); this.log.clearSessionWorktrees(job.issueIdentifier); this.log.markCleanupDone(job.id); - } else - this.log.retainCleanup( + this.options.onIssueFinalized?.(); + } else { + preservationAttempt = true; + const result = await this.worktrees.preserveAndRemove( + job.issueIdentifier, + this.options.bundlesDir ?? `${this.worktreesRoot}/../worktree-bundles`, + { alwaysPreserve: true }, + ); + preservationAttempt = false; + this.log.clearSessionWorktrees(job.issueIdentifier); + this.log.noteCleanupOutcome( job.id, - `Worktree retained because it is dirty: ${session.worktreePath}`, + `Worktree cleanup completed; state ${result.detail}.`, this.now(), ); + this.log.markCleanupDone(job.id); + this.options.onIssueFinalized?.(); + } } catch (error) { const message = error instanceof Error ? error.message : String(error); if ( @@ -132,10 +172,18 @@ export class CleanupWorker { this.now() + Math.min(60_000, 1000 * 2 ** Math.min(job.attempts, 6)), ); else { - this.log.failCleanup(job.id, message); + if (preservationAttempt) + this.log.retainCleanup( + job.id, + `Worktree retained because preservation failed: ${message}`, + this.now(), + ); + else this.log.failCleanup(job.id, message); this.logger.error( JSON.stringify({ - event: "cleanup_failed", + event: preservationAttempt + ? "cleanup_preservation_failed" + : "cleanup_failed", jobId: job.id, error: message, }), diff --git a/daemon/src/config.ts b/daemon/src/config.ts index 4a5dadf..f14b632 100644 --- a/daemon/src/config.ts +++ b/daemon/src/config.ts @@ -47,6 +47,8 @@ export interface Config { apps: Record; sessionsEnabled: boolean; worktreesRoot: string; + worktreeUnlinkedGraceMs: number; + worktreeBundlesDir: string; targetRepoPath?: string; claudeArgv: string[]; claudexArgv?: string[]; @@ -195,6 +197,15 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { apps: { planner: appConfig(env, "planner", testMode), implementer: appConfig(env, "implementer", testMode) }, sessionsEnabled, worktreesRoot: env.WORKTREES_ROOT?.trim() || `${dirname(dbPath)}/worktrees`, + worktreeUnlinkedGraceMs: + positiveInteger(env, "WORKTREE_UNLINKED_GRACE_DAYS", 14) * + 24 * + 60 * + 60 * + 1000, + worktreeBundlesDir: + env.WORKTREE_BUNDLES_DIR?.trim() || + `${dirname(dbPath)}/worktree-bundles`, ...(targetRepoPath ? { targetRepoPath } : {}), claudeArgv, ...(claudexArgv ? { claudexArgv } : {}), diff --git a/daemon/src/eventlog.ts b/daemon/src/eventlog.ts index 0f75f6d..033416c 100644 --- a/daemon/src/eventlog.ts +++ b/daemon/src/eventlog.ts @@ -250,6 +250,12 @@ export interface CleanupNotificationRow { nextAttemptAt: number; createdAt: number; } +export interface WorktreeUnlinkedObservation { + identifier: string; + path: string; + firstObservedAt: number; + lastSeenAt: number; +} export interface StopAckRow { sourceActivityId: string; eventId: number; @@ -430,6 +436,12 @@ export class EventLog { activity_id TEXT NOT NULL UNIQUE, body TEXT NOT NULL, status TEXT NOT NULL CHECK(status IN ('pending','posted','failed')), attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL, error TEXT ); + CREATE TABLE IF NOT EXISTS worktree_unlinked_observations ( + identifier TEXT PRIMARY KEY, + path TEXT NOT NULL, + first_observed_at INTEGER NOT NULL, + last_seen_at INTEGER NOT NULL + ); CREATE TABLE IF NOT EXISTS stop_acks ( source_activity_id TEXT PRIMARY KEY, event_id INTEGER NOT NULL REFERENCES events(id), app TEXT NOT NULL CHECK(app IN ('planner','implementer')), linear_session_id TEXT NOT NULL, @@ -1992,6 +2004,92 @@ export class EventLog { randomUUID(), ); } + enqueueCleanupFromReconcile( + issueId: string, + identifier: string, + now = Date.now(), + ): boolean { + const session = this.sessionByIssueIdentifier(identifier); + if (!session) return false; + return ( + this.db + .prepare( + `INSERT OR IGNORE INTO cleanup_jobs + (issue_id,issue_identifier,linear_session_id,app,status,next_attempt_at,created_at,notify_activity_id) + VALUES (?,?,?,?, 'pending',?,?,?)`, + ) + .run( + issueId, + identifier, + session.linearSessionId, + session.app, + now, + now, + randomUUID(), + ).changes > 0 + ); + } + cleanupJobByIssueIdentifier( + identifier: string, + ): CleanupJobRow | undefined { + return this.db + .prepare( + `SELECT id,issue_id issueId,issue_identifier issueIdentifier,linear_session_id linearSessionId, + app,status,attempts,created_at createdAt,claimed_at claimedAt,notify_activity_id notifyActivityId + FROM cleanup_jobs WHERE issue_identifier=? ORDER BY id DESC LIMIT 1`, + ) + .get(identifier) as CleanupJobRow | undefined; + } + retainedCleanups(): CleanupJobRow[] { + return this.db + .prepare( + `SELECT id,issue_id issueId,issue_identifier issueIdentifier,linear_session_id linearSessionId, + app,status,attempts,created_at createdAt,claimed_at claimedAt,notify_activity_id notifyActivityId + FROM cleanup_jobs WHERE status='retained' ORDER BY id`, + ) + .all() as CleanupJobRow[]; + } + repromoteRetainedCleanup(id: number, now = Date.now()): boolean { + return ( + this.db + .prepare( + `UPDATE cleanup_jobs SET status='pending',claimed_at=NULL, + next_attempt_at=?,error=NULL WHERE id=? AND status='retained'`, + ) + .run(now, id).changes > 0 + ); + } + observeUnlinkedWorktree( + identifier: string, + path: string, + now = Date.now(), + ): WorktreeUnlinkedObservation { + this.db + .prepare( + `INSERT INTO worktree_unlinked_observations + (identifier,path,first_observed_at,last_seen_at) VALUES (?,?,?,?) + ON CONFLICT(identifier) DO UPDATE SET path=excluded.path,last_seen_at=excluded.last_seen_at`, + ) + .run(identifier, path, now, now); + return this.unlinkedObservation(identifier)!; + } + clearUnlinkedObservation(identifier: string): void { + this.db + .prepare( + "DELETE FROM worktree_unlinked_observations WHERE identifier=?", + ) + .run(identifier); + } + unlinkedObservation( + identifier: string, + ): WorktreeUnlinkedObservation | undefined { + return this.db + .prepare( + `SELECT identifier,path,first_observed_at firstObservedAt, + last_seen_at lastSeenAt FROM worktree_unlinked_observations WHERE identifier=?`, + ) + .get(identifier) as WorktreeUnlinkedObservation | undefined; + } claimNextCleanup(now = Date.now()): CleanupJobRow | undefined { return this.db.transaction(() => { const candidate = this.db @@ -2078,6 +2176,29 @@ export class EventLog { .run(id); })(); } + noteCleanupOutcome(id: number, body: string, now = Date.now()): void { + this.db.transaction(() => { + const job = this.cleanupById(id); + if (!job) throw new Error(`Missing cleanup ${id}`); + this.db + .prepare( + `INSERT INTO cleanup_notifications + (job_id,app,linear_session_id,activity_id,body,status,next_attempt_at,created_at) + VALUES (?,?,?,?,?,'pending',?,?) + ON CONFLICT(job_id) DO UPDATE SET body=excluded.body,status='pending', + attempts=0,next_attempt_at=excluded.next_attempt_at,created_at=excluded.created_at,error=NULL`, + ) + .run( + id, + job.app, + job.linearSessionId, + job.notifyActivityId, + body, + now, + now, + ); + })(); + } pendingCleanupNotifications(now = Date.now()): CleanupNotificationRow[] { return this.db .prepare( diff --git a/daemon/src/index.ts b/daemon/src/index.ts index 06b28fc..04b1aa7 100644 --- a/daemon/src/index.ts +++ b/daemon/src/index.ts @@ -15,6 +15,8 @@ import { ArtifactStore } from "./artifacts.js"; import { OtlpRelay } from "./otel-relay.js"; import { resolveOtlpTraces } from "./otel.js"; import { LinearMcpMonitor } from "./linear-mcp-monitor.js"; +import { WorktreeManager } from "./worktrees.js"; +import { WorktreeReconciler } from "./worktree-reconciler.js"; const config = loadConfig(); let log: EventLog; @@ -30,6 +32,10 @@ const gateway = new LinearGateway( const worker = new AckWorker(log, gateway); let cleanupWorker: CleanupWorker | undefined; let sessionWorker: SessionWorker | undefined; +let worktreeReconciler: WorktreeReconciler | undefined; +const worktrees = config.sessionsEnabled + ? new WorktreeManager(config.worktreesRoot, config.targetRepoPath!) + : undefined; const linearMcpMonitor = config.sessionsEnabled ? new LinearMcpMonitor({ url: config.linearMcpUrl, @@ -74,7 +80,11 @@ await relay?.start(); sessionWorker = config.sessionsEnabled ? new SessionWorker(log, gateway, config, { ...(relay ? { relay } : {}), - onTurnComplete: () => void cleanupWorker?.trigger(), + worktrees: worktrees!, + onTurnComplete: () => { + void cleanupWorker?.trigger(); + void worktreeReconciler?.trigger(); + }, }) : undefined; cleanupWorker = config.sessionsEnabled @@ -85,11 +95,28 @@ cleanupWorker = config.sessionsEnabled config.targetRepoPath!, { ...(relay ? { relay } : {}), + worktrees: worktrees!, + bundlesDir: config.worktreeBundlesDir, + onIssueFinalized: () => void worktreeReconciler?.trigger(), ingestDispatches: () => sessionWorker?.ingestDispatches() ?? Promise.resolve(), }, ) : undefined; +worktreeReconciler = config.sessionsEnabled + ? new WorktreeReconciler( + log, + gateway, + worktrees!, + { + worktreesRoot: config.worktreesRoot, + worktreeUnlinkedGraceMs: config.worktreeUnlinkedGraceMs, + worktreeBundlesDir: config.worktreeBundlesDir, + ...(config.ntfyUrl ? { ntfyUrl: config.ntfyUrl } : {}), + }, + { onEnqueued: () => void cleanupWorker?.trigger() }, + ) + : undefined; const triggerWorkers = () => { worker.trigger(); sessionWorker?.trigger(); @@ -109,6 +136,7 @@ const server = new WebhookServer({ config, log, onInserted: triggerWorkers, + onIssueCompleted: () => void worktreeReconciler?.trigger(), onStop, ...(artifactStore ? { artifactStore } : {}), }); @@ -135,6 +163,7 @@ worker.start(); linearMcpMonitor?.start(); await sessionWorker?.start(); cleanupWorker?.start(); +worktreeReconciler?.start(); reconcileWorker?.start(); const address = await server.listen(); console.log( @@ -161,6 +190,7 @@ async function shutdown(signal: string): Promise { }), ); await reconcileWorker?.stop(); + await worktreeReconciler?.stop(); await linearMcpMonitor?.stop(); await server.close(); await worker.stop(); diff --git a/daemon/src/linear.ts b/daemon/src/linear.ts index 4f79f47..6632d41 100644 --- a/daemon/src/linear.ts +++ b/daemon/src/linear.ts @@ -35,6 +35,10 @@ export interface AgentPromptActivity { createdAt: number; signal?: string; } +export type IssueStateResult = + | { kind: "found"; issueId: string; stateType: string } + | { kind: "not_found" } + | { kind: "error"; error: string }; interface Logger { warn(...args: unknown[]): void; } interface TokenResponse { access_token?: unknown; expires_in?: unknown; error?: unknown; error_description?: unknown; } @@ -310,6 +314,49 @@ export class LinearGateway { }); } + async issueState( + app: AppName, + identifier: string, + deadlineAt = this.now() + 10_000, + ): Promise { + try { + const data = await this.withLinearClient( + app, + deadlineAt, + "Linear issue-state request", + (client) => + this.rawRequest<{ + issue: { + id: string; + identifier: string; + state: { type: string } | null; + } | null; + }>( + client, + `query WorktreeIssueState($id: String!) { + issue(id: $id) { id identifier state { type } } + }`, + { id: identifier }, + ), + ); + if (!data.issue) return { kind: "not_found" }; + return { + kind: "found", + issueId: data.issue.id, + stateType: data.issue.state?.type ?? "", + }; + } catch (error) { + const message = errorMessage(error); + if ( + /entity.+not found|issue.+not found|could not find.+issue|does not exist/i.test( + message, + ) + ) + return { kind: "not_found" }; + return { kind: "error", error: message }; + } + } + async listDelegatedIssueAgentSessions(app: AppName, appActorId: string, deadlineAt = this.now() + 10_000): Promise { return this.withLinearClient(app, deadlineAt, "Linear delegated issue request", async client => { const sessions = new Map(); diff --git a/daemon/src/server.ts b/daemon/src/server.ts index 904ccf2..4835b78 100644 --- a/daemon/src/server.ts +++ b/daemon/src/server.ts @@ -16,6 +16,7 @@ export interface WebhookServerOptions { log: EventLog; artifactStore?: ArtifactStore; onInserted?: () => void; + onIssueCompleted?: () => void; onStop?: (agentSessionId: string) => void; logger?: Pick; } @@ -114,6 +115,12 @@ export class WebhookServer { inserted: result.inserted })); this.json(response, 200, { ok: true }); if (result.inserted) this.options.onInserted?.(); + if ( + result.inserted && + event.type === "Issue" && + event.stateType === "completed" + ) + this.options.onIssueCompleted?.(); if (result.stop) this.options.onStop?.(result.stop.agentSessionId); } catch (error) { this.logger.error(JSON.stringify({ level: "error", event: "request_failed", error: error instanceof Error ? error.message : String(error) })); diff --git a/daemon/src/sessions.ts b/daemon/src/sessions.ts index 670c2d7..88b6670 100644 --- a/daemon/src/sessions.ts +++ b/daemon/src/sessions.ts @@ -61,6 +61,7 @@ export interface SessionWorkerOptions { attachmentTimeoutMs?: number; onTurnComplete?: () => void; relay?: OtlpRelay; + worktrees?: WorktreeManager; } export type ShutdownPolicy = "recover" | "hard_restart"; @@ -350,10 +351,9 @@ export class SessionWorker { ) { this.now = options.now ?? Date.now; this.logger = options.logger ?? console; - this.worktrees = new WorktreeManager( - config.worktreesRoot, - config.targetRepoPath!, - ); + this.worktrees = + options.worktrees ?? + new WorktreeManager(config.worktreesRoot, config.targetRepoPath!); } async start(): Promise { this.stopped = false; diff --git a/daemon/src/worktree-reconciler.ts b/daemon/src/worktree-reconciler.ts new file mode 100644 index 0000000..e8d6363 --- /dev/null +++ b/daemon/src/worktree-reconciler.ts @@ -0,0 +1,205 @@ +import { readdir } from "node:fs/promises"; +import { resolve } from "node:path"; +import type { EventLog } from "./eventlog.js"; +import type { LinearGateway } from "./linear.js"; +import type { WorktreeManager } from "./worktrees.js"; + +interface Logger { + log(...args: unknown[]): void; + error(...args: unknown[]): void; +} + +export interface WorktreeReconcilerConfig { + worktreesRoot: string; + worktreeUnlinkedGraceMs: number; + worktreeBundlesDir: string; + ntfyUrl?: string; +} + +export interface WorktreeReconcilerOptions { + now?: () => number; + logger?: Logger; + onEnqueued?: () => void; +} + +export class WorktreeReconciler { + private stopped = false; + private running: Promise | undefined; + private readonly now: () => number; + private readonly logger: Logger; + + constructor( + private readonly log: EventLog, + private readonly gateway: LinearGateway, + private readonly worktrees: WorktreeManager, + private readonly config: WorktreeReconcilerConfig, + private readonly options: WorktreeReconcilerOptions = {}, + ) { + this.now = options.now ?? Date.now; + this.logger = options.logger ?? console; + } + + start(): void { + this.stopped = false; + void this.trigger(); + } + + trigger(): Promise { + if (this.stopped) return Promise.resolve(); + return (this.running ??= this.sweep().finally(() => { + this.running = undefined; + })); + } + + async stop(): Promise { + this.stopped = true; + await this.running; + } + + private async sweep(): Promise { + let entries: Array<{ name: string; isDirectory(): boolean }> = []; + try { + entries = await readdir(this.config.worktreesRoot, { + withFileTypes: true, + }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + const identifiers = new Set( + entries.filter((entry) => entry.isDirectory()).map((entry) => entry.name), + ); + for (const job of this.log.retainedCleanups()) + identifiers.add(job.issueIdentifier); + + for (const identifier of identifiers) + await this.reconcileCandidate(identifier); + } + + private async reconcileCandidate(identifier: string): Promise { + const path = resolve(this.config.worktreesRoot, identifier); + if (!(await this.worktrees.isOwned(path))) { + this.logger.log( + JSON.stringify({ + event: "worktree_reconcile_skipped", + identifier, + reason: "foreign_or_invalid", + }), + ); + return; + } + + const state = /^[A-Za-z0-9]+-[0-9]+$/.test(identifier) + ? await this.gateway.issueState( + "implementer", + identifier, + this.now() + 10_000, + ) + : ({ kind: "not_found" } as const); + if (state.kind === "error") { + this.logger.error( + JSON.stringify({ + event: "worktree_reconcile_lookup_failed", + identifier, + error: state.error, + }), + ); + return; + } + if (state.kind === "not_found") { + const observation = this.log.observeUnlinkedWorktree( + identifier, + path, + this.now(), + ); + if ( + this.now() - observation.firstObservedAt < + this.config.worktreeUnlinkedGraceMs + ) + return; + try { + const result = await this.worktrees.preserveAndRemove( + identifier, + this.config.worktreeBundlesDir, + { alwaysPreserve: true }, + ); + this.log.clearUnlinkedObservation(identifier); + this.notify(identifier, result.detail, "unlinked_grace_expired"); + } catch (error) { + this.logger.error( + JSON.stringify({ + event: "worktree_reconcile_preservation_failed", + identifier, + error: error instanceof Error ? error.message : String(error), + }), + ); + } + return; + } + + this.log.clearUnlinkedObservation(identifier); + if (!["completed", "canceled"].includes(state.stateType)) return; + const existing = this.log.cleanupJobByIssueIdentifier(identifier); + if (existing?.status === "retained") { + if (this.log.repromoteRetainedCleanup(existing.id, this.now())) + this.options.onEnqueued?.(); + return; + } + const session = this.log.sessionByIssueIdentifier(identifier); + if (session) { + if ( + this.log.enqueueCleanupFromReconcile( + state.issueId, + identifier, + this.now(), + ) + ) + this.options.onEnqueued?.(); + return; + } + try { + const result = await this.worktrees.preserveAndRemove( + identifier, + this.config.worktreeBundlesDir, + { alwaysPreserve: true }, + ); + this.notify(identifier, result.detail, "resolved_without_session"); + } catch (error) { + this.logger.error( + JSON.stringify({ + event: "worktree_reconcile_preservation_failed", + identifier, + error: error instanceof Error ? error.message : String(error), + }), + ); + } + } + + private notify(identifier: string, detail: string, reason: string): void { + const body = `Worktree ${identifier} reconciled; state ${detail}.`; + this.logger.log( + JSON.stringify({ + event: "worktree_reconciled", + identifier, + reason, + destination: detail, + }), + ); + if (!this.config.ntfyUrl) return; + void fetch(this.config.ntfyUrl, { + method: "POST", + headers: { + Title: "Linear daemon worktree reconciled", + Priority: "default", + }, + body, + }).catch((error) => + this.logger.error( + JSON.stringify({ + event: "worktree_reconcile_notification_failed", + identifier, + error: error instanceof Error ? error.message : String(error), + }), + ), + ); + } +} diff --git a/daemon/src/worktrees.ts b/daemon/src/worktrees.ts index 2ddc591..4d97134 100644 --- a/daemon/src/worktrees.ts +++ b/daemon/src/worktrees.ts @@ -6,6 +6,10 @@ import { promisify } from "node:util"; const exec = promisify(execFile); export interface Worktree { path: string; branch: string; } +export interface PreservationResult { + preserved: "pushed" | "bundled" | "none"; + detail: string; +} export class WorktreeManager { private mutation: Promise = Promise.resolve(); @@ -27,6 +31,15 @@ export class WorktreeManager { async isPresent(path: string): Promise { return this.exists(path); } + async isOwned(path: string): Promise { + try { + await this.validate(path); + return true; + } catch { + return false; + } + } + async remove(rawIdentifier: string): Promise { const previous = this.mutation; let release!: () => void; @@ -47,6 +60,120 @@ export class WorktreeManager { } finally { release(); } } + async preserveAndRemove( + rawIdentifier: string, + bundlesDir: string, + opts: { alwaysPreserve: boolean }, + ): Promise { + const previous = this.mutation; + let release!: () => void; + this.mutation = new Promise((resolve) => { + release = resolve; + }); + await previous; + try { + const identifier = + rawIdentifier.replace(/[^A-Za-z0-9-]/g, "-") || "issue"; + const path = resolve(this.root, identifier); + const branch = `agents/${identifier}`; + let result: PreservationResult = { + preserved: "none", + detail: "no preservation required", + }; + if (await this.exists(path)) { + await this.validate(path); + const wasDirty = + (await this.git(["status", "--porcelain"], path)).trim() !== ""; + if (wasDirty) { + await this.git(["add", "-A"], path); + await this.git( + [ + "-c", + "user.email=daemon", + "-c", + "user.name=daemon", + "commit", + "-m", + "preserve: agent worktree state before cleanup", + ], + path, + ); + } + if (wasDirty || opts.alwaysPreserve) { + try { + await this.git( + ["push", "origin", `${branch}:${branch}`], + path, + ); + result = { + preserved: "pushed", + detail: `pushed to ${branch}`, + }; + } catch { + await mkdir(bundlesDir, { recursive: true }); + const bundlePath = resolve( + bundlesDir, + `${identifier}-${Date.now()}.bundle`, + ); + await this.git(["bundle", "create", bundlePath, "--all"], path); + result = { + preserved: "bundled", + detail: `bundled at ${bundlePath}`, + }; + } + } + await rm(resolve(path, ".linear-attachments"), { + recursive: true, + force: true, + }); + await rm(resolve(path, ".codex-dispatches"), { + recursive: true, + force: true, + }); + await this.git(["worktree", "remove", "--force", path], this.repo); + } else if ( + opts.alwaysPreserve && + (await this.gitOk( + ["show-ref", "--verify", `refs/heads/${branch}`], + this.repo, + )) + ) { + try { + await this.git(["push", "origin", `${branch}:${branch}`], this.repo); + result = { + preserved: "pushed", + detail: `pushed to ${branch}`, + }; + } catch { + await mkdir(bundlesDir, { recursive: true }); + const bundlePath = resolve( + bundlesDir, + `${identifier}-${Date.now()}.bundle`, + ); + await this.git(["bundle", "create", bundlePath, "--all"], this.repo); + result = { + preserved: "bundled", + detail: `bundled at ${bundlePath}`, + }; + } + } + try { + await this.git(["branch", "-D", branch], this.repo); + } catch (error) { + if ( + await this.gitOk( + ["show-ref", "--verify", `refs/heads/${branch}`], + this.repo, + ) + ) + throw error; + } + return result; + } finally { + release(); + } + } + private async ensureWorktreeLocked(rawIdentifier: string): Promise { const identifier = rawIdentifier.replace(/[^A-Za-z0-9-]/g, "-") || "issue"; const path = resolve(this.root, identifier); diff --git a/daemon/test/cleanup.test.ts b/daemon/test/cleanup.test.ts index c148819..234053b 100644 --- a/daemon/test/cleanup.test.ts +++ b/daemon/test/cleanup.test.ts @@ -8,7 +8,7 @@ import { } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { CleanupWorker } from "../src/cleanup.js"; import { inFlightDispatches } from "../src/dispatches.js"; import { EventLog } from "../src/eventlog.js"; @@ -128,12 +128,13 @@ describe("CleanupWorker", () => { const plannerTurn = s.log.claimNextTurn(3)!; s.log.finishTurn(plannerTurn.id, "response", "done", 3); s.log.markTurnActivityPosted(plannerTurn.id, 3); + const onIssueFinalized = vi.fn(); const worker = new CleanupWorker( s.log, new Poster() as unknown as LinearGateway, s.root, s.repo, - { pollMs: 10, reconcileMs: 20 }, + { pollMs: 10, reconcileMs: 20, onIssueFinalized }, ); worker.start(); await waitFor(() => s.log.cleanupStates()[0]?.status === "done"); @@ -144,9 +145,10 @@ describe("CleanupWorker", () => { expect(() => git(["show-ref", "--verify", "refs/heads/agents/ENG-42"], s.repo), ).toThrow(); + expect(onIssueFinalized).toHaveBeenCalledOnce(); s.log.close(); }); - it("retains a clean present worktree when no pull request URL was recorded", async () => { + it("preserves and removes a clean present worktree when no pull request URL was recorded", async () => { const s = await setup(); complete(s.log); const poster = new Poster(); @@ -162,13 +164,12 @@ describe("CleanupWorker", () => { () => s.log.cleanupNotificationStates()[0]?.status === "posted", ); await worker.stop(); - expect(s.log.cleanupStates()[0]?.status).toBe("retained"); - expect(poster.posts[0]).toContain("no pull request was recorded"); - expect(poster.posts[0]).toContain(s.tree.path); - expect(existsSync(s.tree.path)).toBe(true); + expect(s.log.cleanupStates()[0]?.status).toBe("done"); + expect(poster.posts[0]).toContain("pushed to agents/ENG-42"); + expect(existsSync(s.tree.path)).toBe(false); s.log.close(); }); - it("AC6 retains a dirty worktree and durably posts its path", async () => { + it("preserves and removes a dirty worktree and durably posts its destination", async () => { const s = await setup(); writeFileSync(join(s.tree.path, "dirty.txt"), "x"); complete(s.log); @@ -185,8 +186,33 @@ describe("CleanupWorker", () => { () => s.log.cleanupNotificationStates()[0]?.status === "posted", ); await worker.stop(); + expect(s.log.cleanupStates()[0]?.status).toBe("done"); + expect(poster.posts[0]).toContain("pushed to agents/ENG-42"); + expect(existsSync(s.tree.path)).toBe(false); + s.log.close(); + }); + it("retains only when preservation exhausts its retry window", async () => { + const s = await setup(); + writeFileSync(join(s.tree.path, "dirty.txt"), "x"); + git(["remote", "set-url", "origin", join(s.dir, "missing.git")], s.repo); + const blockedBundles = join(s.dir, "not-a-directory"); + writeFileSync(blockedBundles, "x"); + complete(s.log); + const poster = new Poster(); + const worker = new CleanupWorker( + s.log, + poster as unknown as LinearGateway, + s.root, + s.repo, + { + now: () => 10, + retryWindowMs: 1, + bundlesDir: blockedBundles, + }, + ); + await worker.trigger(); expect(s.log.cleanupStates()[0]?.status).toBe("retained"); - expect(poster.posts[0]).toContain(s.tree.path); + expect(poster.posts[0]).toContain("preservation failed"); expect(existsSync(s.tree.path)).toBe(true); s.log.close(); }); diff --git a/daemon/test/config.test.ts b/daemon/test/config.test.ts index 650b3f6..6bc0936 100644 --- a/daemon/test/config.test.ts +++ b/daemon/test/config.test.ts @@ -43,6 +43,10 @@ describe("loadConfig", () => { providerStateStaleMs: 300_000, providerInitialProbeTimeoutMs: 5_000 }); expect(config).toMatchObject({ bashDefaultTimeoutMs: 900_000, bashMaxTimeoutMs: 900_000 }); expect(config).toMatchObject({doPermissionMode:"bypassPermissions",doMaxTurns:300}); + expect(config.worktreeUnlinkedGraceMs).toBe(14 * 24 * 60 * 60 * 1000); + expect(config.worktreeBundlesDir).toBe( + "/var/lib/linear-agent-daemon/worktree-bundles", + ); }); it("loads independent harness preferences and names invalid settings", () => { const config = loadConfig({ ...base, PLANNER_HARNESS: "claudex", IMPLEMENTER_HARNESS: "claude" }); @@ -118,6 +122,23 @@ describe("loadConfig", () => { claudeArgv: ["node", "fixture.mjs"], claudePermissionMode: "bypassPermissions", claudeMaxTurns: 100, sessionConcurrency: 2, keepaliveMs: 900_000, attachmentsEnabled: true, attachmentHosts: ["uploads.linear.app"] }); }); + it("loads and validates worktree reconciliation settings", () => { + expect( + loadConfig({ + ...base, + DB_PATH: "/state/events.db", + WORKTREE_UNLINKED_GRACE_DAYS: "3", + WORKTREE_BUNDLES_DIR: " /archive/bundles ", + }), + ).toMatchObject({ + worktreeUnlinkedGraceMs: 3 * 24 * 60 * 60 * 1000, + worktreeBundlesDir: "/archive/bundles", + }); + for (const value of ["0", "-1", "1.5", "nope"]) + expect(() => + loadConfig({ ...base, WORKTREE_UNLINKED_GRACE_DAYS: value }), + ).toThrow("WORKTREE_UNLINKED_GRACE_DAYS"); + }); it("names missing variables", () => { expect(() => loadConfig({ ...base, PLANNER_WEBHOOK_SECRET: "" })).toThrow("PLANNER_WEBHOOK_SECRET"); }); diff --git a/daemon/test/linear.test.ts b/daemon/test/linear.test.ts index 1fab29c..dcc7214 100644 --- a/daemon/test/linear.test.ts +++ b/daemon/test/linear.test.ts @@ -394,4 +394,57 @@ describe("LinearGateway", () => { expect(api.requests.filter(request => request.url === "/graphql")).toHaveLength(0); await api.close(); db.close(); }); + + it("returns found, not_found, and error issue-state outcomes without throwing", async () => { + const api = await stub((request) => { + const id = (request.body.variables as { id: string }).id; + if (id === "ENG-1") + return { + body: { + data: { + issue: { + id: "issue-1", + identifier: "ENG-1", + state: { type: "completed" }, + }, + }, + }, + }; + if (id === "ENG-404") return { body: { data: { issue: null } } }; + return { status: 503, body: { error: "unavailable" } }; + }); + const db = log(); + const gateway = new LinearGateway( + db, + { + planner: { + name: "planner", + webhookSecret: "p", + staticToken: "token", + }, + implementer: { + name: "implementer", + webhookSecret: "i", + staticToken: "i", + }, + }, + api.graphqlUrl, + api.tokenUrl, + ); + await expect( + gateway.issueState("implementer", "ENG-1", Date.now() + 1000), + ).resolves.toEqual({ + kind: "found", + issueId: "issue-1", + stateType: "completed", + }); + await expect( + gateway.issueState("implementer", "ENG-404", Date.now() + 1000), + ).resolves.toEqual({ kind: "not_found" }); + await expect( + gateway.issueState("implementer", "ENG-500", Date.now() + 1000), + ).resolves.toMatchObject({ kind: "error" }); + await api.close(); + db.close(); + }); }); diff --git a/daemon/test/server.test.ts b/daemon/test/server.test.ts index cfe522d..34fcb60 100644 --- a/daemon/test/server.test.ts +++ b/daemon/test/server.test.ts @@ -31,10 +31,10 @@ function setup() { implementer: { name: "implementer", webhookSecret: "implementer-secret", staticToken: "i" }, }, }; - const onInserted = vi.fn(); const onStop = vi.fn(); + const onInserted = vi.fn(); const onIssueCompleted = vi.fn(); const onStop = vi.fn(); const logger = { log: vi.fn(), error: vi.fn() }; - const server = new WebhookServer({ config, log, onInserted, onStop, logger }); - return { config, log, server, onInserted, onStop, logger, managementKey }; + const server = new WebhookServer({ config, log, onInserted, onIssueCompleted, onStop, logger }); + return { config, log, server, onInserted, onIssueCompleted, onStop, logger, managementKey }; } function signed(body: string, secret = "planner-secret", delivery = "delivery-1") { @@ -63,9 +63,27 @@ async function waitForHealth(port: number, child: ChildProcess): Promise { describe("webhook HTTP integration", () => { it("persists a signed Issue webhook without acking or creating a turn",async()=>{ - const {log,server}=setup();const address=await server.listen();const body=JSON.stringify({webhookTimestamp:Date.now(),type:"Issue",action:"update",data:{id:"issue",identifier:"ENG-42",state:{type:"completed"}}}); + const {log,server,onIssueCompleted}=setup();const address=await server.listen();const body=JSON.stringify({webhookTimestamp:Date.now(),type:"Issue",action:"update",data:{id:"issue",identifier:"ENG-42",state:{type:"completed"}}}); const response=await fetch(`http://127.0.0.1:${address.port}/webhook/implementer`,{method:"POST",headers:signed(body,"implementer-secret","issue-delivery"),body}); - expect(response.status).toBe(200);expect(log.count()).toBe(1);expect(log.ackCount()).toBe(0);expect(log.turnStates()).toHaveLength(0);await server.close();log.close(); + expect(response.status).toBe(200);expect(log.count()).toBe(1);expect(log.ackCount()).toBe(0);expect(log.turnStates()).toHaveLength(0);expect(onIssueCompleted).toHaveBeenCalledOnce();await server.close();log.close(); + }); + it("does not fire the Issue-completed callback for non-completed events", async () => { + const { log, server, onIssueCompleted } = setup(); + const address = await server.listen(); + const body = JSON.stringify({ + webhookTimestamp: Date.now(), + type: "Issue", + action: "update", + data: { id: "issue", identifier: "ENG-42", state: { type: "started" } }, + }); + await fetch(`http://127.0.0.1:${address.port}/webhook/implementer`, { + method: "POST", + headers: signed(body, "implementer-secret", "started-delivery"), + body, + }); + expect(onIssueCompleted).not.toHaveBeenCalled(); + await server.close(); + log.close(); }); it("AC1: responds under 5s and persists a signed fresh event", async () => { const { log, server } = setup(); const address = await server.listen(); diff --git a/daemon/test/sessions.test.ts b/daemon/test/sessions.test.ts index ae16ea7..7ff7dd2 100644 --- a/daemon/test/sessions.test.ts +++ b/daemon/test/sessions.test.ts @@ -3186,12 +3186,13 @@ describe("SessionWorker", () => { server.listen(0, "127.0.0.1", resolve), ); const ntfyUrl = `http://127.0.0.1:${(server.address() as { port: number }).port}`; + const onTurnComplete = vi.fn(); append(log, "d1", "session", "created"); const worker = new SessionWorker( log, poster as unknown as LinearGateway, { ...config, ntfyUrl }, - { pollMs: 10, reconcileMs: 20 }, + { pollMs: 10, reconcileMs: 20, onTurnComplete }, ); await worker.start(); await waitFor( @@ -3204,6 +3205,7 @@ describe("SessionWorker", () => { expect(received[0].title).toContain("ENG-42"); expect(received[0].priority).toBe("default"); expect(received[0].body).toBe("planner answer"); + expect(onTurnComplete).toHaveBeenCalledOnce(); }); it("AC1/AC2/AC3-contract: creates worktree, posts response, then resumes in the same cwd", async () => { const { dir, log, config } = setup(); diff --git a/daemon/test/worktree-reconciler.test.ts b/daemon/test/worktree-reconciler.test.ts new file mode 100644 index 0000000..533e3c4 --- /dev/null +++ b/daemon/test/worktree-reconciler.test.ts @@ -0,0 +1,350 @@ +import { execFileSync } from "node:child_process"; +import { + existsSync, + mkdirSync, + mkdtempSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { CleanupWorker } from "../src/cleanup.js"; +import { EventLog } from "../src/eventlog.js"; +import type { IssueStateResult, LinearGateway } from "../src/linear.js"; +import { WorktreeReconciler } from "../src/worktree-reconciler.js"; +import { WorktreeManager } from "../src/worktrees.js"; + +const dirs: string[] = []; +afterEach(() => { + vi.useRealTimers(); + for (const dir of dirs.splice(0)) + rmSync(dir, { recursive: true, force: true }); +}); + +function git(args: string[], cwd?: string): string { + return execFileSync("git", args, { cwd, encoding: "utf8" }); +} + +function harness(now = 1_000) { + const dir = mkdtempSync(join(tmpdir(), "worktree-reconciler-")); + dirs.push(dir); + const seed = join(dir, "seed"); + const origin = join(dir, "origin.git"); + const repo = join(dir, "repo"); + const root = join(dir, "trees"); + const bundles = join(dir, "bundles"); + mkdirSync(seed); + git(["init", "-b", "main"], seed); + git(["config", "user.email", "test@example.com"], seed); + git(["config", "user.name", "Test"], seed); + git(["commit", "--allow-empty", "-m", "initial"], seed); + git(["clone", "--bare", seed, origin]); + git(["clone", origin, repo]); + const log = new EventLog(join(dir, "events.db")); + const worktrees = new WorktreeManager(root, repo); + const states = new Map(); + const gateway = { + issueState: vi.fn(async (_app: string, identifier: string) => + states.get(identifier) ?? { kind: "not_found" }, + ), + } as unknown as LinearGateway; + const enqueued = vi.fn(); + const logger = { log: vi.fn(), error: vi.fn() }; + const clock = { value: now }; + const reconciler = new WorktreeReconciler( + log, + gateway, + worktrees, + { + worktreesRoot: root, + worktreeUnlinkedGraceMs: 100, + worktreeBundlesDir: bundles, + }, + { now: () => clock.value, onEnqueued: enqueued, logger }, + ); + return { + dir, + origin, + repo, + root, + bundles, + log, + worktrees, + states, + gateway, + enqueued, + logger, + clock, + reconciler, + }; +} + +async function addSession( + h: ReturnType, + identifier: string, + issueId: string, + withPath = true, +) { + const tree = await h.worktrees.ensureWorktree(identifier); + h.log.append({ + deliveryId: `session-${identifier}`, + app: "implementer", + action: "created", + agentSessionId: `session-${identifier}`, + issueId, + issueIdentifier: identifier, + receivedAt: 1, + rawBody: Buffer.from("{}"), + }); + if (withPath) + h.log.updateSessionWorktree( + `session-${identifier}`, + tree.path, + tree.branch, + 2, + ); + const turn = h.log.claimNextTurn(2)!; + h.log.finishTurn(turn.id, "response", "done", 2); + h.log.markTurnActivityPosted(turn.id, 2); + return tree; +} + +describe("WorktreeReconciler", () => { + it("closes a resolved session-backed worktree end-to-end without an Issue webhook", async () => { + const h = harness(); + const tree = await addSession(h, "ENG-1", "issue-1"); + h.log.stageExternalUrl( + "session-ENG-1", + "implementer", + "Pull Request", + "https://github.com/dcouple/example/pull/1", + 3, + ); + h.states.set("ENG-1", { + kind: "found", + issueId: "issue-1", + stateType: "completed", + }); + expect(h.log.count()).toBe(1); + await h.reconciler.trigger(); + expect(h.log.cleanupStates()).toEqual([ + { id: 1, status: "pending", issueId: "issue-1" }, + ]); + expect(h.enqueued).toHaveBeenCalledOnce(); + const cleanup = new CleanupWorker( + h.log, + h.gateway, + h.root, + h.repo, + { + worktrees: h.worktrees, + bundlesDir: h.bundles, + now: () => h.clock.value, + }, + ); + await cleanup.trigger(); + expect(h.log.count()).toBe(1); + expect(h.log.cleanupStates()).toEqual([ + { id: 1, status: "done", issueId: "issue-1" }, + ]); + expect(existsSync(tree.path)).toBe(false); + expect(() => + git(["show-ref", "--verify", "refs/heads/agents/ENG-1"], h.repo), + ).toThrow(); + await cleanup.stop(); + h.log.close(); + }); + + it.each(["completed", "canceled"])( + "directly preserves and removes a no-session %s worktree", + async (stateType) => { + const h = harness(); + const tree = await h.worktrees.ensureWorktree( + stateType === "completed" ? "ENG-2" : "ENG-3", + ); + writeFileSync(join(tree.path, "dirty.txt"), stateType); + const identifier = tree.branch.replace("agents/", ""); + h.states.set(identifier, { + kind: "found", + issueId: `issue-${identifier}`, + stateType, + }); + await h.reconciler.trigger(); + expect(existsSync(tree.path)).toBe(false); + expect( + git( + ["show", `refs/heads/${tree.branch}:dirty.txt`], + h.origin, + ).trim(), + ).toBe(stateType); + expect(h.logger.log).toHaveBeenCalledWith( + expect.stringContaining("worktree_reconciled"), + ); + h.log.close(); + }, + ); + + it("starts and expires only the persisted unlinked grace clock", async () => { + const h = harness(); + const tree = await h.worktrees.ensureWorktree("ENG-4"); + await h.reconciler.trigger(); + expect(h.log.unlinkedObservation("ENG-4")?.firstObservedAt).toBe(1_000); + expect(existsSync(tree.path)).toBe(true); + h.clock.value = 1_099; + await h.reconciler.trigger(); + expect(existsSync(tree.path)).toBe(true); + h.clock.value = 1_100; + await h.reconciler.trigger(); + expect(existsSync(tree.path)).toBe(false); + expect(h.log.unlinkedObservation("ENG-4")).toBeUndefined(); + h.log.close(); + }); + + it("skips lookup errors without writing the grace clock", async () => { + const h = harness(); + const tree = await h.worktrees.ensureWorktree("ENG-5"); + h.states.set("ENG-5", { kind: "error", error: "timeout" }); + await h.reconciler.trigger(); + expect(existsSync(tree.path)).toBe(true); + expect(h.log.unlinkedObservation("ENG-5")).toBeUndefined(); + h.log.close(); + }); + + it("leaves unresolved worktrees and clears stale unlinked observations", async () => { + const h = harness(); + const tree = await h.worktrees.ensureWorktree("ENG-6"); + h.log.observeUnlinkedWorktree("ENG-6", tree.path, 1); + h.states.set("ENG-6", { + kind: "found", + issueId: "issue-6", + stateType: "started", + }); + await h.reconciler.trigger(); + expect(existsSync(tree.path)).toBe(true); + expect(h.log.unlinkedObservation("ENG-6")).toBeUndefined(); + expect(h.log.cleanupStates()).toEqual([]); + h.log.close(); + }); + + it("never touches a foreign directory", async () => { + const h = harness(); + mkdirSync(h.root); + const foreign = join(h.root, "ENG-7"); + git(["clone", h.origin, foreign]); + await h.reconciler.trigger(); + expect(existsSync(foreign)).toBe(true); + expect(h.gateway.issueState).not.toHaveBeenCalled(); + h.log.close(); + }); + + it("repromotes a retained cleanup and signals the cleanup worker", async () => { + const h = harness(); + await addSession(h, "ENG-8", "issue-8"); + h.log.enqueueCleanupFromReconcile("issue-8", "ENG-8", 3); + const job = h.log.claimNextCleanup(3)!; + h.log.retainCleanup(job.id, "old preservation failure", 3); + h.states.set("ENG-8", { + kind: "found", + issueId: "issue-8", + stateType: "completed", + }); + await h.reconciler.trigger(); + expect(h.log.cleanupStates()[0]?.status).toBe("pending"); + expect(h.enqueued).toHaveBeenCalledOnce(); + h.log.close(); + }); + + it("start performs one immediate coalesced sweep with no standing timer", async () => { + vi.useFakeTimers(); + const h = harness(); + await h.worktrees.ensureWorktree("ENG-9"); + h.reconciler.start(); + const first = h.reconciler.trigger(); + const second = h.reconciler.trigger(); + expect(first).toBe(second); + await first; + expect(vi.getTimerCount()).toBe(0); + await h.reconciler.stop(); + h.log.close(); + }); + + it("stop awaits an in-flight lookup and shared mutations stay coherent", async () => { + const h = harness(); + const tree = await h.worktrees.ensureWorktree("ENG-10"); + let release!: (result: IssueStateResult) => void; + const lookup = new Promise((resolve) => { + release = resolve; + }); + vi.mocked(h.gateway.issueState).mockReturnValueOnce(lookup); + h.reconciler.start(); + while (!vi.mocked(h.gateway.issueState).mock.calls.length) + await new Promise((resolve) => setTimeout(resolve, 0)); + let stopped = false; + const stop = h.reconciler.stop().then(() => { + stopped = true; + }); + await Promise.resolve(); + expect(stopped).toBe(false); + release({ + kind: "found", + issueId: "issue-10", + stateType: "completed", + }); + await Promise.all([ + stop, + h.worktrees.ensureWorktree("ENG-10").catch(() => undefined), + ]); + if (existsSync(tree.path)) { + expect( + git(["branch", "--show-current"], tree.path).trim(), + ).toBe("agents/ENG-10"); + } else { + expect(() => + git( + ["show-ref", "--verify", "refs/heads/agents/ENG-10"], + h.repo, + ), + ).toThrow(); + } + h.log.close(); + }); + + it("is sequentially idempotent after a completed-state preservation and removal", async () => { + const h = harness(); + const tree = await h.worktrees.ensureWorktree("ENG-11"); + writeFileSync(join(tree.path, "dirty.txt"), "preserve once"); + h.states.set("ENG-11", { + kind: "found", + issueId: "issue-11", + stateType: "completed", + }); + const preserve = vi.spyOn(h.worktrees, "preserveAndRemove"); + + await h.reconciler.trigger(); + expect(preserve).toHaveBeenCalledOnce(); + expect(existsSync(tree.path)).toBe(false); + expect( + git( + ["show", "refs/heads/agents/ENG-11:dirty.txt"], + h.origin, + ).trim(), + ).toBe("preserve once"); + const jobsAfterFirst = h.log.cleanupStates(); + const observationAfterFirst = h.log.unlinkedObservation("ENG-11"); + const logsAfterFirst = h.logger.log.mock.calls.length; + const errorsAfterFirst = h.logger.error.mock.calls.length; + + await h.reconciler.trigger(); + + expect(preserve).toHaveBeenCalledOnce(); + expect(h.log.cleanupStates()).toEqual(jobsAfterFirst); + expect(h.log.unlinkedObservation("ENG-11")).toEqual( + observationAfterFirst, + ); + expect(h.logger.log.mock.calls.length).toBe(logsAfterFirst); + expect(h.logger.error.mock.calls.length).toBe(errorsAfterFirst); + expect(existsSync(h.bundles)).toBe(false); + h.log.close(); + }); +}); diff --git a/daemon/test/worktrees.test.ts b/daemon/test/worktrees.test.ts index d6f9728..b5bc5eb 100644 --- a/daemon/test/worktrees.test.ts +++ b/daemon/test/worktrees.test.ts @@ -50,4 +50,38 @@ describe("WorktreeManager", () => { rmSync(join(tree.path,"dirty.txt")); git(["worktree","remove",tree.path],setup.repo); await manager.remove("ENG-45"); expect(()=>git(["show-ref","--verify","refs/heads/agents/ENG-45"],setup.repo)).toThrow(); }); + it("commits and pushes dirty state before force removal", async () => { + const setup = repository(); + const manager = new WorktreeManager(setup.root, setup.repo); + const tree = await manager.ensureWorktree("ENG-46"); + writeFileSync(join(tree.path, "dirty.txt"), "preserved"); + const result = await manager.preserveAndRemove( + "ENG-46", + join(setup.dir, "bundles"), + { alwaysPreserve: false }, + ); + expect(result).toEqual({ + preserved: "pushed", + detail: "pushed to agents/ENG-46", + }); + expect(existsSync(tree.path)).toBe(false); + expect( + git(["show", "refs/heads/agents/ENG-46:dirty.txt"], setup.origin), + ).toBe("preserved"); + }); + it("falls back to a newly created bundle directory when push fails", async () => { + const setup = repository(); + const manager = new WorktreeManager(setup.root, setup.repo); + const tree = await manager.ensureWorktree("ENG-47"); + writeFileSync(join(tree.path, "dirty.txt"), "preserved"); + git(["remote", "set-url", "origin", join(setup.dir, "missing.git")], setup.repo); + const result = await manager.preserveAndRemove( + "ENG-47", + join(setup.dir, "new", "bundles"), + { alwaysPreserve: true }, + ); + expect(result.preserved).toBe("bundled"); + expect(existsSync(result.detail.replace("bundled at ", ""))).toBe(true); + expect(existsSync(tree.path)).toBe(false); + }); }); diff --git a/docs/linear-agent-daemon-setup.md b/docs/linear-agent-daemon-setup.md index 102ae86..8b15cc6 100644 --- a/docs/linear-agent-daemon-setup.md +++ b/docs/linear-agent-daemon-setup.md @@ -84,6 +84,9 @@ For **both** apps: For **bloom-implementer only**, additionally check **Issues** under **Data change events** — that webhook is how the daemon learns an issue was completed so it can clean up the worktree and branch. +An event-driven reconciliation sweep also runs at startup and after completed +turns to backstop missed delivery; the Issues subscription remains required +for immediate cleanup. After saving each app, record three values: the **client ID**, the **client secret**, and the **webhook signing secret** (`lin_wh_…`). @@ -189,6 +192,8 @@ LINEAR_API_KEY=... SESSIONS_ENABLED=1 TARGET_REPO_PATH=/var/lib/linear-agent-daemon/repos/ WORKTREES_ROOT=/var/lib/linear-agent-daemon/worktrees +WORKTREE_UNLINKED_GRACE_DAYS=14 +WORKTREE_BUNDLES_DIR=/var/lib/linear-agent-daemon/worktree-bundles PLANNER_HARNESS=claude IMPLEMENTER_HARNESS=claude CLAUDE_BIN=/var/lib/linear-agent-daemon/.local/bin/claude @@ -250,7 +255,8 @@ have the exact SQL/GraphQL evidence queries. unattended on branch `agents/`, opens a PR, and the PR appears on the session. Reply to an implementer question — the session resumes. Move the issue to a `completed` state; the worktree and - branch are cleaned up (dirty worktrees are retained and reported). + branch are cleaned up. Dirty or PR-less state is first pushed to the + issue's agent branch or saved as a restorable git bundle and reported. 4. If `NTFY_URL` is set, each terminal response/error arrives as a phone notification. From adcf60d4805eec7da71a9c91c8547b6edfa06918 Mon Sep 17 00:00:00 2001 From: Tyler Brown Date: Thu, 30 Jul 2026 12:05:00 -0700 Subject: [PATCH 2/3] fix(daemon): reset stale grace clocks and guard reconcile removal against live sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review fixes for PR #129: a recreated worktree now always gets a fresh unlinked grace period (durable directory identity resets the clock), and preserveAndRemove revalidates worktree identity/generation inside the WorktreeManager mutation lock, skipping removal when ensureWorktree interleaved — a session's live worktree can no longer be deleted from under it. Co-Authored-By: Claude Fable 5 --- daemon/src/eventlog.ts | 38 +++++++++-- daemon/src/worktree-reconciler.ts | 27 ++++++-- daemon/src/worktrees.ts | 90 +++++++++++++++++++++++-- daemon/test/worktree-reconciler.test.ts | 51 +++++++++----- 4 files changed, 174 insertions(+), 32 deletions(-) diff --git a/daemon/src/eventlog.ts b/daemon/src/eventlog.ts index 033416c..544c45d 100644 --- a/daemon/src/eventlog.ts +++ b/daemon/src/eventlog.ts @@ -253,6 +253,7 @@ export interface CleanupNotificationRow { export interface WorktreeUnlinkedObservation { identifier: string; path: string; + identity: string | null; firstObservedAt: number; lastSeenAt: number; } @@ -439,6 +440,7 @@ export class EventLog { CREATE TABLE IF NOT EXISTS worktree_unlinked_observations ( identifier TEXT PRIMARY KEY, path TEXT NOT NULL, + identity TEXT NOT NULL, first_observed_at INTEGER NOT NULL, last_seen_at INTEGER NOT NULL ); @@ -566,6 +568,7 @@ export class EventLog { this.migrateTurnColumns(); this.migrateAckColumns(); this.migrateTurnActivityColumns(); + this.migrateWorktreeObservationColumns(); this.recoverAmbiguousOutbox(); } @@ -612,6 +615,22 @@ export class EventLog { })(); } + private migrateWorktreeObservationColumns(): void { + const columns = new Set( + ( + this.db + .prepare("PRAGMA table_info(worktree_unlinked_observations)") + .all() as Array<{ name: string }> + ).map((column) => column.name), + ); + if (!columns.has("identity")) + this.db + .prepare( + "ALTER TABLE worktree_unlinked_observations ADD COLUMN identity TEXT", + ) + .run(); + } + private migrateEventColumns(): void { const columns = new Set( ( @@ -2062,15 +2081,24 @@ export class EventLog { observeUnlinkedWorktree( identifier: string, path: string, + identity: string, now = Date.now(), ): WorktreeUnlinkedObservation { this.db .prepare( `INSERT INTO worktree_unlinked_observations - (identifier,path,first_observed_at,last_seen_at) VALUES (?,?,?,?) - ON CONFLICT(identifier) DO UPDATE SET path=excluded.path,last_seen_at=excluded.last_seen_at`, - ) - .run(identifier, path, now, now); + (identifier,path,identity,first_observed_at,last_seen_at) VALUES (?,?,?,?,?) + ON CONFLICT(identifier) DO UPDATE SET + path=excluded.path, + identity=excluded.identity, + first_observed_at=CASE + WHEN worktree_unlinked_observations.identity IS excluded.identity + THEN worktree_unlinked_observations.first_observed_at + ELSE excluded.first_observed_at + END, + last_seen_at=excluded.last_seen_at`, + ) + .run(identifier, path, identity, now, now); return this.unlinkedObservation(identifier)!; } clearUnlinkedObservation(identifier: string): void { @@ -2085,7 +2113,7 @@ export class EventLog { ): WorktreeUnlinkedObservation | undefined { return this.db .prepare( - `SELECT identifier,path,first_observed_at firstObservedAt, + `SELECT identifier,path,identity,first_observed_at firstObservedAt, last_seen_at lastSeenAt FROM worktree_unlinked_observations WHERE identifier=?`, ) .get(identifier) as WorktreeUnlinkedObservation | undefined; diff --git a/daemon/src/worktree-reconciler.ts b/daemon/src/worktree-reconciler.ts index e8d6363..d1efd74 100644 --- a/daemon/src/worktree-reconciler.ts +++ b/daemon/src/worktree-reconciler.ts @@ -2,7 +2,10 @@ import { readdir } from "node:fs/promises"; import { resolve } from "node:path"; import type { EventLog } from "./eventlog.js"; import type { LinearGateway } from "./linear.js"; -import type { WorktreeManager } from "./worktrees.js"; +import { + WorktreeChangedError, + type WorktreeManager, +} from "./worktrees.js"; interface Logger { log(...args: unknown[]): void; @@ -77,7 +80,8 @@ export class WorktreeReconciler { private async reconcileCandidate(identifier: string): Promise { const path = resolve(this.config.worktreesRoot, identifier); - if (!(await this.worktrees.isOwned(path))) { + const snapshot = await this.worktrees.snapshot(identifier); + if (!snapshot) { this.logger.log( JSON.stringify({ event: "worktree_reconcile_skipped", @@ -109,6 +113,7 @@ export class WorktreeReconciler { const observation = this.log.observeUnlinkedWorktree( identifier, path, + snapshot.identity, this.now(), ); if ( @@ -120,11 +125,12 @@ export class WorktreeReconciler { const result = await this.worktrees.preserveAndRemove( identifier, this.config.worktreeBundlesDir, - { alwaysPreserve: true }, + { alwaysPreserve: true, expectedSnapshot: snapshot }, ); this.log.clearUnlinkedObservation(identifier); this.notify(identifier, result.detail, "unlinked_grace_expired"); } catch (error) { + if (this.logChangedCandidate(identifier, error)) return; this.logger.error( JSON.stringify({ event: "worktree_reconcile_preservation_failed", @@ -160,10 +166,11 @@ export class WorktreeReconciler { const result = await this.worktrees.preserveAndRemove( identifier, this.config.worktreeBundlesDir, - { alwaysPreserve: true }, + { alwaysPreserve: true, expectedSnapshot: snapshot }, ); this.notify(identifier, result.detail, "resolved_without_session"); } catch (error) { + if (this.logChangedCandidate(identifier, error)) return; this.logger.error( JSON.stringify({ event: "worktree_reconcile_preservation_failed", @@ -174,6 +181,18 @@ export class WorktreeReconciler { } } + private logChangedCandidate(identifier: string, error: unknown): boolean { + if (!(error instanceof WorktreeChangedError)) return false; + this.logger.log( + JSON.stringify({ + event: "worktree_reconcile_skipped", + identifier, + reason: "candidate_changed", + }), + ); + return true; + } + private notify(identifier: string, detail: string, reason: string): void { const body = `Worktree ${identifier} reconciled; state ${detail}.`; this.logger.log( diff --git a/daemon/src/worktrees.ts b/daemon/src/worktrees.ts index 4d97134..970aa87 100644 --- a/daemon/src/worktrees.ts +++ b/daemon/src/worktrees.ts @@ -1,4 +1,5 @@ import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; import { mkdir, readFile, realpath, rm, stat, writeFile } from "node:fs/promises"; import { isAbsolute, join, resolve } from "node:path"; import { promisify } from "node:util"; @@ -10,9 +11,19 @@ export interface PreservationResult { preserved: "pushed" | "bundled" | "none"; detail: string; } +export interface WorktreeSnapshot { + identity: string; + generation: number; +} +export class WorktreeChangedError extends Error { + constructor(identifier: string) { + super(`Worktree changed after reconciliation lookup: ${identifier}`); + } +} export class WorktreeManager { private mutation: Promise = Promise.resolve(); + private readonly generations = new Map(); constructor(private readonly root: string, private readonly repo: string) {} async ensureWorktree(rawIdentifier: string): Promise { @@ -20,7 +31,14 @@ export class WorktreeManager { let release!: () => void; this.mutation = new Promise(resolve => { release = resolve; }); await previous; - try { return await this.ensureWorktreeLocked(rawIdentifier); } + try { + const identifier = this.identifier(rawIdentifier); + this.generations.set( + identifier, + (this.generations.get(identifier) ?? 0) + 1, + ); + return await this.ensureWorktreeLocked(rawIdentifier); + } finally { release(); } } @@ -40,6 +58,30 @@ export class WorktreeManager { } } + async snapshot(rawIdentifier: string): Promise { + const previous = this.mutation; + let release!: () => void; + this.mutation = new Promise((resolve) => { + release = resolve; + }); + await previous; + try { + const identifier = this.identifier(rawIdentifier); + const path = resolve(this.root, identifier); + if (!(await this.exists(path))) return undefined; + try { + return { + identity: await this.identityLocked(path), + generation: this.generations.get(identifier) ?? 0, + }; + } catch { + return undefined; + } + } finally { + release(); + } + } + async remove(rawIdentifier: string): Promise { const previous = this.mutation; let release!: () => void; @@ -63,7 +105,10 @@ export class WorktreeManager { async preserveAndRemove( rawIdentifier: string, bundlesDir: string, - opts: { alwaysPreserve: boolean }, + opts: { + alwaysPreserve: boolean; + expectedSnapshot?: WorktreeSnapshot; + }, ): Promise { const previous = this.mutation; let release!: () => void; @@ -72,14 +117,22 @@ export class WorktreeManager { }); await previous; try { - const identifier = - rawIdentifier.replace(/[^A-Za-z0-9-]/g, "-") || "issue"; + const identifier = this.identifier(rawIdentifier); const path = resolve(this.root, identifier); const branch = `agents/${identifier}`; let result: PreservationResult = { preserved: "none", detail: "no preservation required", }; + if (opts.expectedSnapshot) { + const generation = this.generations.get(identifier) ?? 0; + if ( + generation !== opts.expectedSnapshot.generation || + !(await this.exists(path)) || + (await this.identityLocked(path)) !== opts.expectedSnapshot.identity + ) + throw new WorktreeChangedError(identifier); + } if (await this.exists(path)) { await this.validate(path); const wasDirty = @@ -175,7 +228,7 @@ export class WorktreeManager { } private async ensureWorktreeLocked(rawIdentifier: string): Promise { - const identifier = rawIdentifier.replace(/[^A-Za-z0-9-]/g, "-") || "issue"; + const identifier = this.identifier(rawIdentifier); const path = resolve(this.root, identifier); const branch = `agents/${identifier}`; await mkdir(this.root, { recursive: true }); @@ -215,6 +268,30 @@ export class WorktreeManager { if (common !== expected) throw new Error(`Existing worktree belongs to a foreign repository: ${path}`); } + private async identityLocked(path: string): Promise { + await this.validate(path); + const gitDirRaw = (await this.git(["rev-parse", "--git-dir"], path)).trim(); + const gitDir = isAbsolute(gitDirRaw) + ? gitDirRaw + : resolve(path, gitDirRaw); + const markerPath = resolve(gitDir, "orchestra-worktree-id"); + let marker: string; + try { + marker = (await readFile(markerPath, "utf8")).trim(); + } catch { + marker = randomUUID(); + await writeFile(markerPath, `${marker}\n`, { flag: "wx" }).catch( + async (error: NodeJS.ErrnoException) => { + if (error.code !== "EEXIST") throw error; + marker = (await readFile(markerPath, "utf8")).trim(); + }, + ); + } + const branch = (await this.git(["branch", "--show-current"], path)).trim(); + const head = (await this.git(["rev-parse", "HEAD"], path)).trim(); + return `${marker}:${branch}:${head}`; + } + private async excludeTransientDirectories(path: string): Promise { const raw = (await this.git(["rev-parse", "--git-path", "info/exclude"], path)).trim(); const exclude = isAbsolute(raw) ? raw : resolve(path, raw); @@ -228,6 +305,9 @@ export class WorktreeManager { } private async exists(path: string): Promise { try { await stat(path); return true; } catch { return false; } } + private identifier(rawIdentifier: string): string { + return rawIdentifier.replace(/[^A-Za-z0-9-]/g, "-") || "issue"; + } private async git(args: string[], cwd: string): Promise { const { stdout } = await exec("git", args, { cwd }); return stdout; } diff --git a/daemon/test/worktree-reconciler.test.ts b/daemon/test/worktree-reconciler.test.ts index 533e3c4..2f39e11 100644 --- a/daemon/test/worktree-reconciler.test.ts +++ b/daemon/test/worktree-reconciler.test.ts @@ -201,6 +201,23 @@ describe("WorktreeReconciler", () => { h.log.close(); }); + it("gives a recreated unlinked worktree a fresh grace clock", async () => { + const h = harness(); + const original = await h.worktrees.ensureWorktree("ENG-12"); + await h.reconciler.trigger(); + expect(h.log.unlinkedObservation("ENG-12")?.firstObservedAt).toBe(1_000); + + await h.worktrees.remove("ENG-12"); + expect(existsSync(original.path)).toBe(false); + const recreated = await h.worktrees.ensureWorktree("ENG-12"); + h.clock.value = 10_000; + await h.reconciler.trigger(); + + expect(existsSync(recreated.path)).toBe(true); + expect(h.log.unlinkedObservation("ENG-12")?.firstObservedAt).toBe(10_000); + h.log.close(); + }); + it("skips lookup errors without writing the grace clock", async () => { const h = harness(); const tree = await h.worktrees.ensureWorktree("ENG-5"); @@ -214,7 +231,8 @@ describe("WorktreeReconciler", () => { it("leaves unresolved worktrees and clears stale unlinked observations", async () => { const h = harness(); const tree = await h.worktrees.ensureWorktree("ENG-6"); - h.log.observeUnlinkedWorktree("ENG-6", tree.path, 1); + const snapshot = await h.worktrees.snapshot("ENG-6"); + h.log.observeUnlinkedWorktree("ENG-6", tree.path, snapshot!.identity, 1); h.states.set("ENG-6", { kind: "found", issueId: "issue-6", @@ -269,7 +287,7 @@ describe("WorktreeReconciler", () => { h.log.close(); }); - it("stop awaits an in-flight lookup and shared mutations stay coherent", async () => { + it("keeps a worktree when ensureWorktree reuses it during the lookup", async () => { const h = harness(); const tree = await h.worktrees.ensureWorktree("ENG-10"); let release!: (result: IssueStateResult) => void; @@ -286,27 +304,24 @@ describe("WorktreeReconciler", () => { }); await Promise.resolve(); expect(stopped).toBe(false); + await h.worktrees.ensureWorktree("ENG-10"); release({ kind: "found", issueId: "issue-10", stateType: "completed", }); - await Promise.all([ - stop, - h.worktrees.ensureWorktree("ENG-10").catch(() => undefined), - ]); - if (existsSync(tree.path)) { - expect( - git(["branch", "--show-current"], tree.path).trim(), - ).toBe("agents/ENG-10"); - } else { - expect(() => - git( - ["show-ref", "--verify", "refs/heads/agents/ENG-10"], - h.repo, - ), - ).toThrow(); - } + await stop; + expect(existsSync(tree.path)).toBe(true); + expect(git(["branch", "--show-current"], tree.path).trim()).toBe( + "agents/ENG-10", + ); + expect( + git(["show-ref", "--verify", "refs/heads/agents/ENG-10"], h.repo).trim(), + ).toContain("refs/heads/agents/ENG-10"); + expect(h.logger.log).toHaveBeenCalledWith( + expect.stringContaining('"reason":"candidate_changed"'), + ); + expect(h.logger.error).not.toHaveBeenCalled(); h.log.close(); }); From 3dd4b020b83d2bf529661225d70992acce8b14d9 Mon Sep 17 00:00:00 2001 From: Tyler Brown Date: Thu, 30 Jul 2026 12:10:42 -0700 Subject: [PATCH 3/3] fix(daemon): revalidate worktree snapshot in every CleanupWorker removal Review pass 2 follow-up for PR #129: CleanupWorker now captures a fresh worktree identity/generation snapshot after finalization and passes it to every removal; a generation change from an interleaving ensureWorktree preserves the live worktree and re-pends the job instead of deleting it. Co-Authored-By: Claude Fable 5 --- daemon/src/cleanup.ts | 30 +++++++++++++++---- daemon/src/worktrees.ts | 58 ++++++++++++++++++++++++++++--------- daemon/test/cleanup.test.ts | 52 ++++++++++++++++++++++++++++++++- 3 files changed, 121 insertions(+), 19 deletions(-) diff --git a/daemon/src/cleanup.ts b/daemon/src/cleanup.ts index 87a448f..449ee5a 100644 --- a/daemon/src/cleanup.ts +++ b/daemon/src/cleanup.ts @@ -4,7 +4,7 @@ import type { CleanupNotificationRow, } from "./eventlog.js"; import type { LinearGateway } from "./linear.js"; -import { WorktreeManager } from "./worktrees.js"; +import { WorktreeChangedError, WorktreeManager } from "./worktrees.js"; import { buildInvocationSpan, buildSessionRoot, @@ -97,6 +97,14 @@ export class CleanupWorker { ); return; } + const expectedSnapshot = await this.worktrees.snapshot( + job.issueIdentifier, + { includeAbsent: true }, + ); + if (!expectedSnapshot) + throw new Error( + `Refusing cleanup without an owned worktree snapshot: ${job.issueIdentifier}`, + ); const session = this.log.sessionByIssueIdentifier(job.issueIdentifier); if ( !session?.worktreePath || @@ -106,7 +114,7 @@ export class CleanupWorker { const result = await this.worktrees.preserveAndRemove( job.issueIdentifier, this.options.bundlesDir ?? `${this.worktreesRoot}/../worktree-bundles`, - { alwaysPreserve: true }, + { alwaysPreserve: true, expectedSnapshot }, ); preservationAttempt = false; this.log.clearSessionWorktrees(job.issueIdentifier); @@ -126,7 +134,7 @@ export class CleanupWorker { job.issueIdentifier, this.options.bundlesDir ?? `${this.worktreesRoot}/../worktree-bundles`, - { alwaysPreserve: true }, + { alwaysPreserve: true, expectedSnapshot }, ); preservationAttempt = false; this.log.clearSessionWorktrees(job.issueIdentifier); @@ -139,7 +147,7 @@ export class CleanupWorker { this.options.onIssueFinalized?.(); return; } - await this.worktrees.remove(job.issueIdentifier); + await this.worktrees.remove(job.issueIdentifier, { expectedSnapshot }); this.log.clearSessionWorktrees(job.issueIdentifier); this.log.markCleanupDone(job.id); this.options.onIssueFinalized?.(); @@ -148,7 +156,7 @@ export class CleanupWorker { const result = await this.worktrees.preserveAndRemove( job.issueIdentifier, this.options.bundlesDir ?? `${this.worktreesRoot}/../worktree-bundles`, - { alwaysPreserve: true }, + { alwaysPreserve: true, expectedSnapshot }, ); preservationAttempt = false; this.log.clearSessionWorktrees(job.issueIdentifier); @@ -162,6 +170,18 @@ export class CleanupWorker { } } catch (error) { const message = error instanceof Error ? error.message : String(error); + if (error instanceof WorktreeChangedError) { + this.log.retryCleanup(job.id, message, this.now() + 1000); + this.logger.log( + JSON.stringify({ + event: "cleanup_revalidation_retry", + jobId: job.id, + issueIdentifier: job.issueIdentifier, + reason: "worktree_changed", + }), + ); + return; + } if ( this.now() < job.createdAt + (this.options.retryWindowMs ?? 30 * 60_000) diff --git a/daemon/src/worktrees.ts b/daemon/src/worktrees.ts index 970aa87..b35c7f7 100644 --- a/daemon/src/worktrees.ts +++ b/daemon/src/worktrees.ts @@ -14,6 +14,7 @@ export interface PreservationResult { export interface WorktreeSnapshot { identity: string; generation: number; + present: boolean; } export class WorktreeChangedError extends Error { constructor(identifier: string) { @@ -58,7 +59,10 @@ export class WorktreeManager { } } - async snapshot(rawIdentifier: string): Promise { + async snapshot( + rawIdentifier: string, + opts: { includeAbsent?: boolean } = {}, + ): Promise { const previous = this.mutation; let release!: () => void; this.mutation = new Promise((resolve) => { @@ -68,11 +72,19 @@ export class WorktreeManager { try { const identifier = this.identifier(rawIdentifier); const path = resolve(this.root, identifier); - if (!(await this.exists(path))) return undefined; + if (!(await this.exists(path))) + return opts.includeAbsent + ? { + identity: "", + generation: this.generations.get(identifier) ?? 0, + present: false, + } + : undefined; try { return { identity: await this.identityLocked(path), generation: this.generations.get(identifier) ?? 0, + present: true, }; } catch { return undefined; @@ -82,14 +94,22 @@ export class WorktreeManager { } } - async remove(rawIdentifier: string): Promise { + async remove( + rawIdentifier: string, + opts: { expectedSnapshot?: WorktreeSnapshot } = {}, + ): Promise { const previous = this.mutation; let release!: () => void; this.mutation = new Promise(resolve => { release = resolve; }); await previous; try { - const identifier = rawIdentifier.replace(/[^A-Za-z0-9-]/g, "-") || "issue"; + const identifier = this.identifier(rawIdentifier); const path = resolve(this.root, identifier); + await this.assertExpectedSnapshotLocked( + identifier, + path, + opts.expectedSnapshot, + ); if (await this.exists(path)) { await this.validate(path); if (!(await this.isClean(path))) throw new Error(`Refusing to remove dirty worktree: ${path}`); @@ -124,15 +144,11 @@ export class WorktreeManager { preserved: "none", detail: "no preservation required", }; - if (opts.expectedSnapshot) { - const generation = this.generations.get(identifier) ?? 0; - if ( - generation !== opts.expectedSnapshot.generation || - !(await this.exists(path)) || - (await this.identityLocked(path)) !== opts.expectedSnapshot.identity - ) - throw new WorktreeChangedError(identifier); - } + await this.assertExpectedSnapshotLocked( + identifier, + path, + opts.expectedSnapshot, + ); if (await this.exists(path)) { await this.validate(path); const wasDirty = @@ -292,6 +308,22 @@ export class WorktreeManager { return `${marker}:${branch}:${head}`; } + private async assertExpectedSnapshotLocked( + identifier: string, + path: string, + expected?: WorktreeSnapshot, + ): Promise { + if (!expected) return; + const generation = this.generations.get(identifier) ?? 0; + const present = await this.exists(path); + if ( + generation !== expected.generation || + present !== expected.present || + (present && (await this.identityLocked(path)) !== expected.identity) + ) + throw new WorktreeChangedError(identifier); + } + private async excludeTransientDirectories(path: string): Promise { const raw = (await this.git(["rev-parse", "--git-path", "info/exclude"], path)).trim(); const exclude = isAbsolute(raw) ? raw : resolve(path, raw); diff --git a/daemon/test/cleanup.test.ts b/daemon/test/cleanup.test.ts index 234053b..f28da58 100644 --- a/daemon/test/cleanup.test.ts +++ b/daemon/test/cleanup.test.ts @@ -54,7 +54,7 @@ async function setup() { const turn = log.claimNextTurn(2)!; log.finishTurn(turn.id, "response", "done", 2); log.markTurnActivityPosted(turn.id, 2); - return { dir, repo, root, log, tree }; + return { dir, repo, root, log, manager, tree }; } class Poster { posts: string[] = []; @@ -148,6 +148,56 @@ describe("CleanupWorker", () => { expect(onIssueFinalized).toHaveBeenCalledOnce(); s.log.close(); }); + it("AC5 retries instead of removing a session worktree reused after cleanup is claimed", async () => { + const s = await setup(); + s.log.stageExternalUrl( + "session", + "implementer", + "Pull Request", + "https://github.com/dcouple/example/pull/42", + 3, + ); + complete(s.log); + const enteredCleanCheck = deferred(); + const releaseCleanCheck = deferred(); + const originalIsClean = s.manager.isClean.bind(s.manager); + vi.spyOn(s.manager, "isClean").mockImplementation(async (path) => { + enteredCleanCheck.resolve(); + await releaseCleanCheck.promise; + return originalIsClean(path); + }); + const logger = { log: vi.fn(), error: vi.fn() }; + const worker = new CleanupWorker( + s.log, + new Poster() as unknown as LinearGateway, + s.root, + s.repo, + { + now: () => 10, + logger, + worktrees: s.manager, + }, + ); + + const drain = worker.trigger(); + await enteredCleanCheck.promise; + expect(s.log.cleanupStates()[0]?.status).toBe("running"); + await s.manager.ensureWorktree("ENG-42"); + releaseCleanCheck.resolve(); + await drain; + + expect(existsSync(s.tree.path)).toBe(true); + expect(() => + git(["show-ref", "--verify", "refs/heads/agents/ENG-42"], s.repo), + ).not.toThrow(); + expect(s.log.getSession("session")?.worktreePath).toBe(s.tree.path); + expect(s.log.cleanupStates()[0]?.status).toBe("pending"); + expect(logger.error).not.toHaveBeenCalled(); + expect(logger.log).toHaveBeenCalledWith( + expect.stringContaining('"event":"cleanup_revalidation_retry"'), + ); + s.log.close(); + }); it("preserves and removes a clean present worktree when no pull request URL was recorded", async () => { const s = await setup(); complete(s.log);