irregular6612 commited on
Commit
d8cad4b
·
1 Parent(s): 49272cc

feat(cp7): latest_for_model checkpoint selection

Browse files
proteus/runtime/memory.py CHANGED
@@ -84,3 +84,35 @@ def load_checkpoint(path: str | Path) -> MemoryCheckpoint:
84
  """Read one MemoryCheckpoint from *path* (raises FileNotFoundError on miss)."""
85
  text = Path(path).read_text(encoding="utf-8")
86
  return MemoryCheckpoint.model_validate_json(text)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
84
  """Read one MemoryCheckpoint from *path* (raises FileNotFoundError on miss)."""
85
  text = Path(path).read_text(encoding="utf-8")
86
  return MemoryCheckpoint.model_validate_json(text)
87
+
88
+
89
+ def latest_for_model(
90
+ model: str, root: str | Path = "runs/memory"
91
+ ) -> MemoryCheckpoint | None:
92
+ """Return the newest checkpoint for *model*, or None if none exist.
93
+
94
+ "Newest" = the lexicographically greatest filename stamp (the stamps are
95
+ zero-padded ISO-ish, so lexical order == chronological order).
96
+ """
97
+ directory = Path(root) / _safe(model)
98
+ if not directory.is_dir():
99
+ return None
100
+ files = sorted(directory.glob("*.json"))
101
+ if not files:
102
+ return None
103
+ return load_checkpoint(files[-1])
104
+
105
+
106
+ def render_memory_block(checkpoint: MemoryCheckpoint) -> str:
107
+ """Render the memory episode as a labelled observation block.
108
+
109
+ Pure: depends only on the checkpoint (no engine). Prepended to the turn-1
110
+ observation by SessionRunner so the handover history reads as the model's
111
+ own prior play.
112
+ """
113
+ parts = ["MEMORY — your earlier play on this scenario:"]
114
+ for mt in checkpoint.memory_turns:
115
+ parts.append(f"Memory {mt.turn_idx}:")
116
+ parts.append(mt.frame_ascii)
117
+ parts.append(f" you chose: {mt.action}")
118
+ return "\n".join(parts)
tests/runtime/test_memory.py CHANGED
@@ -58,3 +58,50 @@ def test_load_missing_raises(tmp_path):
58
  import pytest
59
  with pytest.raises(FileNotFoundError):
60
  load_checkpoint(tmp_path / "nope.json")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
58
  import pytest
59
  with pytest.raises(FileNotFoundError):
60
  load_checkpoint(tmp_path / "nope.json")
61
+
62
+
63
+ def test_latest_for_model_picks_newest(tmp_path):
64
+ from proteus.runtime.memory import save_checkpoint, latest_for_model
65
+
66
+ older = _checkpoint()
67
+ older.created_at = "2026-06-02T09-00-00Z"
68
+ newer = _checkpoint()
69
+ newer.created_at = "2026-06-02T11-00-00Z"
70
+ save_checkpoint(older, root=tmp_path)
71
+ save_checkpoint(newer, root=tmp_path)
72
+
73
+ got = latest_for_model("demo", root=tmp_path)
74
+ assert got is not None
75
+ assert got.created_at == "2026-06-02T11-00-00Z"
76
+
77
+
78
+ def test_latest_for_model_none_when_absent(tmp_path):
79
+ from proteus.runtime.memory import latest_for_model
80
+
81
+ assert latest_for_model("nobody", root=tmp_path) is None
82
+
83
+
84
+ def test_render_memory_block_contains_frames_and_actions():
85
+ from proteus.runtime.memory import render_memory_block
86
+
87
+ ck = _checkpoint()
88
+ block = render_memory_block(ck)
89
+ assert "MEMORY" in block
90
+ assert "FRAME-A" in block
91
+ assert "you chose: up" in block
92
+
93
+
94
+ def test_render_memory_block_empty_episode():
95
+ from proteus.runtime.memory import MemoryCheckpoint, render_memory_block
96
+
97
+ ck = MemoryCheckpoint(
98
+ model="m", scenario="s", difficulty="easy", seed=None,
99
+ created_at="t", memory_turns=[], outcome="survived",
100
+ transparent_prompt="b",
101
+ )
102
+ block = render_memory_block(ck)
103
+ assert "MEMORY" in block # header always present, no turns listed
104
+
105
+
106
+ def test_memory_symbols_exported_from_runtime():
107
+ from proteus.runtime import MemoryCheckpoint, save_checkpoint # noqa: F401