Lane server tools, lift task notes, parse ANSI

Three streams of data were being lost or mangled in the feed.

WebSearch is not purely client-side: it issues a nested /v1/messages that
declares Anthropic's hosted web_search under the parent session id and with
no agent-id header. Read as a turn start it pushed a fake user prompt,
clobbered the main lane's system/tools signatures and downgraded a [1m]
session to the short window on the next resume. Requests are now classified
three ways (Turn / ServerTool / Side) from the tools array shape alone, and a
nested call gets its own lane, readable only in the A popup. Its result and
citations arrive complete in the stream and are echoed back in no later
request body, so they are attached as the stream delivers them.

An Agent call returns its tool_result immediately ("async agent launched"),
so the real completion is a <task-notification> injected into the parent's
next user turn. Those are lifted out of the prompt: the report moves onto the
Agent entry it answers, the usage totals onto the lane, and the status
becomes one glyph-led note line. A finished agent used to keep reading as
running.

Tool output we do not control carries SGR codes. A self-contained parser maps
them to styles instead of leaving [1m as literal text; filled blocks keep
their own colours and take only the attributes.

Also: a non-2xx upstream response now surfaces as an error entry instead of a
silent stall, tool renderers cover the file/shell/task/prompt/web families,
and the fake upstream answers the nested hosted-tool request so both search
paths run offline.

dev/fake_upstream.py

@@ -8,7 +8,12 @@ ExitPlanMode, TodoWrite/Task*) on demand, so the pane's frame detector in
 `src/term.rs` can be developed against what Ink actually draws.
 
 Scenario is picked per turn from `CT_FAKE_SCENARIO`
-(ask | plan | todo | taskupdate | agent | text).
+(ask | plan | todo | taskupdate | agent | websearch | ansi | text).
+
+`websearch` also answers the *nested* request Claude Code makes to run WebSearch:
+that call declares Anthropic's server-side `web_search` tool, so it is replied to
+with `server_tool_use` + `web_search_tool_result` + `citations_delta` — the block
+types only a hosted tool produces.
 Each incoming request is logged to `dev/fake_upstream.log` (declared tool names
 + the trailing user text) so we can see what CC sends.
 """
@@ -61,6 +66,9 @@ def stream_text(text):
     yield sse("message_stop", {"type": "message_stop"})
 
 
+_TOOL_SEQ = 0
+
+
 def stream_tool(name, tool_input, lead="Working on it."):
     """A turn that calls one client-side tool."""
     yield from stream_tools([(name, tool_input)], lead)
@@ -78,8 +86,13 @@ def stream_tools(calls, lead="Working on it."):
               "delta": {"type": "text_delta", "text": lead}})
     yield sse("content_block_stop", {"type": "content_block_stop", "index": 0})
     for n, (name, tool_input) in enumerate(calls, start=1):
+        # Ids must be unique across the whole session, exactly as the real API
+        # guarantees: Claude Code resends the full history every request, so a
+        # reused id makes an *old* tool_result re-attach to the newest call.
+        global _TOOL_SEQ
+        _TOOL_SEQ += 1
         yield sse("content_block_start", {"type": "content_block_start", "index": n,
-                  "content_block": {"type": "tool_use", "id": f"toolu_fake{n}",
+                  "content_block": {"type": "tool_use", "id": f"toolu_fake{_TOOL_SEQ}",
                                     "name": name, "input": {}}})
         blob = json.dumps(tool_input)
         for i in range(0, len(blob), 40):
@@ -92,6 +105,63 @@ def stream_tools(calls, lead="Working on it."):
     yield sse("message_stop", {"type": "message_stop"})
 
 
+def stream_server_websearch():
+    """What a *hosted* web_search turn looks like: `server_tool_use`, then a
+    complete `web_search_tool_result` block (no deltas — the whole payload
+    rides in `content_block_start`), then cited text. Claude Code issues this
+    nested request itself when it runs the client-side `WebSearch` tool."""
+    yield sse("message_start", {"type": "message_start", "message": {
+        "id": "msg_ws", "type": "message", "role": "assistant", "model": MODEL,
+        "content": [], "stop_reason": None, "stop_sequence": None,
+        "usage": {"input_tokens": 50, "output_tokens": 1}}})
+    yield sse("content_block_start", {"type": "content_block_start", "index": 0,
+              "content_block": {"type": "server_tool_use", "id": "srvtoolu_fake1",
+                                "name": "web_search", "input": {}}})
+    blob = json.dumps({"query": "ratatui scrollbar thumb"})
+    yield sse("content_block_delta", {"type": "content_block_delta", "index": 0,
+              "delta": {"type": "input_json_delta", "partial_json": blob}})
+    yield sse("content_block_stop", {"type": "content_block_stop", "index": 0})
+    yield sse("content_block_start", {"type": "content_block_start", "index": 1,
+              "content_block": {
+                  "type": "web_search_tool_result", "tool_use_id": "srvtoolu_fake1",
+                  "content": [
+                      {"type": "web_search_result", "title": "Ratatui Scrollbar docs",
+                       "url": "https://ratatui.rs/widgets/scrollbar", "page_age": "2 days"},
+                      {"type": "web_search_result", "title": "Scrollbar example",
+                       "url": "https://ratatui.rs/examples/scrollbar", "page_age": None},
+                  ]}})
+    yield sse("content_block_stop", {"type": "content_block_stop", "index": 1})
+    yield sse("content_block_start", {"type": "content_block_start", "index": 2,
+              "content_block": {"type": "text", "text": ""}})
+    for chunk in "Ratatui renders the thumb through its Scrollbar widget. ".split(" "):
+        yield sse("content_block_delta", {"type": "content_block_delta", "index": 2,
+                  "delta": {"type": "text_delta", "text": chunk + " "}})
+    yield sse("content_block_delta", {"type": "content_block_delta", "index": 2,
+              "delta": {"type": "citations_delta", "citation": {
+                  "type": "web_search_result_location",
+                  "url": "https://ratatui.rs/widgets/scrollbar",
+                  "title": "Ratatui Scrollbar docs",
+                  "cited_text": "Scrollbar renders a thumb over the track."}}})
+    yield sse("content_block_stop", {"type": "content_block_stop", "index": 2})
+    yield sse("message_delta", {"type": "message_delta",
+              "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+              "usage": {"output_tokens": 30, "server_tool_use": {"web_search_requests": 1}}})
+    yield sse("message_stop", {"type": "message_stop"})
+
+
+# A command whose *output* carries real SGR codes, so the feed's ANSI handling
+# is exercised by a genuine tool_result rather than a hand-written fixture.
+ANSI_INPUT = {
+    "command": (
+        "printf '\\033[1mbold heading\\033[22m\\n'; "
+        "printf '\\033[31m- removed line\\033[0m\\n'; "
+        "printf '\\033[32m+ added line\\033[0m\\n'; "
+        "printf '\\033[38;5;208m256-colour orange\\033[0m\\n'"
+    ),
+    "description": "Print coloured output",
+}
+
+
 ASK_INPUT = {"questions": [{
     "question": "The compact pane currently crops the top of this prompt. Which framing "
                 "should the pane use when an interactive question is on screen, given that "
@@ -163,7 +233,14 @@ class Handler(BaseHTTPRequestHandler):
             body = json.loads(raw)
         except Exception:
             body = {}
-        tools = [t.get("name") for t in body.get("tools", []) or []]
+        raw_tools = body.get("tools", []) or []
+        tools = [t.get("name") for t in raw_tools]
+        # A hosted tool carries a `type` and no `input_schema`; that is the
+        # nested WebSearch call, not a turn start.
+        hosted = bool(raw_tools) and all(
+            t.get("input_schema") is None and t.get("type") not in (None, "custom")
+            for t in raw_tools
+        )
         msgs = body.get("messages", []) or []
         tail = json.dumps(msgs[-1])[:300] if msgs else ""
         # Only the immediate reply to *our* canned tool call ends the turn with
@@ -192,8 +269,11 @@ class Handler(BaseHTTPRequestHandler):
             return
 
         scenario = read_scenario()
+        # The nested hosted-tool request answers itself, whatever the scenario.
+        if hosted:
+            gen = stream_server_websearch()
         # A request with no tools is CC's side/title call — answer with text.
-        if not tools or has_result:
+        elif not tools or has_result:
             gen = stream_text("Done. Ask me anything else.")
         elif scenario == "ask":
             gen = stream_tool("AskUserQuestion", ASK_INPUT, "Let me check how you want this framed.")
@@ -215,6 +295,11 @@ class Handler(BaseHTTPRequestHandler):
             gen = stream_tools(
                 [("Agent", a) for a in AGENT_INPUTS], "Delegating this."
             )
+        elif scenario == "websearch":
+            gen = stream_tool("WebSearch", {"query": "ratatui scrollbar thumb"},
+                              "Let me search for that.")
+        elif scenario == "ansi":
+            gen = stream_tool("Bash", ANSI_INPUT, "Printing coloured output.")
         elif scenario == "taskupdate":
             gen = stream_tool("TaskUpdate", {"taskId": "1", "status": "in_progress"},
                               "Starting the first task.")
@@ -241,5 +326,5 @@ class Handler(BaseHTTPRequestHandler):
 
 if __name__ == "__main__":
     port = int(sys.argv[1]) if len(sys.argv) > 1 else 9911
-    print(f"fake upstream on 127.0.0.1:{port} scenario={os.environ.get('CT_FAKE_SCENARIO', 'ask')}")
+    print(f"fake upstream on 127.0.0.1:{port} scenario={read_scenario()}")
     ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()