Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 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
4 changes: 4 additions & 0 deletions packages/acp-bridge/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,10 @@
"types": "./dist/internal/testUtils.d.ts",
"import": "./dist/internal/testUtils.js"
},
"./compactionEngine": {
"types": "./dist/compactionEngine.d.ts",
"import": "./dist/compactionEngine.js"
},
"./package.json": "./package.json"
},
"scripts": {
Expand Down
5 changes: 5 additions & 0 deletions packages/acp-bridge/src/bridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -818,6 +818,9 @@ describe('createHttpAcpBridge', () => {
clientId: expect.stringMatching(/^client_/),
createdAt: expect.any(String),
state: { configOptions: [] },
compactedReplay: [],
liveJournal: [],
lastEventId: 0,
});
expect(handles[0]?.agent.loadSessionCalls).toEqual([
{ sessionId: 'persisted-1', cwd: WS_A, mcpServers: [] },
Expand Down Expand Up @@ -919,6 +922,7 @@ describe('createHttpAcpBridge', () => {
clientId: expect.stringMatching(/^client_/),
createdAt: expect.any(String),
state: { modes: null },
lastEventId: 0,
});
expect(handles[0]?.agent.loadSessionCalls).toHaveLength(0);
expect(handles[0]?.agent.resumeSessionCalls).toEqual([
Expand Down Expand Up @@ -962,6 +966,7 @@ describe('createHttpAcpBridge', () => {
clientId: expect.stringMatching(/^client_/),
createdAt: expect.any(String),
state: { _meta: { tag: 'restored-foo' } },
lastEventId: expect.any(Number),
});
expect(attached.clientId).not.toBe(loaded.clientId);
expect(handles[0]?.agent.loadSessionCalls).toHaveLength(1);
Expand Down
30 changes: 28 additions & 2 deletions packages/acp-bridge/src/bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import {
import type { ShellCommandResult } from './bridgeTypes.js';
import type { AcpChannel } from './channel.js';
import { EventBus, DEFAULT_RING_SIZE, type BridgeEvent } from './eventBus.js';
import { TurnBoundaryCompactionEngine } from './compactionEngine.js';
import {
BridgeChannelClosedError,
BridgeTimeoutError,
Expand Down Expand Up @@ -1669,11 +1670,14 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
}
};

const createSessionEventBus = (): EventBus =>
new EventBus(eventRingSize, undefined, new TurnBoundaryCompactionEngine());

const createSessionEntry = (
ci: ChannelInfo,
sessionId: string,
workspaceCwd: string,
events = new EventBus(eventRingSize),
events = createSessionEventBus(),
): SessionEntry => {
const entry: SessionEntry = {
sessionId,
Expand Down Expand Up @@ -1732,6 +1736,25 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
);
};

const replayFieldsFor = (
Comment thread
doudouOUC marked this conversation as resolved.
Resolved
entry: { events: EventBus },
action: 'load' | 'resume',
): Pick<
BridgeRestoredSession,
'compactedReplay' | 'liveJournal' | 'lastEventId'
> => {
const snapshot = entry.events.snapshotReplay();
if (!snapshot) return { lastEventId: entry.events.lastEventId };
if (action === 'load') {
return {
compactedReplay: snapshot.compactedTurns,
liveJournal: snapshot.liveJournal,
lastEventId: snapshot.lastEventId,
};
}
return { lastEventId: snapshot.lastEventId };
};

async function restoreSession(
action: 'load' | 'resume',
req: BridgeRestoreSessionRequest,
Expand All @@ -1754,6 +1777,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
// Late attachers get the same ACP state the original restore
// caller saw; spawn-only sessions don't carry a state payload.
state: existing.restoreState ?? {},
...replayFieldsFor(existing, action),
};
}

Expand Down Expand Up @@ -1820,7 +1844,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
throw new SessionLimitExceededError(maxSessions);
}

const restoreEvents = new EventBus(eventRingSize);
const restoreEvents = createSessionEventBus();
let registeredEntry: SessionEntry | undefined;
let ci: ChannelInfo | undefined;
// Live counter shared with coalesced waiters (see InFlightRestore
Expand Down Expand Up @@ -1937,6 +1961,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
clientId,
createdAt: racedEntry.createdAt,
state: racedEntry.restoreState ?? {},
...replayFieldsFor(racedEntry, action),
};
}

Expand Down Expand Up @@ -1970,6 +1995,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
clientId,
createdAt: entry.createdAt,
state,
...replayFieldsFor(entry, action),
};
})().finally(() => {
ci?.pendingRestoreIds.delete(req.sessionId);
Expand Down
6 changes: 6 additions & 0 deletions packages/acp-bridge/src/bridgeTypes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,12 @@ export type BridgeSessionState = LoadSessionResponse | ResumeSessionResponse;
export interface BridgeRestoredSession extends BridgeSession {
/** ACP state returned by `session/load` / `session/resume`. */
state: BridgeSessionState;
/** Compacted events for all completed turns (O(turns) size). */
compactedReplay?: BridgeEvent[];
/** Raw events since last turn boundary (current incomplete turn). */
liveJournal?: BridgeEvent[];
/** High-water mark event ID — client uses this as initial SSE cursor. */
lastEventId?: number;
}

/** Sparse summary used by `GET /workspace/:id/sessions`. */
Expand Down
Loading