Spaces:
Sleeping
Sleeping
| """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 | |