mirror of
https://github.com/openclaw/lobster.git
synced 2026-08-14 00:48:09 +00:00
feat: add command-level input requests
Adds state-backed command-level requestInput resume across CLI/tool/SDK pipelines and workflow pipeline steps. Closes #101.
This commit is contained in:
Peter Steinberger
@@ -4,6 +4,7 @@ All notable changes to Lobster will be documented in this file.
|
||||
|
||||
## Unreleased
|
||||
|
||||
- Add command-level `ctx.requestInput(...)` for CLI/tool/SDK pipeline commands, with state-backed same-command resume, bounded command-input replay, and workflow `pipeline:` propagation (Issue [#101](https://github.com/openclaw/lobster/issues/101)).
|
||||
- Warn when LLM usage records an unknown or missing model ID or invalid `LOBSTER_LLM_PRICING_JSON`, keeping zero-cost fallback behavior visible for `cost_limit` users. Thanks to [@KrasimirKralev](https://github.com/KrasimirKralev) (Issue [#107](https://github.com/openclaw/lobster/issues/107)).
|
||||
- Require Node.js 22 or newer for the npm package, matching release CI.
|
||||
- Write Lobster state files atomically while preserving restricted file modes, preventing truncated resume/session state after process termination. Thanks to [@KrasimirKralev](https://github.com/KrasimirKralev) (Issues [#108](https://github.com/openclaw/lobster/issues/108), [#109](https://github.com/openclaw/lobster/issues/109), PR [#110](https://github.com/openclaw/lobster/pull/110)).
|
||||
|
||||
@@ -218,6 +218,12 @@ Notes:
|
||||
- `LOBSTER_APPROVAL_INITIATED_BY` can provide a default initiator id at run time.
|
||||
- `LOBSTER_APPROVAL_APPROVED_BY` is used at resume/approval time for identity checks.
|
||||
|
||||
### Command-level input requests
|
||||
|
||||
Pipeline commands can call `ctx.requestInput({ prompt, responseSchema, defaults, subject, suspendedState })` to pause in tool mode, workflows, or the SDK and resume the same command after a structured response. CLI/tool resume tokens store only a state key; the persisted state validates the suspended request metadata before returning the submitted response to the command. SDK same-command resumes store the command frame in the configured SDK state directory.
|
||||
|
||||
Commands are re-run on resume, so they must be idempotent until `requestInput` returns. Array-backed command input is snapshotted with bounds for replay; lazy stream input is not buffered and requires a compact JSON `suspendedState` supplied by the command. On resume, call `ctx.requestInput.getSuspendedState()` before reading lazy input to restore that command-owned continuation state.
|
||||
|
||||
## Visualizing workflows
|
||||
|
||||
Use `lobster graph` to inspect workflow structure before execution.
|
||||
|
||||
+28
-9
@@ -294,6 +294,10 @@ async function handleRun({ argv, registry }) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (output.halted && isPipelineInputRequest(output.items)) {
|
||||
throw new Error("requestInput requires --mode tool when stdin is not interactive");
|
||||
}
|
||||
|
||||
// Human mode: if the last command didn't render, print JSON.
|
||||
if (!output.rendered) {
|
||||
process.stdout.write(JSON.stringify(output.items, null, 2));
|
||||
@@ -313,6 +317,12 @@ async function handleRun({ argv, registry }) {
|
||||
}
|
||||
}
|
||||
|
||||
function isPipelineInputRequest(items) {
|
||||
return (
|
||||
items.length === 1 && items[0]?.type === "input_request" && items[0]?.commandInput !== undefined
|
||||
);
|
||||
}
|
||||
|
||||
function parseRunArgs(argv) {
|
||||
const rest = [];
|
||||
let mode = "human";
|
||||
@@ -697,10 +707,24 @@ async function handleResume({ argv, registry }) {
|
||||
}
|
||||
}
|
||||
|
||||
const isSameStageInput =
|
||||
resumeState.haltType === "input_request" && resumeState.resumeMode === "same_stage";
|
||||
const remaining = resumeState.pipeline.slice(resumeState.resumeAtIndex);
|
||||
const input = streamFromItems(
|
||||
resumeState.haltType === "input_request" ? [response] : resumeState.items,
|
||||
);
|
||||
const input = isSameStageInput
|
||||
? resumeState.items
|
||||
: resumeState.haltType === "input_request"
|
||||
? [response]
|
||||
: resumeState.items;
|
||||
const requestInputResume = isSameStageInput
|
||||
? {
|
||||
state: resumeState.commandInput!,
|
||||
response,
|
||||
onConsumed: async () => {
|
||||
await cleanupIndex();
|
||||
await deleteStateJson({ env: process.env, key: previousStateKey });
|
||||
},
|
||||
}
|
||||
: undefined;
|
||||
|
||||
try {
|
||||
const output = await runPipeline({
|
||||
@@ -712,6 +736,7 @@ async function handleResume({ argv, registry }) {
|
||||
env: process.env,
|
||||
mode,
|
||||
input,
|
||||
requestInputResume,
|
||||
});
|
||||
await cleanupIndex();
|
||||
const finalized = await finalizePipelineToolRun({
|
||||
@@ -737,12 +762,6 @@ async function handleResume({ argv, registry }) {
|
||||
}
|
||||
}
|
||||
|
||||
function streamFromItems(items: unknown[]) {
|
||||
return (async function* () {
|
||||
for (const item of items) yield item;
|
||||
})();
|
||||
}
|
||||
|
||||
async function readVersion() {
|
||||
const { readFile } = await import("node:fs/promises");
|
||||
const { fileURLToPath } = await import("node:url");
|
||||
|
||||
@@ -78,9 +78,6 @@ export const askCommand = {
|
||||
const subjectFromStdin = Boolean(args["subject-from-stdin"] ?? args.subjectFromStdin);
|
||||
const schemaRaw = typeof args.schema === "string" ? args.schema : null;
|
||||
|
||||
const items = [];
|
||||
for await (const item of input) items.push(item);
|
||||
|
||||
const defaultSchema = {
|
||||
type: "object",
|
||||
properties: {
|
||||
@@ -105,7 +102,22 @@ export const askCommand = {
|
||||
}
|
||||
const responseValidator = compileAskValidator(responseSchema);
|
||||
|
||||
let subject;
|
||||
const forceEmit = Boolean(args.emit);
|
||||
const emit = forceEmit || ctx.mode === "tool" || !isInteractive(ctx.stdin);
|
||||
const canRequestInput =
|
||||
!forceEmit && ctx.mode === "tool" && typeof ctx.requestInput === "function";
|
||||
const restoredState = canRequestInput ? ctx.requestInput.getSuspendedState?.() : undefined;
|
||||
const restoredAskState =
|
||||
restoredState && typeof restoredState === "object" && restoredState.type === "ask"
|
||||
? restoredState
|
||||
: null;
|
||||
|
||||
const items = [];
|
||||
if (!restoredAskState) {
|
||||
for await (const item of input) items.push(item);
|
||||
}
|
||||
|
||||
let subject = restoredAskState?.subject;
|
||||
if (subjectFromStdin && items.length > 0) {
|
||||
const preview = items
|
||||
.map((item) => (typeof item === "string" ? item : JSON.stringify(item)))
|
||||
@@ -114,8 +126,18 @@ export const askCommand = {
|
||||
subject = { text: preview };
|
||||
}
|
||||
|
||||
const emit = Boolean(args.emit) || ctx.mode === "tool" || !isInteractive(ctx.stdin);
|
||||
if (emit) {
|
||||
if (canRequestInput) {
|
||||
const response = await ctx.requestInput({
|
||||
prompt,
|
||||
responseSchema,
|
||||
...(subject ? { subject } : null),
|
||||
suspendedState: { type: "ask", ...(subject ? { subject } : null) },
|
||||
});
|
||||
return {
|
||||
output: asStream([response]),
|
||||
};
|
||||
}
|
||||
return {
|
||||
halt: true,
|
||||
output: (async function* () {
|
||||
@@ -151,3 +173,7 @@ export const askCommand = {
|
||||
throw lastError ?? new Error("ask response failed schema validation");
|
||||
},
|
||||
};
|
||||
|
||||
async function* asStream(items) {
|
||||
for (const item of items) yield item;
|
||||
}
|
||||
|
||||
+18
-11
@@ -277,10 +277,24 @@ export async function resumeToolRequest({
|
||||
}
|
||||
}
|
||||
|
||||
const isSameStageInput =
|
||||
resumeState.haltType === "input_request" && resumeState.resumeMode === "same_stage";
|
||||
const remaining = resumeState.pipeline.slice(resumeState.resumeAtIndex);
|
||||
const input = streamFromItems(
|
||||
resumeState.haltType === "input_request" ? [response] : resumeState.items,
|
||||
);
|
||||
const input = isSameStageInput
|
||||
? resumeState.items
|
||||
: resumeState.haltType === "input_request"
|
||||
? [response]
|
||||
: resumeState.items;
|
||||
const requestInputResume = isSameStageInput
|
||||
? {
|
||||
state: resumeState.commandInput!,
|
||||
response,
|
||||
onConsumed: async () => {
|
||||
await cleanupIndex();
|
||||
await deleteStateJson({ env: runtime.env, key: payload.stateKey });
|
||||
},
|
||||
}
|
||||
: undefined;
|
||||
|
||||
try {
|
||||
const output = await runPipeline({
|
||||
@@ -295,6 +309,7 @@ export async function resumeToolRequest({
|
||||
llmAdapters: runtime.llmAdapters,
|
||||
signal: runtime.signal,
|
||||
input,
|
||||
requestInputResume,
|
||||
});
|
||||
|
||||
await cleanupIndex();
|
||||
@@ -362,14 +377,6 @@ function errorEnvelope(type: string, message: string): ToolEnvelope {
|
||||
};
|
||||
}
|
||||
|
||||
function streamFromItems(items: unknown[]) {
|
||||
return (async function* () {
|
||||
for (const item of items) {
|
||||
yield item;
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
async function resolveWorkflowFile(candidate: string, cwd: string) {
|
||||
const { stat } = await import("node:fs/promises");
|
||||
const resolved = path.isAbsolute(candidate) ? candidate : path.resolve(cwd, candidate);
|
||||
|
||||
@@ -0,0 +1,527 @@
|
||||
import { stableStringify } from "./state/store.js";
|
||||
import { compileCached } from "./validation.js";
|
||||
|
||||
export type RequestInputParams = {
|
||||
prompt: string;
|
||||
responseSchema: unknown;
|
||||
defaults?: unknown;
|
||||
subject?: unknown;
|
||||
suspendedState?: unknown;
|
||||
};
|
||||
|
||||
export type RequestInputMetadata = {
|
||||
prompt: string;
|
||||
responseSchema: unknown;
|
||||
defaults?: unknown;
|
||||
subject?: unknown;
|
||||
signature: string;
|
||||
};
|
||||
|
||||
export type CommandInputHistoryEntry = {
|
||||
requestIndex: number;
|
||||
metadata: RequestInputMetadata;
|
||||
suspendedState?: unknown;
|
||||
response: unknown;
|
||||
};
|
||||
|
||||
export type CommandInputPendingRequest = {
|
||||
requestIndex: number;
|
||||
metadata: RequestInputMetadata;
|
||||
suspendedState?: unknown;
|
||||
};
|
||||
|
||||
export type CommandInputState = {
|
||||
pending: CommandInputPendingRequest;
|
||||
history: CommandInputHistoryEntry[];
|
||||
};
|
||||
|
||||
export type CommandInputResume = {
|
||||
state: CommandInputState;
|
||||
response: unknown;
|
||||
consumed?: boolean;
|
||||
onConsumed?: () => void | Promise<void>;
|
||||
};
|
||||
|
||||
export type PipelineCommandInputRequest = {
|
||||
type: "input_request";
|
||||
prompt: string;
|
||||
responseSchema: unknown;
|
||||
defaults?: unknown;
|
||||
subject?: unknown;
|
||||
items: unknown[];
|
||||
commandInput: CommandInputState;
|
||||
};
|
||||
|
||||
const MAX_REPLAY_ITEMS = 1000;
|
||||
const MAX_REPLAY_BYTES = 1024 * 1024;
|
||||
const MAX_REQUEST_HISTORY = 100;
|
||||
|
||||
export class InputRequestSuspension extends Error {
|
||||
stageIndex: number;
|
||||
request: PipelineCommandInputRequest;
|
||||
|
||||
constructor(stageIndex: number, request: PipelineCommandInputRequest) {
|
||||
super("Input request suspended");
|
||||
this.name = "InputRequestSuspension";
|
||||
this.stageIndex = stageIndex;
|
||||
this.request = request;
|
||||
}
|
||||
}
|
||||
|
||||
export class RequestInputResumeError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "RequestInputResumeError";
|
||||
}
|
||||
}
|
||||
|
||||
export function createInputTracker(input: AsyncIterable<unknown> | Iterable<unknown>) {
|
||||
const knownItems = Array.isArray(input) ? input : null;
|
||||
let iterator: AsyncIterator<unknown> | null = null;
|
||||
let completed = false;
|
||||
let closed = false;
|
||||
let replayEnabled = true;
|
||||
let replaySnapshotItems: unknown[] = [];
|
||||
let replaySnapshotIndex = 0;
|
||||
let replaySnapshotError: Error | null = null;
|
||||
let replaySnapshotBytes = 0;
|
||||
|
||||
const iterable = {
|
||||
async *[Symbol.asyncIterator]() {
|
||||
const iter = getIterator();
|
||||
let hasPrimaryError = false;
|
||||
let yieldedIndex = 0;
|
||||
try {
|
||||
while (true) {
|
||||
const next = await iter.next();
|
||||
if (next.done) {
|
||||
completed = true;
|
||||
return;
|
||||
}
|
||||
if (knownItems) {
|
||||
snapshotKnownItemsThrough(yieldedIndex);
|
||||
yieldedIndex += 1;
|
||||
}
|
||||
yield next.value;
|
||||
}
|
||||
} catch (err) {
|
||||
hasPrimaryError = true;
|
||||
throw err;
|
||||
} finally {
|
||||
await closeTrackedIterator({ suppressErrors: hasPrimaryError });
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
return {
|
||||
iterable,
|
||||
getReplayItems(hasSuspendedState: boolean) {
|
||||
if (!replayEnabled) {
|
||||
throw new Error("requestInput replay is no longer available after command output");
|
||||
}
|
||||
if (!knownItems) {
|
||||
if (hasSuspendedState) return [];
|
||||
throw new Error("requestInput requires suspendedState when command input is streaming");
|
||||
}
|
||||
snapshotKnownItemsThrough(knownItems.length - 1);
|
||||
if (replaySnapshotError) throw replaySnapshotError;
|
||||
return snapshotArray(replaySnapshotItems, "requestInput replay input");
|
||||
},
|
||||
disableReplay() {
|
||||
replayEnabled = false;
|
||||
},
|
||||
async close(options: { suppressErrors?: boolean } = {}) {
|
||||
await closeTrackedIterator({ suppressErrors: options.suppressErrors === true });
|
||||
},
|
||||
};
|
||||
|
||||
function getIterator() {
|
||||
iterator ??= toAsyncIterator(input);
|
||||
return iterator;
|
||||
}
|
||||
|
||||
async function closeTrackedIterator({ suppressErrors }: { suppressErrors: boolean }) {
|
||||
if (!iterator || completed || closed) return;
|
||||
closed = true;
|
||||
await closeIterator(iterator, { suppressErrors });
|
||||
}
|
||||
|
||||
function snapshotKnownItemsThrough(index: number) {
|
||||
if (!knownItems || replaySnapshotError) return;
|
||||
while (replaySnapshotIndex <= index && replaySnapshotIndex < knownItems.length) {
|
||||
try {
|
||||
const snapshot = snapshotJson(knownItems[replaySnapshotIndex], "requestInput replay input");
|
||||
replaySnapshotBytes += Buffer.byteLength(JSON.stringify(snapshot), "utf8");
|
||||
if (
|
||||
replaySnapshotItems.length + 1 > MAX_REPLAY_ITEMS ||
|
||||
replaySnapshotBytes > MAX_REPLAY_BYTES
|
||||
) {
|
||||
throw new Error("requestInput replay limit exceeded");
|
||||
}
|
||||
replaySnapshotItems.push(snapshot);
|
||||
replaySnapshotIndex += 1;
|
||||
} catch (err) {
|
||||
replaySnapshotError = err instanceof Error ? err : new Error(String(err));
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function createStageRequestInput({
|
||||
ctx,
|
||||
stageIndex,
|
||||
mode,
|
||||
inputTracker,
|
||||
isCommandActive,
|
||||
getInactiveReason,
|
||||
isOutputStarted,
|
||||
resume,
|
||||
}: {
|
||||
ctx: any;
|
||||
stageIndex: number;
|
||||
mode: string;
|
||||
inputTracker: ReturnType<typeof createInputTracker>;
|
||||
isCommandActive: () => boolean;
|
||||
getInactiveReason?: () => string | undefined;
|
||||
isOutputStarted: () => boolean;
|
||||
resume?: CommandInputResume;
|
||||
}) {
|
||||
let requestIndex = 0;
|
||||
const history: CommandInputHistoryEntry[] = [...(resume?.state.history ?? [])];
|
||||
|
||||
const requestInput = async function requestInput(params: RequestInputParams) {
|
||||
if (!isCommandActive()) {
|
||||
throw new Error(
|
||||
getInactiveReason?.() ?? "requestInput cannot run after the command has completed",
|
||||
);
|
||||
}
|
||||
|
||||
const metadata = snapshotRequestMetadata(params);
|
||||
const requestedSuspendedState =
|
||||
params.suspendedState === undefined
|
||||
? undefined
|
||||
: snapshotJson(params.suspendedState, "requestInput suspendedState");
|
||||
const historical = history[requestIndex];
|
||||
if (historical) {
|
||||
assertMetadataMatches(historical.requestIndex, historical.metadata, requestIndex, metadata);
|
||||
assertSuspendedStateMatches(historical.suspendedState, requestedSuspendedState);
|
||||
const response = snapshotJson(historical.response, "requestInput response");
|
||||
validateRequestInputResponse(metadata.responseSchema, response, "requestInput");
|
||||
requestIndex += 1;
|
||||
return response;
|
||||
}
|
||||
|
||||
if (resume && !resume.consumed) {
|
||||
assertMetadataMatches(
|
||||
resume.state.pending.requestIndex,
|
||||
resume.state.pending.metadata,
|
||||
requestIndex,
|
||||
metadata,
|
||||
);
|
||||
assertSuspendedStateMatches(resume.state.pending.suspendedState, requestedSuspendedState);
|
||||
const response = snapshotJson(resume.response, "requestInput response");
|
||||
validateRequestInputResponse(metadata.responseSchema, response, "requestInput");
|
||||
const historyResponse = snapshotJson(response, "requestInput response");
|
||||
await resume.onConsumed?.();
|
||||
resume.consumed = true;
|
||||
history.push({
|
||||
requestIndex,
|
||||
metadata,
|
||||
...(requestedSuspendedState !== undefined
|
||||
? { suspendedState: requestedSuspendedState }
|
||||
: null),
|
||||
response: historyResponse,
|
||||
});
|
||||
requestIndex += 1;
|
||||
return response;
|
||||
}
|
||||
|
||||
if (mode === "human" && isInteractive(ctx.stdin)) {
|
||||
return requestInputInteractively(ctx, metadata);
|
||||
}
|
||||
|
||||
if (isOutputStarted()) {
|
||||
throw new Error("requestInput cannot suspend after this command has produced output");
|
||||
}
|
||||
if (history.length >= MAX_REQUEST_HISTORY) {
|
||||
throw new Error("requestInput replay history limit exceeded");
|
||||
}
|
||||
|
||||
const items = inputTracker.getReplayItems(requestedSuspendedState !== undefined);
|
||||
const pending: CommandInputPendingRequest = {
|
||||
requestIndex,
|
||||
metadata,
|
||||
...(requestedSuspendedState !== undefined
|
||||
? { suspendedState: requestedSuspendedState }
|
||||
: null),
|
||||
};
|
||||
throw new InputRequestSuspension(stageIndex, {
|
||||
type: "input_request",
|
||||
prompt: metadata.prompt,
|
||||
responseSchema: metadata.responseSchema,
|
||||
...(metadata.defaults !== undefined ? { defaults: metadata.defaults } : null),
|
||||
...(metadata.subject !== undefined ? { subject: metadata.subject } : null),
|
||||
items,
|
||||
commandInput: {
|
||||
pending,
|
||||
history,
|
||||
},
|
||||
});
|
||||
};
|
||||
requestInput.getSuspendedState = function getSuspendedState() {
|
||||
if (requestIndex < history.length) {
|
||||
return snapshotOptionalState(history[requestIndex].suspendedState);
|
||||
}
|
||||
if (resume && !resume.consumed && resume.state.pending.requestIndex === requestIndex) {
|
||||
return snapshotOptionalState(resume.state.pending.suspendedState);
|
||||
}
|
||||
return undefined;
|
||||
};
|
||||
return requestInput;
|
||||
}
|
||||
|
||||
export function assertRequestInputResumeConsumed(resume?: CommandInputResume) {
|
||||
if (resume && !resume.consumed) {
|
||||
throw new RequestInputResumeError("resume input response was not consumed by requestInput");
|
||||
}
|
||||
}
|
||||
|
||||
export function snapshotRequestMetadata(params: RequestInputParams): RequestInputMetadata {
|
||||
validateRequestInputParams(params);
|
||||
const responseSchema = snapshotJson(params.responseSchema, "requestInput responseSchema");
|
||||
const defaults =
|
||||
params.defaults === undefined
|
||||
? undefined
|
||||
: snapshotJson(params.defaults, "requestInput defaults");
|
||||
const subject =
|
||||
params.subject === undefined ? undefined : snapshotJson(params.subject, "requestInput subject");
|
||||
const unsigned = {
|
||||
prompt: params.prompt,
|
||||
responseSchema,
|
||||
...(defaults !== undefined ? { defaults } : null),
|
||||
...(subject !== undefined ? { subject } : null),
|
||||
};
|
||||
return {
|
||||
...unsigned,
|
||||
signature: stableStringify(unsigned),
|
||||
};
|
||||
}
|
||||
|
||||
export function validateCommandInputState(value: unknown): CommandInputState {
|
||||
if (!value || typeof value !== "object") throw new Error("Invalid pipeline resume state");
|
||||
const data = value as Partial<CommandInputState>;
|
||||
validatePending(data.pending);
|
||||
if (!Array.isArray(data.history)) throw new Error("Invalid pipeline resume state");
|
||||
if (data.history.length !== data.pending.requestIndex) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
data.history.forEach(validateHistoryEntry);
|
||||
return data as CommandInputState;
|
||||
}
|
||||
|
||||
export function validateRequestInputResponse(schema: unknown, response: unknown, label: string) {
|
||||
let validator;
|
||||
try {
|
||||
validator = compileCached(schema as any);
|
||||
} catch {
|
||||
throw new Error(`${label} response schema is invalid`);
|
||||
}
|
||||
if (validator(response)) return;
|
||||
const first = validator.errors?.[0];
|
||||
const pathValue = first?.instancePath || "/";
|
||||
const reason = first?.message ? ` ${first.message}` : "";
|
||||
throw new Error(`${label} response failed schema validation at ${pathValue}:${reason}`);
|
||||
}
|
||||
|
||||
function validateRequestInputParams(params: RequestInputParams) {
|
||||
if (!params || typeof params !== "object") {
|
||||
throw new Error("requestInput params must be an object");
|
||||
}
|
||||
if (typeof params.prompt !== "string" || params.prompt.length === 0) {
|
||||
throw new Error("requestInput prompt is required");
|
||||
}
|
||||
if (params.responseSchema === undefined) {
|
||||
throw new Error("requestInput responseSchema is required");
|
||||
}
|
||||
try {
|
||||
compileCached(params.responseSchema as any);
|
||||
} catch {
|
||||
throw new Error("requestInput response schema is invalid");
|
||||
}
|
||||
}
|
||||
|
||||
function validatePending(value: unknown): asserts value is CommandInputPendingRequest {
|
||||
if (!value || typeof value !== "object") throw new Error("Invalid pipeline resume state");
|
||||
const data = value as Partial<CommandInputPendingRequest>;
|
||||
if (
|
||||
typeof data.requestIndex !== "number" ||
|
||||
!Number.isInteger(data.requestIndex) ||
|
||||
data.requestIndex < 0
|
||||
) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
validateStoredMetadata(data.metadata);
|
||||
if (data.suspendedState !== undefined) {
|
||||
snapshotJson(data.suspendedState, "requestInput suspendedState");
|
||||
}
|
||||
}
|
||||
|
||||
function validateHistoryEntry(value: unknown, index: number) {
|
||||
if (!value || typeof value !== "object") throw new Error("Invalid pipeline resume state");
|
||||
const data = value as Partial<CommandInputHistoryEntry>;
|
||||
if (data.requestIndex !== index) throw new Error("Invalid pipeline resume state");
|
||||
validateStoredMetadata(data.metadata);
|
||||
if (data.suspendedState !== undefined) {
|
||||
snapshotJson(data.suspendedState, "requestInput suspendedState");
|
||||
}
|
||||
if (data.response === undefined) throw new Error("Invalid pipeline resume state");
|
||||
snapshotJson(data.response, "requestInput response");
|
||||
}
|
||||
|
||||
function validateStoredMetadata(value: unknown): asserts value is RequestInputMetadata {
|
||||
if (!value || typeof value !== "object") throw new Error("Invalid pipeline resume state");
|
||||
const data = value as Partial<RequestInputMetadata>;
|
||||
if (typeof data.prompt !== "string" || data.prompt.length === 0) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (typeof data.signature !== "string" || data.signature.length === 0) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (data.responseSchema === undefined) throw new Error("Invalid pipeline resume state");
|
||||
const actual = snapshotRequestMetadata({
|
||||
prompt: data.prompt,
|
||||
responseSchema: data.responseSchema,
|
||||
...(data.defaults !== undefined ? { defaults: data.defaults } : null),
|
||||
...(data.subject !== undefined ? { subject: data.subject } : null),
|
||||
});
|
||||
if (actual.signature !== data.signature) throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
|
||||
function assertMetadataMatches(
|
||||
storedIndex: number,
|
||||
stored: RequestInputMetadata,
|
||||
actualIndex: number,
|
||||
actual: RequestInputMetadata,
|
||||
) {
|
||||
if (
|
||||
storedIndex !== actualIndex ||
|
||||
stored.prompt !== actual.prompt ||
|
||||
stored.signature !== actual.signature
|
||||
) {
|
||||
throw new RequestInputResumeError(
|
||||
"requestInput resume request does not match suspended request",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function assertSuspendedStateMatches(stored: unknown, actual: unknown) {
|
||||
if (stableStringify(stored) !== stableStringify(actual)) {
|
||||
throw new RequestInputResumeError("requestInput resume state does not match suspended request");
|
||||
}
|
||||
}
|
||||
|
||||
function snapshotOptionalState(state: unknown) {
|
||||
return state === undefined ? undefined : snapshotJson(state, "requestInput suspendedState");
|
||||
}
|
||||
|
||||
function snapshotArray(items: readonly unknown[], label: string) {
|
||||
if (items.length > MAX_REPLAY_ITEMS) throw new Error("requestInput replay item limit exceeded");
|
||||
let totalBytes = 0;
|
||||
return items.map((item) => {
|
||||
const snapshot = snapshotJson(item, label);
|
||||
totalBytes += Buffer.byteLength(JSON.stringify(snapshot), "utf8");
|
||||
if (totalBytes > MAX_REPLAY_BYTES) {
|
||||
throw new Error("requestInput replay byte limit exceeded");
|
||||
}
|
||||
return snapshot;
|
||||
});
|
||||
}
|
||||
|
||||
function snapshotJson(value: unknown, label: string): unknown {
|
||||
assertJsonSerializable(value, label, new WeakSet());
|
||||
const text = JSON.stringify(value);
|
||||
if (text === undefined) throw new Error(`${label} must be JSON-serializable`);
|
||||
return JSON.parse(text);
|
||||
}
|
||||
|
||||
function assertJsonSerializable(value: unknown, label: string, seen: WeakSet<object>) {
|
||||
if (value === null) return;
|
||||
const type = typeof value;
|
||||
if (type === "string" || type === "boolean") return;
|
||||
if (type === "number") {
|
||||
if (!Number.isFinite(value)) throw new Error(`${label} must be JSON-serializable`);
|
||||
return;
|
||||
}
|
||||
if (type === "undefined" || type === "function" || type === "symbol" || type === "bigint") {
|
||||
throw new Error(`${label} must be JSON-serializable`);
|
||||
}
|
||||
const object = value as object;
|
||||
if (seen.has(object)) throw new Error(`${label} must be JSON-serializable`);
|
||||
const prototype = Object.getPrototypeOf(object);
|
||||
if (prototype !== Object.prototype && prototype !== null && !Array.isArray(value)) {
|
||||
throw new Error(`${label} must be JSON-serializable`);
|
||||
}
|
||||
seen.add(object);
|
||||
if (Array.isArray(value)) {
|
||||
for (let index = 0; index < value.length; index++) {
|
||||
if (!(index in value)) throw new Error(`${label} must be JSON-serializable`);
|
||||
assertJsonSerializable(value[index], label, seen);
|
||||
}
|
||||
} else {
|
||||
for (const key of Object.keys(value as Record<string, unknown>)) {
|
||||
assertJsonSerializable((value as Record<string, unknown>)[key], label, seen);
|
||||
}
|
||||
}
|
||||
seen.delete(object);
|
||||
}
|
||||
|
||||
async function requestInputInteractively(ctx: any, metadata: RequestInputMetadata) {
|
||||
ctx.stdout.write(`${metadata.prompt}\n> `);
|
||||
const { readLineFromStream } = await import("./read_line.js");
|
||||
const raw = await readLineFromStream(ctx.stdin, { timeoutMs: 0 });
|
||||
let response;
|
||||
try {
|
||||
response = JSON.parse(String(raw ?? "").trim());
|
||||
} catch {
|
||||
throw new Error("requestInput response must be valid JSON");
|
||||
}
|
||||
validateRequestInputResponse(metadata.responseSchema, response, "requestInput");
|
||||
return snapshotJson(response, "requestInput response");
|
||||
}
|
||||
|
||||
function isInteractive(stdin: any) {
|
||||
return Boolean(stdin?.isTTY);
|
||||
}
|
||||
|
||||
async function closeIterator(
|
||||
iterator: AsyncIterator<unknown>,
|
||||
{ suppressErrors }: { suppressErrors: boolean },
|
||||
) {
|
||||
if (typeof iterator.return !== "function") return;
|
||||
try {
|
||||
await iterator.return();
|
||||
} catch (err) {
|
||||
if (!suppressErrors) throw err;
|
||||
// Cleanup must not mask the original command error or suspension.
|
||||
}
|
||||
}
|
||||
|
||||
function toAsyncIterator(input: AsyncIterable<unknown> | Iterable<unknown>) {
|
||||
if (typeof (input as any)[Symbol.asyncIterator] === "function") {
|
||||
return (input as AsyncIterable<unknown>)[Symbol.asyncIterator]();
|
||||
}
|
||||
if (typeof (input as any)[Symbol.iterator] === "function") {
|
||||
const iterator = (input as Iterable<unknown>)[Symbol.iterator]();
|
||||
return {
|
||||
async next() {
|
||||
return iterator.next();
|
||||
},
|
||||
async return() {
|
||||
if (typeof iterator.return === "function") iterator.return();
|
||||
return { done: true, value: undefined };
|
||||
},
|
||||
};
|
||||
}
|
||||
throw new Error("input is not iterable");
|
||||
}
|
||||
@@ -9,14 +9,17 @@ import {
|
||||
writeStateJson,
|
||||
} from "./state/store.js";
|
||||
import { compileCached } from "./validation.js";
|
||||
import { validateCommandInputState, type CommandInputState } from "./input_request.js";
|
||||
|
||||
export type PipelineResumeState = {
|
||||
pipeline: Array<{ name: string; args: Record<string, unknown>; raw: string }>;
|
||||
resumeAtIndex: number;
|
||||
items: unknown[];
|
||||
haltType?: "approval_request" | "input_request";
|
||||
resumeMode?: "next_stage" | "same_stage";
|
||||
inputSchema?: unknown;
|
||||
prompt?: string;
|
||||
commandInput?: CommandInputState;
|
||||
createdAt: string;
|
||||
};
|
||||
|
||||
@@ -34,6 +37,7 @@ export type PipelineInputRequest = {
|
||||
defaults?: unknown;
|
||||
subject?: unknown;
|
||||
items?: unknown[];
|
||||
commandInput?: CommandInputState;
|
||||
};
|
||||
|
||||
export type PipelineRunOutput = {
|
||||
@@ -134,13 +138,19 @@ export async function finalizePipelineToolRun(params: {
|
||||
}
|
||||
|
||||
if (inputRequest) {
|
||||
const resumeMode = inputRequest.commandInput ? "same_stage" : "next_stage";
|
||||
const nextStateKey = await savePipelineResumeState(params.env, {
|
||||
pipeline: params.pipeline,
|
||||
resumeAtIndex: (params.output.haltedAt?.index ?? -1) + 1,
|
||||
items: [],
|
||||
resumeAtIndex:
|
||||
resumeMode === "same_stage"
|
||||
? (params.output.haltedAt?.index ?? -1)
|
||||
: (params.output.haltedAt?.index ?? -1) + 1,
|
||||
items: resumeMode === "same_stage" ? (inputRequest.items ?? []) : [],
|
||||
haltType: "input_request",
|
||||
resumeMode,
|
||||
inputSchema: inputRequest.responseSchema,
|
||||
prompt: inputRequest.prompt,
|
||||
...(inputRequest.commandInput ? { commandInput: inputRequest.commandInput } : null),
|
||||
createdAt: new Date().toISOString(),
|
||||
});
|
||||
if (params.previousStateKey) {
|
||||
@@ -199,7 +209,15 @@ export async function loadPipelineResumeState(
|
||||
}
|
||||
const data = stored as Partial<PipelineResumeState>;
|
||||
if (!Array.isArray(data.pipeline)) throw new Error("Invalid pipeline resume state");
|
||||
if (typeof data.resumeAtIndex !== "number") throw new Error("Invalid pipeline resume state");
|
||||
validatePipelineShape(data.pipeline);
|
||||
if (
|
||||
typeof data.resumeAtIndex !== "number" ||
|
||||
!Number.isInteger(data.resumeAtIndex) ||
|
||||
data.resumeAtIndex < 0 ||
|
||||
data.resumeAtIndex > data.pipeline.length
|
||||
) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (!Array.isArray(data.items)) throw new Error("Invalid pipeline resume state");
|
||||
if (
|
||||
data.haltType !== undefined &&
|
||||
@@ -207,6 +225,24 @@ export async function loadPipelineResumeState(
|
||||
) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (data.resumeMode !== undefined && !["next_stage", "same_stage"].includes(data.resumeMode)) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (data.haltType === "input_request") {
|
||||
if (data.inputSchema === undefined || typeof data.prompt !== "string") {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (data.resumeMode === "same_stage") {
|
||||
if (data.resumeAtIndex >= data.pipeline.length) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
data.commandInput = validateCommandInputState(data.commandInput);
|
||||
} else if (data.commandInput !== undefined) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
} else if (data.resumeMode === "same_stage" || data.commandInput !== undefined) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
return data as PipelineResumeState;
|
||||
}
|
||||
|
||||
@@ -227,3 +263,17 @@ export function validatePipelineInputResponse(schema: unknown, response: unknown
|
||||
const reason = first?.message ? ` ${first.message}` : "";
|
||||
throw new Error(`pipeline input response failed schema validation at ${pathValue}:${reason}`);
|
||||
}
|
||||
|
||||
function validatePipelineShape(pipeline: unknown[]) {
|
||||
for (const stage of pipeline) {
|
||||
if (!stage || typeof stage !== "object") throw new Error("Invalid pipeline resume state");
|
||||
const data = stage as Record<string, unknown>;
|
||||
if (typeof data.name !== "string" || data.name.length === 0) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (!data.args || typeof data.args !== "object" || Array.isArray(data.args)) {
|
||||
throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
if (typeof data.raw !== "string") throw new Error("Invalid pipeline resume state");
|
||||
}
|
||||
}
|
||||
|
||||
+182
-9
@@ -1,4 +1,12 @@
|
||||
import { createJsonRenderer } from "./renderers/json.js";
|
||||
import {
|
||||
InputRequestSuspension,
|
||||
RequestInputResumeError,
|
||||
assertRequestInputResumeConsumed,
|
||||
createInputTracker,
|
||||
createStageRequestInput,
|
||||
type CommandInputResume,
|
||||
} from "./input_request.js";
|
||||
|
||||
export async function runPipeline({
|
||||
pipeline,
|
||||
@@ -13,6 +21,8 @@ export async function runPipeline({
|
||||
llmAdapters = undefined,
|
||||
signal = undefined,
|
||||
dryRun = false,
|
||||
requestInputResume = undefined,
|
||||
requestInputEnabled = true,
|
||||
}: {
|
||||
pipeline: any[];
|
||||
registry: any;
|
||||
@@ -26,17 +36,20 @@ export async function runPipeline({
|
||||
llmAdapters?: Record<string, any> | undefined;
|
||||
signal?: AbortSignal | undefined;
|
||||
dryRun?: boolean;
|
||||
requestInputResume?: CommandInputResume | undefined;
|
||||
requestInputEnabled?: boolean;
|
||||
}) {
|
||||
if (dryRun) {
|
||||
return dryRunPipeline({ pipeline, registry, stderr });
|
||||
}
|
||||
|
||||
let stream = input ?? emptyStream();
|
||||
let stream = input ?? [];
|
||||
let rendered = false;
|
||||
let halted = false;
|
||||
let haltedAt = null;
|
||||
let pipelineOutputStarted = false;
|
||||
|
||||
const ctx = {
|
||||
const baseCtx = {
|
||||
stdin,
|
||||
stdout,
|
||||
stderr,
|
||||
@@ -46,7 +59,6 @@ export async function runPipeline({
|
||||
cwd,
|
||||
llmAdapters,
|
||||
signal,
|
||||
render: createJsonRenderer(stdout),
|
||||
};
|
||||
|
||||
for (let idx = 0; idx < pipeline.length; idx++) {
|
||||
@@ -56,26 +68,124 @@ export async function runPipeline({
|
||||
throw new Error(`Unknown command: ${stage.name}`);
|
||||
}
|
||||
|
||||
const result = await command.run({ input: stream, args: stage.args, ctx });
|
||||
const inputTracker = createInputTracker(stream);
|
||||
const stageResume = idx === 0 ? requestInputResume : undefined;
|
||||
let commandActive = true;
|
||||
let inactiveReason: string | undefined;
|
||||
let commandOutputStarted = false;
|
||||
let stageFinished = false;
|
||||
async function finishStage({ assertResume = true, suppressCloseErrors = false } = {}) {
|
||||
if (stageFinished) return;
|
||||
stageFinished = true;
|
||||
commandActive = false;
|
||||
inputTracker.disableReplay();
|
||||
await inputTracker.close({ suppressErrors: suppressCloseErrors });
|
||||
if (assertResume) assertRequestInputResumeConsumed(stageResume);
|
||||
}
|
||||
const stageStdout = trackWritableOutput(stdout, () => {
|
||||
pipelineOutputStarted = true;
|
||||
});
|
||||
const ctx = {
|
||||
...baseCtx,
|
||||
stdout: stageStdout,
|
||||
render: createJsonRenderer(stageStdout),
|
||||
};
|
||||
const stageCtx = {
|
||||
...ctx,
|
||||
requestInput: requestInputEnabled
|
||||
? createStageRequestInput({
|
||||
ctx,
|
||||
stageIndex: idx,
|
||||
mode,
|
||||
inputTracker,
|
||||
isCommandActive: () => commandActive,
|
||||
getInactiveReason: () => inactiveReason,
|
||||
isOutputStarted: () => pipelineOutputStarted || commandOutputStarted,
|
||||
resume: stageResume,
|
||||
})
|
||||
: createUnsupportedRequestInput(),
|
||||
};
|
||||
|
||||
let result;
|
||||
try {
|
||||
result = await command.run({ input: inputTracker.iterable, args: stage.args, ctx: stageCtx });
|
||||
} catch (err) {
|
||||
await finishStage({ assertResume: false, suppressCloseErrors: true });
|
||||
if (haltForInputRequest(err)) break;
|
||||
assertNoUnconsumedResumeAfterError(stageResume, err);
|
||||
throw err;
|
||||
}
|
||||
|
||||
if (result?.rendered) {
|
||||
rendered = true;
|
||||
}
|
||||
|
||||
const output = result?.output;
|
||||
if (Array.isArray(output)) {
|
||||
stream = output;
|
||||
await finishStage();
|
||||
} else if (output && !result?.halt && idx < pipeline.length - 1) {
|
||||
commandActive = false;
|
||||
inactiveReason = "requestInput cannot suspend from lazy output before downstream stages";
|
||||
assertRequestInputResumeConsumed(stageResume);
|
||||
stream = trackCommandOutput(
|
||||
output,
|
||||
() => {
|
||||
commandOutputStarted = true;
|
||||
},
|
||||
() => assertRequestInputResumeConsumed(stageResume),
|
||||
(err) => assertNoUnconsumedResumeAfterError(stageResume, err),
|
||||
finishStage,
|
||||
);
|
||||
} else {
|
||||
stream = output
|
||||
? trackCommandOutput(
|
||||
output,
|
||||
() => {
|
||||
commandOutputStarted = true;
|
||||
},
|
||||
() => assertRequestInputResumeConsumed(stageResume),
|
||||
(err) => assertNoUnconsumedResumeAfterError(stageResume, err),
|
||||
finishStage,
|
||||
)
|
||||
: [];
|
||||
if (!output) await finishStage();
|
||||
}
|
||||
|
||||
if (result?.halt) {
|
||||
halted = true;
|
||||
haltedAt = { index: idx, stage };
|
||||
stream = result.output ?? emptyStream();
|
||||
break;
|
||||
}
|
||||
|
||||
stream = result?.output ?? emptyStream();
|
||||
}
|
||||
|
||||
const items = [];
|
||||
for await (const item of stream) items.push(item);
|
||||
try {
|
||||
for await (const item of stream) items.push(item);
|
||||
} catch (err) {
|
||||
if (haltForInputRequest(err)) {
|
||||
items.length = 0;
|
||||
for await (const item of stream) items.push(item);
|
||||
} else {
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
assertRequestInputResumeConsumed(requestInputResume);
|
||||
|
||||
return { items, rendered, halted, haltedAt };
|
||||
|
||||
function haltForInputRequest(err: unknown) {
|
||||
if (!(err instanceof InputRequestSuspension)) return false;
|
||||
const stageIndex = err.stageIndex;
|
||||
halted = true;
|
||||
haltedAt = {
|
||||
index: stageIndex,
|
||||
stage: pipeline[stageIndex],
|
||||
inPlace: true,
|
||||
};
|
||||
stream = streamFromItems([err.request]);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
function dryRunPipeline({
|
||||
@@ -122,4 +232,67 @@ function formatStageArgs(args: Record<string, unknown>) {
|
||||
return parts.join(", ");
|
||||
}
|
||||
|
||||
async function* emptyStream() {}
|
||||
function streamFromItems(items: unknown[]) {
|
||||
return (async function* () {
|
||||
for (const item of items) yield item;
|
||||
})();
|
||||
}
|
||||
|
||||
function trackCommandOutput(
|
||||
output: AsyncIterable<unknown> | Iterable<unknown>,
|
||||
markOutput: () => void,
|
||||
assertResumeConsumed: () => void,
|
||||
assertNoUnconsumedResumeAfterError: (err: unknown) => void,
|
||||
finishStage: (options?: {
|
||||
assertResume?: boolean;
|
||||
suppressCloseErrors?: boolean;
|
||||
}) => Promise<void>,
|
||||
) {
|
||||
return (async function* () {
|
||||
let completed = false;
|
||||
try {
|
||||
for await (const item of output) {
|
||||
assertResumeConsumed();
|
||||
markOutput();
|
||||
yield item;
|
||||
}
|
||||
completed = true;
|
||||
} catch (err) {
|
||||
await finishStage({ assertResume: false, suppressCloseErrors: true });
|
||||
assertNoUnconsumedResumeAfterError(err);
|
||||
throw err;
|
||||
} finally {
|
||||
await finishStage({ assertResume: completed });
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
function assertNoUnconsumedResumeAfterError(resume: CommandInputResume | undefined, err: unknown) {
|
||||
if (err instanceof RequestInputResumeError) return;
|
||||
assertRequestInputResumeConsumed(resume);
|
||||
}
|
||||
|
||||
function trackWritableOutput(stdout: any, markOutput: () => void) {
|
||||
return new Proxy(stdout, {
|
||||
get(target, prop, receiver) {
|
||||
const value = Reflect.get(target, prop, receiver);
|
||||
if (prop === "write" || prop === "end") {
|
||||
return (...args: unknown[]) => {
|
||||
markOutput();
|
||||
return value.apply(target, args);
|
||||
};
|
||||
}
|
||||
return typeof value === "function" ? value.bind(target) : value;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function createUnsupportedRequestInput() {
|
||||
const requestInput = async function requestInput() {
|
||||
throw new Error("requestInput is not supported in this pipeline context");
|
||||
};
|
||||
requestInput.getSuspendedState = function getSuspendedState() {
|
||||
return undefined;
|
||||
};
|
||||
return requestInput;
|
||||
}
|
||||
|
||||
+123
-12
@@ -1,6 +1,9 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { runPipelineInternal } from "./runtime.js";
|
||||
import { encodeToken, decodeToken } from "./token.js";
|
||||
import { compileCached } from "../validation.js";
|
||||
import { validateCommandInputState, type CommandInputState } from "../input_request.js";
|
||||
import { deleteStateJson, readStateJson, writeStateJson } from "../state/store.js";
|
||||
|
||||
type SdkResumePayload = {
|
||||
protocolVersion: 1;
|
||||
@@ -11,6 +14,16 @@ type SdkResumePayload = {
|
||||
prompt?: string;
|
||||
inputSchema?: unknown;
|
||||
inputSubject?: unknown;
|
||||
resumeMode?: "next_stage" | "same_stage";
|
||||
stateKey?: string;
|
||||
};
|
||||
|
||||
type SdkCommandInputResumeState = {
|
||||
resumeAtIndex: number;
|
||||
items: unknown[];
|
||||
inputSchema: unknown;
|
||||
inputSubject?: unknown;
|
||||
commandInput: CommandInputState;
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -109,13 +122,30 @@ export class Lobster {
|
||||
|
||||
if (result.halted && result.items.length === 1 && result.items[0]?.type === "input_request") {
|
||||
const input = result.items[0];
|
||||
const resumeMode = input.commandInput ? "same_stage" : "next_stage";
|
||||
const resumeAtIndex =
|
||||
resumeMode === "same_stage"
|
||||
? (result.haltedAt?.index ?? -1)
|
||||
: (result.haltedAt?.index ?? -1) + 1;
|
||||
const stateKey =
|
||||
resumeMode === "same_stage"
|
||||
? await saveSdkCommandInputResumeState(this.#options, {
|
||||
resumeAtIndex,
|
||||
items: input.items ?? [],
|
||||
inputSchema: input.responseSchema,
|
||||
...(input.subject !== undefined ? { inputSubject: input.subject } : null),
|
||||
commandInput: input.commandInput,
|
||||
})
|
||||
: undefined;
|
||||
const resumeToken = encodeToken({
|
||||
protocolVersion: 1,
|
||||
v: 1,
|
||||
stageIndex: result.haltedAt?.index ?? -1,
|
||||
resumeAtIndex: (result.haltedAt?.index ?? -1) + 1,
|
||||
items: [],
|
||||
inputSchema: input.responseSchema,
|
||||
resumeAtIndex,
|
||||
resumeMode,
|
||||
...(resumeMode === "same_stage"
|
||||
? { stateKey }
|
||||
: { items: [], inputSchema: input.responseSchema }),
|
||||
inputSubject: input.subject,
|
||||
});
|
||||
|
||||
@@ -174,7 +204,15 @@ export class Lobster {
|
||||
|
||||
const payload = decodeSdkResumePayload(token);
|
||||
|
||||
let sdkCommandInputState: SdkCommandInputResumeState | undefined;
|
||||
if (payload.resumeMode === "same_stage") {
|
||||
sdkCommandInputState = await loadSdkCommandInputResumeState(this.#options, payload.stateKey!);
|
||||
}
|
||||
|
||||
if (cancel === true) {
|
||||
if (payload.resumeMode === "same_stage") {
|
||||
await deleteStateJson({ env: sdkStateEnv(this.#options), key: payload.stateKey! });
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
status: "cancelled",
|
||||
@@ -184,7 +222,7 @@ export class Lobster {
|
||||
};
|
||||
}
|
||||
|
||||
const expectsInput = payload.inputSchema !== undefined;
|
||||
const expectsInput = payload.inputSchema !== undefined || payload.resumeMode === "same_stage";
|
||||
if (expectsInput) {
|
||||
if (approved !== undefined) {
|
||||
throw new Error("resume token expects an input response, not approved");
|
||||
@@ -210,10 +248,11 @@ export class Lobster {
|
||||
}
|
||||
}
|
||||
|
||||
const resumeIndex = payload.resumeAtIndex ?? 0;
|
||||
let resumeItems = payload.items ?? [];
|
||||
const resumeIndex = sdkCommandInputState?.resumeAtIndex ?? payload.resumeAtIndex ?? 0;
|
||||
let resumeItems = sdkCommandInputState?.items ?? payload.items ?? [];
|
||||
let requestInputResume;
|
||||
if (response !== undefined) {
|
||||
const schema = payload.inputSchema;
|
||||
const schema = sdkCommandInputState?.inputSchema ?? payload.inputSchema;
|
||||
if (schema === undefined) {
|
||||
throw new Error("resume token does not support input responses");
|
||||
}
|
||||
@@ -230,7 +269,18 @@ export class Lobster {
|
||||
`response does not match schema at ${first?.instancePath || "/"}: ${first?.message || "invalid"}`,
|
||||
);
|
||||
}
|
||||
resumeItems = [response];
|
||||
if (payload.resumeMode === "same_stage") {
|
||||
resumeItems = sdkCommandInputState!.items;
|
||||
requestInputResume = {
|
||||
state: sdkCommandInputState!.commandInput,
|
||||
response,
|
||||
onConsumed: async () => {
|
||||
await deleteStateJson({ env: sdkStateEnv(this.#options), key: payload.stateKey! });
|
||||
},
|
||||
};
|
||||
} else {
|
||||
resumeItems = [response];
|
||||
}
|
||||
}
|
||||
|
||||
const remainingStages = this.#stages.slice(resumeIndex);
|
||||
@@ -245,6 +295,7 @@ export class Lobster {
|
||||
stages: remainingStages,
|
||||
ctx,
|
||||
input: resumeItems,
|
||||
requestInputResume,
|
||||
});
|
||||
|
||||
if (
|
||||
@@ -277,13 +328,27 @@ export class Lobster {
|
||||
|
||||
if (result.halted && result.items.length === 1 && result.items[0]?.type === "input_request") {
|
||||
const input = result.items[0];
|
||||
const inputStageIndex = resumeIndex + (result.haltedAt?.index ?? 0);
|
||||
const resumeMode = input.commandInput ? "same_stage" : "next_stage";
|
||||
const stateKey =
|
||||
resumeMode === "same_stage"
|
||||
? await saveSdkCommandInputResumeState(this.#options, {
|
||||
resumeAtIndex: inputStageIndex,
|
||||
items: input.items ?? [],
|
||||
inputSchema: input.responseSchema,
|
||||
...(input.subject !== undefined ? { inputSubject: input.subject } : null),
|
||||
commandInput: input.commandInput,
|
||||
})
|
||||
: undefined;
|
||||
const resumeToken = encodeToken({
|
||||
protocolVersion: 1,
|
||||
v: 1,
|
||||
stageIndex: resumeIndex + (result.haltedAt?.index ?? 0),
|
||||
resumeAtIndex: resumeIndex + (result.haltedAt?.index ?? 0) + 1,
|
||||
items: [],
|
||||
inputSchema: input.responseSchema,
|
||||
stageIndex: inputStageIndex,
|
||||
resumeAtIndex: resumeMode === "same_stage" ? inputStageIndex : inputStageIndex + 1,
|
||||
resumeMode,
|
||||
...(resumeMode === "same_stage"
|
||||
? { stateKey }
|
||||
: { items: [], inputSchema: input.responseSchema }),
|
||||
inputSubject: input.subject,
|
||||
});
|
||||
|
||||
@@ -351,5 +416,51 @@ function decodeSdkResumePayload(token: string): SdkResumePayload {
|
||||
if (data.items !== undefined && !Array.isArray(data.items)) {
|
||||
throw new Error("Invalid token");
|
||||
}
|
||||
if (
|
||||
data.resumeMode !== undefined &&
|
||||
(typeof data.resumeMode !== "string" || !["next_stage", "same_stage"].includes(data.resumeMode))
|
||||
) {
|
||||
throw new Error("Invalid token");
|
||||
}
|
||||
if (data.resumeMode === "same_stage") {
|
||||
if (typeof data.stateKey !== "string" || data.stateKey.length === 0) {
|
||||
throw new Error("Invalid token");
|
||||
}
|
||||
} else if (data.stateKey !== undefined) {
|
||||
throw new Error("Invalid token");
|
||||
}
|
||||
if (data.commandInput !== undefined) throw new Error("Invalid token");
|
||||
return data as unknown as SdkResumePayload;
|
||||
}
|
||||
|
||||
function sdkStateEnv(options: any) {
|
||||
return options.stateDir
|
||||
? { ...(options.env ?? process.env), LOBSTER_STATE_DIR: options.stateDir }
|
||||
: (options.env ?? process.env);
|
||||
}
|
||||
|
||||
async function saveSdkCommandInputResumeState(options: any, state: SdkCommandInputResumeState) {
|
||||
const stateKey = `sdk_resume_${randomUUID()}`;
|
||||
await writeStateJson({ env: sdkStateEnv(options), key: stateKey, value: state });
|
||||
return stateKey;
|
||||
}
|
||||
|
||||
async function loadSdkCommandInputResumeState(
|
||||
options: any,
|
||||
stateKey: string,
|
||||
): Promise<SdkCommandInputResumeState> {
|
||||
const stored = await readStateJson({ env: sdkStateEnv(options), key: stateKey });
|
||||
if (!stored || typeof stored !== "object") throw new Error("SDK resume state not found");
|
||||
const data = stored as Partial<SdkCommandInputResumeState>;
|
||||
if (
|
||||
typeof data.resumeAtIndex !== "number" ||
|
||||
!Number.isInteger(data.resumeAtIndex) ||
|
||||
data.resumeAtIndex < 0
|
||||
) {
|
||||
throw new Error("Invalid SDK resume state");
|
||||
}
|
||||
if (!Array.isArray(data.items)) throw new Error("Invalid SDK resume state");
|
||||
if (data.inputSchema === undefined) throw new Error("Invalid SDK resume state");
|
||||
data.commandInput = validateCommandInputState(data.commandInput);
|
||||
return data as SdkCommandInputResumeState;
|
||||
}
|
||||
|
||||
+96
-99
@@ -1,9 +1,12 @@
|
||||
/**
|
||||
* SDK Runtime - Executes Lobster pipelines
|
||||
*
|
||||
* This is adapted from the CLI runtime but designed for SDK use.
|
||||
* This adapts SDK stages to the core runtime so command-level suspension rules
|
||||
* stay identical across CLI, tool mode, and SDK entry points.
|
||||
*/
|
||||
|
||||
import { runPipeline as runCorePipeline } from "../runtime.js";
|
||||
|
||||
/**
|
||||
* @typedef {Object} StageResult
|
||||
* @property {AsyncIterable|any[]} [output] - Output items
|
||||
@@ -18,39 +21,6 @@
|
||||
* @property {Object|null} haltedAt - Stage where halt occurred
|
||||
*/
|
||||
|
||||
/**
|
||||
* Convert various inputs to an async iterable
|
||||
* @param {any} input
|
||||
* @returns {AsyncIterable}
|
||||
*/
|
||||
async function* toAsyncIterable(input) {
|
||||
if (input === null || input === undefined) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (Array.isArray(input)) {
|
||||
for (const item of input) {
|
||||
yield item;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (typeof input[Symbol.asyncIterator] === "function") {
|
||||
yield* input;
|
||||
return;
|
||||
}
|
||||
|
||||
if (typeof input[Symbol.iterator] === "function") {
|
||||
for (const item of input) {
|
||||
yield item;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Single item
|
||||
yield input;
|
||||
}
|
||||
|
||||
/**
|
||||
* Collect async iterable to array
|
||||
* @param {AsyncIterable} iterable
|
||||
@@ -64,6 +34,29 @@ async function collectItems(iterable) {
|
||||
return items;
|
||||
}
|
||||
|
||||
function normalizeSdkOutput(output) {
|
||||
if (output === null || output === undefined) return [];
|
||||
if (Array.isArray(output)) return output;
|
||||
if (
|
||||
typeof output?.[Symbol.asyncIterator] === "function" ||
|
||||
typeof output?.[Symbol.iterator] === "function"
|
||||
) {
|
||||
return output;
|
||||
}
|
||||
return [output];
|
||||
}
|
||||
|
||||
function createNullWritable() {
|
||||
return {
|
||||
write() {
|
||||
return true;
|
||||
},
|
||||
end() {
|
||||
return undefined;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a pipeline of stages
|
||||
*
|
||||
@@ -73,53 +66,72 @@ async function collectItems(iterable) {
|
||||
* @param {any[]} [options.input] - Initial input items
|
||||
* @returns {Promise<PipelineResult>}
|
||||
*/
|
||||
export async function runPipelineInternal({ stages, ctx, input = [] }) {
|
||||
let stream = toAsyncIterable(input);
|
||||
let halted = false;
|
||||
let haltedAt = null;
|
||||
export async function runPipelineInternal({
|
||||
stages,
|
||||
ctx,
|
||||
input = [],
|
||||
requestInputResume = undefined,
|
||||
}) {
|
||||
const runtimeCtx = ctx ?? {};
|
||||
const pipeline = stages.map((_stage, index) => ({
|
||||
name: `sdk.stage.${index}`,
|
||||
args: {},
|
||||
raw: `sdk.stage.${index}`,
|
||||
}));
|
||||
const commands = new Map(
|
||||
stages.map((stage, index) => [
|
||||
`sdk.stage.${index}`,
|
||||
{
|
||||
async run({ input, ctx }) {
|
||||
const stageCtx = { ...runtimeCtx, ...ctx };
|
||||
if (typeof stage === "function") {
|
||||
const isGenerator =
|
||||
stage.constructor?.name === "AsyncGeneratorFunction" ||
|
||||
stage.constructor?.name === "GeneratorFunction";
|
||||
|
||||
for (let idx = 0; idx < stages.length; idx++) {
|
||||
const stage = stages[idx];
|
||||
if (isGenerator) {
|
||||
return { output: normalizeSdkOutput(stage(input, stageCtx)) };
|
||||
}
|
||||
|
||||
let result;
|
||||
const items = await collectItems(input);
|
||||
return { output: normalizeSdkOutput(await stage(items, stageCtx)) };
|
||||
}
|
||||
|
||||
if (typeof stage === "function") {
|
||||
// Check if it's a generator function
|
||||
const isGenerator =
|
||||
stage.constructor?.name === "AsyncGeneratorFunction" ||
|
||||
stage.constructor?.name === "GeneratorFunction";
|
||||
if (typeof stage?.run === "function") {
|
||||
const result = await stage.run({ input, ctx: stageCtx });
|
||||
return result && "output" in result
|
||||
? { ...result, output: normalizeSdkOutput(result.output) }
|
||||
: result;
|
||||
}
|
||||
|
||||
if (isGenerator) {
|
||||
// Generator function - pass the stream directly
|
||||
result = { output: stage(stream, ctx) };
|
||||
} else {
|
||||
// Regular function - collect items first, then call
|
||||
const items = await collectItems(stream);
|
||||
const output = await stage(items, ctx);
|
||||
result = { output: toAsyncIterable(output) };
|
||||
}
|
||||
} else if (typeof stage?.run === "function") {
|
||||
// Stage object with run method (primitives)
|
||||
result = await stage.run({ input: stream, ctx });
|
||||
} else {
|
||||
throw new Error(`Invalid stage at index ${idx}: must be a function or have run() method`);
|
||||
}
|
||||
throw new Error(
|
||||
`Invalid stage at index ${index}: must be a function or have run() method`,
|
||||
);
|
||||
},
|
||||
},
|
||||
]),
|
||||
);
|
||||
const stdout = runtimeCtx.stdout ?? createNullWritable();
|
||||
const stderr = runtimeCtx.stderr ?? createNullWritable();
|
||||
|
||||
// Handle halt
|
||||
if (result?.halt) {
|
||||
halted = true;
|
||||
haltedAt = { index: idx, stage };
|
||||
stream = result.output ?? toAsyncIterable([]);
|
||||
break;
|
||||
}
|
||||
|
||||
stream = result?.output ?? toAsyncIterable([]);
|
||||
}
|
||||
|
||||
// Collect final output
|
||||
const items = await collectItems(stream);
|
||||
|
||||
return { items, halted, haltedAt };
|
||||
return runCorePipeline({
|
||||
pipeline,
|
||||
registry: {
|
||||
get(name) {
|
||||
return commands.get(name);
|
||||
},
|
||||
},
|
||||
stdin: runtimeCtx.stdin ?? { isTTY: false },
|
||||
stdout,
|
||||
stderr,
|
||||
env: runtimeCtx.env ?? process.env,
|
||||
mode: runtimeCtx.mode ?? "sdk",
|
||||
cwd: runtimeCtx.cwd,
|
||||
llmAdapters: runtimeCtx.llmAdapters,
|
||||
signal: runtimeCtx.signal,
|
||||
input: normalizeSdkOutput(input),
|
||||
requestInputResume,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -134,34 +146,19 @@ export async function runPipeline({
|
||||
env,
|
||||
mode = "human",
|
||||
input,
|
||||
requestInputResume = undefined,
|
||||
requestInputEnabled = true,
|
||||
}) {
|
||||
// This wraps the CLI-style pipeline execution
|
||||
// Convert pipeline stages to functions using registry
|
||||
|
||||
const stages = pipeline.map((stage) => {
|
||||
const command = registry.get(stage.name);
|
||||
if (!command) {
|
||||
throw new Error(`Unknown command: ${stage.name}`);
|
||||
}
|
||||
|
||||
return {
|
||||
run: async ({ input, ctx }) => {
|
||||
return command.run({ input, args: stage.args, ctx });
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const ctx = {
|
||||
return runCorePipeline({
|
||||
pipeline,
|
||||
registry,
|
||||
stdin,
|
||||
stdout,
|
||||
stderr,
|
||||
env,
|
||||
mode,
|
||||
};
|
||||
|
||||
return runPipelineInternal({
|
||||
stages,
|
||||
ctx,
|
||||
input: input ? await collectItems(input) : [],
|
||||
input,
|
||||
requestInputResume,
|
||||
requestInputEnabled,
|
||||
});
|
||||
}
|
||||
|
||||
+264
-25
@@ -22,6 +22,11 @@ import { CostTracker } from "../core/cost_tracker.js";
|
||||
import type { CostLimit, CostSummary } from "../core/cost_tracker.js";
|
||||
import { withRetry, resolveRetryConfig } from "../core/retry.js";
|
||||
import type { RetryConfig } from "../core/retry.js";
|
||||
import {
|
||||
RequestInputResumeError,
|
||||
validateCommandInputState,
|
||||
type CommandInputState,
|
||||
} from "../input_request.js";
|
||||
|
||||
export type WorkflowFile = {
|
||||
name?: string;
|
||||
@@ -178,8 +183,10 @@ export type WorkflowResumePayload = {
|
||||
approvalStepId?: string;
|
||||
approvalIdentity?: WorkflowApprovalIdentity;
|
||||
inputStepId?: string;
|
||||
inputKind?: "workflow_step" | "pipeline_command";
|
||||
inputSchema?: unknown;
|
||||
inputSubject?: unknown;
|
||||
pipelineInput?: WorkflowPipelineInputResumeState;
|
||||
};
|
||||
|
||||
type WorkflowResumeState = {
|
||||
@@ -190,11 +197,20 @@ type WorkflowResumeState = {
|
||||
approvalStepId?: string;
|
||||
approvalIdentity?: WorkflowApprovalIdentity;
|
||||
inputStepId?: string;
|
||||
inputKind?: "workflow_step" | "pipeline_command";
|
||||
inputSchema?: unknown;
|
||||
inputSubject?: unknown;
|
||||
pipelineInput?: WorkflowPipelineInputResumeState;
|
||||
createdAt: string;
|
||||
};
|
||||
|
||||
type WorkflowPipelineInputResumeState = {
|
||||
pipeline: Array<{ name: string; args: Record<string, unknown>; raw: string }>;
|
||||
resumeAtIndex: number;
|
||||
items: unknown[];
|
||||
commandInput: CommandInputState;
|
||||
};
|
||||
|
||||
export class WorkflowResumeArgumentError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
@@ -202,6 +218,33 @@ export class WorkflowResumeArgumentError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
class WorkflowPipelineInputSuspension extends Error {
|
||||
stepId: string;
|
||||
request: {
|
||||
prompt: string;
|
||||
responseSchema: unknown;
|
||||
defaults?: unknown;
|
||||
subject?: unknown;
|
||||
};
|
||||
pipelineInput: WorkflowPipelineInputResumeState;
|
||||
|
||||
constructor({
|
||||
stepId,
|
||||
request,
|
||||
pipelineInput,
|
||||
}: {
|
||||
stepId: string;
|
||||
request: WorkflowPipelineInputSuspension["request"];
|
||||
pipelineInput: WorkflowPipelineInputResumeState;
|
||||
}) {
|
||||
super(`Workflow step ${stepId} pipeline requested input`);
|
||||
this.name = "WorkflowPipelineInputSuspension";
|
||||
this.stepId = stepId;
|
||||
this.request = request;
|
||||
this.pipelineInput = pipelineInput;
|
||||
}
|
||||
}
|
||||
|
||||
export async function loadWorkflowFile(filePath: string): Promise<WorkflowFile> {
|
||||
const text = await fsp.readFile(filePath, "utf8");
|
||||
const ext = path.extname(filePath).toLowerCase();
|
||||
@@ -726,6 +769,13 @@ export async function runWorkflowFile({
|
||||
results[resumeState.approvalStepId] = previous;
|
||||
}
|
||||
|
||||
let resumedPipelineInput: {
|
||||
stepId: string;
|
||||
response: unknown;
|
||||
pipelineInput: WorkflowPipelineInputResumeState;
|
||||
onConsumed?: () => Promise<void>;
|
||||
} | null = null;
|
||||
|
||||
if (resumeState?.inputStepId) {
|
||||
if (approved !== undefined) {
|
||||
throw new WorkflowResumeArgumentError(
|
||||
@@ -737,24 +787,61 @@ export async function runWorkflowFile({
|
||||
"Workflow resume requires --response-json for input requests",
|
||||
);
|
||||
}
|
||||
const inputStep = steps[stepIndexById.get(resumeState.inputStepId) ?? -1];
|
||||
if (!inputStep || !isInputStep(inputStep.input)) {
|
||||
throw new Error(`Invalid input step in resume state: ${resumeState.inputStepId}`);
|
||||
}
|
||||
try {
|
||||
validateInputResponse({
|
||||
schema: resumeState.inputSchema ?? inputStep.input.responseSchema,
|
||||
if (resumeState.inputKind === "pipeline_command") {
|
||||
const resumedStepIndex = stepIndexById.get(resumeState.inputStepId);
|
||||
if (resumedStepIndex !== startIndex) {
|
||||
throw new RequestInputResumeError("workflow input step changed since input request");
|
||||
}
|
||||
const pipelineStep = steps[resumedStepIndex];
|
||||
if (!pipelineStep || typeof pipelineStep.pipeline !== "string") {
|
||||
throw new Error(
|
||||
`Invalid pipeline input step in resume state: ${resumeState.inputStepId}`,
|
||||
);
|
||||
}
|
||||
if (!evaluateCondition(pipelineStep.when ?? pipelineStep.condition, results)) {
|
||||
throw new RequestInputResumeError(
|
||||
"workflow input step condition changed since input request",
|
||||
);
|
||||
}
|
||||
try {
|
||||
validateInputResponse({
|
||||
schema: resumeState.inputSchema,
|
||||
response,
|
||||
stepId: pipelineStep.id,
|
||||
});
|
||||
} catch (err: any) {
|
||||
throw new WorkflowResumeArgumentError(err?.message ?? String(err));
|
||||
}
|
||||
resumedPipelineInput = {
|
||||
stepId: resumeState.inputStepId,
|
||||
response,
|
||||
stepId: inputStep.id,
|
||||
});
|
||||
} catch (err: any) {
|
||||
throw new WorkflowResumeArgumentError(err?.message ?? String(err));
|
||||
pipelineInput: resumeState.pipelineInput!,
|
||||
onConsumed: consumedResumeStateKey
|
||||
? async () => {
|
||||
await deleteStateJson({ env: ctx.env, key: consumedResumeStateKey });
|
||||
}
|
||||
: undefined,
|
||||
};
|
||||
} else {
|
||||
const inputStep = steps[stepIndexById.get(resumeState.inputStepId) ?? -1];
|
||||
if (!inputStep || !isInputStep(inputStep.input)) {
|
||||
throw new Error(`Invalid input step in resume state: ${resumeState.inputStepId}`);
|
||||
}
|
||||
try {
|
||||
validateInputResponse({
|
||||
schema: resumeState.inputSchema ?? inputStep.input.responseSchema,
|
||||
response,
|
||||
stepId: inputStep.id,
|
||||
});
|
||||
} catch (err: any) {
|
||||
throw new WorkflowResumeArgumentError(err?.message ?? String(err));
|
||||
}
|
||||
const previous = results[resumeState.inputStepId] ?? { id: resumeState.inputStepId };
|
||||
previous.subject = resumeState.inputSubject ?? null;
|
||||
previous.response = response;
|
||||
delete previous.skipped;
|
||||
results[resumeState.inputStepId] = previous;
|
||||
}
|
||||
const previous = results[resumeState.inputStepId] ?? { id: resumeState.inputStepId };
|
||||
previous.subject = resumeState.inputSubject ?? null;
|
||||
previous.response = response;
|
||||
delete previous.skipped;
|
||||
results[resumeState.inputStepId] = previous;
|
||||
}
|
||||
|
||||
if (ctx.dryRun) {
|
||||
@@ -925,6 +1012,7 @@ export async function runWorkflowFile({
|
||||
ctx,
|
||||
env: subEnv,
|
||||
cwd: subCwd,
|
||||
requestInputEnabled: false,
|
||||
});
|
||||
} else {
|
||||
const inputValue = resolveInputValue(subStep.stdin, resolvedArgs, scopedResults);
|
||||
@@ -1053,6 +1141,7 @@ export async function runWorkflowFile({
|
||||
ctx: { ...ctx, signal: branchSignal },
|
||||
env: branchEnv,
|
||||
cwd: branchCwd,
|
||||
requestInputEnabled: false,
|
||||
});
|
||||
return { branchId: branch.id, result: branchResult };
|
||||
}
|
||||
@@ -1196,6 +1285,14 @@ export async function runWorkflowFile({
|
||||
ctx: { ...ctx, signal: stepSignal },
|
||||
env,
|
||||
cwd,
|
||||
resume:
|
||||
resumedPipelineInput?.stepId === step.id
|
||||
? {
|
||||
pipelineInput: resumedPipelineInput.pipelineInput,
|
||||
response: resumedPipelineInput.response,
|
||||
onConsumed: resumedPipelineInput.onConsumed,
|
||||
}
|
||||
: undefined,
|
||||
});
|
||||
} else {
|
||||
const inputValue = resolveInputValue(step.stdin, resolvedArgs, results);
|
||||
@@ -1216,6 +1313,12 @@ export async function runWorkflowFile({
|
||||
? await withRetry(executeStepAttempt, retryConfig, {
|
||||
signal: ctx.signal,
|
||||
shouldRetry: (error) => {
|
||||
if (
|
||||
error instanceof WorkflowPipelineInputSuspension ||
|
||||
error instanceof RequestInputResumeError
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
const message = error?.message ?? String(error);
|
||||
return !/halted (for approval inside|before completion at) pipeline/.test(
|
||||
message,
|
||||
@@ -1231,6 +1334,51 @@ export async function runWorkflowFile({
|
||||
result = attemptResult.result;
|
||||
parallelBranchResults = attemptResult.parallelBranchResults;
|
||||
} catch (err: any) {
|
||||
if (err instanceof WorkflowPipelineInputSuspension) {
|
||||
const inputRequest = buildNeedsInputRequest({
|
||||
stepId: err.stepId,
|
||||
prompt: err.request.prompt,
|
||||
responseSchema: err.request.responseSchema,
|
||||
defaults: err.request.defaults,
|
||||
subject: err.request.subject,
|
||||
maxEnvelopeBytes: resolveToolEnvelopeMaxBytes(ctx.env),
|
||||
});
|
||||
const stateKey = await saveWorkflowResumeState(ctx.env, {
|
||||
filePath: resolvedFilePath,
|
||||
resumeAtIndex: idx,
|
||||
steps: results,
|
||||
args: resolvedArgs,
|
||||
inputStepId: err.stepId,
|
||||
inputKind: "pipeline_command",
|
||||
inputSchema: err.request.responseSchema,
|
||||
inputSubject: err.request.subject,
|
||||
pipelineInput: err.pipelineInput,
|
||||
createdAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
if (consumedResumeStateKey && consumedResumeStateKey !== stateKey) {
|
||||
await deleteStateJson({ env: ctx.env, key: consumedResumeStateKey });
|
||||
}
|
||||
|
||||
const resumeToken = encodeToken({
|
||||
protocolVersion: 1,
|
||||
v: 1,
|
||||
kind: "workflow-file",
|
||||
stateKey,
|
||||
} satisfies WorkflowResumePayload);
|
||||
|
||||
return {
|
||||
status: "needs_input",
|
||||
output: [],
|
||||
requiresInput: {
|
||||
...inputRequest,
|
||||
resumeToken,
|
||||
},
|
||||
};
|
||||
}
|
||||
if (err instanceof RequestInputResumeError) {
|
||||
throw err;
|
||||
}
|
||||
if (ctx.signal?.aborted && (err?.name === "AbortError" || err?.code === "ABORT_ERR")) {
|
||||
throw err;
|
||||
}
|
||||
@@ -1716,9 +1864,56 @@ async function loadWorkflowResumeState(env: Record<string, string | undefined>,
|
||||
if (!data.steps || typeof data.steps !== "object")
|
||||
throw new Error("Invalid workflow resume state");
|
||||
if (!data.args || typeof data.args !== "object") throw new Error("Invalid workflow resume state");
|
||||
if (
|
||||
data.inputKind !== undefined &&
|
||||
!["workflow_step", "pipeline_command"].includes(data.inputKind)
|
||||
) {
|
||||
throw new Error("Invalid workflow resume state");
|
||||
}
|
||||
if (data.inputKind === "pipeline_command") {
|
||||
if (typeof data.inputStepId !== "string") throw new Error("Invalid workflow resume state");
|
||||
if (data.inputSchema === undefined) throw new Error("Invalid workflow resume state");
|
||||
data.pipelineInput = validateWorkflowPipelineInputResumeState(data.pipelineInput);
|
||||
} else if (data.pipelineInput !== undefined) {
|
||||
throw new Error("Invalid workflow resume state");
|
||||
}
|
||||
return data as WorkflowResumeState;
|
||||
}
|
||||
|
||||
function validateWorkflowPipelineInputResumeState(
|
||||
value: unknown,
|
||||
): WorkflowPipelineInputResumeState {
|
||||
if (!value || typeof value !== "object") throw new Error("Invalid workflow resume state");
|
||||
const data = value as Partial<WorkflowPipelineInputResumeState>;
|
||||
if (!Array.isArray(data.pipeline)) throw new Error("Invalid workflow resume state");
|
||||
validateWorkflowPipelineShape(data.pipeline);
|
||||
if (
|
||||
typeof data.resumeAtIndex !== "number" ||
|
||||
!Number.isInteger(data.resumeAtIndex) ||
|
||||
data.resumeAtIndex < 0 ||
|
||||
data.resumeAtIndex >= data.pipeline.length
|
||||
) {
|
||||
throw new Error("Invalid workflow resume state");
|
||||
}
|
||||
if (!Array.isArray(data.items)) throw new Error("Invalid workflow resume state");
|
||||
data.commandInput = validateCommandInputState(data.commandInput);
|
||||
return data as WorkflowPipelineInputResumeState;
|
||||
}
|
||||
|
||||
function validateWorkflowPipelineShape(pipeline: unknown[]) {
|
||||
for (const stage of pipeline) {
|
||||
if (!stage || typeof stage !== "object") throw new Error("Invalid workflow resume state");
|
||||
const data = stage as Record<string, unknown>;
|
||||
if (typeof data.name !== "string" || data.name.length === 0) {
|
||||
throw new Error("Invalid workflow resume state");
|
||||
}
|
||||
if (!data.args || typeof data.args !== "object" || Array.isArray(data.args)) {
|
||||
throw new Error("Invalid workflow resume state");
|
||||
}
|
||||
if (typeof data.raw !== "string") throw new Error("Invalid workflow resume state");
|
||||
}
|
||||
}
|
||||
|
||||
function mergeEnv(
|
||||
base: Record<string, string | undefined>,
|
||||
workflowEnv: WorkflowFile["env"],
|
||||
@@ -2652,6 +2847,8 @@ async function runPipelineStep({
|
||||
ctx,
|
||||
env,
|
||||
cwd,
|
||||
resume,
|
||||
requestInputEnabled = true,
|
||||
}: {
|
||||
stepId: string;
|
||||
pipelineText: string;
|
||||
@@ -2659,11 +2856,26 @@ async function runPipelineStep({
|
||||
ctx: RunContext;
|
||||
env: Record<string, string | undefined>;
|
||||
cwd?: string;
|
||||
resume?: {
|
||||
pipelineInput: WorkflowPipelineInputResumeState;
|
||||
response: unknown;
|
||||
onConsumed?: () => Promise<void>;
|
||||
};
|
||||
requestInputEnabled?: boolean;
|
||||
}) {
|
||||
let pipeline;
|
||||
try {
|
||||
pipeline = parsePipeline(pipelineText);
|
||||
const currentPipeline = parsePipeline(pipelineText);
|
||||
if (resume) {
|
||||
if (!isDeepStrictEqual(currentPipeline, resume.pipelineInput.pipeline)) {
|
||||
throw new RequestInputResumeError("workflow pipeline changed since input request");
|
||||
}
|
||||
pipeline = resume.pipelineInput.pipeline;
|
||||
} else {
|
||||
pipeline = currentPipeline;
|
||||
}
|
||||
} catch (err: any) {
|
||||
if (err instanceof RequestInputResumeError) throw err;
|
||||
throw new Error(
|
||||
`Workflow step ${stepId} pipeline parse failed: ${err?.message ?? String(err)}`,
|
||||
);
|
||||
@@ -2676,8 +2888,10 @@ async function runPipelineStep({
|
||||
renderedStdout += String(chunk);
|
||||
});
|
||||
|
||||
const pipelineStartIndex = resume ? resume.pipelineInput.resumeAtIndex : 0;
|
||||
const remainingPipeline = pipeline.slice(pipelineStartIndex);
|
||||
const result = await runPipeline({
|
||||
pipeline,
|
||||
pipeline: remainingPipeline,
|
||||
registry: ctx.registry,
|
||||
stdin: ctx.stdin,
|
||||
stdout,
|
||||
@@ -2687,7 +2901,15 @@ async function runPipelineStep({
|
||||
cwd,
|
||||
signal: ctx.signal,
|
||||
llmAdapters: ctx.llmAdapters,
|
||||
input: inputValueToStream(inputValue),
|
||||
input: resume ? resume.pipelineInput.items : inputValueToPipelineItems(inputValue),
|
||||
requestInputEnabled,
|
||||
requestInputResume: resume
|
||||
? {
|
||||
state: resume.pipelineInput.commandInput,
|
||||
response: resume.response,
|
||||
onConsumed: resume.onConsumed,
|
||||
}
|
||||
: undefined,
|
||||
});
|
||||
stdout.end();
|
||||
|
||||
@@ -2698,6 +2920,27 @@ async function runPipelineStep({
|
||||
`Workflow step ${stepId} halted for approval inside pipeline stage ${haltedName}. Use a separate approval step in the workflow file.`,
|
||||
);
|
||||
}
|
||||
const request =
|
||||
result.items.length === 1 && result.items[0]?.type === "input_request"
|
||||
? (result.items[0] as Record<string, any>)
|
||||
: null;
|
||||
if (request?.commandInput) {
|
||||
throw new WorkflowPipelineInputSuspension({
|
||||
stepId,
|
||||
request: {
|
||||
prompt: String(request.prompt),
|
||||
responseSchema: request.responseSchema,
|
||||
...(request.defaults !== undefined ? { defaults: request.defaults } : null),
|
||||
...(request.subject !== undefined ? { subject: request.subject } : null),
|
||||
},
|
||||
pipelineInput: {
|
||||
pipeline,
|
||||
resumeAtIndex: pipelineStartIndex + (result.haltedAt?.index ?? 0),
|
||||
items: Array.isArray(request.items) ? request.items : [],
|
||||
commandInput: request.commandInput,
|
||||
},
|
||||
});
|
||||
}
|
||||
throw new Error(
|
||||
`Workflow step ${stepId} halted before completion at pipeline stage ${haltedName}`,
|
||||
);
|
||||
@@ -2772,12 +3015,8 @@ function* inputValueToItems(value: unknown) {
|
||||
yield value;
|
||||
}
|
||||
|
||||
function inputValueToStream(value: unknown) {
|
||||
return (async function* () {
|
||||
for (const item of inputValueToItems(value)) {
|
||||
yield item;
|
||||
}
|
||||
})();
|
||||
function inputValueToPipelineItems(value: unknown) {
|
||||
return [...inputValueToItems(value)];
|
||||
}
|
||||
|
||||
function serializePipelineItemsToStdout(items: unknown[]) {
|
||||
|
||||
@@ -83,6 +83,28 @@ test("for_each supports custom item_var and index_var", async () => {
|
||||
]);
|
||||
});
|
||||
|
||||
test("for_each pipeline sub-steps reject command-level requestInput", async () => {
|
||||
await assert.rejects(
|
||||
() =>
|
||||
runWorkflow({
|
||||
steps: [
|
||||
{ id: "vals", command: 'node -e "process.stdout.write(JSON.stringify([1]))"' },
|
||||
{
|
||||
id: "loop",
|
||||
for_each: "$vals.json",
|
||||
steps: [
|
||||
{
|
||||
id: "review",
|
||||
pipeline: "ask --prompt 'Review?'",
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
}),
|
||||
/requestInput is not supported in this pipeline context/,
|
||||
);
|
||||
});
|
||||
|
||||
test("for_each throws when source is not an array", async () => {
|
||||
await assert.rejects(
|
||||
() =>
|
||||
|
||||
@@ -122,6 +122,24 @@ test("parallel wait=all propagates branch failure", async () => {
|
||||
);
|
||||
});
|
||||
|
||||
test("parallel pipeline branches reject command-level requestInput", async () => {
|
||||
await assert.rejects(
|
||||
() =>
|
||||
runWorkflow({
|
||||
steps: [
|
||||
{
|
||||
id: "p",
|
||||
parallel: {
|
||||
wait: "all",
|
||||
branches: [{ id: "review", pipeline: "ask --prompt 'Review?'" }],
|
||||
},
|
||||
},
|
||||
],
|
||||
}),
|
||||
/requestInput is not supported in this pipeline context/,
|
||||
);
|
||||
});
|
||||
|
||||
test("parallel validation rejects empty branches", async () => {
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-parallel-"));
|
||||
const filePath = path.join(tmpDir, "bad.lobster");
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,7 +1,12 @@
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { promises as fsp } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import { Lobster } from "../src/sdk/Lobster.js";
|
||||
import { stateSet } from "../src/sdk/primitives/state.js";
|
||||
import { decodeToken, encodeToken } from "../src/token.js";
|
||||
|
||||
test("sdk Lobster.resume accepts structured input responses", async () => {
|
||||
const workflow = new Lobster().pipe({
|
||||
@@ -65,3 +70,137 @@ test("sdk Lobster.resume rejects invalid structured input responses", async () =
|
||||
/does not match schema/i,
|
||||
);
|
||||
});
|
||||
|
||||
test("sdk Lobster.resume replays command-level requestInput with a fresh instance", async () => {
|
||||
const stateDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-sdk-input-"));
|
||||
let runs = 0;
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { choice: { type: "string", enum: ["red", "blue"] } },
|
||||
required: ["choice"],
|
||||
};
|
||||
const choose = {
|
||||
async run({ input, ctx }: any) {
|
||||
runs += 1;
|
||||
const items = [];
|
||||
for await (const item of input) items.push(item);
|
||||
const response = await ctx.requestInput({
|
||||
prompt: "Pick a color",
|
||||
responseSchema: schema,
|
||||
suspendedState: { count: items.length },
|
||||
});
|
||||
return { output: [{ runs, items, choice: response.choice }] };
|
||||
},
|
||||
};
|
||||
const createWorkflow = () => new Lobster({ stateDir }).pipe(choose);
|
||||
|
||||
const first = await createWorkflow().run([{ id: 1 }]);
|
||||
assert.equal(first.ok, true);
|
||||
assert.equal(first.status, "needs_input");
|
||||
assert.ok(first.requiresInput?.resumeToken);
|
||||
|
||||
const payload = decodeToken(first.requiresInput.resumeToken) as any;
|
||||
assert.equal(payload.resumeMode, "same_stage");
|
||||
assert.equal(payload.resumeAtIndex, 0);
|
||||
assert.match(payload.stateKey, /^sdk_resume_/);
|
||||
assert.equal(payload.commandInput, undefined);
|
||||
|
||||
const resumed = await createWorkflow().resume(first.requiresInput.resumeToken, {
|
||||
response: { choice: "red" },
|
||||
});
|
||||
assert.equal(resumed.ok, true);
|
||||
assert.equal(resumed.status, "ok");
|
||||
assert.deepEqual(resumed.output, [{ runs: 2, items: [{ id: 1 }], choice: "red" }]);
|
||||
});
|
||||
|
||||
test("sdk Lobster.resume rejects command-level requestInput tampering and metadata drift", async () => {
|
||||
const stateDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-sdk-input-tamper-"));
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { choice: { type: "string" } },
|
||||
required: ["choice"],
|
||||
};
|
||||
let prompt = "Original";
|
||||
const workflow = new Lobster({ stateDir }).pipe({
|
||||
async run({ ctx }: any) {
|
||||
const response = await ctx.requestInput({ prompt, responseSchema: schema });
|
||||
return { output: [response] };
|
||||
},
|
||||
});
|
||||
|
||||
const first = await workflow.run();
|
||||
assert.equal(first.status, "needs_input");
|
||||
const payload = decodeToken(first.requiresInput!.resumeToken) as any;
|
||||
payload.stateKey = "sdk_resume_forged";
|
||||
const forged = encodeToken(payload);
|
||||
|
||||
await assert.rejects(
|
||||
() => workflow.resume(forged, { response: { choice: "red" } }),
|
||||
/SDK resume state not found/,
|
||||
);
|
||||
|
||||
prompt = "Changed";
|
||||
const resumed = await workflow.resume(first.requiresInput!.resumeToken, {
|
||||
response: { choice: "red" },
|
||||
});
|
||||
assert.equal(resumed.ok, false);
|
||||
assert.equal(resumed.status, "error");
|
||||
assert.match(resumed.error?.message ?? "", /does not match suspended request/);
|
||||
});
|
||||
|
||||
test("sdk Lobster stores chained requestInput history outside the resume token", async () => {
|
||||
const stateDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-sdk-input-chain-"));
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { choice: { type: "string", enum: ["red", "blue"] } },
|
||||
required: ["choice"],
|
||||
};
|
||||
const createWorkflow = () =>
|
||||
new Lobster({ stateDir }).pipe({
|
||||
async run({ ctx }: any) {
|
||||
const first = await ctx.requestInput({ prompt: "First", responseSchema: schema });
|
||||
const second = await ctx.requestInput({
|
||||
prompt: `Second after ${first.choice}`,
|
||||
responseSchema: schema,
|
||||
});
|
||||
return { output: [{ first: first.choice, second: second.choice }] };
|
||||
},
|
||||
});
|
||||
|
||||
const first = await createWorkflow().run();
|
||||
assert.equal(first.status, "needs_input");
|
||||
const second = await createWorkflow().resume(first.requiresInput!.resumeToken, {
|
||||
response: { choice: "red" },
|
||||
});
|
||||
assert.equal(second.status, "needs_input");
|
||||
|
||||
const payload = decodeToken(second.requiresInput!.resumeToken) as any;
|
||||
assert.equal(payload.resumeMode, "same_stage");
|
||||
assert.match(payload.stateKey, /^sdk_resume_/);
|
||||
assert.equal(payload.commandInput, undefined);
|
||||
assert.equal(payload.items, undefined);
|
||||
|
||||
const final = await createWorkflow().resume(second.requiresInput!.resumeToken, {
|
||||
response: { choice: "blue" },
|
||||
});
|
||||
assert.equal(final.ok, true);
|
||||
assert.equal(final.status, "ok");
|
||||
assert.deepEqual(final.output, [{ first: "red", second: "blue" }]);
|
||||
});
|
||||
|
||||
test("sdk Lobster preserves custom stateDir while using core runtime", async () => {
|
||||
const stateDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-sdk-state-dir-"));
|
||||
const workflow = new Lobster({ stateDir })
|
||||
.pipe(() => ({ saved: true }))
|
||||
.pipe(stateSet("custom-key"));
|
||||
|
||||
const result = await workflow.run();
|
||||
|
||||
assert.equal(result.ok, true);
|
||||
assert.equal(result.status, "ok");
|
||||
assert.deepEqual(result.output, [{ saved: true }]);
|
||||
assert.equal(
|
||||
await fsp.readFile(path.join(stateDir, "custom-key.json"), "utf8"),
|
||||
'{\n "saved": true\n}\n',
|
||||
);
|
||||
});
|
||||
|
||||
@@ -8,6 +8,13 @@ import os from "node:os";
|
||||
import { createDefaultRegistry } from "../src/commands/registry.js";
|
||||
import { runWorkflowFile } from "../src/workflows/file.js";
|
||||
import { decodeResumeToken } from "../src/resume.js";
|
||||
import { readStateJson } from "../src/state/store.js";
|
||||
|
||||
function streamOf(items: unknown[]) {
|
||||
return (async function* () {
|
||||
for (const item of items) yield item;
|
||||
})();
|
||||
}
|
||||
|
||||
test("workflow file runs with approval and resume", async () => {
|
||||
const workflow = {
|
||||
@@ -288,6 +295,549 @@ test("workflow file input steps pause and resume with structured responses", asy
|
||||
assert.deepEqual(resumed.output, [{ decision: "approve", subject: "hello" }]);
|
||||
});
|
||||
|
||||
test("workflow pipeline command input pauses and resumes the same pipeline step", async () => {
|
||||
const schema = JSON.stringify({
|
||||
type: "object",
|
||||
properties: { decision: { type: "string", enum: ["approve", "reject"] } },
|
||||
required: ["decision"],
|
||||
});
|
||||
const workflow = {
|
||||
steps: [
|
||||
{
|
||||
id: "draft",
|
||||
run: "node -e \"process.stdout.write(JSON.stringify({text:'hello'}))\"",
|
||||
},
|
||||
{
|
||||
id: "review",
|
||||
pipeline: `ask --subject-from-stdin --prompt 'Review draft?' --schema ${JSON.stringify(schema)} | pick decision`,
|
||||
stdin: "$draft.json",
|
||||
},
|
||||
{
|
||||
id: "finish",
|
||||
run: 'node -e "process.stdout.write(JSON.stringify({decision:process.env.DECISION}))"',
|
||||
env: {
|
||||
DECISION: "$review.json.decision",
|
||||
},
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-workflow-pipeline-input-"));
|
||||
const stateDir = path.join(tmpDir, "state");
|
||||
const filePath = path.join(tmpDir, "workflow.lobster");
|
||||
await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), "utf8");
|
||||
|
||||
const env = { ...process.env, LOBSTER_STATE_DIR: stateDir };
|
||||
const first = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry: createDefaultRegistry(),
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(first.status, "needs_input");
|
||||
assert.deepEqual(first.requiresInput?.subject, { text: '{"text":"hello"}' });
|
||||
const payload = decodeResumeToken(first.requiresInput?.resumeToken ?? "");
|
||||
assert.equal(payload.kind, "workflow-file");
|
||||
const state = (await readStateJson({ env, key: payload.stateKey! })) as any;
|
||||
assert.equal(state.resumeAtIndex, 1);
|
||||
assert.equal(state.inputKind, "pipeline_command");
|
||||
assert.equal(state.inputStepId, "review");
|
||||
assert.equal(state.pipelineInput.resumeAtIndex, 0);
|
||||
assert.deepEqual(state.pipelineInput.items, [{ text: "hello" }]);
|
||||
assert.deepEqual(state.pipelineInput.commandInput.pending.suspendedState, {
|
||||
type: "ask",
|
||||
subject: { text: '{"text":"hello"}' },
|
||||
});
|
||||
|
||||
const resumed = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry: createDefaultRegistry(),
|
||||
},
|
||||
resume: payload,
|
||||
response: { decision: "approve" },
|
||||
});
|
||||
|
||||
assert.equal(resumed.status, "ok");
|
||||
assert.deepEqual(resumed.output, [{ decision: "approve" }]);
|
||||
});
|
||||
|
||||
test("workflow pipeline requestInput resume invariant bypasses on_error", async () => {
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { decision: { type: "string" } },
|
||||
required: ["decision"],
|
||||
};
|
||||
let calls = 0;
|
||||
let sideEffects = 0;
|
||||
const choose = {
|
||||
name: "choose",
|
||||
async run({ ctx }: any) {
|
||||
calls += 1;
|
||||
if (calls > 1) return { output: streamOf([{ skipped: true }]) };
|
||||
await ctx.requestInput({ prompt: "Review?", responseSchema: schema });
|
||||
return { output: streamOf([]) };
|
||||
},
|
||||
};
|
||||
const side = {
|
||||
name: "side",
|
||||
async run() {
|
||||
sideEffects += 1;
|
||||
return { output: streamOf([{ sideEffects }]) };
|
||||
},
|
||||
};
|
||||
const registry = {
|
||||
get(name: string) {
|
||||
return name === "choose" ? choose : name === "side" ? side : undefined;
|
||||
},
|
||||
list() {
|
||||
return ["choose", "side"];
|
||||
},
|
||||
};
|
||||
const workflow = {
|
||||
name: "sample",
|
||||
steps: [
|
||||
{
|
||||
id: "review",
|
||||
pipeline: "choose",
|
||||
on_error: "continue",
|
||||
},
|
||||
{
|
||||
id: "side",
|
||||
pipeline: "side",
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-workflow-pipeline-invariant-"));
|
||||
const stateDir = path.join(tmpDir, "state");
|
||||
const filePath = path.join(tmpDir, "workflow.lobster");
|
||||
await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), "utf8");
|
||||
const env = { ...process.env, LOBSTER_STATE_DIR: stateDir };
|
||||
|
||||
const first = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
});
|
||||
assert.equal(first.status, "needs_input");
|
||||
const payload = decodeResumeToken(first.requiresInput?.resumeToken ?? "");
|
||||
assert.equal(payload.kind, "workflow-file");
|
||||
|
||||
await assert.rejects(
|
||||
runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
resume: payload,
|
||||
response: { decision: "approve" },
|
||||
}),
|
||||
/not consumed/,
|
||||
);
|
||||
assert.equal(sideEffects, 0);
|
||||
await fsp.access(path.join(stateDir, `${payload.stateKey}.json`));
|
||||
});
|
||||
|
||||
test("workflow pipeline requestInput resume rejects changed pipeline", async () => {
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { decision: { type: "string" } },
|
||||
required: ["decision"],
|
||||
};
|
||||
let sideEffects = 0;
|
||||
const choose = {
|
||||
name: "choose",
|
||||
async run({ ctx }: any) {
|
||||
const response = await ctx.requestInput({ prompt: "Review?", responseSchema: schema });
|
||||
return { output: streamOf([{ decision: response.decision }]) };
|
||||
},
|
||||
};
|
||||
const side = {
|
||||
name: "side",
|
||||
async run({ input }: any) {
|
||||
sideEffects += 1;
|
||||
return { output: input };
|
||||
},
|
||||
};
|
||||
const registry = {
|
||||
get(name: string) {
|
||||
return name === "choose" ? choose : name === "side" ? side : undefined;
|
||||
},
|
||||
list() {
|
||||
return ["choose", "side"];
|
||||
},
|
||||
};
|
||||
const workflow = {
|
||||
name: "sample",
|
||||
steps: [
|
||||
{
|
||||
id: "review",
|
||||
pipeline: "choose | side",
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-workflow-pipeline-change-"));
|
||||
const stateDir = path.join(tmpDir, "state");
|
||||
const filePath = path.join(tmpDir, "workflow.lobster");
|
||||
await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), "utf8");
|
||||
const env = { ...process.env, LOBSTER_STATE_DIR: stateDir };
|
||||
|
||||
const first = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
});
|
||||
assert.equal(first.status, "needs_input");
|
||||
const payload = decodeResumeToken(first.requiresInput?.resumeToken ?? "");
|
||||
assert.equal(payload.kind, "workflow-file");
|
||||
|
||||
workflow.steps[0].pipeline = "choose";
|
||||
await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), "utf8");
|
||||
|
||||
await assert.rejects(
|
||||
runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
resume: payload,
|
||||
response: { decision: "approve" },
|
||||
}),
|
||||
/pipeline changed/,
|
||||
);
|
||||
assert.equal(sideEffects, 0);
|
||||
await fsp.access(path.join(stateDir, `${payload.stateKey}.json`));
|
||||
});
|
||||
|
||||
test("workflow pipeline requestInput keeps full pipeline across repeated suspensions", async () => {
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { decision: { type: "string" } },
|
||||
required: ["decision"],
|
||||
};
|
||||
const produce = {
|
||||
name: "produce",
|
||||
async run() {
|
||||
return { output: streamOf([{ id: 1 }]) };
|
||||
},
|
||||
};
|
||||
const choose = {
|
||||
name: "choose",
|
||||
async run({ ctx }: any) {
|
||||
const first = await ctx.requestInput({
|
||||
prompt: "First?",
|
||||
responseSchema: schema,
|
||||
suspendedState: { phase: "first" },
|
||||
});
|
||||
const second = await ctx.requestInput({
|
||||
prompt: `Second after ${first.decision}`,
|
||||
responseSchema: schema,
|
||||
suspendedState: { phase: "second" },
|
||||
});
|
||||
return { output: streamOf([{ first: first.decision, second: second.decision }]) };
|
||||
},
|
||||
};
|
||||
const registry = {
|
||||
get(name: string) {
|
||||
return name === "produce" ? produce : name === "choose" ? choose : undefined;
|
||||
},
|
||||
list() {
|
||||
return ["produce", "choose"];
|
||||
},
|
||||
};
|
||||
const workflow = {
|
||||
name: "sample",
|
||||
steps: [
|
||||
{
|
||||
id: "review",
|
||||
pipeline: "produce | choose",
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-workflow-pipeline-repeat-"));
|
||||
const stateDir = path.join(tmpDir, "state");
|
||||
const filePath = path.join(tmpDir, "workflow.lobster");
|
||||
await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), "utf8");
|
||||
const env = { ...process.env, LOBSTER_STATE_DIR: stateDir };
|
||||
|
||||
const first = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
});
|
||||
assert.equal(first.status, "needs_input");
|
||||
const firstPayload = decodeResumeToken(first.requiresInput?.resumeToken ?? "");
|
||||
assert.equal(firstPayload.kind, "workflow-file");
|
||||
|
||||
const second = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
resume: firstPayload,
|
||||
response: { decision: "approve" },
|
||||
});
|
||||
assert.equal(second.status, "needs_input");
|
||||
const secondPayload = decodeResumeToken(second.requiresInput?.resumeToken ?? "");
|
||||
assert.equal(secondPayload.kind, "workflow-file");
|
||||
const state = (await readStateJson({ env, key: secondPayload.stateKey! })) as any;
|
||||
assert.equal(state.pipelineInput.resumeAtIndex, 1);
|
||||
assert.equal(state.pipelineInput.pipeline.length, 2);
|
||||
|
||||
const done = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
resume: secondPayload,
|
||||
response: { decision: "ship" },
|
||||
});
|
||||
assert.equal(done.status, "ok");
|
||||
assert.deepEqual(done.output, [{ first: "approve", second: "ship" }]);
|
||||
});
|
||||
|
||||
test("workflow pipeline requestInput resume rejects condition bypass", async () => {
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { decision: { type: "string" } },
|
||||
required: ["decision"],
|
||||
};
|
||||
let sideEffects = 0;
|
||||
const choose = {
|
||||
name: "choose",
|
||||
async run({ ctx }: any) {
|
||||
const response = await ctx.requestInput({ prompt: "Review?", responseSchema: schema });
|
||||
return { output: streamOf([{ decision: response.decision }]) };
|
||||
},
|
||||
};
|
||||
const side = {
|
||||
name: "side",
|
||||
async run() {
|
||||
sideEffects += 1;
|
||||
return { output: streamOf([{ sideEffects }]) };
|
||||
},
|
||||
};
|
||||
const registry = {
|
||||
get(name: string) {
|
||||
return name === "choose" ? choose : name === "side" ? side : undefined;
|
||||
},
|
||||
list() {
|
||||
return ["choose", "side"];
|
||||
},
|
||||
};
|
||||
const workflow = {
|
||||
name: "sample",
|
||||
steps: [
|
||||
{
|
||||
id: "gate",
|
||||
run: 'node -e "process.stdout.write(JSON.stringify({ok:true}))"',
|
||||
},
|
||||
{
|
||||
id: "review",
|
||||
pipeline: "choose",
|
||||
condition: "$gate.json.ok",
|
||||
},
|
||||
{
|
||||
id: "side",
|
||||
pipeline: "side",
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-workflow-pipeline-condition-"));
|
||||
const stateDir = path.join(tmpDir, "state");
|
||||
const filePath = path.join(tmpDir, "workflow.lobster");
|
||||
await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), "utf8");
|
||||
const env = { ...process.env, LOBSTER_STATE_DIR: stateDir };
|
||||
|
||||
const first = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
});
|
||||
assert.equal(first.status, "needs_input");
|
||||
const payload = decodeResumeToken(first.requiresInput?.resumeToken ?? "");
|
||||
assert.equal(payload.kind, "workflow-file");
|
||||
|
||||
workflow.steps[1].condition = "$gate.json.missing";
|
||||
await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), "utf8");
|
||||
|
||||
await assert.rejects(
|
||||
runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
resume: payload,
|
||||
response: { decision: "approve" },
|
||||
}),
|
||||
/condition changed/,
|
||||
);
|
||||
assert.equal(sideEffects, 0);
|
||||
await fsp.access(path.join(stateDir, `${payload.stateKey}.json`));
|
||||
});
|
||||
|
||||
test("workflow pipeline command input preserves replayable stdin without suspended state", async () => {
|
||||
const schema = {
|
||||
type: "object",
|
||||
properties: { decision: { type: "string" } },
|
||||
required: ["decision"],
|
||||
};
|
||||
const reviewCommand = {
|
||||
name: "review_input",
|
||||
async run({ input, ctx }: any) {
|
||||
const response = await ctx.requestInput({ prompt: "Review?", responseSchema: schema });
|
||||
const items = [];
|
||||
for await (const item of input) items.push(item);
|
||||
return { output: streamOf([{ items, decision: response.decision }]) };
|
||||
},
|
||||
};
|
||||
const registry = {
|
||||
get(name: string) {
|
||||
return name === reviewCommand.name ? reviewCommand : undefined;
|
||||
},
|
||||
list() {
|
||||
return [reviewCommand.name];
|
||||
},
|
||||
};
|
||||
|
||||
async function runCase({
|
||||
sourceStep,
|
||||
stdin,
|
||||
prefix,
|
||||
}: {
|
||||
sourceStep?: Record<string, unknown>;
|
||||
stdin?: string;
|
||||
prefix: string;
|
||||
}) {
|
||||
const steps = [
|
||||
...(sourceStep ? [sourceStep] : []),
|
||||
{
|
||||
id: "review",
|
||||
pipeline: "review_input",
|
||||
...(stdin ? { stdin } : null),
|
||||
},
|
||||
];
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), prefix));
|
||||
const stateDir = path.join(tmpDir, "state");
|
||||
const filePath = path.join(tmpDir, "workflow.lobster");
|
||||
await fsp.writeFile(filePath, JSON.stringify({ name: "sample", steps }, null, 2), "utf8");
|
||||
const env = { ...process.env, LOBSTER_STATE_DIR: stateDir };
|
||||
|
||||
const first = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
});
|
||||
assert.equal(first.status, "needs_input");
|
||||
|
||||
const payload = decodeResumeToken(first.requiresInput?.resumeToken ?? "");
|
||||
assert.equal(payload.kind, "workflow-file");
|
||||
const state = (await readStateJson({ env, key: payload.stateKey! })) as any;
|
||||
const resumed = await runWorkflowFile({
|
||||
filePath,
|
||||
ctx: {
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env,
|
||||
mode: "tool",
|
||||
registry,
|
||||
},
|
||||
resume: payload,
|
||||
response: { decision: "approve" },
|
||||
});
|
||||
|
||||
return { state, resumed };
|
||||
}
|
||||
|
||||
const noStdin = await runCase({ prefix: "lobster-workflow-pipeline-no-stdin-" });
|
||||
assert.deepEqual(noStdin.state.pipelineInput.items, []);
|
||||
assert.equal(noStdin.resumed.status, "ok");
|
||||
assert.deepEqual(noStdin.resumed.output, [{ items: [], decision: "approve" }]);
|
||||
|
||||
const withArrayStdin = await runCase({
|
||||
prefix: "lobster-workflow-pipeline-array-stdin-",
|
||||
sourceStep: {
|
||||
id: "draft",
|
||||
run: 'node -e "process.stdout.write(JSON.stringify([{id:1}]))"',
|
||||
},
|
||||
stdin: "$draft.json",
|
||||
});
|
||||
assert.deepEqual(withArrayStdin.state.pipelineInput.items, [{ id: 1 }]);
|
||||
assert.equal(withArrayStdin.resumed.status, "ok");
|
||||
assert.deepEqual(withArrayStdin.resumed.output, [{ items: [{ id: 1 }], decision: "approve" }]);
|
||||
});
|
||||
|
||||
test("workflow input resumes preserve the full subject even when the tool envelope preview is truncated", async () => {
|
||||
const longText = "x".repeat(250_000);
|
||||
const workflow = {
|
||||
|
||||
Reference in New Issue
Block a user