From 83bbd974ab9bf4ae59f7ce38a69434b9002e6666 Mon Sep 17 00:00:00 2001 From: Aaron Trowbridge Date: Fri, 16 Oct 2026 04:17:54 -0400 Subject: [PATCH] =?UTF-8?q?fix(hub):=20forward=20the=20engine=20child's=20?= =?UTF-8?q?stdout=20to=20the=20shard=20log=20=E2=80=94=20SSE-TICK/SELF-STA?= =?UTF-8?q?LL=20reach=20production=20(#1808)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mute sink was never the telemetry module: the service runner spawns the engine with piped stdio and consumed both pipes into a private engineLog string surfaced only on boot failure, so the default console.log sink (engine stdout) never reached the runner's stdout — the surface the systemd unit appends to the shard log. Forward every chunk as it arrives (injectable engineOutput for tests; the diagnostics buffer stays), and add the AMICODE_SSE_TICK_STALL=1 drill override that parks the /event heartbeat tap mid-fiber so the out-of-tree SELF-STALL watchdog can be rehearsed against the production wiring end-to-end. --- packages/app-bundle/manifest.json | 6 +- .../routes/instance/httpapi/handlers/event.ts | 24 ++++-- .../test/server/httpapi-event.test.ts | 79 +++++++++++++++++++ .../extension/src/amicode_service_runner.ts | 26 +++++- .../test/amicode_service_runner.test.ts | 66 ++++++++++++++++ 5 files changed, 189 insertions(+), 12 deletions(-) diff --git a/packages/app-bundle/manifest.json b/packages/app-bundle/manifest.json index ab712920b..58379c62f 100644 --- a/packages/app-bundle/manifest.json +++ b/packages/app-bundle/manifest.json @@ -417,7 +417,7 @@ "packages/opencode/test/server/gateway-secret.test.ts": "8d38ec187098be99f698c5e795b867a0e2fe5e2ef2aa3a9e2c03a8b0147d4473", "packages/opencode/test/server/httpapi-compression.test.ts": "5f7603d7018e0b95f1c2f36c01dd297c01770bfcdd91ce29778e03b7d6143f75", "packages/opencode/test/server/httpapi-event-replay.test.ts": "c94c13dbfb4ca5d5e9b80f4c1a17955d1530a8bf9141b9a83a810ee56743269f", - "packages/opencode/test/server/httpapi-event.test.ts": "36263284302434461c4c6cf3e6652e094f4a02782acbe5de26c489748fe4cb50", + "packages/opencode/test/server/httpapi-event.test.ts": "274619e3489c253b1693f4f0d42309f767fcce2074a0da406f8f9cfd27afb042", "packages/opencode/test/server/httpapi-file-write-gate.test.ts": "0c91b7417e615850edfde3f2878207dab6b6624d4cfbb6e6783b2e6d9aa4dff0", "packages/opencode/test/server/httpapi-file.test.ts": "d3386129bc81467ecbcc57a3861f53767ee56ac41b4ff6ad58d8b168fb3b3373", "packages/opencode/test/server/httpapi-instance-route-auth.test.ts": "4b73a89d2431c02bc3b665d93f9002914e51eb1add979dc1867efbcbc0c36e25", @@ -970,7 +970,7 @@ "packages/opencode/src/server/routes/instance/httpapi/groups/mcp.ts": "048efa7b74d43c4ad4a75d0f8421c808760c8c0dcd593f0051a87044e40487d6", "packages/opencode/src/server/routes/instance/httpapi/groups/provider.ts": "34c6f1ab1a085d1b047c220bbff32a36032dc5e924784e9272cda166cf0ec84b", "packages/opencode/src/server/routes/instance/httpapi/groups/session.ts": "1421f8c07948b121c7e2005bd9b0fcd6b2c456565ddd1da32d8e6c35d03fac9c", - "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "5824d237bedb868d97ca24f68da6b137b7c2d84e42111d7c17637619ce87c866", + "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "35a218c1e40081fc970ba88a6b19b56ded9d21efd6d74928342ebbb26343a7bd", "packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts": "69ca3dd48d95157fc822e1f76eb42ac873df50f22616b85194619ca658613867", "packages/opencode/src/server/routes/instance/httpapi/handlers/file.ts": "10bcb79b117beebee04184fb48c1ad789e782e2f7ae4b0c68603edc00f03aee9", "packages/opencode/src/server/routes/instance/httpapi/handlers/mcp.ts": "7da08d58765cdf2b6ce4648fc70de371402949f5bb7b75696f674d43f2d5365b", @@ -993,7 +993,7 @@ "packages/ui/src/v2/components/text-shimmer-v2.tsx" ], "exceptions": [], - "extracted_at": "2026-10-15T07:32:24.625Z", + "extracted_at": "2026-10-16T08:14:25.680Z", "per_package": { "packages/app": { "A": 251, diff --git a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts index d366bdd40..4e89af26d 100644 --- a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts +++ b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts @@ -152,6 +152,14 @@ function eventResponse(events: EventV2.Interface) { // the contract in sse-tick-telemetry.ts is the one place both must agree. const connId = eventID() sseTickTelemetry.registerConnection(EventPaths.event, connId) + // #1808 stall drill: AMICODE_SSE_TICK_STALL=1 parks the heartbeat tap + // mid-fiber BEFORE its first noteTick (Effect.never — the #1798 + // blocked-shared-emit-loop shape), so the out-of-tree SELF-STALL watchdog + // can be rehearsed against the PRODUCTION wiring: the connection stays + // registered, the wire goes heartbeat-silent, and the watchdog's line + // must ride the production sink (engine stdout → the runner's forward → + // the shard log). Ops drill only — never set on a live shard. + const tickStallDrill = process.env.AMICODE_SSE_TICK_STALL === "1" const heartbeat = Stream.tick(sseTickTelemetry.cadenceMs).pipe( Stream.drop(1), Stream.map(() => ({ id: eventID(), type: "server.heartbeat", properties: {} })), @@ -160,13 +168,15 @@ function eventResponse(events: EventV2.Interface) { // discriminates the blocked-shared-emit-loop hypothesis), alongside the // watchdog's lag and RSS samples. Stream.tap(() => - Effect.gen(function* () { - const fiberId = yield* Effect.fiberId - sseTickTelemetry.noteTick(EventPaths.event, connId, { - queueDepth: Queue.sizeUnsafe(queue), - fiberId, - }) - }), + tickStallDrill + ? Effect.never + : Effect.gen(function* () { + const fiberId = yield* Effect.fiberId + sseTickTelemetry.noteTick(EventPaths.event, connId, { + queueDepth: Queue.sizeUnsafe(queue), + fiberId, + }) + }), ), ) diff --git a/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-event.test.ts b/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-event.test.ts index 952aa3d8b..ee22f0b8e 100644 --- a/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-event.test.ts +++ b/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-event.test.ts @@ -526,3 +526,82 @@ sys.exit(1) 45_000, ) }) + +// ── #1808: the production stall drill ──────────────────────────────────────── +// +// The mute-sink postmortem (the third #1798 occurrence, 2026-10-16 03:27): +// Phase A's lines were perfect under the injected sink and invisible in +// production, so the drill must wedge the tick source through the SAME env +// override ops uses against a real boot — never through the injectable test +// sink — and assert the watchdog names the stall through whatever sink the +// production singleton carries. This suite rides the fast singleton for +// cadence only; the real-process proof (engine binary → runner forward → +// scratch log) is the gate's end-to-end half. +describe("self-stall drill (#1808 — wedged tick source, production wiring)", () => { + afterEach(() => { + // bun shares one process across test files — the drill env must not leak. + delete process.env.AMICODE_SSE_TICK_STALL + resetSseTickTelemetry() + }) + + it.instance( + "AMICODE_SSE_TICK_STALL=1 wedges the tick source: no SSE-TICK, no heartbeat on the wire, the watchdog SELF-STALLs", + () => + Effect.gen(function* () { + const lines: Array = [] + const telemetry = makeSseTickTelemetry({ + cadenceMs: 100, + watchdogIntervalMs: 300, + probeIntervalMs: 500, + sink: (line) => lines.push(line), + }) + setSseTickTelemetry(telemetry) + process.env.AMICODE_SSE_TICK_STALL = "1" + + const { directory } = yield* TestInstance + // Inner scope: the drill's connection lives only as long as the + // assertions need it live, and its finalizer unregisters against the + // fast singleton INSIDE the test — bun shares one process across + // files, so a leaked registration or raw timer must not outlive it. + yield* Effect.gen(function* () { + const { response, reader } = yield* openEventStream(directory) + expect(response.status).toBe(200) + expect(yield* readEvent(reader)).toMatchObject({ type: "server.connected", properties: {} }) + + // The wedge parks the heartbeat tap BEFORE its first noteTick: no + // heartbeat frame ever reaches the wire — the silent-recycle shape + // (connection open, frontdoor cadence telemetry sees zero ticks). + const frame = yield* Queue.take(reader).pipe( + Effect.as("frame" as const), + Effect.timeoutOrElse({ duration: "250 millis", orElse: () => Effect.succeed("silent" as const) }), + ) + expect(frame).toBe("silent") + + // The out-of-tree watchdog names the stall while the tap fiber + // sits parked and the connection stays registered: a full check + // window with zero ticks from the live connection. + yield* Effect.sleep("700 millis") + const stalls = lines.filter((line) => line.startsWith("SELF-STALL")) + expect(stalls.length).toBeGreaterThanOrEqual(1) + expect(stalls[0]).toContain("seen=0") + expect(stalls[0]).toContain("route=/event") + expect(lines.some((line) => line.startsWith("SSE-TICK"))).toBe(false) + }).pipe(Effect.scoped) + + // The finalizer is async — poll until the registry drains, then stop + // the drill telemetry's raw timers (watchdog + probe) for good. + yield* Effect.forEach( + [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19], + () => + Effect.gen(function* () { + if (telemetry.snapshot().connections > 0) yield* Effect.sleep("100 millis") + }), + { discard: true }, + ) + expect(telemetry.snapshot().connections).toBe(0) + telemetry.stop() + }), + { git: true, config: { formatter: false, lsp: false } }, + 10_000, + ) +}) diff --git a/packages/extension/src/amicode_service_runner.ts b/packages/extension/src/amicode_service_runner.ts index 5575a35fa..f48e761cb 100644 --- a/packages/extension/src/amicode_service_runner.ts +++ b/packages/extension/src/amicode_service_runner.ts @@ -134,6 +134,15 @@ export interface AmicodeServiceRunnerOptions { * ships without a real catalog; only a truly unresolvable shipped dir leaves * skills.paths omitted (suppression still applies). */ machineSkillRoot?: string; + /** #1808: where the engine child's stdout/stderr chunks are forwarded AS + * THEY ARRIVE. Default: the runner's own stdout/stderr — the systemd unit + * appends those to the shard log, and the engine's telemetry lines + * (SSE-TICK / SSE-DROP / the SELF-STALL watchdog) ride its stdout. Before + * #1808 the pipes were consumed into the private engineLog buffer and only + * surfaced on boot failure — every Phase A line was mute in production while + * its wire heartbeats demonstrably flowed. The buffer is kept regardless + * (health-failure diagnostics); forwarding is fire-and-forget. */ + engineOutput?: (chunk: string, stream: "stdout" | "stderr") => void; /** Log sink (the structural-interface convention — vscode-free). */ log?: (line: string) => void; } @@ -466,9 +475,22 @@ export async function bootAmicodeServiceRunner(opts: AmicodeServiceRunnerOptions env: engineEnv, stdio: ["ignore", "pipe", "pipe"], }); + // #1808: the engine's stdout is a telemetry surface, not a private buffer. + // Forward every chunk to the runner's own stdio (which the systemd unit + // appends to the shard log) as it arrives — SSE-TICK / SSE-DROP / SELF-STALL + // lines must be greppable on a LIVE engine, not only in a failed boot's + // post-mortem. The engineLog buffer stays: it is the boot-failure diagnostic. + const forwardEngineOutput = + opts.engineOutput ?? ((chunk: string, stream: "stdout" | "stderr") => (stream === "stderr" ? process.stderr : process.stdout).write(chunk)); let engineLog = ""; - engine.stdout?.on("data", (d: Buffer) => (engineLog += d)); - engine.stderr?.on("data", (d: Buffer) => (engineLog += d)); + engine.stdout?.on("data", (d: Buffer) => { + engineLog += d; + forwardEngineOutput(d.toString(), "stdout"); + }); + engine.stderr?.on("data", (d: Buffer) => { + engineLog += d; + forwardEngineOutput(d.toString(), "stderr"); + }); // ── PID-file write (#1578): record the engine child's PID ─────────────── if (opts.pidFile && engine.pid !== undefined) { diff --git a/packages/extension/test/amicode_service_runner.test.ts b/packages/extension/test/amicode_service_runner.test.ts index 99fc7761c..77c93e091 100644 --- a/packages/extension/test/amicode_service_runner.test.ts +++ b/packages/extension/test/amicode_service_runner.test.ts @@ -511,6 +511,72 @@ createServer((_req, res) => { }, 30_000); }); +// ── #1808: the engine child's output is a telemetry surface, not a private buffer ── +// +// The mute-sink root cause: the runner spawned the engine with piped stdio and +// buffered both pipes into a private string surfaced only on boot failure — so +// the engine's SSE-TICK / SSE-DROP / SELF-STALL lines (console.log → engine +// stdout) never reached the runner's stdout, which is what the systemd unit +// appends to the shard log. The runner must FORWARD every chunk as it arrives. +describe("engine output forwarding (#1808 — the production telemetry sink)", () => { + const boots: AmicodeServiceRunnerBoot[] = []; + afterAll(async () => { + for (const b of boots.splice(0)) await b.shutdown().catch(() => undefined); + }); + + /** A fake engine that prints telemetry-shaped lines on BOTH pipes after it + * starts listening — the child-side half of the forwarding contract. */ + function writeTelemetryEngine(dir: string): string { + const bin = join(dir, "fake-engine"); + writeFileSync( + bin, + `#!/usr/bin/env node +const { createServer } = require("node:http"); +const port = Number(process.argv[4] ?? 0); +createServer((_req, res) => { + res.writeHead(200, { "content-type": "text/plain" }); + res.end("fake engine up"); +}).listen(port, "127.0.0.1", () => { + console.log("SSE-TICK route=/event conn=drill cadence_ms=10000 queue_depth=0 fiber=7 lag_ms=0 rss_bytes=42 ts_ms=123"); + console.error("SELF-STALL ticks_expected=1 seen=0 conns=1 route=/event since_last_tick_ms=120 last_queue_depth=0 lag_ms=0 rss_bytes=42 ts_ms=124"); +}); +`, + ); + chmodSync(bin, 0o755); + return bin; + } + + function writeStubShelf(dir: string): string { + writeFileSync(join(dir, "index.html"), "stub shelf"); + return dir; + } + + it("forwards the engine child's stdout AND stderr chunks to the runner's output surface as they arrive", async () => { + const dir = mkdtempSync(join(tmpdir(), "amicode-runner-tick-fwd-")); + const forwarded: Array<{ chunk: string; stream: string }> = []; + const boot = await bootAmicodeServiceRunner({ + engineBin: writeTelemetryEngine(dir), + appDistRoot: writeStubShelf(mkdtempSync(join(tmpdir(), "amicode-runner-tick-shelf-"))), + engineUnarmed: true, + healthTimeoutMs: 10_000, + servicePort: 0, + enginePort: 0, + log: () => undefined, + engineOutput: (chunk, stream) => forwarded.push({ chunk, stream }), + }); + boots.push(boot); + + // Chunk boundaries may split lines — join per stream and wait for both. + const joined = (stream: string) => + forwarded.filter((f) => f.stream === stream).map((f) => f.chunk).join(""); + expect( + await waitUntil(() => joined("stdout").includes("SSE-TICK route=/event") && joined("stderr").includes("SELF-STALL")), + ).toBe(true); + expect(joined("stdout")).toContain("cadence_ms=10000"); + expect(joined("stderr")).toContain("seen=0"); + }, 30_000); +}); + describe("amicode service runner (fail-loud, headless — no engine needed)", () => { it("a missing engine binary fails with the named reason BEFORE spawning anything", async () => { const err = await bootAmicodeServiceRunner({