fix(cli): wait for armed goal completion

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,144 @@ 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<void>();
+    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('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<void>();
+    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<void>();
+    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<void>();
+    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<void>();
+    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 +1007,8 @@ function runFixture(input: {
   onGraphStop?: () => void;
   initialMessages?: StoredMessage[];
   finalMessages?: StoredMessage[];
+  goal?: GoalProjection | null;
+  onGoalWaitStarted?: () => void;
 }) {
   const switches: string[] = [];
   const moves: string[] = [];
@@ -876,6 +1017,10 @@ 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 sessionFailureListeners = new Set<(error: Error) => void>();
+  const startedTurnListeners = new Set<(turn: { events: AsyncIterable<SessionEvent> }) => void>();
+  const goalControls: Array<Record<string, unknown>> = [];
   const transcriptListeners = new Set<
     (
       sessionId: string,
@@ -885,6 +1030,7 @@ function runFixture(input: {
     ) => void
   >();
   let messageReads = 0;
+  let currentGoal = input.goal ?? null;
   const preparedMaxSteps: Array<number | undefined> = [];
   const driver = {
     createSession: async () => sessionSummary('session-created'),
@@ -934,6 +1080,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 {
@@ -978,7 +1130,14 @@ function runFixture(input: {
     stop: async () => {
       turnStops += 1;
     },
-    subscribeStartedTurns: () => () => {},
+    subscribeStartedTurns: (listener: (turn: { events: AsyncIterable<SessionEvent> }) => void) => {
+      startedTurnListeners.add(listener);
+      return () => startedTurnListeners.delete(listener);
+    },
+    subscribeSessionFailures: (listener: (error: Error) => void) => {
+      sessionFailureListeners.add(listener);
+      return () => sessionFailureListeners.delete(listener);
+    },
     subscribeTranscriptReplacements: (
       listener: (
         sessionId: string,
@@ -1002,6 +1161,17 @@ function runFixture(input: {
         await input.graphQueryGate;
         return { status: input.graphQueryStatus ?? 'completed' };
       }
+      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?.();
@@ -1046,6 +1216,7 @@ function runFixture(input: {
     switches,
     moves,
     graphStops,
+    goalControls,
     exactTurnStops,
     preparedMaxSteps,
     sandboxResponses,
@@ -1062,6 +1233,16 @@ 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));
+    },
+    publishSessionFailure(error: Error) {
+      for (const listener of sessionFailureListeners) listener(error);
+    },
+    publishStartedTurn(events: AsyncIterable<SessionEvent>) {
+      for (const listener of startedTurnListeners) listener({ events });
+    },
     get turnStops() {
       return turnStops;
     },
@@ -1327,6 +1508,56 @@ function graphMessages(includeTerminal = true): StoredMessage[] {
   return messages;
 }
 
+function goalProjection(overrides: Partial<GoalProjection> = {}): 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,

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<Error>();
+    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(

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<void>;
+  };
   agentGraph?: {
     reserveActivity(sessionId: string): { release(): void };
     waitForCompletion(sessionId: string): Promise<void>;
@@ -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;