From e2dd51921f919f1deaafee77dc81114cb77ba03d Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Tue, 18 Aug 2026 20:17:06 +0200
Subject: [PATCH 1/5] feat(observability): authenticate hosted reporter

---
 src/cli/fleet.test.ts                     | 158 +++++++++++++++++++++-
 src/cli/fleet.ts                          |  48 +++++--
 src/mount/relayfile-cloud-mount-client.ts |  17 ++-
 3 files changed, 208 insertions(+), 15 deletions(-)

diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index 4c2b007a..8af92819 100644
--- a/src/cli/fleet.test.ts
+++ b/src/cli/fleet.test.ts
@@ -15,7 +15,12 @@ import type {
   FactoryPorts,
   createFactory,
 } from '../index'
-import { FactoryConfigSchema, LiveDispatchStateChangedError, stateResolutionFromIds } from '../index'
+import {
+  FactoryConfigSchema,
+  FileFactoryCloudEventOutbox,
+  LiveDispatchStateChangedError,
+  stateResolutionFromIds,
+} from '../index'
 import { MountAuthScopeError, mountAuthRemediation } from '../mount/mount-auth-error'
 import { DocumentStateStore, FileStateStore } from '../state/file-state-store'
 import { FakeFleetClient, FakeMountClient, withDeadline } from '../testing'
@@ -1525,6 +1530,157 @@ describe('fleet CLI runtime', () => {
     }
   })
 
+  it('uses the hosted rotating path token for Cloud reporting without local login', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-hosted-reporting-'))
+    try {
+      const outboxPath = join(root, 'factory-cloud-events.json')
+      const configPath = await writeConfig(root, {
+        workspaceId: 'rw_7ccfea89',
+        loop: {
+          heartbeatPath: join(root, 'heartbeat.json'),
+          registryPath: join(root, 'registry.json'),
+        },
+        reporting: {
+          enabled: true,
+          instanceName: 'factory-khaliq-cloud',
+          outboxPath,
+          batchSize: 100,
+          requestTimeoutMs: 1_000,
+        },
+      })
+      const factory = {
+        start: vi.fn(),
+        stop: vi.fn(),
+        runLoop: vi.fn(async () => []),
+        runOnce: vi.fn(async () => ({ pulled: [], triaged: [], dispatched: [], skipped: [], dryRun: true })),
+        status: vi.fn(),
+        triageIssue: vi.fn(),
+        dispatch: vi.fn(),
+        on: vi.fn(),
+        dispose: vi.fn(async () => {}),
+      } as unknown as Factory
+      let capturedReporter: FactoryEventReporter | undefined
+      const createFactorySpy = vi.fn((_config, ports: FactoryPorts) => {
+        capturedReporter = ports.reporter
+        return factory
+      }) as typeof createFactory
+      const cloudSessionProvider = vi.fn(async () => {
+        throw new Error('local Cloud login must not be consulted')
+      })
+      const cloudAccessTokenFetch = vi.fn(async () =>
+        Response.json({ accessToken: 'relay_pa_test' }))
+      const batches: Array<Record<string, unknown>> = []
+      const cloudReporterFetch = vi.fn(async (request: string | URL | Request, init?: RequestInit) => {
+        expect(String(request)).toBe('https://cloud.example/api/v1/factory/events')
+        expect(new Headers(init?.headers).get('authorization')).toBe(
+          'Bearer relay_pa_test',
+        )
+        const batch = JSON.parse(String(init?.body)) as Record<string, unknown>
+        batches.push(batch)
+        const accepted = Array.isArray(batch.events) ? batch.events.length : 0
+        return Response.json({ accepted, duplicates: 0 }, { status: 201 })
+      })
+
+      const code = await runFleetCli(['run-once', '--dry-run', '--config', configPath], {
+        env: {
+          FACTORY_CLOUD_ACCESS_TOKEN_URL: 'http://factory-auth.do/factory-primary/v1/access',
+          CLOUD_API_URL: 'https://cloud.example',
+        },
+        fleet: new FakeFleetClient(),
+        mount: new FakeMountClient(),
+        createFactory: createFactorySpy,
+        cloudSessionProvider,
+        cloudAccessTokenFetch: cloudAccessTokenFetch as unknown as typeof fetch,
+        cloudReporterFetch: cloudReporterFetch as unknown as typeof fetch,
+        stdout: buffer(),
+        stderr: buffer(),
+      })
+
+      expect(code).toBe(0)
+      expect(capturedReporter).toBeDefined()
+      expect(factory.runOnce).toHaveBeenCalledOnce()
+      expect(cloudSessionProvider).not.toHaveBeenCalled()
+      expect(cloudAccessTokenFetch).toHaveBeenCalledWith(
+        new URL('http://factory-auth.do/factory-primary/v1/access'),
+        expect.objectContaining({ method: 'GET', redirect: 'error' }),
+      )
+      expect(cloudReporterFetch).toHaveBeenCalled()
+      const events = batches.flatMap((batch) => (
+        Array.isArray(batch.events) ? batch.events as Array<{ type?: string }> : []
+      ))
+      expect(events.map((event) => event.type)).toEqual(expect.arrayContaining([
+        'instance.started',
+        'instance.stopping',
+        'instance.stopped',
+      ]))
+      expect(batches[0]?.instance).toMatchObject({
+        metadata: expect.objectContaining({ name: 'factory-khaliq-cloud' }),
+      })
+      await expect(new FileFactoryCloudEventOutbox({ path: outboxPath }).stats())
+        .resolves.toMatchObject({ pending: 0 })
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('keeps dispatch successful and the event pending when hosted telemetry returns 503', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-hosted-reporting-fail-open-'))
+    try {
+      const outboxPath = join(root, 'factory-cloud-events.json')
+      const configPath = await writeConfig(root, {
+        workspaceId: 'rw_7ccfea89',
+        loop: {
+          heartbeatPath: join(root, 'heartbeat.json'),
+          registryPath: join(root, 'registry.json'),
+        },
+        reporting: {
+          enabled: true,
+          instanceName: 'factory-khaliq-cloud',
+          outboxPath,
+          batchSize: 100,
+          requestTimeoutMs: 100,
+        },
+      })
+      const factory = {
+        start: vi.fn(),
+        stop: vi.fn(),
+        runLoop: vi.fn(async () => []),
+        runOnce: vi.fn(async () => ({ pulled: [], triaged: [], dispatched: [], skipped: [], dryRun: true })),
+        status: vi.fn(),
+        triageIssue: vi.fn(),
+        dispatch: vi.fn(),
+        on: vi.fn(),
+        dispose: vi.fn(async () => {}),
+      } as unknown as Factory
+      const cloudAccessTokenFetch = vi.fn(async () =>
+        Response.json({ accessToken: 'relay_pa_test' }))
+      const cloudReporterFetch = vi.fn(async () =>
+        new Response('unavailable', { status: 503 }))
+
+      const code = await runFleetCli(['run-once', '--dry-run', '--config', configPath], {
+        env: {
+          FACTORY_CLOUD_ACCESS_TOKEN_URL: 'http://factory-auth.do/factory-primary/v1/access',
+          CLOUD_API_URL: 'https://cloud.example',
+        },
+        fleet: new FakeFleetClient(),
+        mount: new FakeMountClient(),
+        createFactory: () => factory,
+        cloudAccessTokenFetch: cloudAccessTokenFetch as unknown as typeof fetch,
+        cloudReporterFetch: cloudReporterFetch as unknown as typeof fetch,
+        stdout: buffer(),
+        stderr: buffer(),
+      })
+
+      expect(code).toBe(0)
+      expect(factory.runOnce).toHaveBeenCalledOnce()
+      expect(cloudReporterFetch).toHaveBeenCalled()
+      const stats = await new FileFactoryCloudEventOutbox({ path: outboxPath }).stats()
+      expect(stats.pending).toBeGreaterThan(0)
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
   it('infers clonePath from cwd for internal dispatch and logs the checkout root', async () => {
     const root = await mkdtemp(join(tmpdir(), 'fleet-cli-infer-'))
     try {
diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts
index 5314a3c2..17785380 100644
--- a/src/cli/fleet.ts
+++ b/src/cli/fleet.ts
@@ -87,6 +87,11 @@ import type { FactoryIntegrationProvider } from '../ports'
 import type { StateStore } from '../ports/state'
 import { checkMountStaleness } from '../mount/relayfile-binary'
 import { MountAuthScopeError } from '../mount/mount-auth-error'
+import {
+  createHostedCloudAccessTokenProvider,
+  FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV,
+  resolveHostedCloudApiUrl,
+} from '../mount/relayfile-cloud-mount-client'
 import { resolveRelayWorkspaceKey } from '../fleet/relay-workspace-key'
 import {
   isMeaningfullyBehind,
@@ -149,6 +154,10 @@ export interface FleetCliDeps {
   localClonePathOptions?: LocalClonePathOptions
   reporter?: FactoryEventReporter
   cloudSessionProvider?: (options?: Parameters<typeof ensureCloudSession>[0]) => Promise<CloudSession>
+  /** Hermetic private hosted-token endpoint transport for CLI integration tests. */
+  cloudAccessTokenFetch?: typeof fetch
+  /** Hermetic Cloud telemetry transport for CLI integration tests. */
+  cloudReporterFetch?: typeof fetch
   isInteractive?: () => boolean
   confirmIntegrationConnect?: (provider: FactoryIntegrationProvider) => Promise<boolean>
   openIntegrationUrl?: (url: string) => void | Promise<void>
@@ -1582,10 +1591,20 @@ async function buildFactoryCloudReporter(input: {
   deps: FleetCliDeps
 }): Promise<FactoryEventReporter | undefined> {
   if (!input.config.reporting.enabled) return undefined
-  if (hasInjectedFactoryRuntime(input.deps) && !input.deps.cloudSessionProvider) return undefined
+  const runtimeEnv = input.deps.env ?? process.env
+  const hasHostedAccessTokenConfig = Object.prototype.hasOwnProperty.call(
+    runtimeEnv,
+    FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV,
+  )
+  if (
+    hasInjectedFactoryRuntime(input.deps)
+    && !input.deps.cloudSessionProvider
+    && !hasHostedAccessTokenConfig
+  ) return undefined
 
   try {
-    const activeWorkspace = await (input.deps.resolveWorkspace ?? resolveFactoryWorkspace)()
+    const activeWorkspace = await (input.deps.resolveWorkspace
+      ?? (() => resolveFactoryWorkspace(undefined, runtimeEnv)))()
     const activeWorkspaceIds = new Set([
       activeWorkspace.workspaceId,
       activeWorkspace.cloudWorkspaceId,
@@ -1594,15 +1613,28 @@ async function buildFactoryCloudReporter(input: {
       input.logger.warn?.('[factory] Cloud progress reporting skipped because the active account workspace differs from Factory config')
       return undefined
     }
-    const session = await (input.deps.cloudSessionProvider ?? ensureCloudSession)({ interactive: false })
     const outboxPath = input.config.reporting.outboxPath
       ?? join(dirname(input.config.loop.registryPath), 'factory-cloud-events.json')
     const instanceId = await loadOrCreateFactoryInstanceId(`${outboxPath}.instance-id`)
     const instanceName = resolveFactoryInstanceName(input.config)
-    const cloudFetch: typeof fetch = async (_request, init) =>
-      session.client.fetch('/api/v1/factory/events', init)
+    let apiUrl: string
+    let getAccessToken: () => Promise<string>
+    let cloudFetch: typeof fetch | undefined
+    if (hasHostedAccessTokenConfig) {
+      apiUrl = resolveHostedCloudApiUrl(runtimeEnv)
+      getAccessToken = createHostedCloudAccessTokenProvider({
+        url: runtimeEnv[FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV]?.trim() ?? '',
+        fetchImpl: input.deps.cloudAccessTokenFetch ?? fetch,
+      })
+      cloudFetch = input.deps.cloudReporterFetch
+    } else {
+      const session = await (input.deps.cloudSessionProvider ?? ensureCloudSession)({ interactive: false })
+      apiUrl = session.auth.apiUrl
+      getAccessToken = async () => session.client.snapshot().accessToken
+      cloudFetch = async (_request, init) => session.client.fetch('/api/v1/factory/events', init)
+    }
     return new FactoryCloudReporter({
-      apiUrl: session.auth.apiUrl,
+      apiUrl,
       instance: {
         id: instanceId,
         bootId: randomUUID(),
@@ -1615,8 +1647,8 @@ async function buildFactoryCloudReporter(input: {
         },
       },
       outbox: new FileFactoryCloudEventOutbox({ path: outboxPath }),
-      getAccessToken: async () => session.client.snapshot().accessToken,
-      fetch: cloudFetch,
+      getAccessToken,
+      ...(cloudFetch ? { fetch: cloudFetch } : {}),
       logger: input.logger,
       batchSize: input.config.reporting.batchSize,
       requestTimeoutMs: input.config.reporting.requestTimeoutMs,
diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts
index 83c3e559..fe2cde62 100644
--- a/src/mount/relayfile-cloud-mount-client.ts
+++ b/src/mount/relayfile-cloud-mount-client.ts
@@ -83,6 +83,10 @@ export const FACTORY_RELAYFILE_SCOPES = [
   'relayfile:fs:write:/factory/observability/**',
 ] as const
 
+export const resolveHostedCloudApiUrl = (
+  env: NodeJS.ProcessEnv = process.env,
+): string => env.CLOUD_API_URL?.trim() || defaultApiUrl()
+
 export type CloudSessionProvider = (options?: CloudSessionOptions) => Promise<CloudSession>
 
 export type ActiveWorkspaceResolver = (
@@ -412,7 +416,7 @@ export class RelayfileCloudMountClient implements MountClient {
     }
     const cloudApiUrl = config.cloudApiUrl
       ?? initialSession?.auth.apiUrl
-      ?? (hostedTokenProvider ? (runtimeEnv.CLOUD_API_URL?.trim() || defaultApiUrl()) : undefined)
+      ?? (hostedTokenProvider ? resolveHostedCloudApiUrl(runtimeEnv) : undefined)
     if (!cloudApiUrl) {
       throw new Error('Relayfile hosted access requires cloudApiUrl with cloudAccessTokenProvider')
     }
@@ -1015,11 +1019,12 @@ const createValidatedHostedAccessTokenProvider = (
   return accessToken
 }
 
-const createHostedCloudAccessTokenProvider = (options: {
+export const createHostedCloudAccessTokenProvider = (options: {
   url: string
   fetchImpl: typeof fetch
-  timeoutMs: number
+  timeoutMs?: number
 }): (() => Promise<string>) => {
+  const timeoutMs = options.timeoutMs ?? DEFAULT_HOSTED_ACCESS_TOKEN_TIMEOUT_MS
   let url: URL
   try {
     url = new URL(options.url)
@@ -1029,13 +1034,13 @@ const createHostedCloudAccessTokenProvider = (options: {
   if (url.protocol !== 'http:' && url.protocol !== 'https:') {
     throw new Error(`${FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV} must use http or https`)
   }
-  if (!Number.isFinite(options.timeoutMs) || options.timeoutMs <= 0) {
+  if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
     throw new Error('hosted Cloud access-token timeout must be positive')
   }
 
   return async (): Promise<string> => {
     const controller = new AbortController()
-    const timer = setTimeout(() => controller.abort(), options.timeoutMs)
+    const timer = setTimeout(() => controller.abort(), timeoutMs)
     try {
       const response = await options.fetchImpl(url, {
         method: 'GET',
@@ -1062,7 +1067,7 @@ const createHostedCloudAccessTokenProvider = (options: {
       return accessToken
     } catch (error) {
       if (error instanceof Error && error.name === 'AbortError') {
-        throw new Error(`hosted Cloud access-token provider timed out after ${String(options.timeoutMs)}ms`)
+        throw new Error(`hosted Cloud access-token provider timed out after ${String(timeoutMs)}ms`)
       }
       throw error
     } finally {

From 5d7c816e0104252c06c054996038f296615952ee Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Wed, 19 Aug 2026 23:10:49 +0200
Subject: [PATCH 2/5] fix(observability): bind hosted credential fetches to
 reporter shutdown

Reporter close({deadlineMs}) only stopped awaiting the flush; the hosted
token request it left behind kept a referenced 10s abort timer and a live
socket, so the process could outlive the shutdown deadline by ~30s across
three delivery attempts.

- FactoryCloudReporter now owns a shutdown AbortController. Token and event
  requests are bound to it and to the flush deadline, and a provider that
  rejects because shutdown cancelled it stops delivery instead of retrying.
- createHostedCloudAccessTokenProvider accepts a caller signal, cancels the
  in-flight fetch when it aborts, and unrefs its timeout timer.
- The default retry sleep is unref'd too.

Also gate hosted reporting on a trimmed non-empty FACTORY_CLOUD_ACCESS_TOKEN_URL
rather than the variable's presence. A blank value entered the hosted branch and
threw on `new URL('')`, discarding telemetry instead of falling back to the
local-session path. resolveFactoryWorkspace() uses the same gate.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Session-Id: fc9722a8-8ef1-47f2-82d7-470a4ecfb9b1
---
 src/cli/fleet.test.ts                         | 138 ++++++++++++++++++
 src/cli/fleet.ts                              |  19 +--
 .../relayfile-cloud-mount-client.test.ts      |  50 +++++++
 src/mount/relayfile-cloud-mount-client.ts     |  53 ++++++-
 src/observability/cloud-reporter.test.ts      |  49 +++++++
 src/observability/cloud-reporter.ts           |  60 +++++++-
 6 files changed, 346 insertions(+), 23 deletions(-)

diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index 8af92819..3fd52f4f 100644
--- a/src/cli/fleet.test.ts
+++ b/src/cli/fleet.test.ts
@@ -1,4 +1,5 @@
 import { describe, expect, it, vi } from 'vitest'
+import type { CloudSession } from '@agent-relay/cloud'
 import { existsSync } from 'node:fs'
 import { mkdir, mkdtemp, readFile, rm, symlink, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
@@ -1623,6 +1624,143 @@ describe('fleet CLI runtime', () => {
     }
   })
 
+  it('cancels a hanging hosted credential request when the CLI shuts down', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-hosted-token-shutdown-'))
+    try {
+      const outboxPath = join(root, 'factory-cloud-events.json')
+      const configPath = await writeConfig(root, {
+        workspaceId: 'rw_7ccfea89',
+        loop: {
+          heartbeatPath: join(root, 'heartbeat.json'),
+          registryPath: join(root, 'registry.json'),
+        },
+        reporting: {
+          enabled: true,
+          outboxPath,
+          batchSize: 100,
+          requestTimeoutMs: 100,
+        },
+      })
+      const factory = {
+        start: vi.fn(),
+        stop: vi.fn(),
+        runLoop: vi.fn(async () => []),
+        runOnce: vi.fn(async () => ({ pulled: [], triaged: [], dispatched: [], skipped: [], dryRun: true })),
+        status: vi.fn(),
+        triageIssue: vi.fn(),
+        dispatch: vi.fn(),
+        on: vi.fn(),
+        dispose: vi.fn(async () => {}),
+      } as unknown as Factory
+      const tokenSignals: Array<AbortSignal | undefined> = []
+      // The private credential endpoint accepts the connection and never answers.
+      const cloudAccessTokenFetch = vi.fn(async (_url: unknown, init?: RequestInit) => {
+        tokenSignals.push(init?.signal ?? undefined)
+        return await new Promise<Response>((_resolve, reject) => {
+          init?.signal?.addEventListener('abort', () => {
+            reject(Object.assign(new Error('aborted'), { name: 'AbortError' }))
+          }, { once: true })
+        })
+      })
+
+      const code = await runFleetCli(['run-once', '--dry-run', '--config', configPath], {
+        env: {
+          FACTORY_CLOUD_ACCESS_TOKEN_URL: 'http://factory-auth.do/factory-primary/v1/access',
+          CLOUD_API_URL: 'https://cloud.example',
+        },
+        fleet: new FakeFleetClient(),
+        mount: new FakeMountClient(),
+        createFactory: () => factory,
+        cloudAccessTokenFetch: cloudAccessTokenFetch as unknown as typeof fetch,
+        stdout: buffer(),
+        stderr: buffer(),
+      })
+
+      expect(code).toBe(0)
+      expect(tokenSignals.length).toBeGreaterThan(0)
+      // Shutdown must leave no referenced credential request behind, or the
+      // process outlives the reporter's close() deadline.
+      expect(tokenSignals.map((signal) => signal?.aborted)).toEqual(tokenSignals.map(() => true))
+    } finally {
+      // The cancelled flush unwinds its outbox writes just after the CLI
+      // returns; retry rmdir rather than race it.
+      await rm(root, { recursive: true, force: true, maxRetries: 20, retryDelay: 25 })
+    }
+  })
+
+  it('falls back to the local Cloud session when the hosted credential URL is blank', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-hosted-blank-url-'))
+    try {
+      const outboxPath = join(root, 'factory-cloud-events.json')
+      const configPath = await writeConfig(root, {
+        workspaceId: 'rw_7ccfea89',
+        loop: {
+          heartbeatPath: join(root, 'heartbeat.json'),
+          registryPath: join(root, 'registry.json'),
+        },
+        reporting: {
+          enabled: true,
+          outboxPath,
+          batchSize: 100,
+          requestTimeoutMs: 1_000,
+        },
+      })
+      const factory = {
+        start: vi.fn(),
+        stop: vi.fn(),
+        runLoop: vi.fn(async () => []),
+        runOnce: vi.fn(async () => ({ pulled: [], triaged: [], dispatched: [], skipped: [], dryRun: true })),
+        status: vi.fn(),
+        triageIssue: vi.fn(),
+        dispatch: vi.fn(),
+        on: vi.fn(),
+        dispose: vi.fn(async () => {}),
+      } as unknown as Factory
+      const sessionBatches: Array<Record<string, unknown>> = []
+      const cloudSessionProvider = vi.fn(async () => ({
+        auth: { apiUrl: 'https://cloud.example' },
+        client: {
+          snapshot: () => ({ accessToken: 'relay_pa_session' }),
+          fetch: async (_path: string, init?: RequestInit) => {
+            const batch = JSON.parse(String(init?.body)) as Record<string, unknown>
+            sessionBatches.push(batch)
+            const accepted = Array.isArray(batch.events) ? batch.events.length : 0
+            return Response.json({ accepted, duplicates: 0 }, { status: 201 })
+          },
+        },
+      }) as unknown as CloudSession)
+      const cloudAccessTokenFetch = vi.fn(async () => {
+        throw new Error('hosted credential endpoint must not be consulted')
+      })
+
+      const code = await runFleetCli(['run-once', '--dry-run', '--config', configPath], {
+        env: {
+          FACTORY_CLOUD_ACCESS_TOKEN_URL: '   ',
+          CLOUD_API_URL: 'https://cloud.example',
+        },
+        fleet: new FakeFleetClient(),
+        mount: new FakeMountClient(),
+        createFactory: () => factory,
+        resolveWorkspace: async () => ({ workspaceId: 'rw_7ccfea89' }),
+        cloudSessionProvider,
+        cloudAccessTokenFetch: cloudAccessTokenFetch as unknown as typeof fetch,
+        stdout: buffer(),
+        stderr: buffer(),
+      })
+
+      expect(code).toBe(0)
+      // A blank URL is not hosted configuration: the local-session path owns
+      // telemetry instead of `new URL('')` throwing the reporter away.
+      expect(cloudSessionProvider).toHaveBeenCalledOnce()
+      expect(cloudAccessTokenFetch).not.toHaveBeenCalled()
+      expect(sessionBatches.length).toBeGreaterThan(0)
+      await expect(new FileFactoryCloudEventOutbox({ path: outboxPath }).stats())
+        .resolves.toMatchObject({ pending: 0 })
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
   it('keeps dispatch successful and the event pending when hosted telemetry returns 503', async () => {
     const root = await mkdtemp(join(tmpdir(), 'fleet-cli-hosted-reporting-fail-open-'))
     try {
diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts
index 17785380..007c5af8 100644
--- a/src/cli/fleet.ts
+++ b/src/cli/fleet.ts
@@ -53,6 +53,7 @@ import {
   resolveFactoryWorkspace,
   type Capability,
   type Factory,
+  type FactoryCloudAccessTokenProvider,
   type FactoryEventReporter,
   type FactoryInFlightDispatchStatus,
   type FactoryInFlightRegistry,
@@ -89,7 +90,7 @@ import { checkMountStaleness } from '../mount/relayfile-binary'
 import { MountAuthScopeError } from '../mount/mount-auth-error'
 import {
   createHostedCloudAccessTokenProvider,
-  FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV,
+  hostedCloudAccessTokenUrl,
   resolveHostedCloudApiUrl,
 } from '../mount/relayfile-cloud-mount-client'
 import { resolveRelayWorkspaceKey } from '../fleet/relay-workspace-key'
@@ -1592,14 +1593,14 @@ async function buildFactoryCloudReporter(input: {
 }): Promise<FactoryEventReporter | undefined> {
   if (!input.config.reporting.enabled) return undefined
   const runtimeEnv = input.deps.env ?? process.env
-  const hasHostedAccessTokenConfig = Object.prototype.hasOwnProperty.call(
-    runtimeEnv,
-    FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV,
-  )
+  // Hosted mode is a usable endpoint, not a defined-but-blank variable: an
+  // empty value must fall through to the local-session path instead of
+  // reaching createHostedCloudAccessTokenProvider, which throws on `new URL('')`.
+  const hostedAccessTokenUrl = hostedCloudAccessTokenUrl(runtimeEnv)
   if (
     hasInjectedFactoryRuntime(input.deps)
     && !input.deps.cloudSessionProvider
-    && !hasHostedAccessTokenConfig
+    && !hostedAccessTokenUrl
   ) return undefined
 
   try {
@@ -1618,12 +1619,12 @@ async function buildFactoryCloudReporter(input: {
     const instanceId = await loadOrCreateFactoryInstanceId(`${outboxPath}.instance-id`)
     const instanceName = resolveFactoryInstanceName(input.config)
     let apiUrl: string
-    let getAccessToken: () => Promise<string>
+    let getAccessToken: FactoryCloudAccessTokenProvider
     let cloudFetch: typeof fetch | undefined
-    if (hasHostedAccessTokenConfig) {
+    if (hostedAccessTokenUrl) {
       apiUrl = resolveHostedCloudApiUrl(runtimeEnv)
       getAccessToken = createHostedCloudAccessTokenProvider({
-        url: runtimeEnv[FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV]?.trim() ?? '',
+        url: hostedAccessTokenUrl,
         fetchImpl: input.deps.cloudAccessTokenFetch ?? fetch,
       })
       cloudFetch = input.deps.cloudReporterFetch
diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts
index 0747eb4d..e388b654 100644
--- a/src/mount/relayfile-cloud-mount-client.test.ts
+++ b/src/mount/relayfile-cloud-mount-client.test.ts
@@ -12,6 +12,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 
 import {
+  createHostedCloudAccessTokenProvider,
   FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV,
   FACTORY_RELAYFILE_SCOPES,
   RelayfileCloudMountClient,
@@ -1172,6 +1173,55 @@ describe('RelayfileCloudMountClient', () => {
     expect(activeWorkspaceResolver).not.toHaveBeenCalled()
   })
 
+  it('cancels the hosted credential request when the caller aborts', async () => {
+    let requestSignal: AbortSignal | undefined
+    const fetchImpl = vi.fn(async (_url: unknown, init?: RequestInit) => {
+      requestSignal = init?.signal ?? undefined
+      return await new Promise<Response>((_resolve, reject) => {
+        init?.signal?.addEventListener('abort', () => {
+          reject(Object.assign(new Error('aborted'), { name: 'AbortError' }))
+        }, { once: true })
+      })
+    })
+    const provider = createHostedCloudAccessTokenProvider({
+      url: 'http://factory-auth.do/v1/access',
+      fetchImpl: fetchImpl as unknown as typeof fetch,
+      timeoutMs: 60_000,
+    })
+
+    const controller = new AbortController()
+    const pending = provider({ signal: controller.signal })
+    await vi.waitFor(() => { expect(fetchImpl).toHaveBeenCalledOnce() })
+    controller.abort()
+
+    await expect(pending).rejects.toThrow('hosted Cloud access-token request was cancelled')
+    expect(requestSignal?.aborted).toBe(true)
+  })
+
+  it('refuses a hosted credential request whose caller signal is already aborted', async () => {
+    const fetchImpl = vi.fn(async () => Response.json({ accessToken: 'relay_pa_never' }))
+    const provider = createHostedCloudAccessTokenProvider({
+      url: 'http://factory-auth.do/v1/access',
+      fetchImpl: fetchImpl as unknown as typeof fetch,
+    })
+
+    await expect(provider({ signal: AbortSignal.abort() }))
+      .rejects.toThrow('hosted Cloud access-token request was cancelled')
+    expect(fetchImpl).not.toHaveBeenCalled()
+  })
+
+  it('resolves a local Cloud workspace when the hosted credential variable is blank', async () => {
+    const activeWorkspaceResolver = vi.fn(async () => ({
+      relayfileWorkspaceId: 'rw_local',
+      cloudWorkspaceId: 'ws-uuid',
+    }))
+
+    await expect(resolveFactoryWorkspace(activeWorkspaceResolver, {
+      [FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV]: '   ',
+    })).resolves.toEqual({ workspaceId: 'rw_local', cloudWorkspaceId: 'ws-uuid' })
+    expect(activeWorkspaceResolver).toHaveBeenCalledOnce()
+  })
+
   it('cancels a failed hosted credential response body without reading it', async () => {
     const response = new Response('sensitive failure details', { status: 503 })
     const cancel = vi.spyOn(response.body!, 'cancel')
diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts
index fe2cde62..d8314567 100644
--- a/src/mount/relayfile-cloud-mount-client.ts
+++ b/src/mount/relayfile-cloud-mount-client.ts
@@ -83,6 +83,16 @@ export const FACTORY_RELAYFILE_SCOPES = [
   'relayfile:fs:write:/factory/observability/**',
 ] as const
 
+/**
+ * Hosted mode is defined by a usable endpoint, not by the variable merely
+ * existing. A blank or whitespace-only value is not hosted configuration — it
+ * must fall through to the local-session path rather than steer callers into a
+ * branch that throws on `new URL('')`.
+ */
+export const hostedCloudAccessTokenUrl = (
+  env: NodeJS.ProcessEnv = process.env,
+): string | undefined => env[FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV]?.trim() || undefined
+
 export const resolveHostedCloudApiUrl = (
   env: NodeJS.ProcessEnv = process.env,
 ): string => env.CLOUD_API_URL?.trim() || defaultApiUrl()
@@ -121,7 +131,7 @@ export async function resolveFactoryWorkspace(
   // that mode; doing so can invoke AGENT_RELAY_BIN before fromConfig() gets the
   // chance to use the hosted provider. fromConfig() remains responsible for
   // validating the configured endpoint URL and failing closed when malformed.
-  if (Object.prototype.hasOwnProperty.call(env, FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV)) {
+  if (hostedCloudAccessTokenUrl(env)) {
     return { workspaceId: DEFAULT_WORKSPACE_ID }
   }
   try {
@@ -211,7 +221,7 @@ export interface RelayfileCloudMountClientConfig {
    * Hosted runtimes inject a rotating, fixed RelayAuth path-token provider;
    * when present, Factory never reads the local Cloud login or shells out.
    */
-  cloudAccessTokenProvider?: () => Promise<string>
+  cloudAccessTokenProvider?: HostedCloudAccessTokenProvider
   /** Private host endpoint that returns the current fixed RelayAuth access token. */
   cloudAccessTokenUrl?: string
   /** Internal fetch override for hosted credential-provider tests. */
@@ -1009,21 +1019,36 @@ const createDefaultRelayfileSetup: RelayfileSetupFactory = ({ cloudApiUrl, token
     accessToken: tokenProvider,
   }) as unknown as RelayfileSetupLike
 
+/**
+ * Options a caller may pass per token request. `signal` binds the request's
+ * lifetime to the caller's: a reporter that has hit its shutdown deadline
+ * cancels the credential fetch instead of leaving it holding the event loop.
+ */
+export interface HostedCloudAccessTokenRequestOptions {
+  signal?: AbortSignal
+}
+
+export type HostedCloudAccessTokenProvider = (
+  options?: HostedCloudAccessTokenRequestOptions,
+) => Promise<string>
+
 const createValidatedHostedAccessTokenProvider = (
-  provider: () => Promise<string>,
-): (() => Promise<string>) => async () => {
-  const accessToken = (await provider()).trim()
+  provider: HostedCloudAccessTokenProvider,
+): HostedCloudAccessTokenProvider => async (requestOptions) => {
+  const accessToken = (await provider(requestOptions)).trim()
   if (!accessToken.startsWith('relay_pa_')) {
     throw new Error('hosted Cloud access-token provider returned an invalid token class')
   }
   return accessToken
 }
 
+const HOSTED_ACCESS_TOKEN_CANCELLED = 'hosted Cloud access-token request was cancelled'
+
 export const createHostedCloudAccessTokenProvider = (options: {
   url: string
   fetchImpl: typeof fetch
   timeoutMs?: number
-}): (() => Promise<string>) => {
+}): HostedCloudAccessTokenProvider => {
   const timeoutMs = options.timeoutMs ?? DEFAULT_HOSTED_ACCESS_TOKEN_TIMEOUT_MS
   let url: URL
   try {
@@ -1038,9 +1063,21 @@ export const createHostedCloudAccessTokenProvider = (options: {
     throw new Error('hosted Cloud access-token timeout must be positive')
   }
 
-  return async (): Promise<string> => {
+  return async (requestOptions?: HostedCloudAccessTokenRequestOptions): Promise<string> => {
+    const callerSignal = requestOptions?.signal
+    if (callerSignal?.aborted) throw new Error(HOSTED_ACCESS_TOKEN_CANCELLED)
     const controller = new AbortController()
+    let cancelled = false
+    const abortForCaller = (): void => {
+      cancelled = true
+      controller.abort()
+    }
+    callerSignal?.addEventListener('abort', abortForCaller, { once: true })
     const timer = setTimeout(() => controller.abort(), timeoutMs)
+    // An unreferenced timer cannot outlive the caller. Combined with the
+    // caller signal above, a hung credential endpoint can no longer keep the
+    // process alive past a reporter's shutdown deadline.
+    timer.unref?.()
     try {
       const response = await options.fetchImpl(url, {
         method: 'GET',
@@ -1066,12 +1103,14 @@ export const createHostedCloudAccessTokenProvider = (options: {
       }
       return accessToken
     } catch (error) {
+      if (cancelled) throw new Error(HOSTED_ACCESS_TOKEN_CANCELLED)
       if (error instanceof Error && error.name === 'AbortError') {
         throw new Error(`hosted Cloud access-token provider timed out after ${String(timeoutMs)}ms`)
       }
       throw error
     } finally {
       clearTimeout(timer)
+      callerSignal?.removeEventListener('abort', abortForCaller)
     }
   }
 }
diff --git a/src/observability/cloud-reporter.test.ts b/src/observability/cloud-reporter.test.ts
index 430739d1..bc8bb7f2 100644
--- a/src/observability/cloud-reporter.test.ts
+++ b/src/observability/cloud-reporter.test.ts
@@ -234,6 +234,55 @@ describe('FactoryCloudReporter', () => {
     expect(JSON.stringify(warnings)).not.toContain('private server rejection detail')
   })
 
+  it('cancels an in-flight access-token request when close() gives up at its deadline', async () => {
+    const signals: Array<AbortSignal | undefined> = []
+    const reporter = await createReporter({
+      autoFlush: true,
+      getAccessToken: async (options) => {
+        signals.push(options?.signal)
+        // The hosted credential endpoint never answers.
+        await new Promise<void>((resolve) => {
+          options?.signal?.addEventListener('abort', () => { resolve() }, { once: true })
+        })
+        throw new Error('token request cancelled')
+      },
+      fetch: vi.fn(async () => response(201, { accepted: 1, duplicates: 0 })),
+    })
+
+    await reporter.report(progress('event-shutdown'))
+    await vi.waitFor(() => { expect(signals).toHaveLength(1) })
+
+    await expect(reporter.close({ deadlineMs: 20 })).resolves.toMatchObject({
+      stoppedReason: 'deadline',
+    })
+
+    // close() must not merely stop awaiting the automatic flush: the token
+    // request it left behind has to be cancelled, or it keeps the process alive.
+    expect(signals[0]?.aborted).toBe(true)
+  })
+
+  it('cancels an in-flight event request when close() gives up at its deadline', async () => {
+    const signals: Array<AbortSignal | undefined> = []
+    const reporter = await createReporter({
+      autoFlush: true,
+      fetch: vi.fn(async (_url, init) => {
+        signals.push(init?.signal ?? undefined)
+        return await new Promise<Response>((_resolve, reject) => {
+          init?.signal?.addEventListener('abort', () => {
+            reject(Object.assign(new Error('aborted'), { name: 'AbortError' }))
+          }, { once: true })
+        })
+      }),
+    })
+
+    await reporter.report(progress('event-shutdown-post'))
+    await vi.waitFor(() => { expect(signals).toHaveLength(1) })
+
+    await reporter.close({ deadlineMs: 20 })
+
+    expect(signals[0]?.aborted).toBe(true)
+  })
+
   it('preserves a Cloud deployment base path when resolving the endpoint', async () => {
     const requests: string[] = []
     const reporter = await createReporter({
diff --git a/src/observability/cloud-reporter.ts b/src/observability/cloud-reporter.ts
index 9af52c96..0e83e729 100644
--- a/src/observability/cloud-reporter.ts
+++ b/src/observability/cloud-reporter.ts
@@ -27,7 +27,7 @@ const ingestResponseSchema = z.object({
 }).passthrough()
 
 export type FactoryCloudAccessTokenProvider = (
-  options?: { forceRefresh?: boolean },
+  options?: { forceRefresh?: boolean, signal?: AbortSignal },
 ) => string | Promise<string>
 
 export interface FactoryCloudReporterOptions {
@@ -84,6 +84,7 @@ export class FactoryCloudReporter implements FactoryEventReporter {
   readonly #now: () => number
   readonly #random: () => number
   readonly #sleep: (ms: number) => Promise<void>
+  readonly #shutdown = new AbortController()
   #flushInFlight?: Promise<FactoryEventReportResult>
   #retryTimer?: ReturnType<typeof setTimeout>
   #closed = false
@@ -110,7 +111,12 @@ export class FactoryCloudReporter implements FactoryEventReporter {
     this.#autoFlush = options.autoFlush ?? true
     this.#now = options.now ?? Date.now
     this.#random = options.random ?? Math.random
-    this.#sleep = options.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms)))
+    // An unreferenced retry timer cannot hold the process open once the CLI
+    // has stopped awaiting the reporter.
+    this.#sleep = options.sleep ?? ((ms) => new Promise((resolve) => {
+      const timer = setTimeout(resolve, ms)
+      timer.unref?.()
+    }))
   }
 
   async report(rawEvent: FactoryCloudEventInputV1): Promise<void> {
@@ -170,7 +176,14 @@ export class FactoryCloudReporter implements FactoryEventReporter {
     this.#closed = true
     if (this.#retryTimer) clearTimeout(this.#retryTimer)
     this.#retryTimer = undefined
-    return await this.flush(options)
+    try {
+      return await this.flush(options)
+    } finally {
+      // A deadline only stops us *awaiting* the flush; the requests it started
+      // keep running. Cancel them here so nothing — a hung credential endpoint
+      // in particular — survives shutdown and holds the event loop open.
+      this.#shutdown.abort(new FactoryCloudDeadlineError())
+    }
   }
 
   async #flush(deadlineAt: number): Promise<FactoryEventReportResult> {
@@ -232,10 +245,7 @@ export class FactoryCloudReporter implements FactoryEventReporter {
       if (this.#now() >= deadlineAt) return { delivered: false, attempts: attempt - 1, stoppedReason: 'deadline' }
       let response: Response
       try {
-        const token = await this.#withinDeadline(
-          Promise.resolve().then(async () => await this.#getAccessToken({ forceRefresh: attempt > 1 })),
-          deadlineAt,
-        )
+        const token = await this.#requestAccessToken(attempt, deadlineAt)
         if (!token.trim()) throw new Error('Cloud access token is empty')
         response = await this.#fetchWithTimeout(batch, token, deadlineAt)
       } catch (error) {
@@ -301,12 +311,44 @@ export class FactoryCloudReporter implements FactoryEventReporter {
     return { delivered: false, attempts: this.#maxAttempts, stoppedReason: 'retry-exhausted', retryDelayMs }
   }
 
+  /**
+   * Resolve a credential without letting the provider outlive this flush.
+   * `#withinDeadline` alone only stops awaiting it, so the controller is
+   * aborted both when the deadline lapses and when `close()` gives up.
+   */
+  async #requestAccessToken(attempt: number, deadlineAt: number): Promise<string> {
+    if (this.#shutdown.signal.aborted) throw new FactoryCloudDeadlineError()
+    const controller = new AbortController()
+    const abortForShutdown = (): void => { controller.abort(new FactoryCloudDeadlineError()) }
+    this.#shutdown.signal.addEventListener('abort', abortForShutdown, { once: true })
+    try {
+      return await this.#withinDeadline(
+        Promise.resolve().then(async () => await this.#getAccessToken({
+          forceRefresh: attempt > 1,
+          signal: controller.signal,
+        })),
+        deadlineAt,
+      )
+    } catch (error) {
+      controller.abort(error)
+      // A provider that rejected because shutdown cancelled it is a deadline
+      // stop, not a transient failure worth another attempt.
+      if (this.#shutdown.signal.aborted && !isDeadlineExceeded(error)) throw new FactoryCloudDeadlineError()
+      throw error
+    } finally {
+      this.#shutdown.signal.removeEventListener('abort', abortForShutdown)
+    }
+  }
+
   async #fetchWithTimeout(batch: FactoryCloudEventBatchV1, token: string, deadlineAt: number): Promise<Response> {
+    if (this.#shutdown.signal.aborted) throw new FactoryCloudDeadlineError()
     const controller = new AbortController()
     const remainingMs = Math.max(1, deadlineAt - this.#now())
     const timeoutMs = Math.min(this.#requestTimeoutMs, remainingMs)
     const timer = setTimeout(() => controller.abort(), timeoutMs)
     timer.unref?.()
+    const abortForShutdown = (): void => { controller.abort(new FactoryCloudDeadlineError()) }
+    this.#shutdown.signal.addEventListener('abort', abortForShutdown, { once: true })
     try {
       return await this.#fetch(this.#endpoint, {
         method: 'POST',
@@ -317,8 +359,12 @@ export class FactoryCloudReporter implements FactoryEventReporter {
         body: JSON.stringify(batch),
         signal: controller.signal,
       })
+    } catch (error) {
+      if (this.#shutdown.signal.aborted && !isDeadlineExceeded(error)) throw new FactoryCloudDeadlineError()
+      throw error
     } finally {
       clearTimeout(timer)
+      this.#shutdown.signal.removeEventListener('abort', abortForShutdown)
     }
   }
 

From 65afec6769c2576b7c5bc13045cf01e52f957d3f Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Thu, 20 Aug 2026 01:33:41 +0200
Subject: [PATCH 3/5] fix(observability): bind the response body read to
 reporter shutdown
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Headers arriving is not delivery. #fetchWithTimeout removed the shutdown
listener and cleared the request timeout as soon as the response resolved,
then #deliver read the body outside that scope. A Cloud reply that stalls
mid-body therefore survived close(): #withinDeadline stopped awaiting it,
but nothing cancelled the read, so the connection held the event loop open
past the shutdown deadline.

The request's cancellation now spans the whole round trip. #beginRequest
opens the scope, #sendBatch performs the fetch, and the caller releases it
only after the body is read or abandoned — release() also aborts, which
discards a body no branch consumed. Release happens before any retry sleep,
so a waiting attempt never holds a connection open.

Shutdown aborts now carry FactoryCloudShutdownError (a FactoryCloudDeadlineError,
so control flow is unchanged) and the hosted token provider forwards the
caller's abort reason. That makes the *cause* of a cancellation observable:
the same controller is aborted both by a lapsed flush deadline and by
shutdown, so `signal.aborted` alone could not distinguish them — the
fleet CLI guard now asserts the reason instead, and cannot pass if the
shutdown binding regresses.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Session-Id: 4124d0ae-4989-420b-972a-2bbdf4fd1484
---
 src/cli/fleet.test.ts                     |  10 +-
 src/mount/relayfile-cloud-mount-client.ts |   5 +-
 src/observability/cloud-reporter.test.ts  |  55 ++++++++
 src/observability/cloud-reporter.ts       | 163 +++++++++++++++-------
 4 files changed, 178 insertions(+), 55 deletions(-)

diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index 3fd52f4f..2a276d95 100644
--- a/src/cli/fleet.test.ts
+++ b/src/cli/fleet.test.ts
@@ -1679,8 +1679,14 @@ describe('fleet CLI runtime', () => {
       expect(code).toBe(0)
       expect(tokenSignals.length).toBeGreaterThan(0)
       // Shutdown must leave no referenced credential request behind, or the
-      // process outlives the reporter's close() deadline.
-      expect(tokenSignals.map((signal) => signal?.aborted)).toEqual(tokenSignals.map(() => true))
+      // process outlives the reporter's close() deadline. `aborted` alone does
+      // not prove that: the reporter also aborts this same controller when the
+      // flush deadline lapses, so an unbound request still reports aborted and
+      // the assertion could not fail for the reason it guards. Assert the abort
+      // *reason* instead — only the shutdown path produces it.
+      expect(tokenSignals.map((signal) => (signal?.reason as Error | undefined)?.name)).toEqual(
+        tokenSignals.map(() => 'FactoryCloudShutdownError'),
+      )
     } finally {
       // The cancelled flush unwinds its outbox writes just after the CLI
       // returns; retry rmdir rather than race it.
diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts
index d8314567..c9741a5e 100644
--- a/src/mount/relayfile-cloud-mount-client.ts
+++ b/src/mount/relayfile-cloud-mount-client.ts
@@ -1070,7 +1070,10 @@ export const createHostedCloudAccessTokenProvider = (options: {
     let cancelled = false
     const abortForCaller = (): void => {
       cancelled = true
-      controller.abort()
+      // Forward the caller's reason so the request signal records *why* it was
+      // cancelled: a reporter shutting down and a lapsed flush deadline are
+      // different stops, and only the caller can tell them apart.
+      controller.abort(callerSignal?.reason)
     }
     callerSignal?.addEventListener('abort', abortForCaller, { once: true })
     const timer = setTimeout(() => controller.abort(), timeoutMs)
diff --git a/src/observability/cloud-reporter.test.ts b/src/observability/cloud-reporter.test.ts
index bc8bb7f2..8bff8128 100644
--- a/src/observability/cloud-reporter.test.ts
+++ b/src/observability/cloud-reporter.test.ts
@@ -283,6 +283,61 @@ describe('FactoryCloudReporter', () => {
     expect(signals[0]?.aborted).toBe(true)
   })
 
+  it('records shutdown, not the lapsed deadline, as the reason a token request was cancelled', async () => {
+    const signals: AbortSignal[] = []
+    const reporter = await createReporter({
+      getAccessToken: async (options) => {
+        if (options?.signal) signals.push(options.signal)
+        // The hosted credential endpoint never answers.
+        await new Promise<never>(() => {})
+        return 'cloud-token'
+      },
+      fetch: vi.fn(),
+    })
+
+    await reporter.report(progress('event-shutdown-reason'))
+    await reporter.close({ deadlineMs: 20 })
+
+    // `aborted` alone cannot guard the shutdown binding: this controller is
+    // also aborted when the flush deadline lapses, so a request unbound from
+    // shutdown still ends up aborted. Only the reason tells the two apart.
+    await vi.waitFor(() => { expect(signals[0]?.aborted).toBe(true) })
+    expect((signals[0]?.reason as Error | undefined)?.name).toBe('FactoryCloudShutdownError')
+  })
+
+  it('cancels a response body that stalls after headers when close() gives up at its deadline', async () => {
+    const requestSignals: Array<AbortSignal | undefined> = []
+    const bodyStops: string[] = []
+    const reporter = await createReporter({
+      autoFlush: true,
+      fetch: vi.fn(async (_url, init) => {
+        requestSignals.push(init?.signal ?? undefined)
+        // Cloud answers with headers and then never sends the body. A real
+        // fetch tears the body down when the request signal aborts.
+        const body = new ReadableStream<Uint8Array>({
+          start(controller) {
+            init?.signal?.addEventListener('abort', () => {
+              bodyStops.push('aborted')
+              controller.error(Object.assign(new Error('aborted'), { name: 'AbortError' }))
+            }, { once: true })
+          },
+          cancel() { bodyStops.push('cancelled') },
+        })
+        return new Response(body, { status: 201, headers: { 'Content-Type': 'application/json' } })
+      }),
+    })
+
+    await reporter.report(progress('event-stalled-body'))
+    await vi.waitFor(() => { expect(requestSignals).toHaveLength(1) })
+
+    await reporter.close({ deadlineMs: 20 })
+
+    // Headers arriving is not delivery: the body read runs on the same socket,
+    // so shutdown has to cancel it too or the process outlives close().
+    expect(bodyStops.length).toBeGreaterThan(0)
+    expect(requestSignals[0]?.aborted).toBe(true)
+  })
+
   it('preserves a Cloud deployment base path when resolving the endpoint', async () => {
     const requests: string[] = []
     const reporter = await createReporter({
diff --git a/src/observability/cloud-reporter.ts b/src/observability/cloud-reporter.ts
index 0e83e729..356f52e8 100644
--- a/src/observability/cloud-reporter.ts
+++ b/src/observability/cloud-reporter.ts
@@ -53,6 +53,16 @@ export interface FactoryCloudReporterOptions {
   sleep?: (ms: number) => Promise<void>
 }
 
+/**
+ * Cancellation that outlives the round trip. `release()` detaches the
+ * shutdown/timeout wiring and aborts, which also tears down a body that was
+ * never read.
+ */
+type CloudRequestLifetime = {
+  signal: AbortSignal
+  release: () => void
+}
+
 type DeliveryResult = {
   delivered: boolean
   attempts: number
@@ -182,7 +192,7 @@ export class FactoryCloudReporter implements FactoryEventReporter {
       // A deadline only stops us *awaiting* the flush; the requests it started
       // keep running. Cancel them here so nothing — a hung credential endpoint
       // in particular — survives shutdown and holds the event loop open.
-      this.#shutdown.abort(new FactoryCloudDeadlineError())
+      this.#shutdown.abort(new FactoryCloudShutdownError())
     }
   }
 
@@ -244,11 +254,14 @@ export class FactoryCloudReporter implements FactoryEventReporter {
     for (let attempt = 1; attempt <= this.#maxAttempts; attempt += 1) {
       if (this.#now() >= deadlineAt) return { delivered: false, attempts: attempt - 1, stoppedReason: 'deadline' }
       let response: Response
+      let request: CloudRequestLifetime | undefined
       try {
         const token = await this.#requestAccessToken(attempt, deadlineAt)
         if (!token.trim()) throw new Error('Cloud access token is empty')
-        response = await this.#fetchWithTimeout(batch, token, deadlineAt)
+        request = this.#beginRequest(deadlineAt)
+        response = await this.#sendBatch(batch, token, request)
       } catch (error) {
+        request?.release()
         if (isDeadlineExceeded(error)) {
           return { delivered: false, attempts: attempt, stoppedReason: 'deadline' }
         }
@@ -266,43 +279,51 @@ export class FactoryCloudReporter implements FactoryEventReporter {
         continue
       }
 
-      if (response.status === 201) {
-        const payload = ingestResponseSchema.safeParse(
-          await this.#withinDeadline(readJson(response), deadlineAt),
-        )
-        if (payload.success && payload.data.accepted + payload.data.duplicates === batch.events.length) {
-          return { delivered: true, attempts: attempt }
+      // Response headers are not the response. The body arrives on the same
+      // connection, so the request's cancellation stays armed until the body is
+      // read or discarded; released early, a Cloud reply that stalls mid-body
+      // survives close() and holds the event loop open. Released here rather
+      // than in a whole-attempt finally so a retry never sleeps with the
+      // connection still open.
+      try {
+        if (response.status === 201) {
+          const payload = ingestResponseSchema.safeParse(await this.#readBody(response, deadlineAt))
+          if (payload.success && payload.data.accepted + payload.data.duplicates === batch.events.length) {
+            return { delivered: true, attempts: attempt }
+          }
+          this.#logger?.warn?.('[factory] cloud progress response was incomplete', { status: response.status })
+          return { delivered: false, attempts: attempt, stoppedReason: 'rejected', retryDelayMs: this.#retryMaxMs }
         }
-        this.#logger?.warn?.('[factory] cloud progress response was incomplete', { status: response.status })
-        return { delivered: false, attempts: attempt, stoppedReason: 'rejected', retryDelayMs: this.#retryMaxMs }
-      }
 
-      // A Cloud CLI token can expire between provider resolution and the
-      // request. Retry once through the provider's forced-refresh path without
-      // delaying the durable queue.
-      if (response.status === 401 && attempt < this.#maxAttempts) {
-        continue
-      }
+        // A Cloud CLI token can expire between provider resolution and the
+        // request. Retry once through the provider's forced-refresh path without
+        // delaying the durable queue.
+        if (response.status === 401 && attempt < this.#maxAttempts) {
+          continue
+        }
 
-      if (response.status === 401) {
-        return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
-      }
+        if (response.status === 401) {
+          return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
+        }
 
-      if (!isRetryableStatus(response.status)) {
-        this.#logger?.warn?.('[factory] cloud progress batch rejected', { status: response.status })
-        return {
-          delivered: false,
-          attempts: attempt,
-          stoppedReason: 'rejected',
-          retryDelayMs: this.#retryMaxMs,
-          discard: isPermanentPayloadRejectionStatus(response.status),
-          status: response.status,
+        if (!isRetryableStatus(response.status)) {
+          this.#logger?.warn?.('[factory] cloud progress batch rejected', { status: response.status })
+          return {
+            delivered: false,
+            attempts: attempt,
+            stoppedReason: 'rejected',
+            retryDelayMs: this.#retryMaxMs,
+            discard: isPermanentPayloadRejectionStatus(response.status),
+            status: response.status,
+          }
         }
-      }
 
-      retryDelayMs = retryAfterMs(response, this.#now()) ?? this.#nextRetryDelay(attempt)
-      if (attempt >= this.#maxAttempts) {
-        return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
+        retryDelayMs = retryAfterMs(response, this.#now()) ?? this.#nextRetryDelay(attempt)
+        if (attempt >= this.#maxAttempts) {
+          return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
+        }
+      } finally {
+        request?.release()
       }
       if (!await this.#waitForRetry(retryDelayMs, deadlineAt)) {
         return { delivered: false, attempts: attempt, stoppedReason: 'deadline' }
@@ -317,9 +338,9 @@ export class FactoryCloudReporter implements FactoryEventReporter {
    * aborted both when the deadline lapses and when `close()` gives up.
    */
   async #requestAccessToken(attempt: number, deadlineAt: number): Promise<string> {
-    if (this.#shutdown.signal.aborted) throw new FactoryCloudDeadlineError()
+    if (this.#shutdown.signal.aborted) throw new FactoryCloudShutdownError()
     const controller = new AbortController()
-    const abortForShutdown = (): void => { controller.abort(new FactoryCloudDeadlineError()) }
+    const abortForShutdown = (): void => { controller.abort(new FactoryCloudShutdownError()) }
     this.#shutdown.signal.addEventListener('abort', abortForShutdown, { once: true })
     try {
       return await this.#withinDeadline(
@@ -333,22 +354,44 @@ export class FactoryCloudReporter implements FactoryEventReporter {
       controller.abort(error)
       // A provider that rejected because shutdown cancelled it is a deadline
       // stop, not a transient failure worth another attempt.
-      if (this.#shutdown.signal.aborted && !isDeadlineExceeded(error)) throw new FactoryCloudDeadlineError()
+      if (this.#shutdown.signal.aborted && !isDeadlineExceeded(error)) throw new FactoryCloudShutdownError()
       throw error
     } finally {
       this.#shutdown.signal.removeEventListener('abort', abortForShutdown)
     }
   }
 
-  async #fetchWithTimeout(batch: FactoryCloudEventBatchV1, token: string, deadlineAt: number): Promise<Response> {
-    if (this.#shutdown.signal.aborted) throw new FactoryCloudDeadlineError()
+  /**
+   * Open a cancellation scope for one round trip: request timeout, remaining
+   * deadline, and reporter shutdown all abort it. The caller releases it once
+   * the response body is consumed or abandoned.
+   */
+  #beginRequest(deadlineAt: number): CloudRequestLifetime {
+    if (this.#shutdown.signal.aborted) throw new FactoryCloudShutdownError()
     const controller = new AbortController()
     const remainingMs = Math.max(1, deadlineAt - this.#now())
     const timeoutMs = Math.min(this.#requestTimeoutMs, remainingMs)
     const timer = setTimeout(() => controller.abort(), timeoutMs)
     timer.unref?.()
-    const abortForShutdown = (): void => { controller.abort(new FactoryCloudDeadlineError()) }
+    const abortForShutdown = (): void => { controller.abort(new FactoryCloudShutdownError()) }
     this.#shutdown.signal.addEventListener('abort', abortForShutdown, { once: true })
+    return {
+      signal: controller.signal,
+      release: () => {
+        clearTimeout(timer)
+        this.#shutdown.signal.removeEventListener('abort', abortForShutdown)
+        // Aborting on release discards any body left unread: an unconsumed body
+        // holds its connection open exactly like a hung request would.
+        controller.abort()
+      },
+    }
+  }
+
+  async #sendBatch(
+    batch: FactoryCloudEventBatchV1,
+    token: string,
+    request: CloudRequestLifetime,
+  ): Promise<Response> {
     try {
       return await this.#fetch(this.#endpoint, {
         method: 'POST',
@@ -357,14 +400,25 @@ export class FactoryCloudReporter implements FactoryEventReporter {
           'Content-Type': 'application/json',
         },
         body: JSON.stringify(batch),
-        signal: controller.signal,
+        signal: request.signal,
       })
     } catch (error) {
-      if (this.#shutdown.signal.aborted && !isDeadlineExceeded(error)) throw new FactoryCloudDeadlineError()
+      if (this.#shutdown.signal.aborted && !isDeadlineExceeded(error)) throw new FactoryCloudShutdownError()
       throw error
-    } finally {
-      clearTimeout(timer)
-      this.#shutdown.signal.removeEventListener('abort', abortForShutdown)
+    }
+  }
+
+  /**
+   * Read the acknowledgement body. A body that is malformed is a rejection; a
+   * body cut short by shutdown or the deadline is a stop, not a rejection.
+   */
+  async #readBody(response: Response, deadlineAt: number): Promise<unknown> {
+    try {
+      return await this.#withinDeadline(response.json(), deadlineAt)
+    } catch (error) {
+      if (isDeadlineExceeded(error)) throw error
+      if (this.#shutdown.signal.aborted) throw new FactoryCloudShutdownError()
+      return undefined
     }
   }
 
@@ -442,6 +496,19 @@ class FactoryCloudDeadlineError extends Error {
   }
 }
 
+/**
+ * A stop caused by `close()` rather than by a lapsed flush deadline. It is a
+ * deadline stop for control-flow purposes, and a distinguishable abort reason
+ * for anyone — a token provider, a test — asking *why* a request was cancelled.
+ */
+class FactoryCloudShutdownError extends FactoryCloudDeadlineError {
+  constructor() {
+    super()
+    this.name = 'FactoryCloudShutdownError'
+    this.message = 'Factory Cloud reporting stopped at shutdown'
+  }
+}
+
 const isDeadlineExceeded = (error: unknown): error is FactoryCloudDeadlineError =>
   error instanceof FactoryCloudDeadlineError
 
@@ -468,14 +535,6 @@ function retryAfterMs(response: Response, nowMs: number): number | undefined {
   return Number.isFinite(dateMs) ? Math.max(0, dateMs - nowMs) : undefined
 }
 
-async function readJson(response: Response): Promise<unknown> {
-  try {
-    return await response.json()
-  } catch {
-    return undefined
-  }
-}
-
 function positiveInteger(value: number, name: string): number {
   if (!Number.isSafeInteger(value) || value <= 0) throw new Error(`${name} must be a positive safe integer`)
   return value

From 53d0d1996b8f403f0d5660a1daafed9f374fce11 Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Thu, 20 Aug 2026 03:14:31 +0200
Subject: [PATCH 4/5] fix(observability): retry a timed-out acknowledgement
 body
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Binding the 201 body read to the request lifetime was right, but it left a
timed-out body on a success response taking the malformed-body path: the
abort surfaced as an unparseable body, so the reporter marked the batch
'rejected' and parked it for retryMaxMs. Nothing rejected it — the
acknowledgement simply never finished arriving, and ingestion is idempotent.

#readBody now tells a request-timeout abort apart from a malformed body and
raises FactoryCloudRequestTimeoutError, which #deliver routes onto the same
transient backoff a failed send takes. The retry tail is now shared by both
transient paths, so the connection is still released before any sleep.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Session-Id: b59d7dc4-689d-49d1-b4fc-4cdba699c9e6
---
 src/observability/cloud-reporter.test.ts |  39 ++++++++
 src/observability/cloud-reporter.ts      | 117 ++++++++++++++---------
 2 files changed, 113 insertions(+), 43 deletions(-)

diff --git a/src/observability/cloud-reporter.test.ts b/src/observability/cloud-reporter.test.ts
index 8bff8128..8baa3313 100644
--- a/src/observability/cloud-reporter.test.ts
+++ b/src/observability/cloud-reporter.test.ts
@@ -167,6 +167,45 @@ describe('FactoryCloudReporter', () => {
     expect(fetch).toHaveBeenCalledTimes(2)
   })
 
+  it('retries a 201 whose acknowledgement body is cut short by the request timeout', async () => {
+    let calls = 0
+    const waits: number[] = []
+    const fetch = vi.fn(async (_url, init) => {
+      calls += 1
+      if (calls > 1) return response(201, { accepted: 1, duplicates: 0 })
+      // Cloud answers 201 and then stalls mid-body. A real fetch tears the body
+      // down when the request timeout aborts the signal.
+      const body = new ReadableStream<Uint8Array>({
+        start(controller) {
+          init?.signal?.addEventListener('abort', () => {
+            controller.error(Object.assign(new Error('aborted'), { name: 'AbortError' }))
+          }, { once: true })
+        },
+      })
+      return new Response(body, { status: 201, headers: { 'Content-Type': 'application/json' } })
+    })
+    const reporter = await createReporter({
+      fetch,
+      requestTimeoutMs: 10,
+      maxAttempts: 2,
+      retryBaseMs: 5,
+      sleep: async (ms) => { waits.push(ms) },
+    })
+    await reporter.report(progress('event-body-timeout'))
+
+    // A timed-out acknowledgement is not a rejection: nothing said the batch was
+    // bad, so it belongs on the transient retry path rather than parked for
+    // retryMaxMs. Ingestion is idempotent, so a re-send is safe.
+    expect(await reporter.flush()).toMatchObject({
+      delivered: 1,
+      pending: 0,
+      attempts: 2,
+      stoppedReason: 'empty',
+    })
+    expect(waits).toEqual([5])
+    expect(fetch).toHaveBeenCalledTimes(2)
+  })
+
   it('bounds close even when an earlier unbounded flush is already running', async () => {
     let resolveTokenRequested!: () => void
     const tokenRequested = new Promise<void>((resolve) => {
diff --git a/src/observability/cloud-reporter.ts b/src/observability/cloud-reporter.ts
index 356f52e8..4a6f83cb 100644
--- a/src/observability/cloud-reporter.ts
+++ b/src/observability/cloud-reporter.ts
@@ -253,8 +253,12 @@ export class FactoryCloudReporter implements FactoryEventReporter {
     let retryDelayMs = this.#retryBaseMs
     for (let attempt = 1; attempt <= this.#maxAttempts; attempt += 1) {
       if (this.#now() >= deadlineAt) return { delivered: false, attempts: attempt - 1, stoppedReason: 'deadline' }
-      let response: Response
+      let response: Response | undefined
       let request: CloudRequestLifetime | undefined
+      // The attempt failed in a way that says nothing about the batch: the
+      // request never completed, or its acknowledgement was cut short. Both
+      // back off and try again rather than parking the batch.
+      let transient = false
       try {
         const token = await this.#requestAccessToken(attempt, deadlineAt)
         if (!token.trim()) throw new Error('Cloud access token is empty')
@@ -269,14 +273,7 @@ export class FactoryCloudReporter implements FactoryEventReporter {
           attempt,
           errorClass: errorClass(error),
         })
-        if (attempt >= this.#maxAttempts) {
-          return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
-        }
-        if (!await this.#waitForRetry(retryDelayMs, deadlineAt)) {
-          return { delivered: false, attempts: attempt, stoppedReason: 'deadline' }
-        }
-        retryDelayMs = this.#nextRetryDelay(attempt)
-        continue
+        transient = true
       }
 
       // Response headers are not the response. The body arrives on the same
@@ -285,49 +282,69 @@ export class FactoryCloudReporter implements FactoryEventReporter {
       // survives close() and holds the event loop open. Released here rather
       // than in a whole-attempt finally so a retry never sleeps with the
       // connection still open.
-      try {
-        if (response.status === 201) {
-          const payload = ingestResponseSchema.safeParse(await this.#readBody(response, deadlineAt))
-          if (payload.success && payload.data.accepted + payload.data.duplicates === batch.events.length) {
-            return { delivered: true, attempts: attempt }
+      if (request && response) {
+        try {
+          if (response.status === 201) {
+            const payload = ingestResponseSchema.safeParse(await this.#readBody(response, request, deadlineAt))
+            if (payload.success && payload.data.accepted + payload.data.duplicates === batch.events.length) {
+              return { delivered: true, attempts: attempt }
+            }
+            this.#logger?.warn?.('[factory] cloud progress response was incomplete', { status: response.status })
+            return { delivered: false, attempts: attempt, stoppedReason: 'rejected', retryDelayMs: this.#retryMaxMs }
           }
-          this.#logger?.warn?.('[factory] cloud progress response was incomplete', { status: response.status })
-          return { delivered: false, attempts: attempt, stoppedReason: 'rejected', retryDelayMs: this.#retryMaxMs }
-        }
 
-        // A Cloud CLI token can expire between provider resolution and the
-        // request. Retry once through the provider's forced-refresh path without
-        // delaying the durable queue.
-        if (response.status === 401 && attempt < this.#maxAttempts) {
-          continue
-        }
+          // A Cloud CLI token can expire between provider resolution and the
+          // request. Retry once through the provider's forced-refresh path without
+          // delaying the durable queue.
+          if (response.status === 401 && attempt < this.#maxAttempts) {
+            continue
+          }
 
-        if (response.status === 401) {
-          return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
-        }
+          if (response.status === 401) {
+            return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
+          }
 
-        if (!isRetryableStatus(response.status)) {
-          this.#logger?.warn?.('[factory] cloud progress batch rejected', { status: response.status })
-          return {
-            delivered: false,
-            attempts: attempt,
-            stoppedReason: 'rejected',
-            retryDelayMs: this.#retryMaxMs,
-            discard: isPermanentPayloadRejectionStatus(response.status),
-            status: response.status,
+          if (!isRetryableStatus(response.status)) {
+            this.#logger?.warn?.('[factory] cloud progress batch rejected', { status: response.status })
+            return {
+              delivered: false,
+              attempts: attempt,
+              stoppedReason: 'rejected',
+              retryDelayMs: this.#retryMaxMs,
+              discard: isPermanentPayloadRejectionStatus(response.status),
+              status: response.status,
+            }
           }
-        }
 
-        retryDelayMs = retryAfterMs(response, this.#now()) ?? this.#nextRetryDelay(attempt)
-        if (attempt >= this.#maxAttempts) {
-          return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
+          retryDelayMs = retryAfterMs(response, this.#now()) ?? this.#nextRetryDelay(attempt)
+          if (attempt >= this.#maxAttempts) {
+            return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
+          }
+        } catch (error) {
+          // The acknowledgement never finished arriving. Cloud did not reject
+          // anything, so treating it like a malformed body would park a healthy
+          // batch for retryMaxMs; ingestion is idempotent, so re-send instead.
+          if (!(error instanceof FactoryCloudRequestTimeoutError)) throw error
+          this.#logger?.warn?.('[factory] cloud progress acknowledgement timed out', { attempt })
+          transient = true
+        } finally {
+          request.release()
         }
-      } finally {
-        request?.release()
+        if (!transient) {
+          if (!await this.#waitForRetry(retryDelayMs, deadlineAt)) {
+            return { delivered: false, attempts: attempt, stoppedReason: 'deadline' }
+          }
+          continue
+        }
+      }
+
+      if (attempt >= this.#maxAttempts) {
+        return { delivered: false, attempts: attempt, stoppedReason: 'retry-exhausted', retryDelayMs }
       }
       if (!await this.#waitForRetry(retryDelayMs, deadlineAt)) {
         return { delivered: false, attempts: attempt, stoppedReason: 'deadline' }
       }
+      retryDelayMs = this.#nextRetryDelay(attempt)
     }
     return { delivered: false, attempts: this.#maxAttempts, stoppedReason: 'retry-exhausted', retryDelayMs }
   }
@@ -410,14 +427,16 @@ export class FactoryCloudReporter implements FactoryEventReporter {
 
   /**
    * Read the acknowledgement body. A body that is malformed is a rejection; a
-   * body cut short by shutdown or the deadline is a stop, not a rejection.
+   * body cut short by shutdown, the deadline, or the request timeout is a stop
+   * or a transient failure, not a rejection.
    */
-  async #readBody(response: Response, deadlineAt: number): Promise<unknown> {
+  async #readBody(response: Response, request: CloudRequestLifetime, deadlineAt: number): Promise<unknown> {
     try {
       return await this.#withinDeadline(response.json(), deadlineAt)
     } catch (error) {
       if (isDeadlineExceeded(error)) throw error
       if (this.#shutdown.signal.aborted) throw new FactoryCloudShutdownError()
+      if (request.signal.aborted) throw new FactoryCloudRequestTimeoutError()
       return undefined
     }
   }
@@ -509,6 +528,18 @@ class FactoryCloudShutdownError extends FactoryCloudDeadlineError {
   }
 }
 
+/**
+ * The per-request timeout fired before the round trip finished. Deliberately
+ * not a deadline error: the flush may still have time, and the batch's fate is
+ * simply unknown, so it retries like any other transient failure.
+ */
+class FactoryCloudRequestTimeoutError extends Error {
+  constructor() {
+    super('Factory Cloud request timed out')
+    this.name = 'FactoryCloudRequestTimeoutError'
+  }
+}
+
 const isDeadlineExceeded = (error: unknown): error is FactoryCloudDeadlineError =>
   error instanceof FactoryCloudDeadlineError
 

From 75481a03dbe60b11eaf821f3e8f12b7902e11c5f Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Thu, 20 Aug 2026 03:14:31 +0200
Subject: [PATCH 5/5] test(observability): stop racing the shutdown abort
 reason
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The shutdown-reason test drove the credential request from close()'s own
bounded flush, so two timers armed at the same deadline decided the abort
reason: #requestAccessToken aborts with FactoryCloudDeadlineError when the
flush deadline lapses first, and close() only aborts #shutdown after flush()
returns. Whichever fired first froze the reason — the assertion failed on
2 of 15 local runs.

Put the credential request under an unbounded flush instead. It then has no
deadline of its own, so shutdown is the only thing that can cancel it and the
recorded reason is unambiguous. 0 failures in 40 runs, and the test still
fails when the shutdown binding is removed from #requestAccessToken.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Session-Id: b59d7dc4-689d-49d1-b4fc-4cdba699c9e6
---
 src/observability/cloud-reporter.test.ts | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

diff --git a/src/observability/cloud-reporter.test.ts b/src/observability/cloud-reporter.test.ts
index 8baa3313..e1ff6bb0 100644
--- a/src/observability/cloud-reporter.test.ts
+++ b/src/observability/cloud-reporter.test.ts
@@ -324,23 +324,33 @@ describe('FactoryCloudReporter', () => {
 
   it('records shutdown, not the lapsed deadline, as the reason a token request was cancelled', async () => {
     const signals: AbortSignal[] = []
+    let resolveTokenRequested!: () => void
+    const tokenRequested = new Promise<void>((resolve) => {
+      resolveTokenRequested = resolve
+    })
     const reporter = await createReporter({
       getAccessToken: async (options) => {
         if (options?.signal) signals.push(options.signal)
+        resolveTokenRequested()
         // The hosted credential endpoint never answers.
-        await new Promise<never>(() => {})
-        return 'cloud-token'
+        return await new Promise<string>(() => {})
       },
       fetch: vi.fn(),
     })
 
     await reporter.report(progress('event-shutdown-reason'))
+    // The credential request has to be in flight under an *unbounded* flush.
+    // Give it a deadline of its own and the two cancellations race: whichever
+    // fires first freezes the reason, so the assertion below would be flaky.
+    void reporter.flush()
+    await tokenRequested
+
     await reporter.close({ deadlineMs: 20 })
 
     // `aborted` alone cannot guard the shutdown binding: this controller is
     // also aborted when the flush deadline lapses, so a request unbound from
     // shutdown still ends up aborted. Only the reason tells the two apart.
-    await vi.waitFor(() => { expect(signals[0]?.aborted).toBe(true) })
+    expect(signals[0]?.aborted).toBe(true)
     expect((signals[0]?.reason as Error | undefined)?.name).toBe('FactoryCloudShutdownError')
   })