fix: enforce work product primary invariant

This commit is contained in:
ZacharyZhang-NY committed 2026-04-23 21:08:07 -04:00
1 parent 6c757f2081
commit c850d70bfc
5 files changed
+356 -48

No files matched your search

+69 -47
View File
@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto";
import { statSync } from "node:fs";
import { isAbsolute, resolve as resolvePath, relative as relativePath } from "node:path";
import type { DatabaseSync, SQLInputValue } from "node:sqlite";
import type { ReadAll } from "./helpers";
import type { ReadAll, Transaction } from "./helpers";
import { nowIso } from "./helpers";
function resolveWorkspaceRoot(
@@ -76,6 +76,7 @@ function joinWorkspaceRelative(root: string, rel: string): string | null {
export function buildWorkProductMethods(
getDb: () => DatabaseSync,
readAll: ReadAll,
transaction: Transaction,
) {
return {
createWorkProduct(input: {
@@ -86,26 +87,29 @@ export function buildWorkProductMethods(
isPrimary?: boolean; summary?: string | null; metadataJson?: string | null;
createdByRunId?: string | null;
}): string {
const id = randomUUID();
const ts = nowIso();
if (input.isPrimary && input.taskId) {
return transaction(() => {
const id = randomUUID();
const ts = nowIso();
const status = input.status ?? "active";
if (input.isPrimary && input.taskId && status !== "archived") {
getDb().prepare(
"UPDATE work_products SET is_primary = 0, updated_at = ? WHERE company_id = ? AND task_id = ? AND type = ? AND status <> 'archived'",
).run(ts, input.companyId, input.taskId, input.type);
}
getDb().prepare(
"UPDATE work_products SET is_primary = 0, updated_at = ? WHERE company_id = ? AND task_id = ? AND type = ?",
).run(ts, input.companyId, input.taskId, input.type);
}
getDb().prepare(
`INSERT INTO work_products (id, company_id, project_id, task_id, workspace_id, type, provider, external_id, title, url, status, review_state, is_primary, summary, metadata_json, created_by_run_id, last_touched_by_run_id, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
).run(
id, input.companyId, input.projectId ?? null, input.taskId ?? null,
input.workspaceId ?? null,
input.type, input.provider, input.externalId ?? null,
input.title, input.url ?? null, input.status ?? "active",
input.reviewState ?? "none", input.isPrimary ? 1 : 0,
input.summary ?? null, input.metadataJson ?? null,
input.createdByRunId ?? null, input.createdByRunId ?? null, ts, ts,
);
return id;
`INSERT INTO work_products (id, company_id, project_id, task_id, workspace_id, type, provider, external_id, title, url, status, review_state, is_primary, summary, metadata_json, created_by_run_id, last_touched_by_run_id, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
).run(
id, input.companyId, input.projectId ?? null, input.taskId ?? null,
input.workspaceId ?? null,
input.type, input.provider, input.externalId ?? null,
input.title, input.url ?? null, status,
input.reviewState ?? "none", input.isPrimary ? 1 : 0,
input.summary ?? null, input.metadataJson ?? null,
input.createdByRunId ?? null, input.createdByRunId ?? null, ts, ts,
);
return id;
});
},
findWorkProductByExternalId(
@@ -152,34 +156,52 @@ export function buildWorkProductMethods(
lastTouchedByRunId?: string | null;
},
): number {
const ts = nowIso();
const sets: string[] = ["updated_at = ?"];
const values: SQLInputValue[] = [ts];
const map: Array<[keyof typeof patch, string, (v: unknown) => SQLInputValue]> = [
["type", "type", (v) => (v as string) ?? null],
["provider", "provider", (v) => (v as string) ?? null],
["externalId", "external_id", (v) => (v as string | null) ?? null],
["workspaceId", "workspace_id", (v) => (v as string | null) ?? null],
["title", "title", (v) => (v as string) ?? null],
["url", "url", (v) => (v as string | null) ?? null],
["status", "status", (v) => (v as string) ?? null],
["reviewState", "review_state", (v) => (v as string) ?? null],
["isPrimary", "is_primary", (v) => (v ? 1 : 0)],
["summary", "summary", (v) => (v as string | null) ?? null],
["metadataJson", "metadata_json", (v) => (v as string | null) ?? null],
["lastTouchedByRunId", "last_touched_by_run_id", (v) => (v as string | null) ?? null],
];
for (const [key, col, coerce] of map) {
if (key in patch) {
sets.push(`${col} = ?`);
values.push(coerce(patch[key]));
return transaction(() => {
const current = getDb().prepare(
"SELECT task_id as taskId, type, status, is_primary as isPrimary FROM work_products WHERE id = ? AND company_id = ?",
).get(id, companyId) as
| { taskId: string | null; type: string; status: string; isPrimary: number }
| undefined;
if (!current) return 0;
const ts = nowIso();
const nextType = patch.type ?? current.type;
const nextStatus = patch.status ?? current.status;
const nextIsPrimary = patch.isPrimary ?? Boolean(current.isPrimary);
if (nextIsPrimary && current.taskId && nextStatus !== "archived") {
getDb().prepare(
"UPDATE work_products SET is_primary = 0, updated_at = ? WHERE company_id = ? AND task_id = ? AND type = ? AND status <> 'archived' AND id <> ?",
).run(ts, companyId, current.taskId, nextType, id);
}
}
values.push(id, companyId);
const result = getDb()
.prepare(`UPDATE work_products SET ${sets.join(", ")} WHERE id = ? AND company_id = ?`)
.run(...values);
return Number(result.changes ?? 0);
const sets: string[] = ["updated_at = ?"];
const values: SQLInputValue[] = [ts];
const map: Array<[keyof typeof patch, string, (v: unknown) => SQLInputValue]> = [
["type", "type", (v) => (v as string) ?? null],
["provider", "provider", (v) => (v as string) ?? null],
["externalId", "external_id", (v) => (v as string | null) ?? null],
["workspaceId", "workspace_id", (v) => (v as string | null) ?? null],
["title", "title", (v) => (v as string) ?? null],
["url", "url", (v) => (v as string | null) ?? null],
["status", "status", (v) => (v as string) ?? null],
["reviewState", "review_state", (v) => (v as string) ?? null],
["isPrimary", "is_primary", (v) => (v ? 1 : 0)],
["summary", "summary", (v) => (v as string | null) ?? null],
["metadataJson", "metadata_json", (v) => (v as string | null) ?? null],
["lastTouchedByRunId", "last_touched_by_run_id", (v) => (v as string | null) ?? null],
];
for (const [key, col, coerce] of map) {
if (key in patch) {
sets.push(`${col} = ?`);
values.push(coerce(patch[key]));
}
}
values.push(id, companyId);
const result = getDb()
.prepare(`UPDATE work_products SET ${sets.join(", ")} WHERE id = ? AND company_id = ?`)
.run(...values);
return Number(result.changes ?? 0);
});
},
resolveWorkProductTarget(
+1 -1
View File
@@ -55,7 +55,7 @@ export function buildDocumentMethods(
) {
return {
...buildConfigRevisionMethods(getDb, readAll, transaction),
...buildWorkProductMethods(getDb, readAll),
...buildWorkProductMethods(getDb, readAll, transaction),
saveDocument(input: {
id?: string | null; companyId: string; type: string; title: string; content?: string;
authorAgentId?: string | null; reviewerAgentId?: string | null; projectId?: string | null;
+48
View File
@@ -0,0 +1,48 @@
import log from "electron-log/main.js";
import type { Migration, MigrationDeps } from "./types";
import { tableExists } from "./types";
export function createMigrationsV56V56(deps: MigrationDeps): Migration[] {
const { db } = deps;
return [
{
version: 56,
name: "work_products_primary_invariant",
fn: () => {
if (!tableExists(db, "work_products")) {
log.info("[migration v56] work_products table missing; skipping");
return;
}
db.exec(`
UPDATE work_products
SET is_primary = 0,
updated_at = datetime('now')
WHERE id IN (
SELECT id FROM (
SELECT
id,
row_number() OVER (
PARTITION BY company_id, task_id, type
ORDER BY updated_at DESC, created_at DESC, id DESC
) AS rank_in_group
FROM work_products
WHERE task_id IS NOT NULL
AND is_primary = 1
AND status <> 'archived'
)
WHERE rank_in_group > 1
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_work_products_primary_active
ON work_products(company_id, task_id, type)
WHERE task_id IS NOT NULL
AND is_primary = 1
AND status <> 'archived';
`);
log.info("[migration v56] work_products active primary invariant installed");
},
},
];
}
+2
View File
@@ -21,6 +21,7 @@ import { createMigrationsV52V52 } from "./migrations/v52-v52";
import { createMigrationsV53V53 } from "./migrations/v53-v53";
import { createMigrationsV54V54 } from "./migrations/v54-v54";
import { createMigrationsV55V55 } from "./migrations/v55-v55";
import { createMigrationsV56V56 } from "./migrations/v56-v56";
import {
columnExists,
nowIso,
@@ -223,6 +224,7 @@ export function runMigrations(
...createMigrationsV53V53(deps),
...createMigrationsV54V54(deps),
...createMigrationsV55V55(deps),
...createMigrationsV56V56(deps),
];
// Pre-migration snapshot. If any migration is pending we take a full
@@ -0,0 +1,236 @@
import { mkdtemp, rm } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { DatabaseSync } from "node:sqlite";
import { afterEach, describe, expect, it } from "vitest";
import { AppDatabase } from "../../src/main/database";
const tempDirs: string[] = [];
async function createTempDir(prefix: string) {
const dir = await mkdtemp(join(tmpdir(), prefix));
tempDirs.push(dir);
return dir;
}
afterEach(async () => {
await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true })));
});
async function createHarness() {
const profileDir = await createTempDir("agentcompany-primary-wp-");
const db = new AppDatabase(profileDir);
db.init();
const companyId = db.core.company.saveCompany({
name: "Primary Work Product Co",
description: "Tests primary invariant",
status: "active",
});
const agentId = db.core.agents.saveAgent({
companyId,
name: "Writer",
role: "Engineer",
title: "Writer",
status: "active",
reportsTo: null,
connectorId: "openai-codex",
model: null,
capabilities: "Writes artifacts",
budgetMonthlyUsd: 100,
});
const taskId = db.core.tasks.saveTask({
companyId,
title: "Create deliverables",
description: "",
assigneeAgentId: agentId,
priority: "medium",
status: "in_progress",
});
return { db, companyId, taskId };
}
function createWorkProduct(db: AppDatabase, companyId: string, taskId: string, title: string, isPrimary = false, type = "file") {
return db.core.documents.createWorkProduct({
companyId,
taskId,
type,
provider: "filesystem",
externalId: `${title}.md`,
title,
isPrimary,
});
}
function listPrimaryIds(db: AppDatabase, companyId: string, taskId: string) {
return listPrimaryIdsByType(db, companyId, taskId, "file");
}
function listPrimaryIdsByType(db: AppDatabase, companyId: string, taskId: string, type: string) {
return db.queryAll<{ id: string }>(
"select id from work_products where company_id = ? and task_id = ? and type = ? and is_primary = 1 and status <> 'archived' order by updated_at desc",
companyId,
taskId,
type,
).map((row) => row.id);
}
describe("work product primary invariant", () => {
it("installs the database unique index for active primary rows", async () => {
const { db } = await createHarness();
try {
const index = db.queryAll<{ sql: string | null }>(
"select sql from sqlite_master where type = 'index' and name = 'idx_work_products_primary_active' limit 1",
)[0];
expect(index?.sql).toContain("ON work_products(company_id, task_id, type)");
expect(index?.sql).toContain("is_primary = 1");
expect(index?.sql).toContain("status <> 'archived'");
} finally {
db.close();
}
});
it("normalizes duplicate active primaries during v56 migration", async () => {
const profileDir = await createTempDir("agentcompany-primary-migration-");
const dbPath = join(profileDir, "profile.sqlite");
const rawDb = new DatabaseSync(dbPath);
const ts = "2026-04-24T00:00:00.000Z";
try {
rawDb.exec(`
create table settings (key text primary key, value text not null);
create table companies (id text primary key, name text not null, description text not null, status text not null, created_at text not null, updated_at text not null);
create table agents (id text primary key, company_id text not null, status text not null, reports_to text, department text, connector_id text not null default 'openai-codex', updated_at text not null);
create table workspaces (id text primary key, company_id text not null, local_path text not null, state text not null, lock_holder_run_id text, lock_acquired_at text, updated_at text not null);
create table agent_messages (id text primary key, company_id text not null, channel text not null, channel_target_id text, created_at text not null);
create table model_capabilities (
provider TEXT NOT NULL,
model_id TEXT NOT NULL,
reasoning INTEGER NOT NULL,
coding INTEGER NOT NULL,
writing INTEGER NOT NULL,
tier TEXT NOT NULL,
cost_per_mtok_out REAL NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY (provider, model_id)
);
create table work_products (
id TEXT PRIMARY KEY,
company_id TEXT NOT NULL,
project_id TEXT,
task_id TEXT,
workspace_id TEXT,
type TEXT NOT NULL,
provider TEXT NOT NULL,
external_id TEXT,
title TEXT NOT NULL,
url TEXT,
status TEXT NOT NULL DEFAULT 'active',
review_state TEXT NOT NULL DEFAULT 'none',
is_primary INTEGER NOT NULL DEFAULT 0,
summary TEXT,
metadata_json TEXT,
created_by_run_id TEXT,
last_touched_by_run_id TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
create table schema_migrations (version integer primary key, name text not null, applied_at text not null);
`);
for (let version = 1; version <= 55; version += 1) {
rawDb
.prepare("insert into schema_migrations (version, name, applied_at) values (?, ?, ?)")
.run(version, `legacy-${version}`, ts);
}
rawDb
.prepare("insert into companies (id, name, description, status, created_at, updated_at) values (?, ?, ?, ?, ?, ?)")
.run("company-1", "Legacy Co", "", "active", ts, ts);
rawDb
.prepare("insert into work_products (id, company_id, task_id, type, provider, external_id, title, status, review_state, is_primary, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
.run("older-primary", "company-1", "task-1", "file", "filesystem", "older.md", "Older", "active", "none", 1, "2026-04-24T00:00:01.000Z", "2026-04-24T00:00:01.000Z");
rawDb
.prepare("insert into work_products (id, company_id, task_id, type, provider, external_id, title, status, review_state, is_primary, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
.run("newer-primary", "company-1", "task-1", "file", "filesystem", "newer.md", "Newer", "active", "none", 1, "2026-04-24T00:00:02.000Z", "2026-04-24T00:00:02.000Z");
} finally {
rawDb.close();
}
const appDb = new AppDatabase(profileDir);
appDb.init();
try {
expect(listPrimaryIdsByType(appDb, "company-1", "task-1", "file")).toEqual(["newer-primary"]);
const older = appDb.queryAll<{ isPrimary: number }>("select is_primary as isPrimary from work_products where id = 'older-primary'")[0];
expect(older?.isPrimary).toBe(0);
const migration = appDb.queryAll<{ version: number }>("select version from schema_migrations where version = 56")[0];
expect(migration?.version).toBe(56);
const index = appDb.queryAll<{ sql: string | null }>(
"select sql from sqlite_master where type = 'index' and name = 'idx_work_products_primary_active' limit 1",
)[0];
expect(index?.sql).toContain("idx_work_products_primary_active");
} finally {
appDb.close();
}
});
it("rejects direct SQL writes that violate the primary invariant", async () => {
const { db, companyId, taskId } = await createHarness();
try {
createWorkProduct(db, companyId, taskId, "first", true);
expect(() => db.queryAll(
`insert into work_products (
id, company_id, task_id, type, provider, external_id, title, status, review_state, is_primary, created_at, updated_at
) values (
'direct-primary', ?, ?, 'file', 'filesystem', 'direct.md', 'Direct', 'active', 'none', 1, '2026-04-24T00:00:00.000Z', '2026-04-24T00:00:00.000Z'
)`,
companyId,
taskId,
)).toThrow(/UNIQUE constraint failed/);
} finally {
db.close();
}
});
it("moves primary status when PATCH promotes another work product", async () => {
const { db, companyId, taskId } = await createHarness();
try {
const firstId = createWorkProduct(db, companyId, taskId, "first", true);
const secondId = createWorkProduct(db, companyId, taskId, "second", false);
db.core.documents.updateWorkProduct(secondId, companyId, { isPrimary: true });
expect(listPrimaryIds(db, companyId, taskId)).toEqual([secondId]);
const first = db.queryAll<{ isPrimary: number }>("select is_primary as isPrimary from work_products where id = ?", firstId)[0];
expect(first?.isPrimary).toBe(0);
} finally {
db.close();
}
});
it("keeps the primary invariant when create inserts a new primary", async () => {
const { db, companyId, taskId } = await createHarness();
try {
const firstId = createWorkProduct(db, companyId, taskId, "first", true);
const secondId = createWorkProduct(db, companyId, taskId, "second", true);
expect(listPrimaryIds(db, companyId, taskId)).toEqual([secondId]);
const first = db.queryAll<{ isPrimary: number }>("select is_primary as isPrimary from work_products where id = ?", firstId)[0];
expect(first?.isPrimary).toBe(0);
} finally {
db.close();
}
});
it("moves primary status when a primary work product changes type", async () => {
const { db, companyId, taskId } = await createHarness();
try {
const firstId = createWorkProduct(db, companyId, taskId, "first", true, "file");
const secondId = createWorkProduct(db, companyId, taskId, "second", true, "link");
db.core.documents.updateWorkProduct(firstId, companyId, { type: "link" });
expect(listPrimaryIdsByType(db, companyId, taskId, "link")).toEqual([firstId]);
const second = db.queryAll<{ isPrimary: number }>("select is_primary as isPrimary from work_products where id = ?", secondId)[0];
expect(second?.isPrimary).toBe(0);
} finally {
db.close();
}
});
});