Skip to content

Commit 28ce10a

Browse files
mmabroukmmabrouk
committed
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
1 parent 386c63e commit 28ce10a

19 files changed

Lines changed: 947 additions & 667 deletions

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

Lines changed: 11 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -98,11 +98,7 @@ import {
9898
routeSessionEventToActiveTurn,
9999
} from "./session-events.ts";
100100
import { buildSandboxProvider } from "./provider.ts";
101-
import {
102-
clearSandboxPointer,
103-
readStoredSandboxPointer,
104-
writeSandboxPointer,
105-
} from "./sandbox-reconnect.ts";
101+
import { readStoredSandboxPointer } from "./sandbox-reconnect.ts";
106102
import type {
107103
AcquireEnvironmentResult,
108104
SandboxAgentDeps,
@@ -641,31 +637,13 @@ export async function acquireEnvironment(
641637
logger(
642638
`reconnect failed sandbox=${storedSandboxPointer.sandboxId}, creating fresh: ${conciseError(err, plan.harness)}`,
643639
);
644-
if (
645-
err instanceof DaytonaReconnectTerminalError &&
646-
sessionForMount &&
647-
runCred
648-
) {
649-
// The post-hydrate write later in acquire is authoritative. This clear only prevents
650-
// repeated doomed reconnects if acquire fails before reaching that write. Hydrate
651-
// first: after a runner restart the in-memory store is behind the durable
652-
// latest_turn_index, and an unhydrated guard token would be rejected as stale.
653-
await (
654-
deps.hydrateHarnessSessionFromDurable ??
655-
hydrateHarnessSessionFromDurable
656-
)(
657-
sessionForMount,
658-
plan.harness,
659-
deps.sessionContinuityStore ?? sessionContinuityStore,
660-
{ authorization: runCred, log: logger },
661-
);
662-
await (deps.clearSandboxPointer ?? clearSandboxPointer)(
663-
sessionForMount,
664-
nextTurnIndex(
665-
sessionForMount,
666-
deps.sessionContinuityStore ?? sessionContinuityStore,
667-
),
668-
{ authorization: runCred, log: logger },
640+
// No explicit pointer clear needed: turns are append-only, so the fresh sandbox this
641+
// turn creates below gets its own turn row at completion, and that row's higher
642+
// turn_index naturally supersedes the dead one on the next `latest_turn` read — the
643+
// staleness guard the old states model needed dissolves with the ordering.
644+
if (err instanceof DaytonaReconnectTerminalError) {
645+
logger(
646+
`terminal Daytona state '${err.state}' for sandbox=${storedSandboxPointer.sandboxId}, not retrying reconnect`,
669647
);
670648
}
671649
} finally {
@@ -970,24 +948,9 @@ export async function acquireEnvironment(
970948
environment.continuityTurnIndex = continuitySessionKey
971949
? nextTurnIndex(continuitySessionKey, continuityStore)
972950
: undefined;
973-
// Daytona only: a local run must not overwrite a conversation's remote pointer (switching
974-
// sandboxes mid-conversation would strand the parked Daytona instance).
975-
if (plan.isDaytona && sessionForMount && runCred) {
976-
const liveSandboxId = environment.sandbox?.sandboxId ?? plan.sandboxId;
977-
const pointerWriteOutcome = await (
978-
deps.writeSandboxPointer ?? writeSandboxPointer
979-
)(
980-
sessionForMount,
981-
{
982-
sandboxId: liveSandboxId,
983-
turnIndex: environment.continuityTurnIndex ?? 0,
984-
},
985-
{ authorization: runCred, log: logger },
986-
);
987-
logger(
988-
`sandbox pointer write ${pointerWriteOutcome} session=${sessionForMount} sandbox=${liveSandboxId}`,
989-
);
990-
}
951+
// The live sandbox id rides forward as a field on the turn-append row written at turn end
952+
// (see `appendSessionTurn` call in `runTurn`), not a separate pre-turn pointer PUT: the
953+
// turns table is append-only, so there is nothing to overwrite mid-conversation.
991954
let loadedFromContinuity = false;
992955
if (priorAgentSessionId && localSessionId) {
993956
await persist.updateSession({

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

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ import {
6161
shouldSuppressPausedToolCallUpdate,
6262
} from "./runtime-policy.ts";
6363
import {
64-
syncHarnessSessionDurable,
64+
appendSessionTurn,
6565
} from "./session-continuity-durable.ts";
6666
import { sessionContinuityStore } from "./session-continuity.ts";
6767
import { priorMessages } from "./transcript.ts";
@@ -552,16 +552,25 @@ export async function runTurn(
552552
env.session.agentSessionId,
553553
env.continuityTurnIndex,
554554
);
555-
// Mirror the record durably so it survives a runner restart; fire-and-forget.
555+
// Append this turn to the durable turns log; fire-and-forget (a plain INSERT, no race).
556556
const syncCred = runCredential(request);
557-
if (syncCred) {
558-
void (deps.syncHarnessSessionDurable ?? syncHarnessSessionDurable)(
557+
if (syncCred && request.streamId) {
558+
const workflowRefs = buildWorkflowReferences(
559+
request.runContext?.workflow,
560+
);
561+
void (deps.appendSessionTurn ?? appendSessionTurn)(
559562
sessionId,
560563
plan.harness,
561-
env.session.agentSessionId,
562564
env.continuityTurnIndex,
565+
{
566+
streamId: request.streamId,
567+
agentSessionId: env.session.agentSessionId,
568+
sandboxId: env.sandbox?.sandboxId,
569+
references: workflowRefs ? Object.values(workflowRefs) : undefined,
570+
traceId: run.traceId(),
571+
},
563572
{ authorization: syncCred, log: logger },
564-
);
573+
).catch(() => {});
565574
}
566575
} else if (stopReason === "paused") {
567576
// A pause stopped mid-turn, after the harness may have written a partial turn natively.

‎services/runner/src/engines/sandbox_agent/runtime-contracts.ts‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,8 @@ import { PendingApprovalPauseController } from "./pause.ts";
1818
import { buildSandboxProvider } from "./provider.ts";
1919
import { createRunLimits, resolveRunLimits } from "./run-limits.ts";
2020
import { type BuildRunPlanDeps, type RunPlan } from "./run-plan.ts";
21-
import { clearSandboxPointer, readStoredSandboxPointer, writeSandboxPointer } from "./sandbox-reconnect.ts";
22-
import { hydrateHarnessSessionFromDurable, syncHarnessSessionDurable } from "./session-continuity-durable.ts";
21+
import { readStoredSandboxPointer } from "./sandbox-reconnect.ts";
22+
import { appendSessionTurn, hydrateHarnessSessionFromDurable } from "./session-continuity-durable.ts";
2323
import { type SessionContinuityStore } from "./session-continuity.ts";
2424
import { type TeardownReason } from "./teardown.ts";
2525
import { uploadToolMcpAssets } from "./tool-mcp-assets.ts";
@@ -57,13 +57,12 @@ export interface SandboxAgentDeps extends BuildRunPlanDeps {
5757
createRunLimits?: typeof createRunLimits;
5858
/** Session-continuity store override (tests inject their own; default is the process singleton). */
5959
sessionContinuityStore?: SessionContinuityStore;
60-
/** Durable read-back/write-forward of the continuity store (tests inject fakes). */
60+
/** Durable read-back/append-forward of the continuity store (tests inject fakes). */
6161
hydrateHarnessSessionFromDurable?: typeof hydrateHarnessSessionFromDurable;
62-
syncHarnessSessionDurable?: typeof syncHarnessSessionDurable;
63-
/** Durable read/write of the sandbox pointer, for the remote reconnect ladder. */
62+
appendSessionTurn?: typeof appendSessionTurn;
63+
/** Durable read of the sandbox pointer (the latest turn's sandbox_id), for the remote
64+
* reconnect ladder. The write side is folded into `appendSessionTurn`. */
6465
readStoredSandboxPointer?: typeof readStoredSandboxPointer;
65-
clearSandboxPointer?: typeof clearSandboxPointer;
66-
writeSandboxPointer?: typeof writeSandboxPointer;
6766
/**
6867
* Resolve `{replicaId, ownerReplicaId}` for a session-owned local-sandbox run, so
6968
* `acquireEnvironment` can fail loudly instead of silently cold-starting on a non-owner

‎services/runner/src/engines/sandbox_agent/sandbox-reconnect.ts‎

Lines changed: 15 additions & 105 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,16 @@
11
/**
2-
* Daytona sandbox reconnect: read the stored sandbox id back so a resumed session restarts the
3-
* parked (stopped/archived) sandbox instead of provisioning a fresh one, and write the live id
4-
* forward after start. Best-effort throughout: a missing/unreadable id, or a failed reconnect,
5-
* degrades to a fresh create (the dead rung), never a hard error.
2+
* Daytona sandbox reconnect: read the latest turn's stored sandbox id so a resumed session
3+
* restarts the parked (stopped/archived) sandbox instead of provisioning a fresh one.
4+
* Best-effort throughout: a missing/unreadable id, or a failed reconnect, degrades to a fresh
5+
* create (the dead rung), never a hard error.
6+
*
7+
* The live id is written forward as a field on the turn-append row (see
8+
* `session-continuity-durable.ts` `appendSessionTurn`), not through a separate pointer PUT: the
9+
* turns table is append-only, so "the latest turn's sandbox_id" IS the current pointer — a late
10+
* lower-index write can never win `ORDER BY turn_index DESC`, dissolving the old atomic
11+
* staleness guard.
612
*/
7-
import { apiBase } from "../../apiBase.ts";
13+
import { fetchLatestSessionTurn } from "./session-continuity-durable.ts";
814

915
export interface SandboxPointerDeps {
1016
apiBase?: string;
@@ -13,10 +19,6 @@ export interface SandboxPointerDeps {
1319
log?: (msg: string) => void;
1420
}
1521

16-
function defaultLog(msg: string): void {
17-
process.stderr.write(`[sandbox-reconnect] ${msg}\n`);
18-
}
19-
2022
/**
2123
* The stored sandbox instance id for this session, or undefined when none is recorded (first
2224
* turn, storage disabled, or unreachable). The id is a provider-scoped handle, so reconnect is
@@ -26,104 +28,12 @@ export interface StoredSandboxPointer {
2628
sandboxId: string;
2729
}
2830

29-
export interface SandboxPointerWrite {
30-
sandboxId: string;
31-
turnIndex: number;
32-
}
33-
34-
export type SandboxPointerWriteOutcome = "applied" | "rejected" | "failed";
35-
3631
export async function readStoredSandboxPointer(
3732
sessionId: string,
3833
deps: SandboxPointerDeps,
3934
): Promise<StoredSandboxPointer | undefined> {
40-
const log = deps.log ?? defaultLog;
41-
const doFetch = deps.fetchImpl ?? fetch;
42-
const base = deps.apiBase ?? apiBase();
43-
try {
44-
const res = await doFetch(
45-
`${base}/sessions/states/?session_id=${encodeURIComponent(sessionId)}`,
46-
{ method: "GET", headers: { authorization: deps.authorization } },
47-
);
48-
if (!res.ok) return undefined;
49-
const body = (await res.json()) as {
50-
session_state?: {
51-
sandbox_id?: string | null;
52-
} | null;
53-
};
54-
const id = body.session_state?.sandbox_id;
55-
if (typeof id !== "string" || id.length === 0) return undefined;
56-
return { sandboxId: id };
57-
} catch (err) {
58-
log(
59-
`read failed session=${sessionId}: ${String(err instanceof Error ? err.message : err).slice(0, 120)}`,
60-
);
61-
return undefined;
62-
}
63-
}
64-
65-
/**
66-
* Shared guarded pointer PUT. `write` and `clear` differ only in the sandbox_id they send and
67-
* the returned-row value that counts as applied; the transport, guard token, and error handling
68-
* must stay identical, so they live here once.
69-
*/
70-
async function putSandboxPointer(
71-
sessionId: string,
72-
sandboxId: string | null,
73-
turnIndex: number,
74-
deps: SandboxPointerDeps,
75-
logPrefix: string,
76-
): Promise<SandboxPointerWriteOutcome> {
77-
const log = deps.log ?? defaultLog;
78-
const doFetch = deps.fetchImpl ?? fetch;
79-
const base = deps.apiBase ?? apiBase();
80-
try {
81-
const res = await doFetch(
82-
`${base}/sessions/states/?session_id=${encodeURIComponent(sessionId)}`,
83-
{
84-
method: "PUT",
85-
headers: { "content-type": "application/json", authorization: deps.authorization },
86-
body: JSON.stringify({
87-
sandbox_id: sandboxId,
88-
sandbox_turn_index: turnIndex,
89-
}),
90-
},
91-
);
92-
if (!res.ok) {
93-
log(`${logPrefix} HTTP ${res.status} session=${sessionId}`);
94-
return "failed";
95-
}
96-
const body = (await res.json()) as {
97-
session_state?: { sandbox_id?: string | null } | null;
98-
};
99-
const stored = body.session_state?.sandbox_id;
100-
const applied = sandboxId === null ? stored == null : stored === sandboxId;
101-
return applied ? "applied" : "rejected";
102-
} catch (err) {
103-
log(
104-
`${logPrefix} failed session=${sessionId}: ${String(err instanceof Error ? err.message : err).slice(0, 120)}`,
105-
);
106-
return "failed";
107-
}
108-
}
109-
110-
/**
111-
* Write the live sandbox instance id forward (best-effort) so the next turn can reconnect it.
112-
* Only Daytona runs record a pointer; a local run leaves the row untouched.
113-
*/
114-
export async function writeSandboxPointer(
115-
sessionId: string,
116-
pointer: SandboxPointerWrite,
117-
deps: SandboxPointerDeps,
118-
): Promise<SandboxPointerWriteOutcome> {
119-
return putSandboxPointer(sessionId, pointer.sandboxId, pointer.turnIndex, deps, "write");
120-
}
121-
122-
/** Clear a terminal sandbox pointer under the same turn-index guard as pointer writes. */
123-
export async function clearSandboxPointer(
124-
sessionId: string,
125-
turnIndex: number,
126-
deps: SandboxPointerDeps,
127-
): Promise<SandboxPointerWriteOutcome> {
128-
return putSandboxPointer(sessionId, null, turnIndex, deps, "clear");
35+
const latest = await fetchLatestSessionTurn(sessionId, undefined, deps);
36+
const id = latest?.sandbox_id;
37+
if (typeof id !== "string" || id.length === 0) return undefined;
38+
return { sandboxId: id };
12939
}

0 commit comments

Comments
 (0)