File size: 5,444 Bytes
33dab44 5ddf32a 33dab44 ff538d0 33dab44 5ddf32a 33dab44 5ddf32a 33dab44 5ddf32a 0a8c2ae 33dab44 5ddf32a 33dab44 | 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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """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
|