Add flow runtime and Codex release flows

apps/flow-backend-systemd-local/package.json

@@ -0,0 +1,25 @@
+{
+	"name": "codex-flow-systemd-local",
+	"version": "0.1.0",
+	"description": "Local flow execution backend suitable for a systemd-managed service.",
+	"type": "module",
+	"private": true,
+	"license": "Apache-2.0",
+	"bin": {
+		"codex-flow-systemd-local": "./src/index.ts"
+	},
+	"scripts": {
+		"build": "tsc --noEmit",
+		"check:types": "tsc --noEmit",
+		"start": "bun ./src/index.ts serve",
+		"test": "bun test test/*.test.ts"
+	},
+	"dependencies": {
+		"@peezy.tech/flow-runtime": "workspace:*"
+	},
+	"devDependencies": {
+		"@types/bun": "catalog:",
+		"@types/node": "catalog:",
+		"typescript": "catalog:"
+	}
+}

apps/flow-backend-systemd-local/src/backend.ts

@@ -0,0 +1,174 @@
+import { createHash } from "node:crypto";
+import { mkdir } from "node:fs/promises";
+import path from "node:path";
+import {
+	discoverFlows,
+	matchingSteps,
+	type FlowEvent,
+	type FlowStep,
+	type LoadedFlow,
+} from "@peezy.tech/flow-runtime";
+import type { FlowBackendConfig } from "./config.ts";
+import { executeCommand, flowCommand, parseRunnerResult } from "./executor.ts";
+import { FlowBackendStore, type FlowRunRecord } from "./store.ts";
+
+export type DispatchFlowEventOptions = {
+	config: FlowBackendConfig;
+	store: FlowBackendStore;
+	event: FlowEvent;
+	wait?: boolean;
+	env?: Record<string, string | undefined>;
+};
+
+export type DispatchFlowEventResult = {
+	status: "accepted" | "duplicate";
+	eventId: string;
+	runIds: string[];
+	matched: number;
+};
+
+export async function dispatchFlowEvent(options: DispatchFlowEventOptions): Promise<DispatchFlowEventResult> {
+	const inserted = options.store.insertEvent(options.event);
+	if (!inserted) {
+		return {
+			status: "duplicate",
+			eventId: options.event.id,
+			runIds: options.store.listRunsByEvent(options.event.id).map((run) => run.id),
+			matched: 0,
+		};
+	}
+
+	const eventPath = await writeEventFile(options.config.dataDir, options.event);
+	const flows = await discoverFlows({ cwd: options.config.cwd });
+	const matches = await matchingSteps(flows, options.event);
+	const promises: Array<Promise<void>> = [];
+	for (const match of matches) {
+		const run = createRunRecord(options.config, options.event, match.flow, match.step, eventPath);
+		options.store.createRun(run);
+		const promise = executeAndRecord({
+			config: options.config,
+			store: options.store,
+			run,
+			env: options.env,
+		});
+		if (options.wait) {
+			promises.push(promise);
+		} else {
+			promise.catch((error) => {
+				options.store.markRunCompleted(run.id, {
+					status: "failed",
+					stdout: "",
+					stderr: "",
+					error: error instanceof Error ? error.message : String(error),
+				});
+			});
+		}
+	}
+	if (promises.length > 0) {
+		await Promise.all(promises);
+	}
+	return {
+		status: "accepted",
+		eventId: options.event.id,
+		runIds: matches.map((match) => runId(options.event.id, match.flow.manifest.name, match.step.name)),
+		matched: matches.length,
+	};
+}
+
+export async function readFlowEvent(pathValue: string): Promise<FlowEvent> {
+	return normalizeFlowEvent(JSON.parse(await Bun.file(path.resolve(pathValue)).text()) as unknown);
+}
+
+export function normalizeFlowEvent(value: unknown): FlowEvent {
+	if (!isRecord(value) || typeof value.id !== "string" || typeof value.type !== "string") {
+		throw new Error("FlowEvent requires string id and type");
+	}
+	return {
+		receivedAt: typeof value.receivedAt === "string" ? value.receivedAt : new Date().toISOString(),
+		payload: isRecord(value.payload) ? value.payload : {},
+		...value,
+	} as FlowEvent;
+}
+
+async function executeAndRecord(options: {
+	config: FlowBackendConfig;
+	store: FlowBackendStore;
+	run: FlowRunRecord;
+	env?: Record<string, string | undefined>;
+}): Promise<void> {
+	const command = flowCommand({
+		config: options.config,
+		runId: options.run.id,
+		eventPath: options.run.eventPath,
+		flowName: options.run.flowName,
+		stepName: options.run.stepName,
+		env: options.env,
+	});
+	options.store.markRunRunning(options.run.id, JSON.stringify(command), command.unit);
+	let result: Awaited<ReturnType<typeof executeCommand>>;
+	try {
+		result = await executeCommand(command, options.config, options.env);
+	} catch (error) {
+		options.store.markRunCompleted(options.run.id, {
+			status: "failed",
+			stdout: "",
+			stderr: "",
+			error: error instanceof Error ? error.message : String(error),
+		});
+		return;
+	}
+	const status = result.exitCode === 0 ? "completed" : "failed";
+	options.store.markRunCompleted(options.run.id, {
+		status,
+		resultJson: parseRunnerResult(result.stdout),
+		stdout: result.stdout,
+		stderr: result.stderr,
+		...(status === "failed" ? { error: `flow runner exited with ${result.exitCode ?? "unknown"}` } : {}),
+	});
+}
+
+function createRunRecord(
+	config: FlowBackendConfig,
+	event: FlowEvent,
+	flow: LoadedFlow,
+	step: FlowStep,
+	eventPath: string,
+): FlowRunRecord {
+	return {
+		id: runId(event.id, flow.manifest.name, step.name),
+		eventId: event.id,
+		flowName: flow.manifest.name,
+		stepName: step.name,
+		status: "queued",
+		backend: "systemd-local",
+		executor: config.executor,
+		eventPath,
+		createdAt: new Date().toISOString(),
+	};
+}
+
+function runId(eventId: string, flowName: string, stepName: string): string {
+	const hash = createHash("sha256")
+		.update(`${eventId}\0${flowName}\0${stepName}`)
+		.digest("hex")
+		.slice(0, 12);
+	return `run_${hash}`;
+}
+
+async function writeEventFile(dataDir: string, event: FlowEvent): Promise<string> {
+	const directory = path.join(dataDir, "events");
+	await mkdir(directory, { recursive: true });
+	const filePath = path.join(directory, `${safeFileName(event.id)}.json`);
+	await Bun.write(filePath, JSON.stringify(event, null, 2));
+	return filePath;
+}
+
+function safeFileName(value: string): string {
+	const hash = createHash("sha256").update(value).digest("hex").slice(0, 12);
+	const base = value.toLowerCase().replace(/[^a-z0-9._-]+/g, "-").replace(/^-+|-+$/g, "").slice(0, 64);
+	return `${base || "event"}-${hash}`;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+	return typeof value === "object" && value !== null && !Array.isArray(value);
+}

apps/flow-backend-systemd-local/src/config.ts

@@ -0,0 +1,191 @@
+import path from "node:path";
+
+export type FlowBackendExecutor = "direct" | "systemd-run";
+
+export type FlowBackendConfig = {
+	cwd: string;
+	dataDir: string;
+	host: string;
+	port: number;
+	secret?: string;
+	executor: FlowBackendExecutor;
+	bunCommand: string;
+	flowRunnerPath: string;
+	forwardEnv: string[];
+};
+
+export type FlowBackendCli =
+	| { kind: "help" }
+	| { kind: "serve"; config: FlowBackendConfig }
+	| { kind: "dispatch"; config: FlowBackendConfig; eventPath: string; wait: boolean };
+
+export function readConfig(
+	env: Record<string, string | undefined> = process.env,
+	overrides: Partial<FlowBackendConfig> = {},
+): FlowBackendConfig {
+	const cwd = path.resolve(overrides.cwd ?? env.CODEX_FLOW_BACKEND_CWD ?? process.cwd());
+	const dataDir = path.resolve(overrides.dataDir ?? env.CODEX_FLOW_BACKEND_DATA_DIR ?? path.join(cwd, ".codex", "flow-backend"));
+	return {
+		cwd,
+		dataDir,
+		host: overrides.host ?? env.CODEX_FLOW_BACKEND_HOST ?? "127.0.0.1",
+		port: overrides.port ?? numberEnv(env.CODEX_FLOW_BACKEND_PORT, 7345),
+		...(overrides.secret ?? env.CODEX_FLOW_BACKEND_SECRET
+			? { secret: overrides.secret ?? env.CODEX_FLOW_BACKEND_SECRET }
+			: {}),
+		executor: overrides.executor ?? executorEnv(env.CODEX_FLOW_BACKEND_EXECUTOR),
+		bunCommand: overrides.bunCommand ?? env.CODEX_FLOW_BACKEND_BUN ?? process.execPath,
+		flowRunnerPath: path.resolve(
+			overrides.flowRunnerPath ?? env.CODEX_FLOW_RUNNER_PATH ?? defaultFlowRunnerPath(),
+		),
+		forwardEnv: overrides.forwardEnv ?? forwardEnv(env.CODEX_FLOW_BACKEND_FORWARD_ENV),
+	};
+}
+
+export function parseCli(argv: string[], env: Record<string, string | undefined> = process.env): FlowBackendCli {
+	const command = argv[0];
+	if (!command || command === "help" || command === "-h" || command === "--help") {
+		return { kind: "help" };
+	}
+
+	let cwd: string | undefined;
+	let dataDir: string | undefined;
+	let host: string | undefined;
+	let port: number | undefined;
+	let secret: string | undefined;
+	let executor: FlowBackendExecutor | undefined;
+	let bunCommand: string | undefined;
+	let flowRunnerPath: string | undefined;
+	let wait = false;
+	let eventPath: string | undefined;
+	const rest = argv.slice(1);
+	for (let index = 0; index < rest.length; index += 1) {
+		const arg = rest[index];
+		if (!arg) {
+			continue;
+		}
+		if (arg === "--cwd") {
+			cwd = required(rest, ++index, arg);
+			continue;
+		}
+		if (arg === "--data-dir") {
+			dataDir = required(rest, ++index, arg);
+			continue;
+		}
+		if (arg === "--host") {
+			host = required(rest, ++index, arg);
+			continue;
+		}
+		if (arg === "--port") {
+			port = Number(required(rest, ++index, arg));
+			continue;
+		}
+		if (arg === "--secret") {
+			secret = required(rest, ++index, arg);
+			continue;
+		}
+		if (arg === "--executor") {
+			executor = executorEnv(required(rest, ++index, arg));
+			continue;
+		}
+		if (arg === "--bun") {
+			bunCommand = required(rest, ++index, arg);
+			continue;
+		}
+		if (arg === "--flow-runner") {
+			flowRunnerPath = required(rest, ++index, arg);
+			continue;
+		}
+		if (arg === "--event") {
+			eventPath = required(rest, ++index, arg);
+			continue;
+		}
+		if (arg === "--wait") {
+			wait = true;
+			continue;
+		}
+		throw new Error(`Unknown option: ${arg}`);
+	}
+
+	const config = readConfig(env, {
+		...(cwd ? { cwd } : {}),
+		...(dataDir ? { dataDir } : {}),
+		...(host ? { host } : {}),
+		...(port !== undefined ? { port } : {}),
+		...(secret ? { secret } : {}),
+		...(executor ? { executor } : {}),
+		...(bunCommand ? { bunCommand } : {}),
+		...(flowRunnerPath ? { flowRunnerPath } : {}),
+	});
+	if (command === "serve") {
+		return { kind: "serve", config };
+	}
+	if (command === "dispatch") {
+		if (!eventPath) {
+			throw new Error("dispatch requires --event <path>");
+		}
+		return { kind: "dispatch", config, eventPath, wait };
+	}
+	throw new Error(`Unknown command: ${command}`);
+}
+
+export function defaultFlowRunnerPath(): string {
+	return path.resolve(import.meta.dir, "..", "..", "flow-runner", "src", "index.ts");
+}
+
+export function helpText(): string {
+	return [
+		"Usage:",
+		"  codex-flow-systemd-local serve [--cwd <dir>] [--data-dir <dir>] [--host <host>] [--port <port>]",
+		"  codex-flow-systemd-local dispatch --event <event.json> [--cwd <dir>] [--data-dir <dir>] [--wait]",
+		"",
+		"Environment:",
+		"  CODEX_FLOW_BACKEND_SECRET       Optional HMAC secret for HTTP dispatches",
+		"  CODEX_FLOW_BACKEND_EXECUTOR     direct or systemd-run",
+		"  CODEX_FLOWS_ENABLE_CODE_MODE    Enables runner = \"code-mode\" steps",
+		"",
+	].join("\n");
+}
+
+function numberEnv(value: string | undefined, fallback: number): number {
+	if (!value) {
+		return fallback;
+	}
+	const parsed = Number(value);
+	return Number.isFinite(parsed) ? parsed : fallback;
+}
+
+function executorEnv(value: string | undefined): FlowBackendExecutor {
+	if (value === "systemd-run") {
+		return "systemd-run";
+	}
+	if (!value || value === "direct") {
+		return "direct";
+	}
+	throw new Error("executor must be direct or systemd-run");
+}
+
+function forwardEnv(value: string | undefined): string[] {
+	const defaults = [
+		"CODEX_FLOWS_ENABLE_CODE_MODE",
+		"CODEX_APP_SERVER_CODEX_COMMAND",
+		"CODEX_HOME",
+		"HOME",
+		"PATH",
+	];
+	if (!value) {
+		return defaults;
+	}
+	return value
+		.split(",")
+		.map((entry) => entry.trim())
+		.filter(Boolean);
+}
+
+function required(args: string[], index: number, flag: string): string {
+	const value = args[index];
+	if (!value) {
+		throw new Error(`${flag} requires a value`);
+	}
+	return value;
+}

apps/flow-backend-systemd-local/src/executor.ts

@@ -0,0 +1,113 @@
+import { createHash } from "node:crypto";
+import type { FlowBackendConfig } from "./config.ts";
+
+export type FlowCommandSpec = {
+	command: string;
+	args: string[];
+	unit?: string;
+};
+
+export type ExecuteFlowRunOptions = {
+	config: FlowBackendConfig;
+	runId: string;
+	eventPath: string;
+	flowName: string;
+	stepName: string;
+	env?: Record<string, string | undefined>;
+};
+
+export type ExecuteFlowRunResult = {
+	command: FlowCommandSpec;
+	exitCode: number | null;
+	stdout: string;
+	stderr: string;
+};
+
+export async function executeFlowRun(options: ExecuteFlowRunOptions): Promise<ExecuteFlowRunResult> {
+	const command = flowCommand(options);
+	const result = await executeCommand(command, options.config, options.env);
+	return { command, ...result };
+}
+
+export async function executeCommand(
+	command: FlowCommandSpec,
+	config: FlowBackendConfig,
+	env: Record<string, string | undefined> = process.env,
+): Promise<Omit<ExecuteFlowRunResult, "command">> {
+	const child = Bun.spawn([command.command, ...command.args], {
+		cwd: config.cwd,
+		env: forwardedEnv(config, env),
+		stdout: "pipe",
+		stderr: "pipe",
+	});
+	const [stdout, stderr, exitCode] = await Promise.all([
+		new Response(child.stdout).text(),
+		new Response(child.stderr).text(),
+		child.exited,
+	]);
+	return { exitCode, stdout, stderr };
+}
+
+export function flowCommand(options: ExecuteFlowRunOptions): FlowCommandSpec {
+	const runnerArgs = [
+		options.config.flowRunnerPath,
+		"--cwd",
+		options.config.cwd,
+		"run",
+		options.flowName,
+		options.stepName,
+		"--event",
+		options.eventPath,
+	];
+	if (options.config.executor === "direct") {
+		return { command: options.config.bunCommand, args: runnerArgs };
+	}
+	const unit = `codex-flow-${safeUnit(options.runId)}`;
+	return {
+		command: "systemd-run",
+		unit,
+		args: [
+			"--user",
+			"--collect",
+			"--wait",
+			`--unit=${unit}`,
+			`--working-directory=${options.config.cwd}`,
+			...systemdSetEnvArgs(options.config, options.env ?? process.env),
+			options.config.bunCommand,
+			...runnerArgs,
+		],
+	};
+}
+
+export function parseRunnerResult(stdout: string): string | undefined {
+	const trimmed = stdout.trim();
+	if (!trimmed.startsWith("{")) {
+		return undefined;
+	}
+	try {
+		return JSON.stringify(JSON.parse(trimmed));
+	} catch {
+		return undefined;
+	}
+}
+
+function forwardedEnv(config: FlowBackendConfig, env: Record<string, string | undefined>): Record<string, string> {
+	const next: Record<string, string> = {};
+	const source: Record<string, string | undefined> = { ...process.env, ...env };
+	for (const name of config.forwardEnv) {
+		const value = source[name];
+		if (value !== undefined) {
+			next[name] = value;
+		}
+	}
+	return next;
+}
+
+function systemdSetEnvArgs(config: FlowBackendConfig, env: Record<string, string | undefined>): string[] {
+	return Object.entries(forwardedEnv(config, env)).map(([key, value]) => `--setenv=${key}=${value}`);
+}
+
+function safeUnit(value: string): string {
+	const hash = createHash("sha256").update(value).digest("hex").slice(0, 10);
+	return `${value.toLowerCase().replace(/[^a-z0-9-]+/g, "-").replace(/^-+|-+$/g, "").slice(0, 48)}-${hash}`;
+}

apps/flow-backend-systemd-local/src/index.ts

@@ -0,0 +1,38 @@
+#!/usr/bin/env bun
+import path from "node:path";
+import { dispatchFlowEvent, readFlowEvent } from "./backend.ts";
+import { helpText, parseCli } from "./config.ts";
+import { serveFlowBackend } from "./server.ts";
+import { FlowBackendStore } from "./store.ts";
+
+await main().catch((error) => {
+	process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
+	process.exitCode = 1;
+});
+
+async function main(): Promise<void> {
+	const cli = parseCli(Bun.argv.slice(2));
+	if (cli.kind === "help") {
+		process.stdout.write(helpText());
+		return;
+	}
+	if (cli.kind === "serve") {
+		const server = serveFlowBackend(cli.config);
+		process.stdout.write(`codex-flow-systemd-local listening on http://${server.hostname}:${server.port}\n`);
+		return new Promise(() => undefined);
+	}
+	const store = new FlowBackendStore(path.join(cli.config.dataDir, "flow-backend.sqlite"));
+	try {
+		const event = await readFlowEvent(cli.eventPath);
+		const result = await dispatchFlowEvent({
+			config: cli.config,
+			store,
+			event,
+			wait: cli.wait,
+			env: process.env,
+		});
+		process.stdout.write(`${JSON.stringify(result, null, 2)}\n`);
+	} finally {
+		store.close();
+	}
+}

apps/flow-backend-systemd-local/src/server.ts

@@ -0,0 +1,43 @@
+import path from "node:path";
+import type { FlowBackendConfig } from "./config.ts";
+import { dispatchFlowEvent, normalizeFlowEvent } from "./backend.ts";
+import { requestSignature, verifyBodySignature } from "./signature.ts";
+import { FlowBackendStore } from "./store.ts";
+
+export function serveFlowBackend(config: FlowBackendConfig): ReturnType<typeof Bun.serve> {
+	const store = new FlowBackendStore(path.join(config.dataDir, "flow-backend.sqlite"));
+	return Bun.serve({
+		hostname: config.host,
+		port: config.port,
+		async fetch(request) {
+			const url = new URL(request.url);
+			if (request.method === "GET" && url.pathname === "/healthz") {
+				return json({ ok: true });
+			}
+			if (request.method === "POST" && (url.pathname === "/events" || url.pathname === "/flow-events")) {
+				const body = await request.text();
+				if (config.secret && !verifyBodySignature(config.secret, body, requestSignature(request.headers))) {
+					return json({ error: "invalid signature" }, 401);
+				}
+				const event = normalizeFlowEvent(JSON.parse(body) as unknown);
+				const result = await dispatchFlowEvent({ config, store, event });
+				return json(result, 202);
+			}
+			if (request.method === "GET" && url.pathname === "/runs") {
+				const eventId = url.searchParams.get("eventId");
+				if (!eventId) {
+					return json({ error: "missing eventId" }, 400);
+				}
+				return json({ eventId, runs: store.listRunsByEvent(eventId) });
+			}
+			return json({ error: "not found" }, 404);
+		},
+	});
+}
+
+function json(value: unknown, status = 200): Response {
+	return new Response(JSON.stringify(value, null, 2), {
+		status,
+		headers: { "content-type": "application/json" },
+	});
+}

apps/flow-backend-systemd-local/src/signature.ts

@@ -0,0 +1,18 @@
+import { createHmac, timingSafeEqual } from "node:crypto";
+
+export function signBody(secret: string, body: string): string {
+	return `sha256=${createHmac("sha256", secret).update(body).digest("hex")}`;
+}
+
+export function verifyBodySignature(secret: string, body: string, signature: string | null): boolean {
+	if (!signature?.startsWith("sha256=")) {
+		return false;
+	}
+	const expected = Buffer.from(signBody(secret, body));
+	const actual = Buffer.from(signature);
+	return expected.length === actual.length && timingSafeEqual(expected, actual);
+}
+
+export function requestSignature(headers: Headers): string | null {
+	return headers.get("x-flow-signature-256") ?? headers.get("x-patchbay-flow-signature-256");
+}

apps/flow-backend-systemd-local/src/store.ts

@@ -0,0 +1,198 @@
+import { mkdirSync } from "node:fs";
+import path from "node:path";
+import { Database } from "bun:sqlite";
+import type { FlowEvent } from "@peezy.tech/flow-runtime";
+
+export type FlowRunStatus = "queued" | "running" | "completed" | "failed";
+
+export type FlowRunRecord = {
+	id: string;
+	eventId: string;
+	flowName: string;
+	stepName: string;
+	status: FlowRunStatus;
+	backend: "systemd-local";
+	executor: string;
+	unit?: string;
+	eventPath: string;
+	commandJson?: string;
+	resultJson?: string;
+	stdout?: string;
+	stderr?: string;
+	error?: string;
+	createdAt: string;
+	startedAt?: string;
+	completedAt?: string;
+};
+
+export class FlowBackendStore {
+	readonly dbPath: string;
+	#db: Database;
+
+	constructor(dbPath: string) {
+		this.dbPath = dbPath;
+		mkdirSync(path.dirname(dbPath), { recursive: true });
+		this.#db = new Database(dbPath);
+		this.#db.exec(`
+			create table if not exists flow_events (
+				id text primary key,
+				type text not null,
+				source text,
+				occurred_at text,
+				received_at text not null,
+				payload_json text not null,
+				raw_json text not null,
+				created_at text not null
+			);
+			create table if not exists flow_runs (
+				id text primary key,
+				event_id text not null,
+				flow_name text not null,
+				step_name text not null,
+				status text not null,
+				backend text not null,
+				executor text not null,
+				unit text,
+				event_path text not null,
+				command_json text,
+				result_json text,
+				stdout text,
+				stderr text,
+				error text,
+				created_at text not null,
+				started_at text,
+				completed_at text
+			);
+			create index if not exists flow_runs_event_id_idx on flow_runs(event_id);
+		`);
+	}
+
+	insertEvent(event: FlowEvent): boolean {
+		const result = this.#db
+			.query(
+				`insert or ignore into flow_events
+					(id, type, source, occurred_at, received_at, payload_json, raw_json, created_at)
+					values ($id, $type, $source, $occurredAt, $receivedAt, $payloadJson, $rawJson, $createdAt)`,
+			)
+			.run({
+				$id: event.id,
+				$type: event.type,
+				$source: event.source ?? null,
+				$occurredAt: event.occurredAt ?? null,
+				$receivedAt: event.receivedAt,
+				$payloadJson: JSON.stringify(event.payload),
+				$rawJson: JSON.stringify(event),
+				$createdAt: new Date().toISOString(),
+			});
+		return result.changes > 0;
+	}
+
+	createRun(record: FlowRunRecord): void {
+		this.#db
+			.query(
+				`insert into flow_runs
+					(id, event_id, flow_name, step_name, status, backend, executor, unit, event_path,
+						command_json, result_json, stdout, stderr, error, created_at, started_at, completed_at)
+					values
+					($id, $eventId, $flowName, $stepName, $status, $backend, $executor, $unit, $eventPath,
+						$commandJson, $resultJson, $stdout, $stderr, $error, $createdAt, $startedAt, $completedAt)`,
+			)
+			.run(runParams(record));
+	}
+
+	markRunRunning(id: string, commandJson: string, unit?: string): void {
+		this.#db
+			.query(
+				`update flow_runs
+					set status = 'running', started_at = $startedAt, command_json = $commandJson, unit = $unit
+					where id = $id`,
+			)
+			.run({
+				$id: id,
+				$startedAt: new Date().toISOString(),
+				$commandJson: commandJson,
+				$unit: unit ?? null,
+			});
+	}
+
+	markRunCompleted(id: string, values: { status: FlowRunStatus; resultJson?: string; stdout: string; stderr: string; error?: string }): void {
+		this.#db
+			.query(
+				`update flow_runs
+					set status = $status, completed_at = $completedAt, result_json = $resultJson,
+						stdout = $stdout, stderr = $stderr, error = $error
+					where id = $id`,
+			)
+			.run({
+				$id: id,
+				$status: values.status,
+				$completedAt: new Date().toISOString(),
+				$resultJson: values.resultJson ?? null,
+				$stdout: values.stdout,
+				$stderr: values.stderr,
+				$error: values.error ?? null,
+			});
+	}
+
+	listRunsByEvent(eventId: string): FlowRunRecord[] {
+		return this.#db
+			.query("select * from flow_runs where event_id = $eventId order by created_at, id")
+			.all({ $eventId: eventId })
+			.map(rowToRunRecord);
+	}
+
+	close(): void {
+		this.#db.close();
+	}
+}
+
+function runParams(record: FlowRunRecord): Record<string, string | null> {
+	return {
+		$id: record.id,
+		$eventId: record.eventId,
+		$flowName: record.flowName,
+		$stepName: record.stepName,
+		$status: record.status,
+		$backend: record.backend,
+		$executor: record.executor,
+		$unit: record.unit ?? null,
+		$eventPath: record.eventPath,
+		$commandJson: record.commandJson ?? null,
+		$resultJson: record.resultJson ?? null,
+		$stdout: record.stdout ?? null,
+		$stderr: record.stderr ?? null,
+		$error: record.error ?? null,
+		$createdAt: record.createdAt,
+		$startedAt: record.startedAt ?? null,
+		$completedAt: record.completedAt ?? null,
+	};
+}
+
+function rowToRunRecord(row: unknown): FlowRunRecord {
+	if (!isRecord(row)) {
+		throw new Error("invalid run row");
+	}
+	return {
+		id: String(row.id),
+		eventId: String(row.event_id),
+		flowName: String(row.flow_name),
+		stepName: String(row.step_name),
+		status: String(row.status) as FlowRunStatus,
+		backend: "systemd-local",
+		executor: String(row.executor),
+		...(typeof row.unit === "string" ? { unit: row.unit } : {}),
+		eventPath: String(row.event_path),
+		...(typeof row.command_json === "string" ? { commandJson: row.command_json } : {}),
+		...(typeof row.result_json === "string" ? { resultJson: row.result_json } : {}),
+		...(typeof row.stdout === "string" ? { stdout: row.stdout } : {}),
+		...(typeof row.stderr === "string" ? { stderr: row.stderr } : {}),
+		...(typeof row.error === "string" ? { error: row.error } : {}),
+		createdAt: String(row.created_at),
+		...(typeof row.started_at === "string" ? { startedAt: row.started_at } : {}),
+		...(typeof row.completed_at === "string" ? { completedAt: row.completed_at } : {}),
+	};
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+	return typeof value === "object" && value !== null && !Array.isArray(value);
+}

apps/flow-backend-systemd-local/test/backend.test.ts

@@ -0,0 +1,121 @@
+import { expect, test } from "bun:test";
+import { mkdir, mkdtemp, rm } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { dispatchFlowEvent } from "../src/backend.ts";
+import { readConfig } from "../src/config.ts";
+import { flowCommand } from "../src/executor.ts";
+import { signBody, verifyBodySignature } from "../src/signature.ts";
+import { FlowBackendStore } from "../src/store.ts";
+
+test("signs and verifies dispatch bodies", () => {
+	const body = JSON.stringify({ id: "event-1" });
+	const signature = signBody("secret", body);
+
+	expect(verifyBodySignature("secret", body, signature)).toBe(true);
+	expect(verifyBodySignature("secret", `${body}\n`, signature)).toBe(false);
+});
+
+test("dispatches matching flow steps and records runs", async () => {
+	const directory = await mkdtemp(path.join(os.tmpdir(), "flow-backend-"));
+	try {
+		await writeFlow(directory);
+		const config = readConfig(
+			{},
+			{
+				cwd: directory,
+				dataDir: path.join(directory, ".codex", "flow-backend"),
+				executor: "direct",
+				bunCommand: process.execPath,
+			},
+		);
+		const store = new FlowBackendStore(path.join(config.dataDir, "flow-backend.sqlite"));
+		try {
+			const result = await dispatchFlowEvent({
+				config,
+				store,
+				wait: true,
+				env: {},
+				event: {
+					id: "event-1",
+					type: "demo.event",
+					receivedAt: "2026-05-13T00:00:00.000Z",
+					payload: { name: "Ada" },
+				},
+			});
+
+			expect(result).toMatchObject({ status: "accepted", eventId: "event-1", matched: 1 });
+			const runs = store.listRunsByEvent("event-1");
+			expect(runs).toHaveLength(1);
+			expect(runs[0]).toMatchObject({
+				flowName: "demo",
+				stepName: "hello",
+				status: "completed",
+			});
+			expect(runs[0]?.stdout).toContain("hello Ada");
+		} finally {
+			store.close();
+		}
+	} finally {
+		await rm(directory, { recursive: true, force: true });
+	}
+});
+
+test("builds systemd-run commands without executing them", () => {
+	const config = readConfig({}, { cwd: "/tmp/project", executor: "systemd-run", bunCommand: "/usr/bin/bun" });
+	const command = flowCommand({
+		config,
+		runId: "run_123",
+		eventPath: "/tmp/event.json",
+		flowName: "demo",
+		stepName: "hello",
+		env: { CODEX_FLOWS_ENABLE_CODE_MODE: "1" },
+	});
+
+	expect(command.command).toBe("systemd-run");
+	expect(command.args).toContain("--user");
+	expect(command.args).toContain("--wait");
+	expect(command.args).toContain("--setenv=CODEX_FLOWS_ENABLE_CODE_MODE=1");
+	expect(command.args).toContain("/usr/bin/bun");
+});
+
+async function writeFlow(root: string): Promise<void> {
+	const flowRoot = path.join(root, "flows", "demo");
+	await mkdir(path.join(flowRoot, "exec"), { recursive: true });
+	await mkdir(path.join(flowRoot, "schemas"), { recursive: true });
+	await Bun.write(
+		path.join(flowRoot, "flow.toml"),
+		[
+			'name = "demo"',
+			"version = 1",
+			"",
+			"[[steps]]",
+			'name = "hello"',
+			'runner = "bun"',
+			'script = "exec/hello.ts"',
+			"timeout_ms = 30000",
+			"",
+			"[steps.trigger]",
+			'type = "demo.event"',
+			'schema = "schemas/demo-event.schema.json"',
+			"",
+		].join("\n"),
+	);
+	await Bun.write(
+		path.join(flowRoot, "schemas/demo-event.schema.json"),
+		JSON.stringify({
+			type: "object",
+			required: ["name"],
+			properties: { name: { type: "string" } },
+		}),
+	);
+	await Bun.write(
+		path.join(flowRoot, "exec/hello.ts"),
+		[
+			"const context = JSON.parse(await Bun.stdin.text());",
+			"const name = context.flow.event.payload.name;",
+			"console.log(`FLOW_RESULT ${JSON.stringify({ status: 'completed', message: `hello ${name}` })}`);",
+			"",
+		].join("\n"),
+	);
+}

apps/flow-backend-systemd-local/tsconfig.json

@@ -0,0 +1,21 @@
+{
+	"compilerOptions": {
+		"module": "ESNext",
+		"moduleResolution": "Bundler",
+		"allowImportingTsExtensions": true,
+		"esModuleInterop": true,
+		"resolveJsonModule": true,
+		"target": "ES2022",
+		"lib": ["ESNext"],
+		"strict": true,
+		"skipLibCheck": true,
+		"noUncheckedIndexedAccess": true,
+		"isolatedModules": true,
+		"verbatimModuleSyntax": true,
+		"forceConsistentCasingInFileNames": true,
+		"noEmit": true,
+		"types": ["node", "bun"],
+		"rootDir": "."
+	},
+	"include": ["src/**/*.ts", "test/**/*.ts"]
+}

apps/flow-runner/package.json

@@ -0,0 +1,24 @@
+{
+	"name": "codex-flow-runner",
+	"version": "0.1.0",
+	"description": "CLI for listing, firing, and running Codex flow packages.",
+	"type": "module",
+	"private": true,
+	"license": "Apache-2.0",
+	"bin": {
+		"codex-flow-runner": "./src/index.ts"
+	},
+	"scripts": {
+		"build": "tsc --noEmit",
+		"check:types": "tsc --noEmit",
+		"test": "bun test test/*.test.ts"
+	},
+	"dependencies": {
+		"@peezy.tech/flow-runtime": "workspace:*"
+	},
+	"devDependencies": {
+		"@types/bun": "catalog:",
+		"@types/node": "catalog:",
+		"typescript": "catalog:"
+	}
+}

apps/flow-runner/src/index.ts

@@ -0,0 +1,172 @@
+#!/usr/bin/env bun
+import path from "node:path";
+import {
+	discoverFlows,
+	matchingSteps,
+	runFlowStep,
+	type FlowEvent,
+	type LoadedFlow,
+	type FlowStep,
+} from "@peezy.tech/flow-runtime";
+
+type Cli =
+	| { kind: "help" }
+	| { kind: "list"; cwd: string }
+	| { kind: "fire"; cwd: string; eventPath: string }
+	| { kind: "run"; cwd: string; flow: string; step: string; eventPath: string };
+
+await main().catch((error) => {
+	process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
+	process.exitCode = 1;
+});
+
+async function main(): Promise<void> {
+	const cli = parseArgs(Bun.argv.slice(2));
+	if (cli.kind === "help") {
+		process.stdout.write(helpText());
+		return;
+	}
+	const flows = await discoverFlows({ cwd: cli.cwd });
+	if (cli.kind === "list") {
+		for (const flow of flows) {
+			process.stdout.write(`${flow.manifest.name}\t${flow.root}\n`);
+		}
+		return;
+	}
+	const event = await readEvent(cli.eventPath);
+	if (cli.kind === "fire") {
+		const matches = await matchingSteps(flows, event);
+		const results = [];
+		for (const match of matches) {
+			results.push(await runAndReport(match.flow, match.step, event));
+		}
+		process.stdout.write(`${JSON.stringify({ eventId: event.id, results }, null, 2)}\n`);
+		return;
+	}
+	const flow = requireFlow(flows, cli.flow);
+	const step = requireStep(flow, cli.step);
+	const result = await runAndReport(flow, step, event);
+	process.stdout.write(`${JSON.stringify(result, null, 2)}\n`);
+}
+
+async function runAndReport(flow: LoadedFlow, step: FlowStep, event: FlowEvent): Promise<Record<string, unknown>> {
+	const result = await runFlowStep({
+		flow,
+		step,
+		event,
+		env: process.env,
+		codeMode: {
+			codexCommand: process.env.CODEX_APP_SERVER_CODEX_COMMAND,
+			codexHome: process.env.CODEX_HOME,
+			stream: true,
+		},
+	});
+	return {
+		flow: flow.manifest.name,
+		step: step.name,
+		result,
+	};
+}
+
+async function readEvent(eventPath: string): Promise<FlowEvent> {
+	const parsed = JSON.parse(await Bun.file(path.resolve(eventPath)).text()) as unknown;
+	if (!isRecord(parsed) || typeof parsed.id !== "string" || typeof parsed.type !== "string") {
+		throw new Error("event file must contain at least string id and type");
+	}
+	return {
+		receivedAt: new Date().toISOString(),
+		payload: {},
+		...parsed,
+	} as FlowEvent;
+}
+
+function requireFlow(flows: LoadedFlow[], name: string): LoadedFlow {
+	const flow = flows.find((entry) => entry.manifest.name === name);
+	if (!flow) {
+		throw new Error(`Unknown flow: ${name}`);
+	}
+	return flow;
+}
+
+function requireStep(flow: LoadedFlow, name: string): FlowStep {
+	const step = flow.manifest.steps.find((entry) => entry.name === name);
+	if (!step) {
+		throw new Error(`Unknown step ${name} in flow ${flow.manifest.name}`);
+	}
+	return step;
+}
+
+function parseArgs(argv: string[]): Cli {
+	let cwd = process.cwd();
+	const args: string[] = [];
+	for (let index = 0; index < argv.length; index += 1) {
+		const arg = argv[index];
+		if (!arg) {
+			continue;
+		}
+		if (arg === "--cwd") {
+			cwd = path.resolve(required(argv, ++index, arg));
+			continue;
+		}
+		if (arg.startsWith("--cwd=")) {
+			cwd = path.resolve(arg.slice("--cwd=".length));
+			continue;
+		}
+		args.push(arg);
+	}
+
+	const command = args[0];
+	if (!command || command === "-h" || command === "--help" || command === "help") {
+		return { kind: "help" };
+	}
+	if (command === "list") {
+		return { kind: "list", cwd };
+	}
+	if (command === "fire") {
+		return { kind: "fire", cwd, eventPath: eventPathArg(args, 1) };
+	}
+	if (command === "run") {
+		const flow = args[1];
+		const step = args[2];
+		if (!flow || !step) {
+			throw new Error("run requires <flow> <step>");
+		}
+		return { kind: "run", cwd, flow, step, eventPath: eventPathArg(args, 3) };
+	}
+	throw new Error(`Unknown command: ${command}`);
+}
+
+function eventPathArg(args: string[], start: number): string {
+	for (let index = start; index < args.length; index += 1) {
+		const arg = args[index];
+		if (arg === "--event") {
+			return required(args, index + 1, "--event");
+		}
+		if (arg?.startsWith("--event=")) {
+			return arg.slice("--event=".length);
+		}
+	}
+	throw new Error("missing --event <path>");
+}
+
+function required(args: string[], index: number, flag: string): string {
+	const value = args[index];
+	if (!value) {
+		throw new Error(`${flag} requires a value`);
+	}
+	return value;
+}
+
+function helpText(): string {
+	return [
+		"Usage:",
+		"  codex-flow-runner [--cwd <dir>] list",
+		"  codex-flow-runner [--cwd <dir>] fire --event <event.json>",
+		"  codex-flow-runner [--cwd <dir>] run <flow> <step> --event <event.json>",
+		"",
+	].join("\n");
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+	return typeof value === "object" && value !== null && !Array.isArray(value);
+}

apps/flow-runner/tsconfig.json

@@ -0,0 +1,21 @@
+{
+	"compilerOptions": {
+		"module": "ESNext",
+		"moduleResolution": "Bundler",
+		"allowImportingTsExtensions": true,
+		"esModuleInterop": true,
+		"resolveJsonModule": true,
+		"target": "ES2022",
+		"lib": ["ESNext"],
+		"strict": true,
+		"skipLibCheck": true,
+		"noUncheckedIndexedAccess": true,
+		"isolatedModules": true,
+		"verbatimModuleSyntax": true,
+		"forceConsistentCasingInFileNames": true,
+		"noEmit": true,
+		"types": ["node", "bun"],
+		"rootDir": "."
+	},
+	"include": ["src/**/*.ts", "test/**/*.ts"]
+}