Skip to content

feat(observability): authenticate hosted Factory reporter #289

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
khaliqgant wants to merge 1 commit into
base: main
Choose a base branch
from
+208 −15
Open

feat(observability): authenticate hosted Factory reporter #289

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
158 changes: 157 additions & 1 deletion src/cli/fleet.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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 {
Expand Down
48 changes: 40 additions & 8 deletions src/cli/fleet.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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>
Expand Down Expand Up @@ -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,
Expand All @@ -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({

@cubic-dev-ai cubic-dev-ai Bot Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: Propagate reporter cancellation into createHostedCloudAccessTokenProvider; otherwise a hung hosted-token fetch continues past close({ deadlineMs: 2_000 }) and can keep the CLI process alive until its timeout and retries finish.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. At src/cli/fleet.ts, line 1625:

<comment>Propagate reporter cancellation into `createHostedCloudAccessTokenProvider`; otherwise a hung hosted-token fetch continues past `close({ deadlineMs: 2_000 })` and can keep the CLI process alive until its timeout and retries finish.</comment>

<file context>
@@ -1594,15 +1613,28 @@ async function buildFactoryCloudReporter(input: {
+    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,
</file context>

url: runtimeEnv[FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV]?.trim() ?? '',

@cubic-dev-ai cubic-dev-ai Bot Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: Hosted mode is selected by the presence of FACTORY_CLOUD_ACCESS_TOKEN_URL, not by a non-empty value. If the variable is set to an empty or whitespace-only string, the code enters the hosted branch and createHostedCloudAccessTokenProvider throws synchronously on new URL(''), so hosted telemetry is bypassed/disabled (or the command errors) instead of falling back to the local-session path. Gate the hosted branch on a trimmed non-empty value and pass that value to the provider.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. At src/cli/fleet.ts, line 1626:

<comment>Hosted mode is selected by the presence of FACTORY_CLOUD_ACCESS_TOKEN_URL, not by a non-empty value. If the variable is set to an empty or whitespace-only string, the code enters the hosted branch and createHostedCloudAccessTokenProvider throws synchronously on `new URL('')`, so hosted telemetry is bypassed/disabled (or the command errors) instead of falling back to the local-session path. Gate the hosted branch on a trimmed non-empty value and pass that value to the provider.</comment>

<file context>
@@ -1594,15 +1613,28 @@ async function buildFactoryCloudReporter(input: {
+    if (hasHostedAccessTokenConfig) {
+      apiUrl = resolveHostedCloudApiUrl(runtimeEnv)
+      getAccessToken = createHostedCloudAccessTokenProvider({
+        url: runtimeEnv[FACTORY_CLOUD_ACCESS_TOKEN_URL_ENV]?.trim() ?? '',
+        fetchImpl: input.deps.cloudAccessTokenFetch ?? fetch,
+      })
</file context>

fetchImpl: input.deps.cloudAccessTokenFetch ?? fetch,
})
Comment on lines +1625 to +1628

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Bound hosted-token fetches to reporter shutdown

When the private token endpoint hangs while an automatic flush is already running, close({ deadlineMs: 2_000 }) only stops awaiting the reporter operation; it does not cancel this hosted-token provider. The provider retains a referenced 10-second abort timer, and the in-flight reporter can perform three attempts, so main() may set the exit code while Node remains alive for roughly 30 seconds. Thread reporter cancellation into the token request, or otherwise ensure the request cannot keep the process alive after the shutdown deadline.

Useful? React with 👍 / 👎.

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(),
Expand All @@ -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,
Expand Down
17 changes: 11 additions & 6 deletions src/mount/relayfile-cloud-mount-client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = (
Expand Down Expand Up @@ -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')
}
Expand Down Expand Up @@ -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)
Expand All @@ -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',
Expand All @@ -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 {
Expand Down