fix(providers/pi): wire env injection + harden silent-failure paths (#1296)

Four defensive fixes to the Pi community provider to match the
Claude/Codex contract and eliminate silent error swallowing.

1. envInjection now actually wired (capability was declared but unused)
   Pi's SDK has no top-level `env` option on createAgentSession, so
   per-project env vars were being dropped. Routes requestOptions.env
   through a BashSpawnHook that merges caller env over the inherited
   baseline (caller wins, matching Claude/Codex semantics). When env is
   present with no allow/deny, resolvePiTools now explicitly returns Pi's
   4 default tools so the pre-constructed default bashTool is replaced
   with an env-aware one.

2. AsyncQueue no longer leaks on consumer abort. Added close() that
   drains pending waiters with { done: true } so iterate() exits instead
   of hanging forever when the producer's finally fires before the next
   push. bridgeSession calls queue.close() in its finally block.

3. buildResultChunk no longer reports silent success when agent_end fires
   with no assistant message. Now returns { isError: true, errorSubtype:
   'missing_assistant_message' } and logs a warn event so broken Pi
   sessions don't masquerade as clean completions.

4. session-resolver no longer swallows arbitrary errors from
   SessionManager.list(). Narrowed the catch to ENOENT/ENOTDIR (the only
   "session dir doesn't exist yet" signals); permission errors, parse
   failures, and other unexpected errors now propagate.

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: coleam00

packages/providers/src/community/pi/capabilities.ts

@@ -1,13 +1,13 @@
 import type { ProviderCapabilities } from '../../types';
 
 /**
- * Pi v1 capabilities — intentionally conservative. Declared flags must reflect
+ * Pi capabilities — intentionally conservative. Declared flags must reflect
  * wired-up behavior, not potential support. The dag-executor uses these to
  * warn users when a workflow node specifies a feature the provider ignores.
  *
- * Roadmap (v2+): thinkingControl, skills, envInjection can be flipped once
- * the corresponding nodeConfig fields are intentionally translated to Pi's
- * runtime options.
+ * envInjection covers both auth-key passthrough (setRuntimeApiKey for mapped
+ * provider env vars) and bash tool subprocess env (BashSpawnHook merges the
+ * caller's env over Pi's inherited baseline), matching Claude/Codex semantics.
  */
 export const PI_CAPABILITIES: ProviderCapabilities = {
   sessionResume: true,

packages/providers/src/community/pi/event-bridge.test.ts

@@ -55,6 +55,40 @@ describe('AsyncQueue', () => {
     q[Symbol.asyncIterator]();
     expect(() => q[Symbol.asyncIterator]()).toThrow(/single-consumer/);
   });
+
+  test('close() terminates pending waiter so consumer exits loop', async () => {
+    const q = new AsyncQueue<number>();
+    const iter = q[Symbol.asyncIterator]();
+    const pending = iter.next();
+    queueMicrotask(() => q.close());
+    const result = await pending;
+    expect(result.done).toBe(true);
+  });
+
+  test('close() drains buffered items before terminating', async () => {
+    const q = new AsyncQueue<number>();
+    q.push(1);
+    q.push(2);
+    q.close();
+    const received: number[] = [];
+    for await (const n of q) received.push(n);
+    expect(received).toEqual([1, 2]);
+  });
+
+  test('push after close is a no-op (does not leak past close)', async () => {
+    const q = new AsyncQueue<number>();
+    const iter = q[Symbol.asyncIterator]();
+    q.close();
+    q.push(42); // Must not resurrect the closed queue.
+    const r = await iter.next();
+    expect(r.done).toBe(true);
+  });
+
+  test('close() is idempotent', () => {
+    const q = new AsyncQueue<number>();
+    q.close();
+    expect(() => q.close()).not.toThrow();
+  });
 });
 
 // ─── serializeToolResult ───────────────────────────────────────────────────
@@ -114,9 +148,17 @@ describe('buildResultChunk', () => {
     cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0.01 },
   };
 
-  test('returns bare result chunk if no assistant message', () => {
-    expect(buildResultChunk([])).toEqual({ type: 'result' });
-    expect(buildResultChunk([{ role: 'user', content: [] }])).toEqual({ type: 'result' });
+  test('flags isError when no assistant message is present', () => {
+    // agent_end with no assistant message in the transcript is anomalous —
+    // must surface as an error so the orchestrator doesn't treat a broken
+    // session as a clean success.
+    const expected = {
+      type: 'result',
+      isError: true,
+      errorSubtype: 'missing_assistant_message',
+    };
+    expect(buildResultChunk([])).toEqual(expected);
+    expect(buildResultChunk([{ role: 'user', content: [] }])).toEqual(expected);
   });
 
   test('extracts usage from last assistant message', () => {

packages/providers/src/community/pi/event-bridge.ts

@@ -28,15 +28,32 @@ function getLog(): ReturnType<typeof createLogger> {
  */
 export class AsyncQueue<T> implements AsyncIterable<T> {
   private readonly buffer: T[] = [];
-  private readonly waiters: ((item: T) => void)[] = [];
+  private readonly waiters: ((result: IteratorResult<T>) => void)[] = [];
   private consumed = false;
+  private closed = false;
 
   push(item: T): void {
+    if (this.closed) return;
     const waiter = this.waiters.shift();
-    if (waiter) waiter(item);
+    if (waiter) waiter({ value: item, done: false });
     else this.buffer.push(item);
   }
 
+  /**
+   * Terminate iteration cleanly. Drains any pending waiters with
+   * `{ done: true }` so the consumer exits the `for await` loop instead of
+   * hanging forever when the producer's finally block fires before a new
+   * item arrives (e.g. consumer abort mid-iteration).
+   */
+  close(): void {
+    if (this.closed) return;
+    this.closed = true;
+    while (this.waiters.length > 0) {
+      const waiter = this.waiters.shift();
+      if (waiter) waiter({ value: undefined, done: true });
+    }
+  }
+
   [Symbol.asyncIterator](): AsyncIterator<T> {
     if (this.consumed) {
       // Throw synchronously at the call site (not lazily on first .next())
@@ -56,10 +73,12 @@ export class AsyncQueue<T> implements AsyncIterable<T> {
         yield next;
         continue;
       }
-      const item = await new Promise<T>(resolve => {
+      if (this.closed) return;
+      const result = await new Promise<IteratorResult<T>>(resolve => {
         this.waiters.push(resolve);
       });
-      yield item;
+      if (result.done) return;
+      yield result.value;
     }
   }
 }
@@ -111,7 +130,12 @@ function isAssistantMessage(m: unknown): m is AssistantMessage {
 export function buildResultChunk(messages: readonly unknown[]): MessageChunk {
   const last = [...messages].reverse().find(isAssistantMessage);
   if (!last) {
-    return { type: 'result' };
+    // agent_end fired with no assistant message in the transcript. This
+    // shouldn't happen in healthy Pi runs — surface it as a loud error
+    // rather than a silent success so orchestrators don't treat a broken
+    // session as a clean completion.
+    getLog().warn('pi.event-bridge.result_missing_assistant_message');
+    return { type: 'result', isError: true, errorSubtype: 'missing_assistant_message' };
   }
 
   const tokens = usageToTokens(last.usage);
@@ -274,6 +298,10 @@ export async function* bridgeSession(
       }
     }
   } finally {
+    // Close the queue first so any producer push() still in flight becomes
+    // a no-op and pending iterate() waiters resolve — otherwise a consumer
+    // abort mid-iteration would leak this generator on the promise forever.
+    queue.close();
     unsubscribe();
     if (abortSignal) {
       abortSignal.removeEventListener('abort', onAbort);

packages/providers/src/community/pi/options-translator.test.ts

@@ -133,6 +133,23 @@ describe('resolvePiTools', () => {
     expect(result.tools).toHaveLength(1); // only 'read'
     expect(result.unknownTools).toEqual(['UnknownA', 'UnknownB']);
   });
+
+  test('no allow/deny with non-empty env → returns Pi default 4-tool set with env-aware bash', () => {
+    const result = resolvePiTools(cwd, undefined, { DATABASE_URL: 'postgres://x' });
+    expect(result.tools).toHaveLength(4); // read/bash/edit/write
+    expect(result.unknownTools).toEqual([]);
+  });
+
+  test('no allow/deny with empty env → still returns undefined (Pi defaults)', () => {
+    expect(resolvePiTools(cwd, undefined, {})).toEqual({ tools: undefined, unknownTools: [] });
+    expect(resolvePiTools(cwd, {}, {})).toEqual({ tools: undefined, unknownTools: [] });
+  });
+
+  test('env passthrough does not affect unknown tool reporting', () => {
+    const result = resolvePiTools(cwd, { allowed_tools: ['read', 'WebFetch'] }, { FOO: 'bar' });
+    expect(result.tools).toHaveLength(1);
+    expect(result.unknownTools).toEqual(['WebFetch']);
+  });
 });
 
 // ─── resolvePiSkills ───────────────────────────────────────────────────────

packages/providers/src/community/pi/options-translator.ts

@@ -11,6 +11,8 @@ import {
   createLsTool,
   createReadTool,
   createWriteTool,
+  type BashSpawnContext,
+  type BashSpawnHook,
 } from '@mariozechner/pi-coding-agent';
 import type { ThinkingLevel } from '@mariozechner/pi-ai';
 
@@ -113,13 +115,27 @@ export function resolvePiThinkingLevel(nodeConfig?: NodeConfig): ResolvedThinkin
 const PI_TOOL_NAMES = ['read', 'bash', 'edit', 'write', 'grep', 'find', 'ls'] as const;
 export type PiToolName = (typeof PI_TOOL_NAMES)[number];
 
+/**
+ * Build a Pi `spawnHook` that merges managed env vars into every bash
+ * subprocess. Matches Claude/Codex precedence: caller-provided env keys
+ * override Pi's inherited baseline. Returns undefined when `env` is empty
+ * so bash spawns without an unnecessary hook allocation.
+ */
+function buildBashSpawnHook(env: Record<string, string> | undefined): BashSpawnHook | undefined {
+  if (!env || Object.keys(env).length === 0) return undefined;
+  return (context: BashSpawnContext): BashSpawnContext => ({
+    ...context,
+    env: { ...context.env, ...env },
+  });
+}
+
 /** Map a normalized (lowercase) Pi tool name to its Pi-internal factory. */
-function buildPiTool(name: PiToolName, cwd: string): PiTool {
+function buildPiTool(name: PiToolName, cwd: string, spawnHook: BashSpawnHook | undefined): PiTool {
   switch (name) {
     case 'read':
       return createReadTool(cwd);
     case 'bash':
-      return createBashTool(cwd);
+      return spawnHook ? createBashTool(cwd, { spawnHook }) : createBashTool(cwd);
     case 'edit':
       return createEditTool(cwd);
     case 'write':
@@ -144,24 +160,51 @@ export interface ResolvedTools {
   unknownTools: string[];
 }
 
+/** Pi's default coding-tool set (mirrors `codingTools` export: read/bash/edit/write). */
+const PI_DEFAULT_TOOL_NAMES = [
+  'read',
+  'bash',
+  'edit',
+  'write',
+] as const satisfies readonly PiToolName[];
+
 /**
  * Filter Pi's built-in tool set against Archon's `allowed_tools` /
- * `denied_tools` node config.
+ * `denied_tools` node config, with managed env injected into any bash tool.
  *
  * Semantics:
- *   - neither set → return undefined (Pi's default tools)
+ *   - neither allow/deny set, no env → return undefined (Pi's default tools)
+ *   - neither allow/deny set, env present → return Pi's default 4 tools with
+ *     an env-aware bash, so codebase env vars reach bash subprocesses
  *   - allowed_tools: [] → return [] (explicit no-tools; valid Archon idiom)
  *   - allowed_tools: [X, Y] → only X, Y (normalized to lowercase)
  *   - denied_tools subtracts from allowed_tools (or full set if allowed_tools absent)
  *   - tool names not in Pi's built-in set are silently dropped but reported
  *     via `unknownTools` so the caller can surface a warning.
+ *
+ * The `env` parameter is the caller's `requestOptions.env` merged with any
+ * relevant defaults; when non-empty, it is injected into every bash spawn via
+ * a `BashSpawnHook`, matching Claude's `options.env` and Codex's constructor
+ * `env` behavior so codebase-scoped env vars reach tool subprocesses.
  */
-export function resolvePiTools(cwd: string, nodeConfig?: NodeConfig): ResolvedTools {
+export function resolvePiTools(
+  cwd: string,
+  nodeConfig?: NodeConfig,
+  env?: Record<string, string>
+): ResolvedTools {
   const allowed = nodeConfig?.allowed_tools;
   const denied = nodeConfig?.denied_tools;
+  const spawnHook = buildBashSpawnHook(env);
 
   if (allowed === undefined && denied === undefined) {
-    return { tools: undefined, unknownTools: [] };
+    // No restrictions. Match Pi's default tool set unless env injection forces
+    // a custom bash tool (Pi's default bashTool is pre-constructed with no
+    // spawnHook and there's no way to retrofit env onto it).
+    if (!spawnHook) return { tools: undefined, unknownTools: [] };
+    return {
+      tools: PI_DEFAULT_TOOL_NAMES.map(n => buildPiTool(n, cwd, spawnHook)),
+      unknownTools: [],
+    };
   }
 
   const knownSet = new Set<PiToolName>(PI_TOOL_NAMES);
@@ -199,7 +242,7 @@ export function resolvePiTools(cwd: string, nodeConfig?: NodeConfig): ResolvedTo
   });
 
   return {
-    tools: unique.map(n => buildPiTool(n, cwd)),
+    tools: unique.map(n => buildPiTool(n, cwd, spawnHook)),
     unknownTools,
   };
 }

packages/providers/src/community/pi/provider.test.ts

@@ -93,7 +93,7 @@ const MockDefaultResourceLoader = mock(function (_opts: unknown) {
 // Tool factory mocks — each returns an opaque object tagged with the tool
 // name so assertions can verify which tools the provider selected.
 const mockCreateReadTool = mock((_cwd: string) => ({ __piTool: 'read' }));
-const mockCreateBashTool = mock((_cwd: string) => ({ __piTool: 'bash' }));
+const mockCreateBashTool = mock((_cwd: string, _options?: unknown) => ({ __piTool: 'bash' }));
 const mockCreateEditTool = mock((_cwd: string) => ({ __piTool: 'edit' }));
 const mockCreateWriteTool = mock((_cwd: string) => ({ __piTool: 'write' }));
 const mockCreateGrepTool = mock((_cwd: string) => ({ __piTool: 'grep' }));
@@ -765,6 +765,80 @@ describe('PiProvider', () => {
     expect('tools' in callArgs).toBe(false);
   });
 
+  test('requestOptions.env with no tool restrictions overrides Pi defaults with env-aware bash', async () => {
+    process.env.GEMINI_API_KEY = 'sk-test';
+    resetScript(scriptedAgentEnd());
+
+    await consume(
+      new PiProvider().sendQuery('hi', '/tmp', undefined, {
+        model: 'google/gemini-2.5-pro',
+        env: { DATABASE_URL: 'postgres://managed' },
+      })
+    );
+
+    const [callArgs] = mockCreateAgentSession.mock.calls[0] as [Record<string, unknown>];
+    // Env present → we override Pi's built-in codingTools so bash sees the env.
+    const tools = callArgs.tools as Array<{ __piTool: string }>;
+    expect(Array.isArray(tools)).toBe(true);
+    expect(tools.map(t => t.__piTool).sort()).toEqual(['bash', 'edit', 'read', 'write']);
+
+    const bashCall = mockCreateBashTool.mock.calls.find(call => call[1] !== undefined);
+    expect(bashCall).toBeDefined();
+    const bashOptions = bashCall![1] as { spawnHook: (c: unknown) => unknown };
+    expect(typeof bashOptions.spawnHook).toBe('function');
+
+    // The spawnHook must merge caller env OVER Pi's inherited baseline, matching
+    // Claude's { ...subprocessEnv, ...requestOptions.env } and Codex's buildCodexEnv.
+    const merged = bashOptions.spawnHook({
+      command: 'echo',
+      cwd: '/tmp',
+      env: { PATH: '/usr/bin', DATABASE_URL: 'postgres://stale' },
+    }) as { env: Record<string, string> };
+    expect(merged.env.PATH).toBe('/usr/bin');
+    expect(merged.env.DATABASE_URL).toBe('postgres://managed');
+  });
+
+  test('requestOptions.env threads through to bash tool when allowed_tools includes bash', async () => {
+    process.env.GEMINI_API_KEY = 'sk-test';
+    resetScript(scriptedAgentEnd());
+
+    await consume(
+      new PiProvider().sendQuery('hi', '/tmp', undefined, {
+        model: 'google/gemini-2.5-pro',
+        nodeConfig: { allowed_tools: ['read', 'bash'] },
+        env: { STRIPE_KEY: 'sk_test_abc' },
+      })
+    );
+
+    const bashCall = mockCreateBashTool.mock.calls.find(call => call[1] !== undefined);
+    expect(bashCall).toBeDefined();
+    const bashOptions = bashCall![1] as { spawnHook: (c: unknown) => unknown };
+    const merged = bashOptions.spawnHook({
+      command: 'echo',
+      cwd: '/tmp',
+      env: { PATH: '/usr/bin' },
+    }) as { env: Record<string, string> };
+    expect(merged.env.STRIPE_KEY).toBe('sk_test_abc');
+    expect(merged.env.PATH).toBe('/usr/bin');
+  });
+
+  test('empty requestOptions.env does NOT construct a spawnHook', async () => {
+    process.env.GEMINI_API_KEY = 'sk-test';
+    resetScript(scriptedAgentEnd());
+
+    await consume(
+      new PiProvider().sendQuery('hi', '/tmp', undefined, {
+        model: 'google/gemini-2.5-pro',
+        env: {},
+      })
+    );
+
+    // Every createBashTool call in this test path is either (cwd) or (cwd, undefined).
+    for (const call of mockCreateBashTool.mock.calls) {
+      expect(call[1]).toBeUndefined();
+    }
+  });
+
   test('requestOptions.systemPrompt threads through to DefaultResourceLoader', async () => {
     process.env.GEMINI_API_KEY = 'sk-test';
     resetScript(scriptedAgentEnd());

packages/providers/src/community/pi/provider.ts

@@ -169,7 +169,14 @@ export class PiProvider implements IAgentProvider {
     //    4b. tools: covers allowed_tools / denied_tools. `undefined` leaves Pi
     //        defaults; an explicit empty array means "no tools" (valid idiom
     //        matching e2e-claude-smoke's `allowed_tools: []`).
-    const { tools: filteredTools, unknownTools } = resolvePiTools(cwd, nodeConfig);
+    //        requestOptions.env (codebase-scoped env vars from .archon/config.yaml)
+    //        is injected into bash subprocesses via a BashSpawnHook, mirroring
+    //        Claude's options.env and Codex's constructor env.
+    const { tools: filteredTools, unknownTools } = resolvePiTools(
+      cwd,
+      nodeConfig,
+      requestOptions?.env
+    );
     if (unknownTools.length > 0) {
       yield {
         type: 'system',