From 4cb4007441fb9e54541d3eca33f4124101efd8e3 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Wed, 22 Jul 2026 00:34:08 +0530 Subject: [PATCH] refactor(workflow): strengthen contracts and zod boundaries --- package-lock.json | 6 +- package.json | 3 +- src/json-types.ts | 42 ++++++ src/local-agent-adapters.test.ts | 2 +- src/local-agent-adapters.ts | 10 +- src/local-agent-capabilities.ts | 51 +++++++ src/local-agent-profiles.ts | 8 +- src/local-agent-runtime.ts | 8 +- src/workflow-api.ts | 88 +++++------ src/workflow-cli.ts | 25 +++- src/workflow-contracts.test.ts | 126 ++++++++++++++++ src/workflow-contracts.ts | 247 +++++++++++++++++++++++++++++++ src/workflow-engine.ts | 18 +-- src/workflow-files.ts | 9 +- src/workflow-replay.ts | 3 +- src/workflow-sandbox.test.ts | 10 +- src/workflow-sandbox.ts | 22 ++- src/workflow-schema.test.ts | 8 +- src/workflow-schema.ts | 39 +++-- src/workflow-script.test.ts | 8 +- src/workflow-script.ts | 97 ++---------- src/workflow-store.test.ts | 9 +- src/workflow-store.ts | 31 ++-- src/workflow-tools.ts | 9 +- src/workflow-types.ts | 129 +++++----------- 25 files changed, 687 insertions(+), 321 deletions(-) create mode 100644 src/json-types.ts create mode 100644 src/local-agent-capabilities.ts create mode 100644 src/workflow-contracts.test.ts create mode 100644 src/workflow-contracts.ts diff --git a/package-lock.json b/package-lock.json index 2029860a..39efe57c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -24,6 +24,7 @@ "diff": "^8.0.3", "drizzle-orm": "^0.45.2", "express": "^5.2.1", + "json-schema-to-ts": "^3.1.1", "lucide": "^1.24.0", "react": "^19.2.6", "react-dom": "^19.2.6", @@ -217,7 +218,6 @@ "resolved": "https://registry.npmjs.org/@babel/runtime/-/runtime-7.29.7.tgz", "integrity": "sha512-Nq8OhGWiZIZGV6hLHoyAKLLcJihP/xFeBMGJoUrxTX2psI8dCifzLhZISFb+VWS3wFMRDmCGw5R+dOySCqPLhw==", "license": "MIT", - "peer": true, "engines": { "node": ">=6.9.0" } @@ -4513,7 +4513,6 @@ "resolved": "https://registry.npmjs.org/json-schema-to-ts/-/json-schema-to-ts-3.1.1.tgz", "integrity": "sha512-+DWg8jCJG2TEnpy7kOm/7/AxaYoaRbjVB4LFZLySZlWn8exGs3A4OLJR966cVvU26N7X9TWxl+Jsw7dzAqKT6g==", "license": "MIT", - "peer": true, "dependencies": { "@babel/runtime": "^7.18.3", "ts-algebra": "^2.0.0" @@ -5841,8 +5840,7 @@ "version": "2.0.0", "resolved": "https://registry.npmjs.org/ts-algebra/-/ts-algebra-2.0.0.tgz", "integrity": "sha512-FPAhNPFMrkwz76P7cdjdmiShwMynZYN6SgOujD1urY4oNm80Ou9oMdmbR45LotcKOXoy7wSmHkRFE6Mxbrhefw==", - "license": "MIT", - "peer": true + "license": "MIT" }, "node_modules/tslib": { "version": "2.8.1", diff --git a/package.json b/package.json index f43b5dff..07f242f6 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,7 @@ "dev": "node scripts/dev-server.mjs", "postinstall": "node scripts/fix-node-pty-permissions.mjs", "start": "node dist/cli.js serve", - "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts && tsx src/workflow-types.test.ts && tsx src/workflow-store.test.ts && tsx src/workflow-script.test.ts && tsx src/workflow-sandbox.test.ts && tsx src/workflow-engine.test.ts && tsx src/workflow-files.test.ts && tsx src/workflow-replay.test.ts && tsx src/workflow-schema.test.ts", + "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts && tsx src/workflow-contracts.test.ts && tsx src/workflow-types.test.ts && tsx src/workflow-store.test.ts && tsx src/workflow-script.test.ts && tsx src/workflow-sandbox.test.ts && tsx src/workflow-engine.test.ts && tsx src/workflow-files.test.ts && tsx src/workflow-replay.test.ts && tsx src/workflow-schema.test.ts", "typecheck": "tsc -p tsconfig.json --noEmit" }, "keywords": [], @@ -49,6 +49,7 @@ "diff": "^8.0.3", "drizzle-orm": "^0.45.2", "express": "^5.2.1", + "json-schema-to-ts": "^3.1.1", "lucide": "^1.24.0", "react": "^19.2.6", "react-dom": "^19.2.6", diff --git a/src/json-types.ts b/src/json-types.ts new file mode 100644 index 00000000..a701edd4 --- /dev/null +++ b/src/json-types.ts @@ -0,0 +1,42 @@ +import type { JSONSchema } from "json-schema-to-ts"; +import * as z from "zod/v4"; + +export type JsonPrimitive = string | number | boolean | null; + +export type JsonValue = + | JsonPrimitive + | JsonValue[] + | { [key: string]: JsonValue }; + +export type JsonObject = { [key: string]: JsonValue }; + +/** JSON Schema is the portable contract shared with provider SDKs and Ajv. */ +export type JsonSchema = JSONSchema; + +export const jsonValueSchema: z.ZodType = z.lazy(() => + z.union([ + z.string(), + z.number().finite(), + z.boolean(), + z.null(), + z.array(jsonValueSchema), + z.record(z.string(), jsonValueSchema), + ]), +); + +export const jsonObjectSchema: z.ZodType = z.record( + z.string(), + jsonValueSchema, +); + +export const jsonSchemaSchema = jsonObjectSchema.transform( + (value): JsonSchema => value as JsonSchema, +); + +export function parseJsonValue(value: unknown): JsonValue { + return jsonValueSchema.parse(value); +} + +export function parseJsonText(text: string): JsonValue { + return parseJsonValue(JSON.parse(text) as unknown); +} diff --git a/src/local-agent-adapters.test.ts b/src/local-agent-adapters.test.ts index 4438f6cc..93cf4fc0 100644 --- a/src/local-agent-adapters.test.ts +++ b/src/local-agent-adapters.test.ts @@ -397,7 +397,7 @@ assert.equal( type: "object", properties: { n: { type: "number" } }, required: ["n"], - }; + } as const; assert.deepEqual(claudeOutputFormatOptions(schema), { outputFormat: { type: "json_schema", schema }, }); diff --git a/src/local-agent-adapters.ts b/src/local-agent-adapters.ts index fe56e81d..d452c1ba 100644 --- a/src/local-agent-adapters.ts +++ b/src/local-agent-adapters.ts @@ -1,7 +1,11 @@ import { spawn, spawnSync, type ChildProcessWithoutNullStreams } from "node:child_process"; import { resolve } from "node:path"; import { Readable, Writable } from "node:stream"; -import type { EffortLevel } from "@anthropic-ai/claude-agent-sdk"; +import type { + EffortLevel, + OutputFormat, +} from "@anthropic-ai/claude-agent-sdk"; +import type { JsonSchema } from "./json-types.js"; import type { LocalAgentProvider } from "./local-agent-profiles.js"; import { removeDevspaceNodeModulesBinFromPath } from "./local-agent-path.js"; import { @@ -116,8 +120,8 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter { /** Build Claude SDK outputFormat when a JSON Schema is requested. */ export function claudeOutputFormatOptions( - schema: object | undefined, -): { outputFormat: { type: "json_schema"; schema: Record } } | Record { + schema: JsonSchema | undefined, +): { outputFormat: OutputFormat } | Record { if (!schema) return {}; return { outputFormat: { diff --git a/src/local-agent-capabilities.ts b/src/local-agent-capabilities.ts new file mode 100644 index 00000000..f9a5f34a --- /dev/null +++ b/src/local-agent-capabilities.ts @@ -0,0 +1,51 @@ +import type { LocalAgentProvider } from "./local-agent-profiles.js"; + +export interface LocalAgentProviderCapabilities { + structuredOutput: "native" | "prompt"; + resumableSessions: boolean; + cancellation: "signal" | "process"; + supportsWorkspaceIsolation: boolean; +} + +export const LOCAL_AGENT_PROVIDER_CAPABILITIES = { + codex: { + structuredOutput: "native", + resumableSessions: true, + cancellation: "signal", + supportsWorkspaceIsolation: true, + }, + claude: { + structuredOutput: "native", + resumableSessions: true, + cancellation: "signal", + supportsWorkspaceIsolation: true, + }, + opencode: { + structuredOutput: "prompt", + resumableSessions: true, + cancellation: "process", + supportsWorkspaceIsolation: true, + }, + pi: { + structuredOutput: "prompt", + resumableSessions: true, + cancellation: "process", + supportsWorkspaceIsolation: true, + }, + cursor: { + structuredOutput: "prompt", + resumableSessions: true, + cancellation: "process", + supportsWorkspaceIsolation: true, + }, + copilot: { + structuredOutput: "prompt", + resumableSessions: true, + cancellation: "process", + supportsWorkspaceIsolation: true, + }, +} as const satisfies Record; + +export function supportsNativeStructuredOutput(provider: LocalAgentProvider): boolean { + return LOCAL_AGENT_PROVIDER_CAPABILITIES[provider].structuredOutput === "native"; +} diff --git a/src/local-agent-profiles.ts b/src/local-agent-profiles.ts index f29316dd..9cf846f6 100644 --- a/src/local-agent-profiles.ts +++ b/src/local-agent-profiles.ts @@ -4,16 +4,16 @@ import { basename, join, resolve } from "node:path"; import { parse as parseYaml } from "yaml"; import type { ServerConfig } from "./config.js"; -export type LocalAgentProvider = "codex" | "claude" | "opencode" | "pi" | "cursor" | "copilot"; - -export const LOCAL_AGENT_PROVIDERS: readonly LocalAgentProvider[] = [ +export const LOCAL_AGENT_PROVIDERS = [ "codex", "claude", "opencode", "pi", "cursor", "copilot", -]; +] as const; + +export type LocalAgentProvider = (typeof LOCAL_AGENT_PROVIDERS)[number]; export interface LocalAgentProfile { name: string; diff --git a/src/local-agent-runtime.ts b/src/local-agent-runtime.ts index 5690e469..0f3896ed 100644 --- a/src/local-agent-runtime.ts +++ b/src/local-agent-runtime.ts @@ -7,6 +7,8 @@ import type { ThreadOptions, TurnOptions, } from "@openai/codex-sdk"; +import type { JsonSchema } from "./json-types.js"; +import type { LocalAgentProvider } from "./local-agent-profiles.js"; export type LocalAgentWriteMode = "read_only" | "allowed" | "full_access"; @@ -19,11 +21,11 @@ export interface LocalAgentRunInput { /** Provider-native effort / reasoning level (was thinking). */ effort?: string; /** JSON Schema for native structured output (codex/claude). */ - schema?: object; + schema?: JsonSchema; } export interface LocalAgentRunResult { - provider: string; + provider: LocalAgentProvider; providerSessionId: string | null; finalResponse: string; items: unknown[]; @@ -32,7 +34,7 @@ export interface LocalAgentRunResult { } export interface LocalAgentRuntime { - readonly provider: string; + readonly provider: LocalAgentProvider; run(input: LocalAgentRunInput): Promise; } diff --git a/src/workflow-api.ts b/src/workflow-api.ts index 6e0eddaf..b1b09c93 100644 --- a/src/workflow-api.ts +++ b/src/workflow-api.ts @@ -1,6 +1,9 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { createHash } from "node:crypto"; import type { WorkflowSandboxApi } from "./workflow-sandbox.js"; +import type { LocalAgentProvider } from "./local-agent-profiles.js"; +import type { JsonSchema, JsonValue } from "./json-types.js"; +import { jsonValueSchema } from "./json-types.js"; import { WORKFLOW_LIMITS, WORKFLOW_MAX_ITEMS, @@ -9,15 +12,17 @@ import { createStubBudget, type AgentIsolationMode, type AgentOpts, + type AppendWorkflowEventInput, type WorkflowMeta, } from "./workflow-types.js"; +import { agentOptsSchema } from "./workflow-contracts.js"; // --------------------------------------------------------------------------- // Host deps (injected by engine; fakes OK in tests) // --------------------------------------------------------------------------- export interface WorkflowProviderRunInput { - provider: string; + provider: LocalAgentProvider; prompt: string; providerSessionId?: string; model?: string; @@ -27,7 +32,7 @@ export interface WorkflowProviderRunInput { label?: string; phase?: string; /** JSON Schema for native structured output (codex/claude). */ - schema?: object; + schema?: JsonSchema; } export interface WorkflowProviderRunResult { @@ -55,7 +60,7 @@ export type CreateAgentWorktree = (input: { }) => Promise; export interface WorkflowReplayHit { - value: unknown; + value: JsonValue; responseText?: string; structuredJson?: string; providerSessionId?: string; @@ -66,27 +71,14 @@ export interface WorkflowReplay { } export interface WorkflowJournal { - appendEvent(input: { - runId: string; - type: - | "phase_started" - | "log" - | "agent_call_started" - | "agent_call_completed" - | "agent_call_failed" - | "agent_call_cached" - | "schema_retry" - | "worktree_created" - | "worktree_finalized"; - phase?: string; - label?: string; - data?: unknown; - }): unknown; + appendEvent( + input: Extract, + ): unknown; beginAgentCall(input: { runId: string; callIndex: number; cacheKey: string; - provider: string; + provider: LocalAgentProvider; model?: string; effort?: string; label?: string; @@ -118,13 +110,13 @@ export interface WorkflowApiDeps { runId: string; journal: WorkflowJournal; meta: WorkflowMeta; - args: unknown; + args: JsonValue | undefined; concurrency: number; signal: AbortSignal; workspaceRoot: string; baseSha?: string; /** Already-filtered enabled ∩ live provider ids, preference order. */ - enabledProviders: string[]; + enabledProviders: LocalAgentProvider[]; runProvider: WorkflowRunProvider; createWorktree?: CreateAgentWorktree; replay?: WorkflowReplay; @@ -133,7 +125,7 @@ export interface WorkflowApiDeps { /** Run a nested script sharing semaphore/callIndex. */ executeNested?: (input: { source: string; - args: unknown; + args: JsonValue | undefined; nestDepth: number; }) => Promise; nestDepth?: number; @@ -558,11 +550,17 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { throw new WorkflowEngineError("internal", "nested workflow() is not configured on this host"); } const nameOrRef = args[0] as string | { scriptPath: string }; - const childArgs = args[1]; + const childArgsResult = jsonValueSchema.optional().safeParse(args[1]); + if (!childArgsResult.success) { + throw new WorkflowEngineError( + "internal", + `workflow() args must be JSON-serializable: ${childArgsResult.error.issues[0]?.message ?? "invalid value"}`, + ); + } const source = await deps.resolveNestedSource(nameOrRef); return deps.executeNested({ source, - args: childArgs, + args: childArgsResult.data, nestDepth: nestDepth + 1, }); }; @@ -592,10 +590,10 @@ export function hashCacheKey(input: ReturnType): } export function resolveProvider( - optsProvider: string | undefined, + optsProvider: LocalAgentProvider | undefined, meta: WorkflowMeta, - enabledProviders: string[], -): string { + enabledProviders: LocalAgentProvider[], +): LocalAgentProvider { if (optsProvider) { if (!enabledProviders.includes(optsProvider)) { throw new WorkflowEngineError( @@ -623,32 +621,18 @@ export function resolveProvider( function normalizeAgentOpts(opts: unknown): AgentOpts { if (opts === undefined || opts === null) return {}; - if (typeof opts !== "object" || Array.isArray(opts)) { - throw new WorkflowEngineError("internal", "agent opts must be an object"); - } - const record = opts as Record; - const out: AgentOpts = {}; - if (typeof record.label === "string") out.label = record.label; - if (typeof record.phase === "string") out.phase = record.phase; - if (record.schema !== undefined) { - if (!record.schema || typeof record.schema !== "object" || Array.isArray(record.schema)) { - throw new WorkflowEngineError("schema", "agent opts.schema must be an object"); - } - out.schema = record.schema as object; - } - if (typeof record.model === "string") out.model = record.model; - if (typeof record.effort === "string") out.effort = record.effort; - if (typeof record.provider === "string") out.provider = record.provider; - if (record.isolation !== undefined) { - if (record.isolation !== "worktree") { - throw new WorkflowEngineError("worktree", 'agent opts.isolation must be "worktree" when set'); - } - out.isolation = "worktree"; - } - if ("writeMode" in record) { + if (typeof opts === "object" && opts !== null && "writeMode" in opts) { throw new WorkflowEngineError("internal", "writeMode is not supported on agent() (v1)"); } - return out; + const parsed = agentOptsSchema.safeParse(opts); + if (parsed.success) return parsed.data; + const issue = parsed.error.issues[0]; + const path = issue?.path.join(".") || "opts"; + const kind = path === "schema" ? "schema" : path === "isolation" ? "worktree" : "internal"; + throw new WorkflowEngineError( + kind, + `Invalid agent ${path}: ${issue?.message ?? "validation failed"}`, + ); } function assertMaxItems(count: number, label: string): void { diff --git a/src/workflow-cli.ts b/src/workflow-cli.ts index e1c459c0..68d9f917 100644 --- a/src/workflow-cli.ts +++ b/src/workflow-cli.ts @@ -4,11 +4,13 @@ import { availableParallelism } from "node:os"; import { resolve } from "node:path"; import { fileURLToPath } from "node:url"; import type { ServerConfig } from "./config.js"; +import { parseJsonText, type JsonObject, type JsonValue } from "./json-types.js"; import { runLocalAgentProvider } from "./local-agent-adapters.js"; import { getLocalAgentProviderAvailabilitySnapshot } from "./local-agent-availability.js"; import { isLocalAgentProvider, LOCAL_AGENT_PROVIDERS, + type LocalAgentProvider, } from "./local-agent-profiles.js"; import { executeWorkflow, mapEngineErrorKind } from "./workflow-engine.js"; import { @@ -30,6 +32,7 @@ import { type WorkflowRunRecord, type WorkflowRunSource, } from "./workflow-types.js"; +import { parseWorkflowEventPayload } from "./workflow-contracts.js"; import { createWorkflowWorktreeFactory, resolveWorkspaceHead, @@ -119,7 +122,8 @@ async function runWorkflowRun(args: string[], config: ServerConfig): Promise row.available).map((row) => row.name)); if (!agentProviders) { return LOCAL_AGENT_PROVIDERS.filter((id) => live.has(id)); } - return agentProviders.enabled.filter((id) => live.has(id as never)); + return agentProviders.enabled.filter((id) => live.has(id)); } function splitFlags(args: string[]): { @@ -542,3 +547,7 @@ function collectArgTokens(args: string[]): string[] { function sleep(ms: number): Promise { return new Promise((resolveSleep) => setTimeout(resolveSleep, ms)); } + +function isJsonObject(value: JsonValue): value is JsonObject { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/workflow-contracts.test.ts b/src/workflow-contracts.test.ts new file mode 100644 index 00000000..417035b4 --- /dev/null +++ b/src/workflow-contracts.test.ts @@ -0,0 +1,126 @@ +import assert from "node:assert/strict"; +import { + agentOptsSchema, + localAgentProviderSchema, + parseWorkflowEventPayload, + workflowMetaSchema, + type WorkflowAgent, + type WorkflowParallel, +} from "./workflow-contracts.js"; +import { + jsonSchemaSchema, + jsonValueSchema, +} from "./json-types.js"; +import { + LOCAL_AGENT_PROVIDER_CAPABILITIES, +} from "./local-agent-capabilities.js"; +import { LOCAL_AGENT_PROVIDERS } from "./local-agent-profiles.js"; + +assert.deepEqual(localAgentProviderSchema.options, LOCAL_AGENT_PROVIDERS); +assert.deepEqual( + Object.keys(LOCAL_AGENT_PROVIDER_CAPABILITIES).sort(), + [...LOCAL_AGENT_PROVIDERS].sort(), +); + +assert.deepEqual( + workflowMetaSchema.parse({ + name: "typed-review", + description: "Review with typed contracts", + defaultProvider: "codex", + phases: [{ title: "Review" }], + }), + { + name: "typed-review", + description: "Review with typed contracts", + defaultProvider: "codex", + phases: [{ title: "Review" }], + }, +); + +assert.throws( + () => + workflowMetaSchema.parse({ + name: "typed-review", + description: "d", + unknown: true, + }), + /Unrecognized key/, +); + +assert.throws( + () => agentOptsSchema.parse({ provider: "made-up" }), + /Invalid option/, +); +assert.throws(() => agentOptsSchema.parse({ schema: [] }), /expected record/i); +assert.throws(() => jsonValueSchema.parse(new Date()), /invalid input/i); +assert.throws(() => jsonValueSchema.parse(() => undefined), /invalid input/i); + +assert.deepEqual( + jsonSchemaSchema.parse({ + type: "object", + properties: { count: { type: "number" } }, + required: ["count"], + }), + { + type: "object", + properties: { count: { type: "number" } }, + required: ["count"], + }, +); + +assert.deepEqual( + parseWorkflowEventPayload("agent_call_completed", { + callIndex: 2, + provider: "claude", + isolation: "shared", + fromCache: false, + }), + { + callIndex: 2, + provider: "claude", + isolation: "shared", + fromCache: false, + }, +); +assert.throws( + () => + parseWorkflowEventPayload("run_completed", { + provider: "codex", + }), + /callCount/, +); + +declare const agent: WorkflowAgent; +declare const parallel: WorkflowParallel; + +if (false) { + const output = await agent("Return a count", { + schema: { + type: "object", + properties: { + count: { type: "number" }, + }, + required: ["count"], + additionalProperties: false, + } as const, + }); + const count: number = output.count; + void count; + + // @ts-expect-error schema-derived output has no `missing` field + void output.missing; + + // @ts-expect-error providers are exhaustive + await agent("x", { provider: "made-up" }); + + const tuple = await parallel([ + async () => "text", + async () => 42, + ] as const); + const first: string | null = tuple[0]; + const second: number | null = tuple[1]; + void first; + void second; +} + +console.log("workflow-contracts.test.ts: ok"); diff --git a/src/workflow-contracts.ts b/src/workflow-contracts.ts new file mode 100644 index 00000000..bf43d58e --- /dev/null +++ b/src/workflow-contracts.ts @@ -0,0 +1,247 @@ +import type { FromSchema } from "json-schema-to-ts"; +import * as z from "zod/v4"; +import { LOCAL_AGENT_PROVIDERS } from "./local-agent-profiles.js"; +import type { LocalAgentProvider } from "./local-agent-profiles.js"; +import { jsonSchemaSchema, type JsonSchema, type JsonValue } from "./json-types.js"; + +export const localAgentProviderSchema = z.enum(LOCAL_AGENT_PROVIDERS); + +export const workflowMetaSchema = z + .object({ + name: z.string().trim().min(1).regex(/^[a-z0-9-]+$/), + description: z.string().trim().min(1), + phases: z + .array( + z + .object({ + title: z.string().trim().min(1), + detail: z.string().trim().min(1).optional(), + }) + .strict(), + ) + .optional(), + whenToUse: z.string().trim().min(1).optional(), + defaultProvider: localAgentProviderSchema.optional(), + concurrency: z.number().finite().int().positive().optional(), + }) + .strict(); + +export type WorkflowMeta = z.infer; +export type WorkflowPhaseMeta = NonNullable[number]; + +export const agentIsolationModeSchema = z.enum(["shared", "worktree"]); +export type AgentIsolationMode = z.infer; + +export const workflowRunStatusSchema = z.enum([ + "starting", + "running", + "completed", + "failed", + "cancelled", +]); +export type WorkflowRunStatus = z.infer; + +export const workflowAgentCallStatusSchema = z.enum([ + "running", + "completed", + "failed", + "cancelled", + "from_cache", +]); +export type WorkflowAgentCallStatus = z.infer; + +export const workflowRunSourceSchema = z.enum(["inline", "named", "resume"]); +export type WorkflowRunSource = z.infer; + +export const agentOptsSchema = z + .object({ + label: z.string().trim().min(1).optional(), + phase: z.string().trim().min(1).optional(), + schema: jsonSchemaSchema.optional(), + model: z.string().trim().min(1).optional(), + effort: z.string().trim().min(1).optional(), + provider: localAgentProviderSchema.optional(), + isolation: z.literal("worktree").optional(), + }) + .strict(); + +export type AgentOpts = Omit< + z.infer, + "schema" +> & { + schema?: S; +}; + +export interface WorkflowAgent { + ( + prompt: string, + opts: AgentOpts & { schema: S }, + ): Promise>; + (prompt: string, opts?: AgentOpts): Promise; +} + +export type WorkflowTask = () => T | Promise; + +export interface WorkflowParallel { + ( + tasks: T, + ): Promise<{ + [K in keyof T]: Awaited> | null; + }>; +} + +export interface WorkflowPipeline { + ( + items: readonly T[], + stage: (previous: T, item: T, index: number) => R | Promise, + ): Promise | null>>; + ( + items: readonly T[], + first: (previous: T, item: T, index: number) => A | Promise, + second: (previous: Awaited, item: T, index: number) => R | Promise, + ): Promise | null>>; + (...args: unknown[]): Promise>; +} + +export interface WorkflowNested { + (nameOrRef: string | { scriptPath: string }, args?: JsonValue): Promise; +} + +export const workflowErrorKindSchema = z.enum([ + "syntax", + "meta", + "determinism", + "provider_disabled", + "provider_unavailable", + "no_provider", + "provider", + "schema", + "cancelled", + "timeout", + "heartbeat", + "worktree", + "nest_depth", + "path", + "result_too_large", + "args_too_large", + "script_too_large", + "internal", +]); +export type WorkflowErrorKind = z.infer; + +export const WORKFLOW_EVENT_TYPES = [ + "run_started", + "run_completed", + "run_failed", + "run_cancelled", + "phase_started", + "log", + "agent_call_started", + "agent_call_completed", + "agent_call_failed", + "agent_call_cached", + "schema_retry", + "worktree_created", + "worktree_finalized", +] as const; + +export const workflowEventTypeSchema = z.enum(WORKFLOW_EVENT_TYPES); +export type WorkflowEventType = z.infer; + +export const workflowEventPayloadSchemas = { + run_started: z + .object({ + name: z.string(), + scriptHash: z.string(), + concurrency: z.number().int().positive(), + }) + .strict(), + run_completed: z.object({ callCount: z.number().int().nonnegative() }).strict(), + run_failed: z + .object({ error: z.string(), errorKind: workflowErrorKindSchema }) + .strict(), + run_cancelled: z.object({ reason: z.string().optional() }).strict(), + phase_started: z.object({ title: z.string().min(1) }).strict(), + log: z.object({ message: z.string() }).strict(), + agent_call_started: z + .object({ + callIndex: z.number().int().nonnegative(), + cacheKey: z.string(), + provider: localAgentProviderSchema, + isolation: agentIsolationModeSchema, + worktreePath: z.string().optional(), + }) + .strict(), + agent_call_completed: z + .object({ + callIndex: z.number().int().nonnegative(), + provider: localAgentProviderSchema, + isolation: agentIsolationModeSchema, + worktreePath: z.string().optional(), + dirty: z.boolean().optional(), + fromCache: z.boolean(), + }) + .strict(), + agent_call_failed: z + .object({ + callIndex: z.number().int().nonnegative(), + error: z.string(), + isolation: agentIsolationModeSchema, + worktreePath: z.string().optional(), + }) + .strict(), + agent_call_cached: z + .object({ + callIndex: z.number().int().nonnegative(), + cacheKey: z.string(), + provider: localAgentProviderSchema, + }) + .strict(), + schema_retry: z + .object({ + callIndex: z.number().int().nonnegative(), + attempt: z.number().int().positive(), + errors: z.string(), + mode: z.enum(["native", "prompt"]), + }) + .strict(), + worktree_created: z + .object({ + callIndex: z.number().int().nonnegative(), + worktreePath: z.string(), + isolation: z.literal("worktree"), + }) + .strict(), + worktree_finalized: z + .object({ + callIndex: z.number().int().nonnegative(), + worktreePath: z.string().optional(), + dirty: z.boolean(), + removed: z.boolean(), + outcome: z.literal("failure").optional(), + }) + .strict(), +} as const satisfies Record; + +export type WorkflowEventPayloads = { + [K in WorkflowEventType]: z.infer<(typeof workflowEventPayloadSchemas)[K]>; +}; + +export type AppendWorkflowEventInput = { + [P in K]: { + runId: string; + type: P; + phase?: string; + label?: string; + data: WorkflowEventPayloads[P]; + }; +}[K]; + +export function parseWorkflowEventPayload( + type: K, + data: unknown, +): WorkflowEventPayloads[K] { + return workflowEventPayloadSchemas[type].parse(data) as WorkflowEventPayloads[K]; +} + +export type WorkflowProviderId = LocalAgentProvider; diff --git a/src/workflow-engine.ts b/src/workflow-engine.ts index 856dc292..ff2a4726 100644 --- a/src/workflow-engine.ts +++ b/src/workflow-engine.ts @@ -1,4 +1,6 @@ import { availableParallelism } from "node:os"; +import type { LocalAgentProvider } from "./local-agent-profiles.js"; +import type { JsonValue } from "./json-types.js"; import { parseWorkflowScript, type ParsedWorkflowScript } from "./workflow-script.js"; import { runWorkflowSandbox } from "./workflow-sandbox.js"; import { @@ -23,21 +25,13 @@ export interface ExecuteWorkflowOptions { source?: string; filename?: string; runId: string; - journal: WorkflowJournal & { - appendEvent(input: { - runId: string; - type: string; - phase?: string; - label?: string; - data?: unknown; - }): unknown; - }; - args?: unknown; + journal: WorkflowJournal; + args?: JsonValue; concurrency?: number; signal?: AbortSignal; workspaceRoot: string; baseSha?: string; - enabledProviders: string[]; + enabledProviders: LocalAgentProvider[]; runProvider: WorkflowRunProvider; createWorktree?: CreateAgentWorktree; replay?: WorkflowReplay; @@ -143,7 +137,7 @@ async function executeNestedOnApi(input: { parentOptions: ExecuteWorkflowOptions; parentApi: WorkflowApi; source: string; - args: unknown; + args: JsonValue | undefined; nestDepth: number; }): Promise { if (input.nestDepth > WORKFLOW_MAX_NEST_DEPTH_LOCAL) { diff --git a/src/workflow-files.ts b/src/workflow-files.ts index 12b12228..7615980f 100644 --- a/src/workflow-files.ts +++ b/src/workflow-files.ts @@ -2,6 +2,7 @@ import { createHash, randomBytes } from "node:crypto"; import { access, mkdir, readFile, writeFile } from "node:fs/promises"; import { basename, dirname, extname, isAbsolute, join, resolve } from "node:path"; import { hashSource } from "./workflow-script.js"; +import { jsonValueSchema, type JsonValue } from "./json-types.js"; export class WorkflowPathError extends Error { constructor(message: string) { @@ -120,10 +121,10 @@ export async function resolveWorkflowScriptFromPathOrName(input: { } export function parseWorkflowArgFlags(tokens: string[]): { - args: Record; + args: Record; rest: string[]; } { - const args: Record = {}; + const args: Record = {}; const rest: string[] = []; for (let i = 0; i < tokens.length; i += 1) { const token = tokens[i]!; @@ -150,9 +151,9 @@ export function parseWorkflowArgFlags(tokens: string[]): { return { args, rest }; } -function coerceArgValue(raw: string): unknown { +function coerceArgValue(raw: string): JsonValue { try { - return JSON.parse(raw); + return jsonValueSchema.parse(JSON.parse(raw) as unknown); } catch { return raw; } diff --git a/src/workflow-replay.ts b/src/workflow-replay.ts index bff7b935..3038d1b0 100644 --- a/src/workflow-replay.ts +++ b/src/workflow-replay.ts @@ -1,5 +1,6 @@ import type { WorkflowAgentCallRecord } from "./workflow-types.js"; import type { WorkflowReplay, WorkflowReplayHit } from "./workflow-api.js"; +import { parseJsonText } from "./json-types.js"; /** * Resume matcher: @@ -63,7 +64,7 @@ function toHit(call: WorkflowAgentCallRecord): WorkflowReplayHit { if (call.structuredJson) { try { return { - value: JSON.parse(call.structuredJson), + value: parseJsonText(call.structuredJson), responseText: call.responseText, structuredJson: call.structuredJson, providerSessionId: call.providerSessionId, diff --git a/src/workflow-sandbox.test.ts b/src/workflow-sandbox.test.ts index 9620cd1d..7805aa75 100644 --- a/src/workflow-sandbox.test.ts +++ b/src/workflow-sandbox.test.ts @@ -1,9 +1,13 @@ import assert from "node:assert/strict"; import { parseWorkflowScript } from "./workflow-script.js"; -import { createStubBudget, type WorkflowMeta } from "./workflow-types.js"; +import { + createStubBudget, + type WorkflowMeta, +} from "./workflow-types.js"; +import type { WorkflowSandboxApi } from "./workflow-sandbox.js"; import { runWorkflowSandbox, WorkflowDeterminismError } from "./workflow-sandbox.js"; -function api(meta: WorkflowMeta, logs?: string[]) { +function api(meta: WorkflowMeta, logs?: string[]): WorkflowSandboxApi { return { agent: async () => "", parallel: async () => [], @@ -16,7 +20,7 @@ function api(meta: WorkflowMeta, logs?: string[]) { budget: createStubBudget(), workflow: async () => null, meta, - }; + } as unknown as WorkflowSandboxApi; } { diff --git a/src/workflow-sandbox.ts b/src/workflow-sandbox.ts index f05a17f1..656e128c 100644 --- a/src/workflow-sandbox.ts +++ b/src/workflow-sandbox.ts @@ -1,6 +1,14 @@ import vm from "node:vm"; import type { ParsedWorkflowScript } from "./workflow-script.js"; -import type { WorkflowBudget, WorkflowMeta } from "./workflow-types.js"; +import type { JsonValue } from "./json-types.js"; +import type { + WorkflowAgent, + WorkflowBudget, + WorkflowMeta, + WorkflowNested, + WorkflowParallel, + WorkflowPipeline, +} from "./workflow-types.js"; export class WorkflowDeterminismError extends Error { constructor(message: string) { @@ -10,14 +18,14 @@ export class WorkflowDeterminismError extends Error { } export interface WorkflowSandboxApi { - agent: (...args: unknown[]) => unknown; - parallel: (...args: unknown[]) => unknown; - pipeline: (...args: unknown[]) => unknown; - phase: (...args: unknown[]) => unknown; + agent: WorkflowAgent; + parallel: WorkflowParallel; + pipeline: WorkflowPipeline; + phase: (title: string) => void; log: (...args: unknown[]) => unknown; - args: unknown; + args: JsonValue | undefined; budget: WorkflowBudget; - workflow: (...args: unknown[]) => unknown; + workflow: WorkflowNested; /** Host bookkeeping only; script binds its own `const meta`. */ meta: WorkflowMeta; } diff --git a/src/workflow-schema.test.ts b/src/workflow-schema.test.ts index 2084f513..8e92ffcd 100644 --- a/src/workflow-schema.test.ts +++ b/src/workflow-schema.test.ts @@ -3,10 +3,10 @@ import { augmentPromptForSchema, enforceAgentSchema, formatAjvErrors, - NATIVE_SCHEMA_PROVIDERS, } from "./workflow-schema.js"; import { WorkflowEngineError } from "./workflow-api.js"; import { ProviderSchemaUnsupportedError } from "./local-agent-runtime.js"; +import { supportsNativeStructuredOutput } from "./local-agent-capabilities.js"; { const prompt = augmentPromptForSchema("find bugs", { @@ -23,9 +23,9 @@ assert.equal( "/n must be number", ); -assert.ok(NATIVE_SCHEMA_PROVIDERS.has("codex")); -assert.ok(NATIVE_SCHEMA_PROVIDERS.has("claude")); -assert.ok(!NATIVE_SCHEMA_PROVIDERS.has("opencode")); +assert.ok(supportsNativeStructuredOutput("codex")); +assert.ok(supportsNativeStructuredOutput("claude")); +assert.ok(!supportsNativeStructuredOutput("opencode")); { let attempts = 0; diff --git a/src/workflow-schema.ts b/src/workflow-schema.ts index 1d93fe5f..b5234b45 100644 --- a/src/workflow-schema.ts +++ b/src/workflow-schema.ts @@ -3,14 +3,18 @@ import { WORKFLOW_MAX_SCHEMA_RETRIES } from "./workflow-types.js"; import { tryExtractJson, WorkflowEngineError } from "./workflow-api.js"; import type { WorkflowProviderRunResult, WorkflowRunProvider } from "./workflow-api.js"; import { isProviderSchemaUnsupportedError } from "./local-agent-runtime.js"; +import { supportsNativeStructuredOutput } from "./local-agent-capabilities.js"; +import type { LocalAgentProvider } from "./local-agent-profiles.js"; +import { + jsonValueSchema, + type JsonSchema, + type JsonValue, +} from "./json-types.js"; const require = createRequire(import.meta.url); -/** Providers with a real structured-output API (hardcoded — no capability probe). */ -export const NATIVE_SCHEMA_PROVIDERS = new Set(["codex", "claude"]); - type AjvLike = new (opts?: object) => { - compile: (schema: object) => ((data: unknown) => boolean) & { + compile: (schema: JsonSchema) => ((data: unknown) => boolean) & { errors?: Array<{ instancePath?: string; message?: string }> | null; }; }; @@ -30,13 +34,13 @@ function loadAjv(): AjvLike { export type SchemaEnforceMode = "native" | "prompt"; export interface EnforceSchemaInput { - schema: object; + schema: JsonSchema; prompt: string; /** * Provider id for native-vs-prompt policy. When in NATIVE_SCHEMA_PROVIDERS, * attempt 0 uses raw prompt + native structured path; later attempts repair via prompt. */ - provider?: string; + provider?: LocalAgentProvider; run: ( prompt: string, opts: { @@ -53,7 +57,7 @@ export interface EnforceSchemaInput { } export interface EnforceSchemaResult { - value: unknown; + value: JsonValue; finalResponse: string; providerSessionId?: string; attempts: number; @@ -71,7 +75,7 @@ export async function enforceAgentSchema( const ajv = new Ajv({ allErrors: true, strict: false }); const validate = ajv.compile(input.schema); const maxRetries = input.maxRetries ?? WORKFLOW_MAX_SCHEMA_RETRIES; - const native = Boolean(input.provider && NATIVE_SCHEMA_PROVIDERS.has(input.provider)); + const native = Boolean(input.provider && supportsNativeStructuredOutput(input.provider)); const basePrompt = augmentPromptForSchema(input.prompt, input.schema); let lastErrors = "unknown validation error"; @@ -134,7 +138,7 @@ export async function enforceAgentSchema( ); } -export function augmentPromptForSchema(prompt: string, schema: object): string { +export function augmentPromptForSchema(prompt: string, schema: JsonSchema): string { return [ prompt, "", @@ -158,7 +162,7 @@ export function formatAjvErrors( /** Helper for wiring into agent(): wrap a one-shot provider as retrying schema runner. */ export function schemaAwareRunProvider( runProvider: WorkflowRunProvider, - schema: object, + schema: JsonSchema, base: Parameters[0], onRetry?: EnforceSchemaInput["onRetry"], ): Promise { @@ -177,16 +181,21 @@ export function schemaAwareRunProvider( }); } -function structuredCandidates(result: WorkflowProviderRunResult): unknown[] { - const candidates: unknown[] = []; +function structuredCandidates(result: WorkflowProviderRunResult): JsonValue[] { + const candidates: JsonValue[] = []; if (result.structured !== undefined) { - candidates.push(result.structured); + const structured = jsonValueSchema.safeParse(result.structured); + if (structured.success) candidates.push(structured.data); if (typeof result.structured === "string") { const parsed = tryExtractJson(result.structured); - if (parsed !== undefined && parsed !== result.structured) candidates.push(parsed); + const parsedJson = jsonValueSchema.safeParse(parsed); + if (parsedJson.success && parsedJson.data !== result.structured) { + candidates.push(parsedJson.data); + } } } const fromText = tryExtractJson(result.finalResponse); - if (fromText !== undefined) candidates.push(fromText); + const textJson = jsonValueSchema.safeParse(fromText); + if (textJson.success) candidates.push(textJson.data); return candidates; } diff --git a/src/workflow-script.test.ts b/src/workflow-script.test.ts index c8e61423..d64c5cfc 100644 --- a/src/workflow-script.test.ts +++ b/src/workflow-script.test.ts @@ -1,7 +1,11 @@ import assert from "node:assert/strict"; import { parseWorkflowScript, WorkflowScriptError } from "./workflow-script.js"; import { createStubBudget } from "./workflow-types.js"; -import { runWorkflowSandbox, WorkflowDeterminismError } from "./workflow-sandbox.js"; +import { + runWorkflowSandbox, + WorkflowDeterminismError, + type WorkflowSandboxApi, +} from "./workflow-sandbox.js"; { const parsed = parseWorkflowScript(` @@ -91,7 +95,7 @@ async function runBody(source: string): Promise { budget: createStubBudget(), workflow: async () => null, meta: parsed.meta, - }, + } as WorkflowSandboxApi, }); } diff --git a/src/workflow-script.ts b/src/workflow-script.ts index 5847f08a..cf949568 100644 --- a/src/workflow-script.ts +++ b/src/workflow-script.ts @@ -1,11 +1,7 @@ import { createHash } from "node:crypto"; import vm from "node:vm"; -import { - LOCAL_AGENT_PROVIDERS, - type LocalAgentProvider, - isLocalAgentProvider, -} from "./local-agent-profiles.js"; import { WORKFLOW_LIMITS, type WorkflowMeta } from "./workflow-types.js"; +import { workflowMetaSchema } from "./workflow-contracts.js"; export class WorkflowScriptError extends Error { constructor( @@ -28,7 +24,6 @@ export interface ParsedWorkflowScript { } const META_EXPORT = /export\s+const\s+meta\s*=/; -const NAME_RE = /^[a-z0-9-]+$/; /** * Parse + compile a workflow script. @@ -189,87 +184,21 @@ function evaluateMetaLiteral(literal: string, filename: string): unknown { } function validateMeta(value: unknown): WorkflowMeta { - if (!value || typeof value !== "object" || Array.isArray(value)) { - throw new WorkflowScriptError("meta", "meta must be an object"); - } - const record = value as Record; - const name = readRequiredString(record, "name"); - if (!NAME_RE.test(name)) { - throw new WorkflowScriptError( - "meta", - `meta.name must match ${NAME_RE} (got ${JSON.stringify(name)})`, - ); - } - const description = readRequiredString(record, "description"); - - let phases: WorkflowMeta["phases"]; - if (record.phases !== undefined) { - if (!Array.isArray(record.phases)) { - throw new WorkflowScriptError("meta", "meta.phases must be an array"); - } - const hostPhases: NonNullable = []; - for (let index = 0; index < record.phases.length; index += 1) { - const phase = record.phases[index]; - if (!phase || typeof phase !== "object" || Array.isArray(phase)) { - throw new WorkflowScriptError("meta", `meta.phases[${index}] must be an object`); - } - const p = phase as Record; - const title = readRequiredString(p, "title", `meta.phases[${index}].title`); - const detail = optionalString(p.detail); - hostPhases.push(detail === undefined ? { title } : { title, detail }); - } - phases = hostPhases; - } - - const whenToUse = optionalString(record.whenToUse); - let defaultProvider: LocalAgentProvider | undefined; - if (record.defaultProvider !== undefined) { - if (typeof record.defaultProvider !== "string" || !isLocalAgentProvider(record.defaultProvider)) { - throw new WorkflowScriptError( - "meta", - `meta.defaultProvider must be one of: ${LOCAL_AGENT_PROVIDERS.join(", ")}`, - ); - } - defaultProvider = record.defaultProvider; - } + const parsed = workflowMetaSchema.safeParse(value); + if (parsed.success) return parsed.data; - let concurrency: number | undefined; - if (record.concurrency !== undefined) { - if (typeof record.concurrency !== "number" || !Number.isFinite(record.concurrency)) { - throw new WorkflowScriptError("meta", "meta.concurrency must be a number"); - } - concurrency = Math.floor(record.concurrency); - if (concurrency < 1) { - throw new WorkflowScriptError("meta", "meta.concurrency must be >= 1"); - } + const issue = parsed.error.issues[0]; + const path = issue?.path.length ? `meta.${issue.path.join(".")}` : "meta"; + if (issue?.code === "invalid_type" && issue.input === undefined) { + throw new WorkflowScriptError("meta", `${path} is required`); } - - return { - name, - description, - ...(phases ? { phases } : {}), - ...(whenToUse ? { whenToUse } : {}), - ...(defaultProvider ? { defaultProvider } : {}), - ...(concurrency !== undefined ? { concurrency } : {}), - }; -} - -function readRequiredString( - record: Record, - key: string, - label = `meta.${key}`, -): string { - const value = record[key]; - if (typeof value !== "string" || !value.trim()) { - throw new WorkflowScriptError("meta", `${label} is required`); + if (issue?.code === "invalid_format" && issue.format === "regex") { + throw new WorkflowScriptError("meta", `${path} must match /^[a-z0-9-]+$/`); } - return value.trim(); -} - -function optionalString(value: unknown): string | undefined { - if (typeof value !== "string") return undefined; - const trimmed = value.trim(); - return trimmed || undefined; + throw new WorkflowScriptError( + "meta", + `${path}: ${issue?.message ?? "validation failed"}`, + ); } function parseErrorLine(message: string): number | undefined { diff --git a/src/workflow-store.test.ts b/src/workflow-store.test.ts index f436a5ca..314ab289 100644 --- a/src/workflow-store.test.ts +++ b/src/workflow-store.test.ts @@ -36,12 +36,17 @@ try { store.setHeartbeat(run.id); assert.ok(store.getRun(run.id)?.heartbeatAt); - const e1 = store.appendEvent({ runId: run.id, type: "run_started", data: { ok: true } }); + const e1 = store.appendEvent({ + runId: run.id, + type: "run_started", + data: { name: run.name, scriptHash: run.scriptHash, concurrency: 1 }, + }); const e2 = store.appendEvent({ runId: run.id, type: "phase_started", phase: "Review", label: "r1", + data: { title: "Review" }, }); const e3 = store.appendEvent({ runId: run.id, type: "log", data: { message: "hello" } }); assert.equal(e1.seq, 1); @@ -147,7 +152,7 @@ try { workspaceRoot: join(root, "project"), }); const seqs = [0, 1, 2, 3, 4].map(() => - store.appendEvent({ runId: run4.id, type: "log", data: { n: 1 } }).seq, + store.appendEvent({ runId: run4.id, type: "log", data: { message: "1" } }).seq, ); assert.deepEqual(seqs, [1, 2, 3, 4, 5]); diff --git a/src/workflow-store.ts b/src/workflow-store.ts index 4e57e184..90d866c3 100644 --- a/src/workflow-store.ts +++ b/src/workflow-store.ts @@ -5,15 +5,23 @@ import type { ServerConfig } from "./config.js"; import { WORKFLOW_LIMITS, type AgentIsolationMode, + type AppendWorkflowEventInput, type WorkflowAgentCallRecord, type WorkflowAgentCallStatus, type WorkflowErrorKind, type WorkflowEventRecord, - type WorkflowEventType, type WorkflowRunRecord, type WorkflowRunSource, type WorkflowRunStatus, } from "./workflow-types.js"; +import { + localAgentProviderSchema, + parseWorkflowEventPayload, + workflowAgentCallStatusSchema, + workflowEventTypeSchema, + workflowRunSourceSchema, + workflowRunStatusSchema, +} from "./workflow-contracts.js"; export interface CreateWorkflowRunInput { name: string; @@ -27,14 +35,6 @@ export interface CreateWorkflowRunInput { baseSha?: string; } -export interface AppendWorkflowEventInput { - runId: string; - type: WorkflowEventType; - phase?: string; - label?: string; - data?: unknown; -} - export interface BeginAgentCallInput { runId: string; callIndex: number; @@ -338,7 +338,8 @@ export class WorkflowStore { } appendEvent(input: AppendWorkflowEventInput): WorkflowEventRecord { - const dataJson = truncateJson(input.data ?? {}, WORKFLOW_LIMITS.eventDataJsonBytes); + const payload = parseWorkflowEventPayload(input.type, input.data); + const dataJson = truncateJson(payload, WORKFLOW_LIMITS.eventDataJsonBytes); const createdAt = isoNow(); const insert = this.database.sqlite.transaction(() => { @@ -562,13 +563,13 @@ function rowToRun(row: WorkflowRunRow): WorkflowRunRecord { return { id: row.id, name: row.name, - source: row.source as WorkflowRunSource, + source: workflowRunSourceSchema.parse(row.source), scriptPath: row.script_path, scriptHash: row.script_hash, workspaceRoot: row.workspace_root, workspaceId: row.workspace_id ?? undefined, argsJson: row.args_json, - status: row.status as WorkflowRunStatus, + status: workflowRunStatusSchema.parse(row.status), error: row.error ?? undefined, errorKind: (row.error_kind as WorkflowErrorKind | null) ?? undefined, resultJson: row.result_json ?? undefined, @@ -588,7 +589,7 @@ function rowToEvent(row: WorkflowEventRow): WorkflowEventRecord { return { runId: row.run_id, seq: row.seq, - type: row.type as WorkflowEventType, + type: workflowEventTypeSchema.parse(row.type), phase: row.phase ?? undefined, label: row.label ?? undefined, dataJson: row.data_json, @@ -601,12 +602,12 @@ function rowToAgentCall(row: WorkflowAgentCallRow): WorkflowAgentCallRecord { runId: row.run_id, callIndex: row.call_index, cacheKey: row.cache_key, - provider: row.provider, + provider: localAgentProviderSchema.parse(row.provider), model: row.model ?? undefined, effort: row.effort ?? undefined, label: row.label ?? undefined, phase: row.phase ?? undefined, - status: row.status as WorkflowAgentCallStatus, + status: workflowAgentCallStatusSchema.parse(row.status), fromCache: row.from_cache === "true", providerSessionId: row.provider_session_id ?? undefined, responseText: row.response_text ?? undefined, diff --git a/src/workflow-tools.ts b/src/workflow-tools.ts index 8551701f..faefce75 100644 --- a/src/workflow-tools.ts +++ b/src/workflow-tools.ts @@ -3,6 +3,7 @@ import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { registerAppTool } from "@modelcontextprotocol/ext-apps/server"; import * as z from "zod/v4"; import type { ServerConfig } from "./config.js"; +import { jsonValueSchema, parseJsonText, type JsonValue } from "./json-types.js"; import type { WorkspaceRegistry } from "./workspaces.js"; import { persistWorkflowScript, @@ -62,7 +63,7 @@ export function registerWorkflowTools( .describe("Inline workflow script source (export const meta = …)."), name: z.string().optional().describe("Named workflow under .devspace/workflows/.js"), resumeFromRunId: z.string().optional().describe("Prior run id to resume (new run + cache)."), - args: z.unknown().optional().describe("Args object/array passed to script as `args`."), + args: jsonValueSchema.optional().describe("JSON args passed to script as `args`."), yieldTimeMs: z .number() .int() @@ -102,7 +103,7 @@ export function registerWorkflowTools( runSource = "resume"; if (args === undefined && prior.argsJson && prior.argsJson !== "null") { try { - args = JSON.parse(prior.argsJson); + args = parseJsonText(prior.argsJson); } catch { // keep undefined } @@ -288,9 +289,9 @@ function toolResult(page: { }; } -function safeJson(text: string): unknown { +function safeJson(text: string): JsonValue { try { - return JSON.parse(text); + return parseJsonText(text); } catch { return text; } diff --git a/src/workflow-types.ts b/src/workflow-types.ts index 11b14193..ad9b9d8a 100644 --- a/src/workflow-types.ts +++ b/src/workflow-types.ts @@ -10,6 +10,38 @@ */ import type { LocalAgentProvider } from "./local-agent-profiles.js"; +import type { JsonSchema } from "./json-types.js"; +import type { + AgentIsolationMode, + AgentOpts, + WorkflowErrorKind, + WorkflowAgentCallStatus, + WorkflowEventType, + WorkflowMeta, + WorkflowPhaseMeta, + WorkflowRunSource, + WorkflowRunStatus, +} from "./workflow-contracts.js"; + +export type { JsonObject, JsonPrimitive, JsonSchema, JsonValue } from "./json-types.js"; +export type { + AgentIsolationMode, + AgentOpts, + AppendWorkflowEventInput, + WorkflowAgent, + WorkflowAgentCallStatus, + WorkflowErrorKind, + WorkflowEventPayloads, + WorkflowEventType, + WorkflowMeta, + WorkflowNested, + WorkflowParallel, + WorkflowPhaseMeta, + WorkflowPipeline, + WorkflowRunSource, + WorkflowRunStatus, + WorkflowTask, +} from "./workflow-contracts.js"; // --------------------------------------------------------------------------- // Limits @@ -58,101 +90,14 @@ export interface AgentProvidersConfig { lastProbe?: AgentProviderProbe[]; } -// --------------------------------------------------------------------------- -// Script meta + agent opts -// --------------------------------------------------------------------------- - -export interface WorkflowPhaseMeta { - title: string; - detail?: string; -} - -export interface WorkflowMeta { - name: string; - description: string; - phases?: WorkflowPhaseMeta[]; - whenToUse?: string; - /** DevSpace extension */ - defaultProvider?: AgentProviderId; - /** DevSpace extension; clamped to engine max */ - concurrency?: number; -} - -/** - * Public agent() options. Deliberately no writeMode. - */ -export interface AgentOpts { - label?: string; - phase?: string; - schema?: object; - model?: string; - /** Provider-native effort/reasoning level (was thinking). */ - effort?: string; - provider?: AgentProviderId | string; - isolation?: "worktree"; -} - -export type AgentIsolationMode = "shared" | "worktree"; - // --------------------------------------------------------------------------- // Status / events // --------------------------------------------------------------------------- -export type WorkflowRunStatus = - | "starting" - | "running" - | "completed" - | "failed" - | "cancelled"; - -export type WorkflowAgentCallStatus = - | "running" - | "completed" - | "failed" - | "cancelled" - | "from_cache"; - -export type WorkflowEventType = - | "run_started" - | "run_completed" - | "run_failed" - | "run_cancelled" - | "phase_started" - | "log" - | "agent_call_started" - | "agent_call_completed" - | "agent_call_failed" - | "agent_call_cached" - | "schema_retry" - | "worktree_created" - | "worktree_finalized"; - -export type WorkflowErrorKind = - | "syntax" - | "meta" - | "determinism" - | "provider_disabled" - | "provider_unavailable" - | "no_provider" - | "provider" - | "schema" - | "cancelled" - | "timeout" - | "heartbeat" - | "worktree" - | "nest_depth" - | "path" - | "result_too_large" - | "args_too_large" - | "script_too_large" - | "internal"; - // --------------------------------------------------------------------------- // Journal row shapes (behavioral; store maps snake_case) // --------------------------------------------------------------------------- -export type WorkflowRunSource = "inline" | "named" | "resume"; - export interface WorkflowRunRecord { id: string; name: string; @@ -192,7 +137,7 @@ export interface WorkflowAgentCallRecord { runId: string; callIndex: number; cacheKey: string; - provider: string; + provider: AgentProviderId; model?: string; effort?: string; label?: string; @@ -222,19 +167,19 @@ export interface WorkflowAgentCallRecord { */ export interface AgentCacheKeyInput { prompt: string; - provider: string; + provider: AgentProviderId; model: string | null; effort: string | null; - schema: object | null; + schema: JsonSchema | null; isolation: AgentIsolationMode; } export function buildAgentCacheKeyInput(input: { prompt: string; - provider: string; + provider: AgentProviderId; model?: string | null; effort?: string | null; - schema?: object | null; + schema?: JsonSchema | null; isolation?: AgentIsolationMode | "worktree" | null; }): AgentCacheKeyInput { const isolation: AgentIsolationMode =