From dd6e899b0bae0ea06e90bc5c058c0a1598ccbe5c Mon Sep 17 00:00:00 2001 From: Wang Date: Sun, 30 Aug 2026 00:16:32 +0800 Subject: [PATCH] feat(runtime-host): admit approved guest Turn requests Persist immutable Guest Turn requests and Owner decisions, then recover approved exact intents through the canonical Turn admission path. Generated-by: OpenAI Codex --- apps/desktop/src/main/runtime-host-client.ts | 22 + .../runtime-host-operator-command.test.ts | 2 + .../__tests__/root-turn-coordinator.test.ts | 42 ++ .../session-collaboration-authority.test.ts | 532 +++++++++++++++++- packages/runtime-host/src/protocol/index.ts | 4 +- .../runtime-host/src/protocol/operations.ts | 2 + .../src/protocol/session-collaboration.ts | 274 +++++++++ packages/runtime-host/src/protocol/turn.ts | 2 +- .../src/server/access-authority.ts | 497 +++++++++++++++- .../src/server/access-credential-store.ts | 55 +- .../src/server/connection-authority.ts | 6 +- .../src/server/execution-composition.ts | 23 +- .../runtime-host/src/server/host-kernel.ts | 41 +- .../src/server/operation-dispatcher.ts | 30 + .../src/server/root-admission-owner.ts | 4 + .../src/server/root-turn-coordinator.ts | 23 +- ...session-turn-access-request-coordinator.ts | 143 +++++ packages/storage/src/agent-run-store.ts | 62 +- packages/storage/src/execution-stores.ts | 2 + 19 files changed, 1724 insertions(+), 42 deletions(-) create mode 100644 packages/runtime-host/src/server/session-turn-access-request-coordinator.ts diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index 74ef6579d8..235e4fe2bc 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -108,7 +108,11 @@ import { type CollaborationAccessQueryResult, type CollaborationInvitationPrepareResult, type CollaborationPrincipalRevokeResult, + type CollaborationTurnRequestDecideResult, + type CollaborationTurnRequestQueryResult, type SessionCollaborationGrantKind, + type SessionTurnAccessRequest, + type SessionTurnRequestIntent, type SessionConfigurationPatch, type SessionAssistantStreamIdentity, type SessionContinuitySnapshot, @@ -296,6 +300,24 @@ export class DesktopRuntimeHostClient { return this.request('collaboration.principal.revoke', { principalId }); } + createCollaborationTurnRequest(intent: SessionTurnRequestIntent): Promise { + return this.request('collaboration.turn-request.create', { intent }); + } + + queryCollaborationTurnRequests(sessionId: string): Promise { + return this.request('collaboration.turn-request.query', { sessionId }); + } + + decideCollaborationTurnRequest( + requestId: string, + decision: 'approve' | 'reject', + ): Promise { + return this.request('collaboration.turn-request.decide', { + requestId, + decision, + }); + } + subscribeConfigurationChanges(listener: (revision: number) => void): () => void { this.#assertOpen(); return this.connection.subscribeConfigurationChanges(listener); diff --git a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts index b99cedecb4..9179097b5b 100644 --- a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts +++ b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts @@ -378,6 +378,8 @@ describe('Runtime Host operator commands', () => { 'access.credential.rotation.prepare', 'access.credential.rotation.revoke', 'access.principal.revoke', + 'collaboration.turn-request.acknowledge', + 'collaboration.turn-request.create', 'host.upgrade.prepare', 'hosted.execution.cancel', 'hosted.execution.start', 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 791b9f8a6e..3abf3d6db7 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -826,6 +826,48 @@ test('turn.start durably applies one exact per-Turn orchestration override', asy } }); +test('turn.start durably binds a Guest request approval to the admitted Turn', async () => { + const fixture = await createFailureFixture({ + registerBackend: (backends) => + backends.register('ai-sdk', (context) => new FakeBackend(context)), + }); + const input = { + sessionId: fixture.sessionId, + turnId: 'turn-collaboration-request', + content: { text: 'Run the exact approved request.' }, + }; + const authorization = { + kind: 'session_turn_access_request' as const, + requestId: 'request-1', + principalId: 'session_guest:guest-1', + grantId: 'grant-1', + approvedAt: 1_788_000_000_000, + approvedBy: 'local_owner', + }; + try { + const started = await fixture.interactiveTurns.handlers['turn.start'](input, { + ...operationContext(fixture.hostEpoch, fixture.acquireResidency), + principal: authorization.principalId, + turnAdmissionAuthorization: authorization, + }); + assertStartedTurn(started); + assert.deepEqual( + (await fixture.stores.agentRunStore.readRootTurnAdmission(fixture.sessionId, input.turnId)) + ?.authorization, + authorization, + ); + + const conflictingRetry = await fixture.interactiveTurns.handlers['turn.start']( + input, + operationContext(fixture.hostEpoch, fixture.acquireResidency), + ); + assert.equal(conflictingRetry.ok, false); + if (!conflictingRetry.ok) assert.equal(conflictingRetry.error.code, 'operation_conflict'); + } finally { + await fixture.dispose(); + } +}); + test('turn.start resolves explicit Skills once before durable admission and replays the result', async () => { let preparationCount = 0; let blocked = false; 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 9f3573e370..7faa139454 100644 --- a/packages/runtime-host/src/__tests__/session-collaboration-authority.test.ts +++ b/packages/runtime-host/src/__tests__/session-collaboration-authority.test.ts @@ -22,8 +22,31 @@ import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import test from 'node:test'; -import { decodeCollaborationInvitationCode } from '../protocol/index.js'; -import { openRuntimeHostAccessAuthority } from '../server/access-authority.js'; +import { + decodeCollaborationInvitationCode, + HOST_OPERATION_SPECS, + type RequestFrame, +} from '../protocol/index.js'; +import { + openRuntimeHostAccessAuthority, + queryCollaborationTurnRequests, + type RuntimeHostAccessAuthority, +} from '../server/access-authority.js'; +import { + RuntimeHostAccessCommitOutcomeUnknownError, + writeAccessCredentialFile, +} from '../server/access-credential-store.js'; +import { authorizeRuntimeHostOperation } from '../server/connection-authority.js'; +import { SessionTurnAccessRequestCoordinator } from '../server/session-turn-access-request-coordinator.js'; + +const LOCAL_OWNER = { + principalId: 'local_owner', + principalKind: 'local_owner', +} as const; + +function sessionGuest(principalId: string) { + return { principalId, principalKind: 'session_guest' as const }; +} test('Session Guest invitation, grants, and revocation form one durable authority lifecycle', async () => { const directory = await mkdtemp(join(tmpdir(), 'maka-session-collaboration-')); @@ -42,9 +65,13 @@ test('Session Guest invitation, grants, and revocation form one durable authorit const credentialId = authority.authenticate(invitation.credential)?.credentialId; assert.ok(credentialId); await authority.finalize(credentialId, 'guest-client'); - assert.deepEqual(authority.authenticate(invitation.credential)?.operationGrants, [ + const activeGuest = authority.authenticate(invitation.credential); + assert.deepEqual(activeGuest?.operationGrants, [ 'host.status', 'artifact.query', + 'collaboration.turn-request.create', + 'collaboration.turn-request.acknowledge', + 'collaboration.turn-request.query', 'runtime.resource.query', 'session.shared.query', 'subscription.open', @@ -52,6 +79,36 @@ test('Session Guest invitation, grants, and revocation form one durable authorit 'session.transcript.page', 'session.transcript.overlay.release', ]); + assert.ok(activeGuest); + const unidentifiedQuery = queryCollaborationTurnRequests( + authority, + { principalId: activeGuest.principalId, principalKind: undefined }, + { sessionId: 'session-1' }, + ); + assert.equal(unidentifiedQuery.ok, false); + if (!unidentifiedQuery.ok) assert.equal(unidentifiedQuery.error.code, 'operation_unavailable'); + assert.equal( + authorizeRuntimeHostOperation(activeGuest, { + requestId: 'request-1', + operation: 'collaboration.turn-request.create', + input: { + intent: { + sessionId: 'session-1', + turnId: 'turn-1', + content: { text: 'Continue' }, + }, + }, + } as RequestFrame), + true, + ); + assert.equal( + authorizeRuntimeHostOperation(activeGuest, { + requestId: 'request-2', + operation: 'turn.start', + input: {}, + } as RequestFrame), + false, + ); const observation = prepared.grants.find((grant) => grant.kind === 'session_observation')!; assert.equal( @@ -60,7 +117,11 @@ test('Session Guest invitation, grants, and revocation form one durable authorit observation.grantId, ); assert.equal( - (await authority.revokeCollaborationGrant({ grantId: observation.grantId })).revoked, + ( + await authority.revokeCollaborationGrant({ + grantId: observation.grantId, + }) + ).revoked, true, ); assert.equal( @@ -78,3 +139,466 @@ test('Session Guest invitation, grants, and revocation form one durable authorit await rm(directory, { recursive: true, force: true }); } }); + +test('Turn requests reject execution input that the Owner cannot review', () => { + assert.throws( + () => + HOST_OPERATION_SPECS['collaboration.turn-request.create'].decodeInput({ + intent: { + sessionId: 'session-1', + turnId: 'turn-1', + content: { text: 'Looks harmless' }, + skillSelection: { mode: 'all' }, + }, + }), + /Unknown Session Turn request intent field/u, + ); +}); + +test('an approved exact Turn request survives restart and is admitted once', async () => { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-turn-request-')); + let authority = await openRuntimeHostAccessAuthority(directory); + try { + const principalId = await activateTurnGuest(authority); + const intent = { + sessionId: 'session-1', + turnId: 'turn-1', + content: { text: 'Run the approved task' }, + }; + const request = await authority.createTurnAccessRequest(principalId, { + intent, + }); + const decision = await authority.decideTurnAccessRequest('local_owner', { + requestId: request.requestId, + decision: 'approve', + }); + assert.equal(decision.kind, 'decided'); + assert.equal(decision.kind, 'decided'); + if (decision.kind !== 'decided') assert.fail('Expected an approved request'); + assert.equal(decision.request.state.kind, 'approved'); + if (decision.request.state.kind !== 'approved') assert.fail('Expected an approved request'); + await authority.close(); + + authority = await openRuntimeHostAccessAuthority(directory); + const admitted: unknown[] = []; + const authorizations: unknown[] = []; + const coordinator = new SessionTurnAccessRequestCoordinator({ + authority, + hostEpoch: 'epoch-1', + acquireResidency: () => ({ release: () => undefined }), + requestDrain: () => undefined, + whenIdle: () => undefined, + startTurn: async (input, context) => { + admitted.push(input); + authorizations.push(context.turnAdmissionAuthorization); + return { + ok: true, + result: { + kind: 'blocked', + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }, + }; + }, + }); + await coordinator.recover(); + await coordinator.close(); + + assert.deepEqual(admitted, [intent]); + assert.deepEqual(authorizations, [ + { + kind: 'session_turn_access_request', + requestId: request.requestId, + principalId, + grantId: request.grantId, + approvedAt: Date.parse(decision.request.state.decidedAt), + approvedBy: 'local_owner', + }, + ]); + const completed = authority.queryTurnAccessRequests(LOCAL_OWNER, { + sessionId: 'session-1', + }).requests[0]; + assert.equal( + completed?.state.kind === 'approved' ? completed.state.admission : undefined, + 'blocked', + ); + } finally { + await authority.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('Turn access request creation is idempotent for one exact Guest intent', async () => { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-turn-request-idempotent-')); + const authority = await openRuntimeHostAccessAuthority(directory); + try { + const principalId = await activateTurnGuest(authority); + const intent = { + sessionId: 'session-1', + turnId: 'stable-turn', + content: { text: 'Run this once' }, + }; + const created = await authority.createTurnAccessRequest(principalId, { intent }); + assert.deepEqual(await authority.createTurnAccessRequest(principalId, { intent }), created); + assert.equal( + authority.queryTurnAccessRequests(sessionGuest(principalId), { sessionId: 'session-1' }) + .requests.length, + 1, + ); + await assert.rejects( + authority.createTurnAccessRequest(principalId, { + intent: { ...intent, content: { text: 'Different work' } }, + }), + /different content/u, + ); + } finally { + await authority.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('Turn access request results remain until acknowledged without consuming unbounded state', async () => { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-turn-request-bound-')); + const authority = await openRuntimeHostAccessAuthority(directory); + try { + const principalId = await activateTurnGuest(authority); + const rejectedIds: string[] = []; + for (let index = 0; index < 4; index += 1) { + const request = await authority.createTurnAccessRequest(principalId, { + intent: { + sessionId: 'session-1', + turnId: `rejected-${index}`, + content: { text: `Rejected ${index}` }, + }, + }); + await authority.decideTurnAccessRequest('local_owner', { + requestId: request.requestId, + decision: 'reject', + }); + rejectedIds.push(request.requestId); + } + await assert.rejects( + authority.createTurnAccessRequest(principalId, { + intent: { + sessionId: 'session-1', + turnId: 'unacknowledged-overflow', + content: { text: 'Review the earlier results first' }, + }, + }), + /Review earlier Turn access request results/u, + ); + assert.deepEqual( + authority + .queryTurnAccessRequests(sessionGuest(principalId), { sessionId: 'session-1' }) + .requests.map((request) => request.requestId), + rejectedIds, + ); + assert.deepEqual( + await authority.acknowledgeTurnAccessRequest(principalId, { requestId: rejectedIds[0]! }), + { acknowledged: true }, + ); + assert.equal( + authority.queryTurnAccessRequests(sessionGuest(principalId), { sessionId: 'session-1' }) + .requests.length, + 3, + ); + const replacement = await authority.createTurnAccessRequest(principalId, { + intent: { + sessionId: 'session-1', + turnId: 'pending-after-acknowledgement', + content: { text: 'Acknowledged history no longer blocks new work' }, + }, + }); + assert.equal(replacement.state.kind, 'pending'); + + const otherPrincipal = await activateTurnGuest(authority); + for (let index = 0; index < 3; index += 1) { + await authority.createTurnAccessRequest(otherPrincipal, { + intent: { + sessionId: 'session-1', + turnId: `pending-${index}`, + content: { text: `Pending ${index}` }, + }, + }); + } + await assert.rejects( + authority.createTurnAccessRequest(otherPrincipal, { + intent: { + sessionId: 'session-1', + turnId: 'pending-overflow', + content: { text: 'Pending overflow' }, + }, + }), + /Too many Turn access requests/u, + ); + assert.equal( + authority + .queryTurnAccessRequests(sessionGuest(principalId), { sessionId: 'session-1' }) + .requests.every((request) => request.principalId === principalId), + true, + ); + assert.equal( + authority + .queryTurnAccessRequests(LOCAL_OWNER, { sessionId: 'session-1' }) + .requests.some((request) => request.principalId === otherPrincipal), + true, + ); + await authority.revokeCollaborationPrincipal(otherPrincipal); + assert.equal( + authority + .queryTurnAccessRequests(LOCAL_OWNER, { sessionId: 'session-1' }) + .requests.some((request) => request.principalId === otherPrincipal), + false, + ); + } finally { + await authority.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('an uncertain access decision is recovered without premature Turn admission', async () => { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-turn-request-uncertain-')); + let failAfterWrite = false; + let authority = await openRuntimeHostAccessAuthority(directory, { + writeFile: async (path, file) => { + await writeAccessCredentialFile(path, file); + if (failAfterWrite) throw new RuntimeHostAccessCommitOutcomeUnknownError(new Error('fsync')); + }, + }); + try { + const principalId = await activateTurnGuest(authority); + const request = await authority.createTurnAccessRequest(principalId, { + intent: { + sessionId: 'session-1', + turnId: 'turn-1', + content: { text: 'Run once durable' }, + }, + }); + const published: string[] = []; + authority.subscribeApprovedTurnAccessRequests((changed) => published.push(changed.requestId)); + + failAfterWrite = true; + await assert.rejects( + authority.decideTurnAccessRequest('local_owner', { + requestId: request.requestId, + decision: 'approve', + }), + RuntimeHostAccessCommitOutcomeUnknownError, + ); + assert.deepEqual(published, []); + await authority.close(); + + authority = await openRuntimeHostAccessAuthority(directory); + assert.equal(authority.approvedTurnAccessRequests().length, 1); + } finally { + await authority.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('recovery stays ready while an approved Turn request waits for the Session', async () => { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-turn-request-busy-')); + const authority = await openRuntimeHostAccessAuthority(directory); + try { + const principalId = await activateTurnGuest(authority); + const request = await authority.createTurnAccessRequest(principalId, { + intent: { + sessionId: 'session-1', + turnId: 'turn-after-active-root', + content: { text: 'Start when the current Turn finishes' }, + }, + }); + await authority.decideTurnAccessRequest('local_owner', { + requestId: request.requestId, + decision: 'approve', + }); + + let attempts = 0; + let drained = false; + let releaseIdle!: () => void; + const idle = new Promise((resolve) => { + releaseIdle = resolve; + }); + let observeSecondAttempt!: () => void; + const secondAttempt = new Promise((resolve) => { + observeSecondAttempt = resolve; + }); + const coordinator = new SessionTurnAccessRequestCoordinator({ + authority, + hostEpoch: 'epoch-1', + acquireResidency: () => ({ release: () => undefined }), + requestDrain: () => { + drained = true; + }, + whenIdle: () => idle, + startTurn: async () => { + attempts += 1; + if (attempts === 2) observeSecondAttempt(); + return attempts === 1 + ? { + ok: false, + error: { + code: 'session_busy', + message: 'Session already has an active root Turn', + retryable: true, + }, + } + : { + ok: true, + result: { + kind: 'blocked', + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }, + }; + }, + }); + coordinator.recover(); + assert.equal(attempts, 1); + releaseIdle(); + await secondAttempt; + await coordinator.close(); + + assert.equal(attempts, 2); + assert.equal(drained, false); + const completed = authority.queryTurnAccessRequests(LOCAL_OWNER, { + sessionId: 'session-1', + }).requests[0]; + assert.equal( + completed?.state.kind === 'approved' ? completed.state.admission : undefined, + 'blocked', + ); + } finally { + await authority.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('an idle-wait failure drains without losing the approved Turn request', async () => { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-turn-request-idle-failure-')); + const authority = await openRuntimeHostAccessAuthority(directory); + try { + const principalId = await activateTurnGuest(authority); + const request = await authority.createTurnAccessRequest(principalId, { + intent: { + sessionId: 'session-1', + turnId: 'turn-after-failed-root', + content: { text: 'Retry after Host recovery' }, + }, + }); + await authority.decideTurnAccessRequest('local_owner', { + requestId: request.requestId, + decision: 'approve', + }); + + let drained = false; + const coordinator = new SessionTurnAccessRequestCoordinator({ + authority, + hostEpoch: 'epoch-1', + acquireResidency: () => ({ release: () => undefined }), + requestDrain: () => { + drained = true; + }, + whenIdle: () => Promise.reject(new Error('active Turn authority failed')), + startTurn: async () => ({ + ok: false, + error: { + code: 'session_busy', + message: 'Session already has an active root Turn', + retryable: true, + }, + }), + }); + coordinator.recover(); + await coordinator.close(); + + assert.equal(drained, true); + const pending = authority.queryTurnAccessRequests(LOCAL_OWNER, { + sessionId: 'session-1', + }).requests[0]; + assert.equal( + pending?.state.kind === 'approved' ? pending.state.admission : undefined, + 'pending', + ); + } finally { + await authority.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('drain does not terminalize an in-flight admission failure', async () => { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-turn-request-drain-')); + const authority = await openRuntimeHostAccessAuthority(directory); + try { + const principalId = await activateTurnGuest(authority); + let finishAdmission!: (result: { + ok: false; + error: { code: 'host_draining'; message: string; retryable: boolean }; + }) => void; + const admission = new Promise<{ + ok: false; + error: { code: 'host_draining'; message: string; retryable: boolean }; + }>((resolve) => { + finishAdmission = resolve; + }); + let started!: () => void; + const startObserved = new Promise((resolve) => { + started = resolve; + }); + const coordinator = new SessionTurnAccessRequestCoordinator({ + authority, + hostEpoch: 'epoch-1', + acquireResidency: () => ({ release: () => undefined }), + requestDrain: () => undefined, + whenIdle: () => undefined, + startTurn: async () => { + started(); + return admission; + }, + }); + const request = await authority.createTurnAccessRequest(principalId, { + intent: { + sessionId: 'session-1', + turnId: 'turn-during-drain', + content: { text: 'Retry after restart' }, + }, + }); + await authority.decideTurnAccessRequest('local_owner', { + requestId: request.requestId, + decision: 'approve', + }); + await startObserved; + coordinator.beginDrain(); + finishAdmission({ + ok: false, + error: { + code: 'host_draining', + message: 'Runtime Host is draining', + retryable: true, + }, + }); + await coordinator.close(); + + const retained = authority.queryTurnAccessRequests(LOCAL_OWNER, { + sessionId: 'session-1', + }).requests[0]; + assert.equal(retained?.state.kind, 'approved'); + assert.equal( + retained?.state.kind === 'approved' ? retained.state.admission : undefined, + 'pending', + ); + } finally { + await authority.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +async function activateTurnGuest(authority: RuntimeHostAccessAuthority): Promise { + const prepared = await authority.prepareCollaborationInvitation('root-1', { + sessionId: 'session-1', + grantKinds: ['session_turn_request'], + }); + const invitation = decodeCollaborationInvitationCode(prepared.invitationCode); + const credentialId = authority.authenticate(invitation.credential)?.credentialId; + assert.ok(credentialId); + await authority.finalize(credentialId, 'guest-client'); + return prepared.principalId; +} diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 9b16632344..651871f718 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 = 70 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 71 as const; +// 71: Session Guests can submit durable exact Turn access requests and Owners +// can decide them. Older peers do not understand this execution-authority flow. // 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 diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index ed7af8d936..8476927962 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -241,6 +241,8 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'collaboration.grant.revoke', 'collaboration.invitation.prepare', 'collaboration.principal.revoke', + 'collaboration.turn-request.decide', + 'collaboration.turn-request.query', 'connection.catalog.create', 'connection.catalog.query', 'connection.catalog.remove', diff --git a/packages/runtime-host/src/protocol/session-collaboration.ts b/packages/runtime-host/src/protocol/session-collaboration.ts index 25cdb43ea1..df0c4a3779 100644 --- a/packages/runtime-host/src/protocol/session-collaboration.ts +++ b/packages/runtime-host/src/protocol/session-collaboration.ts @@ -21,11 +21,13 @@ import { requireEntityId, requireExactRecord, requireId, + requireRecord, requireShapedRecord, requireUtf8String, } from './codec.js'; import { invalidProtocolFrame } from './errors.js'; import { defineOperation } from './operation-spec.js'; +import { decodeTurnStartInput } from './turn.js'; export const COLLABORATION_INVITATION_SCHEMA_VERSION = 1 as const; export const COLLABORATION_INVITATION_CODE_MAX_BYTES = 16 * 1024; @@ -100,6 +102,66 @@ export interface CollaborationPrincipalRevokeResult { readonly revoked: boolean; } +export type SessionTurnAccessRequestState = + | { readonly kind: 'pending' } + | { + readonly kind: 'rejected'; + readonly decidedAt: string; + readonly decidedBy: string; + } + | { + readonly kind: 'approved'; + readonly decidedAt: string; + readonly decidedBy: string; + readonly admission: 'pending' | 'started' | 'blocked' | 'failed'; + }; + +export interface SessionTurnRequestIntent { + readonly sessionId: string; + readonly turnId: string; + readonly content: { readonly text: string }; +} + +export interface SessionTurnAccessRequest { + readonly requestId: string; + readonly principalId: string; + readonly grantId: string; + readonly intent: SessionTurnRequestIntent; + readonly createdAt: string; + readonly state: SessionTurnAccessRequestState; +} + +export interface CollaborationTurnRequestCreateInput { + readonly intent: SessionTurnRequestIntent; +} + +export interface CollaborationTurnRequestQueryInput { + readonly sessionId: string; +} + +export interface CollaborationTurnRequestQueryResult { + readonly requests: readonly SessionTurnAccessRequest[]; +} + +export interface CollaborationTurnRequestDecideInput { + readonly requestId: string; + readonly decision: 'approve' | 'reject'; +} + +export type CollaborationTurnRequestDecideResult = + | { readonly kind: 'not_found' } + | { + readonly kind: 'decided' | 'already_decided'; + readonly request: SessionTurnAccessRequest; + }; + +export interface CollaborationTurnRequestAcknowledgeInput { + readonly requestId: string; +} + +export interface CollaborationTurnRequestAcknowledgeResult { + readonly acknowledged: boolean; +} const COLLABORATION_ERRORS = [ 'host_not_ready', 'host_draining', @@ -155,6 +217,50 @@ export const SESSION_COLLABORATION_OPERATION_SPECS = { decodeInput: decodeCollaborationPrincipalRevokeInput, decodeOutput: decodeCollaborationPrincipalRevokeResult, }), + 'collaboration.turn-request.create': defineOperation< + CollaborationTurnRequestCreateInput, + SessionTurnAccessRequest, + (typeof COLLABORATION_ERRORS)[number] + >({ + mode: 'command', + availability: 'ready', + errors: COLLABORATION_ERRORS, + decodeInput: decodeCollaborationTurnRequestCreateInput, + decodeOutput: decodeSessionTurnAccessRequest, + }), + 'collaboration.turn-request.query': defineOperation< + CollaborationTurnRequestQueryInput, + CollaborationTurnRequestQueryResult, + (typeof COLLABORATION_ERRORS)[number] + >({ + mode: 'query', + availability: 'ready', + errors: COLLABORATION_ERRORS, + decodeInput: decodeCollaborationTurnRequestQueryInput, + decodeOutput: decodeCollaborationTurnRequestQueryResult, + }), + 'collaboration.turn-request.acknowledge': defineOperation< + CollaborationTurnRequestAcknowledgeInput, + CollaborationTurnRequestAcknowledgeResult, + (typeof COLLABORATION_ERRORS)[number] + >({ + mode: 'command', + availability: 'ready', + errors: COLLABORATION_ERRORS, + decodeInput: decodeCollaborationTurnRequestAcknowledgeInput, + decodeOutput: decodeCollaborationTurnRequestAcknowledgeResult, + }), + 'collaboration.turn-request.decide': defineOperation< + CollaborationTurnRequestDecideInput, + CollaborationTurnRequestDecideResult, + (typeof COLLABORATION_ERRORS)[number] + >({ + mode: 'command', + availability: 'ready', + errors: COLLABORATION_ERRORS, + decodeInput: decodeCollaborationTurnRequestDecideInput, + decodeOutput: decodeCollaborationTurnRequestDecideResult, + }), } as const; export function encodeCollaborationInvitationCode(payload: CollaborationInvitationPayload): string { @@ -307,6 +413,174 @@ function decodeCollaborationPrincipalRevokeResult( return { revoked: record.revoked }; } +function decodeCollaborationTurnRequestCreateInput( + value: unknown, +): CollaborationTurnRequestCreateInput { + const record = requireExactRecord(value, 'collaboration Turn request input', ['intent']); + return { intent: decodeSessionTurnRequestIntent(record.intent) }; +} + +function decodeCollaborationTurnRequestQueryInput( + value: unknown, +): CollaborationTurnRequestQueryInput { + const record = requireExactRecord(value, 'collaboration Turn request query', ['sessionId']); + return { sessionId: requireEntityId(record.sessionId, 'sessionId') }; +} + +function decodeCollaborationTurnRequestQueryResult( + value: unknown, +): CollaborationTurnRequestQueryResult { + const record = requireExactRecord(value, 'collaboration Turn request query result', ['requests']); + if (!Array.isArray(record.requests)) { + throw invalidProtocolFrame('Invalid collaboration Turn request query result'); + } + return { requests: record.requests.map(decodeSessionTurnAccessRequest) }; +} + +function decodeCollaborationTurnRequestAcknowledgeInput( + value: unknown, +): CollaborationTurnRequestAcknowledgeInput { + const record = requireExactRecord(value, 'collaboration Turn request acknowledgement input', [ + 'requestId', + ]); + return { requestId: requireId(record.requestId, 'requestId') }; +} + +function decodeCollaborationTurnRequestAcknowledgeResult( + value: unknown, +): CollaborationTurnRequestAcknowledgeResult { + const record = requireExactRecord(value, 'collaboration Turn request acknowledgement result', [ + 'acknowledged', + ]); + if (typeof record.acknowledged !== 'boolean') { + throw invalidProtocolFrame('Invalid collaboration Turn request acknowledgement result'); + } + return { acknowledged: record.acknowledged }; +} + +function decodeCollaborationTurnRequestDecideInput( + value: unknown, +): CollaborationTurnRequestDecideInput { + const record = requireExactRecord(value, 'collaboration Turn request decision', [ + 'requestId', + 'decision', + ]); + if (record.decision !== 'approve' && record.decision !== 'reject') { + throw invalidProtocolFrame('Invalid collaboration Turn request decision'); + } + return { + requestId: requireId(record.requestId, 'requestId'), + decision: record.decision, + }; +} + +function decodeCollaborationTurnRequestDecideResult( + value: unknown, +): CollaborationTurnRequestDecideResult { + const candidate = requireRecord(value, 'collaboration Turn request decision result'); + const record = requireExactRecord( + value, + 'collaboration Turn request decision result', + candidate.kind === 'not_found' ? ['kind'] : ['kind', 'request'], + ); + if ( + record.kind !== 'decided' && + record.kind !== 'already_decided' && + record.kind !== 'not_found' + ) { + throw invalidProtocolFrame('Invalid collaboration Turn request decision result'); + } + return record.kind === 'not_found' + ? { kind: record.kind } + : { kind: record.kind, request: decodeSessionTurnAccessRequest(record.request) }; +} + +export function decodeSessionTurnAccessRequest(value: unknown): SessionTurnAccessRequest { + const record = requireExactRecord(value, 'Session Turn access request', [ + 'requestId', + 'principalId', + 'grantId', + 'intent', + 'createdAt', + 'state', + ]); + return { + requestId: requireId(record.requestId, 'requestId'), + principalId: decodePrincipalId(record.principalId), + grantId: requireId(record.grantId, 'grantId'), + intent: decodeSessionTurnRequestIntent(record.intent), + createdAt: decodeIsoTimestamp(record.createdAt, 'createdAt'), + state: decodeSessionTurnAccessRequestState(record.state), + }; +} + +function decodeSessionTurnRequestIntent(value: unknown): SessionTurnRequestIntent { + const record = requireExactRecord(value, 'Session Turn request intent', [ + 'sessionId', + 'turnId', + 'content', + ]); + const content = requireExactRecord(record.content, 'Session Turn request content', ['text']); + const decoded = decodeTurnStartInput({ + sessionId: record.sessionId, + turnId: record.turnId, + content: { text: content.text }, + }); + return { + sessionId: decoded.sessionId, + turnId: decoded.turnId, + content: { text: decoded.content.text }, + }; +} + +function decodeSessionTurnAccessRequestState(value: unknown): SessionTurnAccessRequestState { + const kind = requireShapedRecord( + value, + 'Session Turn access request state', + ['kind'], + ['decidedAt', 'decidedBy', 'admission'], + ).kind; + if (kind === 'pending') { + requireExactRecord(value, 'pending Session Turn access request', ['kind']); + return { kind }; + } + if (kind === 'rejected') { + const record = requireExactRecord(value, 'rejected Session Turn access request', [ + 'kind', + 'decidedAt', + 'decidedBy', + ]); + return { + kind, + decidedAt: decodeIsoTimestamp(record.decidedAt, 'decidedAt'), + decidedBy: decodePrincipalId(record.decidedBy), + }; + } + if (kind !== 'approved') { + throw invalidProtocolFrame('Invalid Session Turn access request state'); + } + const candidate = requireExactRecord(value, 'approved Session Turn access request', [ + 'kind', + 'decidedAt', + 'decidedBy', + 'admission', + ]); + if ( + candidate.admission !== 'pending' && + candidate.admission !== 'started' && + candidate.admission !== 'blocked' && + candidate.admission !== 'failed' + ) { + throw invalidProtocolFrame('Invalid Session Turn access request admission'); + } + return { + kind, + decidedAt: decodeIsoTimestamp(candidate.decidedAt, 'decidedAt'), + decidedBy: decodePrincipalId(candidate.decidedBy), + admission: candidate.admission, + }; +} + export function decodeSessionCollaborationGrant(value: unknown): SessionCollaborationGrant { const record = requireExactRecord(value, 'Session collaboration grant', [ 'kind', diff --git a/packages/runtime-host/src/protocol/turn.ts b/packages/runtime-host/src/protocol/turn.ts index e1a4a5b41e..def4e1a4bc 100644 --- a/packages/runtime-host/src/protocol/turn.ts +++ b/packages/runtime-host/src/protocol/turn.ts @@ -344,7 +344,7 @@ export const TURN_OPERATION_SPECS = { }), } as const; -function decodeTurnStartInput(value: unknown): TurnStartInput { +export function decodeTurnStartInput(value: unknown): TurnStartInput { const record = requireShapedRecord( value, 'turn.start input', diff --git a/packages/runtime-host/src/server/access-authority.ts b/packages/runtime-host/src/server/access-authority.ts index 3ccdfad22b..000a8868f3 100644 --- a/packages/runtime-host/src/server/access-authority.ts +++ b/packages/runtime-host/src/server/access-authority.ts @@ -43,9 +43,17 @@ import { type CollaborationInvitationPrepareInput, type CollaborationInvitationPrepareResult, type CollaborationPrincipalRevokeResult, + type CollaborationTurnRequestAcknowledgeInput, + type CollaborationTurnRequestAcknowledgeResult, + type CollaborationTurnRequestCreateInput, + type CollaborationTurnRequestDecideInput, + type CollaborationTurnRequestDecideResult, + type CollaborationTurnRequestQueryInput, + type CollaborationTurnRequestQueryResult, encodeCollaborationInvitationCode, type SessionCollaborationGrant, type SessionCollaborationGrantKind, + type SessionTurnAccessRequest, } from '../protocol/index.js'; import { createRuntimeHostConnectionAuthority, @@ -73,12 +81,22 @@ import { const ACCESS_CREDENTIAL_PREFIX = 'maka_rh_'; const PENDING_CREDENTIAL_LIFETIME_MS = 15 * 60_000; +const TURN_ACCESS_REQUEST_ACTIVE_MAX = 4; const CAPABILITY_PROVIDER_GRANTS = new Set([ 'host.status', 'client.capability.replace', 'client.capability.unregister', ]); +function createNextAccessCredentialFile( + current: AccessCredentialFile, + credentials: readonly StoredAccessCredential[], + sessionGrants: readonly SessionCollaborationGrant[], + turnAccessRequests: readonly SessionTurnAccessRequest[] = current.turnAccessRequests, +): AccessCredentialFile { + return createAccessCredentialFile(credentials, sessionGrants, turnAccessRequests); +} + export interface RuntimeHostAccessAuthority { authenticate(credential: string): RuntimeHostConnectionAuthority | undefined; issue(input: AccessCredentialIssueInput): Promise; @@ -102,6 +120,27 @@ export interface RuntimeHostAccessAuthority { input: CollaborationGrantRevokeInput, ): Promise; revokeCollaborationPrincipal(principalId: string): Promise; + createTurnAccessRequest( + principalId: string, + input: CollaborationTurnRequestCreateInput, + ): Promise; + queryTurnAccessRequests( + principal: Pick, + input: CollaborationTurnRequestQueryInput, + ): CollaborationTurnRequestQueryResult; + acknowledgeTurnAccessRequest( + principalId: string, + input: CollaborationTurnRequestAcknowledgeInput, + ): Promise; + decideTurnAccessRequest( + principalId: string, + input: CollaborationTurnRequestDecideInput, + ): Promise; + completeTurnAccessRequest( + requestId: string, + admission: 'started' | 'blocked' | 'failed', + ): Promise; + approvedTurnAccessRequests(): readonly SessionTurnAccessRequest[]; activeSessionGrant( principalId: string, sessionId: string, @@ -113,6 +152,9 @@ export interface RuntimeHostAccessAuthority { ): SessionCollaborationGrant | undefined; subscribeRevocations(listener: (credentialId: string) => void): () => void; subscribeGrantRevocations(listener: (grant: SessionCollaborationGrant) => void): () => void; + subscribeApprovedTurnAccessRequests( + listener: (request: SessionTurnAccessRequest) => void, + ): () => void; close(): Promise; } @@ -142,6 +184,9 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { #closed = false; readonly #revocationListeners = new Set<(credentialId: string) => void>(); readonly #grantRevocationListeners = new Set<(grant: SessionCollaborationGrant) => void>(); + readonly #approvedTurnAccessRequestListeners = new Set< + (request: SessionTurnAccessRequest) => void + >(); constructor( controlDirectory: string, @@ -233,7 +278,8 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { createdAt, }), ); - const nextFile = createAccessCredentialFile( + const nextFile = createNextAccessCredentialFile( + this.#file, [...this.#file.credentials, stored], [...this.#file.sessionGrants, ...grants], ); @@ -290,9 +336,16 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { const current = this.#file.sessionGrants.find((grant) => grant.grantId === input.grantId); if (!current) return { revoked: false }; await this.#commit( - createAccessCredentialFile( + createNextAccessCredentialFile( + this.#file, this.#file.credentials, this.#file.sessionGrants.filter((grant) => grant !== current), + current.kind === 'session_turn_request' + ? removePendingTurnAccessRequests( + this.#file.turnAccessRequests, + (request) => request.grantId === current.grantId, + ) + : this.#file.turnAccessRequests, ), [], [current], @@ -302,7 +355,204 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { } revokeCollaborationPrincipal(principalId: string): Promise { - return this.revokePrincipal({ principalKind: 'session_guest', principalId }); + return this.revokePrincipal({ + principalKind: 'session_guest', + principalId, + }); + } + + createTurnAccessRequest( + principalId: string, + input: CollaborationTurnRequestCreateInput, + ): Promise { + return this.#mutate(async () => { + const grant = this.activeSessionGrant( + principalId, + input.intent.sessionId, + 'session_turn_request', + ); + if (!grant) { + throw new RuntimeHostAccessInputError('This Guest cannot request a Turn in this Session'); + } + const retainedRequests = this.#file.turnAccessRequests; + const existing = retainedRequests.find( + (request) => + request.principalId === principalId && + request.intent.sessionId === input.intent.sessionId && + request.intent.turnId === input.intent.turnId, + ); + if (existing) { + if (existing.intent.content.text !== input.intent.content.text) { + throw new RuntimeHostAccessInputError( + 'A Turn access request already uses this Turn identity with different content', + ); + } + return existing; + } + if ( + retainedRequests.filter(isActiveTurnAccessRequest).length >= TURN_ACCESS_REQUEST_ACTIVE_MAX + ) { + throw new RuntimeHostAccessInputError( + 'Too many Turn access requests are awaiting an Owner decision or admission', + ); + } + if ( + retainedRequests.filter( + (request) => request.principalId === principalId && !isActiveTurnAccessRequest(request), + ).length >= TURN_ACCESS_REQUEST_ACTIVE_MAX + ) { + throw new RuntimeHostAccessInputError( + 'Review earlier Turn access request results before creating another request', + ); + } + const request: SessionTurnAccessRequest = { + requestId: randomUUID(), + principalId, + grantId: grant.grantId, + intent: input.intent, + createdAt: new Date().toISOString(), + state: { kind: 'pending' }, + }; + const nextFile = createAccessCredentialFile( + this.#file.credentials, + this.#file.sessionGrants, + [...retainedRequests, request], + ); + assertAccessCredentialFileCapacity(nextFile); + await this.#commit(nextFile); + return request; + }); + } + + queryTurnAccessRequests( + principal: Pick, + input: CollaborationTurnRequestQueryInput, + ): CollaborationTurnRequestQueryResult { + const guest = principal.principalKind === 'session_guest'; + return { + requests: this.#file.turnAccessRequests.filter( + (request) => + (!guest || request.principalId === principal.principalId) && + request.intent.sessionId === input.sessionId, + ), + }; + } + + acknowledgeTurnAccessRequest( + principalId: string, + input: CollaborationTurnRequestAcknowledgeInput, + ): Promise { + return this.#mutate(async () => { + const current = this.#file.turnAccessRequests.find( + (request) => request.requestId === input.requestId && request.principalId === principalId, + ); + if ( + !current || + current.state.kind === 'pending' || + (current.state.kind === 'approved' && current.state.admission === 'pending') + ) { + return { acknowledged: false }; + } + await this.#commit( + createAccessCredentialFile( + this.#file.credentials, + this.#file.sessionGrants, + this.#file.turnAccessRequests.filter((request) => request !== current), + ), + ); + return { acknowledged: true }; + }); + } + + decideTurnAccessRequest( + principalId: string, + input: CollaborationTurnRequestDecideInput, + ): Promise { + return this.#mutate(async () => { + const current = this.#file.turnAccessRequests.find( + (request) => request.requestId === input.requestId, + ); + if (!current) return { kind: 'not_found' }; + if (current.state.kind !== 'pending') { + return { kind: 'already_decided', request: current }; + } + if ( + input.decision === 'approve' && + !this.activeSessionGrant( + current.principalId, + current.intent.sessionId, + 'session_turn_request', + ) + ) { + throw new RuntimeHostAccessInputError('The Guest can no longer request this Turn'); + } + const decidedAt = new Date().toISOString(); + const request: SessionTurnAccessRequest = { + ...current, + state: + input.decision === 'approve' + ? { + kind: 'approved', + decidedAt, + decidedBy: principalId, + admission: 'pending', + } + : { kind: 'rejected', decidedAt, decidedBy: principalId }, + }; + await this.#commit( + createAccessCredentialFile( + this.#file.credentials, + this.#file.sessionGrants, + replaceTurnAccessRequest(this.#file.turnAccessRequests, current, request), + ), + [], + [], + input.decision === 'approve' ? [request] : [], + ); + return { kind: 'decided', request }; + }); + } + + completeTurnAccessRequest( + requestId: string, + admission: 'started' | 'blocked' | 'failed', + ): Promise { + return this.#mutate(async () => { + const current = this.#file.turnAccessRequests.find( + (request) => request.requestId === requestId, + ); + if (!current || current.state.kind !== 'approved' || current.state.admission !== 'pending') { + return; + } + const guestCanObserve = this.#file.credentials.some( + (credential) => + credential.principalKind === 'session_guest' && + credential.principalId === current.principalId && + credential.status !== 'revoked', + ); + const request: SessionTurnAccessRequest = { + ...current, + state: { + ...current.state, + admission, + }, + }; + await this.#commit( + createAccessCredentialFile( + this.#file.credentials, + this.#file.sessionGrants, + guestCanObserve + ? replaceTurnAccessRequest(this.#file.turnAccessRequests, current, request) + : this.#file.turnAccessRequests.filter((candidate) => candidate !== current), + ), + ); + }); + } + + approvedTurnAccessRequests(): readonly SessionTurnAccessRequest[] { + return this.#file.turnAccessRequests.filter( + (request) => request.state.kind === 'approved' && request.state.admission === 'pending', + ); } activeSessionGrant( @@ -418,7 +668,11 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { replaced.length === 0 ? this.#file.credentials : this.#file.credentials.filter((candidate) => !replaced.includes(candidate)); - const nextFile = createAccessCredentialFile([...retained, stored], this.#file.sessionGrants); + const nextFile = createNextAccessCredentialFile( + this.#file, + [...retained, stored], + this.#file.sessionGrants, + ); assertAccessCredentialFileCapacity(nextFile); const deliveryId = await createAccessCredentialDelivery( this.#controlDirectory, @@ -474,9 +728,13 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { return [{ ...revoked, status: 'revoked' as const, revokedAt }]; }); await this.#commit( - createAccessCredentialFile( + createNextAccessCredentialFile( + this.#file, credentials, this.#file.sessionGrants.filter((grant) => !activeGrants.includes(grant)), + input.principalKind === 'session_guest' + ? retirePrincipalTurnAccessRequests(this.#file.turnAccessRequests, input.principalId) + : this.#file.turnAccessRequests, ), [...matchedIds], activeGrants, @@ -536,7 +794,7 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { }; }); await this.#commit( - createAccessCredentialFile(credentials, this.#file.sessionGrants), + createNextAccessCredentialFile(this.#file, credentials, this.#file.sessionGrants), [current, ...pendingForPrincipal].map((credential) => credential.credentialId), ); return { credentialId, revoked: true }; @@ -580,7 +838,8 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { credential.principalKind === retained.principalKind && credential.principalId === retained.principalId, ); - const finalized = createAccessCredentialFile( + const finalized = createNextAccessCredentialFile( + this.#file, this.#file.credentials .filter((credential) => !revoked.includes(credential)) .map((credential) => @@ -594,7 +853,9 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { finalized, revoked.map((credential) => credential.credentialId), ); - return { reconnectRequired: retained.bindClientInstanceOnFinalize === true }; + return { + reconnectRequired: retained.bindClientInstanceOnFinalize === true, + }; } subscribeRevocations(listener: (credentialId: string) => void): () => void { @@ -609,6 +870,14 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { return () => this.#grantRevocationListeners.delete(listener); } + subscribeApprovedTurnAccessRequests( + listener: (request: SessionTurnAccessRequest) => void, + ): () => void { + if (this.#closed) return () => undefined; + this.#approvedTurnAccessRequestListeners.add(listener); + return () => this.#approvedTurnAccessRequestListeners.delete(listener); + } + close(): Promise { if (!this.#closed) { this.#closed = true; @@ -616,6 +885,7 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { this.#expiryTimer = undefined; this.#revocationListeners.clear(); this.#grantRevocationListeners.clear(); + this.#approvedTurnAccessRequestListeners.clear(); } return this.#mutation; } @@ -636,6 +906,7 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { file: AccessCredentialFile, revokedCredentialIds: readonly string[] = [], revokedGrants: readonly SessionCollaborationGrant[] = [], + approvedTurnAccessRequests: readonly SessionTurnAccessRequest[] = [], ): Promise { let outcomeUnknown: RuntimeHostAccessCommitOutcomeUnknownError | undefined; try { @@ -648,6 +919,11 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { this.#schedulePendingExpiry(); for (const credentialId of revokedCredentialIds) this.#publishRevocation(credentialId); for (const grant of revokedGrants) this.#publishGrantRevocation(grant); + if (!outcomeUnknown) { + for (const request of approvedTurnAccessRequests) { + for (const listener of this.#approvedTurnAccessRequestListeners) listener(request); + } + } if (outcomeUnknown) throw outcomeUnknown; } @@ -669,7 +945,8 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { expiredGuestPrincipals.has(grant.principalId), ); await this.#commit( - createAccessCredentialFile( + createNextAccessCredentialFile( + this.#file, this.#file.credentials.filter((credential) => !expired.includes(credential)), this.#file.sessionGrants.filter((grant) => !expiredGrants.includes(grant)), ), @@ -726,6 +1003,39 @@ class FileRuntimeHostAccessAuthority implements RuntimeHostAccessAuthority { } } +function isActiveTurnAccessRequest(request: SessionTurnAccessRequest): boolean { + return ( + request.state.kind === 'pending' || + (request.state.kind === 'approved' && request.state.admission === 'pending') + ); +} + +function replaceTurnAccessRequest( + requests: readonly SessionTurnAccessRequest[], + current: SessionTurnAccessRequest, + replacement: SessionTurnAccessRequest, +): readonly SessionTurnAccessRequest[] { + return [...requests.filter((request) => request !== current), replacement]; +} + +function removePendingTurnAccessRequests( + requests: readonly SessionTurnAccessRequest[], + matches: (request: SessionTurnAccessRequest) => boolean, +): readonly SessionTurnAccessRequest[] { + return requests.filter((request) => request.state.kind !== 'pending' || !matches(request)); +} + +function retirePrincipalTurnAccessRequests( + requests: readonly SessionTurnAccessRequest[], + principalId: string, +): readonly SessionTurnAccessRequest[] { + return requests.filter( + (request) => + request.principalId !== principalId || + (request.state.kind === 'approved' && request.state.admission === 'pending'), + ); +} + function activatePendingCredential( credential: StoredAccessCredential, clientInstanceId: string, @@ -767,7 +1077,10 @@ export async function issueAccessCredential( return { ok: true, result: await authority.issue(input) }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -786,7 +1099,10 @@ export async function replaceAccessCredential( return { ok: true, result: await authority.replace(input) }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -805,7 +1121,10 @@ export async function prepareAccessCredential( return { ok: true, result: await authority.prepare(input) }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -824,7 +1143,10 @@ export async function prepareAccessCredentialRotation( return { ok: true, result: await authority.prepareRotation(input) }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -843,7 +1165,10 @@ export async function revokeAccessCredential( return { ok: true, result: await authority.revoke(input) }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -878,7 +1203,10 @@ export async function revokeAccessCredentialRotation( return { ok: true, result: await authority.revokeRotation(input) }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -897,20 +1225,32 @@ export async function finalizeAccessCredential( if (!credentialId) { return { ok: false, - error: { code: 'invalid_request', message: 'A remote access credential is required' }, + error: { + code: 'invalid_request', + message: 'A remote access credential is required', + }, }; } if (!clientInstanceId) { return { ok: false, - error: { code: 'invalid_request', message: 'A Client identity is required' }, + error: { + code: 'invalid_request', + message: 'A Client identity is required', + }, }; } try { - return { ok: true, result: await authority.finalize(credentialId, clientInstanceId) }; + return { + ok: true, + result: await authority.finalize(credentialId, clientInstanceId), + }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -927,10 +1267,16 @@ export async function prepareCollaborationInvitation( ): Promise> { if (!authority) return collaborationUnavailable('collaboration.invitation.prepare'); try { - return { ok: true, result: await authority.prepareCollaborationInvitation(rootId, input) }; + return { + ok: true, + result: await authority.prepareCollaborationInvitation(rootId, input), + }; } catch (error) { if (error instanceof RuntimeHostAccessInputError) { - return { ok: false, error: { code: 'invalid_request', message: error.message } }; + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; } return accessPersistenceFailure( error, @@ -940,13 +1286,109 @@ export async function prepareCollaborationInvitation( } } +export async function createCollaborationTurnRequest( + authority: RuntimeHostAccessAuthority | undefined, + principalId: string, + input: CollaborationTurnRequestCreateInput, +): Promise> { + if (!authority) return collaborationUnavailable('collaboration.turn-request.create'); + try { + return { + ok: true, + result: await authority.createTurnAccessRequest(principalId, input), + }; + } catch (error) { + if (error instanceof RuntimeHostAccessInputError) { + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; + } + return accessPersistenceFailure( + error, + 'Turn access request outcome is unknown', + 'Turn access request could not be created', + ); + } +} + +export function queryCollaborationTurnRequests( + authority: RuntimeHostAccessAuthority | undefined, + principal: { + readonly principalId: string; + readonly principalKind: RuntimeHostConnectionAuthority['principalKind'] | undefined; + }, + input: CollaborationTurnRequestQueryInput, +): OperationOutcome<'collaboration.turn-request.query'> { + return authority && principal.principalKind + ? { + ok: true, + result: authority.queryTurnAccessRequests( + { + principalId: principal.principalId, + principalKind: principal.principalKind, + }, + input, + ), + } + : collaborationUnavailable('collaboration.turn-request.query'); +} + +export async function acknowledgeCollaborationTurnRequest( + authority: RuntimeHostAccessAuthority | undefined, + principalId: string, + input: CollaborationTurnRequestAcknowledgeInput, +): Promise> { + if (!authority) return collaborationUnavailable('collaboration.turn-request.acknowledge'); + try { + return { + ok: true, + result: await authority.acknowledgeTurnAccessRequest(principalId, input), + }; + } catch (error) { + return accessPersistenceFailure( + error, + 'Turn access acknowledgement outcome is unknown', + 'Turn access request could not be acknowledged', + ); + } +} + +export async function decideCollaborationTurnRequest( + authority: RuntimeHostAccessAuthority | undefined, + principalId: string, + input: CollaborationTurnRequestDecideInput, +): Promise> { + if (!authority) return collaborationUnavailable('collaboration.turn-request.decide'); + try { + return { + ok: true, + result: await authority.decideTurnAccessRequest(principalId, input), + }; + } catch (error) { + if (error instanceof RuntimeHostAccessInputError) { + return { + ok: false, + error: { code: 'invalid_request', message: error.message }, + }; + } + return accessPersistenceFailure( + error, + 'Turn access decision outcome is unknown', + 'Turn access request could not be decided', + ); + } +} export async function revokeCollaborationGrant( authority: RuntimeHostAccessAuthority | undefined, input: CollaborationGrantRevokeInput, ): Promise> { if (!authority) return collaborationUnavailable('collaboration.grant.revoke'); try { - return { ok: true, result: await authority.revokeCollaborationGrant(input) }; + return { + ok: true, + result: await authority.revokeCollaborationGrant(input), + }; } catch (error) { return accessPersistenceFailure( error, @@ -962,7 +1404,10 @@ export async function revokeCollaborationPrincipal( ): Promise> { if (!authority) return collaborationUnavailable('collaboration.principal.revoke'); try { - return { ok: true, result: await authority.revokeCollaborationPrincipal(principalId) }; + return { + ok: true, + result: await authority.revokeCollaborationPrincipal(principalId), + }; } catch (error) { return accessPersistenceFailure( error, @@ -976,7 +1421,11 @@ function collaborationUnavailable< K extends | 'collaboration.invitation.prepare' | 'collaboration.grant.revoke' - | 'collaboration.principal.revoke', + | 'collaboration.principal.revoke' + | 'collaboration.turn-request.create' + | 'collaboration.turn-request.acknowledge' + | 'collaboration.turn-request.decide' + | 'collaboration.turn-request.query', >(operation: K): OperationOutcome { return { ok: false, diff --git a/packages/runtime-host/src/server/access-credential-store.ts b/packages/runtime-host/src/server/access-credential-store.ts index e4fe1b819d..292cede830 100644 --- a/packages/runtime-host/src/server/access-credential-store.ts +++ b/packages/runtime-host/src/server/access-credential-store.ts @@ -25,10 +25,12 @@ import { HOST_OPERATION_SPECS, operationAllowsRemoteOwner, type SessionCollaborationGrant, + decodeSessionTurnAccessRequest, + type SessionTurnAccessRequest, type OperationKey, } from '../protocol/index.js'; -const ACCESS_FILE_SCHEMA_VERSION = 2; +const ACCESS_FILE_SCHEMA_VERSION = 3; const ACCESS_FILE_MAX_BYTES = 512 * 1024; const LEGACY_TRANSCRIPT_QUERY_GRANT = 'session.transcript.query'; const TRANSCRIPT_QUERY_REPLACEMENT_GRANTS = [ @@ -58,6 +60,9 @@ export const ACCESS_FILE_NAME = 'runtime-host-access.json'; export const SESSION_GUEST_OPERATION_GRANTS = Object.freeze([ 'host.status', 'artifact.query', + 'collaboration.turn-request.create', + 'collaboration.turn-request.acknowledge', + 'collaboration.turn-request.query', 'runtime.resource.query', 'session.shared.query', 'subscription.open', @@ -86,6 +91,7 @@ export interface AccessCredentialFile { readonly schemaVersion: typeof ACCESS_FILE_SCHEMA_VERSION; readonly credentials: readonly StoredAccessCredential[]; readonly sessionGrants: readonly SessionCollaborationGrant[]; + readonly turnAccessRequests: readonly SessionTurnAccessRequest[]; } export class RuntimeHostAccessInputError extends Error { @@ -112,8 +118,14 @@ export class RuntimeHostAccessCommitOutcomeUnknownError extends Error { export function createAccessCredentialFile( credentials: readonly StoredAccessCredential[], sessionGrants: readonly SessionCollaborationGrant[] = [], + turnAccessRequests: readonly SessionTurnAccessRequest[] = [], ): AccessCredentialFile { - return { schemaVersion: ACCESS_FILE_SCHEMA_VERSION, credentials, sessionGrants }; + return { + schemaVersion: ACCESS_FILE_SCHEMA_VERSION, + credentials, + sessionGrants, + turnAccessRequests, + }; } export function issuedAccessGrants(grants: readonly OperationKey[]): readonly OperationKey[] { @@ -132,10 +144,28 @@ export function assertAccessCredentialFileCapacity(file: AccessCredentialFile): }, ), file.sessionGrants, + file.turnAccessRequests.map(reserveTurnAccessRequestCompletionCapacity), ); serializeAccessCredentialFile(fullyRevoked); } +function reserveTurnAccessRequestCompletionCapacity( + request: SessionTurnAccessRequest, +): SessionTurnAccessRequest { + if (request.state.kind !== 'pending' && request.state.kind !== 'approved') return request; + if (request.state.kind === 'approved' && request.state.admission !== 'pending') return request; + return { + ...request, + state: { + kind: 'approved', + decidedAt: + request.state.kind === 'approved' ? request.state.decidedAt : '9999-12-31T23:59:59.999Z', + decidedBy: request.state.kind === 'approved' ? request.state.decidedBy : 'x'.repeat(128), + admission: 'failed', + }, + }; +} + export async function readAccessCredentialFile(path: string): Promise { let handle: FileHandle; try { @@ -193,7 +223,10 @@ function serializeAccessCredentialFile(file: AccessCredentialFile): string { } function decodeAccessFile(value: unknown): AccessCredentialFile { - if (!isRecord(value) || (value.schemaVersion !== 1 && value.schemaVersion !== 2)) { + if ( + !isRecord(value) || + (value.schemaVersion !== 1 && value.schemaVersion !== 2 && value.schemaVersion !== 3) + ) { throw new Error('Unsupported Runtime Host access file'); } if (!Array.isArray(value.credentials)) throw new Error('Invalid Runtime Host access file'); @@ -228,7 +261,21 @@ function decodeAccessFile(value: unknown): AccessCredentialFile { } sessionByGuest.set(grant.principalId, grant.sessionId); } - return createAccessCredentialFile(credentials, sessionGrants); + const turnAccessRequests = + value.schemaVersion < 3 + ? [] + : Array.isArray(value.turnAccessRequests) + ? value.turnAccessRequests.map(decodeSessionTurnAccessRequest) + : (() => { + throw new Error('Invalid Runtime Host Turn access requests'); + })(); + if ( + new Set(turnAccessRequests.map((request) => request.requestId)).size !== + turnAccessRequests.length + ) { + throw new Error('Duplicate Runtime Host Turn access request identity'); + } + return createAccessCredentialFile(credentials, sessionGrants, turnAccessRequests); } function decodeStoredCredential(value: unknown): StoredAccessCredential { diff --git a/packages/runtime-host/src/server/connection-authority.ts b/packages/runtime-host/src/server/connection-authority.ts index 3b7c9f5b02..8e7f83a1bf 100644 --- a/packages/runtime-host/src/server/connection-authority.ts +++ b/packages/runtime-host/src/server/connection-authority.ts @@ -81,7 +81,11 @@ export function authorizeRuntimeHostOperation( authority: RuntimeHostConnectionAuthority, frame: RequestFrame, ): boolean { - if (authority.principalKind !== 'local_owner' && !operationAllowsRemoteOwner(frame.operation)) { + if ( + (authority.principalKind === 'remote_owner' || + authority.principalKind === 'capability_provider') && + !operationAllowsRemoteOwner(frame.operation) + ) { return false; } if (authority.operationGrants !== 'all' && !authority.operationGrants.includes(frame.operation)) { diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index edc2ff8418..ac1b5e85f7 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -141,6 +141,7 @@ import { } from './host-composition.js'; import { HostInteractionCoordinator } from './interaction-coordinator.js'; import { HostInteractiveTurnCoordinator } from './interactive-turn-coordinator.js'; +import { SessionTurnAccessRequestCoordinator } from './session-turn-access-request-coordinator.js'; import { ensureBootstrapRuntimePolicy } from './bootstrap-runtime-policy.js'; import { hostedExecutionRunProfile } from './hosted-execution-tool-profile.js'; import { HostMemoryCoordinator } from './memory-coordinator.js'; @@ -493,7 +494,9 @@ export async function createExecutionRuntimeHostComposition( const projects = new HostProjectCatalogCoordinator( openedProjectCatalog, { publish: () => hostChanges.publishProjectCatalog() }, - { publish: (sessionId: string) => hostChanges.publishSessionCatalog(sessionId) }, + { + publish: (sessionId: string) => hostChanges.publishSessionCatalog(sessionId), + }, projectMembership, context.requestDrain, new HostProjectDirectoryAuthority(options.projectDirectoryRoots), @@ -1177,6 +1180,16 @@ export async function createExecutionRuntimeHostComposition( turns: stores.agentRunStore, runtime: manager, }); + const turnAccessRequests = context.sessionAccessAuthority + ? new SessionTurnAccessRequestCoordinator({ + authority: context.sessionAccessAuthority, + startTurn: interactiveTurns.handlers['turn.start'], + hostEpoch: context.hostEpoch, + acquireResidency: () => context.acquireResidency('collaboration-turn-request'), + requestDrain: context.requestDrain, + whenIdle: (sessionId) => coordinator.whenIdle(sessionId), + }) + : undefined; const graphExecutions = new HostAgentGraphExecutionCoordinator({ executions: coordinator, runtime: manager, @@ -1687,10 +1700,12 @@ export async function createExecutionRuntimeHostComposition( await messages.recoverPendingAfterHostRestart( recoverySessions.map((session) => session.id), ); + await turnAccessRequests?.recover(); rootRecoveryCompleted = true; }, }, drain: [ + () => turnAccessRequests?.beginDrain(), () => rootCoordinator?.beginDrain(), () => workspaceExecution?.beginDrain(), () => runtimeResources?.beginDrain(), @@ -1709,6 +1724,7 @@ export async function createExecutionRuntimeHostComposition( () => sessionEffects?.close(), () => messages.close(), () => interactions.close(), + () => turnAccessRequests?.close(), () => continuityCoordinator.close(), ], releaseConnection: [(connectionId) => runtimeResources?.releaseConnection(connectionId)], @@ -1877,7 +1893,10 @@ function adaptWorkspaceFilesystemWorker( async execute(input) { // Read-only operations never participate in CAS; the adapter says so // explicitly (#3484) instead of relying on an absent optional field. - const result = await worker.execute({ ...input, expectedIdentity: 'unchecked' }); + const result = await worker.execute({ + ...input, + expectedIdentity: 'unchecked', + }); switch (result.kind) { case 'read': case 'read_image': diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index c0be10e94f..4ab9603da7 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -59,8 +59,12 @@ import { } from './operation-dispatcher.js'; import { issueAccessCredential, + acknowledgeCollaborationTurnRequest, + createCollaborationTurnRequest, + decideCollaborationTurnRequest, finalizeAccessCredential, prepareCollaborationInvitation, + queryCollaborationTurnRequests, prepareAccessCredential, prepareAccessCredentialRotation, replaceAccessCredential, @@ -121,7 +125,12 @@ export interface RuntimeHostCompositionContext { requestDrain(): void; sessionAccessAuthority?: Pick< RuntimeHostAccessAuthority, - 'activeSessionGrant' | 'activeSessionGrantForPrincipal' | 'subscribeGrantRevocations' + | 'activeSessionGrant' + | 'activeSessionGrantForPrincipal' + | 'approvedTurnAccessRequests' + | 'completeTurnAccessRequest' + | 'subscribeGrantRevocations' + | 'subscribeApprovedTurnAccessRequests' >; waitForResidencies?(): Promise; waitForResidenciesExcept?(excludedLabel: string): Promise; @@ -723,7 +732,10 @@ export class RuntimeHostKernel { ), 'collaboration.access.query': async (input) => this.#options.accessAuthority - ? { ok: true, result: this.#options.accessAuthority.queryCollaborationAccess(input) } + ? { + ok: true, + result: this.#options.accessAuthority.queryCollaborationAccess(input), + } : { ok: false, error: { @@ -739,6 +751,31 @@ export class RuntimeHostKernel { this.#settleAccessCredentialMutation( revokeCollaborationPrincipal(this.#options.accessAuthority, input.principalId), ), + 'collaboration.turn-request.create': async (input, context) => + this.#settleAccessCredentialMutation( + createCollaborationTurnRequest(this.#options.accessAuthority, context.principal, input), + ), + 'collaboration.turn-request.query': async (input, context) => + queryCollaborationTurnRequests( + this.#options.accessAuthority, + { + principalId: context.principal, + principalKind: context.principalKind, + }, + input, + ), + 'collaboration.turn-request.acknowledge': async (input, context) => + this.#settleAccessCredentialMutation( + acknowledgeCollaborationTurnRequest( + this.#options.accessAuthority, + context.principal, + input, + ), + ), + 'collaboration.turn-request.decide': async (input, context) => + this.#settleAccessCredentialMutation( + decideCollaborationTurnRequest(this.#options.accessAuthority, context.principal, input), + ), }, createPeerMeshOperationHandlers(this.#options.peerMesh, { requestDrain: () => this.#requestDrain(), diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index 856afe48bd..b752189df3 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -19,6 +19,7 @@ import { truncateUtf8 } from '@maka/core/diagnostic-log'; import { redactSecrets } from '@maka/core/redaction'; +import type { RootTurnAdmissionAuthorization } from '@maka/storage/execution-stores'; import { HOST_OPERATION_SPECS, decodeOperationOutcome, @@ -45,6 +46,7 @@ export interface ConnectionContext { principalKind?: RuntimeHostConnectionAuthority['principalKind']; credentialId?: string; clientInstanceId?: string; + turnAdmissionAuthorization?: RootTurnAdmissionAuthorization; acquireResidency(): OperationResidency; } @@ -366,6 +368,34 @@ export function createUnavailableAccessAuthorityOperationHandlers(): AccessAutho message: 'Runtime Host collaboration authority is unavailable', }, }), + 'collaboration.turn-request.create': async () => ({ + ok: false, + error: { + code: 'operation_unavailable', + message: 'Runtime Host collaboration authority is unavailable', + }, + }), + 'collaboration.turn-request.acknowledge': async () => ({ + ok: false, + error: { + code: 'operation_unavailable', + message: 'Runtime Host collaboration authority is unavailable', + }, + }), + 'collaboration.turn-request.query': async () => ({ + ok: false, + error: { + code: 'operation_unavailable', + message: 'Runtime Host collaboration authority is unavailable', + }, + }), + 'collaboration.turn-request.decide': async () => ({ + ok: false, + error: { + code: 'operation_unavailable', + message: 'Runtime Host collaboration authority is unavailable', + }, + }), }; } diff --git a/packages/runtime-host/src/server/root-admission-owner.ts b/packages/runtime-host/src/server/root-admission-owner.ts index b199fb930b..759fa3ee2c 100644 --- a/packages/runtime-host/src/server/root-admission-owner.ts +++ b/packages/runtime-host/src/server/root-admission-owner.ts @@ -120,6 +120,7 @@ function sameRootAdmission(left: RootTurnAdmission, right: RootTurnAdmission): b isDeepStrictEqual(left.execution, right.execution) && isDeepStrictEqual(left.turnOrchestration, right.turnOrchestration) && isDeepStrictEqual(left.skillInvocation, right.skillInvocation) && + isDeepStrictEqual(left.authorization, right.authorization) && left.previousRootTurnId === right.previousRootTurnId && (left.normalizedInput === null || right.normalizedInput === null ? left.normalizedInput === right.normalizedInput @@ -154,6 +155,9 @@ function snapshotAdmission(admission: RootTurnAdmission): RootTurnAdmission { ...(admission.turnOrchestration ? { turnOrchestration: Object.freeze({ ...admission.turnOrchestration }) } : {}), + ...(admission.authorization + ? { authorization: Object.freeze({ ...admission.authorization }) } + : {}), normalizedInput: admission.normalizedInput === null ? null : snapshotMessageContent(admission.normalizedInput), sourceMessages: Object.freeze(sourceMessages), diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index 2406f4ee1c..86d7ceed9c 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -1438,7 +1438,14 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { ? request.content : requireHostedExecutionMessageContent(existing); } - if (!rootMessageAdmissionMatches(existing, request, content)) { + if ( + !rootMessageAdmissionMatches( + existing, + request, + content, + context.turnAdmissionAuthorization, + ) + ) { return completedStart( operationConflict('Turn identity was already admitted with a different payload'), ); @@ -1543,6 +1550,9 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { normalizedInput: canonicalContent.content, ...(request.turnOrchestration ? { turnOrchestration: request.turnOrchestration } : {}), ...(prepared.skillInvocation ? { skillInvocation: prepared.skillInvocation } : {}), + ...(context.turnAdmissionAuthorization + ? { authorization: context.turnAdmissionAuthorization } + : {}), sourceMessages: [], admittedAt: Date.now(), }); @@ -1551,7 +1561,14 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { operationConflict('Turn identity belongs to a different execution kind'), ); } - if (!rootMessageAdmissionMatches(admitted.admission, request, canonicalContent.content)) { + if ( + !rootMessageAdmissionMatches( + admitted.admission, + request, + canonicalContent.content, + context.turnAdmissionAuthorization, + ) + ) { return completedStart( operationConflict('Turn identity was already admitted with a different payload'), ); @@ -2726,6 +2743,7 @@ function rootMessageAdmissionMatches( admission: RootTurnAdmission, request: RootMessageStartRequest, content: MessageContent, + authorization: ConnectionContext['turnAdmissionAuthorization'], ): boolean { return ( isDeepStrictEqual(admission.execution, request.execution) && @@ -2733,6 +2751,7 @@ function rootMessageAdmissionMatches( ? true : messageContentsEqual(requireHostedExecutionMessageContent(admission), content)) && isDeepStrictEqual(admission.turnOrchestration, request.turnOrchestration) && + isDeepStrictEqual(admission.authorization, authorization) && admission.sourceMessages.length === 0 ); } diff --git a/packages/runtime-host/src/server/session-turn-access-request-coordinator.ts b/packages/runtime-host/src/server/session-turn-access-request-coordinator.ts new file mode 100644 index 0000000000..6cd33f19b2 --- /dev/null +++ b/packages/runtime-host/src/server/session-turn-access-request-coordinator.ts @@ -0,0 +1,143 @@ +/* + * 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 { SessionTurnAccessRequest } from '../protocol/index.js'; +import type { RuntimeHostAccessAuthority } from './access-authority.js'; +import type { + ConnectionContext, + OperationHandler, + OperationResidency, +} from './operation-dispatcher.js'; + +type TurnAccessRequestAuthority = Pick< + RuntimeHostAccessAuthority, + 'approvedTurnAccessRequests' | 'completeTurnAccessRequest' | 'subscribeApprovedTurnAccessRequests' +>; + +export class SessionTurnAccessRequestCoordinator { + readonly #authority: TurnAccessRequestAuthority; + readonly #startTurn: OperationHandler<'turn.start'>; + readonly #acquireResidency: () => OperationResidency; + readonly #requestDrain: () => void; + readonly #whenIdle: (sessionId: string) => Promise | undefined; + readonly #hostEpoch: string; + readonly #tasks = new Map>(); + readonly #unsubscribe: () => void; + #draining = false; + + constructor(input: { + readonly authority: TurnAccessRequestAuthority; + readonly startTurn: OperationHandler<'turn.start'>; + readonly acquireResidency: () => OperationResidency; + readonly requestDrain: () => void; + readonly whenIdle: (sessionId: string) => Promise | undefined; + readonly hostEpoch: string; + }) { + this.#authority = input.authority; + this.#startTurn = input.startTurn; + this.#acquireResidency = input.acquireResidency; + this.#requestDrain = input.requestDrain; + this.#whenIdle = input.whenIdle; + this.#hostEpoch = input.hostEpoch; + this.#unsubscribe = input.authority.subscribeApprovedTurnAccessRequests((request) => { + this.#schedule(request); + }); + } + + recover(): void { + for (const request of this.#authority.approvedTurnAccessRequests()) { + this.#schedule(request); + } + } + + beginDrain(): void { + if (this.#draining) return; + this.#draining = true; + this.#unsubscribe(); + } + + async close(): Promise { + this.beginDrain(); + await this.#settled(); + } + + #schedule(request: SessionTurnAccessRequest): void { + if (this.#draining || this.#tasks.has(request.requestId)) return; + const task = this.#admit(request).finally(() => { + this.#tasks.delete(request.requestId); + }); + this.#tasks.set(request.requestId, task); + } + + async #admit(request: SessionTurnAccessRequest): Promise { + if (request.state.kind !== 'approved' || request.state.admission !== 'pending') return; + const residency = this.#acquireResidency(); + const context: ConnectionContext = { + hostEpoch: this.#hostEpoch, + connectionId: `collaboration:${request.requestId}`, + principal: request.principalId, + turnAdmissionAuthorization: { + kind: 'session_turn_access_request', + requestId: request.requestId, + principalId: request.principalId, + grantId: request.grantId, + approvedAt: Date.parse(request.state.decidedAt), + approvedBy: request.state.decidedBy, + }, + acquireResidency: this.#acquireResidency, + }; + try { + let outcome: Awaited>>; + for (;;) { + try { + outcome = await this.#startTurn(request.intent, context); + } catch { + this.#requestDrain(); + return; + } + if (outcome.ok || outcome.error.code !== 'session_busy') break; + const whenIdle = this.#whenIdle(request.intent.sessionId); + if (whenIdle) { + try { + await whenIdle; + } catch { + this.#requestDrain(); + return; + } + } + if (this.#draining) return; + } + if (this.#draining && !outcome.ok) return; + try { + await this.#authority.completeTurnAccessRequest( + request.requestId, + outcome.ok ? outcome.result.kind : 'failed', + ); + } catch { + this.#requestDrain(); + } + } finally { + residency.release(); + } + } + + #settled(): Promise { + return Promise.all(this.#tasks.values()).then(() => undefined); + } +} diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index 6feb75144b..27481f3fec 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -123,10 +123,20 @@ export interface RootTurnAdmission { normalizedInput: MessageContent | null; turnOrchestration?: TurnOrchestration; skillInvocation?: SkillInvocationResult; + authorization?: RootTurnAdmissionAuthorization; sourceMessages: readonly RootTurnSourceMessage[]; admittedAt: number; } +export interface RootTurnAdmissionAuthorization { + readonly kind: 'session_turn_access_request'; + readonly requestId: string; + readonly principalId: string; + readonly grantId: string; + readonly approvedAt: number; + readonly approvedBy: string; +} + export interface RootTurnStartRejection { schemaVersion: 1; sessionId: string; @@ -146,6 +156,7 @@ export interface AdmitRootTurnInput { normalizedInput: MessageContent | null; turnOrchestration?: TurnOrchestration; skillInvocation?: SkillInvocationResult; + authorization?: RootTurnAdmissionAuthorization; sourceMessages: readonly RootTurnSourceMessage[]; admittedAt: number; } @@ -1213,6 +1224,7 @@ function normalizeAdmitRootTurnInput(input: AdmitRootTurnInput): RootTurnAdmissi input.skillInvocation === undefined ? undefined : decodeSkillInvocationResult(input.skillInvocation); + const authorization = normalizeRootTurnAdmissionAuthorization(input.authorization); const execution = normalizeRootExecutionDescriptor(input.execution); if (execution.kind === 'legacy_automation') { throw new Error('New root admission cannot use removed Automation authority'); @@ -1228,6 +1240,7 @@ function normalizeAdmitRootTurnInput(input: AdmitRootTurnInput): RootTurnAdmissi normalizedInput, ...(turnOrchestration ? { turnOrchestration } : {}), ...(skillInvocation ? { skillInvocation } : {}), + ...(authorization ? { authorization } : {}), sourceMessages, admittedAt: input.admittedAt, }; @@ -1442,6 +1455,7 @@ function normalizeRootTurnAdmission( record.skillInvocation === undefined ? undefined : decodeSkillInvocationResult(record.skillInvocation); + const authorization = normalizeRootTurnAdmissionAuthorization(record.authorization); const admission: RootTurnAdmission = { schemaVersion: ROOT_TURN_ADMISSION_SCHEMA_VERSION, sessionId, @@ -1453,6 +1467,7 @@ function normalizeRootTurnAdmission( normalizedInput, ...(turnOrchestration ? { turnOrchestration } : {}), ...(skillInvocation ? { skillInvocation } : {}), + ...(authorization ? { authorization } : {}), sourceMessages, admittedAt: record.admittedAt as number, }; @@ -1693,6 +1708,7 @@ function rootTurnAdmissionPayloadsEqual( isDeepStrictEqual(left.execution, right.execution) && isDeepStrictEqual(left.turnOrchestration, right.turnOrchestration) && isDeepStrictEqual(left.skillInvocation, right.skillInvocation) && + isDeepStrictEqual(left.authorization, right.authorization) && (left.normalizedInput === null || right.normalizedInput === null ? left.normalizedInput === right.normalizedInput : messageContentsEqual(left.normalizedInput, right.normalizedInput)) && @@ -1763,6 +1779,11 @@ function assertRootTurnAdmissionContract(admission: RootTurnAdmission): void { 'Invalid root turn admission contract: Skill invocation requires external message execution', ); } + if (admission.authorization && execution.kind !== 'external_message') { + throw new Error( + 'Invalid root turn admission contract: authorization proof requires external message execution', + ); + } if (execution.kind === 'claimed_agent_graph_intent') { if ( execution.claim.targetSessionId !== admission.sessionId || @@ -1824,6 +1845,7 @@ function deepFreezeRootTurnAdmission(admission: RootTurnAdmission): RootTurnAdmi Object.freeze(admission.execution); if (admission.turnOrchestration) Object.freeze(admission.turnOrchestration); if (admission.skillInvocation) Object.freeze(admission.skillInvocation); + if (admission.authorization) Object.freeze(admission.authorization); if (admission.normalizedInput) deepFreezeRootTurnMessageContent(admission.normalizedInput); for (const sourceMessage of admission.sourceMessages) { deepFreezeRootTurnMessageContent(sourceMessage.content); @@ -1846,6 +1868,44 @@ function normalizeTurnOrchestration(value: unknown): TurnOrchestration | undefin return Object.freeze({ mode: value.mode, source: value.source }); } +function normalizeRootTurnAdmissionAuthorization( + value: unknown, +): RootTurnAdmissionAuthorization | undefined { + if (value === undefined) return undefined; + if ( + !isPlainRecord(value) || + !hasExactKeys(value, [ + 'kind', + 'requestId', + 'principalId', + 'grantId', + 'approvedAt', + 'approvedBy', + ]) || + value.kind !== 'session_turn_access_request' || + typeof value.requestId !== 'string' || + !isSafeId(value.requestId) || + typeof value.principalId !== 'string' || + !isGraphControlIdentity(value.principalId) || + typeof value.grantId !== 'string' || + !isSafeId(value.grantId) || + !Number.isSafeInteger(value.approvedAt) || + (value.approvedAt as number) < 0 || + typeof value.approvedBy !== 'string' || + !isGraphControlIdentity(value.approvedBy) + ) { + throw new Error('Invalid root turn admission authorization'); + } + return Object.freeze({ + kind: value.kind, + requestId: value.requestId, + principalId: value.principalId, + grantId: value.grantId, + approvedAt: value.approvedAt as number, + approvedBy: value.approvedBy, + }); +} + function hasRootTurnAdmissionKeys(record: Record): boolean { const keys = [ 'schemaVersion', @@ -1859,7 +1919,7 @@ function hasRootTurnAdmissionKeys(record: Record): boolean { 'sourceMessages', 'admittedAt', ]; - const optionalKeys = ['turnOrchestration', 'skillInvocation'].filter((key) => + const optionalKeys = ['turnOrchestration', 'skillInvocation', 'authorization'].filter((key) => Object.hasOwn(record, key), ); return hasExactKeys(record, [...keys, ...optionalKeys]); diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index f50ffee65e..58b86e1346 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -40,6 +40,7 @@ import { type DurableRuntimeEventStore, type EvidenceReadBudget, type RootTurnAdmission, + type RootTurnAdmissionAuthorization, type RootTurnSourceMessageReceipt, } from './agent-run-store.js'; import { @@ -102,6 +103,7 @@ export type { EvidenceReadBudget, ImmutableSteeringMessageProof, RootTurnAdmission, + RootTurnAdmissionAuthorization, RootTurnAdmissionStore, RootTurnStartRejectionStore, RootTurnSourceMessage,