From 8c482357523e14237d926da30ebe8b993b03b8f1 Mon Sep 17 00:00:00 2001
From: Mahmoud Mabrouk <mahmoud@agenta.ai>
Date: Sat, 18 Jul 2026 18:26:15 +0200
Subject: [PATCH 1/3] feat(sessions): rebase session turns/streams runner onto
 v0.105.5

Rebases the runner third of JP's feat/sessions-extensions (PR #5363) onto
current main, ported into the post-decomposition module layout (PR #5369).

JP wrote these against the old monolithic sandbox_agent.ts; they are re-homed
by region into the decomposed modules:
- The SandboxAgentDeps interface change (syncHarnessSessionDurable renamed to
  appendSessionTurn; the write/clear sandbox-pointer deps dropped) goes into
  runtime-contracts.ts.
- The two acquireEnvironment branches (the reconnect-failure path and the
  post-hydrate pointer write, both now append-only with no pre-turn pointer PUT)
  go into environment.ts.
- The turn-completion sync (syncHarnessSessionDurable replaced by appendSessionTurn
  with the streamId gate, workflow references, sandbox id, and trace id) goes into
  run-turn.ts.
The sandbox-reconnect.ts and session-continuity-durable.ts rewrites apply as-is.
protocol.ts, sessions/alive.ts, sessions/persist.ts, version.ts, and the tests
are JP's versions verbatim.

Ported from feat/sessions-extensions@1532ec7fe5 (JP).

Reconciliations with the release main:
- server.ts: kept main's session-identity / session-pool import split from the
  decomposition and applied JP's watchdog changes (startAliveWatchdog is now
  awaited and takes an onInterrupted callback that aborts the controller,
  request.streamId is threaded from the heartbeat, and buildPersistingEmitter
  gets turnId and span_id).
- tracing/otel.ts, package.json, pnpm-lock.yaml: kept main's. The OTel 2.x bump
  JP carried is already in main, and main additionally has the #5364
  trace-id-on-done feature that JP's branch predates.

Verified: pnpm run typecheck clean, pnpm test 1202/1202 pass, build:extension ok.

Claude-Session: https://claude.ai/code/session_01DnWRxU3dCJ11hgDidm26vW
---
 .../src/engines/sandbox_agent/environment.ts  |  59 +--
 .../src/engines/sandbox_agent/run-turn.ts     |  21 +-
 .../sandbox_agent/runtime-contracts.ts        |  13 +-
 .../sandbox_agent/sandbox-reconnect.ts        | 120 +----
 .../session-continuity-durable.ts             | 253 +++++-----
 services/runner/src/protocol.ts               |   7 +
 services/runner/src/server.ts                 |  15 +-
 services/runner/src/sessions/alive.ts         |  71 ++-
 services/runner/src/sessions/persist.ts       |  33 +-
 services/runner/src/version.ts                |   4 +-
 .../tests/acceptance/server-contract.test.ts  |  18 +-
 .../tests/integration/server-smoke.test.ts    |  12 +-
 .../tests/unit/sandbox-lifecycle.test.ts      | 111 ++---
 .../tests/unit/sandbox-reconnect.test.ts      | 132 ++---
 services/runner/tests/unit/server.test.ts     |  22 +-
 .../unit/session-alive-interrupt.test.ts      | 121 +++++
 .../runner/tests/unit/session-alive.test.ts   |  76 ++-
 .../unit/session-continuity-durable.test.ts   | 466 +++++++++++-------
 .../runner/tests/unit/session-persist.test.ts |  60 +++
 19 files changed, 947 insertions(+), 667 deletions(-)
 create mode 100644 services/runner/tests/unit/session-alive-interrupt.test.ts

diff --git a/services/runner/src/engines/sandbox_agent/environment.ts b/services/runner/src/engines/sandbox_agent/environment.ts
index 7d38fcb5c0..cad062774b 100644
--- a/services/runner/src/engines/sandbox_agent/environment.ts
+++ b/services/runner/src/engines/sandbox_agent/environment.ts
@@ -99,11 +99,7 @@ import {
   routeSessionEventToActiveTurn,
 } from "./session-events.ts";
 import { buildSandboxProvider } from "./provider.ts";
-import {
-  clearSandboxPointer,
-  readStoredSandboxPointer,
-  writeSandboxPointer,
-} from "./sandbox-reconnect.ts";
+import { readStoredSandboxPointer } from "./sandbox-reconnect.ts";
 import type {
   AcquireEnvironmentResult,
   SandboxAgentDeps,
@@ -642,31 +638,13 @@ export async function acquireEnvironment(
         logger(
           `reconnect failed sandbox=${storedSandboxPointer.sandboxId}, creating fresh: ${conciseError(err, plan.harness)}`,
         );
-        if (
-          err instanceof DaytonaReconnectTerminalError &&
-          sessionForMount &&
-          runCred
-        ) {
-          // The post-hydrate write later in acquire is authoritative. This clear only prevents
-          // repeated doomed reconnects if acquire fails before reaching that write. Hydrate
-          // first: after a runner restart the in-memory store is behind the durable
-          // latest_turn_index, and an unhydrated guard token would be rejected as stale.
-          await (
-            deps.hydrateHarnessSessionFromDurable ??
-            hydrateHarnessSessionFromDurable
-          )(
-            sessionForMount,
-            plan.harness,
-            deps.sessionContinuityStore ?? sessionContinuityStore,
-            { authorization: runCred, log: logger },
-          );
-          await (deps.clearSandboxPointer ?? clearSandboxPointer)(
-            sessionForMount,
-            nextTurnIndex(
-              sessionForMount,
-              deps.sessionContinuityStore ?? sessionContinuityStore,
-            ),
-            { authorization: runCred, log: logger },
+        // No explicit pointer clear needed: turns are append-only, so the fresh sandbox this
+        // turn creates below gets its own turn row at completion, and that row's higher
+        // turn_index naturally supersedes the dead one on the next `latest_turn` read — the
+        // staleness guard the old states model needed dissolves with the ordering.
+        if (err instanceof DaytonaReconnectTerminalError) {
+          logger(
+            `terminal Daytona state '${err.state}' for sandbox=${storedSandboxPointer.sandboxId}, not retrying reconnect`,
           );
         }
       } finally {
@@ -984,24 +962,9 @@ export async function acquireEnvironment(
     environment.continuityTurnIndex = continuitySessionKey
       ? nextTurnIndex(continuitySessionKey, continuityStore)
       : undefined;
-    // Daytona only: a local run must not overwrite a conversation's remote pointer (switching
-    // sandboxes mid-conversation would strand the parked Daytona instance).
-    if (plan.isDaytona && sessionForMount && runCred) {
-      const liveSandboxId = environment.sandbox?.sandboxId ?? plan.sandboxId;
-      const pointerWriteOutcome = await (
-        deps.writeSandboxPointer ?? writeSandboxPointer
-      )(
-        sessionForMount,
-        {
-          sandboxId: liveSandboxId,
-          turnIndex: environment.continuityTurnIndex ?? 0,
-        },
-        { authorization: runCred, log: logger },
-      );
-      logger(
-        `sandbox pointer write ${pointerWriteOutcome} session=${sessionForMount} sandbox=${liveSandboxId}`,
-      );
-    }
+    // 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.
     let loadedFromContinuity = false;
     if (priorAgentSessionId && localSessionId) {
       await persist.updateSession({
diff --git a/services/runner/src/engines/sandbox_agent/run-turn.ts b/services/runner/src/engines/sandbox_agent/run-turn.ts
index 7139732658..b7e50689a4 100644
--- a/services/runner/src/engines/sandbox_agent/run-turn.ts
+++ b/services/runner/src/engines/sandbox_agent/run-turn.ts
@@ -64,7 +64,7 @@ import {
   shouldSuppressPausedToolCallUpdate,
 } from "./runtime-policy.ts";
 import {
-  syncHarnessSessionDurable,
+  appendSessionTurn,
 } from "./session-continuity-durable.ts";
 import { sessionContinuityStore } from "./session-continuity.ts";
 import { priorMessages } from "./transcript.ts";
@@ -567,16 +567,25 @@ export async function runTurn(
         env.session.agentSessionId,
         env.continuityTurnIndex,
       );
-      // Mirror the record durably so it survives a runner restart; fire-and-forget.
+      // Append this turn to the durable turns log; fire-and-forget (a plain INSERT, no race).
       const syncCred = runCredential(request);
-      if (syncCred) {
-        void (deps.syncHarnessSessionDurable ?? syncHarnessSessionDurable)(
+      if (syncCred && request.streamId) {
+        const workflowRefs = buildWorkflowReferences(
+          request.runContext?.workflow,
+        );
+        void (deps.appendSessionTurn ?? appendSessionTurn)(
           sessionId,
           plan.harness,
-          env.session.agentSessionId,
           env.continuityTurnIndex,
+          {
+            streamId: request.streamId,
+            agentSessionId: env.session.agentSessionId,
+            sandboxId: env.sandbox?.sandboxId,
+            references: workflowRefs ? Object.values(workflowRefs) : undefined,
+            traceId: run.traceId(),
+          },
           { authorization: syncCred, log: logger },
-        );
+        ).catch(() => {});
       }
     } else if (stopReason === "paused") {
       // A pause stopped mid-turn, after the harness may have written a partial turn natively.
diff --git a/services/runner/src/engines/sandbox_agent/runtime-contracts.ts b/services/runner/src/engines/sandbox_agent/runtime-contracts.ts
index 01e82d9b1e..04479499a1 100644
--- a/services/runner/src/engines/sandbox_agent/runtime-contracts.ts
+++ b/services/runner/src/engines/sandbox_agent/runtime-contracts.ts
@@ -18,8 +18,8 @@ import { PendingApprovalPauseController } from "./pause.ts";
 import { buildSandboxProvider } from "./provider.ts";
 import { createRunLimits, resolveRunLimits } from "./run-limits.ts";
 import { type BuildRunPlanDeps, type RunPlan } from "./run-plan.ts";
-import { clearSandboxPointer, readStoredSandboxPointer, writeSandboxPointer } from "./sandbox-reconnect.ts";
-import { hydrateHarnessSessionFromDurable, syncHarnessSessionDurable } from "./session-continuity-durable.ts";
+import { readStoredSandboxPointer } from "./sandbox-reconnect.ts";
+import { appendSessionTurn, hydrateHarnessSessionFromDurable } from "./session-continuity-durable.ts";
 import { type SessionContinuityStore } from "./session-continuity.ts";
 import { type TeardownReason } from "./teardown.ts";
 import { uploadToolMcpAssets } from "./tool-mcp-assets.ts";
@@ -57,13 +57,12 @@ export interface SandboxAgentDeps extends BuildRunPlanDeps {
   createRunLimits?: typeof createRunLimits;
   /** Session-continuity store override (tests inject their own; default is the process singleton). */
   sessionContinuityStore?: SessionContinuityStore;
-  /** Durable read-back/write-forward of the continuity store (tests inject fakes). */
+  /** Durable read-back/append-forward of the continuity store (tests inject fakes). */
   hydrateHarnessSessionFromDurable?: typeof hydrateHarnessSessionFromDurable;
-  syncHarnessSessionDurable?: typeof syncHarnessSessionDurable;
-  /** Durable read/write of the sandbox pointer, for the remote reconnect ladder. */
+  appendSessionTurn?: typeof appendSessionTurn;
+  /** Durable read of the sandbox pointer (the latest turn's sandbox_id), for the remote
+   * reconnect ladder. The write side is folded into `appendSessionTurn`. */
   readStoredSandboxPointer?: typeof readStoredSandboxPointer;
-  clearSandboxPointer?: typeof clearSandboxPointer;
-  writeSandboxPointer?: typeof writeSandboxPointer;
   /**
    * Resolve `{replicaId, ownerReplicaId}` for a session-owned local-sandbox run, so
    * `acquireEnvironment` can fail loudly instead of silently cold-starting on a non-owner
diff --git a/services/runner/src/engines/sandbox_agent/sandbox-reconnect.ts b/services/runner/src/engines/sandbox_agent/sandbox-reconnect.ts
index 0154458146..6c2fbdc8db 100644
--- a/services/runner/src/engines/sandbox_agent/sandbox-reconnect.ts
+++ b/services/runner/src/engines/sandbox_agent/sandbox-reconnect.ts
@@ -1,10 +1,16 @@
 /**
- * Daytona sandbox reconnect: read the stored sandbox id back so a resumed session restarts the
- * parked (stopped/archived) sandbox instead of provisioning a fresh one, and write the live id
- * forward after start. Best-effort throughout: a missing/unreadable id, or a failed reconnect,
- * degrades to a fresh create (the dead rung), never a hard error.
+ * Daytona sandbox reconnect: read the latest turn's stored sandbox id so a resumed session
+ * restarts the parked (stopped/archived) sandbox instead of provisioning a fresh one.
+ * Best-effort throughout: a missing/unreadable id, or a failed reconnect, degrades to a fresh
+ * create (the dead rung), never a hard error.
+ *
+ * The live id is written forward as a field on the turn-append row (see
+ * `session-continuity-durable.ts` `appendSessionTurn`), not through a separate pointer PUT: the
+ * turns table is append-only, so "the latest turn's sandbox_id" IS the current pointer — a late
+ * lower-index write can never win `ORDER BY turn_index DESC`, dissolving the old atomic
+ * staleness guard.
  */
-import { apiBase } from "../../apiBase.ts";
+import { fetchLatestSessionTurn } from "./session-continuity-durable.ts";
 
 export interface SandboxPointerDeps {
   apiBase?: string;
@@ -13,10 +19,6 @@ export interface SandboxPointerDeps {
   log?: (msg: string) => void;
 }
 
-function defaultLog(msg: string): void {
-  process.stderr.write(`[sandbox-reconnect] ${msg}\n`);
-}
-
 /**
  * The stored sandbox instance id for this session, or undefined when none is recorded (first
  * turn, storage disabled, or unreachable). The id is a provider-scoped handle, so reconnect is
@@ -26,104 +28,12 @@ export interface StoredSandboxPointer {
   sandboxId: string;
 }
 
-export interface SandboxPointerWrite {
-  sandboxId: string;
-  turnIndex: number;
-}
-
-export type SandboxPointerWriteOutcome = "applied" | "rejected" | "failed";
-
 export async function readStoredSandboxPointer(
   sessionId: string,
   deps: SandboxPointerDeps,
 ): Promise<StoredSandboxPointer | undefined> {
-  const log = deps.log ?? defaultLog;
-  const doFetch = deps.fetchImpl ?? fetch;
-  const base = deps.apiBase ?? apiBase();
-  try {
-    const res = await doFetch(
-      `${base}/sessions/states/?session_id=${encodeURIComponent(sessionId)}`,
-      { method: "GET", headers: { authorization: deps.authorization } },
-    );
-    if (!res.ok) return undefined;
-    const body = (await res.json()) as {
-      session_state?: {
-        sandbox_id?: string | null;
-      } | null;
-    };
-    const id = body.session_state?.sandbox_id;
-    if (typeof id !== "string" || id.length === 0) return undefined;
-    return { sandboxId: id };
-  } catch (err) {
-    log(
-      `read failed session=${sessionId}: ${String(err instanceof Error ? err.message : err).slice(0, 120)}`,
-    );
-    return undefined;
-  }
-}
-
-/**
- * Shared guarded pointer PUT. `write` and `clear` differ only in the sandbox_id they send and
- * the returned-row value that counts as applied; the transport, guard token, and error handling
- * must stay identical, so they live here once.
- */
-async function putSandboxPointer(
-  sessionId: string,
-  sandboxId: string | null,
-  turnIndex: number,
-  deps: SandboxPointerDeps,
-  logPrefix: string,
-): Promise<SandboxPointerWriteOutcome> {
-  const log = deps.log ?? defaultLog;
-  const doFetch = deps.fetchImpl ?? fetch;
-  const base = deps.apiBase ?? apiBase();
-  try {
-    const res = await doFetch(
-      `${base}/sessions/states/?session_id=${encodeURIComponent(sessionId)}`,
-      {
-        method: "PUT",
-        headers: { "content-type": "application/json", authorization: deps.authorization },
-        body: JSON.stringify({
-          sandbox_id: sandboxId,
-          sandbox_turn_index: turnIndex,
-        }),
-      },
-    );
-    if (!res.ok) {
-      log(`${logPrefix} HTTP ${res.status} session=${sessionId}`);
-      return "failed";
-    }
-    const body = (await res.json()) as {
-      session_state?: { sandbox_id?: string | null } | null;
-    };
-    const stored = body.session_state?.sandbox_id;
-    const applied = sandboxId === null ? stored == null : stored === sandboxId;
-    return applied ? "applied" : "rejected";
-  } catch (err) {
-    log(
-      `${logPrefix} failed session=${sessionId}: ${String(err instanceof Error ? err.message : err).slice(0, 120)}`,
-    );
-    return "failed";
-  }
-}
-
-/**
- * Write the live sandbox instance id forward (best-effort) so the next turn can reconnect it.
- * Only Daytona runs record a pointer; a local run leaves the row untouched.
- */
-export async function writeSandboxPointer(
-  sessionId: string,
-  pointer: SandboxPointerWrite,
-  deps: SandboxPointerDeps,
-): Promise<SandboxPointerWriteOutcome> {
-  return putSandboxPointer(sessionId, pointer.sandboxId, pointer.turnIndex, deps, "write");
-}
-
-/** Clear a terminal sandbox pointer under the same turn-index guard as pointer writes. */
-export async function clearSandboxPointer(
-  sessionId: string,
-  turnIndex: number,
-  deps: SandboxPointerDeps,
-): Promise<SandboxPointerWriteOutcome> {
-  return putSandboxPointer(sessionId, null, turnIndex, deps, "clear");
+  const latest = await fetchLatestSessionTurn(sessionId, undefined, deps);
+  const id = latest?.sandbox_id;
+  if (typeof id !== "string" || id.length === 0) return undefined;
+  return { sandboxId: id };
 }
diff --git a/services/runner/src/engines/sandbox_agent/session-continuity-durable.ts b/services/runner/src/engines/sandbox_agent/session-continuity-durable.ts
index cdf417033d..bb204dbd05 100644
--- a/services/runner/src/engines/sandbox_agent/session-continuity-durable.ts
+++ b/services/runner/src/engines/sandbox_agent/session-continuity-durable.ts
@@ -1,13 +1,12 @@
 /**
  * Durable mirror of `session-continuity.ts`'s in-memory store, so continuity survives a
- * runner restart. Reads the `session_states` row back into the in-memory store at session
- * setup, and writes the just-completed turn's record forward after
- * `SessionContinuityStore.record()`. Any failure (row absent, API unreachable) is best-effort:
- * it degrades to cold text replay, never a hard error.
+ * runner restart. Reads the session's LATEST `session_turns` row back into the in-memory
+ * store at session setup, and appends a fresh turn row after
+ * `SessionContinuityStore.record()`.
  *
- * The API row's `data` is replaced wholesale, not per-key-patched server-side, so
- * `syncHarnessSessionDurable` does a read-modify-write: GET the current row, splice in this
- * harness's fresh entry, PUT the whole `data` object back.
+ * `appendSessionTurn` is a plain INSERT (`POST /sessions/turns/`) — no read-modify-write, no
+ * race. It replaces the old `syncHarnessSessionDurable`, which GET-then-PUT the whole
+ * `session_states.data` blob.
  */
 import { apiBase } from "../../apiBase.ts";
 import type { SessionContinuityStore } from "./session-continuity.ts";
@@ -16,19 +15,19 @@ function defaultLog(msg: string): void {
   process.stderr.write(`[session-continuity/durable] ${msg}\n`);
 }
 
-interface WireHarnessSessionRecord {
+/** A platform entity reference (the API `Reference` shape). */
+export type TurnReference = { id?: string; slug?: string; version?: string };
+
+export interface WireSessionTurn {
+  harness_kind?: string;
   agent_session_id?: string;
+  sandbox_id?: string;
   turn_index?: number;
 }
 
-interface SessionStateDataWire {
-  latest_agent_session_id?: string;
-  harness_sessions?: Record<string, WireHarnessSessionRecord>;
-  latest_turn_index?: number;
-}
-
-interface SessionStateWire {
-  data?: SessionStateDataWire | null;
+interface SessionTurnsQueryResponseWire {
+  count?: number;
+  turns?: WireSessionTurn[];
 }
 
 export interface DurableContinuityDeps {
@@ -38,138 +37,162 @@ export interface DurableContinuityDeps {
   log?: (msg: string) => void;
 }
 
+/** The fields the turn-append write carries, beyond the (session, harness, turnIndex) key. */
+export interface SessionTurnAppend {
+  streamId: string;
+  agentSessionId?: string;
+  sandboxId?: string;
+  references?: TurnReference[];
+  traceId?: string;
+  spanId?: string;
+  startTime?: string;
+  endTime?: string;
+}
+
 /**
- * Read the durable row back into `store` for ONE harness, so a resume after a runner restart
- * (the in-memory map is empty) still sees a prior turn's eligibility exactly as if the process
- * had stayed up. Best-effort: any failure (row absent, API unreachable, storage disabled)
- * leaves `store` untouched — the caller then behaves exactly as it does today with an empty
- * store (cold replay), never throwing for a missing durable record.
+ * Fetch the LATEST turn for a session, optionally scoped to one harness. Ordered by `id`
+ * (uuid7 — insertion order, equivalent to `turn_index DESC` for an append-only table) via
+ * `windowing: {limit: 1, order: "descending"}`. Returns undefined on any failure (row absent,
+ * API unreachable) — every caller treats this as best-effort, degrading to cold replay.
  */
-export async function hydrateHarnessSessionFromDurable(
+export async function fetchLatestSessionTurn(
   sessionId: string,
-  harness: string,
-  store: SessionContinuityStore,
+  harness: string | undefined,
   deps: DurableContinuityDeps,
-): Promise<void> {
+): Promise<WireSessionTurn | undefined> {
   const log = deps.log ?? defaultLog;
   const doFetch = deps.fetchImpl ?? fetch;
   const base = deps.apiBase ?? apiBase();
   try {
-    const res = await doFetch(
-      `${base}/sessions/states/?session_id=${encodeURIComponent(sessionId)}`,
-      {
-        method: "GET",
-        headers: { authorization: deps.authorization },
+    const res = await doFetch(`${base}/sessions/turns/query`, {
+      method: "POST",
+      headers: {
+        "content-type": "application/json",
+        authorization: deps.authorization,
       },
-    );
+      body: JSON.stringify({
+        query: {
+          session_id: sessionId,
+          ...(harness ? { harness_kind: harness } : {}),
+        },
+        windowing: { limit: 1, order: "descending" },
+      }),
+    });
     if (!res.ok) {
-      log(`hydrate HTTP ${res.status} session=${sessionId} — starting cold`);
-      return;
+      log(
+        `latest-turn HTTP ${res.status} session=${sessionId} harness=${harness ?? "-"}`,
+      );
+      return undefined;
     }
-    const body = (await res.json()) as {
-      session_state?: SessionStateWire | null;
-    };
-    const data = body.session_state?.data;
-
-    // Restore the cross-harness latest-turn counter FIRST, independent of whether THIS harness
-    // has a record: it may exceed this harness's own turn (another harness ran the later turn),
-    // and understating it would make `isHarnessLoadEligible` wrongly pass a stale harness after
-    // a restart.
-    if (data?.latest_turn_index !== undefined) {
-      store.restoreLatestTurn(sessionId, data.latest_turn_index);
-    }
-
-    const record = data?.harness_sessions?.[harness];
-    if (!record?.agent_session_id || record.turn_index === undefined) {
-      return;
-    }
-    // Only seed the store when it has NOTHING for this (session, harness) yet — a live
-    // in-process record (this restart never happened) is always fresher than the durable
-    // mirror and must not be clobbered by a stale read.
-    if (store.get(sessionId, harness)) return;
-    store.record(
-      sessionId,
-      harness,
-      record.agent_session_id,
-      record.turn_index,
-    );
-    log(
-      `hydrated session=${sessionId} harness=${harness} turn=${record.turn_index}`,
-    );
+    const body = (await res.json()) as SessionTurnsQueryResponseWire;
+    return body.turns?.[0];
   } catch (err) {
     log(
-      `hydrate failed session=${sessionId} harness=${harness}: ${String(err instanceof Error ? err.message : err).slice(0, 160)}`,
+      `latest-turn failed session=${sessionId} harness=${harness ?? "-"}: ${String(err instanceof Error ? err.message : err).slice(0, 160)}`,
     );
+    return undefined;
+  }
+}
+
+/**
+ * Read the durable turn log back into `store` for ONE harness, so a resume after a runner
+ * restart (the in-memory map is empty) still sees a prior turn's eligibility exactly as if the
+ * process had stayed up. Best-effort: any failure (no turns yet, API unreachable) leaves
+ * `store` untouched — the caller then behaves exactly as it does today with an empty store
+ * (cold replay), never throwing for a missing durable record.
+ */
+export async function hydrateHarnessSessionFromDurable(
+  sessionId: string,
+  harness: string,
+  store: SessionContinuityStore,
+  deps: DurableContinuityDeps,
+): Promise<void> {
+  const log = deps.log ?? defaultLog;
+
+  // Restore the cross-harness latest-turn counter FIRST, independent of whether THIS harness
+  // authored it: another harness may have run the later turn, and understating the counter
+  // would make `isHarnessLoadEligible` wrongly pass a stale harness after a restart.
+  const latestOverall = await fetchLatestSessionTurn(
+    sessionId,
+    undefined,
+    deps,
+  );
+  if (latestOverall?.turn_index !== undefined) {
+    store.restoreLatestTurn(sessionId, latestOverall.turn_index);
+  }
+
+  // Only seed the store when it has NOTHING for this (session, harness) yet — a live
+  // in-process record (this restart never happened) is always fresher than the durable
+  // mirror and must not be clobbered by a stale read.
+  if (store.get(sessionId, harness)) return;
+
+  const latestForHarness =
+    latestOverall?.harness_kind === harness
+      ? latestOverall
+      : await fetchLatestSessionTurn(sessionId, harness, deps);
+  if (
+    !latestForHarness?.agent_session_id ||
+    latestForHarness.turn_index === undefined
+  ) {
+    return;
   }
+  store.record(
+    sessionId,
+    harness,
+    latestForHarness.agent_session_id,
+    latestForHarness.turn_index,
+  );
+  log(
+    `hydrated session=${sessionId} harness=${harness} turn=${latestForHarness.turn_index}`,
+  );
 }
 
 /**
- * Write this harness's just-completed-turn record forward to the durable row (read-modify-write
- * on `harness_sessions`, full-PUT semantics — see module doc). Call AFTER
- * `SessionContinuityStore.record()` so the value persisted matches the in-memory one exactly.
- * Best-effort and fire-and-forget from the caller's perspective: a failure here only means the
- * NEXT restart cold-replays this harness turn, never a broken current turn.
+ * Append this harness's just-completed-turn as a new `session_turns` row (a plain INSERT —
+ * no read, no merge, no race). Call AFTER `SessionContinuityStore.record()` so the persisted
+ * turn_index matches the in-memory one exactly. Best-effort and fire-and-forget from the
+ * caller's perspective: a failure here only means the NEXT restart cold-replays this harness
+ * turn, never a broken current turn.
  */
-export async function syncHarnessSessionDurable(
+export async function appendSessionTurn(
   sessionId: string,
   harness: string,
-  agentSessionId: string,
   turnIndex: number,
+  turn: SessionTurnAppend,
   deps: DurableContinuityDeps,
 ): Promise<void> {
   const log = deps.log ?? defaultLog;
   const doFetch = deps.fetchImpl ?? fetch;
   const base = deps.apiBase ?? apiBase();
   try {
-    const getRes = await doFetch(
-      `${base}/sessions/states/?session_id=${encodeURIComponent(sessionId)}`,
-      { method: "GET", headers: { authorization: deps.authorization } },
-    );
-    const existingData: SessionStateDataWire = getRes.ok
-      ? (((await getRes.json()) as { session_state?: SessionStateWire | null })
-          .session_state?.data ?? {})
-      : {};
-    const existing = existingData.harness_sessions ?? {};
-
-    const merged: Record<string, WireHarnessSessionRecord> = {
-      ...existing,
-      [harness]: {
-        agent_session_id: agentSessionId,
-        turn_index: turnIndex,
+    const res = await doFetch(`${base}/sessions/turns/`, {
+      method: "POST",
+      headers: {
+        "content-type": "application/json",
+        authorization: deps.authorization,
       },
-    };
-
-    // Never regress the cross-harness counter: another harness may already have persisted a
-    // higher latest turn. Take the max so the durable `latest_turn_index` is monotonic.
-    const latestTurnIndex = Math.max(
-      existingData.latest_turn_index ?? -1,
-      turnIndex,
-    );
-
-    const res = await doFetch(
-      `${base}/sessions/states/?session_id=${encodeURIComponent(sessionId)}`,
-      {
-        method: "PUT",
-        headers: {
-          "content-type": "application/json",
-          authorization: deps.authorization,
-        },
-        body: JSON.stringify({
-          data: {
-            ...existingData,
-            latest_agent_session_id: agentSessionId,
-            latest_turn_index: latestTurnIndex,
-            harness_sessions: merged,
-          },
-        }),
-      },
-    );
+      body: JSON.stringify({
+        session_id: sessionId,
+        stream_id: turn.streamId,
+        turn_index: turnIndex,
+        harness_kind: harness,
+        ...(turn.agentSessionId
+          ? { agent_session_id: turn.agentSessionId }
+          : {}),
+        ...(turn.sandboxId ? { sandbox_id: turn.sandboxId } : {}),
+        ...(turn.references?.length ? { references: turn.references } : {}),
+        ...(turn.traceId ? { trace_id: turn.traceId } : {}),
+        ...(turn.spanId ? { span_id: turn.spanId } : {}),
+        ...(turn.startTime ? { start_time: turn.startTime } : {}),
+        ...(turn.endTime ? { end_time: turn.endTime } : {}),
+      }),
+    });
     log(
-      `sync ${res.ok ? "OK" : `HTTP ${res.status}`} session=${sessionId} harness=${harness} turn=${turnIndex}`,
+      `append ${res.ok ? "OK" : `HTTP ${res.status}`} session=${sessionId} harness=${harness} turn=${turnIndex}`,
     );
   } catch (err) {
     log(
-      `sync failed session=${sessionId} harness=${harness}: ${String(err instanceof Error ? err.message : err).slice(0, 160)}`,
+      `append failed session=${sessionId} harness=${harness}: ${String(err instanceof Error ? err.message : err).slice(0, 160)}`,
     );
   }
 }
diff --git a/services/runner/src/protocol.ts b/services/runner/src/protocol.ts
index bce0dfbc89..cdf95a6c08 100644
--- a/services/runner/src/protocol.ts
+++ b/services/runner/src/protocol.ts
@@ -519,6 +519,13 @@ export interface AgentRunRequest {
    * the runner can include it in heartbeat and record-ingest calls. Absent otherwise.
    */
   projectId?: string;
+  /**
+   * The session's `session_streams` row id, captured for free from the alive-watchdog's
+   * heartbeat response (`sessions/alive.ts`) and threaded here before the engine runs. Present
+   * only for session-owned streaming runs; consumed at the turn-append write site
+   * (`engines/sandbox_agent.ts`) as `session_turns.stream_id`.
+   */
+  streamId?: string;
 }
 
 export interface AgentRunResult {
diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts
index d2ddc2987e..235f730559 100644
--- a/services/runner/src/server.ts
+++ b/services/runner/src/server.ts
@@ -906,12 +906,23 @@ async function runAndStreamWithApiBaseResolved(
     // The runner authenticates session calls AS the invoke caller (the run credential),
     // refreshing it for the turn's lifetime — never the admin key. Project scope is
     // resolved server-side from the credential, so no project_id rides the request.
-    const watchdog = startAliveWatchdog(
+    //
+    // onInterrupted (W7.4): a cancel/steer/kill against this session (via
+    // `POST /sessions/streams/` or the runner's own `/kill`) drops this turn's alive lock.
+    // The next heartbeat surfaces that as `is_current_turn: false`; wiring it to
+    // `controller.abort()` is what makes the control-plane signal actually reach this
+    // in-flight run — before this, a session-owned run's controller was never aborted.
+    // Awaited (WP3) so the first heartbeat's stream_id is ready before the turn starts.
+    const watchdog = await startAliveWatchdog(
       sessionId,
       turnId,
       runCredential(request),
+      () => controller.abort(),
     );
     aliveWatchdog = watchdog;
+    // The heartbeat response already carries the session_streams row id — free, no extra
+    // round-trip. Thread it onto the request so the engine's turn-append write has it.
+    request.streamId = watchdog.streamId();
     // A new turn supersedes any prior turn's unanswered gate: cancel stale pending
     // interactions (sparing this turn's own, plus a parked gate this turn answers in-band —
     // the resume resolves that one). Best-effort, never blocks the turn.
@@ -933,6 +944,8 @@ async function runAndStreamWithApiBaseResolved(
       watchdog.credential,
       liveEmit,
       seedForRun(request),
+      turnId,
+      request.runContext?.trace?.span_id,
     );
     // Record the inbound user turn first so the session record is the full conversation,
     // not just agent output. Interaction replies ride tool_result blocks (no text) and are
diff --git a/services/runner/src/sessions/alive.ts b/services/runner/src/sessions/alive.ts
index dd63c4681f..d5843c8611 100644
--- a/services/runner/src/sessions/alive.ts
+++ b/services/runner/src/sessions/alive.ts
@@ -49,13 +49,20 @@ function log(msg: string): void {
  * container `replica_id` (refreshes `owner` affinity) and the `turn_id` (proves alive ownership).
  * Authenticates AS the invoke caller (the run credential) — project scope is resolved server-side
  * from that credential, so no `project_id` rides the request.
+ *
+ * Returns both signals the one response body carries: `streamId` (the `session_streams` row
+ * uuid — the free gift of a call the runner already makes every turn, no new round-trip) and
+ * `interrupted: true` when the API reports `is_current_turn: false` (a cancel/steer/kill took
+ * this turn's alive/running lock since the last beat — W7.4, the control-signal path). A
+ * network/HTTP failure yields `{ streamId: undefined, interrupted: false }` (fail-open: a
+ * transient API blip must neither abort a healthy run nor fabricate a stream id).
  */
 async function sendHeartbeat(
   sessionId: string,
   turnId: string,
   authorization: string,
   isRunning = true,
-): Promise<void> {
+): Promise<{ streamId: string | undefined; interrupted: boolean }> {
   try {
     const url = `${apiBase()}/sessions/streams/heartbeat`;
     const res = await fetch(url, {
@@ -73,15 +80,27 @@ async function sendHeartbeat(
     });
     if (!res.ok) {
       log(`heartbeat HTTP ${res.status} session=${sessionId} turn=${turnId}`);
-    } else {
-      log(
-        `heartbeat OK session=${sessionId} turn=${turnId} running=${isRunning}`,
-      );
+      return { streamId: undefined, interrupted: false };
     }
+    const body = (await res.json()) as {
+      stream?: { id?: unknown } | null;
+      is_current_turn?: unknown;
+    };
+    const rawStreamId = body.stream?.id;
+    const streamId =
+      typeof rawStreamId === "string" && rawStreamId.length > 0
+        ? rawStreamId
+        : undefined;
+    const interrupted = body.is_current_turn === false;
+    log(
+      `heartbeat OK session=${sessionId} turn=${turnId} running=${isRunning}${interrupted ? " INTERRUPTED" : ""}`,
+    );
+    return { streamId, interrupted };
   } catch (err) {
     log(
       `heartbeat failed session=${sessionId} turn=${turnId}: ${String(err instanceof Error ? err.message : err).slice(0, 120)}`,
     );
+    return { streamId: undefined, interrupted: false };
   }
 }
 
@@ -132,20 +151,49 @@ export async function claimSessionOwnership(
  * inherits ownership via `turnId`. This watchdog heartbeats the API on the contract interval,
  * keeping the lock's TTL refreshed and the stream row `running`.
  *
- * Returns a `release()` function the caller MUST await in the run's `finally` so the
- * heartbeat stops and the session row is marked `ended`.
+ * `onInterrupted` (W7.4) fires AT MOST ONCE, the first time a beat reports the lock was taken
+ * by a cancel/steer/kill since the last beat — the caller wires this to `controller.abort()` so
+ * a control-plane cancel actually reaches the in-flight run. Before this, losing the alive lock
+ * was invisible to the runner process: a heartbeat's nx=True re-acquire silently re-armed the
+ * same lock under the same turn_id and the run continued as if nothing happened.
+ *
+ * Awaits the FIRST heartbeat (only) so its response's `stream_id` is ready before the caller
+ * starts the turn — every later heartbeat stays fire-and-forget. Returns a `release()` function
+ * the caller MUST await in the run's `finally` so the heartbeat stops and the row is marked
+ * `ended`.
  */
-export function startAliveWatchdog(
+export async function startAliveWatchdog(
   sessionId: string,
   turnId: string,
   authorization: string,
-): { release: () => Promise<void>; credential: () => string } {
+  onInterrupted?: () => void,
+): Promise<{
+  release: () => Promise<void>;
+  credential: () => string;
+  streamId: () => string | undefined;
+}> {
   // The run credential is an ephemeral Secret (~15-min TTL). Hold it mutably and refresh it
   // every Nth heartbeat (re-/check mints a fresh-expiry token) so a long turn never 401s.
   let credential = authorization;
   let beats = 0;
+  let interruptedFired = false;
+  let streamId: string | undefined;
+
+  const handleBeat = (result: {
+    streamId: string | undefined;
+    interrupted: boolean;
+  }): void => {
+    if (result.streamId) streamId = result.streamId;
+    if (result.interrupted && !interruptedFired) {
+      interruptedFired = true;
+      log(`interrupted session=${sessionId} turn=${turnId} -> aborting`);
+      onInterrupted?.();
+    }
+  };
 
-  void sendHeartbeat(sessionId, turnId, credential);
+  // Await the FIRST beat so streamId is ready before the caller starts the turn.
+  const first = await sendHeartbeat(sessionId, turnId, credential);
+  handleBeat(first);
 
   const interval = setInterval(() => {
     void (async () => {
@@ -154,7 +202,7 @@ export function startAliveWatchdog(
         const fresh = await refreshCredential(apiBase(), credential);
         if (fresh) credential = fresh;
       }
-      void sendHeartbeat(sessionId, turnId, credential);
+      handleBeat(await sendHeartbeat(sessionId, turnId, credential));
     })();
   }, REFRESH_INTERVAL_MS);
 
@@ -170,5 +218,6 @@ export function startAliveWatchdog(
       await sendHeartbeat(sessionId, turnId, credential, false);
     },
     credential: () => credential,
+    streamId: () => streamId,
   };
 }
diff --git a/services/runner/src/sessions/persist.ts b/services/runner/src/sessions/persist.ts
index 77a6c1f1d6..740e1c66a8 100644
--- a/services/runner/src/sessions/persist.ts
+++ b/services/runner/src/sessions/persist.ts
@@ -43,6 +43,8 @@ async function postEvent(
   eventIndex: number,
   sender: string,
   recordId?: string,
+  turnId?: string,
+  spanId?: string,
 ): Promise<void> {
   const url = `${apiBase()}/sessions/records/ingest`;
   let lastErr: unknown;
@@ -64,6 +66,10 @@ async function postEvent(
           record_source: sender,
           record_type: event.type,
           attributes: event,
+          // Tags the record for turn-grouping; span_id bridges to observability when
+          // the run has one in scope (both forward-fill only, absent is expected).
+          ...(turnId ? { turn_id: turnId } : {}),
+          ...(spanId ? { span_id: spanId } : {}),
         }),
       });
       if (!res.ok) throw new Error(`HTTP ${res.status}`);
@@ -94,12 +100,23 @@ export function persistEvent(
   sender: string = "agent",
   recordId?: string,
   redactor?: Redactor,
+  turnId?: string,
+  spanId?: string,
 ): void {
   // Redact at the sink: the durable copy is scrubbed; the live/in-memory event the harness
   // and the client stream still hold is untouched.
   const durable = redactor ? redactor.redactJson(event, "records") : event;
   const tail = (persistChains.get(sessionId) ?? Promise.resolve()).then(() =>
-    postEvent(sessionId, auth, durable, eventIndex, sender, recordId),
+    postEvent(
+      sessionId,
+      auth,
+      durable,
+      eventIndex,
+      sender,
+      recordId,
+      turnId,
+      spanId,
+    ),
   );
   persistChains.set(sessionId, tail);
 }
@@ -145,6 +162,8 @@ export function buildPersistingEmitter(
   auth: () => string,
   liveEmit?: (event: AgentEvent) => void,
   redactor?: Redactor,
+  turnId?: string,
+  spanId?: string,
 ): {
   emit: (event: AgentEvent) => void;
   /** Persist an out-of-band record (e.g. the inbound user turn) through the same
@@ -179,6 +198,8 @@ export function buildPersistingEmitter(
       "agent",
       stableRecordId(sessionId, id, "tool_call"),
       redactor,
+      turnId,
+      spanId,
     );
   };
 
@@ -234,6 +255,8 @@ export function buildPersistingEmitter(
           "agent",
           undefined,
           redactor,
+          turnId,
+          spanId,
         );
         return;
       }
@@ -262,6 +285,8 @@ export function buildPersistingEmitter(
           "agent",
           undefined,
           redactor,
+          turnId,
+          spanId,
         );
         return;
       }
@@ -281,6 +306,8 @@ export function buildPersistingEmitter(
         "agent",
         stableRecordId(sessionId, event.id, event.type),
         redactor,
+        turnId,
+        spanId,
       );
       return;
     }
@@ -294,6 +321,8 @@ export function buildPersistingEmitter(
       "agent",
       undefined,
       redactor,
+      turnId,
+      spanId,
     );
   };
 
@@ -308,6 +337,8 @@ export function buildPersistingEmitter(
       sender,
       undefined,
       redactor,
+      turnId,
+      spanId,
     );
   };
 
diff --git a/services/runner/src/version.ts b/services/runner/src/version.ts
index 35f2a55b67..185460f3fe 100644
--- a/services/runner/src/version.ts
+++ b/services/runner/src/version.ts
@@ -12,7 +12,7 @@ import pkg from "../package.json";
 export const PROTOCOL_VERSION = 1;
 export const RUNNER_VERSION: string = pkg.version;
 export const ENGINES = ["sandbox-agent"] as const;
-export const HARNESSES = ["pi_core", "claude", "pi_agenta"] as const;
+export const HARNESS_KINDS = ["pi_core", "claude", "pi_agenta"] as const;
 
 export interface RunnerInfo {
   status: "ok";
@@ -30,6 +30,6 @@ export function runnerInfo(): RunnerInfo {
     runner: RUNNER_VERSION,
     protocol: PROTOCOL_VERSION,
     engines: ENGINES,
-    harnesses: HARNESSES,
+    harnesses: HARNESS_KINDS,
   };
 }
diff --git a/services/runner/tests/acceptance/server-contract.test.ts b/services/runner/tests/acceptance/server-contract.test.ts
index a18ecfcd5a..b15a3e5685 100644
--- a/services/runner/tests/acceptance/server-contract.test.ts
+++ b/services/runner/tests/acceptance/server-contract.test.ts
@@ -41,8 +41,12 @@ const AUTH = { authorization: `Bearer ${TEST_TOKEN}` };
 
 async function listen(
   run: RunAgent,
+  token: string | null = TEST_TOKEN,
 ): Promise<{ url: string; close: () => Promise<void> }> {
-  if (!process.env[TOKEN_ENV]) process.env[TOKEN_ENV] = TEST_TOKEN;
+  // Force the configured token unconditionally (`null` = leave the env as the test set it,
+  // for the pre-set-secret / tokenless cases). A loaded dev env sets AGENTA_RUNNER_TOKEN=replace-me;
+  // a "set only if unset" guard would let that leak in and 401 every default-token request.
+  if (token !== null) process.env[TOKEN_ENV] = token;
   const server = createAgentServer(run);
   await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
   const { port } = server.address() as AddressInfo;
@@ -84,7 +88,7 @@ describe("GET /health contract", () => {
 
   it("is reachable without a bearer token even when AGENTA_RUNNER_TOKEN is set", async () => {
     process.env[TOKEN_ENV] = "health-gate-test";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/health`);
       assert.equal(res.status, 200);
@@ -112,7 +116,7 @@ describe("POST /run auth contract", () => {
 
   it("token configured, no header → 401 {ok:false, error:/Unauthorized/}", async () => {
     process.env[TOKEN_ENV] = "tok-abc";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/run`, { method: "POST", body: "{}" });
       assert.equal(res.status, 401);
@@ -126,7 +130,7 @@ describe("POST /run auth contract", () => {
 
   it("token configured, wrong bearer → 401", async () => {
     process.env[TOKEN_ENV] = "tok-abc";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -141,7 +145,7 @@ describe("POST /run auth contract", () => {
 
   it("token configured, correct Authorization: Bearer → 200", async () => {
     process.env[TOKEN_ENV] = "tok-abc";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -156,7 +160,7 @@ describe("POST /run auth contract", () => {
 
   it("token configured, correct X-Agenta-Runner-Token → 200", async () => {
     process.env[TOKEN_ENV] = "tok-abc";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -337,7 +341,7 @@ describe("POST /kill contract", () => {
 
   it("returns 401 when token is set and no bearer supplied", async () => {
     process.env[TOKEN_ENV] = "kill-tok";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/kill`, {
         method: "POST",
diff --git a/services/runner/tests/integration/server-smoke.test.ts b/services/runner/tests/integration/server-smoke.test.ts
index a5791e96fc..74f836ad8b 100644
--- a/services/runner/tests/integration/server-smoke.test.ts
+++ b/services/runner/tests/integration/server-smoke.test.ts
@@ -47,8 +47,12 @@ const AUTH = { authorization: `Bearer ${TEST_TOKEN}` };
 
 async function listen(
   run: RunAgent,
+  token: string | null = TEST_TOKEN,
 ): Promise<{ url: string; close: () => Promise<void> }> {
-  if (!process.env[TOKEN_ENV]) process.env[TOKEN_ENV] = TEST_TOKEN;
+  // Force the configured token unconditionally (`null` = leave the env as the test set it,
+  // for the tokenless/pre-set-secret cases). A loaded dev env sets AGENTA_RUNNER_TOKEN=replace-me;
+  // a "set only if unset" guard would let that leak in and 401 the AUTH-bearer stream test.
+  if (token !== null) process.env[TOKEN_ENV] = token;
   const server = createAgentServer(run);
   await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
   const { port } = server.address() as AddressInfo;
@@ -96,7 +100,7 @@ describe("server integration — in-process server with fake engine", () => {
 
   it("POST /run returns 401 when AGENTA_RUNNER_TOKEN is set and no bearer supplied", async () => {
     process.env[TOKEN_ENV] = "integration-secret";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -113,7 +117,7 @@ describe("server integration — in-process server with fake engine", () => {
 
   it("POST /run accepts the correct bearer token", async () => {
     process.env[TOKEN_ENV] = "integration-secret";
-    const s = await listen(okRun);
+    const s = await listen(okRun, null);
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -161,7 +165,7 @@ describe("session record persist — skipped when Redis unavailable", () => {
     // The alive watchdog POSTs to the API; without a live API server the fetch
     // will fail, but the module must swallow failures and never throw.
     const { startAliveWatchdog } = await import("../../src/sessions/alive.ts");
-    const watchdog = startAliveWatchdog("integ-sess", "integ-turn", "tok");
+    const watchdog = await startAliveWatchdog("integ-sess", "integ-turn", "tok");
     await assert.doesNotReject(() => watchdog.release());
   });
 
diff --git a/services/runner/tests/unit/sandbox-lifecycle.test.ts b/services/runner/tests/unit/sandbox-lifecycle.test.ts
index ab9799ef73..4d2add811d 100644
--- a/services/runner/tests/unit/sandbox-lifecycle.test.ts
+++ b/services/runner/tests/unit/sandbox-lifecycle.test.ts
@@ -45,12 +45,17 @@ function fakeSandbox(sandboxId: string | undefined, opts: FakeOpts = {}) {
     paused: 0,
     destroyed: 0,
     disposed: 0,
-    wrote: [] as Array<{ sandboxId: string; turnIndex: number }>,
-    cleared: [] as Array<{ sessionId: string; turnIndex: number }>,
+    appended: [] as Array<{
+      sessionId: string;
+      harness: string;
+      turnIndex: number;
+      sandboxId: string | undefined;
+    }>,
     logs: [] as string[],
   };
   const session = {
     id: "session-1",
+    agentSessionId: "agent-fake-1",
     onEvent() {},
     onPermissionRequest() {},
     async prompt() {
@@ -106,7 +111,14 @@ function fakeSandbox(sandboxId: string | undefined, opts: FakeOpts = {}) {
     createPersist: () => ({}) as any,
     sessionContinuityStore: continuityStore,
     hydrateHarnessSessionFromDurable: async () => {},
-    syncHarnessSessionDurable: async () => {},
+    appendSessionTurn: async (sessionId, harness, turnIndex, turn) => {
+      calls.appended.push({
+        sessionId,
+        harness,
+        turnIndex,
+        sandboxId: turn.sandboxId,
+      });
+    },
     startSandboxAgent: (async (startOpts: any) => {
       calls.starts.push({ sandboxId: startOpts.sandboxId });
       // The dead rung: Daytona already archived/deleted the parked sandbox, so reconnect
@@ -164,17 +176,6 @@ function fakeSandbox(sandboxId: string | undefined, opts: FakeOpts = {}) {
     // Lifecycle seam under test:
     readStoredSandboxPointer: async () =>
       sandboxId ? { sandboxId } : undefined,
-    clearSandboxPointer: async (sessionId, turnIndex) => {
-      calls.cleared.push({ sessionId, turnIndex });
-      return "applied";
-    },
-    writeSandboxPointer: async (_sessionId, pointer) => {
-      calls.wrote.push({
-        sandboxId: pointer.sandboxId,
-        turnIndex: pointer.turnIndex,
-      });
-      return "applied";
-    },
   };
   return { calls, deps };
 }
@@ -183,8 +184,9 @@ const daytonaRequest: AgentRunRequest = {
   harness: "claude",
   sandbox: "daytona",
   sessionId: "sess-1",
+  streamId: "stream-1",
   messages: [{ role: "user", content: "hello" }],
-  // A session-owned run always carries the invoke credential; the read/write helpers need it.
+  // A session-owned run always carries the invoke credential; the read/append helpers need it.
   telemetry: {
     exporters: { otlp: { headers: { authorization: "ApiKey abc" } } },
   } as any,
@@ -235,7 +237,7 @@ describe("remote sandbox reconnect ladder", () => {
     assert.equal(calls.starts[0].sandboxId, undefined);
   });
 
-  it("falls through to a fresh create without clearing on a transient reconnect failure", async () => {
+  it("falls through to a fresh create on a transient reconnect failure", async () => {
     const { calls, deps } = fakeSandbox("sbx-gone", { reconnectThrows: true });
     const result = await runSandboxAgent(
       daytonaRequest,
@@ -259,10 +261,9 @@ describe("remote sandbox reconnect ladder", () => {
       undefined,
       "the retry carries no id",
     );
-    assert.deepEqual(calls.cleared, []);
   });
 
-  it("clears the pointer before a fresh create on a terminal reconnect failure", async () => {
+  it("falls through to a fresh create on a terminal reconnect failure, no pointer clear needed", async () => {
     const { calls, deps } = fakeSandbox("sbx-gone", {
       reconnectTerminalState: "not-found",
     });
@@ -276,62 +277,37 @@ describe("remote sandbox reconnect ladder", () => {
 
     assert.equal(result.ok, true);
     assert.equal(calls.starts.length, 2);
-    assert.deepEqual(calls.cleared, [{ sessionId: "sess-1", turnIndex: 0 }]);
-  });
-
-  it("hydrates the turn index before the guarded clear (post-restart token)", async () => {
-    const { calls, deps } = fakeSandbox("sbx-gone", {
-      reconnectTerminalState: "not-found",
-    });
-    // A restarted runner has an empty in-memory store; the durable row knows turn 5.
-    deps.hydrateHarnessSessionFromDurable = async (
-      sessionId,
-      _harness,
-      store,
-    ) => {
-      store.restoreLatestTurn(sessionId, 5);
-    };
-
-    const result = await runSandboxAgent(
-      daytonaRequest,
-      undefined,
-      undefined,
-      deps,
-    );
-
-    assert.equal(result.ok, true);
-    assert.deepEqual(
-      calls.cleared,
-      [{ sessionId: "sess-1", turnIndex: 6 }],
-      "the clear must carry the hydrated index, not the cold store's 0",
-    );
+    // No explicit clear call exists anymore: the fresh sandbox this turn creates gets its own
+    // turn row at completion, whose higher turn_index naturally supersedes the dead pointer on
+    // the next `latest_turn` read.
+    assert.equal(calls.appended.length, 1);
   });
 
-  it("does not write a pointer for a local run", async () => {
+  it("does not append a turn for a local run without a streamId", async () => {
     const { calls, deps } = fakeSandbox(undefined);
 
     const result = await runSandboxAgent(
-      { ...daytonaRequest, sandbox: "local" },
+      { ...daytonaRequest, sandbox: "local", streamId: undefined },
       undefined,
       undefined,
       deps,
     );
 
     assert.equal(result.ok, true);
-    assert.deepEqual(
-      calls.wrote,
-      [],
-      "a local run must not overwrite a conversation's remote pointer",
-    );
+    assert.deepEqual(calls.appended, []);
   });
 
-  it("writes the live sandbox id forward for the next turn", async () => {
+  it("appends the live sandbox id forward on the completed turn's row", async () => {
     const { calls, deps } = fakeSandbox("sbx-99");
     await runSandboxAgent(daytonaRequest, undefined, undefined, deps);
-    assert.deepEqual(calls.wrote, [{ sandboxId: "sbx-99", turnIndex: 0 }]);
+    assert.equal(calls.appended.length, 1);
+    assert.equal(calls.appended[0].sessionId, "sess-1");
+    assert.equal(calls.appended[0].harness, "claude");
+    assert.equal(calls.appended[0].turnIndex, 0);
+    assert.equal(calls.appended[0].sandboxId, "sbx-99");
   });
 
-  it("writes the next turn index after durable continuity hydration", async () => {
+  it("appends at the next turn index after durable continuity hydration", async () => {
     const { calls, deps } = fakeSandbox("sbx-99");
     deps.hydrateHarnessSessionFromDurable = async (
       sessionId,
@@ -343,7 +319,8 @@ describe("remote sandbox reconnect ladder", () => {
 
     await runSandboxAgent(daytonaRequest, undefined, undefined, deps);
 
-    assert.deepEqual(calls.wrote, [{ sandboxId: "sbx-99", turnIndex: 6 }]);
+    assert.equal(calls.appended.length, 1);
+    assert.equal(calls.appended[0].turnIndex, 6);
   });
 
   it("trusts a stored pointer and reconnects by id without a compatibility check", async () => {
@@ -362,27 +339,21 @@ describe("remote sandbox reconnect ladder", () => {
     );
   });
 
-  it("awaits a rejected pointer write and logs the outcome without failing", async () => {
-    const { calls, deps } = fakeSandbox(undefined);
-    let writeFinished = false;
-    deps.writeSandboxPointer = async () => {
-      await Promise.resolve();
-      writeFinished = true;
-      return "rejected";
+  it("never throws when the turn-append call fails", async () => {
+    const { deps } = fakeSandbox(undefined);
+    deps.appendSessionTurn = async () => {
+      throw new Error("network error");
     };
 
+    // appendSessionTurn is called fire-and-forget (`void`), so a rejection here must never
+    // surface as an unhandled rejection or fail the run.
     const result = await runSandboxAgent(
       daytonaRequest,
       undefined,
       undefined,
       deps,
     );
-
     assert.equal(result.ok, true);
-    assert.equal(writeFinished, true);
-    assert.ok(
-      calls.logs.some((message) => message.includes("pointer write rejected")),
-    );
   });
 });
 
diff --git a/services/runner/tests/unit/sandbox-reconnect.test.ts b/services/runner/tests/unit/sandbox-reconnect.test.ts
index ab329882b5..d860402fba 100644
--- a/services/runner/tests/unit/sandbox-reconnect.test.ts
+++ b/services/runner/tests/unit/sandbox-reconnect.test.ts
@@ -1,22 +1,24 @@
 /**
- * Unit tests for the sandbox-id read/write helpers that back the remote reconnect ladder.
- * Exercised through injected fetch/deps -- no real HTTP.
+ * Unit tests for the sandbox-id read helper that backs the remote reconnect ladder. The write
+ * side (the live sandbox id landing on the next turn's row) is covered by
+ * `session-continuity-durable.test.ts`'s `appendSessionTurn` tests. Exercised through injected
+ * fetch/deps -- no real HTTP.
  *
  * Run: pnpm exec vitest run tests/unit/sandbox-reconnect.test.ts
  */
 import { describe, it } from "vitest";
 import assert from "node:assert/strict";
 
-import {
-  clearSandboxPointer,
-  readStoredSandboxPointer,
-  writeSandboxPointer,
-} from "../../src/engines/sandbox_agent/sandbox-reconnect.ts";
+import { readStoredSandboxPointer } from "../../src/engines/sandbox_agent/sandbox-reconnect.ts";
 
 const SILENT = () => {};
 
 function okResponse(body: unknown): Response {
-  return { ok: true, status: 200, json: async () => body } as unknown as Response;
+  return {
+    ok: true,
+    status: 200,
+    json: async () => body,
+  } as unknown as Response;
 }
 
 function errResponse(status: number): Response {
@@ -24,144 +26,86 @@ function errResponse(status: number): Response {
 }
 
 describe("readStoredSandboxPointer", () => {
-  it("returns the stored pointer from the durable row", async () => {
+  it("returns the latest turn's sandbox_id", async () => {
     const pointer = await readStoredSandboxPointer("sess-1", {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
       fetchImpl: (async () =>
         okResponse({
-          session_state: {
-            sandbox_id: "sbx-42",
-          },
+          turns: [{ sandbox_id: "sbx-42", turn_index: 3 }],
         })) as unknown as typeof fetch,
       log: SILENT,
     });
     assert.deepEqual(pointer, { sandboxId: "sbx-42" });
   });
 
-  it("returns undefined when the row has no sandbox_id", async () => {
-    const id = await readStoredSandboxPointer("sess-1", {
+  it("queries unscoped by harness (the pointer is session-wide, not per-harness)", async () => {
+    let body: Record<string, unknown> | undefined;
+    await readStoredSandboxPointer("sess-1", {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
-      fetchImpl: (async () =>
-        okResponse({ session_state: { sandbox_id: null } })) as unknown as typeof fetch,
+      fetchImpl: (async (_url: string, init?: RequestInit) => {
+        body = JSON.parse(init!.body as string);
+        return okResponse({ turns: [] });
+      }) as unknown as typeof fetch,
       log: SILENT,
     });
-    assert.equal(id, undefined);
+    assert.deepEqual(body!["query"], { session_id: "sess-1" });
+    assert.deepEqual(body!["windowing"], { limit: 1, order: "descending" });
   });
 
-  it("returns undefined on a non-2xx (degrades to a fresh create)", async () => {
+  it("returns undefined when no turn has a sandbox_id", async () => {
     const id = await readStoredSandboxPointer("sess-1", {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
-      fetchImpl: (async () => errResponse(503)) as unknown as typeof fetch,
+      fetchImpl: (async () =>
+        okResponse({ turns: [{ turn_index: 0 }] })) as unknown as typeof fetch,
       log: SILENT,
     });
     assert.equal(id, undefined);
   });
 
-  it("returns undefined when fetch throws, no throw out", async () => {
+  it("returns undefined when there are no turns yet", async () => {
     const id = await readStoredSandboxPointer("sess-1", {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
-      fetchImpl: (async () => {
-        throw new Error("ECONNREFUSED");
-      }) as unknown as typeof fetch,
+      fetchImpl: (async () =>
+        okResponse({ turns: [] })) as unknown as typeof fetch,
       log: SILENT,
     });
     assert.equal(id, undefined);
   });
 
-  it("ignores an empty-string id", async () => {
+  it("returns undefined on a non-2xx (degrades to a fresh create)", async () => {
     const id = await readStoredSandboxPointer("sess-1", {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
-      fetchImpl: (async () =>
-        okResponse({ session_state: { sandbox_id: "" } })) as unknown as typeof fetch,
+      fetchImpl: (async () => errResponse(503)) as unknown as typeof fetch,
       log: SILENT,
     });
     assert.equal(id, undefined);
   });
-});
 
-describe("clearSandboxPointer", () => {
-  it("PUTs null pointer fields with the guard token", async () => {
-    let body: Record<string, unknown> | undefined;
-    const outcome = await clearSandboxPointer("sess-1", 7, {
-      apiBase: "http://api:8000",
-      authorization: "ApiKey abc",
-      fetchImpl: (async (_url: string, init?: RequestInit) => {
-        body = JSON.parse(init!.body as string);
-        return okResponse({ session_state: { sandbox_id: null } });
-      }) as unknown as typeof fetch,
-      log: SILENT,
-    });
-
-    assert.deepEqual(body, {
-      sandbox_id: null,
-      sandbox_turn_index: 7,
-    });
-    assert.equal(outcome, "applied");
-  });
-});
-
-describe("writeSandboxPointer", () => {
-  it("PUTs the sandbox pointer and guard token on the row", async () => {
-    let body: Record<string, unknown> | undefined;
-    const outcome = await writeSandboxPointer("sess-1", {
-      sandboxId: "sbx-42",
-      turnIndex: 3,
-    }, {
+  it("returns undefined when fetch throws, no throw out", async () => {
+    const id = await readStoredSandboxPointer("sess-1", {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
-      fetchImpl: (async (_url: string, init?: RequestInit) => {
-        body = JSON.parse(init!.body as string);
-        return okResponse({ session_state: { sandbox_id: "sbx-42" } });
+      fetchImpl: (async () => {
+        throw new Error("ECONNREFUSED");
       }) as unknown as typeof fetch,
       log: SILENT,
     });
-    assert.deepEqual(body, {
-      sandbox_id: "sbx-42",
-      sandbox_turn_index: 3,
-    });
-    assert.equal(outcome, "applied");
+    assert.equal(id, undefined);
   });
 
-  it("returns rejected when the response keeps another sandbox id", async () => {
-    const outcome = await writeSandboxPointer("sess-1", {
-      sandboxId: "sbx-stale",
-      turnIndex: 1,
-    }, {
+  it("ignores an empty-string id", async () => {
+    const id = await readStoredSandboxPointer("sess-1", {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
       fetchImpl: (async () =>
-        okResponse({ session_state: { sandbox_id: "sbx-current" } })) as unknown as typeof fetch,
+        okResponse({ turns: [{ sandbox_id: "" }] })) as unknown as typeof fetch,
       log: SILENT,
     });
-    assert.equal(outcome, "rejected");
-  });
-
-  it("never throws when the PUT fails", async () => {
-    await assert.doesNotReject(() =>
-      writeSandboxPointer("sess-1", { sandboxId: "sbx-42", turnIndex: 0 }, {
-        apiBase: "http://api:8000",
-        authorization: "ApiKey abc",
-        fetchImpl: (async () => errResponse(503)) as unknown as typeof fetch,
-        log: SILENT,
-      }),
-    );
-  });
-
-  it("never throws when fetch throws", async () => {
-    await assert.doesNotReject(() =>
-      writeSandboxPointer("sess-1", { sandboxId: "sbx-42", turnIndex: 0 }, {
-        apiBase: "http://api:8000",
-        authorization: "ApiKey abc",
-        fetchImpl: (async () => {
-          throw new Error("ECONNREFUSED");
-        }) as unknown as typeof fetch,
-        log: SILENT,
-      }),
-    );
+    assert.equal(id, undefined);
   });
 });
diff --git a/services/runner/tests/unit/server.test.ts b/services/runner/tests/unit/server.test.ts
index 851740ae21..6b9d88abb5 100644
--- a/services/runner/tests/unit/server.test.ts
+++ b/services/runner/tests/unit/server.test.ts
@@ -43,8 +43,13 @@ const AUTH = { authorization: `Bearer ${TEST_TOKEN}` };
 
 async function listen(
   run: RunAgent,
+  token: string | null = TEST_TOKEN,
 ): Promise<{ url: string; close: () => Promise<void> }> {
-  if (!process.env[TOKEN_ENV]) process.env[TOKEN_ENV] = TEST_TOKEN;
+  // Force the configured token unconditionally (default TEST_TOKEN; `null` = leave the env
+  // as the test set it, for the tokenless-boot case). A loaded dev env (`load-env` before
+  // the suite) sets AGENTA_RUNNER_TOKEN=replace-me; a "set only if unset" guard would let
+  // that leak in and 401 every AUTH request. afterEach restores the pre-suite value.
+  if (token !== null) process.env[TOKEN_ENV] = token;
   const server = createAgentServer(run);
   await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
   const { port } = server.address() as AddressInfo;
@@ -142,8 +147,7 @@ describe("createAgentServer", () => {
   });
 
   it("POST /run without the token returns 401 when a token is configured", async () => {
-    process.env[TOKEN_ENV] = "s3cret";
-    const s = await listen(okRun);
+    const s = await listen(okRun, "s3cret");
     try {
       const res = await fetch(`${s.url}/run`, { method: "POST", body: "{}" });
       assert.equal(res.status, 401);
@@ -156,8 +160,7 @@ describe("createAgentServer", () => {
   });
 
   it("POST /run with a wrong token returns 401", async () => {
-    process.env[TOKEN_ENV] = "s3cret";
-    const s = await listen(okRun);
+    const s = await listen(okRun, "s3cret");
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -171,8 +174,7 @@ describe("createAgentServer", () => {
   });
 
   it("POST /run accepts the matching token via Authorization: Bearer", async () => {
-    process.env[TOKEN_ENV] = "s3cret";
-    const s = await listen(okRun);
+    const s = await listen(okRun, "s3cret");
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -186,8 +188,7 @@ describe("createAgentServer", () => {
   });
 
   it("POST /run accepts the matching token via X-Agenta-Runner-Token", async () => {
-    process.env[TOKEN_ENV] = "s3cret";
-    const s = await listen(okRun);
+    const s = await listen(okRun, "s3cret");
     try {
       const res = await fetch(`${s.url}/run`, {
         method: "POST",
@@ -202,8 +203,7 @@ describe("createAgentServer", () => {
 
   it("GET /health is reachable without the token even when one is configured", async () => {
     // Health is for liveness probes and carries no secrets, so the token gate is on /run only.
-    process.env[TOKEN_ENV] = "s3cret";
-    const s = await listen(okRun);
+    const s = await listen(okRun, "s3cret");
     try {
       const res = await fetch(`${s.url}/health`);
       assert.equal(res.status, 200);
diff --git a/services/runner/tests/unit/session-alive-interrupt.test.ts b/services/runner/tests/unit/session-alive-interrupt.test.ts
new file mode 100644
index 0000000000..72bd2c0380
--- /dev/null
+++ b/services/runner/tests/unit/session-alive-interrupt.test.ts
@@ -0,0 +1,121 @@
+/**
+ * W7.4: a cancel/steer/kill against a session-owned run must reach the runner process.
+ *
+ * The API's heartbeat response carries `is_current_turn: false` when this turn's alive/
+ * running lock was gone or reassigned since the last beat. `startAliveWatchdog`'s
+ * `onInterrupted` callback is how `server.ts` wires that into `controller.abort()` — these
+ * tests pin the watchdog's half of that contract via fetch interception.
+ */
+import { describe, it, beforeEach, afterEach, vi } from "vitest";
+import assert from "node:assert/strict";
+
+const fetchCalls: Array<{ url: string; body: unknown }> = [];
+let nextIsCurrentTurn: boolean | undefined = true;
+
+vi.stubGlobal("fetch", async (url: string, init?: RequestInit) => {
+  const body = init?.body ? JSON.parse(init.body as string) : undefined;
+  fetchCalls.push({ url, body });
+  const payload: Record<string, unknown> = { ok: true };
+  if (nextIsCurrentTurn !== undefined) {
+    payload.is_current_turn = nextIsCurrentTurn;
+  }
+  return new Response(JSON.stringify(payload), { status: 200 });
+});
+
+const { startAliveWatchdog } = await import("../../src/sessions/alive.ts");
+
+function flushMicrotasks(): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, 0));
+}
+
+beforeEach(() => {
+  fetchCalls.length = 0;
+  nextIsCurrentTurn = true;
+});
+
+afterEach(() => {
+  vi.restoreAllMocks();
+});
+
+describe("startAliveWatchdog onInterrupted", () => {
+  it("does not fire onInterrupted while is_current_turn stays true", async () => {
+    const onInterrupted = vi.fn();
+    const watchdog = await startAliveWatchdog(
+      "sess-ok",
+      "turn-ok",
+      "proj-1",
+      onInterrupted,
+    );
+    await flushMicrotasks();
+
+    assert.equal(onInterrupted.mock.calls.length, 0);
+    await watchdog.release();
+  });
+
+  it("fires onInterrupted when a beat reports is_current_turn: false", async () => {
+    const onInterrupted = vi.fn();
+    const watchdog = await startAliveWatchdog(
+      "sess-cancelled",
+      "turn-cancelled",
+      "proj-1",
+      onInterrupted,
+    );
+    await flushMicrotasks();
+    assert.equal(onInterrupted.mock.calls.length, 0, "not interrupted yet");
+
+    // A cancel/steer/kill landed: the next beat reports the lock was taken.
+    nextIsCurrentTurn = false;
+    // Directly drive another beat by releasing and re-starting is unnecessary — the interval
+    // path is exercised in the interval test below; here we simulate the first beat itself
+    // being interrupted (e.g. the lock was already gone before the watchdog's first heartbeat
+    // observed it — a steer that raced session start).
+    const watchdog2 = await startAliveWatchdog(
+      "sess-cancelled-2",
+      "turn-cancelled-2",
+      "proj-1",
+      onInterrupted,
+    );
+    await flushMicrotasks();
+
+    assert.equal(onInterrupted.mock.calls.length, 1);
+
+    await watchdog.release();
+    await watchdog2.release();
+  });
+
+  it("fires onInterrupted at most once even if later beats keep reporting interrupted", async () => {
+    nextIsCurrentTurn = false;
+    const onInterrupted = vi.fn();
+    const watchdog = await startAliveWatchdog(
+      "sess-repeat",
+      "turn-repeat",
+      "proj-1",
+      onInterrupted,
+    );
+    await flushMicrotasks();
+    assert.equal(onInterrupted.mock.calls.length, 1);
+
+    // A second beat still reporting interrupted (e.g. release()'s final heartbeat) must not
+    // fire the callback again — the caller's controller.abort() is idempotent but the
+    // callback itself should still only ever fire once per turn.
+    await watchdog.release();
+    assert.equal(onInterrupted.mock.calls.length, 1);
+  });
+
+  it("treats a network/HTTP failure as NOT interrupted (fail-open)", async () => {
+    vi.stubGlobal("fetch", async () => {
+      throw new Error("network down");
+    });
+    const onInterrupted = vi.fn();
+    const watchdog = await startAliveWatchdog(
+      "sess-neterr",
+      "turn-neterr",
+      "proj-1",
+      onInterrupted,
+    );
+    await flushMicrotasks();
+
+    assert.equal(onInterrupted.mock.calls.length, 0);
+    await assert.doesNotReject(() => watchdog.release());
+  });
+});
diff --git a/services/runner/tests/unit/session-alive.test.ts b/services/runner/tests/unit/session-alive.test.ts
index 168e7aecfd..1cff659c27 100644
--- a/services/runner/tests/unit/session-alive.test.ts
+++ b/services/runner/tests/unit/session-alive.test.ts
@@ -3,20 +3,28 @@
  *
  * Verifies observable behaviors via fetch interception:
  *  - the watchdog fires an immediate heartbeat on start
+ *  - the FIRST heartbeat's response `stream.id` is captured and exposed via `streamId()`
  *  - release() calls the heartbeat endpoint with is_running=false and status=ended
- *  - heartbeat failures are swallowed (no throw)
+ *  - heartbeat failures are swallowed (no throw), and `streamId()` stays undefined
  */
 import { describe, it, beforeEach, afterEach, vi } from "vitest";
 import assert from "node:assert/strict";
 
 const fetchCalls: Array<{ url: string; body: unknown }> = [];
 let fetchShouldFail = false;
+let streamIdToReturn: string | undefined = "stream-default";
 
 vi.stubGlobal("fetch", async (url: string, init?: RequestInit) => {
   const body = init?.body ? JSON.parse(init.body as string) : undefined;
   fetchCalls.push({ url, body });
   if (fetchShouldFail) return new Response("", { status: 500 });
-  return new Response(JSON.stringify({ ok: true }), { status: 200 });
+  return new Response(
+    JSON.stringify({
+      ok: true,
+      stream: streamIdToReturn ? { id: streamIdToReturn } : undefined,
+    }),
+    { status: 200 },
+  );
 });
 
 const { startAliveWatchdog } = await import("../../src/sessions/alive.ts");
@@ -29,6 +37,7 @@ function flushMicrotasks(): Promise<void> {
 beforeEach(() => {
   fetchCalls.length = 0;
   fetchShouldFail = false;
+  streamIdToReturn = "stream-default";
 });
 
 afterEach(() => {
@@ -37,8 +46,7 @@ afterEach(() => {
 
 describe("startAliveWatchdog", () => {
   it("sends an immediate heartbeat on start carrying turn_id and is_running=true", async () => {
-    const watchdog = startAliveWatchdog("sess-1", "turn-abc", "proj-1");
-    await flushMicrotasks();
+    const watchdog = await startAliveWatchdog("sess-1", "turn-abc", "proj-1");
 
     const hb = fetchCalls.find((c) => c.url.includes("heartbeat"));
     assert.ok(hb, "expected a heartbeat call");
@@ -53,11 +61,35 @@ describe("startAliveWatchdog", () => {
     await watchdog.release();
   });
 
+  it("captures stream_id from the first heartbeat's response — no extra round-trip", async () => {
+    streamIdToReturn = "stream-xyz";
+    const watchdog = await startAliveWatchdog("sess-1", "turn-abc", "proj-1");
+
+    assert.equal(watchdog.streamId(), "stream-xyz");
+    // Exactly one heartbeat call was needed to obtain it.
+    const heartbeats = fetchCalls.filter((c) => c.url.includes("heartbeat"));
+    assert.equal(heartbeats.length, 1);
+
+    await watchdog.release();
+  });
+
+  it("streamId() is undefined when the heartbeat response carries no stream", async () => {
+    streamIdToReturn = undefined;
+    const watchdog = await startAliveWatchdog("sess-1", "turn-abc", "proj-1");
+    assert.equal(watchdog.streamId(), undefined);
+    await watchdog.release();
+  });
+
+  it("streamId() is undefined when the first heartbeat fails", async () => {
+    fetchShouldFail = true;
+    const watchdog = await startAliveWatchdog("sess-1", "turn-abc", "proj-1");
+    assert.equal(watchdog.streamId(), undefined);
+    await watchdog.release();
+  });
+
   it("uses a stable replica_id across turns (distinct from turn_id)", async () => {
-    const w1 = startAliveWatchdog("sess-a", "turn-1", "proj-1");
-    await flushMicrotasks();
-    const w2 = startAliveWatchdog("sess-b", "turn-2", "proj-1");
-    await flushMicrotasks();
+    const w1 = await startAliveWatchdog("sess-a", "turn-1", "proj-1");
+    const w2 = await startAliveWatchdog("sess-b", "turn-2", "proj-1");
 
     const replicas = fetchCalls
       .filter((c) => c.url.includes("heartbeat"))
@@ -71,8 +103,7 @@ describe("startAliveWatchdog", () => {
   });
 
   it("release() sends a final heartbeat with is_running=false (status derived server-side)", async () => {
-    const watchdog = startAliveWatchdog("sess-2", "turn-xyz", "proj-2");
-    await flushMicrotasks();
+    const watchdog = await startAliveWatchdog("sess-2", "turn-xyz", "proj-2");
     fetchCalls.length = 0; // reset after the start heartbeat
 
     await watchdog.release();
@@ -82,15 +113,34 @@ describe("startAliveWatchdog", () => {
     const body = final.body as Record<string, unknown>;
     assert.equal(body["is_running"], false);
     assert.equal(body["turn_id"], "turn-xyz");
-    assert.ok(!("status" in body), "status was dropped from the heartbeat contract");
+    assert.ok(
+      !("status" in body),
+      "status was dropped from the heartbeat contract",
+    );
   });
 
   it("swallows heartbeat failures — never throws", async () => {
     fetchShouldFail = true;
-    const watchdog = startAliveWatchdog("sess-3", "run-fail", "proj-3");
-    await flushMicrotasks();
+    const watchdog = await startAliveWatchdog("sess-3", "run-fail", "proj-3");
 
     // release() should not throw even if fetch fails.
     await assert.doesNotReject(() => watchdog.release());
   });
+
+  it("a later heartbeat updates streamId() when it returns a fresh id", async () => {
+    vi.useFakeTimers();
+    try {
+      streamIdToReturn = "stream-first";
+      const watchdog = await startAliveWatchdog("sess-1", "turn-abc", "proj-1");
+      assert.equal(watchdog.streamId(), "stream-first");
+
+      streamIdToReturn = "stream-second";
+      await vi.advanceTimersByTimeAsync(30_000);
+      assert.equal(watchdog.streamId(), "stream-second");
+
+      await watchdog.release();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
 });
diff --git a/services/runner/tests/unit/session-continuity-durable.test.ts b/services/runner/tests/unit/session-continuity-durable.test.ts
index db85f41c9f..f4ec30c5ca 100644
--- a/services/runner/tests/unit/session-continuity-durable.test.ts
+++ b/services/runner/tests/unit/session-continuity-durable.test.ts
@@ -1,9 +1,9 @@
 /**
- * Unit tests for the durable mirror (`engines/sandbox_agent/session-continuity-durable.ts`).
+ * Unit tests for the durable turn log (`engines/sandbox_agent/session-continuity-durable.ts`).
  *
  * Exercised through injected fetch/deps -- no real HTTP. Covers: hydrate seeds an empty store
- * from the durable row and never clobbers a live in-process record; both hydrate and sync are
- * best-effort (a 503/throw never propagates out of a run).
+ * from the latest turn and never clobbers a live in-process record; append is a plain POST (no
+ * GET, no read-modify-write); both are best-effort (a 503/throw never propagates out of a run).
  *
  * Run: pnpm exec vitest run tests/unit/session-continuity-durable.test.ts
  */
@@ -11,8 +11,9 @@ import { describe, it } from "vitest";
 import assert from "node:assert/strict";
 
 import {
+  appendSessionTurn,
+  fetchLatestSessionTurn,
   hydrateHarnessSessionFromDurable,
-  syncHarnessSessionDurable,
 } from "../../src/engines/sandbox_agent/session-continuity-durable.ts";
 import {
   SessionContinuityStore,
@@ -22,28 +23,116 @@ import {
 const SILENT = () => {};
 
 function okResponse(body: unknown): Response {
-  return { ok: true, status: 200, json: async () => body } as unknown as Response;
+  return {
+    ok: true,
+    status: 200,
+    json: async () => body,
+  } as unknown as Response;
 }
 
 function errResponse(status: number): Response {
   return { ok: false, status, json: async () => ({}) } as unknown as Response;
 }
 
+describe("fetchLatestSessionTurn", () => {
+  it("POSTs a query scoped to session (+harness when given), windowed to the latest one", async () => {
+    let body: Record<string, unknown> | undefined;
+    let url: string | undefined;
+    await fetchLatestSessionTurn("sess-1", "claude", {
+      apiBase: "http://api:8000",
+      authorization: "ApiKey abc",
+      fetchImpl: (async (u: string, init?: RequestInit) => {
+        url = u;
+        body = JSON.parse(init!.body as string);
+        return okResponse({ count: 0, turns: [] });
+      }) as unknown as typeof fetch,
+      log: SILENT,
+    });
+    assert.equal(url, "http://api:8000/sessions/turns/query");
+    assert.deepEqual(body, {
+      query: { session_id: "sess-1", harness_kind: "claude" },
+      windowing: { limit: 1, order: "descending" },
+    });
+  });
+
+  it("omits harness from the query when not given", async () => {
+    let body: Record<string, unknown> | undefined;
+    await fetchLatestSessionTurn("sess-1", undefined, {
+      apiBase: "http://api:8000",
+      authorization: "ApiKey abc",
+      fetchImpl: (async (_u: string, init?: RequestInit) => {
+        body = JSON.parse(init!.body as string);
+        return okResponse({ turns: [] });
+      }) as unknown as typeof fetch,
+      log: SILENT,
+    });
+    assert.deepEqual(body!["query"], { session_id: "sess-1" });
+  });
+
+  it("returns the first (latest) turn from the response", async () => {
+    const turn = await fetchLatestSessionTurn("sess-1", "claude", {
+      apiBase: "http://api:8000",
+      authorization: "ApiKey abc",
+      fetchImpl: (async () =>
+        okResponse({
+          count: 1,
+          turns: [
+            {
+              harness_kind: "claude",
+              agent_session_id: "agent-1",
+              sandbox_id: "sbx-1",
+              turn_index: 3,
+            },
+          ],
+        })) as unknown as typeof fetch,
+      log: SILENT,
+    });
+    assert.deepEqual(turn, {
+      harness_kind: "claude",
+      agent_session_id: "agent-1",
+      sandbox_id: "sbx-1",
+      turn_index: 3,
+    });
+  });
+
+  it("returns undefined on a non-2xx response, no throw", async () => {
+    const turn = await fetchLatestSessionTurn("sess-1", "claude", {
+      apiBase: "http://api:8000",
+      authorization: "ApiKey abc",
+      fetchImpl: (async () => errResponse(503)) as unknown as typeof fetch,
+      log: SILENT,
+    });
+    assert.equal(turn, undefined);
+  });
+
+  it("returns undefined when fetch throws, no throw out", async () => {
+    const turn = await fetchLatestSessionTurn("sess-1", "claude", {
+      apiBase: "http://api:8000",
+      authorization: "ApiKey abc",
+      fetchImpl: (async () => {
+        throw new Error("ECONNREFUSED");
+      }) as unknown as typeof fetch,
+      log: SILENT,
+    });
+    assert.equal(turn, undefined);
+  });
+});
+
 describe("hydrateHarnessSessionFromDurable", () => {
-  it("seeds an empty store from the durable row", async () => {
+  it("seeds an empty store from the latest turn for this harness", async () => {
     const store = new SessionContinuityStore();
     await hydrateHarnessSessionFromDurable("sess-1", "claude", store, {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
       fetchImpl: (async () =>
         okResponse({
-          session_state: {
-            data: {
-              harness_sessions: {
-                claude: { agent_session_id: "agent-restored", turn_index: 2 },
-              },
+          turns: [
+            {
+              harness_kind: "claude",
+              agent_session_id: "agent-restored",
+              turn_index: 2,
             },
-          },
+          ],
         })) as unknown as typeof fetch,
       log: SILENT,
     });
@@ -62,13 +151,13 @@ describe("hydrateHarnessSessionFromDurable", () => {
       authorization: "ApiKey abc",
       fetchImpl: (async () =>
         okResponse({
-          session_state: {
-            data: {
-              harness_sessions: {
-                claude: { agent_session_id: "agent-OLD", turn_index: 0 },
-              },
+          turns: [
+            {
+              harness_kind: "claude",
+              agent_session_id: "agent-OLD",
+              turn_index: 0,
             },
-          },
+          ],
         })) as unknown as typeof fetch,
       log: SILENT,
     });
@@ -108,24 +197,13 @@ describe("hydrateHarnessSessionFromDurable", () => {
     assert.equal(store.get("sess-1", "claude"), undefined);
   });
 
-  it("does nothing when the row has no record for this harness", async () => {
+  it("does nothing when there are no turns yet", async () => {
     const store = new SessionContinuityStore();
     await hydrateHarnessSessionFromDurable("sess-1", "claude", store, {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
       fetchImpl: (async () =>
-        okResponse({ session_state: { data: { harness_sessions: {} } } })) as unknown as typeof fetch,
-      log: SILENT,
-    });
-    assert.equal(store.get("sess-1", "claude"), undefined);
-  });
-
-  it("does nothing when session_state is null (no row yet)", async () => {
-    const store = new SessionContinuityStore();
-    await hydrateHarnessSessionFromDurable("sess-1", "claude", store, {
-      apiBase: "http://api:8000",
-      authorization: "ApiKey abc",
-      fetchImpl: (async () => okResponse({ session_state: null })) as unknown as typeof fetch,
+        okResponse({ turns: [] })) as unknown as typeof fetch,
       log: SILENT,
     });
     assert.equal(store.get("sess-1", "claude"), undefined);
@@ -133,28 +211,48 @@ describe("hydrateHarnessSessionFromDurable", () => {
 
   it("restores the cross-harness latest_turn_index, keeping a stale harness INELIGIBLE after restart", async () => {
     // Durable state: claude ran turns 0-1, pi ran turn 2 (the latest). A fresh runner (empty
-    // store) returns to claude. Without restoring latest_turn_index, hydrate would set latest=1
-    // (claude's own turn) and wrongly report claude eligible — the double-switch bug on restart.
+    // store) returns to claude. Without restoring the cross-harness counter, hydrate would set
+    // latest=1 (claude's own turn) and wrongly report claude eligible — the double-switch bug.
     const store = new SessionContinuityStore();
-    const durableRow = {
-      session_state: {
-        data: {
-          latest_turn_index: 2,
-          harness_sessions: {
-            claude: { agent_session_id: "agent-claude", turn_index: 1 },
-            pi: { agent_session_id: "agent-pi", turn_index: 2 },
-          },
-        },
-      },
-    };
-    const fetchRow = (async () => okResponse(durableRow)) as unknown as typeof fetch;
+    const fetchImpl = (async (url: string, init?: RequestInit) => {
+      const body = JSON.parse(init!.body as string) as {
+        query: { harness_kind?: string };
+      };
+      if (body.query.harness_kind === "claude") {
+        return okResponse({
+          turns: [
+            {
+              harness_kind: "claude",
+              agent_session_id: "agent-claude",
+              turn_index: 1,
+            },
+          ],
+        });
+      }
+      if (body.query.harness_kind === "pi") {
+        return okResponse({
+          turns: [
+            { harness_kind: "pi", agent_session_id: "agent-pi", turn_index: 2 },
+          ],
+        });
+      }
+      // overall latest (no harness filter): pi's turn 2 wins.
+      return okResponse({
+        turns: [{ harness_kind: "pi", agent_session_id: "agent-pi", turn_index: 2 }],
+      });
+    }) as unknown as typeof fetch;
+
     await hydrateHarnessSessionFromDurable("sess-1", "claude", store, {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
-      fetchImpl: fetchRow,
+      fetchImpl,
       log: SILENT,
     });
-    assert.equal(store.latestTurn("sess-1"), 2, "latest must reflect pi's turn 2, not claude's 1");
+    assert.equal(
+      store.latestTurn("sess-1"),
+      2,
+      "latest must reflect pi's turn 2, not claude's 1",
+    );
     assert.equal(
       isHarnessLoadEligible("sess-1", "claude", store),
       false,
@@ -164,7 +262,7 @@ describe("hydrateHarnessSessionFromDurable", () => {
     await hydrateHarnessSessionFromDurable("sess-1", "pi", store, {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
-      fetchImpl: fetchRow,
+      fetchImpl,
       log: SILENT,
     });
     assert.equal(isHarnessLoadEligible("sess-1", "pi", store), true);
@@ -173,158 +271,182 @@ describe("hydrateHarnessSessionFromDurable", () => {
   it("restores latest_turn_index even for a harness with no record of its own", async () => {
     const store = new SessionContinuityStore();
     await hydrateHarnessSessionFromDurable("sess-1", "codex", store, {
-      apiBase: "http://api:8000",
-      authorization: "ApiKey abc",
-      fetchImpl: (async () =>
-        okResponse({
-          session_state: {
-            data: {
-              latest_turn_index: 4,
-              harness_sessions: {
-                claude: { agent_session_id: "agent-claude", turn_index: 4 },
-              },
-            },
-          },
-        })) as unknown as typeof fetch,
-      log: SILENT,
-    });
-    assert.equal(store.get("sess-1", "codex"), undefined, "codex has no record of its own");
-    assert.equal(store.latestTurn("sess-1"), 4, "but the conversation counter is still restored");
-  });
-});
-
-describe("syncHarnessSessionDurable", () => {
-  it("PUTs a read-modify-write merge that preserves other harnesses' entries", async () => {
-    const calls: Array<{ method: string; body?: unknown }> = [];
-    await syncHarnessSessionDurable("sess-1", "claude", "agent-new", 3, {
-      apiBase: "http://api:8000",
-      authorization: "ApiKey abc",
-      fetchImpl: (async (url: string, init?: RequestInit) => {
-        const method = init?.method ?? "GET";
-        if (method === "GET") {
-          calls.push({ method });
-          return okResponse({
-            session_state: {
-              data: {
-                harness_sessions: {
-                  pi: { agent_session_id: "agent-pi", turn_index: 1 },
-                },
-              },
-            },
-          });
-        }
-        calls.push({ method, body: JSON.parse(init!.body as string) });
-        return okResponse({});
-      }) as unknown as typeof fetch,
-      log: SILENT,
-    });
-
-    const put = calls.find((c) => c.method === "PUT")!;
-    const data = (put.body as Record<string, unknown>)["data"] as Record<string, unknown>;
-    assert.equal(data["latest_agent_session_id"], "agent-new");
-    assert.equal(data["latest_turn_index"], 3);
-    const harnessSessions = data["harness_sessions"] as Record<string, unknown>;
-    assert.deepEqual(harnessSessions["claude"], {
-      agent_session_id: "agent-new",
-      turn_index: 3,
-    });
-    assert.deepEqual(harnessSessions["pi"], {
-      agent_session_id: "agent-pi",
-      turn_index: 1,
-    });
-  });
-
-  it("never regresses latest_turn_index below the durable row's existing value", async () => {
-    // This harness completes turn 2, but the durable row already recorded latest_turn_index=5
-    // (another harness ran a later turn). The PUT must keep 5, not overwrite it down to 2.
-    let putBody: Record<string, unknown> | undefined;
-    await syncHarnessSessionDurable("sess-1", "claude", "agent-new", 2, {
       apiBase: "http://api:8000",
       authorization: "ApiKey abc",
       fetchImpl: (async (_url: string, init?: RequestInit) => {
-        if ((init?.method ?? "GET") === "GET") {
-          return okResponse({
-            session_state: {
-              data: {
-                latest_turn_index: 5,
-                harness_sessions: {
-                  pi: { agent_session_id: "agent-pi", turn_index: 5 },
-                },
-              },
-            },
-          });
+        const body = JSON.parse(init!.body as string) as {
+          query: { harness_kind?: string };
+        };
+        if (body.query.harness_kind === "codex") {
+          return okResponse({ turns: [] });
         }
-        putBody = JSON.parse(init!.body as string);
-        return okResponse({});
+        return okResponse({
+          turns: [
+            {
+              harness_kind: "claude",
+              agent_session_id: "agent-claude",
+              turn_index: 4,
+            },
+          ],
+        });
       }) as unknown as typeof fetch,
       log: SILENT,
     });
-    const putData = putBody!["data"] as Record<string, unknown>;
-    assert.equal(putData["latest_turn_index"], 5, "must not regress the counter below 5");
     assert.equal(
-      (putData["harness_sessions"] as Record<string, unknown>)["claude"] !== undefined,
-      true,
-      "claude's own turn-2 record is still written",
+      store.get("sess-1", "codex"),
+      undefined,
+      "codex has no record of its own",
+    );
+    assert.equal(
+      store.latestTurn("sess-1"),
+      4,
+      "but the conversation counter is still restored",
     );
   });
+});
 
-  it("advances latest_turn_index when this turn is the highest", async () => {
-    let putBody: Record<string, unknown> | undefined;
-    await syncHarnessSessionDurable("sess-1", "claude", "agent-new", 6, {
-      apiBase: "http://api:8000",
-      authorization: "ApiKey abc",
-      fetchImpl: (async (_url: string, init?: RequestInit) => {
-        if ((init?.method ?? "GET") === "GET") {
-          return okResponse({
-            session_state: { data: { latest_turn_index: 5, harness_sessions: {} } },
+describe("appendSessionTurn", () => {
+  it("POSTs a plain create — no GET, one call only", async () => {
+    const calls: Array<{ method: string; url: string; body?: unknown }> = [];
+    await appendSessionTurn(
+      "sess-1",
+      "claude",
+      3,
+      { streamId: "stream-1", agentSessionId: "agent-new" },
+      {
+        apiBase: "http://api:8000",
+        authorization: "ApiKey abc",
+        fetchImpl: (async (url: string, init?: RequestInit) => {
+          calls.push({
+            method: init?.method ?? "GET",
+            url,
+            body: init?.body ? JSON.parse(init.body as string) : undefined,
           });
-        }
-        putBody = JSON.parse(init!.body as string);
-        return okResponse({});
-      }) as unknown as typeof fetch,
-      log: SILENT,
+          return okResponse({ turn: {} });
+        }) as unknown as typeof fetch,
+        log: SILENT,
+      },
+    );
+    assert.equal(
+      calls.length,
+      1,
+      "append must be a single POST, no read-modify-write",
+    );
+    assert.equal(calls[0].method, "POST");
+    assert.equal(calls[0].url, "http://api:8000/sessions/turns/");
+    assert.deepEqual(calls[0].body, {
+      session_id: "sess-1",
+      stream_id: "stream-1",
+      turn_index: 3,
+      harness_kind: "claude",
+      agent_session_id: "agent-new",
     });
-    assert.equal((putBody!["data"] as Record<string, unknown>)["latest_turn_index"], 6);
   });
 
-  it("never throws when the GET fails (falls back to an empty merge base)", async () => {
-    await assert.doesNotReject(() =>
-      syncHarnessSessionDurable("sess-1", "claude", "agent-new", 0, {
+  it("carries sandbox_id, references, and trace_id when given", async () => {
+    let body: Record<string, unknown> | undefined;
+    await appendSessionTurn(
+      "sess-1",
+      "claude",
+      1,
+      {
+        streamId: "stream-1",
+        agentSessionId: "agent-1",
+        sandboxId: "sbx-1",
+        references: [{ id: "wf-1" }, { id: "rev-1", version: "v1" }],
+        traceId: "trace-abc",
+      },
+      {
         apiBase: "http://api:8000",
         authorization: "ApiKey abc",
         fetchImpl: (async (_url: string, init?: RequestInit) => {
-          if ((init?.method ?? "GET") === "GET") return errResponse(503);
-          return okResponse({});
+          body = JSON.parse(init!.body as string);
+          return okResponse({ turn: {} });
         }) as unknown as typeof fetch,
         log: SILENT,
-      }),
+      },
     );
+    assert.equal(body!["sandbox_id"], "sbx-1");
+    assert.deepEqual(body!["references"], [
+      { id: "wf-1" },
+      { id: "rev-1", version: "v1" },
+    ]);
+    assert.equal(body!["trace_id"], "trace-abc");
   });
 
-  it("never throws when the whole call chain throws (network error)", async () => {
+  it("never throws when the POST fails (503)", async () => {
     await assert.doesNotReject(() =>
-      syncHarnessSessionDurable("sess-1", "claude", "agent-new", 0, {
-        apiBase: "http://api:8000",
-        authorization: "ApiKey abc",
-        fetchImpl: (async () => {
-          throw new Error("ECONNREFUSED");
-        }) as unknown as typeof fetch,
-        log: SILENT,
-      }),
+      appendSessionTurn(
+        "sess-1",
+        "claude",
+        0,
+        { streamId: "stream-1" },
+        {
+          apiBase: "http://api:8000",
+          authorization: "ApiKey abc",
+          fetchImpl: (async () => errResponse(503)) as unknown as typeof fetch,
+          log: SILENT,
+        },
+      ),
     );
   });
 
-  it("never throws when the PUT itself returns a non-2xx (503)", async () => {
+  it("never throws when fetch itself throws (network error)", async () => {
     await assert.doesNotReject(() =>
-      syncHarnessSessionDurable("sess-1", "claude", "agent-new", 0, {
-        apiBase: "http://api:8000",
-        authorization: "ApiKey abc",
-        fetchImpl: (async (_url: string, init?: RequestInit) => {
-          if ((init?.method ?? "GET") === "GET") return errResponse(404);
-          return errResponse(503);
-        }) as unknown as typeof fetch,
-        log: SILENT,
-      }),
+      appendSessionTurn(
+        "sess-1",
+        "claude",
+        0,
+        { streamId: "stream-1" },
+        {
+          apiBase: "http://api:8000",
+          authorization: "ApiKey abc",
+          fetchImpl: (async () => {
+            throw new Error("ECONNREFUSED");
+          }) as unknown as typeof fetch,
+          log: SILENT,
+        },
+      ),
+    );
+  });
+});
+
+describe("two turns append cleanly (no RMW race)", () => {
+  it("appending turn N+1 does not read or depend on turn N's row", async () => {
+    // The defining behavioral change: unlike the old GET-then-PUT, two sequential appends never
+    // issue a GET, so there is nothing for a concurrent writer to race.
+    const posts: Array<Record<string, unknown>> = [];
+    const deps = {
+      apiBase: "http://api:8000",
+      authorization: "ApiKey abc",
+      fetchImpl: (async (_url: string, init?: RequestInit) => {
+        assert.equal(
+          init?.method,
+          "POST",
+          "every call must be a POST, never a GET",
+        );
+        posts.push(JSON.parse(init!.body as string));
+        return okResponse({ turn: {} });
+      }) as unknown as typeof fetch,
+      log: SILENT,
+    };
+    await appendSessionTurn(
+      "sess-1",
+      "claude",
+      0,
+      { streamId: "stream-1", agentSessionId: "agent-a" },
+      deps,
+    );
+    await appendSessionTurn(
+      "sess-1",
+      "claude",
+      1,
+      { streamId: "stream-1", agentSessionId: "agent-b" },
+      deps,
     );
+    assert.equal(posts.length, 2);
+    assert.equal(posts[0]["turn_index"], 0);
+    assert.equal(posts[1]["turn_index"], 1);
+    assert.equal(posts[1]["agent_session_id"], "agent-b");
   });
 });
diff --git a/services/runner/tests/unit/session-persist.test.ts b/services/runner/tests/unit/session-persist.test.ts
index 5db1789fc7..3f07f821cd 100644
--- a/services/runner/tests/unit/session-persist.test.ts
+++ b/services/runner/tests/unit/session-persist.test.ts
@@ -223,6 +223,66 @@ describe("buildPersistingEmitter", () => {
   });
 });
 
+describe("buildPersistingEmitter turn/span tagging", () => {
+  it("stamps turn_id and span_id on every posted record when both are supplied", async () => {
+    const { emit, persist, flush } = buildPersistingEmitter(
+      "sess-turn",
+      () => "Secret t",
+      undefined,
+      undefined,
+      "turn-abc",
+      "span-def",
+    );
+
+    persist({ type: "message", text: "hi" }, "user");
+    emit({ type: "message", text: "hello" });
+    emit({ type: "done" });
+    await flush();
+
+    const bodies = postedBodies as Array<Record<string, unknown>>;
+    assert.equal(bodies.length, 3);
+    for (const body of bodies) {
+      assert.equal(body["turn_id"], "turn-abc");
+      assert.equal(body["span_id"], "span-def");
+    }
+  });
+
+  it("omits turn_id/span_id from the body when neither is supplied", async () => {
+    const { emit, flush } = buildPersistingEmitter("sess-noturn", () => "Secret t");
+
+    emit({ type: "message", text: "hello" });
+    await flush();
+
+    const body = postedBodies[0] as Record<string, unknown>;
+    assert.equal("turn_id" in body, false);
+    assert.equal("span_id" in body, false);
+  });
+
+  it("stamps turn_id on coalesced and tool-call records too", async () => {
+    const { emit, flush } = buildPersistingEmitter(
+      "sess-turn-tc",
+      () => "Secret t",
+      undefined,
+      undefined,
+      "turn-tc",
+    );
+
+    emit({ type: "message_start", id: "m1" });
+    emit({ type: "message_delta", id: "m1", delta: "hi" });
+    emit({ type: "message_end", id: "m1" });
+    emit({ type: "tool_call", id: "call_1", name: "bash", input: { command: "ls" } });
+    emit({ type: "tool_result", id: "call_1", output: "ok" });
+    await flush();
+
+    const bodies = postedBodies as Array<Record<string, unknown>>;
+    assert.equal(bodies.length, 3);
+    for (const body of bodies) {
+      assert.equal(body["turn_id"], "turn-tc");
+      assert.equal("span_id" in body, false);
+    }
+  });
+});
+
 describe("drainPersist", () => {
   it("resolves immediately when no events are pending", async () => {
     await assert.doesNotReject(() => drainPersist("no-session"));

From 4363de24d05bc8fd3003328ef13e46517e2eb7ad Mon Sep 17 00:00:00 2001
From: Mahmoud Mabrouk <mahmoud@agenta.ai>
Date: Sat, 18 Jul 2026 19:28:50 +0200
Subject: [PATCH 2/3] test(sessions): pin the record-ingest span_id contract in
 the runner persist test

Add a persist test whose fetch stub enforces the API contract (span_id must be a
16-hex OTel span id, else 422), so a regression to a non-span-shaped span_id fails
here instead of silently dropping records. The old stub returned 200 for any body
and hid the mismatch. Also switch the turn/span tagging test to a real 16-hex span
id.

Claude-Session: https://claude.ai/code/session_01DnWRxU3dCJ11hgDidm26vW
---
 .../runner/tests/unit/session-persist.test.ts | 166 +++++++++++++++---
 1 file changed, 144 insertions(+), 22 deletions(-)

diff --git a/services/runner/tests/unit/session-persist.test.ts b/services/runner/tests/unit/session-persist.test.ts
index 3f07f821cd..b5672af3ed 100644
--- a/services/runner/tests/unit/session-persist.test.ts
+++ b/services/runner/tests/unit/session-persist.test.ts
@@ -25,10 +25,8 @@ vi.stubGlobal("fetch", async (_url: string, init?: RequestInit) => {
   return new Response(JSON.stringify({ ok: true }), { status: 200 });
 });
 
-const {
-  buildPersistingEmitter,
-  drainPersist,
-} = await import("../../src/sessions/persist.ts");
+const { buildPersistingEmitter, drainPersist } =
+  await import("../../src/sessions/persist.ts");
 
 beforeEach(() => {
   postedBodies.length = 0;
@@ -89,8 +87,8 @@ describe("buildPersistingEmitter", () => {
     const live: unknown[] = [];
     const { emit, flush } = buildPersistingEmitter(
       "sess-2",
-      () => "Secret t", (e) =>
-      live.push(e),
+      () => "Secret t",
+      (e) => live.push(e),
     );
 
     emit({ type: "message_start", id: "m1" });
@@ -147,8 +145,18 @@ describe("buildPersistingEmitter", () => {
 
     // A tool call streams a growing partial-args snapshot for one id.
     emit({ type: "tool_call", id: "call_1", name: "bash", input: {} });
-    emit({ type: "tool_call", id: "call_1", name: "bash", input: { command: "fi" } });
-    emit({ type: "tool_call", id: "call_1", name: "bash", input: { command: "find ." } });
+    emit({
+      type: "tool_call",
+      id: "call_1",
+      name: "bash",
+      input: { command: "fi" },
+    });
+    emit({
+      type: "tool_call",
+      id: "call_1",
+      name: "bash",
+      input: { command: "find ." },
+    });
     // A different step closes the open call.
     emit({ type: "tool_result", id: "call_1", output: "ok" });
     emit({ type: "done" });
@@ -173,10 +181,23 @@ describe("buildPersistingEmitter", () => {
   });
 
   it("a different tool id flushes the previous open call", async () => {
-    const { emit, flush } = buildPersistingEmitter("sess-tc2", () => "Secret t");
+    const { emit, flush } = buildPersistingEmitter(
+      "sess-tc2",
+      () => "Secret t",
+    );
 
-    emit({ type: "tool_call", id: "call_a", name: "read", input: { path: "/x" } });
-    emit({ type: "tool_call", id: "call_b", name: "read", input: { path: "/y" } });
+    emit({
+      type: "tool_call",
+      id: "call_a",
+      name: "read",
+      input: { path: "/x" },
+    });
+    emit({
+      type: "tool_call",
+      id: "call_b",
+      name: "read",
+      input: { path: "/y" },
+    });
     await flush();
 
     const bodies = postedBodies as Array<Record<string, unknown>>;
@@ -184,14 +205,25 @@ describe("buildPersistingEmitter", () => {
       (b) => (b["attributes"] as Record<string, unknown>)["input"],
     );
     assert.deepEqual(inputs, [{ path: "/x" }, { path: "/y" }]);
-    assert.deepEqual(bodies.map((b) => b["record_index"]), [0, 1]);
+    assert.deepEqual(
+      bodies.map((b) => b["record_index"]),
+      [0, 1],
+    );
   });
 
   it("flushes an open (paused) tool_call on drain", async () => {
-    const { emit, flush } = buildPersistingEmitter("sess-tc3", () => "Secret t");
+    const { emit, flush } = buildPersistingEmitter(
+      "sess-tc3",
+      () => "Secret t",
+    );
 
     // A paused call ends the turn with its slot still open.
-    emit({ type: "tool_call", id: "call_p", name: "bash", input: { command: "ls" } });
+    emit({
+      type: "tool_call",
+      id: "call_p",
+      name: "bash",
+      input: { command: "ls" },
+    });
     await flush();
 
     const bodies = postedBodies as Array<Record<string, unknown>>;
@@ -205,15 +237,28 @@ describe("buildPersistingEmitter", () => {
   it("flushes an open tool_call when the idle TTL fires", async () => {
     vi.useFakeTimers();
     try {
-      const { emit, flush } = buildPersistingEmitter("sess-tc4", () => "Secret t");
+      const { emit, flush } = buildPersistingEmitter(
+        "sess-tc4",
+        () => "Secret t",
+      );
 
-      emit({ type: "tool_call", id: "call_ttl", name: "bash", input: { command: "x" } });
+      emit({
+        type: "tool_call",
+        id: "call_ttl",
+        name: "bash",
+        input: { command: "x" },
+      });
       // Nothing follows; only the TTL can close it.
       assert.equal(postedBodies.length, 0);
       await vi.advanceTimersByTimeAsync(3000);
       assert.equal(postedBodies.length, 1);
       assert.equal(
-        ((postedBodies[0] as Record<string, unknown>)["attributes"] as Record<string, unknown>)["type"],
+        (
+          (postedBodies[0] as Record<string, unknown>)["attributes"] as Record<
+            string,
+            unknown
+          >
+        )["type"],
         "tool_call",
       );
       await flush();
@@ -225,13 +270,16 @@ describe("buildPersistingEmitter", () => {
 
 describe("buildPersistingEmitter turn/span tagging", () => {
   it("stamps turn_id and span_id on every posted record when both are supplied", async () => {
+    // A real 16-hex OTel span id (the runContext.trace.span_id shape), NOT a UUID — the API
+    // types this field as a 16-hex string, so anything else 422s (FINDING-record-ingest-422).
+    const spanId = "a1b2c3d4e5f6a7b8";
     const { emit, persist, flush } = buildPersistingEmitter(
       "sess-turn",
       () => "Secret t",
       undefined,
       undefined,
       "turn-abc",
-      "span-def",
+      spanId,
     );
 
     persist({ type: "message", text: "hi" }, "user");
@@ -243,12 +291,15 @@ describe("buildPersistingEmitter turn/span tagging", () => {
     assert.equal(bodies.length, 3);
     for (const body of bodies) {
       assert.equal(body["turn_id"], "turn-abc");
-      assert.equal(body["span_id"], "span-def");
+      assert.equal(body["span_id"], spanId);
     }
   });
 
   it("omits turn_id/span_id from the body when neither is supplied", async () => {
-    const { emit, flush } = buildPersistingEmitter("sess-noturn", () => "Secret t");
+    const { emit, flush } = buildPersistingEmitter(
+      "sess-noturn",
+      () => "Secret t",
+    );
 
     emit({ type: "message", text: "hello" });
     await flush();
@@ -270,7 +321,12 @@ describe("buildPersistingEmitter turn/span tagging", () => {
     emit({ type: "message_start", id: "m1" });
     emit({ type: "message_delta", id: "m1", delta: "hi" });
     emit({ type: "message_end", id: "m1" });
-    emit({ type: "tool_call", id: "call_1", name: "bash", input: { command: "ls" } });
+    emit({
+      type: "tool_call",
+      id: "call_1",
+      name: "bash",
+      input: { command: "ls" },
+    });
     emit({ type: "tool_result", id: "call_1", output: "ok" });
     await flush();
 
@@ -283,13 +339,79 @@ describe("buildPersistingEmitter turn/span tagging", () => {
   });
 });
 
+describe("buildPersistingEmitter API contract", () => {
+  // The API's SessionRecordIngestRequest types span_id as a 16-hex OTel span id. A stub that
+  // returns 200 for ANY body (like the other tests here) hid the real bug: the field used to
+  // be typed as UUID and every ingest 422'd. This stub enforces the real contract, so a
+  // regression to a non-span-shaped span_id fails here instead of silently dropping records.
+  const OTEL_SPAN_ID = /^[0-9a-fA-F]{16}$/;
+
+  it("emits a span_id the API contract accepts (16-hex OTel span id, not a UUID)", async () => {
+    const accepted: Array<Record<string, unknown>> = [];
+    const rejected: Array<Record<string, unknown>> = [];
+    const contractFetch = (async (_url: string, init?: RequestInit) => {
+      const body = init?.body
+        ? (JSON.parse(init.body as string) as Record<string, unknown>)
+        : {};
+      if (
+        body.span_id !== undefined &&
+        !OTEL_SPAN_ID.test(String(body.span_id))
+      ) {
+        rejected.push(body);
+        return new Response(JSON.stringify({ detail: "uuid_parsing" }), {
+          status: 422,
+        });
+      }
+      accepted.push(body);
+      return new Response(JSON.stringify({ ok: true }), { status: 200 });
+    }) as typeof fetch;
+
+    const prior = globalThis.fetch;
+    globalThis.fetch = contractFetch;
+    try {
+      const spanId = "a1b2c3d4e5f6a7b8"; // the runContext.trace.span_id shape
+      const { emit, persist, flush } = buildPersistingEmitter(
+        "sess-contract",
+        () => "Secret t",
+        undefined,
+        undefined,
+        "turn-1",
+        spanId,
+      );
+      persist({ type: "message", text: "hi" }, "user");
+      emit({ type: "message", text: "hello" });
+      emit({ type: "done" });
+      await flush();
+
+      // Every record cleared the contract-validating stub; nothing 422'd or dropped.
+      assert.equal(rejected.length, 0);
+      assert.equal(accepted.length, 3);
+      for (const body of accepted)
+        assert.match(String(body.span_id), OTEL_SPAN_ID);
+    } finally {
+      globalThis.fetch = prior;
+    }
+  });
+
+  it("the contract stub has teeth: a non-span-shaped span_id would 422", () => {
+    // Guards the test above: the old mock returned 200 for any body, so a UUID (or any
+    // arbitrary string) sailed through. The real contract rejects both.
+    assert.equal(OTEL_SPAN_ID.test("a1b2c3d4e5f6a7b8"), true);
+    assert.equal(OTEL_SPAN_ID.test("span-def"), false);
+    assert.equal(OTEL_SPAN_ID.test("f".repeat(32)), false); // a UUID (32 hex) is not a span id
+  });
+});
+
 describe("drainPersist", () => {
   it("resolves immediately when no events are pending", async () => {
     await assert.doesNotReject(() => drainPersist("no-session"));
   });
 
   it("waits for all queued events before resolving", async () => {
-    const { emit, flush } = buildPersistingEmitter("sess-drain", () => "Secret t");
+    const { emit, flush } = buildPersistingEmitter(
+      "sess-drain",
+      () => "Secret t",
+    );
     emit({ type: "message", text: "x" });
     emit({ type: "done" });
     await flush(); // same as drainPersist internally

From cae71e49ce3b8c08638c1f171118bc2422725bd8 Mon Sep 17 00:00:00 2001
From: Mahmoud Mabrouk <mahmoud@agenta.ai>
Date: Sun, 19 Jul 2026 16:19:34 +0200
Subject: [PATCH 3/3] fix(runner): make turn_index a per-conversation-turn
 counter

---
 .../src/engines/sandbox_agent/environment.ts  |   6 -
 .../src/engines/sandbox_agent/run-turn.ts     |  11 +-
 .../unit/session-keepalive-engine.test.ts     | 108 +++++++++++++++++-
 3 files changed, 117 insertions(+), 8 deletions(-)

diff --git a/services/runner/src/engines/sandbox_agent/environment.ts b/services/runner/src/engines/sandbox_agent/environment.ts
index cad062774b..e77fe382bb 100644
--- a/services/runner/src/engines/sandbox_agent/environment.ts
+++ b/services/runner/src/engines/sandbox_agent/environment.ts
@@ -113,7 +113,6 @@ import {
 import { hydrateHarnessSessionFromDurable } from "./session-continuity-durable.ts";
 import {
   eligibleAgentSessionId,
-  nextTurnIndex,
   sessionContinuityStore,
 } from "./session-continuity.ts";
 import { projectScopeFor } from "./session-identity.ts";
@@ -957,11 +956,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.
diff --git a/services/runner/src/engines/sandbox_agent/run-turn.ts b/services/runner/src/engines/sandbox_agent/run-turn.ts
index b7e50689a4..4e4f5ffe4f 100644
--- a/services/runner/src/engines/sandbox_agent/run-turn.ts
+++ b/services/runner/src/engines/sandbox_agent/run-turn.ts
@@ -66,7 +66,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";
 
@@ -86,6 +89,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 = [];
diff --git a/services/runner/tests/unit/session-keepalive-engine.test.ts b/services/runner/tests/unit/session-keepalive-engine.test.ts
index 28e26decf0..2ec1365398 100644
--- a/services/runner/tests/unit/session-keepalive-engine.test.ts
+++ b/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();