fix(llama-cpp): stop a generation whose stream is gone

grpc::ServerWriter::Write() returns false once the peer is gone, and
PredictStream ignored that result at every call site. The handler kept
pulling decoded tokens and writing them into a dead stream, so the
llama.cpp slot stayed busy until the generation ended on its own terms.

A model configured with max_tokens 0 and a large context ends on its own
terms only at the context limit. On a 35B model at ~41 t/s a 120k context
is about fifty minutes, and a slot held that long is a slot every other
request for that model queues behind. Two abandoned requests were enough
to make a node with free VRAM and a healthy control plane serve nothing:
new requests timed out waiting for a slot, each timeout abandoned another
generation, and the node fell further behind the longer it ran.

Track the peer instead. The first failed write retires it for good, since
a stream never recovers, and the RPC's own cancellation flag folds into
the same predicate so the loop has one condition to test. Returning early
is what frees the slot: ~server_response_reader() posts
SERVER_TASK_TYPE_CANCEL for whatever is still decoding.

TTSStream already checked Write(); this brings PredictStream in line.
Cancellation stays cooperative and is checked between decoded results, so
a batch already in flight may finish before the request stops.

Assisted-by: Claude:claude-opus-5

backend/cpp/llama-cpp/stream_peer_test.cpp

@@ -0,0 +1,67 @@
+#include "stream_peer.h"
+
+#include <cstdio>
+
+namespace {
+
+int failures = 0;
+
+void check(bool condition, const char *what) {
+    if (!condition) {
+        std::fprintf(stderr, "FAIL: %s\n", what);
+        ++failures;
+    }
+}
+
+} // namespace
+
+int main() {
+    {
+        llama_grpc::StreamPeer peer;
+        check(peer.alive(), "a fresh peer is alive");
+        check(!peer.gone(), "a fresh peer is not gone");
+    }
+
+    {
+        llama_grpc::StreamPeer peer;
+        peer.observe_write(true);
+        peer.observe_write(true);
+        check(peer.alive(), "successful writes keep the peer alive");
+    }
+
+    {
+        llama_grpc::StreamPeer peer;
+        peer.observe_write(false);
+        check(peer.gone(), "a failed write marks the peer gone");
+    }
+
+    {
+        // The whole point of the guard: a stream never comes back, so a later
+        // success must not resurrect a peer an earlier failure retired.
+        llama_grpc::StreamPeer peer;
+        peer.observe_write(false);
+        peer.observe_write(true);
+        check(peer.gone(), "a failed write is sticky across later writes");
+    }
+
+    {
+        llama_grpc::StreamPeer peer;
+        peer.observe_cancelled(false);
+        check(peer.alive(), "an uncancelled RPC keeps the peer alive");
+        peer.observe_cancelled(true);
+        check(peer.gone(), "cancellation marks the peer gone");
+    }
+
+    {
+        llama_grpc::StreamPeer peer;
+        peer.observe_cancelled(true);
+        peer.observe_cancelled(false);
+        check(peer.gone(), "cancellation is sticky across later checks");
+    }
+
+    if (failures != 0) {
+        std::fprintf(stderr, "%d check(s) failed\n", failures);
+        return 1;
+    }
+    return 0;
+}