diff --git a/src/main/database/migrations/081-connector-executions.ts b/src/main/database/migrations/081-connector-executions.ts new file mode 100644 index 00000000..348eee8a --- /dev/null +++ b/src/main/database/migrations/081-connector-executions.ts @@ -0,0 +1,18 @@ +export const connectorExecutionsSchema = ` + create table if not exists connector_executions ( + id text primary key, + org_id text not null references organizations(id) on delete cascade, + agent_id text not null, + run_id text, + action text not null, + arguments_hash text not null, + state text not null check (state in ('pending', 'succeeded', 'failed', 'unknown', 'resolved')), + detail text, + created_at text not null, + updated_at text not null + ); + create index if not exists connector_executions_call on connector_executions(org_id, action, arguments_hash, state); + + insert into schema_migrations(version, name, applied_at) + values (81, 'connector_executions', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index 266cb5ce..c024dd05 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -79,6 +79,7 @@ import { modelChecksSchema } from "./migrations/077-model-checks"; import { requestErrorCodesSchema } from "./migrations/078-request-error-codes"; import { eraseRetiredCredentialsSchema } from "./migrations/079-erase-retired-credentials"; import { runCapabilityWarningsSchema } from "./migrations/080-run-capability-warnings"; +import { connectorExecutionsSchema } from "./migrations/081-connector-executions"; interface Migration { version: number; @@ -167,6 +168,7 @@ const MIGRATIONS: Migration[] = [ { version: 78, name: "request_error_codes", sql: requestErrorCodesSchema }, { version: 79, name: "erase_retired_credentials", sql: eraseRetiredCredentialsSchema }, { version: 80, name: "run_capability_warnings", sql: runCapabilityWarningsSchema }, + { version: 81, name: "connector_executions", sql: connectorExecutionsSchema }, ]; const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version; diff --git a/src/main/services/composio/composio-client.ts b/src/main/services/composio/composio-client.ts index 11687b66..5a1c48fc 100644 --- a/src/main/services/composio/composio-client.ts +++ b/src/main/services/composio/composio-client.ts @@ -29,6 +29,13 @@ export interface ComposioConnectedAccount { /** Composio answers 200 for business failures too; only `successful` tells them apart. */ export type ComposioToolExecutionResult = { successful: true; data: unknown } | { successful: false; error: string }; +/** Composio answered with an HTTP error status; below 500 the request was refused, so nothing ran. */ +export class ComposioHttpError extends Error { + constructor(message: string, readonly status: number) { + super(message); + } +} + const MAX_RESPONSE_BYTES = 2_000_000; const REQUEST_TIMEOUT_MS = 30_000; const EXECUTE_TIMEOUT_MS = 120_000; @@ -204,7 +211,7 @@ export class ComposioClient { return; } if (response.statusCode && response.statusCode >= 400) { - finish(new Error(`Composio API ${method} ${url} returned ${response.statusCode}: ${responseData.slice(0, 200)}`)); + finish(new ComposioHttpError(`Composio API ${method} ${url} returned ${response.statusCode}: ${responseData.slice(0, 200)}`, response.statusCode)); return; } try { diff --git a/src/main/services/composio/composio-tool-bridge.ts b/src/main/services/composio/composio-tool-bridge.ts index d5ad8e99..7b9fd428 100644 --- a/src/main/services/composio/composio-tool-bridge.ts +++ b/src/main/services/composio/composio-tool-bridge.ts @@ -1,6 +1,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { ToolDefinition, ToolCall } from "../provider-tool-turn"; -import { ComposioClient, type ComposioToolSchema } from "./composio-client"; +import { ComposioClient, ComposioHttpError, type ComposioToolExecutionResult, type ComposioToolSchema } from "./composio-client"; +import { beginConnectorExecution, settleConnectorExecution } from "./connector-executions"; import type { IntegrationKeyStore } from "../integration-key-store"; import { ConnectorGrantsService } from "./connector-grants"; import { currentRunId } from "../run-control"; @@ -48,6 +49,7 @@ export function loadComposioTools(db: DatabaseSync, keys: IntegrationKeyStore, o properties: { action: { type: "string", description: "The action's name exactly as found, e.g. GITHUB_CREATE_A_PULL_REQUEST." }, arguments: { type: "object", description: "The action's input, following the schema the search returned." }, + confirm_not_done: { type: "boolean", description: "Only after an identical call's reply was lost and you checked in the service that it did not take effect." }, }, required: ["action", "arguments"], }, @@ -113,11 +115,31 @@ async function runAction(db: DatabaseSync, apiKey: string, orgId: string, agentI if (!account) return `ERROR: The "${toolkit}" connector is no longer connected. Ask the board to reconnect it on the Connectors page.`; const input = typeof args.arguments === "object" && args.arguments !== null && !Array.isArray(args.arguments) ? args.arguments as Record : null; if (!input) return "ERROR: arguments must be an object matching the action's input."; - const result = await new ComposioClient(apiKey).executeTool({ toolName: action.name, arguments: input, userId: getComposioUserId(db, orgId), connectedAccountId: account.id }); + if (args.confirm_not_done !== undefined && typeof args.confirm_not_done !== "boolean") return "ERROR: confirm_not_done must be true or false."; + const started = beginConnectorExecution(db, { orgId, agentId, runId: currentRunId(), action: action.name, args: input, confirmedNotDone: args.confirm_not_done === true }); + if ("blockedBy" in started) { + const lost = started.blockedBy.detail ? `: ${started.blockedBy.detail}` : ""; + return `ERROR: An identical ${action.name} call at ${started.blockedBy.createdAt} may already have run; its reply was lost${lost}. Check in ${toolkit} whether it took effect. If it did not, call again with confirm_not_done: true.`; + } + let result: ComposioToolExecutionResult; + try { + result = await new ComposioClient(apiKey).executeTool({ toolName: action.name, arguments: input, userId: getComposioUserId(db, orgId), connectedAccountId: account.id }); + } catch (error) { + const reason = (error instanceof Error ? error.message : String(error)).replace(/\s+/g, " ").replace(/\.$/, ""); + if (error instanceof ComposioHttpError && error.status < 500) { + settleConnectorExecution(db, started.id, "failed", reason); + throw error; + } + settleConnectorExecution(db, started.id, "unknown", reason); + console.warn(`[connectors] ${action.name} outcome unknown: ${reason}`); + return `ERROR: ${action.name} may or may not have run; its reply was lost (${reason}). Check in ${toolkit} whether it took effect before calling it again.`; + } if (!result.successful) { + settleConnectorExecution(db, started.id, "failed", result.error); console.warn(`[connectors] ${action.name} failed: ${result.error.replace(/\s+/g, " ")}`); return `ERROR: ${action.name} failed: ${result.error}`; } + settleConnectorExecution(db, started.id, "succeeded", null); return JSON.stringify(result.data); } diff --git a/src/main/services/composio/connector-executions.ts b/src/main/services/composio/connector-executions.ts new file mode 100644 index 00000000..2f55ab6d --- /dev/null +++ b/src/main/services/composio/connector-executions.ts @@ -0,0 +1,45 @@ +import { createHash } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +import { createId } from "../../../shared/ids"; + +/** How an action run ended: unknown means it was sent and its reply was lost, so it may have taken effect. */ +export type ConnectorExecutionOutcome = "succeeded" | "failed" | "unknown"; + +/** Records an action run before it is sent; an identical call whose outcome is pending or unknown blocks it until the agent confirms it did not happen. */ +export function beginConnectorExecution( + db: DatabaseSync, + input: { orgId: string; agentId: string; runId: string | null; action: string; args: Record; confirmedNotDone: boolean }, +): { id: string } | { blockedBy: { createdAt: string; detail: string | null } } { + const hash = createHash("sha256").update(canonicalJson(input.args)).digest("hex"); + const open = db.prepare(` + select created_at, detail from connector_executions + where org_id = ? and action = ? and arguments_hash = ? and state in ('pending', 'unknown') order by rowid desc limit 1 + `).get(input.orgId, input.action, hash) as { created_at: string; detail: string | null } | undefined; + const now = new Date().toISOString(); + if (open && !input.confirmedNotDone) return { blockedBy: { createdAt: open.created_at, detail: open.detail } }; + if (open) { + db.prepare("update connector_executions set state = 'resolved', updated_at = ? where org_id = ? and action = ? and arguments_hash = ? and state in ('pending', 'unknown')") + .run(now, input.orgId, input.action, hash); + } + const id = createId("cex"); + db.prepare(` + insert into connector_executions(id, org_id, agent_id, run_id, action, arguments_hash, state, created_at, updated_at) + values (?, ?, ?, ?, ?, ?, 'pending', ?, ?) + `).run(id, input.orgId, input.agentId, input.runId, input.action, hash, now, now); + return { id }; +} + +export function settleConnectorExecution(db: DatabaseSync, id: string, outcome: ConnectorExecutionOutcome, detail: string | null): void { + db.prepare("update connector_executions set state = ?, detail = ?, updated_at = ? where id = ? and state = 'pending'") + .run(outcome, detail, new Date().toISOString(), id); +} + +/** JSON with object keys sorted at every level, so the same arguments always hash the same. */ +export function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`; + if (value && typeof value === "object") { + const entries = Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${canonicalJson((value as Record)[key])}`); + return `{${entries.join(",")}}`; + } + return JSON.stringify(value); +} diff --git a/tests/unit/composio-tool-pages.test.ts b/tests/unit/composio-tool-pages.test.ts index 3ebe1389..5b812505 100644 --- a/tests/unit/composio-tool-pages.test.ts +++ b/tests/unit/composio-tool-pages.test.ts @@ -20,6 +20,10 @@ vi.mock("electron", () => ({ request.end = () => { requested.push(url); if (url.includes("/tools/execute/GITHUB_HANG")) return; + if (url.includes("/tools/execute/GITHUB_DROP")) { + request.emit("error", new Error("socket hang up")); + return; + } const query = new URL(url).searchParams; const page = url.includes("/tools/execute/") ? executions.get(url.split("/tools/execute/")[1]) ?? { data: { number: 42 }, error: null, successful: true } : pages.get(`${query.get("toolkit_slug")}:${query.get("cursor") ?? ""}`); const response = new EventEmitter() as EventEmitter & { statusCode: number }; @@ -40,6 +44,7 @@ const { OrganizationService } = await import("../../src/main/services/organizati const { IntegrationKeyStore } = await import("../../src/main/services/integration-key-store"); const { ConnectorGrantsService } = await import("../../src/main/services/composio/connector-grants"); const { recordRunAction } = await import("../../src/main/services/run-journal"); +const { beginConnectorExecution } = await import("../../src/main/services/composio/connector-executions"); const tool = (slug: string) => ({ slug, name: slug, description: "", input_parameters: {}, toolkit: { slug: "github" } }); @@ -119,7 +124,38 @@ describe("agents reach every action through search and run", () => { executions.set("GITHUB_ODD", { data: { ok: 1 } }); const h = company("ak_failed"); expect(await h.call("composio_run_action", { action: "GITHUB_MERGE", arguments: {} })).toBe("ERROR: GITHUB_MERGE failed: Repository not found"); - expect(await h.call("composio_run_action", { action: "GITHUB_ODD", arguments: {} })).toBe("ERROR: Composio returned no execution status for GITHUB_ODD."); + expect(await h.call("composio_run_action", { action: "GITHUB_ODD", arguments: {} })).toBe("ERROR: GITHUB_ODD may or may not have run; its reply was lost (Composio returned no execution status for GITHUB_ODD). Check in github whether it took effect before calling it again."); + h.connection.close(); + }); + + test("a call whose reply was lost is not repeated until the agent confirms it did not happen", async () => { + pages.set("github:", { items: [tool("GITHUB_DROP"), tool("GITHUB_MERGE")], next_cursor: null }); + executions.set("GITHUB_MERGE", { data: {}, error: "Repository not found", successful: false }); + const h = company("ak_lost"); + const lost = await h.call("composio_run_action", { action: "GITHUB_DROP", arguments: { body: "hi", to: "a" } }); + expect(lost).toBe("ERROR: GITHUB_DROP may or may not have run; its reply was lost (socket hang up). Check in github whether it took effect before calling it again."); + const sent = written.length; + const again = await h.call("composio_run_action", { action: "GITHUB_DROP", arguments: { to: "a", body: "hi" } }); + expect(again).toMatch(/^ERROR: An identical GITHUB_DROP call at .+ may already have run; its reply was lost: socket hang up\. Check in github whether it took effect\. If it did not, call again with confirm_not_done: true\.$/); + expect(written.length).toBe(sent); + expect(await h.call("composio_run_action", { action: "GITHUB_DROP", arguments: { to: "a", body: "hi" }, confirm_not_done: "true" })).toBe("ERROR: confirm_not_done must be true or false."); + expect(written.length).toBe(sent); + expect(await h.call("composio_run_action", { action: "GITHUB_DROP", arguments: { to: "b", body: "hi" } })).toContain("may or may not have run"); + expect(await h.call("composio_run_action", { action: "GITHUB_DROP", arguments: { to: "a", body: "hi" }, confirm_not_done: true })).toContain("may or may not have run"); + expect(written.length).toBe(sent + 2); + // A definite failure does not block the same call again. + expect(await h.call("composio_run_action", { action: "GITHUB_MERGE", arguments: {} })).toBe("ERROR: GITHUB_MERGE failed: Repository not found"); + expect(await h.call("composio_run_action", { action: "GITHUB_MERGE", arguments: {} })).toBe("ERROR: GITHUB_MERGE failed: Repository not found"); + const states = h.connection.db.prepare("select action, state from connector_executions order by rowid").all() as Array<{ action: string; state: string }>; + expect(states.map((row) => `${row.action}:${row.state}`)).toEqual(["GITHUB_DROP:resolved", "GITHUB_DROP:unknown", "GITHUB_DROP:unknown", "GITHUB_MERGE:failed", "GITHUB_MERGE:failed"]); + h.connection.close(); + }); + + test("a call left pending by a crash blocks the same call", async () => { + pages.set("github:", { items: [tool("GITHUB_SEND")], next_cursor: null }); + const h = company("ak_crash"); + beginConnectorExecution(h.connection.db, { orgId: h.orgId, agentId: h.ceoId, runId: null, action: "GITHUB_SEND", args: { text: "x" }, confirmedNotDone: false }); + expect(await h.call("composio_run_action", { action: "GITHUB_SEND", arguments: { text: "x" } })).toMatch(/^ERROR: An identical GITHUB_SEND call at .+ may already have run; its reply was lost\. Check in github/); h.connection.close(); }); diff --git a/tests/unit/domain-services.test.ts b/tests/unit/domain-services.test.ts index 975e9c1a..f08fa97c 100644 --- a/tests/unit/domain-services.test.ts +++ b/tests/unit/domain-services.test.ts @@ -24,7 +24,7 @@ describe("domain services", () => { const connection = createConnection(); const rows = connection.db.prepare("select version from schema_migrations order by version asc").all() as Array<{ version: number }>; - expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80]); + expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81]); connection.close(); });