diff --git a/docs/architecture/runtime-managed-workspace-continuation-admission-v1.zh-CN.md b/docs/architecture/runtime-managed-workspace-continuation-admission-v1.zh-CN.md new file mode 100644 index 00000000000..38a8fda0bd7 --- /dev/null +++ b/docs/architecture/runtime-managed-workspace-continuation-admission-v1.zh-CN.md @@ -0,0 +1,52 @@ +# Managed workspace continuation admission v1 + +## 范围 + +本切片只证明一个主要不变量: + +> 对 `managed-coding-v1` 会话,continuation claim 必须同时绑定不可变 RuntimeEvent replay 边界和当前已接受的 Gitoxide workspace head;任一边界变化都不得启动 provider。 + +它不改变 provider replay 的 high-water 语义,也不创建新的 workspace baseline。 + +## Owner 与事实源 + +- RuntimeEvent replay 边界由 SQLite continuation authority 从不可变事件前缀签发。 +- accepted workspace 边界由 workspace version authority 从 epoch、head、version 和 storage-root binding 重建。 +- Runtime Host 只负责重新观测 Gitoxide `refs/maka/accepted`,并证明它与 SQLite head 一致。 +- `ContinuationClaimV2` 使用 domain-separated digest 绑定上述两个边界;target Run 使用 `continuation_source_v3`,同时保留独立的 replay manifest digest。 + +调用者不能提交裸 workspace digest,也不能用 v1 claim 替代 v2 claim。 + +## 原子性边界 + +SQLite 在一个事务中校验: + +1. immutable RuntimeEvent prefix; +2. storage-root binding; +3. workspace epoch/head/version; +4. claim target identity; +5. continuation claim 的唯一性。 + +claim 与 continuation-start 是两个 durable 边界。若进程停在两者之间,recovery 只能在重新观测到完全相同的 workspace boundary 后写入 repair start;它不会调 provider。 + +## 失败状态与回滚 + +- workspace head、accepted ref、receipt 或 storage root 不一致:park / fail closed。 +- workspace-bound authority 缺失:park;禁止回退 v1。 +- v2 claim 已提交、start 未提交:创建或核对 claim 中的 target Run,写 repair start,再写可审计失败终态。 +- normal start 已提交但 provider 状态未知:保持 indeterminate,不自动重放 provider。 +- claim 事务内崩溃:SQLite 回滚,不留下半 claim。 + +## 平台能力 + +| 平台 | 本切片承诺 | +| --- | --- | +| Linux | SQLite claim/start 原子性;Gitoxide accepted-ref 重验;process-crash 收敛 | +| macOS | 与 Linux 相同;路径以 `realpath` 后身份为准 | +| Windows | SQLite claim/start 原子性;Gitoxide accepted-ref 重验;不依赖 POSIX inode 语义 | + +断电持久性沿用 SQLite/Gitoxide owner 各自已经声明的平台合同,本切片不扩大承诺。 + +## 交付状态 + +这是 stacked Draft。Desktop/CLI 自动 continuation 仍受现有 feature gate 控制;在 production-shaped Host kill/reopen 证据进入三平台 recovery inventory 前,不转 Ready。 diff --git a/packages/core/src/runtime-event-store.ts b/packages/core/src/runtime-event-store.ts index cf359de1449..bf3a48557fb 100644 --- a/packages/core/src/runtime-event-store.ts +++ b/packages/core/src/runtime-event-store.ts @@ -166,6 +166,26 @@ export interface RuntimeWorkspaceBoundContinuationAuthorityStore extends Runtime readWorkspaceBoundContinuationClaimByBoundary( boundaryDigest: RuntimeBoundaryDigest, ): Promise; + readWorkspaceBoundContinuationClaimStateByBoundary( + boundaryDigest: RuntimeBoundaryDigest, + ): Promise; + listWorkspaceBoundContinuationClaimsForRecovery( + sessionId: string, + ): Promise; + commitWorkspaceBoundContinuationStart(input: { + claim: ContinuationClaimV2; + event: RuntimeEvent; + }): Promise<{ created: boolean; runtimeEventSeq: number }>; + commitWorkspaceBoundContinuationRepairStart(input: { + claim: ContinuationClaimV2; + event: RuntimeEvent; + }): Promise<{ created: boolean; runtimeEventSeq: number }>; +} + +export interface ContinuationClaimStateV2 { + claim: ContinuationClaimV2; + startEventId?: string; + startKind?: 'runtime_admission' | 'claim_repair'; } export interface RuntimeWorkspaceVersionAuthorityStore extends RuntimeEventStore { diff --git a/packages/runtime-host/src/__tests__/gitoxide-managed-mutation-session.test.ts b/packages/runtime-host/src/__tests__/gitoxide-managed-mutation-session.test.ts index 147f1dac598..ba4df4aaf5b 100644 --- a/packages/runtime-host/src/__tests__/gitoxide-managed-mutation-session.test.ts +++ b/packages/runtime-host/src/__tests__/gitoxide-managed-mutation-session.test.ts @@ -36,6 +36,7 @@ import { issueGitoxideHelperReleaseArtifactClaimInternal, } from '../server/gitoxide-helper-artifact-authority-internal.js'; import { + inspectGitoxideManagedContinuationBoundary, openGitoxideManagedMutationSession, recoverGitoxideManagedMutationBeforeRunClosureInternal, } from '../server/gitoxide-managed-mutation-session.js'; @@ -100,8 +101,12 @@ test('opens one durable Gitoxide baseline and exactly reuses it for the session' const first = await openGitoxideManagedMutationSession(input); const reopened = await openGitoxideManagedMutationSession(input); + const initialContinuationBoundary = await inspectGitoxideManagedContinuationBoundary(input); assert.deepEqual(reopened.head, first.head); + assert.equal(initialContinuationBoundary?.commitOid, first.head.commitOid); + assert.equal(initialContinuationBoundary?.treeOid, first.head.treeOid); + assert.equal(initialContinuationBoundary?.revision, first.head.revision); assert.equal(first.head.revision, 1); assert.notEqual(first.head.commitOid, git(sourceRoot, ['rev-parse', 'HEAD'])); assert.equal(first.head.treeOid, git(sourceRoot, ['rev-parse', 'HEAD^{tree}'])); @@ -161,6 +166,10 @@ test('opens one durable Gitoxide baseline and exactly reuses it for the session' 'after\n', ); assert.equal(await readFile(join(sourceRoot, 'notes.txt'), 'utf8'), 'baseline\n'); + const successorContinuationBoundary = await inspectGitoxideManagedContinuationBoundary(input); + assert.equal(successorContinuationBoundary?.commitOid, afterChange.head.commitOid); + assert.equal(successorContinuationBoundary?.treeOid, afterChange.head.treeOid); + assert.equal(successorContinuationBoundary?.revision, 3); const noChange = await executeManagedWrite({ stores, diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 7ed7cc36327..cbf78021fe5 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -185,6 +185,7 @@ import { import { resolvePackagedGitoxideHelperInternal } from './packaged-gitoxide-helper-internal.js'; import type { GitoxideHelperInvocationCapability } from './gitoxide-helper-artifact-authority-internal.js'; import { + inspectGitoxideManagedContinuationBoundary, openGitoxideManagedMutationSession, recoverGitoxideManagedMutationBeforeRunClosureInternal, } from './gitoxide-managed-mutation-session.js'; @@ -1004,6 +1005,19 @@ export async function createExecutionRuntimeHostComposition( resourcesLive || graphLive || graphWake.hasLiveSessionState(sessionId) || descendantLive ); }, + readManagedWorkspaceBoundary: async (sessionId) => { + const header = await stores.sessionStore.readHeaderSnapshot(sessionId); + if (header.toolProfile !== 'managed-coding-v1') return undefined; + const runtime = requireGitoxideManagedMutationRuntime(gitoxideManagedMutationRuntime); + return inspectGitoxideManagedContinuationBoundary({ + storageRootLease: context.owner.lease, + sourceRoot: header.cwd, + sessionId, + invocationOwnerToken: runtime.invocationOwnerToken, + helperCapability: runtime.helperCapability, + settlementAuthority: requireExecutionStoresWorkspaceMutationAuthorityInternal(stores), + }); + }, }), runBackendActivation: (operation) => runtimePolicyActivation.runBackendActivation(operation), messageAuthority: runtimeAuthority, diff --git a/packages/runtime-host/src/server/gitoxide-managed-mutation-session.ts b/packages/runtime-host/src/server/gitoxide-managed-mutation-session.ts index c171f201493..c63d588330d 100644 --- a/packages/runtime-host/src/server/gitoxide-managed-mutation-session.ts +++ b/packages/runtime-host/src/server/gitoxide-managed-mutation-session.ts @@ -24,6 +24,7 @@ import { dirname, join } from 'node:path'; import { workspaceMutationPolicyHashV1 } from '@maka/core/workspace-version-authority'; import { GITOXIDE_MANAGED_MUTATION_TRANSFORM_PROFILE_DIGEST } from '@maka/runtime/managed-mutation-transform'; import type { ToolRuntimeInput } from '@maka/runtime/tool-runtime'; +import type { ManagedWorkspaceContinuationBoundaryV1 } from '@maka/core/runtime-boundary'; import type { WorkspaceHeadRecordV1 } from '@maka/core/workspace-version-authority'; import type { ExecutionStoresWorkspaceMutationAuthorityInternal } from '@maka/storage/execution-stores-workspace-authority-internal'; import { runWithStorageRootLease, type StorageRootLease } from '@maka/storage/root-authority'; @@ -133,6 +134,72 @@ export async function recoverGitoxideManagedMutationBeforeRunClosureInternal(inp } } +/** + * Re-observes an already accepted managed workspace without creating a + * baseline. This is the only Host seam allowed to issue the workspace half of + * a continuation boundary. + */ +export async function inspectGitoxideManagedContinuationBoundary(input: { + readonly storageRootLease: StorageRootLease<'interactive', 'write'>; + readonly sourceRoot: string; + readonly sessionId: string; + readonly invocationOwnerToken: object; + readonly helperCapability: GitoxideHelperInvocationCapability; + readonly settlementAuthority: ExecutionStoresWorkspaceMutationAuthorityInternal; + readonly abortSignal?: AbortSignal; +}): Promise { + input.abortSignal?.throwIfAborted(); + const [storageRoot, sourceRoot, helper] = await Promise.all([ + runWithStorageRootLease(input.storageRootLease, 'interactive', 'write', async (root) => root), + realpath(input.sourceRoot), + verifyGitoxideHelperArtifactForInvocationInternal( + input.invocationOwnerToken, + input.helperCapability, + ), + ]); + const materializationProfileDigest = sha256( + `maka-gitoxide-materialization-v1\0${helper.artifactSha256}\0`, + ); + const workspacePolicyHash = workspaceMutationPolicyHashV1( + materializationProfileDigest, + GITOXIDE_MANAGED_MUTATION_TRANSFORM_PROFILE_DIGEST, + ); + const identity = managedMutationIdentity(sourceRoot, input.sessionId); + const boundary = await input.settlementAuthority.readContinuationBoundary( + identity.workspaceId, + identity.workspaceEpochId, + GITOXIDE_MANAGED_MUTATION_TRANSFORM_PROFILE_DIGEST, + ); + if (!boundary) return undefined; + if ( + boundary.repositoryId !== identity.repositoryId || + boundary.workspaceId !== identity.workspaceId || + boundary.workspaceEpochId !== identity.workspaceEpochId || + boundary.workspaceInstanceId !== identity.workspaceInstanceId + ) { + throw new Error('Gitoxide managed continuation boundary conflicts with session identity'); + } + const repositoryPath = gitoxideManagedRepositoryPathInternal(storageRoot, identity); + const receipt = await readBaselineReceipt(join(dirname(repositoryPath), 'baseline-receipt.json')); + if ( + !receipt || + receipt.repositoryId !== boundary.repositoryId || + receipt.workspaceId !== boundary.workspaceId || + receipt.workspaceEpochId !== boundary.workspaceEpochId || + receipt.workspaceInstanceId !== boundary.workspaceInstanceId || + receipt.helperArtifactSha256 !== helper.artifactSha256 || + boundary.materializationProfileDigest !== materializationProfileDigest || + boundary.executionProfileDigest !== + GITOXIDE_MANAGED_MUTATION_TRANSFORM_PROFILE_DIGEST || + boundary.policyHash !== workspacePolicyHash + ) { + throw new Error('Gitoxide managed continuation baseline receipt is unavailable'); + } + await verifyAcceptedRef(input, repositoryPath, boundary.commitOid, boundary.treeOid); + input.abortSignal?.throwIfAborted(); + return boundary; +} + /** * Opens one explicit managed-coding session. The source observation is frozen * before import, Gitoxide owns the immutable repository, SQLite owns accepted diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index c05a8a019d8..a46c9d722eb 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -21,6 +21,7 @@ import { createHash, randomUUID } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import type { BackendStopMode } from '@maka/core/backend-types'; import type { AgentRunHeader, RootExecutionDescriptor } from '@maka/core/agent-run'; +import { digestWorkspaceBoundContinuationBoundary } from '@maka/core/runtime-boundary'; import { INLINE_REFERENCE_MAX_COUNT, messageContentsEqual, @@ -2665,12 +2666,18 @@ function requirePlannedContinuation(plan: SafeBoundaryContinuationPlan): Runtime function continuationExecutionDescriptor( continuation: RuntimeContinuation, ): Extract { - const boundaryDigest = continuation.boundary?.manifestDigest; - if (!continuation.claimId || !boundaryDigest || !continuation.providerReplayDigest) { + const replayManifestDigest = continuation.boundary?.manifestDigest; + if (!continuation.claimId || !replayManifestDigest || !continuation.providerReplayDigest) { throw new RuntimeMessageAuthorityInvariantError( 'Authoritative continuation plan omitted its durable replay proof', ); } + const boundaryDigest = continuation.workspaceBoundary + ? digestWorkspaceBoundContinuationBoundary( + continuation.boundary!, + continuation.workspaceBoundary, + ) + : replayManifestDigest; return { kind: 'safe_boundary_continuation', sourceInvocationId: continuation.sourceInvocationId, @@ -2679,6 +2686,7 @@ function continuationExecutionDescriptor( sourceRuntimeEventHighWater: continuation.sourceRuntimeEventHighWater, claimId: continuation.claimId, boundaryDigest, + ...(continuation.workspaceBoundary ? { replayManifestDigest } : {}), providerReplayDigest: continuation.providerReplayDigest, safetyDigest: continuationSafetyDigest(continuation), targetInvocationId: continuation.invocationId, @@ -2695,6 +2703,7 @@ export function continuationSafetyDigest(continuation: RuntimeContinuation): `sh snapshot.workspaceCheckpoint ? [snapshot.workspaceCheckpoint.ref, snapshot.workspaceCheckpoint.runtimeEventHighWater] : null, + ...(snapshot.workspaceBoundary ? [snapshot.workspaceBoundary] : []), ]); return `sha256:${createHash('sha256').update(body, 'utf8').digest('hex')}`; } diff --git a/packages/runtime/src/__tests__/runtime-continuation.test.ts b/packages/runtime/src/__tests__/runtime-continuation.test.ts index 76879d086b1..2c35fab4231 100644 --- a/packages/runtime/src/__tests__/runtime-continuation.test.ts +++ b/packages/runtime/src/__tests__/runtime-continuation.test.ts @@ -25,6 +25,7 @@ import { createRuntimeBoundaryCursor, runtimePrefixSegment, type ImmutableRuntimePrefixV1, + type ManagedWorkspaceContinuationBoundaryV1, } from '@maka/core/runtime-boundary'; import type { RuntimeEvent } from '@maka/core/runtime-event'; import type { AgentRunHeader } from '@maka/core/agent-run'; @@ -337,6 +338,89 @@ test('RuntimeContinuationPlanner reads the durable source boundary and allocates }); }); +test('RuntimeContinuationPlanner binds managed continuation admission to the observed accepted head', async () => { + const sourcePrefix = immutablePrefix([ + event({ + id: 'source-user', + role: 'user', + author: 'user', + content: { kind: 'text', text: 'continue managed work' }, + }), + event({ + id: 'source-terminal', + role: 'system', + author: 'system', + status: 'failed', + actions: { endInvocation: true }, + }), + ]); + const workspaceBoundary = managedWorkspaceBoundary(); + let readWorkspaceClaim = false; + const ids = ['invocation-2', 'run-2', 'turn-2', 'claim-2']; + const planner = new RuntimeContinuationPlanner({ + readSourceRun: async () => runHeader('run-1'), + readImmutableRuntimePrefix: async () => sourcePrefix, + readWorkspaceBoundContinuationClaimStateByBoundary: async () => { + readWorkspaceClaim = true; + return undefined; + }, + newId: () => ids.shift() ?? 'unexpected-id', + }); + + const plan = await planner.plan({ + sessionId: 'session-1', + sourceRunId: 'run-1', + currentCwd: '/workspace/repo', + sourceWorkspaceIdentity: 'workspace-1', + currentWorkspaceIdentity: 'workspace-1', + backgroundOperationsSettled: true, + availableToolNames: ['Write', 'Edit'], + workspaceBoundary, + }); + + assert.equal(plan.disposition, 'continue'); + assert.equal(readWorkspaceClaim, true); + assert.deepEqual(plan.continuation?.workspaceBoundary, workspaceBoundary); + assert.deepEqual(plan.continuation?.safetySnapshot.workspaceBoundary, workspaceBoundary); +}); + +test('RuntimeContinuationPlanner never downgrades a managed boundary to the v1 claim authority', async () => { + let readV1Claim = false; + const planner = new RuntimeContinuationPlanner({ + readSourceRun: async () => runHeader('run-1'), + readImmutableRuntimePrefix: async () => + immutablePrefix([ + event({ + id: 'source-terminal', + role: 'system', + author: 'system', + status: 'failed', + actions: { endInvocation: true }, + }), + ]), + readContinuationClaimStateByBoundary: async () => { + readV1Claim = true; + return undefined; + }, + newId: () => 'unused', + }); + + const plan = await planner.plan({ + sessionId: 'session-1', + sourceRunId: 'run-1', + currentCwd: '/workspace/repo', + sourceWorkspaceIdentity: 'workspace-1', + currentWorkspaceIdentity: 'workspace-1', + backgroundOperationsSettled: true, + availableToolNames: [], + workspaceBoundary: managedWorkspaceBoundary(), + }); + + assert.equal(plan.disposition, 'park'); + assert.deepEqual(plan.rejectionReasons, ['continuation_authority_unavailable']); + assert.equal(readV1Claim, false); +}); + test('RuntimeRunner rejects a continuation envelope whose high-water is behind its replay context', async () => { const runner = new RuntimeRunner({ flow: { @@ -1149,6 +1233,26 @@ function runHeader(runId: string, overrides: Partial = {}): Agen }; } +function managedWorkspaceBoundary(): ManagedWorkspaceContinuationBoundaryV1 { + return { + protocol: 'managed_workspace_continuation_boundary_v1', + storageRootId: 'a'.repeat(64), + repositoryId: `repository_${'1'.repeat(32)}`, + workspaceId: `workspace_${'2'.repeat(32)}`, + workspaceEpochId: `epoch_${'3'.repeat(32)}`, + workspaceInstanceId: `instance_${'4'.repeat(32)}`, + workspaceVersionId: `version_${'5'.repeat(32)}`, + acceptedEventId: 'accepted-event-1', + revision: 2, + objectFormat: 'sha1', + commitOid: 'b'.repeat(40), + treeOid: 'c'.repeat(40), + materializationProfileDigest: `sha256:${'d'.repeat(64)}`, + policyHash: `sha256:${'e'.repeat(64)}`, + executionProfileDigest: `sha256:${'f'.repeat(64)}`, + }; +} + function event(overrides: Partial): RuntimeEvent { return { id: 'event', diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 1702134a22e..b34784ccd90 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -5365,6 +5365,64 @@ describe('SessionManager permission mode updates', () => { expect(plan.rejectionReasons).toEqual(['safety_observation_unavailable']); }); + test('never downgrades a managed session when its workspace boundary is unavailable', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends: new BackendRegistry(), + safeBoundaryResumeEnabled: true, + inspectContinuationSafety: async () => ({ + workspaceIdentity: 'workspace-managed', + backgroundOperationsSettled: true, + availableToolNames: ['Write', 'Edit'], + }), + newId: nextId(), + now: nextNow(6_533), + }); + const session = await manager.createSession(makeInput({ toolProfile: 'managed-coding-v1' })); + const header = await store.readHeader(session.id); + expect(header.toolProfile).toBe('managed-coding-v1'); + const sourceRunId = 'source-run-managed-boundary-missing'; + const sourceTurnId = 'source-turn-managed-boundary-missing'; + await seedRuntimeRun( + runStore, + makeRunHeader({ + runId: sourceRunId, + sessionId: session.id, + turnId: sourceTurnId, + status: 'failed', + cwd: header.cwd, + workspaceIdentity: 'workspace-managed', + createdAt: 1, + updatedAt: 2, + completedAt: 2, + failureClass: 'app_restarted', + }), + [ + runtimeEvent({ + id: 'source-terminal-managed-boundary-missing', + invocationId: 'source-invocation-managed-boundary-missing', + runId: sourceRunId, + sessionId: session.id, + turnId: sourceTurnId, + ts: 2, + status: 'failed', + actions: { endInvocation: true, stateDelta: { failureClass: 'app_restarted' } }, + }), + ], + ); + + const plan = await manager.planAuthoritativeSafeBoundaryContinuation(session.id, { + sourceRunId, + }); + + expect(plan.disposition).toBe('park'); + expect(plan.rejectionReasons).toEqual(['workspace_boundary_unavailable']); + }); + test('keeps the authoritative continuation entry disabled unless the host enables it', async () => { const store = new MemorySessionStore(); const backends = new BackendRegistry(); @@ -17484,6 +17542,7 @@ class MemorySessionStore implements SessionStore { permissionMode: input.permissionMode, collaborationMode: input.collaborationMode ?? 'agent', orchestrationMode: input.orchestrationMode ?? 'default', + ...(input.toolProfile !== undefined ? { toolProfile: input.toolProfile } : {}), schemaVersion: 1, }; this.headers.set(header.id, header); diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts index 0a43b140690..a8368ac5ac5 100644 --- a/packages/runtime/src/agent-run.ts +++ b/packages/runtime/src/agent-run.ts @@ -25,6 +25,7 @@ import type { } from '@maka/core/agent-run'; import type { RuntimeEvent, ToolBoundaryProtocol } from '@maka/core/runtime-event'; import type { RuntimeEventStore } from '@maka/core/runtime-event-store'; +import { digestWorkspaceBoundContinuationBoundary } from '@maka/core/runtime-boundary'; import type { RunCompositionSnapshot } from '@maka/core/run-composition'; import { decodeRunCompositionSnapshot } from '@maka/core/run-composition'; import { DurableStoreWriteError, RunSealedError } from '@maka/core/runtime-event-store'; @@ -1116,9 +1117,16 @@ export class AgentRun { continuationSource: continuation.claimId && continuation.boundary ? { - protocol: 'continuation_source_v2' as const, + protocol: continuation.workspaceBoundary + ? ('continuation_source_v3' as const) + : ('continuation_source_v2' as const), claimId: continuation.claimId, - boundaryDigest: continuation.boundary.manifestDigest, + boundaryDigest: continuation.workspaceBoundary + ? digestWorkspaceBoundContinuationBoundary( + continuation.boundary, + continuation.workspaceBoundary, + ) + : continuation.boundary.manifestDigest, sourceInvocationId: continuation.sourceInvocationId, sourceRunId: continuation.sourceRunId, sourceTurnId: continuation.sourceTurnId, diff --git a/packages/runtime/src/continuation-safety.ts b/packages/runtime/src/continuation-safety.ts index 1e5b06934c0..26f773df817 100644 --- a/packages/runtime/src/continuation-safety.ts +++ b/packages/runtime/src/continuation-safety.ts @@ -32,6 +32,9 @@ export interface LocalContinuationSafetyInspectorDeps { readWorkspaceCheckpoint?: ( sessionId: string, ) => Promise; + readManagedWorkspaceBoundary?: ( + sessionId: string, + ) => Promise; } export function createLocalContinuationSafetyInspector( @@ -39,19 +42,26 @@ export function createLocalContinuationSafetyInspector( ): (sessionId: string) => Promise { return async (sessionId) => { const cwd = await deps.readSessionCwd(sessionId); - const [workspace, availableToolNames, hasPendingBackgroundOperations, workspaceCheckpoint] = - await Promise.all([ - deps.resolveWorkspaceIdentity(cwd), - deps.listAvailableToolNames(sessionId), - deps.hasPendingBackgroundOperations(sessionId), - deps.readWorkspaceCheckpoint?.(sessionId), - ]); + const [ + workspace, + availableToolNames, + hasPendingBackgroundOperations, + workspaceCheckpoint, + workspaceBoundary, + ] = await Promise.all([ + deps.resolveWorkspaceIdentity(cwd), + deps.listAvailableToolNames(sessionId), + deps.hasPendingBackgroundOperations(sessionId), + deps.readWorkspaceCheckpoint?.(sessionId), + deps.readManagedWorkspaceBoundary?.(sessionId), + ]); return { workspaceIdentity: workspace.workspaceIdentity, workspacePath: workspace.canonicalPath, backgroundOperationsSettled: !hasPendingBackgroundOperations, availableToolNames: [...new Set(availableToolNames)].sort(), ...(workspaceCheckpoint ? { workspaceCheckpoint } : {}), + ...(workspaceBoundary ? { workspaceBoundary } : {}), }; }; } diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts index fa6043b72c2..36b04f03730 100644 --- a/packages/runtime/src/runtime-kernel.ts +++ b/packages/runtime/src/runtime-kernel.ts @@ -18,11 +18,18 @@ */ import type { AgentRunHeader, AgentRunStore } from '@maka/core/agent-run'; -import type { ContinuationClaimV1, ImmutableRuntimePrefixV1 } from '@maka/core/runtime-boundary'; +import { + digestWorkspaceBoundContinuationBoundary, + type ContinuationClaim, + type ContinuationClaimV1, + type ContinuationClaimV2, + type ImmutableRuntimePrefixV1, +} from '@maka/core/runtime-boundary'; import type { RuntimeEvent, ToolBoundaryProtocol } from '@maka/core/runtime-event'; import type { RuntimeContinuationAuthorityStore, RuntimeEventStore, + RuntimeWorkspaceBoundContinuationAuthorityStore, } from '@maka/core/runtime-event-store'; import { isSessionInlineRun } from '@maka/core/agent-run'; import type { @@ -824,7 +831,14 @@ export class RuntimeKernel implements RuntimeKernelLike { claimedAt, }); const claim = continuationClaimForExecution(continuation, claimedAt, targetRunHeader); - const claimResult = await continuationAuthority.claimContinuation({ claim }); + const workspaceContinuationAuthority = + claim.protocol === 'continuation_claim_v2' + ? requireRuntimeWorkspaceBoundContinuationAuthority(this.deps.runtimeEventStore) + : undefined; + const claimResult = + claim.protocol === 'continuation_claim_v2' + ? await workspaceContinuationAuthority!.claimWorkspaceBoundContinuation({ claim }) + : await continuationAuthority.claimContinuation({ claim }); if (claimResult.kind !== 'acquired') { throw new RuntimeContinuationRevalidationError( 'continuation_claim_conflict', @@ -878,7 +892,9 @@ export class RuntimeKernel implements RuntimeKernelLike { commitContinuationStart: async (startedAt) => { const source = claim.boundary.segments.at(-1)!; const eventId = this.deps.newId(); - const result = await continuationAuthority.commitContinuationStart({ + const result = await commitContinuationStartForClaim({ + continuationAuthority, + workspaceContinuationAuthority, claim, event: { id: eventId, @@ -1332,9 +1348,14 @@ export class RuntimeKernel implements RuntimeKernelLike { claimedAt, }); const claim = continuationClaimForExecution(continuation, claimedAt, targetRunHeader); - const claimResult = await continuationAuthority.claimContinuation({ - claim, - }); + const workspaceContinuationAuthority = + claim.protocol === 'continuation_claim_v2' + ? requireRuntimeWorkspaceBoundContinuationAuthority(this.deps.runtimeEventStore) + : undefined; + const claimResult = + claim.protocol === 'continuation_claim_v2' + ? await workspaceContinuationAuthority!.claimWorkspaceBoundContinuation({ claim }) + : await continuationAuthority.claimContinuation({ claim }); if (claimResult.kind !== 'acquired') { throw new RuntimeContinuationRevalidationError( 'continuation_claim_conflict', @@ -1353,7 +1374,9 @@ export class RuntimeKernel implements RuntimeKernelLike { commitContinuationStart: async (startedAt) => { const source = claim.boundary.segments.at(-1)!; const eventId = this.deps.newId(); - const result = await continuationAuthority.commitContinuationStart({ + const result = await commitContinuationStartForClaim({ + continuationAuthority, + workspaceContinuationAuthority, claim, event: { id: eventId, @@ -3362,6 +3385,45 @@ function requireRuntimeContinuationAuthority( return candidate as RuntimeContinuationAuthorityStore; } +function requireRuntimeWorkspaceBoundContinuationAuthority( + store: RuntimeEventStore, +): RuntimeWorkspaceBoundContinuationAuthorityStore { + const candidate = store as Partial; + if ( + candidate.workspaceBoundContinuationAuthorityCapability !== + 'runtime_workspace_bound_continuation_authority_v1' || + typeof candidate.claimWorkspaceBoundContinuation !== 'function' || + typeof candidate.readWorkspaceBoundContinuationClaimStateByBoundary !== 'function' || + typeof candidate.listWorkspaceBoundContinuationClaimsForRecovery !== 'function' || + typeof candidate.commitWorkspaceBoundContinuationStart !== 'function' || + typeof candidate.commitWorkspaceBoundContinuationRepairStart !== 'function' + ) { + throw new Error('Managed Runtime continuation requires workspace-bound SQLite authority'); + } + return candidate as RuntimeWorkspaceBoundContinuationAuthorityStore; +} + +async function commitContinuationStartForClaim(input: { + continuationAuthority: RuntimeContinuationAuthorityStore; + workspaceContinuationAuthority?: RuntimeWorkspaceBoundContinuationAuthorityStore; + claim: ContinuationClaim; + event: RuntimeEvent; +}): Promise<{ created: boolean; runtimeEventSeq: number }> { + if (input.claim.protocol === 'continuation_claim_v2') { + if (!input.workspaceContinuationAuthority) { + throw new Error('Workspace-bound continuation start authority is unavailable'); + } + return input.workspaceContinuationAuthority.commitWorkspaceBoundContinuationStart({ + claim: input.claim, + event: input.event, + }); + } + return input.continuationAuthority.commitContinuationStart({ + claim: input.claim, + event: input.event, + }); +} + async function revalidateContinuationBoundary( store: RuntimeContinuationAuthorityStore, continuation: RuntimeContinuation, @@ -3421,7 +3483,7 @@ function continuationClaimForExecution( continuation: RuntimeContinuation, claimedAt: number, targetRunHeader: AgentRunHeader, -): ContinuationClaimV1 { +): ContinuationClaim { if ( !continuation.claimId || !continuation.boundary || @@ -3433,10 +3495,14 @@ function continuationClaimForExecution( 'Runtime continuation is missing its durable claim identity', ); } - return { - protocol: 'continuation_claim_v1', + const common = { claimId: continuation.claimId, - boundaryDigest: continuation.boundary.manifestDigest, + boundaryDigest: continuation.workspaceBoundary + ? digestWorkspaceBoundContinuationBoundary( + continuation.boundary, + continuation.workspaceBoundary, + ) + : continuation.boundary.manifestDigest, boundary: continuation.boundary, providerProjectionVersion: continuation.providerProjectionVersion, providerReplayDigest: continuation.providerReplayDigest, @@ -3449,6 +3515,17 @@ function continuationClaimForExecution( targetRunHeader, claimedAt, }; + if (continuation.workspaceBoundary) { + return { + protocol: 'continuation_claim_v2', + ...common, + workspaceBoundary: continuation.workspaceBoundary, + } satisfies ContinuationClaimV2; + } + return { + protocol: 'continuation_claim_v1', + ...common, + } satisfies ContinuationClaimV1; } function continuationTargetRunHeaderForExecution(input: { @@ -3475,6 +3552,12 @@ function continuationTargetRunHeaderForExecution(input: { ); } const source = continuation.boundary.segments.at(-1)!; + const boundaryDigest = continuation.workspaceBoundary + ? digestWorkspaceBoundContinuationBoundary( + continuation.boundary, + continuation.workspaceBoundary, + ) + : continuation.boundary.manifestDigest; return { runId: continuation.runId, invocationId: continuation.invocationId, @@ -3507,9 +3590,11 @@ function continuationTargetRunHeaderForExecution(input: { ...(userInput.agentId ? { agentId: userInput.agentId } : {}), ...(userInput.agentName ? { agentName: userInput.agentName } : {}), continuationSource: { - protocol: 'continuation_source_v2', + protocol: continuation.workspaceBoundary + ? 'continuation_source_v3' + : 'continuation_source_v2', claimId: continuation.claimId, - boundaryDigest: continuation.boundary.manifestDigest, + boundaryDigest, sourceInvocationId: source.identity.invocationId, sourceRunId: source.identity.runId, sourceTurnId: source.identity.turnId, @@ -3594,6 +3679,14 @@ function assertContinuationSafetyUnchanged( 'Runtime continuation workspace identity changed after planning', ); } + if ( + !isDeepStrictEqual(observation.workspaceBoundary ?? null, snapshot.workspaceBoundary ?? null) + ) { + throw new RuntimeContinuationRevalidationError( + 'workspace_version_changed', + 'Runtime continuation accepted workspace boundary changed after planning', + ); + } if (!observation.backgroundOperationsSettled) { throw new RuntimeContinuationRevalidationError( 'background_operation_started', diff --git a/packages/runtime/src/runtime-resume.ts b/packages/runtime/src/runtime-resume.ts index 6e861da307e..c9255c39c58 100644 --- a/packages/runtime/src/runtime-resume.ts +++ b/packages/runtime/src/runtime-resume.ts @@ -27,13 +27,18 @@ import { type RuntimeEventFunctionResponseContent, } from '@maka/core/runtime-event'; import type { - ContinuationClaimV1, + ContinuationClaim, ImmutableRuntimePrefixV1, + ManagedWorkspaceContinuationBoundaryV1, RuntimeBoundaryCursorV1, RuntimeBoundaryDigest, } from '@maka/core/runtime-boundary'; +import { digestWorkspaceBoundContinuationBoundary } from '@maka/core/runtime-boundary'; import type { AgentRunHeader } from '@maka/core/agent-run'; -import type { ContinuationClaimStateV1 } from '@maka/core/runtime-event-store'; +import type { + ContinuationClaimStateV1, + ContinuationClaimStateV2, +} from '@maka/core/runtime-event-store'; import { isDeepStrictEqual } from 'node:util'; import { buildContinuationReplayPlan, @@ -77,6 +82,7 @@ export type RuntimeContinuationRevalidationCode = | 'source_ledger_identity_changed' | 'source_replay_changed' | 'workspace_identity_changed' + | 'workspace_version_changed' | 'background_operation_started' | 'tool_catalog_changed' | 'workspace_checkpoint_changed'; @@ -123,6 +129,7 @@ export type ResumePlanDiagnosticCode = | 'source_run_unreadable' | 'continuation_already_exists' | 'continuation_authority_unavailable' + | 'workspace_boundary_unavailable' | 'continuation_claim_repair_required' | 'continuation_started_indeterminate' | 'workspace_identity_missing' @@ -166,6 +173,7 @@ export type ResumeRejectionReason = | 'source_run_unreadable' | 'continuation_already_exists' | 'continuation_authority_unavailable' + | 'workspace_boundary_unavailable' | 'continuation_claim_repair_required' | 'continuation_started_indeterminate' | 'workspace_identity_missing' @@ -300,6 +308,7 @@ export interface SafeBoundaryContinuationFacts { restored: boolean; runtimeEventHighWater: number; }; + workspaceBoundary?: ManagedWorkspaceContinuationBoundaryV1; } export interface RuntimeContinuation { @@ -322,6 +331,7 @@ export interface RuntimeContinuation { /** Identity of the exact provider-facing replay projection. */ providerReplayDigest?: RuntimeBoundaryDigest; providerProjectionVersion?: typeof PROVIDER_REPLAY_PROJECTION_VERSION; + workspaceBoundary?: ManagedWorkspaceContinuationBoundaryV1; safetySnapshot: RuntimeContinuationSafetySnapshot; } @@ -333,6 +343,7 @@ export interface RuntimeContinuationSafetySnapshot { ref: string; runtimeEventHighWater: number; }; + workspaceBoundary?: ManagedWorkspaceContinuationBoundaryV1; } export interface RuntimeContinuationSafetyObservation { @@ -346,6 +357,7 @@ export interface RuntimeContinuationSafetyObservation { restored: boolean; runtimeEventHighWater: number; }; + workspaceBoundary?: ManagedWorkspaceContinuationBoundaryV1; } export interface SafeBoundaryContinuationPlan { @@ -365,6 +377,9 @@ export interface RuntimeContinuationPlannerInput { availableToolNames: readonly string[]; expectedRuntimeEventHighWater?: number; workspaceCheckpoint?: SafeBoundaryContinuationFacts['workspaceCheckpoint']; + workspaceBoundary?: ManagedWorkspaceContinuationBoundaryV1; + /** Managed coding requires an authenticated workspace boundary and may never downgrade to v1. */ + workspaceBoundaryRequirement?: 'optional' | 'required'; } export interface RuntimeContinuationPlannerDeps { @@ -377,6 +392,9 @@ export interface RuntimeContinuationPlannerDeps { readContinuationClaimStateByBoundary?( boundaryDigest: RuntimeBoundaryDigest, ): Promise; + readWorkspaceBoundContinuationClaimStateByBoundary?( + boundaryDigest: RuntimeBoundaryDigest, + ): Promise; findExistingContinuation?( sessionId: string, sourceRunId: string, @@ -389,6 +407,12 @@ export class RuntimeContinuationPlanner { constructor(private readonly deps: RuntimeContinuationPlannerDeps) {} async plan(input: RuntimeContinuationPlannerInput): Promise { + if (input.workspaceBoundaryRequirement === 'required' && !input.workspaceBoundary) { + return parkedPlan( + 'workspace_boundary_unavailable', + 'managed continuation requires an authoritative workspace boundary', + ); + } let sourceRun: Awaited>; try { sourceRun = await this.deps.readSourceRun(input.sessionId, input.sourceRunId); @@ -435,11 +459,22 @@ export class RuntimeContinuationPlanner { `continuation replay segment ${replay.segmentIndex} is not replayable: ${replay.reason}`, ); } - let durableClaimState: ContinuationClaimStateV1 | undefined; - try { - durableClaimState = await this.deps.readContinuationClaimStateByBoundary?.( - replay.plan.boundary.manifestDigest, + const plannedBoundaryDigest = input.workspaceBoundary + ? digestWorkspaceBoundContinuationBoundary(replay.plan.boundary, input.workspaceBoundary) + : replay.plan.boundary.manifestDigest; + if (input.workspaceBoundary && !this.deps.readWorkspaceBoundContinuationClaimStateByBoundary) { + return parkedPlan( + 'continuation_authority_unavailable', + 'workspace-bound continuation authority is unavailable', ); + } + let durableClaimState: ContinuationClaimStateV1 | ContinuationClaimStateV2 | undefined; + try { + durableClaimState = input.workspaceBoundary + ? await this.deps.readWorkspaceBoundContinuationClaimStateByBoundary?.( + plannedBoundaryDigest, + ) + : await this.deps.readContinuationClaimStateByBoundary?.(plannedBoundaryDigest); } catch { return parkedPlan( 'continuation_authority_unavailable', @@ -449,8 +484,11 @@ export class RuntimeContinuationPlanner { if (durableClaimState) { const claim = durableClaimState.claim; if ( - claim.boundaryDigest !== replay.plan.boundary.manifestDigest || + claim.boundaryDigest !== plannedBoundaryDigest || !isDeepStrictEqual(claim.boundary, replay.plan.boundary) || + (claim.protocol === 'continuation_claim_v2' && + !isDeepStrictEqual(claim.workspaceBoundary, input.workspaceBoundary)) || + (claim.protocol === 'continuation_claim_v1' && input.workspaceBoundary !== undefined) || claim.providerProjectionVersion !== replay.plan.providerProjectionVersion || claim.providerReplayDigest !== replay.plan.providerReplayDigest ) { @@ -500,12 +538,15 @@ export class RuntimeContinuationPlanner { ...(input.workspaceCheckpoint !== undefined ? { workspaceCheckpoint: input.workspaceCheckpoint } : {}), + ...(input.workspaceBoundary !== undefined + ? { workspaceBoundary: input.workspaceBoundary } + : {}), }); } private async classifyExistingClaim( sessionId: string, - state: ContinuationClaimStateV1, + state: ContinuationClaimStateV1 | ContinuationClaimStateV2, ): Promise { const { claim } = state; const detail = { @@ -621,13 +662,15 @@ export class RuntimeContinuationPlanner { }); const segments: ImmutableRuntimePrefixV1[] = [immediate]; const seen = new Set([sourceRunId]); - const v2Edges: Array<{ + const canonicalEdges: Array<{ childRunId: string; childRunHeader: AgentRunHeader; startEvent: RuntimeEvent; startKind: 'runtime_admission' | 'claim_repair'; + protocol: 'continuation_source_v2' | 'continuation_source_v3'; claimId: string; boundaryDigest: RuntimeBoundaryDigest; + replayManifestDigest: RuntimeBoundaryDigest; providerProjectionVersion: typeof PROVIDER_REPLAY_PROJECTION_VERSION; providerReplayDigest: RuntimeBoundaryDigest; }> = []; @@ -638,41 +681,46 @@ export class RuntimeContinuationPlanner { while (true) { const current = childRun.continuationSource; const start = childPrefix.events[0]?.actions?.continuationStart; - const currentV2 = - current && 'protocol' in current && current.protocol === 'continuation_source_v2' + const canonicalSource = + current && + 'protocol' in current && + (current.protocol === 'continuation_source_v2' || + current.protocol === 'continuation_source_v3') ? current : undefined; - if (start && !currentV2) { + if (start && !canonicalSource) { throw new RuntimeLineageError( 'runtime_lineage_start_mismatch', `canonical continuation-start cannot be downgraded to legacy lineage for ${childRunId}`, ); } - if (currentV2) { + if (canonicalSource) { if ( !start || - start.claimId !== currentV2.claimId || - start.boundaryDigest !== currentV2.boundaryDigest || - start.replayManifestDigest !== currentV2.replayManifestDigest || + start.claimId !== canonicalSource.claimId || + start.boundaryDigest !== canonicalSource.boundaryDigest || + start.replayManifestDigest !== canonicalSource.replayManifestDigest || start.immediateSource.sessionId !== sessionId || - start.immediateSource.invocationId !== currentV2.sourceInvocationId || - start.immediateSource.runId !== currentV2.sourceRunId || - start.immediateSource.turnId !== currentV2.sourceTurnId || - start.immediateSource.highWater !== currentV2.sourceRuntimeEventHighWater || - start.immediateSource.prefixDigest !== currentV2.sourcePrefixDigest + start.immediateSource.invocationId !== canonicalSource.sourceInvocationId || + start.immediateSource.runId !== canonicalSource.sourceRunId || + start.immediateSource.turnId !== canonicalSource.sourceTurnId || + start.immediateSource.highWater !== canonicalSource.sourceRuntimeEventHighWater || + start.immediateSource.prefixDigest !== canonicalSource.sourcePrefixDigest ) { throw new RuntimeLineageError( 'runtime_lineage_start_mismatch', `continuation-start does not authenticate lineage edge for ${childRunId}`, ); } - v2Edges.push({ + canonicalEdges.push({ childRunId, childRunHeader: childRun, startEvent: childPrefix.events[0]!, startKind: start.provenance, + protocol: canonicalSource.protocol, claimId: start.claimId, - boundaryDigest: currentV2.boundaryDigest, + boundaryDigest: canonicalSource.boundaryDigest, + replayManifestDigest: canonicalSource.replayManifestDigest, providerProjectionVersion: start.providerProjectionVersion, providerReplayDigest: start.providerReplayDigest, }); @@ -722,7 +770,8 @@ export class RuntimeContinuationPlanner { } if ( 'protocol' in current && - current.protocol === 'continuation_source_v2' && + (current.protocol === 'continuation_source_v2' || + current.protocol === 'continuation_source_v3') && current.sourcePrefixDigest !== prefix.prefixDigest ) { throw new RuntimeLineageError( @@ -736,7 +785,7 @@ export class RuntimeContinuationPlanner { childRunId = current.sourceRunId; depth += 1; } - for (const edge of v2Edges) { + for (const edge of canonicalEdges) { const childIndex = segments.findIndex((prefix) => prefix.identity.runId === edge.childRunId); if (childIndex <= 0) { throw new RuntimeLineageError( @@ -753,7 +802,7 @@ export class RuntimeContinuationPlanner { }); if ( edgeReplay.kind !== 'replayable' || - edgeReplay.plan.boundary.manifestDigest !== edge.boundaryDigest || + edgeReplay.plan.boundary.manifestDigest !== edge.replayManifestDigest || edgeReplay.plan.providerReplayDigest !== edge.providerReplayDigest ) { throw new RuntimeLineageError( @@ -761,15 +810,19 @@ export class RuntimeContinuationPlanner { `continuation provider replay changed before ${edge.childRunId}`, ); } - if (!this.deps.readContinuationClaimStateByBoundary) { + const readClaimState = + edge.protocol === 'continuation_source_v3' + ? this.deps.readWorkspaceBoundContinuationClaimStateByBoundary + : this.deps.readContinuationClaimStateByBoundary; + if (!readClaimState) { throw new RuntimeLineageError( 'continuation_authority_unavailable', `durable continuation authority is unavailable for ${edge.childRunId}`, ); } - let state: ContinuationClaimStateV1 | undefined; + let state: ContinuationClaimStateV1 | ContinuationClaimStateV2 | undefined; try { - state = await this.deps.readContinuationClaimStateByBoundary(edge.boundaryDigest); + state = await readClaimState(edge.boundaryDigest); } catch { throw new RuntimeLineageError( 'continuation_authority_unavailable', @@ -1169,6 +1222,7 @@ export function buildSafeBoundaryContinuationPlan( : {}), runtimeContext: [...modelRuntimeContext], ...(facts.continuationClaimId ? { claimId: facts.continuationClaimId } : {}), + ...(facts.workspaceBoundary ? { workspaceBoundary: facts.workspaceBoundary } : {}), ...(compositeReplay ? { boundary: compositeReplay.boundary, @@ -1188,6 +1242,7 @@ export function buildSafeBoundaryContinuationPlan( }, } : {}), + ...(facts.workspaceBoundary ? { workspaceBoundary: facts.workspaceBoundary } : {}), }, }, }; @@ -1466,7 +1521,7 @@ function hasMatchingCall( return call !== undefined && call.name === response.name; } -function claimTargetRunHeaderMatches(actual: AgentRunHeader, claim: ContinuationClaimV1): boolean { +function claimTargetRunHeaderMatches(actual: AgentRunHeader, claim: ContinuationClaim): boolean { const candidate = actual as unknown as Record; const expected = claim.targetRunHeader as unknown as Record; const immutable = (header: Record) => { @@ -1487,7 +1542,7 @@ function claimTargetRunHeaderMatches(actual: AgentRunHeader, claim: Continuation function continuationStartMatchesClaim( event: RuntimeEvent | undefined, - claim: ContinuationClaimV1, + claim: ContinuationClaim, startKind: ContinuationClaimStateV1['startKind'], ): boolean { const start = event?.actions?.continuationStart; diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index ee9027a7835..aaeda3e8558 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -125,9 +125,11 @@ import type { RootExecutionDescriptor, } from '@maka/core/agent-run'; import type { ArtifactRecord } from '@maka/core/artifacts'; -import type { ContinuationClaimV1 } from '@maka/core/runtime-boundary'; +import type { ContinuationClaim } from '@maka/core/runtime-boundary'; import type { ContinuationClaimStateV1, + ContinuationClaimStateV2, + RuntimeWorkspaceBoundContinuationAuthorityStore, RuntimeEventStore, RuntimeContinuationAuthorityStore, } from '@maka/core/runtime-event-store'; @@ -260,6 +262,21 @@ function runtimeContinuationAuthority( : undefined; } +function runtimeWorkspaceBoundContinuationAuthority( + store: RuntimeEventStore | undefined, +): RuntimeWorkspaceBoundContinuationAuthorityStore | undefined { + const candidate = store as Partial | undefined; + return candidate?.workspaceBoundContinuationAuthorityCapability === + 'runtime_workspace_bound_continuation_authority_v1' && + typeof candidate.claimWorkspaceBoundContinuation === 'function' && + typeof candidate.readWorkspaceBoundContinuationClaimStateByBoundary === 'function' && + typeof candidate.listWorkspaceBoundContinuationClaimsForRecovery === 'function' && + typeof candidate.commitWorkspaceBoundContinuationStart === 'function' && + typeof candidate.commitWorkspaceBoundContinuationRepairStart === 'function' + ? (candidate as RuntimeWorkspaceBoundContinuationAuthorityStore) + : undefined; +} + function runtimeCommitSinkFromEventStore( store: RuntimeEventStore | undefined, ): RuntimeCommitSink | undefined { @@ -1535,6 +1552,7 @@ export class SessionManager { continuationClaimRecovered = await this.recoverContinuationClaimsBeforeProvider( session.id, continuationAuthority, + runtimeWorkspaceBoundContinuationAuthority(this.deps.runtimeEventStore), policy, ); } catch (error) { @@ -2150,6 +2168,13 @@ export class SessionManager { if (!authority) throw new Error('Continuation authority is not configured'); return authority.readContinuationClaimStateByBoundary(boundaryDigest); }, + readWorkspaceBoundContinuationClaimStateByBoundary: async (boundaryDigest) => { + const authority = runtimeWorkspaceBoundContinuationAuthority(this.deps.runtimeEventStore); + if (!authority) { + throw new Error('Workspace-bound continuation authority is not configured'); + } + return authority.readWorkspaceBoundContinuationClaimStateByBoundary(boundaryDigest); + }, findExistingContinuation: async ( targetSessionId, sourceRunId, @@ -2245,12 +2270,17 @@ export class SessionManager { currentWorkspaceIdentity: observation.workspaceIdentity, backgroundOperationsSettled: observation.backgroundOperationsSettled, availableToolNames: observation.availableToolNames, + workspaceBoundaryRequirement: + header.toolProfile === 'managed-coding-v1' ? 'required' : 'optional', ...(input.expectedRuntimeEventHighWater !== undefined ? { expectedRuntimeEventHighWater: input.expectedRuntimeEventHighWater } : {}), ...(observation.workspaceCheckpoint ? { workspaceCheckpoint: observation.workspaceCheckpoint } : {}), + ...(observation.workspaceBoundary + ? { workspaceBoundary: observation.workspaceBoundary } + : {}), }); } @@ -4082,7 +4112,11 @@ export class SessionManager { const authority = runtimeContinuationAuthority(this.deps.runtimeEventStore); const abandonedBeforeProvider = sourceRun.failureClass === 'continuation_abandoned_before_provider_dispatch' - ? await isProvenRecoveredContinuationAbandonment(authority, sourceRun) + ? await isProvenRecoveredContinuationAbandonment( + authority, + runtimeWorkspaceBoundContinuationAuthority(this.deps.runtimeEventStore), + sourceRun, + ) : false; if ( sourceRun.status !== 'failed' || @@ -4286,6 +4320,13 @@ export class SessionManager { input.authority.readImmutableRuntimePrefix(prefixInput), readContinuationClaimStateByBoundary: (boundaryDigest) => input.authority.readContinuationClaimStateByBoundary(boundaryDigest), + readWorkspaceBoundContinuationClaimStateByBoundary: async (boundaryDigest) => { + const authority = runtimeWorkspaceBoundContinuationAuthority(this.deps.runtimeEventStore); + if (!authority) { + throw new Error('Workspace-bound continuation authority is not configured'); + } + return authority.readWorkspaceBoundContinuationClaimStateByBoundary(boundaryDigest); + }, findExistingContinuation: async (_sessionId, sourceRunId, sourceRuntimeEventHighWater) => input.runs.find( (run) => @@ -4306,7 +4347,10 @@ export class SessionManager { currentWorkspaceIdentity: safety.workspaceIdentity, backgroundOperationsSettled: safety.backgroundOperationsSettled, availableToolNames: input.availableToolNames, + workspaceBoundaryRequirement: + targetHeader.toolProfile === 'managed-coding-v1' ? 'required' : 'optional', ...(safety.workspaceCheckpoint ? { workspaceCheckpoint: safety.workspaceCheckpoint } : {}), + ...(safety.workspaceBoundary ? { workspaceBoundary: safety.workspaceBoundary } : {}), }); if (retryPlan.disposition !== 'continue' || !retryPlan.continuation) { throw new Error( @@ -4666,9 +4710,17 @@ export class SessionManager { const continuationAuthority = runtimeContinuationAuthority(this.deps.runtimeEventStore); if (continuationAuthority) { - const claimState = ( - await continuationAuthority.listContinuationClaimsForRecovery(input.sessionId) - ).find( + const workspaceAuthority = runtimeWorkspaceBoundContinuationAuthority( + this.deps.runtimeEventStore, + ); + const claimState = [ + ...(await continuationAuthority.listContinuationClaimsForRecovery(input.sessionId)), + ...(workspaceAuthority + ? await workspaceAuthority.listWorkspaceBoundContinuationClaimsForRecovery( + input.sessionId, + ) + : []), + ].find( (candidate) => candidate.claim.target.runId === input.runId || candidate.claim.target.turnId === input.turnId, @@ -5571,13 +5623,30 @@ export class SessionManager { private async recoverContinuationClaimsBeforeProvider( sessionId: string, authority: RuntimeContinuationAuthorityStore, + workspaceAuthority: RuntimeWorkspaceBoundContinuationAuthorityStore | undefined, _policy: RecoveryPolicy, ): Promise { if (!this.deps.runStore) return false; - const states = await authority.listContinuationClaimsForRecovery(sessionId); + const states: Array = [ + ...(await authority.listContinuationClaimsForRecovery(sessionId)), + ...(workspaceAuthority + ? await workspaceAuthority.listWorkspaceBoundContinuationClaimsForRecovery(sessionId) + : []), + ]; let recovered = false; for (const initialState of states) { const { claim } = initialState; + if (claim.protocol === 'continuation_claim_v2') { + if (!workspaceAuthority || !this.deps.inspectContinuationSafety) { + throw new Error('Workspace-bound continuation recovery authority is unavailable'); + } + const observation = await this.deps.inspectContinuationSafety(sessionId); + if (!isDeepStrictEqual(observation.workspaceBoundary, claim.workspaceBoundary)) { + throw new Error( + `Workspace-bound continuation claim ${claim.claimId} no longer matches accepted head`, + ); + } + } let run: AgentRunHeader; try { run = await this.deps.runStore.readRun(sessionId, claim.target.runId); @@ -5595,9 +5664,13 @@ export class SessionManager { } recovered = true; } - let state = - (await authority.readContinuationClaimStateByBoundary(claim.boundaryDigest)) ?? - initialState; + const readState = () => + claim.protocol === 'continuation_claim_v2' + ? workspaceAuthority!.readWorkspaceBoundContinuationClaimStateByBoundary( + claim.boundaryDigest, + ) + : authority.readContinuationClaimStateByBoundary(claim.boundaryDigest); + let state = (await readState()) ?? initialState; if ( state.startEventId ? !claimTargetRunHeaderIsCompatible(run, claim.targetRunHeader) @@ -5620,9 +5693,16 @@ export class SessionManager { ); } const repairStart = buildContinuationRepairStartEvent(claim); - await authority.commitContinuationRepairStart({ claim, event: repairStart }); + if (claim.protocol === 'continuation_claim_v2') { + await workspaceAuthority!.commitWorkspaceBoundContinuationRepairStart({ + claim, + event: repairStart, + }); + } else { + await authority.commitContinuationRepairStart({ claim, event: repairStart }); + } state = - (await authority.readContinuationClaimStateByBoundary(claim.boundaryDigest)) ?? + (await readState()) ?? (() => { throw new Error(`Continuation claim ${claim.claimId} disappeared during repair`); })(); @@ -6007,7 +6087,7 @@ function continuationRepairEventId( .slice(0, 32)}`; } -function buildContinuationRepairStartEvent(claim: ContinuationClaimV1): RuntimeEvent { +function buildContinuationRepairStartEvent(claim: ContinuationClaim): RuntimeEvent { const source = claim.boundary.segments.at(-1)!; return { id: continuationRepairEventId('start', claim.claimId), @@ -6045,7 +6125,7 @@ function assertClaimOwnsHostedLinkedChildAdmission( runId: string; execution: Exclude; }, - claim: ContinuationClaimV1, + claim: ContinuationClaim, ): void { if ( input.execution.kind !== 'linked_child_resume' && @@ -6063,10 +6143,11 @@ function assertClaimOwnsHostedLinkedChildAdmission( const header = claim.targetRunHeader; const source = claim.boundary.segments.at(-1)!; const continuationSource = header.continuationSource; - const continuationSourceV2 = + const canonicalContinuationSource = continuationSource !== undefined && 'protocol' in continuationSource && - continuationSource.protocol === 'continuation_source_v2' + (continuationSource.protocol === 'continuation_source_v2' || + continuationSource.protocol === 'continuation_source_v3') ? continuationSource : undefined; if ( @@ -6079,10 +6160,10 @@ function assertClaimOwnsHostedLinkedChildAdmission( header.agentName !== input.execution.agentName || source.identity.sessionId !== input.sessionId || source.identity.runId !== input.execution.sourceRunId || - !continuationSourceV2 || - continuationSourceV2.claimId !== claim.claimId || - continuationSourceV2.boundaryDigest !== claim.boundaryDigest || - continuationSourceV2.sourceRunId !== input.execution.sourceRunId + !canonicalContinuationSource || + canonicalContinuationSource.claimId !== claim.claimId || + canonicalContinuationSource.boundaryDigest !== claim.boundaryDigest || + canonicalContinuationSource.sourceRunId !== input.execution.sourceRunId ) { throw new Error('Linked child admission continuation claim identity is inconsistent'); } @@ -6099,28 +6180,35 @@ function assertClaimOwnsHostedLinkedChildAdmission( async function isProvenRecoveredContinuationAbandonment( authority: RuntimeContinuationAuthorityStore | undefined, + workspaceAuthority: RuntimeWorkspaceBoundContinuationAuthorityStore | undefined, run: AgentRunHeader, ): Promise { const continuationSource = run.continuationSource; - const continuationSourceV2 = + const canonicalContinuationSource = continuationSource !== undefined && 'protocol' in continuationSource && - continuationSource.protocol === 'continuation_source_v2' + (continuationSource.protocol === 'continuation_source_v2' || + continuationSource.protocol === 'continuation_source_v3') ? continuationSource : undefined; if ( !authority || - !continuationSourceV2 || + !canonicalContinuationSource || run.failureClass !== 'continuation_abandoned_before_provider_dispatch' ) { return false; } - let state: ContinuationClaimStateV1 | undefined; + let state: ContinuationClaimStateV1 | ContinuationClaimStateV2 | undefined; let events: RuntimeEvent[]; try { - state = await authority.readContinuationClaimStateByBoundary( - continuationSourceV2.boundaryDigest, - ); + state = + canonicalContinuationSource.protocol === 'continuation_source_v3' + ? await workspaceAuthority?.readWorkspaceBoundContinuationClaimStateByBoundary( + canonicalContinuationSource.boundaryDigest, + ) + : await authority.readContinuationClaimStateByBoundary( + canonicalContinuationSource.boundaryDigest, + ); events = await authority.readImmutableRuntimeEvents(run.sessionId, run.runId); } catch { return false; @@ -6140,7 +6228,7 @@ async function isProvenRecoveredContinuationAbandonment( }); const terminalFact = classifyRuntimeEventTerminalFact(run, events).fact; return ( - claim.claimId === continuationSourceV2.claimId && + claim.claimId === canonicalContinuationSource.claimId && claim.target.sessionId === run.sessionId && claim.target.invocationId === run.invocationId && claim.target.runId === run.runId && diff --git a/packages/storage/src/__tests__/sqlite-runtime-crash.test.ts b/packages/storage/src/__tests__/sqlite-runtime-crash.test.ts index dbc96eb57d0..5de618b2c5c 100644 --- a/packages/storage/src/__tests__/sqlite-runtime-crash.test.ts +++ b/packages/storage/src/__tests__/sqlite-runtime-crash.test.ts @@ -244,6 +244,24 @@ if (childMode) { assert.equal((await store.claimWorkspaceBoundContinuation({ claim })).kind, 'existing'); }); }); + + it('rolls back a workspace-bound continuation start when killed inside its transaction', { + timeout: 30_000, + }, async () => { + await withKilledChild('inside_workspace_continuation_start', async (store) => { + const claim = workspaceBoundContinuationClaim(); + const state = await store.readWorkspaceBoundContinuationClaimStateByBoundary( + claim.boundaryDigest, + ); + assert.ok(state); + assert.equal(state.startEventId, undefined); + const committed = await store.commitWorkspaceBoundContinuationStart({ + claim, + event: workspaceBoundContinuationStartEvent(claim), + }); + assert.equal(committed.created, true); + }); + }); }); } @@ -337,6 +355,12 @@ async function runCrashChild(mode: string): Promise { ) { blockUntilKilled(); } + if ( + point === 'after_continuation_start_insert' && + mode === 'inside_workspace_continuation_start' + ) { + blockUntilKilled(); + } }; const store = createSqliteRuntimeStore(dbPath, { failpoint }); if ( @@ -346,7 +370,8 @@ async function runCrashChild(mode: string): Promise { mode === 'inside_workspace_successor' || mode === 'after_workspace_successor_commit' || mode === 'inside_workspace_continuation_claim' || - mode === 'after_workspace_continuation_claim_commit' + mode === 'after_workspace_continuation_claim_commit' || + mode === 'inside_workspace_continuation_start' ) { bindWorkspaceBaselineAuthorityStoreRootInternal(store, 'a'.repeat(64)); await commitWorkspaceBaselineInternal(store, workspaceBaselineInput()); @@ -362,11 +387,19 @@ async function runCrashChild(mode: string): Promise { } if ( mode === 'inside_workspace_continuation_claim' || - mode === 'after_workspace_continuation_claim_commit' + mode === 'after_workspace_continuation_claim_commit' || + mode === 'inside_workspace_continuation_start' ) { await persistWorkspaceContinuationSource(store); - await store.claimWorkspaceBoundContinuation({ claim: workspaceBoundContinuationClaim() }); + const claim = workspaceBoundContinuationClaim(); + await store.claimWorkspaceBoundContinuation({ claim }); if (mode === 'after_workspace_continuation_claim_commit') blockUntilKilled(); + if (mode === 'inside_workspace_continuation_start') { + await store.commitWorkspaceBoundContinuationStart({ + claim, + event: workspaceBoundContinuationStartEvent(claim), + }); + } } throw new Error(`Workspace baseline crash mode ${mode} missed its failpoint`); } @@ -517,6 +550,37 @@ function workspaceBoundContinuationClaim(): ContinuationClaimV2 { }; } +function workspaceBoundContinuationStartEvent(claim: ContinuationClaimV2): RuntimeEvent { + const source = claim.boundary.segments.at(-1)!; + return { + id: 'workspace-continuation-start', + ...claim.target, + ts: claim.claimedAt + 1, + partial: false, + role: 'system', + author: 'system', + actions: { + continuationStart: { + protocol: 'continuation_start_v2', + provenance: 'runtime_admission', + claimId: claim.claimId, + boundaryDigest: claim.boundaryDigest, + immediateSource: { + sessionId: source.identity.sessionId, + invocationId: source.identity.invocationId, + runId: source.identity.runId, + turnId: source.identity.turnId, + highWater: source.position.lastEventSeq, + prefixDigest: source.prefixDigest, + }, + replayManifestDigest: claim.boundary.manifestDigest, + providerProjectionVersion: claim.providerProjectionVersion, + providerReplayDigest: claim.providerReplayDigest, + }, + }, + }; +} + function workspaceContinuationSourcePrefix() { const identity = { sessionId: 'continuation-session', diff --git a/packages/storage/src/__tests__/sqlite-runtime-store.test.ts b/packages/storage/src/__tests__/sqlite-runtime-store.test.ts index 36962b4c16b..f5e0676b166 100644 --- a/packages/storage/src/__tests__/sqlite-runtime-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-runtime-store.test.ts @@ -31,6 +31,7 @@ import { createRuntimeBoundaryCursor, digestWorkspaceBoundContinuationBoundary, runtimePrefixSegment, + type ContinuationClaim, type ContinuationClaimV2, type ContinuationClaimV1, type ImmutableRuntimePrefixV1, @@ -908,6 +909,62 @@ describe('SqliteRuntimeStore', () => { }); }); + it('commits a workspace-bound continuation start only through its dedicated authority', async () => { + await withStore(async (store) => { + const workspaceBoundary = await openContinuationWorkspaceBoundary(store); + const claim = workspaceBoundContinuationClaim(workspaceBoundary); + await persistImmutablePrefix(store, continuationSourcePrefix()); + assert.equal((await store.claimWorkspaceBoundContinuation({ claim })).kind, 'acquired'); + const event = continuationStartEvent(claim); + + const committed = await store.commitWorkspaceBoundContinuationStart({ claim, event }); + const retry = await store.commitWorkspaceBoundContinuationStart({ claim, event }); + const state = await store.readWorkspaceBoundContinuationClaimStateByBoundary( + claim.boundaryDigest, + ); + + assert.equal(committed.created, true); + assert.equal(retry.created, false); + assert.deepEqual(state, { + claim, + startEventId: event.id, + startKind: 'runtime_admission', + }); + assert.equal(await store.readContinuationClaimByBoundary(claim.boundaryDigest), undefined); + }); + }); + + it('revalidates the accepted workspace boundary when committing continuation start', async () => { + await withStore(async (store, dbPath) => { + const workspaceBoundary = await openContinuationWorkspaceBoundary(store); + const claim = workspaceBoundContinuationClaim(workspaceBoundary); + await persistImmutablePrefix(store, continuationSourcePrefix()); + assert.equal((await store.claimWorkspaceBoundContinuation({ claim })).kind, 'acquired'); + + const tamper = new DatabaseSync(dbPath); + try { + tamper + .prepare( + `UPDATE runtime_workspace_heads + SET revision = revision + 1 + WHERE workspace_id = ? AND workspace_epoch_id = ?`, + ) + .run(workspaceBoundary.workspaceId, workspaceBoundary.workspaceEpochId); + } finally { + tamper.close(); + } + + await assert.rejects( + store.commitWorkspaceBoundContinuationStart({ + claim, + event: continuationStartEvent(claim), + }), + /workspace version projection is incomplete|workspace boundary no longer matches accepted authority/i, + ); + assert.deepEqual(await store.readRuntimeEvents(claim.target.sessionId, claim.target.runId), []); + }); + }); + it('rejects a workspace-bound claim when caller evidence differs from accepted authority', async () => { await withStore(async (store) => { const workspaceBoundary = await openContinuationWorkspaceBoundary(store); @@ -2298,7 +2355,7 @@ async function persistImmutablePrefix( } function continuationStartEvent( - claim: ContinuationClaimV1, + claim: ContinuationClaim, overrides: { id?: string; provenance?: 'runtime_admission' | 'claim_repair'; diff --git a/packages/storage/src/__tests__/workspace-version-authority-persistence.test.ts b/packages/storage/src/__tests__/workspace-version-authority-persistence.test.ts index 85b1ff504c6..c77f7f6385c 100644 --- a/packages/storage/src/__tests__/workspace-version-authority-persistence.test.ts +++ b/packages/storage/src/__tests__/workspace-version-authority-persistence.test.ts @@ -537,7 +537,7 @@ describe('workspace version persistence authority', () => { const upgraded = createSqliteRuntimeStore(dbPath); bindWorkspaceBaselineAuthorityStoreRootInternal(upgraded, TEST_STORAGE_ROOT_ID); try { - assert.equal(upgraded.schemaVersion(), 14); + assert.equal(upgraded.schemaVersion(), 15); assert.equal( ( await upgraded.readWorkspaceHead( @@ -1212,6 +1212,8 @@ function recreateWorkspaceTablesAsSchema12(database: DatabaseSync): void { DROP TABLE runtime_workspace_heads_schema_13; DROP TABLE runtime_workspace_versions_schema_13; DROP TABLE runtime_managed_mutation_reservations; + DELETE FROM runtime_capabilities + WHERE capability = 'runtime_workspace_bound_continuation_authority'; PRAGMA user_version = 12; COMMIT; PRAGMA foreign_keys = ON; diff --git a/packages/storage/src/execution-stores-workspace-authority-internal.ts b/packages/storage/src/execution-stores-workspace-authority-internal.ts index 9261c08f78a..46a3a801158 100644 --- a/packages/storage/src/execution-stores-workspace-authority-internal.ts +++ b/packages/storage/src/execution-stores-workspace-authority-internal.ts @@ -19,6 +19,10 @@ import type { RuntimeEvent } from '@maka/core/runtime-event'; import type { RuntimeWorkspaceVersionAuthorityStore } from '@maka/core/runtime-event-store'; +import type { + ManagedWorkspaceContinuationBoundaryV1, + RuntimeBoundaryDigest, +} from '@maka/core/runtime-boundary'; import type { WorkspaceBaselineAuthorityInput, WorkspaceBaselineCommitResult, @@ -31,6 +35,7 @@ import { commitWorkspaceBaselineInternal, commitWorkspaceSuccessorInternal, readActiveManagedMutationInternal, + readWorkspaceContinuationBoundaryInternal, type ManagedMutationReservationRecordV1, type ManagedMutationTerminalCommitInput, type ManagedMutationTerminalCommitResult, @@ -60,6 +65,11 @@ export interface ExecutionStoresWorkspaceMutationAuthorityInternal { commitTerminal( input: ManagedMutationTerminalCommitInput, ): Promise; + readContinuationBoundary( + workspaceId: string, + workspaceEpochId: string, + executionProfileDigest: RuntimeBoundaryDigest, + ): Promise; } interface RegisteredWorkspaceAuthority { @@ -103,5 +113,16 @@ export function requireExecutionStoresWorkspaceMutationAuthorityInternal( commitWorkspaceSuccessorInternal(authority, input), commitTerminal: (input: ManagedMutationTerminalCommitInput) => commitManagedMutationTerminalInternal(authority, input), + readContinuationBoundary: ( + workspaceId: string, + workspaceEpochId: string, + executionProfileDigest: RuntimeBoundaryDigest, + ) => + readWorkspaceContinuationBoundaryInternal( + authority, + workspaceId, + workspaceEpochId, + executionProfileDigest, + ), }); } diff --git a/packages/storage/src/sqlite-runtime-store.ts b/packages/storage/src/sqlite-runtime-store.ts index 01c98876c71..0d2f27feffb 100644 --- a/packages/storage/src/sqlite-runtime-store.ts +++ b/packages/storage/src/sqlite-runtime-store.ts @@ -59,6 +59,7 @@ import { TOOL_RECOVERY_BUNDLE_CAPABILITY_V1, type ContinuationClaimResult, type ContinuationClaimStateV1, + type ContinuationClaimStateV2, type RuntimeContinuationAuthorityStore, type RuntimeRecoveryBundleCommit, type RuntimeRecoveryBundleStore, @@ -1052,6 +1053,12 @@ export class SqliteRuntimeStore async readWorkspaceBoundContinuationClaimByBoundary( boundaryDigest: RuntimeBoundaryDigest, ): Promise | undefined> { + return (await this.readWorkspaceBoundContinuationClaimStateByBoundary(boundaryDigest))?.claim; + } + + async readWorkspaceBoundContinuationClaimStateByBoundary( + boundaryDigest: RuntimeBoundaryDigest, + ): Promise { if (!/^sha256:[0-9a-f]{64}$/.test(boundaryDigest)) { throw new Error('Invalid continuation boundary digest'); } @@ -1060,11 +1067,7 @@ export class SqliteRuntimeStore boundaryDigest, ); if (!row) return undefined; - const claim = decodeContinuationClaimRow(row); - if (claim.protocol !== 'continuation_claim_v2') { - throw new Error('Workspace-bound continuation claim protocol mismatch'); - } - return claim; + return this.decodeWorkspaceBoundContinuationClaimStateRow(row); } async readContinuationClaimStateByBoundary( @@ -1113,6 +1116,13 @@ export class SqliteRuntimeStore return rows.map((row) => this.decodeContinuationClaimStateRow(row)); } + async listWorkspaceBoundContinuationClaimsForRecovery( + sessionId: string, + ): Promise { + const rows = this.readContinuationClaimRowsForSession(sessionId, 2); + return rows.map((row) => this.decodeWorkspaceBoundContinuationClaimStateRow(row)); + } + async commitContinuationStart(input: { claim: ContinuationClaimV1; event: RuntimeEvent; @@ -1127,16 +1137,30 @@ export class SqliteRuntimeStore return this.commitContinuationStartOfKind(input, 'claim_repair'); } + async commitWorkspaceBoundContinuationStart(input: { + claim: Extract; + event: RuntimeEvent; + }): Promise { + return this.commitContinuationStartOfKind(input, 'runtime_admission'); + } + + async commitWorkspaceBoundContinuationRepairStart(input: { + claim: Extract; + event: RuntimeEvent; + }): Promise { + return this.commitContinuationStartOfKind(input, 'claim_repair'); + } + private commitContinuationStartOfKind( input: { - claim: ContinuationClaimV1; + claim: ContinuationClaim; event: RuntimeEvent; }, startKind: 'runtime_admission' | 'claim_repair', ): ToolCommitResult { const claim = decodeContinuationClaim(input.claim); - if (claim.protocol !== 'continuation_claim_v1') { - throw new Error('Legacy continuation start requires a v1 claim'); + if (claim.protocol !== input.claim.protocol) { + throw new Error('Continuation start claim protocol changed during decoding'); } const event = canonicalizeRuntimeEventForStorage(input.event); assertNoReservedWorkspaceAuthorityAppend(event); @@ -1150,6 +1174,12 @@ export class SqliteRuntimeStore if (!isDeepStrictEqual(storedClaim, claim)) { throw new Error('Continuation start claim identity conflict'); } + if (claim.protocol === 'continuation_claim_v2') { + // Claim acquisition and continuation start are separate durable boundaries. + // Re-observe the accepted head in this transaction so a stale managed claim + // can never start merely because an in-process execution lease survived. + this.assertContinuationWorkspaceBoundaryMatchesAuthority(claim.workspaceBoundary); + } if (row.start_event_id) { if (row.start_event_id !== event.id || row.start_kind !== startKind) { throw new Error('Continuation claim already has a different start event'); @@ -1698,9 +1728,33 @@ export class SqliteRuntimeStore (rootId) => this.#adoptWorkspaceStorageRoot(rootId), readWorkspaceHead, (workspaceInstanceId) => this.#readActiveManagedMutation(workspaceInstanceId), + (workspaceId, workspaceEpochId, rootId, executionProfileDigest) => + this.#readWorkspaceContinuationBoundary( + workspaceId, + workspaceEpochId, + rootId, + executionProfileDigest, + ), ); } + async #readWorkspaceContinuationBoundary( + workspaceId: string, + workspaceEpochId: string, + rootId: string, + executionProfileDigest: `sha256:${string}`, + ): Promise { + return this.readTransaction(() => { + this.#assertWorkspaceStorageRootBinding(rootId); + return this.currentWorkspaceContinuationBoundarySync( + workspaceId, + workspaceEpochId, + rootId, + executionProfileDigest, + ); + }); + } + async #readActiveManagedMutation( workspaceInstanceId: string, ): Promise< @@ -3202,18 +3256,44 @@ export class SqliteRuntimeStore .all() as unknown as ContinuationClaimStorageRow[]; } + private readContinuationClaimRowsForSession( + sessionId: string, + protocolVersion: 1 | 2, + ): ContinuationClaimStorageRow[] { + return this.db + .prepare(` + SELECT + claim_id, + source_session_id, + source_invocation_id, + source_run_id, + source_turn_id, + source_event_high_water, + source_prefix_digest, + boundary_digest, + boundary_json, + workspace_boundary_json, + provider_projection_version, + provider_replay_digest, + target_session_id, + target_invocation_id, + target_run_id, + target_turn_id, + target_run_header_json, + claimed_at, + start_event_id, + start_kind, + protocol_version + FROM runtime_continuation_claims + WHERE target_session_id = ? AND protocol_version = ? + ORDER BY claimed_at ASC, claim_id ASC + `) + .all(sessionId, protocolVersion) as unknown as ContinuationClaimStorageRow[]; + } + private assertContinuationAuthorityIntegrity(): void { for (const row of this.readContinuationClaimRows()) { - const claim = decodeContinuationClaimRow(row); - if (claim.protocol === 'continuation_claim_v2') { - if (row.start_event_id !== null || row.start_kind !== null) { - throw new Error( - `Workspace-bound continuation claim has an unsupported start event for ${claim.claimId}`, - ); - } - continue; - } - this.decodeContinuationClaimStateRow(row); + this.decodeAnyContinuationClaimStateRow(row); } } @@ -3254,10 +3334,29 @@ export class SqliteRuntimeStore private decodeContinuationClaimStateRow( row: ContinuationClaimStorageRow, ): ContinuationClaimStateV1 { - const claim = decodeContinuationClaimRow(row); - if (claim.protocol !== 'continuation_claim_v1') { - throw new Error(`Legacy continuation reader encountered ${claim.protocol}`); + const state = this.decodeAnyContinuationClaimStateRow(row); + if (state.claim.protocol !== 'continuation_claim_v1') { + throw new Error(`Legacy continuation reader encountered ${state.claim.protocol}`); } + return state as ContinuationClaimStateV1; + } + + private decodeWorkspaceBoundContinuationClaimStateRow( + row: ContinuationClaimStorageRow, + ): ContinuationClaimStateV2 { + const state = this.decodeAnyContinuationClaimStateRow(row); + if (state.claim.protocol !== 'continuation_claim_v2') { + throw new Error(`Workspace-bound continuation reader encountered ${state.claim.protocol}`); + } + return state as ContinuationClaimStateV2; + } + + private decodeAnyContinuationClaimStateRow(row: ContinuationClaimStorageRow): { + claim: ContinuationClaim; + startEventId?: string; + startKind?: 'runtime_admission' | 'claim_repair'; + } { + const claim = decodeContinuationClaimRow(row); if (!row.start_event_id) { if (row.start_kind !== null) { throw new Error(`Continuation claim start kind exists without event for ${claim.claimId}`); @@ -3335,21 +3434,35 @@ export class SqliteRuntimeStore throw new Error('Continuation workspace boundary storage-root identity conflict'); } + const expected = this.currentWorkspaceContinuationBoundarySync( + boundary.workspaceId, + boundary.workspaceEpochId, + storageRoot.root_id, + boundary.executionProfileDigest, + ); + if (!expected || !isDeepStrictEqual(boundary, expected)) { + throw new Error('Continuation workspace boundary no longer matches accepted authority'); + } + } + + private currentWorkspaceContinuationBoundarySync( + workspaceId: string, + workspaceEpochId: string, + storageRootId: string, + executionProfileDigest: `sha256:${string}`, + ): ManagedWorkspaceContinuationBoundaryV1 | undefined { const authority = this.readCanonicalWorkspaceAuthoritySync(); this.assertWorkspaceProjectionsMatchSync(authority); const epoch = authority.baselines.find( (candidate) => - candidate.epoch.workspaceId === boundary.workspaceId && - candidate.epoch.workspaceEpochId === boundary.workspaceEpochId, + candidate.epoch.workspaceId === workspaceId && + candidate.epoch.workspaceEpochId === workspaceEpochId, ); const head = authority.heads.find( (candidate) => - candidate.workspaceId === boundary.workspaceId && - candidate.workspaceEpochId === boundary.workspaceEpochId, + candidate.workspaceId === workspaceId && candidate.workspaceEpochId === workspaceEpochId, ); - if (!epoch || !head) { - throw new Error('Continuation workspace boundary is unavailable'); - } + if (!epoch || !head) return undefined; const baseline = authority.baselines.find( (candidate) => @@ -3369,16 +3482,16 @@ export class SqliteRuntimeStore epoch.epoch.policyHash !== version.policyHash || workspaceMutationPolicyHashV1( epoch.epoch.materializationProfileDigest, - boundary.executionProfileDigest, + executionProfileDigest, ) !== epoch.epoch.policyHash || (successor !== undefined && - successor.successor.executionProfileDigest !== boundary.executionProfileDigest) + successor.successor.executionProfileDigest !== executionProfileDigest) ) { throw new Error('Continuation workspace boundary execution profile conflict'); } - const expected: ManagedWorkspaceContinuationBoundaryV1 = { + return { protocol: 'managed_workspace_continuation_boundary_v1', - storageRootId: storageRoot.root_id, + storageRootId, repositoryId: epoch.epoch.repositoryId, workspaceId: epoch.epoch.workspaceId, workspaceEpochId: epoch.epoch.workspaceEpochId, @@ -3391,11 +3504,8 @@ export class SqliteRuntimeStore treeOid: head.treeOid, materializationProfileDigest: epoch.epoch.materializationProfileDigest, policyHash: version.policyHash, - executionProfileDigest: boundary.executionProfileDigest, + executionProfileDigest, }; - if (!isDeepStrictEqual(boundary, expected)) { - throw new Error('Continuation workspace boundary no longer matches accepted authority'); - } } private assertToolLedgerTransition( @@ -4227,7 +4337,7 @@ function assertRecoveryAuthorityCapability(db: DatabaseSync): void { } function assertContinuationStartEvent( - claim: ContinuationClaimV1, + claim: ContinuationClaim, event: RuntimeEvent, startKind: 'runtime_admission' | 'claim_repair', ): void { diff --git a/packages/storage/src/workspace-version-authority-internal.ts b/packages/storage/src/workspace-version-authority-internal.ts index 8369ce4713b..a0193a77f36 100644 --- a/packages/storage/src/workspace-version-authority-internal.ts +++ b/packages/storage/src/workspace-version-authority-internal.ts @@ -24,6 +24,7 @@ import type { WorkspaceSuccessorAuthorityInput, } from '@maka/core/workspace-version-authority'; import type { RuntimeEvent } from '@maka/core/runtime-event'; +import type { ManagedWorkspaceContinuationBoundaryV1 } from '@maka/core/runtime-boundary'; import { lstatSync } from 'node:fs'; import { realpath } from 'node:fs/promises'; import { basename, dirname, join, normalize, resolve } from 'node:path'; @@ -88,6 +89,12 @@ export interface ManagedMutationReservationRecordV1 { type ManagedMutationReservationReader = ( workspaceInstanceId: string, ) => Promise; +type WorkspaceContinuationBoundaryReader = ( + workspaceId: string, + workspaceEpochId: string, + rootId: string, + executionProfileDigest: `sha256:${string}`, +) => Promise; interface WorkspaceBaselineAuthorityRegistration { readonly writer: WorkspaceBaselineAuthorityWriter; @@ -95,6 +102,7 @@ interface WorkspaceBaselineAuthorityRegistration { readonly terminalWriter: ManagedMutationTerminalWriter; readonly readHead: WorkspaceHeadReader; readonly readActiveManagedMutation: ManagedMutationReservationReader; + readonly readContinuationBoundary: WorkspaceContinuationBoundaryReader; readonly bindStorageRoot: WorkspaceStorageRootBinder; readonly adoptStorageRoot: WorkspaceStorageRootAdopter; readonly databasePath: string; @@ -117,6 +125,7 @@ export function registerWorkspaceBaselineAuthorityWriterInternal( adoptStorageRoot: WorkspaceStorageRootAdopter, readHead: WorkspaceHeadReader, readActiveManagedMutation: ManagedMutationReservationReader, + readContinuationBoundary: WorkspaceContinuationBoundaryReader, ): void { if (workspaceBaselineAuthorityWriters.has(store)) { throw new Error('Workspace baseline authority writer is already registered'); @@ -128,6 +137,7 @@ export function registerWorkspaceBaselineAuthorityWriterInternal( terminalWriter, readHead, readActiveManagedMutation, + readContinuationBoundary, bindStorageRoot, adoptStorageRoot, databasePath: resolvedDatabasePath, @@ -154,6 +164,25 @@ export function readWorkspaceHeadInternal( return registration.readHead(workspaceId, workspaceEpochId); } +export function readWorkspaceContinuationBoundaryInternal( + store: object, + workspaceId: string, + workspaceEpochId: string, + executionProfileDigest: `sha256:${string}`, +): Promise { + const registration = workspaceBaselineAuthorityWriters.get(store); + if (!registration) throw new Error('Workspace continuation boundary reader is unavailable'); + if (!registration.boundRootId) { + throw new Error('Workspace continuation boundary store has no durable storage-root binding'); + } + return registration.readContinuationBoundary( + workspaceId, + workspaceEpochId, + registration.boundRootId, + executionProfileDigest, + ); +} + /** * Storage-internal authority seam. This module is deliberately absent from the * @maka/storage package exports. Production composition reaches it only through