[eric] tests: one shared logger capture, and the shaper's names follow the p_ convention

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018foyDoK19jjbYdudfzQVkZ

backend/apps/agents/core/fault_injection.py

@@ -37,7 +37,7 @@ def armed(name: str) -> bool:
     return name in (wanted & KNOWN_FAULTS)
 
 
-_FIRED: Set[str] = set()
+P_FIRED: Set[str] = set()
 
 
 def armed_once(name: str) -> bool:
@@ -46,15 +46,15 @@ def armed_once(name: str) -> bool:
     A fault that fires on every turn cannot drill a recovery: the retry hits the same wall and the
     drill only ever proves the failure, never the heal. Recoverable classes (a dead pipe, a rotated
     token) want one hit and then a clear road."""
-    if not armed(name) or name in _FIRED:
+    if not armed(name) or name in P_FIRED:
         return False
-    _FIRED.add(name)
+    P_FIRED.add(name)
     return True
 
 
 def reset_fired() -> None:
     """Test-only: forget what has fired so a case can arm the same one-shot again."""
-    _FIRED.clear()
+    P_FIRED.clear()
 
 
 def unknown_faults() -> Set[str]:

backend/apps/agents/manager/streaming/tool_output_shaper.py

@@ -114,41 +114,42 @@ def shape_for_model(session: object, session_id: str, response: object, msg_id:
     # a user who hits a bad shape has a lever that is not "downgrade". It announces itself: a guard
     # that stops guarding in silence is the bug class this module was written under.
     if os.environ.get("OSW_TOOL_SHAPING") == "off":
-        p_bump(session, "disabled", 1)
-        if getattr(session, "_shaping_off_said", False) is False:
+        bump_shaping_stat(session, "disabled", 1)
+        if getattr(session, "p_shaping_off_said", False) is False:
             logger.warning("tool-output shaping is OFF (OSW_TOOL_SHAPING=off); every tool result "
                            "will be sent to the model in full")
             try:
-                session._shaping_off_said = True  # type: ignore[attr-defined]
+                session.p_shaping_off_said = True  # type: ignore[attr-defined]
             except Exception:
                 pass
         return None
     probe, _, _ = shape_tool_response(response, "")
     if probe is None:
-        p_bump(session, "seen", 1)
+        bump_shaping_stat(session, "seen", 1)
         return None
 
-    p_body = _payload_text(response)
+    p_body = p_payload_text(response)
     blob = write_blob(p_body, session_id, msg_id, suffix="-model") if p_body else None
     if blob is None:
         # No recovery path means the cut would be unrecoverable, which is the one thing this must
         # never do. Spend the tokens instead.
         logger.warning(f"tool-output shaping skipped for {session_id}: full body could not be parked")
-        p_bump(session, "skipped_no_recovery", 1)
+        bump_shaping_stat(session, "skipped_no_recovery", 1)
         return None
 
     shaped, before, after = shape_tool_response(response, blob)
     if shaped is None:
         return None
-    p_bump(session, "seen", 1)
-    p_bump(session, "shaped", 1)
-    p_bump(session, "bytes_before", before)
-    p_bump(session, "bytes_after", after)
+    bump_shaping_stat(session, "seen", 1)
+    bump_shaping_stat(session, "shaped", 1)
+    bump_shaping_stat(session, "bytes_before", before)
+    bump_shaping_stat(session, "bytes_after", after)
     logger.info(f"shaped {tool_name} result for the model: {before} -> {after} bytes (full copy at {blob})")
     return shaped
 
 
-def _payload_text(response: object) -> str:
+@typechecked
+def p_payload_text(response: object) -> str:
     """The field shape_tool_response would rewrite, so the parked copy is the thing being cut."""
     if isinstance(response, str):
         return response
@@ -166,7 +167,8 @@ def _payload_text(response: object) -> str:
     return ""
 
 
-def p_bump(session: object, key: str, n: int) -> None:
+@typechecked
+def bump_shaping_stat(session: object, key: str, n: int) -> None:
     stats = getattr(session, "tool_shaping", None)
     if not isinstance(stats, dict):
         stats = {}

backend/tests/log_capture.py

@@ -0,0 +1,32 @@
+"""Capture ONE logger's records, regardless of what the rest of the run did to the logging tree.
+
+`caplog` reads the ROOT logger, so it goes blind the moment any other test sets propagate=False on
+the logger under test: two assertions here passed in isolation and failed in the full suite while
+the line was plainly visible in captured stderr. A logging assertion must not depend on the other
+three thousand tests.
+"""
+
+import logging
+from typing import List
+
+
+class LogCapture:
+    def __init__(self, name: str) -> None:
+        self.logger = logging.getLogger(name)
+        self.records: List[logging.LogRecord] = []
+        self.handler = logging.Handler()
+        self.handler.emit = lambda record: self.records.append(record)  # type: ignore[method-assign]
+
+    def __enter__(self) -> "LogCapture":
+        self.prev = self.logger.level
+        self.logger.setLevel(logging.WARNING)
+        self.logger.addHandler(self.handler)
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self.logger.removeHandler(self.handler)
+        self.logger.setLevel(self.prev)
+
+    @property
+    def text(self) -> str:
+        return "\n".join(r.getMessage() for r in self.records)

backend/tests/test_fault_injection.py

@@ -11,6 +11,8 @@ import pytest
 
 import builtins
 
+from backend.tests.log_capture import LogCapture
+
 from backend.apps.agents.core.fault_injection import (
     KNOWN_FAULTS, announce, armed, armed_once, reset_fired, unknown_faults,
 )
@@ -145,11 +147,11 @@ def test_arming_is_announced_and_a_typo_is_named(monkeypatch):
     # The harness built to kill row-6 silence had it: unknown_faults() existed and NOTHING called it,
     # so a mistyped name armed nothing while the drill exercised the untouched happy path.
     monkeypatch.setenv("OSW_FAULT", "policy_block,plicy_blok")
-    with p_capture("backend.apps.agents.core.fault_injection") as cap:
+    with LogCapture("backend.apps.agents.core.fault_injection") as cap:
         announce()
     assert "policy_block" in cap.text and "plicy_blok" in cap.text
     monkeypatch.delenv("OSW_FAULT")
-    with p_capture("backend.apps.agents.core.fault_injection") as cap2:
+    with LogCapture("backend.apps.agents.core.fault_injection") as cap2:
         announce()
     assert cap2.text == "", "a shipped build must say nothing at all"
 

backend/tests/test_tool_output_shaper.py

@@ -7,12 +7,12 @@ without naming the file that still holds it. These pin both halves, plus the sha
 enforces in silence.
 """
 
-import json
-
 import pytest
 
+from backend.tests.log_capture import LogCapture
+
 from backend.apps.agents.manager.streaming.tool_output_shaper import (
-    SHAPE_OVER_BYTES, shape_text, shape_tool_response, shaping_report, p_bump,
+    SHAPE_OVER_BYTES, bump_shaping_stat, shape_text, shape_tool_response, shaping_report,
 )
 
 HOOK = "backend/apps/agents/manager/streaming/post_tool_hook.py"
@@ -84,11 +84,11 @@ def test_it_says_so_when_it_cut_nothing_at_depth():
         pass
     s = S()
     for _ in range(45):
-        p_bump(s, "seen", 1)
+        bump_shaping_stat(s, "seen", 1)
     assert "0 of 45" in (shaping_report(s) or "")
-    p_bump(s, "shaped", 1)
-    p_bump(s, "bytes_before", 9_000)
-    p_bump(s, "bytes_after", 2_000)
+    bump_shaping_stat(s, "shaped", 1)
+    bump_shaping_stat(s, "bytes_before", 9_000)
+    bump_shaping_stat(s, "bytes_after", 2_000)
     assert "7,000 bytes" in (shaping_report(s) or "")
 
 
@@ -96,7 +96,7 @@ def test_a_shallow_session_says_nothing():
     class S:
         pass
     s = S()
-    p_bump(s, "seen", 3)
+    bump_shaping_stat(s, "seen", 3)
     assert shaping_report(s) is None
 
 
@@ -120,11 +120,10 @@ def test_the_hook_shapes_the_pristine_response_not_the_flattened_one():
 
 def test_the_off_switch_is_declared_and_announces_itself(monkeypatch):
     from backend.apps.agents.manager.streaming import tool_output_shaper as mod
-    from backend.tests.test_fault_injection import p_capture
 
     class S:
         id = "s1"
     monkeypatch.setenv("OSW_TOOL_SHAPING", "off")
-    with p_capture("backend.apps.agents.manager.streaming.tool_output_shaper") as cap:
+    with LogCapture("backend.apps.agents.manager.streaming.tool_output_shaper") as cap:
         assert mod.shape_for_model(S(), "s1", {"stdout": p_big()}, "m1", "Bash") is None
     assert "OFF" in cap.text, "a guard that stops guarding must say which sessions it stopped protecting"