mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-16 20:31:33 +08:00
infstellar
257 lines
16 KiB
Python
257 lines
16 KiB
Python
# Executor 的 OpenCode 只读观测，读取范围来自当前任务的可信执行绑定。
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import sqlite3
|
||
import time
|
||
from contextlib import closing
|
||
from pathlib import Path
|
||
from threading import Event, Thread
|
||
from typing import Any, Callable
|
||
|
||
from loguru import logger
|
||
from opentelemetry import metrics, trace
|
||
from opentelemetry.context import Context
|
||
from pydantic import BaseModel, ConfigDict
|
||
|
||
from runtime_telemetry import emit_event, protected_content, record_content
|
||
|
||
|
||
class SavedContent(BaseModel):
|
||
"""读取固定 OpenCode 版本的结构化 JSON，保留来源提供的扩展字段。"""
|
||
|
||
model_config = ConfigDict(extra="allow")
|
||
role: str | None = None
|
||
type: str | None = None
|
||
time: dict[str, Any] | None = None
|
||
state: dict[str, Any] | None = None
|
||
error: dict[str, Any] | None = None
|
||
modelID: str | None = None
|
||
providerID: str | None = None
|
||
|
||
|
||
def read_saved_records(database: Path, *, directory: Path, session_id: str) -> list[dict[str, Any]]:
|
||
"""读取已绑定主 session 和相同目录的子 session；250ms 等待并限制查询时间。"""
|
||
if not session_id:
|
||
raise ValueError("sqlite_session_binding_missing")
|
||
if not database.is_file() or database.is_symlink():
|
||
raise ValueError("sqlite_database_missing_or_symlink")
|
||
started = time.monotonic()
|
||
with closing(sqlite3.connect(database.absolute().as_uri() + "?mode=ro", uri=True, timeout=0.25)) as connection:
|
||
connection.execute("PRAGMA query_only = ON")
|
||
connection.set_progress_handler(lambda: int(time.monotonic() - started > 0.25), 1000)
|
||
connection.execute("BEGIN")
|
||
root = connection.execute("SELECT parent_id, directory FROM session WHERE id = ?", (session_id,)).fetchone()
|
||
if root is None or root[0] is not None or Path(root[1]).resolve() != directory.resolve():
|
||
raise ValueError("sqlite_session_directory_mismatch")
|
||
sessions = connection.execute("""
|
||
WITH RECURSIVE bound(id, directory) AS (
|
||
SELECT id, directory FROM session WHERE id = ?
|
||
UNION SELECT session.id, session.directory FROM session JOIN bound ON session.parent_id = bound.id
|
||
) SELECT id, directory FROM bound
|
||
""", (session_id,)).fetchall()
|
||
if any(Path(row[1]).resolve() != directory.resolve() for row in sessions):
|
||
raise ValueError("sqlite_child_session_directory_mismatch")
|
||
identifiers = [row[0] for row in sessions]
|
||
placeholders = ",".join("?" for identifier in identifiers)
|
||
session_rows = connection.execute(f"SELECT id, time_created, time_updated, parent_id, title, version, cost, tokens_input, tokens_output, tokens_reasoning, tokens_cache_read, tokens_cache_write FROM session WHERE id IN ({placeholders}) ORDER BY time_updated, id", identifiers)
|
||
records = [{"kind": "session", "id": row[0], "session_id": row[0], "created_ms": row[1], "updated_ms": row[2], "data": dict(zip(("parent_id", "title", "version", "cost", "tokens_input", "tokens_output", "tokens_reasoning", "tokens_cache_read", "tokens_cache_write"), row[3:]))} for row in session_rows]
|
||
for table in ("message", "part"):
|
||
message_column = "message_id" if table == "part" else "id"
|
||
rows = connection.execute(f"SELECT id, session_id, time_created, time_updated, data, {message_column} FROM {table} WHERE session_id IN ({placeholders}) ORDER BY time_updated, id", identifiers)
|
||
for identifier, bound_session, created, updated, data, message_id in rows:
|
||
content = SavedContent.model_validate_json(data).model_dump(exclude_none=True)
|
||
records.append({"kind": table, "id": identifier, "message_id": message_id, "session_id": bound_session, "created_ms": created, "updated_ms": updated, "data": content})
|
||
connection.rollback()
|
||
return records
|
||
|
||
|
||
class SQLiteTelemetry:
|
||
"""后台采集当前 attempt 的 SQLite 更新，恢复读取时建立终态基线。"""
|
||
|
||
def __init__(self, *, database: Path, directory: Path, session: Callable[[], str | None], attributes: dict[str, Any], context: Context, started_at: float, secrets: tuple[str, ...]):
|
||
self.database = database
|
||
self.directory = directory
|
||
self.session = session
|
||
self.attributes = attributes
|
||
self.context = context
|
||
self.started_at = started_at
|
||
self.secrets = secrets
|
||
self.stop = Event()
|
||
self.thread = Thread(target=self._loop, name="opencode-sqlite-telemetry", daemon=True)
|
||
self.versions: dict[tuple[str, str], tuple[int, str]] = {}
|
||
self.terminals: set[tuple[str, str]] = set()
|
||
self.baseline = True
|
||
self.failed_since: float | None = None
|
||
self.last_success = 0.0
|
||
self.usage_values: dict[str, dict[str, float]] = {}
|
||
self.usage_baselines: set[str] = set()
|
||
self.message_models: dict[str, dict[str, str]] = {}
|
||
meter = metrics.get_meter("mememeow.agent")
|
||
self.success_time = meter.create_gauge("agent_sqlite_last_success", unit="s")
|
||
self.failures = meter.create_counter("agent_sqlite_read_errors", unit="{error}")
|
||
self.active_failures = meter.create_up_down_counter("agent_sqlite_active_failures", unit="{source}")
|
||
self.active_sources = meter.create_up_down_counter("agent_sqlite_active_sources", unit="{source}")
|
||
self.terminal_count = meter.create_counter("agent_saved_terminals", unit="{record}")
|
||
self.duration = meter.create_histogram("agent_saved_duration", unit="s")
|
||
self.tokens = meter.create_counter("agent_saved_tokens", unit="{token}")
|
||
self.cost = meter.create_counter("agent_saved_cost", unit="USD")
|
||
self.usage_issues = meter.create_counter("agent_usage_issues", unit="{field}")
|
||
self.tool_issues = meter.create_counter("agent_tool_record_issues", unit="{record}")
|
||
self.baseline_time = meter.create_gauge("agent_usage_baseline_time", unit="s")
|
||
|
||
def start(self) -> None:
|
||
"""在进程外启动独立读取线程，业务执行不等待 SQLite 观测。"""
|
||
self.active_sources.add(1)
|
||
self.thread.start()
|
||
|
||
def close(self) -> None:
|
||
"""停止周期读取并完成最后一次有界读取，保留终止时的来源状态。"""
|
||
self.stop.set()
|
||
self.thread.join()
|
||
self.active_sources.add(-1)
|
||
if self.failed_since is not None:
|
||
self.active_failures.add(-1)
|
||
|
||
def _loop(self) -> None:
|
||
"""每六秒取得绑定快照，结束时读取最终保存的记录。"""
|
||
while not self.stop.is_set():
|
||
self._poll()
|
||
self.stop.wait(6.0)
|
||
self._poll(final=True)
|
||
|
||
def _poll(self, *, final: bool = False) -> None:
|
||
"""记录真实读取错误及中断区间，恢复后已有终态用于调查。"""
|
||
session_id = self.session()
|
||
observed_at = time.time()
|
||
try:
|
||
if session_id is None:
|
||
if final:
|
||
raise ValueError("sqlite_session_binding_missing_at_attempt_end")
|
||
return
|
||
records = read_saved_records(self.database, directory=self.directory, session_id=session_id)
|
||
self.message_models = {record["id"]: {label: record["data"][field] for label, field in (("provider", "providerID"), ("model", "modelID")) if isinstance(record["data"].get(field), str)} for record in records if record["kind"] == "message"}
|
||
if self.baseline:
|
||
self.baseline_time.set(observed_at)
|
||
emit_event("sqlite_diagnostics", {"event": "usage_baseline", "observed_at": observed_at, "started_at": self.started_at, "measurement_scope": "已有完成步骤建立基线；后续完成步骤及有效正向修订计数", **self.attributes})
|
||
for record in records:
|
||
self._record(record, observed_at)
|
||
except (sqlite3.Error, OSError, ValueError) as error:
|
||
self.baseline = True
|
||
self.failures.add(1, {"error_type": type(error).__name__})
|
||
if self.failed_since is None:
|
||
self.failed_since = observed_at
|
||
self.active_failures.add(1)
|
||
with trace.get_tracer("mememeow.agent").start_as_current_span("agent.sqlite.read_error", context=self.context, attributes=self.attributes) as span:
|
||
record_content(span, "source.error", {"reason": str(error), "error_code": getattr(error, "sqlite_errorname", type(error).__name__), "missing_since": self.failed_since, "last_success": self.last_success}, secrets=self.secrets)
|
||
logger.error("agent_sqlite_read_error: {}", protected_content(str(error), self.secrets))
|
||
emit_event("sqlite_diagnostics", {"event": "read_error", "reason": str(error), "error_code": getattr(error, "sqlite_errorname", type(error).__name__), "missing_since": self.failed_since, "last_success": self.last_success, **self.attributes})
|
||
return
|
||
if self.failed_since is not None:
|
||
with trace.get_tracer("mememeow.agent").start_as_current_span("agent.sqlite.read_recovered", context=self.context, attributes=self.attributes) as span:
|
||
record_content(span, "source.missing_interval", {"from": self.failed_since, "to": observed_at}, secrets=self.secrets)
|
||
emit_event("sqlite_diagnostics", {"event": "read_recovered", "missing_from": self.failed_since, "missing_to": observed_at, **self.attributes})
|
||
self.failed_since = None
|
||
self.active_failures.add(-1)
|
||
self.last_success = observed_at
|
||
self.success_time.set(observed_at)
|
||
self.baseline = False
|
||
|
||
def _usage_issue(self, record: dict[str, Any], field: str, reason: str) -> None:
|
||
"""用有限类别记录字段覆盖问题，来源标识只进入受控诊断。"""
|
||
self.usage_issues.add(1, {"field": field, "reason": reason})
|
||
emit_event("sqlite_diagnostics", {"event": "usage_issue", "field": field, "reason": reason, "record_id": record["id"], "session_id": record["session_id"], **self.attributes})
|
||
|
||
def _record_usage(self, record: dict[str, Any]) -> None:
|
||
"""完成步骤提供唯一用量来源，按字段累计有效新增量并保存读取基线。"""
|
||
content = record["data"]
|
||
tokens = content.get("tokens")
|
||
tokens = tokens if isinstance(tokens, dict) else {}
|
||
cache = tokens.get("cache")
|
||
cache = cache if isinstance(cache, dict) else {}
|
||
values = {"input": tokens.get("input"), "output": tokens.get("output"), "reasoning": tokens.get("reasoning"), "cache_read": cache.get("read"), "cache_write": cache.get("write"), "cost": content.get("cost")}
|
||
labels = {"provider": "unknown", "model": "unknown", **self.message_models.get(record.get("message_id"), {})}
|
||
if "unknown" in labels.values():
|
||
self._usage_issue(record, "association", "message_model_missing")
|
||
previous = self.usage_values.setdefault(record["id"], {})
|
||
establish_baseline = self.baseline and (self.last_success > 0 or self.failed_since is not None or record["created_ms"] / 1000 < self.started_at)
|
||
if establish_baseline:
|
||
self.usage_baselines.add(record["id"])
|
||
for field, value in values.items():
|
||
if value is None:
|
||
self._usage_issue(record, field, "missing")
|
||
continue
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0 or (field != "cost" and value != int(value)):
|
||
self._usage_issue(record, field, "invalid")
|
||
continue
|
||
before = previous.get(field, 0)
|
||
if value < before:
|
||
self._usage_issue(record, field, "decreased")
|
||
continue
|
||
missing_baseline = record["id"] in self.usage_baselines and field not in previous
|
||
previous[field] = value
|
||
if establish_baseline or missing_baseline:
|
||
if missing_baseline and not establish_baseline:
|
||
self._usage_issue(record, field, "baseline_field_became_valid")
|
||
continue
|
||
delta = value - before
|
||
if field == "cost":
|
||
self.cost.add(delta, labels)
|
||
else:
|
||
self.tokens.add(delta, {**labels, "token_type": field})
|
||
|
||
def _record(self, record: dict[str, Any], observed_at: float) -> None:
|
||
"""按来源 ID 统计新的终态；工具按保存名称分组，缺失名称进入诊断。"""
|
||
identity = (record["kind"], record["id"])
|
||
content = record["data"]
|
||
version = (record["updated_ms"], json.dumps(content, sort_keys=True))
|
||
if self.versions.get(identity) == version and not (self.baseline and content.get("type") == "step-finish"):
|
||
return
|
||
self.versions[identity] = version
|
||
if record["kind"] == "part" and content.get("type") == "step-finish":
|
||
self._record_usage(record)
|
||
unit = "session" if record["kind"] == "session" else "message" if record["kind"] == "message" else "tool" if content.get("type") == "tool" else "step" if content.get("type") == "step-finish" else "content"
|
||
clock = content.get("time") or (content.get("state") or {}).get("time") or {}
|
||
terminal = (record["kind"] == "message" and content.get("role") == "assistant" and (clock.get("completed") is not None or content.get("error") is not None)) or (unit == "tool" and (content.get("state") or {}).get("status") in {"completed", "error"}) or unit == "step"
|
||
labels = {"unit": unit, "result": "failed" if content.get("error") or (content.get("state") or {}).get("status") == "error" else "succeeded"}
|
||
labels.update({label: content[field] for label, field in (("provider", "providerID"), ("model", "modelID")) if isinstance(content.get(field), str)})
|
||
tool = content.get("tool") if unit == "tool" else None
|
||
if unit == "tool":
|
||
labels.update(self.message_models.get(record.get("message_id"), {}))
|
||
labels["result"] = "failed" if (content.get("state") or {}).get("status") == "error" else "succeeded"
|
||
if isinstance(tool, str) and tool.strip():
|
||
labels["tool"] = tool
|
||
for result in ("succeeded", "failed"):
|
||
self.terminal_count.add(0, {**labels, "result": result})
|
||
else:
|
||
reason = "missing" if tool is None else "invalid"
|
||
self.tool_issues.add(1, {"field": "tool", "reason": reason})
|
||
emit_event("sqlite_diagnostics", {"event": "tool_record_issue", "field": "tool", "reason": reason, "record_id": record["id"], "session_id": record["session_id"], **self.attributes})
|
||
if terminal and identity not in self.terminals:
|
||
self.terminals.add(identity)
|
||
if not self.baseline or (self.last_success == 0 and self.failed_since is None and record["created_ms"] / 1000 >= self.started_at):
|
||
self.terminal_count.add(1, labels)
|
||
start = clock.get("created", clock.get("start"))
|
||
end = clock.get("completed", clock.get("end"))
|
||
if isinstance(start, (int, float)) and isinstance(end, (int, float)) and end >= start:
|
||
self.duration.record((end - start) / 1000, {"unit": unit})
|
||
associated = record["created_ms"] / 1000 >= self.started_at and record["updated_ms"] / 1000 <= observed_at
|
||
attributes = dict(self.attributes)
|
||
if not associated:
|
||
attributes.pop("attempt.id", None)
|
||
attributes.update({"source": "opencode_sqlite", "observation.unit": unit, "record.id": record["id"], "session.id": record["session_id"], "source.observed_at": observed_at, "attempt.association": "confirmed" if associated else "undetermined"})
|
||
if unit == "tool":
|
||
if "tool" in labels:
|
||
attributes["tool.name"] = labels["tool"]
|
||
status = (content.get("state") or {}).get("status")
|
||
if isinstance(status, str):
|
||
attributes["tool.status"] = status
|
||
with trace.get_tracer("mememeow.agent").start_as_current_span(f"agent.saved.{unit}", context=self.context if associated else Context(), attributes=attributes) as span:
|
||
record_content(span, "saved.record", record, secrets=self.secrets)
|
||
span.set_attribute("content.source", "opencode_sqlite")
|
||
if not clock:
|
||
span.set_attribute("content.missing_fields", "saved_time_fields")
|
||
elif terminal and not (isinstance(clock.get("created", clock.get("start")), (int, float)) and isinstance(clock.get("completed", clock.get("end")), (int, float))):
|
||
span.set_attribute("content.missing_fields", "saved_duration_fields")
|