diff --git a/packages/db/src/issue-comment-derived-attribution-migration.test.ts b/packages/db/src/issue-comment-derived-attribution-migration.test.ts new file mode 100644 index 0000000000..f208dced2f --- /dev/null +++ b/packages/db/src/issue-comment-derived-attribution-migration.test.ts @@ -0,0 +1,345 @@ +import { createHash, randomUUID } from "node:crypto"; +import fs from "node:fs"; +import { afterEach, describe, expect, it } from "vitest"; +import postgres from "postgres"; +import { + applyPendingMigrations, + inspectMigrations, +} from "./client.js"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./test-embedded-postgres.js"; + +const DERIVED_ATTRIBUTION_MIGRATION = "0132_issue_comment_derived_attribution_fast.sql"; + +const cleanups: Array<() => Promise> = []; +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +async function createTempDatabase(): Promise { + const db = await startEmbeddedPostgresTestDatabase("paperclip-derived-attribution-"); + cleanups.push(db.cleanup); + return db.connectionString; +} + +async function migrationHash(migrationFile: string): Promise { + const content = await fs.promises.readFile( + new URL(`./migrations/${migrationFile}`, import.meta.url), + "utf8", + ); + return createHash("sha256").update(content).digest("hex"); +} + +async function makeDerivedAttributionMigrationPending( + sql: ReturnType, +): Promise { + const hash = await migrationHash(DERIVED_ATTRIBUTION_MIGRATION); + await sql` + DELETE FROM "drizzle"."__drizzle_migrations" + WHERE "hash" = ${hash} + `; +} + +async function dropDerivedAttributionSchema(sql: ReturnType): Promise { + await sql`ALTER TABLE "issue_comments" DROP CONSTRAINT IF EXISTS "issue_comments_derived_author_agent_id_agents_id_fk"`; + await sql`ALTER TABLE "issue_comments" DROP CONSTRAINT IF EXISTS "issue_comments_derived_created_by_run_id_heartbeat_runs_id_fk"`; + await sql`ALTER TABLE "issue_comments" DROP COLUMN IF EXISTS "derived_author_agent_id"`; + await sql`ALTER TABLE "issue_comments" DROP COLUMN IF EXISTS "derived_created_by_run_id"`; + await sql`ALTER TABLE "issue_comments" DROP COLUMN IF EXISTS "derived_author_source"`; +} + +async function createSeedGraph(sql: ReturnType, label: string) { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const runId = randomUUID(); + + await sql` + INSERT INTO "companies" ("id", "name", "issue_prefix") + VALUES (${companyId}, ${`Company ${label}`}, ${`T${label}`}) + `; + await sql` + INSERT INTO "agents" ("id", "company_id", "name", "role", "adapter_type", "adapter_config") + VALUES (${agentId}, ${companyId}, ${`Agent ${label}`}, 'engineer', 'process', '{}'::jsonb) + `; + await sql` + INSERT INTO "issues" ("id", "company_id", "title", "identifier") + VALUES (${issueId}, ${companyId}, ${`Issue ${label}`}, ${`T${label}-1`}) + `; + await sql` + INSERT INTO "heartbeat_runs" ("id", "company_id", "agent_id", "status") + VALUES (${runId}, ${companyId}, ${agentId}, 'succeeded') + `; + + return { companyId, agentId, issueId, runId }; +} + +async function expectDerivedAttributionSchema(sql: ReturnType): Promise { + const columns = await sql<{ column_name: string; data_type: string; is_nullable: string }[]>` + SELECT "column_name", "data_type", "is_nullable" + FROM "information_schema"."columns" + WHERE "table_schema" = 'public' + AND "table_name" = 'issue_comments' + AND "column_name" IN ( + 'derived_author_agent_id', + 'derived_created_by_run_id', + 'derived_author_source' + ) + ORDER BY "column_name" + `; + expect(columns).toEqual([ + { column_name: "derived_author_agent_id", data_type: "uuid", is_nullable: "YES" }, + { column_name: "derived_author_source", data_type: "text", is_nullable: "YES" }, + { column_name: "derived_created_by_run_id", data_type: "uuid", is_nullable: "YES" }, + ]); + + const constraints = await sql<{ conname: string; delete_rule: string }[]>` + SELECT tc."constraint_name" AS "conname", rc."delete_rule" + FROM "information_schema"."table_constraints" tc + JOIN "information_schema"."referential_constraints" rc + ON rc."constraint_schema" = tc."constraint_schema" + AND rc."constraint_name" = tc."constraint_name" + WHERE tc."table_schema" = 'public' + AND tc."table_name" = 'issue_comments' + AND tc."constraint_name" IN ( + 'issue_comments_derived_author_agent_id_agents_id_fk', + 'issue_comments_derived_created_by_run_id_heartbeat_runs_id_fk' + ) + ORDER BY tc."constraint_name" + `; + expect(constraints).toEqual([ + { + conname: "issue_comments_derived_author_agent_id_agents_id_fk", + delete_rule: "SET NULL", + }, + { + conname: "issue_comments_derived_created_by_run_id_heartbeat_runs_id_fk", + delete_rule: "SET NULL", + }, + ]); +} + +afterEach(async () => { + while (cleanups.length > 0) { + const cleanup = cleanups.pop(); + await cleanup?.(); + } +}); + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres derived attribution migration tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("issue comment derived attribution migration", () => { + it( + "fresh installs include the relocated schema and no deleted 0126 migration", + async () => { + const connectionString = await createTempDatabase(); + const state = await inspectMigrations(connectionString); + + expect(state.status).toBe("upToDate"); + expect(state.availableMigrations).not.toContain("0126_issue_comment_derived_attribution.sql"); + expect(state.availableMigrations).toContain(DERIVED_ATTRIBUTION_MIGRATION); + + const sql = postgres(connectionString, { max: 1, onnotice: () => {} }); + try { + await expectDerivedAttributionSchema(sql); + const supportIndexes = await sql<{ indexname: string }[]>` + SELECT "indexname" + FROM "pg_indexes" + WHERE "schemaname" = 'public' + AND "indexname" = 'issue_comments_derived_attribution_backfill_idx' + `; + expect(supportIndexes).toEqual([]); + } finally { + await sql.end(); + } + }, + 20_000, + ); + + it( + "is idempotent for a database that already has 0126 schema and data", + async () => { + const connectionString = await createTempDatabase(); + const sql = postgres(connectionString, { max: 1, onnotice: () => {} }); + const alreadyBackfilledCommentId = randomUUID(); + const timingTierCommentId = randomUUID(); + const realUserCommentId = randomUUID(); + + try { + await makeDerivedAttributionMigrationPending(sql); + const { companyId, agentId, issueId, runId } = await createSeedGraph(sql, "OLD126"); + await sql` + INSERT INTO "user" ("id", "name", "email", "email_verified", "created_at", "updated_at") + VALUES ('real-user', 'Real User', 'real-user@example.test', true, now(), now()) + ON CONFLICT ("id") DO NOTHING + `; + await sql` + INSERT INTO "issue_comments" ( + "id", + "company_id", + "issue_id", + "author_user_id", + "created_by_run_id", + "derived_author_agent_id", + "derived_created_by_run_id", + "derived_author_source", + "body" + ) + VALUES + (${alreadyBackfilledCommentId}, ${companyId}, ${issueId}, 'local-board', ${runId}, ${agentId}, ${runId}, 'run_id', 'already attributed'), + (${timingTierCommentId}, ${companyId}, ${issueId}, 'local-board', NULL, ${agentId}, ${runId}, 'run_window_unique', 'timing tier'), + (${realUserCommentId}, ${companyId}, ${issueId}, 'real-user', ${runId}, NULL, NULL, NULL, 'human comment') + `; + } finally { + await sql.end(); + } + + const pendingState = await inspectMigrations(connectionString); + expect(pendingState).toMatchObject({ + status: "needsMigrations", + pendingMigrations: [DERIVED_ATTRIBUTION_MIGRATION], + reason: "pending-migrations", + }); + + await applyPendingMigrations(connectionString); + + const verifySql = postgres(connectionString, { max: 1, onnotice: () => {} }); + try { + const rows = await verifySql<{ + id: string; + derived_author_agent_id: string | null; + derived_created_by_run_id: string | null; + derived_author_source: string | null; + }[]>` + SELECT + "id", + "derived_author_agent_id", + "derived_created_by_run_id", + "derived_author_source" + FROM "issue_comments" + WHERE "id" IN (${alreadyBackfilledCommentId}, ${timingTierCommentId}, ${realUserCommentId}) + ORDER BY "body" + `; + + expect(rows).toEqual([ + expect.objectContaining({ + id: alreadyBackfilledCommentId, + derived_author_source: "run_id", + }), + { + id: realUserCommentId, + derived_author_agent_id: null, + derived_created_by_run_id: null, + derived_author_source: null, + }, + { + id: timingTierCommentId, + derived_author_agent_id: null, + derived_created_by_run_id: null, + derived_author_source: null, + }, + ]); + } finally { + await verifySql.end(); + } + + const finalState = await inspectMigrations(connectionString); + expect(finalState.status).toBe("upToDate"); + }, + 20_000, + ); + + it( + "completes a partially backfilled pre-0131 database", + async () => { + const connectionString = await createTempDatabase(); + const sql = postgres(connectionString, { max: 1, onnotice: () => {} }); + const eligibleCommentId = randomUUID(); + const deletedUserCommentId = randomUUID(); + const agentAuthoredCommentId = randomUUID(); + + try { + await dropDerivedAttributionSchema(sql); + await makeDerivedAttributionMigrationPending(sql); + const { companyId, agentId, issueId, runId } = await createSeedGraph(sql, "PARTIAL"); + await sql` + INSERT INTO "issue_comments" ( + "id", + "company_id", + "issue_id", + "author_user_id", + "created_by_run_id", + "body" + ) + VALUES + (${eligibleCommentId}, ${companyId}, ${issueId}, 'local-board', ${runId}, 'eligible local-board'), + (${deletedUserCommentId}, ${companyId}, ${issueId}, 'deleted-user', ${runId}, 'eligible deleted user') + `; + await sql` + INSERT INTO "issue_comments" ( + "id", + "company_id", + "issue_id", + "author_agent_id", + "author_user_id", + "created_by_run_id", + "body" + ) + VALUES (${agentAuthoredCommentId}, ${companyId}, ${issueId}, ${agentId}, 'local-board', ${runId}, 'already agent-authored') + `; + } finally { + await sql.end(); + } + + await applyPendingMigrations(connectionString); + + const verifySql = postgres(connectionString, { max: 1, onnotice: () => {} }); + try { + await expectDerivedAttributionSchema(verifySql); + const rows = await verifySql<{ + id: string; + derived_author_agent_id: string | null; + derived_created_by_run_id: string | null; + derived_author_source: string | null; + }[]>` + SELECT + "id", + "derived_author_agent_id", + "derived_created_by_run_id", + "derived_author_source" + FROM "issue_comments" + WHERE "id" IN (${eligibleCommentId}, ${deletedUserCommentId}, ${agentAuthoredCommentId}) + ORDER BY "body" + `; + + expect(rows).toEqual([ + { + id: agentAuthoredCommentId, + derived_author_agent_id: null, + derived_created_by_run_id: null, + derived_author_source: null, + }, + expect.objectContaining({ + id: deletedUserCommentId, + derived_author_source: "run_id", + }), + expect.objectContaining({ + id: eligibleCommentId, + derived_author_source: "run_id", + }), + ]); + } finally { + await verifySql.end(); + } + + const finalState = await inspectMigrations(connectionString); + expect(finalState.status).toBe("upToDate"); + }, + 20_000, + ); +}); diff --git a/packages/db/src/migrations/0126_issue_comment_derived_attribution.sql b/packages/db/src/migrations/0126_issue_comment_derived_attribution.sql deleted file mode 100644 index 564c2ad7a6..0000000000 --- a/packages/db/src/migrations/0126_issue_comment_derived_attribution.sql +++ /dev/null @@ -1,69 +0,0 @@ -ALTER TABLE "issue_comments" ADD COLUMN IF NOT EXISTS "derived_author_agent_id" uuid;--> statement-breakpoint -ALTER TABLE "issue_comments" ADD COLUMN IF NOT EXISTS "derived_created_by_run_id" uuid;--> statement-breakpoint -ALTER TABLE "issue_comments" ADD COLUMN IF NOT EXISTS "derived_author_source" text;--> statement-breakpoint -DO $$ BEGIN - IF NOT EXISTS ( - SELECT 1 FROM "pg_constraint" WHERE "conname" = 'issue_comments_derived_author_agent_id_agents_id_fk' - ) THEN - ALTER TABLE "issue_comments" ADD CONSTRAINT "issue_comments_derived_author_agent_id_agents_id_fk" FOREIGN KEY ("derived_author_agent_id") REFERENCES "public"."agents"("id") ON DELETE set null ON UPDATE no action; - END IF; -END $$;--> statement-breakpoint -DO $$ BEGIN - IF NOT EXISTS ( - SELECT 1 FROM "pg_constraint" WHERE "conname" = 'issue_comments_derived_created_by_run_id_heartbeat_runs_id_fk' - ) THEN - ALTER TABLE "issue_comments" ADD CONSTRAINT "issue_comments_derived_created_by_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("derived_created_by_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE set null ON UPDATE no action; - END IF; -END $$;--> statement-breakpoint --- Backfill agent attribution for historical non-human ("Board") comments so old --- threads stop rendering as blue board bubbles and the read path stops --- re-scanning run logs. Two SQL-computable tiers, both guarded to NEVER touch a --- comment whose author maps to a genuine user profile. The log-marker tier is --- handled lazily on read (it needs object-storage log bodies). --- --- Tier `run_id`: the comment's own authoring run resolves to an agent (lossless). --- Batched to keep lock/WAL footprint bounded on large histories. -DO $$ -DECLARE - affected integer; -BEGIN - LOOP - WITH batch AS ( - SELECT c.id AS comment_id, hr.agent_id, hr.id AS run_id - FROM issue_comments c - JOIN heartbeat_runs hr ON hr.id = c.created_by_run_id - WHERE c.author_agent_id IS NULL - AND c.derived_author_agent_id IS NULL - AND c.author_user_id IS NOT NULL - -- Only the non-human board sentinel or non-`user` authors are - -- eligible. `local-board` IS a row in "user" (the implicit board - -- admin), so it must be allowed explicitly; genuine signups are not. - AND (c.author_user_id = 'local-board' - OR NOT EXISTS (SELECT 1 FROM "user" u WHERE u.id = c.author_user_id)) - LIMIT 5000 - ) - UPDATE issue_comments c - SET derived_author_agent_id = b.agent_id, - derived_created_by_run_id = b.run_id, - derived_author_source = 'run_id' - FROM batch b - WHERE c.id = b.comment_id; - GET DIAGNOSTICS affected = ROW_COUNT; - EXIT WHEN affected = 0; - END LOOP; -END $$;--> statement-breakpoint --- Option A: the pure run-window TIMING tiers are intentionally NOT applied. --- Because agents post through the `local-board` subprocess, an agent comment and --- a genuine human board comment are indistinguishable rows, so any timing-overlap --- guess mis-attributes human board comments that merely coincided with an agent --- run (e.g. a human board reply typed while an agent run was in flight). --- Only the lossless `run_id` backfill above (and the read-path `run_log_comment_post` --- tier) attribute history; everything else stays "Board". --- --- This statement also reverts any attribution a PRIOR revision of this migration --- persisted via the timing tiers, so re-applying / upgrading is idempotent. -UPDATE issue_comments -SET derived_author_agent_id = NULL, - derived_created_by_run_id = NULL, - derived_author_source = NULL -WHERE derived_author_source IN ('run_window_unique', 'run_window_agent_unique'); diff --git a/packages/db/src/migrations/0132_issue_comment_derived_attribution_fast.sql b/packages/db/src/migrations/0132_issue_comment_derived_attribution_fast.sql new file mode 100644 index 0000000000..4882c76654 --- /dev/null +++ b/packages/db/src/migrations/0132_issue_comment_derived_attribution_fast.sql @@ -0,0 +1,82 @@ +ALTER TABLE "issue_comments" ADD COLUMN IF NOT EXISTS "derived_author_agent_id" uuid;--> statement-breakpoint +ALTER TABLE "issue_comments" ADD COLUMN IF NOT EXISTS "derived_created_by_run_id" uuid;--> statement-breakpoint +ALTER TABLE "issue_comments" ADD COLUMN IF NOT EXISTS "derived_author_source" text;--> statement-breakpoint +DO $$ BEGIN + IF NOT EXISTS ( + SELECT 1 FROM "pg_constraint" WHERE "conname" = 'issue_comments_derived_author_agent_id_agents_id_fk' + ) THEN + ALTER TABLE "issue_comments" ADD CONSTRAINT "issue_comments_derived_author_agent_id_agents_id_fk" FOREIGN KEY ("derived_author_agent_id") REFERENCES "public"."agents"("id") ON DELETE set null ON UPDATE no action; + END IF; +END $$;--> statement-breakpoint +DO $$ BEGIN + IF NOT EXISTS ( + SELECT 1 FROM "pg_constraint" WHERE "conname" = 'issue_comments_derived_created_by_run_id_heartbeat_runs_id_fk' + ) THEN + ALTER TABLE "issue_comments" ADD CONSTRAINT "issue_comments_derived_created_by_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("derived_created_by_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE set null ON UPDATE no action; + END IF; +END $$;--> statement-breakpoint +-- Temporary support for the forward-only backfill. The keyset loop below +-- advances over this partial index by comment id, so each eligible slice is +-- visited once instead of re-scanning issue_comments from the beginning on +-- every batch. +CREATE INDEX IF NOT EXISTS "issue_comments_derived_attribution_backfill_idx" + ON "issue_comments" USING btree ("id") + WHERE "author_agent_id" IS NULL + AND "derived_author_agent_id" IS NULL + AND "author_user_id" IS NOT NULL + AND "created_by_run_id" IS NOT NULL;--> statement-breakpoint +ANALYZE "issue_comments";--> statement-breakpoint +DO $$ +DECLARE + last_comment_id uuid := '00000000-0000-0000-0000-000000000000'::uuid; + next_last_comment_id uuid; +BEGIN + LOOP + next_last_comment_id := NULL; + + WITH batch AS MATERIALIZED ( + SELECT c."id" AS comment_id, hr."agent_id", hr."id" AS run_id + FROM "issue_comments" c + JOIN "heartbeat_runs" hr ON hr."id" = c."created_by_run_id" + LEFT JOIN "user" u ON u."id" = c."author_user_id" + WHERE c."id" > last_comment_id + AND c."author_agent_id" IS NULL + AND c."derived_author_agent_id" IS NULL + AND c."author_user_id" IS NOT NULL + AND c."created_by_run_id" IS NOT NULL + AND ( + c."author_user_id" = 'local-board' + OR u."id" IS NULL + ) + ORDER BY c."id" + LIMIT 5000 + ), + updated AS ( + UPDATE "issue_comments" c + SET "derived_author_agent_id" = b."agent_id", + "derived_created_by_run_id" = b."run_id", + "derived_author_source" = 'run_id' + FROM batch b + WHERE c."id" = b."comment_id" + RETURNING c."id" + ) + SELECT b."comment_id" + INTO next_last_comment_id + FROM batch b + LEFT JOIN updated u ON u."id" = b."comment_id" + ORDER BY b."comment_id" DESC + LIMIT 1; + + EXIT WHEN next_last_comment_id IS NULL; + last_comment_id := next_last_comment_id; + END LOOP; +END $$;--> statement-breakpoint +DROP INDEX IF EXISTS "issue_comments_derived_attribution_backfill_idx";--> statement-breakpoint +-- Keep the Option-A cleanup at the end as well, matching the original 0126 +-- terminal state if any timing-tier rows are introduced before this migration +-- is retried. +UPDATE "issue_comments" +SET "derived_author_agent_id" = NULL, + "derived_created_by_run_id" = NULL, + "derived_author_source" = NULL +WHERE "derived_author_source" IN ('run_window_unique', 'run_window_agent_unique'); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 81ba9a4972..f1094e4725 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -884,13 +884,6 @@ "tag": "0125_environment_custom_image_templates", "breakpoints": true }, - { - "idx": 126, - "version": "7", - "when": 1782526400000, - "tag": "0126_issue_comment_derived_attribution", - "breakpoints": true - }, { "idx": 127, "version": "7", @@ -925,6 +918,13 @@ "when": 1783025324120, "tag": "0131_repair_run_responsible_user_context_refs", "breakpoints": true + }, + { + "idx": 132, + "version": "7", + "when": 1783025424120, + "tag": "0132_issue_comment_derived_attribution_fast", + "breakpoints": true } ] }