feat(attachments): orphan GC sweep with periodic scheduler (TASK-886) (#307)

* feat(attachments): orphan GC sweep with periodic scheduler (TASK-886)

Background job that reclaims attachments past the grace period. Two
qualification criteria, both with a 30-day default grace:

  - item_id IS NULL AND deleted_at IS NULL AND created_at < cutoff
    (never-attached uploads — editor uploaded then tab-closed before
    attaching to an item)
  - deleted_at IS NOT NULL AND deleted_at < cutoff
    (soft-deleted via the Settings → Storage delete button or the
    DELETE /attachments/{id} endpoint)

Reclamation is dedupe-aware: content-addressed storage means the same
hash can be referenced by multiple rows, so the on-disk blob is only
removed when the GC'd row is the LAST live reference to its
content_hash. Otherwise the row drops and the blob stays for the
remaining references. CountLiveAttachmentsForHash is the predicate.

Per-row failures (resolve backend, blob delete, hard-delete) are
logged and skipped; the sweep keeps making progress. Catastrophic
errors (DB failure) return up to the loop, which logs and waits for
the next tick rather than crashing the server.

Lifecycle:
- SetOrphanGCConfig overrides the default 24h interval / 30-day
  grace. cmd/pad reads PAD_ORPHAN_GC_INTERVAL / PAD_ORPHAN_GC_GRACE
  (Go duration syntax — 1m, 24h, 720h) so operators can tune
  without recompiling and tests can crank the interval down to 1ms
  to see sweeps land in CI.
- StartOrphanGC kicks the loop. Idempotent — second call is a
  no-op so a misconfigured caller can't double-spawn.
- Server.Stop() now signals the loop via stopOrphanGC() before
  s.bg.Wait(), so process shutdown drains the goroutine cleanly
  (BUG-842 invariant).
- Each tick wraps the sweep in a 30m context timeout so a slow
  scan can't pin the goroutine across multiple intervals.

Tests:
- TestOrphanGC_ReclaimsSoftDeleted: upload → soft-delete → sweep
  with future cutoff → DB row gone + blob gone from FSStore.
- TestOrphanGC_ReclaimsLongOrphans: upload → backdate created_at
  31d → sweep with 30d grace cutoff → row reclaimed.
- TestOrphanGC_KeepsRecentRows: upload → soft-delete → sweep with
  past cutoff → row stays. Catches a typo in the WHERE clause that
  would silently destroy live attachments.
- TestOrphanGC_PreservesSharedBlob: two uploads with identical
  bytes (same hash, same blob), soft-delete only one → sweep →
  one row reclaimed BUT BlobsReclaimed=0 because the other row
  still references the blob. Pin for content-addressed dedupe.
- TestOrphanGC_StartStop: loop spins up at 1ms interval, second
  StartOrphanGC is a no-op, Stop drains via testServer's cleanup.

Parent: PLAN-866. Closes the phase 1 plan with full export →
import → orphan-cleanup round-trip.

* fix(attachments): protect referenced/in-flight blobs from orphan GC per Codex (round 1)

Two real correctness issues Codex caught on PR #307:

P1. The editor's normal upload flow leaves attachments.item_id NULL.
The canonical association lives in markdown content (the editor
PATCHes "pad-attachment:UUID" into the item) — but the GC's
"never-attached past 30d" predicate only checked item_id. So a
legitimate inline image could be hard-deleted 30 days after upload
even though item content still references it.

Added store.AttachmentReferencedInItems(workspaceID, attachmentID)
that scans items.content + items.fields for "pad-attachment:UUID".
The GC sweep now runs this check before reclaiming any
never-attached row; if any live item references the attachment,
the row is left alone (and re-checked next sweep).

P2. Race between concurrent upload and GC. Upload calls
AttachmentStore.Put (blob lands on disk) → THEN inserts the DB row.
Between those two steps an orphan-GC sweep could count zero live
refs for the hash, delete the blob, and the upload's row insert
would then point at a missing blob.

Added Server.inFlightUploadHashes (sync.Map of *atomic.Int64
counters) with markUploadInFlight / uploadInFlight helpers. Every
Put + CreateAttachment site fences itself via markUploadInFlight:
the upload handler, the transform handler, the thumbnail
derivation pipeline, and the bundle-import rehydrate path. The GC
sweep treats an in-flight hash as "another live ref" so it leaves
the blob alone.

Tests:
- TestOrphanGC_KeepsReferencedNeverAttachedRows: upload (item_id
  NULL) → create item with pad-attachment: ref → backdate 31d →
  sweep with 30d cutoff → row stays.
- TestOrphanGC_RespectsInFlightUploads: upload → soft-delete →
  register an in-flight upload at the same hash → sweep → DB row
  goes (it's tombstoned past grace) but blob stays so the
  in-flight upload can complete cleanly.

The DB row still gets reclaimed in the in-flight case because the
soft-deleted row is independently past grace; only the blob delete
is fenced. That's correct: the blob remains usable for the
incoming upload and the new upload will register its own
attachments row.

* fix(attachments): mutex-protect in-flight tracker + portable JSONB scan per Codex (round 2)

Two fixes for the round-2 findings on PR #307:

P1. Same-hash race in the in-flight upload tracker. The sync.Map +
*atomic.Int64 design split increment from LoadOrStore-then-add and
release-decrement from delete, so a release could see "0" and start
deleting while another upload concurrently reloaded the same map
entry and incremented to "1" — the second upload's signal then
lived in a doomed map slot, invisible to subsequent uploadInFlight
calls.

Replaced with a plain map[string]int64 + sync.Mutex. Inc, dec,
delete-when-zero all run under one critical section, so any
inspection sees a consistent snapshot. Net cost is one mutex per
mark/release; uncontended this is ~10ns and the upload path is
already doing far more expensive work (Put + DB insert).

Stress test: 20 goroutines × 500 iterations of mark→check→release
on a shared hash. Every check must observe in-flight=true while
the calling goroutine holds the mark. Final state must be empty.
Runs cleanly under -race -count=3.

P2. Postgres JSONB compatibility. items.fields is TEXT on SQLite
but JSONB on PostgreSQL (per pgmigrations/001_initial.sql). LIKE
on JSONB fails with a type error, so the orphan GC's reference
scan would error on Postgres and skip every never-attached row —
breaking orphan reclamation for those rows entirely.

Cast fields::text in the Postgres dialect path:

  fieldsExpr := "fields"
  if s.dialect.Driver() == DriverPostgres {
      fieldsExpr = "fields::text"
  }

Same approach used elsewhere in the store for dialect-sensitive
text searches.

* fix(attachments): close GC/upload TOCTOU + protect in-grace peers per Codex (round 3)

P1 round 3: TOCTOU race between uploadInFlight check and store.Delete.
The mutex protected the in-flight counter but not the GC's
check-and-delete sequence. A new upload could call markUploadInFlight
between our check and our blob delete, then run Put after the blob
was gone — its CreateAttachment would insert a live row pointing at
the missing hash.

Fixed by holding inFlightHashesMu across the check + FS Delete:

  s.inFlightHashesMu.Lock()
  inFlight := s.inFlightHashes[hash] > 0
  if !inFlight && others == 0 {
      store.Delete(ctx, key)
  }
  s.inFlightHashesMu.Unlock()

A concurrent markUploadInFlight blocks until either we skip (because
we observed in-flight) or finish deleting. Lock window is ms-class
on FSStore; a per-hash lock can replace this server-wide mutex when
S3 lands in Phase 2.

P2 round 3: CountLiveAttachmentsForHash counted only live rows, so
GC could reclaim the blob from row A (soft-deleted 31d ago) even
when row B was also soft-deleted but only 1 day old — within
grace, so its blob must stay reachable until its own grace lapses.

Replaced with CountProtectingAttachmentsForHash which counts rows
where deleted_at IS NULL OR deleted_at >= graceCutoff. The blob is
preserved until every soft-deleted peer has aged past its own
grace window.

Tests:
- TestOrphanGC_RespectsSoftDeletedInGracePeer: two rows sharing a
  hash, soft-delete both, backdate only one past 30d → sweep with
  30d cutoff → older row reclaimed but blob stays for the still-in-
  grace peer.
- existing TestOrphanGC_RespectsInFlightUploads still passes
  (still uses the in-flight signal correctly).

* fix(attachments): dedupe blob-reclaim metric across same-hash peers per Codex (round 4)

Codex round 4 noted that when multiple soft-deleted peers share a
content_hash and all are past grace, the GC sweep would inflate
BlobsReclaimed and BytesReclaimed: AttachmentStore.Delete treats a
missing key as success, so the second peer's idempotent no-op
delete still bumped the counter.

Functional cleanup was correct (the blob really was gone after the
first peer); only the metric / log line was wrong, which makes
operator dashboards report fictitious bytes-reclaimed values.

Track per-sweep reclaimed hashes in a map and skip the Delete call
+ counter increment for repeats. The DB row still gets hard-deleted
on each peer.

Test: TestOrphanGC_DedupesBlobReclaimMetric uploads twice with
identical bytes (single shared blob), soft-deletes both, backdates
deleted_at past grace → sweep deletes 2 rows and reports
BlobsReclaimed=1 / BytesReclaimed=blobLen rather than 2 / 2*blobLen.

cmd/pad/main.go

@@ -390,6 +390,19 @@ func serveCmd() *cobra.Command {
 				}
 			}
 
+			// Orphan GC (TASK-886). Periodic sweep that reclaims
+			// attachments tombstoned past the grace period, plus
+			// uploads that were never associated with an item.
+			// Defaults: 24h interval, 30-day grace. Both override-
+			// able via env (e.g. PAD_ORPHAN_GC_INTERVAL=1m for tests
+			// where you want to see the sweep land within a CI run).
+			gcInterval := parseDurationEnv("PAD_ORPHAN_GC_INTERVAL", 0)
+			gcGrace := parseDurationEnv("PAD_ORPHAN_GC_GRACE", 0)
+			if gcInterval != 0 || gcGrace != 0 {
+				srv.SetOrphanGCConfig(gcInterval, gcGrace)
+			}
+			srv.StartOrphanGC()
+
 			// Wire the image processor used for thumbnail derivation
 			// (TASK-878) and the editor's rotate/crop tools (TASK-879/880).
 			// The default build picks the pure-Go backend (no cgo);
@@ -1029,6 +1042,23 @@ workspaces with no owner).`,
 // humanBytes formats a byte count with the smallest IEC unit that
 // keeps the value under 1024 — matches the convention used across
 // the web UI's storage bar so CLI and browser reads agree.
+// parseDurationEnv reads a duration env var (Go syntax: 1h, 30m,
+// 24h, 720h, etc). Returns the default when the var is unset; logs
+// a warning and returns the default on a parse error so a typo
+// doesn't silently break the GC schedule.
+func parseDurationEnv(name string, def time.Duration) time.Duration {
+	v := os.Getenv(name)
+	if v == "" {
+		return def
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil {
+		slog.Warn(name+" ignored — not a valid Go duration", "value", v, "error", err)
+		return def
+	}
+	return d
+}
+
 func humanBytes(n int64) string {
 	if n < 0 {
 		return "?"

internal/server/handlers_attachments.go

@@ -217,6 +217,15 @@ func (s *Server) handleUploadAttachment(w http.ResponseWriter, r *http.Request)
 		writeInternalError(w, fmt.Errorf("resolve attachment store: %w", err))
 		return
 	}
+	// Fence Put + CreateAttachment against the orphan GC. The
+	// in-flight tracker keeps the GC from reclaiming a blob with
+	// this hash between Put (blob lands on disk) and
+	// CreateAttachment (DB row inserted) — without it, a GC sweep
+	// of an old soft-deleted row sharing the same hash could delete
+	// the blob and strand the new live row. Released after the
+	// CreateAttachment call below (whether it succeeds or fails).
+	releaseInFlight := s.markUploadInFlight(hash)
+	defer releaseInFlight()
 	storageKey, err := store.Put(r.Context(), hash, entry.MIME, tmp)
 	if err != nil {
 		writeInternalError(w, fmt.Errorf("attachment store.Put: %w", err))

internal/server/handlers_attachments_thumbnails.go

@@ -161,6 +161,12 @@ func (s *Server) persistThumbnail(
 	if err != nil {
 		return fmt.Errorf("resolve storage backend: %w", err)
 	}
+	// Fence the Put + CreateAttachment pair against orphan-GC blob
+	// deletion. See handlers_attachments.go for the upload-handler
+	// rationale; thumbnails hit the same race when an old soft-
+	// deleted thumbnail shares the same hash as a freshly-derived one.
+	releaseInFlight := s.markUploadInFlight(hash)
+	defer releaseInFlight()
 	storageKey, err := store.Put(ctx, hash, attachments.ThumbnailMime(format), bytes.NewReader(buf.Bytes()))
 	if err != nil {
 		return fmt.Errorf("put thumbnail blob: %w", err)

internal/server/handlers_attachments_transform.go

@@ -175,6 +175,11 @@ func (s *Server) handleTransformAttachment(w http.ResponseWriter, r *http.Reques
 		writeInternalError(w, fmt.Errorf("resolve destination backend: %w", err))
 		return
 	}
+	// Same orphan-GC fence as the upload handler — see the comment
+	// at handlers_attachments.go's store.Put callsite. Released
+	// after CreateAttachment below regardless of outcome.
+	releaseInFlight := s.markUploadInFlight(hash)
+	defer releaseInFlight()
 	storageKey, err := dstStore.Put(r.Context(), hash, attachments.ThumbnailMime(outFormat), bytes.NewReader(buf.Bytes()))
 	if err != nil {
 		writeInternalError(w, fmt.Errorf("put transformed blob: %w", err))

internal/server/handlers_import_bundle.go

@@ -373,6 +373,12 @@ func (s *Server) rehydrateAttachment(
 	if err != nil {
 		return "", fmt.Errorf("resolve attachment store: %w", err)
 	}
+	// Fence the Put + CreateAttachment pair against orphan-GC blob
+	// deletion (Codex P2 on PR #307). The bundle import races GC the
+	// same way uploads do — possibly more so, since a workspace
+	// re-import touches thousands of hashes in quick succession.
+	releaseInFlight := s.markUploadInFlight(hash)
+	defer releaseInFlight()
 	storageKey, err := store.Put(ctx, hash, allowed.MIME, strings.NewReader(string(blob)))
 	if err != nil {
 		return "", fmt.Errorf("store.Put: %w", err)

internal/server/orphan_gc.go

@@ -0,0 +1,283 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/PerpetualSoftware/pad/internal/attachments"
+)
+
+// Default GC parameters. Operators override via env vars wired in
+// cmd/pad/main.go (PAD_ORPHAN_GC_INTERVAL / PAD_ORPHAN_GC_GRACE).
+const (
+	defaultOrphanGCInterval = 24 * time.Hour
+	defaultOrphanGCGrace    = 30 * 24 * time.Hour
+)
+
+// orphanGCResult records what one sweep accomplished. Returned from
+// runOrphanGCSweep so tests can assert on the counters and the
+// periodic logger can summarize a run in one line.
+type orphanGCResult struct {
+	Scanned        int   // rows considered (matched the orphan SELECT)
+	Deleted        int   // rows hard-deleted from the DB
+	BlobsReclaimed int   // on-disk blobs Delete'd through the storage backend
+	BytesReclaimed int64 // sum of size_bytes for reclaimed blobs
+	Skipped        int   // rows skipped due to mid-sweep errors
+}
+
+// runOrphanGCSweep walks the orphaned-attachments query and reclaims
+// rows past the grace period. Two reclamation paths:
+//
+//   - DB row only. content_hash is still referenced by another live
+//     row (dedup hit). Drop the row, leave the blob on disk.
+//   - DB row + blob. No other live row references the hash. Delete
+//     the blob through the storage backend, then drop the row.
+//
+// Failures within a single row are logged and skipped — the sweep
+// keeps making progress. A genuine catastrophic error (e.g. DB
+// connection lost) returns up so the caller can decide whether to
+// retry the whole sweep.
+//
+// Splitting this out from the periodic loop lets tests drive a
+// single sweep deterministically. Pass a graceCutoff so tests can
+// inject a known time without waiting for real elapsed grace.
+func (s *Server) runOrphanGCSweep(ctx context.Context, graceCutoff time.Time) (*orphanGCResult, error) {
+	if s.attachments == nil {
+		return nil, errors.New("attachments registry not configured")
+	}
+	res := &orphanGCResult{}
+
+	orphans, err := s.store.OrphanedAttachments(graceCutoff)
+	if err != nil {
+		return nil, err
+	}
+	res.Scanned = len(orphans)
+
+	// Track hashes whose blob has already been deleted earlier in
+	// this same sweep so we don't double-count. Without this, two
+	// soft-deleted peers sharing a content_hash would both report
+	// BlobsReclaimed=1 — AttachmentStore.Delete treats a missing
+	// key as success, so the second row's Delete returns nil and
+	// the counter increments again. Functional cleanup is correct
+	// (idempotent); only the metric was wrong. Codex round 4.
+	reclaimedThisSweep := make(map[string]bool)
+
+	for _, a := range orphans {
+		if err := ctx.Err(); err != nil {
+			return res, err
+		}
+		// "Never-attached" rows (item_id IS NULL, deleted_at IS NULL)
+		// can still be referenced from item content via
+		// `pad-attachment:UUID` — the editor uploads first, then
+		// PATCHes content with the reference, but the attachments
+		// row's item_id stays NULL. Scan items.content + items.fields
+		// before reclaiming so the GC doesn't destroy a legitimate
+		// reference. Codex P1 on PR #307 round 1.
+		if a.ItemID == nil && a.DeletedAt == nil {
+			referenced, err := s.store.AttachmentReferencedInItems(a.WorkspaceID, a.ID)
+			if err != nil {
+				slog.Warn("orphan GC: ref-scan failed",
+					"attachment_id", a.ID, "workspace_id", a.WorkspaceID, "error", err)
+				res.Skipped++
+				continue
+			}
+			if referenced {
+				// Item content references the attachment — leave it
+				// alone. Bonus side effect: the row will be picked
+				// up next sweep if the reference goes away.
+				continue
+			}
+		}
+
+		// Decide whether the on-disk blob can also be reclaimed.
+		// Two protections to consider:
+		//
+		//   1. content-addressed dedupe: another row at the same
+		//      hash may still need the blob. CountProtecting includes
+		//      both LIVE rows and soft-deleted rows still inside
+		//      their own grace window — the latter case keeps the
+		//      blob around for un-delete / inspection until each
+		//      row's own grace lapses.
+		//
+		//   2. in-flight uploads: an upload that called
+		//      AttachmentStore.Put but hasn't yet inserted its DB
+		//      row. markUploadInFlight registers the hash before
+		//      Put; we MUST observe that under the same mutex we
+		//      use to gate blob deletion, otherwise a TOCTOU race
+		//      between our check and store.Delete lets a new
+		//      upload's Put land on a blob we're about to remove.
+		//      Codex P1 round 3.
+		others, err := s.store.CountProtectingAttachmentsForHash(a.ContentHash, a.ID, graceCutoff)
+		if err != nil {
+			slog.Warn("orphan GC: count protecting refs failed",
+				"attachment_id", a.ID, "hash", a.ContentHash, "error", err)
+			res.Skipped++
+			continue
+		}
+
+		// Critical section: hold the in-flight mutex across the
+		// uploadInFlight check AND the FS Delete so a concurrent
+		// markUploadInFlight blocks until we either skip (because
+		// it's in flight) or finish deleting. The lock window is
+		// ms-class on FSStore; for S3 backends in Phase 2 a
+		// per-hash lock will replace this server-wide mutex.
+		blobDeleted := false
+		alreadyReclaimed := reclaimedThisSweep[a.ContentHash]
+		s.inFlightHashesMu.Lock()
+		inFlight := s.inFlightHashes[a.ContentHash] > 0
+		if others == 0 && !inFlight && !alreadyReclaimed {
+			store, resolveErr := s.attachments.Resolve(a.StorageKey)
+			if resolveErr != nil {
+				slog.Warn("orphan GC: resolve backend failed",
+					"attachment_id", a.ID, "storage_key", a.StorageKey, "error", resolveErr)
+				s.inFlightHashesMu.Unlock()
+				res.Skipped++
+				continue
+			}
+			if delErr := store.Delete(ctx, a.StorageKey); delErr != nil {
+				// AttachmentStore.Delete documents that deleting a
+				// missing key is NOT an error, so anything reaching
+				// here is a real failure (permission, IO, etc.).
+				// Still drop the DB row — keeping it strands the
+				// row indefinitely; the operator will have to clean
+				// the disk by hand either way.
+				slog.Warn("orphan GC: blob delete failed",
+					"attachment_id", a.ID, "storage_key", a.StorageKey, "error", delErr)
+			} else {
+				blobDeleted = true
+			}
+		}
+		s.inFlightHashesMu.Unlock()
+
+		if blobDeleted {
+			res.BlobsReclaimed++
+			res.BytesReclaimed += a.SizeBytes
+			reclaimedThisSweep[a.ContentHash] = true
+		}
+
+		if err := s.store.HardDeleteAttachment(a.ID); err != nil {
+			slog.Warn("orphan GC: hard delete failed",
+				"attachment_id", a.ID, "error", err)
+			res.Skipped++
+			continue
+		}
+		res.Deleted++
+	}
+
+	return res, nil
+}
+
+// orphanGCConfig captures runtime knobs for the periodic loop.
+// Stored on Server via SetOrphanGCConfig so tests + cmd/pad can
+// override defaults independently.
+type orphanGCConfig struct {
+	mu       sync.Mutex
+	interval time.Duration
+	grace    time.Duration
+	stop     chan struct{}
+	running  bool
+}
+
+// SetOrphanGCConfig overrides the default sweep interval (24h) and
+// grace period (30d). Pass 0 for either to keep the package default.
+// Must be called before StartOrphanGC.
+func (s *Server) SetOrphanGCConfig(interval, grace time.Duration) {
+	s.orphanGC.mu.Lock()
+	defer s.orphanGC.mu.Unlock()
+	if interval > 0 {
+		s.orphanGC.interval = interval
+	}
+	if grace > 0 {
+		s.orphanGC.grace = grace
+	}
+}
+
+// StartOrphanGC kicks off the periodic sweep loop. Idempotent —
+// calling twice is a no-op (existing loop continues, second call
+// returns silently). Must be called AFTER SetAttachments; the loop
+// no-ops sweeps when the registry isn't wired so a server without
+// attachment storage doesn't log spurious errors.
+//
+// The loop is tracked by Server.bg so Stop() drains it before the
+// process exits / SQLite is closed (BUG-842 invariant).
+func (s *Server) StartOrphanGC() {
+	s.orphanGC.mu.Lock()
+	if s.orphanGC.running {
+		s.orphanGC.mu.Unlock()
+		return
+	}
+	if s.orphanGC.interval == 0 {
+		s.orphanGC.interval = defaultOrphanGCInterval
+	}
+	if s.orphanGC.grace == 0 {
+		s.orphanGC.grace = defaultOrphanGCGrace
+	}
+	s.orphanGC.stop = make(chan struct{})
+	s.orphanGC.running = true
+	interval := s.orphanGC.interval
+	grace := s.orphanGC.grace
+	stop := s.orphanGC.stop
+	s.orphanGC.mu.Unlock()
+
+	slog.Info("orphan GC started",
+		"interval", interval.String(), "grace", grace.String())
+
+	s.bg.Add(1)
+	go func() {
+		defer s.bg.Done()
+		t := time.NewTicker(interval)
+		defer t.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-t.C:
+				s.runOrphanGCTick(grace)
+			}
+		}
+	}()
+}
+
+// stopOrphanGC signals the loop to exit. Called from Server.Stop().
+// Safe to call when the loop never started.
+func (s *Server) stopOrphanGC() {
+	s.orphanGC.mu.Lock()
+	defer s.orphanGC.mu.Unlock()
+	if !s.orphanGC.running {
+		return
+	}
+	close(s.orphanGC.stop)
+	s.orphanGC.running = false
+}
+
+// runOrphanGCTick is one tick of the periodic loop. Wrapped with a
+// 30-minute cap on the sweep so a long-running scan can't pin the
+// goroutine across multiple intervals. Logged at info on success,
+// warn on failure.
+func (s *Server) runOrphanGCTick(grace time.Duration) {
+	if s.attachments == nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+	cutoff := time.Now().UTC().Add(-grace)
+	res, err := s.runOrphanGCSweep(ctx, cutoff)
+	if err != nil {
+		slog.Warn("orphan GC sweep failed", "error", err)
+		return
+	}
+	slog.Info("orphan GC sweep",
+		"scanned", res.Scanned,
+		"deleted", res.Deleted,
+		"blobs_reclaimed", res.BlobsReclaimed,
+		"bytes_reclaimed", res.BytesReclaimed,
+		"skipped", res.Skipped)
+}
+
+// _ keeps the attachments import alive even if every callsite ends
+// up only touching s.store — the storage-backend Resolve call lives
+// inside runOrphanGCSweep regardless.
+var _ = attachments.ErrNotFound

internal/server/orphan_gc_test.go

@@ -0,0 +1,530 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/PerpetualSoftware/pad/internal/attachments"
+)
+
+// TestOrphanGC_ReclaimsSoftDeleted pins TASK-886's main case: a
+// soft-deleted attachment past the grace period gets hard-deleted
+// from the DB AND its blob removed from the storage backend.
+func TestOrphanGC_ReclaimsSoftDeleted(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+
+	body := realPNG()
+	rr := doMultipartUpload(srv, slug, "doomed.png", body)
+	if rr.Code != 201 {
+		t.Fatalf("upload: %d %s", rr.Code, rr.Body.String())
+	}
+	id := getOnlyAttachmentID(t, srv, workspaceIDForSlug(t, srv, slug))
+
+	// Soft-delete via the user-facing endpoint.
+	rr = doRequest(srv, "DELETE", "/api/v1/workspaces/"+slug+"/attachments/"+id, nil)
+	if rr.Code != 204 {
+		t.Fatalf("delete: %d %s", rr.Code, rr.Body.String())
+	}
+
+	// Sanity check: row exists, soft-deleted.
+	att, err := srv.store.GetAttachment(id)
+	if err != nil {
+		t.Fatalf("GetAttachment: %v", err)
+	}
+	if att == nil || att.DeletedAt == nil {
+		t.Fatalf("expected soft-deleted row, got %+v", att)
+	}
+	storageKey := att.StorageKey
+	store, err := srv.attachments.Resolve(storageKey)
+	if err != nil {
+		t.Fatalf("resolve backend: %v", err)
+	}
+	if _, err := store.Stat(context.Background(), storageKey); err != nil {
+		t.Fatalf("blob missing before GC: %v", err)
+	}
+
+	// Run the sweep with a graceCutoff in the future so the soft-
+	// deleted row qualifies immediately.
+	res, err := srv.runOrphanGCSweep(context.Background(), time.Now().Add(time.Hour))
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted < 1 {
+		t.Errorf("Deleted=%d, want >= 1", res.Deleted)
+	}
+	if res.BlobsReclaimed < 1 {
+		t.Errorf("BlobsReclaimed=%d, want >= 1", res.BlobsReclaimed)
+	}
+	if res.BytesReclaimed != int64(len(body)) {
+		t.Errorf("BytesReclaimed=%d, want %d", res.BytesReclaimed, len(body))
+	}
+
+	// DB row gone.
+	att, err = srv.store.GetAttachment(id)
+	if err != nil {
+		t.Fatalf("post-GC GetAttachment: %v", err)
+	}
+	if att != nil {
+		t.Errorf("row still present after GC: %+v", att)
+	}
+	// Blob gone too.
+	if _, err := store.Stat(context.Background(), storageKey); !errors.Is(err, attachments.ErrNotFound) {
+		t.Errorf("blob still on disk after GC; Stat err=%v", err)
+	}
+}
+
+// TestOrphanGC_ReclaimsLongOrphans pins the never-attached path:
+// rows with item_id IS NULL AND deleted_at IS NULL aged past the
+// grace period get reclaimed.
+func TestOrphanGC_ReclaimsLongOrphans(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+
+	if rr := doMultipartUpload(srv, slug, "orphan.png", realPNG()); rr.Code != 201 {
+		t.Fatalf("upload: %d", rr.Code)
+	}
+	id := getOnlyAttachmentID(t, srv, workspaceIDForSlug(t, srv, slug))
+
+	// Push the row's created_at into the past via direct SQL — the
+	// upload handler stamps "now" and there's no API to backdate.
+	pastTs := time.Now().UTC().Add(-31 * 24 * time.Hour).Format(time.RFC3339)
+	if _, err := srv.store.DB().Exec(
+		`UPDATE attachments SET created_at = ? WHERE id = ?`, pastTs, id,
+	); err != nil {
+		t.Fatalf("backdate: %v", err)
+	}
+
+	// Sweep with a 30-day grace cutoff.
+	cutoff := time.Now().UTC().Add(-30 * 24 * time.Hour)
+	res, err := srv.runOrphanGCSweep(context.Background(), cutoff)
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted < 1 {
+		t.Errorf("Deleted=%d, want >= 1", res.Deleted)
+	}
+	if got, _ := srv.store.GetAttachment(id); got != nil {
+		t.Errorf("orphan still present after GC: %+v", got)
+	}
+}
+
+// TestOrphanGC_KeepsRecentRows pins the safety case: rows still
+// inside the grace window MUST NOT be reclaimed. Catches a typo
+// in the WHERE clause that would silently destroy live attachments.
+func TestOrphanGC_KeepsRecentRows(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+
+	if rr := doMultipartUpload(srv, slug, "fresh.png", realPNG()); rr.Code != 201 {
+		t.Fatalf("upload: %d", rr.Code)
+	}
+	id := getOnlyAttachmentID(t, srv, workspaceIDForSlug(t, srv, slug))
+
+	// Soft-delete it but use a grace cutoff way in the past so the
+	// row is NOT yet past grace.
+	rr := doRequest(srv, "DELETE", "/api/v1/workspaces/"+slug+"/attachments/"+id, nil)
+	if rr.Code != 204 {
+		t.Fatalf("delete: %d", rr.Code)
+	}
+	cutoff := time.Now().UTC().Add(-365 * 24 * time.Hour)
+	res, err := srv.runOrphanGCSweep(context.Background(), cutoff)
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted != 0 {
+		t.Errorf("Deleted=%d, want 0 (row still in grace)", res.Deleted)
+	}
+	// Row must still exist (soft-deleted).
+	if att, _ := srv.store.GetAttachment(id); att == nil {
+		t.Errorf("row hard-deleted while still in grace")
+	}
+}
+
+// TestOrphanGC_PreservesSharedBlob pins the dedupe-safety case:
+// when two rows reference the same content_hash and only one is
+// orphan, the row gets hard-deleted but the blob stays on disk so
+// the other row keeps working.
+func TestOrphanGC_PreservesSharedBlob(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+
+	body := realPNG()
+	// Two uploads with identical bytes → same content_hash → one
+	// physical blob on disk, two attachment rows.
+	if rr := doMultipartUpload(srv, slug, "a.png", body); rr.Code != 201 {
+		t.Fatalf("upload a: %d", rr.Code)
+	}
+	if rr := doMultipartUpload(srv, slug, "b.png", body); rr.Code != 201 {
+		t.Fatalf("upload b: %d", rr.Code)
+	}
+	wsID := workspaceIDForSlug(t, srv, slug)
+
+	// Pull both row IDs via direct SQL — easier than the public list
+	// API, which paginates and doesn't expose storage_key.
+	var firstID, secondID, sharedKey string
+	dbRows, err := srv.store.DB().Query(
+		`SELECT id, storage_key FROM attachments WHERE workspace_id = ? AND deleted_at IS NULL ORDER BY created_at, id`, wsID)
+	if err != nil {
+		t.Fatalf("list rows: %v", err)
+	}
+	for dbRows.Next() {
+		var id, key string
+		if err := dbRows.Scan(&id, &key); err != nil {
+			t.Fatalf("scan: %v", err)
+		}
+		if firstID == "" {
+			firstID = id
+			sharedKey = key
+		} else {
+			secondID = id
+		}
+	}
+	dbRows.Close()
+	if firstID == "" || secondID == "" || firstID == secondID {
+		t.Fatalf("expected two distinct row ids; got %q / %q", firstID, secondID)
+	}
+
+	// Soft-delete the second row FIRST while item_id is still NULL
+	// (the delete handler's orphan branch is happy with workspace
+	// owner role and doesn't run the requireItemVisible check that
+	// would otherwise 404 on a synthetic item_id).
+	rr := doRequest(srv, "DELETE", "/api/v1/workspaces/"+slug+"/attachments/"+secondID, nil)
+	if rr.Code != 204 {
+		t.Fatalf("delete: %d", rr.Code)
+	}
+
+	// Now tag the FIRST (still-live) row with a synthetic item_id so
+	// the never-attached-orphan path doesn't reclaim it under the
+	// future grace cutoff. The test is about dedupe-aware blob
+	// preservation, not the orphan-from-start case (covered by
+	// TestOrphanGC_ReclaimsLongOrphans).
+	if _, err := srv.store.DB().Exec(
+		`UPDATE attachments SET item_id = ? WHERE id = ?`,
+		"synthetic-item", firstID,
+	); err != nil {
+		t.Fatalf("attach first row: %v", err)
+	}
+
+	res, err := srv.runOrphanGCSweep(context.Background(), time.Now().Add(time.Hour))
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted < 1 {
+		t.Errorf("Deleted=%d, want >= 1", res.Deleted)
+	}
+	if res.BlobsReclaimed != 0 {
+		t.Errorf("BlobsReclaimed=%d, want 0 (other row still references the blob)",
+			res.BlobsReclaimed)
+	}
+
+	// First row still works — blob still on disk.
+	store, err := srv.attachments.Resolve(sharedKey)
+	if err != nil {
+		t.Fatalf("resolve: %v", err)
+	}
+	if _, err := store.Stat(context.Background(), sharedKey); err != nil {
+		t.Errorf("shared blob disappeared after GC: %v", err)
+	}
+}
+
+// TestOrphanGC_KeepsReferencedNeverAttachedRows pins Codex P1 on
+// PR #307 round 1: the editor's normal upload flow leaves
+// attachments.item_id NULL and only the markdown reference inside
+// item.content connects them. So a "never-attached" row past the
+// grace period might still be referenced — the GC must scan item
+// content before reclaiming.
+func TestOrphanGC_KeepsReferencedNeverAttachedRows(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+	wsID := workspaceIDForSlug(t, srv, slug)
+
+	if rr := doMultipartUpload(srv, slug, "kept.png", realPNG()); rr.Code != 201 {
+		t.Fatalf("upload: %d", rr.Code)
+	}
+	id := getOnlyAttachmentID(t, srv, wsID)
+
+	// Create an item whose content references the attachment, but
+	// don't update attachments.item_id — exactly mirrors the
+	// production editor flow (upload → PATCH content with the ref).
+	rr := doRequest(srv, "POST", "/api/v1/workspaces/"+slug+"/collections/docs/items",
+		map[string]any{"title": "Holds Image", "content": "ref: pad-attachment:" + id})
+	if rr.Code != 201 {
+		t.Fatalf("create item: %d %s", rr.Code, rr.Body.String())
+	}
+
+	// Backdate the attachment's created_at past the 30-day grace
+	// so the orphan SELECT picks it up.
+	pastTs := time.Now().UTC().Add(-31 * 24 * time.Hour).Format(time.RFC3339)
+	if _, err := srv.store.DB().Exec(
+		`UPDATE attachments SET created_at = ? WHERE id = ?`, pastTs, id,
+	); err != nil {
+		t.Fatalf("backdate: %v", err)
+	}
+
+	res, err := srv.runOrphanGCSweep(context.Background(),
+		time.Now().UTC().Add(-30*24*time.Hour))
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted != 0 {
+		t.Errorf("Deleted=%d, want 0 (item content references the attachment)", res.Deleted)
+	}
+	if got, _ := srv.store.GetAttachment(id); got == nil {
+		t.Errorf("referenced attachment was hard-deleted by GC; row gone")
+	}
+}
+
+// TestOrphanGC_RespectsInFlightUploads pins Codex P2 on PR #307
+// round 1: an upload that called Put but hasn't yet inserted the
+// attachments row must NOT lose its blob to GC reclamation of an
+// older soft-deleted row sharing the same hash.
+//
+// We simulate the race by registering an in-flight hash directly,
+// running a sweep against an old soft-deleted row at that hash,
+// and asserting the blob stayed.
+func TestOrphanGC_RespectsInFlightUploads(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+
+	if rr := doMultipartUpload(srv, slug, "victim.png", realPNG()); rr.Code != 201 {
+		t.Fatalf("upload: %d", rr.Code)
+	}
+	id := getOnlyAttachmentID(t, srv, workspaceIDForSlug(t, srv, slug))
+
+	att, _ := srv.store.GetAttachment(id)
+	if att == nil {
+		t.Fatal("expected attachment row")
+	}
+
+	// Soft-delete it.
+	rr := doRequest(srv, "DELETE", "/api/v1/workspaces/"+slug+"/attachments/"+id, nil)
+	if rr.Code != 204 {
+		t.Fatalf("delete: %d", rr.Code)
+	}
+
+	// Pretend an upload is in flight for the same hash. (Production
+	// upload code would have called this between Put and
+	// CreateAttachment; the GC must see the in-flight signal.)
+	release := srv.markUploadInFlight(att.ContentHash)
+	defer release()
+
+	res, err := srv.runOrphanGCSweep(context.Background(), time.Now().Add(time.Hour))
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted < 1 {
+		t.Errorf("Deleted=%d, want >= 1 (DB row should still go)", res.Deleted)
+	}
+	if res.BlobsReclaimed != 0 {
+		t.Errorf("BlobsReclaimed=%d, want 0 (in-flight upload protects the blob)",
+			res.BlobsReclaimed)
+	}
+	// Blob still on disk so the in-flight upload can complete.
+	store, err := srv.attachments.Resolve(att.StorageKey)
+	if err != nil {
+		t.Fatalf("resolve: %v", err)
+	}
+	if _, err := store.Stat(context.Background(), att.StorageKey); err != nil {
+		t.Errorf("blob disappeared despite in-flight signal: %v", err)
+	}
+}
+
+// TestOrphanGC_RespectsSoftDeletedInGracePeer pins Codex P2 round
+// 3: when two rows share a content_hash, GC reclaims the older one
+// past grace but MUST NOT delete the blob if the second row is
+// still inside its own grace window — the second row could be
+// restored / inspected and would otherwise hit a missing blob.
+func TestOrphanGC_RespectsSoftDeletedInGracePeer(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+	wsID := workspaceIDForSlug(t, srv, slug)
+
+	body := realPNG()
+	if rr := doMultipartUpload(srv, slug, "a.png", body); rr.Code != 201 {
+		t.Fatalf("upload a: %d", rr.Code)
+	}
+	if rr := doMultipartUpload(srv, slug, "b.png", body); rr.Code != 201 {
+		t.Fatalf("upload b: %d", rr.Code)
+	}
+
+	dbRows, err := srv.store.DB().Query(
+		`SELECT id, storage_key, content_hash FROM attachments WHERE workspace_id = ? AND deleted_at IS NULL ORDER BY created_at, id`, wsID)
+	if err != nil {
+		t.Fatalf("list rows: %v", err)
+	}
+	var firstID, secondID, sharedKey, sharedHash string
+	for dbRows.Next() {
+		var id, key, hash string
+		if err := dbRows.Scan(&id, &key, &hash); err != nil {
+			t.Fatalf("scan: %v", err)
+		}
+		if firstID == "" {
+			firstID = id
+			sharedKey = key
+			sharedHash = hash
+		} else {
+			secondID = id
+		}
+	}
+	dbRows.Close()
+	if firstID == "" || secondID == "" {
+		t.Fatalf("expected two rows; got %q / %q", firstID, secondID)
+	}
+
+	// Soft-delete BOTH rows. Then backdate ONLY the first row's
+	// deleted_at past the 30-day grace; the second stays "fresh".
+	for _, id := range []string{firstID, secondID} {
+		rr := doRequest(srv, "DELETE", "/api/v1/workspaces/"+slug+"/attachments/"+id, nil)
+		if rr.Code != 204 {
+			t.Fatalf("delete %s: %d", id, rr.Code)
+		}
+	}
+	pastTs := time.Now().UTC().Add(-31 * 24 * time.Hour).Format(time.RFC3339)
+	if _, err := srv.store.DB().Exec(
+		`UPDATE attachments SET deleted_at = ? WHERE id = ?`, pastTs, firstID,
+	); err != nil {
+		t.Fatalf("backdate first: %v", err)
+	}
+
+	// Sweep with a 30-day cutoff. Only the first row qualifies.
+	cutoff := time.Now().UTC().Add(-30 * 24 * time.Hour)
+	res, err := srv.runOrphanGCSweep(context.Background(), cutoff)
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted != 1 {
+		t.Errorf("Deleted=%d, want 1 (only the older row qualifies)", res.Deleted)
+	}
+	if res.BlobsReclaimed != 0 {
+		t.Errorf("BlobsReclaimed=%d, want 0 (newer soft-deleted peer is still in grace)",
+			res.BlobsReclaimed)
+	}
+	// Blob still on disk so the still-in-grace row's hypothetical
+	// undelete works.
+	store, err := srv.attachments.Resolve(sharedKey)
+	if err != nil {
+		t.Fatalf("resolve: %v", err)
+	}
+	if _, err := store.Stat(context.Background(), sharedKey); err != nil {
+		t.Errorf("shared blob disappeared while peer still in grace: %v", err)
+	}
+	_ = sharedHash
+}
+
+// TestOrphanGC_DedupesBlobReclaimMetric pins Codex round 4: when
+// multiple soft-deleted peers share a content_hash and all are
+// past grace, the blob is deleted on the first peer and the
+// remaining peers' Delete calls are idempotent no-ops. The earlier
+// version still bumped BlobsReclaimed / BytesReclaimed for each
+// no-op, inflating the sweep metrics.
+func TestOrphanGC_DedupesBlobReclaimMetric(t *testing.T) {
+	srv, slug := testServerWithAttachments(t)
+	wsID := workspaceIDForSlug(t, srv, slug)
+
+	body := realPNG()
+	if rr := doMultipartUpload(srv, slug, "a.png", body); rr.Code != 201 {
+		t.Fatalf("upload a: %d", rr.Code)
+	}
+	if rr := doMultipartUpload(srv, slug, "b.png", body); rr.Code != 201 {
+		t.Fatalf("upload b: %d", rr.Code)
+	}
+
+	// Soft-delete both rows + backdate both deleted_at past 30d so
+	// they BOTH qualify for reclamation in the same sweep.
+	dbRows, _ := srv.store.DB().Query(
+		`SELECT id FROM attachments WHERE workspace_id = ? AND deleted_at IS NULL`, wsID)
+	var ids []string
+	for dbRows.Next() {
+		var id string
+		dbRows.Scan(&id)
+		ids = append(ids, id)
+	}
+	dbRows.Close()
+	for _, id := range ids {
+		rr := doRequest(srv, "DELETE", "/api/v1/workspaces/"+slug+"/attachments/"+id, nil)
+		if rr.Code != 204 {
+			t.Fatalf("delete %s: %d", id, rr.Code)
+		}
+	}
+	pastTs := time.Now().UTC().Add(-31 * 24 * time.Hour).Format(time.RFC3339)
+	if _, err := srv.store.DB().Exec(
+		`UPDATE attachments SET deleted_at = ?`, pastTs,
+	); err != nil {
+		t.Fatalf("backdate: %v", err)
+	}
+
+	cutoff := time.Now().UTC().Add(-30 * 24 * time.Hour)
+	res, err := srv.runOrphanGCSweep(context.Background(), cutoff)
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if res.Deleted < 2 {
+		t.Errorf("Deleted=%d, want >= 2 (both rows past grace)", res.Deleted)
+	}
+	if res.BlobsReclaimed != 1 {
+		t.Errorf("BlobsReclaimed=%d, want 1 (single shared blob)", res.BlobsReclaimed)
+	}
+	if res.BytesReclaimed != int64(len(body)) {
+		t.Errorf("BytesReclaimed=%d, want %d (single shared blob's size)",
+			res.BytesReclaimed, len(body))
+	}
+}
+
+// TestInFlightUploadHashes_ConcurrentReleaseReacquire pins Codex P1
+// round 2 on PR #307: the prior sync.Map version raced when one
+// release's decrement-to-zero ran in parallel with another upload's
+// LoadOrStore-then-increment, leaving an in-flight upload's signal
+// invisible to the GC. The mutex-protected map closes the window.
+//
+// Stress test: hammer a single hash with overlapping
+// markUploadInFlight / release pairs. At every observation point,
+// uploadInFlight must report > 0 whenever ANY goroutine is between
+// its mark and its release.
+func TestInFlightUploadHashes_ConcurrentReleaseReacquire(t *testing.T) {
+	srv, _ := testServerWithAttachments(t)
+
+	const goroutines = 20
+	const iterations = 500
+	hash := "race-test-hash"
+
+	var wg sync.WaitGroup
+	wg.Add(goroutines)
+	// Spawn goroutines that each loop mark/release; all share the
+	// same hash so the inc/dec interleaving is maximized.
+	for i := 0; i < goroutines; i++ {
+		go func() {
+			defer wg.Done()
+			for j := 0; j < iterations; j++ {
+				release := srv.markUploadInFlight(hash)
+				// At least one goroutine (this one) is in flight.
+				if !srv.uploadInFlight(hash) {
+					t.Errorf("uploadInFlight=false while one goroutine holds the mark")
+				}
+				release()
+			}
+		}()
+	}
+	wg.Wait()
+
+	// After everyone finishes, the counter should be exactly zero
+	// and the map entry deleted.
+	if srv.uploadInFlight(hash) {
+		t.Errorf("uploadInFlight=true after all releases; counter leaked")
+	}
+}
+
+// TestOrphanGC_StartStop pins the lifecycle: StartOrphanGC kicks the
+// loop, Stop signals it to exit, and Server.Stop() actually drains.
+// Catches regressions where a leaked goroutine would compound across
+// every Stop cycle (BUG-851 echo).
+func TestOrphanGC_StartStop(t *testing.T) {
+	srv, _ := testServerWithAttachments(t)
+	srv.SetOrphanGCConfig(1*time.Millisecond, 24*time.Hour)
+	srv.StartOrphanGC()
+
+	// Calling start a second time is a no-op.
+	srv.StartOrphanGC()
+
+	// Stop drains in the t.Cleanup hook from testServer; just give
+	// the loop a tick to actually run a sweep.
+	time.Sleep(10 * time.Millisecond)
+	// If we got here without deadlocking on Stop, the loop drains
+	// correctly. testServer's t.Cleanup will exercise Stop.
+}

internal/server/server.go

@@ -86,6 +86,27 @@ type Server struct {
 	// exports can opt in without recompiling.
 	importBundleMaxBytes int64
 
+	// orphanGC holds the periodic-sweep config + lifecycle for the
+	// attachment orphan garbage collector (TASK-886). Configured via
+	// SetOrphanGCConfig and started via StartOrphanGC. Stop() signals
+	// the loop to exit and waits for it via the bg WaitGroup.
+	orphanGC orphanGCConfig
+
+	// inFlightUploadHashes tracks content_hash values for uploads
+	// that have called AttachmentStore.Put but not yet inserted the
+	// attachments row. Without this, the orphan GC could delete a
+	// blob between Put and CreateAttachment, leaving a live row that
+	// references a missing blob (Codex P2 on PR #307 round 1).
+	//
+	// A plain map + mutex rather than sync.Map: counters need
+	// atomic-with-delete semantics (decrement-then-delete-if-zero
+	// must be one critical section, not two — sync.Map.CompareAndDelete
+	// addresses the entry but not the inc/dec interleaving). Codex
+	// P1 round 2 caught the prior sync.Map version racing on
+	// release-vs-reload of the same hash.
+	inFlightHashesMu sync.Mutex
+	inFlightHashes   map[string]int64
+
 	// bg tracks fire-and-forget goroutines spawned by request handlers
 	// (TouchUserActivity in middleware_auth, async email sends, etc.) so
 	// the server can drain them before shutdown / test cleanup. Without
@@ -113,6 +134,10 @@ func (s *Server) goAsync(fn func()) {
 // Store.Close() so in-flight DB writes don't race a closed connection
 // (or worse, the SQLite -wal/-shm file removal in t.TempDir cleanup).
 func (s *Server) Stop() {
+	// Signal long-running background loops (orphan GC, etc.) to exit.
+	// Each loop registers itself on s.bg, so the Wait() below blocks
+	// until they actually finish and any in-flight goroutines drain.
+	s.stopOrphanGC()
 	s.bg.Wait()
 	s.rateLimiters.Stop() // nil-safe via the RateLimiters receiver guard
 }
@@ -286,6 +311,45 @@ func (s *Server) SetImageProcessor(p attachments.Processor) {
 	s.imageProcessor = p
 }
 
+// markUploadInFlight increments the in-flight counter for a content
+// hash. Returns a release func the caller MUST defer; the release
+// decrements and removes the entry once it hits zero. Used by the
+// upload handler to fence Put + CreateAttachment against orphan-GC
+// blob deletions of the same hash.
+//
+// Increment + map-store + decrement + delete all run under one
+// mutex so a concurrent uploadInFlight call can't observe a stale
+// "0" between the last release-decrement and the next-upload
+// increment. The earlier sync.Map version split increment from
+// LoadOrStore-then-atomic-add and missed that window (Codex P1 on
+// PR #307 round 2).
+func (s *Server) markUploadInFlight(hash string) func() {
+	s.inFlightHashesMu.Lock()
+	if s.inFlightHashes == nil {
+		s.inFlightHashes = make(map[string]int64)
+	}
+	s.inFlightHashes[hash]++
+	s.inFlightHashesMu.Unlock()
+	return func() {
+		s.inFlightHashesMu.Lock()
+		defer s.inFlightHashesMu.Unlock()
+		s.inFlightHashes[hash]--
+		if s.inFlightHashes[hash] <= 0 {
+			delete(s.inFlightHashes, hash)
+		}
+	}
+}
+
+// uploadInFlight reports whether any upload is currently materializing
+// a blob with the given hash. The orphan GC consults this before
+// deleting a blob — if an upload just finished Put but hasn't
+// inserted the row yet, GC must NOT reclaim the blob.
+func (s *Server) uploadInFlight(hash string) bool {
+	s.inFlightHashesMu.Lock()
+	defer s.inFlightHashesMu.Unlock()
+	return s.inFlightHashes[hash] > 0
+}
+
 // SetImportBundleMaxBytes overrides the default 2 GiB cap on a
 // single workspace import bundle. Set to 0 to fall back to the
 // default. Wired from PAD_IMPORT_BUNDLE_MAX_BYTES in cmd/pad/main.go

internal/store/attachments.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/PerpetualSoftware/pad/internal/models"
 )
@@ -508,6 +509,140 @@ func (s *Store) WorkspaceAttachments(workspaceID string, filters AttachmentListF
 	return out, total, nil
 }
 
+// OrphanedAttachments returns rows eligible for orphan GC reclamation
+// (TASK-886). Two cases qualify:
+//
+//   - Never-attached uploads: item_id IS NULL AND deleted_at IS NULL
+//     AND created_at < grace cutoff. The editor uploads first and
+//     PATCHes the item content second; a tab-close in between leaves
+//     a row in this state. 30 days is comfortable headroom for that
+//     race plus any deferred-attachment workflow we add later.
+//
+//   - Soft-deleted past grace: deleted_at IS NOT NULL AND
+//     deleted_at < grace cutoff. Delete handlers tombstone rows
+//     immediately so undelete is possible; GC reclaims after the
+//     grace period.
+//
+// Both filters compare the timestamp column (TEXT, ISO 8601 UTC)
+// against the cutoff string lexicographically — UTC RFC3339 collates
+// in chronological order without parsing. The cutoff is computed by
+// the caller so tests can inject a deterministic time.
+//
+// Returned rows include thumbnail variants (parent_id != NULL) when
+// they meet either criterion — soft-deleting an original cascades
+// to its thumbnails (SoftDeleteAttachment), so they all share the
+// same deleted_at and reach the GC together.
+func (s *Store) OrphanedAttachments(graceCutoff time.Time) ([]models.Attachment, error) {
+	cutoffStr := graceCutoff.UTC().Format(time.RFC3339)
+	rows, err := s.db.Query(s.q(`
+		SELECT `+attachmentColumns+`
+		FROM attachments
+		WHERE
+		  (item_id IS NULL AND deleted_at IS NULL AND created_at < ?)
+		  OR
+		  (deleted_at IS NOT NULL AND deleted_at < ?)
+		ORDER BY created_at, id
+	`), cutoffStr, cutoffStr)
+	if err != nil {
+		return nil, fmt.Errorf("orphaned attachments: %w", err)
+	}
+	defer rows.Close()
+
+	var out []models.Attachment
+	for rows.Next() {
+		a, err := scanAttachment(rows)
+		if err != nil {
+			return nil, fmt.Errorf("scan orphan: %w", err)
+		}
+		out = append(out, *a)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate orphans: %w", err)
+	}
+	return out, nil
+}
+
+// HardDeleteAttachment removes the attachments row outright. Used by
+// orphan GC after the grace period; never call from a request
+// handler — soft-delete is the safe default for user-facing flows.
+func (s *Store) HardDeleteAttachment(id string) error {
+	_, err := s.db.Exec(s.q(`DELETE FROM attachments WHERE id = ?`), id)
+	if err != nil {
+		return fmt.Errorf("hard delete attachment: %w", err)
+	}
+	return nil
+}
+
+// CountProtectingAttachmentsForHash returns the number of rows
+// pointing at the given content_hash whose presence requires the
+// on-disk blob to stay. A row protects the blob when it is either:
+//
+//   - live (deleted_at IS NULL), or
+//   - soft-deleted but still inside the grace window
+//     (deleted_at >= graceCutoff)
+//
+// excludeID is the row currently being GC'd; we don't count it
+// against itself. Codex P2 round 3 caught the earlier version's
+// gap: counting only deleted_at IS NULL would have GC reclaim the
+// blob from row A (soft-deleted 31d ago) even though row B is
+// also soft-deleted but still 1 day old — within grace, so its
+// blob must stay reachable until its own grace expires.
+func (s *Store) CountProtectingAttachmentsForHash(hash, excludeID string, graceCutoff time.Time) (int, error) {
+	var n int
+	cutoff := graceCutoff.UTC().Format(time.RFC3339)
+	err := s.db.QueryRow(s.q(`
+		SELECT COUNT(*) FROM attachments
+		WHERE content_hash = ? AND id <> ?
+		  AND (deleted_at IS NULL OR deleted_at >= ?)
+	`), hash, excludeID, cutoff).Scan(&n)
+	if err != nil {
+		return 0, fmt.Errorf("count protecting attachments for hash: %w", err)
+	}
+	return n, nil
+}
+
+// AttachmentReferencedInItems returns true when any live item in the
+// workspace mentions "pad-attachment:<id>" in its content or fields
+// JSON. The editor upload flow leaves attachments.item_id NULL —
+// the canonical association is the markdown reference inside the
+// item's content, NOT a column in the attachments table — so the
+// orphan GC has to look at item content directly before reclaiming
+// a "never-attached" row, otherwise it'd hard-delete attachments
+// that markdown still points at.
+//
+// Scoped to one workspace because a "pad-attachment:UUID" reference
+// only resolves within the workspace where the attachment lives;
+// cross-workspace references are intentionally not supported.
+//
+// Dialect note: items.fields is TEXT on SQLite but JSONB on
+// PostgreSQL (see migrations + pgmigrations). LIKE doesn't work on
+// JSONB so the Postgres path casts to text first. Codex P1 round 2
+// caught this — without the cast, the GC's reference scan errored
+// on Postgres and every never-attached row got skipped.
+func (s *Store) AttachmentReferencedInItems(workspaceID, attachmentID string) (bool, error) {
+	if workspaceID == "" || attachmentID == "" {
+		return false, nil
+	}
+	needle := "pad-attachment:" + attachmentID
+	pattern := "%" + needle + "%"
+
+	fieldsExpr := "fields"
+	if s.dialect.Driver() == DriverPostgres {
+		fieldsExpr = "fields::text"
+	}
+
+	var n int
+	err := s.db.QueryRow(s.q(`
+		SELECT COUNT(*) FROM items
+		WHERE workspace_id = ? AND deleted_at IS NULL
+		  AND (content LIKE ? OR `+fieldsExpr+` LIKE ?)
+	`), workspaceID, pattern, pattern).Scan(&n)
+	if err != nil {
+		return false, fmt.Errorf("attachment referenced in items: %w", err)
+	}
+	return n > 0, nil
+}
+
 // SoftDeleteAttachment marks the given attachment row deleted (and
 // every variant whose parent_id points at it) so the orphan GC will
 // reclaim the bytes after the grace period. Returns sql.ErrNoRows if