Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
16 changes: 8 additions & 8 deletions services/runner/src/engines/sandbox_agent/environment-setup.ts
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ import {
resolvesToLocalProvider,
} from "./session-identity.ts";
import { loadRunnerConfig } from "../../config/runner-config.ts";
import { createTimingLog } from "../../environment/timing.ts";

function defaultLog(message: string): void {
process.stderr.write(`[sandbox-agent] ${message}\n`);
Expand All @@ -62,14 +63,13 @@ export async function prepareEnvironmentSetup(
) {
const logger = deps.log ?? defaultLog;
const acquireStartedAt = Date.now();
const timingLog = (stage: string, startedAt: number, fields = ""): void => {
const sandboxId = environment?.sandbox?.sandboxId ?? "-";
const sessionId =
environment?.sessionId ?? request.sessionId?.trim() ?? "-";
logger(
`[timing] stage=${stage} ms=${Math.round(Date.now() - startedAt)} sandbox=${sandboxId} session=${sessionId}${fields}`,
);
};
// The stage names are matched by dashboards; see `environment/timing.ts`. The accessors are
// read at call time on purpose: the sandbox does not exist yet, and the session id changes
// during acquire, so capturing either by value would log a stale `-`.
const timingLog = createTimingLog(logger, {
sandboxId: () => environment?.sandbox?.sandboxId,
sessionId: () => environment?.sessionId ?? request.sessionId?.trim(),
});

// Local multi-runner fails loudly. Session-owned + local-sandbox only (a non-session run
// has no cross-replica identity to protect, and a remote sandbox has no runner-local pooled
Expand Down
146 changes: 55 additions & 91 deletions services/runner/src/engines/sandbox_agent/environment.ts
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,14 @@ import {
} from "./session-continuity.ts";
import { mountExpiryMs, projectScopeFor } from "./session-identity.ts";
import { teardownDisposition, type TeardownReason } from "./teardown.ts";
import {
cleanup as cleanupWorkspace,
materialize as materializeWorkspace,
} from "../../environment/workspace-manager.ts";
import {
acquire as acquireSandbox,
teardown as teardownSandbox,
} from "../../environment/sandbox-lifecycle.ts";
import { uploadToolMcpAssets, type ToolMcpAssets } from "./tool-mcp-assets.ts";
import { prepareWorkspace } from "./workspace.ts";
import { prepareEnvironmentSetup } from "./environment-setup.ts";
Expand Down Expand Up @@ -307,34 +315,16 @@ export async function acquireEnvironment(
await environment.sandbox
?.destroySession?.(environment.session.id)
.catch(() => {});
const disposition = teardownDisposition(opts?.reason ?? "failed-turn");
let parked = false;
if (
disposition === "stop" &&
plan.isDaytona &&
environment.sandbox?.pauseSandbox
) {
const sandboxLogId = environment.sandbox.sandboxId ?? plan.sandboxId;
try {
await environment.sandbox.pauseSandbox();
parked = true;
logger(`parked sandbox=${sandboxLogId}`);
} catch (err) {
logger(
`pause failed sandbox=${sandboxLogId}: ${conciseError(err, plan.harness)}`,
);
}
}
if (!parked) {
// Record the id BEFORE the delete call, and record it even when the call throws. A delete
// that failed may still have removed the sandbox, so reconnecting to it is a wasted round
// trip either way. See `markSandboxDestroyed`.
markSandboxDestroyed(
environment.sandbox?.sandboxId ?? plan.sandboxId ?? undefined,
);
await environment.sandbox?.destroySandbox().catch(() => {});
}
await environment.sandbox?.dispose().catch(() => {});
// SandboxLifecycle owns park-versus-delete. It returns `parked` because the mount teardown
// below is gated on it: a parked Daytona sandbox keeps its agent mount.
const { parked } = await teardownSandbox({
sandbox: environment.sandbox,
plannedSandboxId: plan.sandboxId,
isDaytona: plan.isDaytona,
harness: plan.harness,
reason: opts?.reason,
log: logger,
});
// Unmount the durable cwd BEFORE removing the dir: data lives in the store, only the host
// mountpoint is torn down. If unmount is not CONFIRMED gone, skip the delete: rmSync must
// never run against a possibly-live FUSE mount into the durable store.
Expand Down Expand Up @@ -365,7 +355,7 @@ export async function acquireEnvironment(
`durable cwd unmount not confirmed, skipping workspace cleanup cwd=${plan.workspace.cwd}`,
);
} else {
await environment.workspace?.cleanup().catch(() => {});
await cleanupWorkspace(environment.workspace);
}
// The per-run Agenta agent dir (skills isolation) is throwaway; remove it too. This is only
// ever a temp dir: a subscription run leaves `runAgentDir` undefined precisely so that the
Expand Down Expand Up @@ -667,52 +657,29 @@ export async function acquireEnvironment(
? (deps.createCookieFetch ?? createCookieFetch)()
: (deps.createAcpFetch ?? createAcpFetch)(),
};
// A stored sandbox id is trusted: reconnect it by id and let reconnect converge its network
// policy to this run's plan. Any reconnect failure falls through to a fresh create. Snapshot
// and image drift are accepted as per-conversation version pinning, not grounds for a rebuild.
const storedSandboxPointer =
plan.isDaytona && sessionForMount && runCred
? await (deps.readStoredSandboxPointer ?? readStoredSandboxPointer)(
sessionForMount,
{ authorization: runCred, log: logger },
)
: undefined;
if (storedSandboxPointer) {
const sandboxStartStartedAt = Date.now();
try {
environment.sandbox = await startSandboxAgent({
...startOptions,
sandboxId: storedSandboxPointer.sandboxId,
});
logger(
`reconnected sandbox=${storedSandboxPointer.sandboxId} session=${sessionForMount}`,
);
} catch (err) {
logger(
`reconnect failed sandbox=${storedSandboxPointer.sandboxId}, creating fresh: ${conciseError(err, plan.harness)}`,
);
// 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 {
timingLog("sandbox_start", sandboxStartStartedAt, " mode=reconnect");
}
}
if (!environment.sandbox) {
const sandboxStartStartedAt = Date.now();
try {
environment.sandbox = await startSandboxAgent(startOptions);
} finally {
timingLog("sandbox_start", sandboxStartStartedAt, " mode=create");
}
}
environment.resumable = Boolean(plan.isDaytona && sessionForMount);
// SandboxLifecycle owns the reconnect ladder, the fresh-create fallback, and both
// `sandbox_start` timing marks. See `environment/sandbox-lifecycle.ts`.
const acquiredSandbox = await acquireSandbox(
{
startOptions,
isDaytona: plan.isDaytona,
harness: plan.harness,
sessionForMount,
runCred,
log: logger,
timingLog,
},
{
startSandboxAgent: startSandboxAgent as unknown as (
options: Record<string, unknown>,
) => Promise<unknown>,
...(deps.readStoredSandboxPointer
? { readStoredSandboxPointer: deps.readStoredSandboxPointer }
: {}),
},
);
environment.sandbox = acquiredSandbox.sandbox;
environment.resumable = acquiredSandbox.resumable;
// Track the live handle so a shutdown signal handler can delete it if `destroy` is skipped by
// a process KILL; removed in `destroy` on every normal exit so it is never double-deleted.
if (environment.sandbox) inFlightSandboxes.add(environment);
Expand Down Expand Up @@ -880,15 +847,16 @@ export async function acquireEnvironment(
}

const prepareWorkspaceStartedAt = Date.now();
// WorkspaceManager owns the write; the retry stays here because it re-signs a MOUNT, which is
// the mount unit's concern, not the workspace's.
const workspaceInput = {
sandbox: environment.sandbox,
plan,
piSkillSnapshot,
log: logger,
};
try {
environment.workspace = await (deps.prepareWorkspace ?? prepareWorkspace)(
{
sandbox: environment.sandbox,
plan,
piSkillSnapshot,
log: logger,
},
);
environment.workspace = await materializeWorkspace(workspaceInput, deps);
} catch (err) {
if (
!plan.isDaytona &&
Expand All @@ -899,14 +867,10 @@ export async function acquireEnvironment(
logger(
`retrying workspace preparation after local durable cwd remount`,
);
environment.workspace = await (
deps.prepareWorkspace ?? prepareWorkspace
)({
sandbox: environment.sandbox,
plan,
piSkillSnapshot,
log: logger,
});
environment.workspace = await materializeWorkspace(
workspaceInput,
deps,
);
} else {
throw err;
}
Expand Down
180 changes: 180 additions & 0 deletions services/runner/src/environment/sandbox-lifecycle.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,180 @@
/**
* `SandboxLifecycle` — the provider instance.
*
* LIFECYCLE MIGRATION, STEP 5. This unit owns the `sandbox_start` acquire stage and the sandbox
* half of teardown. It is a pure code move: the reconnect ladder, the fresh-create fallback, the
* park-versus-delete decision, and the in-flight registry all behave exactly as they did inline.
*
* TWO EVENTS, ONE STAGE NAME. `acquire` may reconnect a parked sandbox or create a fresh one. Both
* emit `sandbox_start`, and the mode rides the ` mode=...` field. A dashboard grouping by stage
* therefore sees one series with a mode dimension, which is what the existing queries expect.
*
* THE RECONNECT LADDER NEVER FAILS A TURN. A stored id that will not reconnect degrades to a fresh
* create. That is why the reconnect `catch` swallows: a dead sandbox is an ordinary outcome, not
* an error, and the only cost is the round trip.
*/
import { conciseError } from "../engines/sandbox_agent/errors.ts";
import { DaytonaReconnectTerminalError } from "../engines/sandbox_agent/daytona-provider.ts";
import {
markSandboxDestroyed,
readStoredSandboxPointer,
} from "../engines/sandbox_agent/sandbox-reconnect.ts";
import {
teardownDisposition,
type TeardownReason,
} from "../engines/sandbox_agent/teardown.ts";
import type { Log, TimingLog } from "./timing.ts";

/** What `acquire` needs. Deliberately narrow: this unit never sees credentials or a workspace. */
export interface SandboxAcquireInput {
/** Provider-agnostic start options, already built by the composer. */
startOptions: Record<string, unknown>;
isDaytona: boolean;
harness: string;
/** The session whose stored pointer may name a parked sandbox. Undefined disables reconnect. */
sessionForMount: string | undefined;
/** The run credential the pointer read needs. Undefined disables reconnect. */
runCred: string | undefined;
log: Log;
timingLog: TimingLog;
}

export interface SandboxAcquireDeps {
startSandboxAgent: (options: Record<string, unknown>) => Promise<unknown>;
readStoredSandboxPointer?: typeof readStoredSandboxPointer;
}

export interface SandboxAcquireResult {
sandbox: unknown;
/** True when this sandbox may be parked and reconnected on a later turn. */
resumable: boolean;
/** Which path produced the handle. Reported for the composer's logs and for tests. */
mode: "reconnect" | "create";
}

/**
* Get a sandbox: reconnect a parked one when a pointer names it, otherwise create a fresh one.
*
* Byte-for-byte the inline behavior, including the swallowed reconnect failure and the extra log
* line for a confirmed terminal Daytona state.
*/
export async function acquire(

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.

This function is a code move. The reconnect ladder, the fresh create fallback, and the extra log line for a terminal Daytona state all behave exactly as they did inline.

The swallowed reconnect failure is deliberate and predates this PR. A stored id that will not reconnect degrades to a fresh create. A dead sandbox is an ordinary outcome, not an error, and the only cost is one round trip. If that catch stopped swallowing, a stale pointer would fail a turn that used to succeed.

The helper takes an explicit input object rather than reading shared state. That is what lets the unit be tested without an environment.

input: SandboxAcquireInput,
deps: SandboxAcquireDeps,
): Promise<SandboxAcquireResult> {
const { isDaytona, sessionForMount, runCred, log, timingLog } = input;

// A stored sandbox id is trusted: reconnect it by id and let reconnect converge its network
// policy to this run's plan. Any reconnect failure falls through to a fresh create. Snapshot
// and image drift are accepted as per-conversation version pinning, not grounds for a rebuild.
const storedSandboxPointer =
isDaytona && sessionForMount && runCred
? await (deps.readStoredSandboxPointer ?? readStoredSandboxPointer)(
sessionForMount,
{ authorization: runCred, log },
)
: undefined;

let sandbox: unknown;
let mode: "reconnect" | "create" = "create";

if (storedSandboxPointer) {
const sandboxStartStartedAt = Date.now();
try {
sandbox = await deps.startSandboxAgent({
...input.startOptions,
sandboxId: storedSandboxPointer.sandboxId,
});
mode = "reconnect";
log(
`reconnected sandbox=${storedSandboxPointer.sandboxId} session=${sessionForMount}`,
);
} catch (err) {
log(
`reconnect failed sandbox=${storedSandboxPointer.sandboxId}, creating fresh: ${conciseError(err, input.harness)}`,
);
// 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.
if (err instanceof DaytonaReconnectTerminalError) {
log(
`terminal Daytona state '${err.state}' for sandbox=${storedSandboxPointer.sandboxId}, not retrying reconnect`,
);
}
} finally {
timingLog("sandbox_start", sandboxStartStartedAt, " mode=reconnect");
}
}

if (!sandbox) {
const sandboxStartStartedAt = Date.now();
mode = "create";
try {
sandbox = await deps.startSandboxAgent(input.startOptions);
} finally {
timingLog("sandbox_start", sandboxStartStartedAt, " mode=create");
}
}

return {
sandbox,
resumable: Boolean(isDaytona && sessionForMount),
mode,
};
}

export interface SandboxTeardownInput {
sandbox: {
sandboxId?: string;
pauseSandbox?: () => Promise<unknown>;
destroySandbox?: () => Promise<unknown>;
dispose?: () => Promise<unknown>;
} | undefined;
/** The plan's id, used when the live handle carries none. */
plannedSandboxId: string | undefined;
isDaytona: boolean;
harness: string;
reason: TeardownReason | undefined;
log: Log;
}

/**
* Stop or delete the sandbox, and say which happened.
*
* `parked` is returned because the caller needs it: a parked Daytona sandbox keeps its agent
* mount, so the mount unit's teardown is gated on this answer. That coupling is why the composer
* still sequences the units rather than each unit tearing itself down independently.
*
* Never throws. Teardown must always complete.
*/
export async function teardown(
input: SandboxTeardownInput,
): Promise<{ parked: boolean }> {
const { sandbox, log } = input;
const disposition = teardownDisposition(input.reason ?? "failed-turn");
let parked = false;

if (disposition === "stop" && input.isDaytona && sandbox?.pauseSandbox) {
const sandboxLogId = sandbox.sandboxId ?? input.plannedSandboxId;
try {
await sandbox.pauseSandbox();
parked = true;
log(`parked sandbox=${sandboxLogId}`);
} catch (err) {
log(
`pause failed sandbox=${sandboxLogId}: ${conciseError(err, input.harness)}`,
);
}
}

if (!parked) {
// Record the id BEFORE the delete call, and record it even when the call throws. A delete
// that failed may still have removed the sandbox, so reconnecting to it is a wasted round
// trip either way. See `markSandboxDestroyed`.
markSandboxDestroyed(sandbox?.sandboxId ?? input.plannedSandboxId ?? undefined);
await sandbox?.destroySandbox?.().catch(() => {});
}
await sandbox?.dispose?.().catch(() => {});

return { parked };
}
Loading
Loading