diff --git a/README.md b/README.md index 47d7a7e..5ebb9a5 100644 --- a/README.md +++ b/README.md @@ -198,6 +198,7 @@ hatch is deliberately page-specific; there is no title or content heuristic. "version": 1, "mountRoot": ".integrations/notion", "workerMountRoot": ".integrations/notion", + "workerMountTransport": { "kind": "relay-channel" }, "statePath": ".factory/notion-intake-state.json", "tasks": [ { "page": "https://app.notion.com/p/Reconcile-3b36800c1c90801db1cfc8f2e1cff7cf" } @@ -206,8 +207,15 @@ 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; all three default to the -values shown above. `page` accepts a Notion URL or a bare page ID. +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. +`page` accepts a Notion URL or a bare page ID. Plan without writes, then dispatch: diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index c5ea817..17558e1 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -582,12 +582,13 @@ describe('fleet CLI runtime', () => { `Project-Paths: ${projectPath}`, ].join('\n')) const manifestPath = join(root, 'notion.json') - await writeFile(manifestPath, JSON.stringify({ + const manifest = { version: 1, mountRoot: './notion', statePath: './state.json', tasks: [{ page: '3b36800c1c90801db1cfc8f2e1cff7cf' }], - })) + } + await writeFile(manifestPath, JSON.stringify(manifest)) const output = buffer() const fleet = new FakeFleetClient() @@ -603,6 +604,79 @@ describe('fleet CLI runtime', () => { ok: true, results: [{ status: 'dispatched', target: { projectPath } }], }) + + await writeFile(manifestPath, JSON.stringify({ + ...manifest, + workerMountTransport: { kind: 'relay-channel' }, + })) + const contracts = { + publish: vi.fn(async () => ({ + kind: 'relay-channel' as const, + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1'], + encoding: 'base64-chunks-v1' as const, + })), + dispose: vi.fn(async () => { throw new Error('cleanup failed') }), + } + const migratedOutput = buffer() + const migratedErrors = buffer() + const migratedCode = await runFleetCli(['intake', 'notion', manifestPath], { + fleet, + notionContracts: contracts, + env: {}, + stdout: migratedOutput, + stderr: migratedErrors, + }) + + expect(migratedCode).toBe(0) + expect(fleet.spawns).toHaveLength(1) + expect(fleet.messages).toEqual([expect.objectContaining({ + to: expect.stringContaining('notion-e1cff7cf'), + text: expect.stringContaining('factory-notion-e1cff7cf-aabbccddee'), + mode: 'steer', + })]) + expect(contracts.dispose).toHaveBeenCalledOnce() + expect(migratedErrors.text()).toContain('Notion contract publisher failed during shutdown') + expect(JSON.parse(migratedOutput.text())).toMatchObject({ + ok: true, + results: [{ status: 'already-dispatched', target: { projectPath } }], + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('does not borrow an ambient workspace when an explicit intake environment has no key', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-cli-notion-no-key-')) + try { + const mountedPage = join(root, 'notion', 'pages', '3b36800c-1c90-801d-b1cf-c8f2e1cff7cf') + await mkdir(mountedPage, { recursive: true }) + await writeFile(join(mountedPage, 'content.md'), [ + '# Chief Spec', + 'Status: ready', + 'Title: Verify isolated credentials', + 'Summary: Refuse ambient workspace credentials.', + 'Recipe: single', + 'Repos: AgentWorkforce/cloud', + ].join('\n')) + const manifestPath = join(root, 'notion.json') + await writeFile(manifestPath, JSON.stringify({ + version: 1, + mountRoot: './notion', + workerMountTransport: { kind: 'relay-channel' }, + tasks: [{ page: '3b36800c1c90801db1cfc8f2e1cff7cf' }], + })) + const errors = buffer() + + const code = await runFleetCli(['intake', 'notion', manifestPath], { + env: {}, + createFleet: () => { throw new Error('missing key must fail before fleet construction') }, + stdout: buffer(), + stderr: errors, + }) + + expect(code).toBe(1) + expect(errors.text()).toContain('requires an active Agent Relay workspace') } finally { await rm(root, { recursive: true, force: true }) } diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 0cbba58..fd6fc12 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -70,10 +70,13 @@ import { import type { FactoryIntegrationProvider } from '../ports' import { checkMountStaleness } from '../mount/relayfile-binary' import { MountAuthScopeError } from '../mount/mount-auth-error' +import { resolveRelayWorkspaceKey } from '../fleet/relay-workspace-key' import { GhCliIssuePublisher, + RelayChannelNotionContractPublisher, loadNotionIntakeManifest, runNotionIntake, + type NotionContractPublisher, type WorkspaceTaskDispatcher, } from '../intake' @@ -111,6 +114,8 @@ interface FleetCliDeps { confirmIntegrationConnect?: (provider: FactoryIntegrationProvider) => Promise openIntegrationUrl?: (url: string) => void | Promise featureMapCheck?: (options?: CheckFeatureMapOptions) => Promise + /** Hermetic portable Notion contract publisher for intake tests and alternate runtimes. */ + notionContracts?: NotionContractPublisher /** Hermetic verification-environment sweep for CLI tests and alternate runtimes. */ reapEnvironments?: typeof reapFactoryEnvironmentsOnce } @@ -151,6 +156,7 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom let fleet: FleetClient | undefined let mount: MountClient | undefined let reporter: FactoryEventReporter | undefined + let notionContracts: NotionContractPublisher | undefined try { if (argv.some(isHelpFlag)) { @@ -180,6 +186,19 @@ 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') { + notionContracts = deps.notionContracts + if (!notionContracts) { + const workspaceKey = 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 }) + } + } const workspace: WorkspaceTaskDispatcher = { dispatch: async (task) => { fleet ??= await buildFleet(globals, undefined, deps) @@ -196,11 +215,33 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom fleet.preserveInfrastructureOnDispose?.() return { agent: spawned.name, node: spawned.node, status: 'spawned' } }, + redispatch: async (task) => { + fleet ??= await buildFleet(globals, undefined, deps) + const running = (await fleet.roster()).agents.find((agent) => agent.name === task.name) + if (running) { + await fleet.sendMessage({ to: `@${running.name}`, text: task.task, mode: 'steer' }) + return { agent: running.name, node: running.node, status: 'updated-running' } + } + const spawned = await fleet.spawn({ + name: task.name, + capability: 'spawn:codex', + node: task.node ?? 'self', + task: task.task, + cwd: task.projectPath, + invocationId: task.invocationId, + }) + fleet.preserveInfrastructureOnDispose?.() + return { agent: spawned.name, node: spawned.node, status: 'respawned' } + }, } const report = await runNotionIntake({ manifest, dispatch: !globals.dryRun, - ...(!globals.dryRun ? { github: new GhCliIssuePublisher(), workspace } : {}), + ...(!globals.dryRun ? { + github: new GhCliIssuePublisher(), + workspace, + ...(notionContracts ? { contracts: notionContracts } : {}), + } : {}), }) writeJson(out, report) return report.ok ? 0 : 1 @@ -406,24 +447,32 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom return 1 } 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/intake/index.ts b/src/intake/index.ts index c12499d..b8bf8c4 100644 --- a/src/intake/index.ts +++ b/src/intake/index.ts @@ -11,7 +11,15 @@ export { type NotionIntakeReport, type NotionIntakeResult, type NotionIntakeTarget, + type NotionContractDelivery, + type NotionContractPublisher, type NotionRecipe, type NormalizedNotionTask, type WorkspaceTaskDispatcher, } from './notion' + +export { + RelayChannelNotionContractPublisher, + contractChannelName, + contractMarkerPrefix, +} from './notion-relay-contract' diff --git a/src/intake/notion-relay-contract.test.ts b/src/intake/notion-relay-contract.test.ts new file mode 100644 index 0000000..e0349e2 --- /dev/null +++ b/src/intake/notion-relay-contract.test.ts @@ -0,0 +1,180 @@ +import { createHash } from 'node:crypto' + +import { describe, expect, it, vi } from 'vitest' + +import { RelayChannelNotionContractPublisher, contractChannelName, contractMarkerPrefix } from './notion-relay-contract' + +type StoredMessage = { id: string; text: string } + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((resolvePromise) => { resolve = resolvePromise }) + return { promise, resolve } +} + +function fakeRelaySurface() { + const channels = new Map() + let registrations = 0 + let sends = 0 + let nextMessage = 1 + const createRelay = vi.fn((options: { agentToken?: string }) => ({ + agents: { + register: vi.fn(async () => ({ token: `agent-token-${++registrations}` })), + delete: vi.fn(async () => undefined), + }, + channels: { + 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 new Error('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 }) => { + sends += 1 + const message = { id: `message-${nextMessage++}`, text } + channels.get(channel)?.push(message) + return message + }), + }, + messaging: { events: { disconnect: vi.fn(async () => undefined) } }, + options, + }) as never) + return { + channels, + createRelay, + registrationCount: () => registrations, + sendCount: () => sends, + } +} + +function contractInput(content: string, sourceKey = 'notion:page:repo:agentworkforce/cloud') { + return { + pageId: '3b36800c-1c90-801d-b1cf-c8f2e1cff7cf', + sourceKey, + content, + contentDigest: createHash('sha256').update(content).digest('hex'), + } +} + +describe('RelayChannelNotionContractPublisher', () => { + it('rejects a digest mismatch before creating a Relay client', async () => { + const fake = fakeRelaySurface() + const publisher = new RelayChannelNotionContractPublisher({ + workspaceKey: 'workspace-key', + createRelay: fake.createRelay, + }) + + await expect(publisher.publish({ + ...contractInput('private body'), + contentDigest: '0'.repeat(64), + })).rejects.toThrow('changed before portable mount publication') + expect(fake.createRelay).not.toHaveBeenCalled() + }) + + it('chunks exact bytes and reuses their message ids across publisher identities', async () => { + const fake = fakeRelaySurface() + const input = contractInput('private body '.repeat(600)) + const firstPublisher = new RelayChannelNotionContractPublisher({ + workspaceKey: 'workspace-key', + publisherName: 'publisher-one', + createRelay: fake.createRelay, + }) + const first = await firstPublisher.publish(input) + await firstPublisher.dispose() + const sendsAfterFirst = fake.sendCount() + const secondPublisher = new RelayChannelNotionContractPublisher({ + workspaceKey: 'workspace-key', + publisherName: 'publisher-two', + createRelay: fake.createRelay, + }) + const second = await secondPublisher.publish(input) + + expect(first.messageIds.length).toBeGreaterThan(1) + expect(second).toEqual(first) + expect(fake.sendCount()).toBe(sendsAfterFirst) + const byId = new Map((fake.channels.get(first.channel) ?? []).map((message) => [message.id, message.text])) + const encoded = first.messageIds.map((id) => byId.get(id)?.split('\n')[2]).join('') + expect(Buffer.from(encoded, 'base64').toString('utf8')).toBe(input.content) + }) + + it('uses a new digest channel and cache entry when one source changes', async () => { + const fake = fakeRelaySurface() + const publisher = new RelayChannelNotionContractPublisher({ + workspaceKey: 'workspace-key', + createRelay: fake.createRelay, + }) + const first = await publisher.publish(contractInput('first revision')) + const second = await publisher.publish(contractInput('second revision')) + + expect(second.channel).not.toBe(first.channel) + expect(second.messageIds).not.toEqual(first.messageIds) + expect(fake.sendCount()).toBe(2) + }) + + it('rejects an existing digest marker whose payload bytes differ', async () => { + const fake = fakeRelaySurface() + const input = contractInput('expected content') + const channel = contractChannelName(input.pageId, input.sourceKey, input.contentDigest) + fake.channels.set(channel, [{ + id: 'tampered-message', + text: `${contractMarkerPrefix(input.pageId, input.contentDigest)}1/1\n` + + '---BEGIN FACTORY NOTION CONTRACT BASE64---\ndGFtcGVyZWQ=\n---END FACTORY NOTION CONTRACT BASE64---', + }]) + const publisher = new RelayChannelNotionContractPublisher({ + workspaceKey: 'workspace-key', + createRelay: fake.createRelay, + }) + + await expect(publisher.publish(input)).rejects.toThrow('does not match its digest-bound marker') + expect(fake.sendCount()).toBe(0) + }) + + it('shares one in-flight agent registration across overlapping publications', async () => { + const fake = fakeRelaySurface() + const publisher = new RelayChannelNotionContractPublisher({ + workspaceKey: 'workspace-key', + createRelay: fake.createRelay, + }) + + await Promise.all([ + publisher.publish(contractInput('first body', 'notion:first')), + publisher.publish(contractInput('second body', 'notion:second')), + ]) + + expect(fake.registrationCount()).toBe(1) + }) + + it('cleans up an agent registration that completes during disposal', async () => { + const registration = deferred<{ token: string }>() + const deleteAgent = vi.fn(async () => undefined) + const createRelay = vi.fn((options: { agentToken?: string }) => ({ + agents: { + register: vi.fn(async () => registration.promise), + delete: deleteAgent, + }, + channels: { join: vi.fn(), create: vi.fn() }, + messages: { list: vi.fn(), send: vi.fn() }, + messaging: { events: { disconnect: vi.fn() } }, + options, + }) as never) + const publisher = new RelayChannelNotionContractPublisher({ + workspaceKey: 'workspace-key', + publisherName: 'disposing-publisher', + createRelay, + }) + + const publication = publisher.publish(contractInput('private body')) + const rejectedPublication = expect(publication).rejects.toThrow('disposed during initialization') + const disposal = publisher.dispose() + registration.resolve({ token: 'agent-token' }) + + await Promise.all([rejectedPublication, disposal]) + expect(createRelay).toHaveBeenCalledOnce() + expect(deleteAgent).toHaveBeenCalledWith('disposing-publisher') + }) +}) diff --git a/src/intake/notion-relay-contract.ts b/src/intake/notion-relay-contract.ts new file mode 100644 index 0000000..a8f7507 --- /dev/null +++ b/src/intake/notion-relay-contract.ts @@ -0,0 +1,187 @@ +import { createHash } from 'node:crypto' + +import { AgentRelay } from '@agent-relay/sdk' +import type { RelayMessage } from '@agent-relay/sdk' + +import type { NotionContractDelivery, NotionContractPublisher } from './notion' + +const CONTRACT_CHUNK_CHARACTERS = 6_000 +const CONTRACT_BEGIN = '---BEGIN FACTORY NOTION CONTRACT BASE64---' +const CONTRACT_END = '---END FACTORY NOTION CONTRACT BASE64---' + +type RelayChannelContractPublisherOptions = { + workspaceKey: string + baseUrl?: string + publisherName?: string + createRelay?: (options: { workspaceKey: string; baseUrl?: string; agentToken?: string }) => ContractRelay +} + +type ContractRelay = Pick + +/** + * Publishes digest-bound Notion bytes to a workspace-private Relay channel. + * Workers can reconstruct a read-only local mount snapshot on any fleet node + * without exposing the private page through a public lifecycle issue. + */ +export class RelayChannelNotionContractPublisher implements NotionContractPublisher { + readonly #workspaceKey: string + readonly #baseUrl?: string + readonly #publisherName: string + readonly #createRelay: NonNullable + readonly #cache = new Map() + #workspaceRelay?: ContractRelay + #agentRelay?: ContractRelay + #relayReady?: Promise + #disposed = false + + constructor(options: RelayChannelContractPublisherOptions) { + this.#workspaceKey = options.workspaceKey + this.#baseUrl = options.baseUrl + this.#publisherName = options.publisherName ?? + `factory-notion-intake-${process.pid}-${Date.now().toString(36)}` + this.#createRelay = options.createRelay ?? ((relayOptions) => new AgentRelay(relayOptions)) + } + + async publish(input: { + pageId: string + sourceKey: string + content: string + contentDigest: string + }): Promise { + if (this.#disposed) throw new Error('Notion contract publisher has been disposed') + const observedDigest = createHash('sha256').update(input.content).digest('hex') + if (observedDigest !== input.contentDigest) { + throw new Error('Notion contract changed before portable mount publication') + } + const cacheKey = `${input.sourceKey}\0${input.contentDigest}` + const cached = this.#cache.get(cacheKey) + if (cached) return cached + + const relay = await this.#relay() + const channel = contractChannelName(input.pageId, input.sourceKey, input.contentDigest) + try { + await relay.channels.join(channel) + } catch (joinError) { + try { + await relay.channels.create({ + name: channel, + topic: `Read-only Notion contract ${input.pageId}`, + }) + } catch (createError) { + try { + await relay.channels.join(channel) + } catch (finalJoinError) { + throw new Error(`unable to join or create Notion contract channel ${channel}`, { + cause: new AggregateError([joinError, createError, finalJoinError]), + }) + } + } + } + + const encoded = Buffer.from(input.content, 'utf8').toString('base64') + const chunks = splitContract(encoded) + const markerPrefix = contractMarkerPrefix(input.pageId, input.contentDigest) + const existing = await listAllMessages(relay, channel) + const messageIds: string[] = [] + + for (let index = 0; index < chunks.length; index += 1) { + const marker = `${markerPrefix}${index + 1}/${chunks.length}` + const expectedText = `${marker}\n${CONTRACT_BEGIN}\n${chunks[index]}\n${CONTRACT_END}` + const idempotencyKey = createHash('sha256') + .update(`${input.sourceKey}\0${input.contentDigest}\0${index + 1}\0${chunks.length}`) + .digest('hex') + const prior = existing.find((message) => message.text.startsWith(`${marker}\n`)) + const message = prior ?? await relay.messages.send({ + channel, + text: expectedText, + idempotencyKey: `factory-notion-contract-v1:${idempotencyKey}`, + }) + if (message.text !== expectedText) { + throw new Error(`portable Notion contract chunk ${index + 1} does not match its digest-bound marker`) + } + messageIds.push(message.id) + } + + const delivery: NotionContractDelivery = { + kind: 'relay-channel', + channel, + messageIds, + encoding: 'base64-chunks-v1', + } + this.#cache.set(cacheKey, delivery) + return delivery + } + + 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 contract publisher was disposed during initialization') + } + this.#workspaceRelay = workspaceRelay + this.#agentRelay = this.#createRelay({ ...options, agentToken: registration.token }) + return this.#agentRelay + } +} + +export function contractChannelName(pageId: string, sourceKey: string, contentDigest: string): string { + const sourceSuffix = createHash('sha256').update(sourceKey).digest('hex').slice(0, 8) + return `factory-notion-${pageId.slice(-8)}-${sourceSuffix}-${contentDigest.slice(0, 10)}` +} + +export function contractMarkerPrefix(pageId: string, contentDigest: string): string { + return `factory-notion-contract-v1:${pageId}:${contentDigest}:part:` +} + +function splitContract(encoded: string): string[] { + const chunks: string[] = [] + for (let offset = 0; offset < encoded.length; offset += CONTRACT_CHUNK_CHARACTERS) { + chunks.push(encoded.slice(offset, offset + CONTRACT_CHUNK_CHARACTERS)) + } + return chunks.length > 0 ? chunks : [''] +} + +async function listAllMessages(relay: ContractRelay, 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('portable Notion contract message pagination did not advance') + } + before = nextBefore + } + throw new Error('portable Notion contract digest channel exceeds the 10,000-message safety limit') +} diff --git a/src/intake/notion.test.ts b/src/intake/notion.test.ts index ad379ce..88cf35b 100644 --- a/src/intake/notion.test.ts +++ b/src/intake/notion.test.ts @@ -5,6 +5,7 @@ import { join } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' import { + loadNotionIntakeManifest, normalizeNotionManifest, normalizeNotionPageId, parseChiefSpecHeader, @@ -12,6 +13,7 @@ import { type GithubIssuePublisher, type NotionIntakeManifest, type NotionIntakeTarget, + type NotionContractPublisher, type WorkspaceTaskDispatcher, } from './notion' @@ -33,6 +35,20 @@ describe('Notion spec intake', () => { ) }) + it('defaults an omitted worker mount transport to local when loading existing manifests', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-notion-manifest-')) + roots.push(root) + const manifestPath = join(root, 'notion.json') + await writeFile(manifestPath, JSON.stringify({ + version: 1, + tasks: [{ page: pageId }], + })) + + const manifest = await loadNotionIntakeManifest(manifestPath) + + expect(manifest.workerMountTransport).toEqual({ kind: 'local' }) + }) + it('requires an explicit, ready Chief Spec header and parses both destination kinds', () => { expect(() => parseChiefSpecHeader('Continue this work')).toThrow('first line must be exactly "# Chief Spec"') expect(() => parseChiefSpecHeader('# Chief Spec\nStatus: draft\n')).toThrow('Status must be ready') @@ -165,6 +181,226 @@ describe('Notion spec intake', () => { expect(vi.mocked(github.createIssue).mock.calls[0]![0].title).toBe('[factory] Resume the checkpoint') }) + it('delivers private mounted bytes through a portable Relay channel without copying them to GitHub', async () => { + const { root, manifest } = await fixtureManifest('private mounted implementation detail', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + manifest.workerMountTransport = { kind: 'relay-channel' } + const github = fakeGithub({ visibility: 'private' }) + const contracts: NotionContractPublisher = { + publish: vi.fn(async () => ({ + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1', 'message-2'], + encoding: 'base64-chunks-v1', + })), + } + + const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + + expect(report.results[0]).toMatchObject({ status: 'dispatched' }) + expect(contracts.publish).toHaveBeenCalledWith(expect.objectContaining({ + pageId, + content: 'private mounted implementation detail', + })) + const body = vi.mocked(github.createIssue).mock.calls[0]![0].body + expect(body).toContain('workspace-private Agent Relay channel') + expect(body).toContain('factory-notion-e1cff7cf-aabbccddee') + expect(body).toContain('message-1,message-2') + expect(body).toContain('chmod the file 0444') + expect(body).not.toContain('private mounted implementation detail') + const stored = JSON.parse(await readFile(manifest.statePath, 'utf8')) + expect(stored.receipts[`notion:${pageId}:repo:agentworkforce/cloud`].delivery).toEqual({ + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1', 'message-2'], + encoding: 'base64-chunks-v1', + }) + }) + + it('migrates an untouched lifecycle issue to a portable mount without dispatching it again', async () => { + const { root, manifest } = await fixtureManifest('private mounted implementation detail', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + const github = fakeGithub({ visibility: 'private' }) + await runNotionIntake({ manifest, dispatch: true, github }) + const originalBody = vi.mocked(github.createIssue).mock.calls[0]![0].body + vi.mocked(github.findBySource).mockResolvedValue({ + number: 42, + url: 'https://github.test/issues/42', + body: originalBody, + }) + 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, dispatch: true, github, contracts }) + + expect(report.results[0]).toMatchObject({ status: 'already-dispatched', issue: { number: 42 } }) + expect(github.createIssue).toHaveBeenCalledTimes(1) + expect(github.updateIssue).toHaveBeenCalledWith(expect.objectContaining({ + repo: 'AgentWorkforce/cloud', + number: 42, + body: expect.stringContaining('factory-notion-e1cff7cf-aabbccddee'), + })) + const stored = JSON.parse(await readFile(manifest.statePath, 'utf8')) + expect(stored.receipts[`notion:${pageId}:repo:agentworkforce/cloud`].delivery).toEqual({ + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1'], + encoding: 'base64-chunks-v1', + }) + }) + + it('reconciles a portable issue marker when the receipt write was interrupted', async () => { + const { root, manifest } = await fixtureManifest('private mounted implementation detail', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + const github = fakeGithub({ visibility: 'private' }) + await runNotionIntake({ manifest, dispatch: true, github }) + const originalBody = vi.mocked(github.createIssue).mock.calls[0]![0].body + vi.mocked(github.findBySource).mockResolvedValue({ + number: 42, + url: 'https://github.test/issues/42', + body: originalBody, + }) + 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', + })), + } + await runNotionIntake({ manifest, dispatch: true, 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 + await writeFile(manifest.statePath, JSON.stringify(interruptedState)) + vi.mocked(github.findBySource).mockResolvedValue({ + number: 42, + url: 'https://github.test/issues/42', + body: migratedBody, + }) + + const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + + expect(report.results[0]).toMatchObject({ status: 'already-dispatched', issue: { number: 42 } }) + expect(github.updateIssue).toHaveBeenCalledTimes(1) + const reconciled = JSON.parse(await readFile(manifest.statePath, 'utf8')) + expect(reconciled.receipts[`notion:${pageId}:repo:agentworkforce/cloud`].delivery.messageIds).toEqual(['message-1']) + }) + + it('refuses to overwrite a manually edited lifecycle issue during portable mount migration', async () => { + const { root, manifest } = await fixtureManifest('private mounted implementation detail', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + const github = fakeGithub({ visibility: 'private' }) + await runNotionIntake({ manifest, dispatch: true, github }) + const originalBody = vi.mocked(github.createIssue).mock.calls[0]![0].body + vi.mocked(github.findBySource).mockResolvedValue({ + number: 42, + url: 'https://github.test/issues/42', + body: `${originalBody}\noperator note`, + }) + 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, dispatch: true, github, contracts }) + + expect(report.results[0]).toMatchObject({ + status: 'blocked', + reason: expect.stringContaining('refusing to overwrite'), + }) + expect(contracts.publish).not.toHaveBeenCalled() + expect(github.updateIssue).not.toHaveBeenCalled() + }) + + it('blocks portable dispatch when no Relay contract publisher is configured', async () => { + const { root, manifest } = await fixtureManifest('private mounted body', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + manifest.workerMountTransport = { kind: 'relay-channel' } + const github = fakeGithub({ visibility: 'private' }) + + const report = await runNotionIntake({ manifest, dispatch: true, github }) + + expect(report.results[0]).toMatchObject({ + status: 'blocked', + reason: expect.stringContaining('requires an Agent Relay contract publisher'), + }) + expect(github.createIssue).not.toHaveBeenCalled() + }) + + it('blocks a portable publisher response with no contract messages', async () => { + const { root, manifest } = await fixtureManifest('private mounted body', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + manifest.workerMountTransport = { kind: 'relay-channel' } + const github = fakeGithub({ visibility: 'private' }) + const contracts: NotionContractPublisher = { + publish: vi.fn(async () => ({ + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: [], + encoding: 'base64-chunks-v1', + })), + } + + const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + + expect(report.results[0]).toMatchObject({ + status: 'blocked', + reason: expect.stringContaining('at least one message id'), + }) + expect(github.createIssue).not.toHaveBeenCalled() + }) + + it('blocks portable delivery identifiers that cannot round-trip through an issue marker', async () => { + const { root, manifest } = await fixtureManifest('private mounted body', { + bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), + }) + roots.push(root) + manifest.workerMountTransport = { kind: 'relay-channel' } + const github = fakeGithub({ visibility: 'private' }) + const contracts: NotionContractPublisher = { + publish: vi.fn(async () => ({ + kind: 'relay-channel', + channel: 'factory-notion:unsafe', + messageIds: ['message-1,forged'], + encoding: 'base64-chunks-v1', + })), + } + + const report = await runNotionIntake({ manifest, dispatch: true, github, contracts }) + + expect(report.results[0]).toMatchObject({ + status: 'blocked', + reason: expect.stringContaining('marker-safe ASCII alphabet'), + }) + expect(github.createIssue).not.toHaveBeenCalled() + }) + it('blocks a source marker that has no authoritative intake receipt', async () => { const { root, manifest } = await fixtureManifest('private mounted body', { bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), @@ -238,6 +474,7 @@ describe('Notion spec intake', () => { roots.push(root) const workspace: WorkspaceTaskDispatcher = { dispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' })), + redispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'respawned' })), } const first = await runNotionIntake({ @@ -264,17 +501,88 @@ describe('Notion spec intake', () => { expect(second.results[0]).toMatchObject({ status: 'already-dispatched', agent: 'benchmark-agent' }) expect(workspace.dispatch).toHaveBeenCalledTimes(1) + 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 migrated = await runNotionIntake({ + manifest, + dispatch: true, + workspace, + contracts, + now: () => new Date('2026-08-05T23:00:00.000Z'), + }) + expect(migrated.results[0]).toMatchObject({ + status: 'already-dispatched', + agent: 'benchmark-agent', + node: 'kjg-laptop', + }) + expect(workspace.redispatch).toHaveBeenCalledWith(expect.objectContaining({ + name: 'benchmark-agent', + node: 'kjg-laptop', + task: expect.stringContaining('factory-notion-e1cff7cf-aabbccddee'), + })) + expect(workspace.dispatch).toHaveBeenCalledTimes(1) + expect(contracts.publish).toHaveBeenCalledOnce() + const migratedState = JSON.parse(await readFile(manifest.statePath, 'utf8')) + expect(migratedState.receipts[`notion:${pageId}:workspace:/work/benchmark`]).toMatchObject({ + kind: 'workspace', + agent: 'benchmark-agent', + dispatchedAt: '2026-08-05T23:00:00.000Z', + delivery: { + kind: 'relay-channel', + channel: 'factory-notion-e1cff7cf-aabbccddee', + messageIds: ['message-1'], + encoding: 'base64-chunks-v1', + }, + }) + await writeFile( join(manifest.mountRoot, 'pages', pageId, 'content.md'), 'workspace body changed after dispatch', ) - const changed = await runNotionIntake({ manifest, dispatch: true, workspace }) + const changed = await runNotionIntake({ manifest, dispatch: true, workspace, contracts }) expect(changed.results[0]).toMatchObject({ status: 'blocked', agent: 'benchmark-agent', reason: 'mounted spec changed after workspace dispatch', }) expect(workspace.dispatch).toHaveBeenCalledTimes(1) + expect(workspace.redispatch).toHaveBeenCalledTimes(1) + }) + + 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' }), + }) + roots.push(root) + const workspace: WorkspaceTaskDispatcher = { + dispatch: vi.fn(async () => ({ agent: 'benchmark-agent', node: 'kjg-laptop', status: 'spawned' })), + } + await runNotionIntake({ manifest, dispatch: true, workspace }) + 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, dispatch: true, workspace, contracts }) + + expect(report.results[0]).toMatchObject({ + status: 'blocked', + reason: 'portable workspace mount migration requires a workspace redispatcher', + }) + const stored = JSON.parse(await readFile(manifest.statePath, 'utf8')) + expect(stored.receipts[`notion:${pageId}:workspace:/work/benchmark`].delivery).toBeUndefined() }) it('retains per-destination results when a later publisher fails', async () => { @@ -317,6 +625,7 @@ async function fixtureManifest( version: 1, mountRoot, workerMountRoot: '.integrations/notion', + workerMountTransport: { kind: 'local' }, statePath: join(root, 'state.json'), tasks: [{ page: pageId, ...task }], }, @@ -343,5 +652,6 @@ function fakeGithub(input: { visibility: 'public' | 'private' | 'internal' }): G missingLabels: vi.fn(async () => []), findBySource: vi.fn(async () => undefined), createIssue: vi.fn(async () => ({ number: 42, url: 'https://github.test/issues/42' })), + updateIssue: vi.fn(async () => undefined), } } diff --git a/src/intake/notion.ts b/src/intake/notion.ts index fba7d5e..775e51e 100644 --- a/src/intake/notion.ts +++ b/src/intake/notion.ts @@ -23,6 +23,23 @@ const workspaceTargetSchema = z.object({ const targetSchema = z.union([repoTargetSchema, workspaceTargetSchema]) +const workerMountTransportSchema = z.discriminatedUnion('kind', [ + z.object({ kind: z.literal('local') }).strict(), + z.object({ kind: z.literal('relay-channel') }).strict(), +]).default({ kind: 'local' }) + +const contractMarkerTokenSchema = z.string().trim().min(1).regex( + /^[A-Za-z0-9._-]+$/u, + 'portable delivery identifiers must use the marker-safe ASCII alphabet', +) + +const contractDeliverySchema = z.object({ + kind: z.literal('relay-channel'), + channel: contractMarkerTokenSchema, + messageIds: z.array(contractMarkerTokenSchema).min(1, 'portable delivery must include at least one message id'), + encoding: z.literal('base64-chunks-v1'), +}).strict() + const bootstrapSchema = z.object({ authorizedPageId: z.string().trim().min(1), reason: z.string().trim().min(1), @@ -37,6 +54,7 @@ const manifestSchema = z.object({ version: z.literal(1), mountRoot: z.string().trim().min(1).default('.integrations/notion'), workerMountRoot: z.string().trim().min(1).default('.integrations/notion'), + workerMountTransport: workerMountTransportSchema, statePath: z.string().trim().min(1).default('.factory/notion-intake-state.json'), tasks: z.array(z.object({ page: z.string().trim().min(1), @@ -53,6 +71,7 @@ export interface NormalizedNotionTask { sourceKey: string sourcePath: string workerSourcePath: string + content: string authorizationDigestInput?: string contentDigest: string digest: string @@ -79,6 +98,23 @@ export interface GithubIssuePublisher { body: string labels: readonly string[] }): Promise<{ number: number; url: string }> + updateIssue(input: { + repo: string + number: number + body: string + }): Promise +} + +export type NotionContractDelivery = z.infer + +export interface NotionContractPublisher { + publish(input: { + pageId: string + sourceKey: string + content: string + contentDigest: string + }): Promise + dispose?(): Promise } export interface WorkspaceTaskDispatcher { @@ -90,6 +126,14 @@ export interface WorkspaceTaskDispatcher { title: string task: string }): Promise<{ agent: string; node?: string; status?: string }> + redispatch?(input: { + name: string + invocationId: string + node?: string + projectPath: string + title: string + task: string + }): Promise<{ agent: string; node?: string; status?: string }> } export type NotionIntakeResult = { @@ -114,12 +158,14 @@ type IntakeReceipt = { kind: 'github' digest: string issue: { number: number; url: string } + delivery?: NotionContractDelivery dispatchedAt: string } | { kind: 'workspace' digest: string agent: string node?: string + delivery?: NotionContractDelivery dispatchedAt: string } @@ -146,6 +192,7 @@ export async function runNotionIntake(input: { dispatch: boolean github?: GithubIssuePublisher workspace?: WorkspaceTaskDispatcher + contracts?: NotionContractPublisher now?: () => Date }): Promise { if (!input.dispatch) return await runNotionIntakeUnlocked(input) @@ -174,14 +221,19 @@ async function runNotionIntakeUnlocked( for (const task of tasks) { try { await assertMountedTaskUnchanged(task) - const hadReceipt = Boolean(state.receipts[task.sourceKey]) + const receiptBefore = state.receipts[task.sourceKey] + ? JSON.stringify(state.receipts[task.sourceKey]) + : undefined let result: NotionIntakeResult if ('repo' in task.target) { result = await publishRepoTask(task, input, state) } else { result = await dispatchWorkspaceTask(task, input, state) } - if (input.dispatch && !hadReceipt && state.receipts[task.sourceKey]) { + const receiptAfter = state.receipts[task.sourceKey] + ? JSON.stringify(state.receipts[task.sourceKey]) + : undefined + if (input.dispatch && receiptAfter && receiptAfter !== receiptBefore) { await writeIntakeState(input.manifest.statePath, state) } results.push(result) @@ -229,9 +281,10 @@ export async function normalizeNotionManifest(manifest: NotionIntakeManifest): P pageId, sourceKey, sourcePath, - workerSourcePath: 'repo' in target + workerSourcePath: 'repo' in target || manifest.workerMountTransport.kind === 'relay-channel' ? join(manifest.workerMountRoot, 'pages', pageId, 'content.md') : sourcePath, + content, ...(authorizationDigestInput ? { authorizationDigestInput } : {}), contentDigest, digest, @@ -334,6 +387,10 @@ export class GhCliIssuePublisher implements GithubIssuePublisher { if (!Number.isInteger(number) || number <= 0) throw new Error(`GitHub issue create returned an unexpected URL: ${url}`) return { number, url } } + + async updateIssue(input: { repo: string; number: number; body: string }): Promise { + await runGh(['issue', 'edit', String(input.number), '--repo', input.repo, '--body-file', '-'], input.body) + } } async function publishRepoTask( @@ -375,6 +432,41 @@ async function publishRepoTask( if (currentDigest !== task.digest) { return { ...base, status: 'blocked', issue: existing, reason: 'mounted spec changed after the lifecycle issue was created' } } + const bodyDelivery = contractDeliveryFromBody(existing.body) + if (input.manifest.workerMountTransport.kind === 'local') { + if (receipt.delivery || bodyDelivery) { + return { ...base, status: 'blocked', issue: existing, reason: 'portable Notion delivery cannot be downgraded to a local worker mount' } + } + return { ...base, status: 'already-dispatched', issue: existing } + } + const visibility = await input.github.repositoryVisibility(target.repo) + if (visibility === 'public' && !target.publicSummary) { + return { ...base, status: 'blocked', issue: existing, reason: 'public repository requires an explicit publicSummary; mounted content was not copied' } + } + const summary = visibility === 'public' ? target.publicSummary! : task.summary + const bodyWasEdited = existing.body !== renderIssueBody(task, summary, bodyDelivery) + if (bodyWasEdited) { + return { + ...base, + status: 'blocked', + issue: existing, + 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' } + } + const delivery = await prepareContractDelivery(task, input, receipt.delivery ?? bodyDelivery) + if (delivery && !bodyDelivery) { + 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 } + } return { ...base, status: 'already-dispatched', issue: existing } } if (receipt) { @@ -390,16 +482,18 @@ async function publishRepoTask( if (missing.length > 0) { return { ...base, status: 'blocked', reason: `missing required GitHub labels: ${missing.join(', ')}` } } + const delivery = await prepareContractDelivery(task, input) const issue = await input.github.createIssue({ repo: target.repo, title: factoryIssueTitle(task.title), labels, - body: renderIssueBody(task, visibility === 'public' ? target.publicSummary! : task.summary), + body: renderIssueBody(task, visibility === 'public' ? target.publicSummary! : task.summary, delivery), }) state.receipts[task.sourceKey] = { kind: 'github', digest: task.digest, issue, + ...(delivery ? { delivery } : {}), dispatchedAt: (input.now?.() ?? new Date()).toISOString(), } return { ...base, status: 'dispatched', issue } @@ -431,10 +525,40 @@ 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' } } - return { ...base, status: 'already-dispatched', agent: receipt.agent, node: receipt.node } + const delivery = await prepareContractDelivery(task, input, receipt.delivery) + if (delivery && !sameContractDelivery(receipt.delivery, delivery)) { + if (!input.workspace?.redispatch) { + return { + ...base, + status: 'blocked', + agent: receipt.agent, + node: receipt.node, + reason: 'portable workspace mount migration requires a workspace redispatcher', + } + } + const refreshed = await input.workspace.redispatch({ + name: receipt.agent, + invocationId: `factory:${task.sourceKey}:${task.digest}:portable-mount`, + node: receipt.node ?? target.node, + projectPath: target.projectPath, + title: task.title, + task: renderWorkspaceTask(task, delivery), + }) + state.receipts[task.sourceKey] = { + ...receipt, + agent: refreshed.agent, + ...(refreshed.node ? { node: refreshed.node } : {}), + delivery, + dispatchedAt: (input.now?.() ?? new Date()).toISOString(), + } + } + const currentReceipt = state.receipts[task.sourceKey] as Extract + return { ...base, status: 'already-dispatched', agent: currentReceipt.agent, node: currentReceipt.node } } 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 result = await input.workspace.dispatch({ @@ -443,13 +567,14 @@ async function dispatchWorkspaceTask( node: target.node, projectPath: target.projectPath, title: task.title, - task: renderWorkspaceTask(task), + task: renderWorkspaceTask(task, delivery), }) state.receipts[task.sourceKey] = { kind: 'workspace', digest: task.digest, agent: result.agent, ...(result.node ? { node: result.node } : {}), + ...(delivery ? { delivery } : {}), dispatchedAt: (input.now?.() ?? new Date()).toISOString(), } return { @@ -468,14 +593,17 @@ function normalizedBootstrapSpec(bootstrap: z.infer, pag return { ...bootstrap, authorizedPageId } } -function renderIssueBody(task: NormalizedNotionTask, summary: string): string { +function renderIssueBody( + task: NormalizedNotionTask, + summary: string, + delivery?: NotionContractDelivery, +): string { return [ '## Factory intake', '', summary, '', - 'The complete authorized spec is available to workers through the read-only Relayfile mount:', - `\`${task.workerSourcePath}\``, + ...renderWorkerMountInstructions(task, delivery), '', 'Treat the mounted page as the execution contract. Preserve every safety gate in it. Do not write back to Notion.', `Before executing, SHA-256 hash the mounted file's UTF-8 bytes and refuse the task unless it matches \`${task.contentDigest}\`.`, @@ -483,16 +611,17 @@ function renderIssueBody(task: NormalizedNotionTask, summary: string): string { `Source identity: \`notion:${task.pageId}\``, `Source digest: \`${task.digest}\``, sourceMarker(task.sourceKey), + ...(delivery ? [contractDeliveryMarker(delivery)] : []), ].join('\n') } -function renderWorkspaceTask(task: NormalizedNotionTask): string { +function renderWorkspaceTask(task: NormalizedNotionTask, delivery?: NotionContractDelivery): string { return [ task.title, '', task.summary, '', - `Read the full execution contract from the authorized read-only Notion mount at ${task.workerSourcePath}.`, + ...renderWorkerMountInstructions(task, delivery), `Before executing, SHA-256 hash that file's UTF-8 bytes and refuse the task unless it matches ${task.contentDigest}.`, 'Preserve every safety gate in that page. Do not write back to Notion.', `Factory source: ${task.sourceKey}`, @@ -500,6 +629,77 @@ function renderWorkspaceTask(task: NormalizedNotionTask): string { ].join('\n') } +async function prepareContractDelivery( + task: NormalizedNotionTask, + input: Parameters[0], + existing?: NotionContractDelivery, +): Promise { + if (input.manifest.workerMountTransport.kind === 'local') return undefined + if (!input.contracts) { + throw new Error('relay-channel worker mount transport requires an Agent Relay contract publisher') + } + const published = await input.contracts.publish({ + pageId: task.pageId, + sourceKey: task.sourceKey, + content: task.content, + contentDigest: task.contentDigest, + }) + const parsed = contractDeliverySchema.safeParse(published) + if (!parsed.success) { + const details = [...new Set(parsed.error.issues.map((issue) => issue.message))].join('; ') + throw new Error(`portable Notion contract publisher returned invalid delivery: ${details}`) + } + const delivery = parsed.data + if (existing && !sameContractDelivery(existing, delivery)) { + throw new Error('portable Notion contract delivery changed after dispatch') + } + return delivery +} + +function renderWorkerMountInstructions( + task: NormalizedNotionTask, + delivery?: NotionContractDelivery, +): string[] { + if (!delivery) { + return [ + 'The complete authorized spec is available to workers through the read-only Relayfile mount:', + `\`${task.workerSourcePath}\``, + ] + } + return [ + 'The complete authorized spec was snapshotted from the read-only Notion mount into a workspace-private Agent Relay channel:', + `- channel: \`${delivery.channel}\``, + `- ordered message ids: \`${delivery.messageIds.join(',')}\``, + `- encoding: \`${delivery.encoding}\``, + '', + `Join that channel, concatenate the base64 payload from those exact messages in order, decode it to \`${task.workerSourcePath}\`, chmod the file 0444, and then apply the SHA-256 gate below. Never copy the contract to GitHub or another public surface.`, + ] +} + +function sameContractDelivery( + left: NotionContractDelivery | undefined, + right: NotionContractDelivery, +): boolean { + return Boolean(left && left.kind === right.kind && left.channel === right.channel && + left.encoding === right.encoding && left.messageIds.join('\0') === right.messageIds.join('\0')) +} + +function contractDeliveryMarker(delivery: NotionContractDelivery): string { + return `` +} + +function contractDeliveryFromBody(body: string): NotionContractDelivery | undefined { + const match = //u.exec(body) + if (!match) return undefined + const parsed = contractDeliverySchema.safeParse({ + kind: 'relay-channel', + channel: match[1], + messageIds: match[2]?.split(','), + encoding: 'base64-chunks-v1', + }) + return parsed.success ? parsed.data : undefined +} + function sourceMarker(sourceKey: string): string { return `` } @@ -523,6 +723,7 @@ async function readIntakeState(path: string): Promise { const common = { digest: z.string().regex(/^[0-9a-f]{64}$/u), dispatchedAt: z.string().datetime(), + delivery: contractDeliverySchema.optional(), } return z.object({ version: z.literal(1),