diff --git a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts index aea43a2024..561672602f 100644 --- a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts @@ -230,6 +230,7 @@ test('keeps a bounded contiguous window while moving between history and the tai events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 4, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrapPage, overlay: { ...page(null), source: 'overlay' }, @@ -294,6 +295,7 @@ test('delivers a mid-session tail append even while a history window is resident events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 4, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrapPage, overlay: { ...page(null), source: 'overlay' }, @@ -364,6 +366,7 @@ test('does not resurrect a discarded replica when a tail re-anchor is in flight' events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 4, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrapPage, overlay: { ...page(null), source: 'overlay' }, @@ -446,6 +449,7 @@ test('does not resurrect a discarded replica when a history load is in flight', events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 4, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrapPage, overlay: { ...page(null), source: 'overlay' }, @@ -518,6 +522,7 @@ test('does not drive a discarded replica terminal when a contiguous catch-up is events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 4, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrapPage, overlay: { ...page(null, 4), source: 'overlay' }, @@ -572,6 +577,7 @@ test('loads a history target with newer messages available below it', async () = events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 4, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrapPage, overlay: { ...transcriptPage('older', null, 4), source: 'overlay' }, @@ -624,6 +630,7 @@ test('keeps an oversized transcript sparse while moving between indexed prompts' events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 15, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrapPage, overlay: { ...transcriptPage('older', null, 15), source: 'overlay' }, @@ -727,6 +734,7 @@ test('keeps history resident when an active overlay uses its own cache budget', events: { async *[Symbol.asyncIterator]() {} }, transcriptBootstrap: { throughSequence: 1, + durableCoverage: 'complete', overlayMessageCount: 1, durable: bootstrapPage, overlay: { ...transcriptPage('older', null, 1), source: 'overlay' }, diff --git a/apps/desktop/src/main/__tests__/runtime-host-client.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client.test.ts index 01fd41646e..4318b9d004 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client.test.ts @@ -163,6 +163,7 @@ function subscription( activeAssistantStreams: [], transcriptBootstrap: { throughSequence: null, + durableCoverage: 'complete', overlayMessageCount: 0, durable: emptyTranscriptPage(sessionId, 'durable'), overlay: emptyTranscriptPage(sessionId, 'overlay'), diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts index 5d0fb4e4fb..b8c7e90dc5 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts @@ -594,6 +594,7 @@ test('fences transcript range failures across same-source replica recovery', asy events, transcriptBootstrap: { throughSequence: 1, + durableCoverage: 'complete', overlayMessageCount: 0, durable: bootstrap, overlay: { ...bootstrap, source: 'overlay', nextCursor: null }, @@ -887,6 +888,7 @@ test('finishes transcript open and replays a stale range request after replaceme events, transcriptBootstrap: { throughSequence: 0, + durableCoverage: 'complete', overlayMessageCount: 0, durable: { kind: 'page', diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts b/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts index c213a0ee45..6ea30bb2d9 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts @@ -46,6 +46,7 @@ export function runtimeHostSessionFixture(input: { activeAssistantStreams: input.activeAssistantStreams ?? [], transcriptBootstrap: input.transcriptBootstrap ?? { throughSequence: null, + durableCoverage: 'complete', overlayMessageCount: 0, durable: emptyPage(sessionId, 'durable'), overlay: emptyPage(sessionId, 'overlay'), diff --git a/apps/desktop/src/main/runtime-host-session-subscription-owner.ts b/apps/desktop/src/main/runtime-host-session-subscription-owner.ts index 21d95402db..096683bf58 100644 --- a/apps/desktop/src/main/runtime-host-session-subscription-owner.ts +++ b/apps/desktop/src/main/runtime-host-session-subscription-owner.ts @@ -404,11 +404,13 @@ export class RuntimeHostSessionSubscriptionOwner { } function subscriptionClosedError( - reason: "slow_consumer" | "session_removed", + reason: "access_revoked" | "slow_consumer" | "session_removed", ): Error { - return reason === "session_removed" + return reason !== "slow_consumer" ? new SessionRemovedSubscriptionError( - "Runtime Host Session was removed while it was observed", + reason === "access_revoked" + ? "Runtime Host Session access was revoked" + : "Runtime Host Session was removed while it was observed", ) : new RuntimeHostSubscriptionError( "slow_consumer", diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 686ad8493c..cbfc7efec3 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -780,6 +780,17 @@ export function userFacingText(message: Pick { + const root = await mkdtemp(join(tmpdir(), 'maka-artifact-shared-read-')); + const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + try { + const store = await openInteractiveArtifactStoreForWrite(owner.lease); + await store.recover(); + await store.create({ + id: 'shared-image', + sessionId: 'session-1', + turnId: 'turn-1', + name: 'shared.png', + kind: 'image', + mimeType: 'image/png', + source: 'user_upload', + content: Buffer.from('image'), + now: 1, + }); + await store.create({ + id: 'private-artifact', + sessionId: 'session-1', + turnId: 'turn-1', + name: 'private.txt', + kind: 'file', + source: 'provider_request_capture', + content: Buffer.from('private'), + now: 2, + }); + let active = true; + let revokeAfterAuthorization = false; + const coordinator = new HostArtifactCoordinator( + store, + () => assert.fail('successful read must not request Host drain'), + new SessionAdmissionGate(), + { probeSessionRemoval: async () => ({ kind: 'present' }) }, + Date.now, + { + activeSessionGrant: () => { + if (!active) return; + if (revokeAfterAuthorization) { + revokeAfterAuthorization = false; + queueMicrotask(() => { + active = false; + }); + } + return { + kind: 'session_observation', + grantId: 'grant-1', + principalId: 'guest-1', + sessionId: 'session-1', + createdAt: '2026-08-30T00:00:00.000Z', + }; + }, + }, + ); + const guest = { + ...connectionContext, + principal: 'guest-1', + principalKind: 'session_guest' as const, + }; + + const visible = await coordinator.handlers['artifact.query']( + { kind: 'get', sessionId: 'session-1', artifactId: 'shared-image' }, + guest, + ); + assert.equal(visible.ok && visible.result.kind === 'artifact', true); + assert.equal( + ( + await coordinator.handlers['artifact.query']( + { kind: 'get', sessionId: 'session-1', artifactId: 'private-artifact' }, + guest, + ) + ).ok, + false, + ); + assert.equal( + ( + await coordinator.handlers['artifact.query']( + { kind: 'list_start', sessionId: 'session-1' }, + guest, + ) + ).ok, + false, + ); + + revokeAfterAuthorization = true; + assert.equal( + ( + await coordinator.handlers['artifact.query']( + { kind: 'get', sessionId: 'session-1', artifactId: 'shared-image' }, + guest, + ) + ).ok, + false, + ); + assert.equal( + ( + await coordinator.handlers['artifact.query']( + { kind: 'get', sessionId: 'session-1', artifactId: 'shared-image' }, + guest, + ) + ).ok, + false, + ); + store.close(); + } finally { + await owner.close(); + await rm(root, { recursive: true, force: true }); + } +}); + function digest(bytes: Uint8Array): `sha256:${string}` { return `sha256:${createHash('sha256').update(bytes).digest('hex')}`; } diff --git a/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts b/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts index 878675d6c7..fb75ef9e90 100644 --- a/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts +++ b/packages/runtime-host/src/__tests__/authenticated-websocket.test.ts @@ -39,6 +39,7 @@ import { INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, RUNTIME_HOST_PROTOCOL_VERSION, SESSION_CATALOG_LIVE_RUN_STATE_SCHEMA_VERSION, + decodeCollaborationInvitationCode, type RequestFrame, } from '../protocol/index.js'; import { openRuntimeHostAccessAuthority } from '../server/access-authority.js'; @@ -70,6 +71,7 @@ test('one Local IPC owner and one authenticated WebSocket Client control the sam }); let local: RuntimeHostConnection | undefined; let remote: RuntimeHostConnection | undefined; + let guest: RuntimeHostConnection | undefined; try { local = requireConnection(await connectRuntimeHost({ rootPath: root, protocol: PROTOCOL })); const issued = await local.request('access.credential.issue', { @@ -217,6 +219,58 @@ test('one Local IPC owner and one authenticated WebSocket Client control the sam modelTarget: { kind: 'default' }, }); assert.ok(!('kind' in created)); + const preparedGuest = await local.request('collaboration.invitation.prepare', { + sessionId: 'shared-session', + grantKinds: ['session_observation'], + }); + const guestInvitation = decodeCollaborationInvitationCode(preparedGuest.invitationCode); + const pendingGuest = await connectRemoteRuntimeHost({ + url, + credential: guestInvitation.credential, + clientInstanceId: 'guest-client', + expectedRootId: capability.rootId, + compositionId: INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, + protocol: PROTOCOL, + }); + assert.equal(pendingGuest.kind, 'connected', JSON.stringify(pendingGuest)); + if (pendingGuest.kind !== 'connected') assert.fail('Session Guest did not connect'); + await pendingGuest.connection.request('access.credential.finalize', {}); + await pendingGuest.connection.close(); + const activeGuest = await connectRemoteRuntimeHost({ + url, + credential: guestInvitation.credential, + clientInstanceId: 'guest-client', + expectedRootId: capability.rootId, + compositionId: INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, + protocol: PROTOCOL, + }); + assert.equal(activeGuest.kind, 'connected', JSON.stringify(activeGuest)); + if (activeGuest.kind !== 'connected') assert.fail('Session Guest did not reconnect'); + guest = activeGuest.connection; + const sharedCatalog = await guest.request('session.shared.query', {}); + assert.equal(sharedCatalog.session?.id, 'shared-session'); + assert.equal('workspace' in sharedCatalog.session!, false); + await assert.rejects( + guest.request('session.catalog.query', { kind: 'list_start' }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'unauthorized', + ); + const guestSubscription = await guest.openSessionSubscription({ + sessionId: 'shared-session', + transcript: { kind: 'none' }, + }); + const observationGrant = preparedGuest.grants.find( + (grant) => grant.kind === 'session_observation', + )!; + await local.request('collaboration.grant.revoke', { + grantId: observationGrant.grantId, + }); + const closed = await guestSubscription[Symbol.asyncIterator]().next(); + assert.equal(closed.done, false); + assert.equal(closed.value?.kind, 'subscription.closed'); + if (closed.value?.kind === 'subscription.closed') { + assert.equal(closed.value.reason, 'access_revoked'); + } assert.deepEqual( await remote.request('session.catalog.query', { kind: 'get', @@ -409,7 +463,7 @@ test('one Local IPC owner and one authenticated WebSocket Client control the sam { credentialId: candidate.credentialId, revoked: true }, ); } finally { - await Promise.allSettled([remote?.close(), local?.close()]); + await Promise.allSettled([guest?.close(), remote?.close(), local?.close()]); await host.close().catch(() => undefined); await rm(join(resolveRootControlNamespace(), capability.rootId), { recursive: true, diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index c749055c17..f6c0715795 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -1632,6 +1632,7 @@ function transcriptBootstrapFor(sessionId: string) { const contents = Buffer.from('t'.repeat(16 * 1024)); return { throughSequence: 0, + durableCoverage: 'complete' as const, overlayMessageCount: 0, durable: { kind: 'page' as const, diff --git a/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts b/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts index bc86eb9954..8d2dbecb96 100644 --- a/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts +++ b/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts @@ -101,6 +101,40 @@ export function transcriptReader( } return { throughSequence, fragments, rawBytes, next }; }, + readDurableRecords: async (_sessionId, request) => { + const throughSequence = + request.throughSequence === undefined + ? durable.length === 0 + ? null + : durable.length - 1 + : request.throughSequence; + if (throughSequence === null) { + return { throughSequence: null, records: [], nextPosition: null }; + } + const position = request.position ?? (request.direction === 'older' ? throughSequence : 0); + const candidates = durable + .map((message, sequence) => ({ sequence, message })) + .filter( + ({ sequence }) => + sequence <= throughSequence && + (request.direction === 'older' ? sequence <= position : sequence >= position), + ) + .sort((left, right) => + request.direction === 'older' + ? right.sequence - left.sequence + : left.sequence - right.sequence, + ); + const records = candidates.slice(0, request.maxMessages); + const last = records.at(-1); + return { + throughSequence, + records, + nextPosition: + last && records.length < candidates.length + ? last.sequence + (request.direction === 'older' ? -1 : 1) + : null, + }; + }, readDurableMessagesById: async (_sessionId, request) => request.throughSequence === null ? [] diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index a0f2f62842..791b9f8a6e 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -5162,6 +5162,7 @@ function operationContext( hostEpoch, connectionId, principal: 'local_os_user' as const, + principalKind: 'local_owner' as const, acquireResidency, }; } diff --git a/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts b/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts index 4716466df1..a9597a56bd 100644 --- a/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts @@ -161,6 +161,72 @@ describe('Host Runtime Resource coordinator', () => { assert.equal(harness.listReadCount, 0); }); + test('fences Guest reads to the exact active observation grant', async () => { + let active = true; + let releaseRead!: () => void; + let markReadStarted!: () => void; + const readBarrier = new Promise((resolve) => { + releaseRead = resolve; + }); + const readStarted = new Promise((resolve) => { + markReadStarted = resolve; + }); + const harness = createHarness({ + sessionAccessAuthority: { + activeSessionGrant(principalId, sessionId, kind) { + return active && + principalId === 'guest-1' && + sessionId === SESSION_ID && + kind === 'session_observation' + ? { + kind, + grantId: 'grant-1', + principalId, + sessionId, + createdAt: '2026-01-01T00:00:00.000Z', + } + : undefined; + }, + }, + }); + harness.pointReadBarrier = readBarrier; + harness.pointReadStarted = markReadStarted; + const guest = guestConnection('guest-1'); + + const outsideGrant = await harness.coordinator.handlers['runtime.resource.query']( + { kind: 'list_start', sessionId: 'session-2' }, + guest, + ); + assert.equal(outsideGrant.ok, false); + assert.equal(!outsideGrant.ok && outsideGrant.error.code, 'not_found'); + assert.equal(harness.listReadCount, 0); + + harness.updates = [resourceUpdate(0), resourceUpdate(1, { sessionId: 'parent-session' })]; + const visible = await harness.coordinator.handlers['runtime.resource.query']( + { kind: 'list_start', sessionId: SESSION_ID }, + guest, + ); + assert.equal(visible.ok, true); + assert.deepEqual( + visible.ok && visible.result.kind === 'page' + ? visible.result.resources.map((resource) => resource.result.ref) + : [], + [shellRef(0)], + ); + + const pending = harness.coordinator.handlers['runtime.resource.query']( + { kind: 'get', sessionId: SESSION_ID, ref: shellRef(0) }, + guest, + ); + await readStarted; + assert.equal(harness.pointReadCount, 1); + active = false; + releaseRead(); + const revoked = await pending; + assert.equal(revoked.ok, false); + assert.equal(!revoked.ok && revoked.error.code, 'not_found'); + }); + test('drains for canonical state failure but keeps projection failure scoped to its query', async () => { const harness = createHarness(); harness.updates = [ @@ -638,7 +704,12 @@ describe('Host Runtime Resource coordinator', () => { }); }); -function createHarness(options: Pick = {}) { +function createHarness( + options: Pick< + HostRuntimeResourceCoordinatorInput, + 'resolveShell' | 'sessionAccessAuthority' + > = {}, +) { let backgroundCompletion: ShellRunBashInput['onCompletion']; let currentSnapshot = ptySnapshot(); let lastStartedSnapshot: ShellRunSnapshotResult | undefined; @@ -651,6 +722,8 @@ function createHarness(options: Pick | undefined, + pointReadStarted: undefined as (() => void) | undefined, stateReadFailure: undefined as Error | undefined, inspectFailure: undefined as Error | undefined, activeResidencies: 0, @@ -748,6 +821,8 @@ function createHarness(options: Pick { state.pointReadCount += 1; + state.pointReadStarted?.(); + await state.pointReadBarrier; if (state.stateReadFailure) throw state.stateReadFailure; return structuredClone(state.updates.find((update) => update.result.ref === ref) ?? null); }, @@ -805,6 +880,14 @@ function connection(connectionId: string): ConnectionContext { }; } +function guestConnection(principal: string): ConnectionContext { + return { + ...connection('guest-connection'), + principal, + principalKind: 'session_guest', + }; +} + function backgroundInput(onCompletion?: ShellRunBashInput['onCompletion']): ShellRunBashInput { return { sessionId: SESSION_ID, diff --git a/packages/runtime-host/src/__tests__/session-collaboration-authority.test.ts b/packages/runtime-host/src/__tests__/session-collaboration-authority.test.ts index 2ac5c2ecdb..9f3573e370 100644 --- a/packages/runtime-host/src/__tests__/session-collaboration-authority.test.ts +++ b/packages/runtime-host/src/__tests__/session-collaboration-authority.test.ts @@ -44,6 +44,13 @@ test('Session Guest invitation, grants, and revocation form one durable authorit await authority.finalize(credentialId, 'guest-client'); assert.deepEqual(authority.authenticate(invitation.credential)?.operationGrants, [ 'host.status', + 'artifact.query', + 'runtime.resource.query', + 'session.shared.query', + 'subscription.open', + 'subscription.close', + 'session.transcript.page', + 'session.transcript.overlay.release', ]); const observation = prepared.grants.find((grant) => grant.kind === 'session_observation')!; diff --git a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts index ccb25b8115..04b4852b31 100644 --- a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts @@ -47,6 +47,16 @@ import { transcriptReader } from './fixtures/session-transcript-reader.js'; const HOST_EPOCH = 'host-epoch'; const SESSION_ID = 'session-1'; +const TEST_OWNER_IDENTITY = { + principalId: 'local_owner', + principalKind: 'local_owner', +} as const; +type TestIdentity = + | typeof TEST_OWNER_IDENTITY + | { + readonly principalId: string; + readonly principalKind: 'session_guest'; + }; test('open is an inactive publication barrier and live sequence starts at nextSequence', async () => { const read = deferred(); @@ -88,6 +98,53 @@ test('open is an inactive publication barrier and live sequence starts at nextSe coordinator.close(); }); +test('Guest revocation wins a concurrent subscription open', async () => { + const read = deferred(); + const grant = { + kind: 'session_observation' as const, + grantId: 'grant-1', + principalId: 'guest-1', + sessionId: SESSION_ID, + createdAt: '2026-08-30T00:00:00.000Z', + }; + let active = true; + let publishRevocation: ((revoked: typeof grant) => void) | undefined; + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + () => read.promise, + new SessionAdmissionGate(), + undefined, + undefined, + undefined, + { + activeSessionGrant: () => (active ? grant : undefined), + subscribeGrantRevocations: (listener) => { + publishRevocation = listener; + return () => undefined; + }, + }, + ); + coordinator.attachConnection('guest-connection', new RecordingSink()); + + const opening = coordinator.handlers['subscription.open']( + { sessionId: SESSION_ID, transcript: { kind: 'none' } }, + connectionContext('guest-connection', { + principalId: 'guest-1', + principalKind: 'session_guest', + }), + ); + await delayImmediate(); + active = false; + publishRevocation?.(grant); + read.resolve(canonical()); + + assert.deepEqual(await opening, { + ok: false, + error: { code: 'not_found', message: 'Session was not found' }, + }); + coordinator.close(); +}); + test('forwards the durable steering echo to subscribers as a session event', async () => { const sink = new RecordingSink(); const coordinator = new SessionContinuityCoordinator( @@ -121,6 +178,86 @@ test('forwards the durable steering echo to subscribers as a session event', asy coordinator.close(); }); +test('projects model-only user content out of Guest queue and steering frames', async () => { + const grant = { + kind: 'session_observation' as const, + grantId: 'grant-1', + principalId: 'guest-1', + sessionId: SESSION_ID, + createdAt: '2026-08-30T00:00:00.000Z', + }; + const sink = new RecordingSink(); + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + async () => + canonical({ + queue: { + hostEpoch: HOST_EPOCH, + queueRevision: 1, + steering: [], + followup: [ + { + entryId: 'entry-1', + messageId: 'message-1', + content: { text: 'private skill body', displayText: 'visible prompt' }, + placement: 'next_turn', + state: 'queued', + }, + ], + }, + }), + new SessionAdmissionGate(), + undefined, + undefined, + undefined, + { + activeSessionGrant: () => grant, + subscribeGrantRevocations: () => () => undefined, + }, + ); + const connection = coordinator.attachConnection('guest-connection', sink); + const opened = await open( + coordinator, + 'guest-connection', + { kind: 'none' }, + { + principalId: grant.principalId, + principalKind: 'session_guest', + }, + ); + assert.deepEqual(opened.snapshot.queue.followup[0]?.content, { text: 'visible prompt' }); + connection.activate(opened.subscriptionId); + coordinator.enqueueAgentGraphChanged({ + rootSessionId: SESSION_ID, + graphId: 'private-graph', + reason: 'observation', + }); + await delayImmediate(); + assert.equal(sink.frames.length, 0); + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', { + type: 'steering_message', + id: 'steering-event-1', + turnId: 'turn-1', + ts: 7, + messageId: 'steering-message-1', + content: { text: 'private skill body', displayText: 'visible steer' }, + }); + + const frame = sink.frames.find((candidate) => candidate.kind === 'subscription.session_event'); + assert.equal(frame?.kind, 'subscription.session_event'); + if (frame?.kind === 'subscription.session_event') { + assert.deepEqual(frame.event, { + type: 'steering_message', + id: 'steering-event-1', + turnId: 'turn-1', + ts: 7, + messageId: 'steering-message-1', + content: { text: 'visible steer' }, + }); + } + coordinator.close(); +}); + test('open snapshot includes pending Interactions from the canonical projection', async () => { const pending = pendingInteraction(); const coordinator = new SessionContinuityCoordinator( @@ -597,10 +734,24 @@ test('coalesces typed domain invalidations without publishing continuity project test('fans one bounded Runtime Resource burst out to an inherited Session view', async () => { const childSessionId = 'child-session'; + const grant = { + kind: 'session_observation' as const, + grantId: 'grant-1', + principalId: 'guest-1', + sessionId: childSessionId, + createdAt: '2026-08-30T00:00:00.000Z', + }; const coordinator = new SessionContinuityCoordinator( HOST_EPOCH, async (sessionId) => canonicalFor(sessionId), new SessionAdmissionGate(), + undefined, + undefined, + undefined, + { + activeSessionGrant: () => grant, + subscribeGrantRevocations: () => () => undefined, + }, ); const sink = new RecordingSink(); const connection = coordinator.attachConnection('connection-1', sink); @@ -611,6 +762,18 @@ test('fans one bounded Runtime Resource burst out to an inherited Session view', assert.equal(outcome.ok, true); if (!outcome.ok) return; connection.activate(outcome.result.subscriptionId); + const guestSink = new RecordingSink(); + const guestConnection = coordinator.attachConnection('guest-connection', guestSink); + const guestOutcome = await coordinator.handlers['subscription.open']( + { sessionId: childSessionId, transcript: { kind: 'none' } }, + connectionContext('guest-connection', { + principalId: grant.principalId, + principalKind: 'session_guest', + }), + ); + assert.equal(guestOutcome.ok, true); + if (!guestOutcome.ok) return; + guestConnection.activate(guestOutcome.result.subscriptionId); const updates = Array.from({ length: 64 }, (_, index) => { const update = shellRunUpdate({ sessionId: 'parent-session', @@ -619,6 +782,7 @@ test('fans one bounded Runtime Resource burst out to an inherited Session view', update.result.ref = `shell:run-${index}`; return update; }); + updates[updates.length - 1]!.sessionId = childSessionId; for (const update of updates) coordinator.enqueueRuntimeResourceChanged(update); await waitFor(() => sink.frames.length === 1); @@ -635,6 +799,20 @@ test('fans one bounded Runtime Resource burst out to an inherited Session view', ref: update.result.ref, })), }); + assert.deepEqual(guestSink.frames[0], { + kind: 'subscription.session_domain_changed', + hostEpoch: HOST_EPOCH, + subscriptionId: guestOutcome.result.subscriptionId, + sequence: 1, + sessionId: childSessionId, + domain: 'runtime_resource', + resources: [ + { + sourceSessionId: childSessionId, + ref: updates[updates.length - 1]!.result.ref, + }, + ], + }); coordinator.close(); }); @@ -1672,6 +1850,87 @@ test('an in-flight transcript page cannot outlive its owning connection', async coordinator.close(); }); +test('an in-flight transcript page cannot outlive its Guest observation grant', async () => { + const message = assistantMessage('界'.repeat(20_000)); + const continued = deferred(); + const baseReader = transcriptReader([message]); + let blockPage = false; + const reader: SessionTranscriptReader = { + ...baseReader, + readDurableRecords: async (sessionId, request) => { + if (blockPage) await continued.promise; + return baseReader.readDurableRecords(sessionId, request); + }, + }; + const grant = { + kind: 'session_observation' as const, + grantId: 'grant-1', + principalId: 'guest-1', + sessionId: SESSION_ID, + createdAt: '2026-08-30T00:00:00.000Z', + }; + let active = true; + let publishRevocation: ((revoked: typeof grant) => void) | undefined; + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + async () => canonical(), + new SessionAdmissionGate(), + undefined, + reader, + undefined, + { + activeSessionGrant: () => (active ? grant : undefined), + subscribeGrantRevocations: (listener) => { + publishRevocation = listener; + return () => undefined; + }, + }, + ); + coordinator.attachConnection('guest-connection', new RecordingSink()); + const opened = await open( + coordinator, + 'guest-connection', + { + kind: 'tail', + maxBytes: SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES, + }, + { + principalId: grant.principalId, + principalKind: 'session_guest', + }, + ); + const cursor = opened.transcript?.durable.nextCursor; + assert.ok(cursor); + if (!opened.transcript || !cursor) return; + + blockPage = true; + const reading = coordinator.handlers['session.transcript.page']( + { + subscriptionId: opened.subscriptionId, + source: 'durable', + direction: 'older', + throughSequence: opened.transcript.throughSequence, + cursor, + anchorSequence: null, + maxBytes: 1024, + }, + connectionContext('guest-connection', { + principalId: grant.principalId, + principalKind: 'session_guest', + }), + ); + await delayImmediate(); + active = false; + publishRevocation?.(grant); + continued.resolve(); + + assert.deepEqual(await reading, { + ok: false, + error: { code: 'not_found', message: 'Session subscription was not found' }, + }); + coordinator.close(); +}); + test('a durable append refresh advances transcript before its completion event', async () => { const durable = [assistantMessage('first')]; const reader = transcriptReader(durable); @@ -2048,10 +2307,11 @@ async function open( transcript: { readonly kind: 'none' } | { readonly kind: 'tail'; readonly maxBytes: number } = { kind: 'none', }, + identity: TestIdentity = TEST_OWNER_IDENTITY, ) { const outcome = await coordinator.handlers['subscription.open']( { sessionId: SESSION_ID, transcript }, - connectionContext(connectionId), + connectionContext(connectionId, identity), ); if (!outcome.ok) throw new Error(outcome.error.message); assert.equal(outcome.ok, true); @@ -2153,11 +2413,15 @@ async function consumeBootstrapOverlay( ); } -function connectionContext(connectionId: string): ConnectionContext { +function connectionContext( + connectionId: string, + identity: TestIdentity = TEST_OWNER_IDENTITY, +): ConnectionContext { return { hostEpoch: HOST_EPOCH, connectionId, - principal: 'local_os_user', + principal: identity.principalId, + principalKind: identity.principalKind, acquireResidency: () => ({ release() {} }), }; } diff --git a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts index 1a777646ef..6ca64650b9 100644 --- a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts +++ b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts @@ -392,6 +392,7 @@ test('reassembles a large message from bounded backward pages', async () => { const openRequest = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId); const opened = openResult(hostEpoch, 'subscription-fragmented', { throughSequence: 0, + durableCoverage: 'complete', overlayMessageCount: 0, durable: transcriptPage({ rawBytes: encoded.byteLength - splitAt, @@ -472,6 +473,7 @@ test('decodes one bounded page without walking the remaining transcript', async const subscription = new ClientSessionSubscription( openResult('host-1', 'subscription-bounded-page', { throughSequence: 4, + durableCoverage: 'complete', overlayMessageCount: 0, durable: { ...transcriptPage({ @@ -811,6 +813,7 @@ test('rejects a durable sequence gap', async () => { const gap = new ClientSessionSubscription( openResult('host-1', 'subscription-gap', { throughSequence: 1, + durableCoverage: 'complete', overlayMessageCount: 0, durable: { ...transcriptPage({ @@ -832,6 +835,54 @@ test('rejects a durable sequence gap', async () => { ); }); +test('loads a projected durable transcript with intentionally sparse sequences', async () => { + const messages = [0, 2].map((sequence) => + Buffer.from( + JSON.stringify({ + type: 'user', + id: `user-${sequence}`, + turnId: 'turn-1', + ts: sequence + 1, + text: `visible-${sequence}`, + }), + 'utf8', + ), + ); + const subscription = new ClientSessionSubscription( + openResult('host-1', 'subscription-projected', { + throughSequence: 2, + durableCoverage: 'projected', + overlayMessageCount: 0, + durable: { + ...transcriptPage({ + rawBytes: messages.reduce((total, message) => total + message.byteLength, 0), + fragments: messages + .map((message, index) => ({ + kind: 'durable' as const, + sequence: index * 2, + byteOffset: 0, + totalBytes: message.byteLength, + payloadDigest: null, + data: message.toString('base64'), + })) + .reverse(), + }), + throughSequence: 2, + }, + overlay: { ...transcriptPage({ source: 'overlay' }), throughSequence: 2 }, + }), + async () => undefined, + async () => { + throw new Error('unexpected page request'); + }, + ); + + assert.deepEqual( + (await subscription.loadTranscript(decodeStoredMessage)).map((message) => message.id), + ['user-0', 'user-2'], + ); +}); + test('rejects a durable message that does not match its payload digest', async () => { const message = Buffer.from( JSON.stringify({ @@ -846,6 +897,7 @@ test('rejects a durable message that does not match its payload digest', async ( const subscription = new ClientSessionSubscription( openResult('host-1', 'subscription-digest-mismatch', { throughSequence: 0, + durableCoverage: 'complete', overlayMessageCount: 0, durable: transcriptPage({ rawBytes: message.byteLength, @@ -910,6 +962,7 @@ test('rejects a transcript cursor that does not advance', async () => { const subscription = new ClientSessionSubscription( openResult('host-1', 'subscription-stuck-cursor', { throughSequence: 0, + durableCoverage: 'complete', overlayMessageCount: 0, durable: repeated, overlay: transcriptPage({ source: 'overlay' }), @@ -939,6 +992,7 @@ test('rejects an overlay that terminates before its declared high-water', async const subscription = new ClientSessionSubscription( openResult('host-1', 'subscription-truncated-overlay', { throughSequence: null, + durableCoverage: 'complete', overlayMessageCount: 2, durable: { ...transcriptPage(), throughSequence: null }, overlay: { @@ -1030,6 +1084,7 @@ test('acknowledges a complete overlay before waiting for durable continuation pa const subscription = new ClientSessionSubscription( openResult('host-1', 'subscription-overlay-release-before-durable', { throughSequence: 0, + durableCoverage: 'complete', overlayMessageCount: 1, durable: transcriptPage({ rawBytes: durableMessage.byteLength - split, @@ -1103,6 +1158,7 @@ test('close stops transcript pagination after the in-flight page', async () => { const subscription = new ClientSessionSubscription( openResult('host-1', 'subscription-closing', { throughSequence: 0, + durableCoverage: 'complete', overlayMessageCount: 0, durable: transcriptPage({ rawBytes: Math.floor(message.byteLength / 2), @@ -1293,6 +1349,7 @@ function openResult( function transcriptBootstrap(message: Buffer): SessionTranscriptBootstrap { return { throughSequence: 0, + durableCoverage: 'complete', overlayMessageCount: 0, durable: transcriptPage({ rawBytes: message.byteLength, @@ -1314,6 +1371,7 @@ function transcriptBootstrap(message: Buffer): SessionTranscriptBootstrap { function overlayBootstrap(message: Buffer): SessionTranscriptBootstrap { return { throughSequence: null, + durableCoverage: 'complete', overlayMessageCount: 1, durable: { ...transcriptPage(), throughSequence: null }, overlay: { diff --git a/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts b/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts index e4ea72480a..9dd2b10fcd 100644 --- a/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts +++ b/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts @@ -28,6 +28,7 @@ import { updateSubscriberTranscriptHighWater, } from '../server/session-transcript-pager.js'; import type { SessionTranscriptReader } from '../server/session-transcript-reader.js'; +import { projectSharedSessionTranscriptMessage } from '../server/shared-session-transcript.js'; import { transcriptReader } from './fixtures/session-transcript-reader.js'; test('reads newly durable messages forward from an announced watermark', async () => { @@ -41,6 +42,7 @@ test('reads newly durable messages forward from an announced watermark', async ( rootTurn: null, activeAssistantStreams: [], maxBytes: 1024, + projection: 'owner', }); assert.equal(bootstrap.throughSequence, 1); @@ -68,6 +70,151 @@ test('reads newly durable messages forward from an announced watermark', async ( assert.equal(page.nextCursor, null); }); +test('projects durable and active transcript records before sharing them', async () => { + const durable: StoredMessage[] = [ + { + ...assistantMessage(0), + providerOptions: { replay: 'private' }, + thinking: { + text: 'visible thought', + signature: 'private-signature', + providerOptions: { replay: 'private' }, + }, + }, + { + type: 'tool_result', + id: 'result-1', + turnId: 'turn-0', + ts: 2, + toolUseId: 'tool-1', + isError: false, + content: { kind: 'text', text: 'visible result' }, + modelVisibility: 'hidden', + providerOutput: { replay: 'private' }, + }, + { + type: 'system_note', + id: 'audit-1', + ts: 3, + kind: 'mode_change', + data: { previousSessionId: 'private-session' }, + }, + { + type: 'user', + id: 'user-1', + turnId: 'turn-0', + ts: 4, + text: 'private composed skill instructions', + displayText: 'visible attachment', + steeringEventId: 'steering-event-1', + attachments: [ + { + kind: 'code', + name: 'visible.ts', + mimeType: 'text/typescript', + bytes: 7, + ref: { kind: 'session_file', sessionId: 'session-1', relativePath: 'visible.ts' }, + }, + { + kind: 'code', + name: 'private.ts', + mimeType: 'text/typescript', + bytes: 8, + ref: { kind: 'external_file', absolutePath: '/private/private.ts' }, + }, + ], + }, + ]; + const overlay: StoredMessage[] = [ + { + ...assistantMessage(1), + providerOptions: { replay: 'private' }, + }, + ]; + const reader = transcriptReader(durable, overlay); + const owner = await createSessionTranscriptBootstrap({ + reader, + sessionId: 'session-1', + subscriptionId: 'owner', + throughSequence: 2, + rootTurn: null, + activeAssistantStreams: [], + maxBytes: 16 * 1024, + projection: 'owner', + }); + const shared = await createSessionTranscriptBootstrap({ + reader, + sessionId: 'session-1', + subscriptionId: 'shared', + throughSequence: 3, + rootTurn: null, + activeAssistantStreams: [], + maxBytes: 16 * 1024, + projection: 'shared', + }); + + assert.equal( + (decodeBootstrap(owner.bootstrap.durable)[0] as { data?: unknown }).data !== undefined, + true, + ); + assert.equal( + (decodeBootstrap(owner.bootstrap.overlay)[0] as { providerOptions?: unknown }) + .providerOptions !== undefined, + true, + ); + const sharedDurable = decodeBootstrap(shared.bootstrap.durable); + assert.deepEqual( + sharedDurable.map((message) => message.type), + ['user', 'tool_result', 'assistant'], + ); + const sharedAttachments = sharedDurable[0]?.attachments; + assert.deepEqual( + Array.isArray(sharedAttachments) + ? sharedAttachments.map((item) => (item as { name: string }).name) + : [], + ['visible.ts'], + ); + assert.equal(sharedDurable[0]?.text, 'visible attachment'); + assert.equal('displayText' in sharedDurable[0]!, false); + assert.equal(sharedDurable[0]?.steeringEventId, 'steering-event-1'); + assert.equal('providerOutput' in sharedDurable[1]!, false); + assert.equal('providerOptions' in sharedDurable[2]!, false); + assert.deepEqual(sharedDurable[2]!.thinking, { text: 'visible thought' }); + assert.equal('providerOptions' in decodeBootstrap(shared.bootstrap.overlay)[0]!, false); + const projectedState = projectSharedSessionTranscriptMessage( + { + type: 'turn_state', + id: 'state-1', + turnId: 'turn-0', + ts: 5, + status: 'aborted', + abortedAt: 5, + abortSource: 'stop_button', + partialOutputRetained: true, + }, + 'session-1', + ); + assert.equal(projectedState?.type, 'turn_state'); + if (projectedState?.type === 'turn_state') { + assert.equal(projectedState.abortSource, 'stop_button'); + } + const projectedInput = projectSharedSessionTranscriptMessage( + { + type: 'tool_call', + id: 'tool-1', + turnId: 'turn-0', + ts: 6, + toolName: 'WriteStdin', + args: { ref: 'shell-1', input: 'secret=sk-example-value' }, + }, + 'session-1', + ); + assert.equal(projectedInput?.type, 'tool_call'); + if (projectedInput?.type === 'tool_call') { + assert.equal(JSON.stringify(projectedInput.args).includes('sk-example-value'), false); + } +}); + test('rejects cursor tampering and cross-subscription replay', async () => { const reader = transcriptReader([userMessage(0, 'x'.repeat(2_000))]); const first = await createSessionTranscriptBootstrap({ @@ -78,6 +225,7 @@ test('rejects cursor tampering and cross-subscription replay', async () => { rootTurn: null, activeAssistantStreams: [], maxBytes: 128, + projection: 'owner', }); const second = await createSessionTranscriptBootstrap({ reader, @@ -87,6 +235,7 @@ test('rejects cursor tampering and cross-subscription replay', async () => { rootTurn: null, activeAssistantStreams: [], maxBytes: 128, + projection: 'owner', }); const cursor = first.bootstrap.durable.nextCursor; assert.ok(cursor); @@ -127,6 +276,7 @@ test('keeps a durable continuation when overlay bytes reduce the bootstrap budge rootTurn: null, activeAssistantStreams: [], maxBytes: Buffer.byteLength(JSON.stringify(durable[0]), 'utf8') * 2, + projection: 'owner', }); assert.ok(bootstrap.overlay.rawBytes > 0); assert.ok(bootstrap.durable.nextCursor); @@ -143,6 +293,7 @@ test('shrinks the raw bootstrap until it fits its aggregate encoded budget', asy activeAssistantStreams: [], maxBytes: 16 * 1024, maxEncodedBytes: 4 * 1024, + projection: 'owner', }); assert.ok(Buffer.byteLength(JSON.stringify(bootstrap), 'utf8') <= 4 * 1024); assert.ok(bootstrap.durable.nextCursor); @@ -230,3 +381,11 @@ function assistantMessage(index: number): Extract>['bootstrap']['durable'], +): Array> { + return page.fragments.map((fragment) => + JSON.parse(Buffer.from(fragment.data, 'base64').toString('utf8')), + ); +} diff --git a/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts b/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts index d6b9886a6f..78e73ce6b4 100644 --- a/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts @@ -70,6 +70,7 @@ test('Session transcript protocol accepts bounded correlated pages and bootstrap const bootstrap = { throughSequence: 3, + durableCoverage: 'complete' as const, overlayMessageCount: 0, durable: { ...page, direction: 'older' as const }, overlay: { @@ -181,6 +182,7 @@ test('Session transcript protocol rejects malformed and uncorrelated values', () () => decodeSessionTranscriptBootstrap({ throughSequence: 3, + durableCoverage: 'complete', overlayMessageCount: 0, durable: page, overlay: { ...page, source: 'overlay', throughSequence: 2 }, diff --git a/packages/runtime-host/src/client/session-subscription.ts b/packages/runtime-host/src/client/session-subscription.ts index 569261d0af..3a61369a59 100644 --- a/packages/runtime-host/src/client/session-subscription.ts +++ b/packages/runtime-host/src/client/session-subscription.ts @@ -317,7 +317,9 @@ export class ClientSessionSubscription } const overlay = await this.#consumeTranscriptOverlay(bootstrap); const durable = await this.#loadTranscriptSource(bootstrap.durable); - assertCompleteIdentities(durable, bootstrap.throughSequence); + if (bootstrap.durableCoverage === 'complete') { + assertCompleteIdentities(durable, bootstrap.throughSequence); + } const messages = durable.map((entry) => entry.value); const indexById = new Map(); for (const [index, message] of messages.entries()) { diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 7fa57c3559..9b16632344 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -95,7 +95,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 69 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 70 as const; +// 70: Session Guest connections receive resource-scoped shared catalog and +// continuity projections. Older peers cannot enforce the Session grant fence. // 69: Runtime Host access authority recognizes restricted Session Guest // principals and typed Session collaboration grants. Older Hosts would either // reject the new operations or misclassify the authenticated principal. diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 67c05a29f7..ed7af8d936 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -305,6 +305,7 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'session.create', 'session.execution_boundary.query', 'session.lifecycle.set', + 'session.shared.query', 'session.metadata.update', 'session.read_marker.set', 'session.recap.generate', diff --git a/packages/runtime-host/src/protocol/session-catalog.ts b/packages/runtime-host/src/protocol/session-catalog.ts index 62c5309a48..da2ffcbbd6 100644 --- a/packages/runtime-host/src/protocol/session-catalog.ts +++ b/packages/runtime-host/src/protocol/session-catalog.ts @@ -253,8 +253,29 @@ export interface UnsupportedLegacySessionCatalogRecord { readonly reason: 'not_wire_representable'; } +export interface SharedSessionCatalogProjection { + readonly kind: 'shared_session'; + readonly id: string; + readonly revision: number; + readonly createdAt: number; + readonly activityAt: number; + readonly name: string; + readonly lastMessageAt?: number; + readonly lastMessagePreview?: string; + readonly status: SessionStatus; + readonly liveRunState?: SessionCatalogLiveRunState; + readonly blockedReason?: SessionBlockedReason; + readonly statusUpdatedAt?: number; +} + export type SessionCatalogItem = SessionCatalogProjection | UnsupportedLegacySessionCatalogRecord; +export type SharedSessionCatalogQueryInput = Record; + +export interface SharedSessionCatalogQueryResult { + readonly session: SharedSessionCatalogProjection | null; +} + export type SessionCatalogQueryResult = | { readonly kind: 'page'; @@ -281,6 +302,17 @@ export type SessionUpdateResult = }; export const SESSION_CATALOG_OPERATION_SPECS = { + 'session.shared.query': defineOperation< + SharedSessionCatalogQueryInput, + SharedSessionCatalogQueryResult, + (typeof QUERY_ERRORS)[number] + >({ + mode: 'query', + availability: 'ready', + errors: QUERY_ERRORS, + decodeInput: decodeSharedSessionCatalogQueryInput, + decodeOutput: decodeSharedSessionCatalogQueryResult, + }), 'session.catalog.query': defineOperation< SessionCatalogQueryInput, SessionCatalogQueryResult, @@ -371,6 +403,49 @@ export const SESSION_CATALOG_OPERATION_SPECS = { }), } as const; +function decodeSharedSessionCatalogQueryInput(value: unknown): SharedSessionCatalogQueryInput { + requireExactRecord(value, 'shared Session catalog query input', []); + return {}; +} + +function decodeSharedSessionCatalogQueryResult(value: unknown): SharedSessionCatalogQueryResult { + const record = requireExactRecord(value, 'shared Session catalog query result', ['session']); + const session = + record.session === null ? null : decodeSharedSessionCatalogProjection(record.session); + requireEncodedByteLimit( + session, + 'shared Session catalog result', + SESSION_CATALOG_RESULT_MAX_BYTES, + ); + return { session }; +} + +export function decodeSharedSessionCatalogProjection( + value: unknown, +): SharedSessionCatalogProjection { + const exact = requireShapedRecord( + value, + 'shared Session catalog projection', + ['kind', 'id', 'revision', 'createdAt', 'activityAt', 'name', 'status'], + ['lastMessageAt', 'lastMessagePreview', 'liveRunState', 'blockedReason', 'statusUpdatedAt'], + ); + if (exact.kind !== 'shared_session') throw invalidProtocolFrame('Invalid shared Session kind'); + return { + kind: 'shared_session', + id: requireEntityId(exact.id, 'Session id'), + revision: positiveRevision(exact.revision, 'Session revision'), + createdAt: timestamp(exact.createdAt, 'Session createdAt'), + activityAt: timestamp(exact.activityAt, 'Session activityAt'), + name: sessionName(exact.name), + ...optionalTimestamp(exact, 'lastMessageAt'), + ...optionalText(exact, 'lastMessagePreview', SESSION_CATALOG_PREVIEW_MAX_BYTES), + status: decodeSessionStatus(exact.status), + ...optionalLiveRunState(exact), + ...optionalBlockedReason(exact), + ...optionalTimestamp(exact, 'statusUpdatedAt'), + }; +} + export function decodeSessionExecutionBoundaryQueryInput( value: unknown, ): SessionExecutionBoundaryQueryInput { diff --git a/packages/runtime-host/src/protocol/session-continuity.ts b/packages/runtime-host/src/protocol/session-continuity.ts index 116f875cad..5cd42d7927 100644 --- a/packages/runtime-host/src/protocol/session-continuity.ts +++ b/packages/runtime-host/src/protocol/session-continuity.ts @@ -293,7 +293,7 @@ export interface AgentGraphChangedFrame extends SubscriptionEnvelope { export interface SubscriptionClosedFrame extends SubscriptionEnvelope { kind: 'subscription.closed'; - reason: 'slow_consumer' | 'session_removed'; + reason: 'slow_consumer' | 'session_removed' | 'access_revoked'; } export type SubscriptionFrame = @@ -492,7 +492,11 @@ export function decodeSubscriptionFrame(value: unknown): SubscriptionFrame { 'sequence', 'reason', ]); - if (record.reason !== 'slow_consumer' && record.reason !== 'session_removed') { + if ( + record.reason !== 'slow_consumer' && + record.reason !== 'session_removed' && + record.reason !== 'access_revoked' + ) { throw invalidProtocolFrame('Invalid subscription close reason'); } frame = { kind: record.kind, ...envelope, reason: record.reason }; diff --git a/packages/runtime-host/src/protocol/session-transcript.ts b/packages/runtime-host/src/protocol/session-transcript.ts index 6369f73344..d54f791491 100644 --- a/packages/runtime-host/src/protocol/session-transcript.ts +++ b/packages/runtime-host/src/protocol/session-transcript.ts @@ -69,6 +69,8 @@ export interface SessionTranscriptPage { export interface SessionTranscriptBootstrap { readonly throughSequence: number | null; + /** Whether every durable sequence is present or policy projection may leave gaps. */ + readonly durableCoverage: 'complete' | 'projected'; readonly overlayMessageCount: number; readonly durable: SessionTranscriptPage; readonly overlay: SessionTranscriptPage; @@ -186,6 +188,7 @@ export function decodeSessionTranscriptPageInput(value: unknown): SessionTranscr export function decodeSessionTranscriptBootstrap(value: unknown): SessionTranscriptBootstrap { const bootstrap = requireExactRecord(value, 'Session transcript bootstrap', [ 'throughSequence', + 'durableCoverage', 'overlayMessageCount', 'durable', 'overlay', @@ -194,6 +197,9 @@ export function decodeSessionTranscriptBootstrap(value: unknown): SessionTranscr bootstrap.throughSequence === null ? null : requireCount(bootstrap.throughSequence, 'Session transcript watermark'); + if (bootstrap.durableCoverage !== 'complete' && bootstrap.durableCoverage !== 'projected') { + throw invalidProtocolFrame('Invalid Session transcript durable coverage'); + } const overlayMessageCount = requireCount( bootstrap.overlayMessageCount, 'Session transcript overlay message count', @@ -216,7 +222,13 @@ export function decodeSessionTranscriptBootstrap(value: unknown): SessionTranscr if (durable.rawBytes + overlay.rawBytes > SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES) { throw invalidProtocolFrame('Session transcript bootstrap exceeds byte limit'); } - return { throughSequence, overlayMessageCount, durable, overlay }; + return { + throughSequence, + durableCoverage: bootstrap.durableCoverage, + overlayMessageCount, + durable, + overlay, + }; } export function decodeSessionTranscriptPage(value: unknown): SessionTranscriptPage { diff --git a/packages/runtime-host/src/server/access-authority.ts b/packages/runtime-host/src/server/access-authority.ts index 0d25be84fc..3ccdfad22b 100644 --- a/packages/runtime-host/src/server/access-authority.ts +++ b/packages/runtime-host/src/server/access-authority.ts @@ -107,6 +107,10 @@ export interface RuntimeHostAccessAuthority { sessionId: string, kind: SessionCollaborationGrantKind, ): SessionCollaborationGrant | undefined; + activeSessionGrantForPrincipal( + principalId: string, + kind: SessionCollaborationGrantKind, + ): SessionCollaborationGrant | undefined; subscribeRevocations(listener: (credentialId: string) => void): () => void; subscribeGrantRevocations(listener: (grant: SessionCollaborationGrant) => void): () => void; close(): Promise; @@ -312,6 +316,15 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { ); } + activeSessionGrantForPrincipal( + principalId: string, + kind: SessionCollaborationGrantKind, + ): SessionCollaborationGrant | undefined { + return this.#file.sessionGrants.find( + (grant) => grant.principalId === principalId && grant.kind === kind, + ); + } + prepareRotation( input: AccessCredentialRotationPrepareInput, ): Promise { diff --git a/packages/runtime-host/src/server/access-credential-store.ts b/packages/runtime-host/src/server/access-credential-store.ts index 4b18e86cfa..e4fe1b819d 100644 --- a/packages/runtime-host/src/server/access-credential-store.ts +++ b/packages/runtime-host/src/server/access-credential-store.ts @@ -57,6 +57,13 @@ export const ACCESS_FILE_NAME = 'runtime-host-access.json'; export const SESSION_GUEST_OPERATION_GRANTS = Object.freeze([ 'host.status', + 'artifact.query', + 'runtime.resource.query', + 'session.shared.query', + 'subscription.open', + 'subscription.close', + 'session.transcript.page', + 'session.transcript.overlay.release', ] as const satisfies readonly OperationKey[]); export interface StoredAccessCredential { diff --git a/packages/runtime-host/src/server/artifact-coordinator.ts b/packages/runtime-host/src/server/artifact-coordinator.ts index 76cd0a4834..195d5fb40c 100644 --- a/packages/runtime-host/src/server/artifact-coordinator.ts +++ b/packages/runtime-host/src/server/artifact-coordinator.ts @@ -42,7 +42,8 @@ import { type OperationOutcome, } from '../protocol/index.js'; import { encodeArtifactProjection } from '../protocol/artifact.js'; -import type { ArtifactOperationHandlerMap } from './operation-dispatcher.js'; +import type { RuntimeHostAccessAuthority } from './access-authority.js'; +import type { ArtifactOperationHandlerMap, ConnectionContext } from './operation-dispatcher.js'; import { SessionAdmissionGate } from './session-admission-gate.js'; import type { SessionPresenceReader } from './session-presence.js'; import { ConnectionBoundChunkUploads } from './connection-bound-chunk-uploads.js'; @@ -50,6 +51,7 @@ import { ConnectionBoundChunkUploads } from './connection-bound-chunk-uploads.js const MAX_ACTIVE_ARTIFACT_UPLOADS = 16; const MAX_STAGED_ARTIFACT_UPLOAD_BYTES = 128 * 1024 * 1024; const ARTIFACT_UPLOAD_TTL_MS = 5 * 60 * 1000; +const SHARED_ARTIFACT_SOURCES = new Set(['user_upload', 'tool_result']); interface ArtifactUploadMetadata { readonly attachmentKind: AttachmentRef['kind']; @@ -63,8 +65,8 @@ export class HostArtifactCoordinator { readonly handlers: ArtifactOperationHandlerMap = { 'artifact.ingest': (input, context) => this.#sessionAdmission.run(input.sessionId, () => this.#ingest(input, context)), - 'artifact.query': (input) => - this.#sessionAdmission.run(input.sessionId, () => this.#query(input)), + 'artifact.query': (input, context) => + this.#sessionAdmission.run(input.sessionId, () => this.#query(input, context)), 'artifact.delete': (input) => this.#sessionAdmission.run(input.sessionId, () => this.#delete(input)), }; @@ -73,6 +75,9 @@ export class HostArtifactCoordinator { readonly #requestDrain: () => void; readonly #sessionAdmission: SessionAdmissionGate; readonly #sessions: SessionPresenceReader; + readonly #sessionAccessAuthority: + | Pick + | undefined; readonly #uploads: ConnectionBoundChunkUploads; constructor( @@ -81,11 +86,13 @@ export class HostArtifactCoordinator { sessionAdmission: SessionAdmissionGate, sessions: SessionPresenceReader, now: () => number = Date.now, + sessionAccessAuthority?: Pick, ) { this.#store = authenticateInteractiveArtifactStoreWriter(store); this.#requestDrain = requestDrain; this.#sessionAdmission = sessionAdmission; this.#sessions = sessions; + this.#sessionAccessAuthority = sessionAccessAuthority; this.#uploads = new ConnectionBoundChunkUploads( { maxActive: MAX_ACTIVE_ARTIFACT_UPLOADS, @@ -310,11 +317,19 @@ export class HostArtifactCoordinator { return { kind: 'committed', record }; } - async #query(input: ArtifactQueryInput): Promise> { + async #query( + input: ArtifactQueryInput, + context: ConnectionContext, + ): Promise> { try { if ((await this.#sessions.probeSessionRemoval(input.sessionId)).kind !== 'present') { return notFound('artifact.query', 'Session was not found'); } + let sharedGrantId: string | undefined; + if (context.principalKind === 'session_guest') { + sharedGrantId = await this.#sharedArtifactGrantId(context.principal, input); + if (!sharedGrantId) return notFound('artifact.query', 'Artifact was not found'); + } if (input.kind === 'read_text' || input.kind === 'read_binary') { if (input.kind === 'read_text') { const preview = await this.#store.readTextInSession(input.sessionId, input.artifactId, { @@ -356,6 +371,9 @@ export class HostArtifactCoordinator { } return persistenceFailure('artifact.query', 'Artifact content is unavailable'); } + if (!this.#sharedGrantRemainsActive(context.principal, input.sessionId, sharedGrantId)) { + return notFound('artifact.query', 'Artifact was not found'); + } return querySuccess( encodeArtifactQueryResult({ kind: 'chunk', @@ -371,6 +389,9 @@ export class HostArtifactCoordinator { if (input.kind === 'get') { const entry = await this.#store.getInSession(input.sessionId, input.artifactId); + if (!this.#sharedGrantRemainsActive(context.principal, input.sessionId, sharedGrantId)) { + return notFound('artifact.query', 'Artifact was not found'); + } return querySuccess( encodeArtifactQueryResult({ kind: 'artifact', @@ -410,6 +431,40 @@ export class HostArtifactCoordinator { } } + async #sharedArtifactGrantId( + principalId: string, + input: ArtifactQueryInput, + ): Promise { + if (input.kind !== 'get' && input.kind !== 'read_chunk') return; + const grant = this.#sessionAccessAuthority?.activeSessionGrant( + principalId, + input.sessionId, + 'session_observation', + ); + if (!grant) return; + const entry = await this.#store.getInSession(input.sessionId, input.artifactId); + return entry.record?.status === 'live' && + entry.record.source !== undefined && + SHARED_ARTIFACT_SOURCES.has(entry.record.source) + ? grant.grantId + : undefined; + } + + #sharedGrantRemainsActive( + principalId: string, + sessionId: string, + expectedGrantId: string | undefined, + ): boolean { + if (!expectedGrantId) return true; + return ( + this.#sessionAccessAuthority?.activeSessionGrant( + principalId, + sessionId, + 'session_observation', + )?.grantId === expectedGrantId + ); + } + async #delete(input: { readonly sessionId: string; readonly artifactId: string; diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts index 4f54f122d3..28b431f3d9 100644 --- a/packages/runtime-host/src/server/connection-session.ts +++ b/packages/runtime-host/src/server/connection-session.ts @@ -222,6 +222,7 @@ export class RuntimeHostConnectionSession { const response = await dispatchOperation(frame, this.#options.resolveHandlers(), { ...this.#options.connection, principal: this.#options.connection.authority.principalId, + principalKind: this.#options.connection.authority.principalKind, ...(this.#options.connection.authority.credentialId ? { credentialId: this.#options.connection.authority.credentialId } : {}), @@ -330,10 +331,11 @@ export class RuntimeHostConnectionSession { this.#options.connection.authority, 'project.catalog.query', ), - sessionCatalog: hasRuntimeHostOperationGrant( - this.#options.connection.authority, - 'session.catalog.query', - ), + sessionCatalog: + hasRuntimeHostOperationGrant( + this.#options.connection.authority, + 'session.catalog.query', + ) && this.#options.connection.authority.principalKind !== 'session_guest', scheduledTask: hasRuntimeHostOperationGrant( this.#options.connection.authority, 'scheduled-task.query', diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index cd1bb2b91e..edc2ff8418 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -378,6 +378,9 @@ export async function createExecutionRuntimeHostComposition( sessionAdmission, acquireResidency: () => context.acquireResidency('runtime-resource'), requestDrain: context.requestDrain, + ...(context.sessionAccessAuthority + ? { sessionAccessAuthority: context.sessionAccessAuthority } + : {}), resolveShell: async () => resolveShellPlan((await runtimePolicyStores.runtimePolicy.getSnapshot()).policy.shell), onProjectionChanged: (update) => @@ -556,6 +559,7 @@ export async function createExecutionRuntimeHostComposition( context.requestDrain, createSessionTranscriptReader({ stores, canonicalPermissionOutcomes }), (sessionId) => hostChanges.publishSessionCatalog(sessionId), + context.sessionAccessAuthority, ); const continuityCoordinator = continuity; unsubscribeTranscriptChanges = stores.sessionStore.subscribeTranscriptChanges((sessionId) => @@ -1108,6 +1112,8 @@ export async function createExecutionRuntimeHostComposition( context.requestDrain, sessionAdmission, stores.sessionStore, + Date.now, + context.sessionAccessAuthority, ); rootCoordinator = new RootTurnCoordinator( manager, @@ -1270,6 +1276,9 @@ export async function createExecutionRuntimeHostComposition( continuity: continuityCoordinator, workspaceResolver, requestDrain: context.requestDrain, + ...(context.sessionAccessAuthority + ? { sessionAccessAuthority: context.sessionAccessAuthority } + : {}), }); const workHubCoordination = new HostWorkHubCoordinationCoordinator({ stateRoot: context.owner.capability.canonicalPath, diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index d7b4e67a57..004bf6ad6a 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -119,6 +119,10 @@ export interface RuntimeHostCompositionContext { /** Irreversible fail-stop latch; normal residency still uses acquireResidency(). */ retainUntilProcessExit(): void; requestDrain(): void; + sessionAccessAuthority?: Pick< + RuntimeHostAccessAuthority, + 'activeSessionGrant' | 'activeSessionGrantForPrincipal' | 'subscribeGrantRevocations' + >; waitForResidencies?(): Promise; waitForResidenciesExcept?(excludedLabel: string): Promise; } @@ -346,6 +350,9 @@ export class RuntimeHostKernel { acquireResidency: (label) => this.#acquireResidency(label), retainUntilProcessExit: () => this.#retainUntilProcessExit(), requestDrain: () => this.#requestDrain(), + ...(this.#options.accessAuthority + ? { sessionAccessAuthority: this.#options.accessAuthority } + : {}), waitForResidencies: () => this.#waitForResidencies(), waitForResidenciesExcept: (excludedLabel) => this.#waitForResidenciesExcept(excludedLabel), diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index 552233e799..856afe48bd 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -36,11 +36,13 @@ import { ACCESS_AUTHORITY_OPERATION_SPECS } from '../protocol/access-authority.j import { SESSION_COLLABORATION_OPERATION_SPECS } from '../protocol/session-collaboration.js'; import { PEER_MESH_OPERATION_SPECS } from '../protocol/peer-mesh.js'; import { createPeerMeshOperationHandlers } from './peer-mesh-authority.js'; +import type { RuntimeHostConnectionAuthority } from './connection-authority.js'; export interface ConnectionContext { hostEpoch: string; connectionId: string; principal: string; + principalKind?: RuntimeHostConnectionAuthority['principalKind']; credentialId?: string; clientInstanceId?: string; acquireResidency(): OperationResidency; diff --git a/packages/runtime-host/src/server/runtime-resource-coordinator.ts b/packages/runtime-host/src/server/runtime-resource-coordinator.ts index 6dd125c54a..95499ef4ad 100644 --- a/packages/runtime-host/src/server/runtime-resource-coordinator.ts +++ b/packages/runtime-host/src/server/runtime-resource-coordinator.ts @@ -59,6 +59,7 @@ import type { ConnectionContext, RuntimeResourceOperationHandlerMap, } from './operation-dispatcher.js'; +import type { RuntimeHostAccessAuthority } from './access-authority.js'; import { SessionAdmissionGate } from './session-admission-gate.js'; import { boundedRuntimeResourceSnapshot, @@ -99,6 +100,7 @@ export interface HostRuntimeResourceCoordinatorInput { readonly sessionAdmission: SessionAdmissionGate; readonly acquireResidency: () => RuntimeHostResidency; readonly requestDrain: () => void; + readonly sessionAccessAuthority?: Pick; readonly onProjectionChanged?: (update: ShellRunUpdate) => void; /** * Fallback shell resolution for callers that do not carry a plan (e.g. @@ -129,7 +131,7 @@ export class HostRuntimeResourceCoordinator implements ShellRunLauncher, RuntimeResourceReader, BackgroundTaskStopper, PtyControlWriter { readonly handlers: RuntimeResourceOperationHandlerMap = { - 'runtime.resource.query': (input) => this.#query(input), + 'runtime.resource.query': (input, context) => this.#query(input, context), 'runtime.resource.start': (input) => this.#start(input), 'runtime.resource.controller.acquire': (input, context) => this.#acquire(input, context), 'runtime.resource.controller.control': (input, context) => this.#control(input, context), @@ -143,6 +145,9 @@ export class HostRuntimeResourceCoordinator readonly #sessionAdmission: SessionAdmissionGate; readonly #acquireResidency: () => RuntimeHostResidency; readonly #requestDrain: () => void; + readonly #sessionAccessAuthority: + | Pick + | undefined; readonly #onProjectionChanged: (update: ShellRunUpdate) => void; readonly #resolveShell: () => Promise | ShellPlan; readonly #resourceQueue = new ResourceSerialQueue(); @@ -159,6 +164,7 @@ export class HostRuntimeResourceCoordinator this.#sessionAdmission = input.sessionAdmission; this.#acquireResidency = input.acquireResidency; this.#requestDrain = input.requestDrain; + this.#sessionAccessAuthority = input.sessionAccessAuthority; this.#onProjectionChanged = input.onProjectionChanged ?? (() => undefined); this.#resolveShell = input.resolveShell ?? defaultShellPlan; } @@ -285,73 +291,100 @@ export class HostRuntimeResourceCoordinator return updates.some((update) => isActiveShellRunStatus(update.result.status)); } - #query(input: RuntimeResourceQueryInput): Promise> { + async #query( + input: RuntimeResourceQueryInput, + context: ConnectionContext, + ): Promise> { if (input.kind === 'get' && !isShellRunResourceRef(input.ref)) { - return Promise.resolve( - queryFailure('invalid_request', 'Runtime Resource ref is unsupported'), - ); + return queryFailure('invalid_request', 'Runtime Resource ref is unsupported'); } - return this.#sessionAdmission.run(input.sessionId, async () => { - try { - await this.#sessionHeaders.readHeader(input.sessionId); - } catch (error) { - if (isSessionNotFoundError(error)) { - return queryFailure('not_found', 'Session was not found'); + const guestGrantId = this.#guestObservationGrantId(context, input.sessionId); + if (context.principalKind === 'session_guest' && !guestGrantId) { + return queryFailure('not_found', 'Session was not found'); + } + const outcome: OperationOutcome<'runtime.resource.query'> = await this.#sessionAdmission.run( + input.sessionId, + async () => { + try { + await this.#sessionHeaders.readHeader(input.sessionId); + } catch (error) { + if (isSessionNotFoundError(error)) { + return queryFailure('not_found', 'Session was not found'); + } + this.#requestDrain(); + return queryFailure('internal_failure', 'Session state is unavailable'); } - this.#requestDrain(); - return queryFailure('internal_failure', 'Session state is unavailable'); - } - if (input.kind === 'get') { + if (input.kind === 'get') { + try { + const resource = await this.#sessions.getShellRunUpdate(input.sessionId, input.ref); + const visible = + context.principalKind !== 'session_guest' || resource?.sessionId === input.sessionId + ? resource + : null; + const canonical = visible ? (canonicalRuntimeResources([visible])[0] ?? null) : null; + return { + ok: true, + result: decodeRuntimeResourceQueryResult({ + kind: 'resource', + sessionId: input.sessionId, + revision: runtimeResourceRevision(canonical ? [canonical] : []), + resource: canonical, + }), + }; + } catch { + this.#requestDrain(); + return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); + } + } + let updates: ShellRunUpdate[]; try { - const resource = await this.#sessions.getShellRunUpdate(input.sessionId, input.ref); - const canonical = resource ? (canonicalRuntimeResources([resource])[0] ?? null) : null; - return { - ok: true, - result: decodeRuntimeResourceQueryResult({ - kind: 'resource', - sessionId: input.sessionId, - revision: runtimeResourceRevision(canonical ? [canonical] : []), - resource: canonical, - }), - }; + updates = await this.#sessions.listShellRunUpdates(input.sessionId); + if (context.principalKind === 'session_guest') { + updates = updates.filter((update) => update.sessionId === input.sessionId); + } } catch { this.#requestDrain(); return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); } - } - let updates: ShellRunUpdate[]; - try { - updates = await this.#sessions.listShellRunUpdates(input.sessionId); - } catch { - this.#requestDrain(); - return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); - } - try { - const resources = canonicalRuntimeResources(updates); - const revision = runtimeResourceRevision(resources); - if (input.kind === 'list_continue' && input.revision !== revision) { + try { + const resources = canonicalRuntimeResources(updates); + const revision = runtimeResourceRevision(resources); + if (input.kind === 'list_continue' && input.revision !== revision) { + return { + ok: true, + result: { kind: 'revision_changed', expected: input.revision, actual: revision }, + }; + } + const offset = input.kind === 'list_start' ? 0 : decodeCursor(input.cursor); + if ( + offset === undefined || + offset > resources.length || + (input.kind === 'list_continue' && offset === 0) || + (input.kind === 'list_continue' && offset === resources.length) + ) { + return queryFailure('invalid_request', 'Runtime Resource cursor is invalid'); + } return { ok: true, - result: { kind: 'revision_changed', expected: input.revision, actual: revision }, + result: createRuntimeResourcePage(input.sessionId, revision, resources, offset), }; + } catch { + return queryFailure('internal_failure', 'Runtime Resource projection is unavailable'); } - const offset = input.kind === 'list_start' ? 0 : decodeCursor(input.cursor); - if ( - offset === undefined || - offset > resources.length || - (input.kind === 'list_continue' && offset === 0) || - (input.kind === 'list_continue' && offset === resources.length) - ) { - return queryFailure('invalid_request', 'Runtime Resource cursor is invalid'); - } - return { - ok: true, - result: createRuntimeResourcePage(input.sessionId, revision, resources, offset), - }; - } catch { - return queryFailure('internal_failure', 'Runtime Resource projection is unavailable'); - } - }); + }, + ); + return guestGrantId && this.#guestObservationGrantId(context, input.sessionId) !== guestGrantId + ? queryFailure('not_found', 'Session was not found') + : outcome; + } + + #guestObservationGrantId(context: ConnectionContext, sessionId: string): string | undefined { + if (context.principalKind !== 'session_guest') return; + return this.#sessionAccessAuthority?.activeSessionGrant( + context.principal, + sessionId, + 'session_observation', + )?.grantId; } async #start( diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts index 10dd15752b..4069522ce2 100644 --- a/packages/runtime-host/src/server/session-catalog-coordinator.ts +++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts @@ -58,6 +58,7 @@ import { } from '@maka/runtime/session-manager'; import { decodeSessionCatalogProjection, + decodeSharedSessionCatalogProjection, SESSION_CATALOG_LIVE_RUN_STATE_SCHEMA_VERSION, SESSION_CATALOG_LABEL_MAX_BYTES, SESSION_CATALOG_LABEL_MAX_ITEMS, @@ -70,6 +71,7 @@ import { type SessionCatalogItem, type SessionCatalogLiveRunState, type SessionCatalogProjection, + type SharedSessionCatalogProjection, type SessionCatalogQueryInput, type SessionCatalogQueryResult, type SessionCatalogRevision, @@ -88,6 +90,7 @@ import { projectSessionTurnContributionForWire, } from '../protocol/index.js'; import type { SessionCatalogOperationHandlerMap } from './operation-dispatcher.js'; +import type { RuntimeHostAccessAuthority } from './access-authority.js'; import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admission-gate.js'; import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js'; import { type HostWorkspaceResolver, WorkspaceResolutionError } from './workspace-resolver.js'; @@ -154,6 +157,10 @@ export interface HostSessionCatalogCoordinatorOptions { readonly continuity: SessionContinuity; readonly workspaceResolver: HostWorkspaceResolver; readonly requestDrain: () => void; + readonly sessionAccessAuthority?: Pick< + RuntimeHostAccessAuthority, + 'activeSessionGrantForPrincipal' + >; } interface ResolvedSessionModel { @@ -165,6 +172,7 @@ interface ResolvedSessionModel { /** Host-owned Session catalog, creation, and configuration authority. */ export class HostSessionCatalogCoordinator { readonly handlers: SessionCatalogOperationHandlerMap = { + 'session.shared.query': (_input, context) => this.#querySharedSession(context.principal), 'session.catalog.query': (input) => this.#query(input), 'session.create': (input) => this.#create(input), 'session.metadata.update': (input) => this.#updateMetadata(input), @@ -183,6 +191,9 @@ export class HostSessionCatalogCoordinator { readonly #continuity: SessionContinuity; readonly #workspaceResolver: HostWorkspaceResolver; readonly #requestDrain: () => void; + readonly #sessionAccessAuthority: + | Pick + | undefined; constructor(options: HostSessionCatalogCoordinatorOptions) { this.#stores = options.stores; @@ -192,6 +203,7 @@ export class HostSessionCatalogCoordinator { this.#continuity = options.continuity; this.#workspaceResolver = options.workspaceResolver; this.#requestDrain = options.requestDrain; + this.#sessionAccessAuthority = options.sessionAccessAuthority; } async resolveExternalSessionImportTarget(): Promise> { @@ -286,6 +298,48 @@ export class HostSessionCatalogCoordinator { } } + async #querySharedSession( + principalId: string, + ): Promise> { + if (!this.#sessionAccessAuthority) { + return { + ok: false, + error: { code: 'operation_unavailable', message: 'Session sharing is unavailable' }, + }; + } + const grant = this.#sessionAccessAuthority.activeSessionGrantForPrincipal( + principalId, + 'session_observation', + ); + if (!grant) return { ok: true, result: { session: null } }; + try { + const record = await this.#readCatalogRecordIfPresent(grant.sessionId); + const currentGrant = this.#sessionAccessAuthority.activeSessionGrantForPrincipal( + principalId, + 'session_observation', + ); + if (currentGrant?.grantId !== grant.grantId) { + return { ok: true, result: { session: null } }; + } + return { + ok: true, + result: { + session: record + ? projectSharedSessionCatalogRecord( + record, + projectCatalogLiveRunState(this.#manager.runningTurnIds(record.header.id)), + ) + : null, + }, + }; + } catch { + return { + ok: false, + error: { code: 'persistence_failed', message: 'Shared Session catalog is unavailable' }, + }; + } + } + #projectCatalogQueryRecord(record: SessionCatalogRecord): SessionCatalogItem { return projectSessionCatalogRecord( record, @@ -1107,6 +1161,30 @@ export function projectSessionCatalogRecord( } } +function projectSharedSessionCatalogRecord( + record: SessionCatalogRecord, + liveRunState?: SessionCatalogLiveRunState, +): SharedSessionCatalogProjection { + const { header, summary } = record; + const shared: SharedSessionCatalogProjection = { + kind: 'shared_session', + id: header.id, + revision: record.revision, + createdAt: header.createdAt, + activityAt: record.activityAt, + name: header.name, + ...(summary.lastMessageAt === undefined ? {} : { lastMessageAt: summary.lastMessageAt }), + ...(summary.lastMessagePreview === undefined + ? {} + : { lastMessagePreview: summary.lastMessagePreview }), + status: header.status, + ...(liveRunState === undefined ? {} : { liveRunState }), + ...(header.blockedReason === undefined ? {} : { blockedReason: header.blockedReason }), + ...(header.statusUpdatedAt === undefined ? {} : { statusUpdatedAt: header.statusUpdatedAt }), + }; + return decodeSharedSessionCatalogProjection(shared); +} + function projectCatalogLiveRunState( runningTurnIds: readonly string[], ): SessionCatalogLiveRunState | undefined { diff --git a/packages/runtime-host/src/server/session-continuity-coordinator.ts b/packages/runtime-host/src/server/session-continuity-coordinator.ts index cb3b9fc815..2596e1a8af 100644 --- a/packages/runtime-host/src/server/session-continuity-coordinator.ts +++ b/packages/runtime-host/src/server/session-continuity-coordinator.ts @@ -54,7 +54,11 @@ import { type TurnProviderRetry, type TurnSnapshot, } from '../protocol/index.js'; -import type { SessionContinuityOperationHandlerMap } from './operation-dispatcher.js'; +import type { + ConnectionContext, + SessionContinuityOperationHandlerMap, +} from './operation-dispatcher.js'; +import type { RuntimeHostAccessAuthority } from './access-authority.js'; import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admission-gate.js'; import { type CanonicalSessionProjection, @@ -77,6 +81,7 @@ import { ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES, type SessionTranscriptReader, } from './session-transcript-reader.js'; +import { projectSharedSessionMessageContent } from './shared-session-transcript.js'; const MAX_CONNECTION_SUBSCRIPTIONS = 16; const MAX_SUBSCRIBER_QUEUED_FRAMES = 32; @@ -152,6 +157,8 @@ interface QueuedSubscriptionFrame { interface Subscriber { connectionId: string; + principalId: string; + principalKind: NonNullable; sessionId: string; subscriptionId: string; sink: SessionContinuityFrameSink; @@ -230,7 +237,7 @@ interface PendingSessionDomainChanges { export class SessionContinuityCoordinator implements SessionContinuityService { readonly handlers: SessionContinuityOperationHandlerMap = { 'subscription.open': async (input, context) => { - const result = await this.#open(context.connectionId, input); + const result = await this.#open(context, input); return result.ok ? { ok: true, result: result.value } : { ok: false, error: { code: result.code, message: result.message } }; @@ -277,6 +284,10 @@ export class SessionContinuityCoordinator implements SessionContinuityService { #closed = false; #preparingTranscriptOverlayBytes = 0; #retainedTranscriptOverlayBytes = 0; + readonly #sessionAccessAuthority: + | Pick + | undefined; + readonly #unsubscribeGrantRevocations: (() => void) | undefined; constructor( hostEpoch: string, @@ -285,10 +296,28 @@ export class SessionContinuityCoordinator implements SessionContinuityService { private readonly onPublicationFailure: (error: unknown) => void = () => undefined, transcriptReader?: SessionTranscriptReader, private readonly onCatalogChanged: (sessionId: string) => void = () => undefined, + sessionAccessAuthority?: Pick< + RuntimeHostAccessAuthority, + 'activeSessionGrant' | 'subscribeGrantRevocations' + >, ) { this.#hostEpoch = hostEpoch; this.#readCanonical = readCanonical; this.#transcriptReader = transcriptReader; + this.#sessionAccessAuthority = sessionAccessAuthority; + this.#unsubscribeGrantRevocations = sessionAccessAuthority?.subscribeGrantRevocations( + (grant) => { + if (grant.kind !== 'session_observation') return; + for (const subscriber of this.#subscriptions.values()) { + if ( + subscriber.principalId === grant.principalId && + subscriber.sessionId === grant.sessionId + ) { + this.#closeSubscriber(subscriber, 'access_revoked'); + } + } + }, + ); } attachConnection( @@ -394,6 +423,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { const state = this.#sessions.get(event.rootSessionId); if (!state) return; for (const subscriber of state.subscribers.values()) { + if (subscriber.principalKind === 'session_guest') continue; const frame: AgentGraphChangedFrame = { kind: 'subscription.agent_graph_changed', hostEpoch: this.#hostEpoch, @@ -518,12 +548,24 @@ export class SessionContinuityCoordinator implements SessionContinuityService { } for (const change of frames) { for (const subscriber of state.subscribers.values()) { + const projected = + change.domain === 'runtime_resource' && subscriber.principalKind === 'session_guest' + ? { + ...change, + resources: change.resources.filter( + (resource) => resource.sourceSessionId === subscriber.sessionId, + ), + } + : change; + if (projected.domain === 'runtime_resource' && projected.resources.length === 0) { + continue; + } const frame: SessionDomainChangedFrame = { kind: 'subscription.session_domain_changed', hostEpoch: this.#hostEpoch, subscriptionId: subscriber.subscriptionId, sequence: subscriber.nextSequence, - ...change, + ...projected, }; this.#enqueue(subscriber, frame); } @@ -739,7 +781,6 @@ export class SessionContinuityCoordinator implements SessionContinuityService { } else if (event.type === 'tool_result') { state.toolResultPreviews.delete(event.toolUseId); } - const projected = projectSessionEvent(event); for (const subscriber of state.subscribers.values()) { const frame: SessionEventFrame = { kind: 'subscription.session_event', @@ -748,7 +789,11 @@ export class SessionContinuityCoordinator implements SessionContinuityService { sequence: subscriber.nextSequence, sessionId, runId, - event: projected, + event: projectSessionEvent( + event, + sessionId, + subscriber.principalKind === 'session_guest', + ), }; this.#enqueue(subscriber, frame); } @@ -779,6 +824,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { close(): void { if (this.#closed) return; this.#closed = true; + this.#unsubscribeGrantRevocations?.(); this.#cancelTranscriptOverlayPreparationWaiters(); for (const connectionId of [...this.#connections.keys()]) this.#closeConnection(connectionId); for (const state of this.#sessions.values()) this.#invalidateTranscriptOverlay(state); @@ -790,7 +836,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { } async #open( - connectionId: string, + context: ConnectionContext, input: SubscriptionOpenInput, ): Promise< | { ok: true; value: SubscriptionOpenResult } @@ -800,9 +846,14 @@ export class SessionContinuityCoordinator implements SessionContinuityService { message: string; } > { + const connectionId = context.connectionId; + const identity = connectionIdentity(context); const sessionId = input.sessionId; const connection = this.#connections.get(connectionId); if (!connection) throw new Error('Runtime Host connection is not attached to continuity'); + if (!this.#canObserve(identity, sessionId)) { + return { ok: false, code: 'not_found', message: 'Session was not found' }; + } if ( connection.subscriptionIds.size + connection.pendingOpenCount >= MAX_CONNECTION_SUBSCRIPTIONS @@ -878,6 +929,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { throw new Error('Runtime Host connection closed during subscription open'); }), ]); + const snapshot = projectSessionSnapshot(committed.value, identity.principalKind); const created = await createSessionTranscriptBootstrap({ reader: this.#transcriptReader, sessionId, @@ -887,11 +939,12 @@ export class SessionContinuityCoordinator implements SessionContinuityService { activeAssistantStreams: committed.state.assistantStreams.values(), maxBytes: input.transcript.maxBytes, preparedOverlayMessages: retainedTranscriptOverlay.messages, + projection: identity.principalKind === 'session_guest' ? 'shared' : 'owner', maxEncodedBytes: subscriptionOpenTranscriptBudget({ hostEpoch: this.#hostEpoch, subscriptionId, nextSequence: 1, - snapshot: committed.value, + snapshot, activeAssistantStreams, transcript: null, }), @@ -921,7 +974,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { hostEpoch: this.#hostEpoch, subscriptionId, nextSequence: 1, - snapshot: committed.value, + snapshot: projectSessionSnapshot(committed.value, identity.principalKind), activeAssistantStreams, transcript: transcriptBootstrap, }; @@ -935,8 +988,17 @@ export class SessionContinuityCoordinator implements SessionContinuityService { message: 'Session subscription state exceeds the transport limit', }; } + if (!this.#canObserve(identity, sessionId)) { + return { + ok: false as const, + code: 'not_found' as const, + message: 'Session was not found', + }; + } const subscriber: Subscriber = { connectionId, + principalId: identity.principalId, + principalKind: identity.principalKind, sessionId, subscriptionId, sink: connection.sink, @@ -973,7 +1035,11 @@ export class SessionContinuityCoordinator implements SessionContinuityService { sequence: subscriber.nextSequence, sessionId, runId: rootTurn.runId, - event: projectSessionEvent(preview), + event: projectSessionEvent( + preview, + sessionId, + subscriber.principalKind === 'session_guest', + ), }; this.#enqueue(subscriber, frame); } @@ -1041,9 +1107,18 @@ export class SessionContinuityCoordinator implements SessionContinuityService { error: { code: 'operation_unavailable', message: 'Session transcript is unavailable' }, }; } + const connection = this.#connections.get(connectionId); + if (!connection || !this.#canObserve(subscriber, subscriber.sessionId)) { + this.#closeSubscriber(subscriber, 'access_revoked'); + return transcriptSubscriptionNotFound(); + } const transcript = subscriber.transcript; return this.sessionAdmission.run(subscriber.sessionId, async () => { - if (this.#ownedSubscriber(connectionId, input.subscriptionId) !== subscriber) { + if ( + this.#ownedSubscriber(connectionId, input.subscriptionId) !== subscriber || + this.#connections.get(connectionId) !== connection || + !this.#canObserve(subscriber, subscriber.sessionId) + ) { return transcriptSubscriptionNotFound(); } try { @@ -1052,7 +1127,11 @@ export class SessionContinuityCoordinator implements SessionContinuityService { state: transcript, request: input, }); - if (this.#ownedSubscriber(connectionId, input.subscriptionId) !== subscriber) { + if ( + this.#ownedSubscriber(connectionId, input.subscriptionId) !== subscriber || + this.#connections.get(connectionId) !== connection || + !this.#canObserve(subscriber, subscriber.sessionId) + ) { return transcriptSubscriptionNotFound(); } return { ok: true, result: page }; @@ -1426,6 +1505,10 @@ export class SessionContinuityCoordinator implements SessionContinuityService { } #evictSlowSubscriber(subscriber: Subscriber): void { + this.#closeSubscriber(subscriber, 'slow_consumer'); + } + + #closeSubscriber(subscriber: Subscriber, reason: 'slow_consumer' | 'access_revoked'): void { if (subscriber.phase !== 'open') return; subscriber.phase = 'closing'; const inFlight = subscriber.pumping ? subscriber.queue[0] : undefined; @@ -1437,7 +1520,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { hostEpoch: this.#hostEpoch, subscriptionId: subscriber.subscriptionId, sequence: subscriber.nextSequence, - reason: 'slow_consumer', + reason, }; subscriber.nextSequence += 1; subscriber.terminalQueued = true; @@ -1451,6 +1534,20 @@ export class SessionContinuityCoordinator implements SessionContinuityService { if (subscriber.activated) this.#pump(subscriber); } + #canObserve( + identity: { readonly principalId: string; readonly principalKind: Subscriber['principalKind'] }, + sessionId: string, + ): boolean { + return ( + identity.principalKind !== 'session_guest' || + this.#sessionAccessAuthority?.activeSessionGrant( + identity.principalId, + sessionId, + 'session_observation', + ) !== undefined + ); + } + #enqueueAssistantDelta( subscriber: Subscriber, sessionId: string, @@ -1741,7 +1838,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { hostEpoch: this.#hostEpoch, subscriptionId: subscriber.subscriptionId, sequence: subscriber.nextSequence, - snapshot, + snapshot: projectSessionSnapshot(snapshot, subscriber.principalKind), }); } } @@ -1925,6 +2022,37 @@ function jsonStringContentBytes(value: string): number { return Buffer.byteLength(encoded.slice(1, -1), 'utf8'); } +function connectionIdentity(context: ConnectionContext): { + readonly principalId: string; + readonly principalKind: Subscriber['principalKind']; +} { + if (!context.principalKind) { + throw new Error('Runtime Host connection has no authenticated principal kind'); + } + return { principalId: context.principal, principalKind: context.principalKind }; +} + +function projectSessionSnapshot( + snapshot: SessionContinuitySnapshot, + principalKind: Subscriber['principalKind'], +): SessionContinuitySnapshot { + if (principalKind !== 'session_guest') return snapshot; + const projectEntry = ( + entry: T, + ): T => ({ + ...entry, + content: projectSharedSessionMessageContent(entry.content, snapshot.session.sessionId), + }); + return { + ...snapshot, + queue: { + ...snapshot.queue, + steering: snapshot.queue.steering.map(projectEntry), + followup: snapshot.queue.followup.map(projectEntry), + }, + }; +} + function projectSessionEvent( event: Exclude< RuntimeSessionForwardedEvent, @@ -1937,6 +2065,8 @@ function projectSessionEvent( | 'provider_retry'; } >, + sessionId: string, + shared = false, ): SessionToolEvent | SessionSteeringEvent { if (event.type === 'steering_message') { // The durable steering echo: forwarded verbatim so subscribers render the @@ -1948,7 +2078,9 @@ function projectSessionEvent( turnId: event.turnId, ts: event.ts, messageId: event.messageId, - content: structuredClone(event.content), + content: shared + ? projectSharedSessionMessageContent(event.content, sessionId) + : structuredClone(event.content), }; } const identity = { diff --git a/packages/runtime-host/src/server/session-transcript-pager.ts b/packages/runtime-host/src/server/session-transcript-pager.ts index 01f5618bc8..5c7dba3e0e 100644 --- a/packages/runtime-host/src/server/session-transcript-pager.ts +++ b/packages/runtime-host/src/server/session-transcript-pager.ts @@ -34,6 +34,9 @@ import { ACTIVE_TRANSCRIPT_OVERLAY_MAX_MESSAGES, type SessionTranscriptReader, } from './session-transcript-reader.js'; +import { projectSharedSessionTranscriptMessage } from './shared-session-transcript.js'; + +type SessionTranscriptProjection = 'owner' | 'shared'; interface TranscriptCursorState { readonly version: 1; @@ -53,6 +56,7 @@ export interface SubscriberTranscriptState { overlayMessages: readonly Buffer[] | undefined; readonly cursorSecret: Buffer; durableThroughSequence: number | null; + readonly projection: SessionTranscriptProjection; } export interface ActiveTranscriptAssistantStream { @@ -78,9 +82,17 @@ export async function createSessionTranscriptBootstrap(input: { maxBytes: number; maxEncodedBytes?: number; preparedOverlayMessages?: readonly Buffer[]; + projection: SessionTranscriptProjection; }): Promise<{ bootstrap: SessionTranscriptBootstrap; state: SubscriberTranscriptState }> { - const overlayMessages = + const projection = input.projection; + const preparedOverlayMessages = input.preparedOverlayMessages ?? (await prepareSessionTranscriptOverlay(input)); + const overlayMessages = + projection === 'shared' + ? preparedOverlayMessages.flatMap((message) => + projectEncodedSharedMessage(message, input.sessionId), + ) + : preparedOverlayMessages; const cursorSecret = randomBytes(32); let rawBudget = input.maxBytes; for (;;) { @@ -93,12 +105,16 @@ export async function createSessionTranscriptBootstrap(input: { overlayBudget, ); const durableBudget = rawBudget - selectedOverlay.rawBytes; - const durableStorage = await input.reader.readDurablePage(input.sessionId, { + const durableRequest = { direction: 'older', throughSequence: input.throughSequence, maxBytes: durableBudget, maxMessages: SESSION_TRANSCRIPT_PAGE_MAX_MESSAGES, - }); + } as const; + const durableStorage = + projection === 'shared' + ? await readSharedDurablePage(input.reader, input.sessionId, durableRequest) + : await input.reader.readDurablePage(input.sessionId, durableRequest); if (durableStorage.throughSequence !== input.throughSequence) { throw new Error('Session transcript durable watermark changed during bootstrap'); } @@ -109,9 +125,11 @@ export async function createSessionTranscriptBootstrap(input: { durableThroughSequence: input.throughSequence, overlayMessages, cursorSecret, + projection, }; const bootstrap: SessionTranscriptBootstrap = { throughSequence: input.throughSequence, + durableCoverage: projection === 'shared' ? 'projected' : 'complete', overlayMessageCount: overlayMessages.length, durable: pageFromSelection(state, 'durable', 'older', storageSelection(durableStorage)), overlay: pageFromSelection(state, 'overlay', 'older', selectedOverlay), @@ -193,14 +211,18 @@ export async function readSessionTranscriptPage(input: { ); } if (request.throughSequence === null) return emptyPage(state, request); - const storage = await input.reader.readDurablePage(state.sessionId, { + const durableRequest = { direction: request.direction, throughSequence: request.throughSequence, position: position.position, ...(position.byteOffset === null ? {} : { byteOffset: position.byteOffset }), maxBytes: request.maxBytes, maxMessages: SESSION_TRANSCRIPT_PAGE_MAX_MESSAGES, - }); + } as const; + const storage = + state.projection === 'shared' + ? await readSharedDurablePage(input.reader, state.sessionId, durableRequest) + : await input.reader.readDurablePage(state.sessionId, durableRequest); return pageFromSelection( state, 'durable', @@ -210,6 +232,82 @@ export async function readSessionTranscriptPage(input: { ); } +async function readSharedDurablePage( + reader: SessionTranscriptReader, + sessionId: string, + request: Parameters[1], +): ReturnType { + const position = + request.position ?? + (request.direction === 'older' ? (request.throughSequence ?? undefined) : 0); + const scanned = await reader.readDurableRecords(sessionId, { + direction: request.direction, + ...(request.throughSequence === undefined ? {} : { throughSequence: request.throughSequence }), + ...(position === undefined ? {} : { position }), + maxStoredBytes: ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES, + maxMessages: SESSION_TRANSCRIPT_PAGE_MAX_MESSAGES, + }); + const fragments: Awaited< + ReturnType + >['fragments'][number][] = []; + let rawBytes = 0; + let next: { position: number; byteOffset: number | null } | null = null; + let recordIndex = 0; + for (; recordIndex < scanned.records.length; recordIndex += 1) { + const record = scanned.records[recordIndex]!; + const projected = projectSharedSessionTranscriptMessage(record.message, sessionId); + if (!projected) continue; + const bytes = Buffer.from(JSON.stringify(projected), 'utf8'); + const continuationOffset = + record.sequence === position && request.byteOffset !== undefined ? request.byteOffset : null; + const selected = selectBuffer( + bytes, + request.direction, + continuationOffset, + request.maxBytes - rawBytes, + ); + if (!selected) break; + fragments.push({ + sequence: record.sequence, + byteOffset: selected.byteOffset, + totalBytes: bytes.byteLength, + payloadDigest: null, + data: selected.data, + }); + rawBytes += selected.data.byteLength; + if (!selected.complete) { + next = { position: record.sequence, byteOffset: selected.nextOffset }; + break; + } + if (fragments.length === request.maxMessages || rawBytes === request.maxBytes) { + recordIndex += 1; + break; + } + } + if (!next) { + next = + recordIndex < scanned.records.length + ? { position: scanned.records[recordIndex]!.sequence, byteOffset: null } + : scanned.nextPosition === null + ? null + : { position: scanned.nextPosition, byteOffset: null }; + } + return { + throughSequence: scanned.throughSequence, + fragments, + rawBytes, + next, + }; +} + +function projectEncodedSharedMessage(bytes: Buffer, sessionId: string): Buffer[] { + const projected = projectSharedSessionTranscriptMessage( + JSON.parse(bytes.toString('utf8')), + sessionId, + ); + return projected ? [Buffer.from(JSON.stringify(projected), 'utf8')] : []; +} + export function updateSubscriberTranscriptHighWater( state: SubscriberTranscriptState, throughSequence: number | null, diff --git a/packages/runtime-host/src/server/session-transcript-reader.ts b/packages/runtime-host/src/server/session-transcript-reader.ts index 311cf5dc91..442d3b8851 100644 --- a/packages/runtime-host/src/server/session-transcript-reader.ts +++ b/packages/runtime-host/src/server/session-transcript-reader.ts @@ -32,6 +32,8 @@ import type { ExecutionStoresWriter, SessionTranscriptMessageLookupRequest, SessionTranscriptPageRequest, + SessionTranscriptRecordScanPage, + SessionTranscriptRecordScanRequest, SessionTranscriptStoragePage, } from '@maka/storage/execution-stores'; import { SESSION_TRANSCRIPT_OVERLAY_MAX_MESSAGES, type TurnSnapshot } from '../protocol/index.js'; @@ -51,6 +53,8 @@ export function createSessionTranscriptReader(input: { input.stores.sessionStore.readTranscriptHighWaterSnapshot(sessionId), readDurablePage: (sessionId, request) => input.stores.sessionStore.readTranscriptPageSnapshot(sessionId, request), + readDurableRecords: (sessionId, request) => + input.stores.sessionStore.readTranscriptRecordsSnapshot(sessionId, request), readDurableMessagesById: (sessionId, request) => input.stores.sessionStore.readTranscriptMessagesSnapshot(sessionId, request), readActiveOverlay: async (sessionId, rootTurn) => { @@ -81,6 +85,10 @@ export interface SessionTranscriptReader { sessionId: string, request: SessionTranscriptPageRequest, ): Promise; + readDurableRecords( + sessionId: string, + request: SessionTranscriptRecordScanRequest, + ): Promise; readDurableMessagesById( sessionId: string, request: SessionTranscriptMessageLookupRequest, diff --git a/packages/runtime-host/src/server/shared-session-transcript.ts b/packages/runtime-host/src/server/shared-session-transcript.ts new file mode 100644 index 0000000000..5c59dd8b0e --- /dev/null +++ b/packages/runtime-host/src/server/shared-session-transcript.ts @@ -0,0 +1,182 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { AttachmentRef, MessageContent } from '@maka/core/events'; +import { projectToolActivityArgs } from '@maka/core/tool-activity-args'; +import { + isUserVisibleSessionSystemNote, + type AssistantThinking, + type StoredMessage, + userFacingText, +} from '@maka/core/session'; + +/** Removes model-only composition while preserving what the user actually sent. */ +export function projectSharedSessionMessageContent( + content: MessageContent, + sessionId: string, +): MessageContent { + const attachments = content.attachments?.filter((attachment) => + isSharedSessionAttachment(attachment, sessionId), + ); + return { + text: userFacingText(content), + ...(attachments === undefined ? {} : { attachments: structuredClone(attachments) }), + ...(content.quotes === undefined ? {} : { quotes: structuredClone(content.quotes) }), + ...(content.inlineReferences === undefined + ? {} + : { inlineReferences: structuredClone(content.inlineReferences) }), + }; +} + +/** Projects the canonical transcript fields that the conversation UI can present. */ +export function projectSharedSessionTranscriptMessage( + message: StoredMessage, + sessionId: string, +): StoredMessage | null { + switch (message.type) { + case 'user': { + return { + ...projectSharedSessionMessageContent(message, sessionId), + type: message.type, + id: message.id, + turnId: message.turnId, + ts: message.ts, + ...(message.steeringEventId === undefined + ? {} + : { steeringEventId: message.steeringEventId }), + ...(message.origin === undefined ? {} : { origin: message.origin }), + }; + } + case 'assistant': + return { + type: message.type, + id: message.id, + turnId: message.turnId, + ts: message.ts, + text: message.text, + modelId: message.modelId, + ...(message.thinking === undefined ? {} : { thinking: projectThinking(message.thinking) }), + ...(message.contentOrder === undefined ? {} : { contentOrder: message.contentOrder }), + }; + case 'tool_call': + return { + type: message.type, + id: message.id, + turnId: message.turnId, + ts: message.ts, + toolName: message.toolName, + args: projectToolActivityArgs(message.toolName, message.args), + ...(message.activityKind === undefined ? {} : { activityKind: message.activityKind }), + ...(message.displayName === undefined ? {} : { displayName: message.displayName }), + ...(message.intent === undefined ? {} : { intent: message.intent }), + ...(message.stepId === undefined ? {} : { stepId: message.stepId }), + ...(message.origin === undefined ? {} : { origin: message.origin }), + ...(message.modelVisibility === undefined + ? {} + : { modelVisibility: message.modelVisibility }), + ...(message.parentToolCallId === undefined + ? {} + : { parentToolCallId: message.parentToolCallId }), + ...(message.parentOperationId === undefined + ? {} + : { parentOperationId: message.parentOperationId }), + }; + case 'tool_result': + return { + type: message.type, + id: message.id, + turnId: message.turnId, + ts: message.ts, + toolUseId: message.toolUseId, + isError: message.isError, + content: message.content, + ...(message.durationMs === undefined ? {} : { durationMs: message.durationMs }), + ...(message.origin === undefined ? {} : { origin: message.origin }), + ...(message.modelVisibility === undefined + ? {} + : { modelVisibility: message.modelVisibility }), + ...(message.parentToolCallId === undefined + ? {} + : { parentToolCallId: message.parentToolCallId }), + ...(message.parentOperationId === undefined + ? {} + : { parentOperationId: message.parentOperationId }), + }; + case 'turn_state': + return { + type: message.type, + id: message.id, + turnId: message.turnId, + ts: message.ts, + status: message.status, + ...(message.parentTurnId === undefined ? {} : { parentTurnId: message.parentTurnId }), + ...(message.retriedFromTurnId === undefined + ? {} + : { retriedFromTurnId: message.retriedFromTurnId }), + ...(message.regeneratedFromTurnId === undefined + ? {} + : { regeneratedFromTurnId: message.regeneratedFromTurnId }), + ...(message.branchOfTurnId === undefined ? {} : { branchOfTurnId: message.branchOfTurnId }), + ...(message.abortedAt === undefined ? {} : { abortedAt: message.abortedAt }), + ...(message.abortSource === undefined ? {} : { abortSource: message.abortSource }), + ...(message.errorClass === undefined ? {} : { errorClass: message.errorClass }), + partialOutputRetained: message.partialOutputRetained, + }; + case 'token_usage': + return { + type: message.type, + id: message.id, + turnId: message.turnId, + ts: message.ts, + input: message.input, + output: message.output, + ...(message.cacheMissInput === undefined ? {} : { cacheMissInput: message.cacheMissInput }), + ...(message.cacheRead === undefined ? {} : { cacheRead: message.cacheRead }), + ...(message.cacheCreation === undefined ? {} : { cacheCreation: message.cacheCreation }), + ...(message.reasoning === undefined ? {} : { reasoning: message.reasoning }), + ...(message.costUsd === undefined ? {} : { costUsd: message.costUsd }), + }; + case 'system_note': + return isUserVisibleSessionSystemNote(message.kind) + ? { + type: message.type, + id: message.id, + ...(message.turnId === undefined ? {} : { turnId: message.turnId }), + ts: message.ts, + kind: message.kind, + } + : null; + case 'permission_decision': + case 'workhub_coordination': + return null; + } +} + +function isSharedSessionAttachment(attachment: AttachmentRef, sessionId: string): boolean { + return attachment.ref.kind === 'session_file' && attachment.ref.sessionId === sessionId; +} + +function projectThinking(thinking: AssistantThinking): AssistantThinking { + return { + text: thinking.text, + ...(thinking.parts === undefined + ? {} + : { parts: thinking.parts.map((part) => ({ text: part.text })) }), + }; +} diff --git a/packages/storage/src/__tests__/session-store.test.ts b/packages/storage/src/__tests__/session-store.test.ts index 39499758ba..c59ea4a8f5 100644 --- a/packages/storage/src/__tests__/session-store.test.ts +++ b/packages/storage/src/__tests__/session-store.test.ts @@ -661,6 +661,15 @@ describe('SQLite SessionStore', () => { ); assert.deepEqual(tail.next, { position: 1, byteOffset: null }); + const decodedTail = await store.readTranscriptRecordsSnapshot(session.id, { + direction: 'older', + maxStoredBytes: 1, + maxMessages: 2, + }); + assert.equal(decodedTail.throughSequence, 3); + assert.deepEqual(decodedTail.records, [{ sequence: 3, message: messages[3] }]); + assert.equal(decodedTail.nextPosition, 2); + await store.appendMessage(session.id, { type: 'user', id: 'message-4', diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 047e5ca4eb..f50ffee65e 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -126,6 +126,8 @@ export type { SessionHeaderSnapshot, SessionTranscriptMessageLookupRequest, SessionTranscriptPageRequest, + SessionTranscriptRecordScanPage, + SessionTranscriptRecordScanRequest, SessionTranscriptStoragePage, SessionTranscriptStorageFragment, } from './session-store.js'; @@ -390,6 +392,8 @@ async function createExecutionStoresForWrite run(() => sessionStore.readMessagesSnapshot(sessionId)), readTranscriptPageSnapshot: (sessionId, request) => run(() => sessionStore.readTranscriptPageSnapshot(sessionId, request)), + readTranscriptRecordsSnapshot: (sessionId, request) => + run(() => sessionStore.readTranscriptRecordsSnapshot(sessionId, request)), readTranscriptMessagesSnapshot: (sessionId, request) => run(() => sessionStore.readTranscriptMessagesSnapshot(sessionId, request)), readTranscriptHighWaterSnapshot: (sessionId) => diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 5605ed6ccc..55eed4d1d4 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -255,6 +255,20 @@ export interface SessionTranscriptStoragePage { } | null; } +export interface SessionTranscriptRecordScanRequest { + readonly direction: 'older' | 'newer'; + readonly throughSequence?: number | null; + readonly position?: number; + readonly maxStoredBytes: number; + readonly maxMessages: number; +} + +export interface SessionTranscriptRecordScanPage { + readonly throughSequence: number | null; + readonly records: readonly { readonly sequence: number; readonly message: StoredMessage }[]; + readonly nextPosition: number | null; +} + export interface SessionTurnContribution { readonly turnId: string; readonly firstSequence: number; @@ -331,6 +345,11 @@ export interface SessionStore { } export interface SessionAuthorityStore extends SessionStore, MessageAdmissionStore { + /** Decode a bounded ledger range for an authority-owned wire projection. */ + readTranscriptRecordsSnapshot( + sessionId: string, + request: SessionTranscriptRecordScanRequest, + ): Promise; /** Read a bounded set of durable messages at an inclusive transcript watermark. */ readTranscriptMessagesSnapshot( sessionId: string, @@ -891,6 +910,14 @@ class SqliteSessionStore implements SessionAuthorityStore { return this.metadata.readTranscriptMessages(sessionId, request); } + async readTranscriptRecordsSnapshot( + sessionId: string, + request: SessionTranscriptRecordScanRequest, + ): Promise { + await this.ensureReady(); + return this.metadata.readTranscriptRecords(sessionId, request); + } + async readTranscriptHighWaterSnapshot(sessionId: string): Promise { await this.ensureReady(); return this.metadata.readTranscriptHighWater(sessionId); diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index 07d238eb16..bd707e0a7c 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -135,6 +135,8 @@ import { type ExternalSessionImportLookupResult, type SessionTranscriptMessageLookupRequest, type SessionTranscriptPageRequest, + type SessionTranscriptRecordScanPage, + type SessionTranscriptRecordScanRequest, type SessionTranscriptStoragePage, type SessionTurnContribution, type SessionTurnContributionPage, @@ -2540,6 +2542,86 @@ export class SqliteSessionMetadataStore { }); } + async readTranscriptRecords( + sessionId: string, + request: SessionTranscriptRecordScanRequest, + ): Promise { + this.assertOpen(); + assertSafeSessionId(sessionId); + assertTranscriptRecordScanRequest(request); + return this.readTransaction(() => { + if (!this.readRecordSync(sessionId)) throw new SessionNotFoundError(sessionId); + const highWaterRow = this.db + .prepare('SELECT MAX(sequence) AS high_water FROM session_messages WHERE session_id = ?') + .get(sessionId) as { high_water?: unknown }; + const actualHighWater = nullableStoredMessageSequence(highWaterRow.high_water, sessionId); + const throughSequence = + request.throughSequence === undefined ? actualHighWater : request.throughSequence; + if (throughSequence === null) { + return { throughSequence: null, records: [], nextPosition: null }; + } + if (actualHighWater === null || throughSequence > actualHighWater) { + throw new Error(`Session transcript watermark is ahead of durable storage: ${sessionId}`); + } + const position = request.position ?? (request.direction === 'older' ? throughSequence : 0); + const comparison = request.direction === 'older' ? '<=' : '>='; + const order = request.direction === 'older' ? 'DESC' : 'ASC'; + const rows = this.db + .prepare( + ` + SELECT message.sequence, + coalesce(payload.record_bytes, length(CAST(message.record_json AS BLOB))) AS stored_bytes + FROM session_messages AS message + LEFT JOIN session_message_payloads AS payload + ON payload.session_id = message.session_id AND payload.sequence = message.sequence + WHERE message.session_id = ? + AND message.sequence <= ? + AND message.sequence ${comparison} ? + ORDER BY message.sequence ${order} + LIMIT ? + `, + ) + .all(sessionId, throughSequence, position, request.maxMessages + 1) as Array<{ + sequence?: unknown; + stored_bytes?: unknown; + }>; + const selected: number[] = []; + let storedBytes = 0; + for (const row of rows) { + if (selected.length >= request.maxMessages) break; + const sequence = requireStoredMessageSequence(row.sequence, sessionId); + const bytes = requireTranscriptRecordByteLength(row.stored_bytes, sessionId, sequence); + if (selected.length > 0 && storedBytes + bytes > request.maxStoredBytes) break; + selected.push(sequence); + storedBytes += bytes; + } + const decoded = new Map(); + for (const row of readStoredMessageRows(this.db, sessionId, selected)) { + try { + decoded.set(row.sequence, decodeStoredMessage(JSON.parse(row.recordJson) as unknown)); + } catch (error) { + throw new StoredSessionMessageIncompatibleError(sessionId, row.sequence, { + cause: error, + }); + } + } + const records = selected.map((sequence) => { + const message = decoded.get(sequence); + if (!message) throw new StoredSessionMessageIncompatibleError(sessionId, sequence); + return { sequence, message }; + }); + const last = selected.at(-1); + return { + throughSequence, + records, + nextPosition: + last !== undefined && rows.length > selected.length + ? last + (request.direction === 'older' ? -1 : 1) + : null, + }; + }); + } + async readTranscriptHighWater(sessionId: string): Promise { this.assertOpen(); assertSafeSessionId(sessionId); @@ -7036,3 +7118,36 @@ function assertTranscriptPageRequest(request: SessionTranscriptPageRequest): voi throw new Error('Session transcript page message limit must be between 1 and 256'); } } + +function assertTranscriptRecordScanRequest(request: SessionTranscriptRecordScanRequest): void { + if (request.direction !== 'older' && request.direction !== 'newer') { + throw new Error('Invalid Session transcript record direction'); + } + if ( + request.throughSequence !== undefined && + request.throughSequence !== null && + (!Number.isSafeInteger(request.throughSequence) || request.throughSequence < 0) + ) { + throw new Error('Invalid Session transcript watermark'); + } + if ( + request.position !== undefined && + (!Number.isSafeInteger(request.position) || request.position < 0) + ) { + throw new Error('Invalid Session transcript position'); + } + if ( + !Number.isSafeInteger(request.maxStoredBytes) || + request.maxStoredBytes < 1 || + request.maxStoredBytes > 16 * 1024 * 1024 + ) { + throw new Error('Invalid Session transcript record byte limit'); + } + if ( + !Number.isSafeInteger(request.maxMessages) || + request.maxMessages < 1 || + request.maxMessages > 256 + ) { + throw new Error('Invalid Session transcript record count limit'); + } +} diff --git a/packages/ui/src/materialize.ts b/packages/ui/src/materialize.ts index 916e3db1d4..13c15cff93 100644 --- a/packages/ui/src/materialize.ts +++ b/packages/ui/src/materialize.ts @@ -17,7 +17,7 @@ * under the License. */ -import { deriveTurnRecords } from '@maka/core/session'; +import { deriveTurnRecords, isUserVisibleSessionSystemNote } from '@maka/core/session'; import { isInFlightToolStatus, toolResultActivityStatus, @@ -138,16 +138,6 @@ export interface ToolActivityItem { shellRunSource?: "owned" | "unavailable"; } -// system_note kinds that we surface inline to the user. Everything else -// (session_resume, connection_locked, mode_change-as-internal-audit, …) -// stays in the JSONL audit trail but is hidden from the chat surface so -// the conversation reads like a conversation, not a debug log. -const VISIBLE_SYSTEM_NOTES = new Set([ - "context_compacted", - "context_compaction_failed_open", - "step_limit", -]); - function systemNoteLabel(kind: string, locale: UiLocale): string { const copy = getConversationCopy(locale).messages.systemNotes; if (kind === "context_compacted") return copy.contextCompacted; @@ -189,7 +179,7 @@ export function materializeChat( }); if ( message.type === "system_note" && - VISIBLE_SYSTEM_NOTES.has(message.kind) + isUserVisibleSessionSystemNote(message.kind) ) { items.push({ id: message.id, @@ -774,7 +764,7 @@ export function materializeTurns( } } else if ( message.type === "system_note" && - VISIBLE_SYSTEM_NOTES.has(message.kind) + isUserVisibleSessionSystemNote(message.kind) ) { turn.notes.push({ id: message.id,