diff --git a/docs/streaming-architecture.md b/docs/streaming-architecture.md
index 66591b58..433d08be 100644
--- a/docs/streaming-architecture.md
+++ b/docs/streaming-architecture.md
@@ -66,14 +66,15 @@ the connection without a success acknowledgement, and the row is marked failed.
 
 ### Resume after a network failure
 
-Producers that declare `protocolVersion: 2` in `start_mission` get lossless
+Producers that declare `protocolVersion: 2` or later in `start_mission` get lossless
 recovery from network failures. Older producers keep the behavior described in
 the next section.
 
 - `start_mission` carries a `missionId` (a random UUID per recording) and a
   `producerId` (a random UUID per extension process). Every envelope carries a
   `seq`, starting with 1 for `start_mission`.
-- The server acknowledges `start_mission` with `protocolVersion: 2`, and then
+- The server acknowledges `start_mission` with `protocolVersion: 3` (resume
+  plus `discard_mission`, see below), and then
   acknowledges received envelopes cumulatively (`{"type":"ack","seq":N}`) every
   `streaming.ackInterval` and during bursts. The extension holds every envelope
   in memory until it is acknowledged.
@@ -86,17 +87,25 @@ the next section.
   a web restart), the extension starts a new segment: it sends `start_mission`
   again followed by every envelope the server never acknowledged. The skipped
   numbers mark the new recording `partial`.
-- If the mission was already saved but the `end_mission` acknowledgement was
-  lost, the server answers `resume_completed` instead.
+- If the mission was already saved, or discarded, but the `end_mission` or
+  `discard_mission` acknowledgement was lost, the server answers
+  `resume_completed` instead.
+- Servers at protocol version 3 put `protocolVersion: 3` on every resume reply.
 - A connection that resumes a mission closes any older connection still bound to
   it, such as a half-open socket. The extension also abandons a socket whose
   written envelopes stay unacknowledged for 20 seconds.
 - Sessions are keyed only by mission id, so any number of game servers can stream
   to one web server. Only a session of the same producer is affected by a new
   `start_mission`: its earlier mission is saved as `partial` at once instead of
-  waiting for the grace period.
+  waiting for the grace period. A start marked `"recovered": true` resends an
+  older mission from the extension's spool. It supersedes nothing, and the
+  producer's next start does not supersede it either: the producer may be
+  recording its next mission meanwhile. A `resume` marked the same way makes
+  the session it continues exempt too.
 - A session not resumed within the grace period is saved with `partial` appended
-  to its tags.
+  to its tags. A session superseded or expired before it received anything
+  after `start_mission`, or ended as interrupted with nothing in it, is
+  removed instead: an empty recording is not kept.
 
 | Setting | Default | Meaning |
 |---|---|---|
@@ -147,9 +156,56 @@ still replaces the earlier partial segments when nothing is missing. Missions
 whose save was never confirmed are recovered the same way, with their own
 `end_mission`.
 
+### Discarding a mission
+
+The addon enforces its save limits, such as `OCAP_settings_minMissionTime`, by
+not saving a recording. A streamed recording already lives on the server,
+which would save it as `partial` once its producer is gone. So instead of
+`end_mission` the producer sends a numbered `discard_mission`, whose payload
+names the mission (`{"missionId":"…","producerId":"…"}`):
+
+- The server removes the recording's `streaming` row and files, along with the
+  partial and failed segments earlier attempts of the same mission id left
+  behind, and acknowledges it like `end_mission`
+  (`{"type":"ack","for":"discard_mission","seq":N}`). Recordings marked as
+  saved through `end_mission` (`stream_ended`, set before the session leaves
+  the registry) are never removed.
+- It also ends a legacy session (no numbering) the same way; the disconnect that
+  follows no longer saves it as `partial`.
+- Like `end_mission`, it is acknowledged only once the recording is gone: until
+  then progress acknowledgements stop short of its number. If the removal
+  fails, the server closes the connection without acknowledging it, marks the
+  session's row `failed` (it is removed last), and the producer repeats the
+  discard. A legacy session has no mission id to repeat it by; its row stays
+  `failed`.
+- The mission id is remembered as finished for an hour, in memory only, so a
+  producer whose acknowledgement was lost gets `resume_completed`. After a web
+  restart it gets `resume_unknown` and repeats the discard, which finds nothing
+  left to remove. A live connection repeats it by resending the mission from
+  its spool into a new session first, which the discard then drops with the
+  rest; recovery repeats it without a session.
+- An unconfirmed discard stays in the extension's spool. Recovery sends only the
+  `discard_mission` again, never the mission data. When the server no longer
+  holds the session (`resume_unknown`), for example because the resume grace
+  expired and the mission was saved as `partial`, or the web server restarted
+  and marked it `failed`, the discard is sent without a session. The server
+  then removes every recording of that mission id not saved through
+  `end_mission`, and acknowledges it without `seq`. It holds off a start of the
+  mission id meanwhile. A session of the mission still waiting for its producer
+  (`producerId` must match) is dropped in its place, and one being saved is
+  waited for. One a connection still feeds, or another producer's, is left
+  alone, and the discard is not acknowledged.
+- Files are removed on a best-effort basis, as for a deleted operation: once the
+  row is gone, the discard counts as done.
+
+The extension sends `discard_mission` only to servers that acknowledged
+`start_mission`, or answered `resume` for the session-less form, with
+`protocolVersion: 3` or later. With an older server it closes the connection
+instead, and that server saves the recording as `partial`.
+
 ### Legacy producers
 
-For producers without `protocolVersion: 2`, disconnecting without `end_mission`,
+For producers without `protocolVersion: 2` or later, disconnecting without `end_mission`,
 including an addon/extension crash or a normal WebSocket close, finalizes
 received data and appends `partial` to the mission's existing tags. Repeated
 starts cannot silently discard an active mission. Their reconnect starts a new
diff --git a/go.mod b/go.mod
index c8f218cf..2d9a7fea 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,7 @@ module github.com/OCAP2/web
 go 1.27.2
 
 require (
-	github.com/OCAP2/extension/v5 v5.0.2-0.20261019201552-0ebd39eb0267
+	github.com/OCAP2/extension/v5 v5.0.2-0.20261027093953-0a7ce6a9b9a7
 	github.com/getkin/kin-openapi v0.151.0
 	github.com/go-fuego/fuego v0.21.0
 	github.com/golang-jwt/jwt/v5 v5.3.1
diff --git a/go.sum b/go.sum
index cdc01df3..5a396de7 100644
--- a/go.sum
+++ b/go.sum
@@ -1,7 +1,7 @@
-github.com/OCAP2/extension/v5 v5.0.2-0.20261019173050-9ef6f2444ee8 h1:FgLuVMC2vSqwhfELiMGXuEJaN36qNY9N4m/0e5ioqQU=
-github.com/OCAP2/extension/v5 v5.0.2-0.20261019173050-9ef6f2444ee8/go.mod h1:+sonvAIzk+dRa7sFzKfjpcKcCFpDo7wyQF6M63XxUsg=
-github.com/OCAP2/extension/v5 v5.0.2-0.20261019201552-0ebd39eb0267 h1:14dPRuYh/Cr+POYlnJWMyNo9PVnRT7K31MBLJw4TkNU=
-github.com/OCAP2/extension/v5 v5.0.2-0.20261019201552-0ebd39eb0267/go.mod h1:+sonvAIzk+dRa7sFzKfjpcKcCFpDo7wyQF6M63XxUsg=
+github.com/OCAP2/extension/v5 v5.0.2-0.20261027013159-6e1878442892 h1:Adl0GAaAj5Eg3up22xmvz1m7v3i7hOpYpwdBMeogT4I=
+github.com/OCAP2/extension/v5 v5.0.2-0.20261027013159-6e1878442892/go.mod h1:+sonvAIzk+dRa7sFzKfjpcKcCFpDo7wyQF6M63XxUsg=
+github.com/OCAP2/extension/v5 v5.0.2-0.20261027093953-0a7ce6a9b9a7 h1:s4z0SDRxEJo9zNVTD6UaQvgEBYv5R8kq/LErQZztjTY=
+github.com/OCAP2/extension/v5 v5.0.2-0.20261027093953-0a7ce6a9b9a7/go.mod h1:+sonvAIzk+dRa7sFzKfjpcKcCFpDo7wyQF6M63XxUsg=
 github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI=
 github.com/dlclark/regexp2 v1.11.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
diff --git a/internal/server/handler_stream.go b/internal/server/handler_stream.go
index 12ccbd63..a0eb1d42 100644
--- a/internal/server/handler_stream.go
+++ b/internal/server/handler_stream.go
@@ -175,7 +175,7 @@ func (h *Handler) streamLoop(ws *websocket.Conn) {
 			ack := streaming.AckMessage{Type: streaming.ServerAck, For: streaming.TypeStartMission}
 			if started.resumable() {
 				ack.Seq = started.lastSeq.Load()
-				ack.ProtocolVersion = streaming.ProtocolVersionResume
+				ack.ProtocolVersion = streaming.ProtocolVersionDiscard
 			}
 			writeJSON(ack)
 
@@ -212,6 +212,31 @@ func (h *Handler) streamLoop(ws *websocket.Conn) {
 			writeJSON(ack)
 			return
 
+		case streaming.TypeDiscardMission:
+			if ss == nil {
+				if !h.discardStreamed(envelope.Payload) {
+					fail("mission could not be discarded")
+					return
+				}
+				writeJSON(streaming.AckMessage{Type: streaming.ServerAck, For: streaming.TypeDiscardMission})
+				return
+			}
+			switch h.discardStream(ss, conn, envelope) {
+			case endTakenOver:
+				return
+			case endFailed:
+				current.Store(nil)
+				fail("mission could not be discarded")
+				return
+			}
+			current.Store(nil)
+			ack := streaming.AckMessage{Type: streaming.ServerAck, For: streaming.TypeDiscardMission}
+			if ss.resumable() {
+				ack.Seq = ss.lastSeq.Load()
+			}
+			writeJSON(ack)
+			return
+
 		default:
 			// All other messages require an active session.
 			if ss == nil {
@@ -259,16 +284,20 @@ func (h *Handler) startStream(envelope streaming.Envelope, conn *streamConn) (*s
 				return nil, "mission id in use"
 			}
 			// The producer repeats a start whose acknowledgement it missed.
-			if !h.takeOver(existing, conn) {
+			if !h.takeOver(existing, conn, payload.Recovered) {
 				return nil, "mission already saved"
 			}
 			slog.Info("stream: mission start repeated", "missionId", payload.MissionID)
 			return existing, ""
 		}
 		// A producer records one mission at a time: a new start supersedes
-		// whatever it left behind. Other producers are unaffected.
-		for _, old := range reg.byProducer(payload.ProducerID, payload.MissionID) {
-			go h.expireStream(old, 0)
+		// whatever it left behind. Other producers are unaffected. Resends
+		// from the spool are not the live mission: a recovered start
+		// supersedes nothing, and is not superseded (see byProducer).
+		if !payload.Recovered {
+			for _, old := range reg.byProducer(payload.ProducerID, payload.MissionID) {
+				go h.expireStream(old, 0)
+			}
 		}
 	}
 	// settle ends the reservation: ss nil when the mission was refused.
@@ -323,6 +352,7 @@ func (h *Handler) startStream(envelope streaming.Envelope, conn *streamConn) (*s
 	if resumable {
 		ss.missionID = payload.MissionID
 		ss.producer = payload.ProducerID
+		ss.recovered.Store(payload.Recovered)
 		if err := h.repoOperation.SetStreamMission(context.Background(), op.ID, ss.missionID); err != nil {
 			slog.Warn("stream: cannot record mission id", "error", err)
 		}
@@ -334,12 +364,19 @@ func (h *Handler) startStream(envelope streaming.Envelope, conn *streamConn) (*s
 
 // takeOver makes conn the session's connection, closing any previous one and
 // cancelling a pending grace expiry. It fails once the session is saved.
-func (h *Handler) takeOver(ss *streamSession, conn *streamConn) bool {
+// recovered marks the session continued from the producer's spool: not the
+// mission it records live, which its next start supersedes. It is set under
+// the session's lock, so a supersede already scheduled sees it (see
+// expireStream).
+func (h *Handler) takeOver(ss *streamSession, conn *streamConn, recovered bool) bool {
 	ss.mu.Lock()
 	defer ss.mu.Unlock()
 	if ss.done {
 		return false
 	}
+	if recovered {
+		ss.recovered.Store(true)
+	}
 	if ss.owner != nil && ss.owner != conn {
 		ss.owner.kick() // a half-open predecessor must not feed the session any more
 	}
@@ -352,22 +389,27 @@ func (h *Handler) takeOver(ss *streamSession, conn *streamConn) bool {
 }
 
 // resumeStream answers resume and returns the session conn now feeds, if any.
+// Every reply carries the protocol version: a producer finishing a discard
+// learns whether the server takes discard_mission without a session.
 func (h *Handler) resumeStream(raw json.RawMessage, conn *streamConn) (streaming.AckMessage, *streamSession) {
+	reply := func(msgType string, seq uint64) streaming.AckMessage {
+		return streaming.AckMessage{Type: msgType, Seq: seq, ProtocolVersion: streaming.ProtocolVersionDiscard}
+	}
 	var p streaming.ResumePayload
 	if err := json.Unmarshal(raw, &p); err != nil || p.MissionID == "" {
-		return streaming.AckMessage{Type: streaming.ServerResumeUnknown}, nil
+		return reply(streaming.ServerResumeUnknown, 0), nil
 	}
 	reg := h.streamRegistry()
-	if ss := reg.get(p.MissionID); ss != nil && ss.producer == p.ProducerID && h.takeOver(ss, conn) {
+	if ss := reg.get(p.MissionID); ss != nil && ss.producer == p.ProducerID && h.takeOver(ss, conn, p.Recovered) {
 		seq := ss.lastSeq.Load()
-		slog.Info("stream: mission resumed", "missionId", p.MissionID, "seq", seq)
-		return streaming.AckMessage{Type: streaming.ServerResumeOK, Seq: seq}, ss
+		slog.Info("stream: mission resumed", "missionId", p.MissionID, "seq", seq, "recovered", p.Recovered)
+		return reply(streaming.ServerResumeOK, seq), ss
 	}
 	if reg.isCompleted(p.MissionID) || h.streamEnded(p.MissionID) {
-		return streaming.AckMessage{Type: streaming.ServerResumeCompleted}, nil
+		return reply(streaming.ServerResumeCompleted, 0), nil
 	}
 	slog.Warn("stream: resume of unknown mission", "missionId", p.MissionID)
-	return streaming.AckMessage{Type: streaming.ServerResumeUnknown}, nil
+	return reply(streaming.ServerResumeUnknown, 0), nil
 }
 
 // acceptSeq applies envelope numbering. It reports false for a duplicate the
@@ -475,10 +517,32 @@ func (h *Handler) endStream(ss *streamSession, conn *streamConn, envelope stream
 	if ss.gap || interrupted {
 		tag = withPartialTag(tag)
 	}
+	if interrupted && len(ss.counts) == 0 && ss.resumable() {
+		// A mission recovered after a crash right after its start: an empty
+		// recording is not kept, as on grace expiry.
+		slog.Info("stream: dropping interrupted mission that recorded nothing", "missionId", ss.missionID)
+		ss.done = true
+		h.dropEmptySession(ss, true)
+		if envelope.Seq > ss.lastSeq.Load() {
+			ss.lastSeq.Store(envelope.Seq)
+		}
+		return endSaved
+	}
 	slog.Info("stream: mission ended", "missionId", ss.missionID, "counts", ss.counts)
 
 	ss.done = true
 	err := h.saveStream(ss, tag)
+	ctx := context.Background()
+	if err == nil && ss.resumable() {
+		// Marked before the session leaves the registry, so a discard of the
+		// mission id never takes the saved recording for an unended one.
+		// Without the mark the save does not count: end_mission is not
+		// acknowledged, and the producer resends the mission, whose
+		// recording replaces this one, or discards it.
+		if err = h.markStreamEnded(ctx, ss.opID); err != nil {
+			slog.Error("stream: cannot mark mission ended", "missionId", ss.missionID, "id", ss.opID, "error", err)
+		}
+	}
 	if ss.resumable() {
 		h.streamRegistry().remove(ss, err == nil)
 	}
@@ -488,18 +552,23 @@ func (h *Handler) endStream(ss *streamSession, conn *streamConn, envelope stream
 	if envelope.Seq > 0 {
 		ss.lastSeq.Store(envelope.Seq)
 	}
-	if ss.resumable() {
-		ctx := context.Background()
-		if err := h.repoOperation.MarkStreamEnded(ctx, ss.opID); err != nil {
-			slog.Warn("stream: cannot mark mission ended", "error", err)
-		}
-		if !ss.gap {
-			h.replaceStreamSegments(ctx, ss)
-		}
+	if ss.resumable() && !ss.gap {
+		h.replaceStreamSegments(ctx, ss)
 	}
 	return endSaved
 }
 
+// markStreamEnded records that the session's row was saved through
+// end_mission.
+func (h *Handler) markStreamEnded(ctx context.Context, id int64) error {
+	if hook := streamTestHooks.beforeMarkEnded; hook != nil {
+		if err := hook(); err != nil {
+			return err
+		}
+	}
+	return h.repoOperation.MarkStreamEnded(ctx, id)
+}
+
 // replaceStreamSegments removes the partial recordings an earlier attempt of
 // the mission left behind, once a complete recording of it is saved. They
 // exist when the producer resent the mission in full, for example after a web
@@ -511,15 +580,208 @@ func (h *Handler) replaceStreamSegments(ctx context.Context, ss *streamSession)
 		return
 	}
 	for _, seg := range segments {
-		if err := h.repoOperation.Delete(ctx, seg.ID); err != nil && !errors.Is(err, ErrNotFound) {
+		if err := h.removeStreamOperation(ctx, seg.ID, seg.Filename); err != nil {
 			slog.Warn("stream: cannot remove earlier segment", "id", seg.ID, "error", err)
 			continue
 		}
-		removeRecordingArtifacts(h.setting.Data, seg.Filename)
 		slog.Info("stream: replaced partial segment with complete recording", "missionId", ss.missionID, "removed", seg.Filename, "kept", filepath.Base(ss.outputDir))
 	}
 }
 
+// discardStream drops the session for discard_mission: the producer decided
+// the recording does not meet its save limits, for example the minimum
+// mission duration. The streaming row and its files go, along with any
+// partial segments an earlier attempt of the mission left behind. A failure
+// leaves the discard unacknowledged, so the producer repeats it.
+func (h *Handler) discardStream(ss *streamSession, conn *streamConn, envelope streaming.Envelope) endResult {
+	ss.mu.Lock()
+	defer ss.mu.Unlock()
+	if ss.owner != conn || ss.done {
+		return endTakenOver
+	}
+	// Like end_mission, discard_mission is acknowledged only once the
+	// recording is gone.
+	checkSeq(ss, envelope.Seq)
+	ss.done = true
+	if h.dropSession(ss) != nil {
+		return endFailed
+	}
+	if envelope.Seq > ss.lastSeq.Load() {
+		ss.lastSeq.Store(envelope.Seq)
+	}
+	slog.Info("stream: mission discarded", "missionId", ss.missionID)
+	return endSaved
+}
+
+// dropEmptySession removes a session that received nothing after its start
+// instead of saving an empty recording, and releases it; finished remembers
+// the mission id as finished for resume. Caller holds ss.mu and has set
+// ss.done.
+func (h *Handler) dropEmptySession(ss *streamSession, finished bool) {
+	ctx := context.Background()
+	if err := h.removeStreamOperation(ctx, ss.opID, filepath.Base(ss.outputDir)); err != nil {
+		slog.Error("stream: cannot remove empty mission", "missionId", ss.missionID, "id", ss.opID, "error", err)
+		if ss.opID > 0 {
+			h.repoOperation.UpdateConversionStatus(ctx, ss.opID, ConversionStatusFailed)
+		}
+	}
+	h.streamRegistry().release()
+	if ss.resumable() {
+		h.streamRegistry().remove(ss, finished)
+	}
+}
+
+// dropSession removes a discarded session's recording and earlier unended
+// segments of its mission, and releases the session. On failure the row is
+// marked failed, no longer live, and the producer repeats the discard: on a
+// new session after resume_unknown, or without one from its spool's
+// recovery. Either removes it. Caller holds ss.mu and has set ss.done.
+func (h *Handler) dropSession(ss *streamSession) error {
+	slog.Info("stream: discarding mission", "missionId", ss.missionID, "mission", ss.session.Mission().MissionName, "counts", ss.counts)
+	ctx := context.Background()
+	// The session's own row goes last: after a failure it is still there to
+	// be marked failed.
+	var err error
+	if ss.resumable() {
+		err = h.removeUnendedSegments(ctx, ss.missionID, ss.opID)
+	}
+	if err == nil {
+		err = h.removeStreamOperation(ctx, ss.opID, filepath.Base(ss.outputDir))
+	}
+	h.streamRegistry().release() // before leaving the registry, as saveStream does
+	if ss.resumable() {
+		// Remembered as finished, so a producer whose acknowledgement was
+		// lost learns from resume that the discard went through.
+		h.streamRegistry().remove(ss, err == nil)
+	}
+	if err != nil {
+		slog.Error("stream: cannot remove discarded mission", "missionId", ss.missionID, "id", ss.opID, "error", err)
+		if ss.opID > 0 {
+			h.repoOperation.UpdateConversionStatus(ctx, ss.opID, ConversionStatusFailed)
+		}
+	}
+	return err
+}
+
+// discardStreamed handles discard_mission without a session: the producer
+// finishes a discard the server did not confirm after the session ended, for
+// example when its producer stayed away past the resume grace and the
+// mission was saved as partial. What the server kept of the mission goes;
+// recordings saved through end_mission stay. It reports false when they
+// could not be removed, or a connection still feeds the mission.
+func (h *Handler) discardStreamed(raw json.RawMessage) bool {
+	var p streaming.DiscardMissionPayload
+	if len(raw) == 0 || string(raw) == "null" {
+		slog.Info("stream: mission discarded without an active session")
+		return true // like end_mission without a session: nothing named
+	}
+	if err := json.Unmarshal(raw, &p); err != nil || p.MissionID == "" {
+		// Not acknowledged: the producer keeps its spool rather than
+		// believing the mission dropped.
+		slog.Warn("stream: invalid discard_mission payload", "error", err)
+		return false
+	}
+	reg := h.streamRegistry()
+	// Reserving the mission id holds off a start of it while its recordings
+	// go. A session of the mission is dropped in its place, unless a
+	// connection still feeds it.
+	// A finalized session leaves the registry before it unlocks, so one retry
+	// is normally enough; the bound only guards against a broken invariant.
+	for range 3 {
+		existing, err := reg.reserve(p.MissionID, streamReserveWait)
+		if err != nil {
+			slog.Warn("stream: cannot discard mission, start in progress", "missionId", p.MissionID)
+			return false
+		}
+		if existing == nil {
+			err := h.removeUnendedSegments(context.Background(), p.MissionID, 0)
+			reg.settle(p.MissionID, nil)
+			if err != nil {
+				slog.Error("stream: cannot remove discarded mission", "missionId", p.MissionID, "error", err)
+				return false
+			}
+			slog.Info("stream: mission discarded after its session ended", "missionId", p.MissionID)
+			return true
+		}
+		if hook := streamTestHooks.beforeDiscardSession; hook != nil {
+			hook()
+		}
+		if dropped, retry := h.discardDetached(existing, p.ProducerID); !retry {
+			return dropped
+		}
+	}
+	slog.Warn("stream: cannot discard mission, its session is still being finalized", "missionId", p.MissionID)
+	return false
+}
+
+// discardDetached drops a session waiting for its producer to resume, for a
+// session-less discard_mission. It refuses one a connection feeds or another
+// producer recorded, and asks for a retry when the session was finalized
+// meanwhile: it then leaves the registry.
+func (h *Handler) discardDetached(ss *streamSession, producer string) (dropped, retry bool) {
+	ss.mu.Lock()
+	defer ss.mu.Unlock()
+	if ss.done {
+		return false, true
+	}
+	if ss.owner != nil {
+		slog.Warn("stream: cannot discard mission, a connection feeds it", "missionId", ss.missionID)
+		return false, false
+	}
+	if ss.producer != producer {
+		slog.Warn("stream: cannot discard mission of another producer", "missionId", ss.missionID)
+		return false, false
+	}
+	if ss.grace != nil {
+		ss.grace.Stop()
+		ss.grace = nil
+	}
+	ss.done = true
+	return h.dropSession(ss) == nil, false
+}
+
+// removeUnendedSegments removes the recordings of the producer mission not
+// saved through end_mission, except the one with id exceptID. It returns the
+// first error and removes what it can.
+func (h *Handler) removeUnendedSegments(ctx context.Context, missionID string, exceptID int64) error {
+	if h.repoOperation == nil {
+		return nil
+	}
+	segments, err := h.repoOperation.UnendedStreamSegments(ctx, missionID, exceptID)
+	if err != nil {
+		return fmt.Errorf("list segments: %w", err)
+	}
+	var first error
+	for _, seg := range segments {
+		if err := h.removeStreamOperation(ctx, seg.ID, seg.Filename); err != nil {
+			slog.Warn("stream: cannot remove segment", "id", seg.ID, "error", err)
+			if first == nil {
+				first = err
+			}
+			continue
+		}
+		slog.Info("stream: removed segment of discarded mission", "missionId", missionID, "removed", seg.Filename)
+	}
+	return first
+}
+
+// removeStreamOperation deletes a streamed operation row and its files. The
+// files go on a best-effort basis, as for a deleted operation.
+func (h *Handler) removeStreamOperation(ctx context.Context, id int64, filename string) error {
+	if hook := streamTestHooks.beforeRemove; hook != nil {
+		if err := hook(); err != nil {
+			return err
+		}
+	}
+	if id > 0 {
+		if err := h.repoOperation.Delete(ctx, id); err != nil && !errors.Is(err, ErrNotFound) {
+			return err
+		}
+	}
+	removeRecordingArtifacts(h.setting.Data, filename)
+	return nil
+}
+
 // streamEnded reports whether the database holds a saved recording of the
 // producer mission, which outlives the in-memory registry across restarts.
 func (h *Handler) streamEnded(missionID string) bool {
@@ -571,6 +833,11 @@ func (h *Handler) expireStream(ss *streamSession, gen uint64) {
 	if gen != 0 && (ss.owner != nil || gen != ss.detachGen) {
 		return // resumed, or detached again with a newer timer
 	}
+	if gen == 0 && ss.recovered.Load() {
+		// Recovery resumed it after the new start chose it: a resend from
+		// the spool is not superseded (see byProducer).
+		return
+	}
 	if ss.owner != nil {
 		ss.owner.kick()
 		ss.owner = nil
@@ -583,8 +850,15 @@ func (h *Handler) expireStream(ss *streamSession, gen uint64) {
 	if gen == 0 {
 		reason = "superseded by a new mission"
 	}
-	slog.Warn("stream: saving abandoned mission as partial", "missionId", ss.missionID, "reason", reason)
 	ss.done = true
+	if len(ss.counts) == 0 {
+		// Nothing was received after the start, for example a start the
+		// producer gave up waiting for.
+		slog.Info("stream: dropping abandoned mission that recorded nothing", "missionId", ss.missionID, "reason", reason)
+		h.dropEmptySession(ss, false)
+		return
+	}
+	slog.Warn("stream: saving abandoned mission as partial", "missionId", ss.missionID, "reason", reason)
 	_ = h.saveStream(ss, "partial")
 	h.streamRegistry().remove(ss, false)
 }
diff --git a/internal/server/handler_stream_discard_test.go b/internal/server/handler_stream_discard_test.go
new file mode 100644
index 00000000..a07d72ce
--- /dev/null
+++ b/internal/server/handler_stream_discard_test.go
@@ -0,0 +1,650 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"net"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/OCAP2/extension/v5/pkg/streaming"
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// recordingFiles lists what the data directory holds besides the database.
+func (rh *resumeHarness) recordingFiles() []string {
+	rh.t.Helper()
+	entries, err := os.ReadDir(rh.dir)
+	require.NoError(rh.t, err)
+	var out []string
+	for _, e := range entries {
+		if !strings.HasPrefix(e.Name(), "test.db") {
+			out = append(out, e.Name())
+		}
+	}
+	return out
+}
+
+// requireClosed checks that the server closed the connection.
+func requireClosed(t *testing.T, c *websocket.Conn) {
+	t.Helper()
+	c.SetReadDeadline(time.Now().Add(5 * time.Second))
+	for {
+		if _, _, err := c.ReadMessage(); err != nil {
+			requireNotTimeout(t, err)
+			return
+		}
+	}
+}
+
+// requireNotTimeout fails when a read ended by its deadline: the server was
+// expected to close the connection.
+func requireNotTimeout(t *testing.T, err error) {
+	t.Helper()
+	var netErr net.Error
+	require.False(t, errors.As(err, &netErr) && netErr.Timeout(), "the server did not close the connection: %v", err)
+}
+
+func TestStreamDiscard_RemovesRecording(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Discard", "m-d", "p-1")
+	last := sendStates(t, c, 2, 1, 5)
+	rh.waitSeq("m-d", last)
+	require.Len(t, rh.ops(), 1, "a streaming row exists while the mission records")
+
+	sendEnv(t, c, streaming.TypeDiscardMission, last+1, nil)
+	ack := readUntil(t, c, isAckFor(streaming.TypeDiscardMission))
+	assert.Equal(t, last+1, ack.Seq, "the discard is acknowledged with everything before it")
+	requireClosed(t, c)
+
+	assert.Empty(t, rh.ops())
+	assert.Empty(t, rh.recordingFiles(), "no files are left behind")
+	reg := rh.h.streamRegistry()
+	assert.Nil(t, reg.get("m-d"))
+	assert.True(t, reg.isCompleted("m-d"))
+	assert.Zero(t, reg.activeCount(), "the session's slot is released")
+}
+
+func TestStreamDiscard_LostAckResolvedByResume(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Lost Ack", "m-lost", "p-1")
+	sendEnv(t, c, streaming.TypeDiscardMission, 2, nil)
+	readUntil(t, c, isAckFor(streaming.TypeDiscardMission))
+
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-lost", ProducerID: "p-1"})
+	reply := readUntil(t, c2, isType(streaming.ServerResumeCompleted))
+	assert.Equal(t, uint8(streaming.ProtocolVersionDiscard), reply.ProtocolVersion)
+	assert.Empty(t, rh.ops())
+}
+
+func TestStreamDiscard_ResumedSession(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Resumed", "m-res", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-res", last)
+	c.Close()
+
+	// The producer finishes an unconfirmed discard on a new connection.
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-res", ProducerID: "p-1"})
+	reply := readUntil(t, c2, isType(streaming.ServerResumeOK))
+	assert.Equal(t, uint8(streaming.ProtocolVersionDiscard), reply.ProtocolVersion, "the producer learns the server still takes discard_mission")
+	sendEnv(t, c2, streaming.TypeDiscardMission, reply.Seq+1, nil)
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+
+	assert.Empty(t, rh.ops())
+	assert.Empty(t, rh.recordingFiles())
+	reg := rh.h.streamRegistry()
+	assert.True(t, reg.isCompleted("m-res"))
+	assert.Zero(t, reg.activeCount())
+}
+
+func TestStreamDiscard_RemovesEarlierSegments(t *testing.T) {
+	rh := newResumeHarness(t, 100*time.Millisecond)
+	c := rh.dial()
+	startResumable(t, c, "Segments", "m-seg", "p-1")
+	sendStates(t, c, 2, 1, 5)
+	rh.waitSeq("m-seg", 6)
+	c.Close()
+	// The producer stays away past the grace period: a partial segment is saved.
+	require.Eventually(t, func() bool { return rh.h.streamRegistry().get("m-seg") == nil }, 5*time.Second, 10*time.Millisecond)
+	require.Len(t, rh.opsByMission("m-seg"), 1)
+
+	// It returns, resends the mission and then discards it.
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-seg", ProducerID: "p-1"})
+	readUntil(t, c2, isType(streaming.ServerResumeUnknown))
+	startResumable(t, c2, "Segments", "m-seg", "p-1")
+	last := sendStates(t, c2, 2, 1, 8)
+	sendEnv(t, c2, streaming.TypeDiscardMission, last+1, nil)
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+
+	assert.Empty(t, rh.ops(), "every segment of the discarded mission is removed")
+	assert.Empty(t, rh.recordingFiles())
+}
+
+func TestStreamDiscard_LeavesOtherMissionsAlone(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	other := rh.dial()
+	replayMission(t, other, "Kept", "m-kept", "p-2", 3)
+
+	c := rh.dial()
+	startResumable(t, c, "Dropped", "m-drop", "p-1")
+	sendEnv(t, c, streaming.TypeDiscardMission, 2, nil)
+	readUntil(t, c, isAckFor(streaming.TypeDiscardMission))
+
+	ops := rh.ops()
+	require.Len(t, ops, 1)
+	assert.Equal(t, "Kept", ops[0].MissionName)
+}
+
+func TestStreamDiscard_WithoutSession(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	sendEnv(t, c, streaming.TypeDiscardMission, 0, nil)
+	readUntil(t, c, isAckFor(streaming.TypeDiscardMission))
+	requireClosed(t, c)
+}
+
+func TestStreamDiscard_LegacySession(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	sendStartMission(t, c, "Legacy", "altis")
+	sendEnv(t, c, streaming.TypeSoldierState, 0, soldierState(1))
+	sendEnv(t, c, streaming.TypeDiscardMission, 0, nil)
+	readUntil(t, c, isAckFor(streaming.TypeDiscardMission))
+	requireClosed(t, c)
+
+	assert.Empty(t, rh.ops(), "the disconnect that follows does not save it as partial")
+	assert.Empty(t, rh.recordingFiles())
+	assert.Zero(t, rh.h.streamRegistry().activeCount())
+}
+
+// gate holds the hook calls it is installed in until it opens; reached
+// closes on the first call. Any number of calls is safe, including a stray
+// one from a session an earlier test left behind.
+type gate struct {
+	reached             chan struct{}
+	release             chan struct{}
+	reachOnce, openOnce sync.Once
+}
+
+func newGate(t *testing.T) *gate {
+	g := &gate{reached: make(chan struct{}), release: make(chan struct{})}
+	t.Cleanup(g.open)
+	return g
+}
+
+func (g *gate) hold() {
+	g.reachOnce.Do(func() { close(g.reached) })
+	<-g.release
+}
+
+func (g *gate) open() { g.openOnce.Do(func() { close(g.release) }) }
+
+// discardPayload names a mission in discard_mission.
+func discardPayload(missionID, producerID string) streaming.DiscardMissionPayload {
+	return streaming.DiscardMissionPayload{MissionID: missionID, ProducerID: producerID}
+}
+
+func TestStreamDiscard_AfterSessionEndedRemovesPartial(t *testing.T) {
+	rh := newResumeHarness(t, 100*time.Millisecond)
+	c := rh.dial()
+	startResumable(t, c, "Expired", "m-exp", "p-1")
+	sendStates(t, c, 2, 1, 5)
+	rh.waitSeq("m-exp", 6)
+	c.Close()
+	// The discard never arrived and the grace expired: saved as partial.
+	require.Eventually(t, func() bool { return rh.h.streamRegistry().get("m-exp") == nil }, 5*time.Second, 10*time.Millisecond)
+	require.Len(t, rh.opsByMission("m-exp"), 1)
+
+	// The producer's recovery finishes the discard without a session.
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-exp", ProducerID: "p-1"})
+	reply := readUntil(t, c2, isType(streaming.ServerResumeUnknown))
+	assert.Equal(t, uint8(streaming.ProtocolVersionDiscard), reply.ProtocolVersion, "tells the producer it may discard without a session")
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-exp", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	requireClosed(t, c2)
+
+	assert.Empty(t, rh.ops())
+	assert.Empty(t, rh.recordingFiles())
+}
+
+func TestStreamDiscard_AfterRestartRemovesFailedRecording(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Restarted", "m-rst", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-rst", last)
+	c.Close()
+	rh.restart()
+	_, err := FailInterruptedStreams(context.Background(), rh.h.repoOperation)
+	require.NoError(t, err)
+	require.Len(t, rh.ops(), 1, "the interrupted recording is kept as failed")
+
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-rst", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	assert.Empty(t, rh.ops())
+}
+
+func TestStreamDiscard_AfterSessionEndedKeepsSavedRecording(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	replayMission(t, c, "Saved", "m-saved", "p-1", 3)
+
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-saved", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	require.Len(t, rh.ops(), 1, "a recording saved through end_mission is never discarded")
+}
+
+func TestStreamDiscard_NotAcknowledgedBeforeRemoval(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Slow", "m-slow-d", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-slow-d", last)
+
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	unblock := func() { releaseOnce.Do(func() { close(release) }) }
+	streamTestHooks.beforeRemove = func() error { <-release; return nil }
+	t.Cleanup(func() { unblock(); streamTestHooks.beforeRemove = nil })
+
+	msgs := make(chan streaming.AckMessage, 64)
+	go func() {
+		for {
+			var m streaming.AckMessage
+			if c.ReadJSON(&m) != nil {
+				close(msgs)
+				return
+			}
+			msgs <- m
+		}
+	}()
+
+	discardSeq := last + 1
+	sendEnv(t, c, streaming.TypeDiscardMission, discardSeq, discardPayload("m-slow-d", "p-1"))
+	// Several acknowledgement ticks pass while the removal is held.
+	hold := time.After(300 * time.Millisecond)
+holding:
+	for {
+		select {
+		case m := <-msgs:
+			require.Less(t, m.Seq, discardSeq, "discard_mission acknowledged before the recording was removed: %+v", m)
+		case <-hold:
+			break holding
+		}
+	}
+	unblock()
+	for m := range msgs {
+		if m.For == streaming.TypeDiscardMission {
+			assert.Equal(t, discardSeq, m.Seq)
+			assert.Empty(t, rh.ops())
+			return
+		}
+	}
+	t.Fatal("discard_mission was not acknowledged")
+}
+
+func TestStreamDiscard_FailedRemovalIsRepeated(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Locked", "m-fail", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-fail", last)
+
+	streamTestHooks.beforeRemove = func() error { return errors.New("database is locked") }
+	t.Cleanup(func() { streamTestHooks.beforeRemove = nil })
+	sendEnv(t, c, streaming.TypeDiscardMission, last+1, discardPayload("m-fail", "p-1"))
+	c.SetReadDeadline(time.Now().Add(5 * time.Second))
+	for {
+		var m streaming.AckMessage
+		err := c.ReadJSON(&m)
+		if err != nil {
+			require.True(t, websocket.IsCloseError(err, websocket.CloseInternalServerErr), "%v", err)
+			break
+		}
+		require.NotEqual(t, streaming.TypeDiscardMission, m.For, "a failed discard is not acknowledged")
+		require.Less(t, m.Seq, last+1)
+	}
+	ops := rh.ops()
+	require.Len(t, ops, 1)
+	assert.Equal(t, ConversionStatusFailed, ops[0].ConversionStatus, "no longer shown as live")
+	reg := rh.h.streamRegistry()
+	assert.False(t, reg.isCompleted("m-fail"))
+	assert.Zero(t, reg.activeCount())
+	streamTestHooks.beforeRemove = nil
+
+	// The producer repeats the discard: resume_unknown, then without a session.
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-fail", ProducerID: "p-1"})
+	readUntil(t, c2, isType(streaming.ServerResumeUnknown))
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-fail", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	assert.Empty(t, rh.ops())
+	assert.Empty(t, rh.recordingFiles())
+}
+
+// requireClosedUnacknowledged checks that the server closes the connection
+// with an error and never acknowledges discard_mission.
+func requireClosedUnacknowledged(t *testing.T, c *websocket.Conn) {
+	t.Helper()
+	c.SetReadDeadline(time.Now().Add(5 * time.Second))
+	for {
+		var m streaming.AckMessage
+		if err := c.ReadJSON(&m); err != nil {
+			requireNotTimeout(t, err)
+			return
+		}
+		require.NotEqual(t, streaming.TypeDiscardMission, m.For, "discard_mission must not be acknowledged")
+	}
+}
+
+func TestStreamDiscard_WithoutSessionDropsWaitingSession(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Waiting", "m-wait", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-wait", last)
+	c.Close()
+	reg := rh.h.streamRegistry()
+	rh.detachedSession("m-wait")
+
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-wait", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+
+	assert.Empty(t, rh.ops())
+	assert.Empty(t, rh.recordingFiles())
+	assert.Nil(t, reg.get("m-wait"))
+	assert.True(t, reg.isCompleted("m-wait"))
+	assert.Zero(t, reg.activeCount())
+}
+
+// detachedSession waits until the mission's session waits for its producer.
+func (rh *resumeHarness) detachedSession(missionID string) {
+	rh.t.Helper()
+	require.Eventually(rh.t, func() bool {
+		ss := rh.h.streamRegistry().get(missionID)
+		if ss == nil {
+			return false
+		}
+		ss.mu.Lock()
+		defer ss.mu.Unlock()
+		return ss.owner == nil
+	}, 5*time.Second, 10*time.Millisecond)
+}
+
+func TestStreamDiscard_WithoutSessionLeavesOtherProducersSession(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Theirs", "m-theirs", "p-1")
+	rh.waitSeq("m-theirs", sendStates(t, c, 2, 1, 3))
+	c.Close()
+	rh.detachedSession("m-theirs")
+
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-theirs", "p-2"))
+	requireClosedUnacknowledged(t, c2)
+	require.Len(t, rh.ops(), 1)
+	assert.NotNil(t, rh.h.streamRegistry().get("m-theirs"), "its producer may still resume it")
+}
+
+func TestStreamDiscard_WithoutSessionWaitsForExpiry(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Expiring", "m-expiring", "p-1")
+	rh.waitSeq("m-expiring", sendStates(t, c, 2, 1, 3))
+	c.Close()
+	rh.detachedSession("m-expiring")
+	ss := rh.h.streamRegistry().get("m-expiring")
+
+	// The grace expires and the partial is being saved when the discard
+	// arrives: it waits for the save, then removes the partial.
+	save := newGate(t)
+	streamTestHooks.beforeSave = save.hold
+	t.Cleanup(func() { streamTestHooks.beforeSave = nil })
+	go rh.h.expireStream(ss, 0)
+	<-save.reached
+
+	found := newGate(t)
+	found.open() // only notes that the discard got there
+	streamTestHooks.beforeDiscardSession = found.hold
+	t.Cleanup(func() { streamTestHooks.beforeDiscardSession = nil })
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-expiring", "p-1"))
+	<-found.reached // the discard found the session being saved
+	save.open()
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	assert.Empty(t, rh.ops(), "the partial saved meanwhile is removed")
+}
+
+func TestStreamDiscard_WithoutSessionKeepsRecordingSavedMeanwhile(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Saved", "m-saving", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-saving", last)
+
+	// end_mission is being saved when a session-less discard of the
+	// mission arrives: the saved recording stays.
+	save := newGate(t)
+	streamTestHooks.beforeSave = save.hold
+	t.Cleanup(func() { streamTestHooks.beforeSave = nil })
+	sendEnv(t, c, streaming.TypeEndMission, last+1, nil)
+	<-save.reached
+
+	found := newGate(t)
+	found.open()
+	streamTestHooks.beforeDiscardSession = found.hold
+	t.Cleanup(func() { streamTestHooks.beforeDiscardSession = nil })
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-saving", "p-1"))
+	<-found.reached
+	save.open()
+	readUntil(t, c, isAckFor(streaming.TypeEndMission))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	require.Len(t, rh.ops(), 1, "a recording saved through end_mission is never discarded")
+}
+
+func TestStreamDiscard_EndNotAcknowledgedWithoutEndedMark(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Unmarked", "m-unmarked", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-unmarked", last)
+
+	// The recording is saved, but marking it ended fails: the discard
+	// protection is not durable, so the save does not count.
+	streamTestHooks.beforeMarkEnded = func() error { return errors.New("database is locked") }
+	t.Cleanup(func() { streamTestHooks.beforeMarkEnded = nil })
+	sendEnv(t, c, streaming.TypeEndMission, last+1, nil)
+	c.SetReadDeadline(time.Now().Add(5 * time.Second))
+	for {
+		var m streaming.AckMessage
+		err := c.ReadJSON(&m)
+		if err != nil {
+			require.True(t, websocket.IsCloseError(err, websocket.CloseInternalServerErr), "%v", err)
+			break
+		}
+		require.NotEqual(t, streaming.TypeEndMission, m.For, "end_mission is not acknowledged without the mark")
+		require.Less(t, m.Seq, last+1)
+	}
+	assert.False(t, rh.h.streamRegistry().isCompleted("m-unmarked"))
+	streamTestHooks.beforeMarkEnded = nil
+
+	// The producer was never told the recording is kept: resume finds it
+	// unknown, and the producer's discard removes it.
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-unmarked", ProducerID: "p-1"})
+	readUntil(t, c2, isType(streaming.ServerResumeUnknown))
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-unmarked", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	assert.Empty(t, rh.ops())
+}
+
+func TestStreamDiscard_WithoutSessionHoldsOffStart(t *testing.T) {
+	rh := newResumeHarness(t, 100*time.Millisecond)
+	c := rh.dial()
+	startResumable(t, c, "Expired", "m-held", "p-1")
+	rh.waitSeq("m-held", sendStates(t, c, 2, 1, 3))
+	c.Close()
+	require.Eventually(t, func() bool { return rh.h.streamRegistry().get("m-held") == nil }, 5*time.Second, 10*time.Millisecond)
+
+	// The session-less discard is removing the partial when a start of the
+	// same mission arrives: the start waits for it.
+	remove := newGate(t)
+	streamTestHooks.beforeRemove = func() error { remove.hold(); return nil }
+	t.Cleanup(func() { streamTestHooks.beforeRemove = nil })
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-held", "p-1"))
+	<-remove.reached
+
+	c3 := rh.dial()
+	sendEnv(t, c3, streaming.TypeStartMission, 1, startPayload("Restarted", "m-held", "p-1"))
+	assert.Never(t, func() bool { return rh.h.streamRegistry().get("m-held") != nil }, 200*time.Millisecond, 10*time.Millisecond, "the start waits for the discard")
+	remove.open()
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	readUntil(t, c3, isAckFor(streaming.TypeStartMission))
+	ops := rh.ops()
+	require.Len(t, ops, 1, "only the new start's recording is left")
+	assert.Equal(t, "Restarted", ops[0].MissionName)
+}
+
+func TestStreamDiscard_FailedSegmentRemovalMarksRowFailed(t *testing.T) {
+	rh := newResumeHarness(t, 100*time.Millisecond)
+	c := rh.dial()
+	startResumable(t, c, "Earlier", "m-part", "p-1")
+	rh.waitSeq("m-part", sendStates(t, c, 2, 1, 3))
+	c.Close()
+	require.Eventually(t, func() bool { return rh.h.streamRegistry().get("m-part") == nil }, 5*time.Second, 10*time.Millisecond)
+
+	// A resend of the mission is discarded, but its earlier segment cannot go.
+	c2 := rh.dial()
+	startResumable(t, c2, "Resent", "m-part", "p-1")
+	last := sendStates(t, c2, 2, 1, 3)
+	rh.waitSeq("m-part", last)
+	streamTestHooks.beforeRemove = func() error { return errors.New("database is locked") }
+	t.Cleanup(func() { streamTestHooks.beforeRemove = nil })
+	sendEnv(t, c2, streaming.TypeDiscardMission, last+1, discardPayload("m-part", "p-1"))
+	requireClosedUnacknowledged(t, c2)
+
+	statuses := map[string]string{}
+	for _, op := range rh.ops() {
+		statuses[op.MissionName] = op.ConversionStatus
+	}
+	assert.Len(t, statuses, 2, "nothing was removed")
+	assert.Equal(t, ConversionStatusFailed, statuses["Resent"], "the discarded session's row is no longer live")
+}
+
+func TestStreamDiscard_WithoutSessionLeavesFedSession(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Fed", "m-fed", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	rh.waitSeq("m-fed", last)
+
+	other := rh.dial()
+	sendEnv(t, other, streaming.TypeDiscardMission, 0, discardPayload("m-fed", "p-1"))
+	requireClosedUnacknowledged(t, other)
+
+	require.Len(t, rh.ops(), 1, "the live recording stays")
+	sendEnv(t, c, streaming.TypeEndMission, last+1, nil)
+	readUntil(t, c, isAckFor(streaming.TypeEndMission))
+}
+
+func TestStreamDiscard_LosesToSupersede(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	startResumable(t, c, "Superseded", "m-old-d", "p-1")
+	last := sendStates(t, c, 2, 1, 3)
+	ss := rh.waitSeq("m-old-d", last)
+	ss.mu.Lock()
+	owner := ss.owner
+	ss.mu.Unlock()
+
+	// The producer's next mission supersedes this one before its discard is
+	// handled: the discard finds the session taken and is not acknowledged.
+	rh.h.expireStream(ss, 0)
+	env := streaming.Envelope{Type: streaming.TypeDiscardMission, Seq: last + 1}
+	assert.Equal(t, endTakenOver, rh.h.discardStream(ss, owner, env))
+	requireClosedUnacknowledged(t, c)
+	require.Len(t, rh.opsByMission("m-old-d"), 1, "saved as partial by the supersede")
+
+	// The producer finishes the discard without a session.
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-old-d", ProducerID: "p-1"})
+	readUntil(t, c2, isType(streaming.ServerResumeUnknown))
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-old-d", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	assert.Empty(t, rh.opsByMission("m-old-d"))
+}
+
+func TestStreamDiscard_WithoutSessionFailedRemoval(t *testing.T) {
+	rh := newResumeHarness(t, 100*time.Millisecond)
+	c := rh.dial()
+	startResumable(t, c, "Expired", "m-exp-f", "p-1")
+	rh.waitSeq("m-exp-f", sendStates(t, c, 2, 1, 3))
+	c.Close()
+	require.Eventually(t, func() bool { return rh.h.streamRegistry().get("m-exp-f") == nil }, 5*time.Second, 10*time.Millisecond)
+
+	streamTestHooks.beforeRemove = func() error { return errors.New("database is locked") }
+	t.Cleanup(func() { streamTestHooks.beforeRemove = nil })
+	c2 := rh.dial()
+	sendEnv(t, c2, streaming.TypeDiscardMission, 0, discardPayload("m-exp-f", "p-1"))
+	requireClosedUnacknowledged(t, c2)
+	require.Len(t, rh.ops(), 1)
+}
+
+func TestStreamDiscard_WithoutSessionInvalidPayload(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	for _, payload := range []any{map[string]any{"missionId": 7}, map[string]any{"producerId": "p-1"}} {
+		c := rh.dial()
+		sendEnv(t, c, streaming.TypeDiscardMission, 0, payload)
+		requireClosedUnacknowledged(t, c) // the producer keeps its spool
+	}
+}
+
+func TestStreamDiscard_KeepsSavedRecordingOfTheMission(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	replayMission(t, c, "Saved", "m-twice", "p-1", 3)
+
+	// A later attempt of the same mission id is discarded on its session.
+	c2 := rh.dial()
+	startResumable(t, c2, "Again", "m-twice", "p-1")
+	last := sendStates(t, c2, 2, 1, 2)
+	sendEnv(t, c2, streaming.TypeDiscardMission, last+1, discardPayload("m-twice", "p-1"))
+	readUntil(t, c2, isAckFor(streaming.TypeDiscardMission))
+	ops := rh.ops()
+	require.Len(t, ops, 1, "the recording saved through end_mission stays")
+	assert.Equal(t, "Saved", ops[0].MissionName)
+}
+
+func TestStreamDiscard_EmptyMissionRemovalFailure(t *testing.T) {
+	rh := newResumeHarness(t, 100*time.Millisecond)
+	streamTestHooks.beforeRemove = func() error { return errors.New("database is locked") }
+	t.Cleanup(func() { streamTestHooks.beforeRemove = nil })
+	c := rh.dial()
+	startResumable(t, c, "Empty", "m-empty-f", "p-1")
+	c.Close()
+
+	reg := rh.h.streamRegistry()
+	require.Eventually(t, func() bool { return reg.get("m-empty-f") == nil }, 5*time.Second, 10*time.Millisecond)
+	assert.Zero(t, reg.activeCount(), "the slot is released")
+	ops := rh.ops()
+	require.Len(t, ops, 1)
+	assert.Equal(t, ConversionStatusFailed, ops[0].ConversionStatus, "no longer shown as live")
+}
diff --git a/internal/server/handler_stream_resume_test.go b/internal/server/handler_stream_resume_test.go
index 6f71606b..b568c507 100644
--- a/internal/server/handler_stream_resume_test.go
+++ b/internal/server/handler_stream_resume_test.go
@@ -33,7 +33,30 @@ func newResumeHarness(t *testing.T, grace time.Duration) *resumeHarness {
 	h.setting.Streaming.AckInterval = 20 * time.Millisecond
 	srv := httptest.NewServer(mux)
 	t.Cleanup(srv.Close)
-	return &resumeHarness{t: t, h: h, srv: srv, dir: dir}
+	rh := &resumeHarness{t: t, h: h, srv: srv, dir: dir}
+	t.Cleanup(func() { retire(rh.h) })
+	return rh
+}
+
+// retire keeps a handler's sessions from ever being finalized, so a grace
+// timer a test leaves behind cannot fire into a later test's hooks.
+func retire(h *Handler) {
+	reg := h.streamRegistry()
+	reg.mu.Lock()
+	sessions := make([]*streamSession, 0, len(reg.sessions))
+	for _, ss := range reg.sessions {
+		sessions = append(sessions, ss)
+	}
+	reg.mu.Unlock()
+	for _, ss := range sessions {
+		ss.mu.Lock()
+		ss.done = true
+		if ss.grace != nil {
+			ss.grace.Stop()
+			ss.grace = nil
+		}
+		ss.mu.Unlock()
+	}
 }
 
 func (rh *resumeHarness) dial() *websocket.Conn {
@@ -133,7 +156,7 @@ func TestStreamResume_StartAckAdvertisesResume(t *testing.T) {
 	rh := newResumeHarness(t, time.Minute)
 	c := rh.dial()
 	ack := startResumable(t, c, "Advertise", "m-1", "p-1")
-	assert.Equal(t, uint8(streaming.ProtocolVersionResume), ack.ProtocolVersion)
+	assert.Equal(t, uint8(streaming.ProtocolVersionDiscard), ack.ProtocolVersion)
 	assert.Equal(t, uint64(1), ack.Seq)
 	assert.NotNil(t, rh.h.streamRegistry().get("m-1"))
 }
@@ -330,6 +353,7 @@ func TestStreamResume_NewMissionSupersedesProducersOldOne(t *testing.T) {
 	rh := newResumeHarness(t, time.Hour)
 	c := rh.dial()
 	startResumable(t, c, "Old", "m-old", "p-1")
+	rh.waitSeq("m-old", sendStates(t, c, 2, 1, 3))
 	other := rh.dial()
 	startResumable(t, other, "Other", "m-other", "p-2")
 	c.Close() // waiting for resume, an hour of grace
@@ -349,6 +373,144 @@ func TestStreamResume_NewMissionSupersedesProducersOldOne(t *testing.T) {
 	assert.Equal(t, "TvT,partial", old.Tag)
 }
 
+func TestStreamResume_AbandonedEmptyMissionIsDropped(t *testing.T) {
+	rh := newResumeHarness(t, 100*time.Millisecond)
+	c := rh.dial()
+	startResumable(t, c, "Expired empty", "m-empty-x", "p-1")
+	c.Close()
+	empty := rh.dial()
+	startResumable(t, empty, "Superseded empty", "m-empty-s", "p-2")
+	next := rh.dial()
+	startResumable(t, next, "Next", "m-next-e", "p-2")
+
+	reg := rh.h.streamRegistry()
+	require.Eventually(t, func() bool { return reg.get("m-empty-x") == nil && reg.get("m-empty-s") == nil }, 5*time.Second, 10*time.Millisecond)
+	ops := rh.ops()
+	require.Len(t, ops, 1, "neither the expired nor the superseded empty mission is kept")
+	assert.Equal(t, "Next", ops[0].MissionName)
+	assert.Equal(t, 1, reg.activeCount())
+}
+
+func TestStreamResume_InterruptedEmptyMissionIsDropped(t *testing.T) {
+	rh := newResumeHarness(t, time.Minute)
+	c := rh.dial()
+	// Recovery after a crash right after the start: nothing but the start.
+	start := startPayload("Crashed early", "m-crash-empty", "p-1")
+	start["recovered"] = true
+	sendEnv(t, c, streaming.TypeStartMission, 1, start)
+	readUntil(t, c, isAckFor(streaming.TypeStartMission))
+	sendEnv(t, c, streaming.TypeEndMission, 2, streaming.EndMissionPayload{Interrupted: true})
+	ack := readUntil(t, c, isAckFor(streaming.TypeEndMission))
+	assert.Equal(t, uint64(2), ack.Seq)
+
+	assert.Empty(t, rh.ops(), "an empty recording is not kept")
+	reg := rh.h.streamRegistry()
+	assert.True(t, reg.isCompleted("m-crash-empty"), "a lost ack is answered by resume_completed")
+	assert.Zero(t, reg.activeCount())
+}
+
+func TestStreamResume_RecoveredStartLeavesLiveMission(t *testing.T) {
+	rh := newResumeHarness(t, time.Hour)
+	live := rh.dial()
+	startResumable(t, live, "Live", "m-live", "p-1")
+	rh.waitSeq("m-live", sendStates(t, live, 2, 1, 3))
+
+	reg := rh.h.streamRegistry()
+	running := func(missionID string) bool {
+		ss := reg.get(missionID)
+		if ss == nil {
+			return false
+		}
+		ss.mu.Lock()
+		defer ss.mu.Unlock()
+		return !ss.done && ss.owner != nil
+	}
+
+	// The producer resends an older mission from its spool.
+	c := rh.dial()
+	start := startPayload("Spooled", "m-spooled", "p-1")
+	start["recovered"] = true
+	sendEnv(t, c, streaming.TypeStartMission, 1, start)
+	readUntil(t, c, isAckFor(streaming.TypeStartMission))
+	assert.Never(t, func() bool { return !running("m-live") }, 300*time.Millisecond, 10*time.Millisecond, "the live mission is not superseded")
+
+	// The producer's next live mission supersedes the live one, not the resend.
+	next := rh.dial()
+	startResumable(t, next, "Next", "m-next", "p-1")
+	require.Eventually(t, func() bool { return reg.get("m-live") == nil }, 5*time.Second, 10*time.Millisecond)
+	assert.Never(t, func() bool { return !running("m-spooled") }, 300*time.Millisecond, 10*time.Millisecond, "the resend is not superseded")
+	last := sendStates(t, c, 2, 1, 2)
+	sendEnv(t, c, streaming.TypeEndMission, last+1, streaming.EndMissionPayload{Interrupted: true})
+	readUntil(t, c, isAckFor(streaming.TypeEndMission))
+
+	tags := map[string]string{}
+	for _, op := range rh.ops() {
+		tags[op.MissionName] = op.Tag
+	}
+	assert.Equal(t, "TvT,partial", tags["Live"], "the superseded live mission is saved as partial")
+	assert.Equal(t, "TvT,partial", tags["Spooled"], "the resend is saved through its interrupted end")
+}
+
+func TestStreamResume_RecoveredResumeIsNotSuperseded(t *testing.T) {
+	rh := newResumeHarness(t, time.Hour)
+	c := rh.dial()
+	startResumable(t, c, "Left", "m-left", "p-1")
+	rh.waitSeq("m-left", sendStates(t, c, 2, 1, 3))
+	c.Close()
+
+	// The producer's recovery continues the mission from its spool...
+	rec := rh.dial()
+	sendEnv(t, rec, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-left", ProducerID: "p-1", Recovered: true})
+	readUntil(t, rec, isType(streaming.ServerResumeOK))
+
+	// ...while it starts recording its next mission.
+	next := rh.dial()
+	startResumable(t, next, "Next", "m-next-r", "p-1")
+	reg := rh.h.streamRegistry()
+	assert.Never(t, func() bool {
+		ss := reg.get("m-left")
+		if ss == nil {
+			return true
+		}
+		ss.mu.Lock()
+		defer ss.mu.Unlock()
+		return ss.done
+	}, 300*time.Millisecond, 10*time.Millisecond, "the resumed recovery is not superseded")
+	sendEnv(t, rec, streaming.TypeEndMission, 5, streaming.EndMissionPayload{Interrupted: true})
+	readUntil(t, rec, isAckFor(streaming.TypeEndMission))
+}
+
+func TestStreamResume_RecoveredResumeAfterSupersedeChoseIt(t *testing.T) {
+	rh := newResumeHarness(t, time.Hour)
+	c := rh.dial()
+	startResumable(t, c, "Left", "m-chosen", "p-1")
+	rh.waitSeq("m-chosen", sendStates(t, c, 2, 1, 3))
+	c.Close()
+	rh.detachedSession("m-chosen")
+
+	// The producer's next start chooses the waiting mission to supersede...
+	reg := rh.h.streamRegistry()
+	chosen := reg.byProducer("p-1", "m-next-c")
+	require.Len(t, chosen, 1)
+
+	// ...its recovery resumes the mission before the supersede runs...
+	rec := rh.dial()
+	sendEnv(t, rec, streaming.TypeResume, 0, streaming.ResumePayload{MissionID: "m-chosen", ProducerID: "p-1", Recovered: true})
+	readUntil(t, rec, isType(streaming.ServerResumeOK))
+
+	// ...and then the supersede runs.
+	rh.h.expireStream(chosen[0], 0)
+	ss := reg.get("m-chosen")
+	require.NotNil(t, ss, "the resumed recovery stays registered")
+	ss.mu.Lock()
+	done, owned := ss.done, ss.owner != nil
+	ss.mu.Unlock()
+	assert.False(t, done, "the resumed recovery is not saved")
+	assert.True(t, owned, "its connection is not closed")
+	sendEnv(t, rec, streaming.TypeEndMission, 5, streaming.EndMissionPayload{Interrupted: true})
+	readUntil(t, rec, isAckFor(streaming.TypeEndMission))
+}
+
 func TestStreamResume_RepeatedStartReusesSession(t *testing.T) {
 	rh := newResumeHarness(t, time.Minute)
 	c := rh.dial()
@@ -405,6 +567,7 @@ func TestStreamRegistryCompletedExpires(t *testing.T) {
 func (rh *resumeHarness) restart() {
 	rh.t.Helper()
 	rh.srv.Close()
+	retire(rh.h) // the old process is gone
 	h := &Handler{repoOperation: rh.h.repoOperation, setting: rh.h.setting}
 	mux := http.NewServeMux()
 	mux.HandleFunc("GET /api/v1/stream", h.HandleStream)
@@ -611,6 +774,10 @@ func TestStreamResume_SlotsReleasedOnEveryEnding(t *testing.T) {
 	old.Close()
 	superseding := rh.dial()
 	startResumable(t, superseding, "New", "m-new", "p-3") // supersedes m-old
+	discarded := rh.dial()
+	startResumable(t, discarded, "Discarded", "m-d", "p-4")
+	sendEnv(t, discarded, streaming.TypeDiscardMission, 2, nil)
+	readUntil(t, discarded, isAckFor(streaming.TypeDiscardMission))
 
 	require.Eventually(t, func() bool { return reg.activeCount() == 1 }, 5*time.Second, 10*time.Millisecond, "only the newest mission is still active")
 }
@@ -743,7 +910,7 @@ func TestStreamResume_ConcurrentRepeatedStartsShareOneSession(t *testing.T) {
 	unblock()
 
 	ack := readUntil(t, second, isAckFor(streaming.TypeStartMission))
-	assert.Equal(t, uint8(streaming.ProtocolVersionResume), ack.ProtocolVersion)
+	assert.Equal(t, uint8(streaming.ProtocolVersionDiscard), ack.ProtocolVersion)
 	assert.Len(t, rh.ops(), 1, "one recording for one mission")
 	assert.Equal(t, 1, rh.h.streamRegistry().activeCount(), "one admission for one mission")
 
diff --git a/internal/server/operation.go b/internal/server/operation.go
index dac2899b..7d63eec1 100644
--- a/internal/server/operation.go
+++ b/internal/server/operation.go
@@ -358,11 +358,22 @@ type StreamSegment struct {
 // StreamSegments returns the operations recorded for the producer mission,
 // except the one with id exceptID.
 func (r *RepoOperation) StreamSegments(ctx context.Context, missionID string, exceptID int64) ([]StreamSegment, error) {
+	return r.streamSegments(ctx, `stream_mission_id = ? AND id != ?`, missionID, exceptID)
+}
+
+// UnendedStreamSegments returns the operations recorded for the producer
+// mission that were not saved through end_mission, except the one with id
+// exceptID: partial and failed recordings, and live ones.
+func (r *RepoOperation) UnendedStreamSegments(ctx context.Context, missionID string, exceptID int64) ([]StreamSegment, error) {
+	return r.streamSegments(ctx, `stream_mission_id = ? AND id != ? AND stream_ended = 0`, missionID, exceptID)
+}
+
+func (r *RepoOperation) streamSegments(ctx context.Context, where string, missionID string, exceptID int64) ([]StreamSegment, error) {
 	if missionID == "" {
 		return nil, nil
 	}
 	rows, err := r.db.QueryContext(ctx,
-		`SELECT id, filename FROM operations WHERE stream_mission_id = ? AND id != ? ORDER BY id`, missionID, exceptID)
+		`SELECT id, filename FROM operations WHERE `+where+` ORDER BY id`, missionID, exceptID)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/server/stream_registry.go b/internal/server/stream_registry.go
index d53ea739..fb5baf0e 100644
--- a/internal/server/stream_registry.go
+++ b/internal/server/stream_registry.go
@@ -27,6 +27,15 @@ const streamReserveWait = 30 * time.Second
 var streamTestHooks struct {
 	beforeCreate func() // a start reserved its mission and is about to set it up
 	beforeSave   func() // a session is about to be finalized
+	// beforeRemove runs before a streamed recording is removed; an error
+	// stands in for a failed removal.
+	beforeRemove func() error
+	// beforeMarkEnded runs before a saved recording is marked ended; an
+	// error stands in for a failed write.
+	beforeMarkEnded func() error
+	// beforeDiscardSession runs before a session-less discard_mission
+	// handles a session of its mission.
+	beforeDiscardSession func()
 }
 
 // streamConn is one producer WebSocket connection.
@@ -52,6 +61,7 @@ type streamSession struct {
 	mu        sync.Mutex
 	missionID string // empty for legacy sessions
 	producer  string
+	recovered atomic.Bool // started or resumed by a resend from the producer's spool
 	session   *ingestion.Session
 	outputDir string
 	opID      int64
@@ -185,13 +195,15 @@ func (r *streamRegistry) isCompleted(missionID string) bool {
 	return ok && time.Since(at) <= completedTTL
 }
 
-// byProducer returns the producer's active sessions except keep.
+// byProducer returns the producer's active sessions except keep, and except
+// recovered ones: a resend from the spool is not the mission the producer
+// records live, which a new start supersedes.
 func (r *streamRegistry) byProducer(producer string, keep string) []*streamSession {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	var out []*streamSession
 	for id, s := range r.sessions {
-		if id != keep && s.producer == producer {
+		if id != keep && s.producer == producer && !s.recovered.Load() {
 			out = append(out, s)
 		}
 	}