You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/08/31 13:16:20 UTC
(maka) 01/02: fix(runtime): prepare provider target once per activation
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch fix/reasoning-replay-provenance
in repository https://gitbox.apache.org/repos/asf/maka.git
commit c14f69a5d1dcb6787969a33e0cd38e09cfdd2205
Author: AstroHan <le...@outlook.com>
AuthorDate: Mon Aug 31 21:15:24 2026 +0800
fix(runtime): prepare provider target once per activation
Generated-by: Codex
---
.../connection-effect-coordinator.test.ts | 21 ++-
.../__tests__/execution-model-composition.test.ts | 208 +++++++++++----------
.../src/server/execution-composition.ts | 92 +++++----
.../src/server/execution-model-authority.ts | 22 +--
.../src/server/execution-model-composition.ts | 57 +++++-
.../computer-use-provider-protocol.test.ts | 9 +
.../__tests__/overflow-reactive-recovery.test.ts | 4 +
.../runtime/src/__tests__/session-manager.test.ts | 18 +-
packages/runtime/src/agent-run.ts | 17 +-
packages/runtime/src/runtime-kernel.ts | 72 ++++++-
packages/runtime/src/session-manager.ts | 49 +++--
11 files changed, 356 insertions(+), 213 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/connection-effect-coordinator.test.ts b/packages/runtime-host/src/__tests__/connection-effect-coordinator.test.ts
index be950a3bad..904e644b58 100644
--- a/packages/runtime-host/src/__tests__/connection-effect-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/connection-effect-coordinator.test.ts
@@ -40,7 +40,7 @@ import { HostConnectionEffectCoordinator } from '../server/connection-effect-coo
import { HostOAuthExecutionAuthority } from '../server/oauth-execution-authority.js';
import type { ConnectionContext } from '../server/operation-dispatcher.js';
import { RuntimePolicyActivationGate } from '../server/runtime-policy-activation-gate.js';
-import { resolveExecutionProviderStateIdentity } from '../server/execution-model-authority.js';
+import { resolveExecutionTarget } from '../server/execution-model-authority.js';
import type { ConnectionOnboardingSaveResult, OperationOutcome } from '../protocol/index.js';
const context: ConnectionContext = {
@@ -502,7 +502,18 @@ test('provider state identity follows endpoint, credential, and request-header o
llmConnectionSlug: connection.slug,
model: 'gpt-5',
};
- const initial = await resolveExecutionProviderStateIdentity(header, stores);
+ const resolveIdentity = async () =>
+ (
+ await resolveExecutionTarget(
+ header,
+ stores,
+ new HostOAuthExecutionAuthority(stores),
+ () => {
+ throw new Error('API-key provider must not create an OAuth refresh transport');
+ },
+ )
+ ).providerStateIdentity;
+ const initial = await resolveIdentity();
const moved = await stores.connectionCatalog.update({
expected: { connectionId: connection.connectionId, revision: connection.revision },
@@ -514,7 +525,7 @@ test('provider state identity follows endpoint, credential, and request-header o
},
});
assert.equal(moved.kind, 'committed');
- const afterEndpoint = await resolveExecutionProviderStateIdentity(header, stores);
+ const afterEndpoint = await resolveIdentity();
assert.notEqual(afterEndpoint, initial);
const status = await connectionCredentialStatus(stores, connection);
@@ -526,7 +537,7 @@ test('provider state identity follows endpoint, credential, and request-header o
secret: 'key-b',
});
assert.equal(rotated.kind, 'committed');
- const afterCredential = await resolveExecutionProviderStateIdentity(header, stores);
+ const afterCredential = await resolveIdentity();
assert.notEqual(afterCredential, afterEndpoint);
const headers = await stores.credentialVault.set({
@@ -539,7 +550,7 @@ test('provider state identity follows endpoint, credential, and request-header o
secret: JSON.stringify({ 'X-Relay-Account': 'account-b' }),
});
assert.equal(headers.kind, 'committed');
- const afterHeaders = await resolveExecutionProviderStateIdentity(header, stores);
+ const afterHeaders = await resolveIdentity();
assert.notEqual(afterHeaders, afterCredential);
});
});
diff --git a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts
index a1a11e39c6..e6a05e772b 100644
--- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts
@@ -91,6 +91,7 @@ import {
} from '../server/execution-model-authority.js';
import {
createHostAiSdkBackend,
+ prepareHostAiSdkBackend,
resolveCollaborationPermissionMode,
type HostAiSdkBackendInput,
} from '../server/execution-model-composition.js';
@@ -161,18 +162,22 @@ test('backend creation resolves a bound Session by immutable Connection identity
});
});
-test('backend creation rejects provider state drift after Run admission', async () => {
- await assert.rejects(
- createHostAiSdkBackend(
- backendCreationFixture({
- abortSignal: new AbortController().signal,
- providerStateIdentity: `sha256:${'f'.repeat(64)}`,
- resolveExecutionConnection: async () => readyExecutionConnection(),
- readPricing: async () => ({ revision: 0, overrides: [] }),
- }),
- ),
- /Provider state changed after AgentRun admission/,
- );
+test('prepared backend activation builds from its admitted provider snapshot', async () => {
+ let providerReadAvailable = true;
+ const input = backendCreationFixture({
+ abortSignal: new AbortController().signal,
+ resolveExecutionConnection: async () => {
+ if (!providerReadAvailable) throw new Error('provider state was read after admission');
+ return readyExecutionConnection();
+ },
+ readPricing: async () => ({ revision: 0, overrides: [] }),
+ });
+ const { context, ...dependencies } = input;
+ const prepared = await prepareHostAiSdkBackend({ context, ...dependencies });
+ providerReadAvailable = false;
+
+ const backend = await prepared.build(context);
+ await backend.dispose();
});
test('backend creation aborts a stalled canonical connection read', async () => {
@@ -829,95 +834,99 @@ test('Codex OAuth history compaction falls back to a text checkpoint after nativ
resolve: async () => oauthTokens,
}),
} as unknown as HostOAuthExecutionAuthority;
- const backend = await createHostAiSdkBackend(
- backendCreationFixture({
- abortSignal: new AbortController().signal,
- modelId,
- oauthCredentials,
- resolveExecutionConnection: async () => ({
- kind: 'ready',
- connection: {
- slug: 'backend-creation-connection',
- providerType: 'openai-codex',
- enabledModelIds: [modelId],
- models: [
- {
- id: modelId,
- capabilities: { chat: true, functionCalling: true },
- contextWindow: 32_768,
- maxOutputTokens: 1_024,
- },
- ],
- },
- networkProxy: { enabled: false },
- secretMaterial: { connection: { secret: 'oauth-material' } },
- }),
- readPricing: async () => ({ revision: 0, overrides: [] }),
- recordHistoryCompactCheckpoint: async (checkpoint) => {
- recordedTextCheckpoint = 'summary' in checkpoint;
- },
- recordModelCallAttempt: async ({ attempt }) => {
- attempts.push(attempt);
+ const fixture = backendCreationFixture({
+ abortSignal: new AbortController().signal,
+ modelId,
+ oauthCredentials,
+ resolveExecutionConnection: async () => ({
+ kind: 'ready',
+ connection: {
+ slug: 'backend-creation-connection',
+ providerType: 'openai-codex',
+ enabledModelIds: [modelId],
+ models: [
+ {
+ id: modelId,
+ capabilities: { chat: true, functionCalling: true },
+ contextWindow: 32_768,
+ maxOutputTokens: 1_024,
+ },
+ ],
},
- createFetchTransport: () => ({
- fetch: async (url, init) => {
- const body = JSON.parse(String(init?.body)) as Record<string, unknown>;
- requests.push({
- url: String(url),
- body,
- });
- const providerInput = Array.isArray(body.input) ? body.input : [];
- if (
- !providerInput.some(
- (item) =>
- typeof item === 'object' &&
- item !== null &&
- 'type' in item &&
- item.type === 'compaction_trigger',
- )
- ) {
- return Response.json({
- id: 'resp-text-fallback',
- object: 'response',
- created_at: 1,
- status: 'completed',
- model: modelId,
- output: [
- {
- type: 'message',
- id: 'msg-text-fallback',
- status: 'completed',
- role: 'assistant',
- content: [
- {
- type: 'output_text',
- text: fallbackSummary,
- annotations: [],
- logprobs: [],
- },
- ],
- },
- ],
- usage: { input_tokens: 4_000, output_tokens: 60, total_tokens: 4_060 },
- });
- }
- return Response.json(
- {
- error: {
- message: 'request rejected without echoing this body',
- code: 'missing_required_parameter',
+ networkProxy: { enabled: false },
+ secretMaterial: { connection: { secret: 'oauth-material' } },
+ }),
+ readPricing: async () => ({ revision: 0, overrides: [] }),
+ recordHistoryCompactCheckpoint: async (checkpoint) => {
+ recordedTextCheckpoint = 'summary' in checkpoint;
+ },
+ recordModelCallAttempt: async ({ attempt }) => {
+ attempts.push(attempt);
+ },
+ createFetchTransport: () => ({
+ fetch: async (url, init) => {
+ const body = JSON.parse(String(init?.body)) as Record<string, unknown>;
+ requests.push({
+ url: String(url),
+ body,
+ });
+ const providerInput = Array.isArray(body.input) ? body.input : [];
+ if (
+ !providerInput.some(
+ (item) =>
+ typeof item === 'object' &&
+ item !== null &&
+ 'type' in item &&
+ item.type === 'compaction_trigger',
+ )
+ ) {
+ return Response.json({
+ id: 'resp-text-fallback',
+ object: 'response',
+ created_at: 1,
+ status: 'completed',
+ model: modelId,
+ output: [
+ {
+ type: 'message',
+ id: 'msg-text-fallback',
+ status: 'completed',
+ role: 'assistant',
+ content: [
+ {
+ type: 'output_text',
+ text: fallbackSummary,
+ annotations: [],
+ logprobs: [],
+ },
+ ],
},
+ ],
+ usage: { input_tokens: 4_000, output_tokens: 60, total_tokens: 4_060 },
+ });
+ }
+ return Response.json(
+ {
+ error: {
+ message: 'request rejected without echoing this body',
+ code: 'missing_required_parameter',
},
- {
- status: 400,
- headers: { 'x-request-id': 'req-codex-compact' },
- },
- );
- },
- close: async () => undefined,
- }),
+ },
+ {
+ status: 400,
+ headers: { 'x-request-id': 'req-codex-compact' },
+ },
+ );
+ },
+ close: async () => undefined,
}),
- );
+ });
+ const { context, ...dependencies } = fixture;
+ const prepared = await prepareHostAiSdkBackend({ context, ...dependencies });
+ const providerStateIdentity = prepared.providerStateIdentity;
+ assert.ok(providerStateIdentity);
+ const backend = await prepared.build(context);
+ assert.ok(backend.compactHistory);
try {
const runtimeContext: RuntimeEvent[] = [
@@ -1053,6 +1062,7 @@ test('Codex OAuth history compaction falls back to a text checkpoint after nativ
llmConnectionId: '11111111-1111-4111-8111-111111111111',
llmConnectionSlug: 'backend-creation-connection',
modelId,
+ providerStateIdentity,
cwd: '/workspace',
permissionMode: 'bypass',
createdAt: 2,
@@ -3871,7 +3881,6 @@ async function publishConnectionModel(
function backendCreationFixture(input: {
abortSignal: AbortSignal;
connectionId?: string;
- providerStateIdentity?: `sha256:${string}`;
resolveExecutionConnection: (ref?: unknown) => Promise<unknown>;
readPricing: () => Promise<unknown>;
runtimePolicy?: RuntimePolicyStoresWriter;
@@ -3941,9 +3950,6 @@ function backendCreationFixture(input: {
permissionMode: 'bypass',
},
abortSignal: input.abortSignal,
- ...(input.providerStateIdentity
- ? { providerStateIdentity: input.providerStateIdentity }
- : {}),
...(input.tools ? { tools: input.tools } : {}),
...(input.loadTurnRuntimeEvents
? { loadTurnRuntimeEvents: input.loadTurnRuntimeEvents }
diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts
index f9a5538c7f..d6590b724a 100644
--- a/packages/runtime-host/src/server/execution-composition.ts
+++ b/packages/runtime-host/src/server/execution-composition.ts
@@ -41,6 +41,7 @@ import {
BackendRegistry,
SessionManager,
type BackendFactory,
+ type BackendPreparationContext,
} from '@maka/runtime/session-manager';
import { buildToolsForAgentDefinition } from '@maka/runtime/agent-catalog';
import { buildHistoryTools } from '@maka/runtime/history-tools';
@@ -109,8 +110,7 @@ import { HostContextCoordinator } from './context-coordinator.js';
import { HostClientCapabilityCoordinator } from './client-capability-coordinator.js';
import { HostDeepResearchCoordinator } from './deep-research-coordinator.js';
import { HostDailyReviewCoordinator } from './daily-review-coordinator.js';
-import { createHostAiSdkBackend } from './execution-model-composition.js';
-import { resolveExecutionProviderStateIdentity } from './execution-model-authority.js';
+import { prepareHostAiSdkBackend } from './execution-model-composition.js';
import {
createInteractiveRunComposer,
createInteractiveRunComposerFactory,
@@ -665,51 +665,51 @@ export async function createExecutionRuntimeHostComposition(
lane: memoryExtractionLane,
acquireResidency: () => context.acquireResidency('memory-extraction'),
});
+ const hostAiSdkBackendInput = <T extends BackendPreparationContext>(backendContext: T) => ({
+ context: backendContext,
+ runtimePolicy: runtimePolicyStores,
+ oauthCredentials,
+ createRunComposer: createInteractiveRunComposerFactory({
+ skills,
+ memory: requireMemory(memory),
+ taskLedger,
+ clientCapabilities: requireClientCapabilities(clientCapabilities),
+ resolveTavilyWebSearchReadiness: () =>
+ resolveHostTavilyWebSearchReadiness(runtimePolicyStores.operations),
+ ...(scheduledTaskTool ? { scheduledTaskTool } : {}),
+ planStore: openedPlanStore,
+ deepResearchTools: requireDeepResearch(deepResearch).toolsForSession(
+ backendContext.sessionId,
+ ),
+ goalTools: requireGoal(goal).tools,
+ builtinTools,
+ hostTools,
+ resolveRootTools: (sessionId) =>
+ requireGraphCoordinator(graphCoordinator).toolsForSession(sessionId),
+ parentAgentTools: childAgentTools.parentTools,
+ childTools: childAgentTools.childTools,
+ worktreePatchWriteBackAvailable: true,
+ }),
+ ...(hostedExecutionRunProfile(backendContext.header.toolProfile)?.memoryExtraction === false
+ ? {}
+ : { memoryExtraction }),
+ artifacts: openedArtifactStore,
+ ...(openedContextOffloadReader ? { contextOffload: openedContextOffloadReader } : {}),
+ ...(storage.contextOffloadUnavailable ? { contextOffloadUnavailable: true } : {}),
+ executionArtifacts,
+ usage: openedUsageStores,
+ childAgents: bindHostChildAgentBackend(
+ requireSessionManager(manager),
+ backendContext.sessionId,
+ ),
+ runtimeCommitSink: stores.runtimeEventStore,
+ requestDrain: context.requestDrain,
+ });
backends.register(
'ai-sdk',
- dependencies.primaryBackendFactory ??
- ((backendContext) =>
- createHostAiSdkBackend({
- context: backendContext,
- runtimePolicy: runtimePolicyStores,
- oauthCredentials,
- createRunComposer: createInteractiveRunComposerFactory({
- skills,
- memory: requireMemory(memory),
- taskLedger,
- clientCapabilities: requireClientCapabilities(clientCapabilities),
- resolveTavilyWebSearchReadiness: () =>
- resolveHostTavilyWebSearchReadiness(runtimePolicyStores.operations),
- ...(scheduledTaskTool ? { scheduledTaskTool } : {}),
- planStore: openedPlanStore,
- deepResearchTools: requireDeepResearch(deepResearch).toolsForSession(
- backendContext.sessionId,
- ),
- goalTools: requireGoal(goal).tools,
- builtinTools,
- hostTools,
- resolveRootTools: (sessionId) =>
- requireGraphCoordinator(graphCoordinator).toolsForSession(sessionId),
- parentAgentTools: childAgentTools.parentTools,
- childTools: childAgentTools.childTools,
- worktreePatchWriteBackAvailable: true,
- }),
- ...(hostedExecutionRunProfile(backendContext.header.toolProfile)?.memoryExtraction ===
- false
- ? {}
- : { memoryExtraction }),
- artifacts: openedArtifactStore,
- ...(openedContextOffloadReader ? { contextOffload: openedContextOffloadReader } : {}),
- ...(storage.contextOffloadUnavailable ? { contextOffloadUnavailable: true } : {}),
- executionArtifacts,
- usage: openedUsageStores,
- childAgents: bindHostChildAgentBackend(
- requireSessionManager(manager),
- backendContext.sessionId,
- ),
- runtimeCommitSink: stores.runtimeEventStore,
- requestDrain: context.requestDrain,
- })),
+ dependencies.primaryBackendFactory ?? {
+ prepare: (backendContext) => prepareHostAiSdkBackend(hostAiSdkBackendInput(backendContext)),
+ },
);
const runtimeAuthority: RuntimeHostedRootAuthority = {
bindRun: (identity) => messages.bindRun(identity),
@@ -982,8 +982,6 @@ export async function createExecutionRuntimeHostComposition(
},
}),
runBackendActivation: (operation) => runtimePolicyActivation.runBackendActivation(operation),
- resolveProviderStateIdentity: (header) =>
- resolveExecutionProviderStateIdentity(header, runtimePolicyStores),
messageAuthority: runtimeAuthority,
hostedAgentGraphExecution: {
readAgentGraphIntentClaim: (graphId, intentId) =>
diff --git a/packages/runtime-host/src/server/execution-model-authority.ts b/packages/runtime-host/src/server/execution-model-authority.ts
index b4d8ef7646..5003be4e2c 100644
--- a/packages/runtime-host/src/server/execution-model-authority.ts
+++ b/packages/runtime-host/src/server/execution-model-authority.ts
@@ -731,7 +731,7 @@ class AuxiliaryModelCallConfigurationError extends Error {
}
}
-interface ResolvedExecutionTarget {
+export interface ResolvedExecutionTarget {
readonly connection: RuntimeExecutionConnection;
readonly model: string;
readonly apiKey: string;
@@ -775,26 +775,6 @@ function providerStateIdentityForResolvedExecution(
});
}
-export async function resolveExecutionProviderStateIdentity(
- header: ExecutionRouteHeader,
- runtimePolicy: {
- readonly operations: Pick<
- RuntimePolicyStoresWriter['operations'],
- 'resolveExecutionConnection'
- >;
- },
-): Promise<`sha256:${string}`> {
- const resolved = await runtimePolicy.operations.resolveExecutionConnection(
- executionConnectionRef(header),
- );
- if (resolved.kind !== 'ready') {
- throw new AuxiliaryModelCallConfigurationError(
- `Runtime Host model connection is not ready: ${resolved.kind}`,
- );
- }
- return providerStateIdentityForResolvedExecution(resolved);
-}
-
async function resolveDailyReviewHeader(
runtimePolicy: RuntimePolicyStoresWriter,
modelKey: string,
diff --git a/packages/runtime-host/src/server/execution-model-composition.ts b/packages/runtime-host/src/server/execution-model-composition.ts
index 04e2603de8..2d52137ac6 100644
--- a/packages/runtime-host/src/server/execution-model-composition.ts
+++ b/packages/runtime-host/src/server/execution-model-composition.ts
@@ -44,7 +44,11 @@ import {
} from '@maka/runtime/network/scoped-fetch-transport';
import { stableHash, toolCatalogHash } from '@maka/runtime/request-shape';
import { toolAvailabilityHash } from '@maka/runtime/tool-availability';
-import { type BackendFactoryContext } from '@maka/runtime/session-manager';
+import {
+ type BackendFactoryContext,
+ type BackendPreparationContext,
+ type PreparedBackendActivation,
+} from '@maka/runtime/session-manager';
import { type RuntimeCommitSink } from '@maka/runtime/runtime-commit-sink';
import {
createAttachmentByteReader,
@@ -63,7 +67,11 @@ import {
import type { HostChildAgentBackendCapabilities } from './child-agent-composition.js';
import type { HostExecutionArtifactServices } from './execution-artifacts.js';
import type { HostMemoryExtractionCoordinator } from './memory-extraction-coordinator.js';
-import { readDuringBackendCreation, resolveExecutionTarget } from './execution-model-authority.js';
+import {
+ readDuringBackendCreation,
+ resolveExecutionTarget,
+ type ResolvedExecutionTarget,
+} from './execution-model-authority.js';
import { toRuntimePolicyProxy } from './runtime-policy-proxy.js';
import type { HostRunComposer, HostRunComposerFactory } from './host-run-composer.js';
@@ -84,6 +92,10 @@ export interface HostAiSdkBackendInput {
readonly createFetchTransport?: (proxy: ProxiedFetchProxy | null) => ProxiedFetchTransport;
}
+export type HostAiSdkBackendPreparationInput = Omit<HostAiSdkBackendInput, 'context'> & {
+ readonly context: BackendPreparationContext;
+};
+
type HostExecutionRuntimePolicyAuthority = {
readonly operations: Pick<RuntimePolicyStoresWriter['operations'], 'resolveExecutionConnection'>;
readonly runtimePolicy: Pick<RuntimePolicyStoresWriter['runtimePolicy'], 'getSnapshot'>;
@@ -116,12 +128,14 @@ export async function createHostAiSdkBackend(input: HostAiSdkBackendInput): Prom
),
input.context.abortSignal,
);
- if (
- input.context.providerStateIdentity !== undefined &&
- input.context.providerStateIdentity !== target.providerStateIdentity
- ) {
- throw new Error('Provider state changed after AgentRun admission');
- }
+ return await buildHostAiSdkBackend(input, target);
+}
+
+async function buildHostAiSdkBackend(
+ input: HostAiSdkBackendInput,
+ target: ResolvedExecutionTarget,
+): Promise<AiSdkBackend> {
+ const createFetchTransport = input.createFetchTransport ?? createProxiedFetchTransport;
const pricingSnapshot = await readDuringBackendCreation(
() => input.usage.pricing.snapshot(),
input.context.abortSignal,
@@ -486,6 +500,33 @@ export async function createHostAiSdkBackend(input: HostAiSdkBackendInput): Prom
}
}
+export async function prepareHostAiSdkBackend(
+ input: HostAiSdkBackendPreparationInput,
+): Promise<PreparedBackendActivation> {
+ const createFetchTransport = input.createFetchTransport ?? createProxiedFetchTransport;
+ const preparedTarget = await readDuringBackendCreation(
+ () =>
+ resolveExecutionTarget(
+ input.context.header,
+ input.runtimePolicy,
+ input.oauthCredentials,
+ createFetchTransport,
+ ),
+ input.context.abortSignal,
+ );
+ return {
+ providerStateIdentity: preparedTarget.providerStateIdentity,
+ build: (context) =>
+ buildHostAiSdkBackend(
+ {
+ ...input,
+ context,
+ },
+ preparedTarget,
+ ),
+ };
+}
+
class HostAiSdkBackend extends AiSdkBackend {
constructor(
input: ConstructorParameters<typeof AiSdkBackend>[0],
diff --git a/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts b/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts
index 81884d8cdb..2efc952c62 100644
--- a/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts
+++ b/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts
@@ -46,6 +46,7 @@ import { createTestAiSdkBackend } from './execution-boundary-test-helpers.js';
import { latestObservationIn } from './observation-text-reader.js';
const servers: Array<{ close(): Promise<void> }> = [];
+const PROVIDER_STATE_IDENTITY = `sha256:${'1'.repeat(64)}` as const;
after(async () => {
await Promise.all(servers.map((server) => server.close()));
@@ -96,6 +97,7 @@ describe('Anthropic-compatible Computer Use product loops', () => {
appendMessage: async () => {},
connection: providerConnection,
apiKey: 'test-key',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
modelId: 'claude-sonnet-4-5-20250929',
modelFactory: (input) => getAIModel(input),
tools: [],
@@ -114,6 +116,7 @@ describe('Anthropic-compatible Computer Use product loops', () => {
llmConnectionId: 'connection-anthropic',
llmConnectionSlug: 'anthropic',
modelId: 'claude-sonnet-4-5-20250929',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
cwd: '/tmp/maka',
permissionMode: 'bypass',
createdAt: 1,
@@ -480,6 +483,7 @@ describe('OpenAI-compatible product loops', () => {
appendMessage: async () => {},
connection: providerConnection,
apiKey: 'test-key',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
modelId: 'gpt-5.4',
modelFactory: (input) => getAIModel(input),
tools: [],
@@ -497,6 +501,7 @@ describe('OpenAI-compatible product loops', () => {
llmConnectionId: 'connection-copilot',
llmConnectionSlug: 'github-copilot',
modelId: 'gpt-5.4',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
cwd: '/tmp/maka',
permissionMode: 'bypass',
createdAt: 1,
@@ -689,6 +694,7 @@ describe('OpenAI-compatible product loops', () => {
llmConnectionId: 'test-connection-id',
llmConnectionSlug: providerConnection.slug,
modelId: 'k3',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
cwd: '/tmp/maka',
permissionMode: 'ask',
createdAt: 1,
@@ -746,6 +752,7 @@ describe('OpenAI-compatible product loops', () => {
appendMessage: async () => {},
connection: providerConnection,
apiKey: 'test-key',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
modelId: 'k3',
modelFactory: (input) => getAIModel(input),
providerOptions: buildProviderOptions(providerConnection, 'k3'),
@@ -834,6 +841,7 @@ describe('OpenAI-compatible product loops', () => {
llmConnectionId: 'test-connection-id',
llmConnectionSlug: providerConnection.slug,
modelId: 'k3',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
cwd: '/tmp/maka',
permissionMode: 'ask',
createdAt: firstTurn.anchor.ts,
@@ -850,6 +858,7 @@ describe('OpenAI-compatible product loops', () => {
},
connection: providerConnection,
apiKey: 'test-key',
+ providerStateIdentity: PROVIDER_STATE_IDENTITY,
modelId: 'k3',
modelFactory: (input) => getAIModel(input),
providerOptions: buildProviderOptions(providerConnection, 'k3'),
diff --git a/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts
index 02a150ff41..f1c2ebc488 100644
--- a/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts
+++ b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts
@@ -62,6 +62,7 @@ function reactiveStructuredSummary(foldedRuntimeEvents: RuntimeEvent[]): string
const RAW_SPAN_ONE = 'RAW_SPAN_ONE_'.repeat(24);
const ANCHOR_TEXT = 'reactive overflow recovery keep my exact words';
const OVERFLOW_MESSAGE = 'prompt is too long: 213462 tokens > 200000 maximum';
+const PROVIDER_STATE_IDENTITY = `sha256:${'1'.repeat(64)}` as const;
/**
* Per-provider-request script. Each entry drives one `doStream` invocation:
@@ -584,6 +585,7 @@ function buildReactiveFixture(options: ReactiveFixtureOptions): ReactiveFixture
models: [{ id: 'mock-model-id', contextWindow }],
},
apiKey: 'sk-test',
+ ...(options.reasoningReplayTail ? { providerStateIdentity: PROVIDER_STATE_IDENTITY } : {}),
modelId: 'mock-model-id',
modelFactory: () => model,
...(options.imagePrior || options.currentImage || options.liveImageResult
@@ -1912,6 +1914,8 @@ function priorRunHeader(runId: string, llmConnectionId: string, modelId: string)
llmConnectionId,
llmConnectionSlug: 'anthropic-source',
modelId,
+ providerStateIdentity:
+ runId === 'same-route-prior-run' ? PROVIDER_STATE_IDENTITY : `sha256:${'2'.repeat(64)}`,
cwd: '/tmp/maka',
permissionMode: 'ask',
createdAt: 1,
diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts
index 4893edbef2..5742e2b28e 100644
--- a/packages/runtime/src/__tests__/session-manager.test.ts
+++ b/packages/runtime/src/__tests__/session-manager.test.ts
@@ -4567,6 +4567,9 @@ describe('SessionManager permission mode updates', () => {
const store = new MemorySessionStore();
const runStore = new MemoryAgentRunStore();
const backends = new BackendRegistry();
+ backends.register('ai-sdk', () => {
+ throw new Error('continuation planning must not build a backend');
+ });
const manager = new SessionManager({
store,
runStore,
@@ -4709,6 +4712,9 @@ describe('SessionManager permission mode updates', () => {
const store = new MemorySessionStore();
const runStore = new MemoryAgentRunStore();
const backends = new BackendRegistry();
+ backends.register('ai-sdk', () => {
+ throw new Error('continuation planning must not build a backend');
+ });
const manager = new SessionManager({
store,
runStore,
@@ -5058,9 +5064,14 @@ describe('SessionManager permission mode updates', () => {
const lifecycleEvents: Array<{ type: string }> = [];
let backend: FinalTextTestBackend | undefined;
const providerStateIdentity = `sha256:${'c'.repeat(64)}` as const;
- backends.register('ai-sdk', (ctx) => {
- backend = new FinalTextTestBackend(ctx);
- return backend;
+ backends.register('ai-sdk', {
+ prepare: async () => ({
+ providerStateIdentity,
+ build: (ctx) => {
+ backend = new FinalTextTestBackend(ctx);
+ return backend;
+ },
+ }),
});
const manager = new SessionManager({
store,
@@ -5074,7 +5085,6 @@ describe('SessionManager permission mode updates', () => {
backgroundOperationsSettled: true,
availableToolNames: ['Read'],
}),
- resolveProviderStateIdentity: async () => providerStateIdentity,
onContinuationLifecycleEvent: (event) => {
lifecycleEvents.push(event);
},
diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts
index c0b2333088..90f903c275 100644
--- a/packages/runtime/src/agent-run.ts
+++ b/packages/runtime/src/agent-run.ts
@@ -154,7 +154,6 @@ export interface AgentRunInput {
newId: () => string;
now: () => number;
workspaceIdentity?: string;
- resolveProviderStateIdentity?: (header: SessionHeader) => Promise<`sha256:${string}` | undefined>;
continuationFailpoint?: (point: RuntimeContinuationFailpoint) => Promise<void>;
/** Exact target header already committed inside the durable continuation claim. */
claimedRunHeader?: AgentRunHeader;
@@ -330,8 +329,17 @@ export class AgentRun {
return this.stopped;
}
- resolvedProviderStateIdentity(): `sha256:${string}` | undefined {
- return this.providerStateIdentity;
+ headerSnapshot(): SessionHeader {
+ return this.header;
+ }
+
+ bindProviderStateIdentity(identity: `sha256:${string}` | undefined): void {
+ const claimed = this.input.claimedRunHeader?.providerStateIdentity;
+ const expected = claimed ?? this.providerStateIdentity;
+ if (expected !== undefined && expected !== identity) {
+ throw new Error('Prepared backend provider state does not match the AgentRun admission');
+ }
+ this.providerStateIdentity = identity;
}
isSessionInline(): boolean {
@@ -1103,8 +1111,7 @@ export class AgentRun {
? this.input.claimedRunHeader.createdAt
: this.input.now();
const providerStateIdentity =
- this.input.claimedRunHeader?.providerStateIdentity ??
- (await this.input.resolveProviderStateIdentity?.(this.header));
+ this.input.claimedRunHeader?.providerStateIdentity ?? this.providerStateIdentity;
this.providerStateIdentity = providerStateIdentity;
const computedHeader: AgentRunHeader = {
runId: this.runId,
diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts
index 2fabcdd38d..e309c696b0 100644
--- a/packages/runtime/src/runtime-kernel.ts
+++ b/packages/runtime/src/runtime-kernel.ts
@@ -85,6 +85,7 @@ import type {
BackendFactoryContext,
BackendRegistry,
CompactSessionInput,
+ PreparedBackendActivation,
ResolvedChildToolActivation,
SessionStore,
StopSessionInput,
@@ -269,7 +270,6 @@ export interface RuntimeKernelDeps {
safeBoundaryResumeEnabled?: boolean;
continuationFailpoint?: (point: RuntimeContinuationFailpoint) => Promise<void>;
runBackendActivation?: BackendActivationBoundary;
- resolveProviderStateIdentity?: (header: SessionHeader) => Promise<`sha256:${string}` | undefined>;
/** Hosted composition capability. When present, the Host owns all message queues. */
messageAuthority?: RuntimeMessageAuthority;
/** Hosted composition capability. Omit for embedded interaction ownership. */
@@ -283,6 +283,7 @@ interface BackendGeneration extends AgentRunActiveSession {
generation: number;
phase: 'active' | 'stopping' | 'disposing' | 'failed' | 'terminated';
backend: AgentBackend;
+ providerStateIdentity?: `sha256:${string}`;
stopBackend: AgentBackend['stop'];
stopState:
| { kind: 'idle' }
@@ -349,6 +350,7 @@ interface PendingExecutionClaim {
rejectSettled(error: unknown): void;
phase: 'pending' | 'attached' | 'reserved' | 'released' | 'failed';
run?: AgentRun;
+ backendPreparation?: PreparedBackendActivation;
stopIntent?: SessionStopIntent;
finalization?: ExecutionClaimOutcome;
}
@@ -659,7 +661,6 @@ export class RuntimeKernel implements RuntimeKernelLike {
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
- resolveProviderStateIdentity: this.deps.resolveProviderStateIdentity,
...(workspaceIdentity ? { workspaceIdentity } : {}),
hooks: {
reserveRun: async (targetSessionId, nextHeader, activeRun) => {
@@ -744,7 +745,14 @@ export class RuntimeKernel implements RuntimeKernelLike {
this.deps.runStore.readRun(continuation.sessionId, continuation.sourceRunId),
this.deps.runStore.listSessionRuns(continuation.sessionId),
]);
- const targetProviderStateIdentity = await this.deps.resolveProviderStateIdentity?.(header);
+ const targetProviderStateIdentity = (
+ await this.deps.backends.prepare(header.backend, {
+ sessionId: continuation.sessionId,
+ workspaceRoot: header.workspaceRoot,
+ header,
+ abortSignal: execution.abortController.signal,
+ })
+ ).providerStateIdentity;
const admissionRoute: ContinuationReplayAdmissionRoute = {
runHeaders: sessionRuns,
targetProviderStateIdentity,
@@ -823,7 +831,6 @@ export class RuntimeKernel implements RuntimeKernelLike {
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
- resolveProviderStateIdentity: this.deps.resolveProviderStateIdentity,
workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
effectiveOrchestration,
claimedRunHeader: claim.targetRunHeader,
@@ -977,7 +984,6 @@ export class RuntimeKernel implements RuntimeKernelLike {
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
- resolveProviderStateIdentity: this.deps.resolveProviderStateIdentity,
effectiveOrchestration: resolveEffectiveOrchestration('default', undefined),
hooks: {
reserveRun: async (targetSessionId, nextHeader, activeRun) => {
@@ -1010,7 +1016,12 @@ export class RuntimeKernel implements RuntimeKernelLike {
runId: run.runId,
});
}
- begin = await this.runBackendActivation(() => run.beginOperation());
+ begin = await this.runBackendActivation(async () => {
+ run.bindProviderStateIdentity(
+ await this.prepareBackendForExecution(sessionId, header, execution),
+ );
+ return await run.beginOperation();
+ });
await input.hostedRoot?.onRunStarted?.();
this.settleReservedExecutionClaim(execution, run, { ok: true });
} catch (error) {
@@ -1137,6 +1148,9 @@ export class RuntimeKernel implements RuntimeKernelLike {
}
begin = await this.runBackendActivation(async () => {
await prepareBackendActivation?.();
+ run.bindProviderStateIdentity(
+ await this.prepareBackendForExecution(sessionId, run.headerSnapshot(), execution),
+ );
const started = await run.begin();
await owners.bindInteraction(this.deps.interactionAuthority, {
sessionId,
@@ -1289,6 +1303,13 @@ export class RuntimeKernel implements RuntimeKernelLike {
throw new Error('Durable continuation omitted final safety revalidation');
}
await revalidateSafety();
+ run.bindProviderStateIdentity(
+ await this.prepareBackendForExecution(
+ continuation.sessionId,
+ run.headerSnapshot(),
+ execution,
+ ),
+ );
const started = await run.beginContinuation(continuation);
await owners.bindInteraction(this.deps.interactionAuthority, {
sessionId: continuation.sessionId,
@@ -2310,12 +2331,28 @@ export class RuntimeKernel implements RuntimeKernelLike {
};
}
+ private async prepareBackendForExecution(
+ sessionId: string,
+ header: SessionHeader,
+ execution: PendingExecutionClaim,
+ ): Promise<`sha256:${string}` | undefined> {
+ const existing = this.active.get(sessionId);
+ if (existing) return existing.providerStateIdentity;
+ const prepared = await this.deps.backends.prepare(header.backend, {
+ sessionId,
+ workspaceRoot: header.workspaceRoot,
+ header,
+ abortSignal: execution.abortController.signal,
+ });
+ execution.backendPreparation = prepared;
+ return prepared.providerStateIdentity;
+ }
+
private async ensureActive(
sessionId: string,
header: SessionHeader,
execution: PendingExecutionClaim,
): Promise<BackendGeneration> {
- const providerStateIdentity = execution.run?.resolvedProviderStateIdentity();
await this.clearBackendQuarantineForActivation(sessionId, execution);
let existing = this.active.get(sessionId);
if (existing) {
@@ -2331,12 +2368,20 @@ export class RuntimeKernel implements RuntimeKernelLike {
const entry = await this.shareBackendActivation(`parent:${sessionId}`, async () => {
const current = this.active.get(sessionId);
if (current) return current;
+ const prepared =
+ execution.backendPreparation ??
+ (await this.deps.backends.prepare(header.backend, {
+ sessionId,
+ workspaceRoot: header.workspaceRoot,
+ header,
+ abortSignal: execution.abortController.signal,
+ }));
+ execution.run?.bindProviderStateIdentity(prepared.providerStateIdentity);
const subagent = await this.resolveSubagentActivation(header);
- const backend = await this.deps.backends.build(header.backend, {
+ const backend = await prepared.build({
sessionId,
workspaceRoot: header.workspaceRoot,
header,
- ...(providerStateIdentity ? { providerStateIdentity } : {}),
store: this.deps.store,
abortSignal: execution.abortController.signal,
...(subagent
@@ -2352,7 +2397,12 @@ export class RuntimeKernel implements RuntimeKernelLike {
allowMidTurnHistoryCompaction: Boolean(this.deps.runtimeEventStore),
});
await this.rejectCancelledBackendActivation(backend, header, execution);
- const generation = this.createBackendGeneration(sessionId, backend, header);
+ const generation = this.createBackendGeneration(
+ sessionId,
+ backend,
+ header,
+ prepared.providerStateIdentity,
+ );
this.active.set(sessionId, generation);
return generation;
});
@@ -2428,12 +2478,14 @@ export class RuntimeKernel implements RuntimeKernelLike {
sessionId: string,
backend: AgentBackend,
header: SessionHeader,
+ providerStateIdentity?: `sha256:${string}`,
): BackendGeneration {
const active: BackendGeneration = {
sessionId,
generation: ++this.nextBackendGeneration,
phase: 'active',
backend,
+ ...(providerStateIdentity ? { providerStateIdentity } : {}),
stopBackend: undefined as never,
stopState: { kind: 'idle' },
cachedHeader: header,
diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts
index aec76185d6..850029e095 100644
--- a/packages/runtime/src/session-manager.ts
+++ b/packages/runtime/src/session-manager.ts
@@ -641,8 +641,6 @@ export interface BackendFactoryContext {
sessionId: string;
workspaceRoot: string;
header: SessionHeader;
- /** Provider state identity already frozen into the activating AgentRun. */
- providerStateIdentity?: `sha256:${string}`;
store: SessionStore;
/** Process-local cancellation for the execution that owns this activation. */
abortSignal?: AbortSignal;
@@ -704,21 +702,43 @@ export interface BackendFactoryContext {
export type BackendFactory = (ctx: BackendFactoryContext) => AgentBackend | Promise<AgentBackend>;
+export type BackendPreparationContext = Pick<
+ BackendFactoryContext,
+ 'sessionId' | 'workspaceRoot' | 'header' | 'abortSignal'
+>;
+
+export interface PreparedBackendActivation {
+ readonly providerStateIdentity?: `sha256:${string}`;
+ build(ctx: BackendFactoryContext): AgentBackend | Promise<AgentBackend>;
+}
+
+export interface PreparedBackendFactory {
+ prepare(ctx: BackendPreparationContext): Promise<PreparedBackendActivation>;
+}
+
+type BackendRegistration = BackendFactory | PreparedBackendFactory;
+
export class BackendRegistry {
- private readonly factories = new Map<PersistedBackendKind, BackendFactory>();
+ private readonly registrations = new Map<PersistedBackendKind, BackendRegistration>();
- register(kind: PersistedBackendKind, factory: BackendFactory): void {
- this.factories.set(kind, factory);
+ register(kind: PersistedBackendKind, registration: BackendRegistration): void {
+ this.registrations.set(kind, registration);
}
- async build(kind: PersistedBackendKind, ctx: BackendFactoryContext): Promise<AgentBackend> {
- const f = this.factories.get(kind);
- if (!f) throw new Error(`No backend factory registered for kind="${kind}"`);
- return await f(ctx);
+ async prepare(
+ kind: PersistedBackendKind,
+ ctx: BackendPreparationContext,
+ ): Promise<PreparedBackendActivation> {
+ const registration = this.registrations.get(kind);
+ if (!registration) throw new Error(`No backend factory registered for kind="${kind}"`);
+ if (typeof registration === 'function') {
+ return { build: registration };
+ }
+ return await registration.prepare(ctx);
}
has(kind: PersistedBackendKind): boolean {
- return this.factories.has(kind);
+ return this.registrations.has(kind);
}
}
@@ -775,7 +795,6 @@ interface SessionManagerBaseDeps {
inspectContinuationSafety?: (sessionId: string) => Promise<RuntimeContinuationSafetyObservation>;
continuationFailpoint?: (point: RuntimeContinuationFailpoint) => Promise<void>;
runBackendActivation?: BackendActivationBoundary;
- resolveProviderStateIdentity?: (header: SessionHeader) => Promise<`sha256:${string}` | undefined>;
safeBoundaryResumeEnabled?: boolean;
/** Hosted composition capability. Omit for the production embedded queue. */
messageAuthority?: RuntimeMessageAuthority;
@@ -1999,7 +2018,13 @@ export class SessionManager {
this.deps.store.readHeader(sessionId),
this.deps.runStore.listSessionRuns(sessionId),
]);
- const targetProviderStateIdentity = await this.deps.resolveProviderStateIdentity?.(header);
+ const targetProviderStateIdentity = (
+ await this.deps.backends.prepare(header.backend, {
+ sessionId,
+ workspaceRoot: header.workspaceRoot,
+ header,
+ })
+ ).providerStateIdentity;
admissionRoute = {
runHeaders,
targetProviderStateIdentity,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org