feat(diagnostics): chunked report upload fallback for proxy body caps (#494)

* feat(diagnostics): chunked report upload fallback for proxy body caps Diagnostics bundles can be up to max_bundle_bytes (10 MiB default), but a reverse proxy in front of Silo commonly caps request bodies at nginx's default client_max_body_size of 1 MiB. Such a proxy answers the single-shot multipart upload with its own 413 before Silo ever sees the request, so any report over the cap could never be delivered. Add a chunked upload fallback under /api/v1/diagnostics/reports/uploads: - POST / {manifest, bundle_bytes} opens a session - PUT /{id}/chunks/{index} streams one ≤768 KiB chunk (proxy-safe) - POST /{id}/complete ingests the assembled bundle - DELETE /{id} best-effort abandon The assembled bundle goes through the exact same Ingest path as the single-shot endpoint, so every content check (manifest contract, archive sha/bytes/entries, quotas, profile attribution) applies identically. Sessions reuse internal/uploads (the plugin chunked-upload spool manager) plus a small owner map for per-user isolation; they spool to disk, expire after 15 minutes, cap at one per user / 16 global, and complete shares the existing per-user + global in-flight ingest limiter. /diagnostics/status now advertises upload_chunk_bytes so clients can detect support; older servers omit the field and clients treat that as unsupported. The demo guard's diagnostics prefix gains PUT to cover the chunk route. Verified end to end against an OpenResty proxy with a 1m body cap: the single-shot upload 413s, the same 1.6 MiB bundle uploads in three chunks and lands as an accepted report; also exercised from the tvOS client's fallback path in the simulator. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(diagnostics): harden chunked upload sessions per review - Reserve the per-user slot and global cap atomically in init (a reservation map counted with live sessions), so concurrent inits by one account can no longer fan out past one session or transiently exceed the cap. Creation failures roll the reservation back. - Move chunk body I/O outside the uploads.Manager mutex: a slow client streaming one chunk no longer serializes every other session's chunk writes, completes, and cancels. A per-chunk in-flight flag rejects duplicate concurrent writes to the same offset (ErrChunkBusy → 409), and cancel/expiry defer spool-directory removal to the last finishing writer. - Chunk arrivals refresh the session expiry, making the TTL an idle timeout instead of an absolute deadline so a slow-but-progressing upload cannot expire mid-transfer. - Extend the request read deadline on chunk PUTs and both deadlines on complete, matching the single-shot handler's slow-uplink handling. - Keep the session when complete's availability re-check fails transiently (status load error → 500): only definitive disabled/storage-unavailable answers discard the spool, so a retried complete succeeds without re-uploading every chunk. - Reclaim orphaned spool directories at startup (a restart previously stranded the old process's partial uploads forever) and sweep expired sessions on a timer instead of only from later init traffic. - Document that session state is process-local and what that means for multi-replica deployments. Adds concurrency/race tests (go test -race) for atomic admission, same-chunk write exclusion, expiry refresh, transient-status retry, and startup reclaim. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(diagnostics): count detached chunk writers and lift chunk PUT write deadline Second review round: - A canceled session whose slow chunk writer was still draining held a connection and spool disk but vanished from every count, so a cancel-and-reinit loop could stack unbounded live writers behind the 16-session cap. The uploads manager now parks such sessions in a detached set (exposed as DetachedWriterSessions) until their last writer returns, and diagnostics init counts them in its admission gate. - Chunk PUTs now extend the write deadline as well as the read deadline: on an uplink slow enough to eat the server's 120s WriteTimeout, the stored chunk's JSON acknowledgement would otherwise be lost and the client would retry an already-accepted chunk. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-27 12:07:37 -04:00
parent 8dea4b9056
commit f235524365
10 changed files with 1388 additions and 62 deletions
@@ -8,6 +8,8 @@ import (
 	"mime"
 	"mime/multipart"
 	"net/http"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -41,18 +43,24 @@ type DiagnosticsService interface {
 }

 type DiagnosticsHandler struct {
-	service      DiagnosticsService
-	adminService AdminDiagnosticsService
-	inflight     *diagnosticsInFlightLimiter
-	logger       *slog.Logger
+	service       DiagnosticsService
+	adminService  AdminDiagnosticsService
+	inflight      *diagnosticsInFlightLimiter
+	chunkSessions *diagnosticsChunkSessions
+	logger        *slog.Logger
 }

 func NewDiagnosticsHandler(service DiagnosticsService) *DiagnosticsHandler {
 	handler := &DiagnosticsHandler{
-		service:  service,
-		inflight: newDiagnosticsInFlightLimiter(4),
-		logger:   slog.Default(),
+		service:       service,
+		inflight:      newDiagnosticsInFlightLimiter(4),
+		chunkSessions: newDiagnosticsChunkSessions(filepath.Join(os.TempDir(), "silo-diagnostics-uploads")),
+		logger:        slog.Default(),
 	}
+	// Reclaim abandoned chunk spool bytes on a timer, not only from later API
+	// traffic. The handler lives for the process, so the sweeper needs no
+	// stop signal.
+	handler.chunkSessions.startSweeper(nil)
 	if adminService, ok := service.(AdminDiagnosticsService); ok {
 		handler.adminService = adminService
 	}
@@ -72,9 +80,14 @@ func (h *DiagnosticsHandler) HandleStatus(w http.ResponseWriter, r *http.Request
 	writeJSON(w, http.StatusOK, status)
 }

-func (h *DiagnosticsHandler) HandleUpload(w http.ResponseWriter, r *http.Request) {
-	// Extend the read deadline for this request only; the server-wide
-	// ReadTimeout is too short for slow, large bundle uploads (see the constant).
+// extendDiagnosticsUploadDeadlines lifts this request's read (and, when
+// includeWrite, write) deadline to diagnosticsUploadReadTimeout. The
+// server-wide ReadTimeout (30s) is too short for slow bundle/chunk uploads,
+// and the WriteTimeout (120s) can expire after Ingest has already stored the
+// report — the lost 201 would make the client retry an upload that
+// succeeded. Shared by the single-shot upload, chunk PUT (read only), and
+// chunked complete (both).
+func (h *DiagnosticsHandler) extendDiagnosticsUploadDeadlines(w http.ResponseWriter, r *http.Request, includeWrite bool) {
 	rc := http.NewResponseController(w)
 	if err := rc.SetReadDeadline(time.Now().Add(diagnosticsUploadReadTimeout)); err != nil {
 		h.diagnosticsLogger().WarnContext(r.Context(), "diagnostics upload read deadline not extended",
@@ -82,17 +95,19 @@ func (h *DiagnosticsHandler) HandleUpload(w http.ResponseWriter, r *http.Request
 			"error", err,
 		)
 	}
-	// Extend the write deadline the same way. cmd/silo's integrated server sets
-	// WriteTimeout 120s, but a slow upload can take longer than that within the
-	// read window above; Ingest may store and mark the report ready and then the
-	// final writeJSON would miss the 120s write deadline, so the client sees a
-	// timeout and retries a report that already succeeded.
+	if !includeWrite {
+		return
+	}
 	if err := rc.SetWriteDeadline(time.Now().Add(diagnosticsUploadReadTimeout)); err != nil {
 		h.diagnosticsLogger().WarnContext(r.Context(), "diagnostics upload write deadline not extended",
 			"component", "diagnostics",
 			"error", err,
 		)
 	}
+}
+
+func (h *DiagnosticsHandler) HandleUpload(w http.ResponseWriter, r *http.Request) {
+	h.extendDiagnosticsUploadDeadlines(w, r, true)

 	userID, ok := diagnosticsUserID(w, r)
 	if !ok {
@@ -0,0 +1,482 @@
+package handlers
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+
+	"github.com/Silo-Server/silo-server/internal/diagnostics"
+	"github.com/Silo-Server/silo-server/internal/uploads"
+)
+
+// Chunked diagnostics upload: a fallback path for bundles that a reverse proxy
+// in front of Silo refuses as a single request (nginx's default
+// client_max_body_size is 1 MiB; /diagnostics/status advertises bundles up to
+// 10 MiB). The client:
+//
+//  1. POST   /diagnostics/reports/uploads                      {manifest, bundle_bytes}
+//  2. PUT    /diagnostics/reports/uploads/{id}/chunks/{index}  raw bundle bytes, ≤ upload_chunk_bytes each
+//  3. POST   /diagnostics/reports/uploads/{id}/complete        → same 201 as the single-shot upload
+//     DELETE /diagnostics/reports/uploads/{id}                 best-effort abandon
+//
+// The assembled bundle goes through the exact same Ingest path as the
+// single-shot endpoint, so every content check (manifest contract, archive
+// sha/bytes/entries, quotas, profile attribution) applies identically. The
+// manifest is validated only for size at init; Ingest judges it at complete.
+//
+// Sessions spool to local disk and expire after diagnosticsChunkSessionTTL of
+// inactivity (chunk arrivals refresh the deadline). They do NOT hold the
+// shared in-flight ingest slot while chunks stream in — only complete
+// acquires it, for the ingest itself — so a client that dies mid-upload
+// leaves nothing behind but spool bytes the TTL sweep reclaims, and never
+// blocks its user's later uploads. Concurrency is bounded by one session per
+// user (a new init replaces the previous session) plus a global session cap,
+// both reserved atomically in init.
+//
+// Session state (ownership map + spool files) is process-local. That is fine
+// for the integrated single-process server this feature targets; a
+// multi-replica API deployment would need sticky routing for the few requests
+// of one chunked upload, or a chunk PUT landing on another replica answers
+// 404 and the client restarts from init.
+const (
+	// diagnosticsChunkSessionTTL bounds how long an in-progress chunked upload
+	// may sit idle before its spooled bytes are reclaimed. Generous enough for
+	// a slow uplink to push 10 MiB in 768 KiB chunks, small enough that
+	// abandoned sessions don't hold disk for long.
+	diagnosticsChunkSessionTTL = 15 * time.Minute
+	// diagnosticsChunkSessionCap bounds concurrent chunked sessions across all
+	// users, capping worst-case spool disk at cap × max_bundle_bytes
+	// (≈ 160 MiB at defaults).
+	diagnosticsChunkSessionCap = 16
+)
+
+type diagnosticsChunkInitRequest struct {
+	// Manifest is the same part-1 manifest the multipart endpoint takes,
+	// embedded verbatim. Deferring all content validation to Ingest keeps one
+	// authority for what a valid manifest is.
+	Manifest json.RawMessage `json:"manifest"`
+	// BundleBytes is the exact assembled bundle size the client will upload.
+	// Declared up front so over-limit uploads fail at init, before any chunk
+	// bytes are spent.
+	BundleBytes int64 `json:"bundle_bytes"`
+}
+
+type diagnosticsChunkInitResponse struct {
+	UploadID    string `json:"upload_id"`
+	ChunkBytes  int64  `json:"chunk_bytes"`
+	TotalChunks int    `json:"total_chunks"`
+	ExpiresAt   string `json:"expires_at"`
+}
+
+type diagnosticsChunkStateResponse struct {
+	ReceivedChunks int `json:"received_chunks"`
+	TotalChunks    int `json:"total_chunks"`
+}
+
+// diagnosticsChunkSessions pairs the generic upload session manager with the
+// per-session diagnostics state it does not track: the manifest captured at
+// init and the owning user (sessions must not be readable or writable across
+// accounts).
+type diagnosticsChunkSessions struct {
+	manager *uploads.Manager
+
+	mu     sync.Mutex
+	owners map[string]diagnosticsChunkOwner
+	byUser map[int]string
+	// reserved counts init calls that have claimed a cap slot but not yet
+	// registered their created session in owners. Counting reservations and
+	// registrations together makes the per-user + global-cap admission atomic:
+	// concurrent inits cannot each pass the checks before any of them
+	// registers.
+	reserved map[int]struct{}
+}
+
+type diagnosticsChunkOwner struct {
+	userID   int
+	manifest []byte
+}
+
+func newDiagnosticsChunkSessions(spoolDir string) *diagnosticsChunkSessions {
+	sessions := &diagnosticsChunkSessions{
+		manager: uploads.NewManager(uploads.ManagerOptions{
+			RootDir:      spoolDir,
+			TTL:          diagnosticsChunkSessionTTL,
+			MaxChunkSize: diagnostics.UploadChunkBytes,
+			// MaxSize is enforced per-init against the live max_bundle_bytes
+			// setting; the manager-level bound is just a hard backstop.
+			MaxSize: 256 << 20,
+		}),
+		owners:   make(map[string]diagnosticsChunkOwner),
+		byUser:   make(map[int]string),
+		reserved: make(map[int]struct{}),
+	}
+	// A restart leaves the previous process's spool directories on disk with
+	// no session map entry to ever expire them; reclaim them now.
+	sessions.manager.ReclaimOrphanedDirs()
+	return sessions
+}
+
+// startSweeper begins the timed expiry sweep. Split from the constructor so
+// tests can run without background goroutines.
+func (s *diagnosticsChunkSessions) startSweeper(stop <-chan struct{}) {
+	s.manager.StartExpirySweeper(time.Minute, stop)
+}
+
+// owner returns the session owner entry when id belongs to userID.
+func (s *diagnosticsChunkSessions) owner(id string, userID int) (diagnosticsChunkOwner, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	owner, ok := s.owners[id]
+	if !ok || owner.userID != userID {
+		return diagnosticsChunkOwner{}, false
+	}
+	return owner, true
+}
+
+// reserve atomically claims the user's session slot and a global cap slot,
+// evicting the user's own previous session first (its id is returned for the
+// caller to cancel outside the lock — the caller cancels it whether or not
+// admission succeeds, since the eviction already happened). A false capOK
+// means the cap is full, or a concurrent init by the same user holds an
+// unfinished reservation — that racing call must not create a second session.
+//
+// The cap counts live sessions, unfinished reservations, AND the manager's
+// detached-writer sessions: a canceled session whose slow chunk writer is
+// still draining holds real disk and a connection until the writer's read
+// deadline, so a cancel-and-reinit loop must stall at the cap rather than
+// stack unbounded live writers behind it.
+func (s *diagnosticsChunkSessions) reserve(userID int) (previousID string, capOK bool) {
+	detached := s.manager.DetachedWriterSessions()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, inFlight := s.reserved[userID]; inFlight {
+		return "", false
+	}
+	if id, ok := s.byUser[userID]; ok {
+		previousID = id
+		delete(s.owners, id)
+		delete(s.byUser, userID)
+	}
+	if len(s.owners)+len(s.reserved)+detached >= diagnosticsChunkSessionCap {
+		return previousID, false
+	}
+	s.reserved[userID] = struct{}{}
+	return previousID, true
+}
+
+// commit registers the created session under an existing reservation.
+func (s *diagnosticsChunkSessions) commit(userID int, id string, manifest []byte) {
+	s.mu.Lock()
+	delete(s.reserved, userID)
+	s.owners[id] = diagnosticsChunkOwner{userID: userID, manifest: manifest}
+	s.byUser[userID] = id
+	s.mu.Unlock()
+}
+
+// unreserve rolls back a reservation whose session creation failed.
+func (s *diagnosticsChunkSessions) unreserve(userID int) {
+	s.mu.Lock()
+	delete(s.reserved, userID)
+	s.mu.Unlock()
+}
+
+// drop removes the owner entry. Idempotent; the manager session is the
+// caller's to cancel.
+func (s *diagnosticsChunkSessions) drop(id string) {
+	s.mu.Lock()
+	owner, ok := s.owners[id]
+	delete(s.owners, id)
+	if ok && s.byUser[owner.userID] == id {
+		delete(s.byUser, owner.userID)
+	}
+	s.mu.Unlock()
+}
+
+// sweepExpired drops owner entries whose manager session has expired or is
+// gone, so the owner map (and the cap headroom it counts against) doesn't
+// leak alongside the manager's own spool cleanup. Called from init, the only
+// entry point that needs the freed headroom.
+func (s *diagnosticsChunkSessions) sweepExpired() {
+	s.mu.Lock()
+	stale := make([]string, 0, len(s.owners))
+	for id := range s.owners {
+		if _, err := s.manager.Peek(id); errors.Is(err, uploads.ErrNotFound) || errors.Is(err, uploads.ErrExpired) {
+			stale = append(stale, id)
+		}
+	}
+	s.mu.Unlock()
+	for _, id := range stale {
+		s.drop(id)
+	}
+}
+
+// HandleChunkedUploadInit handles POST /diagnostics/reports/uploads.
+func (h *DiagnosticsHandler) HandleChunkedUploadInit(w http.ResponseWriter, r *http.Request) {
+	userID, ok := diagnosticsUserID(w, r)
+	if !ok {
+		return
+	}
+
+	status, ok := h.diagnosticsUploadStatus(w, r, userID)
+	if !ok {
+		return
+	}
+	maxBundleBytes := status.MaxBundleBytes
+	if maxBundleBytes <= 0 {
+		maxBundleBytes = diagnostics.DefaultMaxBundleBytes
+	}
+
+	// Manifest cap plus a small envelope allowance keeps init requests tiny —
+	// they must themselves fit under restrictive proxy body caps.
+	r.Body = http.MaxBytesReader(w, r.Body, diagnostics.MaxManifestBytes+diagnosticsMultipartOverheadBytes)
+	var req diagnosticsChunkInitRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		var maxBytesErr *http.MaxBytesError
+		if errors.As(err, &maxBytesErr) {
+			h.logRejected(r.Context(), userID, "too_large")
+			writeError(w, http.StatusRequestEntityTooLarge, "too_large", "Diagnostics manifest is too large")
+			return
+		}
+		writeError(w, http.StatusBadRequest, "invalid_bundle", "Invalid diagnostics upload init")
+		return
+	}
+	if len(req.Manifest) == 0 || int64(len(req.Manifest)) > diagnostics.MaxManifestBytes {
+		h.logRejected(r.Context(), userID, "too_large")
+		writeError(w, http.StatusRequestEntityTooLarge, "too_large", "Diagnostics manifest is too large")
+		return
+	}
+	if req.BundleBytes <= 0 {
+		writeError(w, http.StatusBadRequest, "invalid_bundle", "bundle_bytes must be positive")
+		return
+	}
+	if req.BundleBytes > maxBundleBytes {
+		h.logRejected(r.Context(), userID, "too_large")
+		writeError(w, http.StatusRequestEntityTooLarge, "too_large", "Diagnostics upload is too large")
+		return
+	}
+
+	h.chunkSessions.sweepExpired()
+
+	// Atomically evict the user's previous session (a new init replaces it,
+	// so a client that died mid-upload can start over immediately) and claim
+	// a cap slot. Cancel the evicted spool outside the lock either way.
+	previousID, capOK := h.chunkSessions.reserve(userID)
+	if previousID != "" {
+		h.abortChunkedUpload(previousID)
+	}
+	if !capOK {
+		h.logRejected(r.Context(), userID, "busy")
+		w.Header().Set("Retry-After", diagnosticsBusyRetryAfter)
+		writeError(w, http.StatusServiceUnavailable, "busy", "Diagnostics upload capacity is busy")
+		return
+	}
+
+	session, err := h.chunkSessions.manager.Create(uploads.CreateRequest{
+		Filename:  "bundle.tar.gz",
+		SizeBytes: req.BundleBytes,
+		ChunkSize: diagnostics.UploadChunkBytes,
+	})
+	if err != nil {
+		h.chunkSessions.unreserve(userID)
+		statusCode, message := uploadErrorResponse(err)
+		writeError(w, statusCode, "upload_error", message)
+		return
+	}
+	h.chunkSessions.commit(userID, session.ID, append([]byte(nil), req.Manifest...))
+
+	writeJSON(w, http.StatusCreated, diagnosticsChunkInitResponse{
+		UploadID:    session.ID,
+		ChunkBytes:  session.ChunkSize,
+		TotalChunks: session.TotalChunks,
+		ExpiresAt:   session.ExpiresAt.UTC().Format(time.RFC3339),
+	})
+}
+
+// HandleChunkedUploadChunk handles PUT /diagnostics/reports/uploads/{upload_id}/chunks/{chunk_index}.
+func (h *DiagnosticsHandler) HandleChunkedUploadChunk(w http.ResponseWriter, r *http.Request) {
+	// The integrated server's 30s ReadTimeout kills a 768 KiB chunk arriving
+	// below ~26 KiB/s — exactly the slow uplinks the chunked fallback exists
+	// for. The write deadline needs lifting too: the 120s WriteTimeout starts
+	// at request start, so on a sufficiently slow uplink the stored chunk's
+	// JSON acknowledgement would miss it and the client would retry an
+	// already-accepted chunk.
+	h.extendDiagnosticsUploadDeadlines(w, r, true)
+
+	userID, ok := diagnosticsUserID(w, r)
+	if !ok {
+		return
+	}
+	uploadID := chi.URLParam(r, "upload_id")
+	chunkIndex, err := strconv.Atoi(chi.URLParam(r, "chunk_index"))
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", "Invalid chunk index")
+		return
+	}
+	if _, ok := h.chunkSessions.owner(uploadID, userID); !ok {
+		writeError(w, http.StatusNotFound, "upload_error", "Upload session not found")
+		return
+	}
+
+	r.Body = http.MaxBytesReader(w, r.Body, diagnostics.UploadChunkBytes+1)
+	defer r.Body.Close()
+
+	session, err := h.chunkSessions.manager.PutChunk(r.Context(), uploadID, chunkIndex, r.Body, r.ContentLength)
+	if err != nil {
+		if errors.Is(err, uploads.ErrNotFound) || errors.Is(err, uploads.ErrExpired) {
+			h.chunkSessions.drop(uploadID)
+		}
+		statusCode, message := uploadErrorResponse(err)
+		writeError(w, statusCode, "upload_error", message)
+		return
+	}
+	writeJSON(w, http.StatusOK, diagnosticsChunkStateResponse{
+		ReceivedChunks: session.ReceivedChunks,
+		TotalChunks:    session.TotalChunks,
+	})
+}
+
+// HandleChunkedUploadComplete handles POST /diagnostics/reports/uploads/{upload_id}/complete.
+func (h *DiagnosticsHandler) HandleChunkedUploadComplete(w http.ResponseWriter, r *http.Request) {
+	// Ingest + object-store write can outlast the server's 120s WriteTimeout;
+	// without the extension the stored report's 201 would be lost and the
+	// client would retry an upload that already succeeded (see the single-shot
+	// handler).
+	h.extendDiagnosticsUploadDeadlines(w, r, true)
+
+	userID, ok := diagnosticsUserID(w, r)
+	if !ok {
+		return
+	}
+	uploadID := chi.URLParam(r, "upload_id")
+	owner, ok := h.chunkSessions.owner(uploadID, userID)
+	if !ok {
+		writeError(w, http.StatusNotFound, "upload_error", "Upload session not found")
+		return
+	}
+
+	// Availability can have changed since init (admin toggle, storage loss);
+	// re-check so a completed spool is not ingested into a disabled feature.
+	// Only a definitive non-available answer discards the session — a
+	// transient status load failure (500) keeps it so a retried complete can
+	// succeed without re-uploading every chunk.
+	status, statusErr := h.service.Status(r.Context(), userID)
+	if statusErr != nil {
+		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load diagnostics status")
+		return
+	}
+	if status.Status != diagnostics.StatusAvailable {
+		h.abortChunkedUpload(uploadID)
+		switch status.Status {
+		case diagnostics.StatusDisabled:
+			writeError(w, http.StatusForbidden, "disabled", "Diagnostics uploads are disabled")
+		default:
+			writeError(w, http.StatusServiceUnavailable, "storage_unavailable", "Diagnostics storage is not configured")
+		}
+		return
+	}
+
+	// The ingest itself shares capacity with single-shot uploads. Check
+	// before consuming the manager session so a busy answer leaves the
+	// session intact for a client-side retry of complete.
+	release, acquired := h.inflight.acquire(userID)
+	if !acquired {
+		h.logRejected(r.Context(), userID, "busy")
+		w.Header().Set("Retry-After", diagnosticsBusyRetryAfter)
+		writeError(w, http.StatusServiceUnavailable, "busy", "Diagnostics upload capacity is busy")
+		return
+	}
+	defer release()
+
+	upload, err := h.chunkSessions.manager.Complete(uploadID)
+	if err != nil {
+		if errors.Is(err, uploads.ErrNotFound) || errors.Is(err, uploads.ErrExpired) {
+			h.chunkSessions.drop(uploadID)
+		}
+		statusCode, message := uploadErrorResponse(err)
+		writeError(w, statusCode, "upload_error", message)
+		return
+	}
+	// The manager session is consumed; a failed ingest is retried by the
+	// client from a fresh init, never by re-completing a spent session.
+	defer h.chunkSessions.drop(uploadID)
+	defer upload.Cleanup()
+
+	bundle, err := os.Open(upload.Path)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "internal_error", "Diagnostics upload failed")
+		return
+	}
+	defer bundle.Close()
+
+	profileID := strings.TrimSpace(r.Header.Get("X-Profile-Id"))
+	var profileIDPtr *string
+	if profileID != "" {
+		profileIDPtr = &profileID
+	}
+
+	result, err := h.service.Ingest(r.Context(), userID, profileIDPtr, owner.manifest, io.Reader(bundle))
+	if err != nil {
+		writeDiagnosticsServiceError(w, err)
+		return
+	}
+	writeJSON(w, http.StatusCreated, result)
+}
+
+// HandleChunkedUploadAbort handles DELETE /diagnostics/reports/uploads/{upload_id}.
+func (h *DiagnosticsHandler) HandleChunkedUploadAbort(w http.ResponseWriter, r *http.Request) {
+	userID, ok := diagnosticsUserID(w, r)
+	if !ok {
+		return
+	}
+	uploadID := chi.URLParam(r, "upload_id")
+	if _, ok := h.chunkSessions.owner(uploadID, userID); !ok {
+		// Unknown or foreign session: abort is a best-effort courtesy, and
+		// idempotent success leaks nothing about other users' session ids.
+		w.WriteHeader(http.StatusNoContent)
+		return
+	}
+	h.abortChunkedUpload(uploadID)
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func (h *DiagnosticsHandler) abortChunkedUpload(uploadID string) {
+	if err := h.chunkSessions.manager.Cancel(uploadID); err != nil && !errors.Is(err, uploads.ErrNotFound) {
+		h.diagnosticsLogger().Warn("diagnostics chunked upload cancel failed",
+			"component", "diagnostics",
+			"error", err,
+		)
+	}
+	h.chunkSessions.drop(uploadID)
+}
+
+// diagnosticsUploadStatus loads the feature status and writes the
+// disabled/storage-unavailable rejection when uploads cannot proceed,
+// mirroring the single-shot endpoint's gate.
+func (h *DiagnosticsHandler) diagnosticsUploadStatus(w http.ResponseWriter, r *http.Request, userID int) (diagnostics.Status, bool) {
+	status, err := h.service.Status(r.Context(), userID)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load diagnostics status")
+		return diagnostics.Status{}, false
+	}
+	switch status.Status {
+	case diagnostics.StatusDisabled:
+		writeError(w, http.StatusForbidden, "disabled", "Diagnostics uploads are disabled")
+		return diagnostics.Status{}, false
+	case diagnostics.StatusStorageUnavailable:
+		writeError(w, http.StatusServiceUnavailable, "storage_unavailable", "Diagnostics storage is not configured")
+		return diagnostics.Status{}, false
+	case diagnostics.StatusAvailable:
+		return status, true
+	default:
+		writeError(w, http.StatusServiceUnavailable, "storage_unavailable", "Diagnostics storage is not available")
+		return diagnostics.Status{}, false
+	}
+}
@@ -0,0 +1,424 @@
+package handlers
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/go-chi/chi/v5"
+
+	apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
+	"github.com/Silo-Server/silo-server/internal/auth"
+	"github.com/Silo-Server/silo-server/internal/diagnostics"
+)
+
+var errFakeStatusUnavailable = errors.New("status store temporarily unavailable")
+
+// newChunkedTestHandler returns a handler whose chunk spool lives in the
+// test's temp dir so parallel test runs never share state.
+func newChunkedTestHandler(t *testing.T, service DiagnosticsService) *DiagnosticsHandler {
+	t.Helper()
+	handler := NewDiagnosticsHandler(service)
+	handler.chunkSessions = newDiagnosticsChunkSessions(t.TempDir())
+	return handler
+}
+
+func chunkedRouter(handler *DiagnosticsHandler) chi.Router {
+	r := chi.NewRouter()
+	r.Post("/uploads", handler.HandleChunkedUploadInit)
+	r.Put("/uploads/{upload_id}/chunks/{chunk_index}", handler.HandleChunkedUploadChunk)
+	r.Post("/uploads/{upload_id}/complete", handler.HandleChunkedUploadComplete)
+	r.Delete("/uploads/{upload_id}", handler.HandleChunkedUploadAbort)
+	return r
+}
+
+func doChunked(t *testing.T, router chi.Router, method, path string, body []byte, claims *auth.Claims) *httptest.ResponseRecorder {
+	t.Helper()
+	req := httptest.NewRequest(method, path, bytes.NewReader(body))
+	req.Header.Set("Authorization", "Bearer token")
+	if method == http.MethodPost && strings.HasSuffix(path, "/uploads") {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	if claims != nil {
+		req = req.WithContext(apimw.SetClaims(req.Context(), claims))
+	}
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, req)
+	return rec
+}
+
+func initChunkedUpload(t *testing.T, router chi.Router, bundleBytes int, claims *auth.Claims) diagnosticsChunkInitResponse {
+	t.Helper()
+	body := []byte(fmt.Sprintf(`{"manifest":{"ok":true},"bundle_bytes":%d}`, bundleBytes))
+	rec := doChunked(t, router, http.MethodPost, "/uploads", body, claims)
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("init status = %d, want %d; body=%s", rec.Code, http.StatusCreated, rec.Body.String())
+	}
+	var resp diagnosticsChunkInitResponse
+	if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
+		t.Fatalf("decode init response: %v", err)
+	}
+	return resp
+}
+
+func TestDiagnosticsChunkedUploadHappyPath(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	bundle := bytes.Repeat([]byte("b"), int(diagnostics.UploadChunkBytes)+512)
+	session := initChunkedUpload(t, router, len(bundle), accessClaims())
+	if session.ChunkBytes != diagnostics.UploadChunkBytes {
+		t.Fatalf("chunk_bytes = %d, want %d", session.ChunkBytes, diagnostics.UploadChunkBytes)
+	}
+	if session.TotalChunks != 2 {
+		t.Fatalf("total_chunks = %d, want 2", session.TotalChunks)
+	}
+
+	for index := 0; index < session.TotalChunks; index++ {
+		start := int64(index) * session.ChunkBytes
+		end := start + session.ChunkBytes
+		if end > int64(len(bundle)) {
+			end = int64(len(bundle))
+		}
+		rec := doChunked(t, router, http.MethodPut,
+			fmt.Sprintf("/uploads/%s/chunks/%d", session.UploadID, index),
+			bundle[start:end], accessClaims())
+		if rec.Code != http.StatusOK {
+			t.Fatalf("chunk %d status = %d; body=%s", index, rec.Code, rec.Body.String())
+		}
+	}
+
+	rec := doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("complete status = %d; body=%s", rec.Code, rec.Body.String())
+	}
+	var resp diagnostics.IngestResult
+	if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
+		t.Fatalf("decode complete response: %v", err)
+	}
+	if resp.ShortID != "SILO-ABCDEF123456" {
+		t.Fatalf("short_id = %q", resp.ShortID)
+	}
+	if service.ingestCalls != 1 {
+		t.Fatalf("ingest calls = %d, want 1", service.ingestCalls)
+	}
+	if got := string(service.lastManifest); got != `{"ok":true}` {
+		t.Fatalf("manifest passed to ingest = %q", got)
+	}
+	if int(service.lastBundleBytes) != len(bundle) {
+		t.Fatalf("bundle bytes passed to ingest = %d, want %d", service.lastBundleBytes, len(bundle))
+	}
+
+	// The ingest slot must be free again after completion.
+	release, acquired := handler.inflight.acquire(42)
+	if !acquired {
+		t.Fatal("in-flight slot still held after complete")
+	}
+	release()
+}
+
+func TestDiagnosticsChunkedUploadInitOverBundleLimit(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	service.status.MaxBundleBytes = 1024
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	rec := doChunked(t, router, http.MethodPost, "/uploads",
+		[]byte(`{"manifest":{"ok":true},"bundle_bytes":2048}`), accessClaims())
+	assertDiagnosticsError(t, rec, http.StatusRequestEntityTooLarge, "too_large")
+
+	// Rejected init must not leave the limiter slot held.
+	release, acquired := handler.inflight.acquire(42)
+	if !acquired {
+		t.Fatal("in-flight slot held after rejected init")
+	}
+	release()
+}
+
+func TestDiagnosticsChunkedUploadInitDisabled(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	service.status.Status = diagnostics.StatusDisabled
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	rec := doChunked(t, router, http.MethodPost, "/uploads",
+		[]byte(`{"manifest":{"ok":true},"bundle_bytes":10}`), accessClaims())
+	assertDiagnosticsError(t, rec, http.StatusForbidden, "disabled")
+}
+
+func TestDiagnosticsChunkedUploadForeignSessionRejected(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	session := initChunkedUpload(t, router, 10, accessClaims())
+
+	other := &auth.Claims{UserID: 7, TokenType: auth.TokenTypeAccess}
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		[]byte("0123456789"), other)
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("foreign chunk status = %d, want 404; body=%s", rec.Code, rec.Body.String())
+	}
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, other)
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("foreign complete status = %d, want 404", rec.Code)
+	}
+	// A foreign abort reports success but must not destroy the session.
+	rec = doChunked(t, router, http.MethodDelete, "/uploads/"+session.UploadID, nil, other)
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("foreign abort status = %d, want 204", rec.Code)
+	}
+	rec = doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		[]byte("0123456789"), accessClaims())
+	if rec.Code != http.StatusOK {
+		t.Fatalf("owner chunk after foreign abort = %d, want 200; body=%s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestDiagnosticsChunkedUploadIncompleteComplete(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	bundle := bytes.Repeat([]byte("b"), int(diagnostics.UploadChunkBytes)+512)
+	session := initChunkedUpload(t, router, len(bundle), accessClaims())
+
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		bundle[:diagnostics.UploadChunkBytes], accessClaims())
+	if rec.Code != http.StatusOK {
+		t.Fatalf("chunk status = %d", rec.Code)
+	}
+
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	if rec.Code != http.StatusConflict {
+		t.Fatalf("incomplete complete status = %d, want 409; body=%s", rec.Code, rec.Body.String())
+	}
+	if service.ingestCalls != 0 {
+		t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
+	}
+}
+
+func TestDiagnosticsChunkedUploadReinitReplacesPreviousSession(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	first := initChunkedUpload(t, router, 10, accessClaims())
+	second := initChunkedUpload(t, router, 10, accessClaims())
+	if first.UploadID == second.UploadID {
+		t.Fatal("re-init returned the same session id")
+	}
+
+	// The replaced session is gone; the new one accepts chunks.
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+first.UploadID+"/chunks/0",
+		[]byte("0123456789"), accessClaims())
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("replaced-session chunk status = %d, want 404", rec.Code)
+	}
+	rec = doChunked(t, router, http.MethodPut, "/uploads/"+second.UploadID+"/chunks/0",
+		[]byte("0123456789"), accessClaims())
+	if rec.Code != http.StatusOK {
+		t.Fatalf("new-session chunk status = %d, want 200; body=%s", rec.Code, rec.Body.String())
+	}
+
+	rec = doChunked(t, router, http.MethodDelete, "/uploads/"+second.UploadID, nil, accessClaims())
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("abort status = %d, want 204", rec.Code)
+	}
+}
+
+func TestDiagnosticsChunkedCompleteBusyKeepsSessionRetryable(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	session := initChunkedUpload(t, router, 10, accessClaims())
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		[]byte("0123456789"), accessClaims())
+	if rec.Code != http.StatusOK {
+		t.Fatalf("chunk status = %d", rec.Code)
+	}
+
+	// Occupy the user's ingest slot: complete must answer busy without
+	// consuming the spooled session.
+	release, acquired := handler.inflight.acquire(42)
+	if !acquired {
+		t.Fatal("could not occupy in-flight slot")
+	}
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	assertDiagnosticsError(t, rec, http.StatusServiceUnavailable, "busy")
+	release()
+
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("retried complete status = %d; body=%s", rec.Code, rec.Body.String())
+	}
+	if service.ingestCalls != 1 {
+		t.Fatalf("ingest calls = %d, want 1", service.ingestCalls)
+	}
+}
+
+func TestDiagnosticsChunkedUploadOversizedChunkRejected(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	bundle := bytes.Repeat([]byte("b"), int(diagnostics.UploadChunkBytes)*2)
+	session := initChunkedUpload(t, router, len(bundle), accessClaims())
+
+	oversized := bytes.Repeat([]byte("x"), int(diagnostics.UploadChunkBytes)+1)
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		oversized, accessClaims())
+	if rec.Code != http.StatusBadRequest && rec.Code != http.StatusRequestEntityTooLarge {
+		t.Fatalf("oversized chunk status = %d, want 400 or 413; body=%s", rec.Code, rec.Body.String())
+	}
+	if service.ingestCalls != 0 {
+		t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
+	}
+}
+
+func TestDiagnosticsChunkedUploadManifestTooLarge(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	manifest := `{"pad":"` + strings.Repeat("x", int(diagnostics.MaxManifestBytes)) + `"}`
+	rec := doChunked(t, router, http.MethodPost, "/uploads",
+		[]byte(`{"manifest":`+manifest+`,"bundle_bytes":10}`), accessClaims())
+	assertDiagnosticsError(t, rec, http.StatusRequestEntityTooLarge, "too_large")
+}
+
+func TestDiagnosticsChunkedCompleteIngestErrorMapsLikeSingleShot(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	service.ingestErr = diagnostics.ErrStaleConsent
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	session := initChunkedUpload(t, router, 10, accessClaims())
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		[]byte("0123456789"), accessClaims())
+	if rec.Code != http.StatusOK {
+		t.Fatalf("chunk status = %d", rec.Code)
+	}
+
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	assertDiagnosticsError(t, rec, http.StatusBadRequest, "stale_consent")
+
+	// The failed session is consumed; a retry needs a fresh init and the slot
+	// must be free for it.
+	initChunkedUpload(t, router, 10, accessClaims())
+}
+
+func TestDiagnosticsChunkedUploadConcurrentInitsRespectSlotAtomically(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	// Fire concurrent inits for the same user. The reservation must admit at
+	// most one at a time — the losers get busy — so a single account can
+	// never fan out to multiple sessions and eat the global cap.
+	const attempts = 8
+	var wg sync.WaitGroup
+	codes := make([]int, attempts)
+	for i := 0; i < attempts; i++ {
+		wg.Add(1)
+		go func(slot int) {
+			defer wg.Done()
+			rec := doChunked(t, router, http.MethodPost, "/uploads",
+				[]byte(`{"manifest":{"ok":true},"bundle_bytes":10}`), accessClaims())
+			codes[slot] = rec.Code
+		}(i)
+	}
+	wg.Wait()
+
+	created := 0
+	for _, code := range codes {
+		switch code {
+		case http.StatusCreated:
+			created++
+		case http.StatusServiceUnavailable:
+		default:
+			t.Fatalf("unexpected init status %d", code)
+		}
+	}
+	if created < 1 {
+		t.Fatal("no init succeeded")
+	}
+	// However the race resolved, the user must end up with at most one live
+	// session and no leaked cap slots: after aborting it, a fresh init and
+	// full upload must succeed.
+	handler.chunkSessions.mu.Lock()
+	owned := len(handler.chunkSessions.owners)
+	reserved := len(handler.chunkSessions.reserved)
+	handler.chunkSessions.mu.Unlock()
+	if owned > 1 || reserved != 0 {
+		t.Fatalf("owners = %d (want ≤1), reserved = %d (want 0)", owned, reserved)
+	}
+
+	session := initChunkedUpload(t, router, 10, accessClaims())
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		[]byte("0123456789"), accessClaims())
+	if rec.Code != http.StatusOK {
+		t.Fatalf("post-race chunk status = %d; body=%s", rec.Code, rec.Body.String())
+	}
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("post-race complete status = %d; body=%s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestDiagnosticsChunkedCompleteTransientStatusErrorKeepsSession(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	handler := newChunkedTestHandler(t, service)
+	router := chunkedRouter(handler)
+
+	session := initChunkedUpload(t, router, 10, accessClaims())
+	rec := doChunked(t, router, http.MethodPut, "/uploads/"+session.UploadID+"/chunks/0",
+		[]byte("0123456789"), accessClaims())
+	if rec.Code != http.StatusOK {
+		t.Fatalf("chunk status = %d", rec.Code)
+	}
+
+	// A transient status failure must answer 500 WITHOUT destroying the fully
+	// uploaded spool; the retried complete succeeds with no re-upload.
+	service.setStatusErr(errFakeStatusUnavailable)
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	assertDiagnosticsError(t, rec, http.StatusInternalServerError, "internal_error")
+
+	service.setStatusErr(nil)
+	rec = doChunked(t, router, http.MethodPost, "/uploads/"+session.UploadID+"/complete", nil, accessClaims())
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("retried complete status = %d; body=%s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestDiagnosticsStatusAdvertisesUploadChunkBytes(t *testing.T) {
+	service := newFakeDiagnosticsService()
+	service.status.UploadChunkBytes = diagnostics.UploadChunkBytes
+	handler := newChunkedTestHandler(t, service)
+
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/diagnostics/status", nil)
+	req.Header.Set("Authorization", "Bearer token")
+	req = req.WithContext(apimw.SetClaims(req.Context(), accessClaims()))
+	rec := httptest.NewRecorder()
+	handler.HandleStatus(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	var payload map[string]any
+	if err := json.NewDecoder(rec.Body).Decode(&payload); err != nil {
+		t.Fatalf("decode status: %v", err)
+	}
+	got, ok := payload["upload_chunk_bytes"].(float64)
+	if !ok || int64(got) != diagnostics.UploadChunkBytes {
+		t.Fatalf("upload_chunk_bytes = %v, want %d", payload["upload_chunk_bytes"], diagnostics.UploadChunkBytes)
+	}
+}
@@ -704,29 +704,32 @@ func adminClaims() *auth.Claims {
 }

 type fakeDiagnosticsService struct {
-	mu            sync.Mutex
-	status        diagnostics.Status
-	statusCalls   int
-	ingestCalls   int
-	ingestErr     error
-	started       chan struct{}
-	block         chan struct{}
-	listCalls     int
-	listFilters   diagnostics.ListFilters
-	listResult    diagnostics.ListResult
-	listErr       error
-	getReport     *diagnostics.Report
-	getErr        error
-	presignURL    string
-	presignErr    error
-	presignCalls  int
-	presignExpiry time.Duration
-	effectiveTTL  time.Duration
-	openCalls     int
-	openData      []byte
-	openErr       error
-	deleteReport  *diagnostics.Report
-	deleteErr     error
+	mu              sync.Mutex
+	status          diagnostics.Status
+	statusErr       error
+	statusCalls     int
+	ingestCalls     int
+	ingestErr       error
+	lastManifest    []byte
+	lastBundleBytes int64
+	started         chan struct{}
+	block           chan struct{}
+	listCalls       int
+	listFilters     diagnostics.ListFilters
+	listResult      diagnostics.ListResult
+	listErr         error
+	getReport       *diagnostics.Report
+	getErr          error
+	presignURL      string
+	presignErr      error
+	presignCalls    int
+	presignExpiry   time.Duration
+	effectiveTTL    time.Duration
+	openCalls       int
+	openData        []byte
+	openErr         error
+	deleteReport    *diagnostics.Report
+	deleteErr       error
 }

 func newFakeDiagnosticsService() *fakeDiagnosticsService {
@@ -747,12 +750,22 @@ func (f *fakeDiagnosticsService) Status(context.Context, int) (diagnostics.Statu
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.statusCalls++
+	if f.statusErr != nil {
+		return diagnostics.Status{}, f.statusErr
+	}
 	return f.status, nil
 }

-func (f *fakeDiagnosticsService) Ingest(_ context.Context, _ int, _ *string, _ []byte, bundle io.Reader) (diagnostics.IngestResult, error) {
+func (f *fakeDiagnosticsService) setStatusErr(err error) {
+	f.mu.Lock()
+	f.statusErr = err
+	f.mu.Unlock()
+}
+
+func (f *fakeDiagnosticsService) Ingest(_ context.Context, _ int, _ *string, manifestJSON []byte, bundle io.Reader) (diagnostics.IngestResult, error) {
 	f.mu.Lock()
 	f.ingestCalls++
+	f.lastManifest = append([]byte(nil), manifestJSON...)
 	started := f.started
 	block := f.block
 	err := f.ingestErr
@@ -767,9 +780,13 @@ func (f *fakeDiagnosticsService) Ingest(_ context.Context, _ int, _ *string, _ [
 	if err != nil {
 		return diagnostics.IngestResult{}, err
 	}
-	if _, err := io.ReadAll(bundle); err != nil {
+	consumed, err := io.Copy(io.Discard, bundle)
+	if err != nil {
 		return diagnostics.IngestResult{}, err
 	}
+	f.mu.Lock()
+	f.lastBundleBytes = consumed
+	f.mu.Unlock()
 	return diagnostics.IngestResult{
 		ReportID: "11111111-1111-1111-1111-111111111111",
 		ShortID:  "SILO-ABCDEF123456",
@@ -855,6 +855,8 @@ func uploadErrorResponse(err error) (int, string) {
 		return http.StatusConflict, "Upload session is incomplete"
 	case errors.Is(err, uploads.ErrAlreadyCompleted):
 		return http.StatusConflict, "Upload session is already complete"
+	case errors.Is(err, uploads.ErrChunkBusy):
+		return http.StatusConflict, "This chunk is already being uploaded"
 	case errors.Is(err, uploads.ErrInvalidChunk), errors.Is(err, uploads.ErrInvalidRequest):
 		return http.StatusBadRequest, err.Error()
 	default:
@@ -43,8 +43,9 @@ var demoBlockedRoutes = []blockedRoute{
 	{methods: []string{"POST"}, prefix: "/api/v1/subtitles/upload"},
 	{methods: []string{"DELETE"}, prefix: "/api/v1/subtitles/"},
 	// Diagnostics report uploads/deletes write DB rows and private-bucket blobs;
-	// GET /api/v1/diagnostics/status is unaffected (GETs always pass).
-	{methods: []string{"POST", "DELETE"}, prefix: "/api/v1/diagnostics/reports"},
+	// GET /api/v1/diagnostics/status is unaffected (GETs always pass). PUT
+	// covers the chunked-upload chunk route under /reports/uploads.
+	{methods: []string{"POST", "PUT", "DELETE"}, prefix: "/api/v1/diagnostics/reports"},
 }

 // Guard is an HTTP middleware that enforces demo mode restrictions.
@@ -1830,6 +1830,15 @@ func NewRouter(deps Dependencies) chi.Router {
 				r.Route("/diagnostics", func(r chi.Router) {
 					r.Get("/status", diagnosticsHandler.HandleStatus)
 					r.Post("/reports", diagnosticsHandler.HandleUpload)
+					// Chunked fallback for bundles a fronting proxy's
+					// request-body cap rejects as one request. Same
+					// ingest/validation path; see diagnostics_chunked.go.
+					r.Route("/reports/uploads", func(r chi.Router) {
+						r.Post("/", diagnosticsHandler.HandleChunkedUploadInit)
+						r.Put("/{upload_id}/chunks/{chunk_index}", diagnosticsHandler.HandleChunkedUploadChunk)
+						r.Post("/{upload_id}/complete", diagnosticsHandler.HandleChunkedUploadComplete)
+						r.Delete("/{upload_id}", diagnosticsHandler.HandleChunkedUploadAbort)
+					})
 				})
 			})
 		}
@@ -18,8 +18,14 @@ import (
 )

 const (
-	MaxManifestBytes       = int64(64 * 1024)
-	BundleContentType      = "application/gzip"
+	MaxManifestBytes  = int64(64 * 1024)
+	BundleContentType = "application/gzip"
+	// UploadChunkBytes is the fixed chunk payload size for chunked bundle
+	// uploads. Deliberately well under 1 MiB: chunked uploads exist to pass
+	// reverse proxies that keep nginx's default `client_max_body_size 1m`, so
+	// each chunk request — payload plus multipart/header overhead — must stay
+	// clearly below that floor.
+	UploadChunkBytes       = int64(768 * 1024)
 	diagnosticObjectPrefix = "diagnostics"

 	failedUploadCompensationTimeout = 10 * time.Second
@@ -54,6 +60,12 @@ type Status struct {
 	MaxManifestBytes       int64              `json:"max_manifest_bytes"`
 	RetentionDays          int                `json:"retention_days"`
 	ConsentNoticeVersion   int                `json:"consent_notice_version"`
+	// UploadChunkBytes advertises chunked-upload support and the fixed chunk
+	// payload size. Clients that see it may split a bundle across
+	// POST /diagnostics/reports/uploads + chunk PUTs when the single-shot
+	// upload is rejected by a reverse proxy's request-body cap. Absent/zero on
+	// older servers, which clients must treat as "chunking unsupported".
+	UploadChunkBytes int64 `json:"upload_chunk_bytes"`
 }

 type IngestResult struct {
@@ -306,6 +318,7 @@ func (s *Service) statusFromSettings(settings Settings) Status {
 		MaxManifestBytes:       MaxManifestBytes,
 		RetentionDays:          settings.RetentionDays,
 		ConsentNoticeVersion:   settings.ConsentNoticeVersion,
+		UploadChunkBytes:       UploadChunkBytes,
 	}
 }

@@ -23,6 +23,7 @@ const (

 var (
 	ErrAlreadyCompleted = errors.New("upload session is already completed")
+	ErrChunkBusy        = errors.New("upload chunk already in progress")
 	ErrExpired          = errors.New("upload session expired")
 	ErrIncomplete       = errors.New("upload session is incomplete")
 	ErrInvalidChunk     = errors.New("invalid upload chunk")
@@ -73,6 +74,13 @@ type Manager struct {
 	maxChunkSize int64
 	now          func() time.Time
 	sessions     map[string]*session
+	// detached holds sessions removed from `sessions` (canceled or expired)
+	// whose chunk writers are still draining a request body into the spool.
+	// They keep holding real disk and a connection until the writer finishes
+	// or hits the read deadline, so callers enforcing admission caps must
+	// count them (DetachedWriterSessions) or a cancel-and-recreate loop could
+	// stack unbounded live writers behind a fixed session cap.
+	detached map[*session]struct{}
 }

 type session struct {
@@ -88,6 +96,16 @@ type session struct {
 	dir            string
 	expiresAt      time.Time
 	completed      bool
+	// writing marks chunk indexes with a body copy in flight. Body I/O happens
+	// outside the manager mutex (a slow client must not stall every other
+	// session), so these flags are what stops a duplicate concurrent write to
+	// the same offset and keeps Complete/Cancel from consuming a session
+	// while bytes are still landing in its file.
+	writing map[int]struct{}
+}
+
+func (s *session) writingCount() int {
+	return len(s.writing)
 }

 func NewManager(opts ManagerOptions) *Manager {
@@ -118,6 +136,7 @@ func NewManager(opts ManagerOptions) *Manager {
 		maxChunkSize: maxChunkSize,
 		now:          now,
 		sessions:     make(map[string]*session),
+		detached:     make(map[*session]struct{}),
 	}
 }

@@ -199,67 +218,127 @@ func (m *Manager) Create(req CreateRequest) (SessionInfo, error) {
 	return s.info(), nil
 }

+// PutChunk streams one chunk into the session file. The body copy runs
+// OUTSIDE the manager mutex — network reads from one slow client must not
+// serialize every other session's chunk writes, completes, and cancels behind
+// it. A per-chunk in-flight flag prevents duplicate concurrent writes to the
+// same offset. Each begun and each accepted chunk refreshes the session
+// expiry, making the TTL an idle timeout rather than an absolute deadline so
+// a slow-but-progressing upload stays alive.
 func (m *Manager) PutChunk(ctx context.Context, id string, index int, body io.Reader, contentLength int64) (SessionInfo, error) {
 	if body == nil {
 		return SessionInfo{}, fmt.Errorf("%w: chunk body is required", ErrInvalidChunk)
 	}

 	m.mu.Lock()
-	defer m.mu.Unlock()
 	m.cleanupExpiredLocked(m.now())

 	s, err := m.getLocked(id)
 	if err != nil {
+		m.mu.Unlock()
 		return SessionInfo{}, err
 	}
 	if s.completed {
+		m.mu.Unlock()
 		return SessionInfo{}, ErrAlreadyCompleted
 	}
 	if index < 0 || index >= s.totalChunks {
+		m.mu.Unlock()
 		return SessionInfo{}, fmt.Errorf("%w: chunk index is out of range", ErrInvalidChunk)
 	}

 	expectedSize := s.expectedChunkSize(index)
 	if contentLength >= 0 && contentLength != expectedSize {
+		m.mu.Unlock()
 		return SessionInfo{}, fmt.Errorf("%w: chunk size must be %d bytes", ErrInvalidChunk, expectedSize)
 	}
 	if expectedSize > m.maxChunkSize {
+		m.mu.Unlock()
 		return SessionInfo{}, fmt.Errorf("%w: maximum chunk size is %d bytes", ErrTooLarge, m.maxChunkSize)
 	}
 	if s.received[index] {
-		return s.info(), nil
+		info := s.info()
+		m.mu.Unlock()
+		return info, nil
 	}
+	if _, inFlight := s.writing[index]; inFlight {
+		m.mu.Unlock()
+		return SessionInfo{}, ErrChunkBusy
+	}
+	if s.writing == nil {
+		s.writing = make(map[int]struct{})
+	}
+	s.writing[index] = struct{}{}
+	s.expiresAt = m.now().Add(m.ttl)
+	path := s.path
+	offset := int64(index) * s.chunkSize
+	m.mu.Unlock()

-	file, err := os.OpenFile(s.path, os.O_WRONLY, 0o600)
+	written, copyErr := copyChunkBody(ctx, path, offset, body, expectedSize)
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(s.writing, index)
+	// The session may have been canceled or expired while the copy ran. Its
+	// directory removal was deferred to the last finishing writer (see
+	// removeDetachedDirLocked); report the session gone either way.
+	if s.completed {
+		return SessionInfo{}, ErrAlreadyCompleted
+	}
+	if m.sessions[id] != s {
+		m.removeDetachedDirLocked(s)
+		return SessionInfo{}, ErrNotFound
+	}
+	if copyErr != nil {
+		return SessionInfo{}, copyErr
+	}
+	s.received[index] = true
+	s.receivedChunks++
+	s.receivedBytes += written
+	s.expiresAt = m.now().Add(m.ttl)
+	return s.info(), nil
+}
+
+// removeDetachedDirLocked reclaims a detached session once its last in-flight
+// writer has returned: drops it from the detached count and removes the spool
+// directory. Safe to call for every finishing writer of a dropped session.
+func (m *Manager) removeDetachedDirLocked(s *session) {
+	if s.writingCount() != 0 {
+		return
+	}
+	delete(m.detached, s)
+	_ = os.RemoveAll(s.dir)
+}
+
+// copyChunkBody writes exactly expectedSize bytes from body at offset,
+// rejecting short or oversized chunks. Runs without the manager lock held.
+func copyChunkBody(ctx context.Context, path string, offset int64, body io.Reader, expectedSize int64) (int64, error) {
+	file, err := os.OpenFile(path, os.O_WRONLY, 0o600)
 	if err != nil {
-		return SessionInfo{}, fmt.Errorf("open upload session file: %w", err)
+		return 0, fmt.Errorf("open upload session file: %w", err)
 	}
 	defer file.Close()

-	writer := io.NewOffsetWriter(file, int64(index)*s.chunkSize)
+	writer := io.NewOffsetWriter(file, offset)
 	written, err := io.Copy(writer, io.LimitReader(body, expectedSize))
 	if err != nil {
-		return SessionInfo{}, err
+		return 0, err
 	}
 	if err := ctx.Err(); err != nil {
-		return SessionInfo{}, err
+		return 0, err
 	}
 	if written != expectedSize {
-		return SessionInfo{}, fmt.Errorf("%w: chunk size must be %d bytes", ErrInvalidChunk, expectedSize)
+		return 0, fmt.Errorf("%w: chunk size must be %d bytes", ErrInvalidChunk, expectedSize)
 	}
 	var extra [1]byte
 	extraBytes, err := body.Read(extra[:])
 	if err != nil && !errors.Is(err, io.EOF) {
-		return SessionInfo{}, err
+		return 0, err
 	}
 	if extraBytes > 0 {
-		return SessionInfo{}, fmt.Errorf("%w: chunk size must be %d bytes", ErrInvalidChunk, expectedSize)
+		return 0, fmt.Errorf("%w: chunk size must be %d bytes", ErrInvalidChunk, expectedSize)
 	}
-
-	s.received[index] = true
-	s.receivedChunks++
-	s.receivedBytes += written
-	return s.info(), nil
+	return written, nil
 }

 func (m *Manager) Complete(id string) (*CompletedUpload, error) {
@@ -291,6 +370,20 @@ func (m *Manager) Complete(id string) (*CompletedUpload, error) {
 	}, nil
 }

+// Peek reports the session's current state without mutating it beyond the
+// shared expiry cleanup. Callers holding per-session state outside the
+// manager (e.g. diagnostics chunked uploads) use it to notice sessions the
+// manager has expired so their own bookkeeping can be released too.
+func (m *Manager) Peek(id string) (SessionInfo, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	s, err := m.getLocked(id)
+	if err != nil {
+		return SessionInfo{}, err
+	}
+	return s.info(), nil
+}
+
 func (m *Manager) Cancel(id string) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -299,7 +392,24 @@ func (m *Manager) Cancel(id string) error {
 		return ErrNotFound
 	}
 	delete(m.sessions, id)
-	return os.RemoveAll(s.dir)
+	// With a chunk write in flight the file handle is still open; park the
+	// session as detached — still counted by DetachedWriterSessions — and the
+	// last finishing writer removes the directory.
+	if s.writingCount() == 0 {
+		return os.RemoveAll(s.dir)
+	}
+	m.detached[s] = struct{}{}
+	return nil
+}
+
+// DetachedWriterSessions counts canceled/expired sessions whose chunk writers
+// are still draining request bodies into spool files. Admission caps must
+// include them: they hold connections and disk until each writer finishes or
+// times out.
+func (m *Manager) DetachedWriterSessions() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return len(m.detached)
 }

 func (m *Manager) getLocked(id string) (*session, error) {
@@ -313,7 +423,11 @@ func (m *Manager) getLocked(id string) (*session, error) {
 	}
 	if !m.now().Before(s.expiresAt) {
 		delete(m.sessions, id)
-		_ = os.RemoveAll(s.dir)
+		if s.writingCount() == 0 {
+			_ = os.RemoveAll(s.dir)
+		} else {
+			m.detached[s] = struct{}{}
+		}
 		return nil, ErrExpired
 	}
 	return s, nil
@@ -325,10 +439,71 @@ func (m *Manager) cleanupExpiredLocked(now time.Time) {
 			continue
 		}
 		delete(m.sessions, id)
-		_ = os.RemoveAll(s.dir)
+		if s.writingCount() == 0 {
+			_ = os.RemoveAll(s.dir)
+		} else {
+			m.detached[s] = struct{}{}
+		}
 	}
 }

+// ReclaimOrphanedDirs deletes session directories under the manager's root
+// that no live session owns. Call once at startup: after a process restart
+// the in-memory session map is empty, but the previous process's partially
+// uploaded spool directories survive on disk and would otherwise accumulate
+// forever (no sweep ever discovers them). Returns the number of directories
+// removed.
+func (m *Manager) ReclaimOrphanedDirs() int {
+	m.mu.Lock()
+	live := make(map[string]struct{}, len(m.sessions))
+	for id := range m.sessions {
+		live[id] = struct{}{}
+	}
+	rootDir := m.rootDir
+	m.mu.Unlock()
+
+	entries, err := os.ReadDir(rootDir)
+	if err != nil {
+		return 0
+	}
+	removed := 0
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			continue
+		}
+		if _, ok := live[entry.Name()]; ok {
+			continue
+		}
+		if os.RemoveAll(filepath.Join(rootDir, entry.Name())) == nil {
+			removed++
+		}
+	}
+	return removed
+}
+
+// StartExpirySweeper runs cleanup on the given interval until stop is closed.
+// Without it, expiry only triggers from later API calls — sessions abandoned
+// during a quiet period would otherwise hold their spool bytes indefinitely.
+func (m *Manager) StartExpirySweeper(interval time.Duration, stop <-chan struct{}) {
+	if interval <= 0 {
+		interval = time.Minute
+	}
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-ticker.C:
+				m.mu.Lock()
+				m.cleanupExpiredLocked(m.now())
+				m.mu.Unlock()
+			}
+		}
+	}()
+}
+
 func (s *session) isComplete() bool {
 	return s.receivedChunks == s.totalChunks && s.receivedBytes == s.sizeBytes
 }
@@ -4,7 +4,9 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"io"
 	"os"
+	"path/filepath"
 	"testing"
 	"time"
 )
@@ -167,6 +169,192 @@ func TestManagerAcceptsDuplicateReceivedChunk(t *testing.T) {
 	}
 }

+func TestManagerChunkActivityRefreshesExpiry(t *testing.T) {
+	now := time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)
+	manager := NewManager(ManagerOptions{
+		RootDir:      t.TempDir(),
+		MaxSize:      16,
+		MaxChunkSize: 4,
+		TTL:          10 * time.Minute,
+		Now:          func() time.Time { return now },
+	})
+
+	session, err := manager.Create(CreateRequest{Filename: "plugin.bin", SizeBytes: 8, ChunkSize: 4})
+	if err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+
+	// 9 minutes per chunk: an absolute deadline would expire before the
+	// second chunk; an idle timeout refreshed by chunk arrivals must not.
+	now = now.Add(9 * time.Minute)
+	if _, err := manager.PutChunk(context.Background(), session.ID, 0, bytes.NewReader([]byte("abcd")), 4); err != nil {
+		t.Fatalf("PutChunk(0) error = %v", err)
+	}
+	now = now.Add(9 * time.Minute)
+	if _, err := manager.PutChunk(context.Background(), session.ID, 1, bytes.NewReader([]byte("efgh")), 4); err != nil {
+		t.Fatalf("PutChunk(1) after refresh error = %v", err)
+	}
+	if _, err := manager.Complete(session.ID); err != nil {
+		t.Fatalf("Complete() error = %v", err)
+	}
+}
+
+func TestManagerReclaimsOrphanedDirsFromPreviousProcess(t *testing.T) {
+	root := t.TempDir()
+	stale := filepath.Join(root, "deadbeefdeadbeefdeadbeefdeadbeef")
+	if err := os.MkdirAll(stale, 0o700); err != nil {
+		t.Fatalf("mkdir stale dir: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(stale, "upload.bin"), []byte("leftover"), 0o600); err != nil {
+		t.Fatalf("write stale spool: %v", err)
+	}
+
+	// A "restarted" manager knows nothing about the stale dir; reclaim must
+	// remove it while leaving live sessions untouched.
+	manager := NewManager(ManagerOptions{RootDir: root, MaxSize: 16, MaxChunkSize: 4, Now: fixedNow()})
+	session, err := manager.Create(CreateRequest{Filename: "plugin.bin", SizeBytes: 4, ChunkSize: 4})
+	if err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+
+	if removed := manager.ReclaimOrphanedDirs(); removed != 1 {
+		t.Fatalf("ReclaimOrphanedDirs() = %d, want 1", removed)
+	}
+	if _, err := os.Stat(stale); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("stale dir still exists: %v", err)
+	}
+	if _, err := manager.PutChunk(context.Background(), session.ID, 0, bytes.NewReader([]byte("abcd")), 4); err != nil {
+		t.Fatalf("live session broken after reclaim: %v", err)
+	}
+}
+
+func TestManagerRejectsConcurrentWritesToSameChunk(t *testing.T) {
+	manager := NewManager(ManagerOptions{
+		RootDir:      t.TempDir(),
+		MaxSize:      16,
+		MaxChunkSize: 4,
+		Now:          fixedNow(),
+	})
+	session, err := manager.Create(CreateRequest{Filename: "plugin.bin", SizeBytes: 4, ChunkSize: 4})
+	if err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+
+	// First writer blocks mid-body (outside the manager lock); a second
+	// writer for the same chunk must be turned away, and writers for other
+	// sessions must not be blocked behind the stalled copy.
+	firstStarted := make(chan struct{})
+	release := make(chan struct{})
+	done := make(chan error, 1)
+	go func() {
+		_, err := manager.PutChunk(context.Background(), session.ID, 0, &gatedReader{
+			started: firstStarted,
+			release: release,
+			data:    []byte("abcd"),
+		}, 4)
+		done <- err
+	}()
+	<-firstStarted
+
+	if _, err := manager.PutChunk(context.Background(), session.ID, 0, bytes.NewReader([]byte("zzzz")), 4); !errors.Is(err, ErrChunkBusy) {
+		t.Fatalf("concurrent same-chunk PutChunk error = %v, want ErrChunkBusy", err)
+	}
+
+	other, err := manager.Create(CreateRequest{Filename: "other.bin", SizeBytes: 4, ChunkSize: 4})
+	if err != nil {
+		t.Fatalf("Create(other) error = %v", err)
+	}
+	if _, err := manager.PutChunk(context.Background(), other.ID, 0, bytes.NewReader([]byte("wxyz")), 4); err != nil {
+		t.Fatalf("other session blocked behind stalled writer: %v", err)
+	}
+
+	close(release)
+	if err := <-done; err != nil {
+		t.Fatalf("stalled writer error = %v", err)
+	}
+	upload, err := manager.Complete(session.ID)
+	if err != nil {
+		t.Fatalf("Complete() error = %v", err)
+	}
+	defer upload.Cleanup()
+	data, err := os.ReadFile(upload.Path)
+	if err != nil {
+		t.Fatalf("ReadFile() error = %v", err)
+	}
+	if string(data) != "abcd" {
+		t.Fatalf("assembled upload = %q, want %q", data, "abcd")
+	}
+}
+
+// gatedReader signals when Read is first called and then blocks until
+// released, letting tests hold a chunk body copy open mid-flight.
+type gatedReader struct {
+	started chan struct{}
+	release chan struct{}
+	data    []byte
+	signal  bool
+	offset  int
+}
+
+func (g *gatedReader) Read(p []byte) (int, error) {
+	if !g.signal {
+		g.signal = true
+		close(g.started)
+		<-g.release
+	}
+	if g.offset >= len(g.data) {
+		return 0, io.EOF
+	}
+	n := copy(p, g.data[g.offset:])
+	g.offset += n
+	return n, nil
+}
+
+func TestManagerCountsDetachedWritersUntilTheyFinish(t *testing.T) {
+	manager := NewManager(ManagerOptions{
+		RootDir:      t.TempDir(),
+		MaxSize:      16,
+		MaxChunkSize: 4,
+		Now:          fixedNow(),
+	})
+	session, err := manager.Create(CreateRequest{Filename: "plugin.bin", SizeBytes: 4, ChunkSize: 4})
+	if err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+
+	// Stall a chunk write, then cancel the session out from under it — the
+	// cancel-and-recreate pattern a client uses to replace an upload. The
+	// stalled writer still holds a connection and the spool file, so it must
+	// stay visible to admission caps until it finishes.
+	firstStarted := make(chan struct{})
+	release := make(chan struct{})
+	done := make(chan error, 1)
+	go func() {
+		_, err := manager.PutChunk(context.Background(), session.ID, 0, &gatedReader{
+			started: firstStarted,
+			release: release,
+			data:    []byte("abcd"),
+		}, 4)
+		done <- err
+	}()
+	<-firstStarted
+
+	if err := manager.Cancel(session.ID); err != nil {
+		t.Fatalf("Cancel() error = %v", err)
+	}
+	if got := manager.DetachedWriterSessions(); got != 1 {
+		t.Fatalf("DetachedWriterSessions() during write = %d, want 1", got)
+	}
+
+	close(release)
+	if err := <-done; !errors.Is(err, ErrNotFound) {
+		t.Fatalf("stalled writer error = %v, want ErrNotFound", err)
+	}
+	if got := manager.DetachedWriterSessions(); got != 0 {
+		t.Fatalf("DetachedWriterSessions() after writer drained = %d, want 0", got)
+	}
+}
+
 func fixedNow() func() time.Time {
 	now := time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)
 	return func() time.Time {