fix(server): watcher device-pin authz + table-schema drift test runs in CI

packages/server/src/__tests__/integration/watchers/watchers-crud.test.ts

@@ -14,19 +14,25 @@ import {
   createTestUser,
 } from '../../setup/test-fixtures';
 import { TestApiClient } from '../../setup/test-mcp-client';
-import { cleanupTestDatabase } from '../../setup/test-db';
+import { cleanupTestDatabase, getTestDb } from '../../setup/test-db';
 
 describe('watcher CRUD', () => {
   let owner: TestApiClient;
   let intruder: TestApiClient;
   let entityId: number;
   let agentId: string;
+  let ownerOrgId: string;
+  let ownerUserId: string;
+  let otherOrgId: string;
+  let otherUserId: string;
 
   beforeAll(async () => {
     await cleanupTestDatabase();
     const org = await createTestOrganization({ name: 'Watcher Test Org' });
     const user = await createTestUser({ email: 'watcher-owner@test.com' });
     await addUserToOrganization(user.id, org.id, 'owner');
+    ownerOrgId = org.id;
+    ownerUserId = user.id;
     owner = await TestApiClient.for({
       organizationId: org.id,
       userId: user.id,
@@ -38,6 +44,8 @@ describe('watcher CRUD', () => {
     const otherOrg = await createTestOrganization({ name: 'Watcher Other Org' });
     const otherUser = await createTestUser({ email: 'watcher-other@test.com' });
     await addUserToOrganization(otherUser.id, otherOrg.id, 'owner');
+    otherOrgId = otherOrg.id;
+    otherUserId = otherUser.id;
     intruder = await TestApiClient.for({
       organizationId: otherOrg.id,
       userId: otherUser.id,
@@ -288,4 +296,116 @@ describe('watcher CRUD', () => {
       /admin|owner|access/i
     );
   });
+
+  // Issue #1060: a device pin (watchers.device_worker_id) runs the watcher's
+  // agent CLI on the device owner's machine, so create/update must verify the
+  // caller may target that device. The exhaustive role × ownership matrix is in
+  // src/__tests__/unit/watcher-device-access.test.ts; this proves the gate is
+  // wired into the handlers end-to-end (device_worker_id only reaches the
+  // handler via the raw `manage()` escape hatch — the typed input omits it).
+  describe('device_worker_id ownership gate', () => {
+    async function seedDevice(opts: {
+      userId: string;
+      organizationId: string | null;
+      worker: string;
+    }): Promise<string> {
+      const sql = getTestDb();
+      const rows = (await sql`
+        INSERT INTO device_workers (user_id, worker_id, platform, capabilities, label, organization_id)
+        VALUES (${opts.userId}, ${opts.worker}, 'macos', ${sql.json([])}, 'Seed Device', ${opts.organizationId})
+        RETURNING id
+      `) as unknown as Array<{ id: string }>;
+      return String(rows[0].id);
+    }
+
+    it('lets an org owner pin a foreign device attached to their org', async () => {
+      // A different user's device, but it lives in the owner's org → allowed
+      // for an owner/admin role.
+      const deviceId = await seedDevice({
+        userId: otherUserId,
+        organizationId: ownerOrgId,
+        worker: 'dev-in-org',
+      });
+      const created = (await owner.watchers.manage({
+        action: 'create',
+        entity_id: entityId,
+        slug: 'device-pin-allowed',
+        name: 'Device Pin Allowed',
+        prompt: 'x',
+        extraction_schema: { type: 'object', properties: {} },
+        agent_id: agentId,
+        device_worker_id: deviceId,
+      })) as { watcher_id: string };
+      expect(created.watcher_id).toBeDefined();
+
+      const got = (await owner.watchers.get(created.watcher_id)) as {
+        watcher?: { device_worker_id?: string | null };
+      };
+      expect(got.watcher?.device_worker_id).toBe(deviceId);
+      await owner.watchers.delete([created.watcher_id]);
+    });
+
+    it("rejects pinning to a device in another org (create)", async () => {
+      // Device owned by another user AND attached to another org — even an owner
+      // cannot pin it; this is the privilege-escalation case.
+      const foreignDeviceId = await seedDevice({
+        userId: otherUserId,
+        organizationId: otherOrgId,
+        worker: 'dev-foreign-org',
+      });
+      await expect(
+        owner.watchers.manage({
+          action: 'create',
+          entity_id: entityId,
+          slug: 'device-pin-foreign',
+          name: 'Device Pin Foreign',
+          prompt: 'x',
+          extraction_schema: { type: 'object', properties: {} },
+          agent_id: agentId,
+          device_worker_id: foreignDeviceId,
+        })
+      ).rejects.toThrow(/device you own|not found or not accessible/i);
+    });
+
+    it('rejects pinning to a nonexistent device (create)', async () => {
+      await expect(
+        owner.watchers.manage({
+          action: 'create',
+          entity_id: entityId,
+          slug: 'device-pin-missing',
+          name: 'Device Pin Missing',
+          prompt: 'x',
+          extraction_schema: { type: 'object', properties: {} },
+          agent_id: agentId,
+          device_worker_id: '00000000-0000-0000-0000-000000000000',
+        })
+      ).rejects.toThrow(/not found or not accessible/i);
+    });
+
+    it('rejects re-pinning an existing watcher to a foreign-org device (update)', async () => {
+      const created = (await owner.watchers.create({
+        entity_id: entityId,
+        slug: 'device-pin-update',
+        name: 'Device Pin Update',
+        prompt: 'x',
+        extraction_schema: { type: 'object', properties: {} },
+        agent_id: agentId,
+      })) as { watcher_id: string };
+
+      const foreignDeviceId = await seedDevice({
+        userId: otherUserId,
+        organizationId: otherOrgId,
+        worker: 'dev-foreign-update',
+      });
+      await expect(
+        owner.watchers.manage({
+          action: 'update',
+          watcher_id: created.watcher_id,
+          device_worker_id: foreignDeviceId,
+        })
+      ).rejects.toThrow(/device you own|not found or not accessible/i);
+
+      await owner.watchers.delete([created.watcher_id]);
+    });
+  });
 });

packages/server/src/__tests__/setup/global-setup.ts

@@ -10,15 +10,29 @@
  * migrations, fixtures, and assertions are identical either way.
  */
 
+import type { GlobalSetupContext } from 'vitest/node';
 import { closeDbSingleton } from '../../db/client';
 import { type EmbeddedBackend, startEmbeddedBackend } from './embedded-postgres-backend';
 import { closeTestDb, setupTestDatabase } from './test-db';
 
+// Make the resolved test DATABASE_URL available to forked test workers via
+// vitest's `inject()`. Env-var propagation from this setup process to forks
+// is timing-dependent (and was the reason the table-schema drift test could
+// silently self-skip in CI); `provide`/`inject` is the vitest-blessed,
+// transport-level channel that always reaches the workers.
+declare module 'vitest' {
+  interface ProvidedContext {
+    /** Connectable Postgres URL for the suite, or null when DB setup was skipped. */
+    databaseUrl: string | null;
+  }
+}
+
 let embedded: EmbeddedBackend | null = null;
 
-export async function setup(): Promise<void> {
+export async function setup({ provide }: GlobalSetupContext): Promise<void> {
   if (process.env.SKIP_TEST_DB_SETUP === '1') {
     console.log('\n⚠️  Skipping test database setup (SKIP_TEST_DB_SETUP=1).\n');
+    provide('databaseUrl', null);
     return;
   }
 
@@ -34,6 +48,9 @@ export async function setup(): Promise<void> {
     console.log(`✅ Embedded Postgres ready at ${embedded.url}`);
   }
 
+  // Authoritative signal for forks (see ProvidedContext augmentation above).
+  provide('databaseUrl', process.env.DATABASE_URL ?? null);
+
   // Deterministic 32-byte hex key for AES-256-GCM in tests. Same value the
   // gateway's secret-store test harness uses so behavior is aligned.
   if (!process.env.ENCRYPTION_KEY) {

packages/server/src/__tests__/unit/watcher-device-access.test.ts

@@ -0,0 +1,121 @@
+/**
+ * Unit coverage for the device-worker ownership gate on watcher pins
+ * (watchers.device_worker_id). A device pin runs the watcher's agent CLI on the
+ * device owner's machine, so a member-write actor must not be able to pin a
+ * watcher to another user's device. This pins the pure decision matrix
+ * (owner/admin/member/system × owned/foreign/missing device); the DB-backed
+ * wrapper + persistence is exercised in the integration suite.
+ */
+
+import { describe, expect, it } from 'bun:test';
+import {
+  type DeviceWorkerAccessCaller,
+  type DeviceWorkerOwnershipRow,
+  evaluateDeviceWorkerAccess,
+} from '../../tools/admin/watcher-device-access';
+
+const ORG = 'org-1';
+const OTHER_ORG = 'org-2';
+
+const owner: DeviceWorkerAccessCaller = {
+  memberRole: 'owner',
+  userId: 'u-owner',
+  organizationId: ORG,
+  isAuthenticated: true,
+};
+const admin: DeviceWorkerAccessCaller = {
+  memberRole: 'admin',
+  userId: 'u-admin',
+  organizationId: ORG,
+  isAuthenticated: true,
+};
+const member: DeviceWorkerAccessCaller = {
+  memberRole: 'member',
+  userId: 'u-member',
+  organizationId: ORG,
+  isAuthenticated: true,
+};
+// apply / automation / default-provisioning: authenticated, no user/role.
+const system: DeviceWorkerAccessCaller = {
+  memberRole: null,
+  userId: null,
+  organizationId: ORG,
+  isAuthenticated: true,
+};
+
+function device(
+  overrides: Partial<DeviceWorkerOwnershipRow> = {}
+): DeviceWorkerOwnershipRow {
+  return { id: 'dev-1', user_id: 'u-member', organization_id: ORG, ...overrides };
+}
+
+describe('evaluateDeviceWorkerAccess — own device', () => {
+  it('allows any caller to pin a device they own', () => {
+    expect(evaluateDeviceWorkerAccess(device({ user_id: 'u-member' }), member)).toBeNull();
+    expect(evaluateDeviceWorkerAccess(device({ user_id: 'u-owner' }), owner)).toBeNull();
+    // Even when the owned device is attached to a different org.
+    expect(
+      evaluateDeviceWorkerAccess(
+        device({ user_id: 'u-member', organization_id: OTHER_ORG }),
+        member
+      )
+    ).toBeNull();
+  });
+});
+
+describe('evaluateDeviceWorkerAccess — foreign device', () => {
+  it("blocks a member from pinning another user's device", () => {
+    const reason = evaluateDeviceWorkerAccess(
+      device({ user_id: 'someone-else', organization_id: ORG }),
+      member
+    );
+    expect(reason).toMatch(/device you own/i);
+  });
+
+  it('allows an org owner to pin a foreign device attached to their org', () => {
+    expect(
+      evaluateDeviceWorkerAccess(device({ user_id: 'someone-else', organization_id: ORG }), owner)
+    ).toBeNull();
+  });
+
+  it('allows an org admin to pin a foreign device attached to their org', () => {
+    expect(
+      evaluateDeviceWorkerAccess(device({ user_id: 'someone-else', organization_id: ORG }), admin)
+    ).toBeNull();
+  });
+
+  it('blocks an owner from pinning a device in another org', () => {
+    const reason = evaluateDeviceWorkerAccess(
+      device({ user_id: 'someone-else', organization_id: OTHER_ORG }),
+      owner
+    );
+    expect(reason).toMatch(/device you own/i);
+  });
+
+  it('blocks an admin from pinning a device with no org attachment they do not own', () => {
+    const reason = evaluateDeviceWorkerAccess(
+      device({ user_id: 'someone-else', organization_id: null }),
+      admin
+    );
+    expect(reason).toMatch(/device you own/i);
+  });
+});
+
+describe('evaluateDeviceWorkerAccess — missing device', () => {
+  it('rejects a non-member caller when the device row is absent', () => {
+    expect(evaluateDeviceWorkerAccess(null, member)).toMatch(/not found or not accessible/i);
+    expect(evaluateDeviceWorkerAccess(null, owner)).toMatch(/not found or not accessible/i);
+  });
+});
+
+describe('evaluateDeviceWorkerAccess — system/internal caller', () => {
+  it('allows system callers regardless of ownership', () => {
+    expect(
+      evaluateDeviceWorkerAccess(device({ user_id: 'whoever', organization_id: OTHER_ORG }), system)
+    ).toBeNull();
+  });
+
+  it('allows system callers even when the device is missing', () => {
+    expect(evaluateDeviceWorkerAccess(null, system)).toBeNull();
+  });
+});

packages/server/src/tools/admin/manage_watchers.ts

@@ -61,6 +61,7 @@ import { validateTemplate } from '../../watchers/renderer';
 import { validateClassifierSourcePaths, validateExtractionSchema } from '../../watchers/validator';
 import type { ToolContext } from '../registry';
 import { routeAction } from './action-router';
+import { assertDeviceWorkerAccess } from './watcher-device-access';
 import {
   assertValidExecutionConfig,
   WatcherExecutionConfigSchema,
@@ -953,6 +954,10 @@ async function handleCreate(
     throw new ToolUserError('extraction_schema is required for create action');
   }
   assertValidExecutionConfig(args.execution_config, ctx);
+  // A device pin runs the watcher's agent CLI on the device owner's machine —
+  // validate the caller may target this device (own it, or org owner/admin
+  // over a device attached to the org) before storing it.
+  await assertDeviceWorkerAccess(sql, args.device_worker_id, ctx);
 
   // entity_id is optional: omit it for an org-scoped/global watcher.
   const entityId = args.entity_id;
@@ -1286,6 +1291,10 @@ async function handleUpdate(
     throw new Error('watcher_id is required for update action');
   }
   assertValidExecutionConfig(args.execution_config, ctx);
+  // Re-pinning to a device targets that device owner's machine — validate the
+  // caller may pin it (own it, or org owner/admin over an org-attached device).
+  // undefined = unchanged and null = clear the pin both pass without a lookup.
+  await assertDeviceWorkerAccess(sql, args.device_worker_id, ctx);
 
   await requireExists(sql, 'watchers', args.watcher_id, 'Watcher');