diff --git a/src/agents/pi-embedded-runner.ts b/src/agents/pi-embedded-runner.ts index 164a27343..49ee978bb 100644 --- a/src/agents/pi-embedded-runner.ts +++ b/src/agents/pi-embedded-runner.ts @@ -378,16 +378,16 @@ export async function runEmbeddedPiAgent(params: { const apiKey = await getApiKeyForModel(model, authStorage); authStorage.setRuntimeApiKey(model.provider, apiKey); - const thinkingLevel = mapThinkingLevel(params.thinkLevel); + const thinkingLevel = mapThinkingLevel(params.thinkLevel); - logVerbose( - `embedded run start: runId=${params.runId} sessionId=${params.sessionId} provider=${provider} model=${modelId} surface=${params.surface ?? "unknown"}`, - ); + logVerbose( + `embedded run start: runId=${params.runId} sessionId=${params.sessionId} provider=${provider} model=${modelId} surface=${params.surface ?? "unknown"}`, + ); - await fs.mkdir(resolvedWorkspace, { recursive: true }); - await ensureSessionHeader({ - sessionFile: params.sessionFile, - sessionId: params.sessionId, + await fs.mkdir(resolvedWorkspace, { recursive: true }); + await ensureSessionHeader({ + sessionFile: params.sessionFile, + sessionId: params.sessionId, cwd: resolvedWorkspace, }); @@ -510,20 +510,23 @@ export async function runEmbeddedPiAgent(params: { }); let abortWarnTimer: NodeJS.Timeout | undefined; - const abortTimer = setTimeout(() => { - defaultRuntime.warn?.( - `embedded run timeout: runId=${params.runId} sessionId=${params.sessionId} timeoutMs=${params.timeoutMs}`, - ); - abortRun(); - if (!abortWarnTimer) { - abortWarnTimer = setTimeout(() => { - if (!session.isStreaming) return; - defaultRuntime.warn?.( - `embedded run abort still streaming: runId=${params.runId} sessionId=${params.sessionId}`, - ); - }, 10_000); - } - }, Math.max(1, params.timeoutMs)); + const abortTimer = setTimeout( + () => { + defaultRuntime.warn?.( + `embedded run timeout: runId=${params.runId} sessionId=${params.sessionId} timeoutMs=${params.timeoutMs}`, + ); + abortRun(); + if (!abortWarnTimer) { + abortWarnTimer = setTimeout(() => { + if (!session.isStreaming) return; + defaultRuntime.warn?.( + `embedded run abort still streaming: runId=${params.runId} sessionId=${params.sessionId}`, + ); + }, 10_000); + } + }, + Math.max(1, params.timeoutMs), + ); let messagesSnapshot: AgentMessage[] = []; let sessionIdUsed = session.sessionId; diff --git a/src/agents/pi-embedded-subscribe.ts b/src/agents/pi-embedded-subscribe.ts index 539a48b22..07768eae0 100644 --- a/src/agents/pi-embedded-subscribe.ts +++ b/src/agents/pi-embedded-subscribe.ts @@ -6,13 +6,13 @@ import { createToolDebouncer, formatToolAggregate, } from "../auto-reply/tool-meta.js"; +import { logVerbose } from "../globals.js"; import { emitAgentEvent } from "../infra/agent-events.js"; import { splitMediaFromOutput } from "../media/parse.js"; import { extractAssistantText, inferToolMetaFromArgs, } from "./pi-embedded-utils.js"; -import { logVerbose } from "../globals.js"; const THINKING_TAG_RE = /<\s*\/?\s*think(?:ing)?\s*>/gi; const THINKING_OPEN_RE = /<\s*think(?:ing)?\s*>/i; diff --git a/src/agents/pi-tools.ts b/src/agents/pi-tools.ts index 5de878a9c..aeb6e06e7 100644 --- a/src/agents/pi-tools.ts +++ b/src/agents/pi-tools.ts @@ -144,12 +144,11 @@ function mergePropertySchemas(existing: unknown, incoming: unknown): unknown { function cleanSchemaForGemini(schema: unknown): unknown { if (!schema || typeof schema !== "object") return schema; if (Array.isArray(schema)) return schema.map(cleanSchemaForGemini); - + const obj = schema as Record; const hasAnyOf = "anyOf" in obj && Array.isArray(obj.anyOf); - const hasConst = "const" in obj; const cleaned: Record = {}; - + for (const [key, value] of Object.entries(obj)) { // Skip unsupported schema features for Gemini: // - patternProperties: not in OpenAPI 3.0 subset @@ -158,44 +157,48 @@ function cleanSchemaForGemini(schema: unknown): unknown { // Gemini doesn't support patternProperties - skip it continue; } - + // Convert const to enum (Gemini doesn't support const) if (key === "const") { cleaned.enum = [value]; continue; } - + // Skip 'type' if we have 'anyOf' — Gemini doesn't allow both if (key === "type" && hasAnyOf) { continue; } - + if (key === "properties" && value && typeof value === "object") { // Recursively clean nested properties const props = value as Record; cleaned[key] = Object.fromEntries( - Object.entries(props).map(([k, v]) => [k, cleanSchemaForGemini(v)]) + Object.entries(props).map(([k, v]) => [k, cleanSchemaForGemini(v)]), ); } else if (key === "items" && value && typeof value === "object") { // Recursively clean array items schema cleaned[key] = cleanSchemaForGemini(value); } else if (key === "anyOf" && Array.isArray(value)) { // Clean each anyOf variant - cleaned[key] = value.map(v => cleanSchemaForGemini(v)); + cleaned[key] = value.map((v) => cleanSchemaForGemini(v)); } else if (key === "oneOf" && Array.isArray(value)) { // Clean each oneOf variant - cleaned[key] = value.map(v => cleanSchemaForGemini(v)); + cleaned[key] = value.map((v) => cleanSchemaForGemini(v)); } else if (key === "allOf" && Array.isArray(value)) { // Clean each allOf variant - cleaned[key] = value.map(v => cleanSchemaForGemini(v)); - } else if (key === "additionalProperties" && value && typeof value === "object") { + cleaned[key] = value.map((v) => cleanSchemaForGemini(v)); + } else if ( + key === "additionalProperties" && + value && + typeof value === "object" + ) { // Recursively clean additionalProperties schema cleaned[key] = cleanSchemaForGemini(value); } else { cleaned[key] = value; } } - + return cleaned; } @@ -205,16 +208,20 @@ function normalizeToolParameters(tool: AnyAgentTool): AnyAgentTool { ? (tool.parameters as Record) : undefined; if (!schema) return tool; - + // If schema already has type + properties (no top-level anyOf to merge), // still clean it for Gemini compatibility - if ("type" in schema && "properties" in schema && !Array.isArray(schema.anyOf)) { + if ( + "type" in schema && + "properties" in schema && + !Array.isArray(schema.anyOf) + ) { return { ...tool, parameters: cleanSchemaForGemini(schema), }; } - + if (!Array.isArray(schema.anyOf)) return tool; const mergedProperties: Record = {}; const requiredCounts = new Map(); diff --git a/src/auto-reply/reply.ts b/src/auto-reply/reply.ts index dda1eb09b..0daa397db 100644 --- a/src/auto-reply/reply.ts +++ b/src/auto-reply/reply.ts @@ -782,10 +782,7 @@ export async function getReplyFromConfig( const typingIntervalSeconds = typeof configuredTypingSeconds === "number" ? configuredTypingSeconds : 6; const typingIntervalMs = typingIntervalSeconds * 1000; - const typingTtlMs = Math.min( - Math.max(15_000, typingIntervalMs * 5), - 60_000, - ); + const typingTtlMs = Math.min(Math.max(15_000, typingIntervalMs * 5), 60_000); const cleanupTyping = () => { if (typingTtlTimer) { clearTimeout(typingTtlTimer); diff --git a/src/hooks/gmail-setup-utils.test.ts b/src/hooks/gmail-setup-utils.test.ts index ae0618b98..9768843dc 100644 --- a/src/hooks/gmail-setup-utils.test.ts +++ b/src/hooks/gmail-setup-utils.test.ts @@ -22,7 +22,7 @@ describe("resolvePythonExecutablePath", () => { const shim = path.join(shimDir, "python3"); await fs.writeFile( shim, - `#!/bin/sh\nif [ \"$1\" = \"-c\" ]; then\n echo \"${realPython}\"\n exit 0\nfi\nexit 1\n`, + `#!/bin/sh\nif [ "$1" = "-c" ]; then\n echo "${realPython}"\n exit 0\nfi\nexit 1\n`, "utf-8", ); await fs.chmod(shim, 0o755); diff --git a/src/hooks/gmail-setup-utils.ts b/src/hooks/gmail-setup-utils.ts index bf0b7e132..83249db9a 100644 --- a/src/hooks/gmail-setup-utils.ts +++ b/src/hooks/gmail-setup-utils.ts @@ -58,7 +58,9 @@ function ensureGcloudOnPath(): boolean { return false; } -export async function resolvePythonExecutablePath(): Promise { +export async function resolvePythonExecutablePath(): Promise< + string | undefined +> { if (cachedPythonPath !== undefined) { return cachedPythonPath ?? undefined; } @@ -171,7 +173,14 @@ export async function ensureSubscription( pushEndpoint: string, ) { const describe = await runGcloudCommand( - ["pubsub", "subscriptions", "describe", subscription, "--project", projectId], + [ + "pubsub", + "subscriptions", + "describe", + subscription, + "--project", + projectId, + ], 30_000, ); if (describe.code === 0) { diff --git a/src/logging.ts b/src/logging.ts index faaed464e..ac3b24246 100644 --- a/src/logging.ts +++ b/src/logging.ts @@ -396,10 +396,12 @@ const SUBSYSTEM_COLORS = [ "magenta", "red", ] as const; -const SUBSYSTEM_COLOR_OVERRIDES: Record = - { - "gmail-watcher": "blue", - }; +const SUBSYSTEM_COLOR_OVERRIDES: Record< + string, + (typeof SUBSYSTEM_COLORS)[number] +> = { + "gmail-watcher": "blue", +}; const SUBSYSTEM_PREFIXES_TO_DROP = ["gateway", "providers"] as const; const SUBSYSTEM_MAX_SEGMENTS = 2; diff --git a/src/web/inbound.ts b/src/web/inbound.ts index cc3b463bf..b7f0eb0d2 100644 --- a/src/web/inbound.ts +++ b/src/web/inbound.ts @@ -190,7 +190,7 @@ export async function monitorWebInbox(options: { await sock.readMessages([ { remoteJid, id, participant, fromMe: false }, ]); - if (shouldLogVerbose()) { + if (shouldLogVerbose()) { const suffix = participant ? ` (participant ${participant})` : ""; logVerbose( `Marked message ${id} as read for ${remoteJid}${suffix}`, diff --git a/test/gateway.multi.e2e.test.ts b/test/gateway.multi.e2e.test.ts index 22a63d77c..7d5b08790 100644 --- a/test/gateway.multi.e2e.test.ts +++ b/test/gateway.multi.e2e.test.ts @@ -1,12 +1,15 @@ +import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; import fs from "node:fs/promises"; import { request as httpRequest } from "node:http"; import net from "node:net"; import os from "node:os"; import path from "node:path"; import { afterAll, describe, expect, it } from "vitest"; -import { approveNodePairing, listNodePairing } from "../src/infra/node-pairing.js"; +import { + approveNodePairing, + listNodePairing, +} from "../src/infra/node-pairing.js"; type GatewayInstance = { name: string; @@ -96,7 +99,9 @@ const spawnGatewayInstance = async (name: string): Promise => { const port = await getFreePort(); const bridgePort = await getFreePort(); const hookToken = `token-${name}-${randomUUID()}`; - const homeDir = await fs.mkdtemp(path.join(os.tmpdir(), `clawdis-e2e-${name}-`)); + const homeDir = await fs.mkdtemp( + path.join(os.tmpdir(), `clawdis-e2e-${name}-`), + ); const configDir = path.join(homeDir, ".clawdis"); await fs.mkdir(configDir, { recursive: true }); const configPath = path.join(configDir, "clawdis.json"); @@ -226,8 +231,11 @@ const runCliJson = async ( child.stderr?.setEncoding("utf8"); child.stdout?.on("data", (d) => stdout.push(String(d))); child.stderr?.on("data", (d) => stderr.push(String(d))); - const result = await new Promise<{ code: number | null; signal: string | null }>( - (resolve) => child.once("exit", (code, signal) => resolve({ code, signal })), + const result = await new Promise<{ + code: number | null; + signal: string | null; + }>((resolve) => + child.once("exit", (code, signal) => resolve({ code, signal })), ); const out = stdout.join("").trim(); if (result.code !== 0) { @@ -249,41 +257,43 @@ const runCliJson = async ( const postJson = async (url: string, body: unknown) => { const payload = JSON.stringify(body); const parsed = new URL(url); - return await new Promise<{ status: number; json: unknown }>((resolve, reject) => { - const req = httpRequest( - { - method: "POST", - hostname: parsed.hostname, - port: Number(parsed.port), - path: `${parsed.pathname}${parsed.search}`, - headers: { - "Content-Type": "application/json", - "Content-Length": Buffer.byteLength(payload), + return await new Promise<{ status: number; json: unknown }>( + (resolve, reject) => { + const req = httpRequest( + { + method: "POST", + hostname: parsed.hostname, + port: Number(parsed.port), + path: `${parsed.pathname}${parsed.search}`, + headers: { + "Content-Type": "application/json", + "Content-Length": Buffer.byteLength(payload), + }, }, - }, - (res) => { - let data = ""; - res.setEncoding("utf8"); - res.on("data", (chunk) => { - data += chunk; - }); - res.on("end", () => { - let json: unknown = null; - if (data.trim()) { - try { - json = JSON.parse(data); - } catch { - json = data; + (res) => { + let data = ""; + res.setEncoding("utf8"); + res.on("data", (chunk) => { + data += chunk; + }); + res.on("end", () => { + let json: unknown = null; + if (data.trim()) { + try { + json = JSON.parse(data); + } catch { + json = data; + } } - } - resolve({ status: res.statusCode ?? 0, json }); - }); - }, - ); - req.on("error", reject); - req.write(payload); - req.end(); - }); + resolve({ status: res.statusCode ?? 0, json }); + }); + }, + ); + req.on("error", reject); + req.write(payload); + req.end(); + }, + ); }; const createLineReader = (socket: net.Socket) => { @@ -437,26 +447,14 @@ describe("gateway multi-instance e2e", () => { const [nodeListA, nodeListB] = (await Promise.all([ runCliJson( - [ - "nodes", - "status", - "--json", - "--url", - `ws://127.0.0.1:${gwA.port}`, - ], + ["nodes", "status", "--json", "--url", `ws://127.0.0.1:${gwA.port}`], { CLAWDIS_GATEWAY_TOKEN: "", CLAWDIS_GATEWAY_PASSWORD: "", }, ), runCliJson( - [ - "nodes", - "status", - "--json", - "--url", - `ws://127.0.0.1:${gwB.port}`, - ], + ["nodes", "status", "--json", "--url", `ws://127.0.0.1:${gwB.port}`], { CLAWDIS_GATEWAY_TOKEN: "", CLAWDIS_GATEWAY_PASSWORD: "", @@ -466,17 +464,13 @@ describe("gateway multi-instance e2e", () => { expect( nodeListA.nodes?.some( (n) => - n.nodeId === "node-a" && - n.connected === true && - n.paired === true, + n.nodeId === "node-a" && n.connected === true && n.paired === true, ), ).toBe(true); expect( nodeListB.nodes?.some( (n) => - n.nodeId === "node-b" && - n.connected === true && - n.paired === true, + n.nodeId === "node-b" && n.connected === true && n.paired === true, ), ).toBe(true);