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
12 changes: 10 additions & 2 deletions apps/server/integration/transferBudgetV2.integration.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
// @effect-diagnostics nodeBuiltinImport:off - Measures real local HTTP/WebSocket sockets.
import * as NodeHttp from "node:http";
import * as NodeUtil from "node:util";
import { assert, it } from "@effect/vitest";
import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
Expand Down Expand Up @@ -299,9 +300,16 @@ it.live(
const events = turnEvents(provider, TRANSFER_HISTORY_TURN_COUNT, true);
const stored = yield* sink.write({ events });
const last = stored.at(-1)!;
const terminal = (item: StreamItem) =>
const threadTerminal = (item: StreamItem) =>
"sequence" in item && item.sequence >= last.sequence;
for (const queue of [tq, sq, ttq, ssq]) yield* collectUntil(queue, terminal);
// Shell items carry the shell read when their batch is projected, and unchanged
// shells are skipped, so a turn split across batches may never yield an item at
// the last sequence. Wait for the final shell instead.
const finalShell = yield* projection.getThreadShell(THREAD_ID);
const shellTerminal = (item: StreamItem) =>
item.kind === "thread.updated" && NodeUtil.isDeepStrictEqual(item.thread, finalShell);
for (const queue of [tq, ttq]) yield* collectUntil(queue, threadTerminal);
for (const queue of [sq, ssq]) yield* collectUntil(queue, shellTerminal);
const measuredTurnWebSocket = difference(thread.recorder.totals(), threadBefore);
const measuredTurnShellWebSocket = difference(shell.recorder.totals(), shellBefore);
const measuredTurnSecondClientWebSocket = difference(
Expand Down
17 changes: 12 additions & 5 deletions apps/server/scripts/acp-mock-agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1383,26 +1383,32 @@ const program = Effect.gen(function* () {
(emitRunningCommandThenHangOnFirstPrompt && promptCount === 1)
) {
const toolCallId = "tool-call-running-1";
let runningCommandPublished: Promise<void> = Promise.resolve();
if (runningCommandPidPath !== undefined) {
const command = runningCommandIgnoresTerm
? 'trap "" TERM; bash -c \'trap "" TERM; while :; do sleep 1; done\' & child=$!; printf "%s %s\n" "$$" "$child" > "$1"; wait "$child"'
: 'sleep 120 & child=$!; printf "%s %s\n" "$$" "$child" > "$1"; wait "$child"';
if (runningCommandSeparateSession) {
const launcher = [
'const { spawn } = require("node:child_process");',
"const child = spawn(process.argv[1], process.argv.slice(2), { stdio: 'ignore' });",
"const child = spawn(process.argv[1], process.argv.slice(2), { stdio: ['ignore', 'inherit', 'ignore'] });",
"child.once('exit', (code, signal) => process.exitCode = code ?? (signal ? 1 : 0));",
].join(" ");
const detachedCommand = runningCommandIgnoresTerm
? 'trap "" TERM; bash -c \'trap "" TERM; while :; do sleep 1; done\' & child=$!; printf "%s %s %s\n" "$PPID" "$$" "$child" > "$1"; wait "$child"'
: 'sleep 120 & child=$!; printf "%s %s %s\n" "$PPID" "$$" "$child" > "$1"; wait "$child"';
? 'trap "" TERM; bash -c \'trap "" TERM; while :; do sleep 1; done\' & child=$!; printf "%s %s %s\n" "$PPID" "$$" "$child" > "$1"; echo published; wait "$child"'
: 'sleep 120 & child=$!; printf "%s %s %s\n" "$PPID" "$$" "$child" > "$1"; echo published; wait "$child"';
// Nested bash publishes "$PPID $$ $child" once it starts. Do not
// write the launcher PID alone here: that races with bash and can
// clobber the triple that interrupt tests wait for.
// clobber the triple that interrupt tests wait for. It then prints a
// line, so an agent that exits after launch only exits once the
// triple exists instead of racing a slow launcher start.
const detachedLauncher = NodeChildProcess.spawn(
process.execPath,
["-e", launcher, "bash", "-c", detachedCommand, "bash", runningCommandPidPath],
{ detached: true, stdio: "ignore" },
{ detached: true, stdio: ["ignore", "pipe", "ignore"] },
);
runningCommandPublished = new Promise((resolve) =>
detachedLauncher.stdout.once("data", () => resolve()),
);
detachedLauncher.unref();
} else {
Expand Down Expand Up @@ -1460,6 +1466,7 @@ const program = Effect.gen(function* () {
},
});
if (exitAfterRunningCommandLaunch) {
yield* Effect.promise(() => runningCommandPublished);
yield* Effect.sleep("100 millis");
return yield* Effect.sync(() => process.exit(0));
}
Expand Down
37 changes: 15 additions & 22 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17499,11 +17499,13 @@ describe("ClaudeAdapterV2 background wake turns", () => {
const NESTED_TOOL_USE_ID = "toolu-steer-nested";
const FINISHED_TOOL_USE_ID = "toolu-steer-finished";
const calls: Array<string> = [];
const backgroundRequested = yield* Deferred.make<void>();
let beforeAnswer: Effect.Effect<void> = Effect.void;
const lateAnswer = yield* Deferred.make<void>();
const harness = yield* makeWakeHarnessWithOptions({
backgroundTasks: (toolUseId) =>
Effect.sync(() => calls.push(toolUseId)).pipe(
Effect.andThen(Deferred.succeed(backgroundRequested, undefined)),
Effect.andThen(
backgrounding === "answers after the turn ended" ||
backgrounding === "answers after the subagent ran again"
Expand Down Expand Up @@ -17671,29 +17673,19 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);
}
// The steer arms its 5-second backgrounding timeout after the request
// starts on another fiber, so one TestClock.adjust can land before that
// timer exists and leave the steer waiting forever. Tick until it ends.
const awaitSteerTimeout = <A, E>(fiber: Fiber.Fiber<A, E>) =>
Effect.gen(function* () {
for (let tick = 0; tick < 100 && fiber.pollUnsafe() === undefined; tick++) {
yield* TestClock.adjust("1 second");
yield* Effect.yieldNow;
}
if (fiber.pollUnsafe() === undefined) {
return yield* Effect.die("Timed out waiting for the steer.");
}
return yield* Fiber.await(fiber);
});
const steering = yield* Effect.forkScoped(steer);
const timesOut =
// Started at once so the steer reads its backgrounding deadline before
// this test moves the clock. Its message is built with real file I/O,
// so a scheduled start could read the clock after the moves below.
const steering = yield* Effect.forkScoped(steer, { startImmediately: true });
if (
backgrounding === "never answers" ||
backgrounding === "answers after the steer went out";
if (timesOut) {
yield* awaitUntil(() => calls.length === 1, "background request");
backgrounding === "answers after the steer went out"
) {
yield* Deferred.await(backgroundRequested);
assert.equal(harness.offeredMessages.length, offered);
yield* TestClock.adjust("5 seconds");
}
const steered = yield* timesOut ? awaitSteerTimeout(steering) : Fiber.await(steering);
const steered = yield* Fiber.await(steering);

assert.deepEqual(calls, [FOREGROUND_TOOL_USE_ID]);
if (backgrounding === "answers after the turn ended") {
Expand All @@ -17706,8 +17698,9 @@ describe("ClaudeAdapterV2 background wake turns", () => {
assert.equal(harness.offeredMessages.at(-1)?.priority, "now");
if (backgrounding === "never answers") {
for (let repeat = 1; repeat <= 2; repeat++) {
const again = yield* Effect.forkScoped(steer);
assert.isTrue(Exit.isSuccess(yield* awaitSteerTimeout(again)));
const again = yield* Effect.forkScoped(steer, { startImmediately: true });
yield* TestClock.adjust("5 seconds");
assert.isTrue(Exit.isSuccess(yield* Fiber.await(again)));
assert.equal(harness.offeredMessages.length, offered + 1 + repeat);
}
assert.deepEqual(calls, [FOREGROUND_TOOL_USE_ID]);
Expand Down
15 changes: 13 additions & 2 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,7 @@ import {
} from "@t3tools/contracts";

import * as Cause from "effect/Cause";
import * as Clock from "effect/Clock";
import * as Context from "effect/Context";
import * as Crypto from "effect/Crypto";
import * as DateTime from "effect/DateTime";
Expand Down Expand Up @@ -304,6 +305,9 @@ const CLAUDE_WAKE_TURN_START_DEADLINE = "2 minutes";
// reply can land after the progress frame that settles it. Its transcript
// stays the source of its answer this long after it settles.
const WORKFLOW_MEMBER_TRANSCRIPT_FLUSH_MS = 500;
// A steer waits at most this long, from its request, for its foreground
// subagents to move to the background before it is sent.
const BACKGROUND_MOVE_TIMEOUT_MS = 5_000;

/**
* When the CLI exits it stops the MCP servers it started, but not their children: a plugin's
Expand Down Expand Up @@ -9684,6 +9688,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
// its result arrives as a task notification like any background agent.
const backgroundForegroundSubagents = Effect.fnUntraced(function* (
live: ClaudeLiveQueryContext,
deadline: number,
) {
const foreground: Array<{
readonly taskId: string;
Expand Down Expand Up @@ -9739,11 +9744,17 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
backgroundMovesByToolUseId.set(toolUseId, move);
moves.push(move);
}
yield* Fiber.joinAll(moves).pipe(Effect.timeoutOption("5 seconds"));
const remaining = deadline - (yield* Clock.currentTimeMillis);
yield* Fiber.joinAll(moves).pipe(Effect.timeoutOption(Math.max(0, remaining)));
});

const steerTurn = Effect.fn("ClaudeAdapterV2.steerTurn")(
function* (turnInput: ProviderAdapter.ProviderAdapterV2SteerInput) {
// Measured from the request, not from when the moves are forked:
// building the message and the moves can take clock time a timer
// started later would not count, and the steer would wait past it.
const backgroundDeadline =
(yield* Clock.currentTimeMillis) + BACKGROUND_MOVE_TIMEOUT_MS;
const existing = yield* Ref.get(queryContext);
if (existing === null) {
return yield* new ProviderAdapter.ProviderAdapterProtocolError({
Expand All @@ -9769,7 +9780,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
fileSystem,
skills: yield* userInvocableSkills(currentTurn.input.runtimePolicy.cwd),
});
yield* backgroundForegroundSubagents(existing);
yield* backgroundForegroundSubagents(existing, backgroundDeadline);
// The turn can end while the moves are answered. A steer offered
// then would start a new turn the orchestrator never asked for, so
// the steer fails and the orchestrator starts the follow-up turn.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -77,9 +77,7 @@ it("keeps only the usage limit attached while an active goal waits in the collap
const limit = page.getByText("Usage limit reached", { exact: true });
const goal = page.getByText("Goal active", { exact: true });

// A retained pointer can expand the stack as it mounts; start outside the hover target.
await userEvent.hover(limit);
await expect.poll(visibleBanners).toHaveLength(1);
expect(visibleBanners()).toHaveLength(1);
await expect.element(limit).toBeVisible();
await expect.element(goal).not.toBeVisible();
expect(container.querySelector('[data-slot="composer-banner-attachment"]')?.textContent).toBe(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,15 @@ vi.mock(import("~/state/query"), async (original) => ({
})) as never,
}));

// The first Mermaid render loads the library and its flowchart chunks, which took over
// 10 seconds on a busy release runner. Pay that while the file loads, not inside a test.
{
const { default: mermaid } = await import("mermaid");
mermaid.initialize({ startOnLoad: false });
await mermaid.render("mermaid-warmup", "flowchart LR\n A --> B");
document.getElementById("dmermaid-warmup")?.remove();
}

const ENVIRONMENT_ID = EnvironmentId.make("environment-local");
const CREATED_AT = "2026-03-17T19:12:28.000Z";

Expand Down
Loading
Loading