jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(server): clean up failed provider session initialization

Browse Source 
Closes #2347
This commit is contained in:
维她命@
2026-07-23 19:38:33 +02:00
committed by GitHub
parent 1ffa2f821a
commit 2bffd6e71e
10 changed files with 400 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-loading.test.ts
+82
View File
@@ -0,0 +1,82 @@
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { expect, test } from "vitest";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { AgentManager } from "./agent-manager.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import { AgentStorage } from "./agent-storage.js";
import type {
AgentClient,
AgentLaunchContext,
AgentPersistenceHandle,
AgentResumeSessionOptions,
AgentSession,
AgentSessionConfig,
} from "./agent-sdk-types.js";
import { createTestAgentClients } from "../test-utils/fake-agent-client.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-"));
const logger = createTestLogger();
const storage = new AgentStorage(path.join(root, "agents"), logger);
const baseClient = createTestAgentClients().codex;
if (!baseClient) {
throw new Error("expected Codex test client");
}
const resumeOptions: Array<AgentResumeSessionOptions | undefined> = [];
const client: AgentClient = {
provider: baseClient.provider,
capabilities: baseClient.capabilities,
createSession: async (
config: AgentSessionConfig,
launchContext?: AgentLaunchContext,
): Promise<AgentSession> => await baseClient.createSession(config, launchContext),
resumeSession: async (
handle: AgentPersistenceHandle,
overrides?: Partial<AgentSessionConfig>,
launchContext?: AgentLaunchContext,
options?: AgentResumeSessionOptions,
): Promise<AgentSession> => {
resumeOptions.push(options);
return await baseClient.resumeSession(handle, overrides, launchContext);
},
fetchCatalog: async (options) => await baseClient.fetchCatalog(options),
isAvailable: async () => await baseClient.isAvailable(),
};
const manager = new AgentManager({
clients: { codex: client },
registry: storage,
logger,
});
const archivedId = "00000000-0000-4000-8000-000000000301";
const activeId = "00000000-0000-4000-8000-000000000302";
try {
const archived = await manager.createAgent({ provider: "codex", cwd: root }, archivedId, {
workspaceId: "workspace-archived",
});
await manager.archiveAgent(archived.id);
const active = await manager.createAgent({ provider: "codex", cwd: root }, activeId, {
workspaceId: "workspace-active",
});
await manager.closeAgent(active.id);
await ensureAgentLoaded(archived.id, { agentManager: manager, agentStorage: storage, logger });
await ensureAgentLoaded(active.id, { agentManager: manager, agentStorage: storage, logger });
expect(resumeOptions).toEqual([{ purpose: "history" }, undefined]);
} finally {
await Promise.all([
manager.closeAgent(archivedId).catch(() => undefined),
manager.closeAgent(activeId).catch(() => undefined),
]);
await manager.flush().catch(() => undefined);
await storage.flush().catch(() => undefined);
await rm(root, { recursive: true, force: true });
}
});
packages/server/src/server/agent/agent-loading.ts
+1
View File
@@ -111,6 +111,7 @@ export async function ensureAgentLoaded(
buildConfigOverrides(record),
agentId,
extractTimestamps(record),
record.archivedAt ? { purpose: "history" } : undefined,
);
deps.logger.info({ agentId, provider: record.provider }, "Agent resumed from persistence");
} else {
packages/server/src/server/agent/agent-manager.ts
+10 -2
View File
@@ -19,6 +19,7 @@ import {
type AgentCapabilityFlags,
type AgentClient,
type AgentCreateSessionOptions,
type AgentResumeSessionOptions,
type AgentFeature,
type AgentLaunchContext,
type AgentSlashCommand,
@@ -1078,9 +1079,10 @@ export class AgentManager {
workspaceId?: string;
owner?: AgentOwner;
},
resumeOptions?: AgentResumeSessionOptions,
): Promise<ManagedAgent> {
return this.trackAgentRegistrationOperation(
this.resumeAgentFromPersistenceInternal(handle, overrides, agentId, options),
this.resumeAgentFromPersistenceInternal(handle, overrides, agentId, options, resumeOptions),
);
}
@@ -1096,6 +1098,7 @@ export class AgentManager {
workspaceId?: string;
owner?: AgentOwner;
},
resumeOptions?: AgentResumeSessionOptions,
): Promise<ManagedAgent> {
this.assertAcceptingAgentRegistrations();
const resolvedAgentId = validateAgentId(
@@ -1122,7 +1125,12 @@ export class AgentManager {
}
const launchContext = await this.buildLaunchContext(resolvedAgentId, client);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
const session = await client.resumeSession(handle, providerLaunchConfig, launchContext);
const session = await client.resumeSession(
handle,
providerLaunchConfig,
launchContext,
resumeOptions,
);
return this.registerSession(session, storedConfig, resolvedAgentId, {
...options,
persistence: handle,
packages/server/src/server/agent/agent-sdk-types.ts
+7
View File
@@ -602,6 +602,12 @@ export interface AgentCreateSessionOptions {
persistSession?: boolean;
}
/** Runtime-only intent for a persisted-session resume. Never persist this option. */
export interface AgentResumeSessionOptions {
/** Defaults to interactive. History loading may be read-only for archived native sessions. */
purpose?: "interactive" | "history";
}
/**
* Returned by respondToPermission when the permission resolution requires
* a follow-up turn (e.g. Codex plan approval → implementation).
@@ -688,6 +694,7 @@ export interface AgentClient {
handle: AgentPersistenceHandle,
overrides?: Partial<AgentSessionConfig>,
launchContext?: AgentLaunchContext,
options?: AgentResumeSessionOptions,
): Promise<AgentSession>;
/**
* Discover models and modes together. Implementations may use one upstream
packages/server/src/server/agent/provider-registry.ts
+2 -1
View File
@@ -408,7 +408,7 @@ function wrapClientProvider(
launchContext,
),
),
resumeSession: async (handle, overrides, launchContext) =>
resumeSession: async (handle, overrides, launchContext, options) =>
wrapSessionProvider(
provider,
await inner.resumeSession(
@@ -423,6 +423,7 @@ function wrapClientProvider(
}
: undefined,
launchContext,
options,
),
),
fetchCatalog: async (options) => {
packages/server/src/server/agent/providers/acp-agent.test.ts
+75
View File
@@ -2742,6 +2742,81 @@ describe("ACPAgentSession close() tree-kill", () => {
});
});
describe("ACPAgentSession initialization cleanup", () => {
test("terminates the ACP process when session/new fails", async () => {
const terminator = new FakeTerminator();
const child = createProbeChildStub();
class FailingNewSession extends ACPAgentSession {
protected override async spawnProcess(): Promise<SpawnedACPProcess> {
return {
child,
connection: {
newSession: vi.fn().mockRejectedValue(new Error("session/new failed")),
} as unknown as ClientSideConnection,
initialize: { agentCapabilities: {} },
};
}
}
const session = new FailingNewSession(
{ provider: "copilot", cwd: "/tmp/paseo-acp-test" },
{
provider: "copilot",
logger: createTestLogger(),
defaultCommand: ["copilot", "--acp"],
defaultModes: [],
capabilities: {
supportsStreaming: true,
supportsSessionPersistence: true,
},
terminateProcess: terminator.terminate,
},
);
await expect(session.initializeNewSession()).rejects.toThrow("session/new failed");
expect(terminator.terminated).toContain(child);
});
test("terminates the ACP process when session/load fails", async () => {
const terminator = new FakeTerminator();
const child = createProbeChildStub();
class FailingLoadSession extends ACPAgentSession {
protected override async spawnProcess(): Promise<SpawnedACPProcess> {
return {
child,
connection: {
loadSession: vi.fn().mockRejectedValue(new Error("session/load failed")),
} as unknown as ClientSideConnection,
initialize: { agentCapabilities: { loadSession: true } },
};
}
}
const session = new FailingLoadSession(
{ provider: "cursor", cwd: "/tmp/paseo-acp-test" },
{
provider: "cursor",
logger: createTestLogger(),
defaultCommand: ["cursor-agent", "acp"],
defaultModes: [],
capabilities: {
supportsStreaming: true,
supportsSessionPersistence: true,
},
handle: { provider: "cursor", sessionId: "session-1" },
terminateProcess: terminator.terminate,
},
);
await expect(session.initializeResumedSession()).rejects.toThrow("session/load failed");
expect(terminator.terminated).toContain(child);
});
});
describe("ACPAgentClient probe cleanup", () => {
afterEach(() => {
vi.restoreAllMocks();
packages/server/src/server/agent/providers/acp-agent.ts
+73 -49
View File
@@ -1376,21 +1376,25 @@ export class ACPAgentSession implements AgentSession, ACPClient {
}
async initializeNewSession(): Promise<void> {
const spawned = await this.spawnProcess();
this.child = spawned.child;
this.connection = spawned.connection;
this.agentCapabilities = spawned.initialize.agentCapabilities ?? null;
try {
const spawned = await this.spawnProcess();
this.child = spawned.child;
this.connection = spawned.connection;
this.agentCapabilities = spawned.initialize.agentCapabilities ?? null;
const response = await this.runACPRequest(() =>
this.connection!.newSession({
cwd: this.config.cwd,
mcpServers: this.acpMcpServers(),
}),
);
this.sessionId = response.sessionId;
this.bootstrapThreadEventPending = true;
this.applySessionState(response);
await this.applyConfiguredOverrides();
const response = await this.runACPRequest(() =>
this.connection!.newSession({
cwd: this.config.cwd,
mcpServers: this.acpMcpServers(),
}),
);
this.sessionId = response.sessionId;
this.bootstrapThreadEventPending = true;
this.applySessionState(response);
await this.applyConfiguredOverrides();
} catch (error) {
await this.closeAfterInitializationFailure(error);
}
}
/**
@@ -1401,45 +1405,61 @@ export class ACPAgentSession implements AgentSession, ACPClient {
* from these calls regardless of capabilities.
*/
async initializeResumedSession(): Promise<void> {
const handle = this.initialHandle;
if (!handle) {
throw new Error("Resume requested without persistence handle");
try {
const handle = this.initialHandle;
if (!handle) {
throw new Error("Resume requested without persistence handle");
}
const spawned = await this.spawnProcess();
this.child = spawned.child;
this.connection = spawned.connection;
this.agentCapabilities = spawned.initialize.agentCapabilities ?? null;
this.sessionId = handle.sessionId;
this.bootstrapThreadEventPending = true;
const sessionCapabilities = this.agentCapabilities?.sessionCapabilities;
if (this.agentCapabilities?.loadSession) {
this.replayingHistory = true;
const response = await this.runACPRequest(() =>
this.connection!.loadSession({
sessionId: handle.sessionId,
cwd: this.config.cwd,
mcpServers: this.acpMcpServers(),
}),
);
this.replayingHistory = false;
this.historyPending = this.persistedHistory.length > 0;
this.applySessionState(response);
} else if (sessionCapabilities?.resume) {
const response = await this.runACPRequest(() =>
this.connection!.unstable_resumeSession({
sessionId: handle.sessionId,
cwd: this.config.cwd,
mcpServers: this.acpMcpServers(),
}),
);
this.applySessionState(response);
} else {
throw new Error(`${this.provider} does not support ACP session resume`);
}
await this.applyConfiguredOverrides();
} catch (error) {
await this.closeAfterInitializationFailure(error);
}
}
const spawned = await this.spawnProcess();
this.child = spawned.child;
this.connection = spawned.connection;
this.agentCapabilities = spawned.initialize.agentCapabilities ?? null;
this.sessionId = handle.sessionId;
this.bootstrapThreadEventPending = true;
const sessionCapabilities = this.agentCapabilities?.sessionCapabilities;
if (this.agentCapabilities?.loadSession) {
this.replayingHistory = true;
const response = await this.runACPRequest(() =>
this.connection!.loadSession({
sessionId: handle.sessionId,
cwd: this.config.cwd,
mcpServers: this.acpMcpServers(),
}),
private async closeAfterInitializationFailure(error: unknown): Promise<never> {
try {
await this.close();
} catch (closeError) {
this.logger.warn(
{ err: closeError, initializationError: error },
"Failed to close ACP process after session initialization failure",
);
this.replayingHistory = false;
this.historyPending = this.persistedHistory.length > 0;
this.applySessionState(response);
} else if (sessionCapabilities?.resume) {
const response = await this.runACPRequest(() =>
this.connection!.unstable_resumeSession({
sessionId: handle.sessionId,
cwd: this.config.cwd,
mcpServers: this.acpMcpServers(),
}),
);
this.applySessionState(response);
} else {
throw new Error(`${this.provider} does not support ACP session resume`);
}
await this.applyConfiguredOverrides();
throw error;
}
async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
@@ -2335,6 +2355,10 @@ export class ACPAgentSession implements AgentSession, ACPClient {
{ logger: this.logger, provider: this.provider },
);
const connection = new ClientSideConnection(() => this, stream);
// Take ownership before initialize so the outer initialization guard can
// close the process even when the ACP handshake itself rejects.
this.child = child;
this.connection = connection;
const initialize = await this.runACPRequest(() =>
connection.initialize({
protocolVersion: PROTOCOL_VERSION,
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
+92 -6
View File
@@ -106,6 +106,37 @@ function createSession(
return session;
}
function createProviderWithFakeAppServer(appServer: FakeCodexAppServer): CodexAppServerAgentClient {
const provider = new CodexAppServerAgentClient(createTestLogger());
const internals = castInternals<{
goalsEnabledPromise: Promise<boolean> | null;
autoReviewEnabledPromise: Promise<boolean> | null;
spawnAppServer: () => Promise<ChildProcessWithoutNullStreams>;
}>(provider);
internals.goalsEnabledPromise = Promise.resolve(false);
internals.autoReviewEnabledPromise = Promise.resolve(false);
internals.spawnAppServer = async () => appServer.child;
return provider;
}
function archivedThreadHandle() {
return {
sessionId: "archived-thread-id",
metadata: {
cwd: "/tmp/codex-question-test",
modeId: "auto",
model: "gpt-5.4",
},
};
}
function archivedThreadErrorMessage(threadId: string): string {
return (
`session ${threadId} is archived. ` +
`Run \`codex unarchive ${threadId}\` to unarchive it first.`
);
}
function asInternals(session: CodexTestSession): CodexSessionTestAccess {
return castInternals<CodexSessionTestAccess>(session);
}
@@ -871,6 +902,66 @@ describe("Codex app-server provider", () => {
await session.close();
});
test("loads archived Codex history without resuming the native thread", async () => {
const threadRequests: string[] = [];
const appServer = createFakeCodexAppServer({
"thread/loaded/list": () => {
threadRequests.push("thread/loaded/list");
return { data: [] };
},
"thread/resume": () => {
threadRequests.push("thread/resume");
return Promise.reject(new Error(archivedThreadErrorMessage("archived-thread-id")));
},
"thread/read": () => {
threadRequests.push("thread/read");
return { thread: { turns: [] } };
},
});
const provider = createProviderWithFakeAppServer(appServer);
const session = await provider.resumeSession(archivedThreadHandle(), undefined, undefined, {
purpose: "history",
});
expect(threadRequests).toEqual(["thread/loaded/list", "thread/resume", "thread/read"]);
await session.close();
appServer.assertNoErrors();
});
test("closes Codex app-server when an interactive resume fails", async () => {
const appServer = createFakeCodexAppServer({
"thread/resume": () =>
Promise.reject(new Error(archivedThreadErrorMessage("archived-thread-id"))),
});
const killSpy = vi.spyOn(appServer.child, "kill");
const provider = createProviderWithFakeAppServer(appServer);
await expect(provider.resumeSession(archivedThreadHandle())).rejects.toThrow(
archivedThreadErrorMessage("archived-thread-id"),
);
expect(killSpy).toHaveBeenCalledWith("SIGTERM");
appServer.assertNoErrors();
});
test("closes Codex app-server when archived history hydration fails", async () => {
const appServer = createFakeCodexAppServer({
"thread/resume": () =>
Promise.reject(new Error(archivedThreadErrorMessage("archived-thread-id"))),
"thread/read": () => Promise.reject(new Error("thread history is unavailable")),
});
const killSpy = vi.spyOn(appServer.child, "kill");
const provider = createProviderWithFakeAppServer(appServer);
await expect(
provider.resumeSession(archivedThreadHandle(), undefined, undefined, { purpose: "history" }),
).rejects.toThrow("thread history is unavailable");
expect(killSpy).toHaveBeenCalledWith("SIGTERM");
appServer.assertNoErrors();
});
test("unarchives a persisted Codex thread through app-server", async () => {
const threadRequests: Array<{ method: string; params: unknown }> = [];
const appServer = createFakeCodexAppServer({
@@ -1113,12 +1204,7 @@ describe("Codex app-server provider", () => {
};
},
});
const provider = new CodexAppServerAgentClient(createTestLogger());
castInternals<{ goalsEnabledPromise: Promise<boolean> | null }>(provider).goalsEnabledPromise =
Promise.resolve(false);
castInternals<{ spawnAppServer: () => Promise<ChildProcessWithoutNullStreams> }>(
provider,
).spawnAppServer = async () => appServer.child;
const provider = createProviderWithFakeAppServer(appServer);
const outcome = await Promise.race([
provider
packages/server/src/server/agent/providers/codex-app-server-agent.ts
+48 -10
View File
@@ -6,6 +6,7 @@ import {
type AgentCreateSessionOptions,
type AgentFeature,
type AgentLaunchContext,
type AgentResumeSessionOptions,
type AgentMode,
type AgentModelDefinition,
type McpServerConfig,
@@ -110,6 +111,14 @@ function isRecord(value: unknown): value is Record<string, unknown> {
return value != null && typeof value === "object" && !Array.isArray(value);
}
function isArchivedCodexThreadResumeError(error: unknown, threadId: string): boolean {
if (!(error instanceof Error)) return false;
const expectedMessage =
`session ${threadId} is archived. ` +
`Run \`codex unarchive ${threadId}\` to unarchive it first.`;
return error.message === expectedMessage;
}
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}`);
@@ -3174,6 +3183,7 @@ export class CodexAppServerAgentSession implements AgentSession {
private readonly goalsEnabled: boolean = false,
private readonly autoReviewEnabled: boolean = false,
private readonly agentId?: string,
private readonly initialResumePurpose: "interactive" | "history" = "interactive",
) {
this.logger = logger.child({
module: "agent",
@@ -3220,18 +3230,32 @@ export class CodexAppServerAgentSession implements AgentSession {
this.client.setNotificationHandler((method, params) => this.handleNotification(method, params));
this.registerRequestHandlers();
await this.client.request("initialize", buildCodexAppServerInitializeParams());
this.client.notify("initialized", {});
try {
await this.client.request("initialize", buildCodexAppServerInitializeParams());
this.client.notify("initialized", {});
await this.loadCollaborationModes();
await this.loadSkills();
await this.loadCollaborationModes();
await this.loadSkills();
if (this.currentThreadId) {
await this.ensureThreadLoaded();
await this.loadPersistedHistory();
if (this.currentThreadId) {
await this.ensureThreadLoaded({
allowArchivedHistory: this.initialResumePurpose === "history",
});
await this.loadPersistedHistory();
}
this.connected = true;
} catch (error) {
try {
await this.close();
} catch (closeError) {
this.logger.warn(
{ err: closeError, connectError: error },
"Failed to close Codex app-server after connection failure",
);
}
throw error;
}
this.connected = true;
}
private traceContext(): CodexAppServerTraceContext {
@@ -3537,7 +3561,9 @@ export class CodexAppServerAgentSession implements AgentSession {
}
}
private async ensureThreadLoaded(): Promise<void> {
private async ensureThreadLoaded(
options: { allowArchivedHistory?: boolean } = {},
): Promise<void> {
if (!this.client || !this.currentThreadId) return;
try {
const loaded = toObjectRecord(await this.client.request("thread/loaded/list", {}));
@@ -3561,6 +3587,16 @@ export class CodexAppServerAgentSession implements AgentSession {
} catch (error) {
const threadId = this.currentThreadId;
const message = error instanceof Error ? error.message : String(error);
if (
options.allowArchivedHistory === true &&
isArchivedCodexThreadResumeError(error, threadId)
) {
this.logger.info(
{ threadId },
"Loading archived Codex thread history without resuming the native session",
);
return;
}
this.logger.warn({ error, threadId }, "Failed to resume persisted Codex thread");
throw new Error(`Failed to resume Codex thread ${threadId}: ${message}`, { cause: error });
}
@@ -6303,6 +6339,7 @@ export class CodexAppServerAgentClient implements AgentClient {
handle: { sessionId: string; metadata?: Record<string, unknown> },
overrides?: Partial<AgentSessionConfig>,
launchContext?: AgentLaunchContext,
options?: AgentResumeSessionOptions,
): Promise<AgentSession> {
const storedConfig = (handle.metadata ?? {}) as Partial<AgentSessionConfig>;
const merged: AgentSessionConfig = {
@@ -6324,6 +6361,7 @@ export class CodexAppServerAgentClient implements AgentClient {
goalsEnabled,
autoReviewEnabled,
launchContext?.agentId,
options?.purpose ?? "interactive",
);
await session.connect();
return session;