From 3c4f245f970989e6a49a80eaacfa46dccc36cb82 Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 11 Sep 2026 20:29:20 +0200 Subject: [PATCH] feat(sdk): prove journaled agent channel posts Session-Id: 01a091b4-085a-7f93-a4b5-011e627868c1 --- docs/SURFACE.md | 38 ++++++++ evidence/spec-R-channel-verification.txt | 29 ++++++ packages/sdk/src/effect-channel.ts | 97 ++++++++++++++++++++ packages/sdk/tests/effect-channel.test.ts | 103 ++++++++++++++++++++++ 4 files changed, 267 insertions(+) create mode 100644 evidence/spec-R-channel-verification.txt create mode 100644 packages/sdk/src/effect-channel.ts create mode 100644 packages/sdk/tests/effect-channel.test.ts diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 00af787cf..e231ff694 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -78,6 +78,34 @@ No process runs between events: the handler wakes, executes to its next await, p failed calls retain their MCP diagnostic in `trajectory_tail` and complete with `worker_error`. + **Agent channels (contract; initial post proof).** The target authoring API is + `const channel = f.channel('review-panel')`, followed by + `await channel.post('security-lens', 'this looks off at line 42')` or + `await channel.recv('maintainability-lens', { timeout: '30s' })`. + Each send and receive is one journaled effect boundary. Helpers lower to + the existing agent effect record/confirm path, never a new `StepKind`. + The message ID is the transport idempotency key; the kernel still uses its + issued attempt key to authorize the effect claim. The broker transports + messages; the journal records their truth and ordering. + + The initial internal SDK proof (`effect-channel.ts`) implements one post + per declared agent step through Agent Relay's `messages.dm` interface, + with the logical channel in message metadata. It validates recipients + against a supplied participant inventory before producing the spec, and + again before transport. Credentials stay in the supplied broker client; + the completed output contains only the message envelope. A confirmed + post can complete after interruption without fetching or resending it. + An unconfirmed retry reuses the same broker idempotency key. + + This proof does **not** expose `Ctx.channel` yet. Public authored lowering, + `flows check` participant discovery, receive/acknowledgement, concurrent + per-channel ordering and a real broker SIGKILL/resume test remain follow-up. + Automatic workspaces also remain follow-up: provision on first use with a + run-ID-derived identity, inject credentials during step setup, retain on + park until resume or lease expiry, clean up at terminal completion, and + reconstruct broker state from journal after broker loss. The proof accepts + an already provisioned run-scoped client and makes no workspace API changes. + The first typed codegen slice covers Slack's four existing dispatcher methods. `Ctx` composes the generated helper namespace map; argument shapes come from the pinned relayfile ergonomic client and results retain journal-backed `Step` @@ -661,3 +689,13 @@ kernel primitives. - Are YAML helper verbs (`slack:`, `mcp:`) core spec vocabulary or compile-time expansion into `run`/effect steps? Leaning: expansion — the kernel spec stays seven words; helpers stay a surface concern. - Helper generation cadence: generated from relayfile adapter manifests at build time vs published per-adapter packages. Leaning: generated, with hand-tuned verb names for the top providers. + +## 7. Broker transport covenant + +Broker interactions with a v2 flow are legal only as journaled effect steps. +Any bypass of that journaling makes the run non-replayable and must set +`step.completed.human_intervention: true`. + +This is the required contract for channel rollout. Detecting bypasses and +carrying that marker through the completion protocol remain implementation +work; the initial post proof does not claim to enforce uninstrumented agent I/O. diff --git a/evidence/spec-R-channel-verification.txt b/evidence/spec-R-channel-verification.txt new file mode 100644 index 000000000..dca618c1f --- /dev/null +++ b/evidence/spec-R-channel-verification.txt @@ -0,0 +1,29 @@ +$ (cd packages/sdk && npm run typecheck) + +> @relayflows/sdk@2.0.8 typecheck +> tsc --noEmit && tsc -p tsconfig.type-tests.json + +Exit code: 0 + +$ (cd packages/sdk && npm run typecheck:tests) + +> @relayflows/sdk@2.0.8 typecheck:tests +> tsc -p tsconfig.tests.json + +Exit code: 0 + +$ (cd packages/sdk && ./node_modules/.bin/vitest run tests/effect-channel.test.ts tests/journal-client.test.ts tests/worker-lease.test.ts) + + RUN v2.1.9 /Users/khaliqgant/flows-spec-R-channel/packages/sdk + + ✓ tests/worker-lease.test.ts (7 tests) 6ms + ✓ tests/journal-client.test.ts (14 tests) 66ms + ✓ tests/effect-channel.test.ts (5 tests) 297ms + + Test Files 3 passed (3) + Tests 26 passed (26) + Start at 20:29:12 + Duration 633ms (transform 118ms, setup 0ms, collect 313ms, tests 370ms, environment 0ms, prepare 96ms) + +Exit code: 0 + diff --git a/packages/sdk/src/effect-channel.ts b/packages/sdk/src/effect-channel.ts new file mode 100644 index 000000000..4425a8ee0 --- /dev/null +++ b/packages/sdk/src/effect-channel.ts @@ -0,0 +1,97 @@ +import { createHash } from 'node:crypto'; +import { compileSpec, toKernelSpec } from './compile.js'; +import type { JournalClient } from './journal-client.js'; +import type { StepDispatchEvent } from './protocol.js'; +import { SPEC_SCHEMA_VERSION } from './spec.js'; +import { withWorkerLease } from './worker-lease.js'; +import { snapshotJsonValue } from './json-value.js'; + +/** Structural subset of Agent Relay's messages.dm; credentials stay in the client. */ +export interface ChannelBroker { + messages: { + dm(input: { + to: string; + text: string; + idempotencyKey: string; + metadata: { channel: string; messageId: string }; + }): Promise; + }; +} + +export interface ChannelPost { + channel: string; + to: string; + text: string; +} + +function validatePost(post: ChannelPost, participants: readonly string[]): void { + if (typeof post.channel !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9_-]*$/.test(post.channel)) { + throw new Error('channel_invalid'); + } + if (typeof post.to !== 'string' || !post.to.trim() || !participants.includes(post.to)) { + throw new Error(`channel_participant_unresolved: ${post.to}`); + } + if (typeof post.text !== 'string') throw new Error('channel_text_invalid'); +} + +/** One message boundary per step. Participant resolution precedes journal submission. */ +export function channelPostSpec( + name: string, stepId: string, post: ChannelPost, participants: readonly string[], +) { + post = snapshotJsonValue(post, 'channel post') as unknown as ChannelPost; + validatePost(post, participants); + return toKernelSpec(compileSpec({ + version: SPEC_SCHEMA_VERSION, name, + steps: [{ id: stepId, type: 'agent', + instruction: JSON.stringify({ type: 'effect', provider: 'channel', verb: 'post', + channel: post.channel, to: post.to, text: post.text }), + maxIterations: 3, recoveryMode: 'reset', + surfaces: { streams: [{ stream: `channel-${post.channel}` }], external: [`/channel/${post.channel}`] }, + }], + })); +} + +/** + * Internal post-only proof, for an already provisioned run-scoped broker client. + * The broker must honor idempotencyKey across worker retries. The application + * message ID is deliberately distinct from any provider-generated record ID. + */ +export async function completeChannelPost( + client: JournalClient, dispatch: StepDispatchEvent, broker: ChannelBroker, + participants: readonly string[], +): Promise { + const spec = dispatch.spec as { instruction?: string; surfaces?: { external?: string[] } }; + if (dispatch.step_type !== 'agent' || typeof spec.instruction !== 'string') { + throw new Error('channel_dispatch_invalid'); + } + const call = JSON.parse(spec.instruction) as ChannelPost & { type: string; provider: string; verb: string }; + if (call.type !== 'effect' || call.provider !== 'channel' || call.verb !== 'post') { + throw new Error('channel_dispatch_invalid'); + } + validatePost(call, participants); + const surfacePath = `/channel/${call.channel}`; + if (!spec.surfaces?.external?.includes(surfacePath)) throw new Error('channel_surface_undeclared'); + const messageId = createHash('sha256') + .update(JSON.stringify([dispatch.run_id, dispatch.step_id, call.channel])).digest('hex'); + await withWorkerLease(client, dispatch, async signal => { + await client.performEffect({ + runId: dispatch.run_id, stepId: dispatch.step_id, attempt: dispatch.attempt, + idempotencyKey: dispatch.idempotency_key, surfacePath, + revisionBefore: 'pending', revisionAfter: messageId, + }, async () => { + signal.throwIfAborted(); + await broker.messages.dm({ to: call.to, text: call.text, idempotencyKey: messageId, + metadata: { channel: call.channel, messageId } }); + signal.throwIfAborted(); + }); + }); + // All receipt fields already exist in the journaled spec and stable run identity. + // A crash after confirm needs neither a local receipt cache nor a broker read. + await client.stepComplete(dispatch.run_id, dispatch.step_id, dispatch.attempt, + dispatch.idempotency_key, 'success', { + output: { type: 'effect', provider: 'channel', verb: 'post', + messageId, channel: call.channel, to: call.to, text: call.text }, + started_pins: dispatch.pins, end_pins: dispatch.pins, + effects: [{ surface_path: surfacePath, idempotency_key: dispatch.idempotency_key }], + }); +} diff --git a/packages/sdk/tests/effect-channel.test.ts b/packages/sdk/tests/effect-channel.test.ts new file mode 100644 index 000000000..330bfd03e --- /dev/null +++ b/packages/sdk/tests/effect-channel.test.ts @@ -0,0 +1,103 @@ +import { spawn, spawnSync, type ChildProcess } from 'node:child_process'; +import { once } from 'node:events'; +import { existsSync, mkdtempSync, realpathSync, rmSync } from 'node:fs'; +import { homedir, tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; +import { afterEach, expect, it, vi } from 'vitest'; +import { channelPostSpec, completeChannelPost, type ChannelBroker } from '../src/effect-channel.js'; +import { JournalClient } from '../src/journal-client.js'; +import { socketPathFor } from '../src/daemon-connection.js'; +import type { StepDispatchEvent } from '../src/protocol.js'; + +const root = resolve('../..'); +const key = spawnSync('cksum', { input: realpathSync(root), encoding: 'utf8' }).stdout.trim().split(' ')[0]!; +const binary = process.env.RELAYFLOWD_BIN ?? join(process.env.CARGO_TARGET_DIR + ?? join(process.env.RELAYFLOWS_TOOLCHAIN_HOME ?? join(homedir(), '.relayflows-toolchain'), 'target', key), 'debug/relayflowd'); +const directories: string[] = []; +const children: ChildProcess[] = []; +const clients: JournalClient[] = []; +afterEach(async () => { + vi.restoreAllMocks(); + for (const client of clients.splice(0)) client.close(); + for (const child of children.splice(0)) { + if (child.exitCode !== null || child.signalCode !== null) continue; + const exited = once(child, 'exit'); child.kill('SIGKILL'); await exited; + } + for (const dir of directories.splice(0)) rmSync(dir, { recursive: true, force: true }); +}); + +it('lowers to an agent effect and refuses unresolved participants before submission', () => { + const post = { channel: 'review-panel', to: 'security-lens', text: 'line 42' }; + expect(() => channelPostSpec('review', 'post', post, [])).toThrow('channel_participant_unresolved'); + expect(channelPostSpec('review', 'post', post, ['security-lens']).steps[0]).toMatchObject({ + type: 'agent', surfaces: { external: ['/channel/review-panel'] }, + }); + expect(() => channelPostSpec('review', 'post', { ...post, channel: '../escape' }, ['security-lens'])) + .toThrow('channel_invalid'); +}); + +async function start(dir: string) { + const daemon = spawn(binary, ['--data-dir', dir, 'serve'], { stdio: ['ignore', 'pipe', 'pipe'] }); + children.push(daemon); + let client: JournalClient | undefined; + let stderr = ''; + daemon.stderr!.on('data', data => { stderr += String(data); }); + for (let attempt = 0; attempt < 100; attempt++) { + const candidate = new JournalClient(socketPathFor(dir), { connectTimeoutMs: 100 }); + try { await candidate.connect(); await candidate.hello('channel-test'); client = candidate; break; } + catch { candidate.close(); await delay(20); } + } + if (!client) throw new Error(`daemon startup failed: ${stderr}`); + clients.push(client); + const worker = client.createPeer(); clients.push(worker); + await worker.connect(); await worker.hello('channel-worker'); + const dispatched = once(worker, 'step.dispatch') as Promise<[StepDispatchEvent]>; + await worker.workerAttach('channel-worker', ['agent'], { + workspace: [], streams: [{ stream: 'channel-review-panel', read_offset: 0 }], + }, 1); + return { daemon, client, worker, dispatched }; +} + +it.each(['none', 'record', 'confirm', 'complete'] as const)('journals one post across interruption before %s', async boundary => { + expect(existsSync(binary), `Build relayflowd first: ${binary}`).toBe(true); + const dir = mkdtempSync(join(tmpdir(), 'channel-effect-')); directories.push(dir); + let active = await start(dir); + const spec = channelPostSpec('review', 'post', { + channel: 'review-panel', to: 'security-lens', text: 'this looks off at line 42', + }, ['security-lens']); + const run = await active.client.runStart(spec); + let [dispatch] = await active.dispatched; + const delivered = new Map(); + const dm = vi.fn(async (input: Parameters[0]) => { + if (!delivered.has(input.idempotencyKey)) delivered.set(input.idempotencyKey, input); + return { credential: 'must-not-leak', providerId: 'provider-record' }; + }); + const broker = { messages: { dm } }; + if (boundary !== 'none') { + vi.spyOn(active.worker, boundary === 'record' ? 'effectRecord' : boundary === 'confirm' ? 'effectConfirm' : 'stepComplete') + .mockRejectedValueOnce(new Error('injected interruption')); + await expect(completeChannelPost(active.worker, dispatch, broker, ['security-lens'])) + .rejects.toThrow('injected interruption'); + if (boundary === 'record') expect(dm).not.toHaveBeenCalled(); + active.worker.close(); active.client.close(); + const exited = once(active.daemon, 'exit'); active.daemon.kill('SIGKILL'); await exited; + active = await start(dir); + await active.client.runResume(run.run_id); + [dispatch] = await active.dispatched; + expect(dispatch.attempt).toBe(2); + } + await completeChannelPost(active.worker, dispatch, broker, ['security-lens']); + expect(delivered.size).toBe(1); + expect(dm).toHaveBeenCalledTimes(boundary === 'confirm' ? 2 : 1); + if (boundary === 'confirm') expect(dm.mock.calls[0]).toEqual(dm.mock.calls[1]); + const entries = (await active.client.journalRead(run.run_id, 1)).entries as Array<{ + entry_type: string; payload: { output?: { messageId: string }; completionReason?: string }; + }>; + expect(entries.filter(e => e.entry_type === 'effect.confirmed')).toHaveLength(1); + const completed = entries.filter(e => e.entry_type === 'step.completed' && e.payload.completionReason === 'success'); + expect(completed).toHaveLength(1); + expect(completed[0]!.payload.completionReason).toBe('success'); + expect(completed[0]!.payload.output!.messageId).toBe([...delivered.keys()][0]); + expect(JSON.stringify(entries)).not.toContain('must-not-leak'); +}, 15_000);