feat(surface,sdk,kernel): agent-to-agent effect channel via broker transport (#335)

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.

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
+

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<unknown>;
+  };
+}
+
+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<void> {
+  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 }],
+    });
+}

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<string, unknown>();
+  const dm = vi.fn(async (input: Parameters<ChannelBroker['messages']['dm']>[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);