From 41bb94ba773ebc02c846e3683d0139f09ed1b1d8 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 11 Aug 2026 03:32:16 -0700 Subject: [PATCH] FN-8957: isolate PostgreSQL agents by project Keep durable PostgreSQL agent records and analytics isolated to their bound project. - Scope agent reads, mutations, satellite records, ratings, reassignment links, and analytics task data by project. - Add a project-local agent-ratings migration and schema coverage. - Exercise duplicate agent IDs, ratings, and token analytics across shared PostgreSQL projects. Files changed: .changeset/fn-8957-agent-project-isolation.md | 7 + docs/multi-project.md | 9 + .../postgres/agent-project-isolation.pg.test.ts | 207 +++++++++++++++++++++ .../src/__tests__/postgres/schema-applier.test.ts | 13 +- packages/core/src/agents/agent-store.ts | 52 +++--- .../core/src/async-stores/async-agent-store.ts | 118 ++++++++++-- packages/core/src/board/team-analytics.ts | 20 +- ...0054_fn_8957_agent_rating_project_isolation.sql | 32 ++++ packages/core/src/postgres/schema-applier.ts | 13 +- packages/core/src/postgres/schema/project.ts | 6 +- packages/core/src/task-store/task-artifacts-ops.ts | 19 +- 11 files changed, 443 insertions(+), 53 deletions(-) Fusion-Task-Id: FN-8957 Fusion-Task-Lineage: 02cb3f13-e5d1-4cea-8afb-72bcd9fae890 Co-authored-by: Fusion (runfusion.ai) --- .changeset/fn-8957-agent-project-isolation.md | 7 + docs/multi-project.md | 9 + .../agent-project-isolation.pg.test.ts | 207 ++++++++++++++++++ .../__tests__/postgres/schema-applier.test.ts | 13 +- packages/core/src/agents/agent-store.ts | 52 +++-- .../src/async-stores/async-agent-store.ts | 118 ++++++++-- packages/core/src/board/team-analytics.ts | 20 +- ...fn_8957_agent_rating_project_isolation.sql | 32 +++ packages/core/src/postgres/schema-applier.ts | 13 +- packages/core/src/postgres/schema/project.ts | 6 +- .../core/src/task-store/task-artifacts-ops.ts | 19 +- 11 files changed, 443 insertions(+), 53 deletions(-) create mode 100644 .changeset/fn-8957-agent-project-isolation.md create mode 100644 packages/core/src/__tests__/postgres/agent-project-isolation.pg.test.ts create mode 100644 packages/core/src/postgres/migrations/0054_fn_8957_agent_rating_project_isolation.sql diff --git a/.changeset/fn-8957-agent-project-isolation.md b/.changeset/fn-8957-agent-project-isolation.md new file mode 100644 index 0000000000..9c37e6889c --- /dev/null +++ b/.changeset/fn-8957-agent-project-isolation.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep durable agent data isolated to the active project in shared PostgreSQL. +category: fix +dev: Agent reads, mutations, satellite tables, analytics, and reassignment links now apply the bound project scope. diff --git a/docs/multi-project.md b/docs/multi-project.md index 0bb42cc23a..5387f9771c 100644 --- a/docs/multi-project.md +++ b/docs/multi-project.md @@ -48,6 +48,15 @@ Core `central` tables (names as exposed by the data layer; SQL uses snake_case): Per-project task data is keyed by `projectId` in PostgreSQL's `project` schema. Each repo keeps `.fusion/project.json` as its filesystem identity marker; `.fusion/fusion.db` is read only by the one-time legacy migrator. +### Agent ownership predicates + + + +Bound agent-store layers scope those operations with `projectScopeFor(..., projectId)`. An unbound or blank layer is intentionally a no-op scope for compatibility and cross-project analytics callers; it must not be converted to a literal empty `project_id` filter or a throwing project-id accessor. + Use PostgreSQL-native backup/restore tooling for authoritative runtime data. Legacy `fn backup` SQLite artifacts remain migration/recovery inputs; restoring one does not replace the live PostgreSQL registry. `taskClaims` is the central cross-node lease mutex introduced by FN-4819 §2: claim acquisition/renewal/release happen in PostgreSQL, while per-project lease fields mirror the central winner for local scheduler/runtime consumption. diff --git a/packages/core/src/__tests__/postgres/agent-project-isolation.pg.test.ts b/packages/core/src/__tests__/postgres/agent-project-isolation.pg.test.ts new file mode 100644 index 0000000000..773b879e1b --- /dev/null +++ b/packages/core/src/__tests__/postgres/agent-project-isolation.pg.test.ts @@ -0,0 +1,207 @@ +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 { AgentStore } from "../../agents/agent-store.js"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import * as schema from "../../postgres/schema/index.js"; +import { TaskStore } from "../../store.js"; +import type { Agent } from "../../types.js"; +import { aggregateTeamAnalytics } from "../../board/team-analytics.js"; +import { + addRating, + appendConfigRevision, + clearLastBlockedState, + deleteAgent, + deleteRating, + findAgentRowsByName, + getAllBlockedStates, + getHeartbeatHistory, + getLastBlockedState, + getRatings, + getTaskSession, + insertApiKey, + listAgentRows, + readAgent, + readApiKeys, + readConfigRevisions, + recordHeartbeat, + setLastBlockedState, + upsertTaskSession, + writeAgent, +} from "../../async-stores/async-agent-store.js"; + +pgDescribe("agent project isolation", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_agent_project_isolation", + }); + + beforeAll(h.beforeAll); + afterAll(h.afterAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + + it("keeps duplicate agent ids and satellite rows inside the bound project", async () => { + /* + FNXC:MultiProjectIsolation 2026-08-11-09:13: + Runfusion/Fusion#3414 reproduced duplicate durable ids in a shared owner-connected + PostgreSQL cluster. This fixture exercises production AgentStore and TaskStore paths, + proving every bound helper uses its partition while unbound compatibility remains cross-project. + */ + const bind = (projectId: string): AsyncDataLayer => ({ ...h.layer(), projectId }); + const projectA = bind("agents-project-a"); + const projectB = bind("agents-project-b"); + const sharedId = "agent-shared"; + const now = "2026-08-11T09:13:00.000Z"; + const agent = (name: string, state: Agent["state"], taskId?: string): Agent => ({ + id: sharedId, + name, + role: "executor", + roles: ["executor"], + state, + taskId, + createdAt: now, + updatedAt: now, + metadata: { project: name }, + runtimeConfig: { model: `${name}-model` }, + }); + const readRaw = async (projectId: string) => (await h.adminDb() + .select({ + projectId: schema.project.agents.projectId, + name: schema.project.agents.name, + state: schema.project.agents.state, + taskId: schema.project.agents.taskId, + metadata: schema.project.agents.metadata, + data: schema.project.agents.data, + }) + .from(schema.project.agents) + .where(and(eq(schema.project.agents.id, sharedId), eq(schema.project.agents.projectId, projectId))))[0]; + + await writeAgent(projectA.db, agent("Agent A", "idle", "task-a"), projectA.projectId); + await writeAgent(projectB.db, agent("Agent B", "active", "task-b"), projectB.projectId); + await writeAgent(projectB.db, { ...agent("Only B", "idle"), id: "agent-only-b" }, projectB.projectId); + + expect((await listAgentRows(projectA.db, undefined, projectA.projectId)).map((row) => row.name)).toEqual(["Agent A"]); + expect((await readAgent(projectA.db, sharedId, projectA.projectId))?.name).toBe("Agent A"); + expect(await readAgent(projectA.db, "agent-only-b", projectA.projectId)).toBeNull(); + expect(await findAgentRowsByName(projectA.db, "Agent B", projectA.projectId)).toEqual([]); + + // Direct writeAgent upsert updates all durable identity fields only in project A. + const beforeBDirectUpdate = await readRaw(projectB.projectId); + await writeAgent(projectA.db, { + ...agent("Agent A direct update", "paused", "task-a-updated"), + metadata: { project: "A-direct" }, + runtimeConfig: { model: "a-direct-model" }, + }, projectA.projectId); + expect(await readRaw(projectB.projectId)).toEqual(beforeBDirectUpdate); + + // AgentStore is the dashboard/engine update entry point; its delegation must retain A's layer id. + const agentStoreA = new AgentStore({ rootDir: h.rootDir(), asyncLayer: projectA }); + const beforeBAgentStoreUpdate = await readRaw(projectB.projectId); + await agentStoreA.updateAgent(sharedId, { + name: "Agent A store update", + runtimeConfig: { model: "a-store-model" }, + metadata: { project: "A-store" }, + }); + await agentStoreA.updateAgentState(sharedId, "active"); + expect((await readAgent(projectA.db, sharedId, projectA.projectId))?.name).toBe("Agent A store update"); + expect(await readRaw(projectB.projectId)).toEqual(beforeBAgentStoreUpdate); + + // Repeat through B's production update path and prove it cannot mutate A. + const agentStoreB = new AgentStore({ rootDir: h.rootDir(), asyncLayer: projectB }); + const beforeBUpdateA = await readRaw(projectA.projectId); + await agentStoreB.updateAgent(sharedId, { + name: "Agent B store update", + runtimeConfig: { model: "b-store-model" }, + metadata: { project: "B-store" }, + }); + await agentStoreB.updateAgentState(sharedId, "paused"); + expect(await readRaw(projectA.projectId)).toEqual(beforeBUpdateA); + expect((await readAgent(projectB.db, sharedId, projectB.projectId))?.name).toBe("Agent B store update"); + + await recordHeartbeat(projectA.db, { agentId: sharedId, timestamp: now, status: "ok", runId: "run-a" }, projectA.projectId); + await recordHeartbeat(projectB.db, { agentId: sharedId, timestamp: now, status: "missed", runId: "run-b" }, projectB.projectId); + expect((await getHeartbeatHistory(projectA.db, sharedId, 50, projectA.projectId)).map((row) => row.runId)).toEqual(["run-a"]); + + await upsertTaskSession(projectA.db, { agentId: sharedId, taskId: "session", createdAt: now, updatedAt: now } as never, projectA.projectId); + await upsertTaskSession(projectB.db, { agentId: sharedId, taskId: "session", createdAt: now, updatedAt: now, model: "b" } as never, projectB.projectId); + expect((await getTaskSession(projectA.db, sharedId, "session", projectA.projectId) as { model?: string } | null)?.model).toBeUndefined(); + + await insertApiKey(projectA.db, { id: "key", agentId: sharedId, tokenHash: "a", createdAt: now }, projectA.projectId); + await insertApiKey(projectB.db, { id: "key", agentId: sharedId, tokenHash: "b", createdAt: now }, projectB.projectId); + expect((await readApiKeys(projectA.db, sharedId, projectA.projectId)).map((key) => key.tokenHash)).toEqual(["a"]); + + await appendConfigRevision(projectA.db, { id: "revision-a", agentId: sharedId, createdAt: now } as never, projectA.projectId); + await appendConfigRevision(projectB.db, { id: "revision-b", agentId: sharedId, createdAt: now } as never, projectB.projectId); + expect((await readConfigRevisions(projectA.db, sharedId, projectA.projectId)).map((row) => row.id)).toEqual(expect.arrayContaining(["revision-a"])); + + await addRating(projectA.db, { id: "rating", agentId: sharedId, raterType: "user", score: 5, createdAt: now }, projectA.projectId); + await addRating(projectB.db, { id: "rating", agentId: sharedId, raterType: "user", score: 1, createdAt: now }, projectB.projectId); + expect((await getRatings(projectA.db, sharedId, undefined, projectA.projectId)).map((rating) => rating.score)).toEqual([5]); + expect(await deleteRating(projectA.db, "rating", projectA.projectId)).toBe(true); + expect((await getRatings(projectB.db, sharedId, undefined, projectB.projectId)).map((rating) => rating.score)).toEqual([1]); + + await setLastBlockedState(projectA.db, sharedId, { taskId: "blocked-a" } as never, projectA.projectId); + await setLastBlockedState(projectB.db, sharedId, { taskId: "blocked-b" } as never, projectB.projectId); + expect((await getLastBlockedState(projectA.db, sharedId, projectA.projectId) as { taskId?: string } | null)?.taskId).toBe("blocked-a"); + expect((await getAllBlockedStates(projectA.db, projectA.projectId)).map((row) => row.agentId)).toEqual([sharedId]); + await clearLastBlockedState(projectA.db, sharedId, projectA.projectId); + expect(await getLastBlockedState(projectB.db, sharedId, projectB.projectId)).not.toBeNull(); + + await h.adminDb().insert(schema.project.tasks).values([ + { + projectId: projectA.projectId, + id: "analytics-a", + description: "A analytics fixture", + column: "done", + assignedAgentId: sharedId, + tokenUsageInputTokens: 5, + tokenUsageTotalTokens: 5, + tokenUsageLastUsedAt: now, + createdAt: now, + updatedAt: now, + }, + { + projectId: projectB.projectId, + id: "analytics-b", + description: "B analytics fixture", + column: "done", + assignedAgentId: sharedId, + tokenUsageInputTokens: 99, + tokenUsageTotalTokens: 99, + tokenUsageLastUsedAt: now, + createdAt: now, + updatedAt: now, + }, + ]); + const analytics = await aggregateTeamAnalytics(projectA, { now: Date.parse(now) }); + expect(analytics.agents.map((row) => row.agentName)).toEqual(["Agent A store update"]); + expect(analytics.agents[0]?.tokens.totalTokens).toBe(5); + + // Use TaskStore's public reassignment path rather than a cast mock so both UPDATEs are covered. + const taskStoreA = new TaskStore(h.rootDir(), undefined, { asyncLayer: projectA }); + const beforeReassignmentB = await readRaw(projectB.projectId); + await taskStoreA.syncAgentTaskLinkOnReassignment("task-a-updated", sharedId, sharedId); + expect(await readRaw(projectB.projectId)).toEqual(beforeReassignmentB); + expect((await readAgent(projectA.db, sharedId, projectA.projectId))?.taskId).toBe("task-a-updated"); + + expect(await deleteAgent(projectA.db, sharedId, projectA.projectId)).toBe(true); + expect(await readAgent(projectA.db, sharedId, projectA.projectId)).toBeNull(); + expect((await readAgent(projectB.db, sharedId, projectB.projectId))?.name).toBe("Agent B store update"); + + // An undefined project id intentionally remains unscoped: it sees B's row and writes its legacy partition. + const unboundStore = new AgentStore({ + rootDir: h.rootDir(), + asyncLayer: { ...h.layer(), projectId: undefined }, + }); + expect((await listAgentRows(h.layer().db)).map((row) => row.name).sort()).toEqual(["Agent B store update", "Only B"]); + expect((await findAgentRowsByName(h.layer().db, "Agent B store update")).map((row) => row.id)).toEqual([sharedId]); + await unboundStore.updateAgent(sharedId, { name: "Agent unbound update" }); + expect((await readAgent(h.layer().db, sharedId))?.name).toBe("Agent B store update"); + expect((await listAgentRows(h.layer().db)).map((row) => row.name)).toContain("Agent unbound update"); + expect((await readAgent(projectB.db, sharedId, projectB.projectId))?.name).toBe("Agent B store update"); + }); +}); diff --git a/packages/core/src/__tests__/postgres/schema-applier.test.ts b/packages/core/src/__tests__/postgres/schema-applier.test.ts index ca14d81500..62b9682741 100644 --- a/packages/core/src/__tests__/postgres/schema-applier.test.ts +++ b/packages/core/src/__tests__/postgres/schema-applier.test.ts @@ -96,6 +96,7 @@ import { SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION, MEMORY_RECALL_RECORDS_VERSION, MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, + AGENT_RATING_PROJECT_ISOLATION_VERSION, } from "../../postgres/schema-applier.js"; import { ProjectPartitionRekeyError, rekeyFallbackProjectPartition } from "../../postgres/migration-stamping.js"; import type { PluginSchemaInitHook } from "../../postgres/plugin-schema-hook.js"; @@ -125,7 +126,8 @@ describe("schema-applier: immutable migration identities", () => { expect(SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION).toBe("0051"); expect(MEMORY_RECALL_RECORDS_VERSION).toBe("0052"); expect(MISSION_FEATURE_SPEC_ALIGNMENT_VERSION).toBe("0053"); - expect(SCHEMA_BASELINE_VERSION).toBe("0053"); + expect(AGENT_RATING_PROJECT_ISOLATION_VERSION).toBe("0054"); + expect(SCHEMA_BASELINE_VERSION).toBe("0054"); }); it("keeps monitor and approval isolation assigned to version 0003", () => { @@ -1794,6 +1796,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION, MEMORY_RECALL_RECORDS_VERSION, MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, + AGENT_RATING_PROJECT_ISOLATION_VERSION, ]); expect((await applySchemaBaseline(ctx.db, { pluginHooks: [] })).applied).toBe(false); }); @@ -1873,6 +1876,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION, MEMORY_RECALL_RECORDS_VERSION, MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, + AGENT_RATING_PROJECT_ISOLATION_VERSION, ]); }); @@ -2085,6 +2089,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION, MEMORY_RECALL_RECORDS_VERSION, MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, + AGENT_RATING_PROJECT_ISOLATION_VERSION, ]); }); @@ -2178,6 +2183,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION, MEMORY_RECALL_RECORDS_VERSION, MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, + AGENT_RATING_PROJECT_ISOLATION_VERSION, ]); }); @@ -2271,6 +2277,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION, MEMORY_RECALL_RECORDS_VERSION, MISSION_FEATURE_SPEC_ALIGNMENT_VERSION, + AGENT_RATING_PROJECT_ISOLATION_VERSION, ]); }); }); @@ -2354,8 +2361,8 @@ pgDescribe("schema-applier: VAL-SCHEMA-005 CHECK constraints preserved and enfor await applySchemaBaseline(ctx.db); await expectPgError( ctx.db.execute(sql` - INSERT INTO project.agent_ratings (id, agent_id, rater_type, score, created_at) - VALUES ('r1', 'a1', 'user', 99, '2026-01-01') + INSERT INTO project.agent_ratings (project_id, id, agent_id, rater_type, score, created_at) + VALUES ('schema-test', 'r1', 'a1', 'user', 99, '2026-01-01') `), /score_check|check constraint/i, ); diff --git a/packages/core/src/agents/agent-store.ts b/packages/core/src/agents/agent-store.ts index 855a712b12..ff88c3483d 100644 --- a/packages/core/src/agents/agent-store.ts +++ b/packages/core/src/agents/agent-store.ts @@ -769,7 +769,11 @@ export class AgentStore extends EventEmitter { async findAgentByName(name: string, executor?: QueryHandle): Promise { // FNXC:SqliteFinalRemoval 2026-06-25-23:45: // Backend mode: read via async Drizzle helper, filter ephemeral in-memory. - const agents = await findAgentRowsByNameAsync(executor ?? this.asyncLayer!.db, name); + const agents = await findAgentRowsByNameAsync( + executor ?? this.asyncLayer!.db, + name, + this.workflowProjectId, + ); for (const agent of agents) { if (!isEphemeralAgent(agent)) { return this.parseAgent(agent as unknown as AgentData); @@ -889,7 +893,7 @@ export class AgentStore extends EventEmitter { FNXC:SqliteDualPathCleanup 2026-07-26-14:05: Agent reads are PostgreSQL-only via readAgentAsync. Populate getCachedAgent memory so sync heartbeat resolveAgentConfig can honor per-agent runtimeConfig without a SQLite handle. */ - const agent = await readAgentAsync(this.asyncLayer!.db, agentId); + const agent = await readAgentAsync(this.asyncLayer!.db, agentId, this.workflowProjectId); const parsed = agent ? this.parseAgent(agent) : null; if (parsed) this.agentMemoryCache.set(agentId, parsed); else this.agentMemoryCache.delete(agentId); @@ -1025,7 +1029,7 @@ export class AgentStore extends EventEmitter { * Backend-mode: delegate to async Drizzle addRating helper. The score CHECK * constraint is enforced by PostgreSQL (VAL-SCHEMA-005). */ - const saved = await addRatingAsync(this.asyncLayer!.db, rating); + const saved = await addRatingAsync(this.asyncLayer!.db, rating, this.workflowProjectId); this.emit("rating:added", saved); return saved; } @@ -1035,7 +1039,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:15: * Backend-mode: delegate to async Drizzle getRatings helper. */ - return getRatingsAsync(this.asyncLayer!.db, agentId, options); + return getRatingsAsync(this.asyncLayer!.db, agentId, options, this.workflowProjectId); } async getRatingSummary(agentId: string): Promise { @@ -1103,7 +1107,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:15: * Backend-mode: delegate to async Drizzle deleteRating helper. */ - await deleteRatingAsync(this.asyncLayer!.db, ratingId); + await deleteRatingAsync(this.asyncLayer!.db, ratingId, this.workflowProjectId); return; } @@ -2006,7 +2010,11 @@ export class AgentStore extends EventEmitter { // FNXC:WorkflowAgentRouting 2026-08-07-03:12: // Role-pool membership is canonical multi-tag state, so SQL must not use the // deprecated singular projection to exclude a matching durable principal. - const agents = await listAgentRowsAsync(executor ?? this.asyncLayer!.db, { state: filter?.state }); + const agents = await listAgentRowsAsync( + executor ?? this.asyncLayer!.db, + { state: filter?.state }, + this.workflowProjectId, + ); return agents .map((a) => this.parseAgent(a as unknown as AgentData)) .filter((agent) => !filter?.role || agent.roles.includes(filter.role)) @@ -2159,7 +2167,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:20: * Backend-mode: delegate to async Drizzle insertApiKey helper. */ - await insertApiKeyAsync(this.asyncLayer!.db, key); + await insertApiKeyAsync(this.asyncLayer!.db, key, this.workflowProjectId); return { key, token }; }); @@ -2208,7 +2216,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:20: * Backend-mode: delegate to async Drizzle revokeApiKeyRow helper. */ - await revokeApiKeyRowAsync(this.asyncLayer!.db, keyId, agentId, revoked); + await revokeApiKeyRowAsync(this.asyncLayer!.db, keyId, agentId, revoked, this.workflowProjectId); return revoked; }); @@ -2243,7 +2251,7 @@ export class AgentStore extends EventEmitter { // FNXC:SqliteFinalRemoval 2026-06-25-23:55: // Backend mode: delete via async Drizzle helper (cascading FKs handle // heartbeats, runs, task sessions, API keys, config revisions, etc.). - await deleteAgentAsync(this.asyncLayer!.db, agentId); + await deleteAgentAsync(this.asyncLayer!.db, agentId, this.workflowProjectId); // FN-7723: keep this instance's own change-detection snapshot in sync // with its own delete so a later poll never mistakes the row's absence @@ -2297,7 +2305,7 @@ export class AgentStore extends EventEmitter { timestamp: event.timestamp, status: event.status, runId: event.runId, - }); + }, this.workflowProjectId); // Update agent's lastHeartbeatAt if status is ok if (status === "ok") { @@ -2330,8 +2338,8 @@ export class AgentStore extends EventEmitter { async getHeartbeatHistory(agentId: string, limit = 50): Promise { // FNXC:SqliteFinalRemoval 2026-06-26-00:05: // Backend mode: read via async Drizzle helper. - void this.backendProjectId; - return getHeartbeatHistoryAsync(this.asyncLayer!.db, agentId, limit); + void this.backendProjectId; + return getHeartbeatHistoryAsync(this.asyncLayer!.db, agentId, limit, this.workflowProjectId); } /** @@ -2455,7 +2463,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:30: * Backend-mode: delegate to async Drizzle getTaskSession helper. */ - return getTaskSessionAsync(this.asyncLayer!.db, agentId, taskId); + return getTaskSessionAsync(this.asyncLayer!.db, agentId, taskId, this.workflowProjectId); } /** @@ -2477,7 +2485,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:30: * Backend-mode: delegate to async Drizzle upsertTaskSession helper. */ - await upsertTaskSessionAsync(this.asyncLayer!.db, saved); + await upsertTaskSessionAsync(this.asyncLayer!.db, saved, this.workflowProjectId); return saved; } @@ -2492,7 +2500,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:30: * Backend-mode: delegate to async Drizzle deleteTaskSession helper. */ - await deleteTaskSessionAsync(this.asyncLayer!.db, agentId, taskId); + await deleteTaskSessionAsync(this.asyncLayer!.db, agentId, taskId, this.workflowProjectId); return; } @@ -2731,7 +2739,7 @@ export class AgentStore extends EventEmitter { */ async getLastBlockedState(agentId: string): Promise { // FNXC:PostgresCutover 2026-07-04: delegate to async Drizzle helper in backend mode. - return getLastBlockedStateAsync(this.asyncLayer!.db, agentId); + return getLastBlockedStateAsync(this.asyncLayer!.db, agentId, this.workflowProjectId); } /** @@ -2740,7 +2748,7 @@ export class AgentStore extends EventEmitter { async setLastBlockedState(agentId: string, state: BlockedStateSnapshot): Promise { await this.withLock(agentId, async () => { // FNXC:PostgresCutover 2026-07-04: delegate to async Drizzle helper in backend mode. - await setLastBlockedStateAsync(this.asyncLayer!.db, agentId, state); + await setLastBlockedStateAsync(this.asyncLayer!.db, agentId, state, this.workflowProjectId); return; }); } @@ -2751,7 +2759,7 @@ export class AgentStore extends EventEmitter { async clearLastBlockedState(agentId: string): Promise { await this.withLock(agentId, async () => { // FNXC:PostgresCutover 2026-07-04: delegate to async Drizzle helper in backend mode. - await clearLastBlockedStateAsync(this.asyncLayer!.db, agentId); + await clearLastBlockedStateAsync(this.asyncLayer!.db, agentId, this.workflowProjectId); return; }); } @@ -2766,13 +2774,13 @@ export class AgentStore extends EventEmitter { private async appendConfigRevision(revision: AgentConfigRevision): Promise { // FNXC:SqliteFinalRemoval 2026-06-26-00:10: backend mode async delegation. - await appendConfigRevisionAsync(this.asyncLayer!.db, revision); + await appendConfigRevisionAsync(this.asyncLayer!.db, revision, this.workflowProjectId); return; } private async readConfigRevisions(agentId: string): Promise { // FNXC:SqliteFinalRemoval 2026-06-26-00:10: backend mode async delegation. - return readConfigRevisionsAsync(this.asyncLayer!.db, agentId); + return readConfigRevisionsAsync(this.asyncLayer!.db, agentId, this.workflowProjectId); } private createConfigRevision(params: { @@ -2862,7 +2870,7 @@ export class AgentStore extends EventEmitter { private async findConfigRevisionAcrossAgents(revisionId: string): Promise { // FNXC:PostgresCutover 2026-07-04: delegate to async Drizzle helper in backend mode. - return findConfigRevisionByIdAsync(this.asyncLayer!.db, revisionId); + return findConfigRevisionByIdAsync(this.asyncLayer!.db, revisionId, this.workflowProjectId); } private computeNextResetAt(period: AgentBudgetConfig["budgetPeriod"], resetDay?: number): string | null { @@ -3098,7 +3106,7 @@ export class AgentStore extends EventEmitter { * FNXC:SqliteFinalRemoval 2026-06-26-09:20: * Backend-mode: delegate to async Drizzle readApiKeys helper. */ - return readApiKeysAsync(this.asyncLayer!.db, agentId); + return readApiKeysAsync(this.asyncLayer!.db, agentId, this.workflowProjectId); } private readAgent(_agentId: string): Agent | null { diff --git a/packages/core/src/async-stores/async-agent-store.ts b/packages/core/src/async-stores/async-agent-store.ts index eef3b0fbe6..16a4112568 100644 --- a/packages/core/src/async-stores/async-agent-store.ts +++ b/packages/core/src/async-stores/async-agent-store.ts @@ -45,7 +45,7 @@ */ import { and, asc, desc, eq, inArray, sql } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; -import { projectOwnershipPartition, type AsyncDataLayer, type DbTransaction } from "../postgres/data-layer.js"; +import { projectOwnershipPartition, projectScopeFor, type AsyncDataLayer, type DbTransaction } from "../postgres/data-layer.js"; import type { Agent, AgentState, @@ -226,17 +226,30 @@ export async function writeAgent(handle: QueryHandle, agent: Agent, projectId?: /** * Read a single agent by id, or null if not found. * + * FNXC:MultiProjectIsolation 2026-08-11-09:08: + * Runfusion/Fusion#3414 requires reads and deletes to carry the same ownership + * predicate as writes because external PostgreSQL owner/superuser connections + * bypass RLS. An unbound layer deliberately passes no scope for compatibility + * and cross-project analytics callers. + * * FNXC:AgentStore 2026-06-24-14:15: * The jsonb `data` column holds the extended fields; the indexed columns hold * the identity/state fields. The two are merged back into an Agent. The caller * is responsible for applying ephemeral/permission-policy normalization * (parseAgent in the sync store) — this helper returns the raw merged shape. */ -export async function readAgent(handle: QueryHandle, agentId: string): Promise { +export async function readAgent( + handle: QueryHandle, + agentId: string, + projectId?: string, +): Promise { const rows = await handle .select(agentColumns) .from(schema.project.agents) - .where(eq(schema.project.agents.id, agentId)); + .where(and( + eq(schema.project.agents.id, agentId), + projectScopeFor(schema.project.agents.projectId, projectId), + )); const row = rows[0] as AgentRow | undefined; if (!row) return null; return mergeAgentRow(row); @@ -269,8 +282,9 @@ export function mergeAgentRow(row: AgentRow): Agent { export async function listAgentRows( handle: QueryHandle, filter?: { state?: AgentState; role?: AgentCapability }, + projectId?: string, ): Promise { - const conditions = []; + const conditions = [projectScopeFor(schema.project.agents.projectId, projectId)]; if (filter?.state) { conditions.push(eq(schema.project.agents.state, filter.state)); } @@ -293,11 +307,15 @@ export async function listAgentRows( export async function findAgentRowsByName( handle: QueryHandle, name: string, + projectId?: string, ): Promise { const rows = await handle .select(agentColumns) .from(schema.project.agents) - .where(eq(schema.project.agents.name, name)) + .where(and( + eq(schema.project.agents.name, name), + projectScopeFor(schema.project.agents.projectId, projectId), + )) .orderBy(desc(schema.project.agents.createdAt), desc(schema.project.agents.id)); return rows.map((row) => mergeAgentRow(row as AgentRow)); } @@ -306,10 +324,17 @@ export async function findAgentRowsByName( * Delete an agent by id. Cascading foreign keys remove heartbeats, runs, * task sessions, API keys, config revisions, and blocked states. */ -export async function deleteAgent(handle: QueryHandle, agentId: string): Promise { +export async function deleteAgent( + handle: QueryHandle, + agentId: string, + projectId?: string, +): Promise { const result = await handle .delete(schema.project.agents) - .where(eq(schema.project.agents.id, agentId)) + .where(and( + eq(schema.project.agents.id, agentId), + projectScopeFor(schema.project.agents.projectId, projectId), + )) .returning({ id: schema.project.agents.id }); return result.length > 0; } @@ -324,8 +349,10 @@ export async function deleteAgent(handle: QueryHandle, agentId: string): Promise export async function recordHeartbeat( handle: QueryHandle, event: { agentId: string; timestamp: string; status: AgentHeartbeatEvent["status"]; runId: string }, + projectId?: string, ): Promise { await handle.insert(schema.project.agentHeartbeats).values({ + projectId: projectId?.trim() || "", agentId: event.agentId, timestamp: event.timestamp, status: event.status, @@ -345,11 +372,15 @@ export async function getHeartbeatHistory( handle: QueryHandle, agentId: string, limit = 50, + projectId?: string, ): Promise { const rows = await handle .select(heartbeatColumns) .from(schema.project.agentHeartbeats) - .where(eq(schema.project.agentHeartbeats.agentId, agentId)) + .where(and( + eq(schema.project.agentHeartbeats.agentId, agentId), + projectScopeFor(schema.project.agentHeartbeats.projectId, projectId), + )) .orderBy(desc(schema.project.agentHeartbeats.timestamp)) .limit(limit); return (rows as AgentHeartbeatRow[]).map((row) => ({ @@ -582,6 +613,7 @@ export async function getTaskSession( handle: QueryHandle, agentId: string, taskId: string, + projectId?: string, ): Promise { const rows = await handle .select({ data: schema.project.agentTaskSessions.data }) @@ -590,6 +622,7 @@ export async function getTaskSession( and( eq(schema.project.agentTaskSessions.agentId, agentId), eq(schema.project.agentTaskSessions.taskId, taskId), + projectScopeFor(schema.project.agentTaskSessions.projectId, projectId), ), ); return (rows[0]?.data as AgentTaskSession | undefined) ?? null; @@ -602,9 +635,10 @@ export async function getTaskSession( export async function upsertTaskSession( handle: QueryHandle, session: AgentTaskSession, + projectId?: string, ): Promise { const now = new Date().toISOString(); - const existing = await getTaskSession(handle, session.agentId, session.taskId); + const existing = await getTaskSession(handle, session.agentId, session.taskId, projectId); const saved: AgentTaskSession = { ...session, createdAt: existing?.createdAt ?? now, @@ -613,6 +647,7 @@ export async function upsertTaskSession( await handle .insert(schema.project.agentTaskSessions) .values({ + projectId: projectId?.trim() || "", agentId: session.agentId, taskId: session.taskId, data: saved, @@ -640,6 +675,7 @@ export async function deleteTaskSession( handle: QueryHandle, agentId: string, taskId: string, + projectId?: string, ): Promise { await handle .delete(schema.project.agentTaskSessions) @@ -647,6 +683,7 @@ export async function deleteTaskSession( and( eq(schema.project.agentTaskSessions.agentId, agentId), eq(schema.project.agentTaskSessions.taskId, taskId), + projectScopeFor(schema.project.agentTaskSessions.projectId, projectId), ), ); } @@ -659,11 +696,15 @@ export async function deleteTaskSession( export async function readApiKeys( handle: QueryHandle, agentId: string, + projectId?: string, ): Promise { const rows = await handle .select({ data: schema.project.agentApiKeys.data }) .from(schema.project.agentApiKeys) - .where(eq(schema.project.agentApiKeys.agentId, agentId)) + .where(and( + eq(schema.project.agentApiKeys.agentId, agentId), + projectScopeFor(schema.project.agentApiKeys.projectId, projectId), + )) .orderBy(asc(schema.project.agentApiKeys.createdAt)); return rows .map((row) => (row.data as AgentApiKey | null) ?? null) @@ -678,8 +719,10 @@ export async function readApiKeys( export async function insertApiKey( handle: QueryHandle, key: AgentApiKey, + projectId?: string, ): Promise { await handle.insert(schema.project.agentApiKeys).values({ + projectId: projectId?.trim() || "", id: key.id, agentId: key.agentId, data: key, @@ -696,6 +739,7 @@ export async function revokeApiKeyRow( keyId: string, agentId: string, revoked: AgentApiKey, + projectId?: string, ): Promise { await handle .update(schema.project.agentApiKeys) @@ -704,6 +748,7 @@ export async function revokeApiKeyRow( and( eq(schema.project.agentApiKeys.id, keyId), eq(schema.project.agentApiKeys.agentId, agentId), + projectScopeFor(schema.project.agentApiKeys.projectId, projectId), ), ); } @@ -716,8 +761,10 @@ export async function revokeApiKeyRow( export async function appendConfigRevision( handle: QueryHandle, revision: AgentConfigRevision, + projectId?: string, ): Promise { await handle.insert(schema.project.agentConfigRevisions).values({ + projectId: projectId?.trim() || "", id: revision.id, agentId: revision.agentId, data: revision, @@ -731,11 +778,15 @@ export async function appendConfigRevision( export async function readConfigRevisions( handle: QueryHandle, agentId: string, + projectId?: string, ): Promise { const rows = await handle .select({ data: schema.project.agentConfigRevisions.data }) .from(schema.project.agentConfigRevisions) - .where(eq(schema.project.agentConfigRevisions.agentId, agentId)) + .where(and( + eq(schema.project.agentConfigRevisions.agentId, agentId), + projectScopeFor(schema.project.agentConfigRevisions.projectId, projectId), + )) .orderBy(asc(schema.project.agentConfigRevisions.createdAt)); return rows .map((row) => (row.data as AgentConfigRevision | null) ?? null) @@ -748,11 +799,15 @@ export async function readConfigRevisions( export async function findConfigRevisionById( handle: QueryHandle, revisionId: string, + projectId?: string, ): Promise { const rows = await handle .select({ data: schema.project.agentConfigRevisions.data }) .from(schema.project.agentConfigRevisions) - .where(eq(schema.project.agentConfigRevisions.id, revisionId)); + .where(and( + eq(schema.project.agentConfigRevisions.id, revisionId), + projectScopeFor(schema.project.agentConfigRevisions.projectId, projectId), + )); return (rows[0]?.data as AgentConfigRevision | undefined) ?? null; } @@ -762,12 +817,18 @@ export async function findConfigRevisionById( * FNXC:AgentStore 2026-06-24-14:55: * Add a rating. The `score` CHECK constraint (BETWEEN 1 AND 5) is enforced by * PostgreSQL (VAL-SCHEMA-005); a violation rejects the insert. + * + * FNXC:MultiProjectIsolation 2026-08-11-10:25: + * Ratings share project-local agent IDs, so each write and lookup receives the + * bound project scope even when an owner connection bypasses PostgreSQL RLS. */ export async function addRating( handle: QueryHandle, rating: AgentRating, + projectId?: string, ): Promise { await handle.insert(schema.project.agentRatings).values({ + projectId: projectId?.trim() || "", id: rating.id, agentId: rating.agentId, raterType: rating.raterType, @@ -805,8 +866,12 @@ export async function getRatings( handle: QueryHandle, agentId: string, options?: { limit?: number; category?: string }, + projectId?: string, ): Promise { - const conditions = [eq(schema.project.agentRatings.agentId, agentId)]; + const conditions = [ + eq(schema.project.agentRatings.agentId, agentId), + projectScopeFor(schema.project.agentRatings.projectId, projectId), + ]; if (options?.category !== undefined) { conditions.push(eq(schema.project.agentRatings.category, options.category)); } @@ -824,10 +889,17 @@ export async function getRatings( /** * Delete a rating by id. */ -export async function deleteRating(handle: QueryHandle, ratingId: string): Promise { +export async function deleteRating( + handle: QueryHandle, + ratingId: string, + projectId?: string, +): Promise { const result = await handle .delete(schema.project.agentRatings) - .where(eq(schema.project.agentRatings.id, ratingId)) + .where(and( + eq(schema.project.agentRatings.id, ratingId), + projectScopeFor(schema.project.agentRatings.projectId, projectId), + )) .returning({ id: schema.project.agentRatings.id }); return result.length > 0; } @@ -840,11 +912,15 @@ export async function deleteRating(handle: QueryHandle, ratingId: string): Promi export async function getLastBlockedState( handle: QueryHandle, agentId: string, + projectId?: string, ): Promise { const rows = await handle .select({ data: schema.project.agentBlockedStates.data }) .from(schema.project.agentBlockedStates) - .where(eq(schema.project.agentBlockedStates.agentId, agentId)); + .where(and( + eq(schema.project.agentBlockedStates.agentId, agentId), + projectScopeFor(schema.project.agentBlockedStates.projectId, projectId), + )); return (rows[0]?.data as BlockedStateSnapshot | undefined) ?? null; } @@ -856,11 +932,13 @@ export async function setLastBlockedState( handle: QueryHandle, agentId: string, state: BlockedStateSnapshot, + projectId?: string, ): Promise { const updatedAt = new Date().toISOString(); await handle .insert(schema.project.agentBlockedStates) .values({ + projectId: projectId?.trim() || "", agentId, data: state, updatedAt, @@ -880,10 +958,14 @@ export async function setLastBlockedState( export async function clearLastBlockedState( handle: QueryHandle, agentId: string, + projectId?: string, ): Promise { await handle .delete(schema.project.agentBlockedStates) - .where(eq(schema.project.agentBlockedStates.agentId, agentId)); + .where(and( + eq(schema.project.agentBlockedStates.agentId, agentId), + projectScopeFor(schema.project.agentBlockedStates.projectId, projectId), + )); } /** @@ -892,6 +974,7 @@ export async function clearLastBlockedState( */ export async function getAllBlockedStates( handle: QueryHandle, + projectId?: string, ): Promise> { const rows = await handle .select({ @@ -900,6 +983,7 @@ export async function getAllBlockedStates( updatedAt: schema.project.agentBlockedStates.updatedAt, }) .from(schema.project.agentBlockedStates) + .where(projectScopeFor(schema.project.agentBlockedStates.projectId, projectId)) .orderBy(asc(schema.project.agentBlockedStates.updatedAt), asc(schema.project.agentBlockedStates.agentId)); return rows .map((row) => { diff --git a/packages/core/src/board/team-analytics.ts b/packages/core/src/board/team-analytics.ts index 532ad56dd6..77e40c2f53 100644 --- a/packages/core/src/board/team-analytics.ts +++ b/packages/core/src/board/team-analytics.ts @@ -360,8 +360,18 @@ async function aggregateTeamAnalyticsAsync( interpolated as SQL text. */ const inList = (lanes: readonly string[]) => sql.join(lanes.map((lane) => sql`${lane}`), sql`, `); + /* + FNXC:MultiProjectIsolation 2026-08-11-09:13: + Runfusion/Fusion#3414 requires analytics to carry the agent ownership predicate: + owner/superuser PostgreSQL connections bypass RLS. An unbound layer remains an + intentional cross-project analytics reader, so its predicate is a no-op. + */ + const agentProjectId = layer.projectId?.trim(); + const agentScope = agentProjectId ? sql`WHERE project_id = ${agentProjectId}` : sql``; + // FNXC:MultiProjectIsolation 2026-08-11-10:25: Bound analytics must scope every task-derived metric, not only its agent directory, because duplicate agent IDs can otherwise import another project's token usage. + const taskScope = agentProjectId ? sql`AND project_id = ${agentProjectId}` : sql``; const agents = (await layer.db.execute( - sql`SELECT id, name, role, state FROM project.agents ORDER BY id`, + sql`SELECT id, name, role, state FROM project.agents ${agentScope} ORDER BY id`, )) as unknown as AgentRow[]; const tokFrom = query.from !== undefined ? sql`AND token_usage_last_used_at >= ${query.from}` : sql``; @@ -379,7 +389,7 @@ async function aggregateTeamAnalyticsAsync( token_usage_model_provider AS "tokenUsageModelProvider", token_usage_model_id AS "tokenUsageModelId" FROM project.tasks - WHERE assigned_agent_id IS NOT NULL AND token_usage_last_used_at IS NOT NULL ${tokFrom} ${tokTo}`, + WHERE assigned_agent_id IS NOT NULL AND token_usage_last_used_at IS NOT NULL ${taskScope} ${tokFrom} ${tokTo}`, )) as Array>; const tokenRows: TaskTokenRow[] = tokenRowsRaw.map((r) => ({ agentId: String(r.agentId), @@ -399,14 +409,14 @@ async function aggregateTeamAnalyticsAsync( const completedRows = (await layer.db.execute( sql`SELECT assigned_agent_id AS "agentId", count(*)::int AS count FROM project.tasks - WHERE assigned_agent_id IS NOT NULL AND "column" IN (${inList(completeLanes)}) AND column_moved_at IS NOT NULL ${compFrom} ${compTo} + WHERE assigned_agent_id IS NOT NULL AND "column" IN (${inList(completeLanes)}) AND column_moved_at IS NOT NULL ${taskScope} ${compFrom} ${compTo} GROUP BY assigned_agent_id`, )) as unknown as CountByAgentRow[]; const currentRows = (await layer.db.execute( sql`SELECT assigned_agent_id AS "agentId", "column" AS "columnName", count(*)::int AS count FROM project.tasks - WHERE assigned_agent_id IS NOT NULL AND "column" IN (${inList(activeLanes)}) + WHERE assigned_agent_id IS NOT NULL AND "column" IN (${inList(activeLanes)}) ${taskScope} GROUP BY assigned_agent_id, "column"`, )) as unknown as Array; @@ -419,7 +429,7 @@ async function aggregateTeamAnalyticsAsync( AND modified_files IS NOT NULL AND jsonb_typeof(modified_files) = 'array' AND jsonb_array_length(modified_files) > 0 - ${filesFrom} ${filesTo}`, + ${taskScope} ${filesFrom} ${filesTo}`, )) as Array<{ agentId: string; modifiedFiles: unknown }>; const fileRows: ModifiedFilesRow[] = fileRowsRaw.map((r) => ({ agentId: String(r.agentId), diff --git a/packages/core/src/postgres/migrations/0054_fn_8957_agent_rating_project_isolation.sql b/packages/core/src/postgres/migrations/0054_fn_8957_agent_rating_project_isolation.sql new file mode 100644 index 0000000000..02912d7f75 --- /dev/null +++ b/packages/core/src/postgres/migrations/0054_fn_8957_agent_rating_project_isolation.sql @@ -0,0 +1,32 @@ +/* +FNXC:MultiProjectIsolation 2026-08-11-10:25: +FN-8957 requires agent ratings to use the same project-local identity as durable agents. Owner and superuser PostgreSQL connections bypass RLS, so the application predicates and composite primary key must prevent duplicate agent and rating IDs from crossing projects. +*/ +DO $$ +BEGIN + -- Partial historical schemas used by upgrade tests may not have this satellite table yet. + IF to_regclass('project.agent_ratings') IS NULL THEN + RETURN; + END IF; + + ALTER TABLE project.agent_ratings + ADD COLUMN IF NOT EXISTS project_id text; + + UPDATE project.agent_ratings + SET project_id = COALESCE( + NULLIF(project_id, ''), + NULLIF(current_setting('fusion.project_id', true), ''), + '__legacy_unscoped__' + ) + WHERE project_id IS NULL OR project_id = ''; + + ALTER TABLE project.agent_ratings + ALTER COLUMN project_id SET DEFAULT COALESCE(NULLIF(current_setting('fusion.project_id', true), ''), '__legacy_unscoped__'), + ALTER COLUMN project_id SET NOT NULL, + DROP CONSTRAINT IF EXISTS agent_ratings_pkey, + ADD CONSTRAINT agent_ratings_pkey PRIMARY KEY (project_id, id); + + DROP INDEX IF EXISTS project."idxAgentRatingsAgentId"; + CREATE INDEX "idxAgentRatingsAgentId" + ON project.agent_ratings(project_id, agent_id); +END $$; diff --git a/packages/core/src/postgres/schema-applier.ts b/packages/core/src/postgres/schema-applier.ts index c1ad96e3ad..c5601fc9d1 100644 --- a/packages/core/src/postgres/schema-applier.ts +++ b/packages/core/src/postgres/schema-applier.ts @@ -61,7 +61,8 @@ capacity-model table drop that landed while this PR was open. /* FNXC:SpecLock 2026-08-09-18:17: 0050 stores immutable plan history and 0051 widens source revisions before Date.now()-based writes. */ /* 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. */ -export const SCHEMA_BASELINE_VERSION = "0053"; +/* 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 = "0054"; /** 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"; @@ -210,6 +211,8 @@ export const SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION = "0051"; export const MEMORY_RECALL_RECORDS_VERSION = "0052"; /** FNXC:SpecLockMissionAlignment 2026-08-10-16:17: durable feature alignment is registered after all existing migration identities. */ export const MISSION_FEATURE_SPEC_ALIGNMENT_VERSION = "0053"; +/** FNXC:MultiProjectIsolation 2026-08-11-10:25: keep rating identity project-local after the universal ownership migration. */ +export const AGENT_RATING_PROJECT_ISOLATION_VERSION = "0054"; /** SECURITY DEFINER helper that only inserts LEGACY_ADOPTION_DRAINED_MARKER. */ export const LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION = "fusion_mark_legacy_adoption_drained"; @@ -437,6 +440,7 @@ const SPEC_LOCK_DRIFT_REPORT_MIGRATION_PATH = join(MIGRATIONS_DIR, "0050_spec_lo const SPEC_LOCK_SOURCE_REVISION_BIGINT_MIGRATION_PATH = join(MIGRATIONS_DIR, "0051_spec_lock_source_revision_bigint.sql"); const MEMORY_RECALL_RECORDS_MIGRATION_PATH = join(MIGRATIONS_DIR, "0052_fn_8922_memory_recall_records.sql"); 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"); /** * Ensure the migration bookkeeping table exists. Lives in the public schema so @@ -560,6 +564,7 @@ export async function applySchemaBaseline( const specLockSourceRevisionBigintAlreadyApplied = applied.includes(SPEC_LOCK_SOURCE_REVISION_BIGINT_VERSION); const memoryRecallRecordsAlreadyApplied = applied.includes(MEMORY_RECALL_RECORDS_VERSION); const missionFeatureSpecAlignmentAlreadyApplied = applied.includes(MISSION_FEATURE_SPEC_ALIGNMENT_VERSION); + const agentRatingProjectIsolationAlreadyApplied = applied.includes(AGENT_RATING_PROJECT_ISOLATION_VERSION); assertBinaryNotOlderThanDatabase(applied); let schemaChanged = false; @@ -1230,6 +1235,12 @@ export async function applySchemaBaseline( await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${MISSION_FEATURE_SPEC_ALIGNMENT_VERSION}) ON CONFLICT (version) DO NOTHING`); schemaChanged = true; } + if (!agentRatingProjectIsolationAlreadyApplied) { + const migrationSql = await readFile(AGENT_RATING_PROJECT_ISOLATION_MIGRATION_PATH, "utf8"); + await tx.execute(sql.raw(migrationSql)); + await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${AGENT_RATING_PROJECT_ISOLATION_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 10481f6893..8107a0007c 100644 --- a/packages/core/src/postgres/schema/project.ts +++ b/packages/core/src/postgres/schema/project.ts @@ -2042,7 +2042,8 @@ export const messages = projectSchema.table("messages", { ]); export const agentRatings = projectSchema.table("agent_ratings", { - id: text("id").primaryKey(), + projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`), + id: text("id").notNull(), agentId: text("agent_id").notNull(), raterType: text("rater_type").notNull(), raterId: text("rater_id"), @@ -2053,8 +2054,9 @@ export const agentRatings = projectSchema.table("agent_ratings", { taskId: text("task_id"), createdAt: text("created_at").notNull(), }, (t) => [ + primaryKey({ columns: [t.projectId, t.id] }), check("agent_ratings_score_check", sql`${t.score} BETWEEN 1 AND 5`), - index("idxAgentRatingsAgentId").on(t.agentId), + index("idxAgentRatingsAgentId").on(t.projectId, t.agentId), index("idxAgentRatingsCreatedAt").on(t.createdAt), ]); diff --git a/packages/core/src/task-store/task-artifacts-ops.ts b/packages/core/src/task-store/task-artifacts-ops.ts index 2f16113705..c0f1daf561 100644 --- a/packages/core/src/task-store/task-artifacts-ops.ts +++ b/packages/core/src/task-store/task-artifacts-ops.ts @@ -16,7 +16,7 @@ import {resolveWorkflowIrForTask} from "../workflows/workflow-ir-resolver.js"; import {toTaskMoveLanes} from "../workflows/workflow-lifecycle-traits.js"; import { countAgentLogEntries, readAgentLogEntries } from "../agents/agent-log-file-store.js"; import { toJsonNullable } from "../db/db.js"; -import { DbTransaction, recordRunAuditEventWithinTransaction } from "../postgres/data-layer.js"; +import { DbTransaction, projectScopeFor, recordRunAuditEventWithinTransaction } from "../postgres/data-layer.js"; import { and, eq, inArray, isNull, ne } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; import { runCommandAsync } from "../process/run-command.js"; @@ -272,19 +272,32 @@ export async function syncAgentTaskLinkOnReassignmentImpl(store: TaskStore, /* FNXC:PostgresCutover 2026-07-04-00:00: Backend-mode agent-task-link sync: update the agents.taskId column via async Drizzle. Only the dedicated taskId column is authoritative in PG (agent.data jsonb is not read for the link), so the SQLite json_set/json_remove on data is not mirrored. + + FNXC:MultiProjectIsolation 2026-08-11-09:13: + Runfusion/Fusion#3414 requires reassignment updates to carry project ownership because + owner/superuser PostgreSQL connections bypass RLS. An unbound store intentionally leaves + the scope empty for compatibility callers. */ const db = store.asyncLayer!.db; + const projectId = store.asyncLayer?.projectId; if (previousAgentId) { await db .update(schema.project.agents) .set({ taskId: null, updatedAt }) - .where(and(eq(schema.project.agents.id, previousAgentId), eq(schema.project.agents.taskId, taskId))); + .where(and( + eq(schema.project.agents.id, previousAgentId), + eq(schema.project.agents.taskId, taskId), + projectScopeFor(schema.project.agents.projectId, projectId), + )); } if (newAgentId) { await db .update(schema.project.agents) .set({ taskId, updatedAt }) - .where(eq(schema.project.agents.id, newAgentId)); + .where(and( + eq(schema.project.agents.id, newAgentId), + projectScopeFor(schema.project.agents.projectId, projectId), + )); } return; }