Batch 3a: library_run.py 支持 --work-dir（工作根可指定，默认常量不变）

_apply_work_dir() 在 argv 解析后重绑 WORK/STAGE/LEDGER/LOCKF/POLICY；--stage-dir 仍只覆盖暂存。
不传 --work-dir 时行为与以前逐位一致（控制面 docs/DESIGN.md §18 是规则出处）。
STATE.md 记录本轮。验收：控制面 scripts/acceptance_work_dir.py 真跑 copy 档 PASS 8/8，
控制面 124 用例 OK。回滚点 tag pre-batch3a。

scripts/library_run.py

@@ -99,10 +99,32 @@ def run_pinned(cmd, timeout=None, **kw):
         raise subprocess.TimeoutExpired(cmd, timeout, output=out, stderr=err)
     return subprocess.CompletedProcess(cmd, p.returncode, out, err)
 ROOT = r"Z:\Class\樊登读书\樊登读书会（每周更新）"
+# The work root: ledger.jsonl + gate_policy.json + the per-tier locks + (by default) the scratch dir.
+# `--work-dir` re-points ALL of it BEFORE a run starts (see _apply_work_dir below). The constant here
+# is the compiled-in DEFAULT, kept so a hand-run engine writes exactly where it always did: an
+# existing ledger and an existing gate_policy.json - the only record of what is parked and accepted -
+# must never be silently swapped for an empty directory. The control plane passes --work-dir
+# explicitly (Batch 3a), which is what makes `settings.work_root` authoritative instead of a number
+# the two repositories had to keep in sync.
 WORK = r"C:\AI\transcode-run"
 STAGE = os.path.join(WORK, "stage")
 LEDGER = os.path.join(WORK, "ledger.jsonl")
 LOCKF = os.path.join(WORK, "run.lock")
+
+
+def _apply_work_dir(path):
+    """Re-point WORK and everything derived from it. Called from main() AFTER argparse.
+
+    Rebound as module globals so every existing call site keeps working unchanged; the ledger, the
+    locks and the policy file are all read through those names at call time, never captured at
+    import time.
+    """
+    global WORK, STAGE, LEDGER, LOCKF, POLICY
+    WORK = os.path.abspath(path)
+    STAGE = os.path.join(WORK, "stage")
+    LEDGER = os.path.join(WORK, "ledger.jsonl")
+    LOCKF = os.path.join(WORK, "run.lock")
+    POLICY = os.path.join(WORK, "gate_policy.json")
 VIDEO_EXT = {".mkv", ".mp4", ".mov", ".avi", ".webm", ".flv", ".ts", ".m4v", ".wmv",
              ".mpg", ".mpeg", ".m2ts", ".3gp", ".ogv", ".vob", ".rmvb", ".rm", ".divx"}
 DECODE_CHECK = True
@@ -1334,7 +1356,7 @@ def collect(root):
 
 
 def main():
-    global DECODE_CHECK, STAGE
+    global DECODE_CHECK, STAGE            # WORK/STAGE/LEDGER/LOCKF/POLICY move via _apply_work_dir
     ap = argparse.ArgumentParser()
     ap.add_argument("--cq", type=int, default=CQ_NVENC, help="nvenc tier quality (measured default 30)")
     ap.add_argument("--crf", type=int, default=CRF_X265, help="x265 tier quality (measured default 24)")
@@ -1347,9 +1369,13 @@ def main():
                     help="hex CPU mask for the x265 tier (default = the E-core mask); the mask is "
                          "inherited by the ffmpeg children and must agree with --pools")
     ap.add_argument("--stage-dir", default=None,
-                    help="scratch directory for this run (default %s). The ledger, the locks and "
-                         "the policy file stay in WORK - only the disposable per-item scratch "
-                         "moves, so the dashboard can point it at a roomier drive." % STAGE)
+                    help="scratch directory for this run (default <work-dir>/stage). The ledger, the "
+                         "locks and the policy file stay in --work-dir - only the disposable "
+                         "per-item scratch moves, so the dashboard can point it at a roomier drive.")
+    ap.add_argument("--work-dir", default=None,
+                    help="work root for this run: ledger.jsonl, gate_policy.json and run-*.lock "
+                         "(default %s, this file's compiled-in constant, so a hand-run engine is "
+                         "unchanged). An explicit value moves ALL of them." % WORK)
     ap.add_argument("--tier", choices=["copy", "nvenc", "x265"],
                     help="only touch files classified into this tier (run tiers separately)")
     ap.add_argument("--force-tier", choices=["copy", "nvenc", "x265"], default=None,
@@ -1384,6 +1410,9 @@ def main():
                          "THROW THE OUTPUT AWAY - nothing is uploaded and nothing is replaced")
     args = ap.parse_args()
     DECODE_CHECK = not args.no_decode_check
+    if args.work_dir:
+        _apply_work_dir(args.work_dir)
+        log("=== work dir: %s (ledger, locks, gate_policy.json)" % WORK)
     if args.stage_dir:
         STAGE = os.path.abspath(args.stage_dir)
         log("=== stage dir: %s (ledger + locks stay in %s)" % (STAGE, WORK))