feat(sdk): YAML helper verbs compile to effect steps (#345)

docs/YAML-HELPERS.md

@@ -0,0 +1,57 @@
+# YAML helper verbs
+
+Helpers compile into ordinary agent steps that execute journal-backed effects
+through `@relayfile/relay-helpers`. Slack uses the same writeback path as
+`f.slack`. No provider step type is added to the kernel.
+
+```yaml
+version: 0.1.0
+steps:
+  - id: notify
+    slack:
+      post:
+        channel: "#test"
+        text: hi
+  - id: ticket
+    dependsOn: [notify]
+    linear:
+      createIssue:
+        teamId: engineering
+        title: Follow up
+```
+
+Each step has an `id` and exactly one provider containing exactly one verb.
+Optional step fields are `dependsOn`, `maxIterations`, `verification`, and
+`output`. Arguments are literal JSON-compatible data; templates and dynamic
+argument bindings are not supported. Unknown fields, verbs, and malformed
+arguments fail compilation. `YamlFlowSpec` and `YamlHelperStepSpec` describe
+the authoring shapes; `compileSpec` returns the normalized `FlowSpec`.
+
+Supported verbs:
+
+| Provider | Verbs |
+| --- | --- |
+| Slack | `post`, `dm`, `reply`, `react` |
+| GitHub | `comment`, `createIssue`, `createPullRequest`, `closePullRequest` |
+| Linear | `comment`, `createIssue`, `updateIssue` |
+
+Single-object client arguments appear directly under the verb. Slack's
+positional arguments become named fields (`channel`, `text`, `opts`, etc.).
+GitHub `comment` takes `{target: {owner, repo, number}, body}`; Linear `comment`
+takes `{issueId, body}`, and `updateIssue` takes `{issueId, args}`.
+
+Configure a relayfile mount containing the provider directory using the same
+mount environment variables as TS Slack helpers (`RELAYFILE_MOUNT_PATH`,
+`WORKSPACE_ROOT`, `WORKFORCE_SANDBOX_ROOT`, `RELAYFILE_MOUNT_ROOT`, or
+`RELAYFILE_ROOT`). `flows check` refuses missing mounts. To run locally:
+
+```sh
+flows run notify.yaml --local-agent
+```
+
+An attached SDK `AgentWorker` can also execute these steps; it must have a
+`dataDir` for durable receipts. Completion output contains the effect call,
+its stable idempotency key, and `receipt`. As with TS Slack, receipts are saved
+before effect confirmation, allowing unfinished attempts to recover without
+repeating a confirmed provider write. Keep the worker's data directory across
+restarts.

evidence/spec-V/typecheck.txt

@@ -0,0 +1,4 @@
+
+> @relayflows/sdk@2.0.8 typecheck
+> tsc --noEmit && tsc -p tsconfig.type-tests.json
+

packages/sdk/src/authored-slack-effect.ts

@@ -8,9 +8,9 @@ import type { AuthoredFlowJournalStep } from './authored-flow-executor.js';
 import { compileSpec, toKernelSpec } from './compile.js';
 import { SPEC_SCHEMA_VERSION, type KernelAgentStep } from './spec.js';
 import type { StepDispatchEvent } from './protocol.js';
-import { withWorkerLease } from './worker-lease.js';
+import { completeHelperDispatch } from './yaml-helper-effect.js';
 import { checkSlackHelpers } from './slack-preflight.js';
-import { atomicJson, readSlackReceipt, receiptPath, slackWriteback, type SlackCall } from './slack-writeback.js';
+import { atomicJson, type SlackCall } from './slack-writeback.js';
 
 export function assertSlackCredentials(): void {
   const report = checkSlackHelpers({ header: { tools: { slack: true } }, body() {} });
@@ -75,7 +75,7 @@ async function driveSlackEffect(
   client.on('step.dispatch', (dispatch: StepDispatchEvent) => {
     if (executing || dispatch.run_id !== runId || dispatch.step_id !== step.id) return;
     executing = true;
-    void completeSlackDispatch(client, dispatch, call, dataDir).then(resolve, reject);
+    void completeHelperDispatch(client, dispatch, call, dataDir).then(resolve, reject);
   });
   client.on('error', reject);
   try {
@@ -93,29 +93,3 @@ async function driveSlackEffect(
     throw new AuthoredFlowExecutionError('step_failed', error instanceof Error ? error.message : 'Slack effect failed', 'worker_error', runId);
   } finally { client.close(); }
 }
-
-async function completeSlackDispatch(client: JournalClient, dispatch: StepDispatchEvent, call: SlackCall, dataDir: string): Promise<void> {
-  const output = await withWorkerLease(client, dispatch, async signal => {
-    const file = receiptPath(dataDir, dispatch.run_id, dispatch.step_id);
-    let receipt: unknown;
-    await client.performEffect({
-      runId: dispatch.run_id, stepId: dispatch.step_id, attempt: dispatch.attempt,
-      idempotencyKey: dispatch.idempotency_key, surfacePath: '/slack',
-      revisionBefore: 'pending', revisionAfter: `${dispatch.run_id}:${dispatch.step_id}`,
-    }, async () => {
-      try { receipt = await readSlackReceipt(file); }
-      catch (error) {
-        if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
-        receipt = await slackWriteback(call, dataDir, dispatch.run_id, dispatch.step_id, signal);
-        await atomicJson(file, receipt);
-      }
-      signal.throwIfAborted();
-    });
-    // Also required after a confirmed election followed by a crash before step.complete.
-    if (receipt === undefined) receipt = await readSlackReceipt(file);
-    return { ...call, idempotencyKey: `${dispatch.run_id}:${dispatch.step_id}`, receipt };
-  });
-  await client.stepComplete(dispatch.run_id, dispatch.step_id, dispatch.attempt,
-    dispatch.idempotency_key, 'success', { output, started_pins: dispatch.pins, end_pins: dispatch.pins,
-      effects: [{ surface_path: '/slack', idempotency_key: dispatch.idempotency_key }] });
-}

packages/sdk/src/cli/check.ts

@@ -3,6 +3,7 @@ import { dirname, isAbsolute, join, parse as parsePath, resolve } from 'node:pat
 import { spawnSync } from 'node:child_process';
 import { parse as parseYaml } from 'yaml';
 import { CompileError, compileSpec, kernelToAuthoring } from '../compile.js';
+import { helperReady } from '../yaml-helper-effect.js';
 import {
   adapterIdentification,
   authenticationProbe,
@@ -270,6 +271,7 @@ function findConfig(start: string): string | undefined {
 
 function systemProbes(flowDirectory: string, config: ProjectConfig): PreflightProbes {
   return {
+    helper: helperReady,
     cli: (cli, source, model) => probeCli(cli, source === 'project' ? config.directory : flowDirectory, model),
     executor: (trigger) => config.executors.includes(trigger.executor),
     command: (binary) => executableExists(binary, flowDirectory),

packages/sdk/src/compile.ts

@@ -40,6 +40,7 @@ import { canonicalize, specHash } from './canonical.js';
 import { validateOutputDeclaration } from './output-schema.js';
 import { validateSpec, type ValidationResult } from './validate.js';
 import { snapshotJsonValue } from './json-value.js';
+import { expandYamlHelpers } from './yaml-helpers.js';
 
 export class CompileError extends Error {
   readonly errors: string[];
@@ -122,6 +123,7 @@ export function compileSpec(spec: unknown): CompiledFlowSpec {
   let snapshot: unknown;
   try {
     snapshot = snapshotJsonValue(spec, 'spec');
+    snapshot = expandYamlHelpers(snapshot);
   } catch (error) {
     throw new CompileError([
       error instanceof Error ? error.message : 'spec: expected JSON-compatible data',

packages/sdk/src/failure-kinds.ts

@@ -6,6 +6,7 @@ const PREFLIGHT_ENVIRONMENT_FAILURE_KINDS = [
   ...PLUGIN_FAILURE_KINDS,
   'helper_slack.credential_missing',
   'helper_slack.mount_required',
+  'helper_mount_required',
   'mcp_undeclared_server',
   'mcp_unreachable',
   'budget_syntax_invalid',

packages/sdk/src/index.ts

@@ -12,6 +12,9 @@ export type {
   DeterministicStepSpec,
   ExitCodeGate,
   FlowSpec,
+  YamlFlowSpec,
+  YamlHelperStepSpec,
+  YamlHelperParams,
   FlowsJson,
   McpServerConfig,
   JsonSchemaGate,

packages/sdk/src/preflight.ts

@@ -7,6 +7,7 @@ import { budgetDiagnostics } from './budget-preflight.js';
 import type { TriggerSource } from '@relayflows/surface';
 import { acceptsAnyOutput, inspectStepGate, type StepGateInspection } from './gate-contract.js';
 import { compileSpec, CompileError } from './compile.js';
+import { helperCall } from './yaml-helpers.js';
 import type {
   PreflightFailureKind,
   PreflightWarningKind,
@@ -54,6 +55,8 @@ type CliProbeOutcome =
  * emits `probe_failed` (or `command_unprovable` for a deterministic command).
  */
 export interface PreflightProbes {
+  /** Whether the provider's relayfile mount is available to the helper worker. */
+  helper?(provider: string): boolean;
   /**
    * Resolve relative paths against the file implied by `source`, then probe
    * `auth status`. When the step declared a `model`, the probe runs with that
@@ -216,6 +219,7 @@ function preflightSync(flow: unknown, options: PreflightOptions): PreflightResul
   // earlier command, provider/model, or trigger probe may run first.
   for (const step of compiled.steps) {
     if (step.type === 'deterministic') continue;
+    if (step.type === 'agent' && helperCall(step) !== undefined) continue;
     const resolution = resolveCli(step, compiled, options.projectCli);
     if (resolution === undefined) {
       diagnostics.push({
@@ -237,6 +241,22 @@ function preflightSync(flow: unknown, options: PreflightOptions): PreflightResul
     warnOnVacuousGate(step, diagnostics);
     warnOnUnprovableEffects(step, options.probes, diagnostics);
     if (step.type === 'deterministic') continue;
+    const helper = step.type === 'agent' ? helperCall(step) : undefined;
+    if (helper !== undefined) {
+      try {
+        if (options.probes.helper === undefined) {
+          diagnostics.push({ severity: 'warning', kind: 'unprovable_effects', stepId: step.id,
+            message: `${helper.provider} helper requires a relayfile mount and an SDK agent worker.` });
+        } else if (!options.probes.helper(helper.provider)) {
+          diagnostics.push({ severity: 'refusal', kind: 'helper_mount_required', stepId: step.id,
+            message: `${helper.provider} helper requires a relayfile mount.` });
+        }
+      } catch {
+        diagnostics.push({ severity: 'refusal', kind: 'probe_failed', stepId: step.id,
+          message: `${helper.provider} helper mount could not be checked.` });
+      }
+      continue;
+    }
     const resolution = resolutionByStep.get(step.id)!;
     probeResolvedCli(resolution, options.probes, cliProbeResults, diagnostics);
   }

packages/sdk/src/spec.ts

@@ -229,6 +229,33 @@ export interface AgentStepSpec extends BaseStepSpec {
 
 export type StepSpec = DeterministicStepSpec | LlmStepSpec | AgentStepSpec;
 
+/** YAML argument maps use the same pinned client types as the TS helpers. */
+export interface YamlHelperParams {
+  slack: { [V in import('./slack-writeback.js').SlackCall['verb']]:
+    Extract<import('./slack-writeback.js').SlackCall, { verb: V }>['params'] };
+  github: {
+    comment: { target: Parameters<import('@relayfile/relay-helpers').GithubClient['comment']>[0]; body: string };
+    createIssue: Parameters<import('@relayfile/relay-helpers').GithubClient['createIssue']>[0];
+    createPullRequest: Parameters<import('@relayfile/relay-helpers').GithubClient['createPullRequest']>[0];
+    closePullRequest: Parameters<import('@relayfile/relay-helpers').GithubClient['closePullRequest']>[0];
+  };
+  linear: {
+    comment: { issueId: string; body: string };
+    createIssue: Parameters<import('@relayfile/relay-helpers').LinearClient['createIssue']>[0];
+    updateIssue: { issueId: string; args: Parameters<import('@relayfile/relay-helpers').LinearClient['updateIssue']>[1] };
+  };
+}
+
+type OneKey<T> = { [K in keyof T]: Pick<T, K> & Partial<Record<Exclude<keyof T, K>, never>> }[keyof T];
+
+/** Helper sugar is removed before validation of the three kernel step types. */
+export type YamlHelperStepSpec = Pick<BaseStepSpec, 'id' | 'dependsOn' | 'maxIterations'> & {
+  verification?: OutputVerificationSpec;
+  output?: JsonOutputSchema;
+} & OneKey<{ [P in keyof YamlHelperParams]: OneKey<YamlHelperParams[P]> }>;
+
+export type YamlFlowSpec = Omit<FlowSpec, 'steps'> & { steps: Array<StepSpec | YamlHelperStepSpec> };
+
 /**
  * Reusable authoring declaration for an agent CLI/model pair. Both fields are
  * required so selecting a named agent can never inherit a host model. The

packages/sdk/src/worker.ts

@@ -7,6 +7,8 @@ import type { KernelAgentStep } from './spec.js';
 import { runAgentCli } from './worker-cli.js';
 import { withWorkerLease } from './worker-lease.js';
 import { workerInstruction } from './worker-input.js';
+import { helperCall } from './yaml-helpers.js';
+import { completeHelperDispatch } from './yaml-helper-effect.js';
 
 export { MODEL_ENV, WAKE_CONTEXT_ENV } from './worker-cli.js';
 
@@ -96,6 +98,12 @@ export class AgentWorker extends EventEmitter {
 
   private async execute(dispatch: StepDispatchEvent): Promise<void> {
     const spec = dispatch.spec as Partial<KernelAgentStep>;
+    const helper = helperCall(spec);
+    if (helper !== undefined) {
+      if (this.options.dataDir === undefined) throw new Error('Helper worker requires a data directory for durable receipts');
+      await completeHelperDispatch(this.client, dispatch, helper, this.options.dataDir);
+      return;
+    }
     let humanIntervention = false;
     const completed: WorkerCliResult = await withWorkerLease(this.client, dispatch, signal =>
       typeof spec.cli === 'string' && typeof spec.instruction === 'string'

packages/sdk/src/yaml-helper-effect.ts

@@ -0,0 +1,78 @@
+import { createHash } from 'node:crypto';
+import { statSync } from 'node:fs';
+import { join } from 'node:path';
+import { writeJsonFile } from '@relayfile/adapter-core/vfs-client';
+import type { RelayTransport } from '@relayfile/relay-helpers';
+import type { JournalClient } from './journal-client.js';
+import type { StepDispatchEvent } from './protocol.js';
+import { atomicJson, readSlackReceipt, receiptPath, slackWriteback } from './slack-writeback.js';
+import { invokeHelper, type HelperCall } from './yaml-helpers.js';
+import { withWorkerLease } from './worker-lease.js';
+
+export function helperMount(provider: string, env = process.env): string | undefined {
+  const root = [env.RELAYFILE_MOUNT_PATH, env.WORKSPACE_ROOT, env.WORKFORCE_SANDBOX_ROOT,
+    env.RELAYFILE_MOUNT_ROOT, env.RELAYFILE_ROOT].find(value => value?.trim());
+  if (!root) return undefined;
+  try { return statSync(join(root, provider)).isDirectory() ? root : undefined; }
+  catch { return undefined; }
+}
+
+export function helperReady(provider: string): boolean {
+  return (provider === 'slack' && process.env.RELAYFLOWS_SLACK_MOCK === '1') || helperMount(provider) !== undefined;
+}
+
+async function writeback(call: HelperCall, dataDir: string, runId: string, stepId: string, signal: AbortSignal) {
+  // Exactly the TS surface's client, transport, idempotency stamp and receipt checks.
+  if (call.provider === 'slack') return slackWriteback(call, dataDir, runId, stepId, signal);
+  const mount = helperMount(call.provider);
+  if (mount === undefined) throw new Error(`${call.provider} helper requires a relayfile mount`);
+  const idempotencyKey = `${runId}:${stepId}`;
+  const transport: RelayTransport = {
+    async read() { throw new Error('Helper effect transport is write-only'); },
+    async list() { throw new Error('Helper effect transport is write-only'); },
+    async write(request) {
+      signal.throwIfAborted();
+      const body = { ...request.body as Record<string, unknown>, idempotencyKey };
+      // Item updates keep the client's canonical path. Creates use a stable draft.
+      const path = request.path.endsWith('.json') ? request.path
+        : `${request.path}/draft-${createHash('sha256').update(idempotencyKey).digest('hex')}.json`;
+      const result = await writeJsonFile({ relayfileMountRoot: mount }, request.provider,
+        `write.${request.resource}`, path, body);
+      if (result.deliveryStatus !== 'confirmed' || !result.receipt) {
+        throw new Error(`${call.provider} writeback is pending; no delivery receipt`);
+      }
+      signal.throwIfAborted();
+      return result;
+    },
+  };
+  return invokeHelper(call, transport);
+}
+
+/** Existing lease/effect election protocol; provider verbs never enter the kernel. */
+export async function completeHelperDispatch(
+  client: JournalClient, dispatch: StepDispatchEvent, call: HelperCall, dataDir: string,
+): Promise<void> {
+  const surfacePath = `/${call.provider}`;
+  const output = await withWorkerLease(client, dispatch, async signal => {
+    const file = receiptPath(dataDir, dispatch.run_id, dispatch.step_id);
+    let receipt: unknown;
+    await client.performEffect({
+      runId: dispatch.run_id, stepId: dispatch.step_id, attempt: dispatch.attempt,
+      idempotencyKey: dispatch.idempotency_key, surfacePath,
+      revisionBefore: 'pending', revisionAfter: `${dispatch.run_id}:${dispatch.step_id}`,
+    }, async () => {
+      try { receipt = await readSlackReceipt(file); }
+      catch (error) {
+        if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
+        receipt = await writeback(call, dataDir, dispatch.run_id, dispatch.step_id, signal);
+        await atomicJson(file, receipt);
+      }
+      signal.throwIfAborted();
+    });
+    if (receipt === undefined) receipt = await readSlackReceipt(file);
+    return { ...call, idempotencyKey: `${dispatch.run_id}:${dispatch.step_id}`, receipt };
+  });
+  await client.stepComplete(dispatch.run_id, dispatch.step_id, dispatch.attempt,
+    dispatch.idempotency_key, 'success', { output, started_pins: dispatch.pins, end_pins: dispatch.pins,
+      effects: [{ surface_path: surfacePath, idempotency_key: dispatch.idempotency_key }] });
+}