Request accounting and prompt timing marks: yielded prompt legs summed once, admission time without a double count, unknown stays unknown (P0.4)

Reporting only (class E): no engine speedup, no change to what the engine computes.

serve/server.py
- A prompt read that gives way (BYIELD) and resumes keeps the request's external prefix boundary, and the prompt
  time and tokens of every leg are summed exactly once (prefill_leg_stats; PR #5318).
- The admission's DONE times one token and BDONE counts that token but times only the slot: the slot time now
  includes the admission time, and the output token is not counted twice. A figure the engine never reported
  (a close at the admission token, in the slot, a carry back to the solo path) stays unknown (None) instead of
  turning into 0 or into the API's output count (PR #5307).
- The two patches overlap in the request finalization; ended_on_yield is taken before the unknown-figures step so
  the identity check still sees the yield's own record.

PfTimer::mark(): a mark in the phase already running is not recorded (PR #4006). The totals are the same, a
fraction of the event records; under WDDM every record is a submission of its own and the launch gaps they made
were charged to "dequant". The rule is a pure function in prefill/timer_marks.hpp with a host test.

Tests: serve/test_yield_accounting.py (new: two yields, solo and admission yields, cancel after one and two
yields), serve/test_parallel.py BatchLegAccounting (EOS and length finishes, zero slot time, close at the
admission token and in the slot, slot to solo carry), serve/test_request_accounting.py (unknown stays unknown,
explicit zero), tests/prefill/timer_marks_test.cpp (same totals with 18 of 23 records skipped).

Co-authored-by: YusenZhang0601 <[email protected]>
Co-authored-by: shoucandanghehe <[email protected]>
Co-authored-by: Vop3p <[email protected]>

CMakeLists.txt

@@ -410,6 +410,11 @@ if(STRATA_BUILD_TESTS)
   target_link_libraries(routed_stream_test PRIVATE Threads::Threads)
   add_test(NAME routed_stream_test COMMAND routed_stream_test)
 
+  # PR #4006: STRATA_PREFILL_TIMING does not record a mark in the phase already running (header-only, no GPU)
+  add_executable(timer_marks_test tests/prefill/timer_marks_test.cpp)
+  target_include_directories(timer_marks_test PRIVATE include)
+  add_test(NAME timer_marks_test COMMAND timer_marks_test)
+
   # the Windows HIP auto prefill chunk cap (#1630) and the commit-limit verdict (#1607): header-only, no GPU
   add_executable(host_limits_test tests/platform/host_limits_test.cpp)
   target_include_directories(host_limits_test PRIVATE include)

include/strata/prefill/timer_marks.hpp

@@ -0,0 +1,20 @@
+// include/strata/prefill/timer_marks.hpp - which STRATA_PREFILL_TIMING marks are recorded (PR #4006, @Vop3p).
+//
+// PfTimer charges the gap between two consecutive marks to the phase of the earlier one.  Two marks in a row in the same
+// phase therefore add nothing to the totals: the gap between them is charged to that phase either way.  Skipping the
+// second one leaves the totals as they were with a fraction of the event records.  That matters under WDDM, where every
+// cudaEventRecord is a submission of its own (~80 us on an RX 6900 XT, Windows 11): compute() marks "dequant" once per
+// expert, which put ~16 records per MMQ group on the stream and charged the launch gaps they made to "dequant".
+// Pure functions, so a host test can check the rule and the totals without a GPU.
+#pragma once
+
+#include <cstddef>
+
+namespace strata::prefill {
+
+/// A mark for `phase` is redundant when the newest recorded mark (ph[used - 1]) is already in that phase.
+inline bool timer_mark_repeats(const int* ph, size_t used, int phase) {
+    return used > 0 && ph[used - 1] == phase;
+}
+
+}  // namespace strata::prefill

serve/server.py

@@ -1926,10 +1926,12 @@ class StrataEngine:
         return None
 
     def _apply_bdone(self, f: list[str]) -> None:
-        """`BDONE <slot> <produced> <finish> <ms>` into this request's figures."""
+        """BDONE counts the admission token too, but its clock starts after the admission's DONE."""
         if len(f) >= 5 and isinstance(self.last, dict):
-            self.last = {**self.last, "finish": f[3], "decode_ms": float(f[4]),
-                         "generated": int(f[2]) if f[2].isdigit() else self.last.get("generated")}
+            admission_ms = self.last.get("decode_ms")
+            self.last = {**self.last, "finish": f[3],
+                         "decode_ms": admission_ms + float(f[4]) if admission_ms is not None else None,
+                         "generated": int(f[2]) if f[2].isdigit() else None}
 
     def _raise_if_fatal_gpu(self, line: str) -> None:
         """#4350: the engine's ERR line "fatal GPU error: ..." means it is ending (its GPU context is gone).  The request in flight fails
@@ -2251,6 +2253,7 @@ class StrataEngine:
         orig_len = len(prompt)  # a leg that continues (prompt + what it generated) names where the reply began: pen_from, the penalty window's start
         reused0 = None      # the first read's reused tokens: a continued leg's count includes this request's own output
         carry = {}          # #1869: the figures of the legs that ended in a hand-back to the solo path (the final DONE adds to them)
+        yielded_legs = []    # prompt reads that gave way before producing output
         ok = yield from self._take_control(cancel, len(prompt), new=True,
                                            prompt=prompt if (COALESCE_PREFILL > 1 and not embeddings) else None)
         if not ok:
@@ -2266,6 +2269,7 @@ class StrataEngine:
             slot, cancel._strata_preslot = pre, None
         phase = "none"            # solo -> (admit -> slot) ; "done" once the engine has finished with this request
         stop_sent = False
+        admission_pending, admission_before = False, None   # the admission's DONE is partial until BADM 0 or BDONE
         yields, solo_again, refused = 0, 0, 0
         admit_wait, admit_since, admit_said = False, 0.0, False   # #1603: waiting for a slot, since when, said
         try:
@@ -2326,6 +2330,7 @@ class StrataEngine:
                     if self._done_finish() == "nonfinite":    # #879 (never promoted to a slot)
                         raise EngineNonFinite("the engine's logits became non-finite (#879)")
                     if self._yielded is not None and reserved is not None and self._yielded[0] == reserved:
+                        yielded_legs.append(self.last)
                         slot, reserved = reserved, None     # gave way: the read goes on in that slot (below)
                     elif reserved is not None:              # it did not give way: the slot is free again
                         with self.slot_cv:
@@ -2431,6 +2436,7 @@ class StrataEngine:
                     if started is not None:
                         started()
                     self.slot_held[slot] = []               # the admission overwrites what the slot held
+                    admission_pending, admission_before = True, self.last
                     phase = "admit"
                     self._ctl_mode, self._ctl_result, self._yielded = "batch", None, None
                     self._ctl_erred = False
@@ -2478,6 +2484,7 @@ class StrataEngine:
                         slot = None
                         continue
                     cont = bool(self._ctl_result and self._ctl_result[1])
+                    admission_pending = cont
                     phase = "slot" if cont else "none"
                     while pending:
                         t = pending.pop(0)
@@ -2486,6 +2493,7 @@ class StrataEngine:
                     if not cont and self._done_finish() == "nonfinite":   # #879
                         raise EngineNonFinite("the engine's logits became non-finite (#879)")
                     if not cont and self._yielded is not None and self._yielded[0] == slot and not cancel.is_set():
+                        yielded_legs.append(self.last)
                         continue                            # gave way: again once the shorter request is in
                     break
                 self.ctl.release()
@@ -2528,6 +2536,7 @@ class StrataEngine:
                         phase = "none"
                         f = line.split()
                         self._apply_bdone(f)
+                        admission_pending = False
                         # what the slot's sessions hold now: the prompt and every token fed (all but the last one)
                         self.slot_held[slot] = list(prompt) + out[gen0:-1] if len(out) > gen0 else []
                         if f[3:4] == ["nonfinite"]:              # #879: the engine dropped the slot's state
@@ -2569,6 +2578,8 @@ class StrataEngine:
                         undrained = "DONE"
                 elif phase == "admit":
                     line = self._drain_control("BADM", born=born, lines_q=lines_q)
+                    if line and line.startswith("BADM "):
+                        admission_pending = line.split()[2:3] == ["1"]
                     if line and line.startswith("BADM ") and line.split()[2:3] == ["1"]:
                         phase = "slot"
                     elif line is None and self.gen == born and self.alive():
@@ -2586,11 +2597,19 @@ class StrataEngine:
                 bd = self._bdone_after_eos(slot, q=sq[slot])    # (#1869) the slot's figures, not the admission's
                 if bd is not None:
                     self._apply_bdone(bd.split())
+                    admission_pending = False
                     bdone_in = True
+            ended_on_yield = bool(yielded_legs and self.last is yielded_legs[-1])
+            if admission_pending and isinstance(self.last, dict) and (self.last is not admission_before or carry):
+                # A background drain owns the rest. The admission's one token and time are not this request's totals.
+                self.last = {**self.last, "generated": None, "decode_ms": None}
             if reused0 is not None and isinstance(self.last, dict):
                 self.last = {**self.last, "reused": reused0}
             if carry and isinstance(self.last, dict):
                 self.last = add_carry(carry, self.last)
+            earlier_reads = yielded_legs[:-1] if ended_on_yield else yielded_legs
+            if earlier_reads and isinstance(self.last, dict):
+                self.last = prefill_leg_stats(earlier_reads, self.last, len(ids))
             if holding:
                 self.ctl.release()
             if admit_wait:
@@ -6768,8 +6787,9 @@ def prompt_tokens_seen(prompt_tokens: int, last: dict) -> int:
 def carry_leg(carry: dict, leg: dict) -> None:
     """#1869: a request that decoded in a batch slot and went back to the solo path ends with the solo path's DONE, which
     replaces the engine's figures: keep the slot leg's decode time, tokens and prompt figures to add to it."""
-    carry["decode_ms"] = carry.get("decode_ms", 0.0) + float(leg.get("decode_ms") or 0.0)
-    carry["generated"] = carry.get("generated", 0) + int(leg.get("generated") or 0)
+    for key in ("decode_ms", "generated"):
+        previous, value = carry.get(key, 0), leg.get(key)
+        carry[key] = previous + value if previous is not None and value is not None else None
     carry.setdefault("prompt_ms", float(leg.get("prompt_ms") or 0.0))
     if leg.get("prompt_read") is not None:
         carry.setdefault("prompt_read", int(leg["prompt_read"]))
@@ -6778,9 +6798,10 @@ def carry_leg(carry: dict, leg: dict) -> None:
 def add_carry(carry: dict, last: dict) -> dict:
     """The request's totals: the earlier legs' decode time and tokens plus the final one's; the prompt figures of the leg
     that read the prompt (the solo continuation reads next to nothing)."""
-    out = {**last, "decode_ms": float(last.get("decode_ms") or 0.0) + carry.get("decode_ms", 0.0),
-           "generated": int(last.get("generated") or 0) + carry.get("generated", 0),
-           "prompt_ms": max(float(last.get("prompt_ms") or 0.0), carry.get("prompt_ms", 0.0))}
+    out = {**last, "prompt_ms": max(float(last.get("prompt_ms") or 0.0), carry.get("prompt_ms", 0.0))}
+    for key in ("decode_ms", "generated"):
+        previous, value = carry.get(key, 0), last.get(key)
+        out[key] = previous + value if previous is not None and value is not None else None
     if "prompt_read" in carry:
         out["prompt_read"] = carry["prompt_read"]
     return out
@@ -6811,14 +6832,14 @@ def request_timings(prompt_tokens: int, generated: int, last: dict) -> dict | No
         # #1870: the engine says how many prompt tokens it read: the slot's own session may have held the rest
         # (not the conversation cache), so the rate is not made of tokens it never read
         cache_n = max(cache_n, max(0, prompt_tokens - int(last["prompt_read"])))
-    prompt_n, prompt_ms, decode_ms = max(0, prompt_tokens - cache_n), float(last["prompt_ms"]), float(last.get("decode_ms") or 0)
-    decoded = int(last.get("generated") or generated)          # the engine's count gives its rate, as /metrics does
+    prompt_n, prompt_ms, decode_ms = max(0, prompt_tokens - cache_n), float(last["prompt_ms"]), last.get("decode_ms")
+    decoded = last.get("generated")          # explicit zero and missing native work are not the API's output count
     return {"cache_n": cache_n, "prompt_n": prompt_n, "prompt_ms": round(prompt_ms, 1),
             "prompt_per_token_ms": round(prompt_ms / prompt_n, 3) if prompt_n else None,
             "prompt_per_second": round(prompt_n / (prompt_ms / 1000), 1) if prompt_n and prompt_ms > 0 else None,
-            "predicted_n": generated, "predicted_ms": round(decode_ms, 1),
-            "predicted_per_token_ms": round(decode_ms / decoded, 3) if decoded else None,
-            "predicted_per_second": round(decoded / (decode_ms / 1000), 1) if decoded and decode_ms > 0 else None,
+            "predicted_n": generated, "predicted_ms": round(decode_ms, 1) if decode_ms is not None else None,
+            "predicted_per_token_ms": round(decode_ms / decoded, 3) if decoded and decode_ms is not None else None,
+            "predicted_per_second": round(decoded / (decode_ms / 1000), 1) if decoded and decode_ms and decode_ms > 0 else None,
             # the speculative drafts, as llama.cpp names them (from PR #83, @mikicvi): only when the engine reported them
             **({"draft_n": int(last["drafts_offered"]), "draft_n_accepted": int(last["drafts_accepted"])}
                if last.get("drafts_offered") is not None else {})}
@@ -6836,11 +6857,23 @@ def request_stats(segments: list[dict]) -> dict:
             result.pop(key, None)
     for key in ("prompt_ms", "decode_ms", "generated", "drafts_offered", "drafts_accepted",
                 "hits", "lookups", "offloaded", "ram_blobs", "file_blobs", "file_mb", "experts_streamed"):
-        if any(segment.get(key) is not None for segment in segments):
+        if key in ("decode_ms", "generated") and any(segment.get(key) is None for segment in segments):
+            result[key] = None
+        elif any(segment.get(key) is not None for segment in segments):
             result[key] = sum(segment.get(key) or 0 for segment in segments)
     return result
 
 
+def prefill_leg_stats(earlier: list[dict], last: dict, prompt_tokens: int) -> dict:
+    """Total yielded reads once, retaining only reuse from before this API request."""
+    legs = [*earlier, last]
+    total = request_stats(legs)
+    if all(leg.get("prompt_read") is not None for leg in legs):
+        fresh = max(0, prompt_tokens - int(total.get("reused") or 0))
+        total["prompt_read"] = min(fresh, sum(int(leg["prompt_read"]) for leg in legs))
+    return total
+
+
 def _debug_req(api, req, messages, tools, max_new, thinking, prompt_tokens):
     """One compact line per request while diagnosing blank/empty turns. Set STRATA_DEBUG=1 to enable."""
     if not os.environ.get("STRATA_DEBUG"):

serve/test_parallel.py

@@ -32,6 +32,7 @@ fail = "--fail-window" in args
 groups = int(args[args.index("--says-groups") + 1]) if "--says-groups" in args else 0   # INFO batch_groups=G (--batch-groups auto)        # #997: the first window over two slots fails
 STEP = 0.02
 CH = 32
+BDONE_MS = " 0.0" if "--zero-bdone-ms" in args else " 1.0"
 lines, stop = queue.Queue(), threading.Event()
 def reader():
     for l in sys.stdin:
@@ -72,7 +73,7 @@ def window():        # one batch window: every active slot one token
         left, max_new, produced = active[b]
         if b in nf_slots and produced >= 2:
             nf_slots.discard(b)
-            print(f"BDONE {b} {produced} nonfinite 1.0", flush=True)
+            print(f"BDONE {b} {produced} nonfinite{BDONE_MS}", flush=True)
             del active[b]
             continue
         t = left.pop(0) if left else 257
@@ -81,7 +82,7 @@ def window():        # one batch window: every active slot one token
         if t == 257 or produced >= max_new:
             fin = 'stop' if t == 257 else 'cancel' if b in stopped else 'length'
             stopped.discard(b)
-            print(f"BDONE {b} {produced} {fin} 1.0", flush=True)
+            print(f"BDONE {b} {produced} {fin}{BDONE_MS}", flush=True)
             del active[b]
         else:
             active[b] = [left, max_new, produced]
@@ -583,6 +584,102 @@ class FakeEngineCase(unittest.TestCase):
         return res
 
 
+class BatchLegAccounting(FakeEngineCase):
+    """The admission's DONE times one token; BDONE counts that token but times only the slot."""
+
+    def queued_chats(self, texts=("first", "second"), max_tokens=10, stop=None):
+        replies, errors = {}, []
+
+        def request(text):
+            body = {"messages": [{"role": "user", "content": text}], "max_tokens": max_tokens,
+                    "reasoning_effort": "none", **({"stop": [stop]} if stop else {})}
+            req = urllib.request.Request(self.base + "/v1/chat/completions", data=json.dumps(body).encode(),
+                                         headers={"Content-Type": "application/json"})
+            try:
+                with urllib.request.urlopen(req, timeout=15) as response:
+                    replies[text] = json.loads(response.read())
+            except Exception as error:
+                errors.append(repr(error))
+
+        workers = [threading.Thread(target=request, args=(text,)) for text in texts]
+        # Both requests must be waiting before either takes control: the first cannot take the solo shortcut.
+        with self.engine.ctl:
+            for worker in workers:
+                worker.start()
+            deadline = time.monotonic() + 3
+            while self.engine.waiting < 2 and time.monotonic() < deadline:
+                time.sleep(0.01)
+            self.assertEqual(self.engine.waiting, 2)
+        for worker in workers:
+            worker.join(15)
+            self.assertFalse(worker.is_alive())
+        self.assertEqual(errors, [])
+        self.assertEqual(len(replies), 2)
+        self.assertEqual([line.split()[0] for line in self.log.read_text().splitlines()][:2], ["BGEN", "BGEN"])
+        return replies
+
+    def test_eos_adds_admission_time_without_counting_its_token_twice(self):
+        self.start(2)
+        with mock.patch.object(server, "EOS_IDS", {257}):
+            replies = self.queued_chats()
+        for reply in replies.values():
+            timings = reply["timings"]
+            self.assertEqual((timings["predicted_n"], timings["predicted_ms"]), (10, 21.0))
+            self.assertEqual(timings["predicted_per_second"], 476.2)
+        metrics = self.get("/metrics")
+        self.assertEqual([(row["engine_generated"], row["decode_ms"]) for row in metrics["requests"]],
+                         [(10, 21.0), (10, 21.0)])
+        self.assertEqual(metrics["totals"]["decode_ms"], 42.0)
+
+    def test_length_finish_also_adds_admission_time(self):
+        self.start(2)
+        for reply in self.queued_chats(max_tokens=8).values():
+            self.assertEqual((reply["timings"]["predicted_n"], reply["timings"]["predicted_ms"]), (8, 21.0))
+
+    def test_zero_slot_time_is_known_and_still_includes_admission(self):
+        self.start(2, more=("--zero-bdone-ms",))
+        with mock.patch.object(server, "EOS_IDS", {257}):
+            replies = self.queued_chats()
+        for reply in replies.values():
+            self.assertEqual(reply["timings"]["predicted_ms"], 20.0)
+
+    def assert_incomplete_after_stop(self, stop):
+        self.start(2)
+        replies = self.queued_chats(stop=stop)
+        for reply in replies.values():
+            self.assertIsNone(reply["timings"]["predicted_ms"])
+            self.assertIsNone(reply["timings"]["predicted_per_second"])
+        for row in self.get("/metrics")["requests"]:
+            self.assertIsNone(row["engine_generated"])
+            self.assertIsNone(row["decode_ms"])
+            self.assertIsNone(row["decode_tok_s"])
+        self.assertEqual(self.get("/metrics")["totals"]["decode_ms"], 0.0)
+
+    def test_close_at_admission_token_does_not_publish_partial_totals(self):
+        self.assert_incomplete_after_stop("o")
+
+    def test_close_in_slot_does_not_publish_partial_totals(self):
+        self.assert_incomplete_after_stop("k, ")
+
+    def test_slot_to_solo_carry_includes_admission_once(self):
+        self.start(2, slot_cache=True)
+        with mock.patch.object(server, "EOS_IDS", {257}):
+            replies = self.queued_chats(("short", "LONGREPLY"), max_tokens=200)
+        lines = [line.split() for line in self.log.read_text().splitlines()]
+        solos = [line for line in lines if line[0] == "GEN"]
+        self.assertEqual(len(solos), 1, lines)
+        prompt_lengths = [int(line[2]) for line in lines if line[0] == "BGEN"]
+        slot_tokens = int(solos[0][2]) - max(prompt_lengths)
+        self.assertGreaterEqual(slot_tokens, 2)
+        self.assertLess(slot_tokens, 55)
+        # "ok, " + 15 * "la " + "done." + EOS: 55 total tokens. The solo leg times its remaining tokens at 20 ms each.
+        expected_ms = 21.0 + (55 - slot_tokens) * 20.0
+        reply = replies["LONGREPLY"]
+        self.assertEqual((reply["timings"]["predicted_n"], reply["timings"]["predicted_ms"]), (55, expected_ms))
+        row = max(self.get("/metrics")["requests"], key=lambda item: item["output_tokens"])
+        self.assertEqual((row["engine_generated"], row["decode_ms"]), (55, expected_ms))
+
+
 class ParallelService(FakeEngineCase):
     """The real StrataEngine and Service over HTTP, the fake engine behind them: slots, queues, benchmarks, telemetry.
     The yielding, restart and failure tests are in serve/test_parallel_yield.py (this file alone ran past 100 s)."""

serve/test_request_accounting.py

@@ -9,7 +9,7 @@ import urllib.request
 from pathlib import Path
 
 from serve.frontend import ChatTemplate
-from serve.server import ByteTokenizer, EngineDied, Service, add_carry, carry_leg, request_timings, serve
+from serve.server import ByteTokenizer, EngineDied, Service, add_carry, carry_leg, request_stats, request_timings, serve
 from serve.test_server import ThinkingEngine
 
 
@@ -204,6 +204,33 @@ class SlotLegAccounting(unittest.TestCase):
         self.assertEqual((total["generated"], total["decode_ms"], total["prompt_ms"]), (35, 450.0, 50.0))
         self.assertNotIn("prompt_read", total)
 
+    def test_missing_slot_time_stays_unknown_after_solo_carry(self):
+        carry = {}
+        carry_leg(carry, {"generated": 10, "decode_ms": None, "prompt_ms": 50.0})
+        total = add_carry(carry, {"generated": 5, "decode_ms": 100.0, "prompt_ms": 1.0})
+        self.assertEqual(total["generated"], 15)
+        self.assertIsNone(total["decode_ms"])
+
+    def test_missing_final_figures_stay_unknown_after_carry_and_segments(self):
+        carry = {}
+        carry_leg(carry, {"generated": 10, "decode_ms": 20.0, "prompt_ms": 50.0})
+        total = add_carry(carry, {"generated": None, "decode_ms": None, "prompt_ms": 1.0})
+        combined = request_stats([{"generated": 5, "decode_ms": 100.0, "prompt_ms": 10.0}, total])
+        self.assertIsNone(combined["generated"])
+        self.assertIsNone(combined["decode_ms"])
+
+    def test_unknown_decode_time_is_not_reported_as_zero(self):
+        timings = request_timings(100, 10, {"generated": 10, "decode_ms": None, "prompt_ms": 50.0})
+        self.assertIsNone(timings["predicted_ms"])
+        self.assertIsNone(timings["predicted_per_token_ms"])
+        self.assertIsNone(timings["predicted_per_second"])
+
+    def test_explicit_zero_native_count_is_not_replaced_by_api_output_count(self):
+        timings = request_timings(100, 5, {"generated": 0, "decode_ms": 0.0, "prompt_ms": 50.0})
+        self.assertEqual((timings["predicted_n"], timings["predicted_ms"]), (5, 0.0))
+        self.assertIsNone(timings["predicted_per_token_ms"])
+        self.assertIsNone(timings["predicted_per_second"])
+
     def test_prompt_n_is_what_was_read(self):
         # the slot already held the whole prompt: 1 token read, the conversation cache off (reused 0)
         last = {"prompt_ms": 563.5, "decode_ms": 700.0, "generated": 30, "reused": 0, "prompt_read": 1, "finish": "stop"}

serve/test_yield_accounting.py

@@ -0,0 +1,132 @@
+"""Real serving flow over a fake native engine with accurate partial-read DONE fields."""
+
+import threading, time, unittest
+from unittest import mock
+from serve import test_parallel as tp
+
+
+def accurate_fake():
+    source = tp.FAKE_BATCH
+    source = source.replace("def reused(ids):", "partial = {}\ndef reused(ids):", 1)
+    old = "    return 0 if not REUSE else len(ids) - 1 if continued(ids)[1] else 3"
+    assert source.count(old) == 1
+    source = source.replace(
+        old,
+        """    if not REUSE: return 0
+    kept = max((len(h) for h in partial.values() if ids[:len(h)] == h), default=0)
+    return max(kept, len(ids) - 1 if continued(ids)[1] else 3)""",
+    )
+    old = "        given = None\n        for q in range(CH, len(ids) - 1, CH):"
+    assert source.count(old) == 1
+    source = source.replace(
+        old,
+        "        read_from = reused(ids)\n        given = None\n        for q in range((read_from // CH + 1) * CH, len(ids) - 1, CH):",
+    )
+    old = '            print(f"DONE 0 {len(ids)} 5.0 0.0 cancel 0 0 {reused(ids)}", flush=True)'
+    assert source.count(old) == 1
+    source = source.replace(
+        old,
+        """            partial[given[0]] = ids[:given[1]]
+            print(f"DONE 0 {len(ids)} 5.0 0.0 cancel 0 0 {read_from}" +
+                  (f" 0 0 0 0 0.0 {given[1] - read_from}" if READF else ""), flush=True)""",
+    )
+    source = source.replace("{len(ids) - reused(ids)}", "{len(ids) - read_from}")
+    source = source.replace("{reused(ids)}", "{read_from}")
+    compile(source, "accurate_fake", "exec")
+    return source
+
+
+class YieldAccounting(tp.FakeEngineCase):
+    def start_accurate(self, slots):
+        with mock.patch.object(tp, "FAKE_BATCH", accurate_fake()):
+            self.start(slots, reuse=True, more=["--readfield"])
+
+    def race_long(self):
+        replies = []
+        errors = []
+
+        def long():
+            try:
+                replies.append(self.chat("long " * 600, max_tokens=64))
+            except Exception as error:
+                errors.append(repr(error))
+
+        thread = threading.Thread(target=long)
+        thread.start()
+        time.sleep(0.15)
+        self.chat("short", max_tokens=64)
+        thread.join(20)
+        self.assertFalse(thread.is_alive())
+        self.assertEqual(errors, [])
+        self.assertEqual(len(replies), 1)
+        return replies[0]
+
+    def check_full_input(self, response):
+        n = response["usage"]["prompt_tokens"]
+        log = self.log.read_text().splitlines()
+        self.assertTrue(any(x.startswith("YIELD ") for x in log), log)
+        legs = [x for x in log if x.startswith(("GEN ", "BGEN ")) and int(x.split()[2]) == n]
+        self.assertGreaterEqual(len(legs), 2)
+        self.assertEqual(response["usage"]["prompt_tokens_details"]["cached_tokens"], 3)
+        self.assertEqual((response["timings"]["cache_n"], response["timings"]["prompt_n"]), (3, n - 3))
+        self.assertEqual(response["timings"]["prompt_ms"], 5.0 * len(legs))
+        row = max(self.svc.history, key=lambda x: x["prompt_tokens"])
+        self.assertEqual((row["reused"], row["prompt_read"]), (3, n - 3))
+
+    def test_solo_yield_keeps_original_cache_and_sums_prompt_work(self):
+        self.start_accurate(2)
+        self.check_full_input(self.race_long())
+
+    def test_admission_yield_keeps_original_cache_and_sums_prompt_work(self):
+        self.start_accurate(3)
+        errors = []
+
+        def background():
+            try:
+                self.chat("LONGREPLY background", max_tokens=60)
+            except Exception as error:
+                errors.append(repr(error))
+
+        bg = threading.Thread(target=background)
+        bg.start()
+        time.sleep(0.05)
+        response = self.race_long()
+        bg.join(20)
+        self.assertFalse(bg.is_alive())
+        self.assertEqual(errors, [])
+        self.check_full_input(response)
+
+    def cancel_after_yields(self, count):
+        self.start_accurate(2)
+        original = self.engine._take_control
+        seen = 0
+
+        def take(cancel, plen, after_epoch=None, **kwargs):
+            nonlocal seen
+            if after_epoch is not None:
+                seen += 1
+                if seen == count:
+                    cancel.set()
+                    return False
+                return (yield from original(cancel, plen, **kwargs))
+            return (yield from original(cancel, plen, **kwargs))
+
+        with mock.patch.object(self.engine, "_shorter_waiting", return_value=True), mock.patch.object(
+            self.engine, "_take_control", side_effect=take
+        ):
+            self.chat("long " * 600, max_tokens=64)
+        positions = [int(x.split()[2]) for x in self.log.read_text().splitlines() if x.startswith("YIELD ")]
+        self.assertEqual(len(positions), count)
+        self.assertEqual(positions, sorted(set(positions)))
+        row = self.svc.history[-1]
+        self.assertEqual(row["finish"], "cancel")
+        self.assertEqual(
+            (row["prompt_tokens"], row["reused"], row["prompt_read"]), (positions[-1], 3, positions[-1] - 3)
+        )
+        self.assertEqual(row["prompt_ms"], 5.0 * count)
+
+    def test_cancel_after_one_yield_counts_that_done_once(self):
+        self.cancel_after_yields(1)
+
+    def test_cancel_after_two_yields_counts_each_done_once(self):
+        self.cancel_after_yields(2)

src/prefill/prefill.cpp

@@ -2,6 +2,7 @@
 #include "strata/kernels/fused_gr.hpp"
 #include "strata/prefill/prefill.hpp"
 #include "strata/prefill/copy_events.hpp"
+#include "strata/prefill/timer_marks.hpp"
 #include "strata/prefill/pf_fused_override.hpp"
 #if defined(STRATA_HAVE_DFLASH2)
 #include "strata/kernels/dflash2.hpp"
@@ -3002,6 +3003,11 @@ struct PfTimer {
     double ms[kPfCount] = {};
     void mark(int phase, cudaStream_t s) {
         if (!on) return;
+        // a mark in the phase already running adds nothing to the totals (the gaps are charged to the earlier mark's
+        // phase either way), but every event record is a submission of its own under WDDM (~80 us on a 6900 XT,
+        // Windows 11): compute() marks "dequant" once per expert, which put ~16 records per MMQ group on the stream
+        // and charged the launch gaps they made to "dequant".  The same totals, a fraction of the records.
+        if (strata::prefill::timer_mark_repeats(ph.data(), used, phase)) return;
         if (used == ev.size()) {
             cudaEvent_t e = nullptr;
             cudaEventCreate(&e);

tests/prefill/timer_marks_test.cpp

@@ -0,0 +1,85 @@
+// STRATA_PREFILL_TIMING mark coalescing (PR #4006): repeated marks in the running phase add nothing to the totals.
+// No GPU: the timeline is simulated, one fake clock tick per launch.
+#include <cstdio>
+#include <vector>
+
+#include "strata/prefill/timer_marks.hpp"
+
+using strata::prefill::timer_mark_repeats;
+
+static int failed = 0;
+#define CHECK(x) do { if (!(x)) { std::fprintf(stderr, "FAIL %s:%d: %s\n", __FILE__, __LINE__, #x); ++failed; } } while (0)
+
+enum { kIdle, kDequant, kGemm, kWaitCopy, kPhases };
+
+struct Mark { int phase; double t; };
+
+// PfTimer::fold(): the gap between mark i and mark i+1 is charged to the phase of mark i.
+static std::vector<double> fold_totals(const std::vector<Mark>& marks) {
+    std::vector<double> ms(kPhases, 0.0);
+    for (size_t i = 0; i + 1 < marks.size(); ++i) ms[marks[i].phase] += marks[i + 1].t - marks[i].t;
+    return ms;
+}
+
+// PfTimer::mark(): record unless the newest recorded mark is in the same phase.
+static std::vector<Mark> record(const std::vector<Mark>& calls, size_t* skipped) {
+    std::vector<Mark> out;
+    std::vector<int> ph;
+    *skipped = 0;
+    for (const Mark& m : calls) {
+        if (timer_mark_repeats(ph.data(), ph.size(), m.phase)) { ++*skipped; continue; }
+        out.push_back(m);
+        ph.push_back(m.phase);
+    }
+    return out;
+}
+
+int main() {
+    // the predicate
+    const int ph[] = {kGemm, kDequant};
+    CHECK(!timer_mark_repeats(ph, 0, kDequant));        // nothing recorded yet: always record
+    CHECK(timer_mark_repeats(ph, 2, kDequant));         // the newest mark is already dequant
+    CHECK(!timer_mark_repeats(ph, 2, kGemm));           // an older mark's phase does not count, only the newest
+    CHECK(!timer_mark_repeats(ph, 1, kDequant));
+    CHECK(timer_mark_repeats(ph, 1, kGemm));
+
+    // a compute() like timeline: gemm, then one dequant mark per expert (16 of them), then a copy wait, then dequant again
+    std::vector<Mark> calls;
+    double t = 0;
+    calls.push_back({kGemm, t});
+    t += 5;
+    for (int e = 0; e < 16; ++e) { calls.push_back({kDequant, t}); t += 0.5 + e * 0.01; }
+    calls.push_back({kWaitCopy, t});
+    t += 3;
+    for (int e = 0; e < 4; ++e) { calls.push_back({kDequant, t}); t += 1.0; }
+    calls.push_back({kIdle, t});                         // the run's closing mark
+
+    size_t skipped = 0;
+    const std::vector<Mark> kept = record(calls, &skipped);
+    const std::vector<double> full = fold_totals(calls), coal = fold_totals(kept);
+    CHECK(skipped == 15 + 3);                            // 18 records saved of 23
+    CHECK(kept.size() == 5);                             // gemm, dequant, wait copy, dequant, idle
+    for (int p = 0; p < kPhases; ++p) CHECK(full[p] - coal[p] < 1e-9 && coal[p] - full[p] < 1e-9);   // the totals are the same
+
+    // alternating phases: nothing is skipped
+    std::vector<Mark> alt;
+    for (int i = 0; i < 10; ++i) alt.push_back({i % 2 ? kGemm : kDequant, (double) i});
+    record(alt, &skipped);
+    CHECK(skipped == 0);
+
+    // the same phase for the whole run: one record plus the closing mark
+    std::vector<Mark> same;
+    for (int i = 0; i < 100; ++i) same.push_back({kDequant, (double) i});
+    same.push_back({kIdle, 100.0});
+    const std::vector<Mark> k2 = record(same, &skipped);
+    CHECK(skipped == 99 && k2.size() == 2);
+    CHECK(fold_totals(k2)[kDequant] == fold_totals(same)[kDequant]);
+
+    // after a fold the timer keeps its last mark (used == 1): a repeat of that phase is still skipped
+    const int kept_ph[] = {kWaitCopy};
+    CHECK(timer_mark_repeats(kept_ph, 1, kWaitCopy));
+
+    if (failed) { std::fprintf(stderr, "%d check(s) failed\n", failed); return 1; }
+    std::puts("timer_marks_test: ok");
+    return 0;
+}