fixes

sanchitrk · Aug 16, 2026 · Aug 11, 2026 · Aug 12, 2026 · Aug 13, 2026 · Aug 13, 2026
commit 697a4ae85d68d832512e7f35b7293ad076ab39bd
diff --git a/packages/harness/src/session/input/input.ts b/packages/harness/src/session/input/input.ts
@@ -93,13 +93,19 @@ export interface Interface {
   readonly hasPending: (sessionId: SessionSchema.ID, delivery: Delivery) => Effect.Effect<boolean>;
   /**
    * Promotes every pending steer admitted at or before `cutoff`, in admission
-   * order, and returns how many were promoted. The cutoff keeps a steer that
-   * lands mid-promotion out of a request that is already being assembled.
+   * order. The cutoff keeps a steer that lands mid-promotion out of a request
+   * that is already being assembled.
+   *
+   * Counts promotions *this* caller won. A row another drainer promoted first
+   * is not counted: callers treat a positive count as new work to answer, and
+   * counting a promotion someone else committed would have two workers answer
+   * one prompt.
    */
   readonly promoteSteers: (sessionId: SessionSchema.ID, cutoff: number) => Effect.Effect<number>;
   /**
    * Promotes the single oldest pending follow-up, if any. One per run, unlike
    * steers, which is what makes the follow-up lane strictly turn-at-a-time.
+   * False when there was none, or when another caller promoted it first.
    */
   readonly promoteFollowUp: (sessionId: SessionSchema.ID) => Effect.Effect<boolean>;
 }
@@ -295,9 +301,10 @@ export const make = Effect.gen(function* () {
     sessionId: SessionSchema.ID,
     rows: ReadonlyArray<SessionInputRow>,
   ) {
+    let promoted = 0;
     for (const row of rows) {
       const id = SessionMessageSchema.ID.make(row.id);
-      yield* events
+      const won = yield* events
         .publish(EventList.Prompted, {
           sessionId,
           timestamp: row.createdAt,
@@ -306,22 +313,25 @@ export const make = Effect.gen(function* () {
           delivery: row.delivery,
         })
         .pipe(
-          // Someone else promoted this row first. That is the intended outcome,
-          // so long as it really is promoted now.
+          Effect.as(true),
+          // Someone else promoted this row first. That is the intended outcome
+          // so long as it really is promoted now -- but it is their promotion,
+          // not ours, so it does not count as work for this caller.
           Effect.catchDefect((defect) =>
             isLifecycleConflict(defect)
               ? find(id).pipe(
                   Effect.flatMap((stored) =>
                     Option.isSome(stored) && stored.value.promotedSeq !== undefined
-                      ? Effect.void
+                      ? Effect.succeed(false)
                       : Effect.die(defect),
                   ),
                 )
               : Effect.die(defect),
           ),
         );
+      if (won) promoted += 1;
     }
-    return rows.length;
+    return promoted;
   });
 
   const hasPending = Effect.fn("SessionInput.hasPending")(function* (
@@ -345,8 +355,7 @@ export const make = Effect.gen(function* () {
   ) {
     const rows = yield* selectPendingFollowUp(sessionId).pipe(Effect.orDie);
     if (rows.length === 0) return false;
-    yield* publishPrompted(sessionId, rows);
-    return true;
+    return (yield* publishPrompted(sessionId, rows)) > 0;
   });
 
   return {

diff --git a/packages/harness/test/runner.loop.test.ts b/packages/harness/test/runner.loop.test.ts
@@ -219,6 +219,42 @@ const loopSpec = (name: string, backend: LoopBackend) => {
 			}),
 		);
 
+		/**
+		 * The delivery order the whole design turns on, and the one thing that must
+		 * not drift when the real loop replaces this mock: steers drain as a batch
+		 * while the lane keeps producing, and only once it is quiet does a single
+		 * follow-up go, one per pass. Interleaved admission does not reorder them.
+		 */
+		it(
+			"delivers all pending steers before follow-ups, one follow-up per pass",
+			Effect.gen(function* () {
+				const execution = yield* RunnerExecution.Service;
+				const sessions = yield* Session.Service;
+				const sessionId = yield* seedSession(backend);
+
+				// Admitted interleaved, so lane precedence — not arrival — decides.
+				yield* enqueue({ id: "msg_s1", sessionId, delivery: "steer" });
+				yield* enqueue({ id: "msg_f1", sessionId, delivery: "followUp" });
+				yield* enqueue({ id: "msg_s2", sessionId, delivery: "steer" });
+				yield* enqueue({ id: "msg_f2", sessionId, delivery: "followUp" });
+				yield* enqueue({ id: "msg_s3", sessionId, delivery: "steer" });
+
+				yield* execution.resume(sessionId);
+
+				// Steers in admission order, then follow-ups in admission order.
+				const path = yield* sessions.path(sessionId);
+				expect(path.map((h) => h.entry.id)).toEqual(["msg_s1", "msg_s2", "msg_s3", "msg_f1", "msg_f2"]);
+
+				// Conversation order is delivery order: the entry positions ascend in
+				// exactly the sequence the inbox recorded them as promoted.
+				const rows = yield* delivered(sessionId);
+				const byPromotion = [...rows]
+					.filter((row) => row.promotedSeq !== null)
+					.sort((a, b) => a.promotedSeq! - b.promotedSeq!);
+				expect(byPromotion.map((row) => row.id)).toEqual(path.map((h) => h.entry.id));
+			}),
+		);
+
 		it(
 			"wake with nothing eligible idles; an explicit resume still takes a turn",
 			Effect.gen(function* () {

diff --git a/packages/harness/test/session.flow.integration.test.ts b/packages/harness/test/session.flow.integration.test.ts
@@ -69,16 +69,27 @@ const drain = (
 ) =>
 	Effect.gen(function* () {
 		const passes: number[] = [];
+		let lane: "steer" | "followUp" | undefined = (yield* inputs.hasPending(sessionId, "steer"))
+			? "steer"
+			: "followUp";
 		for (;;) {
 			const cutoff = yield* events.latestSequence(sessionId);
-			let promoted = yield* inputs.promoteSteers(sessionId, cutoff);
-			if (promoted === 0 && (yield* inputs.hasPending(sessionId, "followUp"))) {
-				promoted += Number(yield* inputs.promoteFollowUp(sessionId));
-			}
+			// Follow-up first within a follow-up pass, then any steer that arrived
+			// since — the order a queue turn uses, so the one prompt this pass owes
+			// an answer to is ahead of anything that landed while it started.
+			let promoted = 0;
+			if (lane === "followUp") promoted += Number(yield* inputs.promoteFollowUp(sessionId));
+			promoted += yield* inputs.promoteSteers(sessionId, cutoff);
 			if (promoted === 0) break;
 			passes.push(promoted);
 			// Stands in for the turn a real loop would run here.
 			yield* Effect.yieldNow;
+			if (yield* inputs.hasPending(sessionId, "steer")) {
+				lane = "steer";
+				continue;
+			}
+			lane = (yield* inputs.hasPending(sessionId, "followUp")) ? "followUp" : undefined;
+			if (lane === undefined) break;
 		}
 		return passes;
 	});

diff --git a/packages/harness/test/session.prompt.test.ts b/packages/harness/test/session.prompt.test.ts
@@ -141,14 +141,17 @@ describe("Session.prompt", () => {
 			const inputs = yield* SessionInput.make;
 			const admitted = yield* sessions.prompt({ sessionId, id: messageId, prompt: prompt("Promote once") });
 
-			yield* Effect.all(
+			const counts = yield* Effect.all(
 				[
 					inputs.promoteSteers(sessionId, Number.MAX_SAFE_INTEGER),
 					inputs.promoteSteers(sessionId, Number.MAX_SAFE_INTEGER),
 				],
 				{ concurrency: "unbounded" },
 			);
 
+			// Exactly one caller won it. A count is "work this caller promoted", so
+			// the loser reporting 1 would have both of them answer the same prompt.
+			expect(counts[0] + counts[1]).toBe(1);
 			expect(yield* eventCount("session.next.prompt.promoted.1")).toBe(1);
 			const promoted = some(yield* inputs.find(admitted.id));
 			expect(promoted.promotedSeq).toBeDefined();
@@ -161,6 +164,22 @@ describe("Session.prompt", () => {
 			expect(JSON.parse(path[0]!.parts[0]!.data)).toMatchObject({ type: "text", text: "Promote once" });
 		}));
 
+	it("reports the follow-up to whichever caller won it", () =>
+		Effect.gen(function* () {
+			const { sessions, sessionId } = yield* setup;
+			const inputs = yield* SessionInput.make;
+			yield* sessions.prompt({ sessionId, prompt: prompt("Later"), delivery: "followUp" });
+
+			const won = yield* Effect.all(
+				[inputs.promoteFollowUp(sessionId), inputs.promoteFollowUp(sessionId)],
+				{ concurrency: "unbounded" },
+			);
+
+			expect(won.filter(Boolean).length).toBe(1);
+			expect(yield* eventCount("session.next.prompt.promoted.1")).toBe(1);
+			expect(yield* inputs.hasPending(sessionId, "followUp")).toBe(false);
+		}));
+
 	it("promotes steers only through the captured cutoff", () =>
 		Effect.gen(function* () {
 			const { sessions, sessionId } = yield* setup;