fix(extensions): skip AdvancedSQLiteSession usage store when the branch has no turn

src/agents/extensions/memory/advanced_sqlite_session.py

@@ -1895,16 +1895,24 @@ async def _update_turn_usage_internal(
             branch_id: The branch the turn was read from. Defaults to the current
                 branch when not provided.
             turn_anchor: The id of the turn's first ``message_structure`` row,
-                captured when the turn was read. When provided, the write is
-                skipped unless that exact row still exists for the given
-                branch/turn, so usage is never recorded against a turn that was
-                removed — even if a new turn reused the same numeric id. Because
-                the check is scoped to this branch/turn, unrelated removals (e.g.
-                delete_branch on another branch) do not drop this write.
+                captured when the turn was read. The write is skipped unless that
+                exact row still exists for the given branch/turn, so usage is
+                never recorded against a turn that was removed — even if a new
+                turn reused the same numeric id. Because the check is scoped to
+                this branch/turn, unrelated removals (e.g. delete_branch on
+                another branch) do not drop this write. ``None`` means the branch
+                had no turn when it was read, so there is nothing to attribute
+                the usage to and the write is skipped.
         """
 
         target_branch = branch_id if branch_id is not None else self._current_branch_id
 
+        if turn_anchor is None:
+            # ``_capture_current_turn`` returns no anchor only when the branch has no
+            # turn rows; recording usage would invent a phantom turn 0.
+            self._logger.debug("Skipping usage store: no current turn on branch %r", target_branch)
+            return
+
         def _update_sync():
             """Synchronous helper to update turn usage data."""
             with self._write_connection() as conn:

tests/extensions/memory/test_advanced_sqlite_session.py

@@ -3442,6 +3442,46 @@ async def test_store_run_usage_survives_unrelated_branch_deletion(usage_data: Us
         session.close()
 
 
+async def test_store_run_usage_skips_when_current_branch_has_no_turn(usage_data: Usage):
+    """A branch without any turn rows has no turn to attribute a run's usage to, so
+    store_run_usage skips the write instead of recording a phantom turn 0.
+    """
+    session = AdvancedSQLiteSession(session_id="usage_no_turn_test", create_tables=True)
+
+    try:
+        # A fresh session has no turn on the current branch.
+        await session.store_run_usage(create_mock_run_result(usage_data))
+        assert await session.get_session_usage() is None
+        assert await session.get_turn_usage() == []
+
+        # A branch whose only turn was popped away has no turn either.
+        await session.add_items(
+            [
+                {"role": "user", "content": "u1"},
+                {"role": "assistant", "content": "a1"},
+            ]
+        )
+        await session.pop_item()
+        await session.pop_item()
+        await session.store_run_usage(create_mock_run_result(usage_data))
+        assert await session.get_session_usage() is None
+        assert _count_rows(session, "turn_usage") == 0
+
+        # Once a real turn exists, usage is recorded against it.
+        await session.add_items([{"role": "user", "content": "u2"}])
+        second_usage = Usage(requests=2, input_tokens=20, output_tokens=5, total_tokens=25)
+        await session.store_run_usage(create_mock_run_result(second_usage))
+        session_usage = await session.get_session_usage()
+        assert session_usage is not None
+        assert session_usage["requests"] == 2
+        assert session_usage["total_turns"] == 1
+        turn_usage = await session.get_turn_usage()
+        assert isinstance(turn_usage, list)
+        assert [row["user_turn_number"] for row in turn_usage] == [1]
+    finally:
+        session.close()
+
+
 async def test_clear_session_resets_current_branch_to_main():
     """Regression: clear_session must reset the in-memory branch pointer to 'main'
     (inside the locked operation) since every branch was removed.