Archived
feat(sessions): persist shared unread state
This commit is contained in:
Federico Jaramillo Martinez
@@ -33,7 +33,7 @@ import { deterministicSessionName, fallbackSessionName, generateShortSessionName
|
||||
import { computeEditPreview, type EditPreviewResult } from "./editPreview.js";
|
||||
import { attachmentsToInlineImages, saveAttachmentsToWorkspace } from "./attachmentService.js";
|
||||
import { parsePromptAttachments } from "../../shared/promptAttachments.js";
|
||||
import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH } from "../../shared/apiTypes.js";
|
||||
import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_LIMIT } from "../../shared/apiTypes.js";
|
||||
import type {
|
||||
SavedPromptAttachment,
|
||||
SessionBulkArchiveResponse,
|
||||
@@ -45,6 +45,8 @@ import type {
|
||||
SessionNotificationDismissAllRequest,
|
||||
SessionNotificationDismissRequest,
|
||||
SessionNotificationInboxSnapshot,
|
||||
SessionUnreadAcknowledgeRequest,
|
||||
SessionUnreadCatalogSnapshot,
|
||||
SessionWarning,
|
||||
} from "../../shared/apiTypes.js";
|
||||
import type { SessionRouteLookup, SessionRouteRef, SessionRouteService } from "./sessionService.js";
|
||||
@@ -63,6 +65,7 @@ import {
|
||||
type SessionNotificationMutation,
|
||||
} from "./sessionNotificationStore.js";
|
||||
import { plainTextTheme } from "./plainTextTheme.js";
|
||||
import { SessionUnreadStore, type SessionUnreadMutation } from "./sessionUnreadStore.js";
|
||||
|
||||
/**
|
||||
* Minimal structured-logging seam, shaped like Fastify's logger so sessiond can
|
||||
@@ -74,6 +77,9 @@ export interface PiSessionLogger {
|
||||
}
|
||||
|
||||
const noopLogger: PiSessionLogger = { info() { /* no-op */ } };
|
||||
const DEFAULT_UNREAD_PUBLICATION_RETRY_MS = 1_000;
|
||||
const MAX_UNREAD_PUBLICATION_RETRY_MS = 30_000;
|
||||
const MAX_PENDING_UNREAD_MUTATIONS = SESSION_UNREAD_LIMIT + 1;
|
||||
|
||||
function noop(): void {
|
||||
// Intentionally empty default unsubscribe callback.
|
||||
@@ -648,6 +654,10 @@ export interface PiSessionServiceDependencies {
|
||||
now?: () => Date;
|
||||
/** Daemon-lifetime notification state, injected by sessiond in production. */
|
||||
notificationStore?: SessionNotificationStore;
|
||||
/** Durable daemon-owned unread state; defaults to an in-memory store in tests. */
|
||||
unreadStore?: SessionUnreadStore;
|
||||
/** Initial retry delay for durable unread publication failures. */
|
||||
unreadPublicationRetryDelayMs?: number;
|
||||
}
|
||||
|
||||
export class PiSessionService implements SessionRouteService {
|
||||
@@ -694,6 +704,15 @@ export class PiSessionService implements SessionRouteService {
|
||||
private readonly now: () => Date;
|
||||
private readonly notificationStore: SessionNotificationStore;
|
||||
private readonly notificationGenerationBySession = new WeakMap<PiAgentSession, SessionNotificationGeneration>();
|
||||
private readonly unreadStore: SessionUnreadStore;
|
||||
private readonly unreadPublicationRetryInitialMs: number;
|
||||
private readonly pendingUnreadMutations: SessionUnreadMutation[] = [];
|
||||
private unreadPublication: Promise<void> | undefined;
|
||||
private unreadPublicationFailure: unknown;
|
||||
private unreadPublicationFlushRequested = false;
|
||||
private unreadPublicationRetryTimer: NodeJS.Timeout | undefined;
|
||||
private unreadPublicationRetryDelayMs: number;
|
||||
private unreadPublicationStopped = false;
|
||||
|
||||
constructor(private readonly events: SessionEventHub, deps: PiSessionServiceDependencies) {
|
||||
this.archiveStore = deps.archiveStore ?? new SessionArchiveStore();
|
||||
@@ -704,6 +723,12 @@ export class PiSessionService implements SessionRouteService {
|
||||
this.logger = deps.logger ?? noopLogger;
|
||||
this.now = deps.now ?? (() => new Date());
|
||||
this.notificationStore = deps.notificationStore ?? new SessionNotificationStore();
|
||||
this.unreadStore = deps.unreadStore ?? new SessionUnreadStore();
|
||||
this.unreadPublicationRetryInitialMs = Math.max(
|
||||
0,
|
||||
deps.unreadPublicationRetryDelayMs ?? DEFAULT_UNREAD_PUBLICATION_RETRY_MS,
|
||||
);
|
||||
this.unreadPublicationRetryDelayMs = this.unreadPublicationRetryInitialMs;
|
||||
// Subsessions are a beta capability gated behind their own flag, and they
|
||||
// also require the spawn capability (they share its project-scope resolver).
|
||||
const subsessionsActive = this.spawnTargets !== undefined && deps.subsessionsEnabled === true;
|
||||
@@ -761,6 +786,20 @@ export class PiSessionService implements SessionRouteService {
|
||||
return this.notificationStore.catalogSnapshot();
|
||||
}
|
||||
|
||||
async unreadCatalog(): Promise<SessionUnreadCatalogSnapshot> {
|
||||
await this.publishUnreadMutations([]);
|
||||
return this.unreadStore.durableCatalogSnapshot();
|
||||
}
|
||||
|
||||
async acknowledgeUnread(sessionId: string, request: SessionUnreadAcknowledgeRequest): Promise<SessionUnreadCatalogSnapshot> {
|
||||
const result = this.unreadStore.acknowledge(sessionId, {
|
||||
...request,
|
||||
cwd: canonicalizeStoredCwd(request.cwd),
|
||||
});
|
||||
await this.publishUnreadMutations(result.mutations);
|
||||
return this.unreadStore.durableCatalogSnapshot();
|
||||
}
|
||||
|
||||
notificationInbox(ref: PiSessionRef): SessionNotificationInboxSnapshot {
|
||||
return this.notificationStore.inboxSnapshot(ref.id, canonicalizeStoredCwd(ref.cwd));
|
||||
}
|
||||
@@ -818,6 +857,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
}
|
||||
await this.archiveStoreArchiveMany(readyArchiveInputs);
|
||||
archiveInputs.push(...readyArchiveInputs);
|
||||
await this.forgetUnreadSessions(readyArchiveInputs);
|
||||
|
||||
for (const record of plan.deleteRecords) {
|
||||
if (this.activeSessionHasWork(record.sessionId)) {
|
||||
@@ -830,6 +870,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
await this.ensureArchivedRecordsMoved(readyDeleteRecords);
|
||||
const deletedSessionIds = new Set(await this.archiveStoreDeleteArchivedMany(readyDeleteRecords.map((record) => record.sessionId)));
|
||||
deleteRecords.push(...readyDeleteRecords.filter((record) => deletedSessionIds.has(record.sessionId)));
|
||||
await this.forgetUnreadSessions(deleteRecords);
|
||||
|
||||
return summarizeSessionCleanupExecution({
|
||||
archiveInputs,
|
||||
@@ -841,11 +882,14 @@ export class PiSessionService implements SessionRouteService {
|
||||
}
|
||||
|
||||
async dispose(): Promise<void> {
|
||||
this.unreadPublicationStopped = true;
|
||||
this.clearUnreadPublicationRetry();
|
||||
clearInterval(this.heartbeat);
|
||||
this.clearCompactionDrainTimers();
|
||||
const pendingOpens = this.pendingSessionOpenPromises();
|
||||
if (pendingOpens.length > 0) await Promise.allSettled(pendingOpens);
|
||||
const activeSessions = Array.from(new Set(this.active.values()));
|
||||
for (const active of activeSessions) this.forgetUnreadActivity(active.runtime.session);
|
||||
this.active.clear();
|
||||
this.pendingSessionOpens.clear();
|
||||
this.activities.clear();
|
||||
@@ -867,6 +911,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
await active.runtime.dispose();
|
||||
}
|
||||
}));
|
||||
await this.publishUnreadMutations([]);
|
||||
}
|
||||
|
||||
async list(cwd: string): Promise<ClientSession[]> {
|
||||
@@ -882,7 +927,9 @@ export class PiSessionService implements SessionRouteService {
|
||||
this.publishNotificationMutations(this.notificationStore.clearSession(record.sessionId, "archive-reconcile"));
|
||||
}
|
||||
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 reconcilableSessionIds = this.reconcilableSessionIds(cwd, unarchivedSessions.map((session) => session.id), archivedById);
|
||||
this.workspaceActivity?.reconcileSessionActivity(cwd, reconcilableSessionIds);
|
||||
await this.publishUnreadMutations(this.unreadStore.reconcileCwd(canonicalizeStoredCwd(cwd), reconcilableSessionIds));
|
||||
const archivedSessions = archivedForCwd
|
||||
.sort(compareArchivedRecords)
|
||||
.map((record) => clientSessionFromArchivedRecord(record, sessionsById.get(record.sessionId)))
|
||||
@@ -964,7 +1011,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
...(parentSessionFile === undefined ? {} : { parentSessionFile }),
|
||||
cwd: decision.cwd,
|
||||
};
|
||||
this.registerVerifiedSubsession(link);
|
||||
await this.registerVerifiedSubsession(link);
|
||||
this.persistSubsessionLink(link);
|
||||
this.persistSubsessionChildMarker(input.parentSessionId, created.id);
|
||||
await this.prompt(created.id, input.prompt);
|
||||
@@ -1046,7 +1093,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
return sessionFileMatches(session, link.childSessionFile) ? link : undefined;
|
||||
}
|
||||
|
||||
private registerVerifiedSubsession(link: TrackedSubsessionLink): void {
|
||||
private async registerVerifiedSubsession(link: TrackedSubsessionLink): Promise<void> {
|
||||
const { childSessionId, parentSessionId } = link;
|
||||
const previousParentId = this.subsessionParents.get(childSessionId);
|
||||
if (previousParentId !== undefined && previousParentId !== parentSessionId) {
|
||||
@@ -1062,6 +1109,25 @@ export class PiSessionService implements SessionRouteService {
|
||||
|
||||
this.subsessionLinks.set(childSessionId, link);
|
||||
if (!this.subsessionNotifyArmed.has(childSessionId)) this.subsessionNotifyArmed.set(childSessionId, false);
|
||||
|
||||
const cwd = this.cwdForVerifiedSubsession(link);
|
||||
await this.publishUnreadMutations(this.unreadStore.excludeSession(childSessionId, cwd));
|
||||
}
|
||||
|
||||
private cwdForVerifiedSubsession(link: TrackedSubsessionLink): string {
|
||||
const activeCwd = this.activeChildForSubsessionLink(link)?.runtime.session.sessionManager.getCwd();
|
||||
const linkedCwd = nonEmptyString(activeCwd) ?? nonEmptyString(link.cwd);
|
||||
if (linkedCwd !== undefined) return canonicalizeStoredCwd(linkedCwd);
|
||||
|
||||
const childSessionFile = link.childSessionFile;
|
||||
if (childSessionFile !== undefined) {
|
||||
try {
|
||||
return canonicalizeStoredCwd(this.sessionManager.open(childSessionFile).getCwd());
|
||||
} catch (error: unknown) {
|
||||
throw new Error("Could not resolve cwd for verified tracked sub-session", { cause: error });
|
||||
}
|
||||
}
|
||||
throw new Error("Could not resolve cwd for verified tracked sub-session");
|
||||
}
|
||||
|
||||
private unregisterSubsession(childSessionId: string): void {
|
||||
@@ -1110,39 +1176,45 @@ export class PiSessionService implements SessionRouteService {
|
||||
const activeParent = this.active.get(parentSessionId);
|
||||
if (activeParent !== undefined && (parentSessionFile === undefined || activeSessionFileMatches(activeParent, parentSessionFile))) {
|
||||
const activeParentFile = nonEmptyString(activeParent.runtime.session.sessionFile);
|
||||
await this.registerPersistedSubsessionLinks(parentSessionId, activeParent.runtime.session.sessionManager, activeParentFile);
|
||||
this.subsessionHydratedParents.add(hydrationKey);
|
||||
const complete = await this.registerPersistedSubsessionLinks(
|
||||
parentSessionId,
|
||||
activeParent.runtime.session.sessionManager,
|
||||
activeParentFile,
|
||||
);
|
||||
if (complete) this.subsessionHydratedParents.add(hydrationKey);
|
||||
return;
|
||||
}
|
||||
|
||||
if (parentSessionFile === undefined) return;
|
||||
if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) {
|
||||
this.subsessionHydratedParents.add(hydrationKey);
|
||||
return;
|
||||
}
|
||||
if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) return;
|
||||
|
||||
let parentManager: PiSessionManager;
|
||||
try {
|
||||
parentManager = this.sessionManager.open(parentSessionFile);
|
||||
} catch {
|
||||
this.subsessionHydratedParents.add(hydrationKey);
|
||||
return;
|
||||
}
|
||||
await this.registerPersistedSubsessionLinks(parentSessionId, parentManager, parentSessionFile);
|
||||
this.subsessionHydratedParents.add(hydrationKey);
|
||||
const complete = await this.registerPersistedSubsessionLinks(parentSessionId, parentManager, parentSessionFile);
|
||||
if (complete) this.subsessionHydratedParents.add(hydrationKey);
|
||||
}
|
||||
|
||||
private async registerPersistedSubsessionLinks(parentSessionId: string, parentManager: PiSessionManager, parentSessionFile: string | undefined): Promise<void> {
|
||||
private async registerPersistedSubsessionLinks(parentSessionId: string, parentManager: PiSessionManager, parentSessionFile: string | undefined): Promise<boolean> {
|
||||
// Parent custom links are the authoritative recovery record: verify the
|
||||
// exact live child file/header before tracking.
|
||||
// exact live child file/header before tracking. Do not negatively cache a
|
||||
// scan while a candidate child is temporarily unavailable.
|
||||
const entries = parentManager.getEntries?.() ?? parentManager.getBranch();
|
||||
let complete = true;
|
||||
for (const entry of entries) {
|
||||
const link = parsePersistedParentSubsessionLink(entry);
|
||||
if (link === undefined) continue;
|
||||
if (link?.spawnedBySessionId !== parentSessionId) continue;
|
||||
const verified = await this.verifiedSubsessionLinkFromParentLink(parentSessionId, parentSessionFile, link);
|
||||
if (verified === undefined) continue;
|
||||
this.registerVerifiedSubsession(verified);
|
||||
if (verified === undefined) {
|
||||
complete = false;
|
||||
continue;
|
||||
}
|
||||
await this.registerVerifiedSubsession(verified);
|
||||
}
|
||||
return complete;
|
||||
}
|
||||
|
||||
private async verifiedSubsessionLinkFromParentLink(parentSessionId: string, parentSessionFile: string | undefined, link: PersistedParentSubsessionLink): Promise<TrackedSubsessionLink | undefined> {
|
||||
@@ -1160,7 +1232,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
private async recoverSubsessionTrackingForOpenedSession(session: PiAgentSession): Promise<void> {
|
||||
const link = await this.verifiedSubsessionLinkFromOpenedChild(session);
|
||||
if (link === undefined) return;
|
||||
this.registerVerifiedSubsession(link);
|
||||
await this.registerVerifiedSubsession(link);
|
||||
}
|
||||
|
||||
private verifiedSubsessionLinkFromOpenedChild(session: PiAgentSession): Promise<TrackedSubsessionLink | undefined> {
|
||||
@@ -1611,6 +1683,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
const archiveInput = await this.archiveInputForSession(session);
|
||||
await this.closeActive(session.sessionId, { kind: "clear", reason: "archive" });
|
||||
await this.archiveStore.archive(archiveInput);
|
||||
await this.forgetUnreadSessions([archiveInput]);
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -1623,6 +1696,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
]);
|
||||
const failures: SessionBulkFailure[] = [];
|
||||
const alreadyArchivedSessionIds: string[] = [];
|
||||
const unreadArchivedIdentities: { sessionId: string; cwd: string }[] = [];
|
||||
const planItems: BulkArchivePlanItem[] = [];
|
||||
|
||||
for (const ref of uniqueRefs) {
|
||||
@@ -1630,6 +1704,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
if (archived !== undefined) {
|
||||
this.publishNotificationMutations(this.notificationStore.clearSession(archived.sessionId, "archive"));
|
||||
alreadyArchivedSessionIds.push(archived.sessionId);
|
||||
unreadArchivedIdentities.push(archived);
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -1685,11 +1760,13 @@ export class PiSessionService implements SessionRouteService {
|
||||
try {
|
||||
const archived = await this.archiveStoreArchiveMany(readyInputs);
|
||||
archivedSessionIds.push(...archived.map((record) => record.sessionId));
|
||||
unreadArchivedIdentities.push(...archived);
|
||||
} catch (error: unknown) {
|
||||
for (const input of readyInputs) failures.push({ sessionId: input.sessionId, error: errorMessage(error) });
|
||||
}
|
||||
},
|
||||
);
|
||||
await this.forgetUnreadSessions(unreadArchivedIdentities);
|
||||
|
||||
return {
|
||||
archived: true,
|
||||
@@ -1722,6 +1799,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
await this.archiveStoreArchiveMany(archiveInputs);
|
||||
},
|
||||
);
|
||||
await this.forgetUnreadSessions(plan.targets.map((target) => ({ sessionId: target.id, cwd: target.cwd })));
|
||||
|
||||
return {
|
||||
archived: true,
|
||||
@@ -1736,6 +1814,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
if (archived === undefined) throw new Error("Session not found");
|
||||
await this.closeActive(archived.sessionId, { kind: "clear", reason: "restore" });
|
||||
await this.archiveStore.restore(archived.sessionId);
|
||||
await this.forgetUnreadSessions([archived]);
|
||||
}
|
||||
|
||||
async deleteArchived(ref: PiSessionLookup): Promise<void> {
|
||||
@@ -1746,6 +1825,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
await this.closeActive(record.sessionId, { kind: "clear", reason: "delete" });
|
||||
if (record.archivePath === undefined) await this.ensureArchivedRecordMoved(record);
|
||||
await this.archiveStore.deleteArchived(record.sessionId);
|
||||
await this.forgetUnreadSessions([record]);
|
||||
}
|
||||
|
||||
async deleteArchivedMany(refs: readonly SessionBulkMutationRef[]): Promise<SessionBulkDeleteArchivedResponse> {
|
||||
@@ -1794,6 +1874,8 @@ export class PiSessionService implements SessionRouteService {
|
||||
} catch (error: unknown) {
|
||||
for (const sessionId of deleteIds) failures.push({ sessionId, error: errorMessage(error) });
|
||||
}
|
||||
const deletedIdSet = new Set(deletedSessionIds);
|
||||
await this.forgetUnreadSessions(readyRecords.filter((record) => deletedIdSet.has(record.sessionId)));
|
||||
|
||||
return {
|
||||
deleted: true,
|
||||
@@ -1846,6 +1928,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
await clearParentSession(sessionFile);
|
||||
clearParentSessionHeader(session.sessionManager);
|
||||
this.unregisterSubsession(session.sessionId);
|
||||
await this.forgetUnreadSessions([{ sessionId: session.sessionId, cwd: session.sessionManager.getCwd() }]);
|
||||
}
|
||||
|
||||
async clearQueue(ref: PiSessionLookup): Promise<ClientSessionStatus> {
|
||||
@@ -2090,6 +2173,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
this.publishNotificationMutations(mutations);
|
||||
}
|
||||
if (!active) return;
|
||||
this.forgetUnreadActivity(active.runtime.session);
|
||||
this.active.delete(sessionId);
|
||||
this.activities.delete(sessionId);
|
||||
this.workspaceActivity?.removeSession(sessionId, active.runtime.session.sessionManager.getCwd());
|
||||
@@ -2226,6 +2310,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
...(options.initialModel === undefined ? {} : { initialModel: options.initialModel }),
|
||||
});
|
||||
const active: ActiveSession<PiSessionRuntime> = { runtime, unsubscribe: noop };
|
||||
let boundSession = runtime.session;
|
||||
let notificationGeneration = options.notificationGeneration;
|
||||
let notificationOwnership: "disabled" | "external" | "registered" | "replacement" = options.notifications === "disabled"
|
||||
? "disabled"
|
||||
@@ -2254,19 +2339,29 @@ export class PiSessionService implements SessionRouteService {
|
||||
if (notificationGeneration !== undefined) this.notificationGenerationBySession.set(runtime.session, notificationGeneration);
|
||||
|
||||
try {
|
||||
if (options.creationProvenance === "tracked-subsession") {
|
||||
await this.publishUnreadMutations(this.unreadStore.excludeSession(
|
||||
runtime.session.sessionId,
|
||||
canonicalizeStoredCwd(runtime.session.sessionManager.getCwd()),
|
||||
));
|
||||
} else {
|
||||
await this.recoverSubsessionTrackingForOpenedSession(runtime.session);
|
||||
}
|
||||
await this.bindSessionExtensions(runtime.session, notificationGeneration);
|
||||
this.bindRuntime(active);
|
||||
runtime.setRebindSession(async (session) => {
|
||||
const priorGeneration = notificationGeneration;
|
||||
let candidateGeneration: SessionNotificationGeneration | undefined;
|
||||
try {
|
||||
await this.prepareUnreadRuntimeRebind(boundSession, session);
|
||||
await this.recoverSubsessionTrackingForOpenedSession(session);
|
||||
if (priorGeneration !== undefined) {
|
||||
candidateGeneration = this.notificationStore.beginReplacement(priorGeneration, notificationIdentityForSession(session));
|
||||
this.notificationGenerationBySession.set(session, candidateGeneration);
|
||||
}
|
||||
this.bindRuntime(active, session);
|
||||
boundSession = session;
|
||||
await this.bindSessionExtensions(session, candidateGeneration);
|
||||
await this.recoverSubsessionTrackingForOpenedSession(session);
|
||||
if (candidateGeneration !== undefined) {
|
||||
this.publishNotificationMutations(this.notificationStore.commitReplacement(candidateGeneration));
|
||||
notificationGeneration = candidateGeneration;
|
||||
@@ -2281,7 +2376,6 @@ export class PiSessionService implements SessionRouteService {
|
||||
}
|
||||
});
|
||||
this.active.set(runtime.session.sessionId, active);
|
||||
await this.recoverSubsessionTrackingForOpenedSession(runtime.session);
|
||||
if (notificationOwnership === "replacement" && notificationGeneration !== undefined) {
|
||||
this.publishNotificationMutations(this.notificationStore.commitReplacement(notificationGeneration));
|
||||
notificationOwnership = "external";
|
||||
@@ -2297,6 +2391,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
}
|
||||
}
|
||||
active.unsubscribe();
|
||||
this.forgetUnreadActivity(boundSession);
|
||||
let removedActive = false;
|
||||
for (const [sessionId, candidate] of this.active.entries()) {
|
||||
if (candidate !== active) continue;
|
||||
@@ -2382,6 +2477,133 @@ export class PiSessionService implements SessionRouteService {
|
||||
}
|
||||
}
|
||||
|
||||
private async prepareUnreadRuntimeRebind(previous: PiAgentSession, next: PiAgentSession): Promise<void> {
|
||||
const previousCwd = canonicalizeStoredCwd(previous.sessionManager.getCwd());
|
||||
this.unreadStore.forgetActivity(previous.sessionId, previousCwd);
|
||||
const nextCwd = canonicalizeStoredCwd(next.sessionManager.getCwd());
|
||||
if (previous.sessionId === next.sessionId && cwdPathsEqual(previousCwd, nextCwd)) return;
|
||||
await this.publishUnreadMutations(this.unreadStore.forgetSession(previous.sessionId, previousCwd));
|
||||
}
|
||||
|
||||
private forgetUnreadActivity(session: PiAgentSession): void {
|
||||
this.unreadStore.forgetActivity(
|
||||
session.sessionId,
|
||||
canonicalizeStoredCwd(session.sessionManager.getCwd()),
|
||||
);
|
||||
}
|
||||
|
||||
private async forgetUnreadSessions(identities: readonly { sessionId: string; cwd: string }[]): Promise<void> {
|
||||
const mutations: SessionUnreadMutation[] = [];
|
||||
for (const identity of identities) {
|
||||
mutations.push(...this.unreadStore.forgetSession(
|
||||
identity.sessionId,
|
||||
canonicalizeStoredCwd(identity.cwd),
|
||||
));
|
||||
}
|
||||
await this.publishUnreadMutations(mutations);
|
||||
}
|
||||
|
||||
private observeUnreadActivityState(session: PiAgentSession): void {
|
||||
const mutations = this.unreadStore.observeActivityState(
|
||||
session.sessionId,
|
||||
canonicalizeStoredCwd(session.sessionManager.getCwd()),
|
||||
this.hasActiveWork(session),
|
||||
);
|
||||
if (mutations.length === 0) return;
|
||||
void this.publishUnreadMutations(mutations).catch(() => undefined);
|
||||
}
|
||||
|
||||
private publishUnreadMutations(mutations: readonly SessionUnreadMutation[]): Promise<void> {
|
||||
this.enqueueUnreadMutations(mutations);
|
||||
this.unreadPublicationFlushRequested = true;
|
||||
if (this.unreadPublication === undefined && this.unreadPublicationRetryTimer !== undefined) {
|
||||
const failure = this.unreadPublicationFailure;
|
||||
return Promise.reject(failure instanceof Error
|
||||
? failure
|
||||
: new Error("Session unread publication is awaiting retry", { cause: failure }));
|
||||
}
|
||||
return this.ensureUnreadPublication();
|
||||
}
|
||||
|
||||
private ensureUnreadPublication(): Promise<void> {
|
||||
const existing = this.unreadPublication;
|
||||
if (existing !== undefined) return existing;
|
||||
|
||||
const publication = this.drainUnreadPublication();
|
||||
this.unreadPublication = publication;
|
||||
void publication.then(
|
||||
() => {
|
||||
if (this.unreadPublication === publication) this.unreadPublication = undefined;
|
||||
},
|
||||
(error: unknown) => {
|
||||
if (this.unreadPublication === publication) this.unreadPublication = undefined;
|
||||
this.unreadPublicationFailure = error;
|
||||
this.logger.info(
|
||||
{ error: error instanceof Error ? error.message : String(error) },
|
||||
"failed to publish durable session unread mutations",
|
||||
);
|
||||
this.scheduleUnreadPublicationRetry();
|
||||
},
|
||||
);
|
||||
return publication;
|
||||
}
|
||||
|
||||
private async drainUnreadPublication(): Promise<void> {
|
||||
while (this.unreadPublicationFlushRequested || this.pendingUnreadMutations.length > 0) {
|
||||
this.unreadPublicationFlushRequested = false;
|
||||
const batch = this.pendingUnreadMutations.splice(0);
|
||||
let publishedCount = 0;
|
||||
try {
|
||||
await this.unreadStore.flush();
|
||||
for (const mutation of batch) {
|
||||
this.events.publishGlobal(mutation.event);
|
||||
publishedCount += 1;
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
this.prependUnreadMutations(batch.slice(publishedCount));
|
||||
this.unreadPublicationFlushRequested = true;
|
||||
throw error;
|
||||
}
|
||||
this.unreadPublicationFailure = undefined;
|
||||
this.clearUnreadPublicationRetry();
|
||||
}
|
||||
}
|
||||
|
||||
private enqueueUnreadMutations(mutations: readonly SessionUnreadMutation[]): void {
|
||||
this.pendingUnreadMutations.push(...mutations);
|
||||
this.trimPendingUnreadMutations();
|
||||
}
|
||||
|
||||
private prependUnreadMutations(mutations: readonly SessionUnreadMutation[]): void {
|
||||
this.pendingUnreadMutations.unshift(...mutations);
|
||||
this.trimPendingUnreadMutations();
|
||||
}
|
||||
|
||||
private trimPendingUnreadMutations(): void {
|
||||
const excess = this.pendingUnreadMutations.length - MAX_PENDING_UNREAD_MUTATIONS;
|
||||
if (excess > 0) this.pendingUnreadMutations.splice(0, excess);
|
||||
}
|
||||
|
||||
private scheduleUnreadPublicationRetry(): void {
|
||||
if (this.unreadPublicationStopped || this.unreadPublicationRetryTimer !== undefined) return;
|
||||
const delay = this.unreadPublicationRetryDelayMs;
|
||||
this.unreadPublicationRetryDelayMs = Math.min(
|
||||
Math.max(delay * 2, this.unreadPublicationRetryInitialMs),
|
||||
Math.max(MAX_UNREAD_PUBLICATION_RETRY_MS, this.unreadPublicationRetryInitialMs),
|
||||
);
|
||||
this.unreadPublicationRetryTimer = setTimeout(() => {
|
||||
this.unreadPublicationRetryTimer = undefined;
|
||||
void this.ensureUnreadPublication().catch(() => undefined);
|
||||
}, delay);
|
||||
this.unreadPublicationRetryTimer.unref();
|
||||
}
|
||||
|
||||
private clearUnreadPublicationRetry(): void {
|
||||
if (this.unreadPublicationRetryTimer !== undefined) clearTimeout(this.unreadPublicationRetryTimer);
|
||||
this.unreadPublicationRetryTimer = undefined;
|
||||
this.unreadPublicationRetryDelayMs = this.unreadPublicationRetryInitialMs;
|
||||
}
|
||||
|
||||
private bindRuntime(active: ActiveSession<PiSessionRuntime>, session: PiAgentSession = active.runtime.session): void {
|
||||
active.unsubscribe();
|
||||
for (const [sessionId, candidate] of this.active.entries()) {
|
||||
@@ -2602,12 +2824,14 @@ export class PiSessionService implements SessionRouteService {
|
||||
): Promise<T> {
|
||||
const sessionIds = new Set<string>();
|
||||
const runtimes = new Set<PiSessionRuntime>();
|
||||
const sessions = new Set<PiAgentSession>();
|
||||
for (const target of targets) {
|
||||
const runtime = target.runtime ?? (target.session === undefined ? undefined : this.activeRuntimeForSession(target.session));
|
||||
const session = target.session ?? runtime?.session;
|
||||
if (session !== undefined && this.hasActiveWork(session)) throw new Error(activeError);
|
||||
sessionIds.add(target.sessionId);
|
||||
if (runtime !== undefined) runtimes.add(runtime);
|
||||
if (session !== undefined) sessions.add(session);
|
||||
}
|
||||
|
||||
for (const sessionId of sessionIds) {
|
||||
@@ -2616,12 +2840,16 @@ export class PiSessionService implements SessionRouteService {
|
||||
for (const runtime of runtimes) {
|
||||
this.treeExclusiveRuntimeOperationCounts.set(runtime, (this.treeExclusiveRuntimeOperationCounts.get(runtime) ?? 0) + 1);
|
||||
}
|
||||
for (const session of sessions) this.observeUnreadActivityState(session);
|
||||
|
||||
try {
|
||||
return await operation();
|
||||
} finally {
|
||||
for (const runtime of runtimes) decrementWeakCount(this.treeExclusiveRuntimeOperationCounts, runtime);
|
||||
for (const sessionId of sessionIds) decrementMapCount(this.treeExclusiveSessionOperationCounts, sessionId);
|
||||
for (const session of sessions) {
|
||||
if (this.isCurrentActiveSession(session)) this.observeUnreadActivityState(session);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2658,12 +2886,14 @@ export class PiSessionService implements SessionRouteService {
|
||||
private beginSessionEntryMutation(session: PiAgentSession, action: string): void {
|
||||
this.assertTreeNavigationInactive(session, action);
|
||||
this.sessionEntryMutationCounts.set(session, (this.sessionEntryMutationCounts.get(session) ?? 0) + 1);
|
||||
this.observeUnreadActivityState(session);
|
||||
}
|
||||
|
||||
private endSessionEntryMutation(session: PiAgentSession): void {
|
||||
const remaining = (this.sessionEntryMutationCounts.get(session) ?? 1) - 1;
|
||||
if (remaining <= 0) this.sessionEntryMutationCounts.delete(session);
|
||||
else this.sessionEntryMutationCounts.set(session, remaining);
|
||||
this.observeUnreadActivityState(session);
|
||||
}
|
||||
|
||||
private isSessionEntryMutationActive(session: PiAgentSession): boolean {
|
||||
@@ -2705,6 +2935,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
this.workspaceActivity?.applySessionActivity(session.sessionManager.getCwd(), activity);
|
||||
this.events.publish(session.sessionId, { type: "activity.update", activity });
|
||||
this.events.publishGlobal({ type: "activity.update", activity });
|
||||
this.observeUnreadActivityState(session);
|
||||
}
|
||||
|
||||
private publishStatus(session: PiAgentSession): void {
|
||||
@@ -2713,6 +2944,7 @@ export class PiSessionService implements SessionRouteService {
|
||||
this.workspaceActivity?.applySessionStatus(session.sessionManager.getCwd(), status);
|
||||
this.events.publish(session.sessionId, { type: "status.update", status });
|
||||
this.events.publishGlobal({ type: "status.update", status });
|
||||
this.observeUnreadActivityState(session);
|
||||
}
|
||||
|
||||
private clearStaleActiveActivity(session: PiAgentSession): void {
|
||||
|
||||
@@ -0,0 +1,675 @@
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { PiSessionService, type PiAgentSession } from "./piSessionService.js";
|
||||
import {
|
||||
CapturingSessionEventHub,
|
||||
emptyArchiveStore,
|
||||
fakeRuntime,
|
||||
fakeSessionManager,
|
||||
runtimeCreator,
|
||||
sessionGateway,
|
||||
sessionRecord,
|
||||
sessionRef,
|
||||
testModelRuntime,
|
||||
type RuntimeCreator,
|
||||
} from "./piSessionService.testSupport.js";
|
||||
import {
|
||||
SessionUnreadStore,
|
||||
type SessionUnreadPersistedState,
|
||||
type SessionUnreadPersistence,
|
||||
} from "./sessionUnreadStore.js";
|
||||
|
||||
const TEST_AGENT_DIR = "/tmp/pi-web-test-agent";
|
||||
const tempRoots: string[] = [];
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(tempRoots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
describe("PiSessionService daemon-owned unread state", () => {
|
||||
it("records one durable completion and keeps stale acknowledgements from clearing newer work", async () => {
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
const hub = new CapturingSessionEventHub();
|
||||
const fake = fakeRuntime("session-1");
|
||||
const service = new PiSessionService(hub, {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(fake.runtime),
|
||||
sessionManager: sessionGateway([sessionRecord("session-1")]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("session-1"));
|
||||
completeRuntimeWork(fake);
|
||||
completeRuntimeWork(fake);
|
||||
|
||||
const secondSnapshot = await service.unreadCatalog();
|
||||
const current = secondSnapshot.sessions[0];
|
||||
expect(current).toMatchObject({ sessionId: "session-1", cwd: "/workspace", completionOrder: 2 });
|
||||
expect(unreadEvents(hub).map((event) => event.catalogRevision)).toEqual([1, 2]);
|
||||
|
||||
const staleSnapshot = await service.acknowledgeUnread("session-1", {
|
||||
cwd: "/workspace",
|
||||
catalogId: "catalog-test",
|
||||
throughCompletionOrder: 1,
|
||||
});
|
||||
expect(staleSnapshot.sessions).toEqual(secondSnapshot.sessions);
|
||||
expect(unreadEvents(hub).map((event) => event.catalogRevision)).toEqual([1, 2]);
|
||||
|
||||
const acknowledged = await service.acknowledgeUnread("session-1", {
|
||||
cwd: "/workspace",
|
||||
catalogId: "catalog-test",
|
||||
throughCompletionOrder: current?.completionOrder ?? 0,
|
||||
});
|
||||
expect(acknowledged.sessions).toEqual([]);
|
||||
expect(unreadEvents(hub).at(-1)).toMatchObject({ catalogRevision: 3, sessionId: "session-1", unread: null });
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("tracks service-owned activity even while runtime status flags look idle", async () => {
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
const fake = fakeRuntime("session-1");
|
||||
let finishBash: (() => void) | undefined;
|
||||
fake.session.executeBash = () => new Promise((resolve) => {
|
||||
finishBash = () => { resolve({ output: "done", exitCode: 0, cancelled: false, truncated: false }); };
|
||||
});
|
||||
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(fake.runtime),
|
||||
sessionManager: sessionGateway([sessionRecord("session-1")]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("session-1"));
|
||||
await service.shell(sessionRef("session-1"), "!echo done");
|
||||
expect(fake.session.isStreaming).toBe(false);
|
||||
expect(fake.session.isBashRunning).toBe(false);
|
||||
expect((await service.unreadCatalog()).sessions).toEqual([]);
|
||||
|
||||
finishBash?.();
|
||||
await Promise.resolve();
|
||||
|
||||
expect((await service.unreadCatalog()).sessions).toMatchObject([{ sessionId: "session-1", cwd: "/workspace", completionOrder: 1 }]);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("publishes completion revisions in order only after their captured state is durable", async () => {
|
||||
const persistence = new BlockingUnreadPersistence();
|
||||
const unreadStore = new SessionUnreadStore({ persistence, createCatalogId: () => "catalog-test" });
|
||||
await unreadStore.load();
|
||||
const hub = new CapturingSessionEventHub();
|
||||
const fake = fakeRuntime("session-1");
|
||||
const service = new PiSessionService(hub, {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(fake.runtime),
|
||||
sessionManager: sessionGateway([sessionRecord("session-1")]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("session-1"));
|
||||
const blockedSave = persistence.blockNextSave();
|
||||
completeRuntimeWork(fake);
|
||||
completeRuntimeWork(fake);
|
||||
await Promise.resolve();
|
||||
|
||||
expect(unreadEvents(hub)).toEqual([]);
|
||||
|
||||
blockedSave.resolve();
|
||||
const snapshot = await service.unreadCatalog();
|
||||
|
||||
expect(snapshot.sessions).toMatchObject([{ sessionId: "session-1", completionOrder: 2 }]);
|
||||
expect(unreadEvents(hub).map((event) => event.catalogRevision)).toEqual([1, 2]);
|
||||
expect(persistence.savedStates.at(-1)).toMatchObject({ catalogRevision: 2, nextCompletionOrder: 2 });
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not publish a mutation queued after the current batch became durable", async () => {
|
||||
const persistence = new BlockingUnreadPersistence();
|
||||
const unreadStore = new SessionUnreadStore({ persistence, createCatalogId: () => "catalog-test" });
|
||||
await unreadStore.load();
|
||||
const hub = new CapturingSessionEventHub();
|
||||
const fake = fakeRuntime("session-1");
|
||||
const service = new PiSessionService(hub, {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(fake.runtime),
|
||||
sessionManager: sessionGateway([sessionRecord("session-1")]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
const flush = unreadStore.flush.bind(unreadStore);
|
||||
let blockedSecondSave: Deferred | undefined;
|
||||
let injectedSecondCompletion = false;
|
||||
vi.spyOn(unreadStore, "flush").mockImplementation(async () => {
|
||||
await flush();
|
||||
if (injectedSecondCompletion) return;
|
||||
injectedSecondCompletion = true;
|
||||
blockedSecondSave = persistence.blockNextSave();
|
||||
completeRuntimeWork(fake);
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("session-1"));
|
||||
completeRuntimeWork(fake);
|
||||
await drainMicrotasks();
|
||||
|
||||
expect(persistence.savedStates.at(-1)).toMatchObject({ catalogRevision: 1, nextCompletionOrder: 1 });
|
||||
expect(unreadEvents(hub).map((event) => event.catalogRevision)).toEqual([1]);
|
||||
|
||||
if (blockedSecondSave === undefined) throw new Error("Expected the second unread save to be blocked");
|
||||
blockedSecondSave.resolve();
|
||||
await service.unreadCatalog();
|
||||
|
||||
expect(persistence.savedStates.at(-1)).toMatchObject({ catalogRevision: 2, nextCompletionOrder: 2 });
|
||||
expect(unreadEvents(hub).map((event) => event.catalogRevision)).toEqual([1, 2]);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("retries failed durable publication without waiting for another client request", async () => {
|
||||
vi.useFakeTimers();
|
||||
const persistence = new RecoveringUnreadPersistence(2);
|
||||
const unreadStore = new SessionUnreadStore({ persistence, createCatalogId: () => "unused-catalog" });
|
||||
await unreadStore.load();
|
||||
const hub = new CapturingSessionEventHub();
|
||||
const fake = fakeRuntime("session-1");
|
||||
const service = new PiSessionService(hub, {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(fake.runtime),
|
||||
sessionManager: sessionGateway([sessionRecord("session-1")]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
unreadPublicationRetryDelayMs: 100,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("session-1"));
|
||||
completeRuntimeWork(fake);
|
||||
await drainMicrotasks();
|
||||
|
||||
expect(persistence.saveCalls).toBe(2);
|
||||
expect(unreadEvents(hub)).toEqual([]);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
await drainMicrotasks();
|
||||
|
||||
expect(persistence.saveCalls).toBe(3);
|
||||
expect(persistence.persistedState()).toMatchObject({ catalogRevision: 1, nextCompletionOrder: 1 });
|
||||
expect(unreadEvents(hub)).toMatchObject([{ catalogRevision: 1, sessionId: "session-1" }]);
|
||||
} finally {
|
||||
try {
|
||||
await service.dispose();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("forgets a closing runtime latch without manufacturing a stop completion and preserves unread across reload work", async () => {
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
const hub = new CapturingSessionEventHub();
|
||||
const runtimes = [fakeRuntime("session-1"), fakeRuntime("session-1"), fakeRuntime("session-1")];
|
||||
let runtimeIndex = 0;
|
||||
const createAgentRuntime: RuntimeCreator = () => {
|
||||
const next = runtimes[runtimeIndex++];
|
||||
if (next === undefined) throw new Error("Unexpected extra runtime creation");
|
||||
return Promise.resolve(next.runtime);
|
||||
};
|
||||
const service = new PiSessionService(hub, {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime,
|
||||
sessionManager: sessionGateway([sessionRecord("session-1")]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("session-1"));
|
||||
const initial = runtimes[0];
|
||||
if (initial === undefined) throw new Error("Expected an initial runtime");
|
||||
initial.session.isStreaming = true;
|
||||
initial.emit({ type: "agent_start" });
|
||||
await service.stop(sessionRef("session-1"));
|
||||
initial.session.isStreaming = false;
|
||||
expect((await service.unreadCatalog()).sessions).toEqual([]);
|
||||
|
||||
completeStoreWork(unreadStore, "session-1", "/workspace");
|
||||
const beforeReload = (await service.unreadCatalog()).sessions[0];
|
||||
await service.reload(sessionRef("session-1"));
|
||||
const afterReload = (await service.unreadCatalog()).sessions[0];
|
||||
|
||||
expect(beforeReload).toBeDefined();
|
||||
expect(afterReload).toMatchObject({ sessionId: "session-1", cwd: "/workspace" });
|
||||
expect(afterReload?.completionOrder).toBeGreaterThan(beforeReload?.completionOrder ?? 0);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("clears stale unread when a runtime rebind changes logical session identity", async () => {
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
completeStoreWork(unreadStore, "session-old", "/workspace");
|
||||
const original = fakeRuntime("session-old");
|
||||
let rebindSession: ((session: PiAgentSession) => Promise<void>) | undefined;
|
||||
original.runtime.setRebindSession = (callback) => { rebindSession = callback; };
|
||||
const replacement = fakeRuntime("session-new");
|
||||
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(original.runtime),
|
||||
sessionManager: sessionGateway([sessionRecord("session-old")]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("session-old"));
|
||||
if (rebindSession === undefined) throw new Error("Expected runtime rebind callback");
|
||||
await rebindSession(replacement.session);
|
||||
|
||||
expect((await service.unreadCatalog()).sessions).toEqual([]);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("cleans unread state through archive, restore, delete, and cwd reconciliation", async () => {
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
for (const sessionId of ["archive-me", "restore-me", "delete-me", "orphan"]) {
|
||||
completeStoreWork(unreadStore, sessionId, "/workspace");
|
||||
}
|
||||
const archived = new Map([
|
||||
["restore-me", { sessionId: "restore-me", cwd: "/workspace", archivedAt: "2026-07-01T00:00:00.000Z", archivePath: "/archive/restore-me.jsonl" }],
|
||||
["delete-me", { sessionId: "delete-me", cwd: "/workspace", archivedAt: "2026-07-01T00:00:00.000Z", archivePath: "/archive/delete-me.jsonl" }],
|
||||
]);
|
||||
const fake = fakeRuntime("archive-me");
|
||||
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(fake.runtime),
|
||||
sessionManager: sessionGateway([sessionRecord("archive-me")]),
|
||||
archiveStore: {
|
||||
list: () => Promise.resolve([...archived.values()]),
|
||||
get: (sessionId) => Promise.resolve([...archived.values()].find((record) => record.sessionId.startsWith(sessionId))),
|
||||
archive: (input) => {
|
||||
const record = { sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-07-20T00:00:00.000Z", archivePath: `/archive/${input.sessionId}.jsonl` };
|
||||
archived.set(input.sessionId, record);
|
||||
return Promise.resolve(record);
|
||||
},
|
||||
restore: (sessionId) => { archived.delete(sessionId); return Promise.resolve(); },
|
||||
deleteArchived: (sessionId) => { archived.delete(sessionId); return Promise.resolve(); },
|
||||
isArchived: (sessionId) => Promise.resolve(archived.has(sessionId)),
|
||||
},
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("archive-me"));
|
||||
await service.archive(sessionRef("archive-me"));
|
||||
expect((await service.unreadCatalog()).sessions.map((summary) => summary.sessionId)).toEqual([
|
||||
"orphan",
|
||||
"delete-me",
|
||||
"restore-me",
|
||||
]);
|
||||
|
||||
await service.restore(sessionRef("restore-me"));
|
||||
expect((await service.unreadCatalog()).sessions.map((summary) => summary.sessionId)).toEqual([
|
||||
"orphan",
|
||||
"delete-me",
|
||||
]);
|
||||
|
||||
await service.deleteArchived(sessionRef("delete-me"));
|
||||
expect((await service.unreadCatalog()).sessions.map((summary) => summary.sessionId)).toEqual(["orphan"]);
|
||||
|
||||
await service.list("/workspace");
|
||||
expect((await service.unreadCatalog()).sessions).toEqual([]);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("excludes live tracked sub-sessions, then restores ordinary tracking after detach", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "pi-web-unread-live-subsessions-"));
|
||||
tempRoots.push(root);
|
||||
const parentFile = join(root, "parent.jsonl");
|
||||
const childFile = join(root, "child.jsonl");
|
||||
await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
const hub = new CapturingSessionEventHub();
|
||||
const parent = fakeRuntime("parent-1", { sessionFile: parentFile });
|
||||
const child = fakeRuntime("child-1", {
|
||||
sessionFile: childFile,
|
||||
sessionManager: fakeSessionManager("/workspace-feature"),
|
||||
});
|
||||
child.session.prompt = () => {
|
||||
completeRuntimeWork(child);
|
||||
return Promise.resolve();
|
||||
};
|
||||
const runtimes = [parent.runtime, child.runtime];
|
||||
let runtimeIndex = 0;
|
||||
const createAgentRuntime: RuntimeCreator = () => Promise.resolve(runtimes[runtimeIndex++] ?? child.runtime);
|
||||
const service = new PiSessionService(hub, {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime,
|
||||
sessionManager: sessionGateway([]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
spawnTargets: { resolveSpawnTarget: () => Promise.resolve({ allowed: true, cwd: "/workspace-feature" }) },
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.start("/workspace");
|
||||
await service.spawnSubsession({
|
||||
spawningCwd: "/workspace",
|
||||
parentSessionId: "parent-1",
|
||||
parentSessionFile: parentFile,
|
||||
prompt: "do the slice",
|
||||
cwd: "/workspace-feature",
|
||||
});
|
||||
completeRuntimeWork(child);
|
||||
|
||||
expect((await service.unreadCatalog()).sessions.some((summary) => summary.sessionId === "child-1")).toBe(false);
|
||||
expect(unreadEvents(hub).some((event) => event.sessionId === "child-1" && event.unread !== null)).toBe(false);
|
||||
|
||||
await service.detachParent(sessionRef("child-1", "/workspace-feature"));
|
||||
completeRuntimeWork(child);
|
||||
expect((await service.unreadCatalog()).sessions).toContainEqual(expect.objectContaining({
|
||||
sessionId: "child-1",
|
||||
cwd: "/workspace-feature",
|
||||
}));
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("clears accidental unread when a reciprocal persisted tracked link is verified after restart", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "pi-web-unread-subsessions-"));
|
||||
tempRoots.push(root);
|
||||
const parentFile = join(root, "parent.jsonl");
|
||||
const childFile = join(root, "child.jsonl");
|
||||
await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8");
|
||||
await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
|
||||
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
completeStoreWork(unreadStore, "child-1", "/workspace-feature");
|
||||
const hub = new CapturingSessionEventHub();
|
||||
const parentManager = fakeSessionManager("/workspace", {
|
||||
getEntries: () => [{
|
||||
type: "custom",
|
||||
customType: "pi-web.subsession.link",
|
||||
data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace-feature" },
|
||||
}],
|
||||
});
|
||||
const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager });
|
||||
const service = new PiSessionService(hub, {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(parent.runtime),
|
||||
sessionManager: {
|
||||
create: () => parentManager,
|
||||
list: () => Promise.resolve([]),
|
||||
listAll: () => Promise.resolve([]),
|
||||
open: () => fakeSessionManager("/workspace-feature"),
|
||||
},
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.start("/workspace");
|
||||
await expect(service.listSubsessions("parent-1", parentFile)).resolves.toEqual([
|
||||
{ sessionId: "child-1", cwd: "/workspace-feature", status: "idle" },
|
||||
]);
|
||||
|
||||
expect((await service.unreadCatalog()).sessions).toEqual([]);
|
||||
expect(unreadEvents(hub).at(-1)).toMatchObject({ sessionId: "child-1", cwd: "/workspace-feature", unread: null });
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("retries tracked-child hydration after a linked child is temporarily unavailable", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "pi-web-unread-subsessions-retry-"));
|
||||
tempRoots.push(root);
|
||||
const parentFile = join(root, "parent.jsonl");
|
||||
const childFile = join(root, "child.jsonl");
|
||||
await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8");
|
||||
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
completeStoreWork(unreadStore, "child-1", "/workspace-feature");
|
||||
const parentManager = fakeSessionManager("/workspace", {
|
||||
getEntries: () => [{
|
||||
type: "custom",
|
||||
customType: "pi-web.subsession.link",
|
||||
data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace-feature" },
|
||||
}],
|
||||
});
|
||||
const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager });
|
||||
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(parent.runtime),
|
||||
sessionManager: {
|
||||
create: () => parentManager,
|
||||
list: () => Promise.resolve([]),
|
||||
listAll: () => Promise.resolve([]),
|
||||
open: () => fakeSessionManager("/workspace-feature"),
|
||||
},
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.start("/workspace");
|
||||
await expect(service.listSubsessions("parent-1", parentFile)).resolves.toEqual([]);
|
||||
|
||||
await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
|
||||
await expect(service.listSubsessions("parent-1", parentFile)).resolves.toEqual([
|
||||
{ sessionId: "child-1", cwd: "/workspace-feature", status: "idle" },
|
||||
]);
|
||||
expect((await service.unreadCatalog()).sessions).toEqual([]);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not re-exclude a detached child from persisted markers after restart", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "pi-web-unread-detached-subsessions-"));
|
||||
tempRoots.push(root);
|
||||
const parentFile = join(root, "parent.jsonl");
|
||||
const childFile = join(root, "child.jsonl");
|
||||
await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8");
|
||||
await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature" })}\n`, "utf8");
|
||||
|
||||
const parentManager = fakeSessionManager("/workspace", {
|
||||
getSessionId: () => "parent-1",
|
||||
getSessionFile: () => parentFile,
|
||||
getEntries: () => [{
|
||||
type: "custom",
|
||||
customType: "pi-web.subsession.link",
|
||||
data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace-feature" },
|
||||
}],
|
||||
});
|
||||
const childManager = fakeSessionManager("/workspace-feature", {
|
||||
getSessionId: () => "child-1",
|
||||
getSessionFile: () => childFile,
|
||||
getEntries: () => [{
|
||||
type: "custom",
|
||||
customType: "pi-web.subsession.spawned",
|
||||
data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" },
|
||||
}],
|
||||
});
|
||||
const childRecord = { ...sessionRecord("child-1", "/workspace-feature"), path: childFile };
|
||||
const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager });
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
completeStoreWork(unreadStore, "child-1", "/workspace-feature");
|
||||
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(child.runtime),
|
||||
sessionManager: {
|
||||
create: () => childManager,
|
||||
list: () => Promise.resolve([childRecord]),
|
||||
listAll: () => Promise.resolve([childRecord]),
|
||||
open: (path) => path === parentFile ? parentManager : childManager,
|
||||
},
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("child-1", "/workspace-feature"));
|
||||
await expect(service.listSubsessions("parent-1", parentFile)).resolves.toEqual([]);
|
||||
expect((await service.unreadCatalog()).sessions).toMatchObject([
|
||||
{ sessionId: "child-1", cwd: "/workspace-feature" },
|
||||
]);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not exclude a generic parentSessionPath descendant without verified tracked markers", async () => {
|
||||
const unreadStore = new SessionUnreadStore({ createCatalogId: () => "catalog-test" });
|
||||
completeStoreWork(unreadStore, "branch-1", "/workspace");
|
||||
const branch = fakeRuntime("branch-1", {
|
||||
sessionFile: "/tmp/branch-1.jsonl",
|
||||
sessionManager: fakeSessionManager("/workspace", { getBranch: () => [] }),
|
||||
});
|
||||
const genericDescendant = { ...sessionRecord("branch-1"), parentSessionPath: "/tmp/parent.jsonl" };
|
||||
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||
agentDir: TEST_AGENT_DIR,
|
||||
modelRuntime: testModelRuntime,
|
||||
createAgentRuntime: runtimeCreator(branch.runtime),
|
||||
sessionManager: sessionGateway([genericDescendant]),
|
||||
archiveStore: emptyArchiveStore(),
|
||||
heartbeatIntervalMs: 60_000,
|
||||
unreadStore,
|
||||
});
|
||||
|
||||
try {
|
||||
await service.status(sessionRef("branch-1"));
|
||||
expect((await service.unreadCatalog()).sessions).toMatchObject([{ sessionId: "branch-1", cwd: "/workspace" }]);
|
||||
} finally {
|
||||
await service.dispose();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
function completeRuntimeWork(runtime: ReturnType<typeof fakeRuntime>): void {
|
||||
runtime.session.isStreaming = true;
|
||||
runtime.emit({ type: "agent_start" });
|
||||
runtime.session.isStreaming = false;
|
||||
runtime.emit({ type: "turn_end" });
|
||||
}
|
||||
|
||||
function completeStoreWork(store: SessionUnreadStore, sessionId: string, cwd: string): void {
|
||||
store.observeActivityState(sessionId, cwd, true);
|
||||
store.observeActivityState(sessionId, cwd, false);
|
||||
}
|
||||
|
||||
function unreadEvents(hub: CapturingSessionEventHub) {
|
||||
return hub.globalEvents.filter((event) => event.type === "sessions.unread");
|
||||
}
|
||||
|
||||
interface Deferred {
|
||||
promise: Promise<void>;
|
||||
resolve(): void;
|
||||
}
|
||||
|
||||
class RecoveringUnreadPersistence implements SessionUnreadPersistence {
|
||||
saveCalls = 0;
|
||||
private value: SessionUnreadPersistedState = {
|
||||
version: 1,
|
||||
catalogId: "catalog-test",
|
||||
catalogRevision: 0,
|
||||
nextCompletionOrder: 0,
|
||||
sessions: [],
|
||||
};
|
||||
|
||||
constructor(private readonly failures: number) {}
|
||||
|
||||
load(): Promise<unknown> {
|
||||
return Promise.resolve(structuredClone(this.value));
|
||||
}
|
||||
|
||||
save(state: SessionUnreadPersistedState): Promise<void> {
|
||||
this.saveCalls += 1;
|
||||
if (this.saveCalls <= this.failures) return Promise.reject(new Error("unread persistence unavailable"));
|
||||
this.value = structuredClone(state);
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
persistedState(): SessionUnreadPersistedState {
|
||||
return structuredClone(this.value);
|
||||
}
|
||||
}
|
||||
|
||||
class BlockingUnreadPersistence implements SessionUnreadPersistence {
|
||||
readonly savedStates: SessionUnreadPersistedState[] = [];
|
||||
private persistedState: SessionUnreadPersistedState | undefined;
|
||||
private nextSaveGate: Deferred | undefined;
|
||||
|
||||
load(): Promise<unknown> {
|
||||
return Promise.resolve(this.persistedState);
|
||||
}
|
||||
|
||||
async save(state: SessionUnreadPersistedState): Promise<void> {
|
||||
const gate = this.nextSaveGate;
|
||||
this.nextSaveGate = undefined;
|
||||
if (gate !== undefined) await gate.promise;
|
||||
const saved = structuredClone(state);
|
||||
this.persistedState = saved;
|
||||
this.savedStates.push(saved);
|
||||
}
|
||||
|
||||
blockNextSave(): Deferred {
|
||||
const gate = deferred();
|
||||
this.nextSaveGate = gate;
|
||||
return gate;
|
||||
}
|
||||
}
|
||||
|
||||
async function drainMicrotasks(): Promise<void> {
|
||||
for (let index = 0; index < 20; index += 1) await Promise.resolve();
|
||||
}
|
||||
|
||||
function deferred(): Deferred {
|
||||
let resolvePromise: (() => void) | undefined;
|
||||
const promise = new Promise<void>((resolve) => { resolvePromise = resolve; });
|
||||
return {
|
||||
promise,
|
||||
resolve() { resolvePromise?.(); },
|
||||
};
|
||||
}
|
||||
@@ -2,7 +2,7 @@ import { resolve } from "node:path";
|
||||
import Fastify, { type FastifyInstance } from "fastify";
|
||||
import fastifyWebsocket from "@fastify/websocket";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH } from "../../shared/apiTypes.js";
|
||||
import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_CATALOG_ID_MAX_LENGTH } from "../../shared/apiTypes.js";
|
||||
import type {
|
||||
MessagePage,
|
||||
SessionBulkArchiveResponse,
|
||||
@@ -17,6 +17,8 @@ import type {
|
||||
SessionStatus,
|
||||
SessionStreamSnapshot,
|
||||
SessionTreeNavigateRequest,
|
||||
SessionUnreadAcknowledgeRequest,
|
||||
SessionUnreadCatalogSnapshot,
|
||||
SessionTreeNavigateResult,
|
||||
} from "../../shared/apiTypes.js";
|
||||
import { SessionEventHub } from "../realtime/sessionEventHub.js";
|
||||
@@ -71,6 +73,86 @@ describe("session routes", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("returns unread snapshots and validates race-safe acknowledgement cutoffs", async () => {
|
||||
const routeApp = Fastify({ logger: false });
|
||||
await routeApp.register(fastifyWebsocket);
|
||||
const eventHub = new SessionEventHub();
|
||||
const routeService = new CapturingRouteSessionService();
|
||||
registerSessionRoutes(routeApp, routeService, eventHub);
|
||||
|
||||
try {
|
||||
const requestCwd = resolve("/repo");
|
||||
const catalog = await routeApp.inject({ method: "GET", url: "/sessions/unread" });
|
||||
const acknowledged = await routeApp.inject({
|
||||
method: "POST",
|
||||
url: "/sessions/session-1/unread/acknowledge",
|
||||
payload: { cwd: requestCwd, catalogId: "catalog-test", throughCompletionOrder: 7 },
|
||||
});
|
||||
const invalid = await routeApp.inject({
|
||||
method: "POST",
|
||||
url: "/sessions/session-1/unread/acknowledge",
|
||||
payload: { cwd: requestCwd, catalogId: "catalog-test", throughCompletionOrder: 0 },
|
||||
});
|
||||
const oversized = await routeApp.inject({
|
||||
method: "POST",
|
||||
url: "/sessions/session-1/unread/acknowledge",
|
||||
payload: {
|
||||
cwd: requestCwd,
|
||||
catalogId: "x".repeat(SESSION_UNREAD_CATALOG_ID_MAX_LENGTH + 1),
|
||||
throughCompletionOrder: 7,
|
||||
},
|
||||
});
|
||||
|
||||
expect(catalog.statusCode).toBe(200);
|
||||
expect(catalog.json()).toEqual(routeService.unreadCatalogResponse);
|
||||
expect(acknowledged.statusCode).toBe(200);
|
||||
expect(acknowledged.json()).toEqual(routeService.unreadCatalogResponse);
|
||||
expect(invalid.statusCode).toBe(400);
|
||||
expect(invalid.json()).toEqual({ error: "throughCompletionOrder field must be positive" });
|
||||
expect(oversized.statusCode).toBe(400);
|
||||
expect(oversized.json()).toEqual({ error: "catalogId field is too long" });
|
||||
expect(routeService.acknowledgeUnreadCalls).toEqual([{
|
||||
sessionId: "session-1",
|
||||
request: { cwd: requestCwd, catalogId: "catalog-test", throughCompletionOrder: 7 },
|
||||
}]);
|
||||
} finally {
|
||||
await routeService.dispose();
|
||||
await routeApp.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("reports unread backend failures as unavailable while keeping validation errors at 400", async () => {
|
||||
const routeApp = Fastify({ logger: false });
|
||||
await routeApp.register(fastifyWebsocket);
|
||||
const eventHub = new SessionEventHub();
|
||||
const routeService = new CapturingRouteSessionService();
|
||||
routeService.unreadError = new Error("unread persistence unavailable");
|
||||
registerSessionRoutes(routeApp, routeService, eventHub);
|
||||
|
||||
try {
|
||||
const catalog = await routeApp.inject({ method: "GET", url: "/sessions/unread" });
|
||||
const acknowledgement = await routeApp.inject({
|
||||
method: "POST",
|
||||
url: "/sessions/session-1/unread/acknowledge",
|
||||
payload: { cwd: resolve("/repo"), catalogId: "catalog-test", throughCompletionOrder: 7 },
|
||||
});
|
||||
const invalid = await routeApp.inject({
|
||||
method: "POST",
|
||||
url: "/sessions/session-1/unread/acknowledge",
|
||||
payload: { cwd: "relative", catalogId: "catalog-test", throughCompletionOrder: 7 },
|
||||
});
|
||||
|
||||
expect(catalog.statusCode).toBe(503);
|
||||
expect(acknowledgement.statusCode).toBe(503);
|
||||
expect(catalog.json()).toEqual({ error: "unread persistence unavailable" });
|
||||
expect(acknowledgement.json()).toEqual({ error: "unread persistence unavailable" });
|
||||
expect(invalid.statusCode).toBe(400);
|
||||
} finally {
|
||||
await routeService.dispose();
|
||||
await routeApp.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("validates and forwards idempotent notification dismissal cutoffs", async () => {
|
||||
const routeApp = Fastify({ logger: false });
|
||||
await routeApp.register(fastifyWebsocket);
|
||||
@@ -610,9 +692,12 @@ class CapturingRouteSessionService implements SessionRouteService {
|
||||
readonly clearQueueCalls: SessionRouteLookup[] = [];
|
||||
readonly dismissWarningCalls: { lookup: SessionRouteLookup; dismissId: string }[] = [];
|
||||
readonly notificationInboxCalls: SessionRef[] = [];
|
||||
readonly acknowledgeUnreadCalls: { sessionId: string; request: SessionUnreadAcknowledgeRequest }[] = [];
|
||||
readonly unreadCatalogResponse: SessionUnreadCatalogSnapshot = { catalogId: "catalog-test", catalogRevision: 1, sessions: [] };
|
||||
readonly dismissNotificationCalls: { ref: SessionRef; request: Omit<SessionNotificationDismissRequest, "cwd"> }[] = [];
|
||||
readonly dismissAllNotificationCalls: { ref: SessionRef; request: Omit<SessionNotificationDismissAllRequest, "cwd"> }[] = [];
|
||||
dismissWarningError: Error | undefined;
|
||||
unreadError: Error | undefined;
|
||||
messagesResponse: unknown[] | MessagePage = [];
|
||||
streamSnapshotResponse: SessionStreamSnapshot = { seq: 0, partial: null };
|
||||
readonly streamSnapshotCalls: SessionRouteLookup[] = [];
|
||||
@@ -658,6 +743,19 @@ class CapturingRouteSessionService implements SessionRouteService {
|
||||
return { daemonInstanceId: "daemon-test", catalogRevision: 0, sessions: [] };
|
||||
}
|
||||
|
||||
unreadCatalog(): Promise<SessionUnreadCatalogSnapshot> {
|
||||
return this.unreadError === undefined
|
||||
? Promise.resolve(this.unreadCatalogResponse)
|
||||
: Promise.reject(this.unreadError);
|
||||
}
|
||||
|
||||
acknowledgeUnread(sessionId: string, request: SessionUnreadAcknowledgeRequest): Promise<SessionUnreadCatalogSnapshot> {
|
||||
this.acknowledgeUnreadCalls.push({ sessionId, request });
|
||||
return this.unreadError === undefined
|
||||
? Promise.resolve(this.unreadCatalogResponse)
|
||||
: Promise.reject(this.unreadError);
|
||||
}
|
||||
|
||||
notificationInbox(ref: SessionRef): SessionNotificationInboxSnapshot {
|
||||
this.notificationInboxCalls.push(ref);
|
||||
return notificationSnapshot(ref);
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { FastifyInstance } from "fastify";
|
||||
import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, type SessionBulkMutationRequest, type SessionBulkMutationRef, type SessionCleanupRequest, type SessionTreeNavigateRequest, type SessionTreeSummaryChoice } from "../../shared/apiTypes.js";
|
||||
import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_CATALOG_ID_MAX_LENGTH, SESSION_UNREAD_CWD_MAX_LENGTH, SESSION_UNREAD_SESSION_ID_MAX_LENGTH, type SessionBulkMutationRequest, type SessionBulkMutationRef, type SessionCleanupRequest, type SessionTreeNavigateRequest, type SessionTreeSummaryChoice, type SessionUnreadAcknowledgeRequest } from "../../shared/apiTypes.js";
|
||||
import { projectBrowserMessageResponse } from "../browserMessageProjection.js";
|
||||
import { normalizeRequestCwd } from "../workingDirectory.js";
|
||||
import type { SessionEventHub } from "../realtime/sessionEventHub.js";
|
||||
@@ -62,6 +62,35 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: SessionRou
|
||||
}
|
||||
});
|
||||
|
||||
app.get(`${prefix}/sessions/unread`, async (_request, reply) => {
|
||||
try {
|
||||
return await sessions.unreadCatalog();
|
||||
} catch (error) {
|
||||
return reply.code(503).send({ error: errorMessage(error) });
|
||||
}
|
||||
});
|
||||
|
||||
app.post<{ Params: { sessionId: string }; Body: Record<string, unknown> | undefined }>(`${prefix}/sessions/:sessionId/unread/acknowledge`, async (request, reply) => {
|
||||
let sessionId: string;
|
||||
let acknowledgement: SessionUnreadAcknowledgeRequest;
|
||||
try {
|
||||
const body = requireRecord(request.body);
|
||||
sessionId = requireNonEmptyBoundedString(request.params.sessionId, "sessionId", SESSION_UNREAD_SESSION_ID_MAX_LENGTH);
|
||||
acknowledgement = {
|
||||
cwd: normalizeRequestCwd(requireNonEmptyBoundedString(body["cwd"], "cwd", SESSION_UNREAD_CWD_MAX_LENGTH)),
|
||||
catalogId: requireNonEmptyBoundedString(body["catalogId"], "catalogId", SESSION_UNREAD_CATALOG_ID_MAX_LENGTH),
|
||||
throughCompletionOrder: requirePositiveSafeInteger(body["throughCompletionOrder"], "throughCompletionOrder"),
|
||||
};
|
||||
} catch (error) {
|
||||
return reply.code(400).send({ error: errorMessage(error) });
|
||||
}
|
||||
try {
|
||||
return await sessions.acknowledgeUnread(sessionId, acknowledgement);
|
||||
} catch (error) {
|
||||
return reply.code(503).send({ error: errorMessage(error) });
|
||||
}
|
||||
});
|
||||
|
||||
app.post<{ Body: SessionCleanupRequest | undefined }>(`${prefix}/sessions/cleanup/preview`, async (request, reply) => {
|
||||
try {
|
||||
return await sessions.cleanupPreview(normalizeSessionCleanupRequest(optionalRecord(request.body)));
|
||||
@@ -509,6 +538,12 @@ function requireNonNegativeSafeInteger(value: unknown, field: string): number {
|
||||
return value;
|
||||
}
|
||||
|
||||
function requirePositiveSafeInteger(value: unknown, field: string): number {
|
||||
const parsed = requireNonNegativeSafeInteger(value, field);
|
||||
if (parsed === 0) throw new Error(`${field} field must be positive`);
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function requireThinkingLevel(value: unknown): string {
|
||||
if (typeof value !== "string" || value === "") throw new Error("level field is invalid");
|
||||
return value;
|
||||
|
||||
@@ -7,6 +7,8 @@ import type {
|
||||
SessionNotificationDismissAllRequest,
|
||||
SessionNotificationDismissRequest,
|
||||
SessionNotificationInboxSnapshot,
|
||||
SessionUnreadAcknowledgeRequest,
|
||||
SessionUnreadCatalogSnapshot,
|
||||
} from "../../shared/apiTypes.js";
|
||||
import type {
|
||||
ClientArchiveSessionsResponse,
|
||||
@@ -43,6 +45,8 @@ export interface SessionRouteService {
|
||||
status(ref: SessionRouteLookup): Promise<ClientSessionStatus>;
|
||||
streamSnapshot(ref: SessionRouteLookup): Promise<SessionStreamSnapshot>;
|
||||
notificationCatalog(): SessionNotificationCatalogSnapshot | Promise<SessionNotificationCatalogSnapshot>;
|
||||
unreadCatalog(): Promise<SessionUnreadCatalogSnapshot>;
|
||||
acknowledgeUnread(sessionId: string, request: SessionUnreadAcknowledgeRequest): Promise<SessionUnreadCatalogSnapshot>;
|
||||
notificationInbox(ref: SessionRouteRef): SessionNotificationInboxSnapshot | Promise<SessionNotificationInboxSnapshot>;
|
||||
dismissNotification(ref: SessionRouteRef, request: Omit<SessionNotificationDismissRequest, "cwd">): SessionNotificationInboxSnapshot | Promise<SessionNotificationInboxSnapshot>;
|
||||
dismissAllNotifications(ref: SessionRouteRef, request: Omit<SessionNotificationDismissAllRequest, "cwd">): SessionNotificationInboxSnapshot | Promise<SessionNotificationInboxSnapshot>;
|
||||
|
||||
@@ -0,0 +1,593 @@
|
||||
import { mkdtemp, readdir, readFile, rm, stat, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { SESSION_UNREAD_LIMIT, SESSION_UNREAD_SESSION_ID_MAX_LENGTH } from "../../shared/apiTypes.js";
|
||||
import {
|
||||
FileSessionUnreadPersistence,
|
||||
SessionUnreadStore,
|
||||
defaultSessionUnreadFilePath,
|
||||
type SessionUnreadPersistedState,
|
||||
type SessionUnreadPersistence,
|
||||
} from "./sessionUnreadStore.js";
|
||||
|
||||
const roots: string[] = [];
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
describe("SessionUnreadStore", () => {
|
||||
it("marks only known active-to-idle transitions unread", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-a");
|
||||
|
||||
expect(store.observeActivityState("session-1", "/repo", false)).toEqual([]);
|
||||
expect(store.observeActivityState("session-1", "/repo", true)).toEqual([]);
|
||||
expect(store.observeActivityState("session-1", "/repo", true)).toEqual([]);
|
||||
const completed = store.observeActivityState("session-1", "/repo", false);
|
||||
|
||||
expect(completed).toMatchObject([{
|
||||
event: {
|
||||
type: "sessions.unread",
|
||||
catalogId: "catalog-a",
|
||||
catalogRevision: 1,
|
||||
sessionId: "session-1",
|
||||
cwd: "/repo",
|
||||
unread: { completionOrder: 1, completedAt: "2026-07-20T00:00:00.000Z" },
|
||||
},
|
||||
}]);
|
||||
expect(store.catalogSnapshot()).toMatchObject({
|
||||
catalogId: "catalog-a",
|
||||
catalogRevision: 1,
|
||||
sessions: [{ sessionId: "session-1", cwd: "/repo", completionOrder: 1 }],
|
||||
});
|
||||
expect(store.observeActivityState("session-1", "/repo", false)).toEqual([]);
|
||||
});
|
||||
|
||||
it("uses monotonic completion orders so stale acknowledgements cannot clear newer work", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-a");
|
||||
complete(store, "session-1", "/repo");
|
||||
const firstOrder = currentOrder(store, "session-1", "/repo");
|
||||
|
||||
complete(store, "session-1", "/repo");
|
||||
const secondOrder = currentOrder(store, "session-1", "/repo");
|
||||
|
||||
expect(secondOrder).toBeGreaterThan(firstOrder);
|
||||
expect(store.acknowledge("session-1", {
|
||||
cwd: "/repo",
|
||||
catalogId: "catalog-a",
|
||||
throughCompletionOrder: firstOrder,
|
||||
}).mutations).toEqual([]);
|
||||
expect(currentOrder(store, "session-1", "/repo")).toBe(secondOrder);
|
||||
|
||||
const acknowledged = store.acknowledge("session-1", {
|
||||
cwd: "/repo",
|
||||
catalogId: "catalog-a",
|
||||
throughCompletionOrder: secondOrder,
|
||||
});
|
||||
expect(acknowledged.mutations).toMatchObject([{
|
||||
event: {
|
||||
type: "sessions.unread",
|
||||
catalogId: "catalog-a",
|
||||
catalogRevision: 3,
|
||||
sessionId: "session-1",
|
||||
unread: null,
|
||||
},
|
||||
}]);
|
||||
expect(store.catalogSnapshot().sessions).toEqual([]);
|
||||
expect(store.acknowledge("session-1", {
|
||||
cwd: "/repo",
|
||||
catalogId: "catalog-a",
|
||||
throughCompletionOrder: secondOrder,
|
||||
}).mutations).toEqual([]);
|
||||
});
|
||||
|
||||
it("rejects stale acknowledgements from a reset catalog epoch", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-new");
|
||||
complete(store, "session-1", "/repo");
|
||||
const current = currentOrder(store, "session-1", "/repo");
|
||||
|
||||
const stale = store.acknowledge("session-1", {
|
||||
cwd: "/repo",
|
||||
catalogId: "catalog-old",
|
||||
throughCompletionOrder: Number.MAX_SAFE_INTEGER,
|
||||
});
|
||||
|
||||
expect(stale.mutations).toEqual([]);
|
||||
expect(currentOrder(store, "session-1", "/repo")).toBe(current);
|
||||
});
|
||||
|
||||
it("scopes lifecycle and acknowledgements to the canonical id and cwd pair", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-a");
|
||||
complete(store, "session-1", "/repo-a");
|
||||
complete(store, "session-1", "/repo-b");
|
||||
|
||||
const repoBOrder = currentOrder(store, "session-1", "/repo-b");
|
||||
store.acknowledge("session-1", {
|
||||
cwd: "/repo-b",
|
||||
catalogId: "catalog-a",
|
||||
throughCompletionOrder: repoBOrder,
|
||||
});
|
||||
|
||||
expect(store.catalogSnapshot().sessions).toMatchObject([
|
||||
{ sessionId: "session-1", cwd: "/repo-a", completionOrder: 1 },
|
||||
]);
|
||||
});
|
||||
|
||||
it("forgets a closing runtime's active latch without manufacturing a completion", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-a");
|
||||
store.observeActivityState("session-1", "/repo", true);
|
||||
|
||||
store.forgetActivity("session-1", "/repo");
|
||||
|
||||
expect(store.observeActivityState("session-1", "/repo", false)).toEqual([]);
|
||||
expect(store.catalogSnapshot().sessions).toEqual([]);
|
||||
});
|
||||
|
||||
it("excludes verified tracked sub-sessions and clears accidental lifecycle state", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-a");
|
||||
store.observeActivityState("tracked", "/repo", true);
|
||||
|
||||
expect(store.excludeSession("tracked", "/repo")).toEqual([]);
|
||||
expect(store.observeActivityState("tracked", "/repo", false)).toEqual([]);
|
||||
expect(store.observeActivityState("tracked", "/repo", true)).toEqual([]);
|
||||
expect(store.observeActivityState("tracked", "/repo", false)).toEqual([]);
|
||||
|
||||
store.forgetSession("tracked", "/repo");
|
||||
complete(store, "tracked", "/repo");
|
||||
const removed = store.excludeSession("tracked", "/repo");
|
||||
|
||||
expect(removed).toMatchObject([{
|
||||
event: { catalogId: "catalog-a", sessionId: "tracked", cwd: "/repo", unread: null },
|
||||
}]);
|
||||
expect(store.catalogSnapshot().sessions).toEqual([]);
|
||||
});
|
||||
|
||||
it("removes durable and transient state when a cwd is reconciled", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-a");
|
||||
complete(store, "keep", "/repo");
|
||||
complete(store, "remove", "/repo");
|
||||
store.observeActivityState("active-orphan", "/repo", true);
|
||||
store.excludeSession("excluded-orphan", "/repo");
|
||||
|
||||
const mutations = store.reconcileCwd("/repo", ["keep"]);
|
||||
|
||||
expect(mutations).toMatchObject([{ event: { sessionId: "remove", unread: null } }]);
|
||||
expect(store.catalogSnapshot().sessions.map((summary) => summary.sessionId)).toEqual(["keep"]);
|
||||
expect(store.observeActivityState("active-orphan", "/repo", false)).toEqual([]);
|
||||
complete(store, "excluded-orphan", "/repo");
|
||||
expect(currentOrder(store, "excluded-orphan", "/repo")).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("bounds the catalog and emits an authoritative removal when pruning", () => {
|
||||
const store = storeAt("2026-07-20T00:00:00.000Z", "catalog-a");
|
||||
let finalMutations = store.observeActivityState("baseline", "/repo", false);
|
||||
for (let index = 0; index <= SESSION_UNREAD_LIMIT; index += 1) {
|
||||
const sessionId = `session-${index.toString()}`;
|
||||
store.observeActivityState(sessionId, "/repo", true);
|
||||
finalMutations = store.observeActivityState(sessionId, "/repo", false);
|
||||
}
|
||||
|
||||
const snapshot = store.catalogSnapshot();
|
||||
expect(snapshot.sessions).toHaveLength(SESSION_UNREAD_LIMIT);
|
||||
expect(snapshot.sessions.some((summary) => summary.sessionId === "session-0")).toBe(false);
|
||||
expect(snapshot.sessions[0]).toMatchObject({ sessionId: `session-${SESSION_UNREAD_LIMIT.toString()}`, completionOrder: SESSION_UNREAD_LIMIT + 1 });
|
||||
expect(finalMutations).toMatchObject([
|
||||
{ event: { unread: { sessionId: `session-${SESSION_UNREAD_LIMIT.toString()}` } } },
|
||||
{ event: { sessionId: "session-0", unread: null } },
|
||||
]);
|
||||
});
|
||||
|
||||
it("persists the catalog epoch, revisions, and completion order across store instances", async () => {
|
||||
const persistence = new MemoryPersistence(undefined);
|
||||
const first = persistedStore(persistence, "catalog-a", "2026-07-20T00:00:00.000Z");
|
||||
await Promise.all([first.load(), first.load()]);
|
||||
expect(persistence.loadCalls).toBe(1);
|
||||
complete(first, "session-1", "/repo");
|
||||
await first.flush();
|
||||
|
||||
const second = persistedStore(persistence, "unused-catalog-b", "2026-07-20T01:00:00.000Z");
|
||||
await second.load();
|
||||
expect(second.catalogSnapshot()).toEqual(first.catalogSnapshot());
|
||||
|
||||
const order = currentOrder(second, "session-1", "/repo");
|
||||
second.acknowledge("session-1", {
|
||||
cwd: "/repo",
|
||||
catalogId: "catalog-a",
|
||||
throughCompletionOrder: order,
|
||||
});
|
||||
await second.flush();
|
||||
|
||||
const third = persistedStore(persistence, "unused-catalog-c", "2026-07-20T02:00:00.000Z");
|
||||
await third.load();
|
||||
expect(third.catalogSnapshot()).toMatchObject({ catalogId: "catalog-a", catalogRevision: 2, sessions: [] });
|
||||
complete(third, "session-1", "/repo");
|
||||
expect(currentOrder(third, "session-1", "/repo")).toBe(2);
|
||||
});
|
||||
|
||||
it("repairs malformed persistence with a fresh epoch and keeps stale old-epoch acks harmless", async () => {
|
||||
const errors: { operation: "load" | "save"; error: unknown }[] = [];
|
||||
const persistence = new MemoryPersistence({ version: 999, catalogId: "catalog-old" });
|
||||
const store = new SessionUnreadStore({
|
||||
persistence,
|
||||
createCatalogId: () => "catalog-reset",
|
||||
onPersistenceError: (operation, error) => { errors.push({ operation, error }); },
|
||||
now: () => new Date("2026-07-20T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
expect(() => store.catalogSnapshot()).toThrow("must be loaded");
|
||||
await store.load();
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(errors[0]?.operation).toBe("load");
|
||||
expect(errors[0]?.error).toBeInstanceOf(Error);
|
||||
expect(store.catalogSnapshot()).toEqual({ catalogId: "catalog-reset", catalogRevision: 0, sessions: [] });
|
||||
expect(persistence.valueSnapshot()).toMatchObject({
|
||||
version: 1,
|
||||
catalogId: "catalog-reset",
|
||||
catalogRevision: 0,
|
||||
nextCompletionOrder: 0,
|
||||
sessions: [],
|
||||
});
|
||||
|
||||
complete(store, "session-1", "/repo");
|
||||
expect(store.acknowledge("session-1", {
|
||||
cwd: "/repo",
|
||||
catalogId: "catalog-old",
|
||||
throughCompletionOrder: Number.MAX_SAFE_INTEGER,
|
||||
}).mutations).toEqual([]);
|
||||
expect(store.catalogSnapshot().sessions).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("resets persisted protocol fields that exceed shared bounds and rejects oversized runtime identities", async () => {
|
||||
const persistence = new MemoryPersistence({
|
||||
version: 1,
|
||||
catalogId: "catalog-old",
|
||||
catalogRevision: 1,
|
||||
nextCompletionOrder: 1,
|
||||
sessions: [{
|
||||
sessionId: "x".repeat(SESSION_UNREAD_SESSION_ID_MAX_LENGTH + 1),
|
||||
cwd: "/repo",
|
||||
completionOrder: 1,
|
||||
completedAt: "2026-07-20T00:00:00.000Z",
|
||||
}],
|
||||
});
|
||||
const errors: unknown[] = [];
|
||||
const store = new SessionUnreadStore({
|
||||
persistence,
|
||||
createCatalogId: () => "catalog-reset",
|
||||
onPersistenceError: (_operation, error) => { errors.push(error); },
|
||||
});
|
||||
|
||||
await store.load();
|
||||
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(store.catalogSnapshot()).toEqual({ catalogId: "catalog-reset", catalogRevision: 0, sessions: [] });
|
||||
expect(() => store.observeActivityState(
|
||||
"x".repeat(SESSION_UNREAD_SESSION_ID_MAX_LENGTH + 1),
|
||||
"/repo",
|
||||
true,
|
||||
)).toThrow("sessionId exceeds its length limit");
|
||||
});
|
||||
|
||||
it("rejects an oversized persisted catalog instead of loading unbounded state", async () => {
|
||||
const sessions = Array.from({ length: SESSION_UNREAD_LIMIT + 1 }, (_, index) => ({
|
||||
sessionId: `session-${index.toString()}`,
|
||||
cwd: "/repo",
|
||||
completionOrder: index + 1,
|
||||
completedAt: "2026-07-20T00:00:00.000Z",
|
||||
}));
|
||||
const persistence = new MemoryPersistence({
|
||||
version: 1,
|
||||
catalogId: "catalog-old",
|
||||
catalogRevision: sessions.length,
|
||||
nextCompletionOrder: sessions.length,
|
||||
sessions,
|
||||
});
|
||||
const errors: unknown[] = [];
|
||||
const store = new SessionUnreadStore({
|
||||
persistence,
|
||||
createCatalogId: () => "catalog-reset",
|
||||
onPersistenceError: (_operation, error) => { errors.push(error); },
|
||||
});
|
||||
|
||||
await store.load();
|
||||
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(store.catalogSnapshot()).toEqual({ catalogId: "catalog-reset", catalogRevision: 0, sessions: [] });
|
||||
});
|
||||
|
||||
it("serializes writes and captures each mutation's state", async () => {
|
||||
const persistence = new BlockingPersistence(emptyPersistedState("catalog-a"));
|
||||
const store = persistedStore(persistence, "unused-catalog", "2026-07-20T00:00:00.000Z");
|
||||
await store.load();
|
||||
complete(store, "session-1", "/repo");
|
||||
const order = currentOrder(store, "session-1", "/repo");
|
||||
store.acknowledge("session-1", {
|
||||
cwd: "/repo",
|
||||
catalogId: "catalog-a",
|
||||
throughCompletionOrder: order,
|
||||
});
|
||||
|
||||
await vi.waitFor(() => { expect(persistence.savedStates).toHaveLength(1); });
|
||||
expect(persistence.maximumConcurrentSaves).toBe(1);
|
||||
expect(persistence.savedStates[0]?.sessions).toHaveLength(1);
|
||||
|
||||
persistence.releaseNextSave();
|
||||
await vi.waitFor(() => { expect(persistence.savedStates).toHaveLength(2); });
|
||||
expect(persistence.maximumConcurrentSaves).toBe(1);
|
||||
expect(persistence.savedStates[1]?.sessions).toEqual([]);
|
||||
|
||||
const flushed = store.flush();
|
||||
persistence.releaseNextSave();
|
||||
await flushed;
|
||||
});
|
||||
|
||||
it("coalesces a mutation burst behind one in-flight persistence write", async () => {
|
||||
const persistence = new BlockingPersistence(emptyPersistedState("catalog-a"));
|
||||
const store = persistedStore(persistence, "unused-catalog", "2026-07-20T00:00:00.000Z");
|
||||
await store.load();
|
||||
|
||||
complete(store, "session-0", "/repo");
|
||||
for (let index = 1; index <= 200; index += 1) complete(store, `session-${index.toString()}`, "/repo");
|
||||
|
||||
await vi.waitFor(() => { expect(persistence.savedStates).toHaveLength(1); });
|
||||
expect(persistence.savedStates[0]).toMatchObject({ catalogRevision: 1, nextCompletionOrder: 1 });
|
||||
|
||||
persistence.releaseNextSave();
|
||||
await vi.waitFor(() => { expect(persistence.savedStates).toHaveLength(2); });
|
||||
expect(persistence.savedStates[1]).toMatchObject({ catalogRevision: 201, nextCompletionOrder: 201 });
|
||||
|
||||
persistence.releaseNextSave();
|
||||
await store.flush();
|
||||
expect(persistence.savedStates).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("holds one latest snapshot without retrying every mutation during a storage outage", async () => {
|
||||
const saveError = new Error("storage unavailable");
|
||||
const save = vi.fn<SessionUnreadPersistence["save"]>(() => Promise.reject(saveError));
|
||||
const store = persistedStore({
|
||||
load: () => Promise.resolve(emptyPersistedState("catalog-a")),
|
||||
save,
|
||||
}, "unused-catalog", "2026-07-20T00:00:00.000Z");
|
||||
await store.load();
|
||||
|
||||
complete(store, "session-0", "/repo");
|
||||
await vi.waitFor(() => { expect(save).toHaveBeenCalledOnce(); });
|
||||
for (let index = 1; index <= 100; index += 1) complete(store, `session-${index.toString()}`, "/repo");
|
||||
await Promise.resolve();
|
||||
|
||||
expect(save).toHaveBeenCalledOnce();
|
||||
await expect(store.flush()).rejects.toBe(saveError);
|
||||
expect(save).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("retries the latest snapshot before exposing state after a transient save failure", async () => {
|
||||
const persistence = new FailOncePersistence(emptyPersistedState("catalog-a"));
|
||||
const errors: unknown[] = [];
|
||||
const store = new SessionUnreadStore({
|
||||
persistence,
|
||||
createCatalogId: () => "unused-catalog",
|
||||
now: () => new Date("2026-07-20T00:00:00.000Z"),
|
||||
onPersistenceError: (_operation, error) => { errors.push(error); },
|
||||
});
|
||||
await store.load();
|
||||
complete(store, "session-1", "/repo");
|
||||
|
||||
const snapshot = await store.durableCatalogSnapshot();
|
||||
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(persistence.saveCalls).toBe(2);
|
||||
expect(persistence.valueSnapshot()).toMatchObject({
|
||||
catalogId: "catalog-a",
|
||||
catalogRevision: 1,
|
||||
nextCompletionOrder: 1,
|
||||
sessions: [{ sessionId: "session-1", completionOrder: 1 }],
|
||||
});
|
||||
expect(snapshot.sessions).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("rejects operational load failures without overwriting the unread file", async () => {
|
||||
const loadError = Object.assign(new Error("read failed"), { code: "EIO" });
|
||||
const save = vi.fn<SessionUnreadPersistence["save"]>(() => Promise.resolve());
|
||||
const errors: { operation: "load" | "save"; error: unknown }[] = [];
|
||||
const store = new SessionUnreadStore({
|
||||
persistence: { load: () => Promise.reject(loadError), save },
|
||||
createCatalogId: () => "catalog-reset",
|
||||
onPersistenceError: (operation, error) => { errors.push({ operation, error }); },
|
||||
});
|
||||
|
||||
await expect(store.load()).rejects.toBe(loadError);
|
||||
|
||||
expect(save).not.toHaveBeenCalled();
|
||||
expect(errors).toEqual([{ operation: "load", error: loadError }]);
|
||||
expect(() => store.catalogSnapshot()).toThrow("must be loaded");
|
||||
});
|
||||
|
||||
it("rejects startup when a missing or corrupt catalog cannot persist its fresh epoch", async () => {
|
||||
const saveError = new Error("disk full");
|
||||
const errors: { operation: "load" | "save"; error: unknown }[] = [];
|
||||
const store = new SessionUnreadStore({
|
||||
persistence: {
|
||||
load: () => Promise.resolve(undefined),
|
||||
save: () => Promise.reject(saveError),
|
||||
},
|
||||
createCatalogId: () => "catalog-reset",
|
||||
onPersistenceError: (operation, error) => { errors.push({ operation, error }); },
|
||||
});
|
||||
|
||||
await expect(store.load()).rejects.toBe(saveError);
|
||||
|
||||
expect(errors).toEqual([{ operation: "save", error: saveError }]);
|
||||
expect(() => store.catalogSnapshot()).toThrow("must be loaded");
|
||||
});
|
||||
});
|
||||
|
||||
describe("FileSessionUnreadPersistence", () => {
|
||||
it("repairs malformed JSON with a fresh persisted catalog epoch", async () => {
|
||||
const root = await temporaryRoot();
|
||||
const filePath = join(root, "session-unread.json");
|
||||
await writeFile(filePath, "{not-json", "utf8");
|
||||
const errors: unknown[] = [];
|
||||
const store = new SessionUnreadStore({
|
||||
persistence: new FileSessionUnreadPersistence(filePath),
|
||||
createCatalogId: () => "catalog-reset",
|
||||
onPersistenceError: (_operation, error) => { errors.push(error); },
|
||||
});
|
||||
|
||||
await store.load();
|
||||
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(store.catalogSnapshot()).toEqual({ catalogId: "catalog-reset", catalogRevision: 0, sessions: [] });
|
||||
expect(JSON.parse(await readFile(filePath, "utf8"))).toMatchObject({ catalogId: "catalog-reset", sessions: [] });
|
||||
});
|
||||
|
||||
it("uses PI_WEB_DATA_DIR and atomically reloads a private state file", async () => {
|
||||
const root = await temporaryRoot();
|
||||
expect(defaultSessionUnreadFilePath({ PI_WEB_DATA_DIR: "state" }, root)).toBe(join(root, "state", "session-unread.json"));
|
||||
const filePath = join(root, "state", "custom-unread.json");
|
||||
const persistence = new FileSessionUnreadPersistence(filePath);
|
||||
const store = new SessionUnreadStore({
|
||||
persistence,
|
||||
createCatalogId: () => "catalog-a",
|
||||
now: () => new Date("2026-07-20T00:00:00.000Z"),
|
||||
});
|
||||
await store.load();
|
||||
complete(store, "session-1", "/repo");
|
||||
await store.flush();
|
||||
|
||||
const persisted: unknown = JSON.parse(await readFile(filePath, "utf8"));
|
||||
expect(persisted).toMatchObject({
|
||||
version: 1,
|
||||
catalogId: "catalog-a",
|
||||
catalogRevision: 1,
|
||||
nextCompletionOrder: 1,
|
||||
sessions: [{ sessionId: "session-1", cwd: "/repo", completionOrder: 1 }],
|
||||
});
|
||||
expect((await stat(filePath)).mode & 0o777).toBe(0o600);
|
||||
expect((await readdir(join(root, "state"))).filter((name) => name.endsWith(".tmp"))).toEqual([]);
|
||||
|
||||
const reloaded = new SessionUnreadStore({ persistence, createCatalogId: () => "unused-catalog" });
|
||||
await reloaded.load();
|
||||
expect(reloaded.catalogSnapshot()).toMatchObject({
|
||||
catalogId: "catalog-a",
|
||||
sessions: [{ sessionId: "session-1", completionOrder: 1 }],
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
function storeAt(iso: string, catalogId: string): SessionUnreadStore {
|
||||
return new SessionUnreadStore({ now: () => new Date(iso), createCatalogId: () => catalogId });
|
||||
}
|
||||
|
||||
function persistedStore(
|
||||
persistence: SessionUnreadPersistence,
|
||||
catalogId: string,
|
||||
iso: string,
|
||||
): SessionUnreadStore {
|
||||
return new SessionUnreadStore({
|
||||
persistence,
|
||||
createCatalogId: () => catalogId,
|
||||
now: () => new Date(iso),
|
||||
});
|
||||
}
|
||||
|
||||
function complete(store: SessionUnreadStore, sessionId: string, cwd: string): void {
|
||||
store.observeActivityState(sessionId, cwd, true);
|
||||
store.observeActivityState(sessionId, cwd, false);
|
||||
}
|
||||
|
||||
function currentOrder(store: SessionUnreadStore, sessionId: string, cwd: string): number {
|
||||
return store.catalogSnapshot().sessions.find((summary) => summary.sessionId === sessionId && summary.cwd === cwd)?.completionOrder ?? 0;
|
||||
}
|
||||
|
||||
function emptyPersistedState(catalogId: string): SessionUnreadPersistedState {
|
||||
return {
|
||||
version: 1,
|
||||
catalogId,
|
||||
catalogRevision: 0,
|
||||
nextCompletionOrder: 0,
|
||||
sessions: [],
|
||||
};
|
||||
}
|
||||
|
||||
class MemoryPersistence implements SessionUnreadPersistence {
|
||||
loadCalls = 0;
|
||||
|
||||
constructor(private value: unknown) {}
|
||||
|
||||
load(): Promise<unknown> {
|
||||
this.loadCalls += 1;
|
||||
return Promise.resolve(structuredClone(this.value));
|
||||
}
|
||||
|
||||
save(state: SessionUnreadPersistedState): Promise<void> {
|
||||
this.value = structuredClone(state);
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
valueSnapshot(): unknown {
|
||||
return structuredClone(this.value);
|
||||
}
|
||||
}
|
||||
|
||||
class FailOncePersistence implements SessionUnreadPersistence {
|
||||
saveCalls = 0;
|
||||
private value: SessionUnreadPersistedState;
|
||||
|
||||
constructor(initial: SessionUnreadPersistedState) {
|
||||
this.value = structuredClone(initial);
|
||||
}
|
||||
|
||||
load(): Promise<unknown> {
|
||||
return Promise.resolve(structuredClone(this.value));
|
||||
}
|
||||
|
||||
save(state: SessionUnreadPersistedState): Promise<void> {
|
||||
this.saveCalls += 1;
|
||||
if (this.saveCalls === 1) return Promise.reject(new Error("transient save failure"));
|
||||
this.value = structuredClone(state);
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
valueSnapshot(): unknown {
|
||||
return structuredClone(this.value);
|
||||
}
|
||||
}
|
||||
|
||||
class BlockingPersistence implements SessionUnreadPersistence {
|
||||
readonly savedStates: SessionUnreadPersistedState[] = [];
|
||||
maximumConcurrentSaves = 0;
|
||||
private concurrentSaves = 0;
|
||||
private readonly releases: (() => void)[] = [];
|
||||
|
||||
constructor(private readonly initial: SessionUnreadPersistedState) {}
|
||||
|
||||
load(): Promise<unknown> {
|
||||
return Promise.resolve(structuredClone(this.initial));
|
||||
}
|
||||
|
||||
save(state: SessionUnreadPersistedState): Promise<void> {
|
||||
this.concurrentSaves += 1;
|
||||
this.maximumConcurrentSaves = Math.max(this.maximumConcurrentSaves, this.concurrentSaves);
|
||||
this.savedStates.push(structuredClone(state));
|
||||
return new Promise<void>((resolve) => {
|
||||
this.releases.push(() => {
|
||||
this.concurrentSaves -= 1;
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
releaseNextSave(): void {
|
||||
const release = this.releases.shift();
|
||||
if (release === undefined) throw new Error("No blocked persistence save to release");
|
||||
release();
|
||||
}
|
||||
}
|
||||
|
||||
async function temporaryRoot(): Promise<string> {
|
||||
const root = await mkdtemp(join(tmpdir(), "pi-web-unread-"));
|
||||
roots.push(root);
|
||||
return root;
|
||||
}
|
||||
@@ -0,0 +1,603 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { dirname, join } from "node:path";
|
||||
import { piWebDataDir } from "../../config.js";
|
||||
import {
|
||||
SESSION_UNREAD_CATALOG_ID_MAX_LENGTH,
|
||||
SESSION_UNREAD_COMPLETED_AT_MAX_LENGTH,
|
||||
SESSION_UNREAD_CWD_MAX_LENGTH,
|
||||
SESSION_UNREAD_LIMIT,
|
||||
SESSION_UNREAD_SESSION_ID_MAX_LENGTH,
|
||||
type SessionUnreadAcknowledgeRequest,
|
||||
type SessionUnreadCatalogSnapshot,
|
||||
type SessionUnreadEvent,
|
||||
type SessionUnreadSummary,
|
||||
} from "../../shared/apiTypes.js";
|
||||
|
||||
const SESSION_UNREAD_STATE_VERSION = 1;
|
||||
const SESSION_UNREAD_FILE_MODE = 0o600;
|
||||
|
||||
export interface SessionUnreadPersistedState {
|
||||
version: typeof SESSION_UNREAD_STATE_VERSION;
|
||||
catalogId: string;
|
||||
catalogRevision: number;
|
||||
nextCompletionOrder: number;
|
||||
sessions: SessionUnreadSummary[];
|
||||
}
|
||||
|
||||
export interface SessionUnreadPersistence {
|
||||
load(): Promise<unknown>;
|
||||
save(state: SessionUnreadPersistedState): Promise<void>;
|
||||
}
|
||||
|
||||
export interface SessionUnreadStoreOptions {
|
||||
now?: (() => Date) | undefined;
|
||||
persistence?: SessionUnreadPersistence | undefined;
|
||||
createCatalogId?: (() => string) | undefined;
|
||||
onPersistenceError?: ((operation: "load" | "save", error: unknown) => void) | undefined;
|
||||
}
|
||||
|
||||
export interface SessionUnreadMutation {
|
||||
event: SessionUnreadEvent;
|
||||
}
|
||||
|
||||
export interface SessionUnreadAcknowledgeResult {
|
||||
mutations: SessionUnreadMutation[];
|
||||
}
|
||||
|
||||
interface SessionUnreadIdentity {
|
||||
sessionId: string;
|
||||
cwd: string;
|
||||
}
|
||||
|
||||
interface PendingSessionUnreadPersistence {
|
||||
generation: number;
|
||||
state: SessionUnreadPersistedState;
|
||||
}
|
||||
|
||||
/**
|
||||
* Daemon-owned unread completion catalog.
|
||||
*
|
||||
* Completion orders are global and never reused within a catalog epoch. An
|
||||
* acknowledgement must carry both the epoch and the observed completion order,
|
||||
* so neither an old completion nor a client from reset state can clear newer
|
||||
* work.
|
||||
*/
|
||||
export class SessionUnreadStore {
|
||||
private readonly now: () => Date;
|
||||
private readonly persistence: SessionUnreadPersistence | undefined;
|
||||
private readonly createCatalogId: () => string;
|
||||
private readonly onPersistenceError: (operation: "load" | "save", error: unknown) => void;
|
||||
private readonly unreadByIdentity = new Map<string, SessionUnreadSummary>();
|
||||
private readonly activeByIdentity = new Map<string, SessionUnreadIdentity>();
|
||||
private readonly excludedByIdentity = new Map<string, SessionUnreadIdentity>();
|
||||
private catalogId: string;
|
||||
private catalogRevision = 0;
|
||||
private nextCompletionOrder = 0;
|
||||
private persistenceWorker: Promise<void> | undefined;
|
||||
private pendingPersistence: PendingSessionUnreadPersistence | undefined;
|
||||
private persistenceGeneration = 0;
|
||||
private durablePersistenceGeneration = 0;
|
||||
private persistenceFailure: { error: unknown } | undefined;
|
||||
private loadPromise: Promise<void> | undefined;
|
||||
private loaded: boolean;
|
||||
|
||||
constructor(options: SessionUnreadStoreOptions = {}) {
|
||||
this.now = options.now ?? (() => new Date());
|
||||
this.persistence = options.persistence;
|
||||
this.createCatalogId = options.createCatalogId ?? randomUUID;
|
||||
this.onPersistenceError = options.onPersistenceError ?? (() => undefined);
|
||||
this.loaded = this.persistence === undefined;
|
||||
// A persisted store receives its epoch from disk or creates one during
|
||||
// load; synchronous in-memory stores are ready immediately.
|
||||
this.catalogId = this.loaded ? this.freshCatalogId() : "";
|
||||
}
|
||||
|
||||
load(): Promise<void> {
|
||||
if (this.loaded) return Promise.resolve();
|
||||
if (this.loadPromise !== undefined) return this.loadPromise;
|
||||
const loadPromise = this.loadPersistedState();
|
||||
this.loadPromise = loadPromise;
|
||||
return loadPromise;
|
||||
}
|
||||
|
||||
/** Current in-memory state. Transport boundaries should use `durableCatalogSnapshot`. */
|
||||
catalogSnapshot(): SessionUnreadCatalogSnapshot {
|
||||
this.requireLoaded();
|
||||
return {
|
||||
catalogId: this.catalogId,
|
||||
catalogRevision: this.catalogRevision,
|
||||
sessions: [...this.unreadByIdentity.values()]
|
||||
.sort((left, right) => right.completionOrder - left.completionOrder)
|
||||
.map((summary) => ({ ...summary })),
|
||||
};
|
||||
}
|
||||
|
||||
observeActivityState(sessionId: string, cwd: string, active: boolean): SessionUnreadMutation[] {
|
||||
this.requireLoaded();
|
||||
const identity = requireIdentity(sessionId, cwd);
|
||||
const key = sessionIdentityKey(identity);
|
||||
if (this.excludedByIdentity.has(key)) {
|
||||
this.activeByIdentity.delete(key);
|
||||
return [];
|
||||
}
|
||||
if (active) {
|
||||
this.activeByIdentity.set(key, identity);
|
||||
return [];
|
||||
}
|
||||
if (!this.activeByIdentity.has(key)) return [];
|
||||
|
||||
const completionOrder = incrementSafe(this.nextCompletionOrder, "Session unread completion order exhausted");
|
||||
const willExceedLimit = !this.unreadByIdentity.has(key) && this.unreadByIdentity.size >= SESSION_UNREAD_LIMIT;
|
||||
this.assertRevisionCapacity(willExceedLimit ? 2 : 1);
|
||||
const completedAt = this.now().toISOString();
|
||||
|
||||
this.activeByIdentity.delete(key);
|
||||
this.nextCompletionOrder = completionOrder;
|
||||
const summary: SessionUnreadSummary = {
|
||||
sessionId,
|
||||
cwd,
|
||||
completionOrder,
|
||||
completedAt,
|
||||
};
|
||||
// Reinsert existing identities so map order remains completion order.
|
||||
this.unreadByIdentity.delete(key);
|
||||
this.unreadByIdentity.set(key, summary);
|
||||
const mutations = [this.mutation(identity, summary), ...this.trimToLimit()];
|
||||
this.schedulePersist();
|
||||
return mutations;
|
||||
}
|
||||
|
||||
/** Clear only the transient active latch for a runtime that is closing or rebinding. */
|
||||
forgetActivity(sessionId: string, cwd: string): void {
|
||||
this.requireLoaded();
|
||||
const identity = requireIdentity(sessionId, cwd);
|
||||
this.activeByIdentity.delete(sessionIdentityKey(identity));
|
||||
}
|
||||
|
||||
/**
|
||||
* Suppress unread tracking until this identity is explicitly forgotten and
|
||||
* remove state recorded before it was verified as a tracked sub-session.
|
||||
*/
|
||||
excludeSession(sessionId: string, cwd: string): SessionUnreadMutation[] {
|
||||
this.requireLoaded();
|
||||
const identity = requireIdentity(sessionId, cwd);
|
||||
const key = sessionIdentityKey(identity);
|
||||
const current = this.unreadByIdentity.get(key);
|
||||
this.assertRevisionCapacity(current === undefined ? 0 : 1);
|
||||
|
||||
this.excludedByIdentity.set(key, identity);
|
||||
this.activeByIdentity.delete(key);
|
||||
if (current === undefined) return [];
|
||||
|
||||
this.unreadByIdentity.delete(key);
|
||||
const mutations = [this.mutation(identity, null)];
|
||||
this.schedulePersist();
|
||||
return mutations;
|
||||
}
|
||||
|
||||
acknowledge(sessionId: string, request: SessionUnreadAcknowledgeRequest): SessionUnreadAcknowledgeResult {
|
||||
this.requireLoaded();
|
||||
const identity = requireIdentity(sessionId, request.cwd);
|
||||
requireCatalogId(request.catalogId);
|
||||
requirePositiveSafeInteger(request.throughCompletionOrder, "throughCompletionOrder");
|
||||
if (request.catalogId !== this.catalogId) return { mutations: [] };
|
||||
|
||||
const key = sessionIdentityKey(identity);
|
||||
const current = this.unreadByIdentity.get(key);
|
||||
if (current === undefined || current.completionOrder > request.throughCompletionOrder) {
|
||||
return { mutations: [] };
|
||||
}
|
||||
|
||||
this.assertRevisionCapacity(1);
|
||||
this.unreadByIdentity.delete(key);
|
||||
const mutations = [this.mutation(identity, null)];
|
||||
this.schedulePersist();
|
||||
return { mutations };
|
||||
}
|
||||
|
||||
/** Remove durable unread and all transient lifecycle state for one identity. */
|
||||
forgetSession(sessionId: string, cwd: string): SessionUnreadMutation[] {
|
||||
this.requireLoaded();
|
||||
const identity = requireIdentity(sessionId, cwd);
|
||||
const key = sessionIdentityKey(identity);
|
||||
const current = this.unreadByIdentity.get(key);
|
||||
this.assertRevisionCapacity(current === undefined ? 0 : 1);
|
||||
|
||||
this.activeByIdentity.delete(key);
|
||||
this.excludedByIdentity.delete(key);
|
||||
if (current === undefined) return [];
|
||||
|
||||
this.unreadByIdentity.delete(key);
|
||||
const mutations = [this.mutation(identity, null)];
|
||||
this.schedulePersist();
|
||||
return mutations;
|
||||
}
|
||||
|
||||
reconcileCwd(cwd: string, sessionIds: Iterable<string>): SessionUnreadMutation[] {
|
||||
this.requireLoaded();
|
||||
const boundedCwd = requireBoundedNonEmptyString(cwd, "cwd", SESSION_UNREAD_CWD_MAX_LENGTH);
|
||||
const retained = new Set(sessionIds);
|
||||
const removed = [...this.unreadByIdentity.entries()]
|
||||
.filter(([, summary]) => summary.cwd === boundedCwd && !retained.has(summary.sessionId));
|
||||
this.assertRevisionCapacity(removed.length);
|
||||
|
||||
for (const [key, identity] of this.activeByIdentity) {
|
||||
if (identity.cwd === boundedCwd && !retained.has(identity.sessionId)) this.activeByIdentity.delete(key);
|
||||
}
|
||||
for (const [key, identity] of this.excludedByIdentity) {
|
||||
if (identity.cwd === boundedCwd && !retained.has(identity.sessionId)) this.excludedByIdentity.delete(key);
|
||||
}
|
||||
|
||||
const mutations: SessionUnreadMutation[] = [];
|
||||
for (const [key, summary] of removed) {
|
||||
this.unreadByIdentity.delete(key);
|
||||
mutations.push(this.mutation(summary, null));
|
||||
}
|
||||
if (mutations.length > 0) this.schedulePersist();
|
||||
return mutations;
|
||||
}
|
||||
|
||||
/** Wait until all currently queued state is durably represented or throw. */
|
||||
async flush(): Promise<void> {
|
||||
this.requireLoaded();
|
||||
await this.waitForPersistenceWorker();
|
||||
if (this.persistenceFailure === undefined
|
||||
&& this.persistenceGeneration === this.durablePersistenceGeneration) return;
|
||||
|
||||
// Retry the latest complete snapshot once. Failed/intermediate snapshots
|
||||
// are coalesced because completion orders and revisions are cumulative.
|
||||
this.ensurePersistenceWorker();
|
||||
await this.waitForPersistenceWorker();
|
||||
this.throwIfPersistenceFailed();
|
||||
}
|
||||
|
||||
/** Snapshot safe to expose to a client that may subsequently acknowledge it. */
|
||||
async durableCatalogSnapshot(): Promise<SessionUnreadCatalogSnapshot> {
|
||||
let snapshot: SessionUnreadCatalogSnapshot;
|
||||
do {
|
||||
await this.flush();
|
||||
snapshot = this.catalogSnapshot();
|
||||
} while (this.persistenceGeneration !== this.durablePersistenceGeneration);
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
private async loadPersistedState(): Promise<void> {
|
||||
const persistence = this.persistence;
|
||||
if (persistence === undefined) {
|
||||
this.loaded = true;
|
||||
return;
|
||||
}
|
||||
|
||||
let value: unknown;
|
||||
let resetState = false;
|
||||
try {
|
||||
value = await persistence.load();
|
||||
} catch (error: unknown) {
|
||||
this.reportPersistenceError("load", error);
|
||||
if (!(error instanceof SessionUnreadPersistenceCorruptionError)) throw error;
|
||||
resetState = true;
|
||||
}
|
||||
|
||||
if (!resetState && value !== undefined) {
|
||||
try {
|
||||
this.installPersistedState(parsePersistedState(value));
|
||||
} catch (error: unknown) {
|
||||
this.reportPersistenceError("load", error);
|
||||
resetState = true;
|
||||
}
|
||||
} else if (value === undefined) {
|
||||
resetState = true;
|
||||
}
|
||||
|
||||
if (resetState) {
|
||||
this.resetInMemoryState();
|
||||
// Persist even an empty epoch so the catalog identity itself survives a
|
||||
// clean daemon restart and a corrupt file is repaired once.
|
||||
this.schedulePersist();
|
||||
await this.waitForPersistenceWorker();
|
||||
try {
|
||||
this.throwIfPersistenceFailed();
|
||||
} catch (error: unknown) {
|
||||
this.loaded = false;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
this.loaded = true;
|
||||
}
|
||||
|
||||
private resetInMemoryState(): void {
|
||||
this.catalogId = this.freshCatalogId();
|
||||
this.catalogRevision = 0;
|
||||
this.nextCompletionOrder = 0;
|
||||
this.unreadByIdentity.clear();
|
||||
this.activeByIdentity.clear();
|
||||
this.excludedByIdentity.clear();
|
||||
}
|
||||
|
||||
private mutation(identity: SessionUnreadIdentity, unread: SessionUnreadSummary | null): SessionUnreadMutation {
|
||||
this.catalogRevision = incrementSafe(this.catalogRevision, "Session unread catalog revision exhausted");
|
||||
return {
|
||||
event: {
|
||||
type: "sessions.unread",
|
||||
catalogId: this.catalogId,
|
||||
catalogRevision: this.catalogRevision,
|
||||
sessionId: identity.sessionId,
|
||||
cwd: identity.cwd,
|
||||
unread: unread === null ? null : { ...unread },
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private trimToLimit(): SessionUnreadMutation[] {
|
||||
const mutations: SessionUnreadMutation[] = [];
|
||||
while (this.unreadByIdentity.size > SESSION_UNREAD_LIMIT) {
|
||||
let oldestKey: string | undefined;
|
||||
let oldest: SessionUnreadSummary | undefined;
|
||||
for (const [key, summary] of this.unreadByIdentity) {
|
||||
if (oldest === undefined || summary.completionOrder < oldest.completionOrder) {
|
||||
oldestKey = key;
|
||||
oldest = summary;
|
||||
}
|
||||
}
|
||||
if (oldestKey === undefined || oldest === undefined) break;
|
||||
this.unreadByIdentity.delete(oldestKey);
|
||||
mutations.push(this.mutation(oldest, null));
|
||||
}
|
||||
return mutations;
|
||||
}
|
||||
|
||||
private assertRevisionCapacity(count: number): void {
|
||||
if (!Number.isSafeInteger(this.catalogRevision + count)) {
|
||||
throw new Error("Session unread catalog revision exhausted");
|
||||
}
|
||||
}
|
||||
|
||||
private schedulePersist(): void {
|
||||
if (this.persistence === undefined) return;
|
||||
const generation = incrementSafe(this.persistenceGeneration, "Session unread persistence generation exhausted");
|
||||
this.persistenceGeneration = generation;
|
||||
this.pendingPersistence = { generation, state: this.persistedState() };
|
||||
// Once a save fails, mutations continue replacing the one pending snapshot
|
||||
// but do not hammer storage; the service's backoff (or an explicit flush)
|
||||
// owns the next retry attempt.
|
||||
if (this.persistenceFailure === undefined) this.ensurePersistenceWorker();
|
||||
}
|
||||
|
||||
private ensurePersistenceWorker(): void {
|
||||
if (this.persistence === undefined || this.persistenceWorker !== undefined || this.pendingPersistence === undefined) return;
|
||||
const worker = this.runPersistenceWorker();
|
||||
this.persistenceWorker = worker;
|
||||
void worker.finally(() => {
|
||||
if (this.persistenceWorker === worker) this.persistenceWorker = undefined;
|
||||
});
|
||||
}
|
||||
|
||||
private async runPersistenceWorker(): Promise<void> {
|
||||
const persistence = this.persistence;
|
||||
if (persistence === undefined) return;
|
||||
while (this.pendingPersistence !== undefined) {
|
||||
const pending = this.pendingPersistence;
|
||||
this.pendingPersistence = undefined;
|
||||
try {
|
||||
await persistence.save(pending.state);
|
||||
this.durablePersistenceGeneration = pending.generation;
|
||||
this.persistenceFailure = undefined;
|
||||
} catch (error: unknown) {
|
||||
// A newer pending snapshot subsumes this failed one. Otherwise retain
|
||||
// this exact snapshot so a later flush can retry without a new mutation.
|
||||
this.pendingPersistence ??= pending;
|
||||
this.persistenceFailure = { error };
|
||||
this.reportPersistenceError("save", error);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async waitForPersistenceWorker(): Promise<void> {
|
||||
let worker = this.persistenceWorker;
|
||||
while (worker !== undefined) {
|
||||
await worker;
|
||||
worker = this.persistenceWorker;
|
||||
}
|
||||
}
|
||||
|
||||
private throwIfPersistenceFailed(): void {
|
||||
const failure = this.persistenceFailure;
|
||||
if (failure !== undefined) throw failure.error;
|
||||
}
|
||||
|
||||
private persistedState(): SessionUnreadPersistedState {
|
||||
return {
|
||||
version: SESSION_UNREAD_STATE_VERSION,
|
||||
catalogId: this.catalogId,
|
||||
catalogRevision: this.catalogRevision,
|
||||
nextCompletionOrder: this.nextCompletionOrder,
|
||||
sessions: [...this.unreadByIdentity.values()].map((summary) => ({ ...summary })),
|
||||
};
|
||||
}
|
||||
|
||||
private installPersistedState(state: SessionUnreadPersistedState): void {
|
||||
this.catalogId = state.catalogId;
|
||||
this.catalogRevision = state.catalogRevision;
|
||||
this.nextCompletionOrder = state.nextCompletionOrder;
|
||||
this.unreadByIdentity.clear();
|
||||
for (const summary of [...state.sessions].sort((left, right) => left.completionOrder - right.completionOrder)) {
|
||||
this.unreadByIdentity.set(sessionIdentityKey(summary), { ...summary });
|
||||
}
|
||||
}
|
||||
|
||||
private freshCatalogId(): string {
|
||||
return requireCatalogId(this.createCatalogId());
|
||||
}
|
||||
|
||||
private requireLoaded(): void {
|
||||
if (!this.loaded) throw new Error("Session unread store must be loaded before use");
|
||||
}
|
||||
|
||||
private reportPersistenceError(operation: "load" | "save", error: unknown): void {
|
||||
try {
|
||||
this.onPersistenceError(operation, error);
|
||||
} catch {
|
||||
// Error reporting must not poison future serialized persistence work.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class SessionUnreadPersistenceCorruptionError extends Error {
|
||||
constructor(cause: unknown) {
|
||||
super("Session unread persistence contains invalid JSON", { cause });
|
||||
}
|
||||
}
|
||||
|
||||
export class FileSessionUnreadPersistence implements SessionUnreadPersistence {
|
||||
constructor(readonly filePath = defaultSessionUnreadFilePath()) {}
|
||||
|
||||
async load(): Promise<unknown> {
|
||||
let source: string;
|
||||
try {
|
||||
source = await readFile(this.filePath, "utf8");
|
||||
} catch (error: unknown) {
|
||||
if (isNodeError(error) && error.code === "ENOENT") return undefined;
|
||||
throw error;
|
||||
}
|
||||
try {
|
||||
const value: unknown = JSON.parse(source);
|
||||
return value;
|
||||
} catch (error: unknown) {
|
||||
throw new SessionUnreadPersistenceCorruptionError(error);
|
||||
}
|
||||
}
|
||||
|
||||
async save(state: SessionUnreadPersistedState): Promise<void> {
|
||||
await mkdir(dirname(this.filePath), { recursive: true });
|
||||
const tempPath = `${this.filePath}.${process.pid.toString()}-${randomUUID()}.tmp`;
|
||||
try {
|
||||
await writeFile(tempPath, `${JSON.stringify(state, null, 2)}\n`, {
|
||||
encoding: "utf8",
|
||||
mode: SESSION_UNREAD_FILE_MODE,
|
||||
flag: "wx",
|
||||
});
|
||||
await rename(tempPath, this.filePath);
|
||||
} finally {
|
||||
await rm(tempPath, { force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function defaultSessionUnreadFilePath(env: NodeJS.ProcessEnv = process.env, cwd = process.cwd()): string {
|
||||
return join(piWebDataDir(env, cwd), "session-unread.json");
|
||||
}
|
||||
|
||||
function parsePersistedState(value: unknown): SessionUnreadPersistedState {
|
||||
const record = requireRecord(value, "Session unread state must be an object");
|
||||
if (record["version"] !== SESSION_UNREAD_STATE_VERSION) throw new Error("Unsupported session unread state version");
|
||||
const catalogId = requireCatalogId(record["catalogId"]);
|
||||
const catalogRevision = requireNonNegativeSafeInteger(record["catalogRevision"], "catalogRevision");
|
||||
const nextCompletionOrder = requireNonNegativeSafeInteger(record["nextCompletionOrder"], "nextCompletionOrder");
|
||||
const rawSessions = record["sessions"];
|
||||
if (!Array.isArray(rawSessions)) throw new Error("Session unread sessions must be an array");
|
||||
if (rawSessions.length > SESSION_UNREAD_LIMIT) throw new Error("Session unread state exceeds its session limit");
|
||||
|
||||
const sessions = rawSessions.map(parseSummary);
|
||||
const identities = new Set<string>();
|
||||
const orders = new Set<number>();
|
||||
for (const summary of sessions) {
|
||||
const key = sessionIdentityKey(summary);
|
||||
if (identities.has(key)) throw new Error("Duplicate session unread identity");
|
||||
if (orders.has(summary.completionOrder)) throw new Error("Duplicate session unread completion order");
|
||||
identities.add(key);
|
||||
orders.add(summary.completionOrder);
|
||||
}
|
||||
const maxOrder = sessions.reduce((maximum, summary) => Math.max(maximum, summary.completionOrder), 0);
|
||||
if (nextCompletionOrder < maxOrder) throw new Error("Session unread completion order is inconsistent");
|
||||
if (catalogRevision < nextCompletionOrder) throw new Error("Session unread catalog revision is inconsistent");
|
||||
return {
|
||||
version: SESSION_UNREAD_STATE_VERSION,
|
||||
catalogId,
|
||||
catalogRevision,
|
||||
nextCompletionOrder,
|
||||
sessions,
|
||||
};
|
||||
}
|
||||
|
||||
function parseSummary(value: unknown): SessionUnreadSummary {
|
||||
const record = requireRecord(value, "Session unread summary must be an object");
|
||||
const completedAt = requireBoundedNonEmptyString(
|
||||
record["completedAt"],
|
||||
"completedAt",
|
||||
SESSION_UNREAD_COMPLETED_AT_MAX_LENGTH,
|
||||
);
|
||||
const completedDate = new Date(completedAt);
|
||||
if (!Number.isFinite(completedDate.getTime()) || completedDate.toISOString() !== completedAt) {
|
||||
throw new Error("Session unread completedAt must be a canonical ISO timestamp");
|
||||
}
|
||||
return {
|
||||
sessionId: requireBoundedNonEmptyString(
|
||||
record["sessionId"],
|
||||
"sessionId",
|
||||
SESSION_UNREAD_SESSION_ID_MAX_LENGTH,
|
||||
),
|
||||
cwd: requireBoundedNonEmptyString(record["cwd"], "cwd", SESSION_UNREAD_CWD_MAX_LENGTH),
|
||||
completionOrder: requirePositiveSafeInteger(record["completionOrder"], "completionOrder"),
|
||||
completedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function requireRecord(value: unknown, message: string): Record<string, unknown> {
|
||||
if (!isRecord(value)) throw new Error(message);
|
||||
return value;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function requireNonEmptyString(value: unknown, field: string): string {
|
||||
if (typeof value !== "string" || value === "") throw new Error(`Session unread ${field} must be a non-empty string`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function requireBoundedNonEmptyString(value: unknown, field: string, maxLength: number): string {
|
||||
const parsed = requireNonEmptyString(value, field);
|
||||
if (parsed.length > maxLength) throw new Error(`Session unread ${field} exceeds its length limit`);
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function requireCatalogId(value: unknown): string {
|
||||
return requireBoundedNonEmptyString(value, "catalogId", SESSION_UNREAD_CATALOG_ID_MAX_LENGTH);
|
||||
}
|
||||
|
||||
function requireNonNegativeSafeInteger(value: unknown, field: string): number {
|
||||
if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) {
|
||||
throw new Error(`Session unread ${field} must be a non-negative safe integer`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function requirePositiveSafeInteger(value: unknown, field: string): number {
|
||||
const parsed = requireNonNegativeSafeInteger(value, field);
|
||||
if (parsed === 0) throw new Error(`Session unread ${field} must be positive`);
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function requireIdentity(sessionId: string, cwd: string): SessionUnreadIdentity {
|
||||
return {
|
||||
sessionId: requireBoundedNonEmptyString(sessionId, "sessionId", SESSION_UNREAD_SESSION_ID_MAX_LENGTH),
|
||||
cwd: requireBoundedNonEmptyString(cwd, "cwd", SESSION_UNREAD_CWD_MAX_LENGTH),
|
||||
};
|
||||
}
|
||||
|
||||
function sessionIdentityKey(identity: SessionUnreadIdentity): string {
|
||||
return JSON.stringify([identity.sessionId, identity.cwd]);
|
||||
}
|
||||
|
||||
function incrementSafe(value: number, message: string): number {
|
||||
const next = value + 1;
|
||||
if (!Number.isSafeInteger(next)) throw new Error(message);
|
||||
return next;
|
||||
}
|
||||
|
||||
function isNodeError(error: unknown): error is NodeJS.ErrnoException {
|
||||
return error instanceof Error && "code" in error;
|
||||
}
|
||||
Reference in New Issue
Block a user