Skip to content

feat(runner): rebase JP's session turns/streams runner onto the decomposed runner #5376

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mmabrouk wants to merge 3 commits into from
Closed

feat(runner): rebase JP's session turns/streams runner onto the decomposed runner #5376

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8c48235
feat(sessions): rebase session turns/streams runner onto v0.105.5
mmabrouk Jul 18, 2026
4363de2
test(sessions): pin the record-ingest span_id contract in the runner …
mmabrouk Jul 18, 2026
cae71e4
fix(runner): make turn_index a per-conversation-turn counter
mmabrouk Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 11 additions & 54 deletions services/runner/src/engines/sandbox_agent/environment.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -117,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";
Expand Down Expand Up @@ -642,31 +637,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 {
Expand Down Expand Up @@ -979,29 +956,9 @@ 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;
// 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({
Expand Down
32 changes: 25 additions & 7 deletions services/runner/src/engines/sandbox_agent/run-turn.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,9 +64,12 @@ import {
shouldSuppressPausedToolCallUpdate,
} from "./runtime-policy.ts";
import {
syncHarnessSessionDurable,
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";

Expand All @@ -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 = [];
Expand Down Expand Up @@ -567,16 +576,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(() => {});

@mmabrouk mmabrouk Jul 18, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The swallow here is acceptable only because sessions/persist.ts logs the terminal failure after its retries. Worth a one-line comment saying so, since this exact silent-drop shape is how the span-id 422 went unnoticed until live QA.

}
} else if (stopReason === "paused") {
// A pause stopped mid-turn, after the harness may have written a partial turn natively.
Expand Down
13 changes: 6 additions & 7 deletions services/runner/src/engines/sandbox_agent/runtime-contracts.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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
Expand Down
120 changes: 15 additions & 105 deletions services/runner/src/engines/sandbox_agent/sandbox-reconnect.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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
Expand All @@ -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 };
}
Loading
Loading