feat(reasoning): expose mid-session style-swap detection via ReasoningAuditor (closes #370)

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

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."""
 

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"

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