feat(runner): event-driven tool relay (fs.watch + Daytona long-poll watch, polling fallback)

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 (`<final>.tmp.<nonce>`) 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` |

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/` |

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 `<id>.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
-`<id>.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 `<id>.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
+`<id>.res.json` response. Every publication is atomic on both sides: full bytes under a temp
+name (`<final>.tmp.<nonce>`), 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.

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");

services/runner/src/engines/sandbox_agent.ts

@@ -418,7 +418,7 @@ const RUN_LIMIT_TRIPPED = Symbol("run-limit-tripped");
 interface CurrentTurn {
   run: ReturnType<typeof createSandboxAgentOtel>;
   pause: PendingApprovalPauseController;
-  toolRelay?: { stop: () => Promise<void> };
+  toolRelay?: { ready?: Promise<void>; stop: () => Promise<void> };
   /** 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,15 +1739,24 @@ 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,
         request.toolCallback as ToolCallbackContext | undefined,
         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

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.

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<string> {
-  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.