diff --git a/src/index.ts b/src/index.ts index e8da791..ea863b5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -185,6 +185,7 @@ export type { } from './safety/factory-scope' export { canonicalMountPaths, + createResourceSubscriptionsHttpClient, createWorkspaceScopedEventClient, deliveryTargetsFor, eventPathGlobsForIntegration, @@ -200,6 +201,9 @@ export { linearScopePredicates, normalizeChangePath, relayfileSdkPathFiltersFor, + ResourceSubscriptionsHttpError, + ResourceSubscriptionsUnavailableError, + isResourceSubscriptionsUnavailable, slackListenDms, subscriptionSpecsFor, } from './subscriptions' @@ -221,6 +225,12 @@ export type { WorkspaceScopedEventClientOptions, WorkspaceScopedSubscribeOptions, ChangeEvent as SubscriptionChangeEvent, + AcceptedResourceDelivery, + ResourceDeliveryClaim, + ResourceSubscription, + ResourceSubscriptionInput, + ResourceSubscriptionsClient, + ResourceSubscriptionsHttpClientOptions, } from './subscriptions' export type { Capability, diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index d0ae44f..b680893 100644 --- a/src/mount/relayfile-cloud-mount-client.test.ts +++ b/src/mount/relayfile-cloud-mount-client.test.ts @@ -268,6 +268,130 @@ describe('RelayfileCloudMountClient', () => { expect(cloudSessionProvider).toHaveBeenCalledTimes(2) }) + it('wires the durable resource-subscription API from the configured Relayfile URL and workspace bearer', async () => { + const fake = new FakeRelayFileClient() + const requests: Array<{ url: string; init?: RequestInit }> = [] + const resourceSubscriptionFetch: typeof fetch = async (input, init) => { + requests.push({ url: String(input), init }) + if (String(input).endsWith('/subscriptions') && init?.method === 'POST') { + return new Response(JSON.stringify({ + id: 'sub-1', + ownerId: 'configured-factory-agent', + subscriberId: 'factory-babysitter:uuid-1', + provider: 'github', + resourceRef: '/github/repos/AgentWorkforce__pear/pulls/by-id/1.json', + eventTypes: ['pull_request_review_comment.created'], + terminalEventTypes: ['pull_request.closed'], + intent: null, + status: 'active', + expiresAt: '2026-12-31T00:00:00.000Z', + }), { status: 200, headers: { 'content-type': 'application/json' } }) + } + if (String(input).endsWith('/subscriptions/deliveries/claim') && init?.method === 'POST') { + return new Response(JSON.stringify({ deliveries: [{ + id: 'delivery-1', + claimToken: 'claim-token-1', + subscriptionId: 'sub-1', + ownerId: 'configured-factory-agent', + subscriberId: 'factory-babysitter:uuid-1', + provider: 'github', + resourceRef: '/github/repos/AgentWorkforce__pear/pulls/by-id/1.json', + event: { type: 'pull_request.closed' }, + terminal: true, + }] }), { status: 200, headers: { 'content-type': 'application/json' } }) + } + if (String(input).includes('/accept') && init?.method === 'POST') { + if (JSON.parse(String(init.body)).claimToken !== 'claim-token-1') { + return new Response(JSON.stringify({ error: 'delivery_claim_mismatch' }), { status: 409, headers: { 'content-type': 'application/json' } }) + } + return new Response(JSON.stringify({ deliveryId: 'delivery-1', subscriptionId: 'sub-1', terminal: true }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }) + } + return new Response(null, { status: 204 }) + } + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client: fake, + resourceSubscriptionFetch, + }) + + const client = mount.resourceSubscriptions! + await expect(client.createOrRenew('rw_test', { + provider: 'github', + resourceRef: '/github/repos/AgentWorkforce__pear/pulls/by-id/1.json', + eventTypes: ['pull_request_review_comment.created'], + terminalEventTypes: ['pull_request.closed'], + subscriberId: 'factory-babysitter:uuid-1', + ttlSeconds: 3600, + })).resolves.toMatchObject({ subscriptionId: 'sub-1', ownerId: 'configured-factory-agent' }) + await expect(client.claimDeliveryClaims('rw_test')).resolves.toEqual([expect.objectContaining({ + deliveryId: 'delivery-1', + claimToken: 'claim-token-1', + terminal: true, + })]) + await expect(client.acceptDelivery('rw_test', { deliveryId: 'delivery-1', claimToken: 'wrong-token' })) + .rejects.toMatchObject({ status: 409 }) + await expect(client.acceptDelivery('rw_test', { deliveryId: 'delivery-1', claimToken: 'claim-token-1' })) + .resolves.toEqual({ deliveryId: 'delivery-1', subscriptionId: 'sub-1', terminal: true }) + await client.cancel('rw_test', { subscriptionId: 'sub-1' }) + + expect(requests.map(({ url, init }) => ({ url, method: init?.method }))).toEqual([ + { url: 'https://relayfile.invalid/v1/workspaces/rw_test/subscriptions', method: 'POST' }, + { url: 'https://relayfile.invalid/v1/workspaces/rw_test/subscriptions/deliveries/claim', method: 'POST' }, + { url: 'https://relayfile.invalid/v1/workspaces/rw_test/subscriptions/deliveries/delivery-1/accept', method: 'POST' }, + { url: 'https://relayfile.invalid/v1/workspaces/rw_test/subscriptions/deliveries/delivery-1/accept', method: 'POST' }, + { url: 'https://relayfile.invalid/v1/workspaces/rw_test/subscriptions/sub-1', method: 'DELETE' }, + ]) + expect(requests[0]?.init?.headers).toMatchObject({ authorization: 'Bearer relayfile-token' }) + expect(JSON.parse(String(requests[0]?.init?.body))).toEqual({ + provider: 'github', + resourceRef: '/github/repos/AgentWorkforce__pear/pulls/by-id/1.json', + eventTypes: ['pull_request_review_comment.created'], + terminalEventTypes: ['pull_request.closed'], + subscriberId: 'factory-babysitter:uuid-1', + ttlSeconds: 3600, + }) + expect(JSON.parse(String(requests[2]?.init?.body))).toEqual({ claimToken: 'wrong-token' }) + expect(JSON.parse(String(requests[3]?.init?.body))).toEqual({ claimToken: 'claim-token-1' }) + }) + + it('fails closed for malformed durable claim envelopes and stalled Relayfile requests', async () => { + const fake = new FakeRelayFileClient() + const malformed = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client: fake, + baseUrl: 'https://relayfile.invalid', + tokenProvider: () => 'relayfile-token', + resourceSubscriptionFetch: async () => new Response(JSON.stringify({ notDeliveries: [] }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + }) + await expect(malformed.resourceSubscriptions!.claimDeliveryClaims('rw_test')) + .rejects.toMatchObject({ status: 502 }) + + const stalled = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client: fake, + baseUrl: 'https://relayfile.invalid', + tokenProvider: () => 'relayfile-token', + resourceSubscriptionRequestTimeoutMs: 10, + resourceSubscriptionFetch: async (_input, init) => await new Promise((_resolve, reject) => { + const signal = init?.signal + if (!signal) throw new Error('subscription request was not abortable') + if (signal.aborted) { + reject(signal.reason) + return + } + signal.addEventListener('abort', () => reject(signal.reason), { once: true }) + }), + }) + await expect(stalled.resourceSubscriptions!.claimDeliveryClaims('rw_test')) + .rejects.toMatchObject({ status: 504 }) + }) + it('coalesces concurrent shared session resolutions for relayfile token refresh', async () => { const setup = { joinWorkspace: vi.fn(async () => ({ diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts index ca0effe..d7e8e54 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -24,11 +24,13 @@ import { import type { EventPage, GithubConnectionWrite, MountClient, ProviderSyncStatus, SubscribeOptions } from '../ports' import { + createResourceSubscriptionsHttpClient, createWorkspaceScopedEventClient, type RelayfileEventClient, type TokenProvider, type WorkspaceEventClientSource, } from '../subscriptions' +import type { ResourceSubscriptionsClient } from '../subscriptions' import { RelayfileGithubConnectionWrite } from './relayfile-github-connection-write' const DEFAULT_WORKSPACE_ID = 'rw_7ccfea89' @@ -113,6 +115,11 @@ export interface RelayfileCloudMountClientConfig { tokenProvider?: TokenProvider baseUrl?: string eventClient?: RelayfileEventClient + /** Override the standard Relayfile Cloud durable-subscription HTTP client. */ + resourceSubscriptions?: ResourceSubscriptionsClient + resourceSubscriptionFetch?: typeof fetch + resourceSubscriptionRequestTimeoutMs?: number + resourceSubscriptionSignal?: AbortSignal isAllowedDraft?: (path: string, content: unknown, opts?: { guarded?: boolean }) => boolean | Promise isAllowedDelete?: (path: string, currentContent: unknown) => boolean | Promise } @@ -136,6 +143,7 @@ export class RelayfileCloudMountClient implements MountClient { readonly workspaceId: string readonly writebackTransport = 'relayfile-cloud' readonly githubWrite: GithubConnectionWrite + readonly resourceSubscriptions?: ResourceSubscriptionsClient readonly #client: RelayFileClientLike readonly #tokenProvider: TokenProvider @@ -155,6 +163,15 @@ export class RelayfileCloudMountClient implements MountClient { this.#tokenProvider = config.tokenProvider ?? (() => this.#client.getToken?.()) this.#baseUrl = config.baseUrl ?? this.#client.getBaseUrl?.() this.#eventClient = config.eventClient + this.resourceSubscriptions = config.resourceSubscriptions ?? (this.#baseUrl + ? createResourceSubscriptionsHttpClient({ + baseUrl: this.#baseUrl, + tokenProvider: this.#tokenProvider, + fetch: config.resourceSubscriptionFetch, + requestTimeoutMs: config.resourceSubscriptionRequestTimeoutMs, + signal: config.resourceSubscriptionSignal, + }) + : undefined) this.#isAllowedDraft = config.isAllowedDraft this.#isAllowedDelete = config.isAllowedDelete this.githubWrite = new RelayfileGithubConnectionWrite({ mount: this }) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index d5db5d9..4fdd0c0 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -33,6 +33,14 @@ import { InMemoryStateStore } from '../state/in-memory-state-store' import { FileStateStore } from '../state/file-state-store' import { LIVE_GITHUB_ISSUE_GLOB, githubIssuePathParts, keyFromPath } from './factory' import { globMatchesPath } from '../subscriptions/globs' +import { + ResourceSubscriptionsUnavailableError, + type AcceptedResourceDelivery, + type ResourceDeliveryClaim, + type ResourceSubscription, + type ResourceSubscriptionInput, + type ResourceSubscriptionsClient, +} from '../subscriptions' import { InternalFleetClient, type HarnessDriverClientLike } from '../fleet/internal-fleet-client' const ready = 'b9bec744-b60c-4745-8022-d90d6ab59ae3' @@ -11999,6 +12007,100 @@ describe('FactoryLoop PR babysitter', () => { }) } + class FakeResourceSubscriptions implements ResourceSubscriptionsClient { + readonly createCalls: Array<{ workspaceId: string; input: ResourceSubscriptionInput }> = [] + readonly claimCalls: Array<{ workspaceId: string; limit?: number }> = [] + readonly accepted: Array<{ workspaceId: string; deliveryId: string; claimToken: string }> = [] + readonly cancelled: Array<{ workspaceId: string; subscriptionId: string }> = [] + readonly records: ResourceSubscription[] = [] + claims: ResourceDeliveryClaim[] = [] + readonly leasedClaims = new Map() + readonly acceptedClaims = new Map() + ownerId = 'configured-factory-agent' + unavailable = false + createFailure?: Error + claimFailure?: Error + onAccept?: (deliveryId: string) => Promise | void + + async createOrRenew(workspaceId: string, input: ResourceSubscriptionInput): Promise { + if (this.unavailable) throw new ResourceSubscriptionsUnavailableError() + if (this.createFailure) throw this.createFailure + this.createCalls.push({ workspaceId, input: structuredClone(input) }) + const identity = JSON.stringify([input.subscriberId, input.resourceRef, [...input.eventTypes].sort()]) + let record = this.records.find((candidate) => JSON.stringify([ + candidate.subscriberId, + candidate.resourceRef, + [...candidate.eventTypes].sort(), + ]) === identity) + if (!record) { + record = { + ...structuredClone(input), + ownerId: this.ownerId, + eventTypes: [...input.eventTypes].sort(), + subscriptionId: `sub-${this.records.length + 1}`, + expiresAt: '2026-12-31T00:00:00.000Z', + } + this.records.push(record) + } + return structuredClone(record) + } + + claimFor( + subscription: ResourceSubscription, + eventType: string, + deliveryId: string, + ): ResourceDeliveryClaim { + return { + deliveryId, + claimToken: `claim-token-${deliveryId}`, + subscriptionId: subscription.subscriptionId, + provider: subscription.provider, + resourceRef: subscription.resourceRef, + eventType, + ownerId: subscription.ownerId, + subscriberId: subscription.subscriberId, + terminal: subscription.terminalEventTypes?.includes(eventType) === true, + } + } + + async claimDeliveryClaims(workspaceId: string, input?: { limit?: number }): Promise { + if (this.unavailable) throw new ResourceSubscriptionsUnavailableError() + if (this.claimFailure) throw this.claimFailure + this.claimCalls.push({ workspaceId, ...input }) + // Model Relayfile's atomic claim lease: overlapping runtime polls can + // observe a delivery at most once, independent of client-side locking. + const claimed = this.claims.splice(0, input?.limit ?? this.claims.length) + for (const claim of claimed) this.leasedClaims.set(claim.deliveryId, claim) + return claimed.map((claim) => structuredClone(claim)) + } + + async acceptDelivery( + workspaceId: string, + input: { deliveryId: string; claimToken: string }, + ): Promise { + if (this.unavailable) throw new ResourceSubscriptionsUnavailableError() + const accepted = this.acceptedClaims.get(input.deliveryId) + if (accepted) { + if (accepted.claimToken !== input.claimToken) throw new Error(`claim ${input.deliveryId} has an invalid token`) + this.accepted.push({ workspaceId, ...input }) + return structuredClone(accepted.receipt) + } + const claim = this.leasedClaims.get(input.deliveryId) + if (!claim) throw new Error(`claim ${input.deliveryId} is unavailable`) + if (claim.claimToken !== input.claimToken) throw new Error(`claim ${input.deliveryId} has an invalid token`) + this.accepted.push({ workspaceId, ...input }) + const receipt = { deliveryId: claim.deliveryId, subscriptionId: claim.subscriptionId, terminal: claim.terminal } + this.acceptedClaims.set(input.deliveryId, { claimToken: input.claimToken, receipt }) + await this.onAccept?.(input.deliveryId) + this.leasedClaims.delete(input.deliveryId) + return structuredClone(receipt) + } + + async cancel(workspaceId: string, input: { subscriptionId: string }): Promise { + this.cancelled.push({ workspaceId, ...input }) + } + } + it('recovers a remote babysitter across the spawn-ack crash gap without replaying the original team tasks', async () => { class BabysitterAckGapFleet extends RemoteLifecycleFleetClient { crashed = false @@ -12377,6 +12479,387 @@ describe('FactoryLoop PR babysitter', () => { expect(fleet.spawns.filter((s) => s.name === 'ar-403-babysit')).toHaveLength(1) }) + it('uses Relayfile by-id delivery claims for renamed PR activity, renews idempotently, and retires terminal claims after the local queue is durable', async () => { + const number = 604 + const issue = realIssueFile(number, ready, { title: 'Real durable resource babysitter' }) + const prPath = `/github/repos/AgentWorkforce/pear/pulls/${number}/metadata.json` + const renamedCommentPath = `/github/repos/AgentWorkforce/pear/pulls/${number}__renamed-after-review/comments/6041.json` + const mount = new FakeMountClient({ [issuePath(number)]: issue }) + const subscriptions = new FakeResourceSubscriptions() + mount.resourceSubscriptions = subscriptions + const fleet = new FakeFleetClient() + const state = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(babysitterConfig(), { mount, fleet, triage: new StaticTriage(), stateStore: state }) + + try { + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) + mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.emit(changeEvent(prPath, 'durable-pr-open')) + await flush() + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain(`ar-${number}-babysit`)) + await vi.waitFor(() => expect(subscriptions.records).toHaveLength(1)) + + const [[sessionKey, initialSession]] = await state.listBabysitterSessions('factory-test') + expect(initialSession.resourceSubscription).toMatchObject({ + resourceRef: `/github/repos/AgentWorkforce__pear/pulls/by-id/${number}.json`, + ownerId: 'configured-factory-agent', + subscriberId: `factory-babysitter:uuid-${number}`, + }) + const subscription = initialSession.resourceSubscription! + expect(subscriptions.createCalls[0]?.input).toMatchObject({ + eventTypes: expect.arrayContaining(['pull_request_review_comment.created']), + terminalEventTypes: ['pull_request.closed'], + }) + expect(subscriptions.createCalls[0]?.input.eventTypes).not.toContain('pull_request.closed') + subscriptions.claims = [ + { + deliveryId: 'delivery-unowned', + claimToken: 'claim-token-unowned', + subscriptionId: 'sub-unowned', + resourceRef: '/github/repos/AgentWorkforce__pear/pulls/by-id/999.json', + eventType: 'pull_request_review_comment.created', + ownerId: 'configured-factory-agent', + subscriberId: 'factory-babysitter:unowned', + provider: 'github', + terminal: false, + }, + subscriptions.claimFor(subscription, 'pull_request_review_comment.created', 'delivery-renamed'), + ] + let sessionAtAcceptance: Awaited> | undefined + subscriptions.onAccept = async (deliveryId) => { + if (deliveryId === 'delivery-renamed') sessionAtAcceptance = await state.listBabysitterSessions('factory-test') + } + + // The raw path carries a new title slug. Factory does not transform it or + // scan its local owners: the service's stable by-id delivery claim picks + // the one session that must wake. + // Two raw events may race in the runtime; the service lease gives the + // babysitter one delivery and therefore one injected wake. + mount.emit(changeEvent(renamedCommentPath, 'renamed-pr-comment-a')) + mount.emit(changeEvent(renamedCommentPath, 'renamed-pr-comment-b')) + await vi.waitFor(() => expect( + fleet.messages.filter((message) => message.text.startsWith(' message.text.startsWith(' entry.deliveryId)).toEqual(['delivery-renamed']) + expect(sessionAtAcceptance?.find(([key]) => key === sessionKey)?.[1]).toMatchObject({ + pendingKinds: ['pull-request-state'], + pendingDeliveryClaims: [{ deliveryId: 'delivery-renamed', claimToken: 'claim-token-delivery-renamed' }], + }) + expect(factory.status().counters.babysitterResourceDeliveriesIgnoredUnowned).toBe(1) + + // A duplicate PR observation renews the same server identity rather than + // making another subscription record. There is no claim, so no wake. + const renamedPrPath = `/github/repos/AgentWorkforce/pear/pulls/${number}__renamed-after-review/metadata.json` + mount.files.set(renamedPrPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.emit(changeEvent(renamedPrPath, 'durable-pr-repeat')) + await vi.waitFor(() => expect(subscriptions.createCalls).toHaveLength(2)) + expect(subscriptions.records).toHaveLength(1) + expect(fleet.messages.filter((message) => message.text.startsWith(' key === sessionKey)![1] + subscriptions.claims = [subscriptions.claimFor( + subscriptions.records[0]!, + 'pull_request.closed', + 'delivery-terminal', + )] + let terminalSessionAtAcceptance: Awaited> | undefined + subscriptions.onAccept = async (deliveryId) => { + if (deliveryId === 'delivery-terminal') terminalSessionAtAcceptance = await state.listBabysitterSessions('factory-test') + } + mount.emit(changeEvent(renamedCommentPath, 'renamed-pr-terminal')) + await vi.waitFor(() => expect(subscriptions.accepted.map((entry) => entry.deliveryId)) + .toEqual(['delivery-renamed', 'delivery-terminal'])) + expect(terminalSessionAtAcceptance?.find(([key]) => key === sessionKey)?.[1].resourceSubscription).toMatchObject({ + terminal: true, + }) + expect((await state.listBabysitterSessions('factory-test')).find(([key]) => key === sessionKey)?.[1].resourceSubscription) + .toMatchObject({ terminal: true }) + expect(factory.status().counters.babysitterResourceSubscriptionsRetiredTerminal).toBe(1) + await vi.waitFor(() => expect( + fleet.messages.filter((message) => message.text.startsWith(' message.text.startsWith(' setTimeout(resolve, 900)) + expect(subscriptions.createCalls).toHaveLength(2) + expect(fleet.messages.filter((message) => message.text.startsWith(' setTimeout(resolve, 900)) + expect(fleet.messages.filter((message) => message.text.startsWith(' { + const number = 605 + const issue = realIssueFile(number, ready, { title: 'Real durable subscription fallback' }) + const prPath = `/github/repos/AgentWorkforce/pear/pulls/${number}/metadata.json` + const commentPath = `/github/repos/AgentWorkforce/pear/pulls/${number}/comments/6051.json` + const mount = new FakeMountClient({ [issuePath(number)]: issue }) + const subscriptions = new FakeResourceSubscriptions() + subscriptions.unavailable = true + mount.resourceSubscriptions = subscriptions + const fleet = new FakeFleetClient() + const factory = createFactory(babysitterConfig(), { mount, fleet, triage: new StaticTriage() }) + + try { + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) + mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.emit(changeEvent(prPath, 'fallback-pr-open')) + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain(`ar-${number}-babysit`)) + + mount.emit(changeEvent(commentPath, 'fallback-comment')) + await vi.waitFor(() => expect( + fleet.messages.filter((message) => message.text.startsWith(' message.to), + ).toEqual([`ar-${number}-babysit`])) + expect(factory.status().counters.babysitterResourceSubscriptionUnavailable).toBeGreaterThan(0) + } finally { + await factory.stop() + } + }) + + it('renews quiet durable subscriptions and retries a transient initial create failure', async () => { + vi.useFakeTimers() + const number = 610 + const issue = realIssueFile(number, ready, { title: 'Real durable renewal' }) + const mount = new FakeMountClient({ [issuePath(number)]: issue }) + seedPrMeta(mount, 'AgentWorkforce/pear', number, { state: 'open', draft: false }) + const subscriptions = new FakeResourceSubscriptions() + subscriptions.createFailure = Object.assign(new Error('transient create failure'), { status: 503 }) + mount.resourceSubscriptions = subscriptions + const fleet = new FakeFleetClient() + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: number }), + }) + + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) + fleet.emitAgentExit(`ar-${number}-impl-pear`, 'worker_exited') + await vi.advanceTimersByTimeAsync(0) + expect(subscriptions.records).toEqual([]) + expect(factory.status().counters.babysitterResourceSubscriptionRenewFailures).toBe(1) + + subscriptions.createFailure = undefined + await vi.advanceTimersByTimeAsync(5_000) + expect(subscriptions.records).toHaveLength(1) + + await vi.advanceTimersByTimeAsync(30 * 60_000) + expect(subscriptions.createCalls).toHaveLength(2) + } finally { + await factory.stop() + vi.useRealTimers() + } + }) + + it('accepts a provider terminal claim before a closed PR tears down its local subscription owner', async () => { + const number = 607 + const issue = realIssueFile(number, ready, { title: 'Real durable terminal close' }) + const prPath = `/github/repos/AgentWorkforce/pear/pulls/${number}/metadata.json` + const mount = new FakeMountClient({ [issuePath(number)]: issue }) + const subscriptions = new FakeResourceSubscriptions() + mount.resourceSubscriptions = subscriptions + const fleet = new FakeFleetClient() + const state = new InMemoryStateStore({ batchSize: 2 }) + seedPrMeta(mount, 'AgentWorkforce/pear', number, { state: 'open', draft: false }) + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + stateStore: state, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: number }), + }) + + try { + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) + fleet.emitAgentExit(`ar-${number}-impl-pear`, 'worker_exited') + await vi.waitFor(() => expect(subscriptions.records).toHaveLength(1)) + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain(`ar-${number}-babysit`)) + + const subscription = subscriptions.records[0]! + subscriptions.claims = [subscriptions.claimFor(subscription, 'pull_request.closed', 'delivery-close')] + mount.files.set(prPath, { content: { number, state: 'closed', draft: false, head_ref: `ar-${number}-fix` } }) + mount.emit(changeEvent(prPath, 'terminal-close')) + + await vi.waitFor(() => expect(subscriptions.accepted.map((entry) => entry.deliveryId)).toEqual(['delivery-close'])) + expect(subscriptions.cancelled).toEqual([{ + workspaceId: 'factory-test', + subscriptionId: subscription.subscriptionId, + }]) + await expect(state.listBabysitterSessions('factory-test')).resolves.toEqual([]) + } finally { + await factory.stop() + } + }) + + it('reuses the persisted claim token after a crash following terminal acceptance', async () => { + class FailingTerminalAcceptancePersistStore extends InMemoryStateStore { + failFinalTerminalPersist = false + + override async setBabysitterSession(...args: Parameters): Promise { + const [, , session] = args + if ( + this.failFinalTerminalPersist && + session.resourceSubscription?.terminal && + !session.pendingDeliveryClaims?.length + ) { + this.failFinalTerminalPersist = false + throw new Error('process stopped after Relayfile accepted the terminal delivery') + } + await super.setBabysitterSession(...args) + } + } + + const number = 609 + const issue = realIssueFile(number, ready, { title: 'Real durable acceptance restart' }) + const prPath = `/github/repos/AgentWorkforce/pear/pulls/${number}/metadata.json` + const mount = new FakeMountClient({ [issuePath(number)]: issue }) + const subscriptions = new FakeResourceSubscriptions() + mount.resourceSubscriptions = subscriptions + const state = new FailingTerminalAcceptancePersistStore({ batchSize: 2 }) + const first = createFactory(babysitterConfig(), { + mount, + fleet: new FakeFleetClient(), + triage: new StaticTriage(), + stateStore: state, + }) + let restarted: ReturnType | undefined + + try { + await first.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(number), issue))) + mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.emit(changeEvent(prPath, 'acceptance-restart-open')) + await flush() + await vi.waitFor(() => expect(subscriptions.records).toHaveLength(1)) + + const subscription = subscriptions.records[0]! + subscriptions.claims = [subscriptions.claimFor(subscription, 'pull_request.closed', 'delivery-accepted-before-crash')] + state.failFinalTerminalPersist = true + mount.emit(changeEvent(`/github/repos/AgentWorkforce/pear/pulls/${number}/comments/6091.json`, 'acceptance-restart-terminal')) + + await vi.waitFor(() => expect(subscriptions.accepted.map((entry) => entry.deliveryId)) + .toEqual(['delivery-accepted-before-crash'])) + await vi.waitFor(async () => expect(await state.listBabysitterSessions('factory-test')).toEqual([ + [expect.any(String), expect.objectContaining({ + resourceSubscription: expect.objectContaining({ terminal: true }), + pendingDeliveryClaims: [{ + deliveryId: 'delivery-accepted-before-crash', + claimToken: 'claim-token-delivery-accepted-before-crash', + }], + })], + ])) + await first.stop() + + restarted = createFactory(babysitterConfig(), { + mount, + fleet: new FakeFleetClient(), + triage: new StaticTriage(), + stateStore: state, + }) + await restarted.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + + await vi.waitFor(() => expect(subscriptions.accepted.map((entry) => entry.deliveryId)) + .toEqual(['delivery-accepted-before-crash', 'delivery-accepted-before-crash'])) + await vi.waitFor(async () => { + const [[, restored]] = await state.listBabysitterSessions('factory-test') + expect(restored?.resourceSubscription).toMatchObject({ terminal: true }) + expect(restored?.pendingDeliveryClaims).toBeUndefined() + }) + } finally { + await first.stop() + await restarted?.stop() + } + }) + + it('cancels the durable subscription before completion clears the babysitter owner', async () => { + const number = 608 + const issue = realIssueFile(number, ready, { title: 'Real durable completion cancellation' }) + const mount = new FakeMountClient({ [issuePath(number)]: issue }) + seedPrMeta(mount, 'AgentWorkforce/pear', number, { state: 'open', draft: false }) + const subscriptions = new FakeResourceSubscriptions() + mount.resourceSubscriptions = subscriptions + const fleet = new FakeFleetClient() + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: number }), + }) + + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) + fleet.emitAgentExit(`ar-${number}-impl-pear`, 'worker_exited') + await vi.waitFor(() => expect(subscriptions.records).toHaveLength(1)) + const subscription = subscriptions.records[0]! + + fleet.emitAgentMessage({ from: `ar-${number}-babysit`, target: 'factory', body: `[factory-pr-ready] AR-${number}` }) + await vi.waitFor(() => expect(subscriptions.cancelled).toEqual([{ + workspaceId: 'factory-test', + subscriptionId: subscription.subscriptionId, + }])) + } finally { + await factory.stop() + } + }) + + it('does not double-route locally when an established durable delivery lookup is transiently unavailable', async () => { + const number = 606 + const issue = realIssueFile(number, ready, { title: 'Real durable transient claim retry' }) + const prPath = `/github/repos/AgentWorkforce/pear/pulls/${number}/metadata.json` + const commentPath = `/github/repos/AgentWorkforce/pear/pulls/${number}__renamed/comments/6061.json` + const mount = new FakeMountClient({ [issuePath(number)]: issue }) + const subscriptions = new FakeResourceSubscriptions() + mount.resourceSubscriptions = subscriptions + const fleet = new FakeFleetClient() + const factory = createFactory(babysitterConfig(), { mount, fleet, triage: new StaticTriage() }) + + try { + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) + mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.emit(changeEvent(prPath, 'transient-pr-open')) + await vi.waitFor(() => expect(subscriptions.records).toHaveLength(1)) + const subscription = subscriptions.records[0]! + + subscriptions.claimFailure = Object.assign(new Error('temporarily unavailable'), { status: 503 }) + mount.emit(changeEvent(commentPath, 'transient-claim-event')) + await new Promise((resolve) => setTimeout(resolve, 900)) + expect(fleet.messages.filter((message) => message.text.startsWith(' expect( + fleet.messages.filter((message) => message.text.startsWith(' message.to), + ).toEqual([`ar-${number}-babysit`])) + expect(subscriptions.accepted.map((entry) => entry.deliveryId)).toEqual(['delivery-after-recovery']) + } finally { + await factory.stop() + } + }) + it('transitions to Human Review on the babysitter ready signal (open PR meta, no gh call)', async () => { const issue = realIssueFile(402, ready, { title: 'Real babysitter ready' }) const mount = new FakeMountClient({ [issuePath(402)]: issue }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 8c07b34..cd599c3 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -45,6 +45,7 @@ import { import { renderAgentTask } from '../dispatch/templates' import { HeuristicTriage, TieredTriage, babysitterSpec, isShapeLabel, scopeFromLabels } from '../triage' import { agentNameForRole, sanitizeAgentSlug } from '../triage/agent-names' +import { isResourceSubscriptionsUnavailable, type ResourceSubscription } from '../subscriptions' import type { DispatchResult, Factory, @@ -91,7 +92,19 @@ type BabysitterWakeKind = | 'checks-failed' | 'merge-conflict' | 'base-diverged' -type BabysitterPrRef = { repo: string; prNumber: number; path?: string; agentName: string } +type BabysitterResourceSubscription = Pick< + ResourceSubscription, + 'subscriptionId' | 'provider' | 'resourceRef' | 'subscriberId' | 'ownerId' | 'expiresAt' +> & { terminal?: boolean } +type BabysitterPendingDeliveryClaim = { deliveryId: string; claimToken: string } +type BabysitterPrRef = { + repo: string + prNumber: number + path?: string + agentName: string + resourceSubscription?: BabysitterResourceSubscription + pendingDeliveryClaims?: BabysitterPendingDeliveryClaim[] +} type BabysitterWakeState = { issue: IssueRef repo: string @@ -214,6 +227,24 @@ const INJECTION_RETRY_ATTEMPT_TIMEOUT_MS = 15_000 const INJECTION_MAX_ATTEMPTS = 6 const BABYSITTER_EVENT_COALESCE_MS = 750 const BABYSITTER_EVENT_RETRY_MS = 1_000 +const BABYSITTER_SUBSCRIPTION_TTL_SECONDS = 60 * 60 +// Relayfile receives provider-native GitHub events, not the materialized file +// changes that the legacy local router consumed. `closed` is separately +// indexed as a terminal event below, so terminal delivery does not need a +// broad normal-event subscription. +const BABYSITTER_SUBSCRIPTION_EVENT_TYPES = [ + 'pull_request.opened', + 'pull_request.reopened', + 'pull_request.synchronize', + 'pull_request.ready_for_review', + 'pull_request_review.submitted', + 'pull_request_review_comment.created', + 'issue_comment.created', + 'check_run.completed', +] +const BABYSITTER_SUBSCRIPTION_TERMINAL_EVENT_TYPES = ['pull_request.closed'] +const BABYSITTER_RESOURCE_DELIVERY_RETRY_MS = 5_000 +const BABYSITTER_RESOURCE_SUBSCRIPTION_RENEW_MS = (BABYSITTER_SUBSCRIPTION_TTL_SECONDS * 1_000) / 2 const CLARIFICATION_WAKE_LEASE_MS = 60_000 const CLARIFICATION_WAKE_RETRY_MS = 1_000 const CLARIFICATION_PARK_RETRY_MS = 5_000 @@ -349,6 +380,13 @@ export class FactoryLoop implements Factory { // webhook-fed mount path so readiness can re-read PR meta without a gh call. readonly #babysitterPr = new Map() readonly #babysitterIssueRefs = new Map() + // Relayfile matches subscription IDs server-side. This direct index means a + // delivery claim never requires the legacy local repo/PR scan to find an + // owning babysitter. + readonly #babysitterSubscriptionOwners = new Map() + #babysitterResourceSubscriptionFault = false + #babysitterResourceDeliveryRetryTimer?: ReturnType + #babysitterResourceSubscriptionRenewTimer?: ReturnType readonly #babysitterWakeStates = new Map() // A babysitter announces this fence before invoking destructive git tooling // and clears it afterward. Event text can be broker-delivered while a prompt @@ -650,6 +688,10 @@ export class FactoryLoop implements Factory { async stop(): Promise { this.#started = false this.#stopping = true + if (this.#babysitterResourceDeliveryRetryTimer) clearTimeout(this.#babysitterResourceDeliveryRetryTimer) + this.#babysitterResourceDeliveryRetryTimer = undefined + if (this.#babysitterResourceSubscriptionRenewTimer) clearTimeout(this.#babysitterResourceSubscriptionRenewTimer) + this.#babysitterResourceSubscriptionRenewTimer = undefined if (this.#dispatchLifecycleRenewTimer) clearInterval(this.#dispatchLifecycleRenewTimer) this.#dispatchLifecycleRenewTimer = undefined for (const timer of this.#dispatchLifecycleRetryTimers.values()) clearTimeout(timer) @@ -696,6 +738,7 @@ export class FactoryLoop implements Factory { this.#babysitterSpawned.clear() this.#babysitterPr.clear() this.#babysitterIssueRefs.clear() + this.#babysitterSubscriptionOwners.clear() this.#babysitterCriticalAgents.clear() const subscription = this.#subscription this.#subscription = undefined @@ -5160,18 +5203,34 @@ export class FactoryLoop implements Factory { prNumber: session.prNumber, path: session.path, agentName: session.agentName, + resourceSubscription: session.resourceSubscription, + pendingDeliveryClaims: session.pendingDeliveryClaims, } this.#babysitterPr.set(persistedKey, ref) + if (ref.resourceSubscription) { + this.#babysitterSubscriptionOwners.set(ref.resourceSubscription.subscriptionId, persistedKey) + } this.#babysitterIssueRefs.set(persistedKey, { ...session.issue }) this.#babysitterSpawned.add(persistedKey) if (session.critical) this.#babysitterCriticalAgents.add(session.agentName) this.#increment('babysitterOwnershipRestored') const pendingKinds = session.pendingKinds.filter(isBabysitterWakeKind) if (pendingKinds.length > 0) { - await this.#queueBabysitterWake(session.issue, ref, pendingKinds, tracked) + // A terminal marker can coexist with the one wake persisted before + // acknowledgement. Rehydrate that durable hand-off once, while later + // raw events remain quarantined by #queueBabysitterWake. + await this.#queueBabysitterWake(session.issue, ref, pendingKinds, tracked, { allowTerminal: true }) this.#increment('babysitterPendingWakesRestored') } + await this.#ensureBabysitterResourceSubscription(session.issue, ref, tracked) } + // A crash after the local queue write but before (or just after) the + // remote acceptance leaves an ID in state. Accept is idempotent once the + // lease was accepted, so settle those durable hand-offs before claiming + // new work; lease expiry is retried below when a server has not released + // the original claim yet. + await this.#retryPendingBabysitterDeliveryAcceptances() + await this.#routeDurableBabysitterDeliveries() } async #drainBabysitterWakesForStop(): Promise { @@ -5194,6 +5253,7 @@ export class FactoryLoop implements Factory { const issue = this.#babysitterIssueRefs.get(issueIdentity) const mayClearDurable = this.#fleet.placementLocality !== 'remote' || Boolean(issue && await this.#assertIssueDispatchLifecycleOwner(issue)) + const ref = this.#babysitterPr.get(issueIdentity) for (const [key, state] of this.#babysitterWakeStates) { if (issueKey(state.issue) !== issueIdentity) continue state.cancelled = true @@ -5202,13 +5262,351 @@ export class FactoryLoop implements Factory { this.#babysitterWakeStates.delete(key) this.#babysitterCriticalAgents.delete(state.agentName) } + if (mayClearDurable && ref?.resourceSubscription) { + this.#babysitterSubscriptionOwners.delete(ref.resourceSubscription.subscriptionId) + const subscriptions = this.#mount.resourceSubscriptions + if (subscriptions) { + try { + await subscriptions.cancel(this.#workspaceId, { + subscriptionId: ref.resourceSubscription.subscriptionId, + }) + this.#increment('babysitterResourceSubscriptionsCancelled') + } catch (error) { + // Cancellation is deliberately idempotent. A terminal acceptance may + // have retired this record already; an outage leaves the bounded TTL + // as the leak backstop and must not prevent local session cleanup. + this.#increment('babysitterResourceSubscriptionCancelFailures') + this.#logger.warn?.('[factory] could not cancel durable babysitter resource subscription', { + issue: issue?.key, + subscriptionId: ref.resourceSubscription.subscriptionId, + error: describeError(error).errorMessage, + }) + } + } + } this.#babysitterPr.delete(issueIdentity) this.#babysitterIssueRefs.delete(issueIdentity) this.#babysitterSpawned.delete(issueIdentity) if (mayClearDurable) await this.#state.clearBabysitterSession(this.#workspaceId, issueIdentity) } + async #ensureBabysitterResourceSubscription( + issue: IssueRef, + ref: BabysitterPrRef, + tracked?: TrackedAgent, + ): Promise { + const subscriptions = this.#mount.resourceSubscriptions + if (!subscriptions || !ref.agentName || !await this.#assertIssueDispatchLifecycleOwner(issue)) return + // A terminal claim is persisted before Relayfile acceptance so a crash in + // that gap can never renew a retired record into a fresh generation. + if (ref.resourceSubscription?.terminal) return + + const resourceRef = babysitterResourceRef(ref.repo, ref.prNumber) + const subscriberId = babysitterSubscriberId(issue) + try { + const subscription = await subscriptions.createOrRenew(this.#workspaceId, { + provider: 'github', + resourceRef, + eventTypes: [...BABYSITTER_SUBSCRIPTION_EVENT_TYPES], + terminalEventTypes: [...BABYSITTER_SUBSCRIPTION_TERMINAL_EVENT_TYPES], + subscriberId, + ttlSeconds: BABYSITTER_SUBSCRIPTION_TTL_SECONDS, + }) + if ( + !subscription.subscriptionId || + subscription.provider !== 'github' || + subscription.resourceRef !== resourceRef || + subscription.subscriberId !== subscriberId || + !subscription.ownerId || + !subscription.expiresAt || + !subscription.terminalEventTypes?.includes('pull_request.closed') + ) { + throw new Error('Relayfile returned an invalid durable resource subscription') + } + if (ref.resourceSubscription?.subscriptionId && ref.resourceSubscription.subscriptionId !== subscription.subscriptionId) { + this.#babysitterSubscriptionOwners.delete(ref.resourceSubscription.subscriptionId) + } + ref.resourceSubscription = { + subscriptionId: subscription.subscriptionId, + provider: subscription.provider, + resourceRef: subscription.resourceRef, + subscriberId: subscription.subscriberId, + ownerId: subscription.ownerId, + expiresAt: subscription.expiresAt, + } + this.#babysitterSubscriptionOwners.set(subscription.subscriptionId, issueKey(issue)) + await this.#persistBabysitterSession(issue, ref, tracked) + this.#babysitterResourceSubscriptionFault = false + this.#scheduleBabysitterResourceSubscriptionRenewal() + this.#increment('babysitterResourceSubscriptionsRenewed') + } catch (error) { + if (isResourceSubscriptionsUnavailable(error)) { + this.#babysitterResourceSubscriptionFault = false + this.#increment('babysitterResourceSubscriptionUnavailable') + return + } + this.#babysitterResourceSubscriptionFault = true + this.#scheduleDurableBabysitterDeliveryRetry() + this.#increment('babysitterResourceSubscriptionRenewFailures') + this.#logger.warn?.('[factory] could not create or renew durable babysitter resource subscription', { + issue: issue.key, + repo: ref.repo, + prNumber: ref.prNumber, + error: describeError(error).errorMessage, + }) + } + } + + async #routeDurableBabysitterDeliveries(): Promise { + const subscriptions = this.#mount.resourceSubscriptions + if (!subscriptions || !this.#config.babysitter.enabled || this.#stopping) return false + // Do not bypass the proven local router until every active babysitter has + // completed its own create/renew. This closes the rollout and transient + // provisioning gap without making a successful API response for some + // other subscription suppress an unregistered PR's wake. + if ([...this.#babysitterPr.values()].some((ref) => ref.agentName && !ref.resourceSubscription)) { + return this.#babysitterResourceSubscriptionFault + } + + let claims: Awaited> + try { + claims = await subscriptions.claimDeliveryClaims(this.#workspaceId) + } catch (error) { + if (isResourceSubscriptionsUnavailable(error)) { + this.#babysitterResourceSubscriptionFault = false + this.#increment('babysitterResourceSubscriptionUnavailable') + } else { + this.#babysitterResourceSubscriptionFault = true + this.#scheduleDurableBabysitterDeliveryRetry() + this.#increment('babysitterResourceDeliveryLookupFailures') + this.#logger.warn?.('[factory] durable babysitter delivery-claim lookup failed; retaining durable delivery retry', { + error: describeError(error).errorMessage, + }) + } + return !isResourceSubscriptionsUnavailable(error) + } + this.#babysitterResourceSubscriptionFault = false + + for (const claim of claims) { + const issueIdentity = this.#babysitterSubscriptionOwners.get(claim.subscriptionId) + const issue = issueIdentity ? this.#babysitterIssueRefs.get(issueIdentity) : undefined + const ref = issueIdentity ? this.#babysitterPr.get(issueIdentity) : undefined + const subscription = ref?.resourceSubscription + if ( + !issue || + !ref || + !subscription || + subscription.subscriptionId !== claim.subscriptionId || + subscription.provider !== claim.provider || + subscription.resourceRef !== claim.resourceRef || + subscription.subscriberId !== claim.subscriberId || + subscription.ownerId !== claim.ownerId + ) { + // The service is owner-isolated, but Factory may have just retired a + // local owner. Never route a stale or other-session claim by resource. + this.#increment('babysitterResourceDeliveriesIgnoredUnowned') + continue + } + if (!await this.#assertIssueDispatchLifecycleOwner(issue)) { + this.#increment('babysitterResourceDeliveriesIgnoredNonOwner') + continue + } + + // A terminal delivery may be reclaimed after a process crash before its + // remote acceptance. Only that already-persisted delivery may finish; + // no later claim is allowed to wake or re-open the retired session. + if (subscription.terminal && !ref.pendingDeliveryClaims?.some((pending) => pending.deliveryId === claim.deliveryId)) { + this.#increment('babysitterResourceDeliveriesIgnoredTerminal') + continue + } + + const batch = await this.#batch() + const tracked = batch.getIssue(issue)?.agents.get(ref.agentName) + ?? [...(batch.getIssue(issue)?.agents.values() ?? [])].find((agent) => agent.spec.role === 'babysitter') + ?? durableBabysitterTrackedAgent({ + issue, + repo: ref.repo, + prNumber: ref.prNumber, + path: ref.path, + agentName: ref.agentName, + critical: false, + pendingKinds: [], + resourceSubscription: subscription, + pendingDeliveryClaims: ref.pendingDeliveryClaims, + }) + + const pendingClaim = ref.pendingDeliveryClaims?.find((pending) => pending.deliveryId === claim.deliveryId) + const alreadyQueued = Boolean(pendingClaim) + if (!alreadyQueued) { + const queued = await this.#queueBabysitterWake(issue, ref, ['pull-request-state'], tracked) + if (!queued) continue + } + if (!pendingClaim || pendingClaim.claimToken !== claim.claimToken) { + ref.pendingDeliveryClaims = [ + ...(ref.pendingDeliveryClaims ?? []).filter((pending) => pending.deliveryId !== claim.deliveryId), + { deliveryId: claim.deliveryId, claimToken: claim.claimToken }, + ] + // The claim lease joins Factory's durable pending-wake state before + // the external acceptance. A crash after this point can retry the + // exact hand-off without delivering the same wake a second time. + await this.#persistBabysitterSession(issue, ref, tracked) + } + + try { + if (claim.terminal && !subscription.terminal) { + subscription.terminal = true + await this.#persistBabysitterSession(issue, ref, tracked) + } + const accepted = await subscriptions.acceptDelivery(this.#workspaceId, { + deliveryId: claim.deliveryId, + claimToken: claim.claimToken, + }) + if (accepted.deliveryId !== claim.deliveryId || accepted.subscriptionId !== claim.subscriptionId) { + throw new Error('Relayfile accepted a different durable delivery claim') + } + if (accepted.terminal || claim.terminal) { + // Keep the terminal marker and subscription identity locally until + // normal PR/session teardown. That quarantines the babysitter from + // both legacy fallback and a restart-time create-or-renew. + subscription.terminal = true + ref.pendingDeliveryClaims = (ref.pendingDeliveryClaims ?? []).filter((pending) => pending.deliveryId !== claim.deliveryId) + await this.#persistBabysitterSession(issue, ref, tracked) + this.#increment('babysitterResourceSubscriptionsRetiredTerminal') + } else { + ref.pendingDeliveryClaims = (ref.pendingDeliveryClaims ?? []).filter((pending) => pending.deliveryId !== claim.deliveryId) + await this.#persistBabysitterSession(issue, ref, tracked) + } + this.#increment('babysitterResourceDeliveriesAccepted') + } catch (error) { + this.#increment('babysitterResourceDeliveryAcceptFailures') + this.#logger.warn?.('[factory] durable babysitter delivery claim remains pending after wake queue', { + issue: issue.key, + subscriptionId: claim.subscriptionId, + deliveryId: claim.deliveryId, + error: describeError(error).errorMessage, + }) + } + } + if ([...this.#babysitterPr.values()].some((ref) => ref.pendingDeliveryClaims?.length)) { + this.#scheduleDurableBabysitterDeliveryRetry() + } + return true + } + + async #retryPendingBabysitterDeliveryAcceptances(): Promise { + const subscriptions = this.#mount.resourceSubscriptions + if (!subscriptions) return + const retrySubscriptionRenewal = this.#babysitterResourceSubscriptionFault + + for (const [issueIdentity, ref] of this.#babysitterPr) { + const issue = this.#babysitterIssueRefs.get(issueIdentity) + if (issue && ref.agentName && !ref.resourceSubscription?.terminal && (!ref.resourceSubscription || retrySubscriptionRenewal)) { + const batch = await this.#batch() + const tracked = batch.getIssue(issue)?.agents.get(ref.agentName) + ?? [...(batch.getIssue(issue)?.agents.values() ?? [])].find((agent) => agent.spec.role === 'babysitter') + await this.#ensureBabysitterResourceSubscription(issue, ref, tracked) + } + const subscription = ref.resourceSubscription + const pendingDeliveryClaims = [...(ref.pendingDeliveryClaims ?? [])] + if (!subscription || !issue || pendingDeliveryClaims.length === 0) continue + const batch = await this.#batch() + const tracked = batch.getIssue(issue)?.agents.get(ref.agentName) + ?? [...(batch.getIssue(issue)?.agents.values() ?? [])].find((agent) => agent.spec.role === 'babysitter') + ?? durableBabysitterTrackedAgent({ + issue, + repo: ref.repo, + prNumber: ref.prNumber, + path: ref.path, + agentName: ref.agentName, + critical: false, + pendingKinds: [], + resourceSubscription: subscription, + pendingDeliveryClaims, + }) + for (const { deliveryId, claimToken } of pendingDeliveryClaims) { + try { + const accepted = await subscriptions.acceptDelivery(this.#workspaceId, { deliveryId, claimToken }) + if (accepted.deliveryId !== deliveryId || accepted.subscriptionId !== subscription.subscriptionId) { + throw new Error('Relayfile accepted a different durable delivery claim') + } + if (accepted.terminal) subscription.terminal = true + ref.pendingDeliveryClaims = (ref.pendingDeliveryClaims ?? []).filter((pending) => pending.deliveryId !== deliveryId) + await this.#persistBabysitterSession(issue, ref, tracked) + this.#increment('babysitterResourceDeliveriesAcceptedAfterRestore') + } catch (error) { + if (isResourceSubscriptionsUnavailable(error)) { + this.#babysitterResourceSubscriptionFault = false + this.#increment('babysitterResourceSubscriptionUnavailable') + } else { + this.#babysitterResourceSubscriptionFault = true + this.#increment('babysitterResourceDeliveryAcceptFailures') + this.#logger.warn?.('[factory] durable babysitter delivery acceptance remains pending after restore', { + issue: issue.key, + subscriptionId: subscription.subscriptionId, + deliveryId, + error: describeError(error).errorMessage, + }) + } + } + } + } + if ([...this.#babysitterPr.values()].some((ref) => ref.pendingDeliveryClaims?.length)) { + this.#scheduleDurableBabysitterDeliveryRetry() + } + } + + #scheduleBabysitterResourceSubscriptionRenewal(): void { + if ( + this.#babysitterResourceSubscriptionRenewTimer || + this.#stopping || + !this.#mount.resourceSubscriptions || + ![...this.#babysitterPr.values()].some((ref) => ref.resourceSubscription && !ref.resourceSubscription.terminal) + ) return + this.#babysitterResourceSubscriptionRenewTimer = setTimeout(() => { + this.#babysitterResourceSubscriptionRenewTimer = undefined + void (async () => { + const batch = await this.#batch() + for (const [issueIdentity, ref] of this.#babysitterPr) { + const issue = this.#babysitterIssueRefs.get(issueIdentity) + if (!issue || !ref.resourceSubscription || ref.resourceSubscription.terminal) continue + const tracked = batch.getIssue(issue)?.agents.get(ref.agentName) + ?? [...(batch.getIssue(issue)?.agents.values() ?? [])].find((agent) => agent.spec.role === 'babysitter') + await this.#ensureBabysitterResourceSubscription(issue, ref, tracked) + } + })().catch((error) => { + this.#logger.warn?.('[factory] durable babysitter subscription renewal rejected', { + error: describeError(error).errorMessage, + }) + }).finally(() => { + this.#scheduleBabysitterResourceSubscriptionRenewal() + }) + }, BABYSITTER_RESOURCE_SUBSCRIPTION_RENEW_MS) + this.#babysitterResourceSubscriptionRenewTimer.unref?.() + } + + #scheduleDurableBabysitterDeliveryRetry(): void { + if (this.#babysitterResourceDeliveryRetryTimer || this.#stopping || !this.#mount.resourceSubscriptions) return + this.#babysitterResourceDeliveryRetryTimer = setTimeout(() => { + this.#babysitterResourceDeliveryRetryTimer = undefined + void (async () => { + await this.#retryPendingBabysitterDeliveryAcceptances() + await this.#routeDurableBabysitterDeliveries() + })().catch((error) => { + this.#logger.warn?.('[factory] durable babysitter delivery retry rejected', { + error: describeError(error).errorMessage, + }) + }) + }, BABYSITTER_RESOURCE_DELIVERY_RETRY_MS) + this.#babysitterResourceDeliveryRetryTimer.unref?.() + } + async #routeBabysitterEvent(path: string, extraKinds: Iterable = []): Promise { + // A successful Relayfile claim lookup is the new exact demux. Fall back to + // the legacy path router only while the capability is absent; transient + // failures retain and retry service claims so they cannot double-deliver, + // preserving rollout and degraded-service behaviour without teaching this + // runtime how to resolve canonical/renamed paths to their by-id anchors. + if (await this.#routeDurableBabysitterDeliveries()) return const event = githubBabysitterEventPathParts(path) if (!event || !this.#config.babysitter.enabled || this.#stopping) return let targets: Array<{ prNumber: number; kinds: BabysitterWakeKind[] }> @@ -5303,10 +5701,15 @@ export class FactoryLoop implements Factory { ref: BabysitterPrRef, kinds: Iterable, tracked: TrackedAgent, - ): Promise { + options: { allowTerminal?: boolean } = {}, + ): Promise { + if (ref.resourceSubscription?.terminal && !options.allowTerminal) { + this.#increment('babysitterEventsIgnoredTerminal') + return false + } if (!await this.#assertIssueDispatchLifecycleOwner(issue)) { this.#increment('babysitterEventsIgnoredNonOwner') - return + return false } // Owner lookup and queueing straddle async mount/state reads. Revalidate // the exact composite owner so a concurrent close/merge cancellation can @@ -5318,7 +5721,7 @@ export class FactoryLoop implements Factory { githubPrIdentity(current.repo, current.prNumber) !== githubPrIdentity(ref.repo, ref.prNumber) ) { this.#increment('babysitterEventsIgnoredStaleOwner') - return + return false } const key = babysitterWakeKey(issue, ref) let state = this.#babysitterWakeStates.get(key) @@ -5346,9 +5749,10 @@ export class FactoryLoop implements Factory { if (state.deferredSubmitTargets || state.inFlight || this.#babysitterCriticalAgents.has(state.agentName)) { this.#increment('babysitterEventWakesDeferred') - return + return true } this.#scheduleBabysitterWake(state, BABYSITTER_EVENT_COALESCE_MS) + return true } async #recordPendingBabysitterWake(state: BabysitterWakeState): Promise { @@ -5393,6 +5797,8 @@ export class FactoryLoop implements Factory { path: ref.path, critical: this.#babysitterCriticalAgents.has(ref.agentName), pendingKinds: pending?.kinds.filter(isBabysitterWakeKind).sort(compareBabysitterWakeKinds) ?? [], + ...(ref.resourceSubscription ? { resourceSubscription: { ...ref.resourceSubscription } } : {}), + ...(ref.pendingDeliveryClaims?.length ? { pendingDeliveryClaims: structuredClone(ref.pendingDeliveryClaims) } : {}), }) } @@ -5608,10 +6014,19 @@ export class FactoryLoop implements Factory { } if (!this.#config.babysitter.enabled) return if (snapshot.state && snapshot.state.trim().toUpperCase() !== 'OPEN') { + // A provider close produces a separately indexed terminal claim. Keep + // the durable owner until it has been accepted (or a transient retry + // has claimed it); do not let closed-state cleanup erase that hand-off. + await this.#routeDurableBabysitterDeliveries() + if (this.#babysitterResourceSubscriptionFault) return await this.#cancelBabysitterWake(ownedKey) return } if (snapshot.draft) this.#increment('babysitterDraftPrSkipped') + // PR meta events are also the normal renewal heartbeat for the durable + // record. The store's identity makes this a create-or-renew, never a + // second subscription for the same babysitter. + await this.#ensureBabysitterResourceSubscription(owned.issue, owned.ref, owned.tracked) await this.#routeBabysitterEvent(path, babysitterWakeKindsFromSnapshot(snapshot)) return } @@ -5650,6 +6065,12 @@ export class FactoryLoop implements Factory { } if (snapshot.state && snapshot.state.trim().toUpperCase() !== 'OPEN') { + // `pull_request.closed` is a separately indexed Relayfile terminal + // event. Claim and accept its durable hand-off before the local closed + // PR cleanup drops the subscription owner. On a transient service fault, + // retain the owner so the retry loop can claim it without local fallback. + await this.#routeDurableBabysitterDeliveries() + if (this.#babysitterResourceSubscriptionFault) return if (babysitterKey && existing) await this.#cancelBabysitterWake(babysitterKey) return } @@ -5834,6 +6255,11 @@ export class FactoryLoop implements Factory { await this.#babysitterSpawnInFlight.get(babysitterKey) const settled = this.#babysitterPr.get(babysitterKey) if (settled && prRef.path) settled.path = prRef.path + if (settled) { + const tracked = record.agents.get(settled.agentName) + ?? [...record.agents.values()].find((agent) => agent.spec.role === 'babysitter') + await this.#ensureBabysitterResourceSubscription(record.issue, settled, tracked) + } return } const trackedBabysitter = [...record.agents.entries()].find(([, agent]) => agent.spec.role === 'babysitter') @@ -5852,7 +6278,9 @@ export class FactoryLoop implements Factory { agentName: tracked.result?.name ?? trackedName, }) this.#babysitterSpawned.add(babysitterKey) - await this.#persistBabysitterSession(record.issue, this.#babysitterPr.get(babysitterKey)!, tracked) + const ref = this.#babysitterPr.get(babysitterKey)! + await this.#persistBabysitterSession(record.issue, ref, tracked) + await this.#ensureBabysitterResourceSubscription(record.issue, ref, tracked) return } // Reserve up-front so concurrent PR events in a drain don't double-spawn. @@ -5904,7 +6332,9 @@ export class FactoryLoop implements Factory { path: prRef.path, agentName: tracked?.result?.name ?? spawned.name, }) - await this.#persistBabysitterSession(record.issue, this.#babysitterPr.get(babysitterKey)!, tracked) + const ref = this.#babysitterPr.get(babysitterKey)! + await this.#persistBabysitterSession(record.issue, ref, tracked) + await this.#ensureBabysitterResourceSubscription(record.issue, ref, tracked) await this.#writeInFlightRegistry() if (!await this.#saveDispatchLifecycle(record, 'running')) return this.#increment('babysittersSpawned') @@ -6172,8 +6602,8 @@ export class FactoryLoop implements Factory { const stateKey = issueStateKey(record.issue) this.#probePrGhBackoffUntilMs.delete(stateKey) this.#probePrResolvedCache.delete(stateKey) - this.#babysitterSpawned.delete(completionKey) - this.#babysitterPr.delete(completionKey) + // Cancellation must see the subscription identity so it can issue the + // idempotent Relayfile DELETE before clearing the local owner maps. await this.#cancelBabysitterWake(completionKey) const durable = await this.#state.getDispatchLifecycle(this.#workspaceId, issueKey(record.issue)).catch(() => undefined) if (this.#fleet.placementLocality !== 'remote' || (durable && isTerminalDispatchLifecycle(durable))) { @@ -9184,6 +9614,19 @@ const validPrNumber = (value: number): boolean => Number.isInteger(value) && val const githubPrIdentity = (repo: string, prNumber: number): string | undefined => validGithubRepo(repo) && validPrNumber(prNumber) ? `${repo.toLowerCase()}#${prNumber}` : undefined +// This is the public Relayfile stable identity for a GitHub pull request. It +// deliberately comes from the PR's repo/number ownership record, never by +// transforming an incoming canonical path (whose title slug can be renamed). +const babysitterResourceRef = (repo: string, prNumber: number): string => { + if (!validGithubRepo(repo) || !validPrNumber(prNumber)) { + throw new Error('Cannot create a durable babysitter subscription for an invalid GitHub PR identity') + } + const [owner, name] = repo.split('/') + return `/github/repos/${owner}__${name}/pulls/by-id/${prNumber}.json` +} + +const babysitterSubscriberId = (issue: IssueRef): string => `factory-babysitter:${issue.uuid}` + const recordMatchesGithubRepo = (record: InFlightIssue, eventRepo: string, defaultOwner?: string): boolean => { if (!validGithubRepo(eventRepo)) return false const wanted = eventRepo.toLowerCase() diff --git a/src/ports/mount.ts b/src/ports/mount.ts index cb52e00..ab37c0e 100644 --- a/src/ports/mount.ts +++ b/src/ports/mount.ts @@ -2,6 +2,7 @@ import type { ChangeEvent as RelayFileChangeEvent, Subscription as RelayFileSubscription, } from '@relayfile/sdk' +import type { ResourceSubscriptionsClient } from '../subscriptions/resource-subscriptions' export type ChangeEvent = RelayFileChangeEvent export type Subscription = RelayFileSubscription @@ -63,6 +64,12 @@ export interface GithubConnectionWrite { export interface MountClient { readonly writebackTransport?: 'relayfile-cloud' | 'test' readonly githubWrite?: GithubConnectionWrite + /** + * Optional durable Relayfile resource-subscription API. Its absence means + * this mount targets a pre-subscription service and consumers retain their + * established local routing behaviour. + */ + readonly resourceSubscriptions?: ResourceSubscriptionsClient readFile(path: string): Promise<{ content: unknown; revision?: string }> writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise deleteFile(path: string): Promise diff --git a/src/ports/state.ts b/src/ports/state.ts index 5eb8fd1..8560d1d 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -64,6 +64,19 @@ export type BabysitterSessionState = { path?: string critical: boolean pendingKinds: string[] + /** Durable Relayfile subscription identity, if the workspace supports it. */ + resourceSubscription?: { + subscriptionId: string + provider: string + resourceRef: string + subscriberId: string + ownerId: string + expiresAt: string + /** Terminal delivery accepted or awaiting acceptance; never renew this record. */ + terminal?: boolean + } + /** Claims already queued locally but not yet accepted by Relayfile. */ + pendingDeliveryClaims?: Array<{ deliveryId: string; claimToken: string }> } export type DispatchAttemptState = { diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index 243b3c5..3f2861c 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -64,6 +64,16 @@ describe('FileStateStore', () => { path: '/github/repos/AgentWorkforce/factory/pulls/87/metadata.json', critical: true, pendingKinds: ['checks-failed', 'review-comment'], + resourceSubscription: { + subscriptionId: 'sub-87', + provider: 'github', + resourceRef: '/github/repos/AgentWorkforce__factory/pulls/by-id/87.json', + subscriberId: 'factory-babysitter:uuid-87', + ownerId: 'factory-runtime', + expiresAt: '2026-12-31T00:00:00.000Z', + terminal: true, + }, + pendingDeliveryClaims: [{ deliveryId: 'delivery-87-terminal', claimToken: 'claim-token-87' }], } const first = new FileStateStore({ batchSize: 2, watchStatePath }) await first.setBabysitterSession('workspace-1', 'AR-87:uuid-87:/linear/issues/AR-87__uuid-87.json', session) diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 7efb645..c0cacf4 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -767,7 +767,23 @@ const parseBabysitterSessions = (value: Record): Record typeof kind === 'string') + !candidate.pendingKinds.every((kind) => typeof kind === 'string') || + (candidate.pendingDeliveryClaims !== undefined && ( + !Array.isArray(candidate.pendingDeliveryClaims) || + !candidate.pendingDeliveryClaims.every((claim) => + isRecord(claim) && typeof claim.deliveryId === 'string' && typeof claim.claimToken === 'string' + ) + )) || + (candidate.resourceSubscription !== undefined && ( + !isRecord(candidate.resourceSubscription) || + typeof candidate.resourceSubscription.subscriptionId !== 'string' || + typeof candidate.resourceSubscription.provider !== 'string' || + typeof candidate.resourceSubscription.resourceRef !== 'string' || + typeof candidate.resourceSubscription.subscriberId !== 'string' || + typeof candidate.resourceSubscription.ownerId !== 'string' || + typeof candidate.resourceSubscription.expiresAt !== 'string' || + (candidate.resourceSubscription.terminal !== undefined && typeof candidate.resourceSubscription.terminal !== 'boolean') + )) ) { throw new Error('Factory GitHub watch state file is invalid') } @@ -779,6 +795,23 @@ const parseBabysitterSessions = (value: Record): Record>).map((claim) => ({ + deliveryId: claim.deliveryId as string, + claimToken: claim.claimToken as string, + })), + }), } } return sessions diff --git a/src/subscriptions/index.ts b/src/subscriptions/index.ts index 6a66034..8f238fb 100644 --- a/src/subscriptions/index.ts +++ b/src/subscriptions/index.ts @@ -35,6 +35,20 @@ export { filesystemEventToChangeEvent, integrationRelayFileSyncOptions, } from './event-client' +export { + createResourceSubscriptionsHttpClient, + isResourceSubscriptionsUnavailable, + ResourceSubscriptionsHttpError, + ResourceSubscriptionsUnavailableError, +} from './resource-subscriptions' +export type { + AcceptedResourceDelivery, + ResourceDeliveryClaim, + ResourceSubscription, + ResourceSubscriptionInput, + ResourceSubscriptionsClient, + ResourceSubscriptionsHttpClientOptions, +} from './resource-subscriptions' export type { ChangeEvent, FilesystemEventLike, diff --git a/src/subscriptions/resource-subscriptions.ts b/src/subscriptions/resource-subscriptions.ts new file mode 100644 index 0000000..f1b5948 --- /dev/null +++ b/src/subscriptions/resource-subscriptions.ts @@ -0,0 +1,262 @@ +/** + * Consumer-facing contract for Relayfile's durable resource-subscription API. + * + * Relayfile owns matching and delivery-claim persistence. A runtime owns the + * meaning of its opaque IDs and must only accept a claim after it has made the + * corresponding wake durable on its side. + */ +export type ResourceSubscriptionInput = { + provider: string + resourceRef: string + eventTypes: string[] + terminalEventTypes?: string[] + subscriberId: string + intent?: string + ttlSeconds: number +} + +export type ResourceSubscription = { + subscriptionId: string + /** Derived by Relayfile from the authenticated workspace bearer token. */ + ownerId: string + provider: string + resourceRef: string + eventTypes: string[] + terminalEventTypes?: string[] + subscriberId: string + intent?: string + expiresAt: string +} + +export type ResourceDeliveryClaim = { + deliveryId: string + /** Opaque, short-lived lease credential required to accept this claim. */ + claimToken: string + subscriptionId: string + resourceRef: string + eventType: string + subscriberId: string + ownerId: string + provider: string + terminal: boolean +} + +export type AcceptedResourceDelivery = { + deliveryId: string + subscriptionId: string + terminal: boolean +} + +export interface ResourceSubscriptionsClient { + createOrRenew(workspaceId: string, input: ResourceSubscriptionInput): Promise + /** Atomically lease pending claims for the authenticated owner. */ + claimDeliveryClaims(workspaceId: string, input?: { limit?: number }): Promise + acceptDelivery(workspaceId: string, input: { deliveryId: string; claimToken: string }): Promise + cancel(workspaceId: string, input: { subscriptionId: string }): Promise +} + +/** The service is absent or too old; callers may retain their legacy route. */ +export class ResourceSubscriptionsUnavailableError extends Error { + constructor(message = 'Relayfile durable resource subscriptions are unavailable') { + super(message) + this.name = 'ResourceSubscriptionsUnavailableError' + } +} + +export class ResourceSubscriptionsHttpError extends Error { + readonly status: number + + constructor(status: number, message: string) { + super(message) + this.name = 'ResourceSubscriptionsHttpError' + this.status = status + } +} + +export const isResourceSubscriptionsUnavailable = (error: unknown): boolean => { + if (error instanceof ResourceSubscriptionsUnavailableError) return true + if (!error || typeof error !== 'object') return false + const record = error as Record + const status = record.status ?? record.statusCode ?? record.httpStatus + return status === 404 +} + +export type ResourceSubscriptionsHttpClientOptions = { + baseUrl: string + tokenProvider: () => string | undefined | Promise + fetch?: typeof fetch + /** Bounds an individual Relayfile API call; defaults to 15 seconds. */ + requestTimeoutMs?: number + /** Optional lifecycle cancellation shared by all requests from this client. */ + signal?: AbortSignal +} + +const DEFAULT_RESOURCE_SUBSCRIPTION_REQUEST_TIMEOUT_MS = 15_000 + +/** + * Minimal REST adapter for Relayfile Cloud's public subscription contract. + * The SDK can adopt these endpoints later without changing Factory's port. + */ +export function createResourceSubscriptionsHttpClient( + options: ResourceSubscriptionsHttpClientOptions, +): ResourceSubscriptionsClient { + const requestTimeoutMs = options.requestTimeoutMs === undefined + ? DEFAULT_RESOURCE_SUBSCRIPTION_REQUEST_TIMEOUT_MS + : Math.max(1, Math.trunc(options.requestTimeoutMs)) + + const request = async (workspaceId: string, path: string, init: RequestInit = {}): Promise => { + const token = await options.tokenProvider() + if (!token) throw new ResourceSubscriptionsHttpError(401, 'Relayfile workspace bearer token is unavailable') + const url = new URL(`/v1/workspaces/${encodeURIComponent(workspaceId)}${path}`, options.baseUrl) + const abort = new AbortController() + const forwardedSignals = [options.signal, init.signal].filter((signal): signal is AbortSignal => Boolean(signal)) + const removeAbortListeners: Array<() => void> = [] + for (const signal of forwardedSignals) { + const forwardAbort = () => abort.abort(signal.reason) + if (signal.aborted) forwardAbort() + else { + signal.addEventListener('abort', forwardAbort, { once: true }) + removeAbortListeners.push(() => signal.removeEventListener('abort', forwardAbort)) + } + } + let timedOut = false + const timeout = setTimeout(() => { + timedOut = true + abort.abort(new Error('Relayfile resource-subscription request timed out')) + }, requestTimeoutMs) + let response: Response + try { + response = await (options.fetch ?? fetch)(url, { + ...init, + signal: abort.signal, + headers: { + accept: 'application/json', + authorization: `Bearer ${token}`, + ...(init.body ? { 'content-type': 'application/json' } : {}), + ...init.headers, + }, + }) + } catch (error) { + if (timedOut) { + throw new ResourceSubscriptionsHttpError(504, `Relayfile resource-subscription request timed out after ${requestTimeoutMs}ms`) + } + throw error + } finally { + clearTimeout(timeout) + for (const remove of removeAbortListeners) remove() + } + if (!response.ok) { + if (response.status === 404) throw new ResourceSubscriptionsUnavailableError() + throw new ResourceSubscriptionsHttpError(response.status, `Relayfile resource-subscription request failed (${response.status})`) + } + if (response.status === 204) return undefined + try { + return await response.json() + } catch { + throw new ResourceSubscriptionsHttpError(response.status, 'Relayfile resource-subscription response was not valid JSON') + } + } + + return { + async createOrRenew(workspaceId, input) { + const payload = await request(workspaceId, '/subscriptions', { + method: 'POST', + body: JSON.stringify(input), + }) + return parseSubscription(payload) + }, + async claimDeliveryClaims(workspaceId, input) { + const payload = await request(workspaceId, '/subscriptions/deliveries/claim', { + method: 'POST', + body: JSON.stringify(input ?? {}), + }) + const envelope = record(payload) + const deliveries = Array.isArray(payload) + ? payload + : Array.isArray(envelope?.deliveries) + ? envelope.deliveries + : undefined + if (!deliveries) { + throw new ResourceSubscriptionsHttpError(502, 'Relayfile delivery claim response was invalid') + } + return deliveries.map(parseDeliveryClaim) + }, + async acceptDelivery(workspaceId, input) { + const payload = await request(workspaceId, `/subscriptions/deliveries/${encodeURIComponent(input.deliveryId)}/accept`, { + method: 'POST', + body: JSON.stringify({ claimToken: input.claimToken }), + }) + return parseAcceptedDelivery(payload) + }, + async cancel(workspaceId, input) { + await request(workspaceId, `/subscriptions/${encodeURIComponent(input.subscriptionId)}`, { method: 'DELETE' }) + }, + } +} + +const record = (value: unknown): Record | undefined => + value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : undefined + +const string = (value: unknown, field: string): string => { + if (typeof value !== 'string' || !value) throw new ResourceSubscriptionsHttpError(502, `Relayfile resource-subscription response omitted ${field}`) + return value +} + +const stringArray = (value: unknown, field: string): string[] => { + if (!Array.isArray(value) || !value.every((item) => typeof item === 'string')) { + throw new ResourceSubscriptionsHttpError(502, `Relayfile resource-subscription response omitted ${field}`) + } + return value +} + +const boolean = (value: unknown, field: string): boolean => { + if (typeof value !== 'boolean') throw new ResourceSubscriptionsHttpError(502, `Relayfile resource-subscription response omitted ${field}`) + return value +} + +const parseSubscription = (value: unknown): ResourceSubscription => { + const source = record(record(value)?.subscription) ?? record(value) + if (!source) throw new ResourceSubscriptionsHttpError(502, 'Relayfile resource-subscription response was invalid') + return { + subscriptionId: string(source.id ?? source.subscriptionId, 'subscription id'), + ownerId: string(source.ownerId, 'ownerId'), + subscriberId: string(source.subscriberId, 'subscriberId'), + provider: string(source.provider, 'provider'), + resourceRef: string(source.resourceRef, 'resourceRef'), + eventTypes: stringArray(source.eventTypes, 'eventTypes'), + ...(source.terminalEventTypes === undefined ? {} : { terminalEventTypes: stringArray(source.terminalEventTypes, 'terminalEventTypes') }), + ...(typeof source.intent === 'string' && source.intent ? { intent: source.intent } : {}), + expiresAt: string(source.expiresAt, 'expiresAt'), + } +} + +const parseDeliveryClaim = (value: unknown): ResourceDeliveryClaim => { + const source = record(value) + if (!source) throw new ResourceSubscriptionsHttpError(502, 'Relayfile delivery claim response was invalid') + const event = record(source.event) + return { + deliveryId: string(source.deliveryId ?? source.id, 'deliveryId'), + claimToken: string(source.claimToken, 'claimToken'), + subscriptionId: string(source.subscriptionId, 'subscriptionId'), + ownerId: string(source.ownerId, 'ownerId'), + subscriberId: string(source.subscriberId, 'subscriberId'), + provider: string(source.provider, 'provider'), + resourceRef: string(source.resourceRef, 'resourceRef'), + eventType: typeof source.eventType === 'string' + ? source.eventType + : string(event?.type, 'event.type'), + terminal: boolean(source.terminal, 'terminal'), + } +} + +const parseAcceptedDelivery = (value: unknown): AcceptedResourceDelivery => { + const source = record(record(value)?.delivery) ?? record(value) + if (!source) throw new ResourceSubscriptionsHttpError(502, 'Relayfile delivery acceptance response was invalid') + return { + deliveryId: string(source.deliveryId ?? source.id, 'deliveryId'), + subscriptionId: string(source.subscriptionId, 'subscriptionId'), + terminal: boolean(source.terminal, 'terminal'), + } +} diff --git a/src/testing/fakes.ts b/src/testing/fakes.ts index eab7d64..dcfee2c 100644 --- a/src/testing/fakes.ts +++ b/src/testing/fakes.ts @@ -13,6 +13,7 @@ import type { Subscription, Capability, } from '../ports' +import type { ResourceSubscriptionsClient } from '../subscriptions' type ExitListener = (name: string, reason?: string) => void type DeliveryFailedListener = (info: { to: string; msgId?: string; reason?: string }) => void @@ -21,6 +22,7 @@ type AgentMessageListener = (message: AgentMessage) => void export class FakeMountClient implements MountClient { readonly writebackTransport = 'test' githubWrite?: GithubConnectionWrite + resourceSubscriptions?: ResourceSubscriptionsClient readonly files = new Map() readonly writes: Array<{ path: string; content: unknown }> = [] readonly deletes: string[] = []