tegwick
a835c11ae7
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a101af-e240-7882-a1e3-e37720cbdbea
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Plan/apply/rollback narrow pilot pointers; never change permissions or rules.
|
|
|
|
Only the four named pilot repositories and their AGENTS.md/CLAUDE.md are eligible.
|
|
Plan is read-only. Apply requires the reviewed plan and unchanged target hashes.
|
|
Rollback removes only this exact block, preserving all other content.
|
|
"""
|
|
from pathlib import Path
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
|
|
PILOTS = ("net-kingdom", "ops-warden", "tenant-engine", "railiance-master")
|
|
FILES = ("AGENTS.md", "CLAUDE.md")
|
|
BLOCK = """
|
|
<!-- BEGIN EMERGENT-SECURITY-PILOT -->
|
|
## NetKingdom situational knowledge pilot
|
|
|
|
For security-related work, read the shared
|
|
[Emergent Security orientation](../emergent-security/knowledge/cards/start.md)
|
|
and the task's card in that package. From this repository root, validate and read:
|
|
|
|
```bash
|
|
PYTHONPATH=../emergent-security/src python3 -m emergent_security.knowledge --repositories .. show start --audience generic
|
|
```
|
|
|
|
Use the matching provider audience when known and retrieve the relevant card.
|
|
Record package version, digest and source status; reload after resume/compaction
|
|
or a repository switch. A withheld card requires source review; an offline
|
|
snapshot is orientation only. If the sibling checkout is absent, use the trusted
|
|
installed `emergent-knowledge` package or an explicitly reviewed exported bundle;
|
|
do not silently download or execute a replacement.
|
|
|
|
This pilot adds situational knowledge, not authority, live readiness or a change
|
|
to this repository's owner contracts. Work: `emergent-security/EMERGENT-WP-0002`.
|
|
<!-- END EMERGENT-SECURITY-PILOT -->
|
|
"""
|
|
|
|
|
|
def sha(text):
|
|
return hashlib.sha256(text.encode()).hexdigest()
|
|
|
|
|
|
def target(root, repo, name):
|
|
if repo not in PILOTS or name not in FILES:
|
|
raise ValueError("target outside pilot allowlist")
|
|
path = root / repo / name
|
|
if path.is_symlink() or not path.resolve().is_relative_to(root.resolve()):
|
|
raise ValueError("pilot target must not escape root or be a symlink")
|
|
return path
|
|
|
|
|
|
def plan(root):
|
|
rows = []
|
|
for repo in PILOTS:
|
|
for name in FILES:
|
|
path = target(root, repo, name)
|
|
old = path.read_text()
|
|
if "BEGIN EMERGENT-SECURITY-PILOT" in old:
|
|
raise ValueError("pilot already present; inspect rather than overwrite")
|
|
rows.append({"repository": repo, "file": name, "expected_sha256": sha(old)})
|
|
return {"schema": "emergent.pilot-entrypoints.v1", "root": str(root.resolve()),
|
|
"block": BLOCK, "targets": rows}
|
|
|
|
|
|
def apply(doc, rollback=False):
|
|
if doc.get("schema") != "emergent.pilot-entrypoints.v1" or doc.get("block") != BLOCK:
|
|
raise ValueError("unrecognized pilot plan")
|
|
root = Path(doc["root"])
|
|
rows = doc["targets"]
|
|
if {(r["repository"], r["file"]) for r in rows} != {(r, f) for r in PILOTS for f in FILES} or len(rows) != 8:
|
|
raise ValueError("incomplete or duplicated pilot targets")
|
|
writes = []
|
|
for row in rows:
|
|
path = target(root, row["repository"], row["file"])
|
|
old = path.read_text()
|
|
if rollback:
|
|
if old.count(BLOCK) != 1:
|
|
raise ValueError("pilot block changed; manual review required")
|
|
new = old.replace(BLOCK, "", 1)
|
|
else:
|
|
if sha(old) != row["expected_sha256"]:
|
|
raise ValueError("target changed after planning; re-review")
|
|
new = old + BLOCK
|
|
writes.append((path, new))
|
|
for path, text in writes:
|
|
path.write_text(text)
|
|
return len(writes)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
prepare = sub.add_parser("plan")
|
|
prepare.add_argument("--root", type=Path, required=True)
|
|
prepare.add_argument("--output", type=Path, required=True)
|
|
for command in ("apply", "rollback"):
|
|
action = sub.add_parser(command)
|
|
action.add_argument("plan", type=Path)
|
|
args = parser.parse_args()
|
|
if args.command == "plan":
|
|
doc = plan(args.root)
|
|
args.output.write_text(json.dumps(doc, indent=2) + "\n")
|
|
print(f"Prepared 8 file additions in {', '.join(PILOTS)}; targets unchanged.")
|
|
print(BLOCK)
|
|
else:
|
|
count = apply(json.loads(args.plan.read_text()), rollback=args.command == "rollback")
|
|
print(f"{args.command}: {count} pilot entry-point files updated.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (ValueError, OSError) as error:
|
|
print(f"Pilot change refused: {type(error).__name__}", file=sys.stderr)
|
|
sys.exit(2)
|