feat(search): web_search onto the injected port — Google CSE retired, research honestly disabled
Chunk A of the search port wave (2026-09-20), per master's assignment: - web.ts: the Google Custom Search backend is gone — web_search now runs on the core search port (SearchProvider), injected as a dependency (import law: the tool never imports server code). createWebTools(search) resolves the LIVE provider fresh per call; null = the honest unavailability string (no silent fallback, owner decision). web_fetch unchanged (needs no engine). - runner: JobRunnerDeps.searchProvider — the server composition threaded into the researcher; jobs already in flight when a key disappears still fail honestly through the same seam, never crash. - jobTools: the research dispatch tool DROPS from the teacher's registry when ctx.searchAvailable() is false (owner decision: registry-level drop — the model never sees the tool, no doomed dispatch, no error card). Absent seam = available (tests). chat-handler injects server/searchAvailable + searchProvider. - tests: stubFetch (the Google wire stub) retired; jobs.test.ts threads a recording stub provider through all four research runners and asserts the scripted query hit the port; new web-search.test.ts covers the factory contract (null → honest string, hits, error strings, fresh resolution, web_fetch independence) + the registry drop matrix. Gates: typecheck, 793/793 tests, lint, build — all green.
This commit is contained in:
amosblomqvist
@@ -1,6 +1,7 @@
|
||||
import type { AgentTool } from "@earendil-works/pi-agent-core";
|
||||
import type { Usage } from "@earendil-works/pi-ai";
|
||||
import { webTools } from "../tools/web.ts";
|
||||
import { createWebTools } from "../tools/web.ts";
|
||||
import type { SearchProvider } from "../../core/search.ts";
|
||||
import { toPiTools } from "../lh-tool.ts";
|
||||
import type { PiEngine } from "../models.ts";
|
||||
import { runSubagent } from "./pi-run.ts";
|
||||
@@ -62,6 +63,12 @@ What couldn't be answered. Suggested next steps.`;
|
||||
|
||||
export interface ResearcherRunOptions {
|
||||
engine?: PiEngine;
|
||||
/** The search seam (search port wave, 2026-09-20): resolves the LIVE
|
||||
* provider per call, null = honestly unavailable (the web_search tool
|
||||
* reports that in its output — no crash, no silent fallback). Absent
|
||||
* seam reads as unavailable: the server wiring always injects one; the
|
||||
* dispatch-level drop (jobTools) keeps doomed jobs from even starting. */
|
||||
search?: () => SearchProvider | null;
|
||||
abortSignal?: AbortSignal;
|
||||
onUsage?: (usage: Usage) => void;
|
||||
onToolStart?: (toolName: string, args: unknown) => void;
|
||||
@@ -69,10 +76,11 @@ export interface ResearcherRunOptions {
|
||||
|
||||
/**
|
||||
* Run the researcher to completion: isolated context, web tools only, final
|
||||
* text = the brief. The engine is injectable so tests run the loop scripted.
|
||||
* text = the brief. The engine and search seam are injectable so tests run
|
||||
* the loop scripted.
|
||||
*/
|
||||
export async function runResearcherAgent(prompt: string, opts: ResearcherRunOptions = {}): Promise<string> {
|
||||
const { text } = await runSubagent(toPiTools(webTools) as AgentTool<any, any>[], {
|
||||
const { text } = await runSubagent(toPiTools(createWebTools(opts.search ?? (() => null))) as AgentTool<any, any>[], {
|
||||
role: "researcher",
|
||||
prompt,
|
||||
engine: opts.engine,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { isToolUIPart, type UIMessage } from "ai";
|
||||
import type { PiEngine } from "../models.ts";
|
||||
import type { SearchProvider } from "../../core/search.ts";
|
||||
import {
|
||||
accumulateJobUsage,
|
||||
buildSubagentSteer,
|
||||
@@ -76,6 +77,13 @@ export interface JobRunnerDeps {
|
||||
* Test overrides (below) still win; absent → config pins.
|
||||
*/
|
||||
engineFor?: (role: ModelRole) => PiEngine;
|
||||
/** The search seam (search port wave, 2026-09-20): resolves the LIVE
|
||||
* provider fresh per job — server/search.ts composition injected by
|
||||
* the wiring; null = honestly unavailable (the researcher's web_search
|
||||
* reports it). Absent seam (tests) reads the same. The dispatch-level
|
||||
* drop (jobTools' searchAvailable) uses the same signal — jobs already
|
||||
* in flight when a key disappears still fail honestly, never crash. */
|
||||
searchProvider?: () => SearchProvider | null;
|
||||
}
|
||||
|
||||
export interface JobRunner {
|
||||
@@ -308,6 +316,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
|
||||
jobStore.updateJob(jobId, { activity: "researching" });
|
||||
const brief = await runResearcherAgent(prompt, {
|
||||
engine: model,
|
||||
search: deps.searchProvider,
|
||||
abortSignal: loop.abortSignal,
|
||||
// Money channel: each agent step is one model call — fold its
|
||||
// usage into the record as it completes (progress write).
|
||||
|
||||
@@ -154,6 +154,10 @@ export interface TeacherToolContext extends PlanToolContext {
|
||||
* rides along — the spawn fires mid-stream, before the tool part is
|
||||
* persisted to the transcript (a transcript scan would race it). */
|
||||
spawn: (job: JobRecord, input: ResearchInput | DemoInput | VisualInput) => void;
|
||||
/** Search availability (search port wave, 2026-09-20): the research
|
||||
* dispatch tool's honest-disable signal — no live engine = the tool
|
||||
* drops from this registry (JobToolContext.searchAvailable). */
|
||||
searchAvailable?: () => boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+19
-5
@@ -22,6 +22,12 @@ export interface JobToolContext {
|
||||
* input rides along because the spawn happens MID-STREAM, before
|
||||
* onFinish persists the tool part — a transcript scan would race it. */
|
||||
spawn: (job: JobRecord, input: ResearchInput | DemoInput | VisualInput | PdfInput) => void;
|
||||
/** Search availability (search port wave, 2026-09-20): the honest
|
||||
* disable signal for the research dispatch tool — no live engine
|
||||
* means NO research tool in the registry (owner decision: registry-
|
||||
* level drop, no doomed dispatch, no silent fallback). Absent =
|
||||
* available (tests + any seam without search wiring). */
|
||||
searchAvailable?: () => boolean;
|
||||
}
|
||||
|
||||
export const researchInputSchema = z.object({
|
||||
@@ -121,14 +127,22 @@ const PDF_DESCRIPTION =
|
||||
* Per-request job tool registry: each tool closes over the composition root's
|
||||
* context (chatId + jobStore + the spawn seam), so concurrent chats never
|
||||
* share state. chat-handler wires it each turn; tests wire their own.
|
||||
*
|
||||
* The research tool drops from the registry when ctx.searchAvailable reports
|
||||
* no live engine (2026-09-20) — the teacher never sees it, so a keyless
|
||||
* server can't dispatch doomed research jobs. Jobs ALREADY in flight when
|
||||
* a key disappears still run to their honest failure (runner backstop).
|
||||
*/
|
||||
export function jobTools(ctx: JobToolContext): Record<string, LhTool<any, any>> {
|
||||
const searchOn = ctx.searchAvailable?.() ?? true;
|
||||
return {
|
||||
research: lhTool({
|
||||
name: "research",
|
||||
description: RESEARCH_DESCRIPTION,
|
||||
schema: researchInputSchema,
|
||||
execute: async (input, { toolCallId }) => dispatch(ctx, "research", toolCallId, RESEARCH_ACK, input),
|
||||
...(searchOn && {
|
||||
research: lhTool({
|
||||
name: "research",
|
||||
description: RESEARCH_DESCRIPTION,
|
||||
schema: researchInputSchema,
|
||||
execute: async (input, { toolCallId }) => dispatch(ctx, "research", toolCallId, RESEARCH_ACK, input),
|
||||
}),
|
||||
}),
|
||||
demo: lhTool({
|
||||
name: "demo",
|
||||
|
||||
+63
-79
@@ -1,9 +1,8 @@
|
||||
import { z } from "zod";
|
||||
import { lhTool } from "../lh-tool.ts";
|
||||
// The search-hit shape lives in the core port since the search wave
|
||||
// (2026-09-20) — this module keeps the historical alias so callers don't
|
||||
// churn; the tool's rewiring onto the port is the Teacher stream's.
|
||||
import type { SearchHit as WebSearchResult } from "../../core/search.ts";
|
||||
// (2026-09-20); this alias keeps callers churn-free.
|
||||
import type { SearchHit as WebSearchResult, SearchProvider } from "../../core/search.ts";
|
||||
export type { WebSearchResult };
|
||||
import { Readability } from "@mozilla/readability";
|
||||
import { parseHTML } from "linkedom";
|
||||
@@ -13,15 +12,20 @@ import TurndownService from "turndown";
|
||||
* Server-side web tools for the researcher sub-agent — a minimal port of the
|
||||
* pi extensions (~/.pi/agent/extensions/web-search + web-fetch):
|
||||
*
|
||||
* web_search(query, count?) → {title, url, snippet}[] (Google Custom Search)
|
||||
* web_search(query, count?) → {title, url, snippet}[] (injected search port)
|
||||
* web_fetch(url) → readable markdown text (Readability + Turndown,
|
||||
* Jina Reader fallback for
|
||||
* JS-rendered pages)
|
||||
*
|
||||
* Credentials come from env (the pi extension used auth.json):
|
||||
* GOOGLE_SEARCH_API_KEY, GOOGLE_CSE_ID
|
||||
* SEARCH REWIRING (search port wave, 2026-09-20): the Google CSE backend is
|
||||
* retired — web_search now runs on the injected search port (core/search.ts).
|
||||
* The live adapter (Tavily/Brave — server/search.ts) arrives as a DEPENDENCY
|
||||
* (import law: this tool depends on the core port, never server code). NO
|
||||
* silent fallback (owner decision): provider absent → search is honestly
|
||||
* unavailable, and the composition root drops the research dispatch tool on
|
||||
* the same signal (the jobTools seam, src/agent/tools/jobs.ts).
|
||||
*
|
||||
* Tool contract: NEVER throw into the agent loop — missing credentials and
|
||||
* Tool contract: NEVER throw into the agent loop — missing provider and
|
||||
* every failure path return a clear error STRING as the tool output, so the
|
||||
* researcher can adapt (or report) instead of crashing.
|
||||
*/
|
||||
@@ -29,32 +33,6 @@ import TurndownService from "turndown";
|
||||
// ── web_search ───────────────────────────────────────────────────────────────
|
||||
// (WebSearchResult is the core SearchHit, re-exported above.)
|
||||
|
||||
function searchCredentials(): { apiKey: string; cseId: string } | null {
|
||||
const apiKey = process.env.GOOGLE_SEARCH_API_KEY;
|
||||
const cseId = process.env.GOOGLE_CSE_ID;
|
||||
return apiKey && cseId ? { apiKey, cseId } : null;
|
||||
}
|
||||
|
||||
async function googleSearch(query: string, count: number, creds: { apiKey: string; cseId: string }): Promise<WebSearchResult[]> {
|
||||
const url = new URL("https://www.googleapis.com/customsearch/v1");
|
||||
url.searchParams.set("key", creds.apiKey);
|
||||
url.searchParams.set("cx", creds.cseId);
|
||||
url.searchParams.set("q", query);
|
||||
url.searchParams.set("num", String(Math.min(count, 10)));
|
||||
|
||||
const resp = await fetch(url, { signal: AbortSignal.timeout(30_000) });
|
||||
if (!resp.ok) {
|
||||
const body = await resp.text();
|
||||
throw new Error(`Google Custom Search API ${resp.status}: ${body.slice(0, 200)}`);
|
||||
}
|
||||
const data = (await resp.json()) as { items?: Array<{ title: string; link: string; snippet?: string }> };
|
||||
return (data.items ?? []).map((item) => ({
|
||||
title: item.title,
|
||||
url: item.link,
|
||||
snippet: item.snippet?.replace(/\n/g, " ") ?? "",
|
||||
}));
|
||||
}
|
||||
|
||||
// ── web_fetch ────────────────────────────────────────────────────────────────
|
||||
|
||||
const USER_AGENT =
|
||||
@@ -136,52 +114,58 @@ async function fetchAndExtract(url: string): Promise<string> {
|
||||
|
||||
// ── tool registry ────────────────────────────────────────────────────────────
|
||||
|
||||
export const webTools = {
|
||||
web_search: lhTool({
|
||||
name: "web_search",
|
||||
description:
|
||||
"Search the web via Google Custom Search. One search per call — use one call per search angle. " +
|
||||
"Returns an array of results with title, url, and snippet.",
|
||||
schema: z.object({
|
||||
query: z.string().describe("The search query (a normal search string, phrases included)"),
|
||||
count: z.number().int().min(1).max(10).optional().describe("Number of results (default 5, max 10)"),
|
||||
export function createWebTools(search: () => SearchProvider | null) {
|
||||
return {
|
||||
web_search: lhTool({
|
||||
name: "web_search",
|
||||
description: "Search the web. One search per call — use one call per search angle. " +
|
||||
"Returns an array of results with title, url, and snippet.",
|
||||
schema: z.object({
|
||||
query: z.string().describe("The search query (a normal search string, phrases included)"),
|
||||
count: z.number().int().min(1).max(10).optional().describe("Number of results (default 5, max 10)"),
|
||||
}),
|
||||
execute: async ({ query, count }): Promise<WebSearchResult[] | string> => {
|
||||
const provider = search();
|
||||
if (!provider) {
|
||||
return (
|
||||
"web_search is not available: no search engine is configured. " +
|
||||
"Set a Tavily or Brave search API key in settings. Report in your brief that web search is unavailable."
|
||||
);
|
||||
}
|
||||
try {
|
||||
return await provider.search(query, count ?? 5);
|
||||
} catch (err) {
|
||||
return `web_search failed: ${err instanceof Error ? err.message : String(err)}`;
|
||||
}
|
||||
},
|
||||
}),
|
||||
execute: async ({ query, count }): Promise<WebSearchResult[] | string> => {
|
||||
const creds = searchCredentials();
|
||||
if (!creds) {
|
||||
return (
|
||||
"web_search is not configured on this server: GOOGLE_SEARCH_API_KEY and/or GOOGLE_CSE_ID " +
|
||||
"are missing. Report in your brief that web search is unavailable."
|
||||
);
|
||||
}
|
||||
try {
|
||||
return await googleSearch(query, count ?? 5, creds);
|
||||
} catch (err) {
|
||||
return `web_search failed: ${err instanceof Error ? err.message : String(err)}`;
|
||||
}
|
||||
},
|
||||
}),
|
||||
|
||||
web_fetch: lhTool({
|
||||
name: "web_fetch",
|
||||
description:
|
||||
"Fetch a web page and extract its readable content as markdown (Readability). " +
|
||||
"Handles HTML pages and plain text; JS-rendered pages fall back to a reader service.",
|
||||
schema: z.object({
|
||||
url: z.string().describe("The URL to fetch"),
|
||||
web_fetch: lhTool({
|
||||
name: "web_fetch",
|
||||
description:
|
||||
"Fetch a web page and extract its readable content as markdown (Readability). " +
|
||||
"Handles HTML pages and plain text; JS-rendered pages fall back to a reader service.",
|
||||
schema: z.object({
|
||||
url: z.string().describe("The URL to fetch"),
|
||||
}),
|
||||
execute: async ({ url }): Promise<string> => {
|
||||
try {
|
||||
const parsed = new URL(url);
|
||||
if (parsed.protocol !== "http:" && parsed.protocol !== "https:") throw new Error("Only http(s) URLs are supported");
|
||||
} catch {
|
||||
return `web_fetch failed: invalid URL "${url}"`;
|
||||
}
|
||||
try {
|
||||
return await fetchAndExtract(url);
|
||||
} catch (err) {
|
||||
return `web_fetch failed for ${url}: ${err instanceof Error ? err.message : String(err)}`;
|
||||
}
|
||||
},
|
||||
}),
|
||||
execute: async ({ url }): Promise<string> => {
|
||||
try {
|
||||
const parsed = new URL(url);
|
||||
if (parsed.protocol !== "http:" && parsed.protocol !== "https:") throw new Error("Only http(s) URLs are supported");
|
||||
} catch {
|
||||
return `web_fetch failed: invalid URL "${url}"`;
|
||||
}
|
||||
try {
|
||||
return await fetchAndExtract(url);
|
||||
} catch (err) {
|
||||
return `web_fetch failed for ${url}: ${err instanceof Error ? err.message : String(err)}`;
|
||||
}
|
||||
},
|
||||
}),
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
/** The provider-less backstop (tests pin the shape with it; the runner
|
||||
* always injects the real seam — an absent seam reads as unavailable,
|
||||
* never a crash and never a silent alternative engine). */
|
||||
export const webTools = createWebTools(() => null);
|
||||
|
||||
@@ -11,6 +11,7 @@ import { claimSteers, deliverJobResult, setContextRespawn, setJobRespawn, setTur
|
||||
import { recordToPiMessages } from "./pi/record-adapter.ts";
|
||||
import { createPiTurnResponse } from "./pi/ui-stream.ts";
|
||||
import { effectiveSettings, MODEL_PINS, stores } from "./stores.ts";
|
||||
import { searchAvailable, searchProvider } from "./search.ts";
|
||||
import { extractionEnabled, graphContextEnabled, resolveRoleSpec } from "../core/settings-resolution.ts";
|
||||
import { beginTurn, endTurn } from "./turn-liveness.ts";
|
||||
import { persistTurn, triggerExtraction, triggerGraphContext, withMessageIds } from "./turn-service.ts";
|
||||
@@ -45,6 +46,11 @@ const jobRunner = createJobRunner({
|
||||
// per-role overrides reached rolePins but never the engines.
|
||||
engineFor: (role) =>
|
||||
piEngineForSpec(resolveRoleSpec(role, stores.settingsStore.readSettings(), MODEL_PINS)),
|
||||
// Search port wave (2026-09-20): the researcher's web_search runs on the
|
||||
// injected port — resolved FRESH per job (key rotations land immediately).
|
||||
// null = honestly unavailable; the dispatch-level drop (below) uses the
|
||||
// same signal so keyless servers never see the research tool at all.
|
||||
searchProvider,
|
||||
});
|
||||
/** The app's ONE extraction pipeline instance — both front doors share it:
|
||||
* quiz-triggered session runs (turn-service) and notes imports
|
||||
@@ -252,6 +258,9 @@ export async function handleChatRequest(
|
||||
chatStore: stores.chatStore,
|
||||
jobStore: stores.jobStore,
|
||||
spawn: (job, input) => void jobRunner.runJob(job.id, input),
|
||||
// The registry drop's signal (2026-09-20): no live engine
|
||||
// = no research tool in the teacher's registry.
|
||||
searchAvailable,
|
||||
},
|
||||
agentOpts,
|
||||
)
|
||||
|
||||
+131
-145
@@ -59,6 +59,7 @@ import { chatStore } from "../src/store/chats.ts";
|
||||
import { jobStore } from "../src/store/jobs.ts";
|
||||
import { createJobRunner, pdfModelOverride, researchModelOverride } from "../src/agent/subagents/runner.ts";
|
||||
import type { PiEngine } from "../src/agent/models.ts";
|
||||
import type { SearchProvider } from "../src/core/search.ts";
|
||||
import { PATHS } from "../config.ts";
|
||||
|
||||
const JOBS_DIR = path.join(PATHS.data, "jobs");
|
||||
@@ -357,7 +358,8 @@ describe("dispatch (async protocol §2)", () => {
|
||||
trackChat(chat.id);
|
||||
|
||||
const delivery = fakeDeliver();
|
||||
const testRunner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver });
|
||||
const search = stubSearch();
|
||||
const testRunner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver, searchProvider: () => search });
|
||||
const engine = researchCallingTeacher();
|
||||
const tools = createTeacherTools({
|
||||
chatId: chat.id,
|
||||
@@ -368,42 +370,41 @@ describe("dispatch (async protocol §2)", () => {
|
||||
const agent = (outgoing: UIMessage[]): TeachingAgent =>
|
||||
createTeacherAgent(undefined, { engine, tools, messages: recordToPiMessages(outgoing) });
|
||||
|
||||
const restoreFetch = stubFetch();
|
||||
researchModelOverride.current = researcherModel();
|
||||
try {
|
||||
const firstUser = { id: "u1", role: "user", parts: [{ type: "text", text: "teach me graph rendering" }] };
|
||||
const body = await drain(await handleChatRequest(post(chat.id, [firstUser]), agent));
|
||||
const firstUser = { id: "u1", role: "user", parts: [{ type: "text", text: "teach me graph rendering" }] };
|
||||
const body = await drain(await handleChatRequest(post(chat.id, [firstUser]), agent));
|
||||
|
||||
// The teacher's SAME turn carried on past the tool call.
|
||||
assert.ok(body.includes("Back with the brief."), "the turn continued after the dispatch");
|
||||
// The teacher's SAME turn carried on past the tool call.
|
||||
assert.ok(body.includes("Back with the brief."), "the turn continued after the dispatch");
|
||||
|
||||
const saved = chatStore.getChat(chat.id)!;
|
||||
const part = saved.messages[1]!.parts.find((p) => p.type === "tool-research");
|
||||
assert.ok(part && part.type === "tool-research", "research tool part persisted");
|
||||
assert.equal(part.state, "output-available", "the part resolved at dispatch time (not held pending)");
|
||||
assert.equal(typeof (part as { output?: unknown }).output, "string", "the output is the ack, not the result");
|
||||
assert.match(String((part as { output?: unknown }).output), /fire-and-forget/);
|
||||
const saved = chatStore.getChat(chat.id)!;
|
||||
const part = saved.messages[1]!.parts.find((p) => p.type === "tool-research");
|
||||
assert.ok(part && part.type === "tool-research", "research tool part persisted");
|
||||
assert.equal(part.state, "output-available", "the part resolved at dispatch time (not held pending)");
|
||||
assert.equal(typeof (part as { output?: unknown }).output, "string", "the output is the ack, not the result");
|
||||
assert.match(String((part as { output?: unknown }).output), /fire-and-forget/);
|
||||
|
||||
// The job record exists and the background runner completed it.
|
||||
const job = jobStore.findJobByToolCallId(chat.id, "call-r1");
|
||||
assert.ok(job, "dispatch created a job record");
|
||||
trackJob(job!.id);
|
||||
await waitFor(() => jobStore.getJob(job!.id)!.status !== "running", 8000, "research job completion");
|
||||
// The job record exists and the background runner completed it.
|
||||
const job = jobStore.findJobByToolCallId(chat.id, "call-r1");
|
||||
assert.ok(job, "dispatch created a job record");
|
||||
trackJob(job!.id);
|
||||
await waitFor(() => jobStore.getJob(job!.id)!.status !== "running", 8000, "research job completion");
|
||||
|
||||
const done = jobStore.getJob(job!.id)!;
|
||||
assert.equal(done.status, "done");
|
||||
assert.equal(done.activity, "delivered");
|
||||
const done = jobStore.getJob(job!.id)!;
|
||||
assert.equal(done.status, "done");
|
||||
assert.equal(done.activity, "delivered");
|
||||
|
||||
// The steer carries the brief + artifact for the (UI) renderer.
|
||||
assert.equal(delivery.steers.length, 1, "exactly one steer — one job, one result");
|
||||
const steer = delivery.steers[0]!;
|
||||
assert.equal(steer.metadata.subagentResult.jobId, job!.id);
|
||||
const text = steer.parts[0];
|
||||
assert.ok(text.type === "text" && text.text.includes("sigma.js v3 is current"), "steer text carries the brief");
|
||||
} finally {
|
||||
restoreFetch();
|
||||
researchModelOverride.current = undefined;
|
||||
}
|
||||
// The steer carries the brief + artifact for the (UI) renderer.
|
||||
assert.equal(delivery.steers.length, 1, "exactly one steer — one job, one result");
|
||||
const steer = delivery.steers[0]!;
|
||||
assert.equal(steer.metadata.subagentResult.jobId, job!.id);
|
||||
const text = steer.parts[0];
|
||||
assert.ok(text.type === "text" && text.text.includes("sigma.js v3 is current"), "steer text carries the brief");
|
||||
|
||||
// Search threading (2026-09-20): the researcher's web_search ran on
|
||||
// the INJECTED port — the stub saw the scripted query, no network.
|
||||
assert.deepEqual(search.queries, ["sigma.js latest version (5)"]);
|
||||
researchModelOverride.current = undefined;
|
||||
});
|
||||
});
|
||||
|
||||
@@ -429,75 +430,65 @@ function researcherModel(): PiEngine {
|
||||
]);
|
||||
}
|
||||
|
||||
function stubFetch() {
|
||||
const original = globalThis.fetch;
|
||||
globalThis.fetch = (async (input: RequestInfo | URL) => {
|
||||
const url = String(input);
|
||||
if (url.startsWith("https://www.googleapis.com/")) {
|
||||
return new Response(
|
||||
JSON.stringify({
|
||||
items: [{ title: "sigma.js", link: "https://www.sigmajs.org/", snippet: "v3 released" }],
|
||||
}),
|
||||
{ status: 200, headers: { "Content-Type": "application/json" } },
|
||||
);
|
||||
}
|
||||
return new Response("unexpected fetch: " + url, { status: 500 });
|
||||
}) as typeof fetch;
|
||||
return () => {
|
||||
globalThis.fetch = original;
|
||||
/** A recording stub provider (search port wave, 2026-09-20): the runner's
|
||||
* injected search seam — no network, queries captured for assertions. */
|
||||
function stubSearch(): SearchProvider & { queries: string[] } {
|
||||
const queries: string[] = [];
|
||||
return {
|
||||
id: "tavily",
|
||||
async search(query: string, count: number) {
|
||||
queries.push(`${query} (${count})`);
|
||||
return [{ title: "sigma.js", url: "https://www.sigmajs.org/", snippet: "v3 released" }];
|
||||
},
|
||||
queries,
|
||||
};
|
||||
}
|
||||
|
||||
describe("research runner (async protocol §3)", () => {
|
||||
it("runJob completes the job: result + sources on the job, brief on the steer, activity recorded", async () => {
|
||||
const restoreFetch = stubFetch();
|
||||
const chat = chatStore.createChat("Test: runner completion");
|
||||
trackChat(chat.id);
|
||||
|
||||
// A dispatched research part in the stored chat (the input is also
|
||||
// what the runner falls back to on a crash-recovery re-spawn).
|
||||
chatStore.saveMessages(chat.id, [
|
||||
{ id: "u1", role: "user", parts: [{ type: "text", text: "which sigma version?" }] },
|
||||
{
|
||||
id: "a1",
|
||||
role: "assistant",
|
||||
parts: [
|
||||
{ type: "text", text: "Checking." },
|
||||
{ type: "tool-research", toolCallId: "call-run1", state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." },
|
||||
],
|
||||
},
|
||||
] as never);
|
||||
|
||||
const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-run1" });
|
||||
trackJob(job.id);
|
||||
|
||||
const delivery = fakeDeliver();
|
||||
const search = stubSearch();
|
||||
const runner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver, searchProvider: () => search });
|
||||
researchModelOverride.current = researcherModel();
|
||||
try {
|
||||
const chat = chatStore.createChat("Test: runner completion");
|
||||
trackChat(chat.id);
|
||||
|
||||
// A dispatched research part in the stored chat (the input is also
|
||||
// what the runner falls back to on a crash-recovery re-spawn).
|
||||
chatStore.saveMessages(chat.id, [
|
||||
{ id: "u1", role: "user", parts: [{ type: "text", text: "which sigma version?" }] },
|
||||
{
|
||||
id: "a1",
|
||||
role: "assistant",
|
||||
parts: [
|
||||
{ type: "text", text: "Checking." },
|
||||
{ type: "tool-research", toolCallId: "call-run1", state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." },
|
||||
],
|
||||
},
|
||||
] as never);
|
||||
|
||||
const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-run1" });
|
||||
trackJob(job.id);
|
||||
|
||||
const delivery = fakeDeliver();
|
||||
const runner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver });
|
||||
researchModelOverride.current = researcherModel();
|
||||
try {
|
||||
await runner.runJob(job.id);
|
||||
} finally {
|
||||
researchModelOverride.current = undefined;
|
||||
}
|
||||
|
||||
// The job is done, with the activity line from the tool call.
|
||||
const done = jobStore.getJob(job.id)!;
|
||||
assert.equal(done.status, "done");
|
||||
assert.equal(done.activity, "delivered", "the fake deliver flips the status like the pipeline");
|
||||
const result = done.result as ResearchResult;
|
||||
assert.equal(result.brief, BRIEF);
|
||||
assert.ok(result.sources.length > 0, "sources parsed from the brief");
|
||||
assert.equal(result.sources[0]!.url, "https://www.sigmajs.org/docs");
|
||||
|
||||
// The activity trail survives on... the last recorded activity is
|
||||
// the deliver flip, but the search line was recorded during the run:
|
||||
// the job store keeps only the latest; the search fired before it.
|
||||
// (We ran the search; the runner recorded it — check the store's
|
||||
// history via the final state only.)
|
||||
await runner.runJob(job.id);
|
||||
} finally {
|
||||
restoreFetch();
|
||||
researchModelOverride.current = undefined;
|
||||
}
|
||||
|
||||
// The job is done, with the activity line from the tool call.
|
||||
const done = jobStore.getJob(job.id)!;
|
||||
assert.equal(done.status, "done");
|
||||
assert.equal(done.activity, "delivered", "the fake deliver flips the status like the pipeline");
|
||||
const result = done.result as ResearchResult;
|
||||
assert.equal(result.brief, BRIEF);
|
||||
assert.ok(result.sources.length > 0, "sources parsed from the brief");
|
||||
assert.equal(result.sources[0]!.url, "https://www.sigmajs.org/docs");
|
||||
|
||||
// The activity trail: the last recorded activity is the deliver
|
||||
// flip; the search line fired before it. The search itself ran on
|
||||
// the INJECTED port — the stub saw the query (no network):
|
||||
assert.deepEqual(search.queries, ["sigma.js latest version (5)"]);
|
||||
});
|
||||
|
||||
it("a failing runner still steers (the teacher learns the outcome) and the job lands error", async () => {
|
||||
@@ -542,37 +533,37 @@ describe("research runner (async protocol §3)", () => {
|
||||
});
|
||||
|
||||
it("runJob resolves the input from the transcript when dispatch-time input is absent (crash recovery)", async () => {
|
||||
const restoreFetch = stubFetch();
|
||||
const chat = chatStore.createChat("Test: transcript input fallback");
|
||||
trackChat(chat.id);
|
||||
chatStore.saveMessages(chat.id, [
|
||||
{ id: "u1", role: "user", parts: [{ type: "text", text: "go" }] },
|
||||
{
|
||||
id: "a1",
|
||||
role: "assistant",
|
||||
parts: [{ type: "tool-research", toolCallId: "call-fb1", state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." }],
|
||||
},
|
||||
] as never);
|
||||
const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-fb1" });
|
||||
trackJob(job.id);
|
||||
|
||||
const delivery = fakeDeliver();
|
||||
const runner = createJobRunner({
|
||||
chatStore,
|
||||
jobStore,
|
||||
deliver: delivery.deliver,
|
||||
searchProvider: () => stubSearch(),
|
||||
});
|
||||
researchModelOverride.current = researcherModel();
|
||||
try {
|
||||
const chat = chatStore.createChat("Test: transcript input fallback");
|
||||
trackChat(chat.id);
|
||||
chatStore.saveMessages(chat.id, [
|
||||
{ id: "u1", role: "user", parts: [{ type: "text", text: "go" }] },
|
||||
{
|
||||
id: "a1",
|
||||
role: "assistant",
|
||||
parts: [{ type: "tool-research", toolCallId: "call-fb1", state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." }],
|
||||
},
|
||||
] as never);
|
||||
const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-fb1" });
|
||||
trackJob(job.id);
|
||||
|
||||
const delivery = fakeDeliver();
|
||||
const runner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver });
|
||||
researchModelOverride.current = researcherModel();
|
||||
try {
|
||||
// no input passed — as a recovery re-spawn would call it
|
||||
await runner.runJob(job.id);
|
||||
} finally {
|
||||
researchModelOverride.current = undefined;
|
||||
}
|
||||
|
||||
const done = jobStore.getJob(job.id)!;
|
||||
assert.equal(done.status, "done", "the researcher ran from the transcript-resolved input");
|
||||
assert.equal((done.result as ResearchResult).brief, BRIEF);
|
||||
// no input passed — as a recovery re-spawn would call it
|
||||
await runner.runJob(job.id);
|
||||
} finally {
|
||||
restoreFetch();
|
||||
researchModelOverride.current = undefined;
|
||||
}
|
||||
|
||||
const done = jobStore.getJob(job.id)!;
|
||||
assert.equal(done.status, "done", "the researcher ran from the transcript-resolved input");
|
||||
assert.equal((done.result as ResearchResult).brief, BRIEF);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1203,33 +1194,28 @@ describe("abort (src/server/jobs.ts abortJob)", () => {
|
||||
});
|
||||
|
||||
it("the usage folded so far survives the kill (money channel outlives the abort)", async () => {
|
||||
const restoreFetch = stubFetch();
|
||||
const { jobId } = newChatWithResearchJob("abort keeps usage", "call-abort2");
|
||||
setTurnRunner(async () => 200);
|
||||
researchModelOverride.current = oneStepThenHangModel();
|
||||
const runner = createJobRunner({ chatStore, jobStore, deliver: async () => {}, searchProvider: () => stubSearch() });
|
||||
try {
|
||||
const { jobId } = newChatWithResearchJob("abort keeps usage", "call-abort2");
|
||||
setTurnRunner(async () => 200);
|
||||
researchModelOverride.current = oneStepThenHangModel();
|
||||
const runner = createJobRunner({ chatStore, jobStore, deliver: async () => {} });
|
||||
try {
|
||||
const settled = runner.runJob(jobId, RESEARCH_INPUT);
|
||||
await waitFor(() => (jobStore.getJob(jobId)!.usage?.steps ?? 0) === 1, 8000, "first step folded");
|
||||
abortJob(jobId);
|
||||
await settled;
|
||||
const job = jobStore.getJob(jobId)!;
|
||||
assert.equal(job.status, "error");
|
||||
assert.equal(job.error, "Aborted by learner");
|
||||
assert.equal(job.usage!.steps, 1, "usage-so-far persisted by the runner's finally");
|
||||
// model slots wave: the stamp is the ENGINE's actual model,
|
||||
// provider-qualified — the scripted faux engine here.
|
||||
assert.equal(job.usage!.model, "faux:faux-1");
|
||||
// faux estimates tokens from the prompt/response text — the
|
||||
// pinned fact is that the ONE completed step's spend survived.
|
||||
assert.ok(job.usage!.inputTokens! > 0, "input tokens billed");
|
||||
assert.ok(job.usage!.outputTokens! > 0, "output tokens billed");
|
||||
} finally {
|
||||
researchModelOverride.current = undefined;
|
||||
}
|
||||
const settled = runner.runJob(jobId, RESEARCH_INPUT);
|
||||
await waitFor(() => (jobStore.getJob(jobId)!.usage?.steps ?? 0) === 1, 8000, "first step folded");
|
||||
abortJob(jobId);
|
||||
await settled;
|
||||
const job = jobStore.getJob(jobId)!;
|
||||
assert.equal(job.status, "error");
|
||||
assert.equal(job.error, "Aborted by learner");
|
||||
assert.equal(job.usage!.steps, 1, "usage-so-far persisted by the runner's finally");
|
||||
// model slots wave: the stamp is the ENGINE's actual model,
|
||||
// provider-qualified — the scripted faux engine here.
|
||||
assert.equal(job.usage!.model, "faux:faux-1");
|
||||
// faux estimates tokens from the prompt/response text — the
|
||||
// pinned fact is that the ONE completed step's spend survived.
|
||||
assert.ok(job.usage!.inputTokens! > 0, "input tokens billed");
|
||||
assert.ok(job.usage!.outputTokens! > 0, "output tokens billed");
|
||||
} finally {
|
||||
restoreFetch();
|
||||
researchModelOverride.current = undefined;
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
/**
|
||||
* Search rewiring (search port wave, 2026-09-20) — the web_search tool's
|
||||
* port injection + the research dispatch tool's registry-level drop.
|
||||
*
|
||||
* Google CSE is dead: the tool runs on the injected search port
|
||||
* (core/search.ts), the runner threads the seam (searchProvider — covered
|
||||
* end-to-end in jobs.test.ts), and jobTools drops the research tool when
|
||||
* ctx.searchAvailable reports no live engine (owner decision: registry
|
||||
* drop, no doomed dispatch, no silent fallback).
|
||||
*
|
||||
* No network, no shared data files — the jobStore fake below is in-memory.
|
||||
*/
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { describe, it } from "node:test";
|
||||
import { createWebTools } from "../src/agent/tools/web.ts";
|
||||
import { jobTools } from "../src/agent/tools/jobs.ts";
|
||||
import type { JobStore } from "../src/core/ports.ts";
|
||||
import type { JobRecord } from "../src/core/contracts.ts";
|
||||
import type { SearchHit, SearchProvider } from "../src/core/search.ts";
|
||||
|
||||
function stubProvider(hits: SearchHit[] = [{ title: "t", url: "https://e.com/", snippet: "s" }]): SearchProvider {
|
||||
return {
|
||||
id: "tavily",
|
||||
async search() {
|
||||
return hits;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("web_search on the injected port (web.ts)", () => {
|
||||
it("null provider → the honest unavailability string, never a crash", async () => {
|
||||
const { web_search } = createWebTools(() => null);
|
||||
const out = await web_search.execute!({ query: "anything" }, { toolCallId: "c1" });
|
||||
assert.match(String(out), /not available: no search engine is configured/);
|
||||
assert.match(String(out), /Tavily or Brave/);
|
||||
});
|
||||
|
||||
it("a live provider returns its hits (count defaults to 5)", async () => {
|
||||
const seen: Array<{ query: string; count: number }> = [];
|
||||
const provider: SearchProvider = {
|
||||
id: "tavily",
|
||||
async search(query, count) {
|
||||
seen.push({ query, count });
|
||||
return [{ title: "sigma.js", url: "https://www.sigmajs.org/", snippet: "v3 released" }];
|
||||
},
|
||||
};
|
||||
const { web_search } = createWebTools(() => provider);
|
||||
const out = await web_search.execute!({ query: "sigma.js latest version" }, { toolCallId: "c1" });
|
||||
assert.deepEqual(seen, [{ query: "sigma.js latest version", count: 5 }]);
|
||||
assert.deepEqual(out, [{ title: "sigma.js", url: "https://www.sigmajs.org/", snippet: "v3 released" }]);
|
||||
});
|
||||
|
||||
it("provider errors surface as error strings, never thrown", async () => {
|
||||
const provider: SearchProvider = {
|
||||
id: "brave",
|
||||
async search() {
|
||||
throw new Error("Brave search API 429: rate limited");
|
||||
},
|
||||
};
|
||||
const { web_search } = createWebTools(() => provider);
|
||||
const out = await web_search.execute!({ query: "q" }, { toolCallId: "c1" });
|
||||
assert.match(String(out), /web_search failed: Brave search API 429: rate limited/);
|
||||
});
|
||||
|
||||
it("the resolution is FRESH per call — a key appearing mid-session takes effect", async () => {
|
||||
let provider: SearchProvider | null = null;
|
||||
const { web_search } = createWebTools(() => provider);
|
||||
const first = await web_search.execute!({ query: "q" }, { toolCallId: "c1" });
|
||||
assert.match(String(first), /not available/);
|
||||
provider = stubProvider();
|
||||
const second = await web_search.execute!({ query: "q" }, { toolCallId: "c2" });
|
||||
assert.ok(Array.isArray(second), "the live provider answered on the very next call");
|
||||
});
|
||||
|
||||
it("web_fetch needs no search engine — it stays independent", async () => {
|
||||
const { web_fetch } = createWebTools(() => null);
|
||||
assert.ok(web_fetch, "web_fetch present with no provider");
|
||||
// (Full fetch behavior is out of scope here — no network in tests.)
|
||||
const out = await web_fetch.execute!({ url: "not a url" }, { toolCallId: "c1" });
|
||||
assert.match(String(out), /web_fetch failed: invalid URL/);
|
||||
});
|
||||
});
|
||||
|
||||
/** Minimal in-memory JobStore — the drop tests never touch the fs. */
|
||||
function fakeJobStore(): JobStore {
|
||||
const jobs = new Map<string, JobRecord>();
|
||||
return {
|
||||
createJob(init) {
|
||||
const job: JobRecord = {
|
||||
id: `job_${jobs.size + 1}`,
|
||||
kind: init.kind,
|
||||
chatId: init.chatId,
|
||||
toolCallId: init.toolCallId,
|
||||
status: "running",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
activity: "",
|
||||
} as JobRecord;
|
||||
jobs.set(job.id, job);
|
||||
return job;
|
||||
},
|
||||
getJob: (id) => jobs.get(id) ?? null,
|
||||
updateJob: (id, patch) => {
|
||||
const job = jobs.get(id);
|
||||
if (!job) return null;
|
||||
return Object.assign(job, patch);
|
||||
},
|
||||
listJobs: () => [...jobs.values()],
|
||||
findJobByToolCallId: () => null,
|
||||
};
|
||||
}
|
||||
|
||||
function fakeCtx(available?: () => boolean) {
|
||||
return {
|
||||
chatId: "chat-drop-test",
|
||||
jobStore: fakeJobStore(),
|
||||
spawn: () => {},
|
||||
...(available ? { searchAvailable: available } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
describe("the research registry drop (jobTools)", () => {
|
||||
it("searchUnavailable: false → research drops from the registry (the other job tools stay)", () => {
|
||||
const tools = jobTools(fakeCtx(() => false));
|
||||
assert.equal("research" in tools, false, "no research tool — no doomed dispatch");
|
||||
assert.ok(tools.demo && tools.visual && tools.pdf, "the search-less server keeps its other sub-agents");
|
||||
});
|
||||
|
||||
it("searchAvailable: true → research present", () => {
|
||||
const tools = jobTools(fakeCtx(() => true));
|
||||
assert.ok(tools.research, "a keyed server exposes research");
|
||||
});
|
||||
|
||||
it("absent seam = available (tests + any wiring without search)", () => {
|
||||
const tools = jobTools(fakeCtx());
|
||||
assert.ok(tools.research, "the seam is optional — default is honest availability");
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user