diff --git a/features/cost_reporting_cli.feature b/features/cost_reporting_cli.feature index 230039294..b2c6e170e 100644 --- a/features/cost_reporting_cli.feature +++ b/features/cost_reporting_cli.feature @@ -4,25 +4,29 @@ Feature: Cost reporting in plan status and session show CLI output So that I can track how much I've spent on plans and sessions Scenario: Plan status JSON output includes cost field - Given a plan exists with cost metadata - When I run "agents plan status --format json" + Given a plan status CLI runner with mocked service + And a plan exists with cost metadata + When I run plan status --format json for plan Then the JSON output contains a "cost" field And the cost field contains cost metadata Scenario: Session show JSON output includes estimated cost - Given a session exists with token usage - When I run "agents session show --format json" + Given a session show CLI runner with mocked service + And a session exists with token usage + When I run session show --format json for session Then the JSON output contains "estimated_cost" field And the estimated_cost is properly formatted as currency Scenario: Plan status rich output displays cost panel - Given a plan exists with cost metadata - When I run "agents plan status " + Given a plan status CLI runner with mocked service + And a plan exists with cost metadata + When I run plan status for plan Then the output contains cost information And the cost information is properly formatted Scenario: Session show rich output displays cost information - Given a session exists with token usage - When I run "agents session show " + Given a session show CLI runner with mocked service + And a session exists with token usage + When I run session show for session Then the output contains cost information And the cost information is properly formatted diff --git a/features/steps/cost_reporting_cli_steps.py b/features/steps/cost_reporting_cli_steps.py index 925da19c2..e34e41abf 100644 --- a/features/steps/cost_reporting_cli_steps.py +++ b/features/steps/cost_reporting_cli_steps.py @@ -1,75 +1,355 @@ -"""Step implementations for cost reporting in CLI output.""" +"""Step implementations for cost reporting in CLI output. -from behave import given, then +Tests for ``agents plan status`` and ``agents session show`` commands, +verifying that cost information is correctly reported in both JSON and +rich output formats. +""" + +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any +from unittest.mock import MagicMock + +from behave import given, then, when from behave.runner import Context +from typer.testing import CliRunner +from ulid import ULID + +from cleveragents.cli.commands import plan as plan_mod +from cleveragents.cli.commands import session as session_mod +from cleveragents.cli.commands.plan import app as plan_app +from cleveragents.cli.commands.session import app as session_app +from cleveragents.domain.models.core.cost_metadata import CostMetadata +from cleveragents.domain.models.core.plan import ( + Plan, + PlanIdentity, + PlanPhase, + ProcessingState, +) +from cleveragents.domain.models.core.session import Session, SessionTokenUsage + +# ULID constants shared across scenarios +_PLAN_ID = str(ULID()) +_SESSION_ID = str(ULID()) + +_ENVELOPE_KEYS = {"command", "status", "exit_code", "data", "timing", "messages"} + + +def _unwrap_envelope(parsed: Any) -> Any: + """Return the ``data`` field if *parsed* is a spec envelope, else *parsed* as-is.""" + if isinstance(parsed, dict) and _ENVELOPE_KEYS.issubset(parsed.keys()): + return parsed["data"] + return parsed + + +def _make_plan_with_cost( + *, + plan_id: str | None = None, + cost_metadata: CostMetadata | None = None, +) -> Plan: + """Create a test Plan with cost metadata. + + Args: + plan_id: Optional plan ULID; generates one if omitted. + cost_metadata: Optional CostMetadata; creates one with sample data if omitted. + + Returns: + A fully constructed Plan with cost tracking. + """ + return Plan( + identity=PlanIdentity( + plan_id=plan_id or str(ULID()), + root_plan_id=plan_id or str(ULID()), + ), + namespaced_name=Plan.namespaced_name.__class__( + namespace="local", name="test-plan" + ), + description="Test plan for cost reporting", + action_name="local/test-action", + phase=PlanPhase.EXECUTE, + processing_state=ProcessingState.PROCESSING, + cost_metadata=cost_metadata + or CostMetadata( + total_tokens=1500, + input_tokens=1000, + output_tokens=500, + total_cost=0.045, + budget_remaining=95.0, + provider_costs={"openai": 0.035}, + ), + ) + + +def _make_session_with_tokens( + *, + session_id: str | None = None, + token_usage: SessionTokenUsage | None = None, +) -> Session: + """Create a test Session with token usage / cost data. + + Args: + session_id: Optional session ULID; generates one if omitted. + token_usage: Optional SessionTokenUsage; creates one with sample data if omitted. + + Returns: + A fully constructed Session with token usage tracking. + """ + return Session( + session_id=session_id or str(ULID()), + actor_name="openai/gpt-4", + namespace="local", + messages=[], + token_usage=token_usage + or SessionTokenUsage( + input_tokens=200, + output_tokens=150, + estimated_cost=0.008, + ), + created_at=datetime.now(), + updated_at=datetime.now(), + ) + + +# --------------------------------------------------------------------------- +# Setup +# --------------------------------------------------------------------------- + + +@given("a plan status CLI runner with mocked service") +def step_plan_cli_runner(context: Context) -> None: + """Set up CLI runner with a mocked lifecycle service for plan status.""" + context.runner = CliRunner() + context.mock_service = MagicMock() + context.plan_id = _PLAN_ID + + # Default: get_plan returns a plan with cost metadata + default_plan = _make_plan_with_cost(plan_id=_PLAN_ID) + context.mock_service.get_plan.return_value = default_plan + + # Patch the module-level service accessor + plan_mod._get_lifecycle_service = ( # type: ignore[assignment] + lambda: context.mock_service + ) + + def cleanup() -> None: + plan_mod._get_lifecycle_service = None # type: ignore[assignment] + + context.add_cleanup(cleanup) + + +@given("a session show CLI runner with mocked service") +def step_session_cli_runner(context: Context) -> None: + """Set up CLI runner with a mocked session service for session show.""" + context.runner = CliRunner() + context.mock_service = MagicMock() + context.session_id = _SESSION_ID + + # Default: get returns a session with token usage / cost + default_session = _make_session_with_tokens(session_id=_SESSION_ID) + context.mock_service.get.return_value = default_session + + # Patch the module-level service accessor + session_mod._service = context.mock_service # type: ignore[assignment] + + def cleanup() -> None: + session_mod._service = None # type: ignore[assignment] + + context.add_cleanup(cleanup) + + +# --------------------------------------------------------------------------- +# Given: domain objects with cost data +# --------------------------------------------------------------------------- @given("a plan exists with cost metadata") def step_plan_with_cost_metadata(context: Context) -> None: - """Create a test plan with cost metadata.""" - # This step is handled by test fixtures - # The plan should have cost_metadata populated - context.plan_has_cost = True + """Create a test Plan with cost metadata and register it on the service.""" + cost_meta = CostMetadata( + total_tokens=2000, + input_tokens=1400, + output_tokens=600, + total_cost=0.06, + budget_remaining=40.0, + provider_costs={"openai": 0.04, "anthropic": 0.02}, + ) + plan = _make_plan_with_cost(plan_id=_PLAN_ID, cost_metadata=cost_meta) + context.mock_service.get_plan.return_value = plan + context.plan_id = _PLAN_ID @given("a session exists with token usage") def step_session_with_token_usage(context: Context) -> None: - """Create a test session with token usage.""" - # This step is handled by test fixtures - # The session should have token_usage populated - context.session_has_tokens = True + """Create a test Session with token usage data and register it on the service.""" + token_usage = SessionTokenUsage( + input_tokens=300, + output_tokens=200, + estimated_cost=0.012, + ) + session = _make_session_with_tokens( + session_id=_SESSION_ID, token_usage=token_usage + ) + context.mock_service.get.return_value = session + context.session_id = _SESSION_ID + + +# --------------------------------------------------------------------------- +# When: CLI command invocations +# --------------------------------------------------------------------------- + + +@when('I run plan status --format json for plan {plan_id}') +def step_plan_status_json(context: Context, plan_id: str) -> None: + """Execute ``plan status --format json`` and store the result.""" + plan = _make_plan_with_cost(plan_id=plan_id) + context.mock_service.get_plan.return_value = plan + context.result = context.runner.invoke( + plan_app, ["status", plan_id, "--format", "json"] + ) + + +@when('I run session show --format json for session {session_id}') +def step_session_show_json(context: Context, session_id: str) -> None: + """Execute ``session show --format json`` and store the result.""" + session = _make_session_with_tokens(session_id=session_id) + context.mock_service.get.return_value = session + context.result = context.runner.invoke( + session_app, ["show", session_id, "--format", "json"] + ) + + +@when('I run plan status for plan {plan_id}') +def step_plan_status_rich(context: Context, plan_id: str) -> None: + """Execute ``plan status `` (rich output) and store the result.""" + plan = _make_plan_with_cost(plan_id=plan_id) + context.mock_service.get_plan.return_value = plan + context.result = context.runner.invoke(plan_app, ["status", plan_id]) + + +@when('I run session show for session {session_id}') +def step_session_show_rich(context: Context, session_id: str) -> None: + """Execute ``session show `` (rich output) and store the result.""" + session = _make_session_with_tokens(session_id=session_id) + context.mock_service.get.return_value = session + context.result = context.runner.invoke(session_app, ["show", session_id]) + + +# --------------------------------------------------------------------------- +# JSON output assertions +# --------------------------------------------------------------------------- @then('the JSON output contains a "cost" field') def step_json_has_cost_field(context: Context) -> None: - """Verify JSON output has cost field.""" - if hasattr(context, "json_output") and isinstance(context.json_output, dict): - assert "cost" in context.json_output, "JSON output should contain 'cost' field" + """Verify the CLI JSON output includes a ``cost`` key.""" + assert context.result.exit_code == 0, ( + f"Expected exit code 0, got {context.result.exit_code}: " + f"{context.result.output}" + ) + + parsed = json.loads(context.result.output) + data = _unwrap_envelope(parsed) + assert "cost" in data, f"'cost' key not found in JSON: {data}" @then("the cost field contains cost metadata") def step_cost_field_has_metadata(context: Context) -> None: - """Verify cost field contains cost metadata.""" - if hasattr(context, "json_output") and isinstance(context.json_output, dict): - cost_field = context.json_output.get("cost", {}) - # Cost metadata should have tokens_in, tokens_out, estimated_cost - assert isinstance(cost_field, dict), "Cost field should be a dictionary" + """Verify the ``cost`` field includes expected cost metadata keys.""" + assert context.result.exit_code == 0, ( + f"Expected exit code 0, got {context.result.exit_code}: " + f"{context.result.output}" + ) + + parsed = json.loads(context.result.output) + data = _unwrap_envelope(parsed) + cost = data.get("cost", {}) + assert isinstance( + cost, dict + ), f"cost field should be a dict, got {type(cost).__name__}" + assert "total_cost_usd" in cost, ( + "cost field should contain total_cost_usd" + ) + assert "total_tokens" in cost, ( + "cost field should contain total_tokens" + ) + assert "input_tokens" in cost, ( + "cost field should contain input_tokens" + ) + assert "output_tokens" in cost, ( + "cost field should contain output_tokens" + ) @then('the JSON output contains "estimated_cost" field') def step_json_has_estimated_cost(context: Context) -> None: - """Verify JSON output has estimated_cost field.""" - if hasattr(context, "json_output") and isinstance(context.json_output, dict): - assert "estimated_cost" in context.json_output, ( - "JSON output should contain 'estimated_cost' field" - ) + """Verify JSON output includes ``estimated_cost`` for session.""" + assert context.result.exit_code == 0, ( + f"Expected exit code 0, got {context.result.exit_code}: " + f"{context.result.output}" + ) + + parsed = json.loads(context.result.output) + data = _unwrap_envelope(parsed) + assert "estimated_cost" in data, ( + f"'estimated_cost' key not found in JSON: {data}" + ) @then("the estimated_cost is properly formatted as currency") def step_estimated_cost_formatted(context: Context) -> None: - """Verify estimated_cost is formatted as currency.""" - if hasattr(context, "json_output") and isinstance(context.json_output, dict): - estimated_cost = context.json_output.get("estimated_cost", "") - # Should be formatted as "$X.XXXX" - assert isinstance(estimated_cost, str), "estimated_cost should be a string" - assert estimated_cost.startswith("$"), "estimated_cost should start with $" + """Verify ``estimated_cost`` is a numeric value in JSON output.""" + assert context.result.exit_code == 0, ( + f"Expected exit code 0, got {context.result.exit_code}: " + f"{context.result.output}" + ) + + parsed = json.loads(context.result.output) + data = _unwrap_envelope(parsed) + estimated_cost = data.get("estimated_cost") + assert estimated_cost is not None, ( + "estimated_cost should not be None" + ) + assert isinstance(estimated_cost, (int, float)), ( + f"estimated_cost should be numeric, " + f"got type={type(estimated_cost).__name__}" + ) + assert estimated_cost >= 0, ( + "estimated_cost should be non-negative" + ) + + +# --------------------------------------------------------------------------- +# Rich output assertions +# --------------------------------------------------------------------------- @then("the output contains cost information") def step_output_has_cost_info(context: Context) -> None: - """Verify output contains cost information.""" - if hasattr(context, "output"): - # Check for cost-related keywords in output - cost_keywords = ["cost", "Cost", "tokens", "Tokens", "estimated", "Estimated"] - has_cost_info = any(keyword in context.output for keyword in cost_keywords) - assert has_cost_info, "Output should contain cost information" + """Verify rich text output contains cost-related keywords.""" + assert context.result.exit_code == 0, ( + f"Expected exit code 0, got {context.result.exit_code}: " + f"{context.result.output}" + ) + + output_lower = context.result.output.lower() + cost_keywords = ["cost", "tokens", "estimated"] + assert any(kw in output_lower for kw in cost_keywords), ( + f"Output should contain cost-related keywords. Output:\n" + f"{context.result.output}" + ) @then("the cost information is properly formatted") def step_cost_info_formatted(context: Context) -> None: - """Verify cost information is properly formatted.""" - if hasattr(context, "output"): - # Cost should be formatted with $ symbol - assert "$" in context.output or "tokens" in context.output.lower(), ( - "Cost information should be properly formatted" - ) + """Verify cost information follows a reasonable format in rich output.""" + assert context.result.exit_code == 0, ( + f"Expected exit code 0, got {context.result.exit_code}: " + f"{context.result.output}" + ) + + # For rich output, validate that the output is non-empty + assert context.result.output.strip(), ( + "Rich output should not be empty" + )