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

5 files changed, 235 insertions, 19 deletions

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