← Source tree

conversation.py

"""Internal typed CLI simulation. Not Alexa, authentication, or owner approval."""
import argparse
import copy
import hashlib
import json
from pathlib import Path

from reviewqueue import load_payload, review


def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True,
                      separators=(",", ":"), allow_nan=False).encode("utf-8")


class ReviewSession:
    """Run the real local model and bind demo confirmation to its full report."""
    def __init__(self, payload):
        snapshot = copy.deepcopy(payload)
        payload_digest = hashlib.sha256(canonical(snapshot)).hexdigest()
        result = review(snapshot)
        document = {
            "simulation": "ReviewQueue typed conversation 0.1",
            "limits": "Internal demo only. Not Alexa, authentication, owner approval, or a submission.",
            "canonical_input_sha256": payload_digest,
            "review": result,
        }
        # Store only immutable serialized bytes; callers cannot alter this snapshot.
        self.__encoded = canonical(document)
        self.__digest = hashlib.sha256(self.__encoded).hexdigest()
        self.__viewed = False
        self.__confirmed = False

    def view(self):
        self.__viewed = True
        self.__confirmed = False
        return {"state": "awaiting_demo_confirmation", "digest": self.__digest,
                "document": json.loads(self.__encoded)}

    def confirm(self, digest):
        if not self.__viewed or digest != self.__digest:
            raise ValueError("Review the complete current report and confirm its exact digest")
        self.__confirmed = True
        return {"state": "demo_confirmed", "digest": self.__digest}

    def export(self):
        if not self.__confirmed:
            raise ValueError("Demo confirmation required; nothing exported")
        return self.__encoded

    def reset(self):
        self.__viewed = False
        self.__confirmed = False
        return {"state": "unreviewed"}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    if args.input.resolve() == args.output.resolve():
        parser.error("input and output must be different")
    try:
        raw = args.input.read_bytes()
        if len(raw) > 8_000_000:
            raise ValueError("input exceeds the 8 MB demonstration limit")
        session = ReviewSession(load_payload(raw))
    except (OSError, ValueError, RecursionError, OverflowError) as exc:
        parser.error(str(exc))
    print("INTERNAL SYNTHETIC DEMO — not Alexa or owner approval.", flush=True)
    print("Commands: review | confirm <digest> | export | reset | quit", flush=True)
    while True:
        try:
            line = input("> ").strip()
        except EOFError:
            break
        try:
            if line == "quit":
                break
            if line == "review":
                print(json.dumps(session.view(), ensure_ascii=False, indent=2))
            elif line.startswith("confirm "):
                print(json.dumps(session.confirm(line.removeprefix("confirm ").strip())))
            elif line == "export":
                encoded = session.export() + b"\n"
                # Confirmed only; never overwrite any existing file or symlink.
                with args.output.open("xb") as target:
                    target.write(encoded)
                print("Internal demo report exported; no submission or client delivery.")
            elif line == "reset":
                print(json.dumps(session.reset()))
            else:
                raise ValueError("Unknown command; source text is never executed")
        except (OSError, ValueError) as exc:
            print(f"Refused: {exc}")


if __name__ == "__main__":
    main()