fix(watchers,feeds): guard id-less source queries and cross-org entity_ids

packages/server/src/__tests__/integration/feeds-cross-org-entity.test.ts

@@ -0,0 +1,125 @@
+/**
+ * manage_feeds create_feed / update_feed — cross-org entity_ids validation.
+ *
+ * Regression (found in prod lobu-crm): feeds were created with `entity_ids`
+ * pointing at an entity owned by a DIFFERENT org. The create/update path did
+ * not validate entity ownership, so synced events linked to a non-existent
+ * in-org entity — a silent data-correctness bug. The fix rejects any entity_id
+ * that does not belong to the requesting org.
+ */
+
+import { beforeAll, describe, expect, it } from "vitest";
+import { cleanupTestDatabase, getTestDb } from "../setup/test-db";
+import {
+	createTestConnection,
+	createTestEntity,
+	createTestOrganization,
+	createTestUser,
+} from "../setup/test-fixtures";
+import { TestApiClient } from "../setup/test-mcp-client";
+
+describe("manage_feeds cross-org entity_ids", () => {
+	let owner: TestApiClient;
+	let ownerOrgId: string;
+	let connectionId: number;
+	let inOrgEntityId: number;
+	let foreignEntityId: number;
+
+	beforeAll(async () => {
+		await cleanupTestDatabase();
+
+		const org = await createTestOrganization({ name: "Feed Owner Org" });
+		ownerOrgId = org.id;
+		const user = await createTestUser({ email: "feed-owner@test.com" });
+		owner = await TestApiClient.for({
+			organizationId: org.id,
+			userId: user.id,
+			memberRole: "owner",
+		});
+
+		const conn = await createTestConnection({
+			organization_id: ownerOrgId,
+			connector_key: "github",
+			created_by: user.id,
+			createDefaultFeed: false,
+		});
+		connectionId = Number(conn.id);
+
+		const inOrg = await createTestEntity({
+			name: "In-Org Entity",
+			entity_type: "company",
+			organization_id: ownerOrgId,
+			created_by: user.id,
+		});
+		inOrgEntityId = Number(inOrg.id);
+
+		// A separate org owns the "foreign" entity.
+		const foreignOrg = await createTestOrganization({ name: "Foreign Org" });
+		const foreignEntity = await createTestEntity({
+			name: "Foreign Entity",
+			entity_type: "company",
+			organization_id: foreignOrg.id,
+		});
+		foreignEntityId = Number(foreignEntity.id);
+	});
+
+	it("rejects create_feed when an entity_id belongs to another org", async () => {
+		const result = (await owner.feeds.create({
+			connection_id: connectionId,
+			feed_key: "default",
+			entity_ids: [foreignEntityId],
+		})) as { error?: string; feed?: unknown };
+
+		expect(result.error).toBeTruthy();
+		expect(result.error).toContain(String(foreignEntityId));
+		expect(result.feed).toBeUndefined();
+
+		// No feed row leaked into the DB.
+		const sql = getTestDb();
+		const rows = await sql<{ id: number }[]>`
+      SELECT id FROM feeds
+      WHERE organization_id = ${ownerOrgId} AND ${foreignEntityId} = ANY(entity_ids)
+    `;
+		expect(rows.length).toBe(0);
+	});
+
+	it("accepts create_feed with an in-org entity_id", async () => {
+		const result = (await owner.feeds.create({
+			connection_id: connectionId,
+			feed_key: "default",
+			entity_ids: [inOrgEntityId],
+		})) as { error?: string; feed?: { id: number } };
+
+		expect(result.error).toBeUndefined();
+		expect(result.feed?.id).toBeDefined();
+	});
+
+	it("rejects update_feed that repoints to another org's entity", async () => {
+		// Create a clean feed with no entity_ids first.
+		const created = (await owner.feeds.create({
+			connection_id: connectionId,
+			feed_key: "default",
+			display_name: "update-target",
+		})) as { feed?: { id: number } };
+		const feedId = Number(created.feed?.id);
+		expect(feedId).toBeGreaterThan(0);
+
+		const result = (await owner.feeds.update({
+			feed_id: feedId,
+			entity_ids: [foreignEntityId],
+		})) as { error?: string };
+
+		expect(result.error).toBeTruthy();
+		expect(result.error).toContain(String(foreignEntityId));
+
+		// The feed's entity_ids were NOT changed.
+		const sql = getTestDb();
+		const [row] = await sql<{ entity_ids: number[] | null }[]>`
+      SELECT entity_ids FROM feeds WHERE id = ${feedId}
+    `;
+		const ids = Array.isArray(row?.entity_ids)
+			? row.entity_ids.map(Number)
+			: [];
+		expect(ids).not.toContain(foreignEntityId);
+	});
+});

packages/server/src/__tests__/integration/watchers/source-id-and-cross-org.test.ts

@@ -0,0 +1,194 @@
+/**
+ * manage_watchers correctness guards:
+ *
+ *  BUG A — a source query that omits `id` is rejected at create/create_version
+ *  time. Watcher-mode aggregation keys rows by `id` and the signed window_token
+ *  only carries those ids, so an id-less source produces content_linked: 0 at
+ *  complete_window and SILENTLY skips the reaction. We reject it on save.
+ *
+ *  BUG B — create_from_version rejects entity_ids that belong to another org.
+ *  A watcher attached to a foreign entity links its content to a non-existent
+ *  in-org entity.
+ */
+
+import { beforeAll, describe, expect, it } from "vitest";
+import { cleanupTestDatabase, getTestDb } from "../../setup/test-db";
+import {
+	addUserToOrganization,
+	createTestAgent,
+	createTestEntity,
+	createTestOrganization,
+	createTestUser,
+} from "../../setup/test-fixtures";
+import { TestApiClient } from "../../setup/test-mcp-client";
+
+describe("manage_watchers source-id + cross-org guards", () => {
+	let owner: TestApiClient;
+	let ownerOrgId: string;
+	let agentId: string;
+	let inOrgEntityId: number;
+	let foreignEntityId: number;
+
+	const schema = {
+		type: "object",
+		properties: { items: { type: "array", items: { type: "string" } } },
+	};
+
+	beforeAll(async () => {
+		await cleanupTestDatabase();
+		const org = await createTestOrganization({ name: "Watcher Guard Org" });
+		ownerOrgId = org.id;
+		const user = await createTestUser({ email: "watcher-guard@test.com" });
+		await addUserToOrganization(user.id, org.id, "owner");
+		owner = await TestApiClient.for({
+			organizationId: org.id,
+			userId: user.id,
+			memberRole: "owner",
+		});
+		const agent = await createTestAgent({
+			organizationId: org.id,
+			ownerUserId: user.id,
+		});
+		agentId = agent.agentId;
+
+		const inOrg = await createTestEntity({
+			name: "In-Org Watcher Target",
+			entity_type: "company",
+			organization_id: ownerOrgId,
+			created_by: user.id,
+		});
+		inOrgEntityId = Number(inOrg.id);
+
+		const foreignOrg = await createTestOrganization({
+			name: "Watcher Foreign Org",
+		});
+		const foreignEntity = await createTestEntity({
+			name: "Foreign Watcher Target",
+			entity_type: "company",
+			organization_id: foreignOrg.id,
+		});
+		foreignEntityId = Number(foreignEntity.id);
+	});
+
+	// ---- BUG A ----
+
+	it("rejects create when a source query omits the id column", async () => {
+		await expect(
+			owner.watchers.create({
+				entity_id: inOrgEntityId,
+				slug: "no-id-source",
+				name: "No Id Source",
+				prompt: "Track stuff.",
+				extraction_schema: schema,
+				agent_id: agentId,
+				sources: [
+					{
+						name: "content",
+						query: "SELECT origin_id, payload_text FROM events",
+					},
+				],
+			}),
+		).rejects.toThrow(/id/i);
+	});
+
+	it("accepts create when the source query projects id", async () => {
+		const created = (await owner.watchers.create({
+			entity_id: inOrgEntityId,
+			slug: "with-id-source",
+			name: "With Id Source",
+			prompt: "Track stuff.",
+			extraction_schema: schema,
+			agent_id: agentId,
+			sources: [
+				{
+					name: "content",
+					query: "SELECT id, origin_id, payload_text FROM events",
+				},
+			],
+		})) as { watcher_id?: string };
+		expect(created.watcher_id).toBeDefined();
+	});
+
+	it("rejects create_version when a source query omits id", async () => {
+		const created = (await owner.watchers.create({
+			entity_id: inOrgEntityId,
+			slug: "version-id-guard",
+			name: "Version Id Guard",
+			prompt: "Track stuff.",
+			extraction_schema: schema,
+			agent_id: agentId,
+			sources: [{ name: "content", query: "SELECT id FROM events" }],
+		})) as { watcher_id: string };
+
+		await expect(
+			owner.watchers.createVersion({
+				watcher_id: created.watcher_id,
+				prompt: "Track stuff v2.",
+				extraction_schema: schema,
+				change_notes: "omit id",
+				sources: [
+					{ name: "content", query: "SELECT payload_text FROM events" },
+				],
+			} as never),
+		).rejects.toThrow(/id/i);
+	});
+
+	// ---- BUG B ----
+
+	it("create_from_version rejects a cross-org entity_id", async () => {
+		const base = (await owner.watchers.create({
+			entity_id: inOrgEntityId,
+			slug: "cfv-base",
+			name: "CFV Base",
+			prompt: "Track stuff.",
+			extraction_schema: schema,
+			agent_id: agentId,
+			sources: [{ name: "content", query: "SELECT id FROM events" }],
+		})) as { watcher_id: string };
+
+		const sql = getTestDb();
+		const [row] = await sql<{ current_version_id: number }[]>`
+      SELECT current_version_id FROM watchers WHERE id = ${base.watcher_id}
+    `;
+		const versionId = Number(row?.current_version_id);
+		expect(versionId).toBeGreaterThan(0);
+
+		await expect(
+			owner.watchers.createFromVersion({
+				version_id: versionId,
+				entity_ids: [foreignEntityId],
+			}),
+		).rejects.toThrow(new RegExp(String(foreignEntityId)));
+
+		// No watcher leaked pointing at the foreign entity.
+		const leaked = await sql<{ id: number }[]>`
+      SELECT id FROM watchers
+      WHERE organization_id = ${ownerOrgId} AND ${foreignEntityId} = ANY(entity_ids)
+    `;
+		expect(leaked.length).toBe(0);
+	});
+
+	it("create_from_version accepts an in-org entity_id", async () => {
+		const base = (await owner.watchers.create({
+			entity_id: inOrgEntityId,
+			slug: "cfv-base-ok",
+			name: "CFV Base OK",
+			prompt: "Track stuff.",
+			extraction_schema: schema,
+			agent_id: agentId,
+			sources: [{ name: "content", query: "SELECT id FROM events" }],
+		})) as { watcher_id: string };
+
+		const sql = getTestDb();
+		const [row] = await sql<{ current_version_id: number }[]>`
+      SELECT current_version_id FROM watchers WHERE id = ${base.watcher_id}
+    `;
+		const versionId = Number(row?.current_version_id);
+
+		const result = (await owner.watchers.createFromVersion({
+			version_id: versionId,
+			entity_ids: [inOrgEntityId],
+		})) as { created: Array<{ watcher_id: string }> };
+		expect(result.created.length).toBe(1);
+	});
+});