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
Original file line number Diff line number Diff line change
Expand Up @@ -197,12 +197,16 @@ describe("ProviderCommandReactor", () => {
model: "gpt-5-codex",
};
const startSessionEffect = input?.startSessionEffect;
// Generation-suffixed names: a re-provision after a teardown creates a NEW
// container, so a test asserting the fresh name can prove a stale cached
// target (naming the destroyed generation) was not served.
let provisionGeneration = 0;
const provisionSandbox = vi.fn(
(request: Parameters<SandboxRuntimeManagerShape["provision"]>[0]) =>
Effect.succeed({
sandboxId: request.bootstrap.threadId,
runtime: "docker" as const,
containerName: `sandbox-${request.bootstrap.threadId}`,
containerName: `sandbox-${request.bootstrap.threadId}-gen${++provisionGeneration}`,
networkName: `network-${request.bootstrap.threadId}`,
workspaceVolumeName: `workspace-${request.bootstrap.threadId}`,
desktopVolumeName: `desktop-${request.bootstrap.threadId}`,
Expand All @@ -213,7 +217,7 @@ describe("ProviderCommandReactor", () => {
services: [],
}),
);
const startSession = vi.fn((_: unknown, input: unknown) => {
const startSession = vi.fn((_: unknown, input: unknown, _executionTarget?: unknown) => {
const sessionIndex = nextSessionIndex++;
const resumeCursor =
typeof input === "object" && input !== null && "resumeCursor" in input
Expand Down Expand Up @@ -890,6 +894,87 @@ describe("ProviderCommandReactor", () => {
);
});

it("never serves a cached execution target that names a torn-down container", async () => {
// The per-thread target cache is only invalidated by validating against
// the current sandbox projection at point of use. Without that, the
// second turn below gets a cache hit for the gen1 container the stop
// already destroyed: no re-provision, and the provider session is started
// against a container name that no longer resolves.
const harness = await createHarness();
const now = "2026-01-01T00:00:00.000Z";
const threadId = ThreadId.make("thread-1");

await harness.runEffect(
harness.engine.dispatch({
type: "thread.turn.start",
commandId: CommandId.make("cmd-turn-start-before-stale-cache"),
threadId,
message: {
messageId: asMessageId("user-message-before-stale-cache"),
role: "user",
text: "hello",
attachments: [],
},
interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
runtimeMode: "approval-required",
createdAt: now,
}),
);
await waitFor(() => harness.startSession.mock.calls.length === 1);
expect(harness.startSession.mock.calls[0]?.[2]).toMatchObject({
kind: "sandbox",
runtimeRef: "sandbox-thread-1-gen1",
});

await harness.runEffect(
harness.engine.dispatch({
type: "sandbox.stop",
commandId: CommandId.make("cmd-sandbox-stop-stale-cache"),
threadId,
createdAt: now,
}),
);
await harness.runEffect(
harness.engine.dispatch({
type: "sandbox.stop.complete",
commandId: CommandId.make("cmd-sandbox-stopped-stale-cache"),
threadId,
expired: false,
createdAt: now,
}),
);
// The real teardown kills the in-container provider process with the
// container; mirror that so the next turn must start a fresh session and
// hand its execution target to the adapter, where it can be observed.
harness.runtimeSessions.length = 0;

await harness.runEffect(
harness.engine.dispatch({
type: "thread.turn.start",
commandId: CommandId.make("cmd-turn-start-after-stale-cache"),
threadId,
message: {
messageId: asMessageId("user-message-after-stale-cache"),
role: "user",
text: "back again",
attachments: [],
},
interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
runtimeMode: "approval-required",
createdAt: now,
}),
);

// A stale cache hit would skip the re-provision entirely (one provision
// call) and hand the dead gen1 name to the new session.
await waitFor(() => harness.startSession.mock.calls.length === 2);
expect(harness.provisionSandbox).toHaveBeenCalledTimes(2);
expect(harness.startSession.mock.calls[1]?.[2]).toMatchObject({
kind: "sandbox",
runtimeRef: "sandbox-thread-1-gen2",
});
});

it("keeps the resume cursor of a legacy-host thread", async () => {
// Same code path -- a host thread has no sandbox, so it takes the same
// branch -- but its conversation store is on the host and outlives
Expand Down
46 changes: 37 additions & 9 deletions apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -359,6 +359,14 @@ export const make = Effect.gen(function* () {
const sandboxRuntimeManager = yield* SandboxRuntimeManager;
const projectFileLoader = yield* T3ProjectFileLoader;
const providerSessionDirectory = yield* ProviderSessionDirectory;
// Both maps hold one tiny per-thread entry (a mutex; a target descriptor)
// and are bounded by the number of sandbox threads this server generation
// ever provisioned, so neither needs an eviction scheme for memory. What the
// target cache DOES need is staleness protection: a stop, an expiry, or a
// failure destroys the container an entry names without touching the map, so
// every read is validated against the current sandbox projection before it
// is served (see the cache hit in `ensureExecutionTarget`). The locks carry
// no container state at all and never go stale.
const sandboxProvisionLocks = new Map<string, Semaphore.Semaphore>();
const provisionedTargets = new Map<
string,
Expand Down Expand Up @@ -408,23 +416,43 @@ export const make = Effect.gen(function* () {
) {
return yield* threadSandboxRuntime.ensureReady(thread, legacyCwd);
}
// Anything cached below describes a container that a stop, an expiry, or
// a failed provision already destroyed. Handing it back would point the
// provider at a container id that no longer resolves, so drop it and let
// this call provision a fresh one.
provisionedTargets.delete(thread.id);
// The lock itself is deliberately kept: it is a per-thread mutex, not
// state about a container, and replacing one another fiber currently
// holds would let two provisions run at once.
// Cache invalidation happens under the lock below, against the CURRENT
// projection, not here against the caller's snapshot: this snapshot was
// read before any lock wait, and evicting on it can throw away an entry
// another fiber just provisioned. The lock itself is deliberately kept
// across provisions: it is a per-thread mutex, not state about a
// container, and replacing one another fiber currently holds would let
// two provisions run at once.
let lock = sandboxProvisionLocks.get(thread.id);
if (lock === undefined) {
lock = yield* Semaphore.make(1);
sandboxProvisionLocks.set(thread.id, lock);
}
return yield* lock.withPermits(1)(
Effect.gen(function* () {
// A hit only proves some fiber provisioned this thread once, not
// that the container still exists: the entry survives every stop,
// expiry, and failure, and this reactor never observes sandbox
// lifecycle events to evict it. Point-of-use validation against the
// CURRENT projection (not the caller's snapshot, which is from
// before the lock wait) is what keeps a dead container's name from
// being handed to the provider. Valid means the projection still
// names this exact container and calls it ready -- anything else
// evicts the entry and re-provisions below.
const cached = provisionedTargets.get(thread.id);
if (cached !== undefined) return cached;
if (cached !== undefined) {
const currentSandbox = Option.getOrUndefined(
yield* projectionSnapshotQuery.getThreadDetailById(thread.id),
)?.sandbox;
if (
currentSandbox != null &&
currentSandbox.lifecycle === "ready" &&
currentSandbox.sandboxId === cached.sandboxId &&
currentSandbox.runtimeRef === cached.runtimeRef
)
return cached;
provisionedTargets.delete(thread.id);
}
const project = yield* resolveProject(thread.projectId);
if (project === undefined) {
return yield* new ProviderAdapterRequestError({
Expand Down
Loading
Loading