fix(server): keep earlier replies visible after agent resume (#9098)

* fix(server): keep earlier replies visible after agent resume

Idle runtime collection retains timeline rows but recreates provider sessions. Scope turn counters to a fresh UUID namespace so resumed sessions cannot reuse persisted turn identities.

* test(server): require a turn ID before inspecting its namespace
This commit is contained in:
Mohamed Boudra
2026-11-08 16:09:45 +07:00
committed by GitHub
parent 02ad59908f
commit 1b0da5fce5
11 changed files with 154 additions and 56 deletions
@@ -469,7 +469,7 @@ class TestAgentSession implements AgentSession {
async startTurn(): Promise<{ turnId: string }> {
this.interrupted = false;
const turnId = `turn-${++this.turnIdCounter}`;
const turnId = `turn-${this.id}-${++this.turnIdCounter}`;
// Use setTimeout so events arrive after the caller sets up the foreground waiter
setTimeout(() => {
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
@@ -13487,3 +13487,35 @@ test("stops only a currently owned public background item with its current token
expect(stopped).toEqual(["one"]);
await manager.closeAgent(agent.id);
});
test("turns keep distinct identities across an idle runtime collection and resume", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-turn-identity-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const client = new TestAgentClient();
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
const collect = runtimeCleanup(manager);
try {
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
const consume = async (prompt: string, clientMessageId: string) => {
for await (const _event of manager.streamAgent(agent.id, prompt, { clientMessageId })) {
}
};
await consume("first", "first");
const later = new Date(Date.now() + 61 * 60_000);
expect((await collect({ now: later })).collected).toHaveLength(1);
await ensureAgentLoaded(agent.id, { agentManager: manager, agentStorage: storage, logger });
await consume("second", "second");
const rows = manager.fetchTimeline(agent.id, { limit: 0 }).rows;
const userTurnIds = rows
.filter((row) => row.item.type === "user_message")
.map((row) => row.turnId);
expect(userTurnIds).toHaveLength(2);
expect(new Set(userTurnIds).size).toBe(2);
} finally {
await manager.closeAgent(manager.listAgents()[0]?.id ?? "missing").catch(() => undefined);
await storage.flush();
rmSync(workdir, { recursive: true, force: true });
}
});
@@ -690,6 +690,7 @@ export interface AgentSession {
* replay them at their original timestamps; restored sessions omit old rows. */
readonly initialTimeline?: ImportedTimelineEntry[];
run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult>;
/** Returns an ID unique across the agent lifetime, including resumed sessions and daemon restarts. */
startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>;
steerActiveTurn?(prompt: AgentPromptInput, options: SteerActiveTurnOptions): Promise<SteerResult>;
subscribe(callback: (event: AgentStreamEvent) => void): () => void;
@@ -1206,7 +1206,7 @@ test("reuses one autonomous run for unbound stream_event bursts with no foregrou
);
const firstRunId = internal.autonomousTurn?.id ?? null;
expect(firstRunId).toBe("autonomous-turn-1");
expect(firstRunId).toEqual(expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/));
expect(internal.nextTurnOrdinal).toBe(2);
await internal.routeSdkMessageFromPump(
@@ -1360,7 +1360,9 @@ test("a main-session frame opens no turn until Claude starts one", async () => {
// session start a turn of its own, with no prompt from the user.
await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
await internal.routeSdkMessageFromPump(unboundTextDelta("picking this up"), queryMock);
expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
expect(internal.autonomousTurn?.id).toEqual(
expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
);
await internal.routeSdkMessageFromPump(successResult(), queryMock);
expect(internal.autonomousTurn).toBeNull();
@@ -1410,7 +1412,9 @@ test("Claude waking up after a helper finishes gets an autonomous turn that its
expect(internal.autonomousTurn).toBeNull();
await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
expect(internal.autonomousTurn?.id).toEqual(
expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
);
expect(internal.turnState).toBe("autonomous");
await internal.routeSdkMessageFromPump(unboundTextDelta("the helper is done"), queryMock);
@@ -1446,7 +1450,9 @@ test("a steer folded into a turn keeps one turn, and a turn Claude starts after
// A message that arrived while Claude wrote its final answer runs as a turn of its own.
await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
expect(internal.autonomousTurn?.id).toBe("autonomous-turn-2");
expect(internal.autonomousTurn?.id).toEqual(
expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-2$/),
);
await internal.routeSdkMessageFromPump(successResult(), queryMock);
expect(internal.autonomousTurn).toBeNull();
expect(internal.turnState).toBe("idle");
@@ -1483,7 +1489,9 @@ test("an idle session state closes an autonomous turn that no result closed", as
session.subscribe((event) => events.push(event));
await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
expect(internal.autonomousTurn?.id).toEqual(
expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
);
// Safety net: Claude says nothing at all is running any more.
await internal.routeSdkMessageFromPump(sessionState("idle"), queryMock);
@@ -1685,7 +1693,9 @@ test("a new turn clears an interrupt flag no result will consume", async () => {
await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
expect(internal.pendingInterruptAbort).toBe(false);
expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
expect(internal.autonomousTurn?.id).toEqual(
expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
);
// This turn's own failure is reported, not swallowed as the interrupted turn's leftover.
await internal.routeSdkMessageFromPump(
@@ -1345,6 +1345,30 @@ describe("ClaudeAgentSession features", () => {
await session.close();
});
test("keeps turn identities distinct across resumed session objects", async () => {
const { queryFactory } = createQueryMock();
const client = new ClaudeAgentClient({
logger,
queryFactory,
resolveBinary: async () => "/test/claude/bin",
});
const handle = {
provider: "claude" as const,
sessionId: "saved-session",
metadata: { cwd: process.cwd() },
};
const first = await client.resumeSession(handle);
const firstTurn = await first.startTurn("first");
await first.close();
const resumed = await client.resumeSession(handle);
try {
const nextTurn = await resumed.startTurn("second");
expect(nextTurn.turnId).not.toBe(firstTurn.turnId);
} finally {
await resumed.close();
}
});
test("disables Claude hooks for internal agents only", async () => {
const { queryFactory } = createQueryMock();
const client = new ClaudeAgentClient({
@@ -2378,6 +2378,7 @@ class ClaudeAgentSession implements AgentSession {
private historySessionId: string | null = null;
private replayingHistory = false;
private turnState: TurnState = "idle";
private readonly turnNamespace = randomUUID();
private nextTurnOrdinal = 1;
private cancelCurrentTurn: (() => void) | null = null;
private cachedRuntimeInfo: AgentRuntimeInfo | null = null;
@@ -3983,7 +3984,7 @@ class ClaudeAgentSession implements AgentSession {
}
private createTurnId(owner: "foreground" | "autonomous"): string {
return `${owner}-turn-${this.nextTurnOrdinal++}`;
return `${owner}-turn-${this.turnNamespace}-${this.nextTurnOrdinal++}`;
}
private isTerminalTurnEvent(event: AgentStreamEvent): boolean {
@@ -1359,7 +1359,7 @@ describe("Codex app-server provider", () => {
const firstSession = await createProviderWithFakeAppServer(beforeRestart).createSession(
createConfig({ thinkingOptionId: undefined }),
);
await firstSession.startTurn("first turn");
const firstTurn = await firstSession.startTurn("first turn");
await expect(beforeRestart.waitForTurnStart()).resolves.toMatchObject({ effort: "medium" });
const saved = waitForNextEvent(firstSession, "thread_started");
beforeRestart.startsTurn({ threadId: "thread-1" });
@@ -1378,7 +1378,8 @@ describe("Codex app-server provider", () => {
await expect(resumed.getRuntimeInfo()).resolves.toMatchObject({
thinkingOptionId: "medium",
});
await resumed.startTurn("turn after restart");
const resumedTurn = await resumed.startTurn("turn after restart");
expect(resumedTurn.turnId).not.toBe(firstTurn.turnId);
await expect(afterRestart.waitForTurnStart()).resolves.toMatchObject({ effort: "medium" });
beforeRestart.assertNoErrors();
afterRestart.assertNoErrors();
@@ -4853,6 +4854,7 @@ describe("Codex app-server provider", () => {
);
try {
const started = waitForNextEvent(session, "turn_started");
const resultPromise = session.run("Delegate the investigation.");
await appServer.waitForTurnStart();
const child = waitForProviderSubagent(session, "legacy-only-child-thread");
@@ -4870,7 +4872,7 @@ describe("Codex app-server provider", () => {
await expect(child).resolves.toMatchObject({
type: "provider_subagent",
provider: "codex",
turnId: "codex-turn-0",
turnId: (await started).turnId,
event: {
type: "upsert",
id: "legacy-only-child-thread",
@@ -4880,7 +4882,7 @@ describe("Codex app-server provider", () => {
await expect(spawn).resolves.toMatchObject({
type: "timeline",
provider: "codex",
turnId: "codex-turn-0",
turnId: (await started).turnId,
item: {
type: "tool_call",
callId: "spawn-legacy-only-child",
@@ -3529,6 +3529,7 @@ export class CodexAppServerAgentSession implements AgentSession {
private threadRollbackAvailable = true;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private startupEvents: AgentStreamEvent[] | null = null;
private readonly turnNamespace = randomUUID();
private nextTurnOrdinal = 0;
private activeForegroundTurnId: string | null = null;
private activeClientMessageId: string | null = null;
@@ -5719,7 +5720,7 @@ export class CodexAppServerAgentSession implements AgentSession {
}
private createTurnId(): string {
return `codex-turn-${this.nextTurnOrdinal++}`;
return `codex-turn-${this.turnNamespace}-${this.nextTurnOrdinal++}`;
}
private pendingNotifications: Promise<void> = Promise.resolve();
@@ -32,6 +32,16 @@ import type {
const EXPECTED_STARTUP_BUDGET_MS = 30_000;
const EXPECTED_EVENT_READINESS_BUDGET_MS = 45_000;
// Keep event correlation and turn ordering assertions exact while accepting a fresh namespace.
function turnIdAt(events: AgentStreamEvent[], ordinal: number): string {
const started = events.find((event) => event.type === "turn_started");
if (!started || started.type !== "turn_started" || typeof started.turnId !== "string") {
throw new Error("Expected a turn_started event with a turn ID");
}
const namespace = started.turnId.slice(0, started.turnId.lastIndexOf("-"));
return `${namespace}-${ordinal}`;
}
function tmpCwd(): string {
const dir = mkdtempSync(path.join(os.tmpdir(), "opencode-agent-test-"));
try {
@@ -757,6 +767,19 @@ describe("OpenCodeAgentClient adapter smoke tests", () => {
}
});
test("keeps turn identities distinct for the same persisted session", async () => {
const first = await createParentSession("ses_persisted");
const firstTurn = await first.parent.startTurn("first");
await first.parent.close();
const resumed = await createParentSession("ses_persisted");
try {
const nextTurn = await resumed.parent.startTurn("second");
expect(nextTurn.turnId).not.toBe(firstTurn.turnId);
} finally {
await resumed.parent.close();
}
});
test.each([
{ saved: "default", variant: undefined },
{ saved: "high", variant: "high" },
@@ -1638,10 +1661,10 @@ describe("OpenCode adapter startTurn error handling", () => {
item: { type: "user_message", clientMessageId: "client-message-1" },
});
expect(turn.events.map((event) => ("turnId" in event ? event.turnId : undefined))).toEqual([
"opencode-turn-0",
"opencode-turn-0",
"opencode-turn-0",
"opencode-turn-0",
turnIdAt(turn.events, 0),
turnIdAt(turn.events, 0),
turnIdAt(turn.events, 0),
turnIdAt(turn.events, 0),
]);
});
@@ -1833,7 +1856,7 @@ describe("OpenCode adapter startTurn error handling", () => {
type: "error",
message: "Provider retry (attempt 1): model does not exist",
},
turnId: "opencode-turn-0",
turnId: turnIdAt(events, 0),
});
expect(events.some((event) => event.type === "turn_failed")).toBe(false);
await session.close();
@@ -2641,11 +2664,11 @@ describe("OpenCode adapter startTurn error handling", () => {
expect(
turnEventSignatures(events.filter((event) => event.type !== "provider_subagent")),
).toEqual([
["turn_started", "opencode-turn-0"],
["turn_canceled", "opencode-turn-0"],
["turn_started", "opencode-turn-1"],
["timeline", "opencode-turn-1"],
["turn_completed", "opencode-turn-1"],
["turn_started", turnIdAt(events, 0)],
["turn_canceled", turnIdAt(events, 0)],
["turn_started", turnIdAt(events, 1)],
["timeline", turnIdAt(events, 1)],
["turn_completed", turnIdAt(events, 1)],
]);
} finally {
settleRetry.resolve();
@@ -2706,7 +2729,7 @@ describe("OpenCode adapter startTurn error handling", () => {
expect(events).toContainEqual(
expect.objectContaining({
type: "timeline",
turnId: "opencode-turn-1",
turnId: turnIdAt(events, 1),
item: expect.objectContaining({ text: "Wake output streams live." }),
}),
),
@@ -2726,11 +2749,11 @@ describe("OpenCode adapter startTurn error handling", () => {
});
await vi.waitFor(() =>
expect(events).toContainEqual(
expect.objectContaining({ type: "turn_completed", turnId: "opencode-turn-1" }),
expect.objectContaining({ type: "turn_completed", turnId: turnIdAt(events, 1) }),
),
);
await expect(session.startTurn("after the wake")).resolves.toEqual({
turnId: "opencode-turn-2",
turnId: turnIdAt(events, 2),
});
} finally {
settleAbort.resolve();
@@ -2843,13 +2866,13 @@ describe("OpenCode adapter startTurn error handling", () => {
expect(events).toContainEqual({
type: "turn_started",
provider: "opencode",
turnId: "opencode-turn-1",
turnId: turnIdAt(events, 1),
});
await vi.waitFor(() =>
expect(events).toContainEqual({
type: "turn_completed",
provider: "opencode",
turnId: "opencode-turn-1",
turnId: turnIdAt(events, 1),
usage: expect.objectContaining({
inputTokens: 7,
outputTokens: 3,
@@ -2858,7 +2881,7 @@ describe("OpenCode adapter startTurn error handling", () => {
}),
);
const autonomousCompletion = events.find(
(event) => event.type === "turn_completed" && event.turnId === "opencode-turn-1",
(event) => event.type === "turn_completed" && event.turnId === turnIdAt(events, 1),
);
expect(autonomousCompletion?.type).toBe("turn_completed");
if (autonomousCompletion?.type === "turn_completed") {
@@ -2867,7 +2890,7 @@ describe("OpenCode adapter startTurn error handling", () => {
expect(events).toContainEqual(
expect.objectContaining({
type: "timeline",
turnId: "opencode-turn-1",
turnId: turnIdAt(events, 1),
item: expect.objectContaining({
type: "assistant_message",
text: "Autonomous wake completed.",
@@ -2875,7 +2898,7 @@ describe("OpenCode adapter startTurn error handling", () => {
}),
);
await expect(session.startTurn("after autonomous wake")).resolves.toEqual({
turnId: "opencode-turn-2",
turnId: turnIdAt(events, 2),
});
} finally {
settleAbort.resolve();
@@ -2954,7 +2977,7 @@ describe("OpenCode adapter startTurn error handling", () => {
type: "turn_canceled",
provider: "opencode",
reason: "interrupted",
turnId: "opencode-turn-0",
turnId: turnIdAt(events, 0),
});
openCode.emitEvent({
@@ -2962,7 +2985,7 @@ describe("OpenCode adapter startTurn error handling", () => {
properties: { sessionID: "ses_retried_stop" },
});
await expect(session.startTurn("replacement")).resolves.toEqual({
turnId: "opencode-turn-1",
turnId: turnIdAt(events, 1),
});
expect(openCode.calls.sessionPromptAsync).toHaveLength(2);
} finally {
@@ -3058,7 +3081,7 @@ describe("OpenCode adapter startTurn error handling", () => {
expect(events).toContainEqual({
type: "turn_started",
provider: "opencode",
turnId: "opencode-turn-1",
turnId: turnIdAt(events, 1),
}),
);
@@ -3767,7 +3790,7 @@ describe("OpenCode adapter startTurn error handling", () => {
];
const replacement = session.startTurn("second");
await vi.advanceTimersByTimeAsync(10_000);
await expect(replacement).resolves.toEqual({ turnId: "opencode-turn-1" });
await expect(replacement).resolves.toEqual({ turnId: turnIdAt(events, 1) });
expect(countEvents(events, "turn_canceled")).toBe(1);
} finally {
vi.useRealTimers();
@@ -4009,6 +4032,9 @@ describe("OpenCode adapter startTurn error handling", () => {
return { data: true };
};
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await session.startTurn("first");
const interrupt = session.interrupt();
@@ -4022,7 +4048,7 @@ describe("OpenCode adapter startTurn error handling", () => {
expect(openCode.calls.sessionStatus).toHaveLength(1);
releaseIdle.resolve();
await expect(replacement).resolves.toEqual({ turnId: "opencode-turn-1" });
await expect(replacement).resolves.toEqual({ turnId: turnIdAt(events, 1) });
expect(openCode.calls.sessionPromptAsync).toHaveLength(2);
expect(openCode.calls.sessionStatus.length).toBeGreaterThanOrEqual(2);
} finally {
@@ -4722,9 +4748,9 @@ describe("OpenCode provider subagent contract", () => {
"turn_completed",
]);
expect(events.map((event) => ("turnId" in event ? event.turnId : undefined))).toEqual([
"opencode-turn-0",
"opencode-turn-0",
"opencode-turn-0",
turnIdAt(events, 0),
turnIdAt(events, 0),
turnIdAt(events, 0),
]);
expect(events).toContainEqual(
expect.objectContaining({
@@ -4760,7 +4786,7 @@ describe("OpenCode provider subagent contract", () => {
expect(events).toContainEqual({
type: "turn_started",
provider: "opencode",
turnId: "opencode-turn-0",
turnId: turnIdAt(events, 0),
});
});
for (const event of assistantTurnEvents({
@@ -4774,9 +4800,9 @@ describe("OpenCode provider subagent contract", () => {
expect(childClient.calls.sessionStatus).toEqual([{ directory: "/workspace/repo" }]);
expect(turnEventSignatures(events)).toEqual([
["turn_started", "opencode-turn-0"],
["timeline", "opencode-turn-0"],
["turn_completed", "opencode-turn-0"],
["turn_started", turnIdAt(events, 0)],
["timeline", turnIdAt(events, 0)],
["turn_completed", turnIdAt(events, 0)],
]);
expect(events).toContainEqual(
expect.objectContaining({
@@ -4881,9 +4907,9 @@ describe("OpenCode provider subagent contract", () => {
await vi.waitFor(() => {
expect(turnEventSignatures(events)).toEqual([
["turn_started", "opencode-turn-0"],
["timeline", "opencode-turn-0"],
["turn_completed", "opencode-turn-0"],
["turn_started", turnIdAt(events, 0)],
["timeline", turnIdAt(events, 0)],
["turn_completed", turnIdAt(events, 0)],
]);
});
expect(events).toContainEqual(
@@ -4927,9 +4953,9 @@ describe("OpenCode provider subagent contract", () => {
await vi.waitFor(() => {
expect(turnEventSignatures(events)).toEqual([
["turn_started", "opencode-turn-0"],
["timeline", "opencode-turn-0"],
["turn_completed", "opencode-turn-0"],
["turn_started", turnIdAt(events, 0)],
["timeline", turnIdAt(events, 0)],
["turn_completed", turnIdAt(events, 0)],
]);
});
} finally {
@@ -5003,7 +5029,7 @@ describe("OpenCode provider subagent contract", () => {
});
await vi.waitFor(() => {
expect(events).toEqual([
{ type: "turn_started", provider: "opencode", turnId: "opencode-turn-0" },
{ type: "turn_started", provider: "opencode", turnId: turnIdAt(events, 0) },
]);
});
@@ -5118,12 +5144,12 @@ describe("OpenCode provider subagent contract", () => {
await streamDrained.promise;
expect(events.filter((event) => event.type !== "provider_subagent")).toEqual([
{ type: "turn_started", provider: "opencode", turnId: "opencode-turn-0" },
{ type: "turn_started", provider: "opencode", turnId: turnIdAt(events, 0) },
{
type: "turn_canceled",
provider: "opencode",
reason: "interrupted",
turnId: "opencode-turn-0",
turnId: turnIdAt(events, 0),
},
]);
} finally {
@@ -5174,7 +5200,7 @@ describe("OpenCode provider subagent contract", () => {
expect.objectContaining({
type: "permission_requested",
request: expect.objectContaining({ id: "perm_child_external" }),
turnId: "opencode-turn-0",
turnId: turnIdAt(events, 0),
}),
);
});
@@ -3409,6 +3409,7 @@ class OpenCodeAgentSession implements AgentSession {
private partTypes = new Map<string, string>();
private availableModesCache: AgentMode[] | null = null;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private readonly turnNamespace = randomUUID();
private nextTurnOrdinal = 0;
private turnState: OpenCodeTurnState = { status: "idle" };
/**
@@ -4886,7 +4887,7 @@ class OpenCodeAgentSession implements AgentSession {
private createTurnId(): string {
this.gapRepairRevision += 1;
return `opencode-turn-${this.nextTurnOrdinal++}`;
return `opencode-turn-${this.turnNamespace}-${this.nextTurnOrdinal++}`;
}
private traceOpenCode(msg: OpenCodeTraceMessage, data: OpenCodeTraceData = {}): void {
+1 -1
View File
@@ -94,7 +94,7 @@ type PluginTurnOutcome =
| Field | Shape / meaning |
| ------------ | -------------------------------------------------------------------------------------- |
| `turnId` | Provider-reported `string` or `null`; can repeat after a session reopens |
| `turnId` | Provider-reported `string` or `null`; unique across resumed sessions |
| `timeline` | Deprecated, always `[]`; fetch history with the SDK. |
| `request` | SDK `AgentPermissionRequest`; `kind` is `tool`, `plan`, `question`, `mode`, or `other` |
| `resolution` | SDK `AgentPermissionResponse` |
+1 -1
View File
@@ -92,7 +92,7 @@ type PluginTurnOutcome =
| 字段 | 结构 / 含义 |
| ------------ | ---------------------------------------------------------------------------------------- |
| `turnId` | Provider 报告的 `string` 或 `null`;会话重新打开后可能重复 |
| `turnId` | Provider 报告的 `string` 或 `null`;在会话恢复后仍保持唯一 |
| `timeline` | 已弃用,始终为 `[]`;通过 SDK 获取历史。 |
| `request` | SDK 的 `AgentPermissionRequest`;`kind` 为 `tool`、`plan`、`question`、`mode` 或 `other` |
| `resolution` | SDK 的 `AgentPermissionResponse` |