fix(workflow): preflight structured output schemas before agents start

AGENTS.md

@@ -45,12 +45,12 @@
 - Opt-in via `createFlowExtension({ workflow })` (compatibility alias: `createSubagentExtension`); defaults to `true`. Set `false` for a subagents-only surface (then only `Agent` registers and no workflow prompt is appended).
 - The `workflow` tool runs a trusted, model-written JavaScript script in an isolated Worker-hosted `node:vm` context so pi can detect stalls and abort unresponsive scripts. Initial synchronous execution is bounded (5s by default), and post-`await` event-loop stalls are caught by a heartbeat watchdog. This is not a security sandbox; saved workflows are trusted code like extensions, and inline workflows are model-written code executed by the local process. Globals: `agent(prompt, opts)`, `parallel(thunks)`, `pipeline(items, ...stages)`, `phase(title)`, `log(message)`, `args`, `cwd`. The script must start with `export const meta = { name, description }` (a plain literal) and call `agent()` at least once.
 - Determinism is a cooperative parse-time lint via an `acorn` AST scan: Date APIs and `Math.random()` uses, including simple aliases/destructuring, are rejected for normal model-written scripts. Dynamic authoring, deterministic-by-convention execution. The scan checks determinism ONLY — it intentionally permits ordinary computed member access (`obj[key]`, `arr[i]`, `{ [k]: v }`) except static `Math['random']`, and does not attempt vm-escape hardening. Do not claim malicious JavaScript is sandboxed.
-- `agent()` reuses the shared spawn core, so a `subagent_type` selects a real profile and the subagent gets that profile's configured backend, model, thinking level, prompt, optional `session_key`, and (for pi-backed profiles only) tool allow-list — not stubbed guidance. `agent({ schema })` returns a schema-validated object: pi-backed subagents receive a terminating `structured_output` tool via `createAgentSession`'s `customTools` with the profile tool allow-list extended to admit it, while Codex-backed subagents use Codex CLI `--output-schema` and Claude-backed subagents use Claude Code `--json-schema`. The first successful structured result is captured; duplicate successful calls are ignored.
+- `agent()` reuses the shared spawn core, so a `subagent_type` selects a real profile and the subagent gets that profile's configured backend, model, thinking level, prompt, optional `session_key`, and (for pi-backed profiles only) tool allow-list — not stubbed guidance. `agent({ schema })` requires a portable strict JSON Schema: every object must have `additionalProperties: false` and list all its properties in `required`; optional values use nullable types. Static schemas (inline object literals and top-level `const` references) are validated at script parse time before any subagent starts; dynamic options are validated at runtime before the requested agent launches. On a valid schema the subagent is forced to return one validated object: pi-backed subagents receive a terminating `structured_output` tool via `createAgentSession`'s `customTools` with the profile tool allow-list extended to admit it, while Codex-backed subagents use Codex CLI `--output-schema` and Claude-backed subagents use Claude Code `--json-schema`. The first successful structured result is captured; duplicate successful calls are ignored.
 - Concurrency is the SAME global cap as `Agent`: both tools share one `ConcurrencyLimiter`. Normal `Agent` calls and workflow `agent()` calls both queue and drain via `acquire`; the cap limits simultaneously running subagents, not total requested subagents. The `workflow` tool itself does not consume a slot; only its `agent()` calls do. A workflow also has hard caps on total `agent()` calls, retained logs, and orchestration-worker memory (512MB old generation by default; subagent/tool subprocess memory is not included).
 - Foreground-only still holds: the `workflow` tool blocks until the script completes. No background execution, polling, steering, or scheduling — orchestration is front-loaded into the script, not a reactive coordinator. V3 adds foreground resume-by-replay using a run journal. Per-call model/thinking *override* remains out of contract; profile-based selection via `subagent_type` is the supported path.
 - Nesting is hard-blocked for pi-backed workflow subagents: they get neither `Agent` nor `workflow`. External CLI backends use their own tool surface; this extension does not try to prevent nested/delegation features inside those CLIs.
 - Do not put exact concurrency values in the model-facing workflow prompt; say fan-out is bounded and queued.
-- Architecture: `src/core/{spawn,concurrency,model,progress,stream}.ts` is the shared core; `src/workflow/{runtime,tool,structured-output}.ts` is the workflow layer; `src/pi-subagent.ts` wires both tools and shares one limiter. Adds an `acorn` dependency (the only runtime dependency).
+- Architecture: `src/core/{spawn,concurrency,model,progress,stream}.ts` is the shared core; `src/workflow/{runtime,tool,structured-output,output-schema}.ts` is the workflow layer; `src/pi-subagent.ts` wires both tools and shares one limiter. Adds an `acorn` dependency (the only runtime dependency).
 - The throttled progress-emit + heartbeat machinery lives ONCE in `progress.ts` as `createProgressEmitter` and is shared by all three backends (`spawn.ts` pi, `codex.ts`, `claude.ts`); do not re-inline per-backend copies. The queued→running and abort emit timing is owned by that emitter.
 - External-CLI backends bound parent-side child output via `createBoundedBuffer` (`stream.ts`): stderr is capped (`MAX_STDERR_CHARS`) and a single newline-free stdout line over `MAX_STDOUT_LINE_CHARS` aborts/fails the run clearly, so one runaway subagent cannot OOM the host pi process. A clean exit (code 0) with usable final text but no recognized terminal event is accepted rather than failed, so a CLI stream-format change does not turn good runs into failures.
 
@@ -64,7 +64,7 @@
 
 - The `workflow` tool now accepts exactly one source: inline `script` for ad-hoc orchestration, `name` for a saved workflow, or `scriptPath` for a persisted script. `args` is still exposed to the script as the `args` global.
 - Saved workflow files are plain JavaScript under `~/.pi/agent/workflows/*.js` (global) and trusted `.pi/workflows/*.js` (project-local). There is no per-workflow slash command surface; the agent discovers saved workflows from the prompt roster and invokes `workflow({ name, args })` from natural language.
-- Project workflows are loaded only when `ctx.isProjectTrusted()` is true. Saved files are realpath-checked to stay inside an allowed workflow root, must end in `.js`, and are parsed with the same `export const meta = { name, description }` plus determinism-lint validator before every run. Never auto-run on discovery.
+- Project workflows are loaded only when `ctx.isProjectTrusted()` is true. Saved files are realpath-checked to stay inside an allowed workflow root, must end in `.js`, and are parsed with the same `export const meta = { name, description }` plus determinism-lint and schema-preflight validators before every run. Never auto-run on discovery.
 - Workflow identity is `meta.name`; valid saved names match lowercase letters/digits plus `_` or `-`. Project workflows override global workflows with the same name.
 - The root prompt includes a compact saved-workflow roster (`name`, `description`) when workflows exist. Put both summary and “when to use” routing guidance in `description`; do not include script bodies in the prompt.
 - Inline workflow runs auto-persist their script under the current persisted session's workflow directory and return `scriptPath`, `runId`, and `journalPath` in tool details. In-memory sessions may run without persistence.

scripts/e2e/workflow-features.mjs

@@ -118,7 +118,7 @@ const analyzed = await parallel(targets.map(function (f) {
     return agent("Read the file " + f + " in this repository and analyze it. Report the file path and the number of exported symbols.", {
       label: "analyze:" + f,
       phase: "collect",
-      schema: { type: "object", required: ["file", "exportCount"], properties: { file: { type: "string" }, exportCount: { type: "number" } } }
+      schema: { type: "object", additionalProperties: false, required: ["file", "exportCount"], properties: { file: { type: "string" }, exportCount: { type: "number" } } }
     });
   };
 }));
@@ -165,7 +165,7 @@ const flags = await parallel(files.map(function (f) {
   return function () {
     return agent("Does the file " + f + " import the 'kleur' package? Answer strictly from its source.", {
       label: "flag:" + f,
-      schema: { type: "object", required: ["file", "importsKleur"], properties: { file: { type: "string" }, importsKleur: { type: "boolean" } } }
+      schema: { type: "object", additionalProperties: false, required: ["file", "importsKleur"], properties: { file: { type: "string" }, importsKleur: { type: "boolean" } } }
     });
   };
 }));
@@ -184,7 +184,7 @@ const flags = await parallel(files.map(function (f) {
   return function () {
     return agent("Does " + f + " import the 'kleur' package? Answer strictly from its source.", {
       label: "scan:" + f,
-      schema: { type: "object", required: ["file", "importsKleur"], properties: { file: { type: "string" }, importsKleur: { type: "boolean" } } }
+      schema: { type: "object", additionalProperties: false, required: ["file", "importsKleur"], properties: { file: { type: "string" }, importsKleur: { type: "boolean" } } }
     });
   };
 }));
@@ -202,7 +202,7 @@ const ROUTE_PROBE = `export const meta = { name: "route_probe", description: "Cl
 const target = args.file;
 const c = await agent("Classify " + target + " as exactly one of: entry (a CLI entry point, e.g. reads process.argv or is declared as a bin), lib (an imported helper module), or test (a test file). Judge strictly from its source and role.", {
   label: "classify",
-  schema: { type: "object", required: ["kind"], properties: { kind: { type: "string", enum: ["entry", "lib", "test"] } } }
+  schema: { type: "object", additionalProperties: false, required: ["kind"], properties: { kind: { type: "string", enum: ["entry", "lib", "test"] } } }
 });
 let follow;
 if (c && c.kind === "entry") follow = await agent("List the command-line argument(s) " + target + " reads. Plain text only.", { label: "route:entry" });

src/prompts.ts

@@ -32,7 +32,7 @@ export const WORKFLOW_PROMPT_GUIDELINES = [
   "parallel() takes functions, not promises: `await parallel(items.map(item => () => agent('...', { label: '...' })))`. Results come back in input order.",
   "pipeline(items, ...stages) runs each item through the stages in order while different items run concurrently; each stage receives (previousValue, originalItem, index). Prefer pipeline() for multi-stage work — there is no barrier between stages. Reach for parallel() only when you genuinely need all results together, e.g. dedup or a zero-count early exit.",
   "Give each agent() a unique short `label` and pick a `subagent_type` (defaults to general-purpose) so it uses that profile's configured backend, model, thinking level, prompt, and pi-backend tool allowlist. Pass `session_key` only when you intentionally want to continue a prior subagent conversation.",
-  "Pass a JSON Schema as agent()'s `schema` option whenever the script must branch, route, filter, or aggregate on a result: the subagent is forced to return one validated object (agent() resolves to that object instead of text), so `if (r.kind === ...)` / `flags.filter(...)` are reliable. Omit `schema` for prose findings you only read or synthesize.",
+  "Pass a portable strict JSON Schema as agent()'s `schema` option whenever the script must branch, route, filter, or aggregate on a result: every object must set `additionalProperties: false`, every property must be listed in `required`, and optional values must use a nullable type. Schemas must be static object literals or top-level consts so workflow preflight can validate all of them before any subagent starts. Omit `schema` for prose findings you only read or synthesize.",
   "When `session_key` is omitted, subagents are fresh one-shot sessions with no parent context. Pi-backed subagents do not receive Agent/workflow; external CLI backends use their own tool surface. Include all needed context and paths in each fresh agent() prompt.",
   "Failed agent()/parallel()/pipeline() branches resolve to null and are logged unless the workflow is aborted; check for nulls before synthesizing.",
 ];
@@ -78,9 +78,9 @@ Inline script contract:
 Each agent() spawns a fresh one-shot subagent unless you pass \`session_key\` to create or continue a resumable child conversation. Set \`subagent_type\` to use a profile's backend, model, thinking, prompt, and pi-backend tool allowlist:
 ${formatAvailableAgents(profiles)}
 
-agent() options: \`label\` (short unique id), \`phase\` (progress group), \`subagent_type\` (profile above), \`session_key\` (caller-chosen key for a resumable child conversation), and \`schema\` (a JSON Schema). Pass \`schema\` when the script must branch, route, filter, or aggregate on the result: the subagent is forced to return one validated object and agent() resolves to that object instead of free text. Omit \`schema\` for prose findings you only synthesize. Example — classify, then dispatch:
+agent() options: \`label\` (short unique id), \`phase\` (progress group), \`subagent_type\` (profile above), \`session_key\` (caller-chosen key for a resumable child conversation), and \`schema\` (a portable strict JSON Schema). Pass \`schema\` when the script must branch, route, filter, or aggregate on the result: the subagent is forced to return one validated object and agent() resolves to that object instead of free text. Every object schema must set \`additionalProperties: false\`, list every property in \`required\`, and represent optional values with nullable types. Define schemas as static object literals or top-level consts so preflight can reject invalid schemas before any subagent starts. Omit \`schema\` for prose findings you only synthesize. Example — classify, then dispatch:
 \`\`\`
-const r = await agent("Classify " + file, { label: "classify", schema: { type: "object", required: ["kind"], properties: { kind: { type: "string", enum: ["entry", "lib", "test"] } } } });
+const r = await agent("Classify " + file, { label: "classify", schema: { type: "object", additionalProperties: false, required: ["kind"], properties: { kind: { type: "string", enum: ["entry", "lib", "test"] } } } });
 if (r.kind === "entry") { /* ... */ }
 \`\`\`
 

src/workflow/output-schema.ts

@@ -0,0 +1,125 @@
+const JSON_SCHEMA_TYPES = new Set(["array", "boolean", "integer", "null", "number", "object", "string"]);
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  if (value === null || typeof value !== "object" || Array.isArray(value)) return false;
+  const prototype = Object.getPrototypeOf(value);
+  return prototype === Object.prototype || prototype === null;
+}
+
+function appendPath(path: string, key: string): string {
+  return /^[A-Za-z_$][A-Za-z0-9_$]*$/.test(key) ? `${path}.${key}` : `${path}[${JSON.stringify(key)}]`;
+}
+
+function fail(path: string, message: string): never {
+  throw new Error(`${path}: ${message}`);
+}
+
+function assertJsonValue(value: unknown, path: string, seen: WeakSet<object>): void {
+  if (value === null || typeof value === "string" || typeof value === "boolean") return;
+  if (typeof value === "number") {
+    if (!Number.isFinite(value)) fail(path, "must contain only finite JSON numbers");
+    return;
+  }
+  if (typeof value !== "object") fail(path, "must contain only JSON values");
+  if (seen.has(value)) fail(path, "must not contain circular references");
+  seen.add(value);
+  try {
+    if (Array.isArray(value)) {
+      value.forEach((item, index) => assertJsonValue(item, `${path}[${index}]`, seen));
+      return;
+    }
+    if (!isRecord(value)) fail(path, "must contain only plain JSON objects");
+    for (const [key, child] of Object.entries(value)) {
+      assertJsonValue(child, appendPath(path, key), seen);
+    }
+  } finally {
+    seen.delete(value);
+  }
+}
+
+function schemaTypes(schema: Record<string, unknown>, path: string): string[] {
+  if (schema.type === undefined) return [];
+  const types = Array.isArray(schema.type) ? schema.type : [schema.type];
+  if (types.length === 0 || types.some((type) => typeof type !== "string" || !JSON_SCHEMA_TYPES.has(type))) {
+    fail(appendPath(path, "type"), "must be a valid JSON Schema type or non-empty array of types");
+  }
+  if (new Set(types).size !== types.length) {
+    fail(appendPath(path, "type"), "must not contain duplicate types");
+  }
+  return types as string[];
+}
+
+function schemaRecord(value: unknown, path: string): Record<string, unknown> {
+  if (!isRecord(value)) fail(path, "must be a schema object");
+  return value;
+}
+
+function validateSchemaNode(schema: Record<string, unknown>, path: string): void {
+  const types = schemaTypes(schema, path);
+  const hasObjectShape = types.includes("object") || schema.properties !== undefined;
+  if (hasObjectShape) {
+    if (!types.includes("object")) {
+      fail(appendPath(path, "type"), 'must include "object" when properties are defined');
+    }
+    if (schema.additionalProperties !== false) {
+      fail(appendPath(path, "additionalProperties"), "must be false for every object schema");
+    }
+    const propertiesPath = appendPath(path, "properties");
+    const properties = schemaRecord(schema.properties, propertiesPath);
+    const propertyNames = Object.keys(properties);
+    if (!Array.isArray(schema.required) || schema.required.some((name) => typeof name !== "string")) {
+      fail(appendPath(path, "required"), "must be an array containing every property name");
+    }
+    const required = schema.required as string[];
+    if (new Set(required).size !== required.length) {
+      fail(appendPath(path, "required"), "must not contain duplicate property names");
+    }
+    const missing = propertyNames.filter((name) => !required.includes(name));
+    if (missing.length > 0) {
+      fail(appendPath(path, "required"), `must include every property; missing: ${missing.join(", ")}`);
+    }
+    const unknown = required.filter((name) => !Object.hasOwn(properties, name));
+    if (unknown.length > 0) {
+      fail(appendPath(path, "required"), `contains names not present in properties: ${unknown.join(", ")}`);
+    }
+    for (const [name, child] of Object.entries(properties)) {
+      const childPath = appendPath(propertiesPath, name);
+      validateSchemaNode(schemaRecord(child, childPath), childPath);
+    }
+  }
+
+  if (schema.items !== undefined) {
+    validateSchemaNode(schemaRecord(schema.items, appendPath(path, "items")), appendPath(path, "items"));
+  }
+  if (schema.anyOf !== undefined) {
+    if (!Array.isArray(schema.anyOf) || schema.anyOf.length === 0) {
+      fail(appendPath(path, "anyOf"), "must be a non-empty array of schema objects");
+    }
+    schema.anyOf.forEach((child, index) => {
+      const childPath = `${appendPath(path, "anyOf")}[${index}]`;
+      validateSchemaNode(schemaRecord(child, childPath), childPath);
+    });
+  }
+  if (schema.allOf !== undefined) {
+    fail(appendPath(path, "allOf"), "allOf is not supported in the portable strict schema subset; use anyOf with nullable types instead");
+  }
+  if (schema.oneOf !== undefined) {
+    fail(appendPath(path, "oneOf"), "oneOf is not supported in the portable strict schema subset; use anyOf with nullable types instead");
+  }
+  if (schema.$defs !== undefined) {
+    const definitions = schemaRecord(schema.$defs, appendPath(path, "$defs"));
+    for (const [name, child] of Object.entries(definitions)) {
+      const childPath = appendPath(appendPath(path, "$defs"), name);
+      validateSchemaNode(schemaRecord(child, childPath), childPath);
+    }
+  }
+}
+
+export function assertPortableOutputSchema(schema: unknown): void {
+  assertJsonValue(schema, "$", new WeakSet());
+  const root = schemaRecord(schema, "$");
+  if (root.type !== "object") {
+    fail("$.type", 'root schema must have type "object"');
+  }
+  validateSchemaNode(root, "$");
+}