Skip to content

Commit e0ba441

Browse files
thymikeethymikee
committed
fix(daemon): report the /rpc backstop through the runtime's injected sinks
The detached `.catch` on the /rpc flow answers a rejection no part of the flow answered. As merged it wrote its reason to `process.stderr` directly, which for a daemon started with an injected error stream is the wrong stream, and its `emitDiagnostic` had no scope to record into. Give `createDaemonHttpServer` an `onUnownedRequestFailure` option and let `openDaemonServers` wire it to the pair every other out-of-request daemon event uses: the runtime's injected `stderr` plus `emitDaemonDiagnostic`. The server now only answers, and the reason reaches a sink that exists. Normalizing the reason could itself throw — the flow's own catches normalize while evaluating their answer, so a reason whose `details` getter throws fails both and the unreadable value lands in the backstop. That second throw escaped the `.catch` and was the fatal `unhandledRejection` all over again, so the reason is normalized defensively and an unreadable one reports as the unknown-error failure. Part of #3453
1 parent 50e45f1 commit e0ba441

7 files changed

Lines changed: 353 additions & 56 deletions

‎src/__tests__/daemon-composition/server/daemon-runtime-rpc-flow-failure.test.ts‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
import fs from 'node:fs';
2+
import { afterEach, expect, test, vi } from 'vitest';
3+
import { resolveDaemonPaths } from '@agent-device/daemon-contracts/daemon-resolution';
4+
import { mkdtempForTestSync } from '@agent-device/test-kit/tmp-dir';
5+
6+
// The HTTP server hands a `/rpc` flow rejection it answered itself to `onUnownedRequestFailure`,
7+
// because that event has no request scope to be recorded in — the request's own scope opens later,
8+
// inside the handler. The runtime owns the pair every other out-of-request daemon event uses, so
9+
// this boots the real runtime and checks both halves of what `openDaemonServers` wires: the
10+
// injected error stream and the daemon log. The daemon's own HTTP route drives it, so a flow that
11+
// stops answering shows up as a refused call rather than a hung one.
12+
const requestProbe = vi.hoisted(() => ({
13+
failNextDispatch: false,
14+
}));
15+
16+
vi.mock('../../../daemon/request-router.ts', async (importOriginal) => {
17+
const actual = await importOriginal<typeof import('../../../daemon/request-router.ts')>();
18+
return {
19+
...actual,
20+
createRequestHandler: (...args: Parameters<typeof actual.createRequestHandler>) => {
21+
const dispatch = actual.createRequestHandler(...args);
22+
return async (request: Parameters<typeof dispatch>[0]) => {
23+
if (requestProbe.failNextDispatch) {
24+
requestProbe.failNextDispatch = false;
25+
throw unwrappableReason();
26+
}
27+
return await dispatch(request);
28+
};
29+
},
30+
};
31+
});
32+
33+
vi.mock('../../../provider-device-runtimes.ts', () => ({
34+
DEFAULT_PROVIDER_RUNTIME_REQUIRED_IDS: [],
35+
createDaemonProviderRuntimeComposition: async () => ({ runtimes: [], platformModules: [] }),
36+
}));
37+
38+
import { startDaemonRuntimeFixture } from '../daemon-runtime-fixture.ts';
39+
40+
afterEach(() => {
41+
requestProbe.failNextDispatch = false;
42+
});
43+
44+
test('a /rpc flow failure the server refused itself reaches the runtime stderr and daemon log', async () => {
45+
const stateDir = mkdtempForTestSync('agent-device-daemon-rpc-flow-failure-');
46+
const paths = resolveDaemonPaths(stateDir);
47+
const stderrLines: string[] = [];
48+
// Records the reason without inspecting it: this one throws on the property read an assertion
49+
// would perform.
50+
const rejections: unknown[] = [];
51+
const observeRejection = (reason: unknown) => {
52+
rejections.push(reason);
53+
};
54+
process.on('unhandledRejection', observeRejection);
55+
let exitCode: number | undefined;
56+
try {
57+
const runtime = await startDaemonRuntimeFixture({
58+
env: {
59+
...process.env,
60+
AGENT_DEVICE_STATE_DIR: stateDir,
61+
AGENT_DEVICE_DAEMON_IDLE_TIMEOUT_MS: '0',
62+
AGENT_DEVICE_DAEMON_SERVER_MODE: 'http',
63+
},
64+
exit: (code) => {
65+
exitCode = code;
66+
},
67+
registerProcessHandlers: false,
68+
stderr: { write: (chunk: string) => void stderrLines.push(chunk) },
69+
stdout: { write: () => {} },
70+
});
71+
expect(runtime).not.toBeNull();
72+
const { httpPort, token } = JSON.parse(fs.readFileSync(paths.infoPath, 'utf8')) as {
73+
httpPort: number;
74+
token: string;
75+
};
76+
77+
requestProbe.failNextDispatch = true;
78+
const refused = await postRpc(httpPort, token, {
79+
jsonrpc: '2.0',
80+
id: 'req-flow-failure',
81+
method: 'agent_device.command',
82+
params: { command: 'session_list', meta: { requestId: 'req-flow-failure' } },
83+
});
84+
expect(refused.status).toBe(500);
85+
expect(refused.body.id).toBeNull();
86+
expect(refused.body.error?.code).toBe(-32000);
87+
88+
// The refusal is answered synchronously while its record is written by the detached report the
89+
// server was handed, so the log line follows the answer by a turn.
90+
const events = await waitForDaemonLogEvent(paths.logPath, 'daemon_rpc_request_flow_failed');
91+
expect(events).toHaveLength(1);
92+
expect(events[0]?.data).toEqual({ code: 'UNKNOWN', message: 'Unknown error' });
93+
// The runtime's injected stream and not the process's: this is the channel the
94+
// `unhandledRejection` handler uses, and a daemon-level event belongs on that one.
95+
expect(stderrLines.filter((line) => line.startsWith('Daemon error:'))).toEqual([
96+
'Daemon error: Unknown error\n',
97+
]);
98+
99+
const next = await postRpc(httpPort, token, 'null');
100+
expect(next.status).toBe(400);
101+
expect(next.body.error?.code).toBe(-32600);
102+
expect(rejections).toEqual([]);
103+
expect(exitCode).toBeUndefined();
104+
} finally {
105+
process.off('unhandledRejection', observeRejection);
106+
await fs.promises.rm(stateDir, { recursive: true, force: true });
107+
}
108+
});
109+
110+
async function postRpc(
111+
httpPort: number,
112+
token: string,
113+
body: unknown,
114+
): Promise<{ status: number; body: { id?: unknown; error?: { code?: number } } }> {
115+
const response = await fetch(`http://127.0.0.1:${httpPort}/rpc`, {
116+
method: 'POST',
117+
headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` },
118+
body: typeof body === 'string' ? body : JSON.stringify(body),
119+
});
120+
return { status: response.status, body: (await response.json()) as never };
121+
}
122+
123+
async function waitForDaemonLogEvent(
124+
logPath: string,
125+
phase: string,
126+
): Promise<Array<{ phase: string; data?: Record<string, unknown> }>> {
127+
const deadline = Date.now() + 2_000;
128+
for (;;) {
129+
const events = readDaemonLogEvents(logPath);
130+
if (events.some((event) => event.phase === phase) || Date.now() >= deadline) return events;
131+
await new Promise((resolve) => setTimeout(resolve, 10));
132+
}
133+
}
134+
135+
function readDaemonLogEvents(
136+
logPath: string,
137+
): Array<{ phase: string; data?: Record<string, unknown> }> {
138+
if (!fs.existsSync(logPath)) return [];
139+
return fs
140+
.readFileSync(logPath, 'utf8')
141+
.trim()
142+
.split('\n')
143+
.filter(Boolean)
144+
.map((line) => JSON.parse(line) as { phase: string; data?: Record<string, unknown> });
145+
}
146+
147+
/**
148+
* A rejection reason that cannot be inspected: the one property `normalizeError` must read to
149+
* redact its diagnostics throws the reason itself, so the flow's own catches fail while computing
150+
* their answer and the failure reaches the runtime's sink unreadable — the shape a daemon-level
151+
* record has to survive. It stays printable, so a run without the guard reports the escaped
152+
* rejection instead of failing to format it.
153+
*/
154+
function unwrappableReason(): Record<string, unknown> {
155+
const hostile: Record<string, unknown> = {};
156+
Object.defineProperty(hostile, 'details', {
157+
enumerable: true,
158+
get(): never {
159+
throw hostile;
160+
},
161+
});
162+
return hostile;
163+
}

‎src/daemon/__tests__/http-server-rpc-handler-failure.test.ts‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -6,13 +6,12 @@ import { postRpcCommand, startRpcServer } from './http-server-rpc.fixtures.ts';
66

77
const TOKEN = 'throwing-handler-token';
88

9-
// The property this PR depends on: a `/rpc` request never leaves a rejection for the detached
10-
// `void handleRpcRequest(...)` call to drop, because the daemon exits on `unhandledRejection`.
11-
// A rejecting handler is the closest reachable case, and the answer comes from the invocation's own
12-
// catch (`invokeAdmittedRpcRequest`), which answered it the same way before this PR — so this pins
13-
// existing behavior and the invariant the new detached `.catch` defends, not that `.catch` itself.
14-
// The flow's refusal for a body that is not a call has to reach the client with the call's own id
15-
// and typed reason, and the daemon must still serve the next request.
9+
// What happens when an admitted call's own handler rejects: the invocation's catch answers it with
10+
// that call's id, its JSON-RPC -32000, and the typed reason intact, and the request lifecycle stays
11+
// usable for the call that follows. It is also the closest reachable stand-in for the invariant the
12+
// detached `.catch` in the server defends — a `/rpc` request never leaves a rejection for the
13+
// process to find, because the daemon exits on `unhandledRejection` — though the answer here comes
14+
// from the invocation, exactly as it did before that `.catch` existed.
1615
test('a rejecting handler is answered for its own call id and never reaches the process', async (t) => {
1716
if (await skipWhenLoopbackUnavailable(t)) return;
1817
const rejections: string[] = [];

‎src/daemon/__tests__/http-server-rpc-invalid-body.test.ts‎

Lines changed: 2 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
import { test } from 'vitest';
22
import assert from 'node:assert/strict';
3-
import http from 'node:http';
43
import { closeLoopbackServer, skipWhenLoopbackUnavailable } from '@agent-device/test-kit/loopback';
54
import { postRpcCommand, postRpcRaw, startRpcServer } from './http-server-rpc.fixtures.ts';
65
import type { DaemonResponse } from '../daemon-request.ts';
@@ -69,10 +68,10 @@ test('the daemon keeps serving after the bodies that never became a call', async
6968
data: {},
7069
}));
7170
try {
72-
assert.equal(await postRpcWithDeadline(port, 'null'), 400, 'the null root is answered');
71+
assert.equal((await postRpcRaw(port, TOKEN, 'null')).status, 400, 'the null root is answered');
7372
for (const root of ['{oops', '']) {
7473
assert.equal(
75-
await postRpcWithDeadline(port, root),
74+
(await postRpcRaw(port, TOKEN, root)).status,
7675
400,
7776
`${root}: a body that never parsed is answered`,
7877
);
@@ -88,27 +87,3 @@ test('the daemon keeps serving after the bodies that never became a call', async
8887
await closeLoopbackServer(server);
8988
}
9089
});
91-
92-
/** `/rpc` POST that gives up after a second: a daemon which never answers fails the caller. */
93-
async function postRpcWithDeadline(port: number, rawBody: string): Promise<number> {
94-
return await new Promise((resolve, reject) => {
95-
const req = http.request(
96-
{
97-
host: '127.0.0.1',
98-
port,
99-
method: 'POST',
100-
path: '/rpc',
101-
headers: { 'content-type': 'application/json', authorization: `Bearer ${TOKEN}` },
102-
},
103-
(res) => {
104-
res.resume();
105-
res.on('end', () => resolve(res.statusCode ?? 0));
106-
},
107-
);
108-
req.setTimeout(1_000, () => {
109-
req.destroy(new Error('the daemon never answered the request'));
110-
});
111-
req.on('error', reject);
112-
req.end(rawBody);
113-
});
114-
}

‎src/daemon/__tests__/http-server-rpc-unowned-failure.test.ts‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
import { test } from 'vitest';
2+
import assert from 'node:assert/strict';
3+
import type { NormalizedError } from '@agent-device/kernel/errors';
4+
import { closeLoopbackServer, skipWhenLoopbackUnavailable } from '@agent-device/test-kit/loopback';
5+
import { postRpcCommand, startRpcServer } from './http-server-rpc.fixtures.ts';
6+
import type { DaemonResponse } from '../daemon-request.ts';
7+
8+
const TOKEN = 'unowned-failure-token';
9+
10+
// `http-server-rpc-handler-failure.test.ts` pins a handler rejection the flow answers with the
11+
// call's own id. This covers the step behind it: the flow rejects with a reason its own catches
12+
// cannot read, so nothing upstream of the detached `.catch` has answered and no call id is safe to
13+
// echo. The refusal carries `id: null`, the failure goes to the sink the server was given rather
14+
// than to a stream this module does not own, and the daemon is still a daemon afterwards.
15+
test('a flow rejection nothing upstream answered is refused without a call id and reported', async (t) => {
16+
if (await skipWhenLoopbackUnavailable(t)) return;
17+
const rejections: unknown[] = [];
18+
// Records the reason without inspecting it: this one throws on every property read, including
19+
// the coercion an assertion would perform on it.
20+
const observeRejection = (reason: unknown) => {
21+
rejections.push(reason);
22+
};
23+
process.on('unhandledRejection', observeRejection);
24+
const reported: NormalizedError[] = [];
25+
const { port, server } = await startRpcServer(
26+
TOKEN,
27+
async (request): Promise<DaemonResponse> => {
28+
if (request.meta?.requestId?.startsWith('req-unowned')) throw unwrappableReason();
29+
return { ok: true, data: {} };
30+
},
31+
{
32+
// Throws on its first event: the sink is the caller's code, and a sink that cannot take the
33+
// event must not cost the client the answer this branch exists to give.
34+
onUnownedRequestFailure: (failure) => {
35+
reported.push(failure);
36+
if (reported.length === 1) throw new Error('the sink is down');
37+
},
38+
},
39+
);
40+
try {
41+
const answer = await postRpcCommand(port, TOKEN, 'req-unowned-failure');
42+
assert.equal(answer.status, 500, 'an unanswered flow rejection is refused with 500');
43+
assert.equal(
44+
answer.body.id,
45+
null,
46+
'a reason that cannot be read leaves no call id safe to echo',
47+
);
48+
assert.equal(answer.body.error?.code, -32000, 'refused as an internal error');
49+
assert.equal(reported.length, 1, 'the sink heard about the failure');
50+
assert.deepEqual(
51+
reported.map((failure) => [failure.code, failure.message]),
52+
[['UNKNOWN', 'Unknown error']],
53+
'an unreadable reason is reported as the unknown-error failure, never as a half-read one',
54+
);
55+
56+
// The sink throwing did not cost the caller its answer, and the next call runs normally: the
57+
// rejection left the request lifecycle as usable as it found it.
58+
const next = await postRpcCommand(port, TOKEN, 'req-after-unowned-failure');
59+
assert.equal(next.status, 200, 'the next call on the same server still runs and answers');
60+
assert.equal(rejections.length, 0, 'a flow rejection never reaches the process');
61+
} finally {
62+
process.off('unhandledRejection', observeRejection);
63+
await closeLoopbackServer(server);
64+
}
65+
});
66+
67+
/**
68+
* A rejection reason that cannot be inspected: the one property `normalizeError` must read to
69+
* redact its diagnostics throws the reason itself, so both of the flow's own catches fail while
70+
* computing their answer. That is the only way to land in the detached `.catch` a reason nothing has
71+
* answered and nothing can name. It stays printable, so a run without the fix reports the escaped
72+
* rejection instead of failing to format it.
73+
*/
74+
function unwrappableReason(): Record<string, unknown> {
75+
const hostile: Record<string, unknown> = {};
76+
Object.defineProperty(hostile, 'details', {
77+
enumerable: true,
78+
get(): never {
79+
throw hostile;
80+
},
81+
});
82+
return hostile;
83+
}

0 commit comments

Comments
 (0)