feat: authoring skill, script iterate loop, and pi 1.3 APIs for workflows

.pi/extensions/pi-workflow-engine/src/agent-runner-types.ts

@@ -38,10 +38,12 @@ export type CreateAgentSession = (options: CreateAgentSessionOptions) => Promise
 /** Every row update after `agentQueued` addresses that row by its id; labels are display text, not unique keys. */
 export interface AgentProgress {
   agentQueued(phase: string | undefined, label: string): number;
-  agentStart(id: number): void;
+  /** `skip` aborts the row's live agent; the tracker offers it to the inspector only while the row runs. */
+  agentStart(id: number, skip?: () => void): void;
   agentTool(id: number, tool: string): void;
   agentDone(id: number): void;
   agentFailed(id: number, error: unknown): void;
+  agentSkipped(id: number): void;
   event(event: WorkflowProgressEvent): void;
   log(message: string): void;
 }

.pi/extensions/pi-workflow-engine/src/agent-runner.ts

@@ -1,6 +1,6 @@
 import { assertWorkflowBudgetAvailable } from "./budget.ts";
 import { WorkflowAgentTimeoutError } from "./agent-limits.ts";
-import { abortReason, linkAbortSignal, throwIfAborted } from "./cancellation.ts";
+import { abortReason, linkAbortSignal, throwIfAborted, WorkflowAgentSkippedError } from "./cancellation.ts";
 import { executeAgentAttempt } from "./agent-attempt.ts";
 import {
   createAgentReplayPlan,
@@ -66,7 +66,7 @@ export async function runAgent(
           const agentRc: RunContext = { ...rc, signal: liveScope.signal };
           throwIfAborted(agentRc.signal);
           if (!isReplayEnabled(replay)) assertWorkflowBudgetAvailable(agentRc.budget);
-          rc.progress.agentStart(rowId);
+          rc.progress.agentStart(rowId, liveScope.skip);
           if (replay.kind === "disabled") {
             rc.progress.log(`${label}: resume disabled for this call (${replay.reason})`);
           }
@@ -107,6 +107,8 @@ export async function runAgent(
               await agentRc.retryScheduler.sleep(delayMs, agentRc.signal);
               continue;
             }
+            // A skip that lands as the attempt finishes still wins, so its result is neither returned nor journaled.
+            liveScope.throwIfSkipped();
             const settlement = await settleAgentAttempt({ rc: agentRc, label, outcome });
             if (settlement.kind === "retry-live") {
               attemptPlan = { kind: "off" };
@@ -125,8 +127,13 @@ export async function runAgent(
       let failure = error;
       if (liveScope.signal.aborted) failure = abortReason(liveScope.signal);
       if (rc.signal?.aborted) failure = abortReason(rc.signal);
-      rc.progress.agentFailed(rowId, failure);
-      rc.progress.log(`${label} failed: ${unknownErrorMessage(failure)}`);
+      if (failure instanceof WorkflowAgentSkippedError) {
+        rc.progress.agentSkipped(rowId);
+        rc.progress.log(`${label} skipped`);
+      } else {
+        rc.progress.agentFailed(rowId, failure);
+        rc.progress.log(`${label} failed: ${unknownErrorMessage(failure)}`);
+      }
       throw failure;
     } finally {
       liveScope.dispose();
@@ -151,6 +158,13 @@ function createAgentLiveScope(rc: RunContext, label: string) {
         rc.agentTimeoutMs,
       );
     },
+    /** Aborts only this agent; the run and its siblings carry on. */
+    skip() {
+      controller.abort(new WorkflowAgentSkippedError(label));
+    },
+    throwIfSkipped() {
+      if (controller.signal.reason instanceof WorkflowAgentSkippedError) throw controller.signal.reason;
+    },
     dispose() {
       if (timer) clearTimeout(timer);
       unlink();

.pi/extensions/pi-workflow-engine/src/agent-session.ts

@@ -12,6 +12,7 @@ import {
   type ToolDefinition,
 } from "@earendil-works/pi-coding-agent";
 import { prepareAgentSkillResources } from "./agent-skills.ts";
+import { createCodemodeSlotExtension, runToolCallsSequentially } from "./codemode-slots.ts";
 import type {
   AgentExecutionOptions,
   AgentProgress,
@@ -35,6 +36,16 @@ export const FINAL_TOOL = "final_answer";
 /** pi's codemode tool; SDK sessions only have it when their resource loader registers the extension. */
 export const CODEMODE_TOOL = "codemode";
 
+/** pi's stock prompt frames a chat with a person; a subagent's last message is program data instead. */
+const RETURN_VALUE_FRAMING =
+  "Your final message is returned verbatim to the workflow program that started you, as data. " +
+  "It is not shown to a person as a chat reply. Make it exactly the output your task asks for, in the format it asks for, " +
+  "with no greeting, preamble, narration of your process, or offer of further help.";
+/** A schema agent returns its final_answer arguments, not its last message. */
+const STRUCTURED_RETURN_VALUE_FRAMING =
+  `Your ${FINAL_TOOL} arguments are returned to the workflow program that started you, as data. ` +
+  "They are not shown to a person as a chat reply. Fill each field with exactly the content it asks for, " +
+  "with no greeting, preamble, narration of your process, or offer of further help.";
 const SCHEMA_REPROMPT =
   `You ended your turn without calling the ${FINAL_TOOL} tool, so no result was recorded. ` +
   `Call ${FINAL_TOOL} now with your final answer as its arguments. Do not reply with plain text.`;
@@ -234,6 +245,7 @@ async function prepareAgentSessionResources(input: {
     sessionManager: SessionManager.inMemory(cwd),
   });
   const createSession = rc.createSession;
+  // Injected test sessions build no resource loader, so they get neither skills nor the return-value framing.
   if (createSession) {
     return {
       selectedSkills: [],
@@ -251,9 +263,15 @@ async function prepareAgentSessionResources(input: {
     log: (message: string) => rc.progress.log(`${label}: ${message}`),
   };
   const preparedSkills = prepareAgentSkillResources(skillOptions);
+  const codemode = opts.tools?.includes(CODEMODE_TOOL) ?? false;
   const services = await createAgentSessionServices({
     cwd,
-    resourceLoaderOptions: { ...preparedSkills.resourceLoaderOptions, ...codemodeExtension(opts) },
+    resourceLoaderOptions: {
+      ...preparedSkills.resourceLoaderOptions,
+      ...(codemode ? codemodeExtensions() : {}),
+      // An override, not `appendSystemPrompt`: passing that replaces pi's discovery of the user's APPEND_SYSTEM.md.
+      appendSystemPromptOverride: (base) => [...base, opts.schema ? STRUCTURED_RETURN_VALUE_FRAMING : RETURN_VALUE_FRAMING],
+    },
   });
   await synchronizeWorkflowModelRuntime({
     host: rc.modelRegistry,
@@ -272,22 +290,31 @@ async function prepareAgentSessionResources(input: {
   const selectedSkills = preparedSkills.resolve(services.resourceLoader);
   return {
     selectedSkills,
-    createSession: (sessionOptions) =>
-      createAgentSessionFromServices({
+    createSession: async (sessionOptions) => {
+      const created = await createAgentSessionFromServices({
         services,
         ...commonSessionOptions(sessionOptions),
-      }),
+      });
+      if (codemode) runToolCallsSequentially(created.session);
+      return created;
+    },
   };
 }
 
 /**
  * SDK sessions load none of pi's built-in extensions, so codemode is registered only for agents whose
  * allowlist names it. Scripts can call just the agent's other allowed tools. Mode "on" keeps those
  * tools declared whatever the user's `codemode.mode`, so the author's allowlist stays what the model sees.
+ * Its calls queue for the process-wide codemode slots instead of failing at pi's worker limit, so the
+ * session runs its tool calls one at a time.
  */
-function codemodeExtension(opts: AgentExecutionOptions): Pick<SessionResourceLoaderOptions, "extensionFactories"> {
-  if (!opts.tools?.includes(CODEMODE_TOOL)) return {};
-  return { extensionFactories: [{ name: CODEMODE_TOOL, factory: createCodemodeExtension({ mode: "on" }), replaceable: true }] };
+function codemodeExtensions(): Pick<SessionResourceLoaderOptions, "extensionFactories"> {
+  return {
+    extensionFactories: [
+      { name: CODEMODE_TOOL, factory: createCodemodeExtension({ mode: "on" }), replaceable: true },
+      { name: "workflow-codemode-slots", factory: createCodemodeSlotExtension(CODEMODE_TOOL), hidden: true },
+    ],
+  };
 }
 
 type SessionResourceLoaderOptions = NonNullable<CreateAgentSessionServicesOptions["resourceLoaderOptions"]>;

.pi/extensions/pi-workflow-engine/src/background-workflow-tool.ts

@@ -1,8 +1,10 @@
+import { join } from "node:path";
 import type { ExtensionContext } from "@earendil-works/pi-coding-agent";
 import { BackgroundWorkflowCoordinator, backgroundOrigin } from "./background-workflows.ts";
-import { createWorkflowRunId } from "./journal.ts";
+import { createWorkflowRunId, WORKFLOW_RUNS_DIR } from "./journal.ts";
 import type { ResolvedWorkflowRunOptions } from "./options.ts";
 import { unknownErrorMessage } from "./unknown-error.ts";
+import { workflowRunScriptFile } from "./workflow-run-record.ts";
 
 export interface BackgroundWorkflowToolResult {
   readonly content: Array<{ readonly type: "text"; readonly text: string }>;
@@ -15,6 +17,8 @@ export interface BackgroundWorkflowToolStartInput {
   readonly ctx: ExtensionContext;
   readonly name: string;
   readonly options: ResolvedWorkflowRunOptions;
+  /** Deliver the completion with a new agent turn instead of silently. */
+  readonly wakeOnComplete?: boolean;
   readonly execute: (ctx: ExtensionContext, options: ResolvedWorkflowRunOptions) => Promise<void>;
 }
 
@@ -39,8 +43,9 @@ export async function startBackgroundWorkflowTool(
     inspect: false,
     resultViewer: "skip",
     runId,
-    background: backgroundOrigin(input.ctx),
+    background: backgroundOrigin(input.ctx, Date.now(), input.wakeOnComplete),
   };
+  let scriptPersisted = false;
   try {
     await input.coordinator.start({
       ctx: input.ctx,
@@ -53,6 +58,7 @@ export async function startBackgroundWorkflowTool(
             ...backgroundOptions,
             signal,
             onRunMetadata(metadata) {
+              scriptPersisted = metadata.scriptPath !== undefined;
               onStarted();
               return backgroundOptions.onRunMetadata?.(metadata);
             },
@@ -71,7 +77,7 @@ export async function startBackgroundWorkflowTool(
   return {
     content: [{
       type: "text",
-      text: `Background workflow "${input.name}" started.\nRun ID: ${runId}\nDurable record: .pi/.workflow-runs/${runId}.run.json`,
+      text: `Background workflow "${input.name}" started.\nRun ID: ${runId}\nDurable record: .pi/.workflow-runs/${runId}.run.json${scriptPersisted ? `\nScript: ${join(WORKFLOW_RUNS_DIR, workflowRunScriptFile(runId))}` : ""}`,
     }],
     details: { background: true, state: "running", name: input.name, runId },
   };

.pi/extensions/pi-workflow-engine/src/background-workflows.ts

@@ -1,6 +1,8 @@
+import { join } from "node:path";
 import { SessionManager, type ExtensionAPI, type ExtensionContext } from "@earendil-works/pi-coding-agent";
 import { WorkflowAbortError, WorkflowPauseError } from "./cancellation.ts";
 import { isRecord } from "./guards.ts";
+import { WORKFLOW_RUNS_DIR } from "./journal.ts";
 import type { WorkflowBackgroundOrigin } from "./types.ts";
 import {
   stopWorkflowRunRecord,
@@ -9,9 +11,11 @@ import {
   type WorkflowRunState,
 } from "./workflow-run-record.ts";
 import { updateWorkflowRunDelivery, type WorkflowRunDelivery } from "./workflow-run-background.ts";
-import { ProjectWorkflowRunStore, type WorkflowRunStore } from "./workflow-run-store.ts";
+import { ProjectWorkflowRunStore, WORKFLOW_RUN_RECORD_SUFFIX, type WorkflowRunStore } from "./workflow-run-store.ts";
 import { truncateText } from "./text.ts";
+import { formatWorkflowUsageLine } from "./usage.ts";
 import { workflowResultSummary } from "./workflow-execution.ts";
+import { formatResultForContext, formatWorkflowRunFileLines } from "./workflow-result-context.ts";
 import { unknownErrorMessage } from "./unknown-error.ts";
 
 const BACKGROUND_DELIVERY_CUSTOM_TYPE = "workflow-result";
@@ -294,20 +298,26 @@ export class BackgroundWorkflowCoordinator {
       this.pi.sendMessage(
         {
           customType: BACKGROUND_DELIVERY_CUSTOM_TYPE,
-          content: formatBackgroundDelivery(details),
+          content: formatBackgroundDelivery(record, details),
           display: true,
           details,
         },
-        { triggerTurn: false },
+        // Delivery waits for an idle session, and pi steers a message sent while a run starts into that run.
+        { triggerTurn: wakesAgent(record) },
       );
     }
     await markDelivery(store, runId, { state: "delivered", deliveredAt: Date.now() });
     return true;
   }
 }
 
-export function backgroundOrigin(ctx: Pick<ExtensionContext, "sessionManager">, requestedAt = Date.now()): WorkflowBackgroundOrigin {
-  return { sessionId: ctx.sessionManager.getSessionId(), requestedAt };
+export function backgroundOrigin(
+  ctx: Pick<ExtensionContext, "sessionManager">,
+  requestedAt = Date.now(),
+  wakeOnComplete = false,
+): WorkflowBackgroundOrigin {
+  const origin = { sessionId: ctx.sessionManager.getSessionId(), requestedAt };
+  return wakeOnComplete ? { ...origin, wakeOnComplete } : origin;
 }
 
 export function backgroundResultDetails(record: WorkflowRunRecord): BackgroundWorkflowResultDetails {
@@ -336,17 +346,41 @@ function backgroundSummary(record: WorkflowRunRecord): string {
   return summary === undefined ? "Workflow completed. Open run history for the retained result." : truncateText(summary, SUMMARY_LIMIT);
 }
 
-function formatBackgroundDelivery(details: BackgroundWorkflowResultDetails): string {
+/**
+ * The host model only sees this text, never `details`. A woken agent acts on it at once, so it gets the
+ * result as the synchronous tool path renders it; a silent delivery keeps the summary. Both name the run's
+ * files so a later turn can read the retained result.
+ */
+function formatBackgroundDelivery(record: WorkflowRunRecord, details: BackgroundWorkflowResultDetails): string {
+  const woken = record.background?.origin.wakeOnComplete === true;
+  const usage = woken && record.state === "completed" ? formatWorkflowUsageLine(record.usage) : undefined;
   return [
     `## Background workflow: ${details.name}`,
     "",
-    `Run ID: ${details.runId}`,
+    // A resume, including an automatic one, has a new run id; name the run the agent started.
+    details.resumedFromRunId ? `Run ID: ${details.runId} (resumed from ${details.resumedFromRunId})` : `Run ID: ${details.runId}`,
     `State: ${details.status}`,
     "",
-    details.summary,
+    woken ? wokenDeliveryBody(record, details.summary) : details.summary,
+    "",
+    ...backgroundRunFileLines(record),
+    ...(usage === undefined ? [] : [usage]),
   ].join("\n");
 }
 
+function wokenDeliveryBody(record: WorkflowRunRecord, summary: string): string {
+  if (record.state !== "completed") return `Workflow ${record.state}: ${record.message}`;
+  return record.result.kind === "value" ? formatResultForContext(record.result.value) : summary;
+}
+
+/** The retained record, script and journal, relative to the workflow cwd like the synchronous result's lines. */
+function backgroundRunFileLines(record: WorkflowRunRecord): string[] {
+  return [
+    `Record: ${join(WORKFLOW_RUNS_DIR, `${record.runId}${WORKFLOW_RUN_RECORD_SUFFIX}`)}`,
+    ...formatWorkflowRunFileLines(record.runId, record.scriptFile !== undefined),
+  ];
+}
+
 function sessionHasBackgroundDelivery(ctx: ExtensionContext, runId: string): boolean {
   for (const entry of ctx.sessionManager.getEntries()) {
     if (entry.type !== "message" || entry.message.role !== "custom") continue;
@@ -361,6 +395,15 @@ function isPendingBackgroundOutcome(record: WorkflowRunRecord): record is Workfl
   return record.background?.delivery.state === "pending" && isDeliverableState(record.state);
 }
 
+/**
+ * A run started under Dynamax wakes the agent to act on its result; others arrive silently. So does a run
+ * the user stopped, or one paused with a resume already scheduled, whose successor wakes it instead.
+ */
+function wakesAgent(record: WorkflowRunRecord): boolean {
+  if (record.background?.origin.wakeOnComplete !== true || record.state === "stopped") return false;
+  return !(record.state === "paused" && record.pause?.autoResume === true);
+}
+
 function isDeliverableState(state: WorkflowRunState): state is "completed" | "failed" | "stopped" | "paused" {
   return state === "completed" || state === "failed" || state === "stopped" || state === "paused";
 }

.pi/extensions/pi-workflow-engine/src/cancellation.ts

@@ -13,6 +13,20 @@ export class WorkflowPauseError extends WorkflowAbortError {
   }
 }
 
+/**
+ * Abort reason for one agent the user skipped from the inspector. It deliberately does not extend
+ * WorkflowAbortError: the run carries on, so `parallel()`/`pipeline()` turn the skipped branch into a
+ * `null` slot while a direct `await agent()` sees this error.
+ */
+export class WorkflowAgentSkippedError extends Error {
+  override readonly name = "WorkflowAgentSkippedError";
+  readonly code = "WORKFLOW_AGENT_SKIPPED";
+
+  constructor(readonly label: string) {
+    super(`Workflow agent "${label}" was skipped by the user.`);
+  }
+}
+
 export function abortReason(signal: AbortSignal | undefined): Error {
   const reason = signal?.reason;
   if (safeInstanceOf(reason, Error)) return reason;