diff --git a/docs/design/agent-workflows/documentation/tools.md b/docs/design/agent-workflows/documentation/tools.md index bda719864b..8f27b61454 100644 --- a/docs/design/agent-workflows/documentation/tools.md +++ b/docs/design/agent-workflows/documentation/tools.md @@ -243,14 +243,50 @@ never leave the service — the same safety property gateway tools have. The wor relays a `callback` spec exactly as it does a gateway tool (direct or via the Daytona file relay). There is one transport wrinkle. On Daytona the in-sandbox process cannot reach Agenta over the -network. So the call is relayed through files instead: the in-sandbox tool writes a request -file to a relay directory, the runner (which can reach Agenta) reads it, performs the same -`/tools/call` POST, and writes the answer back (`relayToolCall` in `dispatch.ts`, -`startToolRelay` in `tools/relay.ts`). Same callback, same envelope, different delivery. The +network. So the call is relayed through files instead: the in-sandbox tool publishes a request +file into a relay directory, the runner (which can reach Agenta) picks it up, performs the same +`/tools/call` POST, and publishes the answer back. The writer is `relayToolCall` in +`tools/relay-client.ts` (re-exported by `dispatch.ts`); the wire protocol (file suffixes, +request and response shapes, byte-pinned request serialization) is `tools/relay-protocol.ts`; +the runner-side loop is `startToolRelay` in `tools/relay.ts`. The two writer modules import +node builtins only, so the Pi extension bundle and the future in-sandbox MCP shim (#5234) +consume them from inside the sandbox. Same callback, same envelope, different delivery. The non-Pi internal MCP channel (a loopback HTTP MCP server the runner serves) uses this same relay even on local runs, because the harness calling it is kept blind to the private spec — only public metadata crosses the channel, and execution relays back to the runner. +The relay files carry three guarantees. Publication is atomic in both directions: each side +writes the full bytes to a temp name (`.tmp.`) and renames it to the final name +in the same directory, so a reader never sees partial JSON (on Daytona the daemon's `moveFs` +is `rename(2)` underneath). The runner deletes each request file right after reading it, so a +request executes at most once per publication; a request lost to a runner crash surfaces as a +writer timeout and a tool error, never a redelivery. And each turn's relay loop treats its +first directory listing as a snapshot: request files that predate the turn are deleted, never +executed, so a crashed earlier turn cannot leak a stale call into a warm-continued one. + +Pickup is event-driven, with polling kept as the fallback. The writer arms one coalescing +`fs.watch` on the relay dir before its first response check, so the runner's answer wakes it +instantly; the 300 ms poll survives as the racing safety timer. The runner's loop wakes the +same way: locally from an in-process `fs.watch` (unflagged; it only shortens the poll sleeps), +and on Daytona, behind a flag, from one bounded watch exec per window inside the sandbox. +While the Daytona watch is healthy the runner suspends its remote `ls` polling and keeps a +30 s safety poll; failures demote the turn to classic polling with jittered backoff and one +log line. Three env vars control the wakes: + +- `AGENTA_AGENT_TOOLS_RELAY_RESPONSE_WATCH_ENABLED`: the in-sandbox response watch (hop 1). + Default true; only the exact strings `false` and `0` disable it. Forwarded into the sandbox + env only when the operator set it. +- `AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_ENABLED`: the Daytona watch exec (hop 2). Default + false; only the exact strings `true` and `1` enable it. +- `AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_WINDOW_MS`: one watch exec window. Default 25000, + clamped to [5000, 120000], with downward-only jitter of up to 20 percent. Keep it below the + 30 s safety poll: a window of 30 s or more still works but degrades pickup latency and can + demote a healthy watch. + +The existing relay vars (`AGENTA_AGENT_TOOLS_RELAY_POLLING`, `_POLLING_MAX`, +`_IDLE_GROW_AFTER`, `_TIMEOUT`) keep their names and meanings; they now describe the fallback +poll mode and the safety timers. + ### Platform tools: the runner calls an existing Agenta endpoint directly A `type: "platform"` tool exposes an EXISTING Agenta endpoint to the agent — a thin wrapper, no @@ -576,7 +612,9 @@ never drift from the files that exist. The canonical playbook format lives in th | Runtime dispatch (branch on `kind`) | `services/agent/src/tools/dispatch.ts` | | Callback transport | `services/agent/src/tools/callback.ts` | | Code execution | `services/agent/src/tools/code.ts` | -| Daytona/non-Pi relay | `services/agent/src/tools/relay.ts` | +| Daytona/non-Pi relay (runner-side loop) | `services/runner/src/tools/relay.ts` | +| In-sandbox relay writer + wire protocol | `services/runner/src/tools/relay-client.ts`, `relay-protocol.ts` | +| Relay wake sources (local `fs.watch`, Daytona watch exec) | `services/runner/src/tools/relay-watch.ts` | | Pi native delivery | `services/agent/src/extensions/agenta.ts` | | `agenta-tools` server for non-Pi harnesses | `services/agent/src/tools/mcp-bridge.ts`, `services/agent/src/tools/mcp-server.ts` | | Capability probe | `services/agent/src/engines/sandbox_agent/capabilities.ts` | diff --git a/docs/design/agent-workflows/interfaces/README.md b/docs/design/agent-workflows/interfaces/README.md index 57048a960a..9292015546 100644 --- a/docs/design/agent-workflows/interfaces/README.md +++ b/docs/design/agent-workflows/interfaces/README.md @@ -47,7 +47,7 @@ page. `Status` is read from each page's prose: **stable** (wired and unlikely to | [Agent config schema](public-edge/agent-config-schema.md) | public | `agent/schemas.py`, `sdk/utils/types.py`, `agents/dtos.py` (`HARNESS_IDENTITIES`) | stable | `unit/agents/test_dtos_agent_config.py`, `unit/agents/test_harness_identity.py` | | [`/run`](cross-service/service-to-agent-runner.md) | cross-service (the spine) | `protocol.ts`, `utils/wire.py`, `utils/ts_runner.py`, `server.ts`/`cli.ts` | stable (pinned by golden) | `unit/agents/test_wire_contract.py` + `golden/`, `services/agent/tests/unit/wire-contract.test.ts` | | [Runner to harness](cross-service/runner-to-harness.md) | cross-service (ACP) | `engines/sandbox_agent.ts` + `sandbox_agent/{run-plan,capabilities,permissions}.ts` | evolving | `services/agent/tests/unit/sandbox-agent-*.test.ts` | -| [Runner to MCP server](cross-service/runner-to-mcp-server.md) | cross-service | `agents/mcp/`, `engines/sandbox_agent/mcp.ts`, `tools/{mcp-bridge,mcp-server,relay}.ts` | evolving (stdio wired; remote deferred) | `services/agent/tests/unit/mcp-servers.test.ts` | +| [Runner to MCP server](cross-service/runner-to-mcp-server.md) | cross-service | `agents/mcp/`, `engines/sandbox_agent/mcp.ts`, `tools/{mcp-bridge,mcp-server,relay,relay-client,relay-protocol,relay-watch}.ts` | evolving (stdio wired; remote deferred) | `services/agent/tests/unit/mcp-servers.test.ts` | | [Runner to tool callback](cross-service/runner-to-tool-callback.md) | cross-service | `tools/{callback,dispatch,direct}.ts`, `apis/fastapi/tools/router.py` (`/tools/call`, `/tools/discover`, `_call_reserved_agenta_tool`), `core/tools/{discovery,service,platform_handlers}.py`, `agent/tools/resolver.py` | evolving (the `call` descriptor is wired and platform ops emit it; the legacy `tools.agenta.find_capabilities` route is deleted; reserved refs now dispatch server handlers, resolution flag-gated off until the runner half lands) | `services/agent/tests/unit/{code-tool,extension-tools}.test.ts`, `api unit/tools/{test_workflow_tool_call,test_discovery,test_platform_handlers}.py`, `unit/agents/platform/test_op_catalog.py` | | [Service and runner trace export](cross-service/service-and-runner-trace-export.md) | cross-service | `agent/tracing.py`, `tracing/otel.ts`, `extensions/agenta.ts` | stable | `services/agent/tests/unit/` | | [Service to vault and tool providers](cross-service/service-to-vault-and-tool-providers.md) | cross-service (external) | `agent/app.py`, `platform/{resolve,connections}.py`, `agents/capabilities.py`, `tools/router.py` | stable | `unit/agents/connections/`, `unit/agents/platform/`, `unit/agents/tools/` | diff --git a/docs/design/agent-workflows/interfaces/cross-service/runner-to-mcp-server.md b/docs/design/agent-workflows/interfaces/cross-service/runner-to-mcp-server.md index a1744bd9c6..4c6223f5d3 100644 --- a/docs/design/agent-workflows/interfaces/cross-service/runner-to-mcp-server.md +++ b/docs/design/agent-workflows/interfaces/cross-service/runner-to-mcp-server.md @@ -63,10 +63,18 @@ local", so an unknown remote provider fails closed too. A user http MCP server ( harness dials directly) is NOT loopback-bound and stays delivered on Daytona unchanged. **The file relay.** A resolved tool may need to run privately rather than inside the harness -process. The relay moves the call across that boundary: the child writes a `.req.json` -request into the relay directory, the runner polls (every ~300ms), executes the tool through -the `RelayHost` (local `node:fs` or the Daytona sandbox filesystem), and writes a -`.res.json` response. A tool slower than the relay timeout surfaces as a tool error. +process. The relay moves the call across that boundary: the child publishes a `.req.json` +request into the relay directory, the runner picks it up, executes the tool through the +`RelayHost` (local `node:fs` or the Daytona sandbox filesystem), and publishes a +`.res.json` response. Every publication is atomic on both sides: full bytes under a temp +name (`.tmp.`), then a same-directory rename, so neither side ever reads partial +JSON. The runner deletes the request file as soon as it has read it, so a request executes at +most once per publication; the writer still deletes the response after reading it (its own +request unlink is now usually a no-op). Crash redelivery is gone on both backends: a runner +crash after pickup loses the request, the writer times out, and the call surfaces as a tool +error. Pickup is event-driven with polling as the fallback: an in-process `fs.watch` locally, +a flagged bounded watch exec on Daytona, and the ~300 ms poll as the safety timer. A tool +slower than the relay timeout surfaces as a tool error. **User-declared servers.** `mcpServers` in `/run` carries each user server with its transport, command or url, args, env (secrets already injected by the Python resolver), tool @@ -109,7 +117,12 @@ allowlist, and permission. Two transports, opposite states: - `services/runner/src/tools/spec-schema.ts`: the shared `specInputSchema` accessor + arg validation. - `services/runner/src/tools/mcp-server.ts`: the removed stdio JSON-RPC server (refusing stub). -- `services/runner/src/tools/relay.ts`: the file relay loop and hosts (idle-poll backoff). +- `services/runner/src/tools/relay.ts`: the runner-side relay loop and hosts + (delete-on-pickup; idle-poll backoff in fallback mode). +- `services/runner/src/tools/relay-client.ts` and `relay-protocol.ts`: the bundle-safe + in-sandbox writer and wire protocol (atomic publication; the hop-1 response watch). +- `services/runner/src/tools/relay-watch.ts`: the hop-2 wake sources (local `fs.watch`; the + flagged Daytona watch exec). ## Watch for when changing @@ -127,8 +140,9 @@ allowlist, and permission. Two transports, opposite states: - **The remote-tools gate.** A non-Pi remote-sandbox run carrying ANY custom tool (client kind included) is refused in `run-plan.ts`. Swap it for a real in-sandbox delivery path when one exists; do not widen it. -- **The relay.** Polling interval, idle backoff, timeout, and the local-versus-Daytona host. A - slow tool must fail cleanly. +- **The relay.** Atomic temp-plus-rename publication, delete-on-pickup (at most one execution + per publication), the wake sources and their flags, polling interval, idle backoff, timeout, + and the local-versus-Daytona host. A slow tool must fail cleanly. - **HTTP MCP delivery.** `toAcpMcpServers` routes the resolved secret from `env` into a request header and builds the ACP `type: "http"` entry. Changing the env-to-header mapping or the ACP variant shape changes which auth reaches the remote server. diff --git a/services/runner/scripts/build-extension.mjs b/services/runner/scripts/build-extension.mjs index debdae88d7..56ec98684a 100644 --- a/services/runner/scripts/build-extension.mjs +++ b/services/runner/scripts/build-extension.mjs @@ -6,14 +6,16 @@ * Run: pnpm run build:extension -> dist/extensions/agenta.js */ import { build } from "esbuild"; +import { readFile } from "node:fs/promises"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; const root = join(dirname(fileURLToPath(import.meta.url)), ".."); +const outfile = join(root, "dist/extensions/agenta.js"); await build({ entryPoints: [join(root, "src/extensions/agenta.ts")], - outfile: join(root, "dist/extensions/agenta.js"), + outfile, bundle: true, platform: "node", format: "esm", @@ -27,4 +29,25 @@ await build({ logLevel: "info", }); +// Bundle-safety gate: the extension runs INSIDE the sandbox and must only pull in the +// bundle-safe relay modules (relay-client.ts, relay-protocol.ts). If a server-side +// relay symbol shows up in the bundle, someone imported runner-side code (relay.ts / +// relay-watch.ts, or a sandbox host's deleteFsEntry surface) into the extension graph. +const FORBIDDEN_SERVER_SYMBOLS = [ + "daytonaRelayActivitySource", + "startToolRelay", + "deleteFsEntry", +]; +const bundle = await readFile(outfile, "utf-8"); +const leaked = FORBIDDEN_SERVER_SYMBOLS.filter((symbol) => + bundle.includes(symbol), +); +if (leaked.length > 0) { + process.stderr.write( + `[build-extension] FATAL: server-side relay symbols leaked into the sandbox bundle: ${leaked.join(", ")}\n` + + "[build-extension] the extension may import only tools/relay-client.ts and tools/relay-protocol.ts; never tools/relay.ts or tools/relay-watch.ts\n", + ); + process.exit(1); +} + process.stderr.write("[build-extension] wrote dist/extensions/agenta.js\n"); diff --git a/services/runner/src/engines/sandbox_agent.ts b/services/runner/src/engines/sandbox_agent.ts index c744179602..476ee31982 100644 --- a/services/runner/src/engines/sandbox_agent.ts +++ b/services/runner/src/engines/sandbox_agent.ts @@ -418,7 +418,7 @@ const RUN_LIMIT_TRIPPED = Symbol("run-limit-tripped"); interface CurrentTurn { run: ReturnType; pause: PendingApprovalPauseController; - toolRelay?: { stop: () => Promise }; + toolRelay?: { ready?: Promise; stop: () => Promise }; /** Route a session/update for the active turn (suppress + handleUpdate + pause re-sweep). */ handleUpdate: (update: unknown) => void; /** Route a permission reverse-RPC for the active turn (built by attachPermissionResponder). */ @@ -1739,7 +1739,9 @@ export async function runTurn( if (plan.useToolRelay) { turn.toolRelay = (deps.startToolRelay ?? startToolRelay)( plan.isDaytona - ? (deps.sandboxRelayHost ?? sandboxRelayHost)(env.sandbox) + ? (deps.sandboxRelayHost ?? sandboxRelayHost)(env.sandbox, { + log: logger, + }) : (deps.localRelayHost ?? localRelayHost)(), plan.relayDir, plan.toolSpecs, @@ -1747,7 +1749,14 @@ export async function runTurn( request.runContext, env.clientToolRelayRef.current, relayGuard, + { log: logger }, ); + // Ordering invariant: the relay's stale-file sweep must complete before the + // resume's respondPermission or the fresh prompt below can cause a legitimate + // request, so nothing legitimate can predate the sweep and be swallowed as + // stale. Optional-chained so a fake relay without `ready` is tolerated, and a + // sweep failure never kills the turn. + await turn.toolRelay?.ready?.catch?.(() => {}); } // The prompt promise this turn races against the pause signal. A normal/continuation turn diff --git a/services/runner/src/engines/sandbox_agent/pi-assets.ts b/services/runner/src/engines/sandbox_agent/pi-assets.ts index f1f40da335..63d9340a61 100644 --- a/services/runner/src/engines/sandbox_agent/pi-assets.ts +++ b/services/runner/src/engines/sandbox_agent/pi-assets.ts @@ -70,6 +70,13 @@ export function buildPiExtensionEnv( if (specs.length && opts.relayDir) { env.AGENTA_AGENT_TOOLS_PUBLIC_SPECS = JSON.stringify(specs); env.AGENTA_AGENT_TOOLS_RELAY_DIR = opts.relayDir; + // Hop-1 response-watch kill switch (event-driven-tool-relay plan, decision 7): the + // in-sandbox writer defaults it to true, so it is only forwarded — verbatim — when + // the operator set it on the runner. + const responseWatch = + process.env.AGENTA_AGENT_TOOLS_RELAY_RESPONSE_WATCH_ENABLED; + if (responseWatch !== undefined) + env.AGENTA_AGENT_TOOLS_RELAY_RESPONSE_WATCH_ENABLED = responseWatch; } // Builtin gating needs no relay dir: the gate rides the extension's `ctx.ui.confirm` // dialog onto the ACP permission plane (Pi approval parking), not the file relay. diff --git a/services/runner/src/tools/dispatch.ts b/services/runner/src/tools/dispatch.ts index 3abe105986..40ae9d26de 100644 --- a/services/runner/src/tools/dispatch.ts +++ b/services/runner/src/tools/dispatch.ts @@ -16,30 +16,18 @@ * so the call is relayed through the runner via files (see tools/relay.ts) when `relayDir` * is set; otherwise it POSTs directly. * - * `relayToolCall` lives here (not in extensions/agenta.ts) so this module is the single - * dispatch home with no import cycle back into a call site. + * `relayToolCall` now lives in tools/relay-client.ts (the bundle-safe in-sandbox writer) + * and is re-exported here so existing importers keep working. */ -import { - existsSync, - mkdirSync, - readFileSync, - unlinkSync, - writeFileSync, -} from "node:fs"; - import type { ResolvedToolSpec } from "../protocol.ts"; import { callAgentaTool } from "./callback.ts"; import { CODE_TOOL_UNSUPPORTED_MESSAGE } from "./code.ts"; +import { relayToolCall } from "./relay-client.ts"; import { assertRequiredArguments } from "./spec-schema.ts"; -import { - RELAY_POLL_MS, - RELAY_REQ_SUFFIX, - RELAY_RES_SUFFIX, - RELAY_TIMEOUT_MS, - sanitizeRelayId, - sleep, - type RelayResponse, -} from "./relay.ts"; + +// Compatibility re-export: the writer moved to `relay-client.ts`; importers that still +// reach it through this module keep working while they migrate. +export { relayToolCall } from "./relay-client.ts"; /** Options for executing a resolved tool. `endpoint`/`authorization`/`relayDir` only matter for callbacks. */ export interface RunResolvedToolOpts { @@ -55,57 +43,6 @@ export interface RunResolvedToolOpts { signal?: AbortSignal; } -/** - * Daytona tool call: the in-sandbox process can't reach Agenta, so write the request to a - * file the runner watches and poll for the response it writes back (see tools/relay.ts). - */ -export async function relayToolCall( - dir: string, - toolName: string, - toolCallId: string, - params: unknown, - timeoutMs?: number, - signal?: AbortSignal, -): Promise { - const id = sanitizeRelayId(toolCallId); - const reqPath = `${dir}/${id}${RELAY_REQ_SUFFIX}`; - const resPath = `${dir}/${id}${RELAY_RES_SUFFIX}`; - try { - mkdirSync(dir, { recursive: true }); - } catch { - // The runner also creates it; a race here is harmless. - } - writeFileSync( - reqPath, - JSON.stringify({ toolName, toolCallId, args: params ?? {} }), - "utf-8", - ); - - const deadline = - Date.now() + - (timeoutMs && timeoutMs > 0 ? timeoutMs + 10_000 : RELAY_TIMEOUT_MS); - while (Date.now() < deadline) { - if (signal?.aborted) throw new Error("aborted"); - if (existsSync(resPath)) { - const res = JSON.parse(readFileSync(resPath, "utf-8")) as RelayResponse; - try { - unlinkSync(reqPath); - } catch { - /* best-effort cleanup */ - } - try { - unlinkSync(resPath); - } catch { - /* best-effort cleanup */ - } - if (res.ok) return res.text ?? ""; - throw new Error(res.error || `tool relay failed for ${toolName}`); - } - await sleep(RELAY_POLL_MS); - } - throw new Error(`tool relay timed out for ${toolName}`); -} - /** * Execute one resolved tool and return its result text. Throws on failure; every call site * turns the throw into a tool-error result so the model loop continues rather than crashing. diff --git a/services/runner/src/tools/relay-client.ts b/services/runner/src/tools/relay-client.ts new file mode 100644 index 0000000000..5a2bc35f9b --- /dev/null +++ b/services/runner/src/tools/relay-client.ts @@ -0,0 +1,270 @@ +/** + * In-sandbox relay writer client: publish one `.req.json` request file and wait for + * the `.res.json` response the runner writes back (see tools/relay.ts for the + * runner-side loop). + * + * This module runs INSIDE the sandbox (bundled into the Pi extension, and consumed by + * the future in-sandbox MCP shim, #5234), so it MUST stay bundle-safe: it may import + * ONLY node builtins and ./relay-protocol.ts — never server-side runner modules. + */ +import { + existsSync, + mkdirSync, + readFileSync, + renameSync, + unlinkSync, + watch, + writeFileSync, + type FSWatcher, +} from "node:fs"; +import { dirname } from "node:path"; + +import { + RELAY_POLL_MS, + RELAY_REQ_SUFFIX, + RELAY_RES_SUFFIX, + RELAY_TIMEOUT_MS, + relayEnvFlag, + relayTempPath, + sanitizeRelayId, + serializeRelayRequest, + sleep, + type ExecuteRelayRequest, + type RelayResponse, +} from "./relay-protocol.ts"; + +/** + * Write one execute request file into the relay dir. Returns the request path and the + * response path the runner will answer on. Publication is atomic (plan decision 2): the + * bytes go to a temp name first and a same-directory rename publishes the final name, so + * the runner's reader can never observe partial JSON. The final on-disk bytes are + * unchanged (the golden test pins them). + */ +export function publishRelayRequest( + dir: string, + req: ExecuteRelayRequest, +): { reqPath: string; resPath: string } { + const id = sanitizeRelayId(req.toolCallId); + const reqPath = `${dir}/${id}${RELAY_REQ_SUFFIX}`; + const resPath = `${dir}/${id}${RELAY_RES_SUFFIX}`; + try { + mkdirSync(dir, { recursive: true }); + } catch { + // The runner also creates it; a race here is harmless. + } + const tmpPath = relayTempPath(reqPath); + writeFileSync(tmpPath, serializeRelayRequest(req), "utf-8"); + renameSync(tmpPath, reqPath); + return { reqPath, resPath }; +} + +/** Thrown by `waitForRelayResponse` when the deadline passes with no response file. */ +export class RelayTimeoutError extends Error { + constructor(resPath: string) { + super(`tool relay timed out waiting on ${resPath}`); + this.name = "RelayTimeoutError"; + } +} + +/** + * Hop-1 response-watch kill switch (plan decision 7), read at CALL time so a test or an + * operator restart takes effect immediately. Default true; only the exact strings + * "false" and "0" disable it. + */ +export function responseWatchEnabled(): boolean { + return relayEnvFlag("AGENTA_AGENT_TOOLS_RELAY_RESPONSE_WATCH_ENABLED", true); +} + +export interface RelayDirWatch { + /** + * Wait for the next directory event or `timeoutMs`, whichever comes first. A pending + * event that arrived while no waiter was armed resolves immediately ("activity"). + */ + wait: (timeoutMs: number) => Promise<"activity" | "timeout">; + close: () => void; +} + +/** + * Coalescing directory watch (the plan's decision-3 activity-source shape, single-waiter + * form): ONE `fs.watch` is armed at creation and any event — eventType and filename are + * ignored entirely, filenames may be absent per the Node docs — sets a sticky pending + * bit and wakes the current waiter, if any. `wait()` consumes the sticky bit, so an + * event that lands between two waits stays observable; a timer win clears the waiter and + * an event win clears the timer, so thousands of consecutive waits accumulate zero + * listeners and zero timers (the watch callback is attached exactly once, at creation). + * + * Degradation, never rejection: a synchronous `fs.watch` throw returns undefined (the + * caller falls back to a plain poll); a watcher "error" after creation closes the + * watcher, and every later `wait()` resolves via its timer only. + * + * Single consumer: at most one `wait()` may be pending at a time (a second concurrent + * `wait()` orphans the first waiter's wake), and a `wait()` after `close()` resolves + * "timeout" immediately — a caller looping on `wait()` must stop once it closed the + * watch. + */ +export function createRelayDirWatch(dir: string): RelayDirWatch | undefined { + let pending = false; + let waiter: ((outcome: "activity" | "timeout") => void) | undefined; + let closed = false; + + const notify = (): void => { + if (waiter) { + const resolve = waiter; + waiter = undefined; + resolve("activity"); + } else { + pending = true; + } + }; + + let watcher: FSWatcher; + try { + watcher = watch(dir, notify); + } catch { + return undefined; + } + // Degrade on a post-creation watcher error: stop watching, let waits time out. Never + // reject and never leave the "error" event unhandled (an unlistened EventEmitter + // "error" would throw). + watcher.on("error", () => { + try { + watcher.close(); + } catch { + // Already closed; nothing to release. + } + }); + + return { + wait: (timeoutMs) => { + if (pending) { + pending = false; + return Promise.resolve("activity"); + } + if (closed) return Promise.resolve("timeout"); + return new Promise((resolve) => { + const timer = setTimeout(() => { + waiter = undefined; + resolve("timeout"); + }, timeoutMs); + waiter = (outcome) => { + clearTimeout(timer); + resolve(outcome); + }; + }); + }, + close: () => { + if (closed) return; + closed = true; + try { + watcher.close(); + } catch { + // Already closed; nothing to release. + } + if (waiter) { + // Resolve an in-flight wait as a timer win; the hop-1 caller just re-checks + // existsSync and hits its deadline as usual. + const resolve = waiter; + waiter = undefined; + resolve("timeout"); + } + }, + }; +} + +/** + * Wait for the runner's response file until it appears, the signal aborts, or the + * deadline passes. The caller computes the total timeout; this function checks the + * abort per iteration, reads and parses the file when it exists, and otherwise waits up + * to RELAY_POLL_MS between checks. Throws `Error("aborted")` on abort and a + * `RelayTimeoutError` at the deadline. + * + * Hop 1 of the event-driven relay (plan decisions 3, 6, 7): when the response watch is + * enabled (default), a directory watch is armed BEFORE the first existsSync check — + * arm-then-check closes the created-before-armed race — and each sleep becomes a wait + * that a directory event cuts short. The RELAY_POLL_MS cadence survives as the racing + * safety timer, so the watch only shortens sleeps, never lengthens them; abort is still + * noticed at worst one poll interval later, exactly as today. A watch that cannot be + * created degrades to the plain poll. + */ +export async function waitForRelayResponse( + resPath: string, + opts: { timeoutMs: number; signal?: AbortSignal }, +): Promise { + const deadline = Date.now() + opts.timeoutMs; + if (opts.signal?.aborted) throw new Error("aborted"); + // Fast path: a response that already exists needs no watcher at all (common for + // quick tools whose runner answered before this waiter started). The miss path + // below keeps arm-then-check: the watch is created BEFORE the next existsSync so a + // file created between the two is still observed. + if (existsSync(resPath)) { + return JSON.parse(readFileSync(resPath, "utf-8")) as RelayResponse; + } + const dirWatch = responseWatchEnabled() + ? createRelayDirWatch(dirname(resPath)) + : undefined; + try { + while (Date.now() < deadline) { + if (opts.signal?.aborted) throw new Error("aborted"); + if (existsSync(resPath)) { + return JSON.parse(readFileSync(resPath, "utf-8")) as RelayResponse; + } + if (dirWatch) await dirWatch.wait(RELAY_POLL_MS); + else await sleep(RELAY_POLL_MS); + } + // Deadline-coincident wake: the last wait may have been cut short by the very + // event that published the response, with Date.now() landing on/after the + // deadline. One final check turns that from a spurious timeout into the answer. + if (existsSync(resPath)) { + return JSON.parse(readFileSync(resPath, "utf-8")) as RelayResponse; + } + throw new RelayTimeoutError(resPath); + } finally { + dirWatch?.close(); + } +} + +/** + * Daytona tool call: the in-sandbox process can't reach Agenta, so write the request to a + * file the runner watches and poll for the response it writes back (see tools/relay.ts). + */ +export async function relayToolCall( + dir: string, + toolName: string, + toolCallId: string, + params: unknown, + timeoutMs?: number, + signal?: AbortSignal, +): Promise { + const { reqPath, resPath } = publishRelayRequest(dir, { + toolName, + toolCallId, + args: params, + }); + + const totalTimeoutMs = + timeoutMs && timeoutMs > 0 ? timeoutMs + 10_000 : RELAY_TIMEOUT_MS; + let res: RelayResponse; + try { + res = await waitForRelayResponse(resPath, { + timeoutMs: totalTimeoutMs, + signal, + }); + } catch (err) { + if (err instanceof RelayTimeoutError) { + throw new Error(`tool relay timed out for ${toolName}`); + } + throw err; + } + try { + unlinkSync(reqPath); + } catch { + /* best-effort cleanup */ + } + try { + unlinkSync(resPath); + } catch { + /* best-effort cleanup */ + } + if (res.ok) return res.text ?? ""; + throw new Error(res.error || `tool relay failed for ${toolName}`); +} diff --git a/services/runner/src/tools/relay-protocol.ts b/services/runner/src/tools/relay-protocol.ts new file mode 100644 index 0000000000..9924447673 --- /dev/null +++ b/services/runner/src/tools/relay-protocol.ts @@ -0,0 +1,82 @@ +/** + * Relay wire protocol: the file-name suffixes, request/response JSON shapes, request + * serialization, and the env-derived timing constants shared by every relay writer and + * the runner-side reader (tools/relay.ts). + * + * This module is the writer-shared relay protocol and MUST stay dependency-free (node + * builtins only, no imports from other src modules) so it stays bundle-safe: the Pi + * extension bundle and the future in-sandbox MCP shim (#5234) consume it from inside + * the sandbox, where server-side code must never be pulled in. + */ + +export const RELAY_REQ_SUFFIX = ".req.json"; +export const RELAY_RES_SUFFIX = ".res.json"; +export const RELAY_POLL_MS = Number( + process.env.AGENTA_AGENT_TOOLS_RELAY_POLLING ?? 300, +); +export const RELAY_TIMEOUT_MS = Number( + process.env.AGENTA_AGENT_TOOLS_RELAY_TIMEOUT ?? 60000, +); + +export interface ExecuteRelayRequest { + kind?: "execute"; + toolName: string; + toolCallId: string; + args: unknown; +} +export type RelayRequest = ExecuteRelayRequest; + +export interface ExecuteRelayResponse { + kind?: "execute"; + ok: boolean; + text?: string; + error?: string; +} +export type RelayResponse = ExecuteRelayResponse; + +/** Make a tool-call id safe to use as a filename (and bounded). */ +export function sanitizeRelayId(id: string): string { + return id.replace(/[^A-Za-z0-9_-]/g, "_").slice(0, 120) || "tool"; +} + +/** + * Temporary publication name for one relay file (atomic publication, plan decision 2). + * Both directions write the full contents to this name first and then rename it to + * `finalPath`. Temp names never match the `.req.json`/`.res.json` suffix filters, so no + * reader or watcher ever lists them, and a same-directory rename is atomic on POSIX, so + * a final-name file always holds complete bytes — a wake can never observe partial JSON. + */ +export function relayTempPath(finalPath: string): string { + const nonce = Math.random().toString(36).slice(2, 10); + return `${finalPath}.tmp.${nonce}`; +} + +/** + * Serialize one execute request to the exact bytes every relay writer produces. The key + * order (toolName, toolCallId, args) and the `args ?? {}` default are part of the + * cross-writer contract; the golden test in tests/unit/relay-client.test.ts pins them. + */ +export function serializeRelayRequest(req: ExecuteRelayRequest): string { + return JSON.stringify({ + toolName: req.toolName, + toolCallId: req.toolCallId, + args: req.args ?? {}, + }); +} + +export const sleep = (ms: number): Promise => + new Promise((r) => setTimeout(r, ms)); + +/** + * Read one boolean relay env flag at CALL time (so a test or an operator restart takes + * effect immediately). The exact strings "false"/"0" disable, "true"/"1" enable, and + * anything else (unset, empty, garbage) falls back to `defaultValue`. Shared by the + * hop-1 response-watch flag (default true) and the hop-2 remote-watch flag (default + * false) so the two flags cannot drift in parsing. + */ +export function relayEnvFlag(name: string, defaultValue: boolean): boolean { + const value = process.env[name]; + if (value === "false" || value === "0") return false; + if (value === "true" || value === "1") return true; + return defaultValue; +} diff --git a/services/runner/src/tools/relay-watch.ts b/services/runner/src/tools/relay-watch.ts new file mode 100644 index 0000000000..94a43d967a --- /dev/null +++ b/services/runner/src/tools/relay-watch.ts @@ -0,0 +1,514 @@ +/** + * Hop 2 wake sources for the runner's relay loop (tools/relay.ts, plan decisions 3, 4, + * 6, 7 of docs/design/agent-workflows/projects/event-driven-tool-relay/plan.md). + * + * Two implementations of one contract (`RelayActivitySource`): + * + * - `localRelayActivitySource`: an in-process `fs.watch` on the relay dir for the local + * backend. No flag; a wake only shortens the poll sleep, the cadence is unchanged. + * - `daytonaRelayActivitySource`: a re-issued bounded watch exec (`node -e` inside the + * sandbox via `sandbox.runProcess`) that REPLACES the runner's remote polling while + * healthy, backed by a 30 s safety poll and demoted to classic polling after repeated + * failure. Behind `AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_ENABLED` (default false). + * + * This module is SERVER-SIDE: it may import ./relay-client.ts and ./relay-protocol.ts, + * but must never be imported by them (they are bundled into the sandbox; + * scripts/build-extension.mjs fails the build if this module's symbols appear in the + * extension bundle). + */ +import { relayEnvFlag } from "./relay-protocol.ts"; +import { createRelayDirWatch } from "./relay-client.ts"; + +export interface RelayActivitySource { + wait(options: { + timeoutMs: number; + signal?: AbortSignal; + }): Promise<"activity" | "timeout" | "closed">; + close(): void; + /** False once the source is demoted or closed; the relay loop then uses classic polling. */ + isHealthy(): boolean; + /** Whether a healthy source replaces the remote poll (Daytona watch: true; local fs.watch: false). */ + readonly suspendsPolling: boolean; + /** Optional: a request discovered by the safety poll while healthy counts as a watch miss (feeds demotion). */ + noteMiss?(): void; +} + +/** + * Hop 2 remote-watch kill switch (plan decision 7), read at call time. Default FALSE + * (flips to true after the QA pass); only the exact strings "true" and "1" enable it. + */ +export function remoteWatchEnabled(): boolean { + return relayEnvFlag("AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_ENABLED", false); +} + +/** + * The runner's remote `ls` cadence while a suspended-polling watch is healthy (plan + * decision 4): pickup latency stays bounded by this no matter what the watch subsystem + * does, including lying about success. + */ +export const RELAY_SAFETY_POLL_MS = 30_000; + +/** Default bounded lifetime of one watch exec window (plan decision 7). */ +export const RELAY_REMOTE_WATCH_WINDOW_DEFAULT_MS = 25_000; +/** Clamp floor: below 5 s the re-issue cadence becomes the request storm this removes. */ +export const RELAY_REMOTE_WATCH_WINDOW_MIN_MS = 5_000; +/** Clamp ceiling: above 2 min the bounded-lifetime and orphan-cleanup assumptions stop holding. */ +export const RELAY_REMOTE_WATCH_WINDOW_MAX_MS = 120_000; +/** Daemon-timeout grace on top of the window so normal expiry never reads as a daemon timeout. */ +const RELAY_REMOTE_WATCH_GRACE_MS = 5_000; +/** The in-sandbox script's own periodic readdir (the in-window correctness fallback). */ +const RELAY_WATCH_READDIR_POLL_MS = 2_000; +/** Consecutive failures before the source demotes itself to classic polling. */ +const RELAY_WATCH_DEMOTION_THRESHOLD = 3; +const RELAY_WATCH_BACKOFF_BASE_MS = 1_000; +const RELAY_WATCH_BACKOFF_CAP_MS = 30_000; + +/** + * Parse and clamp `AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_WINDOW_MS` (plan decision 7). + * Read once at source creation. Missing -> default silently; set-but-unparseable -> + * default with one warning; out of range -> clamped with one warning. + */ +export function resolveRemoteWatchWindowMs( + log: (msg: string) => void = () => {}, +): number { + const raw = process.env.AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_WINDOW_MS; + if (raw === undefined || raw === "") + return RELAY_REMOTE_WATCH_WINDOW_DEFAULT_MS; + const parsed = Number.parseInt(raw, 10); + if (!Number.isFinite(parsed)) { + log( + `[relay] unparseable AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_WINDOW_MS '${raw}'; using default ${RELAY_REMOTE_WATCH_WINDOW_DEFAULT_MS}`, + ); + return RELAY_REMOTE_WATCH_WINDOW_DEFAULT_MS; + } + if (parsed < RELAY_REMOTE_WATCH_WINDOW_MIN_MS) { + log( + `[relay] AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_WINDOW_MS ${parsed} below minimum; clamped to ${RELAY_REMOTE_WATCH_WINDOW_MIN_MS}`, + ); + return RELAY_REMOTE_WATCH_WINDOW_MIN_MS; + } + if (parsed > RELAY_REMOTE_WATCH_WINDOW_MAX_MS) { + log( + `[relay] AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_WINDOW_MS ${parsed} above maximum; clamped to ${RELAY_REMOTE_WATCH_WINDOW_MAX_MS}`, + ); + return RELAY_REMOTE_WATCH_WINDOW_MAX_MS; + } + return parsed; +} + +/** + * DOWNWARD-ONLY jitter (−20%..0%): windows and backoffs never re-issue in lockstep + * (fleet desync is preserved), and a jittered window is always <= the nominal window, + * so a 25 s default window always completes before the 30 s safety wait — an upward + * draw used to race the safety poll and read as a false watch miss. Deliberate + * deviation from the plan's ±20% (decision 4/7), recorded here. + */ +export function applyRelayWatchJitter(ms: number): number { + return Math.round(ms * (0.8 + Math.random() * 0.2)); +} + +/** + * Local-backend hop 2 wake source: the in-process `fs.watch` adapter over the shared + * coalescing dir watch (relay-client.ts). It never suspends polling — a wake only + * shortens the sleep, so a degraded inner watch just means waits resolve by timer and + * the loop's cadence is unchanged; no demotion machinery needed (isHealthy stays true + * until closed). A dir that cannot be watched returns undefined (plain poll). + */ +export function localRelayActivitySource( + dir: string, +): RelayActivitySource | undefined { + const dirWatch = createRelayDirWatch(dir); + if (!dirWatch) return undefined; + let closed = false; + return { + suspendsPolling: false, + isHealthy: () => !closed, + wait: async ({ timeoutMs }) => { + if (closed) return "closed"; + // The inner watch resolves a wait cut short by close() as "timeout"; the closed + // flag remaps it (and any later wait) to "closed" for the activity-source contract. + const outcome = await dirWatch.wait(timeoutMs); + return closed ? "closed" : outcome; + }, + close: () => { + if (closed) return; + closed = true; + dirWatch.close(); + }, + }; +} + +/** + * The in-sandbox watch script (plan decision 4, hardening list). Plain CommonJS for + * `node -e`. Argv only — the relay dir is NEVER interpolated into this source (a path + * with quotes, whitespace, newlines, or shell metacharacters must be safe; runProcess + * passes argv directly with no shell). Order matters: arm `fs.watch` FIRST, then the + * sync readdir check (a file landing during either fires the already-armed watch), then + * the interval + window timer. `finish()` is idempotent, guards not-yet-assigned + * handles, never calls process.exit() and never writes stdout: it just releases every + * handle so the event loop drains and the process exits 0 — the exec COMPLETION is the + * wake signal. A synchronous `fs.watch` throw (dir missing) leaves the interval + timer + * running: the script degrades to an in-sandbox readdir poll for the window. + */ +export const RELAY_WATCH_SCRIPT = `"use strict"; +const fs = require("fs"); +const dir = process.argv[1]; +const windowMs = Number(process.argv[2]); +const readdirPollMs = Number(process.argv[3]); +let finished = false; +let watcher; +let interval; +let timer; +function finish() { + if (finished) return; + finished = true; + if (watcher) { + try { + watcher.close(); + } catch (e) {} + } + if (interval) clearInterval(interval); + if (timer) clearTimeout(timer); +} +try { + watcher = fs.watch(dir, finish); + watcher.on("error", finish); +} catch (e) {} +function hasReq() { + try { + return fs.readdirSync(dir).some(function (n) { + return n.endsWith(".req.json"); + }); + } catch (e) { + return false; + } +} +if (hasReq()) finish(); +if (!finished) { + interval = setInterval(function () { + if (hasReq()) finish(); + }, readdirPollMs); + timer = setTimeout(finish, windowMs); +} +`; + +/** + * Build the exec argv for one watch window. `node -e` argv semantics, verified + * empirically (`node -e 'console.log(process.argv)' a b` prints + * `[execPath, "a", "b"]`): the script sees the trailing arguments starting at + * `process.argv[1]`, so argv[1] = relay dir, argv[2] = window ms, argv[3] = readdir + * poll ms. The dir rides argv, never the script text. + */ +export function buildRelayWatchScriptArgs( + dir: string, + windowMs: number, + readdirPollMs: number, +): { command: string; args: string[] } { + return { + command: "node", + args: [ + "-e", + RELAY_WATCH_SCRIPT, + dir, + String(windowMs), + String(readdirPollMs), + ], + }; +} + +/** The one daemon call this source issues; matches sandbox-agent's `runProcess`. */ +export interface RelayWatchSandbox { + runProcess: (request: { + command: string; + args: string[]; + timeoutMs: number; + }) => Promise< + { exitCode?: number | null; timedOut?: boolean } | undefined | null + >; +} + +/** Runner-side slack on top of the daemon bound before a never-settling exec is abandoned. */ +const RELAY_WATCH_OUTER_BOUND_MARGIN_MS = 2_000; + +export interface DaytonaRelayActivitySourceOptions { + windowMs?: number; + readdirPollMs?: number; + backoffBaseMs?: number; + backoffCapMs?: number; + /** Daemon-timeout grace over the jittered window (default 5 s); injectable for tests. */ + graceMs?: number; + /** Runner-side outer-bound margin over window + grace (default 2 s); injectable for tests. */ + outerBoundMarginMs?: number; + log?: (msg: string) => void; +} + +/** + * Daytona hop 2 wake source (plan decisions 3 + 4): the single-flight window loop. + * While healthy it replaces the runner's remote `ls` polling (`suspendsPolling`): each + * `wait()` arms at most one bounded watch exec in the sandbox and treats the exec's + * COMPLETION as the wake. Repeated failure (exec rejection, daemon timeout, or a + * safety-poll discovery the watch missed) demotes the source for the turn, with + * jittered exponential backoff on the way there, and the loop reverts to classic + * polling. + */ +export function daytonaRelayActivitySource( + sandbox: RelayWatchSandbox, + dir: string, + opts?: DaytonaRelayActivitySourceOptions, +): RelayActivitySource { + const log = opts?.log ?? (() => {}); + const windowMs = opts?.windowMs ?? resolveRemoteWatchWindowMs(log); + const readdirPollMs = opts?.readdirPollMs ?? RELAY_WATCH_READDIR_POLL_MS; + const backoffBaseMs = opts?.backoffBaseMs ?? RELAY_WATCH_BACKOFF_BASE_MS; + const backoffCapMs = opts?.backoffCapMs ?? RELAY_WATCH_BACKOFF_CAP_MS; + const graceMs = opts?.graceMs ?? RELAY_REMOTE_WATCH_GRACE_MS; + const outerBoundMarginMs = + opts?.outerBoundMarginMs ?? RELAY_WATCH_OUTER_BOUND_MARGIN_MS; + + let closed = false; + let demoted = false; + let stickyActivity = false; + let consecutiveFailures = 0; + let nextArmAt = 0; + // Window generations: each armWindow gets a fresh generation, and only the LIVE + // generation may settle. `liveGeneration !== 0` is also the single source of truth + // for "a window is in flight". The outer-bound timer abandons a never-settling exec + // by killing its generation, so a late settle from the abandoned promise is ignored + // entirely (no wake, no counter reset, no in-flight mutation). + let armGeneration = 0; + let liveGeneration = 0; // 0 = no live window + let outerBoundTimer: ReturnType | undefined; + let deferredArmTimer: ReturnType | undefined; + let waiter: + | ((outcome: "activity" | "timeout" | "closed") => void) + | undefined; + + const clearOuterBound = (): void => { + if (outerBoundTimer !== undefined) { + clearTimeout(outerBoundTimer); + outerBoundTimer = undefined; + } + }; + + const clearDeferredArm = (): void => { + if (deferredArmTimer !== undefined) { + clearTimeout(deferredArmTimer); + deferredArmTimer = undefined; + } + }; + + /** + * Deferred arm: (re)schedule the single source-level timer for `nextArmAt`. At most + * one such timer ever exists (a reschedule replaces it), and it is cleared when the + * current wait settles and on close(). The timer body goes back through `tryArm`, + * which re-defers if it fired marginally early (Node truncates delays) or the gate + * moved (another failure pushed `nextArmAt`) while the wait was parked. + */ + const scheduleDeferredArm = (): void => { + clearDeferredArm(); + deferredArmTimer = setTimeout( + () => { + deferredArmTimer = undefined; + tryArm(); + }, + Math.max(1, nextArmAt - Date.now()), + ); + }; + + /** + * The single arm gate (fix 4 of the slice-3 review): arm a window now when the + * source can (not closed, not demoted, no live window) and the backoff gate is + * open; when only the gate blocks it, fall back to the deferred-arm timer. Used by + * (a) wait() entry, (b) the deferred-arm timer body, and (c) countFailure while a + * waiter is parked — so a window that fails 1 s into a 30 s wait re-arms after its + * backoff inside that same wait instead of leaving the source windowless until the + * safety timer. + */ + const tryArm = (): void => { + if (closed || demoted || liveGeneration !== 0) return; + if (Date.now() >= nextArmAt) { + clearDeferredArm(); + armWindow(); + return; + } + scheduleDeferredArm(); + }; + + /** + * The single failure account (finding 4 of the slice-2 review): every failure — + * exec rejection, nullish result, daemon timeout, nonzero/nullish exit, outer-bound + * expiry, safety-poll miss — increments the counter, applies jittered exponential + * backoff to the next arm, logs one line, and demotes at the threshold (exactly one + * demotion log ever). A failure that lands while a waiter is parked re-enters the + * arm gate so the wait is not left windowless (the gate defers past the backoff). + */ + const countFailure = (reason: string): void => { + consecutiveFailures += 1; + log(`[relay] watch failure: ${reason}`); + const raw = Math.min( + backoffBaseMs * 2 ** Math.max(0, consecutiveFailures - 1), + backoffCapMs, + ); + nextArmAt = Date.now() + applyRelayWatchJitter(raw); + if (!demoted && consecutiveFailures >= RELAY_WATCH_DEMOTION_THRESHOLD) { + demoted = true; + log( + `[relay] relay watch demoted to classic polling after ${consecutiveFailures} consecutive failures (${reason})`, + ); + } + if (waiter && !closed && !demoted) tryArm(); + }; + + /** Window completion or a straggler wake: resolve the waiter or set the sticky bit. */ + const wake = (): void => { + if (closed) return; + if (waiter) { + const resolve = waiter; + waiter = undefined; + resolve("activity"); + } else { + stickyActivity = true; + } + }; + + const armWindow = (): void => { + const gen = ++armGeneration; + liveGeneration = gen; + const jitteredWindow = applyRelayWatchJitter(windowMs); + const { command, args } = buildRelayWatchScriptArgs( + dir, + jitteredWindow, + readdirPollMs, + ); + // Issue the exec synchronously (single-flight accounting and the counting test + // observe it at arm time); a synchronously-throwing runProcess still lands in the + // .catch below instead of escaping armWindow. + let window: Promise< + { exitCode?: number | null; timedOut?: boolean } | undefined | null + >; + try { + window = Promise.resolve( + sandbox.runProcess({ + command, + args, + timeoutMs: jitteredWindow + graceMs, + }), + ); + } catch (err) { + window = Promise.reject(err); + } + // Runner-side outer bound: the daemon's timeoutMs is a promise the daemon may + // never keep (a blackholed proxy leaves runProcess pending forever, pinning the + // live window and starving demotion). Past window + grace + margin a still-pending + // window is abandoned: counted as a failure, generation killed. + outerBoundTimer = setTimeout( + () => { + outerBoundTimer = undefined; + if (closed || gen !== liveGeneration) return; + liveGeneration = 0; + countFailure("window outer bound expired (exec never settled)"); + }, + jitteredWindow + graceMs + outerBoundMarginMs, + ); + window + .then((result) => { + if (gen !== liveGeneration) return; // abandoned by the outer bound: dead gen + liveGeneration = 0; + clearOuterBound(); + if (closed) return; // abandoned window (see close()); its wake is meaningless now + // Completion classification (fix 2 of the slice-3 review). A nullish result + // is a broken daemon path, not a wake: an insta-resolving runProcess that + // returns nothing must feed demotion, never reset the counter and storm. + if (result === null || result === undefined) { + countFailure("exec resolved with no result"); + return; + } + if (result.timedOut === true) { + // The daemon killed the exec past window + grace: the script never expired on + // its own timer, so this is a failure, not a wake. + countFailure("daemon timeout"); + return; + } + if (result.exitCode === 0) { + consecutiveFailures = 0; + } else { + // Nonzero, null (signal-killed / OOM), or MISSING exit is still a wake (the + // list pass is harmless) but counts as a failure so a script that keeps + // dying — or a daemon that stops reporting exit codes — demotes the source; + // an absent exitCode must not read as success. + countFailure(`watch script exited with code ${result.exitCode}`); + } + wake(); + }) + .catch((err) => { + if (gen !== liveGeneration) return; // abandoned: keep the rejection handled, mutate nothing + liveGeneration = 0; + clearOuterBound(); + if (closed) return; + // The exec could not run at all: failure, backoff, and NO wake (the waiter's + // own timer resolves "timeout"). Never rethrown: a rejecting exec must never + // reject wait() or become an unhandled rejection. + countFailure( + `exec rejected: ${err instanceof Error ? err.message : String(err)}`, + ); + }); + }; + + return { + suspendsPolling: true, + isHealthy: () => !demoted && !closed, + noteMiss: () => { + if (closed || demoted) return; + // The safety poll found a request while a window claimed healthy: the watch lied. + countFailure("safety poll found a request the watch missed"); + }, + wait: ({ timeoutMs, signal }) => { + if (closed) return Promise.resolve("closed"); + if (stickyActivity) { + stickyActivity = false; + return Promise.resolve("activity"); + } + if (signal?.aborted) return Promise.resolve("closed"); + // The arm gate: arm now when it can, or defer to the timer when a wait that + // starts inside a backoff gap would otherwise sit windowless for its whole + // timeout (up to the 30 s safety wait). + tryArm(); + return new Promise((resolve) => { + let settled = false; + const settle = (outcome: "activity" | "timeout" | "closed"): void => { + if (settled) return; + settled = true; + clearTimeout(timer); + clearDeferredArm(); + if (waiter === settle) waiter = undefined; + signal?.removeEventListener("abort", onAbort); + resolve(outcome); + }; + const onAbort = (): void => settle("closed"); + // The timer win clears the waiter and the abort listener via settle, so + // thousands of consecutive waits accumulate zero listeners and zero timers. + const timer = setTimeout(() => settle("timeout"), timeoutMs); + waiter = settle; + signal?.addEventListener("abort", onAbort, { once: true }); + }); + }, + close: () => { + if (closed) return; + closed = true; + clearOuterBound(); + clearDeferredArm(); + // The in-flight runProcess CANNOT be aborted: the SDK's runProcess takes no + // per-call AbortSignal (verified against node_modules/sandbox-agent/dist/ + // index.d.ts; only connection-level signals exist). Deviation from plan + // invariant 5 ("aborts the in-flight held exec request"), softened to + // abandon-with-bounded-lifetime: the completion handler above no-ops once + // closed, its .catch keeps a late rejection handled, and the in-sandbox script + // dies at its own window timer (<= window + grace). + if (waiter) { + const resolve = waiter; + waiter = undefined; + resolve("closed"); + } + }, + }; +} diff --git a/services/runner/src/tools/relay.ts b/services/runner/src/tools/relay.ts index 7bcc9a00ac..72f1e951aa 100644 --- a/services/runner/src/tools/relay.ts +++ b/services/runner/src/tools/relay.ts @@ -1,5 +1,5 @@ /** - * Daytona tool relay. + * Daytona tool relay — the RUNNER side. * * Tool child processes do not receive private resolved specs, executable code, scoped env, * callback endpoints, or callback auth. They receive only public tool metadata plus this @@ -14,12 +14,21 @@ * ──▶ write `.res.json` * * The same loop supports local filesystem relays and Daytona sandbox filesystem relays. + * + * The relay is split across three modules: the bundle-safe wire protocol (suffixes, + * request/response shapes, serialization, shared timing) lives in `relay-protocol.ts`; + * the in-sandbox writer client lives in `relay-client.ts`; this module keeps the + * runner-side consumer/executor loop. The protocol pieces are re-exported below so + * existing importers keep compiling. */ import { existsSync, mkdirSync, readdirSync, readFileSync, + renameSync, + statSync, + unlinkSync, writeFileSync, } from "node:fs"; @@ -39,6 +48,23 @@ import type { ToolCallbackContext, } from "../protocol.ts"; import type { ClientToolRelay } from "./client-tool-relay.ts"; +import { + RELAY_POLL_MS, + RELAY_REQ_SUFFIX, + RELAY_RES_SUFFIX, + relayTempPath, + sleep, + type ExecuteRelayRequest, + type RelayRequest, + type RelayResponse, +} from "./relay-protocol.ts"; +import { + RELAY_SAFETY_POLL_MS, + daytonaRelayActivitySource, + localRelayActivitySource, + remoteWatchEnabled, + type RelayActivitySource, +} from "./relay-watch.ts"; import { assertRequiredArguments } from "./spec-schema.ts"; // Compatibility re-export: the type moved to `client-tool-relay.ts` (a pure type module); @@ -48,14 +74,23 @@ export type { ClientToolRelayRequest, } from "./client-tool-relay.ts"; -export const RELAY_REQ_SUFFIX = ".req.json"; -export const RELAY_RES_SUFFIX = ".res.json"; -export const RELAY_POLL_MS = Number( - process.env.AGENTA_AGENT_TOOLS_RELAY_POLLING ?? 300, -); -export const RELAY_TIMEOUT_MS = Number( - process.env.AGENTA_AGENT_TOOLS_RELAY_TIMEOUT ?? 60000, -); +// Compatibility re-export: the wire protocol moved to `relay-protocol.ts` (bundle-safe, +// shared with the in-sandbox writer); importers that still reach it through this module +// keep working while they migrate. +export { + RELAY_POLL_MS, + RELAY_REQ_SUFFIX, + RELAY_RES_SUFFIX, + RELAY_TIMEOUT_MS, + sanitizeRelayId, + sleep, +} from "./relay-protocol.ts"; +export type { + ExecuteRelayRequest, + ExecuteRelayResponse, + RelayRequest, + RelayResponse, +} from "./relay-protocol.ts"; /** * Idle-backoff cap for the runner relay poll. The loop polls `host.list(relayDir)` every * `RELAY_POLL_MS` (300 ms) for the whole turn — on Daytona that `list` is a remote `ls` exec @@ -79,21 +114,6 @@ export function relayPollDelayMs(idlePolls: number): number { return Math.min(RELAY_POLL_MS * factor, RELAY_POLL_MAX_MS); } -export interface ExecuteRelayRequest { - kind?: "execute"; - toolName: string; - toolCallId: string; - args: unknown; -} -export type RelayRequest = ExecuteRelayRequest; - -export interface ExecuteRelayResponse { - kind?: "execute"; - ok: boolean; - text?: string; - error?: string; -} -export type RelayResponse = ExecuteRelayResponse; const PAUSED = Symbol("paused"); /** @@ -164,18 +184,40 @@ export function redactContextBoundArgs( return redacted; } -/** Make a tool-call id safe to use as a filename (and bounded). */ -export function sanitizeRelayId(id: string): string { - return id.replace(/[^A-Za-z0-9_-]/g, "_").slice(0, 120) || "tool"; -} - -export const sleep = (ms: number): Promise => - new Promise((r) => setTimeout(r, ms)); - export interface RelayHost { list: (dir: string) => Promise; read: (path: string) => Promise; write: (path: string, contents: string) => Promise; + /** + * Atomically publish a fully written file under its final name (plan decision 2): + * the response is written to a `relayTempPath` name first and this rename makes it + * visible, so the in-sandbox writer can never read partial JSON. + */ + rename: (from: string, to: string) => Promise; + /** + * Delete one relay file. Used for delete-on-pickup: the runner removes a request + * file as soon as it has read it, so a watch that wakes on ANY `*.req.json` present + * (the Daytona watch exec) does not insta-complete and rearm for the whole + * execution. Call sites guard with their own try/catch (pickup removal is + * best-effort); implementations may throw. + */ + remove: (path: string) => Promise; + /** + * Optional mtime probe for one relay file, in epoch milliseconds; used only for the + * stage=relay_pickup telemetry line (pickup latency = now − request-file mtime). + * Undefined (no capability, missing file, or any error) makes the telemetry report + * pickup_ms=-1 — never an execution failure. Cost: one stat per executed request + * (local: free; Daytona: +1 daemon call per tool call, small next to the removed + * polling). + */ + statMtimeMs?: (path: string) => Promise; + /** + * Optional hop 2 wake source for the relay dir (plan decision 3). Undefined means + * the loop is byte-for-byte today's poll loop. A source that suspends polling + * (Daytona watch exec) replaces the remote poll with the 30 s safety poll while + * healthy; one that does not (local fs.watch) only shortens the poll sleep. + */ + createActivitySource?: (dir: string) => RelayActivitySource | undefined; } /** Relay host for child processes running on the same filesystem as the runner. */ @@ -190,11 +232,30 @@ export function localRelayHost(): RelayHost { mkdirSync(path.slice(0, path.lastIndexOf("/")), { recursive: true }); writeFileSync(path, contents, "utf-8"); }, + rename: async (from, to) => { + renameSync(from, to); + }, + remove: async (path) => { + unlinkSync(path); + }, + statMtimeMs: async (path) => { + try { + return statSync(path).mtimeMs; + } catch { + return undefined; + } + }, + // Unflagged (plan decision 7, last paragraph): the local watch's failure mode is + // "fall back to the poll" and the poll cadence is unchanged either way. + createActivitySource: (dir) => localRelayActivitySource(dir), }; } /** Relay host for child processes running inside a Daytona sandbox. */ -export function sandboxRelayHost(sandbox: any): RelayHost { +export function sandboxRelayHost( + sandbox: any, + opts?: { log?: (msg: string) => void }, +): RelayHost { return { list: async (dir) => { const ls = await sandbox.runProcess({ @@ -216,6 +277,36 @@ export function sandboxRelayHost(sandbox: any): RelayHost { write: async (path, contents) => { await sandbox.writeFsFile({ path }, contents); }, + rename: async (from, to) => { + // Verified against the daemon source (v0.4.2 router.rs): /v1/fs/move is Rust + // std::fs::rename, i.e. rename(2), atomic for a same-directory move (plan open + // question 2 resolved). `overwrite: true` only guards a pathological duplicate + // response; the final name never pre-exists in normal operation. + await sandbox.moveFs({ from, to, overwrite: true }); + }, + remove: async (path) => { + await sandbox.deleteFsEntry({ path }); + }, + statMtimeMs: async (path) => { + // FsStat.modified is an RFC3339 string | null (verified against + // node_modules/sandbox-agent/dist/index.d.ts); absent/null/unparseable -> undefined. + try { + const stat = await sandbox.statFs({ path }); + const modified = stat?.modified; + if (typeof modified !== "string") return undefined; + const parsed = Date.parse(modified); + return Number.isFinite(parsed) ? parsed : undefined; + } catch { + return undefined; + } + }, + // Flagged (plan decision 7): the remote watch changes what the runner asks the + // daemon to do, so it ships behind AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_ENABLED + // (default false). Off means today's poll loop, byte for byte. + createActivitySource: (dir) => + remoteWatchEnabled() + ? daytonaRelayActivitySource(sandbox, dir, { log: opts?.log }) + : undefined, }; } @@ -315,11 +406,77 @@ async function executeAllowedRelayedTool( ); } +/** A relay-owned file name: request, response, or either one's atomic-publication temp + * name. The sweep below may remove ONLY these — the relay dir also holds non-relay + * files (Pi's usage file lives there) that must never be touched. */ +function isRelayFileName(name: string): boolean { + return ( + name.endsWith(RELAY_REQ_SUFFIX) || + name.endsWith(RELAY_RES_SUFFIX) || + name.includes(`${RELAY_REQ_SUFFIX}.tmp.`) || + name.includes(`${RELAY_RES_SUFFIX}.tmp.`) + ); +} + +/** + * Clear pre-turn relay residue from a reused relay dir (a warm-continued turn skips the + * cold build's rm -rf, so a crashed prior turn can leave files behind). Every relay + * file already present is stale — requests would re-execute under this turn's fresh + * `seen` set, temp names are dead atomic-publication residue, and stale RESPONSES are + * dangerous too: a resumed approval reuses its original toolCallId, so a crashed prior + * attempt's `.res.json` would satisfy the new wait instantly with stale bytes. + * Non-relay names are never touched. + * + * The listing is retried up to 3 times (150 ms apart): a missing dir throws on some + * hosts and holds no stale files anyway, and a transiently unlistable dir accepts the + * pre-existing-residue risk rather than the swallow-a-live-request risk — the caller + * (startToolRelay) guarantees no LEGITIMATE request can exist before this sweep + * settles, so sweeping only what an early list shows is always safe, and giving up + * after 3 failures is too. + */ +export async function sweepStaleRelayFiles( + host: RelayHost, + relayDir: string, + log: (msg: string) => void, +): Promise { + for (let attempt = 1; attempt <= 3; attempt += 1) { + let names: string[]; + try { + names = await host.list(relayDir); + } catch { + if (attempt < 3) { + await sleep(150); + continue; + } + log("[relay] stale sweep skipped: relay dir unlistable after 3 attempts"); + return; + } + const stale = names.filter(isRelayFileName); + if (stale.length > 0) { + // Best-effort and concurrent: a failed removal is accepted (a leftover request + // is only re-armed watch noise once the loop's seen set has it; a leftover + // response is inert for fresh toolCallIds). + await Promise.allSettled( + stale.map((name) => host.remove(`${relayDir}/${name}`)), + ); + log( + `[relay] cleared ${stale.length} stale relay file(s) predating the turn`, + ); + } + return; + } +} + +/** The loop's wait outcome AT DISCOVERY time, for the stage=relay_pickup line. */ +type RelayWakeTag = "activity" | "timeout" | "closed" | "poll"; + /** - * Runner-side relay loop. Polls the sandbox relay dir for request files, executes each - * against the private spec in memory, and writes the response file - * the in-sandbox extension is waiting on. Returns `stop()` to end the loop and drain any - * in-flight executions; call it once the prompt resolves. + * Runner-side relay loop. Sweeps pre-turn residue, then polls the sandbox relay dir + * for request files, executes each against the private spec in memory, and writes the + * response file the in-sandbox extension is waiting on. Returns `ready` (resolves once + * the stale sweep settled — the caller must not allow a legitimate request before + * then) and `stop()` to end the loop and drain any in-flight executions; call it once + * the prompt resolves. */ export function startToolRelay( host: RelayHost, @@ -329,17 +486,43 @@ export function startToolRelay( runContext?: RunContext, clientToolRelay?: ClientToolRelay, guard?: RelayExecutionGuard, -): { stop: () => Promise } { + opts?: { log?: (msg: string) => void }, +): { ready: Promise; stop: () => Promise } { let active = true; + const log = opts?.log ?? (() => {}); + // Telemetry gate: without a log sink there is nowhere for pickup_ms to go, so the + // stat (a daemon round-trip on Daytona) is skipped entirely. + const telemetry = opts?.log !== undefined; const seen = new Set(); + // Request names whose delete-on-pickup remove rejected: retried (best-effort, + // concurrently) on each later list pass until the listing no longer shows them, so + // a lingering picked-up file cannot insta-wake watch windows forever. + const removeFailed = new Set(); const inflight: Promise[] = []; const specsByName = new Map(specs.map((spec) => [spec.name, spec])); - const handle = async (reqName: string): Promise => { - const id = reqName.slice(0, -RELAY_REQ_SUFFIX.length); + const writeResponse = async ( + id: string, + res: RelayResponse, + ): Promise => { + try { + // Atomic publication (plan decision 2): full bytes under a temp name, then a + // same-directory rename to the final name the in-sandbox writer waits on. + const finalResPath = `${relayDir}/${id}${RELAY_RES_SUFFIX}`; + const tmpResPath = relayTempPath(finalResPath); + await host.write(tmpResPath, JSON.stringify(res)); + await host.rename(tmpResPath, finalResPath); + } catch { + // The extension will time out and surface a tool error; nothing else to do here. + } + }; + + // Execute phase: runs in the background (pushed to `inflight`); the loop never + // waits on it. Parses, executes against the private spec, and publishes the + // response (or the error) the in-sandbox writer is waiting on. + const execute = async (id: string, raw: string): Promise => { let res: RelayResponse; try { - const raw = await host.read(`${relayDir}/${reqName}`); const req = JSON.parse(raw) as RelayRequest; const spec = specsByName.get(req.toolName); if (!spec) throw new Error(`unknown tool '${req.toolName}'`); @@ -359,43 +542,195 @@ export function startToolRelay( error: err instanceof Error ? err.message : String(err), }; } + await writeResponse(id, res); + }; + + // Pickup phase: read the request file, then clear it from the dir. The loop AWAITS + // every pickup before its next wait (fix 3 of the slice-3 review), so a watch exec + // can never arm while a picked-up request file still exists on disk — the window + // would insta-complete on it and rearm at network speed for the whole execution. + // The execute phase above starts as soon as the read returns; only read+stat+remove + // gate the loop. + const pickup = async (reqName: string, wake: RelayWakeTag): Promise => { + const id = reqName.slice(0, -RELAY_REQ_SUFFIX.length); + const reqPath = `${relayDir}/${reqName}`; + let raw: string; try { - await host.write( - `${relayDir}/${id}${RELAY_RES_SUFFIX}`, - JSON.stringify(res), + raw = await host.read(reqPath); + } catch (err) { + // Nothing was picked up; surface the read failure as the tool's response so the + // in-sandbox writer fails fast instead of waiting out its timeout. + inflight.push( + writeResponse(id, { + ok: false, + error: err instanceof Error ? err.message : String(err), + }), + ); + return; + } + // stage=relay_pickup telemetry: the stat is STARTED before the remove (afterwards + // the file is gone and the stat can only miss); both then run concurrently. Clock + // caveat: on Daytona the mtime is sandbox-clock while `Date.now()` is + // runner-clock, so pickup_ms is approximate — good for QA latency distributions, + // not billing. Cost: one stat per executed request, and none at all without a log + // sink (see `telemetry` above). + let statPromise: Promise | undefined; + if (telemetry && host.statMtimeMs) { + try { + statPromise = host.statMtimeMs(reqPath).catch(() => undefined); + } catch { + statPromise = undefined; + } + } + // Delete-on-pickup: remove the request file BEFORE the next watch window can arm + // (the loop awaits this pickup). This deliberately ends + // crash-redelivery-by-re-listing — a request is executed at most once per + // publication, consistent with the stale-sweep decision (a restarted turn must + // not re-execute stale requests; the writer times out and surfaces a tool error + // instead). A failed removal is recorded for retry on later list passes. + let removeDone: Promise; + try { + removeDone = host.remove(reqPath).then( + () => undefined, + () => { + removeFailed.add(reqName); + }, ); } catch { - // The extension will time out and surface a tool error; nothing else to do here. + removeFailed.add(reqName); + removeDone = Promise.resolve(); + } + // The execute phase starts NOW (as soon as the read returned) and runs in the + // background; the pickup itself only awaits stat + remove. + inflight.push(execute(id, raw)); + const mtimeMs = statPromise ? await statPromise : undefined; + log( + `[relay] stage=relay_pickup id=${id} pickup_ms=${ + mtimeMs === undefined ? -1 : Date.now() - mtimeMs + } wake=${wake}`, + ); + await removeDone; + }; + + // Loop-owned safety bound (fix 6 of the slice-3 review): the plan's 30 s pickup + // bound must hold even if a wait() implementation wedges, so every wait races a + // timer slightly past its own timeout. The timer is cleared as soon as the race + // settles, so nothing leaks and nothing outlives the loop. + const boundedWait = async ( + activitySource: RelayActivitySource, + timeoutMs: number, + ): Promise<"activity" | "timeout" | "closed"> => { + let timer: ReturnType | undefined; + const bound = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), timeoutMs + 1_000); + }); + try { + return await Promise.race([activitySource.wait({ timeoutMs }), bound]); + } finally { + clearTimeout(timer); } }; + // Hop 2 wake source (plan decision 3). Undefined (no capability, or the remote watch + // flag is off, or fs.watch failed) leaves the loop below byte-for-byte today's poll. + const source = host.createActivitySource?.(relayDir); + + // Stale sweep FIRST (fix 1 of the slice-3 review): the old "first successful list + // is the snapshot" raced the resume flow — on Daytona the snapshot ls exec's READ + // time is unordered against respondPermission, so a resume's approved request could + // land first and be swallowed, and a transiently rejecting first list deferred the + // snapshot behind later, legitimate requests. Now the sweep runs before the + // discovery loop, and `ready` lets the engine hold respondPermission/prompt until + // it settles — so nothing legitimate can predate the sweep, and after it EVERY + // listed request is legitimate. + const ready = sweepStaleRelayFiles(host, relayDir, log); + const loop = (async () => { + // The discovery loop starts only after the sweep settled (see `ready` above). + await ready.catch(() => {}); // Idle-poll backoff: a quiet turn (e.g. waiting on a browser-fulfilled client-tool pause) // grows the delay up to RELAY_POLL_MAX_MS instead of polling at 300 ms forever; any new // request resets it. This cuts the remote `ls` rate on Daytona without delaying a real call. let idlePolls = 0; - while (active) { - let sawNew = false; - try { - const names = await host.list(relayDir); - for (const name of names) { - if (!name.endsWith(RELAY_REQ_SUFFIX) || seen.has(name)) continue; - seen.add(name); - sawNew = true; - inflight.push(handle(name)); + let lastWaitOutcome: "activity" | "timeout" | "closed" | undefined; + try { + while (active) { + let sawNew = false; + // This iteration's pickup phases (and remove retries): all awaited below, + // BEFORE the next wait, so no watch window arms over a lingering file. + const pickups: Promise[] = []; + try { + const names = await host.list(relayDir); + // Retry earlier failed delete-on-pickup removals until gone from the listing. + for (const name of [...removeFailed]) { + if (!names.includes(name)) { + removeFailed.delete(name); + continue; + } + try { + pickups.push( + host.remove(`${relayDir}/${name}`).then( + () => { + removeFailed.delete(name); + }, + () => undefined, + ), + ); + } catch { + // Still failing synchronously; keep it in the retry set. + } + } + for (const name of names) { + if (!name.endsWith(RELAY_REQ_SUFFIX) || seen.has(name)) continue; + seen.add(name); + sawNew = true; + pickups.push(pickup(name, lastWaitOutcome ?? "poll")); + } + } catch { + // Transient (dir not created yet, or a poll raced sandbox teardown): retry. + } + if (pickups.length > 0) await Promise.allSettled(pickups); + // A safety-poll ("timeout") wake that FOUND work while the suspended watch + // claimed healthy is a watch miss (plan decision 4): it feeds demotion. + if ( + sawNew && + source?.suspendsPolling && + source.isHealthy() && + lastWaitOutcome === "timeout" + ) { + source.noteMiss?.(); + } + idlePolls = sawNew ? 0 : idlePolls + 1; + if (source && source.isHealthy() && source.suspendsPolling) { + // Healthy remote watch: the watch exec's completion is the wake; the remote + // poll is suspended and only the 30 s safety poll remains (plan decision 6). + lastWaitOutcome = await boundedWait(source, RELAY_SAFETY_POLL_MS); + } else if (source && source.isHealthy()) { + // Local watch: shortens the sleep only; the poll cadence is unchanged. + lastWaitOutcome = await boundedWait( + source, + relayPollDelayMs(idlePolls), + ); + } else { + // Classic loop, byte for byte (no source, demoted, or closed). + await sleep(relayPollDelayMs(idlePolls)); + lastWaitOutcome = undefined; } - } catch { - // Transient (dir not created yet, or a poll raced sandbox teardown): retry. } - idlePolls = sawNew ? 0 : idlePolls + 1; - await sleep(relayPollDelayMs(idlePolls)); + } finally { + // No timer or watcher may outlive the loop, however it exits. + source?.close(); } await Promise.allSettled(inflight); })(); return { + ready, stop: async () => { active = false; + // Close before awaiting the loop so a held 30 s safety-poll wait resolves + // ("closed") immediately instead of pinning stop() on its timer. + source?.close(); await loop.catch(() => {}); }, }; diff --git a/services/runner/tests/unit/relay-atomic-publish.test.ts b/services/runner/tests/unit/relay-atomic-publish.test.ts new file mode 100644 index 0000000000..a29eae5473 --- /dev/null +++ b/services/runner/tests/unit/relay-atomic-publish.test.ts @@ -0,0 +1,211 @@ +/** + * Atomic relay publication (event-driven-tool-relay slice 1, plan decision 2). + * + * Both relay directions publish via a temp name plus a same-directory rename, so a + * reader or watcher can never observe partial JSON: the writer's request in + * tools/relay-client.ts, and the runner's response in tools/relay.ts through the + * `RelayHost.rename` capability. These tests pin the temp-name scheme's invisibility to + * the suffix filters, the absence of temp residue, the runner's write-then-rename order, + * and the Daytona host's rename -> moveFs mapping. + * + * Run: pnpm test (or: pnpm exec vitest run tests/unit/relay-atomic-publish.test.ts) + */ +import { describe, it } from "vitest"; +import assert from "node:assert/strict"; +import { + mkdtempSync, + readdirSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { publishRelayRequest } from "../../src/tools/relay-client.ts"; +import { + RELAY_REQ_SUFFIX, + RELAY_RES_SUFFIX, + relayTempPath, + sleep, +} from "../../src/tools/relay-protocol.ts"; +import { + localRelayHost, + sandboxRelayHost, + startToolRelay, + type RelayHost, +} from "../../src/tools/relay.ts"; + +const tempDir = () => mkdtempSync(join(tmpdir(), "agenta-relay-atomic-test-")); + +describe("relayTempPath", () => { + it("never matches the req/res suffix filters", () => { + for (const finalPath of [ + `/relay/call-1${RELAY_REQ_SUFFIX}`, + `/relay/call-1${RELAY_RES_SUFFIX}`, + ]) { + const tmp = relayTempPath(finalPath); + assert.ok(tmp.startsWith(`${finalPath}.tmp.`), tmp); + assert.ok( + !tmp.endsWith(RELAY_REQ_SUFFIX), + "invisible to the .req.json filter", + ); + assert.ok( + !tmp.endsWith(RELAY_RES_SUFFIX), + "invisible to the .res.json filter", + ); + } + }); + + it("uses a fresh nonce per call", () => { + const finalPath = `/relay/x${RELAY_REQ_SUFFIX}`; + assert.notEqual(relayTempPath(finalPath), relayTempPath(finalPath)); + }); +}); + +describe("publishRelayRequest (atomic request publication)", () => { + it("leaves only the final name in the dir — no *.tmp.* residue", () => { + const dir = tempDir(); + try { + publishRelayRequest(dir, { + toolName: "x", + toolCallId: "call-1", + args: { a: 1 }, + }); + assert.deepEqual(readdirSync(dir), [`call-1${RELAY_REQ_SUFFIX}`]); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +}); + +describe("startToolRelay response publication (write temp, then rename)", () => { + it("writes the full response to a non-.res.json path and renames it to .res.json", async () => { + const relayDir = "/relay"; + const reqName = `call-1${RELAY_REQ_SUFFIX}`; + const ops: Array<{ + op: "read" | "remove" | "write" | "rename"; + path: string; + extra: string; + }> = []; + // The request appears on the SECOND list: the first list feeds the stale-file + // sweep (a relay file already present there is cleared as pre-turn residue, + // never executed) — mirroring production, where the loop starts before the prompt. + let listCalls = 0; + const host: RelayHost = { + list: async () => (++listCalls === 2 ? [reqName] : []), + read: async (path) => { + ops.push({ op: "read", path, extra: "" }); + return JSON.stringify({ + toolName: "nope", + toolCallId: "call-1", + args: {}, + }); + }, + remove: async (path) => void ops.push({ op: "remove", path, extra: "" }), + write: async (path, contents) => + void ops.push({ op: "write", path, extra: contents }), + rename: async (from, to) => + void ops.push({ op: "rename", path: from, extra: to }), + }; + + // No spec named "nope" is registered, so the relay answers with an ok:false + // response — which exercises the publication path without any network. + const relay = startToolRelay(host, relayDir, [], undefined); + const deadline = Date.now() + 5000; + while (Date.now() < deadline && !ops.some((o) => o.op === "rename")) { + await sleep(10); + } + await relay.stop(); + + // Full pickup-to-publication order: read the request, remove it (delete-on-pickup, + // so the watch exec cannot insta-wake on it for the whole execution), write the + // response under a temp name, then rename it to the final name. + assert.deepEqual( + ops.map((o) => o.op), + ["read", "remove", "write", "rename"], + "read -> remove(req) -> write(tmp) -> rename(final)", + ); + const reqPath = `${relayDir}/${reqName}`; + assert.equal(ops[0].path, reqPath, "reads the request file"); + assert.equal(ops[1].path, reqPath, "removes exactly the request file"); + const [, , write, rename] = ops; + const finalResPath = `${relayDir}/call-1${RELAY_RES_SUFFIX}`; + assert.ok( + write.path.startsWith(`${finalResPath}.tmp.`), + `temp write path: ${write.path}`, + ); + assert.ok( + !write.path.endsWith(RELAY_RES_SUFFIX), + "the written path is invisible to the .res.json filter", + ); + assert.equal( + rename.path, + write.path, + "renames exactly the written temp file", + ); + assert.equal(rename.extra, finalResPath, "renames to the exact final name"); + const res = JSON.parse(write.extra); + assert.equal(res.ok, false); + assert.match(res.error, /unknown tool 'nope'/); + }); +}); + +describe("RelayHost rename implementations", () => { + it("localRelayHost.rename renames on the local filesystem", async () => { + const dir = tempDir(); + try { + const from = join(dir, "x.tmp.abc"); + const to = join(dir, `x${RELAY_RES_SUFFIX}`); + writeFileSync(from, '{"ok":true}', "utf-8"); + await localRelayHost().rename(from, to); + assert.deepEqual(readdirSync(dir), [`x${RELAY_RES_SUFFIX}`]); + assert.equal(readFileSync(to, "utf-8"), '{"ok":true}'); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("sandboxRelayHost.rename calls the daemon's moveFs with overwrite", async () => { + const calls: unknown[] = []; + const sandbox = { + moveFs: async (arg: unknown) => void calls.push(arg), + }; + await sandboxRelayHost(sandbox).rename("/relay/x.tmp.abc", "/relay/x"); + assert.deepEqual(calls, [ + { from: "/relay/x.tmp.abc", to: "/relay/x", overwrite: true }, + ]); + }); +}); + +describe("RelayHost remove implementations (delete-on-pickup)", () => { + it("localRelayHost.remove unlinks the file on the local filesystem", async () => { + const dir = tempDir(); + try { + const path = join(dir, `x${RELAY_REQ_SUFFIX}`); + writeFileSync(path, "{}", "utf-8"); + await localRelayHost().remove(path); + assert.deepEqual(readdirSync(dir), []); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("localRelayHost.remove throws on a missing file (call sites guard)", async () => { + const dir = tempDir(); + try { + await assert.rejects(localRelayHost().remove(join(dir, "missing.json"))); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("sandboxRelayHost.remove calls the daemon's deleteFsEntry", async () => { + const calls: unknown[] = []; + const sandbox = { + deleteFsEntry: async (arg: unknown) => void calls.push(arg), + }; + await sandboxRelayHost(sandbox).remove("/relay/x.req.json"); + assert.deepEqual(calls, [{ path: "/relay/x.req.json" }]); + }); +}); diff --git a/services/runner/tests/unit/relay-client.test.ts b/services/runner/tests/unit/relay-client.test.ts new file mode 100644 index 0000000000..f31dda52c8 --- /dev/null +++ b/services/runner/tests/unit/relay-client.test.ts @@ -0,0 +1,406 @@ +/** + * Unit tests for the in-sandbox relay writer client (tools/relay-client.ts) and the + * wire protocol it writes (tools/relay-protocol.ts). + * + * The golden test pins the exact request-file bytes; the rest exercise the writer's + * round-trip, error, abort, and timeout behavior against a real temp dir (no network, + * no harness — the test plays the runner side by writing the `.res.json` file). + * + * Run: pnpm test (or: pnpm exec vitest run tests/unit/relay-client.test.ts) + */ +import { describe, it } from "vitest"; +import assert from "node:assert/strict"; +import { + existsSync, + mkdtempSync, + readFileSync, + renameSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { + createRelayDirWatch, + publishRelayRequest, + relayToolCall, + waitForRelayResponse, +} from "../../src/tools/relay-client.ts"; +import { + RELAY_POLL_MS, + RELAY_REQ_SUFFIX, + RELAY_RES_SUFFIX, + relayEnvFlag, + relayTempPath, + serializeRelayRequest, + sleep, +} from "../../src/tools/relay-protocol.ts"; + +const tempDir = () => mkdtempSync(join(tmpdir(), "agenta-relay-client-test-")); + +const WATCH_FLAG = "AGENTA_AGENT_TOOLS_RELAY_RESPONSE_WATCH_ENABLED"; + +/** Publish a response file the atomic way the runner now does: temp write + rename. */ +function publishResponseAtomically(resPath: string, res: unknown): void { + const tmpPath = relayTempPath(resPath); + writeFileSync(tmpPath, JSON.stringify(res), "utf-8"); + renameSync(tmpPath, resPath); +} + +describe("serializeRelayRequest / publishRelayRequest (golden request bytes)", () => { + // These bytes are the cross-writer contract (Pi extension, local Claude loopback, + // future MCP shim per #5234); changing them breaks this golden ON PURPOSE. Key order + // is toolName, toolCallId, args; args keys keep their insertion order. + const golden = + '{"toolName":"x","toolCallId":"call-1","args":{"b":2,"a":"1"}}'; + + it("serializes the exact golden bytes", () => { + const out = serializeRelayRequest({ + toolName: "x", + toolCallId: "call-1", + args: { b: 2, a: "1" }, + }); + assert.equal(out, golden); + }); + + it("writes byte-identical content to disk", () => { + const dir = tempDir(); + try { + const { reqPath, resPath } = publishRelayRequest(dir, { + toolName: "x", + toolCallId: "call-1", + args: { b: 2, a: "1" }, + }); + assert.equal(reqPath, join(dir, `call-1${RELAY_REQ_SUFFIX}`)); + assert.equal(resPath, join(dir, `call-1${RELAY_RES_SUFFIX}`)); + assert.equal(readFileSync(reqPath, "utf-8"), golden); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("defaults missing args to an empty object", () => { + const out = serializeRelayRequest({ + toolName: "x", + toolCallId: "call-1", + args: undefined, + }); + assert.equal(out, '{"toolName":"x","toolCallId":"call-1","args":{}}'); + }); +}); + +describe("relayToolCall (writer round-trip)", () => { + it("returns the response text and deletes both files", async () => { + const dir = tempDir(); + try { + const reqPath = join(dir, `call-rt${RELAY_REQ_SUFFIX}`); + const resPath = join(dir, `call-rt${RELAY_RES_SUFFIX}`); + // Play the runner: answer shortly after the request file appears. + setTimeout(() => { + assert.ok( + existsSync(reqPath), + "request file was written before the response", + ); + writeFileSync( + resPath, + JSON.stringify({ ok: true, text: "round-trip-ok" }), + ); + }, 50); + const out = await relayToolCall(dir, "myTool", "call-rt", { a: 1 }); + assert.equal(out, "round-trip-ok"); + assert.ok(!existsSync(reqPath), "request file was cleaned up"); + assert.ok(!existsSync(resPath), "response file was cleaned up"); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("rejects with the response error message on ok:false", async () => { + const dir = tempDir(); + try { + const resPath = join(dir, `call-err${RELAY_RES_SUFFIX}`); + writeFileSync( + resPath, + JSON.stringify({ ok: false, error: "boom from runner" }), + ); + await assert.rejects( + () => relayToolCall(dir, "myTool", "call-err", {}), + /boom from runner/, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("rejects with 'aborted' on an already-aborted signal", async () => { + const dir = tempDir(); + try { + const controller = new AbortController(); + controller.abort(); + await assert.rejects( + () => + relayToolCall( + dir, + "myTool", + "call-abort", + {}, + undefined, + controller.signal, + ), + /^Error: aborted$/, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +}); + +describe("waitForRelayResponse", () => { + // relayToolCall adds +10s to any positive timeoutMs (per-tool grace), so the timeout + // path is tested here directly with a tiny deadline instead. + it("throws a timeout error when no response appears before the deadline", async () => { + const dir = tempDir(); + try { + const resPath = join(dir, `never${RELAY_RES_SUFFIX}`); + await assert.rejects( + () => waitForRelayResponse(resPath, { timeoutMs: 50 }), + /tool relay timed out/, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("a deadline-coincident response still wins: the post-loop check returns it instead of throwing", async () => { + // Pin the FINAL post-loop existsSync (fix 5): the watch is disabled so the poll + // sleep (RELAY_POLL_MS, 300 ms) cannot be cut short, and the deadline (150 ms) + // expires DURING that first sleep. The response lands mid-sleep, so neither the + // pre-loop fast path (file absent at entry) nor an in-loop check (the loop exits + // by deadline before re-checking) can explain a successful return. + const dir = tempDir(); + const previous = process.env[WATCH_FLAG]; + process.env[WATCH_FLAG] = "false"; + try { + const resPath = join(dir, `call-deadline${RELAY_RES_SUFFIX}`); + setTimeout(() => { + publishResponseAtomically(resPath, { ok: true, text: "just-in-time" }); + }, 50); + const res = await waitForRelayResponse(resPath, { + timeoutMs: Math.min(150, RELAY_POLL_MS - 50), + }); + assert.equal(res.text, "just-in-time"); + } finally { + if (previous === undefined) delete process.env[WATCH_FLAG]; + else process.env[WATCH_FLAG] = previous; + rmSync(dir, { recursive: true, force: true }); + } + }); +}); + +describe("waitForRelayResponse (hop-1 response watch)", () => { + it("a directory event wakes it well before one poll interval", async () => { + const dir = tempDir(); + try { + const resPath = join(dir, `call-watch${RELAY_RES_SUFFIX}`); + setTimeout(() => { + publishResponseAtomically(resPath, { ok: true, text: "watched" }); + }, 50); + const started = Date.now(); + const res = await waitForRelayResponse(resPath, { timeoutMs: 5000 }); + const elapsed = Date.now() - started; + assert.equal(res.ok, true); + assert.equal(res.text, "watched"); + // The response landed at ~50 ms; the plain poll would sleep a full RELAY_POLL_MS + // (300 ms) before re-checking, so finishing well under one interval proves the + // watch — not the poll — woke the waiter. The bound derives from RELAY_POLL_MS so + // the test keeps its meaning if CI ever overrides the poll cadence env. + assert.ok( + elapsed < RELAY_POLL_MS - 50, + `expected a watch wake well under RELAY_POLL_MS (${RELAY_POLL_MS} ms), took ${elapsed} ms`, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("resolves immediately when the response pre-exists (arm-before-check order)", async () => { + const dir = tempDir(); + try { + const resPath = join(dir, `call-pre${RELAY_RES_SUFFIX}`); + publishResponseAtomically(resPath, { ok: true, text: "already-there" }); + const started = Date.now(); + const res = await waitForRelayResponse(resPath, { timeoutMs: 5000 }); + assert.equal(res.text, "already-there"); + assert.ok( + Date.now() - started < 100, + "no wait at all for an existing file", + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("a pre-aborted signal still rejects 'aborted' with the watch armed", async () => { + const dir = tempDir(); + try { + const controller = new AbortController(); + controller.abort(); + await assert.rejects( + () => + waitForRelayResponse(join(dir, `call-a${RELAY_RES_SUFFIX}`), { + timeoutMs: 5000, + signal: controller.signal, + }), + /^Error: aborted$/, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("watch disabled via env: a later-arriving response is still picked up by the poll", async () => { + const dir = tempDir(); + const previous = process.env[WATCH_FLAG]; + process.env[WATCH_FLAG] = "false"; + try { + const resPath = join(dir, `call-poll${RELAY_RES_SUFFIX}`); + setTimeout(() => { + publishResponseAtomically(resPath, { ok: true, text: "polled" }); + }, 100); + const res = await waitForRelayResponse(resPath, { timeoutMs: 5000 }); + assert.equal(res.text, "polled"); + } finally { + if (previous === undefined) delete process.env[WATCH_FLAG]; + else process.env[WATCH_FLAG] = previous; + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("watch disabled via env: a pre-existing response resolves", async () => { + const dir = tempDir(); + const previous = process.env[WATCH_FLAG]; + process.env[WATCH_FLAG] = "false"; + try { + const resPath = join(dir, `call-poll-pre${RELAY_RES_SUFFIX}`); + publishResponseAtomically(resPath, { ok: true, text: "pre" }); + const res = await waitForRelayResponse(resPath, { timeoutMs: 5000 }); + assert.equal(res.text, "pre"); + } finally { + if (previous === undefined) delete process.env[WATCH_FLAG]; + else process.env[WATCH_FLAG] = previous; + rmSync(dir, { recursive: true, force: true }); + } + }); +}); + +describe("relayEnvFlag (shared call-time flag parsing)", () => { + const FLAG = "AGENTA_TEST_RELAY_FLAG"; + + it("explicit 'false'/'0' disable and 'true'/'1' enable, regardless of the default", () => { + try { + for (const value of ["false", "0"]) { + process.env[FLAG] = value; + assert.equal(relayEnvFlag(FLAG, true), false, `'${value}' disables`); + assert.equal(relayEnvFlag(FLAG, false), false); + } + for (const value of ["true", "1"]) { + process.env[FLAG] = value; + assert.equal(relayEnvFlag(FLAG, false), true, `'${value}' enables`); + assert.equal(relayEnvFlag(FLAG, true), true); + } + } finally { + delete process.env[FLAG]; + } + }); + + it("anything else (unset, empty, garbage, wrong case) falls back to the default", () => { + try { + delete process.env[FLAG]; + assert.equal(relayEnvFlag(FLAG, true), true); + assert.equal(relayEnvFlag(FLAG, false), false); + for (const value of ["", "yes", "TRUE", "False", "2"]) { + process.env[FLAG] = value; + assert.equal(relayEnvFlag(FLAG, true), true, `'${value}' -> default`); + assert.equal(relayEnvFlag(FLAG, false), false, `'${value}' -> default`); + } + } finally { + delete process.env[FLAG]; + } + }); +}); + +describe("createRelayDirWatch (coalescing invariants)", () => { + it("returns undefined when fs.watch throws (missing dir) so callers degrade to the poll", () => { + assert.equal( + createRelayDirWatch(join(tmpdir(), "agenta-relay-no-such-dir-xyz")), + undefined, + ); + }); + + it("an event with no waiter sticks: the next wait resolves 'activity', the one after times out", async () => { + const dir = tempDir(); + const dirWatch = createRelayDirWatch(dir); + try { + assert.ok(dirWatch, "watch armed on an existing dir"); + writeFileSync(join(dir, "poke.txt"), "x", "utf-8"); + await sleep(50); // let the event deliver while NO waiter is armed + assert.equal(await dirWatch.wait(50), "activity"); + assert.equal( + await dirWatch.wait(30), + "timeout", + "sticky bit consumed once", + ); + } finally { + dirWatch?.close(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("hundreds of timer-win waits accumulate no listeners and the watch still wakes after", async () => { + const dir = tempDir(); + const dirWatch = createRelayDirWatch(dir); + const warnings: Error[] = []; + const onWarning = (warning: Error) => { + if (warning.name === "MaxListenersExceededWarning") + warnings.push(warning); + }; + process.on("warning", onWarning); + try { + assert.ok(dirWatch, "watch armed on an existing dir"); + for (let i = 0; i < 200; i += 1) { + assert.equal(await dirWatch.wait(1), "timeout"); + } + assert.deepEqual( + warnings, + [], + "200 consecutive timer wins emitted no MaxListenersExceededWarning", + ); + const waiting = dirWatch.wait(2000); + writeFileSync(join(dir, "poke.txt"), "x", "utf-8"); + assert.equal( + await waiting, + "activity", + "the watch still wakes after the loop", + ); + } finally { + process.off("warning", onWarning); + dirWatch?.close(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("close() resolves an in-flight wait as a timer win and stops future waits", async () => { + const dir = tempDir(); + const dirWatch = createRelayDirWatch(dir); + try { + assert.ok(dirWatch, "watch armed on an existing dir"); + const waiting = dirWatch.wait(60_000); + dirWatch.close(); + assert.equal(await waiting, "timeout"); + assert.equal(await dirWatch.wait(1), "timeout"); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/services/runner/tests/unit/relay-loop.test.ts b/services/runner/tests/unit/relay-loop.test.ts new file mode 100644 index 0000000000..d748ba818e --- /dev/null +++ b/services/runner/tests/unit/relay-loop.test.ts @@ -0,0 +1,814 @@ +/** + * Unit tests for the startToolRelay loop's wake-source integration + * (src/tools/relay.ts + src/tools/relay-watch.ts, event-driven-tool-relay plan + * decisions 3, 4, 6): with a fake host and a fake activity source, the loop must + * + * - handle a request on a wake without waiting out a poll interval, + * - pin every suspended-mode wait to RELAY_SAFETY_POLL_MS (never relayPollDelayMs), + * - use relayPollDelayMs for a non-suspending (local) source, + * - revert to the classic sleep once the source demotes (wait never called again), + * - resolve stop() promptly out of a held wait (a wait that only resolves on close), + * - count a safety-poll discovery ("timeout" outcome, then a request found) as a miss, + * - keep the seen-set dedup across wake and poll pickups, + * - remove a request file on pickup so a slow execution cannot rearm-storm the + * Daytona watch exec (delete-on-pickup, slice-2 review finding 1), + * - still run a list pass on a "closed" wait outcome while the loop is active, + * - and stay byte-for-byte classic with no createActivitySource at all. + * + * Plus the slice-3 additions: + * + * - the stale-file sweep: every relay file (request, response, temp) already present + * when startToolRelay runs predates the turn and is swept before the discovery loop + * starts (`ready` resolves once the sweep settled); non-relay names are never + * touched; a transiently failing list is retried; a sweep whose listing never + * succeeds is skipped and the loop still serves later requests normally, + * - pickup-before-rearm (fix 3): the loop awaits read+stat+remove of every discovered + * request BEFORE its next wait, retries failed removals on later list passes, and + * skips the stat entirely without a log sink, + * - stage=relay_pickup telemetry: one log line per executed request, with pickup_ms + * from host.statMtimeMs (stat started BEFORE the delete-on-pickup remove) and the + * wake tag. + * + * Run: pnpm test (or: pnpm exec vitest run tests/unit/relay-loop.test.ts) + */ +import { describe, it } from "vitest"; +import assert from "node:assert/strict"; + +import { + RELAY_POLL_MS, + startToolRelay, + sweepStaleRelayFiles, + type RelayHost, +} from "../../src/tools/relay.ts"; +import { + RELAY_SAFETY_POLL_MS, + daytonaRelayActivitySource, + type RelayActivitySource, +} from "../../src/tools/relay-watch.ts"; + +const sleep = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + +async function until( + condition: () => boolean, + what: string, + timeoutMs = 3_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!condition()) { + if (Date.now() > deadline) assert.fail(`timed out waiting for ${what}`); + await sleep(5); + } +} + +/** In-memory relay host; reads are counted so a dedup test can pin exactly-once. */ +function fakeHost( + source?: RelayActivitySource, + opts?: { writeDelayMs?: number }, +): { + host: RelayHost; + files: Map; + readCounts: Map; + removed: string[]; +} { + const files = new Map(); + const readCounts = new Map(); + const removed: string[] = []; + const host: RelayHost = { + list: async (dir) => + [...files.keys()] + .filter((path) => path.startsWith(`${dir}/`)) + .map((path) => path.slice(dir.length + 1)), + read: async (path) => { + readCounts.set(path, (readCounts.get(path) ?? 0) + 1); + const contents = files.get(path); + if (contents === undefined) throw new Error(`missing ${path}`); + return contents; + }, + remove: async (path) => { + removed.push(path); + files.delete(path); + }, + write: async (path, contents) => { + if (opts?.writeDelayMs) await sleep(opts.writeDelayMs); + files.set(path, contents); + }, + rename: async (from, to) => { + const contents = files.get(from); + if (contents === undefined) throw new Error(`missing ${from}`); + files.delete(from); + files.set(to, contents); + }, + createActivitySource: source ? () => source : undefined, + }; + return { host, files, readCounts, removed }; +} + +/** + * Controllable activity source: waits resolve ONLY via wake()/expire()/close(). Records + * every wait's timeoutMs so the loop's cadence choice is pinned. + */ +function fakeSource(suspendsPolling: boolean): { + source: RelayActivitySource; + wake: () => void; + expire: () => void; + demote: () => void; + waitTimeouts: number[]; + missCount: () => number; +} { + let waiter: ((o: "activity" | "timeout" | "closed") => void) | undefined; + let healthy = true; + let closed = false; + let misses = 0; + const waitTimeouts: number[] = []; + const settle = (outcome: "activity" | "timeout" | "closed"): void => { + const resolve = waiter; + waiter = undefined; + resolve?.(outcome); + }; + return { + source: { + suspendsPolling, + isHealthy: () => healthy && !closed, + noteMiss: () => { + misses += 1; + }, + wait: ({ timeoutMs }) => { + waitTimeouts.push(timeoutMs); + if (closed) return Promise.resolve("closed"); + return new Promise((resolve) => { + waiter = resolve; + }); + }, + close: () => { + closed = true; + settle("closed"); + }, + }, + wake: () => settle("activity"), + expire: () => settle("timeout"), + demote: () => { + healthy = false; + }, + waitTimeouts, + missCount: () => misses, + }; +} + +const DIR = "/relay"; + +/** An unknown-tool request: the loop handles it and writes an ok:false response. */ +function putRequest(files: Map, id: string): void { + files.set( + `${DIR}/${id}.req.json`, + JSON.stringify({ toolName: "nope", toolCallId: id, args: {} }), + ); +} + +describe("startToolRelay with an activity source", () => { + it("suspended mode: requests are handled on wake, and every wait is the 30 s safety poll", async () => { + const fake = fakeSource(true); + const { host, files } = fakeHost(fake.source); + const relay = startToolRelay(host, DIR, [], undefined); + + // First pass listed (empty) and parked on the source. + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + + putRequest(files, "call-1"); + fake.wake(); + // Handled well before any poll interval could have elapsed: the wake is the pickup. + await until( + () => files.has(`${DIR}/call-1.res.json`), + "the response file", + RELAY_POLL_MS - 50, + ); + + await until(() => fake.waitTimeouts.length === 2, "the loop re-parked"); + assert.ok( + fake.waitTimeouts.every((t) => t === RELAY_SAFETY_POLL_MS), + `suspended waits pinned to RELAY_SAFETY_POLL_MS, got ${fake.waitTimeouts}`, + ); + + await relay.stop(); + }); + + it("non-suspending (local) mode: waits use the poll cadence, not the safety poll", async () => { + const fake = fakeSource(false); + const { host, files } = fakeHost(fake.source); + const relay = startToolRelay(host, DIR, [], undefined); + + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + assert.equal( + fake.waitTimeouts[0], + RELAY_POLL_MS, + "local watch only shortens the poll sleep; cadence unchanged", + ); + + putRequest(files, "call-1"); + fake.wake(); + await until(() => files.has(`${DIR}/call-1.res.json`), "the response"); + + await relay.stop(); + }); + + it("a demoted source reverts the loop to the classic sleep (wait never called again)", async () => { + const fake = fakeSource(true); + const { host, files } = fakeHost(fake.source); + const relay = startToolRelay(host, DIR, [], undefined); + + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + fake.demote(); + fake.expire(); // release the held wait; the next iteration re-checks isHealthy + + // Classic mode still picks requests up (by poll), and never consults the source. + const waitsAtDemotion = fake.waitTimeouts.length; + putRequest(files, "call-1"); + await until(() => files.has(`${DIR}/call-1.res.json`), "poll pickup"); + await sleep(RELAY_POLL_MS + 100); + assert.equal( + fake.waitTimeouts.length, + waitsAtDemotion, + "no wait() after demotion", + ); + + await relay.stop(); + }); + + it("stop() during a held wait returns promptly (close resolves the wait)", async () => { + const fake = fakeSource(true); + const { host } = fakeHost(fake.source); + const relay = startToolRelay(host, DIR, [], undefined); + + await until(() => fake.waitTimeouts.length === 1, "the held wait"); + const start = Date.now(); + await relay.stop(); + assert.ok( + Date.now() - start < 1_000, + "stop() did not sit out the 30 s safety-poll timer", + ); + }); + + it("a safety-poll discovery (timeout outcome, then a request found) counts as a watch miss", async () => { + const fake = fakeSource(true); + const { host, files } = fakeHost(fake.source); + const relay = startToolRelay(host, DIR, [], undefined); + + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + // The request lands, but the watch never wakes: the safety poll finds it. + putRequest(files, "call-1"); + fake.expire(); + await until(() => files.has(`${DIR}/call-1.res.json`), "safety pickup"); + assert.equal(fake.missCount(), 1, "the miss fed noteMiss"); + + // A request found after a real wake is NOT a miss. + await until(() => fake.waitTimeouts.length === 2, "re-parked"); + putRequest(files, "call-2"); + fake.wake(); + await until(() => files.has(`${DIR}/call-2.res.json`), "wake pickup"); + assert.equal(fake.missCount(), 1, "an 'activity' pickup is not a miss"); + + await relay.stop(); + }); + + it("seen-set dedup holds across wake and poll pickups of the same file", async () => { + const fake = fakeSource(true); + const { host, files, readCounts } = fakeHost(fake.source); + const relay = startToolRelay(host, DIR, [], undefined); + + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + putRequest(files, "call-1"); + fake.wake(); + await until(() => files.has(`${DIR}/call-1.res.json`), "first pickup"); + + // Delete-on-pickup already removed the file, and even if a stale list still + // returned it, the seen set dedups: never re-read on later wakes and expiries. + await until(() => fake.waitTimeouts.length === 2, "re-parked"); + fake.wake(); + await until(() => fake.waitTimeouts.length === 3, "re-parked again"); + fake.expire(); + await until(() => fake.waitTimeouts.length === 4, "and once more"); + assert.equal(readCounts.get(`${DIR}/call-1.req.json`), 1); + + await relay.stop(); + }); + + it("delete-on-pickup: a slow execution does not rearm-storm the Daytona watch exec", async () => { + // End-to-end against the REAL daytonaRelayActivitySource: the fake daemon exec + // insta-completes while any *.req.json is present (exactly what the in-sandbox + // watch script does), parks forever otherwise. Before delete-on-pickup, the + // request file stayed on disk for the whole execution, so every window + // insta-completed and rearmed at network speed; now the pickup removes it and the + // window count stays small across a slow (~150 ms) execution. + const files = new Map(); + const removed: string[] = []; + let runProcessCalls = 0; + let testDone = false; + const sandbox = { + runProcess: (_request: { + command: string; + args: string[]; + timeoutMs: number; + }): Promise<{ exitCode?: number | null; timedOut?: boolean }> => { + runProcessCalls += 1; + // Mirrors the in-sandbox watch script's readdir interval: the window + // completes as soon as ANY *.req.json is present, including one that lands + // MID-window. (Since the stale-file sweep, the request must be written after + // the loop starts, so it can no longer be guaranteed present at arm time.) + // `testDone` drains an idle parked window at the end so no timer chain + // outlives the test. + return new Promise((resolve) => { + const check = (): void => { + const hasReq = [...files.keys()].some((path) => + path.endsWith(".req.json"), + ); + if (hasReq || testDone) { + resolve({ exitCode: 0, timedOut: false }); + return; + } + setTimeout(check, 5); + }; + check(); + }); + }, + }; + const source = daytonaRelayActivitySource(sandbox, DIR, { + windowMs: 60_000, + }); + const host: RelayHost = { + list: async (dir) => + [...files.keys()] + .filter((path) => path.startsWith(`${dir}/`)) + .map((path) => path.slice(dir.length + 1)), + read: async (path) => { + const contents = files.get(path); + if (contents === undefined) throw new Error(`missing ${path}`); + return contents; + }, + remove: async (path) => { + removed.push(path); + files.delete(path); + }, + // The slow part of the execution: the response write takes ~150 ms, so the + // request's pickup-to-response span covers many potential rearm windows. + write: async (path, contents) => { + await sleep(150); + files.set(path, contents); + }, + rename: async (from, to) => { + const contents = files.get(from); + if (contents === undefined) throw new Error(`missing ${from}`); + files.delete(from); + files.set(to, contents); + }, + createActivitySource: () => source, + }; + + const relay = startToolRelay(host, DIR, [], undefined); + // Written AFTER the loop starts: the stale-file sweep (slice 3) treats any relay + // file already present when startToolRelay runs as pre-turn residue and removes + // it. The sweep's listing is taken synchronously inside startToolRelay (this + // host's list body is synchronous), so this write is reliably post-sweep — + // exactly how a real request arrives (the loop starts before the prompt is + // issued). + putRequest(files, "call-1"); + await until(() => files.has(`${DIR}/call-1.res.json`), "the response"); + + assert.ok( + runProcessCalls <= 3, + `watch exec count stayed small across the execution, got ${runProcessCalls}`, + ); + assert.deepEqual( + removed, + [`${DIR}/call-1.req.json`], + "pickup removed exactly the request file", + ); + assert.ok( + !files.has(`${DIR}/call-1.req.json`), + "the request file is gone from the fake fs, so later lists never return it", + ); + + testDone = true; + await relay.stop(); + }); + + it("a 'closed' wait outcome while the loop is active still runs a list pass", async () => { + const fake = fakeSource(true); + const { host, files } = fakeHost(fake.source); + const relay = startToolRelay(host, DIR, [], undefined); + + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + // The request lands and the source closes itself (e.g. the watch subsystem tears + // down) while the loop is still active: the "closed" outcome must not skip the + // list pass — the request is still picked up and answered. + putRequest(files, "call-1"); + fake.source.close(); + await until( + () => files.has(`${DIR}/call-1.res.json`), + "post-closed pickup", + ); + + await relay.stop(); + }); + + it("no createActivitySource: the classic loop still serves requests (bare-host parity)", async () => { + const { host, files } = fakeHost(undefined); + const relay = startToolRelay(host, DIR, [], undefined); + + // Post-sweep write (see the delete-on-pickup test): a request seeded before + // startToolRelay would now be cleared as pre-turn residue instead of served. + putRequest(files, "call-1"); + await until(() => files.has(`${DIR}/call-1.res.json`), "classic pickup"); + const res = JSON.parse(files.get(`${DIR}/call-1.res.json`) ?? "{}"); + assert.equal(res.ok, false); + assert.match(res.error ?? "", /unknown tool/); + + await relay.stop(); + }); +}); + +/** startToolRelay with only the trailing log option set (positional args unchanged). */ +function startRelayWithLog( + host: RelayHost, + logs: string[], +): { ready: Promise; stop: () => Promise } { + return startToolRelay( + host, + DIR, + [], + undefined, + undefined, + undefined, + undefined, + { log: (msg) => logs.push(msg) }, + ); +} + +describe("startToolRelay stale-file sweep (a turn only executes requests created after it started)", () => { + it("sweeps pre-existing request, response, AND temp names; never other names; ready resolves after the sweep", async () => { + const logs: string[] = []; + const { host, files, readCounts } = fakeHost(undefined); + // Residue of a crashed prior turn in a reused relay dir (warm continuation skips + // the cold-build rm -rf): a request, a stale RESPONSE (a resumed approval reuses + // its original toolCallId, so a leftover res file would satisfy the new wait + // instantly with stale bytes), and both directions' atomic-publication temp + // names. Pi's usage file shares the dir and must survive. + putRequest(files, "stale-1"); + files.set(`${DIR}/stale-1.res.json`, '{"ok":true,"text":"stale"}'); + files.set(`${DIR}/stale-2.req.json.tmp.abc123`, "{"); + files.set(`${DIR}/stale-2.res.json.tmp.def456`, "{"); + files.set(`${DIR}/pi-usage.json`, '{"tokens":1}'); + + const relay = startRelayWithLog(host, logs); + await relay.ready; + + assert.ok(!files.has(`${DIR}/stale-1.req.json`), "request swept"); + assert.ok(!files.has(`${DIR}/stale-1.res.json`), "stale response swept"); + assert.ok( + !files.has(`${DIR}/stale-2.req.json.tmp.abc123`), + "req temp swept", + ); + assert.ok( + !files.has(`${DIR}/stale-2.res.json.tmp.def456`), + "res temp swept", + ); + assert.ok(files.has(`${DIR}/pi-usage.json`), "non-relay file untouched"); + assert.equal( + readCounts.get(`${DIR}/stale-1.req.json`), + undefined, + "the stale request was never read (no execution path started)", + ); + + // A request written after `ready` (the engine holds prompt/respondPermission on + // it) is legitimate and served normally. + putRequest(files, "fresh-1"); + await until(() => files.has(`${DIR}/fresh-1.res.json`), "fresh response"); + assert.ok( + !files.has(`${DIR}/stale-1.res.json`), + "no response was resurrected for the stale request", + ); + + await relay.stop(); + }); + + it("a transiently failing list is retried: the residue is swept on the retry", async () => { + const logs: string[] = []; + const { host, files, readCounts, removed } = fakeHost(undefined); + const baseList = host.list; + let listCalls = 0; + // The dir is transiently unlistable when the sweep starts: the FIRST list + // rejects; the residue is only visible from the second attempt. + host.list = async (dir) => { + listCalls += 1; + if (listCalls === 1) throw new Error("relay dir not created yet"); + return baseList(dir); + }; + putRequest(files, "stale-1"); + const relay = startRelayWithLog(host, logs); + + await relay.ready; + assert.ok( + removed.includes(`${DIR}/stale-1.req.json`), + "the retry attempt swept the residue", + ); + assert.ok(listCalls >= 2, "the rejecting list was retried"); + assert.equal( + readCounts.get(`${DIR}/stale-1.req.json`), + undefined, + "still treated as stale: never executed", + ); + + await relay.stop(); + }); + + it("all list attempts fail: the sweep is skipped once, and the loop still serves later requests normally", async () => { + const logs: string[] = []; + const { host, files } = fakeHost(undefined); + const baseList = host.list; + let listCalls = 0; + // All 3 sweep attempts reject; the loop's own lists then succeed. + host.list = async (dir) => { + listCalls += 1; + if (listCalls <= 3) throw new Error("dir unlistable"); + return baseList(dir); + }; + const relay = startRelayWithLog(host, logs); + await relay.ready; + assert.equal(listCalls, 3, "exactly 3 sweep attempts"); + assert.deepEqual( + logs.filter((msg) => msg.includes("stale sweep skipped")), + ["[relay] stale sweep skipped: relay dir unlistable after 3 attempts"], + "one skip line", + ); + + // A request arriving after the failed sweep is NOT treated as stale: the loop + // serves it normally (the sweep never defers behind live traffic). + putRequest(files, "fresh-1"); + await until(() => files.has(`${DIR}/fresh-1.res.json`), "fresh response"); + + await relay.stop(); + }); + + it("the stale log line fires once, with the count", async () => { + const logs: string[] = []; + const { host, files, removed } = fakeHost(undefined); + putRequest(files, "stale-1"); + putRequest(files, "stale-2"); + const relay = startRelayWithLog(host, logs); + + await until(() => removed.length === 2, "both stale removals"); + // Serve one real request so several more list passes have run by the assertion: + // the stale line must not repeat on later (post-sweep) passes. + putRequest(files, "fresh-1"); + await until(() => files.has(`${DIR}/fresh-1.res.json`), "fresh response"); + + assert.deepEqual( + logs.filter((msg) => msg.includes("stale relay file")), + ["[relay] cleared 2 stale relay file(s) predating the turn"], + "exactly one stale line, carrying the count", + ); + + await relay.stop(); + }); + + it("sweepStaleRelayFiles removals settle before it resolves; non-relay names survive", async () => { + const { host, files, removed } = fakeHost(undefined); + putRequest(files, "stale-1"); + putRequest(files, "stale-2"); + files.set(`${DIR}/keep.txt`, "x"); + const logs: string[] = []; + await sweepStaleRelayFiles(host, DIR, (msg) => logs.push(msg)); + assert.deepEqual( + removed.sort(), + [`${DIR}/stale-1.req.json`, `${DIR}/stale-2.req.json`], + "both removed by the time the sweep resolves", + ); + assert.ok(files.has(`${DIR}/keep.txt`), "non-relay name untouched"); + assert.equal(logs.length, 1); + }); +}); + +describe("startToolRelay pickup-before-rearm (fix 3)", () => { + it("the loop does not re-wait (so no watch window can arm) until the pickup's remove settled", async () => { + const fake = fakeSource(true); + const files = new Map(); + let readCount = 0; + let releaseRemove: (() => void) | undefined; + const removeGate = new Promise((resolve) => { + releaseRemove = resolve; + }); + const host: RelayHost = { + list: async (dir) => + [...files.keys()] + .filter((path) => path.startsWith(`${dir}/`)) + .map((path) => path.slice(dir.length + 1)), + read: async (path) => { + readCount += 1; + const contents = files.get(path); + if (contents === undefined) throw new Error(`missing ${path}`); + return contents; + }, + remove: async (path) => { + await removeGate; + files.delete(path); + }, + write: async (path, contents) => { + files.set(path, contents); + }, + rename: async (from, to) => { + const contents = files.get(from); + if (contents === undefined) throw new Error(`missing ${from}`); + files.delete(from); + files.set(to, contents); + }, + createActivitySource: () => fake.source, + }; + + const relay = startToolRelay(host, DIR, [], undefined); + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + putRequest(files, "call-1"); + fake.wake(); + + // The request is read (execution can start), but the remove is still pending: + // the loop must NOT have armed another wait — a watch exec issued now would + // insta-complete on the still-present request file. + await until(() => readCount === 1, "the pickup read"); + await sleep(50); + assert.equal( + fake.waitTimeouts.length, + 1, + "no second wait while the pickup remove is pending", + ); + + releaseRemove?.(); + await until(() => fake.waitTimeouts.length === 2, "re-parked after pickup"); + await until(() => files.has(`${DIR}/call-1.res.json`), "the response"); + + await relay.stop(); + }); + + it("a failed delete-on-pickup remove is retried on later list passes until gone from the listing", async () => { + const { host, files, readCounts } = fakeHost(undefined); + const baseRemove = host.remove; + let removeAttempts = 0; + host.remove = async (path) => { + removeAttempts += 1; + if (removeAttempts === 1) throw new Error("EBUSY"); + return baseRemove(path); + }; + const relay = startToolRelay(host, DIR, [], undefined); + await relay.ready; + + putRequest(files, "call-1"); + await until(() => files.has(`${DIR}/call-1.res.json`), "the response"); + await until( + () => !files.has(`${DIR}/call-1.req.json`), + "the retried removal", + ); + assert.ok(removeAttempts >= 2, "the failed remove was retried"); + assert.equal( + readCounts.get(`${DIR}/call-1.req.json`), + 1, + "retries never re-execute (seen-set dedup holds)", + ); + + await relay.stop(); + }); +}); + +describe("startToolRelay stage=relay_pickup telemetry", () => { + /** In-memory host that records the per-path op order and answers statMtimeMs. */ + function telemetryHost(mtimeMs: (path: string) => number | undefined): { + host: RelayHost; + files: Map; + ops: Array<{ op: "read" | "stat" | "remove"; path: string }>; + } { + const files = new Map(); + const ops: Array<{ op: "read" | "stat" | "remove"; path: string }> = []; + const host: RelayHost = { + list: async (dir) => + [...files.keys()] + .filter((path) => path.startsWith(`${dir}/`)) + .map((path) => path.slice(dir.length + 1)), + read: async (path) => { + ops.push({ op: "read", path }); + const contents = files.get(path); + if (contents === undefined) throw new Error(`missing ${path}`); + return contents; + }, + remove: async (path) => { + ops.push({ op: "remove", path }); + files.delete(path); + }, + write: async (path, contents) => { + files.set(path, contents); + }, + rename: async (from, to) => { + const contents = files.get(from); + if (contents === undefined) throw new Error(`missing ${from}`); + files.delete(from); + files.set(to, contents); + }, + statMtimeMs: async (path) => { + ops.push({ op: "stat", path }); + return mtimeMs(path); + }, + }; + return { host, files, ops }; + } + + it("logs one pickup line per executed request with pickup_ms >= 0 and the wake tag; stat runs BEFORE remove", async () => { + const logs: string[] = []; + const { host, files, ops } = telemetryHost(() => Date.now() - 50); + const relay = startRelayWithLog(host, logs); + + putRequest(files, "call-1"); + await until(() => files.has(`${DIR}/call-1.res.json`), "first response"); + putRequest(files, "call-2"); + await until(() => files.has(`${DIR}/call-2.res.json`), "second response"); + await relay.stop(); + + const pickups = logs.filter((msg) => msg.includes("stage=relay_pickup")); + assert.equal( + pickups.length, + 2, + `one pickup line per executed request, got ${JSON.stringify(pickups)}`, + ); + const match = + /^\[relay\] stage=relay_pickup id=call-1 pickup_ms=(\d+) wake=poll$/.exec( + pickups[0], + ); + assert.ok(match, `pickup line shape: ${pickups[0]}`); + assert.ok(Number(match?.[1]) >= 0, "a plausible non-negative pickup_ms"); + + // Order pin: the stat must run BEFORE the delete-on-pickup remove — afterwards + // the file is gone and the stat could only miss. + const reqPath = `${DIR}/call-1.req.json`; + assert.deepEqual( + ops.filter((op) => op.path === reqPath).map((op) => op.op), + ["read", "stat", "remove"], + "read, then stat, then remove", + ); + }); + + it("no log sink -> no stat at all (no daemon round-trip when nothing consumes pickup_ms)", async () => { + const { host, files, ops } = telemetryHost(() => Date.now()); + // No opts.log: the telemetry gate must skip host.statMtimeMs entirely. + const relay = startToolRelay(host, DIR, [], undefined); + + putRequest(files, "call-1"); + await until(() => files.has(`${DIR}/call-1.res.json`), "the response"); + await relay.stop(); + + assert.deepEqual( + ops.filter((op) => op.op === "stat"), + [], + "statMtimeMs never called without a log sink", + ); + }); + + it("a throwing statMtimeMs degrades to pickup_ms=-1 and never fails the execution", async () => { + const logs: string[] = []; + const { host, files } = telemetryHost(() => { + throw new Error("stat failed"); + }); + const relay = startRelayWithLog(host, logs); + + putRequest(files, "call-1"); + await until(() => files.has(`${DIR}/call-1.res.json`), "the response"); + await relay.stop(); + + const pickup = logs.find((msg) => msg.includes("stage=relay_pickup")); + assert.ok(pickup?.includes("pickup_ms=-1"), `degraded line: ${pickup}`); + assert.ok(pickup?.includes("wake="), "the wake tag is still present"); + }); + + it("a wake-driven pickup carries wake=activity (captured at discovery time); no statMtimeMs -> -1", async () => { + const fake = fakeSource(true); + const logs: string[] = []; + // fakeHost has NO statMtimeMs: the optional capability degrades to pickup_ms=-1. + const { host, files } = fakeHost(fake.source); + const relay = startToolRelay( + host, + DIR, + [], + undefined, + undefined, + undefined, + undefined, + { log: (msg) => logs.push(msg) }, + ); + + await until(() => fake.waitTimeouts.length === 1, "the first wait"); + putRequest(files, "call-1"); + fake.wake(); + await until(() => files.has(`${DIR}/call-1.res.json`), "wake pickup"); + + const pickup = logs.find((msg) => + msg.includes("stage=relay_pickup id=call-1"), + ); + assert.ok(pickup?.includes("wake=activity"), `wake tag: ${pickup}`); + assert.ok(pickup?.includes("pickup_ms=-1"), `no-stat fallback: ${pickup}`); + + await relay.stop(); + }); +}); diff --git a/services/runner/tests/unit/relay-watch.test.ts b/services/runner/tests/unit/relay-watch.test.ts new file mode 100644 index 0000000000..c0cd53be48 --- /dev/null +++ b/services/runner/tests/unit/relay-watch.test.ts @@ -0,0 +1,782 @@ +/** + * Unit tests for the hop 2 wake sources (src/tools/relay-watch.ts), pinning the plan's + * activity-source invariants (event-driven-tool-relay plan, decisions 3, 4, 7): + * + * - The Daytona window loop: single-flight (at most ONE concurrent runProcess, ever), + * sticky coalescing, close-abandons-the-window with no unhandled rejection, demotion + * after exactly 3 consecutive failures with backoff-gated rearm, wake/failure + * classification per completion shape (rejected / nullish result / timedOut / + * nonzero-null-or-missing exit / exit 0), the runner-side outer bound abandoning a + * never-settling exec, the deferred arm inside a wait that starts during a backoff + * gap, and the mid-wait re-arm after a failure lands while a waiter is parked. + * - Window config parsing and clamping, and the downward-only (−20%..0%) jitter bounds. + * - The script args builder: the relay dir rides argv only, never the script text. + * - The watch script AS A REAL PROCESS (execFile, argv, no shell) against temp dirs. + * - The local fs.watch adapter. + * + * Run: pnpm test (or: pnpm exec vitest run tests/unit/relay-watch.test.ts) + */ +import { afterEach, describe, it } from "vitest"; +import assert from "node:assert/strict"; +import { execFile } from "node:child_process"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { + RELAY_REMOTE_WATCH_WINDOW_DEFAULT_MS, + RELAY_REMOTE_WATCH_WINDOW_MAX_MS, + RELAY_REMOTE_WATCH_WINDOW_MIN_MS, + RELAY_SAFETY_POLL_MS, + RELAY_WATCH_SCRIPT, + applyRelayWatchJitter, + buildRelayWatchScriptArgs, + daytonaRelayActivitySource, + localRelayActivitySource, + remoteWatchEnabled, + resolveRemoteWatchWindowMs, +} from "../../src/tools/relay-watch.ts"; + +const WINDOW_ENV = "AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_WINDOW_MS"; +const ENABLED_ENV = "AGENTA_AGENT_TOOLS_RELAY_REMOTE_WATCH_ENABLED"; + +const sleep = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); +/** Let the source's runProcess .then/.catch handlers run after a fake settles. */ +const tick = (): Promise => + new Promise((resolve) => setImmediate(resolve)); + +interface FakeExec { + request: { command: string; args: string[]; timeoutMs: number }; + resolve: (result: { exitCode?: number | null; timedOut?: boolean }) => void; + reject: (err: unknown) => void; +} + +/** Controllable runProcess fake that counts concurrency (pins single-flight). */ +function fakeSandbox(): { + sandbox: { + runProcess: (request: { + command: string; + args: string[]; + timeoutMs: number; + }) => Promise<{ exitCode?: number | null; timedOut?: boolean }>; + }; + calls: FakeExec[]; + maxConcurrent: () => number; +} { + const calls: FakeExec[] = []; + let concurrent = 0; + let max = 0; + return { + sandbox: { + runProcess: (request) => { + concurrent += 1; + max = Math.max(max, concurrent); + return new Promise((resolve, reject) => { + calls.push({ + request, + resolve: (result) => { + concurrent -= 1; + resolve(result); + }, + reject: (err) => { + concurrent -= 1; + reject(err); + }, + }); + }); + }, + }, + calls, + maxConcurrent: () => max, + }; +} + +afterEach(() => { + delete process.env[WINDOW_ENV]; + delete process.env[ENABLED_ENV]; +}); + +describe("daytonaRelayActivitySource invariants", () => { + it("coalesces: a window completing with no waiter resolves the NEXT wait immediately; the one after times out", async () => { + const { sandbox, calls } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + }); + + // Arm a window via a wait that times out before the window completes. + const w1 = await source.wait({ timeoutMs: 20 }); + assert.equal(w1, "timeout"); + assert.equal(calls.length, 1); + + // The window completes while NO waiter is present: sticky bit. + calls[0].resolve({ exitCode: 0, timedOut: false }); + await tick(); + + const w2 = await source.wait({ timeoutMs: 20 }); + assert.equal(w2, "activity", "sticky wake consumed"); + assert.equal(calls.length, 1, "sticky consumption did not arm a window"); + + const w3 = await source.wait({ timeoutMs: 20 }); + assert.equal(w3, "timeout", "the sticky bit was consumed exactly once"); + assert.equal(calls.length, 2, "the next wait armed a fresh window"); + + source.close(); + calls[1].resolve({ exitCode: 0 }); + await tick(); + }); + + it("single-flight: at most ONE concurrent runProcess ever, across many waits and completions", async () => { + const { sandbox, calls, maxConcurrent } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + }); + + for (let i = 0; i < 25; i += 1) { + const wait = source.wait({ timeoutMs: 5_000 }); + assert.equal(calls.length, i + 1, "each idle wait arms exactly one"); + calls[i].resolve({ exitCode: 0, timedOut: false }); + assert.equal(await wait, "activity"); + await tick(); + } + // Waits while a window is already in flight never arm a second exec. + const held = source.wait({ timeoutMs: 10 }); + assert.equal(calls.length, 26); + assert.equal(await held, "timeout"); + const heldAgain = source.wait({ timeoutMs: 10 }); + assert.equal(calls.length, 26, "in-flight window: no second exec"); + assert.equal(await heldAgain, "timeout"); + + assert.equal(maxConcurrent(), 1); + source.close(); + calls[25].resolve({ exitCode: 0 }); + await tick(); + }); + + it("close() during a held window resolves 'closed'; the abandoned exec rejection stays handled", async () => { + const { sandbox, calls } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + }); + + const held = source.wait({ timeoutMs: 60_000 }); + assert.equal(calls.length, 1); + source.close(); + assert.equal(await held, "closed"); + assert.equal(source.isHealthy(), false, "closed is not healthy"); + assert.equal(await source.wait({ timeoutMs: 10 }), "closed"); + + // Abandon: the in-flight exec cannot be aborted (no per-call signal in the SDK); + // its late rejection must never become an unhandled rejection. + const unhandled: unknown[] = []; + const trap = (reason: unknown): void => { + unhandled.push(reason); + }; + process.on("unhandledRejection", trap); + try { + calls[0].reject(new Error("daemon went away")); + await sleep(20); + } finally { + process.off("unhandledRejection", trap); + } + assert.equal(unhandled.length, 0); + }); + + it("repeated rejection demotes after exactly 3, with ONE demotion log line; a rejecting exec never rejects wait()", async () => { + const { sandbox, calls } = fakeSandbox(); + const logs: string[] = []; + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 1, + backoffCapMs: 2, + log: (msg) => logs.push(msg), + }); + + for (let failure = 1; failure <= 3; failure += 1) { + const wait = source.wait({ timeoutMs: 30 }); + assert.equal(calls.length, failure, "rearm after backoff"); + calls[failure - 1].reject(new Error("no exec for you")); + // A rejection is NOT a wake: the waiter's own timer resolves "timeout". + assert.equal(await wait, "timeout"); + assert.equal(source.isHealthy(), failure < 3); + await sleep(10); // past the tiny injected backoff + } + + assert.equal(source.isHealthy(), false, "demoted after exactly 3"); + const demotions = logs.filter((line) => line.includes("demoted")); + assert.equal(demotions.length, 1, "exactly one demotion log line"); + + // Demoted: waits never arm again. + assert.equal(await source.wait({ timeoutMs: 10 }), "timeout"); + assert.equal(calls.length, 3); + source.close(); + }); + + it("backoff gates rearm: a wait inside the backoff window does not arm", async () => { + const { sandbox, calls } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 200, + backoffCapMs: 400, + }); + + const w1 = source.wait({ timeoutMs: 20 }); + calls[0].reject(new Error("boom")); + assert.equal(await w1, "timeout"); + + // Still inside the ~200 ms (jittered >= 160 ms) backoff: no new exec. + assert.equal(await source.wait({ timeoutMs: 20 }), "timeout"); + assert.equal(calls.length, 1, "no rearm inside the backoff window"); + + await sleep(300); + const w3 = source.wait({ timeoutMs: 1_000 }); + assert.equal(calls.length, 2, "rearm after the backoff elapsed"); + calls[1].resolve({ exitCode: 0 }); + assert.equal(await w3, "activity"); + source.close(); + }); + + it("a nonzero exit is a wake AND a failure: three in a row wake the waiter each time, then demote", async () => { + const { sandbox, calls } = fakeSandbox(); + const logs: string[] = []; + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 1, + backoffCapMs: 2, + log: (msg) => logs.push(msg), + }); + + for (let i = 0; i < 3; i += 1) { + const wait = source.wait({ timeoutMs: 5_000 }); + calls[i].resolve({ exitCode: 1, timedOut: false }); + assert.equal(await wait, "activity", "abnormal exit still wakes"); + await sleep(10); // past the tiny injected backoff (every failure backs off now) + } + assert.equal(source.isHealthy(), false, "three abnormal exits demote"); + // The demotion line repeats the reason, so count only per-failure lines. + assert.equal( + logs.filter((l) => l.includes("watch failure: watch script exited with")) + .length, + 3, + ); + assert.equal(logs.filter((l) => l.includes("demoted")).length, 1); + source.close(); + }); + + it("a null exit (signal-killed / OOM) is a wake AND a failure, never a success reset", async () => { + const { sandbox, calls } = fakeSandbox(); + const logs: string[] = []; + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 1, + backoffCapMs: 2, + log: (msg) => logs.push(msg), + }); + + for (let i = 0; i < 3; i += 1) { + const wait = source.wait({ timeoutMs: 5_000 }); + calls[i].resolve({ exitCode: null, timedOut: false }); + assert.equal(await wait, "activity", "a killed script still wakes"); + await sleep(10); + } + assert.equal(source.isHealthy(), false, "three null exits demote"); + assert.equal( + logs.filter((l) => + l.includes("watch failure: watch script exited with code null"), + ).length, + 3, + ); + source.close(); + }); + + it("a nullish runProcess RESULT is a failure with NO wake, never a success reset; three demote", async () => { + // A broken daemon path can insta-resolve runProcess with undefined. If that read + // as success, the counter would pin at 0 and the source would storm windows + // forever; it must count as a failure and feed demotion instead. + const { sandbox, calls } = fakeSandbox(); + const logs: string[] = []; + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 1, + backoffCapMs: 2, + log: (msg) => logs.push(msg), + }); + + for (let i = 0; i < 3; i += 1) { + const wait = source.wait({ timeoutMs: 30 }); + calls[i].resolve(undefined as unknown as { exitCode?: number | null }); + assert.equal(await wait, "timeout", "a nullish result never wakes"); + await sleep(10); + } + assert.equal(source.isHealthy(), false, "three nullish results demote"); + assert.equal( + logs.filter((l) => + l.includes("watch failure: exec resolved with no result"), + ).length, + 3, + ); + assert.equal(logs.filter((l) => l.includes("demoted")).length, 1); + source.close(); + }); + + it("timedOut:true is a failure with NO wake; a later success resets the counter", async () => { + const { sandbox, calls } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 1, + backoffCapMs: 2, + }); + + // Two daemon timeouts: failures, no sticky wake left behind. + for (let i = 0; i < 2; i += 1) { + const wait = source.wait({ timeoutMs: 30 }); + calls[i].resolve({ exitCode: null, timedOut: true }); + assert.equal(await wait, "timeout", "a daemon timeout never wakes"); + await sleep(10); + } + assert.equal(source.isHealthy(), true, "two failures < threshold"); + + // A success resets the counter... + const w3 = source.wait({ timeoutMs: 5_000 }); + calls[2].resolve({ exitCode: 0, timedOut: false }); + assert.equal(await w3, "activity"); + await tick(); + + // ...so two MORE failures still do not demote. + for (let i = 3; i < 5; i += 1) { + const wait = source.wait({ timeoutMs: 30 }); + calls[i].resolve({ timedOut: true }); + assert.equal(await wait, "timeout"); + await sleep(10); + } + assert.equal(source.isHealthy(), true, "the success reset the counter"); + source.close(); + }); + + it("noteMiss() feeds the same demotion counter", () => { + const { sandbox } = fakeSandbox(); + const logs: string[] = []; + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + log: (msg) => logs.push(msg), + }); + + source.noteMiss?.(); + source.noteMiss?.(); + assert.equal(source.isHealthy(), true); + source.noteMiss?.(); + assert.equal(source.isHealthy(), false); + assert.equal(logs.filter((l) => l.includes("demoted")).length, 1); + // Idempotent after demotion: no second demotion line. + source.noteMiss?.(); + assert.equal(logs.filter((l) => l.includes("demoted")).length, 1); + source.close(); + }); + + it("outer bound: a never-settling exec counts as a failure, re-arms inside the SAME wait, demotes after three, and late settles change nothing", async () => { + const settlers: Array<{ + resolve: (r: { exitCode?: number | null; timedOut?: boolean }) => void; + reject: (err: unknown) => void; + }> = []; + const sandbox = { + runProcess: (): Promise<{ + exitCode?: number | null; + timedOut?: boolean; + }> => + new Promise((resolve, reject) => settlers.push({ resolve, reject })), + }; + const logs: string[] = []; + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 40, + graceMs: 20, + outerBoundMarginMs: 20, + backoffBaseMs: 1, + backoffCapMs: 2, + log: (msg) => logs.push(msg), + }); + + // ONE long wait: each outer bound (jittered window + grace + margin, ~72-80 ms + // here) abandons the never-settling exec, counts a failure with NO wake, and the + // mid-wait re-arm (fix 4) arms the next window after the tiny backoff — all + // inside this same parked wait, with no safety-poll help. Three abandoned + // windows demote the source; demotion stops the chain at exactly three. + const wait = source.wait({ timeoutMs: 600 }); + assert.equal(settlers.length, 1, "wait entry armed the first window"); + assert.equal(await wait, "timeout", "outer-bound expiry never wakes"); + assert.equal(settlers.length, 3, "three windows chained, then demotion"); + assert.equal(source.isHealthy(), false, "demoted after three"); + assert.equal( + logs.filter((l) => l.includes("watch failure: window outer bound")) + .length, + 3, + ); + assert.equal(logs.filter((l) => l.includes("demoted")).length, 1); + + // Late settles from the abandoned execs are IGNORED entirely: no wake, no counter + // mutation, no rearm, and a late rejection never becomes unhandled. + const unhandled: unknown[] = []; + const trap = (reason: unknown): void => void unhandled.push(reason); + process.on("unhandledRejection", trap); + try { + settlers[0].resolve({ exitCode: 0, timedOut: false }); + settlers[1].reject(new Error("daemon woke up late")); + settlers[2].resolve({ exitCode: 1, timedOut: false }); + await sleep(20); + } finally { + process.off("unhandledRejection", trap); + } + assert.equal(unhandled.length, 0, "late rejection stays handled"); + assert.equal(source.isHealthy(), false, "still demoted"); + assert.equal( + await source.wait({ timeoutMs: 10 }), + "timeout", + "no sticky wake from a late settle", + ); + assert.equal(settlers.length, 3, "no window armed by a late settle"); + source.close(); + }); + + it("deferred arm: a wait that starts inside a backoff gap arms when the gap ends, within that same wait", async () => { + const { sandbox, calls } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 100, + backoffCapMs: 100, + }); + + const w1 = source.wait({ timeoutMs: 20 }); + calls[0].reject(new Error("boom")); + assert.equal(await w1, "timeout"); + + // A long wait starting inside the ~80-100 ms backoff gap: the gate blocks the + // immediate arm, but the deferred arm fires inside this SAME wait once the gap + // ends — the wait is not left windowless for its whole timeout. + const w2 = source.wait({ timeoutMs: 2_000 }); + assert.equal(calls.length, 1, "the backoff gate blocked the immediate arm"); + await sleep(300); // well past the <= 100 ms backoff, with load headroom + assert.equal(calls.length, 2, "the deferred arm fired inside the wait"); + calls[1].resolve({ exitCode: 0, timedOut: false }); + assert.equal(await w2, "activity", "the deferred window wakes the wait"); + source.close(); + }); + + it("a failure landing MID-wait re-arms after the backoff inside that same wait (fix 4)", async () => { + // The gap this closes: an exec that fails 1 s into a 30 s safety wait used to + // leave the source windowless until the wait's own timer — no deferred arm was + // scheduled because the wait STARTED with a window in flight, outside any backoff + // gap. Now countFailure re-enters the arm gate while a waiter is parked. + const { sandbox, calls } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 60_000, + backoffBaseMs: 50, + backoffCapMs: 50, + }); + + const wait = source.wait({ timeoutMs: 2_000 }); + assert.equal(calls.length, 1, "wait entry armed a window"); + calls[0].reject(new Error("exec died mid-wait")); + await tick(); + assert.equal( + calls.length, + 1, + "no instant rearm: the backoff gates the next window", + ); + + await sleep(200); // well past the <= 50 ms jittered backoff + assert.equal( + calls.length, + 2, + "the failure re-armed inside the SAME parked wait", + ); + calls[1].resolve({ exitCode: 0, timedOut: false }); + assert.equal(await wait, "activity", "the re-armed window wakes the wait"); + source.close(); + }); + + it("suspendsPolling is true and the daemon timeoutMs carries the grace over the jittered window", async () => { + const { sandbox, calls } = fakeSandbox(); + const source = daytonaRelayActivitySource(sandbox, "/relay", { + windowMs: 10_000, + }); + assert.equal(source.suspendsPolling, true); + + const wait = source.wait({ timeoutMs: 10 }); + const request = calls[0].request; + const scriptWindow = Number(request.args[3]); + assert.ok( + scriptWindow >= 8_000 && scriptWindow <= 10_000, + "downward-only jitter", + ); + assert.equal( + request.timeoutMs, + scriptWindow + 5_000, + "daemon bound = jittered window + 5 s grace", + ); + assert.equal(await wait, "timeout"); + source.close(); + calls[0].resolve({ exitCode: 0 }); + await tick(); + }); +}); + +describe("remote watch window config", () => { + it("unset -> default, no warning", () => { + const logs: string[] = []; + delete process.env[WINDOW_ENV]; + assert.equal( + resolveRemoteWatchWindowMs((m) => logs.push(m)), + RELAY_REMOTE_WATCH_WINDOW_DEFAULT_MS, + ); + assert.equal(logs.length, 0); + }); + + it("garbage -> default with one warning", () => { + const logs: string[] = []; + process.env[WINDOW_ENV] = "soon"; + assert.equal( + resolveRemoteWatchWindowMs((m) => logs.push(m)), + RELAY_REMOTE_WATCH_WINDOW_DEFAULT_MS, + ); + assert.equal(logs.length, 1); + assert.match(logs[0], /unparseable/); + }); + + it("1000 -> clamped to the 5000 floor with one warning", () => { + const logs: string[] = []; + process.env[WINDOW_ENV] = "1000"; + assert.equal( + resolveRemoteWatchWindowMs((m) => logs.push(m)), + RELAY_REMOTE_WATCH_WINDOW_MIN_MS, + ); + assert.equal(logs.length, 1); + assert.match(logs[0], /clamped/); + }); + + it("500000 -> clamped to the 120000 ceiling with one warning", () => { + const logs: string[] = []; + process.env[WINDOW_ENV] = "500000"; + assert.equal( + resolveRemoteWatchWindowMs((m) => logs.push(m)), + RELAY_REMOTE_WATCH_WINDOW_MAX_MS, + ); + assert.equal(logs.length, 1); + assert.match(logs[0], /clamped/); + }); + + it("an in-range value passes through unclamped and unwarned", () => { + const logs: string[] = []; + process.env[WINDOW_ENV] = "30000"; + assert.equal( + resolveRemoteWatchWindowMs((m) => logs.push(m)), + 30_000, + ); + assert.equal(logs.length, 0); + }); + + it("jitter is downward-only (−20%..0%) across many draws", () => { + // Deliberate deviation from the plan's ±20%: an upward draw would let a jittered + // 25 s window outlast the 30 s safety wait and read as a false watch miss. + let min = Infinity; + let max = -Infinity; + for (let i = 0; i < 1_000; i += 1) { + const drawn = applyRelayWatchJitter(10_000); + min = Math.min(min, drawn); + max = Math.max(max, drawn); + } + assert.ok(min >= 8_000, `min ${min} >= 8000`); + assert.ok( + max <= 10_000, + `max ${max} <= 10000 (never above the nominal window)`, + ); + assert.ok(max > min, "jitter actually varies"); + }); + + it("remoteWatchEnabled defaults false; only 'true'/'1' enable", () => { + delete process.env[ENABLED_ENV]; + assert.equal(remoteWatchEnabled(), false); + for (const value of ["false", "0", "yes", "TRUE", ""]) { + process.env[ENABLED_ENV] = value; + assert.equal(remoteWatchEnabled(), false, `'${value}' must not enable`); + } + for (const value of ["true", "1"]) { + process.env[ENABLED_ENV] = value; + assert.equal(remoteWatchEnabled(), true, `'${value}' enables`); + } + }); + + it("the safety poll constant is 30 s", () => { + assert.equal(RELAY_SAFETY_POLL_MS, 30_000); + }); +}); + +describe("buildRelayWatchScriptArgs", () => { + const METACHAR_DIR = "we ird\"'`$(rm -rf /);|&\ndir"; + + it("the script text never contains the dir; the dir rides argv as one element", () => { + const { command, args } = buildRelayWatchScriptArgs( + METACHAR_DIR, + 25_000, + 2_000, + ); + assert.equal(command, "node"); + assert.deepEqual(args, [ + "-e", + RELAY_WATCH_SCRIPT, + METACHAR_DIR, + "25000", + "2000", + ]); + assert.ok( + !RELAY_WATCH_SCRIPT.includes(METACHAR_DIR), + "dir never interpolated into the script source", + ); + }); + + it("the script has no process.exit and writes nothing to stdout", () => { + assert.ok(!RELAY_WATCH_SCRIPT.includes("process.exit")); + assert.ok(!RELAY_WATCH_SCRIPT.includes("console.")); + assert.ok(!RELAY_WATCH_SCRIPT.includes("stdout")); + }); +}); + +describe("the watch script as a real process", () => { + function runScript( + dir: string, + windowMs: number, + readdirPollMs = 100, + ): Promise<{ exitCode: number; durationMs: number }> { + const { command, args } = buildRelayWatchScriptArgs( + dir, + windowMs, + readdirPollMs, + ); + const start = Date.now(); + return new Promise((resolve) => { + // argv, no shell: exactly how sandbox.runProcess passes it. + execFile(command, args, (error) => { + resolve({ + exitCode: + error && typeof error.code === "number" + ? error.code + : error + ? 1 + : 0, + durationMs: Date.now() - start, + }); + }); + }); + } + + let dir: string | undefined; + afterEach(() => { + if (dir) rmSync(dir, { recursive: true, force: true }); + dir = undefined; + }); + + it("a pre-existing .req.json exits immediately (exit 0)", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-watch-")); + writeFileSync(join(dir, "call-1.req.json"), "{}"); + const { exitCode, durationMs } = await runScript(dir, 1_500); + assert.equal(exitCode, 0); + assert.ok( + durationMs < 1_000, + `exited in ${durationMs}ms, before the window`, + ); + }); + + it("a file created after spawn exits promptly on the watch event", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-watch-")); + const target = dir; + // Long window and slow readdir poll: only the fs.watch event can explain a fast exit. + const running = runScript(target, 4_000, 3_500); + setTimeout(() => { + writeFileSync(join(target, "call-2.req.json"), "{}"); + }, 100); + const { exitCode, durationMs } = await running; + assert.equal(exitCode, 0); + assert.ok( + durationMs < 2_000, + `woke at ${durationMs}ms, well under the window`, + ); + }); + + it("no file: exits at the window bound (exit 0)", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-watch-")); + const { exitCode, durationMs } = await runScript(dir, 1_200); + assert.equal(exitCode, 0); + assert.ok(durationMs >= 1_100, `held the full window (${durationMs}ms)`); + assert.ok(durationMs < 4_000, "and exited at the bound, not later"); + }); + + it("a metacharacter-rich dir path still wakes on the event", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-watch-")); + const weird = join(dir, "we ird\"'`$(rm -rf x);|&\ndir"); + mkdirSync(weird); + const running = runScript(weird, 4_000, 3_500); + setTimeout(() => { + writeFileSync(join(weird, "call-3.req.json"), "{}"); + }, 100); + const { exitCode, durationMs } = await running; + assert.equal(exitCode, 0); + assert.ok(durationMs < 2_000, `woke at ${durationMs}ms in the weird dir`); + }); + + it("a nonexistent dir does not crash; it degrades and exits at the window bound", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-watch-")); + const missing = join(dir, "never-created"); + const { exitCode, durationMs } = await runScript(missing, 1_200); + assert.equal(exitCode, 0); + assert.ok( + durationMs >= 1_100, + `degraded poll held the window (${durationMs}ms)`, + ); + }); +}); + +describe("localRelayActivitySource", () => { + let dir: string | undefined; + afterEach(() => { + if (dir) rmSync(dir, { recursive: true, force: true }); + dir = undefined; + }); + + it("a missing dir returns undefined (plain poll fallback)", () => { + assert.equal(localRelayActivitySource("/definitely/not/a/dir"), undefined); + }); + + it("an event during a wait resolves 'activity'; it does not suspend polling", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-local-")); + const source = localRelayActivitySource(dir); + assert.ok(source); + assert.equal(source.suspendsPolling, false); + assert.equal(source.isHealthy(), true); + + const target = dir; + const wait = source.wait({ timeoutMs: 5_000 }); + setTimeout(() => { + writeFileSync(join(target, "call-1.req.json"), "{}"); + }, 30); + assert.equal(await wait, "activity"); + source.close(); + }); + + it("no event resolves 'timeout' at the deadline", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-local-")); + const source = localRelayActivitySource(dir); + assert.ok(source); + assert.equal(await source.wait({ timeoutMs: 30 }), "timeout"); + source.close(); + }); + + it("close() resolves a held wait 'closed' and later waits 'closed'; isHealthy flips false", async () => { + dir = mkdtempSync(join(tmpdir(), "agenta-relay-local-")); + const source = localRelayActivitySource(dir); + assert.ok(source); + + const held = source.wait({ timeoutMs: 60_000 }); + source.close(); + assert.equal(await held, "closed"); + assert.equal(source.isHealthy(), false); + assert.equal(await source.wait({ timeoutMs: 10 }), "closed"); + }); +}); diff --git a/services/runner/tests/unit/sandbox-agent-orchestration.test.ts b/services/runner/tests/unit/sandbox-agent-orchestration.test.ts index e88c16fc44..0538d90bcb 100644 --- a/services/runner/tests/unit/sandbox-agent-orchestration.test.ts +++ b/services/runner/tests/unit/sandbox-agent-orchestration.test.ts @@ -707,6 +707,12 @@ describe("runSandboxAgent orchestration", () => { // A Pi run passes the execution guard: the relay dir is sandbox-writable, so every execute // record is re-checked runner-side (a forged record must not run an ask/deny tool). assert.equal(typeof calls.toolRelayArgs?.[6], "function"); + // The 8th argument carries the log sink: without it the relay skips pickup + // telemetry (and its per-request stat) entirely, so the engine must pass one. + assert.equal( + typeof (calls.toolRelayArgs?.[7] as { log?: unknown } | undefined)?.log, + "function", + ); assert.equal( calls.toolRelayStops, 2, diff --git a/services/runner/tests/unit/sandbox-agent-pi-assets.test.ts b/services/runner/tests/unit/sandbox-agent-pi-assets.test.ts index 1357a397fc..662f66a324 100644 --- a/services/runner/tests/unit/sandbox-agent-pi-assets.test.ts +++ b/services/runner/tests/unit/sandbox-agent-pi-assets.test.ts @@ -235,6 +235,35 @@ describe("buildPiExtensionEnv", () => { ); }); + describe("hop-1 response-watch kill switch forwarding", () => { + const FLAG = "AGENTA_AGENT_TOOLS_RELAY_RESPONSE_WATCH_ENABLED"; + const previous = process.env[FLAG]; + const relayRequest = { + customTools: [{ name: "safe_tool", kind: "callback" }], + } as AgentRunRequest; + + afterEach(() => { + if (previous === undefined) delete process.env[FLAG]; + else process.env[FLAG] = previous; + }); + + it("forwards the flag verbatim into the sandbox env when the operator set it", () => { + process.env[FLAG] = "false"; + const env = buildPiExtensionEnv(relayRequest, false, { + relayDir: "/tmp/relay", + }); + assert.equal(env[FLAG], "false"); + }); + + it("omits the flag when the operator did not set it (writer defaults to true)", () => { + delete process.env[FLAG]; + const env = buildPiExtensionEnv(relayRequest, false, { + relayDir: "/tmp/relay", + }); + assert.equal(env[FLAG], undefined); + }); + }); + it("never leaks the bearer into env when no auth file path is given", () => { const env = buildPiExtensionEnv( { diff --git a/services/runner/tests/unit/tool-callref-bindings.test.ts b/services/runner/tests/unit/tool-callref-bindings.test.ts index bdff826da5..0c83fd6523 100644 --- a/services/runner/tests/unit/tool-callref-bindings.test.ts +++ b/services/runner/tests/unit/tool-callref-bindings.test.ts @@ -62,6 +62,16 @@ async function relayOnce(input: { const dir = mkdtempSync(join(tmpdir(), "agenta-callref-relay-")); try { const id = "call-1"; + const relay = startToolRelay( + localRelayHost(), + dir, + [input.spec], + { endpoint: ENDPOINT, authorization: "ApiKey secret" }, + input.runContext, + ); + // Written AFTER startToolRelay: the stale-file sweep (whose listing is taken + // synchronously inside startToolRelay for this synchronous-list host) clears any + // request already present as pre-turn residue instead of executing it. writeFileSync( join(dir, `${id}.req.json`), JSON.stringify({ @@ -70,13 +80,6 @@ async function relayOnce(input: { args: input.args, }), ); - const relay = startToolRelay( - localRelayHost(), - dir, - [input.spec], - { endpoint: ENDPOINT, authorization: "ApiKey secret" }, - input.runContext, - ); const resPath = join(dir, `${id}.res.json`); const deadline = Date.now() + 5000; while (Date.now() < deadline && !existsSync(resPath)) { diff --git a/services/runner/tests/unit/tool-direct.test.ts b/services/runner/tests/unit/tool-direct.test.ts index 3b3f4bd5e2..7645c2f6c3 100644 --- a/services/runner/tests/unit/tool-direct.test.ts +++ b/services/runner/tests/unit/tool-direct.test.ts @@ -570,10 +570,6 @@ async function relayOnce( const dir = mkdtempSync(join(tmpdir(), "agenta-direct-relay-")); try { const id = "call-1"; - writeFileSync( - join(dir, `${id}.req.json`), - JSON.stringify({ toolName: spec.name, toolCallId: id, args }), - ); const relay = startToolRelay( localRelayHost(), dir, @@ -581,6 +577,13 @@ async function relayOnce( callback, runContext, ); + // Written AFTER startToolRelay: the stale-file sweep (whose listing is taken + // synchronously inside startToolRelay for this synchronous-list host) clears any + // request already present as pre-turn residue instead of executing it. + writeFileSync( + join(dir, `${id}.req.json`), + JSON.stringify({ toolName: spec.name, toolCallId: id, args }), + ); const resPath = join(dir, `${id}.res.json`); const deadline = Date.now() + 5000; while (Date.now() < deadline && !existsSync(resPath)) { diff --git a/services/runner/tests/unit/tool-relay-guard.test.ts b/services/runner/tests/unit/tool-relay-guard.test.ts index 7f1788e156..51d42310b4 100644 --- a/services/runner/tests/unit/tool-relay-guard.test.ts +++ b/services/runner/tests/unit/tool-relay-guard.test.ts @@ -116,14 +116,6 @@ async function relayOnce(input: { const dir = mkdtempSync(join(tmpdir(), "agenta-relay-guard-")); try { const id = "call-1"; - writeFileSync( - join(dir, `${id}.req.json`), - JSON.stringify({ - toolName: input.spec.name, - toolCallId: id, - args: input.args, - }), - ); const relay = startToolRelay( localRelayHost(), dir, @@ -133,6 +125,18 @@ async function relayOnce(input: { undefined, input.guard, ); + // Written AFTER startToolRelay: the stale-file sweep (whose listing is taken + // synchronously inside startToolRelay for this synchronous-list host) clears any + // request already present as pre-turn residue instead of executing it. A real + // forged record can also only appear after the loop starts. + writeFileSync( + join(dir, `${id}.req.json`), + JSON.stringify({ + toolName: input.spec.name, + toolCallId: id, + args: input.args, + }), + ); const resPath = join(dir, `${id}.res.json`); const deadline = Date.now() + 5000; while (Date.now() < deadline && !existsSync(resPath)) {