diff --git a/DEVLOG.md b/DEVLOG.md index 4586aa6e..50848fde 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -2,6 +2,35 @@ > Append-only session log. Read at session start. Update at session end. +## 2026-07-20 — AXI-134 legacy connection-key migration and dispatch repair + +Traced AXI-129's mention-triggered Victor runs to a dispatch-time contract +mismatch: the managed runtime retained its deterministic `ac_*` connection id, +while the new typed delivery-conflict request required a Prisma CUID. Request +validation rolled the parking transaction back, so the global dispatch sweep +retried until the run watchdog eventually marked each attempt stalled. + +Added an atomic key-normalization migration for every historical `ac_*` +AgentConnection. Existing foreign keys cascade the primary-key update across +runs, run events, Delivery sessions, and participants; typed conflict payloads +and dedupe keys are rewritten explicitly. Each normalized connection retains +its old id as an immutable `legacyId`, keeping append-only audit/event +history correlatable without rewriting it. Connection ids are now validated as +opaque bounded identifiers across conflict, join, and handoff contracts so a +rolling deployment remains compatible with pre-migration records. + +Delivery-conflict materialization is also isolated per candidate. If decision +creation fails again, Forge keeps ownership fail-closed, parks only that run, +records diagnostic evidence, and allows the rest of the dispatch sweep to +continue. + +Verification used only local Postgres, Redis, and synthetic data. A rollbacked +SQL fixture proved the migration across the immutable alias, runs, run events, +session ownership, participants, typed payloads, and dedupe keys. Focused +validation and real-database suites passed 34/34; the full serial Vitest suite passed 1,480 +tests with 12 intentional skips. Prisma validation, lint, and typecheck passed +with pre-existing warnings only. + ## 2026-07-19 — v0.29.1 release preparation Prepared the patch release containing AXI-131 and merged PR #74. Release diff --git a/docs/engineering/work-management.md b/docs/engineering/work-management.md index 6db9aa9d..ac5baed5 100644 --- a/docs/engineering/work-management.md +++ b/docs/engineering/work-management.md @@ -29,6 +29,12 @@ connection is primary; additional connections must explicitly join as a contributor or reviewer, or receive an audited handoff before changing the branch or advancing delivery state. +Agent connection ids are opaque stable identifiers. Callers and typed action +payloads must not assume they are Prisma CUIDs: historical managed-runtime and +webhook connections used deterministic `ac_*` ids. Data migrations may +normalize storage keys, but must preserve the immutable `legacyId` alias and update +relational plus actionable embedded references atomically. + Delivery provenance records separate facts: actor/profile, invocation source, connector/transport, and execution runtime. MCP client identity comes from the negotiated MCP `clientInfo`; missing identity is displayed as an unidentified diff --git a/prisma/migrations/20260720113000_normalize_agent_connection_ids/migration.sql b/prisma/migrations/20260720113000_normalize_agent_connection_ids/migration.sql new file mode 100644 index 00000000..31c1cdca --- /dev/null +++ b/prisma/migrations/20260720113000_normalize_agent_connection_ids/migration.sql @@ -0,0 +1,79 @@ +-- Normalize deterministic connection-backfill ids to the same CUID-shaped +-- primary keys Prisma creates for new endpoints. All relational references +-- were created with ON UPDATE CASCADE; the temporary map also updates the +-- actionable JSON/dedupe references that are not protected by foreign keys. + +CREATE TEMPORARY TABLE "_AgentConnectionIdMigration" ( + "oldId" TEXT PRIMARY KEY, + "newId" TEXT UNIQUE NOT NULL +) ON COMMIT DROP; + +ALTER TABLE "AgentConnection" ADD COLUMN "legacyId" TEXT; + +INSERT INTO "_AgentConnectionIdMigration" ("oldId", "newId") +SELECT + "id", + 'c' || substring(md5('forge-agent-connection-v1:' || "id") FROM 1 FOR 24) +FROM "AgentConnection" +WHERE "id" LIKE 'ac\_%' ESCAPE '\'; + +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM "_AgentConnectionIdMigration" migration + JOIN "AgentConnection" connection ON connection."id" = migration."newId" + WHERE connection."id" <> migration."oldId" + ) THEN + RAISE EXCEPTION 'Cannot normalize AgentConnection ids: generated id collision'; + END IF; +END $$; + +-- Retain a durable, immutable alias so historical audit/event JSON that names +-- the old connection can still be correlated without rewriting history or +-- relying on mutable heartbeat metadata. +UPDATE "AgentConnection" connection +SET "legacyId" = migration."oldId" +FROM "_AgentConnectionIdMigration" migration +WHERE connection."id" = migration."oldId"; + +CREATE UNIQUE INDEX "AgentConnection_legacyId_key" + ON "AgentConnection"("legacyId"); + +-- Typed delivery-conflict payloads hold live identifiers outside relational +-- columns. Rewrite both connection roles before changing the primary key. +UPDATE "ActionRequest" request +SET "payload" = jsonb_set( + request."payload", + '{expectedOwnerConnectionId}', + to_jsonb(migration."newId"), + false +) +FROM "_AgentConnectionIdMigration" migration +WHERE request."kind" = 'DELIVERY_CONNECTION_CONFLICT' + AND request."payload"->>'expectedOwnerConnectionId' = migration."oldId"; + +UPDATE "ActionRequest" request +SET "payload" = jsonb_set( + request."payload", + '{candidateConnectionId}', + to_jsonb(migration."newId"), + false +) +FROM "_AgentConnectionIdMigration" migration +WHERE request."kind" = 'DELIVERY_CONNECTION_CONFLICT' + AND request."payload"->>'candidateConnectionId' = migration."oldId"; + +-- Historical FREE_FORM conflict rows may contain the candidate id only in +-- their dedupe key. Updating all matching keys prevents the legacy alias from +-- splitting one logical conflict into two identities. +UPDATE "ActionRequest" request +SET "dedupeKey" = replace(request."dedupeKey", migration."oldId", migration."newId") +FROM "_AgentConnectionIdMigration" migration +WHERE request."dedupeKey" IS NOT NULL + AND strpos(request."dedupeKey", migration."oldId") > 0; + +UPDATE "AgentConnection" connection +SET "id" = migration."newId" +FROM "_AgentConnectionIdMigration" migration +WHERE connection."id" = migration."oldId"; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index f7d41290..89e9f27d 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -2669,6 +2669,9 @@ model ExternalResourceLink { /// execution (AgentRun/WorkSession) remain separate concerns. model AgentConnection { id String @id @default(cuid()) + /// Historical storage key retained after normalization so append-only audit + /// and event payloads can still be correlated without rewriting history. + legacyId String? @unique workspaceId String agentId String kind AgentConnectionKind diff --git a/src/server/routers/work-session.ts b/src/server/routers/work-session.ts index 16f78943..2f43a85b 100644 --- a/src/server/routers/work-session.ts +++ b/src/server/routers/work-session.ts @@ -10,6 +10,7 @@ import { touchWorkSession, } from "@/server/services/work-session"; import { handoffWorkSession, joinWorkSession } from "@/server/services/work-session-participant"; +import { agentConnectionIdSchema } from "@/server/validators"; const repoSchema = z .string() @@ -100,7 +101,7 @@ export const workSessionRouter = router({ .input( z.object({ sessionId: z.string().cuid(), - connectionId: z.string().cuid(), + connectionId: agentConnectionIdSchema, role: z.enum(["CONTRIBUTOR", "REVIEWER"]), }), ) @@ -119,7 +120,7 @@ export const workSessionRouter = router({ .input( z.object({ sessionId: z.string().cuid(), - targetConnectionId: z.string().cuid(), + targetConnectionId: agentConnectionIdSchema, reason: z.string().max(500).nullable().optional(), }), ) diff --git a/src/server/services/__tests__/run-dispatcher.test.ts b/src/server/services/__tests__/run-dispatcher.test.ts index 8d70d17b..b33fe279 100644 --- a/src/server/services/__tests__/run-dispatcher.test.ts +++ b/src/server/services/__tests__/run-dispatcher.test.ts @@ -4,6 +4,8 @@ import { AgentRunStatus, EngagementMode, EventKind, + AgentConnectionKind, + AgentConnectionLiveness, RuntimeKind, RunEngine, } from "@prisma/client"; @@ -396,6 +398,139 @@ describe("runs dispatcher", () => { } }); + it("parks a conflicting runtime that still has a legacy connection id", async () => { + const previousE2E = process.env.FORGE_E2E; + process.env.FORGE_E2E = "1"; + + try { + const fixture = await createWorkspaceFixture({ keyPrefix: "RLC" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const runtime = await prisma.runtime.create({ + data: { + workspaceId: fixture.workspace.id, + ownerId: fixture.user.id, + name: "legacy runtime", + kind: RuntimeKind.LOCAL_DAEMON, + adapterKey: "mock-runs", + providersAvailable: [AgentProvider.HERMES], + }, + }); + const runtimeAgent = await prisma.agent.create({ + data: { + workspaceId: fixture.workspace.id, + name: "legacy runner", + profileKey: "legacy-runner", + provider: AgentProvider.HERMES, + runEngine: RunEngine.RUNS, + runtimeId: runtime.id, + status: "ONLINE", + }, + }); + const ownerAgent = await prisma.agent.create({ + data: { + workspaceId: fixture.workspace.id, + name: "primary owner", + profileKey: "primary-owner", + provider: AgentProvider.CODEX, + }, + }); + const legacyConnection = await prisma.agentConnection.create({ + data: { + id: "ac_269a426f3d6cf83523c051e35f1adf93", + workspaceId: fixture.workspace.id, + agentId: runtimeAgent.id, + kind: AgentConnectionKind.MANAGED_RUNTIME, + livenessModel: AgentConnectionLiveness.HEARTBEAT, + runtimeId: runtime.id, + instanceKey: `runtime:${runtime.id}`, + }, + }); + const ownerConnection = await prisma.agentConnection.create({ + data: { + workspaceId: fixture.workspace.id, + agentId: ownerAgent.id, + kind: AgentConnectionKind.MCP_CLIENT, + livenessModel: AgentConnectionLiveness.LEASE, + instanceKey: "test-owner", + }, + }); + const issue = await createIssue(fixture, { title: "preserve primary delivery" }); + const session = await prisma.workSession.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + ownerUserId: fixture.user.id, + ownerAgentId: ownerAgent.id, + ownerConnectionId: ownerConnection.id, + source: "MCP", + status: "IN_PROGRESS", + repoFullName: "Codename-11/Forge", + branch: `codex/${issue.id}-primary`, + participants: { + create: { + workspaceId: fixture.workspace.id, + connectionId: ownerConnection.id, + agentId: ownerAgent.id, + role: "PRIMARY", + }, + }, + }, + }); + const trigger = await prisma.activityEvent.create({ + data: { + workspaceId: fixture.workspace.id, + kind: EventKind.COMMENT_CREATED, + actorId: fixture.user.id, + subjectType: "issue", + subjectId: issue.id, + payload: { mentions: { agentIds: [runtimeAgent.id] } }, + }, + }); + const run = await prisma.agentRun.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: runtimeAgent.id, + status: AgentRunStatus.ACTIVE, + triggerKind: EventKind.COMMENT_CREATED, + triggerEventId: trigger.id, + engagementMode: EngagementMode.EXECUTE, + }, + }); + + await expect(ingestRunsDispatch()).resolves.toBeDefined(); + + await expect( + prisma.agentRun.findUniqueOrThrow({ where: { id: run.id } }), + ).resolves.toMatchObject({ + status: AgentRunStatus.WAITING, + connectionId: legacyConnection.id, + externalRunId: null, + currentStep: "blocked by another primary delivery connection", + }); + const request = await prisma.actionRequest.findFirstOrThrow({ + where: { + issueId: issue.id, + kind: "DELIVERY_CONNECTION_CONFLICT", + status: "OPEN", + }, + }); + expect(request.sourceId).toBe(session.id); + expect(request.payload).toMatchObject({ + expectedOwnerConnectionId: ownerConnection.id, + candidateConnectionId: legacyConnection.id, + queuedRunId: run.id, + }); + } finally { + if (previousE2E === undefined) { + delete process.env.FORGE_E2E; + } else { + process.env.FORGE_E2E = previousE2E; + } + } + }); + it("resumes a WAITING run when an operator reply lands after it paused", async () => { const previousE2E = process.env.FORGE_E2E; process.env.FORGE_E2E = "1"; diff --git a/src/server/services/action-request-service.ts b/src/server/services/action-request-service.ts index 85aaa39a..7386892e 100644 --- a/src/server/services/action-request-service.ts +++ b/src/server/services/action-request-service.ts @@ -26,7 +26,7 @@ import { openOrTouchRun, appendRunEvent, finishRun } from "@/server/services/age import { resolveRunEngineWithSource } from "@/server/services/dispatch/registry"; import { FORGE_RUN_CONTRACT_VERSION } from "@/server/services/engagement-mode"; import { archiveIssue } from "@/server/services/issue-archive"; -import { agentIdSchema } from "@/server/validators"; +import { agentConnectionIdSchema, agentIdSchema } from "@/server/validators"; /** * Per-kind payload schemas. The MCP + tRPC entry points strip any @@ -93,8 +93,8 @@ const runtimeToolGrantPayload = z.object({ const deliveryConnectionConflictPayload = z.object({ version: z.literal(1), workSessionId: z.string().cuid(), - expectedOwnerConnectionId: z.string().cuid(), - candidateConnectionId: z.string().cuid(), + expectedOwnerConnectionId: agentConnectionIdSchema, + candidateConnectionId: agentConnectionIdSchema, queuedRunId: z.string().cuid(), triggerEventId: z.string().cuid(), attemptedMode: z.enum(["EXECUTE", "REVIEW"]), diff --git a/src/server/services/dispatch/run-dispatcher.ts b/src/server/services/dispatch/run-dispatcher.ts index d1686739..f2d15580 100644 --- a/src/server/services/dispatch/run-dispatcher.ts +++ b/src/server/services/dispatch/run-dispatcher.ts @@ -266,84 +266,146 @@ async function parkDeliveryConflict(params: { triggerEventId: string; attemptedMode: "EXECUTE" | "REVIEW"; }): Promise { - const parked = await db.$transaction(async (tx) => { - const changed = await tx.agentRun.updateMany({ - where: { - id: params.queuedRunId, - workspaceId: params.workspaceId, - issueId: params.issueId, - agentId: params.agentId, - externalRunId: null, - status: AgentRunStatus.ACTIVE, - OR: [{ connectionId: null }, { connectionId: params.candidateConnectionId }], - }, - data: { - status: AgentRunStatus.WAITING, - connectionId: params.candidateConnectionId, - lifecycleConfidence: "CONFIRMED", - triggerEventId: params.triggerEventId, - currentStep: "blocked by another primary delivery connection", - }, - }); - if (changed.count === 1) { - await appendRunEvent(tx, { - runId: params.queuedRunId, + let parked: boolean; + try { + parked = await db.$transaction(async (tx) => { + const changed = await tx.agentRun.updateMany({ + where: { + id: params.queuedRunId, + workspaceId: params.workspaceId, + issueId: params.issueId, + agentId: params.agentId, + externalRunId: null, + status: AgentRunStatus.ACTIVE, + OR: [{ connectionId: null }, { connectionId: params.candidateConnectionId }], + }, + data: { + status: AgentRunStatus.WAITING, + connectionId: params.candidateConnectionId, + lifecycleConfidence: "CONFIRMED", + triggerEventId: params.triggerEventId, + currentStep: "blocked by another primary delivery connection", + }, + }); + if (changed.count === 1) { + await appendRunEvent(tx, { + runId: params.queuedRunId, + workspaceId: params.workspaceId, + issueId: params.issueId, + agentId: params.agentId, + kind: "DELIVERY_CONNECTION_BLOCKED", + currentStep: "blocked by another primary delivery connection", + payload: { + workSessionId: params.workSessionId, + expectedOwnerConnectionId: params.expectedOwnerConnectionId, + candidateConnectionId: params.candidateConnectionId, + triggerEventId: params.triggerEventId, + }, + }); + } + const current = await tx.agentRun.findFirst({ + where: { + id: params.queuedRunId, + workspaceId: params.workspaceId, + issueId: params.issueId, + agentId: params.agentId, + connectionId: params.candidateConnectionId, + externalRunId: null, + status: AgentRunStatus.WAITING, + }, + select: { id: true }, + }); + if (!current) return false; + await createActionRequest(tx, { workspaceId: params.workspaceId, - issueId: params.issueId, - agentId: params.agentId, - kind: "DELIVERY_CONNECTION_BLOCKED", - currentStep: "blocked by another primary delivery connection", + actorId: null, + title: `${params.issueKey} already has a primary delivery connection`, + body: + `${params.runtimeName} was not started because another connection owns the active delivery session. ` + + (params.attemptedMode === "REVIEW" + ? "Join the existing delivery as a reviewer or cancel this dispatch." + : "Join as a reviewer or contributor, hand off primary ownership, or cancel this dispatch."), + severity: NotificationSeverity.WARNING, + kind: ActionRequestKind.DELIVERY_CONNECTION_CONFLICT, payload: { + version: 1, workSessionId: params.workSessionId, expectedOwnerConnectionId: params.expectedOwnerConnectionId, candidateConnectionId: params.candidateConnectionId, + queuedRunId: params.queuedRunId, triggerEventId: params.triggerEventId, + attemptedMode: params.attemptedMode, }, - }); - } - const current = await tx.agentRun.findFirst({ - where: { - id: params.queuedRunId, - workspaceId: params.workspaceId, + assignedUserId: params.assignedUserId, + assignedAgentId: params.agentId, + sourceType: "work-session", + sourceId: params.workSessionId, + dedupeKey: `work-session-execution-conflict:${params.workSessionId}:${params.candidateConnectionId}`, issueId: params.issueId, - agentId: params.agentId, - connectionId: params.candidateConnectionId, - externalRunId: null, - status: AgentRunStatus.WAITING, - }, - select: { id: true }, + systemOwned: true, + }); + return true; }); - if (!current) return false; - await createActionRequest(tx, { - workspaceId: params.workspaceId, - actorId: null, - title: `${params.issueKey} already has a primary delivery connection`, - body: - `${params.runtimeName} was not started because another connection owns the active delivery session. ` + - (params.attemptedMode === "REVIEW" - ? "Join the existing delivery as a reviewer or cancel this dispatch." - : "Join as a reviewer or contributor, hand off primary ownership, or cancel this dispatch."), - severity: NotificationSeverity.WARNING, - kind: ActionRequestKind.DELIVERY_CONNECTION_CONFLICT, - payload: { - version: 1, + } catch (err) { + logger.error( + { + err, + runId: params.queuedRunId, workSessionId: params.workSessionId, - expectedOwnerConnectionId: params.expectedOwnerConnectionId, candidateConnectionId: params.candidateConnectionId, - queuedRunId: params.queuedRunId, - triggerEventId: params.triggerEventId, - attemptedMode: params.attemptedMode, }, - assignedUserId: params.assignedUserId, - assignedAgentId: params.agentId, - sourceType: "work-session", - sourceId: params.workSessionId, - dedupeKey: `work-session-execution-conflict:${params.workSessionId}:${params.candidateConnectionId}`, - issueId: params.issueId, - systemOwned: true, - }); - return true; - }); + "delivery conflict decision failed to materialize", + ); + + // Ownership remains fail-closed even if presentation/decision creation is + // malformed. Isolate the candidate from the rest of the sweep and leave a + // durable run event so an operator can diagnose it from the issue. + try { + await db.$transaction(async (tx) => { + const changed = await tx.agentRun.updateMany({ + where: { + id: params.queuedRunId, + workspaceId: params.workspaceId, + issueId: params.issueId, + agentId: params.agentId, + externalRunId: null, + status: AgentRunStatus.ACTIVE, + OR: [{ connectionId: null }, { connectionId: params.candidateConnectionId }], + }, + data: { + status: AgentRunStatus.WAITING, + connectionId: params.candidateConnectionId, + lifecycleConfidence: "CONFIRMED", + triggerEventId: params.triggerEventId, + currentStep: "blocked by another primary delivery connection; decision unavailable", + }, + }); + if (changed.count === 1) { + await appendRunEvent(tx, { + runId: params.queuedRunId, + workspaceId: params.workspaceId, + issueId: params.issueId, + agentId: params.agentId, + kind: "DELIVERY_CONNECTION_BLOCKED", + currentStep: "blocked by another primary delivery connection; decision unavailable", + payload: { + workSessionId: params.workSessionId, + expectedOwnerConnectionId: params.expectedOwnerConnectionId, + candidateConnectionId: params.candidateConnectionId, + triggerEventId: params.triggerEventId, + decisionMaterialized: false, + }, + }); + } + }); + } catch (fallbackErr) { + logger.error( + { err: fallbackErr, runId: params.queuedRunId }, + "delivery conflict fallback parking failed", + ); + } + return; + } if (!parked) { logger.warn( { runId: params.queuedRunId, candidateConnectionId: params.candidateConnectionId }, diff --git a/src/server/services/mcp.ts b/src/server/services/mcp.ts index 092f8309..7d8a61e7 100644 --- a/src/server/services/mcp.ts +++ b/src/server/services/mcp.ts @@ -2,6 +2,7 @@ import "server-only"; import { createHash, randomBytes } from "node:crypto"; import { TRPCError } from "@trpc/server"; import { z } from "zod"; +import { agentConnectionIdSchema } from "@/server/validators"; import { AgentProvider, AgentRunStatus, @@ -8453,7 +8454,7 @@ export const mcpTools = { scopes: ["WRITE_ISSUES"] as const, input: z.object({ sessionId: z.string().cuid(), - targetConnectionId: z.string().cuid(), + targetConnectionId: agentConnectionIdSchema, }), async run(input: { sessionId: string; targetConnectionId: string }, ctx: McpContext) { const agentId = ctx.apiKey?.linkedAgentId; diff --git a/src/server/validators.ts b/src/server/validators.ts index 6cbb1e35..e106ced3 100644 --- a/src/server/validators.ts +++ b/src/server/validators.ts @@ -13,3 +13,15 @@ export const agentIdSchema = z .min(1) .max(40) .regex(/^[a-zA-Z0-9_-]+$/); + +/** + * AgentConnection ids are stable opaque identifiers, not a transport-level + * CUID contract. The original connection backfill used deterministic `ac_*` + * ids, while newly registered endpoints use Prisma CUIDs. Accept both during + * rolling upgrades and validate ownership through tenant-scoped lookups. + */ +export const agentConnectionIdSchema = z + .string() + .min(1) + .max(64) + .regex(/^[a-zA-Z0-9_-]+$/); diff --git a/tests/unit/agent-id-validator.test.ts b/tests/unit/agent-id-validator.test.ts index 69852490..4170b7e5 100644 --- a/tests/unit/agent-id-validator.test.ts +++ b/tests/unit/agent-id-validator.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; import { z } from "zod"; -import { agentIdSchema } from "@/server/validators"; +import { agentConnectionIdSchema, agentIdSchema } from "@/server/validators"; /** * Regression guard: agents are NOT always Prisma cuids. Hermes-seeded @@ -32,3 +32,19 @@ describe("agentIdSchema", () => { expect(agentIdSchema.safeParse("drop;table").success).toBe(false); }); }); + +describe("agentConnectionIdSchema", () => { + it("accepts both historical backfill ids and current cuids", () => { + expect( + agentConnectionIdSchema.safeParse("ac_269a426f3d6cf83523c051e35f1adf93").success, + ).toBe(true); + expect(agentConnectionIdSchema.safeParse("cmohnxz9h0001n407mytknqgv").success).toBe(true); + }); + + it("rejects empty, overlong, and unsafe identifiers", () => { + expect(agentConnectionIdSchema.safeParse("").success).toBe(false); + expect(agentConnectionIdSchema.safeParse("a".repeat(65)).success).toBe(false); + expect(agentConnectionIdSchema.safeParse("connection/id").success).toBe(false); + expect(agentConnectionIdSchema.safeParse("connection id").success).toBe(false); + }); +});