#!/usr/bin/env python3
"""Offline replay of a public Sno receipt; --run explicitly calls the real local service."""
from __future__ import annotations
import argparse
import hashlib
import html
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tempfile
import textwrap
ROOT = Path(__file__).resolve().parent
COMMIT = "5c4ee17f3ef4f12dc6514e2bd78ec6e580b9c363"
def read_json(path):
return json.loads(path.read_text(encoding="utf-8"))
def validate(receipt):
if receipt["sourceCommit"] != COMMIT or receipt["version"] != "1.6.0":
raise ValueError("unexpected Sno source version")
if receipt["fixtureSha256"] != hashlib.sha256((ROOT / "fixture.json").read_bytes()).hexdigest():
raise ValueError("public input differs from the real service recording")
if receipt["realAgentsInvoked"] is not False or receipt["telemetry"] is not False:
raise ValueError("recording boundary changed")
if receipt["mode"] != "local-first" or receipt["rerank"] != "none" or receipt["llmCallsConfigured"] != "off":
raise ValueError("this replay requires isolated Local First with model calls disabled")
if receipt.get("status") == "blocked":
if receipt["steps"] or not receipt.get("reason") or receipt.get("memoryOperationsVerified") is not False:
raise ValueError("a blocked receipt must not claim successful memory operations")
return
if receipt.get("status") != "passed":
raise ValueError("unknown validation status")
expected = ["store", "get", "correct", "get-history", "recall-current", "store-and-forget"]
if [step["operation"] for step in receipt["steps"]] != expected or not all(step["passed"] is True for step in receipt["steps"]):
raise ValueError("missing successful real service operations")
fixture = read_json(ROOT / "fixture.json")
steps = receipt["steps"]
if steps[0]["text"] != fixture["original"] or steps[1]["text"] != fixture["original"] or steps[2]["text"] != fixture["corrected"]:
raise ValueError("recorded memory text does not match the public fixture")
if (steps[3]["text"] != fixture["original"] + "\nretired; superseded by memory-v2"
or steps[3]["supersededBy"] != "memory-v2"
or steps[3].get("normalization") != "Verified runtime successor ID in annotated text and metadata is replaced with alias memory-v2."):
raise ValueError("missing correction history")
if (steps[4]["currentText"] != fixture["corrected"] or steps[4]["oldTextExcluded"] is not True
or steps[4].get("source") != "auto" or steps[4].get("injectionPhase") != "prompt"):
raise ValueError("missing current-only recall check")
if steps[5]["readerGet"] is not None:
raise ValueError("forgotten temporary memory remains readable")
def render(receipt):
esc = lambda value: html.escape(str(value), quote=True)
if receipt.get("status") == "blocked":
fixture = read_json(ROOT / "fixture.json")
return '''共享记忆接棒 · 实验输入与真实验证状态