fix: normalize legacy connection keys (AXI-134)

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

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

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";

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

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

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";

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"]),