ba6b18e7d9
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.
331 lines
16 KiB
Python
331 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline stand-in for api.anthropic.com — zero API usage.
|
|
|
|
Point the proxy at it (`CT_UPSTREAM=http://127.0.0.1:9911`) and every request a
|
|
real `claude` child makes is answered locally with a canned SSE stream. That
|
|
lets us make Claude Code render its client-side tool UIs (AskUserQuestion,
|
|
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 | 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.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
LOG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fake_upstream.log")
|
|
MODEL = "claude-opus-4-5-20251101"
|
|
|
|
|
|
def log(msg):
|
|
with open(LOG, "a") as f:
|
|
f.write(msg + "\n")
|
|
|
|
|
|
def read_scenario():
|
|
"""Scenario for the *next* turn: `dev/.fake_scenario` wins over the env, so
|
|
you can switch scenarios without restarting the server."""
|
|
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".fake_scenario")
|
|
try:
|
|
with open(path) as f:
|
|
return f.read().strip()
|
|
except OSError:
|
|
return os.environ.get("CT_FAKE_SCENARIO", "ask")
|
|
|
|
|
|
def sse(event, data):
|
|
return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode()
|
|
|
|
|
|
def stream_text(text):
|
|
"""A plain assistant text turn."""
|
|
yield sse("message_start", {"type": "message_start", "message": {
|
|
"id": "msg_fake", "type": "message", "role": "assistant", "model": MODEL,
|
|
"content": [], "stop_reason": None, "stop_sequence": None,
|
|
"usage": {"input_tokens": 12, "output_tokens": 1}}})
|
|
yield sse("content_block_start", {"type": "content_block_start", "index": 0,
|
|
"content_block": {"type": "text", "text": ""}})
|
|
for chunk in text.split(" "):
|
|
yield sse("content_block_delta", {"type": "content_block_delta", "index": 0,
|
|
"delta": {"type": "text_delta", "text": chunk + " "}})
|
|
yield sse("content_block_stop", {"type": "content_block_stop", "index": 0})
|
|
yield sse("message_delta", {"type": "message_delta",
|
|
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
|
|
"usage": {"output_tokens": 9}})
|
|
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)
|
|
|
|
|
|
def stream_tools(calls, lead="Working on it."):
|
|
"""A turn that calls several client-side tools in one message."""
|
|
yield sse("message_start", {"type": "message_start", "message": {
|
|
"id": "msg_fake", "type": "message", "role": "assistant", "model": MODEL,
|
|
"content": [], "stop_reason": None, "stop_sequence": None,
|
|
"usage": {"input_tokens": 12, "output_tokens": 1}}})
|
|
yield sse("content_block_start", {"type": "content_block_start", "index": 0,
|
|
"content_block": {"type": "text", "text": ""}})
|
|
yield sse("content_block_delta", {"type": "content_block_delta", "index": 0,
|
|
"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{_TOOL_SEQ}",
|
|
"name": name, "input": {}}})
|
|
blob = json.dumps(tool_input)
|
|
for i in range(0, len(blob), 40):
|
|
yield sse("content_block_delta", {"type": "content_block_delta", "index": n,
|
|
"delta": {"type": "input_json_delta", "partial_json": blob[i:i + 40]}})
|
|
yield sse("content_block_stop", {"type": "content_block_stop", "index": n})
|
|
yield sse("message_delta", {"type": "message_delta",
|
|
"delta": {"stop_reason": "tool_use", "stop_sequence": None},
|
|
"usage": {"output_tokens": 40}})
|
|
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 "
|
|
"the question text itself can wrap over several rows?",
|
|
"header": "Framing",
|
|
"multiSelect": False,
|
|
"options": [
|
|
{"label": "Measure the box", "description": "Frame from the question box's own top border down to the input box, so nothing is cut."},
|
|
{"label": "Fixed 75% height", "description": "Always give the pane three quarters of the screen while a question waits, however tall it really is."},
|
|
{"label": "Estimate from JSON", "description": "Keep guessing the row count from the tool input, as today, and accept that wrapped text breaks the guess."},
|
|
{"label": "Fullscreen the pane", "description": "Take the whole screen for as long as a question is waiting for an answer."},
|
|
]}, {
|
|
"question": "And what should happen when the prompt is taller than the pane can ever be?",
|
|
"header": "Overflow",
|
|
"multiSelect": False,
|
|
"options": [
|
|
{"label": "Keep the selection", "description": "Slide the window so the highlighted option stays visible."},
|
|
{"label": "Crop the tail", "description": "Always top-anchor and let the last options fall off."},
|
|
]}]}
|
|
|
|
PLAN_INPUT = {"plan": "## Plan\n\n1. Size the PTY to the whole screen.\n2. Measure the "
|
|
"rendered box instead of guessing rows.\n3. Extend the frame over an "
|
|
"active task list.\n\nThis is a long enough plan to need several rows."}
|
|
|
|
TODO_INPUT = {"todos": [
|
|
{"content": "Capture ground truth screens", "status": "completed", "activeForm": "Capturing ground truth screens"},
|
|
{"content": "Fix interactive pane sizing", "status": "in_progress", "activeForm": "Fixing interactive pane sizing"},
|
|
{"content": "Expand pane while a task list is active", "status": "pending", "activeForm": "Expanding pane for task lists"},
|
|
{"content": "Add regression tests", "status": "pending", "activeForm": "Adding regression tests"},
|
|
{"content": "Update CLAUDE.md", "status": "pending", "activeForm": "Updating CLAUDE.md"},
|
|
]}
|
|
|
|
|
|
TASK_INPUTS = [
|
|
{"subject": "Capture ground truth screens", "description": "Drive a real claude child offline.",
|
|
"activeForm": "Capturing ground truth screens"},
|
|
{"subject": "Fix interactive pane sizing", "description": "Measure the box instead of guessing rows.",
|
|
"activeForm": "Fixing interactive pane sizing"},
|
|
{"subject": "Expand pane while a task list is active", "description": "Show task status near the prompt.",
|
|
"activeForm": "Expanding pane for task lists"},
|
|
{"subject": "Add regression tests", "description": "Pin the framing math.",
|
|
"activeForm": "Adding regression tests"},
|
|
]
|
|
|
|
|
|
# Two parallel subagents with the *same* type and byte-identical prompts: the
|
|
# worst case for correlation (only Claude Code's `x-claude-code-agent-id` tells
|
|
# them apart), and what makes the subagent rows worth having.
|
|
AGENT_INPUTS = [
|
|
{"subagent_type": "Explore", "description": "find the retry helper",
|
|
"prompt": "Locate the retry helper and report back.", "run_in_background": False},
|
|
{"subagent_type": "Explore", "description": "find the retry helper",
|
|
"prompt": "Locate the retry helper and report back.", "run_in_background": False},
|
|
{"subagent_type": "oracle", "description": "review the lane design",
|
|
"prompt": "Review the lane design and report back.", "run_in_background": True},
|
|
]
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def log_message(self, *a): # keep stdout clean
|
|
pass
|
|
|
|
def do_POST(self):
|
|
n = int(self.headers.get("content-length", 0))
|
|
raw = self.rfile.read(n) if n else b"{}"
|
|
try:
|
|
body = json.loads(raw)
|
|
except Exception:
|
|
body = {}
|
|
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
|
|
# text. A user message can carry both the tool_result and fresh user
|
|
# text (a new prompt typed after declining), and that is a new turn —
|
|
# so require the tool_result to be the *whole* last message.
|
|
last = msgs[-1] if msgs else {}
|
|
blocks = last.get("content") if isinstance(last.get("content"), list) else []
|
|
has_result = bool(blocks) and all(
|
|
b.get("type") == "tool_result" for b in blocks if isinstance(b, dict)
|
|
) and "toolu_fake" in json.dumps(blocks)
|
|
# `model` + `anthropic-beta` show how a `--model opus[1m]` pick reaches
|
|
# the wire (the 1M context window is a beta header, not a model id).
|
|
log(
|
|
f"[{time.strftime('%H:%M:%S')}] {self.path} model={body.get('model')} "
|
|
f"betas={self.headers.get('anthropic-beta', '')} tools={tools} tail={tail}"
|
|
)
|
|
|
|
if "count_tokens" in self.path:
|
|
out = json.dumps({"input_tokens": 100}).encode()
|
|
self.send_response(200)
|
|
self.send_header("content-type", "application/json")
|
|
self.send_header("content-length", str(len(out)))
|
|
self.end_headers()
|
|
self.wfile.write(out)
|
|
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.
|
|
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.")
|
|
elif scenario == "plan":
|
|
gen = stream_tool("ExitPlanMode", PLAN_INPUT, "Here is the plan.")
|
|
elif scenario == "todo":
|
|
# Which task tool exists depends on the child's agent: a plain CC
|
|
# session has TodoWrite, an agent like `minimal` has the newer
|
|
# TaskCreate/TaskUpdate set instead.
|
|
if "TodoWrite" in tools:
|
|
gen = stream_tool("TodoWrite", TODO_INPUT, "Setting up the task list.")
|
|
else:
|
|
gen = stream_tools(
|
|
[("TaskCreate", t) for t in TASK_INPUTS], "Setting up the task list."
|
|
)
|
|
elif scenario == "agent":
|
|
# Spawn subagents. A real child then issues their requests itself,
|
|
# each carrying `x-claude-code-agent-id`.
|
|
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.")
|
|
else:
|
|
gen = stream_text("Hello from the fake upstream.")
|
|
|
|
self.send_response(200)
|
|
self.send_header("content-type", "text/event-stream")
|
|
self.send_header("cache-control", "no-cache")
|
|
self.send_header("transfer-encoding", "chunked")
|
|
self.end_headers()
|
|
for part in gen:
|
|
self.wfile.write(b"%x\r\n" % len(part) + part + b"\r\n")
|
|
self.wfile.flush()
|
|
time.sleep(0.02)
|
|
self.wfile.write(b"0\r\n\r\n")
|
|
self.wfile.flush()
|
|
|
|
def do_GET(self):
|
|
self.send_response(404)
|
|
self.send_header("content-length", "0")
|
|
self.end_headers()
|
|
|
|
|
|
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={read_scenario()}")
|
|
ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()
|