624 lines
34 KiB
Python
624 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Canonical campaign controls. Standard library only; local records are not authentication."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
class ControlError(ValueError):
|
|
pass
|
|
|
|
|
|
def now():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def timestamp(value):
|
|
try:
|
|
result = datetime.fromisoformat(value)
|
|
if result.tzinfo is None:
|
|
raise ValueError("timezone required")
|
|
return result.timestamp()
|
|
except (TypeError, ValueError) as exc:
|
|
raise ControlError(f"invalid timestamp: {value!r}") from exc
|
|
|
|
|
|
def read_json(path):
|
|
def pairs(items):
|
|
result = {}
|
|
for key, value in items:
|
|
if key in result:
|
|
raise ControlError(f"duplicate JSON key: {key}")
|
|
result[key] = value
|
|
return result
|
|
try:
|
|
return json.loads(Path(path).read_text(), object_pairs_hook=pairs,
|
|
parse_constant=lambda x: (_ for _ in ()).throw(ControlError(f"invalid JSON: {x}")))
|
|
except (OSError, ValueError) as exc:
|
|
raise ControlError(f"{path}: {exc}") from exc
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
def file_hash(path):
|
|
h = hashlib.sha256()
|
|
with Path(path).open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def git_value(path, *args):
|
|
result = subprocess.run(["git", "-C", str(path), *args], capture_output=True, text=True)
|
|
if result.returncode:
|
|
raise ControlError(f"source git identity unavailable: {path}: {result.stderr.strip()}")
|
|
return result.stdout.rstrip("\n")
|
|
|
|
|
|
def source_manifest(path, kind):
|
|
"""Fixed exclusions, never caller-selected: mutable state cannot hash itself."""
|
|
root = Path(path).resolve()
|
|
if Path(git_value(root, "rev-parse", "--show-toplevel")).resolve() != root:
|
|
raise ControlError("source path must be a Git worktree root")
|
|
files = {}
|
|
names = git_value(root, "ls-files", "-z", "--cached", "--others", "--exclude-standard").split("\0")
|
|
for name in sorted(set(names) - {""}):
|
|
parts = Path(name).parts
|
|
if any(p in {"__pycache__", ".pytest_cache"} for p in parts):
|
|
continue
|
|
if kind == "re" and (name.startswith("verify/results/") or
|
|
(name.startswith("campaign/") and not
|
|
(name in {"campaign/models.json", "campaign/contract.schema.json"} or
|
|
name.startswith("campaign/agents/")))):
|
|
continue
|
|
p = root / name
|
|
if not p.resolve().is_relative_to(root):
|
|
raise ControlError(f"source link escapes repository: {name}")
|
|
if p.is_symlink() or p.is_dir():
|
|
raise ControlError(f"source symlink/submodule unsupported: {name}")
|
|
files[name] = {"sha256": file_hash(p), "mode": p.stat().st_mode & 0o777} if p.is_file() else None
|
|
return {"path": str(root), "commit": git_value(root, "rev-parse", "HEAD"), "sha256": digest(files)}
|
|
|
|
|
|
def atomic_json(path, value):
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temporary = tempfile.mkstemp(prefix=".write-", dir=path.parent)
|
|
try:
|
|
with os.fdopen(fd, "w") as stream:
|
|
json.dump(value, stream, indent=2, sort_keys=True, allow_nan=False)
|
|
stream.write("\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
|
|
def validate_schema(value, schema, root=None, location="$"):
|
|
"""Implement the JSON Schema subset used by contract.schema.json, fail closed."""
|
|
root = schema if root is None else root
|
|
if "$ref" in schema:
|
|
target = root
|
|
for part in schema["$ref"].removeprefix("#/").split("/"):
|
|
target = target[part]
|
|
return validate_schema(value, target, root, location)
|
|
kinds = {"object": dict, "array": list, "string": str, "null": type(None), "boolean": bool}
|
|
if "type" in schema:
|
|
types = schema["type"] if isinstance(schema["type"], list) else [schema["type"]]
|
|
if not any(type(value) is kinds[t] for t in types):
|
|
raise ControlError(f"{location}: expected {types}")
|
|
if "enum" in schema and value not in schema["enum"]:
|
|
raise ControlError(f"{location}: invalid value {value!r}")
|
|
if isinstance(value, dict):
|
|
missing = set(schema.get("required", [])) - value.keys()
|
|
extra = value.keys() - schema.get("properties", {}).keys()
|
|
if missing or (extra and schema.get("additionalProperties") is False):
|
|
raise ControlError(f"{location}: missing {sorted(missing)}, unknown {sorted(extra)}")
|
|
for key, child in schema.get("properties", {}).items():
|
|
if key in value:
|
|
validate_schema(value[key], child, root, f"{location}.{key}")
|
|
if isinstance(value, list):
|
|
if schema.get("uniqueItems") and len({digest(v) for v in value}) != len(value):
|
|
raise ControlError(f"{location}: duplicate items")
|
|
for index, item in enumerate(value):
|
|
validate_schema(item, schema.get("items", {}), root, f"{location}[{index}]")
|
|
if isinstance(value, str):
|
|
if len(value) < schema.get("minLength", 0) or len(value) > schema.get("maxLength", float("inf")):
|
|
raise ControlError(f"{location}: invalid string length")
|
|
if "pattern" in schema and not re.search(schema["pattern"], value):
|
|
raise ControlError(f"{location}: invalid string pattern")
|
|
|
|
|
|
MODELS = {"lead": "openai/gpt-5.6-sol", "resolver": "openai/gpt-6-astra",
|
|
"architecture-review": "openai/gpt-5.6-sol", "analyst": "openai/gpt-5.6-sol",
|
|
"implementer": "openai/gpt-5.6-sol", "verifier": "openai/gpt-5.6-sol", "lab": "openai/gpt-5.6-sol"}
|
|
ROLE_STATUS = {"lead": {"proposed", "ready", "implementing", "verification", "integration", "blocked", "needs-revision"},
|
|
"resolver": {"blocked", "needs-revision"}, "architecture-review": {"proposed", "ready", "implementing", "verification", "needs-revision"},
|
|
"analyst": {"proposed", "ready", "implementing", "needs-revision"},
|
|
"implementer": {"implementing"}, "verifier": {"verification", "integration"}, "lab": {"verification", "integration"}}
|
|
|
|
|
|
class Campaign:
|
|
def __init__(self, root):
|
|
supplied = Path(root)
|
|
if not supplied.is_absolute():
|
|
raise ControlError("--state-root must be an absolute canonical RE repository path")
|
|
self.root = supplied.resolve()
|
|
self.base = self.root / "campaign"
|
|
if not (self.base / "contract.schema.json").is_file():
|
|
raise ControlError("state root does not contain campaign/contract.schema.json")
|
|
self.schema = read_json(self.base / "contract.schema.json")
|
|
self.models = read_json(self.base / "models.json")
|
|
if self.models.get("schema") != "sots-models/1" or self.models.get("roles") != MODELS:
|
|
raise ControlError("model policy mismatch; no fallback allowed")
|
|
if self.models.get("max_steps") != 40 or self.models.get("implementation_wip") != 2:
|
|
raise ControlError("bounded sessions require 40 steps and implementation WIP 2")
|
|
expected = {r: "sots-" + ("lead" if r == "architecture-review" else r) for r in MODELS}
|
|
if self.models.get("agents") != expected:
|
|
raise ControlError("invalid role agent routing")
|
|
|
|
def path(self, relative):
|
|
if not isinstance(relative, str) or not relative.startswith("campaign/") or Path(relative).is_absolute():
|
|
raise ControlError("durable paths must be campaign-relative")
|
|
path = (self.root / relative).resolve()
|
|
if not path.is_relative_to(self.base.resolve()):
|
|
raise ControlError("durable path escapes campaign/")
|
|
return path
|
|
|
|
def artifact_path(self, relative):
|
|
if not isinstance(relative, str) or not relative or Path(relative).is_absolute() or ".." in Path(relative).parts:
|
|
raise ControlError("artifact must be canonical RE-relative")
|
|
path = (self.root / relative).resolve()
|
|
if not path.is_relative_to(self.root):
|
|
raise ControlError("artifact path escapes canonical RE")
|
|
for part in Path(relative).parts + path.relative_to(self.root).parts:
|
|
if part.lower() in {".git", ".ssh", ".gnupg", "secrets", "credentials", "auth.json"} or part.lower().startswith(".env") or part.lower().endswith((".pem", ".key", ".p12", ".pfx")):
|
|
raise ControlError("secret/private artifact path prohibited")
|
|
return path
|
|
|
|
@contextmanager
|
|
def lock(self):
|
|
path = self.path("campaign/runtime/control.lock")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a+") as stream:
|
|
fcntl.flock(stream, fcntl.LOCK_EX)
|
|
yield
|
|
|
|
@staticmethod
|
|
def identifier(value):
|
|
if not isinstance(value, str) or not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,79}", value):
|
|
raise ControlError("invalid identifier")
|
|
return value
|
|
|
|
def contract_path(self, cid):
|
|
return self.path(f"campaign/contracts/{self.identifier(cid)}.json")
|
|
|
|
def validate(self, contract):
|
|
validate_schema(contract, self.schema)
|
|
baseline = contract["baseline"]
|
|
if Path(baseline["re"]["path"]).resolve() != self.root:
|
|
raise ControlError("baseline RE path must equal canonical state root")
|
|
if Path(baseline["engine"]["path"]).resolve() == self.root:
|
|
raise ControlError("engine and RE source repositories must be distinct")
|
|
if contract["id"] in contract["dependencies"]:
|
|
raise ControlError("self dependency")
|
|
for field in ("acceptance", "evidence"):
|
|
ids = [item["id"] for item in contract.get(field, [])]
|
|
if len(ids) != len(set(ids)):
|
|
raise ControlError(f"duplicate {field} id")
|
|
if contract["checkpoint"] is not None:
|
|
self.path(contract["checkpoint"])
|
|
for item in contract.get("evidence", []):
|
|
self.artifact_path(item["path"])
|
|
return contract
|
|
|
|
def load(self, cid):
|
|
c = self.validate(read_json(self.contract_path(cid)))
|
|
if c["id"] != cid:
|
|
raise ControlError("contract filename/id mismatch")
|
|
return c
|
|
|
|
def contracts(self):
|
|
return [self.load(p.stem) for p in sorted((self.base / "contracts").glob("*.json"))]
|
|
|
|
def save(self, contract):
|
|
self.validate(contract)
|
|
atomic_json(self.contract_path(contract["id"]), contract)
|
|
|
|
def records(self, kind, cid):
|
|
result = []
|
|
for path in sorted(self.path(f"campaign/runtime/{kind}").glob("*.json")):
|
|
record = read_json(path)
|
|
if not isinstance(record, dict) or "contract" not in record:
|
|
raise ControlError(f"malformed {kind} record: {path}")
|
|
if record["contract"] == cid:
|
|
result.append(record)
|
|
return result
|
|
|
|
def open_surprises(self, cid):
|
|
records = self.records("surprises", cid)
|
|
if any(r.get("status") not in {"open", "resolved"} for r in records):
|
|
raise ControlError("invalid surprise status")
|
|
return [r for r in records if r["status"] == "open"]
|
|
|
|
def identity(self, actor, role, model):
|
|
if not isinstance(actor, str) or not actor.strip() or len(actor) > 120:
|
|
raise ControlError("nonempty bounded actor required")
|
|
if role not in self.models["roles"] or self.models["roles"][role] != model:
|
|
raise ControlError("role/model identity mismatch; claimed identity is not authentication")
|
|
|
|
def authorize(self, c, actor, role, model):
|
|
self.identity(actor, role, model)
|
|
if role not in {"lead", "resolver"} and (actor != c["owner"]["name"] or role != c["owner"]["role"]):
|
|
raise ControlError("contract owner or explicit lead/resolver required")
|
|
|
|
@staticmethod
|
|
def basis(c):
|
|
return digest({k: v for k, v in c.items() if k not in {"status", "checkpoint", "evidence"}})
|
|
|
|
def artifact(self, relative):
|
|
path = self.artifact_path(relative)
|
|
if not path.is_file():
|
|
raise ControlError(f"artifact is missing: {relative}")
|
|
return {"path": relative, "sha256": file_hash(path)}
|
|
|
|
def check_artifact(self, item):
|
|
if not isinstance(item, dict) or set(item) != {"path", "sha256"}:
|
|
raise ControlError("invalid artifact identity")
|
|
if self.artifact(item["path"]) != item:
|
|
raise ControlError(f"artifact hash mismatch: {item['path']}")
|
|
|
|
def checkpoint(self, cid, actor, role, model, session, summary, artifacts, next_action):
|
|
self.identity(actor, role, model)
|
|
if not session or len(session) > 240 or not summary.strip() or len(summary) > 6000 or not next_action.strip() or len(next_action) > 2000 or len(artifacts) > 32:
|
|
raise ControlError("checkpoint requires bounded summary, session, artifacts and exact next action")
|
|
with self.lock():
|
|
c = self.load(cid)
|
|
if role not in {"lead", "resolver", "verifier", "lab"}:
|
|
self.authorize(c, actor, role, model)
|
|
if any(self.artifact_path(p) == self.contract_path(cid) for p in artifacts):
|
|
raise ControlError("checkpoint cannot hash its own mutable contract; use the basis digest")
|
|
record = {"schema": "sots-checkpoint/1", "id": secrets.token_hex(12), "contract": cid,
|
|
"actor": actor, "role": role, "model": model, "session": session,
|
|
"timestamp": now(), "basis": self.basis(c), "summary": summary,
|
|
"artifacts": [self.artifact(p) for p in artifacts], "next_action": next_action}
|
|
relative = f"campaign/runtime/checkpoints/{cid}-{record['id']}.json"
|
|
atomic_json(self.path(relative), record)
|
|
c["checkpoint"] = relative
|
|
self.save(c)
|
|
return record
|
|
|
|
def check_checkpoint(self, c, since=None, session=None, model=None, require_artifacts=True, fresh=True):
|
|
if not c["checkpoint"]:
|
|
raise ControlError("missing durable checkpoint")
|
|
cp = read_json(self.path(c["checkpoint"]))
|
|
required = {"schema", "id", "contract", "actor", "role", "model", "session", "timestamp", "basis", "summary", "artifacts", "next_action"}
|
|
if not isinstance(cp, dict) or set(cp) != required or cp.get("schema") != "sots-checkpoint/1" or cp["contract"] != c["id"] or cp["basis"] != self.basis(c):
|
|
raise ControlError("checkpoint identity/basis mismatch")
|
|
self.identity(cp["actor"], cp["role"], cp["model"])
|
|
age = timestamp(now()) - timestamp(cp["timestamp"])
|
|
if age < 0 or (fresh and age > 900) or (since and timestamp(cp["timestamp"]) <= timestamp(since)):
|
|
raise ControlError("stale checkpoint: require fresh checkpoint within 15 minutes and after run/start")
|
|
if (session and cp["session"] != session) or (model and cp["model"] != model):
|
|
raise ControlError("checkpoint session/model mismatch")
|
|
if not cp["summary"] or not cp["next_action"] or (require_artifacts and not cp["artifacts"]):
|
|
raise ControlError("checkpoint requires summary, artifacts, next action")
|
|
for artifact in cp["artifacts"]:
|
|
self.check_artifact(artifact)
|
|
return cp
|
|
|
|
def source_binding(self, c, paths=None):
|
|
paths = paths or {k: v["path"] for k, v in c["baseline"].items()}
|
|
result = {}
|
|
for kind in ("engine", "re"):
|
|
path = Path(paths[kind])
|
|
if not path.is_absolute():
|
|
raise ControlError("source path must be absolute")
|
|
canonical = c["baseline"][kind]["path"]
|
|
if git_value(path, "rev-parse", "--path-format=absolute", "--git-common-dir") != git_value(canonical, "rev-parse", "--path-format=absolute", "--git-common-dir"):
|
|
raise ControlError("source is not a paired repository")
|
|
result[kind] = source_manifest(path, kind)
|
|
return result
|
|
|
|
def evidence(self, cid, record, actor, role, model):
|
|
with self.lock():
|
|
c = self.load(cid)
|
|
self.authorize(c, actor, role, model)
|
|
if self.open_surprises(cid) or c["status"] not in {"implementing", "verification", "integration"}:
|
|
raise ControlError("evidence requires active, unblocked contract")
|
|
validate_schema(record, self.schema["$defs"]["evidence"], self.schema)
|
|
if record["integrated"] and (role != "lead" or c["status"] != "integration"):
|
|
raise ControlError("integrated evidence requires lead in integration")
|
|
self.check_evidence(c, [record])
|
|
c["evidence"] = [e for e in c.get("evidence", []) if e["id"] != record["id"]] + [record]
|
|
self.save(c)
|
|
return record
|
|
|
|
def check_evidence(self, c, records=None, integrated=False):
|
|
full_package = records is None
|
|
records = c.get("evidence", []) if records is None else records
|
|
for e in records:
|
|
if e["source"] != c["baseline"]:
|
|
raise ControlError("evidence source does not match contract baseline")
|
|
self.check_artifact({"path": e["path"], "sha256": e["sha256"]})
|
|
binding = e["source_binding"]
|
|
if self.source_binding(c, {k: v["path"] for k, v in binding.items()}) != binding:
|
|
raise ControlError("evidence source content changed")
|
|
if e["integrated"] and any(binding[k]["path"] != c["baseline"][k]["path"] for k in ("engine", "re")):
|
|
raise ControlError("integrated source must be canonical engine and RE")
|
|
for field in ("binaries", "inputs"):
|
|
if not e[field]:
|
|
raise ControlError(f"evidence requires immutable {field}")
|
|
for artifact in e[field]:
|
|
self.check_artifact(artifact)
|
|
expected = {a["id"] for a in c["acceptance"] if a["axis"] == e["axis"]}
|
|
outcomes = e["outcomes"]
|
|
if not expected or {o["criterion"] for o in outcomes} != expected or len(outcomes) != len(expected):
|
|
raise ControlError("per-criterion outcomes must exactly cover evidence axis")
|
|
for outcome in outcomes:
|
|
self.check_artifact(outcome["artifact"])
|
|
if outcome["status"] != "pass":
|
|
raise ControlError("criterion did not pass")
|
|
bindings = {digest(e["source_binding"]) for e in records if e["integrated"]}
|
|
if len(bindings) > 1:
|
|
raise ControlError("final integrated evidence must share one source binding")
|
|
axes = {e["axis"] for e in records if not integrated or e["integrated"]}
|
|
if full_package and not {a["axis"] for a in c["acceptance"]} <= axes:
|
|
raise ControlError("missing required evidence axes")
|
|
return records
|
|
|
|
def verdict(self, cid, actor, role, model, session, verdict, explanation):
|
|
self.identity(actor, role, model)
|
|
if role != "verifier" or not explanation.strip() or len(explanation) > 6000 or not session:
|
|
raise ControlError("independent verifier identity/session and explanation required")
|
|
with self.lock():
|
|
c = self.load(cid)
|
|
if c["status"] not in {"verification", "integration"} or self.open_surprises(cid):
|
|
raise ControlError("verdict requires unblocked verification/integration")
|
|
if actor == c["owner"]["name"]:
|
|
raise ControlError("verifier must be independent of owner")
|
|
workers = [r for r in self.records("checkpoints", cid) if r.get("role") in {"implementer", "analyst", "architecture-review"}]
|
|
if any(r.get("actor") == actor or r.get("session") == session for r in workers):
|
|
raise ControlError("verifier overlaps worker actor/session")
|
|
if verdict not in {"pass", "fail"}:
|
|
raise ControlError("verdict must be pass or fail")
|
|
self.check_evidence(c)
|
|
record = {"schema": "sots-verdict/1", "contract": cid, "actor": actor, "role": role,
|
|
"model": model, "session": session, "timestamp": now(), "verdict": verdict,
|
|
"explanation": explanation, "basis": self.basis(c), "evidence_digest": digest(c.get("evidence", []))}
|
|
record["source_bindings_digest"] = digest([e["source_binding"] for e in c.get("evidence", [])])
|
|
atomic_json(self.path(f"campaign/runtime/verdicts/{cid}.json"), record)
|
|
return record
|
|
|
|
def check_verdict(self, c):
|
|
self.check_evidence(c)
|
|
v = read_json(self.path(f"campaign/runtime/verdicts/{c['id']}.json"))
|
|
self.identity(v.get("actor"), v.get("role"), v.get("model"))
|
|
if v.get("source_bindings_digest") != digest([e["source_binding"] for e in c.get("evidence", [])]):
|
|
raise ControlError("verdict source bindings mismatch")
|
|
if v.get("role") != "verifier" or v.get("actor") == c["owner"]["name"] or not v.get("session") or not v.get("explanation") or v.get("contract") != c["id"] or v.get("verdict") != "pass" or v.get("basis") != self.basis(c) or v.get("evidence_digest") != digest(c.get("evidence", [])):
|
|
raise ControlError("missing independent passing verifier verdict bound to current source/evidence")
|
|
workers = [r for r in self.records("checkpoints", c["id"]) if r.get("role") in {"implementer", "analyst", "architecture-review"}]
|
|
if any(r.get("actor") == v["actor"] or r.get("session") == v["session"] for r in workers):
|
|
raise ControlError("verifier overlaps worker actor/session")
|
|
|
|
@staticmethod
|
|
def check_baseline(c):
|
|
commons = []
|
|
for source in c["baseline"].values():
|
|
for arguments in (("cat-file", "-t", source["commit"]), ("rev-parse", "--path-format=absolute", "--git-common-dir")):
|
|
result = subprocess.run(["git", "-C", source["path"], *arguments], text=True, capture_output=True)
|
|
if result.returncode:
|
|
raise ControlError("baseline repository/commit unavailable")
|
|
if arguments[0] == "cat-file" and result.stdout.strip() != "commit":
|
|
raise ControlError("baseline identity is not a commit")
|
|
if arguments[0] == "rev-parse":
|
|
commons.append(Path(result.stdout.strip()).resolve())
|
|
if commons[0] == commons[1]:
|
|
raise ControlError("baseline engine/RE must be distinct repositories")
|
|
|
|
def transition(self, cid, target, actor, role, model):
|
|
edges = {"proposed": {"ready", "blocked"}, "ready": {"implementing", "blocked", "needs-revision"},
|
|
"implementing": {"verification", "blocked", "needs-revision"}, "verification": {"integration", "blocked", "needs-revision"},
|
|
"integration": {"accepted", "blocked", "needs-revision"}, "blocked": {"needs-revision"},
|
|
"needs-revision": {"ready", "blocked"}, "accepted": {"needs-revision", "blocked"}}
|
|
with self.lock():
|
|
c = self.load(cid)
|
|
self.authorize(c, actor, role, model)
|
|
if target not in edges[c["status"]]:
|
|
raise ControlError(f"illegal lifecycle transition {c['status']} -> {target}")
|
|
if target != "blocked" and self.open_surprises(cid):
|
|
raise ControlError("unresolved surprise blocks transition")
|
|
if target in {"integration", "accepted"} and role != "lead":
|
|
raise ControlError("integration/acceptance requires explicit lead")
|
|
if target in {"ready", "implementing"}:
|
|
self.check_baseline(c)
|
|
if not all(c[k] for k in ("scope", "inputs", "acceptance", "stop_conditions")):
|
|
raise ControlError("ready requires scope, inputs, acceptance and stop conditions")
|
|
for dependency in c["dependencies"]:
|
|
if self.load(dependency)["status"] != "accepted" or self.open_surprises(dependency):
|
|
raise ControlError(f"dependency not accepted: {dependency}")
|
|
if target == "implementing":
|
|
if sum(x["status"] == "implementing" for x in self.contracts()) >= self.models["implementation_wip"]:
|
|
raise ControlError("implementation WIP bound reached (2)")
|
|
if target == "verification":
|
|
starts = self.records("transitions", cid)
|
|
since = max((x["timestamp"] for x in starts if x.get("to") == "implementing"), default=None)
|
|
self.check_checkpoint(c, since=since)
|
|
if target in {"integration", "accepted"}:
|
|
self.check_checkpoint(c)
|
|
if not c["acceptance"] or not c.get("evidence"):
|
|
raise ControlError("acceptance criteria and evidence required")
|
|
self.check_evidence(c, integrated=target == "accepted")
|
|
self.check_verdict(c)
|
|
event = {"contract": cid, "from": c["status"], "to": target, "timestamp": now(), "actor": actor, "role": role, "model": model}
|
|
# Write blocking intent first; unresolved surprises also guard against interrupted writes.
|
|
atomic_json(self.path(f"campaign/runtime/transitions/{cid}-{secrets.token_hex(12)}.json"), event)
|
|
c["status"] = target
|
|
if target == "needs-revision":
|
|
c["evidence"] = []
|
|
c["checkpoint"] = None
|
|
self.save(c)
|
|
return c
|
|
|
|
def surprise(self, cid, actor, role, model, summary, probe):
|
|
self.identity(actor, role, model)
|
|
if not summary.strip() or not probe.strip() or len(summary) > 6000 or len(probe) > 2000:
|
|
raise ControlError("surprise requires bounded explanation and discriminating probe")
|
|
with self.lock():
|
|
c = self.load(cid)
|
|
sid = "s-" + secrets.token_hex(12)
|
|
record = {"schema": "sots-surprise/1", "id": sid, "contract": cid, "status": "open", "summary": summary,
|
|
"probe": probe, "actor": actor, "role": role, "model": model, "timestamp": now()}
|
|
atomic_json(self.path(f"campaign/runtime/surprises/{sid}.json"), record)
|
|
c["status"] = "blocked"
|
|
self.save(c)
|
|
return record
|
|
|
|
def resolve(self, sid, actor, role, model, explanation, probe):
|
|
self.identity(actor, role, model)
|
|
if role not in {"lead", "resolver"} or not explanation.strip() or not probe.strip() or len(explanation) > 6000 or len(probe) > 2000:
|
|
raise ControlError("Astra lead/resolver explanation and discriminating probe required")
|
|
with self.lock():
|
|
path = self.path(f"campaign/runtime/surprises/{self.identifier(sid)}.json")
|
|
s = read_json(path)
|
|
if s.get("status") != "open":
|
|
raise ControlError("surprise is not open")
|
|
c = self.load(s["contract"])
|
|
decision = {"schema": "sots-decision/1", "id": "d-" + secrets.token_hex(12), "contract": c["id"], "surprise": sid,
|
|
"actor": actor, "role": role, "model": model, "timestamp": now(), "explanation": explanation, "probe": probe,
|
|
"invalidated_evidence": c.get("evidence", []), "invalidated_checkpoint": c["checkpoint"]}
|
|
atomic_json(self.path(f"campaign/runtime/decisions/{decision['id']}.json"), decision)
|
|
c["evidence"] = []
|
|
c["checkpoint"] = None
|
|
c["status"] = "blocked"
|
|
self.save(c)
|
|
vpath = self.path(f"campaign/runtime/verdicts/{c['id']}.json")
|
|
if vpath.exists():
|
|
verdict = read_json(vpath)
|
|
verdict["verdict"] = "invalidated"
|
|
verdict["decision"] = decision["id"]
|
|
atomic_json(vpath, verdict)
|
|
s.update(status="resolved", decision=decision["id"])
|
|
atomic_json(path, s)
|
|
if not self.open_surprises(c["id"]):
|
|
c["status"] = "needs-revision"
|
|
self.save(c)
|
|
return decision
|
|
|
|
def lease(self, action, resource, actor=None, token=None, lead_release=False, role=None, model=None, reason=None):
|
|
path = self.path(f"campaign/runtime/leases/{self.identifier(resource)}.json")
|
|
with self.lock():
|
|
current = read_json(path) if path.exists() else None
|
|
if current is not None and (not isinstance(current, dict) or current.get("schema") != "sots-lease/1" or current.get("resource") != resource or current.get("status") not in {"held", "released"} or not current.get("owner") or not current.get("token")):
|
|
raise ControlError("malformed existing lease; explicit repair required")
|
|
if action == "show":
|
|
return current
|
|
self.identity(actor, role, model)
|
|
if action == "acquire":
|
|
if current and current.get("status") == "held":
|
|
raise ControlError("lease already held; no automatic stealing, lead must explicitly release stale lease")
|
|
record = {"schema": "sots-lease/1", "resource": resource, "owner": actor, "role": role, "model": model,
|
|
"token": secrets.token_hex(32), "timestamp": now(), "status": "held"}
|
|
atomic_json(path, record)
|
|
return record
|
|
if action != "release" or not current or current.get("status") != "held":
|
|
raise ControlError("no held lease to release")
|
|
if lead_release:
|
|
if role != "lead" or not reason or not reason.strip():
|
|
raise ControlError("explicit stale release requires Astra lead and reason")
|
|
elif actor != current["owner"] or not token or not secrets.compare_digest(token, current["token"]):
|
|
raise ControlError("lease release requires matching owner and token")
|
|
current.update(status="released", released_at=now(), released_by=actor, release_reason=reason,
|
|
explicit_lead_release=lead_release)
|
|
atomic_json(path, current)
|
|
return current
|
|
|
|
|
|
def parser():
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument("--state-root", required=True, help="absolute canonical RE repository")
|
|
subs = p.add_subparsers(dest="command", required=True)
|
|
subs.add_parser("list")
|
|
s = subs.add_parser("status"); s.add_argument("contract", nargs="?")
|
|
s = subs.add_parser("validate"); s.add_argument("contracts", nargs="*")
|
|
s = subs.add_parser("source-binding"); s.add_argument("contract")
|
|
s.add_argument("--engine-worktree"); s.add_argument("--re-worktree")
|
|
def identified(name):
|
|
s = subs.add_parser(name)
|
|
s.add_argument("--actor", required=True); s.add_argument("--role", required=True, choices=MODELS)
|
|
s.add_argument("--model", required=True)
|
|
return s
|
|
s = identified("checkpoint"); s.add_argument("contract"); s.add_argument("--session", required=True)
|
|
s.add_argument("--summary", required=True); s.add_argument("--artifact", action="append", default=[])
|
|
s.add_argument("--next-action", required=True)
|
|
s = identified("transition"); s.add_argument("contract"); s.add_argument("target")
|
|
s = identified("surprise"); s.add_argument("contract"); s.add_argument("--summary", required=True); s.add_argument("--probe", required=True)
|
|
s = identified("resolve"); s.add_argument("surprise"); s.add_argument("--explanation", required=True); s.add_argument("--probe", required=True)
|
|
s = identified("evidence"); s.add_argument("contract"); s.add_argument("--record", required=True, help="campaign-relative evidence JSON")
|
|
s = identified("verdict"); s.add_argument("contract"); s.add_argument("--session", required=True)
|
|
s.add_argument("--verdict", required=True, choices=["pass", "fail"]); s.add_argument("--explanation", required=True)
|
|
s = subs.add_parser("lease"); s.add_argument("action", choices=["acquire", "release", "show"]); s.add_argument("resource")
|
|
s.add_argument("--actor"); s.add_argument("--role", choices=MODELS); s.add_argument("--model")
|
|
s.add_argument("--token"); s.add_argument("--lead-release", action="store_true"); s.add_argument("--reason")
|
|
return p
|
|
|
|
|
|
def main(argv=None):
|
|
a = parser().parse_args(argv)
|
|
try:
|
|
c = Campaign(a.state_root)
|
|
identity = {k: getattr(a, k) for k in ("actor", "role", "model") if hasattr(a, k)}
|
|
if a.command in {"list", "status"}:
|
|
contracts = [c.load(a.contract)] if getattr(a, "contract", None) else c.contracts()
|
|
result = [{**v, "open_surprises": c.open_surprises(v["id"])} for v in contracts]
|
|
elif a.command == "validate":
|
|
result = [c.load(cid)["id"] for cid in a.contracts] if a.contracts else [v["id"] for v in c.contracts()]
|
|
elif a.command == "checkpoint":
|
|
result = c.checkpoint(a.contract, **identity, session=a.session, summary=a.summary, artifacts=a.artifact, next_action=a.next_action)
|
|
elif a.command == "source-binding":
|
|
if bool(a.engine_worktree) != bool(a.re_worktree):
|
|
raise ControlError("supply both source worktrees or neither")
|
|
result = c.source_binding(c.load(a.contract), {"engine": a.engine_worktree, "re": a.re_worktree} if a.engine_worktree else None)
|
|
elif a.command == "transition":
|
|
result = c.transition(a.contract, a.target, **identity)
|
|
elif a.command == "surprise":
|
|
result = c.surprise(a.contract, **identity, summary=a.summary, probe=a.probe)
|
|
elif a.command == "resolve":
|
|
result = c.resolve(a.surprise, **identity, explanation=a.explanation, probe=a.probe)
|
|
elif a.command == "evidence":
|
|
result = c.evidence(a.contract, read_json(c.path(a.record)), **identity)
|
|
elif a.command == "verdict":
|
|
result = c.verdict(a.contract, **identity, session=a.session, verdict=a.verdict, explanation=a.explanation)
|
|
else:
|
|
result = c.lease(a.action, a.resource, **identity, token=a.token, lead_release=a.lead_release, reason=a.reason)
|
|
print(json.dumps(result, indent=2, sort_keys=True))
|
|
return 0
|
|
except (ControlError, OSError, KeyError, TypeError) as exc:
|
|
print(f"campaign: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|