diff --git a/packages/app/src/panels/agent-panel-load-state.test.ts b/packages/app/src/panels/agent-panel-load-state.test.ts new file mode 100644 index 000000000..3bab4818a --- /dev/null +++ b/packages/app/src/panels/agent-panel-load-state.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; +import type { AgentScreenMissingState } from "@/hooks/use-agent-screen-state-machine"; +import { + clearHistorySyncErrorAfterSuccessfulSync, + reconcileMissingAgentStateWithPresentAgent, +} from "./agent-panel-load-state"; + +describe("reconcileMissingAgentStateWithPresentAgent", () => { + it("clears lookup-only states once the agent record is present", () => { + expect(reconcileMissingAgentStateWithPresentAgent({ kind: "resolving" })).toEqual({ + kind: "idle", + }); + expect( + reconcileMissingAgentStateWithPresentAgent({ + kind: "not_found", + message: "Agent not found: agent-1", + }), + ).toEqual({ kind: "idle" }); + }); + + it("preserves history sync errors while the agent record is present", () => { + const state: AgentScreenMissingState = { + kind: "error", + message: "Failed to get logs: session is archived", + }; + + expect(reconcileMissingAgentStateWithPresentAgent(state)).toBe(state); + }); +}); + +describe("clearHistorySyncErrorAfterSuccessfulSync", () => { + it("clears a sync error after a later successful refresh", () => { + expect( + clearHistorySyncErrorAfterSuccessfulSync({ + kind: "error", + message: "Failed to get logs: session is archived", + }), + ).toEqual({ kind: "idle" }); + }); + + it("leaves non-error states alone", () => { + const state: AgentScreenMissingState = { kind: "resolving" }; + + expect(clearHistorySyncErrorAfterSuccessfulSync(state)).toBe(state); + }); +}); diff --git a/packages/app/src/panels/agent-panel-load-state.ts b/packages/app/src/panels/agent-panel-load-state.ts new file mode 100644 index 000000000..9a77f0492 --- /dev/null +++ b/packages/app/src/panels/agent-panel-load-state.ts @@ -0,0 +1,19 @@ +import type { AgentScreenMissingState } from "@/hooks/use-agent-screen-state-machine"; + +export function reconcileMissingAgentStateWithPresentAgent( + state: AgentScreenMissingState, +): AgentScreenMissingState { + if (state.kind === "resolving" || state.kind === "not_found") { + return { kind: "idle" }; + } + return state; +} + +export function clearHistorySyncErrorAfterSuccessfulSync( + state: AgentScreenMissingState, +): AgentScreenMissingState { + if (state.kind === "error") { + return { kind: "idle" }; + } + return state; +} diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index 2dd9048b2..52710109a 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -53,6 +53,10 @@ import { import { useArchiveAgent } from "@/hooks/use-archive-agent"; import { useKeyboardShiftStyle } from "@/hooks/use-keyboard-shift-style"; import { useContainerWidthBelow } from "@/hooks/use-container-width"; +import { + clearHistorySyncErrorAfterSuccessfulSync, + reconcileMissingAgentStateWithPresentAgent, +} from "@/panels/agent-panel-load-state"; import { usePaneContext, usePaneFocus } from "@/panels/pane-context"; import type { PanelDescriptor, PanelRegistration } from "@/panels/panel-registry"; import { RenderProfile } from "@/utils/render-profiler"; @@ -847,9 +851,15 @@ function ChatAgentContent({ if (!agentId) { return; } - ensureAgentIsInitialized(agentId).catch((error) => { - handleHistorySyncFailure({ origin, error }); - }); + ensureAgentIsInitialized(agentId) + .then(() => { + setMissingAgentState(clearHistorySyncErrorAfterSuccessfulSync); + return undefined; + }) + .catch((error) => { + handleHistorySyncFailure({ origin, error }); + return undefined; + }); }, [agentId, ensureAgentIsInitialized, handleHistorySyncFailure], ); @@ -1011,8 +1021,8 @@ function ChatAgentContent({ return; } if (agentState.id) { - if (missingAgentState.kind !== "idle") { - setMissingAgentState({ kind: "idle" }); + if (missingAgentState.kind === "resolving" || missingAgentState.kind === "not_found") { + setMissingAgentState(reconcileMissingAgentStateWithPresentAgent); } return; } diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index d073a8dd5..502edc47f 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -134,6 +134,28 @@ class TestAgentClient implements AgentClient { } } +class NativeArchiveRecordingClient extends TestAgentClient { + readonly archivedHandles: AgentPersistenceHandle[] = []; + readonly unarchivedHandles: AgentPersistenceHandle[] = []; + readArchivedAtDuringUnarchive: (() => Promise) | null = null; + archivedAtDuringUnarchive: string | null | undefined; + unarchiveFailure: Error | null = null; + + async archiveNativeSession(handle: AgentPersistenceHandle): Promise { + this.archivedHandles.push(handle); + } + + async unarchiveNativeSession(handle: AgentPersistenceHandle): Promise { + this.unarchivedHandles.push(handle); + if (this.readArchivedAtDuringUnarchive) { + this.archivedAtDuringUnarchive = await this.readArchivedAtDuringUnarchive(); + } + if (this.unarchiveFailure) { + throw this.unarchiveFailure; + } + } +} + class EnvProbeAgentClient extends TestAgentClient { probe: Promise<{ probe: string | null; agentId: string | null }> | null = null; @@ -4443,6 +4465,113 @@ test("fires onAgentArchived for stored-only snapshot archives", async () => { expect(archivedIds).toEqual([storedOnly.id]); }); +test("unarchiveSnapshot skips native provider unarchive for active records", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-unarchive-active-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const client = new NativeArchiveRecordingClient(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + }); + + const agent = await manager.createAgent({ + provider: "codex", + cwd: workdir, + title: "Active unarchive target", + }); + + const unarchived = await manager.unarchiveSnapshot(agent.id); + + expect(unarchived).toBe(false); + expect(client.unarchivedHandles).toEqual([]); +}); + +test("unarchiveSnapshot unarchives native provider storage before clearing archivedAt", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-native-unarchive-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const client = new NativeArchiveRecordingClient(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + }); + + const agent = await manager.createAgent({ + provider: "codex", + cwd: workdir, + title: "Native unarchive target", + }); + await manager.archiveAgent(agent.id); + client.readArchivedAtDuringUnarchive = async () => (await storage.get(agent.id))?.archivedAt; + + const unarchived = await manager.unarchiveSnapshot(agent.id); + const stored = await storage.get(agent.id); + + expect(unarchived).toBe(true); + expect(client.archivedHandles).toHaveLength(1); + expect(client.unarchivedHandles).toEqual(client.archivedHandles); + expect(client.archivedAtDuringUnarchive).toEqual(expect.any(String)); + expect(stored?.archivedAt).toBeNull(); +}); + +test("unarchiveSnapshotByHandle unarchives native provider storage for the matched snapshot", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-native-unarchive-handle-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const client = new NativeArchiveRecordingClient(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + }); + + const agent = await manager.createAgent({ + provider: "codex", + cwd: workdir, + title: "Native unarchive by handle target", + }); + await manager.archiveAgent(agent.id); + const archived = await storage.get(agent.id); + if (!archived?.persistence) { + throw new Error("expected archived snapshot to have persistence"); + } + + await manager.unarchiveSnapshotByHandle(archived.persistence); + + const stored = await storage.get(agent.id); + expect(client.unarchivedHandles).toEqual(client.archivedHandles); + expect(stored?.archivedAt).toBeNull(); +}); + +test("unarchiveSnapshot keeps the stored record archived when native unarchive fails", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-native-unarchive-failure-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const client = new NativeArchiveRecordingClient(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + }); + + const agent = await manager.createAgent({ + provider: "codex", + cwd: workdir, + title: "Native unarchive failure target", + }); + await manager.archiveAgent(agent.id); + client.unarchiveFailure = new Error("provider still archived"); + + await expect(manager.unarchiveSnapshot(agent.id)).rejects.toThrow("provider still archived"); + + const stored = await storage.get(agent.id); + expect(stored?.archivedAt).toEqual(expect.any(String)); + expect(client.unarchivedHandles).toHaveLength(1); +}); + test("archiveAgent cascade archives in-memory children with the full archive contract", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-contract-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 9842d4715..09aa33a63 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1503,9 +1503,12 @@ export class AgentManager { return false; } + await this.unarchiveNativeSession(record.provider, record.persistence); + await registry.upsert({ ...record, archivedAt: null, + updatedAt: new Date().toISOString(), }); if (this.getAgent(agentId)) { @@ -3769,6 +3772,16 @@ export class AgentManager { } } + private async unarchiveNativeSession( + provider: AgentProvider, + persistence: AgentPersistenceHandle | null | undefined, + ): Promise { + if (!persistence) return; + const client = this.clients.get(provider); + if (!client?.unarchiveNativeSession) return; + await client.unarchiveNativeSession(persistence); + } + private requireAgent(id: string): LiveManagedAgent { const normalizedId = validateAgentId(id, "requireAgent"); const agent = this.agents.get(normalizedId); diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index e7c7d9973..74664b630 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -90,20 +90,12 @@ export function startAgentRun( * an archived agent unarchives it the same way. */ export async function unarchiveAgentState( - agentStorage: AgentStorage, + _agentStorage: AgentStorage, agentManager: AgentManager, agentId: string, ): Promise { - const record = await agentStorage.get(agentId); - if (!record || !record.archivedAt) { - return false; - } - const updatedAt = new Date().toISOString(); - await agentStorage.upsert({ - ...record, - archivedAt: null, - updatedAt, - }); + const unarchived = await agentManager.unarchiveSnapshot(agentId); + if (!unarchived) return false; agentManager.notifyAgentState(agentId); return true; } diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index ed91b3bbf..e411332b7 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -683,6 +683,11 @@ export interface AgentClient { * Called when Paseo archives an agent so the provider's own UI reflects the same state. */ archiveNativeSession?(handle: AgentPersistenceHandle): Promise; + /** + * Unarchive a persisted session in the native provider. + * Called before Paseo clears its archived flag so provider resume can succeed. + */ + unarchiveNativeSession?(handle: AgentPersistenceHandle): Promise; /** * Release any provider-owned resources held by this client (background * processes, sockets, cached subprocesses, etc.). Called when the daemon diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 879a8651b..f7472ef28 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -583,6 +583,123 @@ describe("Codex app-server provider", () => { await session.close(); }); + test("unarchives a persisted Codex thread through app-server", async () => { + const threadRequests: Array<{ method: string; params: unknown }> = []; + const appServer = createFakeCodexAppServer({ + "thread/unarchive": (params) => { + threadRequests.push({ method: "thread/unarchive", params }); + return { thread: { id: "native-thread-id" } }; + }, + }); + const provider = new CodexAppServerAgentClient(createTestLogger()); + castInternals<{ spawnAppServer: () => Promise }>( + provider, + ).spawnAppServer = async () => appServer.child; + + await provider.unarchiveNativeSession({ + provider: "codex", + sessionId: "persisted-thread-id", + nativeHandle: "native-thread-id", + }); + + expect(threadRequests).toEqual([ + { method: "thread/unarchive", params: { threadId: "native-thread-id" } }, + ]); + appServer.assertNoErrors(); + }); + + test("unarchives a persisted Codex thread using sessionId when nativeHandle is absent", async () => { + const threadRequests: Array<{ method: string; params: unknown }> = []; + const appServer = createFakeCodexAppServer({ + "thread/unarchive": (params) => { + threadRequests.push({ method: "thread/unarchive", params }); + return { thread: { id: "persisted-thread-id" } }; + }, + }); + const provider = new CodexAppServerAgentClient(createTestLogger()); + castInternals<{ spawnAppServer: () => Promise }>( + provider, + ).spawnAppServer = async () => appServer.child; + + await provider.unarchiveNativeSession({ + provider: "codex", + sessionId: "persisted-thread-id", + }); + + expect(threadRequests).toEqual([ + { method: "thread/unarchive", params: { threadId: "persisted-thread-id" } }, + ]); + appServer.assertNoErrors(); + }); + + test("treats a readable Codex thread as already unarchived", async () => { + const threadRequests: Array<{ method: string; params: unknown }> = []; + const appServer = createFakeCodexAppServer({ + "thread/unarchive": (params) => { + threadRequests.push({ method: "thread/unarchive", params }); + return Promise.reject( + new Error( + "failed to unarchive thread: no archived rollout found for thread id active-thread-id", + ), + ); + }, + "thread/read": (params) => { + threadRequests.push({ method: "thread/read", params }); + return { thread: { id: "active-thread-id", turns: [] } }; + }, + }); + const provider = new CodexAppServerAgentClient(createTestLogger()); + castInternals<{ spawnAppServer: () => Promise }>( + provider, + ).spawnAppServer = async () => appServer.child; + + await provider.unarchiveNativeSession({ + provider: "codex", + sessionId: "active-thread-id", + }); + + expect(threadRequests).toEqual([ + { method: "thread/unarchive", params: { threadId: "active-thread-id" } }, + { method: "thread/read", params: { threadId: "active-thread-id" } }, + ]); + appServer.assertNoErrors(); + }); + + test("propagates Codex unarchive failure when the thread cannot be read", async () => { + const threadRequests: Array<{ method: string; params: unknown }> = []; + const appServer = createFakeCodexAppServer({ + "thread/unarchive": (params) => { + threadRequests.push({ method: "thread/unarchive", params }); + return Promise.reject( + new Error( + "failed to unarchive thread: no archived rollout found for thread id missing-thread-id", + ), + ); + }, + "thread/read": (params) => { + threadRequests.push({ method: "thread/read", params }); + return Promise.reject(new Error("thread not found")); + }, + }); + const provider = new CodexAppServerAgentClient(createTestLogger()); + castInternals<{ spawnAppServer: () => Promise }>( + provider, + ).spawnAppServer = async () => appServer.child; + + await expect( + provider.unarchiveNativeSession({ + provider: "codex", + sessionId: "missing-thread-id", + }), + ).rejects.toThrow("no archived rollout found for thread id missing-thread-id"); + + expect(threadRequests).toEqual([ + { method: "thread/unarchive", params: { threadId: "missing-thread-id" } }, + { method: "thread/read", params: { threadId: "missing-thread-id" } }, + ]); + appServer.assertNoErrors(); + }); + test("rewinds the conversation to a freshly emitted Codex user message id", async () => { const appServer = createFakeCodexAppServer(); const session = new CodexAppServerAgentSession( diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index af411a644..0be6713f2 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -108,6 +108,11 @@ function isRecord(value: unknown): value is Record { return value != null && typeof value === "object" && !Array.isArray(value); } +function isCodexAlreadyUnarchivedError(error: unknown, threadId: string): boolean { + const message = error instanceof Error ? error.message : String(error); + return message.includes(`no archived rollout found for thread id ${threadId}`); +} + const TURN_START_TIMEOUT_MS = 90 * 1000; const INTERRUPT_TIMEOUT_MS = 2_000; const CODEX_PROVIDER = "codex" as const; @@ -5603,6 +5608,33 @@ export class CodexAppServerAgentClient implements AgentClient { } } + async unarchiveNativeSession(handle: AgentPersistenceHandle): Promise { + const threadId = handle.nativeHandle ?? handle.sessionId; + if (!threadId) return; + + const child = await this.spawnAppServer(); + const client = new CodexAppServerClient(child, this.logger); + + try { + await client.request("initialize", buildCodexAppServerInitializeParams()); + client.notify("initialized", {}); + try { + await client.request("thread/unarchive", { threadId }); + } catch (error) { + if (!isCodexAlreadyUnarchivedError(error, threadId)) { + throw error; + } + try { + await client.request("thread/read", { threadId }); + } catch { + throw error; + } + } + } finally { + await client.dispose(); + } + } + async isAvailable(): Promise { const launch = await resolveCodexLaunch(this.runtimeSettings); const availability = await checkCodexLaunchAvailable(launch);