Fix Pi research timeouts when runtime or tools are unavailable

apps/electron/src/main/kernelHost.ts

@@ -84,6 +84,7 @@ import {
   CapabilityExecutor,
   computeSkillReadiness,
   createAgentEvaluator,
+  createAgentSynthesizer,
   createFullRegistry,
   createLocalThesisEvaluator,
   createRouterFetchers,
@@ -364,7 +365,12 @@ export class AgentKernelHost {
       synthesizer:
         provider === 'local'
           ? new LocalResearchSynthesizer()
-          : { synthesize: this.runResearchSynthesis },
+          : createAgentSynthesizer(this.runResearchSynthesis, {
+              onFallback: (error) => mainErrorLog.push({
+                at: Date.now(), source: 'main', stack: null,
+                message: 'Research used local synthesis: ' + redactSecrets(error instanceof Error ? error.message : String(error)),
+              }),
+            }),
       repository: new ResearchReportRepository(new JsonFileStore(join(userData, 'store'))),
       getIdentity: async () => {
         const state = provider === 'local' ? undefined : await this.requireLlm().getState();

packages/shared/src/agent/pi-rpc-client.ts

@@ -1,4 +1,6 @@
 import { randomUUID } from 'node:crypto';
+import { existsSync } from 'node:fs';
+import { join } from 'node:path';
 import { spawn, type ChildProcessWithoutNullStreams, type SpawnOptionsWithoutStdio } from 'node:child_process';
 import type { AgentTraceEvent, LlmModel, ToolCallRecord } from '@finagent/core';
 import { createCodeError } from './errors.ts';
@@ -351,7 +353,10 @@ export class PiRpcClient {
       try {
         proc = await spawnPromise;
       } catch (error) {
-        pending.reject(error);
+        // Spawn failures must settle and remove this prompt immediately. If we
+        // only reject the iterator, the old request timer can remain active and
+        // turn a missing runtime into a misleading 60-second timeout.
+        self.finishPrompt(pending, error);
         throw error;
       }
       self.writeJson(proc, { type: 'prompt', id, message: content }, pending);
@@ -781,8 +786,18 @@ function parseArgs(value: string | undefined) {
   return value.split(' ').map((part) => part.trim()).filter(Boolean);
 }
 
-function readDefaultPiCommand() {
-  return process.env.FINAGENT_PI_COMMAND ?? 'bunx';
+function readDefaultPiCommand(env: NodeJS.ProcessEnv = process.env, platform = process.platform) {
+  const configured = env.FINAGENT_PI_COMMAND;
+  if (configured) return configured;
+  if (platform !== 'win32') return 'bunx';
+
+  // Desktop launches do not inherit start-trail.cmd's PATH edits. Resolve the
+  // Bun shim directly so a normal shortcut/installer launch can still start Pi.
+  const candidates = [
+    env.APPDATA ? join(env.APPDATA, 'npm', 'node_modules', 'bun', 'bin', 'bunx.exe') : undefined,
+    env.LOCALAPPDATA ? join(env.LOCALAPPDATA, 'Programs', 'bun', 'bunx.exe') : undefined,
+  ].filter((candidate): candidate is string => Boolean(candidate));
+  return candidates.find((candidate) => existsSync(candidate)) ?? 'bunx';
 }
 
 /**

packages/shared/src/agent/pi-runtime-agent-backend.test.ts

@@ -1,7 +1,8 @@
 import { EventEmitter } from 'node:events';
 import { PassThrough } from 'node:stream';
-import { mkdir, writeFile } from 'node:fs/promises';
-import { join } from 'node:path';
+import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { dirname, join, resolve } from 'node:path';
 import { afterEach, describe, expect, it } from 'bun:test';
 import type { SpawnOptionsWithoutStdio } from 'node:child_process';
 import { PiRuntimeAdapter } from './pi-runtime-adapter.ts';
@@ -107,7 +108,8 @@ describe('PiRpcClient', () => {
   it('uses the project default pi runtime command and args', () => {
     process.env.ANTHROPIC_MODEL = 'MiniMax-M2.7';
 
-    expect(readDefaultPiCommand()).toBe('bunx');
+    expect(readDefaultPiCommand({}, 'linux')).toBe('bunx');
+    expect(readDefaultPiCommand({}, 'win32')).toBe('bunx');
     expect(readDefaultPiArgs()).toEqual([
       '@mariozechner/pi-coding-agent',
       '--mode',
@@ -121,6 +123,42 @@ describe('PiRpcClient', () => {
     ]);
   });
 
+  it('resolves a native Windows Bun shim without a launcher PATH override', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'trail-pi-command-'));
+    const native = join(root, 'npm', 'node_modules', 'bun', 'bin', 'bunx.exe');
+    try {
+      await mkdir(dirname(native), { recursive: true });
+      await writeFile(native, 'test executable placeholder');
+      expect(readDefaultPiCommand({ APPDATA: root }, 'win32')).toBe(native);
+      expect(readDefaultPiCommand({ APPDATA: root, FINAGENT_PI_COMMAND: 'custom-pi' }, 'win32')).toBe('custom-pi');
+    } finally {
+      if (dirname(resolve(root)) !== resolve(tmpdir()) || !root.includes('trail-pi-command-')) {
+        throw new Error('Refusing to remove an unexpected fixture directory.');
+      }
+      await rm(root, { recursive: true, force: true });
+    }
+  });
+
+  it('removes a failed streaming spawn before the next id-less reply arrives', async () => {
+    let starts = 0;
+    const client = new PiRpcClient({
+      requestTimeoutMs: 100,
+      spawnProcess: createSpawn(() => {
+        if (++starts === 1) throw Object.assign(new Error('missing runtime'), { code: 'ENOENT' });
+        return new FakePiProcess((line, proc) => {
+          if (line.type === 'prompt') proc.writeEvent({
+            type: 'agent_end', messages: [{ role: 'assistant', content: [{ type: 'text', text: 'recovered' }] }],
+          });
+        });
+      }),
+    });
+    try {
+      const stream = client.promptStreaming('first');
+      await expect(stream[Symbol.asyncIterator]().next()).rejects.toMatchObject({ code: 'PI_RUNTIME_NOT_FOUND' });
+      await expect(client.prompt('second')).resolves.toMatchObject({ answer: 'recovered' });
+    } finally { await client.dispose(); }
+  });
+
   it('honors explicit FINAGENT_PI_COMMAND and FINAGENT_PI_ARGS overrides', () => {
     process.env.FINAGENT_PI_COMMAND = 'pi';
     process.env.FINAGENT_PI_ARGS = '--mode rpc --provider openai';

packages/shared/src/capabilities/executor.test.ts

@@ -62,6 +62,19 @@ describe('CapabilityExecutor.run', () => {
     expect(record.status).toBe('unavailable');
   });
 
+  it('maps missing account permissions to unavailable', async () => {
+    const executor = new CapabilityExecutor();
+    const cap = makeCap('portfolio.positions', 'get_positions', async () => {
+      throw Object.assign(new Error('Longbridge account scope is missing'), {
+        code: 'PERMISSION_DENIED',
+      });
+    });
+
+    const { record } = await executor.run(cap, {});
+
+    expect(record.status).toBe('unavailable');
+  });
+
   it('records cancelled when a capability exceeds its timeout', async () => {
     const executor = new CapabilityExecutor();
     const cap = makeCap('market.quote', 'get_quote', async (_input, ctx) => {

packages/shared/src/capabilities/executor.ts

@@ -16,6 +16,10 @@ const UNAVAILABLE_ERROR_CODES = {
   LONGBRIDGE_NOT_AUTHED: true,
   LONGBRIDGE_RATE_LIMITED: true,
   LONGBRIDGE_TIMEOUT: true,
+  LONGBRIDGE_PERMISSION_DENIED: true,
+  PERMISSION_DENIED: true,
+  AUTH_EXPIRED: true,
+  CONFIG_MISSING: true,
 } as const;
 
 export interface CapabilityExecutorOptions {

packages/shared/src/capabilities/pi-tools.test.ts

@@ -118,4 +118,84 @@ describe('createCapabilityTools', () => {
       expect.objectContaining({ title: expect.stringContaining('Revenue rises') }),
     ]);
   });
+
+  it('returns an unavailable tool result when a provider call fails', async () => {
+    const [tool] = createCapabilityTools([{
+      id: 'market.quote',
+      name: 'Quote',
+      description: 'Read a quote.',
+      category: 'market',
+      riskLevel: 'read',
+      auth: 'public',
+      toolName: 'get_quote',
+      inputSchema: Type.Object({}),
+      async execute() {
+        throw Object.assign(new Error('account scope is missing'), {
+          code: 'PERMISSION_DENIED',
+        });
+      },
+    }]);
+
+    const result = await tool!.execute('tool-1', {}, new AbortController().signal);
+    expect(result.details).toMatchObject({ status: 'unavailable', code: 'PERMISSION_DENIED' });
+    expect(result.content[0]!.text).toContain('[TOOL_UNAVAILABLE]');
+    expect(result.content[0]!.text).toContain('remaining tools');
+  });
+
+  it('keeps cancellation terminal instead of converting it into a data gap', async () => {
+    const controller = new AbortController();
+    controller.abort(new Error('cancelled'));
+    const [tool] = createCapabilityTools([{
+      id: 'market.quote',
+      name: 'Quote',
+      description: 'Read a quote.',
+      category: 'market',
+      riskLevel: 'read',
+      auth: 'public',
+      toolName: 'get_quote',
+      inputSchema: Type.Object({}),
+      async execute() {
+        return {
+          data: { price: 1 },
+          provenance: { provider: 'test', fetchedAt: 1, stale: false },
+        };
+      },
+    }]);
+
+    await expect(tool!.execute('tool-1', {}, controller.signal)).rejects.toThrow('cancelled');
+  });
+  it('times out an unresponsive tool and allows the next tool to execute', async () => {
+    let childSignal: AbortSignal | undefined;
+    const cap = defineCapability({
+      id: 'market.quote', name: 'Quote', description: 'Read a quote.',
+      category: 'market', riskLevel: 'read', auth: 'public', toolName: 'get_quote',
+      inputSchema: Type.Object({}),
+      async execute(_input, ctx) {
+        childSignal = ctx?.signal;
+        return await new Promise<never>(() => {});
+      },
+    });
+    const [slow, next] = createCapabilityTools([cap, {
+      ...cap, id: 'market.status', toolName: 'get_market_status',
+      async execute() { return { data: ['closed'], provenance: { provider: 'test', fetchedAt: 1, stale: false } }; },
+    }], { timeoutMs: 5 });
+    const result = await slow.execute('slow', {}, new AbortController().signal);
+    expect(result.details).toMatchObject({ status: 'unavailable', code: 'CAPABILITY_TIMEOUT' });
+    expect(childSignal?.aborted).toBe(true);
+    expect((await next.execute('next', {}, new AbortController().signal)).details).toEqual(['closed']);
+  });
+
+  it('cancels in-flight waiting even when the capability ignores its signal', async () => {
+    const controller = new AbortController();
+    const cap = defineCapability({
+      id: 'market.quote', name: 'Quote', description: 'Read a quote.',
+      category: 'market', riskLevel: 'read', auth: 'public', toolName: 'get_quote',
+      inputSchema: Type.Object({}), async execute() { return await new Promise<never>(() => {}); },
+    });
+    const [tool] = createCapabilityTools([cap], { timeoutMs: 100 });
+    const running = tool.execute('cancel', {}, controller.signal);
+    controller.abort(new Error('user cancelled'));
+    await expect(running).rejects.toThrow('user cancelled');
+  });
+
 });

packages/shared/src/capabilities/pi-tools.ts

@@ -2,6 +2,9 @@ import type { FinanceCapability, NewsItem } from '@finagent/core';
 import { validateInput } from './validate.ts';
 import { sanitizeNewsItem, sanitizeUntrustedText } from '../research/sanitize.ts';
 
+/** Keep a stalled provider call from holding the Pi agent turn forever. */
+export const CAPABILITY_TOOL_TIMEOUT_MS = 20_000;
+
 /** Shape the Pi Agent runtime expects for a registered tool. */
 export interface CapabilityTool {
   name: string;
@@ -25,34 +28,101 @@ export interface CapabilityTool {
  * `summary` plus a `DATA: <json>` block, so the agent reads facts while product
  * workflows use the structured `data` from the registry directly.
  */
-export function createCapabilityTools(capabilities: FinanceCapability[]): CapabilityTool[] {
+export function createCapabilityTools(
+  capabilities: FinanceCapability[],
+  options: { timeoutMs?: number } = {}
+): CapabilityTool[] {
   return capabilities.map((cap) => ({
     name: cap.toolName,
     label: cap.name,
     description: cap.description,
     parameters: cap.inputSchema,
     async execute(toolCallId, rawParams, signal) {
       const input = validateInput(cap.inputSchema, rawParams);
-      const result = await cap.execute(input, { signal });
-      // Security (defense in depth): external-text payloads such as news are
-      // sanitized again at the LLM boundary. The capability manifest already
-      // sanitizes at ingestion; this wrapper covers any provider that does not.
-      const data = isNewsItemList(result.data) ? result.data.map(sanitizeNewsItem) : result.data;
-      const summary = result.summary ? sanitizeUntrustedText(result.summary).text : undefined;
-      const json = JSON.stringify(data);
-      const base = summary ? `${summary}\n\nDATA: ${json}` : `DATA: ${json}`;
-      // surface the tool-call id so the model can cite this exact origin.
-      const text = `${base}\n\nEVIDENCE: ${toolCallId}`;
-      return {
-        content: [{ type: 'text', text }],
-        details: data,
-        provenance: result.provenance,
-        evidence: result.evidence,
-      };
+      if (signal.aborted) throw signal.reason ?? new Error('Tool execution cancelled.');
+      try {
+        const result = await runCapabilityWithTimeout(cap, input, signal, options.timeoutMs ?? CAPABILITY_TOOL_TIMEOUT_MS);
+        // Security (defense in depth): external-text payloads such as news are
+        // sanitized again at the LLM boundary. The capability manifest already
+        // sanitizes at ingestion; this wrapper covers any provider that does not.
+        const data = isNewsItemList(result.data) ? result.data.map(sanitizeNewsItem) : result.data;
+        const summary = result.summary ? sanitizeUntrustedText(result.summary).text : undefined;
+        const json = JSON.stringify(data);
+        const base = summary ? `${summary}\n\nDATA: ${json}` : `DATA: ${json}`;
+        // surface the tool-call id so the model can cite this exact origin.
+        const text = `${base}\n\nEVIDENCE: ${toolCallId}`;
+        return {
+          content: [{ type: 'text', text }],
+          details: data,
+          provenance: result.provenance,
+          evidence: result.evidence,
+        };
+      } catch (error) {
+        if (signal.aborted) throw error;
+        // A missing account scope, unavailable provider, or stalled tool is a
+        // data gap. Return it as tool content so Pi can continue with the other
+        // tools and the final answer can disclose the missing dimension.
+        return unavailableToolResult(cap, error);
+      }
     },
   }));
 }
 
+async function runCapabilityWithTimeout<TInput, TOutput>(
+  cap: FinanceCapability<TInput, TOutput>,
+  input: TInput,
+  parentSignal: AbortSignal,
+  timeoutMs: number
+) {
+  const controller = new AbortController();
+  let onAbort: (() => void) | undefined;
+  const abort = () => controller.abort(parentSignal.reason);
+  parentSignal.addEventListener('abort', abort, { once: true });
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    if (parentSignal.aborted) abort();
+    const aborted = new Promise<never>((_, reject) => {
+      onAbort = () => reject(controller.signal.reason ?? new Error('Tool execution cancelled.'));
+      if (controller.signal.aborted) onAbort();
+      else controller.signal.addEventListener('abort', onAbort, { once: true });
+    });
+    timer = setTimeout(() => {
+      controller.abort(Object.assign(new Error(`Capability ${cap.id} timed out after ${timeoutMs}ms.`), {
+        code: 'CAPABILITY_TIMEOUT',
+      }));
+    }, timeoutMs);
+    controller.signal.throwIfAborted();
+    return await Promise.race([cap.execute(input, { signal: controller.signal }), aborted]);
+  } finally {
+    clearTimeout(timer);
+    if (onAbort) controller.signal.removeEventListener('abort', onAbort);
+    parentSignal.removeEventListener('abort', abort);
+  }
+}
+
+function unavailableToolResult(cap: FinanceCapability, error: unknown) {
+  const code = readErrorCode(error) ?? 'UNAVAILABLE';
+  const message = readErrorMessage(error) ?? 'The data source could not be reached.';
+  return {
+    content: [{
+      type: 'text' as const,
+      text: `[TOOL_UNAVAILABLE] ${cap.name} was skipped (${code}): ${message}. Continue with the remaining tools and mark this data dimension unavailable.`,
+    }],
+    details: { status: 'unavailable', capabilityId: cap.id, code, message },
+  };
+}
+
+function readErrorCode(error: unknown): string | undefined {
+  if (!error || typeof error !== 'object' || !('code' in error)) return undefined;
+  const code = (error as { code?: unknown }).code;
+  return typeof code === 'string' ? code : undefined;
+}
+
+function readErrorMessage(error: unknown): string | undefined {
+  if (error instanceof Error) return error.message;
+  return typeof error === 'string' ? error : undefined;
+}
+
 /** Structural check for news-shaped payloads (title/summary/url items). */
 function isNewsItemList(data: unknown): data is NewsItem[] {
   return Array.isArray(data)

packages/shared/src/research/agent-synth.test.ts

@@ -72,4 +72,26 @@ describe('createAgentSynthesizer', () => {
     expect(synthesis.stance).toBe('bullish');
     expect(synthesis.sections[0].verdict).toBe('positive');
   });
+  it('discloses local fallback and retains the model failure for diagnostics', async () => {
+    const failure = new Error('model unavailable');
+    const seen: unknown[] = [];
+    const synthesizer = createAgentSynthesizer(async () => { throw failure; }, { onFallback: (error) => { seen.push(error); } });
+    const synthesis = await synthesizer.synthesize(synthesisInput());
+    expect(synthesis.summary).toContain('local rules');
+    expect(synthesis.risks[0]).toContain('Model synthesis unavailable');
+    expect(seen).toEqual([failure]);
+  });
+
+  it('does not turn user cancellation into a fallback report', async () => {
+    const controller = new AbortController();
+    const failure = new Error('user cancelled');
+    let fellBack = false;
+    const synthesizer = createAgentSynthesizer(async () => {
+      controller.abort(failure);
+      throw failure;
+    }, { onFallback: () => { fellBack = true; } });
+    await expect(synthesizer.synthesize(synthesisInput(), controller.signal)).rejects.toBe(failure);
+    expect(fellBack).toBe(false);
+  });
+
 });