GameCult/Mimir
Watch
9
0
Fork
You've already forked Mimir
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Restore persistence helpers

Browse source
This commit is contained in:
Meta Vens 2026-05-22 16:54:02 +01:00
commit f6d01829a2
4 changed files with 602 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

26
AGENTS.md
View file

@ -29,6 +29,18 @@ VoidBot heartbeat contract, read `docs/mimir-face.md`.
changes. Put it in a separate state commit so Face memory and status do not
stay stranded as local-only residue.
## Persistence Commands
Use a real Python if the Windows Store alias gets cute. In this Codex
workstation, the bundled runtime is:
`C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe`.
```powershell
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\localcast_state.py status
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\localcast_state.py add-evidence --type decision --status ok --note "..."
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\localcast_prepare_compaction.py
```
## Session Bootstrap And Re-entry Protocol
On a fresh session, do this before editing:
@ -39,23 +51,23 @@ On a fresh session, do this before editing:
- `notes/current-system-map.md`
- `docs/implementation-plan.md`
2. run:
- `git status --short --branch`
- `git log --oneline -5`
- `Get-Content .\state\evidence.jsonl -Tail 8`
- `& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\localcast_state.py status`
3. restate the current mechanism, intended change, and next bounded move before broad edits
After compaction, resume, or suspicious continuity loss:
1. rerun `git status --short --branch`
1. rerun `& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\localcast_state.py status`
2. reread `state/map.yaml` and `notes/fresh-workspace-handoff.md`
3. treat the persisted next action as orientation, not permission to wander
When the user says to prepare for imminent compaction:
1. immediately write the hot context into a collision-proof scratch file under `state/`
2. use `state/map.yaml`, `notes/fresh-workspace-handoff.md`, `notes/current-system-map.md`, `state/scratch.md`, `state/evidence.jsonl`, and git status as the checklist for map, handoff, scratch, evidence, and git hygiene
3. update only the state that actually changed
4. commit the completed persistence pass unless the work is deliberately mid-surgery
2. run `tools/localcast_prepare_compaction.py` before editing persistence surfaces
3. use its warnings as the checklist for map, handoff, scratch, evidence, and git hygiene
4. update only the state that actually changed
5. run `tools/localcast_prepare_compaction.py` again after edits
6. commit the completed persistence pass unless the work is deliberately mid-surgery
## Operating Discipline

1
state/evidence.jsonl
View file

@ -48,3 +48,4 @@
{"ts":"2026-05-22T17:05:00+01:00","type":"implementation","status":"ok","note":"Removed the legacy synchronization pipelines from live code. MimirAudioSynchronizationAnalyzer no longer contains chirplet-energy lag search, normalized cross-correlation, approximate runtime clock placement, known-event observation, or integer aligned-frame construction; it only emits timing reports from at least three matched decoded timeline anchors. Deleted MimirAlignedAudioFrame and removed BuildAlignedAudioFrame from MimirSynchronizationHub and BufferSmoke. MimirChirpletTimeline no longer exposes DecodeReferenceWindow, ObserveKnownEvents, BuildTimelineEnergyTrace, or one-best-symbol triplet placement. approximateTimelineSeconds was removed from sync analysis signatures because canonical chirplet anchors now own placement."}
{"ts":"2026-05-22T17:35:00+01:00","type":"implementation","status":"ok","note":"Separated chirplet symbol ownership into MimirChirpletSymbolCodebook and added per-candidate refined sample offsets to transform frames. The codebook now guarantees each of the 32 symbols has a unique chirp shape; rhythm is additional code evidence rather than an acoustic alias. ClassifySymbolsAt now does coarse dictionary search, local sample-resolution search, and parabolic sub-sample refinement per symbol candidate. Synthetic two-second emitted timeline check now detects all 15 emitted chirps, decodes 13 possible triplet anchors for events 0-12, and fits clock=48001.819 Hz with confidence=0.952 and MAE=5.171 samples, improved from 11 anchors / 47978.350 Hz / MAE=8.741."}
{"ts":"2026-05-22T17:45:00+01:00","type":"implementation","status":"ok","note":"Replaced single-phase chirplet matching with quadrature sine/cosine kernels and fixed emitted timeline segment gain so canonical chirps are not amplitude-renormalized per half-second segment. Added `Mimir.BufferSmoke --chirplet-self-test` as a first-class synthetic invariant check. The check renders two seconds of canonical timeline audio, detects all 15 emitted chirps, keeps 13 triplet anchors for events 0-12, fits clock=47999.999990 Hz with confidence=1.000, and holds anchor MAE to 0.000014 samples. The next proof is the same deterministic anchor stability through actual loopback and microphone paths."}
{"ts": "2026-05-22T15:52:26+00:00", "type": "implementation", "status": "ok", "note": "Restored the repo-local Python persistence helpers from the Epiphany-style helper family after checking VoidBot and sibling repos. Mimir now has tools/localcast_state.py for status/evidence/branch edits and tools/localcast_prepare_compaction.py for pre-compaction state audits; AGENTS.md points future agents at both helpers."}

364
tools/localcast_prepare_compaction.py Normal file
View file

@ -0,0 +1,364 @@
from __future__ import annotations
import argparse
import json
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
STATE_DIR = ROOT / "state"
NOTES_DIR = ROOT / "notes"
DOCS_DIR = ROOT / "docs"
MAP_PATH = STATE_DIR / "map.yaml"
SCRATCH_PATH = STATE_DIR / "scratch.md"
BRANCHES_PATH = STATE_DIR / "branches.json"
EVIDENCE_PATH = STATE_DIR / "evidence.jsonl"
HANDOFF_PATH = NOTES_DIR / "fresh-workspace-handoff.md"
SYSTEM_MAP_PATH = NOTES_DIR / "current-system-map.md"
PLAN_PATH = DOCS_DIR / "implementation-plan.md"
AGENTS_PATH = ROOT / "AGENTS.md"
@dataclass
class Finding:
level: str
message: str
def read_text(path: Path) -> str:
return path.read_text(encoding="utf-8-sig")
def run_git(args: list[str]) -> str:
completed = subprocess.run(
["git", *args],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
return completed.stdout.strip()
def extract_map_value(path: list[str]) -> str | None:
lines = read_text(MAP_PATH).splitlines()
current_path: list[str] = []
for index, line in enumerate(lines):
if not line.strip() or line.lstrip().startswith("#"):
continue
indent = len(line) - len(line.lstrip(" "))
level = indent // 2
stripped = line.strip()
if stripped.startswith("- ") or ":" not in stripped:
continue
key, value = stripped.split(":", 1)
current_path = current_path[:level] + [key]
if current_path != path:
continue
value = value.strip()
if value in {">", "|"}:
return collect_block(lines, index + 1, indent)
return value.strip("\"'")
return None
def collect_block(lines: list[str], start: int, parent_indent: int) -> str:
parts: list[str] = []
for line in lines[start:]:
if not line.strip():
if parts:
parts.append("")
continue
indent = len(line) - len(line.lstrip(" "))
if indent <= parent_indent:
break
parts.append(line.strip())
return " ".join(part for part in parts if part).strip()
def extract_list(section: str) -> list[str]:
lines = read_text(MAP_PATH).splitlines()
results: list[str] = []
current: list[str] = []
in_section = False
section_indent = 0
for line in lines:
if not in_section:
if line.strip() == f"{section}:":
in_section = True
section_indent = len(line) - len(line.lstrip(" "))
continue
indent = len(line) - len(line.lstrip(" "))
if line.startswith(" " * (section_indent + 2) + "- "):
if current:
results.append(" ".join(current).strip())
current = [line.strip()[2:].strip()]
continue
if current and line.strip() and indent > section_indent + 2:
current.append(line.strip())
continue
if line.strip() and indent <= section_indent:
break
if current:
results.append(" ".join(current).strip())
return results
def current_scratch_subgoal() -> str | None:
lines = read_text(SCRATCH_PATH).splitlines()
for index, line in enumerate(lines):
if line.strip() != "## Current Subgoal":
continue
for candidate in lines[index + 1 :]:
stripped = candidate.strip()
if stripped:
return stripped
return None
def load_evidence() -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
for line_number, line in enumerate(read_text(EVIDENCE_PATH).splitlines(), start=1):
if not line.strip():
continue
try:
record = json.loads(line)
except json.JSONDecodeError as exc:
raise ValueError(f"evidence line {line_number} is invalid JSON: {exc}") from exc
if not isinstance(record, dict):
raise ValueError(f"evidence line {line_number} is not a JSON object")
records.append(record)
return records
def load_branches() -> dict[str, Any]:
data = json.loads(read_text(BRANCHES_PATH))
if not isinstance(data, dict):
raise ValueError("branches.json must be a JSON object")
branches = data.get("branches")
if not isinstance(branches, list):
raise ValueError("branches.json must contain a branches list")
return data
def add_required_file_checks(findings: list[Finding]) -> None:
for path in [
MAP_PATH,
SCRATCH_PATH,
BRANCHES_PATH,
EVIDENCE_PATH,
HANDOFF_PATH,
SYSTEM_MAP_PATH,
PLAN_PATH,
AGENTS_PATH,
]:
if path.exists():
findings.append(Finding("ok", f"found {path.relative_to(ROOT)}"))
else:
findings.append(Finding("error", f"missing {path.relative_to(ROOT)}"))
def add_content_checks(findings: list[Finding]) -> list[dict[str, Any]]:
summary = extract_map_value(["current_status", "summary"])
next_action = extract_map_value(["current_status", "next_action"])
if summary:
findings.append(Finding("ok", "state/map.yaml has current_status.summary"))
else:
findings.append(Finding("error", "state/map.yaml is missing current_status.summary"))
if next_action:
findings.append(Finding("ok", "state/map.yaml has current_status.next_action"))
else:
findings.append(Finding("error", "state/map.yaml is missing current_status.next_action"))
subgoals = extract_list("active_subgoals")
if subgoals:
findings.append(Finding("ok", f"state/map.yaml has {len(subgoals)} active subgoal(s)"))
else:
findings.append(Finding("warn", "state/map.yaml has no active_subgoals entries"))
scratch_subgoal = current_scratch_subgoal()
if scratch_subgoal == "No active scratch subgoal.":
findings.append(Finding("ok", "state/scratch.md has no stale active scratch subgoal"))
elif scratch_subgoal:
findings.append(Finding("warn", f"state/scratch.md has active scratch subgoal: {scratch_subgoal}"))
else:
findings.append(Finding("warn", "state/scratch.md has no Current Subgoal value"))
evidence = load_evidence()
findings.append(Finding("ok", f"state/evidence.jsonl parses ({len(evidence)} record(s))"))
if EVIDENCE_PATH.stat().st_size > 40_000:
findings.append(Finding("warn", "state/evidence.jsonl is larger than 40 KB; consider distillation"))
branches = load_branches()
active = [
branch
for branch in branches.get("branches", [])
if isinstance(branch, dict) and branch.get("status") == "active"
]
findings.append(Finding("ok", f"state/branches.json parses ({len(active)} active branch(es))"))
return evidence
def add_handoff_checks(findings: list[Finding]) -> None:
text = read_text(HANDOFF_PATH)
if HANDOFF_PATH.stat().st_size > 10_000:
findings.append(Finding("warn", "handoff is larger than 10 KB; move detail into durable docs"))
else:
findings.append(Finding("ok", "handoff is compact enough for re-entry"))
stale_branch = re.search(r"Current branch before .*ahead \d+", text)
stale_head = re.search(r"Current HEAD before .*\b[0-9a-f]{7,40}\b", text)
if stale_branch or stale_head:
findings.append(Finding("error", "handoff embeds an exact branch or HEAD snapshot; use git commands instead"))
else:
findings.append(Finding("ok", "handoff avoids exact branch or HEAD snapshots"))
for phrase in [
"Do not trust this file",
"Immediate Re-entry Instruction",
]:
if phrase in text:
findings.append(Finding("ok", f"handoff contains: {phrase}"))
else:
findings.append(Finding("warn", f"handoff missing: {phrase}"))
def add_agents_checks(findings: list[Finding]) -> None:
text = read_text(AGENTS_PATH)
if "localcast_prepare_compaction.py" in text:
findings.append(Finding("ok", "AGENTS.md tells agents to use the compaction helper"))
else:
findings.append(Finding("warn", "AGENTS.md does not mention tools/localcast_prepare_compaction.py"))
if "localcast_state.py" in text:
findings.append(Finding("ok", "AGENTS.md tells agents to use the state helper"))
else:
findings.append(Finding("warn", "AGENTS.md does not mention tools/localcast_state.py"))
if "prepare for imminent compaction" in text.lower():
findings.append(Finding("ok", "AGENTS.md names the imminent-compaction trigger"))
else:
findings.append(Finding("warn", "AGENTS.md does not name the imminent-compaction trigger phrase"))
def add_git_checks(findings: list[Finding]) -> tuple[str, str]:
status = run_git(["status", "--short", "--branch"])
try:
log = run_git(["log", "--oneline", "-5"])
except subprocess.CalledProcessError:
log = "(no commits yet)"
dirty_lines = [line for line in status.splitlines()[1:] if line.strip()]
if dirty_lines:
findings.append(Finding("warn", "git worktree has uncommitted changes; commit or explain before compaction"))
else:
findings.append(Finding("ok", "git worktree is clean"))
return status, log
def render_report(findings: list[Finding], status: str, log: str, evidence: list[dict[str, Any]]) -> str:
summary = extract_map_value(["current_status", "summary"]) or "(missing)"
next_action = extract_map_value(["current_status", "next_action"]) or "(missing)"
subgoals = extract_list("active_subgoals")
counts = {
"ok": sum(1 for finding in findings if finding.level == "ok"),
"warn": sum(1 for finding in findings if finding.level == "warn"),
"error": sum(1 for finding in findings if finding.level == "error"),
}
lines = [
"Mimir pre-compaction persistence check",
f"Workspace: {ROOT}",
f"Findings: {counts['ok']} ok, {counts['warn']} warn, {counts['error']} error",
"",
"Git status:",
status,
"",
"Recent commits:",
log,
"",
f"Summary: {summary}",
f"Next action: {next_action}",
]
if subgoals:
lines.append("Active subgoals:")
lines.extend(f"- {item}" for item in subgoals)
if evidence:
latest = evidence[-1]
lines.extend(
[
"",
"Latest distilled evidence:",
f"- {latest.get('ts', '(missing ts)')} {latest.get('type', '(missing type)')}/{latest.get('status', '(missing status)')}: {latest.get('note', '(missing note)')}",
]
)
lines.append("")
lines.append("Findings:")
for finding in findings:
lines.append(f"[{finding.level.upper()}] {finding.message}")
lines.extend(
[
"",
"Pre-compaction checklist:",
"- Update state/map.yaml only if current understanding changed.",
"- Refresh notes/fresh-workspace-handoff.md if re-entry instructions changed.",
"- Add distilled evidence only for a belief-changing lesson, verification, rejected path, or scar.",
"- Keep exact branch or HEAD out of handoff prose; git commands own volatile truth.",
"- Commit completed persistence changes, or state why the worktree must stay dirty.",
"- Re-run this helper after edits before yielding to compaction.",
]
)
return "\n".join(lines)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Audit Mimir persistent state before imminent compaction."
)
parser.add_argument(
"--strict",
action="store_true",
help="Exit nonzero on warnings as well as errors.",
)
return parser
def main() -> int:
args = build_parser().parse_args()
findings: list[Finding] = []
evidence: list[dict[str, Any]] = []
add_required_file_checks(findings)
try:
evidence = add_content_checks(findings)
except (OSError, json.JSONDecodeError, ValueError) as exc:
findings.append(Finding("error", f"state content check failed: {exc}"))
try:
add_handoff_checks(findings)
except OSError as exc:
findings.append(Finding("error", f"handoff check failed: {exc}"))
try:
add_agents_checks(findings)
except OSError as exc:
findings.append(Finding("error", f"AGENTS check failed: {exc}"))
try:
status, log = add_git_checks(findings)
except (OSError, subprocess.CalledProcessError) as exc:
findings.append(Finding("error", f"git check failed: {exc}"))
status = "(git status unavailable)"
log = "(git log unavailable)"
print(render_report(findings, status, log, evidence))
has_error = any(finding.level == "error" for finding in findings)
has_warning = any(finding.level == "warn" for finding in findings)
if has_error or (args.strict and has_warning):
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())

218
tools/localcast_state.py Normal file
View file

@ -0,0 +1,218 @@
from __future__ import annotations
import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
STATE_DIR = ROOT / "state"
MAP_PATH = STATE_DIR / "map.yaml"
BRANCHES_PATH = STATE_DIR / "branches.json"
EVIDENCE_PATH = STATE_DIR / "evidence.jsonl"
def read_text(path: Path) -> str:
return path.read_text(encoding="utf-8-sig")
def write_text(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
def load_branches() -> dict[str, Any]:
return json.loads(read_text(BRANCHES_PATH))
def save_branches(data: dict[str, Any]) -> None:
write_text(BRANCHES_PATH, json.dumps(data, indent=2) + "\n")
def append_evidence(record: dict[str, Any]) -> None:
with EVIDENCE_PATH.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(record, ensure_ascii=True) + "\n")
def local_stamp() -> str:
return datetime.now(timezone.utc).astimezone().replace(microsecond=0).isoformat()
def cmd_status(_: argparse.Namespace) -> int:
branches = load_branches().get("branches", [])
active = [branch for branch in branches if branch.get("status") == "active"]
summary = extract_map_value(["current_status", "summary"]) or "(missing)"
next_action = extract_map_value(["current_status", "next_action"]) or "(missing)"
subgoals = extract_list("active_subgoals")
print(f"Workspace: {ROOT}")
print(f"Summary: {summary}")
print(f"Next action: {next_action}")
print(f"Active branches: {len(active)} / {len(branches)}")
if subgoals:
print("Active subgoals:")
for item in subgoals:
print(f"- {item}")
return 0
def cmd_add_evidence(args: argparse.Namespace) -> int:
record: dict[str, Any] = {
"ts": local_stamp(),
"type": args.type,
"status": args.status,
"note": args.note,
}
if args.branch:
record["branch"] = args.branch
append_evidence(record)
print("Appended evidence record.")
return 0
def cmd_add_branch(args: argparse.Namespace) -> int:
data = load_branches()
branches = data.setdefault("branches", [])
if any(branch.get("id") == args.id for branch in branches):
raise SystemExit(f"Branch '{args.id}' already exists.")
branches.append(
{
"id": args.id,
"hypothesis": args.hypothesis,
"status": "active",
"artifacts": args.artifact or [],
"notes": args.note or "",
}
)
save_branches(data)
print(f"Added branch '{args.id}'.")
return 0
def cmd_close_branch(args: argparse.Namespace) -> int:
data = load_branches()
for branch in data.get("branches", []):
if branch.get("id") != args.id:
continue
branch["status"] = args.status
if args.note:
branch["notes"] = args.note
save_branches(data)
print(f"Updated branch '{args.id}' to status '{args.status}'.")
return 0
raise SystemExit(f"Branch '{args.id}' was not found.")
def extract_map_value(path: list[str]) -> str | None:
lines = read_text(MAP_PATH).splitlines()
current_path: list[str] = []
for index, line in enumerate(lines):
if not line.strip() or line.lstrip().startswith("#"):
continue
indent = len(line) - len(line.lstrip(" "))
level = indent // 2
stripped = line.strip()
if stripped.startswith("- ") or ":" not in stripped:
continue
key, value = stripped.split(":", 1)
current_path = current_path[:level] + [key]
if current_path != path:
continue
value = value.strip()
if value in {">", "|"}:
return collect_block(lines, index + 1, indent)
return value.strip("\"'")
return None
def collect_block(lines: list[str], start: int, parent_indent: int) -> str:
parts: list[str] = []
for line in lines[start:]:
if not line.strip():
if parts:
parts.append("")
continue
indent = len(line) - len(line.lstrip(" "))
if indent <= parent_indent:
break
parts.append(line.strip())
return " ".join(part for part in parts if part).strip()
def extract_list(section: str) -> list[str]:
lines = read_text(MAP_PATH).splitlines()
results: list[str] = []
current: list[str] = []
in_section = False
section_indent = 0
for line in lines:
if not in_section:
if line.strip() == f"{section}:":
in_section = True
section_indent = len(line) - len(line.lstrip(" "))
continue
indent = len(line) - len(line.lstrip(" "))
if line.startswith(" " * (section_indent + 2) + "- "):
if current:
results.append(" ".join(current).strip())
current = [line.strip()[2:].strip()]
continue
if current and line.strip() and indent > section_indent + 2:
current.append(line.strip())
continue
if line.strip() and indent <= section_indent:
break
if current:
results.append(" ".join(current).strip())
return results
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Inspect and update Mimir repo state files."
)
subparsers = parser.add_subparsers(dest="command", required=True)
status_parser = subparsers.add_parser("status", help="Show a compact state summary.")
status_parser.set_defaults(func=cmd_status)
evidence_parser = subparsers.add_parser(
"add-evidence", help="Append one JSONL evidence record."
)
evidence_parser.add_argument("--type", required=True)
evidence_parser.add_argument("--status", required=True)
evidence_parser.add_argument("--note", required=True)
evidence_parser.add_argument("--branch")
evidence_parser.set_defaults(func=cmd_add_evidence)
add_branch_parser = subparsers.add_parser(
"add-branch", help="Create a new active branch entry."
)
add_branch_parser.add_argument("--id", required=True)
add_branch_parser.add_argument("--hypothesis", required=True)
add_branch_parser.add_argument("--artifact", action="append")
add_branch_parser.add_argument("--note")
add_branch_parser.set_defaults(func=cmd_add_branch)
close_branch_parser = subparsers.add_parser(
"close-branch", help="Close an existing branch."
)
close_branch_parser.add_argument("--id", required=True)
close_branch_parser.add_argument(
"--status", required=True, choices=["accepted", "rejected", "archived"]
)
close_branch_parser.add_argument("--note")
close_branch_parser.set_defaults(func=cmd_close_branch)
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
return args.func(args)
if __name__ == "__main__":
raise SystemExit(main())