Implements the operator chain.pi

Published 2026-04-19T04:42:57Z UTC by Jacques / SPRAXXX

#!/usr/bin/env python3 """ pios_acceptance_chain.py

Implements the operator chain:

Objective Live state Previous move Next move Proof of previous move Acceptable Accepted Proceed to next move Entropy Time out Return to last safe move Proof of that safe move Live state Reset objectives No rollback necessary Viewed museum files Proceed """

from __future__ import annotations

import argparse import json import os import socket import sys from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional

UTC = timezone.utc

def now_utc() -> str: return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")

def hostname() -> str: return socket.gethostname()

@dataclass class ChainRecord: id: str created_utc: str updated_utc: str node: str

objective: str = "" live_state: str = "" previous_move: str = "" next_move: str = ""

proof_of_previous_move: list[str] = field(default_factory=list) acceptable: Optional[bool] = None accepted: Optional[bool] = None proceed_to_next_move: Optional[bool] = None

entropy: str = "" timeout: bool = False

last_safe_move: str = "" proof_of_safe_move: list[str] = field(default_factory=list)

reset_objectives: bool = False no_rollback_necessary: bool = False viewed_museum_files: list[str] = field(default_factory=list)

status: str = "open" notes: str = ""

def refresh(self) -> None: self.updated_utc = now_utc()

def decision(self) -> str: if self.accepted is True and self.proceed_to_next_move is True: return "PROCEED" if self.timeout is True: return "TIMEOUT" if self.acceptable is False or self.accepted is False: return "HOLD" if self.last_safe_move: return "RETURN_TO_SAFE_MOVE" return "REVIEW"

class ChainStore: def __init__(self, root: Path): self.root = root self.root.mkdir(parents=True, exist_ok=True)

def path_for(self, record_id: str) -> Path: return self.root / f"{record_id}.json"

def create(self, objective: str) -> ChainRecord: ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") rid = f"chain-{ts}" rec = ChainRecord( id=rid, created_utc=now_utc(), updated_utc=now_utc(), node=hostname(), objective=objective, ) self.save(rec) return rec

def load(self, record_id: str) -> ChainRecord: p = self.path_for(record_id) if not p.exists(): raise FileNotFoundError(f"record not found: {p}") data = json.loads(p.read_text()) return ChainRecord(**data)

def save(self, rec: ChainRecord) -> None: rec.refresh() p = self.path_for(rec.id) p.write_text(json.dumps(asdict(rec), indent=2) + "\n")

def list_ids(self) -> list[str]: return sorted(p.stem for p in self.root.glob("*.json"))

def cmd_init(args: argparse.Namespace, store: ChainStore) -> int: rec = store.create(args.objective) print(rec.id) print(store.path_for(rec.id)) return 0

def cmd_show(args: argparse.Namespace, store: ChainStore) -> int: rec = store.load(args.id) print(json.dumps(asdict(rec), indent=2)) print(f"\nDECISION={rec.decision()}") return 0

def cmd_list(args: argparse.Namespace, store: ChainStore) -> int: for rid in store.list_ids(): rec = store.load(rid) print(f"{rec.id}\t{rec.status}\t{rec.decision()}\t{rec.objective}") return 0

def apply_updates(rec: ChainRecord, args: argparse.Namespace) -> None: scalar_fields = [ "objective", "live_state", "previous_move", "next_move", "entropy", "last_safe_move", "notes", "status", ] for field_name in scalar_fields: value = getattr(args, field_name, None) if value is not None: setattr(rec, field_name, value)

bool_fields = [ "acceptable", "accepted", "proceed_to_next_move", "timeout", "reset_objectives", "no_rollback_necessary", ] for field_name in bool_fields: value = getattr(args, field_name, None) if value is not None: setattr(rec, field_name, value)

for item in args.add_previous_proof or []: rec.proof_of_previous_move.append(item)

for item in args.add_safe_proof or []: rec.proof_of_safe_move.append(item)

for item in args.add_museum_file or []: rec.viewed_museum_files.append(item)

def cmd_update(args: argparse.Namespace, store: ChainStore) -> int: rec = store.load(args.id) apply_updates(rec, args) store.save(rec) print(f"saved {store.path_for(rec.id)}") print(f"DECISION={rec.decision()}") return 0

def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="PIOS acceptance chain") p.add_argument( "--root", default=os.environ.get("PIOS_CHAIN_ROOT", "/pios/registry/acceptance_chain"), help="storage root", )

sub = p.add_subparsers(dest="cmd", required=True)

sp = sub.add_parser("init", help="create a new chain record") sp.add_argument("--objective", required=True) sp.set_defaults(func=cmd_init)

sp = sub.add_parser("show", help="show a record") sp.add_argument("id") sp.set_defaults(func=cmd_show)

sp = sub.add_parser("list", help="list records") sp.set_defaults(func=cmd_list)

sp = sub.add_parser("update", help="update a record") sp.add_argument("id")

sp.add_argument("--objective") sp.add_argument("--live-state") sp.add_argument("--previous-move") sp.add_argument("--next-move") sp.add_argument("--entropy") sp.add_argument("--last-safe-move") sp.add_argument("--notes") sp.add_argument("--status")

sp.add_argument("--add-previous-proof", action="append") sp.add_argument("--add-safe-proof", action="append") sp.add_argument("--add-museum-file", action="append")

sp.add_argument("--acceptable", type=lambda x: x.lower() == "true") sp.add_argument("--accepted", type=lambda x: x.lower() == "true") sp.add_argument("--proceed-to-next-move", type=lambda x: x.lower() == "true") sp.add_argument("--timeout", type=lambda x: x.lower() == "true") sp.add_argument("--reset-objectives", type=lambda x: x.lower() == "true") sp.add_argument("--no-rollback-necessary", type=lambda x: x.lower() == "true")

# map dashed names to dataclass fields sp.set_defaults(func=cmd_update)

return p

def normalize_args(ns: argparse.Namespace) -> argparse.Namespace: mapping = { "live_state": getattr(ns, "live_state", None), "previous_move": getattr(ns, "previous_move", None), "next_move": getattr(ns, "next_move", None), "last_safe_move": getattr(ns, "last_safe_move", None), "add_previous_proof": getattr(ns, "add_previous_proof", None), "add_safe_proof": getattr(ns, "add_safe_proof", None), "add_museum_file": getattr(ns, "add_museum_file", None), "proceed_to_next_move": getattr(ns, "proceed_to_next_move", None), "reset_objectives": getattr(ns, "reset_objectives", None), "no_rollback_necessary": getattr(ns, "no_rollback_necessary", None), } for k, v in mapping.items(): setattr(ns, k, v) return ns

def main() -> int: parser = build_parser() args = normalize_args(parser.parse_args()) store = ChainStore(Path(args.root)) return args.func(args, store)

if __name__ == "__main__": sys.exit(main())