From 9c5176f55e8dc2fbd1a551e65edda4b0e1825fc7 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 11 Aug 2026 19:34:01 -0700 Subject: [PATCH] FN-8997: enforce project ownership for project tables Enforce project-scoped ownership for chat and workflow records. - Add project ownership columns, migration backfill, and schema validation. - Scope chat-room and workflow CRUD operations to the active project. - Add PostgreSQL regression coverage, storage documentation, and a release changeset. Files changed: .../fn-8997-project-ownership-declaration-drift.md | 7 ++ docs/storage.md | 1 + .../project-ownership-declaration-drift.pg.test.ts | 42 +++++++++ .../src/__tests__/postgres/schema-applier.test.ts | 9 +- packages/core/src/async-stores/async-chat-store.ts | 100 ++++++++++++--------- packages/core/src/chat/chat-store.ts | 38 ++++---- ...fn_8997_project_ownership_declaration_drift.sql | 17 ++++ packages/core/src/postgres/schema-applier.ts | 12 ++- packages/core/src/postgres/schema/project.ts | 43 +++++++-- packages/core/src/task-store/project-store-ops.ts | 7 +- .../core/src/task-store/workflow-definitions.ts | 4 +- packages/core/src/task-store/workflow-ops.ts | 13 +-- .../src/task-store/workflow-task-create-ops.ts | 2 +- 13 files changed, 210 insertions(+), 85 deletions(-) Fusion-Task-Id: FN-8997 Fusion-Task-Lineage: 01cd71b1-8f86-41e5-abdb-5ab00ff91c83 Co-authored-by: Fusion (runfusion.ai) --- ...997-project-ownership-declaration-drift.md | 7 ++ docs/storage.md | 1 + ...ect-ownership-declaration-drift.pg.test.ts | 42 ++++++++ .../__tests__/postgres/schema-applier.test.ts | 9 +- .../core/src/async-stores/async-chat-store.ts | 100 ++++++++++-------- packages/core/src/chat/chat-store.ts | 38 +++---- ...97_project_ownership_declaration_drift.sql | 17 +++ packages/core/src/postgres/schema-applier.ts | 12 ++- packages/core/src/postgres/schema/project.ts | 43 ++++++-- .../core/src/task-store/project-store-ops.ts | 7 +- .../src/task-store/workflow-definitions.ts | 4 +- packages/core/src/task-store/workflow-ops.ts | 13 +-- .../task-store/workflow-task-create-ops.ts | 2 +- 13 files changed, 210 insertions(+), 85 deletions(-) create mode 100644 .changeset/fn-8997-project-ownership-declaration-drift.md create mode 100644 packages/core/src/__tests__/postgres/project-ownership-declaration-drift.pg.test.ts create mode 100644 packages/core/src/postgres/migrations/0056_fn_8997_project_ownership_declaration_drift.sql diff --git a/.changeset/fn-8997-project-ownership-declaration-drift.md b/.changeset/fn-8997-project-ownership-declaration-drift.md new file mode 100644 index 0000000000..15e8c800ff --- /dev/null +++ b/.changeset/fn-8997-project-ownership-declaration-drift.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Prevent project-bound workflow and chat data from crossing project partitions. +category: fix +dev: Adds idempotent migration 0056 for partition-prefixed ownership indexes. diff --git a/docs/storage.md b/docs/storage.md index 410e6e788a..8bc2e3f663 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -36,6 +36,7 @@ See the [2026-07-14 PostgreSQL runtime cutover review](./postgres-migration-revi - Lock acquire is all-or-nothing over normalized keys. Held unexpired rows owned by another task return their owner as a conflict, while expired/released rows may be reclaimed. Renewal and release are owner-scoped and release is idempotent. - The `0000_initial.sql` baseline defines the table and indexes only. The later `0025_symbol_locks.sql` migration enables and forces RLS, creates `fusion_project_isolation`, and attaches `fusion_assign_project_id` after `0006_project_ownership.sql` creates that function/policy machinery. Both fresh full-applier and upgrade paths therefore end with the same project-isolation contract. - `project.agent_ratings` is project-owned with composite `(project_id, id)` identity, allowing the same rating id in separate projects without cross-project reads or deletes. The dynamic `0006_project_ownership.sql` migration reconciles the physical table; `0055_fn_8988_agent_ratings_project_partition.sql` repeats that guarantee idempotently for historical drift. Bound `addRating`, `getRatings`, and `deleteRating` apply the project ownership partition, while unbound compatibility layers retain trigger-stamped writes and unscoped reads/deletes. +- FN-8997 audited `workflow_steps`, `chat_room_members`, and `chat_room_messages`: their Drizzle declarations now model 0006's `project_id` and composite keys, and bound workflow/chat helpers scope reads and mutations on that partition. Chat isolation requires both membership/message predicates **and** the parent `chat_rooms.project_id` predicate; either leg alone can resolve a foreign row when room IDs collide. `plugins` remains an intentionally unmodeled compatibility table because it has no runtime Drizzle path. Migration `0056_fn_8997_project_ownership_declaration_drift.sql` is idempotent and adds only partition-prefixed predicate indexes; it does not rewrite healthy 0006 ownership columns or keys. - Startup and Batch 1 self-healing expire locks when their lease elapsed or the owner task is terminal/missing. They never move a task or alter scheduler, worktree, semaphore, or verification state. Run-audit events are `symbol-lock:acquired`, `symbol-lock:acquire-conflict`, `symbol-lock:renewed`, `symbol-lock:released`, `symbol-lock:reconcile-stale`, and deduplicated `symbol-lock:reconcile-stale-no-action`; metadata uses only counts/outcomes and normalized opaque keys. - FN-8405 adds `Task.declaredSymbols` as the durable, normalized task declaration source. `## Declared Symbols` in PROMPT.md is parsed only on create/update writes: an absent key may hydrate from the prompt, while a present `undefined`, `null` (update), or `[]` clears and suppresses hydration; a non-empty explicit array wins. Store resolution (`resolveTaskSymbols` and `resolveTaskSymbolsForWorkItem({ taskId })`) reads only the durable field, and slim projections plus archive/restore retain it. Scheduler admission remains a separate FN-8306 consumer; File Scope is never treated as a symbol source. diff --git a/packages/core/src/__tests__/postgres/project-ownership-declaration-drift.pg.test.ts b/packages/core/src/__tests__/postgres/project-ownership-declaration-drift.pg.test.ts new file mode 100644 index 0000000000..031983e37d --- /dev/null +++ b/packages/core/src/__tests__/postgres/project-ownership-declaration-drift.pg.test.ts @@ -0,0 +1,42 @@ +import { afterAll, afterEach, beforeAll, beforeEach, expect, it } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../../__test-utils__/pg-test-harness.js"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import * as schema from "../../postgres/schema/index.js"; +import { TaskStore } from "../../store.js"; +import { ChatStore } from "../../chat/chat-store.js"; + +pgDescribe("project ownership declaration drift", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_fn8997" }); + beforeAll(h.beforeAll); afterAll(h.afterAll); beforeEach(h.beforeEach); afterEach(h.afterEach); + + it("keeps duplicated workflow, room, member, and message ids in a bound partition", async () => { + /* + FNXC:MultiProjectIsolation 2026-08-12-02:12: + FN-8997 reproduces duplicate ids on an owner-connected cluster, where RLS bypass cannot mask missing ORM predicates. Both membership and parent-room legs must be scoped for the duplicated room id assertion to hold. + */ + const bind = (projectId: string): AsyncDataLayer => ({ ...h.layer(), projectId }); + const a = bind("ownership-a"); const b = bind("ownership-b"); + const now = "2026-08-12T02:12:00.000Z"; + for (const [layer, name] of [[a, "A"], [b, "B"]] as const) { + await layer.db.insert(schema.project.workflowSteps).values({ projectId: layer.projectId, id: "ws-1", templateId: "shared-template", name, description: name, createdAt: now, updatedAt: now }); + await layer.db.insert(schema.project.chatRooms).values({ projectId: layer.projectId, id: "room-1", name: `Room ${name}`, slug: `room-${name}`, ownerProjectId: `domain-${name}`, status: "active", createdAt: now, updatedAt: now }); + await layer.db.insert(schema.project.chatRoomMembers).values({ projectId: layer.projectId, roomId: "room-1", agentId: "agent-shared", addedAt: now }); + await layer.db.insert(schema.project.chatRoomMessages).values({ projectId: layer.projectId, id: "message-1", roomId: "room-1", role: "user", content: name, createdAt: now }); + } + const storeA = new TaskStore(h.rootDir(), undefined, { asyncLayer: a }); await storeA.init(); + expect((await storeA.listWorkflowSteps()).filter((step) => step.id === "ws-1").map((step) => step.name)).toEqual(["A"]); + expect((await storeA.getWorkflowStep("ws-1"))?.name).toBe("A"); + expect((await storeA.getWorkflowStep("shared-template"))?.name).toBe("A"); + await storeA.updateWorkflowStep("ws-1", { name: "A renamed" }); + expect((await h.adminDb().select().from(schema.project.workflowSteps).where(and(eq(schema.project.workflowSteps.projectId, b.projectId), eq(schema.project.workflowSteps.id, "ws-1"))))[0]?.name).toBe("B"); + await storeA.deleteWorkflowStep("ws-1"); + expect((await h.adminDb().select().from(schema.project.workflowSteps).where(and(eq(schema.project.workflowSteps.projectId, b.projectId), eq(schema.project.workflowSteps.id, "ws-1")))).length).toBe(1); + + const chatA = new ChatStore(a); + expect((await chatA.listRoomsForAgent("agent-shared", { status: "active" })).map((room) => room.name)).toEqual(["Room A"]); + expect((await chatA.getRoom("room-1"))?.name).toBe("Room A"); + expect((await chatA.listRoomMembers("room-1")).map((member) => member.agentId)).toEqual(["agent-shared"]); + expect((await chatA.getRoomMessages("room-1")).map((message) => message.content)).toEqual(["A"]); + }); +}); diff --git a/packages/core/src/__tests__/postgres/schema-applier.test.ts b/packages/core/src/__tests__/postgres/schema-applier.test.ts index 0d428344a9..a96282f9c9 100644 --- a/packages/core/src/__tests__/postgres/schema-applier.test.ts +++ b/packages/core/src/__tests__/postgres/schema-applier.test.ts @@ -98,6 +98,7 @@ import { MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, AGENT_RATING_PROJECT_ISOLATION_VERSION, AGENT_RATINGS_PROJECT_PARTITION_VERSION, + PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION, } from "../../postgres/schema-applier.js"; import { ProjectPartitionRekeyError, rekeyFallbackProjectPartition } from "../../postgres/migration-stamping.js"; import type { PluginSchemaInitHook } from "../../postgres/plugin-schema-hook.js"; @@ -129,7 +130,8 @@ describe("schema-applier: immutable migration identities", () => { expect(MISSION_FEATURE_SPEC_ALIGNMENT_VERSION).toBe("0053"); expect(AGENT_RATING_PROJECT_ISOLATION_VERSION).toBe("0054"); expect(AGENT_RATINGS_PROJECT_PARTITION_VERSION).toBe("0055"); - expect(SCHEMA_BASELINE_VERSION).toBe("0055"); + expect(PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION).toBe("0056"); + expect(SCHEMA_BASELINE_VERSION).toBe("0056"); }); it("keeps monitor and approval isolation assigned to version 0003", () => { @@ -1800,6 +1802,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, AGENT_RATING_PROJECT_ISOLATION_VERSION, AGENT_RATINGS_PROJECT_PARTITION_VERSION, + PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION, ]); expect((await applySchemaBaseline(ctx.db, { pluginHooks: [] })).applied).toBe(false); }); @@ -1881,6 +1884,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, AGENT_RATING_PROJECT_ISOLATION_VERSION, AGENT_RATINGS_PROJECT_PARTITION_VERSION, + PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION, ]); }); @@ -2095,6 +2099,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, AGENT_RATING_PROJECT_ISOLATION_VERSION, AGENT_RATINGS_PROJECT_PARTITION_VERSION, + PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION, ]); }); @@ -2190,6 +2195,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, AGENT_RATING_PROJECT_ISOLATION_VERSION, AGENT_RATINGS_PROJECT_PARTITION_VERSION, + PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION, ]); }); @@ -2285,6 +2291,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, AGENT_RATING_PROJECT_ISOLATION_VERSION, AGENT_RATINGS_PROJECT_PARTITION_VERSION, + PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION, ]); }); }); diff --git a/packages/core/src/async-stores/async-chat-store.ts b/packages/core/src/async-stores/async-chat-store.ts index ef4f7196e6..e2eba1a306 100644 --- a/packages/core/src/async-stores/async-chat-store.ts +++ b/packages/core/src/async-stores/async-chat-store.ts @@ -19,7 +19,7 @@ import { randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, ilike, inArray, isNull, lte, ne, or as orFn, sql as drizzleSql } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; -import type { AsyncDataLayer, DbTransaction } from "../postgres/data-layer.js"; +import { projectScopeFor, type AsyncDataLayer, type DbTransaction } from "../postgres/data-layer.js"; import { sanitizeTextValue, sanitizeJsonbValue } from "../postgres/nul-sanitize.js"; import type { ChatAttachment, @@ -422,6 +422,7 @@ export async function createChatRoom( const now = room.createdAt; await layer.transactionImmediate(async (tx) => { await tx.insert(schema.project.chatRooms).values({ + projectId: layer.projectId?.trim() ?? "", id: room.id, name: room.name, slug: room.slug, @@ -436,6 +437,7 @@ export async function createChatRoom( for (const agentId of memberAgentIds) { const role: RoomMemberRole = room.createdBy !== null && agentId === room.createdBy ? "owner" : "member"; await tx.insert(schema.project.chatRoomMembers).values({ + projectId: layer.projectId?.trim() ?? "", roomId: room.id, agentId, role, @@ -443,18 +445,19 @@ export async function createChatRoom( }); } }); - const members = await listChatRoomMembers(layer.db, room.id); + const members = await listChatRoomMembers(layer.db, room.id, layer.projectId); return { room, members }; } /** * Get a chat room by id. */ -export async function getChatRoom(handle: QueryHandle, id: string): Promise { +export async function getChatRoom(handle: QueryHandle, id: string, + projectId?: string): Promise { const rows = await handle .select() .from(schema.project.chatRooms) - .where(eq(schema.project.chatRooms.id, id)); + .where(and(eq(schema.project.chatRooms.id, id), projectScopeFor(schema.project.chatRooms.projectId, projectId))); return rows[0] ? rowToRoom(rows[0]) : undefined; } @@ -463,12 +466,12 @@ export async function getChatRoom(handle: QueryHandle, id: string): Promise { - const conditions = [eq(schema.project.chatRooms.slug, slug)]; - if (projectId !== null) { - conditions.push(eq(schema.project.chatRooms.ownerProjectId, projectId)); + projectId?: string): Promise { + const conditions = [eq(schema.project.chatRooms.slug, slug), projectScopeFor(schema.project.chatRooms.projectId, projectId)]; + if (ownerProjectId !== null) { + conditions.push(eq(schema.project.chatRooms.ownerProjectId, ownerProjectId)); } else { conditions.push(isNull(schema.project.chatRooms.ownerProjectId)); } @@ -485,8 +488,9 @@ export async function getChatRoomBySlug( export async function listChatRooms( handle: QueryHandle, options?: { projectId?: string; status?: ChatRoomStatus }, + projectId?: string, ): Promise { - const conditions: ReturnType[] = []; + const conditions = [projectScopeFor(schema.project.chatRooms.projectId, 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 @@ -500,10 +504,11 @@ export async function listChatRooms( /** * Delete a chat room by id. Returns true if a row was deleted. */ -export async function deleteChatRoom(handle: QueryHandle, id: string): Promise { +export async function deleteChatRoom(handle: QueryHandle, id: string, + projectId?: string): Promise { const result = await handle .delete(schema.project.chatRooms) - .where(eq(schema.project.chatRooms.id, id)) + .where(and(eq(schema.project.chatRooms.id, id), projectScopeFor(schema.project.chatRooms.projectId, projectId))) .returning({ id: schema.project.chatRooms.id }); return result.length > 0; } @@ -521,10 +526,10 @@ export async function addChatRoomMember( agentId: string, role: RoomMemberRole, addedAt: string, -): Promise { + projectId?: string): Promise { await handle .insert(schema.project.chatRoomMembers) - .values({ roomId, agentId, role, addedAt }) + .values({ projectId: projectId?.trim() ?? "", roomId, agentId, role, addedAt }) .onConflictDoNothing(); } @@ -535,13 +540,14 @@ export async function removeChatRoomMember( handle: QueryHandle, roomId: string, agentId: string, -): Promise { + projectId?: string): Promise { const result = await handle .delete(schema.project.chatRoomMembers) .where( and( eq(schema.project.chatRoomMembers.roomId, roomId), eq(schema.project.chatRoomMembers.agentId, agentId), + projectScopeFor(schema.project.chatRoomMembers.projectId, projectId), ), ) .returning({ roomId: schema.project.chatRoomMembers.roomId }); @@ -551,11 +557,12 @@ export async function removeChatRoomMember( /** * List room members ordered by addedAt ASC. */ -export async function listChatRoomMembers(handle: QueryHandle, roomId: string): Promise { +export async function listChatRoomMembers(handle: QueryHandle, roomId: string, + projectId?: string): Promise { const rows = await handle .select() .from(schema.project.chatRoomMembers) - .where(eq(schema.project.chatRoomMembers.roomId, roomId)) + .where(and(eq(schema.project.chatRoomMembers.roomId, roomId), projectScopeFor(schema.project.chatRoomMembers.projectId, projectId))) .orderBy(asc(schema.project.chatRoomMembers.addedAt)); return rows.map(rowToRoomMember); } @@ -569,7 +576,7 @@ export async function listChatRoomMembers(handle: QueryHandle, roomId: string): export async function addChatRoomMessage( handle: QueryHandle, message: ChatRoomMessage, -): Promise { + projectId?: string): Promise { // FNXC:PostgresMigrationNulSanitize 2026-07-20: same NUL-byte hazard as // addChatMessage above — sanitize before insert, and return the sanitized // value so the in-memory result matches what was persisted. @@ -584,6 +591,7 @@ export async function addChatRoomMessage( attachments: sanitizedAttachments, }; await handle.insert(schema.project.chatRoomMessages).values({ + projectId: projectId?.trim() ?? "", id: sanitized.id, roomId: sanitized.roomId, role: sanitized.role, @@ -598,18 +606,19 @@ export async function addChatRoomMessage( await handle .update(schema.project.chatRooms) .set({ updatedAt: sanitized.createdAt }) - .where(eq(schema.project.chatRooms.id, sanitized.roomId)); + .where(and(eq(schema.project.chatRooms.id, sanitized.roomId), projectScopeFor(schema.project.chatRooms.projectId, projectId))); return sanitized; } /** * Get a room message by id. */ -export async function getChatRoomMessage(handle: QueryHandle, id: string): Promise { +export async function getChatRoomMessage(handle: QueryHandle, id: string, + projectId?: string): Promise { const rows = await handle .select() .from(schema.project.chatRoomMessages) - .where(eq(schema.project.chatRoomMessages.id, id)); + .where(and(eq(schema.project.chatRoomMessages.id, id), projectScopeFor(schema.project.chatRoomMessages.projectId, projectId))); return rows[0] ? rowToRoomMessage(rows[0]) : undefined; } @@ -620,8 +629,8 @@ export async function getChatRoomMessages( handle: QueryHandle, roomId: string, filter?: { limit?: number; offset?: number; before?: string; order?: "asc" | "desc" }, -): Promise { - const conditions: ReturnType[] = [eq(schema.project.chatRoomMessages.roomId, roomId)]; + projectId?: string): Promise { + const conditions = [eq(schema.project.chatRoomMessages.roomId, roomId), projectScopeFor(schema.project.chatRoomMessages.projectId, projectId)]; if (filter?.before) { conditions.push(lte(schema.project.chatRoomMessages.createdAt, filter.before)); } @@ -642,10 +651,11 @@ export async function getChatRoomMessages( * FNXC:ChatStore 2026-06-24-09:35: * Clear all room messages. Returns the count of deleted messages. */ -export async function clearChatRoomMessages(handle: QueryHandle, roomId: string): Promise { +export async function clearChatRoomMessages(handle: QueryHandle, roomId: string, + projectId?: string): Promise { const result = await handle .delete(schema.project.chatRoomMessages) - .where(eq(schema.project.chatRoomMessages.roomId, roomId)) + .where(and(eq(schema.project.chatRoomMessages.roomId, roomId), projectScopeFor(schema.project.chatRoomMessages.projectId, projectId))) .returning({ id: schema.project.chatRoomMessages.id }); return result.length; } @@ -976,8 +986,8 @@ export async function updateChatRoom( status?: ChatRoomStatus; thinkingLevel?: ChatRoom["thinkingLevel"] | null; }, -): Promise { - const existing = await getChatRoom(handle, id); + projectId?: string): Promise { + const existing = await getChatRoom(handle, id, projectId); if (!existing) return undefined; const setValues: Record = { updatedAt: new Date().toISOString() }; @@ -990,9 +1000,9 @@ export async function updateChatRoom( await handle .update(schema.project.chatRooms) .set(setValues) - .where(eq(schema.project.chatRooms.id, id)); + .where(and(eq(schema.project.chatRooms.id, id), projectScopeFor(schema.project.chatRooms.projectId, projectId))); - return getChatRoom(handle, id); + return getChatRoom(handle, id, projectId); } /** @@ -1004,7 +1014,7 @@ export async function updateChatRoom( export async function cleanupOldChats( handle: QueryHandle, maxAgeMs: number, -): Promise<{ sessionsDeleted: number; roomsDeleted: number; deletedSessionIds: string[]; deletedRoomIds: string[] }> { + projectId?: string): Promise<{ sessionsDeleted: number; roomsDeleted: number; deletedSessionIds: string[]; deletedRoomIds: string[] }> { if (!Number.isFinite(maxAgeMs) || maxAgeMs <= 0) { return { sessionsDeleted: 0, roomsDeleted: 0, deletedSessionIds: [], deletedRoomIds: [] }; } @@ -1017,7 +1027,7 @@ export async function cleanupOldChats( const staleRooms = await handle .delete(schema.project.chatRooms) - .where(lte(schema.project.chatRooms.updatedAt, cutoff)) + .where(and(lte(schema.project.chatRooms.updatedAt, cutoff), projectScopeFor(schema.project.chatRooms.projectId, projectId))) .returning({ id: schema.project.chatRooms.id }); return { @@ -1038,15 +1048,16 @@ export async function listChatRoomsForAgent( handle: QueryHandle, agentId: string, options?: { projectId?: string; status?: ChatRoomStatus }, + projectId?: string, ): Promise { // Use a subquery to find room IDs where the agent is a member, then select // those rooms. This avoids the Drizzle join result-shape complexity. const memberRoomIds = handle .select({ roomId: schema.project.chatRoomMembers.roomId }) .from(schema.project.chatRoomMembers) - .where(eq(schema.project.chatRoomMembers.agentId, agentId)); + .where(and(eq(schema.project.chatRoomMembers.agentId, agentId), projectScopeFor(schema.project.chatRoomMembers.projectId, projectId))); - const conditions: ReturnType[] = [inArray(schema.project.chatRooms.id, memberRoomIds)]; + const conditions = [inArray(schema.project.chatRooms.id, memberRoomIds), projectScopeFor(schema.project.chatRooms.projectId, projectId)]; if (options?.status) conditions.push(eq(schema.project.chatRooms.status, options.status)); if (options?.projectId) conditions.push(eq(schema.project.chatRooms.ownerProjectId, options.projectId)); @@ -1069,9 +1080,10 @@ export async function listChatRoomMessagesSince( roomId: string, sinceIso: string, options?: { excludeSenderAgentId?: string; limit?: number }, -): Promise { - const conditions: ReturnType[] = [ + projectId?: string): Promise { + const conditions = [ eq(schema.project.chatRoomMessages.roomId, roomId), + projectScopeFor(schema.project.chatRoomMessages.projectId, projectId), gt(schema.project.chatRoomMessages.createdAt, sinceIso), ]; if (options?.excludeSenderAgentId) { @@ -1101,14 +1113,14 @@ export async function listChatRoomMessagesSince( export async function deleteChatRoomMessage( handle: QueryHandle, id: string, -): Promise { - const existing = await getChatRoomMessage(handle, id); + projectId?: string): Promise { + const existing = await getChatRoomMessage(handle, id, projectId); if (!existing) return false; - await handle.delete(schema.project.chatRoomMessages).where(eq(schema.project.chatRoomMessages.id, id)); + await handle.delete(schema.project.chatRoomMessages).where(and(eq(schema.project.chatRoomMessages.id, id), projectScopeFor(schema.project.chatRoomMessages.projectId, projectId))); await handle .update(schema.project.chatRooms) .set({ updatedAt: new Date().toISOString() }) - .where(eq(schema.project.chatRooms.id, existing.roomId)); + .where(and(eq(schema.project.chatRooms.id, existing.roomId), projectScopeFor(schema.project.chatRooms.projectId, projectId))); return true; } @@ -1124,8 +1136,8 @@ export async function addChatRoomMessageAttachment( roomId: string, messageId: string, attachment: ChatAttachment, -): Promise { - const message = await getChatRoomMessage(handle, messageId); + projectId?: string): Promise { + const message = await getChatRoomMessage(handle, messageId, projectId); if (!message || message.roomId !== roomId) { throw new Error(`Message ${messageId} not found in room ${roomId}`); } @@ -1136,12 +1148,12 @@ export async function addChatRoomMessageAttachment( await handle .update(schema.project.chatRoomMessages) .set({ attachments: updatedAttachments }) - .where(eq(schema.project.chatRoomMessages.id, messageId)); + .where(and(eq(schema.project.chatRoomMessages.id, messageId), projectScopeFor(schema.project.chatRoomMessages.projectId, projectId))); await handle .update(schema.project.chatRooms) .set({ updatedAt: new Date().toISOString() }) - .where(eq(schema.project.chatRooms.id, roomId)); - const updated = await getChatRoomMessage(handle, messageId); + .where(and(eq(schema.project.chatRooms.id, roomId), projectScopeFor(schema.project.chatRooms.projectId, projectId))); + const updated = await getChatRoomMessage(handle, messageId, projectId); if (!updated) throw new Error(`Failed to update room message ${messageId}`); return updated; } diff --git a/packages/core/src/chat/chat-store.ts b/packages/core/src/chat/chat-store.ts index a6e5085dab..176f5745e4 100644 --- a/packages/core/src/chat/chat-store.ts +++ b/packages/core/src/chat/chat-store.ts @@ -540,15 +540,15 @@ export class ChatStore extends EventEmitter { } async getRoom(id: string): Promise { - return asyncChatStore.getChatRoom(this.asyncLayer.db, id); + return asyncChatStore.getChatRoom(this.asyncLayer.db, id, this.asyncLayer.projectId); } async getRoomBySlug(projectId: string | null, slug: string): Promise { - return asyncChatStore.getChatRoomBySlug(this.asyncLayer.db, projectId, slug); + return asyncChatStore.getChatRoomBySlug(this.asyncLayer.db, projectId, slug, this.asyncLayer.projectId); } async listRooms(options?: { projectId?: string; status?: ChatRoomStatus }): Promise { - return asyncChatStore.listChatRooms(this.asyncLayer.db, options); + return asyncChatStore.listChatRooms(this.asyncLayer.db, options, this.asyncLayer.projectId); } async updateRoom(id: string, input: ChatRoomUpdateInput): Promise { @@ -561,7 +561,7 @@ export class ChatStore extends EventEmitter { if (!slug) throw new Error("Room name must include letters or numbers"); const existing = await this.getRoom(id); if (existing) { - const slugConflict = await asyncChatStore.getChatRoomBySlug(this.asyncLayer.db, existing.projectId, slug); + const slugConflict = await asyncChatStore.getChatRoomBySlug(this.asyncLayer.db, existing.projectId, slug, this.asyncLayer.projectId); if (slugConflict && slugConflict.id !== id) { throw new Error(`Room slug ${slug} already exists in this project`); } @@ -570,19 +570,19 @@ export class ChatStore extends EventEmitter { } if (input.description !== undefined) updateInput.description = input.description; if (input.status !== undefined) updateInput.status = input.status; - const updated = await asyncChatStore.updateChatRoom(this.asyncLayer.db, id, updateInput); + const updated = await asyncChatStore.updateChatRoom(this.asyncLayer.db, id, updateInput, this.asyncLayer.projectId); if (updated) this.emit("chat:room:updated", updated); return updated; } async deleteRoom(id: string): Promise { - const deleted = await asyncChatStore.deleteChatRoom(this.asyncLayer.db, id); + const deleted = await asyncChatStore.deleteChatRoom(this.asyncLayer.db, id, this.asyncLayer.projectId); if (deleted) this.emit("chat:room:deleted", id); return deleted; } async cleanupOldChats(maxAgeMs: number): Promise<{ sessionsDeleted: number; roomsDeleted: number }> { - const result = await asyncChatStore.cleanupOldChats(this.asyncLayer.db, maxAgeMs); + const result = await asyncChatStore.cleanupOldChats(this.asyncLayer.db, maxAgeMs, this.asyncLayer.projectId); for (const sessionId of result.deletedSessionIds) { this.emit("chat:session:deleted", sessionId); } @@ -594,7 +594,7 @@ export class ChatStore extends EventEmitter { async addRoomMember(roomId: string, agentId: string, role: RoomMemberRole = "member"): Promise { const now = new Date().toISOString(); - await asyncChatStore.addChatRoomMember(this.asyncLayer.db, roomId, agentId, role, now); + await asyncChatStore.addChatRoomMember(this.asyncLayer.db, roomId, agentId, role, now, this.asyncLayer.projectId); const members = await this.listRoomMembers(roomId); const member = members.find((m) => m.agentId === agentId); if (!member) throw new Error(`Failed to load room member ${agentId}`); @@ -603,17 +603,17 @@ export class ChatStore extends EventEmitter { } async removeRoomMember(roomId: string, agentId: string): Promise { - const removed = await asyncChatStore.removeChatRoomMember(this.asyncLayer.db, roomId, agentId); + const removed = await asyncChatStore.removeChatRoomMember(this.asyncLayer.db, roomId, agentId, this.asyncLayer.projectId); if (removed) this.emit("chat:room:member:removed", { roomId, agentId }); return removed; } async listRoomMembers(roomId: string): Promise { - return asyncChatStore.listChatRoomMembers(this.asyncLayer.db, roomId); + return asyncChatStore.listChatRoomMembers(this.asyncLayer.db, roomId, this.asyncLayer.projectId); } async listRoomsForAgent(agentId: string, options?: { projectId?: string; status?: ChatRoomStatus }): Promise { - return asyncChatStore.listChatRoomsForAgent(this.asyncLayer.db, agentId, options); + return asyncChatStore.listChatRoomsForAgent(this.asyncLayer.db, agentId, options, this.asyncLayer.projectId); } async addRoomMessage(roomId: string, input: ChatRoomMessageCreateInput): Promise { @@ -634,13 +634,13 @@ export class ChatStore extends EventEmitter { mentions: input.mentions ?? [], createdAt: now, }; - const created = await asyncChatStore.addChatRoomMessage(this.asyncLayer.db, message); + const created = await asyncChatStore.addChatRoomMessage(this.asyncLayer.db, message, this.asyncLayer.projectId); this.emit("chat:room:message:added", created); return created; } async getRoomMessages(roomId: string, filter?: ChatRoomMessagesFilter): Promise { - return asyncChatStore.getChatRoomMessages(this.asyncLayer.db, roomId, filter); + return asyncChatStore.getChatRoomMessages(this.asyncLayer.db, roomId, filter, this.asyncLayer.projectId); } async listRoomMessagesSince( @@ -648,17 +648,17 @@ export class ChatStore extends EventEmitter { sinceIso: string, options?: { excludeSenderAgentId?: string; limit?: number }, ): Promise { - return asyncChatStore.listChatRoomMessagesSince(this.asyncLayer.db, roomId, sinceIso, options); + return asyncChatStore.listChatRoomMessagesSince(this.asyncLayer.db, roomId, sinceIso, options, this.asyncLayer.projectId); } async getRoomMessage(id: string): Promise { - return asyncChatStore.getChatRoomMessage(this.asyncLayer.db, id); + return asyncChatStore.getChatRoomMessage(this.asyncLayer.db, id, this.asyncLayer.projectId); } async deleteRoomMessage(id: string): Promise { - const existing = await asyncChatStore.getChatRoomMessage(this.asyncLayer.db, id); + const existing = await asyncChatStore.getChatRoomMessage(this.asyncLayer.db, id, this.asyncLayer.projectId); if (!existing) return false; - const deleted = await asyncChatStore.deleteChatRoomMessage(this.asyncLayer.db, id); + const deleted = await asyncChatStore.deleteChatRoomMessage(this.asyncLayer.db, id, this.asyncLayer.projectId); if (deleted) { this.emit("chat:room:message:deleted", id); const updatedRoom = await this.getRoom(existing.roomId); @@ -668,13 +668,13 @@ export class ChatStore extends EventEmitter { } async clearRoomMessages(roomId: string): Promise { - const deleted = await asyncChatStore.clearChatRoomMessages(this.asyncLayer.db, roomId); + const deleted = await asyncChatStore.clearChatRoomMessages(this.asyncLayer.db, roomId, this.asyncLayer.projectId); if (deleted > 0) this.emit("chat:room:messages:cleared", { roomId, deletedCount: deleted }); return deleted; } async addRoomMessageAttachment(roomId: string, messageId: string, attachment: ChatAttachment): Promise { - const updated = await asyncChatStore.addChatRoomMessageAttachment(this.asyncLayer.db, roomId, messageId, attachment); + const updated = await asyncChatStore.addChatRoomMessageAttachment(this.asyncLayer.db, roomId, messageId, attachment, this.asyncLayer.projectId); this.emit("chat:room:message:updated", updated); return updated; } diff --git a/packages/core/src/postgres/migrations/0056_fn_8997_project_ownership_declaration_drift.sql b/packages/core/src/postgres/migrations/0056_fn_8997_project_ownership_declaration_drift.sql new file mode 100644 index 0000000000..94adb17bc8 --- /dev/null +++ b/packages/core/src/postgres/migrations/0056_fn_8997_project_ownership_declaration_drift.sql @@ -0,0 +1,17 @@ +/* +FNXC:MultiProjectIsolation 2026-08-12-02:12: +Migration 0006 already establishes project_id ownership and composite keys. FN-8997 adds only idempotent partition-prefixed indexes needed by the newly scoped Drizzle predicates; healthy post-0006 databases require no ownership rewrite. +*/ +DO $$ +BEGIN + IF to_regclass('project.workflow_steps') IS NOT NULL THEN + CREATE INDEX IF NOT EXISTS "idxWorkflowStepsProjectCreatedAt" ON project.workflow_steps(project_id, created_at); + END IF; + IF to_regclass('project.chat_room_members') IS NOT NULL THEN + CREATE INDEX IF NOT EXISTS "idxChatRoomMembersProjectAgentId" ON project.chat_room_members(project_id, agent_id); + END IF; + IF to_regclass('project.chat_room_messages') IS NOT NULL THEN + CREATE INDEX IF NOT EXISTS "idxChatRoomMessagesProjectRoomCreatedAt" ON project.chat_room_messages(project_id, room_id, created_at); + CREATE INDEX IF NOT EXISTS "idxChatRoomMessagesProjectRoomId" ON project.chat_room_messages(project_id, room_id); + END IF; +END $$; diff --git a/packages/core/src/postgres/schema-applier.ts b/packages/core/src/postgres/schema-applier.ts index 298c26cc69..f091df117f 100644 --- a/packages/core/src/postgres/schema-applier.ts +++ b/packages/core/src/postgres/schema-applier.ts @@ -62,7 +62,7 @@ capacity-model table drop that landed while this PR was open. /* FNXC:MemoryRecall 2026-08-10-11:03: Explicit baseline registration prevents the recall migration from being silently skipped. */ /* FNXC:SpecLockMissionAlignment 2026-08-10-16:17: advance the schema ceiling so SQLite and PostgreSQL feature projections retain reconciled drift alignment. */ /* FNXC:MultiProjectIsolation 2026-08-11-10:25: schema startup must register project-local agent ratings before bound stores scope their mutations. */ -export const SCHEMA_BASELINE_VERSION = "0055"; +export const SCHEMA_BASELINE_VERSION = "0056"; /** FNXC:SymbolLock 2026-07-20-10:00: upgrades need durable task declarations before admission resolves symbols. */ export const TASK_DECLARED_SYMBOLS_VERSION = "0028"; const INITIAL_SCHEMA_VERSION = "0000"; @@ -215,6 +215,8 @@ export const MISSION_FEATURE_SPEC_ALIGNMENT_VERSION = "0053"; export const AGENT_RATING_PROJECT_ISOLATION_VERSION = "0054"; /** FNXC:AgentRatingsProjectIsolation 2026-08-12-01:00: targeted idempotent reconciliation protects historical rating ownership drift. */ export const AGENT_RATINGS_PROJECT_PARTITION_VERSION = "0055"; +/** FNXC:MultiProjectIsolation 2026-08-12-02:12: register FN-8997 predicate indexes explicitly; migration files are never auto-discovered. */ +export const PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION = "0056"; /** SECURITY DEFINER helper that only inserts LEGACY_ADOPTION_DRAINED_MARKER. */ export const LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION = "fusion_mark_legacy_adoption_drained"; @@ -444,6 +446,7 @@ const MEMORY_RECALL_RECORDS_MIGRATION_PATH = join(MIGRATIONS_DIR, "0052_fn_8922_ const MISSION_FEATURE_SPEC_ALIGNMENT_MIGRATION_PATH = join(MIGRATIONS_DIR, "0053_mission_feature_spec_alignment.sql"); const AGENT_RATING_PROJECT_ISOLATION_MIGRATION_PATH = join(MIGRATIONS_DIR, "0054_fn_8957_agent_rating_project_isolation.sql"); const AGENT_RATINGS_PROJECT_PARTITION_MIGRATION_PATH = join(MIGRATIONS_DIR, "0055_fn_8988_agent_ratings_project_partition.sql"); +const PROJECT_OWNERSHIP_DECLARATION_DRIFT_MIGRATION_PATH = join(MIGRATIONS_DIR, "0056_fn_8997_project_ownership_declaration_drift.sql"); /** * Ensure the migration bookkeeping table exists. Lives in the public schema so @@ -569,6 +572,7 @@ export async function applySchemaBaseline( const missionFeatureSpecAlignmentAlreadyApplied = applied.includes(MISSION_FEATURE_SPEC_ALIGNMENT_VERSION); const agentRatingProjectIsolationAlreadyApplied = applied.includes(AGENT_RATING_PROJECT_ISOLATION_VERSION); const agentRatingsProjectPartitionAlreadyApplied = applied.includes(AGENT_RATINGS_PROJECT_PARTITION_VERSION); + const projectOwnershipDeclarationDriftAlreadyApplied = applied.includes(PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION); assertBinaryNotOlderThanDatabase(applied); let schemaChanged = false; @@ -1251,6 +1255,12 @@ export async function applySchemaBaseline( await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${AGENT_RATINGS_PROJECT_PARTITION_VERSION}) ON CONFLICT (version) DO NOTHING`); schemaChanged = true; } + if (!projectOwnershipDeclarationDriftAlreadyApplied) { + const migrationSql = await readFile(PROJECT_OWNERSHIP_DECLARATION_DRIFT_MIGRATION_PATH, "utf8"); + await tx.execute(sql.raw(migrationSql)); + await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION}) ON CONFLICT (version) DO NOTHING`); + schemaChanged = true; + } return { applied: schemaChanged, pluginHooksRun: pluginHooks.length }; }); } diff --git a/packages/core/src/postgres/schema/project.ts b/packages/core/src/postgres/schema/project.ts index 3f15cc3bf3..7908532c9c 100644 --- a/packages/core/src/postgres/schema/project.ts +++ b/packages/core/src/postgres/schema/project.ts @@ -678,8 +678,13 @@ export const taskLifecycleConsumerDeadLetters = projectSchema.table("task_lifecy }, (t) => [primaryKey({ columns: [t.projectId, t.consumerId, t.eventId] })]); // ── Workflow step definitions ──────────────────────────────────────── +/* +FNXC:MultiProjectIsolation 2026-08-12-02:12: +Migration 0006 physically partitions workflow steps by project_id and rewrites their key. Declare that ownership here so bound stores can scope colliding per-project ws- identifiers without exposing projectId in public WorkflowStep payloads. +*/ export const workflowSteps = projectSchema.table("workflow_steps", { - id: text("id").primaryKey(), + projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`), + id: text("id").notNull(), templateId: text("template_id"), name: text("name").notNull(), description: text("description").notNull(), @@ -696,7 +701,10 @@ export const workflowSteps = projectSchema.table("workflow_steps", { migratedFragmentId: text("migrated_fragment_id"), createdAt: text("created_at").notNull(), updatedAt: text("updated_at").notNull(), -}); +}, (t) => [ + primaryKey({ columns: [t.projectId, t.id] }), + index("idxWorkflowStepsProjectCreatedAt").on(t.projectId, t.createdAt), +]); export const workflows = projectSchema.table("workflows", { id: text("id").primaryKey(), @@ -1714,6 +1722,10 @@ export const missionEvents = projectSchema.table("mission_events", { ]); // ── Plugins / routines / insights ─────────────────────────────────── +/* +FNXC:MultiProjectIsolation 2026-08-12-02:12: +Audit FN-8997 confirmed 0006 physically partitions this legacy compatibility table, but no runtime Drizzle read/write path reaches it: sqlite migration redirects plugin rows to central registry tables. Keep this declaration intentionally unscoped rather than introducing unused ORM surface. +*/ export const plugins = projectSchema.table("plugins", { id: text("id").primaryKey(), name: text("name").notNull(), @@ -2406,7 +2418,11 @@ export const chatRooms = projectSchema.table("chat_rooms", { name: text("name").notNull(), slug: text("slug").notNull(), description: text("description"), - projectId: text("project_id"), + /* + FNXC:MultiProjectIsolation 2026-08-12-02:12: + Migration 0006 backfills, defaults, and requires the trigger/GUC-owned partition key. Keep it distinct from ownerProjectId, which is a nullable domain attribute introduced by migration 0011. + */ + 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"), createdBy: text("created_by"), @@ -2421,18 +2437,28 @@ export const chatRooms = projectSchema.table("chat_rooms", { index("idxChatRoomsStatus").on(t.status), ]); +/* +FNXC:MultiProjectIsolation 2026-08-12-02:12: +Migration 0006 gives memberships a project-local composite key. The explicit column enables bound member predicates, including the first leg of duplicate room-id isolation. +*/ export const chatRoomMembers = projectSchema.table("chat_room_members", { + projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`), roomId: text("room_id").notNull(), agentId: text("agent_id").notNull(), role: text("role").notNull().default("member"), addedAt: text("added_at").notNull(), }, (t) => [ - primaryKey({ columns: [t.roomId, t.agentId] }), - index("idxChatRoomMembersAgentId").on(t.agentId), + primaryKey({ columns: [t.projectId, t.roomId, t.agentId] }), + index("idxChatRoomMembersAgentId").on(t.projectId, t.agentId), ]); +/* +FNXC:MultiProjectIsolation 2026-08-12-02:12: +Migration 0006 makes message ids project-local. Declare the partition so room/message reads and mutations cannot cross a duplicated room or message id. +*/ export const chatRoomMessages = projectSchema.table("chat_room_messages", { - id: text("id").primaryKey(), + projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`), + id: text("id").notNull(), roomId: text("room_id").notNull(), role: text("role").notNull(), content: text("content").notNull(), @@ -2443,8 +2469,9 @@ export const chatRoomMessages = projectSchema.table("chat_room_messages", { mentions: jsonb("mentions"), createdAt: text("created_at").notNull(), }, (t) => [ - index("idxChatRoomMessagesRoomCreatedAt").on(t.roomId, t.createdAt), - index("idxChatRoomMessagesRoomId").on(t.roomId), + primaryKey({ columns: [t.projectId, t.id] }), + index("idxChatRoomMessagesRoomCreatedAt").on(t.projectId, t.roomId, t.createdAt), + index("idxChatRoomMessagesRoomId").on(t.projectId, t.roomId), ]); /** diff --git a/packages/core/src/task-store/project-store-ops.ts b/packages/core/src/task-store/project-store-ops.ts index 40d77b83da..b7a1ea2e1e 100644 --- a/packages/core/src/task-store/project-store-ops.ts +++ b/packages/core/src/task-store/project-store-ops.ts @@ -25,7 +25,7 @@ import "../builtin-traits.js"; import {normalizeWorkflowIcon, type WorkflowDefinition, type WorkflowDefinitionInput} from "../workflows/workflow-definition-types.js"; import {normalizeTaskPriority} from "../tasks/task-priority.js"; import type {AsyncDataLayer, DbTransaction} from "../postgres/data-layer.js"; -import {recordRunAuditEventWithinTransaction} from "../postgres/data-layer.js"; +import {projectScopeFor, recordRunAuditEventWithinTransaction} from "../postgres/data-layer.js"; import {EvalStore} from "../eval/eval-store.js"; import {AsyncEvalStore} from "../async-stores/async-eval-store.js"; import {BackwardCompat, ProjectRequiredError} from "../central/migration.js"; @@ -712,6 +712,7 @@ export async function listWorkflowStepsImpl(store: TaskStore): Promise store.applyLegacyWorkflowStepOverrides(store.toStoredWorkflowStep({ @@ -751,13 +752,13 @@ export async function getWorkflowStepImpl(store: TaskStore, id: string): Promise const byIdRows = await store.asyncLayer!.db .select() .from(table) - .where(eq(table.id, id)) + .where(and(eq(table.id, id), projectScopeFor(table.projectId, store.asyncLayer!.projectId))) .limit(1); if (byIdRows[0]) return mapRow(byIdRows[0]); const byTemplateRows = await store.asyncLayer!.db .select() .from(table) - .where(eq(table.templateId, id)) + .where(and(eq(table.templateId, id), projectScopeFor(table.projectId, store.asyncLayer!.projectId))) .orderBy(table.createdAt) .limit(1); if (byTemplateRows[0]) return mapRow(byTemplateRows[0]); diff --git a/packages/core/src/task-store/workflow-definitions.ts b/packages/core/src/task-store/workflow-definitions.ts index aa4d51f338..e11a56a9df 100644 --- a/packages/core/src/task-store/workflow-definitions.ts +++ b/packages/core/src/task-store/workflow-definitions.ts @@ -654,7 +654,7 @@ export async function purgeTaskWorkflowSelectionRowsAsyncImpl(store: TaskStore, if (Array.isArray(parsed)) { for (const stepId of parsed) { if (typeof stepId === "string") { - await layer.db.delete(schema.project.workflowSteps).where(eq(schema.project.workflowSteps.id, stepId)); + await layer.db.delete(schema.project.workflowSteps).where(and(eq(schema.project.workflowSteps.id, stepId), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))); } } } @@ -679,7 +679,7 @@ export async function cleanupOrphanedMaterializedStepsImpl(store: TaskStore, ste const layer = store.getAsyncLayer(); if (layer) { try { - await layer.db.delete(schema.project.workflowSteps).where(inArray(schema.project.workflowSteps.id, stepIds)); + await layer.db.delete(schema.project.workflowSteps).where(and(inArray(schema.project.workflowSteps.id, stepIds), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))); } catch { // Best-effort cleanup. } diff --git a/packages/core/src/task-store/workflow-ops.ts b/packages/core/src/task-store/workflow-ops.ts index 14bab63e8b..fea4875318 100644 --- a/packages/core/src/task-store/workflow-ops.ts +++ b/packages/core/src/task-store/workflow-ops.ts @@ -21,7 +21,7 @@ import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js"; import * as schema from "../postgres/schema/index.js"; import {readProjectConfig, writeProjectConfig} from "../task-store/async/async-settings.js"; import {and, eq, inArray} from "drizzle-orm"; -import type {AsyncDataLayer} from "../postgres/data-layer.js"; +import {projectScopeFor, type AsyncDataLayer} from "../postgres/data-layer.js"; export async function createWorkflowStepImpl(store: TaskStore, input: import("../types.js").WorkflowStepInput): Promise { return store.withConfigLock(async () => { @@ -72,6 +72,7 @@ export async function createWorkflowStepImpl(store: TaskStore, input: import(".. }; await layer.db.insert(schema.project.workflowSteps).values({ + projectId: layer.projectId?.trim() ?? "", id: step.id, templateId: step.templateId ?? null, name: step.name, @@ -104,7 +105,7 @@ export async function updateWorkflowStepImpl(store: TaskStore, id: string, updat // FNXC:PostgresCutover 2026-06-28-10:00: // Backend-mode branch: read the step row via Drizzle, apply updates, write back. const layer = store.asyncLayer!; - const rows = await layer.db.select().from(schema.project.workflowSteps).where(eq(schema.project.workflowSteps.id, id)).limit(1); + const rows = await layer.db.select().from(schema.project.workflowSteps).where(and(eq(schema.project.workflowSteps.id, id), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))).limit(1); const pgRow = rows[0]; if (!pgRow) throw new Error(`Workflow step '${id}' not found`); @@ -167,7 +168,7 @@ export async function updateWorkflowStepImpl(store: TaskStore, id: string, updat modelId: step.modelId ?? null, migratedFragmentId: step.migratedFragmentId ?? null, updatedAt: step.updatedAt, - }).where(eq(schema.project.workflowSteps.id, id)); + }).where(and(eq(schema.project.workflowSteps.id, id), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))); store.workflowStepsCache = null; return step; @@ -408,7 +409,7 @@ export async function deleteWorkflowDefinitionImpl(store: TaskStore, id: string) if (Array.isArray(stepIds)) { for (const stepId of stepIds) { if (typeof stepId === "string") { - await layer.db.delete(schema.project.workflowSteps).where(eq(schema.project.workflowSteps.id, stepId)); + await layer.db.delete(schema.project.workflowSteps).where(and(eq(schema.project.workflowSteps.id, stepId), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))); } } } @@ -518,7 +519,7 @@ export async function selectTaskWorkflowImpl(store: TaskStore, taskId: string, w // Delete them before propagating; the prior selection is left untouched. for (const stepId of ids) { try { - await layer.db.delete(schema.project.workflowSteps).where(eq(schema.project.workflowSteps.id, stepId)); + await layer.db.delete(schema.project.workflowSteps).where(and(eq(schema.project.workflowSteps.id, stepId), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))); } catch { // Best-effort cleanup; surface the original error below. } @@ -529,7 +530,7 @@ export async function selectTaskWorkflowImpl(store: TaskStore, taskId: string, w if (priorSelection) { for (const stepId of priorSelection.stepIds) { - await layer.db.delete(schema.project.workflowSteps).where(eq(schema.project.workflowSteps.id, stepId)); + await layer.db.delete(schema.project.workflowSteps).where(and(eq(schema.project.workflowSteps.id, stepId), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))); } store.workflowStepsCache = null; } diff --git a/packages/core/src/task-store/workflow-task-create-ops.ts b/packages/core/src/task-store/workflow-task-create-ops.ts index acc40d6ade..510ce20595 100644 --- a/packages/core/src/task-store/workflow-task-create-ops.ts +++ b/packages/core/src/task-store/workflow-task-create-ops.ts @@ -514,7 +514,7 @@ export async function deleteWorkflowStepImpl(store: TaskStore, id: string): Prom const layer = store.asyncLayer!; const deletedRows = await layer.db .delete(schema.project.workflowSteps) - .where(eq(schema.project.workflowSteps.id, id)) + .where(and(eq(schema.project.workflowSteps.id, id), projectScopeFor(schema.project.workflowSteps.projectId, layer.projectId))) .returning({ id: schema.project.workflowSteps.id }); if (deletedRows.length === 0) { throw new Error(`Workflow step '${id}' not found`);