WS-27: one design language for /projects and /tasks, the themed categorical ramp, the house shell, and the last two Projects tenancy residues

CLAUDE.md

@@ -89,8 +89,19 @@ Rules that make it work:
   numbers). Specs go stale; the code is the fact.
 - **Respect the seams.** Extend the shared seam, never add a parallel one: one
   DB engine (`acb_common.db`), one entitlement intersect, one subject-grammar
-  validator, one task store, one Center registry (`lib/centers.ts`). A second
-  implementation of an existing seam is a defect, not a feature.
+  validator, one task store, one Center registry (`lib/centers.ts`), one status
+  colour vocabulary (`src/lib/statusAccent.ts`). A second implementation of an
+  existing seam is a defect, not a feature.
+- **The UI is one product, themed centrally.** Every app is a projection, never
+  a surface with its own look: no app-local palette, no second colour
+  vocabulary, no hand-rolled control. `workbench/control_plane/DESIGN_SYSTEM.md`
+  is the contract and `AGENTS.md` beside it carries the seven rules and their
+  fences — both are auto-loaded when you touch UI code. Owner directive
+  2026-08-10. Categorical hues (contexts, tags, labels) go through the
+  `--cat-1…8` ramp via `src/lib/categorical.ts`, never a raw Tailwind palette
+  class. The conformance suite checks five regexes and **nothing tests layout
+  or cross-app continuity**, so the theme-switch check (Fluent → Material →
+  Graphite, on your surface *and* its neighbour) is the real gate.
 - **Keep branches short and integrate often.** Long branches are the root cause
   behind the migration-renumber collisions, the green-alone/red-together PRs and
   a duplicated tenancy design. Three or four in flight is the ceiling.

apps/services/gateway/AGENTS.md

@@ -175,6 +175,22 @@ breaking it — **plus R11, which will bite here hardest when WS-29 lands:**
 > directory runs against a tenant-bound session, and the binding happens **once**
 > in `acb_common.db` — so a route that reaches data any other way is the bug.
 > Build shapes: `saas_multitenancy_implementation.md` §2.
+>
+> ⚠️ **A background unit in this directory is NOT a route and must not behave
+> like one.** It has no request to inherit from, so `tenant_session()` with no
+> argument is the bug there — it would act for whoever happened to be in
+> context. The rule (H4) is **resolve on an unbound session, then bind
+> explicitly, and refuse if the resolution finds nothing**: two sessions in that
+> order, the first deciding the tenant and writing nothing. Three built
+> examples, three stored facts: `routes/crm/auto_lead` (the mailbox owner's
+> org), `routes/workflows/service._pm_lifecycle_sweeper` (the workflow owner's
+> org — `workflows` has no `organization_id` column until H3 phase 1) and
+> `routes/projects/agent_dispatch` (the task's own org, carried **on the event
+> payload**, because an event consumer has nowhere legitimate to look one up —
+> the emitter stamps it inside its bound session). A consumer that refuses
+> cannot record its refusal in tenant data either: it logs and returns.
+> `tests/unit/test_db_engine_seam.py` is the fence — `H2_EXEMPT_FILES`,
+> `H2_BASELINE_ELSEWHERE` and the exempt→converted call-edge table.
 
 1. **The app is default-deny; do not opt out to make something reachable.**
    `require_authenticated` is attached at the app level in `main.py`, so a new

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

@@ -21,6 +21,20 @@
 **Only NEW assignees dispatch.** `set_assignees` emits the *added* set, not the
 whole set, so re-asserting an existing assignee cannot start a second run. That
 property lives in the emitter and is relied on here; both sides say so.
+
+**The tenant rides the event, and nothing else will do (WS-27aa / H4).** This
+module was the H2 ratchet's one Projects exemption, on the grounds that an
+event consumer must not inherit the ambient request tenant. That was right, and
+the way out was never "convert it" — it was to give it an EXPLICIT one.
+`set_assignees` now reads the task's own `organization_id` inside the request's
+bound session and puts it on the payload; every session opened here is
+`_tenant_session(org)` with that value. A payload without one is **refused**:
+no run is dispatched and nothing is written, because writing the refusal would
+itself need the unbound session this rule exists to forbid — the refusal is
+therefore a WARNING log line (`projects.agent_dispatch_refused`), not a
+timeline row. Fenced by `tests/unit/test_projects_automation.py`'s
+`test_an_event_without_a_tenant_refuses_*` and by
+`tests/unit/test_db_engine_seam.py`, which no longer exempts this file.
 """
 
 from __future__ import annotations
@@ -30,18 +44,15 @@
 
 from acb_common import get_logger
 
-# ⚠️ 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
+# ⚠️ H4 DONE (WS-27aa), which is why this is `tenant_session` and NOT the
+# ambient no-argument form. 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 — so
+# `tenant_session()` with no argument would inherit whatever tenant happened to
+# be in context, which is precisely what the runbook forbids. Every call here
+# passes the organization the EVENT carried, and there is no path through this
+# module that opens a session without one.
+from gateway.db import tenant_session as _tenant_session
 from gateway.routes.projects.core import record_activity
 from sqlalchemy import text
 
@@ -94,13 +105,35 @@ def build_message(task: Any) -> str:
     return "\n\n".join(parts)
 
 
+def event_tenant(payload: dict[str, Any]) -> str:
+    """The organization the event says it belongs to, or ``""``.
+
+    Exported and tiny so the refusal has ONE definition that both `on_event`
+    and its tests read. The value is a stored fact about the task, stamped by
+    the emitter inside the request's bound session — never a field a caller
+    chose, which is R11 restated for an event payload.
+    """
+    return str(payload.get("organization_id") or "").strip()
+
+
 async def on_event(source: str, event_type: str, payload: dict[str, Any]) -> None:
     """Event sink: start a run for every agent newly assigned to a task.
 
     Registered alongside the workflows dispatcher, so `pm.task.assigned` fans
     out to both. Best-effort like every sink — `emit_event` swallows sink
     errors by default and that default is load-bearing (a webhook must never
     5xx because a sink failed), so this returns rather than raises.
+
+    ⚠️ **Refuses without a tenant on the payload** (WS-27aa / H4). Not
+    "falls back to the ambient one", not "looks it up unbound" — both are the
+    unbounded-leak shape the runbook names. The refusal is a WARNING log line
+    rather than a timeline row, and that asymmetry is deliberate: the timeline
+    lives in `pm_activities`, which is tenant data, so recording the refusal
+    there would need exactly the unbound session being refused. Under RLS
+    phase 4 such a write lands nowhere anyway. `set_assignees` always stamps
+    the field (`pm_tasks.organization_id` is NOT NULL since migration 161), so
+    in practice this fires only for a foreign or replayed emitter — which is
+    the case worth being loud about.
     """
     if source != "projects" or event_type != "pm.task.assigned":
         return
@@ -112,8 +145,17 @@ async def on_event(source: str, event_type: str, payload: dict[str, Any]) -> Non
     if not task_id:
         return
 
-    db = await _get_db()
-    try:
+    organization_id = event_tenant(payload)
+    if not organization_id:
+        _log.warning(
+            "projects.agent_dispatch_refused", task_id=task_id,
+            agents=agents,
+            reason="event carries no organization_id — a sink must not "
+                   "inherit an ambient tenant or resolve one unbound (H4)",
+        )
+        return
+
+    async with _tenant_session(organization_id) as db:
         task = (await db.execute(
             text("SELECT * FROM pm_tasks WHERE id = CAST(:tid AS uuid)"),
             {"tid": task_id},
@@ -123,32 +165,40 @@ async def on_event(source: str, event_type: str, payload: dict[str, Any]) -> Non
         message = build_message(task)
         for name in agents:
             # The timeline entry is written and COMMITTED before the run
-            # starts, so the handoff is visible immediately rather than when
-            # the agent finishes.
+            # starts — the `_tenant_session` block commits on exit, which is
+            # BEFORE the dispatch loop below — so the handoff is visible
+            # immediately rather than when the agent finishes.
             await record_activity(
                 db, activity_type="agent_run", created_by=f"{AGENT_PREFIX}{name}",
                 task_id=task_id, body=f"Assigned to {name}; starting a run.",
                 meta={"agent": name, "state": "started"},
             )
-        await db.commit()
-    finally:
-        await db.close()
 
     for name in agents:
-        await _run_and_record(name, message, task_id)
+        await _run_and_record(name, message, task_id, organization_id)
 
 
-async def _run_and_record(agent: str, message: str, task_id: str) -> None:
+async def _run_and_record(
+    agent: str, message: str, task_id: str, organization_id: str,
+) -> None:
     """Run one agent and close its timeline entry either way.
 
     A dispatch that fails silently is worse than one that never started: the
     task shows a session that appears to still be running and nobody knows to
     pick the work back up. So the failure path writes too.
+
+    ``organization_id`` is threaded down rather than re-read: this coroutine
+    outlives the transaction that started it, so there is nothing left to read
+    it from, and re-resolving it would be a second answer to a question the
+    event already settled.
     """
     try:
         from orchestrator.executor import run_agent
     except Exception as exc:  # pragma: no cover — orchestrator is a hard dep
-        await _record_outcome(task_id, agent, ok=False, detail="orchestrator unavailable")
+        await _record_outcome(
+            task_id, agent, organization_id,
+            ok=False, detail="orchestrator unavailable",
+        )
         _log.warning("projects.agent_dispatch_unavailable", error=str(exc))
         return
 
@@ -157,25 +207,32 @@ async def _run_and_record(agent: str, message: str, task_id: str) -> None:
             run_agent(agent, message), timeout=AGENT_RUN_TIMEOUT_SECONDS,
         )
     except TimeoutError:
-        await _record_outcome(task_id, agent, ok=False, detail="timed out")
+        await _record_outcome(
+            task_id, agent, organization_id, ok=False, detail="timed out",
+        )
         return
     except Exception as exc:
-        await _record_outcome(task_id, agent, ok=False, detail=str(exc)[:300])
+        await _record_outcome(
+            task_id, agent, organization_id, ok=False, detail=str(exc)[:300],
+        )
         return
-    await _record_outcome(task_id, agent, ok=True, detail=str(result or "")[:2000])
+    await _record_outcome(
+        task_id, agent, organization_id,
+        ok=True, detail=str(result or "")[:2000],
+    )
 
 
-async def _record_outcome(task_id: str, agent: str, *, ok: bool, detail: str) -> None:
-    db = await _get_db()
+async def _record_outcome(
+    task_id: str, agent: str, organization_id: str, *, ok: bool, detail: str,
+) -> None:
     try:
-        await record_activity(
-            db, activity_type="agent_run", created_by=f"{AGENT_PREFIX}{agent}",
-            task_id=task_id,
-            body=detail if ok else f"Agent run failed: {detail}",
-            meta={"agent": agent, "state": "finished" if ok else "failed"},
-        )
-        await db.commit()
+        async with _tenant_session(organization_id) as db:
+            await record_activity(
+                db, activity_type="agent_run",
+                created_by=f"{AGENT_PREFIX}{agent}",
+                task_id=task_id,
+                body=detail if ok else f"Agent run failed: {detail}",
+                meta={"agent": agent, "state": "finished" if ok else "failed"},
+            )
     except Exception as exc:  # pragma: no cover — the outcome write is best-effort
         _log.warning("projects.agent_dispatch_record_failed", error=str(exc))
-    finally:
-        await db.close()

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

@@ -40,6 +40,11 @@
 from typing import Any
 from zoneinfo import ZoneInfo
 
+# WS-27aa / H4: the sweep refuses to run without an explicit tenant, and it
+# refuses with the seam's OWN exception rather than a bespoke one — `db.py`
+# already defines "no tenant, never defaulted", and a second exception type
+# would be a second vocabulary for the same refusal.
+from gateway.db import TenantUnbound
 from gateway.routes.projects.core import (
     CLOSING_CATEGORIES,
     TRIAGE_CATEGORY,
@@ -291,12 +296,25 @@ async def _sweep_candidates(
 
 
 async def run_lifecycle_sweep(
-    db: Any, *, actor: str, now: datetime | None = None,
+    db: Any, *, organization_id: str, actor: str, now: datetime | None = None,
 ) -> dict[str, Any]:
-    """Apply every root project's lifecycle policy once. Idempotent.
-
-    For each ROOT project with a policy enabled (both columns NULL — the
-    default — means the project is never touched):
+    """Apply ONE tenant's root-project lifecycle policies once. Idempotent.
+
+    ⚠️ **``organization_id`` is required and is never defaulted** (WS-27aa,
+    H4). This function is reached only from a scheduled ``/workflows`` node, so
+    it has no request to inherit a tenant from — and H4's rule for that
+    category is the whole point: *a job that forgets doesn't leak one row, it
+    leaks unbounded*. Before this argument existed the roots query was
+    ``SELECT * FROM pm_projects WHERE parent_project_id IS NULL`` with no
+    predicate at all, which is one schedule archiving and closing **every
+    customer's** work. A blank tenant raises :class:`TenantUnbound` rather than
+    sweeping wider; the caller
+    (``routes/workflows/service._pm_lifecycle_sweeper``) resolves it from the
+    workflow owner's ``app_user`` row — a stored fact, never request input
+    (R11) — and binds it on the session it hands in.
+
+    For each ROOT project **in that organization** with a policy enabled (both
+    columns NULL — the default — means the project is never touched):
 
     * **archive** — tasks whose status category is done/cancelled and whose
       ``updated_at`` is older than ``archive_after_months`` calendar months
@@ -320,9 +338,27 @@ async def run_lifecycle_sweep(
     ``system:workflow:<id>`` identity), so every change is an ordinary
     timeline row wearing the automation flag.
     """
+    org = str(organization_id or "").strip()
+    if not org:
+        raise TenantUnbound(
+            "run_lifecycle_sweep needs an explicit organization_id — a "
+            "scheduled sweep must never inherit an ambient tenant or default "
+            "to every tenant (saas_multitenancy_handover.md H4)"
+        )
     moment = now or _clock()
+    # The tenant predicate is on the ROOTS query and nowhere else, on purpose:
+    # everything below reaches its rows through `project.id`, so one fence at
+    # the top of the walk bounds the whole sweep. Under RLS phase 4 the bound
+    # session narrows it a second time — belt and braces, in the order H3's
+    # cliff needs (the predicate has to be right BEFORE the policy exists,
+    # because today there is no policy to catch a missing one).
     roots = (await db.execute(
-        text("SELECT * FROM pm_projects WHERE parent_project_id IS NULL"),
+        text(
+            "SELECT * FROM pm_projects "
+            "WHERE parent_project_id IS NULL "
+            "AND organization_id = CAST(:org AS uuid)"
+        ),
+        {"org": org},
     )).fetchall()
 
     swept = archived = closed = 0

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

@@ -622,11 +622,23 @@ async def set_assignees(
 
     The emitted ``pm.task.assigned`` event is what WS-27f keys agent dispatch
     off, so it carries the added assignees rather than the whole set: a
-    re-assert of an existing assignee must not re-dispatch a run.
+    re-assert of an existing assignee must not re-dispatch a run. It also
+    carries the task's ``organization_id`` (WS-27aa): the sink runs with no
+    request behind it, so this is the one place its tenant can come from a
+    stored fact instead of an ambient binding.
     """
     async with _tenant_session() as db:
         vis = await resolve_visibility(db, user)
-        await load_visible_task(db, vis, task_id)
+        task = await load_visible_task(db, vis, task_id)
+        # WS-27aa / H4 — the tenant the dispatch sink will bind, read HERE:
+        # inside the request's already-bound session, off the task's own
+        # `organization_id` (NOT NULL since migration 161). It is a stored
+        # fact about the row, never anything the caller sent (R11), and it is
+        # read here because this is the last place a tenant legitimately
+        # exists: `agent_dispatch.on_event` fires with no request behind it,
+        # and a sink that looked the tenant up itself would have to do so on
+        # an unbound session — the exact thing H4 forbids.
+        task_org = str(getattr(task, "organization_id", "") or "")
 
         wanted = {
             a.strip().lower() for a in payload.assignees if (a or "").strip()
@@ -690,6 +702,9 @@ async def set_assignees(
     if added:
         await emit("pm.task.assigned", {
             "task_id": task_id, "assignees": sorted(added),
+            # The sink's ONLY tenant source. `agent_dispatch.on_event` refuses
+            # a payload without it rather than running unbound.
+            "organization_id": task_org,
         })
     return {
         "task_id": task_id,

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

@@ -91,10 +91,23 @@ def _refuse_lifecycle_on_child(values: dict, parent_project_id: object) -> None:
 #: Seeded on every ROOT project. The owner reshapes these in the app; they exist
 #: so a new project has a working board on its first render rather than an empty
 #: status picker.
+#:
+#: ⚠️ **The colours here must equal what `CATEGORY_HUES` derives from the
+#: category** (`workbench/control_plane/src/lib/statusAccent.ts`), because a
+#: stored colour OUTRANKS the category — that is what lets an owner choose. So a
+#: seed that disagrees is a seed that silently overrides the shared vocabulary on
+#: every project nobody has customised, and /projects goes back to looking
+#: different from /tasks. It did: this tuple used to seed `To do` blue and
+#: `In progress` amber against a category map of gray and blue, and two of the
+#: four default lanes rendered differently in the two apps.
+#:
+#: These are defaults, not decisions. If the shared vocabulary changes, change
+#: them here too; `test_seed_status_colours_match_the_shared_vocabulary` fails
+#: until you do.
 _SEED_STATUSES: tuple[tuple[str, str, int, str, bool], ...] = (
     ("Backlog",     "gray",   10, "backlog",     True),
-    ("To do",       "blue",   20, "todo",        False),
-    ("In progress", "amber",  30, "in_progress", False),
+    ("To do",       "gray",   20, "todo",        False),
+    ("In progress", "blue",   30, "in_progress", False),
     ("Done",        "green",  40, "done",        False),
 )