diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 4a1c6e791..6ef4c77fa 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -351,6 +351,150 @@ describe("AgentManager", () => { expect(afterHydrate).toEqual(beforeReload); }); + test("reloadAgentSession cancels active run and resumes existing session once thread_started is observed", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-active-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + + class DelayedPersistenceSession extends TestAgentSession { + private persistenceReady = false; + private interrupted = false; + private releaseGate: (() => void) | null = null; + private readonly gate = new Promise((resolve) => { + this.releaseGate = resolve; + }); + + constructor( + config: AgentSessionConfig, + private readonly stableSessionId: string, + initiallyReady = false + ) { + super(config); + this.persistenceReady = initiallyReady; + } + + async *stream(): AsyncGenerator { + yield { type: "turn_started", provider: this.provider }; + this.persistenceReady = true; + yield { + type: "thread_started", + provider: this.provider, + sessionId: this.stableSessionId, + }; + await this.gate; + if (this.interrupted) { + yield { type: "turn_canceled", provider: this.provider, reason: "Interrupted" }; + return; + } + yield { type: "turn_completed", provider: this.provider }; + } + + async getRuntimeInfo() { + return { + provider: this.provider, + sessionId: this.persistenceReady ? this.stableSessionId : null, + model: null, + modeId: null, + }; + } + + describePersistence() { + if (!this.persistenceReady) { + return null; + } + return { + provider: this.provider, + sessionId: this.stableSessionId, + }; + } + + async interrupt(): Promise { + this.interrupted = true; + this.releaseGate?.(); + } + + async close(): Promise { + this.interrupted = true; + this.releaseGate?.(); + } + } + + class DelayedPersistenceClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities = TEST_CAPABILITIES; + createSessionCalls = 0; + resumeSessionCalls = 0; + private nextSessionNumber = 1; + + async isAvailable(): Promise { + return true; + } + + async createSession(config: AgentSessionConfig): Promise { + const sessionId = `delayed-session-${this.nextSessionNumber++}`; + this.createSessionCalls += 1; + return new DelayedPersistenceSession(config, sessionId); + } + + async resumeSession( + handle: AgentPersistenceHandle, + overrides?: Partial + ): Promise { + this.resumeSessionCalls += 1; + const metadata = (handle.metadata ?? {}) as Partial; + const merged: AgentSessionConfig = { + ...metadata, + ...overrides, + provider: "codex", + cwd: overrides?.cwd ?? metadata.cwd ?? process.cwd(), + }; + return new DelayedPersistenceSession(merged, handle.sessionId, true); + } + } + + const client = new DelayedPersistenceClient(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000114", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + expect(snapshot.persistence).toBeNull(); + + const stream = manager.streamAgent(snapshot.id, "hello"); + const first = await stream.next(); + expect(first.done).toBe(false); + expect(first.value?.type).toBe("turn_started"); + const second = await stream.next(); + expect(second.done).toBe(false); + expect(second.value?.type).toBe("thread_started"); + + const active = manager.getAgent(snapshot.id); + expect(active?.lifecycle).toBe("running"); + expect(active?.persistence?.sessionId).toBe("delayed-session-1"); + + const reloaded = await manager.reloadAgentSession(snapshot.id, { + systemPrompt: "voice mode on", + }); + + expect(client.createSessionCalls).toBe(1); + expect(client.resumeSessionCalls).toBe(1); + expect(reloaded.persistence?.sessionId).toBe("delayed-session-1"); + + // Drain stream after cancellation to ensure clean shutdown. + while (true) { + const next = await stream.next(); + if (next.done) { + break; + } + } + }); + test("fetchTimeline returns full timeline with reset when cursor epoch is stale", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-stale-")); 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 b6849fc3c..958abd824 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -666,7 +666,11 @@ export class AgentManager { agentId: string, overrides?: Partial ): Promise { - const existing = this.requireAgent(agentId); + let existing = this.requireAgent(agentId); + if (existing.lifecycle === "running" || existing.pendingRun) { + await this.cancelAgentRun(agentId); + existing = this.requireAgent(agentId); + } const timelineState = this.ensureTimelineState(existing); const preservedTimeline = [...existing.timeline]; const preservedTimelineRows = timelineState.rows.map((row) => ({ ...row })); @@ -1561,9 +1565,13 @@ export class AgentManager { // Update persistence with the new session ID from the provider. // persistence.sessionId is the single source of truth for session identity. { + const previousSessionId = agent.persistence?.sessionId ?? null; const handle = agent.session.describePersistence(); if (handle) { agent.persistence = attachPersistenceCwd(handle, agent.cwd); + if (agent.persistence?.sessionId !== previousSessionId) { + this.emitState(agent); + } } } break; diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index 3162a09bf..7bd4ae29d 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -1428,7 +1428,14 @@ class ClaudeAgentSession implements AgentSession { switch (message.type) { case "system": if (message.subtype === "init") { - this.handleSystemMessage(message); + const threadSessionId = this.handleSystemMessage(message); + if (threadSessionId) { + events.push({ + type: "thread_started", + provider: "claude", + sessionId: threadSessionId, + }); + } } else if (message.subtype === "status") { const status = (message as { status?: string }).status; if (status === "compacting") { @@ -1513,9 +1520,9 @@ class ClaudeAgentSession implements AgentSession { return events; } - private handleSystemMessage(message: SDKSystemMessage): void { + private handleSystemMessage(message: SDKSystemMessage): string | null { if (message.subtype !== "init") { - return; + return null; } const msg = message as unknown as { @@ -1533,13 +1540,15 @@ class ClaudeAgentSession implements AgentSession { : ""; const newSessionId = newSessionIdRaw.trim(); if (!newSessionId) { - return; + return null; } const existingSessionId = this.claudeSessionId; + let threadStartedSessionId: string | null = null; if (existingSessionId === null) { // First time setting session ID (empty → filled) - this is expected this.claudeSessionId = newSessionId; + threadStartedSessionId = newSessionId; this.logger.debug( { sessionId: newSessionId }, "Claude session ID set for the first time" @@ -1578,6 +1587,7 @@ class ClaudeAgentSession implements AgentSession { // Invalidate cached runtime info so it picks up the new model this.cachedRuntimeInfo = null; } + return threadStartedSessionId; } private convertUsage(message: SDKResultMessage): AgentUsage | undefined {