fix: cancel pending human takeover on disconnect

Author: szdziedzic

docs/adr/0007-remote-device-leases.md

@@ -68,6 +68,11 @@ their ownership checks. Unknown effects are treated as mutations. A pending acti
 mutations and drains those already admitted before reporting active; advisory execution locks alone
 do not establish this guarantee for fresh sessions.
 
+Activation follows the calling RPC or host HTTP request's cancellation signal. A disconnect while
+draining removes only that request's pending hold, leaving successor and unrelated holds intact;
+canceling activation does not cancel the mutations being drained. Completed holds use their TTL or
+explicit release lifecycle.
+
 Holds, like leases, are in-memory and do not survive daemon restart. Controllers must reconnect and
 re-establish them; no persisted hold store is used. Local takeover is deferred: a future host-global
 human-control fence must coexist with the local session's device claim, not acquire it exclusively.

src/daemon/__tests__/device-mutation-drain.test.ts

@@ -2,6 +2,7 @@ import { createControlLatch } from './human-control-fixtures.ts';
 import assert from 'node:assert/strict';
 import { test } from 'vitest';
 import { DeviceMutationDrain } from '../device-mutation-drain.ts';
+import { getEventListeners } from 'node:events';
 
 test('drain counts concurrent operations, releases on failure, and isolates device keys', async () => {
   const drain = new DeviceMutationDrain();
@@ -24,3 +25,30 @@ test('drain counts concurrent operations, releases on failure, and isolates devi
   assert.equal(idle, true);
   await drain.wait('device-a');
 });
+
+test('canceling a drain waiter detaches its signal without canceling mutations or other waiters', async () => {
+  const drain = new DeviceMutationDrain();
+  const finish = createControlLatch();
+  const mutation = drain.run('device-a', () => finish.promise);
+  const controller = new AbortController();
+  const reason = new Error('canceled waiter');
+  const rejected = assert.rejects(
+    drain.wait('device-a', controller.signal),
+    (error) => error === reason,
+  );
+  let drained = false;
+  const survivor = drain.wait('device-a').then(() => {
+    drained = true;
+  });
+  controller.abort(reason);
+  await rejected;
+  assert.equal(drained, false);
+  assert.equal(getEventListeners(controller.signal, 'abort').length, 0);
+  const successController = new AbortController();
+  const successful = drain.wait('device-a', successController.signal);
+  finish.resolve();
+  await Promise.all([mutation, survivor, successful]);
+  assert.equal(drained, true);
+  assert.equal(getEventListeners(successController.signal, 'abort').length, 0);
+  await assert.rejects(drain.wait('device-a', controller.signal), (error) => error === reason);
+});

src/daemon/__tests__/human-control-http.test.ts

@@ -1,14 +1,18 @@
 import assert from 'node:assert/strict';
-import type http from 'node:http';
-import { test } from 'vitest';
+import http from 'node:http';
+import { test, vi } from 'vitest';
 import {
   closeLoopbackServer,
   listenOnLoopback,
   skipWhenLoopbackUnavailable,
 } from '../../__tests__/test-utils/loopback.ts';
 import { HUMAN_CONTROL_HTTP_PREFIX } from '../human-control-contract.ts';
 import { LeaseRegistry } from '../lease-registry.ts';
-import { HUMAN_CONTROL_SCOPE, humanControlRequest } from './human-control-fixtures.ts';
+import {
+  HUMAN_CONTROL_SCOPE,
+  humanControlRequest,
+  createControlLatch,
+} from './human-control-fixtures.ts';
 import { createHumanControlHarness } from './human-control-router-fixture.ts';
 import { tryHandleHumanControlHttpRoute } from '../human-control-http.ts';
 import { createDaemonHttpServer } from '../server/http-server.ts';
@@ -47,6 +51,76 @@ test('malformed request URLs return a normalized error', async () => {
   assert.equal((JSON.parse(responseBody) as { code?: string }).code, 'INVALID_ARGS');
 });
 
+for (const transport of ['tenant RPC', 'host PUT'] as const) {
+  test(`${transport} disconnect during drain removes its pending hold before the mutation ends`, async (t) => {
+    if (await skipWhenLoopbackUnavailable(t)) return;
+    const { registry, lease, handleRequest } = createHumanControlHarness();
+    const finish = createControlLatch();
+    const disconnected = createControlLatch();
+    let mutationFinished = false;
+    const mutation = registry.runDeviceMutation(lease, async () => {
+      await finish.promise;
+      mutationFinished = true;
+    });
+    const server = await createDaemonHttpServer({
+      token: 'test-token',
+      leaseRegistry: registry,
+      handleRequest,
+    });
+    server.on('request', (_req, res) => {
+      res.once('close', () => {
+        if (!res.writableFinished) disconnected.resolve();
+      });
+    });
+    let request: http.ClientRequest | undefined;
+    try {
+      const port = await listenOnLoopback(server);
+      const isRpc = transport === 'tenant RPC';
+      const body = JSON.stringify(
+        isRpc
+          ? {
+              jsonrpc: '2.0',
+              id: 'disconnected-takeover',
+              method: 'agent_device.command',
+              params: humanControlRequest(lease, 'human_control', ['put', 'disconnected', '{}']),
+            }
+          : { scope: HUMAN_CONTROL_SCOPE },
+      );
+      request = http.request({
+        host: '127.0.0.1',
+        port,
+        path: isRpc ? '/rpc' : `${HUMAN_CONTROL_HTTP_PREFIX}/disconnected`,
+        method: isRpc ? 'POST' : 'PUT',
+        headers: {
+          authorization: 'Bearer test-token',
+          'content-type': 'application/json',
+          'content-length': Buffer.byteLength(body),
+        },
+      });
+      request.on('error', () => undefined);
+      request.end(body);
+      await vi.waitFor(() => {
+        assert.equal(registry.listHumanControlHolds({ kind: 'host' })[0]?.state, 'activating');
+      });
+      request.destroy();
+      await disconnected.promise;
+      await vi.waitFor(() => {
+        assert.deepEqual(registry.listHumanControlHolds({ kind: 'host' }), []);
+      });
+      assert.equal(mutationFinished, false);
+      finish.resolve();
+      await mutation;
+      assert.deepEqual(registry.listHumanControlHolds({ kind: 'host' }), []);
+      assert.equal(await registry.runDeviceMutation(lease, async () => 'resumed'), 'resumed');
+    } finally {
+      request?.destroy();
+      finish.resolve();
+      await mutation;
+      await closeLoopbackServer(server);
+    }
+  });
+}
+
 test('host administration and tenant RPC use the same lease registry with distinct authority', async (t) => {
   if (await skipWhenLoopbackUnavailable(t)) return;
   const { registry, lease, handleRequest } = createHumanControlHarness();

src/daemon/__tests__/lease-registry-scope.test.ts

@@ -1,6 +1,10 @@
 import assert from 'node:assert/strict';
 import { test } from 'vitest';
-import { leaseDeviceBindingKey, normalizeAllocateLeaseRequest } from '../lease-registry-scope.ts';
+import {
+  createLeaseTtlResolver,
+  leaseDeviceBindingKey,
+  normalizeAllocateLeaseRequest,
+} from '../lease-registry-scope.ts';
 import { HUMAN_CONTROL_LEASE_REQUEST, HUMAN_CONTROL_SCOPE } from './human-control-fixtures.ts';
 
 test('allocation and human control share the exact contention identity', () => {
@@ -20,3 +24,20 @@ test('allocation and human control share the exact contention identity', () => {
   );
   assert.equal(leaseDeviceBindingKey({ backend: 'ios-simulator' }), undefined);
 });
+
+test('lease TTL normalization retains defaults, limits, and invalid configuration handling', () => {
+  const defaults = createLeaseTtlResolver({});
+  assert.equal(defaults(undefined), 60_000);
+  assert.equal(defaults(1.5), 60_000);
+  assert.equal(defaults(5_000), 5_000);
+  assert.equal(defaults(600_000), 600_000);
+  for (const ttl of [4_999, 600_001]) assert.throws(() => defaults(ttl), { code: 'INVALID_ARGS' });
+  const configured = createLeaseTtlResolver({
+    defaultLeaseTtlMs: 0,
+    minLeaseTtlMs: 0,
+    maxLeaseTtlMs: -1,
+  });
+  assert.equal(configured(undefined), 1);
+  assert.equal(configured(1), 1);
+  assert.throws(() => configured(2), { code: 'INVALID_ARGS' });
+});

src/daemon/__tests__/lease-registry.test.ts

@@ -1,5 +1,6 @@
 import { test } from 'vitest';
 import assert from 'node:assert/strict';
+import { createRequestCanceledError } from '@agent-device/kernel/errors';
 import { LeaseRegistry } from '../lease-registry.ts';
 import {
   HUMAN_CONTROL_LEASE_REQUEST,
@@ -511,3 +512,55 @@ test('holds do not survive registry restart, including host holds created withou
   assert.deepEqual(restarted.listHumanControlHolds({ kind: 'host' }), []);
   assert.doesNotThrow(() => restarted.allocateLease(HUMAN_CONTROL_LEASE_REQUEST));
 });
+
+test('canceled activation refreshes the protected lease without waiting for the mutation', async () => {
+  let now = 0;
+  const registry = new LeaseRegistry({ now: () => now, defaultLeaseTtlMs: 5_000 });
+  const lease = registry.allocateLease(HUMAN_CONTROL_LEASE_REQUEST);
+  const authority = { kind: 'lease', leaseId: lease.leaseId } as const;
+  const finish = createControlLatch();
+  const mutation = registry.runDeviceMutation(lease, () => finish.promise);
+  const controller = new AbortController();
+  const activation = registry.putHumanControlHold(authority, 'console', {}, controller.signal);
+  const canceled = createRequestCanceledError();
+  const rejected = assert.rejects(activation, (error) => error === canceled);
+  now = 20_000;
+  controller.abort(canceled);
+  await rejected;
+  assert.deepEqual(registry.listHumanControlHolds(authority), []);
+  assert.equal(registry.listActiveLeases()[0]?.expiresAt, 25_000);
+  finish.resolve();
+  await mutation;
+  assert.deepEqual(registry.listHumanControlHolds(authority), []);
+});
+
+test('canceling a superseded activation cannot remove its successor or another hold', async () => {
+  const registry = new LeaseRegistry();
+  const lease = registry.allocateLease(HUMAN_CONTROL_LEASE_REQUEST);
+  const authority = { kind: 'lease', leaseId: lease.leaseId } as const;
+  const finish = createControlLatch();
+  const mutation = registry.runDeviceMutation(lease, () => finish.promise);
+  const controller = new AbortController();
+  const canceled = createRequestCanceledError();
+  const rejected = assert.rejects(
+    registry.putHumanControlHold(authority, 'console', {}, controller.signal),
+    (error) => error === canceled,
+  );
+  const successor = registry.putHumanControlHold(authority, 'console', { reason: 'successor' });
+  const other = registry.putHumanControlHold(authority, 'other', {});
+  controller.abort(canceled);
+  await rejected;
+  assert.deepEqual(
+    registry.listHumanControlHolds(authority).map((hold) => hold.id),
+    ['console', 'other'],
+  );
+  finish.resolve();
+  await mutation;
+  assert.equal((await successor).reason, 'successor');
+  assert.equal((await other).state, 'active');
+  await assert.rejects(
+    registry.putHumanControlHold(authority, 'console', {}, controller.signal),
+    (error) => error === canceled,
+  );
+  assert.equal(registry.listHumanControlHolds(authority)[0]?.reason, 'successor');
+});

src/daemon/device-mutation-drain.ts

@@ -19,12 +19,24 @@ export class DeviceMutationDrain {
     }
   }
 
-  async wait(key: string): Promise<void> {
+  async wait(key: string, signal?: AbortSignal): Promise<void> {
+    signal?.throwIfAborted();
     if (!this.active.has(key)) return;
-    await new Promise<void>((resolve) => {
-      const waiters = this.waiters.get(key) ?? new Set<() => void>();
-      waiters.add(resolve);
-      this.waiters.set(key, waiters);
-    });
+    const waiters = this.waiters.get(key) ?? new Set<() => void>();
+    let drained: () => void = () => {};
+    let aborted: () => void = () => {};
+    try {
+      await new Promise<void>((resolve, reject) => {
+        drained = resolve;
+        aborted = () => reject(signal?.reason);
+        waiters.add(drained);
+        this.waiters.set(key, waiters);
+        signal?.addEventListener('abort', aborted, { once: true });
+      });
+    } finally {
+      signal?.removeEventListener('abort', aborted);
+      waiters.delete(drained);
+      if (waiters.size === 0 && this.waiters.get(key) === waiters) this.waiters.delete(key);
+    }
   }
 }

src/daemon/handlers/human-control.ts

@@ -1,4 +1,5 @@
 import { AppError } from '@agent-device/kernel/errors';
+import { getRequestSignal } from '@agent-device/host-kit/request';
 import { parseHumanControlHoldInput } from '../human-control-contract.ts';
 import type { LeaseRegistry } from '../lease-registry.ts';
 import type { DaemonRequest, DaemonResponse } from '../types.ts';
@@ -21,7 +22,12 @@ export async function handleHumanControlCommand(params: {
       return { ok: true, data: { holds: registry.listHumanControlHolds(authority) } };
     case 'put': {
       assertArgumentCount(positionals, 3);
-      const hold = await registry.putHumanControlHold(authority, holdId, readHoldInput(rawInput));
+      const hold = await registry.putHumanControlHold(
+        authority,
+        holdId,
+        readHoldInput(rawInput),
+        getRequestSignal(req.meta?.requestId),
+      );
       return { ok: true, data: { hold, state: 'active' } };
     }
     case 'remove': {

src/daemon/human-control-http.ts

@@ -1,5 +1,5 @@
 import type http from 'node:http';
-import { AppError, normalizeError } from '@agent-device/kernel/errors';
+import { AppError, createRequestCanceledError, normalizeError } from '@agent-device/kernel/errors';
 import { readNodeHttpRequestBody } from '../utils/node-http.ts';
 import { timingSafeStringEqual } from '../utils/timing-safe-equal.ts';
 import { sendRestJsonError } from './http-errors.ts';
@@ -75,9 +75,27 @@ async function upsertHumanControlHold(
   holdId: string,
   params: HumanControlHttpParams,
 ): Promise<void> {
-  const input = await readHoldInput(params.req);
-  const hold = await params.registry.putHumanControlHold({ kind: 'host' }, holdId, input);
-  sendJson(params.res, { ok: true, hold, state: 'active' });
+  const { req, res } = params;
+  const controller = new AbortController();
+  const cancelIfDisconnected = () => {
+    if (!res.writableFinished) controller.abort(createRequestCanceledError());
+  };
+  req.once('aborted', cancelIfDisconnected);
+  res.once('close', cancelIfDisconnected);
+  if (req.aborted || res.destroyed) cancelIfDisconnected();
+  try {
+    const input = await readHoldInput(req);
+    const hold = await params.registry.putHumanControlHold(
+      { kind: 'host' },
+      holdId,
+      input,
+      controller.signal,
+    );
+    sendJson(res, { ok: true, hold, state: 'active' });
+  } finally {
+    req.off('aborted', cancelIfDisconnected);
+    res.off('close', cancelIfDisconnected);
+  }
 }
 
 function removeHumanControlHold(holdId: string, params: HumanControlHttpParams): void {

src/daemon/lease-registry-scope.ts

@@ -83,11 +83,31 @@ export type NormalizedAllocateLeaseRequest = {
   ttlMs?: number;
 };
 
-export const DEFAULT_LEASE_TTL_MS = 60_000;
-export const MIN_LEASE_TTL_MS = 5_000;
-export const MAX_LEASE_TTL_MS = 10 * 60_000;
+const DEFAULT_LEASE_TTL_MS = 60_000;
+const MIN_LEASE_TTL_MS = 5_000;
+const MAX_LEASE_TTL_MS = 10 * 60_000;
 const DEFAULT_LEASE_PROVIDER = 'default';
 
+export function createLeaseTtlResolver(options: LeaseRegistryOptions) {
+  const defaultTtl = Number.isInteger(options.defaultLeaseTtlMs)
+    ? Math.max(1, Number(options.defaultLeaseTtlMs))
+    : DEFAULT_LEASE_TTL_MS;
+  const minTtl = Number.isInteger(options.minLeaseTtlMs)
+    ? Math.max(1, Number(options.minLeaseTtlMs))
+    : MIN_LEASE_TTL_MS;
+  const maxTtl = Number.isInteger(options.maxLeaseTtlMs)
+    ? Math.max(minTtl, Number(options.maxLeaseTtlMs))
+    : MAX_LEASE_TTL_MS;
+  return (raw: number | undefined): number => {
+    if (!Number.isInteger(raw)) return defaultTtl;
+    const value = Number(raw);
+    if (value < minTtl || value > maxTtl) {
+      throw new AppError('INVALID_ARGS', `Lease ttlMs must be between ${minTtl} and ${maxTtl}.`);
+    }
+    return value;
+  };
+}
+
 function normalizeRunId(raw: string | undefined): string | undefined {
   if (!raw) return undefined;
   const value = raw.trim();