Compare commits

...
Author SHA1 Message Date
owner b70c89217b feat(governed): add /reopenOwnerApproval to silent-release allow-list (V2R2C2 v1.2)
autonomy-smoke-tests pr81 PASS evidence_sha256=e9b29f42e72804aedf5a269d4507aafa7270b39400c6923bbb014d8a335fe788
autonomy-unit-tests pr81 PASS evidence_sha256=068610c15e2716ad9dbda4df158f891693e53cb7da2c57aef3783e722fb6f354
campaign-approval-gate pr81 PASS evidence_sha256=0616d6aa11496391686dcbd658123e51d8d057e1f02661af7ef8512554e404d8
git-diff-check pr81 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
py-compile pr81 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
reconciliation-contract pr81 PASS evidence_sha256=f96a26b816cb58aec4639ca90e87da07ed8cf7d041d82fb67d755bbd5085dfca
release-state-machine pr81 PASS evidence_sha256=6cea88752b3e58ae5cee56ff3b1b13eab34117918d6bad9661194c031e2dc157
secret-scan pr81 PASS evidence_sha256=e85624b4ea35c4f7019bc9d9a3d7d44124afca891c3487fe9f550faf6c743245
signer-validator-contract pr81 PASS evidence_sha256=775b7aa27d0c17f48b3feff20c836f14b6ed8ebcb90ec7e0b76d2a13c579268a
aios.owner-final-merge-authorization-v2 real v2 contract tests passed
Narrow v1.2 delta: add the Owner approval-reopen POST path to
SILENT_RELEASE_POST_PATHS so reopen can proceed in Shadow without enabling
AIOS_GITOPS_WRITES_ENABLED. Route handler, release:approve guard, global
write check and all other allow-list paths are unchanged.

Adds tests/test_reopen_shadow_gate_contract.py covering:
- reopen is a narrow silent-release POST (route requires release:approve)
- global write gate is not enabled by this source patch
- all existing silent-release POST paths remain present
2026-09-12 14:29:33 +08:00
AIOS Governed Continuation dd2bd8fa08 feat(governed): add Owner approval reopen control-plane capability (V2R2C2)
autonomy-smoke-tests pr80 PASS evidence_sha256=c16c68f86ee0c4e8cfd0af3f76078be45441ee151395fbde5800145c08e189b5
autonomy-unit-tests pr80 PASS evidence_sha256=02e00edbe6415cf1d548ef7fb53b67e37c4abd29397b00075c6feb0d48b5f2c5
campaign-approval-gate pr80 PASS evidence_sha256=0616d6aa11496391686dcbd658123e51d8d057e1f02661af7ef8512554e404d8
git-diff-check pr80 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
py-compile pr80 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
reconciliation-contract pr80 PASS evidence_sha256=a98ae8e4b231cd5a106781a85b051ad0543e0481d604919bab8fc1047de342eb
release-state-machine pr80 PASS evidence_sha256=b6c51b401d65a9c69435f8b86cecab752d65ded7188228aab3de3eb010990b60
secret-scan pr80 PASS evidence_sha256=fb45191fc28616fa47b5c8ac1f5d6541f3e8d3163eb6653c9678ab84dff44e61
signer-validator-contract pr80 PASS evidence_sha256=775b7aa27d0c17f48b3feff20c836f14b6ed8ebcb90ec7e0b76d2a13c579268a
aios.owner-final-merge-authorization-v2 real v2 contract tests passed
Add a fail-closed control-plane operation POST /reopenOwnerApproval (scope
release:approve) that reopens one expired Owner approval cycle only after
signed_owner_approval_accepted is the final Saga event and every release byte
(candidate/source/tree/artifact/manifest/release-plan/policy) matches exactly.

- New module release_approval_reopen.py: archives the old approval byte-for-byte
  under approval-history/<candidate>/<fingerprint>.json, then drives
  ReleaseSagaStore.compare_and_set(APPROVED -> READY_FOR_APPROVAL) and removes
  only the old active approval slot (post-CAS). Idempotent per idempotency key.
- ReleaseCandidate v1 durable records do not persist side_effects_executed; this
  accepts an absent marker only when the accepted-approval event is the final
  Saga event, and rejects any explicit non-false value.
- Does not modify ReleaseSagaStore, never calls force_state_for_recovery_test,
  never performs direct DB or legacy-global writes. Publishes control-plane
  capability only; keeps 4642b74 on the mainline ancestry.
2026-09-12 12:13:33 +08:00
5 changed files with 605 additions and 0 deletions
@@ -155,6 +155,7 @@ SILENT_RELEASE_POST_PATHS = frozenset({
'/submitSignedApproval',
'/submitOwnerApproval',
'/startDeployment',
'/reopenOwnerApproval',
})
def reject_non_read_http(handler, method: str, path: str) -> bool:
@@ -974,6 +975,17 @@ class H(BaseHTTPRequestHandler):
return self.sendj(
silent_release_service().submit_owner_approval(data)
)
if path == "/reopenOwnerApproval":
a = self.need("release:approve")
if not a:
return
from release_approval_reopen import ApprovalReopenError, reopen_owner_approval
try:
result = reopen_owner_approval(silent_release_service(), data)
except ApprovalReopenError as exc:
return self.sendj({"error": exc.code, "message": str(exc)}, 409)
return self.sendj(result)
if path == '/startDeployment':
a=self.need('release:deploy')
if not a: return
@@ -0,0 +1,358 @@
#!/usr/bin/env python3
"""Governed, append-only reopen of one expired Owner approval cycle."""
from __future__ import annotations
import hashlib
import json
import os
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping
REQUEST_SCHEMA = "aios.owner-approval-reopen-request.v1"
EVENT_NAME = "owner_approval_reopen_authorized"
_IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
_HEX40 = re.compile(r"^[0-9a-f]{40}$")
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
class ApprovalReopenError(RuntimeError):
def __init__(self, code: str, detail: str | None = None):
self.code = code
self.detail = detail
super().__init__(code if not detail else f"{code}:{detail}")
@dataclass(frozen=True)
class ReopenRequest:
candidate_id: str
expected_approval_fingerprint: str
expected_source_commit: str
expected_tree_oid: str
expected_artifact_sha256: str
expected_manifest_sha256: str
expected_release_plan_sha256: str
expected_policy_sha256: str
reason: str
idempotency_key: str
@classmethod
def from_mapping(cls, raw: Mapping[str, Any]) -> "ReopenRequest":
if raw.get("schema") != REQUEST_SCHEMA:
raise ApprovalReopenError("APPROVAL_REOPEN_SCHEMA_INVALID")
allowed = {
"schema", "candidate_id", "expected_approval_fingerprint",
"expected_source_commit", "expected_tree_oid",
"expected_artifact_sha256", "expected_manifest_sha256",
"expected_release_plan_sha256", "expected_policy_sha256",
"reason", "idempotency_key",
}
if set(raw) != allowed:
raise ApprovalReopenError("APPROVAL_REOPEN_REQUEST_FIELDS_INVALID")
req = cls(
candidate_id=str(raw["candidate_id"]),
expected_approval_fingerprint=str(raw["expected_approval_fingerprint"]),
expected_source_commit=str(raw["expected_source_commit"]),
expected_tree_oid=str(raw["expected_tree_oid"]),
expected_artifact_sha256=str(raw["expected_artifact_sha256"]),
expected_manifest_sha256=str(raw["expected_manifest_sha256"]),
expected_release_plan_sha256=str(raw["expected_release_plan_sha256"]),
expected_policy_sha256=str(raw["expected_policy_sha256"]),
reason=str(raw["reason"]),
idempotency_key=str(raw["idempotency_key"]),
)
if not req.candidate_id or "/" in req.candidate_id or ".." in req.candidate_id:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_ID_INVALID")
if not _HEX64.fullmatch(req.expected_approval_fingerprint):
raise ApprovalReopenError("APPROVAL_REOPEN_FINGERPRINT_INVALID")
if not _HEX40.fullmatch(req.expected_source_commit):
raise ApprovalReopenError("APPROVAL_REOPEN_SOURCE_INVALID")
if not _HEX40.fullmatch(req.expected_tree_oid):
raise ApprovalReopenError("APPROVAL_REOPEN_TREE_INVALID")
for value, code in (
(req.expected_artifact_sha256, "APPROVAL_REOPEN_ARTIFACT_INVALID"),
(req.expected_manifest_sha256, "APPROVAL_REOPEN_MANIFEST_INVALID"),
(req.expected_release_plan_sha256, "APPROVAL_REOPEN_RELEASE_PLAN_INVALID"),
(req.expected_policy_sha256, "APPROVAL_REOPEN_POLICY_INVALID"),
):
if not _HEX64.fullmatch(value):
raise ApprovalReopenError(code)
if not (1 <= len(req.reason.strip()) <= 512):
raise ApprovalReopenError("APPROVAL_REOPEN_REASON_INVALID")
if not _IDEMPOTENCY_RE.fullmatch(req.idempotency_key):
raise ApprovalReopenError("APPROVAL_REOPEN_IDEMPOTENCY_KEY_INVALID")
return req
def _canon(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
def _sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _time(value: Any, code: str) -> datetime:
if not isinstance(value, str) or not value:
raise ApprovalReopenError(code)
try:
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise ApprovalReopenError(code) from exc
if dt.tzinfo is None:
raise ApprovalReopenError(code)
return dt.astimezone(timezone.utc)
def _secure_json(path: Path) -> dict[str, Any]:
if path.is_symlink() or not path.is_file():
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_INVALID")
st = path.stat()
if st.st_mode & 0o022:
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_WRITABLE")
if st.st_uid not in {0, os.geteuid()}:
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_OWNER_INVALID")
try:
obj = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_INVALID") from exc
if not isinstance(obj, dict):
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_INVALID")
return obj
def _fsync_dir(path: Path) -> None:
fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(fd)
finally:
os.close(fd)
def _paths(root: Path, cid: str) -> dict[str, Path]:
return {
"candidate": root / "candidates" / f"{cid}.json",
"saga": root / "sagas" / f"{cid}.json",
"plan": root / "release-plans" / f"{cid}.json",
"acceptance": root / "acceptance" / f"{cid}.json",
"approval": root / "approvals" / f"{cid}.json",
}
def _history(root: Path, cid: str, fp: str) -> Path:
return root / "approval-history" / cid / f"{fp}.json"
def _archive(active: Path, history: Path) -> None:
history.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
try:
os.chmod(history.parent, 0o750)
except PermissionError:
pass
original = active.read_bytes()
if history.exists():
if history.is_symlink() or not history.is_file() or history.read_bytes() != original:
raise ApprovalReopenError("APPROVAL_REOPEN_HISTORY_COLLISION")
return
try:
os.link(active, history)
except OSError:
fd = os.open(history, os.O_WRONLY | os.O_CREAT | os.O_EXCL,
active.stat().st_mode & 0o777)
try:
view = memoryview(original)
while view:
n = os.write(fd, view)
view = view[n:]
os.fsync(fd)
finally:
os.close(fd)
if history.read_bytes() != original:
raise ApprovalReopenError("APPROVAL_REOPEN_HISTORY_COPY_FAILED")
_fsync_dir(history.parent)
def _details(event: Mapping[str, Any]) -> Mapping[str, Any]:
val = event.get("details")
return val if isinstance(val, Mapping) else {}
def _cycle_event(saga: Mapping[str, Any], cycle_id: str) -> Mapping[str, Any] | None:
for event in saga.get("events") or []:
if isinstance(event, Mapping) and event.get("event") == EVENT_NAME \
and _details(event).get("approval_cycle_id") == cycle_id:
return event
return None
def _state(name: str) -> Any:
from silent_release_hardening import SagaState # lazy: avoids import cycle
return getattr(SagaState, name)
def _cycle_id(req: ReopenRequest) -> str:
return "approval-reopen-" + _sha(_canon({
"schema": REQUEST_SCHEMA,
"candidate_id": req.candidate_id,
"old_fingerprint": req.expected_approval_fingerprint,
"source_commit": req.expected_source_commit,
"tree_oid": req.expected_tree_oid,
"artifact_sha256": req.expected_artifact_sha256,
"manifest_sha256": req.expected_manifest_sha256,
"release_plan_sha256": req.expected_release_plan_sha256,
"policy_sha256": req.expected_policy_sha256,
"idempotency_key": req.idempotency_key,
}))[:24]
def _validate(req: ReopenRequest, candidate: Mapping[str, Any],
saga: Mapping[str, Any], plan: Mapping[str, Any],
acceptance: Mapping[str, Any], approval: Mapping[str, Any],
now: datetime) -> None:
if saga.get("journal_integrity") is False:
raise ApprovalReopenError("APPROVAL_REOPEN_JOURNAL_INTEGRITY_INVALID")
if saga.get("state") != "APPROVED":
raise ApprovalReopenError("APPROVAL_REOPEN_SAGA_STATE_INVALID")
if candidate.get("candidate_id") != req.candidate_id:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_MISMATCH")
if candidate.get("state") not in {"READY_FOR_APPROVAL", "APPROVED"}:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_STATE_INVALID")
if candidate.get("approval_state") not in {None, "PENDING"}:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_APPROVAL_STATE_INVALID")
# ReleaseCandidate v1 durable records do not persist side_effects_executed.
# Compatibility rule is intentionally fail-closed:
# - an absent key is accepted only together with the independent Saga
# invariant below that the accepted Owner-approval event is the final
# Saga event;
# - if a future durable schema persists the key, it must be exactly False;
# - explicit True/None/number/string values are rejected.
if "side_effects_executed" in candidate and candidate.get("side_effects_executed") is not False:
raise ApprovalReopenError("APPROVAL_REOPEN_SIDE_EFFECTS_ALREADY_EXECUTED")
checks = {
"source": candidate.get("source_commit") == req.expected_source_commit,
"tree": candidate.get("tree_oid") == req.expected_tree_oid,
"artifact": candidate.get("artifact_sha256") == req.expected_artifact_sha256,
"manifest": candidate.get("manifest_sha256") == req.expected_manifest_sha256,
"approval_fingerprint": approval.get("fingerprint") == req.expected_approval_fingerprint,
"approval_source": approval.get("source_commit") == req.expected_source_commit,
"approval_tree": approval.get("tree_oid") == req.expected_tree_oid,
"approval_artifact": approval.get("artifact_sha256") == req.expected_artifact_sha256,
"approval_manifest": approval.get("manifest_sha256") == req.expected_manifest_sha256,
"approval_release_plan": approval.get("release_plan_sha256") == req.expected_release_plan_sha256,
"approval_policy": approval.get("policy_sha256") == req.expected_policy_sha256,
"acceptance": acceptance.get("status") == "ACCEPTED",
"acceptance_release_plan": acceptance.get("release_plan_sha256") == req.expected_release_plan_sha256,
"plan_sha": (plan.get("sha256") or plan.get("release_plan_sha256")) == req.expected_release_plan_sha256,
"plan_policy": plan.get("policy_sha256") == req.expected_policy_sha256,
}
bad = sorted(k for k, ok in checks.items() if not ok)
if bad:
raise ApprovalReopenError("APPROVAL_REOPEN_BINDING_MISMATCH", ",".join(bad))
if now <= _time(approval.get("expires_at"), "APPROVAL_REOPEN_APPROVAL_EXPIRY_INVALID"):
raise ApprovalReopenError("APPROVAL_REOPEN_APPROVAL_NOT_EXPIRED")
events = [e for e in saga.get("events") or [] if isinstance(e, Mapping)]
accepted = None
for i, event in enumerate(events):
if event.get("event") == "signed_owner_approval_accepted" \
and _details(event).get("approval_fingerprint") == req.expected_approval_fingerprint:
accepted = i
if accepted is None:
raise ApprovalReopenError("APPROVAL_REOPEN_ACCEPTED_EVENT_MISSING")
if accepted != len(events) - 1:
raise ApprovalReopenError("APPROVAL_REOPEN_SAGA_ALREADY_ADVANCED")
def reopen_owner_approval(service: Any, raw: Mapping[str, Any], *,
state_root: Path | None = None,
now: datetime | None = None) -> dict[str, Any]:
req = ReopenRequest.from_mapping(raw)
raw_root = state_root or os.environ.get("AIOS_SILENT_RELEASE_STATE_ROOT", "")
if not str(raw_root):
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_ROOT_MISSING")
root = Path(raw_root).expanduser().resolve()
paths = _paths(root, req.candidate_id)
cycle = _cycle_id(req)
history = _history(root, req.candidate_id, req.expected_approval_fingerprint)
saga = _secure_json(paths["saga"])
prior = _cycle_event(saga, cycle)
if prior is not None:
if saga.get("state") != "READY_FOR_APPROVAL":
raise ApprovalReopenError("APPROVAL_REOPEN_CYCLE_ALREADY_CONSUMED")
if not history.is_file():
raise ApprovalReopenError("APPROVAL_REOPEN_HISTORY_MISSING")
if paths["approval"].exists():
active = _secure_json(paths["approval"])
if active.get("fingerprint") != req.expected_approval_fingerprint:
raise ApprovalReopenError("APPROVAL_REOPEN_ACTIVE_APPROVAL_CHANGED")
paths["approval"].unlink()
_fsync_dir(paths["approval"].parent)
return {
"schema": "aios.owner-approval-reopen-result.v1",
"status": "IDEMPOTENT",
"candidate_id": req.candidate_id,
"approval_cycle_id": cycle,
"supersedes_fingerprint": req.expected_approval_fingerprint,
"saga_state": "READY_FOR_APPROVAL",
"history_ref": str(history.relative_to(root)),
}
candidate = _secure_json(paths["candidate"])
plan = _secure_json(paths["plan"])
acceptance = _secure_json(paths["acceptance"])
approval = _secure_json(paths["approval"])
_validate(req, candidate, saga, plan, acceptance, approval,
(now or datetime.now(timezone.utc)).astimezone(timezone.utc))
_archive(paths["approval"], history)
event_details = {
"schema": "aios.owner-approval-reopen-event.v1",
"approval_cycle_id": cycle,
"supersedes_approval_fingerprint": req.expected_approval_fingerprint,
"expired_at": approval.get("expires_at"),
"release_plan_sha256": req.expected_release_plan_sha256,
"source_commit": req.expected_source_commit,
"tree_oid": req.expected_tree_oid,
"artifact_sha256": req.expected_artifact_sha256,
"reason": req.reason.strip(),
"idempotency_digest": _sha(req.idempotency_key.encode()),
}
try:
service.sagas.compare_and_set(
req.candidate_id,
expected_states={_state("APPROVED")},
target=_state("READY_FOR_APPROVAL"),
event=EVENT_NAME,
details=event_details,
)
except Exception as exc:
raise ApprovalReopenError("APPROVAL_REOPEN_SAGA_CAS_FAILED") from exc
try:
active = _secure_json(paths["approval"])
if active.get("fingerprint") != req.expected_approval_fingerprint:
raise ApprovalReopenError("APPROVAL_REOPEN_ACTIVE_APPROVAL_CHANGED")
paths["approval"].unlink()
_fsync_dir(paths["approval"].parent)
except ApprovalReopenError:
raise
except OSError as exc:
raise ApprovalReopenError("APPROVAL_REOPEN_ACTIVE_APPROVAL_REMOVE_FAILED") from exc
return {
"schema": "aios.owner-approval-reopen-result.v1",
"status": "REOPENED",
"candidate_id": req.candidate_id,
"approval_cycle_id": cycle,
"supersedes_fingerprint": req.expected_approval_fingerprint,
"saga_state": "READY_FOR_APPROVAL",
"history_ref": str(history.relative_to(root)),
"release_plan_sha256": req.expected_release_plan_sha256,
"source_commit": req.expected_source_commit,
"tree_oid": req.expected_tree_oid,
}
+44
View File
@@ -782,3 +782,47 @@ paths:
'200': {description: Safe GitOps plan}
'401': {description: Unauthorized}
x-openai-isConsequential: true
/reopenOwnerApproval:
post:
operationId: reopenOwnerApproval
summary: Reopen one expired Owner approval cycle without rebuilding release bytes
security:
- bearerAuth: []
x-openai-isConsequential: true
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required:
- schema
- candidate_id
- expected_approval_fingerprint
- expected_source_commit
- expected_tree_oid
- expected_artifact_sha256
- expected_manifest_sha256
- expected_release_plan_sha256
- expected_policy_sha256
- reason
- idempotency_key
properties:
schema:
type: string
enum: [aios.owner-approval-reopen-request.v1]
candidate_id: {type: string}
expected_approval_fingerprint: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_source_commit: {type: string, pattern: "^[0-9a-f]{40}$"}
expected_tree_oid: {type: string, pattern: "^[0-9a-f]{40}$"}
expected_artifact_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_manifest_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_release_plan_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_policy_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
reason: {type: string, minLength: 1, maxLength: 512}
idempotency_key: {type: string, minLength: 8, maxLength: 128}
responses:
"200": {description: Reopened or idempotent replay}
"403": {description: Missing release:approve capability}
"409": {description: Governed reopen gate rejected}
+116
View File
@@ -0,0 +1,116 @@
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT/"active-runtime"/"app"))
import release_approval_reopen as reopen
CID="rc-v2r2c2-4642b74af62c"; SRC="4"*40; TREE="a"*40
ART="5"*64; MAN="8"*64; PLAN="f"*64; POL="2"*64; FP="9"*64
NOW=datetime(2026,9,12,3,10,tzinfo=timezone.utc)
class Sagas:
def __init__(self, root): self.root=root
def compare_and_set(self, candidate_id, *, expected_states, target, event, details):
p=self.root/"sagas"/f"{candidate_id}.json"; x=json.loads(p.read_text())
if x["state"] not in expected_states: raise RuntimeError("CAS")
x["state"]=target; x["revision"]=int(x.get("revision",0))+1
x.setdefault("events",[]).append({"event":event,"details":details,"timestamp":"2026-09-12T03:10:00Z"})
p.write_text(json.dumps(x,sort_keys=True)); return x
class Service:
def __init__(self, root): self.sagas=Sagas(root)
def w(p,o):
p.parent.mkdir(parents=True,exist_ok=True); p.write_text(json.dumps(o,sort_keys=True)); p.chmod(0o640)
_MISSING = object()
def state(tmp, expired=True, side_marker=_MISSING, saga="APPROVED", acceptance="ACCEPTED", extra=False):
r=tmp/"silent-release"
for d in ("candidates","sagas","release-plans","acceptance","approvals"):
(r/d).mkdir(parents=True,exist_ok=True,mode=0o750)
candidate={"candidate_id":CID,"state":"READY_FOR_APPROVAL","approval_state":"PENDING",
"source_commit":SRC,"tree_oid":TREE,"artifact_sha256":ART,"manifest_sha256":MAN}
if side_marker is not _MISSING:
candidate["side_effects_executed"]=side_marker
w(r/"candidates"/f"{CID}.json", candidate)
ev=[{"event":"saga_created","details":{}},{"event":"trusted_candidate_ready_for_approval","details":{}},
{"event":"signed_owner_approval_accepted","details":{"approval_fingerprint":FP}}]
if extra: ev.append({"event":"deployment_started","details":{}})
w(r/"sagas"/f"{CID}.json",{"candidate_id":CID,"state":saga,"revision":2,"journal_integrity":True,"events":ev})
w(r/"release-plans"/f"{CID}.json",{"sha256":PLAN,"policy_sha256":POL})
w(r/"acceptance"/f"{CID}.json",{"status":acceptance,"release_plan_sha256":PLAN})
w(r/"approvals"/f"{CID}.json",{"candidate_id":CID,"fingerprint":FP,"source_commit":SRC,"tree_oid":TREE,
"artifact_sha256":ART,"manifest_sha256":MAN,"release_plan_sha256":PLAN,"policy_sha256":POL,
"expires_at":"2026-09-12T02:59:05Z" if expired else "2026-09-12T04:59:05Z"})
return r
def req(key="reopen:test:0001"):
return {"schema":reopen.REQUEST_SCHEMA,"candidate_id":CID,"expected_approval_fingerprint":FP,
"expected_source_commit":SRC,"expected_tree_oid":TREE,"expected_artifact_sha256":ART,
"expected_manifest_sha256":MAN,"expected_release_plan_sha256":PLAN,"expected_policy_sha256":POL,
"reason":"expired approval after pre-side-effect infrastructure recovery","idempotency_key":key}
@pytest.fixture(autouse=True)
def enum(monkeypatch): monkeypatch.setattr(reopen,"_state",lambda x:x)
def test_happy(tmp_path):
r=state(tmp_path); old=(r/"approvals"/f"{CID}.json").read_bytes()
out=reopen.reopen_owner_approval(Service(r),req(),state_root=r,now=NOW)
assert out["status"]=="REOPENED"; s=json.loads((r/"sagas"/f"{CID}.json").read_text())
assert s["state"]=="READY_FOR_APPROVAL" and s["events"][-1]["event"]==reopen.EVENT_NAME
assert not (r/"approvals"/f"{CID}.json").exists()
assert (r/out["history_ref"]).read_bytes()==old
def test_idempotent(tmp_path):
r=state(tmp_path); svc=Service(r); a=reopen.reopen_owner_approval(svc,req(),state_root=r,now=NOW)
b=reopen.reopen_owner_approval(svc,req(),state_root=r,now=NOW)
assert b["status"]=="IDEMPOTENT" and a["approval_cycle_id"]==b["approval_cycle_id"]
@pytest.mark.parametrize("kind",["unexpired","saga","acceptance","side","advanced"])
def test_rejections(tmp_path,kind):
kw={}
expected=""
if kind=="unexpired": kw["expired"]=False; expected="APPROVAL_REOPEN_APPROVAL_NOT_EXPIRED"
if kind=="saga": kw["saga"]="READY_FOR_APPROVAL"; expected="APPROVAL_REOPEN_SAGA_STATE_INVALID"
if kind=="acceptance": kw["acceptance"]="REJECTED"; expected="APPROVAL_REOPEN_BINDING_MISMATCH"
if kind=="side": kw["side_marker"]=True; expected="APPROVAL_REOPEN_SIDE_EFFECTS_ALREADY_EXECUTED"
if kind=="advanced": kw["extra"]=True; expected="APPROVAL_REOPEN_SAGA_ALREADY_ADVANCED"
r=state(tmp_path,**kw)
with pytest.raises(reopen.ApprovalReopenError,match=expected):
reopen.reopen_owner_approval(Service(r),req(),state_root=r,now=NOW)
def test_binding_drift(tmp_path):
r=state(tmp_path); q=req(); q["expected_artifact_sha256"]="6"*64
with pytest.raises(reopen.ApprovalReopenError,match="APPROVAL_REOPEN_BINDING_MISMATCH"):
reopen.reopen_owner_approval(Service(r),q,state_root=r,now=NOW)
def test_history_collision(tmp_path):
r=state(tmp_path); h=r/"approval-history"/CID/f"{FP}.json"; h.parent.mkdir(parents=True); h.write_text("{}")
with pytest.raises(reopen.ApprovalReopenError,match="APPROVAL_REOPEN_HISTORY_COLLISION"):
reopen.reopen_owner_approval(Service(r),req(),state_root=r,now=NOW)
def test_real_durable_schema_without_side_effect_field_is_supported(tmp_path):
r=state(tmp_path)
candidate=json.loads((r/"candidates"/f"{CID}.json").read_text())
assert "side_effects_executed" not in candidate
out=reopen.reopen_owner_approval(
Service(r), req("reopen:test:real-schema"), state_root=r, now=NOW
)
assert out["status"]=="REOPENED"
@pytest.mark.parametrize("value",[True,None,1,"false"])
def test_explicit_non_false_side_effect_marker_fails_closed(tmp_path,value):
r=state(tmp_path,side_marker=value)
with pytest.raises(
reopen.ApprovalReopenError,
match="APPROVAL_REOPEN_SIDE_EFFECTS_ALREADY_EXECUTED"
):
reopen.reopen_owner_approval(
Service(r), req("reopen:test:marker:bad"), state_root=r, now=NOW
)
+75
View File
@@ -0,0 +1,75 @@
from __future__ import annotations
import ast
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
STAGING = ROOT / "active-runtime" / "app" / "aios_model_api_staging.py"
if not STAGING.is_file():
pytest.skip(
"shadow-gate contract runs after package is applied to an AIOS checkout",
allow_module_level=True,
)
def _literal_string_set(node: ast.AST) -> set[str]:
"""Extract string literals from a set or frozenset({...}) literal.
The governed codebase represents allow-lists as ``frozenset({...})``; a
plain ``{...}`` set literal is also accepted. Only string literal members
are allowed in the silent-release allow-list.
"""
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "frozenset":
if len(node.args) == 1:
node = node.args[0]
if not isinstance(node, ast.Set):
raise AssertionError("SILENT_RELEASE_POST_PATHS must remain a set literal")
values: set[str] = set()
for elt in node.elts:
if not isinstance(elt, ast.Constant) or not isinstance(elt.value, str):
raise AssertionError("silent-release allow-list must contain only string literals")
values.add(elt.value)
return values
def test_reopen_is_narrow_silent_release_post():
tree = ast.parse(STAGING.read_text(encoding="utf-8"))
allow = None
for node in tree.body:
if not isinstance(node, ast.Assign):
continue
if any(isinstance(t, ast.Name) and t.id == "SILENT_RELEASE_POST_PATHS" for t in node.targets):
allow = _literal_string_set(node.value)
break
assert allow is not None
assert "/reopenOwnerApproval" in allow
# Existing governed silent-release paths must stay present.
assert {
"/createReleaseCandidate",
"/createReleaseIntent",
"/submitSignedApproval",
"/submitOwnerApproval",
"/startDeployment",
}.issubset(allow)
def test_reopen_route_still_requires_release_approve():
import re
source = STAGING.read_text(encoding="utf-8")
# Locate the route handler (the `== "/reopenOwnerApproval"` comparison),
# independent of single/double-quote style on the allow-list entry.
m = re.search(r'==\s*["\']/reopenOwnerApproval["\']', source)
assert m is not None, "reopen handler route comparison not found"
tail = source[m.start() : m.start() + 1200]
assert 'self.need("release:approve")' in tail
assert "reopen_owner_approval(silent_release_service(), data)" in tail
def test_global_write_gate_is_not_enabled_by_source_patch():
source = STAGING.read_text(encoding="utf-8")
# v1.2 must not hard-code or synthesize a global write enable.
assert 'AIOS_GITOPS_WRITES_ENABLED=true' not in source
assert 'AIOS_GITOPS_WRITES_ENABLED="true"' not in source