diff --git a/docs/agent-managed-compute/roadmap.md b/docs/agent-managed-compute/roadmap.md index 47dbe514..89e2f9f2 100644 --- a/docs/agent-managed-compute/roadmap.md +++ b/docs/agent-managed-compute/roadmap.md @@ -57,7 +57,9 @@ The work is ordered to prove the two-agent atom before adding scale. - Extend the existing run record with revisions, ownership generation, commands, provider references, and coordination events. - Add durable adapters with conditional writes. - Persist dispatch intent before provider creation. - Partially done: `startRetainedRun` blocks on its required `onAdmission` hook after creation and after dispatch, and `recoverRetainedRun` rebuilds a run from the pre-dispatch record. + Partially done: `startRetainedRun` persists admission after creation and dispatch. + `startRetainedRunInEnvironment` applies the same boundary to a fresh session in an existing environment. + `recoverRetainedRun` rebuilds a run from the pre-dispatch record. The pre-creation intent record remains open. - Add a durable provider-command outbox with coordinator generation and command sequence. - Rebuild budget reservations and interaction state on restart. diff --git a/docs/api/primitive-catalog.md b/docs/api/primitive-catalog.md index c1c6ecf5..39106140 100644 --- a/docs/api/primitive-catalog.md +++ b/docs/api/primitive-catalog.md @@ -525,7 +525,7 @@ Import from `@tangle-network/agent-runtime/intelligence` — 166 exports. ### Execution kernel — recursive atom, supervision, executors, round-synchronous loop -Import from `@tangle-network/agent-runtime/kernel` — 755 exports. +Import from `@tangle-network/agent-runtime/kernel` — 757 exports. | Symbol | Kind | Summary | |---|---|---| @@ -715,6 +715,7 @@ Import from `@tangle-network/agent-runtime/kernel` — 755 exports. | `settledWorkerOut` | function | What a settled worker exposes as its output artifact (the blob the brain's | | `spendFromUsageEvents` | function | Fold a normalized `UsageEvent` array into a `Spend`. Tokens and usd are separate | | `startRetainedRun` | function | Dispatch one detached, replayable run and return only after exact durable | +| `startRetainedRunInEnvironment` | function | Dispatch a fresh retained session inside an existing provider environment. | | `stopSentinel` | function | A unique, attributable stop sentinel for a node (ralph-loop style). Deterministic from the | | `streamAgentTurn` | function | Run ONE agent turn on any backend kind and stream its events. Yields the | | `structuralRollout` | function | Build the structuralRollout `Strategy`: k shots → score each by the frozen visible | @@ -987,6 +988,7 @@ Import from `@tangle-network/agent-runtime/kernel` — 755 exports. | `SpawnJournal` | interface | The spawn-tree event source (mirrors `ConversationJournal`'s begin/append/load shape). | | `Spend` | interface | Conserved spend, reconciled from the normalized `UsageEvent` stream. Tokens and usd | | `SpendGap` | interface | One journaled node whose usage accounting is incomplete — the named gap behind a `false` | +| `StartRetainedRunInEnvironmentOptions` | interface | A fresh retained session inside a provider environment that already exists. | | `StartRetainedRunOptions` | interface | A retained start is retry-safe only when environment and turn keys are explicit. | | `SteerableRootHandle` | interface | A Runtime-minted root handle that can deliver raw steering or answers to a live manager inbox. | | `SteerableSandboxSession` | interface | What the steerable session exposes to its executor: the usage stream plus the live reads. | diff --git a/docs/api/runtime.md b/docs/api/runtime.md index 340f8da4..bc408d0b 100644 --- a/docs/api/runtime.md +++ b/docs/api/runtime.md @@ -920,7 +920,7 @@ Manager-scoped assignment identity, including deterministic ids for unkeyed sibl ###### Inherited from -[`NodeSnapshot`](#nodesnapshot).[`identity`](#identity-7) +[`NodeSnapshot`](#nodesnapshot).[`identity`](#identity-8) ##### materialization? @@ -6290,6 +6290,73 @@ so every process derives the same values. *** +### StartRetainedRunInEnvironmentOptions + +**`Stable`** + +A fresh retained session inside a provider environment that already exists. + +#### Properties + +##### provider + +> `readonly` **provider**: `AgentEnvironmentProvider` + +##### environment + +> `readonly` **environment**: `object` + +###### id + +> `readonly` **id**: `string` + +Stable provider environment identifier used by `provider.get`. + +###### idempotencyKey + +> `readonly` **idempotencyKey**: `string` + +Original environment key retained for deterministic run identity and recovery records. + +##### turn + +> `readonly` **turn**: `AgentTurnInput` & `object` + +###### Type Declaration + +###### turnId + +> **turnId**: `string` + +##### identity? + +> `readonly` `optional` **identity?**: `object` + +Explicit fresh-session coordinates. When omitted, the runtime mints them +from `(environment.idempotencyKey, turn.turnId)`. + +###### sessionId + +> `readonly` **sessionId**: `string` + +###### executionId + +> `readonly` **executionId**: `string` + +##### onAdmission + +> `readonly` **onAdmission**: [`RetainedRunAdmissionHook`](#retainedrunadmissionhook) + +##### now? + +> `readonly` `optional` **now?**: () => `number` + +###### Returns + +`number` + +*** + ### ReconnectRetainedRunOptions **`Stable`** @@ -14446,7 +14513,7 @@ Stable identity of this manager's coordination stream. ###### Inherited from -[`SupervisorNodeContext`](#supervisornodecontext).[`identity`](#identity-2) +[`SupervisorNodeContext`](#supervisornodecontext).[`identity`](#identity-3) ##### assignmentId? @@ -22787,6 +22854,30 @@ whose exact reference a crash would lose. *** +### startRetainedRunInEnvironment() + +> **startRetainedRunInEnvironment**(`options`): `Promise`\<[`RetainedRunHandle`](#retainedrunhandle)\> + +**`Stable`** + +Dispatch a fresh retained session inside an existing provider environment. + +This operation reuses only the environment. It does not append to a prior +harness chat and does not claim native conversation continuity. The caller +must use `RetainedRunHandle.continueNative` for a verified same-chat turn. + +#### Parameters + +##### options + +[`StartRetainedRunInEnvironmentOptions`](#startretainedruninenvironmentoptions) + +#### Returns + +`Promise`\<[`RetainedRunHandle`](#retainedrunhandle)\> + +*** + ### recoverRetainedRun() > **recoverRetainedRun**(`options`): `Promise`\<[`RecoverRetainedRunResult`](#recoverretainedrunresult)\> diff --git a/docs/api/testing.md b/docs/api/testing.md index ea20dae4..f38e8d18 100644 --- a/docs/api/testing.md +++ b/docs/api/testing.md @@ -466,7 +466,7 @@ Product authority over every steer/answer instruction (the filter seam). `runGra ###### Inherited from -[`RunGraphOptions`](runtime.md#rungraphoptions).[`now`](runtime.md#now-7) +[`RunGraphOptions`](runtime.md#rungraphoptions).[`now`](runtime.md#now-8) ##### otel? @@ -1187,7 +1187,7 @@ Give the supervisor brain a chapter-lifecycle on its OWN context window (router ###### Inherited from -[`SuperviseOptions`](runtime.md#superviseoptions).[`now`](runtime.md#now-13) +[`SuperviseOptions`](runtime.md#superviseoptions).[`now`](runtime.md#now-14) ##### allowedModels? diff --git a/docs/canonical-api.md b/docs/canonical-api.md index 23ac47f6..bb8f7efd 100644 --- a/docs/canonical-api.md +++ b/docs/canonical-api.md @@ -141,6 +141,7 @@ A general "loop" primitive is the single most common modelling error in this rep | Run **agent-eval fixture folders** through Runtime `runAgentRounds` | agent-eval fixture loading/planning, then `loopCampaignDispatch(...)`: `/kernel`; it starts the Runtime cell inside Eval's paid-call lifecycle | a one-off `runCampaign` dispatch, or attaching a completed `LoopResult` after paid work already ran | | Run a **recursive `supervise()` tree** through an agent-eval profile matrix | `superviseDispatch({ toTask, toSuperviseOptions, ... })`: `/kernel`; it admits the tree through Eval before Runtime spends, then records its receipt only when Runtime proves one model. Mixed or unknown trees fail instead of being relabelled. | a Lab receipt mapper, a second scheduler, or attaching a completed `SupervisedResult` after paid work already ran | | Run + **resume** ONE persistent box across turns | `openSandboxRun(client, opts, deliverable)`: `/kernel` | a per-domain `new Sandbox`+`box.fs.read`+delete copy | +| Start a retry-safe detached run in a new environment, or a fresh harness chat in one existing environment | `startRetainedRun(...)` or `startRetainedRunInEnvironment(...)`: `/kernel`; both persist exact coordinates before and after dispatch, while only `continueNative(...)` may claim same-chat continuity | calling `provider.create/get/dispatch` directly, reusing an environment as proof of chat continuity, or appending to an unverified native session | | Run **ONE agent turn** on any substrate: box (`streamPrompt`), cli-bridge/router `Executor`, or in-process chat backend: as ONE normalized `RuntimeStreamEvent` stream with a guaranteed terminal result+usage event; opt into in-stream `tool_call`/`tool_result` with `preserveToolParts`, or tap the raw sandbox events with `onRawEvent` | `streamAgentTurn(backend, prompt, { signal, timeoutMs, preserveToolParts?, onRawEvent? })` + `collectAgentTurn(stream)`: `/kernel` | a per-provider stream→event mapper zoo, a hand-faked box around a non-box executor, or raw fetch leaking through the turn abstraction | | Use an exact profile and Runtime executor where `runAgentTaskStream` or a conversation expects an `AgentExecutionBackend` | `createProfileExecutionBackend({ profile, executor: createExecutor(config) })`: root `.`; the adapter preserves conversation authorization, recursion-depth, and trace headers | a provider-specific backend constructor or an adapter that reads a second model/prompt configuration | | Pick the **execution transport a driven loop runs on** (`sandbox` box / cli-bridge / router) from a product flag | `resolveSandboxClient({ backend })`: `/kernel` | a per-product `if (backend === 'router') …` branch re-wiring `createExecutor` + `inlineSandboxClient` | diff --git a/src/runtime/index.ts b/src/runtime/index.ts index 7204a078..5f3f60d6 100644 --- a/src/runtime/index.ts +++ b/src/runtime/index.ts @@ -363,8 +363,10 @@ export { type RetainedRunSnapshot, reconnectRetainedRun, recoverRetainedRun, + type StartRetainedRunInEnvironmentOptions, type StartRetainedRunOptions, startRetainedRun, + startRetainedRunInEnvironment, } from './retained-run' // Router requests are an internal transport adapter. Public execution always enters through an // exact AgentProfile (`createExecutor` + `streamAgentTurn`); callers may configure only the diff --git a/src/runtime/retained-run-binding.ts b/src/runtime/retained-run-binding.ts index ea787813..7d60fff1 100644 --- a/src/runtime/retained-run-binding.ts +++ b/src/runtime/retained-run-binding.ts @@ -264,19 +264,31 @@ export function assertEventBinding( controlRef: AgentExactRunControlRef, ): void { const transport = isRecord(source) ? source : undefined - const record = isRecord(source.data) ? source.data : undefined + const payload = isRecord(source.data) ? source.data : undefined const providerEvent = isRecord(source.providerEvent) ? source.providerEvent : undefined - const values = [transport, record, providerEvent].filter( - (value): value is Record => value !== undefined, - ) - for (const value of values) { + const bindings = [ + { value: transport, checkSessionId: true }, + { + value: payload, + // The canonical session.updated payload identifies the harness-native chat. + // It is event content, not the retained provider session coordinate. + checkSessionId: source.type !== 'session.updated', + }, + { value: providerEvent, checkSessionId: true }, + ] + for (const { value, checkSessionId } of bindings) { + if (value === undefined) continue if (value.runId !== undefined && value.runId !== controlRef.runId) { throw new Error('provider returned an event for another retained run') } if (value.executionId !== undefined && value.executionId !== controlRef.executionId) { throw new Error('provider returned an event for another retained execution') } - if (value.sessionId !== undefined && value.sessionId !== controlRef.sessionId) { + if ( + checkSessionId && + value.sessionId !== undefined && + value.sessionId !== controlRef.sessionId + ) { throw new Error('provider returned an event for another retained session') } } diff --git a/src/runtime/retained-run-start.ts b/src/runtime/retained-run-start.ts index 64182381..bf0cb5c1 100644 --- a/src/runtime/retained-run-start.ts +++ b/src/runtime/retained-run-start.ts @@ -3,6 +3,7 @@ import { AgentExactRunControlRefSchema, } from '@tangle-network/agent-interface' import type { + AgentEnvironment, AgentEnvironmentCapabilities, AgentEnvironmentProvider, AgentSession, @@ -22,6 +23,7 @@ import type { RetainedRunAdmission, RetainedRunAdmissionHook, RetainedRunHandle, + StartRetainedRunInEnvironmentOptions, StartRetainedRunOptions, } from './retained-run-types' import { freshTurnInput } from './turn-input' @@ -107,22 +109,101 @@ export async function startRetainedRun( throw new Error(`provider "${options.provider.name}" does not expose detached session control`) } - // The environment admission fires only for a dispatch-capable environment: - // an environment destroyed above must never leave a durable record. + return dispatchRetainedRun({ + provider: options.provider, + environment, + environmentIdempotencyKey: options.environment.idempotencyKey, + turn: options.turn, + identity, + onAdmission: options.onAdmission, + capabilities, + now: options.now, + }) +} + +/** + * Dispatch a fresh retained session inside an existing provider environment. + * + * This operation reuses only the environment. It does not append to a prior + * harness chat and does not claim native conversation continuity. The caller + * must use `RetainedRunHandle.continueNative` for a verified same-chat turn. + * + * @stable + */ +export async function startRetainedRunInEnvironment( + options: StartRetainedRunInEnvironmentOptions, +): Promise { + assertStableText(options.environment.id, 'retained environment id') + assertStableText(options.environment.idempotencyKey, 'environment idempotency key') + assertStableText(options.turn.turnId, 'turn idempotency key') + if (options.identity !== undefined) { + assertStableText(options.identity.sessionId, 'retained session id') + assertStableText(options.identity.executionId, 'retained execution id') + } + if (typeof options.onAdmission !== 'function') { + throw new Error('startRetainedRunInEnvironment requires an awaited onAdmission durability hook') + } + const identity = + options.identity ?? + mintRetainedIdentity(options.environment.idempotencyKey, options.turn.turnId) + const capabilities = await assertRetainedCapabilities(options.provider) + if (!options.provider.get) { + throw new Error(`provider "${options.provider.name}" cannot reconstruct an environment by id`) + } + const environment = await options.provider.get(options.environment.id) + if (!environment) { + throw new Error( + `provider "${options.provider.name}" no longer holds environment "${options.environment.id}"`, + ) + } + if (environment.id !== options.environment.id || environment.provider !== options.provider.name) { + throw new Error('provider reconstructed a different retained environment') + } + if (!environment.dispatch || !environment.session) { + throw new Error(`provider "${options.provider.name}" does not expose detached session control`) + } + + return dispatchRetainedRun({ + provider: options.provider, + environment, + environmentIdempotencyKey: options.environment.idempotencyKey, + turn: options.turn, + identity, + onAdmission: options.onAdmission, + capabilities, + now: options.now, + }) +} + +interface DispatchRetainedRunOptions { + readonly provider: AgentEnvironmentProvider + readonly environment: AgentEnvironment + readonly environmentIdempotencyKey: string + readonly turn: StartRetainedRunOptions['turn'] + readonly identity: { readonly sessionId: string; readonly executionId: string } + readonly onAdmission: RetainedRunAdmissionHook + readonly capabilities: AgentEnvironmentCapabilities + readonly now?: () => number +} + +async function dispatchRetainedRun( + options: DispatchRetainedRunOptions, +): Promise { + const { environment, identity } = options + // The environment admission fires only for a dispatch-capable environment. await admitDurably(options.onAdmission, { phase: 'environment', provider: options.provider.name, environmentId: environment.id, - idempotencyKey: options.environment.idempotencyKey, + idempotencyKey: options.environmentIdempotencyKey, turnId: options.turn.turnId, sessionId: identity.sessionId, executionId: identity.executionId, }) // Once dispatch begins, its outcome may be unknown to this process. Keep the - // idempotently-created environment so a retry or reconnect can recover the - // retained operation instead of destroying work that may already be live. - const reference = await environment.dispatch( + // environment so a retry or reconnect can recover work that may already be live. + const reference = await environment.dispatch!( freshTurnInput(options.turn, { turnId: options.turn.turnId, detach: true, @@ -168,14 +249,14 @@ export async function startRetainedRun( await admitDurably(options.onAdmission, { phase: 'dispatched', controlRef: freezeControlRef(exact.controlRef), - idempotencyKey: options.environment.idempotencyKey, + idempotencyKey: options.environmentIdempotencyKey, turnId: options.turn.turnId, }) return createRetainedRunHandle( environment, exact.session, exact.controlRef, - capabilities, + options.capabilities, options.now, ) } diff --git a/src/runtime/retained-run-types.ts b/src/runtime/retained-run-types.ts index ed88024f..0d74d5f2 100644 --- a/src/runtime/retained-run-types.ts +++ b/src/runtime/retained-run-types.ts @@ -142,6 +142,28 @@ export interface StartRetainedRunOptions { readonly now?: () => number } +/** A fresh retained session inside a provider environment that already exists. @stable */ +export interface StartRetainedRunInEnvironmentOptions { + readonly provider: AgentEnvironmentProvider + readonly environment: { + /** Stable provider environment identifier used by `provider.get`. */ + readonly id: string + /** Original environment key retained for deterministic run identity and recovery records. */ + readonly idempotencyKey: string + } + readonly turn: AgentTurnInput & { turnId: string } + /** + * Explicit fresh-session coordinates. When omitted, the runtime mints them + * from `(environment.idempotencyKey, turn.turnId)`. + */ + readonly identity?: { + readonly sessionId: string + readonly executionId: string + } + readonly onAdmission: RetainedRunAdmissionHook + readonly now?: () => number +} + /** Inputs sufficient to rebuild a control client in a new process. @stable */ export interface ReconnectRetainedRunOptions { readonly provider: AgentEnvironmentProvider diff --git a/src/runtime/retained-run.test.ts b/src/runtime/retained-run.test.ts index 094eeec7..d668489d 100644 --- a/src/runtime/retained-run.test.ts +++ b/src/runtime/retained-run.test.ts @@ -27,6 +27,7 @@ import { reconnectRetainedRun, recoverRetainedRun, startRetainedRun, + startRetainedRunInEnvironment, } from './retained-run' import { mintRetainedIdentity } from './retained-run-start' @@ -247,6 +248,152 @@ describe('retained runtime run control', () => { expect(unusableRecorder.admissions).toEqual([]) }) + it('starts a fresh retained session inside an existing environment', async () => { + const identity = mintRetainedIdentity('durable-environment-key', 'fresh-workspace-turn') + const controlRef = { + runId: 'fresh-run-in-existing-environment', + provider: 'test-provider', + environmentId: 'environment-1', + ...identity, + requestDigest: retainedRequestDigest, + } + const session: AgentSession = { + id: identity.sessionId, + controlRef, + status: async () => 'running', + async *events() { + yield* [] + }, + result: async () => ({ + text: 'same workspace, fresh harness chat', + success: true, + sessionId: identity.sessionId, + metadata: { + runId: controlRef.runId, + executionId: controlRef.executionId, + requestDigest: controlRef.requestDigest, + }, + }), + prompt: async () => ({ text: 'continued', success: true }), + cancel: async () => {}, + } + let createCalls = 0 + let destroyCalls = 0 + const getIds: string[] = [] + let dispatched: AgentTurnInput | undefined + const provider = providerWithEnvironment({ + async dispatch(input) { + dispatched = input + return { id: identity.sessionId, provider: 'test-provider', controlRef } + }, + session(id) { + if (id !== identity.sessionId) throw new Error('unexpected session id') + return session + }, + async destroy() { + destroyCalls += 1 + }, + }) + provider.create = async () => { + createCalls += 1 + throw new Error('existing-environment start must not create') + } + const get = provider.get! + provider.get = async (id) => { + getIds.push(id) + return get(id) + } + const recorder = recordedAdmissions() + + const run = await startRetainedRunInEnvironment({ + provider, + environment: { id: 'environment-1', idempotencyKey: 'durable-environment-key' }, + turn: { prompt: 'inspect the existing workspace', turnId: 'fresh-workspace-turn' }, + onAdmission: recorder.onAdmission, + }) + + expect(createCalls).toBe(0) + expect(destroyCalls).toBe(0) + expect(getIds).toEqual(['environment-1']) + expect(dispatched).toEqual({ + prompt: 'inspect the existing workspace', + turnId: 'fresh-workspace-turn', + detach: true, + ...identity, + }) + expect(recorder.admissions).toMatchObject([ + { + phase: 'environment', + environmentId: 'environment-1', + idempotencyKey: 'durable-environment-key', + turnId: 'fresh-workspace-turn', + ...identity, + }, + { + phase: 'dispatched', + idempotencyKey: 'durable-environment-key', + turnId: 'fresh-workspace-turn', + controlRef, + }, + ]) + await expect(run.result()).resolves.toMatchObject({ + text: 'same workspace, fresh harness chat', + sessionId: identity.sessionId, + }) + }) + + it('fails before admission when an existing retained environment is unavailable or unusable', async () => { + const missing = providerWithEnvironment({}) + missing.get = async () => null + const missingRecorder = recordedAdmissions() + await expect( + startRetainedRunInEnvironment({ + provider: missing, + environment: { id: 'missing-environment', idempotencyKey: 'missing-key' }, + turn: { prompt: 'go', turnId: 'missing-turn' }, + onAdmission: missingRecorder.onAdmission, + }), + ).rejects.toThrow('no longer holds environment') + expect(missingRecorder.admissions).toEqual([]) + + let destroyCalls = 0 + const unusable = providerWithEnvironment({ + dispatch: undefined, + session: undefined, + async destroy() { + destroyCalls += 1 + }, + }) + const unusableRecorder = recordedAdmissions() + await expect( + startRetainedRunInEnvironment({ + provider: unusable, + environment: { id: 'environment-1', idempotencyKey: 'existing-key' }, + turn: { prompt: 'go', turnId: 'existing-turn' }, + onAdmission: unusableRecorder.onAdmission, + }), + ).rejects.toThrow('does not expose detached session control') + expect(destroyCalls).toBe(0) + expect(unusableRecorder.admissions).toEqual([]) + + const foreign = providerWithEnvironment({}) + const getForeign = foreign.get! + foreign.get = async (id) => { + const environment = await getForeign(id) + return environment === null ? null : { ...environment, provider: 'other-provider' } + } + const foreignRecorder = recordedAdmissions() + await expect( + startRetainedRunInEnvironment({ + provider: foreign, + environment: { id: 'environment-1', idempotencyKey: 'existing-key' }, + turn: { prompt: 'go', turnId: 'foreign-turn' }, + onAdmission: foreignRecorder.onAdmission, + }), + ).rejects.toThrow('reconstructed a different retained environment') + expect(foreignRecorder.admissions).toEqual([]) + }) + it('allowlists a fresh retained start when JavaScript supplies stale run fields', async () => { const controlRef = { runId: 'fresh-run', @@ -1232,6 +1379,113 @@ describe('retained runtime run control', () => { ]) }) + it('preserves a harness-native id while binding session.updated to its execution', async () => { + const controlRef = { + runId: 'native-session-run', + provider: 'test-provider', + environmentId: 'environment-1', + sessionId: 'retained-provider-session', + executionId: 'native-session-execution', + requestDigest: retainedRequestDigest, + } + const session: AgentSession = { + id: controlRef.sessionId, + controlRef, + status: async () => 'running', + async *events() { + yield { + id: 'native-session-event', + type: 'session.updated', + data: { + sessionId: 'harness-native-session', + runId: controlRef.runId, + executionId: controlRef.executionId, + }, + normalized: { + type: 'session.updated', + sessionId: 'harness-native-session', + }, + } + yield { + id: 'foreign-native-session-event', + type: 'session.updated', + data: { + sessionId: 'another-harness-session', + runId: controlRef.runId, + executionId: 'foreign-execution', + }, + normalized: { + type: 'session.updated', + sessionId: 'another-harness-session', + }, + } + }, + result: async () => ({ text: 'done', success: true }), + prompt: async () => ({ text: 'continued', success: true }), + cancel: async () => {}, + } + const provider = providerWithEnvironment({ + dispatch: async () => ({ id: session.id, provider: 'test-provider', controlRef }), + session: () => session, + }) + const run = await startRetainedRun({ + provider, + environment: { profile: { name: 'worker' }, idempotencyKey: 'native-session' }, + turn: { prompt: 'go', turnId: 'native-session-turn' }, + onAdmission: recordedAdmissions().onAdmission, + identity: { sessionId: controlRef.sessionId, executionId: controlRef.executionId }, + }) + + const events = run.events()[Symbol.asyncIterator]() + await expect(events.next()).resolves.toMatchObject({ + value: { + eventId: 'native-session-event', + event: { type: 'session.updated', sessionId: 'harness-native-session' }, + }, + }) + await expect(events.next()).rejects.toThrow('another retained execution') + }) + + it('still rejects a foreign retained session on lifecycle event payloads', async () => { + const controlRef = { + runId: 'foreign-session-run', + provider: 'test-provider', + environmentId: 'environment-1', + sessionId: 'expected-retained-session', + executionId: 'foreign-session-execution', + requestDigest: retainedRequestDigest, + } + const session: AgentSession = { + id: controlRef.sessionId, + controlRef, + status: async () => 'running', + async *events() { + yield { + id: 'foreign-session-event', + type: 'status', + data: { sessionId: 'foreign-retained-session' }, + normalized: { type: 'status', status: 'processing' }, + } + }, + result: async () => ({ text: 'done', success: true }), + prompt: async () => ({ text: 'continued', success: true }), + cancel: async () => {}, + } + const provider = providerWithEnvironment({ + dispatch: async () => ({ id: session.id, provider: 'test-provider', controlRef }), + session: () => session, + }) + const run = await startRetainedRun({ + provider, + environment: { profile: { name: 'worker' }, idempotencyKey: 'foreign-session' }, + turn: { prompt: 'go', turnId: 'foreign-session-turn' }, + onAdmission: recordedAdmissions().onAdmission, + identity: { sessionId: controlRef.sessionId, executionId: controlRef.executionId }, + }) + + await expect(collectRetainedEvents(run.events())).rejects.toThrow('another retained session') + }) + it('validates a replay anchor before skipping it', async () => { const controlRef = { runId: 'anchor-run', diff --git a/src/runtime/retained-run.ts b/src/runtime/retained-run.ts index d398e905..1fff575f 100644 --- a/src/runtime/retained-run.ts +++ b/src/runtime/retained-run.ts @@ -5,7 +5,12 @@ * startup, replay, binding checks, and handle operations. */ -export { reconnectRetainedRun, recoverRetainedRun, startRetainedRun } from './retained-run-start' +export { + reconnectRetainedRun, + recoverRetainedRun, + startRetainedRun, + startRetainedRunInEnvironment, +} from './retained-run-start' export type { NativeContextContinuationExecution, NativeContextContinuationInput, @@ -23,5 +28,6 @@ export type { RetainedRunHandle, RetainedRunReplayPoint, RetainedRunSnapshot, + StartRetainedRunInEnvironmentOptions, StartRetainedRunOptions, } from './retained-run-types'