fix(usage): let operator Kimi/Z.AI limits reach the Piri lane

bridge/.env.example

@@ -72,6 +72,16 @@ TELEGRAM_BOT_TOKEN = your_bot_token_here
 # CCC_USAGE_BUDGET_TOKENS_PIRI=0
 # CCC_USAGE_BUDGET_WARN_PERCENT=80
 #
+# Piri reports no token/quota telemetry, so /usage on a Piri node can only show
+# the local meter's estimate — and the synthesized quota windows are keyed by
+# service. The Claude lane reads its service from the ANTHROPIC_BASE_URL host,
+# but Piri's backend is pinned by the per-node launcher's --model argument,
+# which the bridge never sees. Name it here to match the launcher, and the
+# CCC_USAGE_KIMI_* / CCC_USAGE_ZAI_* limits start applying to /usage.
+# Known values: "Kimi Code" (launcher kimi-coding/*), "Z.AI" (launcher zai/*).
+# Unset leaves Piri /usage output unchanged.
+# CCC_USAGE_PIRI_SERVICE=Kimi Code
+#
 # Persisted Claude sessions resume after a bridge restart by default when their
 # SDK transcript still exists. Set false only for a never-resume policy.
 # CCC_RESUME_PERSISTED_SESSIONS=true

bridge/CLAUDE.md

@@ -151,6 +151,7 @@ utils/chat_logger.py Per-session debug chat logging
 | `CCC_CODEX_SKILL_COLLECTOR_MAX_JOBS_PER_SWEEP` | Codex only | Hard cap on provider attempts per collector sweep, bounded to 1–10 (default: 1) |
 | `CCC_USAGE_KIMI_5H_REQUEST_LIMIT` | No | Operator-configured Kimi Code 5-hour request limit (from the Kimi Code Console); when set, `/usage` shows `used/limit req · X% used / Y% left` for the synthesized `Kimi 5-hour` window (local estimate). Unset = count-only |
 | `CCC_USAGE_KIMI_WEEKLY_TOKEN_LIMIT` | No | Operator-configured Kimi Code weekly token limit (Console shows weekly % only; back-compute the limit once from the meter's 7-day totals). When set, `/usage` adds a `Kimi weekly` percent window (local estimate). Unset = weekly window hidden |
+| `CCC_USAGE_PIRI_SERVICE` | Piri only | Names the Piri lane's backing service so the `CCC_USAGE_*_LIMIT` windows above apply to it. The Claude lane infers this from the `ANTHROPIC_BASE_URL` host, but Piri's backend is pinned by the per-node launcher's `--model` argument, which the bridge never sees — so the operator states it here, deliberately next to the limits it selects. Accepts only a known service name (`Kimi Code`, `Z.AI`), case-insensitive; unset or unrecognized leaves `/usage` byte-identical to before. **Set it to match the launcher**: `kimi-coding/*` → `Kimi Code`, `zai/*` → `Z.AI` |
 | `CCC_DEPS_UNLOCKED` | No | Set 1 to skip the default hash-locked install (`requirements.lock.txt` via `pip --require-hashes`) and use the legacy lower-bound `requirements.txt` flow. Regenerate locks with `scripts/ccc-deps-lock.sh` |
 | `OPENAI_API_KEY` | Voice only | API key for Whisper transcription |
 | `OPENAI_BASE_URL` | No | Optional OpenAI-compatible Whisper API base URL |

bridge/core/project_chat.py

@@ -37,6 +37,7 @@
     UsageSnapshot,
     load_claude_status_snapshot,
     local_claude_environment_snapshot,
+    local_piri_environment_snapshot,
     merge_usage,
     parse_claude_rate_limit_event,
     parse_claude_result,
@@ -471,6 +472,14 @@ async def get_usage(self, user_id: int, chat_id: int, session_id: str | None) ->
             if get_usage is not None:
                 return await asyncio.wait_for(get_usage(session_id), timeout=7.0)
             provider = str(getattr(self._config, "agent_provider", "claude"))
+            if provider == "piri":
+                # PiriRuntime exposes no usage endpoint and Piri reports no
+                # token/quota telemetry, so the only usable signal is the
+                # local meter — but synthesis is keyed by service, which the
+                # bare snapshot lacks. Name it, then fill from the meter.
+                return self._fill_local_service_windows(
+                    local_piri_environment_snapshot()
+                )
             if provider != "claude":
                 return UsageSnapshot(provider=provider)
             # Claude adapter path (#584): ClaudeRuntime exposes no usage
@@ -507,26 +516,35 @@ async def get_usage(self, user_id: int, chat_id: int, session_id: str | None) ->
         rate_limit = getattr(self, "_claude_rate_limit", None)
         if rate_limit is not None:
             result = merge_usage(result, rate_limit)
-        # Third-party services (e.g. Kimi Code) publish no quota data, so no
-        # observed window ever arrives; fall back to the meter's local
-        # rolling-window estimate so /usage is not stuck on "unavailable".
-        # Real observed windows always win — synthesis only fills an empty set.
-        if result.service is not None and not result.windows:
-            meter = getattr(self, "_usage_meter", None)
-            if meter is not None:
-                try:
-                    rolling = meter.rolling_usage().get(result.provider)
-                    period = getattr(meter, "period_usage", None)
-                    weekly = period(days=7).get(result.provider) if period is not None else None
-                    windows = synthesize_service_windows(result.service, rolling, weekly)
-                except Exception:
-                    logger.debug("Local service window synthesis failed")
-                    windows = ()
-                if windows:
-                    result = merge_usage(
-                        result,
-                        UsageSnapshot(provider=result.provider, windows=windows),
-                    )
+        return self._fill_local_service_windows(result)
+
+    def _fill_local_service_windows(self, result: UsageSnapshot) -> UsageSnapshot:
+        """Fill empty rate-limit windows from the local meter estimate.
+
+        Third-party services (e.g. Kimi Code) publish no quota data, so no
+        observed window ever arrives; fall back to the meter's local
+        rolling-window estimate so /usage is not stuck on "unavailable".
+        Real observed windows always win — synthesis only fills an empty set,
+        and a snapshot without a service is returned untouched.
+        """
+        if result.service is None or result.windows:
+            return result
+        meter = getattr(self, "_usage_meter", None)
+        if meter is None:
+            return result
+        try:
+            rolling = meter.rolling_usage().get(result.provider)
+            period = getattr(meter, "period_usage", None)
+            weekly = period(days=7).get(result.provider) if period is not None else None
+            windows = synthesize_service_windows(result.service, rolling, weekly)
+        except Exception:
+            logger.debug("Local service window synthesis failed")
+            windows = ()
+        if windows:
+            result = merge_usage(
+                result,
+                UsageSnapshot(provider=result.provider, windows=windows),
+            )
         return result
 
     def _get_conversation_lock(self, user_id: int, chat_id: int) -> asyncio.Lock:

bridge/core/usage.py

@@ -267,6 +267,48 @@ def local_claude_environment_snapshot() -> UsageSnapshot:
 }
 
 
+def detect_piri_service(value: object = None) -> str | None:
+    """Return the backing service for the Piri lane, or ``None`` if unset.
+
+    The Claude lane infers its service from the ``ANTHROPIC_BASE_URL`` host,
+    but Piri talks to its provider directly and exposes no such URL to the
+    bridge: the backend is pinned by the per-node launcher's ``--model``
+    argument, which the bridge never sees. So the operator names it via
+    ``CCC_USAGE_PIRI_SERVICE``, deliberately co-located with the
+    ``CCC_USAGE_*_LIMIT`` values it selects — the two are edited together and
+    assert the same fact, which keeps them from drifting apart.
+
+    Only a name already known to ``_SERVICE_WINDOW_SPECS`` is accepted, and
+    matching is case-insensitive. Unset, unknown, or malformed values return
+    ``None`` so Piri rendering stays byte-identical to before this feature.
+    """
+    raw = value if value is not None else os.environ.get("CCC_USAGE_PIRI_SERVICE")
+    text = _text(raw, maximum=40)
+    if not text:
+        return None
+    folded = text.casefold()
+    for service in _SERVICE_WINDOW_SPECS:
+        if service.casefold() == folded:
+            return service
+    return None
+
+
+def local_piri_environment_snapshot() -> UsageSnapshot:
+    """Base snapshot naming the Piri lane's backing service, when configured.
+
+    Piri reports no token or quota telemetry, so without this the operator's
+    ``CCC_USAGE_KIMI_*``/``CCC_USAGE_ZAI_*`` limits are silently dead on Piri
+    nodes: the window synthesis in ``synthesize_service_windows`` is keyed by
+    service, and the Piri snapshot never carried one. Naming the service lets
+    the existing local-estimate windows render. Returns an empty Piri snapshot
+    when unconfigured, preserving current behavior exactly.
+    """
+    service = detect_piri_service()
+    if service is None:
+        return UsageSnapshot(provider="piri")
+    return UsageSnapshot(provider="piri", service=service, plan_type=service)
+
+
 def synthesize_service_windows(
     service: str | None,
     rolling: Mapping[str, int] | None,

bridge/tests/test_usage.py

@@ -22,8 +22,10 @@
     UsageWindow,
     claude_endpoint_host,
     detect_claude_service,
+    detect_piri_service,
     load_claude_status_snapshot,
     local_claude_environment_snapshot,
+    local_piri_environment_snapshot,
     merge_usage,
     parse_claude_rate_limit_event,
     parse_claude_result,
@@ -832,6 +834,7 @@ async def test_get_usage_tolerates_handler_without_rate_limit_attribute(
     "ANTHROPIC_MODEL",
     "CLAUDE_CODE_EFFORT_LEVEL",
     "CLAUDE_CODE_MAX_CONTEXT_TOKENS",
+    "CCC_USAGE_PIRI_SERVICE",
 )
 
 
@@ -962,6 +965,79 @@ def test_render_piri_usage_states_provider_telemetry_boundary() -> None:
     assert "Session cost" not in rendered
 
 
+def test_detect_piri_service_accepts_only_known_services(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _clear_service_env(monkeypatch)
+    # Unset stays None so Piri rendering is unchanged for unconfigured nodes.
+    assert detect_piri_service() is None
+    monkeypatch.setenv("CCC_USAGE_PIRI_SERVICE", "Kimi Code")
+    assert detect_piri_service() == "Kimi Code"
+    # Case-insensitive, and the canonical spelling is what comes back so the
+    # value can key _SERVICE_WINDOW_SPECS directly.
+    assert detect_piri_service("kimi code") == "Kimi Code"
+    assert detect_piri_service("Z.AI") == "Z.AI"
+    assert detect_piri_service("  z.ai  ") == "Z.AI"
+    # Anything not in the window spec table is rejected rather than trusted,
+    # so a typo cannot silently select the wrong quota limits.
+    assert detect_piri_service("Kimi") is None
+    assert detect_piri_service("Anthropic") is None
+    assert detect_piri_service("") is None
+    assert detect_piri_service(123) is None
+
+
+def test_local_piri_environment_snapshot_is_inert_when_unconfigured(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _clear_service_env(monkeypatch)
+    snapshot = local_piri_environment_snapshot()
+    assert snapshot == UsageSnapshot(provider="piri")
+    assert snapshot.service is None
+    # Byte-identical rendering to the pre-feature Piri output.
+    assert render_usage(snapshot) == render_usage(UsageSnapshot(provider="piri"))
+
+
+def test_local_piri_environment_snapshot_names_configured_service(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _clear_service_env(monkeypatch)
+    monkeypatch.setenv("CCC_USAGE_PIRI_SERVICE", "Kimi Code")
+    snapshot = local_piri_environment_snapshot()
+    assert snapshot.provider == "piri"
+    assert snapshot.service == "Kimi Code"
+    assert snapshot.plan_type == "Kimi Code"
+
+
+def test_render_piri_usage_shows_synthesized_kimi_windows(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The operator's Kimi limits must reach the Piri lane's /usage output.
+
+    Before the service was named, CCC_USAGE_KIMI_* was silently dead on Piri
+    nodes: synthesis is keyed by service and the Piri snapshot carried none.
+    """
+    monkeypatch.setenv("CCC_USAGE_KIMI_5H_REQUEST_LIMIT", "359")
+    monkeypatch.setenv("CCC_USAGE_KIMI_WEEKLY_TOKEN_LIMIT", "292008127")
+    windows = synthesize_service_windows(
+        "Kimi Code",
+        {"requests": 173, "tokens": 2_847_600},
+        {"requests": 371, "tokens": 40_000_000},
+    )
+    rendered = render_usage(
+        UsageSnapshot(
+            provider="piri", service="Kimi Code", plan_type="Kimi Code", windows=windows
+        )
+    )
+    assert rendered.splitlines()[0] == "📊 Usage · Piri"
+    assert "Kimi 5-hour" in rendered
+    assert "Kimi weekly" in rendered
+    assert "173" in rendered
+    # The provider-telemetry boundary is still stated: Piri reports no tokens,
+    # and these windows are local estimates, not provider-reported quota.
+    assert "Provider token/quota telemetry: unavailable" in rendered
+    assert "Kimi quota" in rendered
+
+
 @pytest.mark.anyio
 async def test_get_usage_bases_snapshot_on_kimi_environment(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@@ -986,6 +1062,78 @@ async def test_get_usage_bases_snapshot_on_kimi_environment(
     assert result.windows == ()
 
 
+@pytest.mark.anyio
+async def test_get_usage_piri_lane_synthesizes_windows_from_meter(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """End-to-end: configured Piri service + local meter -> rendered windows.
+
+    PiriRuntime exposes no get_usage, so this exercises the runtime branch
+    that previously returned a bare snapshot and dropped the meter entirely.
+    """
+    _clear_service_env(monkeypatch)
+    monkeypatch.setenv("CCC_USAGE_PIRI_SERVICE", "Kimi Code")
+    monkeypatch.setenv("CCC_USAGE_KIMI_5H_REQUEST_LIMIT", "359")
+    monkeypatch.setenv("CCC_USAGE_KIMI_WEEKLY_TOKEN_LIMIT", "292008127")
+    handler = ProjectChatHandler.__new__(ProjectChatHandler)
+    # A runtime without get_usage — exactly PiriRuntime's shape.
+    handler._agent_runtime = SimpleNamespace()
+    handler._require_runtime = lambda: handler._agent_runtime
+    handler._config = SimpleNamespace(agent_provider="piri")
+    handler._usage_meter = SimpleNamespace(
+        rolling_usage=lambda: {"piri": {"requests": 173, "tokens": 2_847_600}},
+        period_usage=lambda days: {"piri": {"requests": 371, "tokens": 40_000_000}},
+    )
+
+    result = await handler.get_usage(1, 2, None)
+    assert result.provider == "piri"
+    assert result.service == "Kimi Code"
+    assert [w.label for w in result.windows] == ["Kimi 5-hour", "Kimi weekly"]
+    assert (result.windows[0].used_count, result.windows[0].count_limit) == (173, 359)
+    rendered = render_usage(result)
+    assert "Kimi 5-hour" in rendered
+
+
+@pytest.mark.anyio
+async def test_get_usage_piri_lane_unconfigured_stays_bare(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Without CCC_USAGE_PIRI_SERVICE the Piri lane behaves exactly as before."""
+    _clear_service_env(monkeypatch)
+    handler = ProjectChatHandler.__new__(ProjectChatHandler)
+    handler._agent_runtime = SimpleNamespace()
+    handler._require_runtime = lambda: handler._agent_runtime
+    handler._config = SimpleNamespace(agent_provider="piri")
+    handler._usage_meter = SimpleNamespace(
+        rolling_usage=lambda: {"piri": {"requests": 173, "tokens": 2_847_600}},
+        period_usage=lambda days: {"piri": {"requests": 371, "tokens": 40_000_000}},
+    )
+
+    result = await handler.get_usage(1, 2, None)
+    assert result == UsageSnapshot(provider="piri")
+    assert result.windows == ()
+
+
+@pytest.mark.anyio
+async def test_get_usage_non_piri_runtime_provider_unchanged(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Crush/other runtime-backed providers keep returning a bare snapshot."""
+    _clear_service_env(monkeypatch)
+    monkeypatch.setenv("CCC_USAGE_PIRI_SERVICE", "Kimi Code")
+    handler = ProjectChatHandler.__new__(ProjectChatHandler)
+    handler._agent_runtime = SimpleNamespace()
+    handler._require_runtime = lambda: handler._agent_runtime
+    handler._config = SimpleNamespace(agent_provider="crush")
+    handler._usage_meter = SimpleNamespace(
+        rolling_usage=lambda: {"crush": {"requests": 5, "tokens": 10}},
+        period_usage=lambda days: {"crush": {"requests": 9, "tokens": 20}},
+    )
+
+    result = await handler.get_usage(1, 2, None)
+    assert result == UsageSnapshot(provider="crush")
+
+
 def test_synthesize_service_windows_builds_count_only_kimi_window() -> None:
     windows = synthesize_service_windows(
         "Kimi Code", {"requests": 47, "tokens": 29_000_000}