diff --git a/README.md b/README.md index 5ebb9a5..4f03745 100644 --- a/README.md +++ b/README.md @@ -163,11 +163,20 @@ instruction to rerun the Factory command in an interactive terminal instead. ### Mounted Notion specs -Notion is an intake source, not a lifecycle writeback surface. Factory reads a -Relayfile-mounted page, binds it to a stable `notion:` identity and -digest, then either creates a labeled GitHub issue for a repository target or -dispatches an exact-path workspace task through the fleet. Existing GitHub and -Linear discovery continue unchanged. +Notion is an intake source, not a lifecycle issue source or writeback surface. +Factory reads a Relayfile-mounted page, binds it to a stable +`notion:` identity and digest, then either creates a labeled GitHub +issue for a repository target or dispatches an exact-path workspace task +through the fleet. The config field `issueSource` continues to select the +discovery and lifecycle-writeback adapter (`linear` or `github`); Notion intake +normalizes into that lifecycle instead of becoming a third adapter. + +The absence of Notion writeback is deliberate. Relayfile supplies the mounted +page as a read-only execution contract, and Factory has no guarded Notion +property/comment writeback adapter or operator-defined lifecycle mapping. +Repository targets reconcile accepted, dispatched, PR, blocked, and completed +state on their generated GitHub issue. Native Notion lifecycle reconciliation +requires a separate provider capability and product contract. New pages fail closed unless the first block is an explicit ready spec: @@ -206,15 +215,35 @@ hatch is deliberately page-specific; there is no title or content heuristic. } ``` -`mountRoot` and `statePath` resolve relative to the manifest file. `workerMountRoot` -is the repo-relative read-only mount workers receive. With the recommended +`mountRoot` and `statePath` resolve relative to the manifest file. `statePath` +is a local receipt cache, not the dispatch authority: before creating an issue +or spawning an agent, Factory creates one immutable, digest-bound claim in the +active Agent Relay workspace. Workspace-global claim-channel uniqueness stops +two machines with independent caches from dispatching the same source key. A +failed or ambiguous claim write blocks dispatch. All non-dry-run Notion intake +therefore requires a resolvable active Agent Relay workspace key. + +`workerMountRoot` is the repo-relative read-only mount workers receive. With the recommended `relay-channel` transport, Factory base64-chunks the digest-bound mounted bytes into a workspace-private Agent Relay channel. A worker on any fleet machine can reconstruct the exact file at `workerMountRoot`, set it to mode `0444`, and apply the source SHA-256 gate without exposing the page in a public issue. -The field defaults to `{ "kind": "local" }` whenever it is omitted, including -in new manifests. Portable delivery must be selected explicitly and requires a -resolvable active Agent Relay workspace key; otherwise dispatch fails closed. +`workerMountTransport` defaults to `{ "kind": "local" }` whenever it is +omitted, including in new manifests. Portable delivery must be selected +explicitly; otherwise the worker uses the local mount path. Missing portable +delivery or claim capability fails closed. + +If dispatch stops after creating the durable claim but before recording its +receipt, re-running stays blocked rather than guessing whether the downstream +side effect happened. The blocked result includes the source key. A workspace +administrator can derive the claim channel as +`factory-notion-claim-`. After verifying that no matching +lifecycle issue or workspace agent exists, the administrator may delete that +channel in Agent Relay and re-run intake. Never clear a claim merely because +its local receipt is missing: the shared claim, issue marker, and running agent +must be reconciled first. A blocked portable-mount migration uses the same +channel formula with `:portable-mount` as the hashed value; verify +the worker was not redispatched before clearing it. `page` accepts a Notion URL or a bare page ID. Plan without writes, then dispatch: @@ -227,9 +256,9 @@ factory intake notion ./ops/notion-intake.json --backend relay Repository targets require the `factory-ready` and matching `agent:` labels to already exist. Factory automatically prefixes their issue titles with `[factory]` so the hosted brain's independent safety gate can -accept them. Re-running is idempotent: GitHub work -is claimed by a hidden source marker, while exact-path dispatches use a local -digest-bound receipt. A changed mounted spec blocks instead of silently mutating +accept them. Re-running reconciles the workspace-global claim with the GitHub +source marker or the running exact-path agent; the local digest-bound receipt is +only a cache. A changed mounted spec blocks instead of silently mutating already-dispatched work. ### Feature-map validation @@ -538,7 +567,13 @@ Tokens involved — set only the first one on the orchestrator host: ## Configuration -Pass a JSON file via `--config`. Beyond the two required fields above, useful +Factory resolves exactly one contract: the path passed via `--config`, or +`./factory.config.json` in the command's current working directory when the +flag is omitted. It does not search target repositories, walk to a clone root, +or merge multiple configs; a config in another repository is inert unless it +is selected explicitly. + +Beyond the two required fields above, useful knobs include issue **routing** (`repos.byLabel` / `byProject` / `keywordRules` / `default`), the **safety gate** (`safety.requireTitlePrefix`, `safety.requireTeamKey`), `mergePolicy` (defaults to `never`), per-role **model** overrides, and an optional diff --git a/docs/notion-ticket-feeder-assessment.md b/docs/notion-ticket-feeder-assessment.md new file mode 100644 index 0000000..89ad38e --- /dev/null +++ b/docs/notion-ticket-feeder-assessment.md @@ -0,0 +1,96 @@ +# Notion as a first-class Factory ticket feeder + +Date: 2026-08-06 + +## Decision summary + +Notion should remain a separate, explicit intake command that normalizes +repository tasks into GitHub lifecycle issues or dispatches exact-path fleet +work. It should not become a third `issueSource` until Factory has both native +Notion discovery semantics and a guarded lifecycle-writeback adapter. The +immediate correctness gap is cross-dispatcher claim durability. + +## 1. Claim durability + +**Assessment:** defect; fix now. + +The prior `.factory/notion-intake-state.json`-style receipt was protected only +by a same-filesystem lock. Two machines using different state paths could both +observe no receipt and dispatch the same stable `notion::` work +unit. The stable source identity and digest were already correct; the authority +was on the wrong surface. + +**Recommendation and implementation:** use one immutable Agent Relay claim +channel per hashed source key. Channel-name uniqueness in the shared workspace +is the atomic cross-dispatcher gate. The winner must receive an acknowledged, +digest-bound claim record before GitHub issue creation, lifecycle publication, +agent spawn, or redispatch. A second dispatcher observes the record and cannot +spawn. A failed or ambiguous record write leaves an incomplete claim channel +and blocks subsequent dispatch, favoring an operator-visible stranded claim +over duplicate real work. + +The manifest `statePath` remains a local receipt cache for reconciliation and +migration. It is no longer the authority. Agent Relay message idempotency was +not used as the claim primitive because it is sender-scoped and time-bounded; +workspace-global channel uniqueness supplies the required durable exclusion. + +## 2. Lifecycle reconciliation to Notion + +**Assessment:** deliberate boundary worth keeping now, not a defect to paper +over in Factory. + +Relayfile currently presents the Notion page as a read-only execution contract. +Factory has guarded writeback adapters and lifecycle mappings for Linear, +Slack, and GitHub, but no Notion property/comment writeback contract. A native +Notion lifecycle would also need an operator-selected mapping for accepted, +dispatched, PR opened, blocked, and completed states. Inventing that mapping or +assuming a writable provider behind a read-only mount would weaken the existing +fail-closed model. + +**Recommendation:** keep the page immutable and reconcile repository work on +the generated GitHub issue. Treat native Notion lifecycle as a separate product +decision requiring a real provider capability, guarded acknowledgement/readback +semantics, and a configured database/property mapping. + +## 3. `issueSource` and routing-contract visibility + +**Assessment:** terminology/documentation ambiguity; do not add `notion` to the +enum yet. + +`issueSource` selects the orchestrator's discovery and lifecycle-writeback +adapter. Notion intake is intentionally an upstream normalization step, not an +adapter implementing that contract. Accepting `issueSource: "notion"` today +would promise discovery, triage, state transitions, comments, and terminal +reconciliation that do not exist. + +**Recommendation:** retain `linear | github`, explicitly define the field as a +lifecycle issue source, and document that `factory intake notion` normalizes +into GitHub lifecycle or exact-path fleet work. A future native Notion adapter +can extend the enum only when it satisfies the whole lifecycle port. + +## Contract resolution + +Factory resolves exactly one config: the path supplied with `--config`, or +`./factory.config.json` in the command's current working directory. It does not +search the target repository, walk to a clone root, or merge configs. Notion +intake does not create an implicit fallback layer. + +## Proof and current operational limitation + +Automated coverage proves both repository publication and exact-path spawn are +preceded by the shared claim, claim-write failure prevents the external action, +and two manifests with independent local state paths produce only one issue or +one spawn. + +Chief also has prior live evidence: the mounted Notion benchmark page produced +agent `notion-9a84f582-8fc3bc47` on `kjg-laptop`, with a durable fleet invocation +record and a portable Relay delivery receipt. Repository pages produced labeled +GitHub lifecycle issues (for example Cloud #2935 and Relay #1433). + +A fresh production dispatch was intentionally not created during this change: +every page in Chief's active manifest already has a receipt, so another task +would be duplicate real work. `relayfile status` reports the Notion provider as +lagging with no sync cursor/watermark and the Chief snapshots are dated August +5. The remote page is currently readable and its digest matches the local +snapshot checked, but that does not establish fresh provider ingress. A new-page +live proof remains blocked until a fresh Notion sync watermark is observable. diff --git a/src/__tests__/dist-entrypoints.test.ts b/src/__tests__/dist-entrypoints.test.ts index d44f6b0..d8b2d10 100644 --- a/src/__tests__/dist-entrypoints.test.ts +++ b/src/__tests__/dist-entrypoints.test.ts @@ -11,6 +11,7 @@ describe('published dist entrypoints', () => { const featureGuardian = await import('../../dist/feature-guardian/index.js') const main = await import('../../dist/index.js') const hosted = await import('../../dist/hosted/index.js') + const intake = await import('../../dist/intake/index.js') const environments = await import('../../dist/environments/index.js') const testing = await import('../../dist/testing/index.js') const writeback = await import('../../dist/writeback/index.js') @@ -29,6 +30,8 @@ describe('published dist entrypoints', () => { expect(main.KubernetesEnvironmentProvider).toBeTypeOf('function') expect(hosted.createHostedFactory).toBeTypeOf('function') expect(hosted.DurableObjectHostedFactoryStateStore).toBeTypeOf('function') + expect(intake.RelayChannelNotionClaimStore).toBeTypeOf('function') + expect(intake.runNotionIntake).toBeTypeOf('function') expect(environments.KubernetesEnvironmentProvider).toBeTypeOf('function') expect(environments.KubectlEnvironmentProvider).toBeTypeOf('function') expect(environments.VerificationPipeline).toBeTypeOf('function') diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 17558e1..87bc889 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -591,9 +591,21 @@ describe('fleet CLI runtime', () => { await writeFile(manifestPath, JSON.stringify(manifest)) const output = buffer() const fleet = new FakeFleetClient() + const durableClaims = new Map() + const notionClaims = { + get: vi.fn(async (sourceKey: string) => durableClaims.get(sourceKey)), + claim: vi.fn(async (claim: { sourceKey: string; digest: string; claimedAt: string }) => { + const existing = durableClaims.get(claim.sourceKey) + if (existing) return { status: 'existing' as const, claim: existing } + durableClaims.set(claim.sourceKey, claim) + return { status: 'claimed' as const, claim } + }), + dispose: vi.fn(async () => undefined), + } const code = await runFleetCli(['intake', 'notion', manifestPath], { fleet, + notionClaims, stdout: output, stderr: buffer(), }) @@ -622,6 +634,7 @@ describe('fleet CLI runtime', () => { const migratedErrors = buffer() const migratedCode = await runFleetCli(['intake', 'notion', manifestPath], { fleet, + notionClaims, notionContracts: contracts, env: {}, stdout: migratedOutput, @@ -636,11 +649,42 @@ describe('fleet CLI runtime', () => { mode: 'steer', })]) expect(contracts.dispose).toHaveBeenCalledOnce() + expect(notionClaims.dispose).toHaveBeenCalledTimes(2) expect(migratedErrors.text()).toContain('Notion contract publisher failed during shutdown') expect(JSON.parse(migratedOutput.text())).toMatchObject({ ok: true, results: [{ status: 'already-dispatched', target: { projectPath } }], }) + + await writeFile(join(mountedPage, 'content.md'), [ + '# Chief Spec', + 'Status: ready', + 'Title: Changed exact-path dispatch', + 'Summary: This digest must not reuse the durable claim.', + 'Recipe: single', + `Project-Paths: ${projectPath}`, + ].join('\n')) + await writeFile(manifestPath, JSON.stringify({ + ...manifest, + statePath: './changed-state.json', + })) + const changedOutput = buffer() + const changedCode = await runFleetCli(['intake', 'notion', manifestPath], { + fleet, + notionClaims, + stdout: changedOutput, + stderr: buffer(), + }) + + expect(changedCode).toBe(1) + expect(fleet.spawns).toHaveLength(1) + expect(JSON.parse(changedOutput.text())).toMatchObject({ + ok: false, + results: [{ + status: 'blocked', + reason: 'durable Notion claim digest does not match the mounted spec', + }], + }) } finally { await rm(root, { recursive: true, force: true }) } @@ -676,7 +720,7 @@ describe('fleet CLI runtime', () => { }) expect(code).toBe(1) - expect(errors.text()).toContain('requires an active Agent Relay workspace') + expect(errors.text()).toContain('requires an active Agent Relay workspace for its durable shared claim') } finally { await rm(root, { recursive: true, force: true }) } diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index fd6fc12..89d5f67 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -73,9 +73,11 @@ import { MountAuthScopeError } from '../mount/mount-auth-error' import { resolveRelayWorkspaceKey } from '../fleet/relay-workspace-key' import { GhCliIssuePublisher, + RelayChannelNotionClaimStore, RelayChannelNotionContractPublisher, loadNotionIntakeManifest, runNotionIntake, + type NotionIntakeClaimStore, type NotionContractPublisher, type WorkspaceTaskDispatcher, } from '../intake' @@ -116,6 +118,8 @@ interface FleetCliDeps { featureMapCheck?: (options?: CheckFeatureMapOptions) => Promise /** Hermetic portable Notion contract publisher for intake tests and alternate runtimes. */ notionContracts?: NotionContractPublisher + /** Hermetic workspace-global Notion claim store for tests and alternate runtimes. */ + notionClaims?: NotionIntakeClaimStore /** Hermetic verification-environment sweep for CLI tests and alternate runtimes. */ reapEnvironments?: typeof reapFactoryEnvironmentsOnce } @@ -157,6 +161,7 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom let mount: MountClient | undefined let reporter: FactoryEventReporter | undefined let notionContracts: NotionContractPublisher | undefined + let notionClaims: NotionIntakeClaimStore | undefined try { if (argv.some(isHelpFlag)) { @@ -186,20 +191,31 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom if (command.kind === 'notion-intake') { const manifest = await loadNotionIntakeManifest(command.manifestPath) - if (!globals.dryRun && manifest.workerMountTransport.kind === 'relay-channel') { + if (!globals.dryRun) { + notionClaims = deps.notionClaims notionContracts = deps.notionContracts - if (!notionContracts) { - const workspaceKey = resolveRelayWorkspaceKey({ + const needsWorkspaceKey = !notionClaims || + (manifest.workerMountTransport.kind === 'relay-channel' && !notionContracts) + const workspaceKey = needsWorkspaceKey + ? resolveRelayWorkspaceKey({ env: deps.env ?? process.env, ...(deps.env ? { activeWorkspaceKey: () => undefined } : {}), }) - if (!workspaceKey) { - throw new Error('relay-channel worker mount transport requires an active Agent Relay workspace') - } - notionContracts = new RelayChannelNotionContractPublisher({ workspaceKey }) + : undefined + if (needsWorkspaceKey && !workspaceKey) { + throw new Error('Notion dispatch requires an active Agent Relay workspace for its durable shared claim') + } + notionClaims ??= new RelayChannelNotionClaimStore({ workspaceKey: workspaceKey! }) + if (manifest.workerMountTransport.kind === 'relay-channel') { + notionContracts ??= new RelayChannelNotionContractPublisher({ workspaceKey: workspaceKey! }) } } const workspace: WorkspaceTaskDispatcher = { + find: async (name) => { + fleet ??= await buildFleet(globals, undefined, deps) + const running = (await fleet.roster()).agents.find((agent) => agent.name === name) + return running ? { agent: running.name, node: running.node } : undefined + }, dispatch: async (task) => { fleet ??= await buildFleet(globals, undefined, deps) const running = (await fleet.roster()).agents.find((agent) => agent.name === task.name) @@ -240,6 +256,7 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom ...(!globals.dryRun ? { github: new GhCliIssuePublisher(), workspace, + ...(notionClaims ? { claims: notionClaims } : {}), ...(notionContracts ? { contracts: notionContracts } : {}), } : {}), }) @@ -448,30 +465,38 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom } finally { try { try { - await notionContracts?.dispose?.() + await notionClaims?.dispose?.() } catch { - err.write('[factory] warning: Notion contract publisher failed during shutdown\n') + err.write('[factory] warning: Notion claim store failed during shutdown\n') } } finally { try { - await mount?.dispose?.() + try { + await notionContracts?.dispose?.() + } catch { + err.write('[factory] warning: Notion contract publisher failed during shutdown\n') + } } finally { try { - await fleet?.dispose() + await mount?.dispose?.() } finally { - if (reporter) { - try { - await reporter.report(createFactoryCloudEventV1({ - type: 'instance.stopping', - attributes: { component: 'cli', operation: 'stop' }, - })) - await reporter.report(createFactoryCloudEventV1({ - type: 'instance.stopped', - attributes: { component: 'cli', operation: 'stop' }, - })) - await reporter.close?.({ deadlineMs: 2_000 }) - } catch { - err.write('[factory] warning: Cloud progress reporter failed during shutdown\n') + try { + await fleet?.dispose() + } finally { + if (reporter) { + try { + await reporter.report(createFactoryCloudEventV1({ + type: 'instance.stopping', + attributes: { component: 'cli', operation: 'stop' }, + })) + await reporter.report(createFactoryCloudEventV1({ + type: 'instance.stopped', + attributes: { component: 'cli', operation: 'stop' }, + })) + await reporter.close?.({ deadlineMs: 2_000 }) + } catch { + err.write('[factory] warning: Cloud progress reporter failed during shutdown\n') + } } } } diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index d99450a..273cb83 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -237,6 +237,13 @@ describe('FactoryConfigSchema', () => { expect(github.issueSource).toBe('github') }) + it('keeps Notion on the separate intake path instead of accepting it as a lifecycle issue source', () => { + expect(() => FactoryConfigSchema.parse({ + issueSource: 'notion', + repos: { default: 'AgentWorkforce/factory' }, + })).toThrow() + }) + it.each(['app', 'user', 'auto'] as const)('accepts github.identity %s', (identity) => { const parsed = FactoryConfigSchema.parse({ repos: { default: 'AgentWorkforce/factory' }, diff --git a/src/config/schema.ts b/src/config/schema.ts index 81c2dbf..55f5b16 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -244,9 +244,12 @@ const WorkspaceConfigObjectSchema = z.object({ models: modelsSchema, agentCapabilities: agentCapabilitiesSchema, slack: slackSchema, - // Linear remains the default whenever its issue sub-root is connected. When - // omitted, the orchestrator probes that sub-root once and falls back to - // GitHub-native issue lifecycle handling only when Linear is absent. + // This selects the discovery + lifecycle-writeback adapter, not every intake + // surface. Notion deliberately remains a separate intake command that + // normalizes repository work into GitHub lifecycle issues. Linear remains + // the default whenever its issue sub-root is connected. When omitted, the + // orchestrator probes that sub-root once and falls back to GitHub-native + // issue lifecycle handling only when Linear is absent. issueSource: z.enum(['linear', 'github']).optional(), mergePolicy: z.enum(['never', 'on-green-with-review']).default('never'), // Opt-in PR babysitter. When enabled, a sonnet agent is spawned once the diff --git a/src/intake/index.ts b/src/intake/index.ts index b8bf8c4..1da4e62 100644 --- a/src/intake/index.ts +++ b/src/intake/index.ts @@ -8,6 +8,8 @@ export { type GithubIssuePublisher, type ExistingGithubIssue, type NotionIntakeManifest, + type NotionIntakeClaim, + type NotionIntakeClaimStore, type NotionIntakeReport, type NotionIntakeResult, type NotionIntakeTarget, @@ -18,6 +20,11 @@ export { type WorkspaceTaskDispatcher, } from './notion' +export { + RelayChannelNotionClaimStore, + notionClaimChannelName, +} from './notion-relay-claim' + export { RelayChannelNotionContractPublisher, contractChannelName, diff --git a/src/intake/notion-relay-claim.test.ts b/src/intake/notion-relay-claim.test.ts new file mode 100644 index 0000000..b65752c --- /dev/null +++ b/src/intake/notion-relay-claim.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it, vi } from 'vitest' + +import { RelayChannelNotionClaimStore, notionClaimChannelName } from './notion-relay-claim' + +type StoredMessage = { id: string; text: string } + +function fakeRelaySurface(options: { failWrites?: boolean } = {}) { + const channels = new Map() + let registrations = 0 + let sends = 0 + const createRelay = vi.fn((relayOptions: { agentToken?: string }) => ({ + agents: { + register: vi.fn(async () => ({ token: `agent-token-${++registrations}` })), + delete: vi.fn(async () => undefined), + }, + channels: { + get: vi.fn(async (name: string) => { + if (!channels.has(name)) throw Object.assign(new Error('missing'), { code: 'channel_not_found' }) + return { name } + }), + join: vi.fn(async (name: string) => { + if (!channels.has(name)) throw new Error('channel not found') + }), + create: vi.fn(async ({ name }: { name: string }) => { + if (channels.has(name)) throw Object.assign(new Error('exists'), { code: 'channel_already_exists' }) + channels.set(name, []) + return { name } + }), + }, + messages: { + list: vi.fn(async (channel: string) => [...(channels.get(channel) ?? [])].reverse()), + send: vi.fn(async ({ channel, text }: { channel: string; text: string }) => { + if (options.failWrites) throw new Error('claim write unavailable') + sends += 1 + const message = { id: `message-${sends}`, text } + channels.get(channel)?.push(message) + return message + }), + }, + messaging: { events: { disconnect: vi.fn(async () => undefined) } }, + relayOptions, + }) as never) + return { channels, createRelay, sendCount: () => sends } +} + +const claim = { + sourceKey: 'notion:3b36800c-1c90-801d-b1cf-c8f2e1cff7cf:repo:agentworkforce/cloud', + digest: 'a'.repeat(64), + claimedAt: '2026-08-06T20:00:00.000Z', +} + +describe('RelayChannelNotionClaimStore', () => { + it('enforces the workspace-uniqueness contract across dispatcher instances', async () => { + const fake = fakeRelaySurface() + const first = new RelayChannelNotionClaimStore({ + workspaceKey: 'workspace-key', + publisherName: 'dispatcher-one', + createRelay: fake.createRelay, + }) + const second = new RelayChannelNotionClaimStore({ + workspaceKey: 'workspace-key', + publisherName: 'dispatcher-two', + createRelay: fake.createRelay, + }) + + const results = await Promise.all([ + first.claim(claim), + second.claim({ ...claim, claimedAt: '2026-08-06T20:01:00.000Z' }), + ]) + const winner = results.find((result) => result.status === 'claimed')! + const observer = results.find((result) => result.status === 'existing')! + + expect(results.map((result) => result.status).sort()).toEqual(['claimed', 'existing']) + expect(observer.claim).toEqual(winner.claim) + expect(fake.channels.get(notionClaimChannelName(claim.sourceKey))).toHaveLength(1) + expect(fake.sendCount()).toBe(1) + }) + + it('reads an existing durable claim without trying to create another', async () => { + const fake = fakeRelaySurface() + const first = new RelayChannelNotionClaimStore({ workspaceKey: 'workspace-key', createRelay: fake.createRelay }) + const second = new RelayChannelNotionClaimStore({ workspaceKey: 'workspace-key', createRelay: fake.createRelay }) + await first.claim(claim) + + await expect(second.get(claim.sourceKey)).resolves.toEqual(claim) + await expect(second.get('notion:missing')).resolves.toBeUndefined() + expect(fake.sendCount()).toBe(1) + }) + + it('leaves an incomplete durable channel and rejects when the claim record write fails', async () => { + const fake = fakeRelaySurface({ failWrites: true }) + const first = new RelayChannelNotionClaimStore({ workspaceKey: 'workspace-key', createRelay: fake.createRelay }) + const second = new RelayChannelNotionClaimStore({ workspaceKey: 'workspace-key', createRelay: fake.createRelay }) + + await expect(first.claim(claim)).rejects.toThrow('claim write unavailable') + await expect(second.claim(claim)).rejects.toThrow('has 0 immutable claim records; refusing dispatch') + expect(fake.channels.has(notionClaimChannelName(claim.sourceKey))).toBe(true) + }) +}) diff --git a/src/intake/notion-relay-claim.ts b/src/intake/notion-relay-claim.ts new file mode 100644 index 0000000..db7a02e --- /dev/null +++ b/src/intake/notion-relay-claim.ts @@ -0,0 +1,201 @@ +import { createHash } from 'node:crypto' + +import { AgentRelay } from '@agent-relay/sdk' +import type { RelayMessage } from '@agent-relay/sdk' +import { z } from 'zod' + +import type { NotionIntakeClaim, NotionIntakeClaimStore } from './notion' + +const CLAIM_MARKER = '---FACTORY NOTION INTAKE CLAIM V1---' + +const claimRecordSchema = z.object({ + version: z.literal(1), + sourceKey: z.string().min(1), + digest: z.string().regex(/^[0-9a-f]{64}$/u), + claimedAt: z.string().datetime(), +}).strict() + +type RelayChannelClaimStoreOptions = { + workspaceKey: string + baseUrl?: string + publisherName?: string + createRelay?: (options: { workspaceKey: string; baseUrl?: string; agentToken?: string }) => ClaimRelay +} + +type ClaimRelay = Pick + +/** + * Stores one immutable claim per Notion source key in a workspace-global Relay + * channel. Channel-name uniqueness is the cross-dispatcher compare-and-set; + * message idempotency is deliberately not used because it is actor-scoped and + * expires after a bounded interval. + */ +export class RelayChannelNotionClaimStore implements NotionIntakeClaimStore { + readonly #workspaceKey: string + readonly #baseUrl?: string + readonly #publisherName: string + readonly #createRelay: NonNullable + #workspaceRelay?: ClaimRelay + #agentRelay?: ClaimRelay + #relayReady?: Promise + #disposed = false + + constructor(options: RelayChannelClaimStoreOptions) { + this.#workspaceKey = options.workspaceKey + this.#baseUrl = options.baseUrl + this.#publisherName = options.publisherName ?? + `factory-notion-claims-${process.pid}-${Date.now().toString(36)}` + this.#createRelay = options.createRelay ?? ((relayOptions) => new AgentRelay(relayOptions)) + } + + async claim(input: NotionIntakeClaim): Promise<{ + status: 'claimed' | 'existing' + claim: NotionIntakeClaim + }> { + if (this.#disposed) throw new Error('Notion claim store has been disposed') + const record = claimRecordSchema.parse({ version: 1, ...input }) + const claim = publicClaim(record) + const relay = await this.#relay() + const channel = notionClaimChannelName(claim.sourceKey) + + try { + await relay.channels.create({ + name: channel, + topic: `Immutable Factory Notion claim ${claim.digest}`, + }) + } catch (createError) { + try { + await relay.channels.join(channel) + } catch (joinError) { + throw new Error(`unable to create or observe durable Notion claim ${channel}`, { + cause: new AggregateError([createError, joinError]), + }) + } + return { + status: 'existing', + claim: await readExistingClaim(relay, channel, claim.sourceKey), + } + } + + const text = renderClaim(record) + const written = await relay.messages.send({ channel, text }) + if (written.text !== text) { + throw new Error(`durable Notion claim acknowledgement did not match ${channel}`) + } + return { status: 'claimed', claim } + } + + async get(sourceKey: string): Promise { + if (this.#disposed) throw new Error('Notion claim store has been disposed') + const relay = await this.#relay() + const channel = notionClaimChannelName(sourceKey) + try { + await relay.channels.get(channel) + } catch (error) { + if (isMissingChannel(error)) return undefined + throw error + } + await relay.channels.join(channel) + return await readExistingClaim(relay, channel, sourceKey) + } + + async dispose(): Promise { + this.#disposed = true + await this.#relayReady?.catch(() => undefined) + await this.#agentRelay?.messaging.events.disconnect().catch(() => undefined) + await this.#workspaceRelay?.agents.delete(this.#publisherName).catch(() => undefined) + this.#agentRelay = undefined + this.#workspaceRelay = undefined + this.#relayReady = undefined + } + + async #relay(): Promise { + if (this.#agentRelay) return this.#agentRelay + this.#relayReady ??= this.#initializeRelay() + try { + return await this.#relayReady + } catch (error) { + this.#relayReady = undefined + throw error + } + } + + async #initializeRelay(): Promise { + const options = { + workspaceKey: this.#workspaceKey, + ...(this.#baseUrl ? { baseUrl: this.#baseUrl } : {}), + } + const workspaceRelay = this.#createRelay(options) + const registration = await workspaceRelay.agents.register({ + name: this.#publisherName, + type: 'system', + }) + if (this.#disposed) { + await workspaceRelay.agents.delete(this.#publisherName).catch(() => undefined) + throw new Error('Notion claim store was disposed during initialization') + } + this.#workspaceRelay = workspaceRelay + this.#agentRelay = this.#createRelay({ ...options, agentToken: registration.token }) + return this.#agentRelay + } +} + +function isMissingChannel(error: unknown): boolean { + if (!error || typeof error !== 'object') return false + const record = error as { code?: unknown; status?: unknown } + return record.code === 'channel_not_found' || record.status === 404 +} + +export function notionClaimChannelName(sourceKey: string): string { + const suffix = createHash('sha256').update(sourceKey).digest('hex') + return `factory-notion-claim-${suffix}` +} + +function renderClaim(claim: z.infer): string { + return `${CLAIM_MARKER}\n${JSON.stringify(claim)}` +} + +function publicClaim(record: z.infer): NotionIntakeClaim { + return { + sourceKey: record.sourceKey, + digest: record.digest, + claimedAt: record.claimedAt, + } +} + +async function readExistingClaim( + relay: ClaimRelay, + channel: string, + expectedSourceKey: string, +): Promise { + const messages = await listAllMessages(relay, channel) + const records = messages + .filter((message) => message.text.startsWith(`${CLAIM_MARKER}\n`)) + .map((message) => claimRecordSchema.parse(JSON.parse(message.text.slice(CLAIM_MARKER.length + 1)))) + if (records.length !== 1) { + throw new Error( + `durable Notion claim ${channel} has ${records.length} immutable claim records; refusing dispatch`, + ) + } + const [claim] = records + if (claim!.sourceKey !== expectedSourceKey) { + throw new Error(`durable Notion claim ${channel} does not match its source key`) + } + return publicClaim(claim!) +} + +async function listAllMessages(relay: ClaimRelay, channel: string): Promise { + const messages: RelayMessage[] = [] + let before: string | undefined + for (let pageNumber = 0; pageNumber < 100; pageNumber += 1) { + const page = await relay.messages.list(channel, { limit: 100, ...(before ? { before } : {}) }) + messages.push(...page) + if (page.length < 100) return messages + const nextBefore = page.at(-1)?.id + if (!nextBefore || nextBefore === before) { + throw new Error('durable Notion claim message pagination did not advance') + } + before = nextBefore + } + throw new Error('durable Notion claim channel exceeds the 10,000-message safety limit') +} diff --git a/src/intake/notion.test.ts b/src/intake/notion.test.ts index 88cf35b..422d5de 100644 --- a/src/intake/notion.test.ts +++ b/src/intake/notion.test.ts @@ -11,6 +11,7 @@ import { parseChiefSpecHeader, runNotionIntake, type GithubIssuePublisher, + type NotionIntakeClaimStore, type NotionIntakeManifest, type NotionIntakeTarget, type NotionContractPublisher, @@ -19,8 +20,21 @@ import { const pageId = '3b36800c-1c90-801d-b1cf-c8f2e1cff7cf' const roots: string[] = [] +const durableClaims = new Map() +const claims: NotionIntakeClaimStore = { + get: vi.fn(async (sourceKey) => durableClaims.get(sourceKey)), + claim: vi.fn(async (claim) => { + const existing = durableClaims.get(claim.sourceKey) + if (existing) return { status: 'existing' as const, claim: existing } + durableClaims.set(claim.sourceKey, claim) + return { status: 'claimed' as const, claim } + }), +} afterEach(async () => { + durableClaims.clear() + vi.mocked(claims.get).mockClear() + vi.mocked(claims.claim).mockClear() await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))) }) @@ -123,7 +137,7 @@ describe('Notion spec intake', () => { roots.push(root) const github = fakeGithub({ visibility: 'public' }) - const report = await runNotionIntake({ manifest, dispatch: true, github }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github }) expect(report.ok).toBe(false) expect(report.results[0]).toMatchObject({ status: 'blocked', reason: expect.stringContaining('publicSummary') }) @@ -141,7 +155,7 @@ describe('Notion spec intake', () => { manifest.workerMountRoot = 'specs/notion' const github = fakeGithub({ visibility: 'public' }) - const first = await runNotionIntake({ manifest, dispatch: true, github }) + const first = await runNotionIntake({ manifest, dispatch: true, claims, github }) const body = vi.mocked(github.createIssue).mock.calls[0]![0].body expect(first.results[0]).toMatchObject({ status: 'dispatched', issue: { number: 42 } }) expect(body).toContain('Resolve the already-scoped fleet reliability follow-ups.') @@ -161,7 +175,7 @@ describe('Notion spec intake', () => { }) vi.mocked(github.findBySource).mockResolvedValue({ number: 42, url: 'https://github.test/issues/42', body }) - const second = await runNotionIntake({ manifest, dispatch: true, github }) + const second = await runNotionIntake({ manifest, dispatch: true, claims, github }) expect(second.results[0]).toMatchObject({ status: 'already-dispatched', issue: { number: 42 } }) expect(github.createIssue).toHaveBeenCalledTimes(1) }) @@ -176,7 +190,7 @@ describe('Notion spec intake', () => { roots.push(root) const github = fakeGithub({ visibility: 'private' }) - await runNotionIntake({ manifest, dispatch: true, github }) + await runNotionIntake({ manifest, dispatch: true, claims, github }) expect(vi.mocked(github.createIssue).mock.calls[0]![0].title).toBe('[factory] Resume the checkpoint') }) @@ -197,7 +211,7 @@ describe('Notion spec intake', () => { })), } - const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github, contracts }) expect(report.results[0]).toMatchObject({ status: 'dispatched' }) expect(contracts.publish).toHaveBeenCalledWith(expect.objectContaining({ @@ -225,7 +239,7 @@ describe('Notion spec intake', () => { }) roots.push(root) const github = fakeGithub({ visibility: 'private' }) - await runNotionIntake({ manifest, dispatch: true, github }) + await runNotionIntake({ manifest, dispatch: true, claims, github }) const originalBody = vi.mocked(github.createIssue).mock.calls[0]![0].body vi.mocked(github.findBySource).mockResolvedValue({ number: 42, @@ -242,7 +256,7 @@ describe('Notion spec intake', () => { })), } - const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github, contracts }) expect(report.results[0]).toMatchObject({ status: 'already-dispatched', issue: { number: 42 } }) expect(github.createIssue).toHaveBeenCalledTimes(1) @@ -266,7 +280,7 @@ describe('Notion spec intake', () => { }) roots.push(root) const github = fakeGithub({ visibility: 'private' }) - await runNotionIntake({ manifest, dispatch: true, github }) + await runNotionIntake({ manifest, dispatch: true, claims, github }) const originalBody = vi.mocked(github.createIssue).mock.calls[0]![0].body vi.mocked(github.findBySource).mockResolvedValue({ number: 42, @@ -282,7 +296,7 @@ describe('Notion spec intake', () => { encoding: 'base64-chunks-v1', })), } - await runNotionIntake({ manifest, dispatch: true, github, contracts }) + await runNotionIntake({ manifest, dispatch: true, claims, github, contracts }) const migratedBody = vi.mocked(github.updateIssue).mock.calls[0]![0].body const interruptedState = JSON.parse(await readFile(manifest.statePath, 'utf8')) delete interruptedState.receipts[`notion:${pageId}:repo:agentworkforce/cloud`].delivery @@ -293,7 +307,7 @@ describe('Notion spec intake', () => { body: migratedBody, }) - const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github, contracts }) expect(report.results[0]).toMatchObject({ status: 'already-dispatched', issue: { number: 42 } }) expect(github.updateIssue).toHaveBeenCalledTimes(1) @@ -307,7 +321,7 @@ describe('Notion spec intake', () => { }) roots.push(root) const github = fakeGithub({ visibility: 'private' }) - await runNotionIntake({ manifest, dispatch: true, github }) + await runNotionIntake({ manifest, dispatch: true, claims, github }) const originalBody = vi.mocked(github.createIssue).mock.calls[0]![0].body vi.mocked(github.findBySource).mockResolvedValue({ number: 42, @@ -324,7 +338,7 @@ describe('Notion spec intake', () => { })), } - const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github, contracts }) expect(report.results[0]).toMatchObject({ status: 'blocked', @@ -342,7 +356,7 @@ describe('Notion spec intake', () => { manifest.workerMountTransport = { kind: 'relay-channel' } const github = fakeGithub({ visibility: 'private' }) - const report = await runNotionIntake({ manifest, dispatch: true, github }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github }) expect(report.results[0]).toMatchObject({ status: 'blocked', @@ -367,7 +381,7 @@ describe('Notion spec intake', () => { })), } - const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github, contracts }) expect(report.results[0]).toMatchObject({ status: 'blocked', @@ -392,7 +406,7 @@ describe('Notion spec intake', () => { })), } - const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github, contracts }) expect(report.results[0]).toMatchObject({ status: 'blocked', @@ -401,27 +415,79 @@ describe('Notion spec intake', () => { expect(github.createIssue).not.toHaveBeenCalled() }) - it('blocks a source marker that has no authoritative intake receipt', async () => { + it('blocks a source marker that has neither a durable claim nor a local migration receipt', async () => { const { root, manifest } = await fixtureManifest('private mounted body', { bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), }) roots.push(root) const github = fakeGithub({ visibility: 'private' }) + const [task] = await normalizeNotionManifest(manifest) vi.mocked(github.findBySource).mockResolvedValue({ number: 99, url: 'https://github.test/issues/99', - body: `Source digest: \`${'a'.repeat(64)}\`\n`, + body: `Source digest: \`${task!.digest}\`\n`, }) - const report = await runNotionIntake({ manifest, dispatch: true, github }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github }) expect(report.results[0]).toMatchObject({ status: 'blocked', - reason: 'lifecycle issue marker has no authoritative local receipt', + reason: 'lifecycle issue marker has neither a durable shared claim nor a local migration receipt', + }) + expect(github.createIssue).not.toHaveBeenCalled() + }) + + it('requires a durable claim acknowledgement before creating a lifecycle issue', async () => { + const { root, manifest } = await fixtureManifest('private mounted body', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + const github = fakeGithub({ visibility: 'private' }) + const unavailableClaims: NotionIntakeClaimStore = { + get: vi.fn(async () => undefined), + claim: vi.fn(async () => { throw new Error('shared claim write failed') }), + } + + const report = await runNotionIntake({ + manifest, + dispatch: true, + claims: unavailableClaims, + github, }) + + expect(report.results[0]).toMatchObject({ status: 'blocked', reason: 'shared claim write failed' }) expect(github.createIssue).not.toHaveBeenCalled() }) + it('prevents two machines with independent local state from publishing the same page twice', async () => { + const firstFixture = await fixtureManifest('private mounted body', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(firstFixture.root) + const secondStatePath = join(firstFixture.root, 'second-machine', 'state.json') + const secondManifest = { ...firstFixture.manifest, statePath: secondStatePath } + const github = fakeGithub({ visibility: 'private' }) + vi.mocked(github.createIssue).mockImplementation(async () => { + await new Promise((resolvePromise) => setTimeout(resolvePromise, 25)) + return { number: 42, url: 'https://github.test/issues/42' } + }) + + const reports = await Promise.all([ + runNotionIntake({ manifest: firstFixture.manifest, dispatch: true, claims, github }), + runNotionIntake({ manifest: secondManifest, dispatch: true, claims, github }), + ]) + + expect(github.createIssue).toHaveBeenCalledTimes(1) + expect(reports.flatMap((report) => report.results).map((result) => result.status).sort()).toEqual([ + 'blocked', + 'dispatched', + ]) + expect(reports.flatMap((report) => report.results)).toContainEqual(expect.objectContaining({ + status: 'blocked', + reason: expect.stringContaining('durable Notion claim already exists'), + })) + }) + it('serializes overlapping runs and creates one lifecycle issue', async () => { const { root, manifest } = await fixtureManifest('private mounted body', { bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), @@ -439,8 +505,8 @@ describe('Notion spec intake', () => { }) const reports = await Promise.all([ - runNotionIntake({ manifest, dispatch: true, github }), - runNotionIntake({ manifest, dispatch: true, github }), + runNotionIntake({ manifest, dispatch: true, claims, github }), + runNotionIntake({ manifest, dispatch: true, claims, github }), ]) expect(github.createIssue).toHaveBeenCalledTimes(1) @@ -458,7 +524,7 @@ describe('Notion spec intake', () => { const github = fakeGithub({ visibility: 'private' }) vi.mocked(github.missingLabels).mockResolvedValue(['reviewed']) - const report = await runNotionIntake({ manifest, dispatch: true, github }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github }) expect(report.results[0]).toMatchObject({ status: 'blocked', @@ -479,7 +545,7 @@ describe('Notion spec intake', () => { const first = await runNotionIntake({ manifest, - dispatch: true, + dispatch: true, claims, workspace, now: () => new Date('2026-08-05T22:00:00.000Z'), }) @@ -497,7 +563,7 @@ describe('Notion spec intake', () => { dispatchedAt: '2026-08-05T22:00:00.000Z', }) - const second = await runNotionIntake({ manifest, dispatch: true, workspace }) + const second = await runNotionIntake({ manifest, dispatch: true, claims, workspace }) expect(second.results[0]).toMatchObject({ status: 'already-dispatched', agent: 'benchmark-agent' }) expect(workspace.dispatch).toHaveBeenCalledTimes(1) @@ -512,7 +578,7 @@ describe('Notion spec intake', () => { } const migrated = await runNotionIntake({ manifest, - dispatch: true, + dispatch: true, claims, workspace, contracts, now: () => new Date('2026-08-05T23:00:00.000Z'), @@ -546,7 +612,7 @@ describe('Notion spec intake', () => { join(manifest.mountRoot, 'pages', pageId, 'content.md'), 'workspace body changed after dispatch', ) - const changed = await runNotionIntake({ manifest, dispatch: true, workspace, contracts }) + const changed = await runNotionIntake({ manifest, dispatch: true, claims, workspace, contracts }) expect(changed.results[0]).toMatchObject({ status: 'blocked', agent: 'benchmark-agent', @@ -556,6 +622,200 @@ describe('Notion spec intake', () => { expect(workspace.redispatch).toHaveBeenCalledTimes(1) }) + it('writes the shared claim before spawning exact-path work and aborts when that write fails', async () => { + const { root, manifest } = await fixtureManifest('workspace body', { + bootstrap: bootstrap({ projectPath: '/work/benchmark', node: 'kjg-laptop' }), + }) + roots.push(root) + const events: string[] = [] + const unavailableClaims: NotionIntakeClaimStore = { + get: vi.fn(async () => undefined), + claim: vi.fn(async () => { + events.push('claim') + throw new Error('durable claim unavailable') + }), + } + const workspace: WorkspaceTaskDispatcher = { + dispatch: vi.fn(async () => { + events.push('spawn') + return { agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' } + }), + } + + const report = await runNotionIntake({ + manifest, + dispatch: true, + claims: unavailableClaims, + workspace, + }) + + expect(report.results[0]).toMatchObject({ status: 'blocked', reason: 'durable claim unavailable' }) + expect(events).toEqual(['claim']) + expect(workspace.dispatch).not.toHaveBeenCalled() + }) + + it('prevents independent machines from spawning the same exact-path task twice', async () => { + const firstFixture = await fixtureManifest('workspace body', { + bootstrap: bootstrap({ projectPath: '/work/benchmark', node: 'kjg-laptop' }), + }) + roots.push(firstFixture.root) + const secondManifest = { + ...firstFixture.manifest, + statePath: join(firstFixture.root, 'second-machine', 'state.json'), + } + const workspace: WorkspaceTaskDispatcher = { + find: vi.fn(async () => undefined), + dispatch: vi.fn(async () => { + await new Promise((resolvePromise) => setTimeout(resolvePromise, 25)) + return { agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' } + }), + } + + const reports = await Promise.all([ + runNotionIntake({ manifest: firstFixture.manifest, dispatch: true, claims, workspace }), + runNotionIntake({ manifest: secondManifest, dispatch: true, claims, workspace }), + ]) + + expect(workspace.dispatch).toHaveBeenCalledTimes(1) + expect(reports.flatMap((report) => report.results).map((result) => result.status).sort()).toEqual([ + 'blocked', + 'dispatched', + ]) + expect(reports.flatMap((report) => report.results)).toContainEqual(expect.objectContaining({ + status: 'blocked', + reason: 'durable Notion claim already exists but no running workspace agent was found; refusing a second spawn', + })) + }) + + it('reports a missing workspace lookup capability separately from a missing agent', async () => { + const firstFixture = await fixtureManifest('workspace body', { + bootstrap: bootstrap({ projectPath: '/work/benchmark', node: 'kjg-laptop' }), + }) + roots.push(firstFixture.root) + const workspace: WorkspaceTaskDispatcher = { + dispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' })), + } + await runNotionIntake({ manifest: firstFixture.manifest, dispatch: true, claims, workspace }) + + const secondManifest = { + ...firstFixture.manifest, + statePath: join(firstFixture.root, 'second-machine', 'state.json'), + } + const report = await runNotionIntake({ manifest: secondManifest, dispatch: true, claims, workspace }) + + expect(report.results[0]).toMatchObject({ + status: 'blocked', + reason: 'durable Notion claim already exists but this workspace dispatcher cannot look up running agents; refusing a second spawn', + }) + expect(workspace.dispatch).toHaveBeenCalledTimes(1) + }) + + it('uses a durable migration claim to prevent two old receipts from redispatching one worker', async () => { + const firstFixture = await fixtureManifest('workspace body', { + bootstrap: bootstrap({ projectPath: '/work/benchmark', node: 'kjg-laptop' }), + }) + roots.push(firstFixture.root) + const workspace: WorkspaceTaskDispatcher = { + dispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' })), + redispatch: vi.fn(async () => { + await new Promise((resolvePromise) => setTimeout(resolvePromise, 25)) + return { agent: 'benchmark-agent', node: 'kjg-laptop', status: 'respawned' } + }), + } + await runNotionIntake({ manifest: firstFixture.manifest, dispatch: true, claims, workspace }) + const secondStatePath = join(firstFixture.root, 'second-machine', 'state.json') + await mkdir(join(firstFixture.root, 'second-machine'), { recursive: true }) + await writeFile(secondStatePath, await readFile(firstFixture.manifest.statePath, 'utf8')) + firstFixture.manifest.workerMountTransport = { kind: 'relay-channel' } + const secondManifest = { ...firstFixture.manifest, statePath: secondStatePath } + const contracts: NotionContractPublisher = { + publish: vi.fn(async () => ({ + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1'], + encoding: 'base64-chunks-v1', + })), + } + + const reports = await Promise.all([ + runNotionIntake({ manifest: firstFixture.manifest, dispatch: true, claims, workspace, contracts }), + runNotionIntake({ manifest: secondManifest, dispatch: true, claims, workspace, contracts }), + ]) + + expect(workspace.redispatch).toHaveBeenCalledTimes(1) + expect(contracts.publish).toHaveBeenCalledTimes(1) + expect(reports.flatMap((report) => report.results).map((result) => result.status).sort()).toEqual([ + 'already-dispatched', + 'blocked', + ]) + expect(reports.flatMap((report) => report.results)).toContainEqual(expect.objectContaining({ + status: 'blocked', + reason: 'durable portable-mount migration claim already exists; refusing a second workspace redispatch', + })) + const storedStates = await Promise.all([ + firstFixture.manifest.statePath, + secondStatePath, + ].map(async (statePath) => JSON.parse(await readFile(statePath, 'utf8')))) + expect(storedStates + .map((stored) => stored.receipts[`notion:${pageId}:workspace:/work/benchmark`]?.delivery) + .filter(Boolean)).toEqual([{ + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1'], + encoding: 'base64-chunks-v1', + }]) + }) + + it('does not treat an unresolved portable migration claim as a completed receipt on another machine', async () => { + const firstFixture = await fixtureManifest('workspace body', { + bootstrap: bootstrap({ projectPath: '/work/benchmark', node: 'kjg-laptop' }), + }) + roots.push(firstFixture.root) + const workspace: WorkspaceTaskDispatcher = { + find: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop' })), + dispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' })), + redispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'respawned' })), + } + await runNotionIntake({ manifest: firstFixture.manifest, dispatch: true, claims, workspace }) + const firstState = JSON.parse(await readFile(firstFixture.manifest.statePath, 'utf8')) + const sourceKey = `notion:${pageId}:workspace:/work/benchmark` + const primaryClaim = durableClaims.get(sourceKey) + expect(primaryClaim).toBeDefined() + durableClaims.set(`${sourceKey}:portable-mount`, { + ...primaryClaim!, + sourceKey: `${sourceKey}:portable-mount`, + }) + + const secondStatePath = join(firstFixture.root, 'second-machine', 'state.json') + firstFixture.manifest.workerMountTransport = { kind: 'relay-channel' } + const contracts: NotionContractPublisher = { + publish: vi.fn(async () => ({ + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1'], + encoding: 'base64-chunks-v1', + })), + } + + const report = await runNotionIntake({ + manifest: { ...firstFixture.manifest, statePath: secondStatePath }, + dispatch: true, + claims, + workspace, + contracts, + }) + + expect(firstState.receipts[sourceKey].delivery).toBeUndefined() + expect(report.results[0]).toMatchObject({ + status: 'blocked', + agent: 'benchmark-agent', + reason: 'durable portable-mount migration claim exists without a local completion receipt; refusing to assume the running worker was refreshed', + }) + expect(contracts.publish).not.toHaveBeenCalled() + expect(workspace.redispatch).not.toHaveBeenCalled() + await expect(readFile(secondStatePath, 'utf8')).rejects.toMatchObject({ code: 'ENOENT' }) + }) + it('blocks an exact-path portable migration when the existing worker cannot be refreshed', async () => { const { root, manifest } = await fixtureManifest('workspace body', { bootstrap: bootstrap({ projectPath: '/work/benchmark', node: 'kjg-laptop' }), @@ -564,7 +824,7 @@ describe('Notion spec intake', () => { const workspace: WorkspaceTaskDispatcher = { dispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' })), } - await runNotionIntake({ manifest, dispatch: true, workspace }) + await runNotionIntake({ manifest, dispatch: true, claims, workspace }) manifest.workerMountTransport = { kind: 'relay-channel' } const contracts: NotionContractPublisher = { publish: vi.fn(async () => ({ @@ -575,7 +835,7 @@ describe('Notion spec intake', () => { })), } - const report = await runNotionIntake({ manifest, dispatch: true, workspace, contracts }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, workspace, contracts }) expect(report.results[0]).toMatchObject({ status: 'blocked', @@ -601,7 +861,7 @@ describe('Notion spec intake', () => { .mockResolvedValueOnce(undefined) .mockRejectedValueOnce(new Error('GitHub unavailable')) - const report = await runNotionIntake({ manifest, dispatch: true, github }) + const report = await runNotionIntake({ manifest, dispatch: true, claims, github }) expect(report.ok).toBe(false) expect(report.results).toEqual([ diff --git a/src/intake/notion.ts b/src/intake/notion.ts index 775e51e..f8dc7f7 100644 --- a/src/intake/notion.ts +++ b/src/intake/notion.ts @@ -117,7 +117,23 @@ export interface NotionContractPublisher { dispose?(): Promise } +export interface NotionIntakeClaim { + sourceKey: string + digest: string + claimedAt: string +} + +export interface NotionIntakeClaimStore { + get(sourceKey: string): Promise + claim(input: NotionIntakeClaim): Promise<{ + status: 'claimed' | 'existing' + claim: NotionIntakeClaim + }> + dispose?(): Promise +} + export interface WorkspaceTaskDispatcher { + find?(name: string): Promise<{ agent: string; node?: string } | undefined> dispatch(input: { name: string invocationId: string @@ -193,6 +209,7 @@ export async function runNotionIntake(input: { github?: GithubIssuePublisher workspace?: WorkspaceTaskDispatcher contracts?: NotionContractPublisher + claims?: NotionIntakeClaimStore now?: () => Date }): Promise { if (!input.dispatch) return await runNotionIntakeUnlocked(input) @@ -419,11 +436,8 @@ async function publishRepoTask( const existing = await input.github.findBySource(target.repo, task.sourceKey) if (existing) { - if (!receipt) { - return { ...base, status: 'blocked', issue: existing, reason: 'lifecycle issue marker has no authoritative local receipt' } - } - if (existing.number !== receipt.issue.number || existing.url !== receipt.issue.url) { - return { ...base, status: 'blocked', issue: existing, reason: 'lifecycle issue does not match the authoritative local receipt' } + if (receipt && (existing.number !== receipt.issue.number || existing.url !== receipt.issue.url)) { + return { ...base, status: 'blocked', issue: existing, reason: 'lifecycle issue does not match the local receipt cache' } } const currentDigest = digestFromBody(existing.body) if (!currentDigest) { @@ -432,11 +446,29 @@ async function publishRepoTask( if (currentDigest !== task.digest) { return { ...base, status: 'blocked', issue: existing, reason: 'mounted spec changed after the lifecycle issue was created' } } + let claim = await observeNotionClaim(task, input) + if (!claim) { + if (!receipt) { + return { + ...base, + status: 'blocked', + issue: existing, + reason: 'lifecycle issue marker has neither a durable shared claim nor a local migration receipt', + } + } + claim = (await claimNotionTask(task, input)).claim + } const bodyDelivery = contractDeliveryFromBody(existing.body) if (input.manifest.workerMountTransport.kind === 'local') { - if (receipt.delivery || bodyDelivery) { + if (receipt?.delivery || bodyDelivery) { return { ...base, status: 'blocked', issue: existing, reason: 'portable Notion delivery cannot be downgraded to a local worker mount' } } + state.receipts[task.sourceKey] = receipt ?? { + kind: 'github', + digest: task.digest, + issue: { number: existing.number, url: existing.url }, + dispatchedAt: claim.claimedAt, + } return { ...base, status: 'already-dispatched', issue: existing } } const visibility = await input.github.repositoryVisibility(target.repo) @@ -453,24 +485,29 @@ async function publishRepoTask( reason: 'existing lifecycle issue body was edited; refusing to overwrite it during portable mount migration', } } - if (receipt.delivery && bodyDelivery && !sameContractDelivery(receipt.delivery, bodyDelivery)) { - return { ...base, status: 'blocked', issue: existing, reason: 'lifecycle issue portable delivery does not match its authoritative receipt' } + if (receipt?.delivery && bodyDelivery && !sameContractDelivery(receipt.delivery, bodyDelivery)) { + return { ...base, status: 'blocked', issue: existing, reason: 'lifecycle issue portable delivery does not match its local receipt cache' } } - const delivery = await prepareContractDelivery(task, input, receipt.delivery ?? bodyDelivery) + const delivery = await prepareContractDelivery(task, input, receipt?.delivery ?? bodyDelivery) if (delivery && !bodyDelivery) { + await assertMountedTaskUnchanged(task) await input.github.updateIssue({ repo: target.repo, number: existing.number, body: renderIssueBody(task, summary, delivery), }) } - if (delivery && !sameContractDelivery(receipt.delivery, delivery)) { - state.receipts[task.sourceKey] = { ...receipt, delivery } + state.receipts[task.sourceKey] = { + kind: 'github', + digest: task.digest, + issue: { number: existing.number, url: existing.url }, + ...(delivery ? { delivery } : {}), + dispatchedAt: receipt?.dispatchedAt ?? claim.claimedAt, } return { ...base, status: 'already-dispatched', issue: existing } } if (receipt) { - return { ...base, status: 'blocked', issue: receipt.issue, reason: 'authoritative lifecycle receipt exists but its issue marker was not found' } + return { ...base, status: 'blocked', issue: receipt.issue, reason: 'local lifecycle receipt exists but its issue marker was not found' } } const visibility = await input.github.repositoryVisibility(target.repo) @@ -483,6 +520,15 @@ async function publishRepoTask( return { ...base, status: 'blocked', reason: `missing required GitHub labels: ${missing.join(', ')}` } } const delivery = await prepareContractDelivery(task, input) + const claim = await claimNotionTask(task, input) + if (claim.status === 'existing') { + return { + ...base, + status: 'blocked', + reason: 'durable Notion claim already exists; refusing lifecycle issue creation from this dispatcher', + } + } + await assertMountedTaskUnchanged(task) const issue = await input.github.createIssue({ repo: target.repo, title: factoryIssueTitle(task.title), @@ -494,7 +540,7 @@ async function publishRepoTask( digest: task.digest, issue, ...(delivery ? { delivery } : {}), - dispatchedAt: (input.now?.() ?? new Date()).toISOString(), + dispatchedAt: claim.claim.claimedAt, } return { ...base, status: 'dispatched', issue } } @@ -525,6 +571,29 @@ async function dispatchWorkspaceTask( if (receipt.digest !== task.digest) { return { ...base, status: 'blocked', agent: receipt.agent, node: receipt.node, reason: 'mounted spec changed after workspace dispatch' } } + await claimNotionTask(task, input) + const needsPortableMigration = input.manifest.workerMountTransport.kind !== 'local' && !receipt.delivery + if (needsPortableMigration) { + if (!input.workspace?.redispatch) { + return { + ...base, + status: 'blocked', + agent: receipt.agent, + node: receipt.node, + reason: 'portable workspace mount migration requires a workspace redispatcher', + } + } + const migrationClaim = await claimNotionTask(task, input, `${task.sourceKey}:portable-mount`) + if (migrationClaim.status === 'existing') { + return { + ...base, + status: 'blocked', + agent: receipt.agent, + node: receipt.node, + reason: 'durable portable-mount migration claim already exists; refusing a second workspace redispatch', + } + } + } const delivery = await prepareContractDelivery(task, input, receipt.delivery) if (delivery && !sameContractDelivery(receipt.delivery, delivery)) { if (!input.workspace?.redispatch) { @@ -557,10 +626,49 @@ async function dispatchWorkspaceTask( } if (!input.workspace) return { ...base, status: 'blocked', reason: 'workspace task dispatcher is not configured' } - const delivery = await prepareContractDelivery(task, input) - const suffix = createHash('sha256').update(task.sourceKey).digest('hex').slice(0, 8) const name = `notion-${task.pageId.slice(-8)}-${suffix}` + const claim = await claimNotionTask(task, input) + if (claim.status === 'existing') { + if (!input.workspace.find) { + return { + ...base, + status: 'blocked', + reason: 'durable Notion claim already exists but this workspace dispatcher cannot look up running agents; refusing a second spawn', + } + } + const running = await input.workspace.find(name) + if (!running) { + return { + ...base, + status: 'blocked', + reason: 'durable Notion claim already exists but no running workspace agent was found; refusing a second spawn', + } + } + const migrationSourceKey = `${task.sourceKey}:portable-mount` + if (input.manifest.workerMountTransport.kind !== 'local' && + await observeNotionClaim(task, input, migrationSourceKey)) { + return { + ...base, + status: 'blocked', + agent: running.agent, + node: running.node, + reason: 'durable portable-mount migration claim exists without a local completion receipt; refusing to assume the running worker was refreshed', + } + } + const delivery = await prepareContractDelivery(task, input) + state.receipts[task.sourceKey] = { + kind: 'workspace', + digest: task.digest, + agent: running.agent, + ...(running.node ? { node: running.node } : {}), + ...(delivery ? { delivery } : {}), + dispatchedAt: claim.claim.claimedAt, + } + return { ...base, status: 'already-dispatched', agent: running.agent, node: running.node } + } + const delivery = await prepareContractDelivery(task, input) + await assertMountedTaskUnchanged(task) const result = await input.workspace.dispatch({ name, invocationId: `factory:${task.sourceKey}:${task.digest}`, @@ -575,7 +683,7 @@ async function dispatchWorkspaceTask( agent: result.agent, ...(result.node ? { node: result.node } : {}), ...(delivery ? { delivery } : {}), - dispatchedAt: (input.now?.() ?? new Date()).toISOString(), + dispatchedAt: claim.claim.claimedAt, } return { ...base, @@ -585,6 +693,46 @@ async function dispatchWorkspaceTask( } } +async function observeNotionClaim( + task: NormalizedNotionTask, + input: Parameters[0], + sourceKey = task.sourceKey, +): Promise { + if (!input.claims) { + throw new Error('dispatch requires a durable Agent Relay Notion claim store') + } + const claim = await input.claims.get(sourceKey) + if (claim?.sourceKey !== undefined && claim.sourceKey !== sourceKey) { + throw new Error('durable Notion claim does not match the requested source key') + } + if (claim?.digest !== undefined && claim.digest !== task.digest) { + throw new Error('durable Notion claim digest does not match the mounted spec') + } + return claim +} + +async function claimNotionTask( + task: NormalizedNotionTask, + input: Parameters[0], + sourceKey = task.sourceKey, +): Promise<{ status: 'claimed' | 'existing'; claim: NotionIntakeClaim }> { + if (!input.claims) { + throw new Error('dispatch requires a durable Agent Relay Notion claim store') + } + const result = await input.claims.claim({ + sourceKey, + digest: task.digest, + claimedAt: (input.now?.() ?? new Date()).toISOString(), + }) + if (result.claim.sourceKey !== sourceKey) { + throw new Error('durable Notion claim does not match the requested source key') + } + if (result.claim.digest !== task.digest) { + throw new Error('durable Notion claim digest does not match the mounted spec') + } + return result +} + function normalizedBootstrapSpec(bootstrap: z.infer, pageId: string) { const authorizedPageId = normalizeNotionPageId(bootstrap.authorizedPageId) if (authorizedPageId !== pageId) {