Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions docs/design/session-list-persisted-catalog-cache.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,11 @@ Organized and metadata-filtered session lists must load the complete persisted s

The daemon keeps a process-local catalog snapshot keyed by the resolved session runtime root, the exact workspace identity, and active versus archived state. Query, group, source, cursor, trust, and live-merge options are intentionally excluded because they do not change persisted catalog contents.

The first request installs an in-flight Promise before starting the loader. Concurrent requests for the same generation await that Promise. A successful catalog, including worktree sidecars, remains available for two seconds from scan completion. Organization and every live bridge field are merged after lookup on every request. Default numeric pagination and the separate session-info counter do not use the catalog cache.
The first request installs an in-flight load before starting the loader. Concurrent requests for the same generation attach independent waiter promises to that load. A successful catalog, including worktree sidecars, remains available for two seconds from scan completion. Organization and every live bridge field are merged after lookup on every request. Default numeric pagination and the separate session-info counter do not use the catalog cache.

Each physical load owns an `AbortController`; caller signals attach only to their waiter and are never combined directly with the loader signal. Cancelling one waiter rejects only that caller with its original reason. Settlement is first-wins: after the physical load outcome is accepted, a later caller abort cannot replace that result. When the last waiter cancels first, the cache aborts the physical load and synchronously detaches it from the slot, allowing a new request to start a replacement scan immediately. A detached load that ignores cancellation may still settle its existing promise, but identity and generation checks prevent it from installing a snapshot or changing the replacement load.

The core persisted scan checks cancellation around directory enumeration, sorting, JSONL reads, runtime membership reads, sidecar reads, and synchronous title extraction. When a signal is present, directory stat enumeration yields to the event loop every 128 entries so an HTTP disconnect can be observed; callers that do not pass a signal retain the original non-yielding path. REST request disconnects and ACP connection destruction cancel their own waiters. LiveTask callers do not pass a signal and remain non-cancellable waiters.

Each scope has a generation. Explicit metadata, close, delete, archive, and unarchive operations invalidate the affected states. An invalidated in-flight load may finish for requests that already joined it, but its generation cannot repopulate the cache. Failures are never cached and there is no stale-on-error fallback.

Expand All @@ -24,8 +28,8 @@ The cache is not a filesystem transaction. Unknown writers can update a file aft

## Observability

Request spans distinguish physical scans, cache hits, and single-flight waiters before awaiting the shared Promise, so failures retain their cache status. Successful lookups also record archive state, query kind, summary count, scan pages, truncation, and either leader scan duration or cache age. Paths, session identifiers, titles, and source identifiers are never attached.
Request spans distinguish physical scans, cache hits, and single-flight waiters before awaiting the shared load, so failures retain their cache status. Successful lookups also record archive state, query kind, summary count, scan pages, truncation, and either the physical scan duration for scan/single-flight waiters or cache age for cache hits. Paths, session identifiers, titles, and source identifiers are never attached.

## Out of scope

This change does not alter public protocols, Web Shell polling, the session-info scan, cross-workspace scan scheduling, core filesystem APIs, or daemon timeout policy. The outer lifecycle timeout fix remains necessary for a single slow cold scan.
This change does not alter public protocols, Web Shell polling, the session-info scan, cross-workspace scan scheduling, or daemon timeout policy. It does not add per-request ACP cancellation, fixed scan deadlines, asynchronous directory enumeration, concurrent stat calls, worker threads, or cancellation for CLI resume and picker callers.
41 changes: 27 additions & 14 deletions packages/cli/src/serve/acp-http/dispatch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1986,20 +1986,33 @@ export class AcpDispatcher {
if ('error' in parsedSource) {
throw new AcpParamError(parsedSource.error);
}
const result = await listWorkspaceSessionsForResponse(
this.bridge,
workspaceCwd,
{
cursor,
size: metaSize,
archiveState,
view,
group,
parentSessionId,
...parsedSource,
},
{ runtimeBaseDir: this.sessionRuntimeBaseDir },
);
let result: Awaited<
ReturnType<typeof listWorkspaceSessionsForResponse>
>;
try {
conn.abortSignal.throwIfAborted();
result = await listWorkspaceSessionsForResponse(
this.bridge,
workspaceCwd,
{
cursor,
size: metaSize,
archiveState,
view,
group,
parentSessionId,
...parsedSource,
},
{
runtimeBaseDir: this.sessionRuntimeBaseDir,
signal: conn.abortSignal,
},
Comment thread
doudouOUC marked this conversation as resolved.
);
conn.abortSignal.throwIfAborted();
} catch (error) {
Comment thread
doudouOUC marked this conversation as resolved.
if (conn.abortSignal.aborted) return;
throw error;
}
this.replyConn(conn, id, {
sessions: result.sessions.map((s) => ({
sessionId: s.sessionId,
Expand Down
130 changes: 130 additions & 0 deletions packages/cli/src/serve/acp-http/transport.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3257,6 +3257,136 @@ describe('ACP Streamable HTTP transport (over the wire)', () => {
});
});

it('cancels session/list without buffering an error when the connection is destroyed', async () => {
let loadSignal: AbortSignal | undefined;
const listSessionsSpy = vi
.spyOn(SessionService.prototype, 'listSessions')
.mockImplementation(async (options) => {
const signal = options?.signal;
expect(signal).toBeDefined();
loadSignal = signal;
await new Promise<void>((_resolve, reject) => {
if (signal?.aborted) {
reject(signal.reason);
return;
}
signal?.addEventListener('abort', () => reject(signal.reason), {
once: true,
});
});
return { items: [], nextCursor: undefined, hasMore: false };
});

try {
const connId = await initialize();
const conn = acpHandle?.registry.get(connId);
expect(conn).toBeDefined();
const bufferedBefore = conn!.getDiagnostic().bufferedConnectionFrames;
const logCallsBefore = stdioMocks.writeStderrLine.mock.calls.length;
await post(connId, {
jsonrpc: '2.0',
id: 130,
method: 'session/list',
params: { workspaceCwd: TEST_WORKSPACE, view: 'organized' },
});
await waitUntil(() => loadSignal !== undefined);

const deleted = await fetch(`${base}/acp`, {
method: 'DELETE',
headers: { 'acp-connection-id': connId },
});
expect(deleted.status).toBe(202);
await waitUntil(() => loadSignal?.aborted === true);
await new Promise<void>((resolve) => setImmediate(resolve));

expect(conn!.abortSignal.aborted).toBe(true);
expect(conn!.getDiagnostic().bufferedConnectionFrames).toBe(
bufferedBefore,
);
const cancellationLogs = stdioMocks.writeStderrLine.mock.calls
.slice(logCallsBefore)
.flat();
expect(
cancellationLogs.some(
(line) =>
typeof line === 'string' && line.includes('/acp dispatch error'),
),
).toBe(false);
} finally {
listSessionsSpy.mockRestore();
}
});

it('keeps a shared session/list scan alive when one connection is destroyed', async () => {
let loadSignal: AbortSignal | undefined;
let resolveScan!: (value: {
items: [];
nextCursor: undefined;
hasMore: false;
}) => void;
let rejectScan!: (reason?: unknown) => void;
const scan = new Promise<{
items: [];
nextCursor: undefined;
hasMore: false;
}>((resolve, reject) => {
resolveScan = resolve;
rejectScan = reject;
});
const listSessionsSpy = vi
.spyOn(SessionService.prototype, 'listSessions')
.mockImplementation(async (options) => {
const signal = options?.signal;
expect(signal).toBeDefined();
loadSignal = signal;
signal?.addEventListener('abort', () => rejectScan(signal.reason), {
once: true,
});
return scan;
});

try {
const firstConnId = await initialize();
const secondConnId = await initialize();
const secondStream = await openStream(secondConnId);
const secondFrame = takeFrames(secondStream, 1);
await Promise.all([
post(firstConnId, {
jsonrpc: '2.0',
id: 131,
method: 'session/list',
params: { workspaceCwd: TEST_WORKSPACE, view: 'organized' },
}),
post(secondConnId, {
jsonrpc: '2.0',
id: 132,
method: 'session/list',
params: { workspaceCwd: TEST_WORKSPACE, view: 'organized' },
}),
]);
await waitUntil(() => loadSignal !== undefined);
expect(listSessionsSpy).toHaveBeenCalledTimes(1);

const deleted = await fetch(`${base}/acp`, {
method: 'DELETE',
headers: { 'acp-connection-id': firstConnId },
});
expect(deleted.status).toBe(202);
expect(loadSignal?.aborted).toBe(false);

resolveScan({ items: [], nextCursor: undefined, hasMore: false });
await expect(secondFrame).resolves.toEqual([
expect.objectContaining({
id: 132,
result: expect.objectContaining({ sessions: [] }),
}),
]);
} finally {
resolveScan({ items: [], nextCursor: undefined, hasMore: false });
listSessionsSpy.mockRestore();
}
});

it('_qwen/sessions/archive rejects invalid batch params', async () => {
const connId = await initialize();
const connStream = await openStream(connId);
Expand Down
125 changes: 80 additions & 45 deletions packages/cli/src/serve/routes/session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -787,16 +787,22 @@ export function registerSessionRoutes(
return runtime;
};

const hasActivePersistedSessions = async (runtime: WorkspaceRuntime) => {
const hasActivePersistedSessions = async (
runtime: WorkspaceRuntime,
signal: AbortSignal,
) => {
try {
const page = await createWorkspaceRuntimeSessionService(
runtime,
).listSessions({
archiveState: 'active',
size: 1,
signal,
});
signal.throwIfAborted();
return page.items.length > 0;
} catch {
signal.throwIfAborted();
return false;
}
Comment thread
doudouOUC marked this conversation as resolved.
};
Expand Down Expand Up @@ -4371,53 +4377,82 @@ export function registerSessionRoutes(
...(parentSessionId !== undefined ? { parentSessionId } : {}),
...parsedSource,
};
// Organized/archived views always need the persisted store: organized
// cursors are opaque (non-numeric) and archived-only workspaces have no
// active persisted sessions, so the live-only fallback would drop them.
// Metadata filters gather the whole workspace
// (persisted + live) to filter completely and paginates with an opaque
// activity cursor, so the numeric-cursor live fallback can't serve it.
const usePersisted =
readOnlySecondary ||
runtime.primary ||
view === 'organized' ||
archiveState === 'archived' ||
parentSessionId !== undefined ||
parsedSource.sourceType !== undefined ||
(cursor !== undefined && cursor !== ''
? isNumericSessionCursor(cursor)
: await hasActivePersistedSessions(runtime));
// The live path only reads cursor/size; persisted-only options
// (organized view or archived state) would be silently dropped there.
// usePersisted already routes those to the persisted path — assert it so
// a future option added to validation but not to that gate fails loudly.
if (
!usePersisted &&
(view !== undefined ||
const controller = new AbortController();
const onRequestAborted = () => {
controller.abort(new DOMException('Request aborted', 'AbortError'));
};
Comment thread
doudouOUC marked this conversation as resolved.
Comment thread
doudouOUC marked this conversation as resolved.
const onResponseClosed = () => {
if (!res.writableEnded) {
controller.abort(new DOMException('Response closed', 'AbortError'));
}
Comment thread
doudouOUC marked this conversation as resolved.
Comment thread
doudouOUC marked this conversation as resolved.
};
req.once('aborted', onRequestAborted);
res.once('close', onResponseClosed);
if (req.aborted || res.destroyed) onRequestAborted();
Comment on lines +4389 to +4391

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks — not taking this one. An immediate Supertest abort would not prove the pre-check because the route may never run; a valid test needs contrived middleware, and this handler has no await before listener installation. That complexity is not justified as a Critical fix after the current review depth.

try {
// Organized/archived views always need the persisted store: organized
// cursors are opaque (non-numeric) and archived-only workspaces have no
// active persisted sessions, so the live-only fallback would drop them.
// Metadata filters gather the whole workspace
// (persisted + live) to filter completely and paginates with an opaque
// activity cursor, so the numeric-cursor live fallback can't serve it.
const usePersisted =
readOnlySecondary ||
runtime.primary ||
view === 'organized' ||
archiveState === 'archived' ||
parentSessionId !== undefined ||
parsedSource.sourceType !== undefined)
) {
throw new Error(
'session list live path received persisted-only options',
);
parsedSource.sourceType !== undefined ||
(cursor !== undefined && cursor !== ''
? isNumericSessionCursor(cursor)
: await hasActivePersistedSessions(runtime, controller.signal));
// The live path only reads cursor/size; persisted-only options
// (organized view or archived state) would be silently dropped there.
// usePersisted already routes those to the persisted path — assert it so
// a future option added to validation but not to that gate fails loudly.
if (
!usePersisted &&
(view !== undefined ||
archiveState === 'archived' ||
parentSessionId !== undefined ||
parsedSource.sourceType !== undefined)
) {
throw new Error(
'session list live path received persisted-only options',
);
}
const result = usePersisted
? await runWorkspaceInspectionWithLogPolicy(runtime, () =>
listWorkspaceSessionsForResponse(runtime.bridge, key, options, {
mergeLive: !readOnlySecondary,
runtimeBaseDir: runtime.sessionRuntimeBaseDir,
signal: controller.signal,
}),
Comment thread
doudouOUC marked this conversation as resolved.
)
: listLiveWorkspaceSessionsForResponse(
runtime.bridge,
key,
options,
);
controller.signal.throwIfAborted();
if (res.destroyed) return;
res.status(200).json({
sessions: result.sessions,
...(result.nextCursor != null
? { nextCursor: result.nextCursor }
: {}),
...(result.liveMergeFailed ? { liveMergeFailed: true } : {}),
...(result.truncated ? { truncated: true } : {}),
});
} catch (err) {
if (controller.signal.aborted || res.destroyed) {
return;
}
throw err;
Comment thread
doudouOUC marked this conversation as resolved.
} finally {
req.off('aborted', onRequestAborted);
res.off('close', onResponseClosed);
}
const result = usePersisted
? await runWorkspaceInspectionWithLogPolicy(runtime, () =>
listWorkspaceSessionsForResponse(runtime.bridge, key, options, {
mergeLive: !readOnlySecondary,
runtimeBaseDir: runtime.sessionRuntimeBaseDir,
}),
)
: listLiveWorkspaceSessionsForResponse(runtime.bridge, key, options);
res.status(200).json({
sessions: result.sessions,
...(result.nextCursor != null
? { nextCursor: result.nextCursor }
: {}),
...(result.liveMergeFailed ? { liveMergeFailed: true } : {}),
...(result.truncated ? { truncated: true } : {}),
});
} catch (err) {
if (err instanceof InvalidCursorError) {
res.status(400).json({
Expand Down
Loading
Loading