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

Restore archived agents from History (#2316)

Browse Source 
* fix(app): restore archived agents from History

History navigation lost the explicit archived-agent intent during tab reconciliation, and workspace recovery stopped after restoring the workspace. Preserve the selected tab and recover its provider session as one action while serializing provider resume with timeline hydration.

* fix(app): preserve archived agent recovery invariants

* fix(server): preserve timeline hydration call shape

* fix(app): retain agent tabs through lookup

* fix(server): upgrade in-flight timeline broadcast
This commit is contained in:
Mohamed Boudra
2026-07-22 15:03:09 +02:00
committed by GitHub
parent 68993b7ab3
commit 42ee5a5949
22 changed files with 629 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-loading.ts
+27 -7
View File
@@ -11,7 +11,12 @@ import {
toAgentPersistenceHandle,
} from "../persistence-hooks.js";
const pendingAgentInitializations = new Map<string, Promise<ManagedAgent>>();
interface PendingAgentInitialization {
promise: Promise<ManagedAgent>;
options: { broadcastTimeline: boolean };
}
const pendingAgentInitializations = new Map<string, PendingAgentInitialization>();
export type AgentLoaderManager = Pick<
AgentManager,
@@ -27,6 +32,7 @@ export interface EnsureAgentLoadedDeps {
agentManager: AgentLoaderManager;
agentStorage: AgentStorage;
validProviders?: Iterable<AgentProvider>;
broadcastTimeline?: boolean;
logger: Logger;
}
@@ -58,6 +64,13 @@ export async function ensureAgentLoaded(
deps: EnsureAgentLoadedDeps,
): Promise<ManagedAgent> {
await deps.agentManager.waitForAgentClose?.(agentId);
const inflight = pendingAgentInitializations.get(agentId);
if (inflight) {
inflight.options.broadcastTimeline ||= deps.broadcastTimeline === true;
return inflight.promise;
}
const existing =
deps.agentManager.touchAgentActivity?.(agentId) ?? deps.agentManager.getAgent(agentId);
if (existing) {
@@ -69,11 +82,15 @@ export async function ensureAgentLoaded(
// before storage-backed resume begins.
await deps.agentManager.waitForAgentClose?.(agentId);
const inflight = pendingAgentInitializations.get(agentId);
if (inflight) {
return inflight;
const laterInflight = pendingAgentInitializations.get(agentId);
if (laterInflight) {
laterInflight.options.broadcastTimeline ||= deps.broadcastTimeline === true;
return laterInflight.promise;
}
const pendingOptions = {
broadcastTimeline: deps.broadcastTimeline === true,
};
const initPromise = (async () => {
const record = await deps.agentStorage.get(agentId);
if (!record) {
@@ -111,17 +128,20 @@ export async function ensureAgentLoaded(
deps.logger.info({ agentId, provider: record.provider }, "Agent created from stored config");
}
await deps.agentManager.hydrateTimelineFromProvider(agentId);
await deps.agentManager.hydrateTimelineFromProvider(agentId, {
broadcast: () => pendingOptions.broadcastTimeline,
});
return deps.agentManager.getAgent(agentId) ?? snapshot;
})();
pendingAgentInitializations.set(agentId, initPromise);
const pending: PendingAgentInitialization = { promise: initPromise, options: pendingOptions };
pendingAgentInitializations.set(agentId, pending);
try {
return await initPromise;
} finally {
const current = pendingAgentInitializations.get(agentId);
if (current === initPromise) {
if (current === pending) {
pendingAgentInitializations.delete(agentId);
}
}
packages/server/src/server/agent/agent-manager.test.ts
+70
View File
@@ -7518,6 +7518,76 @@ test("ensureUnarchivedAgentLoaded fences an archived agent after joining a share
}
});
test("a shared agent load upgrades provider history hydration to broadcast", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-shared-load-broadcast-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const historyStarted = deferred<void>();
const historyAllowed = deferred<void>();
const client = new (class extends TestAgentClient {
override async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
return new (class extends TestAgentSession {
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
historyStarted.resolve();
await historyAllowed.promise;
yield {
type: "timeline",
provider: "codex",
item: { type: "assistant_message", text: "Recovered history" },
};
}
})({ provider: "codex", cwd: config?.cwd ?? workdir });
}
})();
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
try {
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
await manager.collectIdleAgents({
cutoff: new Date(Date.now() + 1_000),
protectedAgentIds: new Set(),
});
await manager.deleteAgentState(agent.id);
const events: AgentManagerEvent[] = [];
manager.subscribe((event) => events.push(event), { agentId: agent.id, replayState: false });
const quietLoad = ensureAgentLoaded(agent.id, {
agentManager: manager,
agentStorage: storage,
logger,
});
await historyStarted.promise;
const broadcastingLoad = ensureAgentLoaded(agent.id, {
agentManager: manager,
agentStorage: storage,
broadcastTimeline: true,
logger,
});
historyAllowed.resolve();
await Promise.all([quietLoad, broadcastingLoad]);
expect(events).toContainEqual(
expect.objectContaining({
type: "agent_stream",
agentId: agent.id,
event: expect.objectContaining({
type: "timeline",
item: { type: "assistant_message", text: "Recovered history" },
}),
}),
);
} finally {
historyAllowed.resolve();
await manager.flush().catch(() => undefined);
await storage.flush().catch(() => undefined);
rmSync(workdir, { recursive: true, force: true });
}
});
test("collectIdleAgents leaves recent, protected, internal, running, and error agents resident", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-eligibility-"));
const client = new (class extends TestAgentClient {
packages/server/src/server/agent/agent-manager.ts
+44 -7
View File
@@ -185,7 +185,7 @@ export interface SubscribeOptions {
interface HydrateTimelineOptions {
force?: boolean;
broadcast?: boolean;
broadcast?: boolean | (() => boolean);
}
export type ImportablePersistedAgentQueryOptions = ListImportableSessionsOptions & {
@@ -3177,12 +3177,17 @@ export class AgentManager {
return;
}
const broadcast = options?.broadcast ?? false;
if (options?.force) {
await this.forceHydrateTimelineFromLegacyProviderHistory(agent, options.broadcast === true);
await this.forceHydrateTimelineFromLegacyProviderHistory(
agent,
typeof broadcast === "function" ? broadcast() : broadcast,
);
return;
}
await this.primeTimelineFromLegacyProviderHistory(agent, options?.broadcast === true);
await this.primeTimelineFromLegacyProviderHistory(agent, broadcast);
}
private async forceHydrateTimelineFromLegacyProviderHistory(
@@ -3239,15 +3244,24 @@ export class AgentManager {
private async primeTimelineFromLegacyProviderHistory(
agent: ActiveManagedAgent,
broadcast: boolean,
broadcast: boolean | (() => boolean),
): Promise<void> {
const deferredBroadcast = typeof broadcast === "function";
const timelineEvents: Array<{
event: Extract<AgentStreamEvent, { type: "timeline" }>;
row: AgentTimelineRow;
}> = [];
const providerSubagentEvents: AgentManagerEvent[] = [];
agent.historyPrimed = true;
try {
for await (const event of agent.session.streamHistory()) {
if (event.type === "provider_subagent") {
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
if (broadcast) {
this.dispatch({ type: "provider_subagent", event: update });
const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update };
if (deferredBroadcast) {
providerSubagentEvents.push(managerEvent);
} else if (broadcast) {
this.dispatch(managerEvent);
}
continue;
}
@@ -3257,15 +3271,38 @@ export class AgentManager {
if (event.item.type === "user_message" && isSystemInjectedEnvelope(event.item.text)) {
continue;
}
this.recordTimeline(
const row = this.recordTimeline(
agent.id,
event.item,
event.timestamp ? { timestamp: event.timestamp } : undefined,
);
if (deferredBroadcast) {
timelineEvents.push({ event, row });
} else if (broadcast) {
this.dispatchStream(agent.id, event, {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
});
}
}
} catch {
// ignore history failures
}
if (typeof broadcast !== "function" || !broadcast()) {
return;
}
for (const event of providerSubagentEvents) {
this.dispatch(event);
}
for (const { event, row } of timelineEvents) {
this.dispatchStream(agent.id, event, {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
});
}
}
private notifyForegroundTurnWaiters(agentId: string, event: AgentStreamEvent): void {
packages/server/src/server/agent/mcp-server.test.ts
+1
View File
@@ -5632,6 +5632,7 @@ describe("agent snapshot MCP serialization", () => {
expect(spies.agentManager.resumeAgentFromPersistence).toHaveBeenCalled();
expect(spies.agentManager.hydrateTimelineFromProvider).toHaveBeenCalledWith(
"archived-activity-agent",
{ broadcast: expect.any(Function) },
);
});
packages/server/src/server/session.ts
+11 -14
View File
@@ -30,12 +30,7 @@ import { CursorError } from "./pagination/cursor.js";
import { SortablePager, type SortSpec } from "./pagination/sortable-pager.js";
import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech/speech-provider.js";
import type { TurnDetectionProvider } from "./speech/turn-detection-provider.js";
import {
buildConfigOverrides,
extractTimestamps,
isStoredAgentProviderAvailable,
toAgentPersistenceHandle,
} from "./persistence-hooks.js";
import { isStoredAgentProviderAvailable, toAgentPersistenceHandle } from "./persistence-hooks.js";
import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent/agent-loading.js";
import {
formatSystemNotificationPrompt,
@@ -3183,16 +3178,18 @@ export class Session {
if (!isStoredAgentProviderAvailable(record, registeredProviderIds)) {
throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`);
}
const handle = toAgentPersistenceHandle(registeredProviderIds, record.persistence);
if (!handle) {
if (!toAgentPersistenceHandle(registeredProviderIds, record.persistence)) {
throw new Error(`Agent ${agentId} cannot be refreshed because it lacks persistence`);
}
snapshot = await this.agentManager.resumeAgentFromPersistence(
handle,
buildConfigOverrides(record),
agentId,
extractTimestamps(record),
);
// Share the loader's per-agent in-flight operation with timeline fetches.
// Unarchiving publishes the record before provider resume finishes, so
// the agent pane can otherwise race this request and resume it twice.
snapshot = await ensureAgentLoaded(agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
broadcastTimeline: true,
logger: this.sessionLogger,
});
}
await this.agentManager.hydrateTimelineFromProvider(agentId, { broadcast: true });
await this.agentUpdates.forwardLiveAgent(snapshot);