fix(runner): make turn_index a per-conversation-turn counter

Author: mmabrouk

services/runner/src/engines/sandbox_agent/environment.ts

@@ -112,7 +112,6 @@ import {
 import { hydrateHarnessSessionFromDurable } from "./session-continuity-durable.ts";
 import {
   eligibleAgentSessionId,
-  nextTurnIndex,
   sessionContinuityStore,
 } from "./session-continuity.ts";
 import { projectScopeFor } from "./session-identity.ts";
@@ -943,11 +942,6 @@ export async function acquireEnvironment(
     const localSessionId = continuitySessionKey
       ? `${continuitySessionKey}:${plan.harness}`
       : undefined;
-    // The index THIS turn will occupy once it completes: recorded post-turn against the SAME
-    // index read here, so a turn that authors turn N leaves the store agreeing with itself.
-    environment.continuityTurnIndex = continuitySessionKey
-      ? nextTurnIndex(continuitySessionKey, continuityStore)
-      : undefined;
     // The live sandbox id rides forward as a field on the turn-append row written at turn end
     // (see `appendSessionTurn` call in `runTurn`), not a separate pre-turn pointer PUT: the
     // turns table is append-only, so there is nothing to overwrite mid-conversation.

services/runner/src/engines/sandbox_agent/run-turn.ts

@@ -63,7 +63,10 @@ import {
 import {
   appendSessionTurn,
 } from "./session-continuity-durable.ts";
-import { sessionContinuityStore } from "./session-continuity.ts";
+import {
+  nextTurnIndex,
+  sessionContinuityStore,
+} from "./session-continuity.ts";
 import { priorMessages } from "./transcript.ts";
 import { resolveRunUsage } from "./usage.ts";
 
@@ -83,6 +86,12 @@ export async function runTurn(
 ): Promise<AgentRunResult> {
   const { plan, logger, deps } = env;
   const sessionId = env.sessionId;
+  const continuityStore = deps.sessionContinuityStore ?? sessionContinuityStore;
+  // `turn_index` is a true conversation-turn counter, not an acquire counter: it advances once per completed turn across every environment serving the session.
+  // The shared store advances only on `record()` (paused turns record nothing), so park-and-resume consumes one index; compute it at turn start because a warm environment serves many turns.
+  env.continuityTurnIndex = sessionId
+    ? nextTurnIndex(sessionId, continuityStore)
+    : undefined;
   // Reset the per-turn tool-call id record (the park folds the completed turn's ids into the
   // expected next-history fingerprint).
   env.lastTurnToolCallIds = [];

services/runner/tests/unit/session-keepalive-engine.test.ts

@@ -19,10 +19,12 @@ import {
   runTurn,
   type SandboxAgentDeps,
 } from "../../src/engines/sandbox_agent.ts";
+import { SessionContinuityStore } from "../../src/engines/sandbox_agent/session-continuity.ts";
 
 interface FakeOptions {
   probeError?: Error;
   createOtelError?: Error;
+  stopReasons?: string[];
 }
 
 /** One fake otel run per createOtel call, recording which updates it handled. */
@@ -40,9 +42,11 @@ function fakeHarness(options: FakeOptions = {}) {
     sandboxDisposed: 0,
     sessionDestroyed: 0,
     permissionReplies: [] as Array<{ id: string; reply: string }>,
+    appendedTurnIndexes: [] as number[],
     logs: [] as string[],
     runs: [] as FakeRun[],
   };
+  const continuityStore = new SessionContinuityStore();
 
   // Captured session-lifetime handlers, so tests can fire events/permissions at any moment
   // (during a turn, between turns).
@@ -53,6 +57,7 @@ function fakeHarness(options: FakeOptions = {}) {
 
   const session = {
     id: "session-1",
+    agentSessionId: "agent-session-1",
     onEvent(handler: (event: any) => void) {
       captured.onEvent = handler;
     },
@@ -65,7 +70,7 @@ function fakeHarness(options: FakeOptions = {}) {
     async prompt(_blocks: any) {
       calls.promptCount += 1;
       return {
-        stopReason: "complete",
+        stopReason: options.stopReasons?.[calls.promptCount - 1] ?? "complete",
         usage: { inputTokens: 1, outputTokens: 1 },
       };
     },
@@ -163,6 +168,11 @@ function fakeHarness(options: FakeOptions = {}) {
         return { kind: "deny" } as const;
       },
     }),
+    sessionContinuityStore: continuityStore,
+    hydrateHarnessSessionFromDurable: async () => {},
+    appendSessionTurn: async (_sessionId, _harness, turnIndex) => {
+      calls.appendedTurnIndexes.push(turnIndex);
+    },
   };
 
   return { calls, deps, captured };
@@ -173,6 +183,15 @@ const request: AgentRunRequest = {
   messages: [{ role: "user", content: "hello" }],
 };
 
+const continuityRequest: AgentRunRequest = {
+  ...request,
+  sessionId: "sess-1",
+  streamId: "stream-1",
+  telemetry: {
+    exporters: { otlp: { headers: { authorization: "ApiKey abc" } } },
+  } as any,
+};
+
 function updateEvent(update: Record<string, unknown>) {
   return { payload: { update } };
 }
@@ -282,6 +301,93 @@ describe("acquireEnvironment / runTurn split", () => {
   });
 });
 
+describe("conversation turn indexes", () => {
+  it("advances on every completed turn served by the same warm environment", async () => {
+    const { calls, deps } = fakeHarness();
+    const acquired = await acquireEnvironment(continuityRequest, deps);
+    assert.equal(acquired.ok, true);
+    if (!acquired.ok) return;
+
+    for (let turn = 0; turn < 4; turn += 1) {
+      const result = await runTurn(
+        acquired.env,
+        {
+          ...continuityRequest,
+          messages: [{ role: "user", content: "turn " + turn }],
+        },
+        undefined,
+        undefined,
+        { continuation: turn > 0 },
+      );
+      assert.equal(result.ok, true);
+    }
+
+    assert.deepEqual(calls.appendedTurnIndexes, [0, 1, 2, 3]);
+    await acquired.env.destroy();
+  });
+
+  it("shares strictly increasing indexes across two environments for one session", async () => {
+    const { calls, deps } = fakeHarness({
+      stopReasons: ["paused", "complete", "complete", "complete"],
+    });
+    const first = await acquireEnvironment(continuityRequest, deps);
+    const second = await acquireEnvironment(continuityRequest, deps);
+    assert.equal(first.ok, true);
+    assert.equal(second.ok, true);
+    if (!first.ok || !second.ok) return;
+
+    const parked = await runTurn(
+      first.env,
+      continuityRequest,
+      undefined,
+      undefined,
+      {},
+    );
+    assert.equal(parked.stopReason, "paused");
+    await runTurn(second.env, continuityRequest, undefined, undefined, {});
+    await runTurn(first.env, continuityRequest, undefined, undefined, {
+      continuation: true,
+    });
+    await runTurn(second.env, continuityRequest, undefined, undefined, {
+      continuation: true,
+    });
+
+    assert.deepEqual(calls.appendedTurnIndexes, [0, 1, 2]);
+    await first.env.destroy();
+    await second.env.destroy();
+  });
+
+  it("does not consume an index until a paused conversation turn completes", async () => {
+    const { calls, deps } = fakeHarness({
+      stopReasons: ["paused", "complete"],
+    });
+    const acquired = await acquireEnvironment(continuityRequest, deps);
+    assert.equal(acquired.ok, true);
+    if (!acquired.ok) return;
+
+    const parked = await runTurn(
+      acquired.env,
+      continuityRequest,
+      undefined,
+      undefined,
+      {},
+    );
+    assert.equal(parked.stopReason, "paused");
+    assert.deepEqual(calls.appendedTurnIndexes, []);
+
+    const resumed = await runTurn(
+      acquired.env,
+      continuityRequest,
+      undefined,
+      undefined,
+      { continuation: true },
+    );
+    assert.equal(resumed.ok, true);
+    assert.deepEqual(calls.appendedTurnIndexes, [0]);
+    await acquired.env.destroy();
+  });
+});
+
 describe("session-lifetime listener demux (currentTurn swap)", () => {
   it("each turn's sink sees only its own events; between-turns events are dropped by decision", async () => {
     const { calls, deps, captured } = fakeHarness();