feat(chat): settle a cancelled streamed generation on the text it posted

A worker that follows the cancel is never penalised for it, and a client cannot work a volunteer for free: the Horde measures the turned-in work through the progress posts, so the cancel pays and charges on it.

- The native cancel route settles each in-flight generation of a streamed chat request through `TextProcessingGeneration.settle_cancelled`: the row is locked and reloaded, the content the worker posted (`read_posted_progress`, joined from the retained stream in `seq` order, reasoning excluded, the lease's byte count when the stream lost the entries) is priced by the output estimate, and `ProcessingGeneration.cancel` pays the worker and charges the requester once without a dropped job, whichever of submit or cancel commits first; a late submit is refused as `AbortedGen`
- `processing_gens.settled_character_count` records the count the settlement priced, written before `cancel` commits, so cancels can be analysed while the posted text is not kept; `get_things_count` reads it after a stored reply, so a stored reply stays authoritative; the status route does not show it
- `cancel_progress_streams` sets `cancelled` on every lease that still exists in one watched transaction and appends each generation's ended marker, best effort, so a cancel succeeds without Redis and the leases stay for the acknowledgement
- `chat_progress.lua` returns the lease's `cancelled` flag with `accepted` and `duplicate` replies; the progress route's 200 body (`ChatProgressAccepted`) carries it beside `accepted_seq`, so a worker stops generating and does not submit
- Requests that do not stream are paid in full on submit, since the Horde does not measure partial work for them; maintenance mode's abort does not flag leases, since it lets in-flight jobs settle
- Tests cover the flag, the ended marker and the kept lease, the ordered join and the byte fallback, a Redis failure, payment and charge on posted content, the floor before any post, the refused late submit, the untouched stale pass, the recorded count and the unchanged non-streaming cancel; the references, explanation and how-to describe the acknowledgement and the settlement
- The explanation shows the progress lease's states and the settlement of a cancel, `docs/how-to/serve_chat_jobs.md` opens with the worker's pop loop as a diagram keyed to its steps, and `docs/explanation/kudos_accounting.md` describes the settlement of a cancelled streamed generation

docs/explanation/chat_requests.md

@@ -299,6 +299,8 @@ each tool call's name and arguments) divided by four, floored at 20 tokens and c
 ([ADR 0027](../decisions/0027-price-chat-generations-by-the-text-output-estimate.md),
 [Pricing](../reference/chat_requests.md#pricing)). A short reply to a large `max_tokens` is charged for the short
 reply. The basis is computed in one function and stored on every generation as `charge_basis: output_estimate`.
+A streamed generation the client cancels is settled on the same basis on the content its worker posted before the
+cancel, which the Horde measured, so a volunteer worker is paid for what it turned in.
 
 Pricing on the token counts workers report was the alternative
 ([ADR 0028](../decisions/0028-price-chat-generations-on-reported-completion-tokens.md)). It is fairer per reply and
@@ -401,9 +403,33 @@ the lowest kudos priority, and per-source caps on the front, which aggregate use
 A streamed request without tools whose `response_format` is absent or `text` gets its content as the worker posts
 progress, one chunk per post, about one a second. The batches are kept in a retained Redis stream per request, cheap
 and lost on failover
-([ADR 0024](../decisions/0024-stream-chat-deltas-through-a-retained-redis-stream.md)); the alternative that was not
-built is Postgres progress rows, durable and paid for in primary database writes on the connection pool pops and
-submits share ([ADR 0025](../decisions/0025-stream-chat-deltas-through-postgres-progress-rows.md)).
+([ADR 0024](../decisions/0024-stream-chat-deltas-through-a-retained-redis-stream.md)). Postgres progress rows were the
+alternative, durable and paid for in primary database writes on the connection pool pops and submits share
+([ADR 0025](../decisions/0025-stream-chat-deltas-through-postgres-progress-rows.md)).
+
+A lease in Redis, written at pop, authenticates a generation's posts, so the progress route does not read Postgres.
+The lease's life is the streamed generation's life as the worker sees it
+([Streaming progress](../reference/chat_requests.md#streaming-progress)):
+
+```mermaid
+stateDiagram-v2
+    [*] --> Leased: pop of a streamed job
+    Leased --> Leased: post accepted (200)<br/>or resynced after 409
+    Leased --> Submitted: worker submits the whole reply
+    Leased --> Ended: generation faults
+    Leased --> Cancelled: client or front cancels
+    Leased --> PostsStopped: 413 cap reached
+    PostsStopped --> Submitted: worker still submits
+    Submitted --> [*]: lease deleted, stream left to expire
+    Ended --> [*]: ended marker, lease deleted
+    Cancelled --> [*]: settled on the posted text<br/>next ack carries cancelled true<br/>worker stops and does not submit
+```
+
+A cancel settles the generation on the content its worker posted before it, so the worker is paid for that text and
+the requester charged for it. The settlement does not count a dropped job, so a worker that stops on the flag is
+never penalised for a client leaving. The front relays entries as they arrive and reconciles them with the final
+reply: a reply equal to or extending the relayed text completes the stream, and one that diverges ends it with an
+error event.
 
 A request with tools or a non-text `response_format` waits for the whole reply as a non-streamed request does, then
 sends it as

docs/explanation/kudos_accounting.md

@@ -166,6 +166,13 @@ business transaction commits its job state and emitted facts together. The event
 it does not by itself make every producer retry idempotent. Transfers accept a caller idempotency key and validate a
 replay. Other producers still depend on their existing job-state transaction to prevent duplicate settlement.
 
+A streamed chat generation that its requester cancels is settled through the same event, on the content its worker
+posted before the cancel, which the Horde measured. The worker owner is credited and the requester debited for that
+content, and the cancel does not count as a dropped job for the worker. A later submit for the settled generation is
+refused, so the generation settles once. The Horde does not measure part of a non-streamed generation, so it settles
+on submit
+([Chat requests: pricing](chat_requests.md#pricing-by-the-text-output-estimate)).
+
 ## Projection: durable work queue
 
 In ledger mode, rows start with `applied = false`. The periodic projector:

docs/how-to/send_chat_completions.md

@@ -479,7 +479,8 @@ The front differs from OpenAI on purpose in these points:
 
 A client request does not change Horde state beyond the request itself. Point the client's base URL back at its
 previous provider to stop using the Horde. A request the client abandons by closing the connection is cancelled on
-the Horde.
+the Horde. A streamed request cancelled while a worker generates is charged for the content the worker streamed
+before the cancel, priced as a reply of that length.
 
 ## Related
 

docs/how-to/serve_chat_jobs.md

@@ -37,6 +37,38 @@ For why chat travels this way, see [Chat requests on the Horde](../explanation/c
 requests as a client, see [Send chat completions through the OpenAI-compatible front](send_chat_completions.md).
 To run this end to end with koboldcpp for the first time, follow [Serve your first chat job with koboldcpp](../tutorial/first_chat_worker.md).
 
+The steps below build one pop loop. A pass through it, with the step that covers each part:
+
+```mermaid
+flowchart TD
+    P[Pop with the chat declaration<br/>step 1] --> K{chat: true?<br/>step 2}
+    K -->|no| T[Serve the text job<br/>step 8]
+    K -->|yes| M{messages_url?}
+    M -->|yes| FM[Fetch the messages<br/>within 120 s, step 5]
+    M -->|no| IM{Image URLs?}
+    IM -->|yes| FI[Fetch each image<br/>step 5]
+    IM -->|no| R
+    FM --> R[Render and fit the conversation<br/>steps 3 and 4]
+    FI --> R
+    R -->|does not fit, cannot render| F[Fault with a reason<br/>step 7]
+    R --> G[Generate]
+    G --> S{stream_progress: true?}
+    S -->|yes| PP[Post progress about once a second<br/>step 6]
+    PP -->|cancelled true| X[Stop generating<br/>and do not submit]
+    PP -->|404, 413, repeated 503| G2[Stop posting, keep generating]
+    PP -->|generation finished| SB
+    G2 --> SB
+    S -->|no| SB[Submit the structured reply<br/>step 6]
+    G -->|backend fails| F
+    SB -->|400 refusal| F
+    SB --> P
+    F --> P
+    T --> P
+    X --> P
+```
+
+A failed fetch is a fault too, sent without `fault_reason`.
+
 ## 1. Declare chat on pop
 
 Add five fields to every pop body, beside the usual `name`, `models`, `max_length` and `max_context_length`
@@ -349,7 +381,10 @@ apikey: <worker owner's key>
 
 - `seq` is 1 for the job's first post and one more for each post after it. Send a post only after the previous one
   was answered. A lost answer is retried with the same `seq` and content, which returns the same
-  `{"accepted_seq": n}` and does not append an entry.
+  `{"accepted_seq": n, "cancelled": false}` and does not append an entry.
+- `"cancelled": true` in a `200` means the client cancelled the request: stop generating and do not submit. The
+  delta in that post was accepted and is not needed again. The job was settled at the cancel on the content
+  posted before it, so the worker is paid for that content without a dropped job.
 - `409 ChatProgressOutOfSequence` carries `expected_seq`. Changed content under an accepted `seq` is refused the same
   way, so never rewrite text already posted.
 - `404 ChatProgressNotStreaming`, `413 ChatProgressCapReached` (1000 posts, or 16 bytes per `max_length` token) and
@@ -361,7 +396,7 @@ apikey: <worker owner's key>
 - Submit the whole reply as above once the last post was answered. The submit is what the request stores and charges,
   and it ends the job's posts.
 
-Verification: each post returns `200 {"accepted_seq": n}`, and the request's stream
+Verification: each post returns `200 {"accepted_seq": n, "cancelled": false}`, and the request's stream
 `horde:chat:stream:<request-id>` on the chat signal database (Redis logical database 7) holds one entry per accepted
 post with `gen`, `seq`, `content`, `reasoning_content` when the post carried reasoning, `worker` and `model`. Covered
 by `tests/integration/test_chat_progress.py` and `tests/unit/test_chat_progress.py`; the contract is in
@@ -405,10 +440,12 @@ after `400 ChatSubmitInvalid` or `400 ChatModelMismatch` counts as an aborted jo
 <HORDE_URL>/api/v2/workers/name/<worker-name>` shows `nonconformant_replies`; a count that keeps rising means the
 backend does not keep a guarantee the worker declares.
 
-A job the requester cancels while the worker holds it is settled at the cancellation, as a text job is: the worker is
-rewarded and the requester charged for dispatched work
-([ADR 0027](../decisions/0027-price-chat-generations-by-the-text-output-estimate.md)). A submit for a cancelled job,
-or for one whose TTL ran out, returns `400 AbortedGen`.
+A streamed job the requester cancels while the worker holds it is settled at the cancellation on the content the
+worker posted, priced by the output estimate with its floor of 20 tokens: the worker is rewarded and the requester
+charged for that content, and the job does not count as dropped
+([ADR 0027](../decisions/0027-price-chat-generations-by-the-text-output-estimate.md)). A submit for a settled job,
+or for one whose TTL ran out, returns `400 AbortedGen`. A job that does not stream is not settled at the cancel;
+submit it as usual and it is rewarded for the reply.
 
 Verification: the faulted submit returns `{"reward": 0}`, and the next pop by any eligible worker receives a new job
 for the same request. Covered by `test_a_faulted_chat_submit_needs_neither_generation_nor_message` and

docs/reference/chat_front.md

@@ -176,7 +176,7 @@ reasoning, and does not send it otherwise. After exposure:
 | Final reply equal to the relayed text | The finish chunk, the usage chunk when asked for, and `[DONE]` |
 | Final reply starting with the relayed text | The rest as one content chunk, then the finish chunk |
 | Final reply not starting with the relayed text | SSE error event `ChatStreamDiverged`, then `[DONE]` |
-| An entry of another `gen`, the pinned `gen`'s `ended` marker, or a `seq` out of order | SSE error event `ChatStreamDiverged`, then `[DONE]`; the front cancels the request |
+| An entry of another `gen`, the pinned `gen`'s `ended` marker (also written when the request is cancelled through the native route), or a `seq` out of order | SSE error event `ChatStreamDiverged`, then `[DONE]`; the front cancels the request |
 | The request faults | SSE error event `ChatRequestFaulted`, as for a buffered stream |
 
 A restarted attempt starts the reply over, so it is never spliced into a choice the client holds part of

docs/reference/chat_requests.md

@@ -618,7 +618,10 @@ jobs timed out.
 A request is `done` when it is `faulted`, or when no job is waiting or processing and either all requested results
 are stored or the remaining jobs were cancelled. Cancellation retains partial results and lets in-flight jobs finish,
 but faults after cancellation do not restart jobs. A cancelled request can therefore be `done` without a generation;
-the front reports `ChatRequestNoGeneration` when it has no reply to deliver.
+the front reports `ChatRequestNoGeneration` when it does not have a reply to deliver. The cancel route also tells a worker
+streaming progress for the request to stop (see [Streaming progress](#streaming-progress)). For a streamed
+request it settles each in-flight generation on the content its worker posted (see [Pricing](#pricing)); the
+generation is faulted as cancelled, does not count toward the worker's aborted-job threshold and is not restarted.
 
 Faulted generations do not count as results. A job's fault and its restart are committed together, as are a pop's
 taken job and its first generation. A worker-reported image fault that is not restarted and leaves nothing waiting
@@ -743,6 +746,21 @@ submitted, `native_finish_reason`, `prompt_tokens`, `completion_tokens`, `reason
 `queue_wait_seconds`, the seconds from
 the request's creation to the generation's assignment. The text status route does not return any of them.
 
+A streamed generation still in flight when the client cancels is settled at the cancel on the content its worker
+posted, under the same output estimate: `read_posted_progress` (`horde/chat_progress.py`) joins the generation's
+`content` entries from the request's stream in `seq` order, reasoning aside, and
+`TextProcessingGeneration.settle_cancelled` prices those characters, pays the worker, charges the requester with
+the usual extra burn and faults the generation as cancelled. The posted content is measured by the Horde, so it is
+a charge input where a worker-reported count is not. A cancel before any post pays the floor of 20 tokens. When
+the stream does not hold the generation's entries, the lease's accepted `bytes` stand in for the characters;
+they count a multibyte character more than once and include reasoning, so the rare fallback leans toward the
+worker. A Redis failure settles on the floor. The settlement records the character count it priced in the
+generation's `settled_character_count`, which is null on every other generation and is not shown on the status
+route. The posted content is not stored as the generation's reply, so the
+status does not list a generation for it, and `charge_basis` and `finish_reason` stay null on the cancelled
+generation. The Horde does not measure partial work on a request that does not stream: its worker submits after the
+cancel and is paid for the reply as usual.
+
 ## Chat model list
 
 `GET /v2/status/models?type=text&chat=true` lists only the text models that at least one online chat worker serves,
@@ -780,8 +798,9 @@ still submits the whole reply to the submit route.
 | `delta.content` | The text since the previous post; empty only when `delta.reasoning_content` is set        |
 | `delta.reasoning_content` | Optional: the reasoning since the previous post, never part of `delta.content`  |
 
-A delta with neither non-empty `content` nor non-empty `reasoning_content` returns `400 BadRequest`. Reasoning counts
-toward the byte cap with the content. An accepted post returns `200` with `{"accepted_seq": <seq>}`. A post that
+A delta whose `content` and `reasoning_content` are both empty or absent returns `400 BadRequest`. Reasoning counts
+toward the byte cap with the content. An accepted post returns `200` with `{"accepted_seq": <seq>, "cancelled":
+<bool>}`; `cancelled` is `true` once the request was cancelled. A post that
 repeats the last accepted `seq` with the same content and reasoning returns the same body and does not append an entry, so a worker retries a post whose answer it lost by sending it
 again unchanged.
 
@@ -796,7 +815,7 @@ again unchanged.
 
 When a generation carrying the hint is assigned, the pop writes the lease `horde:chat:lease:<generation id>` on the
 chat signal database: a hash of `request`, `worker`, `model`, `owner` (the owner's user id), `key` (a SHA-256 of the owner's
-stored key hash), `max_bytes` and `fake`. It expires `CHAT_PROGRESS_LEASE_GRACE_SECONDS` (90) after the job's TTL,
+stored key hash), `max_bytes`, `fake` and, once the request is cancelled, `cancelled` (`1`). It expires `CHAT_PROGRESS_LEASE_GRACE_SECONDS` (90) after the job's TTL,
 which covers the stale-generation thread's 60 second tick. The route reads only the lease and does not read Postgres; a key reset
 during the job therefore takes effect for posts only once the lease ends. A fake generation's posts are answered as a
 real one's and not appended.
@@ -818,8 +837,19 @@ one. Caps, in `horde/chat_progress.py`:
 A successful submit deletes the lease and leaves the stream to expire, so a reader reconciles what it relayed with
 the final reply. When a generation with a lease faults, on a faulted submit or in the stale-generation thread, the
 stream gets an entry with `gen` and `ended: 1` and the lease is deleted. The route never publishes on the
-completion signal channel. Writing and deleting the lease and the `ended` entry are best effort: a Redis failure is
-logged and never fails the pop, submit or fault.
+completion signal channel.
+
+When the cancel route claims a streamed request's cancellation, it sets `cancelled` on the lease of each generation
+still in flight and appends that generation's `gen` and `ended: 1` entry, in one transaction that writes only leases
+that still exist. The leases are kept, so the worker's next post is still accepted, and its `200` carries `cancelled:
+true`; the delta it carries was generated before the worker read the flag and is harmless after the ended marker.
+The worker then stops generating and does not submit. Before flagging the leases, the route settles each of those
+generations on the content the stream holds for it (see [Pricing](#pricing)), so a submit that arrives after the
+cancel, from a worker that did not post again, returns `400 AbortedGen`. Maintenance mode's abort does not flag the leases, since it
+lets in-flight jobs settle. A fake generation's lease is flagged without an `ended` entry.
+
+Writing, flagging and deleting the lease and the `ended` entry are best effort: a Redis failure is logged and never
+fails the pop, submit, fault or cancel.
 
 ## Completion signal
 
@@ -859,8 +889,9 @@ false), `accepts_prompt_jobs` (boolean, not null, default true), `chat_features`
 `chat_conformance` (varchar, not null, default `strict`), `chat_tools_temperature` (double precision) and
 `chat_messages_offloaded` (boolean, not null, default false); and
 `workers.chat_tools_max_temperature` (double precision); and `waiting_prompts.chat_fault_reason` (varchar 1000) and
-`chat_fault_rc` (varchar 32), `processing_gens.chat_refusal_rc` (varchar 32) and `workers.nonconformant_replies`
-(integer, not null, default 0). Every column is nullable or has a constant default, so no table is rewritten. The
+`chat_fault_rc` (varchar 32), `processing_gens.chat_refusal_rc` (varchar 32) and `settled_character_count`
+(integer), and `workers.nonconformant_replies`
+(integer, not null, default 0). Every column is nullable or has a constant default, so the migration does not rewrite a table. The
 unique partial index `ix_waiting_prompts_user_operation_key` on `(user_id, operation_key) WHERE operation_key IS NOT
 NULL` is built with `CREATE UNIQUE INDEX CONCURRENTLY`, which cannot run inside a transaction block; a failed
 build leaves an invalid index. `db.create_all()` does not add columns to existing tables.

horde/apis/models/kobold_v2.py

@@ -1182,6 +1182,11 @@ class TextModels(v2.Models):
                 "accepted_seq": fields.Integer(
                     description="The seq the Horde holds, the posted one; also for an identical retry of the last one.",
                 ),
+                "cancelled": fields.Boolean(
+                    description=(
+                        "True once the request was cancelled. The delta is still accepted; the worker stops generating and does not submit."
+                    ),
+                ),
             },
         )
         self.response_model_chat_progress_out_of_sequence = api.model(

horde/apis/v2/kobold.py

@@ -44,6 +44,8 @@ from horde.chat_progress import (
     ProgressOutcome,
     api_key_digest,
     append_progress,
+    cancel_progress_streams,
+    read_posted_progress,
     read_stream_lease,
 )
 from horde.chat_signal import ChatEvent, publish_chat_event
@@ -78,6 +80,7 @@ from horde.classes.kobold.chat_request import (
     replace_image_urls,
     rewrite_message_texts,
     serialize_messages,
+    streams_progress,
     verify_chat_reply,
 )
 from horde.classes.kobold.genstats import (
@@ -641,6 +644,14 @@ class TextAsyncStatus(Resource):
             return (wp_status, 200)
         if cancellation_claimed:
             record_request_cancellation_forecast(request_id=str(wp.id), cancelled_at=cancelled_at)
+            if wp.is_chat and streams_progress(wp.chat_extras):
+                # The status above loaded the generations; the ones still pending may hold a lease whose worker reads
+                # the cancel from its next progress acknowledgement. Each is settled on what it posted before its lease
+                # is flagged, so the worker is paid for the content the stream holds and the user is charged for it.
+                pending_procgens = [procgen for procgen in wp.processing_gens if not procgen.is_completed() and not procgen.is_faulted()]
+                for procgen in pending_procgens:
+                    procgen.settle_cancelled(read_posted_progress(str(procgen.id), str(wp.id)))
+                cancel_progress_streams([str(procgen.id) for procgen in pending_procgens], str(wp.id))
             if wp.is_chat:
                 publish_chat_event(str(wp.id), ChatEvent.CANCELLED, wp.count_stored_generations())
         return (wp_status, 200)
@@ -1595,6 +1606,9 @@ class TextJobProgress(Resource):
         `seq` starts at 1 and goes up by one per post. A retry sends the same `seq` and content and is acknowledged
         again without appending it twice.
 
+        The acknowledgement carries `cancelled`, true once the request was cancelled. The delta is still accepted;
+        the worker stops generating and does not submit.
+
         Refused with:
 
         - `ChatProgressNotStreaming` (404): the generation is not streamed, already finished or faulted, expired or
@@ -1631,7 +1645,7 @@ class TextJobProgress(Resource):
             raise e.ServiceUnavailable("The progress store is unavailable.", rc=CHAT_PROGRESS_UNAVAILABLE_RC) from None
         metrics.chat_progress_outcomes.add(1, {"horde.outcome": appended.outcome.value})
         if appended.outcome in (ProgressOutcome.ACCEPTED, ProgressOutcome.DUPLICATE):
-            return {"accepted_seq": appended.seq}, 200
+            return {"accepted_seq": appended.seq, "cancelled": appended.cancelled}, 200
         if appended.outcome is ProgressOutcome.GAP:
             return (
                 {

horde/chat_progress.lua

@@ -4,8 +4,9 @@
 -- ARGV[5]: the stream expiry in seconds; ARGV[6]: the approximate MAXLEN the stream is trimmed to; ARGV[7]: the
 -- reasoning content, empty when the delta carries none.
 -- Returns {outcome, seq}: the accepted seq for 'accepted' and 'duplicate', the expected seq for 'gap', the refused
--- seq for 'over_cap' and 0 for 'no_lease'. Sequence state (seq, bytes, last, last_reasoning) lives in the lease hash
--- and expires with it.
+-- seq for 'over_cap' and 0 for 'no_lease'. 'accepted' and 'duplicate' carry a third element, 1 when the request was
+-- cancelled and 0 otherwise. Sequence state (seq, bytes, last, last_reasoning) lives in the lease hash and expires
+-- with it.
 local lease_key = KEYS[1]
 local stream_key = KEYS[2]
 local seq = tonumber(ARGV[2])
@@ -14,10 +15,16 @@ local reasoning_content = ARGV[7]
 if redis.call('EXISTS', lease_key) == 0 then
     return {'no_lease', 0}
 end
+-- A cancelled lease still takes the delta, since the worker may have generated it before it read the flag; the reply
+-- carries the flag so the worker stops.
+local cancelled = 0
+if redis.call('HGET', lease_key, 'cancelled') == '1' then
+    cancelled = 1
+end
 local last_seq = tonumber(redis.call('HGET', lease_key, 'seq') or '0')
 if last_seq > 0 and seq == last_seq and redis.call('HGET', lease_key, 'last') == content
         and (redis.call('HGET', lease_key, 'last_reasoning') or '') == reasoning_content then
-    return {'duplicate', last_seq}
+    return {'duplicate', last_seq, cancelled}
 end
 if seq ~= last_seq + 1 then
     return {'gap', last_seq + 1}
@@ -49,4 +56,4 @@ if redis.call('HGET', lease_key, 'fake') ~= '1' then
     redis.call('EXPIRE', stream_key, ARGV[5])
 end
 redis.call('HSET', lease_key, 'seq', seq, 'bytes', total_bytes, 'last', content, 'last_reasoning', reasoning_content)
-return {'accepted', seq}
+return {'accepted', seq, cancelled}

horde/chat_progress.py

@@ -16,11 +16,16 @@ delta; an entry appended under a lease without a model leaves it out. The lease
 accepted ``seq``, the bytes accepted so far and the last content and reasoning, so the check does not need a second
 key or a read of the stream. When a generation with a lease faults,
 [`end_progress_stream`][horde.chat_progress.end_progress_stream] appends ``gen`` and ``ended`` (``1``) and deletes the
-lease, so a reader learns at once that the attempt it follows is gone.
+lease, so a reader learns at once that the attempt it follows is gone. When the request is cancelled,
+[`cancel_progress_streams`][horde.chat_progress.cancel_progress_streams] sets ``cancelled`` on each lease it still
+has and appends the same marker for each, and keeps the leases, so the acknowledgement of the worker's next post
+tells the worker to stop. Before that, [`read_posted_progress`][horde.chat_progress.read_posted_progress] reads the
+content each pending generation posted, which the cancel settles the generation on.
 
 Unlike [`horde.chat_signal`][horde.chat_signal], whose messages never carry text, the stream holds generated text. It
-lives on the chat signal database, which the front reads, and expires on its own. Writing the lease, deleting it and
-the ended marker are best effort: a Redis failure is logged and never fails the pop, submit or fault that wrote them.
+lives on the chat signal database, which the front reads, and expires on its own. Writing the lease, deleting it,
+flagging it cancelled and the ended marker are best effort: a Redis failure is logged and never fails the pop, submit,
+fault or cancel that wrote them.
 """
 
 from __future__ import annotations
@@ -28,6 +33,7 @@ from __future__ import annotations
 import enum
 import hashlib
 import threading
+from collections.abc import Sequence
 from dataclasses import dataclass
 from importlib.resources import files
 
@@ -102,6 +108,10 @@ LEASE_FAKE_FIELD = "fake"
 """Lease field that is ``1`` for a fake generation, whose posts are acknowledged and not appended so the worker
 cannot tell it apart from a real one."""
 
+LEASE_CANCELLED_FIELD = "cancelled"
+"""Lease field that is ``1`` once the request was cancelled; absent before. The append script copies it into the
+acknowledgement so the worker stops generating."""
+
 LEASE_SEQ_FIELD = "seq"
 """Lease field the append script keeps the last accepted ``seq`` in; absent before the first post."""
 
@@ -193,6 +203,25 @@ class ProgressAppend:
     seq: int
     """The accepted ``seq`` for an accepted or duplicate post; the ``seq`` the generation expects next otherwise."""
 
+    cancelled: bool = False
+    """Whether the request was cancelled, for an accepted or duplicate post; False for a refused one."""
+
+
+@dataclass(frozen=True)
+class PostedProgress:
+    """Represents the content a generation posted before its request was cancelled, as the server measured it."""
+
+    content: str
+    """The posted content joined in ``seq`` order; empty when the stream no longer holds the generation's entries."""
+
+    chars: int
+    """The characters the generation is settled on: ``len(content)`` when the stream holds the entries, else the
+    lease's accepted bytes."""
+
+
+EMPTY_POSTED_PROGRESS = PostedProgress(content="", chars=0)
+"""The posted progress of a generation that posted nothing, or whose progress could not be read."""
+
 
 APPEND_PROGRESS_SCRIPT: str = files(__package__).joinpath("chat_progress.lua").read_text(encoding="utf-8")
 """The Lua source of the progress append, read from ``chat_progress.lua`` beside this module so that it is
@@ -202,7 +231,8 @@ written and reviewed as Lua. It appends one delta under the generation's lease.
 [`CHAT_PROGRESS_STREAM_TTL_SECONDS`][horde.chat_progress.CHAT_PROGRESS_STREAM_TTL_SECONDS] and
 [`CHAT_PROGRESS_STREAM_MAX_ENTRIES`][horde.chat_progress.CHAT_PROGRESS_STREAM_MAX_ENTRIES] and the reasoning content,
 empty when the delta carries none. Replies with the outcome
-and a ``seq``, as [`ProgressAppend`][horde.chat_progress.ProgressAppend] holds them. The sequence state lives in the
+and a ``seq``, and the ``cancelled`` flag for an accepted or duplicate post, as
+[`ProgressAppend`][horde.chat_progress.ProgressAppend] holds them. The sequence state lives in the
 lease hash, so the script reads one key besides the stream it writes and expires with the lease."""
 
 _client_lock = threading.Lock()
@@ -377,8 +407,12 @@ def append_progress(lease: StreamLease, seq: int, content: str, reasoning_conten
             reasoning_content,
         ],
     )
-    outcome, reply_seq = reply
-    return ProgressAppend(outcome=ProgressOutcome(outcome), seq=int(reply_seq))
+    outcome, reply_seq, *reply_flags = reply
+    return ProgressAppend(
+        outcome=ProgressOutcome(outcome),
+        seq=int(reply_seq),
+        cancelled=bool(reply_flags) and int(reply_flags[0]) == 1,
+    )
 
 
 def end_progress_stream(generation_id: str, request_id: str) -> None:
@@ -420,6 +454,98 @@ def end_progress_stream(generation_id: str, request_id: str) -> None:
         logger.warning(f"Unable to end the delta stream of chat generation {generation_id}: {redis_error}")
 
 
+def cancel_progress_streams(generation_ids: Sequence[str], request_id: str) -> None:
+    """Flag the leases of a cancelled request's in-flight generations and mark each attempt as ended on its stream.
+
+    One call covers every generation of the request, so the flags and markers land in one transaction. A generation
+    without a lease finished, faulted or was not streamed, and is left as it is; a lease is flagged only while it
+    exists, since setting a field on a deleted lease would recreate it without its fields or TTL. Each flagged lease
+    is kept until it is released or expires, so the worker's next post is acknowledged with ``cancelled``. A fake
+    generation's lease is flagged without a marker, since its posts were never appended. Logs rather than raises on a
+    Redis failure, since a cancel succeeds without Redis; the worker then generates and submits as it would without
+    the cancel.
+
+    Args:
+        generation_ids: The request's in-flight generations.
+        request_id: The cancelled chat request.
+
+    Side Effects:
+        Sets ``cancelled`` on each existing lease, and appends ``gen`` and ``ended`` to the stream and refreshes its
+        TTL for each that is not fake.
+    """
+    if not generation_ids:
+        return
+    lease_keys = [chat_lease_key(generation_id) for generation_id in generation_ids]
+    stream_key = chat_stream_key(request_id)
+
+    def flag_live_leases(pipeline: redis.client.Pipeline) -> None:
+        # The leases are watched, so a submit or fault that deletes one between this read and the write retries the
+        # transaction, and a deleted lease is not recreated.
+        fake_flags = [pipeline.hget(lease_key, LEASE_FAKE_FIELD) for lease_key in lease_keys]
+        pipeline.multi()
+        for generation_id, lease_key, fake in zip(generation_ids, lease_keys, fake_flags, strict=True):
+            if fake is None:
+                continue
+            pipeline.hset(lease_key, LEASE_CANCELLED_FIELD, 1)
+            if fake != "1":
+                pipeline.xadd(
+                    stream_key,
+                    {
+                        STREAM_GENERATION_FIELD: generation_id,
+                        STREAM_ENDED_FIELD: 1,
+                    },
+                    maxlen=CHAT_PROGRESS_STREAM_MAX_ENTRIES,
+                    approximate=True,
+                )
+                pipeline.expire(stream_key, CHAT_PROGRESS_STREAM_TTL_SECONDS)
+
+    try:
+        _get_client().transaction(flag_live_leases, *lease_keys)
+    except redis.RedisError as redis_error:
+        logger.warning(f"Unable to cancel the delta stream of chat request {request_id}: {redis_error}")
+
+
+def read_posted_progress(generation_id: str, request_id: str) -> PostedProgress:
+    """Return the content a generation posted, read from the request's stream.
+
+    Reasoning is left out, since a generation is paid on its content alone. When the stream holds no entry for the
+    generation and its lease still exists, the lease's accepted bytes stand in for the characters. The bytes count
+    multibyte characters more than once and include any reasoning, so the fallback leans toward the worker; the stream
+    outlives the lease by design, so the fallback applies only when the stream was trimmed or lost.
+
+    Args:
+        generation_id: The pending generation of the cancelled request.
+        request_id: The chat request whose stream holds the generation's entries.
+
+    Returns:
+        The posted content and its character count, or
+        [`EMPTY_POSTED_PROGRESS`][horde.chat_progress.EMPTY_POSTED_PROGRESS] when the generation posted nothing or
+        Redis could not be reached, which a cancel survives by settling on the formula's floor.
+    """
+    pipeline = _get_client().pipeline(transaction=False)
+    pipeline.xrange(chat_stream_key(request_id))
+    pipeline.hget(chat_lease_key(generation_id), LEASE_BYTES_FIELD)
+    try:
+        stream_entries, lease_bytes = pipeline.execute()
+    except redis.RedisError as redis_error:
+        logger.warning(f"Unable to read the posted progress of chat generation {generation_id}: {redis_error}")
+        return EMPTY_POSTED_PROGRESS
+    # Ended markers carry no seq and no content; only delta entries are paid.
+    generation_deltas = sorted(
+        (
+            (int(entry_fields[STREAM_SEQ_FIELD]), entry_fields.get(STREAM_CONTENT_FIELD, ""))
+            for _entry_id, entry_fields in stream_entries
+            if entry_fields.get(STREAM_GENERATION_FIELD) == generation_id and STREAM_SEQ_FIELD in entry_fields
+        ),
+    )
+    if generation_deltas:
+        content = "".join(delta_content for _seq, delta_content in generation_deltas)
+        return PostedProgress(content=content, chars=len(content))
+    if lease_bytes is None:
+        return EMPTY_POSTED_PROGRESS
+    return PostedProgress(content="", chars=int(lease_bytes))
+
+
 def release_stream_lease(generation_id: str) -> None:
     """Delete a finished generation's lease, logging rather than raising on a Redis failure.
 

horde/classes/kobold/processing_generation.py

@@ -11,7 +11,7 @@ from horde.bridge_reference import (
     is_backed_validated,
     parse_bridge_agent,
 )
-from horde.chat_progress import end_progress_stream, release_stream_lease, write_stream_lease
+from horde.chat_progress import PostedProgress, end_progress_stream, release_stream_lease, write_stream_lease
 from horde.chat_signal import ChatEvent, publish_chat_event
 from horde.classes.base.processing_generation import ProcessingGeneration, json_column_type
 from horde.classes.kobold.chat_request import (
@@ -128,6 +128,12 @@ class TextProcessingGeneration(ProcessingGeneration):
     """The return code the Horde last refused this generation's reply with, written at the refusal. The worker's
     fault submit that follows is counted by it, so a worker cannot claim the exemption a nonconformant reply has from
     dropped jobs. NULL while no reply was refused and on text generations."""
+    settled_character_count = db.Column(db.Integer, nullable=True)
+    """The posted character count a cancelled streamed generation was priced on, written by
+    [`settle_cancelled`][horde.classes.kobold.processing_generation.TextProcessingGeneration.settle_cancelled] and
+    read by ``get_things_count`` in place of a stored reply. The posted text is not retained and a stored
+    ``generation`` would count the faulted generation as a delivered result, so this count is the record cancels are
+    analysed by. NULL on every other generation."""
 
     def __init__(self, *args: object, **kwargs: object) -> None:
         super().__init__(*args, **kwargs)
@@ -359,6 +365,32 @@ class TextProcessingGeneration(ProcessingGeneration):
         if self.wp.is_completed():
             publish_chat_event(request_id, ChatEvent.DONE, generation_count)
 
+    def settle_cancelled(self, posted: PostedProgress) -> float | None:
+        """Settle a pending streamed generation whose request was cancelled on the content its worker posted.
+
+        The posted content is measured by the server, so it is priced by the output estimate like a submitted reply:
+        the worker is paid for what it turned in and the requesting user is charged the same, and the generation is
+        faulted as cancelled, without a dropped-job penalty. The row is locked and reloaded first, so a submit or fault
+        that lands while the cancel runs settles the generation once: whichever commits first wins, and the other finds
+        it finished or faulted.
+
+        Args:
+            posted: The content the generation posted, from
+                [`read_posted_progress`][horde.chat_progress.read_posted_progress].
+
+        Returns:
+            The kudos the worker was paid, or None when the generation had already finished or faulted.
+        """
+        db.session.refresh(self, with_for_update=True)
+        self.settled_character_count = posted.chars
+        kudos = self.cancel()
+        if kudos is None:
+            # A generation that a submit or fault settled first does not keep the count. cancel commits only when it
+            # settles, so this commit releases the row lock the refresh took.
+            self.settled_character_count = None
+            db.session.commit()
+        return kudos
+
     def get_things_count(self, generation=None):
         """Return the output estimate the generation is charged on: its characters divided by 4, rounded up, at least
         20 and at most the requested ``max_length``.
@@ -369,7 +401,7 @@ class TextProcessingGeneration(ProcessingGeneration):
         Returns:
             The estimate. A stored chat reply is counted by
             [`chat_reply_characters`][horde.classes.kobold.chat_request.chat_reply_characters], so tool calls are
-            charged as output.
+            charged as output. A cancelled streamed generation is counted by the posted characters it was settled on.
         """
         if generation is not None:
             character_count = len(generation)
@@ -377,6 +409,8 @@ class TextProcessingGeneration(ProcessingGeneration):
             character_count = chat_reply_characters(self.message)
         elif self.generation is not None:
             character_count = len(self.generation)
+        elif self.settled_character_count is not None:
+            character_count = self.settled_character_count
         else:
             return 0
         quick_token_count = math.ceil(character_count / 4)

sql_statements/5.1.15.txt

@@ -26,7 +26,8 @@
 -- The processing_gens columns hold a chat worker's structured reply (the message with any tool calls,
 -- finish reasons and reported token counts), the basis the generation was charged on, the worker's
 -- declaration as it stood when the generation was assigned, the transport the request's images
--- were sent to it with, and the return code its last reply was refused with. A nullable column without
+-- were sent to it with, the return code its last reply was refused with, and the posted character count
+-- a cancelled streamed generation was settled on. A nullable column without
 -- a default and a column with a constant default are both metadata-only adds; the lock timeout fails
 -- rather than queueing an ACCESS EXCLUSIVE lock behind a long transaction. Rerun on failure.
 SET lock_timeout = '5s';
@@ -63,6 +64,7 @@ ALTER TABLE processing_gens ADD COLUMN IF NOT EXISTS assigned_max_context_length
 ALTER TABLE processing_gens ADD COLUMN IF NOT EXISTS assigned_max_length INTEGER;
 ALTER TABLE processing_gens ADD COLUMN IF NOT EXISTS image_transport VARCHAR(16);
 ALTER TABLE processing_gens ADD COLUMN IF NOT EXISTS chat_refusal_rc VARCHAR(32);
+ALTER TABLE processing_gens ADD COLUMN IF NOT EXISTS settled_character_count INTEGER;
 RESET lock_timeout;
 
 -- One operation key per user. Every existing row has no key, so the partial index starts empty, but the

tests/integration/test_chat_migration.py

@@ -85,6 +85,7 @@ def test_chat_migration_adds_the_chat_columns_with_existing_rows_marked_text_and
                 ("processing_gens", "native_finish_reason", "character varying", 32, "YES"),
                 ("processing_gens", "prompt_tokens", "integer", None, "YES"),
                 ("processing_gens", "reasoning_tokens", "integer", None, "YES"),
+                ("processing_gens", "settled_character_count", "integer", None, "YES"),
                 ("waiting_prompts", "chat_conformance", "character varying", None, "NO"),
                 ("waiting_prompts", "chat_extras", "jsonb", None, "YES"),
                 ("waiting_prompts", "chat_fault_rc", "character varying", 32, "YES"),

tests/integration/test_chat_progress.py

@@ -9,6 +9,8 @@ from __future__ import annotations
 import json
 import uuid
 from collections.abc import Iterator
+from dataclasses import dataclass
+from decimal import Decimal
 from typing import TYPE_CHECKING, Any
 
 import fakeredis
@@ -211,9 +213,9 @@ def test_progress_in_sequence_is_appended_to_the_request_stream_without_a_signal
     second = _post_progress(client, request_headers["apikey"], pop_result["id"], 2, "ue")
 
     assert [response.get_json() for response in (first, retried, second)] == [
-        {"accepted_seq": 1},
-        {"accepted_seq": 1},
-        {"accepted_seq": 2},
+        {"accepted_seq": 1, "cancelled": False},
+        {"accepted_seq": 1, "cancelled": False},
+        {"accepted_seq": 2, "cancelled": False},
     ]
     entries = [fields for _, fields in lua_redis.xrange(chat_stream_key(request_id))]
     assert entries == [
@@ -311,6 +313,226 @@ def test_a_stale_generation_ends_its_stream(
     assert not progress_redis.exists(chat_lease_key(pop_result["id"]))
 
 
+def test_a_cancel_of_a_streamed_request_tells_its_worker_through_the_next_progress_post(
+    client: FlaskClient,
+    request_headers: dict[str, str],
+    chat_worker: FakeChatWorker,
+    lua_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    """Requests left queued by earlier tests may be popped first, so the request cancelled is the popped one."""
+    _submit_chat_request(client, request_headers, stream=True)
+    pop_result = chat_worker.pop()
+    request_id = lua_redis.hget(chat_lease_key(pop_result["id"]), "request")
+    assert request_id is not None
+    first = _post_progress(client, request_headers["apikey"], pop_result["id"], 1, "Bl")
+
+    cancel_response = client.delete(f"/api/v2/generate/chat/status/{request_id}", headers=request_headers)
+    after_cancel = _post_progress(client, request_headers["apikey"], pop_result["id"], 2, "ue")
+
+    assert cancel_response.status_code == 200, cancel_response.get_data(as_text=True)
+    assert first.get_json() == {"accepted_seq": 1, "cancelled": False}
+    assert after_cancel.status_code == 200, after_cancel.get_data(as_text=True)
+    assert after_cancel.get_json() == {"accepted_seq": 2, "cancelled": True}
+    entries = [fields for _, fields in lua_redis.xrange(chat_stream_key(request_id))]
+    assert [(entry["gen"], entry.get("ended"), entry.get("content")) for entry in entries] == [
+        (pop_result["id"], None, "Bl"),
+        (pop_result["id"], "1", None),
+        (pop_result["id"], None, "ue"),
+    ]
+
+
+POSTED_REPLY = "Blue is the colour of a clear sky at noon, of deep water and of the cornflowers in a summer field."
+"""More than 80 characters, so the reply is priced above the output estimate's floor of 20 tokens."""
+
+
+@dataclass(frozen=True)
+class Settlement:
+    """Represents the kudos postings recorded for one generation and the amounts the text formula gives it."""
+
+    credits: list[Decimal]
+    """The positive postings: the worker's pay."""
+
+    debits: list[Decimal]
+    """The negative postings: the requesting user's charge."""
+
+    formula_kudos: Decimal
+    """The text formula's kudos on the expected text."""
+
+    formula_charge: Decimal
+    """``formula_kudos`` with the request's extra burn, as the user is charged it on any settlement."""
+
+    formula_pay: Decimal
+    """``formula_kudos`` with the worker's bridge multiplier, as the worker is paid it."""
+
+    floor_kudos: Decimal
+    """The text formula's kudos on an empty text, the output estimate's floor."""
+
+
+def _serve_own_chat_model(client: FlaskClient, request_headers: dict[str, str]) -> FakeChatWorker:
+    """Bring a worker online for a model no other test requests, so its pops take only this test's requests."""
+    own_model = f"custom/settling-{uuid.uuid4().hex[:8]}"
+    worker = FakeChatWorker(
+        client=client,
+        headers=request_headers,
+        name=f"CICD Settling Scribe {uuid.uuid4().hex[:8]}",
+        models=[own_model],
+    )
+    worker.pop()
+    return worker
+
+
+def _pop_own_request(
+    client: FlaskClient,
+    request_headers: dict[str, str],
+    worker: FakeChatWorker,
+    *,
+    stream: bool,
+) -> tuple[str, str]:
+    """Submit a chat request for the worker's model, pop it and return the generation and request ids."""
+    request_id = _submit_chat_request(client, request_headers, model=worker.models[0], stream=stream, max_tokens=512)
+    generation_id: str = worker.pop()["id"]
+    return generation_id, request_id
+
+
+def _read_settlement(app: Flask, generation_id: str, expected_text: str) -> Settlement:
+    """Read the generation's kudos postings and price the expected text by the text formula on the same generation."""
+    from horde.classes.base.kudos import KudosLedger
+    from horde.classes.kobold.processing_generation import TextProcessingGeneration
+    from horde.flask import db
+
+    def price(generation: TextProcessingGeneration, text: str) -> Decimal:
+        generation.generation = text
+        return Decimal(str(generation.get_gen_kudos())).quantize(Decimal("0.01"))
+
+    with app.app_context():
+        amounts = [
+            amount for (amount,) in db.session.query(KudosLedger.amount).filter(KudosLedger.job_id == uuid.UUID(generation_id)).all()
+        ]
+        generation = db.session.get(TextProcessingGeneration, uuid.UUID(generation_id))
+        assert generation is not None
+        # The text is set on the loaded instance only to price it; the rollback leaves the stored row as it was.
+        with db.session.no_autoflush:
+            bridge_multiplier = Decimal(str(generation.worker.get_bridge_kudos_multiplier()))
+            formula_kudos = price(generation, expected_text)
+            formula_charge = Decimal(str(generation.wp.calculate_extra_kudos_burn(float(formula_kudos)))).quantize(Decimal("0.01"))
+            floor_kudos = price(generation, "")
+        db.session.rollback()
+    return Settlement(
+        credits=[amount for amount in amounts if amount > 0],
+        debits=[amount for amount in amounts if amount < 0],
+        formula_kudos=formula_kudos,
+        formula_charge=formula_charge,
+        formula_pay=(formula_kudos * bridge_multiplier).quantize(Decimal("0.01")),
+        floor_kudos=floor_kudos,
+    )
+
+
+def _read_generation_state(app: Flask, generation_id: str) -> tuple[bool, bool, str | None, int, int | None]:
+    """Return the generation's faulted and cancelled flags, its stored text, its worker's dropped jobs and the
+    character count it was settled on."""
+    from horde.classes.kobold.processing_generation import TextProcessingGeneration
+    from horde.flask import db
+
+    with app.app_context():
+        generation = db.session.get(TextProcessingGeneration, uuid.UUID(generation_id))
+        assert generation is not None
+        return (
+            generation.faulted,
+            generation.cancelled,
+            generation.generation,
+            generation.worker.aborted_jobs,
+            generation.settled_character_count,
+        )
+
+
+def test_a_cancel_settles_a_streamed_generation_on_the_content_its_worker_posted(
+    app: Flask,
+    client: FlaskClient,
+    request_headers: dict[str, str],
+    lua_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    """The worker is paid and the user charged the text formula on the posted content, reasoning aside. The
+    generation is faulted as cancelled without a dropped job and keeps the posted character count it was settled on,
+    and neither a later submit nor the stale pass pays it again."""
+    from datetime import datetime, timedelta
+
+    from horde.classes.kobold.processing_generation import TextProcessingGeneration
+    from horde.database import threads
+    from horde.flask import db
+
+    worker = _serve_own_chat_model(client, request_headers)
+    generation_id, request_id = _pop_own_request(client, request_headers, worker, stream=True)
+    _post_progress(client, request_headers["apikey"], generation_id, 1, "", reasoning_content="The sky is clear today.")
+    _post_progress(client, request_headers["apikey"], generation_id, 2, POSTED_REPLY[:40])
+    _post_progress(client, request_headers["apikey"], generation_id, 3, POSTED_REPLY[40:])
+    aborted_jobs_before = _read_generation_state(app, generation_id)[3]
+
+    cancel_response = client.delete(f"/api/v2/generate/chat/status/{request_id}", headers=request_headers)
+
+    assert cancel_response.status_code == 200, cancel_response.get_data(as_text=True)
+    settlement = _read_settlement(app, generation_id, POSTED_REPLY)
+    assert settlement.formula_kudos > settlement.floor_kudos
+    assert settlement.credits == [settlement.formula_pay]
+    assert settlement.debits == [-settlement.formula_charge]
+    assert _read_generation_state(app, generation_id) == (True, True, None, aborted_jobs_before, len(POSTED_REPLY))
+    assert lua_redis.hget(chat_lease_key(generation_id), "cancelled") == "1"
+
+    late_submit = worker.submit(build_chat_submit_body(generation_id=generation_id, content=POSTED_REPLY, model_id=worker.models[0]))
+
+    assert late_submit.status_code == 400, late_submit.get_data(as_text=True)
+    assert late_submit.get_json()["rc"] == "AbortedGen"
+
+    with app.app_context():
+        generation = db.session.get(TextProcessingGeneration, uuid.UUID(generation_id))
+        assert generation is not None
+        generation.start_time = datetime.utcnow() - timedelta(seconds=generation.job_ttl + 1)
+        db.session.commit()
+        threads.check_waiting_prompts()
+
+    assert _read_settlement(app, generation_id, POSTED_REPLY).credits == [settlement.formula_pay]
+    assert _read_generation_state(app, generation_id) == (True, True, None, aborted_jobs_before, len(POSTED_REPLY))
+
+
+def test_a_cancel_before_any_post_settles_a_streamed_generation_on_the_formula_floor(
+    app: Flask,
+    client: FlaskClient,
+    request_headers: dict[str, str],
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    worker = _serve_own_chat_model(client, request_headers)
+    generation_id, request_id = _pop_own_request(client, request_headers, worker, stream=True)
+
+    cancel_response = client.delete(f"/api/v2/generate/chat/status/{request_id}", headers=request_headers)
+
+    assert cancel_response.status_code == 200, cancel_response.get_data(as_text=True)
+    settlement = _read_settlement(app, generation_id, "")
+    assert settlement.formula_kudos == settlement.floor_kudos
+    assert settlement.debits == [-settlement.formula_charge]
+    assert settlement.credits == [settlement.formula_pay]
+    generation_state = _read_generation_state(app, generation_id)
+    assert generation_state[:2] == (True, True)
+    assert generation_state[4] == 0
+
+
+def test_a_cancel_of_a_request_that_does_not_stream_leaves_its_generation_pending_and_unpaid(
+    app: Flask,
+    client: FlaskClient,
+    request_headers: dict[str, str],
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    worker = _serve_own_chat_model(client, request_headers)
+    generation_id, request_id = _pop_own_request(client, request_headers, worker, stream=False)
+
+    cancel_response = client.delete(f"/api/v2/generate/chat/status/{request_id}", headers=request_headers)
+
+    assert cancel_response.status_code == 200, cancel_response.get_data(as_text=True)
+    settlement = _read_settlement(app, generation_id, "")
+    assert (settlement.credits, settlement.debits) == ([], [])
+    generation_state = _read_generation_state(app, generation_id)
+    assert generation_state[:3] == (False, False, None)
+    assert generation_state[4] is None
+
+
 def test_a_reasoning_delta_is_appended_with_its_reasoning_beside_empty_content(
     client: FlaskClient,
     request_headers: dict[str, str],
@@ -324,7 +546,10 @@ def test_a_reasoning_delta_is_appended_with_its_reasoning_beside_empty_content(
     reasoning = _post_progress(client, request_headers["apikey"], pop_result["id"], 1, "", reasoning_content="Think")
     text = _post_progress(client, request_headers["apikey"], pop_result["id"], 2, "Blue")
 
-    assert [response.get_json() for response in (reasoning, text)] == [{"accepted_seq": 1}, {"accepted_seq": 2}]
+    assert [response.get_json() for response in (reasoning, text)] == [
+        {"accepted_seq": 1, "cancelled": False},
+        {"accepted_seq": 2, "cancelled": False},
+    ]
     entries = [
         fields
         for stream_key in lua_redis.scan_iter(match=chat_stream_key("*"))

tests/unit/test_chat_progress.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 from collections.abc import Iterator
+from types import SimpleNamespace
 
 import fakeredis
 import pytest
@@ -16,14 +17,18 @@ from horde.chat_progress import (
     CHAT_PROGRESS_BYTES_PER_TOKEN,
     CHAT_PROGRESS_LEASE_GRACE_SECONDS,
     CHAT_PROGRESS_STREAM_TTL_SECONDS,
+    EMPTY_POSTED_PROGRESS,
+    PostedProgress,
     ProgressAppend,
     ProgressOutcome,
     StreamLease,
     api_key_digest,
     append_progress,
+    cancel_progress_streams,
     chat_lease_key,
     chat_stream_key,
     end_progress_stream,
+    read_posted_progress,
     read_stream_lease,
     release_stream_lease,
     write_stream_lease,
@@ -318,3 +323,135 @@ def test_reasoning_counts_toward_the_byte_cap_with_the_content(lua_redis: fakere
     half_cap = CHAT_PROGRESS_BYTES_PER_TOKEN // 2
     assert append_progress(_lease(), 1, "x" * half_cap, "y" * (half_cap + 1)) == ProgressAppend(ProgressOutcome.OVER_CAP, 1)
     assert not lua_redis.exists(chat_stream_key(REQUEST_ID))
+
+
+def test_a_cancel_flags_the_live_lease_keeps_it_and_appends_the_ended_marker(
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease()
+    lease_ttl = progress_redis.ttl(chat_lease_key(GENERATION_ID))
+
+    cancel_progress_streams([GENERATION_ID], REQUEST_ID)
+
+    assert progress_redis.hget(chat_lease_key(GENERATION_ID), "cancelled") == "1"
+    assert 0 < progress_redis.ttl(chat_lease_key(GENERATION_ID)) <= lease_ttl
+    assert _stream_entries(progress_redis) == [{"gen": GENERATION_ID, "ended": "1"}]
+    assert 0 < progress_redis.ttl(chat_stream_key(REQUEST_ID)) <= CHAT_PROGRESS_STREAM_TTL_SECONDS
+
+
+def test_a_cancel_of_a_generation_without_a_lease_leaves_redis_as_it_is(
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    cancel_progress_streams([GENERATION_ID], REQUEST_ID)
+
+    assert not progress_redis.exists(chat_lease_key(GENERATION_ID))
+    assert not progress_redis.exists(chat_stream_key(REQUEST_ID))
+
+
+def test_a_cancel_flags_only_the_generations_that_still_hold_a_lease(
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease()
+    released_generation_id = "0d8e7c41-93a5-4d0b-a1f2-6c3b9e8d7f20"
+
+    cancel_progress_streams([released_generation_id, GENERATION_ID], REQUEST_ID)
+
+    assert not progress_redis.exists(chat_lease_key(released_generation_id))
+    assert _stream_entries(progress_redis) == [{"gen": GENERATION_ID, "ended": "1"}]
+
+
+def test_a_cancel_of_a_fake_generation_flags_its_lease_without_a_marker(
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease(fake=True)
+
+    cancel_progress_streams([GENERATION_ID], REQUEST_ID)
+
+    assert progress_redis.hget(chat_lease_key(GENERATION_ID), "cancelled") == "1"
+    assert not progress_redis.exists(chat_stream_key(REQUEST_ID))
+
+
+def test_a_cancel_redis_refuses_is_logged_and_does_not_raise(monkeypatch: pytest.MonkeyPatch) -> None:
+    server = fakeredis.FakeServer()
+    server.connected = False
+    monkeypatch.setattr(chat_progress, "_client", fakeredis.FakeStrictRedis(server=server, decode_responses=True))
+    warnings: list[str] = []
+    monkeypatch.setattr(chat_progress, "logger", SimpleNamespace(warning=warnings.append))
+
+    cancel_progress_streams([GENERATION_ID], REQUEST_ID)
+
+    assert len(warnings) == 1
+    assert REQUEST_ID in warnings[0]
+
+
+def test_a_cancelled_lease_accepts_the_delta_and_its_accept_and_retry_carry_the_flag(
+    lua_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease()
+    append_progress(_lease(), 1, "Hel")
+    cancel_progress_streams([GENERATION_ID], REQUEST_ID)
+
+    accepted = append_progress(_lease(), 2, "lo")
+    retry = append_progress(_lease(), 2, "lo")
+
+    assert accepted == ProgressAppend(ProgressOutcome.ACCEPTED, 2, cancelled=True)
+    assert retry == ProgressAppend(ProgressOutcome.DUPLICATE, 2, cancelled=True)
+    assert [entry.get("ended") or entry["content"] for entry in _stream_entries(lua_redis)] == ["Hel", "1", "lo"]
+
+
+def test_a_refused_post_on_a_cancelled_lease_is_refused_as_before(lua_redis: fakeredis.FakeStrictRedis) -> None:
+    _write_lease()
+    cancel_progress_streams([GENERATION_ID], REQUEST_ID)
+
+    assert append_progress(_lease(), 3, "lo") == ProgressAppend(ProgressOutcome.GAP, 1)
+
+
+def test_a_lease_without_the_flag_acknowledges_the_request_as_not_cancelled(
+    lua_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease()
+
+    assert append_progress(_lease(), 1, "Hel").cancelled is False
+
+
+def test_the_posted_progress_joins_the_generation_content_in_seq_order_without_reasoning(
+    lua_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease(max_length=64)
+    other_generation_id = "0d8e7c41-93a5-4d0b-a1f2-6c3b9e8d7f20"
+    lua_redis.xadd(chat_stream_key(REQUEST_ID), {"gen": other_generation_id, "seq": 1, "content": "Elsewhere"})
+    append_progress(_lease(), 1, "", "Think")
+    append_progress(_lease(), 2, "Hel")
+    append_progress(_lease(), 3, "lo", "more")
+    cancel_progress_streams([GENERATION_ID], REQUEST_ID)
+
+    assert read_posted_progress(GENERATION_ID, REQUEST_ID) == PostedProgress(content="Hello", chars=5)
+
+
+def test_the_posted_progress_falls_back_to_the_lease_bytes_when_the_stream_is_gone(
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease()
+    progress_redis.hset(chat_lease_key(GENERATION_ID), "bytes", 42)
+
+    assert read_posted_progress(GENERATION_ID, REQUEST_ID) == PostedProgress(content="", chars=42)
+
+
+def test_the_posted_progress_of_a_generation_that_posted_nothing_is_empty(
+    progress_redis: fakeredis.FakeStrictRedis,
+) -> None:
+    _write_lease()
+
+    assert read_posted_progress(GENERATION_ID, REQUEST_ID) == EMPTY_POSTED_PROGRESS
+
+
+def test_the_posted_progress_redis_refuses_is_logged_and_empty(monkeypatch: pytest.MonkeyPatch) -> None:
+    server = fakeredis.FakeServer()
+    server.connected = False
+    monkeypatch.setattr(chat_progress, "_client", fakeredis.FakeStrictRedis(server=server, decode_responses=True))
+    warnings: list[str] = []
+    monkeypatch.setattr(chat_progress, "logger", SimpleNamespace(warning=warnings.append))
+
+    assert read_posted_progress(GENERATION_ID, REQUEST_ID) == EMPTY_POSTED_PROGRESS
+    assert len(warnings) == 1
+    assert GENERATION_ID in warnings[0]