fix(server): preserve claude session identity on agent reload
This commit is contained in:
@@ -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<void>((resolve) => {
|
||||
this.releaseGate = resolve;
|
||||
});
|
||||
|
||||
constructor(
|
||||
config: AgentSessionConfig,
|
||||
private readonly stableSessionId: string,
|
||||
initiallyReady = false
|
||||
) {
|
||||
super(config);
|
||||
this.persistenceReady = initiallyReady;
|
||||
}
|
||||
|
||||
async *stream(): AsyncGenerator<AgentStreamEvent> {
|
||||
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<void> {
|
||||
this.interrupted = true;
|
||||
this.releaseGate?.();
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
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<boolean> {
|
||||
return true;
|
||||
}
|
||||
|
||||
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
const sessionId = `delayed-session-${this.nextSessionNumber++}`;
|
||||
this.createSessionCalls += 1;
|
||||
return new DelayedPersistenceSession(config, sessionId);
|
||||
}
|
||||
|
||||
async resumeSession(
|
||||
handle: AgentPersistenceHandle,
|
||||
overrides?: Partial<AgentSessionConfig>
|
||||
): Promise<AgentSession> {
|
||||
this.resumeSessionCalls += 1;
|
||||
const metadata = (handle.metadata ?? {}) as Partial<AgentSessionConfig>;
|
||||
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");
|
||||
|
||||
@@ -666,7 +666,11 @@ export class AgentManager {
|
||||
agentId: string,
|
||||
overrides?: Partial<AgentSessionConfig>
|
||||
): Promise<ManagedAgent> {
|
||||
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;
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user