fix(remote): renew the lease while an artifact upload runs (#2946)

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

@@ -67,6 +67,56 @@ one minute, while a cloud WebDriver connection profile asks for ten. A single co
 longer than its own lease is therefore ordinary on the default and only reachable through a profile
 on the longer one.
 
+## Client-side work that precedes admission
+
+Protecting admitted work covers nothing that happens before a request is admitted. Installing an
+artifact uploads it from the caller while the request that will consume it has not been admitted yet,
+so an upload slower than the lease's inactivity window expired the lease paying for the device the
+bytes were going to (#2946). The caller therefore beats the lease over the ordinary transport for as
+long as that phase runs, naming the lease scope exactly as the command named it and no payload of its
+own. An install names no window, so its beats renew for the window the lease already carries; a
+caller that did name one keeps renewing on it. Resolving an absent window to the registry default
+instead — which is what a heartbeat used to do — quietly shortened every lease allocated above that
+default, which is the other half of why the upload could not survive. Request admission had its own
+copy of that mistake: it named a proxy-specific default for every admitted request, so a lease
+allocated longer than the default lost its window on the next command. Admission renews on the lease's
+window too; the window a lease carries is the one its client named when it allocated.
+
+The first beat is fired when the phase starts, not one cadence in, because a beat is what proves the
+lease the upload is spending time on is still alive — a lease shorter than any fixed cadence would
+otherwise lapse before anything renewed it. A lease already gone is caught early, but not before the
+phase begins: hashing, the preflight, and the start of the stream can all run while the first beat is
+still outstanding, so what the first beat buys is that the loss is learned during the upload rather
+than after it. Each beat answers with the window it just renewed, and the loop arms its successor when
+the beat starts rather than when it settles, so a beat that never answers is abandoned on schedule
+instead of taking the schedule with it. A phase that settles does not wait on a beat still in flight.
+
+Cadence and budget are deliberately two numbers. A beat is armed every third of the window, but it is
+allowed the whole window to answer in. Budgeting a beat at its cadence instead would give a heartbeat
+that needs more than a third of a window for its round trip — a tunneled proxy, an uplink the upload
+itself is saturating — no way to answer at all, and every beat would re-time-out on exactly the slow
+links the beat exists to protect. The window, not the cadence, is the interval in which being alive
+still matters, so it is also the longest a beat's answer is worth waiting for.
+
+That ceiling is the shorter of the window and the heartbeat's own request policy, currently ninety
+seconds. A ten-minute lease therefore gives a beat ninety seconds rather than the full window, which
+costs nothing the beat exists to buy: the daemon renews the lease when it handles the request, before
+it answers, so a round trip abandoned at the cap has still extended the lease if the request ever
+landed — and if it never landed, a transport that has stopped carrying traffic is not served by holding
+the socket open for the rest of a window measured in minutes. What the cap refuses is exactly that
+idle hold. It binds only on windows longer than the policy, and on those the beat already has the
+longest answer deadline any lease gets; shorter windows are bounded by their own window, as above.
+
+A beat is a fresh request each time, never the protected request rewritten: a request identity is
+what a timed-out beat is canceled under, and beats must not inherit each other's cancellation. A beat
+that finds the lease gone, or finds that this request can never renew it — its scope is missing or
+belongs to another lease, or the daemon rejects the scope and window the beat itself was built with,
+which no successor will ask differently — ends the phase with that error and cancels the upload rather
+than finishing bytes against a device nobody owns or a lease that will stop renewing. A beat that fails
+for any other reason is reported and survived, because a later beat covers one lost request — including
+one that was abandoned at its budget rather than answered, which is what makes that promise hold for a
+stalled round trip and not only for one that fails fast.
+
 ## Human control
 
 Human-control holds coexist with an open remote session. They belong to `LeaseRegistry` and use

packages/contracts/src/__tests__/lease-scope.test.ts

@@ -2,6 +2,7 @@ import assert from 'node:assert/strict';
 import { test } from 'vitest';
 import {
   findMissingProxyLeaseFields,
+  isInactiveLeaseError,
   leaseScopeFromOptions,
   leaseScopeFromRequest,
   leaseScopeToCommandFlags,
@@ -215,3 +216,29 @@ test('readLeaseAllocateProviderFlags carries the provider-allocation flags and d
   );
   assert.deepEqual(readLeaseAllocateProviderFlags(undefined), {});
 });
+
+test('isInactiveLeaseError recognizes the lease being gone, and refuses a request mismatch', () => {
+  for (const reason of ['LEASE_NOT_FOUND', 'LEASE_EXPIRED', 'LEASE_REVOKED']) {
+    assert.equal(
+      isInactiveLeaseError({ code: 'UNAUTHORIZED', details: { reason } }),
+      true,
+      `${reason} says the lease is no longer usable`,
+    );
+  }
+  // A mismatch is about the request that asked, not the lease: the lease may be perfectly alive and
+  // held by this same client under another request, so a beat must keep renewing on this answer.
+  assert.equal(
+    isInactiveLeaseError({ code: 'UNAUTHORIZED', details: { reason: 'LEASE_SCOPE_MISMATCH' } }),
+    false,
+    'a scope mismatch is not the lease being gone',
+  );
+  // Both halves are required: the code alone is any unauthorized call, and the reason alone could
+  // ride on an error the client has no business acting on.
+  assert.equal(isInactiveLeaseError({ code: 'UNAUTHORIZED', details: {} }), false);
+  assert.equal(
+    isInactiveLeaseError({ code: 'COMMAND_FAILED', details: { reason: 'LEASE_NOT_FOUND' } }),
+    false,
+  );
+  assert.equal(isInactiveLeaseError(new Error('LEASE_NOT_FOUND')), false);
+  assert.equal(isInactiveLeaseError(undefined), false);
+});

packages/contracts/src/lease-scope.ts

@@ -4,7 +4,6 @@ import type { CloudProviderProfileFields } from './remote-config-fields.ts';
 import type { CommandFlags } from './command-flags.ts';
 
 const PROXY_LEASE_PROVIDER = 'proxy';
-export const DEFAULT_PROXY_LEASE_TTL_MS = 300_000;
 
 const REQUIRED_PROXY_LEASE_FIELDS = [
   'leaseId',
@@ -296,6 +295,41 @@ export function findMissingProxyLeaseFields(scope: LeaseScope): string[] {
   return REQUIRED_PROXY_LEASE_FIELDS.filter((field) => !scope[field]);
 }
 
+/**
+ * The shortest inactivity window a lease can hold by default.
+ *
+ * The registry clamps every ttl to this floor, so it is also the worst case a client plans against
+ * before it has seen a lease answer: work that has to finish inside a lease window it does not know
+ * yet must assume this one, or it can be planning for a window the daemon would never grant.
+ */
+export const MIN_LEASE_WINDOW_MS = 5_000;
+
+/**
+ * Why a lease stopped being ours: it is gone, spent, or taken back.
+ *
+ * This is the whole taxonomy of "the lease is no longer usable" as the daemon reports it, and both
+ * readers ask the same question — a client deciding whether a connection still owns a device, and a
+ * lease beat deciding whether an upload is still worth finishing. A reason naming a mismatch between
+ * a request and a lease is not in here: that says something about the request, not the lease.
+ */
+const INACTIVE_LEASE_REASONS: ReadonlySet<unknown> = new Set([
+  'LEASE_NOT_FOUND',
+  'LEASE_EXPIRED',
+  'LEASE_REVOKED',
+]);
+
+/** Whether `error` says the lease is gone rather than merely unavailable to this request. */
+export function isInactiveLeaseError(error: unknown): boolean {
+  return (
+    typeof error === 'object' &&
+    error !== null &&
+    (error as Readonly<{ code?: unknown }>).code === 'UNAUTHORIZED' &&
+    INACTIVE_LEASE_REASONS.has(
+      (error as Readonly<{ details?: Readonly<{ reason?: unknown }> }>).details?.reason,
+    )
+  );
+}
+
 function leaseScopeToScopedRequest(scope: LeaseScope): LeaseScopedRequestScope {
   return stripUndefined({
     leaseId: scope.leaseId ?? '',

scripts/layering/daemon-client-entry.ts

@@ -43,6 +43,11 @@ export const DAEMON_CLIENT_ENTRY_EDGES: readonly DaemonClientEntryEdge[] = [
     target: 'src/daemon/daemon-request.ts',
     rationale: REQUEST_RATIONALE,
   },
+  {
+    file: 'src/daemon-client/daemon-client-lease-beat.ts',
+    target: 'src/daemon/daemon-request.ts',
+    rationale: REQUEST_RATIONALE,
+  },
   {
     file: 'src/daemon-client/daemon-client-progress.ts',
     target: 'src/daemon/daemon-request.ts',

src/__tests__/upload-client-cancellation.test.ts

@@ -0,0 +1,199 @@
+// Ending an artifact upload when the work it serves is over (#2946).
+//
+// A beat that finds the device's lease gone aborts the upload protecting that lease. An upload is a
+// piped `node:http` request, so the only way to stop bytes already in flight is the request's own
+// abort signal — a rejection the caller swallows would keep streaming a full app bundle to a device
+// nobody owns. Kept out of `upload-client.test.ts`, which is already over the test-file size
+// tripwire and may not grow (docs/agents/testing.md).
+
+import { afterEach, test } from 'vitest';
+import assert from 'node:assert/strict';
+import { createHash } from 'node:crypto';
+import fs from 'node:fs';
+import http, { type IncomingMessage, type ServerResponse } from 'node:http';
+import path from 'node:path';
+import { once } from 'node:events';
+import { uploadArtifact } from '../remote/upload-client.ts';
+import { mkdtempForTestSync } from './test-utils/tmp-dir.ts';
+
+const TEST_TOKEN = 'agent-device-upload-cancel-token';
+const tempDirs: string[] = [];
+
+afterEach(async () => {
+  for (const dir of tempDirs) {
+    await fs.promises.rm(dir, { recursive: true, force: true }).catch(() => {});
+  }
+  tempDirs.length = 0;
+});
+
+test('an aborted signal ends a legacy upload mid-stream instead of finishing the bytes', async () => {
+  // Two megabytes is well past what a paused read lets through: the assertions below are about the
+  // stream stopping, and a smaller payload keeps that off the CPU in a loaded lane.
+  const content = Buffer.alloc(2 * 1024 * 1024, 'x');
+  const artifactPath = createTempFile('app.apk', content);
+  const control = new AbortController();
+  let sawBytes = 0;
+
+  // Preflight reports itself unsupported so the upload takes the legacy stream, the one path that
+  // pipes a file at the daemon and keeps going for as long as the daemon drains it.
+  const server = await startServer(async (req, res) => {
+    if (req.method === 'POST' && req.url === '/upload/preflight') {
+      await readRequestBody(req);
+      res.statusCode = 404;
+      res.end('not found');
+      return;
+    }
+    if (req.method === 'POST' && req.url === '/upload') {
+      req.on('data', (chunk: Buffer) => {
+        sawBytes += chunk.length;
+        // The lease dies once bytes are genuinely in flight — not before the request starts.
+        // Stopping the read applies backpressure so the rest of the file cannot race through
+        // loopback and make "stopped early" a matter of timing.
+        req.pause();
+        if (!control.signal.aborted) control.abort();
+      });
+      return;
+    }
+    res.statusCode = 404;
+    res.end('not found');
+  });
+
+  try {
+    await assert.rejects(
+      async () =>
+        await uploadArtifact({
+          localPath: artifactPath,
+          baseUrl: server.baseUrl,
+          token: TEST_TOKEN,
+          signal: control.signal,
+        }),
+      isAbortError,
+    );
+    assert.ok(sawBytes > 0, 'the upload had started streaming before the abort');
+    assert.ok(sawBytes < content.length, `the stream stopped early, at ${sawBytes} bytes`);
+  } finally {
+    await server.close();
+  }
+});
+
+test('an aborted signal before preflight refuses to ask the daemon for a ticket', async () => {
+  const artifactPath = createTempFile('app.apk', 'payload');
+  const control = new AbortController();
+  control.abort();
+  const requests: string[] = [];
+
+  const server = await startServer(async (req, res) => {
+    requests.push(`${req.method} ${req.url}`);
+    res.statusCode = 404;
+    res.end('not found');
+  });
+
+  try {
+    await assert.rejects(
+      async () =>
+        await uploadArtifact({
+          localPath: artifactPath,
+          baseUrl: server.baseUrl,
+          token: TEST_TOKEN,
+          signal: control.signal,
+        }),
+      isAbortError,
+    );
+    assert.deepEqual(requests, [], 'an upload nobody waits for never reaches the daemon');
+  } finally {
+    await server.close();
+  }
+});
+
+test('an upload with no signal behaves exactly as before', async () => {
+  const content = 'unprotected-payload';
+  const artifactPath = createTempFile('app.apk', content);
+  const expectedHash = createHash('sha256').update(content).digest('hex');
+
+  const server = await startServer(async (req, res) => {
+    if (req.method === 'POST' && req.url === '/upload/preflight') {
+      await readRequestBody(req);
+      res.statusCode = 404;
+      res.end('not found');
+      return;
+    }
+    if (req.method === 'POST' && req.url === '/upload') {
+      assert.equal(req.headers['x-artifact-hash'], expectedHash);
+      await readRequestBody(req);
+      sendJson(res, { ok: true, uploadId: 'upload-uncancelled' });
+      return;
+    }
+    res.statusCode = 404;
+    res.end('not found');
+  });
+
+  try {
+    const uploadId = await uploadArtifact({
+      localPath: artifactPath,
+      baseUrl: server.baseUrl,
+      token: TEST_TOKEN,
+    });
+    assert.equal(uploadId, 'upload-uncancelled');
+  } finally {
+    await server.close();
+  }
+});
+
+function isAbortError(error: unknown): boolean {
+  // The contract on `signal` is that an aborted upload rejects with the signal's own reason. Any
+  // other rejection — a transport failure, the server-error fallback — means the upload stopped for
+  // a reason this test is not about and would let the abort path regress while staying green.
+  return error instanceof DOMException && error.name === 'AbortError';
+}
+
+function createTempFile(filename: string, content: string | Buffer): string {
+  const dir = mkdtempForTestSync('agent-device-upload-cancel-');
+  tempDirs.push(dir);
+  const filePath = path.join(dir, filename);
+  fs.writeFileSync(filePath, content);
+  return filePath;
+}
+
+async function startServer(
+  handler: (req: IncomingMessage, res: ServerResponse) => Promise<void>,
+): Promise<{ baseUrl: string; close: () => Promise<void> }> {
+  const server = http.createServer((req, res) => {
+    void handler(req, res).catch(() => {
+      res.statusCode = 500;
+      res.end();
+    });
+  });
+  // A canceled upload leaves its socket in a state `closeAllConnections` can race; without a bound
+  // idle keep-alive, `close` would then wait out Node's five-second default.
+  server.keepAliveTimeout = 100;
+  server.listen(0, '127.0.0.1');
+  server.unref();
+  await once(server, 'listening');
+  const address = server.address();
+  assert.ok(address && typeof address === 'object');
+  return {
+    baseUrl: `http://127.0.0.1:${address.port}`,
+    close: async () => {
+      // A canceled upload leaves its socket half-open and a pooled keep-alive one behind; without
+      // dropping them, `close` would wait out the server's five-second keep-alive timeout.
+      server.closeAllConnections();
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    },
+  };
+}
+
+async function readRequestBody(req: IncomingMessage): Promise<Buffer> {
+  const chunks: Buffer[] = [];
+  for await (const chunk of req) {
+    chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
+  }
+  return Buffer.concat(chunks);
+}
+
+function sendJson(res: ServerResponse, body: unknown): void {
+  res.statusCode = 200;
+  res.setHeader('content-type', 'application/json');
+  res.end(JSON.stringify(body));
+}

src/cli/commands/connection-runtime.ts

@@ -15,6 +15,7 @@ import {
   type DeviceInfo,
 } from '@agent-device/kernel/device';
 import { shouldAgentCdpUseRemoteBridgeUrl } from './agent-cdp.ts';
+import { isInactiveLeaseError } from '@agent-device/contracts/lease-scope';
 import {
   buildRemoteConnectionDaemonState,
   buildRemoteConnectionRequestMetadata,
@@ -991,12 +992,3 @@ async function heartbeatOrAllocateLease(
     throw error;
   }
 }
-
-function isInactiveLeaseError(error: unknown): boolean {
-  if (!(error instanceof AppError) || error.code !== 'UNAUTHORIZED') return false;
-  return (
-    error.details?.reason === 'LEASE_NOT_FOUND' ||
-    error.details?.reason === 'LEASE_EXPIRED' ||
-    error.details?.reason === 'LEASE_REVOKED'
-  );
-}