diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index e79ebaa3279..a7c4c28f474 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1893,3 +1893,11 @@ unavailable. Preserve current ownership and newer-work fences. See endpoints delegate to sources while retaining response shapes. - GitHub.com, manual refresh only. No upstream editing, polling, webhook sync, commits, or pull-request creation in this milestone. + +### Experimental AI connection routing + +Opt-in plugin routers may represent a pool as an AI runtime binding. Core keeps +company and credential authorization, atomically records task/agent affinity and +a pool cursor, and persists concrete native recovery evidence. The full contract +is in [AI-CONNECTION-ROUTERS.md](connections/AI-CONNECTION-ROUTERS.md). Disabled +routing cannot allocate new tasks; already admitted native runs remain recoverable. diff --git a/doc/SPEC.md b/doc/SPEC.md index ab162e3df95..737abe61f7f 100644 --- a/doc/SPEC.md +++ b/doc/SPEC.md @@ -686,3 +686,10 @@ requirements, and advisory warnings for missing or external references. New upstream skills require reviewed selection; removed or deselected skills remain installed. Editing starts with an independent copy. Write-back and PR publication are a later milestone; exact path and commit provenance provide their base. + +### Experimental connection routing + +A virtual AI connection can rotate new task/agent allocations through an +authorized pool while preserving session affinity. Admission, credentials and +durable recovery remain host responsibilities; policy can be supplied by an +opt-in plugin. See [the experimental contract](connections/AI-CONNECTION-ROUTERS.md). diff --git a/doc/connections/AI-CONNECTION-ROUTERS.md b/doc/connections/AI-CONNECTION-ROUTERS.md new file mode 100644 index 00000000000..4bf4f4d78b7 --- /dev/null +++ b/doc/connections/AI-CONNECTION-ROUTERS.md @@ -0,0 +1,47 @@ +# Experimental AI connection routers + +AI routers are virtual, company-scoped connections owned by a capability-gated +plugin (`ai.connections.route`). The instance flag `enableAiConnectionRouters` +defaults to false. Pools also default to disabled. The host must implement this +contract; the plugin's minimum version alone does not establish compatibility. + +The host authorizes members using the ordinary company, user, sharing, install, +health and harness checks. It sends authorized metadata and sanitized usage +observations to `onRouteAiConnection`. The plugin proposes an opaque member ID; +it cannot receive credentials or expand authorization. Pure round robin does +not probe usage. Usage-aware selection has a shared 15-second probe budget, +60-second freshness cache and ordinary grant/secret freshness invalidation. + +One cursor spans all agents in a pool. An allocation transaction locks the +cursor, checks config and cursor revisions, rechecks authorization, and writes +the task pin and cursor advance together. Pins use company, pool, agent and the +existing task key (including `__heartbeat__`). Wakes without a task key retain +the original run ID as an affinity key in retry context. Conflicts retry at most +20 times; contention beyond that returns an actionable conflict. + +Pins snapshot the concrete binding and member profile. Removing or editing a +member affects future allocations. Composer changes can change a supported +model or effort, with a note on fallback, but never the account or harness. +Session reset and compaction retain the pin. Revocation requires operator repair. +A pre-existing managed session can adopt its saved account when it is an +eligible member; otherwise the operator must explicitly reset the session. + +Credential `ai_session_epoch` changes on reconnect or manual rotation. Only +verified runtime refresh write-back preserves it. Session fingerprints use the +epoch while authentication failure attribution still uses the token generation. +Native recovery retains concrete routing evidence and can finish after the +router flag or plugin is disabled or uninstalled; it revalidates underlying +account access and never makes a new allocation. + +The private Cloud plugin owns round-robin and quota policy. Its configuration +page uses authenticated, company-scoped Core pool APIs. New allocations over +the chosen threshold wait; known exhaustion defers pinned turns. The deferred +run schedules `ai_connection_pool_wait` without spending the failure retry +budget, and the UI labels that retry as **Pool exhausted**. + +Configuration and committed selection are recorded in activity records. Runs +record the selected member/profile and override notes in the local run log and +recovery context. These records stay in the instance database. + +Tests: `cd server && pnpm exec vitest run src/__tests__/ai-connection-router.test.ts` +plus the existing AI connection, retry accounting, run-dispatch and UI suites. diff --git a/packages/db/src/migrations/0295_cuddly_dazzler.sql b/packages/db/src/migrations/0295_cuddly_dazzler.sql new file mode 100644 index 00000000000..5f15b95940c --- /dev/null +++ b/packages/db/src/migrations/0295_cuddly_dazzler.sql @@ -0,0 +1,64 @@ +CREATE TABLE IF NOT EXISTS "ai_connection_pools" ( + "id" uuid PRIMARY KEY NOT NULL, + "company_id" uuid NOT NULL, + "plugin_key" text NOT NULL, + "config" jsonb NOT NULL, + "revision" integer DEFAULT 1 NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT "ai_connection_pools_company_id_uq" UNIQUE("company_id","id") +); +--> statement-breakpoint +CREATE TABLE IF NOT EXISTS "ai_connection_router_cursors" ( + "pool_id" uuid PRIMARY KEY NOT NULL, + "company_id" uuid NOT NULL, + "last_member_id" text, + "version" integer DEFAULT 0 NOT NULL +); +--> statement-breakpoint +CREATE TABLE IF NOT EXISTS "ai_connection_task_pins" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "company_id" uuid NOT NULL, + "pool_id" uuid NOT NULL, + "agent_id" uuid NOT NULL, + "task_key" text NOT NULL, + "selection" jsonb NOT NULL, + "member" jsonb NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +ALTER TABLE "company_secrets" ADD COLUMN IF NOT EXISTS "ai_session_epoch" integer DEFAULT 0 NOT NULL;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_pools" ADD CONSTRAINT "ai_connection_pools_id_tool_connections_id_fk" FOREIGN KEY ("id") REFERENCES "public"."tool_connections"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_pools" ADD CONSTRAINT "ai_connection_pools_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_pools" ADD CONSTRAINT "ai_connection_pools_company_connection_fk" FOREIGN KEY ("company_id","id") REFERENCES "public"."tool_connections"("company_id","id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_router_cursors" ADD CONSTRAINT "ai_connection_router_cursors_pool_id_ai_connection_pools_id_fk" FOREIGN KEY ("pool_id") REFERENCES "public"."ai_connection_pools"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_router_cursors" ADD CONSTRAINT "ai_connection_router_cursors_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_router_cursors" ADD CONSTRAINT "ai_connection_router_cursors_company_pool_fk" FOREIGN KEY ("company_id","pool_id") REFERENCES "public"."ai_connection_pools"("company_id","id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_task_pins" ADD CONSTRAINT "ai_connection_task_pins_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_task_pins" ADD CONSTRAINT "ai_connection_task_pins_pool_id_ai_connection_pools_id_fk" FOREIGN KEY ("pool_id") REFERENCES "public"."ai_connection_pools"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_task_pins" ADD CONSTRAINT "ai_connection_task_pins_agent_id_agents_id_fk" FOREIGN KEY ("agent_id") REFERENCES "public"."agents"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_task_pins" ADD CONSTRAINT "ai_connection_task_pins_company_pool_fk" FOREIGN KEY ("company_id","pool_id") REFERENCES "public"."ai_connection_pools"("company_id","id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +DO $$ BEGIN +ALTER TABLE "ai_connection_task_pins" ADD CONSTRAINT "ai_connection_task_pins_company_agent_fk" FOREIGN KEY ("company_id","agent_id") REFERENCES "public"."agents"("company_id","id") ON DELETE cascade ON UPDATE no action; +EXCEPTION WHEN duplicate_object THEN NULL; END $$;--> statement-breakpoint +CREATE UNIQUE INDEX IF NOT EXISTS "ai_connection_task_pins_affinity_uq" ON "ai_connection_task_pins" USING btree ("company_id","pool_id","agent_id","task_key"); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0290_snapshot.json b/packages/db/src/migrations/meta/0295_snapshot.json similarity index 98% rename from packages/db/src/migrations/meta/0290_snapshot.json rename to packages/db/src/migrations/meta/0295_snapshot.json index 63ef4c6fddf..180a3497174 100644 --- a/packages/db/src/migrations/meta/0290_snapshot.json +++ b/packages/db/src/migrations/meta/0295_snapshot.json @@ -1,6 +1,6 @@ { - "id": "26315a87-4675-4dea-9634-efe5e878cc7d", - "prevId": "3960ae6b-5bf1-4fd7-91db-6a18fc7eaa19", + "id": "18dbd936-096c-4e1d-ab23-10e223ba3a42", + "prevId": "66a0b3c9-91bb-46b8-a868-a6facedbc14b", "version": "7", "dialect": "postgresql", "tables": { @@ -2903,6 +2903,362 @@ }, "isRLSEnabled": false }, + "public.ai_connection_pools": { + "name": "ai_connection_pools", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "plugin_key": { + "name": "plugin_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "config": { + "name": "config", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "ai_connection_pools_id_tool_connections_id_fk": { + "name": "ai_connection_pools_id_tool_connections_id_fk", + "tableFrom": "ai_connection_pools", + "tableTo": "tool_connections", + "columnsFrom": [ + "id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_pools_company_id_companies_id_fk": { + "name": "ai_connection_pools_company_id_companies_id_fk", + "tableFrom": "ai_connection_pools", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_pools_company_connection_fk": { + "name": "ai_connection_pools_company_connection_fk", + "tableFrom": "ai_connection_pools", + "tableTo": "tool_connections", + "columnsFrom": [ + "company_id", + "id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "ai_connection_pools_company_id_uq": { + "name": "ai_connection_pools_company_id_uq", + "nullsNotDistinct": false, + "columns": [ + "company_id", + "id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.ai_connection_router_cursors": { + "name": "ai_connection_router_cursors", + "schema": "", + "columns": { + "pool_id": { + "name": "pool_id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "last_member_id": { + "name": "last_member_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + } + }, + "indexes": {}, + "foreignKeys": { + "ai_connection_router_cursors_pool_id_ai_connection_pools_id_fk": { + "name": "ai_connection_router_cursors_pool_id_ai_connection_pools_id_fk", + "tableFrom": "ai_connection_router_cursors", + "tableTo": "ai_connection_pools", + "columnsFrom": [ + "pool_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_router_cursors_company_id_companies_id_fk": { + "name": "ai_connection_router_cursors_company_id_companies_id_fk", + "tableFrom": "ai_connection_router_cursors", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_router_cursors_company_pool_fk": { + "name": "ai_connection_router_cursors_company_pool_fk", + "tableFrom": "ai_connection_router_cursors", + "tableTo": "ai_connection_pools", + "columnsFrom": [ + "company_id", + "pool_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.ai_connection_task_pins": { + "name": "ai_connection_task_pins", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "pool_id": { + "name": "pool_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "agent_id": { + "name": "agent_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "task_key": { + "name": "task_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "selection": { + "name": "selection", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "member": { + "name": "member", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "ai_connection_task_pins_affinity_uq": { + "name": "ai_connection_task_pins_affinity_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "pool_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "agent_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "task_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "ai_connection_task_pins_company_id_companies_id_fk": { + "name": "ai_connection_task_pins_company_id_companies_id_fk", + "tableFrom": "ai_connection_task_pins", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_task_pins_pool_id_ai_connection_pools_id_fk": { + "name": "ai_connection_task_pins_pool_id_ai_connection_pools_id_fk", + "tableFrom": "ai_connection_task_pins", + "tableTo": "ai_connection_pools", + "columnsFrom": [ + "pool_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_task_pins_agent_id_agents_id_fk": { + "name": "ai_connection_task_pins_agent_id_agents_id_fk", + "tableFrom": "ai_connection_task_pins", + "tableTo": "agents", + "columnsFrom": [ + "agent_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_task_pins_company_pool_fk": { + "name": "ai_connection_task_pins_company_pool_fk", + "tableFrom": "ai_connection_task_pins", + "tableTo": "ai_connection_pools", + "columnsFrom": [ + "company_id", + "pool_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "ai_connection_task_pins_company_agent_fk": { + "name": "ai_connection_task_pins_company_agent_fk", + "tableFrom": "ai_connection_task_pins", + "tableTo": "agents", + "columnsFrom": [ + "company_id", + "agent_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.ai_provider_defaults": { "name": "ai_provider_defaults", "schema": "", @@ -12347,6 +12703,13 @@ "primaryKey": false, "notNull": false }, + "ai_session_epoch": { + "name": "ai_session_epoch", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, "latest_version": { "name": "latest_version", "type": "integer", @@ -12737,6 +13100,373 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.company_skill_source_entries": { + "name": "company_skill_source_entries", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "source_id": { + "name": "source_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "path": { + "name": "path", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "inspection": { + "name": "inspection", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "skill_id": { + "name": "skill_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "selection": { + "name": "selection", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'new'" + }, + "present": { + "name": "present", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "company_skill_source_entries_source_path_idx": { + "name": "company_skill_source_entries_source_path_idx", + "columns": [ + { + "expression": "source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "path", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "company_skill_source_entries_company_skill_idx": { + "name": "company_skill_source_entries_company_skill_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "skill_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "company_skill_source_entries_company_id_companies_id_fk": { + "name": "company_skill_source_entries_company_id_companies_id_fk", + "tableFrom": "company_skill_source_entries", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "company_skill_source_entries_source_id_company_skill_sources_id_fk": { + "name": "company_skill_source_entries_source_id_company_skill_sources_id_fk", + "tableFrom": "company_skill_source_entries", + "tableTo": "company_skill_sources", + "columnsFrom": [ + "source_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "company_skill_source_entries_skill_id_company_skills_id_fk": { + "name": "company_skill_source_entries_skill_id_company_skills_id_fk", + "tableFrom": "company_skill_source_entries", + "tableTo": "company_skills", + "columnsFrom": [ + "skill_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.company_skill_sources": { + "name": "company_skill_sources", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "repository_id": { + "name": "repository_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "repository_url": { + "name": "repository_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "full_name": { + "name": "full_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "tracking_ref": { + "name": "tracking_ref", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "connection_id": { + "name": "connection_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "excluded_folders": { + "name": "excluded_folders", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "enabled": { + "name": "enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "last_attempt_at": { + "name": "last_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "last_success_at": { + "name": "last_success_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "last_scan_commit": { + "name": "last_scan_commit", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "last_error": { + "name": "last_error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "lease_token": { + "name": "lease_token", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "lease_expires_at": { + "name": "lease_expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "company_skill_sources_repository_ref_idx": { + "name": "company_skill_sources_repository_ref_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "repository_url", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "tracking_ref", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "company_skill_sources_identity_ref_idx": { + "name": "company_skill_sources_identity_ref_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "repository_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "tracking_ref", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "company_skill_sources_company_id_companies_id_fk": { + "name": "company_skill_sources_company_id_companies_id_fk", + "tableFrom": "company_skill_sources", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "company_skill_sources_connection_id_tool_connections_id_fk": { + "name": "company_skill_sources_connection_id_tool_connections_id_fk", + "tableFrom": "company_skill_sources", + "tableTo": "tool_connections", + "columnsFrom": [ + "connection_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.company_skill_comments": { "name": "company_skill_comments", "schema": "", @@ -31098,6 +31828,13 @@ "primaryKey": false, "notNull": true }, + "title_needs_generation": { + "name": "title_needs_generation", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, "description": { "name": "description", "type": "text", @@ -45835,6 +46572,27 @@ "method": "btree", "with": {} }, + "tool_mcp_gateway_tokens_expiry_idx": { + "name": "tool_mcp_gateway_tokens_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_mcp_gateway_tokens_gateway_idx": { "name": "tool_mcp_gateway_tokens_gateway_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 15430d35565..9f1f1e22f6d 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2052,6 +2052,13 @@ "when": 1790891286527, "tag": "0294_chilly_marvel_apes", "breakpoints": true + }, + { + "idx": 295, + "version": "7", + "when": 1790967927385, + "tag": "0295_cuddly_dazzler", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/ai_connection_routing.ts b/packages/db/src/schema/ai_connection_routing.ts new file mode 100644 index 00000000000..524e432aa11 --- /dev/null +++ b/packages/db/src/schema/ai_connection_routing.ts @@ -0,0 +1,38 @@ +import { foreignKey, integer, jsonb, pgTable, text, timestamp, unique, uniqueIndex, uuid } from "drizzle-orm/pg-core"; +import type { AiConnectionPoolConfig, AiConnectionRouterSelection } from "@paperclipai/shared"; +import { companies } from "./companies.js"; +import { agents } from "./agents.js"; +import { toolConnections } from "./tool_access.js"; + +export const aiConnectionPools = pgTable("ai_connection_pools", { + id: uuid("id").primaryKey().references(() => toolConnections.id, { onDelete: "cascade" }), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + pluginKey: text("plugin_key").notNull(), + config: jsonb("config").$type().notNull(), + revision: integer("revision").notNull().default(1), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), +}, (table) => [ + unique("ai_connection_pools_company_id_uq").on(table.companyId, table.id), + foreignKey({ columns: [table.companyId, table.id], foreignColumns: [toolConnections.companyId, toolConnections.id], name: "ai_connection_pools_company_connection_fk" }).onDelete("cascade"), +]); +export const aiConnectionRouterCursors = pgTable("ai_connection_router_cursors", { + poolId: uuid("pool_id").primaryKey().references(() => aiConnectionPools.id, { onDelete: "cascade" }), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + lastMemberId: text("last_member_id"), + version: integer("version").notNull().default(0), +}, (table) => [foreignKey({ columns: [table.companyId, table.poolId], foreignColumns: [aiConnectionPools.companyId, aiConnectionPools.id], name: "ai_connection_router_cursors_company_pool_fk" }).onDelete("cascade")]); +export const aiConnectionTaskPins = pgTable("ai_connection_task_pins", { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + poolId: uuid("pool_id").notNull().references(() => aiConnectionPools.id, { onDelete: "cascade" }), + agentId: uuid("agent_id").notNull().references(() => agents.id, { onDelete: "cascade" }), + taskKey: text("task_key").notNull(), + selection: jsonb("selection").$type().notNull(), + member: jsonb("member").$type().notNull(), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), +}, (table) => [ + uniqueIndex("ai_connection_task_pins_affinity_uq").on(table.companyId, table.poolId, table.agentId, table.taskKey), + foreignKey({ columns: [table.companyId, table.poolId], foreignColumns: [aiConnectionPools.companyId, aiConnectionPools.id], name: "ai_connection_task_pins_company_pool_fk" }).onDelete("cascade"), + foreignKey({ columns: [table.companyId, table.agentId], foreignColumns: [agents.companyId, agents.id], name: "ai_connection_task_pins_company_agent_fk" }).onDelete("cascade"), +]); diff --git a/packages/db/src/schema/company_secrets.ts b/packages/db/src/schema/company_secrets.ts index 9ca340d2240..6d3a9ddc29d 100644 --- a/packages/db/src/schema/company_secrets.ts +++ b/packages/db/src/schema/company_secrets.ts @@ -21,6 +21,7 @@ export const companySecrets = pgTable( externalRef: text("external_ref"), providerConfigId: uuid("provider_config_id").references(() => companySecretProviderConfigs.id, { onDelete: "set null" }), providerMetadata: jsonb("provider_metadata").$type>(), + aiSessionEpoch: integer("ai_session_epoch").notNull().default(0), latestVersion: integer("latest_version").notNull().default(1), description: text("description"), lastResolvedAt: timestamp("last_resolved_at", { withTimezone: true }), diff --git a/packages/db/src/schema/index.ts b/packages/db/src/schema/index.ts index 19df9235905..977c3b1b8f7 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -218,3 +218,5 @@ export { browserUseSettings, browserUseSessions, browserUseRuns, browserUseBrows export * from "./company_skill_sources.js"; + +export * from "./ai_connection_routing.js"; diff --git a/packages/plugins/sdk/src/define-plugin.ts b/packages/plugins/sdk/src/define-plugin.ts index ef5c5dc6f9b..947a67cbb55 100644 --- a/packages/plugins/sdk/src/define-plugin.ts +++ b/packages/plugins/sdk/src/define-plugin.ts @@ -1,3 +1,4 @@ +import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paperclipai/shared"; /** * `definePlugin` — the top-level helper for authoring a Paperclip plugin. * @@ -345,6 +346,9 @@ export interface PluginDefinition { * * Requires `external.objects.read`. */ + /** Propose a member from host-authorized candidates. Requires ai.connections.route. */ + onRouteAiConnection?(params: AiConnectionRouterRequest): Promise; + onResolveExternalObject?( params: ResolveExternalObjectParams, ): Promise; diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 788247f39ad..57f86281178 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -451,3 +451,5 @@ export { export { PluginEnvironmentCreationCleanupError, environmentCreationCleanupErrorData, readEnvironmentCreationCleanupError } from "./environment-creation-cleanup.js"; export type { PluginEnvironmentCreationCleanup } from "./environment-creation-cleanup.js"; + +export type { AiConnectionPool, AiConnectionPoolConfig, AiConnectionPoolMember, AiConnectionRouterRequest, AiConnectionRouterResult, AiConnectionRouterSelection } from "@paperclipai/shared"; diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 5476c10becb..9c08a45c42b 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -1,3 +1,4 @@ +import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paperclipai/shared"; /** * JSON-RPC 2.0 message types and protocol helpers for the host ↔ worker IPC * channel. @@ -1350,6 +1351,7 @@ export interface HostToWorkerMethods { params: DetectExternalObjectsParams, result: DetectExternalObjectsResult, ]; + routeAiConnection: [params: AiConnectionRouterRequest, result: AiConnectionRouterResult]; resolveExternalObject: [ params: ResolveExternalObjectParams, result: PluginExternalObjectResolveResult, @@ -1478,6 +1480,7 @@ export const HOST_TO_WORKER_OPTIONAL_METHODS: readonly HostToWorkerMethodName[] "performAction", "executeTool", "detectExternalObjects", + "routeAiConnection", "resolveExternalObject", "refreshExternalObjects", "environmentValidateConfig", diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 78d8308a297..0f0d4eb4cfd 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -1,3 +1,4 @@ +import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paperclipai/shared"; import { environmentCreationCleanupErrorData } from "./environment-creation-cleanup.js"; /** * Worker-side RPC host — runs inside the child process spawned by the host. @@ -1620,6 +1621,9 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost return handleExecuteTool(params as ExecuteToolParams); case "detectExternalObjects": return handleDetectExternalObjects(params as DetectExternalObjectsParams); + case "routeAiConnection": + if (!plugin.definition.onRouteAiConnection) throw methodNotImplemented("routeAiConnection"); + return plugin.definition.onRouteAiConnection(params as AiConnectionRouterRequest); case "resolveExternalObject": return handleResolveExternalObject(params as ResolveExternalObjectParams); case "refreshExternalObjects": @@ -1736,6 +1740,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (plugin.definition.onShutdown) supportedMethods.push("shutdown"); if (plugin.definition.onApiRequest) supportedMethods.push("handleApiRequest"); if (plugin.definition.onDetectExternalObjects) supportedMethods.push("detectExternalObjects"); + if (plugin.definition.onRouteAiConnection) supportedMethods.push("routeAiConnection"); if (plugin.definition.onResolveExternalObject) supportedMethods.push("resolveExternalObject"); if (plugin.definition.onRefreshExternalObjects) supportedMethods.push("refreshExternalObjects"); if (plugin.definition.onEnvironmentValidateConfig) supportedMethods.push("environmentValidateConfig"); diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index b504a703628..cce6a3ae1a7 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -1115,3 +1115,25 @@ describe("worker duplex channel dispatch", () => { } }); }); + + +describe("AI connection router RPC", () => { + it("advertises and calls only an implemented routing hook", async () => { + const input = new PassThrough(), output = new PassThrough(); + const reader = createInterface({ input: output }); + const request = { companyId: "company", taskKey: "task", candidates: [], memberOrder: [] }; + let received: unknown; + const plugin = definePlugin({ async setup() {}, onRouteAiConnection(params) { received = params; return { kind: "selected", memberId: "authorized-member" }; } }); + const worker = startWorkerRpcHost({ plugin, stdin: input, stdout: output }); + let sequence = 0; + const pending = new Map void>(); + reader.on("line", line => { const response = parseMessage(line); if (isJsonRpcResponse(response)) { pending.get(String(response.id))?.(response); pending.delete(String(response.id)); } }); + const call = (method: string, params: unknown) => new Promise(resolve => { const id = String(++sequence); pending.set(id, value => resolve(value as JsonRpcResponse)); input.write(serializeMessage(createRequest(method, params, id))); }); + try { + const initialized = await call("initialize", { manifest: { id: "fixture.router", apiVersion: 1, version: "1.0.0", displayName: "Router", description: "Fixture", author: "Tests", categories: ["connector"], capabilities: ["ai.connections.route"], entrypoints: {} }, config: {}, databaseNamespace: null }); + expect((initialized as { result: { supportedMethods: string[] } }).result.supportedMethods).toContain("routeAiConnection"); + expect((await call("routeAiConnection", request) as { result: unknown }).result).toEqual({ kind: "selected", memberId: "authorized-member" }); + expect(received).toEqual(request); + } finally { worker.stop(); reader.close(); input.destroy(); output.destroy(); } + }); +}); diff --git a/packages/shared/src/ai-connection-router.ts b/packages/shared/src/ai-connection-router.ts new file mode 100644 index 00000000000..34b12476860 --- /dev/null +++ b/packages/shared/src/ai-connection-router.ts @@ -0,0 +1,77 @@ +import { z } from "zod"; +import { aiConnectionBindingSchema, type AiConnectionBinding } from "./ai-connections.js"; +import type { AiConnectionUsage } from "./ai-connection-usage.js"; + +export const aiConnectionRouterBindingSchema = z.object({ + mode: z.literal("router"), + connectionId: z.string().uuid(), +}).strict(); +export const aiRuntimeConnectionBindingSchema = z.union([aiConnectionBindingSchema, aiConnectionRouterBindingSchema]); +export type AiRuntimeConnectionBinding = z.infer; + +export const aiConnectionPoolMemberSchema = z.object({ + id: z.string().uuid(), + binding: aiConnectionBindingSchema.refine((v) => v.mode !== "responsible_user", "Select an explicit account grant"), + profile: z.object({ + provider: z.enum(["codex", "acpx", "opencode"]), + acpxAgent: z.enum(["claude", "grok"]).optional(), + model: z.string().trim().min(1).max(256), + effort: z.string().trim().min(1).max(32).optional(), + }).strict(), +}).strict(); +export const aiConnectionPoolConfigSchema = z.object({ + name: z.string().trim().min(1).max(160), + enabled: z.boolean().default(false), + mode: z.enum(["round_robin", "usage_aware"]).default("round_robin"), + thresholdPercent: z.number().int().min(1).max(100).default(90), + members: z.array(aiConnectionPoolMemberSchema).min(1).max(100), +}).strict().superRefine((v, ctx) => { + const ids = new Set(); + const grants = new Set(); + const connections = new Set(); + for (const [index, member] of v.members.entries()) { + const grant = `${member.binding.connectionId}:${member.binding.grantId}`; + if (ids.has(member.id) || grants.has(grant) || connections.has(member.binding.connectionId)) ctx.addIssue({ code: "custom", path: ["members", index], message: "Duplicate pool member" }); + ids.add(member.id); grants.add(grant); connections.add(member.binding.connectionId); + if (member.profile.provider === "acpx" && !member.profile.acpxAgent) ctx.addIssue({ code: "custom", path: ["members", index, "profile"], message: "Choose an ACP harness" }); + } +}); +export type AiConnectionPoolMember = z.infer; +export type AiConnectionPoolConfig = z.infer; +export interface AiConnectionPool extends AiConnectionPoolConfig { + id: string; + companyId: string; + pluginKey: string; + revision: number; +} +/** Host-authorized metadata only. No credentials or raw provider responses. */ +export interface AiConnectionRouterRequest { + companyId: string; + pool: AiConnectionPool; + agentId: string; + taskKey: string; + lastMemberId: string | null; + cursorVersion: number; + /** Opaque IDs preserve order without exposing unauthorized account metadata. */ + memberOrder: string[]; + pinnedMemberId?: string; + candidates: Array<{ member: AiConnectionPoolMember; runtimeConfig: Record; notes: string[]; usage?: AiConnectionUsage }>; + now: string; +} +export type AiConnectionRouterResult = + | { kind: "selected"; memberId: string; skipped?: Record } + | { kind: "exhausted"; retryAt: string; skipped: Record } + | { kind: "unavailable"; skipped: Record }; +export interface AiConnectionPoolSaveInput { + companyId: string; + id?: string; + expectedRevision?: number; + config: AiConnectionPoolConfig; +} +export interface AiConnectionRouterSelection { + poolId: string; + memberId: string; + binding: AiConnectionBinding; + runtimeConfig: Record; + notes: string[]; +} diff --git a/packages/shared/src/ai-connections.ts b/packages/shared/src/ai-connections.ts index 59176427752..898a5f246e4 100644 --- a/packages/shared/src/ai-connections.ts +++ b/packages/shared/src/ai-connections.ts @@ -176,6 +176,7 @@ export interface AiManagedConnectionSummary { } export interface AiConnectionList { currentUserId: string; + pools?: import("./ai-connection-router.js").AiConnectionPool[]; canManageConnections: boolean; connections: AiManagedConnectionSummary[]; } diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index 61a2b2b5bad..3e83f263130 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -1320,6 +1320,7 @@ export type PluginCategory = (typeof PLUGIN_CATEGORIES)[number]; * @see PLUGIN_SPEC.md §15 — Capability Model */ export const PLUGIN_CAPABILITIES = [ + "ai.connections.route", // Data Read "companies.read", "projects.read", diff --git a/packages/shared/src/feature-catalog.ts b/packages/shared/src/feature-catalog.ts index f2c404a3bd8..593f3a18fd7 100644 --- a/packages/shared/src/feature-catalog.ts +++ b/packages/shared/src/feature-catalog.ts @@ -50,6 +50,11 @@ export const INSTANCE_FEATURE_CATALOG: Record>; +let db: ReturnType; +let home: string; +const companyId = randomUUID(), otherCompanyId = randomUUID(), agentId = randomUUID(), secondAgentId = randomUUID(), pluginId = randomUUID(); +const pluginKey = "fixture.ai-router"; +const manifest: PaperclipPluginManifestV1 = { id: pluginKey, apiVersion: 1, version: "0.1.0", displayName: "Fixture router", description: "Fixture", author: "Tests", categories: ["connector"], capabilities: ["ai.connections.route"], entrypoints: { worker: "worker.js" } }; +let members: AiConnectionPoolMember[]; +const proposal = vi.fn(async (_id: string, _method: string, request: AiConnectionRouterRequest): Promise => { + const next = (request.memberOrder.indexOf(request.lastMemberId ?? "") + 1) % request.memberOrder.length; + const order = request.pinnedMemberId ? [request.pinnedMemberId] : [...request.memberOrder.slice(next), ...request.memberOrder.slice(0, next)]; + const selected = order.find(id => request.candidates.some(candidate => candidate.member.id === id)); + return selected ? { kind: "selected", memberId: selected } : { kind: "unavailable", skipped: {} }; +}); +const worker = { isRunning: () => true, call: proposal } as unknown as PluginWorkerManager; +const service = () => aiConnectionRouterService(db, worker); +const resolve = (poolId: string, taskKey: string, extra = {}) => service().resolve({ companyId, poolId, agentId, userId: "alice", adapterType: "paperclip_runner", taskKey, ...extra }); +const makePool = async (config = {}) => service().save(pluginKey, { companyId, config: { name: randomUUID(), enabled: true, mode: "round_robin", thresholdPercent: 90, members, ...config } }, "alice"); +beforeAll(async () => { + home = await mkdtemp(path.join(os.tmpdir(), "paperclip-router-tests-")); + vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_INSTANCE_ID", "router-fixture"); + database = await startEmbeddedPostgresTestDatabase("paperclip-router-db-"); db = createDb(database.connectionString); + await db.insert(companies).values([{ id: companyId, name: "Router fixture", issuePrefix: "RTF" }, { id: otherCompanyId, name: "Other", issuePrefix: "RTO" }]); + await db.insert(agents).values([agentId, secondAgentId].map(id => ({ id, companyId, name: "Router agent", adapterType: "paperclip_runner" }))); + await db.insert(companyMemberships).values(["alice", "bob"].map(principalId => ({ companyId, principalId, principalType: "user", status: "active", membershipRole: "member" }))); + await db.insert(plugins).values({ id: pluginId, pluginKey, packageName: pluginKey, version: "0.1.0", manifestJson: manifest, status: "ready" }); + const accounts = aiConnectionService(db); + const codex = await accounts.save(companyId, "alice", { provider: "openai", method: "api_key", ownership: "personal", name: "Codex", apiKey: "fixture", agentIds: [], allAgents: true }, "fixture-codex"); + const claude = await accounts.save(companyId, "alice", { provider: "anthropic", method: "subscription", ownership: "personal", name: "Claude", loginSessionId: "fixture", agentIds: [], allAgents: true }, "fixture-claude"); + members = [ + { id: randomUUID(), binding: { mode: "delegated", provider: "openai", method: "api_key", ...codex }, profile: { provider: "codex", model: "gpt-5.6-sol", effort: "low" } }, + { id: randomUUID(), binding: { mode: "delegated", provider: "anthropic", method: "subscription", ...claude }, profile: { provider: "acpx", acpxAgent: "claude", model: "claude-sonnet-5" } }, + ]; +}, 90000); +afterAll(async () => { await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); }); + +describe("durable, authorized connection routing", () => { + it("defaults off and does not expose selectable pools before opt-in", async () => { + expect((await instanceSettingsService(db).getExperimental()).enableAiConnectionRouters).toBe(false); + await expect(makePool()).rejects.toThrow("Experimental"); + expect(await service().selectable(companyId, "alice")).toEqual([]); + await instanceSettingsService(db).updateExperimental({ enableAiConnectionRouters: true }); + }); + it("requires company connection management authority on pool configuration routes", async () => { + const app = express(); app.use(express.json()); + let actor = { type: "board", userId: "alice", source: "session", companyIds: [companyId], memberships: [{ companyId, status: "active", membershipRole: "owner" }] }; + app.use((req, _res, next) => { req.actor = actor as typeof req.actor; next(); }); app.use("/api", aiConnectionRoutes(db)); app.use(errorHandler); + const pool = await makePool(); const url = `/api/companies/${companyId}/ai-connection-pools`; + expect((await request(app).get(url)).status).toBe(200); + expect((await request(app).get(`/api/companies/${otherCompanyId}/ai-connection-pools`)).status).toBe(403); + actor = { ...actor, memberships: [{ companyId, status: "active", membershipRole: "viewer" }] }; + expect((await request(app).get(url)).status).toBe(403); + expect((await request(app).post(url).send({ pluginKey, config: pool })).status).toBe(403); + expect((await request(app).get(`${url}/${pool.id}/inspection`)).status).toBe(403); + actor = { type: "agent", agentId, companyId, source: "agent_jwt" } as unknown as typeof actor; + expect((await request(app).get(url)).status).toBe(403); + }); + it("rotates new tasks across qualified backends and retains pins on turns, resets and service restart", async () => { + const pool = await makePool(); + expect((await resolve(pool.id, "one")).memberId).toBe(members[0]!.id); + expect((await resolve(pool.id, "two")).memberId).toBe(members[1]!.id); + expect((await resolve(pool.id, "one", { overrides: { model: "claude-sonnet-5", modelReasoningEffort: "high" } })).memberId).toBe(members[0]!.id); + expect((await aiConnectionRouterService(db, worker).resolve({ companyId, poolId: pool.id, agentId, userId: "alice", adapterType: "paperclip_runner", taskKey: "one" })).memberId).toBe(members[0]!.id); + const [cursor] = await db.select().from(aiConnectionRouterCursors).where(eq(aiConnectionRouterCursors.poolId, pool.id)); expect(cursor?.version).toBe(2); + }); + it("commits one allocation for duplicate starts and serializes concurrent tasks", async () => { + const pool = await makePool(); + const same = await Promise.all(Array.from({ length: 8 }, () => resolve(pool.id, "duplicate"))); + expect(new Set(same.map(s => s.memberId)).size).toBe(1); + const many = await Promise.all(Array.from({ length: 8 }, (_, n) => resolve(pool.id, `concurrent-${n}`))); + expect(many.filter(s => s.memberId === members[0]!.id)).toHaveLength(4); + const pins = await db.select().from(aiConnectionTaskPins).where(eq(aiConnectionTaskPins.poolId, pool.id)); expect(pins).toHaveLength(9); + const [cursor] = await db.select().from(aiConnectionRouterCursors).where(eq(aiConnectionRouterCursors.poolId, pool.id)); expect(cursor?.version).toBe(9); + }); + it("rolls back both pin and cursor on a failure during commit", async () => { + const pool = await makePool(); + await db.execute(`CREATE FUNCTION fixture_router_crash() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture commit crash'; END $$`); + await db.execute(`CREATE TRIGGER fixture_router_crash BEFORE UPDATE ON ai_connection_router_cursors FOR EACH ROW EXECUTE FUNCTION fixture_router_crash()`); + try { await expect(resolve(pool.id, "crash")).rejects.toThrow(); } finally { await db.execute(`DROP TRIGGER fixture_router_crash ON ai_connection_router_cursors`); await db.execute(`DROP FUNCTION fixture_router_crash()`); } + expect(await db.select().from(aiConnectionTaskPins).where(eq(aiConnectionTaskPins.poolId, pool.id))).toHaveLength(0); + const selected = await resolve(pool.id, "crash"); expect(selected.memberId).toBe(members[0]!.id); + const [cursor] = await db.select().from(aiConnectionRouterCursors).where(eq(aiConnectionRouterCursors.poolId, pool.id)); expect(cursor?.version).toBe(1); + }); + it("rechecks config revisions after a proposal and preserves removed member snapshots", async () => { + const pool = await makePool(); const old = await resolve(pool.id, "existing"); + const saved = await service().save(pluginKey, { companyId, id: pool.id, expectedRevision: pool.revision, config: { name: "Edited", enabled: true, mode: "round_robin", thresholdPercent: 90, members: [members[1]!] } }, "alice"); + await expect(service().save(pluginKey, { companyId, id: pool.id, expectedRevision: pool.revision, config: saved }, "alice")).rejects.toThrow(); + expect((await resolve(pool.id, "existing")).binding).toEqual(old.binding); + expect((await resolve(pool.id, "future")).memberId).toBe(members[1]!.id); + }); + it("keeps account and harness pinned through independent composer fallback", async () => { + const pool = await makePool(); + const first = await resolve(pool.id, "composer"); + const changed = await resolve(pool.id, "composer", { overrides: { model: "claude-sonnet-5", modelReasoningEffort: "high" } }); + expect(changed.binding).toEqual(first.binding); expect(changed.runtimeConfig).toMatchObject({ provider: "codex", model: "gpt-5.6-sol", modelReasoningEffort: "high" }); expect(changed.notes).toHaveLength(1); + const claude = poolMemberRuntimeConfig(members[1]!, "paperclip_runner", { model: "gpt-5.6-sol", modelReasoningEffort: "ultra" }); + expect(claude.config).toMatchObject({ provider: "acpx", acpxAgent: "claude", model: "claude-sonnet-5" }); expect(claude.notes).toHaveLength(2); + const overrides = applyAiConnectionRouterTaskSettings({ model: "claude-sonnet-5", effort: "invalid", cwd: "/fixture" }, changed); + const native = resolvePaperclipRunnerNativeProviderInput({ backend: "codex_app_server", adapterConfig: projectPaperclipRunnerTaskConfig("codex_app_server", changed.runtimeConfig, overrides) }); + expect(native).toMatchObject({ provider: "codex", model: "gpt-5.6-sol", codexReasoningEffort: "high" }); + expect(overrides).toMatchObject({ cwd: "/fixture" }); expect(overrides).not.toHaveProperty("effort"); + }); + it("retries a changed config proposal without advancing the abandoned selection", async () => { + const pool = await makePool(); + proposal.mockImplementationOnce(async () => { + await service().save(pluginKey, { companyId, id: pool.id, expectedRevision: pool.revision, config: { name: pool.name, enabled: true, mode: pool.mode, thresholdPercent: pool.thresholdPercent, members: [members[1]!] } }, "alice"); + return { kind: "selected", memberId: members[0]!.id }; + }); + expect((await resolve(pool.id, "revision-conflict")).memberId).toBe(members[1]!.id); + const [cursor] = await db.select().from(aiConnectionRouterCursors).where(eq(aiConnectionRouterCursors.poolId, pool.id)); expect(cursor?.version).toBe(1); + }); + it("fails closed when the experimental flag changes during a proposal", async () => { + const pool = await makePool(); + proposal.mockImplementationOnce(async () => { await instanceSettingsService(db).updateExperimental({ enableAiConnectionRouters: false }); return { kind: "selected", memberId: members[0]!.id }; }); + try { await expect(resolve(pool.id, "disabled-in-flight")).rejects.toThrow("disabled during allocation"); expect(await db.select().from(aiConnectionTaskPins).where(eq(aiConnectionTaskPins.poolId, pool.id))).toHaveLength(0); } + finally { await instanceSettingsService(db).updateExperimental({ enableAiConnectionRouters: true }); } + }); + it("uses only compatible legacy members and makes reassignment an independent pin", async () => { + const pool = await makePool(); + expect((await resolve(pool.id, "legacy", { adapterType: "claude_local" })).memberId).toBe(members[1]!.id); + expect((await resolve(pool.id, "legacy", { agentId: secondAgentId, adapterType: "codex_local" })).memberId).toBe(members[0]!.id); + const pins = await db.select().from(aiConnectionTaskPins).where(eq(aiConnectionTaskPins.poolId, pool.id)); expect(pins).toHaveLength(2); + }); + it("adopts a valid existing account and requires explicit reset for an unrelated session", async () => { + const pool = await makePool(); + expect((await resolve(pool.id, "adopt", { requireExisting: true, existingGrantId: members[1]!.binding.grantId })).memberId).toBe(members[1]!.id); + await expect(resolve(pool.id, "foreign-session", { requireExisting: true, existingGrantId: randomUUID() })).rejects.toThrow("Reset"); + expect((await resolve(pool.id, "foreign-session")).memberId).toBe(members[0]!.id); + }); + it("rejects cross-company access and unauthorized plugin proposals", async () => { + const pool = await makePool(); + await expect(resolve(pool.id, "foreign", { companyId: otherCompanyId })).rejects.toThrow(); + proposal.mockResolvedValueOnce({ kind: "selected", memberId: randomUUID() }); + await expect(resolve(pool.id, "bad-proposal")).rejects.toThrow("unauthorized"); + expect(await db.select().from(aiConnectionTaskPins).where(eq(aiConnectionTaskPins.poolId, pool.id))).toHaveLength(0); + }); + it("does not grant members to another responsible user or move revoked pins", async () => { + const pool = await makePool(); const selected = await resolve(pool.id, "revoke"); + await expect(resolve(pool.id, "revoke", { userId: "bob" })).rejects.toThrow(); + await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, selected.binding.mode === "responsible_user" ? "" : selected.binding.grantId)); + try { await expect(resolve(pool.id, "revoke")).rejects.toThrow(); expect((await resolve(pool.id, "new")).memberId).toBe(members[1]!.id); } + finally { await db.update(connectionGrants).set({ status: "active" }).where(eq(connectionGrants.id, members[0]!.binding.grantId)); } + }); + it("never probes pure rotation, caches authorized observations, and invalidates on credential change", async () => { + const spy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => new Response(JSON.stringify({ five_hour: { utilization: 12 } }))); + try { + const pure = await makePool(); await resolve(pure.id, "pure"); expect(spy).not.toHaveBeenCalled(); + const aware = await makePool({ mode: "usage_aware", members: [members[1]!] }); await resolve(aware.id, "aware"); expect(spy).toHaveBeenCalledTimes(1); await resolve(aware.id, "aware"); expect(spy).toHaveBeenCalledTimes(1); + expect((await service().inspect(companyId, aware.id, "alice"))[members[1]!.id]?.checkedAt).not.toBeNull(); + const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, members[1]!.binding.grantId)); const ref = grant!.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!; + await secretService(db).rotate(ref.secretId, { value: "fixture-refreshed", preserveAiSessionEpoch: true }, { userId: "alice" }); + expect((await service().inspect(companyId, aware.id, "alice"))[members[1]!.id]?.checkedAt).toBeNull(); + await resolve(aware.id, "aware"); expect(spy).toHaveBeenCalledTimes(2); + proposal.mockResolvedValueOnce({ kind: "exhausted", retryAt: new Date(Date.now() + 60_000).toISOString(), skipped: {} }); + await expect(resolve(aware.id, "aware")).rejects.toBeInstanceOf(AiConnectionPoolExhausted); + const pins = await db.select().from(aiConnectionTaskPins).where(eq(aiConnectionTaskPins.poolId, aware.id)); expect(pins).toHaveLength(1); + } finally { spy.mockRestore(); } + }); + it("expires a usage observation at a reported reset", async () => { + const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, members[1]!.binding.grantId)); + const ref = grant!.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!; + await secretService(db).rotate(ref.secretId, { value: "fixture-reset-cache", preserveAiSessionEpoch: true }, { userId: "alice" }); + const spy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => new Response(JSON.stringify({ five_hour: { utilization: 12, resets_at: new Date(Date.now() + 200).toISOString() } }))); + try { + const pool = await makePool({ mode: "usage_aware", members: [members[1]!] }); await resolve(pool.id, "reset-cache"); expect(spy).toHaveBeenCalledTimes(1); + await new Promise(resolve => setTimeout(resolve, 250)); await resolve(pool.id, "reset-cache"); expect(spy).toHaveBeenCalledTimes(2); + } finally { spy.mockRestore(); } + }); + it("treats unfinished probes as unknown within the shared selection budget", async () => { + const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, members[1]!.binding.grantId)); + const ref = grant!.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!; + await secretService(db).rotate(ref.secretId, { value: "fixture-slow-probe", preserveAiSessionEpoch: true }, { userId: "alice" }); + const spy = vi.spyOn(globalThis, "fetch").mockImplementation(() => new Promise(() => {})); + try { const pool = await makePool({ mode: "usage_aware", members: [members[1]!] }); const start = Date.now(); expect((await resolve(pool.id, "slow-probe")).memberId).toBe(members[1]!.id); expect(Date.now() - start).toBeLessThan(20_000); expect(proposal.mock.calls.at(-1)?.[2].candidates[0]?.usage).toBeUndefined(); } + finally { spy.mockRestore(); } + }, 25_000); + it("rejects disabled routing but recovers server-owned native evidence after disable/uninstall", async () => { + const pool = await makePool(); const selected = await resolve(pool.id, "recovery"); + await instanceSettingsService(db).updateExperimental({ enableAiConnectionRouters: false }); + try { await expect(resolve(pool.id, "recovery")).rejects.toThrow("Experimental"); expect(await resolve(pool.id, "recovery", { persisted: selected })).toEqual(selected); } + finally { await instanceSettingsService(db).updateExperimental({ enableAiConnectionRouters: true }); } + await db.update(plugins).set({ status: "disabled" }).where(eq(plugins.id, pluginId)); + try { await expect(resolve(pool.id, "new")).rejects.toThrow("plugin"); expect(await resolve(pool.id, "recovery", { persisted: selected })).toEqual(selected); await db.delete(toolConnections).where(eq(toolConnections.id, pool.id)); expect(await resolve(pool.id, "recovery", { persisted: selected })).toEqual(selected); } + finally { await db.update(plugins).set({ status: "ready" }).where(eq(plugins.id, pluginId)); } + }); + it("keeps session identity and fingerprints stable on authenticated refresh, but changes them on manual replacement", async () => { + const input = { companyId, agentId, responsibleUserId: "alice", adapterType: "paperclip_runner", binding: members[0]!.binding, config: { provider: "codex", model: "gpt-5.6-sol" } }; + const first = await prepareManagedAiRuntime(db, input); + const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, members[0]!.binding.grantId)); const ref = grant!.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!; + try { + await secretService(db).rotate(ref.secretId, { value: "fixture-new-token", preserveAiSessionEpoch: true }, { userId: "alice" }); + const refreshed = await prepareManagedAiRuntime(db, input); + try { expect(refreshed.identity).not.toBe(first.identity); expect(refreshed.sessionIdentity).toBe(first.sessionIdentity); expect(managedAiSessionFingerprintConfig(refreshed.config, refreshed.home)).toEqual(managedAiSessionFingerprintConfig(first.config, first.home)); } finally { await refreshed.cleanup(); } + await secretService(db).rotate(ref.secretId, { value: "fixture-replacement" }, { userId: "alice" }); + const replaced = await prepareManagedAiRuntime(db, input); try { expect(replaced.sessionIdentity).not.toBe(first.sessionIdentity); } finally { await replaced.cleanup(); } + const [secret] = await db.select().from(companySecrets).where(and(eq(companySecrets.companyId, companyId), eq(companySecrets.id, ref.secretId))); expect(secret!.aiSessionEpoch).toBeGreaterThan(0); + } finally { await first.cleanup(); } + }); +}); diff --git a/server/src/__tests__/instance-settings-service.test.ts b/server/src/__tests__/instance-settings-service.test.ts index 31b6d1751ba..ce050c21274 100644 --- a/server/src/__tests__/instance-settings-service.test.ts +++ b/server/src/__tests__/instance-settings-service.test.ts @@ -20,6 +20,7 @@ describe("instance settings service", () => { expect(normalizeExperimentalSettings({ enableEnvironments: true, enableNativeRunner: false, + enableAiConnectionRouters: false, enableManagedSandboxOnly: false, enableIsolatedWorkspaces: true, enableIssuePlanDecompositions: true, @@ -35,6 +36,7 @@ describe("instance settings service", () => { })).toEqual({ enableEnvironments: true, enableNativeRunner: false, + enableAiConnectionRouters: false, enableManagedSandboxOnly: false, enableIsolatedWorkspaces: true, enableIsolatedWorkspacesByDefault: false, diff --git a/server/src/app.ts b/server/src/app.ts index d7107c260c8..c99839e4fab 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -897,7 +897,7 @@ export async function createApp( lifecycleManager: lifecycle, instanceInfo: { instanceId: opts.instanceId ?? "default", - hostVersion: opts.hostVersion ?? "0.0.0", + hostVersion: opts.hostVersion ?? serverVersion, deploymentMode: opts.deploymentMode, deploymentExposure: opts.deploymentExposure, }, diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index b882b09baca..adaf1a97ce6 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -127,7 +127,7 @@ function readNonEmptyString(value: unknown): string | null { function classifyRetryReasonKind(retryReason: string | null): RetryReasonKind { if (retryReason === MAX_TURN_CONTINUATION_RETRY_REASON) return "max_turn_continuation"; if (retryReason === ISSUE_DISPOSITION_REPAIR_RETRY_REASON) return "disposition_repair"; - if (retryReason === "ai_connection_busy") return "ai_connection_wait"; + if ((retryReason === "ai_connection_busy" || retryReason === "ai_connection_pool_wait")) return "ai_connection_wait"; if (retryReason === "native_safe_replacement") return "native_safe_replacement"; return "other"; } diff --git a/server/src/modules/run-dispatch/domain/wake-context.test.ts b/server/src/modules/run-dispatch/domain/wake-context.test.ts index 223ee6fb8bd..ac6b84eb68e 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.test.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.test.ts @@ -91,3 +91,8 @@ describe("wake context", () => { }, ); }); + + it("preserves non-assignee pool wait authority", () => { + expect(isNonAssigneeWorkspaceBusyRetry("ai_connection_pool_wait", { aiConnectionBusyDeferredWhileAssignee: false })).toBe(true); + expect(isNonAssigneeWorkspaceBusyRetry("ai_connection_pool_wait", { aiConnectionBusyDeferredWhileAssignee: true })).toBe(false); + }); diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index b0dd653959e..2100b357d9c 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -17,6 +17,7 @@ function readNonEmptyString(value: unknown): string | null { export const MAX_TURN_CONTINUATION_RETRY_REASON = "max_turns_continuation"; export const WORKSPACE_BUSY_RETRY_REASON = "workspace_busy"; +export const AI_CONNECTION_POOL_WAIT_RETRY_REASON = "ai_connection_pool_wait"; export const AI_CONNECTION_BUSY_RETRY_REASON = "ai_connection_busy"; export const INTERACTION_CONTINUATION_INFRA_RETRY_REASON = "interaction_continuation_infra_retry"; export const INTERACTION_CONTINUATION_INFRA_WAKE_REASON = "interaction_continuation_infra_retry"; @@ -41,7 +42,7 @@ export function isNonAssigneeWorkspaceBusyRetry( return ( (retryReason === WORKSPACE_BUSY_RETRY_REASON && contextSnapshot.workspaceBusyDeferredWhileAssignee === false) || - (retryReason === AI_CONNECTION_BUSY_RETRY_REASON && + ((retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) && contextSnapshot.aiConnectionBusyDeferredWhileAssignee === false) ); } diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts index d6fda59345a..d29b0783231 100644 --- a/server/src/modules/run-dispatch/index.ts +++ b/server/src/modules/run-dispatch/index.ts @@ -13,6 +13,7 @@ export { MAX_TURN_CONTINUATION_RETRY_REASON, WORKSPACE_BUSY_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON, + AI_CONNECTION_POOL_WAIT_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_WAKE_REASON, WAKE_COMMENT_IDS_KEY, diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index c3f1b8a422c..1dc7026a626 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -1,3 +1,4 @@ +import { aiConnectionRouterService, poolMemberRuntimeConfig } from "../services/ai-connection-router.js"; import { connectionIntentService } from "../services/connection-intents.js"; import { completeConnectionIntentSchema } from "@paperclipai/shared"; import { agentFileStore, agentFileTokenFromHash } from "../services/agent-file-store.js"; @@ -5,7 +6,7 @@ import { pipeline } from "node:stream/promises"; import { resolveAgentAppearance, agentAvatarUrl } from "@paperclipai/shared"; import { listOpenRouterModels } from "../services/openrouter-models.js"; import { prepareManagedAiRuntime, assertManagedAiProjectAuth, stripAiAuthBindings } from "../services/ai-connection-runtime.js"; -import { ADAPTER_AUTH_MISSING_CHECK_CODE, AI_CONNECTION_CAPABILITIES, aiConnectionBindingSchema, type AiConnectionBinding } from "@paperclipai/shared"; +import { ADAPTER_AUTH_MISSING_CHECK_CODE, AI_CONNECTION_CAPABILITIES, aiConnectionBindingSchema, type AiConnectionBinding, aiRuntimeConnectionBindingSchema, type AiRuntimeConnectionBinding } from "@paperclipai/shared"; import { toolConnections } from "@paperclipai/db"; import { aiConnectionService } from "../services/ai-connections.js"; import { defaultAiConnectionForHire } from "../services/agent-ai-connection-default.js"; @@ -3241,6 +3242,21 @@ export function agentRoutes( const companyId = req.params.companyId as string; assertCompanyAccess(req, companyId); const type = assertKnownAdapterType(req.params.type as string); + const poolId = asNonEmptyString(req.query.poolId); + if (poolId) { + const pool = (await aiConnectionRouterService(db).selectable(companyId, getActorInfo(req).actorId)).find(pool => pool.id === poolId); + if (!pool) throw unprocessable("Enable an accessible AI connection pool"); + const catalog = new Map(); + for (const member of pool.members) { + const memberAdapter = member.binding.provider === "openai" ? "codex_local" : member.binding.provider === "anthropic" ? "claude_local" : member.binding.provider === "xai" ? "grok_local" : "opencode_local"; + const models = [{ id: member.profile.model, label: member.profile.model }, ...await listAdapterModels(memberAdapter)]; + for (const model of models) { + try { poolMemberRuntimeConfig({ ...member, profile: { ...member.profile, model: model.id } }, type); catalog.set(model.id, model); } catch { /* Only qualified profiles belong in the pool catalog. */ } + } + } + res.json([...catalog.values()]); + return; + } const refresh = typeof req.query.refresh === "string" ? ["1", "true", "yes"].includes(req.query.refresh.toLowerCase()) : false; @@ -3379,7 +3395,16 @@ export function agentRoutes( return result; } - async function validateManagedAgentBinding(req: Request, companyId: string, agentId: string, adapterType: string, config: Record, binding: AiConnectionBinding, environmentId: string | null | undefined, test: boolean, newAgent = false) { + async function validateManagedAgentBinding(req: Request, companyId: string, agentId: string, adapterType: string, config: Record, binding: AiRuntimeConnectionBinding, environmentId: string | null | undefined, test: boolean, newAgent = false) { + if (binding.mode === "router") { + + const settings = await import("../services/instance-settings.js"); + if (!(await settings.instanceSettingsService(db).getExperimental()).enableAiConnectionRouters) throw unprocessable("Enable AI connection routers in Experimental settings"); + const pool = (await aiConnectionRouterService(db).selectable(companyId, responsibleUserForAiRequest(req) ?? "")).find((p) => p.id === binding.connectionId && p.enabled); + if (!pool) throw unprocessable("Enable the selected connection pool"); + if (!pool.members.some((m) => { try { poolMemberRuntimeConfig(m, adapterType); return true; } catch { return false; } })) throw unprocessable("This pool has no compatible harness"); + return pool.id; + } const userId = responsibleUserForAiRequest(req); const allowUninstalledShared = newAgent && await canInstallSharedAiConnectionForNewAgent(db, req, companyId, binding); const selection = await aiConnectionService(db).select({ companyId, agentId, userId, adapterType, model: config.model, runnerProvider: config.provider, acpxAgent: config.acpxAgent, binding, allowUninstalledPersonal: newAgent, allowUninstalledShared, allowLegacyValidation: test }).catch((error: unknown) => { @@ -3422,7 +3447,8 @@ export function agentRoutes( const adapter = requireServerAdapter(type); - const aiBinding = req.body.aiConnection ? aiConnectionBindingSchema.parse(req.body.aiConnection) : undefined; + const requestedBinding = req.body.aiConnection ? aiRuntimeConnectionBindingSchema.parse(req.body.aiConnection) : undefined; + const aiBinding = requestedBinding?.mode === "router" ? undefined : requestedBinding; if (aiBinding && req.body.testCredentials && Object.keys(req.body.testCredentials).length) throw unprocessable("A managed connection test cannot override its credentials"); const inputAdapterConfig = aiBinding ? { ...req.body.adapterConfig, env: stripAiAuthBindings(req.body.adapterConfig?.env) } : (req.body?.adapterConfig ?? {}) as Record; const savedAgentId = typeof req.body.agentId === "string" ? req.body.agentId : null; @@ -3434,6 +3460,12 @@ export function agentRoutes( if (savedAgent.companyId !== companyId) throw notFound("Agent not found"); await assertCanUpdateAgent(req, savedAgent); } + const poolBinding = requestedBinding ?? (savedAgent?.runtimeConfig?.aiConnection ? aiRuntimeConnectionBindingSchema.parse(savedAgent.runtimeConfig.aiConnection) : undefined); + if (poolBinding?.mode === "router") { + await validateManagedAgentBinding(req, companyId, savedAgentId ?? "", type, req.body.adapterConfig ?? {}, poolBinding, req.body.environmentId, false); + res.json({ adapterType: type, status: "warn", testedAt: new Date().toISOString(), checks: [{ code: "ai_connection_pool_task_test_required", level: "warn", message: "Pool configuration is available. Run a task to verify the selected account and harness; this preview does not allocate a task or contact a provider." }] }); + return; + } const requestedEnvironmentId = await resolveAdapterTestEnvironmentId( companyId, // Omission tests the saved selection. Explicit null tests a prospective @@ -4632,7 +4664,7 @@ export function agentRoutes( const requiresApproval = company.requireBoardApprovalForNewAgents; const status = requiresApproval ? "pending_approval" : "idle"; - const managedHireBinding = normalizedHireInput.runtimeConfig?.aiConnection ? aiConnectionBindingSchema.parse(normalizedHireInput.runtimeConfig.aiConnection) : undefined; + const managedHireBinding = normalizedHireInput.runtimeConfig?.aiConnection ? aiRuntimeConnectionBindingSchema.parse(normalizedHireInput.runtimeConfig.aiConnection) : undefined; const managedHireConnectionId = managedHireBinding ? await validateManagedAgentBinding(req, companyId, hiredAgentId, normalizedHireInput.adapterType, normalizedHireInput.adapterConfig, managedHireBinding, normalizedHireInput.defaultEnvironmentId, false, true) : undefined; const createdAgent = await svc.create( companyId, @@ -4881,7 +4913,7 @@ export function agentRoutes( allowedSandboxProviders: allowedSandboxProvidersForAgent(createInput.adapterType), }); - const managedBinding = normalizedRuntimeConfig.aiConnection ? aiConnectionBindingSchema.parse(normalizedRuntimeConfig.aiConnection) : undefined; + const managedBinding = normalizedRuntimeConfig.aiConnection ? aiRuntimeConnectionBindingSchema.parse(normalizedRuntimeConfig.aiConnection) : undefined; const managedConnectionId = managedBinding ? await validateManagedAgentBinding(req, companyId, agentId, createInput.adapterType, normalizedAdapterConfig, managedBinding, createInput.defaultEnvironmentId, false, true) : undefined; const createdAgent = await svc.create( companyId, @@ -5560,12 +5592,12 @@ export function agentRoutes( }); } if (existing.runtimeConfig.aiConnection && requestedRuntimeConfig && !requestedRuntimeConfig.aiConnection) requestedRuntimeConfig.aiConnection = existing.runtimeConfig.aiConnection; - const nextAiBinding = aiConnectionBindingSchema.safeParse(requestedRuntimeConfig?.aiConnection ?? existing.runtimeConfig.aiConnection).data; + const nextAiBinding = aiRuntimeConnectionBindingSchema.safeParse(requestedRuntimeConfig?.aiConnection ?? existing.runtimeConfig.aiConnection).data; if (nextAiBinding) { await assertCanUpdateAgent(req, existing); const changed = JSON.stringify(nextAiBinding) !== JSON.stringify(existing.runtimeConfig.aiConnection); const aiConfig = (patchData.adapterConfig ?? existing.adapterConfig) as Record; - if (!isAiConnectionCompatible(nextAiBinding, requestedAdapterType, aiConfig.model, aiConfig.provider, aiConfig.acpxAgent)) throw unprocessable("Select an AI connection compatible with the new harness and model"); + if (nextAiBinding.mode !== "router" && !isAiConnectionCompatible(nextAiBinding, requestedAdapterType, aiConfig.model, aiConfig.provider, aiConfig.acpxAgent)) throw unprocessable("Select an AI connection compatible with the new harness and model"); if (changed) await validateManagedAgentBinding(req, existing.companyId, existing.id, requestedAdapterType, aiConfig, nextAiBinding, (patchData.defaultEnvironmentId !== undefined ? patchData.defaultEnvironmentId : existing.defaultEnvironmentId) as string | null, true); } if (requestedRuntimeConfig) patchData.runtimeConfig = requestedRuntimeConfig; diff --git a/server/src/routes/ai-connections.ts b/server/src/routes/ai-connections.ts index 335ed8067b5..1a6a3725f51 100644 --- a/server/src/routes/ai-connections.ts +++ b/server/src/routes/ai-connections.ts @@ -22,11 +22,13 @@ import { type AiProvider, type AiConnectionBinding, type AiConnectionList, + aiConnectionPoolConfigSchema, } from "@paperclipai/shared"; import { assertBoard, assertCompanyAccess, getActorInfo } from "./authz.js"; import { forbidden, notFound, unprocessable } from "../errors.js"; import { accessService } from "../services/access.js"; import { logActivity } from "../services/activity-log.js"; +import { aiConnectionRouterService } from "../services/ai-connection-router.js"; import { aiConnectionService } from "../services/ai-connections.js"; import { validate } from "../middleware/validate.js"; @@ -173,6 +175,28 @@ export function aiConnectionRoutes(db: Db, options: Parameters { + const companyId = req.params.companyId as string; + assertBoard(req); assertCompanyAccess(req, companyId); + if (!(await canManageAiConnections(db, req, companyId))) throw forbidden("Manage connections permission is required"); + res.setHeader("Cache-Control", "no-store"); + res.json(await pools.list(companyId)); + }); + router.get("/companies/:companyId/ai-connection-pools/:poolId/inspection", async (req, res) => { + const companyId = req.params.companyId as string; + assertBoard(req); assertCompanyAccess(req, companyId); + if (!(await canManageAiConnections(db, req, companyId))) throw forbidden("Manage connections permission is required"); + res.setHeader("Cache-Control", "no-store"); + res.json(await pools.inspect(companyId, z.string().uuid().parse(req.params.poolId), getActorInfo(req).actorId)); + }); + router.post("/companies/:companyId/ai-connection-pools", async (req, res) => { + const companyId = req.params.companyId as string; + assertBoard(req); assertCompanyAccess(req, companyId); + if (!(await canManageAiConnections(db, req, companyId))) throw forbidden("Manage connections permission is required"); + const body = z.object({ pluginKey: z.string().min(1).max(160), id: z.string().uuid().optional(), expectedRevision: z.number().int().positive().optional(), config: aiConnectionPoolConfigSchema }).strict().parse(req.body); + res.json(await pools.save(body.pluginKey, { ...body, companyId }, getActorInfo(req).actorId)); + }); function assertLocalOperator(req: Request) { assertBoard(req); assertCompanyAccess(req, req.params.companyId as string); @@ -216,6 +240,7 @@ export function aiConnectionRoutes(db: Db, options: Parameters(); +const routedRuntimeKeys = new Set(["provider", "acpxAgent", "model", "modelReasoningEffort", "reasoningEffort", "effort", "variant"]); +export function applyAiConnectionRouterTaskSettings(config: Record, selection: AiConnectionRouterSelection) { + return { ...Object.fromEntries(Object.entries(config).filter(([key]) => !routedRuntimeKeys.has(key))), ...selection.runtimeConfig }; +} +export class AiConnectionPoolExhausted extends Error { + readonly code = "ai_connection_pool_exhausted"; + constructor(readonly retryAt: string, readonly poolId: string) { super("All eligible pool accounts are over their usage limit. Waiting for usage to reset."); } +} + +/** Routing can override provider/model only, never environment, permissions, or instructions. */ +export function poolMemberRuntimeConfig(member: AiConnectionPoolMember, adapterType: string, overrides: Record = {}) { + const notes: string[] = []; + const provider = member.binding.provider; + const profile = member.profile; + const expected = provider === "openai" ? profile.provider === "codex" + : provider === "anthropic" ? profile.provider === "acpx" && profile.acpxAgent === "claude" + : provider === "xai" ? profile.provider === "acpx" && profile.acpxAgent === "grok" + : profile.provider === "opencode"; + if (!expected) throw unprocessable("Pool member harness does not match its account"); + const supportsModel = (value: string) => provider === "openai" ? isCodexLocalKnownModel(value) + : provider === "anthropic" ? claudeModels.some((m) => m.id === value) + : provider === "xai" ? grokModels.some((m) => m.id === value) : value.startsWith("openrouter/"); + const config: Record = { provider: profile.provider, ...(profile.acpxAgent ? { acpxAgent: profile.acpxAgent } : {}), model: profile.model }; + const validate = () => { + if (!isAiConnectionCompatible(member.binding, adapterType, config.model, config.provider, config.acpxAgent)) throw unprocessable("Pool member is incompatible with this harness"); + if (adapterType === "paperclip_runner") { + const resolved = resolvePaperclipRunnerProviderProfile(config); + resolvePaperclipRunnerNativeProviderInput({ backend: resolved.backend, adapterConfig: config }); + } + }; + validate(); + if (typeof overrides.model === "string" && overrides.model !== profile.model) { + if (supportsModel(overrides.model)) { + config.model = overrides.model; + try { validate(); } catch { config.model = profile.model; notes.push("Model override is unavailable for this member; using its default."); } + } else notes.push("Model override is unavailable for this member; using its default."); + } + const model = String(config.model); + const efforts = provider === "openai" ? codexLocalReasoningEffortsForModel(model) + : provider === "anthropic" ? claudeLocalReasoningEffortsForModel(model) + : provider === "xai" ? grokLocalReasoningEffortsForModel(model) : []; + // ACPX currently has no qualified effort override; keep that host limitation explicit. + const effortKey = adapterType === "paperclip_runner" ? profile.provider === "codex" ? "modelReasoningEffort" : null + : provider === "anthropic" ? "effort" : provider === "xai" ? "reasoningEffort" : provider === "openai" ? "modelReasoningEffort" : "variant"; + const requested = overrides.modelReasoningEffort ?? overrides.reasoningEffort ?? overrides.effort ?? overrides.variant; + const effort = typeof requested === "string" && efforts.includes(requested) && effortKey ? requested + : profile.effort && efforts.includes(profile.effort) && effortKey ? profile.effort : null; + if (requested && requested !== effort) notes.push("Effort override is unavailable for this member; using its default."); + if (effort && effortKey) config[effortKey] = effort; + validate(); + return { config, notes }; +} + +export function aiConnectionRouterService(db: Db, workerManager?: PluginWorkerManager) { + async function owner(pluginKey: string) { + const [plugin] = await db.select().from(plugins).where(eq(plugins.pluginKey, pluginKey)); + if (!plugin || plugin.status !== "ready" || !plugin.manifestJson.capabilities.includes("ai.connections.route")) throw unprocessable("Enable a compatible AI connection router plugin"); + return plugin; + } + async function enabled() { + if (!(await instanceSettingsService(db).getExperimental()).enableAiConnectionRouters) throw unprocessable("Enable AI connection routers in Experimental settings", { code: "ai_connection_router_disabled" }); + } + async function list(companyId: string) { + const rows = await db.select().from(aiConnectionPools).where(eq(aiConnectionPools.companyId, companyId)); + return rows.map((row): AiConnectionPool => ({ ...row.config, id: row.id, companyId, pluginKey: row.pluginKey, revision: row.revision })); + } + async function selectable(companyId: string, userId: string) { + if (!(await instanceSettingsService(db).getExperimental()).enableAiConnectionRouters) return []; + const accounts = await aiConnectionService(db).list(companyId, userId); + const result: AiConnectionPool[] = []; + for (const pool of await list(companyId)) { + if (!pool.enabled) continue; + try { await owner(pool.pluginKey); } catch { continue; } + const members = pool.members.filter(member => accounts.some(account => account.id === member.binding.connectionId && account.grantId === member.binding.grantId)); + if (members.length) result.push({ ...pool, members }); + } + return result; + } + async function inspect(companyId: string, poolId: string, userId: string) { + const pool = (await list(companyId)).find(pool => pool.id === poolId); + if (!pool) throw notFound("Connection pool not found"); + const visible = await aiConnectionService(db).list(companyId, userId); + const result: Record = {}; + for (const member of pool.members) { + const account = visible.find(account => account.id === member.binding.connectionId && account.grantId === member.binding.grantId); + const key = account ? await usageCacheKey(member, companyId) : null; + const cached = key ? usageCache.get(key) : undefined; + const observation = cached && cached.expires > Date.now() ? cached.value : undefined; + result[member.id] = { reason: account?.status === "connected" ? "Eligible if agent access permits" : account?.status ?? "Not permitted", checkedAt: account && observation ? observation.checkedAt : null, ...(account && observation ? { usage: observation } : {}) }; + } + return result; + } + async function save(pluginKey: string, input: AiConnectionPoolSaveInput, userId: string) { + await enabled(); + const plugin = await owner(pluginKey); + const config = aiConnectionPoolConfigSchema.parse(input.config); + const visible = await aiConnectionService(db).list(input.companyId, userId); + for (const member of config.members) { + if (!visible.some((v) => v.id === member.binding.connectionId && v.grantId === member.binding.grantId)) throw forbidden("Choose an account you may use in this company"); + poolMemberRuntimeConfig(member, "paperclip_runner"); + } + const id = input.id ?? randomUUID(); + await db.transaction(async (tx) => { + if (input.id) { + const [row] = await tx.select().from(aiConnectionPools).where(and(eq(aiConnectionPools.id, id), eq(aiConnectionPools.companyId, input.companyId), eq(aiConnectionPools.pluginKey, pluginKey))).for("update"); + if (!row) throw notFound("Connection pool not found"); + if (input.expectedRevision !== row.revision) throw conflict("Pool changed; reload before saving"); + await tx.update(aiConnectionPools).set({ config, revision: row.revision + 1, updatedAt: new Date() }).where(eq(aiConnectionPools.id, id)); + await tx.update(toolConnections).set({ name: config.name, enabled: config.enabled, updatedAt: new Date() }).where(and(eq(toolConnections.id, id), eq(toolConnections.companyId, input.companyId))); + } else { + const key = `plugin:${pluginKey}:ai-router`; + await tx.insert(toolApplications).values({ companyId: input.companyId, applicationKey: key, name: `${plugin.manifestJson.displayName} connections`, type: "paperclip_plugin", pluginId: plugin.id }).onConflictDoNothing(); + const [application] = await tx.select().from(toolApplications).where(and(eq(toolApplications.companyId, input.companyId), eq(toolApplications.applicationKey, key))); + if (!application) throw new Error("Router application could not be created"); + await tx.insert(toolConnections).values({ id, companyId: input.companyId, applicationId: application.id, name: config.name, uid: id, connectionPurpose: "ai", transport: "runtime_auth", enabled: config.enabled, status: "active", healthStatus: "ok", config: { aiRouter: { pluginKey } }, createdByUserId: userId }); + await tx.insert(aiConnectionPools).values({ id, companyId: input.companyId, pluginKey, config }); + await tx.insert(aiConnectionRouterCursors).values({ poolId: id, companyId: input.companyId }); + await tx.insert(toolConnectionInstalls).values({ companyId: input.companyId, connectionId: id, targetType: "company", targetId: input.companyId, createdByUserId: userId }); + } + await logActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "user", actorId: userId, action: "ai_connection.pool_saved", entityType: "tool_connection", entityId: id, details: { enabled: config.enabled, memberCount: config.members.length, mode: config.mode } }); + }); + return (await list(input.companyId)).find((p) => p.id === id)!; + } + async function usageCacheKey(member: AiConnectionPoolMember, companyId: string) { + const [grant] = await db.select().from(connectionGrants).where(and(eq(connectionGrants.companyId, companyId), eq(connectionGrants.id, member.binding.grantId))); + if (!grant) return null; + const ref = grant.credentialSecretRefs.find((r) => r.configPath === "ai.credential"); + const [secret] = ref ? await db.select({ version: companySecrets.latestVersion }).from(companySecrets).where(and(eq(companySecrets.companyId, companyId), eq(companySecrets.id, ref.secretId))) : []; + return `${companyId}:${grant.id}:${secret?.version}:${grant.updatedAt.toISOString()}`; + } + async function probe(member: AiConnectionPoolMember, companyId: string, userId: string, deadline: number) { + const key = await usageCacheKey(member, companyId); + if (!key) return undefined; + const cached = usageCache.get(key); + if (cached && cached.expires > Date.now()) return cached.value; + let timer: ReturnType | undefined; + try { + const value = await Promise.race([ + aiConnectionService(db).probeUsage(companyId, userId, member.binding.connectionId, member.binding.grantId).catch(() => undefined), + new Promise((resolve) => { timer = setTimeout(() => resolve(undefined), Math.max(0, deadline - Date.now())); }), + ]); + if (value) { + const resets = value.limits.map((v) => Date.parse(v.resetsAt ?? "")).filter((v) => Number.isFinite(v) && v > Date.now()); + if (usageCache.size > 1000) usageCache.clear(); + usageCache.set(key, { value, expires: Math.min(Date.now() + 60_000, ...resets) }); + } + return value; + } finally { if (timer) clearTimeout(timer); } + } + async function resolve(input: { companyId: string; poolId: string; agentId: string; userId: string | null; adapterType: string; taskKey: string; overrides?: Record; existingGrantId?: string; requireExisting?: boolean; persisted?: AiConnectionRouterSelection }) { + const pinWhere = and(eq(aiConnectionTaskPins.companyId, input.companyId), eq(aiConnectionTaskPins.poolId, input.poolId), eq(aiConnectionTaskPins.agentId, input.agentId), eq(aiConnectionTaskPins.taskKey, input.taskKey)); + const validateSelection = async (selection: AiConnectionRouterSelection, client = db) => aiConnectionService(client).select({ companyId: input.companyId, agentId: input.agentId, userId: input.userId, adapterType: input.adapterType, binding: selection.binding, model: selection.runtimeConfig.model, runnerProvider: selection.runtimeConfig.provider, acpxAgent: selection.runtimeConfig.acpxAgent }); + if (input.persisted) { + if (input.persisted.poolId !== input.poolId) throw unprocessable("Persisted pool selection does not match this run"); + const [pin] = await db.select().from(aiConnectionTaskPins).where(pinWhere); + // Uninstall can remove the virtual connection. Server-owned native recovery + // evidence still authorizes the original concrete selection, never a new one. + if (pin && (pin.selection.memberId !== input.persisted.memberId || JSON.stringify(pin.selection.binding) !== JSON.stringify(input.persisted.binding))) throw unprocessable("Persisted pool allocation is inconsistent"); + await validateSelection(input.persisted); + return input.persisted; + } + await enabled(); + for (let attempt = 0; attempt < 20; attempt++) { + const [row] = await db.select().from(aiConnectionPools).where(and(eq(aiConnectionPools.id, input.poolId), eq(aiConnectionPools.companyId, input.companyId))); + if (!row || !row.config.enabled) throw unprocessable("Enable the selected connection pool"); + const plugin = await owner(row.pluginKey); + if (!workerManager?.isRunning(plugin.id)) throw unprocessable("The connection router worker is unavailable; retry when it is ready"); + const [pin] = await db.select().from(aiConnectionTaskPins).where(pinWhere); + const [cursor] = await db.select().from(aiConnectionRouterCursors).where(and(eq(aiConnectionRouterCursors.poolId, row.id), eq(aiConnectionRouterCursors.companyId, input.companyId))); + if (!cursor) throw unprocessable("The connection pool cursor is unavailable"); + const pool: AiConnectionPool = { ...row.config, id: row.id, companyId: row.companyId, pluginKey: row.pluginKey, revision: row.revision }; + const deadline = Date.now() + 15_000; + const candidates: AiConnectionRouterRequest["candidates"] = []; + for (const member of pin ? [pin.member] : pool.members) { + try { + const effective = poolMemberRuntimeConfig(member, input.adapterType, input.overrides); + const selection: AiConnectionRouterSelection = { poolId: row.id, memberId: member.id, binding: member.binding, runtimeConfig: effective.config, notes: effective.notes }; + await validateSelection(selection); + candidates.push({ member, runtimeConfig: effective.config, notes: effective.notes }); + } catch (error) { if (pin) throw error; } + } + if (!pin && input.requireExisting) { + const existing = candidates.find((candidate) => candidate.member.binding.grantId === input.existingGrantId); + if (!existing) throw unprocessable("Reset this task session before adopting the connection pool"); + candidates.splice(0, candidates.length, existing); + } + if (pool.mode === "usage_aware" && input.userId) { + // At most four provider calls at once; the shared deadline bounds the whole pool. + for (let i = 0; i < candidates.length && Date.now() < deadline; i += 4) await Promise.all(candidates.slice(i, i + 4).map(async (candidate) => { candidate.usage = await probe(candidate.member, input.companyId, input.userId!, deadline); })); + } + const authorizedPool = { ...pool, members: candidates.map((candidate) => candidate.member) }; + const result = await workerManager.call(plugin.id, "routeAiConnection", { companyId: input.companyId, pool: authorizedPool, memberOrder: pool.members.map((member) => member.id), agentId: input.agentId, taskKey: input.taskKey, lastMemberId: cursor.lastMemberId, cursorVersion: cursor.version, ...(pin ? { pinnedMemberId: pin.member.id } : {}), candidates, now: new Date().toISOString() }, 5000); + if (result.kind === "exhausted") { + const retry = Date.parse(result.retryAt); + throw new AiConnectionPoolExhausted(new Date(Number.isFinite(retry) ? Math.max(Date.now() + 1000, retry) : Date.now() + 60_000).toISOString(), pool.id); + } + if (result.kind !== "selected") throw unprocessable("No pool account is permitted and compatible for this task"); + const candidate = candidates.find((v) => v.member.id === result.memberId); + if (!candidate || (pin && candidate.member.id !== pin.member.id)) throw unprocessable("The router returned an unauthorized member"); + const selected: AiConnectionRouterSelection = { poolId: row.id, memberId: candidate.member.id, binding: candidate.member.binding, runtimeConfig: candidate.runtimeConfig, notes: candidate.notes }; + if (pin) return selected; + const committed = await db.transaction(async (tx) => { + const [locked] = await tx.select().from(aiConnectionRouterCursors).where(and(eq(aiConnectionRouterCursors.poolId, row.id), eq(aiConnectionRouterCursors.companyId, input.companyId))).for("update"); + const [existing] = await tx.select().from(aiConnectionTaskPins).where(pinWhere); + if (existing) return { retry: true }; + const [current] = await tx.select().from(aiConnectionPools).where(and(eq(aiConnectionPools.id, row.id), eq(aiConnectionPools.companyId, input.companyId))).for("share"); + if (!locked || locked.version !== cursor.version || !current || current.revision !== row.revision) return { retry: true }; + const [currentPlugin] = await tx.select().from(plugins).where(eq(plugins.id, plugin.id)).for("share"); + if (currentPlugin?.status !== "ready") throw unprocessable("The router plugin was disabled during allocation"); + await tx.select({ id: instanceSettings.id }).from(instanceSettings).where(eq(instanceSettings.singletonKey, "default")).for("share"); + if (!(await instanceSettingsService(tx as unknown as Db).getExperimental()).enableAiConnectionRouters) throw unprocessable("AI connection routers were disabled during allocation"); + await validateSelection(selected, tx as unknown as Db); + await tx.insert(aiConnectionTaskPins).values({ companyId: input.companyId, poolId: row.id, agentId: input.agentId, taskKey: input.taskKey, selection: selected, member: candidate.member }); + await tx.update(aiConnectionRouterCursors).set({ lastMemberId: candidate.member.id, version: sql`${aiConnectionRouterCursors.version} + 1` }).where(eq(aiConnectionRouterCursors.poolId, row.id)); + await logActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "system", agentId: input.agentId, action: "ai_connection.pool_selected", entityType: "tool_connection", entityId: row.id, details: { memberId: candidate.member.id, connectionId: candidate.member.binding.connectionId, taskKey: input.taskKey } }); + return { retry: false }; + }); + if (!committed.retry) return selected; + } + throw conflict("Concurrent pool selections changed the cursor; retry this task"); + } + return { list, selectable, inspect, save, resolve }; +} diff --git a/server/src/services/ai-connection-runtime.ts b/server/src/services/ai-connection-runtime.ts index 534f5cf023a..88cd475f2b2 100644 --- a/server/src/services/ai-connection-runtime.ts +++ b/server/src/services/ai-connection-runtime.ts @@ -189,7 +189,13 @@ export function managedAiSessionFingerprintConfig( for (const [key, value] of Object.entries(managedAiHomeEnvironment(managedHome))) { if (env[key] === value) env[key] = stable[key]; } - return { ...config, env }; + const managed = config.managedAiConnection as Record | undefined; + if (managed?.sessionIdentity) { + for (const key of ["ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN", "OPENAI_API_KEY", "CODEX_API_KEY", "OPENROUTER_API_KEY", "XAI_API_KEY", "GROK_API_KEY", "OPENCODE_AUTH_JSON", "OPENCODE_CONFIG_CONTENT"]) { + if (env[key]) env[key] = ""; + } + } + return { ...config, env, ...(managed?.sessionIdentity ? { managedAiConnection: { ...managed, identity: managed.sessionIdentity } } : {}) }; } export async function prepareManagedAiRuntime( @@ -251,7 +257,14 @@ export async function prepareManagedAiRuntime( throw unprocessable( "The selected default changed. Retry this execution.", ); + const credentialRef = selection.grant.credentialSecretRefs.find((ref) => ref.configPath === "ai.credential"); + if (!credentialRef) throw unprocessable("The selected AI credential is unavailable"); + const readFreshness = async () => (await db.select({ epoch: companySecrets.aiSessionEpoch, version: companySecrets.latestVersion }) + .from(companySecrets).where(and(eq(companySecrets.companyId, input.companyId), eq(companySecrets.id, credentialRef.secretId))).limit(1))[0]; + const freshness = await readFreshness(); const value = await service.credential(selection); + const afterRead = await readFreshness(); + if (!freshness || freshness.version !== afterRead?.version || freshness.epoch !== afterRead.epoch) throw unprocessable("The AI credential changed during preparation; retry this execution"); home = await mkdtemp( path.join( os.tmpdir(), @@ -298,11 +311,13 @@ export async function prepareManagedAiRuntime( .digest("hex") .slice(0, 16); const identity = `${selection.grant.id}:${input.responsibleUserId ?? "shared"}:${generation}`; + const sessionIdentity = `${selection.grant.id}:${input.responsibleUserId ?? "shared"}:${freshness.epoch}`; return { + sessionIdentity, config: { ...input.config, env, - managedAiConnection: { ...selection.attribution, identity }, + managedAiConnection: { ...selection.attribution, identity, sessionIdentity }, }, attribution: selection.attribution, accountName: selection.connection.name, @@ -367,7 +382,7 @@ export async function prepareManagedAiRuntime( if (decision !== 10) return; await secretService(tx).rotate( ref.secretId, - { value: refreshed }, + { value: refreshed, preserveAiSessionEpoch: true }, { userId: grant.subjectUserId }, ); await tx diff --git a/server/src/services/execution-recovery-attempt.test.ts b/server/src/services/execution-recovery-attempt.test.ts index 2673e2b33c3..c434e7b60a2 100644 --- a/server/src/services/execution-recovery-attempt.test.ts +++ b/server/src/services/execution-recovery-attempt.test.ts @@ -27,7 +27,7 @@ describe("failure attempts across resource waits", () => { }); describe("persisted independent accounting", () => { - it.each(["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => { + it.each(["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => { expect(executionFailureRetryCount({ scheduledRetryReason, scheduledRetryAttempt: 20, contextSnapshot: { executionRetryAccounting: { version: 1, failureRetries: 2, maxTurnContinuations: 1 } }, })).toBe(2); diff --git a/server/src/services/execution-recovery-attempt.ts b/server/src/services/execution-recovery-attempt.ts index 689137d9d99..dec0d847cf7 100644 --- a/server/src/services/execution-recovery-attempt.ts +++ b/server/src/services/execution-recovery-attempt.ts @@ -28,7 +28,7 @@ function savedAccounting(run: RetryRun): ExecutionRetryAccounting | null { function historicalFailureCount(run: RetryRun): number { if (run.scheduledRetryReason === "max_turns_continuation" || run.scheduledRetryReason === "issue_disposition_repair") return 0; - if (run.scheduledRetryReason === "ai_connection_busy") { + if (run.scheduledRetryReason === "ai_connection_busy" || run.scheduledRetryReason === "ai_connection_pool_wait") { const saved = count(run.contextSnapshot?.failureRetriesBeforeAiConnectionWait); if (saved !== null) return saved; } @@ -42,7 +42,7 @@ function historicalFailureCount(run: RetryRun): number { export function executionRetryAccounting(run: RetryRun): ExecutionRetryAccounting { const saved = savedAccounting(run); - const nonFailureLane = ["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy"].includes(run.scheduledRetryReason ?? ""); + const nonFailureLane = ["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"].includes(run.scheduledRetryReason ?? ""); return { version: 1, failureRetries: Math.max(saved?.failureRetries ?? 0, saved && nonFailureLane ? 0 : historicalFailureCount(run)), @@ -57,7 +57,7 @@ export function executionFailureRetryCount(run: RetryRun): number { } export function executionRetryAttemptCount(run: RetryRun, reason: string): number { - if (reason === "workspace_busy" || reason === "ai_connection_busy") { + if (reason === "workspace_busy" || reason === "ai_connection_busy" || reason === "ai_connection_pool_wait") { return run.scheduledRetryReason === reason ? count(run.scheduledRetryAttempt) ?? 0 : 0; } const accounting = executionRetryAccounting(run); @@ -67,6 +67,6 @@ export function executionRetryAttemptCount(run: RetryRun, reason: string): numbe export function accountingForScheduledRetry(run: RetryRun, reason: string, attempt: number): ExecutionRetryAccounting { const accounting = executionRetryAccounting(run); if (reason === "max_turns_continuation") accounting.maxTurnContinuations = attempt; - else if (reason !== "workspace_busy" && reason !== "ai_connection_busy") accounting.failureRetries = attempt; + else if (reason !== "workspace_busy" && reason !== "ai_connection_busy" && reason !== "ai_connection_pool_wait") accounting.failureRetries = attempt; return accounting; } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f2abc881ff1..63dd48607ee 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -34,7 +34,8 @@ import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAs import { admitExplicitNativeContinuation, undeliveredLegacyUserCommentIds } from "./explicit-native-continuation.js"; import { connectionIntentService } from "./connection-intents.js"; import { managedAiSessionFingerprintConfig, prepareManagedAiRuntime, assertManagedAiProjectAuth, stripAiAuthBindings, isAiConnectionBusy, AI_AUTH_ENV_KEYS } from "./ai-connection-runtime.js"; -import { aiConnectionBindingSchema } from "@paperclipai/shared"; +import { aiConnectionBindingSchema, aiRuntimeConnectionBindingSchema, type AiConnectionRouterSelection } from "@paperclipai/shared"; +import { aiConnectionRouterService, AiConnectionPoolExhausted, applyAiConnectionRouterTaskSettings } from "./ai-connection-router.js"; import { executionBlockerPredicate, getExecutionBlocker } from "./execution-blocker.js"; import { CONVERSATION_CONTINUATION_POLICY, claimedAdapterType, runUsedConversationAdapter, hasConversationContinuationPolicy, isConversationAdapter } from "./conversation-continuation.js"; import { recordExecutionWait } from "./execution-wait.js"; @@ -535,6 +536,7 @@ import { MAX_TURN_CONTINUATION_RETRY_REASON, WORKSPACE_BUSY_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON, + AI_CONNECTION_POOL_WAIT_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_WAKE_REASON, WAKE_COMMENT_IDS_KEY, @@ -15542,7 +15544,7 @@ export function heartbeatService( const requiresIssueGate = isTransientWorkspaceGitScanCode(run.errorCode) || hasConversationContinuationPolicy(run.resultJson) || - retryReason === AI_CONNECTION_BUSY_RETRY_REASON || + (retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) || retryReason === MAX_TURN_CONTINUATION_RETRY_REASON || retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON; if (requiresIssueGate) { @@ -15612,7 +15614,7 @@ export function heartbeatService( executionFailureRetryCount(run), } : {}), - ...(retryReason === AI_CONNECTION_BUSY_RETRY_REASON + ...((retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) ? { failureRetriesBeforeAiConnectionWait: executionFailureRetryCount(run) } : {}), ...(shouldQuarantineWorkspaceForRetry @@ -15938,7 +15940,7 @@ export function heartbeatService( } if ( - retryReason === AI_CONNECTION_BUSY_RETRY_REASON && issueId && + (retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) && issueId && !isNonAssigneeWorkspaceBusyRetry(retryReason, contextSnapshot) ) { // The issue row is locked above. Recheck after the preflight gate so @@ -20433,7 +20435,7 @@ export function heartbeatService( let readFailureReportSecrets: () => string[] = () => []; try { - const agent = await getAgent(run.agentId); + let agent = await getAgent(run.agentId); if (!agent) { await setRunStatus(runId, "failed", { error: "Agent not found", @@ -20684,7 +20686,7 @@ export function heartbeatService( : row; }) : null; - const issueAssigneeOverrides = + let issueAssigneeOverrides = issueContext && issueContext.assigneeAgentId === agent.id ? parseIssueAssigneeAdapterOverrides( issueContext.assigneeAdapterOverrides, @@ -20890,7 +20892,7 @@ export function heartbeatService( // durable for authorization and operation-time credential resolution. context.executionPolicy = retainedTrust.executionPolicy; } - const config = parseObject(agent.adapterConfig); + let config = parseObject(agent.adapterConfig); const taskSession = taskKey ? await getTaskSession( agent.companyId, @@ -20899,6 +20901,46 @@ export function heartbeatService( taskKey, ) : null; + const routerHasPersistedInput = Object.keys(parseObject(parseObject(run.runnerProfileJson).nativeExecutionInput)).length > 0; + const persistedRouterPoolId = routerHasPersistedInput ? readNonEmptyString(parseObject(context.aiRouterSelection).poolId) : null; + const requestedAiBinding = persistedRouterPoolId ? { mode: "router" as const, connectionId: persistedRouterPoolId } : agent.runtimeConfig?.aiConnection ? aiRuntimeConnectionBindingSchema.parse(agent.runtimeConfig.aiConnection) : undefined; + let aiBinding = requestedAiBinding?.mode === "router" ? undefined : requestedAiBinding; + if (requestedAiBinding?.mode === "router") { + try { + context.aiRouterTaskKey = readNonEmptyString(context.aiRouterTaskKey) ?? taskKey ?? run.id; + const savedIdentity = taskSession?.sessionParamsJson?.paperclipAiCredentialIdentity; + const selection = await aiConnectionRouterService(db, options.pluginWorkerManager).resolve({ + companyId: agent.companyId, poolId: requestedAiBinding.connectionId, agentId: agent.id, + userId: responsibleUserId, adapterType: agent.adapterType, taskKey: String(context.aiRouterTaskKey), + overrides: issueAssigneeOverrides?.adapterConfig ?? {}, + existingGrantId: typeof savedIdentity === "string" ? savedIdentity.split(":")[0] : undefined, + requireExisting: Boolean(taskSession?.sessionDisplayId) && !shouldResetTaskSessionForWake(context), + persisted: routerHasPersistedInput ? context.aiRouterSelection as AiConnectionRouterSelection | undefined : undefined, + }); + aiBinding = selection.binding; + // Provider-specific fields from the configured harness must not leak into the selected member. + config = applyAiConnectionRouterTaskSettings(config, selection); + if (issueAssigneeOverrides) issueAssigneeOverrides = { ...issueAssigneeOverrides, adapterConfig: applyAiConnectionRouterTaskSettings(issueAssigneeOverrides.adapterConfig ?? {}, selection) }; + agent = { ...agent, adapterConfig: config }; + context.aiRouterSelection = selection; + await db.update(heartbeatRuns).set({ contextSnapshot: sql`coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) || ${JSON.stringify({ aiRouterTaskKey: context.aiRouterTaskKey, aiRouterSelection: selection })}::jsonb` }).where(eq(heartbeatRuns.id, run.id)); + await appendRunEvent(run, { eventType: "lifecycle", stream: "system", level: "info", message: "Using task-pinned pool account", payload: { poolId: selection.poolId, memberId: selection.memberId, provider: aiBinding.provider, model: config.model, notes: selection.notes } }); + } catch (error) { + if (error instanceof AiConnectionPoolExhausted && !routerHasPersistedInput) { + const now = new Date(); + context.aiConnectionBusyDeferredWhileAssignee = issueContext?.assigneeAgentId === agent.id; + const cancelled = await setRunStatusIfRunning(run.id, "cancelled", { error: error.message, errorCode: error.code, finishedAt: now, resultJson: { executionRecovery: { kind: "ai_connection_wait", providerWorkStarted: false }, retryAt: error.retryAt }, contextSnapshot: context }); + if (cancelled.updated) { + await setWakeupStatus(run.wakeupRequestId, "cancelled", { finishedAt: now, error: error.message }); + const retry = await scheduleBoundedRetryForRun(cancelled.run ?? run, agent, { now, retryReason: AI_CONNECTION_POOL_WAIT_RETRY_REASON, wakeReason: "ai_connection_pool_retry", maxAttempts: (run.scheduledRetryAttempt ?? 0) + 1, delayMs: Math.max(1000, Date.parse(error.retryAt) - Date.now()) }); + if (retry.outcome !== "scheduled") await releaseIssueExecutionAndPromote(cancelled.run ?? run); + await finalizeAgentStatus(run.agentId, "cancelled"); + } + return; + } + throw new ConfigurationIncompleteFailure(error instanceof Error ? error.message : "Configure this connection pool", { configurationIncomplete: { reason: "ai_connection_unavailable", companyId: agent.companyId, agentId: agent.id, responsibleUserId, actionUrl: `/agents/${agent.id}/runtime`, fingerprint: `ai-router:${requestedAiBinding.connectionId}` } }); + } + } if (isConversation(issueContext)) { delete context.resumeSessionParams; delete context.resumeSessionDisplayId; @@ -21569,7 +21611,8 @@ export function heartbeatService( }); const mergedConfig = { ...workspaceManagedConfig, - ...(issueAssigneeOverrides?.adapterConfig ?? {}), + ...Object.fromEntries(Object.entries(issueAssigneeOverrides?.adapterConfig ?? {}).filter(([key]) => requestedAiBinding?.mode !== "router" || !["provider", "acpxAgent", "model", "modelReasoningEffort", "reasoningEffort", "effort", "variant"].includes(key))), + ...(requestedAiBinding?.mode === "router" ? parseObject(parseObject(context.aiRouterSelection).runtimeConfig) : {}), // The base below is already task-owned. Keep directory transport while // preserving isolated mode and the mandatory sandbox preflight. ...(useIsolatedTaskDirectory ? { workspaceStrategy: { type: "project_primary" } } : {}), @@ -21606,7 +21649,7 @@ export function heartbeatService( ["local", "ssh"].includes( selectedEnvironmentForConfig?.driver ?? "local", ); - const aiBinding = agent.runtimeConfig?.aiConnection ? aiConnectionBindingSchema.parse(agent.runtimeConfig.aiConnection) : undefined; + const { resolvedConfig, secretKeys, secretManifest } = await resolveExecutionRunAdapterConfig({ managedAiCredentials: Boolean(aiBinding), @@ -21656,13 +21699,16 @@ export function heartbeatService( fingerprint: `ai:${agent.id}:${responsibleUserId}:${JSON.stringify(aiBinding)}` }, }); } - if (persistedNativeExecutionInput && parseObject(run.contextSnapshot?.aiConnection).identity !== managedAiRuntime.identity) { + const savedAiAccount = parseObject(run.contextSnapshot?.aiConnection); + if (persistedNativeExecutionInput && (savedAiAccount.sessionIdentity + ? savedAiAccount.sessionIdentity !== managedAiRuntime.sessionIdentity + : savedAiAccount.identity !== managedAiRuntime.identity)) { throw new ConfigurationIncompleteFailure("The AI account changed while this native run was suspended. Start a new execution.", { configurationIncomplete: { reason: "ai_connection_changed", actionUrl: `/agents/${agent.id}/runtime` } }); } Object.assign(resolvedConfig, managedAiRuntime.config); for (const key of AI_AUTH_ENV_KEYS) secretKeys.add(key); - context.aiConnection = { ...managedAiRuntime.attribution, identity: managedAiRuntime.identity }; - await db.update(heartbeatRuns).set({ contextSnapshot: sql`coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) || ${JSON.stringify({ aiConnection: context.aiConnection })}::jsonb` }).where(eq(heartbeatRuns.id, run.id)); + context.aiConnection = { ...managedAiRuntime.attribution, accountName: managedAiRuntime.accountName, identity: managedAiRuntime.identity, sessionIdentity: managedAiRuntime.sessionIdentity }; + await db.update(heartbeatRuns).set({ contextSnapshot: sql`coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) || ${JSON.stringify({ aiConnection: context.aiConnection, ...(context.aiRouterTaskKey ? { aiRouterTaskKey: context.aiRouterTaskKey } : {}), ...(context.aiRouterSelection ? { aiRouterSelection: context.aiRouterSelection } : {}) })}::jsonb` }).where(eq(heartbeatRuns.id, run.id)); } if (secretManifest.length > 0) { context.paperclipSecrets = { @@ -23044,10 +23090,10 @@ export function heartbeatService( const taskSessionCredentialCompatible = isTaskSessionCredentialCompatible( taskSession?.sessionParamsJson, - managedAiRuntime?.identity, - ); + managedAiRuntime?.sessionIdentity, + ) || Boolean(managedAiRuntime && isTaskSessionCredentialCompatible(taskSession?.sessionParamsJson, managedAiRuntime.identity)); if (managedAiRuntime) { - sessionConfigMetadata.aiCredentialIdentity = managedAiRuntime.identity; + sessionConfigMetadata.aiCredentialIdentity = managedAiRuntime.sessionIdentity; if (!taskSessionCredentialCompatible) { runtimeSessionIdForAdapter = null; runtimeSessionParamsForAdapter = null; diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 2e7d555f0c2..9f855d123f5 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -223,6 +223,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta return { enableEnvironments: parsed.data.enableEnvironments ?? false, enableNativeRunner: parsed.data.enableNativeRunner ?? true, + enableAiConnectionRouters: parsed.data.enableAiConnectionRouters ?? false, enableManagedSandboxOnly: parsed.data.enableManagedSandboxOnly ?? false, enableIsolatedWorkspaces: parsed.data.enableIsolatedWorkspaces ?? false, enableIsolatedWorkspacesByDefault: parsed.data.enableIsolatedWorkspacesByDefault ?? false, @@ -268,6 +269,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta return { enableEnvironments: false, enableNativeRunner: true, + enableAiConnectionRouters: false, enableManagedSandboxOnly: false, enableIsolatedWorkspaces: false, enableIsolatedWorkspacesByDefault: false, diff --git a/server/src/services/secrets.ts b/server/src/services/secrets.ts index 143275d0941..6549cf69e08 100644 --- a/server/src/services/secrets.ts +++ b/server/src/services/secrets.ts @@ -2397,6 +2397,8 @@ export function secretService(db: Db | DbTransaction) { // update matches the latest version, so a concurrent rotation between the // read and the write cannot pass. A mismatch throws a 409 conflict. expectedLatestVersion?: number; + /** Internal authenticated refresh write-back only; never accepted by HTTP schemas. */ + preserveAiSessionEpoch?: boolean; }, actor?: { userId?: string | null; agentId?: string | null }, ) { @@ -2534,6 +2536,7 @@ export function secretService(db: Db | DbTransaction) { .update(companySecrets) .set({ latestVersion: nextVersion, + aiSessionEpoch: input.preserveAiSessionEpoch ? secret.aiSessionEpoch : sql`${companySecrets.aiSessionEpoch} + 1`, externalRef: prepared.externalRef, providerConfigId, lastRotatedAt: new Date(), @@ -4589,6 +4592,8 @@ export function secretService(db: Db | DbTransaction) { providerVersionRef?: string | null; providerConfigId?: string | null; expectedLatestVersion?: number; + /** Internal authenticated refresh write-back only; never accepted by HTTP schemas. */ + preserveAiSessionEpoch?: boolean; }, actor?: { userId?: string | null; agentId?: string | null }, ) => { diff --git a/ui/src/api/agents.ts b/ui/src/api/agents.ts index d1709a68007..7bd9cce9855 100644 --- a/ui/src/api/agents.ts +++ b/ui/src/api/agents.ts @@ -226,11 +226,12 @@ export const agentsApi = { adapterModels: ( companyId: string, type: string, - options?: { refresh?: boolean; environmentId?: string | null; provider?: string }, + options?: { refresh?: boolean; environmentId?: string | null; provider?: string; poolId?: string }, ) => { const params = new URLSearchParams(); if (options?.refresh) params.set("refresh", "1"); if (options?.provider) params.set("provider", options.provider); + if (options?.poolId) params.set("poolId", options.poolId); if (options?.environmentId) params.set("environmentId", options.environmentId); const query = params.size > 0 ? `?${params.toString()}` : ""; return api.get( @@ -246,7 +247,7 @@ export const agentsApi = { type: string, data: { adapterConfig: Record; - aiConnection?: import("@paperclipai/shared").AiConnectionBinding; + aiConnection?: import("@paperclipai/shared").AiRuntimeConnectionBinding; agentId?: string; testCredentials?: Record; environmentId?: string | null; diff --git a/ui/src/components/AgentConfigForm.tsx b/ui/src/components/AgentConfigForm.tsx index 1c5fa70d8f3..794ede1d714 100644 --- a/ui/src/components/AgentConfigForm.tsx +++ b/ui/src/components/AgentConfigForm.tsx @@ -1,5 +1,5 @@ import { AiConnectionField } from "./ai-connections/AiConnectionField"; -import { aiConnectionBindingSchema } from "@paperclipai/shared"; +import { aiConnectionBindingSchema, aiRuntimeConnectionBindingSchema } from "@paperclipai/shared"; import { testAgentSetup } from "@/lib/test-agent-setup"; import { setupEfforts } from "../lib/agent-setup-fields"; import { RuntimeTestCard } from "./RuntimeTestCard"; @@ -1061,7 +1061,7 @@ export function AgentConfigForm(props: AgentConfigFormProps) { }); const adapterConfig = buildAdapterConfigForTest(adapterConfigPatch); const agentId = isCreate ? undefined : props.agent.id; - const aiConnection = isCreate ? undefined : aiConnectionBindingSchema.safeParse( + const aiConnection = isCreate ? undefined : aiRuntimeConnectionBindingSchema.safeParse( (overlay.runtime.runtimeConfig as Record | undefined)?.aiConnection ?? props.agent.runtimeConfig.aiConnection, ).data; if (props.compactTestFeedback) { @@ -1659,7 +1659,7 @@ export function AgentConfigForm(props: AgentConfigFormProps) { )} {!isCreate && selectedCompanyId && | undefined)?.aiConnection ?? runtimeConfig.aiConnection).data} + routerAdapterType={adapterType} value={aiRuntimeConnectionBindingSchema.safeParse((overlay.runtime.runtimeConfig as Record | undefined)?.aiConnection ?? runtimeConfig.aiConnection).data} model={String(eff("adapterConfig", "model", config.model) ?? "")} environmentId={currentDefaultEnvironmentId || undefined} legacy onChange={binding => mark("runtime", "runtimeConfig", { ...runtimeConfig, aiConnection: binding })} />} diff --git a/ui/src/components/ai-connections/AiConnectionField.tsx b/ui/src/components/ai-connections/AiConnectionField.tsx index adf57ec293a..9754e2b5578 100644 --- a/ui/src/components/ai-connections/AiConnectionField.tsx +++ b/ui/src/components/ai-connections/AiConnectionField.tsx @@ -4,6 +4,7 @@ import { aiConnectionBindingSchema, isAiConnectionCompatible, type AiConnectionBinding, + type AiRuntimeConnectionBinding, type AiAuthMethod, type AiProvider, type AiManagedConnectionSummary, @@ -46,28 +47,30 @@ export function AiConnectionField({ environmentId, legacy = false, readOnly = false, + routerAdapterType, }: { companyId: string; agentId?: string; agentName: string; adapterType: string; model?: string; - value?: AiConnectionBinding; - onChange: (binding: AiConnectionBinding) => void; + value?: AiRuntimeConnectionBinding; + onChange: (binding: AiRuntimeConnectionBinding) => void; environmentId?: string; legacy?: boolean; readOnly?: boolean; + routerAdapterType?: string; }) { const provider = aiProviderForAdapter(adapterType); const returnFocus = useRef(null); const restoreFocus = (event: Event) => { event.preventDefault(); returnFocus.current?.focus(); }; const [adopting, setAdopting] = useState(false); - const [pendingAdoption, setPendingAdoption] = useState(); + const [pendingAdoption, setPendingAdoption] = useState(); const [connecting, setConnecting] = useState(false); const [reconnecting, setReconnecting] = useState(); const [allAgents, setAllAgents] = useState(true); const [savedAccount, setSavedAccount] = useState<{ connectionId: string; grantId: string; method: AiAuthMethod }>(); - const changeBinding = (next: AiConnectionBinding) => { + const changeBinding = (next: AiRuntimeConnectionBinding) => { if (legacy && !value) { if (!connecting) returnFocus.current = document.activeElement as HTMLElement; setPendingAdoption(next); } else onChange(next); }; @@ -99,9 +102,10 @@ export function AiConnectionField({ selectDefault.reset(); setConnecting(true); }; - const method: AiAuthMethod = (value?.mode !== "responsible_user" ? value?.method : undefined) + const method: AiAuthMethod = (value && value.mode !== "router" && value.mode !== "responsible_user" ? value.method : undefined) ?? accounts.data?.connections.find((account) => account.provider === provider && account.isDefault)?.method ?? (provider === "openrouter" ? "api_key" : "subscription"); + const compatiblePools = agentId ? accounts.data?.pools?.filter(pool => pool.enabled && pool.members.some(member => isAiConnectionCompatible(member.binding, routerAdapterType ?? adapterType, member.profile.model, member.profile.provider, member.profile.acpxAgent))) ?? [] : []; if (!provider) return null; if (legacy && !value && !adopting) return ( @@ -112,13 +116,25 @@ export function AiConnectionField({ ); return (
- {value && (adapterType !== "opencode_local" || Boolean(model)) && !isAiConnectionCompatible(value, adapterType, model) && ( + {value && value.mode !== "router" && (adapterType !== "opencode_local" || Boolean(model)) && !isAiConnectionCompatible(value, adapterType, model) && (

This connection does not support the current harness and model. Choose a compatible connection before saving.

)} - 0 || value?.mode === "router") && } + {value?.mode !== "router" && openConnection()} onReconnect={(!value || value.mode === "responsible_user") && personalDefault && personalDefault.status !== "connected" ? () => openConnection(personalDefault) : undefined} onRetry={() => void accounts.refetch()} - /> + />} { @@ -145,15 +161,13 @@ export function AiConnectionField({ Adopt Connections for {agentName} - Saving tests this account in {agentName}’s environment before - replacing its existing authentication. Other agents keep their - current configuration. + Saving validates access to the selected connection. Existing sessions keep their account or require an explicit reset before adoption.

{pendingAdoption?.mode === "responsible_user" ? `Responsible user’s default. For you: ${accounts.data?.connections.find((account) => account.isDefault && account.provider === provider)?.name ?? "Not connected"}. Other users use their own default.` - : accounts.data?.connections.find( + : pendingAdoption?.mode === "router" ? `Task-pinned connection pool. Existing sessions must use a pool member or be reset.` : accounts.data?.connections.find( (account) => account.id === pendingAdoption?.connectionId, )?.name}

diff --git a/ui/src/components/ai-connections/AiConnectionPoolRunDetails.tsx b/ui/src/components/ai-connections/AiConnectionPoolRunDetails.tsx new file mode 100644 index 00000000000..34e75fe2ae0 --- /dev/null +++ b/ui/src/components/ai-connections/AiConnectionPoolRunDetails.tsx @@ -0,0 +1,14 @@ +import type { AiConnectionRouterSelection } from "@paperclipai/shared"; + +export function AiConnectionPoolRunDetails({ context }: { context: Record | null }) { + const selection = context?.aiRouterSelection as AiConnectionRouterSelection | undefined; + if (!selection) return null; + const account = context?.aiConnection as { accountName?: string; connectionId?: string } | undefined; + const config = selection.runtimeConfig; + const effort = config.modelReasoningEffort ?? config.reasoningEffort ?? config.effort ?? config.variant; + return
+

Task-pinned pool account · {account?.accountName ?? account?.connectionId ?? "Selected account"}

+

{[selection.binding.provider, config.provider, config.acpxAgent, config.model, effort].filter(Boolean).map(String).join(" · ")}

+ {selection.notes.length > 0 &&

{selection.notes.join(" ")}

} +
; +} diff --git a/ui/src/components/new-agent/NewAgentSetup.tsx b/ui/src/components/new-agent/NewAgentSetup.tsx index dcb167010b4..d81361da140 100644 --- a/ui/src/components/new-agent/NewAgentSetup.tsx +++ b/ui/src/components/new-agent/NewAgentSetup.tsx @@ -845,7 +845,7 @@ function Setup({
) : ( { setRuntimeAiBinding(binding); resetTest(); }} /> + onChange={binding => { binding.mode !== "router" && setRuntimeAiBinding(binding); resetTest(); }} /> ) )} {models.error &&

Could not load models. Retry or enter a model ID manually.

} diff --git a/ui/src/components/task-chat/ComposerRunSettingsPicker.tsx b/ui/src/components/task-chat/ComposerRunSettingsPicker.tsx index f3b6025126a..7572c7d37fb 100644 --- a/ui/src/components/task-chat/ComposerRunSettingsPicker.tsx +++ b/ui/src/components/task-chat/ComposerRunSettingsPicker.tsx @@ -1,7 +1,7 @@ import { useEffect, useLayoutEffect, useRef, useState, type CSSProperties, type ReactNode, type Ref } from "react"; import { useQuery } from "@tanstack/react-query"; import { ArrowLeft, Check, ChevronDown, Plus, RotateCcw, Search, X, Zap } from "lucide-react"; -import type { Agent, IssueAssigneeAdapterOverrides } from "@paperclipai/shared"; +import { aiRuntimeConnectionBindingSchema, type Agent, type IssueAssigneeAdapterOverrides } from "@paperclipai/shared"; import { agentsApi, type AdapterModel } from "@/api/agents"; import { queryKeys } from "@/lib/queryKeys"; import { cn } from "@/lib/utils"; @@ -51,6 +51,8 @@ function harnessLabel(agent: Agent | undefined): string { if (!agent) return "Choose an agent"; const harness = HARNESS_LABELS[agent.adapterType] ?? agent.adapterType; const provider = composerCatalogProvider(agent); + const binding = aiRuntimeConnectionBindingSchema.safeParse(agent?.runtimeConfig?.aiConnection).data; + const poolId = binding?.mode === "router" ? binding.connectionId : undefined; return provider === "openrouter" ? `${harness} · OpenRouter` : harness; } @@ -99,12 +101,14 @@ export function ComposerRunSettingsPicker({ const agent = agents.get(agentId); const modelSupported = supportsComposerModel(agent); const provider = composerCatalogProvider(agent); + const binding = aiRuntimeConnectionBindingSchema.safeParse(agent?.runtimeConfig?.aiConnection).data; + const poolId = binding?.mode === "router" ? binding.connectionId : undefined; const { data: fetchedModels = [], isPending: modelsPending } = useQuery({ queryKey: agent && modelSupported - ? queryKeys.agents.adapterModels(companyId, agent.adapterType, agent.defaultEnvironmentId ?? null, provider) + ? [...queryKeys.agents.adapterModels(companyId, agent.adapterType, agent.defaultEnvironmentId ?? null, provider), poolId ?? null] : ["agents", "composer-models", "none"], queryFn: () => agentsApi.adapterModels(companyId, agent!.adapterType, { - environmentId: agent!.defaultEnvironmentId ?? null, provider, + environmentId: agent!.defaultEnvironmentId ?? null, provider, poolId, }), enabled: Boolean(agent && modelSupported && !modelOptionsOverride), }); diff --git a/ui/src/components/task-chat/composer-run-settings.ts b/ui/src/components/task-chat/composer-run-settings.ts index 8b5ad5cc915..db0ca63ad67 100644 --- a/ui/src/components/task-chat/composer-run-settings.ts +++ b/ui/src/components/task-chat/composer-run-settings.ts @@ -2,7 +2,7 @@ import { codexLocalReasoningEffortsForModel, isCodexLocalFastModeSupported, isCo import { claudeLocalReasoningEffortsForModel } from "@paperclipai/adapter-claude-local"; import { DEFAULT_GROK_LOCAL_MODEL, grokLocalReasoningEffortsForModel } from "@paperclipai/adapter-grok-local"; import { DEFAULT_KIMI_LOCAL_MODEL, modelSupportsEffort, KIMI_SUPPORTED_EFFORTS } from "@paperclipai/adapter-kimi-local"; -import { aiConnectionBindingSchema, type Agent, type IssueAssigneeAdapterOverrides } from "@paperclipai/shared"; +import { aiConnectionBindingSchema, aiRuntimeConnectionBindingSchema, type Agent, type IssueAssigneeAdapterOverrides } from "@paperclipai/shared"; export interface ComposerRunSettings { model: string | null; @@ -40,6 +40,8 @@ export function composerEfforts(agent: Agent | undefined, model: string, catalog const effectiveModel = model.trim() || (agent.adapterType === "grok_local" ? DEFAULT_GROK_LOCAL_MODEL : agent.adapterType === "kimi_local" ? DEFAULT_KIMI_LOCAL_MODEL : ""); if (!effectiveModel) return []; + const pool = aiRuntimeConnectionBindingSchema.safeParse(agent.runtimeConfig?.aiConnection).data?.mode === "router"; + if (pool && agent.adapterType === "paperclip_runner") return isCodexLocalKnownModel(effectiveModel) ? codexLocalReasoningEffortsForModel(effectiveModel) : []; if (agent.adapterType === "codex_local" || (agent.adapterType === "paperclip_runner" && composerCatalogProvider(agent) === "codex")) { return isCodexLocalKnownModel(effectiveModel) ? codexLocalReasoningEffortsForModel(effectiveModel) : []; } diff --git a/ui/src/lib/runRetryState.ts b/ui/src/lib/runRetryState.ts index 1cc33db1891..ce332b48e9b 100644 --- a/ui/src/lib/runRetryState.ts +++ b/ui/src/lib/runRetryState.ts @@ -19,6 +19,7 @@ export type RunRetryStateSummary = { }; const RETRY_REASON_LABELS: Record = { + ai_connection_pool_wait: "Waiting for pool usage to recover", transient_failure: "Transient failure", missing_issue_comment: "Missing task comment", process_lost: "Process lost", @@ -65,7 +66,7 @@ export function describeRunRetryState(run: RetryAwareRun): RunRetryStateSummary if (run.status === "scheduled_retry") { return { kind: "scheduled", - badgeLabel: isMaxTurnContinuation ? "Continuation scheduled" : "Retry scheduled", + badgeLabel: run.scheduledRetryReason === "ai_connection_pool_wait" ? "Pool exhausted" : isMaxTurnContinuation ? "Continuation scheduled" : "Retry scheduled", tone: "border-blue-500/30 bg-blue-500/10 text-blue-700 dark:text-blue-300", detail: joinFragments([attemptLabel, reasonLabel]), secondary: dueAt diff --git a/ui/src/lib/test-agent-setup.test.ts b/ui/src/lib/test-agent-setup.test.ts index 9154d951ed6..514b8516e0c 100644 --- a/ui/src/lib/test-agent-setup.test.ts +++ b/ui/src/lib/test-agent-setup.test.ts @@ -27,6 +27,12 @@ const ready = { checks: [{ code: "runtime", level: "info", message: "Ready" }], }; beforeEach(() => testEnvironment.mockReset()); +it("does not launch an ambient provider hello test for a pool preview", async () => { + testEnvironment.mockResolvedValue({ ...ready, status: "warn", checks: [{ code: "ai_connection_pool_task_test_required", level: "warn", message: "Run a task" }] }); + await testAgentSetup({ ...input, aiConnection: { mode: "router", connectionId: "pool-id" } }); + expect(testEnvironment).toHaveBeenCalledTimes(1); + expect(testEnvironment.mock.calls[0]?.[2].aiConnection).toEqual({ mode: "router", connectionId: "pool-id" }); +}); it("does not report a connection when runtime readiness passes but provider authentication fails", async () => { testEnvironment .mockResolvedValueOnce(ready) diff --git a/ui/src/lib/test-agent-setup.ts b/ui/src/lib/test-agent-setup.ts index e56cd8bbcb9..8e49fc51f30 100644 --- a/ui/src/lib/test-agent-setup.ts +++ b/ui/src/lib/test-agent-setup.ts @@ -10,7 +10,7 @@ export async function testAgentSetup(input: { adapterType: string; providerAdapter: string; adapterConfig: Record; - aiConnection?: import("@paperclipai/shared").AiConnectionBinding; + aiConnection?: import("@paperclipai/shared").AiRuntimeConnectionBinding; testCredentials?: Record; environmentId: string | null; }): Promise { @@ -27,6 +27,7 @@ export async function testAgentSetup(input: { payload, ); if ( + input.aiConnection?.mode === "router" || runtime.status === "fail" || runtime.checks.some( (check) => check.code === ADAPTER_AUTH_MISSING_CHECK_CODE, diff --git a/ui/src/pages/AgentDetail.tsx b/ui/src/pages/AgentDetail.tsx index 939e482e9eb..ebb1159eeb6 100644 --- a/ui/src/pages/AgentDetail.tsx +++ b/ui/src/pages/AgentDetail.tsx @@ -1,3 +1,4 @@ +import { AiConnectionPoolRunDetails } from "@/components/ai-connections/AiConnectionPoolRunDetails"; import type { AgentInstructionCandidate, AgentInstructionsBundle } from "@paperclipai/shared"; import { InstructionHistory } from "../components/InstructionHistory"; import { AgentCharacter } from "../components/AgentCharacter"; @@ -3576,6 +3577,7 @@ function RunDetail({ run: initialRun, agentRouteId, adapterType, adapterConfig } git workspace it could not validate, wired to the same reconcile / repair / re-issue / break-glass handlers as the task detail page. */} + {/* Run summary card */}
diff --git a/ui/src/pages/InstanceExperimentalSettings.test.tsx b/ui/src/pages/InstanceExperimentalSettings.test.tsx index 8e7d7d9654a..c574bd2738d 100644 --- a/ui/src/pages/InstanceExperimentalSettings.test.tsx +++ b/ui/src/pages/InstanceExperimentalSettings.test.tsx @@ -72,6 +72,7 @@ function defaultExperimentalSettings(): InstanceExperimentalSettingsPayload { return { enableEnvironments: false, enableNativeRunner: false, + enableAiConnectionRouters: false, enableManagedSandboxOnly: false, enableIsolatedWorkspaces: false, enableIsolatedWorkspacesByDefault: false, diff --git a/ui/src/pages/InstanceExperimentalSettings.tsx b/ui/src/pages/InstanceExperimentalSettings.tsx index 0d8bd95e366..a9e9e47db1e 100644 --- a/ui/src/pages/InstanceExperimentalSettings.tsx +++ b/ui/src/pages/InstanceExperimentalSettings.tsx @@ -444,6 +444,17 @@ export function InstanceExperimentalSettings() { ariaLabel="Toggle memory connectors experimental setting" /> + toggleMutation.mutate({ enableAiConnectionRouters: checked })} + disabled={toggleMutation.isPending} + settingKey="enableAiConnectionRouters" + managed={managedKeys.enableAiConnectionRouters} + ariaLabel="Toggle AI connection routers experimental setting" + /> +