DongHyeonkaandClaude Opus 5
96d7fbc57a
이어받은 단계에 앞 세션이 관문을 써도 그대로 들어갔다. 원자성 문제가 아니다 — 파일은 온전한 채로 두 세션의 기록이 섞인다. 그리고 관문마다 누가 적었는지가 없어서 나중에 원장을 읽어도 가릴 수 없었다. begin 이 세대를 올리고 주인을 적는다. 낮은 세대나 다른 주인의 쓰기는 거절한다. bin/task.py 의 attempt 와 같은 자리다. gate 와 end 도 --session 을 받아 적는다. 관문마다 session·generation·at 이 남고 end 는 finishedBy 를 남긴다. 주인이 없는 단계는 그대로 받는다. 세션을 안 쓰는 단일 세션 사용이 깨지지 않는다. 회귀에 그 대조를 넣었다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Q4vKjQo9KKBBokzxqXLCfk
208 lines
11 KiB
Python
208 lines
11 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_a_superseded_session_cannot_write(self):
|
|
"""파일이 안 깨지는 것과 두 세션의 기록이 안 섞이는 것은 다른 일이다.
|
|
|
|
`plan/02` B-005 — 「이전 작업자의 종료를 확인하고 재배정한 시도 번호와 맞는
|
|
결과만 받는다.」 `bin/task.py` 의 attempt 와 같은 자리다.
|
|
"""
|
|
_cli("begin", self.led, "--stage", "S3", "--session", "sess-1")
|
|
_cli("begin", self.led, "--stage", "S3", "--session", "sess-2")
|
|
for args in (("gate", self.led, "--stage", "S3", "--cmd", "stale", "--exit", "0"),
|
|
("end", self.led, "--stage", "S3", "--status", "DONE", "--echo", "x")):
|
|
with self.subTest(cmd=args[0]):
|
|
p = _cli(*args, "--session", "sess-1")
|
|
self.assertEqual(2, p.returncode)
|
|
self.assertIn("sess-2", p.stderr)
|
|
|
|
def test_a_stale_generation_is_refused(self):
|
|
_cli("begin", self.led, "--stage", "S3", "--session", "sess-1")
|
|
_cli("begin", self.led, "--stage", "S3", "--session", "sess-2")
|
|
p = _cli("gate", self.led, "--stage", "S3", "--cmd", "x", "--exit", "0",
|
|
"--session", "sess-2", "--generation", "1")
|
|
self.assertEqual(2, p.returncode)
|
|
|
|
def test_an_anonymous_write_to_an_owned_stage_is_refused(self):
|
|
"""누가 썼는지 안 밝히면 나중에 원장을 읽을 수 없다."""
|
|
_cli("begin", self.led, "--stage", "S3", "--session", "sess-1")
|
|
p = _cli("gate", self.led, "--stage", "S3", "--cmd", "x", "--exit", "0")
|
|
self.assertEqual(2, p.returncode)
|
|
|
|
def test_the_current_owner_writes_and_is_recorded(self):
|
|
"""대조군. 지금 주인의 쓰기는 들어가고, 누가 어느 세대에 썼는지 남는다."""
|
|
_cli("begin", self.led, "--stage", "S3", "--session", "sess-1")
|
|
p = _cli("gate", self.led, "--stage", "S3", "--cmd", "ok", "--exit", "0",
|
|
"--session", "sess-1", "--generation", "1")
|
|
self.assertEqual(0, p.returncode, p.stderr)
|
|
gate = self._stage("S3")["gates"][0]
|
|
self.assertEqual(("sess-1", 1), (gate["session"], gate["generation"]))
|
|
|
|
def test_an_unowned_stage_still_accepts_writes(self):
|
|
"""주인이 없으면 (세션을 안 쓰는 단일 세션 사용) 그대로 쓴다. 기존 쓰임을 안 깬다."""
|
|
_cli("begin", self.led, "--stage", "S3")
|
|
self.assertEqual(0, _cli("gate", self.led, "--stage", "S3",
|
|
"--cmd", "x", "--exit", "0").returncode)
|
|
|
|
# ── 옛 원장을 깨지 않는다 ─────────────────────────────────────────
|
|
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()
|