feat(cli): run authored flows with direct input

docs/SURFACE.md

@@ -299,6 +299,7 @@ Gate 1 ships three CLI verbs over the journal protocol:
 ```text
 flows check [--json] <flow.yaml|spec.json>
 flows run [--json] [--data-dir <dir>] <flow.yaml|spec.json>
+flows run [--json] [--data-dir <dir>] <flow.ts> --input <inline-json-or-file>
 flows resume [--json] [--data-dir <dir>] <run-id>
 ```
 
@@ -309,6 +310,17 @@ existing run from its journal. The data directory defaults to `.relayflowd`.
 Neither verb starts the daemon implicitly. `--json` writes one report-shaped
 object to stdout while diagnostics remain on stderr.
 
+A direct `.flow.ts` run requires `--input`. When its argument names an existing
+regular file, the CLI parses that file as JSON; otherwise it parses the argument
+itself as inline JSON. Direct input is limited to 1,048,576 UTF-8 bytes; file
+size is checked before the file is read. Missing, invalid, or oversized input is
+refused before the CLI contacts `relayflowd`. After the journal connection is
+established, the authored body receives the parsed value as its second argument.
+Each awaited `f.run` executes through the journal-backed authored runtime, and
+JavaScript control flow observes the output read from `step.completed`.
+Unsupported headers, verbs, gates, and completion reasons fail closed rather
+than running through a second speculative compiler.
+
 The exit codes are part of the surface contract:
 
 | Exit | Outcome |

sdk/src/authored-flow-executor.ts

@@ -79,11 +79,12 @@ type JournalStepUsesStepCompletionReason = Assert<
  * `JournalClient`; values are read back from `step.completed` journal entries.
  * Unsupported headers, verbs, gates, or completion lowering fail closed.
  */
-export async function executeAuthoredFlow(
+export async function executeAuthoredFlow<Input = undefined>(
   handle: FlowHandle,
   journal: JournalClient,
+  input?: Input,
 ): Promise<AuthoredFlowExecutionResult> {
-  const definition = getAuthoredFlowDefinition(handle);
+  const definition = getAuthoredFlowDefinition<Input>(handle);
   const headerFields = Object.keys(definition.header);
   if (headerFields.length > 0) {
     throw new AuthoredFlowExecutionError(
@@ -182,7 +183,7 @@ export async function executeAuthoredFlow(
   let bodyFailed = false;
   let bodyFailure: unknown;
   try {
-    const bodyPromise = lifecycle.runBody(() => definition.body(context));
+    const bodyPromise = lifecycle.runBody(() => definition.body(context, input as Input));
     await bodyPromise;
   } catch (error) {
     bodyFailed = true;

sdk/src/authored-flow-loader.ts

@@ -0,0 +1,44 @@
+import { accessSync, constants } from 'node:fs';
+import { resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+import { getAuthoredFlowDefinition, type FlowHandle } from './authored-flow.js';
+
+export class AuthoredFlowLoadError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = 'AuthoredFlowLoadError';
+  }
+}
+
+/** Import and validate a direct-run module without executing its authored body. */
+export async function loadAuthoredFlow(path: string): Promise<FlowHandle> {
+  const absolutePath = resolve(path);
+  try {
+    accessSync(absolutePath, constants.R_OK);
+  } catch {
+    throw new AuthoredFlowLoadError(`Flow "${path}" is not readable.`);
+  }
+
+  let authoredModule: Record<string, unknown>;
+  try {
+    authoredModule = await import(pathToFileURL(absolutePath).href) as Record<string, unknown>;
+  } catch (error) {
+    throw new AuthoredFlowLoadError(
+      `Flow "${path}" could not be imported: ${errorMessage(error)}`,
+    );
+  }
+
+  const handle = authoredModule['default'] as FlowHandle;
+  try {
+    getAuthoredFlowDefinition(handle);
+  } catch (error) {
+    throw new AuthoredFlowLoadError(
+      `Flow "${path}" must default-export flow(...): ${errorMessage(error)}`,
+    );
+  }
+  return handle;
+}
+
+function errorMessage(error: unknown): string {
+  return error instanceof Error ? error.message : 'unknown authored-flow error';
+}

sdk/src/authored-flow.ts

@@ -12,10 +12,10 @@ import {
  * progress has no durable root journal. Internal lowering tests recover the
  * definition here without making that seam a supported runner.
  */
-export function getAuthoredFlowDefinition(
+export function getAuthoredFlowDefinition<Input = unknown>(
   handle: FlowHandle,
-): AuthoredFlowDefinition {
-  return getFlowDefinition(handle);
+): AuthoredFlowDefinition<Input> {
+  return getFlowDefinition<Input>(handle);
 }
 
 export type { AuthoredFlowDefinition, FlowHandle };

sdk/src/cli.ts

@@ -13,6 +13,8 @@ import {
   type RunExecution,
   type RunReport,
 } from './cli/run.js';
+import { runDirectFlow } from './cli/direct-run.js';
+import { isAuthoredFlowPath } from './direct-input.js';
 import { runHnMonitor } from './cli/hn-monitor.js';
 import { runTickRunner } from './cli/tick-runner.js';
 
@@ -26,7 +28,8 @@ export interface CliIo {
 type CliExitCode = 0 | 1 | 2 | 3;
 type ParsedArgs =
   | { command: 'check'; json: boolean; value: string }
-  | { command: 'run' | 'resume'; dataDir: string; json: boolean; value: string }
+  | { command: 'run'; dataDir: string; input: string | undefined; json: boolean; value: string }
+  | { command: 'resume'; dataDir: string; json: boolean; value: string }
   | { command: 'hn-monitor'; sub: 'start'; dataDir: string; specPath: string; pollIntervalMs: number | undefined }
   | { command: 'tick'; sub: 'start'; dataDir: string; specPath: string; scheduleId: string;
       intervalMs: number; epochMs: number | undefined; maxCatchUp: number | undefined;
@@ -37,6 +40,7 @@ const USAGE = [
   'Usage:',
   'flows check [--json] <flow.yaml|spec.json>',
   'flows run [--json] [--data-dir <dir>] <flow.yaml|spec.json>',
+  'flows run [--json] [--data-dir <dir>] <flow.ts> --input <inline-json-or-file>',
   'flows tick start --schedule-id <id> --interval-ms <ms> [--epoch-ms <ms>] [--max-catch-up <n>] [--poll-interval-ms <ms>] [--data-dir <dir>] <spec.json>',
   'flows resume [--json] [--data-dir <dir>] <run-id>',
   'flows hn-monitor start [--data-dir <dir>] [--poll-interval-ms <n>] <spec.json>',
@@ -107,7 +111,14 @@ export async function runCli(
   }
 
   const execution = parsed.command === 'run'
-    ? await runFlow(parsed.value, parsed.dataDir, { onWait: (progress) => emitWait(progress, io) })
+    ? isAuthoredFlowPath(parsed.value)
+      ? await runDirectFlow(
+          parsed.value,
+          parsed.input,
+          parsed.dataDir,
+          { onWait: (progress) => emitWait(progress, io) },
+        )
+      : await runFlow(parsed.value, parsed.dataDir, { onWait: (progress) => emitWait(progress, io) })
     : await resumeFlow(parsed.value, parsed.dataDir, { onWait: (progress) => emitWait(progress, io) });
   emitRunReport(execution, parsed.json, io);
   return execution.exitCode;
@@ -132,6 +143,8 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined {
   let json = false;
   let dataDir = DEFAULT_DATA_DIR;
   let sawDataDir = false;
+  let input: string | undefined;
+  let sawInput = false;
   const positionals: string[] = [];
   for (let index = 1; index < args.length; index += 1) {
     const argument = args[index]!;
@@ -148,14 +161,25 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined {
       index += 1;
       continue;
     }
+    if (argument === '--input') {
+      const value = args[index + 1];
+      if (command !== 'run' || sawInput || value === undefined || value.startsWith('--')) return undefined;
+      input = value;
+      sawInput = true;
+      index += 1;
+      continue;
+    }
     if (argument.startsWith('-')) return undefined;
     positionals.push(argument);
   }
   if (positionals.length !== 1) return undefined;
 
+  if (command === 'run' && input !== undefined && !isAuthoredFlowPath(positionals[0]!)) return undefined;
   return command === 'check'
     ? { command, json, value: positionals[0]! }
-    : { command, dataDir, json, value: positionals[0]! };
+    : command === 'run'
+      ? { command, dataDir, input, json, value: positionals[0]! }
+      : { command, dataDir, json, value: positionals[0]! };
 }
 
 function parseHnMonitorArgs(rest: readonly string[]): ParsedArgs | undefined {

sdk/src/cli/check.ts

@@ -63,7 +63,20 @@ class CheckFailure extends Error {
 export function checkFlow(path: string): CheckExecution {
   const absolutePath = resolve(path);
   try {
-    const authoring = readFlow(absolutePath);
+    const flow = readFlow(absolutePath);
+    return checkAuthoredFlow(flow, path);
+  } catch (error) {
+    const failure = error instanceof CheckFailure
+      ? error
+      : new CheckFailure('invalid_spec', `Flow "${path}" could not be checked as a Relayflow spec.`);
+    return { report: inputFailureReport(failure, path) };
+  }
+}
+
+/** Preflight a validated authored flow through the same path as YAML/JSON. */
+export function checkAuthoredFlow(authoring: FlowSpec, path: string): CheckExecution {
+  const absolutePath = resolve(path);
+  try {
     const config = readProjectConfig(dirname(absolutePath));
     const probes = systemProbes(dirname(absolutePath), config);
     const result = preflight(authoring, {

sdk/src/cli/direct-run.ts

@@ -0,0 +1,86 @@
+import {
+  AuthoredFlowExecutionError,
+  executeAuthoredFlow,
+} from '../authored-flow-executor.js';
+import { AuthoredFlowLoadError, loadAuthoredFlow } from '../authored-flow-loader.js';
+import { DirectInputError, parseDirectInput } from '../direct-input.js';
+import { JournalClient } from '../journal-client.js';
+import { inputFailureReport } from './check.js';
+import {
+  connect,
+  emptyReport,
+  fromCheckReport,
+  protocolFailure,
+  socketFor,
+  type RunExecution,
+  type RunLifecycleOptions,
+  type RunReport,
+} from './run.js';
+
+export async function runDirectFlow(
+  path: string,
+  inputArgument: string | undefined,
+  dataDir: string,
+  _options: RunLifecycleOptions = {},
+): Promise<RunExecution> {
+  let input: unknown;
+  try {
+    input = parseDirectInput(inputArgument);
+  } catch (error) {
+    const failure = error instanceof DirectInputError ? error : {
+      kind: 'input_invalid' as const,
+      message: 'Direct input could not be parsed.',
+    };
+    return {
+      exitCode: 2,
+      report: fromCheckReport('run', inputFailureReport(failure, path)),
+    };
+  }
+
+  const socketPath = socketFor(dataDir);
+  const base: RunReport = { ...emptyReport('run'), path };
+  const client = new JournalClient(socketPath);
+  const connected = await connect(client, 'run', dataDir, base);
+  if (connected !== undefined) return connected;
+
+  try {
+    const handle = await loadAuthoredFlow(path);
+    const result = await executeAuthoredFlow(handle, client, input);
+    const terminal = result.journalSteps.at(-1);
+    if (terminal === undefined) {
+      return protocolFailure('run', base, socketPath, new Error(
+        `authored flow "${result.name}" completed without a journal step`,
+      ));
+    }
+    return {
+      exitCode: 0,
+      report: {
+        ...base,
+        ok: true,
+        runId: terminal.runId,
+        socketPath,
+        status: 'completed',
+        completionReason: result.completionReason,
+        completedSteps: result.journalSteps.length,
+      },
+    };
+  } catch (error) {
+    if (error instanceof AuthoredFlowLoadError
+      || (error instanceof AuthoredFlowExecutionError && error.code === 'unsupported_header')) {
+      return {
+        exitCode: 2,
+        report: {
+          ...fromCheckReport('run', inputFailureReport({
+            kind: 'invalid_spec',
+            message: error.message,
+          }, path)),
+          socketPath,
+        },
+      };
+    }
+    const runId = error instanceof AuthoredFlowExecutionError ? error.runId : undefined;
+    return protocolFailure('run', base, socketPath, error, runId);
+  } finally {
+    client.close();
+  }
+}

sdk/src/cli/run.ts

@@ -71,13 +71,21 @@ export async function runFlow(
     return { exitCode: 2, report: fromCheckReport('run', checked.report) };
   }
 
+  return executeCheckedFlow(checked, dataDir, options);
+}
+
+async function executeCheckedFlow(
+  checked: ReturnType<typeof checkFlow>,
+  dataDir: string,
+  options: RunLifecycleOptions,
+): Promise<RunExecution> {
   const socketPath = socketFor(dataDir);
   const client = new JournalClient(socketPath);
   const connected = await connect(client, 'run', dataDir, checked.report);
   if (connected !== undefined) return connected;
 
   try {
-    const spec = toKernelSpec(checked.flow);
+    const spec = toKernelSpec(checked.flow!);
     const outcome = await client.runStart(spec);
     return await classifyOutcome(client, 'run', outcome, checked.report, socketPath, options);
   } catch (error) {
@@ -123,7 +131,7 @@ export async function resumeFlow(
   }
 }
 
-async function connect(
+export async function connect(
   client: JournalClient,
   command: RunCommand,
   dataDir: string,
@@ -318,7 +326,7 @@ async function waitForRunningStep(
   }
 }
 
-function protocolFailure(
+export function protocolFailure(
   command: RunCommand,
   base: CheckReport | RunReport,
   socketPath: string,
@@ -340,7 +348,7 @@ function protocolFailure(
   };
 }
 
-function fromCheckReport(command: RunCommand, report: CheckReport): RunReport {
+export function fromCheckReport(command: RunCommand, report: CheckReport): RunReport {
   return {
     ok: false,
     command,
@@ -351,15 +359,15 @@ function fromCheckReport(command: RunCommand, report: CheckReport): RunReport {
   };
 }
 
-function emptyReport(command: RunCommand): RunReport {
+export function emptyReport(command: RunCommand): RunReport {
   return { ok: false, command, resolutions: [], diagnostics: [] };
 }
 
 function fromBase(command: RunCommand, base: CheckReport | RunReport): RunReport {
   return 'command' in base ? base : fromCheckReport(command, base);
 }
 
-function socketFor(dataDir: string): string {
+export function socketFor(dataDir: string): string {
   return join(resolve(dataDir), 'relayflowd.sock');
 }