From 482e94ecb214abe7a8d6e223d21d4b6775ae430e Mon Sep 17 00:00:00 2001 From: ciregenz Date: Mon, 24 Aug 2026 01:06:56 -0700 Subject: [PATCH] [eric] tests: one shared logger capture, and the shaper's names follow the p_ convention Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018foyDoK19jjbYdudfzQVkZ --- backend/apps/agents/core/fault_injection.py | 8 ++--- .../manager/streaming/tool_output_shaper.py | 26 ++++++++------- backend/tests/log_capture.py | 32 +++++++++++++++++++ backend/tests/test_fault_injection.py | 6 ++-- backend/tests/test_tool_output_shaper.py | 19 ++++++----- 5 files changed, 63 insertions(+), 28 deletions(-) create mode 100644 backend/tests/log_capture.py diff --git a/backend/apps/agents/core/fault_injection.py b/backend/apps/agents/core/fault_injection.py index aa2353ff..f855aaf8 100644 --- a/backend/apps/agents/core/fault_injection.py +++ b/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]: diff --git a/backend/apps/agents/manager/streaming/tool_output_shaper.py b/backend/apps/agents/manager/streaming/tool_output_shaper.py index 2ea6af5e..3b624a52 100644 --- a/backend/apps/agents/manager/streaming/tool_output_shaper.py +++ b/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 = {} diff --git a/backend/tests/log_capture.py b/backend/tests/log_capture.py new file mode 100644 index 00000000..52b6a6b5 --- /dev/null +++ b/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) diff --git a/backend/tests/test_fault_injection.py b/backend/tests/test_fault_injection.py index fed6f95d..916f488b 100644 --- a/backend/tests/test_fault_injection.py +++ b/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" diff --git a/backend/tests/test_tool_output_shaper.py b/backend/tests/test_tool_output_shaper.py index bd2ee157..c84e0022 100644 --- a/backend/tests/test_tool_output_shaper.py +++ b/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"