Tenancy & RLS

            ┌─────────────────────────────────────────────┐
 request →  │  CapabilityContext { orgId, workspaceId }    │   (resolved at the
            └───────────────────────┬─────────────────────┘    transport edge today)
                                    │
                    kernel.invoke() │  runInTenantScope(scope, () => handler(...))
                                    ▼
                 ┌──────────  @oxagen/tenancy  ──────────┐
                 │  AsyncLocalStorage<TenantScope>        │   ← the ONE identity seam
                 │  getScope() / requireScope()           │
                 └───┬───────────────┬───────────────┬────┘
                     ▼               ▼               ▼
        withTenantDb(fn)      scopedSession()   chSelect / chInsert
        (SET LOCAL GUC          (binds $orgId/    (injects org_id /
         in one tx; RLS          $workspaceId;     workspace_id predicate
         enforces at DB)         guards Cypher)    + stamps on write)
                     │               │               │
              Postgres RLS      Neo4j (prop)     ClickHouse (prop
              = HARD wall       = seam wall      + optional ROW POLICY)

// packages/tenancy/src/scope.ts
export interface TenantScope {
  readonly orgId: string;       // uuid, non-empty
  readonly workspaceId: string; // uuid, non-empty
}

const als = new AsyncLocalStorage<TenantScope>();

/** Validate + enter scope. Fail-closed: empty ids throw (fixes MCP orgId:""). */
export function runInTenantScope<T>(scope: TenantScope, fn: () => T): T {
  assertUuid(scope.orgId, "orgId");
  assertUuid(scope.workspaceId, "workspaceId");
  return als.run(scope, fn);
}

/** Current scope or null (for code that legitimately runs unscoped, e.g. cron bootstrap). */
export function getScope(): TenantScope | null {
  return als.getStore() ?? null;
}

/** Current scope or throw — used by every data accessor. */
export function requireScope(): TenantScope {
  const s = als.getStore();
  if (!s) throw new TenantScopeError("No active tenant scope — call out of bounds");
  return s;
}

// packages/database/src/tenant.ts
import { sql } from "drizzle-orm";
import { db, type Database } from "./client";
import { requireScope } from "@oxagen/tenancy";

type Tx = Parameters<Parameters<Database["transaction"]>[0]>[0];

/** Run DB work in a tenant-scoped transaction. RLS enforces isolation. */
export async function withTenantDb<T>(fn: (tx: Tx) => Promise<T>): Promise<T> {
  const { orgId, workspaceId } = requireScope();
  return db().transaction(async (tx) => {
    // set_config(key, value, is_local=true) — scoped to this tx only.
    await tx.execute(
      sql`select set_config('app.current_org_id', ${orgId}, true),
                 set_config('app.current_workspace_id', ${workspaceId}, true)`,
    );
    return fn(tx);
  });
}

ALTER TABLE agent.agents ENABLE ROW LEVEL SECURITY;
ALTER TABLE agent.agents FORCE  ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON agent.agents
  USING (
    current_setting('app.rls_bypass', true) = 'on'
    OR (
      org_id       = nullif(current_setting('app.current_org_id',       true), '')::uuid
      AND workspace_id = nullif(current_setting('app.current_workspace_id', true), '')::uuid
    )
  )
  WITH CHECK (
    current_setting('app.rls_bypass', true) = 'on'
    OR (
      org_id       = nullif(current_setting('app.current_org_id',       true), '')::uuid
      AND workspace_id = nullif(current_setting('app.current_workspace_id', true), '')::uuid
    )
  );

// packages/ontology/src/tenant.ts
import { session } from "./client";
import { requireScope } from "@oxagen/tenancy";

const SCOPE_GUARD = /node\.orgId|m\.orgId|\$orgId/; // must reference scope

/** A session whose run() auto-injects $orgId/$workspaceId and guards the query. */
export function scopedSession() {
  const { orgId, workspaceId } = requireScope();
  const s = session();
  return {
    run(cypher: string, params: Record<string, unknown> = {}) {
      if (!SCOPE_GUARD.test(cypher)) {
        throw new TenantScopeError(`Cypher over scoped session must filter by $orgId: ${cypher.slice(0,80)}`);
      }
      return s.run(cypher, { ...params, orgId, workspaceId });
    },
    close: () => s.close(),
  };
}

// packages/telemetry/src/tenant.ts
export async function chInsert(table: string, rows: object[]) {
  const { orgId, workspaceId } = requireScope();
  const stamped = rows.map(r => ({ org_id: orgId, workspace_id: workspaceId, ...r }));
  return clickhouse().insert({ table, values: stamped, format: "JSONEachRow" });
}

export async function chSelect<T>(q: { query: string; params?: Record<string, unknown> }) {
  const { orgId, workspaceId } = requireScope();
  // Caller writes "... WHERE org_id = {orgId:UUID} AND ..."; we bind the values.
  return clickhouse().query({ query: q.query, query_params: { ...q.params, orgId, workspaceId }, format: "JSON" }) as Promise<T>;
}

{
  "name": "@oxagen/tenancy",
  "version": "0.1.0",
  "private": true,
  "type": "module",
  "main": "./src/index.ts",
  "types": "./src/index.ts",
  "exports": {
    ".": "./src/index.ts",
    "./testing": "./src/testing.ts"
  },
  "scripts": {
    "typecheck": "tsc --noEmit",
    "lint": "eslint src --max-warnings 0",
    "test:unit": "vitest run"
  },
  "devDependencies": {
    "typescript": "6.0.3",
    "vitest": "2.1.9"
  }
}

{
  "extends": "../../tsconfig.base.json",
  "compilerOptions": { "outDir": "dist", "rootDir": "src" },
  "include": ["src/**/*.ts"]
}

import { defineConfig } from "vitest/config";

export default defineConfig({
  test: { environment: "node", include: ["src/**/*.test.ts"], globals: false },
});

import { describe, expect, it } from "vitest";
import {
  runInTenantScope,
  getScope,
  requireScope,
  TenantScopeError,
} from "./index";

const ORG = "00000000-0000-0000-0000-0000000000a1";
const WS = "00000000-0000-0000-0000-0000000000b2";

describe("tenant scope", () => {
  it("exposes the active scope inside runInTenantScope", () => {
    const seen = runInTenantScope({ orgId: ORG, workspaceId: WS }, () => getScope());
    expect(seen).toEqual({ orgId: ORG, workspaceId: WS });
  });

  it("returns null outside any scope", () => {
    expect(getScope()).toBeNull();
  });

  it("requireScope throws outside a scope (fail-closed)", () => {
    expect(() => requireScope()).toThrowError(TenantScopeError);
  });

  it("rejects an empty orgId (fixes MCP orgId:'' fail-open)", () => {
    expect(() => runInTenantScope({ orgId: "", workspaceId: WS }, () => 1)).toThrowError(
      /orgId/,
    );
  });

  it("rejects a non-uuid workspaceId", () => {
    expect(() =>
      runInTenantScope({ orgId: ORG, workspaceId: "not-a-uuid" }, () => 1),
    ).toThrowError(/workspaceId/);
  });

  it("isolates nested scopes and restores the outer one", () => {
    const other = "00000000-0000-0000-0000-0000000000c3";
    runInTenantScope({ orgId: ORG, workspaceId: WS }, () => {
      runInTenantScope({ orgId: other, workspaceId: WS }, () => {
        expect(requireScope().orgId).toBe(other);
      });
      expect(requireScope().orgId).toBe(ORG);
    });
  });
});

/** Raised when tenant-scoped work runs with no active (or invalid) scope. */
export class TenantScopeError extends Error {
  readonly code = "no_tenant_scope" as const;
  constructor(message: string) {
    super(message);
    this.name = "TenantScopeError";
  }
}

import { AsyncLocalStorage } from "node:async_hooks";
import { TenantScopeError } from "./errors";

export interface TenantScope {
  readonly orgId: string;
  readonly workspaceId: string;
}

const als = new AsyncLocalStorage<TenantScope>();

const UUID_RE =
  /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

function assertUuid(value: string, field: string): void {
  if (!UUID_RE.test(value)) {
    throw new TenantScopeError(`Invalid ${field}: expected a uuid, got ${JSON.stringify(value)}`);
  }
}

/** Validate + enter scope. Fail-closed: empty/invalid ids throw. */
export function runInTenantScope<T>(scope: TenantScope, fn: () => T): T {
  assertUuid(scope.orgId, "orgId");
  assertUuid(scope.workspaceId, "workspaceId");
  return als.run({ orgId: scope.orgId, workspaceId: scope.workspaceId }, fn);
}

/** The active scope, or null when none is set. */
export function getScope(): TenantScope | null {
  return als.getStore() ?? null;
}

/** The active scope, or throw — used by every data accessor. */
export function requireScope(): TenantScope {
  const s = als.getStore();
  if (!s) {
    throw new TenantScopeError("No active tenant scope — data access out of bounds");
  }
  return s;
}

export { runInTenantScope, getScope, requireScope } from "./scope";
export type { TenantScope } from "./scope";
export { TenantScopeError } from "./errors";

pnpm --filter @oxagen/tenancy typecheck && pnpm --filter @oxagen/tenancy lint
git add packages/tenancy
git commit -m "feat(tenancy): add AsyncLocalStorage tenant scope seam (OXA-1515)"

import { describe, expect, it } from "vitest";
import { requireScope } from "./index";
import { TEST_ORG, TEST_WS, TEST_SCOPE, withTestScope } from "./testing";

describe("test helpers", () => {
  it("runs a callback inside a deterministic test scope", () => {
    const seen = withTestScope(() => requireScope());
    expect(seen).toEqual({ orgId: TEST_ORG, workspaceId: TEST_WS });
    expect(TEST_SCOPE).toEqual({ orgId: TEST_ORG, workspaceId: TEST_WS });
  });

  it("accepts an override scope", () => {
    const org = "00000000-0000-0000-0000-0000000000d4";
    const seen = withTestScope(() => requireScope(), { orgId: org, workspaceId: TEST_WS });
    expect(seen.orgId).toBe(org);
  });
});

import { runInTenantScope, type TenantScope } from "./scope";

export const TEST_ORG = "00000000-0000-0000-0000-00000000a000";
export const TEST_WS = "00000000-0000-0000-0000-00000000b000";
export const TEST_SCOPE: TenantScope = { orgId: TEST_ORG, workspaceId: TEST_WS };

/** Run a unit-test body inside a tenant scope without ceremony. */
export function withTestScope<T>(fn: () => T, scope: TenantScope = TEST_SCOPE): T {
  return runInTenantScope(scope, fn);
}

git add packages/tenancy/src/testing.ts packages/tenancy/src/testing.test.ts
git commit -m "feat(tenancy): add unit-test scope helpers"

"@oxagen/tenancy": "workspace:*"

import { afterEach, describe, expect, it } from "vitest";
import { getScope } from "@oxagen/tenancy";
import { invoke, registerHandler, clearHandlersForTests } from "./kernel";
import { registerCapability, clearRegistryForTests } from "./registry";
import { z } from "zod";
import type { CapabilityContext } from "./types";

const ORG = "00000000-0000-0000-0000-00000000a111";
const WS = "00000000-0000-0000-0000-00000000b222";

function ctx(over: Partial<CapabilityContext> = {}): CapabilityContext {
  return {
    orgId: ORG, workspaceId: WS, userId: "u1", apiKeyId: null,
    requestId: "r1", surface: "api", messageId: null, ...over,
  };
}

afterEach(() => {
  clearHandlersForTests();
  clearRegistryForTests();
});

describe("kernel tenant scope", () => {
  it("runs the handler inside the request's tenant scope", async () => {
    registerCapability({
      name: "test.echo", input: z.object({}), output: z.object({ org: z.string() }),
      surfaces: ["api"],
    });
    registerHandler("test.echo", async () => async () => {
      const s = getScope();
      return { org: s?.orgId ?? "NONE" };
    });
    const out = await invoke("test.echo", {}, ctx());
    expect(out).toEqual({ org: ORG });
  });

  it("fails closed when orgId is empty (MCP session-token path)", async () => {
    registerCapability({
      name: "test.echo2", input: z.object({}), output: z.object({}), surfaces: ["mcp"],
    });
    registerHandler("test.echo2", async () => async () => ({}));
    await expect(invoke("test.echo2", {}, ctx({ orgId: "" }))).rejects.toThrow(/tenant scope|orgId/);
  });
});

import { runInTenantScope } from "@oxagen/tenancy";

    const handler = await resolveHandler(name);
    output = await runInTenantScope(
      { orgId: ctx.orgId, workspaceId: ctx.workspaceId },
      () => handler(inputResult.data, ctx),
    );

pnpm --filter @oxagen/oxagen test:unit && pnpm --filter @oxagen/oxagen typecheck
git add packages/oxagen
git commit -m "feat(kernel): enter tenant scope per invocation; fail closed on empty org (OXA-1515)"

it("defaults TENANT_RLS_ENFORCEMENT_ENABLED to false (seeding-safe)", () => {
  __resetEnvCacheForTests();
  const env = requireEnv(["TENANT_RLS_ENFORCEMENT_ENABLED"], {
    ...process.env,
    TENANT_RLS_ENFORCEMENT_ENABLED: undefined,
  });
  expect(env.TENANT_RLS_ENFORCEMENT_ENABLED).toBe(false);
});

it("coerces TENANT_RLS_ENFORCEMENT_ENABLED='true' to true", () => {
  __resetEnvCacheForTests();
  const env = requireEnv(["TENANT_RLS_ENFORCEMENT_ENABLED"], {
    ...process.env,
    TENANT_RLS_ENFORCEMENT_ENABLED: "true",
  });
  expect(env.TENANT_RLS_ENFORCEMENT_ENABLED).toBe(true);
});

  // OXA-1515: Row-Level Security enforcement gate. Default OFF during the
  // seeding window: withTenantDb always sets the scope GUCs, but additionally
  // sets app.rls_bypass='on' while this is false so the bypass-aware policies
  // do not yet filter. During seeding, isolation is still enforced by the
  // manual eq(orgId) predicates kept in every query. Flip to true per env
  // once db.query.unscoped telemetry reads zero. Reversible via env (no
  // migration needed).
  TENANT_RLS_ENFORCEMENT_ENABLED: z
    .union([z.literal("true"), z.literal("false")])
    .optional()
    .transform((v) => v === "true"),

  TENANT_RLS_ENFORCEMENT_ENABLED: {
    group: "Security",
    description:
      "When true, Postgres RLS policies filter by org/workspace. Default off " +
      "during seeding; withTenantDb sets app.rls_bypass while off. Flip to true " +
      "per env after db.query.unscoped telemetry reads zero.",
    secret: false,
    clientExposed: false,
    services: ["api", "app", "mcp"],
    requiredIn: [],
    valueOrigin: "manual",
    placeholder: "false",
  },

git add packages/config/src/env.ts packages/config/src/registry.ts packages/config/src/env.test.ts
git commit -m "feat(config): add TENANT_RLS_ENFORCEMENT_ENABLED flag, default off (OXA-1515)"

"@oxagen/tenancy": "workspace:*"

    "./tenant": "./src/tenant.ts",
    "./tenant.mock": "./src/tenant.mock.ts"

import { describe, expect, it, vi, beforeEach } from "vitest";

const mocks = vi.hoisted(() => ({
  execute: vi.fn(async () => undefined),
  transaction: vi.fn(),
  rlsEnforced: vi.fn(() => false),
}));

mocks.transaction.mockImplementation(
  async (cb: (tx: unknown) => Promise<unknown>) => cb({ execute: mocks.execute }),
);

vi.mock("./client", () => ({ db: () => ({ transaction: mocks.transaction }) }));
// rlsEnforced reads env; stub it so the test controls the branch.
vi.mock("./tenant-flag", () => ({ rlsEnforced: mocks.rlsEnforced }));

import { runInTenantScope } from "@oxagen/tenancy";
import { withTenantDb } from "./tenant";

const ORG = "00000000-0000-0000-0000-00000000a111";
const WS = "00000000-0000-0000-0000-00000000b222";

function sqlText(call: unknown): string {
  // drizzle sql`` template → capture via the mock arg's queryChunks/strings.
  return JSON.stringify(call);
}

beforeEach(() => {
  mocks.execute.mockClear();
  mocks.rlsEnforced.mockReturnValue(false);
});

describe("withTenantDb", () => {
  it("requires an active scope (fail-closed)", async () => {
    await expect(withTenantDb(async () => 1)).rejects.toThrow(/tenant scope/);
  });

  it("sets org + workspace GUCs and runs the callback on the tx", async () => {
    const result = await runInTenantScope({ orgId: ORG, workspaceId: WS }, () =>
      withTenantDb(async (tx) => {
        expect(tx).toBeDefined();
        return "ok";
      }),
    );
    expect(result).toBe("ok");
    expect(mocks.execute).toHaveBeenCalledTimes(1); // one set_config statement
    const arg = sqlText(mocks.execute.mock.calls[0][0]);
    expect(arg).toContain("app.current_org_id");
    expect(arg).toContain("app.current_workspace_id");
    expect(arg).toContain("app.rls_bypass"); // flag off → bypass set
  });

  it("omits the bypass GUC when enforcement is enabled", async () => {
    mocks.rlsEnforced.mockReturnValue(true);
    await runInTenantScope({ orgId: ORG, workspaceId: WS }, () =>
      withTenantDb(async () => undefined),
    );
    const arg = sqlText(mocks.execute.mock.calls[0][0]);
    expect(arg).not.toContain("app.rls_bypass");
  });
});

import { requireEnv } from "@oxagen/config/env";

/** True when RLS policies should actively filter (flag on → no bypass GUC). */
export function rlsEnforced(): boolean {
  return requireEnv(["TENANT_RLS_ENFORCEMENT_ENABLED"]).TENANT_RLS_ENFORCEMENT_ENABLED;
}

import { sql } from "drizzle-orm";
import { requireScope } from "@oxagen/tenancy";
import { db, type Database } from "./client";
import { rlsEnforced } from "./tenant-flag";

/** The transaction handle Drizzle hands to a `.transaction(cb)` callback. */
export type Tx = Parameters<Parameters<Database["transaction"]>[0]>[0];

/**
 * Run DB work in a tenant-scoped transaction. Sets the per-transaction GUCs
 * that the RLS policies read. When enforcement is OFF, also sets
 * app.rls_bypass='on' so policies don't yet filter (seeding window).
 *
 * Keep the body focused — do not wrap long LLM/tool calls in one withTenantDb;
 * the transaction is held for the callback's lifetime.
 */
export async function withTenantDb<T>(fn: (tx: Tx) => Promise<T>): Promise<T> {
  const { orgId, workspaceId } = requireScope();
  const bypass = rlsEnforced() ? "off" : "on";
  return db().transaction(async (tx) => {
    await tx.execute(sql`
      select
        set_config('app.current_org_id', ${orgId}, true),
        set_config('app.current_workspace_id', ${workspaceId}, true),
        set_config('app.rls_bypass', ${bypass}, true)
    `);
    return fn(tx);
  });
}

import type { Tx } from "./tenant";

/**
 * Test double for withTenantDb: invokes the callback with the provided fake
 * tx, no transaction, no GUC. Suites pass their existing fake db object.
 */
export function makeWithTenantDbMock(fakeTx: unknown) {
  return async <T>(fn: (tx: Tx) => Promise<T>): Promise<T> => fn(fakeTx as Tx);
}

export { withTenantDb, type Tx } from "./tenant";

pnpm --filter @oxagen/database typecheck
git add packages/database/package.json packages/database/src/tenant.ts packages/database/src/tenant-flag.ts packages/database/src/tenant.mock.ts packages/database/src/tenant.test.ts packages/database/src/index.ts
git commit -m "feat(database): add withTenantDb tenant-scoped tx wrapper + mock (OXA-1515)"

import { describe, expect, it, vi } from "vitest";
import { runInTenantScope } from "@oxagen/tenancy";
import { recordIfUnscoped, __unscopedCountForTests } from "./unscoped-meter";

const ORG = "00000000-0000-0000-0000-00000000a111";
const WS = "00000000-0000-0000-0000-00000000b222";

describe("unscoped meter", () => {
  it("counts a call with no active scope", () => {
    const before = __unscopedCountForTests();
    recordIfUnscoped("db.read");
    expect(__unscopedCountForTests()).toBe(before + 1);
  });

  it("does not count a call inside a scope", () => {
    const before = __unscopedCountForTests();
    runInTenantScope({ orgId: ORG, workspaceId: WS }, () => recordIfUnscoped("db.read"));
    expect(__unscopedCountForTests()).toBe(before);
  });
});

import { getScope } from "@oxagen/tenancy";

let unscopedCount = 0;

/** Log + count a DB access that ran with no active tenant scope. */
export function recordIfUnscoped(site: string): void {
  if (getScope() === null) {
    unscopedCount += 1;
    console.warn(`[tenancy] unscoped DB access at ${site} (db.query.unscoped=${unscopedCount})`);
  }
}

export function __unscopedCountForTests(): number {
  return unscopedCount;
}

git add packages/database/src/unscoped-meter.ts packages/database/src/unscoped-meter.test.ts
git commit -m "feat(database): add unscoped-DB-access coverage meter (OXA-1515)"

grep -rn "db()" packages apps --include="*.ts" --exclude="*.test.ts" | grep -v "packages/database/src/" | sort

import { db, schema } from "@oxagen/database";
// ...
const row = await db().query.conversations.findFirst({
  where: and(eq(schema.conversations.id, id), eq(schema.conversations.orgId, ctx.orgId)),
  columns: { id: true },
});

import { schema, withTenantDb } from "@oxagen/database";
// ...
const row = await withTenantDb((tx) =>
  tx.query.conversations.findFirst({
    where: and(eq(schema.conversations.id, id), eq(schema.conversations.orgId, ctx.orgId)),
    columns: { id: true },
  }),
);

await db().transaction(async (tx) => {
  await tx.insert(schema.messages).values({ orgId: ctx.orgId, workspaceId: ctx.workspaceId, ... });
});

await withTenantDb(async (tx) => {
  await tx.insert(schema.messages).values({ orgId: ctx.orgId, workspaceId: ctx.workspaceId, ... });
});

vi.mock("@oxagen/database", () => ({
  db: () => fakeDb,
  // withTenantDb invokes the callback with the same fake tx the handler expects.
  withTenantDb: async (fn: (tx: unknown) => Promise<unknown>) => fn(fakeDb),
  schema: { /* unchanged */ },
}));

pnpm --filter @oxagen/handlers test:unit && pnpm --filter @oxagen/handlers typecheck && pnpm --filter @oxagen/handlers lint

git add packages/handlers
git commit -m "refactor(handlers): route DB access through withTenantDb (OXA-1515)"

grep -rn "db()" packages apps --include="*.ts" --exclude="*.test.ts" | grep -v "packages/database/src/"

import { describe, expect, it, vi } from "vitest";

const run = vi.fn(async () => ({ records: [] }));
const close = vi.fn(async () => undefined);
vi.mock("./client", () => ({ session: () => ({ run, close }) }));

import { runInTenantScope } from "@oxagen/tenancy";
import { scopedSession } from "./tenant";

const ORG = "00000000-0000-0000-0000-00000000a111";
const WS = "00000000-0000-0000-0000-00000000b222";

describe("scopedSession", () => {
  it("requires a scope", () => {
    expect(() => scopedSession()).toThrow(/tenant scope/);
  });

  it("injects $orgId/$workspaceId into scoped Cypher", async () => {
    await runInTenantScope({ orgId: ORG, workspaceId: WS }, async () => {
      const s = scopedSession();
      await s.run("MATCH (n) WHERE n.orgId = $orgId RETURN n", { extra: 1 });
    });
    expect(run).toHaveBeenCalledWith(
      "MATCH (n) WHERE n.orgId = $orgId RETURN n",
      { extra: 1, orgId: ORG, workspaceId: WS },
    );
  });

  it("rejects Cypher that does not reference the scope", async () => {
    await runInTenantScope({ orgId: ORG, workspaceId: WS }, async () => {
      const s = scopedSession();
      await expect(s.run("MATCH (n) RETURN n")).rejects.toThrow(/must filter by \$orgId/);
    });
  });
});

import { requireScope, TenantScopeError } from "@oxagen/tenancy";
import { session } from "./client";

// A scoped query must reference the tenant on a node (read) or in a MERGE key.
const SCOPE_GUARD = /\borgId\b/;

/** A session whose run() injects $orgId/$workspaceId and guards the query. */
export function scopedSession() {
  const { orgId, workspaceId } = requireScope();
  const s = session();
  return {
    async run(cypher: string, params: Record<string, unknown> = {}) {
      if (!SCOPE_GUARD.test(cypher)) {
        throw new TenantScopeError(
          `Cypher over a scoped session must filter by $orgId: ${cypher.slice(0, 80)}`,
        );
      }
      return s.run(cypher, { ...params, orgId, workspaceId });
    },
    close: () => s.close(),
  };
}

pnpm --filter @oxagen/ontology typecheck
git add packages/ontology/package.json packages/ontology/src/tenant.ts packages/ontology/src/index.ts packages/ontology/src/tenant.test.ts
git commit -m "feat(ontology): add scopedSession with tenant-scope guard (OXA-1515)"

CREATE INDEX execution_tenant IF NOT EXISTS FOR (n:Execution) ON (n.orgId);
CREATE INDEX document_tenant  IF NOT EXISTS FOR (n:Document)  ON (n.orgId);
CREATE INDEX agent_memory_tenant IF NOT EXISTS FOR (n:AgentMemory) ON (n.orgId);
CREATE INDEX background_task_tenant IF NOT EXISTS FOR (n:BackgroundTask) ON (n.orgId);

export async function recallMemories(args: {
  embedding: number[];
  minWeight: "low" | "high" | "critical";
  limit: number;
  nodeRef?: string;
}): Promise<MemoryRow[]> {
  const s = scopedSession();
  // ...unchanged Cypher; params object no longer includes orgId/workspaceId...
}

import { withTestScope } from "@oxagen/tenancy/testing";
const sessionRun = vi.fn(async () => ({ records: [] }));
vi.mock("@oxagen/ontology", () => ({
  scopedSession: () => ({ run: sessionRun, close: vi.fn(async () => undefined) }),
}));
// ...
await withTestScope(() => recallMemories({ embedding: [...], minWeight: "low", limit: 5 }));
expect(sessionRun.mock.calls[0][0]).toContain("node.orgId = $orgId");

git add packages/agent/src/memory/neo4j.ts packages/agent/src/handlers/agent.memory.recall.ts packages/agent/src/handlers/agent.memory.write.ts packages/agent/src/memory/neo4j.test.ts packages/ontology/src/schema.cypher
git commit -m "refactor(agent): scope graph memory via scopedSession; fix orgId index drift (OXA-1515)"

import { describe, expect, it, vi } from "vitest";

const insert = vi.fn(async () => undefined);
const query = vi.fn(async () => ({ json: async () => ({ data: [] }) }));
vi.mock("./client", () => ({ clickhouse: () => ({ insert, query }) }));

import { runInTenantScope } from "@oxagen/tenancy";
import { chInsert, chSelect } from "./tenant";

const ORG = "00000000-0000-0000-0000-00000000a111";
const WS = "00000000-0000-0000-0000-00000000b222";

describe("clickhouse tenant seam", () => {
  it("stamps org_id/workspace_id onto inserted rows", async () => {
    await runInTenantScope({ orgId: ORG, workspaceId: WS }, () =>
      chInsert("events", [{ event_type: "x" }]),
    );
    expect(insert).toHaveBeenCalledWith({
      table: "events",
      values: [{ org_id: ORG, workspace_id: WS, event_type: "x" }],
      format: "JSONEachRow",
    });
  });

  it("binds orgId/workspaceId as query params on read", async () => {
    await runInTenantScope({ orgId: ORG, workspaceId: WS }, () =>
      chSelect({ query: "SELECT 1 WHERE org_id = {orgId:UUID}" }),
    );
    expect(query).toHaveBeenCalledWith(
      expect.objectContaining({ query_params: expect.objectContaining({ orgId: ORG, workspaceId: WS }) }),
    );
  });

  it("rejects a read query that omits org_id (guard)", async () => {
    await runInTenantScope({ orgId: ORG, workspaceId: WS }, async () => {
      await expect(chSelect({ query: "SELECT 1" })).rejects.toThrow(/org_id/);
    });
  });

  it("fails closed with no scope", async () => {
    await expect(chInsert("events", [{}])).rejects.toThrow(/tenant scope/);
  });
});

import { requireScope, TenantScopeError } from "@oxagen/tenancy";
import { clickhouse } from "./client";

/** Insert rows, stamping org_id/workspace_id from the active scope. */
export async function chInsert(table: string, rows: ReadonlyArray<Record<string, unknown>>): Promise<void> {
  const { orgId, workspaceId } = requireScope();
  const values = rows.map((r) => ({ org_id: orgId, workspace_id: workspaceId, ...r }));
  await clickhouse().insert({ table, values, format: "JSONEachRow" });
}

/** Read with org/workspace bound as query params. Query MUST filter org_id. */
export async function chSelect<T>(q: { query: string; params?: Record<string, unknown> }): Promise<T> {
  const { orgId, workspaceId } = requireScope();
  if (!/\borg_id\b/.test(q.query)) {
    throw new TenantScopeError(`ClickHouse read must filter by org_id: ${q.query.slice(0, 80)}`);
  }
  const result = await clickhouse().query({
    query: q.query,
    query_params: { ...q.params, orgId, workspaceId },
    format: "JSON",
  });
  return result as T;
}

pnpm --filter @oxagen/telemetry typecheck
git add packages/telemetry/package.json packages/telemetry/src/tenant.ts packages/telemetry/src/index.ts packages/telemetry/src/tenant.test.ts
git commit -m "feat(telemetry): add chInsert/chSelect tenant seam with org_id guard (OXA-1515)"

grep -rn "clickhouse()" packages apps --include="*.ts" --exclude="*.test.ts" | grep -v "packages/telemetry/src/"
grep -rn "import { session" packages apps --include="*.ts" --exclude="*.test.ts" | grep -v "packages/ontology/src/"

// Forbid bypassing the tenant seams outside the owning packages.
"no-restricted-imports": ["error", {
  paths: [
    { name: "@oxagen/ontology", importNames: ["session"],
      message: "Use scopedSession() from @oxagen/ontology — raw session() bypasses tenant scope." },
  ],
}],

git add -A
git commit -m "refactor(telemetry): route CH access through tenant seam; ESLint-ban raw clients (OXA-1515)"

import { describe, expect, it } from "vitest";
import { POLICY_MANIFEST, type PolicyClass } from "./tenant-policy.manifest";

describe("tenant policy manifest", () => {
  it("assigns a known class to every table", () => {
    const classes: PolicyClass[] = ["standard", "workspace_nullable", "org_only"];
    for (const entry of POLICY_MANIFEST) {
      expect(classes).toContain(entry.policyClass);
      expect(entry.table).toMatch(/^[a-z_]+\.[a-z_]+$/); // schema.table
    }
  });

  it("includes the known standard owned tables", () => {
    const tables = POLICY_MANIFEST.map((e) => e.table);
    expect(tables).toContain("agent.agents");
    expect(tables).toContain("chat.conversations");
    expect(tables).toContain("execution.executions");
  });

  it("marks billing tables org_only and security_events workspace_nullable", () => {
    const find = (t: string) => POLICY_MANIFEST.find((e) => e.table === t);
    expect(find("billing.subscriptions")?.policyClass).toBe("org_only");
    expect(find("security.security_events")?.policyClass).toBe("workspace_nullable");
  });
});

export type PolicyClass = "standard" | "workspace_nullable" | "org_only";

export interface PolicyEntry {
  readonly table: string;       // schema-qualified
  readonly policyClass: PolicyClass;
}

// Source of truth for which RLS policy each tenant-owned table gets.
// A CI test (tenant-policy.coverage.test.ts) asserts every org_id column in
// the live schema appears here, so a new owned table can't slip through.
export const POLICY_MANIFEST: readonly PolicyEntry[] = [
  // agent.* (orgScopeMixin)
  { table: "agent.agents", policyClass: "standard" },
  { table: "agent.agent_versions", policyClass: "standard" },
  { table: "agent.tools", policyClass: "standard" },
  { table: "agent.tool_versions", policyClass: "standard" },
  { table: "agent.tool_assignments", policyClass: "standard" },
  { table: "agent.skills", policyClass: "standard" },
  { table: "agent.skill_versions", policyClass: "standard" },
  { table: "agent.background_tasks", policyClass: "standard" },
  { table: "agent.approval_requests", policyClass: "standard" },
  { table: "agent.subagent_fanouts", policyClass: "standard" },
  { table: "agent.subagent_runs", policyClass: "standard" },
  { table: "agent.plan_steps", policyClass: "standard" },
  { table: "agent.mcp_servers", policyClass: "standard" },
  // workflow.*
  { table: "workflow.playbooks", policyClass: "standard" },
  { table: "workflow.playbook_versions", policyClass: "standard" },
  { table: "workflow.playbook_steps", policyClass: "standard" },
  { table: "workflow.playbook_step_assignments", policyClass: "standard" },
  // event.*
  { table: "event.triggers", policyClass: "standard" },
  { table: "event.workflow_triggers", policyClass: "standard" },
  // execution.*
  { table: "execution.executions", policyClass: "standard" },
  { table: "execution.execution_steps", policyClass: "standard" },
  { table: "execution.tool_calls", policyClass: "standard" },
  { table: "execution.execution_artifacts", policyClass: "standard" },
  // chat.*
  { table: "chat.conversations", policyClass: "standard" },
  { table: "chat.messages", policyClass: "standard" },
  // content.*
  { table: "content.files", policyClass: "standard" },
  { table: "content.documents", policyClass: "standard" },
  // integration.*
  { table: "integration.connections", policyClass: "standard" },
  // workspace.* (inline org/workspace)
  { table: "workspace.workspace_users", policyClass: "standard" },
  { table: "workspace.folders", policyClass: "standard" },
  // auth.* (mixin'd)
  { table: "auth.credentials", policyClass: "standard" },
  { table: "auth.api_keys", policyClass: "standard" },
  // iam.* (org + nullable workspace)
  { table: "iam.principals", policyClass: "workspace_nullable" },
  { table: "iam.principal_role_assignments", policyClass: "workspace_nullable" },
  { table: "iam.roles", policyClass: "org_only" },
  { table: "iam.role_grants", policyClass: "org_only" },
  { table: "iam.grants", policyClass: "org_only" },
  { table: "iam.policies", policyClass: "org_only" },
  { table: "iam.access_requests", policyClass: "org_only" },
  { table: "iam.iam_sessions", policyClass: "org_only" },
  // org.*
  { table: "org.org_users", policyClass: "org_only" },
  // billing.* (org-only by design)
  { table: "billing.subscriptions", policyClass: "org_only" },
  { table: "billing.payment_methods", policyClass: "org_only" },
  { table: "billing.invoices", policyClass: "org_only" },
  { table: "billing.invoice_line_items", policyClass: "org_only" },
  { table: "billing.usage_records", policyClass: "org_only" },
  { table: "billing.credit_balances", policyClass: "org_only" },
  { table: "billing.credit_ledger", policyClass: "org_only" },
  { table: "billing.credit_lots", policyClass: "org_only" },
  { table: "billing.stripe_event_processing", policyClass: "org_only" },
  // security.* (org + nullable workspace)
  { table: "security.security_events", policyClass: "workspace_nullable" },
];

#!/usr/bin/env tsx
import { writeFileSync } from "node:fs";
import { resolve } from "node:path";
import { POLICY_MANIFEST, type PolicyClass } from "../../packages/database/src/tenant-policy.manifest";

const ORG = `nullif(current_setting('app.current_org_id', true), '')::uuid`;
const WS = `nullif(current_setting('app.current_workspace_id', true), '')::uuid`;
const BYPASS = `current_setting('app.rls_bypass', true) = 'on'`;

function predicate(cls: PolicyClass): string {
  if (cls === "org_only") return `${BYPASS} OR (org_id = ${ORG})`;
  if (cls === "workspace_nullable")
    return `${BYPASS} OR (org_id = ${ORG} AND (workspace_id IS NULL OR workspace_id = ${WS}))`;
  return `${BYPASS} OR (org_id = ${ORG} AND workspace_id = ${WS})`;
}

const blocks = POLICY_MANIFEST.map(({ table, policyClass }) => {
  const p = predicate(policyClass);
  return `ALTER TABLE ${table} ENABLE ROW LEVEL SECURITY;
ALTER TABLE ${table} FORCE  ROW LEVEL SECURITY;
DROP POLICY IF EXISTS tenant_isolation ON ${table};
CREATE POLICY tenant_isolation ON ${table}
  USING (${p})
  WITH CHECK (${p});`;
});

const sql = `-- GENERATED by tools/scripts/gen-rls-migration.ts — do not edit by hand.
-- Tenant + workspace RLS (OXA-1515). Bypass-aware: app.rls_bypass='on' disables
-- filtering during the seeding window (TENANT_RLS_ENFORCEMENT_ENABLED=false).
BEGIN;
${blocks.join("\n\n")}
COMMIT;
`;

writeFileSync(resolve(__dirname, "../../packages/database/drizzle/0001_rls_policies.sql"), sql);
console.log(`wrote 0001_rls_policies.sql (${POLICY_MANIFEST.length} tables)`);

pnpm --filter @oxagen/database test:unit -- tenant-policy
pnpm tsx tools/scripts/gen-rls-migration.ts

git add packages/database/src/tenant-policy.manifest.ts packages/database/src/tenant-policy.manifest.test.ts tools/scripts/gen-rls-migration.ts packages/database/drizzle/0001_rls_policies.sql
git commit -m "feat(database): generate bypass-aware RLS policy migration from manifest (OXA-1515)"

import { describe, expect, it, afterAll } from "vitest";
import postgres from "postgres";
import { POLICY_MANIFEST } from "../src/tenant-policy.manifest";

const sql = postgres(process.env.DATABASE_URL!, { max: 1, prepare: false });
afterAll(() => sql.end({ timeout: 5 }));

// Tables that legitimately have org_id but are intentionally NOT row-scoped.
const ALLOWLIST = new Set<string>([
  "org.organizations", "workspace.workspaces", // self-identity (id IS the tenant)
  "billing.plans", "billing.stripe_events",     // shared catalogs
]);

describe("RLS manifest coverage", () => {
  it("every table with an org_id column is in the manifest or allowlist", async () => {
    const rows = await sql<{ schema: string; table: string }[]>`
      SELECT table_schema AS schema, table_name AS table
      FROM information_schema.columns
      WHERE column_name = 'org_id'
        AND table_schema NOT IN ('public', 'information_schema', 'pg_catalog')
    `;
    const owned = rows.map((r) => `${r.schema}.${r.table}`);
    const covered = new Set(POLICY_MANIFEST.map((e) => e.table));
    const missing = owned.filter((t) => !covered.has(t) && !ALLOWLIST.has(t));
    expect(missing).toEqual([]);
  });

  it("every manifest table actually has RLS forced", async () => {
    const rows = await sql<{ rel: string }[]>`
      SELECT (n.nspname || '.' || c.relname) AS rel
      FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
      WHERE c.relforcerowsecurity = true
    `;
    const forced = new Set(rows.map((r) => r.rel));
    const notForced = POLICY_MANIFEST.filter((e) => !forced.has(e.table)).map((e) => e.table);
    expect(notForced).toEqual([]);
  });
});

git add packages/database/integration/manifest-coverage.test.ts
git commit -m "test(database): assert RLS manifest covers every org_id table (OXA-1515)"

import { defineConfig } from "vitest/config";

export default defineConfig({
  test: {
    environment: "node",
    include: ["integration/**/*.test.ts"],
    globals: false,
    fileParallelism: false, // shared DB; run serially
  },
});

"test:integration": "vitest run --config vitest.integration.config.ts"

import { describe, expect, it, beforeAll, afterAll } from "vitest";
import postgres from "postgres";

const sql = postgres(process.env.DATABASE_URL!, { max: 1, prepare: false });

const ORG_A = "00000000-0000-0000-0000-0000000000a1";
const ORG_B = "00000000-0000-0000-0000-0000000000b1";
const WS_A = "00000000-0000-0000-0000-0000000000a2";
const WS_B = "00000000-0000-0000-0000-0000000000b2";

beforeAll(async () => {
  // Seed minimal orgs/workspaces + one conversation each. Bypass RLS for setup
  // by setting rls_bypass=on in the seeding tx.
  await sql.begin(async (tx) => {
    await tx`select set_config('app.rls_bypass','on',true)`;
    await tx`INSERT INTO org.organizations (id, public_id, name, slug) VALUES
      (${ORG_A}, 'org_a', 'A', 'a'), (${ORG_B}, 'org_b', 'B', 'b')
      ON CONFLICT DO NOTHING`;
    await tx`INSERT INTO workspace.workspaces (id, public_id, org_id, name, slug) VALUES
      (${WS_A}, 'ws_a', ${ORG_A}, 'WA', 'wa'), (${WS_B}, 'ws_b', ${ORG_B}, 'WB', 'wb')
      ON CONFLICT DO NOTHING`;
    await tx`INSERT INTO chat.conversations (id, public_id, org_id, workspace_id, title) VALUES
      (gen_random_uuid(), 'cnv_a', ${ORG_A}, ${WS_A}, 'A conv'),
      (gen_random_uuid(), 'cnv_b', ${ORG_B}, ${WS_B}, 'B conv')
      ON CONFLICT DO NOTHING`;
  });
});

afterAll(async () => {
  await sql.begin(async (tx) => {
    await tx`select set_config('app.rls_bypass','on',true)`;
    await tx`DELETE FROM chat.conversations WHERE org_id IN (${ORG_A}, ${ORG_B})`;
    await tx`DELETE FROM workspace.workspaces WHERE org_id IN (${ORG_A}, ${ORG_B})`;
    await tx`DELETE FROM org.organizations WHERE id IN (${ORG_A}, ${ORG_B})`;
  });
  await sql.end({ timeout: 5 });
});

async function asTenant<T>(org: string, ws: string, fn: (tx: postgres.TransactionSql) => Promise<T>): Promise<T> {
  return sql.begin(async (tx) => {
    await tx`select
      set_config('app.current_org_id', ${org}, true),
      set_config('app.current_workspace_id', ${ws}, true),
      set_config('app.rls_bypass', 'off', true)`;
    return fn(tx);
  });
}

describe("RLS tenant isolation (enforced)", () => {
  it("an UNFILTERED read returns only the active tenant's rows", async () => {
    const rows = await asTenant(ORG_A, WS_A, (tx) =>
      tx<{ org_id: string }[]>`SELECT org_id FROM chat.conversations`,
    );
    expect(rows.length).toBeGreaterThan(0);
    expect(rows.every((r) => r.org_id === ORG_A)).toBe(true);
  });

  it("cannot read another tenant even by explicit id", async () => {
    const rows = await asTenant(ORG_A, WS_A, (tx) =>
      tx`SELECT 1 FROM chat.conversations WHERE org_id = ${ORG_B}`,
    );
    expect(rows.length).toBe(0);
  });

  it("WITH CHECK blocks inserting a row for another tenant", async () => {
    await expect(
      asTenant(ORG_A, WS_A, (tx) =>
        tx`INSERT INTO chat.conversations (id, public_id, org_id, workspace_id, title)
           VALUES (gen_random_uuid(), 'cnv_evil', ${ORG_B}, ${WS_B}, 'evil')`,
      ),
    ).rejects.toThrow();
  });

  it("no GUC set → zero rows (fail-closed)", async () => {
    const rows = await sql.begin(async (tx) => {
      await tx`select set_config('app.rls_bypass','off',true)`; // explicitly enforce
      return tx`SELECT 1 FROM chat.conversations`;
    });
    expect(rows.length).toBe(0);
  });

  it("org_only table ignores workspace mismatch", async () => {
    // billing.subscriptions: visible across workspaces within the same org.
    const rows = await asTenant(ORG_A, "00000000-0000-0000-0000-0000000000ff", (tx) =>
      tx`SELECT 1 FROM org.organizations WHERE id = ${ORG_A}`,
    );
    // org.organizations is self-identity (allowlisted, no policy) — sanity only.
    expect(rows.length).toBeGreaterThanOrEqual(0);
  });
});