diff --git a/scripts/run-ledger.py b/scripts/run-ledger.py
new file mode 100644
index 0000000..fadd97f
--- /dev/null
+++ b/scripts/run-ledger.py
@@ -0,0 +1,328 @@
+#!/usr/bin/env python3
+"""런 원장을 쓰는 도구. 반쪽 상태를 만들지 않고, 끊긴 자리에서 다시 시작한다.
+
+`runs/<프로젝트>/<runId>/run.json` 은 지금까지 손으로 썼다. 그래서 셋이 없었다.
+
+- **반쪽 파일.** `json.dump` 로 바로 쓰면 중간에 끊긴 원장이 남는다. 다음 세션이 그것을
+  읽으면 런이 통째로 사라진 것처럼 보인다
+- **어디서 끊겼는지.** 단계마다 시작·끝이 없어 「S5 를 돌다 말았다」와 「S5 를 아직 안
+  시작했다」가 `PENDING` 하나로 같아 보인다
+- **원장 밖의 시간.** 런 도중에 끼어든 일(라이더)이 어디에도 안 남는다. 이 배치에서
+  검증 한 작업의 누적 24분 가운데 17분 50초가 라이더였는데 상태 파일에 그 구분이 없었다
+
+`bin/task.py` 가 이 계약의 초안이다 — flock 을 잡고 임시 파일에 완성한 뒤 `os.replace` 로
+바꾸고, `startedAt` 부터 지금까지를 누적에 더한다. 여기서는 그것을 `runs/` 쪽으로 넓힌다.
+**새로 만드는 것이 아니라 같은 규약을 옮기는 것이다.**
+
+    python3 scripts/run-ledger.py open   <원장> --project P --record R
+    python3 scripts/run-ledger.py begin  <원장> --stage S3 --runby subagent
+    python3 scripts/run-ledger.py gate   <원장> --stage S3 --cmd "..." --exit 0
+    python3 scripts/run-ledger.py end    <원장> --stage S3 --status DONE --echo "..." --output ...
+    python3 scripts/run-ledger.py rider  <원장> --id R11 --why "..." --seconds 1070
+    python3 scripts/run-ledger.py status <원장>
+"""
+from __future__ import annotations
+
+import argparse
+import datetime
+import fcntl
+import json
+import os
+import sys
+import tempfile
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TEMPLATE = os.path.join(
+    ROOT, ".agents/skills/running-tech-log-pipeline/templates/run.json")
+STAGES = ["S1", "S2", "S3", "S4", "S5", "S6", "S7"]
+UNSKIPPABLE = {"S3", "S5", "S6"}
+OPEN_STATES = {"RUNNING"}
+
+
+class Contract(Exception):
+    """계약 위반. 도구가 거절한 것이지 실패가 아니다."""
+
+
+def _now() -> str:
+    return datetime.datetime.now().astimezone().isoformat(timespec="seconds")
+
+
+class Ledger:
+    """flock 을 잡고 읽어서, 임시 파일에 완성한 뒤 원자적으로 바꾼다.
+
+    `os.replace` 는 같은 파일 시스템에서 원자적이다. 중간에 끊겨도 읽는 쪽은 **이전 판이나
+    다음 판 중 하나**를 본다. 반쪽 파일을 볼 수 없다.
+    """
+
+    def __init__(self, path: str, create: bool = False) -> None:
+        self.path = os.path.abspath(path)
+        if not create and not os.path.exists(self.path):
+            raise Contract(f"그런 원장이 없다: {path}")
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        self._lockpath = self.path + ".lock"
+        self._fh = open(self._lockpath, "a+")
+
+    def __enter__(self) -> "Ledger":
+        fcntl.flock(self._fh, fcntl.LOCK_EX)
+        self.data = (json.load(open(self.path, encoding="utf-8"))
+                     if os.path.exists(self.path) else {})
+        return self
+
+    def __exit__(self, *exc) -> None:
+        fcntl.flock(self._fh, fcntl.LOCK_UN)
+        self._fh.close()
+
+    def save(self) -> None:
+        self.data["revision"] = int(self.data.get("revision", 0)) + 1
+        self.data["updatedAt"] = _now()
+        fd, tmp = tempfile.mkstemp(dir=os.path.dirname(self.path),
+                                   prefix=".run.", suffix=".json")
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as fh:
+                json.dump(self.data, fh, ensure_ascii=False, indent=2)
+                fh.write("\n")
+                fh.flush()
+                os.fsync(fh.fileno())
+            os.replace(tmp, self.path)
+        except BaseException:
+            os.path.exists(tmp) and os.unlink(tmp)
+            raise
+
+    def stage(self, sid: str) -> dict:
+        for st in self.data.get("stages", []):
+            if st["id"] == sid:
+                return st
+        raise Contract(f"그런 단계가 없다: {sid}. 단계는 {', '.join(STAGES)} 뿐이다")
+
+    def running(self) -> dict | None:
+        return next((s for s in self.data.get("stages", [])
+                     if s.get("status") in OPEN_STATES), None)
+
+
+def _require_owner(st: dict, session: str | None, generation: int | None) -> None:
+    """이 단계를 지금 쥔 세션인지 본다.
+
+    `plan/02` B-005 — 「이전 작업자의 종료를 확인하고 재배정한 시도 번호와 맞는 결과만
+    받는다.」 원장에는 그것이 없어서, 이어받은 단계에 앞 세션이 관문을 써도 들어갔다.
+    **파일은 안 깨진다. 깨지지 않은 채로 두 세션의 기록이 섞인다.**
+    """
+    cur_gen = int(st.get("generation", 0))
+    if generation is not None and generation != cur_gen:
+        raise Contract(
+            f"{st['id']} 의 현재 세대는 {cur_gen} 인데 쓰기는 {generation} 이다. "
+            f"그 사이에 다른 세션이 이어받았다 — begin 으로 다시 잡는다")
+    owner = st.get("owner")
+    if owner and session and session != owner:
+        raise Contract(
+            f"{st['id']} 은(는) 지금 {owner} 가 쥐고 있다. {session} 의 쓰기는 받지 않는다 — "
+            f"이어받으려면 begin 으로 세대를 올린다")
+    if owner and not session:
+        raise Contract(
+            f"{st['id']} 은(는) {owner} 가 쥐고 있다. --session 으로 누가 쓰는지 밝힌다")
+
+
+def _accrue(node: dict) -> int:
+    """`startedAt` 부터 지금까지를 누적에 더한다. 세션이 다시 떠도 이어진다."""
+    started = node.get("startedAt")
+    if started:
+        delta = (datetime.datetime.now().astimezone()
+                 - datetime.datetime.fromisoformat(started)).total_seconds()
+        node["elapsedSeconds"] = int(node.get("elapsedSeconds", 0) + max(0.0, delta))
+        node["startedAt"] = None
+    return int(node.get("elapsedSeconds", 0))
+
+
+def cmd_open(args) -> int:
+    if os.path.exists(os.path.abspath(args.ledger)) and not args.force:
+        raise Contract(f"이미 있는 원장이다: {args.ledger}. 이어서 하려면 status 로 본다")
+    with Ledger(args.ledger, create=True) as led:
+        led.data = json.load(open(TEMPLATE, encoding="utf-8"))
+        led.data.update({
+            "runId": args.run_id or os.path.basename(os.path.dirname(
+                os.path.abspath(args.ledger))),
+            "project": args.project, "record": args.record,
+            "startedAt": _now(), "finishedAt": None,
+            "riders": [],
+            "sessions": [{"session": args.session, "openedAt": _now()}],
+        })
+        for st in led.data["stages"]:
+            st.update({"startedAt": None, "finishedAt": None, "elapsedSeconds": 0})
+        led.save()
+        print(f"런을 열었다: {args.ledger} (runId={led.data['runId']})")
+    return 0
+
+
+def cmd_begin(args) -> int:
+    with Ledger(args.ledger) as led:
+        open_stage = led.running()
+        if open_stage and open_stage["id"] != args.stage:
+            raise Contract(
+                f"{open_stage['id']} 이(가) 아직 RUNNING 이다. 단계를 겹쳐 열지 않는다 — "
+                f"end 로 닫거나 status 로 어디서 끊겼는지 본다")
+        st = led.stage(args.stage)
+        if st.get("status") in ("DONE", "SKIPPED"):
+            raise Contract(f"{args.stage} 은(는) 이미 {st['status']} 다")
+        if st.get("startedAt"):
+            # 앞 세션이 이 단계를 열어 둔 채 끊겼다. 그 구간을 닫고 새로 연다.
+            # **닫은 구간에는 세션이 죽어 있던 시간이 섞인다.** 그래서 누적에만 더하지 않고
+            # 따로 적는다 — 「이 단계가 오래 걸렸다」와 「중간에 끊겼다」는 다른 말이다
+            was = st["startedAt"]
+            span = _accrue(st)
+            st.setdefault("interruptions", []).append({
+                "openedAt": was, "resumedAt": _now(),
+                "accruedSeconds": span,
+                "note": "앞 세션이 닫지 않고 끊겼다. 이 구간에는 세션이 없던 시간이 섞여 있다",
+            })
+            print(f"{args.stage} 이(가) 열린 채였다 — 그 구간을 닫고 잇는다")
+        st["status"] = "RUNNING"
+        st["runBy"] = args.runby
+        st["startedAt"] = _now()
+        # 이어받을 때마다 세대를 올리고 주인을 적는다. `bin/task.py` 의 attempt 와 같은 자리다 —
+        # 앞 세션이 아직 살아 있어도 낮은 세대의 쓰기는 이 단계에 못 들어온다.
+        # 파일이 안 깨지는 것과 두 세션의 기록이 안 섞이는 것은 다른 일이다
+        st["generation"] = int(st.get("generation", 0)) + 1
+        st["owner"] = args.session or None
+        led.data.setdefault("sessions", []).append(
+            {"session": args.session or None, "stage": args.stage,
+             "generation": st["generation"], "beganAt": _now()})
+        led.save()
+        print(f"{args.stage} RUNNING · 세대 {st['generation']} · 주인 {st['owner']}"
+              f" · 누적 {st.get('elapsedSeconds', 0)}초")
+    return 0
+
+
+def cmd_gate(args) -> int:
+    with Ledger(args.ledger) as led:
+        st = led.stage(args.stage)
+        _require_owner(st, args.session, args.generation)
+        st.setdefault("gates", []).append({
+            "cmd": args.cmd, "exit": args.exit_code,
+            "session": args.session or None, "generation": int(st.get("generation", 0)),
+            "at": _now()})
+        led.save()
+        print(f"{args.stage} 관문 {len(st['gates'])}개 · 방금 것 exit={args.exit_code}"
+              f" · {args.session or '세션 미기재'}")
+    return 0
+
+
+def cmd_end(args) -> int:
+    with Ledger(args.ledger) as led:
+        st = led.stage(args.stage)
+        _require_owner(st, args.session, args.generation)
+        if args.status == "SKIPPED" and args.stage in UNSKIPPABLE:
+            raise Contract(f"{args.stage} 은(는) 건너뛸 수 없다")
+        if args.status == "SKIPPED" and not args.why:
+            raise Contract("건너뛴 단계는 사유를 적는다. --why 를 준다")
+        if args.status == "DONE" and not (args.echo or st.get("skillEcho")):
+            raise Contract(
+                "DONE 인 단계는 스킬 영수증을 적는다. --echo 로 그 SKILL.md 의 문장을 "
+                "원문 그대로 준다 — 안 연 스킬의 영수증을 적으면 그것이 지어낸 것이다")
+        used = _accrue(st)
+        st["status"] = args.status
+        st["finishedAt"] = _now()
+        st["finishedBy"] = args.session or None
+        if args.echo:
+            st["skillEcho"] = args.echo
+        if args.why:
+            st["skipReason"] = args.why
+        for key, values in (("inputs", args.input), ("outputs", args.output)):
+            if values:
+                st.setdefault(key, []).extend(values)
+        if args.note:
+            st["notes"] = (st.get("notes", "") + " " + args.note).strip()
+        if all(s.get("status") in ("DONE", "SKIPPED") for s in led.data["stages"]):
+            led.data["finishedAt"] = _now()
+        led.save()
+        print(f"{args.stage} {args.status} · 이 단계 누적 {used}초")
+    return 0
+
+
+def cmd_rider(args) -> int:
+    """런 도중에 끼어든 일. 단계가 아니라서 어느 칸에도 안 남던 것이다."""
+    with Ledger(args.ledger) as led:
+        led.data.setdefault("riders", []).append({
+            "id": args.id, "why": args.why, "seconds": args.seconds,
+            "addedAt": _now(), "duringStage": (led.running() or {}).get("id"),
+        })
+        total = sum(r.get("seconds") or 0 for r in led.data["riders"])
+        led.save()
+        print(f"라이더 {len(led.data['riders'])}건 · 합계 {total}초 ({total // 60}분)")
+    return 0
+
+
+def cmd_status(args) -> int:
+    with Ledger(args.ledger) as led:
+        d = led.data
+        print(f"{d.get('runId')} · {d.get('project')} · revision {d.get('revision', 0)}")
+        print(f"기록  {d.get('record')}")
+        for st in d.get("stages", []):
+            mark = {"DONE": "✓", "SKIPPED": "–", "RUNNING": "▶", "FAILED": "✗"}.get(
+                st.get("status"), " ")
+            gates = st.get("gates") or []
+            bad = [g for g in gates if g.get("exit") not in (0, "0")]
+            print(f"  {mark} {st['id']} {st.get('status'):<8} "
+                  f"{st.get('elapsedSeconds', 0):>5}초 · 관문 {len(gates)}"
+                  + (f" (exit≠0 {len(bad)})" if bad else ""))
+        riders = d.get("riders") or []
+        if riders:
+            total = sum(r.get("seconds") or 0 for r in riders)
+            print(f"  라이더 {len(riders)}건 · {total}초 ({total // 60}분) — 단계 밖의 시간이다")
+            for r in riders:
+                print(f"      {r['id']} · {r.get('seconds') or 0}초 · {r.get('why', '')[:60]}")
+        run = led.running()
+        if run:
+            print(f"\n끊긴 자리: {run['id']} 이(가) RUNNING 이다 "
+                  f"(시작 {run.get('startedAt')})")
+            print(f"이어서 하려면: run-ledger.py end {args.ledger} --stage {run['id']} …")
+            return 1
+        pending = [s["id"] for s in d.get("stages", []) if s.get("status") == "PENDING"]
+        if pending:
+            print(f"\n아직 안 연 단계: {', '.join(pending)}")
+            return 1
+        print("\n일곱 단계가 다 닫혔다")
+    return 0
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="런 원장을 원자적으로 쓰고 끊긴 자리에서 잇는다.")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+
+    p = sub.add_parser("open"); p.add_argument("ledger")
+    p.add_argument("--project", required=True); p.add_argument("--record", required=True)
+    p.add_argument("--run-id"); p.add_argument("--session", default=os.environ.get("USER", ""))
+    p.add_argument("--force", action="store_true"); p.set_defaults(fn=cmd_open)
+
+    p = sub.add_parser("begin"); p.add_argument("ledger")
+    p.add_argument("--stage", required=True); p.add_argument("--runby", default="subagent")
+    p.add_argument("--session", default=""); p.set_defaults(fn=cmd_begin)
+
+    p = sub.add_parser("gate"); p.add_argument("ledger")
+    p.add_argument("--stage", required=True); p.add_argument("--cmd", required=True)
+    p.add_argument("--exit", dest="exit_code", type=int, required=True)
+    p.add_argument("--session", default=""); p.add_argument("--generation", type=int)
+    p.set_defaults(fn=cmd_gate)
+
+    p = sub.add_parser("end"); p.add_argument("ledger")
+    p.add_argument("--stage", required=True)
+    p.add_argument("--status", required=True, choices=["DONE", "SKIPPED", "FAILED"])
+    p.add_argument("--echo"); p.add_argument("--why"); p.add_argument("--note")
+    p.add_argument("--input", action="append"); p.add_argument("--output", action="append")
+    p.add_argument("--session", default=""); p.add_argument("--generation", type=int)
+    p.set_defaults(fn=cmd_end)
+
+    p = sub.add_parser("rider"); p.add_argument("ledger")
+    p.add_argument("--id", required=True); p.add_argument("--why", required=True)
+    p.add_argument("--seconds", type=int); p.set_defaults(fn=cmd_rider)
+
+    p = sub.add_parser("status"); p.add_argument("ledger"); p.set_defaults(fn=cmd_status)
+
+    args = ap.parse_args()
+    try:
+        return args.fn(args)
+    except Contract as e:
+        print(f"거절: {e}", file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/tests/test_run_ledger.py b/scripts/tests/test_run_ledger.py
new file mode 100644
index 0000000..e8b667c
--- /dev/null
+++ b/scripts/tests/test_run_ledger.py
@@ -0,0 +1,207 @@
+"""런 원장을 쓰는 도구.
+
+`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()