From 81500b6c268179efec808d801320a52a8dc7ac7f Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 6 Aug 2026 02:32:12 +0200 Subject: [PATCH 1/6] feat(intake): deliver Notion specs across fleet nodes --- README.md | 11 +- src/cli/fleet.ts | 53 ++++++--- src/intake/index.ts | 8 ++ src/intake/notion-relay-contract.ts | 155 +++++++++++++++++++++++++ src/intake/notion.test.ts | 125 ++++++++++++++++++++ src/intake/notion.ts | 169 ++++++++++++++++++++++++++-- 6 files changed, 493 insertions(+), 28 deletions(-) create mode 100644 src/intake/notion-relay-contract.ts diff --git a/README.md b/README.md index 47d7a7e..c10e5c5 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,14 @@ 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. +`{ "kind": "local" }` retains the older shared-filesystem contract and remains +the default for existing manifests. `page` accepts a Notion URL or a bare page +ID. Plan without writes, then dispatch: diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 0cbba58..3fea7fa 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,13 @@ 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') { + const workspaceKey = resolveRelayWorkspaceKey({ env: deps.env ?? process.env }) + if (!workspaceKey) { + throw new Error('relay-channel worker mount transport requires an active Agent Relay workspace') + } + notionContracts = deps.notionContracts ?? new RelayChannelNotionContractPublisher({ workspaceKey }) + } const workspace: WorkspaceTaskDispatcher = { dispatch: async (task) => { fleet ??= await buildFleet(globals, undefined, deps) @@ -200,7 +213,11 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom 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 +423,28 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom return 1 } finally { try { - await mount?.dispose?.() + await notionContracts?.dispose?.() } 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.ts b/src/intake/notion-relay-contract.ts new file mode 100644 index 0000000..0d2a027 --- /dev/null +++ b/src/intake/notion-relay-contract.ts @@ -0,0 +1,155 @@ +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 +} + +/** + * 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 #cache = new Map() + #workspaceRelay?: AgentRelay + #agentRelay?: AgentRelay + + constructor(options: RelayChannelContractPublisherOptions) { + this.#workspaceKey = options.workspaceKey + this.#baseUrl = options.baseUrl + this.#publisherName = options.publisherName ?? + `factory-notion-intake-${process.pid}-${Date.now().toString(36)}` + } + + async publish(input: { + pageId: string + sourceKey: string + content: string + contentDigest: string + }): Promise { + const observedDigest = createHash('sha256').update(input.content).digest('hex') + if (observedDigest !== input.contentDigest) { + throw new Error('Notion contract changed before portable mount publication') + } + const cached = this.#cache.get(input.sourceKey) + if (cached) return cached + + const relay = await this.#relay() + const channel = contractChannelName(input.pageId, input.sourceKey) + try { + await relay.channels.join(channel) + } catch { + try { + await relay.channels.create({ + name: channel, + topic: `Read-only Notion contract ${input.pageId}`, + }) + } catch { + await relay.channels.join(channel) + } + } + + 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(input.sourceKey, delivery) + return delivery + } + + async dispose(): Promise { + await this.#agentRelay?.messaging.events.disconnect().catch(() => undefined) + await this.#workspaceRelay?.agents.delete(this.#publisherName).catch(() => undefined) + this.#agentRelay = undefined + this.#workspaceRelay = undefined + } + + async #relay(): Promise { + if (this.#agentRelay) return this.#agentRelay + const options = { + workspaceKey: this.#workspaceKey, + ...(this.#baseUrl ? { baseUrl: this.#baseUrl } : {}), + } + const workspaceRelay = new AgentRelay(options) + const registration = await workspaceRelay.agents.register({ + name: this.#publisherName, + type: 'system', + }) + this.#workspaceRelay = workspaceRelay + this.#agentRelay = new AgentRelay({ ...options, agentToken: registration.token }) + return this.#agentRelay + } +} + +export function contractChannelName(pageId: string, sourceKey: string): string { + const suffix = createHash('sha256').update(sourceKey).digest('hex').slice(0, 10) + return `factory-notion-${pageId.slice(-8)}-${suffix}` +} + +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: AgentRelay, 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 channel exceeds the 10,000-message safety limit') +} diff --git a/src/intake/notion.test.ts b/src/intake/notion.test.ts index ad379ce..8c320f0 100644 --- a/src/intake/notion.test.ts +++ b/src/intake/notion.test.ts @@ -12,6 +12,7 @@ import { type GithubIssuePublisher, type NotionIntakeManifest, type NotionIntakeTarget, + type NotionContractPublisher, type WorkspaceTaskDispatcher, } from './notion' @@ -165,6 +166,128 @@ 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'), + })) + }) + + 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 source marker that has no authoritative intake receipt', async () => { const { root, manifest } = await fixtureManifest('private mounted body', { bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), @@ -317,6 +440,7 @@ async function fixtureManifest( version: 1, mountRoot, workerMountRoot: '.integrations/notion', + workerMountTransport: { kind: 'local' }, statePath: join(root, 'state.json'), tasks: [{ page: pageId, ...task }], }, @@ -343,5 +467,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..e0454cb 100644 --- a/src/intake/notion.ts +++ b/src/intake/notion.ts @@ -23,6 +23,11 @@ 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 bootstrapSchema = z.object({ authorizedPageId: z.string().trim().min(1), reason: z.string().trim().min(1), @@ -37,6 +42,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 +59,7 @@ export interface NormalizedNotionTask { sourceKey: string sourcePath: string workerSourcePath: string + content: string authorizationDigestInput?: string contentDigest: string digest: string @@ -79,6 +86,28 @@ export interface GithubIssuePublisher { body: string labels: readonly string[] }): Promise<{ number: number; url: string }> + updateIssue(input: { + repo: string + number: number + body: string + }): Promise +} + +export interface NotionContractDelivery { + kind: 'relay-channel' + channel: string + messageIds: string[] + encoding: 'base64-chunks-v1' +} + +export interface NotionContractPublisher { + publish(input: { + pageId: string + sourceKey: string + content: string + contentDigest: string + }): Promise + dispose?(): Promise } export interface WorkspaceTaskDispatcher { @@ -114,12 +143,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 +177,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 +206,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 +266,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 +372,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 +417,36 @@ async function publishRepoTask( if (currentDigest !== task.digest) { return { ...base, status: 'blocked', issue: existing, reason: 'mounted spec changed after the lifecycle issue was created' } } + const visibility = await input.github.repositoryVisibility(target.repo) + const summary = visibility === 'public' ? target.publicSummary! : task.summary + if (input.manifest.workerMountTransport.kind === 'relay-channel' && + !receipt.delivery && existing.body !== renderIssueBody(task, summary)) { + return { + ...base, + status: 'blocked', + issue: existing, + reason: 'existing lifecycle issue body was edited; refusing to overwrite it during portable mount migration', + } + } + const delivery = await prepareContractDelivery(task, input, receipt.delivery) + if (delivery && !issueHasContractDelivery(existing.body, delivery)) { + if (existing.body !== renderIssueBody(task, summary)) { + return { + ...base, + status: 'blocked', + issue: existing, + reason: 'existing lifecycle issue body was edited; refusing to overwrite it during portable mount migration', + } + } + 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 +462,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 +505,16 @@ 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' } } + const delivery = await prepareContractDelivery(task, input, receipt.delivery) + if (delivery && !sameContractDelivery(receipt.delivery, delivery)) { + state.receipts[task.sourceKey] = { ...receipt, delivery } + } return { ...base, status: 'already-dispatched', agent: receipt.agent, node: receipt.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 +523,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 +549,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 +567,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 +585,63 @@ 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 delivery = await input.contracts.publish({ + pageId: task.pageId, + sourceKey: task.sourceKey, + content: task.content, + contentDigest: task.contentDigest, + }) + 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 issueHasContractDelivery(body: string, delivery: NotionContractDelivery): boolean { + return body.includes(contractDeliveryMarker(delivery)) +} + +function contractDeliveryMarker(delivery: NotionContractDelivery): string { + return `` +} + function sourceMarker(sourceKey: string): string { return `` } @@ -520,9 +662,16 @@ function splitField(value: string | undefined): string[] { async function readIntakeState(path: string): Promise { try { + const delivery = z.object({ + kind: z.literal('relay-channel'), + channel: z.string().min(1), + messageIds: z.array(z.string().min(1)).min(1), + encoding: z.literal('base64-chunks-v1'), + }).strict() const common = { digest: z.string().regex(/^[0-9a-f]{64}$/u), dispatchedAt: z.string().datetime(), + delivery: delivery.optional(), } return z.object({ version: z.literal(1), From b2aaef783fe7100bc844efe2893c82e576e31279 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 6 Aug 2026 02:41:25 +0200 Subject: [PATCH 2/6] fix(intake): refresh exact-path workers on mount migration --- src/cli/fleet.test.ts | 39 +++++++++++++++++++++++++-- src/cli/fleet.ts | 18 +++++++++++++ src/intake/notion.test.ts | 55 ++++++++++++++++++++++++++++++++++++++- src/intake/notion.ts | 35 +++++++++++++++++++++++-- 4 files changed, 142 insertions(+), 5 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index c5ea817..2fbec37 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,40 @@ 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 () => undefined), + } + const migratedOutput = buffer() + const migratedCode = await runFleetCli(['intake', 'notion', manifestPath], { + fleet, + notionContracts: contracts, + stdout: migratedOutput, + stderr: buffer(), + }) + + 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(JSON.parse(migratedOutput.text())).toMatchObject({ + ok: true, + results: [{ status: 'already-dispatched', target: { projectPath } }], + }) } finally { await rm(root, { recursive: true, force: true }) } diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 3fea7fa..eb427aa 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -209,6 +209,24 @@ 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, diff --git a/src/intake/notion.test.ts b/src/intake/notion.test.ts index 8c320f0..ca13379 100644 --- a/src/intake/notion.test.ts +++ b/src/intake/notion.test.ts @@ -361,6 +361,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({ @@ -387,17 +388,69 @@ 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 }) + 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) + 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 () => { diff --git a/src/intake/notion.ts b/src/intake/notion.ts index e0454cb..9221137 100644 --- a/src/intake/notion.ts +++ b/src/intake/notion.ts @@ -119,6 +119,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 = { @@ -507,9 +515,32 @@ async function dispatchWorkspaceTask( } const delivery = await prepareContractDelivery(task, input, receipt.delivery) if (delivery && !sameContractDelivery(receipt.delivery, delivery)) { - state.receipts[task.sourceKey] = { ...receipt, 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, + } } - return { ...base, status: 'already-dispatched', agent: receipt.agent, node: receipt.node } + 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' } From 4daae12957dd8ca3b71a0721554db3fa527bf840 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 6 Aug 2026 02:46:20 +0200 Subject: [PATCH 3/6] test(intake): keep injected Relay publisher hermetic --- src/cli/fleet.test.ts | 1 + src/cli/fleet.ts | 11 +++++++---- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 2fbec37..697c93f 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -622,6 +622,7 @@ describe('fleet CLI runtime', () => { const migratedCode = await runFleetCli(['intake', 'notion', manifestPath], { fleet, notionContracts: contracts, + env: {}, stdout: migratedOutput, stderr: buffer(), }) diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index eb427aa..dd5ccfc 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -187,11 +187,14 @@ 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') { - const workspaceKey = resolveRelayWorkspaceKey({ env: deps.env ?? process.env }) - if (!workspaceKey) { - throw new Error('relay-channel worker mount transport requires an active Agent Relay workspace') + notionContracts = deps.notionContracts + if (!notionContracts) { + const workspaceKey = resolveRelayWorkspaceKey({ env: deps.env ?? process.env }) + if (!workspaceKey) { + throw new Error('relay-channel worker mount transport requires an active Agent Relay workspace') + } + notionContracts = new RelayChannelNotionContractPublisher({ workspaceKey }) } - notionContracts = deps.notionContracts ?? new RelayChannelNotionContractPublisher({ workspaceKey }) } const workspace: WorkspaceTaskDispatcher = { dispatch: async (task) => { From b2a5b80b909e9f912cf7185b4f82d6d58d3221dc Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 6 Aug 2026 03:04:48 +0200 Subject: [PATCH 4/6] fix(intake): harden portable Notion delivery --- README.md | 7 +- src/cli/fleet.test.ts | 42 ++++++- src/cli/fleet.ts | 11 +- src/intake/notion-relay-contract.test.ts | 145 +++++++++++++++++++++++ src/intake/notion-relay-contract.ts | 64 ++++++---- src/intake/notion.test.ts | 88 ++++++++++++++ src/intake/notion.ts | 74 +++++++----- 7 files changed, 374 insertions(+), 57 deletions(-) create mode 100644 src/intake/notion-relay-contract.test.ts diff --git a/README.md b/README.md index c10e5c5..5ebb9a5 100644 --- a/README.md +++ b/README.md @@ -212,9 +212,10 @@ is the repo-relative read-only mount workers receive. With the recommended 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. -`{ "kind": "local" }` retains the older shared-filesystem contract and remains -the default for existing manifests. `page` accepts a Notion URL or a bare page -ID. +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 697c93f..17558e1 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -616,15 +616,16 @@ describe('fleet CLI runtime', () => { messageIds: ['message-1'], encoding: 'base64-chunks-v1' as const, })), - dispose: vi.fn(async () => undefined), + 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: buffer(), + stderr: migratedErrors, }) expect(migratedCode).toBe(0) @@ -635,6 +636,7 @@ describe('fleet CLI runtime', () => { 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 } }], @@ -644,6 +646,42 @@ describe('fleet CLI runtime', () => { } }) + 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 }) + } + }) + it('prompts and connects a missing GitHub integration before an interactive triage', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-integration-connect-')) try { diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index dd5ccfc..fd6fc12 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -189,7 +189,10 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom if (!globals.dryRun && manifest.workerMountTransport.kind === 'relay-channel') { notionContracts = deps.notionContracts if (!notionContracts) { - const workspaceKey = resolveRelayWorkspaceKey({ env: deps.env ?? process.env }) + 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') } @@ -444,7 +447,11 @@ export async function runFleetCli(argv: string[], deps: FleetCliDeps = {}): Prom return 1 } finally { try { - await notionContracts?.dispose?.() + try { + await notionContracts?.dispose?.() + } catch { + err.write('[factory] warning: Notion contract publisher failed during shutdown\n') + } } finally { try { await mount?.dispose?.() diff --git a/src/intake/notion-relay-contract.test.ts b/src/intake/notion-relay-contract.test.ts new file mode 100644 index 0000000..def872b --- /dev/null +++ b/src/intake/notion-relay-contract.test.ts @@ -0,0 +1,145 @@ +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 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) + }) +}) diff --git a/src/intake/notion-relay-contract.ts b/src/intake/notion-relay-contract.ts index 0d2a027..f52f526 100644 --- a/src/intake/notion-relay-contract.ts +++ b/src/intake/notion-relay-contract.ts @@ -13,8 +13,11 @@ 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 @@ -24,15 +27,18 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis readonly #workspaceKey: string readonly #baseUrl?: string readonly #publisherName: string + readonly #createRelay: NonNullable readonly #cache = new Map() - #workspaceRelay?: AgentRelay - #agentRelay?: AgentRelay + #workspaceRelay?: ContractRelay + #agentRelay?: ContractRelay + #relayReady?: Promise 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: { @@ -45,21 +51,28 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis if (observedDigest !== input.contentDigest) { throw new Error('Notion contract changed before portable mount publication') } - const cached = this.#cache.get(input.sourceKey) + 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) + const channel = contractChannelName(input.pageId, input.sourceKey, input.contentDigest) try { await relay.channels.join(channel) - } catch { + } catch (joinError) { try { await relay.channels.create({ name: channel, topic: `Read-only Notion contract ${input.pageId}`, }) - } catch { - await relay.channels.join(channel) + } 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]), + }) + } } } @@ -77,10 +90,10 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis .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}`, - }) + 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`) } @@ -93,7 +106,7 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis messageIds, encoding: 'base64-chunks-v1', } - this.#cache.set(input.sourceKey, delivery) + this.#cache.set(cacheKey, delivery) return delivery } @@ -102,28 +115,39 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis await this.#workspaceRelay?.agents.delete(this.#publisherName).catch(() => undefined) this.#agentRelay = undefined this.#workspaceRelay = undefined + this.#relayReady = undefined } - async #relay(): Promise { + 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 = new AgentRelay(options) + const workspaceRelay = this.#createRelay(options) const registration = await workspaceRelay.agents.register({ name: this.#publisherName, type: 'system', }) this.#workspaceRelay = workspaceRelay - this.#agentRelay = new AgentRelay({ ...options, agentToken: registration.token }) + this.#agentRelay = this.#createRelay({ ...options, agentToken: registration.token }) return this.#agentRelay } } -export function contractChannelName(pageId: string, sourceKey: string): string { - const suffix = createHash('sha256').update(sourceKey).digest('hex').slice(0, 10) - return `factory-notion-${pageId.slice(-8)}-${suffix}` +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 { @@ -138,7 +162,7 @@ function splitContract(encoded: string): string[] { return chunks.length > 0 ? chunks : [''] } -async function listAllMessages(relay: AgentRelay, channel: string): Promise { +async function listAllMessages(relay: ContractRelay, channel: string): Promise { const messages: RelayMessage[] = [] let before: string | undefined for (let pageNumber = 0; pageNumber < 100; pageNumber += 1) { @@ -151,5 +175,5 @@ async function listAllMessages(relay: AgentRelay, channel: string): Promise { ) }) + 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') @@ -236,6 +251,54 @@ describe('Notion spec intake', () => { 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 () => { @@ -288,6 +351,31 @@ describe('Notion spec intake', () => { 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 a source marker that has no authoritative intake receipt', async () => { const { root, manifest } = await fixtureManifest('private mounted body', { bootstrap: bootstrap({ repo: 'AgentWorkforce/cloud', labels: [] }), diff --git a/src/intake/notion.ts b/src/intake/notion.ts index 9221137..02b6d42 100644 --- a/src/intake/notion.ts +++ b/src/intake/notion.ts @@ -28,6 +28,13 @@ const workerMountTransportSchema = z.discriminatedUnion('kind', [ z.object({ kind: z.literal('relay-channel') }).strict(), ]).default({ kind: 'local' }) +const contractDeliverySchema = z.object({ + kind: z.literal('relay-channel'), + channel: z.string().trim().min(1), + messageIds: z.array(z.string().trim().min(1)).min(1), + encoding: z.literal('base64-chunks-v1'), +}).strict() + const bootstrapSchema = z.object({ authorizedPageId: z.string().trim().min(1), reason: z.string().trim().min(1), @@ -93,12 +100,7 @@ export interface GithubIssuePublisher { }): Promise } -export interface NotionContractDelivery { - kind: 'relay-channel' - channel: string - messageIds: string[] - encoding: 'base64-chunks-v1' -} +export type NotionContractDelivery = z.infer export interface NotionContractPublisher { publish(input: { @@ -425,10 +427,20 @@ 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 - if (input.manifest.workerMountTransport.kind === 'relay-channel' && - !receipt.delivery && existing.body !== renderIssueBody(task, summary)) { + const bodyWasEdited = existing.body !== renderIssueBody(task, summary, bodyDelivery) + if (bodyWasEdited) { return { ...base, status: 'blocked', @@ -436,16 +448,11 @@ async function publishRepoTask( reason: 'existing lifecycle issue body was edited; refusing to overwrite it during portable mount migration', } } - const delivery = await prepareContractDelivery(task, input, receipt.delivery) - if (delivery && !issueHasContractDelivery(existing.body, delivery)) { - if (existing.body !== renderIssueBody(task, summary)) { - 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, @@ -625,12 +632,17 @@ async function prepareContractDelivery( if (!input.contracts) { throw new Error('relay-channel worker mount transport requires an Agent Relay contract publisher') } - const delivery = await input.contracts.publish({ + 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) { + throw new Error('portable Notion contract publisher must return a channel and at least one message id') + } + const delivery = parsed.data if (existing && !sameContractDelivery(existing, delivery)) { throw new Error('portable Notion contract delivery changed after dispatch') } @@ -665,14 +677,22 @@ function sameContractDelivery( left.encoding === right.encoding && left.messageIds.join('\0') === right.messageIds.join('\0')) } -function issueHasContractDelivery(body: string, delivery: NotionContractDelivery): boolean { - return body.includes(contractDeliveryMarker(delivery)) -} - 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 `` } @@ -693,16 +713,10 @@ function splitField(value: string | undefined): string[] { async function readIntakeState(path: string): Promise { try { - const delivery = z.object({ - kind: z.literal('relay-channel'), - channel: z.string().min(1), - messageIds: z.array(z.string().min(1)).min(1), - encoding: z.literal('base64-chunks-v1'), - }).strict() const common = { digest: z.string().regex(/^[0-9a-f]{64}$/u), dispatchedAt: z.string().datetime(), - delivery: delivery.optional(), + delivery: contractDeliverySchema.optional(), } return z.object({ version: z.literal(1), From c9596414c27a9d676c4a360cba77e14fdef81060 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 6 Aug 2026 03:05:46 +0200 Subject: [PATCH 5/6] test(intake): track portable refresh time --- src/intake/notion.test.ts | 21 ++++++++++++++++++++- src/intake/notion.ts | 1 + 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/src/intake/notion.test.ts b/src/intake/notion.test.ts index cf810fc..7e3fe5b 100644 --- a/src/intake/notion.test.ts +++ b/src/intake/notion.test.ts @@ -485,7 +485,13 @@ describe('Notion spec intake', () => { encoding: 'base64-chunks-v1', })), } - const migrated = await runNotionIntake({ manifest, dispatch: true, workspace, contracts }) + 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', @@ -497,6 +503,19 @@ describe('Notion spec intake', () => { 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'), diff --git a/src/intake/notion.ts b/src/intake/notion.ts index 02b6d42..a0a4b24 100644 --- a/src/intake/notion.ts +++ b/src/intake/notion.ts @@ -544,6 +544,7 @@ async function dispatchWorkspaceTask( agent: refreshed.agent, ...(refreshed.node ? { node: refreshed.node } : {}), delivery, + dispatchedAt: (input.now?.() ?? new Date()).toISOString(), } } const currentReceipt = state.receipts[task.sourceKey] as Extract From 85e24bbe61b721fa4aa7b5b76c0deffa20586f26 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 6 Aug 2026 03:19:51 +0200 Subject: [PATCH 6/6] fix(intake): make Relay delivery lifecycle safe --- src/intake/notion-relay-contract.test.ts | 35 ++++++++++++++++++++++++ src/intake/notion-relay-contract.ts | 8 ++++++ src/intake/notion.test.ts | 25 +++++++++++++++++ src/intake/notion.ts | 14 +++++++--- 4 files changed, 78 insertions(+), 4 deletions(-) diff --git a/src/intake/notion-relay-contract.test.ts b/src/intake/notion-relay-contract.test.ts index def872b..e0349e2 100644 --- a/src/intake/notion-relay-contract.test.ts +++ b/src/intake/notion-relay-contract.test.ts @@ -6,6 +6,12 @@ import { RelayChannelNotionContractPublisher, contractChannelName, contractMarke 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 @@ -142,4 +148,33 @@ describe('RelayChannelNotionContractPublisher', () => { 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 index f52f526..a8f7507 100644 --- a/src/intake/notion-relay-contract.ts +++ b/src/intake/notion-relay-contract.ts @@ -32,6 +32,7 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis #workspaceRelay?: ContractRelay #agentRelay?: ContractRelay #relayReady?: Promise + #disposed = false constructor(options: RelayChannelContractPublisherOptions) { this.#workspaceKey = options.workspaceKey @@ -47,6 +48,7 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis 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') @@ -111,6 +113,8 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis } 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 @@ -139,6 +143,10 @@ export class RelayChannelNotionContractPublisher implements NotionContractPublis 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 diff --git a/src/intake/notion.test.ts b/src/intake/notion.test.ts index 7e3fe5b..88cf35b 100644 --- a/src/intake/notion.test.ts +++ b/src/intake/notion.test.ts @@ -376,6 +376,31 @@ describe('Notion spec intake', () => { 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: [] }), diff --git a/src/intake/notion.ts b/src/intake/notion.ts index a0a4b24..775e51e 100644 --- a/src/intake/notion.ts +++ b/src/intake/notion.ts @@ -28,10 +28,15 @@ const workerMountTransportSchema = z.discriminatedUnion('kind', [ 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: z.string().trim().min(1), - messageIds: z.array(z.string().trim().min(1)).min(1), + channel: contractMarkerTokenSchema, + messageIds: z.array(contractMarkerTokenSchema).min(1, 'portable delivery must include at least one message id'), encoding: z.literal('base64-chunks-v1'), }).strict() @@ -641,7 +646,8 @@ async function prepareContractDelivery( }) const parsed = contractDeliverySchema.safeParse(published) if (!parsed.success) { - throw new Error('portable Notion contract publisher must return a channel and at least one message id') + 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)) { @@ -683,7 +689,7 @@ function contractDeliveryMarker(delivery: NotionContractDelivery): string { } function contractDeliveryFromBody(body: string): NotionContractDelivery | undefined { - const match = //u.exec(body) + const match = //u.exec(body) if (!match) return undefined const parsed = contractDeliverySchema.safeParse({ kind: 'relay-channel',