fix(occupancy): the at-the-cut proof main carries a stray TH.[_]

docs/occupancy/GradedStages.agda

@@ -67,16 +67,17 @@ run-in-turn<cost = reclaimed (gone 9204) (gone 9180) (s≤s z≤n) (s≤s z≤n)
 vsearch-timeline : TH.Timeline
 vsearch-timeline = gone 9204 TH.∷ gone 9180 TH.∷ TH.[]
 
--- Cut anywhere in it: the suffix starts from the prefix's DRIFT, not its peak.
--- The statement Reclaim.occupancy-at-cut proves, at the two-task cut.
+-- Prove the Reclaim.occupancy-at-cut statement for the ASV/OTU cut: the stage
+-- peak is the maximum of the prefix peak and prefix drift plus suffix peak,
+-- with all memory figures in KiB.
 at-the-cut : peak (TH.grade vsearch-timeline) ≡
              max (peak (TH.grade (gone 9204 TH.∷ TH.[])))
                  (drift (TH.grade (gone 9204 TH.∷ TH.[])) +
                   peak (TH.grade (gone 9180 TH.∷ TH.[])))
 at-the-cut =
-  trans (cong peak (TH.cut-factor (gone 9204 TH.∷ TH.[]) (gone 9180 TH.∷ TH.[_])))
+  trans (cong peak (TH.cut-factor (gone 9204 TH.∷ TH.[]) (gone 9180 TH.∷ TH.[])))
         (reclaim-law (TH.grade (gone 9204 TH.∷ TH.[]))
-                     (TH.grade (gone 9180 TH.∷ TH.[_])))
+                     (TH.grade (gone 9180 TH.∷ TH.[])))
 
 -- And occupancy never exceeds cost, at every cut of a pipeline: the slack is
 -- what the cut reclaims (the statement Reclaim.protocol-occupancy≤cost proves,

src/core/memory_probe.jl

@@ -51,7 +51,7 @@ mutable struct TaskTrace
     peak_kib  :: Int
     hwm_kib   :: Int
     # The sample window the task's children were live in: the index of the first
-    # sample taken after the first of them started, and of the one taken after
+    # sample taken after the first of them started, and the last taken before
     # the last was waited for. 0 means never. Two tasks overlap when the later
     # one's `first_sample` falls at or before the earlier one's `last_sample`,
     # which is the interleaving the grade algebra sequences over.
@@ -277,8 +277,8 @@ function watch!(proc::Base.Process)
     end
     lock(p.lock) do
         t.processes += 1
+        t.processes == 1 && (t.first_sample = p.samples + 1)
         if pid !== nothing
-            isempty(t.live) && (t.first_sample = p.samples)
             push!(t.live, pid)
         end
     end
@@ -326,9 +326,12 @@ end
 #     hwm peak drift        with   drift ≤ peak
 #
 #   * `peak`  — the high-water mark this interaction reached, in KiB;
-#   * `drift` — what it still held when the next interaction began;
+#   * `drift` — the net change in memory held by the next interaction's start;
 #   * `release` = peak − drift — what the frontier reclaims.
 #
+# A task's drift subtracts earlier tasks that exited before the next start, so
+# it can be negative. The folded drift is the sum of peaks still live there.
+#
 # Sequencing starts the next interaction from the previous one's drift:
 #
 #     hwm p₁ n₁ ⊗ hwm p₂ n₂ = hwm (max p₁ (n₁ + p₂)) (n₁ + n₂)
@@ -354,7 +357,7 @@ end
 
 const GRADE_ALGEBRA = "ResourceAlgebra.Instances.HWM"
 
-"A grade of the pinned high-water-mark algebra: a peak and the drift it holds."
+"A grade of the pinned high-water-mark algebra: a peak and a net memory change."
 struct Grade
     peak  :: Int
     drift :: Int
@@ -388,10 +391,10 @@ end
 _is_windowed(rec::AbstractDict) = _window(rec) !== nothing
 
 # Whether `prev` was still live when `next` started, so the handover hands its
-# peak on rather than reclaiming it. A `prev` never unwatched (window end 0) was
-# still live; a `next` that never started (window start 0) never overlapped.
+# peak on rather than reclaiming it. A started `prev` never unwatched (window
+# end 0) was still live; a task that never started (window start 0) never overlapped.
 _overlaps(prev::Tuple{Int,Int}, next::Tuple{Int,Int}) =
-    next[1] != 0 && (prev[2] == 0 || next[1] <= prev[2])
+    prev[1] != 0 && next[1] != 0 && (prev[2] == 0 || next[1] <= prev[2])
 
 # Grades and labels in sampled order, or `nothing` when the record carries no
 # windows: a comparison is then skipped rather than made on an assumed order.
@@ -404,18 +407,24 @@ function _grades_and_labels(occ::AbstractDict)
     order  = sort(collect(eachindex(recs)); by = i -> (_window(recs[i])[1], i))
     ordered = collect(recs)[order]
     windows = [_window(r) for r in ordered]
+    peaks   = [Int(get(r, "peak_rss_kib", 0)) for r in ordered]
     grades  = Grade[]
     labels  = String[]
+    live    = Int[]
+    held    = 0
     for (i, rec) in enumerate(ordered)
-        peak = Int(get(rec, "peak_rss_kib", 0))
-        drift = if i == length(ordered)
+        push!(live, i)
+        if i == length(ordered)
             # The stage waits for its children: a task still resident at the end
-            # keeps its peak as the grade's drift, an exited one holds nothing.
-            (windows[i][1] != 0 && windows[i][2] == 0) ? peak : 0
+            # keeps its peak in the folded drift, an exited one holds nothing.
+            filter!(j -> windows[j][1] != 0 && windows[j][2] == 0, live)
         else
-            _overlaps(windows[i], windows[i + 1]) ? peak : 0
+            filter!(j -> _overlaps(windows[j], windows[i + 1]), live)
         end
-        push!(grades, Grade(peak, drift))
+        remaining = sum(j -> peaks[j], live; init = 0)
+        # seq adds drifts, so reclaim expired predecessors as well as this task.
+        push!(grades, Grade(peaks[i], remaining - held))
+        held = remaining
         push!(labels, String(get(rec, "label", "?")))
     end
     grades, labels
@@ -424,9 +433,11 @@ end
 """
     task_grades(occupancy) -> Union{Vector{Grade},Nothing}
 
-One grade per task of a stage's `occupancy` record, in sampled order: a task's
-drift is its peak when it was still live at the next task's start, and 0 when it
-had exited. `nothing` when the record carries no sample windows.
+One grade per task of a stage's `occupancy` record, in sampled order. Its drift
+is the change in the live peak sum by the next task's start, including releases
+of earlier tasks; the final boundary retains only unclosed windows. Folding
+with `seq` carries exactly the peaks still live at each boundary. `nothing` when
+the record carries no sample windows.
 """
 function task_grades(occ::AbstractDict)
     graded = _grades_and_labels(occ)

test/unit/test_occupancy_grade.jl

@@ -21,8 +21,7 @@ const RECORDED_SCHEDULES =
     normpath(joinpath(@__DIR__, "..", "..", "docs", "occupancy", "schedules.yml"))
 
 # One task as the probe records it, with the sample window a schedule gives it.
-# The window is what the grade reads: a task still live when the next one starts
-# hands its peak on as drift, an exited one hands on nothing.
+# The windows determine which peaks remain live at each handover.
 function _occ_task(peak; label = "t", first_sample = 1, last_sample = 2)
     OrderedDict{String,Any}("label" => label, "processes" => 1, "samples" => 3,
                             "peak_rss_kib" => peak, "hwm_kib" => peak,
@@ -80,6 +79,75 @@ _occ_stage(tasks; measured) = OrderedDict{String,Any}(
         @test MG.disagreement(together) === nothing
     end
 
+    @testset "overlap expires at each handover" begin
+        for (windows, predicted, remaining) in (
+            ([(1, 2), (2, 3), (4, 5)], 160, 0), # an overlapping pair, then a serial task
+            ([(1, 3), (2, 5), (4, 6)], 160, 0), # pairwise overlap, never all three
+            ([(1, 6), (2, 3), (4, 5)], 180, 0), # the first survives a shorter task
+            ([(1, 5), (2, 5), (3, 5)], 240, 0), # genuine three-way overlap
+            ([(1, 2), (1, 2), (3, 4)], 160, 0), # children registered before sample 1
+            ([(1, 0), (2, 3), (4, 0)], 180, 180), # unclosed windows remain live
+        )
+            tasks = [_occ_task(peak; label = "t$i", first_sample = window[1],
+                               last_sample = window[2])
+                     for (i, (peak, window)) in enumerate(zip((100, 60, 80), windows))]
+            occ = _occ_stage(tasks; measured = predicted)
+            grades = MG.task_grades(occ)
+            folded = foldl(MG.seq, grades)
+            @test folded.peak == predicted
+            @test folded.drift == remaining
+            @test MG.release(folded) == predicted - remaining
+            @test MG.cost_bound_rss_kib(occ) == 240
+            @test MG.reclaimed_kib(occ) == 240 - predicted
+            @test MG.disagreement(occ) === nothing
+            @test MG.graded(occ)["predicted_peak_kib"] == predicted
+            # Record order does not change the sampled order.
+            @test MG.stage_grade(_occ_stage(reverse(tasks); measured = predicted)) == folded
+        end
+        @test !MG._overlaps((0, 0), (1, 2)) # a never-started task is not live
+        @test !MG._overlaps((1, 0), (0, 0))
+        @test MG._overlaps((1, 2), (1, 2))
+        @test MG._overlaps((1, 2), (2, 3))   # the end sample is included
+        @test !MG._overlaps((1, 2), (3, 4))
+    end
+
+    @testset "registration keeps the first child's sample boundary" begin
+        # No background sampler: watch! takes the first sample deterministically.
+        probe = MG.Probe(0.01, ReentrantLock(), MG.TaskTrace[], MG.TaskTrace("(stage)"),
+                         0, 0, Base.gc_num(), Base.gc_live_bytes(), false, nothing,
+                         nothing, nothing, nothing, Vector{UInt8}(undef, 8192),
+                         Vector{UInt8}(undef, 64))
+        function child!()
+            proc = run(`sleep 0.01`; wait = false)
+            pid = MG.watch!(proc)
+            try
+                wait(proc)
+            finally
+                MG.unwatch!(pid)
+            end
+        end
+        MG.within(probe) do
+            MG.task("two children") do
+                child!()
+                @test probe.samples == 1
+                @test probe.tasks[1].first_sample == 1
+                @test probe.tasks[1].last_sample == 1
+                MG.task("between children") do
+                    child!()
+                end
+                child!()
+            end
+        end
+        first_task, between = probe.tasks
+        @test first_task.processes == 2
+        @test first_task.first_sample == 1
+        @test first_task.last_sample == 3
+        @test between.first_sample == 2
+        @test MG._overlaps((first_task.first_sample, first_task.last_sample),
+                           (between.first_sample, between.last_sample))
+        @test MG.TaskTrace("never started").first_sample == 0
+    end
+
     @testset "the graded record says what it is" begin
         rec = _occ_stage([_occ_task(100; label = "vsearch:asv", first_sample = 1, last_sample = 4),
                           _occ_task(60;  label = "vsearch:otu", first_sample = 2, last_sample = 5)];
@@ -93,7 +161,9 @@ _occ_stage(tasks; measured) = OrderedDict{String,Any}(
         @test g["disagreement"] === nothing
         @test [t["label"] for t in g["tasks"]] == ["vsearch:asv", "vsearch:otu"]
         @test g["tasks"][1]["drift_kib"] == 100       # still resident at the handover
-        @test g["tasks"][2]["release_kib"] == 60      # the last one hands on nothing
+        @test g["tasks"][2]["drift_kib"] == -100     # the earlier task also exited
+        @test g["tasks"][2]["release_kib"] == 160    # both peaks are reclaimed
+        @test g["drift_kib"] == 0
     end
 
     @testset "a measured peak above the prediction is a disagreement" begin