From d68271565ab150ce4c854c5b40c1c7bf1604231b Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 4 Nov 2026 14:10:27 +0700 Subject: [PATCH] Read agent history after its working directory is removed (#8737) * fix(server): read agent history after working directory removal * fix(server): preserve interactive residency when reading agent history * fix(server): serialize history session upgrades * fix(server): upgrade history sessions before changing prompt mode * fix(server): keep archived history out of interactive residency * fix(server): admit configuration changes through interactive residency * fix(server): establish interactive residency before rewinding * fix(server): preserve immediate rewind locking for interactive sessions --- .../src/server/agent/agent-loading.test.ts | 292 ++++++++++++++---- .../server/src/server/agent/agent-loading.ts | 3 +- .../server/src/server/agent/agent-manager.ts | 72 ++++- .../src/server/agent/rewind/rewind.test.ts | 2 +- packages/server/src/server/session.ts | 1 + 5 files changed, 306 insertions(+), 64 deletions(-) diff --git a/packages/server/src/server/agent/agent-loading.test.ts b/packages/server/src/server/agent/agent-loading.test.ts index d75fa8739a..3f56f136c1 100644 --- a/packages/server/src/server/agent/agent-loading.test.ts +++ b/packages/server/src/server/agent/agent-loading.test.ts @@ -1,4 +1,4 @@ -import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { expect, test } from "vitest"; @@ -17,6 +17,7 @@ import type { AgentSessionConfig, } from "./agent-sdk-types.js"; import { createTestAgentClients } from "../test-utils/fake-agent-client.js"; +import { createDaemonTestContext } from "../test-utils/daemon-test-context.js"; test("loads archived records for history and active records with the interactive default", async () => { const root = await mkdtemp(path.join(tmpdir(), "agent-loading-purpose-")); @@ -142,60 +143,247 @@ test("resuming a stored agent keeps its unread flag and its last-activity time", } }); -test("loads an archived agent's history after its working directory is removed", async () => { - const root = await mkdtemp(path.join(tmpdir(), "agent-loading-missing-cwd-")); - const worktree = path.join(root, "managed-worktree"); - await mkdir(worktree, { recursive: true }); - const logger = createTestLogger(); - const storage = new AgentStorage(path.join(root, "agents"), logger); - const manager = new AgentManager({ - clients: createTestAgentClients(), - registry: storage, - logger, - }); - - const agentId = "00000000-0000-4000-8000-000000000501"; - - try { - const agent = await manager.createAgent({ provider: "codex", cwd: worktree }, agentId, { - workspaceId: "workspace-worktree", - }); - await startAgentRun(manager, agent.id, "what did you change", logger, {}); - // Dispatching a run does not finish it: the provider appends the reply to its - // history afterwards, and the turn is finalized only once that append lands. - // Archive after the turn is finalized so the transcript this test reads back is - // already on disk when the worktree goes away. - const finished = await manager.waitForAgentEvent(agent.id); - expect(finished.status).toBe("idle"); - await manager.archiveAgent(agent.id); - await manager.closeAgent(agent.id); - await manager.flush(); - await storage.flush(); - - // Archiving the workspace removes the worktree it owned. The agent's history is - // persisted and reading it must not depend on that directory still being there. - await rm(worktree, { recursive: true, force: true }); - - const loaded = await ensureAgentLoaded(agentId, { - agentManager: manager, - agentStorage: storage, +test.each([false, true])( + "loads history after its working directory is removed (archived: %s)", + async (archived) => { + const root = await mkdtemp(path.join(tmpdir(), "agent-loading-missing-cwd-")); + const worktree = path.join(root, "managed-worktree"); + await mkdir(worktree, { recursive: true }); + const logger = createTestLogger(); + const storage = new AgentStorage(path.join(root, "agents"), logger); + const manager = new AgentManager({ + clients: createTestAgentClients(), + registry: storage, logger, }); - expect(loaded.id).toBe(agentId); - // The transcript is replayed from the provider's persisted history, so the reply the - // agent gave before the worktree went away is still readable. - const replies = manager - .getTimeline(agentId) - .filter((item) => item.type === "assistant_message"); - expect(replies.length).toBeGreaterThan(0); - expect(replies.every((item) => item.type === "assistant_message" && item.text.length > 0)).toBe( - true, + const agentId = "00000000-0000-4000-8000-000000000501"; + + try { + const agent = await manager.createAgent({ provider: "codex", cwd: worktree }, agentId, { + workspaceId: "workspace-worktree", + }); + await startAgentRun(manager, agent.id, "what did you change", logger, {}); + // Dispatching a run does not finish it: the provider appends the reply to its + // history afterwards, and the turn is finalized only once that append lands. + // Archive after the turn is finalized so the transcript this test reads back is + // already on disk when the worktree goes away. + const finished = await manager.waitForAgentEvent(agent.id); + expect(finished.status).toBe("idle"); + if (archived) await manager.archiveAgent(agent.id); + await manager.closeAgent(agent.id); + await manager.flush(); + await storage.flush(); + + // Archiving the workspace removes the worktree it owned. The agent's history is + // persisted and reading it must not depend on that directory still being there. + await rm(worktree, { recursive: true, force: true }); + + const loaded = await ensureAgentLoaded(agentId, { + agentManager: manager, + agentStorage: storage, + logger, + purpose: "history", + }); + + expect(loaded.id).toBe(agentId); + // The transcript is replayed from the provider's persisted history, so the reply the + // agent gave before the worktree went away is still readable. + const replies = manager + .getTimeline(agentId) + .filter((item) => item.type === "assistant_message"); + expect(replies.length).toBeGreaterThan(0); + expect( + replies.every((item) => item.type === "assistant_message" && item.text.length > 0), + ).toBe(true); + } finally { + await manager.closeAgent(agentId).catch(() => undefined); + await manager.flush().catch(() => undefined); + await storage.flush().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } + }, +); + +test("timeline requests read unarchived stored history with a deleted cwd", async () => { + const ctx = await createDaemonTestContext(); + const cwd = await mkdtemp(path.join(tmpdir(), "timeline-deleted-cwd-")); + try { + const agent = await ctx.client.createAgent({ provider: "codex", cwd }); + await ctx.client.sendMessage(agent.id, "remember this conversation"); + await ctx.client.waitForFinish(agent.id, 15000); + const before = await ctx.client.fetchAgentTimeline(agent.id); + await ctx.daemon.daemon.agentManager.closeAgent(agent.id); + await ctx.daemon.daemon.agentManager.flush(); + await rm(cwd, { recursive: true, force: true }); + const after = await ctx.client.fetchAgentTimeline(agent.id); + expect(after.entries).toEqual(before.entries); + expect(after.entries.length).toBeGreaterThan(0); + await expect(ctx.client.sendMessage(agent.id, "continue")).rejects.toThrow( + "Working directory does not exist", ); + await expect(ctx.client.createAgent({ provider: "codex", cwd })).rejects.toThrow( + "Working directory does not exist", + ); + await mkdir(cwd); + await ctx.client.sendMessage(agent.id, "continue after restoring directory"); + expect((await ctx.client.waitForFinish(agent.id, 15000)).status).toBe("idle"); } finally { - await manager.closeAgent(agentId).catch(() => undefined); - await manager.flush().catch(() => undefined); - await storage.flush().catch(() => undefined); - await rm(root, { recursive: true, force: true }); + await ctx.cleanup(); + await rm(cwd, { recursive: true, force: true }); } -}); +}, 30000); + +test("timeline reads keep runnable unarchived agents interactive for the next prompt", async () => { + const clients = createTestAgentClients(); + const base = clients.codex!; + const purposes: Array = []; + const client: AgentClient = { + provider: base.provider, + capabilities: base.capabilities, + createSession: (config, context) => base.createSession(config, context), + resumeSession: (handle, overrides, context, options) => { + purposes.push(options); + return base.resumeSession(handle, overrides, context, options); + }, + fetchCatalog: (options) => base.fetchCatalog(options), + isAvailable: () => base.isAvailable(), + }; + const ctx = await createDaemonTestContext({ agentClients: { codex: client } }); + const cwd = await mkdtemp(path.join(tmpdir(), "timeline-interactive-")); + try { + const agent = await ctx.client.createAgent({ provider: "codex", cwd }); + await ctx.daemon.daemon.agentManager.closeAgent(agent.id); + await ctx.client.fetchAgentTimeline(agent.id); + expect(purposes).toEqual([{ purpose: "interactive" }]); + await ctx.client.sendMessage(agent.id, "continue after opening history"); + expect((await ctx.client.waitForFinish(agent.id, 15000)).status).toBe("idle"); + const timeline = await ctx.client.fetchAgentTimeline(agent.id); + expect(timeline.entries.length).toBeGreaterThan(0); + await ctx.daemon.daemon.agentManager.closeAgent(agent.id); + await rm(cwd, { recursive: true, force: true }); + await ctx.client.fetchAgentTimeline(agent.id); + await mkdir(cwd); + const manager = ctx.daemon.daemon.agentManager; + await Promise.all([ + manager.tryRunOutOfBand(agent.id, "first prompt"), + manager.tryRunOutOfBand(agent.id, "second prompt"), + ]); + expect(purposes).toEqual([ + { purpose: "interactive" }, + { purpose: "history" }, + undefined, // Reload uses the provider's interactive default. + ]); + } finally { + await ctx.cleanup(); + await rm(cwd, { recursive: true, force: true }); + } +}, 30000); + +test.each(["mode", "thinking", "rewind"])( + "restored OMP history accepts %s before starting an interactive turn", + async (setting) => { + const { OmpAgentClient } = await import("./providers/omp/agent.js"); + const { FakeOmp } = await import("./providers/omp/test-utils/fake-omp.js"); + const { sendPreparedPromptToAgent } = await import("./agent-prompt.js"); + const root = await mkdtemp(path.join(tmpdir(), "history-mode-")); + const cwd = path.join(root, "removed-worktree"); + const sessionFile = path.join(root, "session.jsonl"); + await writeFile( + sessionFile, + [ + { type: "session", id: "root", parentId: null }, + { + type: "message", + id: "user-1", + parentId: "root", + message: { role: "user", content: "saved question" }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n") + "\n", + ); + const logger = createTestLogger(); + const storage = new AgentStorage(path.join(root, "agents"), logger); + const runtime = new FakeOmp(); + const manager = new AgentManager({ + clients: { + omp: new OmpAgentClient({ + logger, + runtime, + runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } }, + }), + }, + registry: storage, + logger, + }); + const agentId = "00000000-0000-4000-8000-000000000601"; + const now = new Date().toISOString(); + try { + await storage.upsert({ + id: agentId, + provider: "omp", + cwd, + createdAt: now, + updatedAt: now, + labels: {}, + lastStatus: "closed", + archivedAt: null, + persistence: { + provider: "omp", + sessionId: "history-mode", + nativeHandle: sessionFile, + metadata: { cwd }, + }, + }); + await ensureAgentLoaded(agentId, { + agentManager: manager, + agentStorage: storage, + logger, + purpose: "history", + }); + expect( + runtime.recordedLaunches.filter((launch) => launch.session === sessionFile), + ).toHaveLength(0); + await mkdir(cwd); + if (setting === "rewind") { + await manager.rewind(agentId, "user-1", "conversation"); + expect(runtime.latestSession().branchRequests).toEqual(["user-1"]); + } + if (setting === "thinking") { + await manager.setAgentThinkingOption(agentId, "high"); + expect(manager.getAgent(agentId)?.config.thinkingOptionId).toBe("high"); + expect(runtime.latestSession().setThinkingLevelRequests).toContain("high"); + } + await sendPreparedPromptToAgent({ + agentManager: manager, + agentStorage: storage, + agentId, + logger, + prompt: "continue", + sessionMode: "full", + }); + expect( + runtime.recordedLaunches.filter((launch) => launch.session === sessionFile), + ).toHaveLength(1); + expect(manager.getAgent(agentId)?.currentModeId).toBe("full"); + expect(runtime.latestSession().prompts).toEqual([ + expect.objectContaining({ message: "continue" }), + ]); + runtime.latestSession().finishTurn(); + await manager.waitForAgentEvent(agentId); + await manager.archiveAgent(agentId); + await manager.closeAgent(agentId); + await ensureAgentLoaded(agentId, { agentManager: manager, agentStorage: storage, logger }); + const launchesBeforeModeChange = runtime.recordedLaunches.length; + await expect(manager.setAgentMode(agentId, "full")).rejects.toThrow("Agent is archived"); + expect(runtime.recordedLaunches).toHaveLength(launchesBeforeModeChange); + } finally { + await manager.closeAgent(agentId).catch(() => undefined); + await manager.flush(); + await storage.flush(); + await rm(root, { recursive: true, force: true }); + } + }, + 30000, +); diff --git a/packages/server/src/server/agent/agent-loading.ts b/packages/server/src/server/agent/agent-loading.ts index fdfdbcf031..c683a4ec77 100644 --- a/packages/server/src/server/agent/agent-loading.ts +++ b/packages/server/src/server/agent/agent-loading.ts @@ -41,6 +41,7 @@ export interface EnsureAgentLoadedDeps { agentStorage: AgentStorage; validProviders?: Iterable; broadcastTimeline?: boolean; + purpose?: "history" | "interactive"; logger: Logger; } @@ -118,7 +119,7 @@ export async function ensureAgentLoaded( buildConfigOverrides(record), agentId, { ...extractTimestamps(record), attention: extractAttention(record) }, - record.archivedAt ? { purpose: "history" } : undefined, + deps.purpose ? { purpose: deps.purpose } : undefined, ); deps.logger.info({ agentId, provider: record.provider }, "Agent resumed from persistence"); } else { diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 32fa4ca4a6..387d76082d 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -171,6 +171,23 @@ async function assertUsableWorkingDirectory(cwd: string): Promise { } } +async function resolveResumePurpose( + record: StoredAgentRecord | null, + requested: AgentResumePurpose | undefined, + cwd: string, +): Promise { + if (!record) return requested ?? "interactive"; + if (record.archivedAt) return "history"; + if (requested !== "history") return "interactive"; + try { + await stat(cwd); + return "interactive"; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return "history"; + throw error; + } +} + interface PreparedSessionConfig { storedConfig: AgentSessionConfig; launchConfig: AgentSessionConfig; @@ -783,6 +800,7 @@ export class AgentManager { private readonly backgroundTasks = new Set>(); private readonly agentRegistrationTasks = new Set>(); private readonly inFlightAgentCloses = new Map>(); + private readonly historySessions = new WeakSet(); private readonly reloadedSessionCloses = new WeakMap>(); private readonly runtimeCleanupIdentities = new WeakMap(); private readonly collectingRuntimes = new Set(); @@ -1501,10 +1519,8 @@ export class AgentManager { // settled before the config is prepared, because a history load reads an archived // agent whose working directory may be gone. const record = this.registry ? await this.registry.get(resolvedAgentId) : null; - const currentResumeOptions = record - ? { purpose: record.archivedAt ? ("history" as const) : ("interactive" as const) } - : resumeOptions; - const purpose = currentResumeOptions?.purpose ?? "interactive"; + const purpose = await resolveResumePurpose(record, resumeOptions?.purpose, mergedConfig.cwd); + const currentResumeOptions = { purpose } satisfies AgentResumeSessionOptions; const launchEnv = record?.launchEnv; const { storedConfig, launchConfig, paseoToolPolicy } = await this.prepareSessionConfig( @@ -1541,6 +1557,7 @@ export class AgentManager { launchContext, currentResumeOptions, ); + if (purpose === "history") this.historySessions.add(session); await this.requireExternalMcpSupport(session, providerLaunchConfig, launchContext); return this.registerSession(session, storedConfig, resolvedAgentId, { ...options, @@ -2204,7 +2221,7 @@ export class AgentManager { } async setAgentMode(agentId: string, modeId: string): Promise { - const agent = this.requireSessionAgent(agentId); + const agent = await this.ensureInteractiveSession(agentId); const notice = (await agent.session.setMode(modeId)) ?? null; await this.drainSessionEvents(agentId); const currentMode = (await agent.session.getCurrentMode()) ?? modeId; @@ -2220,7 +2237,7 @@ export class AgentManager { } async setAgentModel(agentId: string, modelId: string | null): Promise { - const agent = this.requireSessionAgent(agentId); + const agent = await this.ensureInteractiveSession(agentId); const normalizedModelId = typeof modelId === "string" && modelId.trim().length > 0 ? modelId : null; @@ -2242,7 +2259,7 @@ export class AgentManager { agentId: string, thinkingOptionId: string | null, ): Promise { - const agent = this.requireSessionAgent(agentId); + const agent = await this.ensureInteractiveSession(agentId); const normalizedThinkingOptionId = typeof thinkingOptionId === "string" && thinkingOptionId.trim().length > 0 ? thinkingOptionId @@ -2273,7 +2290,7 @@ export class AgentManager { } async setAgentFeature(agentId: string, featureId: string, value: unknown): Promise { - const agent = this.requireAgent(agentId); + const agent = await this.ensureInteractiveSession(agentId); if (!agent.session.setFeature) { throw new Error("Agent session does not support setting features"); @@ -2655,7 +2672,7 @@ export class AgentManager { logger: this.logger, }); } - const agent = this.requireSessionAgent(agentId); + const agent = await this.ensureInteractiveSession(agentId); // Providers parse slash commands from the raw text, so a command another agent sent // must reach them without its sender envelope. const agentMessage = typeof prompt === "string" ? parseAgentMessage(prompt) : null; @@ -2784,6 +2801,34 @@ export class AgentManager { } } + private async ensureInteractiveSession(agentId: string): Promise { + const agent = this.requireSessionAgent(agentId); + if (this.historySessions.has(agent.session)) { + await this.trackAgentRegistrationOperation( + this.runLifecycleMutation(agentId, async () => { + const record = await this.registry?.get(agentId); + if (record?.archivedAt) throw new Error(`Agent is archived: ${agentId}`); + const current = this.requireSessionAgent(agentId); + if (this.historySessions.has(current.session)) { + // Recheck inside the lane: another prompt may already have upgraded it. + // Reload validates the directory before releasing readable history. + await this.reloadAgentSessionInternal(agentId); + } + }), + ); + } + return this.requireSessionAgent(agentId); + } + + private async *streamAfterHistoryLoad( + agentId: string, + prompt: AgentPromptInput, + options?: AgentRunOptions, + ): AsyncGenerator { + await this.ensureInteractiveSession(agentId); + yield* this.streamAgent(agentId, prompt, options); + } + private async *streamAfterRuntimeCollection( agentId: string, prompt: AgentPromptInput, @@ -2806,6 +2851,8 @@ export class AgentManager { if (this.collectingRuntimes.has(agentId) || (!this.agents.has(agentId) && this.registry)) return this.streamAfterRuntimeCollection(agentId, prompt, options); const existingAgent = this.requireSessionAgent(agentId); + if (this.historySessions.has(existingAgent.session)) + return this.streamAfterHistoryLoad(agentId, prompt, options); this.logger.trace( { agentId, @@ -3492,7 +3539,12 @@ export class AgentManager { } async rewind(agentId: string, messageId: string, mode: RewindMode): Promise { - const agent = this.requireSessionAgent(agentId); + let agent = this.requireSessionAgent(agentId); + if (this.historySessions.has(agent.session)) { + agent = await this.ensureInteractiveSession(agentId); + } + // An interactive session must acquire its rewind lock without yielding. + const submittedRow = this.timelineStore .getRows(agentId) .find( diff --git a/packages/server/src/server/agent/rewind/rewind.test.ts b/packages/server/src/server/agent/rewind/rewind.test.ts index df5c242251..73d98e8741 100644 --- a/packages/server/src/server/agent/rewind/rewind.test.ts +++ b/packages/server/src/server/agent/rewind/rewind.test.ts @@ -162,7 +162,7 @@ describe("AgentManager rewind", () => { }); }); - test("blocks new prompts until the rehydrate epoch broadcasts", async () => { + test("blocks new prompts immediately and until the rehydrate epoch broadcasts", async () => { const historyGate = new RewindHistoryGate(); historyGate.hold(); const { manager, agentId } = await createRewindHarness({ historyGate }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index d112419e7f..4b260f8550 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -8047,6 +8047,7 @@ export class Session { agentManager: this.agentManager, agentStorage: this.agentStorage, logger: this.sessionLogger, + purpose: "history", }); const agentPayload = await this.buildAgentPayload(snapshot);