枠が読めなかった理由を、読める形で持ち帰る

使用量の取り直しが失敗したとき、画面に出るのが CLI の返事の JSON の頭 300 字
だけだった。相手は人向けの報告を JSON の中の 1 項目に入れてくるので、並び順
次第でセッション id やトークン数だけが残り、肝心の文面が枠の外へ落ちる。
報告のほうを取り出して渡す。長さも 600 字に揃える —— 外部の相手のエラーで
「300 字では切れて打つ手が読めない」と踏んだのと同じ話だった。

JSON の拾い方も固くする。最初の { から後ろを読む作りだったので、混ざった警告に
{ が 1 つでもあると、そこから読み始めて必ず失敗していた。しかも落ち方が
「JSON が無い」と同じなので、警告が出た日にだけ静かに読めなくなる。
{ を順に試して、中身のあるものを採る —— 空の {} は警告に混ざりうるが、
CLI の本体が空であることはない。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: claude

app/usage.py

@@ -177,7 +177,7 @@ async def _elevenlabs(spec, credential: str) -> list[Window]:
     if res.status_code in (401, 403):
         raise UsageError(f"認証情報が受け付けられませんでした(HTTP {res.status_code})")
     if res.status_code >= 400:
-        raise UsageError(f"HTTP {res.status_code}: {res.text[:300]}")
+        raise UsageError(f"HTTP {res.status_code}: {res.text[:REASON_MAX]}")
     try:
         body = res.json()
     except ValueError:
@@ -288,10 +288,18 @@ async def _bridge(spec: providers.Provider) -> list[Window]:
             )
         )
     if not windows:
-        raise UsageError(str(body.get("reason") or "CLI が使用量を返しませんでした")[:300])
+        raise UsageError(
+            str(body.get("reason") or "CLI が使用量を返しませんでした")[:REASON_MAX]
+        )
     return windows
 
 
+# 失敗の理由として持ち帰る長さ。**300 では足りなかった** —— 相手は人向けの報告を
+# 返してくるので、頭で切ると「なぜ駄目だったか」が枠の外へ落ちる(実際にそうなった)。
+# 絵と音の相手のエラー(`app/media_backends.py` の `remote_error`)と同じ長さにしてある。
+REASON_MAX = 600
+
+
 def _bridge_error(body: dict, status: int) -> str:
     """ブリッジが返した理由を取り出す。
 
@@ -302,7 +310,7 @@ def _bridge_error(body: dict, status: int) -> str:
     if isinstance(detail, dict):
         detail = detail.get("error") or detail
     reason = body.get("error") or detail
-    return (str(reason) if reason else f"HTTP {status}")[:300]
+    return (str(reason) if reason else f"HTTP {status}")[:REASON_MAX]
 
 
 async def fetch(spec: providers.Provider) -> list[Window]:

bridge/cli_bridge.py

@@ -1052,11 +1052,8 @@ def _result_of(raw: str) -> tuple[str, dict, str]:
         return _codex_result(raw)
     if CLI not in ("claude", "antigravity"):
         return raw, {}, ""
-    try:
-        payload = json.loads(_json_tail(raw))
-    except ValueError:
-        return raw, {}, ""
-    if not isinstance(payload, dict):
+    payload = _json_object(raw)
+    if payload is None:
         return raw, {}, ""
 
     if CLI == "claude":
@@ -1301,16 +1298,40 @@ async def health(check: bool = False) -> dict:
 )
 
 
-def _json_tail(raw: str) -> str:
-    """先に混ざった警告を落として、JSON の本体だけを返す。
+def _json_object(raw: str) -> dict | None:
+    """混ざった出力の中から、JSON の本体(辞書)を 1 つ取り出す。読めなければ None。
 
     **stderr を stdout に混ぜて読んでいる**(`_run_for_usage`)ので、CLI が
     警告を 1 行吐くだけで `json.loads` が落ちる —— 実測で
     `Warning: no stdin data received in 3s, …` が前に付いた。
-    最初の `{` から後ろを見る。JSON が無ければそのまま返す（人向けの行として読ませる）。
+
+    **「最初の `{` から後ろ」では足りない。** 警告のほうに `{` が 1 つでも
+    混ざっていると、そこから読み始めて必ず失敗する —— しかも落ち方が
+    「JSON が無い」と同じなので、警告が出た日にだけ静かに読めなくなる。
+    **`{` を順に試して、最初に読めたものを採る**(`raw_decode` は途中で
+    終わっていても、そこまでを 1 つの値として返す)。
+
+    **中身のあるものを採る。** 空の `{}` は警告のほうに混ざりうるが、CLI の本体が
+    空であることはない —— 最初に読めたものを無条件に採ると、飾りの `{}` を
+    本体として返してしまう(実際にそうなった)。最後まで空しか無ければそれを返す。
+
+    入れ子の `{` は外側が先に読めるので、いちばん大きいものを探す必要は無い。
     """
+    decoder = json.JSONDecoder()
+    empty: dict | None = None
     at = raw.find("{")
-    return raw[at:] if at >= 0 else raw
+    while at >= 0:
+        try:
+            value, _ = decoder.raw_decode(raw, at)
+        except ValueError:
+            at = raw.find("{", at + 1)
+            continue
+        if isinstance(value, dict):
+            if value:
+                return value
+            empty = empty if empty is not None else value
+        at = raw.find("{", at + 1)
+    return empty
 
 
 # claude が言う戻る時刻。**年を書かない**(実測: `Sep 16, 7:10pm (UTC)` /
@@ -1359,10 +1380,8 @@ def _claude_windows(raw: str) -> list[dict]:
     サブスクの枠ではない。
     """
     text = raw
-    with suppress(ValueError):
-        payload = json.loads(_json_tail(raw))
-        if isinstance(payload, dict):
-            text = str(payload.get("result") or "")
+    if (payload := _json_object(raw)) is not None:
+        text = str(payload.get("result") or "")
     windows = []
     for line in text.splitlines():
         if not (found := _CLAUDE_USAGE_RE.match(line.strip())):
@@ -1691,6 +1710,31 @@ async def _talk() -> tuple[list[dict], str]:
 _LAST_USAGE: dict | None = None
 
 
+# 窓を組めなかったときに持ち帰る理由の長さ。**300 では足りなかった** ——
+# CLI は人向けの報告を JSON の奥に入れてくるので、頭で切ると metadata だけが残り、
+# 「なぜ読めなかったか」が枠の外へ落ちる(実際にそうなった)。
+# 外部の相手のエラー(`app/media_backends.py` の `remote_error`)と同じ長さにしてある。
+REASON_MAX = 600
+
+
+def _usage_reason(raw: str) -> str:
+    """窓を組めなかったときに Chiezo へ渡す理由。**人向けの報告のほうを渡す。**
+
+    CLI が返すのは envelope で、読ませたい文面はその中の 1 項目にある
+    (claude は `result`、agy は `response`)。envelope をそのまま切り詰めると、
+    並び順次第でセッション id やトークン数だけが残る。
+
+    読めなければ元の文字列を返す(codex の JSON-RPC はそれ自体が短い)。
+    """
+    text = raw
+    if (payload := _json_object(raw)) is not None:
+        for key in ("result", "response"):
+            if isinstance(payload.get(key), str) and payload[key].strip():
+                text = payload[key]
+                break
+    return text.strip()[:REASON_MAX]
+
+
 async def _read_usage() -> dict:
     """CLI に枠を聞いて、窓の一覧に直す。**待ち枠は取らない**(呼ぶ側の仕事)。"""
     if CLI == "codex":
@@ -1705,7 +1749,7 @@ async def _read_usage() -> dict:
         "cli": CLI,
         "windows": windows,
         # 窓を組めなかったときだけ意味を持つ(Chiezo がそのまま画面に出す)。
-        "reason": "" if windows else raw[:300],
+        "reason": "" if windows else _usage_reason(raw),
         "taken_at": datetime.now(UTC).isoformat(timespec="seconds"),
         "stale": False,
     }

tests/test_bridge.py

@@ -1560,3 +1560,82 @@ def test_the_same_window_seen_twice_is_still_one(self, bridge):
         windows = server._windows_in({"a": {"primary": same}, "b": {"primary": dict(same)}})
 
         assert [w["id"] for w in windows] == ["primary"]
+
+
+class TestWhyTheQuotaCouldNotBeRead:
+    """窓を組めなかったときに、**なぜ読めなかったか**を持ち帰る。
+
+    CLI は人向けの報告を JSON の奥に入れてくる。envelope をそのまま切り詰めると、
+    並び順次第でセッション id やトークン数だけが残り、肝心の文面が枠の外へ落ちる
+    (実際にそうなった)。
+    """
+
+    def test_the_human_readable_part_is_what_travels(self, bridge):
+        server = bridge(CHIEZO_BRIDGE_CLI="claude")
+        raw = json.dumps({
+            "is_error": False, "num_turns": 0, "session_id": "x" * 200,
+            "usage": {"input_tokens": 0, "output_tokens": 0},
+            "result": "Not logged in · Please run /login",
+        })
+
+        assert server._usage_reason(raw) == "Not logged in · Please run /login"
+
+    def test_antigravity_says_it_in_another_field(self, bridge):
+        server = bridge(CHIEZO_BRIDGE_CLI="antigravity")
+        raw = json.dumps({"status": "failed", "response": "枠を読めませんでした"})
+
+        assert server._usage_reason(raw) == "枠を読めませんでした"
+
+    def test_something_unreadable_travels_as_it_is(self, bridge):
+        """codex の JSON-RPC はそれ自体が短いので、そのまま渡す。"""
+        server = bridge(CHIEZO_BRIDGE_CLI="codex")
+
+        assert server._usage_reason("rate limits unavailable") == "rate limits unavailable"
+
+    def test_it_is_cut_but_not_at_300(self, bridge):
+        """300 では足りなかった —— 頭で切ると打つ手が読めない。"""
+        server = bridge(CHIEZO_BRIDGE_CLI="claude")
+        raw = json.dumps({"result": "あ" * 800})
+
+        assert len(server._usage_reason(raw)) == server.REASON_MAX
+
+
+class TestFindingTheJson:
+    """混ざった出力の中から JSON の本体を拾う(`_json_object`)。
+
+    stderr を stdout に混ぜて読んでいるので、CLI が警告を 1 行吐くだけで
+    素の `json.loads` は落ちる。
+    """
+
+    PANEL: ClassVar[str] = json.dumps({"result": "Current session: 20% used"})
+
+    def test_a_warning_in_front_is_stepped_over(self, bridge):
+        server = bridge(CHIEZO_BRIDGE_CLI="claude")
+        raw = "Warning: no stdin data received in 3s\n" + self.PANEL
+
+        assert server._json_object(raw)["result"].startswith("Current session")
+
+    def test_a_brace_in_the_warning_does_not_break_it(self, bridge):
+        """**ここが「最初の `{` から後ろ」では駄目なところ。** 警告に `{` が 1 つ
+        混ざると、そこから読み始めて必ず失敗する —— しかも落ち方が「JSON が無い」と
+        同じなので、警告が出た日にだけ静かに読めなくなる。
+        """
+        server = bridge(CHIEZO_BRIDGE_CLI="claude")
+        raw = "Warning: config {foo} is deprecated\n" + self.PANEL
+
+        assert server._json_object(raw)["result"].startswith("Current session")
+
+    def test_no_json_at_all_is_not_an_answer(self, bridge):
+        server = bridge(CHIEZO_BRIDGE_CLI="claude")
+
+        assert server._json_object("Not logged in") is None
+
+    def test_the_panel_survives_a_noisy_line(self, bridge):
+        """窓まで組めることを通しで見る(拾えても読めなければ意味が無い)。"""
+        server = bridge(CHIEZO_BRIDGE_CLI="claude")
+        raw = "note: {} \n" + json.dumps({
+            "result": ("Current session: 20% used · resets Sep 16, 7:10pm (UTC)\n"
+                       "Current week (all models): 25% used · resets Sep 20, 5:59am (UTC)"),
+        })
+
+        assert [w["used_percent"] for w in server._claude_windows(raw)] == [20.0, 25.0]