Spaces:
Running
Running
| """Explicit live test: writes synthetic notes and one skill revision to an existing private bucket.""" | |
| import argparse | |
| import importlib.util | |
| import json | |
| from pathlib import Path | |
| import uuid | |
| ROOT = Path(__file__).resolve().parents[1] | |
| spec = importlib.util.spec_from_file_location("memory", ROOT / "memory-use/scripts/memory.py") | |
| m = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(m) | |
| def run(bucket, state): | |
| state.mkdir(parents=True) | |
| first = m.Memory(state / "machine-one") | |
| second = m.Memory(state / "machine-two") | |
| work_a, work_b = state / "checkout-a", state / "different-checkout-b" | |
| work_a.mkdir() | |
| work_b.mkdir() | |
| for client, work, runtime in ((first, work_a, "codex"), (second, work_b, "claude")): | |
| instructions = client.home / ("AGENTS.md" if runtime == "codex" else "CLAUDE.md") | |
| instructions.write_text("Preserve this unrelated instruction.\n") | |
| result = client.bootstrap("memory-client-validation", work, bucket=bucket, runtime=runtime, | |
| skills_dir=client.home / "skills", instructions=instructions) | |
| assert result["status"] == "ready", result | |
| assert instructions.read_text().startswith("Preserve this unrelated instruction.") | |
| note = first.save("Synthetic portability check: independent clients retrieve the same immutable checkpoint.", | |
| "Client portability validation", kind="checkpoint", cwd=work_a) | |
| assert note["status"] == "saved", note | |
| read = second.read(note["id"], cwd=work_b, fresh=True) | |
| assert "Synthetic portability check" in read["text"] | |
| assert second.search("portability", cwd=work_b)["sync"] == "fresh" | |
| skill = state / "memory-client-smoke" | |
| skill.mkdir() | |
| (skill / "SKILL.md").write_text("---\nname: memory-client-smoke\ndescription: Synthetic skill used to verify memory client portability.\n---\nRead references/check.md when validating a memory installation. This skill is test data.\n") | |
| (skill / "references").mkdir() | |
| (skill / "references/check.md").write_text("Check that a note written by one client can be read by another.\n") | |
| published = first.publish(skill, cwd=work_a) | |
| assert published["status"] == "saved", published | |
| second.install_shared("memory-client-smoke", published["revision"], "claude", cwd=work_b) | |
| assert (second.home / "skills/memory-client-smoke/references/check.md").read_text().startswith("Check that") | |
| return {"status": "passed", "bucket": bucket, "checkpoint_id": note["id"], | |
| "skill_revision": published["revision"], "isolated_clients": 2, | |
| "physical_machines": 1, "space_dependency": False, "state": str(state)} | |
| if __name__ == "__main__": | |
| p = argparse.ArgumentParser(description=__doc__) | |
| p.add_argument("--bucket", required=True) | |
| p.add_argument("--state", type=Path, default=ROOT / ".test-work" / ("live-" + uuid.uuid4().hex[:8])) | |
| args = p.parse_args() | |
| print(json.dumps(run(args.bucket, args.state), indent=2)) | |