[Engine] Slot placement policy for STRATA_SLOT_CACHE_AFFINITY (2/3, depends on #2527): protection by history length, retention scores, value decay, no wait

CMakeLists.txt

@@ -1771,6 +1771,9 @@ if(STRATA_BUILD_TESTS OR STRATA_BUILD_CONVERSATION_TESTS)
   add_executable(batch_groups_test src/program/batch_groups_test.cpp)
   target_include_directories(batch_groups_test PRIVATE include)
   add_test(NAME batch_groups_test COMMAND batch_groups_test)
+  add_executable(pipeline_rows_test tests/core/pipeline_rows_test.cpp)
+  target_include_directories(pipeline_rows_test PRIVATE include)
+  add_test(NAME pipeline_rows_test COMMAND pipeline_rows_test)
 endif()
 
 if(STRATA_CPU_X86)

docs/BATCHING.md

@@ -185,7 +185,8 @@ about 10-25% speed per request on this card". `--parallel N` is honoured as aske
   engine copies its state back (50-60 ms for a short conversation) instead of reading the history again - also for
   a client that drops the reply's thinking from the history (the checkpoint matches up to the new turn). A new
   conversation takes an empty slot, else the one used longest ago.
-- Slots are assigned so that consecutive requests land in different pipeline groups (`--batch-groups`).
+- Slots are assigned so that consecutive requests land in different pipeline groups (`--batch-groups`) - with
+  `STRATA_SLOT_CACHE_AFFINITY=1`, by the conversations they hold (below).
 - A client that disconnects stops its slot (`BSTOP`); the others go on.
 
 ### Long prompts that move from solo to a batch slot
@@ -386,6 +387,149 @@ one cold 64K prompt joining four streams: `STRATA_BATCH_DECODE_SHARE=2.0` 2.4 s
 s and 13.8 s. Pieces of one chunk (`PIECE=8192`) cost the long prompt the most. Not measured: other GPU pairs, three or
 more stages, HIP and SYCL (built for CUDA only), sampled decoding, prompts with pictures.
 
+## Keeping conversations in pipelined slots (opt-in)
+
+With a layer split, `--batch` runs the slots in pipeline groups (`--batch-groups`). A group's batch window spans its
+slots up to its last active one, and an idle slot below an active one runs as a pad row that writes its state. So in
+a pipelined group a finished slot is not a reliable conversation cache, and the server places requests by the groups
+alone. Agent clients (Claude Code, Codex) send long conversations - tens of thousands of tokens - in many short turns,
+with short side requests in between, and whatever is lost between two turns is read again.
+
+What that cost on one machine: the engine log of 3 x RTX A4000 16 GB (layer split, `--batch` 3-6 in 3 groups,
+Qwen3.8-Flash-Next IQ3_S, 200K context), two people working with Claude Code and Codex, 2026-10-09 to 2026-10-14.
+Of 3,027 prompts of 8K tokens or more, 538 reused less than half of their previous turn, and 40.7M prompt tokens
+were read again. By prompt lengths (the log has no content, so this is a heuristic):
+
+| Why the previous turn was gone | Prompts | Tokens read again |
+| --- | ---: | ---: |
+| it was in a batch slot that another conversation took | 170 | 16.4M |
+| it ran on the solo path, and the next request (often a short side request) replaced the main session | 244 | 15.0M |
+| the client changed the history (not addressed here) | 63 | 6.1M |
+| no earlier turn (a first turn, or a restart) | 66 | 3.1M |
+
+At this machine's ~2,900 tok/s for long prompts, the first two rows are about three hours of prompt reading in six
+days. What follows addresses them: the engine keeps idle slots and parks the solo path's conversation, the server
+places requests by what the slots hold, and a placement policy (`serve/slot_policy.py`) decides which history to give
+up. Each part is opt-in.
+
+### Engine: only the active slots in a window (`STRATA_PIPELINE_ACTIVE_ROWS=1`)
+
+With `STRATA_PIPELINE_ACTIVE_ROWS=1` (a layer split with `--batch` and more than one group), a group's window lists
+only its active slots (`include/strata/program/pipeline_rows.hpp`, `Verifier::batch_launch_rows`): there is no pad
+row, an idle slot keeps its conversation, and `INFO` says `slot_padding_safe=1`. After restoring a prompt's prefix the
+engine says where it came from: `CACHE_SOURCE <slot> <tokens>` (-1: the main session, a checkpoint or the
+conversation cache).
+
+Measured on the machine above (the same binary with the variable off / on, `tools/serve_load.py`, 6 clients x 2048
+tokens at temperature 0, one run each): 177.4 / 173.5 tok/s in total (-2.2%), first token p50 2.4 / 3.7 s; one client
+67.3 / 66.7 tok/s. With all six slots busy the windows hold the same rows either way; runs of this configuration with
+other builds measured 173.9-178.2 tok/s, so this is close to the spread between runs, but it was not repeated.
+
+### Engine: keep the solo path's conversation (`park=B`)
+
+A request alone runs on the solo path, whose state is the engine's main session, not a slot. When the next request is
+another conversation - an agent's short side request between two turns of a long one - the main session is
+overwritten. With `STRATA_PIPELINE_ACTIVE_ROWS=1` the engine takes `park=B` on `GEN` / `BGEN` (`INFO slot_park=1`):
+when the prompt is another conversation (it does not share what the main session holds up to its last turn - the
+last `<|im_start|>`, where its latest reply begins), it first copies the main session's conversation (and its last
+checkpoint) into idle slot B - never the slot it restores from or is admitted to - and says
+`PARKED B <tokens>`. Copying 36-37K tokens took 414-451 ms (five runs, the machine above).
+
+### Server: placement by what the slots hold (`STRATA_SLOT_CACHE_AFFINITY=1`)
+
+With `"env": {"STRATA_SLOT_CACHE_AFFINITY": "1"}` in the server config and pipelined groups, `pick_slot` asks the
+placement policy in `serve/slot_policy.py` instead of placing by the groups alone. The server side is the same for any
+policy:
+
+- The policy gets each slot's state: busy, the tokens it holds, when it was last used (counted from the request's real
+  end, `BDONE`), whether its state is intact, the spreading order, the group size, the observed prompt rate and
+  whether the engine runs active slots only (`slot_padding_safe=1`). Without that engine capability the server tracks
+  from the `BADM` / `BT` / `BDONE` lines which idle slot a later window of its group has padded over, and marks it.
+- With `INFO slot_park=1` the server tracks what the main session holds; when a request is another conversation (as
+  the engine decides it, with `INFO turn_token`), it asks the policy for a slot and sends `park=B`; `PARKED` updates that slot's state. A claim
+  the engine did not use is released.
+
+The policy's interface is `place`, `park_destination`, `cache_state`, `superseded_slots`, `reliable_flags`,
+`configure` / `settings` (their docstrings say what they answer). The policy is described below.
+
+`/metrics` shows what happened: `live.cache_events` (the last 96: allocation, admission with the tokens reused and
+read, resume and cache source, park claims and `parked`, expiries), each request's key, client
+address, slot and cache source in `requests[]`, `live.peer_ip` and `live.solo_*` (the request on the solo path and
+where its cache came from), and for slots the client of the conversation each one holds; with
+`STRATA_SLOT_CACHE_AFFINITY=1` also each idle slot's protection and scores and `live.slot_policy`; with
+`STRATA_CACHE_TRACE=1` each request's snapshot of how long a start every slot shared with its prompt (lengths only,
+`serve/cache_trace.py`). The client address is the socket's (behind a reverse proxy, the proxy's); `X-Forwarded-For`
+and request fields are not used. No prompt text or token ids are recorded.
+
+### The placement policy (`serve/slot_policy.py`)
+
+Rules alone cannot tell which history to give up when a new conversation needs a slot and none is free of value: the
+one used longest ago may be a 90K-token conversation whose user is reading the answer, the newest a side request's 300
+tokens. The policy weighs length and age, and every threshold is a parameter:
+
+- **Protection grows with length**: 0.1 s at 500 tokens, 2 s at 2K, 10 s at 8K, 45 s at 32K, 65 s at 50K, 90 s at
+  100K, 120 s at 262144 (piecewise linear; `STRATA_SLOT_TTL_MAX_S`, default 120, scales the whole curve). A side
+  request's history is free again almost at once; a long conversation outlasts the time its user reads and types.
+- **Retention score** = (n / 50000)^0.7 x (n / (n + 2000))^2 x (0.25 + 0.75 x freshness), freshness = 1 - age / TTL,
+  and 0 once expired; overwrite score = 1 / (1 + retention). A fresh 50K history is worth about 1, a 500-token one
+  0.002; a fresh 50K history outweighs a 150K one about to expire.
+- **Blank slots** - empty, known invalid, or a superseded copy (its history is the start of another idle slot's: the
+  conversation went on there) - count as expired and worth nothing.
+- **A continued conversation** (an idle slot's whole history is the start of the prompt; matches within 2% of the
+  longest, one second of the prompt rate and 4096 tokens count too) runs in its slot or in a blank slot of a less
+  loaded group - the engine copies it there, so moving costs nothing - and **never in a slot that holds another
+  conversation**, expired or not. On the machine above an earlier version that put group balance first moved two Claude
+  Code conversations, idle for minutes between turns, into each other's slot on every turn, and each turn read 60-100K
+  tokens from token 0.
+- **An expired history is not deleted**: it still matches until it is overwritten. A new conversation takes a blank
+  expired or blank slot: first one not worth keeping (blank, or its score at freshness 0 below
+  `STRATA_SLOT_EXPIRED_KEEP_SCORE`, default 0.01 - about 2.5K tokens), then by group load, a blank slot before a
+  history, and the lowest score.
+- **An older version of the same conversation** - an idle history that shares at least
+  `STRATA_SLOT_SIBLING_SHARE` (default 98%; 0: off) of its own length with the prompt but is not its start (the client
+  rewrote its end) - is replaced in place, protected or not, when no slot's history continues the prompt. Claude
+  Code's safety classifier inserts each query's new tool calls before the closing text of its transcript, so no
+  query continues the previous one; kept side by side, the versions took 4-5 of the 6 slots on the machine above and
+  pushed the other conversations out.
+- **No wait**: when every idle history is still protected, the request overwrites the one that loses least at once
+  (losses within 10% go by group load). An earlier version waited up to 5 s first, once per request; with more
+  requests than slots each wait ran its full 5 s, the slot idle meanwhile, and ended in the same overwrite (7 of 7 in
+  a 15-minute test on the machine above), so it was removed.
+- **Expired value decays**: after its protection an expired history is worth its score at freshness 0 times
+  `1 - x^2`, where x runs from 0 at the end of its protection to 1 at a total idle age of
+  `STRATA_SLOT_IDLE_VALUE_ZERO_S` (default 1800 s; 0 disables decay). Reclaiming and parking use this value; it never
+  changes whether a history can match.
+- **Park** (with `slot_padding_safe=1`): not when the main session's conversation's value at its idle age is below
+  `STRATA_SLOT_PARK_MIN_VALUE` (default 0.01, about 2.6K tokens at expiry; no separate length cut-off), or when a slot
+  holds `STRATA_SLOT_PARK_KEPT_SHARE` (default 98%) of it already. Then its own older copy, else an older version of
+  it (`STRATA_SLOT_SIBLING_SHARE` above, protected or not; the most shared first), else an idle expired slot
+  in the reclaim order above (not worth keeping first, then group load, then the lowest value); a history worth
+  keeping is taken only when the main session's conversation is worth at least `STRATA_SLOT_PARK_VALUE_RATIO` (default
+  2) times more. Never another conversation's protected slot. The parked copy keeps the main session's last-use time.
+- A busy slot is never matched or overwritten. On an engine without `slot_padding_safe=1`, a slot that would pad
+  over a protected idle history below it costs that history's retention too.
+
+Measured on the machine above (0.1.44.1, `--batch 6` in 3 groups, `STRATA_PIPELINE_ACTIVE_ROWS=1`):
+
+- Park: a 37,359-token conversation, a short side request, then its next turn. Without `park=`: 0 tokens reused, the
+  turn took 14 s; with it: 37,352 reused, about 1 s, and the same greedy output (temperature 0) as before the side
+  request.
+- Retention under load: conversation A (37K tokens) parked, two clients generating 1,500-token answers in a loop (so
+  batch mode stays on), conversations B (9K) and C (7.9K) taking turns about every 30 s - past their protection of
+  about 11 and 10 s. Tokens reused on turns 1-3:
+
+| Ordering of expired histories | B | C |
+| --- | --- | --- |
+| by group load first (an earlier version) | 0 every turn | 0 every turn |
+| + one still worth keeping goes last (this policy) | all but the ~20 new | 0 every turn |
+| + wait for a cheaper slot about to leave its protection (removed) | all but the ~20 new | all but the ~20 new |
+
+In the first row a new turn took the other conversation's expired slot for group balance, so B and C overwrote each
+other; in the second, C went to a side request that arrived a second after another one finished, while that one's
+1.5K-token history was about to leave its protection. The third row waited for that protection to end, under the
+same one-per-request wait as above; it went with that wait. The policy accepts this case: a protected history is
+not overwritten while an expired one can be.
+
 ## The VRAM expert tier keeps adapting in batch windows
 
 Batch windows count which experts they route to, and every `--adapt-every` windows (default 4) the engine swaps the
@@ -707,6 +851,10 @@ On top of `GEN` / `GENI`:
 | `BYIELD <slot>` | in | the prompt being read gives way at its next chunk boundary; its part read waits in `<slot>` (the admission's own, or a free slot for a solo request) |
 | `YIELDED <slot> <tokens>` | out | before the `DONE cancel` of a read that gave way: the request is sent again later and goes on from there |
 | `INFO ... batch_slots=N` | out | the slots the engine runs (only with `--batch`) |
+| `INFO ... slot_padding_safe=1` | out | `STRATA_PIPELINE_ACTIVE_ROWS=1`: a pipelined group's window runs its active slots only; an idle slot keeps its conversation; with it `slot_park=1` and `turn_token=<id>` (the token a turn begins with) |
+| `CACHE_SOURCE <slot> <tokens>` | out | `STRATA_PIPELINE_ACTIVE_ROWS=1`, after a prompt's restore: the idle slot its reused prefix came from (-1: the main session, a checkpoint or the conversation cache) and how long it is |
+| `park=B` (a key of `GEN` / `BGEN`) | in | with `INFO slot_park=1`: before a different conversation replaces the main session's, copy that one into idle slot B |
+| `PARKED <slot> <tokens>` | out | `park=` copied that many tokens into the slot (0: the copy failed, the slot holds nothing usable) |
 
 `tools/batch_test.py` drives the engine directly: the same prompts alone, then together, compared token by token,
 and the aggregate rate.

include/strata/core/verify.hpp

@@ -243,6 +243,8 @@ class Verifier {
     // a window in flight from one thread (batch_poll does not block).  Stage k can then run group g while stage k+1
     // runs group g-1.  Rows of group `base` use hand-off rows [base, base + S), so groups never share a hand-off row.
     bool batch_launch(int base, int S, const int32_t* tokens, const int64_t* pos, std::string& err);
+    // STRATA_PIPELINE_ACTIVE_ROWS=1: packed active rows bind actual slot states; hand-off storage still starts at base.
+    bool batch_launch_rows(int base, int S, const int* rows, const int32_t* tokens, const int64_t* pos, std::string& err);
     /// 1 = this stage's window and commit are done (the last stage's picks are in batch_out), 0 = still running,
     /// -1 = an error (err).  Serves every layer that has rung so far.
     int batch_poll(PoolMultiFn pool, void* user, std::string& err);

include/strata/program/pipeline_rows.hpp

@@ -0,0 +1,16 @@
+// STRATA_PIPELINE_ACTIVE_ROWS=1: active-row packing; unlisted slot state is never submitted to the verifier.
+#pragma once
+#include <cstdint>
+#include <cstddef>
+namespace strata::program::pipeline_rows {
+template<class Slots>
+int pack(const Slots& slots, int base, int count, int* rows, int32_t* tokens, int64_t* pos) {
+    int n = 0;
+    for (int b = base; b < base + count; ++b) {
+        const auto& slot = slots[(std::size_t) b];
+        if (!slot.active) continue;
+        rows[n] = b; tokens[n] = slot.x; pos[n] = slot.p; ++n;
+    }
+    return n;
+}
+}  // namespace strata::program::pipeline_rows

serve/cache_trace.py

@@ -0,0 +1,18 @@
+"""Metadata-only cache prefix inspection; no prompt text or token IDs are recorded."""
+from . import slot_policy
+
+def snapshot(prompt, held, busy, valid, used, now):
+    rows=[]
+    for b,h in enumerate(held):
+        hi=min(len(prompt),len(h));lo=0
+        if h[:hi]==prompt[:hi]:lo=hi
+        else:
+            while lo<hi:
+                middle=(lo+hi+1)//2
+                if h[:middle]==prompt[:middle]:lo=middle
+                else:hi=middle-1
+        state=slot_policy.cache_state(len(h),used[b],now,valid[b])
+        rows.append({"slot":b,"held_tokens":len(h),"busy":bool(busy[b]),
+                     "valid_hint":bool(valid[b]),"expired":state["expired"],
+                     "common_prefix_tokens":lo,"full_prefix_match":bool(h) and lo==len(h) and len(h)<len(prompt)})
+    return {"prompt_tokens":len(prompt),"slots":rows}