merge: kéo Delta epic-R04 (gồm cả R01 và R03) vào gamma/refactor
Nam chốt: không chờ Delta merge vào main, lấy sớm để va chạm nhỏ và sửa ngay, thay vì dồn một cục lúc cả hai cùng lên main. R04 chứa trọn R01 và R03 nên một lần merge là đủ cả ba: 96 file, +8260 dòng. Xung đột chỉ 5 file, đều là __init__.py add/add — hai team cùng dựng khung thư mục nên đụng docstring. Giữ docstring của Gamma (nói rõ ràng buộc "không import PySide6"), giữ mọi phần code của Delta. Riêng tests/fakes/__init__.py: bỏ hai dòng import háo hức của Delta (fake_provider, fake_tool_executor). fake_provider dùng `from providers.base import ...` — import tuyệt đối, chỉ chạy được khi cwd là gốc repo — nên nó làm đứt bài test "dùng fake mà không nạp config thật". Không ai import ở cấp package; test của Delta gọi thẳng module nên bỏ đi không ảnh hưởng họ. Đã ghi lý do vào docstring của gói. Delta cũng xoá preview-desktop và "requirements (cloud copy).txt". 430 test xanh sau merge. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Nam Pham Dinh Thanh
@@ -0,0 +1 @@
|
||||
"""Infrastructure config package: ConfigRepository and typed settings facades."""
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure filesystem package: Tool handlers (file, command, fetch tools) and execution workspace."""
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure MCP package: McpToolSourceManager and child process lifecycle."""
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure persistence package."""
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure JSON persistence package: AtomicJsonFile and repositories."""
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure platform adapters package."""
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure Qt platform adapters: QtSchedulerClock."""
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure providers package: LLM provider adapters and ProviderRegistry."""
|
||||
|
||||
@@ -0,0 +1,287 @@
|
||||
"""Central registry of every LLM provider the app can talk to.
|
||||
|
||||
Replaces the bare ``{name: class}`` dict in ``providers/factory.py`` as the
|
||||
single catalogue of providers. Two responsibilities, kept deliberately narrow:
|
||||
|
||||
1. **Lookup** — resolve a provider id (or one of its aliases, or a bare model
|
||||
id) to its :class:`~domain.models.provider_descriptor.ProviderDescriptor`.
|
||||
2. **Construction** — instantiate the concrete adapter class that speaks the
|
||||
descriptor's wire protocol.
|
||||
|
||||
This is infrastructure, not domain: it is allowed to import the concrete
|
||||
``providers/*`` adapters (which pull in ``requests``). The adapters are imported
|
||||
lazily inside :meth:`build` so that merely *reading the catalogue* — which the
|
||||
pure routing service does on every turn — never drags the HTTP stack into the
|
||||
process.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from typing import Any, Dict, Iterable, List, Optional
|
||||
|
||||
from ...domain.models.provider_descriptor import (
|
||||
AuthKind,
|
||||
ProviderDescriptor,
|
||||
WireProtocol,
|
||||
)
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Built-in catalogue.
|
||||
#
|
||||
# Mirrors DEFAULT_CONFIG["providers"] in config.py (ids + default models) and
|
||||
# providers/factory.py (id -> wire protocol). Prices are intentionally absent:
|
||||
# core/routing/metadata.py owns cost, and a guessed price is worse than a
|
||||
# known-unknown (see that module's docstring).
|
||||
# --------------------------------------------------------------------------- #
|
||||
BUILTIN_DESCRIPTORS: tuple = (
|
||||
ProviderDescriptor(
|
||||
provider_id="openai_compat",
|
||||
display_name="OpenAI-compatible gateway",
|
||||
wire_protocol=WireProtocol.OPENAI_COMPAT,
|
||||
auth_kind=AuthKind.API_KEY,
|
||||
default_model="gpt-4o-mini",
|
||||
supports_vision=True,
|
||||
# A generic gateway has no fixed host, so the endpoint MUST be
|
||||
# configured before the provider can be used at all.
|
||||
requires_base_url=True,
|
||||
),
|
||||
ProviderDescriptor(
|
||||
provider_id="anthropic",
|
||||
display_name="Anthropic Claude",
|
||||
wire_protocol=WireProtocol.ANTHROPIC,
|
||||
auth_kind=AuthKind.API_KEY,
|
||||
default_model="claude-sonnet-4-6",
|
||||
# Kept in sync with AnthropicProvider._FALLBACK_MODELS — the list the
|
||||
# provider itself falls back to when /v1/models cannot be reached.
|
||||
models=("claude-opus-4-8", "claude-sonnet-4-6", "claude-haiku-4-5-20251001"),
|
||||
max_context=200000,
|
||||
supports_vision=True,
|
||||
),
|
||||
ProviderDescriptor(
|
||||
provider_id="ollama",
|
||||
display_name="Ollama (local)",
|
||||
wire_protocol=WireProtocol.OPENAI_COMPAT,
|
||||
# A local runtime needs no credential; Settings must not demand one.
|
||||
auth_kind=AuthKind.NONE,
|
||||
default_model="llama3.1",
|
||||
supports_vision=False,
|
||||
requires_base_url=True,
|
||||
),
|
||||
ProviderDescriptor(
|
||||
provider_id="github_copilot",
|
||||
display_name="GitHub Copilot",
|
||||
wire_protocol=WireProtocol.OPENAI_COMPAT,
|
||||
# The credential is a Copilot token minted by an external login flow,
|
||||
# not a self-service API key.
|
||||
auth_kind=AuthKind.OAUTH_TOKEN,
|
||||
default_model="gpt-4o",
|
||||
models=("gpt-4o", "gpt-4o-mini"),
|
||||
max_context=128000,
|
||||
supports_vision=True,
|
||||
),
|
||||
ProviderDescriptor(
|
||||
provider_id="codex",
|
||||
display_name="OpenAI",
|
||||
wire_protocol=WireProtocol.OPENAI_COMPAT,
|
||||
auth_kind=AuthKind.API_KEY,
|
||||
default_model="gpt-4o-mini",
|
||||
models=("gpt-4o", "gpt-4o-mini", "o1", "o3"),
|
||||
max_context=128000,
|
||||
supports_vision=True,
|
||||
# Historic config key: early builds stored this provider as "openai".
|
||||
aliases=("openai",),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class ProviderNotFoundError(LookupError):
|
||||
"""Raised when no descriptor answers to the requested provider id.
|
||||
|
||||
A dedicated type (rather than bare ``KeyError``) lets callers distinguish
|
||||
"this provider is not in the catalogue" from an unrelated dict miss, and
|
||||
keeps the message actionable by listing what IS registered.
|
||||
"""
|
||||
|
||||
|
||||
class ProviderRegistry:
|
||||
"""Thread-safe catalogue of :class:`ProviderDescriptor` records.
|
||||
|
||||
Thread-safety matters because model discovery runs on background worker
|
||||
threads (the routing prober, Settings' "Load models") and republishes an
|
||||
updated descriptor via :meth:`replace`, while chat turns on other threads
|
||||
are reading the catalogue concurrently.
|
||||
"""
|
||||
|
||||
def __init__(self, descriptors: Optional[Iterable[ProviderDescriptor]] = None) -> None:
|
||||
# Keyed by canonical id; alias resolution walks the values so an alias
|
||||
# can never shadow a real provider id.
|
||||
self._by_id: Dict[str, ProviderDescriptor] = {}
|
||||
self._lock = threading.RLock()
|
||||
for descriptor in descriptors or ():
|
||||
self.register(descriptor)
|
||||
|
||||
# -- registration --------------------------------------------------- #
|
||||
def register(self, descriptor: ProviderDescriptor) -> ProviderDescriptor:
|
||||
"""Add a descriptor. Refuses to silently overwrite an existing id so a
|
||||
typo in a plugin cannot hijack a built-in provider; use :meth:`replace`
|
||||
when an update is the actual intent."""
|
||||
with self._lock:
|
||||
existing = self._by_id.get(descriptor.provider_id)
|
||||
if existing is not None and existing != descriptor:
|
||||
raise ValueError(
|
||||
f"Provider '{descriptor.provider_id}' is already registered; "
|
||||
"call replace() to update it."
|
||||
)
|
||||
self._by_id[descriptor.provider_id] = descriptor
|
||||
return descriptor
|
||||
|
||||
def replace(self, descriptor: ProviderDescriptor) -> ProviderDescriptor:
|
||||
"""Register or update a descriptor unconditionally — the path model
|
||||
discovery uses to publish a freshly enumerated model list."""
|
||||
with self._lock:
|
||||
self._by_id[descriptor.provider_id] = descriptor
|
||||
return descriptor
|
||||
|
||||
# -- lookup ---------------------------------------------------------- #
|
||||
def get(self, provider_id: str) -> ProviderDescriptor:
|
||||
"""Descriptor for ``provider_id`` (canonical id or alias).
|
||||
|
||||
Raises :class:`ProviderNotFoundError` rather than returning ``None`` so
|
||||
a misconfigured provider fails loudly at the call site instead of
|
||||
surfacing later as an ``AttributeError`` on ``None``.
|
||||
"""
|
||||
found = self.find(provider_id)
|
||||
if found is None:
|
||||
known = ", ".join(sorted(self._by_id)) or "<empty registry>"
|
||||
raise ProviderNotFoundError(
|
||||
f"Unsupported provider: {provider_id!r}. Registered: {known}"
|
||||
)
|
||||
return found
|
||||
|
||||
def find(self, provider_id: str) -> Optional[ProviderDescriptor]:
|
||||
"""Non-raising :meth:`get` — ``None`` when nothing matches."""
|
||||
needle = (provider_id or "").strip()
|
||||
if not needle:
|
||||
return None
|
||||
with self._lock:
|
||||
direct = self._by_id.get(needle)
|
||||
if direct is not None:
|
||||
return direct
|
||||
# Fall back to a case-insensitive id/alias scan; order is stable
|
||||
# because dicts preserve insertion order, so the earliest-registered
|
||||
# provider wins a tie.
|
||||
for descriptor in self._by_id.values():
|
||||
if descriptor.matches(needle):
|
||||
return descriptor
|
||||
return None
|
||||
|
||||
def find_by_model(self, model_id: str) -> Optional[ProviderDescriptor]:
|
||||
"""Resolve a bare model id back to the provider that serves it.
|
||||
|
||||
This is the "dynamic lookup by model ID" R03-T02 calls for: routing
|
||||
decisions and saved conversations sometimes carry only a model name, and
|
||||
the caller still needs to know which provider to build. Returns ``None``
|
||||
when the model belongs to a gateway whose catalogue we cannot enumerate
|
||||
offline — callers then fall back to the configured active provider.
|
||||
"""
|
||||
needle = (model_id or "").strip()
|
||||
if not needle:
|
||||
return None
|
||||
with self._lock:
|
||||
for descriptor in self._by_id.values():
|
||||
if descriptor.knows_model(needle):
|
||||
return descriptor
|
||||
return None
|
||||
|
||||
def all(self) -> List[ProviderDescriptor]:
|
||||
"""Every registered descriptor, in registration order (snapshot copy —
|
||||
safe to iterate while another thread registers)."""
|
||||
with self._lock:
|
||||
return list(self._by_id.values())
|
||||
|
||||
def ids(self) -> List[str]:
|
||||
"""Canonical provider ids, sorted for stable UI/reporting output."""
|
||||
with self._lock:
|
||||
return sorted(self._by_id)
|
||||
|
||||
def __contains__(self, provider_id: object) -> bool:
|
||||
return isinstance(provider_id, str) and self.find(provider_id) is not None
|
||||
|
||||
def __len__(self) -> int:
|
||||
with self._lock:
|
||||
return len(self._by_id)
|
||||
|
||||
# -- construction ---------------------------------------------------- #
|
||||
def adapter_class(self, provider_id: str):
|
||||
"""Concrete ``Provider`` subclass implementing this provider's protocol.
|
||||
|
||||
The adapters are imported here (not at module import) so the pure
|
||||
routing/domain code can consult the catalogue without loading
|
||||
``requests`` and the whole HTTP stack.
|
||||
"""
|
||||
descriptor = self.get(provider_id)
|
||||
from ...providers.anthropic import AnthropicProvider
|
||||
from ...providers.openai_compat import OpenAICompatProvider
|
||||
|
||||
protocol_to_class = {
|
||||
WireProtocol.OPENAI_COMPAT: OpenAICompatProvider,
|
||||
WireProtocol.ANTHROPIC: AnthropicProvider,
|
||||
}
|
||||
adapter = protocol_to_class.get(descriptor.wire_protocol)
|
||||
if adapter is None: # pragma: no cover — unreachable while the map is total
|
||||
raise ProviderNotFoundError(
|
||||
f"No adapter implements wire protocol {descriptor.wire_protocol!r}"
|
||||
)
|
||||
return adapter
|
||||
|
||||
def build(self, provider_id: str, conf: Dict[str, Any]):
|
||||
"""Instantiate a ready-to-use provider adapter.
|
||||
|
||||
The descriptor's ``default_model`` fills in a missing/blank ``model`` so
|
||||
a half-written config still produces a working provider instead of an
|
||||
empty model id that only fails once the request hits the gateway.
|
||||
"""
|
||||
descriptor = self.get(provider_id)
|
||||
adapter = self.adapter_class(descriptor.provider_id)
|
||||
merged = dict(conf or {})
|
||||
merged["model"] = descriptor.resolve_model(merged.get("model", ""))
|
||||
return adapter(merged)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Process-wide default registry.
|
||||
#
|
||||
# Built lazily under a lock: several UI screens can ask for it during startup
|
||||
# from different threads, and double-construction would hand out two catalogues
|
||||
# whose discovered model lists then drift apart.
|
||||
# --------------------------------------------------------------------------- #
|
||||
_default_registry: Optional[ProviderRegistry] = None
|
||||
_default_lock = threading.Lock()
|
||||
|
||||
|
||||
def default_registry() -> ProviderRegistry:
|
||||
"""The shared registry seeded with :data:`BUILTIN_DESCRIPTORS`."""
|
||||
global _default_registry
|
||||
if _default_registry is None:
|
||||
with _default_lock:
|
||||
if _default_registry is None:
|
||||
_default_registry = ProviderRegistry(BUILTIN_DESCRIPTORS)
|
||||
return _default_registry
|
||||
|
||||
|
||||
def reset_default_registry() -> None:
|
||||
"""Drop the cached registry — test-support hook so one test's registrations
|
||||
cannot leak into the next."""
|
||||
global _default_registry
|
||||
with _default_lock:
|
||||
_default_registry = None
|
||||
|
||||
|
||||
__all__ = [
|
||||
"BUILTIN_DESCRIPTORS",
|
||||
"ProviderNotFoundError",
|
||||
"ProviderRegistry",
|
||||
"default_registry",
|
||||
"reset_default_registry",
|
||||
]
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure sandbox package: OS-specific sandbox capability adapters."""
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Infrastructure telemetry package: CanonicalAuditLogger and token usage sinks."""
|
||||
|
||||
@@ -0,0 +1,288 @@
|
||||
"""Token-usage telemetry as a publish/subscribe seam (R03-T06).
|
||||
|
||||
Before this module every provider adapter reached straight into
|
||||
``core/usage_tracker.py`` and wrote a dashboard row itself, which meant the
|
||||
provider layer owned a telemetry policy decision ("where do usage numbers go?")
|
||||
and no test could observe a turn's token accounting without touching the real
|
||||
``~/.cowork_local/usage/`` files.
|
||||
|
||||
Now a provider only *describes what happened* — it publishes an immutable
|
||||
:class:`UsageEvent` — and subscribers decide what to do with it. The default
|
||||
subscriber, :class:`UsageTrackerSink`, forwards to the existing usage tracker so
|
||||
the Dashboard keeps working byte-for-byte; tests swap in
|
||||
:class:`InMemoryUsageSink` and assert on the events directly.
|
||||
|
||||
Every publish path is failure-tolerant on purpose: telemetry must never be the
|
||||
reason a chat turn dies, which is the same contract
|
||||
``usage_tracker.record()`` already documents.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
|
||||
|
||||
logger = logging.getLogger("cowork_local.telemetry.usage")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class UsageEvent:
|
||||
"""One provider turn's token accounting.
|
||||
|
||||
Frozen so a subscriber cannot mutate an event the next subscriber in the
|
||||
chain is about to receive. ``source``/``label`` stay optional: the usage
|
||||
tracker already derives them from thread-local context set by whoever ran
|
||||
the turn, and a provider adapter has no business knowing which UI surface
|
||||
invoked it.
|
||||
"""
|
||||
|
||||
provider: str
|
||||
model: str
|
||||
input_tokens: int = 0
|
||||
output_tokens: int = 0
|
||||
cached_tokens: int = 0
|
||||
# True when the counts are a ~4-chars-per-token approximation because the
|
||||
# gateway never sent a usage block. Surfaced in the Dashboard so users know
|
||||
# which rows are measured and which are guessed.
|
||||
estimated: bool = False
|
||||
source: Optional[str] = None # None -> tracker's thread-local context
|
||||
label: Optional[str] = None # None -> tracker's thread-local context
|
||||
extras: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def total_tokens(self) -> int:
|
||||
"""Billable token count for this turn (cached tokens are already part
|
||||
of the input count reported by every gateway we support, so adding them
|
||||
again would double-count)."""
|
||||
return int(self.input_tokens) + int(self.output_tokens)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""JSON-friendly view, using the same short keys as the usage tracker's
|
||||
on-disk rows so a caller can diff an event against a stored row."""
|
||||
return {
|
||||
"provider": self.provider,
|
||||
"model": self.model,
|
||||
"in": int(self.input_tokens),
|
||||
"out": int(self.output_tokens),
|
||||
"cache": int(self.cached_tokens),
|
||||
"estimated": bool(self.estimated),
|
||||
"source": self.source or "",
|
||||
"label": self.label or "",
|
||||
}
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class UsageEventSink(Protocol):
|
||||
"""Anything that can receive :class:`UsageEvent`s.
|
||||
|
||||
A ``Protocol`` rather than a base class so a plain object (or a test double,
|
||||
or a Qt-side adapter that re-emits a signal) qualifies without inheriting
|
||||
from infrastructure code.
|
||||
"""
|
||||
|
||||
def emit(self, event: UsageEvent) -> None:
|
||||
"""Handle one usage event. Implementations MUST NOT raise."""
|
||||
|
||||
|
||||
class UsageTrackerSink:
|
||||
"""Default subscriber: writes each event through ``core/usage_tracker.py``.
|
||||
|
||||
Keeps the existing Dashboard/telemetry pipeline (daily JSONL files, shared
|
||||
cross-machine mirror, per-thread accumulator) as the single writer, so
|
||||
routing this through an event seam changed the plumbing without changing
|
||||
a single stored byte.
|
||||
"""
|
||||
|
||||
def __init__(self, recorder=None) -> None:
|
||||
# The recorder is injectable so a test can verify the forwarding
|
||||
# contract without importing the real tracker (and its config paths).
|
||||
self._recorder = recorder
|
||||
|
||||
def _resolve_recorder(self):
|
||||
"""Late-bind ``usage_tracker.record``.
|
||||
|
||||
Imported on first use rather than at module import so telemetry stays
|
||||
out of the import graph of anything that merely *declares* a sink.
|
||||
"""
|
||||
if self._recorder is None:
|
||||
from ...core import usage_tracker as tracker
|
||||
|
||||
self._recorder = tracker.record
|
||||
return self._recorder
|
||||
|
||||
def emit(self, event: UsageEvent) -> None:
|
||||
"""Forward one event; swallow every failure (telemetry is never fatal)."""
|
||||
try:
|
||||
record = self._resolve_recorder()
|
||||
if event.source is None:
|
||||
# Normal path: the worker thread already tagged its own
|
||||
# source/label via set_context(), so record() attributes the row.
|
||||
record(
|
||||
event.provider, event.model,
|
||||
int(event.input_tokens), int(event.output_tokens),
|
||||
int(event.cached_tokens), estimated=bool(event.estimated),
|
||||
)
|
||||
return
|
||||
|
||||
# Event carries its own attribution: apply it for this single write
|
||||
# and restore the thread's previous context afterwards, so a
|
||||
# re-attributed event cannot silently relabel every later turn that
|
||||
# runs on the same worker thread.
|
||||
from ...core import usage_tracker as tracker
|
||||
|
||||
previous_source, previous_label = tracker.current_context()
|
||||
tracker.set_context(event.source, event.label or "")
|
||||
try:
|
||||
record(
|
||||
event.provider, event.model,
|
||||
int(event.input_tokens), int(event.output_tokens),
|
||||
int(event.cached_tokens), estimated=bool(event.estimated),
|
||||
)
|
||||
finally:
|
||||
tracker.set_context(previous_source, previous_label)
|
||||
except Exception: # noqa: BLE001 — usage tracking must never break a turn
|
||||
logger.debug("usage sink: forwarding to usage_tracker failed", exc_info=True)
|
||||
|
||||
|
||||
class InMemoryUsageSink:
|
||||
"""Collects events in a list — the test double for usage assertions."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.events: List[UsageEvent] = []
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def emit(self, event: UsageEvent) -> None:
|
||||
"""Append under a lock: parallel Co4E flows publish from several worker
|
||||
threads at once and ``list.append`` alone would still be atomic, but the
|
||||
lock also makes :meth:`snapshot` a consistent read."""
|
||||
with self._lock:
|
||||
self.events.append(event)
|
||||
|
||||
def snapshot(self) -> List[UsageEvent]:
|
||||
"""A copy of everything received so far."""
|
||||
with self._lock:
|
||||
return list(self.events)
|
||||
|
||||
def clear(self) -> None:
|
||||
with self._lock:
|
||||
self.events.clear()
|
||||
|
||||
@property
|
||||
def total_tokens(self) -> int:
|
||||
return sum(e.total_tokens for e in self.snapshot())
|
||||
|
||||
|
||||
class CompositeUsageSink:
|
||||
"""Fans one event out to several subscribers.
|
||||
|
||||
This is what makes the seam useful beyond the Dashboard: a future consumer
|
||||
(per-workspace budget guard, live cost meter) subscribes alongside the
|
||||
tracker instead of patching provider code again. One failing subscriber is
|
||||
logged and skipped so it cannot starve the others.
|
||||
"""
|
||||
|
||||
def __init__(self, sinks=None) -> None:
|
||||
self._sinks: List[UsageEventSink] = list(sinks or ())
|
||||
self._lock = threading.RLock()
|
||||
|
||||
def add(self, sink: UsageEventSink) -> None:
|
||||
with self._lock:
|
||||
self._sinks.append(sink)
|
||||
|
||||
def remove(self, sink: UsageEventSink) -> None:
|
||||
"""Detach a subscriber; a sink that was never added is ignored so
|
||||
teardown code can call this unconditionally."""
|
||||
with self._lock:
|
||||
if sink in self._sinks:
|
||||
self._sinks.remove(sink)
|
||||
|
||||
def sinks(self) -> List[UsageEventSink]:
|
||||
with self._lock:
|
||||
return list(self._sinks)
|
||||
|
||||
def emit(self, event: UsageEvent) -> None:
|
||||
for sink in self.sinks():
|
||||
try:
|
||||
sink.emit(event)
|
||||
except Exception: # noqa: BLE001 — one bad subscriber must not stop the rest
|
||||
logger.debug("usage sink: subscriber %r failed", sink, exc_info=True)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Process-wide sink.
|
||||
#
|
||||
# Providers publish through the module-level helpers below rather than holding a
|
||||
# sink reference, because a provider instance is created fresh for every turn
|
||||
# (see AppContext.build_provider_for) and would otherwise have to be handed the
|
||||
# telemetry wiring on every construction.
|
||||
# --------------------------------------------------------------------------- #
|
||||
_sink_lock = threading.RLock()
|
||||
_sink: Optional[CompositeUsageSink] = None
|
||||
|
||||
|
||||
def get_usage_sink() -> CompositeUsageSink:
|
||||
"""The shared sink, seeded with :class:`UsageTrackerSink` on first use."""
|
||||
global _sink
|
||||
if _sink is None:
|
||||
with _sink_lock:
|
||||
if _sink is None:
|
||||
_sink = CompositeUsageSink([UsageTrackerSink()])
|
||||
return _sink
|
||||
|
||||
|
||||
def set_usage_sink(sink: Optional[CompositeUsageSink]) -> None:
|
||||
"""Replace the shared sink (``None`` restores the default on next use).
|
||||
|
||||
Used by tests and by the app shell when it wants a different fan-out; kept
|
||||
explicit so nothing silently reconfigures telemetry mid-run.
|
||||
"""
|
||||
global _sink
|
||||
with _sink_lock:
|
||||
_sink = sink
|
||||
|
||||
|
||||
def subscribe(sink: UsageEventSink) -> UsageEventSink:
|
||||
"""Attach an extra subscriber to the shared sink and return it (so callers
|
||||
can keep the handle for a later :func:`unsubscribe`)."""
|
||||
get_usage_sink().add(sink)
|
||||
return sink
|
||||
|
||||
|
||||
def unsubscribe(sink: UsageEventSink) -> None:
|
||||
"""Detach a subscriber previously passed to :func:`subscribe`."""
|
||||
get_usage_sink().remove(sink)
|
||||
|
||||
|
||||
def publish(event: UsageEvent) -> None:
|
||||
"""Publish one usage event to every subscriber.
|
||||
|
||||
Never raises: called from inside a provider's streaming loop, where an
|
||||
exception would abort an otherwise successful turn.
|
||||
"""
|
||||
try:
|
||||
get_usage_sink().emit(event)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.debug("usage sink: publish failed", exc_info=True)
|
||||
|
||||
|
||||
def estimate_tokens(text: str) -> int:
|
||||
"""~4 chars per token approximation, re-exported so provider adapters need
|
||||
exactly ONE telemetry import instead of also importing the tracker."""
|
||||
return max(0, len(text or "") // 4)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"UsageEvent",
|
||||
"UsageEventSink",
|
||||
"UsageTrackerSink",
|
||||
"InMemoryUsageSink",
|
||||
"CompositeUsageSink",
|
||||
"get_usage_sink",
|
||||
"set_usage_sink",
|
||||
"subscribe",
|
||||
"unsubscribe",
|
||||
"publish",
|
||||
"estimate_tokens",
|
||||
]
|
||||
Reference in New Issue
Block a user