Merge pull request #1425 from corbitsdev/cl-10652-arktype-228-generate-inputschema-via-tojsonschema-delete-ha

refactor(tools): derive tool inputSchemas from arktype types

Author: TheGreatAxios

bun.lock

@@ -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",

package.json

@@ -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<boolean> {
   try {
     await stat(path);

src/plugins/delete-file-plugin.test.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 {

src/plugins/delete-file-plugin.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.

src/session/stream-journal.test.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<void> {
+  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<void> = 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}",

src/session/stream-journal.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("<h1>Heading</h1>");
   });
 });
+
+// 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"],
+    });
+  });
+});

src/tools/web-fetch.test.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 {