diff --git a/bun.lock b/bun.lock index b13867cad..ff57a593b 100644 --- a/bun.lock +++ b/bun.lock @@ -247,10 +247,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", @@ -261,9 +262,9 @@ "packages": { "@anthropic-ai/sandbox-runtime": ["@anthropic-ai/sandbox-runtime@0.0.79", "", { "dependencies": { "@pondwader/socks5-server": "^1.0.10", "commander": "^12.1.0", "node-forge": "^1.4.0", "zod": "^3.25.0" }, "bin": { "srt": "dist/cli.js" } }, "sha512-WjBmS9fbTpnQwQkM9SU+JjhM2CcXo9eWEELktY1100RMvTy/jPhP44yZ3NXhh+GHB350OgPbM3+FubE5WVRtSA=="], - "@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=="], @@ -535,9 +536,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 cc1905945..df95e1834 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/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}", 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; +}