fix: clear stale workspace activity

This commit is contained in:
Federico Jaramillo Martinez
2026-05-19 21:41:58 +02:00
parent 1232610bf1
commit 619840a398
5 changed files with 111 additions and 9 deletions
@@ -0,0 +1,5 @@
---
"@jmfederico/pi-web": patch
---
Clear stale workspace activity indicators when sessions become idle or all remaining sessions are archived.
@@ -32,6 +32,47 @@ describe("WorkspaceActivityService", () => {
expect(events.at(-1)).toMatchObject({ type: "workspace.activity", activity: { cwd: "/repo", hasSessionActivity: false, hasTerminalActivity: false } });
});
it("clears stale active activity when an idle status arrives", () => {
const events: RealtimeEvent[] = [];
const service = new WorkspaceActivityService({ publishRealtime: (event) => events.push(event) });
service.applySessionActivity("/repo", { sessionId: "s1", phase: "active", label: "running tool", detail: "read", at: "now" });
service.applySessionStatus("/repo", status({ isStreaming: false }));
expect(service.snapshot().workspaces).toEqual([]);
expect(events.at(-1)).toMatchObject({ type: "workspace.activity", activity: { cwd: "/repo", hasSessionActivity: false, hasTerminalActivity: false } });
});
it("publishes a clear event when removing an already-pruned session with a cwd", () => {
const events: RealtimeEvent[] = [];
const service = new WorkspaceActivityService({ publishRealtime: (event) => events.push(event) });
service.removeSession("missing-session", "/repo");
expect(events.at(-1)).toMatchObject({ type: "workspace.activity", activity: { cwd: "/repo", hasSessionActivity: false, hasTerminalActivity: false } });
});
it("reconciles stale session activity for a workspace", () => {
const events: RealtimeEvent[] = [];
const service = new WorkspaceActivityService({ publishRealtime: (event) => events.push(event) });
service.applySessionActivity("/repo", { sessionId: "s1", phase: "active", label: "running tool", at: "now" });
service.applySessionActivity("/repo", { sessionId: "s2", phase: "active", label: "running tool", at: "now" });
service.applySessionActivity("/other", { sessionId: "s3", phase: "active", label: "running tool", at: "now" });
service.reconcileSessionActivity("/repo", ["s2"]);
expect(service.snapshot().workspaces).toMatchObject([
{ cwd: "/other", hasSessionActivity: true, hasTerminalActivity: false },
{ cwd: "/repo", hasSessionActivity: true, hasTerminalActivity: false },
]);
service.reconcileSessionActivity("/repo", []);
expect(service.snapshot().workspaces).toMatchObject([{ cwd: "/other", hasSessionActivity: true, hasTerminalActivity: false }]);
expect(events.at(-1)).toMatchObject({ type: "workspace.activity", activity: { cwd: "/repo", hasSessionActivity: false, hasTerminalActivity: false } });
});
it("combines sessions and terminals and clears closed terminals", () => {
const events: RealtimeEvent[] = [];
const service = new WorkspaceActivityService({ publishRealtime: (event) => events.push(event) });
@@ -26,6 +26,7 @@ export class WorkspaceActivityService {
const record = this.sessions.get(status.sessionId) ?? { cwd };
record.cwd = cwd;
record.status = status;
if (!isSessionActive(status) && record.activity?.phase === "active") delete record.activity;
this.sessions.set(status.sessionId, record);
this.pruneIdleSession(status.sessionId);
this.publishChangedCwds(previousCwd, cwd);
@@ -41,10 +42,21 @@ export class WorkspaceActivityService {
this.publishChangedCwds(previousCwd, cwd);
}
removeSession(sessionId: string): void {
const cwd = this.sessions.get(sessionId)?.cwd;
removeSession(sessionId: string, cwd?: string): void {
const previousCwd = this.sessions.get(sessionId)?.cwd ?? cwd;
this.sessions.delete(sessionId);
this.publishCwd(cwd);
this.publishCwd(previousCwd);
}
reconcileSessionActivity(cwd: string, sessionIds: Iterable<string>): void {
const knownSessionIds = new Set(sessionIds);
let changed = false;
for (const [sessionId, record] of this.sessions.entries()) {
if (record.cwd !== cwd || knownSessionIds.has(sessionId)) continue;
this.sessions.delete(sessionId);
changed = true;
}
if (changed) this.publishCwd(cwd);
}
updateTerminal(terminal: Pick<TerminalInfo, "id" | "cwd" | "exited">): void {
@@ -206,6 +206,40 @@ describe("PiSessionService", () => {
await service.dispose();
});
it("reconciles workspace activity when listing only archived sessions", async () => {
const reconciliations: { cwd: string; sessionIds: string[] }[] = [];
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([{ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", originalPath: "/sessions/archived.jsonl", archivePath: "/archive/archived.jsonl", created: "2026-01-01T00:00:00.000Z", modified: "2026-01-01T00:01:00.000Z", messageCount: 2, firstMessage: "bye" }]),
get: () => Promise.resolve(undefined),
archive: () => { throw new Error("archive should not be called for moved records"); },
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([]),
listAll: () => Promise.resolve([]),
open: () => fakeSessionManager(),
},
workspaceActivity: {
applySessionStatus: () => undefined,
applySessionActivity: () => undefined,
removeSession: () => undefined,
reconcileSessionActivity: (cwd, sessionIds) => { reconciliations.push({ cwd, sessionIds: [...sessionIds] }); },
},
heartbeatIntervalMs: 60_000,
});
const sessions = await service.list("/workspace");
expect(sessions).toHaveLength(1);
expect(sessions[0]).toMatchObject({ id: "archived", archived: true });
expect(reconciliations).toEqual([{ cwd: "/workspace", sessionIds: [] }]);
await service.dispose();
});
it("sends prompts to an injected runtime without touching the SDK runtime", async () => {
const fake = fakeRuntime("prompt-session");
const service = new PiSessionService(new CapturingSessionEventHub(), {
+16 -6
View File
@@ -164,7 +164,7 @@ export interface PiSessionServiceDependencies {
createAgentRuntime?: CreateAgentRuntime;
modelRegistry?: ModelRegistryInstance;
heartbeatIntervalMs?: number;
workspaceActivity?: Pick<WorkspaceActivityService, "applySessionStatus" | "applySessionActivity" | "removeSession">;
workspaceActivity?: Pick<WorkspaceActivityService, "applySessionStatus" | "applySessionActivity" | "removeSession" | "reconcileSessionActivity">;
}
export class PiSessionService {
@@ -179,7 +179,7 @@ export class PiSessionService {
private readonly createRuntime: CreateAgentSessionRuntimeFactory;
private readonly createAgentRuntime: CreateAgentRuntime;
private readonly modelRegistry: ModelRegistryInstance;
private readonly workspaceActivity: Pick<WorkspaceActivityService, "applySessionStatus" | "applySessionActivity" | "removeSession"> | undefined;
private readonly workspaceActivity: Pick<WorkspaceActivityService, "applySessionStatus" | "applySessionActivity" | "removeSession" | "reconcileSessionActivity"> | undefined;
constructor(private readonly events: SessionEventHub, deps: PiSessionServiceDependencies = {}) {
this.archiveStore = deps.archiveStore ?? new SessionArchiveStore();
@@ -219,7 +219,7 @@ export class PiSessionService {
this.authLossWarnings.clear();
await Promise.all(activeSessions.map(async (active) => {
active.unsubscribe();
this.workspaceActivity?.removeSession(active.runtime.session.sessionId);
this.workspaceActivity?.removeSession(active.runtime.session.sessionId, active.runtime.session.sessionManager.getCwd());
await active.runtime.session.abort();
await active.runtime.dispose();
}));
@@ -234,12 +234,13 @@ export class PiSessionService {
.map((record) => this.ensureArchivedSessionMoved(record, sessionsById.get(record.sessionId))),
);
const archivedById = new Map(archivedForCwd.map((record) => [record.sessionId, record]));
const activeSessions = sessions.filter((session) => !archivedById.has(session.id)).map(clientSessionFromListEntry);
const unarchivedSessions = sessions.filter((session) => !archivedById.has(session.id)).map(clientSessionFromListEntry);
this.workspaceActivity?.reconcileSessionActivity(cwd, this.reconcilableSessionIds(cwd, unarchivedSessions.map((session) => session.id), archivedById));
const archivedSessions = archivedForCwd
.sort(compareArchivedRecords)
.map((record) => clientSessionFromArchivedRecord(record, sessionsById.get(record.sessionId)))
.filter(isDefined);
return [...activeSessions, ...archivedSessions];
return [...unarchivedSessions, ...archivedSessions];
}
async start(cwd: string): Promise<ClientSession> {
@@ -439,6 +440,15 @@ export class PiSessionService {
});
}
private reconcilableSessionIds(cwd: string, listedSessionIds: string[], archivedById: Map<string, ArchivedSessionRecord>): string[] {
const sessionIds = new Set(listedSessionIds);
for (const active of new Set(this.active.values())) {
const session = active.runtime.session;
if (session.sessionManager.getCwd() === cwd && !archivedById.has(session.sessionId)) sessionIds.add(session.sessionId);
}
return [...sessionIds];
}
private async ensureArchivedSessionMoved(record: ArchivedSessionRecord, session: PiSessionListEntry | undefined): Promise<ArchivedSessionRecord> {
if (session === undefined || this.active.has(record.sessionId)) return record;
try {
@@ -471,7 +481,7 @@ export class PiSessionService {
if (!active) return;
this.active.delete(sessionId);
this.activities.delete(sessionId);
this.workspaceActivity?.removeSession(sessionId);
this.workspaceActivity?.removeSession(sessionId, active.runtime.session.sessionManager.getCwd());
this.clearAuthLossWarningsForSession(sessionId);
clearSessionQueue(active.runtime.session);
active.unsubscribe();