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

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

The TUI never rendered a consumed steering message: the host event pump
dropped steering_message (not a transient tool/text event), leaving the
transient in-flight queue observation as the only live render path, which
the coalesced canonical refresh can skip entirely (queued -> consumed).

Forward the durable steering_message on the session-event frame so every
subscriber renders the interjection in place, keep the queue in-flight
synthesis for the attach/rejoin window, and dedup both paths per message.
Bumps the session continuity wire schema to 5.

Fixes #3304

Generated-by: Maka

* refactor(runtime-host): name the forwarding rule, not the durability class

isRuntimeSessionForwardedEvent / RuntimeSessionForwardedEvent describe what
the predicate decides — forward live to subscribers — now that the durable
steering_message belongs to it. Rename the client projectSessionEvent to
match its Host-side counterpart. Addresses review on #3316.

Generated-by: Maka

* fix(runtime-host): suppress the steering echo when the bootstrap already rendered it

subscription.open can bootstrap the durable steering message and install the
subscriber before the Host's forwarded echo arrives; seed the render-dedup
set from steering messages already durable in the transcript so the bootstrapped
render stays the only one. Also rebased onto main and updated the new e2e test
to the post-#3277 single-argument connectClient.

Addresses review on #3316

Generated-by: Maka

* test(runtime-host): restore steering queue fixture helper after rebase

Generated-by: maka

* fix(runtime-host): advance compatibility epoch for the continuity schema bump

Generated-by: maka

* fix(runtime-host): align steering types after rebase

Generated-by: maka

* test(runtime-host): use current turn start request

Generated-by: maka

Author: me2seeks

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

@@ -114,6 +114,66 @@ 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();
+    requireStartedTurn(
+      await client.request('turn.start', {
+        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

@@ -250,7 +250,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',
@@ -467,6 +467,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

@@ -88,6 +88,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

@@ -21,6 +21,7 @@ import assert from 'node:assert/strict';
 import test from 'node:test';
 import type { SessionEvent } from '@maka/core/events';
 import type { StoredMessage } from '@maka/core/session';
+import type { SteeringMessageSnapshot } from '../protocol/message.js';
 import {
   createRuntimeHostSessionProjectionSeed,
   RuntimeHostSessionProjector,
@@ -499,6 +500,162 @@ test('preserves the bounded shell-run correlation on a tool start', () => {
   );
 });
 
+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 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,
@@ -523,6 +680,13 @@ function deltaFrame(
   };
 }
 
+function queue(
+  queueRevision: number,
+  steering: readonly SteeringMessageSnapshot[],
+): SessionContinuitySnapshot['queue'] {
+  return { hostEpoch: 'host-1', queueRevision, steering, followup: [] };
+}
+
 function snapshot(overrides: Partial<SessionContinuitySnapshot> = {}): SessionContinuitySnapshot {
   return {
     schemaVersion: SESSION_CONTINUITY_SCHEMA_VERSION,