fix(daemon): fence timeout recovery at ownership and transport seams

Review findings on #3193:

- The reset deleted daemon.json and daemon.lock unconditionally after a
  probe window during which a replacement daemon can publish. Re-read the
  registration through readRegisteredDaemonOwnership and clear it only on
  `match`; a replacement's record survives. The protocol lock is no longer
  touched at all: ADR 0030 gives reclaim to the acquirer under its mutation
  guard, so an out-of-band delete is the legacy-reclaimer pattern.
- The probe's detached HTTP build folded a malformed port into a negative
  answer instead of an unhandled rejection.
- The HTTP error listener returns when the timeout already claimed the
  outcome, so a destroyed request no longer also diagnoses a transport
  failure. The claim stays with the guarded reject, so a genuine socket
  death still settles.
- The reset hint stops claiming runner children stopped with a daemon killed
  by pid alone.
- Route tests count TCP connections instead of requests, seed ownership
  shaped registrations and a protocol lock dir, and assert no transport
  failure rides along with a timeout. Probe tests pin the /health path,
  gate the socket answer on the HTTP leg's receipt, and cover the malformed
  port.

Author: thymikee

src/daemon-client/__tests__/daemon-client-lifecycle.test.ts

@@ -613,10 +613,15 @@ test('sendRequest timeout cleanup uses resolved daemon paths instead of request
   const daemonPaths = resolveDaemonPaths(daemonStateDir);
   const requestFlagPaths = resolveDaemonPaths(requestFlagStateDir);
   const daemon = await startHangingHttpDaemonFixture();
+  // The cleanup fence only clears a registration that still MATCHES the timed-out daemon on pid
+  // and start time (#3125), so this fixture names one identity on both sides: the record it
+  // publishes and the info `sendRequest` times out on.
+  const timedOutDaemonStartTime = 'timeout-daemon-start';
   writeDaemonInfo(daemonPaths, {
     httpPort: daemon.port,
     transport: 'http',
     pid: 999_999,
+    processStartTime: timedOutDaemonStartTime,
   });
   writeDaemonLock(daemonPaths, { pid: 999_999 });
   writeDaemonInfo(requestFlagPaths, {
@@ -644,6 +649,7 @@ test('sendRequest timeout cleanup uses resolved daemon paths instead of request
           pid: 999_999,
           httpPort: daemon.port,
           transport: 'http',
+          processStartTime: timedOutDaemonStartTime,
         },
         request,
         'http',
@@ -660,7 +666,9 @@ test('sendRequest timeout cleanup uses resolved daemon paths instead of request
     // reset. The fixture refused that probe, which is what authorized the reset below.
     assert.deepEqual(daemon.seenPaths, ['POST /rpc', 'GET /health']);
     assert.equal(fs.existsSync(daemonPaths.infoPath), false);
-    assert.equal(fs.existsSync(daemonPaths.lockPath), false);
+    // The reset owns its registration, not the protocol lock: reclaim belongs to the acquirer
+    // under ADR 0030's mutation guard, so neither state dir's lock is ever swept.
+    assert.equal(fs.existsSync(daemonPaths.lockPath), true);
     assert.equal(fs.existsSync(requestFlagPaths.infoPath), true);
     assert.equal(fs.existsSync(requestFlagPaths.lockPath), true);
   } finally {

src/daemon-client/__tests__/daemon-client-liveness-probe.test.ts

@@ -61,12 +61,18 @@ test('the probe budget stays an order of magnitude under the narrowest reset-eli
 
 test('a daemon answering /health is responsive', async (t) => {
   if (await skipWhenLoopbackUnavailable(t)) return;
-  const server = http.createServer((_req, res) => {
+  // The url is asserted, not just "an answer": a probe that drifted to any other route would
+  // still be answered by a server that answers everything, and the finding would then describe
+  // some other endpoint's liveness rather than the health route the transport also asks.
+  const requestedUrls: string[] = [];
+  const server = http.createServer((req, res) => {
+    requestedUrls.push(String(req.url));
     res.statusCode = 200;
     res.end('{}');
   });
   await withLoopback(server, async (port) => {
     assert.equal(await probeDaemonResponsive(daemonInfo({ httpPort: port })), true);
+    assert.deepEqual(requestedUrls, ['/health']);
   });
 });
 
@@ -96,13 +102,27 @@ test('a silent HTTP leg cannot veto a live socket leg', async (t) => {
   // second endpoint no window — and the all-negative verdict SIGKILLs a daemon the other
   // transport would have proven alive. The wrong verdict here kills every session on the host,
   // which is the bug #3177 is about.
-  const rpcHangingServer = http.createServer(() => {
+  // The HTTP leg must be OBSERVED, not assumed: a verdict reached with no /health request ever
+  // sent would also pass if the probe simply skipped the leg. So the socket waits for the receipt
+  // before it answers — if the HTTP leg never asks, the socket never answers, the probe spends
+  // its window, and the `true` assertion below fails for the right reason.
+  const healthRequested: { value: boolean } = { value: false };
+  const rpcHangingServer = http.createServer((req, res) => {
+    if (req.url === '/health') healthRequested.value = true;
     // Accepts and never answers: this leg will spend the full budget.
+    res.on('error', () => {});
   });
   const liveSocketServer = net.createServer((socket) => {
     socket.on('error', () => {});
     socket.on('data', () => {
-      socket.write(`${JSON.stringify({ jsonrpc: '2.0', id: 'probe', result: { ok: true } })}\n`);
+      const answer = () => {
+        socket.write(`${JSON.stringify({ jsonrpc: '2.0', id: 'probe', result: { ok: true } })}\n`);
+      };
+      const waitForHttpLeg = (): void => {
+        if (healthRequested.value) answer();
+        else setTimeout(waitForHttpLeg, 5);
+      };
+      waitForHttpLeg();
     });
   });
   const httpPort = await listenOnLoopback(rpcHangingServer);
@@ -114,6 +134,7 @@ test('a silent HTTP leg cannot veto a live socket leg', async (t) => {
         true,
         'the socket answer alone is the finding; the silent HTTP leg must not outweigh it',
       );
+      assert.ok(healthRequested.value, 'a verdict with no request on the HTTP leg skipped the leg');
       assert.ok(
         Date.now() - startedAt < LIVENESS_PROBE_BUDGET_MS,
         'the affirmative short-circuits instead of waiting the silent leg out',
@@ -124,6 +145,28 @@ test('a silent HTTP leg cannot veto a live socket leg', async (t) => {
   }
 });
 
+test('a malformed port makes the HTTP leg an endpoint that did not answer, not a crash', async () => {
+  // `readDaemonInfo` accepts any positive integer port, and `transport.request` THROWS
+  // synchronously on one out of range. Inside the leg's detached async build that throw would
+  // surface as an unhandled rejection — from a recovery path, killing the CLI that is still
+  // holding a timeout error. The leg must fold it into a negative answer instead.
+  const probe = probeDaemonResponsive(daemonInfo({ httpPort: 70_000 }));
+  const rejection = probe.then(
+    () => null,
+    (error: unknown) => error,
+  );
+  process.on('unhandledRejection', failFastOnUnhandledRejection);
+  try {
+    assert.equal(await probe, false);
+    assert.equal(await rejection, null, 'the probe never rejects on a malformed record');
+  } finally {
+    process.off('unhandledRejection', failFastOnUnhandledRejection);
+  }
+});
+function failFastOnUnhandledRejection(error: unknown): never {
+  throw error;
+}
+
 test('a slow-trickle endpoint is cut off by the absolute deadline, not extended by its dribble', async (t) => {
   if (await skipWhenLoopbackUnavailable(t)) return;
   // `socket.setTimeout`/`http.request({timeout})` are IDLE timeouts: a daemon that dribbles bytes

src/daemon-client/__tests__/daemon-client-timeout-route.test.ts

@@ -19,17 +19,31 @@ import path from 'node:path';
 import assert from 'node:assert/strict';
 import { beforeEach, afterEach, test, vi } from 'vitest';
 
-const { mockRunCmdSync, mockIsDaemon, mockStop } = vi.hoisted(() => ({
+const { mockRunCmdSync, mockIsDaemon, mockStop, mockEmitDiagnostic } = vi.hoisted(() => ({
   mockRunCmdSync: vi.fn(),
   mockIsDaemon: vi.fn(),
   mockStop: vi.fn(),
+  mockEmitDiagnostic: vi.fn(),
 }));
 vi.mock('../../daemon-process.ts', async (importOriginal) => ({
   ...(await importOriginal<typeof import('../../daemon-process.ts')>()),
   isAgentDeviceDaemonProcess: mockIsDaemon,
   stopDaemonProcess: mockStop,
 }));
 
+// Records what the route reported while still emitting for real: the timeout's own diagnostic is
+// expected, a transport-failure diagnostic for the same request is not.
+vi.mock('@agent-device/host-kit/diagnostics', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@agent-device/host-kit/diagnostics')>();
+  return {
+    ...actual,
+    emitDiagnostic: (...args: Parameters<typeof actual.emitDiagnostic>) => {
+      mockEmitDiagnostic(...args);
+      actual.emitDiagnostic(...args);
+    },
+  };
+});
+
 vi.mock('@agent-device/host-kit/command', async () => {
   const actual = await vi.importActual<typeof import('@agent-device/host-kit/command')>(
     '@agent-device/host-kit/command',
@@ -75,12 +89,33 @@ function dummyStatePaths(): DaemonPaths {
   return paths;
 }
 
-function seedResettableMetadata(paths: DaemonPaths): void {
-  // A reset clears the metadata it can no longer trust; seeding it lets the reset-path row
-  // observe removal instead of asserting an absence that was never a presence.
+// The start time the seeded registration and the request's DaemonInfo agree on. The ownership
+// fence only deletes a record that MATCHES the timed-out daemon on pid and start time, so a row
+// that expects removal has to name both. The pid is the test process, which fails the identity
+// gate (`isAgentDeviceDaemonProcess`) on its real start time: these tests prove WHICH branch the
+// route takes and never signal a process.
+const TEST_DAEMON_START_TIME = 'test-start-time';
+// A registration whose pid and start time BOTH differ from the timed-out daemon: proof of a
+// replacement (#3125), which the ownership fence must refuse to delete.
+const REPLACEMENT_DAEMON_PID = 99_999;
+
+function seedRegistration(paths: DaemonPaths, owner: { pid: number; startTime: string }): void {
   fs.mkdirSync(paths.baseDir, { recursive: true });
-  fs.writeFileSync(paths.infoPath, JSON.stringify({ pid: process.pid }));
-  fs.writeFileSync(paths.lockPath, JSON.stringify({ pid: process.pid }));
+  fs.writeFileSync(
+    paths.infoPath,
+    JSON.stringify({ pid: owner.pid, processStartTime: owner.startTime }),
+  );
+}
+
+function seedProtocolLockDir(paths: DaemonPaths, owner: { pid: number; startTime: string }): void {
+  // `daemon.lock` is the ADR 0030 directory, not a file: seeding it shaped like production means
+  // a reset that still deleted it out-of-band would be RECLAIMING someone else's lock, and the
+  // survival assertion below catches that. (The pre-fix `unlinkSync` even failed on this shape.)
+  fs.mkdirSync(path.join(paths.lockPath), { recursive: true });
+  fs.writeFileSync(
+    path.join(paths.lockPath, 'owner.json'),
+    JSON.stringify({ pid: owner.pid, startTime: owner.startTime, acquiredAtMs: Date.now() }),
+  );
 }
 
 function buildRequest(command: string, platform: 'android' | 'ios' | undefined): DaemonRequest {
@@ -99,17 +134,27 @@ function buildRequest(command: string, platform: 'android' | 'ios' | undefined):
  * client's own envelope cuts the round trip off, and whose LATER connections — the timeout
  * handler's fresh probe — either answer like a live daemon (`answer`) or are destroyed
  * unanswered (`refuse`, the wedged daemon the reset path is built for: fails the probe fast
- * instead of spending its window). `connections` proves whether the route asked at all.
+ * instead of spending its window).
+ *
+ * `connections` counts accepted TCP connections, NOT requests: the probe's whole claim is that it
+ * asks on a FRESH connection, so an HTTP request counter — which an RPC kept alive by keep-alive
+ * would also advance — would let the probe pass by reusing the timed-out socket. The stand-in only
+ * answers `/health` on a connection after the first, so a row that reaches its kept-alive hint is
+ * simultaneously proving a fresh connection carried a health request.
  */
 async function startStandIn(
   transport: 'http' | 'socket',
   afterFirst: 'answer' | 'refuse',
 ): Promise<{ server: LoopbackServer; port: number; connections: () => number }> {
   let connections = 0;
+  // The TCP ordinal lives on the socket the request arrived on, so two requests sharing one
+  // socket (keep-alive) count as one connection: the probe's claim is a FRESH connection, and a
+  // request counter would advance for an RPC kept alive on the timed-out socket too.
+  type OrdinalSocket = { __connectionOrdinal?: number };
   const server: LoopbackServer =
     transport === 'http'
       ? http.createServer((req, res) => {
-          const connection = ++connections;
+          const connection = (req.socket as OrdinalSocket).__connectionOrdinal ?? 0;
           if (connection > 1 && afterFirst === 'answer' && req.url === '/health') {
             res.statusCode = 200;
             res.end('{}');
@@ -137,6 +182,9 @@ async function startStandIn(
         });
   if (transport === 'http') {
     (server as http.Server).on('clientError', (_err, socket) => socket.destroy());
+    (server as http.Server).on('connection', (socket) => {
+      (socket as OrdinalSocket).__connectionOrdinal = ++connections;
+    });
   }
   const port = await listenOnLoopback(server);
   return { server, port, connections: () => connections };
@@ -154,12 +202,25 @@ async function expectRouteError(run: Promise<unknown>, hintPattern: RegExp): Pro
   // The regression this suite exists to catch: no host-wide process sweep, whatever the request
   // declared.
   assert.equal(mockRunCmdSync.mock.calls.length, 0);
+  // The timeout settles the request and then destroys it, so the transport's `error` event lands
+  // AFTER the rejection. A timed-out request must not also be diagnosed as a transport FAILURE —
+  // that describes a canceled request as a broken host and buries the timeout's own reason. The
+  // destroy surfaces on a later tick, so give it one before reading the record.
+  await new Promise((resolve) => setImmediate(() => setImmediate(resolve)));
+  assert.deepEqual(
+    mockEmitDiagnostic.mock.calls
+      .map(([entry]) => (entry as { phase?: string })?.phase)
+      .filter((phase) => phase === 'daemon_request_socket_error'),
+    [],
+    'a timeout must not also diagnose a transport failure',
+  );
 }
 
 beforeEach(() => {
   mockRunCmdSync.mockReset();
   mockIsDaemon.mockReset();
   mockStop.mockReset();
+  mockEmitDiagnostic.mockReset();
 });
 afterEach(() => vi.restoreAllMocks());
 
@@ -244,17 +305,37 @@ for (const row of ROUTE_ROWS) {
       row.transport === 'remote' ? 'http' : row.transport,
       row.afterFirst,
     );
-    // Seeded for every row: a preserved daemon must still OWN its metadata, so the assertion is
-    // two-sided instead of an absence that was never a presence.
+    // Seeded for every row with an ownership-MATCHING record (pid and start time agree with the
+    // request's DaemonInfo): a preserved daemon must still OWN its metadata, and a row expecting
+    // removal must have earned it past the fence — not observed an absence that was never a
+    // presence. The protocol lock dir is seeded too and must survive every verdict: a reset
+    // reclaims nothing (#3122, ADR 0030).
     const statePaths = dummyStatePaths();
-    seedResettableMetadata(statePaths);
+    const owned = { pid: process.pid, startTime: TEST_DAEMON_START_TIME };
+    seedRegistration(statePaths, owned);
+    seedProtocolLockDir(statePaths, owned);
     try {
       const info: DaemonInfo =
         row.transport === 'remote'
-          ? { baseUrl: `http://127.0.0.1:${daemon.port}`, token: 'test-token', pid: process.pid }
+          ? {
+              baseUrl: `http://127.0.0.1:${daemon.port}`,
+              token: 'test-token',
+              pid: process.pid,
+              processStartTime: TEST_DAEMON_START_TIME,
+            }
           : row.transport === 'http'
-            ? { httpPort: daemon.port, token: 'test-token', pid: process.pid }
-            : { port: daemon.port, token: 'test-token', pid: process.pid };
+            ? {
+                httpPort: daemon.port,
+                token: 'test-token',
+                pid: process.pid,
+                processStartTime: TEST_DAEMON_START_TIME,
+              }
+            : {
+                port: daemon.port,
+                token: 'test-token',
+                pid: process.pid,
+                processStartTime: TEST_DAEMON_START_TIME,
+              };
       await expectRouteError(
         sendRequest(
           info,
@@ -266,21 +347,61 @@ for (const row of ROUTE_ROWS) {
         row.hintPattern,
       );
       assert.equal(daemon.connections(), row.connections, 'probe-vs-skip is a route decision');
-      // A preserved daemon keeps the metadata a reset would clear; a proved-unresponsive one
-      // loses both files (its pid fails the identity gate, so nothing is signaled — only the
-      // bookkeeping a dead daemon cannot release changes).
+      // A preserved daemon keeps the registration a reset would clear; a proved-unresponsive one
+      // loses only what the ownership fence proves the killed daemon still owned (its pid fails
+      // the identity gate, so nothing is signaled — only the bookkeeping changes).
       assert.equal(
         fs.existsSync(statePaths.infoPath),
         !row.resets,
-        row.resets ? 'the reset clears the stale registration' : 'no reset touches metadata',
+        row.resets ? 'the reset clears the owned registration' : 'no reset touches metadata',
       );
-      assert.equal(fs.existsSync(statePaths.lockPath), !row.resets);
+      assert.ok(fs.existsSync(statePaths.lockPath), 'the protocol lock is never swept');
     } finally {
       await closeLoopbackServer(daemon.server);
     }
   });
 }
 
+test('request-timeout route: a reset never deletes a registration a replacement daemon published', async (t) => {
+  if (await skipWhenLoopbackUnavailable(t)) return;
+  // The probe window is exactly when another client can start a replacement daemon and publish
+  // ITS record (#3177 review, on top of #3125's fence). The timed-out request's info still names
+  // the wedged pid, so the reset must read the record before clearing it: here it names a
+  // different owner, and deleting it would orphan the live replacement — the same class of
+  // host-scoped damage this PR removes everywhere else.
+  const daemon = await startStandIn('socket', 'refuse');
+  const statePaths = dummyStatePaths();
+  seedRegistration(statePaths, { pid: REPLACEMENT_DAEMON_PID, startTime: 'replacement-start' });
+  seedProtocolLockDir(statePaths, { pid: process.pid, startTime: TEST_DAEMON_START_TIME });
+  try {
+    await expectRouteError(
+      sendRequest(
+        {
+          port: daemon.port,
+          token: 'test-token',
+          pid: process.pid,
+          processStartTime: TEST_DAEMON_START_TIME,
+        },
+        buildRequest(RESET_POLICY_COMMAND, undefined),
+        'socket',
+        statePaths,
+        TIMEOUT_MS,
+      ),
+      /The daemon did not answer the liveness probe and was reset after the timeout/,
+    );
+    assert.ok(
+      fs.existsSync(statePaths.infoPath),
+      'the replacement keeps the registration it published',
+    );
+    assert.equal(
+      JSON.parse(fs.readFileSync(statePaths.infoPath, 'utf8')).pid,
+      REPLACEMENT_DAEMON_PID,
+    );
+  } finally {
+    await closeLoopbackServer(daemon.server);
+  }
+});
+
 test('a refused timeout fallback preserves the timeout without an unhandled rejection', async (t) => {
   if (await skipWhenLoopbackUnavailable(t)) return;
   // The reset branch's fallback: a SIGKILL the kernel refuses goes through the confirmed-retirement