diff --git a/packages/core/src/__tests__/sandbox-boundary.test.ts b/packages/core/src/__tests__/sandbox-boundary.test.ts index 733caeb954..2ef563db1e 100644 --- a/packages/core/src/__tests__/sandbox-boundary.test.ts +++ b/packages/core/src/__tests__/sandbox-boundary.test.ts @@ -27,8 +27,10 @@ import { decodeExecutionBoundary, executionBoundaryContains, executionBoundaryDisplayMode, + executionBoundaryMatchesPermissionMode, validateSandboxBoundaryExpansion, } from '../sandbox-boundary.js'; +import type { PermissionMode } from '../permission.js'; import { canReadPath, canWritePath, @@ -73,6 +75,61 @@ describe('executionBoundaryDisplayMode', () => { ); }); + test('a widened read-only profile no longer matches explore (#3349)', () => { + const widened = applySandboxBoundaryExpansion(createReadOnlyPermissionProfile(), { + filesystem: { entries: [{ path: '/outside/dist', access: 'write', scope: 'subtree' }] }, + }); + const boundary = { kind: 'managed', profile: widened, revision: 1 } as const; + + // The name stayed 'read-only' while the structure became writable: a + // no-op short-circuit keyed on this answer must not bless that divergence. + expect(executionBoundaryMatchesPermissionMode(boundary, 'explore')).toBe(false); + expect(executionBoundaryMatchesPermissionMode(boundary, 'ask')).toBe(true); + }); + + test('matching follows the structural derivation, not the profile name', () => { + const customReadOnly: PermissionProfileManaged = { + ...createReadOnlyPermissionProfile(), + name: 'custom', + }; + expect( + executionBoundaryMatchesPermissionMode( + { kind: 'managed', profile: customReadOnly, revision: 0 }, + 'explore', + ), + ).toBe(true); + expect( + executionBoundaryMatchesPermissionMode( + { kind: 'managed', profile: createWorkspaceWritePermissionProfile(), revision: 0 }, + 'ask', + ), + ).toBe(true); + expect(executionBoundaryMatchesPermissionMode({ kind: 'bypass', revision: 0 }, 'bypass')).toBe( + true, + ); + expect( + executionBoundaryMatchesPermissionMode( + { kind: 'managed', profile: createWorkspaceWritePermissionProfile(), revision: 0 }, + 'bypass', + ), + ).toBe(false); + }); + + test('a legacy persisted execute value never matches and an external boundary is not verifiable', () => { + // 'execute' left the mode vocabulary on main; a stale persisted value must + // still never match, so it always routes through a transition instead of + // being blessed as already-committed. + expect( + executionBoundaryMatchesPermissionMode( + { kind: 'managed', profile: createWorkspaceWritePermissionProfile(), revision: 0 }, + 'execute' as PermissionMode, + ), + ).toBe(false); + expect(executionBoundaryMatchesPermissionMode({ kind: 'external', revision: 0 }, 'ask')).toBe( + false, + ); + }); + test('under-states danger-full-access as Auto rather than naming a mode for it', () => { // A deliberate collapse, NOT a description of this profile: the picker // offers two modes and no third one is being invented for a profile the diff --git a/packages/core/src/permission.ts b/packages/core/src/permission.ts index 6c716cc2cf..7e5039fc83 100644 --- a/packages/core/src/permission.ts +++ b/packages/core/src/permission.ts @@ -62,6 +62,21 @@ export function isPermissionMode(value: unknown): value is PermissionMode { return typeof value === 'string' && (PERMISSION_MODES as readonly string[]).includes(value); } +/** + * The permission mode a tool-facing consumer should act on: a plan-mode + * session presents read-only authority to tools unless it is bypassed. Shared + * by the runtime-host composer (build time) and the tool runtime (dispatch + * time), so both derive the same mode from the same inputs. + */ +export function resolveCollaborationPermissionMode(input: { + readonly collaborationMode: 'agent' | 'plan'; + readonly permissionMode: PermissionMode; +}): PermissionMode { + return input.collaborationMode === 'plan' && input.permissionMode !== 'bypass' + ? 'explore' + : input.permissionMode; +} + /** Canonical category names use Claude SDK terminology. Pi adapter MUST * translate Pi-native tool names into these before they reach the runtime. */ export type ToolCategory = diff --git a/packages/core/src/sandbox-boundary.ts b/packages/core/src/sandbox-boundary.ts index 208fd1f2b3..cb068bd9f4 100644 --- a/packages/core/src/sandbox-boundary.ts +++ b/packages/core/src/sandbox-boundary.ts @@ -226,6 +226,24 @@ export function executionBoundaryDisplayMode( return readOnly ? 'explore' : 'ask'; } +/** + * Whether the durable boundary already expresses the requested permission + * mode, derived through the same structural read (#1611) as the display mode: + * a read-only-named profile widened by an approved expansion no longer reads + * as explore. Callers that short-circuit a no-op configuration update on this + * answer must consult it: comparing the header's stored `permissionMode` + * alone would bless a header/boundary divergence as already-committed. + * Legacy 'execute' never matches — forcing the transition is the safe + * direction — and an external boundary is not locally verifiable. + */ +export function executionBoundaryMatchesPermissionMode( + boundary: ExecutionBoundary, + mode: PermissionMode, +): boolean { + const displayMode = executionBoundaryDisplayMode(boundary); + return displayMode !== undefined && displayMode === mode; +} + export function createGenesisExecutionBoundary(mode: PermissionMode): ExecutionBoundary { if (mode === 'bypass') return { kind: 'bypass', revision: 0 }; return { diff --git a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts index 8c2c6a669c..ad569765b7 100644 --- a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts @@ -391,6 +391,79 @@ test('typed configuration rejection does not request Host drain', async () => { assert.equal(fixture.drainRequests(), 0); }); +test('a no-op configuration update repairs a header/boundary divergence instead of blessing it', async () => { + // The header matches the requested configuration on every field, so only the + // boundary consistency check can tell a genuine no-op from a divergence that + // must be repaired through Runtime authority. + let transitions = 0; + const consistent = createFixture({ + stores: { + readHeaderRecordSnapshot: async () => headerSnapshot(matchingHeader(['user-label']), 3), + readCatalogRecord: async () => catalogRecord(matchingHeader(['user-label']), 3), + readExecutionBoundary: async () => ({ kind: 'bypass', revision: 1 }), + }, + manager: { + transitionSessionConfiguration: async () => { + transitions += 1; + return headerSnapshot(matchingHeader(['user-label']), 3); + }, + }, + }); + const consistentOutcome = await consistent.coordinator.handlers['session.configuration.update']( + bypassConfigurationInput(consistent.sessionId, consistent.revision()), + context, + ); + assert.equal(consistentOutcome.ok, true); + assert.equal(transitions, 0); + + const divergent = createFixture({ + stores: { + readHeaderRecordSnapshot: async () => headerSnapshot(matchingHeader(['user-label']), 3), + readCatalogRecord: async () => catalogRecord(matchingHeader(['user-label']), 3), + // The header says bypass while the durable boundary stays managed. + readExecutionBoundary: async () => createGenesisExecutionBoundary('ask'), + }, + manager: { + transitionSessionConfiguration: async () => { + transitions += 1; + return headerSnapshot(matchingHeader(['user-label']), 3); + }, + }, + }); + const divergentOutcome = await divergent.coordinator.handlers['session.configuration.update']( + bypassConfigurationInput(divergent.sessionId, divergent.revision()), + context, + ); + assert.equal(divergentOutcome.ok, true); + assert.equal(transitions, 1); +}); + +test('an externally isolated session keeps benign no-op updates on the header short-circuit', async () => { + let transitions = 0; + const fixture = createFixture({ + stores: { + readHeaderRecordSnapshot: async () => headerSnapshot(matchingHeader(['user-label']), 3), + readCatalogRecord: async () => catalogRecord(matchingHeader(['user-label']), 3), + readExecutionBoundary: async () => ({ kind: 'external', revision: 0 }), + }, + manager: { + transitionSessionConfiguration: async () => { + transitions += 1; + return headerSnapshot(matchingHeader(['user-label']), 3); + }, + }, + }); + const outcome = await fixture.coordinator.handlers['session.configuration.update']( + bypassConfigurationInput(fixture.sessionId, fixture.revision()), + context, + ); + // The external boundary is not locally verifiable, so the header comparison + // alone decides the no-op: the store would refuse to move an externally + // isolated boundary, and a benign re-apply must not become a failure. + assert.equal(outcome.ok, true); + assert.equal(transitions, 0); +}); + test('creation rejects reserved execution labels before claiming a Session identity', async () => { let createAttempts = 0; const fixture = createFixture({ @@ -1311,6 +1384,22 @@ function configurationInput( }; } +function bypassConfigurationInput( + sessionId: string, + expectedRevision: number, +): SessionConfigurationUpdateInput { + const base = configurationInput(sessionId, expectedRevision); + return { ...base, configuration: { ...base.configuration, permissionMode: 'bypass' } }; +} + +function matchingHeader(labels: readonly string[]): SessionHeader { + return { + ...sessionHeader('session-1', labels), + permissionMode: 'bypass', + orchestrationMode: 'graph', + }; +} + function sessionHeader(sessionId: string, labels: readonly string[]): SessionHeader { return { id: sessionId, diff --git a/packages/runtime-host/src/server/execution-model-composition.ts b/packages/runtime-host/src/server/execution-model-composition.ts index 5c0f3acdaa..fdfb9d96d2 100644 --- a/packages/runtime-host/src/server/execution-model-composition.ts +++ b/packages/runtime-host/src/server/execution-model-composition.ts @@ -23,7 +23,7 @@ import { resolveModelVisionSupport } from '@maka/core/model-metadata'; import { relayModelProfile } from '@maka/core/model-thinking'; import type { ModelCallAttempt } from '@maka/core/model-call-attempt'; import type { ModelCallCommit } from '@maka/core/agent-run'; -import type { PermissionMode } from '@maka/core/permission'; +import { resolveCollaborationPermissionMode } from '@maka/core/permission'; import { AiSdkBackend } from '@maka/runtime/ai-sdk-backend'; import { buildDefaultContextBudgetPolicy, @@ -468,11 +468,4 @@ class HostAiSdkBackend extends AiSdkBackend { } } -export function resolveCollaborationPermissionMode(input: { - readonly collaborationMode: 'agent' | 'plan'; - readonly permissionMode: PermissionMode; -}): PermissionMode { - return input.collaborationMode === 'plan' && input.permissionMode !== 'bypass' - ? 'explore' - : input.permissionMode; -} +export { resolveCollaborationPermissionMode } from '@maka/core/permission'; diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts index 760453eb89..70b47572ad 100644 --- a/packages/runtime-host/src/server/session-catalog-coordinator.ts +++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts @@ -24,6 +24,7 @@ import { isModelExplicitlyUnsupportedForChat } from '@maka/core/model-catalog'; import { thinkingVariantsForConnection } from '@maka/core/model-thinking'; import { executionBoundaryDisplayMode, + executionBoundaryMatchesPermissionMode, type ExecutionBoundary, type ExecutionBoundarySummary, } from '@maka/core/sandbox-boundary'; @@ -470,8 +471,19 @@ export class HostSessionCatalogCoordinator { input.configuration.thinkingLevel ?? undefined, ); const clearsConnectionBlock = current.header.blockedReason === 'NO_REAL_CONNECTION'; + // The boundary must match too: the header's stored permissionMode alone + // cannot bless a no-op, or a header/boundary divergence would be + // short-circuited as already-committed instead of repaired. An external + // boundary is not locally verifiable — the store refuses to move it into + // Auto or Bypass — so for those sessions the header comparison alone + // decides the no-op, keeping benign updates working. + const boundary = await this.#stores.readExecutionBoundary(input.sessionId); + const boundaryMatchesConfiguration = + boundary.kind === 'external' || + executionBoundaryMatchesPermissionMode(boundary, input.configuration.permissionMode); if ( !clearsConnectionBlock && + boundaryMatchesConfiguration && sessionConfigurationMatches(current.header, model, input.configuration) ) { return configurationSuccess({ diff --git a/packages/runtime/src/__tests__/runtime-kernel-queued-quiescent-mutation.test.ts b/packages/runtime/src/__tests__/runtime-kernel-queued-quiescent-mutation.test.ts new file mode 100644 index 0000000000..7f8fec3da2 --- /dev/null +++ b/packages/runtime/src/__tests__/runtime-kernel-queued-quiescent-mutation.test.ts @@ -0,0 +1,390 @@ +/* + * 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 assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; + +import type { SessionEvent } from '@maka/core/events'; +import type { SessionHeader, StoredMessage } from '@maka/core/session'; +import type { AgentBackend, BackendSendInput } from '@maka/core/backend-types'; + +import { RuntimeKernel, SessionQuiescentMutationBusyError } from '../runtime-kernel.js'; +import { + BackendRegistry, + type BackendFactoryContext, + type SessionStore, +} from '../session-manager.js'; + +const SESSION_ID = 'session-queued-quiescent'; +const OTHER_SESSION_ID = 'session-queued-quiescent-other'; + +describe('RuntimeKernel queued quiescent mutation', () => { + test('runs immediately when no execution claim exists', async () => { + const kernel = newKernel(); + assert.equal( + await within(kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => 'committed')), + 'committed', + ); + }); + + test('waits for a claim that already existed when the mutation was requested', async () => { + const kernel = newKernel(); + const claim = kernel.claimExecution(SESSION_ID); + const result = track(kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => 'committed')); + await settleTicks(); + assert.equal(result.settled, false, 'mutation must wait while the claim is held'); + + claim.release(); + assert.equal(await within(result.promise), 'committed'); + }); + + test('re-arms while older claims remain after one releases', async () => { + const kernel = newKernel(); + const first = kernel.claimExecution(SESSION_ID); + const second = kernel.claimExecution(SESSION_ID); + const result = track(kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => 'committed')); + + first.release(); + await settleTicks(); + assert.equal(result.settled, false, 'mutation must keep waiting for the older claim'); + + second.release(); + assert.equal(await within(result.promise), 'committed'); + }); + + test('does not wait for a claim created after the mutation was requested', async () => { + const kernel = newKernel(); + // A preceding in-flight mutation keeps the queued slot reserved-but-not-run, + // so the later claim below is created after the frontier is captured. + const gate = deferred(); + const started = deferred(); + const preceding = kernel.runSessionAdmissionMutation([SESSION_ID], async () => { + started.resolve(); + await gate.promise; + return 'first'; + }); + await started.promise; + + const result = track(kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => 'committed')); + const lateClaim = kernel.claimExecution(SESSION_ID); + gate.resolve(); + + assert.equal(await within(preceding), 'first'); + assert.equal( + await within(result.promise), + 'committed', + 'mutation must not wait for a claim requested after it', + ); + lateClaim.release(); + }); + + test('commit lands between goal-style turns without waiting for the successor claim', async () => { + const kernel = newKernel(); + const predecessor = kernel.claimExecution(SESSION_ID); + const result = track(kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => 'committed')); + // The successor turn's claim arrives while the mutation is queued: it is + // newer than the frontier, so the committed slot must not wait for it. + const successor = kernel.claimExecution(SESSION_ID); + + predecessor.release(); + assert.equal(await within(result.promise), 'committed'); + successor.release(); + }); + + test('waits across every session named by the mutation', async () => { + const kernel = newKernel(); + const first = kernel.claimExecution(SESSION_ID); + const second = kernel.claimExecution(OTHER_SESSION_ID); + const result = track( + kernel.runSessionQueuedQuiescentMutation([SESSION_ID, OTHER_SESSION_ID], () => 'committed'), + ); + + first.release(); + await settleTicks(); + assert.equal(result.settled, false, 'mutation must wait for the other session claim'); + + second.release(); + assert.equal(await within(result.promise), 'committed'); + }); + + test('serializes with other session mutations in request order', async () => { + const kernel = newKernel(); + const order: string[] = []; + await Promise.all([ + kernel.runSessionAdmissionMutation([SESSION_ID], async () => { + order.push('admission'); + }), + kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => { + order.push('queued'); + }), + ]); + assert.deepEqual(order, ['admission', 'queued']); + }); + + test('propagates an operation failure and releases the mutation tail', async () => { + const kernel = newKernel(); + await assert.rejects( + kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => { + throw new Error('commit rejected'); + }), + /commit rejected/, + ); + assert.equal( + await within(kernel.runSessionAdmissionMutation([SESSION_ID], () => 'next')), + 'next', + ); + }); + + test('propagates a failure that happens after the drain wait', async () => { + const kernel = newKernel(); + const claim = kernel.claimExecution(SESSION_ID); + const result = kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => { + throw new Error('commit rejected after drain'); + }); + claim.release(); + await assert.rejects(within(result), /commit rejected after drain/); + assert.equal( + await within(kernel.runSessionAdmissionMutation([SESSION_ID], () => 'next')), + 'next', + ); + }); + + test('keeps the eager quiescent mutation semantics unchanged', async () => { + const kernel = newKernel(); + const claim = kernel.claimExecution(SESSION_ID); + await assert.rejects( + kernel.runSessionQuiescentMutation([SESSION_ID], () => 'committed'), + SessionQuiescentMutationBusyError, + ); + claim.release(); + assert.equal( + await within(kernel.runSessionQuiescentMutation([SESSION_ID], () => 'committed')), + 'committed', + ); + }); + + test('waits for a running turn even after its admission claim has settled', async () => { + const gate = deferred(); + const backends = new BackendRegistry(); + backends.register('ai-sdk', (ctx) => new GatedBackend(ctx, gate.promise)); + let id = 0; + const kernel = new RuntimeKernel({ + store: memoryStore(), + backends, + newId: () => `queued-quiescent-id-${++id}`, + now: () => id, + }); + + // The turn is dispatched: its admission claim settles once the run is bound, + // but the run itself stays active on the backend generation while the gate + // holds the stream open. The mutation must wait for the run. + const iterator = kernel + .startTurn(SESSION_ID, { turnId: 'turn-gated', text: 'start' }) + [Symbol.asyncIterator](); + assert.equal((await iterator.next()).value?.type, 'text_delta'); + + const result = track(kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => 'committed')); + await settleTicks(); + assert.equal(result.settled, false, 'mutation must wait while the run is active'); + + gate.resolve(); + while (!(await iterator.next()).done) {} + assert.equal(await within(result.promise), 'committed'); + }); + + test('an admission mutation a running turn depends on passes while the queued mutation waits', async () => { + // Graph operator provisioning (#3349 review): a running supervisor turn's + // completion can depend on an admission mutation enqueued on its own + // session's tail. The queued mutation must therefore never hold the tail + // while it waits for quiescence — only the admission gate closes. + const gate = deferred(); + const backends = new BackendRegistry(); + backends.register('ai-sdk', (ctx) => new GatedBackend(ctx, gate.promise)); + let id = 0; + const kernel = new RuntimeKernel({ + store: memoryStore(), + backends, + newId: () => `queued-quiescent-id-${++id}`, + now: () => id, + }); + const iterator = kernel + .startTurn(SESSION_ID, { turnId: 'turn-provision', text: 'start' }) + [Symbol.asyncIterator](); + assert.equal((await iterator.next()).value?.type, 'text_delta'); + + const queued = track(kernel.runSessionQueuedQuiescentMutation([SESSION_ID], () => 'committed')); + await settleTicks(); + assert.equal(queued.settled, false); + + const admission = track(kernel.runSessionAdmissionMutation([SESSION_ID], () => 'provisioned')); + assert.equal(await within(admission.promise), 'provisioned'); + assert.equal(queued.settled, false, 'the queued mutation still waits for the run'); + + gate.resolve(); + while (!(await iterator.next()).done) {} + assert.equal(await within(queued.promise), 'committed'); + }); +}); + +function newKernel(): RuntimeKernel { + const store = memoryStore(); + let id = 0; + return new RuntimeKernel({ + store, + backends: new BackendRegistry(), + newId: () => `queued-quiescent-id-${++id}`, + now: () => id, + }); +} + +class GatedBackend implements AgentBackend { + readonly kind = 'ai-sdk' as const; + readonly sessionId: string; + + constructor( + ctx: BackendFactoryContext, + private readonly gate: Promise, + ) { + this.sessionId = ctx.sessionId; + } + + async *send(input: BackendSendInput): AsyncIterable { + yield { + type: 'text_delta', + id: `${input.turnId}-delta`, + turnId: input.turnId, + ts: 1, + messageId: `${input.turnId}-message`, + text: 'ok', + }; + await this.gate; + yield { + type: 'complete', + id: `${input.turnId}-complete`, + turnId: input.turnId, + ts: 2, + stopReason: 'end_turn', + }; + } + + async stop(): Promise {} + + async respondToSandboxBoundary(): Promise {} + + async dispose(): Promise {} +} + +function memoryStore(): SessionStore { + let header: SessionHeader = { + id: SESSION_ID, + workspaceRoot: '/tmp/maka-runtime-kernel-queued-quiescent', + cwd: '/tmp/maka-runtime-kernel-queued-quiescent', + createdAt: 1, + lastUsedAt: 1, + name: 'Queued quiescent mutation', + titleIsManual: true, + isFlagged: false, + labels: [], + isArchived: false, + status: 'active', + statusUpdatedAt: 1, + hasUnread: false, + backend: 'ai-sdk', + llmConnectionSlug: 'test', + connectionLocked: true, + model: 'test', + permissionMode: 'ask', + schemaVersion: 1, + }; + let messages: StoredMessage[] = []; + return { + create: async () => header, + createSubagent: async () => ({ header, created: false }), + setExecutionBoundaryKind: async () => { + throw new Error('not implemented'); + }, + readExecutionBoundary: async () => { + throw new Error('not implemented'); + }, + list: async () => [], + readHeader: async () => header, + readMessages: async () => [...messages], + listTurns: async () => [], + appendMessage: async (_sessionId, message) => { + messages.push(message); + }, + appendMessages: async (_sessionId, next) => { + messages.push(...next); + }, + updateHeader: async (_sessionId, patch) => { + header = { ...header, ...patch }; + return header; + }, + setFlagged: async () => {}, + rename: async () => {}, + remove: async () => {}, + }; +} + +function track(promise: Promise): { promise: Promise; settled: boolean } { + const state = { promise, settled: false }; + void promise.then( + () => { + state.settled = true; + }, + () => { + state.settled = true; + }, + ); + return state; +} + +function deferred(): { + promise: Promise; + resolve(value: T | PromiseLike): void; +} { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +async function settleTicks(): Promise { + for (let tick = 0; tick < 2; tick += 1) { + await new Promise((resolve) => setImmediate(resolve)); + } +} + +async function within(promise: Promise, timeoutMs = 1_000): Promise { + let timeout: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timeout = setTimeout( + () => reject(new Error('queued quiescent mutation timed out')), + timeoutMs, + ); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +} diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index f9fa2e36a6..d017f51fc2 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -24,6 +24,7 @@ import { createHash } from 'node:crypto'; import { applySandboxBoundaryExpansion, createGenesisExecutionBoundary, + executionBoundaryDisplayMode, isSandboxBoundaryRestartClosure, } from '@maka/core/sandbox-boundary'; import { createWorkspaceWritePermissionProfile } from '@maka/core/permission-profile'; @@ -91,6 +92,7 @@ import { type SessionStore, type VersionedSessionHeader, } from '../session-manager.js'; +import { ToolRuntime, type ToolRuntimeInput } from '../tool-runtime.js'; import { RuntimeContextCompactError, RuntimeKernel, @@ -4347,7 +4349,7 @@ describe('SessionManager manual compaction and quiescent session changes', () => await sendPromise; }); - test('configuration transitions fence active runs and unavailable resource side effects', async () => { + test('configuration transitions queue behind held claims and fence resource side effects', async () => { const store = new VersionedConfigurationMemorySessionStore(); const kernel = new DelegatingRuntimeKernel(); const manager = new SessionManager({ @@ -4371,25 +4373,23 @@ describe('SessionManager manual compaction and quiescent session changes', () => orchestrationMode: 'graph' as const, }; - kernel.activeRuns = true; - await assert.rejects( - manager.transitionSessionConfiguration(session.id, { + const heldClaim = kernel.claimExecution(session.id); + let transitionSettled = false; + const queuedTransition = manager + .transitionSessionConfiguration(session.id, { expectedRevision: 1, configuration: baseConfiguration, - }), - (error: unknown) => { - assert.ok(error instanceof SessionConfigurationTransitionError); - assert.equal(error.code, 'session_busy'); - return true; - }, - ); + }) + .then((value) => { + transitionSettled = true; + return value; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(transitionSettled, false); assert.deepEqual(kernel.disposed, []); - kernel.activeRuns = false; - const committed = await manager.transitionSessionConfiguration(session.id, { - expectedRevision: 1, - configuration: baseConfiguration, - }); + heldClaim.release(); + const committed = await queuedTransition; assert.equal(committed.revision, 2); assert.equal(committed.header.orchestrationMode, 'graph'); assert.deepEqual(kernel.disposed, [session.id]); @@ -4478,7 +4478,7 @@ describe('SessionManager manual compaction and quiescent session changes', () => assert.equal((await store.readHeader(session.id)).cwd, '/workspace/new'); }); - test('configuration transitions reject a claimed turn without waiting for it to settle', async () => { + test('configuration transitions wait for a claimed turn to settle before committing', async () => { const store = new VersionedConfigurationMemorySessionStore(); const readStarted = makeGate(); const releaseRead = makeGate(); @@ -4495,7 +4495,8 @@ describe('SessionManager manual compaction and quiescent session changes', () => const turn = drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'start' })); await readStarted.promise; - const transitionResult = await manager + let transitionSettled = false; + const transition = manager .transitionSessionConfiguration(session.id, { expectedRevision: 1, configuration: { @@ -4509,18 +4510,20 @@ describe('SessionManager manual compaction and quiescent session changes', () => orchestrationMode: 'graph', }, }) - .then( - (value) => ({ ok: true as const, value }), - (error: unknown) => ({ ok: false as const, error }), - ); - assert.equal(transitionResult.ok, false); - if (transitionResult.ok) assert.fail('Configuration transition unexpectedly committed'); - assert.ok(transitionResult.error instanceof SessionConfigurationTransitionError); - assert.equal(transitionResult.error.code, 'session_busy'); + .then((value) => { + transitionSettled = true; + return value; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(transitionSettled, false); assert.equal((await store.readHeader(session.id)).orchestrationMode, 'default'); releaseRead.release(); await turn; + const committed = await transition; + assert.equal(committed.revision, 2); + assert.equal(committed.header.orchestrationMode, 'graph'); + assert.equal((await store.readHeader(session.id)).orchestrationMode, 'graph'); }); test('a claimed turn waits for an in-flight session mutation before reading its header', async () => { @@ -5068,7 +5071,7 @@ describe('SessionManager permission mode updates', () => { expect(store.disposeCount).toBe(3); }); - test('keeps mode changes blocked until all overlapping turns finish', async () => { + test('queues mode changes until all overlapping turns finish', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); const backends = new BackendRegistry(); @@ -5103,11 +5106,18 @@ describe('SessionManager permission mode updates', () => { expect(afterFirstRuns.find((run) => run.turnId === 'turn-1')?.status).toBe('completed'); expect(afterFirstRuns.find((run) => run.turnId === 'turn-2')?.status).toBe('running'); - await expectRejects(manager.setPermissionMode(session.id, 'bypass'), /当前任务正在运行/); + let modeChangeSettled = false; + const modeChange = manager.setPermissionMode(session.id, 'bypass').then((result) => { + modeChangeSettled = true; + return result; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(modeChangeSettled).toBe(false); secondGate.release(); await second.next(); await second.next(); + while (!(await second.next()).done) {} expect((await store.readHeader(session.id)).status).toBe('active'); const finalRuns = await runStore.listSessionRuns(session.id); expect(finalRuns.map((run) => [run.turnId, run.status])).toEqual([ @@ -5119,8 +5129,467 @@ describe('SessionManager permission mode updates', () => { expect(firstEvents.map((event) => event.type)).toContain('run_started'); expect(firstEvents.map((event) => event.type)).toContain('run_completed'); + const summary = await modeChange; + expect(summary.permissionMode).toBe('bypass'); + }); + + test('a permission switch requested mid-turn lands before the next turn starts', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const gates: Gate[] = []; + const builtPermissionModes: SessionHeader['permissionMode'][] = []; + backends.register('ai-sdk', (ctx) => { + const gate = makeGate(); + gates.push(gate); + builtPermissionModes.push(ctx.header.permissionMode); + return new TestBackend(ctx, gate); + }); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(8_000), + }); + const session = await manager.createSession(makeInput({ permissionMode: 'ask' })); + + // Turn 1 runs under ask; the gate holds the backend so the claim stays attached. + const first = manager + .sendMessage(session.id, { turnId: 'turn-1', text: 'first' }) + [Symbol.asyncIterator](); + expect((await first.next()).value?.type).toBe('text_delta'); + + // Auto → Bypass requested mid-turn: the switch queues instead of rejecting. + let switchSettled = false; + const switchPromise = manager.setPermissionMode(session.id, 'bypass').then( + (result) => { + switchSettled = true; + return result; + }, + (error) => { + switchSettled = true; + throw error; + }, + ); + await new Promise((resolve) => setImmediate(resolve)); + expect(switchSettled).toBe(false); + expect((await store.readExecutionBoundary(session.id)).kind).toBe('managed'); + + // The switch commits in the gap as soon as turn 1 settles. + gates[0]!.release(); + while (!(await first.next()).done) {} + const summary = await switchPromise; + expect(summary.permissionMode).toBe('bypass'); + expect((await store.readExecutionBoundary(session.id)).kind).toBe('bypass'); + + // The next turn is rebuilt from the committed configuration. + const second = manager + .sendMessage(session.id, { turnId: 'turn-2', text: 'second' }) + [Symbol.asyncIterator](); + expect((await second.next()).value?.type).toBe('text_delta'); + gates[1]!.release(); + while (!(await second.next()).done) {} + expect(builtPermissionModes).toEqual(['ask', 'bypass']); + }); + + test('a boundary revision bump without backend disposal rebuilds on the next activation', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + let builds = 0; + backends.register('ai-sdk', (ctx) => { + builds += 1; + return new TestBackend(ctx); + }); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(8_000), + }); + const session = await manager.createSession(makeInput({ permissionMode: 'ask' })); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'one' })); + expect(builds).toBe(1); + + // A write path that skips backend disposal bumps the durable boundary + // while the generation stays alive. + store.forceExecutionBoundary(session.id, { kind: 'bypass', revision: 5 }); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-2', text: 'two' })); + expect(builds).toBe(2); + expect(store.disposeCount).toBe(1); + + // Once rebuilt against the current revision, the generation is reused again. + await drain(manager.sendMessage(session.id, { turnId: 'turn-3', text: 'three' })); + expect(builds).toBe(2); + }); + + test('a stale generation with live runs flushes after they exit instead of disposing underneath them', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const gates: Gate[] = []; + let builds = 0; + backends.register('ai-sdk', (ctx) => { + builds += 1; + const gate = makeGate(); + gates.push(gate); + return new TestBackend(ctx, gate); + }); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(8_000), + }); + const session = await manager.createSession(makeInput({ permissionMode: 'ask' })); + + const first = manager + .sendMessage(session.id, { turnId: 'turn-1', text: 'one' }) + [Symbol.asyncIterator](); + expect((await first.next()).value?.type).toBe('text_delta'); + expect(builds).toBe(1); + + store.forceExecutionBoundary(session.id, { kind: 'bypass', revision: 5 }); + + // An overlapping activation while turn 1 is live must not dispose the + // generation underneath it: the generation is marked, reused for this + // turn, and flushed once both runs exit. Both turns share the reused + // backend, so a single gate holds them both. + const second = manager + .sendMessage(session.id, { turnId: 'turn-2', text: 'two' }) + [Symbol.asyncIterator](); + expect((await second.next()).value?.type).toBe('text_delta'); + expect(builds).toBe(1); + + gates[0]!.release(); + while (!(await first.next()).done) {} + while (!(await second.next()).done) {} + + const third = manager + .sendMessage(session.id, { turnId: 'turn-3', text: 'three' }) + [Symbol.asyncIterator](); + expect((await third.next()).value?.type).toBe('text_delta'); + gates[1]!.release(); + while (!(await third.next()).done) {} + expect(builds).toBe(2); + }); + + test('narrowing with an active descendant commits promptly and fences the lineage shells', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const childGate = makeGate(); + backends.register('ai-sdk', (ctx) => new TestBackend(ctx, childGate)); + const terminated: string[] = []; + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + // Narrowing fences shell runs through this authority. + shellRuns: { + async terminateSession(sessionId: string) { + terminated.push(sessionId); + return undefined; + }, + async commitSessionClose() {}, + rollbackSessionClose() {}, + resumeSession() {}, + } as never, + newId: nextId(), + now: nextNow(8_000), + }); + const parent = await manager.createSession(makeInput({ permissionMode: 'bypass' })); + const child = await manager.createSession( + makeInput({ + permissionMode: 'ask', + subagentParent: { + kind: 'subagent', + parentSessionId: parent.id, + spawnedBy: { + parentRunId: 'parent-run', + parentTurnId: 'parent-turn', + toolCallId: 'parent-tool', + }, + lifecycle: 'foreground', + }, + subagentRuntime: { + schemaVersion: 1, + definitionVersion: 1, + agentId: 'descendant-agent', + agentName: 'descendant-agent', + profile: 'default', + systemPrompt: '', + toolNames: [], + categoryPolicy: {}, + }, + }), + ); + + // The child session runs a gated turn; narrowing the parent revokes now — + // it does not wait out the child's turn (whose supervisor chain could + // depend on it) and does not need to: the boundary write is immediate. + const childTurn = manager + .sendMessage(child.id, { turnId: 'child-turn', text: 'work' }) + [Symbol.asyncIterator](); + expect((await childTurn.next()).value?.type).toBe('text_delta'); + + const summary = await manager.setPermissionMode(parent.id, 'ask'); + expect(summary.permissionMode).toBe('ask'); + expect(terminated).toEqual([parent.id, child.id]); + + // The live child turn is not stopped; it completes normally. + childGate.release(); + while (!(await childTurn.next()).done) {} + }); + + test('a mid-turn Bypass→Ask narrowing reaches the next dispatch, not the next turn', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const gate = makeGate(); + const composedModes: SessionHeader['permissionMode'][] = []; + const terminated: string[] = []; + backends.register('ai-sdk', (ctx) => { + composedModes.push(ctx.header.permissionMode); + return new TestBackend(ctx, gate); + }); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + shellRuns: { + async terminateSession(sessionId: string) { + terminated.push(sessionId); + return undefined; + }, + async commitSessionClose() {}, + rollbackSessionClose() {}, + resumeSession() {}, + } as never, + newId: nextId(), + now: nextNow(8_000), + }); + const session = await manager.createSession(makeInput({ permissionMode: 'bypass' })); + + // The turn is live under bypass when the user tightens to Auto. + const turn = manager + .sendMessage(session.id, { turnId: 'turn-1', text: 'work' }) + [Symbol.asyncIterator](); + expect((await turn.next()).value?.type).toBe('text_delta'); + + // Revocation is prompt: it resolves without waiting for the turn to end. + const summary = await manager.setPermissionMode(session.id, 'ask'); + expect(summary.permissionMode).toBe('ask'); + expect(terminated).toEqual([session.id]); + + // Before the turn completes, its next write-capable dispatch reads the + // narrower authority through the same store closure a real dispatch uses. + const dispatch = await dispatchProbeTool(store, session.id); + expect(dispatch.boundaryKind).toBe('managed'); + expect(dispatch.permissionMode).toBe('ask'); + + // The live turn is not stopped; it finishes, and its generation rebuilds + // from the committed configuration on the next activation (a separate + // invariant from the dispatch-level revocation above). + gate.release(); + while (!(await turn.next()).done) {} + const successor = manager + .sendMessage(session.id, { turnId: 'turn-2', text: 'next' }) + [Symbol.asyncIterator](); + expect((await successor.next()).value?.type).toBe('text_delta'); + gate.release(); + while (!(await successor.next()).done) {} + expect(composedModes).toEqual(['bypass', 'ask']); + }); + + test('a switch queued behind a turn that pauses on an interaction rejects busy', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const requestGate = makeGate(); + const responseGate = makeGate(); + backends.register( + 'ai-sdk', + (ctx) => new InteractionPauseBackend(ctx, requestGate, responseGate), + ); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(8_000), + }); + const session = await manager.createSession(makeInput({ permissionMode: 'ask' })); + + const turn = manager + .sendMessage(session.id, { turnId: 'turn-1', text: 'work' }) + [Symbol.asyncIterator](); + expect((await turn.next()).value?.type).toBe('text_delta'); + + let switchSettled = false; + const switching = manager.setPermissionMode(session.id, 'bypass').then( + (result) => { + switchSettled = true; + return result; + }, + (error) => { + switchSettled = true; + throw error; + }, + ); + await new Promise((resolve) => setImmediate(resolve)); + expect(switchSettled).toBe(false); + + // The turn pauses on a sandbox boundary approval: quiescence now depends + // on the user answering, so the queued switch must reject, not hang. + // Events are pull-driven, so the request only registers once observed. + requestGate.release(); + let sawRequest = false; + while (!sawRequest) { + const next = await turn.next(); + if (next.done) break; + sawRequest = next.value?.type === 'sandbox_boundary_request'; + } + expect(sawRequest).toBe(true); + expect((await manager.listActiveInteractions(session.id)).length).toBe(1); + await expectRejects(switching, /pending Interaction/); + + await manager.respondToSandboxBoundary(session.id, { + requestId: 'boundary-1', + decision: 'deny', + }); + while (!(await turn.next()).done) {} + }); + + test('an idle Auto→Bypass switch is observed by the next turn and its first tool dispatch', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const composedModes: SessionHeader['permissionMode'][] = []; + backends.register('ai-sdk', (ctx) => { + composedModes.push(ctx.header.permissionMode); + return new TestBackend(ctx); + }); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(8_000), + }); + const session = await manager.createSession(makeInput({ permissionMode: 'ask' })); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'first' })); + + // The plain-session case from the issue: switch between two ordinary + // turns, no Goal, session idle. const summary = await manager.setPermissionMode(session.id, 'bypass'); expect(summary.permissionMode).toBe('bypass'); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-2', text: 'second' })); + expect(composedModes).toEqual(['ask', 'bypass']); + + // The reporter's dual assertion, at the layer a real dispatch reads: a + // tool call resolving this session's boundary — through the same store + // the switch committed to — must see both facts at once. + const dispatch = await dispatchProbeTool(store, session.id); + expect(dispatch.boundaryKind).toBe('bypass'); + expect(dispatch.permissionMode).toBe('bypass'); + }); + + test('seeded switch/turn interleavings always observe the committed mode', async () => { + // A fixed-seed PRNG picks the interleaving class per iteration; every + // checkpoint awaits a deterministic event, so the sweep is reproducible. + let seed = 0x3349; + const random = (): number => { + seed = (seed * 1_103_515_245 + 12_345) % 2_147_483_648; + return seed / 2_147_483_648; + }; + + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const gates: Gate[] = []; + const composedModes: SessionHeader['permissionMode'][] = []; + backends.register('ai-sdk', (ctx) => { + const gate = makeGate(); + gates.push(gate); + composedModes.push(ctx.header.permissionMode); + return new TestBackend(ctx, gate); + }); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + // Narrowing (bypass → ask) fences shell runs through this authority. + shellRuns: { + async terminateSession() { + return undefined; + }, + async commitSessionClose() {}, + rollbackSessionClose() {}, + resumeSession() {}, + } as never, + newId: nextId(), + now: nextNow(9_000), + }); + const session = await manager.createSession(makeInput({ permissionMode: 'ask' })); + + let expected: 'ask' | 'bypass' = 'ask'; + let turnCount = 0; + for (let iteration = 0; iteration < 100; iteration += 1) { + const nextMode: 'ask' | 'bypass' = random() < 0.5 ? 'bypass' : 'ask'; + const interleaving = Math.floor(random() * 3); + + if (interleaving === 0) { + // Switch while the session is idle. + await manager.setPermissionMode(session.id, nextMode); + expected = nextMode; + } else { + // Switch requested while a turn is running; the queued commit lands + // in the gap as the turn settles (class 1 requests it mid-flight, + // class 2 races it with the gate release). + turnCount += 1; + const turn = manager + .sendMessage(session.id, { turnId: `turn-${turnCount}`, text: `t${turnCount}` }) + [Symbol.asyncIterator](); + expect((await turn.next()).value?.type).toBe('text_delta'); + const switching = manager.setPermissionMode(session.id, nextMode); + if (interleaving === 2) gates[gates.length - 1]!.release(); + if (interleaving === 1) gates[gates.length - 1]!.release(); + while (!(await turn.next()).done) {} + await switching; + expected = nextMode; + } + + // Invariant: every turn started after the switch resolved is composed + // from the committed mode, and a tool call against the committed store + // derives the same mode. + turnCount += 1; + const verify = manager + .sendMessage(session.id, { turnId: `turn-${turnCount}`, text: `t${turnCount}` }) + [Symbol.asyncIterator](); + expect((await verify.next()).value?.type).toBe('text_delta'); + expect(composedModes[composedModes.length - 1]).toBe(expected); + const dispatch = await dispatchProbeTool(store, session.id); + expect(dispatch.boundaryKind).toBe(expected === 'bypass' ? 'bypass' : 'managed'); + expect(dispatch.permissionMode).toBe(expected); + gates[gates.length - 1]!.release(); + while (!(await verify.next()).done) {} + } }); test('leaving explore clears the deep research label so visible read-only copy stays truthful', async () => { @@ -12224,7 +12693,10 @@ describe('SessionManager permission mode updates', () => { expect((await store.readHeader(session.id)).status).toBe('waiting_for_user'); const [run] = await runStore.listSessionRuns(session.id); expect(run?.status).toBe('waiting_for_user'); - await expectRejects(manager.setPermissionMode(session.id, 'bypass'), /当前任务正在运行/); + await expectRejects( + manager.setPermissionMode(session.id, 'bypass'), + /当前有工具调用正在等待确认/, + ); expect((await store.readHeader(session.id)).permissionMode).toBe('ask'); await manager.respondToSandboxBoundary(session.id, { @@ -15864,13 +16336,20 @@ class DelegatingRuntimeKernel implements RuntimeKernelLike { constructor(private readonly events: readonly SessionEvent[] = []) {} + private readonly heldClaims = new Map(); + claimExecution(sessionId: string): ReturnType { + this.heldClaims.set(sessionId, (this.heldClaims.get(sessionId) ?? 0) + 1); const stopController = new AbortController(); return { sessionId, stopSignal: stopController.signal, isStopRequested: () => false, - release: () => {}, + release: () => { + const remaining = (this.heldClaims.get(sessionId) ?? 0) - 1; + if (remaining > 0) this.heldClaims.set(sessionId, remaining); + else this.heldClaims.delete(sessionId); + }, }; } @@ -15888,6 +16367,16 @@ class DelegatingRuntimeKernel implements RuntimeKernelLike { return operation(); } + async runSessionQueuedQuiescentMutation( + sessionIds: readonly string[], + operation: () => Promise | T, + ): Promise { + while (sessionIds.some((sessionId) => (this.heldClaims.get(sessionId) ?? 0) > 0)) { + await new Promise((resolve) => setImmediate(resolve)); + } + return operation(); + } + async *startTurn( sessionId: string, input: Parameters[1], @@ -16508,6 +16997,75 @@ class EventBackend implements AgentBackend { async dispose(): Promise {} } +class InteractionPauseBackend implements AgentBackend { + readonly kind = 'ai-sdk' as const; + readonly sessionId: string; + readonly responses: SandboxBoundaryResponse[] = []; + + constructor( + ctx: BackendFactoryContext, + private readonly requestGate: Gate, + private readonly responseGate: Gate, + ) { + this.sessionId = ctx.sessionId; + } + + async *send(input: BackendSendInput): AsyncIterable { + yield { + type: 'text_delta', + id: `${input.turnId}-delta`, + turnId: input.turnId, + ts: 1, + messageId: `${input.turnId}-message`, + text: 'ok', + }; + await this.requestGate.promise; + yield { + type: 'sandbox_boundary_request', + id: `${input.turnId}-request`, + turnId: input.turnId, + ts: 2, + requestId: 'boundary-1', + toolUseId: 'tool-1', + justification: 'Write the requested export.', + expansion: { + filesystem: { + entries: [{ path: '/tmp/export.txt', access: 'write', scope: 'exact' }], + }, + }, + }; + await this.responseGate.promise; + const response = this.responses[0]!; + yield { + type: 'sandbox_boundary_decision_ack', + id: `${input.turnId}-decision`, + turnId: input.turnId, + ts: 3, + requestId: response.requestId, + toolUseId: 'tool-1', + decision: response.decision, + status: response.decision === 'allow' ? 'approved' : 'denied', + revision: response.decision === 'allow' ? 1 : 0, + }; + yield { + type: 'complete', + id: `${input.turnId}-complete`, + turnId: input.turnId, + ts: 4, + stopReason: 'end_turn', + }; + } + + async stop(): Promise {} + + async respondToSandboxBoundary(response: SandboxBoundaryResponse): Promise { + this.responses.push(response); + this.responseGate.release(); + } + + async dispose(): Promise {} +} + class SandboxBoundaryWaitBackend implements AgentBackend { readonly kind = 'ai-sdk' as const; readonly sessionId: string; @@ -17526,6 +18084,11 @@ class MemorySessionStore implements SessionStore { return boundary; } + /** Simulates a config write path that bumps the boundary without disposing backends. */ + forceExecutionBoundary(sessionId: string, boundary: ExecutionBoundary): void { + this.executionBoundaries.set(sessionId, boundary); + } + async createSandboxBoundaryRequest( input: CreateSandboxBoundaryRequest, ): Promise { @@ -19142,6 +19705,59 @@ async function drain(iterable: AsyncIterable): Promise { } } +/** + * Dispatches one probe tool call whose boundary resolves through the given + * session's own store — the same read a real tool dispatch performs — and + * reports both facts a tool acts on: the boundary kind and the derived + * permission mode. + */ +async function dispatchProbeTool( + store: SessionStore, + sessionId: string, +): Promise<{ boundaryKind: ExecutionBoundary['kind']; permissionMode: string | undefined }> { + const observed: Array<{ + kind: ExecutionBoundary['kind'] | undefined; + mode: string | undefined; + }> = []; + const runtime = new ToolRuntime({ + turnId: 'probe-turn', + sessionId, + header: await store.readHeader(sessionId), + connection: { providerType: 'openai', slug: 'probe' } as never, + modelId: 'probe', + appendMessage: async () => {}, + readExecutionBoundary: () => store.readExecutionBoundary(sessionId), + newId: nextId(), + now: () => 1, + getPermissionPauseTarget: () => null, + } as unknown as ToolRuntimeInput); + const tool: MakaTool = { + name: 'Read', + description: 'probe', + parameters: {}, + impl: (_args, context) => { + observed.push({ kind: context.executionBoundary?.kind, mode: context.permissionMode }); + return { ok: true }; + }, + }; + const events: SessionEvent[] = []; + await runtime.settleToolCall({ + tool, + turnId: 'probe-turn', + toolCallId: 'probe-tool', + input: {}, + abortSignal: new AbortController().signal, + eventSink: { + push: (event) => events.push(event), + pushAndWaitUntilConsumed: async (event) => { + events.push(event); + }, + }, + }); + const sample = observed[0]!; + return { boundaryKind: sample.kind as ExecutionBoundary['kind'], permissionMode: sample.mode }; +} + async function collectSessionEvents( iterable: AsyncIterable, ): Promise { diff --git a/packages/runtime/src/__tests__/tool-runtime-permission-mode.test.ts b/packages/runtime/src/__tests__/tool-runtime-permission-mode.test.ts new file mode 100644 index 0000000000..a95b378682 --- /dev/null +++ b/packages/runtime/src/__tests__/tool-runtime-permission-mode.test.ts @@ -0,0 +1,213 @@ +/* + * 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 assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; + +import type { ExecutionBoundary } from '@maka/core/sandbox-boundary'; +import { + createReadOnlyPermissionProfile, + createWorkspaceWritePermissionProfile, +} from '@maka/core/permission-profile'; +import type { SessionEvent } from '@maka/core/events'; +import type { SessionHeader } from '@maka/core/session'; + +import { ToolRuntime, type MakaTool, type ToolRuntimeInput } from '../tool-runtime.js'; + +describe('ToolRuntime permission mode derivation', () => { + test('derives the tool permission mode from the live boundary on every dispatch', async () => { + const writable: ExecutionBoundary = { + kind: 'managed', + profile: createWorkspaceWritePermissionProfile(), + revision: 0, + }; + const bypass: ExecutionBoundary = { kind: 'bypass', revision: 1 }; + const readOnly: ExecutionBoundary = { + kind: 'managed', + profile: createReadOnlyPermissionProfile(), + revision: 2, + }; + // A committed switch flips the durable boundary between two dispatches of + // the same turn; the very next tool call must observe it in both facts. + let boundary: ExecutionBoundary = writable; + const observed: Array> = []; + const runtime = new ToolRuntime({ + turnId: 'turn-1', + sessionId: 'session-1', + header: header(), + connection: { providerType: 'openai', slug: 'test' } as never, + modelId: 'test', + appendMessage: async () => {}, + readExecutionBoundary: async () => boundary, + newId: nextId(), + now: () => 1, + getPermissionPauseTarget: () => null, + } as unknown as ToolRuntimeInput); + const tool: MakaTool = { + name: 'Read', + description: 'test', + parameters: {}, + impl: (_args, context) => { + observed.push({ + executionBoundary: context.executionBoundary, + permissionMode: context.permissionMode, + }); + return { ok: true }; + }, + }; + + await settle(runtime, tool, 'tool-1'); + boundary = bypass; + await settle(runtime, tool, 'tool-2'); + boundary = readOnly; + await settle(runtime, tool, 'tool-3'); + + assert.deepEqual( + observed.map((sample) => [sample.executionBoundary?.kind, sample.permissionMode]), + [ + ['managed', 'ask'], + ['bypass', 'bypass'], + ['managed', 'explore'], + ], + ); + }); + + test('a plan-mode session presents read-only authority at dispatch time unless bypassed', async () => { + let boundary: ExecutionBoundary = { + kind: 'managed', + profile: createWorkspaceWritePermissionProfile(), + revision: 0, + }; + const observed: (string | undefined)[] = []; + const runtime = new ToolRuntime({ + turnId: 'turn-1', + sessionId: 'session-1', + header: header({ collaborationMode: 'plan' }), + connection: { providerType: 'openai', slug: 'test' } as never, + modelId: 'test', + appendMessage: async () => {}, + readExecutionBoundary: async () => boundary, + newId: nextId(), + now: () => 1, + getPermissionPauseTarget: () => null, + } as unknown as ToolRuntimeInput); + const tool: MakaTool = { + name: 'Read', + description: 'test', + parameters: {}, + impl: (_args, context) => { + observed.push(context.permissionMode); + return { ok: true }; + }, + }; + + await settle(runtime, tool, 'tool-1'); + boundary = { kind: 'bypass', revision: 1 }; + await settle(runtime, tool, 'tool-2'); + + assert.deepEqual(observed, ['explore', 'bypass']); + }); + + test('an external boundary falls back to the last known header mode', async () => { + const observed: (string | undefined)[] = []; + const runtime = new ToolRuntime({ + turnId: 'turn-1', + sessionId: 'session-1', + header: header({ permissionMode: 'ask' }), + connection: { providerType: 'openai', slug: 'test' } as never, + modelId: 'test', + appendMessage: async () => {}, + readExecutionBoundary: async () => ({ kind: 'external', revision: 0 }), + newId: nextId(), + now: () => 1, + getPermissionPauseTarget: () => null, + } as unknown as ToolRuntimeInput); + const tool: MakaTool = { + name: 'Read', + description: 'test', + parameters: {}, + impl: (_args, context) => { + observed.push(context.permissionMode); + return { ok: true }; + }, + }; + + await settle(runtime, tool, 'tool-1'); + + assert.deepEqual(observed, ['ask']); + }); +}); + +interface MakaToolContextShape { + executionBoundary: ExecutionBoundary | undefined; + permissionMode: string | undefined; +} + +function header( + overrides: { + permissionMode?: SessionHeader['permissionMode']; + collaborationMode?: SessionHeader['collaborationMode']; + } = {}, +): SessionHeader { + const cwd = process.cwd(); + return { + id: 'session-1', + workspaceRoot: cwd, + cwd, + createdAt: 1, + lastUsedAt: 1, + name: 'test', + titleIsManual: false, + isFlagged: false, + labels: [], + isArchived: false, + status: 'active', + statusUpdatedAt: 1, + hasUnread: false, + backend: 'ai-sdk', + llmConnectionSlug: 'test', + connectionLocked: true, + model: 'test', + permissionMode: overrides.permissionMode ?? 'ask', + ...(overrides.collaborationMode ? { collaborationMode: overrides.collaborationMode } : {}), + schemaVersion: 1, + }; +} + +function nextId(): () => string { + let value = 0; + return () => `id-${++value}`; +} + +async function settle(runtime: ToolRuntime, tool: MakaTool, toolCallId: string): Promise { + const events: SessionEvent[] = []; + await runtime.settleToolCall({ + tool, + turnId: 'turn-1', + toolCallId, + input: {}, + abortSignal: new AbortController().signal, + eventSink: { + push: (event) => events.push(event), + pushAndWaitUntilConsumed: async (event) => { + events.push(event); + }, + }, + }); +} diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts index 14738c8719..c28560e58c 100644 --- a/packages/runtime/src/runtime-kernel.ts +++ b/packages/runtime/src/runtime-kernel.ts @@ -143,6 +143,10 @@ export interface RuntimeKernelLike { sessionIds: readonly string[], operation: () => Promise | T, ): Promise; + runSessionQueuedQuiescentMutation?( + sessionIds: readonly string[], + operation: () => Promise | T, + ): Promise; startTurn( sessionId: string, input: UserMessageInput, @@ -347,6 +351,13 @@ interface BackendGeneration extends AgentRunActiveSession { | { kind: 'failed'; error: unknown }; disposal?: Promise; disposalFailure?: Error; + /** + * The durable boundary revision this generation was composed against. + * `ensureActive` compares it against the store to rebuild when a config + * write skipped backend disposal; `undefined` (unreadable at build) keeps + * the guard dormant for this generation. + */ + boundaryRevision?: number; cachedHeader: SessionHeader; activeRuns: Map; turnToRunId: Map; @@ -398,6 +409,12 @@ type ExecutionClaimOutcome = { ok: true } | { ok: false; error: unknown }; interface PendingExecutionClaim { readonly handle: RuntimeExecutionClaim; readonly sessionId: string; + /** + * Monotonic creation order. A queued quiescent mutation only waits out claims + * that predate its tail reservation (`claimSeq <= frontier`); anything newer + * carries the reservation in its own admission barrier and cannot attach first. + */ + readonly claimSeq: number; readonly abortController: AbortController; readonly cancellation: RuntimeExecutionCancellation; readonly admissionBarrier: Promise; @@ -435,6 +452,12 @@ export class RuntimeKernel implements RuntimeKernelLike { private readonly stopAttempts = new Map>(); private readonly executionClaims = new Map>(); private readonly sessionMutationTails = new Map>(); + private claimSequence = 0; + private readonly sessionQuiescenceWaiters = new Map void>>(); + private readonly sessionAdmissionGates = new Map< + string, + Set<{ promise: Promise; open: () => void }> + >(); private readonly executionClaimStates = new WeakMap< RuntimeExecutionClaim, PendingExecutionClaim @@ -483,9 +506,10 @@ export class RuntimeKernel implements RuntimeKernelLike { const state: PendingExecutionClaim = { handle, sessionId, + claimSeq: ++this.claimSequence, abortController, cancellation, - admissionBarrier: this.sessionMutationTails.get(sessionId) ?? Promise.resolve(), + admissionBarrier: this.admissionBarrierFor(sessionId), settled, resolveSettled, rejectSettled, @@ -520,6 +544,78 @@ export class RuntimeKernel implements RuntimeKernelLike { return this.enqueueSessionMutation(ids, operation); } + /** + * A quiescent mutation that queues instead of bailing: the session's + * admission gate closes right away — claims created from then on capture it + * in their admission barrier — and the operation runs only once the session + * is at rest: every execution claim that already existed when the mutation + * was requested has settled, and no run is active. Unlike + * `runSessionQuiescentMutation`, a busy session delays the operation — by at + * most one turn's lifetime — instead of rejecting it. + * + * Claims only cover admission: a turn's claim settles once its run is bound + * (see `bindInteraction`), so live turns are visible through `hasActiveRuns` + * instead. Admission stays continuously observable because a run registers on + * its backend generation (reserve step) before its claim settles, so there is + * no instant where an in-flight admission is invisible to both checks. + * + * Deadlock freedom rests on the mutation tail NOT being held while + * quiescence is awaited: a running turn may legitimately depend on an + * admission mutation enqueued on its own session's tail (graph operator + * provisioning, #3349 review), so gating happens through the admission gate + * — which claims observe — while mutations pass freely. The operation joins + * the tail only after quiescence, still serialized with other mutations. + * Claims created after the request wait at the gate, so they cannot attach + * before the operation has run, and waiting chains run strictly backwards in + * claim-creation order, so no cycle can close. A session that pauses on an + * interaction while the mutation waits is rejected busy instead: quiescence + * would otherwise depend on the user answering. + * + * The claim frontier must be captured in the same synchronous block as the + * gate closing: that keeps claim-creation order and gate order identical, + * which is what makes "strictly older" meaningful. + */ + async runSessionQueuedQuiescentMutation( + sessionIds: readonly string[], + operation: () => Promise | T, + ): Promise { + const ids = this.normalizeSessionMutationIds(sessionIds); + const claimFrontier = this.claimSequence; + const openGates = ids.map((sessionId) => this.closeAdmissionGate(sessionId)); + try { + await this.waitForSessionQuiescence(ids, claimFrontier); + return await this.enqueueSessionMutation(ids, operation); + } finally { + for (const openGate of openGates) openGate(); + } + } + + private closeAdmissionGate(sessionId: string): () => void { + let open!: () => void; + const promise = new Promise((resolve) => { + open = resolve; + }); + const gate = { promise, open }; + let gates = this.sessionAdmissionGates.get(sessionId); + if (!gates) { + gates = new Set(); + this.sessionAdmissionGates.set(sessionId, gates); + } + gates.add(gate); + return () => { + gates.delete(gate); + if (gates.size === 0) this.sessionAdmissionGates.delete(sessionId); + open(); + }; + } + + private admissionBarrierFor(sessionId: string): Promise { + const tail = this.sessionMutationTails.get(sessionId) ?? Promise.resolve(); + const gates = this.sessionAdmissionGates.get(sessionId); + if (!gates || gates.size === 0) return tail; + return Promise.all([tail, ...[...gates].map((gate) => gate.promise)]).then(() => undefined); + } + private normalizeSessionMutationIds(sessionIds: readonly string[]): string[] { const ids = [...new Set(sessionIds)].sort(); if (ids.length === 0 || ids.some((sessionId) => sessionId.length === 0)) { @@ -558,6 +654,70 @@ export class RuntimeKernel implements RuntimeKernelLike { } } + private hasUnsettledExecutionClaims(sessionId: string, claimFrontier: number): boolean { + for (const claim of this.executionClaims.get(sessionId) ?? []) { + if (claim.claimSeq <= claimFrontier) return true; + } + return false; + } + + private isSessionExecuting(sessionId: string, claimFrontier: number): boolean { + return ( + this.hasUnsettledExecutionClaims(sessionId, claimFrontier) || this.hasActiveRuns(sessionId) + ); + } + + private async waitForSessionQuiescence( + sessionIds: readonly string[], + claimFrontier: number, + ): Promise { + for (;;) { + const blocking = sessionIds.filter((sessionId) => + this.isSessionExecuting(sessionId, claimFrontier), + ); + if (blocking.length === 0) return; + // A session paused on an interaction never reaches quiescence on its + // own — the user must answer first. Reject instead of parking the + // request indefinitely behind that decision (#3349 review). + const interactive = blocking.filter( + (sessionId) => this.listActiveInteractions(sessionId).length > 0, + ); + if (interactive.length > 0) { + throw new SessionQuiescentMutationBusyError(interactive); + } + await new Promise((resolve) => { + const wake = (): void => { + for (const sessionId of blocking) { + this.removeSessionQuiescenceWaiter(sessionId, wake); + } + resolve(); + }; + for (const sessionId of blocking) { + let waiters = this.sessionQuiescenceWaiters.get(sessionId); + if (!waiters) { + waiters = new Set(); + this.sessionQuiescenceWaiters.set(sessionId, waiters); + } + waiters.add(wake); + } + }); + } + } + + private removeSessionQuiescenceWaiter(sessionId: string, wake: () => void): void { + const waiters = this.sessionQuiescenceWaiters.get(sessionId); + if (!waiters) return; + waiters.delete(wake); + if (waiters.size === 0) this.sessionQuiescenceWaiters.delete(sessionId); + } + + private wakeSessionQuiescenceWaiters(sessionId: string): void { + const waiters = this.sessionQuiescenceWaiters.get(sessionId); + if (!waiters) return; + this.sessionQuiescenceWaiters.delete(sessionId); + for (const wake of [...waiters]) wake(); + } + private takeExecutionClaim( sessionId: string, supplied?: RuntimeExecutionClaim, @@ -647,6 +807,7 @@ export class RuntimeKernel implements RuntimeKernelLike { const claims = this.executionClaims.get(execution.sessionId); claims?.delete(execution); if (claims?.size === 0) this.executionClaims.delete(execution.sessionId); + this.wakeSessionQuiescenceWaiters(execution.sessionId); if (outcome.ok) execution.resolveSettled(); else execution.rejectSettled(outcome.error); } @@ -2677,6 +2838,9 @@ export class RuntimeKernel implements RuntimeKernelLike { generation: generation.generation, request: event, }); + // A queued quiescence mutation must re-evaluate: the session now pauses on + // an interaction and would otherwise look eternally busy to it. + this.wakeSessionQuiescenceWaiters(sessionId); } private clearInteractionRequestOwners(sessionId: string, turnId: string): void { @@ -2871,16 +3035,27 @@ export class RuntimeKernel implements RuntimeKernelLike { execution: PendingExecutionClaim, ): Promise { await this.clearBackendQuarantineForActivation(sessionId, execution); + // The boundary revision this activation is composed against. Recorded on + // the generation so a later activation can detect a config write that + // skipped backend disposal (#3349). An unreadable boundary leaves the + // guard dormant rather than blocking activation. + const boundaryRevision = await this.readBoundaryRevision(sessionId); let existing = this.active.get(sessionId); if (existing) { - existing.cachedHeader = header; - return existing; + const reusable = await this.resolveReusableGeneration(sessionId, existing, boundaryRevision); + if (reusable) { + reusable.cachedHeader = header; + return reusable; + } } await this.waitForBackendDisposal(sessionId); existing = this.active.get(sessionId); if (existing) { - existing.cachedHeader = header; - return existing; + const reusable = await this.resolveReusableGeneration(sessionId, existing, boundaryRevision); + if (reusable) { + reusable.cachedHeader = header; + return reusable; + } } const entry = await this.shareBackendActivation(`parent:${sessionId}`, async () => { const current = this.active.get(sessionId); @@ -2914,10 +3089,54 @@ export class RuntimeKernel implements RuntimeKernelLike { this.active.set(sessionId, generation); return generation; }); + // Concurrent activations share one build; the first to arrive stamps the + // revision it read. That stamp may trail the revision actually composed + // against (the reader ran before the builder) — safe direction: a later + // activation rebuilds once more, never reuses a newer composition blindly. + entry.boundaryRevision ??= boundaryRevision; entry.cachedHeader = header; return entry; } + private async readBoundaryRevision(sessionId: string): Promise { + // One dedicated store read per activation: cheaper than widening the + // session header read the turn already performs, and the guard is optional + // defense in depth — an unreadable boundary simply leaves it dormant. + try { + return (await this.deps.store.readExecutionBoundary(sessionId)).revision; + } catch { + return undefined; + } + } + + /** + * Defense in depth against a config write that bumped the durable boundary + * without disposing the backend generation it was composed against: dispose + * and rebuild now when nothing executes on the generation, and when runs are + * still live, mark the generation for invalidation instead — it flushes when + * they exit, and the next activation composes fresh. Tools are unaffected + * meanwhile: they read the boundary live on every call. + */ + private async resolveReusableGeneration( + sessionId: string, + existing: BackendGeneration, + boundaryRevision: number | undefined, + ): Promise { + if ( + boundaryRevision === undefined || + existing.boundaryRevision === undefined || + existing.boundaryRevision === boundaryRevision + ) { + return existing; + } + if (this.hasActiveRuns(sessionId)) { + this.ensureBackendInvalidation(sessionId); + return existing; + } + await this.disposeBackend(sessionId); + return undefined; + } + private async shareBackendActivation( activationKey: string, activate: () => Promise, @@ -3148,6 +3367,7 @@ export class RuntimeKernel implements RuntimeKernelLike { if (active.turnToRunId.get(run.turnId) === run.runId) { active.turnToRunId.delete(run.turnId); } + this.wakeSessionQuiescenceWaiters(active.sessionId); } private async unregisterParentRun(active: AgentRunActiveSession, run: AgentRun): Promise { diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 9edfe06a3f..b3df1df913 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -81,6 +81,7 @@ import type { SandboxBoundarySettlement, SettleSandboxBoundaryRequest, } from '@maka/core/sandbox-boundary'; +import { executionBoundaryMatchesPermissionMode } from '@maka/core/sandbox-boundary'; import type { CollaborationMode } from '@maka/core/collaboration'; import type { OrchestrationMode } from '@maka/core/orchestration'; import { @@ -1646,9 +1647,6 @@ export class SessionManager { return headerToSummary(previous); } - if (this.runtimeKernel.hasActiveRuns(sessionId)) { - throw new Error('当前任务正在运行，等结束后再切换权限模式。'); - } if (previous.status === 'waiting_for_user') { throw new Error('当前有工具调用正在等待确认，处理后再切换权限模式。'); } @@ -1679,9 +1677,6 @@ export class SessionManager { sessionId: string, kind: 'managed' | 'bypass', ): Promise { - if (this.runtimeKernel.hasActiveRuns(sessionId)) { - throw new Error('当前任务正在运行，等结束后再切换沙箱边界。'); - } const header = await this.deps.store.readHeader(sessionId); if (header.status === 'waiting_for_user') { throw new Error('当前有沙箱边界请求正在等待确认，处理后再切换。'); @@ -1719,64 +1714,92 @@ export class SessionManager { prepareCommit: () => Promise<() => Promise>, ): Promise { const initialBoundary = await this.deps.store.readExecutionBoundary(sessionId); - const initiallyNarrows = narrowsExecutionAuthority(initialBoundary, nextPermissionMode); - const initialDescendants = initiallyNarrows - ? await this.listLinkedDescendantSessionIds(sessionId) - : []; - const fencedSessionIds = [sessionId, ...initialDescendants]; + if (!narrowsExecutionAuthority(initialBoundary, nextPermissionMode)) { + return this.commitWideningTransition(sessionId, prepareCommit); + } + const initialDescendants = await this.listLinkedDescendantSessionIds(sessionId); + return this.commitTighteningTransition( + sessionId, + [sessionId, ...initialDescendants], + prepareCommit, + ); + } - return this.runSessionQuiescentMutation(fencedSessionIds, async () => { - const currentBoundary = await this.deps.store.readExecutionBoundary(sessionId); - const narrowsShellAuthority = narrowsExecutionAuthority(currentBoundary, nextPermissionMode); - const descendantSessionIds = narrowsShellAuthority - ? await this.listLinkedDescendantSessionIds(sessionId) - : []; + /** + * Widening grants more authority, so it only needs to reach turns that + * start after the request: it commits in the inter-turn gap, and the + * successor turn — admission-gated until it commits — observes the new + * configuration before its first tool call. A delayed grant is a UX + * tradeoff, not a hazard. + */ + private async commitWideningTransition( + sessionId: string, + prepareCommit: () => Promise<() => Promise>, + ): Promise { + return this.runSessionQueuedQuiescentMutation([sessionId], async () => { + const commit = await prepareCommit(); + await this.runtimeKernel.disposeBackend(sessionId); + return await commit(); + }); + } + + /** + * Tightening revokes authority, and revocation must not wait out the turn + * that is still executing under the wider grant. The durable boundary is + * committed on the mutation tail — serialized with other transitions, never + * waiting for claims or runs — so the live turn's next tool dispatch + * already reads the narrower authority, and background shell authority + * across the lineage is fenced at once. Backend disposal is deferred: a + * live run keeps executing on its generation (tools read the boundary live + * on every call), idle generations dispose through invalidation now, and + * the boundary-revision guard rebuilds stale ones on their next activation. + */ + private async commitTighteningTransition( + sessionId: string, + fencedSessionIds: readonly string[], + prepareCommit: () => Promise<() => Promise>, + ): Promise { + if (!this.runtimeKernel.runSessionAdmissionMutation) { + throw new SessionConfigurationTransitionError( + 'operation_unavailable', + 'Session execution mutation authority is unavailable', + ); + } + return this.runtimeKernel.runSessionAdmissionMutation([sessionId], async () => { + // Re-list the lineage after tail serialization: a descendant provisioned + // while this request queued is a fencing gap, not something to fence + // blindly — reject and let the retry fence the full lineage. + const descendants = await this.listLinkedDescendantSessionIds(sessionId); if ( - descendantSessionIds.some( - (descendantSessionId) => !fencedSessionIds.includes(descendantSessionId), - ) + descendants.some((descendantSessionId) => !fencedSessionIds.includes(descendantSessionId)) ) { throw new SessionConfigurationTransitionError( 'operation_conflict', 'Session lineage changed before the configuration transition', ); } - const lineageSessionIds = [sessionId, ...descendantSessionIds]; - if (lineageSessionIds.some((id) => this.runtimeKernel.hasActiveRuns(id))) { - throw new SessionConfigurationTransitionError( - 'session_busy', - 'Session configuration cannot change while a linked Turn is active', - ); - } - if (narrowsShellAuthority && !this.deps.shellRuns) { + if (!this.deps.shellRuns) { throw new SessionConfigurationTransitionError( 'operation_unavailable', 'Session permission narrowing requires Runtime Resource authority', ); } - const commit = await prepareCommit(); const descendantBoundaries = new Map(); - for (const descendantSessionId of descendantSessionIds) { + for (const descendantSessionId of descendants) { descendantBoundaries.set( descendantSessionId, await this.deps.store.readExecutionBoundary(descendantSessionId), ); } + const lineageSessionIds = [sessionId, ...descendants]; const shellRunCloses: Array>> = []; try { - if (narrowsShellAuthority) { - for (const lineageSessionId of lineageSessionIds) { - const close = await this.deps.shellRuns?.terminateSession(lineageSessionId); - if (close) shellRunCloses.push(close); - } + for (const lineageSessionId of lineageSessionIds) { + const close = await this.deps.shellRuns?.terminateSession(lineageSessionId); + if (close) shellRunCloses.push(close); } - await Promise.all( - lineageSessionIds.map((lineageSessionId) => - this.runtimeKernel.disposeBackend(lineageSessionId), - ), - ); } catch { for (const close of shellRunCloses) this.deps.shellRuns?.rollbackSessionClose(close); throw new SessionConfigurationTransitionError( @@ -1803,6 +1826,14 @@ export class SessionManager { } } } + // Deferred disposal, best-effort by design: an invalidation marked on a + // session with a live run flushes when the run exits; the narrower + // boundary is already durable, so disposal is hygiene, not safety. + await Promise.all( + lineageSessionIds.map((lineageSessionId) => + this.runtimeKernel.invalidateBackend(lineageSessionId).catch(() => undefined), + ), + ); return result; }); } @@ -1830,6 +1861,40 @@ export class SessionManager { } } + /** + * Quiescent mutation that queues behind live execution instead of rejecting: + * the kernel closes the session's admission gate and defers the operation + * until the claims and runs that predate the request settle, so a permission + * switch lands in the next inter-turn gap. Turns admitted after the request + * wait at the gate, which is what lets them observe the committed + * configuration. + */ + private async runSessionQueuedQuiescentMutation( + sessionIds: readonly string[], + operation: () => Promise, + ): Promise { + // A kernel without this method turns permission switching into + // operation_unavailable rather than session_busy; kernel and host ship as + // one versioned unit, so the cliff only matters for injected test doubles. + if (!this.runtimeKernel.runSessionQueuedQuiescentMutation) { + throw new SessionConfigurationTransitionError( + 'operation_unavailable', + 'Session execution mutation authority is unavailable', + ); + } + try { + return await this.runtimeKernel.runSessionQueuedQuiescentMutation(sessionIds, operation); + } catch (error) { + if (error instanceof SessionQuiescentMutationBusyError) { + throw new SessionConfigurationTransitionError( + 'session_busy', + 'Session has a pending Interaction', + ); + } + throw error; + } + } + private async listLinkedDescendantSessionIds(sessionId: string): Promise { const sessions = await this.deps.store.list(); const childrenByParent = new Map(); @@ -6328,17 +6393,6 @@ function claimedAgentGraphIntentResult( }; } -function executionBoundaryMatchesPermissionMode( - boundary: ExecutionBoundary, - mode: PermissionMode, -): boolean { - if (mode === 'bypass') return boundary.kind === 'bypass'; - if (boundary.kind !== 'managed') return false; - return mode === 'explore' - ? boundary.profile.name === 'read-only' - : boundary.profile.name !== 'read-only'; -} - function narrowsExecutionAuthority( boundary: ExecutionBoundary, nextPermissionMode: PermissionMode, diff --git a/packages/runtime/src/tool-runtime.ts b/packages/runtime/src/tool-runtime.ts index 84fab7c475..4dbe1f9337 100644 --- a/packages/runtime/src/tool-runtime.ts +++ b/packages/runtime/src/tool-runtime.ts @@ -23,12 +23,14 @@ import { projectToolActivityArgs } from '@maka/core/tool-activity-args'; import { type CreateSandboxBoundaryRequest, type ExecutionBoundary, + executionBoundaryDisplayMode, type SandboxBoundaryDecision, type SandboxBoundaryExpansion, type SandboxBoundaryRequest, type SandboxBoundarySettlement, type SettleSandboxBoundaryRequest, } from '@maka/core/sandbox-boundary'; +import { resolveCollaborationPermissionMode } from '@maka/core/permission'; import { serializedByteLength } from '@maka/core/serialized-byte-length'; import { encodeToolStepProgress, ToolOutcomeUnknownError } from '@maka/core/events'; import type { @@ -1323,6 +1325,20 @@ export class ToolRuntime { try { const runId = this.input.runId; const executionBoundary = clientCapabilityBoundary ?? (await this.readExecutionBoundary()); + // The boundary is the authority on what this session may do (#1611); + // the mode a tool acts on is derived from it per call, so a committed + // permission switch reaches the very next tool dispatch without + // waiting for a backend rebuild. An external boundary is not locally + // controllable and derives no mode — the last known header mode is the + // best available answer there. + const boundaryMode = executionBoundaryDisplayMode(executionBoundary); + const permissionMode = + boundaryMode === undefined + ? this.input.header.permissionMode + : resolveCollaborationPermissionMode({ + collaborationMode: this.input.header.collaborationMode ?? 'agent', + permissionMode: boundaryMode, + }); const result = await tool.impl(structuredClone(executionArgs) as never, { sessionId: this.input.sessionId, turnId, @@ -1332,7 +1348,7 @@ export class ToolRuntime { : {}), cwd: this.input.header.cwd, executionBoundary, - permissionMode: this.input.header.permissionMode, + permissionMode, toolCallId: toolUseId, // The id the call event actually carries, not the candidate: by here // `prepareDurableToolAttempt` has pushed it on the dispatch lane.