feat(leasing): call lease-end participants before reclaiming a device

docs/CONFIGURATION.md

@@ -31,6 +31,7 @@ edit yourself keeps the mode you give it until the next `config set`; run
 | `lease.requestRetentionMs`        | How long a settled lease request stays stored, counted from when it settled. A client that repeats a request with the same `idempotencyKey` inside this window gets the stored result back; after it, the record is removed and the same key starts a new request. | `10 minutes`                                                    |
 | `lease.maxRequestRecords`         | The most lease-request records Simlock keeps. When a new request would go over it, the oldest settled record is removed. A request that is still open is never removed. | `10000`                                                         |
 | `lease.maxPerRequester`           | How many leases one agent id may hold at once. A positive integer; anything else (`0`, `-1`, `1.5`, a string) stops the daemon from starting. An agent id still has at most one waiting request at a time. At the limit, or with a request already waiting, a new request is refused with `REQUESTER_ALREADY_LEASED`. On a gateway the limit counts the agent's leases across the whole fleet, and workers do not apply their own limit to what the gateway sends them. An agent that holds more than the limit (you lowered it and restarted) keeps its leases and cannot request more until it is below. Read at start; changing it needs a restart. **Risk above `1`:** agents that lease one device after another can block each other, each holding one device while waiting for a second.| `1`                                                             |
+| `lease.endWaitMs` | How long Simlock waits, after a lease ends, for add-ons that act on the device before it is wiped (for example one that ends a session on it). A positive number of milliseconds, at most `2147483647`; anything else stops the daemon from starting. When it passes, the wipe starts anyway. Stopping the daemon and `simlock nuke` do not wait. With no such add-on, nothing waits. Read at start; changing it needs a restart. | `600000` |
 | `gateway.url`                     | **Worker side.** Base URL of the gateway this worker joins, e.g. `wss://gw.example:4700`; the worker dials `<url>/v1/uplink` and upgrades it to a WebSocket. Joining grants that gateway the `admin` role on this daemon, and `gateway.token` rides on the upgrade request as a bearer credential, so use `wss://` — or plain `ws://` only over loopback or inside your own tunnel. `http://` and `https://` are rejected at load. Unset means "do not join a fleet": the default, and the only thing that changes about a joined worker.                    | unset                                                            |
 | `gateway.token`                   | **Worker side.** The join token (`simlock token create --role worker`, minted on the gateway) this worker presents when it opens its uplink. Required whenever `gateway.url` is set. `simlock config` shows it; the daemon log and the `daemon.started` event show `[redacted]`. It is stored in plain text in `config.json`, which `simlock config set` writes owner-only.                                          | unset                                                            |
 | `gateway.label`                   | **Worker side.** Display name for this worker in `simlock worker list`, `status`, the console, and on the lease's `worker` block. Display-only: nothing routes on it and it need not be unique.                              | the worker's own id                                              |

docs/internal/COMPONENTS.md

@@ -101,7 +101,7 @@ modules with.
 | `AcquisitionPlanner` | `src/leasing/acquisition-planner.ts` | The read-only plan for one request (for a physical request: the first fitting ready enrolled device that nothing holds, never a boot, provision or eviction): grant a fitting ready device, wait for a fitting one the warm pool is booting, boot a fitting shut-down one, evict an idle running device, provision, wait, or refuse. Takes capacity reservations and claims and hands them to the caller. | Perform any side effect. |
 | `LeaseLifecycle` | `src/leasing/lease-lifecycle.ts` | Grant, renew, the registry half of a release, expiry scheduling through `LeaseExpiryScheduler`. | Reclaim; wake the queue. |
 | `LeaseExpiryScheduler` | `src/leasing/lease-expiry-scheduler.ts` | TTL timers, delivered to the release coordinator. | Decide what expiry means. |
-| `LeaseReleaseCoordinator` | `src/leasing/lease-release-coordinator.ts` | Release, release-all, expiry, device-lost and the startup ending (reclaim, wait with no reclaim started, or missing in one write) as lease commands: the serialized commit, the backgrounded reclaim with its claim, the drain for nuke and daemon stop, maintenance admission. | Purge or decide warmth: it hands the released device to `ReclaimCoordinator`. |
+| `LeaseReleaseCoordinator` | `src/leasing/lease-release-coordinator.ts` | Release, release-all, expiry, device-lost and the startup ending (reclaim, wait with no reclaim started, or missing in one write) as lease commands: the serialized commit, the lease-end participants called after it (bounded by `lease.endWaitMs`, cut short by nuke, stop and dispose), the backgrounded reclaim with its claim, the drain for nuke and daemon stop, maintenance admission. | Purge or decide warmth: it hands the released device to `ReclaimCoordinator`. |
 | `LeaseHealthMonitor` | `src/leasing/lease-health-monitor.ts` | Observing leased devices on a tick, rebooting a crashed one under its lease, giving the lease up as `device-lost` when recovery is exhausted. | Touch an unleased device. |
 | `PhysicalDeviceWatcher` | `src/leasing/physical-device-watcher.ts` | The presence tick (every `health.probeIntervalMs`, whatever `health.enabled` says), the grace-period count of not-present reads, `absent` and its reason, the device-lost ending of a lease on a device that is gone, and the return reclaim of a device that is back (ADR 0022 §7). | Retry quarantine, grant, or run a driver verb but the presence listing on a leased device. |
 

src/cli/index.test.ts

@@ -5846,6 +5846,7 @@ function testConfig(): Config {
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
     },
     capacity: {
       strategy: "resource",

src/contract/operations.test.ts

@@ -1071,6 +1071,7 @@ describe("operation input/output round trips", () => {
         requestRetentionMs: 1,
         maxRequestRecords: 1,
         maxPerRequester: 1,
+        endWaitMs: 600_000,
       },
       exec: { timeoutMs: 1 },
       diskPressure: { freeBytesThreshold: 1 },
@@ -1126,6 +1127,7 @@ describe("operation input/output round trips", () => {
         requestRetentionMs: 1,
         maxRequestRecords: 1,
         maxPerRequester: 1,
+        endWaitMs: 600_000,
       },
       exec: { timeoutMs: 1 },
       diskPressure: { freeBytesThreshold: 1 },

src/contract/protocol.ts

@@ -10,7 +10,7 @@ export interface ProtocolRange {
 const protocolRangeSchema = z.object({ min: z.number().int(), max: z.number().int() });
 
 /** Widen only once a second version is kept alive side by side with the first; a bump that leaves no shim behind moves both ends. */
-export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 28, max: 28 };
+export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 29, max: 29 };
 
 const LEGACY_DAEMON_PROTOCOL_VERSION = 2;
 

src/contract/schemas.ts

@@ -818,6 +818,7 @@ export const configSchema = z.object({
     requestRetentionMs: z.number(),
     maxRequestRecords: z.number(),
     maxPerRequester: z.number(),
+    endWaitMs: z.number(),
   }),
   exec: z.object({ timeoutMs: z.number() }),
   diskPressure: z.object({ freeBytesThreshold: z.number() }),

src/core/cleanup/idle-destroy.test.ts

@@ -53,6 +53,7 @@ const config: Config = {
     requestRetentionMs: 600_000,
     maxRequestRecords: 10_000,
     maxPerRequester: 1,
+    endWaitMs: 600_000,
   },
   capacity: {
     strategy: "resource",

src/core/cleanup/idle-shutdown.test.ts

@@ -51,6 +51,7 @@ const config: Config = {
     requestRetentionMs: 600_000,
     maxRequestRecords: 10_000,
     maxPerRequester: 1,
+    endWaitMs: 600_000,
   },
   capacity: {
     strategy: "resource",

src/core/config.test.ts

@@ -100,6 +100,7 @@ describe("loadConfig", () => {
         requestRetentionMs: 600_000,
         maxRequestRecords: 10_000,
         maxPerRequester: 1,
+        endWaitMs: 600_000,
       },
       capacity: {
         strategy: "resource",
@@ -683,6 +684,50 @@ describe("loadConfig", () => {
     },
   );
 
+  it("lease.endWaitMs defaults to 600000 and accepts a positive number and 2147483647", async () => {
+    const filesystem = new MemoryFilesystem();
+    await filesystem.mkdirp("/home/agent/.simlock");
+    const unset = await loadConfig({ configPath, filesystem, systemStats: createStats() });
+
+    await filesystem.writeFileAtomic(configPath, JSON.stringify({ lease: { endWaitMs: 5_000 } }));
+    const small = await loadConfig({ configPath, filesystem, systemStats: createStats() });
+    await filesystem.writeFileAtomic(
+      configPath,
+      JSON.stringify({ lease: { endWaitMs: 2_147_483_647 } }),
+    );
+    const largest = await loadConfig({ configPath, filesystem, systemStats: createStats() });
+
+    expect(unset.lease.endWaitMs).toBe(600_000);
+    expect(small.lease.endWaitMs).toBe(5_000);
+    expect(largest.lease.endWaitMs).toBe(2_147_483_647);
+  });
+
+  it.each([[0], [-1], [2_147_483_648], ["600000"]])(
+    "refuses lease.endWaitMs %j with a message naming the key, where 5000 loads",
+    async (value) => {
+      const filesystem = new MemoryFilesystem();
+      await filesystem.mkdirp("/home/agent/.simlock");
+      await filesystem.writeFileAtomic(configPath, JSON.stringify({ lease: { endWaitMs: 5_000 } }));
+      const accepted = await loadConfig({ configPath, filesystem, systemStats: createStats() });
+      expect(accepted.lease.endWaitMs).toBe(5_000);
+      await filesystem.writeFileAtomic(configPath, JSON.stringify({ lease: { endWaitMs: value } }));
+
+      await expect(
+        loadConfig({ configPath, filesystem, systemStats: createStats() }),
+      ).rejects.toThrow("lease.endWaitMs");
+    },
+  );
+
+  it("a loaded config parsed through configSchema keeps lease.endWaitMs", async () => {
+    const config = await loadConfig({
+      configPath,
+      filesystem: new MemoryFilesystem(),
+      systemStats: createStats(),
+    });
+
+    expect((configSchema.parse(config) as unknown as Config).lease.endWaitMs).toBe(600_000);
+  });
+
   it("reads lease.identity per platform and leaves the unset platform reusable", async () => {
     const filesystem = new MemoryFilesystem();
     await filesystem.mkdirp("/home/agent/.simlock");
@@ -763,6 +808,7 @@ describe("loadConfig", () => {
         defaultTtlMs: 15 * 60_000,
         identity: { android: "reusable", ios: "reusable" },
         maxPerRequester: 1,
+        endWaitMs: 600_000,
         maxRequestRecords: 10_000,
         maxTtlMs: 4 * 60 * 60_000,
         requestRetentionMs: 10 * 60_000,

src/core/config.ts

@@ -181,6 +181,7 @@ export interface Config {
      */
     readonly maxRequestRecords: number;
     readonly maxPerRequester: number;
+    readonly endWaitMs: number;
   };
   /**
    * ADR 0005 §19e. Platform-agnostic on purpose: it bounds the *daemon's* willingness to wait
@@ -678,6 +679,7 @@ function defaultConfig(
       requestRetentionMs: DEFAULT_LEASE_REQUEST_RETENTION_MS,
       maxRequestRecords: DEFAULT_LEASE_MAX_REQUEST_RECORDS,
       maxPerRequester: DEFAULT_LEASE_MAX_PER_REQUESTER,
+      endWaitMs: DEFAULT_EXEC_TIMEOUT_MS,
     },
     exec: { timeoutMs: DEFAULT_EXEC_TIMEOUT_MS },
     diskPressure: { freeBytesThreshold: 10 * 1024 ** 3 },
@@ -827,6 +829,7 @@ function configValidators(strategy: CapacityStrategyName): Record<string, Valida
       requestRetentionMs: positiveNumber,
       maxRequestRecords: positiveInteger,
       maxPerRequester: positiveInteger,
+      endWaitMs: endWaitValidator,
     }),
     exec: objectValidator({ timeoutMs: positiveNumber }),
     diskPressure: objectValidator({ freeBytesThreshold: nonNegativeNumber }),
@@ -880,6 +883,17 @@ function configValidators(strategy: CapacityStrategyName): Record<string, Valida
  */
 export type DefaultModels = Partial<Record<DeviceClass, readonly string[]>>;
 
+/** why: `setTimeout` fires after about 1 ms above this delay, which would end the wait at once. */
+const MAXIMUM_TIMER_DELAY_MS = 2_147_483_647;
+
+const endWaitValidator: Validator = (value, path) => {
+  const wait = positiveNumber(value, path);
+  if (wait > MAXIMUM_TIMER_DELAY_MS) {
+    throw invalidValue(path, `a positive number no larger than ${MAXIMUM_TIMER_DELAY_MS}`);
+  }
+  return wait;
+};
+
 /**
  * Each key is a device class and each value one non-empty model name or a non-empty list of
  * them, stored as a list. Any other key is refused rather than warned about: a misspelled class

src/core/doctor.test.ts

@@ -3052,6 +3052,7 @@ function config(stalledTransitionOverrides: Partial<Config["stalledTransition"]>
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
     },
     capacity: {
       strategy: "resource",

src/core/nuke.test.ts

@@ -191,6 +191,7 @@ function config(): Config {
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
     },
     capacity: {
       strategy: "resource",

src/core/reaper.test.ts

@@ -87,6 +87,7 @@ function config(): Config {
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
     },
     capacity: {
       strategy: "resource",

src/daemon-protocol/index.test.ts

@@ -9,7 +9,7 @@ import {
 
 describe("daemon protocol", () => {
   it("keeps the current protocol version and newline framing", () => {
-    expect(DAEMON_PROTOCOL_VERSION).toBe(28);
+    expect(DAEMON_PROTOCOL_VERSION).toBe(29);
     expect(serializeFrame({ id: 1, type: "hello" })).toBe('{"id":1,"type":"hello"}\n');
   });
 

src/daemon/dispatcher.test.ts

@@ -3581,6 +3581,7 @@ function testConfig(
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
       ...leaseOverrides,
     },
     capacity,

src/daemon/server.test.ts

@@ -190,8 +190,9 @@ describe("DaemonServer", () => {
     [24, "cannot send lease.request's physical or read physicalQueueDepth"],
     [26, "cannot read a physical device's presence"],
     [27, "cannot read config.get's lease.maxPerRequester"],
+    [28, "cannot read config.get's lease.endWaitMs"],
   ])(
-    "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 28, to a client on protocol %i, which %s",
+    "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 29, to a client on protocol %i, which %s",
     async (version) => {
       const harness = await createHarness();
       const previous = await createClient(harness.socketPath);
@@ -201,7 +202,7 @@ describe("DaemonServer", () => {
       ).resolves.toMatchObject({
         error: {
           code: "PROTOCOL_VERSION_UNSUPPORTED",
-          details: { client: { min: version, max: version }, daemon: { min: 28, max: 28 } },
+          details: { client: { min: version, max: version }, daemon: { min: 29, max: 29 } },
         },
         ok: false,
       });
@@ -3438,6 +3439,7 @@ function testConfig(
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
       ...leaseOverrides,
     },
     capacity: {

src/gateway/dispatcher.test.ts

@@ -62,6 +62,7 @@ const gatewayConfig = {
     requestRetentionMs: 600_000,
     maxRequestRecords: 10_000,
     maxPerRequester: 1,
+    endWaitMs: 600_000,
   },
   log: { level: "info" as const, rotateBytes: 1 },
   mode: "gateway" as const,

src/gateway/service.test.ts

@@ -1207,7 +1207,7 @@ describe("GatewayService", () => {
       min: PROTOCOL_VERSION_RANGE.max - 1,
       max: PROTOCOL_VERSION_RANGE.max - 1,
     };
-    expect(PROTOCOL_VERSION_RANGE.max).toBe(28);
+    expect(PROTOCOL_VERSION_RANGE.max).toBe(29);
     expect(negotiateProtocolVersion(PROTOCOL_VERSION_RANGE, previous)).toBeUndefined();
     const harness = fleet();
     await harness.service.start();

src/http/test-fakes.ts

@@ -76,6 +76,7 @@ export function testConfig(
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
       ...overrides,
     },
     log: { level: "info", rotateBytes: 5 * 1024 * 1024 },

src/leasing/acquisition-planner.test.ts

@@ -62,6 +62,7 @@ const config: Config = {
     requestRetentionMs: 600_000,
     maxRequestRecords: 10_000,
     maxPerRequester: 1,
+    endWaitMs: 600_000,
   },
   capacity: {
     strategy: "resource",

src/leasing/create-leasing.test.ts

@@ -79,6 +79,7 @@ function config(
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
       ...overrides,
     },
     capacity: {

src/leasing/create-leasing.ts

@@ -24,6 +24,7 @@ import type {
   ClientReleaseReason,
   DeviceModeReader,
   LeaseCommands,
+  LeaseEndParticipant,
   QueueControl,
 } from "./lease-ports.js";
 import { LeaseReleaseCoordinator } from "./lease-release-coordinator.js";
@@ -55,6 +56,7 @@ export interface LeasingOptions {
   readonly modelPreferences?: ModelPreferences | undefined;
   /** why: not started on build: a health probe would shell out while startup convergence does. */
   readonly healthMonitor?: boolean | undefined;
+  readonly leaseEndParticipants?: readonly LeaseEndParticipant[] | undefined;
 }
 
 export interface Leasing extends LeaseCommands, QueueControl, DeviceModeReader {
@@ -160,6 +162,15 @@ export function createLeasing(options: LeasingOptions): Leasing {
   const releaseCoordinator: LeaseReleaseCoordinator = new LeaseReleaseCoordinator({
     claims: core.claims,
     decisions,
+    ...(options.leaseEndParticipants === undefined || options.leaseEndParticipants.length === 0
+      ? {}
+      : {
+          leaseEnd: {
+            clock: options.clock,
+            endWaitMs: options.config.lease.endWaitMs,
+            participants: options.leaseEndParticipants,
+          },
+        }),
     lifecycle: leases,
     ...(options.logger === undefined ? {} : { logger: options.logger }),
     // Called once a backgrounded reclaim has given up its claim; the warm pool can act on the
@@ -256,6 +267,7 @@ export function createLeasing(options: LeasingOptions): Leasing {
       await releaseCoordinator.settleBackgroundReclaims();
     },
     dispose: () => {
+      releaseCoordinator.dispose();
       physicalDeviceWatcher.dispose();
       expiry.dispose();
     },

src/leasing/index.ts

@@ -2,7 +2,15 @@ export { createLeasing } from "./create-leasing.js";
 export { NoCapacityError } from "./lease-acquisition-coordinator.js";
 export { LeaseHealthMonitor } from "./lease-health-monitor.js";
 export { type PhysicalDeviceWatcher } from "./physical-device-watcher.js";
-export { type DeviceModeReader, type LeaseCommands, type QueueControl } from "./lease-ports.js";
+export {
+  type DeviceModeReader,
+  // fallow-ignore-next-line unused-type -- the port for optional modules (#474 registers the first participant).
+  type EndedLease,
+  type LeaseCommands,
+  // fallow-ignore-next-line unused-type -- the port for optional modules (#474 registers the first participant).
+  type LeaseEndParticipant,
+  type QueueControl,
+} from "./lease-ports.js";
 export {
   IdempotencyConflictError,
   InMemoryLeaseRequestStore,

src/leasing/lease-acquisition-coordinator.test.ts

@@ -97,6 +97,7 @@ function config(maxDevices = 1, maxRunning = 1): Config {
       requestRetentionMs: 600_000,
       maxRequestRecords: 10_000,
       maxPerRequester: 1,
+      endWaitMs: 600_000,
     },
     capacity: {
       strategy: "resource",