P4.2: keep review reference rows out of Main when a run has no room

Plan-review and task-acceptance reference rows fell back to Main three ways: an
unbound context, an unroutable candidate, and a normalizer failure all wrote
chat 1. A run with no owner-visible room therefore published its review rows
into the owner's conversation, and a task bound to a project after admission
kept writing them to the chat it was born in (live task 8cae80bb wrote four
rows at chat 1 while its binding named another chat).

_emit_review_reference now resolves the task's durable project binding at
emission through the same resolve_project_chat seam the log-addressing rail
uses, routes binding first and caller chat second, and lands in the hidden
partition when neither answers. Both rails carry the same address, so the
durable progress row and the live invalidation agree without a second reader.
This is a DOCTRINE change, not a default change: review references live in the
hidden partition, never in Main. The documented Main exception for project
question pointers is untouched.

No backfill: rows written before a bind are re-classified on read by the
history thread filter. The acceptance-bookkeeping expectation moves with the
seam - a bound task's durable row now carries its project chat.

Issue I30b; owner decision batch 3 answer 6a=A (review rows without a room go
to the hidden partition); disposition R28.

ouroboros/tools/plan_review_references.py

@@ -8,6 +8,7 @@ import pathlib
 from hashlib import sha256
 from typing import Any, Optional
 
+from ouroboros.contracts.chat_id_policy import HIDDEN_CHAT_ID
 from ouroboros.task_results import (
     load_plan_review_state,
     mark_plan_review_cycles_exhausted,
@@ -47,20 +48,36 @@ def _emit_review_reference(
     ctx: Any, task_id: str, state: Any, *, surface: str,
     state_root: Optional[pathlib.Path] = None, fingerprint: str = "", chat_id: Any = None,
 ) -> None:
-    """Invalidate the existing task-detail read model after its durable write."""
+    """Invalidate the existing task-detail read model after its durable write.
+
+    The row is addressed like every other task-scoped notice: the task's DURABLE
+    project binding first, then the chat the caller named, and the hidden
+    partition when neither answers. Main is never a default — a review row for a
+    run that has no owner-visible room belongs to the hidden partition, not to
+    the owner's conversation (`chat_id_policy`, owner decision 6a=A).
+    """
     event_queue = getattr(ctx, "event_queue", None)
     serialized = json.dumps(state, ensure_ascii=False, sort_keys=True, default=str)
     revision = sha256(serialized.encode("utf-8")).hexdigest()
     try:
+        from supervisor.log_addressing import resolve_project_chat
         from supervisor.message_bus import notification_chat_route
 
+        metadata = getattr(ctx, "task_metadata", None) or {}
+        bound = resolve_project_chat(
+            getattr(ctx, "budget_drive_root", "") or getattr(ctx, "drive_root", ""),
+            task_id,
+            metadata.get("parent_task_id"),
+            metadata.get("root_task_id"),
+        )
         chat_id = notification_chat_route(
-            chat_id if chat_id is not None else getattr(ctx, "current_chat_id", None), 1,
+            bound or None,
+            chat_id if chat_id is not None else getattr(ctx, "current_chat_id", None),
         )
         if chat_id is None:
-            chat_id = 1
+            chat_id = HIDDEN_CHAT_ID
     except (TypeError, ValueError):
-        chat_id = 1
+        chat_id = HIDDEN_CHAT_ID
     ts = utc_now_iso()
     payload = {
         "type": "review_reference", "surface": surface,

tests/test_acceptance_bookkeeping.py

@@ -92,7 +92,10 @@ def test_terminal_pipeline_routes_review_and_downloads_without_bookkeeping_deliv
         assert effective["artifact_bundle"]["status"] == expected_artifacts
 
     reference = next(row for row in captured if row.get("type") == "review_reference")
-    assert reference["chat_id"] == chat_id  # Env intentionally carries no current_chat_id.
+    # Env intentionally carries no current_chat_id. The binding is resolved AT
+    # EMISSION, so a BOUND task's durable row already carries its project chat
+    # instead of the chat the task was born in; an unbound task keeps that chat.
+    assert reference["chat_id"] == (project_chat if project_bound else chat_id)
     assert reference["type"] not in WORKER_LOG_SINK_SUPPRESSED_TYPES
     live = []
     supervisor = SimpleNamespace(RUNNING={"applied": {"task": task}}, DRIVE_ROOT=tmp_path,

tests/test_plan_review_references.py

@@ -55,7 +55,10 @@ def test_plan_reference_uses_shared_best_effort_log_event_seam(monkeypatch):
     }
 
 
-def test_plan_reference_defaults_unbound_context_to_main_chat(monkeypatch):
+def test_review_references_without_a_room_live_in_the_hidden_partition(monkeypatch):
+    """Doctrine, not a default: a run with no owner-visible room keeps its
+    review rows in the hidden partition. Main was a hard fallback, which put
+    plan-review rows of unrelated runs into the owner's conversation."""
     events: queue.Queue = queue.Queue()
     ctx = SimpleNamespace(event_queue=events)
     state = {"current_attempt": {"fingerprint": "review-fingerprint"}, "waves": []}
@@ -68,7 +71,7 @@ def test_plan_reference_defaults_unbound_context_to_main_chat(monkeypatch):
     )
     plan_review_references._emit_plan_review_reference(ctx, "task-1", state)
 
-    assert calls[0]["chat_id"] == 1
+    assert calls[0]["chat_id"] == plan_review_references.HIDDEN_CHAT_ID == 0
 
 
 def test_plan_reference_preserves_explicit_panel_chat_zero(monkeypatch):
@@ -87,6 +90,51 @@ def test_plan_reference_preserves_explicit_panel_chat_zero(monkeypatch):
     assert calls[0]["chat_id"] == 0
 
 
+def test_review_reference_addresses_the_bound_project_chat(tmp_path):
+    """The binding outranks the context chat on BOTH rails: a task turned into a
+    project mid-run keeps its origin chat on ctx, so a row addressed from ctx
+    alone lands outside the room that holds the work."""
+    from ouroboros.projects_registry import bind_task_to_project
+
+    binding = bind_task_to_project(
+        tmp_path, "task-bound", "review-ref-proj", 7373, origin={"absent": "system"},
+    )
+    task_results.write_task_result(tmp_path, "task-bound", "running", result="running")
+    events: queue.Queue = queue.Queue()
+    ctx = SimpleNamespace(event_queue=events, current_chat_id=1, drive_root=tmp_path)
+
+    plan_review_references._record_plan_review_attempt_with_reference(
+        ctx, tmp_path, "task-bound", fingerprint="a" * 64, status="open",
+    )
+
+    rows = [
+        json.loads(line)
+        for line in (tmp_path / "logs" / "progress.jsonl").read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+    assert binding["project_chat_id"] == 7373
+    assert rows[0]["chat_id"] == 7373
+    assert _reference_rows(events)[0]["chat_id"] == 7373
+
+
+def test_review_reference_still_emits_when_the_bindings_read_fails(tmp_path, monkeypatch):
+    """A broken bindings store must not cost the invalidation: the row is still
+    emitted, addressed to the hidden partition instead of guessing Main."""
+    import supervisor.log_addressing as log_addressing
+
+    def fail_resolve(*_args, **_kwargs):
+        raise TypeError("bindings unreadable")
+
+    monkeypatch.setattr(log_addressing, "resolve_project_chat", fail_resolve)
+    events: queue.Queue = queue.Queue()
+    ctx = SimpleNamespace(event_queue=events, current_chat_id=23, drive_root=tmp_path)
+    state = {"current_attempt": {"fingerprint": "review-fingerprint"}, "waves": []}
+
+    plan_review_references._emit_plan_review_reference(ctx, "task-1", state)
+
+    assert _reference_rows(events)[0]["chat_id"] == plan_review_references.HIDDEN_CHAT_ID
+
+
 def test_attempt_helper_publishes_immediately_after_the_canonical_write(monkeypatch):
     ctx = SimpleNamespace(event_queue=queue.Queue())
     calls = []