From bbd769c74e2b1ced62859a89ed08855bfba1692f Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 15:23:49 +0800 Subject: [PATCH] fix(runtime-host): recover approved guest regeneration Admit durable Guest authorization on regenerate roots, and let explicit managed-Host repair safely replace a Host that dies while preparing retirement. Generated-by: OpenAI Codex --- .../runtime-host-local-operator.test.ts | 37 +++++++++ .../src/main/runtime-host-local-operator.ts | 2 +- ...runtime-host-lifecycle-transaction.test.ts | 79 ++++++++++++++++++- .../src/runtime-host-lifecycle-transaction.ts | 56 +++++++++---- .../__tests__/root-turn-coordinator.test.ts | 58 ++++++++++++++ .../regenerate-root-admission.test.ts | 9 +++ packages/storage/src/agent-run-store.ts | 8 +- 7 files changed, 229 insertions(+), 20 deletions(-) diff --git a/apps/desktop/src/main/__tests__/runtime-host-local-operator.test.ts b/apps/desktop/src/main/__tests__/runtime-host-local-operator.test.ts index f32962bf26..89ef5a27bc 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-local-operator.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-local-operator.test.ts @@ -185,6 +185,7 @@ test('local update runs the selected package against the exact managed deploymen deploymentId, }, expectedHost: { hostEpoch: 'older-host', pid: 42 }, + allowManualUpdate: true, }, (phase) => phases.push(phase), ); @@ -200,12 +201,48 @@ test('local update runs the selected package against the exact managed deploymen '--expected-root-path', '/tmp/maka/root', '--expected-root-id', 'a'.repeat(64), '--expected-deployment-id', deploymentId, + '--allow-manual-update', ]); assert.equal( environment?.[RUNTIME_HOST_OPERATOR_PROJECT_DIRECTORY_CONFIGURATION_REQUEST_ENV], '1', ); assert.deepEqual(phases, ['staging']); + + const developmentIntegrity = `sha512-${Buffer.alloc(64, 9).toString('base64')}`; + await operator.runUpdate( + { + setupPackage: { + kind: 'development_archive', + path: '/tmp/maka-agent-development.tgz', + integrity: developmentIntegrity, + }, + target: { + serviceId: 'a'.repeat(64), + rootPath: '/tmp/maka/root', + rootId: 'a'.repeat(64), + deploymentId, + }, + allowManualUpdate: true, + allowInterruptActiveTasks: true, + }, + () => undefined, + ); + + assert.deepEqual(args, [ + 'exec', '--yes', '--package', '/tmp/maka-agent-development.tgz', '--', + 'maka', 'runtime-host', 'service', 'update', '--framed', + '--managed-root-id', 'a'.repeat(64), + '--expected-service-id', 'a'.repeat(64), + '--expected-root-path', '/tmp/maka/root', + '--expected-root-id', 'a'.repeat(64), + '--expected-deployment-id', deploymentId, + '--allow-interrupt-active-tasks', + ]); + assert.equal( + environment?.[RUNTIME_HOST_SETUP_SOURCE_PACKAGE_INTEGRITY_ENV], + developmentIntegrity, + ); }); test('local Peer Mesh join keeps invitations off argv and accepts bounded large results', async (t) => { diff --git a/apps/desktop/src/main/runtime-host-local-operator.ts b/apps/desktop/src/main/runtime-host-local-operator.ts index e8179f9195..6e0613ab60 100644 --- a/apps/desktop/src/main/runtime-host-local-operator.ts +++ b/apps/desktop/src/main/runtime-host-local-operator.ts @@ -425,7 +425,7 @@ export function createDesktopRuntimeHostLocalOperator(input: { ? ['--expected-host-json', JSON.stringify(command.expectedHost)] : []), ...managedTargetArgs(command.target), - ...(command.allowManualUpdate ? ['--allow-manual-update'] : []), + ...(command.allowManualUpdate && targetVersion ? ['--allow-manual-update'] : []), ...(command.allowInterruptActiveTasks ? ['--allow-interrupt-active-tasks'] : []), ], }, diff --git a/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts b/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts index 195ae429ff..ce68751979 100644 --- a/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts +++ b/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts @@ -31,7 +31,10 @@ import { type RuntimeHostManagedDeploymentConfig, type RuntimeHostSupervisorProvider, } from '@maka/runtime-host/operator'; -import type { connectExistingRuntimeHost } from '@maka/runtime-host/client'; +import { + RuntimeHostOperationError, + type connectExistingRuntimeHost, +} from '@maka/runtime-host/client'; import { resolveStorageRoot, tryAcquireStateRootOwner } from '@maka/storage/root-authority'; import type { RuntimeHostLifecycleProvider, @@ -533,6 +536,80 @@ test('does not consume replacement consent after the supervised Host exits', asy assert.equal(retired, false); }); +test('requires explicit interruption authority when a supervised Host drains before diagnostics', async (t) => { + const stateRoot = await mkdtemp(join(tmpdir(), 'maka-lifecycle-diagnostics-drain-')); + t.after(() => rm(stateRoot, { recursive: true, force: true })); + const capability = await resolveStorageRoot({ path: stateRoot, kind: 'interactive' }); + let owner: Awaited> = + await tryAcquireStateRootOwner(capability); + assert.ok(owner); + t.after(async () => owner?.close()); + const events: string[] = []; + const connectExisting = (async () => { + events.push('connect'); + return { + kind: 'connected', + registration: { hostEpoch: 'host-a', pid: 42 }, + connection: { + request: async (operation: string) => { + events.push(operation); + throw new RuntimeHostOperationError( + 'host.diagnostics.query', + 'host_draining', + 'Runtime Host is draining', + ); + }, + close: async () => { + events.push('close'); + }, + }, + } as unknown as Awaited>; + }) as typeof connectExistingRuntimeHost; + const supervisor = { + status: async () => { + events.push('status'); + return { active: true, pid: 42 }; + }, + retire: async () => { + events.push('retire'); + await owner?.close(); + owner = undefined; + }, + }; + + assert.deepEqual( + await retireRuntimeHostLifecycleOwner({ + rootPath: capability.canonicalPath, + rootId: capability.rootId, + connectExisting, + expectedOwner: { hostEpoch: 'host-a', pid: 42 }, + supervisor, + }), + { kind: 'active_tasks' }, + ); + const retired = await retireRuntimeHostLifecycleOwner({ + rootPath: capability.canonicalPath, + rootId: capability.rootId, + connectExisting, + expectedOwner: { hostEpoch: 'host-a', pid: 42 }, + supervisor, + allowInterruptActiveTasks: true, + }); + assert.equal(retired.kind, 'retired'); + if (retired.kind === 'retired') await retired.owner.close(); + assert.deepEqual(events, [ + 'connect', + 'status', + 'host.diagnostics.query', + 'close', + 'connect', + 'status', + 'host.diagnostics.query', + 'retire', + 'close', + ]); +}); + test('requires explicit interruption authority to recover an unreachable supervised transition', async (t) => { const stateRoot = await mkdtemp(join(tmpdir(), 'maka-lifecycle-recovery-consent-')); const capability = await resolveStorageRoot({ path: stateRoot, kind: 'interactive' }); diff --git a/packages/cli/src/runtime-host-lifecycle-transaction.ts b/packages/cli/src/runtime-host-lifecycle-transaction.ts index 3f541919c9..60e24492a6 100644 --- a/packages/cli/src/runtime-host-lifecycle-transaction.ts +++ b/packages/cli/src/runtime-host-lifecycle-transaction.ts @@ -28,6 +28,8 @@ import { import { connectExistingRuntimeHost, prepareConnectedRuntimeHostRetirement, + RuntimeHostOperationError, + RuntimeHostRequestInterruptedError, waitForRuntimeHostReady, } from '@maka/runtime-host/client'; import { RUNTIME_HOST_PROTOCOL_VERSION } from '@maka/runtime-host/protocol'; @@ -316,6 +318,7 @@ function assertRecoveryTarget( export async function retireRuntimeHostLifecycleOwner(input: { readonly rootPath: string; readonly rootId: string; + readonly connectExisting?: typeof connectExistingRuntimeHost; readonly allowInterruptActiveTasks?: boolean; /** * Freshness fence evaluated before a canonical supervised-deployment retirement is admitted. @@ -357,7 +360,7 @@ export async function retireRuntimeHostLifecycleOwner(input: { throw error; } } - const connected = await connectExistingRuntimeHost({ + const connected = await (input.connectExisting ?? connectExistingRuntimeHost)({ rootPath: capability.canonicalPath, protocol: { min: RUNTIME_HOST_PROTOCOL_VERSION, @@ -384,31 +387,44 @@ export async function retireRuntimeHostLifecycleOwner(input: { ); } try { - const diagnostics = await connected.connection.request('host.diagnostics.query', {}); const supervisorStatus = await input.supervisor?.status(); if (supervisorStatus) assertExpectedSupervisorOwner(input.expectedOwner, supervisorStatus); if ( supervisorStatus && - (!supervisorStatus.active || supervisorStatus.pid !== diagnostics.pid) + (!supervisorStatus.active || supervisorStatus.pid !== connected.registration.pid) ) { throw new RuntimeHostLifecycleTransactionError( 'transition_failed', - 'The supervisor and State Root report different Runtime Host processes', + 'The supervisor and Runtime Host registration report different processes', ); } - // The exact Root owner and canonical supervisor now agree while the deployment lock is held. - // This admits retirement of that deployment; a later same-deployment restart is not a new - // authority, but it must not acquire the Root before the supervisor is retired. - const prepared = await prepareConnectedRuntimeHostRetirement( - connected.connection, - input.allowInterruptActiveTasks ? 'interrupt_active_work' : 'refuse_active_work', - ); - if (prepared.kind === 'active_tasks') return prepared; - if (prepared.pid !== diagnostics.pid) { - throw new RuntimeHostLifecycleTransactionError( - 'transition_failed', - 'The Runtime Host process changed while retirement was prepared', + try { + const diagnostics = await connected.connection.request('host.diagnostics.query', {}); + if (diagnostics.pid !== connected.registration.pid) { + throw new RuntimeHostLifecycleTransactionError( + 'transition_failed', + 'The Runtime Host registration and diagnostics report different processes', + ); + } + // The exact Root owner and canonical supervisor now agree while the deployment lock is held. + // This admits retirement of that deployment; a later same-deployment restart is not a new + // authority, but it must not acquire the Root before the supervisor is retired. + const prepared = await prepareConnectedRuntimeHostRetirement( + connected.connection, + input.allowInterruptActiveTasks ? 'interrupt_active_work' : 'refuse_active_work', ); + if (prepared.kind === 'active_tasks') return prepared; + if (prepared.pid !== diagnostics.pid) { + throw new RuntimeHostLifecycleTransactionError( + 'transition_failed', + 'The Runtime Host process changed while retirement was prepared', + ); + } + } catch (error) { + if (!isRuntimeHostRetirementUnavailable(error) || !input.supervisor) throw error; + if (!input.allowInterruptActiveTasks) return { kind: 'active_tasks' }; + await input.supervisor.retire(); + return waitForRuntimeHostLifecycleOwner(capability, input.timeoutMs ?? 45_000); } const retirement = await waitForRuntimeHostLifecycleOwner( capability, @@ -426,6 +442,14 @@ export async function retireRuntimeHostLifecycleOwner(input: { } } +function isRuntimeHostRetirementUnavailable(error: unknown): boolean { + return ( + error instanceof RuntimeHostRequestInterruptedError || + (error instanceof RuntimeHostOperationError && + (error.code === 'host_not_ready' || error.code === 'host_draining')) + ); +} + function assertExpectedRuntimeHostOwner( expected: { readonly hostEpoch: string; readonly pid: number } | undefined, observed: { readonly hostEpoch: string; readonly pid: number } | undefined, diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index 8ba15639e0..810184f4bc 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -875,6 +875,64 @@ test('turn.start durably binds a Guest request approval to the admitted Turn', a } }); +test('turn.regenerate durably binds a Guest request approval to the admitted Turn', async () => { + const fixture = await createFailureFixture({ + registerBackend: (backends) => + backends.register('ai-sdk', (context) => new FakeBackend(context)), + }); + const authorization = { + kind: 'session_turn_access_request' as const, + requestId: 'request-regenerate-1', + principalId: 'session_guest:guest-1', + grantId: 'grant-1', + approvedAt: 1_788_000_000_000, + approvedBy: 'local_owner', + }; + const input = { + sessionId: fixture.sessionId, + sourceTurnId: 'turn-regenerate-source', + turnId: 'turn-regenerate-approved', + }; + try { + assertStartedTurn( + await fixture.interactiveTurns.handlers['turn.start']( + { + sessionId: fixture.sessionId, + turnId: input.sourceTurnId, + content: { text: 'Regenerate this approved request.' }, + }, + operationContext(fixture.hostEpoch, fixture.acquireResidency), + ), + ); + await fixture.coordinator.whenIdle(fixture.sessionId); + + const regenerated = await fixture.interactiveTurns.handlers['turn.regenerate'](input, { + ...operationContext(fixture.hostEpoch, fixture.acquireResidency), + principal: authorization.principalId, + turnAdmissionAuthorization: authorization, + }); + assert.equal(regenerated.ok, true, JSON.stringify(regenerated)); + const admission = await fixture.stores.agentRunStore.readRootTurnAdmission( + fixture.sessionId, + input.turnId, + ); + assert.deepEqual(admission?.execution, { + kind: 'regenerate', + sourceTurnId: input.sourceTurnId, + }); + assert.deepEqual(admission?.authorization, authorization); + + const conflictingRetry = await fixture.interactiveTurns.handlers['turn.regenerate']( + input, + operationContext(fixture.hostEpoch, fixture.acquireResidency), + ); + assert.equal(conflictingRetry.ok, false); + if (!conflictingRetry.ok) assert.equal(conflictingRetry.error.code, 'operation_conflict'); + } finally { + await fixture.dispose(); + } +}); + test('turn.start resolves explicit Skills once before durable admission and replays the result', async () => { let preparationCount = 0; let blocked = false; diff --git a/packages/storage/src/__tests__/regenerate-root-admission.test.ts b/packages/storage/src/__tests__/regenerate-root-admission.test.ts index f1c2d0c501..4429d04de6 100644 --- a/packages/storage/src/__tests__/regenerate-root-admission.test.ts +++ b/packages/storage/src/__tests__/regenerate-root-admission.test.ts @@ -36,6 +36,7 @@ test('regenerate admission durably binds the immutable source Turn', async () => kind: 'regenerate', sourceTurnId: 'source-turn', }); + assert.deepEqual(admitted.admission.authorization, input.authorization); store.close?.(); const reopened = createSqliteAgentRunStore(root); @@ -116,6 +117,14 @@ function admissionInput(overrides: Partial = {}): AdmitRootT previousRootTurnId: null, normalizedInput: { text: 'Original request' }, sourceMessages: [], + authorization: { + kind: 'session_turn_access_request', + requestId: 'request-regenerate', + principalId: 'session_guest:guest-1', + grantId: 'grant-1', + approvedAt: 45, + approvedBy: 'local_owner', + }, admittedAt: 50, ...overrides, }; diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index b10e782776..b227cc2d75 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -1897,9 +1897,13 @@ function assertRootTurnAdmissionContract(admission: RootTurnAdmission): void { 'Invalid root turn admission contract: Skill invocation requires external message execution', ); } - if (admission.authorization && execution.kind !== 'external_message') { + if ( + admission.authorization && + execution.kind !== 'external_message' && + execution.kind !== 'regenerate' + ) { throw new Error( - 'Invalid root turn admission contract: authorization proof requires external message execution', + 'Invalid root turn admission contract: authorization proof requires external message or regenerate execution', ); } if (execution.kind === 'claimed_agent_graph_intent') {