Deny a Critic success the live world contradicts, and stop reporting a ghost as live
0a8c2ae verified | """Persistent, auditable learning memory for each Voyager checkpoint.""" | |
| from __future__ import annotations | |
| import json | |
| import threading | |
| import time | |
| from pathlib import Path | |
| from typing import Any | |
| from voyager_verification import task_spec | |
| SCHEMA_VERSION = 1 | |
| WATCH_INTERVAL_SECONDS = 2.0 | |
| def _read_json(path: Path, default: Any) -> Any: | |
| try: | |
| return json.loads(path.read_text(encoding="utf-8")) | |
| except (OSError, ValueError, TypeError): | |
| return default | |
| def learning_path(checkpoint: str | Path) -> Path: | |
| return Path(checkpoint) / "learning" / "learning_state.json" | |
| def build_learning_state(checkpoint: str | Path, bot_username: str) -> dict[str, Any]: | |
| checkpoint = Path(checkpoint) | |
| completed = _read_json(checkpoint / "curriculum" / "completed_tasks.json", []) | |
| failed = _read_json(checkpoint / "curriculum" / "failed_tasks.json", []) | |
| skills = _read_json(checkpoint / "skill" / "skills.json", {}) | |
| recovery = _read_json(checkpoint / "learning" / "curriculum_recovery.json", {}) | |
| if not isinstance(completed, list): | |
| completed = [] | |
| if not isinstance(failed, list): | |
| failed = [] | |
| if not isinstance(skills, dict): | |
| skills = {} | |
| # Reporting only. The upstream CurriculumAgent owns task selection, so this | |
| # state is never used to filter or gate a proposal. | |
| kinds: dict[str, set[str]] = {} | |
| for task in completed: | |
| spec = task_spec(task) | |
| if not spec: | |
| continue | |
| kind, amount, name = spec | |
| kinds.setdefault(kind, set()).add(f"{amount}:{name}" if kind == "explore" else name) | |
| summary: dict[str, Any] = {} | |
| for kind, values in sorted(kinds.items()): | |
| summary[kind] = { | |
| "distinct_count": len(values), | |
| "examples": sorted(values)[-16:], | |
| } | |
| return { | |
| "schema_version": SCHEMA_VERSION, | |
| "bot_username": str(bot_username), | |
| "checkpoint_dir": str(checkpoint), | |
| "bucket_persistent": str(checkpoint).startswith("/persist/"), | |
| "updated_at": time.time(), | |
| "completed_count": len(completed), | |
| "failed_count": len(failed), | |
| "skill_count": len(skills), | |
| "capabilities": summary, | |
| "known_skills": sorted(str(name) for name in skills)[-128:], | |
| "recent_completed": [str(x) for x in completed[-12:]], | |
| "recent_failed": [str(x) for x in failed[-12:]], | |
| "curriculum_owner": "upstream CurriculumAgent", | |
| # Written by the removed local planner layer. No code in this tree writes | |
| # it any more, so the file survives in the bucket as a ghost and the old | |
| # key name made it read as live behaviour: the status kept reporting | |
| # build_retry_counts for a ten-project catalogue that no longer exists. | |
| # Reported under a name that says what it is, rather than deleted. | |
| "curriculum_recovery_legacy": recovery if isinstance(recovery, dict) else {}, | |
| "sources": { | |
| "completed_tasks": "curriculum/completed_tasks.json", | |
| "failed_tasks": "curriculum/failed_tasks.json", | |
| "skills": "skill/skills.json", | |
| }, | |
| } | |
| def refresh_learning_state(checkpoint: str | Path, bot_username: str) -> dict[str, Any]: | |
| state = build_learning_state(checkpoint, bot_username) | |
| path = learning_path(checkpoint) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| temporary = path.with_suffix(".json.tmp") | |
| temporary.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8") | |
| temporary.replace(path) | |
| return state | |
| def read_learning_state(checkpoint: str | Path) -> dict[str, Any]: | |
| value = _read_json(learning_path(checkpoint), {}) | |
| return value if isinstance(value, dict) else {} | |
| def _signature(checkpoint: Path) -> tuple[Any, ...]: | |
| result = [] | |
| for path in ( | |
| checkpoint / "curriculum" / "completed_tasks.json", | |
| checkpoint / "curriculum" / "failed_tasks.json", | |
| checkpoint / "skill" / "skills.json", | |
| ): | |
| try: | |
| stat = path.stat() | |
| result.append((stat.st_mtime_ns, stat.st_size)) | |
| except OSError: | |
| result.append(None) | |
| return tuple(result) | |
| def start_learning_watcher( | |
| checkpoint: str | Path, | |
| bot_username: str, | |
| *, | |
| interval: float = WATCH_INTERVAL_SECONDS, | |
| ) -> threading.Thread: | |
| checkpoint = Path(checkpoint) | |
| def run() -> None: | |
| previous = None | |
| while True: | |
| current = _signature(checkpoint) | |
| if current != previous: | |
| try: | |
| state = refresh_learning_state(checkpoint, bot_username) | |
| print( | |
| f"[voyager-learning] {bot_username}: " | |
| f"completed={state['completed_count']} skills={state['skill_count']} " | |
| f"kinds={len(state['capabilities'])}", | |
| flush=True, | |
| ) | |
| except Exception as exc: | |
| print( | |
| f"[voyager-learning] refresh failed for {bot_username}: " | |
| f"{type(exc).__name__}: {exc}", | |
| flush=True, | |
| ) | |
| previous = current | |
| time.sleep(max(0.25, float(interval))) | |
| thread = threading.Thread( | |
| target=run, | |
| daemon=True, | |
| name=f"voyager-learning-{bot_username}", | |
| ) | |
| thread.start() | |
| return thread | |