refactor(cli): reorganize CLI commands into four-layer model (#463)

Implement comprehensive CLI refactoring to clarify the four-layer model:
workflow → thread → step → turn

## Breaking Changes

### Renamed Commands
- `uwf workflow put` → `uwf workflow add`
- `uwf thread step` → `uwf thread exec`

### Removed Commands
- `uwf thread running` (merged into `thread list --status running`)
- `uwf thread kill` (split into `thread stop` and `thread cancel`)

### Moved Commands
- `uwf thread steps` → `uwf step list`
- `uwf thread step-details` → `uwf step show`
- `uwf thread fork` → `uwf step fork`

## New Commands

### Thread Commands
- `uwf thread list --status <idle|running|completed>` - Filter threads by status
- `uwf thread stop <thread-id>` - Stop background execution (keep thread active)
- `uwf thread cancel <thread-id>` - Cancel thread (stop + archive to history)

### Step Command Group (New)
- `uwf step list <thread-id>` - List all steps in a thread
- `uwf step show <step-hash>` - Show step details
- `uwf step read <step-hash> [--before N]` - Read step output as markdown
- `uwf step fork <step-hash>` - Fork thread from a step

## Implementation Details

### Files Modified
- `packages/cli-workflow/src/commands/workflow.ts` - Renamed cmdWorkflowPut → cmdWorkflowAdd
- `packages/cli-workflow/src/commands/thread.ts`:
  - Renamed cmdThreadStep → cmdThreadExec
  - Added cmdThreadStop and cmdThreadCancel (split from cmdThreadKill)
  - Updated cmdThreadList to support --status filter with idle/running/completed
  - Removed cmdThreadSteps, cmdThreadStepDetails, cmdThreadFork
- `packages/cli-workflow/src/commands/step.ts` - New module with:
  - cmdStepList (moved from cmdThreadSteps)
  - cmdStepShow (moved from cmdThreadStepDetails)
  - cmdStepFork (moved from cmdThreadFork)
  - cmdStepRead (new, stub implementation pending #462)
- `packages/cli-workflow/src/cli.ts` - Updated all CLI command registrations

### Tests Updated
- `packages/cli-workflow/src/__tests__/thread-step-count.test.ts` - Updated references from "thread step" to "thread exec"
- `packages/cli-workflow/src/__tests__/thread.test.ts` - Updated imports to use cmdStepShow from step.ts

## Test Results
All 124 tests pass in cli-workflow package.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-24 10:40:32 +00:00
parent fc6072c28c
commit 7b50969307
6 changed files with 549 additions and 161 deletions
@@ -22,48 +22,48 @@ function runCli(args: string[]): { stdout: string; stderr: string; exitCode: num
}
}
describe("thread step --count CLI parsing", () => {
describe("thread exec --count CLI parsing", () => {
test("--help shows -c/--count option", () => {
const result = runCli(["thread", "step", "--help"]);
const result = runCli(["thread", "exec", "--help"]);
expect(result.stdout).toContain("--count");
expect(result.stdout).toContain("-c");
});
test("description says 'one or more steps'", () => {
const result = runCli(["thread", "step", "--help"]);
const result = runCli(["thread", "exec", "--help"]);
expect(result.stdout).toContain("one or more steps");
});
});
describe("cmdThreadStep count logic", () => {
describe("cmdThreadExec count logic", () => {
test("count=0 fails with validation error", () => {
const result = runCli(["thread", "step", "FAKE_THREAD_ID", "-c", "0"]);
const result = runCli(["thread", "exec", "FAKE_THREAD_ID", "-c", "0"]);
expect(result.exitCode).not.toBe(0);
expect(result.stderr).toContain("positive integer");
});
test("negative count fails with validation error", () => {
const result = runCli(["thread", "step", "FAKE_THREAD_ID", "-c", "-1"]);
const result = runCli(["thread", "exec", "FAKE_THREAD_ID", "-c", "-1"]);
expect(result.exitCode).not.toBe(0);
expect(result.stderr).toContain("positive integer");
});
test("non-integer count fails with validation error", () => {
const result = runCli(["thread", "step", "FAKE_THREAD_ID", "-c", "1.5"]);
const result = runCli(["thread", "exec", "FAKE_THREAD_ID", "-c", "1.5"]);
expect(result.exitCode).not.toBe(0);
expect(result.stderr).toContain("positive integer");
});
test("count=1 is the default (no -c flag)", () => {
// Without -c, it should attempt to run 1 step (failing on missing thread, not on count validation)
const result = runCli(["thread", "step", "FAKE_THREAD_ID"]);
const result = runCli(["thread", "exec", "FAKE_THREAD_ID"]);
expect(result.exitCode).not.toBe(0);
// Should NOT contain "positive integer" error — should fail on thread lookup instead
expect(result.stderr).not.toContain("positive integer");
});
test("count=3 passes validation (fails on thread lookup)", () => {
const result = runCli(["thread", "step", "FAKE_THREAD_ID", "-c", "3"]);
const result = runCli(["thread", "exec", "FAKE_THREAD_ID", "-c", "3"]);
expect(result.exitCode).not.toBe(0);
// Should NOT contain "positive integer" error — should fail on thread/storage lookup
expect(result.stderr).not.toContain("positive integer");
@@ -7,10 +7,10 @@ import type { CasRef, ThreadId } from "@uncaged/workflow-protocol";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import {
cmdThreadRead,
cmdThreadStepDetails,
extractLastAssistantContent,
THREAD_READ_DEFAULT_QUOTA,
} from "../commands/thread.js";
import { cmdStepShow } from "../commands/step.js";
import { registerUwfSchemas } from "../schemas.js";
import type { UwfStore } from "../store.js";
import { saveThreadsIndex } from "../store.js";
@@ -315,9 +315,9 @@ describe("cmdThreadRead <output> section", () => {
});
});
// ── cmdThreadStepDetails ──────────────────────────────────────────────────────
// ── cmdStepShow ───────────────────────────────────────────────────────────────
describe("cmdThreadStepDetails", () => {
describe("cmdStepShow", () => {
test("returns expanded detail node with turns inlined", async () => {
const uwf = await makeUwfStore(tmpDir);
const detailSchemas = await registerDetailSchemas(uwf.store);
@@ -365,7 +365,7 @@ describe("cmdThreadStepDetails", () => {
agent: "uwf-hermes",
});
const result = await cmdThreadStepDetails(tmpDir, stepHash);
const result = await cmdStepShow(tmpDir, stepHash);
expect(result).toMatchObject({
sessionId: "sess42",
@@ -586,9 +586,9 @@ describe("cmdThreadRead start section / before / quota", () => {
// ── Tests that call process.exit must be last ─────────────────────────────────
describe("cmdThreadStepDetails (process.exit tests - must be last)", () => {
describe("cmdStepShow (process.exit tests - must be last)", () => {
test("throws when step hash does not exist", async () => {
await expect(cmdThreadStepDetails(tmpDir, "nonexistenth0" as CasRef)).rejects.toThrow();
await expect(cmdStepShow(tmpDir, "nonexistenth0" as CasRef)).rejects.toThrow();
});
test("before with unknown hash rejects", async () => {
+76 -47
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env bun
import type { ThreadId } from "@uncaged/workflow-protocol";
import type { CasRef, ThreadId } from "@uncaged/workflow-protocol";
import { Command } from "commander";
import { stringify as yamlStringify } from "yaml";
import {
@@ -17,20 +17,19 @@ import {
import { cmdLogClean, cmdLogList, cmdLogShow } from "./commands/log.js";
import { cmdSetup, cmdSetupInteractive } from "./commands/setup.js";
import { cmdSkillCli } from "./commands/skill.js";
import { cmdStepFork, cmdStepList, cmdStepRead, cmdStepShow } from "./commands/step.js";
import {
cmdThreadFork,
cmdThreadKill,
cmdThreadCancel,
cmdThreadExec,
cmdThreadList,
cmdThreadRead,
cmdThreadRunning,
cmdThreadShow,
cmdThreadStart,
cmdThreadStep,
cmdThreadStepDetails,
cmdThreadSteps,
cmdThreadStop,
THREAD_READ_DEFAULT_QUOTA,
type ThreadStatus,
} from "./commands/thread.js";
import { cmdWorkflowList, cmdWorkflowPut, cmdWorkflowShow } from "./commands/workflow.js";
import { cmdWorkflowAdd, cmdWorkflowList, cmdWorkflowShow } from "./commands/workflow.js";
import { formatOutput, type OutputFormat } from "./format.js";
import { resolveStorageRoot } from "./store.js";
@@ -60,13 +59,13 @@ program.option("--format <fmt>", "Output format: json or yaml", "json");
const workflow = program.command("workflow").description("Workflow registry and CAS");
workflow
.command("put")
.command("add")
.description("Register a workflow from YAML")
.argument("<file>", "Workflow YAML file")
.action((file: string) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const result = await cmdWorkflowPut(storageRoot, file);
const result = await cmdWorkflowAdd(storageRoot, file);
writeOutput(result);
});
});
@@ -110,7 +109,7 @@ thread
});
thread
.command("step")
.command("exec")
.description("Execute one or more steps")
.argument("<thread-id>", "Thread ULID")
.option("--agent <cmd>", "Override agent command")
@@ -134,7 +133,7 @@ thread
const background = opts.background ?? false;
const backgroundWorker = opts._backgroundWorker ?? false;
const results = await cmdThreadStep(
const results = await cmdThreadExec(
storageRoot,
threadId,
agentOverride,
@@ -165,47 +164,49 @@ thread
thread
.command("list")
.description("List active threads")
.option("--all", "Include archived threads")
.action((opts: { all: boolean }) => {
.description("List threads")
.option("--status <status>", "Filter by status: idle, running, or completed")
.action((opts: { status: string | undefined }) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const result = await cmdThreadList(storageRoot, opts.all);
const validStatuses: ThreadStatus[] = ["idle", "running", "completed"];
let statusFilter: ThreadStatus | null = null;
if (opts.status !== undefined) {
if (!validStatuses.includes(opts.status as ThreadStatus)) {
process.stderr.write(
`Invalid status: ${opts.status}. Must be one of: idle, running, completed\n`,
);
process.exit(1);
}
statusFilter = opts.status as ThreadStatus;
}
const result = await cmdThreadList(storageRoot, statusFilter);
writeOutput(result);
});
});
thread
.command("running")
.description("List threads currently executing in the background")
.action(() => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const result = await cmdThreadRunning(storageRoot);
writeOutput(result);
});
});
thread
.command("kill")
.description("Terminate and archive a thread")
.command("stop")
.description("Stop background execution of a thread (keep thread active)")
.argument("<thread-id>", "Thread ULID")
.action((threadId: string) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const result = await cmdThreadKill(storageRoot, threadId);
const result = await cmdThreadStop(storageRoot, threadId);
writeOutput(result);
});
});
thread
.command("steps")
.description("List all steps in a thread")
.command("cancel")
.description("Cancel a thread (stop execution and move to history)")
.argument("<thread-id>", "Thread ULID")
.action((threadId: string) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const result = await cmdThreadSteps(storageRoot, threadId);
const result = await cmdThreadCancel(storageRoot, threadId);
writeOutput(result);
});
});
@@ -239,30 +240,58 @@ thread
},
);
thread
const step = program.command("step").description("Step operations");
step
.command("list")
.description("List all steps in a thread")
.argument("<thread-id>", "Thread ULID")
.action((threadId: string) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const result = await cmdStepList(storageRoot, threadId);
writeOutput(result);
});
});
step
.command("show")
.description("Show details of a specific step")
.argument("<step-hash>", "CAS hash of the StepNode")
.action((stepHash: string) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const detail = await cmdStepShow(storageRoot, stepHash as CasRef);
writeOutput(detail);
});
});
step
.command("read")
.description("Read a step's agent output as markdown")
.argument("<step-hash>", "CAS hash of the StepNode")
.option("--before <n>", "Show only first N turns")
.action((stepHash: string, opts: { before: string | undefined }) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const before = opts.before !== undefined ? Number.parseInt(opts.before, 10) : null;
const markdown = await cmdStepRead(storageRoot, stepHash as CasRef, before);
process.stdout.write(markdown.endsWith("\n") ? markdown : `${markdown}\n`);
});
});
step
.command("fork")
.description("Fork a thread from a specific step")
.argument("<step-hash>", "CAS hash of the StartNode or StepNode to fork from")
.action((stepHash: string) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const result = await cmdThreadFork(storageRoot, stepHash);
const result = await cmdStepFork(storageRoot, stepHash as CasRef);
writeOutput(result);
});
});
thread
.command("step-details")
.description("Dump the full detail node of a step as YAML")
.argument("<step-hash>", "CAS hash of the StepNode")
.action((stepHash: string) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const detail = await cmdThreadStepDetails(storageRoot, stepHash);
process.stdout.write(yamlStringify(detail));
});
});
const skill = program.command("skill").description("Built-in skill references for agents");
skill
+346
View File
@@ -0,0 +1,346 @@
import type { Store as CasStore, JSONSchema } from "@uncaged/json-cas";
import { getSchema } from "@uncaged/json-cas";
import type {
CasRef,
StartEntry,
StartNodePayload,
StepEntry,
StepNodePayload,
ThreadForkOutput,
ThreadId,
ThreadStepsOutput,
} from "@uncaged/workflow-protocol";
import { generateUlid } from "@uncaged/workflow-util";
import { createUwfStore, loadThreadsIndex, saveThreadsIndex, type UwfStore } from "../store.js";
function fail(message: string): never {
process.stderr.write(`${message}\n`);
process.exit(1);
}
type ChainState = {
startHash: CasRef;
start: StartNodePayload;
stepsNewestFirst: StepNodePayload[];
headIsStart: boolean;
};
type OrderedStepItem = {
hash: CasRef;
payload: StepNodePayload;
timestamp: number;
};
function walkChain(uwf: UwfStore, headHash: CasRef): ChainState {
const headNode = uwf.store.get(headHash);
if (headNode === null) {
fail(`CAS node not found: ${headHash}`);
}
if (headNode.type === uwf.schemas.startNode) {
return {
startHash: headHash,
start: headNode.payload as StartNodePayload,
stepsNewestFirst: [],
headIsStart: true,
};
}
if (headNode.type !== uwf.schemas.stepNode) {
fail(`head ${headHash} is not a StartNode or StepNode`);
}
const stepsNewestFirst: StepNodePayload[] = [];
let hash: CasRef | null = headHash;
while (hash !== null) {
const node = uwf.store.get(hash);
if (node === null) {
fail(`CAS node not found while walking chain: ${hash}`);
}
if (node.type !== uwf.schemas.stepNode) {
break;
}
const payload = node.payload as StepNodePayload;
stepsNewestFirst.push(payload);
hash = payload.prev;
}
const newest = stepsNewestFirst[0];
if (newest === undefined) {
fail(`empty step chain at head ${headHash}`);
}
const startNode = uwf.store.get(newest.start);
if (startNode === null || startNode.type !== uwf.schemas.startNode) {
fail(`StartNode not found: ${newest.start}`);
}
return {
startHash: newest.start,
start: startNode.payload as StartNodePayload,
stepsNewestFirst,
headIsStart: false,
};
}
function expandOutput(uwf: UwfStore, outputRef: CasRef): unknown {
const node = uwf.store.get(outputRef);
if (node === null) {
return {};
}
return node.payload;
}
/**
* Recursively expand all cas_ref fields in a CAS node's payload,
* replacing hash strings with the referenced node's expanded payload.
*/
function expandDeep(store: CasStore, hash: CasRef, visited?: Set<string>): unknown {
const seen = visited ?? new Set<string>();
if (seen.has(hash)) return hash; // cycle guard
seen.add(hash);
const node = store.get(hash);
if (node === null) return hash;
const schema = getSchema(store, node.type);
if (schema === null) return node.payload;
return expandValue(store, schema, node.payload, seen);
}
function expandCasRefField(store: CasStore, value: unknown, visited: Set<string>): unknown {
if (typeof value === "string") {
return expandDeep(store, value as CasRef, visited);
}
return value;
}
function expandAnyOfField(
store: CasStore,
schema: JSONSchema,
value: unknown,
visited: Set<string>,
): unknown {
if (!Array.isArray(schema.anyOf)) return value;
for (const sub of schema.anyOf as JSONSchema[]) {
if (sub.format === "cas_ref" && typeof value === "string") {
return expandDeep(store, value as CasRef, visited);
}
}
return value;
}
function expandArrayField(
store: CasStore,
schema: JSONSchema,
value: unknown,
visited: Set<string>,
): unknown {
if (!Array.isArray(value)) return value;
const itemSchema = schema.items as JSONSchema | undefined;
if (itemSchema === undefined) return value;
return value.map((item) => expandValue(store, itemSchema, item, visited));
}
function expandObjectField(
store: CasStore,
schema: JSONSchema,
value: unknown,
visited: Set<string>,
): unknown {
if (typeof value !== "object" || value === null || Array.isArray(value)) return value;
const props = schema.properties as Record<string, JSONSchema> | undefined;
if (props === undefined) return value;
const result: Record<string, unknown> = {};
for (const [key, val] of Object.entries(value)) {
const propSchema = props[key];
result[key] = propSchema !== undefined ? expandValue(store, propSchema, val, visited) : val;
}
return result;
}
function expandValue(
store: CasStore,
schema: JSONSchema,
value: unknown,
visited: Set<string>,
): unknown {
if (schema.format === "cas_ref") {
return expandCasRefField(store, value, visited);
}
if (schema.anyOf !== undefined) {
return expandAnyOfField(store, schema, value, visited);
}
if (schema.type === "array") {
return expandArrayField(store, schema, value, visited);
}
if (schema.type === "object") {
return expandObjectField(store, schema, value, visited);
}
return value;
}
function collectOrderedSteps(
uwf: UwfStore,
headHash: CasRef,
chain: ChainState,
): OrderedStepItem[] {
const reversed = chain.stepsNewestFirst.slice().reverse();
const ordered: OrderedStepItem[] = [];
let hash: CasRef | null = chain.headIsStart ? null : headHash;
for (const payload of reversed) {
if (hash === null) {
fail("unexpected null hash while collecting ordered steps");
}
const node = uwf.store.get(hash);
if (node === null) {
fail(`CAS node not found: ${hash}`);
}
ordered.push({
hash,
payload,
timestamp: node.timestamp,
});
hash = payload.prev;
}
return ordered;
}
async function resolveHeadHash(storageRoot: string, threadId: ThreadId): Promise<CasRef> {
const index = await loadThreadsIndex(storageRoot);
const head = index[threadId];
if (head === undefined) {
fail(`thread not active: ${threadId}`);
}
return head;
}
/**
* List all steps in a thread (previously: thread steps)
*/
export async function cmdStepList(
storageRoot: string,
threadId: ThreadId,
): Promise<ThreadStepsOutput> {
const headHash = await resolveHeadHash(storageRoot, threadId);
const uwf = await createUwfStore(storageRoot);
const chain = walkChain(uwf, headHash);
const startNode = uwf.store.get(chain.startHash);
if (startNode === null) {
fail(`StartNode not found: ${chain.startHash}`);
}
const startEntry: StartEntry = {
hash: chain.startHash,
workflow: chain.start.workflow,
prompt: chain.start.prompt,
timestamp: startNode.timestamp,
};
const stepEntries: StepEntry[] = [];
const ordered = collectOrderedSteps(uwf, headHash, chain);
for (const item of ordered) {
stepEntries.push({
hash: item.hash,
role: item.payload.role,
output: expandOutput(uwf, item.payload.output),
detail: item.payload.detail,
agent: item.payload.agent,
timestamp: item.timestamp,
});
}
return {
thread: threadId,
workflow: chain.start.workflow,
steps: [startEntry, ...stepEntries],
};
}
/**
* Show details of a specific step (previously: thread step-details)
*/
export async function cmdStepShow(storageRoot: string, stepHash: CasRef): Promise<unknown> {
const uwf = await createUwfStore(storageRoot);
const node = uwf.store.get(stepHash);
if (node === null) {
fail(`CAS node not found: ${stepHash}`);
}
if (node.type !== uwf.schemas.stepNode) {
fail(`node ${stepHash} is not a StepNode`);
}
const payload = node.payload as StepNodePayload;
if (!payload.detail) {
fail(`step ${stepHash} has no detail`);
}
return expandDeep(uwf.store, payload.detail);
}
/**
* Fork a thread from a specific step (previously: thread fork)
*/
export async function cmdStepFork(
storageRoot: string,
stepHash: CasRef,
): Promise<ThreadForkOutput> {
const uwf = await createUwfStore(storageRoot);
const node = uwf.store.get(stepHash);
if (node === null) {
fail(`CAS node not found: ${stepHash}`);
}
if (node.type !== uwf.schemas.startNode && node.type !== uwf.schemas.stepNode) {
fail(`node ${stepHash} is not a StartNode or StepNode`);
}
const newThreadId = generateUlid(Date.now()) as ThreadId;
const index = await loadThreadsIndex(storageRoot);
index[newThreadId] = stepHash;
await saveThreadsIndex(storageRoot, index);
return {
thread: newThreadId,
forkedFrom: {
step: stepHash,
},
};
}
/**
* Read a step's agent output as markdown (new command - requires #462)
* TODO: Implement once unified agent detail/turn schema is available
*/
export async function cmdStepRead(
storageRoot: string,
stepHash: CasRef,
before: number | null = null,
): Promise<string> {
const uwf = await createUwfStore(storageRoot);
const node = uwf.store.get(stepHash);
if (node === null) {
fail(`CAS node not found: ${stepHash}`);
}
if (node.type !== uwf.schemas.stepNode) {
fail(`node ${stepHash} is not a StepNode`);
}
const payload = node.payload as StepNodePayload;
if (!payload.output) {
fail(`step ${stepHash} has no output`);
}
// TODO: Implement progressive turn reading with --before N
// For now, return a placeholder
const outputNode = uwf.store.get(payload.output);
if (outputNode === null) {
fail(`output node not found: ${payload.output}`);
}
// Return the output as JSON for now
// Once #462 is implemented, this will properly format frontmatter + markdown
return JSON.stringify(outputNode.payload, null, 2);
}
+109 -96
View File
@@ -346,47 +346,65 @@ export async function cmdThreadShow(storageRoot: string, threadId: ThreadId): Pr
fail(`thread not found: ${threadId}`);
}
export type ThreadStatus = "idle" | "running" | "completed";
export type ThreadListItemWithStatus = ThreadListItem & {
status: ThreadStatus;
};
async function threadListItemFromActive(
storageRoot: string,
uwf: UwfStore,
threadId: ThreadId,
head: CasRef,
): Promise<ThreadListItem | null> {
): Promise<ThreadListItemWithStatus | null> {
const workflow = resolveWorkflowFromHead(uwf, head);
if (workflow === null) {
return null;
}
return { thread: threadId, workflow, head };
// Check if thread is currently running in background
const runningMarker = await isThreadRunning(storageRoot, threadId);
const status: ThreadStatus = runningMarker !== null ? "running" : "idle";
return { thread: threadId, workflow, head, status };
}
export async function cmdThreadList(
storageRoot: string,
includeAll: boolean,
): Promise<ThreadListItem[]> {
statusFilter: ThreadStatus | null,
): Promise<ThreadListItemWithStatus[]> {
const uwf = await createUwfStore(storageRoot);
const index = await loadThreadsIndex(storageRoot);
const items: ThreadListItem[] = [];
const items: ThreadListItemWithStatus[] = [];
// Add active threads
for (const [threadId, head] of Object.entries(index)) {
const item = await threadListItemFromActive(uwf, threadId as ThreadId, head);
const item = await threadListItemFromActive(storageRoot, uwf, threadId as ThreadId, head);
if (item !== null) {
items.push(item);
}
}
if (!includeAll) {
return items;
// Add completed threads if requested
if (statusFilter === "completed" || statusFilter === null) {
const activeIds = new Set(items.map((i) => i.thread));
const history = await loadThreadHistory(storageRoot);
for (const entry of history) {
if (!activeIds.has(entry.thread)) {
items.push({
thread: entry.thread,
workflow: entry.workflow,
head: entry.head,
status: "completed",
});
}
}
}
const activeIds = new Set(items.map((i) => i.thread));
const history = await loadThreadHistory(storageRoot);
for (const entry of history) {
if (!activeIds.has(entry.thread)) {
items.push({
thread: entry.thread,
workflow: entry.workflow,
head: entry.head,
});
}
// Apply status filter if provided
if (statusFilter !== null) {
return items.filter((item) => item.status === statusFilter);
}
return items;
@@ -857,7 +875,7 @@ async function archiveThread(
});
}
export async function cmdThreadStep(
export async function cmdThreadExec(
storageRoot: string,
threadId: ThreadId,
agentOverride: string | null,
@@ -953,7 +971,7 @@ async function cmdThreadStepBackground(
failStep(plog, "unable to determine script path for background execution");
}
const args = ["thread", "step", threadId, "--count", String(count)];
const args = ["thread", "exec", threadId, "--count", String(count)];
if (agentOverride !== null) {
args.push("--agent", agentOverride);
@@ -1085,47 +1103,6 @@ async function resolveHeadHash(storageRoot: string, threadId: ThreadId): Promise
fail(`thread not found: ${threadId}`);
}
export async function cmdThreadSteps(
storageRoot: string,
threadId: ThreadId,
): Promise<ThreadStepsOutput> {
const headHash = await resolveHeadHash(storageRoot, threadId);
const uwf = await createUwfStore(storageRoot);
const chain = walkChain(uwf, headHash);
const startNode = uwf.store.get(chain.startHash);
if (startNode === null) {
fail(`StartNode not found: ${chain.startHash}`);
}
const startEntry: StartEntry = {
hash: chain.startHash,
workflow: chain.start.workflow,
prompt: chain.start.prompt,
timestamp: startNode.timestamp,
};
const stepEntries: StepEntry[] = [];
const ordered = collectOrderedSteps(uwf, headHash, chain);
for (const item of ordered) {
stepEntries.push({
hash: item.hash,
role: item.payload.role,
output: expandOutput(uwf, item.payload.output),
detail: item.payload.detail,
agent: item.payload.agent,
timestamp: item.timestamp,
});
}
return {
thread: threadId,
workflow: chain.start.workflow,
steps: [startEntry, ...stepEntries],
};
}
export async function cmdThreadRead(
storageRoot: string,
threadId: ThreadId,
@@ -1153,49 +1130,85 @@ export async function cmdThreadRead(
});
}
export async function cmdThreadFork(
storageRoot: string,
stepHash: CasRef,
): Promise<ThreadForkOutput> {
const uwf = await createUwfStore(storageRoot);
const node = uwf.store.get(stepHash);
if (node === null) {
fail(`CAS node not found: ${stepHash}`);
}
if (node.type !== uwf.schemas.startNode && node.type !== uwf.schemas.stepNode) {
fail(`node ${stepHash} is not a StartNode or StepNode`);
}
export type StopOutput = {
thread: ThreadId;
stopped: boolean;
};
const newThreadId = generateUlid(Date.now()) as ThreadId;
export type CancelOutput = {
thread: ThreadId;
cancelled: boolean;
};
/**
* Stop background execution of a thread (but keep thread active)
*/
export async function cmdThreadStop(storageRoot: string, threadId: ThreadId): Promise<StopOutput> {
const index = await loadThreadsIndex(storageRoot);
index[newThreadId] = stepHash;
await saveThreadsIndex(storageRoot, index);
const head = index[threadId];
if (head === undefined) {
fail(`thread not active: ${threadId}`);
}
return {
thread: newThreadId,
forkedFrom: {
step: stepHash,
},
};
// Check if thread is running in background and terminate it
const runningMarker = await isThreadRunning(storageRoot, threadId);
if (runningMarker === null) {
process.stderr.write(`Warning: thread ${threadId} is not currently running\n`);
return { thread: threadId, stopped: false };
}
try {
process.kill(runningMarker.pid, "SIGTERM");
} catch {
// Process may have already exited, ignore error
}
await deleteMarker(storageRoot, threadId);
return { thread: threadId, stopped: true };
}
export async function cmdThreadStepDetails(
/**
* Cancel a thread (stop execution + move to history)
*/
export async function cmdThreadCancel(
storageRoot: string,
stepHash: CasRef,
): Promise<unknown> {
threadId: ThreadId,
): Promise<CancelOutput> {
const index = await loadThreadsIndex(storageRoot);
const head = index[threadId];
if (head === undefined) {
fail(`thread not active: ${threadId}`);
}
// Check if thread is running in background and terminate it
const runningMarker = await isThreadRunning(storageRoot, threadId);
if (runningMarker !== null) {
try {
process.kill(runningMarker.pid, "SIGTERM");
} catch {
// Process may have already exited, ignore error
}
await deleteMarker(storageRoot, threadId);
}
const uwf = await createUwfStore(storageRoot);
const node = uwf.store.get(stepHash);
if (node === null) {
fail(`CAS node not found: ${stepHash}`);
const workflow = resolveWorkflowFromHead(uwf, head);
if (workflow === null) {
fail(`failed to resolve workflow from head: ${head}`);
}
if (node.type !== uwf.schemas.stepNode) {
fail(`node ${stepHash} is not a StepNode`);
}
const payload = node.payload as StepNodePayload;
if (!payload.detail) {
fail(`step ${stepHash} has no detail`);
}
return expandDeep(uwf.store, payload.detail);
delete index[threadId];
await saveThreadsIndex(storageRoot, index);
const historyEntry: ThreadHistoryLine = {
thread: threadId,
workflow,
head,
completedAt: Date.now(),
};
await appendThreadHistory(storageRoot, historyEntry);
return { thread: threadId, cancelled: true };
}
export async function cmdThreadKill(storageRoot: string, threadId: ThreadId): Promise<KillOutput> {
@@ -29,7 +29,7 @@ export type WorkflowListEntry = {
origin: WorkflowOrigin;
};
export type WorkflowPutOutput = {
export type WorkflowAddOutput = {
name: string;
hash: CasRef;
};
@@ -111,10 +111,10 @@ export async function materializeWorkflowPayload(
};
}
export async function cmdWorkflowPut(
export async function cmdWorkflowAdd(
storageRoot: string,
filePath: string,
): Promise<WorkflowPutOutput> {
): Promise<WorkflowAddOutput> {
let text: string;
try {
text = await readFile(filePath, "utf8");