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) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-8957-agent-project-isolation.md
Normal file
7
.changeset/fn-8957-agent-project-isolation.md
Normal file
@@ -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.
|
||||
@@ -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
|
||||
|
||||
<!--
|
||||
FNXC:MultiProjectIsolation 2026-08-11-09:31:
|
||||
Runfusion/Fusion#3414 requires every `project.agents` read, update, and delete—plus its agent-owned satellite rows—to carry the same ownership predicate as writes. External PostgreSQL deployments commonly use owner or superuser connections that bypass RLS, so application predicates remain the isolation boundary.
|
||||
-->
|
||||
|
||||
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.
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
);
|
||||
|
||||
@@ -769,7 +769,11 @@ export class AgentStore extends EventEmitter {
|
||||
async findAgentByName(name: string, executor?: QueryHandle): Promise<Agent | null> {
|
||||
// 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<AgentRatingSummary> {
|
||||
@@ -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<AgentHeartbeatEvent[]> {
|
||||
// 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<BlockedStateSnapshot | null> {
|
||||
// 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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
// 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<AgentConfigRevision[]> {
|
||||
// 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<AgentConfigRevision | null> {
|
||||
// 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 {
|
||||
|
||||
@@ -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<Agent | null> {
|
||||
export async function readAgent(
|
||||
handle: QueryHandle,
|
||||
agentId: string,
|
||||
projectId?: string,
|
||||
): Promise<Agent | null> {
|
||||
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<Agent[]> {
|
||||
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<Agent[]> {
|
||||
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<boolean> {
|
||||
export async function deleteAgent(
|
||||
handle: QueryHandle,
|
||||
agentId: string,
|
||||
projectId?: string,
|
||||
): Promise<boolean> {
|
||||
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<AgentHeartbeatEvent> {
|
||||
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<AgentHeartbeatEvent[]> {
|
||||
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<AgentTaskSession | null> {
|
||||
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<AgentTaskSession> {
|
||||
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<void> {
|
||||
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<AgentApiKey[]> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<AgentConfigRevision[]> {
|
||||
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<AgentConfigRevision | null> {
|
||||
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<AgentRating> {
|
||||
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<AgentRating[]> {
|
||||
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<boolean> {
|
||||
export async function deleteRating(
|
||||
handle: QueryHandle,
|
||||
ratingId: string,
|
||||
projectId?: string,
|
||||
): Promise<boolean> {
|
||||
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<BlockedStateSnapshot | null> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<Array<{ agentId: string; state: BlockedStateSnapshot }>> {
|
||||
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) => {
|
||||
|
||||
@@ -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<Record<string, unknown>>;
|
||||
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<CountByAgentRow & { columnName: string }>;
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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 $$;
|
||||
@@ -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 };
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
]);
|
||||
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user