DongHyeonkaandClaude Opus 5
8ca822dc4d
원장을 손으로 써 왔다. 그래서 셋이 없었다. 쓰는 도중에 끊기면 반쪽 파일이 남고, 단계마다 시작·끝이 없어 「돌다 말았다」와 「아직 안 시작했다」가 PENDING 하나로 같아 보이고, 런 도중에 끼어든 일이 어디에도 안 남는다. bin/task.py 가 이 계약의 초안이다 — flock 을 잡고 임시 파일에 완성한 뒤 os.replace 로 바꾸고 startedAt 부터 지금까지를 누적에 더한다. 그 규약을 runs/ 쪽으로 옮겼다. 새로 만든 것이 아니다. 끊긴 단계를 새 세션이 다시 열면 그 구간을 닫고 잇되, 누적에만 더하지 않고 interruptions 에 따로 적는다. 닫은 구간에는 세션이 죽어 있던 시간이 섞인다 — 「이 단계가 오래 걸렸다」와 「중간에 끊겼다」는 다른 말이다. 라이더에 자리를 만들었다. 이 배치에서 검증 한 작업의 누적 24분 가운데 17분 50초가 라이더였는데 상태 파일에 그 구분이 없었다. 단계가 아니라서 어느 칸에도 안 남던 것이다. 상태 전이를 거절한다 — 이미 있는 런을 덮어쓰기, 단계 겹쳐 열기, 건너뛸 수 없는 단계 건너뛰기, 사유 없는 SKIPPED, 영수증 없는 DONE, 종료 코드 없는 관문. 더한 칸이 있어도 기존 검사기가 그대로 읽는다. 통과한 원장에 새 칸을 더해도 통과하는 것을 확인했다. 손으로 쓴 원장도 계속 유효하고 이 도구는 선택적으로 부른다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Q4vKjQo9KKBBokzxqXLCfk
164 lines
8.5 KiB
Python
164 lines
8.5 KiB
Python
"""런 원장을 쓰는 도구.
|
|
|
|
`bin/task.py` 가 이 계약의 초안이다 — flock 을 잡고 임시 파일에 완성한 뒤 원자적으로 바꾸고,
|
|
`startedAt` 부터 지금까지를 누적에 더한다. 여기서는 그것이 `runs/` 쪽에서도 성립하는지 본다.
|
|
"""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
TOOL = os.path.join(ROOT, "scripts", "run-ledger.py")
|
|
_spec = importlib.util.spec_from_file_location("run_ledger", TOOL)
|
|
rl = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(rl)
|
|
|
|
|
|
def _cli(*args, **kw):
|
|
return subprocess.run(["python3", TOOL, *args], cwd=ROOT,
|
|
capture_output=True, text=True, **kw)
|
|
|
|
|
|
class LedgerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = tempfile.mkdtemp()
|
|
self.led = os.path.join(self.dir, "runs", "demo", "2026-01-01-0000", "run.json")
|
|
p = _cli("open", self.led, "--project", "demo", "--record", "docs/demo/x.md")
|
|
self.assertEqual(0, p.returncode, p.stderr)
|
|
|
|
def _read(self):
|
|
return json.load(open(self.led, encoding="utf-8"))
|
|
|
|
def _stage(self, sid):
|
|
return next(s for s in self._read()["stages"] if s["id"] == sid)
|
|
|
|
# ── 상태를 지킨다 ────────────────────────────────────────────────
|
|
def test_opening_twice_is_refused(self):
|
|
"""이미 있는 런을 덮어쓰지 않는다. 이어서 하려면 status 로 본다."""
|
|
p = _cli("open", self.led, "--project", "demo", "--record", "x")
|
|
self.assertEqual(2, p.returncode)
|
|
|
|
def test_two_stages_cannot_be_open_at_once(self):
|
|
"""단계를 겹쳐 열지 않는다. 병렬을 꺼도 같은 상태에서 이어갈 수 있어야 한다."""
|
|
_cli("begin", self.led, "--stage", "S3")
|
|
p = _cli("begin", self.led, "--stage", "S5")
|
|
self.assertEqual(2, p.returncode)
|
|
self.assertIn("RUNNING", p.stderr)
|
|
|
|
def test_an_unskippable_stage_cannot_be_skipped(self):
|
|
_cli("begin", self.led, "--stage", "S3")
|
|
p = _cli("end", self.led, "--stage", "S3", "--status", "SKIPPED", "--why", "x")
|
|
self.assertEqual(2, p.returncode)
|
|
|
|
def test_a_skipped_stage_needs_a_reason(self):
|
|
"""적지 않고 빠뜨린 것과 판단해서 건너뛴 것을 구분해야 한다."""
|
|
_cli("begin", self.led, "--stage", "S4")
|
|
self.assertEqual(2, _cli("end", self.led, "--stage", "S4",
|
|
"--status", "SKIPPED").returncode)
|
|
self.assertEqual(0, _cli("end", self.led, "--stage", "S4",
|
|
"--status", "SKIPPED", "--why", "그림이 필요 없다").returncode)
|
|
|
|
def test_a_done_stage_needs_a_skill_receipt(self):
|
|
"""안 연 스킬의 영수증을 적으면 그것이 지어낸 것이다. 빈 채로 DONE 을 못 찍는다."""
|
|
_cli("begin", self.led, "--stage", "S3")
|
|
self.assertEqual(2, _cli("end", self.led, "--stage", "S3",
|
|
"--status", "DONE").returncode)
|
|
self.assertEqual(0, _cli("end", self.led, "--stage", "S3", "--status", "DONE",
|
|
"--echo", "본문이 있는 종류는 Case 와 Concept 둘뿐이다.").returncode)
|
|
|
|
def test_a_gate_must_carry_an_exit_code(self):
|
|
"""종료 코드 없는 관문을 못 적는다. 돌리지 않고 적는 경로를 막는다."""
|
|
p = _cli("gate", self.led, "--stage", "S3", "--cmd", "x")
|
|
self.assertNotEqual(0, p.returncode)
|
|
|
|
# ── 끊겨도 이어진다 ──────────────────────────────────────────────
|
|
def test_the_file_is_never_half_written(self):
|
|
"""쓰는 도중에 죽여도 읽는 쪽은 이전 판이나 다음 판 중 하나를 본다."""
|
|
for i in range(40):
|
|
pr = subprocess.Popen(["python3", TOOL, "gate", self.led, "--stage", "S6",
|
|
"--cmd", f"x{i}", "--exit", "0"], cwd=ROOT,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
time.sleep(0.001 * (i % 7))
|
|
pr.kill()
|
|
pr.wait()
|
|
json.load(open(self.led, encoding="utf-8")) # 깨졌으면 여기서 터진다
|
|
|
|
def test_concurrent_writes_do_not_lose_updates(self):
|
|
"""락이 없으면 마지막에 쓴 것만 남는다."""
|
|
ps = [subprocess.Popen(["python3", TOOL, "gate", self.led, "--stage", "S6",
|
|
"--cmd", f"c{i}", "--exit", "0"], cwd=ROOT,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
for i in range(20)]
|
|
for p in ps:
|
|
p.wait()
|
|
self.assertEqual(20, len(self._stage("S6")["gates"]))
|
|
|
|
def test_an_interrupted_stage_is_closed_and_continued(self):
|
|
"""앞 세션이 열어 둔 채 끊기면, 그 구간을 닫고 잇는다.
|
|
|
|
닫은 구간에는 세션이 없던 시간이 섞이므로 누적에만 더하지 않고 따로 적는다 —
|
|
「이 단계가 오래 걸렸다」와 「중간에 끊겼다」는 다른 말이다.
|
|
"""
|
|
_cli("begin", self.led, "--stage", "S5")
|
|
d = self._read()
|
|
import datetime
|
|
s5 = next(s for s in d["stages"] if s["id"] == "S5")
|
|
s5["startedAt"] = (datetime.datetime.now().astimezone()
|
|
- datetime.timedelta(seconds=90)).isoformat(timespec="seconds")
|
|
json.dump(d, open(self.led, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
|
|
|
|
p = _cli("begin", self.led, "--stage", "S5")
|
|
self.assertEqual(0, p.returncode, p.stderr)
|
|
after = self._stage("S5")
|
|
self.assertGreaterEqual(after["elapsedSeconds"], 89)
|
|
self.assertEqual(1, len(after["interruptions"]))
|
|
self.assertIn("세션이 없던 시간", after["interruptions"][0]["note"])
|
|
|
|
def test_status_says_where_it_stopped(self):
|
|
"""새 세션이 원장만 읽고 어디서 이어야 하는지 알 수 있어야 한다."""
|
|
_cli("begin", self.led, "--stage", "S3")
|
|
p = _cli("status", self.led)
|
|
self.assertEqual(1, p.returncode, "끊긴 자리가 있으면 0 이 아니다")
|
|
self.assertIn("S3", p.stdout)
|
|
self.assertIn("끊긴 자리", p.stdout)
|
|
|
|
# ── 단계 밖의 시간 ───────────────────────────────────────────────
|
|
def test_a_rider_is_recorded_outside_the_stages(self):
|
|
"""런 도중에 끼어든 일은 단계가 아니라서 어느 칸에도 안 남았다."""
|
|
_cli("rider", self.led, "--id", "R11", "--why", "마스킹 되돌이", "--seconds", "1070")
|
|
riders = self._read()["riders"]
|
|
self.assertEqual(["R11"], [r["id"] for r in riders])
|
|
self.assertEqual(1070, riders[0]["seconds"])
|
|
self.assertIn("라이더", _cli("status", self.led).stdout)
|
|
|
|
def test_a_rider_records_which_stage_it_interrupted(self):
|
|
_cli("begin", self.led, "--stage", "S3")
|
|
_cli("rider", self.led, "--id", "R12", "--why", "x", "--seconds", "60")
|
|
self.assertEqual("S3", self._read()["riders"][0]["duringStage"])
|
|
|
|
# ── 옛 원장을 깨지 않는다 ─────────────────────────────────────────
|
|
def test_the_added_fields_do_not_break_the_verifier(self):
|
|
"""이 도구가 더한 칸이 있어도 검사기가 그대로 읽어야 한다."""
|
|
real = os.path.join(ROOT, "runs/document-haness/2026-09-10-1033/run.json")
|
|
if not os.path.exists(real):
|
|
self.skipTest("견줄 실제 원장이 없다")
|
|
d = json.load(open(real, encoding="utf-8"))
|
|
d.update({"revision": 1, "riders": [], "sessions": [], "updatedAt": "x"})
|
|
for st in d["stages"]:
|
|
st.update({"startedAt": None, "finishedAt": None, "elapsedSeconds": 0})
|
|
out = os.path.join(self.dir, "runs", "document-haness", "2026-09-10-1033")
|
|
os.makedirs(out, exist_ok=True)
|
|
copy = os.path.join(out, "run.json")
|
|
json.dump(d, open(copy, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
|
|
p = subprocess.run(["python3", os.path.join(ROOT, "scripts", "verify-pipeline-run.py"),
|
|
copy], cwd=ROOT, capture_output=True, text=True)
|
|
self.assertEqual(0, p.returncode, p.stdout + p.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|