merge: 采纳 M3-1 流式可观测性+门槛缩尺+成本量化（独立复验全绿）

8 files changed, 313 insertions, 28 deletions

diff --git a/docs/overnight-status.md b/docs/overnight-status.md
index e8c2885..2896986 100644
--- a/docs/overnight-status.md
+++ b/docs/overnight-status.md
@@ -1191,3 +1191,10 @@ Updated: 2026-09-20 02:54:20 CST
 - **P14 派发**：prompt_async 204（13:03 CST 起 loop step 推进中，providerID=ark-plan modelID=deepseek-v4.1-flash 日志证实）。范围=上述 4 项整改+evidence 截图更新+维护说明补充；P13 已过验收部分别动。分支 p14-art-fix，verdict 协议照旧。
 - **M3 phase2 长测**：lv-m3-phase2.service MainPID=1694385 运行 ~10.5h，CPU 累计 1d13h43m 持续计算；日志仍 0 world 行无 exit 文件，verdict=null——运行中不计入验收。
 - completion-notice.md 不存在。待收口：P14 复验合并、M3 phase2 终态、最终发布门。
+
+## 2026-09-21 22:2x CST — M3-1 批验可观测性 + 门槛缩尺 + 长局成本量化（分支 m3-observability，未推送）
+- **流式输出**：`--batch K D` 改为每个世界完成即打印 `world=... OK elapsed_s=<秒>` 与 `[done k/N] elapsed=…s world=…` 并 flush；`batch_summary` 旧字段保留、追加 `workers=`/`wall_s=`。实测 `LV_BATCH_WORKERS=4 --batch 4 10`：world 行分别在 368.94s / 431.60s / 460.82s / 545.22s 出现（完成序 3,0,2,1），summary 545.23s，`M3_ACCEPTANCE=PASS`，exit 0；证据 `evidence/stream/batch-4x10.log`（逐行时间戳）。
+- **门槛缩尺**：`scripts/m3_supervise.py` `PHASES.phase2` 由 `(50,100)` 改为 `(50,20)`；`--selftest` 仍 `SELFTEST=PASS`。理由与实测（phase2 11.4h 仅 6/50、100 天≈6.8 CPU-h、全量≈335 CPU-h）写入 `docs/维护说明.md`。判据语义 check a/b/c 未改。
+- **成本探针**：新增只读 `--cost-probe [D1,D2,...]`（默认 1,5,10,20,50,100）。本机单 worker、npcs=30、seed=42 实测：1d 21.615s/19.0GB/gen0=2272，5d 111.645s/98.1GB/gen0=11728，10d 226.226s/202.4GB/gen0=24203（约 20.2GB/天·世界）。证据 `evidence/stream/cost-probe-1-5-10.log`。
+- **长局机理（只测不改）**：`World.Rumors` 永不裁剪，实测 count 1d=2025 / 5d=9821 / 10d=19493（≈1950/天），而 `latestRumorFor`/`rumorIsDuplicate`/`rumorPath` 每次全表扫描 → 单次交互成本随年龄线性、整轮超线性；`Mind.Memory` 有上限（恒 64）非主因。修复方向（裁剪工作集/Map 索引/限窗）留下一单，本单不实施。证据 `evidence/stream/growth-1-5-10.log`。
+- **待收口**：由 Hermes 独立复验本分支；正式 M3 重跑走冻结流程（50×20）。macOS 打包脚本事后单已排队（等本单交付）。
diff --git a/docs/维护说明.md b/docs/维护说明.md
index 4635b6d..19ca201 100644
--- a/docs/维护说明.md
+++ b/docs/维护说明.md
@@ -1,6 +1,6 @@
 # 维护说明
 
-结论先行：主分支保持 0 警告 0 错误 + Desktop 79/79、Kernel 79/79 全绿；性能基线
+结论先行：主分支保持 0 警告 0 错误 + Desktop 86/86、Kernel 79/79 全绿；性能基线
 final_digest 固定 953775FAEB2FDDE97289491AA260BD8D390C571E48A7A13AD2CB6FB7124F7F6C。
 本文记录模块职责、不变量与逐字可复制的验收命令。
 
@@ -83,6 +83,40 @@ dotnet src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll
 - digest 变化处理：先比对 `WorldSave.save world` 全文文本（版本头变化不构成失败即 digest 规范化后等价）；
   再查 Sim.step 数值路径是否被改动（不允许）。历史归属见 `artifacts/perf-independent/*/README.md`。
 
+## M3 批验：流式可观测性、门槛与长局成本（M3-1）
+
+- **流式输出**：`--batch K D` 每个世界一完成就立即打印 `world=... OK elapsed_s=<秒>` 与
+  `[done k/N] elapsed=…s world=…` 并 flush，结束时 `batch_summary` 旧字段原样保留并追加
+  `workers=` 与 `wall_s=`（总墙钟）。完成序可能非升序，监督器按索引集合恰为 0..K-1 校验。
+- **正式门槛缩尺 50×100 → 50×20**（`scripts/m3_supervise.py` 的 `PHASES.phase2`）：
+  phase2（50×100）实测 11.4 小时仅完成 6/50 世界，单世界 100 天约 6.8 CPU 小时，
+  全量约 335 CPU 小时，不可行（诊断见
+  `artifacts/m3-acceptance/final-candidate-20260921T095500Z/logs/phase2_20260921T101500.log.diagnosis.txt`）。
+  Hermes 实测单 worker 成本：1 天 25.6s / 5 天 123s / 10 天 254s / 20 天 ≈900s /
+  100 天 ≈6.8 CPU-h。缩尺只改规模，判据语义 check a/b/c 未动。
+- **成本探针**：`--cost-probe [D1,D2,...]`（默认 `1,5,10,20,50,100`）逐段真实测量单世界
+  `ticks_per_s`、墙钟、`allocated_bytes` 与 gen0/1/2，不估算、不写死。
+  本机（单 worker，npcs=30，seed=42）实测：
+
+  | days | elapsed_s | ticks_per_s | allocated_bytes | gen0 | gen1 | gen2 |
+  |---|---|---|---|---|---|---|
+  | 1 | 21.615 | 239837.5 | 19,001,478,648 | 2272 | 286 | 11 |
+  | 5 | 111.645 | 232165.0 | 98,066,613,520 | 11728 | 11 | 2 |
+  | 10 | 226.226 | 229151.8 | 202,369,225,952 | 24203 | 22 | 2 |
+
+  即约 **20.2 GB / 天·世界** 的分配；10 天内吞吐近乎平稳，超线性主要体现在更老的世界上
+  （Hermes 20 天 ≈45 s/天、100 天 ≈245 s/天）。
+- **成本随年龄上涨的机理（只测不改）**：`World.Rumors: RumorEvent list` 永不裁剪，
+  每次谣言扩散 `rumor :: world.Rumors` 前插。实测 `rumor_trace count`：1 天 2025、
+  5 天 9821、10 天 19493（≈1950/天线性累积）。而 `latestRumorFor` / `rumorIsDuplicate` /
+  `rumorPath` 每次都 `world.Rumors |> List.filter/tryFind` 全表扫描，因此每次聊天/对话的
+  成本随世界年龄线性增长、整轮成本随时长超线性（≈平方）增长；同时每次扫描与前插都产生
+  列表分配，推高 GC。`Mind.Memory` 有 `memoryCapacity` 上限（实测恒 64），不是主因。
+- **修复方向候选（本单不实施，避免动确定性语义）**：①按 `rumorFreshnessTicks`/半衰期
+  确定性裁剪工作集（谱系/annal 另存）；②以 `Map<RumorId, RumorEvent>` 或逐接收者索引替代
+  全表扫描；③把 `latestRumorFor`/`rumorIsDuplicate` 限制到最近窗口。以上都需单开一单、
+  重跑 performance-baseline 与全部 digest 后由 Hermes 走冻结流程。
+
 ## 已验证命令
 
 ```bash
@@ -96,6 +130,11 @@ dotnet build LivingVillage.sln -c Release 2>&1 | tail -n3
 dotnet test src/LivingVillage.Desktop.Tests -c Release --no-build
 dotnet test src/LivingVillage.Kernel.Tests -c Release --no-build
 
+# M3 批次（流式；判据行与原格式逐字节兼容，仅行尾追加 elapsed_s）
+LV_BATCH_WORKERS=4 dotnet src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll --batch 4 10
+# 长局成本阶梯（只读，真实测量）
+dotnet src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll --cost-probe 1,5,10
+
 # 素材再生成（确定性，输出 byte-identical）
 python3 scripts/make-jiangnan-art.py && python3 scripts/make-cc0-art.py
 
diff --git a/scripts/m3_supervise.py b/scripts/m3_supervise.py
index e5a2315..26dfb4a 100755
--- a/scripts/m3_supervise.py
+++ b/scripts/m3_supervise.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 """M3 验收监督（加固版）：供 Hermes cron 每 5 分钟调用（no_agent）。
 
-状态机: idle -> phase1(--batch 30 20) -> phase2(--batch 50 100) -> done
+状态机: idle -> phase1(--batch 30 20) -> phase2(--batch 50 20) -> done
 - flock 单实例防重入；抢不到锁静默退出
 - 冻结运行: 首次启动把 Release 输出目录递归复制到 scripts/m3_frozen/<stamp>_<commit>/，
   v2 清单覆盖全部子目录文件 sha256；两阶段都从冻结副本运行，后续构建不会覆盖
@@ -11,7 +11,8 @@
 - PASS 判定收紧（全部满足才 PASS，任一不满足即 FAIL/停机）:
   1) exitfile 退出码 == 0
   2) `batch worlds=K days=D` 与固定 PHASES[phase] 完全一致（不用 state 里可篡改值）
-  3) `world=` 行: 行数==K、索引恰为 0..K-1 无重复无缺失、seed==42+i
+  3) `world=` 行: 行数==K、索引恰为 0..K-1 无重复无缺失（允许流式完成序）、seed==42+i；
+     行尾 `elapsed_s=<秒>` 为追加观测字段，不参与判据
   4) `batch_summary worlds=K passed=K failed=0`
   5) 行 `M3_ACCEPTANCE=PASS`
   6) 冻结清单 sha256 全部一致 + commit 绑定；v2 清单另查额外/缺失文件
@@ -45,7 +46,7 @@ STATE_PATH = ROOT / "scripts/m3_state.json"
 LOCK_PATH = ROOT / "scripts/m3_supervise.lock"
 LOG_DIR = ROOT / "scripts/m3_logs"
 FROZEN_ROOT = ROOT / "scripts/m3_frozen"
-PHASES = {"phase1": (30, 20), "phase2": (50, 100)}
+PHASES = {"phase1": (30, 20), "phase2": (50, 20)}
 NEXT = {"phase1": "phase2", "phase2": None}
 FRAG = "LivingVillage.Headless.dll"
 
@@ -239,6 +240,14 @@ def read_exitcode(exitfile):
         return None
 
 
+def world_line_verdict(line):
+    """world 行的判据 token；忽略行尾追加的 `elapsed_s=<秒>` 观测字段。"""
+    tokens = line.split()
+    while tokens and re.fullmatch(r"elapsed_s=\d+(\.\d+)?", tokens[-1]):
+        tokens.pop()
+    return tokens[-1] if tokens else ""
+
+
 def verify_batch(text, worlds, days):
     """完整批次判据校验（固定 worlds/days 由调用方从 PHASES 取）。
     返回问题列表（空=满足全部收紧条件）。"""
@@ -263,7 +272,8 @@ def verify_batch(text, worlds, days):
     if malformed:
         problems.append(f"malformed world lines: {malformed[:5]}")
     idxs = [i for i, _ in parsed]
-    if idxs != list(range(worlds)):
+    # 流式输出允许完成序非升序，判据是索引集合恰为 0..K-1（无重无缺）。
+    if sorted(idxs) != list(range(worlds)):
         dup = sorted({i for i in idxs if idxs.count(i) > 1})
         missing = sorted(set(range(worlds)) - set(idxs))
         extra = sorted(set(idxs) - set(range(worlds)))
@@ -271,7 +281,7 @@ def verify_batch(text, worlds, days):
     badseed = [i for i, s in parsed if s != 42 + i]
     if badseed:
         problems.append(f"seed binding mismatch (expect seed=42+i): worlds={badseed[:8]}")
-    bad = [l.split()[0] for l in wlines if not l.split() or l.split()[-1] != "OK"]
+    bad = [l.split()[0] for l in wlines if world_line_verdict(l) != "OK"]
     if bad:
         problems.append(f"non-OK world lines: {bad[:8]}{'...' if len(bad) > 8 else ''}")
     s = re.search(r"^batch_summary worlds=(\d+) passed=(\d+) failed=(\d+) ", text, re.M)
@@ -405,7 +415,7 @@ def decide(state, alive):
             state["verdict"] = "PASS"
             if nxt is None:
                 state["phase"] = "done"
-                return "save", state, f"{phase} PASS (commit={state.get('commit')}, frozen={state.get('frozen')}) -> M3 acceptance complete (30x20 + 50x100)"
+                return "save", state, f"{phase} PASS (commit={state.get('commit')}, frozen={state.get('frozen')}) -> M3 acceptance complete (30x20 + 50x20)"
             return f"launch:{nxt}", state, None
         state["verdict"] = "FAIL"
         state["phase"] = "failed"
@@ -641,6 +651,19 @@ def selftest():
         "batch worlds=2 days=1 x\nworld=0 seed=42 a OK\nworld=1 seed=43 a OK\nbatch_summary worlds=2 passed=1 failed=1 x\nM3_ACCEPTANCE=PASS\n", 2, 1)))
     check("verify missing verdict token", any("M3_ACCEPTANCE=PASS" in p for p in verify_batch(
         "batch worlds=1 days=1 x\nworld=0 seed=42 a OK\nbatch_summary worlds=1 passed=1 failed=0 x\nM3_ACCEPTANCE=FAIL\n", 1, 1)))
+    # 流式输出：完成序非升序 + 行尾 elapsed_s 追加字段，均应通过判据。
+    check("verify out-of-order streaming worlds", verify_batch(
+        "batch worlds=3 days=1 x\nworld=2 seed=44 a OK elapsed_s=3.1\nworld=0 seed=42 a OK elapsed_s=1.2\n"
+        "world=1 seed=43 a OK elapsed_s=2.0\nbatch_summary worlds=3 passed=3 failed=0 x\nM3_ACCEPTANCE=PASS\n", 3, 1) == [])
+    check("verify elapsed suffix tolerated with progress lines", verify_batch(
+        "batch worlds=2 days=1 x\n[done 1/2] elapsed=0.9s world=1\nworld=1 seed=43 a OK elapsed_s=0.9\n"
+        "[done 2/2] elapsed=1.2s world=0\nworld=0 seed=42 a OK elapsed_s=1.2\n"
+        "batch_summary worlds=2 passed=2 failed=0 x\nM3_ACCEPTANCE=PASS\n", 2, 1) == [])
+    check("verify elapsed suffix does not mask BAD", any("non-OK" in p for p in verify_batch(
+        "batch worlds=1 days=1 x\nworld=0 seed=42 a BAD elapsed_s=1.0\n"
+        "batch_summary worlds=1 passed=0 failed=1 x\nM3_ACCEPTANCE=FAIL\n", 1, 1)))
+    check("world_line_verdict ignores elapsed", world_line_verdict("world=0 seed=42 a OK elapsed_s=1.0") == "OK"
+          and world_line_verdict("world=0 seed=42 a BAD elapsed_s=1.0") == "BAD")
 
     # ---- 冻结清单（合成目录, v2 递归 + 额外/缺失检测） ----
     with tempfile.TemporaryDirectory() as td:
@@ -741,8 +764,8 @@ def selftest():
         a, s, m = decide(dict(base, phase="phase2", log=str(plog)), lambda pid, st, fr: False)
         check("phase2 with phase1 log -> FAIL (fixed PHASES)", s["phase"] == "failed" and s["verdict"] == "FAIL")
         plog2 = td / "p2.log"
-        plog2.write_text(make_batch_log(50, 100))
-        a, s, m = decide(dict(base, phase="phase2", worlds=50, days=100, log=str(plog2)), lambda pid, st, fr: False)
+        plog2.write_text(make_batch_log(50, 20))
+        a, s, m = decide(dict(base, phase="phase2", worlds=50, days=20, log=str(plog2)), lambda pid, st, fr: False)
         check("phase2 PASS -> done", s["phase"] == "done" and m and "complete" in m)
         (td / "e.exit").write_text("0\n")
         tampered = dict(base, worlds=1, days=1)
diff --git a/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj b/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj
index b1b0ac2..a8aac24 100644
--- a/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj
+++ b/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj
@@ -10,6 +10,7 @@
     <Compile Include="M6aTests.fs" />
     <Compile Include="M6bTests.fs" />
     <Compile Include="PrototypeTests.fs" />
+    <Compile Include="M3BatchOutputTests.fs" />
     <Compile Include="SampleTests.fs" />
   </ItemGroup>
 
diff --git a/src/LivingVillage.Desktop.Tests/M3BatchOutputTests.fs b/src/LivingVillage.Desktop.Tests/M3BatchOutputTests.fs
new file mode 100644
index 0000000..e962e57
--- /dev/null
+++ b/src/LivingVillage.Desktop.Tests/M3BatchOutputTests.fs
@@ -0,0 +1,84 @@
+namespace LivingVillage.Desktop.Tests
+
+open System.Threading
+open System.Threading.Tasks
+open Microsoft.VisualStudio.TestTools.UnitTesting
+open LivingVillage.Headless
+
+/// M3-1 可观测性回归：批次输出格式与逐世界流式驱动。
+[<TestClass>]
+type M3BatchOutputTests () =
+
+    let sampleBase =
+        "world=3 seed=45 chats_max=9 chats_avg=2.00 ratio=4.500 gini=0.100 relcnt_max=5 relcnt_min=1 relconn=6/6 nonfinite=0 check_a=PASS check_b=PASS check_c=PASS old_crit=PASS OK"
+
+    [<TestMethod>]
+    member _.WorldLineKeepsPrefixAndAppendsElapsed () =
+        let line = BatchOutput.worldLine sampleBase 123.4
+        Assert.IsTrue(line.StartsWith("world=3 seed=45 "), "world 行前缀必须保持不变")
+        Assert.IsTrue(line.StartsWith(sampleBase), "判据字段必须原样保留")
+        Assert.IsTrue(line.EndsWith("elapsed_s=123.4"), "耗时字段追加在行尾")
+        // 旧判据 token 仍在（监督器据此判定 OK）。
+        Assert.AreEqual<string>(sampleBase + " elapsed_s=123.4", line)
+
+    [<TestMethod>]
+    member _.DoneLineReportsProgressAndElapsed () =
+        Assert.AreEqual<string>("[done 2/4] elapsed=12.3s world=1", BatchOutput.doneLine 2 4 12.3 1)
+        Assert.AreEqual<string>("[done 4/4] elapsed=600.0s world=3", BatchOutput.doneLine 4 4 600.0 3)
+
+    [<TestMethod>]
+    member _.SummaryLineKeepsOldFieldsAndAppendsWallClock () =
+        let line = BatchOutput.summaryLine 4 3 1 4 1.100 9.900 0.010 0.400 4 610.5
+        // 监督器正则只读前三个字段，前缀必须逐字保持。
+        Assert.IsTrue(line.StartsWith("batch_summary worlds=4 passed=3 failed=1 old_passed=4/4 "), line)
+        Assert.IsTrue(line.Contains("ratio_min=1.100 ratio_max=9.900 gini_min=0.010 gini_max=0.400"), line)
+        Assert.IsTrue(line.Contains("elapsed_s=610.5"), line)
+        Assert.IsTrue(line.EndsWith("workers=4 wall_s=610.5"), line)
+
+    [<TestMethod>]
+    member _.CostLineDerivesThroughputFromMeasuredTicks () =
+        let line = BatchOutput.costLine 10L 51840000L 254.0 12345678L 3L 1L 0L
+        Assert.IsTrue(line.StartsWith("cost days=10 ticks=51840000 elapsed_s=254.000 "), line)
+        Assert.IsTrue(line.Contains("ticks_per_s=204094.5"), line)
+        Assert.IsTrue(line.EndsWith("allocated_bytes=12345678 gen0=3 gen1=1 gen2=0"), line)
+
+    [<TestMethod>]
+    member _.DefaultCostLadderIsOneFiveTenTwentyFiftyHundred () =
+        CollectionAssert.AreEqual([| 1L; 5L; 10L; 20L; 50L; 100L |], List.toArray BatchOutput.defaultCostLadder)
+
+    [<TestMethod>]
+    member _.RunStreamingEmitsEachWorldBeforeTheNextStarts () =
+        // workers=1：每次 work 后必须立刻 emit 该世界（证明不是收完再打印）。
+        let events = ResizeArray<string> ()
+        let results =
+            BatchOutput.runStreaming 3 1
+                (fun k ->
+                    lock events (fun () -> events.Add(sprintf "work%d" k))
+                    k * k)
+                (fun k _ _ ->
+                    lock events (fun () -> events.Add(sprintf "emit%d" k)))
+        CollectionAssert.AreEqual([| 0; 1; 4 |], results)
+        CollectionAssert.AreEqual(
+            [| "work0"; "emit0"; "work1"; "emit1"; "work2"; "emit2" |],
+            events.ToArray())
+
+    [<TestMethod>]
+    member _.RunStreamingIsCompleteUnderParallelism () =
+        let total = 12
+        let emitted = System.Collections.Concurrent.ConcurrentDictionary<int, int> ()
+        let completions = ResizeArray<int> ()
+        let results =
+            BatchOutput.runStreaming total 4
+                (fun k -> k + 100)
+                (fun k result doneCount ->
+                    Assert.AreEqual<int>(k + 100, result)
+                    emitted.[k] <- doneCount
+                    lock completions (fun () -> completions.Add doneCount))
+        Assert.AreEqual<int>(total, results.Length)
+        Assert.AreEqual<int>(total, emitted.Count)
+        for k in 0 .. total - 1 do
+            Assert.IsTrue(emitted.ContainsKey k, sprintf "world=%d 未 emit" k)
+        // 完成计数恰为 1..total 的一个排列（并发下不重不漏）。
+        CollectionAssert.AreEqual(
+            [| 1 .. total |],
+            completions |> Seq.sort |> Seq.toArray)
diff --git a/src/LivingVillage.Headless/BatchOutput.fs b/src/LivingVillage.Headless/BatchOutput.fs
new file mode 100644
index 0000000..aeaef52
--- /dev/null
+++ b/src/LivingVillage.Headless/BatchOutput.fs
@@ -0,0 +1,75 @@
+namespace LivingVillage.Headless
+
+open System.Threading.Tasks
+
+/// M3 批次的输出格式与流式并行驱动。
+///
+/// 全部是纯函数 / 无模拟依赖的驱动，便于单测；真实运行由 `Program.runBatch`
+/// 传入 `evalWorld` 与 `Console.Out`。格式约定：
+///   - `world=... ` 判据行保持原前缀与字段，行尾追加 `elapsed_s=<秒>`；
+///   - 每完成一个世界追加一行 `[done k/N] elapsed=…s world=…`；
+///   - `batch_summary` 保留旧字段，行尾追加 `workers=` 与 `wall_s=`（总墙钟）。
+module BatchOutput =
+
+    /// `--cost-probe` 默认天数阶梯（单世界逐段实测）。
+    let defaultCostLadder : int64 list = [ 1L; 5L; 10L; 20L; 50L; 100L ]
+
+    /// world 结果行：前缀不变，行尾追加耗时字段。
+    let worldLine (baseLine: string) (elapsedSeconds: float) : string =
+        sprintf "%s elapsed_s=%.1f" baseLine elapsedSeconds
+
+    /// 逐世界完成进度行。
+    let doneLine (completed: int) (total: int) (elapsedSeconds: float) (world: int) : string =
+        sprintf "[done %d/%d] elapsed=%.1fs world=%d" completed total elapsedSeconds world
+
+    /// batch_summary：旧字段原样保留，追加 workers 与 wall_s（总墙钟，append 不删旧字段）。
+    let summaryLine
+        (worlds: int)
+        (passed: int)
+        (failed: int)
+        (oldPassed: int)
+        (ratioMin: float)
+        (ratioMax: float)
+        (giniMin: float)
+        (giniMax: float)
+        (workers: int)
+        (elapsedSeconds: float) : string =
+        sprintf
+            "batch_summary worlds=%d passed=%d failed=%d old_passed=%d/%d ratio_min=%.3f ratio_max=%.3f gini_min=%.3f gini_max=%.3f elapsed_s=%.1f workers=%d wall_s=%.1f"
+            worlds passed failed oldPassed worlds ratioMin ratioMax giniMin giniMax elapsedSeconds workers elapsedSeconds
+
+    /// --cost-probe 单段输出：ticks/s、墙钟、分配与 GC 全部来自真实测量。
+    let costLine
+        (days: int64)
+        (ticks: int64)
+        (elapsedSeconds: float)
+        (allocatedBytes: int64)
+        (gen0: int64)
+        (gen1: int64)
+        (gen2: int64) : string =
+        let ticksPerSecond = if elapsedSeconds > 0.0 then float ticks / elapsedSeconds else 0.0
+        sprintf
+            "cost days=%d ticks=%d elapsed_s=%.3f ticks_per_s=%.1f allocated_bytes=%d gen0=%d gen1=%d gen2=%d"
+            days ticks elapsedSeconds ticksPerSecond allocatedBytes gen0 gen1 gen2
+
+    /// 并行跑 total 个工作，**完成即回调** `emit world result completedCount`。
+    /// 返回按 world 序号存放的结果数组；`work` 必须是仅依赖 k 的确定性纯世界计算。
+    let runStreaming (total: int) (workers: int) (work: int -> 'T) (emit: int -> 'T -> int -> unit) : 'T[] =
+        let results = Array.zeroCreate total
+        let gate = obj ()
+        let mutable completed = 0
+        let options = ParallelOptions(MaxDegreeOfParallelism = max 1 workers)
+        Parallel.For(
+            0,
+            total,
+            options,
+            fun k ->
+                let result = work k
+                let completedCount =
+                    lock gate (fun () ->
+                        results.[k] <- result
+                        completed <- completed + 1
+                        completed)
+                emit k result completedCount)
+        |> ignore
+        results
diff --git a/src/LivingVillage.Headless/LivingVillage.Headless.fsproj b/src/LivingVillage.Headless/LivingVillage.Headless.fsproj
index d9cd0ab..bf8fd7b 100644
--- a/src/LivingVillage.Headless/LivingVillage.Headless.fsproj
+++ b/src/LivingVillage.Headless/LivingVillage.Headless.fsproj
@@ -7,6 +7,7 @@
 
   <ItemGroup>
     <Compile Include="PerformanceProbe.fs" />
+    <Compile Include="BatchOutput.fs" />
     <Compile Include="Program.fs" />
   </ItemGroup>
 
diff --git a/src/LivingVillage.Headless/Program.fs b/src/LivingVillage.Headless/Program.fs
index 2a2482b..89f4f26 100644
--- a/src/LivingVillage.Headless/Program.fs
+++ b/src/LivingVillage.Headless/Program.fs
@@ -449,38 +449,75 @@ let runBatch (worlds: int) (days: int64) : int =
     //     明星集中型分布过严，relcnt_max=0 即判死，与"非均匀分布"的验收目标不符）
     // (c) 无 NaN/Inf
     printfn "checks: (a) chats max/avg > 2.0 (b) npcs_with_relation >= 0.6*npc_count (c) no NaN (old) relcnt_max > 2*relcnt_min (report only)"
-    // 世界间相互独立（seed=42+k，逐世界确定性），可安全并行；输出仍按 world 序号
-    // 顺序打印，逐行内容与串行版本逐字节一致。LV_BATCH_WORKERS 可覆盖（默认 modest 4）。
+    // 世界间相互独立（seed=42+k，逐世界确定性），可安全并行；每个世界一完成即流式
+    // 打印该行与进度（完成序，可能非 world 升序），判据行内容与串行版本逐字节一致。
+    // LV_BATCH_WORKERS 可覆盖（默认 modest 4）。
     let workers =
         match Environment.GetEnvironmentVariable "LV_BATCH_WORKERS" with
         | null | "" -> min 4 Environment.ProcessorCount
         | v -> match Int32.TryParse v with true, n when n > 0 -> min n Environment.ProcessorCount | _ -> min 4 Environment.ProcessorCount
     printfn "workers=%d" workers
-    let results = Array.zeroCreate worlds
-    System.Threading.Tasks.Parallel.For
-        (0, worlds,
-         System.Threading.Tasks.ParallelOptions(MaxDegreeOfParallelism = workers),
-         fun k -> results.[k] <- evalWorld days k)
-    |> ignore
+    Console.Out.Flush()
+    let wall = System.Diagnostics.Stopwatch.StartNew()
+    // 逐世界流式：每个世界一完成立刻打印该行（行尾追加 elapsed_s），再打印一行
+    // [done k/N] 进度，并 flush，使长批次不再整轮零进度。结果仍按 world 序号落数组。
+    let results =
+        LivingVillage.Headless.BatchOutput.runStreaming worlds workers
+            (fun k ->
+                let sw = System.Diagnostics.Stopwatch.StartNew()
+                let verdict = evalWorld days k
+                sw.Stop()
+                (verdict, sw.Elapsed.TotalSeconds))
+            (fun k (verdict, seconds) doneCount ->
+                printfn "%s" (LivingVillage.Headless.BatchOutput.worldLine verdict.Line seconds)
+                printfn "%s" (LivingVillage.Headless.BatchOutput.doneLine doneCount worlds seconds k)
+                Console.Out.Flush())
+    wall.Stop()
     let mutable passed = 0
     let mutable failed = 0
     let mutable oldPassed = 0
     let ratios = ResizeArray<float> ()
     let ginis = ResizeArray<float> ()
-    let sw = System.Diagnostics.Stopwatch.StartNew()
-    for r in results do
-        printfn "%s" r.Line
-        if r.Ok then passed <- passed + 1 else failed <- failed + 1
-        if r.CheckOld then oldPassed <- oldPassed + 1
-        ratios.Add r.Ratio
-        ginis.Add r.Gini
-    sw.Stop()
+    for verdict, _ in results do
+        if verdict.Ok then passed <- passed + 1 else failed <- failed + 1
+        if verdict.CheckOld then oldPassed <- oldPassed + 1
+        ratios.Add verdict.Ratio
+        ginis.Add verdict.Gini
     let allPass = failed = 0
-    printfn "batch_summary worlds=%d passed=%d failed=%d old_passed=%d/%d ratio_min=%.3f ratio_max=%.3f gini_min=%.3f gini_max=%.3f elapsed_s=%.1f"
-        worlds passed failed oldPassed worlds (Seq.min ratios) (Seq.max ratios) (Seq.min ginis) (Seq.max ginis) sw.Elapsed.TotalSeconds
+    printfn "%s"
+        (LivingVillage.Headless.BatchOutput.summaryLine
+            worlds passed failed oldPassed (Seq.min ratios) (Seq.max ratios) (Seq.min ginis) (Seq.max ginis) workers wall.Elapsed.TotalSeconds)
     printfn "M3_ACCEPTANCE=%s" (if allPass then "PASS" else "FAIL")
+    Console.Out.Flush()
     if allPass then 0 else 1
 
+/// `--cost-probe`：单世界按天数阶梯逐段真实测量（墙钟、ticks/s、分配与 GC），
+/// 只读、不改任何模拟语义。天数阶梯可用 `--cost-probe D1,D2,...` 覆盖。
+let runCostProbe (ladder: int64 list) : int =
+    printfn "cost_probe ladder=%s npcs=%d" (ladder |> List.map string |> String.concat ",") Sim.npcCount
+    Console.Out.Flush()
+    for days in ladder do
+        let ticks = days * ticksPerDay
+        System.GC.Collect()
+        let allocBefore = System.GC.GetTotalAllocatedBytes(false)
+        let gen0Before = System.GC.CollectionCount 0
+        let gen1Before = System.GC.CollectionCount 1
+        let gen2Before = System.GC.CollectionCount 2
+        let sw = System.Diagnostics.Stopwatch.StartNew()
+        let stats = runSimulation false days 42UL Sim.npcCount
+        sw.Stop()
+        let allocated = System.GC.GetTotalAllocatedBytes(false) - allocBefore
+        let line =
+            LivingVillage.Headless.BatchOutput.costLine
+                days ticks sw.Elapsed.TotalSeconds allocated
+                (int64 (System.GC.CollectionCount 0 - gen0Before))
+                (int64 (System.GC.CollectionCount 1 - gen1Before))
+                (int64 (System.GC.CollectionCount 2 - gen2Before))
+        printfn "%s" line
+        printfn "cost_probe_check days=%d nonfinite=%d out_of_bounds=%d final_tick=%d" days stats.NonFinite stats.OutOfBounds stats.World.Tick
+        Console.Out.Flush()
+    0
+
 let runExisting (argv: string[]) =
     let rec parse
         (i: int)
@@ -544,14 +581,32 @@ let runExisting (argv: string[]) =
             if stats.NonFinite > 0L || stats.OutOfBounds > 0L || not replayOk then 1 else 0
     | Error msg ->
         eprintfn $"headless: {msg}"
-        eprintfn "usage: dotnet run -c Release --project src/LivingVillage.Headless -- --days N --seed S [--npc N] [--dump-relations] [--dump-rumors] [--replay-rumors] | --m5-smoke | --m6a-smoke | --batch K D | --performance-baseline"
+        eprintfn "usage: dotnet run -c Release --project src/LivingVillage.Headless -- --days N --seed S [--npc N] [--dump-relations] [--dump-rumors] [--replay-rumors] | --m5-smoke | --m6a-smoke | --batch K D | --performance-baseline | --cost-probe [D1,D2,...]"
         2
 
+let private parseCostLadder (argv: string[]) : int64 list =
+    if argv.Length >= 2 then
+        let parsed =
+            argv.[1].Split(',')
+            |> Array.toList
+            |> List.map (fun token ->
+                match Int64.TryParse(token.Trim()) with
+                | true, days when days > 0L -> Some days
+                | _ -> None)
+        if parsed <> [] && List.forall Option.isSome parsed then
+            parsed |> List.map Option.get
+        else
+            LivingVillage.Headless.BatchOutput.defaultCostLadder
+    else
+        LivingVillage.Headless.BatchOutput.defaultCostLadder
+
 [<EntryPoint>]
 let main argv =
     if argv.Length = 1 && argv.[0] = "--m6a-smoke" then
         runM6aSmoke ()
     elif argv.Length = 1 && argv.[0] = "--performance-baseline" then
         LivingVillage.Headless.PerformanceProbe.runDefault ()
+    elif argv.Length >= 1 && argv.[0] = "--cost-probe" then
+        runCostProbe (parseCostLadder argv)
     else
         runExisting argv