From f6e5b9374a119688cd385aebd4ac92a5061c6217 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 14 Oct 2026 08:39:17 -0700 Subject: [PATCH 1/2] refactor(tools): derive tool inputSchemas from arktype types Hand-written JSON Schema for web_fetch, web_search and delete_file duplicated the arktype arg types and had already drifted (no minLength, documented timeout intent living outside the schema). The types are now the single source via toJsonSchema with descriptions carried as arktype metadata; served shape keeps its exact key set (dialect: null) and generated-vs-hand diffs are pinned in tests. arktype moves 2.2.3 -> 2.2.8; an override keeps the first-party and vendored consumers on one type identity (two arktype versions in one program defeat portable type naming). --- bun.lock | 11 ++--- package.json | 5 ++- src/plugins/delete-file-plugin.test.ts | 26 ++++++++++- src/plugins/delete-file-plugin.ts | 24 +++++----- src/tools/web-fetch.test.ts | 42 ++++++++++++++++++ src/tools/web-fetch.ts | 34 +++++++------- src/tools/web-search.test.ts | 59 +++++++++++++++++++++++++ src/tools/web-search.ts | 61 ++++++++++++-------------- src/util/tool-input-schema.ts | 10 +++++ 9 files changed, 201 insertions(+), 71 deletions(-) create mode 100644 src/util/tool-input-schema.ts diff --git a/bun.lock b/bun.lock index 71b65b73c..261a518a3 100644 --- a/bun.lock +++ b/bun.lock @@ -246,10 +246,11 @@ "@intx/storage-isogit": "workspace:*", "@intx/tools-posix": "workspace:*", "@intx/types": "workspace:*", + "arktype": "^2.2.8", }, "catalog": { "@types/semver": "^7.7.1", - "arktype": "^2.2.3", + "arktype": "^2.2.8", "better-auth": "^1.4.18", "drizzle-orm": "^0.45.1", "hono": "^4.11.9", @@ -259,9 +260,9 @@ "tar": "^7.5.1", }, "packages": { - "@ark/schema": ["@ark/schema@0.56.2", "", { "dependencies": { "@ark/util": "0.56.2" } }, "sha512-Qx4D2JFbBWpntiHZaTv7bGG4H/M2rigiknezKg/WVyDSaLdE4YCcWAOoFB7pjjDqHbbV2OqRfntm1nnXvwMexg=="], + "@ark/schema": ["@ark/schema@0.56.7", "", { "dependencies": { "@ark/util": "0.56.7" } }, "sha512-jmoOJlYIc4k1uv8oYBDNOFnAsKk2a2D9t/D6npF2JUfW4ecPB0v6ce6lruKg7lkBxIBA8Z7UtxfmvephYbhUGQ=="], - "@ark/util": ["@ark/util@0.56.2", "", {}, "sha512-9kU2sUE38FZEGG7l3hamYMBieLYEJh2L1mrYD2eXpT+78EnQSV1bhjxJhnxGBMSTbtwpBSDNSK+K60WvaI/DTQ=="], + "@ark/util": ["@ark/util@0.56.7", "", {}, "sha512-z8pVnrPk/x9GC18yOp5wCe8rR2+Z6us4CTSg7dvxdsDLuA2xXYq+QETD1fO6f3fMuXe+PCjpvtCnhTW9RHLgEw=="], "@babel/code-frame": ["@babel/code-frame@7.29.7", "", { "dependencies": { "@babel/helper-validator-identifier": "^7.29.7", "js-tokens": "^4.0.0", "picocolors": "^1.1.1" } }, "sha512-Aup7aUOfpbAUg2ROOJN6Iw5f9DMBlzu0mIkm/malLQFN/YQgO48wCj0Kxa3sEHJvPVFg7siR+qRInwXd2qhQKw=="], @@ -531,9 +532,9 @@ "ansi-regex": ["ansi-regex@6.3.0", "", {}, "sha512-WpDfL7NO6j7tH88IDBNVdUJxDh9nmCteAVW9dsep846XdwF4naCBK+/tGLX3KJgcpgMRXCFlTM2hKGoK9FsdrQ=="], - "arkregex": ["arkregex@0.0.8", "", { "dependencies": { "@ark/util": "0.56.2" } }, "sha512-PJcx6G1kQTgLKPUbeYlYecDRaKq15AMSGVajlKFYWlPeJRQL+j3dKE6tyMs40HZ99djS1l9Vhl3ezAHy9JBIqQ=="], + "arkregex": ["arkregex@0.0.13", "", { "dependencies": { "@ark/util": "0.56.7" } }, "sha512-c17sqCqiiZauxyoU1KZ2zbzCgeghGmc8dR2CrdQnlEtpq3bLhRLoLrxTL52M/8yQD8e/i1/IBRpRCNAuqJa7xQ=="], - "arktype": ["arktype@2.2.3", "", { "dependencies": { "@ark/schema": "0.56.2", "@ark/util": "0.56.2", "arkregex": "0.0.8" } }, "sha512-7W+0RLTUNJiBFIIZXwOQxSR8Z273IAd6IvqBeG9+gHnQKFsIx2C0iOtGTmMrPnlX4qLXyc5+ll7A0BIj9WrbTg=="], + "arktype": ["arktype@2.2.8", "", { "dependencies": { "@ark/schema": "0.56.7", "@ark/util": "0.56.7", "arkregex": "0.0.13" } }, "sha512-tZyLza7kO2vN8BXBKcj4t43IVMe8/ZyqlITH0chZv7FfEw+edLovpY922tqcK4mCjTpf4oNo2xV6np7tquP1ww=="], "async-lock": ["async-lock@1.4.1", "", {}, "sha512-Az2ZTpuytrtqENulXwO3GGv1Bztugx6TT37NIo7imr/Qo0gsYiGtSdBa2B6fsXhTpVZDNfu1Qn3pk531e3q+nQ=="], diff --git a/package.json b/package.json index e7b6a7606..2fe40ea15 100644 --- a/package.json +++ b/package.json @@ -72,11 +72,12 @@ "@intx/tools-posix": "workspace:*", "@intx/mailbox": "workspace:*", "@intx/harness": "workspace:*", - "@intx/mime": "workspace:*" + "@intx/mime": "workspace:*", + "arktype": "^2.2.8" }, "catalog": { "@types/semver": "^7.7.1", - "arktype": "^2.2.3", + "arktype": "^2.2.8", "better-auth": "^1.4.18", "drizzle-orm": "^0.45.1", "hono": "^4.11.9", diff --git a/src/plugins/delete-file-plugin.test.ts b/src/plugins/delete-file-plugin.test.ts index a550753aa..bd786ecb7 100644 --- a/src/plugins/delete-file-plugin.test.ts +++ b/src/plugins/delete-file-plugin.test.ts @@ -15,7 +15,10 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import type { ToolCall, ToolResult } from "@intx/types/runtime"; -import { deleteFilePlugin } from "./delete-file-plugin.js"; +import { + deleteFilePlugin, + DELETE_FILE_DEFINITION, +} from "./delete-file-plugin.js"; import { pathEscapePlugin } from "./path-escape-plugin.js"; import { createPermissionGate } from "../permission/gate.js"; import { permissionPlugin } from "./permission-plugin.js"; @@ -24,6 +27,27 @@ function call(path: unknown): ToolCall { return { id: "delete-call", name: "delete_file", arguments: { path } }; } +// The served schema must stay the toJsonSchema projection of DeleteFileArgs: +// pinning it here asserts the arktype-generated args type remains the single +// source of truth (descriptions flow from the type's @ metadata, not from a +// hand-written JSON copy). +describe("DELETE_FILE_DEFINITION.inputSchema", () => { + test("is generated from DeleteFileArgs via toJsonSchema", () => { + expect(DELETE_FILE_DEFINITION.inputSchema).toEqual({ + type: "object", + properties: { + path: { + type: "string", + description: + "Path to the file to delete, relative to the working directory or absolute", + minLength: 1, + }, + }, + required: ["path"], + }); + }); +}); + async function exists(path: string): Promise { try { await stat(path); diff --git a/src/plugins/delete-file-plugin.ts b/src/plugins/delete-file-plugin.ts index ca8b2a7cd..bddf23e79 100644 --- a/src/plugins/delete-file-plugin.ts +++ b/src/plugins/delete-file-plugin.ts @@ -5,25 +5,25 @@ import type { ExtraTool, ToolPlugin } from "@intx/tools-posix"; import type { ToolCall, ToolResult } from "@intx/types/runtime"; import { resolveWorkspacePath } from "../permission/path-restriction.js"; import type { RootsProvider } from "../permission/worktree-roots.js"; +import { toolInputSchema } from "../util/tool-input-schema.js"; import { formatChangeDiff } from "./change-diff.js"; -const DeleteFileArgs = type({ path: "string>0" }); +const DeleteFileArgs = type({ + path: [ + "string>0", + "@", + { + description: + "Path to the file to delete, relative to the working directory or absolute", + }, + ], +}); export const DELETE_FILE_DEFINITION = { name: "delete_file", description: "Delete one file. Returns success when the file is deleted or already absent. Refuses directories; use this instead of shell rm for file deletion.", - inputSchema: { - type: "object", - properties: { - path: { - type: "string", - description: - "Path to the file to delete, relative to the working directory or absolute", - }, - }, - required: ["path"], - }, + inputSchema: toolInputSchema(DeleteFileArgs), }; function errorResult(callId: string, message: string): ToolResult { diff --git a/src/tools/web-fetch.test.ts b/src/tools/web-fetch.test.ts index b037154ef..72d19da04 100644 --- a/src/tools/web-fetch.test.ts +++ b/src/tools/web-fetch.test.ts @@ -7,6 +7,7 @@ import { createWebFetchTool, runWebFetch, MAX_FETCH_BYTES, + webFetchDefinition, } from "./web-fetch.js"; let server: Server; @@ -264,3 +265,44 @@ describe("createExaMCPWebFetchTool", () => { expect(html.content).toContain("

Heading

"); }); }); + +// The served schema must stay the toJsonSchema projection of WebFetchArgs: +// pinning it here asserts the arktype-generated args type remains the single +// source of truth (descriptions flow from the type's @ metadata, not from a +// hand-written JSON copy). +describe("webFetchDefinition.inputSchema", () => { + test("is generated from WebFetchArgs via toJsonSchema", () => { + expect(webFetchDefinition.inputSchema).toEqual({ + type: "object", + properties: { + url: { + type: "string", + description: "URL to fetch (http or https only)", + minLength: 1, + }, + format: { + anyOf: [ + { + const: "html", + description: "Output format. Defaults to markdown.", + }, + { + const: "markdown", + description: "Output format. Defaults to markdown.", + }, + { + const: "text", + description: "Output format. Defaults to markdown.", + }, + ], + description: "Output format. Defaults to markdown.", + }, + timeout: { + type: "number", + description: "Timeout in seconds (default 30, max 120).", + }, + }, + required: ["url"], + }); + }); +}); diff --git a/src/tools/web-fetch.ts b/src/tools/web-fetch.ts index 857ab31ec..790b20242 100644 --- a/src/tools/web-fetch.ts +++ b/src/tools/web-fetch.ts @@ -6,6 +6,7 @@ import type { ToolCall, ToolDefinition, ToolResult } from "@intx/types/runtime"; import { checkUrlForSsrf } from "./ssrf-guard.js"; import { htmlToMarkdown, htmlToText } from "./html-convert.js"; import { COMMAND_NAME } from "../branding.js"; +import { toolInputSchema } from "../util/tool-input-schema.js"; import type { MCPClient } from "../mcp/client.js"; import pkg from "../../package.json" with { type: "json" }; @@ -36,9 +37,19 @@ function honestUserAgent(): string { } const WebFetchArgs = type({ - url: "string>0", - "format?": "'text' | 'markdown' | 'html'", - "timeout?": "number", + url: ["string>0", "@", { description: "URL to fetch (http or https only)" }], + "format?": [ + "'text' | 'markdown' | 'html'", + "@", + { description: "Output format. Defaults to markdown." }, + ], + "timeout?": [ + "number", + "@", + { + description: `Timeout in seconds (default ${DEFAULT_TIMEOUT_S}, max ${MAX_TIMEOUT_S}).`, + }, + ], }); type WebFetchFormat = "text" | "markdown" | "html"; @@ -78,22 +89,7 @@ export const webFetchDefinition: ToolDefinition = { name: "web_fetch", description: "Fetch a public web page (docs, articles) and return it as markdown by default. For a URL on a host a connected MCP server covers, use that server via tool_search first; web_fetch is for public pages, not authenticated apps.", - inputSchema: { - type: "object", - properties: { - url: { type: "string", description: "URL to fetch (http or https only)" }, - format: { - type: "string", - enum: ["text", "markdown", "html"], - description: "Output format. Defaults to markdown.", - }, - timeout: { - type: "number", - description: `Timeout in seconds (default ${DEFAULT_TIMEOUT_S}, max ${MAX_TIMEOUT_S}).`, - }, - }, - required: ["url"], - }, + inputSchema: toolInputSchema(WebFetchArgs), }; function acceptHeaderFor(format: WebFetchFormat): string { diff --git a/src/tools/web-search.test.ts b/src/tools/web-search.test.ts index 236a957c5..2992477e4 100644 --- a/src/tools/web-search.test.ts +++ b/src/tools/web-search.test.ts @@ -36,6 +36,7 @@ const { resolveWebSearchProvider, EXA_MCP_URL, PARALLEL_MCP_URL, + webSearchDefinition, } = await import("./web-search.js"); const { createAgentToolset } = await import("../agent/tools.js"); const { resolveMcpServers } = await import("../config/index.js"); @@ -180,3 +181,61 @@ describe("createWebSearchTool", () => { expect(connectConfigs.length).toBe(0); }); }); + +// The served schema must stay the toJsonSchema projection of WebSearchArgs: +// pinning it here asserts the arktype-generated args type remains the single +// source of truth (descriptions flow from the type's @ metadata, not from a +// hand-written JSON copy). +describe("webSearchDefinition.inputSchema", () => { + test("is generated from WebSearchArgs via toJsonSchema", () => { + expect(webSearchDefinition.inputSchema).toEqual({ + type: "object", + properties: { + query: { type: "string", description: "Search query", minLength: 1 }, + numResults: { + type: "number", + description: "Maximum number of results (default 8).", + }, + type: { + anyOf: [ + { + const: "auto", + description: "Search depth/strategy. Defaults to auto.", + }, + { + const: "deep", + description: "Search depth/strategy. Defaults to auto.", + }, + { + const: "fast", + description: "Search depth/strategy. Defaults to auto.", + }, + ], + description: "Search depth/strategy. Defaults to auto.", + }, + livecrawl: { + anyOf: [ + { + const: "fallback", + description: + "Whether to prefer or only fall back to a live crawl. Defaults to fallback.", + }, + { + const: "preferred", + description: + "Whether to prefer or only fall back to a live crawl. Defaults to fallback.", + }, + ], + description: + "Whether to prefer or only fall back to a live crawl. Defaults to fallback.", + }, + contextMaxCharacters: { + type: "number", + description: + "Cap on returned context length in characters (default 10000).", + }, + }, + required: ["query"], + }); + }); +}); diff --git a/src/tools/web-search.ts b/src/tools/web-search.ts index f47541d82..a03dee729 100644 --- a/src/tools/web-search.ts +++ b/src/tools/web-search.ts @@ -4,6 +4,7 @@ import type { AgentTool } from "@intx/agent"; import type { ToolDefinition } from "@intx/types/runtime"; import { connectMCPServer, type MCPClient } from "../mcp/client.js"; +import { toolInputSchema } from "../util/tool-input-schema.js"; import type { MCPServerConfig } from "../config/settings.js"; import { EXA_MCP_URL } from "../mcp/exa.js"; @@ -16,43 +17,39 @@ export const PARALLEL_MCP_URL = "https://search.parallel.ai/mcp"; export type WebSearchProviderId = "exa" | "parallel"; const WebSearchArgs = type({ - query: "string>0", - "numResults?": "number", - "type?": "'auto' | 'fast' | 'deep'", - "livecrawl?": "'fallback' | 'preferred'", - "contextMaxCharacters?": "number", + query: ["string>0", "@", { description: "Search query" }], + "numResults?": [ + "number", + "@", + { description: "Maximum number of results (default 8)." }, + ], + "type?": [ + "'auto' | 'fast' | 'deep'", + "@", + { description: "Search depth/strategy. Defaults to auto." }, + ], + "livecrawl?": [ + "'fallback' | 'preferred'", + "@", + { + description: + "Whether to prefer or only fall back to a live crawl. Defaults to fallback.", + }, + ], + "contextMaxCharacters?": [ + "number", + "@", + { + description: + "Cap on returned context length in characters (default 10000).", + }, + ], }); export const webSearchDefinition: ToolDefinition = { name: "web_search", description: "Search the web; returns ranked results as text.", - inputSchema: { - type: "object", - properties: { - query: { type: "string", description: "Search query" }, - numResults: { - type: "number", - description: "Maximum number of results (default 8).", - }, - type: { - type: "string", - enum: ["auto", "fast", "deep"], - description: "Search depth/strategy. Defaults to auto.", - }, - livecrawl: { - type: "string", - enum: ["fallback", "preferred"], - description: - "Whether to prefer or only fall back to a live crawl. Defaults to fallback.", - }, - contextMaxCharacters: { - type: "number", - description: - "Cap on returned context length in characters (default 10000).", - }, - }, - required: ["query"], - }, + inputSchema: toolInputSchema(WebSearchArgs), }; // Provider selection is env-driven (settings.env now makes this configuration diff --git a/src/util/tool-input-schema.ts b/src/util/tool-input-schema.ts new file mode 100644 index 000000000..72f79e834 --- /dev/null +++ b/src/util/tool-input-schema.ts @@ -0,0 +1,10 @@ +/** + * Serves a tool's inputSchema from its declared arktype args type so the type + * stays the single source of truth for what the model sees. `dialect: null` + * drops the `$schema` doc key the wire shape is validated against without. + */ +export function toolInputSchema(argsType: { + toJsonSchema(options?: { dialect: string | null }): unknown; +}): Record { + return argsType.toJsonSchema({ dialect: null }) as Record; +} From 88acffc560078539f09ac21f8db4da80d1aa4574 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 15 Oct 2026 13:00:49 -0700 Subject: [PATCH 2/2] fix(session): publish partial records without a torn line appendFile creates partial.jsonl before the JSON line is written, so a reader can observe empty or truncated JSON. Rewrite through a temp file and rename, and serialize overlapping flushes so one publish cannot drop the other. --- src/session/stream-journal.test.ts | 54 ++++++++++++++++++++++++++++ src/session/stream-journal.ts | 57 ++++++++++++++++++++++++++---- 2 files changed, 105 insertions(+), 6 deletions(-) diff --git a/src/session/stream-journal.test.ts b/src/session/stream-journal.test.ts index 0e6513d29..999c9fc72 100644 --- a/src/session/stream-journal.test.ts +++ b/src/session/stream-journal.test.ts @@ -124,6 +124,60 @@ describe("createCycleTextRecorder", () => { expect(records[0]?.error?.message).toBe("aborted"); }); + test("a reader never observes a torn partial.jsonl record", async () => { + // appendFile creates the file before the line lands, so a poller can read + // empty or truncated JSON. Every observed byte must already parse. + for (let trial = 0; trial < 30; trial++) { + const temp = createTempDir("stream-journal-race-"); + try { + const recorder = createCycleTextRecorder(() => temp.dir); + recorder.handleEvent({ + type: "inference.error", + data: { + error: { + category: "rate_limit", + message: "429 too many requests", + statusCode: 429, + }, + }, + } as unknown as ReactorEmittedEvent); + + const started = Date.now(); + for (;;) { + let raw: string; + try { + raw = await readFile(join(temp.dir, PARTIAL_FILE), "utf8"); + } catch (err) { + if (Date.now() - started >= 2_000) throw err; + continue; + } + const records = parsePartialRaw(raw); + expect(records).toHaveLength(1); + expect(records[0]?.text).toBe(""); + expect(records[0]?.error?.statusCode).toBe(429); + break; + } + } finally { + temp.cleanup(); + } + } + }); + + test("two flushes append both records", async () => { + const recorder = createCycleTextRecorder(() => dir); + recorder.handleEvent(delta("one")); + await recorder.flush("cancelled"); + recorder.handleEvent(delta("two")); + await recorder.flush("interrupted"); + + const records = await readPartialRecords(); + expect(records.map((record) => record.text)).toEqual(["one", "two"]); + expect(records.map((record) => record.reason)).toEqual([ + "cancelled", + "interrupted", + ]); + }); + test("inference.error with empty cycle text still writes a partial with the error payload", async () => { // Live failures can emit inference.error with no streamed text; the // partial must still land so category/message survive. diff --git a/src/session/stream-journal.ts b/src/session/stream-journal.ts index 367d78e20..90e57d567 100644 --- a/src/session/stream-journal.ts +++ b/src/session/stream-journal.ts @@ -1,4 +1,4 @@ -import { appendFile } from "node:fs/promises"; +import { readFile, rename, rm, writeFile } from "node:fs/promises"; import { join } from "node:path"; import type { ReactorEmittedEvent } from "@intx/inference"; @@ -21,6 +21,44 @@ export const PARTIAL_FILE = "partial.jsonl"; /** Cap on retained cycle text; older text is dropped from the front. */ export const CYCLE_TEXT_CAP_CHARS = 262_144; +function isEnoent(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code?: unknown }).code === "ENOENT" + ); +} + +let partialTmpCounter = 0; + +/** + * Append one JSONL line so readers never observe a torn record. `appendFile` + * creates the destination before the bytes land; a poller then reads empty or + * truncated JSON. Rewrite the whole file through a temp sibling and `rename` + * so the path appears only with complete lines. + */ +async function appendPartialLine(dir: string, line: string): Promise { + const target = join(dir, PARTIAL_FILE); + let prior = ""; + try { + prior = await readFile(target, "utf8"); + } catch (err) { + if (!isEnoent(err)) throw err; + } + const tmp = join( + dir, + `.${PARTIAL_FILE}.${process.pid}.${(partialTmpCounter += 1)}.tmp`, + ); + try { + await writeFile(tmp, prior + line, "utf8"); + await rename(tmp, target); + } catch (err) { + await rm(tmp, { force: true }).catch(() => undefined); + throw err; + } +} + /** Append a streamed token to the cycle buffer, keeping only the tail. */ export function appendCycleText( text: string, @@ -105,6 +143,9 @@ export function createCycleTextRecorder( let cycleText = ""; let cycleThinkingText = ""; let closed = false; + // Overlapping flushes (inference.error is not awaited) must not both read + // the same prior bytes and rename away the other line. + let publishChain: Promise = Promise.resolve(); const writeRecord = async ( reason: PartialFlushReason, @@ -132,12 +173,16 @@ export function createCycleTextRecorder( record.thinkingText = thinkingText; } if (error !== undefined) record.error = error; + const line = `${JSON.stringify(record)}\n`; + const run = publishChain.then(() => + appendPartialLine(resolveContextDir(), line), + ); + publishChain = run.then( + () => undefined, + () => undefined, + ); try { - await appendFile( - join(resolveContextDir(), PARTIAL_FILE), - `${JSON.stringify(record)}\n`, - "utf8", - ); + await run; } catch (err) { getLogger([LOG_NAMESPACE_ROOT, "session", "partial"]).warn( "failed to write partial stream output: {error}",