187 lines
7.6 KiB
Python
187 lines
7.6 KiB
Python
"""Bounded, asynchronous operational diagnostics, separate from business/Trace data.
|
|
|
|
Only explicitly allowed metadata is stored. Never store prompts, tool arguments,
|
|
provider response bodies or raw exception messages in this diagnostic channel.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import math
|
|
import queue
|
|
import re
|
|
import sqlite3
|
|
import threading
|
|
import traceback
|
|
from contextvars import ContextVar
|
|
from contextlib import closing
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from app.config import get_settings
|
|
|
|
request_id: ContextVar[str] = ContextVar('log_request_id', default='')
|
|
agent_run_id: ContextVar[str] = ContextVar('log_agent_run_id', default='')
|
|
_allowed = {'run_id', 'task_id', 'note_id', 'job_id', 'provider_id', 'model',
|
|
'device', 'error_code', 'error_type', 'status', 'duration_ms', 'count',
|
|
'step', 'sequence', 'tool', 'method', 'route', 'request_id', 'fallback',
|
|
'frames', 'source', 'changed_fields'}
|
|
_safe = re.compile(r'[^\w .:/@{}\[\],()=+\-]', re.UNICODE)
|
|
|
|
|
|
def metadata(values: dict) -> dict:
|
|
result = {}
|
|
for key, value in values.items():
|
|
if key not in _allowed or value is None:
|
|
continue
|
|
if isinstance(value, (int, float, bool)):
|
|
if not isinstance(value, float) or math.isfinite(value):
|
|
result[key] = value
|
|
else:
|
|
text = str(value)
|
|
text = re.sub(r'(?i)(?:bearer\s+\S+|sk-[\w-]+)', '[REDACTED]', text)
|
|
result[key] = _safe.sub('', text)[:500]
|
|
return result
|
|
|
|
|
|
class LogStore:
|
|
def __init__(self, path: Path, *, retain: int = 20_000):
|
|
self.path = path
|
|
self.retain = retain
|
|
self.queue: queue.Queue = queue.Queue(maxsize=4096)
|
|
self.dropped = 0
|
|
self.failed = 0
|
|
self.closed = False
|
|
self.state_lock = threading.Lock()
|
|
self.thread = threading.Thread(target=self._write, name='operation-logs', daemon=True)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with closing(self._connect()) as conn, conn:
|
|
conn.execute('CREATE TABLE IF NOT EXISTS logs (id INTEGER PRIMARY KEY, timestamp TEXT NOT NULL, level TEXT NOT NULL, source TEXT NOT NULL, event TEXT NOT NULL, details TEXT NOT NULL)')
|
|
conn.execute('CREATE INDEX IF NOT EXISTS logs_level_id ON logs(level, id)')
|
|
conn.execute('CREATE INDEX IF NOT EXISTS logs_source_id ON logs(source, id)')
|
|
self.thread.start()
|
|
|
|
def _connect(self):
|
|
conn = sqlite3.connect(self.path, timeout=5)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def emit(self, level: str, source: str, event: str, details: dict):
|
|
row = (datetime.now(timezone.utc).isoformat(), level, source[:100], event[:160], json.dumps(metadata(details), ensure_ascii=False))
|
|
with self.state_lock:
|
|
if self.closed:
|
|
return
|
|
try:
|
|
self.queue.put_nowait(row)
|
|
except queue.Full:
|
|
self.dropped += 1
|
|
|
|
def _write(self):
|
|
while True:
|
|
first = self.queue.get()
|
|
batch = [first]
|
|
while len(batch) < 128:
|
|
try:
|
|
batch.append(self.queue.get_nowait())
|
|
except queue.Empty:
|
|
break
|
|
stop = None in batch
|
|
rows = [row for row in batch if row is not None]
|
|
try:
|
|
if rows:
|
|
with closing(self._connect()) as conn, conn:
|
|
conn.executemany('INSERT INTO logs(timestamp,level,source,event,details) VALUES(?,?,?,?,?)', rows)
|
|
conn.execute('DELETE FROM logs WHERE id <= (SELECT id FROM logs ORDER BY id DESC LIMIT 1 OFFSET ?)', (self.retain,))
|
|
except Exception:
|
|
self.failed += len(rows)
|
|
finally:
|
|
for _ in batch:
|
|
self.queue.task_done()
|
|
if stop:
|
|
return
|
|
|
|
def query(self, *, limit=50, before=None, level='', source='', q=''):
|
|
clauses, args = [], []
|
|
for column, value in [('level', level), ('source', source)]:
|
|
if value:
|
|
clauses.append(f'{column} = ?')
|
|
args.append(value)
|
|
if before is not None:
|
|
clauses.append('id < ?')
|
|
args.append(before)
|
|
if q:
|
|
clauses.append('(instr(event, ?) > 0 OR instr(details, ?) > 0)')
|
|
args += [q, q]
|
|
where = ' WHERE ' + ' AND '.join(clauses) if clauses else ''
|
|
with closing(self._connect()) as conn, conn:
|
|
rows = conn.execute('SELECT * FROM logs' + where + ' ORDER BY id DESC LIMIT ?', (*args, limit + 1)).fetchall()
|
|
sources = [row[0] for row in conn.execute('SELECT DISTINCT source FROM logs ORDER BY source')]
|
|
items = [{**dict(row), 'details': json.loads(row['details'])} for row in rows[:limit]]
|
|
return {'items': items, 'next_cursor': items[-1]['id'] if len(rows) > limit else None,
|
|
'sources': sources, 'pending': self.queue.qsize(), 'dropped': self.dropped,
|
|
'write_failures': self.failed, 'retention': self.retain}
|
|
|
|
def close(self):
|
|
with self.state_lock:
|
|
if self.closed:
|
|
return
|
|
self.closed = True
|
|
self.queue.put(None)
|
|
self.thread.join(timeout=15)
|
|
|
|
|
|
_store: LogStore | None = None
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def get_store() -> LogStore:
|
|
global _store
|
|
path = get_settings().data_dir / 'logs' / 'operations.sqlite3'
|
|
with _lock:
|
|
if _store is None or _store.path != path or _store.closed:
|
|
if _store is not None and not _store.closed:
|
|
_store.close()
|
|
_store = LogStore(path)
|
|
return _store
|
|
|
|
|
|
def log_event(module: str, event: str, *, level='INFO', error: BaseException | None = None, **details):
|
|
if request_id.get():
|
|
details.setdefault('request_id', request_id.get())
|
|
if agent_run_id.get():
|
|
details.setdefault('run_id', agent_run_id.get())
|
|
if error:
|
|
details['error_type'] = type(error).__name__
|
|
details.setdefault('error_code', getattr(error, 'code', None))
|
|
details['frames'] = '; '.join(f'{Path(f.filename).name}:{f.lineno}:{f.name}' for f in traceback.extract_tb(error.__traceback__)[-8:])
|
|
try:
|
|
get_store().emit(level, module, event, details)
|
|
except Exception:
|
|
# Logging must not turn a successful save/run into a business failure.
|
|
logging.getLogger('operation_log_storage').error('Operational log storage unavailable')
|
|
|
|
|
|
class ApplicationLogHandler(logging.Handler):
|
|
def emit(self, record):
|
|
if record.name == 'operation_log_storage' or getattr(record, '_notes_operation_logged', False):
|
|
return
|
|
record._notes_operation_logged = True
|
|
# Legacy log messages can include note text/credentials, even in f-strings.
|
|
# Preserve source location and error class; structured call sites carry IDs.
|
|
log_event(record.name, 'application.warning' if record.levelno < 40 else 'application.error',
|
|
level=record.levelname, error=record.exc_info[1] if record.exc_info else None,
|
|
frames=f'{Path(record.pathname).name}:{record.lineno}:{record.funcName}')
|
|
|
|
|
|
def install_logging():
|
|
# Uvicorn's default logger stops propagation before the root logger.
|
|
for name in ('', 'uvicorn'):
|
|
logger = logging.getLogger(name)
|
|
if not any(isinstance(h, ApplicationLogHandler) for h in logger.handlers):
|
|
logger.addHandler(ApplicationLogHandler(level=logging.WARNING))
|
|
|
|
|
|
def shutdown_logging():
|
|
if _store is not None and not _store.closed:
|
|
_store.close()
|