Skip to content

Commit f467b00

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 fa96ba7 commit f467b00

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
@@ -99,11 +99,7 @@ import {
9999
routeSessionEventToActiveTurn,
100100
} from "./session-events.ts";
101101
import { buildSandboxProvider } from "./provider.ts";
102-
import {
103-
clearSandboxPointer,
104-
readStoredSandboxPointer,
105-
writeSandboxPointer,
106-
} from "./sandbox-reconnect.ts";
102+
import { readStoredSandboxPointer } from "./sandbox-reconnect.ts";
107103
import type {
108104
AcquireEnvironmentResult,
109105
SandboxAgentDeps,
@@ -642,31 +638,13 @@ export async function acquireEnvironment(
642638
logger(
643639
`reconnect failed sandbox=${storedSandboxPointer.sandboxId}, creating fresh: ${conciseError(err, plan.harness)}`,
644640
);
645-
if (
646-
err instanceof DaytonaReconnectTerminalError &&
647-
sessionForMount &&
648-
runCred
649-
) {
650-
// The post-hydrate write later in acquire is authoritative. This clear only prevents
651-
// repeated doomed reconnects if acquire fails before reaching that write. Hydrate
652-
// first: after a runner restart the in-memory store is behind the durable
653-
// latest_turn_index, and an unhydrated guard token would be rejected as stale.
654-
await (
655-
deps.hydrateHarnessSessionFromDurable ??
656-
hydrateHarnessSessionFromDurable
657-
)(
658-
sessionForMount,
659-
plan.harness,
660-
deps.sessionContinuityStore ?? sessionContinuityStore,
661-
{ authorization: runCred, log: logger },
662-
);
663-
await (deps.clearSandboxPointer ?? clearSandboxPointer)(
664-
sessionForMount,
665-
nextTurnIndex(
666-
sessionForMount,
667-
deps.sessionContinuityStore ?? sessionContinuityStore,
668-
),
669-
{ authorization: runCred, log: logger },
641+
// No explicit pointer clear needed: turns are append-only, so the fresh sandbox this
642+
// turn creates below gets its own turn row at completion, and that row's higher
643+
// turn_index naturally supersedes the dead one on the next `latest_turn` read — the
644+
// staleness guard the old states model needed dissolves with the ordering.
645+
if (err instanceof DaytonaReconnectTerminalError) {
646+
logger(
647+
`terminal Daytona state '${err.state}' for sandbox=${storedSandboxPointer.sandboxId}, not retrying reconnect`,
670648
);
671649
}
672650
} finally {
@@ -984,24 +962,9 @@ export async function acquireEnvironment(
984962
environment.continuityTurnIndex = continuitySessionKey
985963
? nextTurnIndex(continuitySessionKey, continuityStore)
986964
: undefined;
987-
// Daytona only: a local run must not overwrite a conversation's remote pointer (switching
988-
// sandboxes mid-conversation would strand the parked Daytona instance).
989-
if (plan.isDaytona && sessionForMount && runCred) {
990-
const liveSandboxId = environment.sandbox?.sandboxId ?? plan.sandboxId;
991-
const pointerWriteOutcome = await (
992-
deps.writeSandboxPointer ?? writeSandboxPointer
993-
)(
994-
sessionForMount,
995-
{
996-
sandboxId: liveSandboxId,
997-
turnIndex: environment.continuityTurnIndex ?? 0,
998-
},
999-
{ authorization: runCred, log: logger },
1000-
);
1001-
logger(
1002-
`sandbox pointer write ${pointerWriteOutcome} session=${sessionForMount} sandbox=${liveSandboxId}`,
1003-
);
1004-
}
965+
// The live sandbox id rides forward as a field on the turn-append row written at turn end
966+
// (see `appendSessionTurn` call in `runTurn`), not a separate pre-turn pointer PUT: the
967+
// turns table is append-only, so there is nothing to overwrite mid-conversation.
1005968
let loadedFromContinuity = false;
1006969
if (priorAgentSessionId && localSessionId) {
1007970
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
@@ -64,7 +64,7 @@ import {
6464
shouldSuppressPausedToolCallUpdate,
6565
} from "./runtime-policy.ts";
6666
import {
67-
syncHarnessSessionDurable,
67+
appendSessionTurn,
6868
} from "./session-continuity-durable.ts";
6969
import { sessionContinuityStore } from "./session-continuity.ts";
7070
import { priorMessages } from "./transcript.ts";
@@ -567,16 +567,25 @@ export async function runTurn(
567567
env.session.agentSessionId,
568568
env.continuityTurnIndex,
569569
);
570-
// Mirror the record durably so it survives a runner restart; fire-and-forget.
570+
// Append this turn to the durable turns log; fire-and-forget (a plain INSERT, no race).
571571
const syncCred = runCredential(request);
572-
if (syncCred) {
573-
void (deps.syncHarnessSessionDurable ?? syncHarnessSessionDurable)(
572+
if (syncCred && request.streamId) {
573+
const workflowRefs = buildWorkflowReferences(
574+
request.runContext?.workflow,
575+
);
576+
void (deps.appendSessionTurn ?? appendSessionTurn)(
574577
sessionId,
575578
plan.harness,
576-
env.session.agentSessionId,
577579
env.continuityTurnIndex,
580+
{
581+
streamId: request.streamId,
582+
agentSessionId: env.session.agentSessionId,
583+
sandboxId: env.sandbox?.sandboxId,
584+
references: workflowRefs ? Object.values(workflowRefs) : undefined,
585+
traceId: run.traceId(),
586+
},
578587
{ authorization: syncCred, log: logger },
579-
);
588+
).catch(() => {});
580589
}
581590
} else if (stopReason === "paused") {
582591
// 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)