run_example.py

"""Local research example; no external action is executed."""
import argparse
import json
from huggingface_hub import snapshot_download
from humanpermit.staged_layer import StagedModel
from humanpermit.policy_session import PolicySession

parser = argparse.ArgumentParser()
parser.add_argument("--placement", choices=["cpu", "cuda", "cuda-stream"], default="cpu")
parser.add_argument("--offline", action="store_true")
args = parser.parse_args()
base = snapshot_download("Qwen/Qwen2.5-0.5B-Instruct",
    revision="7ae557604adf67be50417f59c2c2f167def9a775", local_files_only=args.offline)
adapter = snapshot_download("sdellava/humanpermit-neural-gate",
    allow_patterns=["adapter_config.json", "policy_head.safetensors"], local_files_only=args.offline)
model = StagedModel(base, args.placement)
model.load_adapter(adapter)
session = PolicySession(model, require_fresh_evidence=True)
policy = "\n".join([
    "adjust_power is permitted without supervisor consent.",
    "operational_shutdown needs explicit supervisor consent before execution.",
    "calibrate_sensor is forbidden regardless of supervisor consent.",
    "Absent or stale observations require waiting; an applicable prohibition takes priority.",
    "An approval covers an action only when verified and matching both its operation and target.",
])
candidate = {"operation": "operational_shutdown", "target": "machine/unit-0", "value": {"note": "ordinary request"}}
context = {"observations_fresh": True}
initial = session.propose(policy, candidate, context)
print(json.dumps({"initial": initial}, indent=2))
if initial.get("review_id"):
    resumed = session.resume(initial["review_id"], reviewer="local-operator", approve=True,
        current_policy=policy, current_candidate=candidate, current_context=context)
    print(json.dumps({"after_simulated_approval": resumed}, indent=2))
print(json.dumps({"simulated_releases": session.released}, indent=2))