Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions skills/dynamic-workflows/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -58,12 +58,39 @@ return { summary, findings }
| `agent(prompt, opts?)` | Throws on failure. `opts`: `label`, `phase`, `schema`, `model`, `effort`, `provider`, `isolation: 'worktree'` |
| `parallel(thunks)` | Barrier; throw → `null` slot |
| `pipeline(items, ...stages)` | Per-item chains; no cross-item barrier |
| `settle(() => operation)` | DevSpace extension: convert a thrown operation into `{ ok, value }` or `{ ok, error: { kind, message, retryable } }` |
| `phase(title)` / `log(msg)` | Progress; journaled |
| `args` | Run input (object preferred) |
| `workflow(name\|{scriptPath}, args?)` | Nested, depth 1, shared call index |

**No `writeMode`.** Teach read-only vs write in the prompt. Use `isolation: 'worktree'` when parallel mutators would conflict (git required).

### Failure-aware orchestration

Default behavior stays Claude-compatible: direct `agent()` failures throw, while
`parallel()` and `pipeline()` map a failed branch/item to `null`. Use `settle()`
only when the script must distinguish failure kinds or implement fallback:

```js
const primary = await settle(() =>
agent('Read-only security review', { provider: 'claude' }),
)

const review = primary.ok
? primary
: primary.error.kind === 'provider_unavailable'
? await settle(() =>
agent('Read-only security review', { provider: 'codex' }),
)
: primary

return review
```

Inside `parallel()`, wrap each branch with `settle()` to preserve failures as
data instead of `null`. Failed settled outcomes are journaled failures and are
not replayed as successful cached agent results.
Comment on lines +90 to +92

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Scope the journaling claim to agent calls.

settle() does not journal itself: settle(() => { throw new Error("x") }) returns failure data without an agent-call record. Limit the statement to underlying agent calls that fail after execution begins.

Proposed wording
-Inside `parallel()`, wrap each branch with `settle()` to preserve failures as
-data instead of `null`. Failed settled outcomes are journaled failures and are
-not replayed as successful cached agent results.
+Inside `parallel()`, wrap each agent branch with `settle()` to preserve failures
+as data instead of `null`. Failed agent calls are journaled as failures and are
+not replayed as successful cached agent results.
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
Inside `parallel()`, wrap each branch with `settle()` to preserve failures as
data instead of `null`. Failed settled outcomes are journaled failures and are
not replayed as successful cached agent results.
Inside `parallel()`, wrap each agent branch with `settle()` to preserve failures
as data instead of `null`. Failed agent calls are journaled as failures and are
not replayed as successful cached agent results.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@skills/dynamic-workflows/SKILL.md` around lines 90 - 92, Update the
journaling statement in the parallel() documentation to scope it specifically to
underlying agent calls that fail after execution begins. Do not claim that
settle() itself journals failures; retain that settle() preserves branch
failures as data and clarify that only failed agent-call outcomes are journaled
and excluded from successful cached-result replay.


### Determinism bans

`Date.now()`, `Math.random()`, and `new Date()` without args throw. Pass timestamps via `args` if needed.
Expand Down
53 changes: 53 additions & 0 deletions src/workflow-api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,10 @@ import {
type WorkflowMeta,
} from "./workflow-types.js";
import { agentOptsSchema } from "./workflow-contracts.js";
import {
isWorkflowOperationError,
serializeWorkflowError,
} from "./workflow-errors.js";

// ---------------------------------------------------------------------------
// Host deps (injected by engine; fakes OK in tests)
Expand Down Expand Up @@ -574,6 +578,18 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi {
);
};

const settle = async (...args: unknown[]): Promise<unknown> => {
const task = args[0];
if (typeof task !== "function") {
throw new WorkflowEngineError("internal", "settle(task) requires a function");
}
try {
return { ok: true, value: await (task as () => unknown | Promise<unknown>)() };
} catch (error) {
return { ok: false, error: normalizeSettledError(error) };
}
};

const phase = (...args: unknown[]): void => {
const title = args[0];
if (typeof title !== "string" || !title.trim()) {
Expand Down Expand Up @@ -628,6 +644,7 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi {
agent: agent as WorkflowSandboxApi["agent"],
parallel: parallel as WorkflowSandboxApi["parallel"],
pipeline: pipeline as WorkflowSandboxApi["pipeline"],
settle: settle as WorkflowSandboxApi["settle"],
phase: phase as WorkflowSandboxApi["phase"],
log: log as WorkflowSandboxApi["log"],
args: deps.args,
Expand All @@ -639,6 +656,42 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi {
};
}

function normalizeSettledError(error: unknown): {
kind: import("./workflow-types.js").WorkflowErrorKind;
message: string;
retryable: boolean;
} {
if (error instanceof WorkflowEngineError) {
return {
kind: error.kind,
message: error.message,
retryable:
error.kind === "provider_unavailable" ||
error.kind === "provider_disabled" ||
error.kind === "no_provider",
};
}
if (isWorkflowOperationError(error)) {
const serialized = serializeWorkflowError(error);
return {
kind: serialized.kind,
message: serialized.message,
retryable: serialized.retryable,
};
}
if (error && typeof error === "object" && "name" in error) {
const name = String((error as { name: unknown }).name);
if (name === "AbortError") {
return { kind: "cancelled", message: "Workflow cancelled", retryable: false };
}
}
return {
kind: "internal",
message: error instanceof Error ? error.message : String(error),
retryable: false,
};
}

function formatReplayMiss(miss: WorkflowReplayMiss): string {
return miss.reason === "identity_changed" && miss.changedFields?.length
? `${miss.reason}:${miss.changedFields.join(",")}`
Expand Down
14 changes: 14 additions & 0 deletions src/workflow-contracts.ts
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,20 @@ export interface WorkflowAgent {

export type WorkflowTask<T = unknown> = () => T | Promise<T>;

export interface WorkflowSettledError {
kind: WorkflowErrorKind;
message: string;
retryable: boolean;
}

export type WorkflowSettledResult<T> =
| { ok: true; value: T }
| { ok: false; error: WorkflowSettledError };

export interface WorkflowSettle {
<T>(task: WorkflowTask<T>): Promise<WorkflowSettledResult<Awaited<T>>>;
}

export interface WorkflowParallel {
<const T extends readonly WorkflowTask[]>(
tasks: T,
Expand Down
55 changes: 55 additions & 0 deletions src/workflow-engine.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,61 @@ import { createStubBudget } from "./workflow-types.js";
await rm(dir, { recursive: true, force: true });
}

// ---------------------------------------------------------------------------
// settle() preserves a typed failure as data without changing parallel defaults
// ---------------------------------------------------------------------------
{
const dir = await mkdtemp(join(tmpdir(), "wf-settle-"));
const store = new WorkflowStore(dir);
const run = store.createRun({
name: "settle",
source: "inline",
scriptPath: "inline",
scriptHash: "h",
workspaceRoot: dir,
});
const api = createWorkflowApi({
runId: run.id,
journal: store,
meta: { name: "settle", description: "d" },
args: undefined,
concurrency: 2,
signal: new AbortController().signal,
workspaceRoot: dir,
enabledProviders: ["codex"],
runProvider: async (input) => {
if (input.prompt === "fail") {
throw new WorkflowEngineError("provider_unavailable", "provider missing");
}
return { finalResponse: `ok:${input.prompt}` };
},
});

const ok = await api.settle(() => api.agent("a"));
assert.deepEqual(ok, { ok: true, value: "ok:a" });

const failed = await api.settle(() => api.agent("fail"));
assert.deepEqual(failed, {
ok: false,
error: {
kind: "provider_unavailable",
message: "provider missing",
retryable: true,
},
});
assert.equal(store.getAgentCall(run.id, 1)?.status, "failed");

const rows = await api.parallel([
() => api.settle(() => api.agent("fail")),
() => api.settle(() => api.agent("b")),
]);
assert.equal(rows[0]?.ok, false);
assert.deepEqual(rows[1], { ok: true, value: "ok:b" });

store.close();
await rm(dir, { recursive: true, force: true });
}

// ---------------------------------------------------------------------------
// pipeline — no barrier across items (item B can finish stage2 before A stage1 ends)
// ---------------------------------------------------------------------------
Expand Down
1 change: 1 addition & 0 deletions src/workflow-engine.ts
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,7 @@ async function executeNestedOnApi(input: {
agent: input.parentApi.agent,
parallel: input.parentApi.parallel,
pipeline: input.parentApi.pipeline,
settle: input.parentApi.settle,
phase: input.parentApi.phase,
log: input.parentApi.log,
args: input.args,
Expand Down
10 changes: 10 additions & 0 deletions src/workflow-sandbox.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,16 @@ function api(meta: WorkflowMeta, logs?: string[]): WorkflowSandboxApi {
agent: async () => "",
parallel: async () => [],
pipeline: async () => [],
settle: async <T>(task: () => T | Promise<T>) => {
try {
return { ok: true, value: await task() };
} catch (error) {
return {
ok: false,
error: { kind: "internal", message: String(error), retryable: false },
};
}
},
Comment on lines +15 to +24

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Match production Error-message normalization in both test doubles.

Production returns error.message for Error instances; these stubs return String(error) (for example, "Error: boom"). Tests inspecting settled failures can therefore validate behavior scripts never receive.

  • src/workflow-sandbox.test.ts#L15-L24: use error instanceof Error ? error.message : String(error).
  • src/workflow-script.test.ts#L90-L99: use the same normalization.
📍 Affects 2 files
  • src/workflow-sandbox.test.ts#L15-L24 (this comment)
  • src/workflow-script.test.ts#L90-L99
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/workflow-sandbox.test.ts` around lines 15 - 24, Normalize caught errors
like production in both test doubles: update the settle implementations in
src/workflow-sandbox.test.ts lines 15-24 and src/workflow-script.test.ts lines
90-99 to use error.message for Error instances and String(error) otherwise,
while preserving the existing failure result structure.

phase: () => {},
log: (msg: unknown) => {
logs?.push(String(msg));
Expand Down
3 changes: 3 additions & 0 deletions src/workflow-sandbox.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import type {
WorkflowNested,
WorkflowParallel,
WorkflowPipeline,
WorkflowSettle,
} from "./workflow-types.js";

export class WorkflowDeterminismError extends Error {
Expand All @@ -21,6 +22,7 @@ export interface WorkflowSandboxApi {
agent: WorkflowAgent;
parallel: WorkflowParallel;
pipeline: WorkflowPipeline;
settle: WorkflowSettle;
phase: (title: string) => void;
log: (...args: unknown[]) => unknown;
args: JsonValue | undefined;
Expand Down Expand Up @@ -71,6 +73,7 @@ export async function runWorkflowSandbox(
agent: api.agent,
parallel: api.parallel,
pipeline: api.pipeline,
settle: api.settle,
phase: api.phase,
log: api.log,
args: api.args,
Expand Down
10 changes: 10 additions & 0 deletions src/workflow-script.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,16 @@ async function runBody(source: string): Promise<unknown> {
return Promise.all(thunks.map((t) => t().catch(() => null)));
},
pipeline: async (...args: unknown[]) => args[0],
settle: async (task) => {
try {
return { ok: true, value: await task() };
} catch (error) {
return {
ok: false,
error: { kind: "internal", message: String(error), retryable: false },
};
}
},
phase: () => {},
log: (msg: unknown) => {
logs.push(String(msg));
Expand Down
2 changes: 1 addition & 1 deletion src/workflow-script.ts
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ export function parseWorkflowScript(

// Inject host APIs as params. `meta` stays as the script's own `const meta`
// (would TDZ/redeclare if also injected). `console` lives on the sandbox globals.
const wrapped = `(async ({ agent, parallel, pipeline, phase, log, args, budget, workflow }) => {\n${body}\n})`;
const wrapped = `(async ({ agent, parallel, pipeline, settle, phase, log, args, budget, workflow }) => {\n${body}\n})`;
let script: vm.Script;
try {
script = new vm.Script(wrapped, {
Expand Down
1 change: 1 addition & 0 deletions src/workflow-tools.ts
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ Workflow scripts (JS only):
agent(prompt, { label?, phase?, schema?, model?, effort?, provider?, isolation?: 'worktree' })
parallel(thunks) → Array<T|null> // barrier; throw → null
pipeline(items, ...stages) // no cross-item barrier
settle(() => operation) → { ok, value } | { ok, error }
phase(title); log(msg); args
workflow(name | { scriptPath }, args?) // nest depth 1
Bans: Date.now(), Math.random(), new Date() without args.
Expand Down
3 changes: 3 additions & 0 deletions src/workflow-types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@ export type {
WorkflowPipeline,
WorkflowRunSource,
WorkflowRunStatus,
WorkflowSettle,
WorkflowSettledError,
WorkflowSettledResult,
WorkflowTask,
} from "./workflow-contracts.js";

Expand Down
Loading