feat(runtime-host): separate queued successor turns

docs/desktop-message-queue.md

@@ -26,7 +26,7 @@ Desktop used a rendered `streaming` prop to decide whether a composer submit sta
 Runtime Host already owns the durable message semantics:
 
 - `current_turn` queues steering for the next provider boundary.
-- `next_turn` queues a successor turn.
+- `next_turn` queues one successor turn per accepted message.
 - queue projections are authoritative.
 - queue projections carry the canonical queued message content; mutation results return only queue state.
 

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

@@ -287,38 +287,41 @@ test('steering becomes durable and ordered followups automatically start the nex
     }
 
     const chain = await fixture.readAdmissionChain();
-    assert.equal(chain.length, 2);
+    assert.equal(chain.length, 3);
     assert.equal(chain[1]?.previousRootTurnId, firstTurnId);
-    assert.equal(chain[1]?.userMessageId, null);
-    assert.deepEqual(
-      chain[1]?.sourceMessages.map(({ messageId, content, placement, disposition }) => ({
-        messageId,
-        content,
-        placement,
-        disposition,
-      })),
-      orderedFollowupSources.map((source) => ({
-        ...source,
-        placement: 'next_turn',
-        disposition: 'followup',
-      })),
+    assert.equal(chain[2]?.previousRootTurnId, chain[1]?.turnId);
+    for (const [index, source] of orderedFollowupSources.entries()) {
+      const admission = chain[index + 1];
+      assert.equal(admission?.userMessageId, source.messageId);
+      assert.deepEqual(
+        admission?.sourceMessages.map(({ messageId, content, placement, disposition }) => ({
+          messageId,
+          content,
+          placement,
+          disposition,
+        })),
+        [{ ...source, placement: 'next_turn', disposition: 'followup' }],
+      );
+    }
+    const followupTurnIds = chain.slice(1).map((admission) => admission.turnId);
+    const followupLedgers = await Promise.all(
+      followupTurnIds.map((turnId) => fixture.readTurn(turnId)),
     );
-    assert.deepEqual(chain[1]?.normalizedInput, {
-      text: `${orderedFollowupSources[0].content.text}\n\n${orderedFollowupSources[1].content.text}`,
-      displayText: `${orderedFollowupSources[0].content.text}\n\n${orderedFollowupSources[1].content.displayText}`,
-      attachments: orderedFollowupSources[1].content.attachments,
-      quotes: orderedFollowupSources.flatMap((source) => source.content.quotes ?? []),
-    });
-    const followupTurnId = chain[1]?.turnId;
-    assert.ok(followupTurnId);
-    const followupLedger = await fixture.readTurn(followupTurnId);
     const expectedQuotes = orderedFollowupSources.flatMap((source) => source.content.quotes ?? []);
-    assert.equal(followupLedger.userMessages.length, followupSources.length);
     assert.deepEqual(
-      followupLedger.userMessages.flatMap((message) => message.quotes ?? []),
+      followupLedgers.map((ledger) => ledger.userMessages.length),
+      [1, 1],
+    );
+    assert.deepEqual(
+      followupLedgers.flatMap((ledger) =>
+        ledger.userMessages.flatMap((message) => message.quotes ?? []),
+      ),
+      expectedQuotes,
+    );
+    assert.deepEqual(
+      followupLedgers.flatMap((ledger) => userRuntimeContent(ledger.runtimeEvents)?.quotes ?? []),
       expectedQuotes,
     );
-    assert.deepEqual(userRuntimeContent(followupLedger.runtimeEvents)?.quotes, expectedQuotes);
     const sessionUserMessages = await fixture.readSessionUserMessages();
     for (const source of orderedFollowupSources) {
       assert.equal(
@@ -327,13 +330,15 @@ test('steering becomes durable and ordered followups automatically start the nex
       );
     }
     assert.equal(
-      sessionUserMessages.filter((message) => message.turnId === followupTurnId).length,
+      sessionUserMessages.filter((message) => followupTurnIds.includes(message.turnId)).length,
       orderedFollowupSources.length,
     );
     assert.deepEqual(
-      sessionUserMessages
-        .filter((message) => message.turnId === followupTurnId)
-        .map((message) => message.id),
+      followupTurnIds.flatMap((turnId) =>
+        sessionUserMessages
+          .filter((message) => message.turnId === turnId)
+          .map((message) => message.id),
+      ),
       orderedFollowupSources.map((source) => source.messageId),
     );
   });

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

@@ -217,17 +217,18 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as
     await tui.close();
     await fixture.stopHost(host);
     const chain = await fixture.readAdmissionChain();
+    assert.equal(chain.length, 3);
     assert.deepEqual(
-      chain.map((admission) => admission.turnId),
+      chain.slice(0, 2).map((admission) => admission.turnId),
       [firstTurnId, successor.snapshot.rootTurn.turnId],
     );
+    assert.equal(chain[2]?.previousRootTurnId, successor.snapshot.rootTurn.turnId);
     assert.deepEqual(
-      chain[1]?.sourceMessages.map((source) => source.messageId),
-      [desktopFollowupId, tuiFollowupId],
+      chain.slice(1).map((admission) => admission.sourceMessages.map((source) => source.messageId)),
+      [[desktopFollowupId], [tuiFollowupId]],
     );
-    assert.deepEqual(chain[1]?.normalizedInput, {
-      text: `${desktopFollowupContent.text}\n\n${tuiFollowupContent.text}`,
-    });
+    assert.deepEqual(chain[1]?.normalizedInput, desktopFollowupContent);
+    assert.deepEqual(chain[2]?.normalizedInput, tuiFollowupContent);
   });
 });