Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
175 changes: 175 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6522,6 +6522,181 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);

// Claude Workflow members run their own commands (local_bash with
// owned_by_subagent) while the root turn is settled. Their ends must not
// release the process the workflow runs in.
it.effect.each([true, false] as const)(
"keeps a running workflow alive across member command ends (backgrounded: %s)",
(memberBackgrounded) =>
Effect.scoped(
Effect.gen(function* () {
let closes = 0;
const harness = yield* makeWakeHarnessWithOptions({
close: () =>
Effect.sync(() => {
closes += 1;
}),
});
const now = yield* DateTime.now;
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-workflow-members"),
text: "Run the batch workflow",
attachments: [],
}),
);
for (const frame of [
claudeSdkFrame({
type: "assistant",
message: {
model: "claude-sonnet-4-6",
id: "msg_workflow_launch",
type: "message",
role: "assistant",
content: [{ type: "tool_use", id: "toolu-workflow", name: "Workflow", input: {} }],
},
parent_tool_use_id: null,
uuid: "00000000-0000-4000-8000-000000009001",
session_id: WAKE_NATIVE_SESSION,
}),
claudeSdkFrame({
type: "system",
subtype: "task_started",
task_id: "workflow-1",
tool_use_id: "toolu-workflow",
task_type: "local_workflow",
workflow_name: "batch",
description: "Batch workflow",
uuid: "00000000-0000-4000-8000-000000009002",
session_id: WAKE_NATIVE_SESSION,
}),
claudeSdkFrame({
type: "user",
message: {
role: "user",
content: [
{
type: "tool_result",
tool_use_id: "toolu-workflow",
content: "Workflow launched successfully.",
},
],
},
parent_tool_use_id: null,
uuid: "00000000-0000-4000-8000-000000009003",
session_id: WAKE_NATIVE_SESSION,
tool_use_result: {
taskType: "local_workflow",
runId: "wf-run",
scriptPath: "/tmp/wf.ts",
},
}),
makeResultFrame({
uuid: "00000000-0000-4000-8000-000000009004",
result: "Workflow is running.",
}),
]) {
yield* harness.offerAndWait(frame);
}
yield* Queue.take(harness.terminalReceipts);

const memberCommand = (index: number, status: "completed" | "failed") => [
claudeSdkFrame({
type: "system",
subtype: "task_started",
task_id: `member-bash-${index}`,
tool_use_id: `toolu-member-bash-${index}`,
task_type: "local_bash",
description: `Member command ${index}`,
is_backgrounded: memberBackgrounded,
owned_by_subagent: true,
uuid: `00000000-0000-4000-8000-00000000910${index}`,
session_id: WAKE_NATIVE_SESSION,
}),
claudeSdkFrame({
type: "system",
subtype: "task_notification",
task_id: `member-bash-${index}`,
tool_use_id: `toolu-member-bash-${index}`,
status,
output_file: `/tmp/member-bash-${index}.output`,
summary: `Member command ${index}`,
uuid: `00000000-0000-4000-8000-00000000920${index}`,
session_id: WAKE_NATIVE_SESSION,
}),
];
for (const frame of [
...memberCommand(1, "failed"),
// Between stages the workflow reports progress, which the session
// manager counts as activity.
claudeSdkFrame({
type: "system",
subtype: "task_progress",
task_id: "workflow-1",
tool_use_id: "toolu-workflow",
description: "Stage two",
usage: { total_tokens: 10, tool_uses: 1, duration_ms: 100 },
uuid: "00000000-0000-4000-8000-000000009301",
session_id: WAKE_NATIVE_SESSION,
}),
...memberCommand(2, "completed"),
]) {
yield* harness.offerAndWait(frame);
}

yield* awaitUntil(
() =>
harness.events.some(
(event) =>
event.type === "subagent.updated" &&
event.subagent.nativeTaskRef?.nativeId === "workflow-1" &&
event.subagent.progress === "Stage two",
),
"workflow progress",
);
assert.isTrue(yield* harness.hasPendingBackgroundWork);
assert.equal(closes, 0);
assert.lengthOf(harness.continuationRequests, 0);
const workflow = harness.events.findLast(
(event) =>
event.type === "subagent.updated" &&
event.subagent.nativeTaskRef?.nativeId === "workflow-1",
);
assert.equal(
workflow?.type === "subagent.updated" && workflow.subagent.status,
"running",
);
// A new selection would need a new CLI process; the workflow keeps it.
const replaced = yield* Effect.exit(
harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-workflow-members-2"),
text: "Switch models",
attachments: [],
providerTurnOrdinal: 2,
modelSelection: {
instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID,
model: "claude-haiku-4-5",
},
}),
),
);
assert.isTrue(Exit.isFailure(replaced));
assert.equal(closes, 0);
}).pipe(
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
),
),
);

it.effect("routes a subagent that starts while the root turn is idle", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down
75 changes: 75 additions & 0 deletions apps/server/src/orchestration-v2/ProviderSessionManager.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2786,6 +2786,81 @@ it.effect("ProviderSessionManagerV2 releases pinned idle sessions once the pin c
}),
);

it.effect(
"ProviderSessionManagerV2 keeps pinned work that reports progress and records a pin expiry",
() =>
Effect.gen(function* () {
const state = yield* Ref.make(emptyState);
const effect = Effect.gen(function* () {
const eventSink = yield* EventSink.EventSinkV2;
const idAllocator = yield* IdAllocator.IdAllocatorV2;
const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
const now = yield* DateTime.now;
const threadId = yield* idAllocator.allocate.thread({
fixtureName: "provider-session-manager-pin-progress",
projectId: yield* idAllocator.allocate.project({
fixtureName: "provider-session-manager-pin-progress",
}),
});
const providerSessionId = yield* idAllocator.allocate.providerSession({
providerInstanceId: modelSelection.instanceId,
threadId,
});
yield* eventSink.write({
events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
});
const runtime = yield* manager.open({
threadId,
providerSessionId,
modelSelection,
runtimePolicy,
});
const adapterQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
assert.isDefined(adapterQueue);
const step = (duration: Parameters<typeof TestClock.adjust>[0]) =>
Effect.gen(function* () {
yield* TestClock.adjust(duration);
for (let i = 0; i < 20; i += 1) yield* Effect.yieldNow;
});

// A long workflow keeps reporting progress well past the pin cap.
for (let tick = 0; tick < 12; tick += 1) {
yield* step("1500 millis");
yield* Queue.offer(adapterQueue!, {
type: "provider_session.updated",
driver: CODEX_DRIVER,
providerSession: runtime.providerSession,
});
for (let i = 0; i < 20; i += 1) yield* Effect.yieldNow;
}
assert.equal((yield* Ref.get(state)).closeCount, 0);

// Work that goes silent past the cap is stopped, and says so.
for (let tick = 0; tick < 6; tick += 1) {
yield* step("1 second");
}
assert.equal((yield* Ref.get(state)).closeCount, 1);
const released = (yield* projectionStore.getThreadProjection(
threadId,
)).providerSessions.find((session) => session.id === providerSessionId);
assert.equal(released?.status, "stopped");
assert.include(released?.lastError ?? "", "no progress");
});

yield* effect.pipe(
Effect.provide(
layerTest({
state,
idleTimeoutMs: 1000,
maxIdlePinMs: 3000,
hasPendingBackgroundWork: Effect.succeed(true),
}),
),
);
}),
);

it.effect(
"ProviderSessionManagerV2 does not idle-release a session that turns busy during the pending-work check",
() =>
Expand Down
28 changes: 23 additions & 5 deletions apps/server/src/orchestration-v2/ProviderSessionManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ import * as ProjectionStore from "./ProjectionStore.ts";

const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000;
const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000;
const PIN_EXPIRED_RELEASE_DETAIL =
"Stopped background work that reported no progress for too long. Send a message to resume.";
const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000;

const busyTurnPrefix = (providerThreadId: ProviderThreadId) => `${providerThreadId}#`;
Expand Down Expand Up @@ -656,7 +658,10 @@ export const layerWithOptions = (
lastError:
input.reason === "runtime_error"
? (input.detail ?? "Provider runtime failed.")
: null,
: // Only a pin expiry gives an idle release a detail: work was cut off.
input.reason === "idle_timeout"
? (input.detail ?? null)
: null,
};
yield* writeProviderSessionEvents({
runtime: input.entry.runtime,
Expand Down Expand Up @@ -1112,6 +1117,9 @@ export const layerWithOptions = (
yield* releaseEntry({
providerSessionId: input.providerSessionId,
reason: "idle_timeout",
// Past the pin cap this stops work that is still running, so the
// session records why instead of looking like it finished.
...(hasPendingWork ? { detail: PIN_EXPIRED_RELEASE_DETAIL } : {}),
cancelIdleFiber: false,
onlyIfIdleGeneration: input.generation,
}).pipe(
Expand Down Expand Up @@ -1139,7 +1147,13 @@ export const layerWithOptions = (
),
);

const scheduleIdleReleaseInternal = (providerSessionId: ProviderSessionId) =>
// Provider activity restarts the pin budget, so background work that keeps
// reporting progress (a long Claude workflow) is never cut off; only work
// that went silent for maxIdlePinMs is released.
const scheduleIdleReleaseInternal = (
providerSessionId: ProviderSessionId,
options?: { readonly providerActivity?: boolean },
) =>
Effect.gen(function* () {
const key = sessionKey(providerSessionId);
const current = yield* Ref.get(sessions);
Expand All @@ -1166,6 +1180,7 @@ export const layerWithOptions = (
idleGeneration: generation,
idleFiber,
lastActivityAtMs,
...(options?.providerActivity === true ? { pinnedSinceMs: null } : {}),
});
return updated;
});
Expand All @@ -1174,7 +1189,10 @@ export const layerWithOptions = (
const scheduleIdleRelease = (providerSessionId: ProviderSessionId) =>
withActivityError(providerSessionId, scheduleIdleReleaseInternal(providerSessionId));

const touchActivity = (providerSessionId: ProviderSessionId) =>
const touchActivity = (
providerSessionId: ProviderSessionId,
options?: { readonly providerActivity?: boolean },
) =>
withActivityError(
providerSessionId,
Effect.gen(function* () {
Expand All @@ -1191,7 +1209,7 @@ export const layerWithOptions = (
});
return updated;
});
yield* scheduleIdleReleaseInternal(providerSessionId);
yield* scheduleIdleReleaseInternal(providerSessionId, options);
}),
);

Expand Down Expand Up @@ -1895,7 +1913,7 @@ export const layerWithOptions = (
event.providerThreadId,
event.runOrdinal,
)
: touchActivity(entry.runtime.providerSessionId),
: touchActivity(entry.runtime.providerSessionId, { providerActivity: true }),
).pipe(
Effect.andThen(
event.type === "provider_session.updated"
Expand Down
Loading