From 30e65e847d1c83cfbba3883b20077985e8b51dd8 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 23 Jul 2026 14:26:52 +0200 Subject: [PATCH 1/2] feat(hosted): report durable lifecycle telemetry --- .../manifest-contract.test.ts | 1 + .agentworkforce/features/manifest.yaml | 22 +- .agentworkforce/features/verify/procedures.md | 11 +- README.md | 10 +- package.json | 4 + scripts/packed-e2e-scenario.mjs | 33 ++ src/hosted/index.ts | 1 + src/hosted/orchestrator.test.ts | 296 +++++++++++++++++- src/hosted/orchestrator.ts | 221 ++++++++++++- src/hosted/types.ts | 4 + src/hosted/worker-safety.test.ts | 29 +- src/telemetry.ts | 11 + 12 files changed, 624 insertions(+), 19 deletions(-) create mode 100644 src/telemetry.ts diff --git a/.agentworkforce/agents/factory-feature-guardian/manifest-contract.test.ts b/.agentworkforce/agents/factory-feature-guardian/manifest-contract.test.ts index d9e7248..be29b33 100644 --- a/.agentworkforce/agents/factory-feature-guardian/manifest-contract.test.ts +++ b/.agentworkforce/agents/factory-feature-guardian/manifest-contract.test.ts @@ -185,6 +185,7 @@ describe('Factory feature manifest contract', () => { const expected: Record = { '.': 'createFactory()', './observability': '@agent-relay/factory/observability', + './telemetry': '@agent-relay/factory/telemetry', './testing': '@agent-relay/factory/testing', './writeback': 'LinearWriteback / GithubWriteback / SlackWriteback', './node': 'createFactoryNodeDefinition()', diff --git a/.agentworkforce/features/manifest.yaml b/.agentworkforce/features/manifest.yaml index 4d99817..a607df3 100644 --- a/.agentworkforce/features/manifest.yaml +++ b/.agentworkforce/features/manifest.yaml @@ -1,11 +1,11 @@ version: '1.1' -updated: '2026-07-21' +updated: '2026-07-23' catalog: category_count: 25 - feature_count: 302 + feature_count: 304 tier_counts: - 1: 44 - 2: 123 + 1: 45 + 2: 124 3: 11 4: 51 5: 58 @@ -1454,6 +1454,13 @@ categories: location: src/observability/, src/index.ts, package.json verify_tier: 1 + - id: api-telemetry-contract + name: Worker-Safe Telemetry Contract API + api: '@agent-relay/factory/telemetry' + description: Export only the Factory-owned event schemas, creator, and reporter types without pulling filesystem-backed outbox or instance identity modules into Worker consumers + location: src/telemetry.ts, src/observability/events.ts, src/ports/observability.ts, package.json + verify_tier: 1 + - id: api-hosted-control-plane name: Hosted Factory API api: '@agent-relay/factory/hosted' @@ -1494,6 +1501,13 @@ categories: location: src/hosted/orchestrator.ts verify_tier: 2 + - id: hosted-durable-run-telemetry + name: Durable Hosted Run Telemetry + api: hostedFactoryRunId() / HostedFactoryPorts.reporter + description: Persist one deterministic run identity per issue and report lifecycle, agent, clarification, writeback, success, and failure events only after their fenced hosted state writes + location: src/hosted/orchestrator.ts, src/hosted/types.ts, src/telemetry.ts + verify_tier: 2 + - id: hosted-merge-gate-writeback name: Hosted Merge Gate and Idempotent Writeback api: HostedFactoryMergeGate / HostedFactoryWriteback diff --git a/.agentworkforce/features/verify/procedures.md b/.agentworkforce/features/verify/procedures.md index 5b8de24..57f5965 100644 --- a/.agentworkforce/features/verify/procedures.md +++ b/.agentworkforce/features/verify/procedures.md @@ -688,7 +688,7 @@ mkdir "$TMP/consumer" && cd "$TMP/consumer" npm init -y >/dev/null npm install --ignore-scripts "$TMP"/*.tgz >/dev/null node --input-type=module <<'NODE' -for (const subpath of ['', '/observability', '/testing', '/writeback', '/featuremap', '/hosted', '/environments']) { +for (const subpath of ['', '/observability', '/telemetry', '/testing', '/writeback', '/featuremap', '/hosted', '/environments']) { const mod = await import(`@agent-relay/factory${subpath}`) if (Object.keys(mod).length === 0) throw new Error(`empty export ${subpath || '/'}`) } @@ -698,7 +698,7 @@ NODE ``` Back in the checkout, run the focused tests named by each API entry. Assert root -types and all eight package export keys resolve, hosted remains worker-safe, +types and all nine JavaScript package entrypoints resolve, hosted remains worker-safe, feature-map validation includes procedure routing, dependency/worktree ports are exported, and fake clients support a hermetic consumer. Clean only `$TMP`. @@ -718,8 +718,11 @@ npm run verify:e2e ``` Assert reconciliation precedes discovery; scope, dedupe, batch capacity, triage, -deterministic invocation IDs, clarification, spawn, pushed completion, polling -completion, merge gate, and every provider writeback transition. Repeat each +deterministic run and invocation IDs, clarification, spawn, pushed completion, +polling completion, merge gate, and every provider writeback transition. Require +the exact persisted lifecycle event order, no second start or spawn event on a +duplicate sweep, correct succeeded/failed terminal status, and unchanged +dispatch/state when a custom reporter rejects. Repeat each external success immediately before a simulated fenced save and require the same invocation or idempotency key after takeover. Two owners must yield one live epoch; expired owners cannot save. Run the same matrix through in-memory diff --git a/README.md b/README.md index bfd5f91..ae2123d 100644 --- a/README.md +++ b/README.md @@ -434,12 +434,20 @@ import { const state = new DurableObjectHostedFactoryStateStore(durableObjectState.storage) const factory = createHostedFactory( { workspaceId, ownerId: isolateId, config }, - { state, discovery, fleet, completions, mergeGate, writeback }, + { state, discovery, fleet, completions, mergeGate, writeback, reporter }, ) await factory.runOnce() // invoke from cron/alarm and safe webhook wakeups ``` +The optional `reporter` implements the canonical `FactoryEventReporter`. +Hosted runs persist a deterministic run ID per workspace and issue, then emit +lifecycle events only after the corresponding fenced state write. Worker hosts +that only need the Factory-owned wire schema, event creator, and reporter types +should import `@agent-relay/factory/telemetry`; unlike the broader +`@agent-relay/factory/observability` surface, it does not export the +filesystem-backed outbox or instance identity helpers. + The Durable Object adapter stores each workspace independently and performs lease claims plus lifecycle writes in storage transactions. Every mutation is checked against the current owner and monotonically increasing lease epoch; an diff --git a/package.json b/package.json index 8f3e32d..f356b6c 100644 --- a/package.json +++ b/package.json @@ -38,6 +38,10 @@ "types": "./dist/observability/index.d.ts", "import": "./dist/observability/index.js" }, + "./telemetry": { + "types": "./dist/telemetry.d.ts", + "import": "./dist/telemetry.js" + }, "./testing": { "types": "./dist/testing/index.d.ts", "import": "./dist/testing/index.js" diff --git a/scripts/packed-e2e-scenario.mjs b/scripts/packed-e2e-scenario.mjs index 292a254..0ec429e 100644 --- a/scripts/packed-e2e-scenario.mjs +++ b/scripts/packed-e2e-scenario.mjs @@ -4,6 +4,7 @@ import { createHostedFactory, InMemoryHostedFactoryStateStore, } from '@agent-relay/factory/hosted' +import { FactoryCloudEventInputV1Schema } from '@agent-relay/factory/telemetry' const checks = ['public-exports-import'] const workspaceId = 'factory-e2e-workspace' @@ -53,6 +54,7 @@ const state = new InMemoryHostedFactoryStateStore({ now: () => nowMs }) const spawned = [] const invocationStatus = new Map() const writebacks = { clarification: [], dispatched: [], completed: [] } +const telemetryEvents = [] const ports = { discovery: { discoverReady: async () => [structuredClone(issue), structuredClone(issue)] }, state, @@ -72,6 +74,15 @@ const ports = { dispatched: async (input) => writebacks.dispatched.push(structuredClone(input)), completed: async (input) => writebacks.completed.push(structuredClone(input)), }, + reporter: { + report: async (event) => telemetryEvents.push(FactoryCloudEventInputV1Schema.parse(event)), + flush: async () => ({ + delivered: telemetryEvents.length, + pending: 0, + attempts: telemetryEvents.length, + stoppedReason: 'empty', + }), + }, now: () => new Date(nowMs), } const factory = createHostedFactory({ @@ -87,14 +98,28 @@ assert.equal(dispatched.discovered, 1, 'duplicate discovery entries must collaps assert.equal(spawned.length, 2, 'single scope must dispatch implementer plus reviewer') assert.equal(new Set(spawned.map(({ invocationId }) => invocationId)).size, 2) assert.equal(writebacks.dispatched.length, 1) +assert.deepEqual(telemetryEvents.map(({ type }) => type), [ + 'run.started', + 'run.phase_changed', + 'agent.planned', + 'agent.planned', + 'agent.spawned', + 'agent.spawned', + 'run.phase_changed', + 'writeback.applied', +]) +assert.equal(new Set(telemetryEvents.map(({ runId }) => runId)).size, 1) checks.push('discover-triage-dispatch-writeback') +checks.push('hosted-telemetry-dispatch-lifecycle') +const eventCountAfterDispatch = telemetryEvents.length const duplicateSweep = await factory.runOnce() assert.equal(spawned.length, 2, 'repeat discovery must not duplicate fleet spawns') assert.ok(duplicateSweep.skipped.some(({ issueKey, reason }) => issueKey === issue.key && reason === 'already running' )) assert.equal(writebacks.dispatched.length, 1, 'dispatch writeback must be idempotent') +assert.equal(telemetryEvents.length, eventCountAfterDispatch, 'repeat discovery must not duplicate lifecycle events') checks.push('at-least-once-deduplication') nowMs += 5_000 @@ -112,6 +137,14 @@ const terminal = await state.getIssue(workspaceId, issue.uuid) assert.equal(terminal?.phase, 'complete') assert.equal(terminal?.mergeGate?.status, 'ready') assert.equal(writebacks.completed.length, 1) +assert.deepEqual(telemetryEvents.slice(eventCountAfterDispatch).map(({ type }) => type), [ + 'agent.exited', + 'agent.exited', + 'run.phase_changed', + 'writeback.applied', + 'run.succeeded', +]) +assert.equal(telemetryEvents.at(-1)?.status, 'succeeded') checks.push('completion-merge-gate-writeback') const replayed = await factory.ingestCompletion({ diff --git a/src/hosted/index.ts b/src/hosted/index.ts index a7a2f8f..5419b47 100644 --- a/src/hosted/index.ts +++ b/src/hosted/index.ts @@ -2,6 +2,7 @@ export { createHostedFactory, HostedFactoryLoop, hostedFactoryInvocationId, + hostedFactoryRunId, } from './orchestrator.js' export { DurableObjectHostedFactoryStateStore, diff --git a/src/hosted/orchestrator.test.ts b/src/hosted/orchestrator.test.ts index 750ab6d..ab8f001 100644 --- a/src/hosted/orchestrator.test.ts +++ b/src/hosted/orchestrator.test.ts @@ -1,15 +1,18 @@ import { describe, expect, it, vi } from 'vitest' import { FactoryConfigSchema, type FactoryConfig } from '../config/schema' +import type { FactoryCloudEventInputV1 } from '../observability/events' import type { SpawnInput, SpawnResult } from '../ports/fleet' +import type { FactoryEventReporter } from '../ports/observability' import type { LinearIssue, TriageEngine } from '../types' -import { createHostedFactory } from './orchestrator' +import { createHostedFactory, hostedFactoryRunId } from './orchestrator' import { InMemoryHostedFactoryStateStore } from './state-store' import type { HostedFactoryCompletion, HostedFactoryCompletionSource, HostedFactoryDiscovery, HostedFactoryFleet, + HostedFactoryIssueRecord, HostedFactoryWriteback, HostedFactoryWritebackInput, } from './types' @@ -109,12 +112,46 @@ class Writeback implements HostedFactoryWriteback { } } +class RecordingReporter implements FactoryEventReporter { + readonly events: FactoryCloudEventInputV1[] = [] + readonly snapshots: Array<{ + event: FactoryCloudEventInputV1 + record: HostedFactoryIssueRecord | null + }> = [] + readonly #state?: InMemoryHostedFactoryStateStore + + constructor(state?: InMemoryHostedFactoryStateStore) { + this.#state = state + } + + async report(event: FactoryCloudEventInputV1): Promise { + const saved = structuredClone(event) + this.events.push(saved) + this.snapshots.push({ + event: saved, + record: this.#state + ? await this.#state.getIssue(workspaceId, issue().uuid) + : null, + }) + } + + async flush() { + return { + delivered: this.events.length, + pending: 0, + attempts: this.events.length, + stoppedReason: 'empty' as const, + } + } +} + function harness(input: { discovery?: Discovery fleet?: HostedFactoryFleet completions?: Completions writeback?: Writeback state?: InMemoryHostedFactoryStateStore + reporter?: FactoryEventReporter ownerId?: string triage?: TriageEngine } = {}) { @@ -134,6 +171,7 @@ function harness(input: { writeback, state, triage: input.triage, + reporter: input.reporter, now: () => new Date('2026-07-19T12:00:00.000Z'), }) return { factory, discovery, fleet, completions, writeback, state } @@ -174,9 +212,72 @@ describe('HostedFactoryLoop', () => { ) }) + it('reports the exact dispatch lifecycle only after each fenced state save', async () => { + const state = new InMemoryHostedFactoryStateStore() + const reporter = new RecordingReporter(state) + const { factory } = harness({ state, reporter }) + + await expect(factory.runOnce()).resolves.toMatchObject({ + status: 'completed', + dispatched: ['AR-2778'], + }) + + expect(reporter.events.map(({ type }) => type)).toEqual([ + 'run.started', + 'run.phase_changed', + 'agent.planned', + 'agent.planned', + 'agent.spawned', + 'agent.spawned', + 'run.phase_changed', + 'writeback.applied', + ]) + expect(new Set(reporter.events.map(({ runId }) => runId))).toEqual(new Set([ + hostedFactoryRunId(workspaceId, 'issue-2778'), + ])) + expect(reporter.events[0]?.run).toMatchObject({ + source: 'linear', + issueKey: 'AR-2778', + }) + + const started = reporter.snapshots.find(({ event }) => event.type === 'run.started') + expect(started?.record).toMatchObject({ + runId: started?.event.runId, + phase: 'triaging', + invocations: [], + }) + + const planned = reporter.snapshots.filter(({ event }) => event.type === 'agent.planned') + expect(planned).toHaveLength(2) + expect(planned.every(({ record }) => + record?.phase === 'dispatching' && + record.invocations.every(({ status }) => status === 'pending') + )).toBe(true) + + const spawned = reporter.snapshots.filter(({ event }) => event.type === 'agent.spawned') + expect(spawned).toHaveLength(2) + expect(spawned.every(({ event, record }) => + record?.invocations.find(({ invocationId }) => + invocationId === event.attributes?.invocationId + )?.status === 'dispatched' + )).toBe(true) + + const phaseChanges = reporter.snapshots.filter(({ event }) => event.type === 'run.phase_changed') + expect(phaseChanges.map(({ record }) => record?.phase)).toEqual(['dispatching', 'running']) + expect(reporter.snapshots.at(-1)?.record?.dispatchWriteback).toMatchObject({ status: 'posted' }) + + const eventCount = reporter.events.length + await factory.runOnce() + expect(reporter.events).toHaveLength(eventCount) + expect(reporter.events.filter(({ type }) => type === 'run.started')).toHaveLength(1) + expect(reporter.events.filter(({ type }) => type === 'agent.spawned')).toHaveLength(2) + }) + it('polls invocation completion, runs the merge gate, and writes back exactly once', async () => { - const { factory, completions, state, writeback } = harness() + const reporter = new RecordingReporter() + const { factory, completions, state, writeback } = harness({ reporter }) await factory.runOnce() + reporter.events.length = 0 const running = await state.getIssue(workspaceId, 'issue-2778') for (const invocation of running?.invocations ?? []) { completions.values.set(invocation.invocationId, { @@ -196,6 +297,18 @@ describe('HostedFactoryLoop', () => { completionWriteback: { status: 'posted', attempts: 1 }, }) expect(writeback.completions).toHaveLength(1) + expect(reporter.events.map(({ type }) => type)).toEqual([ + 'agent.exited', + 'agent.exited', + 'run.phase_changed', + 'writeback.applied', + 'run.succeeded', + ]) + expect(reporter.events.at(-1)).toMatchObject({ + runId: completed?.runId, + phase: 'complete', + status: 'succeeded', + }) const terminalInvocationId = completed!.invocations[0]!.invocationId await expect(factory.ingestCompletion({ invocationId: terminalInvocationId, @@ -203,14 +316,17 @@ describe('HostedFactoryLoop', () => { error: 'late contradictory delivery', })).resolves.toMatchObject({ status: 'terminal', phase: 'complete' }) expect((await state.getIssue(workspaceId, 'issue-2778'))?.phase).toBe('complete') + expect(reporter.events.filter(({ type }) => type === 'agent.exited')).toHaveLength(2) await factory.runOnce() expect(writeback.completions).toHaveLength(1) }) it('ingests pushed completion and blocks the merge gate when a spawn fails', async () => { - const { factory, state, writeback } = harness() + const reporter = new RecordingReporter() + const { factory, state, writeback } = harness({ reporter }) await factory.runOnce() + reporter.events.length = 0 const running = await state.getIssue(workspaceId, 'issue-2778') const [implementer, reviewer] = running?.invocations ?? [] @@ -229,6 +345,19 @@ describe('HostedFactoryLoop', () => { mergeGate: { status: 'blocked' }, }) expect(writeback.completions).toHaveLength(1) + expect(reporter.events.map(({ type }) => type)).toEqual([ + 'agent.exited', + 'agent.exited', + 'run.phase_changed', + 'writeback.applied', + 'run.failed', + ]) + expect(reporter.events.at(-1)).toMatchObject({ + runId: running?.runId, + phase: 'blocked', + status: 'failed', + level: 'error', + }) expect(await factory.ingestCompletion({ invocationId: 'missing-invocation', status: 'completed', @@ -265,16 +394,143 @@ describe('HostedFactoryLoop', () => { expect((await state.getIssue(workspaceId, 'issue-2778'))?.phase).toBe('running') }) + it('retains run and invocation identity across a partial-dispatch retry', async () => { + class PartialAckLostFleet extends Fleet { + lost = false + + override async spawn(input: SpawnInput & { invocationId: string }) { + this.calls.push(structuredClone(input)) + if (this.calls.length === 2 && !this.lost) { + this.lost = true + throw new Error('second spawn accepted but acknowledgement was lost') + } + return { + name: input.name, + invocationId: input.invocationId, + locality: 'remote' as const, + } + } + } + const reporter = new RecordingReporter() + const fleet = new PartialAckLostFleet() + const { factory, state } = harness({ fleet, reporter }) + + const first = await factory.runOnce() + const partial = await state.getIssue(workspaceId, 'issue-2778') + expect(first.errors[0]?.message).toContain('acknowledgement was lost') + expect(partial?.invocations.map(({ status }) => status)).toEqual(['dispatched', 'pending']) + + const second = await factory.runOnce() + const running = await state.getIssue(workspaceId, 'issue-2778') + expect(second.dispatched).toEqual(['AR-2778']) + expect(running?.runId).toBe(partial?.runId) + expect(running?.invocations.map(({ invocationId }) => invocationId)).toEqual( + partial?.invocations.map(({ invocationId }) => invocationId), + ) + expect(fleet.calls.map(({ invocationId }) => invocationId)).toEqual([ + running?.invocations[0]?.invocationId, + running?.invocations[1]?.invocationId, + running?.invocations[1]?.invocationId, + ]) + expect(reporter.events.filter(({ type }) => type === 'run.started')).toHaveLength(1) + expect(reporter.events.filter(({ type }) => type === 'agent.spawned')).toHaveLength(2) + const failure = reporter.events.find(({ type }) => type === 'factory.failure') + expect(failure).toMatchObject({ + runId: running?.runId, + attributes: { + component: 'orchestrator', + operation: 'issue_sweep', + errorClass: 'Error', + }, + }) + expect(JSON.stringify(failure)).not.toContain('acknowledgement was lost') + }) + + it('retains its deterministic run identity after lease loss and a new host retry', async () => { + class AckSaveLeaseLostState extends InMemoryHostedFactoryStateStore { + saves = 0 + + override async saveIssue(record: HostedFactoryIssueRecord, lease: Parameters[1]) { + this.saves += 1 + if (this.saves === 3) return false + return await super.saveIssue(record, lease) + } + } + const state = new AckSaveLeaseLostState() + const reporter = new RecordingReporter() + const fleet = new Fleet() + const first = harness({ state, reporter, fleet, ownerId: 'host-a' }) + + await expect(first.factory.runOnce()).resolves.toMatchObject({ status: 'lease-lost' }) + const fencedRecord = await state.getIssue(workspaceId, 'issue-2778') + expect(fencedRecord).toMatchObject({ + runId: hostedFactoryRunId(workspaceId, 'issue-2778'), + phase: 'dispatching', + }) + + const second = harness({ state, reporter, fleet, ownerId: 'host-b' }) + await expect(second.factory.runOnce()).resolves.toMatchObject({ + status: 'completed', + dispatched: ['AR-2778'], + }) + expect((await state.getIssue(workspaceId, 'issue-2778'))?.runId).toBe(fencedRecord?.runId) + expect(fleet.calls[0]?.invocationId).toBe(fleet.calls[1]?.invocationId) + expect(reporter.events.filter(({ type }) => type === 'run.started')).toHaveLength(1) + }) + + it('backfills a deterministic run ID before saving a legacy hosted record', async () => { + const state = new InMemoryHostedFactoryStateStore() + const seed = await state.claimRunLease(workspaceId, 'seed', 60_000) + if (!seed.acquired) throw new Error('expected seed lease') + await state.saveIssue({ + workspaceId, + issue: issue(), + phase: 'awaiting-clarification', + invocations: [], + clarificationReason: 'legacy row', + createdAt: '2026-07-18T12:00:00.000Z', + updatedAt: '2026-07-18T12:00:00.000Z', + }, seed.lease) + await state.releaseRunLease(seed.lease) + const reporter = new RecordingReporter() + const { factory } = harness({ + state, + reporter, + discovery: new Discovery([]), + ownerId: 'host-after-upgrade', + }) + + await factory.runOnce() + + expect(await state.getIssue(workspaceId, 'issue-2778')).toMatchObject({ + runId: hostedFactoryRunId(workspaceId, 'issue-2778'), + clarificationWriteback: { status: 'posted' }, + }) + expect(reporter.events).toHaveLength(1) + expect(reporter.events[0]).toMatchObject({ + type: 'writeback.applied', + runId: hostedFactoryRunId(workspaceId, 'issue-2778'), + }) + }) + it('parks thin triage for clarification and dispatches after discovery returns richer context', async () => { const thin = issue({ description: 'Please fix it.' }) const discovery = new Discovery([thin]) - const { factory, fleet, state, writeback } = harness({ discovery }) + const reporter = new RecordingReporter() + const { factory, fleet, state, writeback } = harness({ discovery, reporter }) const first = await factory.runOnce() expect(first.awaitingClarification).toEqual(['AR-2778']) expect(writeback.clarifications).toHaveLength(1) expect((fleet as Fleet).calls).toHaveLength(0) expect((await state.getIssue(workspaceId, 'issue-2778'))?.phase).toBe('awaiting-clarification') + expect(reporter.events.map(({ type }) => type)).toEqual([ + 'run.started', + 'run.phase_changed', + 'run.waiting', + 'clarification.requested', + 'writeback.applied', + ]) discovery.issues = [issue()] const resumed = await factory.runOnce() @@ -283,6 +539,34 @@ describe('HostedFactoryLoop', () => { expect(writeback.clarifications).toHaveLength(1) }) + it('isolates a rejecting reporter from dispatch, durable state, and dedupe', async () => { + const reporter: FactoryEventReporter = { + report: vi.fn().mockRejectedValue(new Error('telemetry unavailable')), + flush: vi.fn().mockResolvedValue({ + delivered: 0, + pending: 1, + attempts: 1, + stoppedReason: 'unavailable', + }), + } + const fleet = new Fleet() + const { factory, state } = harness({ fleet, reporter }) + + await expect(factory.runOnce()).resolves.toMatchObject({ + status: 'completed', + dispatched: ['AR-2778'], + errors: [], + }) + expect(await state.getIssue(workspaceId, 'issue-2778')).toMatchObject({ + phase: 'running', + dispatchWriteback: { status: 'posted' }, + }) + expect(fleet.calls).toHaveLength(2) + + await expect(factory.runOnce()).resolves.toMatchObject({ status: 'completed' }) + expect(fleet.calls).toHaveLength(2) + }) + it('persists an undispatchable custom triage decision for human clarification', async () => { const triage: TriageEngine = { triage: async (candidate) => ({ @@ -382,9 +666,10 @@ describe('HostedFactoryLoop', () => { }) const discovery = new Discovery([issue(), secondIssue]) const fleet = new Fleet() + const reporter = new RecordingReporter() const spawn = vi.spyOn(fleet, 'spawn') spawn.mockRejectedValueOnce(new Error('transient placement failure')) - const { factory } = harness({ discovery, fleet }) + const { factory } = harness({ discovery, fleet, reporter }) const report = await factory.runOnce() @@ -392,5 +677,6 @@ describe('HostedFactoryLoop', () => { { issueKey: 'AR-2778', message: 'transient placement failure' }, ]) expect(report.dispatched).toEqual(['AR-2779']) + expect(reporter.events.filter(({ type }) => type === 'factory.failure')).toHaveLength(1) }) }) diff --git a/src/hosted/orchestrator.ts b/src/hosted/orchestrator.ts index 771c0fa..11411cc 100644 --- a/src/hosted/orchestrator.ts +++ b/src/hosted/orchestrator.ts @@ -1,6 +1,11 @@ import { TieredTriage } from '../triage/tiered' import { isInFactoryScope } from '../safety/factory-scope' +import { createFactoryCloudEventV1 } from '../observability/events' +import type { + FactoryCloudEventInputV1, + FactoryCloudReleaseReasonV1, +} from '../observability/events' import type { AgentSpec } from '../ports/fleet' import type { RepoMapEntry, TriageDecision } from '../types' import type { @@ -101,6 +106,8 @@ export class HostedFactoryLoop implements HostedFactory { } const now = this.#timestamp() + const previousPhase = record?.phase + const created = !record record = record ? { ...record, issue: structuredClone(issue), phase: 'triaging', updatedAt: now } : { @@ -112,6 +119,11 @@ export class HostedFactoryLoop implements HostedFactory { updatedAt: now, } await this.#save(record, lease) + if (created) { + await this.#reportLifecycle(record, 'run.started') + } else if (previousPhase !== record.phase) { + await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase }) + } const decision = await this.#triage.triage(issue, { config: this.#options.config, repoMap: repoMapFromConfig(this.#options.config), @@ -129,6 +141,12 @@ export class HostedFactoryLoop implements HostedFactory { updatedAt: this.#timestamp(), } await this.#save(record, lease) + await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase: 'triaging' }) + await this.#reportLifecycle(record, 'run.waiting') + await this.#reportLifecycle(record, 'clarification.requested', { + component: 'writeback', + operation: 'request_clarification', + }) lease = await this.#ensureWriteback(record, lease, 'clarification') report.awaitingClarification.push(issue.key) records = replaceRecord(records, record) @@ -150,6 +168,12 @@ export class HostedFactoryLoop implements HostedFactory { updatedAt: this.#timestamp(), } await this.#save(record, lease) + await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase: 'triaging' }) + await this.#reportLifecycle(record, 'run.waiting') + await this.#reportLifecycle(record, 'clarification.requested', { + component: 'writeback', + operation: 'request_clarification', + }) lease = await this.#ensureWriteback(record, lease, 'clarification') report.awaitingClarification.push(issue.key) records = replaceRecord(records, record) @@ -165,12 +189,24 @@ export class HostedFactoryLoop implements HostedFactory { updatedAt: this.#timestamp(), } await this.#save(record, lease) + await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase: 'triaging' }) + for (const invocation of record.invocations) { + await this.#reportAgent(record, invocation, 'agent.planned') + } activeCount += 1 lease = await this.#dispatchRecord(record, lease, report) records = replaceRecord(records, record) } catch (error) { if (error instanceof HostedFactoryLeaseLostError) throw error report.errors.push({ issueKey: issue.key, message: errorMessage(error) }) + if (record) { + await this.#reportLifecycle(record, 'factory.failure', { + component: 'orchestrator', + operation: 'issue_sweep', + errorClass: telemetryErrorClass(error), + retryable: true, + }) + } this.#ports.logger?.error?.('[factory:hosted] issue sweep failed', { workspaceId: this.#options.workspaceId, issueKey: issue.key, @@ -196,8 +232,9 @@ export class HostedFactoryLoop implements HostedFactory { return { status: 'fenced', invocationId: completion.invocationId } } let lease = claim.lease + let record: HostedFactoryIssueRecord | null = null try { - const record = await this.#ports.state.findIssueByInvocation( + record = await this.#ports.state.findIssueByInvocation( this.#options.workspaceId, completion.invocationId, ) @@ -210,8 +247,15 @@ export class HostedFactoryLoop implements HostedFactory { phase: record.phase, } } + const invocation = record.invocations.find( + (candidate) => candidate.invocationId === completion.invocationId, + ) + const wasTerminal = invocation ? isTerminalInvocation(invocation) : false applyCompletion(record, completion, this.#timestamp()) await this.#save(record, lease) + if (invocation && !wasTerminal && isTerminalInvocation(invocation)) { + await this.#reportAgent(record, invocation, 'agent.exited') + } if (record.invocations.every(isTerminalInvocation)) { lease = await this.#finishRecord(record, lease) } @@ -225,6 +269,14 @@ export class HostedFactoryLoop implements HostedFactory { if (error instanceof HostedFactoryLeaseLostError) { return { status: 'fenced', invocationId: completion.invocationId } } + if (record) { + await this.#reportLifecycle(record, 'factory.failure', { + component: 'orchestrator', + operation: 'completion_ingest', + errorClass: telemetryErrorClass(error), + retryable: true, + }) + } throw error } finally { await this.#release(lease) @@ -254,6 +306,7 @@ export class HostedFactoryLoop implements HostedFactory { lease = await this.#ensureWriteback(record, lease, 'dispatch') } let changed = false + const exited: HostedFactoryInvocation[] = [] for (const invocation of record.invocations) { if (isTerminalInvocation(invocation)) continue const completion = await this.#ports.completions.getInvocation({ @@ -262,10 +315,14 @@ export class HostedFactoryLoop implements HostedFactory { }) if (!completion || completion.status === invocation.status) continue applyCompletion(record, completion, this.#timestamp()) + if (isTerminalInvocation(invocation)) exited.push(invocation) changed = true } if (changed) { await this.#save(record, lease) + for (const invocation of exited) { + await this.#reportAgent(record, invocation, 'agent.exited') + } report.reconciled.push(record.issue.key) } if (record.invocations.every(isTerminalInvocation)) { @@ -274,6 +331,12 @@ export class HostedFactoryLoop implements HostedFactory { } catch (error) { if (error instanceof HostedFactoryLeaseLostError) throw error report.errors.push({ issueKey: record.issue.key, message: errorMessage(error) }) + await this.#reportLifecycle(record, 'factory.failure', { + component: 'orchestrator', + operation: 'reconciliation', + errorClass: telemetryErrorClass(error), + retryable: true, + }) this.#ports.logger?.error?.('[factory:hosted] reconciliation failed', { workspaceId: this.#options.workspaceId, issueKey: record.issue.key, @@ -309,11 +372,16 @@ export class HostedFactoryLoop implements HostedFactory { record.updatedAt = this.#timestamp() lease = await this.#renew(lease) await this.#save(record, lease) + await this.#reportAgent(record, invocation, 'agent.spawned') } if (record.invocations.every((invocation) => invocation.status !== 'pending')) { + const previousPhase = record.phase record.phase = 'running' record.updatedAt = this.#timestamp() await this.#save(record, lease) + if (previousPhase !== record.phase) { + await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase }) + } lease = await this.#ensureWriteback(record, lease, 'dispatch') if (!report.dispatched.includes(record.issue.key)) report.dispatched.push(record.issue.key) } @@ -325,9 +393,13 @@ export class HostedFactoryLoop implements HostedFactory { initialLease: HostedFactoryLease, ): Promise { let lease = await this.#renew(initialLease) + const previousPhase = record.phase record.phase = 'merge-gate' record.updatedAt = this.#timestamp() await this.#save(record, lease) + if (previousPhase !== record.phase) { + await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase }) + } const mergeGate = await (this.#ports.mergeGate?.evaluate({ workspaceId: this.#options.workspaceId, record: structuredClone(record), @@ -346,6 +418,11 @@ export class HostedFactoryLoop implements HostedFactory { record.phase = mergeGate.status === 'ready' ? 'complete' : 'blocked' record.updatedAt = this.#timestamp() await this.#save(record, lease) + await this.#reportLifecycle( + record, + record.phase === 'complete' ? 'run.succeeded' : 'run.failed', + { previousPhase: 'merge-gate' }, + ) } return lease } @@ -362,6 +439,7 @@ export class HostedFactoryLoop implements HostedFactory { record.updatedAt = this.#timestamp() await this.#save(record, initialLease) let lease = await this.#renew(initialLease) + let applied = false try { const input = { workspaceId: this.#options.workspaceId, @@ -381,6 +459,7 @@ export class HostedFactoryLoop implements HostedFactory { } lease = await this.#renew(lease) record[field] = { status: 'posted', attempts, postedAt: this.#timestamp() } + applied = true } catch (error) { record[field] = { status: 'failed', attempts, error: errorMessage(error) } this.#ports.logger?.warn?.('[factory:hosted] writeback failed', { @@ -392,9 +471,98 @@ export class HostedFactoryLoop implements HostedFactory { } record.updatedAt = this.#timestamp() await this.#save(record, lease) + if (applied) { + await this.#reportLifecycle(record, 'writeback.applied', { + component: 'writeback', + operation: kind, + }) + } return lease } + async #report( + input: Parameters[0], + ): Promise { + if (!this.#ports.reporter) return + try { + await this.#ports.reporter.report(createFactoryCloudEventV1(input, { + now: this.#now, + })) + } catch (error) { + this.#ports.logger?.warn?.('[factory:hosted] progress reporter rejected an event', { + eventType: input.type, + errorClass: telemetryErrorClass(error), + }) + } + } + + async #reportLifecycle( + record: HostedFactoryIssueRecord, + type: FactoryCloudEventInputV1['type'], + options: { + level?: FactoryCloudEventInputV1['level'] + previousPhase?: HostedFactoryIssueRecord['phase'] + component?: string + operation?: string + errorClass?: string + errorCode?: string + retryable?: boolean + } = {}, + ): Promise { + await this.#report({ + type, + level: options.level ?? ( + type === 'run.failed' || type === 'factory.failure' ? 'error' : 'info' + ), + runId: record.runId ?? hostedFactoryRunId(record.workspaceId, record.issue.uuid), + phase: record.phase, + status: hostedTelemetryRunStatus(record.phase, type), + run: hostedTelemetryRun(record), + attributes: { + backend: 'hosted', + component: options.component ?? 'orchestrator', + operation: options.operation ?? 'save_lifecycle', + previousPhase: options.previousPhase, + errorClass: options.errorClass, + errorCode: options.errorCode, + retryable: options.retryable, + trackedAgents: record.invocations.length, + }, + }) + } + + async #reportAgent( + record: HostedFactoryIssueRecord, + invocation: HostedFactoryInvocation, + type: 'agent.planned' | 'agent.spawned' | 'agent.exited', + ): Promise { + await this.#report({ + type, + level: invocation.status === 'failed' ? 'error' : 'info', + runId: record.runId ?? hostedFactoryRunId(record.workspaceId, record.issue.uuid), + phase: record.phase, + status: hostedTelemetryRunStatus(record.phase, type), + run: { + ...hostedTelemetryRun(record), + repository: invocation.spec.repo, + }, + attributes: { + backend: 'hosted', + component: 'fleet', + operation: type.slice('agent.'.length), + agentRole: invocation.spec.role, + invocationId: invocation.invocationId, + nodeId: invocation.node, + releaseReason: hostedInvocationReleaseReason(invocation), + errorCode: invocation.status === 'failed' + ? 'agent_failed' + : invocation.status === 'cancelled' + ? 'agent_cancelled' + : undefined, + }, + }) + } + async #claim(operation: string) { this.#operationSequence += 1 return await this.#ports.state.claimRunLease( @@ -411,6 +579,7 @@ export class HostedFactoryLoop implements HostedFactory { } async #save(record: HostedFactoryIssueRecord, lease: HostedFactoryLease): Promise { + record.runId ??= hostedFactoryRunId(record.workspaceId, record.issue.uuid) if (!await this.#ports.state.saveIssue(record, lease)) { throw new HostedFactoryLeaseLostError() } @@ -455,6 +624,14 @@ export function hostedFactoryInvocationId( return `factory:${stableHash(identity)}` } +export function hostedFactoryRunId(workspaceId: string, issueUuid: string): string { + return `factory:hosted:${stableHash(JSON.stringify([ + 'agent-relay.factory.hosted-run.v1', + workspaceId, + issueUuid, + ]))}` +} + function defaultMergeGate( record: HostedFactoryIssueRecord, decidedAt: string, @@ -547,6 +724,48 @@ function isTerminalIssue(record: HostedFactoryIssueRecord): boolean { return record.phase === 'complete' || record.phase === 'blocked' } +function hostedTelemetryRun( + record: HostedFactoryIssueRecord, +): NonNullable { + return { + source: 'linear', + repository: record.decision?.routes[0]?.repo ?? record.invocations[0]?.spec.repo, + issueKey: record.issue.key, + recipe: record.decision?.scope, + } +} + +function hostedTelemetryRunStatus( + phase: HostedFactoryIssueRecord['phase'], + type: FactoryCloudEventInputV1['type'], +): NonNullable { + if (type === 'run.failed') return 'failed' + if (type === 'run.succeeded') return 'succeeded' + switch (phase) { + case 'awaiting-clarification': + return 'waiting' + case 'blocked': + return 'blocked' + case 'complete': + return 'succeeded' + default: + return 'running' + } +} + +function hostedInvocationReleaseReason( + invocation: HostedFactoryInvocation, +): FactoryCloudReleaseReasonV1 | undefined { + if (invocation.status === 'completed') return 'completed' + if (invocation.status === 'failed' || invocation.status === 'cancelled') return 'other' + return undefined +} + +function telemetryErrorClass(error: unknown): string { + const name = error instanceof Error ? error.name : '' + return /^[A-Za-z][A-Za-z0-9]{0,63}(?:Error|Exception)$/u.test(name) ? name : 'Error' +} + function dedupeIssues(issues: T[]): T[] { return [...new Map(issues.map((issue) => [issue.uuid, issue])).values()] } diff --git a/src/hosted/types.ts b/src/hosted/types.ts index 8bc4a8b..f30bd8e 100644 --- a/src/hosted/types.ts +++ b/src/hosted/types.ts @@ -1,5 +1,6 @@ import type { FactoryConfig } from '../config/schema' import type { AgentSpec, SpawnInput, SpawnResult } from '../ports/fleet' +import type { FactoryEventReporter } from '../ports/observability' import type { Logger } from '../ports/system' import type { LinearIssue, TriageDecision, TriageEngine } from '../types' @@ -63,6 +64,8 @@ export type HostedFactoryWritebackRecord = { export type HostedFactoryIssueRecord = { workspaceId: string + /** Deterministic durable-run identity. Optional only for rows written before hosted telemetry existed. */ + runId?: string issue: LinearIssue phase: HostedFactoryIssuePhase decision?: TriageDecision @@ -167,6 +170,7 @@ export type HostedFactoryPorts = { writeback: HostedFactoryWriteback triage?: TriageEngine mergeGate?: HostedFactoryMergeGate + reporter?: FactoryEventReporter logger?: Logger now?: () => Date } diff --git a/src/hosted/worker-safety.test.ts b/src/hosted/worker-safety.test.ts index f974d37..3b643f9 100644 --- a/src/hosted/worker-safety.test.ts +++ b/src/hosted/worker-safety.test.ts @@ -1,18 +1,39 @@ import { build } from 'esbuild' +import { readFile } from 'node:fs/promises' import { describe, expect, it } from 'vitest' -describe('hosted Factory entrypoint', () => { - it('bundles for a worker/browser runtime without Node built-ins', async () => { +describe('hosted Factory Worker entrypoints', () => { + it('bundles hosted orchestration and narrow telemetry for Workers nodejs_compat', async () => { const result = await build({ - entryPoints: ['src/hosted/index.ts'], + entryPoints: { + hosted: 'src/hosted/index.ts', + telemetry: 'src/telemetry.ts', + }, bundle: true, platform: 'browser', format: 'esm', target: 'es2022', + outdir: 'out', + external: ['node:crypto'], + metafile: true, write: false, logLevel: 'silent', }) - expect(result.outputFiles[0]?.text).toContain('HostedFactoryLoop') + expect(result.outputFiles.find(({ path }) => path.endsWith('/hosted.js'))?.text) + .toContain('HostedFactoryLoop') + expect(result.outputFiles.find(({ path }) => path.endsWith('/telemetry.js'))?.text) + .toContain('FactoryCloudEventInputV1Schema') + expect(Object.keys(result.metafile.inputs)).not.toContain('src/observability/outbox.ts') + expect(Object.keys(result.metafile.inputs)).not.toContain('src/observability/instance-identity.ts') + + const packageJson = JSON.parse(await readFile( + new URL('../../package.json', import.meta.url), + 'utf8', + )) as { exports: Record } + expect(packageJson.exports['./telemetry']).toEqual({ + types: './dist/telemetry.d.ts', + import: './dist/telemetry.js', + }) }) }) diff --git a/src/telemetry.ts b/src/telemetry.ts new file mode 100644 index 0000000..60c7772 --- /dev/null +++ b/src/telemetry.ts @@ -0,0 +1,11 @@ +/** + * Worker-safe telemetry contract surface. + * + * Keep this entrypoint narrow: the observability barrel also exports the + * filesystem-backed outbox and instance identity implementation. + */ +export * from './observability/events.js' +export type { + FactoryEventReporter, + FactoryEventReportResult, +} from './ports/observability.js' From e2f56dccfbf72a4ba768f986e060cb4a01cf0095 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 23 Jul 2026 14:34:28 +0200 Subject: [PATCH 2/2] fix(hosted): dedupe clarification telemetry --- src/hosted/orchestrator.test.ts | 19 +++++++++++++++++ src/hosted/orchestrator.ts | 38 ++++++++++++++++++++------------- 2 files changed, 42 insertions(+), 15 deletions(-) diff --git a/src/hosted/orchestrator.test.ts b/src/hosted/orchestrator.test.ts index ab8f001..4343286 100644 --- a/src/hosted/orchestrator.test.ts +++ b/src/hosted/orchestrator.test.ts @@ -532,11 +532,30 @@ describe('HostedFactoryLoop', () => { 'writeback.applied', ]) + const waitingEventCount = reporter.events.length + const unchanged = await factory.runOnce() + expect(unchanged.awaitingClarification).toEqual(['AR-2778']) + expect(writeback.clarifications).toHaveLength(1) + expect(reporter.events).toHaveLength(waitingEventCount) + discovery.issues = [issue()] const resumed = await factory.runOnce() expect(resumed.dispatched).toEqual(['AR-2778']) expect((fleet as Fleet).calls).toHaveLength(2) expect(writeback.clarifications).toHaveLength(1) + expect(reporter.events.slice(waitingEventCount).map(({ type }) => type)).toEqual([ + 'run.phase_changed', + 'agent.planned', + 'agent.planned', + 'agent.spawned', + 'agent.spawned', + 'run.phase_changed', + 'writeback.applied', + ]) + expect(reporter.events[waitingEventCount]).toMatchObject({ + phase: 'dispatching', + attributes: { previousPhase: 'awaiting-clarification' }, + }) }) it('isolates a rejecting reporter from dispatch, durable state, and dedupe', async () => { diff --git a/src/hosted/orchestrator.ts b/src/hosted/orchestrator.ts index 11411cc..2a9d789 100644 --- a/src/hosted/orchestrator.ts +++ b/src/hosted/orchestrator.ts @@ -121,8 +121,6 @@ export class HostedFactoryLoop implements HostedFactory { await this.#save(record, lease) if (created) { await this.#reportLifecycle(record, 'run.started') - } else if (previousPhase !== record.phase) { - await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase }) } const decision = await this.#triage.triage(issue, { config: this.#options.config, @@ -141,12 +139,16 @@ export class HostedFactoryLoop implements HostedFactory { updatedAt: this.#timestamp(), } await this.#save(record, lease) - await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase: 'triaging' }) - await this.#reportLifecycle(record, 'run.waiting') - await this.#reportLifecycle(record, 'clarification.requested', { - component: 'writeback', - operation: 'request_clarification', - }) + if (previousPhase !== 'awaiting-clarification') { + await this.#reportLifecycle(record, 'run.phase_changed', { + previousPhase: previousPhase ?? 'triaging', + }) + await this.#reportLifecycle(record, 'run.waiting') + await this.#reportLifecycle(record, 'clarification.requested', { + component: 'writeback', + operation: 'request_clarification', + }) + } lease = await this.#ensureWriteback(record, lease, 'clarification') report.awaitingClarification.push(issue.key) records = replaceRecord(records, record) @@ -168,12 +170,16 @@ export class HostedFactoryLoop implements HostedFactory { updatedAt: this.#timestamp(), } await this.#save(record, lease) - await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase: 'triaging' }) - await this.#reportLifecycle(record, 'run.waiting') - await this.#reportLifecycle(record, 'clarification.requested', { - component: 'writeback', - operation: 'request_clarification', - }) + if (previousPhase !== 'awaiting-clarification') { + await this.#reportLifecycle(record, 'run.phase_changed', { + previousPhase: previousPhase ?? 'triaging', + }) + await this.#reportLifecycle(record, 'run.waiting') + await this.#reportLifecycle(record, 'clarification.requested', { + component: 'writeback', + operation: 'request_clarification', + }) + } lease = await this.#ensureWriteback(record, lease, 'clarification') report.awaitingClarification.push(issue.key) records = replaceRecord(records, record) @@ -189,7 +195,9 @@ export class HostedFactoryLoop implements HostedFactory { updatedAt: this.#timestamp(), } await this.#save(record, lease) - await this.#reportLifecycle(record, 'run.phase_changed', { previousPhase: 'triaging' }) + await this.#reportLifecycle(record, 'run.phase_changed', { + previousPhase: previousPhase ?? 'triaging', + }) for (const invocation of record.invocations) { await this.#reportAgent(record, invocation, 'agent.planned') }