"""Content-free, endpoint-scoped plugin diagnostics shared across Codex processes.""" from __future__ import annotations import hashlib import json import os import time from pathlib import Path from reme_config import data_dir, lock LOG_BYTES = 256 * 1024 def digest(value: str) -> str: """Return a deterministic filename-safe identifier.""" return hashlib.sha256(value.encode("utf-8")).hexdigest()[:32] def queue_root(config: dict) -> Path: """Keep queued conversations tied to the service that originally captured them.""" return data_dir() / "queue" / digest(config["endpoint"]) def endpoint_dir(config: dict) -> Path: """Keep operational state separate from conversation retry data.""" return data_dir() / "state" / digest(config["endpoint"]) def log_status(event: str, error: BaseException | None = None, *, config: dict | None = None, turns: int = 0) -> None: """Bound diagnostics without logging prompts, memory, credentials, or exception messages.""" try: with lock(data_dir() / "log.lock") as acquired: if not acquired: return path = data_dir() / "hooks.log" if path.exists() and path.stat().st_size >= LOG_BYTES: path.replace(path.with_suffix(".log.1")) row = { "time": time.time(), "event": event, "error": type(error).__name__ if error else "", "endpoint": digest(config["endpoint"]) if config else "", "turns": turns, } fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as handle: handle.write(json.dumps(row) + "\n") except OSError: pass def recent_activity(config: dict, limit: int = 20) -> list[dict]: """Read only the bounded tail, ignoring torn rows and other services' activity.""" rows = [] for name in ("hooks.log.1", "hooks.log"): path = data_dir() / name try: with path.open("rb") as handle: size = path.stat().st_size handle.seek(max(0, size - LOG_BYTES)) if size > LOG_BYTES: handle.readline() lines = handle.read(LOG_BYTES).splitlines() except FileNotFoundError: continue for line in lines: try: row = json.loads(line) except (ValueError, UnicodeError): continue if isinstance(row, dict) and row.get("endpoint") == digest(config["endpoint"]): rows.append({key: row.get(key) for key in ("time", "event", "error", "turns")}) return rows[-limit:] def local_status(config: dict) -> dict: """Report delivery state without opening queued conversation files.""" current = queue_root(config) sessions = set() pending = other = 0 for path in (data_dir() / "queue").glob("*/*/*.json"): if path.with_suffix(".done").exists(): continue if path.parent.parent == current: pending += 1 sessions.add(path.parent.name) else: other += 1 state = endpoint_dir(config) try: latest = json.loads((state / "dream.json").read_text(encoding="utf-8")) except FileNotFoundError: latest = None if latest and latest.get("status") == "running": with lock(state / "dream.lock") as acquired: if acquired: latest = {**latest, "status": "interrupted"} try: scheduler = json.loads((data_dir() / "scheduler.json").read_text(encoding="utf-8")) except FileNotFoundError: scheduler = {} with lock(data_dir() / "scheduler.lock") as acquired: if acquired: scheduler = {"phase": "stopped", "next_run_at": None} elif ( scheduler.get("endpoint") != digest(config["endpoint"]) or scheduler.get("enabled") != config["autoDreamEnabled"] or scheduler.get("cron") != config["dreamCron"] or scheduler.get("timezone") != config["timezone"] ): scheduler = {"phase": "updating", "next_run_at": None} return { "auto_recall": {"enabled": config["autoRecall"]}, "auto_memory": { "enabled": config["autoMemoryEnabled"], "interval": config["autoMemoryInterval"], "queued_turns": pending, "queued_sessions": len(sessions), "other_endpoint_queued_turns": other, }, "auto_dream": { "scheduler": "codex_mcp", "enabled": config["autoDreamEnabled"], "cron": config["dreamCron"], "timezone": config["timezone"], "phase": scheduler.get("phase", "starting"), "next_run_at": scheduler.get("next_run_at"), "last_run": latest, }, "recent_activity": recent_activity(config), "hook_trust": "Check the Codex Hooks settings; MCP health does not report host hook trust.", }