diff --git a/apps/electron/src/main/kernelHost.ts b/apps/electron/src/main/kernelHost.ts index 696d603..2a7983e 100644 --- a/apps/electron/src/main/kernelHost.ts +++ b/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(); diff --git a/packages/shared/src/agent/pi-rpc-client.ts b/packages/shared/src/agent/pi-rpc-client.ts index 37030fc..62390ee 100644 --- a/packages/shared/src/agent/pi-rpc-client.ts +++ b/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'; } /** diff --git a/packages/shared/src/agent/pi-runtime-agent-backend.test.ts b/packages/shared/src/agent/pi-runtime-agent-backend.test.ts index 67410a1..ef00871 100644 --- a/packages/shared/src/agent/pi-runtime-agent-backend.test.ts +++ b/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'; diff --git a/packages/shared/src/capabilities/executor.test.ts b/packages/shared/src/capabilities/executor.test.ts index ded712a..af52695 100644 --- a/packages/shared/src/capabilities/executor.test.ts +++ b/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) => { diff --git a/packages/shared/src/capabilities/executor.ts b/packages/shared/src/capabilities/executor.ts index 81d91ff..9f606ad 100644 --- a/packages/shared/src/capabilities/executor.ts +++ b/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 { diff --git a/packages/shared/src/capabilities/pi-tools.test.ts b/packages/shared/src/capabilities/pi-tools.test.ts index 3e76c06..f9a2789 100644 --- a/packages/shared/src/capabilities/pi-tools.test.ts +++ b/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(() => {}); + }, + }); + 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(() => {}); }, + }); + 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'); + }); + }); diff --git a/packages/shared/src/capabilities/pi-tools.ts b/packages/shared/src/capabilities/pi-tools.ts index b47ebd6..a7b065a 100644 --- a/packages/shared/src/capabilities/pi-tools.ts +++ b/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,7 +28,10 @@ export interface CapabilityTool { * `summary` plus a `DATA: ` 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, @@ -33,26 +39,90 @@ export function createCapabilityTools(capabilities: FinanceCapability[]): Capabi 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( + cap: FinanceCapability, + 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 | undefined; + try { + if (parentSignal.aborted) abort(); + const aborted = new Promise((_, 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) diff --git a/packages/shared/src/research/agent-synth.test.ts b/packages/shared/src/research/agent-synth.test.ts index b862609..447ad0a 100644 --- a/packages/shared/src/research/agent-synth.test.ts +++ b/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); + }); + }); diff --git a/packages/shared/src/research/agent-synth.ts b/packages/shared/src/research/agent-synth.ts index 82c88f8..7ce2b08 100644 --- a/packages/shared/src/research/agent-synth.ts +++ b/packages/shared/src/research/agent-synth.ts @@ -97,14 +97,21 @@ function scrub(text: string): string { * fallback: any failure (agent unavailable, malformed output, parse error) * degrades to the honest local synthesizer — never fabricated prose. */ -export function createAgentSynthesizer(run: ResearchAgentRunner): ResearchSynthesizer { +export function createAgentSynthesizer( + run: ResearchAgentRunner, + options: { onFallback?: (error: unknown) => void } = {} +): ResearchSynthesizer { const fallback = new LocalResearchSynthesizer(); return { async synthesize(input, signal) { try { return await run(input, signal); - } catch { - return fallback.synthesize(input, signal); + } catch (error) { + if (signal?.aborted) throw error; + try { options.onFallback?.(error); } catch { /* Diagnostics cannot prevent fallback. */ } + const synthesis = await fallback.synthesize(input, signal); + const notice = 'Model synthesis unavailable; this report uses local rules applied to the saved evidence.'; + return { ...synthesis, summary: `${notice} ${synthesis.summary}`, risks: [notice, ...synthesis.risks] }; } }, };