fix(runtime-host): forward the durable steering echo to session subscribers

packages/runtime-host/src/__tests__/execution-host-message.test.ts

@@ -95,6 +95,64 @@ import {
   withTimeout,
 } from './fixtures/execution-host-suite.js';
 
+test('subscribed Clients receive the durable steering echo as a session event', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const host = await fixture.startHost();
+    const client = await connectClient(fixture.root);
+    const subscription = await client.openSessionSubscription({
+      sessionId: fixture.sessionId,
+      transcript: { kind: 'none' },
+    });
+    const probe = new SubscriptionProbe(subscription);
+
+    const turnId = randomUUID();
+    await client.startTurn({
+      sessionId: fixture.sessionId,
+      turnId,
+      content: { text: FAKE_WAIT_FOR_STEERING_PROMPT },
+    });
+    const steeringId = randomUUID();
+    const steeringContent = {
+      text: '<steer>steer mid-turn</steer>',
+      displayText: 'steer mid-turn',
+    };
+    const submitted = await client.request('turn.message.submit', {
+      originHostEpoch: host.hostEpoch,
+      sessionId: fixture.sessionId,
+      messageId: steeringId,
+      content: steeringContent,
+      placement: 'current_turn',
+    });
+    assert.equal(submitted.disposition, 'steering');
+
+    // apache/maka#3304: the steering render must not depend on observing the
+    // transient in-flight queue state; the durable echo is forwarded verbatim.
+    const echoed = await probe.waitFor(
+      (frame) =>
+        frame.kind === 'subscription.session_event' && frame.event.type === 'steering_message',
+      'continuity did not forward the durable steering echo',
+    );
+    assert.equal(echoed.kind, 'subscription.session_event');
+    if (echoed.kind === 'subscription.session_event') {
+      assert.equal(echoed.event.type, 'steering_message');
+      if (echoed.event.type === 'steering_message') {
+        assert.equal(echoed.event.turnId, turnId);
+        assert.equal(echoed.event.messageId, steeringId);
+        assert.deepEqual(echoed.event.content, steeringContent);
+      }
+    }
+
+    assert.equal(
+      (await waitForTerminalTurn(client, fixture.sessionId, turnId)).status,
+      'completed',
+    );
+    await subscription.close();
+    await probe.done;
+    await client.close();
+    await fixture.stopHost(host);
+  });
+});
+
 test('steering becomes durable and ordered followups automatically start the next root', async () => {
   await withExecutionRoot(async (fixture) => {
     const host = await fixture.startHost();

packages/runtime-host/src/__tests__/protocol.test.ts

@@ -106,7 +106,7 @@ describe('Runtime Host bootstrap protocol', () => {
   });
 
   test('keeps the subscription queue Epoch correlated', () => {
-    assert.equal(SESSION_CONTINUITY_SCHEMA_VERSION, 4);
+    assert.equal(SESSION_CONTINUITY_SCHEMA_VERSION, 5);
     const opened = {
       requestId: 'open-1',
       operation: 'subscription.open',
@@ -305,6 +305,37 @@ describe('Runtime Host bootstrap protocol', () => {
     ]) {
       assert.throws(() => decodeHostFrame({ ...envelope, event }), isInvalidFrame);
     }
+
+    // The durable steering echo shares the session-event frame without a
+    // toolUseId; unknown keys stay rejected.
+    const steering = {
+      type: 'steering_message' as const,
+      id: 'steering-event-1',
+      turnId: 'turn-1',
+      ts: 7,
+      messageId: 'steering-message-1',
+      content: { text: 'steer the turn' },
+    };
+    const decodedSteering = decodeHostFrame({ ...envelope, event: steering });
+    assert.ok('kind' in decodedSteering);
+    if ('kind' in decodedSteering) {
+      assert.equal(decodedSteering.kind, 'subscription.session_event');
+      if (decodedSteering.kind === 'subscription.session_event') {
+        assert.deepEqual(decodedSteering.event, steering);
+      }
+    }
+    assert.throws(
+      () => decodeHostFrame({ ...envelope, event: { ...steering, toolUseId: 'tool-1' } }),
+      isInvalidFrame,
+    );
+    assert.throws(
+      () =>
+        decodeHostFrame({
+          ...envelope,
+          event: { ...steering, content: { text: 'x'.repeat(49 * 1024) } },
+        }),
+      isInvalidFrame,
+    );
     assert.throws(
       () =>
         decodeHostFrame({

packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts

@@ -69,6 +69,39 @@ test('open is an inactive publication barrier and live sequence starts at nextSe
   coordinator.close();
 });
 
+test('forwards the durable steering echo to subscribers as a session event', async () => {
+  const sink = new RecordingSink();
+  const coordinator = new SessionContinuityCoordinator(
+    HOST_EPOCH,
+    async () => canonical(),
+    new SessionAdmissionGate(),
+  );
+  const connection = coordinator.attachConnection('connection-1', sink);
+  const opened = await open(coordinator, 'connection-1');
+  connection.activate(opened.subscriptionId);
+  await delayImmediate();
+  sink.frames.length = 0;
+
+  const steering = {
+    type: 'steering_message' as const,
+    id: 'steering-event-1',
+    turnId: 'turn-1',
+    ts: 7,
+    messageId: 'steering-message-1',
+    content: { text: 'steer the turn' },
+  };
+  await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', steering);
+
+  assert.equal(sink.frames.length, 1);
+  const frame = sink.frames[0];
+  assert.equal(frame?.kind, 'subscription.session_event');
+  if (frame?.kind !== 'subscription.session_event') return;
+  assert.deepEqual(frame.event, steering);
+
+  connection.abort(opened.subscriptionId);
+  coordinator.close();
+});
+
 test('open snapshot includes pending Interactions from the canonical projection', async () => {
   const pending = pendingInteraction();
   const coordinator = new SessionContinuityCoordinator(

packages/runtime-host/src/__tests__/session-projector.test.ts

@@ -8,6 +8,7 @@ import {
 import {
   SESSION_CONTINUITY_SCHEMA_VERSION,
   type SessionContinuitySnapshot,
+  type SteeringMessageSnapshot,
   type SubscriptionFrame,
 } from '../protocol/index.js';
 
@@ -243,6 +244,169 @@ test('does not replay settled transcript steps when the active step reaches term
   );
 });
 
+test('projects the durable steering echo even when the in-flight queue state was never observed', () => {
+  // Regression for apache/maka#3304: the coalesced canonical refresh can jump
+  // the queue straight from queued to consumed, so the in-flight synthesis
+  // never fires. The forwarded steering_message event must render the message.
+  const projector = new RuntimeHostSessionProjector(
+    snapshot({ queue: queue(2, [steeringEntry('queued')]) }),
+    createRuntimeHostSessionProjectionSeed([], snapshot()),
+    () => 10,
+  );
+
+  const skipped = projector.accept({
+    kind: 'subscription.session_projection',
+    hostEpoch: 'host-1',
+    subscriptionId: 'subscription-1',
+    sequence: 1,
+    snapshot: snapshot({ queue: queue(4, []) }),
+  });
+  assert.deepEqual(
+    skipped.events.map((event) => event.type),
+    ['queue_update'],
+  );
+
+  const echoed = projector.accept(steeringFrame(2)).events;
+  assert.equal(echoed.length, 1);
+  assert.deepEqual(echoed[0], {
+    type: 'steering_message',
+    id: 'steering-event-1',
+    turnId: 'turn-1',
+    ts: 10,
+    messageId: 'steering-message-1',
+    content: { text: 'steer the turn' },
+  });
+});
+
+test('projects a steering message exactly once across both authoritative paths', () => {
+  // The queue in-flight synthesis and the durable session-event echo race;
+  // whichever projects the message first suppresses the other.
+  const inFlightFirst = new RuntimeHostSessionProjector(
+    snapshot({ queue: queue(2, [steeringEntry('queued')]) }),
+    createRuntimeHostSessionProjectionSeed([], snapshot()),
+    () => 10,
+  );
+  const synthesized = inFlightFirst.accept({
+    kind: 'subscription.session_projection',
+    hostEpoch: 'host-1',
+    subscriptionId: 'subscription-1',
+    sequence: 1,
+    snapshot: snapshot({ queue: queue(3, [steeringEntry('in_flight')]) }),
+  });
+  assert.deepEqual(
+    synthesized.events.map((event) => event.type),
+    ['steering_message', 'queue_update'],
+  );
+  assert.deepEqual(inFlightFirst.accept(steeringFrame(2)).events, []);
+
+  const echoFirst = new RuntimeHostSessionProjector(
+    snapshot({ queue: queue(2, [steeringEntry('queued')]) }),
+    createRuntimeHostSessionProjectionSeed([], snapshot()),
+    () => 10,
+  );
+  assert.equal(echoFirst.accept(steeringFrame(1)).events.length, 1);
+  const suppressed = echoFirst.accept({
+    kind: 'subscription.session_projection',
+    hostEpoch: 'host-1',
+    subscriptionId: 'subscription-1',
+    sequence: 2,
+    snapshot: snapshot({ queue: queue(3, [steeringEntry('in_flight')]) }),
+  });
+  assert.deepEqual(
+    suppressed.events.map((event) => event.type),
+    ['queue_update'],
+  );
+});
+
+test('seeds an unrendered in-flight steering message once on rejoin', () => {
+  const projector = new RuntimeHostSessionProjector(
+    snapshot({ queue: queue(3, [steeringEntry('in_flight')]) }),
+    createRuntimeHostSessionProjectionSeed([], snapshot()),
+    () => 10,
+  );
+  assert.deepEqual(
+    projector.seedActive(false).map((event) => event.type),
+    ['steering_message', 'queue_update'],
+  );
+  // A live echo of the same message arriving after the seed is the duplicate.
+  assert.deepEqual(projector.accept(steeringFrame(1)).events, []);
+});
+
+test('suppresses the live echo for a steering message already durable in the bootstrap', () => {
+  // subscription.open can bootstrap the durable steering message and install
+  // the subscriber while the Host's forwarded echo for it is still pending:
+  // the bootstrapped render must stay the only one (apache/maka#3316 review).
+  const inFlight = snapshot({ queue: queue(3, [steeringEntry('in_flight')]) });
+  const projector = new RuntimeHostSessionProjector(
+    inFlight,
+    createRuntimeHostSessionProjectionSeed(
+      [userSteering('steering-message-1', 'steering-event-1')],
+      inFlight,
+    ),
+    () => 10,
+  );
+
+  // Durable and in-flight: no synthesis seed…
+  assert.deepEqual(
+    projector.seedActive(false).map((event) => event.type),
+    ['queue_update'],
+  );
+  // …and the late echo of the same message is the duplicate.
+  assert.deepEqual(projector.accept(steeringFrame(1)).events, []);
+  // A different steering message still renders normally.
+  assert.equal(projector.accept(steeringFrame(2, 'steering-message-2')).events.length, 1);
+});
+
+function steeringEntry(state: 'queued' | 'in_flight'): SteeringMessageSnapshot {
+  return {
+    entryId: 'entry-1',
+    messageId: 'steering-message-1',
+    content: { text: 'steer the turn' },
+    placement: 'current_turn',
+    state,
+  };
+}
+
+function queue(
+  queueRevision: number,
+  steering: readonly SteeringMessageSnapshot[],
+): SessionContinuitySnapshot['queue'] {
+  return { hostEpoch: 'host-1', queueRevision, steering, followup: [] };
+}
+
+function steeringFrame(sequence: number, messageId = 'steering-message-1'): SubscriptionFrame {
+  return {
+    kind: 'subscription.session_event',
+    hostEpoch: 'host-1',
+    subscriptionId: 'subscription-1',
+    sequence,
+    sessionId: 'session-1',
+    runId: 'run-1',
+    event: {
+      type: 'steering_message',
+      id: 'steering-event-1',
+      turnId: 'turn-1',
+      ts: 10,
+      messageId,
+      content: { text: 'steer the turn' },
+    },
+  };
+}
+
+function userSteering(
+  id: string,
+  steeringEventId: string,
+): Extract<StoredMessage, { type: 'user' }> {
+  return {
+    type: 'user',
+    id,
+    turnId: 'turn-1',
+    ts: 1,
+    text: 'steer the turn',
+    steeringEventId,
+  };
+}
+
 function deltaFrame(
   sequence: number,
   startOffset: number,