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
This commit is contained in:
Mohamed Boudra
2026-11-04 14:10:27 +07:00
committed by GitHub
parent ce7b14a143
commit d68271565a
5 changed files with 306 additions and 64 deletions
@@ -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<AgentResumeSessionOptions | undefined> = [];
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,
);
@@ -41,6 +41,7 @@ export interface EnsureAgentLoadedDeps {
agentStorage: AgentStorage;
validProviders?: Iterable<AgentProvider>;
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 {
@@ -171,6 +171,23 @@ async function assertUsableWorkingDirectory(cwd: string): Promise<void> {
}
}
async function resolveResumePurpose(
record: StoredAgentRecord | null,
requested: AgentResumePurpose | undefined,
cwd: string,
): Promise<AgentResumePurpose> {
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<Promise<void>>();
private readonly agentRegistrationTasks = new Set<Promise<void>>();
private readonly inFlightAgentCloses = new Map<string, Promise<void>>();
private readonly historySessions = new WeakSet<AgentSession>();
private readonly reloadedSessionCloses = new WeakMap<AgentSession, Promise<void>>();
private readonly runtimeCleanupIdentities = new WeakMap<AgentSession, object>();
private readonly collectingRuntimes = new Set<string>();
@@ -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<AgentProviderNotice | null> {
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<void> {
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<AgentProviderNotice | null> {
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<void> {
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<ActiveManagedAgent> {
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<AgentStreamEvent> {
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<void> {
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(
@@ -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 });
+1
View File
@@ -8047,6 +8047,7 @@ export class Session {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.sessionLogger,
purpose: "history",
});
const agentPayload = await this.buildAgentPayload(snapshot);