feat(engine): CC-9 fail-open advisory handoff (M3 / Tier 2)

docs/HOOK_CONTRACT.md

@@ -25,9 +25,21 @@
 - `GRAPH_FLUSH_EVERY`
 - `CHAT_COMPRESSOR_FORWARD_BUDGET`
 - `CHAT_COMPRESSOR_INJECT_P1`
+- `CHAT_COMPRESSOR_ADVISORY_PATH` (optional CC-9 advisory JSON; default `advisory/latest.json` under state root)
 
 Unmanaged lines are preserved. `CURSOR_API_KEY` is never written.
 
 ## Merge policy
 
 When editing `hooks.json`, drop only entries whose `command` contains `chat-compressor`; keep all other hooks. Write via temp-file + rename; create a timestamped `.bak` on first modification of a session.
+
+## CC-9 advisory (fail-open)
+
+Optional file-based handoff from an external router (comPASS). When present and **fresh**, `_compose_additional_context` appends a single `COMPASS_ADVISORY:` line to `additional_context`.
+
+- Default path: `$CHAT_COMPRESSOR_STATE_DIR/advisory/latest.json` (override with `CHAT_COMPRESSOR_ADVISORY_PATH`)
+- Schema: `compass-advisory/v1` (see comPASS `docs/API.md`)
+- Include iff JSON parses, required fields are present, and `expires_at` is in the future
+- **Missing, stale, or corrupt advisory MUST NOT block Agent Chat** — hook still returns the event-safe default (`continue: true` / empty context)
+- Hook process never loads provider keys to read the advisory file
+

engine/env.example

@@ -44,3 +44,10 @@ PRODUCER_B_DIM=128
 # the inject ledger is partitioned per recipient (CC-2); hops reset suppression
 # and warmup (CC-3..CC-5). Absent recipient_id keeps session-scoped 0.2.0 behavior.
 # CHAT_COMPRESSOR_CROSS_TURN_DEDUP=1
+
+# CC-9: optional path to comPASS advisory JSON (compass-advisory/v1).
+# Absolute, or relative to CHAT_COMPRESSOR_STATE_DIR / STATE_DIR.
+# Default when unset: advisory/latest.json under the state root.
+# Missing/stale/corrupt files are ignored (fail-open; never blocks Agent Chat).
+# CHAT_COMPRESSOR_ADVISORY_PATH=advisory/latest.json
+

engine/src/chat_compressor/hook_cli.py

@@ -239,12 +239,99 @@ def _truncate_context(text: str, max_chars: int = _MAX_CONTEXT_CHARS) -> str:
     return text[: max_chars - 3].rstrip() + "..."
 
 
+
+def resolve_advisory_path(state_root: Path) -> Path:
+    """Path to compass-advisory/v1 JSON under the state root (CC-9).
+
+    Override with CHAT_COMPRESSOR_ADVISORY_PATH (absolute, or relative to state root).
+    Default: ``advisory/latest.json``.
+    """
+    load_hook_env()
+    raw = os.environ.get("CHAT_COMPRESSOR_ADVISORY_PATH", "").strip()
+    if raw:
+        candidate = Path(raw).expanduser()
+        if not candidate.is_absolute():
+            candidate = state_root / candidate
+        return candidate
+    return state_root / "advisory" / "latest.json"
+
+
+def _parse_advisory_expiry(value: Any) -> datetime | None:
+    if not isinstance(value, str) or not value.strip():
+        return None
+    text = value.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        dt = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def _advisory_required_ok(payload: Any) -> bool:
+    if not isinstance(payload, dict):
+        return False
+    if payload.get("schema") != "compass-advisory/v1":
+        return False
+    if not payload.get("written_at") or not payload.get("expires_at"):
+        return False
+    if not payload.get("task_class"):
+        return False
+    rec = payload.get("recommendation")
+    if not isinstance(rec, dict) or not rec.get("model_id"):
+        return False
+    return True
+
+
+def _format_advisory_line(payload: dict[str, Any]) -> str:
+    rec = payload.get("recommendation") or {}
+    model_id = rec.get("model_id") or "unknown"
+    task_class = payload.get("task_class") or "general"
+    rationale = (payload.get("rationale") or "").strip()
+    line = (
+        "COMPASS_ADVISORY: "
+        f"task_class={task_class}; recommended_model={model_id}"
+    )
+    if rationale:
+        # Keep a single-line hint; truncate aggressively so we stay under context cap.
+        one = " ".join(rationale.split())
+        if len(one) > 240:
+            one = one[:237].rstrip() + "..."
+        line += f"; {one}"
+    return line
+
+
+def _load_advisory_context_line(state_root: Path) -> str | None:
+    """Return advisory context line when fresh; else None. Never raises (CC-9 fail-open)."""
+    try:
+        path = resolve_advisory_path(state_root)
+        if not path.is_file():
+            return None
+        raw = path.read_text(encoding="utf-8")
+        payload = json.loads(raw)
+        if not _advisory_required_ok(payload):
+            return None
+        expires = _parse_advisory_expiry(payload.get("expires_at"))
+        if expires is None:
+            return None
+        now = datetime.now(timezone.utc)
+        if expires <= now:
+            return None
+        return _format_advisory_line(payload)
+    except Exception:  # noqa: BLE001 — corrupt/unreadable advisory must not block
+        return None
+
+
 def _compose_additional_context(
     handle: PersistentAgentHandle,
     sampled_text: str,
     agent_id: str,
     *,
     method: str = "",
+    state_root: Path | None = None,
 ) -> str:
     parts: list[str] = [
         "CHAT-COMPRESSOR session memory (prefer HOT_SET / typed / ranked FORWARD_GIST; do not restate full history).",
@@ -259,6 +346,10 @@ def _compose_additional_context(
     latest = handle.latest()
     if latest is not None:
         parts.append(f"STATE: agent_id={agent_id} t={latest.t} state_id={latest.state_id}")
+    if state_root is not None:
+        advisory_line = _load_advisory_context_line(state_root)
+        if advisory_line:
+            parts.append(advisory_line)
     return _truncate_context("\n\n".join(parts))
 
 
@@ -312,7 +403,7 @@ def handle_before_submit(
         rank_ms = float(sampled.rank_ms)
         rate = float(sampled.rate)
         budget = int(sampled.budget or budget)
-        context = _compose_additional_context(handle, sampled.text, agent_id, method=method)
+        context = _compose_additional_context(handle, sampled.text, agent_id, method=method, state_root=state_root)
         log_stage(
             state_root,
             _format_stage(
@@ -482,7 +573,7 @@ def handle_session_start(
     except Exception as exc:  # noqa: BLE001 — fail-open path
         error_class = type(exc).__name__
         log_error(state_root, f"sessionStart sample error={exc!r}\n{traceback.format_exc()}")
-    context = _compose_additional_context(handle, gist, agent_id, method=method)
+    context = _compose_additional_context(handle, gist, agent_id, method=method, state_root=state_root)
     log_stage(
         state_root,
         _format_stage(

engine/tests/test_cc9_advisory.py

@@ -0,0 +1,153 @@
+"""CC-9: fail-open advisory inclusion in additional_context."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from chat_compressor import hook_cli
+from chat_compressor.handle import PersistentAgentHandle
+from chat_compressor.producer import EmbeddingProducer
+from chat_compressor.store import StateStore
+
+
+def _iso(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _fresh_payload(*, expires_in: int = 300, model_id: str = "cursor-grok-test") -> dict:
+    now = datetime.now(timezone.utc)
+    return {
+        "schema": "compass-advisory/v1",
+        "written_at": _iso(now),
+        "expires_at": _iso(now + timedelta(seconds=expires_in)),
+        "task_class": "multi_file_refactor",
+        "recommendation": {
+            "model_id": model_id,
+            "provider": "cursor",
+            "model_version_id": "urn:mg:modelversion:test",
+        },
+        "rationale": "Across your last 40 tasks of this class, X scored 0.82 at $0.11/task.",
+        "route_decision_id": "urn:mg:routedecision:test",
+        "scores_summary": [
+            {"model_id": "X", "quality_mean": 0.82, "n": 40, "est_cost_per_task": 0.11}
+        ],
+    }
+
+
+@pytest.fixture()
+def state_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    root = tmp_path / "context-graphs"
+    root.mkdir()
+    (root / "logs").mkdir()
+    monkeypatch.setenv("CHAT_COMPRESSOR_STATE_DIR", str(root))
+    monkeypatch.setenv("K_MAX", "8")
+    monkeypatch.delenv("CHAT_COMPRESSOR_ADVISORY_PATH", raising=False)
+    monkeypatch.delenv("EMBED_MODEL_PATH", raising=False)
+    monkeypatch.delenv("GIST_MODEL_PATH", raising=False)
+    return root
+
+
+def _write_advisory(state_root: Path, payload: dict | str, *, name: str = "latest.json") -> Path:
+    dest = state_root / "advisory" / name
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    if isinstance(payload, str):
+        dest.write_text(payload, encoding="utf-8")
+    else:
+        dest.write_text(json.dumps(payload), encoding="utf-8")
+    return dest
+
+
+def _handle(state_root: Path) -> PersistentAgentHandle:
+    return PersistentAgentHandle(
+        agent_id="cc9-agent",
+        store=StateStore(state_root),
+        producer=EmbeddingProducer(d=64, k_max=8),
+        k_max=8,
+    )
+
+
+def test_fresh_advisory_appears_in_additional_context(state_root: Path) -> None:
+    _write_advisory(state_root, _fresh_payload(model_id="cursor-grok-fresh"))
+    handle = _handle(state_root)
+    handle.step("refactor the auth module across packages", role="user")
+    ctx = hook_cli._compose_additional_context(
+        handle, "packed-forward", "cc9-agent", method="pack", state_root=state_root
+    )
+    assert "COMPASS_ADVISORY:" in ctx
+    assert "recommended_model=cursor-grok-fresh" in ctx
+    assert "multi_file_refactor" in ctx
+
+
+def test_missing_advisory_fail_open(state_root: Path) -> None:
+    handle = _handle(state_root)
+    handle.step("hello world note", role="user")
+    ctx = hook_cli._compose_additional_context(
+        handle, "packed-forward", "cc9-agent", method="pack", state_root=state_root
+    )
+    assert "COMPASS_ADVISORY:" not in ctx
+    assert "CHAT-COMPRESSOR session memory" in ctx
+    # Event-safe beforeSubmit still continues.
+    out = hook_cli.process_payload(
+        {
+            "hook_event_name": "beforeSubmitPrompt",
+            "conversation_id": "cc9-missing",
+            "prompt": "add todo milk to the list",
+        },
+        event="beforeSubmitPrompt",
+    )
+    assert out.get("continue") is True
+
+
+def test_stale_advisory_ignored(state_root: Path) -> None:
+    payload = _fresh_payload(expires_in=-60)  # already expired
+    _write_advisory(state_root, payload)
+    assert hook_cli._load_advisory_context_line(state_root) is None
+    handle = _handle(state_root)
+    ctx = hook_cli._compose_additional_context(
+        handle, "gist", "cc9-agent", method="pack", state_root=state_root
+    )
+    assert "COMPASS_ADVISORY:" not in ctx
+    out = hook_cli.process_payload(
+        {
+            "conversation_id": "cc9-stale",
+            "prompt": "plan the grocery list with milk and bread",
+        },
+        event="beforeSubmitPrompt",
+    )
+    assert out.get("continue") is True
+
+
+def test_corrupt_advisory_ignored(state_root: Path) -> None:
+    _write_advisory(state_root, "{not-json")
+    assert hook_cli._load_advisory_context_line(state_root) is None
+    out = hook_cli.process_payload(
+        {
+            "conversation_id": "cc9-corrupt",
+            "prompt": "note that context graphs store StateNodes for sessions",
+        },
+        event="beforeSubmitPrompt",
+    )
+    assert out.get("continue") is True
+    ctx = out.get("additional_context") or ""
+    assert "COMPASS_ADVISORY:" not in ctx
+
+
+def test_malformed_schema_ignored(state_root: Path) -> None:
+    payload = _fresh_payload()
+    payload["schema"] = "other/v0"
+    _write_advisory(state_root, payload)
+    assert hook_cli._load_advisory_context_line(state_root) is None
+
+
+def test_advisory_path_env_override(state_root: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    custom = state_root / "custom" / "adv.json"
+    custom.parent.mkdir(parents=True)
+    custom.write_text(json.dumps(_fresh_payload(model_id="override-model")), encoding="utf-8")
+    monkeypatch.setenv("CHAT_COMPRESSOR_ADVISORY_PATH", str(custom))
+    line = hook_cli._load_advisory_context_line(state_root)
+    assert line is not None
+    assert "override-model" in line

test-results/cc-9-advisory/exit-evidence.json

@@ -0,0 +1,13 @@
+{
+  "cc": "CC-9",
+  "branch": "compass/cc9-advisory-handoff",
+  "negative_cases": [
+    "missing",
+    "stale_expires_at",
+    "corrupt_json"
+  ],
+  "positive_case": "fresh advisory line in additional_context",
+  "hook_continue_on_negatives": true,
+  "tests_file": "engine/tests/test_cc9_advisory.py",
+  "recorded_at": "2026-09-05T07:34:39Z"
+}

test-results/cc-9-advisory/fail-open-proof.md

@@ -0,0 +1,8 @@
+# CC-9 fail-open proof
+
+From `engine/tests/test_cc9_advisory.py`:
+
+1. **Missing file** — `_load_advisory_context_line` returns None; `beforeSubmitPrompt` still returns `continue: true`.
+2. **Stale `expires_at`** — ignored; no `COMPASS_ADVISORY:` line; `continue: true`.
+3. **Corrupt JSON** — ignored; no advisory line; `continue: true`.
+4. **Fresh file** — `COMPASS_ADVISORY:` appears in `additional_context`.