feat(inbox): C2.5 — inbox hardening + SDK version guard (closes C-F7, C-N3, R-6)

- File-layer decision vocabulary (§4.2 set) with SKIP semantics — an unknown
  decision never reaches the store (C-F7, the review's run proof is the fixture)
- Fail-fast caps (max_files / max_rationale_chars) via InboxLimitError raised
  OUTSIDE the tolerant try — a cap breach is never swallowed as a skip
- R-6 id grammar (mirrors ingest _ID_RE) as a pydantic pattern on
  VerdictDocument.id AND re-checked in write_verdict, since model_copy(update=)
  bypasses model validation — traversal ids can no longer write outside the inbox
- promotion._filename_token: any sanitised id maps to a content hash — 'e/vil'
  can no longer clobber the distinct id 'evil' (restarbeid-funn 2)
- SDK pinned >=0.2.111,<0.3 + version guard test naming the sdk_client.py
  attribute premises; resolved 0.2.120, all premises re-verified against it
- sdk_client read loop bound offline with REAL SDK message types (R-4/R-5):
  text aggregation, error fail-paths, usage/cost extraction, _total_tokens
  fail-closed, non-positive budget guard
- test_sdk_isolation comment no longer claims the --system-prompt ""
  serialization the test body does not bind (honesty rule §1)

Guard-G2 assessment (guard-plan §4): the allowlist + caps + id grammar landed
here are G2's necessary part; an optional scan_output depth pass over
rationale (still a verbatim prose channel into the fold prompt, R-9) remains
relevant as a later additive session — the trigger picture is unchanged.

4 detach proofs red → restored green. Full gate: 389 passed (365→389),
ruff+format+mypy clean; golden + shared/ + runs/s10/ byte-untouched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_sdk_isolation.py

@@ -17,10 +17,11 @@ from __future__ import annotations
 from typing import Any, AsyncIterator
 
 import pytest
+from claude_agent_sdk import AssistantMessage, ResultMessage, TextBlock, ThinkingBlock
 
 from portfolio_optimiser_claude import sdk_client
 from portfolio_optimiser_claude.contracts import ModelMapContract
-from portfolio_optimiser_claude.sdk_client import SdkModelClient, build_call_options
+from portfolio_optimiser_claude.sdk_client import SdkModelClient, _total_tokens, build_call_options
 
 
 class TestBuildCallOptions:
@@ -32,8 +33,11 @@ class TestBuildCallOptions:
         assert options.setting_sources == []
 
     def test_the_system_prompt_is_empty(self) -> None:
-        # None serializes to --system-prompt "" (verified against 0.2.110):
-        # no Claude Code preset, no appended operator instructions.
+        # Pins the OPTION value: None, not the Claude Code preset. That None
+        # reaches the spawned CLI as --system-prompt "" was verified by
+        # READING subprocess_cli.py (0.2.110–0.2.120) — this test does NOT
+        # bind that transport serialization; doing so would couple the suite
+        # to SDK-private API (the F11 fragility this repo retired).
         options = build_call_options("model-x", max_budget_usd=0.25)
         assert options.system_prompt is None
 
@@ -75,3 +79,148 @@ class TestCompleteThreadsIsolatedOptions:
         assert captured["options"].model == "model-default"
         # No usage surfaced by the fake → the reply fails CLOSED (§8).
         assert reply.usage_tokens is None
+
+
+def _stream_of(*messages: Any) -> Any:
+    """A fake ``query`` yielding a scripted stream of REAL SDK message objects."""
+
+    def fake_query(*, prompt: str, options: Any) -> AsyncIterator[Any]:
+        async def _stream() -> AsyncIterator[Any]:
+            for message in messages:
+                yield message
+
+        return _stream()
+
+    return fake_query
+
+
+def _assistant(*blocks: Any, model: str = "model-real", error: Any = None) -> AssistantMessage:
+    return AssistantMessage(content=list(blocks), model=model, error=error)
+
+
+def _result(
+    usage: dict[str, Any] | None = None,
+    total_cost_usd: float | None = None,
+    is_error: bool = False,
+    subtype: str = "success",
+    errors: list[str] | None = None,
+) -> ResultMessage:
+    return ResultMessage(
+        subtype=subtype,
+        duration_ms=1,
+        duration_api_ms=1,
+        is_error=is_error,
+        num_turns=1,
+        session_id="s",
+        usage=usage,
+        total_cost_usd=total_cost_usd,
+        errors=errors,
+    )
+
+
+def _client() -> SdkModelClient:
+    return SdkModelClient(ModelMapContract(profiles={"anthropic": {"default": "model-default"}}))
+
+
+_FULL_USAGE = {
+    "input_tokens": 10,
+    "output_tokens": 5,
+    "cache_creation_input_tokens": 3,
+    "cache_read_input_tokens": 2,
+}
+
+
+class TestCompleteAsyncStreamBinding:
+    """C2.5 (R-4/R-5): the read loop is BOUND offline with real SDK message types.
+
+    Before C2.5 nothing in the suite executed sdk_client's aggregation,
+    error, usage or cost branches — the fake stream (real ``AssistantMessage``
+    / ``ResultMessage`` / ``TextBlock`` objects, so constructor drift also
+    goes red) binds every branch without a key or the network.
+    """
+
+    def test_text_aggregates_and_non_text_blocks_are_ignored(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(
+            sdk_client,
+            "query",
+            _stream_of(
+                _assistant(TextBlock("{"), ThinkingBlock(thinking="hmm", signature="sig")),
+                _assistant(TextBlock("}")),
+                _result(usage=_FULL_USAGE, total_cost_usd=0.01),
+            ),
+        )
+        client = _client()
+        reply = client.complete("p", role="proposer")
+        assert reply.text == "{}"
+        assert reply.model == "model-real"
+        assert client.last_model == "model-real"
+
+    def test_usage_tokens_sum_the_four_provider_fields(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(
+            sdk_client,
+            "query",
+            _stream_of(_assistant(TextBlock("ok")), _result(usage=_FULL_USAGE)),
+        )
+        assert _client().complete("p", role="proposer").usage_tokens == 20
+
+    def test_cost_accumulates_across_calls(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        client = _client()
+        for cost in (0.01, 0.02):
+            monkeypatch.setattr(
+                sdk_client,
+                "query",
+                _stream_of(_assistant(TextBlock("ok")), _result(total_cost_usd=cost)),
+            )
+            client.complete("p", role="proposer")
+        assert client.total_cost_usd == pytest.approx(0.03)
+
+    def test_an_assistant_error_fails_the_call(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(
+            sdk_client, "query", _stream_of(_assistant(TextBlock("x"), error="rate_limit"))
+        )
+        with pytest.raises(RuntimeError, match="rate_limit"):
+            _client().complete("p", role="proposer")
+
+    def test_a_result_error_fails_the_call_naming_subtype_and_errors(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(
+            sdk_client,
+            "query",
+            _stream_of(
+                _assistant(TextBlock("x")),
+                _result(is_error=True, subtype="error_during_execution", errors=["boom"]),
+            ),
+        )
+        with pytest.raises(RuntimeError, match="error_during_execution.*boom"):
+            _client().complete("p", role="proposer")
+
+
+class TestTotalTokensFailsClosed:
+    """§8: the meter is never fed an invented count — no usage stays ``None``."""
+
+    def test_no_usage_dict_is_none(self) -> None:
+        assert _total_tokens(None) is None
+
+    def test_an_empty_usage_dict_is_none(self) -> None:
+        assert _total_tokens({}) is None
+
+    def test_non_int_fields_are_ignored_not_coerced(self) -> None:
+        assert _total_tokens({"input_tokens": "10"}) is None
+        assert _total_tokens({"input_tokens": 10, "output_tokens": "x"}) == 10
+
+
+class TestBudgetGuard:
+    """§8: a non-positive per-call USD cap is refused at construction."""
+
+    @pytest.mark.parametrize("cap", [0.0, -0.5])
+    def test_non_positive_caps_are_rejected(self, cap: float) -> None:
+        with pytest.raises(ValueError):
+            SdkModelClient(
+                ModelMapContract(profiles={"anthropic": {"default": "m"}}),
+                max_budget_usd_per_call=cap,
+            )

tests/test_sdk_version_guard.py

@@ -0,0 +1,82 @@
+"""SDK version guard (C2.5, closes C-N3) — LOAD-BEARING (§11).
+
+The seam this file keeps alive: every SDK attribute premise in
+``sdk_client.py`` was verified against a CONCRETE version range
+(0.2.110 read at source level, release notes through 0.2.120; sdk-review
+2026-07-16). The pin ``claude-agent-sdk>=0.2.111,<0.3`` freezes that range —
+this guard makes an upgrade outside it a RED test naming exactly which
+premises must be re-verified, instead of a silent behaviour drift.
+
+Offline-safe: reads installed package metadata only — no key, no network.
+"""
+
+from __future__ import annotations
+
+import importlib.metadata
+from pathlib import Path
+
+import pytest
+
+_PYPROJECT = Path(__file__).resolve().parents[1] / "pyproject.toml"
+
+# The verified range — MUST match the pyproject pin (bound below).
+_VERIFIED_FLOOR = (0, 2, 111)
+_VERIFIED_CEILING = (0, 3)
+_PIN = "claude-agent-sdk>=0.2.111,<0.3"
+
+# The sdk_client.py attribute premises the verified range vouches for
+# (sdk-review 2026-07-16, verified against package source through 0.2.120).
+_SDK_PREMISES = (
+    "AssistantMessage.error/.model/.content",
+    "ResultMessage.usage/.total_cost_usd/.is_error/.subtype/.errors",
+    "ClaudeAgentOptions.max_budget_usd/.setting_sources/.system_prompt/.max_turns/.model/.tools",
+    "TextBlock.text",
+    "query() yields AssistantMessage then a closing ResultMessage",
+)
+
+
+def _parse(raw: str) -> tuple[int, ...]:
+    return tuple(int(part) for part in raw.split(".")[:3])
+
+
+def check_sdk_version() -> str:
+    """Fail if the installed SDK is outside the verified range — naming the premises."""
+    raw = importlib.metadata.version("claude-agent-sdk")
+    if not (_VERIFIED_FLOOR <= _parse(raw) < _VERIFIED_CEILING):
+        raise AssertionError(
+            f"claude-agent-sdk {raw} is OUTSIDE the verified range >=0.2.111,<0.3. "
+            "Re-verify the sdk_client.py attribute premises against the new version "
+            "BEFORE widening the pin (pyproject.toml + this guard together): "
+            + "; ".join(_SDK_PREMISES)
+        )
+    return raw
+
+
+class TestSdkVersionGuard:
+    def test_the_installed_sdk_is_within_the_verified_range(self) -> None:
+        assert check_sdk_version() == importlib.metadata.version("claude-agent-sdk")
+
+    def test_a_version_above_the_ceiling_trips_the_guard(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(importlib.metadata, "version", lambda name: "0.3.0")
+        with pytest.raises(AssertionError) as err:
+            check_sdk_version()
+        message = str(err.value)
+        assert "sdk_client.py" in message
+        for premise in _SDK_PREMISES:
+            assert premise in message
+
+    def test_a_version_below_the_floor_trips_the_guard(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # 0.2.110 works but misses the 0.2.111 read-loop fixes (NDJSON
+        # whitespace on >64 KiB lines, plain-string content TypeError,
+        # zombie subprocess on cancellation) — below the floor is unverified.
+        monkeypatch.setattr(importlib.metadata, "version", lambda name: "0.2.110")
+        with pytest.raises(AssertionError):
+            check_sdk_version()
+
+    def test_the_pyproject_pin_matches_the_verified_range(self) -> None:
+        # Detach-proof: the pin and this guard cannot drift apart silently.
+        assert _PIN in _PYPROJECT.read_text(encoding="utf-8")

tests/test_step7_async_loop_loadbearing.py

@@ -13,7 +13,9 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
+import pytest
 from _scripted import ScriptedClient, reply
+from pydantic import ValidationError
 
 from portfolio_optimiser_claude.budget import BudgetMeter
 from portfolio_optimiser_claude.contracts import TerminationContract
@@ -25,6 +27,7 @@ from portfolio_optimiser_claude.experience import (
     mint_verdict_id,
 )
 from portfolio_optimiser_claude.inbox import (
+    InboxLimitError,
     ProposalFeatures,
     VerdictDocument,
     load_inbox,
@@ -162,6 +165,98 @@ class TestTolerantLoad:
         assert [d.id for d in load_inbox(tmp_path)] == sorted([late.id, early.id])
 
 
+class TestFileLayerVocabulary:
+    """C2.5 (C-F7): the file layer polices the §4.2 decision set — SKIP, never raise."""
+
+    def test_an_unknown_decision_never_reaches_the_store(self, tmp_path: Path) -> None:
+        # Fixture = the review's RUN C-F7 proof: before C2.5 this string
+        # entered the store and its rationale folded into the next prompt.
+        doc = _document().model_copy(update={"decision": "hva-som-helst"})
+        (tmp_path / f"{doc.id}.json").write_text(json.dumps(doc.model_dump()), encoding="utf-8")
+        store = VerdictStore()
+        assert merge_inbox_into_store(store, tmp_path) == 0
+        assert len(store) == 0
+
+    def test_the_full_file_layer_vocabulary_loads(self, tmp_path: Path) -> None:
+        # Control: exactly the §4.2 set {approved, rejected,
+        # approved_with_adjustment} passes the file layer.
+        decisions = ("approved", "rejected", "approved_with_adjustment")
+        for index, decision in enumerate(decisions):
+            write_verdict(
+                tmp_path, _document(_features(codes=frozenset({f"C{index}"})), decision=decision)
+            )
+        assert {d.decision for d in load_inbox(tmp_path)} == set(decisions)
+
+
+class TestInboxCaps:
+    """C2.5: configurable caps fail FAST with a precise error — never a silent cut."""
+
+    def test_rationale_over_the_cap_fails_fast_with_a_precise_error(self, tmp_path: Path) -> None:
+        doc = _document(rationale="x" * 201)
+        write_verdict(tmp_path, doc)
+        with pytest.raises(InboxLimitError) as err:
+            load_inbox(tmp_path, max_rationale_chars=200)
+        message = str(err.value)
+        assert doc.id in message
+        assert "201" in message
+        assert "200" in message
+
+    def test_rationale_at_the_cap_loads(self, tmp_path: Path) -> None:
+        write_verdict(tmp_path, _document(rationale="x" * 200))
+        assert len(load_inbox(tmp_path, max_rationale_chars=200)) == 1
+
+    def test_more_files_than_the_cap_fails_fast(self, tmp_path: Path) -> None:
+        for index in range(3):
+            write_verdict(tmp_path, _document(_features(codes=frozenset({f"C{index}"}))))
+        with pytest.raises(InboxLimitError) as err:
+            load_inbox(tmp_path, max_files=2)
+        message = str(err.value)
+        assert "3" in message
+        assert "2" in message
+
+    def test_merge_threads_the_caps_through(self, tmp_path: Path) -> None:
+        write_verdict(tmp_path, _document(rationale="x" * 201))
+        with pytest.raises(InboxLimitError):
+            merge_inbox_into_store(VerdictStore(), tmp_path, max_rationale_chars=200)
+
+
+class TestIdGrammar:
+    """C2.5 (R-6): the id grammar is policed BEFORE ``write_verdict`` touches disk."""
+
+    def test_a_traversal_id_is_rejected_at_construction(self) -> None:
+        # The security agent's RUN R-6 proof: '../../escaped' used to
+        # construct fine and write outside the inbox.
+        with pytest.raises(ValidationError):
+            VerdictDocument(
+                id="../../escaped",
+                decision="approved",
+                rationale=RATIONALE,
+                proposal_features=ProposalFeatures(
+                    affected_codes=["E01"],
+                    measure_type="led-retrofit",
+                    claimed_saving_nok=25000.0,
+                    description="surface text",
+                ),
+            )
+
+    def test_write_verdict_refuses_an_escaping_id_writing_nothing(self, tmp_path: Path) -> None:
+        # model_copy(update=...) bypasses model validation — the write seam
+        # must fail closed on its own (detach the grammar here → a file lands
+        # OUTSIDE the inbox → red).
+        inbox = tmp_path / "inbox"
+        doc = _document().model_copy(update={"id": "../../escaped"})
+        with pytest.raises(ValueError):
+            write_verdict(inbox, doc)
+        assert not (tmp_path / "escaped.json").exists()
+        assert not inbox.exists()
+
+    def test_a_loaded_file_with_an_escaping_id_is_skipped(self, tmp_path: Path) -> None:
+        # Tolerant load stays tolerant: a hostile id is SKIPPED, never raised.
+        doc = _document().model_copy(update={"id": "../../escaped"})
+        (tmp_path / "escaped.json").write_text(json.dumps(doc.model_dump()), encoding="utf-8")
+        assert load_inbox(tmp_path) == []
+
+
 class TestMergeIntoStore:
     """§5: merge, never replace — first-write-wins per id, idempotent, read-only."""
 

tests/test_step8_promotion_loadbearing.py

@@ -203,6 +203,18 @@ class TestPathSafety:
         assert not (tmp_path / "evil").exists()
         assert "/" not in path.name and "\\" not in path.name
 
+    def test_sanitised_ids_never_collide_with_a_distinct_id_owning_the_name(
+        self, bundle: Path
+    ) -> None:
+        # C2.5 restarbeid-funn 2: 'e/vil' used to sanitise to 'evil' — the
+        # SAME promoted filename as the distinct id 'evil', so the second
+        # promotion silently overwrote the first curated verdict.
+        first = _promote(_document(verdict_id="e/vil", rationale=f"one ({MARKER})"), bundle)
+        second = _promote(_document(verdict_id="evil", rationale=f"two ({MARKER})"), bundle)
+        assert first != second
+        assert parse_concept_file(first).frontmatter["verdict_id"] == "e/vil"
+        assert parse_concept_file(second).frontmatter["verdict_id"] == "evil"
+
     def test_degenerate_token_falls_back_to_a_content_hash(self, bundle: Path) -> None:
         path = _promote(_document(verdict_id="///"), bundle)
         assert path.parent == bundle