tegwick
d4b77a8c0e
FLEX-WP-0028 admitted a dedicated package. Runtime ConfigMap and NetworkPolicy now pin flex-auth-informed-decision-sitting. Memos are in the live store for the existing named recipient. No human bind. Assistant: grok Assistant-Session: 01a09dc1-b21e-77e1-919e-fcad2f82b267
51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
"""Load bound sitting memos into a review store. Does not present or bind."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from informed_decision.records import memo_from as _memo_from
|
|
from informed_decision.store import Store
|
|
|
|
|
|
def principal_from_store(store: Store) -> str:
|
|
with store._connection() as db:
|
|
rows = list(db.execute("SELECT body FROM presentations"))
|
|
subs = {json.loads(body)["principal_sub"] for (body,) in rows}
|
|
if len(subs) != 1:
|
|
raise ValueError("single live recipient required")
|
|
value = next(iter(subs))
|
|
if not value or value == "pending-human-session":
|
|
raise ValueError("live key-cape subject required")
|
|
return value
|
|
|
|
|
|
def load(store: Store, bound: Path) -> dict:
|
|
index = json.loads((bound / "index.json").read_text())
|
|
if index.get("agent_disposition_forbidden") is not True:
|
|
raise ValueError("bound sitting index required")
|
|
loaded = []
|
|
for row in index["memos"]:
|
|
memo = _memo_from(json.loads((bound / row["memo"]).read_text()))
|
|
packet = (bound / row["packet"]).read_bytes()
|
|
digest = store.put_document(packet)
|
|
if digest != memo.packet[0].hash:
|
|
raise ValueError(f"packet digest mismatch for {memo.id}")
|
|
store.save_memo(memo)
|
|
loaded.append(memo.id)
|
|
return {"status": "loaded", "count": len(loaded), "memo_ids": loaded}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--db", type=Path, required=True)
|
|
parser.add_argument("--bound", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
store = Store(args.db)
|
|
print(json.dumps(load(store, args.bound), indent=2))
|
|
print("No presentations or dispositions created.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|