467 lines
18 KiB
Python
467 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""Append-only acceptance / supersession EVENT log (P1-G, finding #14).
|
||
|
||
문제: report 는 불변 SNAPSHOT 이지만 CompletionRecord 상태(Submitted → Accepted /
|
||
Changes-Requested / Blocked)는 "바뀐다"고만 서술돼 있어서, 별도의 수락 이벤트나
|
||
supersedes 모델이 없었다. 결과: 어느 스냅샷이 최신 시도인지·무엇이 수락됐는지 알 수
|
||
없고, 태그 검색이 이미 대체된(superseded) 낡은 결정을 그대로 노출한다.
|
||
|
||
해법: report 스냅샷은 그대로 불변으로 두고, 상태 변화를 append-only EVENT 로 옮긴다.
|
||
- 각 리뷰 결정(수락/변경요청/차단)은 이 원장에 한 줄(JSON) 이벤트로 append 된다.
|
||
- 리포트는 절대 수정되지 않는다(불변 스냅샷 + append-only 이벤트).
|
||
- "이 workflow/role 의 최신 수락 리포트는?" / "이 리포트는 대체됐나?" 를 이벤트로 질의.
|
||
|
||
원장 위치: <state_dir>/acceptance-events.jsonl (한 줄 = JSON 이벤트)
|
||
|
||
이벤트 필드:
|
||
{
|
||
"acceptance-event-id": "ae-<UTCstamp>-<hex>",
|
||
"report-id": "<결정 대상 report-id>",
|
||
"decision": "accepted" | "changes-requested" | "blocked",
|
||
"accepted-report-id": "<accepted 시 = report-id>",
|
||
"rejected-report-id": "<changes-requested/blocked 시 = report-id>",
|
||
"supersedes-report-id": "<선택: 이 리포트가 대체하는 이전 report-id>",
|
||
"workflow-id": "<질의 필터용>",
|
||
"role-id": "<질의 필터용>",
|
||
"effective-at": "2026-07-10T12:00:00Z"
|
||
}
|
||
|
||
CLI:
|
||
state_engine.py review-artifact --workflow WF --report PATH --decision accepted --reviewer ROLE
|
||
[--supersedes PRIOR-RID] # 권장/신뢰 경로(id+sha+권한+self-review 검증)
|
||
acceptance_log.py append --report-id RID --decision accepted --workflow WF --reviewer ROLE
|
||
[--supersedes PRIOR-RID] # 위 API로 위임하는 호환 entrypoint
|
||
acceptance_log.py latest-accepted --workflow WF [--role ROLE] # 최신 수락 report-id 출력
|
||
acceptance_log.py is-superseded --report-id RID # yes/no 출력
|
||
acceptance_log.py excluded # 대체/거부된 report-id 목록
|
||
|
||
Import-safe API (재사용):
|
||
read_events() -> list[dict]
|
||
validate(event) -> list[str] # 위반 사유(빈 리스트 = 통과), 예외 없음
|
||
latest_accepted(workflow, role) -> str|None
|
||
is_superseded(report_id) -> bool
|
||
excluded_report_ids() -> set[str] # report_tags 가 기본 제외할 대상(대체 ∪ 거부)
|
||
query(kind, **kwargs) # 위 질의들의 디스패처
|
||
|
||
강건성 계약(fail-safe):
|
||
- 절대 파이프라인을 크래시시키지 않는다. 질의/읽기는 어떤 예외에도 안전값으로 degrade.
|
||
- workspace 미설정/파일 부재 -> 질의는 안전값, append/review는 fail-closed.
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
|
||
DECISIONS = ("accepted", "changes-requested", "blocked")
|
||
REJECTING = ("changes-requested", "blocked")
|
||
|
||
|
||
def _log(msg):
|
||
try:
|
||
sys.stderr.write(f"[acceptance_log] {msg}\n")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _events_path(create=False):
|
||
"""<state_dir>/acceptance-events.jsonl. workspace 미해석이면 None(조용히 degrade)."""
|
||
try:
|
||
import _workspace as W # noqa: E402
|
||
sd = W.state_dir()
|
||
except Exception as e: # WorkspaceNotSetError 포함
|
||
_log(f"workspace 미해석 — 이벤트 로그 스킵: {e}")
|
||
return None
|
||
try:
|
||
if create:
|
||
os.makedirs(sd, exist_ok=True)
|
||
except Exception as e:
|
||
_log(f"state_dir 생성 실패: {e}")
|
||
return None
|
||
return os.path.join(sd, "acceptance-events.jsonl")
|
||
|
||
|
||
def _now():
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
||
|
||
def _resolve_report_path(report_id, workflow=None):
|
||
"""report-id -> completion-records/<wf>/<report-id>.report.yaml 실존 경로(없으면 None).
|
||
finding P0-4c: 수락 이벤트가 실존 report 만 참조하도록 하는 데 쓴다(ghost acceptance 차단)."""
|
||
import glob
|
||
try:
|
||
import _workspace as W # noqa: E402
|
||
cr = W.records_dir()
|
||
except Exception:
|
||
return None
|
||
rid = str(report_id)
|
||
fname = rid if rid.endswith(".report.yaml") else f"{rid}.report.yaml"
|
||
if workflow:
|
||
p = os.path.join(cr, str(workflow), fname)
|
||
if os.path.exists(p):
|
||
return p
|
||
hits = glob.glob(os.path.join(cr, "**", fname), recursive=True)
|
||
return hits[0] if hits else None
|
||
|
||
|
||
# ---------------------------------------------------------------- read / write
|
||
|
||
def read_events():
|
||
"""원장의 모든 이벤트를 파일(append) 순서로 반환. 어떤 예외에도 [] 로 degrade."""
|
||
p = _events_path(create=False)
|
||
if not p or not os.path.exists(p):
|
||
return []
|
||
out = []
|
||
try:
|
||
with open(p, encoding="utf-8") as fh:
|
||
for line in fh:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
obj = json.loads(line)
|
||
if isinstance(obj, dict):
|
||
out.append(obj)
|
||
except Exception:
|
||
continue # malformed 한 줄은 건너뛴다(원장을 무너뜨리지 않음)
|
||
except Exception as e:
|
||
_log(f"원장 읽기 실패: {e}")
|
||
return []
|
||
return out
|
||
|
||
|
||
def validate(event):
|
||
"""이벤트 구조 검증. 위반 사유 문자열 리스트 반환(빈 리스트 = 통과). 예외를 던지지 않는다."""
|
||
errs = []
|
||
try:
|
||
if not isinstance(event, dict):
|
||
return ["event 는 dict 여야 한다"]
|
||
if not event.get("report-id"):
|
||
errs.append("report-id 누락")
|
||
dec = event.get("decision")
|
||
if dec not in DECISIONS:
|
||
errs.append(f"decision 은 {DECISIONS} 중 하나여야 한다 (got: {dec!r})")
|
||
if not event.get("acceptance-event-id"):
|
||
errs.append("acceptance-event-id 누락")
|
||
if not event.get("effective-at"):
|
||
errs.append("effective-at 누락")
|
||
except Exception as e: # 방어적 — validate 는 결코 크래시하지 않는다
|
||
return [f"validate 내부 오류: {e}"]
|
||
return errs
|
||
|
||
|
||
def build_event(report_id, decision, workflow=None, role=None, supersedes=None,
|
||
accepted_report_id=None, rejected_report_id=None, effective_at=None,
|
||
report_sha256=None, artifact_kind=None, producer_role_id=None,
|
||
reviewer=None, authorization=None):
|
||
"""이벤트 dict 를 만든다(파일 기록은 하지 않음). accepted/rejected 는 미지정 시 파생."""
|
||
dec = (decision or "").strip().lower()
|
||
ev = {
|
||
"acceptance-event-id": f"ae-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}-{uuid.uuid4().hex[:8]}",
|
||
"report-id": report_id,
|
||
"decision": dec,
|
||
"effective-at": effective_at or _now(),
|
||
}
|
||
if workflow:
|
||
ev["workflow-id"] = workflow
|
||
if role:
|
||
ev["role-id"] = role
|
||
if supersedes:
|
||
ev["supersedes-report-id"] = supersedes
|
||
if report_sha256:
|
||
ev["report-sha256"] = report_sha256
|
||
ev["artifact-sha256"] = report_sha256
|
||
if artifact_kind:
|
||
ev["artifact-kind"] = artifact_kind
|
||
if producer_role_id:
|
||
ev["producer-role-id"] = producer_role_id
|
||
if reviewer:
|
||
ev["reviewer"] = reviewer
|
||
if authorization:
|
||
ev["authorization"] = authorization
|
||
# accepted/rejected 파생(명시 override 우선)
|
||
if accepted_report_id:
|
||
ev["accepted-report-id"] = accepted_report_id
|
||
elif dec == "accepted":
|
||
ev["accepted-report-id"] = report_id
|
||
if rejected_report_id:
|
||
ev["rejected-report-id"] = rejected_report_id
|
||
elif dec in REJECTING:
|
||
ev["rejected-report-id"] = report_id
|
||
return ev
|
||
|
||
|
||
def append_event(event):
|
||
"""이벤트를 원장에 append. 성공 True / degrade False. 절대 크래시하지 않는다."""
|
||
p = _events_path(create=True)
|
||
if not p:
|
||
return False
|
||
try:
|
||
with open(p, "a", encoding="utf-8") as fh:
|
||
try:
|
||
import fcntl
|
||
fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
|
||
except Exception:
|
||
pass
|
||
fh.write(json.dumps(event, ensure_ascii=False) + "\n")
|
||
fh.flush()
|
||
os.fsync(fh.fileno())
|
||
try:
|
||
import fcntl
|
||
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
|
||
except Exception:
|
||
pass
|
||
return True
|
||
except Exception as e:
|
||
_log(f"이벤트 append 실패: {e}")
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------- queries
|
||
|
||
def effective_decision(workflow_id, artifact_id, artifact_sha256=None):
|
||
"""Return the latest effective decision for one immutable revision.
|
||
|
||
An accepted event without the requested sha is not valid for a sha-bound
|
||
query. A later superseding revision invalidates the old artifact even if
|
||
its historical accepted event remains in the append-only ledger.
|
||
"""
|
||
result = None
|
||
for ev in read_events():
|
||
if ev.get("workflow-id") != workflow_id:
|
||
continue
|
||
if ev.get("supersedes-report-id") == artifact_id:
|
||
result = "Superseded"
|
||
continue
|
||
rid = ev.get("report-id")
|
||
if rid != artifact_id:
|
||
continue
|
||
if artifact_sha256 is not None:
|
||
event_sha = ev.get("artifact-sha256") or ev.get("report-sha256")
|
||
if event_sha != artifact_sha256:
|
||
continue
|
||
result = {
|
||
"accepted": "Accepted",
|
||
"changes-requested": "ChangesRequested",
|
||
"blocked": "Blocked",
|
||
}.get(ev.get("decision"))
|
||
return result
|
||
|
||
|
||
def is_effectively_accepted(workflow_id, artifact_id, artifact_sha256=None):
|
||
return effective_decision(workflow_id, artifact_id, artifact_sha256) == "Accepted"
|
||
|
||
|
||
def latest_accepted(workflow=None, role=None):
|
||
"""Latest still-effective accepted report id (not stale/rejected/superseded)."""
|
||
result = None
|
||
for ev in read_events():
|
||
if ev.get("decision") != "accepted":
|
||
continue
|
||
if workflow is not None and ev.get("workflow-id") != workflow:
|
||
continue
|
||
if role is not None and ev.get("role-id") != role:
|
||
continue
|
||
rid = ev.get("accepted-report-id") or ev.get("report-id")
|
||
sha = ev.get("artifact-sha256") or ev.get("report-sha256")
|
||
if effective_decision(ev.get("workflow-id"), rid, sha) == "Accepted":
|
||
result = rid
|
||
return result
|
||
|
||
|
||
def latest_accepted_artifact(workflow, *, producer_role=None, artifact_kind=None):
|
||
"""Return the latest effective acceptance event by artifact identity.
|
||
|
||
``role-id`` is the reviewer in canonical events. Handoff consumers must join
|
||
on ``producer-role-id`` and ``artifact-kind`` instead.
|
||
"""
|
||
result = None
|
||
for ev in read_events():
|
||
if ev.get("decision") != "accepted" or ev.get("workflow-id") != workflow:
|
||
continue
|
||
if producer_role is not None and str(ev.get("producer-role-id") or "").upper() != str(producer_role).upper():
|
||
continue
|
||
if artifact_kind is not None and ev.get("artifact-kind") != artifact_kind:
|
||
continue
|
||
rid = ev.get("accepted-report-id") or ev.get("report-id")
|
||
sha = ev.get("artifact-sha256") or ev.get("report-sha256")
|
||
if effective_decision(workflow, rid, sha) == "Accepted":
|
||
result = ev
|
||
return result
|
||
|
||
|
||
def _latest_decision_by_report(events):
|
||
latest = {}
|
||
for ev in events:
|
||
rid = ev.get("report-id")
|
||
if rid:
|
||
latest[rid] = ev.get("decision")
|
||
return latest
|
||
|
||
|
||
def superseded_report_ids():
|
||
"""어떤 이벤트에서 supersedes-report-id 로 지목된(=새 스냅샷이 대체한) report-id 집합."""
|
||
out = set()
|
||
for ev in read_events():
|
||
sup = ev.get("supersedes-report-id")
|
||
if sup:
|
||
out.add(sup)
|
||
return out
|
||
|
||
|
||
def is_superseded(report_id):
|
||
"""report_id 가 더 새로운 스냅샷에 의해 대체됐는가."""
|
||
if not report_id:
|
||
return False
|
||
return report_id in superseded_report_ids()
|
||
|
||
|
||
def excluded_report_ids():
|
||
"""peer 검색에서 기본 제외할 report-id 집합 = 대체됨(superseded) ∪ 거부됨(rejected).
|
||
|
||
거부됨 = 그 리포트의 최신 자기 결정이 changes-requested/blocked 인 경우.
|
||
(같은 report-id 가 나중에 accepted 되면 제외하지 않는다 — 최신 결정 우선.)"""
|
||
events = read_events()
|
||
excluded = set()
|
||
for ev in events:
|
||
sup = ev.get("supersedes-report-id")
|
||
if sup:
|
||
excluded.add(sup)
|
||
for rid, dec in _latest_decision_by_report(events).items():
|
||
if dec in REJECTING:
|
||
excluded.add(rid)
|
||
excluded.discard(None)
|
||
excluded.discard("")
|
||
return excluded
|
||
|
||
|
||
def is_current(report_id):
|
||
"""report_id 가 현재 유효(대체/거부되지 않음)한가."""
|
||
if not report_id:
|
||
return True # 식별 불가 -> 필터하지 않음(과잉 제외 방지)
|
||
return report_id not in excluded_report_ids()
|
||
|
||
|
||
def query(kind, **kwargs):
|
||
"""질의 디스패처(재사용용)."""
|
||
if kind == "latest-accepted":
|
||
return latest_accepted(kwargs.get("workflow"), kwargs.get("role"))
|
||
if kind == "latest-accepted-artifact":
|
||
return latest_accepted_artifact(
|
||
kwargs.get("workflow"), producer_role=kwargs.get("producer_role"),
|
||
artifact_kind=kwargs.get("artifact_kind"),
|
||
)
|
||
if kind == "is-superseded":
|
||
return is_superseded(kwargs.get("report_id"))
|
||
if kind == "is-current":
|
||
return is_current(kwargs.get("report_id"))
|
||
if kind == "excluded":
|
||
return excluded_report_ids()
|
||
if kind == "events":
|
||
return read_events()
|
||
if kind == "effective-decision":
|
||
return effective_decision(kwargs.get("workflow"), kwargs.get("artifact_id"),
|
||
kwargs.get("artifact_sha256"))
|
||
raise ValueError(f"unknown query kind: {kind}")
|
||
|
||
|
||
# ---------------------------------------------------------------- CLI
|
||
|
||
def _argval(args, flag):
|
||
return args[args.index(flag) + 1] if flag in args and args.index(flag) + 1 < len(args) else None
|
||
|
||
|
||
def main():
|
||
args = sys.argv[1:]
|
||
if not args:
|
||
sys.stderr.write(__doc__)
|
||
return 1
|
||
cmd = args[0]
|
||
|
||
if cmd == "append":
|
||
report_id = _argval(args, "--report-id")
|
||
decision = _argval(args, "--decision")
|
||
if not report_id or not decision:
|
||
sys.stderr.write("usage: acceptance_log.py append --report-id RID --decision "
|
||
"accepted|changes-requested|blocked [--workflow WF] [--role ROLE] "
|
||
"[--supersedes PRIOR-RID] [--report-sha256 HEX]\n")
|
||
return 2
|
||
# Compatibility entrypoint now delegates to the state engine's
|
||
# authorization path. A caller-supplied --role string is never itself
|
||
# proof of reviewer authority.
|
||
import _workspace as W # noqa: E402
|
||
W.require_workspace("acceptance_log")
|
||
wf_arg = _argval(args, "--workflow")
|
||
reviewer = _argval(args, "--reviewer")
|
||
if not wf_arg or not reviewer:
|
||
sys.stderr.write(
|
||
"[acceptance_log] BLOCK: append는 --workflow와 --reviewer가 필수다. "
|
||
"권장 명령: state_engine.py review-artifact --workflow WF --report PATH "
|
||
"--decision DECISION --reviewer ROLE\n")
|
||
return 2
|
||
rpath = _resolve_report_path(report_id, wf_arg)
|
||
if not rpath:
|
||
sys.stderr.write(
|
||
f"[acceptance_log] BLOCK: report-id '{report_id}' 에 해당하는 report 파일이 없다 — "
|
||
"존재하지 않는 report 를 수락/거부할 수 없다(ghost acceptance 차단, P0-4c).\n")
|
||
return 2
|
||
try:
|
||
import state_engine as SE # noqa: E402
|
||
ok, result = SE.review_artifact(
|
||
wf_arg, rpath, decision, reviewer,
|
||
supersedes=_argval(args, "--supersedes"),
|
||
)
|
||
except Exception as exc:
|
||
ok, result = False, str(exc)
|
||
if not ok:
|
||
sys.stderr.write(f"[acceptance_log] BLOCK: {result}\n")
|
||
return 2
|
||
print(result["acceptance-event-id"])
|
||
return 0
|
||
|
||
if cmd == "latest-accepted":
|
||
rid = latest_accepted(_argval(args, "--workflow"), _argval(args, "--role"))
|
||
if rid:
|
||
print(rid)
|
||
return 0
|
||
|
||
if cmd == "is-superseded":
|
||
rid = _argval(args, "--report-id")
|
||
print("yes" if is_superseded(rid) else "no")
|
||
return 0
|
||
|
||
if cmd == "excluded":
|
||
for rid in sorted(excluded_report_ids()):
|
||
print(rid)
|
||
return 0
|
||
|
||
if cmd == "effective-decision":
|
||
value = effective_decision(
|
||
_argval(args, "--workflow"), _argval(args, "--artifact-id"),
|
||
_argval(args, "--artifact-sha256"),
|
||
)
|
||
if value:
|
||
print(value)
|
||
return 0
|
||
|
||
if cmd == "events":
|
||
for ev in read_events():
|
||
print(json.dumps(ev, ensure_ascii=False))
|
||
return 0
|
||
|
||
sys.stderr.write(__doc__)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except Exception as e: # 최종 안전망 — 어떤 경우에도 크래시로 파이프라인을 막지 않는다
|
||
_log(f"unexpected: {e}")
|
||
# Mutation/query ambiguity must fail closed. Read APIs themselves keep
|
||
# returning safe empty values; an unexpected CLI error is never success.
|
||
sys.exit(2)
|