phase: 160_tool_budget_telemetry

All green. Final report:

**Phase 160 completion pass — verified (final verification; all 5 tasks already in `complete/`)**

- **Verified, no code defects found:** budget trailers (`TOOL_BUDGET_REMAINING/LAST/SPENT`), cap-path rejection + `ROUND_CAP_FALLBACK_ANSWER` + `ROUND_CAP_WRAPUP_INSTRUCTION`, `done.budget_exhausted` (ABSENT-when-false), `rounds=/cap_hit=` log fields, `agent_max_rounds=10` default, frontend Continue button + restore + shared-page exclusion, mock markers.
- **Tests/lint:** `uv run pytest` 153/153 green (EXIT 0); `uv run pytest --cov=app --cov-report=term-missing` TOTAL **99%** (>90%); `uv run ruff check .` clean; `uv run pyright` 0 errors/0 warnings.
- **Criteria:** ① decreasing trailer pinned by unit tests (9→…→spent) ✅ ② cap path ends readable, no tool lines after cap (E2E) ✅ ③ `budget_exhausted` true-only-on-cap, absent otherwise (integration) ✅ ④ Continue button → one fresh `continue` turn, fresh budget, record intact (E2E) ✅ ⑤ flag survives refresh; shared chat never renders it (`shared.js` has no `budget_exhausted` path) ✅ ⑥ `max_rounds=0` byte-identical single `tools=None` request; default still 10 ✅ ⑦ full suite/coverage/lint green ✅ ⑧ `test_tool_budget_telemetry.py` **7/7** in isolation; all 6 regression E2E suites green in isolation ✅ ⑨ commit + phase move = harness (I make no git history/index changes).
- **Docs:** README.md — fixed: the "Continue researching" button and cap behavior were undocumented (added a Chat-section paragraph + an Agent-tools-section paragraph, incl. the phase-71 recorded-revision note); env-var table, commands, and architecture all verified consistent otherwise. AGENTS.md — verified consistent, no changes needed.
- **Decisions:** no code changes required — phase was fully implemented; only README docs added. No live-infrastructure touched.
- **Next pending phase:** `161_reasoning_bubble`.

tests/integration/test_chat_api.py

@@ -82,6 +82,7 @@ class FakeRagLLM:
         stream_error: Exception | None = None,
         fail_mid_stream: bool = False,
         tool_script: list[list[StreamPiece | ToolCallPiece]] | None = None,
+        no_tools_script: list[list[StreamPiece | ToolCallPiece]] | None = None,
         embed_fail_count: int = 0,
         stream_fail_count: int = 0,
         answer_sequence: list[str] | None = None,
@@ -123,6 +124,18 @@ class FakeRagLLM:
         #: thinking + answer stream below, so a deflected turn through this
         #: fake is byte-identical to the plain fake's output.
         self.tool_script: list[list[StreamPiece | ToolCallPiece]] = list(tool_script or [])
+        #: Phase 160 (task 02): canned piece lists for requests made
+        #: WITHOUT tools (``tools=None``) — the deflected turn, the
+        #: cap-forced final, the kill-switch single-request path, and the
+        #: phase-71 recovery. Empty (the default) keeps the pre-phase
+        #: behavior: a tools=None request yields the thinking + answer
+        #: stream below, so every existing test is untouched. When set,
+        #: the *i*-th tools=None request consumes ``no_tools_script[i]``
+        #: (so a mock can make a tools=None request return structured
+        #: ``ToolCallPiece``s — the documented ``lite`` behavior).
+        self.no_tools_script: list[list[StreamPiece | ToolCallPiece]] = list(
+            no_tools_script or []
+        )
 
     async def embed(
         self,
@@ -197,6 +210,8 @@ class FakeRagLLM:
         raw: list[StreamPiece | ToolCallPiece]
         if tools is not None and self.tool_script:
             raw = self.tool_script.pop(0)
+        elif tools is None and self.no_tools_script:
+            raw = self.no_tools_script.pop(0)
         elif self.fail_mid_stream:
             raw = [StreamPiece("content", "partial ")]
             mid_stream_drop = True
@@ -980,9 +995,10 @@ marker in the model's context."""
             m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
         ]
         assert tool_msgs, "the executed read must be appended as a tool message"
-        assert (
-            tool_msgs[-1]["content"]
-            == "Document docs/fits.md:\ndate: 2024-06-15\n" + content
+        # Phase 160 (task 01): the budget-telemetry trailer rides the
+        # read result — the raw result is the RAW part.
+        assert tool_msgs[-1]["content"].startswith(
+            "Document docs/fits.md:\ndate: 2024-06-15\n" + content
         )
         assert TRUNCATION_MARKER not in tool_msgs[-1]["content"]
         # Still a successful read.
@@ -1165,9 +1181,10 @@ def test_untruncated_read_streams_no_tool_result_frame(
         m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
     ]
     assert tool_msgs
-    assert (
-        tool_msgs[-1]["content"]
-        == "Document docs/big-read.md:\ndate: 2024-06-15\n" + content
+    # Phase 160 (task 01): the budget-telemetry trailer rides the
+    # read result — the raw result is the RAW part.
+    assert tool_msgs[-1]["content"].startswith(
+        "Document docs/big-read.md:\ndate: 2024-06-15\n" + content
     )
     assert TRUNCATION_MARKER not in tool_msgs[-1]["content"]
 
@@ -1500,6 +1517,176 @@ def test_zero_max_rounds_reproduce_pre_phase_single_request(
     assert "scaffold_stripped=0" in lines[-1]  # phase 71: uniform clean-turn field
 
 
+def test_cap_path_rejects_late_calls_and_fallbacks(
+    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A grounded turn with ``agent_max_rounds=1`` whose forced final
+    (``tools=None``) returns a ``ToolCallPiece`` (the documented
+    ``lite`` behavior — a no-tools request that still comes back with
+    structured calls): the executed round streams ``tool`` ×1, then
+    NOTHING of the sort — no second ``tool`` frame, no ``error`` frame —
+    and the turn settles with ``done`` whose delta text is the
+    ``ROUND_CAP_FALLBACK_ANSWER`` (the model gave no visible content on
+    the forced final). The late call is never executed and never counted
+    in ``tool_calls`` (the executed-only log field stays 1)."""
+    scripted = FakeRagLLM(
+        tool_script=[[ToolCallPiece(id="call_1", name="ls", arguments={})]],
+        no_tools_script=[
+            [
+                ToolCallPiece(
+                    id="call_f1",
+                    name="read",
+                    arguments={"path": "docs/homelab/kubernetes.md"},
+                )
+            ]
+        ],
+    )
+    live = get_settings()
+    monkeypatch.setattr(
+        chat_api,
+        "get_settings",
+        lambda: Settings(
+            _env_file=None,  # pyright: ignore[reportCallIssue]
+            relevance_threshold=live.relevance_threshold,
+            agent_max_rounds=1,
+        ),
+    )
+    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
+    try:
+        caplog.set_level(logging.INFO, logger="app.chat")
+        _, _, frames = _stream_chat(client, QUESTION)
+    finally:
+        fastapi_app.dependency_overrides.clear()
+    types = [f["type"] for f in frames]
+    # Only the executed round 1 call streams as a tool frame; the forced
+    # final's call is rejected — no second tool frame, no error frame.
+    assert types.count("tool") == 1
+    assert "error" not in types
+    # The turn settles with done; the answer is the budget fallback.
+    done = frames[-1]
+    assert done["type"] == "done"
+    assert done["deflected"] is False
+    delta_text = "".join(f.get("text", "") for f in frames if f["type"] == "delta")
+    assert delta_text == agent.ROUND_CAP_FALLBACK_ANSWER
+    # The late ``read`` was never executed: ``tool_calls=1`` (the
+    # executed round only) in the per-turn log line.
+    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
+    assert lines and "tool_calls=1" in lines[-1]
+
+
+def test_cap_reached_turn_reports_budget_exhausted_and_log_fields(
+    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A grounded turn with ``agent_max_rounds=1`` that hits the cap:
+    the ``done`` frame carries ``"budget_exhausted": true`` (the
+    additive key — the ``data:`` line the client sees) and the
+    per-turn log line reads ``rounds=1 cap_hit=1`` (the cap hit,
+    named — the owner can grep how often the cap bites)."""
+    scripted = FakeRagLLM(
+        tool_script=[[ToolCallPiece(id="call_1", name="ls", arguments={})]],
+        no_tools_script=[[StreamPiece("content", "the wrap-up answer")]],
+    )
+    live = get_settings()
+    monkeypatch.setattr(
+        chat_api,
+        "get_settings",
+        lambda: Settings(
+            _env_file=None,  # pyright: ignore[reportCallIssue]
+            relevance_threshold=live.relevance_threshold,
+            agent_max_rounds=1,
+        ),
+    )
+    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
+    try:
+        caplog.set_level(logging.INFO, logger="app.chat")
+        _, _, frames = _stream_chat(client, QUESTION)
+    finally:
+        fastapi_app.dependency_overrides.clear()
+    done = frames[-1]
+    assert done["type"] == "done"
+    assert done["deflected"] is False
+    assert done["budget_exhausted"] is True  # the cap-reached flag
+    assert "error" not in [f["type"] for f in frames]
+    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
+    assert lines and "rounds=1 cap_hit=1" in lines[-1]
+    assert "tool_calls=1" in lines[-1]
+
+
+def test_in_cap_and_deflected_turns_report_no_budget_exhausted(
+    client, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An ordinary grounded turn (answered in-cap) and a deflected
+    turn: the ``done`` frames carry NO ``budget_exhausted`` key
+    (ABSENT, never ``false`` — the byte-identical-off contract), and
+    the per-turn log lines read ``cap_hit=0`` — the grounded one
+    with ``rounds=`` the executed rounds (2), the deflected one
+    ``rounds=0 cap_hit=0`` (the agent never runs)."""
+    scripted = FakeRagLLM(
+        tool_script=[
+            [ToolCallPiece(id="call_1", name="ls", arguments={})],
+            [ToolCallPiece(id="call_2", name="ls", arguments={})],
+        ]
+    )
+    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
+    try:
+        caplog.set_level(logging.INFO, logger="app.chat")
+        _, _, frames = _stream_chat(client, QUESTION)
+        _, _, deflected_frames = _stream_chat(client, OFF_TOPIC)
+    finally:
+        fastapi_app.dependency_overrides.clear()
+    done = frames[-1]
+    assert done["type"] == "done"
+    assert done["deflected"] is False
+    assert "budget_exhausted" not in done  # ABSENT, never false
+    d_done = deflected_frames[-1]
+    assert d_done["type"] == "done"
+    assert d_done["deflected"] is True
+    assert "budget_exhausted" not in d_done
+    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
+    assert len(lines) == 2
+    assert "tool_calls=2 rounds=2 cap_hit=0" in lines[0]  # the grounded turn
+    assert "rounds=0 cap_hit=0" in lines[1]  # the deflected turn
+
+
+def test_deflected_turn_rejects_structured_calls(
+    client, db, seeded_kb: FakeRagLLM
+) -> None:
+    """A deflected turn (``tools=None``, A8) whose model returns a
+    structured ``tool_calls`` (the documented ``lite`` behavior) emits
+    NO ``tool`` frame — the call is rejected at the transport
+    (``_pump`` drops it, ``tools_offered=False``), never rendered, never
+    executed. The turn still settles (``done``) with the model's
+    content (the rejection only drops the call, not the answer)."""
+    scripted = FakeRagLLM(
+        no_tools_script=[
+            [
+                ToolCallPiece(id="call_1", name="ls", arguments={}),
+                StreamPiece("content", "the deflected answer"),
+            ]
+        ],
+    )
+    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
+    try:
+        _, _, frames = _stream_chat(client, OFF_TOPIC)
+    finally:
+        fastapi_app.dependency_overrides.clear()
+    types = [f["type"] for f in frames]
+    assert not any(t == "tool" for t in types)  # no tool frame
+    assert "error" not in types
+    done = frames[-1]
+    assert done["type"] == "done"
+    assert done["deflected"] is True
+    # The model's content still streamed (the rejection only drops the
+    # call, not the answer).
+    delta_text = "".join(f.get("text", "") for f in frames if f["type"] == "delta")
+    assert delta_text == "the deflected answer"
+    # One request, no tools key (the deflected direct path, A8).
+    assert len(scripted.seen_messages) == 1
+    assert scripted.seen_tools == [None]
+
+
 def test_tool_execution_db_failure_yields_error_event(
     client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
 ) -> None: