From 527acd9501d9eae6c32d63b36cd0642cffb28ded Mon Sep 17 00:00:00 2001 From: cat0825 <1759138827@qq.com> Date: Sun, 30 Aug 2026 00:23:43 +0800 Subject: [PATCH 1/2] fix(cli): wait for armed goal completion Keep blocking maka run invocations attached to a self-armed Goal until its durable terminal turn is available. Generated-by: OpenAI Codex --- .../runtime-host-run-command.test.ts | 112 ++++++++++++++++++ packages/cli/src/run-command-core.ts | 6 + packages/cli/src/runtime-host-run-command.ts | 80 +++++++++++++ 3 files changed, 198 insertions(+) diff --git a/packages/cli/src/__tests__/runtime-host-run-command.test.ts b/packages/cli/src/__tests__/runtime-host-run-command.test.ts index 7f5dc70c83..3a8e5eb3a5 100644 --- a/packages/cli/src/__tests__/runtime-host-run-command.test.ts +++ b/packages/cli/src/__tests__/runtime-host-run-command.test.ts @@ -28,6 +28,7 @@ import { import { LOCAL_RUNTIME_HOST_PROFILE, type RuntimeHostConnection } from '@maka/runtime-host/client'; import { SESSION_CONTINUITY_SCHEMA_VERSION, + type GoalProjection, type InteractionPendingSnapshot, type SessionCatalogProjection, type SessionContinuitySnapshot, @@ -414,6 +415,50 @@ describe('Runtime Host maka run adapter', () => { assert.equal(observed.at(-1)?.finalOutput, 'Final graph answer'); }); + test('waits for a self-armed Goal and reports its final Turn', async () => { + const stdout: string[] = []; + const goalWaitStarted = deferred(); + const fixture = runFixture({ + goal: goalProjection(), + onGoalWaitStarted: () => goalWaitStarted.resolve(), + initialMessages: goalMessages(), + }); + const command = runFixtureCommand(fixture, ['arm a goal'], (text) => stdout.push(text)); + + const firstBoundary = await Promise.race([ + command.then(() => 'returned' as const), + goalWaitStarted.promise.then(() => 'waiting' as const), + ]); + assert.equal(firstBoundary, 'waiting', 'maka run returned while its Goal was active'); + + fixture.publishGoal(goalProjection({ revision: 2, status: 'achieved', achievedAt: 10 })); + + assert.equal(await command, 0); + assert.equal(stdout.join(''), 'Final Goal answer\n'); + }); + + test('releases a pending Goal wait when the run context closes', async () => { + const goalWaitStarted = deferred(); + const fixture = runFixture({ + goal: goalProjection(), + onGoalWaitStarted: () => goalWaitStarted.resolve(), + }); + const context = fixture.context; + const session = await context.runtime.createSession({ + cwd: '/workspace', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + permissionMode: 'ask', + }); + const waiting = context.goal?.waitForCompletion(session.id); + assert.ok(waiting); + await goalWaitStarted.promise; + + await context.close(); + + await assert.rejects(waiting, new Error('Runtime Host run context closed')); + }); + test('uses the durable Graph supervisor outcome independently of live projection', async () => { const observed: MakaRunOutcome[] = []; const fixture = runFixture({ observed, graph: true }); @@ -868,6 +913,8 @@ function runFixture(input: { onGraphStop?: () => void; initialMessages?: StoredMessage[]; finalMessages?: StoredMessage[]; + goal?: GoalProjection | null; + onGoalWaitStarted?: () => void; }) { const switches: string[] = []; const moves: string[] = []; @@ -876,6 +923,7 @@ function runFixture(input: { const sandboxResponses: { requestId: string; decision: 'deny' }[] = []; let turnStops = 0; const pendingInteractionListeners = new Set<(pending: InteractionPendingSnapshot) => void>(); + const goalListeners = new Set<(goal: GoalProjection | null) => void>(); const transcriptListeners = new Set< ( sessionId: string, @@ -885,6 +933,7 @@ function runFixture(input: { ) => void >(); let messageReads = 0; + let currentGoal = input.goal ?? null; const preparedMaxSteps: Array = []; const driver = { createSession: async () => sessionSummary('session-created'), @@ -934,6 +983,12 @@ function runFixture(input: { } return () => pendingInteractionListeners.delete(listener); }, + getGoal: () => structuredClone(currentGoal), + subscribeGoalChanges: (listener: (goal: GoalProjection | null) => void) => { + goalListeners.add(listener); + input.onGoalWaitStarted?.(); + return () => goalListeners.delete(listener); + }, switchSession: async (sessionId: string) => { switches.push(sessionId); return { @@ -1002,6 +1057,9 @@ function runFixture(input: { await input.graphQueryGate; return { status: input.graphQueryStatus ?? 'completed' }; } + if (operation === 'goal.query') { + return { sessionId: 'session-created', goal: structuredClone(currentGoal) }; + } if (operation === 'agent.graph.stop') { graphStops.push(String(requestInput.rootSessionId)); input.onGraphStop?.(); @@ -1062,6 +1120,10 @@ function runFixture(input: { listener('session-created', turnId, structuredClone(messages), reason); } }, + publishGoal(goal: GoalProjection | null) { + currentGoal = structuredClone(goal); + for (const listener of goalListeners) listener(structuredClone(goal)); + }, get turnStops() { return turnStops; }, @@ -1327,6 +1389,56 @@ function graphMessages(includeTerminal = true): StoredMessage[] { return messages; } +function goalProjection(overrides: Partial = {}): GoalProjection { + return { + goalId: 'goal-1', + revision: 1, + sessionId: 'session-created', + condition: 'Finish the work', + status: 'active', + setAt: 1, + iterations: 0, + maxIterations: 50, + consecutiveNoProgress: 0, + blockCap: 8, + tokenBudget: null, + tokensSpent: 0, + lastReason: null, + achievedAt: null, + pausedAt: null, + ...overrides, + }; +} + +function goalMessages(): StoredMessage[] { + return [ + { + type: 'user', + id: 'user-goal-turn', + turnId: 'turn-goal', + ts: 3, + text: '[Goal continuation] Keep working.', + origin: { kind: 'goal', goalId: 'goal-1' }, + }, + { + type: 'assistant', + id: 'assistant-goal-turn', + turnId: 'turn-goal', + ts: 4, + text: 'Final Goal answer', + modelId: 'gpt-5', + }, + { + type: 'turn_state', + id: 'state-goal-turn', + turnId: 'turn-goal', + ts: 5, + status: 'completed', + partialOutputRetained: false, + }, + ]; +} + function sandboxBoundaryMessages( failureStepId: string | undefined, successStepId: string | undefined, diff --git a/packages/cli/src/run-command-core.ts b/packages/cli/src/run-command-core.ts index 2fba37b35e..a988cfc2ec 100644 --- a/packages/cli/src/run-command-core.ts +++ b/packages/cli/src/run-command-core.ts @@ -70,6 +70,9 @@ export interface MakaRunRuntime { export interface MakaRunContext { runtime: MakaRunRuntime; target: { connection: { slug: string }; model: string }; + goal?: { + waitForCompletion(sessionId: string): Promise; + }; agentGraph?: { reserveActivity(sessionId: string): { release(): void }; waitForCompletion(sessionId: string): Promise; @@ -426,6 +429,9 @@ export async function runMakaTextCliCore( if (parsed.options.graph && outcome?.status === 'completed') { await Promise.race([context.agentGraph!.waitForCompletion(session.id), stopSignal]); } + if (outcome?.status === 'completed' && context.goal) { + await Promise.race([context.goal.waitForCompletion(session.id), stopSignal]); + } await stopPromise; } catch (error) { streamFailed = true; diff --git a/packages/cli/src/runtime-host-run-command.ts b/packages/cli/src/runtime-host-run-command.ts index 62f8536e42..214b2cfab9 100644 --- a/packages/cli/src/runtime-host-run-command.ts +++ b/packages/cli/src/runtime-host-run-command.ts @@ -33,6 +33,8 @@ import { } from '@maka/runtime-host/client'; import { runtimeHostProfileUsesHostWorkspace } from '@maka/runtime-host/profile-kind'; import type { InteractionPendingSnapshot, SessionCatalogItem } from '@maka/runtime-host/protocol'; +import type { GoalProjection } from '@maka/runtime-host/protocol'; +import { TERMINAL_GOAL_STATUSES } from '@maka/runtime/goal-state'; import { runMakaTextCliCore, type MakaRunContext, @@ -188,6 +190,9 @@ export function createRuntimeHostRunContext( return { runtime, target: { connection: { slug: target.connection.slug }, model: target.model }, + goal: { + waitForCompletion: (sessionId: string) => runtime.waitForGoalCompletion(sessionId), + }, ...(input.enableAgentGraph ? { agentGraph: { @@ -269,6 +274,10 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { timer: ReturnType; }> >(); + readonly #goalWaiters = new Set<{ + reject(error: Error): void; + unsubscribe(): void; + }>(); constructor( connection: RuntimeHostConnection, @@ -415,11 +424,28 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { if (outcome) await this.#observer?.(outcome); } + async waitForGoalCompletion(sessionId: string): Promise { + await this.#attach(sessionId); + const current = (await this.#connection.request('goal.query', { sessionId })).goal; + if (!current || TERMINAL_GOAL_STATUSES.has(current.status)) return; + + await this.#waitForGoalTerminal(sessionId, current.goalId); + const messages = await this.#driver.readMessages(); + const turnId = lastGoalTurnId(messages, current.goalId); + if (!turnId) return; + const outcome = outcomeFromStoredTurn(messages, turnId); + if (!outcome) { + throw new Error('Goal final Turn did not reach a durable terminal boundary'); + } + await this.#observer?.(outcome); + } + close(): Promise { this.#closed = true; this.#interactions.close(); this.#unsubscribeTranscriptReplacements(); this.#cancelGraphTerminalWaiters(new Error('Runtime Host run context closed')); + this.#cancelGoalWaiters(new Error('Runtime Host run context closed')); return Promise.resolve(); } @@ -521,6 +547,44 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { }); } + #waitForGoalTerminal(sessionId: string, goalId: string): Promise { + if (this.#closed) return Promise.reject(new Error('Runtime Host run context closed')); + const subscribe = this.#driver.subscribeGoalChanges; + if (!subscribe) return Promise.reject(new Error('Runtime Host Goal updates are unavailable')); + + return new Promise((resolve, reject) => { + let settled = false; + let unsubscribe = () => {}; + const waiter = { + reject: (error: Error) => finish(undefined, error), + unsubscribe: () => unsubscribe(), + }; + const finish = (goal: GoalProjection | null | undefined, error?: Error) => { + if (settled) return; + settled = true; + unsubscribe(); + this.#goalWaiters.delete(waiter); + if (error) reject(error); + else resolve(goal ?? null); + }; + const accept = (goal: GoalProjection | null) => { + if (goal?.goalId === goalId && !TERMINAL_GOAL_STATUSES.has(goal.status)) return; + finish(goal); + }; + + unsubscribe = subscribe.call(this.#driver, accept); + if (settled) { + unsubscribe(); + return; + } + this.#goalWaiters.add(waiter); + void this.#connection.request('goal.query', { sessionId }).then( + (result) => accept(result.goal), + (error) => finish(undefined, error instanceof Error ? error : new Error(String(error))), + ); + }); + } + async #stopForInteraction(pending: InteractionPendingSnapshot): Promise { this.#stopRequested = true; const stops: Promise[] = [ @@ -547,6 +611,11 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { } this.#graphTerminalWaiters.clear(); } + + #cancelGoalWaiters(error: Error): void { + for (const waiter of [...this.#goalWaiters]) waiter.reject(error); + this.#goalWaiters.clear(); + } } function runtimeHostSessionSummaries(items: readonly SessionCatalogItem[]): SessionSummary[] { @@ -802,6 +871,17 @@ function lastNewGraphSupervisorTurnId( )?.turnId; } +function lastGoalTurnId(messages: readonly StoredMessage[], goalId: string): string | undefined { + return [...messages] + .reverse() + .find( + (message) => + message.type === 'user' && + message.origin?.kind === 'goal' && + message.origin.goalId === goalId, + )?.turnId; +} + function outcomeFromStoredTurn( messages: readonly StoredMessage[], turnId: string, From f23b11dc9fa6e6356da2860edd6d164d2d46157b Mon Sep 17 00:00:00 2001 From: cat0825 <1759138827@qq.com> Date: Sun, 30 Aug 2026 16:02:18 +0800 Subject: [PATCH 2/2] fix(cli): close goal lifecycle gaps Generated-by: OpenAI Codex --- .../runtime-host-run-command.test.ts | 121 +++++++++++++++++- .../runtime-host-session-driver.test.ts | 34 +++++ packages/cli/src/runtime-host-run-command.ts | 70 +++++++++- .../cli/src/runtime-host-session-driver.ts | 11 ++ 4 files changed, 230 insertions(+), 6 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-run-command.test.ts b/packages/cli/src/__tests__/runtime-host-run-command.test.ts index 3a8e5eb3a5..470053b481 100644 --- a/packages/cli/src/__tests__/runtime-host-run-command.test.ts +++ b/packages/cli/src/__tests__/runtime-host-run-command.test.ts @@ -437,6 +437,100 @@ describe('Runtime Host maka run adapter', () => { assert.equal(stdout.join(''), 'Final Goal answer\n'); }); + test('projects an already-terminal Goal final Turn before returning', async () => { + const stdout: string[] = []; + const fixture = runFixture({ + goal: goalProjection({ status: 'achieved', achievedAt: 10 }), + initialMessages: goalMessages(), + }); + + const exitCode = await runFixtureCommand(fixture, ['finish immediately'], (text) => + stdout.push(text), + ); + + assert.equal(exitCode, 0); + assert.equal(stdout.join(''), 'Final Goal answer\n'); + }); + + test('fails instead of waiting forever when a Goal pauses', async () => { + const stderr: string[] = []; + const goalWaitStarted = deferred(); + const fixture = runFixture({ + goal: goalProjection({ status: 'waiting' }), + onGoalWaitStarted: () => goalWaitStarted.resolve(), + }); + const command = runFixtureCommand(fixture, ['pause eventually'], undefined, (text) => + stderr.push(text), + ); + await goalWaitStarted.promise; + + fixture.publishGoal(goalProjection({ revision: 2, status: 'paused', pausedAt: 10 })); + + assert.equal(await command, 1); + assert.equal(stderr.join(''), 'maka run: Goal paused before completion\n'); + }); + + test('fails a pending Goal wait when Session recovery is exhausted', async () => { + const goalWaitStarted = deferred(); + const fixture = runFixture({ + goal: goalProjection(), + onGoalWaitStarted: () => goalWaitStarted.resolve(), + }); + const waiting = fixture.context.goal?.waitForCompletion('session-created'); + assert.ok(waiting); + await goalWaitStarted.promise; + + fixture.publishSessionFailure(new Error('Session subscription recovery exhausted')); + + await assert.rejects(waiting, new Error('Session subscription recovery exhausted')); + }); + + test('pauses a waiting Goal when cancellation lands between Turns', async () => { + const goalWaitStarted = deferred(); + const fixture = runFixture({ + goal: goalProjection({ status: 'waiting' }), + onGoalWaitStarted: () => goalWaitStarted.resolve(), + }); + const context = fixture.context; + const session = await context.runtime.createSession({ + cwd: '/workspace', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + permissionMode: 'ask', + }); + const waiting = context.goal?.waitForCompletion(session.id); + assert.ok(waiting); + await goalWaitStarted.promise; + + await context.runtime.stopSession(session.id); + + await assert.rejects(waiting, new Error('Goal paused before completion')); + assert.deepEqual(fixture.goalControls, [ + { + sessionId: session.id, + goalId: 'goal-1', + expectedRevision: 1, + action: 'pause', + }, + ]); + }); + + test('drains live events from Host-started Goal Turns', async () => { + const fixture = runFixture({}); + void fixture.context; + let consumed = 0; + fixture.publishStartedTurn( + (async function* () { + consumed += 1; + yield* completionEvents('turn-goal', 'end_turn'); + })(), + ); + + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(consumed, 1); + }); + test('releases a pending Goal wait when the run context closes', async () => { const goalWaitStarted = deferred(); const fixture = runFixture({ @@ -924,6 +1018,9 @@ function runFixture(input: { let turnStops = 0; const pendingInteractionListeners = new Set<(pending: InteractionPendingSnapshot) => void>(); const goalListeners = new Set<(goal: GoalProjection | null) => void>(); + const sessionFailureListeners = new Set<(error: Error) => void>(); + const startedTurnListeners = new Set<(turn: { events: AsyncIterable }) => void>(); + const goalControls: Array> = []; const transcriptListeners = new Set< ( sessionId: string, @@ -1033,7 +1130,14 @@ function runFixture(input: { stop: async () => { turnStops += 1; }, - subscribeStartedTurns: () => () => {}, + subscribeStartedTurns: (listener: (turn: { events: AsyncIterable }) => void) => { + startedTurnListeners.add(listener); + return () => startedTurnListeners.delete(listener); + }, + subscribeSessionFailures: (listener: (error: Error) => void) => { + sessionFailureListeners.add(listener); + return () => sessionFailureListeners.delete(listener); + }, subscribeTranscriptReplacements: ( listener: ( sessionId: string, @@ -1060,6 +1164,14 @@ function runFixture(input: { if (operation === 'goal.query') { return { sessionId: 'session-created', goal: structuredClone(currentGoal) }; } + if (operation === 'goal.control') { + goalControls.push(structuredClone(requestInput)); + currentGoal = currentGoal + ? { ...currentGoal, revision: currentGoal.revision + 1, status: 'paused', pausedAt: 10 } + : null; + for (const listener of goalListeners) listener(structuredClone(currentGoal)); + return { sessionId: 'session-created', goal: structuredClone(currentGoal) }; + } if (operation === 'agent.graph.stop') { graphStops.push(String(requestInput.rootSessionId)); input.onGraphStop?.(); @@ -1104,6 +1216,7 @@ function runFixture(input: { switches, moves, graphStops, + goalControls, exactTurnStops, preparedMaxSteps, sandboxResponses, @@ -1124,6 +1237,12 @@ function runFixture(input: { currentGoal = structuredClone(goal); for (const listener of goalListeners) listener(structuredClone(goal)); }, + publishSessionFailure(error: Error) { + for (const listener of sessionFailureListeners) listener(error); + }, + publishStartedTurn(events: AsyncIterable) { + for (const listener of startedTurnListeners) listener({ events }); + }, get turnStops() { return turnStops; }, diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 571afe359e..66023885c6 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -2143,6 +2143,40 @@ describe('Runtime Host Maka Session driver', () => { assert.equal(statuses.at(-1), undefined); }); + test('publishes active Session failure when bounded recovery is exhausted', async () => { + const snapshot = continuitySnapshot({ rootTurn: null }); + const initial = new FakeSubscription(snapshot, Promise.resolve([])); + const ended = Array.from({ length: 8 }, (_, index) => { + const subscription = new FakeSubscription( + { ...snapshot, projectionRevision: index + 2 }, + Promise.resolve([]), + `subscription-${index + 2}`, + ); + void subscription.close(); + return subscription; + }); + const connection = new FakeConnection([initial, ...ended], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionId: 'connection-1', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + }); + await driver.switchSession('session-1'); + const failed = deferred(); + driver.subscribeSessionFailures((error) => failed.resolve(error)); + + await initial.close(); + const error = await Promise.race([ + failed.promise, + delay(3_000).then(() => assert.fail('Timed out waiting for Session recovery exhaustion')), + ]); + + assert.match(error.message, /recovery/i); + assert.equal(connection.openedSubscriptions, 9); + }); + test('reopens a failed Session channel before starting the next turn', async () => { const first = new FakeSubscription(continuitySnapshot({ rootTurn: null }), Promise.resolve([])); const second = new FakeSubscription( diff --git a/packages/cli/src/runtime-host-run-command.ts b/packages/cli/src/runtime-host-run-command.ts index 214b2cfab9..c3890c1c28 100644 --- a/packages/cli/src/runtime-host-run-command.ts +++ b/packages/cli/src/runtime-host-run-command.ts @@ -259,6 +259,8 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { readonly #sessionCwdOverride: MakaRunContextInput['sessionCwdOverride']; readonly #maxSteps: number | undefined; readonly #unsubscribeTranscriptReplacements: () => void; + readonly #unsubscribeSessionFailures: () => void; + readonly #unsubscribeStartedTurns: () => void; #sessionId: string | undefined; #activeTurn: ActiveRuntimeHostTurn | undefined; #stopRequested = false; @@ -278,6 +280,8 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { reject(error: Error): void; unsubscribe(): void; }>(); + #sessionFailure: Error | undefined; + readonly #hostStartedTurnDrains = new Set>(); constructor( connection: RuntimeHostConnection, @@ -302,6 +306,17 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { this.#acceptGraphTranscript(messages); }, ); + this.#unsubscribeSessionFailures = driver.subscribeSessionFailures((error) => { + this.#sessionFailure = error; + this.#cancelGraphTerminalWaiters(error); + this.#cancelGoalWaiters(error); + }); + this.#unsubscribeStartedTurns = driver.subscribeStartedTurns((turn) => { + const drain = collectEvents(turn.events) + .catch(() => undefined) + .finally(() => this.#hostStartedTurnDrains.delete(drain)); + this.#hostStartedTurnDrains.add(drain); + }); } async createSession(input: CreateSessionRequest): Promise { @@ -369,6 +384,7 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { if (this.#graphEnabled) { stops.push(this.#stopGraph(sessionId)); } + stops.push(this.#pauseGoal(sessionId)); const settled = await Promise.allSettled(stops); const failure = settled.find( (result): result is PromiseRejectedResult => result.status === 'rejected', @@ -426,10 +442,14 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { async waitForGoalCompletion(sessionId: string): Promise { await this.#attach(sessionId); - const current = (await this.#connection.request('goal.query', { sessionId })).goal; - if (!current || TERMINAL_GOAL_STATUSES.has(current.status)) return; - - await this.#waitForGoalTerminal(sessionId, current.goalId); + let current = (await this.#connection.request('goal.query', { sessionId })).goal; + if (!current) return; + if (current.status === 'paused') throw new Error('Goal paused before completion'); + if (!TERMINAL_GOAL_STATUSES.has(current.status)) { + current = await this.#waitForGoalTerminal(sessionId, current.goalId); + if (!current) return; + if (current.status === 'paused') throw new Error('Goal paused before completion'); + } const messages = await this.#driver.readMessages(); const turnId = lastGoalTurnId(messages, current.goalId); if (!turnId) return; @@ -444,6 +464,8 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { this.#closed = true; this.#interactions.close(); this.#unsubscribeTranscriptReplacements(); + this.#unsubscribeSessionFailures(); + this.#unsubscribeStartedTurns(); this.#cancelGraphTerminalWaiters(new Error('Runtime Host run context closed')); this.#cancelGoalWaiters(new Error('Runtime Host run context closed')); return Promise.resolve(); @@ -501,6 +523,30 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { } } + async #pauseGoal(sessionId: string): Promise { + for (let attempt = 0; attempt < 3; attempt += 1) { + const goal = (await this.#connection.request('goal.query', { sessionId })).goal; + if (!goal || (goal.status !== 'active' && goal.status !== 'waiting')) return; + try { + await this.#connection.request('goal.control', { + sessionId, + goalId: goal.goalId, + expectedRevision: goal.revision, + action: 'pause', + }); + return; + } catch (error) { + if ( + !(error instanceof RuntimeHostOperationError) || + error.code !== 'operation_conflict' || + attempt === 2 + ) { + throw error; + } + } + } + } + #acceptGraphTranscript(messages: readonly StoredMessage[]): void { this.#latestTranscriptReplacement = messages; for (const [turnId, waiters] of this.#graphTerminalWaiters) { @@ -549,6 +595,7 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { #waitForGoalTerminal(sessionId: string, goalId: string): Promise { if (this.#closed) return Promise.reject(new Error('Runtime Host run context closed')); + if (this.#sessionFailure) return Promise.reject(this.#sessionFailure); const subscribe = this.#driver.subscribeGoalChanges; if (!subscribe) return Promise.reject(new Error('Runtime Host Goal updates are unavailable')); @@ -568,7 +615,13 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { else resolve(goal ?? null); }; const accept = (goal: GoalProjection | null) => { - if (goal?.goalId === goalId && !TERMINAL_GOAL_STATUSES.has(goal.status)) return; + if ( + goal?.goalId === goalId && + goal.status !== 'paused' && + !TERMINAL_GOAL_STATUSES.has(goal.status) + ) { + return; + } finish(goal); }; @@ -618,6 +671,13 @@ class RuntimeHostRunRuntime implements MakaRunRuntime { } } +async function collectEvents(events: AsyncIterable): Promise { + for await (const _event of events) { + // Host-started Goal/Graph Turns are projected from durable state. Drain + // their live queue so a long autonomous run cannot force channel recovery. + } +} + function runtimeHostSessionSummaries(items: readonly SessionCatalogItem[]): SessionSummary[] { return items.flatMap((item) => ('kind' in item ? [] : [projectSessionCatalogSummary(item)])); } diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts index 36ba9707c8..e23f03c0fc 100644 --- a/packages/cli/src/runtime-host-session-driver.ts +++ b/packages/cli/src/runtime-host-session-driver.ts @@ -167,6 +167,7 @@ export interface RuntimeHostMakaSessionDriver extends MakaSessionDriver { resumeLatest(): AsyncIterable; subscribePendingInteractions(listener: (pending: InteractionPendingSnapshot) => void): () => void; subscribeStartedTurns(listener: (turn: MakaAttachedSessionTurn) => void): () => void; + subscribeSessionFailures(listener: (error: Error) => void): () => void; subscribeResolvedInteractions( listener: (sessionId: string, requestId: string) => void, ): () => void; @@ -224,6 +225,7 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { #channelGeneration = 0; #transcriptRefreshSequence = 0; readonly #startedTurnListeners = new Set<(turn: MakaAttachedSessionTurn) => void>(); + readonly #sessionFailureListeners = new Set<(error: Error) => void>(); readonly #goalListeners = new Set<(goal: GoalProjection | null) => void>(); readonly #pendingInteractionListeners = new Set<(pending: InteractionPendingSnapshot) => void>(); readonly #claimedTurnIds = new Set(); @@ -1074,6 +1076,11 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { return () => this.#goalListeners.delete(listener); } + subscribeSessionFailures(listener: (error: Error) => void): () => void { + this.#sessionFailureListeners.add(listener); + return () => this.#sessionFailureListeners.delete(listener); + } + async controlGoal(action: GoalControlAction): Promise { const sessionId = this.#sessionId; if (!sessionId) return null; @@ -1525,6 +1532,10 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { if (this.#sessionId !== sessionId || this.#sessionGeneration !== sessionGeneration) return; for (const listener of this.#goalListeners) listener(goal); }, + onFailed: (error) => { + if (this.#sessionId !== sessionId || this.#sessionGeneration !== sessionGeneration) return; + for (const listener of this.#sessionFailureListeners) listener(error); + }, onRecovered: () => this.#refreshRuntimeResources(sessionId), }); }