WS-29 H2: central tenant binding + routes/projects converted to tenant_session()

apps/services/gateway/gateway/db.py

@@ -20,17 +20,29 @@
 from __future__ import annotations
 
 from acb_common.db import (
+    TenantUnbound,
     async_database_url,
+    bind_tenant,
+    clear_tenant,
+    current_tenant,
     engine_connect_args,
     get_db,
     get_engine,
     get_session_factory,
+    release_tenant,
+    tenant_session,
 )
 
 __all__ = [
+    "TenantUnbound",
     "async_database_url",
+    "bind_tenant",
+    "clear_tenant",
+    "current_tenant",
     "engine_connect_args",
     "get_db",
     "get_engine",
     "get_session_factory",
+    "release_tenant",
+    "tenant_session",
 ]

apps/services/gateway/gateway/main.py

@@ -8,6 +8,7 @@
 from acb_auth import (UserContext, UserRole, get_current_user,
                       require_authenticated, require_role)
 from acb_common import configure_logging, get_logger, get_settings
+from acb_common.db import clear_tenant, release_tenant
 from fastapi import BackgroundTasks, Depends, FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import StreamingResponse
@@ -553,6 +554,31 @@ def docs_enabled(env: str) -> bool:
     openapi_url="/openapi.json" if _docs_enabled else None,
 )
 
+# ── Tenant scope (MT-1c / H2) ── every HTTP request runs inside its own tenant
+# scope: opened empty here, filled in by `_with_resolved_access` when the auth
+# dependency resolves the caller's organization, released after the response.
+# Pure ASGI (no BaseHTTPMiddleware) so the downstream app runs in THIS task and
+# the contextvar token round-trips — which is what makes "one request can never
+# inherit another's binding" true on any server task model, not just uvicorn's
+# task-per-request. Jobs and consumers get no scope from this; they bind
+# explicitly or fail closed with TenantUnbound (H4).
+class TenantScopeMiddleware:
+    def __init__(self, asgi_app):  # noqa: ANN001 — ASGI protocol shape
+        self.asgi_app = asgi_app
+
+    async def __call__(self, scope, receive, send):  # noqa: ANN001
+        if scope["type"] != "http":
+            await self.asgi_app(scope, receive, send)
+            return
+        token = clear_tenant()
+        try:
+            await self.asgi_app(scope, receive, send)
+        finally:
+            release_tenant(token)
+
+
+app.add_middleware(TenantScopeMiddleware)
+
 # ── CORS ── allow workbench dev server (port 3001) and production origin
 app.add_middleware(
     CORSMiddleware,

apps/services/gateway/gateway/routes/projects/activities.py

@@ -24,7 +24,7 @@
 from gateway.routes.projects.core import (
     ActivityModel,
     Page,
-    _get_db,
+    _tenant_session,
     actor,
     emit,
     from_jsonb,
@@ -85,8 +85,7 @@ async def get_timeline(
     never withheld — the history of what happened to a task is not editable by
     the people it happened to.
     """
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         vis = await resolve_visibility(db, user)
         await load_visible_task(db, vis, task_id)
         params = {
@@ -111,8 +110,6 @@ async def get_timeline(
             "rows": [row_to_dict(r, ActivityModel) for r in rows],
             "total": int(total),
         }
-    finally:
-        await db.close()
 
 
 @router.post("/tasks/{task_id}/comments", status_code=201)
@@ -123,8 +120,7 @@ async def add_comment(
     body = (payload.body or "").strip()
     if not body:
         raise HTTPException(status_code=422, detail="A comment needs a body.")
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         vis = await resolve_visibility(db, user)
         await load_visible_task(db, vis, task_id)
         row = await record_activity(
@@ -170,13 +166,10 @@ async def add_comment(
                 task_id=task_id,
                 meta={"mentioned": mention_result["notified"]},
             )
-        await db.commit()
         result = row_to_dict(row, ActivityModel)
         # Surfaced, not swallowed: a mention that reached nobody would
         # otherwise leave the author believing they pulled a colleague in.
         result["not_notified"] = mention_result["skipped"]
-    finally:
-        await db.close()
 
     await emit("pm.task.comment_added", {"task_id": task_id})
     return result
@@ -211,8 +204,7 @@ async def edit_comment(
     body = (payload.body or "").strip()
     if not body:
         raise HTTPException(status_code=422, detail="A comment needs a body.")
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         comment = await _load_own_comment(db, activity_id, user)
         # The task must still be visible: a member removed from a Center keeps
         # authorship of what they wrote but not access to it.
@@ -244,12 +236,9 @@ async def edit_comment(
                 task_id=task_id,
                 meta={"mentioned": mention_result["notified"]},
             )
-        await db.commit()
         result = row_to_dict(row, ActivityModel)
         result["not_notified"] = mention_result["skipped"]
         return result
-    finally:
-        await db.close()
 
 
 @router.delete("/comments/{activity_id}")
@@ -263,17 +252,13 @@ async def delete_comment(
     the row hidden, so "deleted" means the words are gone rather than merely
     filtered out of one read path.
     """
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         await _load_own_comment(db, activity_id, user)
         await update_row(
             db, "pm_activities", activity_id,
             {"deleted_at": now(), "body": None},
         )
-        await db.commit()
         return {"deleted": activity_id}
-    finally:
-        await db.close()
 
 
 def _restore_custom(task: Any, changes: list[dict]) -> dict | None:
@@ -317,8 +302,7 @@ async def revert_change(
     the timeline shows that a revert happened instead of history appearing never
     to have contained the change.
     """
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         row = (await db.execute(
             text(
                 "SELECT * FROM pm_activities WHERE id = CAST(:aid AS uuid) "
@@ -378,7 +362,6 @@ async def revert_change(
             ],
             extra_meta={"reverted_activity_id": activity_id},
         )
-        await db.commit()
         task_id = str(task.id)
         reverted = sorted(
             [k for k in restore if k != "custom_fields"]
@@ -387,8 +370,6 @@ async def revert_change(
                 if str(c.get("field") or "").startswith(_CUSTOM_PREFIX)
             ]
         )
-    finally:
-        await db.close()
 
     await emit("pm.task.updated", {"task_id": task_id})
     return {"task_id": task_id, "reverted": reverted, "skipped": skipped}

apps/services/gateway/gateway/routes/projects/admin.py

@@ -28,7 +28,7 @@
     STATUS_CATEGORIES,
     StatusModel,
     TypeModel,
-    _get_db,
+    _tenant_session,
     clean_payload,
     count_where,
     load_visible_project,
@@ -93,8 +93,7 @@ async def _clear_other_defaults(db: Any, table: str, root: str, keep: str) -> No
 async def list_statuses(
     project_id: str, user: UserContext = Depends(get_current_user),
 ) -> dict:
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         vis = await resolve_visibility(db, user)
         root = await _root_for(db, vis, project_id)
         rows = (await db.execute(
@@ -107,8 +106,6 @@ async def list_statuses(
         return {
             "rows": [row_to_dict(r, StatusModel) for r in rows], "total": len(rows),
         }
-    finally:
-        await db.close()
 
 
 @router.post("/nodes/{project_id}/statuses", status_code=201)
@@ -123,8 +120,7 @@ async def create_status(
     category = values.get("category") or "todo"
     validate_choice(category, STATUS_CATEGORIES, "status category")
 
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         vis = await resolve_visibility(db, user)
         root = await _root_for(db, vis, project_id)
         row = (await db.execute(
@@ -144,10 +140,7 @@ async def create_status(
         )).fetchone()
         if row.is_default:
             await _clear_other_defaults(db, "pm_task_statuses", root, str(row.id))
-        await db.commit()
         return row_to_dict(row, StatusModel)
-    finally:
-        await db.close()
 
 
 @router.patch("/statuses/{status_id}")
@@ -157,8 +150,7 @@ async def patch_status(
 ) -> dict:
     values = clean_payload(payload)
     validate_choice(values.get("category"), STATUS_CATEGORIES, "status category")
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         existing = await require_row(db, "pm_task_statuses", status_id, "Status")
         vis = await resolve_visibility(db, user)
         await load_visible_project(db, vis, str(existing.project_id))
@@ -169,10 +161,7 @@ async def patch_status(
             await _clear_other_defaults(
                 db, "pm_task_statuses", str(row.project_id), status_id,
             )
-        await db.commit()
         return row_to_dict(row, StatusModel)
-    finally:
-        await db.close()
 
 
 @router.delete("/statuses/{status_id}")
@@ -186,8 +175,7 @@ async def delete_status(
     that into a 409 naming how many tasks are in the way, which is the number
     the caller needs to decide what to do next.
     """
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         existing = await require_row(db, "pm_task_statuses", status_id, "Status")
         vis = await resolve_visibility(db, user)
         await load_visible_project(db, vis, str(existing.project_id))
@@ -204,10 +192,7 @@ async def delete_status(
             text("DELETE FROM pm_task_statuses WHERE id = CAST(:sid AS uuid)"),
             {"sid": status_id},
         )
-        await db.commit()
         return {"deleted": status_id, "tasks_affected": 0}
-    finally:
-        await db.close()
 
 
 # ── Types ───────────────────────────────────────────────────────────────────
@@ -216,8 +201,7 @@ async def delete_status(
 async def list_types(
     project_id: str, user: UserContext = Depends(get_current_user),
 ) -> dict:
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         vis = await resolve_visibility(db, user)
         root = await _root_for(db, vis, project_id)
         rows = (await db.execute(
@@ -230,8 +214,6 @@ async def list_types(
         return {
             "rows": [row_to_dict(r, TypeModel) for r in rows], "total": len(rows),
         }
-    finally:
-        await db.close()
 
 
 @router.post("/nodes/{project_id}/types", status_code=201)
@@ -244,8 +226,7 @@ async def create_type(
     if not name:
         raise HTTPException(status_code=422, detail="A task type needs a name.")
 
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         vis = await resolve_visibility(db, user)
         root = await _root_for(db, vis, project_id)
         row = (await db.execute(
@@ -266,10 +247,7 @@ async def create_type(
         # second root-only type — or, worse, a type that claims Epic's exemption.
         if row.is_default:
             await _clear_other_defaults(db, "pm_task_types", root, str(row.id))
-        await db.commit()
         return row_to_dict(row, TypeModel)
-    finally:
-        await db.close()
 
 
 @router.patch("/types/{type_id}")
@@ -278,8 +256,7 @@ async def patch_type(
     user: UserContext = Depends(get_current_user),
 ) -> dict:
     values = clean_payload(payload)
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         existing = await require_row(db, "pm_task_types", type_id, "Task type")
         vis = await resolve_visibility(db, user)
         await load_visible_project(db, vis, str(existing.project_id))
@@ -298,10 +275,7 @@ async def patch_type(
             await _clear_other_defaults(
                 db, "pm_task_types", str(row.project_id), type_id,
             )
-        await db.commit()
         return row_to_dict(row, TypeModel)
-    finally:
-        await db.close()
 
 
 @router.delete("/types/{type_id}")
@@ -315,8 +289,7 @@ async def delete_type(
     lane. The count is reported (R7/R8) because "12 tasks became untyped" is
     not something to discover from a board.
     """
-    db = await _get_db()
-    try:
+    async with _tenant_session() as db:
         existing = await require_row(db, "pm_task_types", type_id, "Task type")
         vis = await resolve_visibility(db, user)
         await load_visible_project(db, vis, str(existing.project_id))
@@ -330,7 +303,4 @@ async def delete_type(
             text("DELETE FROM pm_task_types WHERE id = CAST(:tid AS uuid)"),
             {"tid": type_id},
         )
-        await db.commit()
         return {"deleted": type_id, "tasks_untyped": affected}
-    finally:
-        await db.close()

apps/services/gateway/gateway/routes/projects/agent_dispatch.py

@@ -29,7 +29,20 @@
 from typing import Any
 
 from acb_common import get_logger
-from gateway.routes.projects.core import _get_db, record_activity
+
+# ⚠️ H4, DELIBERATELY NOT H2 (`saas_multitenancy_handover.md`): this module is
+# an EVENT CONSUMER, not a request handler — `on_event` fires from
+# `emit_event`'s sink fan-out and `_run_and_record` outlives the request that
+# scheduled it. The runbook's rule for that category is "do not let a job
+# inherit an ambient tenant", so converting these sites to the ambient
+# `tenant_session()` would be exactly the inheritance it forbids. They stay on
+# the unbound `get_db()` until H4 threads an EXPLICIT tenant through the event
+# payload (`tenant_session(org_id)`), and the H2 ratchet in
+# `test_db_engine_seam.py` carries this module as its one named exemption.
+# Sequencing is safe: RLS phase 4 (which would starve these reads) is gated on
+# H2+H4 both being complete.
+from gateway.db import get_db as _get_db
+from gateway.routes.projects.core import record_activity
 from sqlalchemy import text
 
 _log = get_logger("projects.agent_dispatch")