feat(compression): per-image token cost learned from the provider's own usage (#70328, supersedes #70463)

A flat per-image constant (1500 in the trigger estimator, 1600 in the tail-budget walk) is wrong in
both directions: a screenshot costs ~1,100 tokens on one provider and 4,000+ on a local mmproj
model. In a GUI loop on a 64K window the estimate sat at ~20K while the real prompt passed 80K,
so compaction never fired and the provider rejected every request (#70328).

The provider prices every image exactly on the request that carries it, so the cost is
observable from usage alone, with no vendor formula: with a fresh usage anchor, the residual
between the next real prompt_tokens and anchor + text-only delta is the price of the N images
that delta introduced.

- agent/image_token_cost.py: calibrate_from_usage() runs in record_response_usage before the new
  anchor is captured; the learned value (EMA, plausibility-banded) is kept per model@host in
  ~/.hermes/cache/image_token_costs.json and bound per turn through a ContextVar.
- estimate_messages_tokens_rough, _content_length_for_budget (tail walk) and gateway hygiene all
  read the same bound value, so trigger and walk agree; the per-message memo now caches text
  tokens and image COUNT so a recalibration re-prices cached rows.
- One flat default (1500) remains only until the first vision turn; the duplicate 1600 is gone.

evals/token_accounting/ab_image_cost_calibration.py (real AIAgent, fake provider pricing images
at 4,000, one screenshot per turn, 64K window): main learns nothing (1500) and the tail walk
under-prices its own protected tail by 56.5%; this branch learns 4,374 after one vision turn
and the walk's error is +8.5%.

Reporter and first-fix credit: @JonthanaHanh (#70328, #70463).

agent/context_compressor.py

@@ -954,10 +954,6 @@ def _collect_protected_skill_names(messages: List[Dict[str, Any]], prune_boundar
 
 
 _CHARS_PER_TOKEN = 4
-# Flat per-image token estimate (realistic ceiling; matches Claude Code's constant).
-_IMAGE_TOKEN_ESTIMATE = 1600
-# Same figure in char-budget currency.
-_IMAGE_CHAR_EQUIVALENT = _IMAGE_TOKEN_ESTIMATE * _CHARS_PER_TOKEN
 _SUMMARY_FAILURE_COOLDOWN_SECONDS = 600
 
 # Fallback handoff preserves continuity anchors only, not a transcript copy.
@@ -1072,14 +1068,18 @@ def _bullets(items: list[str], limit: int = 8) -> str:
 
 
 def _content_length_for_budget(raw_content: Any) -> int:
-    """Effective char-length of message content for budgeting: text by length plus ``_IMAGE_CHAR_EQUIVALENT`` per image."""
+    """Effective char-length of message content for budgeting: text by length plus the learned
+    per-image price (``agent.image_token_cost``, same figure the trigger estimator uses) per image."""
     if isinstance(raw_content, str):
         return len(raw_content)
     if not isinstance(raw_content, list):
         return len(str(raw_content or ""))
+    from agent.image_token_cost import current_image_token_cost
+
+    image_chars = current_image_token_cost() * _CHARS_PER_TOKEN
     # Any text-bearing part counts its text; image_url payload size is irrelevant.
     return sum(
-        (_IMAGE_CHAR_EQUIVALENT if _is_image_part(p) else len(p.get("text", "") or "")) if isinstance(p, dict) else len(str(p))
+        (image_chars if _is_image_part(p) else len(p.get("text", "") or "")) if isinstance(p, dict) else len(str(p))
         for p in raw_content
     )
 

agent/image_token_cost.py

@@ -0,0 +1,135 @@
+"""Per-image token cost learned from the provider's own usage, never from a vendor formula.
+
+A flat per-image constant is wrong in both directions: a 1920x1080 screenshot costs ~1,100 tokens
+on one provider and 4,000+ on a local mmproj model. The provider prices every image exactly on the
+request that carries it, so the cost is observable: with a fresh usage anchor (real prompt count of
+the previous response), the residual between the next real ``prompt_tokens`` and
+``anchor + text-only delta`` is the price of the N images that delta introduced (#70328).
+
+The learned value is kept per ``model@host`` in ``~/.hermes/cache/image_token_costs.json`` so a new
+session starts calibrated, and bound per turn through a ContextVar so every estimator
+(preflight trigger, tail-budget walk, gateway hygiene) prices images the same way.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+from contextvars import ContextVar
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_IMAGE_TOKEN_COST = 1500
+# Observations outside this band are text-estimate noise, not an image price.
+_MIN_PLAUSIBLE, _MAX_PLAUSIBLE = 64, 32_768
+_EMA_ALPHA = 0.5
+
+_image_cost_var: ContextVar[Optional[int]] = ContextVar("hermes_image_token_cost", default=None)
+_LEARNED: Dict[str, int] = {}
+_LOADED = False
+
+
+def _cache_path():
+    from agent.model_metadata import _cache_file
+
+    return _cache_file("image_token_costs.json")
+
+
+def _key(model: Any, base_url: Any) -> str:
+    from utils import base_url_hostname
+
+    return f"{model or ''}@{base_url_hostname(base_url or '') or ''}"
+
+
+def _load() -> None:
+    global _LOADED
+    if _LOADED:
+        return
+    _LOADED = True
+    from agent.model_metadata import _load_json_dict
+
+    for k, v in _load_json_dict(_cache_path()).items():
+        if isinstance(v, int) and _MIN_PLAUSIBLE <= v <= _MAX_PLAUSIBLE:
+            _LEARNED[k] = v
+
+
+def learned_image_token_cost(model: Any, base_url: Any) -> int:
+    """Learned per-image cost for ``model@host``, else the flat default."""
+    _load()
+    return _LEARNED.get(_key(model, base_url), DEFAULT_IMAGE_TOKEN_COST)
+
+
+def current_image_token_cost() -> int:
+    """Per-image cost bound for the running turn (see :func:`image_cost_context`), else the default."""
+    bound = _image_cost_var.get()
+    return bound if bound is not None else DEFAULT_IMAGE_TOKEN_COST
+
+
+@contextlib.contextmanager
+def image_cost_context(cost: Optional[int]):
+    token = _image_cost_var.set(cost)
+    try:
+        yield
+    finally:
+        _image_cost_var.reset(token)
+
+
+def bind_image_token_cost(agent: Any) -> None:
+    """Bind the agent's learned per-image cost to the current context for the rest of the turn."""
+    _image_cost_var.set(learned_image_token_cost(getattr(agent, "model", None), getattr(agent, "base_url", None)))
+
+
+def count_images(messages: List[Dict[str, Any]]) -> int:
+    from agent.model_metadata import _count_image_tokens
+
+    return sum(_count_image_tokens(m, 1) for m in messages if isinstance(m, dict))
+
+
+def calibrate_from_usage(agent: Any, messages: List[Dict[str, Any]], prompt_tokens: Any) -> Optional[int]:
+    """Learn the per-image cost from the response that just priced ``messages``.
+
+    Requires the PREVIOUS anchor (real count of the prior request) to still match: the residual
+    ``prompt_tokens - (anchor + text-only delta)`` is then the provider's price for the images the
+    delta introduced. Returns the new learned cost, or None when this response teaches nothing
+    (no anchor, no new images, implausible residual)."""
+    from agent.usage_anchor import anchored_context_tokens
+
+    anchor = getattr(agent, "_usage_anchor", None)
+    try:
+        real = int(prompt_tokens or 0)
+    except (TypeError, ValueError):
+        return None
+    if real <= 0 or not isinstance(anchor, dict) or not isinstance(messages, list):
+        return None
+    base_count = int(anchor.get("base_count") or 0)
+    delta = messages[base_count:]
+    if delta and isinstance(delta[0], dict) and delta[0].get("role") == "assistant":
+        delta = delta[1:]
+    n_images = count_images(delta)
+    if n_images <= 0:
+        return None
+    with image_cost_context(0):
+        text_only = anchored_context_tokens(messages, anchor)
+    if text_only is None:
+        return None
+    per_image = (real - text_only) // n_images
+    if not _MIN_PLAUSIBLE <= per_image <= _MAX_PLAUSIBLE:
+        return None
+    key = _key(getattr(agent, "model", None), getattr(agent, "base_url", None))
+    _load()
+    prior = _LEARNED.get(key)
+    learned = per_image if prior is None else int(prior + _EMA_ALPHA * (per_image - prior))
+    _LEARNED[key] = learned
+    _image_cost_var.set(learned)
+    try:
+        from utils import atomic_json_write
+
+        atomic_json_write(_cache_path(), dict(_LEARNED), indent=0, separators=(",", ":"))
+    except Exception:
+        logger.debug("image token cost persist failed", exc_info=True)
+    logger.info(
+        "Image token cost calibrated from provider usage: %s images priced %s tokens each (learned %s for %s)",
+        n_images, f"{per_image:,}", f"{learned:,}", key,
+    )
+    return learned

agent/model_metadata.py

@@ -1977,15 +1977,18 @@ def estimate_tokens_rough(text: str) -> int:
 
 
 def estimate_messages_tokens_rough(messages: List[Dict[str, Any]], *, charge_stale_thinking: bool = True) -> int:
-    """Rough token estimate for a message list (pre-flight only). Images cost a flat ~1500 tokens
-    each rather than their base64 length. ``charge_stale_thinking=False`` mirrors the tail-budget
+    """Rough token estimate for a message list (pre-flight only). Images cost the per-image price
+    learned from provider usage (``agent.image_token_cost``; flat default before calibration)
+    rather than their base64 length. ``charge_stale_thinking=False`` mirrors the tail-budget
     walk (``context_compressor._estimate_msg_budget_tokens``): on non-echo routes stale reasoning
     rides the wire only for the NEWEST assistant turn, so excluding it keeps the compaction TRIGGER
     in the same size class as the walk — otherwise reasoning-heavy sessions fire preflight forever."""
-    _IMAGE_TOKEN_COST = 1500
+    from agent.image_token_cost import current_image_token_cost
+
+    image_cost = current_image_token_cost()
     if not charge_stale_thinking:
         messages = _strip_stale_thinking_for_estimate(messages)
-    return sum(_estimate_message_tokens_cached(msg, _IMAGE_TOKEN_COST) for msg in messages)
+    return sum(_estimate_message_tokens_cached(msg, image_cost) for msg in messages)
 
 
 # Thinking-text keys replayed for at most the newest assistant turn on non-echo routes — must stay
@@ -2020,7 +2023,7 @@ def _strip_stale_thinking_for_estimate(messages: List[Dict[str, Any]]) -> List[D
 # estimate. Because the api_messages build shallow-copies history dicts each iteration, the copies share the
 # same content strings — so unchanged history messages hit the memo even though the outer dicts are fresh
 # objects every turn.
-_MSG_TOKENS_CACHE: Dict[Any, Tuple[list, int]] = {}
+_MSG_TOKENS_CACHE: Dict[Any, Tuple[list, int, int]] = {}  # pins, text tokens, image count
 _MSG_TOKENS_CACHE_MAX = 4096
 
 
@@ -2041,19 +2044,23 @@ def _msg_fingerprint(value: Any, pins: list) -> Any:
 
 
 def _estimate_message_tokens_cached(msg: Any, image_cost: int) -> int:
-    def _compute() -> int:
-        return _estimate_message_tokens_without_images(msg) + _count_image_tokens(msg, image_cost)
+    """Text tokens + images x ``image_cost``; the memo holds text and image COUNT so a recalibrated
+    per-image price re-prices cached rows without invalidating them."""
+    def _compute() -> Tuple[int, int]:
+        return _estimate_message_tokens_without_images(msg), _count_image_tokens(msg, 1)
     try:
         pins: list = []
         key = _msg_fingerprint(msg, pins)
         hash(key)
     except Exception:
-        return _compute()
+        text, images = _compute()
+        return text + images * image_cost
     cached = _MSG_TOKENS_CACHE.get(key)
     if cached is not None:
-        return cached[1]
-    tokens = _compute()
-    _MSG_TOKENS_CACHE[key] = (pins, tokens)
+        return cached[1] + cached[2] * image_cost
+    text, images = _compute()
+    tokens = text + images * image_cost
+    _MSG_TOKENS_CACHE[key] = (pins, text, images)
     while len(_MSG_TOKENS_CACHE) > _MSG_TOKENS_CACHE_MAX:
         try:
             _MSG_TOKENS_CACHE.pop(next(iter(_MSG_TOKENS_CACHE)))

agent/turn_context.py

@@ -23,6 +23,7 @@ from agent.memory_manager import build_memory_context_block
 from agent.memory_provider import is_trivial_prompt
 from agent.message_metadata import append_message, stamp_message_timestamp
 from agent.model_metadata import estimate_messages_tokens_rough, estimate_request_tokens_rough
+from agent.image_token_cost import bind_image_token_cost
 from agent.usage_anchor import anchored_context_tokens, restore_usage_anchor
 
 logger = logging.getLogger(__name__)
@@ -834,6 +835,8 @@ def build_turn_context(
         persist_user_platform_id, persist_user_display_kind, persist_user_display_metadata,
     )
     _hydrate_from_history(agent, conversation_history)
+    # Every estimator this turn prices images at the cost learned from this model's real usage.
+    bind_image_token_cost(agent)
     # Append the user message now that close persistence is safe.
     append_message(messages, user_msg)
     current_turn_user_idx = len(messages) - 1

agent/turn_usage.py

@@ -15,6 +15,7 @@ from contextlib import suppress
 from dataclasses import dataclass
 from typing import Any, Dict, List
 
+from agent.image_token_cost import calibrate_from_usage
 from agent.usage_anchor import capture_usage_anchor, set_usage_anchor
 from agent.usage_pricing import estimate_usage_cost, normalize_usage
 
@@ -119,6 +120,9 @@ def record_response_usage(
     # transcript (main-loop ONLY; MoA uses pre-fold aggregator usage). The display meter
     # anchors on the turn's FIRST response: later same-turn responses inflate
     # prompt_tokens with replayed thinking. Display-only; compression math uses real usage.
+    # The provider just priced this request exactly: if the delta since the previous anchor
+    # introduced images, the residual is their real per-image cost (learned before re-anchoring).
+    calibrate_from_usage(agent, messages, aggregator_usage.prompt_tokens)
     _new_anchor = capture_usage_anchor(
         aggregator_usage.prompt_tokens, aggregator_usage.output_tokens, messages
     )

evals/token_accounting/ab_image_cost_calibration.py

@@ -0,0 +1,189 @@
+"""A/B probe: does the compaction trigger learn the real per-image cost from provider usage? (#70328)
+
+Real ``AIAgent.run_conversation`` against a local fake chat-completions server whose
+``usage.prompt_tokens`` prices every image at ``IMAGE_REAL`` tokens (a multimodal local model:
+several thousand per screenshot) while the flat default is 1,500. A GUI loop appends one
+screenshot per turn on a small window; the question is whether compaction fires BEFORE the real
+prompt crosses the provider window (the fake returns a context-overflow 400 past it, like
+llama.cpp), and what the estimator believes when it does.
+
+    python evals/token_accounting/ab_image_cost_calibration.py --out /tmp/result.json
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+CONTEXT_LENGTH = 65_536
+THRESHOLD = 40_000
+IMAGE_REAL = 4_000
+TEXT_PER_TURN = 400  # real tokens of text the provider sees per turn (system + user + reply)
+TURNS = 14
+
+
+def _count_images(messages) -> int:
+    n = 0
+    for m in messages:
+        c = m.get("content")
+        if isinstance(c, list):
+            n += sum(1 for p in c if isinstance(p, dict) and p.get("type") in ("image_url", "image", "input_image"))
+    return n
+
+
+class _FakeVisionChat:
+    """Prices a request as text_turns*TEXT_PER_TURN + images*IMAGE_REAL; 400s past the window."""
+
+    def __init__(self) -> None:
+        self.requests: list[dict] = []
+        self.overflows = 0
+        server = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *_a):  # noqa: D401
+                pass
+
+            def do_POST(self):
+                n = int(self.headers.get("content-length", 0))
+                body = json.loads(self.rfile.read(n) or b"{}")
+                msgs = body.get("messages") or []
+                server.requests.append(body)
+                images = _count_images(msgs)
+                prompt = len(msgs) * TEXT_PER_TURN // 2 + images * IMAGE_REAL
+                if os.environ.get("AB_DEBUG"):
+                    sys.stderr.write(f"SERVER msgs={len(msgs)} images={images} prompt={prompt} roles={[m.get('role') for m in msgs]} types={[type(m.get('content')).__name__ for m in msgs]}\n")
+                if prompt > CONTEXT_LENGTH:
+                    server.overflows += 1
+                    data = json.dumps({"error": {"message": f"request ({prompt} tokens) exceeds the available context size ({CONTEXT_LENGTH} tokens)", "type": "exceed_context_size"}}).encode()
+                    self.send_response(400)
+                    self.send_header("content-type", "application/json")
+                    self.send_header("content-length", str(len(data)))
+                    self.end_headers()
+                    self.wfile.write(data)
+                    return
+                usage = {"prompt_tokens": prompt, "completion_tokens": 5, "total_tokens": prompt + 5}
+                if body.get("stream") is True:
+                    self.send_response(200)
+                    self.send_header("content-type", "text/event-stream")
+                    self.end_headers()
+                    for chunk in (
+                        {"id": "m", "object": "chat.completion.chunk", "model": "m",
+                         "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}, "finish_reason": None}]},
+                        {"id": "m", "object": "chat.completion.chunk", "model": "m",
+                         "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], "usage": usage},
+                    ):
+                        self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
+                    self.wfile.write(b"data: [DONE]\n\n")
+                    self.wfile.flush()
+                    return
+                data = json.dumps({
+                    "id": "x", "object": "chat.completion", "created": 0, "model": body.get("model", "m"),
+                    "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
+                    "usage": usage,
+                }).encode()
+                self.send_response(200)
+                self.send_header("content-type", "application/json")
+                self.send_header("content-length", str(len(data)))
+                self.end_headers()
+                self.wfile.write(data)
+
+        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+        self.base_url = f"http://127.0.0.1:{self.server.server_address[1]}/v1"
+
+    def close(self) -> None:
+        self.server.shutdown()
+        self.server.server_close()
+
+
+def _screenshot_turn(i: int) -> list:
+    return [{"type": "text", "text": f"screenshot {i}, click next"},
+            {"type": "image_url", "image_url": {"url": "data:image/png;base64," + "A" * 2_000}}]
+
+
+def run(out_path: str) -> dict:
+    from run_agent import AIAgent
+
+    wire = _FakeVisionChat()
+    agent = AIAgent(
+        api_key="k", base_url=wire.base_url, provider="custom", model="vision-local-ab",
+        quiet_mode=True, skip_context_files=True, skip_memory=True, enabled_toolsets=[], max_iterations=3,
+    )
+    agent.compression_enabled = True
+    cc = agent.context_compressor
+    cc.context_length = CONTEXT_LENGTH
+    cc.threshold_tokens = THRESHOLD
+    compress_calls: list[dict] = []
+    original = agent._compress_context
+
+    def counting(messages, system_message, **kw):
+        # Real compaction would need a summarizer; drop everything but the last 2 rows like one.
+        compress_calls.append({"turn": len(wire.requests), "approx_tokens": int(kw.get("approx_tokens") or 0),
+                               "images_in_history": _count_images(messages), "real_prompt_if_sent": len(messages) * TEXT_PER_TURN // 2 + _count_images(messages) * IMAGE_REAL})
+        kept = [{"role": "user", "content": "[compressed summary]"}, {"role": "assistant", "content": "ok"}] + messages[-2:]
+        sys_prompt = system_message.get("content") if isinstance(system_message, dict) else system_message
+        return kept, kw.get("active_system_prompt") or sys_prompt
+
+    agent._compress_context = counting  # type: ignore[method-assign]
+    from agent.image_token_cost import current_image_token_cost, learned_image_token_cost
+    history: list = []
+    per_turn = []
+    try:
+        for i in range(TURNS):
+            r = agent.run_conversation(_screenshot_turn(i), conversation_history=history)
+            history = r["messages"]
+            per_turn.append({"turn": i + 1, "completed": bool(r.get("completed")), "images": _count_images(history),
+                             "provider_overflows_so_far": wire.overflows, "compress_calls_so_far": len(compress_calls)})
+    finally:
+        wire.close()
+    # Compaction sizing: how many screenshots does the protected tail keep under the tail budget?
+    # With images priced under their real cost the walk protects too many rows and the
+    # "compacted" request is still over the window (the #70328 "cannot compress further" loop).
+    from agent.image_token_cost import image_cost_context
+    walk_history = [m for i in range(40) for m in ({"role": "user", "content": _screenshot_turn(i)}, {"role": "assistant", "content": "ok"})]
+    tail_budget = int(THRESHOLD * cc.summary_target_ratio)
+    with image_cost_context(learned_image_token_cost("vision-local-ab", wire.base_url)):
+        cut = cc._find_tail_cut_by_tokens(walk_history, 0, token_budget=tail_budget)
+    tail = walk_history[cut:]
+    tail_real = _count_images(tail) * IMAGE_REAL + len(tail) * TEXT_PER_TURN // 2
+    result = {
+        "head": subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True).stdout.strip(),
+        "image_real_cost": IMAGE_REAL, "context_length": CONTEXT_LENGTH, "threshold": THRESHOLD,
+        "learned_image_cost_after": learned_image_token_cost("vision-local-ab", wire.base_url),
+        "provider_overflows": wire.overflows, "compress_calls": compress_calls, "per_turn": per_turn,
+        "tail_budget": tail_budget, "tail_images_kept": _count_images(tail), "tail_real_tokens": tail_real,
+        # The tail walk has an 8-row hard floor, so a screenshot-per-row tail can exceed the budget by
+        # design; the invariant is that the walk's own accounting of that tail tracks the provider's.
+        "tail_walk_estimate_error_pct": round(100 * (_count_images(tail) * learned_image_token_cost("vision-local-ab", wire.base_url) + len(tail) * TEXT_PER_TURN // 2 - tail_real) / tail_real, 1),
+        "verdict": "PASS" if wire.overflows == 0 and compress_calls and abs(learned_image_token_cost("vision-local-ab", wire.base_url) - IMAGE_REAL) / IMAGE_REAL < 0.15 else "FAIL",
+    }
+    Path(out_path).write_text(json.dumps(result, indent=2), encoding="utf-8")
+    return result
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--out", required=True)
+    args = ap.parse_args()
+    tmp = Path(tempfile.mkdtemp(prefix="ab-image-cost-"))
+    os.environ["HERMES_HOME"] = str(tmp / "home")
+    (tmp / "home").mkdir(parents=True)
+    # An unknown custom model is treated as non-vision (images replaced by text); declare it.
+    (tmp / "home" / "config.yaml").write_text("model:\n  supports_vision: true\n", encoding="utf-8")
+    result = run(args.out)
+    print(json.dumps({k: v for k, v in result.items() if k != "per_turn"}, indent=2))
+    return 0 if result["verdict"] == "PASS" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

gateway/run_turn.py

@@ -621,19 +621,22 @@ class GatewayTurnMixin:
         # row (real count + delta of what was appended since, survives gateway restarts), else the
         # rough estimate (runs 30-50% high, which only fires hygiene early — safe). Do NOT compensate
         # with a threshold multiplier.
+        from agent.image_token_cost import image_cost_context, learned_image_token_cost
         _anchored = None
-        if session_entry.last_prompt_tokens <= 0:
-            from agent.usage_anchor import persisted_anchor_tokens
-            _session_db = getattr(self, "_session_db", None)
-            _anchored = persisted_anchor_tokens(
-                getattr(_session_db, "_db", _session_db), session_entry.session_id, history,
-            )
-        if session_entry.last_prompt_tokens > 0:
-            _approx_tokens, _token_source = session_entry.last_prompt_tokens, "actual"
-        elif _anchored is not None:
-            _approx_tokens, _token_source = _anchored, "anchored"
-        else:
-            _approx_tokens, _token_source = estimate_messages_tokens_rough(history), "estimated"
+        # Images in any local delta/estimate are priced at the cost learned from this model's usage.
+        with image_cost_context(learned_image_token_cost(hs.model, hs.base_url)):
+            if session_entry.last_prompt_tokens <= 0:
+                from agent.usage_anchor import persisted_anchor_tokens
+                _session_db = getattr(self, "_session_db", None)
+                _anchored = persisted_anchor_tokens(
+                    getattr(_session_db, "_db", _session_db), session_entry.session_id, history,
+                )
+            if session_entry.last_prompt_tokens > 0:
+                _approx_tokens, _token_source = session_entry.last_prompt_tokens, "actual"
+            elif _anchored is not None:
+                _approx_tokens, _token_source = _anchored, "anchored"
+            else:
+                _approx_tokens, _token_source = estimate_messages_tokens_rough(history), "estimated"
 
         # Hard safety valve: force compression at an extreme message count regardless of tokens,
         # breaking the disconnect → no token data → no compression spiral. 5000 clears 1M+ sessions.

tests/agent/test_compressor_image_tokens.py

@@ -9,12 +9,8 @@ creative workflows that iterate on images across many turns.
 from __future__ import annotations
 
 
-from agent.context_compressor import (
-    _CHARS_PER_TOKEN,
-    _IMAGE_CHAR_EQUIVALENT,
-    _IMAGE_TOKEN_ESTIMATE,
-    _content_length_for_budget,
-)
+from agent.context_compressor import _CHARS_PER_TOKEN, _content_length_for_budget
+from agent.image_token_cost import DEFAULT_IMAGE_TOKEN_COST, image_cost_context
 
 
 class TestContentLengthForBudget:
@@ -36,17 +32,13 @@ class TestContentLengthForBudget:
 
 
 
-    def test_image_estimate_constant_is_reasonable(self):
-        """Sanity-check the estimate aligns with real provider billing.
-
-        Anthropic ≈ width*height/750 → ~1600 for 1000×1200.
-        OpenAI GPT-4o high-detail 2048×2048 ≈ 1445.
-        Gemini 258/tile × 6 tiles for a 2048×2048 ≈ 1548.
-        Anything in the 800-2000 range is defensible. Enforce bounds so an
-        accidental edit doesn't drop it to e.g. 16.
-        """
-        assert 800 <= _IMAGE_TOKEN_ESTIMATE <= 2500
-        assert _IMAGE_CHAR_EQUIVALENT == _IMAGE_TOKEN_ESTIMATE * _CHARS_PER_TOKEN
+    def test_image_priced_at_the_learned_cost(self):
+        """The budget walk charges each image at the per-image price learned from provider usage
+        (the same figure the trigger estimator uses), falling back to the flat default."""
+        content = [{"type": "text", "text": "look"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}]
+        assert _content_length_for_budget(content) == 4 + DEFAULT_IMAGE_TOKEN_COST * _CHARS_PER_TOKEN
+        with image_cost_context(4_000):
+            assert _content_length_for_budget(content) == 4 + 4_000 * _CHARS_PER_TOKEN
 
 
 class TestTokenBudgetWithImages:

tests/agent/test_image_token_cost.py

@@ -0,0 +1,65 @@
+"""Per-image token cost learned from provider usage (agent/image_token_cost.py, #70328).
+
+A flat per-image constant undercounts multimodal local models 2-4x (a GUI loop then hits
+provider 400s before compaction can fire) and overcounts providers that downscale. The
+provider prices every image exactly on the request that carries it, so the residual between
+the real prompt count and ``anchor + text-only delta`` teaches the per-image cost.
+"""
+
+from types import SimpleNamespace
+
+from agent import image_token_cost as itc
+from agent.model_metadata import estimate_messages_tokens_rough
+from agent.usage_anchor import anchored_context_tokens, capture_usage_anchor
+
+
+def _img():
+    return {"role": "user", "content": [{"type": "text", "text": "look"},
+            {"type": "image_url", "image_url": {"url": "data:image/png;base64," + "A" * 4000}}]}
+
+
+def _agent(anchor):
+    return SimpleNamespace(_usage_anchor=anchor, model="vision-local", base_url="http://127.0.0.1:8080/v1")
+
+
+def _isolate(monkeypatch, tmp_path):
+    monkeypatch.setattr(itc, "_LEARNED", {})
+    monkeypatch.setattr(itc, "_LOADED", True)
+    monkeypatch.setattr(itc, "_cache_path", lambda: tmp_path / "image_token_costs.json")
+
+
+def test_residual_on_an_image_delta_teaches_the_per_image_cost(monkeypatch, tmp_path):
+    """Two screenshots appended since the last real reading; the provider reports 8,000 tokens
+    more than the text-only projection -> 4,000 per image, and every estimator prices images at
+    that from now on (trigger and budget walk read the same bound value)."""
+    _isolate(monkeypatch, tmp_path)
+    history = [{"role": "user", "content": "start"}, {"role": "assistant", "content": "ok"}]
+    anchor = capture_usage_anchor(10_000, 5, history)
+    history += [_img(), {"role": "assistant", "content": "looking"}, _img()]
+    agent = _agent(anchor)
+    with itc.image_cost_context(0):
+        text_only = anchored_context_tokens(history, anchor)
+    with itc.image_cost_context(None):
+        learned = itc.calibrate_from_usage(agent, history, text_only + 2 * 4_000)
+        assert learned == 4_000
+        assert itc.current_image_token_cost() == 4_000
+        assert estimate_messages_tokens_rough([_img()]) >= 4_000
+    # Persisted per model@host: a fresh process starts calibrated.
+    monkeypatch.setattr(itc, "_LEARNED", {})
+    monkeypatch.setattr(itc, "_LOADED", False)
+    assert itc.learned_image_token_cost("vision-local", "http://127.0.0.1:8080/v1") == 4_000
+    assert itc.learned_image_token_cost("other-model", "http://127.0.0.1:8080/v1") == itc.DEFAULT_IMAGE_TOKEN_COST
+
+
+def test_nothing_learned_without_images_or_anchor(monkeypatch, tmp_path):
+    """Text-only deltas, missing anchors and implausible residuals teach nothing: a text-estimate
+    error must never be mistaken for an image price."""
+    _isolate(monkeypatch, tmp_path)
+    history = [{"role": "user", "content": "start"}, {"role": "assistant", "content": "ok"}]
+    anchor = capture_usage_anchor(10_000, 5, history)
+    history += [{"role": "user", "content": "no image here"}]
+    assert itc.calibrate_from_usage(_agent(anchor), history, 50_000) is None
+    history += [_img()]
+    assert itc.calibrate_from_usage(_agent(None), history, 50_000) is None
+    assert itc.calibrate_from_usage(_agent(anchor), history, 10_001) is None  # residual < plausible floor
+    assert itc._LEARNED == {}

website/docs/developer-guide/context-compression-and-caching.md

@@ -96,6 +96,15 @@ provider-proven overflow all compress immediately.
 Opaque provider blobs (`encrypted_content` on Codex reasoning / compaction
 items) contribute 0 to every local estimate; only real usage ever prices them.
 
+Images are priced at the per-image cost **learned from the provider's usage**
+(`agent/image_token_cost.py`), not a vendor formula: on a response whose delta
+since the previous anchor introduced N images, the residual between the real
+`prompt_tokens` and the text-only projection is N × the provider's price. The
+value is kept per `model@host` in `~/.hermes/cache/image_token_costs.json` and
+bound per turn so the trigger estimator, the tail-budget walk and gateway
+hygiene all use the same figure. Before the first vision turn a flat 1,500
+default applies.
+
 #### Failure cooldown and provider-proven overflow
 
 A failed or stalled summary attempt arms a per-session **failure cooldown**