diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 16c8fe68..02db4e41 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -49,6 +49,57 @@ jobs:
           --with-editable "packages/agami-core[model,server]"
           pytest tests/ -q --cov=plugins --cov=packages/agami-core/src --cov-report=term-missing
 
+  integration-pg:
+    # The F9 safety regression gate that needs a real database: the Postgres-SERVED corpus path
+    # (file/db parity) + the read-only role floor. These skip cleanly in `lint-and-test` (no DB);
+    # here a Postgres service is present so they RUN. Make this a required check for the F9 done-bar.
+    name: integration (Postgres — F9 safety corpus + role floor)
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_DB: shop
+          POSTGRES_USER: agami_test
+          # Ephemeral localhost CI service: trust auth (no password) — nothing to leak, and the tests
+          # assert role PRIVILEGES, not authentication. Avoids a throwaway password literal in the diff.
+          POSTGRES_HOST_AUTH_METHOD: trust
+        ports:
+          - 5432:5432
+        # Wait for the server to accept connections before the steps run.
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      # The env-gated PG tests read these (defaults target the local compose fixture); point them at
+      # the service. AGAMI_IT_PG_PASSWORD being set is what flips the tests from skip to run.
+      AGAMI_IT_PG_HOST: 127.0.0.1
+      AGAMI_IT_PG_PORT: "5432"
+      AGAMI_IT_PG_USER: agami_test
+      # A non-secret sentinel: it only has to be non-empty to flip the tests from skip→run (the service
+      # uses trust auth, so the value is never verified). Not a credential.
+      AGAMI_IT_PG_PASSWORD: trust
+      AGAMI_IT_PG_DB: shop
+      # In this job a DB is REQUIRED: an unavailable/unreachable Postgres FAILS the job instead of
+      # skipping, so an all-skip can't pass green and silently prove nothing (the F9 done-bar gate).
+      AGAMI_IT_PG_REQUIRED: "1"
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
+      - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        with:
+          python-version: "3.12"
+      # The [server] extra pulls psycopg2; run ONLY the DB-dependent safety tests (the file path is
+      # already covered by lint-and-test). A regression in either fails the F9 gate.
+      - name: pytest (Postgres safety corpus + read-only role floor)
+        run: >-
+          uvx --python 3.12
+          --with pytest
+          --with-editable "packages/agami-core[model,server]"
+          pytest tests/e2e/test_safety_corpus.py tests/e2e/test_role_floor_pg.py
+          -q -k "db_path or role"
+
   gitleaks:
     name: gitleaks (secret scan)
     runs-on: ubuntu-latest
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
new file mode 100644
index 00000000..cb247f37
--- /dev/null
+++ b/tests/e2e/conftest.py
@@ -0,0 +1,209 @@
+"""Shared e2e fixtures for the F9 safety corpus (ACE-040): the two transport surfaces and the two
+model paths, wired so the corpus asserts the same Envelope regardless of surface or path."""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+PKG_SRC = REPO_ROOT / "packages" / "agami-core" / "src"
+if str(PKG_SRC) not in sys.path:
+    sys.path.insert(0, str(PKG_SRC))
+if str(REPO_ROOT / "tests") not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT / "tests"))
+
+import harness  # noqa: E402  (tests/e2e is on sys.path during collection)
+
+BASE = "https://demo.example.com"
+
+
+@pytest.fixture
+def presence_auth(monkeypatch):
+    """HTTP bearer-presence mode: PUBLIC_BASE_URL set, no signing secret → 'Bearer present' works.
+    Harmless for the stdio surface (the subprocess inherits the env but doesn't gate on the bearer)."""
+    monkeypatch.setenv("PUBLIC_BASE_URL", BASE)
+    monkeypatch.delenv("AGAMI_SIGNING_SECRET", raising=False)
+
+
+@pytest.fixture(params=["stdio", "http"])
+def surface(request, presence_auth):
+    """Parametrize a test across BOTH transports; yields the driver `(sql, datasource=, max_rows=)`."""
+    return harness.SURFACES[request.param]
+
+
+@pytest.fixture
+def file_safety_env(tmp_path, monkeypatch):
+    """The FILE-served model path: an on-disk model (AGAMI_ARTIFACTS_DIR) + a seeded SQLite datasource
+    (via the DATASOURCE_URL__<PROFILE> env DSN, which both surfaces inherit). No AGAMI_DB_URL ⇒ local
+    (not hosted), so the guards resolve the model from disk. Default (enforce) unscopable posture."""
+    art = tmp_path / "art"
+    (art / "acme").mkdir(parents=True)
+    harness.write_disk_model(art / "acme")
+    db = tmp_path / "shop.db"
+    harness.seed_sqlite(db)
+
+    monkeypatch.delenv("AGAMI_DB_URL", raising=False)
+    monkeypatch.delenv("APP_DATABASE_URL", raising=False)
+    monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(art))
+    monkeypatch.setenv("AGAMI_PROFILE", "acme")
+    monkeypatch.setenv("DATASOURCE_URL__ACME", "sqlite:///" + str(db))
+    monkeypatch.delenv(
+        "AGAMI_SQL_UNSCOPABLE_POSTURE", raising=False
+    )  # enforce (the shipped default)
+    yield
+
+
+# ── the DB-served model path (Postgres-in-Docker) — env-gated, skips without a reachable PG ────
+# The corpus's "both paths" proof: the SAME cases run against a Postgres-SERVED model + a Postgres
+# datasource, so file-served and DB-served can't silently diverge. Opt-in via AGAMI_IT_PG_PASSWORD
+# (the integration-pg CI job / `docker compose -f tests/integration/docker-compose.yml up postgres`).
+# NOTE: these fixtures DROP/CREATE the shared global role `agami_ro` and shared tables in the single
+# `shop` DB, so they are SERIAL-ONLY — do not run this dir under pytest-xdist (`-n`) without per-worker
+# role/table names, or workers would race. The current invocation is serial.
+# The read-only role's password is DERIVED from the (test-only) PG password env, never a hardcoded
+# literal — the fixture Postgres is an ephemeral localhost CI service, and the role-floor tests
+# PRIVILEGES, not auth (the CI service uses trust auth, so this value is not verified anyway).
+_RO_PASSWORD = "ro_" + os.environ.get("AGAMI_IT_PG_PASSWORD", "local")
+
+
+def pg_super_creds() -> dict:
+    """Superuser creds for the fixture Postgres (host/port/user/db default to the compose fixture)."""
+    return {
+        "host": os.environ.get("AGAMI_IT_PG_HOST", "127.0.0.1"),
+        "port": int(os.environ.get("AGAMI_IT_PG_PORT", "55432")),
+        "user": os.environ.get("AGAMI_IT_PG_USER", "agami_test"),
+        "password": os.environ.get("AGAMI_IT_PG_PASSWORD", ""),
+        "dbname": os.environ.get("AGAMI_IT_PG_DB", "shop"),
+    }
+
+
+@pytest.fixture
+def pg_admin():
+    """An autocommit superuser connection to the fixture Postgres. Normally SKIPS (never fails) when
+    no AGAMI_IT_PG_PASSWORD is set or no Postgres is reachable, so the DB-free test job is unaffected.
+
+    BUT when AGAMI_IT_PG_REQUIRED is set (the integration-pg CI job sets it), an unavailable DB FAILS
+    instead of skips — this job carries the ONLY proof of the role-floor + file/db parity + DB-served
+    model, and pytest exits 0 when everything skips, so a service race / env rename / driver hiccup
+    would otherwise turn the F9 done-bar gate green while proving nothing."""
+    psycopg2 = pytest.importorskip("psycopg2")
+    # In the required job, a missing DB is a hard failure — an all-skip must NOT pass as green.
+    unavailable = pytest.fail if os.environ.get("AGAMI_IT_PG_REQUIRED") else pytest.skip
+    if not os.environ.get("AGAMI_IT_PG_PASSWORD"):
+        unavailable("set AGAMI_IT_PG_PASSWORD to run the Postgres safety-corpus / role-floor tests")
+    sc = pg_super_creds()
+    try:
+        conn = psycopg2.connect(connect_timeout=10, **sc)
+    except Exception as exc:  # unreachable DB → skip locally, FAIL in the required CI job
+        unavailable(f"no reachable Postgres ({exc})")
+    conn.autocommit = True
+    try:
+        yield psycopg2, conn, sc
+    finally:
+        conn.close()
+
+
+def _reset_ro_role(cur) -> None:
+    """Drop the read-only role + any grants it holds, tolerating 'does not exist' (setup + teardown)."""
+    for stmt in (
+        "REVOKE ALL ON ALL TABLES IN SCHEMA public FROM agami_ro",
+        "REVOKE ALL ON SCHEMA public FROM agami_ro",
+        "DROP OWNED BY agami_ro",
+        "DROP ROLE IF EXISTS agami_ro",
+    ):
+        try:
+            cur.execute(stmt)
+        except Exception:  # role/grant may not exist yet (autocommit conn → no aborted-txn carry)
+            pass
+
+
+def create_ro_role(cur, dbname: str) -> None:
+    """(Re)create the SELECT-only `agami_ro` role + grants — verbatim from readonly-grants.md."""
+    _reset_ro_role(cur)
+    cur.execute("CREATE ROLE agami_ro LOGIN PASSWORD %s", (_RO_PASSWORD,))
+    cur.execute(f"GRANT CONNECT ON DATABASE {dbname} TO agami_ro")
+    cur.execute("GRANT USAGE ON SCHEMA public TO agami_ro")
+    cur.execute("GRANT SELECT ON ALL TABLES IN SCHEMA public TO agami_ro")
+    cur.execute("ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO agami_ro")
+
+
+@pytest.fixture
+def db_safety_env(pg_admin, tmp_path, monkeypatch):
+    """The DB-served model path: the model is written to the app DB (hosted → model_store), the demo
+    datasource tables live in the same Postgres, and the app connects to them as the read-only role."""
+    psycopg2, conn, sc = pg_admin
+    cur = conn.cursor()
+
+    # 1) demo datasource tables in public (owner-seeded), from the single-sourced SCHEMA.
+    from safety.corpus import SCHEMA
+
+    for name, spec in SCHEMA.items():
+        cur.execute(f"DROP TABLE IF EXISTS {name} CASCADE")
+        pg_types = {"INTEGER": "integer", "REAL": "double precision", "TEXT": "text"}
+        ddl = ", ".join(f"{c} {pg_types[t]}" for c, t in spec["columns"])
+        cur.execute(f"CREATE TABLE {name} ({ddl})")
+        placeholders = ", ".join("%s" for _ in spec["columns"])
+        for row in spec["rows"]:
+            cur.execute(f"INSERT INTO {name} VALUES ({placeholders})", row)
+
+    # 2) the SELECT-only role the app connects to the datasource as.
+    create_ro_role(cur, sc["dbname"])
+
+    # 3) the DB-served model: migrate the app schema + write the org into it (the hosted model source).
+    super_dsn = (
+        f"postgresql://{sc['user']}:{sc['password']}@{sc['host']}:{sc['port']}/{sc['dbname']}"
+    )
+    harness.seed_db_model(super_dsn, ds="acme")
+
+    ro_dsn = f"postgresql://agami_ro:{_RO_PASSWORD}@{sc['host']}:{sc['port']}/{sc['dbname']}"
+    monkeypatch.setenv("AGAMI_DB_URL", super_dsn)  # hosted → model + audit from the app DB
+    monkeypatch.setenv("DATASOURCE_URL__ACME", ro_dsn)  # datasource read as the read-only role
+    monkeypatch.setenv("AGAMI_PROFILE", "acme")
+    monkeypatch.setenv(
+        "AGAMI_ARTIFACTS_DIR", str(tmp_path / "no_disk")
+    )  # DB is the only model source
+    monkeypatch.delenv("AGAMI_SQL_UNSCOPABLE_POSTURE", raising=False)
+    try:
+        yield
+    finally:
+        for name in SCHEMA:
+            try:
+                cur.execute(f"DROP TABLE IF EXISTS {name} CASCADE")
+            except Exception:
+                pass
+        _reset_ro_role(cur)
+
+
+@pytest.fixture
+def pg_ro_conn(pg_admin):
+    """A RAW connection AS the read-only role, plus one seeded table — for the role-floor test. This
+    connection bypasses the app layer entirely (no tool_execute_sql / no app read-only gate), so a
+    write reaching it is stopped by the DATABASE itself — the primary control, proven independent of
+    the app gate."""
+    psycopg2, conn, sc = pg_admin
+    cur = conn.cursor()
+    cur.execute("DROP TABLE IF EXISTS agami_floor CASCADE")
+    cur.execute("CREATE TABLE agami_floor (id integer, label text)")
+    cur.execute("INSERT INTO agami_floor VALUES (1, 'a'), (2, 'b')")
+    create_ro_role(cur, sc["dbname"])
+    ro = psycopg2.connect(
+        host=sc["host"],
+        port=sc["port"],
+        user="agami_ro",
+        password=_RO_PASSWORD,
+        dbname=sc["dbname"],
+        connect_timeout=3,
+    )
+    try:
+        yield psycopg2, ro
+    finally:
+        ro.close()
+        try:
+            cur.execute("DROP TABLE IF EXISTS agami_floor CASCADE")
+        except Exception:
+            pass
+        _reset_ro_role(cur)
diff --git a/tests/e2e/harness.py b/tests/e2e/harness.py
new file mode 100644
index 00000000..27499494
--- /dev/null
+++ b/tests/e2e/harness.py
@@ -0,0 +1,206 @@
+"""Shared e2e harness for the F9 safety corpus (ACE-040): the two transport drivers (stdio + HTTP)
+and the builders that materialize the demo model + datasource from `tests/safety/corpus.SCHEMA`.
+
+Kept as a plain importable module (not conftest) so both `test_safety_corpus.py` and the existing
+`test_safety_envelope.py` can share ONE copy of the drivers — the "both surfaces in sync" proof lives
+in one place, not duplicated per file.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sqlite3
+import subprocess
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+PKG_SRC = REPO_ROOT / "packages" / "agami-core" / "src"
+if str(PKG_SRC) not in sys.path:
+    sys.path.insert(0, str(PKG_SRC))
+if str(REPO_ROOT / "tests") not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT / "tests"))  # so `safety.corpus` resolves (repo convention)
+
+from safety.corpus import SCHEMA  # noqa: E402
+
+_TYPE_MAP = {"INTEGER": "integer", "REAL": "float", "TEXT": "string"}
+
+
+# ── model + datasource builders (single-sourced from SCHEMA) ───────────────────────────────────
+def build_org():
+    """The semantic model the guards scope against — built from SCHEMA (names + sensitive flags)."""
+    from semantic_model import models as m
+
+    def _table(name: str, spec: dict):
+        cols = [
+            m.Column(name=c, type=_TYPE_MAP.get(t, "string"), sensitive=(c in spec["sensitive"]))
+            for c, t in spec["columns"]
+        ]
+        return m.Table(
+            name=name,
+            schema="public",
+            storage_connection="c",
+            grain=["id"],
+            description=name,
+            columns=cols,
+        )
+
+    return m.Organization(
+        organization="Shop",
+        version=1,
+        subject_areas=[
+            m.SubjectArea(name="sales", tables_defined=[_table(n, s) for n, s in SCHEMA.items()])
+        ],
+    )
+
+
+def write_disk_model(root: Path) -> None:
+    """Write the FILE-served model under `root` (an AGAMI_ARTIFACTS_DIR/<profile> dir)."""
+    import yaml
+
+    (root / "subject_areas" / "sales" / "tables").mkdir(parents=True, exist_ok=True)
+    (root / "org.yaml").write_text(
+        yaml.safe_dump(
+            {"organization": "Shop", "version": 1, "subject_areas": ["subject_areas/sales"]}
+        )
+    )
+    (root / "subject_areas" / "sales" / "subject_area.yaml").write_text(
+        yaml.safe_dump(
+            {
+                "name": "sales",
+                "tables": [
+                    {"storage_connection": "c", "schema": "public", "table": t} for t in SCHEMA
+                ],
+            }
+        )
+    )
+    for name, spec in SCHEMA.items():
+        cols = []
+        for cname, ctype in spec["columns"]:
+            col = {"name": cname, "type": _TYPE_MAP.get(ctype, "string")}
+            if cname == "id":
+                col["primary_key"] = True
+            if cname in spec["sensitive"]:
+                col["sensitive"] = True
+            cols.append(col)
+        (root / "subject_areas" / "sales" / "tables" / f"{name}.yaml").write_text(
+            yaml.safe_dump(
+                {
+                    "name": name,
+                    "schema": "public",
+                    "storage_connection": "c",
+                    "grain": ["id"],
+                    "description": name,
+                    "columns": cols,
+                }
+            )
+        )
+
+
+def seed_sqlite(path: Path) -> None:
+    """Create + seed the physical SQLite datasource governed queries execute against."""
+    con = sqlite3.connect(str(path))
+    try:
+        for name, spec in SCHEMA.items():
+            ddl = ", ".join(f"{c} {t}" for c, t in spec["columns"])
+            con.execute(f"CREATE TABLE {name} ({ddl})")
+            placeholders = ", ".join("?" for _ in spec["columns"])
+            con.executemany(f"INSERT INTO {name} VALUES ({placeholders})", spec["rows"])
+        con.commit()
+    finally:
+        con.close()
+
+
+def seed_db_model(url: str, ds: str = "acme") -> None:
+    """Write the DB-served model into an app DB at `url` (the hosted path's model source)."""
+    import model_store
+    from store import Store
+
+    s = Store.connect(url)
+    s.run_migrations()
+    model_store.write_organization(s, ds, build_org())
+    s.close()
+
+
+# ── transport drivers: each returns the execute_sql tool's parsed Envelope ─────────────────────
+def _tool_args(sql: str, datasource: str | None, max_rows: int | None) -> dict:
+    args: dict = {"sql": sql}
+    if datasource:
+        args["datasource"] = datasource
+    if max_rows is not None:
+        args["max_rows"] = max_rows
+    return args
+
+
+def stdio_execute_sql(sql: str, datasource: str | None = None, max_rows: int | None = None) -> dict:
+    """Drive execute_sql over the real stdio server (a subprocess), return the tool's Envelope."""
+    msgs = [
+        {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
+        {
+            "jsonrpc": "2.0",
+            "id": 2,
+            "method": "tools/call",
+            "params": {"name": "execute_sql", "arguments": _tool_args(sql, datasource, max_rows)},
+        },
+    ]
+    stdin = "".join(json.dumps(m) + "\n" for m in msgs)
+    proc = subprocess.run(
+        [sys.executable, "-m", "mcp_harness"],
+        input=stdin,
+        capture_output=True,
+        text=True,
+        timeout=60,
+        env={**os.environ},
+    )
+    by_id = {m.get("id"): m for m in (json.loads(x) for x in proc.stdout.splitlines() if x.strip())}
+    return json.loads(by_id[2]["result"]["content"][0]["text"])
+
+
+def http_execute_sql(sql: str, datasource: str | None = None, max_rows: int | None = None) -> dict:
+    """Drive execute_sql over the real HTTP transport (in-process TestClient), return the Envelope."""
+    import mcp_http
+    from starlette.testclient import TestClient
+
+    headers = {
+        "Authorization": "Bearer present",
+        "Content-Type": "application/json",
+        "Accept": "application/json, text/event-stream",
+    }
+    with TestClient(mcp_http.build_app()) as c:
+        init = c.post(
+            "/mcp",
+            headers=headers,
+            json={
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "initialize",
+                "params": {
+                    "protocolVersion": "2025-06-18",
+                    "capabilities": {},
+                    "clientInfo": {"name": "t", "version": "1"},
+                },
+            },
+        )
+        sid = init.headers.get("mcp-session-id")
+        h2 = {**headers, **({"mcp-session-id": sid} if sid else {})}
+        c.post("/mcp", headers=h2, json={"jsonrpc": "2.0", "method": "notifications/initialized"})
+        r = c.post(
+            "/mcp",
+            headers=h2,
+            json={
+                "jsonrpc": "2.0",
+                "id": 2,
+                "method": "tools/call",
+                "params": {
+                    "name": "execute_sql",
+                    "arguments": _tool_args(sql, datasource, max_rows),
+                },
+            },
+        )
+    rpc = json.loads(re.search(r"\{.*\}", r.text, re.DOTALL).group(0))
+    return json.loads(rpc["result"]["content"][0]["text"])
+
+
+SURFACES = {"stdio": stdio_execute_sql, "http": http_execute_sql}
diff --git a/tests/e2e/test_role_floor_pg.py b/tests/e2e/test_role_floor_pg.py
new file mode 100644
index 00000000..3ea7cee2
--- /dev/null
+++ b/tests/e2e/test_role_floor_pg.py
@@ -0,0 +1,44 @@
+"""ACE-040 — the ROLE FLOOR: the read-only DB role rejects a write even with the APP GATE BYPASSED.
+
+The app-layer read-only guard (`check_read_only`) refuses INSERT / UPDATE / DELETE / DDL before any
+driver call — but that is defense in depth. The PRIMARY control is the database role: agami connects
+as a SELECT-only user (see plugins/agami/shared/readonly-grants.md), so even if the app gate were
+bypassed by a bug or a new code path, the DATABASE itself rejects the write.
+
+This test proves that floor. It issues writes on a RAW psycopg2 connection opened AS the read-only
+role — deliberately NOT through `tool_execute_sql` (which would refuse them at the app gate first) —
+and asserts Postgres raises `InsufficientPrivilege`. A read succeeds, so the floor blocks writes, not
+work. Opt-in: skips unless AGAMI_IT_PG_PASSWORD is set + a Postgres is reachable (the integration-pg
+CI job provides both; locally: `docker compose -f tests/integration/docker-compose.yml up -d postgres`).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+# The read-only role must PERMIT reads and REJECT every write path — at the database, not the app.
+_WRITE_ATTEMPTS = [
+    "INSERT INTO agami_floor VALUES (9, 'x')",
+    "UPDATE agami_floor SET label = 'y'",
+    "DELETE FROM agami_floor",
+    "TRUNCATE agami_floor",
+    "DROP TABLE agami_floor",
+]
+
+
+def test_read_only_role_permits_select(pg_ro_conn):
+    _psycopg2, ro = pg_ro_conn
+    with ro.cursor() as c:
+        c.execute("SELECT id, label FROM agami_floor ORDER BY id")
+        assert c.fetchall() == [(1, "a"), (2, "b")]  # the floor permits reads (agami only SELECTs)
+
+
+@pytest.mark.parametrize("write_sql", _WRITE_ATTEMPTS, ids=[w.split()[0] for w in _WRITE_ATTEMPTS])
+def test_read_only_role_rejects_writes_with_app_gate_bypassed(pg_ro_conn, write_sql):
+    psycopg2, ro = pg_ro_conn
+    # NOTE: issued DIRECTLY on the role's connection — the app read-only gate is NOT in the loop here.
+    # If this write ever SUCCEEDS, the primary control has regressed (the role isn't SELECT-only).
+    with pytest.raises(psycopg2.errors.InsufficientPrivilege):
+        with ro.cursor() as c:
+            c.execute(write_sql)
+    ro.rollback()  # clear the aborted transaction before the fixture tears down
diff --git a/tests/e2e/test_safety_corpus.py b/tests/e2e/test_safety_corpus.py
new file mode 100644
index 00000000..4c3e479e
--- /dev/null
+++ b/tests/e2e/test_safety_corpus.py
@@ -0,0 +1,67 @@
+"""ACE-040 — the F9 safety regression corpus, end-to-end over BOTH surfaces (file-served model path).
+
+Every attack class in `tests/safety/corpus.CASES` is driven through the REAL execute_sql tool on both
+transports (stdio subprocess + in-process HTTP) and asserted against its expected Envelope. This is
+the F9 done-bar for the controls it asserts: a regression in read-only, object-scope, fail-closed
+scopability, or recon fails here on whichever surface it regressed; availability is asserted via the
+row-cap arm (the statement-timeout arm is proven end-to-end in `tests/test_resource_limits.py`). The
+read-only-role floor lives in `test_role_floor_pg.py` (Postgres, env-gated).
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("mcp")
+pytest.importorskip("starlette")
+pytest.importorskip("sqlglot")
+pytest.importorskip("pydantic")
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(REPO_ROOT / "tests") not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT / "tests"))
+
+from safety.corpus import CASES  # noqa: E402
+
+
+def assert_outcome(env: dict, expect: str, sql: str) -> None:
+    """Assert the tool's Envelope matches the case's expected outcome — the single mapping used for
+    every surface and every model path, so a surface/path can't silently diverge."""
+    if expect == "ok":
+        assert env["status"] == "ok", (sql, env)
+        assert "refusal" not in env
+        assert env["audit_id"]
+    elif expect == "bounded":
+        # Availability: a runaway result is bounded — EITHER a resource_limit refusal, OR an ok
+        # Envelope flagged truncated with a row_cap in `applied` (capped + flagged, never silent).
+        if env["status"] == "refused":
+            assert env["refusal"]["kind"] == "resource_limit", (sql, env)
+        else:
+            assert env["status"] == "ok", (sql, env)
+            assert env["data"]["truncated"] is True, (sql, env)
+            assert any("row_cap" in a for a in env.get("applied", [])), (sql, env)
+    else:
+        # A refusal kind: the query is refused with exactly this kind, carries no data, is audited.
+        assert env["status"] == "refused", (sql, env)
+        assert env["refusal"]["kind"] == expect, (sql, env)
+        assert "data" not in env
+        assert env["audit_id"]
+
+
+@pytest.mark.parametrize("case", CASES, ids=[c.id for c in CASES])
+def test_safety_corpus_file_path(case, surface, file_safety_env):
+    # File-served model (disk YAML) + SQLite datasource. Runs in the default (DB-free) test job.
+    env = surface(case.sql, datasource="acme", max_rows=case.max_rows)
+    assert_outcome(env, case.expect, case.sql)
+
+
+@pytest.mark.parametrize("case", CASES, ids=[c.id for c in CASES])
+def test_safety_corpus_db_path(case, surface, db_safety_env):
+    # DB-served model (Postgres app DB) + Postgres datasource read as the read-only role. IDENTICAL
+    # verdicts to the file path prove file/db parity (a control that reads the model can't behave
+    # differently by source). Env-gated: skips unless a Postgres is reachable (the integration-pg job).
+    env = surface(case.sql, datasource="acme", max_rows=case.max_rows)
+    assert_outcome(env, case.expect, case.sql)
diff --git a/tests/e2e/test_safety_envelope.py b/tests/e2e/test_safety_envelope.py
index c9c32f9b..5007d529 100644
--- a/tests/e2e/test_safety_envelope.py
+++ b/tests/e2e/test_safety_envelope.py
@@ -1,17 +1,14 @@
 """End-to-end: execute_sql returns ONE response Envelope on BOTH surfaces (stdio + HTTP).
 
 A safety violation → status=refused + refusal{kind} + no data; a clean query → status=ok + data +
-audit_id. The full adversarial safety corpus + the read-only-DB-role test are out of scope here;
-this locks the cross-surface Envelope shape the shared contract promises — the same shape whether a
-client connects over stdio or HTTP.
+audit_id. The full adversarial safety corpus + the read-only-DB-role floor live in
+`test_safety_corpus.py` / `test_role_floor_pg.py`; this file locks the cross-surface Envelope shape
+the shared contract promises. The stdio + HTTP drivers (and `presence_auth`) are the shared harness —
+one copy — so this file and the corpus exercise the exact same two surfaces.
 """
 
 from __future__ import annotations
 
-import json
-import os
-import re
-import subprocess
 import sys
 from pathlib import Path
 
@@ -24,93 +21,11 @@
 if str(PKG_SRC) not in sys.path:
     sys.path.insert(0, str(PKG_SRC))
 
+import harness  # noqa: E402
 import tools  # noqa: E402
 
-BASE = "https://demo.example.com"
-
-
-class _FakeOkProc:
-    """A successful executor run: RFC-4180 CSV on stdout (header + one row), clean stderr."""
-
-    returncode = 0
-    stdout = "n\n5\n"
-    stderr = ""
-
-
-# --- transport drivers: each returns the execute_sql tool's Envelope (parsed) ----------------
-
-
-def _stdio_execute_sql(sql: str) -> dict:
-    """Drive execute_sql over the real stdio server (a subprocess), return the tool's Envelope."""
-    msgs = [
-        {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
-        {
-            "jsonrpc": "2.0",
-            "id": 2,
-            "method": "tools/call",
-            "params": {"name": "execute_sql", "arguments": {"sql": sql}},
-        },
-    ]
-    stdin = "".join(json.dumps(m) + "\n" for m in msgs)
-    proc = subprocess.run(
-        [sys.executable, "-m", "mcp_harness"],
-        input=stdin,
-        capture_output=True,
-        text=True,
-        timeout=30,
-        env={**os.environ},
-    )
-    by_id = {m.get("id"): m for m in (json.loads(x) for x in proc.stdout.splitlines() if x.strip())}
-    return json.loads(by_id[2]["result"]["content"][0]["text"])
-
-
-def _http_execute_sql(sql: str) -> dict:
-    """Drive execute_sql over the real HTTP transport (in-process TestClient), return the Envelope."""
-    import mcp_http
-    from starlette.testclient import TestClient
-
-    headers = {
-        "Authorization": "Bearer present",
-        "Content-Type": "application/json",
-        "Accept": "application/json, text/event-stream",
-    }
-    with TestClient(mcp_http.build_app()) as c:
-        init = c.post(
-            "/mcp",
-            headers=headers,
-            json={
-                "jsonrpc": "2.0",
-                "id": 1,
-                "method": "initialize",
-                "params": {
-                    "protocolVersion": "2025-06-18",
-                    "capabilities": {},
-                    "clientInfo": {"name": "t", "version": "1"},
-                },
-            },
-        )
-        sid = init.headers.get("mcp-session-id")
-        h2 = {**headers, **({"mcp-session-id": sid} if sid else {})}
-        c.post("/mcp", headers=h2, json={"jsonrpc": "2.0", "method": "notifications/initialized"})
-        r = c.post(
-            "/mcp",
-            headers=h2,
-            json={
-                "jsonrpc": "2.0",
-                "id": 2,
-                "method": "tools/call",
-                "params": {"name": "execute_sql", "arguments": {"sql": sql}},
-            },
-        )
-    rpc = json.loads(re.search(r"\{.*\}", r.text, re.DOTALL).group(0))
-    return json.loads(rpc["result"]["content"][0]["text"])
-
-
-@pytest.fixture
-def presence_auth(monkeypatch):
-    """HTTP bearer-presence mode: PUBLIC_BASE_URL set, no signing secret → 'Bearer present' works."""
-    monkeypatch.setenv("PUBLIC_BASE_URL", BASE)
-    monkeypatch.delenv("AGAMI_SIGNING_SECRET", raising=False)
+_stdio_execute_sql = harness.stdio_execute_sql
+_http_execute_sql = harness.http_execute_sql
 
 
 # --- the cross-surface contract ---------------------------------------------------------------
diff --git a/tests/safety/__init__.py b/tests/safety/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/safety/corpus.py b/tests/safety/corpus.py
new file mode 100644
index 00000000..32a46c7d
--- /dev/null
+++ b/tests/safety/corpus.py
@@ -0,0 +1,143 @@
+"""The F9 safety regression corpus (ACE-040) — the single, canonical source of the adversarial
+attack vectors and the demo schema they run against.
+
+One place defines: (a) the demo model/datasource schema (`SCHEMA`), from which the harness derives
+BOTH the semantic model (the `Organization` the guards scope against) AND the physical datasource
+(the SQLite/Postgres tables governed queries actually execute against), and (b) `CASES` — every
+attack class mapped to its expected `Envelope` outcome. The end-to-end corpus test parametrizes
+`CASES` across both surfaces (stdio + HTTP) and both model paths (file-served + DB-served); the
+outcome is asserted the same way regardless.
+
+`expect` is one of:
+  - a refusal `kind` string (e.g. "permission", "table_out_of_scope") — the query is refused;
+  - "bounded" — an availability control fired: EITHER a `resource_limit` refusal OR an ok Envelope
+    flagged `data.truncated` with a `row_cap` in `applied` (a runaway result was capped, never silent);
+  - "ok" — a governed query returns successfully with no false refusal.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+# ── the demo schema — single-sourced; the harness builds the model AND the datasource from this ──
+# table -> {"columns": [(name, sqlite_type), ...], "sensitive": {col, ...}, "rows": [tuple, ...]}
+SCHEMA: dict[str, dict] = {
+    "orders": {
+        "columns": [
+            ("id", "INTEGER"),
+            ("customer_id", "INTEGER"),
+            ("amount", "REAL"),
+            ("status", "TEXT"),
+        ],
+        "sensitive": set(),
+        "rows": [
+            (1, 10, 100.0, "paid"),
+            (2, 10, 50.0, "open"),
+            (3, 20, 75.0, "paid"),
+        ],
+    },
+    "customers": {
+        "columns": [("id", "INTEGER"), ("name", "TEXT"), ("email", "TEXT"), ("region", "TEXT")],
+        "sensitive": {"email"},  # PII: may be COUNTed/filtered/joined, never PROJECTED raw
+        "rows": [
+            (10, "Acme", "ops@acme.example", "west"),
+            (20, "Globex", "hi@globex.example", "east"),
+        ],
+    },
+}
+
+
+@dataclass(frozen=True)
+class Case:
+    cls: str  # attack-class label (test id + grouping)
+    sql: str
+    expect: str  # a refusal kind, or "bounded", or "ok"
+    note: str = ""
+    max_rows: int | None = None  # per-call row cap (the availability row-cap case lowers it)
+
+    @property
+    def id(self) -> str:
+        return f"{self.cls}:{self.note or self.sql[:40]}"
+
+
+CASES: list[Case] = [
+    # ── class 1: integrity / confinement → refused{permission} ──────────────────────────────────
+    Case("integrity", "DELETE FROM orders", "permission", "delete"),
+    Case("integrity", "UPDATE orders SET amount = 0", "permission", "update"),
+    Case("integrity", "DROP TABLE orders", "permission", "drop"),
+    Case("integrity", "INSERT INTO orders (id) VALUES (9)", "permission", "insert"),
+    Case("integrity", "SELECT 1; DROP TABLE orders", "permission", "multi-statement"),
+    Case("integrity", "SELECT pg_sleep(10)", "permission", "sleep-fn"),
+    Case("integrity", "SELECT id FROM orders FOR UPDATE", "permission", "row-lock"),
+    Case("integrity", "SELECT id INTO x FROM orders", "permission", "select-into"),
+    # ── class 2a: object scope — undeclared table → table_out_of_scope ──────────────────────────
+    Case("table_scope", "SELECT id FROM secret_table", "table_out_of_scope", "undeclared"),
+    Case(
+        "table_scope",
+        "SELECT o.id FROM orders o JOIN secret_table s ON s.id = o.id",
+        "table_out_of_scope",
+        "join",
+    ),
+    Case(
+        "table_scope",
+        "SELECT id FROM orders UNION SELECT id FROM secret_table",
+        "table_out_of_scope",
+        "set-op-arm",
+    ),
+    Case(
+        "table_scope",
+        "SELECT id FROM orders EXCEPT SELECT id FROM secret_table",
+        "table_out_of_scope",
+        "except-arm",
+    ),
+    # ── class 2b: SELECT * → select_star (incl. hidden in a set-op arm) ──────────────────────────
+    Case("select_star", "SELECT * FROM orders", "select_star", "star"),
+    Case("select_star", "SELECT o.* FROM orders o", "select_star", "qualified-star"),
+    Case(
+        "select_star",
+        "SELECT id FROM orders UNION SELECT * FROM customers",
+        "select_star",
+        "set-op-arm-star",
+    ),
+    # ── class 2c: undeclared column → column_out_of_scope ────────────────────────────────────────
+    Case("column_scope", "SELECT bogus FROM orders", "column_out_of_scope", "undeclared-col"),
+    Case(
+        "column_scope",
+        "SELECT id FROM orders UNION SELECT bogus FROM customers",
+        "column_out_of_scope",
+        "set-op-arm-col",
+    ),
+    # ── class 3: fail-closed scopability → unscopable_sql (under enforce, the default posture) ────
+    Case("unscopable", "SELECT x FROM (VALUES (1), (2)) AS v(x)", "unscopable_sql", "values"),
+    Case(
+        "unscopable", "SELECT g FROM generate_series(1, 10) AS t(g)", "unscopable_sql", "table-fn"
+    ),
+    Case(
+        "unscopable",
+        "SELECT o.id FROM orders o, (VALUES (1)) AS v(x)",
+        "unscopable_sql",
+        "comma-join-values",
+    ),
+    # ── class 4: recon / metadata deny-list → recon ─────────────────────────────────────────────
+    Case("recon", "SELECT version()", "recon", "version-fn"),
+    Case("recon", "SELECT current_user", "recon", "current-user"),
+    Case(
+        "recon", "SELECT table_name FROM information_schema.tables", "recon", "information_schema"
+    ),
+    Case("recon", "SELECT relname FROM pg_catalog.pg_class", "recon", "pg_catalog"),
+    # ── class 5: availability — a runaway result is bounded (row-cap truncate+flag OR timeout) ────
+    # A capacity query returning more rows than the (test-lowered) cap must come back flagged, never
+    # silently cut. Driven with a low AGAMI_SQL_MAX_ROWS by the harness. (A timeout variant is a
+    # separate slow-marked case; the row-cap path is the deterministic availability proof.)
+    Case("availability", "SELECT id FROM orders", "bounded", "row-cap-truncate", max_rows=1),
+    # ── class 7: governed queries still pass → ok (no false refusals) ────────────────────────────
+    Case("governed", "SELECT id, amount FROM orders", "ok", "projection"),
+    Case("governed", "SELECT status, COUNT(id) AS n FROM orders GROUP BY status", "ok", "group-by"),
+    Case(
+        "governed",
+        "SELECT c.name, COUNT(o.id) AS n FROM customers c JOIN orders o ON o.customer_id = c.id GROUP BY c.name",
+        "ok",
+        "join-aggregate",
+    ),
+    Case("governed", "SELECT COUNT(email) AS n FROM customers", "ok", "sensitive-in-count-ok"),
+]