Spaces:
Sleeping
Sleeping
File size: 3,238 Bytes
990895d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """Load coach brief and persist coach debug traces as a ring buffer.
Traces are append-only, then trimmed to the newest configured limit.
"""
from __future__ import annotations
import hashlib
from pathlib import Path
from typing import Any
from app.fsutil import append_jsonl, read_jsonl, rewrite_jsonl
from app.paths import Paths
DEFAULT_BRIEF = """# Coach brief (generic)
Role: brief checklist coach. No pity. No pep talk.
Never invent statistics. Use only EVIDENCE and SERVER_PICKS.
Never cancel committed admin/travel process merely from fear.
If self-harm language appears: crisis redirect only.
## Bricks
- A: admin/critical checklist item
- B: environment/hygiene one act
- C: build/earn one ship
- D: logistics one line
- E: boundary one line then silence
- S: stop spiral — water/shower/sleep
Court closed by default. Daydream requires same-day brick.
Urge: delay then brick; ceiling one session/day (policy text only).
"""
class TraceStore:
"""Read coach brief and manage the traces JSONL ring buffer."""
def __init__(self, paths: Paths, *, trace_limit: int = 200) -> None:
self.paths = paths
self.trace_limit = trace_limit
def ensure_brief(self) -> None:
"""Seed a generic brief when missing."""
if not self.paths.coach_brief.exists():
self.paths.coach_brief.write_text(DEFAULT_BRIEF, encoding="utf-8")
def read_brief(self, *, max_bytes: int = 32_768) -> str:
"""Read brief truncated to max_bytes with a truncation marker."""
self.ensure_brief()
raw = self.paths.coach_brief.read_bytes()
if len(raw) > max_bytes:
text = raw[:max_bytes].decode("utf-8", errors="replace")
return text + "\n[TRUNCATED]\n"
return raw.decode("utf-8", errors="replace")
def brief_meta(self, *, include_full: bool) -> dict[str, Any]:
"""Return brief hash, excerpt, and optional full body."""
brief = self.read_brief()
digest = hashlib.sha256(brief.encode("utf-8")).hexdigest()
return {
"brief_sha256": digest,
"brief_excerpt": brief[:500],
"brief_full": brief if include_full else None,
}
def append_trace(self, trace: dict[str, Any]) -> None:
"""Append one trace and trim to the newest N records."""
append_jsonl(self.paths.traces, trace)
records = read_jsonl(self.paths.traces)
if len(records) > self.trace_limit:
rewrite_jsonl(self.paths.traces, records[-self.trace_limit :])
def list_traces(self, limit: int = 20) -> list[dict[str, Any]]:
"""Return newest-first trace summaries."""
records = read_jsonl(self.paths.traces)
records.reverse()
return records[:limit]
def get_trace(self, trace_id: str) -> dict[str, Any] | None:
"""Return one full trace by id."""
for record in reversed(read_jsonl(self.paths.traces)):
if record.get("trace_id") == trace_id:
return record
return None
def latest_trace(self) -> dict[str, Any] | None:
"""Return the newest trace, or None."""
records = read_jsonl(self.paths.traces)
return records[-1] if records else None
|