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

WIP: fix archive tab reconciliation (#158)

Browse Source 
* WIP: fix archive tab reconciliation

* fix: converge archive into AgentManager for cross-session propagation

CLI archive left agent tabs visible in passive app clients because
Session.archiveAgentState only notified the archiving session's own
subscription. LoopService had the same gap via AgentManager.archiveAgent.

Converge on AgentManager.archiveAgent as the single canonical archive
path: persist updatedAt, call notifyAgentState before closeAgent so all
sessions receive the archived snapshot. Delete Session.archiveAgentState
and make handleArchiveAgentRequest a thin wrapper.
This commit is contained in:
Mohamed Boudra
2026-03-31 16:03:36 +07:00
committed by GitHub
parent 4b93f990d2
commit ee50d3b8d0
9 changed files with 716 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-manager.test.ts
+44
View File
@@ -2263,6 +2263,50 @@ describe("AgentManager", () => {
expect(attentionReasons).toEqual(["error", "error"]);
});
test("archiveAgent persists archivedAt and updatedAt before emitting closed state", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-archive-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const manager = new AgentManager({
clients: {
codex: new TestAgentClient(),
},
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000131",
});
const agent = await manager.createAgent({
provider: "codex",
cwd: workdir,
title: "Archive target",
});
const lifecycles: string[] = [];
manager.subscribe(
(event) => {
if (event.type === "agent_state" && event.agent.id === agent.id) {
lifecycles.push(event.agent.lifecycle);
}
},
{ agentId: agent.id, replayState: false },
);
const { archivedAt } = await manager.archiveAgent(agent.id);
const stored = await storage.get(agent.id);
expect(stored).toMatchObject({
id: agent.id,
archivedAt,
updatedAt: archivedAt,
lastStatus: "idle",
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
});
expect(lifecycles.slice(-2)).toEqual(["idle", "closed"]);
});
test("turn_failed emits a system error assistant timeline message and keeps error lifecycle", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-turn-failed-"));
const storagePath = join(workdir, "agents");
packages/server/src/server/agent/agent-manager.ts
+2
View File
@@ -883,11 +883,13 @@ export class AgentManager {
await this.registry.upsert({
...stored,
archivedAt,
updatedAt: archivedAt,
lastStatus: normalizedStatus,
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
});
this.notifyAgentState(agentId);
await this.closeAgent(agentId);
return { archivedAt };
packages/server/src/server/session.ts
+64 -58
View File
@@ -979,6 +979,16 @@ export class Session {
const storedRecord = await this.agentStorage.get(agent.id);
const title = storedRecord?.title ?? storedRecord?.config?.title ?? null;
const payload = toAgentPayload(agent, { title });
const storedUpdatedAt = storedRecord
? this.resolveStoredAgentPayloadUpdatedAt(storedRecord)
: null;
if (storedUpdatedAt) {
const liveUpdatedAt = Date.parse(payload.updatedAt);
const persistedUpdatedAt = Date.parse(storedUpdatedAt);
if (!Number.isNaN(persistedUpdatedAt) && (Number.isNaN(liveUpdatedAt) || persistedUpdatedAt > liveUpdatedAt)) {
payload.updatedAt = storedUpdatedAt;
}
}
payload.archivedAt = storedRecord?.archivedAt ?? null;
return payload;
}
@@ -994,7 +1004,7 @@ export class Session {
} as const;
const createdAt = new Date(record.createdAt);
const updatedAt = new Date(record.lastActivityAt ?? record.updatedAt);
const updatedAt = new Date(this.resolveStoredAgentPayloadUpdatedAt(record));
const lastUserMessageAt = record.lastUserMessageAt ? new Date(record.lastUserMessageAt) : null;
const provider = coerceAgentProvider(this.sessionLogger, record.provider, record.id);
@@ -1045,6 +1055,23 @@ export class Session {
};
}
private resolveStoredAgentPayloadUpdatedAt(record: StoredAgentRecord): string {
const timestamps = [record.updatedAt, record.lastActivityAt]
.filter((value): value is string => typeof value === "string" && value.length > 0)
.map((value) => ({
raw: value,
parsed: Date.parse(value),
}))
.filter((value) => !Number.isNaN(value.parsed));
if (timestamps.length === 0) {
return record.updatedAt;
}
timestamps.sort((a, b) => b.parsed - a.parsed);
return timestamps[0].raw;
}
private async ensureAgentLoaded(agentId: string): Promise<ManagedAgent> {
const existing = this.agentManager.getAgent(agentId);
if (existing) {
@@ -1961,7 +1988,40 @@ export class Session {
private async handleArchiveAgentRequest(agentId: string, requestId: string): Promise<void> {
this.sessionLogger.info({ agentId }, `Archiving agent ${agentId}`);
const { archivedAt } = await this.archiveAgentState(agentId);
if (this.agentManager.getAgent(agentId)) {
await this.interruptAgentIfRunning(agentId);
await this.agentManager.clearAgentAttention(agentId).catch(() => undefined);
}
const { archivedAt } = await this.agentManager.archiveAgent(agentId);
const archivedRecord = await this.agentStorage.get(agentId);
if (!archivedRecord) {
throw new Error(`Agent not found in storage after archive: ${agentId}`);
}
if (this.agentUpdatesSubscription) {
const payload = this.buildStoredAgentPayload(archivedRecord);
const project = await this.buildProjectPlacement(payload.cwd);
const matches = this.matchesAgentFilter({
agent: payload,
project,
filter: this.agentUpdatesSubscription.filter,
});
this.bufferOrEmitAgentUpdate(
this.agentUpdatesSubscription,
matches
? {
kind: "upsert",
agent: payload,
project,
}
: {
kind: "remove",
agentId,
},
);
await this.emitWorkspaceUpdateForCwd(payload.cwd);
}
this.emit({
type: "agent_archived",
@@ -1973,70 +2033,16 @@ export class Session {
});
}
private async archiveAgentState(agentId: string): Promise<{
archivedAt: string;
archivedRecord: StoredAgentRecord;
}> {
if (this.agentManager.getAgent(agentId)) {
await this.interruptAgentIfRunning(agentId);
await this.agentManager.clearAgentAttention(agentId).catch(() => undefined);
}
const archivedAt = new Date().toISOString();
const existing = await this.agentStorage.get(agentId);
let archivedRecord: StoredAgentRecord | null = existing;
if (!archivedRecord) {
const liveAgent = this.agentManager.getAgent(agentId);
if (!liveAgent) {
throw new Error(`Agent not found: ${agentId}`);
}
await this.agentStorage.applySnapshot(liveAgent, {
internal: liveAgent.internal,
});
archivedRecord = await this.agentStorage.get(agentId);
if (!archivedRecord) {
throw new Error(`Agent not found in storage after snapshot: ${agentId}`);
}
}
const normalizedStatus =
archivedRecord.lastStatus === "running" || archivedRecord.lastStatus === "initializing"
? "idle"
: archivedRecord.lastStatus;
const nextRecord: StoredAgentRecord = {
...archivedRecord,
archivedAt,
lastStatus: normalizedStatus,
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
};
await this.agentStorage.upsert(nextRecord);
// Unload the agent from memory — the storage record is the source of truth now.
// This tears down the provider session and drops the hydrated timeline,
// freeing memory. ensureAgentLoaded will re-initialize if needed later.
if (this.agentManager.getAgent(agentId)) {
try {
await this.agentManager.closeAgent(agentId);
} catch (error) {
this.sessionLogger.warn({ err: error, agentId }, "Failed to close agent during archive");
}
}
return { archivedAt, archivedRecord: nextRecord };
}
private async unarchiveAgentState(agentId: string): Promise<boolean> {
const record = await this.agentStorage.get(agentId);
if (!record || !record.archivedAt) {
return false;
}
const updatedAt = new Date().toISOString();
await this.agentStorage.upsert({
...record,
archivedAt: null,
updatedAt,
});
this.agentManager.notifyAgentState(agentId);
return true;
packages/server/src/server/session.workspaces.test.ts
+142
View File
@@ -82,6 +82,9 @@ function createSessionForWorkspaceTests(): Session {
subscribe: () => () => {},
listAgents: () => [],
getAgent: () => null,
archiveAgent: async () => ({ archivedAt: new Date().toISOString() }),
clearAgentAttention: async () => {},
notifyAgentState: () => {},
} as any,
agentStorage: {
list: async () => [],
@@ -130,6 +133,145 @@ function createSessionForWorkspaceTests(): Session {
}
describe("workspace aggregation", () => {
test("archive emits an authoritative agent_update upsert for subscribed clients", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
const archivedRecord = {
id: "agent-1",
provider: "codex",
cwd: "/tmp/repo",
createdAt: "2026-03-30T15:00:00.000Z",
updatedAt: "2026-03-30T15:00:00.000Z",
lastActivityAt: "2026-03-30T15:00:00.000Z",
lastUserMessageAt: null,
lastStatus: "idle",
lastModeId: null,
runtimeInfo: null,
config: {
provider: "codex",
cwd: "/tmp/repo",
},
persistence: null,
title: "Archive me",
labels: {},
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
archivedAt: null,
};
const logger = {
child: () => logger,
trace: vi.fn(),
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
};
const session = new Session({
clientId: "test-client",
onMessage: (message) => emitted.push(message as any),
logger: logger as any,
downloadTokenStore: {} as any,
pushTokenStore: {} as any,
paseoHome: "/tmp/paseo-test",
agentManager: {
subscribe: () => () => {},
listAgents: () => [],
getAgent: () => null,
archiveAgent: async () => {
const archivedAt = new Date().toISOString();
Object.assign(archivedRecord, {
archivedAt,
updatedAt: archivedAt,
});
return { archivedAt };
},
clearAgentAttention: async () => {},
notifyAgentState: () => {},
} as any,
agentStorage: {
list: async () => [archivedRecord],
get: async (agentId: string) => (agentId === archivedRecord.id ? archivedRecord : null),
} as any,
projectRegistry: {
initialize: async () => {},
existsOnDisk: async () => true,
list: async () => [],
get: async () => null,
upsert: async () => {},
archive: async () => {},
remove: async () => {},
} as any,
workspaceRegistry: {
initialize: async () => {},
existsOnDisk: async () => true,
list: async () => [],
get: async () => null,
upsert: async () => {},
archive: async () => {},
remove: async () => {},
} as any,
checkoutDiffManager: {
subscribe: async () => ({
initial: { cwd: "/tmp/repo", files: [], error: null },
unsubscribe: () => {},
}),
scheduleRefreshForCwd: () => {},
getMetrics: () => ({
checkoutDiffTargetCount: 0,
checkoutDiffSubscriptionCount: 0,
checkoutDiffWatcherCount: 0,
checkoutDiffFallbackRefreshTargetCount: 0,
}),
dispose: () => {},
} as any,
createAgentMcpTransport: async () => {
throw new Error("not used");
},
stt: null,
tts: null,
terminalManager: null,
}) as any;
session.agentUpdatesSubscription = {
subscriptionId: "sub-agents",
filter: { includeArchived: true },
isBootstrapping: false,
pendingUpdatesByAgentId: new Map(),
};
session.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: "repo",
checkout: {
cwd,
isGit: false,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
});
await session.handleArchiveAgentRequest("agent-1", "req-archive");
const update = emitted.find((message) => message.type === "agent_update");
expect(update?.payload).toMatchObject({
kind: "upsert",
agent: {
id: "agent-1",
archivedAt: expect.any(String),
},
});
expect(
emitted.find((message) => message.type === "agent_archived")?.payload,
).toMatchObject({
agentId: "agent-1",
archivedAt: expect.any(String),
requestId: "req-archive",
});
});
test("non-git workspace uses deterministic directory name and no unknown branch fallback", async () => {
const session = createSessionForWorkspaceTests() as any;
session.workspaceRegistry.list = async () => [