fix(core): split the domain project field from the RLS partition column (#2165)
## Problem Migration 0006 made `project_id` the RLS isolation partition on every `project`-schema table — stamped by a BEFORE INSERT trigger from the `fusion.project_id` session GUC, with every PK/unique/FK rewritten to composite `(project_id, …)`. Eleven tables **also** carried a caller-supplied domain `projectId` on their TS types and wrote that domain value into the same physical column. When the domain value differs from the session GUC, the parent row lands in the domain partition while child rows (`research_run_events`, `experiment_session_records`, `eval_task_results`, …) land in the session partition — and the composite FK fails with SQLSTATE 23503. Appending an event to a project-owned research run could not persist. ## Fix **Decision (operator): separate domain column; `project_id` stays the partition.** - **Migration `0011_owner_project_id.sql`** adds a nullable `owner_project_id` domain column to the 11 conflated tables (`research_runs`, `experiment_sessions`, `todo_lists`, `eval_runs`, `chat_sessions`, `chat_rooms`, `ai_sessions`, `chat_token_usage`, `project_insights`, `project_insight_runs`, `cli_sessions`), backfills it from `project_id` (identical in production, so exact; the `__legacy_unscoped__` sentinel backfills to NULL), and indexes it. Idempotent, `to_regclass`-guarded per the 0007 pattern. - **Stores** (`async-research-store`, `async-experiment-session-store`, `async-todo-store`, `async-chat-store`, `async-ai-session-store`, `async-eval-store`, `async-insight-store`, `cli-session-store`, …) stop writing `project_id` entirely — the trigger/GUC owns the partition — and map their domain `projectId` field to `owner_project_id` for both reads and filters. TS types unchanged. - **Applier** registers `OWNER_PROJECT_ID_SPLIT_VERSION = "0011"` and advances `SCHEMA_BASELINE_VERSION`. ## Verification (re-run independently of the implementing agent) - Core `tsc --noEmit`: exit 0 · `pnpm lint`: exit 0 · `pnpm check:changesets`: exit 0 · `pnpm test:gate`: 185/185 - Full postgres suite: **5 failed / 807 passed** vs a **7 / 804** baseline — the two conflation round-trips (`satellite-db-injected-stores` ResearchStore + ExperimentSessionStore) go green, zero new failures. The remaining 5 are pre-existing unbound-harness `__meta`/identity failures, unrelated to this change. 🤖 Generated with [Claude Code](https://claude.com/claude-code) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Bug Fixes** * Corrected project-scoped persistence and queries across AI sessions, chats (rooms + token usage), evaluations/experiments, insights, research, and todos by separating domain ownership from RLS partitioning. * Prevented foreign-key and row-level security violations when storing or retrieving project-scoped data, including legacy records. * **Database / New Features** * Added migration 0011 introducing `owner_project_id` and backfilling existing rows to preserve ownership while improving isolation. * **Tests** * Updated migration-parity coverage to include the new baseline step. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
7
.changeset/owner-project-id-split.md
Normal file
7
.changeset/owner-project-id-split.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Fix cross-project data mixups by separating a record's owning project from PostgreSQL isolation.
|
||||
category: fix
|
||||
dev: Migration 0011 adds `owner_project_id` to research_runs, experiment_sessions, todo_lists, eval_runs, chat_sessions, chat_rooms, ai_sessions, chat_token_usage, project_insights, project_insight_runs, and cli_sessions (backfilled from the previously conflated `project_id`, `__legacy_unscoped__` → NULL). Stores now write/read the domain project through `owner_project_id`; `project_id` stays the RLS partition owned by the `fusion_assign_project_id` trigger and the `fusion.project_id` GUC, fixing composite-FK 23503 failures when a caller's domain projectId differed from the session partition.
|
||||
@@ -41,6 +41,7 @@ import {
|
||||
MULTI_PROJECT_CUTOVER_SCHEMA_VERSION,
|
||||
MISSION_FIX_IDEMPOTENCY_VERSION,
|
||||
IMPORT_TRANSLATION_CACHE_VERSION,
|
||||
OWNER_PROJECT_ID_SPLIT_VERSION,
|
||||
PROJECT_OWNERSHIP_SCHEMA_VERSION,
|
||||
SESSION_ADVISOR_ENABLED_SCHEMA_VERSION,
|
||||
SQLITE_SCHEMA_PARITY_VERSION,
|
||||
@@ -99,7 +100,14 @@ describe("schema-applier: immutable migration identities", () => {
|
||||
|
||||
it("keeps the import translation cache assigned to version 0010", () => {
|
||||
expect(IMPORT_TRANSLATION_CACHE_VERSION).toBe("0010");
|
||||
expect(SCHEMA_BASELINE_VERSION).toBe(IMPORT_TRANSLATION_CACHE_VERSION);
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the baseline marker advanced to
|
||||
// 0011; 0010 keeps its immutable identity so its migration cannot be skipped.
|
||||
expect(Number(SCHEMA_BASELINE_VERSION)).toBeGreaterThanOrEqual(Number(IMPORT_TRANSLATION_CACHE_VERSION));
|
||||
});
|
||||
|
||||
it("keeps the owner_project_id domain/partition split assigned to version 0011", () => {
|
||||
expect(OWNER_PROJECT_ID_SPLIT_VERSION).toBe("0011");
|
||||
expect(SCHEMA_BASELINE_VERSION).toBe(OWNER_PROJECT_ID_SPLIT_VERSION);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -953,7 +961,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
const versions = (await ctx.db.execute(sql`
|
||||
SELECT version FROM public.fusion_schema_migrations ORDER BY version
|
||||
`)) as unknown as Array<{ version: string }>;
|
||||
expect(versions.map(({ version }) => version)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", PROJECT_OWNERSHIP_SCHEMA_VERSION, SQLITE_SCHEMA_PARITY_VERSION, SESSION_ADVISOR_ENABLED_SCHEMA_VERSION, MISSION_FIX_IDEMPOTENCY_VERSION, SCHEMA_BASELINE_VERSION]);
|
||||
expect(versions.map(({ version }) => version)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", PROJECT_OWNERSHIP_SCHEMA_VERSION, SQLITE_SCHEMA_PARITY_VERSION, SESSION_ADVISOR_ENABLED_SCHEMA_VERSION, MISSION_FIX_IDEMPOTENCY_VERSION, IMPORT_TRANSLATION_CACHE_VERSION, SCHEMA_BASELINE_VERSION]);
|
||||
expect((await applySchemaBaseline(ctx.db, { pluginHooks: [] })).applied).toBe(false);
|
||||
});
|
||||
|
||||
@@ -977,7 +985,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
applySchemaBaseline(ctx.db, { pluginHooks: [] }),
|
||||
]);
|
||||
expect(results.filter(({ applied }) => applied)).toHaveLength(1);
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", PROJECT_OWNERSHIP_SCHEMA_VERSION, SQLITE_SCHEMA_PARITY_VERSION, SESSION_ADVISOR_ENABLED_SCHEMA_VERSION, MISSION_FIX_IDEMPOTENCY_VERSION, SCHEMA_BASELINE_VERSION]);
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", PROJECT_OWNERSHIP_SCHEMA_VERSION, SQLITE_SCHEMA_PARITY_VERSION, SESSION_ADVISOR_ENABLED_SCHEMA_VERSION, MISSION_FIX_IDEMPOTENCY_VERSION, IMPORT_TRANSLATION_CACHE_VERSION, SCHEMA_BASELINE_VERSION]);
|
||||
});
|
||||
|
||||
it("upgrades a 0001 database by backfilling analytics ownership", async () => {
|
||||
@@ -1013,7 +1021,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
))) as unknown as Array<{ project_id: string }>;
|
||||
expect(rows).toEqual([{ project_id: "project-a" }]);
|
||||
}
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010"]);
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010", "0011"]);
|
||||
});
|
||||
|
||||
/**
|
||||
@@ -1051,7 +1059,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
))) as unknown as Array<{ project_id: string }>;
|
||||
expect(rows).toEqual([{ project_id: "project-a" }]);
|
||||
}
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010"]);
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010", "0011"]);
|
||||
});
|
||||
|
||||
/*
|
||||
@@ -1089,7 +1097,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
"project_auth_users",
|
||||
"task_reviewer_runs",
|
||||
]);
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010"]);
|
||||
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010", "0011"]);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -103,7 +103,8 @@ function rowToSession(row: Record<string, unknown>): AiSessionRow {
|
||||
result: result == null ? null : typeof result === "string" ? result : JSON.stringify(result),
|
||||
thinkingOutput: (row.thinkingOutput as string) ?? "",
|
||||
error: (row.error as string | null) ?? null,
|
||||
projectId: (row.projectId as string | null) ?? null,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the domain projectId now maps to owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011).
|
||||
projectId: (row.ownerProjectId as string | null) ?? null,
|
||||
createdAt: row.createdAt as string,
|
||||
updatedAt: row.updatedAt as string,
|
||||
archived: typeof row.archived === "number" ? row.archived : Number(row.archived ?? 0),
|
||||
@@ -153,7 +154,8 @@ export async function upsertAiSession(handle: QueryHandle, session: AiSessionRow
|
||||
result: resultValue,
|
||||
thinkingOutput: thinking,
|
||||
error: session.error ?? null,
|
||||
...(session.projectId ? { projectId: session.projectId } : {}),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — the trigger/GUC owns the partition (the composite (project_id, id) PK conflict target below is partition-scoped by design).
|
||||
...(session.projectId ? { ownerProjectId: session.projectId } : {}),
|
||||
createdAt: session.createdAt || now,
|
||||
updatedAt: now,
|
||||
})
|
||||
@@ -200,14 +202,14 @@ export async function listActiveAiSessions(
|
||||
inArray(schema.project.aiSessions.status, ["generating", "awaiting_input", "error"]),
|
||||
eq(schema.project.aiSessions.archived, 0),
|
||||
];
|
||||
if (projectId) conditions.push(eq(schema.project.aiSessions.projectId, projectId));
|
||||
if (projectId) conditions.push(eq(schema.project.aiSessions.ownerProjectId, projectId));
|
||||
const rows = await handle
|
||||
.select({
|
||||
id: schema.project.aiSessions.id,
|
||||
type: schema.project.aiSessions.type,
|
||||
status: schema.project.aiSessions.status,
|
||||
title: schema.project.aiSessions.title,
|
||||
projectId: schema.project.aiSessions.projectId,
|
||||
projectId: schema.project.aiSessions.ownerProjectId,
|
||||
updatedAt: schema.project.aiSessions.updatedAt,
|
||||
archived: schema.project.aiSessions.archived,
|
||||
})
|
||||
@@ -234,7 +236,7 @@ export async function listAllAiSessions(
|
||||
if (!options?.includeArchived) {
|
||||
conditions.push(eq(schema.project.aiSessions.archived, 0));
|
||||
}
|
||||
if (projectId) conditions.push(eq(schema.project.aiSessions.projectId, projectId));
|
||||
if (projectId) conditions.push(eq(schema.project.aiSessions.ownerProjectId, projectId));
|
||||
if (options?.type) conditions.push(eq(schema.project.aiSessions.type, options.type));
|
||||
const query = handle
|
||||
.select({
|
||||
@@ -243,7 +245,7 @@ export async function listAllAiSessions(
|
||||
status: schema.project.aiSessions.status,
|
||||
title: schema.project.aiSessions.title,
|
||||
inputPayload: schema.project.aiSessions.inputPayload,
|
||||
projectId: schema.project.aiSessions.projectId,
|
||||
projectId: schema.project.aiSessions.ownerProjectId,
|
||||
updatedAt: schema.project.aiSessions.updatedAt,
|
||||
archived: schema.project.aiSessions.archived,
|
||||
})
|
||||
@@ -263,7 +265,7 @@ export async function listRecoverableAiSessions(
|
||||
const conditions = [
|
||||
inArray(schema.project.aiSessions.status, ["generating", "awaiting_input"]),
|
||||
];
|
||||
if (projectId) conditions.push(eq(schema.project.aiSessions.projectId, projectId));
|
||||
if (projectId) conditions.push(eq(schema.project.aiSessions.ownerProjectId, projectId));
|
||||
const rows = await handle
|
||||
.select()
|
||||
.from(schema.project.aiSessions)
|
||||
|
||||
@@ -44,7 +44,8 @@ function rowToSession(row: Record<string, unknown>): ChatSession {
|
||||
agentId: row.agentId as string,
|
||||
title: (row.title as string | null) ?? null,
|
||||
status: row.status as ChatSessionStatus,
|
||||
projectId: (row.projectId as string | null) ?? null,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the domain projectId now maps to owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011).
|
||||
projectId: (row.ownerProjectId as string | null) ?? null,
|
||||
modelProvider: (row.modelProvider as string | null) ?? null,
|
||||
modelId: (row.modelId as string | null) ?? null,
|
||||
thinkingLevel: (row.thinkingLevel as string | null) ?? null,
|
||||
@@ -75,7 +76,8 @@ function rowToRoom(row: Record<string, unknown>): ChatRoom {
|
||||
name: row.name as string,
|
||||
slug: row.slug as string,
|
||||
description: (row.description as string | null) ?? null,
|
||||
projectId: (row.projectId as string | null) ?? null,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain projectId reads from owner_project_id (see rowToSession).
|
||||
projectId: (row.ownerProjectId as string | null) ?? null,
|
||||
createdBy: (row.createdBy as string | null) ?? null,
|
||||
status: row.status as ChatRoomStatus,
|
||||
// FNXC:Chat-ThinkingLevel 2026-07-13 (merge port): room-level reasoning-effort default.
|
||||
@@ -120,7 +122,8 @@ export async function createChatSession(handle: QueryHandle, session: ChatSessio
|
||||
agentId: session.agentId,
|
||||
title: session.title,
|
||||
status: session.status,
|
||||
projectId: session.projectId,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — the trigger/GUC owns the partition.
|
||||
ownerProjectId: session.projectId,
|
||||
modelProvider: session.modelProvider,
|
||||
modelId: session.modelId,
|
||||
thinkingLevel: session.thinkingLevel ?? null,
|
||||
@@ -153,7 +156,7 @@ export async function listChatSessions(
|
||||
options?: { projectId?: string; agentId?: string; status?: ChatSessionStatus },
|
||||
): Promise<ChatSession[]> {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options?.projectId) conditions.push(eq(schema.project.chatSessions.projectId, options.projectId));
|
||||
if (options?.projectId) conditions.push(eq(schema.project.chatSessions.ownerProjectId, options.projectId));
|
||||
if (options?.agentId) conditions.push(eq(schema.project.chatSessions.agentId, options.agentId));
|
||||
if (options?.status) conditions.push(eq(schema.project.chatSessions.status, options.status));
|
||||
const query = handle
|
||||
@@ -283,7 +286,7 @@ export async function createChatRoom(
|
||||
name: room.name,
|
||||
slug: room.slug,
|
||||
description: room.description,
|
||||
projectId: room.projectId,
|
||||
ownerProjectId: room.projectId,
|
||||
createdBy: room.createdBy,
|
||||
status: room.status,
|
||||
thinkingLevel: room.thinkingLevel ?? null,
|
||||
@@ -325,9 +328,9 @@ export async function getChatRoomBySlug(
|
||||
): Promise<ChatRoom | undefined> {
|
||||
const conditions = [eq(schema.project.chatRooms.slug, slug)];
|
||||
if (projectId !== null) {
|
||||
conditions.push(eq(schema.project.chatRooms.projectId, projectId));
|
||||
conditions.push(eq(schema.project.chatRooms.ownerProjectId, projectId));
|
||||
} else {
|
||||
conditions.push(isNull(schema.project.chatRooms.projectId));
|
||||
conditions.push(isNull(schema.project.chatRooms.ownerProjectId));
|
||||
}
|
||||
const rows = await handle
|
||||
.select()
|
||||
@@ -344,7 +347,7 @@ export async function listChatRooms(
|
||||
options?: { projectId?: string; status?: ChatRoomStatus },
|
||||
): Promise<ChatRoom[]> {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options?.projectId) conditions.push(eq(schema.project.chatRooms.projectId, options.projectId));
|
||||
if (options?.projectId) conditions.push(eq(schema.project.chatRooms.ownerProjectId, options.projectId));
|
||||
if (options?.status) conditions.push(eq(schema.project.chatRooms.status, options.status));
|
||||
const query = handle
|
||||
.select()
|
||||
@@ -857,7 +860,7 @@ export async function listChatRoomsForAgent(
|
||||
|
||||
const conditions: ReturnType<typeof eq>[] = [inArray(schema.project.chatRooms.id, memberRoomIds)];
|
||||
if (options?.status) conditions.push(eq(schema.project.chatRooms.status, options.status));
|
||||
if (options?.projectId) conditions.push(eq(schema.project.chatRooms.projectId, options.projectId));
|
||||
if (options?.projectId) conditions.push(eq(schema.project.chatRooms.ownerProjectId, options.projectId));
|
||||
|
||||
const rows = await handle
|
||||
.select()
|
||||
@@ -986,7 +989,7 @@ export async function findLatestActiveChatSessionForTarget(
|
||||
eq(schema.project.chatSessions.agentId, normalizedAgentId),
|
||||
];
|
||||
if (options.projectId && options.projectId.trim()) {
|
||||
baseConditions.push(eq(schema.project.chatSessions.projectId, options.projectId.trim()));
|
||||
baseConditions.push(eq(schema.project.chatSessions.ownerProjectId, options.projectId.trim()));
|
||||
}
|
||||
|
||||
// Model-targeted: exact provider+model match.
|
||||
|
||||
@@ -55,7 +55,8 @@ type QueryHandle = AsyncDataLayer["db"] | DbTransaction;
|
||||
function rowToRun(row: Record<string, unknown>): EvalRun {
|
||||
return {
|
||||
id: String(row.id),
|
||||
projectId: String(row.projectId),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the domain projectId now maps to owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011).
|
||||
projectId: String(row.ownerProjectId ?? ""),
|
||||
status: row.status as EvalRunStatus,
|
||||
trigger: row.trigger as EvalRun["trigger"],
|
||||
scope: String(row.scope),
|
||||
@@ -123,7 +124,8 @@ export async function createEvalRun(
|
||||
): Promise<EvalRun> {
|
||||
await handle.insert(schema.project.evalRuns).values({
|
||||
id: run.id,
|
||||
projectId: run.projectId,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — domain writes into the partition desynced parent/child partitions and broke composite FKs (23503).
|
||||
ownerProjectId: run.projectId,
|
||||
status: "pending",
|
||||
trigger: run.trigger,
|
||||
scope: run.scope,
|
||||
@@ -144,7 +146,7 @@ export async function createEvalRun(
|
||||
});
|
||||
return rowToRun({
|
||||
id: run.id,
|
||||
projectId: run.projectId,
|
||||
ownerProjectId: run.projectId,
|
||||
status: "pending",
|
||||
trigger: run.trigger,
|
||||
scope: run.scope,
|
||||
@@ -181,7 +183,7 @@ export async function getEvalRun(handle: QueryHandle, id: string): Promise<EvalR
|
||||
*/
|
||||
export async function listEvalRuns(handle: QueryHandle, options: EvalRunListOptions = {}): Promise<EvalRun[]> {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options.projectId) conditions.push(eq(schema.project.evalRuns.projectId, options.projectId));
|
||||
if (options.projectId) conditions.push(eq(schema.project.evalRuns.ownerProjectId, options.projectId));
|
||||
if (options.status) conditions.push(eq(schema.project.evalRuns.status, options.status));
|
||||
if (options.trigger) conditions.push(eq(schema.project.evalRuns.trigger, options.trigger));
|
||||
let query = handle
|
||||
@@ -232,17 +234,22 @@ export async function upsertEvalTaskResult(
|
||||
handle: QueryHandle,
|
||||
result: EvalTaskResult,
|
||||
): Promise<void> {
|
||||
// FNXC:MultiProjectIsolation 2026-07-16-08:05: children inherit the parent run's
|
||||
// project_id partition explicitly. The GUC-driven trigger only covers bound sessions;
|
||||
// unbound/bypass handles (tests, admin, maintenance) or a GUC naming another partition
|
||||
// would otherwise violate the composite (project_id, run_id) FK with SQLSTATE 23503.
|
||||
// Inherit the PARTITION column (projectId), never the domain field (ownerProjectId).
|
||||
const runRows = await handle
|
||||
.select({ projectId: schema.project.evalRuns.projectId })
|
||||
.from(schema.project.evalRuns)
|
||||
.where(eq(schema.project.evalRuns.id, result.runId))
|
||||
.limit(1);
|
||||
const projectId = runRows[0]?.projectId;
|
||||
if (!projectId) throw new Error(`Eval run not found: ${result.runId}`);
|
||||
const parentProjectId = runRows[0]?.projectId;
|
||||
if (!parentProjectId) throw new Error(`Eval run not found: ${result.runId}`);
|
||||
await handle
|
||||
.insert(schema.project.evalTaskResults)
|
||||
.values({
|
||||
projectId,
|
||||
projectId: parentProjectId,
|
||||
id: result.id,
|
||||
runId: result.runId,
|
||||
taskId: result.taskId,
|
||||
@@ -346,13 +353,16 @@ export async function appendEvalRunEvent(
|
||||
input: { id: string; runId: string; type: string; message: string; status?: EvalRunStatus; taskId?: string; metadata?: Record<string, unknown> },
|
||||
): Promise<EvalRunEvent> {
|
||||
return layer.transactionImmediate(async (tx) => {
|
||||
// FNXC:MultiProjectIsolation 2026-07-16-08:05: inherit the parent run's project_id
|
||||
// partition explicitly (see upsertEvalTaskResult) — the ambient GUC is not guaranteed
|
||||
// to match for unbound/bypass handles, and a mismatch breaks the composite parent FK.
|
||||
const runRows = await tx
|
||||
.select({ projectId: schema.project.evalRuns.projectId })
|
||||
.from(schema.project.evalRuns)
|
||||
.where(eq(schema.project.evalRuns.id, input.runId))
|
||||
.limit(1);
|
||||
const projectId = runRows[0]?.projectId;
|
||||
if (!projectId) throw new Error(`Eval run not found: ${input.runId}`);
|
||||
const parentProjectId = runRows[0]?.projectId;
|
||||
if (!parentProjectId) throw new Error(`Eval run not found: ${input.runId}`);
|
||||
const seqRows = await tx
|
||||
.select({ maxSeq: sql<number | null>`max(${schema.project.evalRunEvents.seq})` })
|
||||
.from(schema.project.evalRunEvents)
|
||||
@@ -360,7 +370,7 @@ export async function appendEvalRunEvent(
|
||||
const seq = (seqRows[0]?.maxSeq ?? 0) + 1;
|
||||
const createdAt = new Date().toISOString();
|
||||
await tx.insert(schema.project.evalRunEvents).values({
|
||||
projectId,
|
||||
projectId: parentProjectId,
|
||||
id: input.id,
|
||||
runId: input.runId,
|
||||
seq,
|
||||
|
||||
@@ -35,7 +35,8 @@ function rowToSession(row: Record<string, unknown>): ExperimentSession {
|
||||
return {
|
||||
id: row.id as string,
|
||||
name: row.name as string,
|
||||
projectId: (row.projectId as string | null) ?? undefined,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the domain projectId now maps to owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011).
|
||||
projectId: (row.ownerProjectId as string | null) ?? undefined,
|
||||
status: row.status as ExperimentSessionStatus,
|
||||
metric: (metricRaw as ExperimentSession["metric"]) ?? { name: "unknown", direction: "maximize" },
|
||||
currentSegment: Number(row.currentSegment ?? 1),
|
||||
@@ -74,7 +75,8 @@ export async function createExperimentSession(
|
||||
await handle.insert(schema.project.experimentSessions).values({
|
||||
id: session.id,
|
||||
name: session.name,
|
||||
projectId: session.projectId ?? null,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain project goes to owner_project_id; project_id (the RLS partition) is owned by the fusion_assign_project_id trigger/GUC.
|
||||
ownerProjectId: session.projectId ?? null,
|
||||
status: session.status,
|
||||
metric: JSON.stringify(session.metric),
|
||||
currentSegment: session.currentSegment,
|
||||
@@ -109,7 +111,7 @@ export async function getExperimentSession(handle: QueryHandle, id: string): Pro
|
||||
export async function listExperimentSessions(handle: QueryHandle, options: ExperimentSessionListOptions = {}): Promise<ExperimentSession[]> {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options.status) conditions.push(eq(schema.project.experimentSessions.status, options.status));
|
||||
if (options.projectId) conditions.push(eq(schema.project.experimentSessions.projectId, options.projectId));
|
||||
if (options.projectId) conditions.push(eq(schema.project.experimentSessions.ownerProjectId, options.projectId));
|
||||
const query = handle
|
||||
.select()
|
||||
.from(schema.project.experimentSessions)
|
||||
@@ -126,7 +128,7 @@ export async function persistExperimentSession(handle: QueryHandle, session: Exp
|
||||
.update(schema.project.experimentSessions)
|
||||
.set({
|
||||
name: session.name,
|
||||
projectId: session.projectId ?? null,
|
||||
ownerProjectId: session.projectId ?? null,
|
||||
status: session.status,
|
||||
metric: JSON.stringify(session.metric),
|
||||
currentSegment: session.currentSegment,
|
||||
|
||||
@@ -53,7 +53,8 @@ type QueryHandle = AsyncDataLayer["db"] | DbTransaction;
|
||||
function rowToInsight(row: Record<string, unknown>): Insight {
|
||||
return {
|
||||
id: row.id as string,
|
||||
projectId: row.projectId as string,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the domain projectId now maps to owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011).
|
||||
projectId: (row.ownerProjectId as string | null) ?? "",
|
||||
title: row.title as string,
|
||||
content: (row.content as string | null) ?? null,
|
||||
category: row.category as InsightCategory,
|
||||
@@ -69,7 +70,7 @@ function rowToInsight(row: Record<string, unknown>): Insight {
|
||||
function rowToRun(row: Record<string, unknown>): InsightRun {
|
||||
return {
|
||||
id: row.id as string,
|
||||
projectId: row.projectId as string,
|
||||
projectId: (row.ownerProjectId as string | null) ?? "",
|
||||
trigger: row.trigger as InsightRunTrigger,
|
||||
status: row.status as InsightRunStatus,
|
||||
summary: (row.summary as string | null) ?? null,
|
||||
@@ -97,7 +98,8 @@ export async function createInsight(
|
||||
): Promise<void> {
|
||||
await handle.insert(schema.project.projectInsights).values({
|
||||
id: insight.id,
|
||||
projectId: insight.projectId,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — the trigger/GUC owns the partition, and domain writes into it desynced the composite FK partition of project_insight_run_events.
|
||||
ownerProjectId: insight.projectId,
|
||||
title: insight.title,
|
||||
content: insight.content ?? null,
|
||||
category: insight.category,
|
||||
@@ -127,7 +129,7 @@ export async function getInsight(handle: QueryHandle, id: string): Promise<Insig
|
||||
*/
|
||||
export async function listInsights(handle: QueryHandle, options: InsightListOptions = {}): Promise<Insight[]> {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsights.projectId, options.projectId));
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsights.ownerProjectId, options.projectId));
|
||||
if (options.category !== undefined) conditions.push(eq(schema.project.projectInsights.category, options.category));
|
||||
if (options.status !== undefined) conditions.push(eq(schema.project.projectInsights.status, options.status));
|
||||
if (options.runId !== undefined) conditions.push(eq(schema.project.projectInsights.lastRunId, options.runId));
|
||||
@@ -154,7 +156,7 @@ export async function upsertInsight(
|
||||
.from(schema.project.projectInsights)
|
||||
.where(
|
||||
and(
|
||||
eq(schema.project.projectInsights.projectId, projectId),
|
||||
eq(schema.project.projectInsights.ownerProjectId, projectId),
|
||||
eq(schema.project.projectInsights.fingerprint, input.fingerprint),
|
||||
),
|
||||
);
|
||||
@@ -219,7 +221,7 @@ export async function createInsightRun(
|
||||
): Promise<InsightRun> {
|
||||
await handle.insert(schema.project.projectInsightRuns).values({
|
||||
id: run.id,
|
||||
projectId: run.projectId,
|
||||
ownerProjectId: run.projectId,
|
||||
trigger: run.trigger,
|
||||
status: "pending",
|
||||
summary: null,
|
||||
@@ -270,7 +272,7 @@ export async function getInsightRun(handle: QueryHandle, id: string): Promise<In
|
||||
*/
|
||||
export async function listInsightRuns(handle: QueryHandle, options: InsightRunListOptions = {}): Promise<InsightRun[]> {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsightRuns.projectId, options.projectId));
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsightRuns.ownerProjectId, options.projectId));
|
||||
if (options.status !== undefined) conditions.push(eq(schema.project.projectInsightRuns.status, options.status));
|
||||
if (options.trigger !== undefined) conditions.push(eq(schema.project.projectInsightRuns.trigger, options.trigger));
|
||||
const query = handle
|
||||
@@ -295,7 +297,7 @@ export async function findActiveInsightRun(
|
||||
.from(schema.project.projectInsightRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(schema.project.projectInsightRuns.projectId, projectId),
|
||||
eq(schema.project.projectInsightRuns.ownerProjectId, projectId),
|
||||
eq(schema.project.projectInsightRuns.trigger, trigger),
|
||||
inArray(schema.project.projectInsightRuns.status, ["pending", "running"]),
|
||||
),
|
||||
@@ -315,21 +317,32 @@ export async function appendInsightRunEvent(
|
||||
let seq = 1;
|
||||
const createdAt = new Date().toISOString();
|
||||
await layer.transactionImmediate(async (tx) => {
|
||||
// FNXC:MultiProjectIsolation 2026-07-16-09:10: read the parent run's PARTITION column
|
||||
// (projectId, not ownerProjectId) and stamp it explicitly on the child event insert.
|
||||
// The ambient fusion.project_id GUC is not guaranteed to match the parent's partition —
|
||||
// unbound/bypass handles (tests, admin, maintenance) would otherwise get the trigger's
|
||||
// default (__legacy_unscoped__ or another ambient partition) and fail the composite
|
||||
// (project_id, run_id) FK with SQLSTATE 23503. The trigger only rewrites blank values,
|
||||
// so an explicit non-blank project_id is preserved, and RLS WITH CHECK still passes for
|
||||
// bound sessions (they can only read their own partition's parent run anyway).
|
||||
const runRows = await tx
|
||||
.select({ projectId: schema.project.projectInsightRuns.projectId })
|
||||
.select({
|
||||
id: schema.project.projectInsightRuns.id,
|
||||
projectId: schema.project.projectInsightRuns.projectId,
|
||||
})
|
||||
.from(schema.project.projectInsightRuns)
|
||||
.where(eq(schema.project.projectInsightRuns.id, input.runId))
|
||||
.limit(1);
|
||||
const projectId = runRows[0]?.projectId;
|
||||
if (!projectId) throw new Error(`Insight run not found: ${input.runId}`);
|
||||
if (!runRows[0]) throw new Error(`Insight run not found: ${input.runId}`);
|
||||
const parentPartitionId = runRows[0].projectId as string;
|
||||
const seqRows = await tx
|
||||
.select({ nextSeq: sql<number>`coalesce(max(${schema.project.projectInsightRunEvents.seq}), 0) + 1` })
|
||||
.from(schema.project.projectInsightRunEvents)
|
||||
.where(eq(schema.project.projectInsightRunEvents.runId, input.runId));
|
||||
seq = Number(seqRows[0]?.nextSeq ?? 1);
|
||||
await tx.insert(schema.project.projectInsightRunEvents).values({
|
||||
projectId,
|
||||
id: input.id,
|
||||
projectId: parentPartitionId,
|
||||
runId: input.runId,
|
||||
seq,
|
||||
type: input.type,
|
||||
@@ -465,7 +478,7 @@ export async function updateInsightRun(
|
||||
|
||||
function buildInsightCountConditions(options: Pick<InsightListOptions, "projectId" | "category" | "status" | "runId">): ReturnType<typeof eq>[] {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsights.projectId, options.projectId));
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsights.ownerProjectId, options.projectId));
|
||||
if (options.category !== undefined) conditions.push(eq(schema.project.projectInsights.category, options.category));
|
||||
if (options.status !== undefined) conditions.push(eq(schema.project.projectInsights.status, options.status));
|
||||
if (options.runId !== undefined) conditions.push(eq(schema.project.projectInsights.lastRunId, options.runId));
|
||||
@@ -485,7 +498,7 @@ export async function countInsights(handle: QueryHandle, options: Omit<InsightLi
|
||||
|
||||
function buildRunCountConditions(options: Pick<InsightRunListOptions, "projectId" | "status" | "trigger">): ReturnType<typeof eq>[] {
|
||||
const conditions: ReturnType<typeof eq>[] = [];
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsightRuns.projectId, options.projectId));
|
||||
if (options.projectId !== undefined) conditions.push(eq(schema.project.projectInsightRuns.ownerProjectId, options.projectId));
|
||||
if (options.status !== undefined) conditions.push(eq(schema.project.projectInsightRuns.status, options.status));
|
||||
if (options.trigger !== undefined) conditions.push(eq(schema.project.projectInsightRuns.trigger, options.trigger));
|
||||
return conditions;
|
||||
@@ -517,7 +530,7 @@ export async function listStalePendingRuns(
|
||||
inArray(schema.project.projectInsightRuns.status, ["pending", "running"]),
|
||||
lte(sql`coalesce(${schema.project.projectInsightRuns.startedAt}, ${schema.project.projectInsightRuns.createdAt})`, olderThanIso),
|
||||
];
|
||||
if (options.projectId) conditions.push(eq(schema.project.projectInsightRuns.projectId, options.projectId));
|
||||
if (options.projectId) conditions.push(eq(schema.project.projectInsightRuns.ownerProjectId, options.projectId));
|
||||
const rows = await handle
|
||||
.select()
|
||||
.from(schema.project.projectInsightRuns)
|
||||
|
||||
@@ -57,7 +57,8 @@ function rowToRun(row: Record<string, unknown>): ResearchRun {
|
||||
query: row.query as string,
|
||||
topic: (row.topic as string | null) ?? undefined,
|
||||
status: normalizeStatus((row.status as ResearchRunStatus | "pending") ?? "queued"),
|
||||
projectId: (row.projectId as string | null) ?? undefined,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the domain projectId now maps to owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011).
|
||||
projectId: (row.ownerProjectId as string | null) ?? undefined,
|
||||
trigger: (row.trigger as string | null) ?? undefined,
|
||||
providerConfig: row.providerConfig as ResearchRun["providerConfig"],
|
||||
sources: (row.sources as ResearchSource[]) ?? [],
|
||||
@@ -99,7 +100,8 @@ export async function createResearchRun(
|
||||
query: run.query,
|
||||
topic: run.topic ?? null,
|
||||
status: run.status,
|
||||
projectId: run.projectId ?? null,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — writing domain data into the partition put parents and children in different partitions and broke the composite FKs (23503).
|
||||
ownerProjectId: run.projectId ?? null,
|
||||
trigger: run.trigger ?? null,
|
||||
providerConfig: run.providerConfig ?? null,
|
||||
sources: run.sources,
|
||||
@@ -141,7 +143,7 @@ export async function persistResearchRun(handle: QueryHandle, run: ResearchRun):
|
||||
query: run.query,
|
||||
topic: run.topic ?? null,
|
||||
status: run.status,
|
||||
projectId: run.projectId ?? null,
|
||||
ownerProjectId: run.projectId ?? null,
|
||||
trigger: run.trigger ?? null,
|
||||
providerConfig: run.providerConfig ?? null,
|
||||
sources: run.sources,
|
||||
@@ -251,7 +253,7 @@ export async function getActiveResearchRun(
|
||||
.from(schema.project.researchRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(schema.project.researchRuns.projectId, projectId),
|
||||
eq(schema.project.researchRuns.ownerProjectId, projectId),
|
||||
eq(schema.project.researchRuns.trigger, trigger),
|
||||
inArray(schema.project.researchRuns.status, ["queued", "running", "cancelling", "retry_waiting"]),
|
||||
),
|
||||
|
||||
@@ -84,7 +84,8 @@ function rowToTodoItem(row: TodoItemRow): TodoItem {
|
||||
|
||||
const todoListColumns = {
|
||||
id: schema.project.todoLists.id,
|
||||
projectId: schema.project.todoLists.projectId,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the TodoList domain projectId reads from owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011).
|
||||
projectId: schema.project.todoLists.ownerProjectId,
|
||||
title: schema.project.todoLists.title,
|
||||
createdAt: schema.project.todoLists.createdAt,
|
||||
updatedAt: schema.project.todoLists.updatedAt,
|
||||
@@ -111,7 +112,8 @@ export async function createTodoList(
|
||||
): Promise<TodoList> {
|
||||
await handle.insert(schema.project.todoLists).values({
|
||||
id: list.id,
|
||||
projectId: list.projectId,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — the trigger/GUC owns the partition, and domain writes into it desynced the composite FK partitions of todo_items.
|
||||
ownerProjectId: list.projectId,
|
||||
title: list.title,
|
||||
createdAt: list.createdAt,
|
||||
updatedAt: list.updatedAt,
|
||||
@@ -137,7 +139,7 @@ export async function listTodoLists(handle: QueryHandle, projectId: string): Pro
|
||||
const rows = await handle
|
||||
.select(todoListColumns)
|
||||
.from(schema.project.todoLists)
|
||||
.where(eq(schema.project.todoLists.projectId, projectId))
|
||||
.where(eq(schema.project.todoLists.ownerProjectId, projectId))
|
||||
.orderBy(asc(schema.project.todoLists.createdAt), asc(schema.project.todoLists.id));
|
||||
return rows.map((row) => rowToTodoList(row as TodoListRow));
|
||||
}
|
||||
@@ -181,13 +183,18 @@ export async function createTodoItem(
|
||||
handle: QueryHandle,
|
||||
item: { id: string; listId: string; text: string; completed: boolean; completedAt: string | null; sortOrder: number | undefined; createdAt: string; updatedAt: string },
|
||||
): Promise<TodoItem> {
|
||||
// FNXC:MultiProjectIsolation 2026-07-16-00:10: read the parent list's project_id
|
||||
// partition (NOT owner_project_id) so the item explicitly inherits it. The ambient
|
||||
// fusion.project_id GUC is not guaranteed to match the list's partition — unbound or
|
||||
// RLS-bypass handles can read a list from any partition, and letting the trigger stamp
|
||||
// the GUC would put the item in a different partition and break the composite
|
||||
// (project_id, list_id) FK with SQLSTATE 23503.
|
||||
const listRows = await handle
|
||||
.select({ projectId: schema.project.todoLists.projectId })
|
||||
.select({ id: schema.project.todoLists.id, projectId: schema.project.todoLists.projectId })
|
||||
.from(schema.project.todoLists)
|
||||
.where(eq(schema.project.todoLists.id, item.listId))
|
||||
.limit(1);
|
||||
const projectId = listRows[0]?.projectId;
|
||||
if (!projectId) throw new Error(`Todo list not found: ${item.listId}`);
|
||||
if (!listRows[0]) throw new Error(`Todo list not found: ${item.listId}`);
|
||||
let sortOrder = item.sortOrder;
|
||||
if (sortOrder === undefined) {
|
||||
const maxRows = await handle
|
||||
@@ -197,7 +204,10 @@ export async function createTodoItem(
|
||||
sortOrder = (maxRows[0]?.maxSortOrder ?? -1) + 1;
|
||||
}
|
||||
await handle.insert(schema.project.todoItems).values({
|
||||
projectId,
|
||||
// FNXC:MultiProjectIsolation 2026-07-16-00:10: explicit non-blank project_id is safe —
|
||||
// the assign trigger only rewrites blanks, and RLS WITH CHECK holds because a bound
|
||||
// session can only have read a parent list from its own partition.
|
||||
projectId: listRows[0].projectId,
|
||||
id: item.id,
|
||||
listId: item.listId,
|
||||
text: item.text,
|
||||
|
||||
@@ -621,8 +621,9 @@ export class ChatStore extends EventEmitter<ChatStoreEvents> {
|
||||
};
|
||||
const layer = this.asyncLayer;
|
||||
/* FNXC:PostgresChatUsage 2026-07-14-18:49: Token accounting is durable before a chat turn reports completion; callers await the insert so shutdown and immediate analytics cannot lose or race the record. */
|
||||
/* FNXC:MultiProjectIsolation 2026-07-15-23:40: the record's domain projectId is written to owner_project_id; project_id (the RLS partition) is omitted so the fusion_assign_project_id trigger/GUC owns it (migration 0011). */
|
||||
await layer.db.execute(sql`INSERT INTO project.chat_token_usage (
|
||||
id, source_kind, chat_session_id, room_id, message_id, project_id, agent_id,
|
||||
id, source_kind, chat_session_id, room_id, message_id, owner_project_id, agent_id,
|
||||
model_provider, model_id, input_tokens, output_tokens, cached_tokens,
|
||||
cache_write_tokens, total_tokens, created_at
|
||||
) VALUES (
|
||||
@@ -644,7 +645,7 @@ export class ChatStore extends EventEmitter<ChatStoreEvents> {
|
||||
return rows.map((row) => ({
|
||||
id: row.id, sourceKind: row.sourceKind as ChatTokenUsageSourceKind,
|
||||
chatSessionId: row.chatSessionId, roomId: row.roomId, messageId: row.messageId,
|
||||
projectId: row.projectId, agentId: row.agentId, modelProvider: row.modelProvider,
|
||||
projectId: row.ownerProjectId, agentId: row.agentId, modelProvider: row.modelProvider,
|
||||
modelId: row.modelId, inputTokens: row.inputTokens, outputTokens: row.outputTokens,
|
||||
cachedTokens: row.cachedTokens, cacheWriteTokens: row.cacheWriteTokens,
|
||||
totalTokens: row.totalTokens, createdAt: row.createdAt,
|
||||
|
||||
@@ -45,7 +45,8 @@ function rowToSession(row: CliSessionRow): CliSession {
|
||||
taskId: row.taskId,
|
||||
chatSessionId: row.chatSessionId,
|
||||
purpose: row.purpose as CliSessionPurpose,
|
||||
projectId: row.projectId,
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: the domain projectId now maps to owner_project_id; project_id is the trigger/GUC-owned RLS partition (migration 0011). The store always writes it, so hydrated rows are non-null.
|
||||
projectId: row.ownerProjectId ?? "",
|
||||
adapterId: row.adapterId,
|
||||
agentState: row.agentState as CliAgentState,
|
||||
terminationReason: row.terminationReason as CliTerminationReason | null,
|
||||
@@ -77,7 +78,7 @@ export class CliSessionStore extends EventEmitter<CliSessionStoreEvents> {
|
||||
const rows = await layer.db
|
||||
.select()
|
||||
.from(schema.project.cliSessions)
|
||||
.where(eq(schema.project.cliSessions.projectId, projectId))
|
||||
.where(eq(schema.project.cliSessions.ownerProjectId, projectId))
|
||||
.orderBy(desc(schema.project.cliSessions.updatedAt));
|
||||
for (const row of rows) store.sessions.set(row.id, rowToSession(row));
|
||||
return store;
|
||||
@@ -143,8 +144,12 @@ export class CliSessionStore extends EventEmitter<CliSessionStoreEvents> {
|
||||
updatedAt: now,
|
||||
};
|
||||
this.sessions.set(session.id, session);
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the session's domain project to
|
||||
// owner_project_id and never project_id — the trigger/GUC owns the RLS partition.
|
||||
const { projectId: ownerProjectId, ...columns } = session;
|
||||
this.enqueue(() => this.layer.db.insert(schema.project.cliSessions).values({
|
||||
...session,
|
||||
...columns,
|
||||
ownerProjectId,
|
||||
autonomyPosture: session.autonomyPosture ? JSON.stringify(session.autonomyPosture) : null,
|
||||
}));
|
||||
this.emit("cli-session:created", session);
|
||||
@@ -203,7 +208,7 @@ export class CliSessionStore extends EventEmitter<CliSessionStoreEvents> {
|
||||
})
|
||||
.where(and(
|
||||
eq(schema.project.cliSessions.id, id),
|
||||
eq(schema.project.cliSessions.projectId, this.projectId),
|
||||
eq(schema.project.cliSessions.ownerProjectId, this.projectId),
|
||||
)));
|
||||
this.emit("cli-session:updated", updated);
|
||||
return updated;
|
||||
@@ -215,7 +220,7 @@ export class CliSessionStore extends EventEmitter<CliSessionStoreEvents> {
|
||||
.delete(schema.project.cliSessions)
|
||||
.where(and(
|
||||
eq(schema.project.cliSessions.id, id),
|
||||
eq(schema.project.cliSessions.projectId, this.projectId),
|
||||
eq(schema.project.cliSessions.ownerProjectId, this.projectId),
|
||||
)));
|
||||
this.emit("cli-session:deleted", id);
|
||||
return true;
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
-- 0011_owner_project_id.sql
|
||||
--
|
||||
-- FNXC:MultiProjectIsolation 2026-07-15-23:40:
|
||||
-- Separate the DOMAIN "project" field from the PostgreSQL isolation partition.
|
||||
-- Migration 0006 made `project_id` the RLS partition column on every project-schema
|
||||
-- table (owned by the fusion_assign_project_id BEFORE INSERT trigger + the
|
||||
-- fusion.project_id session GUC, with composite PK/unique/FK keys). A handful of
|
||||
-- tables ALSO carried a caller-supplied domain "projectId" on their TS types and
|
||||
-- wrote that domain value into the same physical `project_id` column. When the
|
||||
-- domain value differed from the session GUC, the parent row landed in the domain
|
||||
-- partition while child rows (research_run_events, experiment_session_records,
|
||||
-- eval_task_results, ...) landed in the session partition, violating the composite
|
||||
-- FKs with SQLSTATE 23503.
|
||||
--
|
||||
-- This migration adds a separate nullable `owner_project_id` domain column to the
|
||||
-- conflated tables. `project_id` remains the isolation partition and stays owned by
|
||||
-- the trigger/GUC; stores stop writing it and read/write the domain field through
|
||||
-- `owner_project_id` instead.
|
||||
--
|
||||
-- Backfill: in production the domain value and the partition were written with the
|
||||
-- same id, so `owner_project_id = project_id` is correct. `'__legacy_unscoped__'`
|
||||
-- is the trigger's "no session project bound" sentinel and never a real domain
|
||||
-- project, and the domain field was nullable, so the sentinel backfills to NULL.
|
||||
--
|
||||
-- Idempotent and re-runnable: to_regclass-guarded (a failed early baseline can
|
||||
-- leave a migration marker before every baseline table exists — absent tables
|
||||
-- receive the column when the idempotent baseline is materialized on the next
|
||||
-- recovery pass, matching the 0007 pattern), ADD COLUMN IF NOT EXISTS, and a
|
||||
-- backfill scoped to owner_project_id IS NULL.
|
||||
|
||||
DO $$
|
||||
DECLARE
|
||||
tbl text;
|
||||
BEGIN
|
||||
FOREACH tbl IN ARRAY ARRAY[
|
||||
'research_runs',
|
||||
'experiment_sessions',
|
||||
'todo_lists',
|
||||
'eval_runs',
|
||||
'chat_sessions',
|
||||
'chat_rooms',
|
||||
'ai_sessions',
|
||||
'chat_token_usage',
|
||||
'project_insights',
|
||||
'project_insight_runs',
|
||||
'cli_sessions'
|
||||
] LOOP
|
||||
IF to_regclass(format('project.%I', tbl)) IS NOT NULL THEN
|
||||
EXECUTE format('ALTER TABLE project.%I ADD COLUMN IF NOT EXISTS owner_project_id text', tbl);
|
||||
EXECUTE format(
|
||||
'UPDATE project.%I SET owner_project_id = NULLIF(project_id, %L) WHERE owner_project_id IS NULL',
|
||||
tbl,
|
||||
'__legacy_unscoped__'
|
||||
);
|
||||
-- Domain-filter index (stores list/filter by the domain project id).
|
||||
EXECUTE format(
|
||||
'CREATE INDEX IF NOT EXISTS %I ON project.%I (owner_project_id)',
|
||||
'idx_' || tbl || '_owner_project_id',
|
||||
tbl
|
||||
);
|
||||
END IF;
|
||||
END LOOP;
|
||||
END
|
||||
$$;
|
||||
@@ -28,10 +28,10 @@ import { runPluginSchemaInitHooks, DEFAULT_PLUGIN_SCHEMA_INIT_HOOKS, type Plugin
|
||||
|
||||
/** The latest PostgreSQL schema version known to this applier. */
|
||||
/*
|
||||
FNXC:GitHubImportTranslate 2026-07-15-09:30:
|
||||
Advances to 0010 with the import-translation cache. Per-migration identities above stay fixed; only this latest-version marker moves.
|
||||
FNXC:MultiProjectIsolation 2026-07-15-23:40:
|
||||
Advances to 0011 with the owner_project_id domain/partition split. Per-migration identities above stay fixed; only this latest-version marker moves.
|
||||
*/
|
||||
export const SCHEMA_BASELINE_VERSION = "0010";
|
||||
export const SCHEMA_BASELINE_VERSION = "0011";
|
||||
const INITIAL_SCHEMA_VERSION = "0000";
|
||||
const AUTOMATION_ISOLATION_SCHEMA_VERSION = "0001";
|
||||
const ANALYTICS_ISOLATION_SCHEMA_VERSION = "0002";
|
||||
@@ -56,6 +56,16 @@ FNXC:GitHubImportTranslate 2026-07-15-09:30:
|
||||
Import-translation cache advances to 0010. Migrations are registered here explicitly (not auto-discovered from the migrations dir), so a new .sql file that is not wired through a version constant + bookkeeping check silently never runs.
|
||||
*/
|
||||
export const IMPORT_TRANSLATION_CACHE_VERSION = "0010";
|
||||
/*
|
||||
FNXC:MultiProjectIsolation 2026-07-15-23:40:
|
||||
Version 0011 splits the domain "project" field from the RLS partition on the tables
|
||||
that conflated them: `project_id` stays the trigger/GUC-owned isolation partition,
|
||||
`owner_project_id` becomes the caller-supplied domain field. Writing domain values
|
||||
into the partition put parent rows and child rows in different partitions and broke
|
||||
the composite FKs (SQLSTATE 23503). Keep this identity fixed when
|
||||
SCHEMA_BASELINE_VERSION advances.
|
||||
*/
|
||||
export const OWNER_PROJECT_ID_SPLIT_VERSION = "0011";
|
||||
|
||||
/** Bookkeeping table for the fresh Drizzle migration history. */
|
||||
export const MIGRATION_BOOKKEEPING_TABLE = "fusion_schema_migrations";
|
||||
@@ -112,6 +122,11 @@ const IMPORT_TRANSLATION_CACHE_MIGRATION_PATH = join(
|
||||
"migrations",
|
||||
"0010_import_translation_cache.sql",
|
||||
);
|
||||
const OWNER_PROJECT_ID_SPLIT_MIGRATION_PATH = join(
|
||||
__dirname,
|
||||
"migrations",
|
||||
"0011_owner_project_id.sql",
|
||||
);
|
||||
|
||||
/**
|
||||
* Ensure the migration bookkeeping table exists. Lives in the public schema so
|
||||
@@ -179,6 +194,7 @@ export async function applySchemaBaseline(
|
||||
const sessionAdvisorEnabledAlreadyApplied = applied.includes(SESSION_ADVISOR_ENABLED_SCHEMA_VERSION);
|
||||
const missionFixIdempotencyAlreadyApplied = applied.includes(MISSION_FIX_IDEMPOTENCY_VERSION);
|
||||
const importTranslationCacheAlreadyApplied = applied.includes(IMPORT_TRANSLATION_CACHE_VERSION);
|
||||
const ownerProjectIdSplitAlreadyApplied = applied.includes(OWNER_PROJECT_ID_SPLIT_VERSION);
|
||||
let schemaChanged = false;
|
||||
|
||||
if (!baselineAlreadyApplied) {
|
||||
@@ -415,6 +431,22 @@ export async function applySchemaBaseline(
|
||||
schemaChanged = true;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:MultiProjectIsolation 2026-07-15-23:40:
|
||||
Apply the owner_project_id domain/partition split independently of earlier
|
||||
schema versions so existing databases gain the domain column (backfilled from
|
||||
the previously conflated partition value) before any store read/write path
|
||||
that now targets owner_project_id runs on boot.
|
||||
*/
|
||||
if (!ownerProjectIdSplitAlreadyApplied) {
|
||||
const ownerProjectIdSplitSql = await readFile(OWNER_PROJECT_ID_SPLIT_MIGRATION_PATH, "utf8");
|
||||
await tx.execute(sql.raw(ownerProjectIdSplitSql));
|
||||
await tx.execute(
|
||||
sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${OWNER_PROJECT_ID_SPLIT_VERSION}) ON CONFLICT (version) DO NOTHING`,
|
||||
);
|
||||
schemaChanged = true;
|
||||
}
|
||||
|
||||
return { applied: schemaChanged, pluginHooksRun: pluginHooks.length };
|
||||
});
|
||||
}
|
||||
|
||||
@@ -931,6 +931,8 @@ export const researchRuns = projectSchema.table("research_runs", {
|
||||
topic: text("topic"),
|
||||
status: text("status").notNull(),
|
||||
projectId: text("project_id"),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
trigger: text("trigger"),
|
||||
providerConfig: jsonb("provider_config"),
|
||||
sources: jsonb("sources").notNull().default([]),
|
||||
@@ -985,6 +987,8 @@ export const experimentSessions = projectSchema.table("experiment_sessions", {
|
||||
id: text("id").primaryKey(),
|
||||
name: text("name").notNull(),
|
||||
projectId: text("project_id"),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
status: text("status").notNull(),
|
||||
metric: text("metric").notNull(),
|
||||
currentSegment: integer("current_segment").notNull().default(1),
|
||||
@@ -1022,7 +1026,10 @@ export const experimentSessionRecords = projectSchema.table("experiment_session_
|
||||
// ── Eval runs ────────────────────────────────────────────────────────
|
||||
export const evalRuns = projectSchema.table("eval_runs", {
|
||||
id: text("id").notNull(),
|
||||
projectId: text("project_id").notNull(),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: reflect the DB default installed by migration 0006 so insert types treat the trigger/GUC-owned partition as optional; stores must not write it.
|
||||
projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
status: text("status").notNull(),
|
||||
trigger: text("trigger").notNull(),
|
||||
scope: text("scope").notNull(),
|
||||
@@ -1425,7 +1432,10 @@ export const routines = projectSchema.table("routines", {
|
||||
|
||||
export const projectInsights = projectSchema.table("project_insights", {
|
||||
id: text("id").primaryKey(),
|
||||
projectId: text("project_id").notNull(),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: reflect the DB default installed by migration 0006 so insert types treat the trigger/GUC-owned partition as optional; stores must not write it.
|
||||
projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
title: text("title").notNull(),
|
||||
content: text("content"),
|
||||
category: text("category").notNull(),
|
||||
@@ -1443,7 +1453,10 @@ export const projectInsights = projectSchema.table("project_insights", {
|
||||
|
||||
export const projectInsightRuns = projectSchema.table("project_insight_runs", {
|
||||
id: text("id").notNull(),
|
||||
projectId: text("project_id").notNull(),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: reflect the DB default installed by migration 0006 so insert types treat the trigger/GUC-owned partition as optional; stores must not write it.
|
||||
projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
trigger: text("trigger").notNull(),
|
||||
status: text("status").notNull(),
|
||||
summary: text("summary"),
|
||||
@@ -1483,7 +1496,10 @@ export const projectInsightRunEvents = projectSchema.table("project_insight_run_
|
||||
// ── Todo lists ───────────────────────────────────────────────────────
|
||||
export const todoLists = projectSchema.table("todo_lists", {
|
||||
id: text("id").notNull(),
|
||||
projectId: text("project_id").notNull(),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: reflect the DB default installed by migration 0006 so insert types treat the trigger/GUC-owned partition as optional; stores must not write it.
|
||||
projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
title: text("title").notNull(),
|
||||
createdAt: text("created_at").notNull(),
|
||||
updatedAt: text("updated_at").notNull(),
|
||||
@@ -1618,6 +1634,8 @@ export const aiSessions = projectSchema.table("ai_sessions", {
|
||||
thinkingOutput: text("thinking_output").default(""),
|
||||
error: text("error"),
|
||||
projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
createdAt: text("created_at").notNull(),
|
||||
updatedAt: text("updated_at").notNull(),
|
||||
/*
|
||||
@@ -1685,6 +1703,8 @@ export const chatSessions = projectSchema.table("chat_sessions", {
|
||||
title: text("title"),
|
||||
status: text("status").notNull().default("active"),
|
||||
projectId: text("project_id"),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
modelProvider: text("model_provider"),
|
||||
modelId: text("model_id"),
|
||||
// FNXC:ChatThinkingLevel 2026-07-10: FN-7775 per-chat thinking-level override
|
||||
@@ -1708,7 +1728,10 @@ export const cliSessions = projectSchema.table("cli_sessions", {
|
||||
taskId: text("task_id"),
|
||||
chatSessionId: text("chat_session_id"),
|
||||
purpose: text("purpose").notNull(),
|
||||
projectId: text("project_id").notNull(),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: reflect the DB default installed by migration 0006 so insert types treat the trigger/GUC-owned partition as optional; stores must not write it.
|
||||
projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
adapterId: text("adapter_id").notNull(),
|
||||
agentState: text("agent_state").notNull().default("starting"),
|
||||
terminationReason: text("termination_reason"),
|
||||
@@ -1749,6 +1772,8 @@ export const chatTokenUsage = projectSchema.table("chat_token_usage", {
|
||||
roomId: text("room_id"),
|
||||
messageId: text("message_id"),
|
||||
projectId: text("project_id"),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
agentId: text("agent_id"),
|
||||
modelProvider: text("model_provider"),
|
||||
modelId: text("model_id"),
|
||||
@@ -1951,6 +1976,8 @@ export const chatRooms = projectSchema.table("chat_rooms", {
|
||||
slug: text("slug").notNull(),
|
||||
description: text("description"),
|
||||
projectId: text("project_id"),
|
||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: domain "project" field, split from the trigger/GUC-owned project_id RLS partition (migration 0011).
|
||||
ownerProjectId: text("owner_project_id"),
|
||||
createdBy: text("created_by"),
|
||||
status: text("status").notNull().default("active"),
|
||||
// FNXC:Chat-ThinkingLevel 2026-07-13 (merge port): room-level reasoning-effort default.
|
||||
|
||||
@@ -1126,7 +1126,19 @@ function resolveColumnMapping(
|
||||
const mapping: ColumnMapping[] = [];
|
||||
const unmappedSourceColumns: string[] = [];
|
||||
for (const sc of sqliteCols) {
|
||||
const pgName = toSnakeCase(sc.name);
|
||||
/*
|
||||
FNXC:MultiProjectIsolation 2026-07-15-23:40:
|
||||
SQLite had no partition concept: a legacy `projectId` column is always the DOMAIN
|
||||
project field. Since migration 0011 split that domain field (`owner_project_id`)
|
||||
from the trigger/GUC-owned RLS partition (`project_id`), route the source value to
|
||||
`owner_project_id` whenever the target declares it. The partition still receives
|
||||
the registry-resolved project id through the existing unmapped-project_id insert
|
||||
path, so cutover preserves the source's domain value instead of coercing it away.
|
||||
*/
|
||||
let pgName = toSnakeCase(sc.name);
|
||||
if (pgName === "project_id" && pgByName.has("owner_project_id")) {
|
||||
pgName = "owner_project_id";
|
||||
}
|
||||
const pgCol = pgByName.get(pgName);
|
||||
if (!pgCol) {
|
||||
/*
|
||||
|
||||
@@ -387,7 +387,7 @@ export async function aggregateTokenAnalytics(
|
||||
chat_session_id AS "chatSessionId",
|
||||
room_id AS "roomId",
|
||||
message_id AS "messageId",
|
||||
project_id AS "projectId",
|
||||
owner_project_id AS "projectId",
|
||||
agent_id AS "agentId",
|
||||
input_tokens AS "inputTokens",
|
||||
output_tokens AS "outputTokens",
|
||||
|
||||
Reference in New Issue
Block a user