#2527 Keep conversations in pipelined batch slots (1/2, policy in #2528): active-row windows, park=, cache-aware placement behind a policy interface

Descrição

## Title
Issue: none open that I found; happy to open one first if you prefer.

## How this PR relates to #2528 (please read first)

This is **part 1 of 2**.

- **This PR**: the engine changes (active-row windows, `park=`) and the server side of cache-aware placement, with
  `serve/slot_policy.py` as an **interface only** - its policy functions raise `NotImplementedError`.
- **Part 2 (#2528)**: our placement policy - the real `serve/slot_policy.py` and the tests that need it. Its branch
  contains this PR's commits with its own on top.
- **Merge order**: this PR first, then part 2. With both merged, the code is byte for byte what we run in production.
- **If you prefer other rules than ours**: merge this PR and write your own `serve/slot_policy.py` against the
  interface instead of part 2; the server side does not change.
- **This PR alone**: nothing changes until `STRATA_SLOT_CACHE_AFFINITY=1` or `STRATA_CACHE_TRACE=1` is set, and
  those then need a policy.

## Summary

On a layer split with `--batch`, conversations are lost between turns in two ways: a pipelined group's window runs
every slot up to its last active one, so an idle slot below an active one is a pad row and loses its state; and a long
conversation on the solo path is overwritten by the next, unrelated request. With agent clients (Claude Code, Codex) -
long conversations in many short turns, short side requests in between - that meant re-reading tens of thousands of
tokens on a large share of turns.

This PR adds the mechanism, all opt-in: the engine keeps idle slots and can park the solo path's conversation, and
the server places requests by what the slots hold. **Which history to give up is left to a placement policy behind an
interface** (`serve/slot_policy.py`): this PR ships the interface only, so you can plug in the rules you prefer. Ours
is #2528; the two together are exactly what we run in production.

## The problem, measured

3 x RTX A4000 16 GB (PCIe 4.0 x16, layer split), AMD EPYC 7543, 62 GB RAM, Ubuntu; Qwen3.8-Flash-Next IQ3_S
(GSQ-RCO), 200K context, int8 KV, `--kv-resident 32768`, `--batch` 3-6 in 3 groups; two people working with Claude Code
and Codex. The engine log of 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. Classified 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 ~2,900 tok/s for long prompts on this machine, the first two rows are about three hours of prompt reading.

## What changed

1. **engine: `STRATA_PIPELINE_ACTIVE_ROWS=1`** - a pipelined group's window lists only its active slots
   (`include/strata/program/pipeline_rows.hpp`, `Verifier::batch_launch_rows` for CUDA and SYCL): no pad row, an idle
   slot keeps its conversation, `INFO slot_padding_safe=1`; after a prompt's restore, `CACHE_SOURCE <slot or -1>
   <tokens>`. `tests/core/pipeline_rows_test.cpp`, registered in CMake.
2. **engine: `park=B`** (`INFO slot_park=1 turn_token=<id>`, with the variable above) - when a prompt is another
   conversation (it does not share what the main session holds up to its last turn, the last `<|im_start|>`: two chats
   of one client share a long system prompt, a next turn without the reply's thinking shares the history), the engine
   first copies the main session's conversation into idle slot B and says `PARKED B <tokens>`. The server's claim uses
   the same test.
3. **server: `STRATA_SLOT_CACHE_AFFINITY=1`** - with pipelined groups `pick_slot` asks `slot_policy.place()`, which
   always answers a free slot (a request waits only when every slot is busy, as before); the server tracks the main
   session and asks `slot_policy.park_destination()` before sending `park=`; without `slot_padding_safe=1` it tracks
   from `BADM`/`BT`/`BDONE` which idle slot a later window has padded over. `/metrics` gains `live.cache_events` (the
   last 96 placement events), request keys and client addresses (the socket's, not `X-Forwarded-For`), `live.solo_*`
   and per-slot cache fields; `STRATA_CACHE_TRACE=1` adds a per-request snapshot of what each slot shared with the
   prompt (`serve/cache_trace.py`, lengths only).

The interface (`serve/slot_policy.py`): `place`, `park_destination`, `cache_state`, `superseded_slots`,
`reliable_flags`, `configure` / `settings` - their docstrings state the
contract. Without a policy, `STRATA_SLOT_CACHE_AFFINITY=1` and `STRATA_CACHE_TRACE=1` raise `NotImplementedError`.

With every variable unset the engine's output and the generation path are unchanged. The new `/metrics` fields are
present with the variables unset too (additive only); if you prefer them behind a variable, say so.

## Results (same machine, with our policy from #2528)

| | Without | With |
| --- | ---: | ---: |
| 37,359-token conversation, next turn after a side request: tokens reused | 0 | 37,352 |
| ... time of that turn | 14 s | about 1 s |
| parking 36-37K tokens (five runs) | - | 414-451 ms |

The greedy output (temperature 0) after the restore was the same as before the side request. Alternating
conversations under load: see #2528.

Throughput of commit 1 (the same binary with `STRATA_PIPELINE_ACTIVE_ROWS` off / on, `tools/serve_load.py`, 6 clients
x 2048 tokens, temperature 0, one run each): 177.4 / 173.5 tok/s total, 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; other runs of this configuration with other builds
measured 173.9-178.2 tok/s, so this is close to the run-to-run spread, but it was not repeated.

## How it was checked

- Each commit: `generate.cpp` and `verify.cpp` compile with the CUDA build's flags (sm_86, Linux);
  `tests/core/pipeline_rows_test` passes; `python -m unittest discover -s serve -t .` passes except
  `serve.test_responses.NoLeakedSampler`, which fails on upstream main on this machine too (5 of 6 runs).
- Not built: SYCL (no oneAPI here) and HIP; the SYCL change mirrors the CUDA one.
- This code runs in production on the machine above since 2026-10-15 (with part 2 on top).

## Extra Notes

A client that rewrites history (for example one that drops the reasoning of earlier turns) cannot match a cached
prefix whatever the placement does; not addressed.

🤖 Generated with [Claude Code](https://claude.com/claude-code)