DongHyeonkaandClaude Opus 5
14afe94d76
R5 — 원장은 관문의 명령과 종료 코드를 적는데 **그 명령이 어느 대상에 돌았는지는 아무도 안 봤다.** 다른 프로젝트에 돌려 받은 exit 0 을 적어도 통과로 셌다. `_wrong_target()` 이 `docs/<이름>/` 경로와 명령 인자의 프로젝트 이름 둘 다 본다. 정상 원장 넷 새 error 0 (B 의 document-haness 원장 PASS · warn 3) 프로젝트 인자를 바꾼 원장 FAIL 「관문이 다른 대상에 돌았다」 기록 경로를 바꾼 원장 FAIL **유효 범위를 docstring 에 적었다** — 「다른 프로젝트」만 본다. 같은 프로젝트의 다른 기록에 돌린 관문은 안 잡는다. 기록 단위까지 보려면 관문마다 대상 단위를 계약이 먼저 정해야 한다. R12 — `check-required-content.py` 를 `OUTPUT_CHECKS` 에 더한다. B-003 이 병합돼 이제 된다. 세 상태 표를 지키는 것을 확인했다 — `ca-tmpl` exit 2 · `keycloak-session-store` exit 0 · 없는 프로젝트 exit 2. `OUTPUT CHECKS` 의 error 가 5 → 7 이 된다. **부채가 늘어난 것이 아니라 세는 자리가 늘었다** — `ca-tmpl` 에 계약이 없다는 한 사실이 `check_evidence` · `check-required-content` · `TECH LOG TREES` 세 곳에서 셈된다. 같은 사실을 세 번 세지 않는다. 회귀 `scripts/tests/test_run_target.py` 2건 (123 → 125). **「정상 원장은 그대로 통과한다」 대조를 함께 넣는다** — 무조건 거절로 성공률을 올리는 것이 R-003 이 든 실패다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Wp9jNbePAmWc5jQwCYhK9v
345 lines
16 KiB
Python
345 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""파이프라인 런 원장이 절차를 지켰는지 본다.
|
|
|
|
이 검사기는 글의 품질을 보지 않는다. **절차의 준수**를 본다 — 단계가 빠졌는지, 그 단계가
|
|
자기 스킬을 실제로 열었는지, 관문이 돌았고 종료 코드가 0 이었는지, 적어 낸 산출물이
|
|
디스크에 있는지.
|
|
|
|
python3 scripts/verify-pipeline-run.py --init runs/<프로젝트>/<runId>/run.json \\
|
|
--project <프로젝트> --record <기록 경로>
|
|
python3 scripts/verify-pipeline-run.py runs/<프로젝트>/<runId>/run.json
|
|
|
|
**`skillEcho` 가 이 검사기의 핵심이다.** 단계마다 그 SKILL.md 에서 한 줄을 원문 그대로
|
|
옮겨 오게 하고, 그 문자열이 실제로 그 파일 안에 있는지 대조한다. 스킬을 안 읽고 결과만
|
|
그럴듯하게 낸 단계는 여기서 걸린다.
|
|
|
|
계약은 `.agents/skills/running-tech-log-pipeline/references/stage-contracts.md` 다.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from techlog import Report # noqa: E402
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
SKILLS = os.path.join(ROOT, ".agents", "skills")
|
|
TEMPLATE = os.path.join(SKILLS, "running-tech-log-pipeline", "templates", "run.json")
|
|
|
|
STATUSES = ("PENDING", "RUNNING", "DONE", "SKIPPED", "FAILED")
|
|
|
|
# 단계마다 어떤 스킬이 맡고, 관문에 어떤 명령이 있어야 하는가.
|
|
# 관문은 명령 문자열에 이 토큰이 들어 있는지로 본다 — 호출형이 조금씩 달라도 같은 검사다.
|
|
STAGES = {
|
|
"S1": {"skill": "analyzing-codebase-for-tech-log",
|
|
"gates": ["verify-project-layout.py"], "skippable": True},
|
|
"S2": {"skill": "deriving-tech-log-root-tree",
|
|
"gates": ["build-tech-log-tree.py", "verify-tech-log-tree.py"], "skippable": True},
|
|
"S3": {"skill": "writing-tech-log-records",
|
|
"gates": ["check_body.mjs", "check_prose.mjs", "check_evidence.mjs"],
|
|
"skippable": False},
|
|
"S4": {"skill": "technical-visualizer",
|
|
"gates": ["lint", "check-figure-text.py", "check-figure-overlap.py",
|
|
"preview-figure.py"],
|
|
"skippable": True},
|
|
# S5·S6 은 문장을 고친 뒤라 S3 관문을 다시 돈다 (stage-contracts.md 「관문 요약」)
|
|
"S5": {"skill": "rewriting-technical-prose-naturally",
|
|
"gates": ["check_prose.mjs", "style_profile.mjs", "check_body.mjs",
|
|
"check_evidence.mjs"],
|
|
"skippable": False},
|
|
"S6": {"skill": "writing-as-the-person-who-did-it",
|
|
"gates": ["check_voice.mjs", "check_prose.mjs", "check_body.mjs",
|
|
"check_evidence.mjs"],
|
|
"skippable": False},
|
|
"S7": {"skill": "publishing-tech-log-to-studio",
|
|
"gates": ["저장됨", "build-tech-log-tree.py", "verify-tech-log-tree.py"],
|
|
"skippable": True},
|
|
}
|
|
|
|
# 측정 관문 — 돌았다는 것은 요구하지만 종료 코드 0 은 요구하지 않는다.
|
|
# 문서 계약이 「error 0」을 붙인 것은 check_prose 뿐이고 style_profile 은 문체 수치를 보여 주는
|
|
# 측정이다 (stage-contracts.md:178·:252). 여기에 0 을 요구하면 정직하게 적은 원장이 실패하고,
|
|
# 0 으로 고쳐 적으면 그건 지어낸 것이 된다.
|
|
MEASUREMENT_GATES = ("style_profile.mjs",)
|
|
|
|
ORDER = ["S1", "S2", "S3", "S4", "S5", "S6", "S7"]
|
|
|
|
|
|
def _known_projects() -> set[str]:
|
|
docs = os.path.join(ROOT, "docs")
|
|
if not os.path.isdir(docs):
|
|
return set()
|
|
return {d for d in os.listdir(docs)
|
|
if os.path.isdir(os.path.join(docs, d)) and not d.startswith(("_", "."))}
|
|
|
|
|
|
def _wrong_target(cmd: str, project: str, known: set[str]) -> str | None:
|
|
"""이 관문이 **다른 프로젝트**에 돌지 않았는지 본다 (R5).
|
|
|
|
원장은 관문의 명령 문자열과 종료 코드를 적는다. 그런데 그 명령이 어느 대상에 돌았는지는
|
|
아무도 안 봤다 — 다른 프로젝트에 돌려 exit 0 을 받아 적어도 통과로 셌다.
|
|
`docs/<이름>/` 경로와 명령 인자로 오는 프로젝트 이름 둘 다 본다.
|
|
|
|
**유효 범위 — 「다른 프로젝트」만 본다.** 같은 프로젝트 안의 **다른 기록**에 돌린 관문은
|
|
안 잡는다. `docs/<프로젝트>/` 가 같으면 통과한다. 기록 단위까지 보려면 원장의 `record` 와
|
|
대조해야 하고, 그러면 프로젝트 단위로 도는 관문(`verify-tech-log-tree.py <프로젝트>`)이
|
|
전부 걸리므로 관문마다 대상 단위를 따로 적어야 한다. 그것은 이 검사기가 아니라 계약이
|
|
먼저 정할 일이다.
|
|
"""
|
|
for m in re.finditer(r"docs/([A-Za-z0-9_.+-]+)/", cmd):
|
|
if m.group(1) != project:
|
|
return f"docs/{m.group(1)}/"
|
|
for other in known:
|
|
if other == project:
|
|
continue
|
|
if re.search(rf"(?<![\w/.-]){re.escape(other)}(?![\w/.-])", cmd):
|
|
return other
|
|
return None
|
|
|
|
|
|
def _norm(text: str) -> str:
|
|
"""공백을 하나로 접는다. 인용을 줄바꿈까지 똑같이 옮기라고 요구하지 않는다."""
|
|
return re.sub(r"\s+", " ", text).strip()
|
|
|
|
|
|
def _skill_text(skill: str) -> str | None:
|
|
"""스킬 폴더 전체(SKILL.md 와 references)의 글자. 영수증을 여기서 찾는다."""
|
|
base = os.path.join(SKILLS, skill)
|
|
if not os.path.isdir(base):
|
|
return None
|
|
out = []
|
|
for dirpath, _, names in os.walk(base):
|
|
for name in sorted(names):
|
|
if name.endswith(".md"):
|
|
try:
|
|
out.append(open(os.path.join(dirpath, name), encoding="utf-8").read())
|
|
except OSError:
|
|
pass
|
|
return _norm("\n".join(out))
|
|
|
|
|
|
def init(path: str, project: str, record: str, run_id: str | None) -> int:
|
|
if os.path.exists(path):
|
|
print(f"이미 있다: {path}", file=sys.stderr)
|
|
return 1
|
|
run = json.load(open(TEMPLATE, encoding="utf-8"))
|
|
now = dt.datetime.now()
|
|
run["runId"] = run_id or now.strftime("%Y-%m-%d-%H%M")
|
|
run["project"] = project
|
|
run["record"] = record
|
|
run["startedAt"] = now.astimezone().isoformat(timespec="seconds")
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
json.dump(run, fh, ensure_ascii=False, indent=2)
|
|
fh.write("\n")
|
|
print(f"런을 열었다: {path} (runId={run['runId']} · project={project})")
|
|
return 0
|
|
|
|
|
|
def _side_proof(rep: Report, st: dict, sid: str, spec: dict, where: str) -> None:
|
|
"""건너뛴 단계의 곁증명(`sideProof`)을 본 단계와 같은 잣대로 검사한다.
|
|
|
|
`outputs` 가 가리키는 `*stage-report.json` 중 `"stage"` 가 이 단계인 것을 곁증명으로
|
|
본다. 곁증명이 없는 것은 정상이다 — 있는데 엉터리인 것만 잡는다.
|
|
"""
|
|
for out in st.get("outputs") or []:
|
|
if not out.endswith(".json"):
|
|
continue
|
|
full = os.path.join(ROOT, out)
|
|
if not os.path.exists(full):
|
|
continue
|
|
try:
|
|
proof = json.load(open(full, encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
rep.error("곁증명을 읽지 못했다", f"{where} — {out}")
|
|
continue
|
|
stage_of = str(proof.get("stage") or "")
|
|
if stage_of != sid and not stage_of.startswith(sid + "-"):
|
|
continue
|
|
if proof.get("skill") != spec["skill"]:
|
|
rep.error("곁증명이 다른 스킬을 썼다",
|
|
f"{where} — {proof.get('skill')!r} · 계약은 {spec['skill']!r}")
|
|
echo = _norm(proof.get("skillEcho") or "")
|
|
text = _skill_text(spec["skill"])
|
|
if not echo:
|
|
rep.error("곁증명에 스킬 영수증이 없다", f"{where} — {out}")
|
|
elif text is not None and echo not in text:
|
|
rep.error("곁증명의 스킬 영수증이 그 스킬의 문장이 아니다",
|
|
f"{where} — {echo[:60]}…")
|
|
gates = proof.get("gates") or []
|
|
if not gates:
|
|
rep.error("곁증명에 관문이 없다", f"{where} — {out}")
|
|
for g in gates:
|
|
cmd = str(g.get("cmd") or "")
|
|
if any(tok in cmd for tok in MEASUREMENT_GATES):
|
|
if g.get("exit") is None:
|
|
rep.error("곁증명의 측정 관문에 종료 코드가 없다", f"{where} — {cmd[:70]}")
|
|
continue
|
|
if g.get("exit") not in (0, "0"):
|
|
rep.error("곁증명의 관문이 통과하지 못했다",
|
|
f"{where} — {cmd[:70]} → exit {g.get('exit')}")
|
|
rep.facts.setdefault("곁증명", []).append(f"{sid}:{os.path.basename(out)}")
|
|
|
|
|
|
def verify(path: str) -> Report:
|
|
rel = os.path.relpath(path, ROOT)
|
|
rep = Report(rel)
|
|
try:
|
|
run = json.load(open(path, encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
rep.error("원장을 읽지 못했다", f"{rel} — {exc}")
|
|
return rep
|
|
|
|
for key in ("project", "record", "stages"):
|
|
if not run.get(key):
|
|
rep.error("원장에 칸이 없다", key)
|
|
project = run.get("project") or ""
|
|
rep.facts["project"] = project or "—"
|
|
known_projects = _known_projects()
|
|
|
|
stages = {s.get("id"): s for s in run.get("stages") or []}
|
|
missing = [sid for sid in ORDER if sid not in stages]
|
|
if missing:
|
|
rep.error("단계가 원장에 없다", " · ".join(missing))
|
|
|
|
counts: dict[str, int] = {}
|
|
for sid in ORDER:
|
|
st = stages.get(sid)
|
|
if st is None:
|
|
continue
|
|
spec = STAGES[sid]
|
|
status = st.get("status") or "PENDING"
|
|
counts[status] = counts.get(status, 0) + 1
|
|
where = f"{sid} {st.get('name') or ''}".strip()
|
|
|
|
if status not in STATUSES:
|
|
rep.error("status 값이 계약 밖이다", f"{where} — {status}")
|
|
continue
|
|
if st.get("skill") != spec["skill"]:
|
|
rep.error("단계가 다른 스킬을 썼다",
|
|
f"{where} — {st.get('skill')!r} · 계약은 {spec['skill']!r}")
|
|
if st.get("runBy") != "subagent":
|
|
rep.warn("서브에이전트가 아닌 것으로 적혀 있다",
|
|
f"{where} — runBy={st.get('runBy')!r}")
|
|
|
|
if status in ("PENDING", "RUNNING"):
|
|
rep.error("끝나지 않은 단계가 있다", f"{where} — {status}")
|
|
continue
|
|
if status == "FAILED":
|
|
rep.error("단계가 실패했다", f"{where} — {st.get('notes') or '사유 없음'}")
|
|
continue
|
|
if status == "SKIPPED":
|
|
if not spec["skippable"]:
|
|
rep.error("건너뛸 수 없는 단계를 건너뛰었다", where)
|
|
elif not (st.get("skipReason") or "").strip():
|
|
rep.error("건너뛴 사유가 없다",
|
|
f"{where} — 판단해서 건너뛴 것과 빠뜨린 것을 구분해야 한다")
|
|
# 이 기록에서는 건너뛰었지만 그 단계가 도는지 따로 증명했으면 그것도 검사한다.
|
|
# 안 그러면 곁증명은 아무도 읽지 않는 파일이 된다
|
|
_side_proof(rep, st, sid, spec, where)
|
|
continue
|
|
|
|
# ── 여기부터 DONE ────────────────────────────────────────────
|
|
echo = _norm(st.get("skillEcho") or "")
|
|
if not echo:
|
|
rep.error("스킬 영수증이 없다",
|
|
f"{where} — SKILL.md 를 열었다는 증거가 원장에 없다")
|
|
else:
|
|
text = _skill_text(spec["skill"])
|
|
if text is None:
|
|
rep.error("스킬 폴더가 없다", f"{where} — {spec['skill']}")
|
|
elif echo not in text:
|
|
rep.error("스킬 영수증이 그 스킬의 문장이 아니다",
|
|
f"{where} — {echo[:60]}…")
|
|
elif len(echo) < 20:
|
|
rep.warn("스킬 영수증이 너무 짧다", f"{where} — {echo}")
|
|
|
|
gates = st.get("gates") or []
|
|
cmds = " ; ".join(str(g.get("cmd") or "") for g in gates)
|
|
for g in gates:
|
|
other = _wrong_target(str(g.get("cmd") or ""), project, known_projects)
|
|
if other:
|
|
rep.error("관문이 다른 대상에 돌았다",
|
|
f"{where} — {str(g.get('cmd'))[:60]} → {other} (원장은 {project})")
|
|
for token in spec["gates"]:
|
|
if token not in cmds:
|
|
rep.error("관문이 빠졌다", f"{where} — {token}")
|
|
for g in gates:
|
|
cmd = str(g.get("cmd") or "")
|
|
if any(tok in cmd for tok in MEASUREMENT_GATES):
|
|
# 측정 관문 — 돌았는지만 본다. exit 칸이 아예 없으면 안 돌린 것이다
|
|
if g.get("exit") is None:
|
|
rep.error("측정 관문의 종료 코드가 없다",
|
|
f"{where} — {cmd[:70]} — 돌렸다는 기록이 없다")
|
|
continue
|
|
if g.get("exit") not in (0, "0"):
|
|
rep.error("관문이 통과하지 못했다",
|
|
f"{where} — {cmd[:70]} → exit {g.get('exit')}")
|
|
|
|
for out in st.get("outputs") or []:
|
|
if not os.path.exists(os.path.join(ROOT, out)):
|
|
rep.error("적어 낸 산출물이 디스크에 없다", f"{where} — {out}")
|
|
# 한 단계를 두 번 돌렸으면 두 번째 것도 같은 잣대로 본다
|
|
_side_proof(rep, st, sid, spec, where)
|
|
|
|
rep.facts["stages"] = counts
|
|
record = run.get("record")
|
|
if record and not os.path.exists(os.path.join(ROOT, record)):
|
|
rep.error("런이 만든다는 기록이 없다", record)
|
|
return rep
|
|
|
|
|
|
def render(rep: Report, samples: int) -> None:
|
|
facts = " · ".join(
|
|
f"{k}={json.dumps(v, ensure_ascii=False) if isinstance(v, dict) else v}"
|
|
for k, v in rep.facts.items())
|
|
print(f" [{rep.project}] {facts or '—'}")
|
|
for label, bucket, mark in (("error", rep.errors, "✗"), ("warn", rep.warns, "!")):
|
|
for rule, details in sorted(bucket.items(), key=lambda kv: -len(kv[1])):
|
|
print(f" {mark} {label} {len(details):>4} {rule}")
|
|
for d in details[:samples]:
|
|
if d:
|
|
print(f" · {d}")
|
|
if samples and len(details) > samples:
|
|
print(f" … 외 {len(details) - samples}건")
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="파이프라인 런 원장이 절차를 지켰는지 본다.")
|
|
ap.add_argument("ledgers", nargs="*", help="run.json 경로")
|
|
ap.add_argument("--init", metavar="PATH", help="틀에서 런 원장을 만든다")
|
|
ap.add_argument("--project")
|
|
ap.add_argument("--record", default="")
|
|
ap.add_argument("--run-id")
|
|
ap.add_argument("--samples", type=int, default=3)
|
|
ap.add_argument("--strict", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
if args.init:
|
|
if not args.project:
|
|
print("--init 에는 --project 가 필요하다", file=sys.stderr)
|
|
return 2
|
|
return init(args.init, args.project, args.record, args.run_id)
|
|
|
|
if not args.ledgers:
|
|
print("검사할 run.json 을 달라", file=sys.stderr)
|
|
return 2
|
|
|
|
reports = [verify(p) for p in args.ledgers]
|
|
e = sum(r.error_count for r in reports)
|
|
w = sum(r.warn_count for r in reports)
|
|
print(f"PIPELINE RUN: {'FAIL' if e or (args.strict and w) else 'PASS'}"
|
|
f" — 런 {len(reports)} · error {e} · warn {w}")
|
|
for r in reports:
|
|
render(r, args.samples)
|
|
return 1 if e or (args.strict and w) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|