Spaces:
Running
Running
File size: 2,995 Bytes
d1d8733 | 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 | """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))
|