diff --git a/bridge/.env.example b/bridge/.env.example index a24c7ee9..e980e656 100644 --- a/bridge/.env.example +++ b/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 diff --git a/bridge/CLAUDE.md b/bridge/CLAUDE.md index 3c9d58de..7befe3a5 100644 --- a/bridge/CLAUDE.md +++ b/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 | diff --git a/bridge/core/project_chat.py b/bridge/core/project_chat.py index ea6a880e..662b5b76 100644 --- a/bridge/core/project_chat.py +++ b/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: diff --git a/bridge/core/usage.py b/bridge/core/usage.py index b4860b5e..20c52ca6 100644 --- a/bridge/core/usage.py +++ b/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, diff --git a/bridge/tests/test_usage.py b/bridge/tests/test_usage.py index e244f1b5..d0cc9e16 100644 --- a/bridge/tests/test_usage.py +++ b/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}