From 662fa1ba0b5470add042d27207f182f9056f98b2 Mon Sep 17 00:00:00 2001 From: arcgod-design Date: Sun, 5 Jul 2026 16:32:59 +0530 Subject: [PATCH 1/3] docs: note SilenceBaseline inlining in CHANGELOG for #500 --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) 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 From 95e48795be1e1368074c302c9d24a62feff0d110 Mon Sep 17 00:00:00 2001 From: arcgod-design Date: Sun, 5 Jul 2026 16:54:01 +0530 Subject: [PATCH 2/3] feat(reasoning): expose mid-session style-swap detection via ReasoningAuditor (closes #370) --- agentwatch/core/schema.py | 52 ++++++++ agentwatch/reasoning/auditor.py | 99 +++++++++++++- tests/test_reasoning_style_fingerprint.py | 153 ++++++++++++++++++++++ 3 files changed, 303 insertions(+), 1 deletion(-) create mode 100644 tests/test_reasoning_style_fingerprint.py 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 From 610bcd916ec6b2c9067ee2c6d4b84971cdbe0527 Mon Sep 17 00:00:00 2001 From: Saketh Suman Bathini Date: Sun, 5 Jul 2026 17:03:34 +0530 Subject: [PATCH 3/3] feat(cli): add cost report command for spend by framework (#339) (#557) * feat(cli): add cost report command for spend by framework (#339) * fix(cli): warn on truncated results, skip malformed sessions, dedup group_by validation (#339) --- agentwatch/cli/main.py | 134 ++++++++++++++++++++++++++++ agentwatch/cost/reporting.py | 166 +++++++++++++++++++++++++++++++++++ tests/test_cost_reporting.py | 157 +++++++++++++++++++++++++++++++++ 3 files changed, 457 insertions(+) create mode 100644 agentwatch/cost/reporting.py create mode 100644 tests/test_cost_reporting.py diff --git a/agentwatch/cli/main.py b/agentwatch/cli/main.py index 5f279259..f6a1a279 100644 --- a/agentwatch/cli/main.py +++ b/agentwatch/cli/main.py @@ -27,6 +27,8 @@ # type-only import keeps the annotation without a hard runtime import. import httpx + from agentwatch.cost.reporting import CostReport + app = typer.Typer( name="agentwatch", help="AgentWatch — Reliability, Safety, and Observability Layer for AI Agents", @@ -51,8 +53,16 @@ no_args_is_help=True, ) +cost_app = typer.Typer( + name="cost", + help="AgentWatch FinOps. Report token usage and API spend across agents and frameworks.", + rich_markup_mode="rich", + no_args_is_help=True, +) + app.add_typer(server_app) app.add_typer(safety_app) +app.add_typer(cost_app) _IN_REPL = False @@ -799,6 +809,130 @@ def safety( ) +# --------------------------------------------- +# cost report command +# --------------------------------------------- + + +@cost_app.command(name="report") +def cost_report( + days: int = typer.Option(30, "--days", help="Reporting window in days (must be >= 1)."), + group_by: str = typer.Option( + "framework", "--group-by", help="Group by: framework, agent, or status." + ), + api_url: str = typer.Option("http://localhost:8000", "--api"), + api_key: str | None = API_KEY_OPTION, + as_json: bool = typer.Option(False, "--json", help="Emit JSON instead of a table."), +) -> None: + """ + [bold]Report[/bold] token usage, USD cost, and cost-per-successful-goal. + + Aggregates recent sessions from the AgentWatch API over the last [b]--days[/b], + grouped by [b]--group-by[/b]. + + [b]Example Usage:[/b] + [dim]python -m agentwatch.cli.main cost report --days 30 --group-by framework[/dim] + """ + from agentwatch.cost.reporting import VALID_GROUP_BY, build_cost_report, parse_sessions + + if group_by not in VALID_GROUP_BY: + console.print( + f"[red]Invalid --group-by {group_by!r}. Choose one of: {', '.join(VALID_GROUP_BY)}.[/red]" + ) + raise typer.Exit(2) + if days < 1: + console.print("[red]--days must be >= 1.[/red]") + raise typer.Exit(2) + + limit = 200 # /api/v1/sessions caps its page size at 200 + + async def _fetch() -> list[dict[str, object]]: + try: + import httpx + except ImportError: + console.print("[red]httpx not installed. Run: pip install httpx[/red]") + raise typer.Exit(1) + + async with httpx.AsyncClient() as client: + try: + resp = await client.get( + f"{api_url}/api/v1/sessions", + params={"since_hours": days * 24, "limit": limit}, + headers=_api_headers(api_key), + timeout=15.0, + ) + resp.raise_for_status() + except httpx.HTTPStatusError as exc: + _handle_http_status_error(exc, api_url) + except httpx.HTTPError as exc: + console.print(f"[red]Failed to connect to API at {api_url}: {exc}[/red]") + raise typer.Exit(1) + + payload = resp.json() + items = payload.get("sessions", []) + return list(items) if isinstance(items, list) else [] + + raw_sessions = asyncio.run(_fetch()) + if len(raw_sessions) >= limit: + console.print( + f"[yellow]Note: the API returned its maximum of {limit} sessions, so this " + f"report may be incomplete. Narrow the window with a smaller --days.[/yellow]" + ) + sessions, skipped = parse_sessions(raw_sessions) + if skipped: + console.print(f"[yellow]Skipped {skipped} session record(s) that failed to parse.[/yellow]") + report = build_cost_report(sessions, group_by=group_by, days=days) + + if as_json: + console.print_json(data=report.to_dict()) + return + + _print_cost_report_table(report) + + +def _print_cost_report_table(report: CostReport) -> None: + table = Table( + title=( + f"[bold green]C O S T R E P O R T[/bold green] " + f"[dim](last {report.days}d · by {report.group_by})[/dim]" + ), + box=box.DOUBLE_EDGE, + border_style="bold cyan", + ) + table.add_column(report.group_by.capitalize(), style="bold cyan") + table.add_column("Sessions", justify="right", style="dim white") + table.add_column("Tokens", justify="right", style="green") + table.add_column("USD", justify="right", style="yellow") + table.add_column("Successful", justify="right", style="cyan") + table.add_column("USD / success", justify="right", style="bold yellow") + + for row in report.rows: + cps = row.cost_per_successful_goal + table.add_row( + row.group, + str(row.sessions), + f"{row.total_tokens:,}", + f"${row.total_usd:,.4f}", + str(row.successful), + "—" if cps is None else f"${cps:,.4f}", + ) + + if report.rows: + table.add_section() + table.add_row( + "[bold]TOTAL[/bold]", + str(report.total_sessions), + f"[bold]{report.total_tokens:,}[/bold]", + f"[bold]${report.total_usd:,.4f}[/bold]", + "", + "", + ) + + console.print(table) + if not report.rows: + console.print("[dim]No sessions found in the selected window.[/dim]") + + # --------------------------------------------- # serve command # --------------------------------------------- diff --git a/agentwatch/cost/reporting.py b/agentwatch/cost/reporting.py new file mode 100644 index 00000000..75db95be --- /dev/null +++ b/agentwatch/cost/reporting.py @@ -0,0 +1,166 @@ +"""Aggregate agent sessions into a grouped cost report (issue #339). + +This module is deliberately pure and side-effect free: :func:`build_cost_report` +takes a list of already-time-filtered :class:`AgentSession` summaries and returns +a :class:`CostReport`. Fetching sessions (from the API) and rendering (Rich table +or JSON) live in the CLI layer, which keeps the aggregation trivially testable. + +Design choices (the issue left these open): +* USD comes from the pre-computed ``session.estimated_cost_usd`` — no per-model + pricing lookup is needed at report time. +* A "successful goal" is a session whose stored ``status`` is + ``ExecutionStatus.SUCCESS``. Session summaries do not carry their events, so an + events-based (dual-eval) definition is intentionally out of scope here. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from pydantic import ValidationError + +from agentwatch.core.schema import AgentSession, ExecutionStatus + +#: Session dimensions the report can group by. +VALID_GROUP_BY: tuple[str, ...] = ("framework", "agent", "status") + + +def parse_sessions(items: list[dict[str, object]]) -> tuple[list[AgentSession], int]: + """Validate raw session dicts into :class:`AgentSession`, skipping bad rows. + + The sessions API returns already-serialized sessions, but legacy or partial + records can still fail validation. Rather than crashing the whole report on a + single bad row, invalid records are skipped and counted so the caller can + surface how many were dropped. + + Args: + items: Raw session dictionaries (e.g. from the sessions API response). + + Returns: + A ``(sessions, skipped)`` tuple: the successfully parsed sessions and the + number of records that failed validation. + """ + sessions: list[AgentSession] = [] + skipped = 0 + for item in items: + try: + sessions.append(AgentSession.model_validate(item)) + except ValidationError: + skipped += 1 + return sessions, skipped + + +def _group_key(session: AgentSession, group_by: str) -> str: + if group_by == "framework": + return session.framework.value + if group_by == "agent": + return session.agent_name or session.agent_id + # "status" is the only remaining option; build_cost_report validates group_by + # against VALID_GROUP_BY before this helper is ever reached. + return session.status.value + + +@dataclass +class CostReportRow: + """Aggregated spend for a single group (e.g. one framework).""" + + group: str + sessions: int = 0 + total_tokens: int = 0 + total_usd: float = 0.0 + successful: int = 0 + + @property + def cost_per_successful_goal(self) -> float | None: + """USD spent per successful session, or ``None`` if none succeeded.""" + if self.successful <= 0: + return None + return self.total_usd / self.successful + + def to_dict(self) -> dict[str, Any]: + cps = self.cost_per_successful_goal + return { + "group": self.group, + "sessions": self.sessions, + "total_tokens": self.total_tokens, + "total_usd": round(self.total_usd, 6), + "successful": self.successful, + "cost_per_successful_goal": None if cps is None else round(cps, 6), + } + + +@dataclass +class CostReport: + """A grouped cost report over a reporting window.""" + + group_by: str + days: int + rows: list[CostReportRow] = field(default_factory=list) + + @property + def total_sessions(self) -> int: + return sum(r.sessions for r in self.rows) + + @property + def total_tokens(self) -> int: + return sum(r.total_tokens for r in self.rows) + + @property + def total_usd(self) -> float: + return sum(r.total_usd for r in self.rows) + + def to_dict(self) -> dict[str, Any]: + return { + "group_by": self.group_by, + "days": self.days, + "totals": { + "sessions": self.total_sessions, + "total_tokens": self.total_tokens, + "total_usd": round(self.total_usd, 6), + }, + "rows": [r.to_dict() for r in self.rows], + } + + +def build_cost_report( + sessions: list[AgentSession], + *, + group_by: str = "framework", + days: int = 30, +) -> CostReport: + """Aggregate sessions into a grouped cost report. + + The function does not filter by time — the caller is expected to have already + restricted ``sessions`` to the reporting window. ``days`` is recorded on the + report purely for display. Rows are returned sorted by total USD descending. + + Args: + sessions: Session summaries to aggregate. + group_by: One of :data:`VALID_GROUP_BY` (``framework``/``agent``/``status``). + days: Reporting window in days, carried through for display. + + Returns: + The aggregated :class:`CostReport`. + + Raises: + ValueError: If ``group_by`` is not one of :data:`VALID_GROUP_BY`. + """ + if group_by not in VALID_GROUP_BY: + raise ValueError(f"unsupported group_by: {group_by!r} (expected one of {VALID_GROUP_BY})") + + rows: dict[str, CostReportRow] = {} + for session in sessions: + key = _group_key(session, group_by) + row = rows.get(key) + if row is None: + row = CostReportRow(group=key) + rows[key] = row + row.sessions += 1 + row.total_tokens += session.total_tokens + row.total_usd += session.estimated_cost_usd + if session.status is ExecutionStatus.SUCCESS: + row.successful += 1 + + ordered = sorted(rows.values(), key=lambda r: r.total_usd, reverse=True) + return CostReport(group_by=group_by, days=days, rows=ordered) diff --git a/tests/test_cost_reporting.py b/tests/test_cost_reporting.py new file mode 100644 index 00000000..72821801 --- /dev/null +++ b/tests/test_cost_reporting.py @@ -0,0 +1,157 @@ +"""Tests for the cost-report aggregation (issue #339).""" + +from __future__ import annotations + +import pytest + +from agentwatch.core.schema import AgentFramework, AgentSession, ExecutionStatus +from agentwatch.cost.reporting import build_cost_report, parse_sessions + + +def _session( + *, + framework: AgentFramework = AgentFramework.LANGCHAIN, + status: ExecutionStatus = ExecutionStatus.SUCCESS, + tokens: int = 0, + usd: float = 0.0, + agent_id: str = "a1", + agent_name: str | None = None, +) -> AgentSession: + return AgentSession( + agent_id=agent_id, + agent_name=agent_name, + framework=framework, + status=status, + total_tokens=tokens, + estimated_cost_usd=usd, + ) + + +def test_groups_by_framework_and_sums_tokens_and_usd() -> None: + sessions = [ + _session(framework=AgentFramework.LANGCHAIN, tokens=100, usd=1.0), + _session(framework=AgentFramework.LANGCHAIN, tokens=200, usd=2.0), + _session(framework=AgentFramework.CREWAI, tokens=50, usd=0.5), + ] + report = build_cost_report(sessions, group_by="framework", days=30) + by_group = {r.group: r for r in report.rows} + assert by_group["langchain"].sessions == 2 + assert by_group["langchain"].total_tokens == 300 + assert by_group["langchain"].total_usd == pytest.approx(3.0) + assert by_group["crewai"].total_tokens == 50 + assert report.total_tokens == 350 + assert report.total_usd == pytest.approx(3.5) + + +def test_cost_per_successful_goal() -> None: + sessions = [ + _session(usd=3.0, status=ExecutionStatus.SUCCESS), + _session(usd=3.0, status=ExecutionStatus.SUCCESS), + _session(usd=4.0, status=ExecutionStatus.FAILURE), + ] + report = build_cost_report(sessions, group_by="framework") + (row,) = report.rows + assert row.successful == 2 + # 10 USD total across 2 successful sessions. + assert row.cost_per_successful_goal == pytest.approx(5.0) + + +def test_zero_successful_yields_none_not_divide_error() -> None: + sessions = [ + _session(usd=2.0, status=ExecutionStatus.FAILURE), + _session(usd=1.0, status=ExecutionStatus.BLOCKED), + ] + report = build_cost_report(sessions) + (row,) = report.rows + assert row.successful == 0 + assert row.cost_per_successful_goal is None + + +def test_only_success_status_counts_as_successful() -> None: + non_success = [ + ExecutionStatus.PENDING, + ExecutionStatus.RUNNING, + ExecutionStatus.FAILURE, + ExecutionStatus.BLOCKED, + ExecutionStatus.ROLLED_BACK, + ExecutionStatus.TIMEOUT, + ] + sessions = [_session(status=s) for s in non_success] + sessions.append(_session(status=ExecutionStatus.SUCCESS)) + report = build_cost_report(sessions) + (row,) = report.rows + assert row.successful == 1 + + +def test_group_by_agent_uses_name_then_id() -> None: + sessions = [ + _session(agent_id="id1", agent_name="Researcher", usd=1.0), + _session(agent_id="id2", agent_name=None, usd=2.0), + ] + report = build_cost_report(sessions, group_by="agent") + assert {r.group for r in report.rows} == {"Researcher", "id2"} + + +def test_group_by_status() -> None: + sessions = [ + _session(status=ExecutionStatus.SUCCESS), + _session(status=ExecutionStatus.FAILURE), + _session(status=ExecutionStatus.FAILURE), + ] + report = build_cost_report(sessions, group_by="status") + assert {r.group: r.sessions for r in report.rows} == {"success": 1, "failure": 2} + + +def test_rows_sorted_by_usd_descending() -> None: + sessions = [ + _session(framework=AgentFramework.LANGCHAIN, usd=1.0), + _session(framework=AgentFramework.CREWAI, usd=5.0), + _session(framework=AgentFramework.CLAUDE_CODE, usd=3.0), + ] + report = build_cost_report(sessions, group_by="framework") + usds = [r.total_usd for r in report.rows] + assert usds == sorted(usds, reverse=True) + assert report.rows[0].group == "crewai" + + +def test_empty_sessions_is_empty_report() -> None: + report = build_cost_report([], group_by="framework", days=7) + assert report.rows == [] + assert report.total_sessions == 0 + assert report.total_tokens == 0 + assert report.total_usd == 0.0 + assert report.days == 7 + + +def test_invalid_group_by_raises() -> None: + with pytest.raises(ValueError): + build_cost_report([_session()], group_by="nonsense") + + +def test_report_to_dict_shape() -> None: + report = build_cost_report( + [_session(framework=AgentFramework.CREWAI, tokens=100, usd=2.0)], + group_by="framework", + days=30, + ) + data = report.to_dict() + assert data["group_by"] == "framework" + assert data["days"] == 30 + assert data["totals"] == {"sessions": 1, "total_tokens": 100, "total_usd": 2.0} + assert data["rows"][0]["group"] == "crewai" + assert data["rows"][0]["cost_per_successful_goal"] == 2.0 + + +def test_parse_sessions_keeps_valid_and_skips_malformed() -> None: + good = _session(tokens=10, usd=0.1).model_dump(mode="json") + items = [good, {}, {"foo": "bar"}] # two rows missing the required agent_id + sessions, skipped = parse_sessions(items) + assert len(sessions) == 1 + assert sessions[0].total_tokens == 10 + assert skipped == 2 + + +def test_parse_sessions_empty() -> None: + sessions, skipped = parse_sessions([]) + assert sessions == [] + assert skipped == 0