From a12fcb8703706c05a4e8a7a4c10d3b2e9c4c5c06 Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 13 Aug 2026 10:34:16 +0200 Subject: [PATCH 1/6] feat(factory): pluggable onTicketDispatch delivery --- src/config/schema.ts | 29 ++++++++ src/delivery/ticket-dispatch.ts | 29 ++++++++ src/orchestrator/factory.test.ts | 82 ++++++++++++++++++++++ src/orchestrator/factory.ts | 112 +++++++++++++++++++++++++++++++ src/ports/fleet.ts | 4 ++ src/triage/schema.ts | 2 + src/types.ts | 3 + 7 files changed, 261 insertions(+) create mode 100644 src/delivery/ticket-dispatch.ts diff --git a/src/config/schema.ts b/src/config/schema.ts index ee15fec..3381207 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -143,6 +143,32 @@ const reportingSchema = z.object({ requestTimeoutMs: z.number().int().min(100).max(60_000).default(15_000), }).default({}) +const ticketDispatchNotificationSchema = z.discriminatedUnion('surface', [ + z.object({ + surface: z.literal('relay'), + channel: z.string().trim().min(1), + }).strict(), + z.object({ + surface: z.literal('slack'), + channel: z.string().trim().min(1).optional(), + dm: z.string().trim().min(1).optional(), + }).strict(), + z.object({ + surface: z.literal('telegram'), + chatId: z.string().trim().min(1), + }).strict(), + z.object({ + surface: z.literal('linear'), + commentOnIssue: z.boolean(), + }).strict(), +]) + +const hooksSchema = z.object({ + onTicketDispatch: z.object({ + notify: z.array(ticketDispatchNotificationSchema).min(1), + }).strict().optional(), +}).strict().optional() + const previewServiceSchema = z.object({ /** Local HTTP port the repository's development server listens on. */ port: z.number().int().min(1).max(65_535), @@ -283,6 +309,9 @@ const WorkspaceConfigObjectSchema = z.object({ // analytics. It defaults on for real CLI sessions and remains no-op when no // Cloud account is available; delivery failure never changes orchestration. reporting: reportingSchema, + // Optional fan-out for the point an agent team has been successfully + // dispatched. Every configured surface is attempted independently. + hooks: hooksSchema, preview: previewSchema, github: githubSchema, verification: verificationSchema, diff --git a/src/delivery/ticket-dispatch.ts b/src/delivery/ticket-dispatch.ts new file mode 100644 index 0000000..aadfaa4 --- /dev/null +++ b/src/delivery/ticket-dispatch.ts @@ -0,0 +1,29 @@ +import { slackClient, telegramClient } from '@relayfile/relay-helpers' + +export interface TicketDispatchDelivery { + slack(input: { channel?: string; dm?: string; text: string }): Promise + telegram(input: { chatId: string; text: string }): Promise +} + +export function createTicketDispatchDelivery(options: { + mountRoot?: string +} = {}): TicketDispatchDelivery { + const clientOptions = options.mountRoot ? { mountRoot: options.mountRoot } : undefined + + return { + async slack({ channel, dm, text }): Promise { + const client = slackClient(clientOptions) + await Promise.all([ + ...(channel ? [client.post(channel, text)] : []), + ...(dm ? [client.dm(dm, text)] : []), + ]) + }, + + async telegram({ chatId, text }): Promise { + const result = await telegramClient(clientOptions).sendMessage(chatId, text) + if (!result.ok || !result.messageId) { + throw new Error(`Telegram delivery to ${chatId} returned no receipt`) + } + }, + } +} diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 3bb9a36..f7dde89 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -9623,6 +9623,88 @@ describe('FactoryLoop', () => { expect(new Set(fleet.spawns.map((spawn) => spawn.invocationId)).size).toBe(2) }) + it('delivers one onTicketDispatch payload to every configured surface', async () => { + const path = issuePath(124) + const mount = new FakeMountClient({ [path]: realIssueFile(124) }) + const fleet = new FakeFleetClient() + fleet.setSessionRef('ar-124-impl-pear', 'session-ar-124-impl-pear') + const timestamp = '2026-08-13T08:13:00.000Z' + const notifications: Array<{ surface: 'slack' | 'telegram'; text: string }> = [] + const linearComments: string[] = [] + const factory = createFactory(config({ + hooks: { + onTicketDispatch: { + notify: [ + { surface: 'relay', channel: 'dispatch-notifications' }, + { surface: 'slack', channel: 'C123', dm: 'U456' }, + { surface: 'telegram', chatId: '-100789' }, + { surface: 'linear', commentOnIssue: true }, + ], + }, + }, + }), { + mount, + fleet, + triage: new StaticTriage(), + linear: { + async setState() {}, + async postComment(_issue, text) { + linearComments.push(text) + }, + async createIssue() { + throw new Error('not used') + }, + async verify() { + return true + }, + }, + ticketDispatchDelivery: { + async slack({ text }) { + notifications.push({ surface: 'slack', text }) + }, + async telegram({ text }) { + notifications.push({ surface: 'telegram', text }) + }, + }, + clock: { now: () => Date.parse(timestamp), sleep: async () => {} }, + }) + const decision = await factory.triageIssue(parseLinearIssue(path, realIssueFile(124))) + decision.implementers[0]!.principal = 'broker' + + await factory.dispatch(decision) + + const relayText = fleet.messages.find((message) => message.to === '#dispatch-notifications')?.text + const linearText = linearComments.find((text) => text.startsWith('Ticket dispatched:')) + expect(relayText).toBeDefined() + expect(notifications.map((notification) => notification.surface)).toEqual(['slack', 'telegram']) + expect(linearText).toBeDefined() + + const payloads = [ + relayText!, + ...notifications.map((notification) => notification.text), + linearText!, + ].map((text) => JSON.parse(text.slice(text.indexOf('\n') + 1))) + + for (const payload of payloads) { + expect(payload).toEqual({ + eventType: 'ticket.dispatched', + summary: 'Ticket dispatched: [factory-e2e] Fix factory issue 124 → ar-124-impl-pear.', + issue: { + id: 'uuid-124', + title: '[factory-e2e] Fix factory issue 124', + url: 'https://linear.app/agent-relay/issue/AR-124/factory-issue-124', + }, + agent: { + name: 'ar-124-impl-pear', + sessionRef: 'session-ar-124-impl-pear', + }, + sessionOwner: 'broker', + timestamp, + }) + } + expect(factory.status().counters.ticketDispatchHookNotifications).toBe(4) + }) + it('derives implementer dispatch identities from repo labels instead of triage implementers', async () => { const routedIssue = realIssueFile(720, ready, { title: '[factory-e2e] Relayfile webhooks to cloud', diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 2b8b4cb..97d9081 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -113,9 +113,18 @@ import { type FactoryCloudEventInputV1, } from '../observability/events' import { boundedRunCostTotal, CostLedger, type RunCostTotal, type UnpricedModelCostRecord } from '../cost/ledger' +import { createTicketDispatchDelivery, type TicketDispatchDelivery } from '../delivery/ticket-dispatch' type FactoryEvent = 'issue-queued' | 'dispatched' | 'issue-done' | 'writeback-verified' | 'error' type Listener = (payload: FactoryEventPayload) => void +type TicketDispatchNotificationPayload = { + eventType: 'ticket.dispatched' + summary: string + issue: { id: string; title: string; url: string } + agent: { name: string; sessionRef: string | null } + sessionOwner: string | null + timestamp: string +} type SlackThreadWatcher = { stop(): Promise } type GithubIssueCommentWatcher = { stop(): Promise } type TerminationRoots = { pids: number[]; status: AgentPidResolution['status'] } @@ -407,6 +416,7 @@ export class FactoryLoop implements Factory { readonly #mount: MountClient readonly #states: FactoryStateResolution readonly #fleet: FleetClient + readonly #ticketDispatchDelivery: TicketDispatchDelivery readonly #triage: TriageEngine readonly #linear: LinearWriteback readonly #github: GithubRead @@ -601,6 +611,9 @@ export class FactoryLoop implements Factory { this.#states = ports.stateResolution ?? stateResolutionFromIds(config.stateIds, config.linear.states) installFactoryDraftPredicate(this.#mount, config) this.#fleet = ports.fleet + this.#ticketDispatchDelivery = ports.ticketDispatchDelivery ?? createTicketDispatchDelivery({ + mountRoot: config.localMountRoot, + }) this.#triage = ports.triage ?? new TieredTriage(new HeuristicTriage()) this.#linear = ports.linear ?? MountLinearWriteback(ports.mount, { safety: config.safety, @@ -2726,6 +2739,9 @@ export class FactoryLoop implements Factory { await this.#saveDispatchLifecycle(record, 'running') this.#increment('dispatched') this.#emit('dispatched', { issue: dispatchDecision.issue, result }) + if (!dryRun && this.#config.hooks?.onTicketDispatch) { + await this.#notifyTicketDispatch(decision, liveIssue, record, result) + } if (!dryRun) { await this.#ensureSlackDispatchThread(record, result) } @@ -11098,6 +11114,81 @@ export class FactoryLoop implements Factory { } } + async #notifyTicketDispatch( + decision: TriageDecision, + issue: LinearIssue, + record: InFlightIssue, + result: DispatchResult, + ): Promise { + const hook = this.#config.hooks?.onTicketDispatch + if (!hook || result.dryRun) return + + const agent = result.agents.find((candidate) => candidate.role === 'implementer') + ?? result.agents.find((candidate) => candidate.role === 'workflow') + ?? result.agents[0] + if (!agent) return + + const tracked = record.agents.get(agent.name) + const payload: TicketDispatchNotificationPayload = { + eventType: 'ticket.dispatched', + summary: `Ticket dispatched: ${issue.title} → ${agent.name}.`, + issue: { + id: issue.uuid, + title: issue.title, + url: dispatchIssueUrl(issue), + }, + agent: { + name: agent.name, + sessionRef: tracked?.sessionRef ?? null, + }, + sessionOwner: dispatchSessionOwner(decision) ?? null, + timestamp: new Date(this.#clock.now()).toISOString(), + } + const text = ticketDispatchNotificationText(payload) + + for (const target of hook.notify) { + if (target.surface === 'linear' && !target.commentOnIssue) continue + + try { + switch (target.surface) { + case 'relay': + await this.#fleet.sendMessage({ + to: `#${target.channel.replace(/^#/u, '')}`, + text, + }) + break + case 'slack': + if (!target.channel && !target.dm) { + throw new Error('Slack ticket-dispatch notification needs channel and/or dm') + } + await this.#ticketDispatchDelivery.slack({ + channel: target.channel, + dm: target.dm, + text, + }) + break + case 'telegram': + await this.#ticketDispatchDelivery.telegram({ chatId: target.chatId, text }) + break + case 'linear': + if (isGithubIssue(issue)) { + throw new Error('Linear ticket-dispatch comments require a Linear issue') + } + await this.#linear.postComment(issue, text) + break + } + this.#increment('ticketDispatchHookNotifications') + } catch (error) { + this.#increment('ticketDispatchHookFailures') + this.#logger.warn?.('[factory] onTicketDispatch notification failed', { + issue: issue.key, + surface: target.surface, + error: describeError(error).errorMessage, + }) + } + } + } + #error(error: unknown, issue?: IssueRef): void { this.#increment('errors') const details = describeError(error) @@ -13620,6 +13711,27 @@ function dispatchSpecs(decision: TriageDecision): AgentSpec[] { return [...decision.implementers, decision.reviewer] } +function dispatchSessionOwner(decision: TriageDecision): string | undefined { + for (const spec of dispatchSpecs(decision)) { + const sessionOwner = spec.principal?.trim() || spec.owner?.trim() + if (sessionOwner) return sessionOwner + } + return undefined +} + +function dispatchIssueUrl(issue: LinearIssue): string { + const payload = wrappedPayload(issue.raw) + const source = asRecord(payload.source) + return stringValue(payload.url) + ?? stringValue(payload.html_url) + ?? stringValue(source?.url) + ?? issue.path +} + +function ticketDispatchNotificationText(payload: TicketDispatchNotificationPayload): string { + return `${payload.summary}\n${JSON.stringify(payload)}` +} + function previewServiceForRepo( config: FactoryConfig, repo: string, diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index 4b736bd..1616c3e 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -185,6 +185,10 @@ export interface FleetClient { export type AgentSpec = { name: string role: 'implementer' | 'reviewer' | 'babysitter' | 'workflow' + /** Principal that initiated the agent session, when supplied by the dispatcher. */ + principal?: string + /** Compatibility alias for dispatchers that identify the initiating principal as an owner. */ + owner?: string capability: Capability model?: string task: string diff --git a/src/triage/schema.ts b/src/triage/schema.ts index 887aaca..6428a25 100644 --- a/src/triage/schema.ts +++ b/src/triage/schema.ts @@ -3,6 +3,8 @@ import { z } from 'zod' export const AgentSpecSchema = z.object({ name: z.string(), role: z.enum(['implementer', 'reviewer', 'babysitter', 'workflow']), + principal: z.string().optional(), + owner: z.string().optional(), capability: z.enum(['spawn:codex', 'spawn:claude', 'workflow:run']), model: z.string().optional(), task: z.string(), diff --git a/src/types.ts b/src/types.ts index fc6be8f..c28a363 100644 --- a/src/types.ts +++ b/src/types.ts @@ -11,6 +11,7 @@ import type { AgentProcessFinder, ProcessIdentity } from './orchestrator/process import type { DispatchRelayflowOptions, RelayflowPolicyRegistry } from './dispatch/relayflow-registry' import type { VerificationGate } from './environments/verification-pipeline' import type { CostLedger } from './cost/ledger' +import type { TicketDispatchDelivery } from './delivery/ticket-dispatch' export interface FactoryPorts { mount: MountClient @@ -23,6 +24,8 @@ export interface FactoryPorts { triage?: TriageEngine linear?: LinearWriteback slack?: SlackWriteback + /** Injectable delivery adapter for onTicketDispatch Slack/Telegram notifications. */ + ticketDispatchDelivery?: TicketDispatchDelivery github?: GithubRead githubWriteback?: GithubWriteback mergeGate?: GithubMergeGate From 7f288d036d2dfa9ec5cdc338417f4f77ea0e65ee Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 13 Aug 2026 20:59:10 +0200 Subject: [PATCH 2/6] fix(factory): fence ticket dispatch notifications --- src/config/schema.test.ts | 12 ++++ src/config/schema.ts | 12 +++- src/orchestrator/factory.test.ts | 98 +++++++++++++++++++++++++++++++- src/orchestrator/factory.ts | 4 +- 4 files changed, 120 insertions(+), 6 deletions(-) diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index d0da32e..f960e92 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -96,6 +96,18 @@ describe('FactoryConfigSchema', () => { expect(parsed.environments).toEqual({}) }) + it('rejects ticket-dispatch notification targets that cannot deliver', () => { + const withTarget = (target: Record) => FactoryConfigSchema.safeParse({ + repos: { default: 'AgentWorkforce/factory' }, + hooks: { onTicketDispatch: { notify: [target] } }, + }) + + expect(withTarget({ surface: 'slack' }).success).toBe(false) + expect(withTarget({ surface: 'linear', commentOnIssue: false }).success).toBe(false) + expect(withTarget({ surface: 'slack', dm: 'U123' }).success).toBe(true) + expect(withTarget({ surface: 'linear', commentOnIssue: true }).success).toBe(true) + }) + it('accepts secret-reference-only Kubernetes BYOC and managed connections', () => { const parsed = FactoryConfigSchema.parse({ repos: { default: 'AgentWorkforce/factory' }, diff --git a/src/config/schema.ts b/src/config/schema.ts index 3381207..eb36d2c 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -159,9 +159,17 @@ const ticketDispatchNotificationSchema = z.discriminatedUnion('surface', [ }).strict(), z.object({ surface: z.literal('linear'), - commentOnIssue: z.boolean(), + commentOnIssue: z.literal(true), }).strict(), -]) +]).superRefine((target, ctx) => { + if (target.surface === 'slack' && !target.channel && !target.dm) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['channel'], + message: 'Slack ticket-dispatch notifications require channel and/or dm', + }) + } +}) const hooksSchema = z.object({ onTicketDispatch: z.object({ diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index f7dde89..b806041 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -9669,7 +9669,7 @@ describe('FactoryLoop', () => { clock: { now: () => Date.parse(timestamp), sleep: async () => {} }, }) const decision = await factory.triageIssue(parseLinearIssue(path, realIssueFile(124))) - decision.implementers[0]!.principal = 'broker' + decision.implementers[0]!.principal = 'stale-triage-owner' await factory.dispatch(decision) @@ -9698,13 +9698,107 @@ describe('FactoryLoop', () => { name: 'ar-124-impl-pear', sessionRef: 'session-ar-124-impl-pear', }, - sessionOwner: 'broker', + sessionOwner: null, timestamp, }) } expect(factory.status().counters.ticketDispatchHookNotifications).toBe(4) }) + it('fans out onTicketDispatch only after the running lifecycle owner is persisted', async () => { + class RejectFirstRunningLifecycleSaveStore extends FileStateStore { + rejectedRunningSave = false + + override async saveDispatchLifecycle( + ...args: Parameters + ): Promise { + const lifecycle = args[5] + if (lifecycle.phase === 'running' && !this.rejectedRunningSave) { + this.rejectedRunningSave = true + return false + } + return await super.saveDispatchLifecycle(...args) + } + } + + const root = await mkdtemp(join(tmpdir(), 'factory-ticket-dispatch-owner-')) + const watchStatePath = join(root, 'state.json') + const path = issuePath(125) + const issue = realIssueFile(125) + const mount = new FakeMountClient({ [path]: issue }) + const stateStore = new RejectFirstRunningLifecycleSaveStore({ batchSize: 2, watchStatePath }) + const clock = new ManualClock() + const notifications: string[] = [] + const factoryConfig = config({ + hooks: { + onTicketDispatch: { + notify: [{ surface: 'slack', channel: 'C123' }], + }, + }, + }) + const linear = { + async setState() {}, + async postComment() {}, + async createIssue() { + throw new Error('not used') + }, + async verify() { + return true + }, + } + const ticketDispatchDelivery = { + async slack({ text }: { text: string }) { + notifications.push(text) + }, + async telegram() {}, + } + const staleOwner = createFactory(factoryConfig, { + mount, + fleet: new RemoteLifecycleFleetClient(), + stateStore, + triage: new StaticTriage(), + linear, + ticketDispatchDelivery, + clock, + }) + let successor: ReturnType | undefined + + try { + const decision = await staleOwner.triageIssue(parseLinearIssue(path, issue)) + decision.implementers[0]!.principal = 'stale-triage-owner' + + await staleOwner.dispatch(decision) + + expect(stateStore.rejectedRunningSave).toBe(true) + expect(notifications).toEqual([]) + + await stateStore.releaseInFlight('factory-test', decision.issue.key) + clock.advance(5 * 60_000 + 1) + successor = createFactory(factoryConfig, { + mount, + fleet: new RemoteLifecycleFleetClient(), + stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), + triage: new StaticTriage(), + linear, + ticketDispatchDelivery, + clock, + }) + + await successor.dispatch(decision) + + expect(notifications).toHaveLength(1) + const payload = JSON.parse(notifications[0]!.slice(notifications[0]!.indexOf('\n') + 1)) + expect(payload).toMatchObject({ + agent: { name: 'ar-125-impl-pear' }, + sessionOwner: null, + }) + } finally { + await successor?.stop() + await staleOwner.stop() + await rm(root, { recursive: true, force: true }) + } + }) + it('derives implementer dispatch identities from repo labels instead of triage implementers', async () => { const routedIssue = realIssueFile(720, ready, { title: '[factory-e2e] Relayfile webhooks to cloud', diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 97d9081..bf51586 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -2736,11 +2736,11 @@ export class FactoryLoop implements Factory { dryRun, } record.result = result - await this.#saveDispatchLifecycle(record, 'running') + if (!await this.#saveDispatchLifecycle(record, 'running')) return result this.#increment('dispatched') this.#emit('dispatched', { issue: dispatchDecision.issue, result }) if (!dryRun && this.#config.hooks?.onTicketDispatch) { - await this.#notifyTicketDispatch(decision, liveIssue, record, result) + await this.#notifyTicketDispatch(dispatchDecision, liveIssue, record, result) } if (!dryRun) { await this.#ensureSlackDispatchThread(record, result) From 8488501126fce23a1eb308da24648b8a4825a804 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 13 Aug 2026 21:34:35 +0200 Subject: [PATCH 3/6] fix(factory): notify after durable dispatch recovery --- src/orchestrator/factory.test.ts | 157 ++++++++++++++++++++++++------- src/orchestrator/factory.ts | 56 +++++++++++ src/ports/state.ts | 9 ++ 3 files changed, 189 insertions(+), 33 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index b806041..7915e8f 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -9705,29 +9705,98 @@ describe('FactoryLoop', () => { expect(factory.status().counters.ticketDispatchHookNotifications).toBe(4) }) - it('fans out onTicketDispatch only after the running lifecycle owner is persisted', async () => { - class RejectFirstRunningLifecycleSaveStore extends FileStateStore { - rejectedRunningSave = false + it('delivers onTicketDispatch once when a fenced running save resumes durably', async () => { + vi.useFakeTimers() + + class RejectFirstRunningLifecycleSaveStore extends InMemoryStateStore { + runningSaveAttempts = 0 override async saveDispatchLifecycle( - ...args: Parameters + ...args: Parameters ): Promise { const lifecycle = args[5] - if (lifecycle.phase === 'running' && !this.rejectedRunningSave) { - this.rejectedRunningSave = true - return false + if (lifecycle.phase === 'running') { + this.runningSaveAttempts += 1 + if (this.runningSaveAttempts === 1) return false } return await super.saveDispatchLifecycle(...args) } } - const root = await mkdtemp(join(tmpdir(), 'factory-ticket-dispatch-owner-')) - const watchStatePath = join(root, 'state.json') const path = issuePath(125) const issue = realIssueFile(125) const mount = new FakeMountClient({ [path]: issue }) - const stateStore = new RejectFirstRunningLifecycleSaveStore({ batchSize: 2, watchStatePath }) - const clock = new ManualClock() + const stateStore = new RejectFirstRunningLifecycleSaveStore({ batchSize: 2 }) + const notifications: string[] = [] + const factory = createFactory(config({ + hooks: { + onTicketDispatch: { + notify: [{ surface: 'slack', channel: 'C123' }], + }, + }, + }), { + mount, + fleet: new RemoteLifecycleFleetClient(), + stateStore, + triage: new StaticTriage(), + linear: { + async setState() {}, + async postComment() {}, + async createIssue() { + throw new Error('not used') + }, + async verify() { + return true + }, + }, + ticketDispatchDelivery: { + async slack({ text }) { + notifications.push(text) + }, + async telegram() {}, + }, + }) + + try { + const decision = await factory.triageIssue(parseLinearIssue(path, issue)) + decision.implementers[0]!.principal = 'stale-triage-owner' + + await factory.dispatch(decision) + + expect(stateStore.runningSaveAttempts).toBe(1) + expect(notifications).toEqual([]) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))) + .resolves.toMatchObject({ phase: 'dispatching' }) + + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(() => expect(notifications).toHaveLength(1)) + + const payload = JSON.parse(notifications[0]!.slice(notifications[0]!.indexOf('\n') + 1)) + expect(payload).toMatchObject({ + agent: { name: 'ar-125-impl-pear' }, + sessionOwner: null, + }) + const lifecycle = await stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue)) + expect(lifecycle).toMatchObject({ + phase: 'running', + ticketDispatchNotification: { workUnitId: lifecycle?.runId }, + }) + + await vi.advanceTimersByTimeAsync(1_000) + expect(notifications).toHaveLength(1) + } finally { + await factory.stop() + vi.useRealTimers() + } + }) + + it('does not repeat an already claimed onTicketDispatch notification during durable resume', async () => { + vi.useFakeTimers() + const root = await mkdtemp(join(tmpdir(), 'factory-ticket-dispatch-idempotency-')) + const watchStatePath = join(root, 'state.json') + const path = issuePath(126) + const issue = realIssueFile(126) + const mount = new FakeMountClient({ [path]: issue }) const notifications: string[] = [] const factoryConfig = config({ hooks: { @@ -9752,50 +9821,72 @@ describe('FactoryLoop', () => { }, async telegram() {}, } - const staleOwner = createFactory(factoryConfig, { + const firstState = new FileStateStore({ batchSize: 2, watchStatePath }) + const first = createFactory(factoryConfig, { mount, fleet: new RemoteLifecycleFleetClient(), - stateStore, + stateStore: firstState, triage: new StaticTriage(), linear, ticketDispatchDelivery, - clock, }) - let successor: ReturnType | undefined + let resumed: ReturnType | undefined try { - const decision = await staleOwner.triageIssue(parseLinearIssue(path, issue)) - decision.implementers[0]!.principal = 'stale-triage-owner' + const decision = await first.triageIssue(parseLinearIssue(path, issue)) + await first.dispatch(decision) - await staleOwner.dispatch(decision) + expect(notifications).toHaveLength(1) + const key = issueKey(decision.issue) + const notified = await firstState.getDispatchLifecycle('factory-test', key) + expect(notified).toMatchObject({ + phase: 'running', + ticketDispatchNotification: { workUnitId: notified?.runId }, + }) - expect(stateStore.rejectedRunningSave).toBe(true) - expect(notifications).toEqual([]) + await first.stop() + const mutator = new FileStateStore({ batchSize: 2, watchStatePath }) + const claim = await mutator.claimDispatchLifecycle( + 'factory-test', + key, + notified!, + 'test-resume-owner', + Date.now(), + 5 * 60_000, + ) + expect(claim.acquired).toBe(true) + expect(await mutator.saveDispatchLifecycle( + 'factory-test', + key, + 'test-resume-owner', + claim.lease!.epoch, + Date.now(), + { ...claim.lifecycle, phase: 'dispatching' }, + )).toBe(true) + await mutator.releaseDispatchLifecycleLease('factory-test', key, 'test-resume-owner', claim.lease!.epoch) - await stateStore.releaseInFlight('factory-test', decision.issue.key) - clock.advance(5 * 60_000 + 1) - successor = createFactory(factoryConfig, { + resumed = createFactory(factoryConfig, { mount, fleet: new RemoteLifecycleFleetClient(), stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), triage: new StaticTriage(), linear, ticketDispatchDelivery, - clock, }) - - await successor.dispatch(decision) + await resumed.start({ mode: 'dispatch-owner' }) + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(async () => { + const lifecycle = await new FileStateStore({ batchSize: 2, watchStatePath }) + .getDispatchLifecycle('factory-test', key) + expect(lifecycle?.phase).toBe('running') + }) expect(notifications).toHaveLength(1) - const payload = JSON.parse(notifications[0]!.slice(notifications[0]!.indexOf('\n') + 1)) - expect(payload).toMatchObject({ - agent: { name: 'ar-125-impl-pear' }, - sessionOwner: null, - }) } finally { - await successor?.stop() - await staleOwner.stop() + await resumed?.stop() + await first.stop() await rm(root, { recursive: true, force: true }) + vi.useRealTimers() } }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index bf51586..9e623fb 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3712,6 +3712,9 @@ export class FactoryLoop implements Factory { releaseReason ?? previous?.releaseReason, cost, ) + if (previous?.ticketDispatchNotification) { + lifecycle.ticketDispatchNotification = structuredClone(previous.ticketDispatchNotification) + } for (const agent of lifecycle.agents) { const previousAgent = previous?.agents.find((candidate) => candidate.name === agent.name) if (previousAgent?.releasedAtMs !== undefined) agent.releasedAtMs = previousAgent.releasedAtMs @@ -4130,6 +4133,10 @@ export class FactoryLoop implements Factory { } if (!await this.#saveDispatchLifecycle(record, 'running')) return if (!record.dryRun) { + if (!liveIssue) { + throw new Error(`Unable to recover durable dispatch ${record.issue.key}: issue is no longer readable`) + } + await this.#notifyTicketDispatch(record.decision, liveIssue, record, record.result) await this.#ensureSlackDispatchThread(record, record.result) for (const tracked of record.agents.values()) { const owned = tracked.spec.ownedPullRequest @@ -11127,6 +11134,7 @@ export class FactoryLoop implements Factory { ?? result.agents.find((candidate) => candidate.role === 'workflow') ?? result.agents[0] if (!agent) return + if (!await this.#claimTicketDispatchNotification(record)) return const tracked = record.agents.get(agent.name) const payload: TicketDispatchNotificationPayload = { @@ -11189,6 +11197,54 @@ export class FactoryLoop implements Factory { } } + async #claimTicketDispatchNotification(record: InFlightIssue): Promise { + if (record.dryRun || !this.#usesDurableDispatchLifecycle()) return true + const key = issueKey(record.issue) + return this.#serializeDispatchLifecyclePersistence(key, async () => { + const epoch = this.#dispatchLifecycleEpochs.get(key) + if (epoch === undefined) { + this.#scheduleDispatchLifecycleRetry(record) + return false + } + const lifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, key) + if (!lifecycle) { + this.#scheduleDispatchLifecycleRetry(record) + return false + } + const workUnitId = lifecycle.runId + if (lifecycle.ticketDispatchNotification?.workUnitId === workUnitId) return false + + // Reserve the work unit before external side effects. Hook delivery is + // best-effort per target, so takeover must prefer at-most-once fan-out + // over replaying a notification whose provider acknowledgement was lost. + const claimedAtMs = this.#clock.now() + const claimedLifecycle: DispatchLifecycle = { + ...lifecycle, + ticketDispatchNotification: { workUnitId, claimedAtMs }, + updatedAtMs: claimedAtMs, + } + const saved = await this.#state.saveDispatchLifecycle( + this.#workspaceId, + key, + this.#dispatchLifecycleOwner, + epoch, + claimedAtMs, + claimedLifecycle, + ) + if (!saved) { + this.#dispatchLifecycleEpochs.delete(key) + this.#increment('dispatchLifecycleFencesRejected') + await this.#reportLifecycle(claimedLifecycle, 'factory.anomaly', { + level: 'error', + errorCode: 'fence_rejected', + }) + this.#scheduleDispatchLifecycleRetry(record) + return false + } + return true + }) + } + #error(error: unknown, issue?: IssueRef): void { this.#increment('errors') const details = describeError(error) diff --git a/src/ports/state.ts b/src/ports/state.ts index af644eb..7a0e22e 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -204,6 +204,15 @@ export type DispatchLifecycle = { releaseReason?: string /** Bounded token/USD aggregate updated with durable usage and finalized at terminal save. */ cost?: RunCostTotal + /** + * Durable at-most-once claim for onTicketDispatch fan-out. The lifecycle + * runId is the work-unit identity, so takeover can distinguish a recovered + * dispatch from a true reopen before issuing external side effects. + */ + ticketDispatchNotification?: { + workUnitId: string + claimedAtMs: number + } lease?: DispatchLifecycleLease updatedAtMs: number } From 0f5a0024af8c44e8998c3f8bb81a8652696aa7b2 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 13 Aug 2026 21:53:05 +0200 Subject: [PATCH 4/6] fix(factory): recover unclaimed dispatch notifications --- src/orchestrator/factory.test.ts | 118 +++++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 27 +++++-- 2 files changed, 140 insertions(+), 5 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 7915e8f..02be0db 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -9790,6 +9790,124 @@ describe('FactoryLoop', () => { } }) + it('delivers an unclaimed running onTicketDispatch notification after startup adoption', async () => { + vi.useFakeTimers() + const root = await mkdtemp(join(tmpdir(), 'factory-ticket-dispatch-crash-window-')) + const watchStatePath = join(root, 'state.json') + const path = issuePath(127) + const issue = realIssueFile(127) + const mount = new FakeMountClient({ [path]: issue }) + const notifications: string[] = [] + const factoryConfig = config({ + hooks: { + onTicketDispatch: { + notify: [{ surface: 'slack', channel: 'C123' }], + }, + }, + }) + const linear = { + async setState() {}, + async postComment() {}, + async createIssue() { + throw new Error('not used') + }, + async verify() { + return true + }, + } + const ticketDispatchDelivery = { + async slack({ text }: { text: string }) { + notifications.push(text) + }, + async telegram() {}, + } + const firstState = new FileStateStore({ batchSize: 2, watchStatePath }) + const first = createFactory(factoryConfig, { + mount, + fleet: new RemoteLifecycleFleetClient(), + stateStore: firstState, + triage: new StaticTriage(), + linear, + ticketDispatchDelivery, + }) + let restarted: ReturnType | undefined + + try { + const decision = await first.triageIssue(parseLinearIssue(path, issue)) + await first.dispatch(decision) + + expect(notifications).toHaveLength(1) + const key = issueKey(decision.issue) + const running = await firstState.getDispatchLifecycle('factory-test', key) + expect(running).toMatchObject({ + phase: 'running', + ticketDispatchNotification: { workUnitId: running?.runId }, + }) + + // Model a process crash after `running` was persisted but before the + // notification work unit was claimed. + await first.stop() + const mutator = new FileStateStore({ batchSize: 2, watchStatePath }) + const claim = await mutator.claimDispatchLifecycle( + 'factory-test', + key, + running!, + 'test-crash-window-owner', + Date.now(), + 5 * 60_000, + ) + expect(claim.acquired).toBe(true) + const { ticketDispatchNotification: _discardedClaim, ...unclaimedRunning } = claim.lifecycle + expect(await mutator.saveDispatchLifecycle( + 'factory-test', + key, + 'test-crash-window-owner', + claim.lease!.epoch, + Date.now(), + unclaimedRunning, + )).toBe(true) + await mutator.releaseDispatchLifecycleLease( + 'factory-test', + key, + 'test-crash-window-owner', + claim.lease!.epoch, + ) + notifications.length = 0 + + restarted = createFactory(factoryConfig, { + mount, + fleet: new RemoteLifecycleFleetClient(), + stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), + triage: new StaticTriage(), + linear, + ticketDispatchDelivery, + }) + await restarted.start({ mode: 'dispatch-owner' }) + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(() => expect(notifications).toHaveLength(1)) + + const payload = JSON.parse(notifications[0]!.slice(notifications[0]!.indexOf('\n') + 1)) + expect(payload).toMatchObject({ + agent: { name: 'ar-127-impl-pear' }, + sessionOwner: null, + }) + const adopted = await new FileStateStore({ batchSize: 2, watchStatePath }) + .getDispatchLifecycle('factory-test', key) + expect(adopted).toMatchObject({ + phase: 'running', + ticketDispatchNotification: { workUnitId: adopted?.runId }, + }) + + await vi.advanceTimersByTimeAsync(1_000) + expect(notifications).toHaveLength(1) + } finally { + await restarted?.stop() + await first.stop() + await rm(root, { recursive: true, force: true }) + vi.useRealTimers() + } + }) + it('does not repeat an already claimed onTicketDispatch notification during durable resume', async () => { vi.useFakeTimers() const root = await mkdtemp(join(tmpdir(), 'factory-ticket-dispatch-idempotency-')) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 9e623fb..578216d 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3040,7 +3040,12 @@ export class FactoryLoop implements Factory { const restored = claim.lifecycle.phase === 'queued' || claim.lifecycle.phase === 'releasing' ? durableRecord : batch.restore(durableRecord) - if (claim.lifecycle.phase !== 'running') this.#scheduleDispatchLifecycleRetry(restored) + if ( + claim.lifecycle.phase !== 'running' || + this.#ticketDispatchNotificationIsPending(claim.lifecycle) + ) { + this.#scheduleDispatchLifecycleRetry(restored) + } // Parking agents are cleanup-only. Hydrating them makes relay // reconciliation report their expected absence as an ordinary exit // before the durable parking driver can release/confirm them. @@ -3926,6 +3931,15 @@ export class FactoryLoop implements Factory { await this.#abandonDurableResume(record, 'source issue became terminal before lifecycle cleanup') return } + if (this.#ticketDispatchNotificationIsPending(lifecycle)) { + if (!liveIssue) { + throw new Error(`Unable to recover ticket-dispatch notification ${record.issue.key}: issue is not currently readable`) + } + if (!record.result) { + throw new Error(`Unable to recover ticket-dispatch notification ${record.issue.key}: dispatch result is missing`) + } + await this.#notifyTicketDispatch(record.decision, liveIssue, record, record.result) + } } if (acquiredNow && lifecycle.phase === 'running') { @@ -4132,10 +4146,7 @@ export class FactoryLoop implements Factory { record.result = { ...record.result, previews: recoveredPreviews } } if (!await this.#saveDispatchLifecycle(record, 'running')) return - if (!record.dryRun) { - if (!liveIssue) { - throw new Error(`Unable to recover durable dispatch ${record.issue.key}: issue is no longer readable`) - } + if (liveIssue) { await this.#notifyTicketDispatch(record.decision, liveIssue, record, record.result) await this.#ensureSlackDispatchThread(record, record.result) for (const tracked of record.agents.values()) { @@ -11245,6 +11256,12 @@ export class FactoryLoop implements Factory { }) } + #ticketDispatchNotificationIsPending(lifecycle: DispatchLifecycle): boolean { + return Boolean(this.#config.hooks?.onTicketDispatch) && + !lifecycle.dryRun && + lifecycle.ticketDispatchNotification?.workUnitId !== lifecycle.runId + } + #error(error: unknown, issue?: IssueRef): void { this.#increment('errors') const details = describeError(error) From d7f0481d0100196d4b38d086f80c069cbd394e90 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 13 Aug 2026 22:00:56 +0200 Subject: [PATCH 5/6] fix(factory): notify before startup reconciliation --- src/orchestrator/factory.test.ts | 15 +++++++++++---- src/orchestrator/factory.ts | 16 +++++++++++++--- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 02be0db..1873e51 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -9790,7 +9790,7 @@ describe('FactoryLoop', () => { } }) - it('delivers an unclaimed running onTicketDispatch notification after startup adoption', async () => { + it('delivers an unclaimed running onTicketDispatch notification before startup roster reconciliation', async () => { vi.useFakeTimers() const root = await mkdtemp(join(tmpdir(), 'factory-ticket-dispatch-crash-window-')) const watchStatePath = join(root, 'state.json') @@ -9798,6 +9798,13 @@ describe('FactoryLoop', () => { const issue = realIssueFile(127) const mount = new FakeMountClient({ [path]: issue }) const notifications: string[] = [] + const notificationsSeenAtReconcile: number[] = [] + class NotificationOrderingFleetClient extends RemoteLifecycleFleetClient { + override async reconcileTrackedAgents(): Promise { + notificationsSeenAtReconcile.push(notifications.length) + await super.reconcileTrackedAgents() + } + } const factoryConfig = config({ hooks: { onTicketDispatch: { @@ -9876,15 +9883,15 @@ describe('FactoryLoop', () => { restarted = createFactory(factoryConfig, { mount, - fleet: new RemoteLifecycleFleetClient(), + fleet: new NotificationOrderingFleetClient(), stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), triage: new StaticTriage(), linear, ticketDispatchDelivery, }) await restarted.start({ mode: 'dispatch-owner' }) - await vi.advanceTimersByTimeAsync(1_000) - await vi.waitFor(() => expect(notifications).toHaveLength(1)) + expect(notificationsSeenAtReconcile).toEqual([1]) + expect(notifications).toHaveLength(1) const payload = JSON.parse(notifications[0]!.slice(notifications[0]!.indexOf('\n') + 1)) expect(payload).toMatchObject({ diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 578216d..0765c37 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3007,12 +3007,13 @@ export class FactoryLoop implements Factory { this.#hydrateCostLedger(claim.lifecycle) if (claim.lifecycle.phase === 'waiting-for-human') continue const durableRecord = inFlightRecordFromLifecycle(claim.lifecycle) + let liveIssue: LinearIssue | undefined if ( !durableRecord.dryRun && claim.lifecycle.phase !== 'writeback-applied' && claim.lifecycle.phase !== 'releasing' ) { - const liveIssue = await this.#readIssue(durableRecord.issue.path) + liveIssue = await this.#readIssue(durableRecord.issue.path) // A babysat Linear issue already at Done may have merged while this // process was down. Let authoritative PR restoration drive the // normal `complete` path so merged work is not mislabeled abandoned. @@ -3041,11 +3042,20 @@ export class FactoryLoop implements Factory { ? durableRecord : batch.restore(durableRecord) if ( - claim.lifecycle.phase !== 'running' || + claim.lifecycle.phase === 'running' && this.#ticketDispatchNotificationIsPending(claim.lifecycle) ) { - this.#scheduleDispatchLifecycleRetry(restored) + if (!liveIssue) { + this.#dispatchLifecycleEpochs.delete(claim.key ?? key) + this.#scheduleDispatchLifecycleRetry(restored) + throw new Error(`Unable to recover ticket-dispatch notification ${restored.issue.key}: issue is not currently readable`) + } + if (!restored.result) { + throw new Error(`Unable to recover ticket-dispatch notification ${restored.issue.key}: dispatch result is missing`) + } + await this.#notifyTicketDispatch(restored.decision, liveIssue, restored, restored.result) } + if (claim.lifecycle.phase !== 'running') this.#scheduleDispatchLifecycleRetry(restored) // Parking agents are cleanup-only. Hydrating them makes relay // reconciliation report their expected absence as an ordinary exit // before the durable parking driver can release/confirm them. From 0e9892f1ecef0292f7b593670c31c39011b2638d Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 13 Aug 2026 22:17:22 +0200 Subject: [PATCH 6/6] docs(factory): track dispatch notification crash window --- src/orchestrator/factory.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 0765c37..13a25b0 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -11238,6 +11238,7 @@ export class FactoryLoop implements Factory { // Reserve the work unit before external side effects. Hook delivery is // best-effort per target, so takeover must prefer at-most-once fan-out // over replaying a notification whose provider acknowledgement was lost. + // Follow-up for provider-idempotent retries: https://github.com/AgentWorkforce/factory/issues/239 const claimedAtMs = this.#clock.now() const claimedLifecycle: DispatchLifecycle = { ...lifecycle,