diff --git a/CHANGELOG.md b/CHANGELOG.md index ef546fd6..abfb9ada 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,9 @@ file and uses it as the GitHub Release body, so keep each version's notes here. - Auto-updating contributors wall (scheduled workflow + `contributors.json`). - PyPI auto-publish workflow on `v*` tags and a PR test/coverage workflow. +### Changed +- Refactor (`#500`): Inlined `SilenceBaseline` attributes into `SilentFailureDetector`. Reduces file complexity and removes the redundant dataclass abstraction (also addresses `#478` and `#498`). + ## [0.2.0-preview] - 2026-05-27 ### Added diff --git a/agentwatch/core/schema.py b/agentwatch/core/schema.py index 902141db..5afbadaa 100644 --- a/agentwatch/core/schema.py +++ b/agentwatch/core/schema.py @@ -66,6 +66,10 @@ class EventType(str, Enum): CONFIDENCE_SCORE = "confidence.score" ANOMALY_DETECTED = "anomaly.detected" + # Reasoning auditor + STYLE_FINGERPRINT_COMPUTED = "reasoning.style_fingerprint" + STYLE_SWAP_DETECTED = "reasoning.style_swap" + # Generic CUSTOM = "custom" @@ -240,6 +244,54 @@ class ConfidenceData(BaseModel): explanation: str | None = None +class ReasoningStyleFingerprint(BaseModel): + """Per-session stylistic profile of the model's reasoning output (RSN-008). + + Captures *observable* artifacts of the reasoning style — not chain-of-thought + contents — so it can be used to detect silent mid-session model swaps + (e.g. a provider rolling out a new model version without changing the + exposed model name). + """ + + mean_planner_tokens: float = 0.0 + lex_diversity: float = 0.0 + tools_per_plan: float = 0.0 + punctuation_rate: float = 0.0 + sample_size: int = 0 # number of PLANNER_OUTPUT events used + + @classmethod + def from_dataclass(cls, other: object) -> ReasoningStyleFingerprint: + """Adapt the dataclass form in ``agentwatch.reasoning.fingerprint``. + + Returns: + A pydantic instance mirroring the dataclass fields. + """ + return cls( + mean_planner_tokens=float(getattr(other, "mean_planner_tokens", 0.0)), + lex_diversity=float(getattr(other, "lex_diversity", 0.0)), + tools_per_plan=float(getattr(other, "tools_per_plan", 0.0)), + punctuation_rate=float(getattr(other, "punctuation_rate", 0.0)), + sample_size=int(getattr(other, "sample_size", 0)), + ) + + +class StyleSwapAlert(BaseModel): + """Alert raised when mid-session fingerprint drift suggests a model swap. + + Distance is the Euclidean-style distance between the first-half and + second-half fingerprints, normalized to be roughly comparable across + sessions with different sizes. + """ + + session_id: str + detected: bool + distance: float + threshold: float + first_half: ReasoningStyleFingerprint + second_half: ReasoningStyleFingerprint + reason: str | None = None + + class AgentMessageData(BaseModel): """Inter-agent message in a multi-agent workflow.""" diff --git a/agentwatch/reasoning/auditor.py b/agentwatch/reasoning/auditor.py index de0821f2..0f498a98 100644 --- a/agentwatch/reasoning/auditor.py +++ b/agentwatch/reasoning/auditor.py @@ -7,7 +7,17 @@ from statistics import mean from typing import Any, cast -from agentwatch.core.schema import AgentEvent, EventType +from agentwatch.core.schema import ( + AgentEvent, + EventType, + ReasoningStyleFingerprint, + StyleSwapAlert, +) +from agentwatch.reasoning.fingerprint import ( + StyleFingerprint, + detect_mid_session_change, + fingerprint, +) JudgeCallback = Callable[[str, AgentEvent], Awaitable[dict[str, Any]]] @@ -50,6 +60,22 @@ def to_dict(self) -> dict[str, object]: } +@dataclass +class FingerprintReport: + """Bundle of style-fingerprint + swap-detection results for a session.""" + + session_id: str + fingerprint: ReasoningStyleFingerprint + swap_alert: StyleSwapAlert + + def to_dict(self) -> dict[str, object]: + return { + "session_id": self.session_id, + "fingerprint": self.fingerprint.model_dump(), + "swap_alert": self.swap_alert.model_dump(), + } + + class ReasoningAuditor: """ Audits observable reasoning artifacts. @@ -100,6 +126,77 @@ async def audit_step(self, step_index: int, event: AgentEvent) -> StepAudit: ) return self._heuristic_audit(step_index, event) + def fingerprint_session( + self, + events: list[AgentEvent], + *, + distance_threshold: float = 1.0, + split_ratio: float = 0.5, + ) -> FingerprintReport: + """Compute the session style fingerprint and detect mid-session swaps. + + Args: + events: All events recorded for the session (any timeline order is + fine — the mid-session split is positional on the list you pass). + distance_threshold: Euclidean-style distance above which a fingerprint + delta is reported as a probable mid-session model swap. + split_ratio: Where to split the events list into halves when + comparing the first vs second half. + + Returns: + A :class:`FingerprintReport` containing both the full-session + fingerprint and the swap-detection alert. + """ + full = fingerprint(events) + detected, distance = detect_mid_session_change( + events, + split_ratio=split_ratio, + distance_threshold=distance_threshold, + ) + # Compute both halves for diagnostic context, even when we cannot + # actually compare them — this lets downstream consumers see *why* + # detection returned ``False``. + cut = int(len(events) * split_ratio) + first, second = events[:cut], events[cut:] + first_fp = self._pydantic_fingerprint(fingerprint(first)) + second_fp = self._pydantic_fingerprint(fingerprint(second)) + session_id = events[0].session_id if events else "" + reason: str | None = None + if not detected: + if len(events) < 6: + reason = "insufficient_events" + elif distance == 0.0: + reason = "insufficient_planner_signal" + else: + reason = "below_threshold" + plans_count = sum(1 for e in events if e.event_type == EventType.PLANNER_OUTPUT) + fp_pydantic = self._pydantic_fingerprint(full, sample_size=plans_count) + return FingerprintReport( + session_id=session_id, + fingerprint=fp_pydantic, + swap_alert=StyleSwapAlert( + session_id=session_id, + detected=detected, + distance=round(distance, 4), + threshold=distance_threshold, + first_half=first_fp, + second_half=second_fp, + reason=reason, + ), + ) + + @staticmethod + def _pydantic_fingerprint( + fp: StyleFingerprint, + *, + sample_size: int | None = None, + ) -> ReasoningStyleFingerprint: + """Bridge from the dataclass fingerprint to the schema model.""" + instance = ReasoningStyleFingerprint.from_dataclass(fp) + if sample_size is not None: + instance = instance.model_copy(update={"sample_size": sample_size}) + return instance + def _build_prompt(self, event: AgentEvent) -> str: return ( "Judge the quality of this agent reasoning artifact on a 0-1 scale.\n" diff --git a/tests/test_reasoning_style_fingerprint.py b/tests/test_reasoning_style_fingerprint.py new file mode 100644 index 00000000..bb6c1beb --- /dev/null +++ b/tests/test_reasoning_style_fingerprint.py @@ -0,0 +1,153 @@ +"""Tests for RSN-008 reasoning style fingerprint integration with the auditor. + +Covers: +- ReasoningStyleFingerprint and StyleSwapAlert schema models. +- ReasoningAuditor.fingerprint_session() end-to-end. +- Auto-derivation of StyleSwapAlert.reason for non-detection cases. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from agentwatch.core.schema import ( + AgentEvent, + EventType, + ReasoningStyleFingerprint, + StyleSwapAlert, +) +from agentwatch.reasoning.auditor import ReasoningAuditor +from agentwatch.reasoning.fingerprint import StyleFingerprint + + +def _plan(text: str, index: int) -> AgentEvent: + return AgentEvent( + session_id="S", + agent_id="A", + event_type=EventType.PLANNER_OUTPUT, + planner_output_preview=text, + step_number=index, + timestamp=datetime(2026, 1, 1, tzinfo=UTC), + ) + + +def _tool(index: int) -> AgentEvent: + return AgentEvent( + session_id="S", + agent_id="A", + event_type=EventType.TOOL_CALL, + step_number=index, + timestamp=datetime(2026, 1, 1, tzinfo=UTC), + ) + + +def test_reasoning_style_fingerprint_from_dataclass(): + """Adapt the dataclass form to the pydantic schema model.""" + fp = StyleFingerprint( + mean_planner_tokens=12.5, + lex_diversity=0.8, + tools_per_plan=2.0, + punctuation_rate=0.05, + ) + pyd = ReasoningStyleFingerprint.from_dataclass(fp) + assert pyd.mean_planner_tokens == 12.5 + assert pyd.lex_diversity == 0.8 + assert pyd.tools_per_plan == 2.0 + assert pyd.punctuation_rate == 0.05 + assert pyd.sample_size == 0 + + +def test_style_swap_alert_serialises(): + """StyleSwapAlert round-trips through model_dump for the API.""" + fp = ReasoningStyleFingerprint( + mean_planner_tokens=10.0, + lex_diversity=0.6, + tools_per_plan=1.5, + punctuation_rate=0.04, + ) + alert = StyleSwapAlert( + session_id="S", + detected=True, + distance=1.2, + threshold=1.0, + first_half=fp, + second_half=fp, + ) + out = alert.model_dump() + assert out["session_id"] == "S" + assert out["detected"] is True + assert out["distance"] == 1.2 + assert out["first_half"]["mean_planner_tokens"] == 10.0 + + +def test_fingerprint_session_no_swap_detected_consistent_plans(): + plan = "Read the configuration file carefully and apply the migration step by step." + events = [_plan(plan, i) for i in range(8)] + auditor = ReasoningAuditor() + report = auditor.fingerprint_session(events) + + assert report.session_id == "S" + assert report.fingerprint.sample_size == 8 + assert report.swap_alert.detected is False + assert report.swap_alert.distance == 0.0 + assert report.swap_alert.reason in {"insufficient_planner_signal", "below_threshold"} + + +def test_fingerprint_session_detects_style_swap(): + """When planner verbosity changes mid-session, detection should fire.""" + short_plans = [_plan("ok", i) for i in range(4)] + verbose_plans = [ + _plan( + "A carefully constructed plan that explores all the relevant " + "details and considerations before committing to a course of " + "action. Multiple sentences? Considerable detail!", + i + 4, + ) + for i in range(4) + ] + events = short_plans + verbose_plans + auditor = ReasoningAuditor() + report = auditor.fingerprint_session(events, distance_threshold=0.3) + + assert report.swap_alert.detected is True + assert report.swap_alert.distance > 0.3 + assert report.swap_alert.threshold == 0.3 + assert report.swap_alert.reason is None # detected -> no reason blob + + +def test_fingerprint_session_short_circuit_records_reason(): + events = [_plan("one", 0), _tool(1)] + auditor = ReasoningAuditor() + report = auditor.fingerprint_session(events) + assert report.swap_alert.detected is False + assert report.swap_alert.reason == "insufficient_events" + + +def test_fingerprint_report_to_dict_roundtrip(): + plan = "A plan with multiple words and detail for testing serialization purposes." + events = [_plan(plan, i) for i in range(6)] + auditor = ReasoningAuditor() + report = auditor.fingerprint_session(events) + data = report.to_dict() + assert "fingerprint" in data + assert "swap_alert" in data + assert isinstance(data["fingerprint"]["sample_size"], int) + + +@pytest.mark.parametrize( + "events, expected_reason", + [ + ([], "insufficient_events"), + ([_plan("x", 0)], "insufficient_events"), + ([_plan("plan", i) for i in range(5)], None), # below threshold in even dist + ], +) +def test_fingerprint_session_handles_edge_cases(events, expected_reason): + auditor = ReasoningAuditor() + report = auditor.fingerprint_session(events) + if expected_reason is None: + assert report.swap_alert.detected is False + else: + assert report.swap_alert.reason == expected_reason