feat: rig/globals ambient call, pipeline null-skip, tsconfig baseUrl fix

package.json

@@ -5,6 +5,7 @@
   "exports": {
     ".": "./skills/rig/rig.ts",
     "./eslint": "./skills/rig/eslint/index.js",
+    "./globals": "./skills/rig/globals.ts",
     "./engines/anthropic": "./skills/rig/engines/anthropic.ts",
     "./engines/codex": "./skills/rig/engines/codex.ts",
     "./engines/gemini": "./skills/rig/engines/gemini.ts",

skills/rig/SKILL.md

@@ -63,6 +63,7 @@ Defaults: `name: "agent"`, `model: "small"`, `maxTurns: 4`, string input/output,
 | One-off prompt inside a workflow | `call.text(prompt)` for a string, `call.json(prompt, schema)` for structured output |
 | Reusable workflow step | Define an `agent({ input, output })` and `call(worker, input, { label, phase })` |
 | Phase or log from an agent program | Import `phase` / `log` from `rig` and call them at top level; the launcher runs every program inside a workflow |
+| Ambient `call` outside `body` | Import `call` from `"rig/globals"`; it routes through the active workflow context automatically. Do not import from `"rig/globals"` unless you need it — this avoids polluting non-workflow code. |
 | Custom model-callable operation | `defineTool(name, { description, parameters, handler })` |
 | Structured-output retries | `maxTurns` on the agent plus `addons: [repair()]` |
 | Retry with final-turn warning | `addons: [steering(), repair()]` in that order |

skills/rig/globals.ts

@@ -0,0 +1,73 @@
+/**
+ * Ambient workflow context helpers.
+ *
+ * Import from `"rig/globals"` to access `call`, `pipeline`, and `parallel`
+ * as module-level functions that automatically delegate to the active workflow
+ * run via `currentWorkflow()`.  This keeps rig programs that port from
+ * Claude dynamic workflows readable without threading context explicitly.
+ *
+ * @example
+ * ```ts
+ * import { call, pipeline } from "rig/globals";
+ * import { agent } from "rig";
+ *
+ * const worker = agent({ name: "worker", instructions: "Do work." });
+ * const results = await pipeline(inputs, (item) => call(worker, item));
+ * ```
+ *
+ * @module rig/globals
+ */
+import type {
+  AgentFn,
+  AgentInputValue,
+  InferSchema,
+  PromptBuilder,
+  Schema,
+  Workflow,
+  WorkflowCall,
+  WorkflowCallOptions,
+  WorkflowNestedOptions,
+} from "rig";
+import { currentWorkflow, parallel, pipeline } from "rig";
+
+function requireContext(label: string): WorkflowCall {
+  const ctx = currentWorkflow();
+  if (ctx === undefined) {
+    throw new Error(`${label} requires an active workflow run (call inside runWorkflow or a launcher program).`);
+  }
+  return ctx.call;
+}
+
+function callImpl<Input, Output>(
+  worker: AgentFn<Input, Output>,
+  input: AgentInputValue<Input>,
+  options?: WorkflowCallOptions,
+): Promise<Output | null> {
+  return requireContext("call()")(worker, input, options);
+}
+
+callImpl.text = (prompt: string | PromptBuilder, options?: WorkflowCallOptions): Promise<string | null> =>
+  requireContext("call.text()").text(prompt, options);
+
+callImpl.json = <const Output extends Schema>(
+  prompt: string | PromptBuilder,
+  output: Output,
+  options?: WorkflowCallOptions,
+): Promise<InferSchema<Output> | null> =>
+  requireContext("call.json()").json(prompt, output, options);
+
+callImpl.workflow = <Input, Output>(
+  child: Workflow<Input, Output>,
+  args?: Input,
+  options?: WorkflowNestedOptions,
+): Promise<Output> =>
+  requireContext("call.workflow()").workflow(child, args, options);
+
+/**
+ * Ambient workflow call.  Delegates to the active `WorkflowContext.call`,
+ * which routes through the shared concurrency limiter and agent budget.
+ * Throws if called outside a workflow run.
+ */
+export const call: WorkflowCall = callImpl as unknown as WorkflowCall;
+
+export { pipeline, parallel };

skills/rig/references/dynamic-workflows.md

@@ -99,6 +99,19 @@ outside a run. A `workflow()` default export is nested into the same run, so it
 shares the launcher's limiter, budget, and event stream instead of starting a
 second run.
 
+To use `call`, `pipeline`, and `parallel` at module scope without destructuring
+from `body`, import them from `"rig/globals"`:
+
+```ts
+import { call, pipeline } from "rig/globals";
+```
+
+These are ambient proxies that delegate to the active workflow context
+automatically.  They throw if no workflow run is active.  Prefer explicit
+`body({ call })` destructuring inside `workflow()` bodies and reserve
+`"rig/globals"` for top-level launcher programs ported from Claude dynamic
+workflows.  Do not import `"rig/globals"` unless you need it.
+
 ## Context
 
 | Member | Behavior |
@@ -135,11 +148,13 @@ brackets the child with `log` events. Restore a phase after the nested run if th
 child called `phase()`.
 
 `parallel` turns rejected thunks into `null` holes. Agent failures passed through
-`pipeline` are already `null` because `call` handles them; other pipeline callback
-errors fail the run rather than hiding programming bugs. `WorkflowLimitError` is
-never converted to `null`: exceeding `maxAgents` fails the whole run so runaway
-scheduling cannot be hidden as an ordinary worker failure. Exceptions thrown
-elsewhere in `body` also fail the run.
+`pipeline` are already `null` because `call` handles them. When a `pipeline`
+stage returns `null`, subsequent stages for that item are skipped and `null`
+propagates to the output — this prevents passing a failed result to the next
+stage. Other pipeline callback errors fail the run rather than hiding programming
+bugs. `WorkflowLimitError` is never converted to `null`: exceeding `maxAgents`
+fails the whole run so runaway scheduling cannot be hidden as an ordinary worker
+failure. Exceptions thrown elsewhere in `body` also fail the run.
 
 ## Limits
 

skills/rig/rig.ts

@@ -2309,6 +2309,7 @@ export async function pipeline<Item>(
   return Promise.all(items.map(async (item, index) => {
     let value: unknown = item;
     for (const stage of stages) {
+      if (value === null) break;
       value = await stage(value, item, index);
     }
     return value;

src/workflow.test.ts

@@ -5,6 +5,7 @@ import {
   currentWorkflow,
   log,
   phase,
+  pipeline,
   WorkflowLimitError,
   parallel,
   runWorkflow,
@@ -13,6 +14,7 @@ import {
   type WorkflowEvent,
 } from "rig";
 import { s } from "rig";
+import { call as ambientCall } from "rig/globals";
 
 function fakeAgent<Input, Output>(
   name: string,
@@ -149,6 +151,14 @@ describe("workflow primitives", () => {
     ])).resolves.toEqual([1, null, 3]);
   });
 
+  it("pipeline skips subsequent stages when a stage returns null", async () => {
+    const stage2 = vi.fn((_prev: unknown, item: number) => item);
+    await expect(
+      pipeline([1, 2, 3], (_item: number) => _item === 2 ? null : _item * 10, stage2),
+    ).resolves.toEqual([1, null, 3]);
+    expect(stage2).toHaveBeenCalledTimes(2);
+  });
+
   it("until stops on completion or repeated progress keys", async () => {
     const complete = vi.fn(async (state: number | undefined) => ({
       state: (state ?? 0) + 1,
@@ -333,3 +343,46 @@ describe("ambient workflow context", () => {
     }).not.toThrow();
   });
 });
+
+describe("rig/globals", () => {
+  it("call() delegates to the active workflow context", async () => {
+    const worker = fakeAgent<number, number>("worker", (value) => value * 2);
+    const definition = workflow({
+      meta: { name: "globals-call", description: "ambient call" },
+      body: () => ambientCall(worker, 7),
+    });
+
+    await expect(runWorkflow(definition)).resolves.toBe(14);
+  });
+
+  it("call.text() delegates to the active workflow context", async () => {
+    configureAgent(() => ({
+      ask: async () => '"pong"',
+      close: async () => {},
+    }));
+    const definition = workflow({
+      meta: { name: "globals-call-text", description: "ambient call.text" },
+      body: () => ambientCall.text("ping"),
+    });
+
+    await expect(runWorkflow(definition)).resolves.toBe("pong");
+  });
+
+  it("call.workflow() delegates to the active workflow context", async () => {
+    const child = workflow({
+      meta: { name: "child-globals", description: "child run" },
+      body: () => 42,
+    });
+    const definition = workflow({
+      meta: { name: "globals-call-workflow", description: "ambient call.workflow" },
+      body: () => ambientCall.workflow(child),
+    });
+
+    await expect(runWorkflow(definition)).resolves.toBe(42);
+  });
+
+  it("call() throws outside a workflow run", () => {
+    const worker = fakeAgent<number, number>("worker", (value) => value);
+    expect(() => ambientCall(worker, 1)).toThrow("requires an active workflow run");
+  });
+});

tsconfig.json

@@ -1,6 +1,5 @@
 {
   "compilerOptions": {
-    "baseUrl": ".",
     "target": "ES2022",
     "module": "Node16",
     "moduleResolution": "Node16",
@@ -20,6 +19,8 @@
     "noEmit": true,
     "paths": {
       "rig": ["./skills/rig/rig.ts"],
+      "rig/eslint": ["./skills/rig/eslint/index.js"],
+      "rig/globals": ["./skills/rig/globals.ts"],
       "rig/engines/anthropic": ["./skills/rig/engines/anthropic.ts"],
       "rig/engines/codex": ["./skills/rig/engines/codex.ts"],
       "rig/engines/gemini": ["./skills/rig/engines/gemini.ts"],

vitest.config.ts

@@ -5,6 +5,7 @@ export default defineConfig({
   resolve: {
     alias: [
       { find: /^rig$/, replacement: resolve(__dirname, "skills/rig/rig.ts") },
+      { find: /^rig\/globals$/, replacement: resolve(__dirname, "skills/rig/globals.ts") },
       { find: /^rig\/engines\/anthropic$/, replacement: resolve(__dirname, "skills/rig/engines/anthropic.ts") },
       { find: /^rig\/engines\/codex$/, replacement: resolve(__dirname, "skills/rig/engines/codex.ts") },
       { find: /^rig\/engines\/gemini$/, replacement: resolve(__dirname, "skills/rig/engines/gemini.ts") },