feat(multimodal): 实现本地模型管线与请求用量配置

backend/app/services/index_service.py

@@ -97,6 +97,7 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob:
                 task_note_links = dict(conn.execute(
                     "SELECT task_id, note_id FROM tasks WHERE note_id IS NOT NULL"
                 ).fetchall())
+                media_links = conn.execute("SELECT job_id,revision,options_hash,note_id FROM media_notes").fetchall()
                 repository.clear_all(conn=conn)
                 await vector_store.clear(conn=conn)
                 for parsed, prepared in prepared_notes:
@@ -107,6 +108,9 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob:
                         "AND EXISTS (SELECT 1 FROM notes WHERE note_id = ?)",
                         (note_id, task_id, note_id),
                     )
+                for link in media_links:
+                    conn.execute("INSERT OR IGNORE INTO media_notes SELECT ?,?,?,? WHERE EXISTS (SELECT 1 FROM notes WHERE note_id=?)",
+                                 (*link, link["note_id"]))
         finally:
             conn.close()
     except BaseException as exc:

backend/app/services/media_notes.py

@@ -0,0 +1,55 @@
+"""Idempotent transcript export without overwriting an edited note."""
+import asyncio
+import hashlib
+from contextlib import closing
+
+from app.config import get_settings
+from app.database.db import connect, transaction
+from app.errors import ApiError
+from app.services import note_service
+from app.services.transcription_service import require_job
+
+_locks = {}
+
+
+async def create_transcript_note(job_id, options):
+    identity = (str(get_settings().db_path), job_id)
+    lock = _locks.setdefault(identity, asyncio.Lock())
+    async with lock:
+        job = require_job(job_id)
+        if job.status != "completed":
+            raise ApiError(409, "TRANSCRIPT_NOT_READY", "Only completed transcripts can become notes.")
+        options_hash = hashlib.sha256(options.model_dump_json().encode()).hexdigest()
+        with closing(connect()) as conn:
+            row = conn.execute("SELECT note_id FROM media_notes WHERE job_id=? AND revision=? AND options_hash=?",
+                               (job_id, job.revision, options_hash)).fetchone()
+        if row:
+            return await note_service.get_note(row[0])
+        marker = f"<!-- transcription:{job_id}:{job.revision}:{options_hash} -->"
+        title = f"{options.title} · {job_id[-8:]}-r{job.revision}-{options_hash[:6]}"
+        lines = [marker, f"# {options.title}", "", f"[源音频](/#/media?job={job_id})", ""]
+        if job.segments:
+            for segment in job.segments:
+                prefix = []
+                if options.include_timestamps:
+                    seconds = segment.start_time
+                    label = f"{int(seconds // 60):02}:{int(seconds % 60):02}"
+                    prefix.append(f"[{label}](/#/media?job={job_id}&time={seconds})")
+                if options.include_speakers and segment.speaker:
+                    prefix.append(job.speaker_names.get(segment.speaker, segment.speaker))
+                lines.append(" ".join([*prefix, segment.text]))
+                lines.append("")
+        else:
+            lines.append(job.text or "")
+        try:
+            note = await note_service.create_note(title=title, markdown="\n".join(lines), folder=options.folder, tags=["转写"])
+        except ApiError as exc:
+            if exc.code != "RESOURCE_CONFLICT" or "note_id" not in exc.details:
+                raise
+            # Recover a crash between successful note creation and linking the job.
+            note = await note_service.get_note(exc.details["note_id"])
+            if note is None or marker not in note.markdown:
+                raise
+        with closing(connect()) as conn, transaction(conn):
+            conn.execute("INSERT OR IGNORE INTO media_notes VALUES (?,?,?,?)", (job_id, job.revision, options_hash, note.note_id))
+        return note

backend/app/services/note_service.py

@@ -17,7 +17,7 @@ from app.contracts import Note, NoteBlock, NoteSummary
 from app.database.db import connect, transaction
 from app.errors import ApiError
 from app.knowledge.parser import ParsedNote, parse_note
-from app.retrieval.embedding import HashEmbeddingProvider
+from app.local_models.runtime import LocalEmbedding
 from app.retrieval import routed_vectors
 from app.retrieval.vectorstore import SqliteVecStore, VectorRecord
 from app.services.coordination import serialized_vault_mutation
@@ -28,8 +28,8 @@ from app.services.vault_paths import (
     safe_note_filename,
 )
 
-# 轻量实现实例（无状态，可直接复用）；接入真实模型后替换为对应 Provider
-embedding = HashEmbeddingProvider()
+# 真实模型接口不在 API 进程加载权重；测试可显式替换该实例。
+embedding = LocalEmbedding()
 vector_store = SqliteVecStore()
 
 
@@ -80,6 +80,10 @@ PreparedIndex = tuple[list[list[float]], routed_vectors.RemoteEmbeddings | None]
 async def prepare_note_index(parsed: ParsedNote) -> PreparedIndex:
     """Compute vectors before opening a write transaction (including API I/O)."""
     texts = [block.content for block in parsed.blocks]
+    if isinstance(embedding, LocalEmbedding):
+        # One routed invocation: API first, validated local fallback. No hash vectors.
+        remote = await routed_vectors.embed_remote(texts, accept_local=True)
+        return [], remote
     vectors = await embedding.embed_documents(texts)
     remote = await routed_vectors.embed_remote(texts)
     return vectors, remote
@@ -127,7 +131,8 @@ async def index_note(
             await vector_store.upsert(records, conn=conn)
             routed_vectors.store_remote(conn, [block.block_id for block in parsed.blocks], remote)
             repository.set_index_meta(
-                {"embedding_model": embedding.model_id, "embedding_dim": str(embedding.dim)},
+                {"embedding_model": remote.space_id if remote and isinstance(embedding, LocalEmbedding) else embedding.model_id,
+                 "embedding_dim": str(remote.dimensions if remote and isinstance(embedding, LocalEmbedding) else embedding.dim)},
                 conn=conn,
             )
     finally:

backend/app/services/transcription_service.py

@@ -1,65 +1,243 @@
-"""转写作业：API 优先，本地模型回退；保留已有 Host 文本入口。"""
-
+"""Persistent media jobs and replayable events; HTTP enqueues, tools await."""
 from __future__ import annotations
-
-from collections import OrderedDict
+import asyncio
+import hashlib
+import json
+from contextlib import closing
 from datetime import datetime, timezone
 from uuid import uuid4
-
-from app.contracts import TranscriptionJob
+from app.config import get_settings
+from app.contracts import TranscriptionJob, TranscriptionRequest, TranscriptEditRequest
+from app.database.db import connect, transaction
 from app.errors import ApiError
 from app.services.attachment_service import attachment_path
 
-_jobs: OrderedDict[str, TranscriptionJob] = OrderedDict()
-MAX_JOBS = 100
+TERMINAL = {"completed", "failed", "cancelled"}
+_tasks: dict[tuple[str, str], asyncio.Task] = {}
 
+def now():
+    return datetime.now(timezone.utc)
 
-async def create_transcription(attachment_id: str, language: str | None = None, *, diarization: bool = False) -> TranscriptionJob:
-    from app.container import container
-
-    source = attachment_path(attachment_id)
-    job = TranscriptionJob(
-        job_id=f"transcription_{uuid4().hex}",
-        attachment_id=attachment_id,
-        status="processing",
-        created_at=datetime.now(timezone.utc),
-    )
-    try:
-        if diarization:
-            # Speaker verification and diarization are different capabilities.
-            raise ApiError(501, "DIARIZATION_NOT_IMPLEMENTED", "说话人分离将在阶段 F 接入，当前不能忽略 diarization 请求。")
-        transcript = source if source.suffix.lower() in {".txt", ".md"} else attachment_path(f"{attachment_id}.txt")
-        # A saved transcript remains an explicit import path, never faked ASR.
-        if transcript.is_file() and (source == transcript or container.model_routing.configuration().transcription is None):
-            with transcript.open("rb") as handle:
-                content = handle.read(1024 * 1024 + 1)
-            if len(content) > 1024 * 1024:
-                raise ApiError(413, "TRANSCRIPT_TOO_LARGE", "Transcript exceeds 1 MiB.")
-            job.text = content.decode("utf-8")
-            if not job.text.strip():
-                raise ApiError(422, "TRANSCRIPT_EMPTY", "Transcript is empty.")
-            job.source = "sidecar"
-        else:
-            result = await container.model_routing.transcribe(source, language)
-            job.text = result.text
-            job.source = result.source
-            job.fallback_reason = result.fallback_reason
-        job.status = "completed"
-    except ApiError as exc:
-        job.status = "failed"
-        job.error_code = exc.code
-        job.error_message = exc.message
-        job.fallback_reason = exc.details.get("fallback_reason")
-    except (OSError, UnicodeError):
-        job.status = "failed"
-        job.error_code = "TRANSCRIPT_UNREADABLE"
-        job.error_message = "Transcript could not be read."
-    _jobs[job.job_id] = job
-    while len(_jobs) > MAX_JOBS:
-        _jobs.popitem(last=False)
-    return job.model_copy(deep=True)
-
+def task_key(job_id):
+    return str(get_settings().db_path), job_id
 
 def get_transcription(job_id: str) -> TranscriptionJob | None:
-    job = _jobs.get(job_id)
-    return job.model_copy(deep=True) if job else None
+    with closing(connect()) as conn:
+        row = conn.execute("SELECT job_json FROM media_jobs WHERE job_id=?", (job_id,)).fetchone()
+    return TranscriptionJob.model_validate_json(row[0]) if row else None
+
+def require_job(job_id):
+    job = get_transcription(job_id)
+    if job is None:
+        raise ApiError(404, "RESOURCE_NOT_FOUND", "Transcription job not found.")
+    return job
+
+def _event(conn, job, event, data=None):
+    sequence = conn.execute("SELECT COALESCE(MAX(sequence),-1)+1 FROM media_events WHERE job_id=?", (job.job_id,)).fetchone()[0]
+    conn.execute("INSERT INTO media_events VALUES (?,?,?,?,?)", (job.job_id, sequence, event,
+        json.dumps(data or {"status": job.status, "progress": job.progress}), now().isoformat()))
+
+def save(job, event):
+    job.updated_at = now()
+    with closing(connect()) as conn, transaction(conn):
+        conn.execute("UPDATE media_jobs SET status=?,job_json=?,updated_at=? WHERE job_id=?",
+            (job.status, job.model_dump_json(), job.updated_at.isoformat(), job.job_id))
+        _event(conn, job, event)
+
+def list_transcriptions(status=None, limit=50, offset=0):
+    where, args = (" WHERE status=?", [status]) if status else ("", [])
+    with closing(connect()) as conn:
+        total = conn.execute("SELECT COUNT(*) FROM media_jobs" + where, args).fetchone()[0]
+        rows = conn.execute("SELECT job_json FROM media_jobs" + where + " ORDER BY created_at DESC LIMIT ? OFFSET ?", [*args, limit, offset]).fetchall()
+    return {"items": [TranscriptionJob.model_validate_json(row[0]) for row in rows], "page": {"total": total, "limit": limit, "offset": offset}}
+
+def events(job_id, after=-1):
+    require_job(job_id)
+    with closing(connect()) as conn:
+        rows = conn.execute("SELECT * FROM media_events WHERE job_id=? AND sequence>? ORDER BY sequence LIMIT 200", (job_id, after)).fetchall()
+    return [{"job_id": job_id, "sequence": r["sequence"], "event": r["event"], "data": json.loads(r["data_json"]), "timestamp": r["timestamp"]} for r in rows]
+
+def recover_interrupted():
+    with closing(connect()) as conn:
+        rows = conn.execute("SELECT job_json FROM media_jobs WHERE status IN ('queued','running','processing')").fetchall()
+    for row in rows:
+        job = TranscriptionJob.model_validate_json(row[0])
+        if task_key(job.job_id) not in _tasks:
+            job.status, job.error_code = "failed", "TRANSCRIPTION_INTERRUPTED"
+            job.error_message = "AI Core stopped before completion. Retry to start a new attempt."
+            job.completed_at = now()
+            save(job, "Failed")
+
+async def shutdown():
+    tasks = [t for k, t in list(_tasks.items()) if k[0] == str(get_settings().db_path)]
+    for task in tasks:
+        task.cancel()
+    await asyncio.gather(*tasks, return_exceptions=True)
+
+async def create_transcription(attachment_id, language=None, *, diarization=False, local_only=False,
+        word_timestamps=False, idempotency_key=None, terminology=None, wait=True, previous_job_id=None):
+    request = TranscriptionRequest(attachment_id=attachment_id, language=language, diarization=diarization,
+        local_only=local_only, word_timestamps=word_timestamps, idempotency_key=idempotency_key, terminology=terminology or {})
+    source = attachment_path(attachment_id)
+    actual = source if source.is_file() else attachment_path(f"{attachment_id}.txt")
+    if not actual.is_file():
+        raise ApiError(404, "ATTACHMENT_NOT_FOUND", "Attachment was not found.")
+    if not 0 < actual.stat().st_size <= 25 * 1024 * 1024:
+        raise ApiError(413, "ATTACHMENT_TOO_LARGE", "Attachment must be between 1 byte and 25 MiB.")
+    digest = await asyncio.to_thread(lambda: hashlib.sha256(actual.read_bytes()).hexdigest())
+    from app.container import container
+    from app.local_models.runtime import configuration
+    from app.local_models.catalog import CATALOG
+    routing = container.model_routing.snapshot()
+    route = routing.configuration()
+    binding = None if local_only else route.transcription
+    snapshot = {"local_runtime": configuration().model_dump(), "models": {k:v.revision for k,v in CATALOG.items()},
+                "transcription": binding.model_dump() if binding else None}
+    if binding:
+        provider = routing.providers.get_any(binding.provider_id).config
+        snapshot["provider"] = provider.model_dump(exclude={"credential_id"})
+    fingerprint = hashlib.sha256((digest + request.model_dump_json(exclude={"idempotency_key"}) + json.dumps(snapshot, sort_keys=True)).encode()).hexdigest()
+    job = TranscriptionJob(job_id=f"transcription_{uuid4().hex}", attachment_id=attachment_id, status="queued",
+        created_at=now(), updated_at=now(), language=language, local_only=local_only, previous_job_id=previous_job_id, model_snapshot=snapshot)
+    existing = None
+    with closing(connect()) as conn, transaction(conn):
+        if idempotency_key:
+            existing = conn.execute("SELECT job_json,fingerprint FROM media_jobs WHERE idempotency_key=?", (idempotency_key,)).fetchone()
+        if existing:
+            if existing["fingerprint"] != fingerprint:
+                raise ApiError(409, "IDEMPOTENCY_CONFLICT", "This key was used for different input.")
+            job = TranscriptionJob.model_validate_json(existing["job_json"])
+        else:
+            conn.execute("INSERT INTO media_jobs VALUES (?,?,?,?,?,?,?,?)", (job.job_id, job.status,
+                job.model_dump_json(), request.model_dump_json(), job.created_at.isoformat(), job.updated_at.isoformat(), idempotency_key, fingerprint))
+            _event(conn, job, "Queued")
+    key = task_key(job.job_id)
+    if not existing:
+        task = asyncio.create_task(_execute(job.job_id, request, routing))
+        _tasks[key] = task
+        task.add_done_callback(lambda finished: _tasks.pop(key, None))
+    if wait and key in _tasks:
+        try:
+            await _tasks[key]
+        except asyncio.CancelledError:
+            await cancel(job.job_id)
+            raise
+        return require_job(job.job_id)
+    return job
+
+async def _execute(job_id, request, routing=None):
+    from app.container import container
+    job = require_job(job_id)
+    if job.status in TERMINAL:
+        return
+    from app.local_models.runtime import runtime_context, runtime_progress, RuntimeConfig
+    from app.contracts import TranscriptSegment
+    token = runtime_context.set(RuntimeConfig.model_validate(job.model_snapshot.get("local_runtime", {})))
+    def progress(message):
+        job.progress = max(0.0, min(0.99, message["progress"]))
+        job.segments.append(TranscriptSegment.model_validate(message["segment"]))
+        save(job, "SegmentReady")
+    progress_token = runtime_progress.set(progress)
+    job.status, job.started_at = "running", now()
+    save(job, "TranscriptionStarted")
+    cancelled = False
+    try:
+        source = attachment_path(job.attachment_id)
+        transcript = source if source.suffix.lower() in {".txt", ".md"} else attachment_path(f"{job.attachment_id}.txt")
+        if transcript.is_file() and (source == transcript or not source.exists()):
+            def read_transcript():
+                with transcript.open("rb") as stream:
+                    return stream.read(1024 * 1024 + 1)
+            content = await asyncio.to_thread(read_transcript)
+            if len(content) > 1024 * 1024:
+                raise ApiError(413, "TRANSCRIPT_TOO_LARGE", "Transcript exceeds 1 MiB.")
+            job.text, job.source = content.decode("utf-8"), "sidecar"
+        else:
+            result = await (routing or container.model_routing).transcribe(source, request.language, local_only=request.local_only)
+            job.text, job.source, job.fallback_reason = result.text, result.source, result.fallback_reason
+            job.segments = getattr(result, "segments", []) or []
+        if not job.text or not job.text.strip():
+            raise ApiError(422, "TRANSCRIPT_EMPTY", "Transcript is empty.")
+        if request.diarization:
+            if job.segments:
+                from app.local_models.runtime import runtime
+                from app.providers.base import ProviderError
+                try:
+                    result = await runtime.infer("eres2netv2", "diarization", {"source": str(source.resolve()),
+                        "segments": [s.model_dump() for s in job.segments]})
+                    for segment, speaker in zip(job.segments, result["speakers"], strict=True):
+                        segment.speaker = speaker
+                    job.warnings.append("DIARIZATION_SEGMENT_LEVEL")
+                except ProviderError:
+                    job.warnings.append("DIARIZATION_UNAVAILABLE")
+            else:
+                job.warnings.append("DIARIZATION_UNAVAILABLE")
+        if request.word_timestamps:
+            job.warnings.append("WORD_TIMESTAMPS_UNAVAILABLE")
+        job.original_text, job.original_segments = job.text, [s.model_copy(deep=True) for s in job.segments]
+        for original, replacement in request.terminology.items():
+            if original and original != replacement and original in job.text:
+                job.text = job.text.replace(original, replacement)
+                for segment in job.segments:
+                    segment.text = segment.text.replace(original, replacement)
+                job.corrections.append({"original": original, "replacement": replacement, "source": "terminology_postprocessing"})
+        job.status, job.progress = "completed", 1
+    except asyncio.CancelledError:
+        cancelled = True
+        job.status, job.error_code = "cancelled", "TRANSCRIPTION_CANCELLED"
+    except ApiError as exc:
+        job.status, job.error_code, job.error_message = "failed", exc.code, exc.message
+        job.fallback_reason = exc.details.get("fallback_reason")
+    except Exception:
+        job.status, job.error_code, job.error_message = "failed", "TRANSCRIPTION_FAILED", "Transcription could not be completed."
+    job.completed_at = now()
+    save(job, {"completed": "Completed", "cancelled": "Cancelled", "failed": "Failed"}[job.status])
+    runtime_context.reset(token)
+    runtime_progress.reset(progress_token)
+    if cancelled:
+        raise asyncio.CancelledError
+
+async def cancel(job_id):
+    job = require_job(job_id)
+    if job.status in TERMINAL:
+        return job
+    task = _tasks.get(task_key(job_id))
+    if task:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+    job = require_job(job_id)
+    if job.status not in TERMINAL:
+        job.status, job.error_code, job.completed_at = "cancelled", "TRANSCRIPTION_CANCELLED", now()
+        save(job, "Cancelled")
+    return job
+
+async def retry(job_id):
+    if require_job(job_id).error_code == "MEDIA_PURGED":
+        raise ApiError(409, "MEDIA_PURGED", "Purged jobs cannot be retried.")
+    if require_job(job_id).status not in {"failed", "cancelled"}:
+        raise ApiError(409, "TRANSCRIPTION_NOT_RETRYABLE", "Only failed or cancelled jobs can be retried.")
+    with closing(connect()) as conn:
+        raw = conn.execute("SELECT request_json FROM media_jobs WHERE job_id=?", (job_id,)).fetchone()[0]
+    request = TranscriptionRequest.model_validate_json(raw)
+    return await create_transcription(**request.model_dump(exclude={"idempotency_key"}), wait=False, previous_job_id=job_id)
+
+def edit(job_id, request: TranscriptEditRequest):
+    with closing(connect()) as conn, transaction(conn):
+        row = conn.execute("SELECT job_json FROM media_jobs WHERE job_id=?", (job_id,)).fetchone()
+        if not row:
+            raise ApiError(404, "RESOURCE_NOT_FOUND", "Transcription job not found.")
+        job = TranscriptionJob.model_validate_json(row[0])
+        if job.status != "completed":
+            raise ApiError(409, "TRANSCRIPT_NOT_READY", "Only completed transcripts can be edited.")
+        if job.revision != request.revision:
+            raise ApiError(409, "VERSION_CONFLICT", "Transcript has changed; reload before saving.")
+        ids = [s.segment_id for s in request.segments]
+        if len(ids) != len(set(ids)) or request.segments != sorted(request.segments, key=lambda s: s.start_time):
+            raise ApiError(422, "INVALID_SEGMENTS", "Segments must have unique IDs and ordered timestamps.")
+        conn.execute("INSERT INTO media_revisions VALUES (?,?,?)", (job_id, job.revision, job.model_dump_json()))
+        job.text, job.segments, job.speaker_names = request.text, request.segments, request.speaker_names
+        job.revision += 1
+        job.updated_at = now()
+        conn.execute("UPDATE media_jobs SET job_json=?,updated_at=? WHERE job_id=?", (job.model_dump_json(), job.updated_at.isoformat(), job_id))
+        _event(conn, job, "Revised", {"revision": job.revision})
+    return job

backend/app/services/usage_service.py

@@ -0,0 +1,132 @@
+"""Application-observed usage per actual HTTP attempt; never an account bill."""
+from __future__ import annotations
+
+import json
+import logging
+from contextlib import closing
+from contextvars import ContextVar
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from app.database.db import connect
+
+METRICS = ("input_tokens", "output_tokens", "total_tokens", "cache_hit_tokens", "cache_miss_tokens", "cache_write_tokens", "reasoning_tokens")
+logger = logging.getLogger(__name__)
+usage_context = ContextVar("usage_context", default=None)
+
+
+def connection():
+    conn = connect()
+    conn.execute("""CREATE TABLE IF NOT EXISTS model_usage (
+        attempt_id TEXT PRIMARY KEY, provider_id TEXT NOT NULL, model TEXT NOT NULL,
+        capability TEXT NOT NULL, source TEXT NOT NULL, started_at TEXT NOT NULL,
+        completed INTEGER NOT NULL, counters_json TEXT NOT NULL, raw_json TEXT NOT NULL)""")
+    conn.execute("CREATE INDEX IF NOT EXISTS usage_time_provider ON model_usage(started_at,provider_id,model)")
+    columns = {row[1] for row in conn.execute("PRAGMA table_info(model_usage)")}
+    for column in ("request_id", "run_id"):
+        if column not in columns:
+            conn.execute(f"ALTER TABLE model_usage ADD COLUMN {column} TEXT")
+    return conn
+
+
+def numeric_leaves(value, prefix=""):
+    """Keep known numerical counters only; vendor usage objects may contain arbitrary text."""
+    result = {}
+    if not isinstance(value, dict):
+        return result
+    allowed = {"prompt_tokens", "completion_tokens", "input_tokens", "output_tokens", "total_tokens", "cached_tokens",
+               "cache_read_input_tokens", "cache_creation_input_tokens", "prompt_cache_hit_tokens", "prompt_cache_miss_tokens",
+               "reasoning_tokens", "prompt_eval_count", "eval_count"}
+    for key, item in value.items():
+        path = f"{prefix}.{key}" if prefix else key
+        if key in allowed and type(item) is int and 0 <= item <= 2 ** 53:
+            result[path] = item
+        elif key in {"prompt_tokens_details", "completion_tokens_details", "input_tokens_details", "output_tokens_details"}:
+            result.update(numeric_leaves(item, path))
+    return result
+
+
+class UsageAttempt:
+    def __init__(self, provider_id, model, protocol, capability="chat", source="api"):
+        self.attempt_id = uuid4().hex
+        self.provider_id, self.model, self.protocol = provider_id, model, protocol
+        self.capability, self.source = capability, source
+        self.started_at = datetime.now(timezone.utc).isoformat()
+        self.raw = {}
+        self.completed = False
+        context = usage_context.get() or {}
+        self.request_id = context.get("request_id") or uuid4().hex
+        self.run_id = context.get("run_id")
+
+    def observe(self, data):
+        if not isinstance(data, dict):
+            return
+        values = [data.get("usage"), (data.get("message") or {}).get("usage") if isinstance(data.get("message"), dict) else None,
+                  (data.get("response") or {}).get("usage") if isinstance(data.get("response"), dict) else None]
+        if self.protocol == "ollama":
+            values.append(data)
+        for value in values:
+            for key, count in numeric_leaves(value).items():
+                self.raw[key] = max(self.raw.get(key, 0), count)
+        if data.get("type") in {"[DONE]", "response.completed", "message_stop"} or data.get("done") is True:
+            self.completed = True
+
+    def counters(self):
+        raw = self.raw
+        def first(*names):
+            return next((raw[name] for name in names if name in raw), None)
+        inputs = first("input_tokens", "prompt_tokens", "prompt_eval_count")
+        outputs = first("output_tokens", "completion_tokens", "eval_count")
+        hit = first("cache_read_input_tokens", "prompt_cache_hit_tokens", "input_tokens_details.cached_tokens", "prompt_tokens_details.cached_tokens")
+        write = first("cache_creation_input_tokens")
+        miss = first("prompt_cache_miss_tokens")
+        if self.protocol == "anthropic_messages":
+            miss = inputs
+            inputs = inputs + hit + write if inputs is not None and hit is not None and write is not None else None
+        elif miss is None and inputs is not None and hit is not None and 0 <= hit <= inputs:
+            miss = inputs - hit
+        if hit is not None and inputs is not None and hit > inputs:
+            hit, miss = None, None
+        return dict(input_tokens=inputs, output_tokens=outputs,
+                    total_tokens=inputs + outputs if inputs is not None and outputs is not None else first("total_tokens"),
+                    cache_hit_tokens=hit, cache_miss_tokens=miss, cache_write_tokens=write,
+                    reasoning_tokens=first("output_tokens_details.reasoning_tokens", "completion_tokens_details.reasoning_tokens"))
+
+    def persist(self):
+        try:
+            with closing(connection()) as conn:
+                conn.execute("INSERT OR REPLACE INTO model_usage VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
+                    self.attempt_id, self.provider_id, self.model, self.capability, self.source, self.started_at,
+                    int(self.completed), json.dumps(self.counters()), json.dumps(self.raw), self.request_id, self.run_id))
+        except Exception:
+            logger.warning("Usage persistence failed; model response remains available")
+
+
+def aggregate(start, end, provider_id=None, model=None, source=None):
+    query = "SELECT counters_json,completed FROM model_usage WHERE started_at>=? AND started_at<?"
+    args = [start.astimezone(timezone.utc).isoformat(), end.astimezone(timezone.utc).isoformat()]
+    for column, value in (("provider_id", provider_id), ("model", model), ("source", source)):
+        if value:
+            query += f" AND {column}=?"
+            args.append(value)
+    with closing(connection()) as conn:
+        rows = conn.execute(query, args).fetchall()
+        options = conn.execute("SELECT DISTINCT provider_id,model,source FROM model_usage ORDER BY provider_id,model").fetchall()
+    totals = {key: None for key in METRICS}
+    coverage = {key: 0 for key in METRICS}
+    hits, eligible_input, cache_requests = 0, 0, 0
+    for row in rows:
+        counts = json.loads(row[0])
+        for key in METRICS:
+            if counts.get(key) is not None:
+                totals[key] = (totals[key] or 0) + counts[key]
+                coverage[key] += 1
+        if counts.get("cache_hit_tokens") is not None and counts.get("cache_miss_tokens") is not None:
+            hits += counts["cache_hit_tokens"]
+            eligible_input += counts["input_tokens"] if counts.get("input_tokens") is not None else counts["cache_hit_tokens"] + counts["cache_miss_tokens"]
+            cache_requests += 1
+    return {"totals": totals, "coverage": coverage, "request_count": len(rows),
+            "complete_requests": sum(row[1] for row in rows), "cache_covered_requests": cache_requests,
+            "cache_hit_rate": hits / eligible_input if eligible_input else None,
+            "options": [dict(row) for row in options], "start": start, "end": end,
+            "scope": "application_observed_usage"}