diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go
index a5bc4e7..6f2aff5 100644
--- a/pkg/hold/pds/scan_broadcaster.go
+++ b/pkg/hold/pds/scan_broadcaster.go
@@ -6,9 +6,11 @@ import (
 	"database/sql"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
+	"net"
 	"net/http"
 	"net/url"
 	"strings"
@@ -95,6 +97,46 @@ const (
 	// a dispatch budget, so it is clamped rather than trusted.
 	maxScannerCapacity = 32
 
+	// scannerPingInterval is how often the hold pings an idle scanner
+	// connection.
+	scannerPingInterval = 30 * time.Second
+
+	// scannerPongWait is how long a connection may go without a pong (or any
+	// other frame) before the hold treats it as dead. Three missed pings.
+	scannerPongWait = 90 * time.Second
+
+	// scannerWriteWait bounds a single write to a scanner.
+	scannerWriteWait = 30 * time.Second
+
+	// scannerMaxMessageSize caps one frame from a scanner. gorilla's default is
+	// unlimited, so a broken or hostile peer can make the hold allocate
+	// whatever it likes.
+	//
+	// The ceiling has to be generous, because a result message carries the
+	// whole SBOM and the whole Grype report inline as JSON strings: exceeding
+	// the limit is not a truncation but a connection close, and a legitimate
+	// result that trips it becomes a permanent retry loop for that image. The
+	// largest report measured in this repo is 1.8 MB for 125 matches; 128 MiB
+	// is roughly seventy times the largest frame anyone has documented here,
+	// while still bounding a single allocation to something the host survives.
+	scannerMaxMessageSize = 128 << 20
+
+	// scannerResultTimeout bounds the blob uploads one result message costs.
+	scannerResultTimeout = 5 * time.Minute
+
+	// scannerRecordTimeout bounds the PDS record write a terminal message
+	// costs. It is deliberately a separate budget from scannerResultTimeout: a
+	// stalled S3 endpoint must not be able to spend the time the record write
+	// needs. The record is the only thing that stops the appview reporting the
+	// manifest as never scanned, so it is the last thing that may be skipped.
+	scannerRecordTimeout = 60 * time.Second
+
+	// scannerStorageQueueDepth is how many terminal messages one connection may
+	// have waiting on storage before the reader has to wait for room. It is
+	// small on purpose: each queued message holds an entire SBOM and Grype
+	// report in memory, and the hold shares a 1 GiB host with the scanner.
+	scannerStorageQueueDepth = 8
+
 	// maxScannerInstanceID bounds the scanner-supplied identity that becomes
 	// assigned_to. Same reasoning: it is client input that ends up in a
 	// database column.
@@ -128,13 +170,30 @@ type ScanBroadcaster struct {
 	ownsDB       bool   // true when this broadcaster opened the connection itself
 
 	// Proactive scan scheduling
-	rescanInterval   time.Duration   // Minimum interval between re-scans (0 = disabled)
-	stopCh           chan struct{}   // Signal to stop background goroutines
-	wg               sync.WaitGroup  // Wait for background goroutines to finish
-	predecessorCache map[string]bool // holdDID → "has this hold been migrated (has successor)?"
-	relayEndpoints   []string        // Relay URLs for listReposByCollection (failover order)
-	relayStartIdx    int             // Rotates per discovery pass so load is shared across endpoints
-	relayStartMu     sync.Mutex
+	rescanInterval time.Duration  // Minimum interval between re-scans (0 = disabled)
+	stopCh         chan struct{}  // Signal to stop background goroutines
+	wg             sync.WaitGroup // Wait for background goroutines to finish
+	// predecessorCache answers "has this hold been migrated into us?" without
+	// re-dialling. Only definitive answers belong here. The cache is never
+	// reset, so a false recorded from an unreachable hold would outlive the
+	// outage and stop that hold's manifests being scanned for the life of the
+	// process.
+	predecessorCache map[string]bool
+
+	// predecessorUnresolved holds the DIDs whose predecessor status could not
+	// be determined during the current discovery pass. It exists only so that
+	// one unreachable hold costs a single 5s timeout per pass rather than one
+	// per manifest, and it is cleared at the start of every pass so a hold that
+	// was down once is re-checked next time instead of being written off.
+	predecessorUnresolved map[string]bool
+
+	// predecessorMu guards both maps. Only the discovery goroutine reaches them
+	// today; the lock is what makes that a property of the code rather than of
+	// the current call graph.
+	predecessorMu  sync.Mutex
+	relayEndpoints []string // Relay URLs for listReposByCollection (failover order)
+	relayStartIdx  int      // Rotates per discovery pass so load is shared across endpoints
+	relayStartMu   sync.Mutex
 
 	// Work queues for proactive scanning (populated by discovery/stale goroutines)
 	unscannedQueue   chan *scanCandidate // Medium priority: manifests with no scan record
@@ -148,6 +207,47 @@ type ScanBroadcaster struct {
 	// activeJobsErrs counts consecutive hasActiveJobs query failures, so a
 	// persistent database fault can fail open instead of freezing dispatch.
 	activeJobsErrs atomic.Int64
+
+	// WebSocket liveness knobs. Zero means "use the package default"; tests
+	// shrink them so a half-open connection can be provoked in milliseconds
+	// rather than in the minute and a half production waits.
+	pingInterval time.Duration
+	pongWait     time.Duration
+	writeWait    time.Duration
+
+	// resultTimeout bounds the off-reader work a terminal message triggers.
+	// Zero means the package default.
+	resultTimeout time.Duration
+}
+
+// pingEvery, pongDeadline, writeDeadline and resultDeadline apply the package
+// defaults to the per-broadcaster knobs above.
+func (sb *ScanBroadcaster) pingEvery() time.Duration {
+	if sb.pingInterval > 0 {
+		return sb.pingInterval
+	}
+	return scannerPingInterval
+}
+
+func (sb *ScanBroadcaster) pongDeadline() time.Duration {
+	if sb.pongWait > 0 {
+		return sb.pongWait
+	}
+	return scannerPongWait
+}
+
+func (sb *ScanBroadcaster) writeDeadline() time.Duration {
+	if sb.writeWait > 0 {
+		return sb.writeWait
+	}
+	return scannerWriteWait
+}
+
+func (sb *ScanBroadcaster) resultDeadline() time.Duration {
+	if sb.resultTimeout > 0 {
+		return sb.resultTimeout
+	}
+	return scannerResultTimeout
 }
 
 // ScanSubscriber represents a connected scanner WebSocket client
@@ -161,6 +261,17 @@ type ScanSubscriber struct {
 	// connect. It is the unit both the proactive dispatch depth and
 	// per-scanner admission control are counted in.
 	capacity int
+
+	// storage carries the work a terminal message triggers off the reader
+	// goroutine, and readerDone says when nothing more will be submitted.
+	//
+	// One goroutine drains storage, so messages are handled in the order they
+	// arrived. That matters: two messages about one job must not be applied out
+	// of order, and the alternative — a pool that hashes seq to a worker — buys
+	// parallelism the PDS cannot use anyway, since CreateScanRecord serialises
+	// on the hold's single per-uid repo lock.
+	storage    chan func()
+	readerDone chan struct{}
 }
 
 // effectiveCapacity is capacity with the pre-declaration default applied.
@@ -533,10 +644,12 @@ func (sb *ScanBroadcaster) enqueue(job *ScanJobEvent, origin string) error {
 func (sb *ScanBroadcaster) Subscribe(conn *websocket.Conn, cursor int64, instanceID string, capacity int) *ScanSubscriber {
 	id := sb.subscriberID(instanceID)
 	sub := &ScanSubscriber{
-		conn: conn,
-		send: make(chan *ScanJobEvent, 20),
-		id:   id,
-		done: make(chan struct{}),
+		conn:       conn,
+		send:       make(chan *ScanJobEvent, 20),
+		id:         id,
+		done:       make(chan struct{}),
+		storage:    make(chan func(), scannerStorageQueueDepth),
+		readerDone: make(chan struct{}),
 	}
 	sub.capacity = capacity
 
@@ -557,9 +670,13 @@ func (sb *ScanBroadcaster) Subscribe(conn *websocket.Conn, cursor int64, instanc
 		"capacity", sub.effectiveCapacity(),
 		"totalSubscribers", total)
 
-	// Start writer goroutine (sends jobs to scanner)
+	// Start writer goroutine (sends jobs to scanner, and pings it)
 	go sb.handleWriter(sub)
 
+	// Start the storage goroutine before the reader, so the reader always has
+	// somewhere to hand a terminal message.
+	go sb.handleStorage(sub)
+
 	// Start reader goroutine (receives acks/results/errors from scanner)
 	go sb.handleReader(sub)
 
@@ -851,6 +968,12 @@ func (sb *ScanBroadcaster) subscriberLoads() (map[string]int, bool) {
 func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) {
 	defer sub.conn.Close()
 
+	// The writer owns every write on this connection, which is what makes the
+	// keepalive possible: gorilla permits one writer at a time, so the ping has
+	// to come from here rather than from a timer of its own.
+	ping := time.NewTicker(sb.pingEvery())
+	defer ping.Stop()
+
 	// done is closed by Unsubscribe, not here — ranging over send would block
 	// forever now that nothing closes it. handleReader always unsubscribes on
 	// its way out, so this goroutine is guaranteed to be released.
@@ -859,6 +982,20 @@ func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) {
 		select {
 		case <-sub.done:
 			return
+		case <-ping.C:
+			// Every write now carries a deadline. Without one a write into a
+			// socket whose peer has stopped reading blocks until the kernel
+			// gives up, which for a half-open connection is on the order of
+			// hours, and this goroutine is what the whole connection's
+			// liveness rests on.
+			if err := sub.conn.WriteControl(websocket.PingMessage, nil,
+				time.Now().Add(sb.writeDeadline())); err != nil {
+				slog.Warn("Failed to ping scanner, dropping the connection",
+					"subscriberId", sub.id, "error", err)
+				sb.Unsubscribe(sub)
+				return
+			}
+			continue
 		case job = <-sub.send:
 		}
 
@@ -868,6 +1005,12 @@ func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) {
 			continue
 		}
 
+		if err := sub.conn.SetWriteDeadline(time.Now().Add(sb.writeDeadline())); err != nil {
+			slog.Error("Failed to set write deadline for scan job",
+				"seq", job.Seq, "subscriberId", sub.id, "error", err)
+			sb.Unsubscribe(sub)
+			return
+		}
 		if err := sub.conn.WriteMessage(websocket.TextMessage, data); err != nil {
 			slog.Error("Failed to write scan job to WebSocket",
 				"seq", job.Seq,
@@ -879,20 +1022,63 @@ func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) {
 	}
 }
 
-// handleReader receives ack/result/error messages from a scanner
+// handleReader receives ack/result/error messages from a scanner.
+//
+// Two rules hold here, and they are the same rule seen from two sides: this
+// goroutine must never stop reading, and what it reads must prove the scanner
+// is alive.
+//
+// It never stops reading because the storage a terminal message triggers — two
+// S3 uploads and a CAR commit — is handed to handleStorage instead of being run
+// inline. It used to run here, and while it ran every other job's ack, started,
+// result and error on that connection sat unread in the socket buffer, with no
+// deadline on any of it.
+//
+// Liveness comes from the read deadline, refreshed by a pong and by any other
+// frame. Nothing used to ask: a connection that was open at this end and gone
+// at the other stayed in sb.subscribers for the life of the process, holding
+// its advertised worker count out of the dispatch budget and winning jobs that
+// could only ever time out. The deadline is three ping intervals, so it takes
+// three unanswered pings to condemn a connection.
 func (sb *ScanBroadcaster) handleReader(sub *ScanSubscriber) {
-	defer sb.Unsubscribe(sub)
+	defer func() {
+		// Before Unsubscribe, so handleStorage learns that nothing more will be
+		// submitted only once this goroutine has genuinely stopped submitting.
+		close(sub.readerDone)
+		sb.Unsubscribe(sub)
+	}()
+
+	sub.conn.SetReadLimit(scannerMaxMessageSize)
+	refreshRead := func() {
+		if err := sub.conn.SetReadDeadline(time.Now().Add(sb.pongDeadline())); err != nil {
+			slog.Debug("Failed to set scanner read deadline",
+				"subscriberId", sub.id, "error", err)
+		}
+	}
+	refreshRead()
+	sub.conn.SetPongHandler(func(string) error {
+		refreshRead()
+		return nil
+	})
 
 	for {
 		_, data, err := sub.conn.ReadMessage()
 		if err != nil {
-			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
+			switch {
+			case isReadTimeout(err):
+				slog.Warn("Scanner stopped answering, dropping the connection",
+					"subscriberId", sub.id,
+					"silentFor", sb.pongDeadline(),
+					"error", err)
+			case websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure):
 				slog.Error("Scanner WebSocket read error",
 					"subscriberId", sub.id,
 					"error", err)
 			}
 			return
 		}
+		// Any frame is proof of life, not just a pong.
+		refreshRead()
 
 		var msg ScannerMessage
 		if err := json.Unmarshal(data, &msg); err != nil {
@@ -904,19 +1090,73 @@ func (sb *ScanBroadcaster) handleReader(sub *ScanSubscriber) {
 
 		switch msg.Type {
 		case "ack":
+			// One guarded UPDATE. Cheap enough to stay on the reader, and
+			// keeping it here means an ack is never delayed behind a scan
+			// record write.
 			sb.handleAck(sub, msg.Seq)
 		case "started":
 			sb.handleStarted(sub, msg.Seq)
 		case "result":
-			sb.handleResult(sub, msg)
+			sb.submitStorage(sub, func() { sb.handleResult(sub, msg) })
 		case "error":
-			sb.handleError(sub, msg)
+			sb.submitStorage(sub, func() { sb.handleError(sub, msg) })
 		case "skipped":
-			sb.handleSkipped(sub, msg)
+			sb.submitStorage(sub, func() { sb.handleSkipped(sub, msg) })
 		default:
 			slog.Warn("Unknown scanner message type",
 				"type", msg.Type,
 				"subscriberId", sub.id)
+			continue
+		}
+
+		// Handing work over can block when storage is backed up, and time
+		// spent waiting for our own storage is not evidence against the
+		// scanner. Restart the clock rather than counting that wait against it.
+		refreshRead()
+	}
+}
+
+// isReadTimeout reports whether a read failed because the deadline expired
+// rather than because the peer or the network did something.
+func isReadTimeout(err error) bool {
+	var netErr net.Error
+	return errors.As(err, &netErr) && netErr.Timeout()
+}
+
+// submitStorage hands a terminal message's work to handleStorage.
+//
+// The send is a plain blocking one, and safe: handleReader is the only sender,
+// and handleStorage does not exit until handleReader has closed readerDone and
+// the queue has drained. A verdict is never dropped — it is the only thing that
+// stops the appview reporting the manifest as never scanned.
+func (sb *ScanBroadcaster) submitStorage(sub *ScanSubscriber, fn func()) {
+	sub.storage <- fn
+}
+
+// handleStorage runs the storage work terminal messages trigger, one at a time
+// and in the order the messages arrived.
+//
+// Serial and ordered is the point, not a limitation. Two messages about one job
+// applied out of order would let a stale verdict overwrite a fresh one, and the
+// concurrency there was worth nothing anyway: every one of these writes ends in
+// CreateScanRecord, which serialises on the hold's single per-uid repo lock.
+func (sb *ScanBroadcaster) handleStorage(sub *ScanSubscriber) {
+	for {
+		select {
+		case fn := <-sub.storage:
+			fn()
+		case <-sub.readerDone:
+			// The reader has stopped. Finish what it already handed over: those
+			// are verdicts the scanner really produced, and dropping one leaves
+			// its manifest looking unscanned until the next stale pass.
+			for {
+				select {
+				case fn := <-sub.storage:
+					fn()
+				default:
+					return
+				}
+			}
 		}
 	}
 }
@@ -1014,9 +1254,23 @@ func (sb *ScanBroadcaster) claimJobForTerminal(sub *ScanSubscriber, seq int64, k
 	return true
 }
 
-// handleResult processes a completed scan result: uploads SBOM blob + stores scan record in PDS
+// handleResult processes a completed scan result: uploads SBOM blob + stores scan record in PDS.
+//
+// It runs on handleStorage's goroutine, not on the reader's, and on a context
+// with a deadline. It used to be neither: an S3 endpoint that accepted the
+// connection and then stalled held the entire scanner connection hostage
+// indefinitely, and because the scanner was still nominally connected nothing
+// requeued the work either.
+//
+// The two budgets are separate on purpose. The uploads get scannerResultTimeout
+// between them; the record write gets its own fresh scannerRecordTimeout, so a
+// stalled upload cannot spend the time the record needs. The record is written
+// whichever way the uploads went — a blob already in S3 with no record pointing
+// at it is orphaned, and a manifest with no record at all reads as never
+// scanned.
 func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) {
-	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(context.Background(), sb.resultDeadline())
+	defer cancel()
 
 	// Before the S3 uploads, not after: a scanner that does not hold this job
 	// should not get its payload stored either.
@@ -1102,7 +1356,10 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage)
 		"atcr-scanner-v1.0.0",
 	)
 
-	rpath, _, err := sb.pds.CreateScanRecord(ctx, scanRecord)
+	recordCtx, recordCancel := context.WithTimeout(context.Background(), scannerRecordTimeout)
+	defer recordCancel()
+
+	rpath, _, err := sb.pds.CreateScanRecord(recordCtx, scanRecord)
 	if err != nil {
 		slog.Error("Failed to store scan record in PDS",
 			"seq", msg.Seq,
@@ -1157,7 +1414,11 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage)
 // loop won't immediately retry. Failed records still get retried on the
 // rescan interval since failures may be transient (network, OOM, etc.).
 func (sb *ScanBroadcaster) handleError(sub *ScanSubscriber, msg ScannerMessage) {
-	ctx := context.Background()
+	// Bounded, and off the reader goroutine: this writes a CAR delta under the
+	// hold's per-uid repo lock, which every layer record, stats increment and
+	// Bluesky post also contends for.
+	ctx, cancel := context.WithTimeout(context.Background(), scannerRecordTimeout)
+	defer cancel()
 
 	if !sb.claimJobForTerminal(sub, msg.Seq, "error") {
 		return
@@ -1206,7 +1467,8 @@ func (sb *ScanBroadcaster) handleError(sub *ScanSubscriber, msg ScannerMessage)
 // status="skipped". The stale-scan loop will leave these records alone — the
 // outcome won't change until the scanner gains support for the artifact type.
 func (sb *ScanBroadcaster) handleSkipped(sub *ScanSubscriber, msg ScannerMessage) {
-	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(context.Background(), scannerRecordTimeout)
+	defer cancel()
 
 	if !sb.claimJobForTerminal(sub, msg.Seq, "skipped") {
 		return
@@ -1706,6 +1968,12 @@ func (sb *ScanBroadcaster) runDiscoveryPass() {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
 	defer cancel()
 
+	// A hold that could not be reached last pass gets asked again this one.
+	// predecessorUnresolved is a within-pass memo, not a verdict.
+	sb.predecessorMu.Lock()
+	sb.predecessorUnresolved = nil
+	sb.predecessorMu.Unlock()
+
 	// Fetch DID list from relay
 	userDIDs := sb.fetchManifestDIDs(ctx)
 	if len(userDIDs) == 0 {
@@ -2238,6 +2506,15 @@ func (sb *ScanBroadcaster) resolveManifestForCandidate(ctx context.Context, cand
 
 // isOurManifest checks if a manifest's holdDID matches this hold directly,
 // or if the manifest's hold has been migrated (has a successor label set).
+//
+// An answer we could not get is not an answer of "no". A predecessor is by
+// definition a retired hold, so being briefly unreachable is its normal
+// condition, and this used to cache the resulting false forever: one slow reply
+// during the first discovery pass after boot meant every manifest naming that
+// hold went unscanned for the life of the process, silently, at Debug level.
+// pkg/hold/gc hit the same defect with a worse blast radius (deleted blobs
+// rather than missed scans) and fixed it with the definitive/unresolved split
+// this follows.
 func (sb *ScanBroadcaster) isOurManifest(ctx context.Context, holdDID string) bool {
 	if holdDID == "" {
 		return false
@@ -2248,20 +2525,48 @@ func (sb *ScanBroadcaster) isOurManifest(ctx context.Context, holdDID string) bo
 		return true
 	}
 
-	// Check predecessor cache
+	// Held across the fetch. The lock is uncontended today, and holding it also
+	// means two callers can never dial the same unreachable hold at once.
+	sb.predecessorMu.Lock()
+	defer sb.predecessorMu.Unlock()
+
 	if isPredecessor, cached := sb.predecessorCache[holdDID]; cached {
 		return isPredecessor
 	}
 
+	// Already unreachable earlier in this pass. Answer the same way without
+	// paying another timeout; the next pass starts fresh and re-checks.
+	if sb.predecessorUnresolved[holdDID] {
+		return false
+	}
+
 	// Fetch captain record from the other hold's PDS to check successor
-	isPredecessor := sb.checkPredecessor(ctx, holdDID)
+	isPredecessor, definitive := sb.checkPredecessor(ctx, holdDID)
+	if !definitive {
+		if sb.predecessorUnresolved == nil {
+			sb.predecessorUnresolved = make(map[string]bool)
+		}
+		sb.predecessorUnresolved[holdDID] = true
+		slog.Warn("Proactive scan: predecessor status unresolved, not adopting this hold's manifests this pass",
+			"holdDID", holdDID)
+		return false
+	}
+
+	if sb.predecessorCache == nil {
+		sb.predecessorCache = make(map[string]bool)
+	}
 	sb.predecessorCache[holdDID] = isPredecessor
 	return isPredecessor
 }
 
 // checkPredecessor fetches a hold's captain record to check if it has a successor label
 // (meaning the hold has been migrated/retired and its manifests should be scanned by us).
-func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) bool {
+//
+// The second return value reports whether the answer is definitive. It is false
+// whenever the hold could not be reached or its reply could not be understood:
+// those are cases where the hold may well be a predecessor and we simply cannot
+// tell. Callers must not read an inconclusive result as "not a predecessor".
+func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) (isPredecessor, definitive bool) {
 	fetchCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
@@ -2269,9 +2574,21 @@ func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string)
 	if err != nil {
 		slog.Debug("Proactive scan: failed to resolve predecessor hold URL",
 			"holdDID", holdDID, "error", err)
-		return false
+		return false, false
 	}
 
+	return sb.checkPredecessorAt(fetchCtx, holdDID, holdURL)
+}
+
+// checkPredecessorAt is checkPredecessor with the hold's base URL already
+// resolved, split out so the fetch-and-parse half can be exercised against a
+// local server. It carries the same contract: the second return value is false
+// whenever the answer is inconclusive rather than negative.
+//
+// A non-200 is inconclusive rather than negative on the same reasoning as
+// pkg/hold/gc: a reachable service that cannot produce its own captain record
+// is malfunctioning, not answering.
+func (sb *ScanBroadcaster) checkPredecessorAt(ctx context.Context, holdDID, holdURL string) (isPredecessor, definitive bool) {
 	// Fetch captain record: com.atproto.repo.getRecord
 	recordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self",
 		holdURL,
@@ -2279,53 +2596,63 @@ func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string)
 		url.QueryEscape(atproto.CaptainCollection),
 	)
 
-	req, err := http.NewRequestWithContext(fetchCtx, "GET", recordURL, nil)
+	req, err := http.NewRequestWithContext(ctx, "GET", recordURL, nil)
 	if err != nil {
-		return false
+		return false, false
 	}
 
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		slog.Debug("Proactive scan: failed to fetch predecessor captain record",
 			"holdDID", holdDID, "error", err)
-		return false
+		return false, false
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return false
+		slog.Debug("Proactive scan: predecessor captain record fetch returned non-200",
+			"holdDID", holdDID, "status", resp.StatusCode)
+		return false, false
 	}
 
 	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // 1MB limit
 	if err != nil {
-		return false
+		slog.Debug("Proactive scan: failed to read predecessor captain record",
+			"holdDID", holdDID, "error", err)
+		return false, false
 	}
 
 	var envelope struct {
 		Value json.RawMessage `json:"value"`
 	}
 	if err := json.Unmarshal(body, &envelope); err != nil {
-		return false
+		slog.Debug("Proactive scan: failed to parse predecessor captain envelope",
+			"holdDID", holdDID, "error", err)
+		return false, false
 	}
 
 	var captain atproto.CaptainRecord
 	if err := json.Unmarshal(envelope.Value, &captain); err != nil {
-		return false
+		slog.Debug("Proactive scan: failed to parse predecessor captain record",
+			"holdDID", holdDID, "error", err)
+		return false, false
 	}
 
+	// The hold answered and declares no successor. This is the one negative we
+	// are entitled to cache.
 	if captain.Successor == "" {
-		return false
+		return false, true
 	}
 
 	if captain.Successor != sb.holdDID {
 		slog.Debug("Proactive scan: hold has successor, but it is not us",
 			"holdDID", holdDID, "successor", captain.Successor, "ourHoldDID", sb.holdDID)
-		return false
+		return false, true
 	}
 
 	slog.Info("Proactive scan: discovered migrated hold pointing at us as successor",
 		"holdDID", holdDID, "successor", captain.Successor)
-	return true
+	return true, true
 }
 
 // hasConnectedScanners returns true if at least one scanner is connected.
diff --git a/pkg/hold/pds/scan_broadcaster_liveness_test.go b/pkg/hold/pds/scan_broadcaster_liveness_test.go
new file mode 100644
index 0000000..ee82f55
--- /dev/null
+++ b/pkg/hold/pds/scan_broadcaster_liveness_test.go
@@ -0,0 +1,453 @@
+package pds
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/gorilla/websocket"
+
+	"atcr.io/pkg/s3"
+)
+
+// wsConnPair is wsPair with the client side handed back too, because these
+// tests need to decide whether the client answers a ping and what it writes.
+// A gorilla connection only replies to a ping from inside ReadMessage, so a
+// client that never reads is exactly a scanner whose process is gone while its
+// TCP connection survives: NAT expiry, a load balancer, a host that vanished.
+func wsConnPair(t *testing.T) (server, client *websocket.Conn) {
+	t.Helper()
+
+	upgrader := websocket.Upgrader{}
+	accepted := make(chan *websocket.Conn, 1)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			t.Errorf("upgrade: %v", err)
+			return
+		}
+		accepted <- conn
+	}))
+	t.Cleanup(srv.Close)
+
+	cl, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(srv.URL, "http"), nil)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	t.Cleanup(func() { _ = cl.Close() })
+
+	select {
+	case conn := <-accepted:
+		return conn, cl
+	case <-time.After(5 * time.Second):
+		t.Fatal("server never completed the upgrade")
+		return nil, nil
+	}
+}
+
+// waitFor polls until cond is true, failing with msg if it never becomes true.
+func waitFor(t *testing.T, within time.Duration, msg string, cond func() bool) {
+	t.Helper()
+
+	deadline := time.Now().Add(within)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(2 * time.Millisecond)
+	}
+	t.Fatal(msg)
+}
+
+// TestScanBroadcaster_HalfOpenScannerIsDisconnected is the hold half of F7.
+//
+// Nothing in the hold ever asked a scanner connection whether it was still
+// there: no ping, no read deadline, no pong handler. A connection that is open
+// at this end and gone at the other therefore stays in sb.subscribers for as
+// long as the process lives. That is not merely untidy now that a scanner
+// declares a worker count on connect: the dead subscriber's advertised
+// capacity is subtracted from the dispatch budget, it keeps winning jobs in
+// selectSubscriberLocked, and every row it wins sits marked as its own until
+// the queueing timeout fires an hour later.
+//
+// The client here never calls ReadMessage, so it never answers a ping — the
+// half-open shape, reproduced without needing to break a network.
+func TestScanBroadcaster_HalfOpenScannerIsDisconnected(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	sb.pingInterval = 20 * time.Millisecond
+	sb.pongWait = 200 * time.Millisecond
+	sb.writeWait = time.Second
+
+	server, _ := wsConnPair(t)
+	sb.Subscribe(server, 0, "silent-scanner", 4)
+
+	waitFor(t, 5*time.Second,
+		"a scanner that answers no pings is still subscribed: its advertised "+
+			"capacity stays out of the dispatch budget and it keeps winning jobs",
+		func() bool { return subscriberCount(sb) == 0 })
+}
+
+// The other side of the same rule: a scanner that is merely idle must not be
+// disconnected. gorilla answers a ping from inside ReadMessage, so a client
+// running a read loop pongs without the test doing anything, which is what the
+// real scanner does.
+func TestScanBroadcaster_IdleButResponsiveScannerStaysSubscribed(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	sb.pingInterval = 20 * time.Millisecond
+	sb.pongWait = 200 * time.Millisecond
+	sb.writeWait = time.Second
+
+	server, client := wsConnPair(t)
+	go func() {
+		for {
+			if _, _, err := client.ReadMessage(); err != nil {
+				return
+			}
+		}
+	}()
+
+	sb.Subscribe(server, 0, "live-scanner", 4)
+
+	// Comfortably more than several pongWait periods. A liveness check that
+	// fires on silence rather than on unanswered pings would have cut this
+	// connection many times over by now.
+	time.Sleep(time.Second)
+	if got := subscriberCount(sb); got != 1 {
+		t.Fatalf("subscriber count = %d, want 1: a responsive but idle scanner was disconnected", got)
+	}
+}
+
+// A read limit must not be able to truncate a legitimate result. Results carry
+// the whole SBOM and the whole Grype report inline as JSON strings, and the
+// largest report measured in this repo is 1.8 MB for 125 matches, with a big
+// image's SBOM larger still. This sends a multi-megabyte SBOM through a real
+// socket and requires it to arrive whole and be stored.
+func TestScanBroadcaster_LargeResultIsNotTruncated(t *testing.T) {
+	sb := newRecordingScanBroadcaster(t)
+	sb.pingInterval = 50 * time.Millisecond
+	sb.pongWait = 5 * time.Second
+	sb.writeWait = 5 * time.Second
+
+	server, client := wsConnPair(t)
+	go func() {
+		for {
+			if _, _, err := client.ReadMessage(); err != nil {
+				return
+			}
+		}
+	}()
+
+	const digest = "sha256:bigsbom"
+	seq := seedJobWithDigest(t, sb, digest)
+
+	const instance = "big-result-scanner"
+	if _, err := sb.db.Exec(
+		`UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`,
+		instance, time.Now(), seq); err != nil {
+		t.Fatalf("assign: %v", err)
+	}
+
+	sb.Subscribe(server, 0, instance, 1)
+
+	// 12 MiB of SBOM: several times the largest report this repo has measured,
+	// and far past any frame size a naive limit would have picked.
+	big := `{"spdxVersion":"SPDX-2.3","filler":"` + strings.Repeat("a", 12<<20) + `"}`
+	if err := client.WriteJSON(map[string]any{
+		"type": "result", "seq": seq, "sbom": big,
+	}); err != nil {
+		t.Fatalf("write large result: %v", err)
+	}
+
+	waitFor(t, 30*time.Second,
+		"a large but legitimate result never completed: the read limit truncated it "+
+			"or the connection was closed under it",
+		func() bool { return jobStatus(t, sb, seq) == "completed" })
+
+	_, record, err := sb.pds.GetScanRecord(context.Background(), digest)
+	if err != nil {
+		t.Fatalf("no scan record for the large result: %v", err)
+	}
+	if record.SbomBlob == nil {
+		t.Error("scan record carries no SBOM blob")
+	}
+}
+
+// blockingS3 is a MockS3Client whose PutObject can be parked until the test
+// lets it go, which is how a slow S3 endpoint is reproduced without a network.
+//
+// blockFirst parks only the first upload. That asymmetry is what makes
+// out-of-order handling visible: if a later message is handled while an earlier
+// one is still uploading, the later one's upload runs straight through and
+// finishes first.
+type blockingS3 struct {
+	*s3.MockS3Client
+	entered    chan struct{}
+	release    chan struct{}
+	completed  chan int64 // ContentLength of each upload, as it finishes
+	blockFirst bool
+
+	mu   sync.Mutex
+	seen int
+}
+
+func (b *blockingS3) PutObject(ctx context.Context, in *awss3.PutObjectInput, opts ...func(*awss3.Options)) (*awss3.PutObjectOutput, error) {
+	b.mu.Lock()
+	b.seen++
+	first := b.seen == 1
+	b.mu.Unlock()
+
+	if !b.blockFirst || first {
+		// Buffered and non-blocking: one result can trigger two uploads, and
+		// the test only ever waits for the first.
+		select {
+		case b.entered <- struct{}{}:
+		default:
+		}
+		select {
+		case <-b.release:
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+
+	out, err := b.MockS3Client.PutObject(ctx, in, opts...)
+	if in.ContentLength != nil {
+		select {
+		case b.completed <- *in.ContentLength:
+		default:
+		}
+	}
+	return out, err
+}
+
+func newBlockingS3(t *testing.T, sb *ScanBroadcaster) *blockingS3 {
+	t.Helper()
+
+	b := &blockingS3{
+		MockS3Client: s3.NewMockS3Client(""),
+		entered:      make(chan struct{}, 64),
+		release:      make(chan struct{}),
+		completed:    make(chan int64, 64),
+	}
+	sb.s3 = &s3.S3Service{Client: b, Bucket: "test-bucket"}
+	return b
+}
+
+// TestScanBroadcaster_ReaderKeepsReadingDuringResultStorage is F6.
+//
+// handleResult ran inline on handleReader's goroutine: two S3 uploads and a CAR
+// commit, on a context.Background() with no deadline. For the whole of that the
+// subscriber's socket is not being read, so every other job's ack, started,
+// result and error sits in the kernel buffer. It also interacts badly with the
+// liveness work above — a scanner stalled behind a slow S3 upload looks exactly
+// like a scanner that died, because the pong that would prove otherwise is only
+// processed from inside ReadMessage.
+//
+// The interleaving is built, not raced: the S3 stand-in parks inside the upload
+// on an unbuffered channel and stays there until the test releases it.
+func TestScanBroadcaster_ReaderKeepsReadingDuringResultStorage(t *testing.T) {
+	sb := newRecordingScanBroadcaster(t)
+	blocker := newBlockingS3(t, sb)
+	sb.pingInterval = 50 * time.Millisecond
+	sb.pongWait = 10 * time.Second
+	sb.writeWait = 5 * time.Second
+
+	server, client := wsConnPair(t)
+	go func() {
+		for {
+			if _, _, err := client.ReadMessage(); err != nil {
+				return
+			}
+		}
+	}()
+
+	const instance = "busy-scanner"
+	slow := seedJobWithDigest(t, sb, "sha256:slowupload")
+	other := seedJobWithDigest(t, sb, "sha256:otherjob")
+	for _, seq := range []int64{slow, other} {
+		if _, err := sb.db.Exec(
+			`UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`,
+			instance, time.Now(), seq); err != nil {
+			t.Fatalf("assign %d: %v", seq, err)
+		}
+	}
+
+	sb.Subscribe(server, 0, instance, 2)
+
+	if err := client.WriteJSON(map[string]any{
+		"type": "result", "seq": slow, "sbom": testSBOM,
+	}); err != nil {
+		t.Fatalf("write result: %v", err)
+	}
+
+	// The upload is now parked. Whatever happens next happens while S3 is slow.
+	select {
+	case <-blocker.entered:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the SBOM upload never started")
+	}
+
+	if err := client.WriteJSON(map[string]any{"type": "ack", "seq": other}); err != nil {
+		t.Fatalf("write ack: %v", err)
+	}
+
+	waitFor(t, 5*time.Second,
+		"a second job's ack went unread while an S3 upload was in flight: the "+
+			"reader goroutine is doing storage work",
+		func() bool { return jobStatus(t, sb, other) == "processing" })
+
+	close(blocker.release)
+
+	waitFor(t, 10*time.Second,
+		"the released result never completed",
+		func() bool { return jobStatus(t, sb, slow) == "completed" })
+}
+
+// The bound on that off-reader work. An S3 endpoint that accepts the connection
+// and then stalls used to hold a result forever, because the context carried no
+// deadline at all. It must now give up and still leave the row in a terminal
+// state rather than in limbo.
+func TestScanBroadcaster_ResultStorageIsBounded(t *testing.T) {
+	sb := newRecordingScanBroadcaster(t)
+	blocker := newBlockingS3(t, sb)
+	sb.resultTimeout = 300 * time.Millisecond
+	sb.pingInterval = 50 * time.Millisecond
+	sb.pongWait = 10 * time.Second
+	sb.writeWait = 5 * time.Second
+
+	server, client := wsConnPair(t)
+	go func() {
+		for {
+			if _, _, err := client.ReadMessage(); err != nil {
+				return
+			}
+		}
+	}()
+
+	const instance = "stalled-s3-scanner"
+	seq := seedJobWithDigest(t, sb, "sha256:stalleds3")
+	if _, err := sb.db.Exec(
+		`UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`,
+		instance, time.Now(), seq); err != nil {
+		t.Fatalf("assign: %v", err)
+	}
+
+	sb.Subscribe(server, 0, instance, 1)
+
+	if err := client.WriteJSON(map[string]any{
+		"type": "result", "seq": seq, "sbom": testSBOM,
+	}); err != nil {
+		t.Fatalf("write result: %v", err)
+	}
+
+	select {
+	case <-blocker.entered:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the SBOM upload never started")
+	}
+	// blocker.release is never closed: S3 has stalled for good.
+
+	waitFor(t, 10*time.Second,
+		"a stalled S3 upload never gave up: the result work carries no deadline",
+		func() bool {
+			s := jobStatus(t, sb, seq)
+			return s == "completed" || s == "failed"
+		})
+
+	// The record write is unconditional. It is the only thing that stops the
+	// appview showing this manifest as never scanned, and it must not be
+	// spent out of the same budget the stalled upload just exhausted.
+	if _, _, err := sb.pds.GetScanRecord(context.Background(), "sha256:stalleds3"); err != nil {
+		t.Errorf("no scan record written after a stalled upload: %v", err)
+	}
+
+	// And the connection is still healthy: a slow backend is the hold's
+	// problem, not evidence against the scanner.
+	if got := subscriberCount(sb); got != 1 {
+		t.Errorf("subscriber count = %d, want 1: a slow S3 upload disconnected a healthy scanner", got)
+	}
+}
+
+// Moving the work off the reader must not lose any of it, and must not let two
+// messages about one job land out of order. Both results below name the same
+// job; the second is the newer verdict and must be applied after the first.
+//
+// Only the first upload is parked, which is what makes the assertion decisive
+// rather than lucky: a handler that starts the second result while the first is
+// still uploading finishes it immediately, because nothing is holding it back.
+func TestScanBroadcaster_ResultsForOneJobStayOrdered(t *testing.T) {
+	sb := newRecordingScanBroadcaster(t)
+	blocker := newBlockingS3(t, sb)
+	blocker.blockFirst = true
+	sb.pingInterval = 50 * time.Millisecond
+	sb.pongWait = 10 * time.Second
+	sb.writeWait = 5 * time.Second
+
+	server, client := wsConnPair(t)
+	go func() {
+		for {
+			if _, _, err := client.ReadMessage(); err != nil {
+				return
+			}
+		}
+	}()
+
+	const instance = "ordered-scanner"
+	seq := seedJobWithDigest(t, sb, "sha256:ordered")
+	if _, err := sb.db.Exec(
+		`UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`,
+		instance, time.Now(), seq); err != nil {
+		t.Fatalf("assign: %v", err)
+	}
+
+	sb.Subscribe(server, 0, instance, 1)
+
+	// Two SBOMs of different, recognisable sizes, and no vulnerability report,
+	// so each result produces exactly one upload.
+	const firstSize, secondSize = 4096, 8192
+	send := func(size int) {
+		t.Helper()
+		if err := client.WriteJSON(map[string]any{
+			"type": "result", "seq": seq, "sbom": strings.Repeat("x", size),
+		}); err != nil {
+			t.Fatalf("write result: %v", err)
+		}
+	}
+
+	send(firstSize)
+	// Park the first result inside its upload, then queue the second behind it.
+	select {
+	case <-blocker.entered:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the first upload never started")
+	}
+	send(secondSize)
+
+	// Nothing may finish while the first result is still in flight.
+	select {
+	case size := <-blocker.completed:
+		t.Fatalf("an upload of %d bytes finished while an earlier result for the "+
+			"same job was still in flight: a stale verdict can overwrite a fresh one", size)
+	case <-time.After(time.Second):
+	}
+
+	close(blocker.release)
+
+	for i, want := range []int64{firstSize, secondSize} {
+		select {
+		case got := <-blocker.completed:
+			if got != want {
+				t.Fatalf("upload %d was %d bytes, want %d: the two results were "+
+					"applied out of arrival order", i, got, want)
+			}
+		case <-time.After(15 * time.Second):
+			t.Fatalf("only %d of 2 results reached storage: a message was dropped", i)
+		}
+	}
+}
diff --git a/pkg/hold/pds/scan_broadcaster_predecessor_test.go b/pkg/hold/pds/scan_broadcaster_predecessor_test.go
new file mode 100644
index 0000000..be6b0c3
--- /dev/null
+++ b/pkg/hold/pds/scan_broadcaster_predecessor_test.go
@@ -0,0 +1,173 @@
+package pds
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+// TestScanPredecessor_UnreachableHoldIsNotCachedAsNegative is F5.
+//
+// isOurManifest cached whatever checkPredecessor returned, and checkPredecessor
+// returned a bare false for every failure mode: DID resolution failed, the HTTP
+// request failed, a non-200 came back, the body would not read, the JSON would
+// not parse. predecessorCache is never invalidated and never expires, so one
+// five-second timeout against a predecessor hold means every manifest naming
+// that hold is treated as not ours for the life of the process. A predecessor
+// is by definition a retired hold, so "briefly unreachable" is its normal
+// condition and the cache is filled on the first discovery pass after boot.
+//
+// pkg/hold/gc solved exactly this: a second return value saying whether the
+// answer is definitive, definitive answers cached, inconclusive ones parked in
+// a per-run map that is cleared each pass. This is the same rule for the scan
+// broadcaster.
+func TestScanPredecessor_UnreachableHoldIsNotCachedAsNegative(t *testing.T) {
+	sb := &ScanBroadcaster{
+		holdDID:          "did:web:us.example.com",
+		predecessorCache: make(map[string]bool),
+	}
+
+	// .invalid is reserved by RFC 2606 and never resolves, so the check cannot
+	// reach a conclusion.
+	const unreachable = "did:web:unreachable.invalid"
+
+	if sb.isOurManifest(context.Background(), unreachable) {
+		t.Fatal("an unreachable hold was adopted: ownership was asserted on no evidence")
+	}
+
+	if _, cached := sb.predecessorCache[unreachable]; cached {
+		t.Error("an inconclusive check was cached as a negative in predecessorCache, " +
+			"which is never reset: one blip means this hold's manifests are never " +
+			"scanned again for the life of the process")
+	}
+
+	if !sb.predecessorUnresolved[unreachable] {
+		t.Error("an inconclusive check was not recorded in predecessorUnresolved: " +
+			"the pass pays the five-second timeout again for every manifest naming this hold")
+	}
+
+	// The second call inside one pass is answered from predecessorUnresolved
+	// rather than re-dialled, and must agree with the first.
+	if sb.isOurManifest(context.Background(), unreachable) {
+		t.Error("the second call disagreed with the first")
+	}
+}
+
+// A pass boundary is what re-opens the question. predecessorUnresolved exists
+// only to stop one unreachable hold costing a timeout per manifest within a
+// single pass; a hold that was down during one pass must be asked again on the
+// next one.
+func TestScanPredecessor_UnresolvedIsClearedEachPass(t *testing.T) {
+	sb := &ScanBroadcaster{
+		holdDID:               "did:web:us.example.com",
+		predecessorCache:      make(map[string]bool),
+		predecessorUnresolved: map[string]bool{"did:web:wasdown.invalid": true},
+		stopCh:                make(chan struct{}),
+	}
+
+	// No relay endpoints, so the pass fetches no DIDs and returns immediately.
+	// Clearing has to happen before that, at the top of the pass.
+	sb.runDiscoveryPass()
+
+	if sb.predecessorUnresolved["did:web:wasdown.invalid"] {
+		t.Error("predecessorUnresolved survived a discovery pass: a hold that was " +
+			"down once is written off instead of being re-checked")
+	}
+}
+
+// The one negative worth caching: the hold answered, and said it has no
+// successor. That is an answer, not a failure, and repeating the request would
+// only cost time.
+func TestScanPredecessor_DefinitiveAnswersAreCached(t *testing.T) {
+	sb := &ScanBroadcaster{
+		holdDID:          "did:web:us.example.com",
+		predecessorCache: make(map[string]bool),
+	}
+
+	tests := []struct {
+		name           string
+		body           string
+		wantPredecesor bool
+		wantDefinitive bool
+	}{
+		{
+			name:           "no successor",
+			body:           `{"uri":"at://x","value":{"successor":""}}`,
+			wantPredecesor: false,
+			wantDefinitive: true,
+		},
+		{
+			name:           "successor is us",
+			body:           `{"uri":"at://x","value":{"successor":"did:web:us.example.com"}}`,
+			wantPredecesor: true,
+			wantDefinitive: true,
+		},
+		{
+			name:           "successor is a third hold",
+			body:           `{"uri":"at://x","value":{"successor":"did:web:elsewhere.example.com"}}`,
+			wantPredecesor: false,
+			wantDefinitive: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := newCaptainServer(t, 200, tc.body)
+			got, definitive := sb.checkPredecessorAt(context.Background(), "did:web:other.example.com", srv)
+			if got != tc.wantPredecesor || definitive != tc.wantDefinitive {
+				t.Errorf("checkPredecessorAt = (%v, %v), want (%v, %v)",
+					got, definitive, tc.wantPredecesor, tc.wantDefinitive)
+			}
+		})
+	}
+}
+
+// A reachable hold that cannot produce its own captain record is malfunctioning,
+// not answering. Reading a 500 as "not a predecessor" is the same permanent
+// mistake as reading a timeout that way.
+func TestScanPredecessor_NonDefinitiveReplies(t *testing.T) {
+	sb := &ScanBroadcaster{
+		holdDID:          "did:web:us.example.com",
+		predecessorCache: make(map[string]bool),
+	}
+
+	tests := []struct {
+		name   string
+		status int
+		body   string
+	}{
+		{"server error", 500, `{"error":"InternalServerError"}`},
+		{"not found", 404, `{"error":"RecordNotFound"}`},
+		{"unparseable envelope", 200, `not json at all`},
+		{"unparseable record", 200, `{"uri":"at://x","value":"not an object"}`},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := newCaptainServer(t, tc.status, tc.body)
+			isPredecessor, definitive := sb.checkPredecessorAt(context.Background(), "did:web:other.example.com", srv)
+			if definitive {
+				t.Errorf("reply treated as definitive; it would be cached forever "+
+					"(isPredecessor=%v)", isPredecessor)
+			}
+			if isPredecessor {
+				t.Error("an inconclusive reply reported as a predecessor")
+			}
+		})
+	}
+}
+
+// newCaptainServer serves one canned reply at the getRecord endpoint and
+// returns its base URL.
+func newCaptainServer(t *testing.T, status int, body string) string {
+	t.Helper()
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(status)
+		_, _ = w.Write([]byte(body))
+	}))
+	t.Cleanup(srv.Close)
+	return srv.URL
+}
diff --git a/pkg/hold/pds/scan_broadcaster_test.go b/pkg/hold/pds/scan_broadcaster_test.go
index 099e7b6..a9b9cfc 100644
--- a/pkg/hold/pds/scan_broadcaster_test.go
+++ b/pkg/hold/pds/scan_broadcaster_test.go
@@ -1,7 +1,10 @@
 package pds
 
 import (
+	"context"
 	"database/sql"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
@@ -44,7 +47,7 @@ func newRecordingScanBroadcaster(t *testing.T) *ScanBroadcaster {
 	sb.inflight = make(map[string]struct{})
 	sb.ackTimeout = 5 * time.Minute
 
-	pds, _ := setupTestPDS(t)
+	pds := setupScanTestPDS(t)
 	sb.pds = pds
 	sb.holdDID = pds.did
 	sb.s3 = &s3.S3Service{Client: s3.NewMockS3Client(""), Bucket: "test-bucket"}
@@ -52,6 +55,40 @@ func newRecordingScanBroadcaster(t *testing.T) *ScanBroadcaster {
 	return sb
 }
 
+// setupScanTestPDS is setupTestPDS on a file-backed database rather than
+// ":memory:".
+//
+// go-libsql's in-memory database is per *connection*, not per database, so the
+// moment two goroutines write through one *sql.DB the second one gets a
+// connection with no tables in it — "no such table: blocks" out of the
+// carstore. That did not matter while every scan record was written on the
+// caller's goroutine. It does now: terminal messages are stored on the
+// subscriber's storage goroutine, so the record write and the test's readback
+// are on different goroutines by construction.
+func setupScanTestPDS(t *testing.T) *HoldPDS {
+	t.Helper()
+
+	ctx := context.Background()
+	tmpDir := t.TempDir()
+
+	keyPath := filepath.Join(tmpDir, "signing-key")
+	if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil {
+		t.Fatalf("write signing key: %v", err)
+	}
+
+	pds, err := NewHoldPDS(ctx, "did:web:hold.example.com", "https://hold.example.com",
+		"https://atcr.io", filepath.Join(tmpDir, "pds.sqlite3"), keyPath, false)
+	if err != nil {
+		t.Fatalf("create test PDS: %v", err)
+	}
+	if err := pds.repomgr.InitNewActor(ctx, pds.uid, "", pds.did, "", "", ""); err != nil {
+		t.Fatalf("initialize test repo: %v", err)
+	}
+	t.Cleanup(func() { pds.Close() })
+
+	return pds
+}
+
 // newTestScanSubscriber mirrors what Subscribe builds, registered with the
 // broadcaster so Unsubscribe finds it.
 func newTestScanSubscriber(t *testing.T, sb *ScanBroadcaster, bufSize int) *ScanSubscriber {
diff --git a/scanner/internal/client/hold.go b/scanner/internal/client/hold.go
index d03290e..2fc79e5 100644
--- a/scanner/internal/client/hold.go
+++ b/scanner/internal/client/hold.go
@@ -22,6 +22,30 @@ import (
 	"github.com/gorilla/websocket"
 )
 
+const (
+	// holdPingInterval is how often the scanner pings an idle hold connection.
+	holdPingInterval = 30 * time.Second
+
+	// holdPongWait is how long the connection may go without a pong (or any
+	// other frame) before the scanner treats the hold as gone. Three missed
+	// pings.
+	holdPongWait = 90 * time.Second
+
+	// holdWriteWait bounds a single write to the hold.
+	holdWriteWait = 30 * time.Second
+
+	// holdMaxMessageSize caps one frame from the hold. gorilla's default is
+	// unlimited, so without it a broken or hostile hold decides how much this
+	// process allocates.
+	//
+	// A job frame is small — a config descriptor and a layer list — so 16 MiB
+	// is several orders of magnitude of headroom over anything the real hold
+	// sends. Note the asymmetry with the hold's own limit: results travel the
+	// other way and carry whole SBOMs inline, so the hold's ceiling has to be
+	// far higher than this one.
+	holdMaxMessageSize = 16 << 20
+)
+
 // httpClient is used for blob downloads and presigned URL requests
 // with a timeout to prevent stalled connections from leaking memory.
 var httpClient = &http.Client{Timeout: 5 * time.Minute}
@@ -46,6 +70,33 @@ type HoldClient struct {
 	// the dispatch budget for this connection; a hold that does not know the
 	// parameter ignores it and assumes one.
 	workers int
+
+	// WebSocket liveness knobs. Zero means "use the package default"; tests
+	// shrink them so a half-open connection can be provoked in milliseconds.
+	pingInterval time.Duration
+	pongWait     time.Duration
+	writeWait    time.Duration
+}
+
+func (c *HoldClient) pingEvery() time.Duration {
+	if c.pingInterval > 0 {
+		return c.pingInterval
+	}
+	return holdPingInterval
+}
+
+func (c *HoldClient) pongDeadline() time.Duration {
+	if c.pongWait > 0 {
+		return c.pongWait
+	}
+	return holdPongWait
+}
+
+func (c *HoldClient) writeDeadline() time.Duration {
+	if c.writeWait > 0 {
+		return c.writeWait
+	}
+	return holdWriteWait
 }
 
 // NewHoldClient creates a new hold client
@@ -153,6 +204,27 @@ func (c *HoldClient) connectOnce(cursor int64) error {
 
 	slog.Info("Connected to hold service")
 
+	// Liveness. This loop used to sit in ReadMessage with no deadline and send
+	// nothing, so a hold that went away without closing the TCP connection left
+	// this process permanently "connected": never reconnecting, so never
+	// re-draining the hold's pending rows, and writing every result it computed
+	// into a socket that went nowhere.
+	conn.SetReadLimit(holdMaxMessageSize)
+	refreshRead := func() {
+		if err := conn.SetReadDeadline(time.Now().Add(c.pongDeadline())); err != nil {
+			slog.Debug("Failed to set hold read deadline", "error", err)
+		}
+	}
+	refreshRead()
+	conn.SetPongHandler(func(string) error {
+		refreshRead()
+		return nil
+	})
+
+	stopPing := make(chan struct{})
+	defer close(stopPing)
+	go c.pingLoop(conn, stopPing)
+
 	// Read messages from hold
 	for {
 		_, data, err := conn.ReadMessage()
@@ -162,8 +234,37 @@ func (c *HoldClient) connectOnce(cursor int64) error {
 			}
 			return err
 		}
+		// Any frame is proof of life, not just a pong.
+		refreshRead()
 
 		c.handleFrame(data)
+		refreshRead()
+	}
+}
+
+// pingLoop keeps the connection provably alive from this end. gorilla permits
+// WriteControl concurrently with every other method, so this needs none of the
+// write mutex and cannot be delayed behind a result being sent.
+func (c *HoldClient) pingLoop(conn *websocket.Conn, stop <-chan struct{}) {
+	ticker := time.NewTicker(c.pingEvery())
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-stop:
+			return
+		case <-c.done:
+			return
+		case <-ticker.C:
+			if err := conn.WriteControl(websocket.PingMessage, nil,
+				time.Now().Add(c.writeDeadline())); err != nil {
+				// Closing is what unblocks the read loop, which then
+				// reconnects. Returning quietly would leave it parked.
+				slog.Warn("Failed to ping hold, dropping the connection", "error", err)
+				_ = conn.Close()
+				return
+			}
+		}
 	}
 }
 
@@ -303,6 +404,18 @@ func (c *HoldClient) SendSkipped(seq int64, reason string) {
 	c.sendJSON(scanner.SkippedMessage{Type: "skipped", Seq: seq, Reason: reason})
 }
 
+// sendJSON writes one message to the hold under the write mutex.
+//
+// The deadline is what stops a wedged write from taking the process with it:
+// a result is megabytes of SBOM, and a hold that has stopped reading fills the
+// socket buffer and blocks the write. Without a deadline that write blocks
+// until the kernel gives up on the connection, which is on the order of hours,
+// with the mutex held the whole time — so every worker that finishes a scan
+// queues up behind it and the scanner goes quiet without ever disconnecting.
+//
+// A failed write closes the connection rather than just logging. The read loop
+// is then released, connectOnce returns, and Connect redials; leaving it open
+// means the next write blocks the same way against the same dead socket.
 func (c *HoldClient) sendJSON(v any) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -312,8 +425,12 @@ func (c *HoldClient) sendJSON(v any) {
 		return
 	}
 
+	if err := c.conn.SetWriteDeadline(time.Now().Add(c.writeDeadline())); err != nil {
+		slog.Error("Failed to set WebSocket write deadline", "error", err)
+	}
 	if err := c.conn.WriteJSON(v); err != nil {
-		slog.Error("Failed to send WebSocket message", "error", err)
+		slog.Error("Failed to send WebSocket message, dropping the connection", "error", err)
+		_ = c.conn.Close()
 	}
 }
 
diff --git a/scanner/internal/client/hold_test.go b/scanner/internal/client/hold_test.go
new file mode 100644
index 0000000..e75e26d
--- /dev/null
+++ b/scanner/internal/client/hold_test.go
@@ -0,0 +1,156 @@
+package client
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"atcr.io/scanner/internal/queue"
+	"github.com/gorilla/websocket"
+)
+
+// upgradeOnly accepts a WebSocket and hands the server side to onConn, which
+// decides how lifelike the peer is. The returned URL is what a HoldClient dials.
+func upgradeOnly(t *testing.T, onConn func(*websocket.Conn)) string {
+	t.Helper()
+
+	upgrader := websocket.Upgrader{}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		onConn(conn)
+	}))
+	t.Cleanup(srv.Close)
+	return "ws" + strings.TrimPrefix(srv.URL, "http")
+}
+
+func newTestHoldClient(t *testing.T, url string) *HoldClient {
+	t.Helper()
+
+	c := NewHoldClient(url, "secret", queue.NewJobQueue(8))
+	c.pingInterval = 20 * time.Millisecond
+	c.pongWait = 200 * time.Millisecond
+	c.writeWait = time.Second
+	t.Cleanup(func() {
+		defer func() { _ = recover() }() // Close panics if already closed
+		c.Close()
+	})
+	return c
+}
+
+// TestHoldClient_DetectsHalfOpenConnection is the scanner half of F7.
+//
+// connectOnce sat in ReadMessage with no deadline and sent no pings, so a hold
+// that went away without closing the TCP connection left this scanner
+// permanently "connected": it never reconnects, never re-drains the hold's
+// pending rows, and every result it computes is written into a socket that goes
+// nowhere. The peer here upgrades and then does nothing at all — it never reads,
+// so it never pongs.
+func TestHoldClient_DetectsHalfOpenConnection(t *testing.T) {
+	held := make(chan struct{})
+	url := upgradeOnly(t, func(conn *websocket.Conn) {
+		<-held // hold the connection open, answering nothing
+		_ = conn.Close()
+	})
+	t.Cleanup(func() { close(held) })
+
+	c := newTestHoldClient(t, url)
+
+	errCh := make(chan error, 1)
+	go func() { errCh <- c.connectOnce(-1) }()
+
+	select {
+	case err := <-errCh:
+		if err == nil {
+			t.Fatal("connectOnce returned nil for a peer that answered nothing")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("a hold that answers no pings never disconnected the scanner: " +
+			"the client waits in ReadMessage with no deadline and sends no pings")
+	}
+}
+
+// The other side of the rule. An idle hold is the normal state — the dfd604b
+// commit message records a hold and scanner that shared one WebSocket for a
+// week — so silence alone must not drop the connection. gorilla answers a ping
+// from inside ReadMessage, which is what the real hold's reader does.
+func TestHoldClient_StaysConnectedWhileHoldAnswers(t *testing.T) {
+	url := upgradeOnly(t, func(conn *websocket.Conn) {
+		for {
+			if _, _, err := conn.ReadMessage(); err != nil {
+				_ = conn.Close()
+				return
+			}
+		}
+	})
+
+	c := newTestHoldClient(t, url)
+
+	errCh := make(chan error, 1)
+	go func() { errCh <- c.connectOnce(-1) }()
+
+	// Many pongWait periods with no traffic but pings.
+	select {
+	case err := <-errCh:
+		t.Fatalf("an idle but responsive hold was dropped: %v", err)
+	case <-time.After(time.Second):
+	}
+
+	c.Close()
+	select {
+	case <-errCh:
+	case <-time.After(5 * time.Second):
+		t.Fatal("connectOnce did not return after Close")
+	}
+}
+
+// A job frame is small — a config descriptor and a layer list — so the read
+// limit that bounds a hostile or broken hold must still admit anything the real
+// one sends with room to spare.
+func TestHoldClient_AcceptsARealisticJobFrame(t *testing.T) {
+	delivered := make(chan struct{})
+	url := upgradeOnly(t, func(conn *websocket.Conn) {
+		// A 128-layer image, which is well past the OCI-practical limit.
+		var layers []string
+		for i := 0; i < 128; i++ {
+			layers = append(layers,
+				`{"digest":"sha256:`+strings.Repeat("a", 64)+`","size":123456,"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip"}`)
+		}
+		frame := `{"type":"job","seq":1,"manifestDigest":"sha256:` + strings.Repeat("b", 64) + `",` +
+			`"repository":"repo","tag":"latest","userDid":"did:plc:user","userHandle":"user.example.com",` +
+			`"holdDid":"did:web:hold.example.com","holdEndpoint":"https://hold.example.com","tier":"deckhand",` +
+			`"config":{"digest":"sha256:` + strings.Repeat("c", 64) + `","size":1024,"mediaType":"application/vnd.oci.image.config.v1+json"},` +
+			`"layers":[` + strings.Join(layers, ",") + `]}`
+		if err := conn.WriteMessage(websocket.TextMessage, []byte(frame)); err != nil {
+			return
+		}
+		for {
+			if _, _, err := conn.ReadMessage(); err != nil {
+				_ = conn.Close()
+				return
+			}
+			select {
+			case delivered <- struct{}{}:
+			default:
+			}
+		}
+	})
+
+	c := newTestHoldClient(t, url)
+	c.pongWait = 5 * time.Second
+	go func() { _ = c.connectOnce(-1) }()
+
+	select {
+	case <-delivered:
+	case <-time.After(5 * time.Second):
+		t.Fatal("a realistic job frame was never acknowledged: the read limit rejected it")
+	}
+
+	if c.queue.Len() != 1 {
+		t.Errorf("queue depth = %d, want 1: the job frame did not survive the read", c.queue.Len())
+	}
+}