Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 24 additions & 4 deletions src/copilot_usage/parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -795,12 +795,32 @@ def _first_pass(events: list[SessionEvent]) -> _FirstPassResult:

for idx, ev in enumerate(events):
etype = ev.type
# Fast path: once tool_model is found, skip all tool-complete events
# with a single None-check instead of traversing the full elif chain.
# Once tool_model is found, skip all tool-complete events with a
# single None-check instead of traversing the full elif chain.
if etype == EventType.TOOL_EXECUTION_COMPLETE:
if tool_model is None:
m = ev.data.get("model")
if isinstance(m, str) and m:
try:
m = ev.as_tool_execution().model
except ValidationError as exc:
logger.debug(
Comment on lines +802 to +805
"event {} — could not parse {} event ({}), skipping",
idx,
ev.type,
exc.error_count(),
)
m = None
except ValueError as exc:
# Defensive: currently unreachable because the etype
# guard above prevents type mismatches, but caught so
# a refactor of the guard can't let ValueError escape.
logger.debug(
"event {} — could not parse {} event ({}), skipping",
idx,
ev.type,
exc,
)
m = None
if m:
tool_model = m
continue
if etype not in _FIRST_PASS_EVENT_TYPES:
Expand Down
89 changes: 70 additions & 19 deletions tests/copilot_usage/test_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -3785,13 +3785,13 @@ def test_session_shutdown_validation_error_logs_debug(self, tmp_path: Path) -> N
for msg in log_messages
)

def test_tool_execution_complete_bad_data_silently_skipped(
self, tmp_path: Path
) -> None:
"""Bad tool.execution_complete data → silently skipped (no Pydantic validation).
def test_tool_execution_complete_bad_data_logs_debug(self, tmp_path: Path) -> None:
"""Bad tool.execution_complete data → debug log emitted.

The optimised path reads ``ev.data.get("model")`` directly; non-string
or missing values are ignored without raising or logging.
The typed ``as_tool_execution()`` accessor raises ``ValidationError``
for malformed payloads; the handler logs a debug message and skips
the event, matching the pattern used for session.start and
session.shutdown.
"""
from loguru import logger

Expand All @@ -3818,7 +3818,7 @@ def test_tool_execution_complete_bad_data_silently_skipped(

# Malformed event is harmlessly ignored; summary still builds.
assert summary is not None
assert not any(
assert any(
"could not parse" in msg and "tool.execution_complete" in msg
for msg in log_messages
)
Expand Down Expand Up @@ -6782,28 +6782,79 @@ def test_malformed_tool_event_skipped(self, tmp_path: Path) -> None:
fp = _first_pass(events)
assert fp.tool_model == "gpt-5.1"

def test_no_pydantic_validation_for_tool_model(self) -> None:
"""Optimised path reads model via dict lookup, not Pydantic validation.
def test_first_pass_tool_model_from_typed_accessor(self) -> None:
"""_first_pass extracts tool_model via the typed as_tool_execution() accessor.

Builds 1 000 TOOL_EXECUTION_COMPLETE events without a ``model`` key
(worst-case) and asserts that ``ToolExecutionData.model_validate`` is
never called — proving the hot loop avoids the Pydantic round-trip.
Creates a TOOL_EXECUTION_COMPLETE event with a valid ``model`` value
and asserts that ``_first_pass`` populates ``tool_model`` correctly
through the typed ``ToolExecutionData`` accessor rather than raw
dict access.
"""
events: list[SessionEvent] = [
SessionEvent(
type=EventType.TOOL_EXECUTION_COMPLETE,
data={"toolCallId": f"tc-{i}", "success": True},
id=f"ev-tool-{i}",
data={
"toolCallId": "tc-typed",
"model": "claude-sonnet-4",
"success": True,
},
id="ev-typed",
timestamp=None,
parentId=None,
)
for i in range(1_000)
),
]
fp = _first_pass(events)
assert fp.tool_model == "claude-sonnet-4"

def test_first_pass_tool_model_validation_error(self) -> None:
"""Malformed data that passes isinstance(str) but fails typed validation.

Creates a TOOL_EXECUTION_COMPLETE event whose ``model`` is a valid
string (would pass the old ``isinstance(m, str)`` check) but whose
``toolTelemetry`` is invalid, causing ``as_tool_execution()`` to
raise ``ValidationError``. Asserts ``tool_model`` remains ``None``
and no exception escapes.
"""
events: list[SessionEvent] = [
SessionEvent(
type=EventType.TOOL_EXECUTION_COMPLETE,
data={
"toolCallId": "tc-bad",
"model": "gpt-5.1",
"toolTelemetry": "not-a-valid-telemetry-object",
},
id="ev-bad",
timestamp=None,
parentId=None,
),
]
fp = _first_pass(events)
assert fp.tool_model is None

def test_first_pass_tool_model_value_error(self) -> None:
"""Defensive ValueError handler keeps tool_model None.

The ``except ValueError`` branch in ``_first_pass`` is currently
unreachable because the ``etype`` guard prevents type mismatches,
but it exists so a future refactor cannot let ``ValueError`` escape.
We exercise it by patching ``as_tool_execution`` to raise
``ValueError``.
"""
events: list[SessionEvent] = [
SessionEvent(
type=EventType.TOOL_EXECUTION_COMPLETE,
data={"toolCallId": "tc-ve", "model": "gpt-5.1"},
id="ev-ve",
timestamp=None,
parentId=None,
),
]
with patch.object(
ToolExecutionData, "model_validate", wraps=ToolExecutionData.model_validate
) as mock_validate:
SessionEvent,
"as_tool_execution",
side_effect=ValueError("simulated type mismatch"),
):
fp = _first_pass(events)
assert mock_validate.call_count == 0
assert fp.tool_model is None


Expand Down
Loading