From 736c45c0eec4a16ae35e42b050b9ff07399e9737 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 16:32:46 -0400 Subject: [PATCH] Q100: reconnecting keeps the working connector account until the new one is authorized Co-Authored-By: Claude Opus 5.5 --- .../082-connector-pending-account.ts | 6 + src/main/database/schema.ts | 2 + .../composio/composio-connection-service.ts | 88 +++++++++---- src/renderer/src/i18n/zh.ts | 1 + .../src/pages/connectors/ConnectedList.tsx | 1 + .../pages/connectors/useConnectorCatalog.ts | 2 +- src/shared/domain.ts | 2 + tests/unit/composio-reconnect.test.ts | 120 ++++++++++++++++++ tests/unit/domain-services.test.ts | 2 +- 9 files changed, 194 insertions(+), 30 deletions(-) create mode 100644 src/main/database/migrations/082-connector-pending-account.ts create mode 100644 tests/unit/composio-reconnect.test.ts diff --git a/src/main/database/migrations/082-connector-pending-account.ts b/src/main/database/migrations/082-connector-pending-account.ts new file mode 100644 index 00000000..734433b1 --- /dev/null +++ b/src/main/database/migrations/082-connector-pending-account.ts @@ -0,0 +1,6 @@ +export const connectorPendingAccountSchema = ` + alter table connector_accounts add column pending_account_id text; + + insert into schema_migrations(version, name, applied_at) + values (82, 'connector_pending_account', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index c024dd05..6421aa20 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -80,6 +80,7 @@ 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"; +import { connectorPendingAccountSchema } from "./migrations/082-connector-pending-account"; interface Migration { version: number; @@ -169,6 +170,7 @@ const MIGRATIONS: Migration[] = [ { version: 79, name: "erase_retired_credentials", sql: eraseRetiredCredentialsSchema }, { version: 80, name: "run_capability_warnings", sql: runCapabilityWarningsSchema }, { version: 81, name: "connector_executions", sql: connectorExecutionsSchema }, + { version: 82, name: "connector_pending_account", sql: connectorPendingAccountSchema }, ]; const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version; diff --git a/src/main/services/composio/composio-connection-service.ts b/src/main/services/composio/composio-connection-service.ts index 655c14ad..8817ac91 100644 --- a/src/main/services/composio/composio-connection-service.ts +++ b/src/main/services/composio/composio-connection-service.ts @@ -1,12 +1,13 @@ import type { DatabaseSync } from "node:sqlite"; import type { ConnectorCatalogEntry, ConnectComposioToolkitResult } from "../../../shared/commands"; -import { ComposioClient } from "./composio-client"; +import { ComposioClient, type ComposioConnectedAccount } from "./composio-client"; import { IntegrationKeyStore } from "../integration-key-store"; import { getComposioUserId } from "./composio-tool-bridge"; import { connectorDisplayName, connectorLogoUrl } from "./connector-brands"; import { normalizeSlug } from "./connector-grants"; const CATALOG_TTL_MS = 10 * 60 * 1000; +const WAITING_STATUSES = new Set(["INITIALIZING", "INITIATED"]); let catalogCache: { fetchedAt: number; keys: string; entries: ConnectorCatalogEntry[] } | null = null; @@ -34,15 +35,7 @@ export class ComposioConnectionService { return this.catalog(this.requireClient()); } - /** - * Starts (or restarts) the company's connection for a toolkit: initiates a - * connected account against the toolkit's configured auth config via the - * link flow and records the pending account locally as INITIALIZING — the - * poll that follows adopts Composio's real status. Any previous account - * for the toolkit is deleted first so reconnecting replaces instead of - * accumulating accounts in Composio. The returned redirect URL completes - * authorization (or API-key entry) in the browser. - */ + /** Starts a toolkit's connection; a reconnect waits beside an active account until refresh finds it authorized. */ connectToolkit(orgId: string, toolkitSlug: string): Promise { return this.serialize(orgId, () => this.connectToolkitNow(orgId, toolkitSlug)); } @@ -61,15 +54,22 @@ export class ComposioConnectionService { const entry = (await this.catalog(client)).find((item) => item.slug === slug); if (!entry) throw new Error(`No auth config for "${slug}" exists in the Composio project.`); const previous = this.db.prepare( - "select id from connector_accounts where org_id = ? and toolkit_slug = ?", - ).get(orgId, slug) as { id: string } | undefined; - if (previous) await client.deleteConnectedAccount(previous.id); + "select id, status, pending_account_id from connector_accounts where org_id = ? and toolkit_slug = ?", + ).get(orgId, slug) as { id: string; status: string; pending_account_id: string | null } | undefined; const account = await client.createConnectedAccount({ authConfigId: entry.authConfigId, userId: getComposioUserId(this.db, orgId), }); if (!account.id) throw new Error("Composio did not return a connected account id."); const now = new Date().toISOString(); + // The local row moves to the new account before any old one is deleted; refresh ignores accounts no row tracks. + if (previous?.status === "ACTIVE") { + this.db.prepare("update connector_accounts set pending_account_id = ?, updated_at = ? where org_id = ? and toolkit_slug = ?") + .run(account.id, now, orgId, slug); + console.info(`[connectors] ${slug} reconnect ${account.id} started; ${previous.id} stays in use until it is authorized`); + if (previous.pending_account_id) await client.deleteConnectedAccount(previous.pending_account_id); + return { accountId: account.id, redirectUrl: account.redirectUrl }; + } this.db.prepare(` insert into connector_accounts(id, org_id, toolkit_slug, toolkit_name, auth_config_id, status, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?) @@ -78,17 +78,15 @@ export class ComposioConnectionService { toolkit_name = excluded.toolkit_name, auth_config_id = excluded.auth_config_id, status = excluded.status, + pending_account_id = null, updated_at = excluded.updated_at `).run(account.id, orgId, slug, entry.name, entry.authConfigId, "INITIALIZING", now, now); + if (previous?.pending_account_id) await client.deleteConnectedAccount(previous.pending_account_id); + if (previous) await client.deleteConnectedAccount(previous.id); return { accountId: account.id, redirectUrl: account.redirectUrl }; } - /** - * Pulls account statuses from Composio and updates the local mirror, in - * both directions: statuses adopt Composio's truth, and local rows whose - * account no longer exists in Composio are removed along with their - * grants — deleting an account in the dashboard cleans up here too. - */ + /** Mirrors Composio: tracked accounts take its status, gone ones are removed with their grants, untracked toolkits are adopted. */ refreshAccounts(orgId: string): Promise { return this.serialize(orgId, () => this.refreshAccountsNow(orgId)); } @@ -98,13 +96,26 @@ export class ComposioConnectionService { const accounts = await client.listConnectedAccounts(getComposioUserId(this.db, orgId)); const now = new Date().toISOString(); const liveIds = new Set(accounts.map((account) => account.id)); + const reconnects = this.db.prepare("select id, toolkit_slug, pending_account_id from connector_accounts where org_id = ? and pending_account_id is not null") + .all(orgId) as Array<{ id: string; toolkit_slug: string; pending_account_id: string }>; + for (const row of reconnects) await this.settleReconnect(client, row, accounts.find((account) => account.id === row.pending_account_id), liveIds.has(row.id)); + const localRows = this.db.prepare("select id, toolkit_slug, pending_account_id from connector_accounts where org_id = ?") + .all(orgId) as Array<{ id: string; toolkit_slug: string; pending_account_id: string | null }>; + for (const row of localRows) { + if (liveIds.has(row.id)) continue; + this.removeLocalAccount(orgId, row.toolkit_slug); + } + const tracked = new Map(localRows.filter((row) => liveIds.has(row.id)).map((row) => [row.toolkit_slug, row])); for (const account of accounts) { if (!account.toolkitSlug) continue; + const row = tracked.get(account.toolkitSlug); + // A toolkit's row follows only its own account; others (a waiting reconnect, an orphan) never replace it. + if (row && row.id !== account.id) continue; + tracked.set(account.toolkitSlug, { id: account.id, toolkit_slug: account.toolkitSlug, pending_account_id: null }); this.db.prepare(` insert into connector_accounts(id, org_id, toolkit_slug, toolkit_name, auth_config_id, status, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?) on conflict(org_id, toolkit_slug) do update set - id = excluded.id, auth_config_id = excluded.auth_config_id, status = excluded.status, updated_at = excluded.updated_at @@ -119,12 +130,27 @@ export class ComposioConnectionService { now, ); } - const localRows = this.db.prepare("select id, toolkit_slug from connector_accounts where org_id = ?") - .all(orgId) as Array<{ id: string; toolkit_slug: string }>; - for (const row of localRows) { - if (liveIds.has(row.id)) continue; - this.removeLocalAccount(orgId, row.toolkit_slug); + } + + /** A waiting reconnect replaces the old account once active (or once the old one is gone); one that failed or vanished is dropped and the old account stays. */ + private async settleReconnect( + client: ComposioClient, + row: { id: string; toolkit_slug: string; pending_account_id: string }, + pending: ComposioConnectedAccount | undefined, + oldIsLive: boolean, + ): Promise { + const now = new Date().toISOString(); + if (pending && (pending.status === "ACTIVE" || !oldIsLive)) { + if (oldIsLive) await client.deleteConnectedAccount(row.id); + this.db.prepare("update connector_accounts set id = ?, auth_config_id = ?, status = ?, pending_account_id = null, updated_at = ? where id = ?") + .run(pending.id, pending.authConfigId, pending.status, now, row.id); + console.info(`[connectors] ${row.toolkit_slug} reconnect ${pending.id} replaced ${row.id}`); + return; } + if (pending && WAITING_STATUSES.has(pending.status)) return; + if (pending) await client.deleteConnectedAccount(pending.id); + this.db.prepare("update connector_accounts set pending_account_id = null, updated_at = ? where id = ?").run(now, row.id); + console.warn(`[connectors] ${row.toolkit_slug} reconnect ${row.pending_account_id} ended ${pending?.status ?? "missing"}; ${row.id} stays in use`); } /** @@ -139,10 +165,16 @@ export class ComposioConnectionService { private async disconnectToolkitNow(orgId: string, toolkitSlug: string): Promise { const slug = normalizeSlug(toolkitSlug); const row = this.db.prepare( - "select id from connector_accounts where org_id = ? and toolkit_slug = ?", - ).get(orgId, slug) as { id: string } | undefined; + "select id, pending_account_id from connector_accounts where org_id = ? and toolkit_slug = ?", + ).get(orgId, slug) as { id: string; pending_account_id: string | null } | undefined; if (!row) return; - await this.requireClient().deleteConnectedAccount(row.id); + const client = this.requireClient(); + // Every account Composio holds for the toolkit goes, so an orphan from a failed delete cannot be adopted back. + const ids = new Set([row.id, ...(row.pending_account_id ? [row.pending_account_id] : [])]); + for (const account of await client.listConnectedAccounts(getComposioUserId(this.db, orgId))) { + if (account.toolkitSlug === slug) ids.add(account.id); + } + for (const id of ids) await client.deleteConnectedAccount(id); this.removeLocalAccount(orgId, slug); } diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index 6051f6a4..d99b488a 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -382,6 +382,7 @@ export const zh: Record = { "Recent messages": "最近消息", "Recently decided": "最近已决", "Reconnect": "重新连接", + "Reconnect waiting for authorization": "重新连接等待授权", "Refresh models": "刷新模型", "Region": "区域", "Rejected": "已驳回", diff --git a/src/renderer/src/pages/connectors/ConnectedList.tsx b/src/renderer/src/pages/connectors/ConnectedList.tsx index 1f7c546d..67a51d13 100644 --- a/src/renderer/src/pages/connectors/ConnectedList.tsx +++ b/src/renderer/src/pages/connectors/ConnectedList.tsx @@ -61,6 +61,7 @@ export function ConnectedList({ /> {account.toolkitName} {account.status.charAt(0) + account.status.slice(1).toLowerCase()} + {account.pendingAccountId && {t("Reconnect waiting for authorization")}}