Compare commits

...

1 Commits

Author SHA1 Message Date
xiaoju 3c1fb1b71f feat(cli): add thread poke command
CI / check (pull_request) Successful in 3m46s
Re-runs the head step's agent with a supplementary prompt and replaces
the head step (rewires new step's prev to old head's prev) instead of
appending. Skips moderator re-route — the role of the head step is
reused.

Fixes #144
2026-06-07 05:59:44 +00:00
4 changed files with 723 additions and 0 deletions
+11
View File
@@ -0,0 +1,11 @@
---
"@united-workforce/cli": minor
---
feat(cli): add `uwf thread poke` command
New subcommand `uwf thread poke <thread-id> -p <prompt>` re-runs the head step's
agent with a supplementary prompt, replacing the head step's output. Unlike
`thread resume`, poke skips the moderator and rewrites the new step's `prev`
pointer so the new head replaces (not appends to) the old head. Works on idle
and suspended threads. Resolves issue #144 (Phase 1).
@@ -0,0 +1,549 @@
import { execFileSync } 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 { putSchema } from "@ocas/core";
import { openStore } from "@ocas/fs";
import type {
CasRef,
StepNodePayload,
ThreadId,
ThreadIndexEntry,
} from "@united-workforce/protocol";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { registerUwfSchemas } from "../schemas.js";
import { seedThreads } from "./thread-test-helpers.js";
const OUTPUT_SCHEMA = {
type: "object" as const,
properties: {
$status: { type: "string" as const },
note: { type: "string" as const },
},
required: ["$status"],
additionalProperties: false,
};
const THREAD_ID = "01POKESTEPTEST00000000" as ThreadId;
let tmpDir: string;
beforeEach(async () => {
tmpDir = await mkdtemp(join(tmpdir(), "cli-uwf-poke-test-"));
});
afterEach(async () => {
await rm(tmpDir, { recursive: true, force: true });
});
type SetupResult = {
casDir: string;
oldStepHash: CasRef;
oldStepPrev: CasRef | null;
oldStepCompletedAtMs: number;
startHash: CasRef;
workflowHash: CasRef;
mockAgentPath: string;
failingAgentPath: string;
promptCapturePath: string;
envCapturePath: string;
};
type SetupOpts = {
threadStatus: ThreadIndexEntry["status"];
multipleSteps: boolean;
newCompletedAtMs: number;
newStatus: string;
// The agent name to record in the head StepNode.agent field. Defaults to mockAgentPath.
stepAgentNameOverride: string | null;
// Whether to seed an actual head StepNode (false → only StartNode is the head).
withHeadStep: boolean;
};
async function setupThread(opts: Partial<SetupOpts> = {}): Promise<SetupResult> {
const cfg: SetupOpts = {
threadStatus: opts.threadStatus ?? "idle",
multipleSteps: opts.multipleSteps ?? false,
newCompletedAtMs: opts.newCompletedAtMs ?? 1716600005000,
newStatus: opts.newStatus ?? "ok",
stepAgentNameOverride: opts.stepAgentNameOverride ?? null,
withHeadStep: opts.withHeadStep ?? true,
};
const casDir = join(tmpDir, "cas");
await mkdir(casDir, { recursive: true });
const store = await openStore(casDir);
const schemas = await registerUwfSchemas(store);
const outputSchemaHash = await putSchema(store, OUTPUT_SCHEMA);
const workflowHash = await store.cas.put(schemas.workflow, {
name: "test-poke",
description: "poke command integration test",
roles: {
worker: {
description: "Worker role",
goal: "Work",
capabilities: [],
procedure: "work",
output: "result",
frontmatter: outputSchemaHash,
},
reviewer: {
description: "Reviewer role",
goal: "Review",
capabilities: [],
procedure: "review",
output: "result",
frontmatter: outputSchemaHash,
},
},
graph: {
$START: {
new: { role: "worker", prompt: "Start work", location: null },
resume: { role: "worker", prompt: "Resume the work", location: null },
},
worker: {
ok: { role: "reviewer", prompt: "Review the work", location: null },
needs_input: {
role: "$SUSPEND",
prompt: "Please clarify",
location: null,
},
},
reviewer: { done: { role: "$END", prompt: "Done", location: null } },
},
});
const startHash = await store.cas.put(schemas.startNode, {
workflow: workflowHash,
prompt: "Test poke task",
cwd: tmpDir,
});
process.env.OCAS_HOME = casDir;
// Paths for mock agent and capture files (set early so we can use mockAgentPath as the recorded agent name)
const promptCapturePath = join(tmpDir, "captured-prompt.txt");
const envCapturePath = join(tmpDir, "captured-env.txt");
const mockAgentPath = join(tmpDir, "mock-agent.sh");
const failingAgentPath = join(tmpDir, "failing-agent.sh");
// Build head StepNode chain
let oldStepPrev: CasRef | null = null;
if (cfg.multipleSteps) {
// First step: prev=null
const firstOutputHash = await store.cas.put(outputSchemaHash, { $status: "ok" });
const firstDetailHash = await store.cas.put(schemas.text, "first detail");
const firstStepHash = await store.cas.put(schemas.stepNode, {
start: startHash,
prev: null,
role: "worker",
output: firstOutputHash,
detail: firstDetailHash,
agent: cfg.stepAgentNameOverride ?? mockAgentPath,
edgePrompt: "Start work",
startedAtMs: 1716600000000,
completedAtMs: 1716600001000,
cwd: tmpDir,
assembledPrompt: null,
usage: null,
});
oldStepPrev = firstStepHash;
}
let oldStepHash: CasRef = startHash;
const oldStepCompletedAtMs = 1716600002000;
if (cfg.withHeadStep) {
const outputHash = await store.cas.put(outputSchemaHash, { $status: "ok" });
const detailHash = await store.cas.put(schemas.text, "head step detail");
oldStepHash = await store.cas.put(schemas.stepNode, {
start: startHash,
prev: oldStepPrev,
role: "worker",
output: outputHash,
detail: detailHash,
agent: cfg.stepAgentNameOverride ?? mockAgentPath,
edgePrompt: "Start work",
startedAtMs: 1716600001500,
completedAtMs: oldStepCompletedAtMs,
cwd: tmpDir,
assembledPrompt: null,
usage: null,
});
}
// Seed thread index entry. For "running" we let the test create the marker separately.
await seedThreads(tmpDir, {
[THREAD_ID]: {
head: oldStepHash,
status: cfg.threadStatus,
suspendedRole: cfg.threadStatus === "suspended" ? "worker" : null,
suspendMessage: cfg.threadStatus === "suspended" ? "Please clarify" : null,
completedAt:
cfg.threadStatus === "completed" || cfg.threadStatus === "cancelled"
? oldStepCompletedAtMs
: null,
},
});
// Mock agent always emits a stepNode keyed off the current thread head (which we
// observe through OCAS_HOME). The script writes prompt/env captures and then prints
// an adapter JSON that references a pre-built stepHash.
// We pre-build the agent's stepHash with prev=oldStepHash (normal append behaviour).
const newOutputHash = await store.cas.put(outputSchemaHash, {
$status: cfg.newStatus,
note: "poked output",
});
const newDetailHash = await store.cas.put(schemas.text, "poked detail");
const agentStepHash = await store.cas.put(schemas.stepNode, {
start: startHash,
prev: cfg.withHeadStep ? oldStepHash : null,
role: "worker",
output: newOutputHash,
detail: newDetailHash,
agent: "mock-agent-output",
edgePrompt: "poke prompt placeholder",
startedAtMs: cfg.newCompletedAtMs - 100,
completedAtMs: cfg.newCompletedAtMs,
cwd: tmpDir,
assembledPrompt: null,
usage: null,
});
const adapterJson = JSON.stringify({
stepHash: agentStepHash,
detailHash: newDetailHash,
role: "worker",
frontmatter: { $status: cfg.newStatus, note: "poked output" },
body: "",
startedAtMs: cfg.newCompletedAtMs - 100,
completedAtMs: cfg.newCompletedAtMs,
usage: null,
});
await writeFile(
mockAgentPath,
`#!/bin/sh
prompt=""
while [ $# -gt 0 ]; do
if [ "$1" = "--prompt" ]; then
prompt="$2"
shift 2
else
shift
fi
done
printf '%s' "$prompt" > '${promptCapturePath}'
printf 'OCAS_HOME=%s\\n' "$OCAS_HOME" > '${envCapturePath}'
echo '${adapterJson}'
`,
{ mode: 0o755 },
);
await writeFile(
failingAgentPath,
`#!/bin/sh
echo "boom" >&2
exit 7
`,
{ mode: 0o755 },
);
const configPath = join(tmpDir, "config.yaml");
await writeFile(
configPath,
`defaultAgent: uwf-hermes\ndefaultModel: test-model\nagentOverrides: null\nagents: {}\nproviders: {}\nmodels: {}\n`,
);
return {
casDir,
oldStepHash,
oldStepPrev,
oldStepCompletedAtMs,
startHash,
workflowHash,
mockAgentPath,
failingAgentPath,
promptCapturePath,
envCapturePath,
};
}
function runUwf(
args: string[],
casDir: string,
): { stdout: string; stderr: string; status: number } {
const cliPath = join(dirname(fileURLToPath(import.meta.url)), "..", "..", "dist", "cli.js");
try {
const stdout = execFileSync(process.execPath, [cliPath, ...args], {
encoding: "utf8",
stdio: ["ignore", "pipe", "pipe"],
env: {
...process.env,
UWF_HOME: tmpDir,
OCAS_HOME: casDir,
},
cwd: tmpDir,
timeout: 30000,
});
return { stdout, stderr: "", status: 0 };
} catch (error) {
const err = error as NodeJS.ErrnoException & {
stdout?: string | Buffer;
stderr?: string | Buffer;
status?: number;
};
return {
stdout: typeof err.stdout === "string" ? err.stdout : (err.stdout?.toString("utf8") ?? ""),
stderr: typeof err.stderr === "string" ? err.stderr : (err.stderr?.toString("utf8") ?? ""),
status: err.status ?? 1,
};
}
}
// ── Group 1: CLI argument validation ───────────────────────────────────────
describe("uwf thread poke - CLI argument validation", () => {
test("1.1 missing -p flag exits non-zero", async () => {
const { casDir } = await setupThread();
const result = runUwf(["thread", "poke", THREAD_ID], casDir);
expect(result.status).not.toBe(0);
expect(result.stderr.toLowerCase()).toMatch(/required|missing|prompt/);
});
test("1.2 -p without --agent succeeds", async () => {
const { casDir } = await setupThread();
const result = runUwf(["thread", "poke", THREAD_ID, "-p", "do it again"], casDir);
expect(result.status).toBe(0);
});
test("1.3 -p with --agent succeeds", async () => {
const { casDir, mockAgentPath } = await setupThread();
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "do it again", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
});
});
// ── Group 2: Guard errors ──────────────────────────────────────────────────
describe("uwf thread poke - guard errors", () => {
test("2.1 thread not found", async () => {
const { casDir } = await setupThread();
const result = runUwf(["thread", "poke", "01NOSUCHTHREAD0000000A", "-p", "prompt"], casDir);
expect(result.status).not.toBe(0);
expect(result.stderr.toLowerCase()).toMatch(/not found|not active/);
});
test("2.2 thread running rejects poke", async () => {
const { casDir, workflowHash } = await setupThread();
// Create background marker to simulate running
const { createMarker } = await import("../background/index.js");
await createMarker(tmpDir, {
thread: THREAD_ID,
workflow: workflowHash,
pid: process.pid,
startedAt: Date.now(),
});
const result = runUwf(["thread", "poke", THREAD_ID, "-p", "prompt"], casDir);
expect(result.status).not.toBe(0);
expect(result.stderr.toLowerCase()).toContain("already executing");
});
test("2.3 completed thread rejects poke", async () => {
const { casDir } = await setupThread({ threadStatus: "completed" });
const result = runUwf(["thread", "poke", THREAD_ID, "-p", "prompt"], casDir);
expect(result.status).not.toBe(0);
expect(result.stderr.toLowerCase()).toMatch(/cannot be poked|completed/);
});
test("2.4 cancelled thread rejects poke", async () => {
const { casDir } = await setupThread({ threadStatus: "cancelled" });
const result = runUwf(["thread", "poke", THREAD_ID, "-p", "prompt"], casDir);
expect(result.status).not.toBe(0);
expect(result.stderr.toLowerCase()).toMatch(/cannot be poked|cancelled/);
});
test("2.5 thread head is StartNode (no StepNode) rejects poke", async () => {
const { casDir } = await setupThread({ withHeadStep: false });
const result = runUwf(["thread", "poke", THREAD_ID, "-p", "prompt"], casDir);
expect(result.status).not.toBe(0);
expect(result.stderr.toLowerCase()).toMatch(/no step|cannot be poked/);
});
});
// ── Group 3: Success happy path ────────────────────────────────────────────
describe("uwf thread poke - success", () => {
test("3.1, 3.4 idle thread → new head differs from old, thread index updated", async () => {
const { casDir, oldStepHash, mockAgentPath } = await setupThread();
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const cliOutput = JSON.parse(result.stdout.trim());
expect(cliOutput.head).not.toBe(oldStepHash);
const { createUwfStore, getThread } = await import("../store.js");
const uwf = await createUwfStore(tmpDir);
const entry = getThread(uwf.varStore, THREAD_ID);
expect(entry?.head).toBe(cliOutput.head);
});
test("3.2 new step's prev equals old head's prev (replace, not append)", async () => {
const { casDir, oldStepPrev, mockAgentPath } = await setupThread({ multipleSteps: true });
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const cliOutput = JSON.parse(result.stdout.trim());
const { createUwfStore } = await import("../store.js");
const uwf = await createUwfStore(tmpDir);
const node = uwf.store.cas.get(cliOutput.head as CasRef);
expect(node).not.toBeNull();
expect(node?.type).toBe(uwf.schemas.stepNode);
const payload = node?.payload as StepNodePayload;
expect(payload.prev).toBe(oldStepPrev);
});
test("3.2b new step's prev is null when old head was the first step", async () => {
// multipleSteps:false means oldHead.prev = null
const { casDir, mockAgentPath } = await setupThread({ multipleSteps: false });
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const cliOutput = JSON.parse(result.stdout.trim());
const { createUwfStore } = await import("../store.js");
const uwf = await createUwfStore(tmpDir);
const node = uwf.store.cas.get(cliOutput.head as CasRef);
const payload = node?.payload as StepNodePayload;
expect(payload.prev).toBeNull();
});
test("3.3 new step's completedAtMs is later than old", async () => {
const { casDir, oldStepCompletedAtMs, mockAgentPath } = await setupThread();
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const cliOutput = JSON.parse(result.stdout.trim());
const { createUwfStore } = await import("../store.js");
const uwf = await createUwfStore(tmpDir);
const node = uwf.store.cas.get(cliOutput.head as CasRef);
const payload = node?.payload as StepNodePayload;
expect(payload.completedAtMs).toBeGreaterThan(oldStepCompletedAtMs);
});
test("3.5 status remains idle after poke (no completion/suspend)", async () => {
const { casDir, mockAgentPath } = await setupThread();
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const cliOutput = JSON.parse(result.stdout.trim());
expect(cliOutput.status).toBe("idle");
expect(cliOutput.done).toBe(false);
expect(cliOutput.suspendedRole).toBeNull();
expect(cliOutput.suspendMessage).toBeNull();
});
test("3.6 currentRole unchanged after poke (no moderator re-route)", async () => {
// Before poke: idle thread with worker step having $status=ok → moderator would route to reviewer.
// After poke (mock returns same $status=ok), moderator routing remains the same.
const { casDir, mockAgentPath } = await setupThread();
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const cliOutput = JSON.parse(result.stdout.trim());
expect(cliOutput.currentRole).toBe("reviewer");
});
});
// ── Group 4: Agent resolution ──────────────────────────────────────────────
describe("uwf thread poke - agent resolution", () => {
test("4.1 without --agent, agent command read from head step's agent field", async () => {
// Head step's agent field points at mockAgentPath (default in setupThread)
const { casDir, promptCapturePath } = await setupThread();
const result = runUwf(["thread", "poke", THREAD_ID, "-p", "redo"], casDir);
expect(result.status).toBe(0);
const captured = await readFile(promptCapturePath, "utf8");
expect(captured).toBe("redo");
});
test("4.2 with --agent, explicit override is used", async () => {
// Head step records "uwf-mock" (which is not a real binary). Override with mockAgentPath.
const { casDir, mockAgentPath } = await setupThread({ stepAgentNameOverride: "uwf-mock" });
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
});
});
// ── Group 5: Prompt passthrough ────────────────────────────────────────────
describe("uwf thread poke - prompt passthrough", () => {
test("5.1 -p value is passed to agent as --prompt", async () => {
const { casDir, mockAgentPath, promptCapturePath } = await setupThread();
const supplement = "Use the REST API instead.";
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", supplement, "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const captured = await readFile(promptCapturePath, "utf8");
expect(captured).toBe(supplement);
});
});
// ── Group 6: Edge cases ────────────────────────────────────────────────────
describe("uwf thread poke - edge cases", () => {
test("6.1 poke succeeds on suspended thread", async () => {
const { casDir, oldStepHash, mockAgentPath } = await setupThread({
threadStatus: "suspended",
});
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", mockAgentPath],
casDir,
);
expect(result.status).toBe(0);
const cliOutput = JSON.parse(result.stdout.trim());
expect(cliOutput.head).not.toBe(oldStepHash);
expect(cliOutput.status).toBe("idle");
expect(cliOutput.suspendedRole).toBeNull();
expect(cliOutput.suspendMessage).toBeNull();
});
test("6.2 agent failure leaves thread head unchanged", async () => {
const { casDir, oldStepHash, failingAgentPath } = await setupThread();
const result = runUwf(
["thread", "poke", THREAD_ID, "-p", "redo", "--agent", failingAgentPath],
casDir,
);
expect(result.status).not.toBe(0);
const { createUwfStore, getThread } = await import("../store.js");
const uwf = await createUwfStore(tmpDir);
const entry = getThread(uwf.varStore, THREAD_ID);
expect(entry?.head).toBe(oldStepHash);
});
});
+21
View File
@@ -17,6 +17,7 @@ import {
cmdThreadCancel, cmdThreadCancel,
cmdThreadExec, cmdThreadExec,
cmdThreadList, cmdThreadList,
cmdThreadPoke,
cmdThreadRead, cmdThreadRead,
cmdThreadResume, cmdThreadResume,
cmdThreadShow, cmdThreadShow,
@@ -290,6 +291,26 @@ thread
}); });
}); });
thread
.command("poke")
.description("Re-run the head step's agent with a supplementary prompt (replaces head step)")
.argument("<thread-id>", "Thread ULID")
.requiredOption("-p, --prompt <text>", "Supplementary prompt for the agent")
.option("--agent <cmd>", "Override agent command (defaults to head step's agent)")
.action((threadId: string, opts: { prompt: string; agent: string | undefined }) => {
const storageRoot = resolveStorageRoot();
runAction(async () => {
const agentOverride = opts.agent ?? null;
const result = await cmdThreadPoke(
storageRoot,
threadId as ThreadId,
opts.prompt,
agentOverride,
);
writeOutput(result);
});
});
thread thread
.command("stop") .command("stop")
.description("Stop background execution of a thread (keep thread active)") .description("Stop background execution of a thread (keep thread active)")
+142
View File
@@ -199,6 +199,7 @@ const PL_THREAD_ARCHIVED = "F4D8Q2K5";
const PL_STEP_ERROR = "B8T5N1V6"; const PL_STEP_ERROR = "B8T5N1V6";
const PL_BACKGROUND_START = "X7Q4W9M2"; const PL_BACKGROUND_START = "X7Q4W9M2";
const PL_THREAD_RESUME = "K2R7M4N8"; const PL_THREAD_RESUME = "K2R7M4N8";
const PL_THREAD_POKE = "P4Q9R3X7";
type ResumeStepConfig = { type ResumeStepConfig = {
role: string; role: string;
@@ -1135,6 +1136,147 @@ export async function cmdThreadResume(
}); });
} }
/**
* Validate that a thread can be poked. Returns the existing entry and the head StepNode payload.
* Fails (process exit) when the thread is missing, running, completed, cancelled, or has no
* StepNode at its head.
*/
async function validatePokePreconditions(
storageRoot: string,
uwf: UwfStore,
threadId: ThreadId,
): Promise<{ entry: ThreadIndexEntry; oldHead: CasRef; oldHeadPayload: StepNodePayload }> {
const runningMarker = await isThreadRunning(storageRoot, threadId);
if (runningMarker !== null) {
fail(`thread already executing in background (PID: ${runningMarker.pid})`);
}
const entry = getThread(uwf.varStore, threadId);
if (entry === null) {
fail(`thread not active: ${threadId}`);
}
if (entry.status === "completed" || entry.status === "cancelled") {
fail(`thread cannot be poked: ${threadId} (status: ${entry.status})`);
}
const oldHead = entry.head;
const oldHeadNode = uwf.store.cas.get(oldHead);
if (oldHeadNode === null) {
fail(`CAS node not found: ${oldHead}`);
}
if (oldHeadNode.type !== uwf.schemas.stepNode) {
fail("thread cannot be poked: no step to replace (head is StartNode)");
}
return { entry, oldHead, oldHeadPayload: oldHeadNode.payload as StepNodePayload };
}
/**
* Resolve the next role from the post-poke chain state, used for the StepOutput.currentRole field.
* Returns null when the next role is $END, evaluation fails, or the result is a suspend.
*/
function resolveCurrentRoleFromChain(
uwfAfter: UwfStore,
workflow: WorkflowPayload,
replacedHash: CasRef,
): string | null {
const chainAfter = walkChain(uwfAfter, replacedHash);
const { lastRole, lastOutput } = resolveEvaluateArgs(uwfAfter, chainAfter);
const afterResult = evaluate(workflow.graph, lastRole, lastOutput);
if (!afterResult.ok || isSuspendResult(afterResult.value)) {
return null;
}
if (afterResult.value.role === END_ROLE) {
return null;
}
return afterResult.value.role;
}
/**
* Poke a thread: re-run the agent on the head step with a supplementary prompt,
* replacing the head step's output. The new step's `prev` points to the OLD head's
* `prev` — semantically replacing (not appending to) the head. The moderator is NOT
* re-evaluated for routing; the role of the head step is re-used.
*/
export async function cmdThreadPoke(
storageRoot: string,
threadId: ThreadId,
prompt: string,
agentOverride: string | null,
): Promise<StepOutput> {
const uwf = await createUwfStore(storageRoot);
const { entry, oldHeadPayload } = await validatePokePreconditions(storageRoot, uwf, threadId);
const chain = walkChain(uwf, entry.head);
const workflowHash = chain.start.workflow;
const threadCwd = chain.start.cwd;
const plog = createProcessLogger({
storageRoot,
context: { thread: threadId, workflow: workflowHash },
});
// Resolve the agent: --agent override wins; otherwise read from old head step's `agent` field.
const config = await loadWorkflowConfig(storageRoot);
const workflow = loadWorkflowPayload(uwf, workflowHash);
const role = oldHeadPayload.role;
const agent =
agentOverride !== null
? resolveAgentConfig(config, workflow, role, agentOverride)
: parseAgentOverride(oldHeadPayload.agent);
const effectiveCwd = oldHeadPayload.cwd !== "" ? oldHeadPayload.cwd : threadCwd;
plog.log(PL_THREAD_POKE, `poke role=${role} agent=${agent.command}`, null);
plog.log(PL_AGENT_SPAWN, `spawning agent command=${agent.command}`, {
args: [...agent.args, threadId, role].join(" "),
});
loadDotenv({ path: getEnvPath(storageRoot) });
// Spawn the agent. The agent will create a new StepNode with prev=oldHead (it reads
// the active thread head). After the agent returns, we rewrite that node's prev so
// that the new head replaces the old head instead of appending after it.
const agentResult = spawnAgent(plog, agent, threadId, role, prompt, effectiveCwd);
const agentStepHash = agentResult.stepHash as CasRef;
plog.log(PL_AGENT_DONE, `agent returned head=${agentStepHash}`, null);
const uwfAfter = await createUwfStore(storageRoot);
const agentNode = uwfAfter.store.cas.get(agentStepHash);
if (agentNode === null || agentNode.type !== uwfAfter.schemas.stepNode) {
failStep(plog, `agent returned hash that is not a StepNode: ${agentStepHash}`);
}
const agentPayload = agentNode.payload as StepNodePayload;
// Rewrite the new step so that its `prev` points to the OLD head's prev (replace semantics).
const replacedPayload: StepNodePayload = {
...agentPayload,
prev: oldHeadPayload.prev,
};
const replacedHash = await uwfAfter.store.cas.put(uwfAfter.schemas.stepNode, replacedPayload);
const replacedNode = uwfAfter.store.cas.get(replacedHash);
if (replacedNode === null || !validate(uwfAfter.store, replacedNode)) {
failStep(plog, "rewritten StepNode failed schema validation");
}
// Update thread head to the replaced step. Status becomes idle (no moderator re-route).
setThread(uwfAfter.varStore, threadId, updateThreadHead(entry, replacedHash));
return {
workflow: workflowHash,
thread: threadId,
head: replacedHash,
status: "idle",
currentRole: resolveCurrentRoleFromChain(uwfAfter, workflow, replacedHash),
suspendedRole: null,
suspendMessage: null,
done: false,
background: null,
};
}
export function validateCount(count: number): void { export function validateCount(count: number): void {
if (count < 1 || !Number.isInteger(count)) { if (count < 1 || !Number.isInteger(count)) {
throw new Error(`--count must be a positive integer, got: ${count}`); throw new Error(`--count must be a positive integer, got: ${count}`);