puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-31 13:42:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(server): share concurrent agent refresh results

Browse Source 
Keep a refresh in flight until provider history has been hydrated so concurrent
requests receive the same reload outcome across connections.
This commit is contained in:
tieo
2026-10-31 10:10:21 +00:00
committed by paseo-bot[bot]
parent fd65985e9d
commit 39f1058005
2 changed files with 126 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/session.ts
+55 -31
View File
@@ -682,6 +682,9 @@ interface ClientActivity {
appVisibilityChangedAt: Date;
}
// The agent refreshes in flight on this daemon, by agent id; see handleRefreshAgentRequest.
const pendingAgentRefreshes = new Map<string, Promise<ManagedAgent>>();
export class Session {
private readonly connectedHostSession: ConnectedHostSession | null;
private routed = false;
@@ -4758,38 +4761,23 @@ export class Session {
this.sessionLogger.info({ agentId }, `Refreshing agent ${agentId} from persistence`);
try {
await this.restoreOwningWorkspaceForLegacyAgentRefresh(agentId);
await unarchiveAgentState(this.agentStorage, this.agentManager, agentId);
let snapshot: ManagedAgent;
const existing = this.agentManager.getAgent(agentId);
if (existing) {
await this.interruptAgentIfRunning(agentId);
snapshot = await this.agentManager.reloadAgentSession(agentId, undefined, {
rehydrateFromDisk: true,
});
} else {
const record = await this.agentStorage.get(agentId);
if (!record) {
throw new Error(`Agent not found: ${agentId}`);
}
const registeredProviderIds = this.providerSnapshotManager.listRegisteredProviderIds();
if (!isStoredAgentProviderAvailable(record, registeredProviderIds)) {
throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`);
}
if (!toAgentPersistenceHandle(registeredProviderIds, record.persistence)) {
throw new Error(`Agent ${agentId} cannot be refreshed because it lacks persistence`);
}
// 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,
});
// A refresh takes the agent out of the manager while it resumes the provider session, so a
// second refresh arriving meanwhile, from any connection, found no agent and failed with
// "Unknown agent". A client that gives up waiting and retries makes exactly that second
// request. It waits for the refresh in flight and answers with its outcome instead.
let refresh = pendingAgentRefreshes.get(agentId);
if (!refresh) {
const started = this.refreshAgentFromPersistence(agentId);
refresh = started;
pendingAgentRefreshes.set(agentId, started);
void started
.catch(() => undefined)
.finally(() => {
if (pendingAgentRefreshes.get(agentId) === started)
pendingAgentRefreshes.delete(agentId);
});
}
await this.agentManager.hydrateTimelineFromProvider(agentId, { broadcast: true });
const snapshot = await refresh;
await this.agentUpdates.forwardLiveAgent(snapshot);
const timelineSize = this.agentManager.getTimeline(agentId).length;
if (requestId) {
@@ -4829,6 +4817,42 @@ export class Session {
}
}
private async refreshAgentFromPersistence(agentId: string): Promise<ManagedAgent> {
await this.restoreOwningWorkspaceForLegacyAgentRefresh(agentId);
await unarchiveAgentState(this.agentStorage, this.agentManager, agentId);
let snapshot: ManagedAgent;
const existing = this.agentManager.getAgent(agentId);
if (existing) {
await this.interruptAgentIfRunning(agentId);
snapshot = await this.agentManager.reloadAgentSession(agentId, undefined, {
rehydrateFromDisk: true,
});
} else {
const record = await this.agentStorage.get(agentId);
if (!record) {
throw new Error(`Agent not found: ${agentId}`);
}
const registeredProviderIds = this.providerSnapshotManager.listRegisteredProviderIds();
if (!isStoredAgentProviderAvailable(record, registeredProviderIds)) {
throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`);
}
if (!toAgentPersistenceHandle(registeredProviderIds, record.persistence)) {
throw new Error(`Agent ${agentId} cannot be refreshed because it lacks persistence`);
}
// 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 });
return snapshot;
}
private async handleCancelAgentRequest(agentId: string, requestId?: string): Promise<void> {
this.sessionLogger.info({ agentId }, `Cancel request received for agent ${agentId}`);
packages/server/src/server/session.workspaces.test.ts
+71
View File
@@ -5426,6 +5426,77 @@ test("refresh_agent_request leaves workspace archival independent when its direc
expect(findByType(emitted, "rpc_error")).toBeUndefined();
});
test("refresh_agent_request arriving while a refresh of the same agent resumes waits for it", async () => {
// A client gives up on a slow refresh after its request timeout and asks again. The refresh in
// flight has taken the agent out of the manager, so a second one used to fail with "Unknown
// agent", and every retry after it did the same while the first was still resuming.
const emitted: SessionOutboundMessage[] = [];
const session = createSessionForWorkspaceTests({
onMessage: (message) => {
if (isSessionOutboundMessage(message)) emitted.push(message);
},
});
const agentId = "agent-refreshed-twice";
const cwd = path.resolve("/tmp/paseo-refresh-twice");
session.filesystem.isDirectory = async () => true;
session.agentStorage.get = async (id: string) =>
id === agentId
? makeStoredAgent({ id: agentId, cwd, updatedAt: "2026-03-10T00:00:00.000Z" })
: null;
session.agentStorage.upsert = async () => {};
const managed = makeManagedAgent({
id: agentId,
cwd,
lifecycle: "idle",
updatedAt: "2026-03-10T00:00:00.000Z",
});
let loaded: ManagedAgent | null = managed;
let releaseResume!: () => void;
const resumed = new Promise<void>((resolve) => {
releaseResume = resolve;
});
let reloads = 0;
session.agentManager.getAgent = () => loaded;
session.interruptAgentIfRunning = async () => undefined;
session.agentManager.reloadAgentSession = async () => {
reloads += 1;
loaded = null;
await resumed;
loaded = managed;
return managed;
};
session.agentManager.hydrateTimelineFromProvider = async () => {
if (!loaded) throw new Error(`Unknown agent '${agentId}'`);
};
session.agentManager.getTimeline = () => [];
session.agentUpdates.forwardLiveAgent = async () => undefined;
const first = session.handleMessage({
type: "refresh_agent_request",
agentId,
requestId: "req-first",
});
await waitForImmediate();
const retry = session.handleMessage({
type: "refresh_agent_request",
agentId,
requestId: "req-retry",
});
await waitForImmediate();
releaseResume();
await Promise.all([first, retry]);
expect(reloads).toBe(1);
expect(findByType(emitted, "rpc_error")).toBeUndefined();
const refreshed = emitted.filter(
(m) => m.type === "status" && (m.payload as { status?: string }).status === "agent_refreshed",
);
expect(refreshed.map((m) => (m.payload as { requestId?: string }).requestId).sort()).toEqual([
"req-first",
"req-retry",
]);
});
test("refresh_agent_request leaves workspace archival independent when its directory is missing", async () => {
const emitted: SessionOutboundMessage[] = [];
const session = createSessionForWorkspaceTests({