From 9116289c24216541ce22f654bf22b07444a9dc82 Mon Sep 17 00:00:00 2001 From: George Pickett Date: Mon, 17 Aug 2026 20:41:16 -0700 Subject: [PATCH 1/3] Add Parallel research subagent to Pi --- README.md | 2 +- packages/pi-extension/README.md | 73 +++- .../pi-extension/agents/parallel-research.md | 17 + packages/pi-extension/package.json | 12 +- .../src/__tests__/package.test.ts | 40 ++ .../src/__tests__/parallel-auth.test.ts | 13 +- .../src/__tests__/parallel-responses.test.ts | 328 +++++++++++++++ packages/pi-extension/src/parallel-auth.ts | 19 +- .../pi-extension/src/parallel-responses.ts | 385 ++++++++++++++++++ 9 files changed, 878 insertions(+), 11 deletions(-) create mode 100644 packages/pi-extension/agents/parallel-research.md create mode 100644 packages/pi-extension/src/__tests__/package.test.ts create mode 100644 packages/pi-extension/src/__tests__/parallel-responses.test.ts create mode 100644 packages/pi-extension/src/parallel-responses.ts diff --git a/README.md b/README.md index f95d625..094ba76 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ Monorepo for @parallel-web npm packages. - [`@parallel-web/ai-sdk-tools`](./packages/ai-sdk-tools) - AI SDK tools for Parallel Web - [`@parallel-web/dsh-web-search`](./packages/dsh-web-search) - Parallel Search provider for DeepSeek Harness - [`@parallel-web/opencode-plugin`](./packages/opencode-plugin) - Opencode plugin for Parallel Web -- [`@parallel-web/pi-extension`](./packages/pi-extension) - pi agent extension for Parallel Web +- [`@parallel-web/pi-extension`](./packages/pi-extension) - Pi web tools and a native Parallel Responses research subagent - `@parallel-web/oauth` - Internal, unpublished shared PKCE OAuth helper. Bundled into the opencode plugin and pi extension at build time (`noExternal`), so it is never installed by consumers and is intentionally marked `private`. ## Development diff --git a/packages/pi-extension/README.md b/packages/pi-extension/README.md index 1034ff1..e6d564b 100644 --- a/packages/pi-extension/README.md +++ b/packages/pi-extension/README.md @@ -1,6 +1,7 @@ # @parallel-web/pi-extension -Pi extension that adds `web_search` and `web_fetch` backed by Parallel. +Pi extension that adds `web_search`, `web_fetch`, and a cited research model +backed by Parallel. Install it with: ``` @@ -11,6 +12,10 @@ pi install npm:@parallel-web/pi-extension - Registers `web_search` - Registers `web_fetch` +- Registers the `parallel/research` model, which makes one stateless Parallel + Responses API call +- Ships a `parallel-research` agent for + [pi-subagents](https://github.com/nicobailon/pi-subagents) - Registers a `parallel` auth provider, so Pi's own `/login parallel` runs the Parallel browser OAuth flow and stores the API key in Pi's auth store (`auth.json`) alongside every other provider credential @@ -26,6 +31,69 @@ Auth resolution order (owned by Pi, not the extension): Requires `@earendil-works/pi-coding-agent` 0.83.0 or newer. +## Parallel Research Subagent + +Install both packages to add the native research agent: + +```bash +pi install npm:pi-subagents +pi install npm:@parallel-web/pi-extension +``` + +This integration requires pi-subagents 0.50.0 or newer. The rest of the Pi +extension still works without pi-subagents. + +Run one research child directly: + +```text +/run parallel-research Compare the current JavaScript runtimes in Node and Bun. Cite primary sources. +``` + +The agent is also an ordinary pi-subagents child in JavaScript code mode. Its +`output` is the cited research text, so a later branch can use it directly: + +```javascript +const research = await runs.run("research", { + agent: "parallel-research", + task: "Which JavaScript runtime currently has stronger Node API compatibility? Cite primary sources.", + thinking: "medium", + context: "fresh", + worktree: false +}); + +if (/Bun/i.test(research.output)) { + return { recommendation: "evaluate-bun", evidence: research.output }; +} +return { recommendation: "stay-on-node", evidence: research.output }; +``` + +The default research effort is `medium`. A run may select `low`, `medium`, or +`high` with its `thinking` option. Current +[prices](https://docs.parallel.ai/getting-started/pricing) per successful +response are: + +| Thinking | Price | Typical use | +| --- | ---: | --- | +| `low` | $0.01 | Focused lookup | +| `medium` | $0.05 | General research | +| `high` | $0.25 | Hard, high-value research | + +The provider makes one `POST /v1/responses` request and does not retry it. It +does not use `previous_response_id`, background jobs, or a remote status loop. +Stopping the child aborts the local HTTP request on a best-effort basis; +Parallel does not expose acknowledged server-side cancellation for Responses. + +The research request contains only the packaged agent instructions and the +latest textual child task. It does not send parent history, local files, cwd, +environment variables, Pi tools, session state, or git worktree data. The +agent cannot read or edit the local filesystem. A worktree therefore adds no +research capability and should normally remain disabled. + +Parallel Responses accepts at most 20,000 combined instruction and input +characters. The adapter fails before making a request when that boundary is +exceeded. It renders the returned URL citations as a deduplicated Markdown +source list. + ## Dogfooding Locally Build the extension first: @@ -50,6 +118,7 @@ If the extension loads successfully, Pi will have: - the `web_fetch` tool - `parallel` listed under `/login` - the `parallel-login` status command +- the `parallel/research` model - per-session Parallel `session_id` reuse inside that Pi session ### Option 2: Symlink It Into Pi Extensions @@ -149,4 +218,6 @@ pnpm --filter @parallel-web/pi-extension typecheck - If automatic callback capture does not complete, the login dialog asks you to paste the callback URL. - Credential storage is entirely Pi's; the extension only reads the resolved key through `ctx.modelRegistry.getApiKeyForProvider("parallel")`. +- The research model is stateless and separate from the Search/Extract + `session_id` used by the web tools. - Skill suppression inside the extension is prompt-level only. If you want a clean dogfooding session without your usual skills list, start Pi with `--no-skills`. diff --git a/packages/pi-extension/agents/parallel-research.md b/packages/pi-extension/agents/parallel-research.md new file mode 100644 index 0000000..6527fb6 --- /dev/null +++ b/packages/pi-extension/agents/parallel-research.md @@ -0,0 +1,17 @@ +--- +name: parallel-research +description: One-shot cited web research through Parallel Responses +model: parallel/research +thinking: medium +systemPromptMode: replace +inheritProjectContext: false +inheritSkills: false +defaultContext: fresh +completionGuard: false +turnBudget: {"maxTurns":1,"graceTurns":0} +acceptance: {"level":"none","reason":"One-shot remote research provider"} +--- + +You are a read-only research agent backed by Parallel Responses. + +Research the user's task using current web sources. Return a direct, evidence-based answer with the citations supplied by the provider. Do not claim to inspect local files, run tools, change code, or access the parent session. diff --git a/packages/pi-extension/package.json b/packages/pi-extension/package.json index a900c66..9820e1a 100644 --- a/packages/pi-extension/package.json +++ b/packages/pi-extension/package.json @@ -1,7 +1,7 @@ { "name": "@parallel-web/pi-extension", "version": "1.2.0", - "description": "Add web search and web fetch to your pi agent", + "description": "Add Parallel web tools and cited research to your pi agent", "author": "Parallel Web", "license": "MIT", "type": "module", @@ -9,10 +9,16 @@ "image": "https://assets.parallel.ai/white-parallel-avatar-540.png", "extensions": [ "./dist/index.js" - ] + ], + "subagents": { + "agents": [ + "./agents" + ] + } }, "files": [ "dist", + "agents", "package.json", "README.md" ], @@ -41,6 +47,8 @@ "pi agent", "extension", "parallel", + "research", + "subagents", "web", "search", "fetch", diff --git a/packages/pi-extension/src/__tests__/package.test.ts b/packages/pi-extension/src/__tests__/package.test.ts new file mode 100644 index 0000000..f062a77 --- /dev/null +++ b/packages/pi-extension/src/__tests__/package.test.ts @@ -0,0 +1,40 @@ +import { readFileSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; + +const packageRoot = resolve( + dirname(fileURLToPath(import.meta.url)), + '..', + '..' +); + +describe('pi-subagents package contract', () => { + it('ships the Parallel research agent through the Pi manifest', () => { + const manifest = JSON.parse( + readFileSync(resolve(packageRoot, 'package.json'), 'utf8') + ); + + expect(manifest.name).toBe('@parallel-web/pi-extension'); + expect(manifest.version).toBe('1.2.0'); + expect(manifest.files).toContain('agents'); + expect(manifest.pi.subagents.agents).toEqual(['./agents']); + }); + + it('pins a one-turn fresh agent to the Parallel research model', () => { + const agent = readFileSync( + resolve(packageRoot, 'agents', 'parallel-research.md'), + 'utf8' + ); + + expect(agent).toContain('name: parallel-research'); + expect(agent).toContain('model: parallel/research'); + expect(agent).toContain('thinking: medium'); + expect(agent).toContain('systemPromptMode: replace'); + expect(agent).toContain('inheritProjectContext: false'); + expect(agent).toContain('inheritSkills: false'); + expect(agent).toContain('defaultContext: fresh'); + expect(agent).toContain('completionGuard: false'); + expect(agent).toContain('turnBudget: {"maxTurns":1,"graceTurns":0}'); + }); +}); diff --git a/packages/pi-extension/src/__tests__/parallel-auth.test.ts b/packages/pi-extension/src/__tests__/parallel-auth.test.ts index 6de427b..6a9a20c 100644 --- a/packages/pi-extension/src/__tests__/parallel-auth.test.ts +++ b/packages/pi-extension/src/__tests__/parallel-auth.test.ts @@ -56,12 +56,21 @@ describe('parallel-auth', () => { } = await import('../parallel-auth.js')); }); - it('registers a Parallel provider that serves no models', () => { + it('registers a Parallel provider with one Responses research model', () => { const provider = registerProvider(); expect(provider.id).toBe('parallel'); expect(provider.name).toBe('Parallel'); - expect(provider.getModels()).toEqual([]); + expect(provider.getModels()).toEqual([ + expect.objectContaining({ + id: 'research', + name: 'Parallel Research', + api: 'parallel-responses', + provider: 'parallel', + reasoning: true, + input: ['text'], + }), + ]); expect(provider.auth.apiKey).toBeDefined(); expect(provider.auth.oauth).toBeUndefined(); }); diff --git a/packages/pi-extension/src/__tests__/parallel-responses.test.ts b/packages/pi-extension/src/__tests__/parallel-responses.test.ts new file mode 100644 index 0000000..94751bd --- /dev/null +++ b/packages/pi-extension/src/__tests__/parallel-responses.test.ts @@ -0,0 +1,328 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { + AssistantMessageEvent, + Context, + SimpleStreamOptions, +} from '@earendil-works/pi-ai'; +import { + PARALLEL_RESEARCH_MODEL, + PARALLEL_RESPONSES_MAX_INPUT_CHARS, + PARALLEL_RESPONSES_URL, + streamParallelResponses, +} from '../parallel-responses.js'; + +function completedResponse( + overrides: Record = {} +): Record { + return { + id: 'resp_test', + status: 'completed', + output: [ + { + type: 'message', + role: 'assistant', + status: 'completed', + content: [ + { + type: 'output_text', + text: 'Parallel found the answer.', + annotations: [ + { + type: 'url_citation', + url: 'https://example.com/source', + title: 'Example [source]', + start_index: 0, + end_index: 14, + }, + { + type: 'url_citation', + url: 'https://example.com/source', + title: 'Duplicate title', + start_index: 15, + end_index: 21, + }, + ], + }, + ], + }, + ], + usage: { + input_tokens: 12, + output_tokens: 34, + total_tokens: 46, + }, + ...overrides, + }; +} + +function response(payload: unknown, status = 200): Response { + return new Response(JSON.stringify(payload), { + status, + headers: { 'x-request-id': 'request-test' }, + }); +} + +function researchContext(): Context { + return { + systemPrompt: 'Research carefully and cite sources.', + messages: [ + { role: 'user', content: 'Do not send this parent-history question.' }, + { + role: 'assistant', + content: [{ type: 'text', text: 'Do not send this prior answer.' }], + api: 'test', + provider: 'test', + model: 'test', + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + stopReason: 'stop', + timestamp: 1, + }, + { + role: 'user', + content: [ + { type: 'text', text: 'Research the current API contract.' }, + { type: 'image', data: 'not-forwarded', mimeType: 'image/png' }, + ], + timestamp: 2, + }, + ], + tools: [ + { + name: 'read', + description: 'Must not be forwarded', + parameters: { type: 'object' }, + }, + ], + } as Context; +} + +async function collect( + options: SimpleStreamOptions, + context = researchContext() +) { + const stream = streamParallelResponses( + PARALLEL_RESEARCH_MODEL, + context, + options + ); + const resultPromise = stream.result(); + const events: AssistantMessageEvent[] = []; + for await (const event of stream) events.push(event); + return { events, result: await resultPromise }; +} + +describe('Parallel Responses model', () => { + it('maps one stateless request and returns cited text with usage and cost', async () => { + const fetchMock = vi.fn(async () => response(completedResponse())); + const onPayload = vi.fn(); + const onResponse = vi.fn(); + + const { events, result } = await collect({ + apiKey: 'test-api-key', + fetch: fetchMock, + onPayload, + onResponse, + reasoning: 'medium', + metadata: { + sessionId: 'not-forwarded', + cwd: '/not-forwarded', + worktree: true, + }, + }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledWith( + PARALLEL_RESPONSES_URL, + expect.objectContaining({ + method: 'POST', + redirect: 'error', + signal: expect.any(AbortSignal), + }) + ); + + const init = fetchMock.mock.calls[0][1] as RequestInit; + expect(JSON.parse(String(init.body))).toEqual({ + model: 'parallel', + input: 'Research the current API contract.', + instructions: 'Research carefully and cite sources.', + reasoning: { effort: 'medium' }, + stream: false, + }); + const headers = new Headers(init.headers); + expect(headers.get('authorization')).toBe('Bearer test-api-key'); + expect(headers.get('content-type')).toBe('application/json'); + expect(headers.get('x-tool-calling-package')).toBe( + 'npm:@parallel-web/pi-extension/v1.2.0' + ); + expect(onPayload).toHaveBeenCalledTimes(1); + expect(onResponse).toHaveBeenCalledWith( + { + status: 200, + headers: expect.objectContaining({ 'x-request-id': 'request-test' }), + }, + PARALLEL_RESEARCH_MODEL + ); + + expect(events.map((event) => event.type)).toEqual([ + 'start', + 'text_start', + 'text_delta', + 'text_end', + 'done', + ]); + expect(result.content).toEqual([ + { + type: 'text', + text: [ + 'Parallel found the answer.', + '', + 'Sources:', + '1. [Example \\[source\\]]()', + ].join('\n'), + }, + ]); + expect(result.usage).toEqual( + expect.objectContaining({ + input: 12, + output: 34, + totalTokens: 46, + cost: expect.objectContaining({ total: 0.05 }), + }) + ); + expect(result.stopReason).toBe('stop'); + }); + + it.each([ + ['minimal', 'low', 0.01], + ['low', 'low', 0.01], + ['medium', 'medium', 0.05], + ['high', 'high', 0.25], + ['xhigh', 'high', 0.25], + ['max', 'high', 0.25], + ] as const)( + 'maps %s thinking to %s effort', + async (thinking, effort, cost) => { + const fetchMock = vi.fn(async () => response(completedResponse())); + + const { result } = await collect({ + apiKey: 'test-api-key', + fetch: fetchMock, + reasoning: thinking, + }); + + const init = fetchMock.mock.calls[0][1] as RequestInit; + expect(JSON.parse(String(init.body)).reasoning).toEqual({ effort }); + expect(result.usage.cost.total).toBe(cost); + } + ); + + it('fails oversized input before making a request', async () => { + const fetchMock = vi.fn(); + const context = researchContext(); + context.systemPrompt = 'x'.repeat(PARALLEL_RESPONSES_MAX_INPUT_CHARS); + + const { events, result } = await collect( + { apiKey: 'test-api-key', fetch: fetchMock }, + context + ); + + expect(fetchMock).not.toHaveBeenCalled(); + expect(events.at(-1)).toEqual( + expect.objectContaining({ type: 'error', reason: 'error' }) + ); + expect(result.stopReason).toBe('error'); + expect(result.errorMessage).toContain('20,000-character limit'); + }); + + it('does not retry HTTP errors and redacts the credential', async () => { + const fetchMock = vi.fn(async () => + response({ error: { message: 'key test-api-key is unauthorized' } }, 401) + ); + + const { result } = await collect({ + apiKey: 'test-api-key', + fetch: fetchMock, + }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(result.stopReason).toBe('error'); + expect(result.errorMessage).toBe( + 'Parallel Responses request failed (401): key [REDACTED] is unauthorized' + ); + expect(result.errorMessage).not.toContain('test-api-key'); + }); + + it('does not retry malformed completed responses', async () => { + const fetchMock = vi.fn(async () => + response(completedResponse({ output: [] })) + ); + + const { result } = await collect({ + apiKey: 'test-api-key', + fetch: fetchMock, + }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(result.stopReason).toBe('error'); + expect(result.errorMessage).toBe( + 'Parallel returned an empty research response.' + ); + }); + + it('propagates caller cancellation to the request signal', async () => { + const controller = new AbortController(); + controller.abort(new Error('cancelled by caller')); + const fetchMock = vi.fn(async (_url: unknown, init?: RequestInit) => { + expect(init?.signal?.aborted).toBe(true); + throw new DOMException('The operation was aborted.', 'AbortError'); + }); + + const { events, result } = await collect({ + apiKey: 'test-api-key', + fetch: fetchMock as typeof fetch, + signal: controller.signal, + }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(events.at(-1)).toEqual( + expect.objectContaining({ type: 'error', reason: 'aborted' }) + ); + expect(result.stopReason).toBe('aborted'); + }); + + it('terminates a hung request at the local timeout without retrying', async () => { + const fetchMock = vi.fn( + async (_url: unknown, init?: RequestInit): Promise => + await new Promise((_resolve, reject) => { + init?.signal?.addEventListener( + 'abort', + () => reject(new DOMException('Timed out', 'AbortError')), + { once: true } + ); + }) + ); + + const { result } = await collect({ + apiKey: 'test-api-key', + fetch: fetchMock as typeof fetch, + timeoutMs: 5, + }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(result.stopReason).toBe('error'); + expect(result.errorMessage).toBe('Parallel Research timed out.'); + }); +}); diff --git a/packages/pi-extension/src/parallel-auth.ts b/packages/pi-extension/src/parallel-auth.ts index d0abbd5..f764077 100644 --- a/packages/pi-extension/src/parallel-auth.ts +++ b/packages/pi-extension/src/parallel-auth.ts @@ -8,8 +8,13 @@ import type { AuthResult, Provider, ProviderAuthInteraction, + SimpleStreamOptions, } from '@earendil-works/pi-ai'; import { loginWithParallel as runParallelOAuth } from '@parallel-web/oauth'; +import { + PARALLEL_RESEARCH_MODEL, + streamParallelResponses, +} from './parallel-responses'; /** Provider id under which Pi stores the Parallel credential in its auth store. */ export const PARALLEL_PROVIDER = 'parallel'; @@ -87,12 +92,16 @@ function createParallelProvider(): Provider { resolve: resolveParallelAuth, }, }, - getModels: () => [], - stream() { - throw new Error('The Parallel provider does not serve models.'); + getModels: () => [PARALLEL_RESEARCH_MODEL], + stream(model, context, options) { + return streamParallelResponses( + model, + context, + options as SimpleStreamOptions + ); }, - streamSimple() { - throw new Error('The Parallel provider does not serve models.'); + streamSimple(model, context, options) { + return streamParallelResponses(model, context, options); }, }; } diff --git a/packages/pi-extension/src/parallel-responses.ts b/packages/pi-extension/src/parallel-responses.ts new file mode 100644 index 0000000..29fc5d3 --- /dev/null +++ b/packages/pi-extension/src/parallel-responses.ts @@ -0,0 +1,385 @@ +declare const __PACKAGE_VERSION__: string; + +import { + createAssistantMessageEventStream, + type Api, + type AssistantMessage, + type AssistantMessageEventStream, + type Context, + type Model, + type SimpleStreamOptions, +} from '@earendil-works/pi-ai'; + +export const PARALLEL_RESPONSES_API = 'parallel-responses'; +export const PARALLEL_RESPONSES_URL = 'https://api.parallel.ai/v1/responses'; +export const PARALLEL_RESPONSES_MAX_INPUT_CHARS = 20_000; +export const PARALLEL_RESPONSES_DEFAULT_TIMEOUT_MS = 120_000; + +export const PARALLEL_RESEARCH_MODEL: Model = { + id: 'research', + name: 'Parallel Research', + api: PARALLEL_RESPONSES_API, + provider: 'parallel', + baseUrl: 'https://api.parallel.ai', + reasoning: true, + thinkingLevelMap: { + off: null, + minimal: 'low', + low: 'low', + medium: 'medium', + high: 'high', + xhigh: 'high', + max: 'high', + }, + input: ['text'], + // Parallel Responses is billed per successful call, not per token. The + // custom stream records the fixed call price in usage.cost.total. + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: PARALLEL_RESPONSES_MAX_INPUT_CHARS, + maxTokens: 32_000, +}; + +type ResearchEffort = 'low' | 'medium' | 'high'; + +const COST_PER_SUCCESSFUL_CALL: Record = { + low: 0.01, + medium: 0.05, + high: 0.25, +}; + +interface UrlCitation { + url: string; + title: string; +} + +interface ParsedResponse { + text: string; + citations: UrlCitation[]; + usage: { + input: number; + output: number; + totalTokens: number; + }; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function safeInteger(value: unknown): number { + return Number.isSafeInteger(value) && Number(value) >= 0 ? Number(value) : 0; +} + +function resolveEffort( + reasoning: SimpleStreamOptions['reasoning'] +): ResearchEffort { + if (reasoning === 'minimal' || reasoning === 'low') return 'low'; + if (reasoning === 'high' || reasoning === 'xhigh' || reasoning === 'max') { + return 'high'; + } + return 'medium'; +} + +function latestUserText(context: Context): string { + for (let index = context.messages.length - 1; index >= 0; index -= 1) { + const message = context.messages[index]; + if (message.role !== 'user') continue; + + const text = + typeof message.content === 'string' + ? message.content + : message.content + .filter((part) => part.type === 'text') + .map((part) => part.text) + .join('\n'); + + if (text.trim()) return text; + } + + throw new Error('Parallel Research requires a non-empty textual user task.'); +} + +function parseCitation(value: unknown): UrlCitation | undefined { + if (!isRecord(value) || value.type !== 'url_citation') return undefined; + if (typeof value.url !== 'string' || typeof value.title !== 'string') { + return undefined; + } + + let url: URL; + try { + url = new URL(value.url); + } catch { + return undefined; + } + if (url.protocol !== 'https:' && url.protocol !== 'http:') return undefined; + + return { + url: url.toString(), + title: value.title, + }; +} + +function parseResponse(payload: unknown): ParsedResponse { + if (!isRecord(payload) || payload.status !== 'completed') { + throw new Error('Parallel returned a response that was not completed.'); + } + if (!Array.isArray(payload.output)) { + throw new Error('Parallel returned a response without output messages.'); + } + + const texts: string[] = []; + const citations: UrlCitation[] = []; + for (const item of payload.output) { + if ( + !isRecord(item) || + item.type !== 'message' || + !Array.isArray(item.content) + ) { + continue; + } + for (const content of item.content) { + if ( + !isRecord(content) || + content.type !== 'output_text' || + typeof content.text !== 'string' + ) { + continue; + } + texts.push(content.text); + if (Array.isArray(content.annotations)) { + for (const annotation of content.annotations) { + const citation = parseCitation(annotation); + if (citation) citations.push(citation); + } + } + } + } + + const text = texts.join('\n\n').trim(); + if (!text) throw new Error('Parallel returned an empty research response.'); + + const usage = isRecord(payload.usage) ? payload.usage : {}; + return { + text, + citations, + usage: { + input: safeInteger(usage.input_tokens), + output: safeInteger(usage.output_tokens), + totalTokens: safeInteger(usage.total_tokens), + }, + }; +} + +function escapeMarkdownLabel(value: string): string { + return value + .replaceAll('\\', '\\\\') + .replaceAll('[', '\\[') + .replaceAll(']', '\\]'); +} + +function markdownUrl(value: string): string { + return value.replaceAll('<', '%3C').replaceAll('>', '%3E'); +} + +function renderCitedResearch(parsed: ParsedResponse): string { + const sources = new Map(); + for (const citation of parsed.citations) { + if (!sources.has(citation.url)) { + sources.set(citation.url, citation.title.trim() || citation.url); + } + } + if (sources.size === 0) return parsed.text; + + const list = [...sources].map( + ([url, title], index) => + `${index + 1}. [${escapeMarkdownLabel(title)}](<${markdownUrl(url)}>)` + ); + return `${parsed.text}\n\nSources:\n${list.join('\n')}`; +} + +function responseHeaders(response: Response): Record { + return Object.fromEntries(response.headers.entries()); +} + +function safeErrorMessage(error: unknown, apiKey: string): string { + let message: string; + try { + message = error instanceof Error ? error.message : String(error); + } catch { + message = 'Unknown provider failure'; + } + + if (apiKey) message = message.replaceAll(apiKey, '[REDACTED]'); + const trimmed = message.trim(); + return (trimmed || 'Unknown provider failure').slice(0, 1_000); +} + +async function httpError(response: Response): Promise { + let message = response.statusText || 'request failed'; + try { + const payload: unknown = await response.json(); + if (isRecord(payload) && isRecord(payload.error)) { + if (typeof payload.error.message === 'string') { + message = payload.error.message; + } + } + } catch { + // Status and statusText remain the useful, bounded diagnostic. + } + return new Error( + `Parallel Responses request failed (${response.status}): ${message}` + ); +} + +function createOutput(model: Model): AssistantMessage { + return { + role: 'assistant', + content: [], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: 'pending', + timestamp: Date.now(), + }; +} + +export function streamParallelResponses( + model: Model, + context: Context, + options?: SimpleStreamOptions +): AssistantMessageEventStream { + const stream = createAssistantMessageEventStream(); + const output = createOutput(model); + + void (async () => { + const apiKey = options?.apiKey ?? ''; + const requestController = new AbortController(); + let cancelledByCaller = false; + let timedOut = false; + let timeout: ReturnType | undefined; + + const abortFromCaller = () => { + cancelledByCaller = true; + requestController.abort(options?.signal?.reason); + }; + + try { + stream.push({ type: 'start', partial: output }); + if (!apiKey) { + throw new Error( + 'Parallel authentication required. Run `/login parallel` in Pi, or set PARALLEL_API_KEY.' + ); + } + + if (options?.signal?.aborted) abortFromCaller(); + options?.signal?.addEventListener('abort', abortFromCaller, { + once: true, + }); + + const timeoutMs = + options?.timeoutMs ?? PARALLEL_RESPONSES_DEFAULT_TIMEOUT_MS; + if (timeoutMs > 0) { + timeout = setTimeout(() => { + timedOut = true; + requestController.abort(new Error('Parallel Research timed out.')); + }, timeoutMs); + timeout.unref?.(); + } + + const input = latestUserText(context); + const instructions = context.systemPrompt?.trim() || undefined; + if ( + input.length + (instructions?.length ?? 0) > + PARALLEL_RESPONSES_MAX_INPUT_CHARS + ) { + throw new Error( + `Parallel Research input exceeds the ${PARALLEL_RESPONSES_MAX_INPUT_CHARS.toLocaleString('en-US')}-character limit.` + ); + } + + const effort = resolveEffort(options?.reasoning); + let payload: unknown = { + model: 'parallel', + input, + ...(instructions ? { instructions } : {}), + reasoning: { effort }, + stream: false, + }; + const replacement = await options?.onPayload?.(payload, model); + if (replacement !== undefined) payload = replacement; + + const headers = new Headers({ + 'Content-Type': 'application/json', + 'X-Tool-Calling-Package': `npm:@parallel-web/pi-extension/v${__PACKAGE_VERSION__ ?? '0.0.0'}`, + }); + for (const [name, value] of Object.entries(options?.headers ?? {})) { + if (value === null) headers.delete(name); + else headers.set(name, value); + } + headers.set('Authorization', `Bearer ${apiKey}`); + + const fetchImpl = options?.fetch ?? globalThis.fetch; + const response = await fetchImpl(PARALLEL_RESPONSES_URL, { + method: 'POST', + headers, + body: JSON.stringify(payload), + signal: requestController.signal, + redirect: 'error', + }); + await options?.onResponse?.( + { status: response.status, headers: responseHeaders(response) }, + model + ); + if (!response.ok) throw await httpError(response); + + const parsed = parseResponse(await response.json()); + const text = renderCitedResearch(parsed); + output.content.push({ type: 'text', text }); + output.usage.input = parsed.usage.input; + output.usage.output = parsed.usage.output; + output.usage.totalTokens = parsed.usage.totalTokens; + output.usage.cost.total = COST_PER_SUCCESSFUL_CALL[effort]; + output.stopReason = 'stop'; + + stream.push({ type: 'text_start', contentIndex: 0, partial: output }); + stream.push({ + type: 'text_delta', + contentIndex: 0, + delta: text, + partial: output, + }); + stream.push({ + type: 'text_end', + contentIndex: 0, + content: text, + partial: output, + }); + stream.push({ type: 'done', reason: 'stop', message: output }); + } catch (error) { + const aborted = cancelledByCaller; + output.stopReason = aborted ? 'aborted' : 'error'; + output.errorMessage = timedOut + ? 'Parallel Research timed out.' + : safeErrorMessage(error, apiKey); + stream.push({ + type: 'error', + reason: aborted ? 'aborted' : 'error', + error: output, + }); + } finally { + if (timeout) clearTimeout(timeout); + options?.signal?.removeEventListener('abort', abortFromCaller); + stream.end(); + } + })(); + + return stream; +} From 52619e5ef9fbc052080a3dc5dac4627297fc8dc9 Mon Sep 17 00:00:00 2001 From: George Pickett Date: Mon, 17 Aug 2026 21:00:52 -0700 Subject: [PATCH 2/3] Tighten Parallel research subagent boundaries --- packages/pi-extension/README.md | 5 +++ .../src/__tests__/parallel-auth.test.ts | 2 + .../src/__tests__/parallel-responses.test.ts | 42 +++++++++++++++++++ packages/pi-extension/src/parallel-auth.ts | 8 ++-- .../pi-extension/src/parallel-responses.ts | 24 +++++++++-- 5 files changed, 74 insertions(+), 7 deletions(-) diff --git a/packages/pi-extension/README.md b/packages/pi-extension/README.md index e6d564b..ef63cfe 100644 --- a/packages/pi-extension/README.md +++ b/packages/pi-extension/README.md @@ -89,6 +89,11 @@ environment variables, Pi tools, session state, or git worktree data. The agent cannot read or edit the local filesystem. A worktree therefore adds no research capability and should normally remain disabled. +These data boundaries describe normal `parallel-research` runs. Pi's low-level +provider API also exposes `onPayload` and custom-header hooks to trusted caller +code. A caller that deliberately uses those hooks to replace or extend the +request owns the resulting data boundary. + Parallel Responses accepts at most 20,000 combined instruction and input characters. The adapter fails before making a request when that boundary is exceeded. It renders the returned URL citations as a deduplicated Markdown diff --git a/packages/pi-extension/src/__tests__/parallel-auth.test.ts b/packages/pi-extension/src/__tests__/parallel-auth.test.ts index 6a9a20c..ea56c4d 100644 --- a/packages/pi-extension/src/__tests__/parallel-auth.test.ts +++ b/packages/pi-extension/src/__tests__/parallel-auth.test.ts @@ -69,6 +69,8 @@ describe('parallel-auth', () => { provider: 'parallel', reasoning: true, input: ['text'], + contextWindow: 37_000, + maxTokens: 32_000, }), ]); expect(provider.auth.apiKey).toBeDefined(); diff --git a/packages/pi-extension/src/__tests__/parallel-responses.test.ts b/packages/pi-extension/src/__tests__/parallel-responses.test.ts index 94751bd..fca04a2 100644 --- a/packages/pi-extension/src/__tests__/parallel-responses.test.ts +++ b/packages/pi-extension/src/__tests__/parallel-responses.test.ts @@ -229,6 +229,48 @@ describe('Parallel Responses model', () => { } ); + it('honors an explicit caller-owned payload replacement', async () => { + const fetchMock = vi.fn(async () => response(completedResponse())); + const replacement = { + model: 'parallel', + input: 'Trusted caller replacement.', + reasoning: { effort: 'low' }, + stream: false, + }; + + await collect({ + apiKey: 'test-api-key', + fetch: fetchMock, + onPayload: () => replacement, + }); + + const init = fetchMock.mock.calls[0][1] as RequestInit; + expect(JSON.parse(String(init.body))).toEqual(replacement); + }); + + it('does not fall back to history for a non-textual latest task', async () => { + const fetchMock = vi.fn(); + const context = researchContext(); + context.messages.push({ + role: 'user', + content: [ + { type: 'image', data: 'latest-image-only', mimeType: 'image/png' }, + ], + timestamp: 3, + }); + + const { result } = await collect( + { apiKey: 'test-api-key', fetch: fetchMock }, + context + ); + + expect(fetchMock).not.toHaveBeenCalled(); + expect(result.stopReason).toBe('error'); + expect(result.errorMessage).toBe( + 'Parallel Research requires a non-empty textual user task.' + ); + }); + it('fails oversized input before making a request', async () => { const fetchMock = vi.fn(); const context = researchContext(); diff --git a/packages/pi-extension/src/parallel-auth.ts b/packages/pi-extension/src/parallel-auth.ts index f764077..6336243 100644 --- a/packages/pi-extension/src/parallel-auth.ts +++ b/packages/pi-extension/src/parallel-auth.ts @@ -76,10 +76,10 @@ async function resolveParallelAuth(input: { } /** - * A provider that exists purely to carry Parallel's credential. Pi owns the - * storage (auth.json), the `/login parallel` and `/logout parallel` flows, and - * the `PARALLEL_API_KEY` fallback; the extension only reads the resolved key. - * It serves no models, so the stream entry points are never reached. + * Parallel's provider owns both the shared credential boundary and the static + * research model. Pi owns auth.json, `/login parallel`, `/logout parallel`, + * and the `PARALLEL_API_KEY` fallback; both web tools and the model reuse that + * resolved credential without adding another auth concept. */ function createParallelProvider(): Provider { return { diff --git a/packages/pi-extension/src/parallel-responses.ts b/packages/pi-extension/src/parallel-responses.ts index 29fc5d3..5bcba42 100644 --- a/packages/pi-extension/src/parallel-responses.ts +++ b/packages/pi-extension/src/parallel-responses.ts @@ -15,6 +15,9 @@ export const PARALLEL_RESPONSES_URL = 'https://api.parallel.ai/v1/responses'; export const PARALLEL_RESPONSES_MAX_INPUT_CHARS = 20_000; export const PARALLEL_RESPONSES_DEFAULT_TIMEOUT_MS = 120_000; +const RESPONSE_USAGE_CHARS_PER_TOKEN = 4; +const RESEARCH_MAX_OUTPUT_TOKENS = 32_000; + export const PARALLEL_RESEARCH_MODEL: Model = { id: 'research', name: 'Parallel Research', @@ -33,10 +36,17 @@ export const PARALLEL_RESEARCH_MODEL: Model = { }, input: ['text'], // Parallel Responses is billed per successful call, not per token. The - // custom stream records the fixed call price in usage.cost.total. + // custom stream records the fixed call price in usage.cost.total. Pi's + // contextWindow includes input plus output tokens, while Responses limits + // input in characters and reports usage with a four-chars-per-token + // estimate. Keep that estimate in catalog metadata; the explicit character + // check below remains the authoritative request limit. cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, - contextWindow: PARALLEL_RESPONSES_MAX_INPUT_CHARS, - maxTokens: 32_000, + contextWindow: + Math.ceil( + PARALLEL_RESPONSES_MAX_INPUT_CHARS / RESPONSE_USAGE_CHARS_PER_TOKEN + ) + RESEARCH_MAX_OUTPUT_TOKENS, + maxTokens: RESEARCH_MAX_OUTPUT_TOKENS, }; type ResearchEffort = 'low' | 'medium' | 'high'; @@ -94,6 +104,11 @@ function latestUserText(context: Context): string { .join('\n'); if (text.trim()) return text; + + // The latest user turn is the task boundary. Never fall back to an older + // user message when the current task is empty or non-textual, because that + // would turn parent history into a new research request. + break; } throw new Error('Parallel Research requires a non-empty textual user task.'); @@ -313,6 +328,9 @@ export function streamParallelResponses( reasoning: { effort }, stream: false, }; + // Pi exposes onPayload as an explicit inspect-or-replace hook. Normal + // pi-subagents use does not replace this minimal body; a low-level caller + // that does return a replacement owns the resulting data boundary. const replacement = await options?.onPayload?.(payload, model); if (replacement !== undefined) payload = replacement; From 709a53bccdbbc84999529ebd760723628ac0c27b Mon Sep 17 00:00:00 2001 From: George Pickett Date: Tue, 18 Aug 2026 09:33:43 -0700 Subject: [PATCH 3/3] Fix Pi auth guidance --- packages/pi-extension/README.md | 7 ++++--- packages/pi-extension/src/__tests__/index.test.ts | 8 ++++++++ packages/pi-extension/src/index.ts | 5 ++++- packages/pi-extension/src/parallel-auth.ts | 6 +++--- 4 files changed, 19 insertions(+), 7 deletions(-) diff --git a/packages/pi-extension/README.md b/packages/pi-extension/README.md index ef63cfe..e128064 100644 --- a/packages/pi-extension/README.md +++ b/packages/pi-extension/README.md @@ -26,8 +26,9 @@ Auth resolution order (owned by Pi, not the extension): 1. The credential Pi stored for provider `parallel` 2. `PARALLEL_API_KEY` -`/logout parallel` removes the stored credential, and -`pi auth check --provider parallel` reports whether it is configured. +Run `/parallel-login` inside Pi to check whether Parallel is configured. To +remove a stored credential, run `/logout` and select Parallel. Environment +variables are not affected by Pi's logout flow. Requires `@earendil-works/pi-coding-agent` 0.83.0 or newer. @@ -169,7 +170,7 @@ Inside Pi, run: That opens the browser for Parallel OAuth. On success, Pi stores the API key in its auth store under `parallel`. Run `/parallel-login` to see the current status, -and `/logout parallel` to remove the credential. +and run `/logout` and select Parallel to remove the stored credential. ### Use Environment Variable Instead diff --git a/packages/pi-extension/src/__tests__/index.test.ts b/packages/pi-extension/src/__tests__/index.test.ts index e0f1912..7cb39ae 100644 --- a/packages/pi-extension/src/__tests__/index.test.ts +++ b/packages/pi-extension/src/__tests__/index.test.ts @@ -239,6 +239,10 @@ describe('@parallel-web/pi-extension', () => { expect.stringContaining('authenticated (stored)'), 'info' ); + expect(ctx.ui.notify).toHaveBeenCalledWith( + expect.stringContaining('`/logout` and select Parallel'), + 'info' + ); }); it('parallel-login should recognize a key that only PARALLEL_API_KEY provides', async () => { @@ -260,6 +264,10 @@ describe('@parallel-web/pi-extension', () => { expect.stringContaining('authenticated (PARALLEL_API_KEY)'), 'info' ); + expect(ctx.ui.notify).toHaveBeenCalledWith( + expect.stringContaining('unset PARALLEL_API_KEY'), + 'info' + ); }); it('web_search should use the stored api key when available', async () => { diff --git a/packages/pi-extension/src/index.ts b/packages/pi-extension/src/index.ts index c836e4d..fd32c58 100644 --- a/packages/pi-extension/src/index.ts +++ b/packages/pi-extension/src/index.ts @@ -130,8 +130,11 @@ export default function (pi: ExtensionAPI) { const status = getParallelAuthStatus(ctx); const source = status.label ?? status.source ?? 'PARALLEL_API_KEY'; + const guidance = status.configured + ? 'Run `/login parallel` to replace the stored credential, or `/logout` and select Parallel to remove it.' + : 'Run `/login parallel` to store a credential, or unset PARALLEL_API_KEY to remove the current one.'; ctx.ui.notify( - `Parallel is authenticated (${source}). Run \`/login parallel\` to replace the credential, or \`/logout parallel\` to remove it.`, + `Parallel is authenticated (${source}). ${guidance}`, 'info' ); }, diff --git a/packages/pi-extension/src/parallel-auth.ts b/packages/pi-extension/src/parallel-auth.ts index 6336243..d3f2f31 100644 --- a/packages/pi-extension/src/parallel-auth.ts +++ b/packages/pi-extension/src/parallel-auth.ts @@ -77,9 +77,9 @@ async function resolveParallelAuth(input: { /** * Parallel's provider owns both the shared credential boundary and the static - * research model. Pi owns auth.json, `/login parallel`, `/logout parallel`, - * and the `PARALLEL_API_KEY` fallback; both web tools and the model reuse that - * resolved credential without adding another auth concept. + * research model. Pi owns auth.json, `/login parallel`, the `/logout` provider + * picker, and the `PARALLEL_API_KEY` fallback; both web tools and the model + * reuse that resolved credential without adding another auth concept. */ function createParallelProvider(): Provider { return {