mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-09 17:19:02 +00:00
feat(cli): Add session owner index for workspace runtimes (#6540)
* feat(cli): Add session owner index for workspace runtimes Route live session ownership through a registry-backed owner index so multi-workspace sessions can resolve active sessions without scanning every bridge first. Expand trusted workspace load/resume and live read routing while keeping non-session surfaces primary-only. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(cli): avoid partial session owner index updates Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * test(cli): relax bridge wiring test timeout Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(cli): tighten workspace session owner routing Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(cli): normalize restore workspace mismatch handling Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(cli): record telemetry for workspace sessions alias Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(cli): preserve workspace selector error contract Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> --------- Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
jinye
GitHub
parent
87cad6f1ae
commit
393943daaf
17 changed files with 1643 additions and 340 deletions
|
|
@ -77,6 +77,81 @@ function deferred<T>(): {
|
|||
}
|
||||
|
||||
describe('createAcpSessionBridge', () => {
|
||||
it('emits lifecycle events when a fresh session is registered and closed', async () => {
|
||||
const events: Array<{
|
||||
type: string;
|
||||
sessionId: string;
|
||||
workspaceCwd: string;
|
||||
reason?: string;
|
||||
}> = [];
|
||||
const bridge = makeBridge({
|
||||
channelFactory: async () => makeChannel().channel,
|
||||
sessionLifecycle: (event) => {
|
||||
events.push(event);
|
||||
},
|
||||
});
|
||||
|
||||
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
|
||||
await bridge.closeSession(session.sessionId);
|
||||
|
||||
expect(events).toEqual([
|
||||
{
|
||||
type: 'registered',
|
||||
sessionId: session.sessionId,
|
||||
workspaceCwd: WS_A,
|
||||
reason: 'spawn',
|
||||
},
|
||||
{
|
||||
type: 'removed',
|
||||
sessionId: session.sessionId,
|
||||
workspaceCwd: WS_A,
|
||||
reason: 'client_close',
|
||||
},
|
||||
]);
|
||||
|
||||
await bridge.shutdown();
|
||||
});
|
||||
|
||||
it('does not emit another registration for attach-only spawnOrAttach calls', async () => {
|
||||
const events: Array<{ type: string; sessionId: string }> = [];
|
||||
const bridge = makeBridge({
|
||||
channelFactory: async () => makeChannel().channel,
|
||||
sessionLifecycle: (event) => {
|
||||
events.push(event);
|
||||
},
|
||||
});
|
||||
|
||||
const first = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
|
||||
const second = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
|
||||
|
||||
expect(second.sessionId).toBe(first.sessionId);
|
||||
expect(events.filter((event) => event.type === 'registered')).toEqual([
|
||||
expect.objectContaining({
|
||||
type: 'registered',
|
||||
sessionId: first.sessionId,
|
||||
}),
|
||||
]);
|
||||
|
||||
await bridge.closeSession(first.sessionId);
|
||||
await bridge.shutdown();
|
||||
});
|
||||
|
||||
it('does not fail session lifecycle when the lifecycle callback throws', async () => {
|
||||
const bridge = makeBridge({
|
||||
channelFactory: async () => makeChannel().channel,
|
||||
sessionLifecycle: () => {
|
||||
throw new Error('lifecycle sink failed');
|
||||
},
|
||||
});
|
||||
|
||||
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
|
||||
await expect(bridge.closeSession(session.sessionId)).resolves.toBe(
|
||||
undefined,
|
||||
);
|
||||
|
||||
await bridge.shutdown();
|
||||
});
|
||||
|
||||
it('accepts a valid BridgeOptions.eventRingSize at construction time', () => {
|
||||
// Smoke: positive finite integers are accepted; the underlying
|
||||
// EventBus ring-size threading is exercised end-to-end in
|
||||
|
|
|
|||
|
|
@ -107,6 +107,7 @@ import type {
|
|||
BridgeFreshSessionAdmissionContext,
|
||||
BridgeFreshSessionReservation,
|
||||
BridgeOptions,
|
||||
BridgeSessionLifecycleEvent,
|
||||
BridgeTelemetry,
|
||||
} from './bridgeOptions.js';
|
||||
import { MCP_RESTART_SERVER_DEADLINE_MS } from './mcpTimeouts.js';
|
||||
|
|
@ -1134,6 +1135,17 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
);
|
||||
}
|
||||
};
|
||||
const emitSessionLifecycle = (event: BridgeSessionLifecycleEvent): void => {
|
||||
try {
|
||||
opts.sessionLifecycle?.(event);
|
||||
} catch (err) {
|
||||
const message = `qwen serve: session lifecycle callback failed: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`;
|
||||
opts.onDiagnosticLine?.(message, 'warn');
|
||||
writeStderrLine(message);
|
||||
}
|
||||
};
|
||||
if (defaultSessionScope !== 'single' && defaultSessionScope !== 'thread') {
|
||||
throw new TypeError(
|
||||
`Invalid sessionScope: ${JSON.stringify(defaultSessionScope)}. ` +
|
||||
|
|
@ -1840,6 +1852,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
}
|
||||
byId.delete(sid);
|
||||
telemetry.metrics?.sessionLifecycle('die');
|
||||
emitSessionLifecycle({
|
||||
type: 'removed',
|
||||
sessionId: sid,
|
||||
workspaceCwd: sessEntry.workspaceCwd,
|
||||
reason: 'channel_closed',
|
||||
});
|
||||
// Tombstone the id so any late `extNotification` from the
|
||||
// dying child can't leak into the early-event buffer for a
|
||||
// future load/resume of the same persisted session id.
|
||||
|
|
@ -2649,7 +2667,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
sessionId: string,
|
||||
workspaceCwd: string,
|
||||
events = createSessionEventBus(),
|
||||
options: { drainEarlyEvents?: boolean } = {},
|
||||
options: { drainEarlyEvents?: boolean; lifecycleReason?: string } = {},
|
||||
): SessionEntry => {
|
||||
const entry: SessionEntry = {
|
||||
sessionId,
|
||||
|
|
@ -2679,6 +2697,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
byId.set(entry.sessionId, entry);
|
||||
touchActivity();
|
||||
telemetry.metrics?.sessionLifecycle('spawn');
|
||||
emitSessionLifecycle({
|
||||
type: 'registered',
|
||||
sessionId: entry.sessionId,
|
||||
workspaceCwd: entry.workspaceCwd,
|
||||
reason: options.lifecycleReason ?? 'spawn',
|
||||
});
|
||||
if (options.drainEarlyEvents !== false) {
|
||||
// Drain any guardrail events that fired during this session's
|
||||
// `newSession` handler (before this entry registered) onto the
|
||||
|
|
@ -3077,7 +3101,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
req.sessionId,
|
||||
workspaceKey,
|
||||
restoreEvents,
|
||||
{ drainEarlyEvents: replayUpdates.length === 0 },
|
||||
{
|
||||
drainEarlyEvents: replayUpdates.length === 0,
|
||||
lifecycleReason: action,
|
||||
},
|
||||
);
|
||||
releaseAdmissionOnce();
|
||||
entry.restoreState = state;
|
||||
|
|
@ -3129,9 +3156,16 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
if (!registeredEntry) {
|
||||
restoreEvents.close();
|
||||
let removedRestoreEntry = false;
|
||||
if (byId.get(req.sessionId)?.events === restoreEvents) {
|
||||
const restoreEntry = byId.get(req.sessionId);
|
||||
if (restoreEntry?.events === restoreEvents) {
|
||||
byId.delete(req.sessionId);
|
||||
ci?.sessionIds.delete(req.sessionId);
|
||||
emitSessionLifecycle({
|
||||
type: 'removed',
|
||||
sessionId: req.sessionId,
|
||||
workspaceCwd: restoreEntry.workspaceCwd,
|
||||
reason: 'restore_failed',
|
||||
});
|
||||
removedRestoreEntry = true;
|
||||
}
|
||||
if (removedRestoreEntry && ci && hasNoChannelWork(ci)) {
|
||||
|
|
@ -3230,6 +3264,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
}
|
||||
byId.delete(sessionId);
|
||||
telemetry.metrics?.sessionLifecycle('close');
|
||||
emitSessionLifecycle({
|
||||
type: 'removed',
|
||||
sessionId,
|
||||
workspaceCwd: entry.workspaceCwd,
|
||||
reason,
|
||||
});
|
||||
// Tombstone the closed sessionId so any late `extNotification`
|
||||
// from the (now-defunct) child can't seed the early-event buffer
|
||||
// and leak into a future load/resume of the same persisted id.
|
||||
|
|
@ -6156,6 +6196,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
if (defaultEntry === entry) defaultEntry = undefined;
|
||||
byId.delete(sessionId);
|
||||
telemetry.metrics?.sessionLifecycle('die');
|
||||
emitSessionLifecycle({
|
||||
type: 'removed',
|
||||
sessionId,
|
||||
workspaceCwd: entry.workspaceCwd,
|
||||
reason: 'killed',
|
||||
});
|
||||
// Detach from the channel. The channel dies only when its LAST
|
||||
// session leaves — other sessions on the same channel keep
|
||||
// running.
|
||||
|
|
@ -6282,8 +6328,17 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
cancelIdleTimer();
|
||||
stopSessionReaper();
|
||||
const channels = Array.from(aliveChannels);
|
||||
const entries = Array.from(byId.values());
|
||||
defaultEntry = undefined;
|
||||
byId.clear();
|
||||
for (const entry of entries) {
|
||||
emitSessionLifecycle({
|
||||
type: 'removed',
|
||||
sessionId: entry.sessionId,
|
||||
workspaceCwd: entry.workspaceCwd,
|
||||
reason: 'kill_all',
|
||||
});
|
||||
}
|
||||
for (const info of channels) {
|
||||
try {
|
||||
info.channel.killSync();
|
||||
|
|
@ -6335,6 +6390,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
|
|||
// and the automatic publish wouldn't fire.
|
||||
for (const e of entries) {
|
||||
telemetry.metrics?.sessionLifecycle('die');
|
||||
emitSessionLifecycle({
|
||||
type: 'removed',
|
||||
sessionId: e.sessionId,
|
||||
workspaceCwd: e.workspaceCwd,
|
||||
reason: 'daemon_shutdown',
|
||||
});
|
||||
try {
|
||||
e.events.publish({
|
||||
type: 'session_died',
|
||||
|
|
|
|||
|
|
@ -48,6 +48,24 @@ export type BridgeFreshSessionAdmission = (
|
|||
context: BridgeFreshSessionAdmissionContext,
|
||||
) => BridgeFreshSessionReservation | undefined;
|
||||
|
||||
export type BridgeSessionLifecycleEvent =
|
||||
| {
|
||||
readonly type: 'registered';
|
||||
readonly sessionId: string;
|
||||
readonly workspaceCwd: string;
|
||||
readonly reason: string;
|
||||
}
|
||||
| {
|
||||
readonly type: 'removed';
|
||||
readonly sessionId: string;
|
||||
readonly workspaceCwd: string;
|
||||
readonly reason: string;
|
||||
};
|
||||
|
||||
export type BridgeSessionLifecycle = (
|
||||
event: BridgeSessionLifecycleEvent,
|
||||
) => void;
|
||||
|
||||
/**
|
||||
* Optional injection seam for daemon-host-specific status cells —
|
||||
* `process.env` snapshots and the daemon-side preflight checks
|
||||
|
|
@ -168,6 +186,13 @@ export interface BridgeOptions {
|
|||
* side effect starts. Attaches bypass this hook.
|
||||
*/
|
||||
freshSessionAdmission?: BridgeFreshSessionAdmission;
|
||||
/**
|
||||
* Host-level live session owner callback. The bridge emits registration
|
||||
* only after a live entry is installed, and removal when that live entry is
|
||||
* removed. Callback failures are diagnostic only and do not fail session
|
||||
* lifecycle operations.
|
||||
*/
|
||||
sessionLifecycle?: BridgeSessionLifecycle;
|
||||
/**
|
||||
* Per-session SSE replay ring depth. Sets `ringSize` on every
|
||||
* `new EventBus(...)` the bridge constructs (both fresh sessions
|
||||
|
|
|
|||
|
|
@ -58,6 +58,8 @@ export type {
|
|||
BridgeFreshSessionAdmission,
|
||||
BridgeFreshSessionAdmissionContext,
|
||||
BridgeFreshSessionReservation,
|
||||
BridgeSessionLifecycle,
|
||||
BridgeSessionLifecycleEvent,
|
||||
BridgeOptions,
|
||||
DaemonStatusProvider,
|
||||
} from '@qwen-code/acp-bridge/bridgeOptions';
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@ interface FakeBridge extends AcpSessionBridge {
|
|||
action: 'load' | 'resume';
|
||||
req: BridgeRestoreSessionRequest;
|
||||
}>;
|
||||
readonly listCalls: string[];
|
||||
}
|
||||
|
||||
function makeSummary(
|
||||
|
|
@ -101,6 +102,7 @@ function makeBridge(
|
|||
const pendingPromptCalls: string[] = [];
|
||||
const removePendingPromptCalls: FakeBridge['removePendingPromptCalls'] = [];
|
||||
const restoreCalls: FakeBridge['restoreCalls'] = [];
|
||||
const listCalls: string[] = [];
|
||||
const bridge = {
|
||||
permissionPolicy: 'first-responder' as const,
|
||||
spawnCalls,
|
||||
|
|
@ -114,6 +116,7 @@ function makeBridge(
|
|||
pendingPromptCalls,
|
||||
removePendingPromptCalls,
|
||||
restoreCalls,
|
||||
listCalls,
|
||||
get sessionCount() {
|
||||
return live.size;
|
||||
},
|
||||
|
|
@ -186,6 +189,7 @@ function makeBridge(
|
|||
};
|
||||
},
|
||||
listWorkspaceSessions(cwd: string) {
|
||||
listCalls.push(cwd);
|
||||
return [...live.values()].filter(
|
||||
(summary) => summary.workspaceCwd === cwd,
|
||||
);
|
||||
|
|
@ -630,7 +634,7 @@ describe('multi-workspace session dispatch', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('keeps persisted load and resume primary-only before touching a bridge', async () => {
|
||||
it('dispatches trusted non-primary persisted load and resume', async () => {
|
||||
const { app, primaryBridge, secondaryBridge } = makeHarness();
|
||||
|
||||
for (const action of ['load', 'resume'] as const) {
|
||||
|
|
@ -639,13 +643,61 @@ describe('multi-workspace session dispatch', () => {
|
|||
.set('Host', host())
|
||||
.send({ cwd: SECONDARY_CWD });
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body.code).toBe('secondary_workspace_load_not_supported');
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.workspaceCwd).toBe(SECONDARY_CWD);
|
||||
}
|
||||
|
||||
expect(primaryBridge.restoreCalls).toEqual([]);
|
||||
expect(secondaryBridge.restoreCalls).toEqual([]);
|
||||
expect(secondaryBridge.restoreCalls).toEqual([
|
||||
{
|
||||
action: 'load',
|
||||
req: expect.objectContaining({
|
||||
sessionId: 'secondary-session',
|
||||
workspaceCwd: SECONDARY_CWD,
|
||||
}),
|
||||
},
|
||||
{
|
||||
action: 'resume',
|
||||
req: expect.objectContaining({
|
||||
sessionId: 'secondary-session',
|
||||
workspaceCwd: SECONDARY_CWD,
|
||||
}),
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('rejects unknown and untrusted restore cwd before touching a bridge', async () => {
|
||||
const unknown = makeHarness();
|
||||
const unknownRes = await request(unknown.app)
|
||||
.post('/session/unknown-restore/load')
|
||||
.set('Host', host())
|
||||
.send({ cwd: UNKNOWN_CWD });
|
||||
|
||||
expect(unknownRes.status).toBe(400);
|
||||
expect(unknownRes.body.code).toBe('workspace_mismatch');
|
||||
expect(unknownRes.body.workspaceCount).toBe(2);
|
||||
expect(unknown.primaryBridge.restoreCalls).toEqual([]);
|
||||
expect(unknown.secondaryBridge.restoreCalls).toEqual([]);
|
||||
|
||||
const daemonLog = makeDaemonLog();
|
||||
const untrusted = makeHarness({ secondaryTrusted: false, daemonLog });
|
||||
const untrustedRes = await request(untrusted.app)
|
||||
.post('/session/untrusted-restore/resume')
|
||||
.set('Host', host())
|
||||
.send({ cwd: SECONDARY_CWD });
|
||||
|
||||
expect(untrustedRes.status).toBe(403);
|
||||
expect(untrustedRes.body.code).toBe('untrusted_workspace');
|
||||
expect(untrusted.primaryBridge.restoreCalls).toEqual([]);
|
||||
expect(untrusted.secondaryBridge.restoreCalls).toEqual([]);
|
||||
expect(daemonLog.warn).toHaveBeenCalledWith(
|
||||
'session routing failed',
|
||||
expect.objectContaining({
|
||||
route: 'POST /session/:id/resume',
|
||||
resolutionKind: 'untrusted_workspace',
|
||||
workspaceCwd: SECONDARY_CWD,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('returns a clear Phase 2a error for non-primary sessions on primary-only routes', async () => {
|
||||
|
|
@ -692,6 +744,44 @@ describe('multi-workspace session dispatch', () => {
|
|||
expect(unknown.body.workspaceCount).toBe(2);
|
||||
});
|
||||
|
||||
it('preserves the legacy invalid workspace selector message', async () => {
|
||||
const { app } = makeHarness();
|
||||
|
||||
const res = await request(app)
|
||||
.get('/workspace/not:an:absolute:path/sessions')
|
||||
.set('Host', host());
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body.error).toBe(
|
||||
'`:id` must decode to a workspace id or absolute path',
|
||||
);
|
||||
});
|
||||
|
||||
it('rejects untrusted non-primary workspace session listing', async () => {
|
||||
const daemonLog = makeDaemonLog();
|
||||
const { app, secondaryBridge } = makeHarness({
|
||||
secondaryTrusted: false,
|
||||
daemonLog,
|
||||
});
|
||||
|
||||
const res = await request(app)
|
||||
.get('/workspaces/secondary-id/sessions')
|
||||
.set('Host', host());
|
||||
|
||||
expect(res.status).toBe(403);
|
||||
expect(res.body.code).toBe('untrusted_workspace');
|
||||
expect(res.body.workspaceCwd).toBe(SECONDARY_CWD);
|
||||
expect(secondaryBridge.listCalls).toEqual([]);
|
||||
expect(daemonLog.warn).toHaveBeenCalledWith(
|
||||
'session routing failed',
|
||||
expect.objectContaining({
|
||||
route: 'GET /workspaces/:workspace/sessions',
|
||||
resolutionKind: 'untrusted_workspace',
|
||||
workspaceCwd: SECONDARY_CWD,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('pages live non-primary workspace sessions with a stable cursor', async () => {
|
||||
const { app } = makeHarness({
|
||||
secondarySummaries: [
|
||||
|
|
|
|||
|
|
@ -219,6 +219,29 @@ describe('rateLimit', () => {
|
|||
expect(res.body).toMatchObject({ tier: 'read' });
|
||||
});
|
||||
|
||||
it('classifies plural workspace session listing as read tier', () => {
|
||||
const next = vi.fn();
|
||||
limiter.middleware(
|
||||
mockReq({
|
||||
method: 'GET',
|
||||
path: '/workspaces/ws-secondary/sessions',
|
||||
}),
|
||||
mockRes(),
|
||||
next,
|
||||
);
|
||||
expect(next).toHaveBeenCalled();
|
||||
const res = mockRes();
|
||||
limiter.middleware(
|
||||
mockReq({
|
||||
method: 'GET',
|
||||
path: '/workspaces/ws-secondary/sessions',
|
||||
}),
|
||||
res,
|
||||
vi.fn(),
|
||||
);
|
||||
expect(res.body).toMatchObject({ tier: 'read' });
|
||||
});
|
||||
|
||||
it('exempts GET /health', () => {
|
||||
const next = vi.fn();
|
||||
for (let i = 0; i < 5; i++) {
|
||||
|
|
|
|||
|
|
@ -223,13 +223,14 @@ export function registerSessionRoutes(
|
|||
const resolveRuntimeFromWorkspaceParam = (
|
||||
req: Request,
|
||||
res: Response,
|
||||
paramName = 'id',
|
||||
): WorkspaceRuntime | null => {
|
||||
const workspaceParam = req.params['id'] ?? '';
|
||||
const workspaceParam = req.params[paramName] ?? '';
|
||||
const byId = workspaceRegistry.getByWorkspaceId(workspaceParam);
|
||||
if (byId) return byId;
|
||||
if (!path.isAbsolute(workspaceParam)) {
|
||||
res.status(400).json({
|
||||
error: '`:id` must decode to a workspace id or absolute path',
|
||||
error: `\`:${paramName}\` must decode to a workspace id or absolute path`,
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
|
@ -248,35 +249,164 @@ export function registerSessionRoutes(
|
|||
return runtime;
|
||||
};
|
||||
|
||||
const resolvePrimaryOnlyWorkspaceCwd = (
|
||||
cwd: string,
|
||||
const sendAmbiguousSessionOwner = (
|
||||
res: Response,
|
||||
route: string,
|
||||
): string | undefined => {
|
||||
sessionId: string,
|
||||
runtimes: readonly WorkspaceRuntime[],
|
||||
): void => {
|
||||
const workspaceIds = runtimes.map((runtime) => runtime.workspaceId);
|
||||
logSessionRoutingFailure(route, 'ambiguous', {
|
||||
sessionId,
|
||||
workspaceIds,
|
||||
});
|
||||
res.status(500).json({
|
||||
error: `Session owner is ambiguous for "${sessionId}"`,
|
||||
code: 'ambiguous_session_owner',
|
||||
sessionId,
|
||||
route,
|
||||
workspaceIds,
|
||||
});
|
||||
};
|
||||
const inFlightRestoreOwners = new Map<
|
||||
string,
|
||||
{ workspaceCwd: string; count: number }
|
||||
>();
|
||||
|
||||
const sendSessionWorkspaceConflict = (
|
||||
res: Response,
|
||||
route: string,
|
||||
sessionId: string,
|
||||
runtime: WorkspaceRuntime,
|
||||
liveRuntime: WorkspaceRuntime,
|
||||
): void => {
|
||||
logSessionRoutingFailure(route, 'workspace_conflict', {
|
||||
sessionId,
|
||||
workspaceId: runtime.workspaceId,
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
liveWorkspaceId: liveRuntime.workspaceId,
|
||||
liveWorkspaceCwd: liveRuntime.workspaceCwd,
|
||||
});
|
||||
res.status(409).json({
|
||||
error: `Session "${sessionId}" is already live or restoring in another workspace runtime.`,
|
||||
code: 'session_workspace_conflict',
|
||||
sessionId,
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
workspaceId: runtime.workspaceId,
|
||||
liveWorkspaceCwd: liveRuntime.workspaceCwd,
|
||||
liveWorkspaceId: liveRuntime.workspaceId,
|
||||
});
|
||||
};
|
||||
|
||||
const enterRestoreOwner = (
|
||||
res: Response,
|
||||
route: string,
|
||||
sessionId: string,
|
||||
runtime: WorkspaceRuntime,
|
||||
): (() => void) | undefined => {
|
||||
const existing = inFlightRestoreOwners.get(sessionId);
|
||||
if (existing && existing.workspaceCwd !== runtime.workspaceCwd) {
|
||||
const existingRuntime =
|
||||
workspaceRegistry.getByWorkspaceCwd(existing.workspaceCwd) ??
|
||||
workspaceRegistry.primary;
|
||||
sendSessionWorkspaceConflict(
|
||||
res,
|
||||
route,
|
||||
sessionId,
|
||||
runtime,
|
||||
existingRuntime,
|
||||
);
|
||||
return undefined;
|
||||
}
|
||||
|
||||
if (existing) {
|
||||
existing.count += 1;
|
||||
} else {
|
||||
inFlightRestoreOwners.set(sessionId, {
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
count: 1,
|
||||
});
|
||||
}
|
||||
let released = false;
|
||||
return () => {
|
||||
if (released) return;
|
||||
released = true;
|
||||
const current = inFlightRestoreOwners.get(sessionId);
|
||||
if (!current || current.workspaceCwd !== runtime.workspaceCwd) return;
|
||||
current.count -= 1;
|
||||
if (current.count <= 0) {
|
||||
inFlightRestoreOwners.delete(sessionId);
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
const resolveRuntimeForSessionRestore = (
|
||||
body: Record<string, unknown>,
|
||||
res: Response,
|
||||
route: string,
|
||||
sessionId: string,
|
||||
): { runtime: WorkspaceRuntime; workspaceCwd: string } | undefined => {
|
||||
const cwd = parseOptionalWorkspaceCwd(body, boundWorkspace, res);
|
||||
if (cwd === undefined) return undefined;
|
||||
let key: string;
|
||||
try {
|
||||
key = canonicalizeWorkspace(cwd);
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, { route });
|
||||
return undefined;
|
||||
}
|
||||
if (key !== boundWorkspace) {
|
||||
const runtime = workspaceRegistry.getByWorkspaceCwd(key);
|
||||
if (runtime && !runtime.primary) {
|
||||
res.status(400).json({
|
||||
error:
|
||||
'Non-primary persisted session load/resume is not supported in Phase 2a.',
|
||||
code: 'secondary_workspace_load_not_supported',
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
workspaceId: runtime.workspaceId,
|
||||
route,
|
||||
if ('cwd' in body) {
|
||||
logSessionRoutingFailure(route, 'workspace_mismatch', {
|
||||
requestedWorkspace: cwd,
|
||||
});
|
||||
sendWorkspaceMismatch(res, cwd);
|
||||
return undefined;
|
||||
}
|
||||
sendBridgeError(res, err, { route, sessionId });
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const runtime = workspaceRegistry.resolveWorkspaceCwd(
|
||||
'cwd' in body ? key : undefined,
|
||||
);
|
||||
if (!runtime) {
|
||||
logSessionRoutingFailure(route, 'workspace_mismatch', {
|
||||
requestedWorkspace: key,
|
||||
});
|
||||
sendWorkspaceMismatch(res, key);
|
||||
return undefined;
|
||||
}
|
||||
return key;
|
||||
if (!runtime.primary && !runtime.trusted) {
|
||||
logSessionRoutingFailure(route, 'untrusted_workspace', {
|
||||
workspaceId: runtime.workspaceId,
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
});
|
||||
res.status(403).json({
|
||||
error: `Workspace "${runtime.workspaceCwd}" is not trusted.`,
|
||||
code: 'untrusted_workspace',
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
workspaceId: runtime.workspaceId,
|
||||
});
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const liveOwner = workspaceRegistry.resolveLiveSessionOwner(sessionId);
|
||||
if (liveOwner.kind === 'ambiguous') {
|
||||
sendAmbiguousSessionOwner(res, route, sessionId, liveOwner.runtimes);
|
||||
return undefined;
|
||||
}
|
||||
if (
|
||||
liveOwner.kind === 'found' &&
|
||||
liveOwner.runtime.workspaceCwd !== runtime.workspaceCwd
|
||||
) {
|
||||
sendSessionWorkspaceConflict(
|
||||
res,
|
||||
route,
|
||||
sessionId,
|
||||
runtime,
|
||||
liveOwner.runtime,
|
||||
);
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return { runtime, workspaceCwd: runtime.workspaceCwd };
|
||||
};
|
||||
|
||||
const resolveLiveSessionRuntime = (
|
||||
|
|
@ -321,9 +451,9 @@ export function registerSessionRoutes(
|
|||
async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
const runtime = resolveLiveSessionRuntime(sessionId, res, route);
|
||||
if (!runtime) return;
|
||||
try {
|
||||
const runtime = resolveLiveSessionRuntime(sessionId, res, route);
|
||||
if (!runtime) return;
|
||||
await archiveCoordinator.runSharedMany([sessionId], async () => {
|
||||
await handler(req, res, sessionId, runtime);
|
||||
});
|
||||
|
|
@ -332,6 +462,28 @@ export function registerSessionRoutes(
|
|||
}
|
||||
};
|
||||
|
||||
const withOwnerReadSession =
|
||||
(
|
||||
route: string,
|
||||
handler: (
|
||||
req: Request,
|
||||
res: Response,
|
||||
sessionId: string,
|
||||
runtime: WorkspaceRuntime,
|
||||
) => Promise<void> | void,
|
||||
): RequestHandler =>
|
||||
async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
const runtime = resolveLiveSessionRuntime(sessionId, res, route);
|
||||
if (!runtime) return;
|
||||
await handler(req, res, sessionId, runtime);
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, { route, sessionId });
|
||||
}
|
||||
};
|
||||
|
||||
const parseSessionIdsBody = (
|
||||
req: Request,
|
||||
res: Response,
|
||||
|
|
@ -523,31 +675,50 @@ export function registerSessionRoutes(
|
|||
const sessionId = requireSessionId(req, res);
|
||||
if (!sessionId) return;
|
||||
const body = safeBody(req);
|
||||
const cwd = parseOptionalWorkspaceCwd(body, boundWorkspace, res);
|
||||
if (cwd === undefined) return;
|
||||
const primaryCwd = resolvePrimaryOnlyWorkspaceCwd(
|
||||
cwd,
|
||||
const route = `POST /session/:id/${action}`;
|
||||
let resolvedRuntime:
|
||||
| { runtime: WorkspaceRuntime; workspaceCwd: string }
|
||||
| undefined;
|
||||
try {
|
||||
resolvedRuntime = resolveRuntimeForSessionRestore(
|
||||
body,
|
||||
res,
|
||||
route,
|
||||
sessionId,
|
||||
);
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, { route, sessionId });
|
||||
return;
|
||||
}
|
||||
if (resolvedRuntime === undefined) return;
|
||||
const { runtime, workspaceCwd } = resolvedRuntime;
|
||||
const releaseRestoreOwner = enterRestoreOwner(
|
||||
res,
|
||||
`POST /session/:id/${action}`,
|
||||
route,
|
||||
sessionId,
|
||||
runtime,
|
||||
);
|
||||
if (primaryCwd === undefined) return;
|
||||
if (!releaseRestoreOwner) return;
|
||||
const clientId = parseClientIdHeader(req, res);
|
||||
if (clientId === null) return;
|
||||
if (clientId === null) {
|
||||
releaseRestoreOwner();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const session = await archiveCoordinator.runSharedMany(
|
||||
[sessionId],
|
||||
async () => {
|
||||
await assertSessionLoadable(primaryCwd, sessionId);
|
||||
await assertSessionLoadable(workspaceCwd, sessionId);
|
||||
return action === 'load'
|
||||
? await bridge.loadSession({
|
||||
? await runtime.bridge.loadSession({
|
||||
sessionId,
|
||||
workspaceCwd: primaryCwd,
|
||||
workspaceCwd,
|
||||
historyReplay: 'response',
|
||||
...(clientId !== undefined ? { clientId } : {}),
|
||||
})
|
||||
: await bridge.resumeSession({
|
||||
: await runtime.bridge.resumeSession({
|
||||
sessionId,
|
||||
workspaceCwd: primaryCwd,
|
||||
workspaceCwd,
|
||||
...(clientId !== undefined ? { clientId } : {}),
|
||||
});
|
||||
},
|
||||
|
|
@ -568,13 +739,13 @@ export function registerSessionRoutes(
|
|||
// restored session in `byId` with no client holding its id.
|
||||
if (!res.writable) {
|
||||
if (!session.attached) {
|
||||
bridge
|
||||
runtime.bridge
|
||||
.killSession(session.sessionId, { requireZeroAttaches: true })
|
||||
.catch(() => {
|
||||
// Best-effort cleanup; channel.exited will eventually reap.
|
||||
});
|
||||
} else {
|
||||
bridge
|
||||
runtime.bridge
|
||||
.detachClient(session.sessionId, session.clientId)
|
||||
.catch(() => {
|
||||
// Best-effort cleanup; channel.exited will eventually reap.
|
||||
|
|
@ -585,9 +756,11 @@ export function registerSessionRoutes(
|
|||
res.status(200).json(session);
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: `POST /session/:id/${action}`,
|
||||
route,
|
||||
sessionId,
|
||||
});
|
||||
} finally {
|
||||
releaseRestoreOwner();
|
||||
}
|
||||
};
|
||||
|
||||
|
|
@ -753,113 +926,106 @@ export function registerSessionRoutes(
|
|||
}
|
||||
});
|
||||
|
||||
app.get('/session/:id/context', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res.status(200).json(await bridge.getSessionContextStatus(sessionId));
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: 'GET /session/:id/context',
|
||||
sessionId,
|
||||
});
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/context',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/context',
|
||||
async (_req, res, sessionId, runtime) => {
|
||||
res
|
||||
.status(200)
|
||||
.json(await runtime.bridge.getSessionContextStatus(sessionId));
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
app.get('/session/:id/context-usage', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res.status(200).json(
|
||||
await bridge.getSessionContextUsageStatus(sessionId, {
|
||||
detail: req.query['detail'] === 'true',
|
||||
}),
|
||||
);
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: 'GET /session/:id/context-usage',
|
||||
sessionId,
|
||||
});
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/context-usage',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/context-usage',
|
||||
async (req, res, sessionId, runtime) => {
|
||||
res.status(200).json(
|
||||
await runtime.bridge.getSessionContextUsageStatus(sessionId, {
|
||||
detail: req.query['detail'] === 'true',
|
||||
}),
|
||||
);
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
app.get('/session/:id/stats', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res.status(200).json(await bridge.getSessionStatsStatus(sessionId));
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: 'GET /session/:id/stats',
|
||||
sessionId,
|
||||
});
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/stats',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/stats',
|
||||
async (_req, res, sessionId, runtime) => {
|
||||
res
|
||||
.status(200)
|
||||
.json(await runtime.bridge.getSessionStatsStatus(sessionId));
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
app.get('/session/:id/supported-commands', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res
|
||||
.status(200)
|
||||
.json(await bridge.getSessionSupportedCommandsStatus(sessionId));
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: 'GET /session/:id/supported-commands',
|
||||
sessionId,
|
||||
});
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/supported-commands',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/supported-commands',
|
||||
async (_req, res, sessionId, runtime) => {
|
||||
res
|
||||
.status(200)
|
||||
.json(
|
||||
await runtime.bridge.getSessionSupportedCommandsStatus(sessionId),
|
||||
);
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
app.get('/session/:id/tasks', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res.status(200).json(await bridge.getSessionTasksStatus(sessionId));
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: 'GET /session/:id/tasks',
|
||||
sessionId,
|
||||
});
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/tasks',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/tasks',
|
||||
async (_req, res, sessionId, runtime) => {
|
||||
res
|
||||
.status(200)
|
||||
.json(await runtime.bridge.getSessionTasksStatus(sessionId));
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
app.get('/session/:id/lsp', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res.status(200).json(await bridge.getSessionLspStatus(sessionId));
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: 'GET /session/:id/lsp',
|
||||
sessionId,
|
||||
});
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/lsp',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/lsp',
|
||||
async (_req, res, sessionId, runtime) => {
|
||||
res
|
||||
.status(200)
|
||||
.json(await runtime.bridge.getSessionLspStatus(sessionId));
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
// GET /session/:id/hooks — read-only session-scoped hook status.
|
||||
app.get('/session/:id/hooks', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res.status(200).json(await bridge.getSessionHooksStatus(sessionId));
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, { route: 'GET /session/:id/hooks', sessionId });
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/hooks',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/hooks',
|
||||
async (_req, res, sessionId, runtime) => {
|
||||
res
|
||||
.status(200)
|
||||
.json(await runtime.bridge.getSessionHooksStatus(sessionId));
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
app.get('/session/:id/artifacts', async (req, res) => {
|
||||
const sessionId = requireSessionId(req, res);
|
||||
if (sessionId === null) return;
|
||||
try {
|
||||
res.status(200).json(await bridge.getSessionArtifacts(sessionId));
|
||||
} catch (err) {
|
||||
sendBridgeError(res, err, {
|
||||
route: 'GET /session/:id/artifacts',
|
||||
sessionId,
|
||||
});
|
||||
}
|
||||
});
|
||||
app.get(
|
||||
'/session/:id/artifacts',
|
||||
withOwnerReadSession(
|
||||
'GET /session/:id/artifacts',
|
||||
async (_req, res, sessionId, runtime) => {
|
||||
res
|
||||
.status(200)
|
||||
.json(await runtime.bridge.getSessionArtifacts(sessionId));
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
app.post(
|
||||
'/session/:id/artifacts',
|
||||
|
|
@ -1516,99 +1682,128 @@ export function registerSessionRoutes(
|
|||
},
|
||||
);
|
||||
|
||||
app.get('/workspace/:id/sessions', async (req, res) => {
|
||||
// Express decodes URL-encoded path params automatically; clients pass
|
||||
// the absolute workspace cwd encoded (e.g.
|
||||
// GET /workspace/%2Fwork%2Fa/sessions).
|
||||
const runtime = resolveRuntimeFromWorkspaceParam(req, res);
|
||||
if (runtime === null) return;
|
||||
const key = runtime.workspaceCwd;
|
||||
try {
|
||||
const cursor =
|
||||
typeof req.query['cursor'] === 'string'
|
||||
? req.query['cursor']
|
||||
: undefined;
|
||||
const size = parseSessionPageSizeQuery(req.query['size']);
|
||||
const rawView = req.query['view'];
|
||||
let view: 'organized' | undefined;
|
||||
if (rawView !== undefined) {
|
||||
if (rawView !== 'organized') {
|
||||
const listWorkspaceSessionsHandler =
|
||||
(paramName: string): RequestHandler =>
|
||||
async (req, res) => {
|
||||
const route =
|
||||
paramName === 'workspace'
|
||||
? 'GET /workspaces/:workspace/sessions'
|
||||
: 'GET /workspace/:id/sessions';
|
||||
// Express decodes URL-encoded path params automatically; clients pass
|
||||
// the absolute workspace cwd encoded (e.g.
|
||||
// GET /workspace/%2Fwork%2Fa/sessions).
|
||||
const runtime = resolveRuntimeFromWorkspaceParam(req, res, paramName);
|
||||
if (runtime === null) return;
|
||||
if (!runtime.primary && !runtime.trusted) {
|
||||
logSessionRoutingFailure(route, 'untrusted_workspace', {
|
||||
workspaceId: runtime.workspaceId,
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
});
|
||||
res.status(403).json({
|
||||
error: `Workspace "${runtime.workspaceCwd}" is not trusted.`,
|
||||
code: 'untrusted_workspace',
|
||||
workspaceCwd: runtime.workspaceCwd,
|
||||
workspaceId: runtime.workspaceId,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const key = runtime.workspaceCwd;
|
||||
try {
|
||||
const cursor =
|
||||
typeof req.query['cursor'] === 'string'
|
||||
? req.query['cursor']
|
||||
: undefined;
|
||||
const size = parseSessionPageSizeQuery(req.query['size']);
|
||||
const rawView = req.query['view'];
|
||||
let view: 'organized' | undefined;
|
||||
if (rawView !== undefined) {
|
||||
if (rawView !== 'organized') {
|
||||
res.status(400).json({
|
||||
error: '`view` must be "organized"',
|
||||
code: 'invalid_session_view',
|
||||
});
|
||||
return;
|
||||
}
|
||||
view = 'organized';
|
||||
}
|
||||
const group =
|
||||
typeof req.query['group'] === 'string'
|
||||
? req.query['group']
|
||||
: undefined;
|
||||
if (group !== undefined && view !== 'organized') {
|
||||
res.status(400).json({
|
||||
error: '`view` must be "organized"',
|
||||
code: 'invalid_session_view',
|
||||
error: '`group` requires `view=organized`',
|
||||
code: 'invalid_session_group_filter',
|
||||
});
|
||||
return;
|
||||
}
|
||||
view = 'organized';
|
||||
}
|
||||
const group =
|
||||
typeof req.query['group'] === 'string' ? req.query['group'] : undefined;
|
||||
if (group !== undefined && view !== 'organized') {
|
||||
res.status(400).json({
|
||||
error: '`group` requires `view=organized`',
|
||||
code: 'invalid_session_group_filter',
|
||||
});
|
||||
return;
|
||||
}
|
||||
const rawArchiveState = req.query['archiveState'];
|
||||
let archiveState: SessionArchiveState | undefined;
|
||||
if (rawArchiveState !== undefined) {
|
||||
if (
|
||||
typeof rawArchiveState !== 'string' ||
|
||||
(rawArchiveState !== 'active' && rawArchiveState !== 'archived')
|
||||
) {
|
||||
const rawArchiveState = req.query['archiveState'];
|
||||
let archiveState: SessionArchiveState | undefined;
|
||||
if (rawArchiveState !== undefined) {
|
||||
if (
|
||||
typeof rawArchiveState !== 'string' ||
|
||||
(rawArchiveState !== 'active' && rawArchiveState !== 'archived')
|
||||
) {
|
||||
res.status(400).json({
|
||||
error: '`archiveState` must be "active" or "archived"',
|
||||
code: 'invalid_archive_state',
|
||||
});
|
||||
return;
|
||||
}
|
||||
archiveState = rawArchiveState;
|
||||
}
|
||||
if (!runtime.primary && (archiveState === 'archived' || view)) {
|
||||
res.status(400).json({
|
||||
error: '`archiveState` must be "active" or "archived"',
|
||||
code: 'invalid_archive_state',
|
||||
error:
|
||||
'Non-primary workspace session listing is live-only in Phase 2a.',
|
||||
code: 'non_primary_live_sessions_only',
|
||||
});
|
||||
return;
|
||||
}
|
||||
archiveState = rawArchiveState;
|
||||
}
|
||||
if (!runtime.primary && (archiveState === 'archived' || view)) {
|
||||
res.status(400).json({
|
||||
error:
|
||||
'Non-primary workspace session listing is live-only in Phase 2a.',
|
||||
code: 'non_primary_live_sessions_only',
|
||||
const options = {
|
||||
...(cursor !== undefined ? { cursor } : {}),
|
||||
...(size !== undefined ? { size } : {}),
|
||||
...(archiveState !== undefined ? { archiveState } : {}),
|
||||
...(view !== undefined ? { view } : {}),
|
||||
...(group !== undefined ? { group } : {}),
|
||||
};
|
||||
const result = runtime.primary
|
||||
? await listWorkspaceSessionsForResponse(runtime.bridge, key, options)
|
||||
: 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 } : {}),
|
||||
});
|
||||
return;
|
||||
}
|
||||
const options = {
|
||||
...(cursor !== undefined ? { cursor } : {}),
|
||||
...(size !== undefined ? { size } : {}),
|
||||
...(archiveState !== undefined ? { archiveState } : {}),
|
||||
...(view !== undefined ? { view } : {}),
|
||||
...(group !== undefined ? { group } : {}),
|
||||
};
|
||||
const result = runtime.primary
|
||||
? await listWorkspaceSessionsForResponse(runtime.bridge, key, options)
|
||||
: 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({
|
||||
error: err.message,
|
||||
code: 'invalid_cursor',
|
||||
} catch (err) {
|
||||
if (err instanceof InvalidCursorError) {
|
||||
res.status(400).json({
|
||||
error: err.message,
|
||||
code: 'invalid_cursor',
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (sendSessionOrganizationError(res, err)) return;
|
||||
writeStderrLine(
|
||||
`qwen serve: failed to list sessions for workspace ${safeLogValue(
|
||||
key,
|
||||
)}: ${safeLogValue(err instanceof Error ? err.message : String(err))}`,
|
||||
);
|
||||
res.status(500).json({
|
||||
error: 'Failed to list sessions',
|
||||
code: 'session_list_failed',
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (sendSessionOrganizationError(res, err)) return;
|
||||
writeStderrLine(
|
||||
`qwen serve: failed to list sessions for workspace ${safeLogValue(
|
||||
key,
|
||||
)}: ${safeLogValue(err instanceof Error ? err.message : String(err))}`,
|
||||
);
|
||||
res.status(500).json({
|
||||
error: 'Failed to list sessions',
|
||||
code: 'session_list_failed',
|
||||
});
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
app.get('/workspace/:id/sessions', listWorkspaceSessionsHandler('id'));
|
||||
app.get(
|
||||
'/workspaces/:workspace/sessions',
|
||||
listWorkspaceSessionsHandler('workspace'),
|
||||
);
|
||||
|
||||
app.post(
|
||||
'/session/:id/model',
|
||||
|
|
|
|||
|
|
@ -799,6 +799,8 @@ async function loadServeRuntimeModules() {
|
|||
createTotalSessionAdmissionController:
|
||||
totalSessionAdmissionModule.createTotalSessionAdmissionController,
|
||||
createWorkspaceRegistry: workspaceRegistryModule.createWorkspaceRegistry,
|
||||
createWorkspaceSessionOwnerIndex:
|
||||
workspaceRegistryModule.createWorkspaceSessionOwnerIndex,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -2456,6 +2458,7 @@ export async function runQwenServe(
|
|||
: [],
|
||||
},
|
||||
);
|
||||
const sessionOwnerIndex = runtime.createWorkspaceSessionOwnerIndex();
|
||||
const persistDisabledToolsFn = (
|
||||
workspace: string,
|
||||
toolName: string,
|
||||
|
|
@ -2536,6 +2539,7 @@ export async function runQwenServe(
|
|||
clientMcpSender: clientMcpSenderRegistry.lookup,
|
||||
maxSessions: opts.maxSessions,
|
||||
freshSessionAdmission: totalSessionAdmission.admit,
|
||||
sessionLifecycle: sessionOwnerIndex.handleBridgeSessionLifecycle,
|
||||
...(opts.maxPendingPromptsPerSession !== undefined
|
||||
? { maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession }
|
||||
: {}),
|
||||
|
|
@ -2803,6 +2807,7 @@ export async function runQwenServe(
|
|||
clientMcpSender: secondaryClientMcpSenderRegistry.lookup,
|
||||
maxSessions: opts.maxSessions,
|
||||
freshSessionAdmission: totalSessionAdmission.admit,
|
||||
sessionLifecycle: sessionOwnerIndex.handleBridgeSessionLifecycle,
|
||||
...(opts.maxPendingPromptsPerSession !== undefined
|
||||
? { maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession }
|
||||
: {}),
|
||||
|
|
@ -2933,7 +2938,9 @@ export async function runQwenServe(
|
|||
}
|
||||
|
||||
const workspaceRegistry: WorkspaceRegistry =
|
||||
runtime.createWorkspaceRegistry(workspaceRuntimes);
|
||||
runtime.createWorkspaceRegistry(workspaceRuntimes, {
|
||||
sessionOwnerIndex,
|
||||
});
|
||||
|
||||
core.registerDaemonGaugeCallbacks({
|
||||
sessionCount: () =>
|
||||
|
|
|
|||
153
packages/cli/src/serve/server-default-bridge-wiring.test.ts
Normal file
153
packages/cli/src/serve/server-default-bridge-wiring.test.ts
Normal file
|
|
@ -0,0 +1,153 @@
|
|||
/**
|
||||
* @license
|
||||
* Copyright 2026 Qwen Team
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import {
|
||||
SessionNotFoundError,
|
||||
type AcpSessionBridge,
|
||||
type BridgeFreshSessionAdmission,
|
||||
type BridgeOptions,
|
||||
type BridgeSessionSummary,
|
||||
} from './acp-session-bridge.js';
|
||||
import type { WorkspaceRegistry } from './workspace-registry.js';
|
||||
|
||||
const WS_BOUND = '/work/bound';
|
||||
|
||||
function makeBridge(
|
||||
sessionCount = 0,
|
||||
liveSessionIds?: ReadonlySet<string>,
|
||||
): AcpSessionBridge {
|
||||
const getSessionSummary = (sessionId: string): BridgeSessionSummary => {
|
||||
if (liveSessionIds && !liveSessionIds.has(sessionId)) {
|
||||
throw new SessionNotFoundError(sessionId);
|
||||
}
|
||||
return {
|
||||
sessionId,
|
||||
workspaceCwd: WS_BOUND,
|
||||
createdAt: '2026-05-17T12:00:00.000Z',
|
||||
clientCount: 1,
|
||||
hasActivePrompt: false,
|
||||
};
|
||||
};
|
||||
|
||||
return {
|
||||
get sessionCount() {
|
||||
return sessionCount;
|
||||
},
|
||||
getSessionSummary,
|
||||
async shutdown() {},
|
||||
killAllSync() {},
|
||||
} as unknown as AcpSessionBridge;
|
||||
}
|
||||
|
||||
describe('createServeApp default bridge wiring', () => {
|
||||
afterEach(() => {
|
||||
vi.doUnmock('./acp-session-bridge.js');
|
||||
vi.resetModules();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('wires the internally-created bridge lifecycle into the workspace registry', async () => {
|
||||
let sessionLifecycle: BridgeOptions['sessionLifecycle'];
|
||||
const liveSessionIds = new Set<string>();
|
||||
const bridge = makeBridge(0, liveSessionIds);
|
||||
vi.doMock('./acp-session-bridge.js', async () => {
|
||||
const actual = await vi.importActual<
|
||||
typeof import('./acp-session-bridge.js')
|
||||
>('./acp-session-bridge.js');
|
||||
return {
|
||||
...actual,
|
||||
createAcpSessionBridge: vi.fn((opts: BridgeOptions) => {
|
||||
sessionLifecycle = opts.sessionLifecycle;
|
||||
return bridge;
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
const { createServeApp } = await import('./server.js');
|
||||
const app = createServeApp(
|
||||
{
|
||||
port: 0,
|
||||
hostname: '127.0.0.1',
|
||||
workspace: WS_BOUND,
|
||||
} as Parameters<typeof createServeApp>[0],
|
||||
() => 0,
|
||||
);
|
||||
const locals = app.locals as { workspaceRegistry?: WorkspaceRegistry };
|
||||
|
||||
expect(sessionLifecycle).toBeDefined();
|
||||
liveSessionIds.add('session-indexed');
|
||||
sessionLifecycle!({
|
||||
type: 'registered',
|
||||
sessionId: 'session-indexed',
|
||||
workspaceCwd: WS_BOUND,
|
||||
reason: 'spawn',
|
||||
});
|
||||
expect(
|
||||
locals.workspaceRegistry!.resolveLiveSessionOwner('session-indexed'),
|
||||
).toEqual({
|
||||
kind: 'found',
|
||||
runtime: locals.workspaceRegistry!.primary,
|
||||
});
|
||||
liveSessionIds.delete('session-indexed');
|
||||
sessionLifecycle!({
|
||||
type: 'removed',
|
||||
sessionId: 'session-indexed',
|
||||
workspaceCwd: WS_BOUND,
|
||||
reason: 'client_close',
|
||||
});
|
||||
expect(
|
||||
locals.workspaceRegistry!.resolveLiveSessionOwner('session-indexed'),
|
||||
).toEqual({
|
||||
kind: 'not_found',
|
||||
});
|
||||
}, 15_000);
|
||||
|
||||
it('wires total admission into the internally-created bridge', async () => {
|
||||
let freshSessionAdmission: BridgeFreshSessionAdmission | undefined;
|
||||
vi.doMock('./acp-session-bridge.js', async () => {
|
||||
const actual = await vi.importActual<
|
||||
typeof import('./acp-session-bridge.js')
|
||||
>('./acp-session-bridge.js');
|
||||
return {
|
||||
...actual,
|
||||
createAcpSessionBridge: vi.fn((opts: BridgeOptions) => {
|
||||
freshSessionAdmission = opts.freshSessionAdmission;
|
||||
return makeBridge(1);
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
const { createServeApp } = await import('./server.js');
|
||||
createServeApp(
|
||||
{
|
||||
port: 0,
|
||||
hostname: '127.0.0.1',
|
||||
workspace: WS_BOUND,
|
||||
maxTotalSessions: 1,
|
||||
} as Parameters<typeof createServeApp>[0],
|
||||
() => 0,
|
||||
);
|
||||
|
||||
expect(freshSessionAdmission).toBeDefined();
|
||||
let rejection: unknown;
|
||||
try {
|
||||
freshSessionAdmission!({
|
||||
operation: 'spawn',
|
||||
workspaceCwd: WS_BOUND,
|
||||
});
|
||||
} catch (err) {
|
||||
rejection = err;
|
||||
}
|
||||
expect(rejection).toMatchObject({
|
||||
name: 'TotalSessionLimitExceededError',
|
||||
limit: 1,
|
||||
scope: 'total',
|
||||
operation: 'spawn',
|
||||
workspaceCwd: WS_BOUND,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -184,6 +184,20 @@ function restoreEnv(key: string, value: string | undefined): void {
|
|||
}
|
||||
}
|
||||
|
||||
function deferred<T = void>(): {
|
||||
promise: Promise<T>;
|
||||
resolve: (value: T | PromiseLike<T>) => void;
|
||||
reject: (reason?: unknown) => void;
|
||||
} {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((res, rej) => {
|
||||
resolve = res;
|
||||
reject = rej;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
// Workspace fixtures must round-trip through `path.resolve` so the
|
||||
// expected values match the canonicalized form the route produces on
|
||||
// every platform. On Windows `path.resolve('/work/bound')` returns
|
||||
|
|
@ -1801,6 +1815,26 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge {
|
|||
};
|
||||
}
|
||||
|
||||
function makeWorkspaceRuntimeForTest(input: {
|
||||
workspaceId: string;
|
||||
workspaceCwd: string;
|
||||
primary: boolean;
|
||||
bridge: AcpSessionBridge;
|
||||
trusted?: boolean;
|
||||
}): WorkspaceRuntime {
|
||||
return {
|
||||
workspaceId: input.workspaceId,
|
||||
workspaceCwd: input.workspaceCwd,
|
||||
primary: input.primary,
|
||||
trusted: input.trusted ?? true,
|
||||
env: { mode: 'parent-process', overlayKeys: [] },
|
||||
bridge: input.bridge,
|
||||
workspaceService: {} as DaemonWorkspaceService,
|
||||
routeFileSystemFactory: {} as WorkspaceFileSystemFactory,
|
||||
clientMcpSenderRegistry: new ClientMcpSenderRegistry(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Wenshao review #4335 / 3272581557 — detectFromLoopback tests.
|
||||
*/
|
||||
|
|
@ -4912,6 +4946,93 @@ describe('createServeApp', () => {
|
|||
expect(bridge.sessionLspCalls).toEqual(['s-1']);
|
||||
});
|
||||
|
||||
it('dispatches read-only session snapshots through the live owner runtime', async () => {
|
||||
const primaryBridge = fakeBridge();
|
||||
const secondaryBridge = fakeBridge({
|
||||
summaryImpl: (sessionId) => ({
|
||||
sessionId,
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
createdAt: '2026-05-17T12:00:00.000Z',
|
||||
clientCount: 1,
|
||||
hasActivePrompt: false,
|
||||
}),
|
||||
sessionContextImpl: async (sessionId) => ({
|
||||
v: 1,
|
||||
sessionId,
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
state: { owner: 'secondary' },
|
||||
}),
|
||||
});
|
||||
const registry = createWorkspaceRegistry([
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-primary',
|
||||
workspaceCwd: WS_BOUND,
|
||||
primary: true,
|
||||
bridge: primaryBridge,
|
||||
}),
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
primary: false,
|
||||
bridge: secondaryBridge,
|
||||
}),
|
||||
]);
|
||||
const app = createServeApp(
|
||||
{ ...baseOpts, workspace: WS_BOUND },
|
||||
undefined,
|
||||
{ workspaceRegistry: registry },
|
||||
);
|
||||
|
||||
const res = await request(app)
|
||||
.get('/session/s-secondary/context')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`);
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toMatchObject({
|
||||
sessionId: 's-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
state: { owner: 'secondary' },
|
||||
});
|
||||
expect(primaryBridge.sessionContextCalls).toEqual([]);
|
||||
expect(secondaryBridge.sessionContextCalls).toEqual(['s-secondary']);
|
||||
});
|
||||
|
||||
it('surfaces live owner scan failures as structured bridge errors on owner-routed reads', async () => {
|
||||
const primaryBridge = fakeBridge();
|
||||
const secondaryBridge = fakeBridge({
|
||||
summaryImpl: () => {
|
||||
throw new Error('summary exploded');
|
||||
},
|
||||
});
|
||||
const registry = createWorkspaceRegistry([
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-primary',
|
||||
workspaceCwd: WS_BOUND,
|
||||
primary: true,
|
||||
bridge: primaryBridge,
|
||||
}),
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
primary: false,
|
||||
bridge: secondaryBridge,
|
||||
}),
|
||||
]);
|
||||
const app = createServeApp(
|
||||
{ ...baseOpts, workspace: WS_BOUND },
|
||||
undefined,
|
||||
{ workspaceRegistry: registry },
|
||||
);
|
||||
|
||||
const res = await request(app)
|
||||
.get('/session/s-secondary/context')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`);
|
||||
|
||||
expect(res.status).toBe(500);
|
||||
expect(res.body.error).toContain('summary exploded');
|
||||
expect(secondaryBridge.sessionContextCalls).toEqual([]);
|
||||
});
|
||||
|
||||
it('returns session context-usage from the bridge', async () => {
|
||||
const usage: ServeSessionContextUsageStatus = {
|
||||
v: 1,
|
||||
|
|
@ -6040,6 +6161,27 @@ describe('createServeApp', () => {
|
|||
}
|
||||
});
|
||||
|
||||
it('400s unknown explicit cwd before touching the bridge', async () => {
|
||||
const missingCwd = path.join(
|
||||
os.tmpdir(),
|
||||
`qwen-missing-workspace-${Date.now()}`,
|
||||
);
|
||||
for (const action of ['load', 'resume'] as const) {
|
||||
const bridge = fakeBridge();
|
||||
const app = createServeApp(baseOpts, undefined, { bridge });
|
||||
const res = await request(app)
|
||||
.post(`/session/persisted-unknown/${action}`)
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: missingCwd });
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body.code).toBe('workspace_mismatch');
|
||||
expect(res.body.requestedWorkspace).toBe(missingCwd);
|
||||
expect(bridge.loadCalls).toHaveLength(0);
|
||||
expect(bridge.resumeCalls).toHaveLength(0);
|
||||
}
|
||||
});
|
||||
|
||||
it('400s a non-string cwd before touching the bridge', async () => {
|
||||
// Mirrors the `POST /session` malformed-`cwd`-shape test: a
|
||||
// client/orchestrator serialization bug (`cwd: null`,
|
||||
|
|
@ -6148,6 +6290,231 @@ describe('createServeApp', () => {
|
|||
expect(bridge.loadCalls).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('loads an explicit trusted non-primary cwd through that workspace runtime', async () => {
|
||||
const primaryBridge = fakeBridge();
|
||||
const secondaryBridge = fakeBridge({
|
||||
loadImpl: async (req) => ({
|
||||
sessionId: req.sessionId,
|
||||
workspaceCwd: req.workspaceCwd,
|
||||
attached: false,
|
||||
clientId: 'client-secondary-load',
|
||||
state: { workspace: 'secondary' },
|
||||
hasActivePrompt: false,
|
||||
}),
|
||||
});
|
||||
const registry = createWorkspaceRegistry([
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-primary',
|
||||
workspaceCwd: WS_BOUND,
|
||||
primary: true,
|
||||
bridge: primaryBridge,
|
||||
}),
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
primary: false,
|
||||
bridge: secondaryBridge,
|
||||
}),
|
||||
]);
|
||||
const app = createServeApp(
|
||||
{ ...baseOpts, workspace: WS_BOUND },
|
||||
undefined,
|
||||
{ workspaceRegistry: registry },
|
||||
);
|
||||
|
||||
const res = await request(app)
|
||||
.post('/session/persisted-secondary/load')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: WS_DIFFERENT });
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toMatchObject({
|
||||
sessionId: 'persisted-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
clientId: 'client-secondary-load',
|
||||
state: { workspace: 'secondary' },
|
||||
});
|
||||
expect(primaryBridge.loadCalls).toHaveLength(0);
|
||||
expect(secondaryBridge.loadCalls).toEqual([
|
||||
{
|
||||
sessionId: 'persisted-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
historyReplay: 'response',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('rejects restore when the session id is already live in another runtime', async () => {
|
||||
const primaryBridge = fakeBridge({
|
||||
summaryImpl: (sessionId) => ({
|
||||
sessionId,
|
||||
workspaceCwd: WS_BOUND,
|
||||
createdAt: '2026-05-17T12:00:00.000Z',
|
||||
clientCount: 1,
|
||||
hasActivePrompt: false,
|
||||
}),
|
||||
});
|
||||
const secondaryBridge = fakeBridge();
|
||||
const registry = createWorkspaceRegistry([
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-primary',
|
||||
workspaceCwd: WS_BOUND,
|
||||
primary: true,
|
||||
bridge: primaryBridge,
|
||||
}),
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
primary: false,
|
||||
bridge: secondaryBridge,
|
||||
}),
|
||||
]);
|
||||
const app = createServeApp(
|
||||
{ ...baseOpts, workspace: WS_BOUND },
|
||||
undefined,
|
||||
{ workspaceRegistry: registry },
|
||||
);
|
||||
|
||||
const res = await request(app)
|
||||
.post('/session/live-primary/load')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: WS_DIFFERENT });
|
||||
|
||||
expect(res.status).toBe(409);
|
||||
expect(res.body).toMatchObject({
|
||||
code: 'session_workspace_conflict',
|
||||
sessionId: 'live-primary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
liveWorkspaceCwd: WS_BOUND,
|
||||
});
|
||||
expect(secondaryBridge.loadCalls).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('rejects concurrent restore into a different runtime before either bridge owns the session', async () => {
|
||||
const secondaryStarted = deferred();
|
||||
const releaseSecondary = deferred();
|
||||
const primaryBridge = fakeBridge();
|
||||
const secondaryBridge = fakeBridge({
|
||||
loadImpl: async (req) => {
|
||||
secondaryStarted.resolve(undefined);
|
||||
await releaseSecondary.promise;
|
||||
return {
|
||||
sessionId: req.sessionId,
|
||||
workspaceCwd: req.workspaceCwd,
|
||||
attached: false,
|
||||
clientId: 'client-secondary-load',
|
||||
state: { workspace: 'secondary' },
|
||||
hasActivePrompt: false,
|
||||
};
|
||||
},
|
||||
});
|
||||
const registry = createWorkspaceRegistry([
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-primary',
|
||||
workspaceCwd: WS_BOUND,
|
||||
primary: true,
|
||||
bridge: primaryBridge,
|
||||
}),
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
primary: false,
|
||||
bridge: secondaryBridge,
|
||||
}),
|
||||
]);
|
||||
const app = createServeApp(
|
||||
{ ...baseOpts, workspace: WS_BOUND },
|
||||
undefined,
|
||||
{ workspaceRegistry: registry },
|
||||
);
|
||||
|
||||
const secondaryRequest = request(app)
|
||||
.post('/session/concurrent-restore/load')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: WS_DIFFERENT })
|
||||
.then((res) => res);
|
||||
await secondaryStarted.promise;
|
||||
|
||||
const primaryRes = await (async () => {
|
||||
try {
|
||||
return await request(app)
|
||||
.post('/session/concurrent-restore/load')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: WS_BOUND });
|
||||
} finally {
|
||||
releaseSecondary.resolve(undefined);
|
||||
}
|
||||
})();
|
||||
|
||||
expect(primaryRes.status).toBe(409);
|
||||
expect(primaryRes.body).toMatchObject({
|
||||
code: 'session_workspace_conflict',
|
||||
sessionId: 'concurrent-restore',
|
||||
workspaceCwd: WS_BOUND,
|
||||
liveWorkspaceCwd: WS_DIFFERENT,
|
||||
});
|
||||
expect(primaryBridge.loadCalls).toHaveLength(0);
|
||||
|
||||
const secondaryRes = await secondaryRequest;
|
||||
|
||||
expect(secondaryRes.status).toBe(200);
|
||||
expect(secondaryRes.body).toMatchObject({
|
||||
sessionId: 'concurrent-restore',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
});
|
||||
expect(secondaryBridge.loadCalls).toEqual([
|
||||
{
|
||||
sessionId: 'concurrent-restore',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
historyReplay: 'response',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('surfaces live owner scan failures before restoring into a workspace runtime', async () => {
|
||||
const root = await fsp.mkdtemp(path.join(os.tmpdir(), 'qws-owner-scan-'));
|
||||
const primaryDir = path.join(root, 'primary');
|
||||
const secondaryDir = path.join(root, 'secondary');
|
||||
await fsp.mkdir(primaryDir);
|
||||
await fsp.mkdir(secondaryDir);
|
||||
const primaryCwd = realpathSync.native(primaryDir);
|
||||
const secondaryCwd = realpathSync.native(secondaryDir);
|
||||
const primaryBridge = fakeBridge({
|
||||
summaryImpl: () => {
|
||||
throw new Error('summary exploded');
|
||||
},
|
||||
});
|
||||
const secondaryBridge = fakeBridge();
|
||||
const registry = createWorkspaceRegistry([
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-primary',
|
||||
workspaceCwd: primaryCwd,
|
||||
primary: true,
|
||||
bridge: primaryBridge,
|
||||
}),
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-secondary',
|
||||
workspaceCwd: secondaryCwd,
|
||||
primary: false,
|
||||
bridge: secondaryBridge,
|
||||
}),
|
||||
]);
|
||||
const app = createServeApp(
|
||||
{ ...baseOpts, workspace: primaryCwd },
|
||||
undefined,
|
||||
{ workspaceRegistry: registry },
|
||||
);
|
||||
|
||||
const res = await request(app)
|
||||
.post('/session/persisted-secondary/load')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: secondaryCwd });
|
||||
|
||||
expect(res.status).toBe(500);
|
||||
expect(res.body.error).toContain('summary exploded');
|
||||
expect(secondaryBridge.loadCalls).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('503 + Retry-After: 5 when the bridge throws SessionLimitExceededError', async () => {
|
||||
const bridge = fakeBridge({
|
||||
resumeImpl: async () => {
|
||||
|
|
@ -6513,6 +6880,54 @@ describe('createServeApp', () => {
|
|||
expect(bridge.listCalls).toEqual([WS_BOUND]);
|
||||
});
|
||||
|
||||
it('supports plural /workspaces/:workspace/sessions for non-primary live sessions', async () => {
|
||||
const primaryBridge = fakeBridge();
|
||||
const secondaryBridge = fakeBridge({
|
||||
listImpl: () => [
|
||||
{
|
||||
sessionId: 's-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
createdAt: '2026-05-17T12:00:00.000Z',
|
||||
clientCount: 1,
|
||||
hasActivePrompt: false,
|
||||
},
|
||||
],
|
||||
});
|
||||
const registry = createWorkspaceRegistry([
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-primary',
|
||||
workspaceCwd: WS_BOUND,
|
||||
primary: true,
|
||||
bridge: primaryBridge,
|
||||
}),
|
||||
makeWorkspaceRuntimeForTest({
|
||||
workspaceId: 'ws-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
primary: false,
|
||||
bridge: secondaryBridge,
|
||||
}),
|
||||
]);
|
||||
const app = createServeApp(
|
||||
{ ...baseOpts, workspace: WS_BOUND },
|
||||
undefined,
|
||||
{ workspaceRegistry: registry },
|
||||
);
|
||||
|
||||
const res = await request(app)
|
||||
.get('/workspaces/ws-secondary/sessions')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`);
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.sessions).toEqual([
|
||||
expect.objectContaining({
|
||||
sessionId: 's-secondary',
|
||||
workspaceCwd: WS_DIFFERENT,
|
||||
}),
|
||||
]);
|
||||
expect(primaryBridge.listCalls).toEqual([]);
|
||||
expect(secondaryBridge.listCalls).toEqual([WS_DIFFERENT]);
|
||||
});
|
||||
|
||||
it('includes persisted sessions from the CLI session store', async () => {
|
||||
const storedOnlyId = '550e8400-e29b-41d4-a716-446655440000';
|
||||
const liveAndStoredId = '6ba7b810-9dad-11d1-80b4-00c04fd430c8';
|
||||
|
|
@ -14255,72 +14670,6 @@ describe('createServeApp ServeAppDeps.fsFactory wiring (#4175 PR 18)', () => {
|
|||
expect(locals.workspaceRegistry!.primary.env).toBe(primaryRuntimeEnv);
|
||||
});
|
||||
|
||||
it('wires total admission into the internally-created bridge', async () => {
|
||||
type AdmissionContext = {
|
||||
operation: 'spawn';
|
||||
workspaceCwd: string;
|
||||
};
|
||||
let freshSessionAdmission:
|
||||
| ((context: AdmissionContext) => { release(): void })
|
||||
| undefined;
|
||||
vi.resetModules();
|
||||
vi.doMock('./acp-session-bridge.js', async () => {
|
||||
const actual = await vi.importActual<
|
||||
typeof import('./acp-session-bridge.js')
|
||||
>('./acp-session-bridge.js');
|
||||
return {
|
||||
...actual,
|
||||
createAcpSessionBridge: vi.fn((opts: unknown) => {
|
||||
freshSessionAdmission = (
|
||||
opts as {
|
||||
freshSessionAdmission?: typeof freshSessionAdmission;
|
||||
}
|
||||
).freshSessionAdmission;
|
||||
return fakeBridge();
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
try {
|
||||
const { createServeApp } = await import('./server.js');
|
||||
createServeApp(
|
||||
{
|
||||
port: 0,
|
||||
hostname: '127.0.0.1',
|
||||
workspace: WS_BOUND,
|
||||
maxTotalSessions: 1,
|
||||
} as Parameters<typeof createServeApp>[0],
|
||||
() => 0,
|
||||
);
|
||||
|
||||
expect(freshSessionAdmission).toBeDefined();
|
||||
const first = freshSessionAdmission!({
|
||||
operation: 'spawn',
|
||||
workspaceCwd: WS_BOUND,
|
||||
});
|
||||
let rejection: unknown;
|
||||
try {
|
||||
freshSessionAdmission!({
|
||||
operation: 'spawn',
|
||||
workspaceCwd: WS_BOUND,
|
||||
});
|
||||
} catch (err) {
|
||||
rejection = err;
|
||||
}
|
||||
expect(rejection).toMatchObject({
|
||||
name: 'TotalSessionLimitExceededError',
|
||||
limit: 1,
|
||||
scope: 'total',
|
||||
operation: 'spawn',
|
||||
workspaceCwd: WS_BOUND,
|
||||
});
|
||||
first.release();
|
||||
} finally {
|
||||
vi.doUnmock('./acp-session-bridge.js');
|
||||
vi.resetModules();
|
||||
}
|
||||
});
|
||||
|
||||
it('uses an injected workspace registry as the primary runtime source', async () => {
|
||||
const { createServeApp } = await import('./server.js');
|
||||
const runtime = makeInjectedWorkspaceRuntime();
|
||||
|
|
|
|||
|
|
@ -125,6 +125,7 @@ import { SessionArchiveCoordinator } from './server/session-archive.js';
|
|||
import { installSelfOriginStripMiddleware } from './server/self-origin.js';
|
||||
import {
|
||||
createSingleWorkspaceRegistry,
|
||||
createWorkspaceSessionOwnerIndex,
|
||||
type WorkspaceRegistry,
|
||||
type WorkspaceRuntimeEnvMetadata,
|
||||
} from './workspace-registry.js';
|
||||
|
|
@ -520,6 +521,9 @@ export function createServeApp(
|
|||
defaultBridgeForAdmission ? [defaultBridgeForAdmission] : [],
|
||||
})
|
||||
: undefined;
|
||||
const defaultSessionOwnerIndex = !injectedWorkspaceRegistry
|
||||
? createWorkspaceSessionOwnerIndex()
|
||||
: undefined;
|
||||
const bridge =
|
||||
injectedWorkspaceRegistry?.primary.bridge ??
|
||||
deps.bridge ??
|
||||
|
|
@ -528,6 +532,12 @@ export function createServeApp(
|
|||
...(totalSessionAdmission
|
||||
? { freshSessionAdmission: totalSessionAdmission.admit }
|
||||
: {}),
|
||||
...(defaultSessionOwnerIndex
|
||||
? {
|
||||
sessionLifecycle:
|
||||
defaultSessionOwnerIndex.handleBridgeSessionLifecycle,
|
||||
}
|
||||
: {}),
|
||||
maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession,
|
||||
eventRingSize: opts.eventRingSize,
|
||||
compactedReplayMaxBytes: opts.compactedReplayMaxBytes,
|
||||
|
|
@ -619,20 +629,25 @@ export function createServeApp(
|
|||
});
|
||||
const workspaceRegistry =
|
||||
injectedWorkspaceRegistry ??
|
||||
createSingleWorkspaceRegistry({
|
||||
workspaceId: hashDaemonWorkspace(boundWorkspace),
|
||||
workspaceCwd: boundWorkspace,
|
||||
primary: true,
|
||||
trusted: deps.primaryWorkspaceTrusted ?? false,
|
||||
env: primaryRuntimeEnvMetadata ?? {
|
||||
mode: 'parent-process',
|
||||
overlayKeys: [],
|
||||
createSingleWorkspaceRegistry(
|
||||
{
|
||||
workspaceId: hashDaemonWorkspace(boundWorkspace),
|
||||
workspaceCwd: boundWorkspace,
|
||||
primary: true,
|
||||
trusted: deps.primaryWorkspaceTrusted ?? false,
|
||||
env: primaryRuntimeEnvMetadata ?? {
|
||||
mode: 'parent-process',
|
||||
overlayKeys: [],
|
||||
},
|
||||
bridge,
|
||||
workspaceService: workspace,
|
||||
routeFileSystemFactory: fsFactory,
|
||||
clientMcpSenderRegistry,
|
||||
},
|
||||
bridge,
|
||||
workspaceService: workspace,
|
||||
routeFileSystemFactory: fsFactory,
|
||||
clientMcpSenderRegistry,
|
||||
});
|
||||
defaultSessionOwnerIndex
|
||||
? { sessionOwnerIndex: defaultSessionOwnerIndex }
|
||||
: {},
|
||||
);
|
||||
(app.locals as { workspaceRegistry?: WorkspaceRegistry }).workspaceRegistry =
|
||||
workspaceRegistry;
|
||||
const primaryRuntime = workspaceRegistry.primary;
|
||||
|
|
|
|||
|
|
@ -96,6 +96,28 @@ describe('daemonTelemetryMiddleware — recordRequest seam', () => {
|
|||
expect(recordRequest).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('maps plural workspace session listing to the existing route label', () => {
|
||||
const recordRequest = vi.fn();
|
||||
const mw = daemonTelemetryMiddleware(() => '/ws', recordRequest);
|
||||
const res = mockRes(200);
|
||||
|
||||
mw(
|
||||
mockReq('GET', '/workspaces/ws-secondary/sessions'),
|
||||
res,
|
||||
vi.fn() as unknown as NextFunction,
|
||||
);
|
||||
res.emit('finish');
|
||||
|
||||
expect(recordRequest).toHaveBeenCalledTimes(1);
|
||||
expect(coreMocks.withDaemonRequestSpan).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
method: 'GET',
|
||||
route: 'GET /workspace/:id/sessions',
|
||||
}),
|
||||
expect.any(Function),
|
||||
);
|
||||
});
|
||||
|
||||
it('excludes the dashboard status poll (GET /daemon/status) from recordRequest', () => {
|
||||
const recordRequest = vi.fn();
|
||||
const mw = daemonTelemetryMiddleware(() => '/ws', recordRequest);
|
||||
|
|
|
|||
|
|
@ -114,6 +114,9 @@ export function resolveDaemonTelemetryRoute(
|
|||
if (req.method === 'GET' && /^\/workspace\/[^/]+\/sessions$/.test(path)) {
|
||||
return { route: 'GET /workspace/:id/sessions' };
|
||||
}
|
||||
if (req.method === 'GET' && /^\/workspaces\/[^/]+\/sessions$/.test(path)) {
|
||||
return { route: 'GET /workspace/:id/sessions' };
|
||||
}
|
||||
if (req.method === 'POST' && path === '/workspace/init') {
|
||||
return { route: 'POST /workspace/init' };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@
|
|||
import { describe, expect, it } from 'vitest';
|
||||
import { SessionNotFoundError } from './acp-session-bridge.js';
|
||||
import {
|
||||
createWorkspaceSessionOwnerIndex,
|
||||
createWorkspaceRegistry,
|
||||
createSingleWorkspaceRegistry,
|
||||
type WorkspaceRuntime,
|
||||
|
|
@ -162,6 +163,77 @@ describe('createWorkspaceRegistry', () => {
|
|||
});
|
||||
});
|
||||
|
||||
it('uses the session owner index before scanning runtime bridges', () => {
|
||||
const primarySummary = vi.fn(() => {
|
||||
throw new SessionNotFoundError('sess-secondary');
|
||||
});
|
||||
const secondarySummary = vi.fn((sessionId: string) => ({
|
||||
sessionId,
|
||||
workspaceCwd: '/work/secondary',
|
||||
}));
|
||||
const primary = makeRuntime('/work/primary', {
|
||||
workspaceId: 'ws-primary',
|
||||
primary: true,
|
||||
bridge: bridgeWithSummary(primarySummary),
|
||||
});
|
||||
const secondary = makeRuntime('/work/secondary', {
|
||||
workspaceId: 'ws-secondary',
|
||||
bridge: bridgeWithSummary(secondarySummary),
|
||||
});
|
||||
const sessionOwnerIndex = createWorkspaceSessionOwnerIndex();
|
||||
sessionOwnerIndex.register('sess-secondary', '/work/secondary');
|
||||
|
||||
const registry = createWorkspaceRegistry([primary, secondary], {
|
||||
sessionOwnerIndex,
|
||||
});
|
||||
|
||||
expect(registry.resolveLiveSessionOwner('sess-secondary')).toEqual({
|
||||
kind: 'found',
|
||||
runtime: secondary,
|
||||
});
|
||||
expect(primarySummary).not.toHaveBeenCalled();
|
||||
expect(secondarySummary).toHaveBeenCalledWith('sess-secondary');
|
||||
});
|
||||
|
||||
it('drops stale indexed owners and caches the fallback scan result', () => {
|
||||
const primarySummary = vi.fn((sessionId: string) => ({
|
||||
sessionId,
|
||||
workspaceCwd: '/work/primary',
|
||||
}));
|
||||
const secondarySummary = vi.fn(() => {
|
||||
throw new SessionNotFoundError('stale');
|
||||
});
|
||||
const primary = makeRuntime('/work/primary', {
|
||||
workspaceId: 'ws-primary',
|
||||
primary: true,
|
||||
bridge: bridgeWithSummary(primarySummary),
|
||||
});
|
||||
const secondary = makeRuntime('/work/secondary', {
|
||||
workspaceId: 'ws-secondary',
|
||||
bridge: bridgeWithSummary(secondarySummary),
|
||||
});
|
||||
const sessionOwnerIndex = createWorkspaceSessionOwnerIndex();
|
||||
sessionOwnerIndex.register('stale', '/work/secondary');
|
||||
|
||||
const registry = createWorkspaceRegistry([primary, secondary], {
|
||||
sessionOwnerIndex,
|
||||
});
|
||||
|
||||
expect(registry.resolveLiveSessionOwner('stale')).toEqual({
|
||||
kind: 'found',
|
||||
runtime: primary,
|
||||
});
|
||||
expect(primarySummary).toHaveBeenCalledTimes(1);
|
||||
expect(secondarySummary).toHaveBeenCalledTimes(2);
|
||||
|
||||
expect(registry.resolveLiveSessionOwner('stale')).toEqual({
|
||||
kind: 'found',
|
||||
runtime: primary,
|
||||
});
|
||||
expect(primarySummary).toHaveBeenCalledTimes(2);
|
||||
expect(secondarySummary).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('fails closed when live session owner resolution is ambiguous', () => {
|
||||
const first = makeRuntime('/work/primary', {
|
||||
workspaceId: 'ws-primary',
|
||||
|
|
@ -201,4 +273,35 @@ describe('createWorkspaceRegistry', () => {
|
|||
|
||||
expect(() => registry.resolveLiveSessionOwner('sess')).toThrow(lookupError);
|
||||
});
|
||||
|
||||
it('does not cache partial scan results when live owner scan fails', () => {
|
||||
const lookupError = new Error('bridge unavailable');
|
||||
const primarySummary = vi.fn((sessionId: string) => ({
|
||||
sessionId,
|
||||
workspaceCwd: '/work/primary',
|
||||
}));
|
||||
const secondarySummary = vi.fn(() => {
|
||||
throw lookupError;
|
||||
});
|
||||
const primary = makeRuntime('/work/primary', {
|
||||
workspaceId: 'ws-primary',
|
||||
primary: true,
|
||||
bridge: bridgeWithSummary(primarySummary),
|
||||
});
|
||||
const secondary = makeRuntime('/work/secondary', {
|
||||
workspaceId: 'ws-secondary',
|
||||
bridge: bridgeWithSummary(secondarySummary),
|
||||
});
|
||||
const sessionOwnerIndex = createWorkspaceSessionOwnerIndex();
|
||||
const registry = createWorkspaceRegistry([primary, secondary], {
|
||||
sessionOwnerIndex,
|
||||
});
|
||||
|
||||
expect(() => registry.resolveLiveSessionOwner('sess')).toThrow(lookupError);
|
||||
expect(sessionOwnerIndex.getWorkspaceCwds('sess')).toEqual([]);
|
||||
|
||||
expect(() => registry.resolveLiveSessionOwner('sess')).toThrow(lookupError);
|
||||
expect(primarySummary).toHaveBeenCalledTimes(2);
|
||||
expect(secondarySummary).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -45,6 +45,25 @@ export type WorkspaceSessionOwnerResolution =
|
|||
readonly runtimes: readonly WorkspaceRuntime[];
|
||||
};
|
||||
|
||||
export type WorkspaceSessionLifecycleEvent =
|
||||
| {
|
||||
readonly type: 'registered';
|
||||
readonly sessionId: string;
|
||||
readonly workspaceCwd: string;
|
||||
}
|
||||
| {
|
||||
readonly type: 'removed';
|
||||
readonly sessionId: string;
|
||||
readonly workspaceCwd: string;
|
||||
};
|
||||
|
||||
export interface WorkspaceSessionOwnerIndex {
|
||||
register(sessionId: string, workspaceCwd: string): void;
|
||||
remove(sessionId: string, workspaceCwd?: string): void;
|
||||
getWorkspaceCwds(sessionId: string): readonly string[];
|
||||
handleBridgeSessionLifecycle(event: WorkspaceSessionLifecycleEvent): void;
|
||||
}
|
||||
|
||||
export interface WorkspaceRegistry {
|
||||
readonly primary: WorkspaceRuntime;
|
||||
list(): readonly WorkspaceRuntime[];
|
||||
|
|
@ -56,8 +75,52 @@ export interface WorkspaceRegistry {
|
|||
resolveLiveSessionOwner(sessionId: string): WorkspaceSessionOwnerResolution;
|
||||
}
|
||||
|
||||
export interface WorkspaceRegistryOptions {
|
||||
readonly sessionOwnerIndex?: WorkspaceSessionOwnerIndex;
|
||||
}
|
||||
|
||||
export function createWorkspaceSessionOwnerIndex(): WorkspaceSessionOwnerIndex {
|
||||
const bySessionId = new Map<string, Set<string>>();
|
||||
|
||||
const register = (sessionId: string, workspaceCwd: string): void => {
|
||||
let owners = bySessionId.get(sessionId);
|
||||
if (!owners) {
|
||||
owners = new Set<string>();
|
||||
bySessionId.set(sessionId, owners);
|
||||
}
|
||||
owners.add(workspaceCwd);
|
||||
};
|
||||
|
||||
const remove = (sessionId: string, workspaceCwd?: string): void => {
|
||||
if (workspaceCwd === undefined) {
|
||||
bySessionId.delete(sessionId);
|
||||
return;
|
||||
}
|
||||
const owners = bySessionId.get(sessionId);
|
||||
if (!owners) return;
|
||||
owners.delete(workspaceCwd);
|
||||
if (owners.size === 0) {
|
||||
bySessionId.delete(sessionId);
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
register,
|
||||
remove,
|
||||
getWorkspaceCwds: (sessionId) => [...(bySessionId.get(sessionId) ?? [])],
|
||||
handleBridgeSessionLifecycle: (event) => {
|
||||
if (event.type === 'registered') {
|
||||
register(event.sessionId, event.workspaceCwd);
|
||||
} else {
|
||||
remove(event.sessionId, event.workspaceCwd);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function createWorkspaceRegistry(
|
||||
inputRuntimes: readonly WorkspaceRuntime[],
|
||||
options: WorkspaceRegistryOptions = {},
|
||||
): WorkspaceRegistry {
|
||||
if (inputRuntimes.length === 0) {
|
||||
throw new Error(
|
||||
|
|
@ -96,6 +159,30 @@ export function createWorkspaceRegistry(
|
|||
|
||||
const runtimes = Object.freeze([...inputRuntimes]);
|
||||
const primary = primaryRuntimes[0]!;
|
||||
const sessionOwnerIndex = options.sessionOwnerIndex;
|
||||
const scanLiveOwners = (
|
||||
sessionId: string,
|
||||
): WorkspaceSessionOwnerResolution => {
|
||||
const matches: WorkspaceRuntime[] = [];
|
||||
for (const runtime of runtimes) {
|
||||
try {
|
||||
runtime.bridge.getSessionSummary(sessionId);
|
||||
matches.push(runtime);
|
||||
} catch (err) {
|
||||
if (err instanceof SessionNotFoundError) continue;
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
for (const match of matches) {
|
||||
sessionOwnerIndex?.register(sessionId, match.workspaceCwd);
|
||||
}
|
||||
if (matches.length === 0) return { kind: 'not_found' };
|
||||
if (matches.length === 1) {
|
||||
return { kind: 'found', runtime: matches[0]! };
|
||||
}
|
||||
return { kind: 'ambiguous', runtimes: matches };
|
||||
};
|
||||
|
||||
return {
|
||||
primary,
|
||||
list: () => runtimes,
|
||||
|
|
@ -104,27 +191,41 @@ export function createWorkspaceRegistry(
|
|||
resolveWorkspaceCwd: (workspaceCwd) =>
|
||||
workspaceCwd === undefined ? primary : byCwd.get(workspaceCwd),
|
||||
resolveLiveSessionOwner: (sessionId) => {
|
||||
const matches: WorkspaceRuntime[] = [];
|
||||
for (const runtime of runtimes) {
|
||||
try {
|
||||
runtime.bridge.getSessionSummary(sessionId);
|
||||
matches.push(runtime);
|
||||
} catch (err) {
|
||||
if (err instanceof SessionNotFoundError) continue;
|
||||
throw err;
|
||||
const indexedCwds = sessionOwnerIndex?.getWorkspaceCwds(sessionId) ?? [];
|
||||
if (indexedCwds.length > 0) {
|
||||
const matches: WorkspaceRuntime[] = [];
|
||||
for (const workspaceCwd of indexedCwds) {
|
||||
const runtime = byCwd.get(workspaceCwd);
|
||||
if (!runtime) {
|
||||
sessionOwnerIndex?.remove(sessionId, workspaceCwd);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
runtime.bridge.getSessionSummary(sessionId);
|
||||
matches.push(runtime);
|
||||
} catch (err) {
|
||||
if (err instanceof SessionNotFoundError) {
|
||||
sessionOwnerIndex?.remove(sessionId, workspaceCwd);
|
||||
continue;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
if (matches.length === 1) {
|
||||
return { kind: 'found', runtime: matches[0]! };
|
||||
}
|
||||
if (matches.length > 1) {
|
||||
return { kind: 'ambiguous', runtimes: matches };
|
||||
}
|
||||
}
|
||||
if (matches.length === 0) return { kind: 'not_found' };
|
||||
if (matches.length === 1) {
|
||||
return { kind: 'found', runtime: matches[0]! };
|
||||
}
|
||||
return { kind: 'ambiguous', runtimes: matches };
|
||||
return scanLiveOwners(sessionId);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function createSingleWorkspaceRegistry(
|
||||
runtime: WorkspaceRuntime,
|
||||
options: WorkspaceRegistryOptions = {},
|
||||
): WorkspaceRegistry {
|
||||
return createWorkspaceRegistry([runtime]);
|
||||
return createWorkspaceRegistry([runtime], options);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue