Q99: connector actions keep an execution record so a lost reply is not resent

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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'));
+`;

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;

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 {

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<string, unknown> : 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);
 }
 

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<string, unknown>; 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<string, unknown>)[key])}`);
+    return `{${entries.join(",")}}`;
+  }
+  return JSON.stringify(value);
+}

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();
   });
 

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();
   });