From bcdc71383ebf25724516bd6fee4eb4e4a2c0df3a Mon Sep 17 00:00:00 2001
From: teahubxyz <boring+teahub@agentmail.to>
Date: Sun, 18 Oct 2026 04:59:20 +0000
Subject: [PATCH] Claude calls go to Anthropic's own Messages API with prompt
 caching; the ledger prices cached tokens (#120)

Co-authored-by: teahubxyz <boring+teahub@agentmail.to>
---
 .github/time/test_time.py |  60 +++++++++++++++++++++--
 .github/time/time.py      | 100 +++++++++++++++++++++++++++++++++++---
 CHANGELOG.md              |   1 +
 3 files changed, 150 insertions(+), 11 deletions(-)

diff --git a/.github/time/test_time.py b/.github/time/test_time.py
index ae58a6e..21615b6 100644
--- a/.github/time/test_time.py
+++ b/.github/time/test_time.py
@@ -513,9 +513,20 @@ class ClaudeWorld(World):
     def __init__(self, *a, ledger=None, **kw):
         super().__init__(*a, **kw); self.label_ids["andon"] = 99; self.vars = {}
         if ledger: self.files[f"time/ledger/{NOW:%Y-%m}.jsonl"] = "".join(json.dumps(r) + "\n" for r in ledger)
-    def http(self, method, url, token, body=None, timeout=0):
+    def http(self, method, url, token, body=None, timeout=0, headers=None):
         if url.startswith(tm.ANTHROPIC):
-            sent.append((url, token, body)); out = self.venice(body); out["usage"] = {"prompt_tokens": 1000, "completion_tokens": 2000}; return out
+            sent.append((url, token, body))
+            if url.endswith("/messages"):  # the native API: the scripted seat speaks OpenAI, so translate both ways
+                assert headers == {"x-api-key": token, "anthropic-version": tm.ANTHROPIC_VERSION}, headers
+                assert "system" in body and body["messages"][0]["role"] == "user" and "temperature" not in body, body
+                oa = {"model": body["model"], "tools": [{"type": "function", "function": {"name": t["name"], "parameters": t["input_schema"]}} for t in body.get("tools") or []],
+                      "messages": [{"role": "system", "content": body["system"][0]["text"]}] + [{"role": m["role"], "content": "".join(b.get("text") or "" for b in m["content"])} for m in body["messages"]]}
+                r = self.venice(oa)["choices"][0]["message"]
+                blocks = ([{"type": "text", "text": r["content"]}] if r.get("content") else []) + \
+                         [{"type": "tool_use", "id": c["id"], "name": c["function"]["name"], "input": json.loads(c["function"]["arguments"])} for c in r.get("tool_calls") or []]
+                return {"content": blocks, "stop_reason": "tool_use" if r.get("tool_calls") else "end_turn",
+                        "usage": {"input_tokens": 1000, "output_tokens": 2000, **getattr(self, "cache_usage", {})}}
+            out = self.venice(body); out["usage"] = {"prompt_tokens": 1000, "completion_tokens": 2000}; return out
         if "/actions/variables/" in url:
             self.vars[url.rsplit("/", 1)[1]] = (method, body["value"]); return {}
         return super().http(method, url, token, body, timeout)
@@ -528,15 +539,54 @@ w.venice = lambda body: {"choices": [{"message": {"role": "assistant", "content"
 tm.think = real_think
 assert tm.think("sonnet", None, "high", "sys", "u") == "hi"
 url, tok, body = sent[-1]
-assert url == "https://api.anthropic.com/v1/chat/completions" and tok == "ak" and body["model"] == "claude-sonnet-5-5"
+assert url == "https://api.anthropic.com/v1/messages" and tok == "ak" and body["model"] == "claude-sonnet-5-5", (url, body)
+assert body["system"] == [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}] and body["max_tokens"] == 4000
+assert body["messages"] == [{"role": "user", "content": [{"type": "text", "text": "u", "cache_control": {"type": "ephemeral"}}]}], body["messages"]
+assert not {"temperature", "venice_parameters", "reasoning_effort", "top_p", "tools"} & set(body), body
+assert tm.LEDGER[-1]["in"] == 1000 and tm.LEDGER[-1]["out"] == 2000 and tm.LEDGER[-1]["usd"] == 0.022 and "cached" not in tm.LEDGER[-1]  # 1000*2 + 2000*10 per Mtok
+# cached tokens are priced: a read at a tenth, a write at a quarter more; the ledger line says how many
+w.cache_usage = {"cache_read_input_tokens": 800, "cache_creation_input_tokens": 100}
+assert tm.think("sonnet", None, None, "sys", "u") == "hi"
+assert tm.LEDGER[-1]["in"] == 1900 and tm.LEDGER[-1]["cached"] == 800 and tm.LEDGER[-1]["written"] == 100, tm.LEDGER[-1]
+assert tm.LEDGER[-1]["usd"] == round((1000 * 2 + 800 * 0.2 + 100 * 2.5 + 2000 * 10) / 1e6, 6), tm.LEDGER[-1]
+w.cache_usage = {}
+# TIME_ANTHROPIC_API=compat: the OpenAI-compatible endpoint, as before (no caching)
+os.environ["TIME_ANTHROPIC_API"] = "compat"
+assert tm.think("sonnet", None, "high", "sys", "u") == "hi"
+url, tok, body = sent[-1]
+assert url == "https://api.anthropic.com/v1/chat/completions" and body["messages"][0]["role"] == "system"
 assert not {"temperature", "venice_parameters", "reasoning_effort", "top_p"} & set(body), body
-assert tm.LEDGER[-1]["in"] == 1000 and tm.LEDGER[-1]["out"] == 2000 and tm.LEDGER[-1]["usd"] == 0.022  # 1000*2 + 2000*10 per Mtok
+os.environ.pop("TIME_ANTHROPIC_API")
+# the translation itself: tools, tool calls and results, the two cache breakpoints, and a turn's blocks kept verbatim
+tm._turns.clear()
+oa_tools = [{"type": "function", "function": {"name": "read_file", "description": "Read", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}]
+rep = tm.from_messages({"content": [{"type": "thinking", "thinking": "hm", "signature": "sig"}, {"type": "text", "text": "Looking."},
+                                     {"type": "tool_use", "id": "tu1", "name": "read_file", "input": {"path": "a.py"}}],
+                         "stop_reason": "tool_use", "usage": {"input_tokens": 10, "output_tokens": 5, "cache_read_input_tokens": 90}})
+m = rep["choices"][0]["message"]
+assert m["content"] == "Looking." and m["tool_calls"] == [{"id": "tu1", "type": "function", "function": {"name": "read_file", "arguments": '{"path": "a.py"}'}}]
+assert rep["usage"] == {"prompt_tokens": 100, "completion_tokens": 5, "cache_read": 90, "cache_write": 0} and rep["choices"][0]["finish_reason"] == "tool_use"
+req = tm.to_messages({"model": "claude-sonnet-5-5", "max_tokens": 16000, "tools": oa_tools, "messages": [
+    {"role": "system", "content": "rules"}, {"role": "user", "content": "Begin."},
+    {"role": "assistant", "content": m["content"], "tool_calls": m["tool_calls"]},
+    {"role": "tool", "tool_call_id": "tu1", "content": "x = 1"}, {"role": "tool", "tool_call_id": "tu2", "content": "y"}]})
+assert req["tools"] == [{"name": "read_file", "description": "Read", "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}}}]
+assert req["messages"][1]["content"][0] == {"type": "thinking", "thinking": "hm", "signature": "sig"}, "the turn goes back as the API gave it"
+assert [b["type"] for b in req["messages"][1]["content"]] == ["thinking", "text", "tool_use"]
+assert req["messages"][2]["role"] == "user" and [b["tool_use_id"] for b in req["messages"][2]["content"]] == ["tu1", "tu2"], "one user turn carries both results"
+assert "cache_control" in req["messages"][2]["content"][-1] and "cache_control" not in req["messages"][2]["content"][0]
+assert "cache_control" not in req["messages"][1]["content"][-1] and tm._turns["tu1"][-1].get("cache_control") is None, "the kept turn is not marked"
+assert req["system"][0]["cache_control"] == {"type": "ephemeral"} and req["max_tokens"] == 16000 and "temperature" not in req
+tm._turns.clear()
+req = tm.to_messages({"model": "claude-sonnet-5-5", "messages": [{"role": "system", "content": "r"}, {"role": "user", "content": "go"},
+    {"role": "assistant", "content": None, "tool_calls": m["tool_calls"]}, {"role": "tool", "tool_call_id": "tu1", "content": "ok"}]})
+assert req["messages"][1]["content"] == [{"type": "tool_use", "id": "tu1", "name": "read_file", "input": {"path": "a.py"}}], "rebuilt when the turn is not kept"
 os.environ["VENICE_API_KEY"] = "vk"; w.venice = lambda body: {"choices": [{"message": {"content": "ok"}}]}
 assert tm.think("kimi", None, None, "s", "u") == "ok" and tm.LEDGER[-1]["usd"] is None and tm.LEDGER[-1]["in"] == 0  # unpriced, no usage
 # the ledger flushes as one appended commit
 tm._ledger_files.clear(); tm.flush_ledger()
 lines = w.files[f"time/ledger/{NOW:%Y-%m}.jsonl"].splitlines()
-assert len(lines) == 2 and tm.LEDGER == [] and json.loads(lines[0])["usd"] == 0.022
+assert len(lines) == 4 and tm.LEDGER == [] and json.loads(lines[0])["usd"] == 0.022 and json.loads(lines[1])["cached"] == 800
 # over the daily cap: Budget before any request
 sent.clear(); tm._ledger_files.clear()
 w.files[f"time/ledger/{NOW:%Y-%m}.jsonl"] = json.dumps({"at": iso(), "model": "claude-opus-5-5", "usd": 6}) + "\n"
diff --git a/.github/time/time.py b/.github/time/time.py
index fd5101d..752593e 100644
--- a/.github/time/time.py
+++ b/.github/time/time.py
@@ -43,10 +43,10 @@ MARK = "<!-- time "
 EFFORTS = {"low": "low", "med": "medium", "medium": "medium", "high": "high", "xhigh": "high", "max": "high"}
 
 
-def http(method: str, url: str, token: str, body=None, timeout: float = 120.0):
+def http(method: str, url: str, token: str, body=None, timeout: float = 120.0, headers: dict | None = None):
     req = urllib.request.Request(url, method=method, data=json.dumps(body).encode() if body is not None else None,
                                  headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json",
-                                          "Accept": "application/json"})
+                                          "Accept": "application/json", **(headers or {})})
     try:
         with urllib.request.urlopen(req, timeout=timeout) as r:
             t = r.read()
@@ -200,11 +200,11 @@ def price_of(model: str) -> list | None:
                  if s.get("price") and (model == s["default"] or model in s.get("aliases", {}).values())), None)
 
 
-def _post(url: str, key: str, body: dict, timeout: float, who: str) -> dict:
+def _post(url: str, key: str, body: dict, timeout: float, who: str, headers: dict | None = None) -> dict:
     """One retry loop for every provider: brief overloads (429, 5xx, Anthropic's 529) and dropped connections back off."""
     for attempt in range(4):
         try:
-            return http("POST", url, key, body, timeout=timeout)
+            return http("POST", url, key, body, timeout=timeout, **({"headers": headers} if headers else {}))
         except (RuntimeError, OSError) as e:
             transient = isinstance(e, OSError) or re.search(r"-> (429|5\d\d)\b", str(e))
             if attempt == 3 or not transient:
@@ -232,10 +232,93 @@ def anthropic_chat(body: dict, timeout: float) -> dict:
     return _post(f"{ANTHROPIC}/chat/completions", key, body, timeout, "anthropic")
 
 
+ANTHROPIC_VERSION = "2023-06-01"
+_turns: dict[str, list] = {}  # tool_use id -> the assistant turn's content blocks as the API gave them (thinking included)
+
+
+def _args(raw) -> dict:
+    try:
+        out = json.loads(raw or "{}")
+    except ValueError:
+        return {}
+    return out if isinstance(out, dict) else {}
+
+
+def to_messages(body: dict) -> dict:
+    """The engine's OpenAI-shaped request as an Anthropic Messages request, with the prompt marked for caching: the tools
+    and system prompt are one breakpoint and the conversation so far is the other, so each step of a worker run pays for
+    the context the step before it already sent at a tenth of the price. An assistant turn that issued tool calls goes
+    back exactly as the API gave it (thinking blocks included), which the API requires; tool results of one turn travel
+    in one user message."""
+    system, msgs = [], []
+    for m in body["messages"]:
+        role, content = m.get("role"), m.get("content")
+        if role == "system":
+            system.append(content or "")
+        elif role == "user":
+            msgs.append({"role": "user", "content": [{"type": "text", "text": content or ""}]})
+        elif role == "assistant":
+            calls = m.get("tool_calls") or []
+            blocks = list(_turns.get(calls[0]["id"]) or []) if calls else []
+            if not blocks:
+                blocks = ([{"type": "text", "text": content}] if content else []) + [
+                    {"type": "tool_use", "id": c["id"], "name": c["function"]["name"], "input": _args(c["function"].get("arguments"))}
+                    for c in calls]
+            if blocks:
+                msgs.append({"role": "assistant", "content": blocks})
+        elif role == "tool":
+            block = {"type": "tool_result", "tool_use_id": m.get("tool_call_id") or "", "content": str(content or "")}
+            if msgs and msgs[-1]["role"] == "user" and msgs[-1]["content"][-1].get("type") == "tool_result":
+                msgs[-1]["content"].append(block)
+            else:
+                msgs.append({"role": "user", "content": [block]})
+    out = {"model": body["model"], "max_tokens": int(body.get("max_tokens") or 4000), "messages": msgs}
+    if body.get("tools"):
+        out["tools"] = [{"name": t["function"]["name"], "description": t["function"].get("description") or "",
+                         "input_schema": t["function"].get("parameters") or {"type": "object", "properties": {}}} for t in body["tools"]]
+    if system:
+        out["system"] = [{"type": "text", "text": "\n\n".join(system), "cache_control": {"type": "ephemeral"}}]
+    if msgs and msgs[-1]["content"]:  # the last block: next step's request shares everything up to here
+        msgs[-1]["content"][-1] = {**msgs[-1]["content"][-1], "cache_control": {"type": "ephemeral"}}
+    return out
+
+
+def from_messages(out: dict) -> dict:
+    """An Anthropic Messages reply in the OpenAI shape, with the cache split kept in usage for the ledger."""
+    blocks = out.get("content") or []
+    text = "".join(b.get("text") or "" for b in blocks if b.get("type") == "text") or None
+    calls = [{"id": b["id"], "type": "function", "function": {"name": b["name"], "arguments": json.dumps(b.get("input") or {})}}
+             for b in blocks if b.get("type") == "tool_use"]
+    if calls:
+        _turns[calls[0]["id"]] = blocks
+    u = out.get("usage") or {}
+    read, written = int(u.get("cache_read_input_tokens") or 0), int(u.get("cache_creation_input_tokens") or 0)
+    usage = {"prompt_tokens": int(u.get("input_tokens") or 0) + read + written, "completion_tokens": int(u.get("output_tokens") or 0),
+             "cache_read": read, "cache_write": written}
+    msg = {"role": "assistant", "content": text, **({"tool_calls": calls} if calls else {})}
+    return {"choices": [{"message": msg, "finish_reason": out.get("stop_reason")}], "usage": usage}
+
+
+def anthropic_messages(body: dict, timeout: float) -> dict:
+    """POST to Anthropic's own Messages API, with prompt caching (the OpenAI-compatible endpoint has none, and a 40-step
+    worker run re-sent its whole context at full price every step: a failed run cost $2.80). Speaks the engine's OpenAI
+    shape on both sides. TIME_ANTHROPIC_API=compat goes back to the compatible endpoint."""
+    key = os.environ.get("ANTHROPIC_API_KEY")
+    if not key:
+        raise LookupError("no ANTHROPIC_API_KEY secret on this repo yet")
+    out = _post(f"{ANTHROPIC}/messages", key, to_messages(body), timeout, "anthropic",
+                headers={"x-api-key": key, "anthropic-version": ANTHROPIC_VERSION})
+    return from_messages(out)
+
+
 def chat(body: dict, timeout: float) -> dict:
     """Every model call goes through here: the budget is checked first and the spend booked after."""
     budget_check()
-    out = (anthropic_chat if provider_of(body["model"]) == "anthropic" else venice_chat)(body, timeout)
+    if provider_of(body["model"]) == "anthropic":
+        native = (os.environ.get("TIME_ANTHROPIC_API") or "messages").strip().lower() != "compat"
+        out = (anthropic_messages if native else anthropic_chat)(body, timeout)
+    else:
+        out = venice_chat(body, timeout)
     book(body["model"], out.get("usage") or {})
     return out
 
@@ -273,12 +356,16 @@ def ledger_lines(month: str) -> list[dict]:
 
 
 def book(model: str, usage: dict) -> None:
+    """One ledger line per call. `in` counts every prompt token; `cached` says how many of them were read from the cache
+    (a tenth of the input price) and `written` how many were written to it (a quarter more than the input price)."""
     p = price_of(model)
     i, o = int(usage.get("prompt_tokens") or 0), int(usage.get("completion_tokens") or 0)
+    r, w = int(usage.get("cache_read") or 0), int(usage.get("cache_write") or 0)
     if not LEDGER:  # first line: every entry point (heartbeat, mention, venture) flushes without remembering to
         atexit.register(flush_ledger)
+    usd = round(((i - r - w) * p[0] + r * p[0] * 0.1 + w * p[0] * 1.25 + o * p[1]) / 1e6, 6) if p else None
     LEDGER.append({"at": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "model": model, "in": i, "out": o,
-                   "usd": round((i * p[0] + o * p[1]) / 1e6, 6) if p else None})
+                   "usd": usd, **({"cached": r, "written": w} if r or w else {})})
 
 
 def flush_ledger() -> None:
@@ -1259,6 +1346,7 @@ def cmd_doctor() -> int:
     row("spend", today < day_cap and month < month_cap, f"${today:.2f} today, ${month:.2f} this month ({ledger_path(dt.datetime.now(dt.timezone.utc).strftime('%Y-%m'))})")
     row("(optional) ANTHROPIC_API_KEY (Claude seats)", bool(os.environ.get("ANTHROPIC_API_KEY")),
         "set" if os.environ.get("ANTHROPIC_API_KEY") else "not set")
+    row("TIME_ANTHROPIC_API", True, os.environ.get("TIME_ANTHROPIC_API") or "messages (default, with prompt caching; compat = the OpenAI-compatible endpoint)")
     row("TIME_AUTOMERGE", True, (os.environ.get("TIME_AUTOMERGE") or "off (default)") + "; merges on Approve + green checks")
     for name in ("queued", "spec", "needs-human"):
         row(f"(optional) label `{name}`", name in labels, "exists" if name in labels else "missing (created on first use)")
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c9685c5..882d32c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@
 - The Venice worker's shell no longer sees secrets: its environment drops tokens and keys, its checkout keeps no token in `.git/config`, and the workflows check out without persisted credentials.
 
 ### Changed
+- Claude calls go to Anthropic's own Messages API with prompt caching. The tools and system prompt are one cache breakpoint and the conversation so far the other, so each step of a worker run pays for the context the step before already sent at a tenth of the price (the OpenAI-compatible endpoint has no caching: a failed 43-step run cost $2.80, most of it the same context re-sent 43 times). The engine still speaks the OpenAI shape inside; an assistant turn that issued tool calls goes back exactly as the API gave it. The ledger prices cached tokens (`cached`, `written` on the line). `TIME_ANTHROPIC_API=compat` goes back to the old endpoint without a code change.
 - The second fix round escalates to Kimi's code model (`venice:kimi code`) instead of grok 4.7: still a different model from the first round, at a fraction of the cost. `TIME_VENTURE_ESCALATE` overrides it.
 
 ### Added