feat: Phase 2 — Thread lifecycle, execution engine, worker, CLI
- types.ts: START/END, RoleMeta, ThreadContext, Role, Moderator, WorkflowDefinition - engine.ts: executeThread with JSONL persistence + AbortSignal - worker.ts: per-bundle process, TCP IPC, kill individual threads - CLI: run/ps/kill/threads/thread/thread rm commands - 32 tests pass, biome clean 小橘 <xiaoju@shazhou.work>
This commit is contained in:
@@ -0,0 +1,100 @@
|
||||
import { afterEach, beforeEach, describe, expect, test } from "bun:test";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { cmdAdd } from "../src/cmd-add.js";
|
||||
import { cmdList, formatListLines } from "../src/cmd-list.js";
|
||||
import { cmdRemove } from "../src/cmd-remove.js";
|
||||
import { cmdShow } from "../src/cmd-show.js";
|
||||
|
||||
describe("cli workflow commands", () => {
|
||||
let prevEnv: string | undefined;
|
||||
let storageRoot: string;
|
||||
|
||||
beforeEach(async () => {
|
||||
prevEnv = process.env.UNCAGED_WORKFLOW_STORAGE_ROOT;
|
||||
storageRoot = await mkdtemp(join(tmpdir(), "uncaged-wf-"));
|
||||
process.env.UNCAGED_WORKFLOW_STORAGE_ROOT = storageRoot;
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
if (prevEnv === undefined) {
|
||||
delete process.env.UNCAGED_WORKFLOW_STORAGE_ROOT;
|
||||
} else {
|
||||
process.env.UNCAGED_WORKFLOW_STORAGE_ROOT = prevEnv;
|
||||
}
|
||||
await rm(storageRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("add / list / show / remove roundtrip", async () => {
|
||||
const bundleDir = join(storageRoot, "src");
|
||||
await mkdir(bundleDir, { recursive: true });
|
||||
const bundlePath = join(bundleDir, "demo.esm.js");
|
||||
await writeFile(
|
||||
bundlePath,
|
||||
`import fs from "node:fs";
|
||||
|
||||
export default {
|
||||
name: "solve-issue",
|
||||
roles: {
|
||||
noop: async () => {
|
||||
fs.existsSync(".");
|
||||
return { content: "ok", meta: { done: true } };
|
||||
},
|
||||
},
|
||||
moderator(ctx) {
|
||||
if (ctx.steps.length === 0) {
|
||||
return "noop";
|
||||
}
|
||||
return "__end__";
|
||||
},
|
||||
};
|
||||
`,
|
||||
"utf8",
|
||||
);
|
||||
|
||||
const added = await cmdAdd(storageRoot, "solve-issue", bundlePath);
|
||||
expect(added.ok).toBe(true);
|
||||
|
||||
const listed = await cmdList(storageRoot);
|
||||
expect(listed.ok).toBe(true);
|
||||
if (listed.ok) {
|
||||
const lines = formatListLines(listed.value);
|
||||
expect(lines.some((l) => l.startsWith("solve-issue\t"))).toBe(true);
|
||||
}
|
||||
|
||||
const shown = await cmdShow(storageRoot, "solve-issue");
|
||||
expect(shown.ok).toBe(true);
|
||||
if (!shown.ok) {
|
||||
return;
|
||||
}
|
||||
expect(shown.value.hash.length).toBe(13);
|
||||
|
||||
const bundleOnDisk = await readFile(
|
||||
join(storageRoot, "bundles", `${shown.value.hash}.esm.js`),
|
||||
"utf8",
|
||||
);
|
||||
expect(bundleOnDisk.length).toBeGreaterThan(0);
|
||||
|
||||
const removed = await cmdRemove(storageRoot, "solve-issue");
|
||||
expect(removed.ok).toBe(true);
|
||||
|
||||
const listedAfter = await cmdList(storageRoot);
|
||||
expect(listedAfter.ok).toBe(true);
|
||||
if (listedAfter.ok) {
|
||||
expect(formatListLines(listedAfter.value)[0]).toBe("(no workflows registered)");
|
||||
}
|
||||
});
|
||||
|
||||
test("add rejects invalid bundles", async () => {
|
||||
const bundlePath = join(storageRoot, "bad.esm.js");
|
||||
await writeFile(
|
||||
bundlePath,
|
||||
'import x from "./local";\nexport default async function run() { return { returnCode: 0, summary: "" }; }\n',
|
||||
"utf8",
|
||||
);
|
||||
const r = await cmdAdd(storageRoot, "solve-issue", bundlePath);
|
||||
expect(r.ok).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,214 @@
|
||||
import { afterEach, beforeEach, describe, expect, test } from "bun:test";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { dirname, join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
import { cmdAdd } from "../src/cmd-add.js";
|
||||
import { cmdKill } from "../src/cmd-kill.js";
|
||||
import { cmdPs } from "../src/cmd-ps.js";
|
||||
import { cmdRun } from "../src/cmd-run.js";
|
||||
import { cmdThreadRemove, cmdThreadShow } from "../src/cmd-thread.js";
|
||||
import { cmdThreads } from "../src/cmd-threads.js";
|
||||
import { pathExists } from "../src/fs-utils.js";
|
||||
|
||||
const fastBundleSource = `export default {
|
||||
name: "solve-issue",
|
||||
roles: {
|
||||
planner: async () => ({ content: "plan", meta: { plan: "x" } }),
|
||||
coder: async () => ({ content: "code", meta: { diff: "y" } }),
|
||||
},
|
||||
moderator(ctx) {
|
||||
if (ctx.steps.length === 0) return "planner";
|
||||
if (ctx.steps.length === 1) return "coder";
|
||||
return "__end__";
|
||||
},
|
||||
};
|
||||
`;
|
||||
|
||||
const slowPlannerBundleSource = `export default {
|
||||
name: "solve-issue",
|
||||
roles: {
|
||||
planner: async () => {
|
||||
await new Promise((r) => setTimeout(r, 400));
|
||||
return { content: "plan", meta: { plan: "x" } };
|
||||
},
|
||||
coder: async () => ({ content: "code", meta: { diff: "y" } }),
|
||||
},
|
||||
moderator(ctx) {
|
||||
if (ctx.steps.length === 0) return "planner";
|
||||
if (ctx.steps.length === 1) return "coder";
|
||||
return "__end__";
|
||||
},
|
||||
};
|
||||
`;
|
||||
|
||||
const cliEntryPath = fileURLToPath(new URL("../src/cli.ts", import.meta.url));
|
||||
|
||||
const abortablePlannerBundleSource = `export default {
|
||||
name: "solve-issue",
|
||||
roles: {
|
||||
planner: async () => {
|
||||
await new Promise((r) => setTimeout(r, 600));
|
||||
return { content: "plan", meta: { plan: "x" } };
|
||||
},
|
||||
coder: async () => ({ content: "code", meta: { diff: "y" } }),
|
||||
},
|
||||
moderator(ctx) {
|
||||
if (ctx.steps.length === 0) return "planner";
|
||||
if (ctx.steps.length === 1) return "coder";
|
||||
return "__end__";
|
||||
},
|
||||
};
|
||||
`;
|
||||
|
||||
describe("cli thread commands", () => {
|
||||
let prevEnv: string | undefined;
|
||||
let storageRoot: string;
|
||||
|
||||
beforeEach(async () => {
|
||||
prevEnv = process.env.UNCAGED_WORKFLOW_STORAGE_ROOT;
|
||||
storageRoot = await mkdtemp(join(tmpdir(), "uncaged-wf-thread-"));
|
||||
process.env.UNCAGED_WORKFLOW_STORAGE_ROOT = storageRoot;
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
if (prevEnv === undefined) {
|
||||
delete process.env.UNCAGED_WORKFLOW_STORAGE_ROOT;
|
||||
} else {
|
||||
process.env.UNCAGED_WORKFLOW_STORAGE_ROOT = prevEnv;
|
||||
}
|
||||
await rm(storageRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("run / threads / thread / thread rm", async () => {
|
||||
const bundleDir = join(storageRoot, "src");
|
||||
await mkdir(bundleDir, { recursive: true });
|
||||
const bundlePath = join(bundleDir, "demo.esm.js");
|
||||
await writeFile(bundlePath, fastBundleSource, "utf8");
|
||||
|
||||
const added = await cmdAdd(storageRoot, "solve-issue", bundlePath);
|
||||
expect(added.ok).toBe(true);
|
||||
if (!added.ok) {
|
||||
return;
|
||||
}
|
||||
|
||||
const ran = await cmdRun(storageRoot, "solve-issue", "hello", false, 5);
|
||||
expect(ran.ok).toBe(true);
|
||||
if (!ran.ok) {
|
||||
return;
|
||||
}
|
||||
|
||||
const threadId = ran.value.threadId;
|
||||
|
||||
let threads = await cmdThreads(storageRoot, []);
|
||||
for (
|
||||
let attempt = 0;
|
||||
attempt < 50 && threads.ok && !threads.value.some((l) => l.includes(threadId));
|
||||
attempt++
|
||||
) {
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
threads = await cmdThreads(storageRoot, []);
|
||||
}
|
||||
expect(threads.ok).toBe(true);
|
||||
if (!threads.ok) {
|
||||
return;
|
||||
}
|
||||
expect(threads.value.some((l) => l.includes(threadId))).toBe(true);
|
||||
|
||||
const shown = await cmdThreadShow(storageRoot, threadId);
|
||||
expect(shown.ok).toBe(true);
|
||||
if (!shown.ok) {
|
||||
return;
|
||||
}
|
||||
expect(shown.value.includes('"threadId"')).toBe(true);
|
||||
|
||||
const removed = await cmdThreadRemove(storageRoot, threadId);
|
||||
expect(removed.ok).toBe(true);
|
||||
|
||||
const dataPath = join(storageRoot, "logs", added.value.hash, `${threadId}.data.jsonl`);
|
||||
expect(await pathExists(dataPath)).toBe(false);
|
||||
});
|
||||
|
||||
test("cli entrypoint dispatches threads / ps (spawn)", () => {
|
||||
const env = { ...process.env, UNCAGED_WORKFLOW_STORAGE_ROOT: storageRoot };
|
||||
const threads = spawnSync(process.execPath, [cliEntryPath, "threads"], {
|
||||
env,
|
||||
encoding: "utf8",
|
||||
});
|
||||
expect(threads.status).toBe(0);
|
||||
|
||||
const ps = spawnSync(process.execPath, [cliEntryPath, "ps"], { env, encoding: "utf8" });
|
||||
expect(ps.status).toBe(0);
|
||||
});
|
||||
|
||||
test("ps lists running threads while planner role is in-flight", async () => {
|
||||
const bundleDir = join(storageRoot, "src");
|
||||
await mkdir(bundleDir, { recursive: true });
|
||||
const bundlePath = join(bundleDir, "demo.esm.js");
|
||||
await writeFile(bundlePath, slowPlannerBundleSource, "utf8");
|
||||
|
||||
const added = await cmdAdd(storageRoot, "solve-issue", bundlePath);
|
||||
expect(added.ok).toBe(true);
|
||||
if (!added.ok) {
|
||||
return;
|
||||
}
|
||||
|
||||
const ran = await cmdRun(storageRoot, "solve-issue", "hello", false, 5);
|
||||
expect(ran.ok).toBe(true);
|
||||
if (!ran.ok) {
|
||||
return;
|
||||
}
|
||||
|
||||
const threadId = ran.value.threadId;
|
||||
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
const psEarly = await cmdPs(storageRoot);
|
||||
expect(psEarly.some((l) => l.includes(threadId))).toBe(true);
|
||||
|
||||
await new Promise((r) => setTimeout(r, 900));
|
||||
|
||||
const psLate = await cmdPs(storageRoot);
|
||||
expect(psLate).toEqual(["(no running threads)"]);
|
||||
});
|
||||
|
||||
test("kill stops thread after the in-flight role (before subsequent roles)", async () => {
|
||||
const bundleDir = join(storageRoot, "src");
|
||||
await mkdir(bundleDir, { recursive: true });
|
||||
const bundlePath = join(bundleDir, "demo.esm.js");
|
||||
await writeFile(bundlePath, abortablePlannerBundleSource, "utf8");
|
||||
|
||||
const added = await cmdAdd(storageRoot, "solve-issue", bundlePath);
|
||||
expect(added.ok).toBe(true);
|
||||
if (!added.ok) {
|
||||
return;
|
||||
}
|
||||
|
||||
const ran = await cmdRun(storageRoot, "solve-issue", "hello", false, 5);
|
||||
expect(ran.ok).toBe(true);
|
||||
if (!ran.ok) {
|
||||
return;
|
||||
}
|
||||
|
||||
const threadId = ran.value.threadId;
|
||||
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
|
||||
const killed = await cmdKill(storageRoot, threadId);
|
||||
expect(killed.ok).toBe(true);
|
||||
|
||||
await new Promise((r) => setTimeout(r, 900));
|
||||
|
||||
const dataPath = join(storageRoot, "logs", added.value.hash, `${threadId}.data.jsonl`);
|
||||
const text = await readFile(dataPath, "utf8");
|
||||
const lines = text
|
||||
.trim()
|
||||
.split("\n")
|
||||
.filter((l) => l !== "");
|
||||
expect(lines.length).toBe(2);
|
||||
|
||||
const runningPath = join(dirname(dataPath), `${threadId}.running`);
|
||||
expect(await pathExists(runningPath)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -6,7 +6,8 @@
|
||||
"uncaged-workflow": "src/cli.ts"
|
||||
},
|
||||
"dependencies": {
|
||||
"@uncaged/workflow": "workspace:*"
|
||||
"@uncaged/workflow": "workspace:*",
|
||||
"yaml": "^2.8.4"
|
||||
},
|
||||
"scripts": {
|
||||
"build": "echo 'TODO'",
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { copyFile, mkdir, readFile, stat } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { err, ok, type Result } from "@uncaged/workflow";
|
||||
|
||||
async function pathExists(path: string): Promise<boolean> {
|
||||
try {
|
||||
await stat(path);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export async function storeWorkflowBundleCopy(
|
||||
storageRoot: string,
|
||||
hash: string,
|
||||
resolvedSourcePath: string,
|
||||
sourceText: string,
|
||||
): Promise<Result<void, string>> {
|
||||
const bundlesDir = join(storageRoot, "bundles");
|
||||
const destPath = join(bundlesDir, `${hash}.esm.js`);
|
||||
|
||||
try {
|
||||
await mkdir(bundlesDir, { recursive: true });
|
||||
} catch (e) {
|
||||
const message = e instanceof Error ? e.message : String(e);
|
||||
return err(`failed to store bundle: ${message}`);
|
||||
}
|
||||
|
||||
if (!(await pathExists(destPath))) {
|
||||
try {
|
||||
await copyFile(resolvedSourcePath, destPath);
|
||||
} catch (e) {
|
||||
const message = e instanceof Error ? e.message : String(e);
|
||||
return err(`failed to store bundle: ${message}`);
|
||||
}
|
||||
return ok(undefined);
|
||||
}
|
||||
|
||||
let existing: string;
|
||||
try {
|
||||
existing = await readFile(destPath, "utf8");
|
||||
} catch (e) {
|
||||
const message = e instanceof Error ? e.message : String(e);
|
||||
return err(`failed to store bundle: ${message}`);
|
||||
}
|
||||
if (existing !== sourceText) {
|
||||
return err(`bundle hash ${hash} already exists with different contents; refusing to overwrite`);
|
||||
}
|
||||
return ok(undefined);
|
||||
}
|
||||
@@ -0,0 +1,228 @@
|
||||
import { printCliError, printCliLine } from "./cli-output.js";
|
||||
import { cmdAdd, formatAddSuccess } from "./cmd-add.js";
|
||||
import { cmdKill } from "./cmd-kill.js";
|
||||
import { cmdList, formatListLines } from "./cmd-list.js";
|
||||
import { cmdPs } from "./cmd-ps.js";
|
||||
import { cmdRemove } from "./cmd-remove.js";
|
||||
import { cmdRun } from "./cmd-run.js";
|
||||
import { cmdShow, formatShowYaml } from "./cmd-show.js";
|
||||
import { cmdThreadRemove, cmdThreadShow } from "./cmd-thread.js";
|
||||
import { cmdThreads } from "./cmd-threads.js";
|
||||
import { parseRunArgv } from "./run-argv.js";
|
||||
|
||||
function usage(): string {
|
||||
return [
|
||||
"Usage:",
|
||||
" uncaged-workflow add <name> <file>",
|
||||
" uncaged-workflow list",
|
||||
" uncaged-workflow show <name>",
|
||||
" uncaged-workflow remove <name>",
|
||||
" uncaged-workflow run <name> [--prompt <text>] [--dry-run] [--max-rounds N]",
|
||||
" uncaged-workflow ps",
|
||||
" uncaged-workflow kill <thread-id>",
|
||||
" uncaged-workflow threads [name]",
|
||||
" uncaged-workflow thread <id>",
|
||||
" uncaged-workflow thread rm <id>",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
async function dispatchAdd(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const name = argv[0];
|
||||
const file = argv[1];
|
||||
if (name === undefined || file === undefined || argv.length > 2) {
|
||||
printCliError(`${usage()}\n\nerror: add requires <name> <file>`);
|
||||
return 1;
|
||||
}
|
||||
const result = await cmdAdd(storageRoot, name, file);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
printCliLine(formatAddSuccess(name, file, result.value.hash));
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchList(storageRoot: string, argv: string[]): Promise<number> {
|
||||
if (argv.length > 0) {
|
||||
printCliError(`${usage()}\n\nerror: list takes no arguments`);
|
||||
return 1;
|
||||
}
|
||||
const result = await cmdList(storageRoot);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
for (const line of formatListLines(result.value)) {
|
||||
printCliLine(line);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchShow(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const name = argv[0];
|
||||
if (name === undefined || argv.length > 1) {
|
||||
printCliError(`${usage()}\n\nerror: show requires <name>`);
|
||||
return 1;
|
||||
}
|
||||
const result = await cmdShow(storageRoot, name);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
printCliLine(formatShowYaml(name, result.value));
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchRemove(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const name = argv[0];
|
||||
if (name === undefined || argv.length > 1) {
|
||||
printCliError(`${usage()}\n\nerror: remove requires <name>`);
|
||||
return 1;
|
||||
}
|
||||
const result = await cmdRemove(storageRoot, name);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
printCliLine(`removed workflow "${name}" from registry`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchRun(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const parsed = parseRunArgv(argv);
|
||||
if (!parsed.ok) {
|
||||
printCliError(`${usage()}\n\nerror: ${parsed.error}`);
|
||||
return 1;
|
||||
}
|
||||
|
||||
const result = await cmdRun(
|
||||
storageRoot,
|
||||
parsed.value.name,
|
||||
parsed.value.prompt,
|
||||
parsed.value.dryRun,
|
||||
parsed.value.maxRounds,
|
||||
);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
|
||||
printCliLine(result.value.threadId);
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchPs(storageRoot: string, argv: string[]): Promise<number> {
|
||||
if (argv.length > 0) {
|
||||
printCliError(`${usage()}\n\nerror: ps takes no arguments`);
|
||||
return 1;
|
||||
}
|
||||
for (const line of await cmdPs(storageRoot)) {
|
||||
printCliLine(line);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchKill(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const threadId = argv[0];
|
||||
if (threadId === undefined || argv.length > 1) {
|
||||
printCliError(`${usage()}\n\nerror: kill requires <thread-id>`);
|
||||
return 1;
|
||||
}
|
||||
const result = await cmdKill(storageRoot, threadId);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
printCliLine(`kill sent for thread ${threadId}`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchThreads(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const result = await cmdThreads(storageRoot, argv);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
for (const line of result.value) {
|
||||
printCliLine(line);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchThread(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const id = argv[0];
|
||||
if (id === undefined || argv.length > 1) {
|
||||
printCliError(`${usage()}\n\nerror: thread requires <id>`);
|
||||
return 1;
|
||||
}
|
||||
const result = await cmdThreadShow(storageRoot, id);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
printCliLine(result.value);
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function dispatchThreadRm(storageRoot: string, argv: string[]): Promise<number> {
|
||||
const id = argv[0];
|
||||
if (id === undefined || argv.length > 1) {
|
||||
printCliError(`${usage()}\n\nerror: thread rm requires <id>`);
|
||||
return 1;
|
||||
}
|
||||
const result = await cmdThreadRemove(storageRoot, id);
|
||||
if (!result.ok) {
|
||||
printCliError(result.error);
|
||||
return 1;
|
||||
}
|
||||
printCliLine(`removed thread ${id}`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
export async function runCli(storageRoot: string, argv: string[]): Promise<number> {
|
||||
if (argv.length === 0) {
|
||||
printCliError(usage());
|
||||
return 1;
|
||||
}
|
||||
const command = argv[0];
|
||||
if (command === undefined) {
|
||||
printCliError(usage());
|
||||
return 1;
|
||||
}
|
||||
const rest = argv.slice(1);
|
||||
|
||||
if (command === "add") {
|
||||
return dispatchAdd(storageRoot, rest);
|
||||
}
|
||||
if (command === "list") {
|
||||
return dispatchList(storageRoot, rest);
|
||||
}
|
||||
if (command === "show") {
|
||||
return dispatchShow(storageRoot, rest);
|
||||
}
|
||||
if (command === "remove") {
|
||||
return dispatchRemove(storageRoot, rest);
|
||||
}
|
||||
if (command === "run") {
|
||||
return dispatchRun(storageRoot, rest);
|
||||
}
|
||||
if (command === "ps") {
|
||||
return dispatchPs(storageRoot, rest);
|
||||
}
|
||||
if (command === "kill") {
|
||||
return dispatchKill(storageRoot, rest);
|
||||
}
|
||||
if (command === "threads") {
|
||||
return dispatchThreads(storageRoot, rest);
|
||||
}
|
||||
if (command === "thread") {
|
||||
const sub = rest[0];
|
||||
if (sub === "rm") {
|
||||
return dispatchThreadRm(storageRoot, rest.slice(1));
|
||||
}
|
||||
return dispatchThread(storageRoot, rest);
|
||||
}
|
||||
|
||||
printCliError(`${usage()}\n\nerror: unknown command ${command}`);
|
||||
return 1;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
export function printCliLine(line: string): void {
|
||||
// biome-ignore lint/suspicious/noConsole: CLI user-facing output
|
||||
console.log(line);
|
||||
}
|
||||
|
||||
export function printCliError(line: string): void {
|
||||
// biome-ignore lint/suspicious/noConsole: CLI user-facing errors
|
||||
console.error(line);
|
||||
}
|
||||
@@ -1,3 +1,9 @@
|
||||
#!/usr/bin/env bun
|
||||
// @uncaged/cli-workflow - uncaged-workflow CLI
|
||||
console.log('uncaged-workflow');
|
||||
|
||||
import { runCli } from "./cli-dispatch.js";
|
||||
import { resolveWorkflowStorageRoot } from "./storage-env.js";
|
||||
|
||||
const argv = process.argv.slice(2);
|
||||
const storageRoot = resolveWorkflowStorageRoot();
|
||||
const code = await runCli(storageRoot, argv);
|
||||
process.exit(code);
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
import { readFile, stat } from "node:fs/promises";
|
||||
import { basename, resolve } from "node:path";
|
||||
|
||||
import {
|
||||
err,
|
||||
hashWorkflowBundleBytes,
|
||||
ok,
|
||||
type Result,
|
||||
readWorkflowRegistry,
|
||||
registerWorkflowVersion,
|
||||
validateWorkflowBundle,
|
||||
writeWorkflowRegistry,
|
||||
} from "@uncaged/workflow";
|
||||
|
||||
import { storeWorkflowBundleCopy } from "./bundle-store.js";
|
||||
import { validateCliWorkflowName } from "./workflow-name.js";
|
||||
|
||||
export async function cmdAdd(
|
||||
storageRoot: string,
|
||||
name: string,
|
||||
filePath: string,
|
||||
): Promise<Result<{ hash: string }, string>> {
|
||||
const nameOk = validateCliWorkflowName(name);
|
||||
if (!nameOk.ok) {
|
||||
return nameOk;
|
||||
}
|
||||
|
||||
let resolvedPath: string;
|
||||
try {
|
||||
resolvedPath = resolve(filePath);
|
||||
await stat(resolvedPath);
|
||||
} catch {
|
||||
return err(`bundle file not found: ${filePath}`);
|
||||
}
|
||||
|
||||
let source: string;
|
||||
try {
|
||||
source = await readFile(resolvedPath, "utf8");
|
||||
} catch (e) {
|
||||
const message = e instanceof Error ? e.message : String(e);
|
||||
return err(`failed to read bundle: ${message}`);
|
||||
}
|
||||
|
||||
const validated = validateWorkflowBundle({
|
||||
filePath: resolvedPath,
|
||||
source,
|
||||
});
|
||||
if (!validated.ok) {
|
||||
return validated;
|
||||
}
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
const bytes = encoder.encode(source);
|
||||
const hash = hashWorkflowBundleBytes(bytes);
|
||||
|
||||
const stored = await storeWorkflowBundleCopy(storageRoot, hash, resolvedPath, source);
|
||||
if (!stored.ok) {
|
||||
return stored;
|
||||
}
|
||||
|
||||
const reg = await readWorkflowRegistry(storageRoot);
|
||||
if (!reg.ok) {
|
||||
return err(reg.error.message);
|
||||
}
|
||||
|
||||
const next = registerWorkflowVersion(reg.value, name, hash, Date.now());
|
||||
const written = await writeWorkflowRegistry(storageRoot, next);
|
||||
if (!written.ok) {
|
||||
return err(written.error.message);
|
||||
}
|
||||
|
||||
return ok({ hash });
|
||||
}
|
||||
|
||||
export function formatAddSuccess(name: string, filePath: string, hash: string): string {
|
||||
return `registered workflow "${name}" from ${basename(filePath)} as ${hash}`;
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
import { join } from "node:path";
|
||||
|
||||
import { err, type Result } from "@uncaged/workflow";
|
||||
|
||||
import { readTextFileIfExists } from "./fs-utils.js";
|
||||
import {
|
||||
resolveRunningHashForThread,
|
||||
sendWorkerTcpCommand,
|
||||
type WorkerCtl,
|
||||
} from "./worker-spawn.js";
|
||||
|
||||
export async function cmdKill(
|
||||
storageRoot: string,
|
||||
threadId: string,
|
||||
): Promise<Result<void, string>> {
|
||||
const hashResult = await resolveRunningHashForThread(storageRoot, threadId);
|
||||
if (!hashResult.ok) {
|
||||
return hashResult;
|
||||
}
|
||||
|
||||
const ctlPath = join(storageRoot, "workers", `${hashResult.value}.json`);
|
||||
const ctlText = await readTextFileIfExists(ctlPath);
|
||||
if (ctlText === null) {
|
||||
return err(`worker control file missing for bundle hash ${hashResult.value}`);
|
||||
}
|
||||
|
||||
let ctl: WorkerCtl;
|
||||
try {
|
||||
ctl = JSON.parse(ctlText) as WorkerCtl;
|
||||
} catch {
|
||||
return err(`corrupt worker control file: ${ctlPath}`);
|
||||
}
|
||||
|
||||
if (typeof ctl.port !== "number" || ctl.port <= 0) {
|
||||
return err(`invalid worker control file: ${ctlPath}`);
|
||||
}
|
||||
|
||||
return await sendWorkerTcpCommand(ctl.port, { type: "kill", threadId });
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
import {
|
||||
err,
|
||||
listRegisteredWorkflowNames,
|
||||
ok,
|
||||
type Result,
|
||||
readWorkflowRegistry,
|
||||
type WorkflowRegistryFile,
|
||||
} from "@uncaged/workflow";
|
||||
|
||||
export async function cmdList(storageRoot: string): Promise<Result<WorkflowRegistryFile, string>> {
|
||||
const reg = await readWorkflowRegistry(storageRoot);
|
||||
if (!reg.ok) {
|
||||
return err(reg.error.message);
|
||||
}
|
||||
return ok(reg.value);
|
||||
}
|
||||
|
||||
export function formatListLines(registry: WorkflowRegistryFile): string[] {
|
||||
const names = listRegisteredWorkflowNames(registry);
|
||||
if (names.length === 0) {
|
||||
return ["(no workflows registered)"];
|
||||
}
|
||||
const lines: string[] = [];
|
||||
for (const name of names) {
|
||||
const entry = registry.workflows[name];
|
||||
if (entry === undefined) {
|
||||
continue;
|
||||
}
|
||||
lines.push(`${name}\t${entry.hash}\t${entry.timestamp}`);
|
||||
}
|
||||
return lines;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
import { listRunningThreads } from "./thread-scan.js";
|
||||
|
||||
export async function cmdPs(storageRoot: string): Promise<string[]> {
|
||||
const rows = await listRunningThreads(storageRoot);
|
||||
if (rows.length === 0) {
|
||||
return ["(no running threads)"];
|
||||
}
|
||||
return rows.map((r) => `${r.threadId}\t${r.hash}\t${r.workflowName ?? "(unknown)"}`);
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
import {
|
||||
err,
|
||||
ok,
|
||||
type Result,
|
||||
readWorkflowRegistry,
|
||||
unregisterWorkflow,
|
||||
writeWorkflowRegistry,
|
||||
} from "@uncaged/workflow";
|
||||
|
||||
import { validateCliWorkflowName } from "./workflow-name.js";
|
||||
|
||||
export async function cmdRemove(storageRoot: string, name: string): Promise<Result<void, string>> {
|
||||
const nameOk = validateCliWorkflowName(name);
|
||||
if (!nameOk.ok) {
|
||||
return nameOk;
|
||||
}
|
||||
|
||||
const reg = await readWorkflowRegistry(storageRoot);
|
||||
if (!reg.ok) {
|
||||
return err(reg.error.message);
|
||||
}
|
||||
|
||||
const next = unregisterWorkflow(reg.value, name);
|
||||
if (!next.ok) {
|
||||
return err(next.error.message);
|
||||
}
|
||||
|
||||
const written = await writeWorkflowRegistry(storageRoot, next.value);
|
||||
if (!written.ok) {
|
||||
return err(written.error.message);
|
||||
}
|
||||
|
||||
return ok(undefined);
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
import { join } from "node:path";
|
||||
|
||||
import {
|
||||
err,
|
||||
generateUlid,
|
||||
getRegisteredWorkflow,
|
||||
ok,
|
||||
type Result,
|
||||
readWorkflowRegistry,
|
||||
} from "@uncaged/workflow";
|
||||
import { ensureWorkerForHash, sendWorkerTcpCommand } from "./worker-spawn.js";
|
||||
import { validateCliWorkflowName } from "./workflow-name.js";
|
||||
|
||||
export async function cmdRun(
|
||||
storageRoot: string,
|
||||
name: string,
|
||||
prompt: string,
|
||||
isDryRun: boolean,
|
||||
maxRounds: number,
|
||||
): Promise<Result<{ threadId: string }, string>> {
|
||||
const nameOk = validateCliWorkflowName(name);
|
||||
if (!nameOk.ok) {
|
||||
return nameOk;
|
||||
}
|
||||
|
||||
const reg = await readWorkflowRegistry(storageRoot);
|
||||
if (!reg.ok) {
|
||||
return err(reg.error.message);
|
||||
}
|
||||
|
||||
const entry = getRegisteredWorkflow(reg.value, name);
|
||||
if (entry === null) {
|
||||
return err(`workflow not registered: ${name}`);
|
||||
}
|
||||
|
||||
const bundlePath = join(storageRoot, "bundles", `${entry.hash}.esm.js`);
|
||||
const worker = await ensureWorkerForHash(storageRoot, entry.hash, bundlePath);
|
||||
if (!worker.ok) {
|
||||
return worker;
|
||||
}
|
||||
|
||||
const threadId = generateUlid(Date.now());
|
||||
const sent = await sendWorkerTcpCommand(worker.value.port, {
|
||||
type: "run",
|
||||
threadId,
|
||||
prompt,
|
||||
options: { isDryRun, maxRounds },
|
||||
});
|
||||
if (!sent.ok) {
|
||||
return sent;
|
||||
}
|
||||
|
||||
return ok({ threadId });
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
import {
|
||||
err,
|
||||
getRegisteredWorkflow,
|
||||
ok,
|
||||
type Result,
|
||||
readWorkflowRegistry,
|
||||
type WorkflowRegistryEntry,
|
||||
} from "@uncaged/workflow";
|
||||
import { stringify } from "yaml";
|
||||
|
||||
import { validateCliWorkflowName } from "./workflow-name.js";
|
||||
|
||||
export async function cmdShow(
|
||||
storageRoot: string,
|
||||
name: string,
|
||||
): Promise<Result<WorkflowRegistryEntry, string>> {
|
||||
const nameOk = validateCliWorkflowName(name);
|
||||
if (!nameOk.ok) {
|
||||
return nameOk;
|
||||
}
|
||||
|
||||
const reg = await readWorkflowRegistry(storageRoot);
|
||||
if (!reg.ok) {
|
||||
return err(reg.error.message);
|
||||
}
|
||||
|
||||
const entry = getRegisteredWorkflow(reg.value, name);
|
||||
if (entry === null) {
|
||||
return err(`workflow not found: ${name}`);
|
||||
}
|
||||
return ok(entry);
|
||||
}
|
||||
|
||||
export function formatShowYaml(name: string, entry: WorkflowRegistryEntry): string {
|
||||
const payload = {
|
||||
[name]: {
|
||||
hash: entry.hash,
|
||||
timestamp: entry.timestamp,
|
||||
history: entry.history,
|
||||
},
|
||||
};
|
||||
return stringify(payload, { indent: 2, defaultStringType: "QUOTE_DOUBLE" });
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import { unlink } from "node:fs/promises";
|
||||
import { dirname, join } from "node:path";
|
||||
|
||||
import { err, ok, type Result } from "@uncaged/workflow";
|
||||
|
||||
import { readTextFileIfExists } from "./fs-utils.js";
|
||||
import { resolveThreadDataPath } from "./thread-scan.js";
|
||||
|
||||
export async function cmdThreadShow(
|
||||
storageRoot: string,
|
||||
threadId: string,
|
||||
): Promise<Result<string, string>> {
|
||||
const dataPath = await resolveThreadDataPath(storageRoot, threadId);
|
||||
if (dataPath === null) {
|
||||
return err(`thread not found: ${threadId}`);
|
||||
}
|
||||
const text = await readTextFileIfExists(dataPath);
|
||||
if (text === null) {
|
||||
return err(`thread data missing: ${threadId}`);
|
||||
}
|
||||
return ok(text.endsWith("\n") ? text.slice(0, -1) : text);
|
||||
}
|
||||
|
||||
export async function cmdThreadRemove(
|
||||
storageRoot: string,
|
||||
threadId: string,
|
||||
): Promise<Result<void, string>> {
|
||||
const dataPath = await resolveThreadDataPath(storageRoot, threadId);
|
||||
if (dataPath === null) {
|
||||
return err(`thread not found: ${threadId}`);
|
||||
}
|
||||
|
||||
const dir = dirname(dataPath);
|
||||
const infoPath = join(dir, `${threadId}.info.jsonl`);
|
||||
const runningPath = join(dir, `${threadId}.running`);
|
||||
|
||||
await unlink(dataPath);
|
||||
await unlink(infoPath).catch(() => {});
|
||||
await unlink(runningPath).catch(() => {});
|
||||
|
||||
return ok(undefined);
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
import { err, ok, type Result } from "@uncaged/workflow";
|
||||
|
||||
import { listHistoricalThreads } from "./thread-scan.js";
|
||||
import { validateCliWorkflowName } from "./workflow-name.js";
|
||||
|
||||
export async function cmdThreads(
|
||||
storageRoot: string,
|
||||
argv: string[],
|
||||
): Promise<Result<string[], string>> {
|
||||
const nameFilter = argv[0];
|
||||
if (argv.length > 1) {
|
||||
return err("threads expects at most one workflow name argument");
|
||||
}
|
||||
|
||||
let workflowNameFilter: string | null = null;
|
||||
if (nameFilter !== undefined) {
|
||||
const nameOk = validateCliWorkflowName(nameFilter);
|
||||
if (!nameOk.ok) {
|
||||
return nameOk;
|
||||
}
|
||||
workflowNameFilter = nameFilter;
|
||||
}
|
||||
|
||||
const rows = await listHistoricalThreads(storageRoot, workflowNameFilter);
|
||||
if (rows.length === 0) {
|
||||
return ok(["(no threads found)"]);
|
||||
}
|
||||
|
||||
const lines = rows.map((r) => `${r.threadId}\t${r.hash}\t${r.workflowName ?? "(unknown)"}`);
|
||||
return ok(lines);
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
import { readFile, stat } from "node:fs/promises";
|
||||
|
||||
export async function readTextFileIfExists(path: string): Promise<string | null> {
|
||||
try {
|
||||
return await readFile(path, "utf8");
|
||||
} catch (e) {
|
||||
const errObj = e as NodeJS.ErrnoException;
|
||||
if (errObj.code === "ENOENT") {
|
||||
return null;
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
export async function pathExists(path: string): Promise<boolean> {
|
||||
try {
|
||||
await stat(path);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
import { err, ok, type Result } from "@uncaged/workflow";
|
||||
|
||||
export type ParsedRunArgv = {
|
||||
name: string;
|
||||
prompt: string;
|
||||
dryRun: boolean;
|
||||
maxRounds: number;
|
||||
};
|
||||
|
||||
type FlagOk =
|
||||
| { kind: "dry-run" }
|
||||
| { kind: "prompt"; value: string }
|
||||
| { kind: "max-rounds"; value: number };
|
||||
|
||||
function parseFlagAt(argv: string[], index: number): Result<FlagOk, string> | null {
|
||||
const flag = argv[index];
|
||||
if (flag === "--dry-run") {
|
||||
return ok({ kind: "dry-run" });
|
||||
}
|
||||
if (flag === "--prompt") {
|
||||
const value = argv[index + 1];
|
||||
if (value === undefined) {
|
||||
return err("missing value for --prompt");
|
||||
}
|
||||
return ok({ kind: "prompt", value });
|
||||
}
|
||||
if (flag === "--max-rounds") {
|
||||
const value = argv[index + 1];
|
||||
if (value === undefined) {
|
||||
return err("missing value for --max-rounds");
|
||||
}
|
||||
const n = Number(value);
|
||||
if (!Number.isFinite(n) || !Number.isInteger(n) || n < 0) {
|
||||
return err("--max-rounds must be a non-negative integer");
|
||||
}
|
||||
return ok({ kind: "max-rounds", value: n });
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export function parseRunArgv(argv: string[]): Result<ParsedRunArgv, string> {
|
||||
let name: string | undefined;
|
||||
let prompt = "";
|
||||
let dryRun = false;
|
||||
let maxRounds = 5;
|
||||
|
||||
let i = 0;
|
||||
const first = argv[0];
|
||||
if (first !== undefined && !first.startsWith("--")) {
|
||||
name = first;
|
||||
i = 1;
|
||||
}
|
||||
|
||||
while (i < argv.length) {
|
||||
const parsed = parseFlagAt(argv, i);
|
||||
if (parsed === null) {
|
||||
const unknown = argv[i];
|
||||
return err(`unknown run flag: ${unknown}`);
|
||||
}
|
||||
if (!parsed.ok) {
|
||||
return parsed;
|
||||
}
|
||||
|
||||
const flag = parsed.value;
|
||||
if (flag.kind === "dry-run") {
|
||||
dryRun = true;
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
if (flag.kind === "prompt") {
|
||||
prompt = flag.value;
|
||||
i += 2;
|
||||
continue;
|
||||
}
|
||||
maxRounds = flag.value;
|
||||
i += 2;
|
||||
}
|
||||
|
||||
if (name === undefined || name === "") {
|
||||
return err("run requires <name>");
|
||||
}
|
||||
|
||||
return ok({ name, prompt, dryRun, maxRounds });
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
import { getDefaultWorkflowStorageRoot } from "@uncaged/workflow";
|
||||
|
||||
/** Resolve storage root, honoring `UNCAGED_WORKFLOW_STORAGE_ROOT` for tests/tools. */
|
||||
export function resolveWorkflowStorageRoot(): string {
|
||||
const override = process.env.UNCAGED_WORKFLOW_STORAGE_ROOT;
|
||||
if (override !== undefined && override !== "") {
|
||||
return override;
|
||||
}
|
||||
return getDefaultWorkflowStorageRoot();
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
import { readdir } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { pathExists, readTextFileIfExists } from "./fs-utils.js";
|
||||
|
||||
export type RunningThreadRow = {
|
||||
threadId: string;
|
||||
hash: string;
|
||||
workflowName: string | null;
|
||||
};
|
||||
|
||||
export type HistoricalThreadRow = {
|
||||
threadId: string;
|
||||
hash: string;
|
||||
workflowName: string | null;
|
||||
};
|
||||
|
||||
async function readWorkflowNameFromDataJsonl(dataPath: string): Promise<string | null> {
|
||||
const text = await readTextFileIfExists(dataPath);
|
||||
if (text === null) {
|
||||
return null;
|
||||
}
|
||||
const firstLine = text.split("\n")[0];
|
||||
if (firstLine === undefined || firstLine.trim() === "") {
|
||||
return null;
|
||||
}
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(firstLine) as unknown;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (parsed === null || typeof parsed !== "object") {
|
||||
return null;
|
||||
}
|
||||
const name = (parsed as Record<string, unknown>).name;
|
||||
return typeof name === "string" ? name : null;
|
||||
}
|
||||
|
||||
/** Threads currently executing — identified via `<threadId>.running` markers. */
|
||||
export async function listRunningThreads(storageRoot: string): Promise<RunningThreadRow[]> {
|
||||
const logsRoot = join(storageRoot, "logs");
|
||||
if (!(await pathExists(logsRoot))) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const hashes = await readdir(logsRoot);
|
||||
const out: RunningThreadRow[] = [];
|
||||
|
||||
for (const hash of hashes) {
|
||||
const dir = join(logsRoot, hash);
|
||||
let entries: string[];
|
||||
try {
|
||||
entries = await readdir(dir);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const fileName of entries) {
|
||||
if (!fileName.endsWith(".running")) {
|
||||
continue;
|
||||
}
|
||||
const threadId = fileName.slice(0, -".running".length);
|
||||
const dataPath = join(dir, `${threadId}.data.jsonl`);
|
||||
const workflowName = await readWorkflowNameFromDataJsonl(dataPath);
|
||||
out.push({ threadId, hash, workflowName });
|
||||
}
|
||||
}
|
||||
|
||||
out.sort((a, b) => {
|
||||
const ha = `${a.hash}/${a.threadId}`;
|
||||
const hb = `${b.hash}/${b.threadId}`;
|
||||
return ha.localeCompare(hb);
|
||||
});
|
||||
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Historical threads discovered via `*.data.jsonl`.
|
||||
* When `workflowNameFilter` is non-null, only threads whose start record `name` matches are returned.
|
||||
*/
|
||||
export async function listHistoricalThreads(
|
||||
storageRoot: string,
|
||||
workflowNameFilter: string | null,
|
||||
): Promise<HistoricalThreadRow[]> {
|
||||
const logsRoot = join(storageRoot, "logs");
|
||||
if (!(await pathExists(logsRoot))) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const hashes = await readdir(logsRoot);
|
||||
const out: HistoricalThreadRow[] = [];
|
||||
|
||||
for (const hash of hashes) {
|
||||
const dir = join(logsRoot, hash);
|
||||
let entries: string[];
|
||||
try {
|
||||
entries = await readdir(dir);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const fileName of entries) {
|
||||
if (!fileName.endsWith(".data.jsonl")) {
|
||||
continue;
|
||||
}
|
||||
const threadId = fileName.slice(0, -".data.jsonl".length);
|
||||
const dataPath = join(dir, fileName);
|
||||
const workflowName = await readWorkflowNameFromDataJsonl(dataPath);
|
||||
if (workflowNameFilter !== null && workflowName !== workflowNameFilter) {
|
||||
continue;
|
||||
}
|
||||
out.push({ threadId, hash, workflowName });
|
||||
}
|
||||
}
|
||||
|
||||
out.sort((a, b) => {
|
||||
const ha = `${a.hash}/${a.threadId}`;
|
||||
const hb = `${b.hash}/${b.threadId}`;
|
||||
return ha.localeCompare(hb);
|
||||
});
|
||||
|
||||
return out;
|
||||
}
|
||||
|
||||
export async function resolveThreadDataPath(
|
||||
storageRoot: string,
|
||||
threadId: string,
|
||||
): Promise<string | null> {
|
||||
const logsRoot = join(storageRoot, "logs");
|
||||
if (!(await pathExists(logsRoot))) {
|
||||
return null;
|
||||
}
|
||||
const hashes = await readdir(logsRoot);
|
||||
for (const hash of hashes) {
|
||||
const candidate = join(logsRoot, hash, `${threadId}.data.jsonl`);
|
||||
if (await pathExists(candidate)) {
|
||||
return candidate;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process";
|
||||
import { mkdir, readdir, unlink, writeFile } from "node:fs/promises";
|
||||
import { createConnection } from "node:net";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { err, getWorkerHostScriptPath, ok, type Result } from "@uncaged/workflow";
|
||||
|
||||
import { pathExists, readTextFileIfExists } from "./fs-utils.js";
|
||||
|
||||
export type WorkerCtl = {
|
||||
pid: number;
|
||||
port: number;
|
||||
};
|
||||
|
||||
function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForReadyLine(
|
||||
childStdout: NodeJS.ReadableStream,
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
): Promise<Result<number, string>> {
|
||||
return await new Promise((resolve) => {
|
||||
let buf = "";
|
||||
let settled = false;
|
||||
|
||||
function finish(result: Result<number, string>): void {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
cleanup();
|
||||
resolve(result);
|
||||
}
|
||||
|
||||
function onData(chunk: Buffer | string): void {
|
||||
buf += typeof chunk === "string" ? chunk : chunk.toString("utf8");
|
||||
const nl = buf.indexOf("\n");
|
||||
if (nl < 0) {
|
||||
return;
|
||||
}
|
||||
const line = buf.slice(0, nl).trim();
|
||||
const prefix = "READY ";
|
||||
if (!line.startsWith(prefix)) {
|
||||
finish(err(`worker did not emit READY line (got: ${line})`));
|
||||
return;
|
||||
}
|
||||
const portText = line.slice(prefix.length);
|
||||
const port = Number(portText);
|
||||
if (!Number.isFinite(port) || port <= 0) {
|
||||
finish(err(`worker READY line had invalid port: ${portText}`));
|
||||
return;
|
||||
}
|
||||
finish(ok(port));
|
||||
}
|
||||
|
||||
function onEnd(): void {
|
||||
finish(err("worker stdout ended before READY line"));
|
||||
}
|
||||
|
||||
function onExit(code: number | null): void {
|
||||
finish(err(`worker exited before READY line (code ${code})`));
|
||||
}
|
||||
|
||||
function cleanup(): void {
|
||||
childStdout.off("data", onData);
|
||||
childStdout.off("end", onEnd);
|
||||
child.off("exit", onExit);
|
||||
}
|
||||
|
||||
childStdout.on("data", onData);
|
||||
childStdout.on("end", onEnd);
|
||||
child.on("exit", onExit);
|
||||
});
|
||||
}
|
||||
|
||||
async function spawnWorkerProcess(
|
||||
bundlePath: string,
|
||||
storageRoot: string,
|
||||
hash: string,
|
||||
): Promise<Result<{ pid: number; port: number }, string>> {
|
||||
const scriptPath = getWorkerHostScriptPath();
|
||||
const child = spawn(process.execPath, [scriptPath, bundlePath, storageRoot, hash], {
|
||||
stdio: ["ignore", "pipe", "inherit"],
|
||||
});
|
||||
|
||||
if (child.stdout === null || child.pid === undefined) {
|
||||
return err("failed to spawn worker process");
|
||||
}
|
||||
|
||||
const pid = child.pid;
|
||||
const ready = await waitForReadyLine(child.stdout, child);
|
||||
if (!ready.ok) {
|
||||
child.kill();
|
||||
return ready;
|
||||
}
|
||||
|
||||
child.unref();
|
||||
child.stdout.destroy();
|
||||
|
||||
return ok({ pid, port: ready.value });
|
||||
}
|
||||
|
||||
export async function ensureWorkerForHash(
|
||||
storageRoot: string,
|
||||
hash: string,
|
||||
bundlePath: string,
|
||||
): Promise<Result<{ port: number }, string>> {
|
||||
const ctlPath = join(storageRoot, "workers", `${hash}.json`);
|
||||
const existingText = await readTextFileIfExists(ctlPath);
|
||||
if (existingText !== null) {
|
||||
try {
|
||||
const ctl = JSON.parse(existingText) as WorkerCtl;
|
||||
if (
|
||||
typeof ctl.pid === "number" &&
|
||||
typeof ctl.port === "number" &&
|
||||
ctl.pid > 0 &&
|
||||
ctl.port > 0 &&
|
||||
isProcessAlive(ctl.pid)
|
||||
) {
|
||||
return ok({ port: ctl.port });
|
||||
}
|
||||
} catch {
|
||||
// Corrupt control file — ignore and respawn.
|
||||
}
|
||||
await unlink(ctlPath).catch(() => {});
|
||||
}
|
||||
|
||||
const spawned = await spawnWorkerProcess(bundlePath, storageRoot, hash);
|
||||
if (!spawned.ok) {
|
||||
return spawned;
|
||||
}
|
||||
|
||||
await mkdir(join(storageRoot, "workers"), { recursive: true });
|
||||
const ctl: WorkerCtl = { pid: spawned.value.pid, port: spawned.value.port };
|
||||
await writeFile(ctlPath, `${JSON.stringify(ctl)}\n`, "utf8");
|
||||
|
||||
return ok({ port: spawned.value.port });
|
||||
}
|
||||
|
||||
export async function sendWorkerTcpCommand(
|
||||
port: number,
|
||||
payload: unknown,
|
||||
): Promise<Result<void, string>> {
|
||||
return await new Promise((resolve) => {
|
||||
let settled = false;
|
||||
const socket = createConnection({ host: "127.0.0.1", port }, () => {
|
||||
socket.write(`${JSON.stringify(payload)}\n`);
|
||||
socket.end();
|
||||
});
|
||||
socket.on("error", (e) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
const message = e instanceof Error ? e.message : String(e);
|
||||
resolve(err(`failed to send worker command: ${message}`));
|
||||
});
|
||||
socket.on("close", () => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
resolve(ok(undefined));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export async function resolveRunningHashForThread(
|
||||
storageRoot: string,
|
||||
threadId: string,
|
||||
): Promise<Result<string, string>> {
|
||||
const logsRoot = join(storageRoot, "logs");
|
||||
if (!(await pathExists(logsRoot))) {
|
||||
return err(`thread not running (no logs dir): ${threadId}`);
|
||||
}
|
||||
const hashes = await readdir(logsRoot);
|
||||
for (const hash of hashes) {
|
||||
const runningPath = join(logsRoot, hash, `${threadId}.running`);
|
||||
if (await pathExists(runningPath)) {
|
||||
return ok(hash);
|
||||
}
|
||||
}
|
||||
return err(`thread not running: ${threadId}`);
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
import { err, ok, type Result } from "@uncaged/workflow";
|
||||
|
||||
const WORKFLOW_NAME_RE = /^[a-z][a-z0-9]*(-[a-z0-9]+)*$/;
|
||||
|
||||
export function validateCliWorkflowName(name: string): Result<void, string> {
|
||||
if (!WORKFLOW_NAME_RE.test(name)) {
|
||||
return err(
|
||||
'invalid workflow name: use verb-first kebab-case (lowercase letters, digits, hyphens), e.g. "solve-issue"',
|
||||
);
|
||||
}
|
||||
return ok(undefined);
|
||||
}
|
||||
Reference in New Issue
Block a user