appview: support OCI manifest DELETE

DELETE /v2/<name>/manifests/<ref> answered UNSUPPORTED before ever
reaching the ATProto-backed stores: distribution v3.1.1's DeleteManifest
handler short-circuits unless app.deleteEnabled is set, which comes from
storage.delete.enabled. Set it (mirrored in the test harness). The
companion storage.EnableDelete option it appends only affects
distribution's built-in store, which RoutingRepository replaces, so it is
a no-op for us.

With the route reachable, make the stores do the right thing:

  - ManifestStore.Delete purges the hold's per-layer, scan and image
    config records on a detached context, since the DELETE handler
    returns immediately and cancels the request context.
  - TagStore.Untag resolves the digest before deleting the tag record,
    then deletes the manifest if that was its last tag and it is not a
    manifest list child, matching the web UI's delete-tag behavior so
    deleting an only-tagged image doesn't orphan the manifest.
  - cleanupUntaggedManifest becomes package-level over *RegistryContext
    so both stores share one implementation.
  - purgeOnHold moves out of handlers into pkg/appview/holdpurge so the
    storage layer can call it: handlers already depends on storage via
    middleware, so storage to handlers would be an import cycle.
  - ProxyBlobStore.Delete returns distribution.ErrUnsupported, so the
    always-registered blob DELETE route gives a clean OCI UNSUPPORTED
    error instead of a generic 500. Layer bytes are reclaimed by the
    hold's refcounted GC.

The cascade's still-tagged re-check pages through the tag records rather
than reading a single capped page. Tags for all of a user's repositories
share one collection, so one page is a per-account budget: past ~100 tags
a live tag fell off the end and the manifest was deleted while still
referenced. Incomplete enumeration now skips the delete, since an
orphaned manifest is recoverable and a deleted live one is not.

This also makes the over-quota delete grant added in 6e426dc load-bearing:
it hands out pull,delete tokens, which could not do anything while
distribution rejected every DELETE.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

internal/testharness/harness.go

@@ -463,6 +463,9 @@ func buildDistributionConfig(addr, baseURL, holdDID string, services []string, c
 				"dryrun":   false,
 			},
 		},
+		// Mirror buildStorageConfig: distribution v3.1.1's DeleteManifest
+		// handler returns UNSUPPORTED unless delete is enabled here.
+		"delete": configuration.Parameters{"enabled": true},
 	}
 	distConfig.Middleware = map[string][]configuration.Middleware{
 		"registry": {{

pkg/appview/config.go

@@ -472,6 +472,17 @@ func buildStorageConfig() configuration.Storage {
 		},
 	}
 
+	// Enable manifest deletion. distribution v3.1.1's DeleteManifest handler
+	// short-circuits with an UNSUPPORTED error unless app.deleteEnabled is set,
+	// which comes from this flag (see registry/handlers/manifests.go). Without
+	// it, `skopeo delete` / OCI DELETE returns "unsupported" before ever reaching
+	// our ATProto-backed ManifestStore.Delete / TagStore.Untag. The companion
+	// storage.EnableDelete option this appends only affects distribution's own
+	// built-in store, which we replace via RoutingRepository, so it's a no-op.
+	storage["delete"] = configuration.Parameters{
+		"enabled": true,
+	}
+
 	return storage
 }
 

pkg/appview/config_test.go

@@ -82,6 +82,17 @@ func TestBuildStorageConfig(t *testing.T) {
 	if purging["enabled"] != false {
 		t.Error("uploadpurging enabled should be false")
 	}
+
+	// Manifest deletion must be enabled: distribution v3.1.1's DeleteManifest
+	// handler short-circuits with UNSUPPORTED unless storage.delete.enabled is
+	// true, which is what lets `skopeo delete` / OCI DELETE reach our stores.
+	deleteCfg, ok := got["delete"]
+	if !ok {
+		t.Fatal("buildStorageConfig() missing delete config — OCI DELETE would return UNSUPPORTED")
+	}
+	if deleteCfg["enabled"] != true {
+		t.Errorf("storage.delete.enabled = %v, want true", deleteCfg["enabled"])
+	}
 }
 
 func TestBuildMiddlewareConfig(t *testing.T) {

pkg/appview/handlers/images.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"atcr.io/pkg/appview/db"
+	"atcr.io/pkg/appview/holdpurge"
 	"atcr.io/pkg/appview/middleware"
 	"atcr.io/pkg/atproto"
 	"github.com/go-chi/render"
@@ -100,7 +101,7 @@ func (h *DeleteTagHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			if err := db.DeleteManifest(h.DB, user.DID, repo, digest); err != nil {
 				slog.Warn("delete-tag: cascade DB delete failed", "did", user.DID, "digest", digest, "error", err)
 			}
-			purgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, atproto.BuildManifestURI(user.DID, digest))
+			holdpurge.PurgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, atproto.BuildManifestURI(user.DID, digest))
 		}
 	}
 
@@ -235,7 +236,7 @@ func (h *DeleteManifestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request
 	// manifest. Best-effort — failures here only mean the hold's lazy GC
 	// will clean up later, so we don't reflect the failure to the user.
 	manifestURI := atproto.BuildManifestURI(user.DID, digest)
-	purgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI)
+	holdpurge.PurgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI)
 
 	w.WriteHeader(http.StatusOK)
 }
@@ -321,7 +322,7 @@ func (h *DeleteUntaggedManifestsHandler) ServeHTTP(w http.ResponseWriter, r *htt
 		}
 
 		manifestURI := atproto.BuildManifestURI(user.DID, digest)
-		purgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI)
+		holdpurge.PurgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI)
 
 		deleted++
 	}

pkg/appview/holdpurge/purge.go

@@ -1,4 +1,8 @@
-package handlers
+// Package holdpurge asks a hold to drop the per-manifest records (layer, scan,
+// image-config) associated with a deleted manifest. It lives in its own leaf
+// package so both the web-UI handlers (pkg/appview/handlers) and the storage
+// routing layer (pkg/appview/storage) can call it without an import cycle.
+package holdpurge
 
 import (
 	"bytes"
@@ -19,7 +23,7 @@ type purgeManifestRequest struct {
 	ManifestURI string `json:"manifestUri"`
 }
 
-// purgeOnHold tells the hold to delete the layer, scan, and image-config
+// PurgeOnHold tells the hold to delete the layer, scan, and image-config
 // records associated with a single manifest. This is best-effort: callers
 // should treat all errors as "log and continue" because lazy GC on the hold
 // will catch up either way (and on third-party holds the user may not even
@@ -29,12 +33,12 @@ type purgeManifestRequest struct {
 // `hold_endpoint` column on the manifests row, or a freshly-resolved value
 // from the manifest record). userDID + pdsEndpoint are the OAuth-acting
 // user — the service token is minted from their PDS with audience = holdDID.
-func purgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEndpoint, holdDID, manifestURI string) {
+func PurgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEndpoint, holdDID, manifestURI string) {
 	if holdDID == "" || manifestURI == "" {
 		return
 	}
 	if refresher == nil {
-		slog.Debug("purgeOnHold: OAuth refresher unavailable; skipping",
+		slog.Debug("PurgeOnHold: OAuth refresher unavailable; skipping",
 			"hold_did", holdDID, "manifest", manifestURI)
 		return
 	}
@@ -44,21 +48,21 @@ func purgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEn
 
 	holdURL, err := atproto.ResolveHoldURL(timeoutCtx, holdDID)
 	if err != nil {
-		slog.Warn("purgeOnHold: failed to resolve hold URL",
+		slog.Warn("PurgeOnHold: failed to resolve hold URL",
 			"hold_did", holdDID, "error", err)
 		return
 	}
 
 	serviceToken, err := auth.GetOrFetchServiceToken(timeoutCtx, refresher, userDID, holdDID, pdsEndpoint)
 	if err != nil {
-		slog.Warn("purgeOnHold: failed to mint service token",
+		slog.Warn("PurgeOnHold: failed to mint service token",
 			"hold_did", holdDID, "user_did", userDID, "error", err)
 		return
 	}
 
 	body, err := json.Marshal(purgeManifestRequest{ManifestURI: manifestURI})
 	if err != nil {
-		slog.Warn("purgeOnHold: failed to marshal request",
+		slog.Warn("PurgeOnHold: failed to marshal request",
 			"hold_did", holdDID, "error", err)
 		return
 	}
@@ -66,7 +70,7 @@ func purgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEn
 	req, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost,
 		holdURL+atproto.HoldPurgeManifest, bytes.NewReader(body))
 	if err != nil {
-		slog.Warn("purgeOnHold: failed to create request",
+		slog.Warn("PurgeOnHold: failed to create request",
 			"hold_did", holdDID, "error", err)
 		return
 	}
@@ -75,7 +79,7 @@ func purgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEn
 
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		slog.Warn("purgeOnHold: request failed",
+		slog.Warn("PurgeOnHold: request failed",
 			"hold_did", holdDID, "manifest", manifestURI, "error", err)
 		return
 	}
@@ -84,13 +88,13 @@ func purgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEn
 	if resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusUnauthorized {
 		// Sailor pushing to a third-party hold won't have captain/crew-admin
 		// rights; that's expected. Lazy GC on that hold will reclaim later.
-		slog.Debug("purgeOnHold: not authorized on hold (lazy GC will handle)",
+		slog.Debug("PurgeOnHold: not authorized on hold (lazy GC will handle)",
 			"hold_did", holdDID, "manifest", manifestURI, "status", resp.StatusCode)
 		return
 	}
 	if resp.StatusCode != http.StatusOK {
 		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
-		slog.Warn("purgeOnHold: hold returned non-OK status",
+		slog.Warn("PurgeOnHold: hold returned non-OK status",
 			"hold_did", holdDID, "manifest", manifestURI,
 			"status", resp.StatusCode, "body", string(body))
 		return
@@ -103,12 +107,12 @@ func purgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEn
 		ImageConfigDeleted bool `json:"imageConfigDeleted"`
 	}
 	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
-		slog.Warn("purgeOnHold: failed to parse response",
+		slog.Warn("PurgeOnHold: failed to parse response",
 			"hold_did", holdDID, "manifest", manifestURI, "error", err)
 		return
 	}
 
-	slog.Info("purgeOnHold: purge succeeded",
+	slog.Info("PurgeOnHold: purge succeeded",
 		"hold_did", holdDID,
 		"manifest", manifestURI,
 		"layers_deleted", out.LayersDeleted,

pkg/appview/storage/manifest_store.go

@@ -13,6 +13,7 @@ import (
 	"sync"
 	"time"
 
+	"atcr.io/pkg/appview/holdpurge"
 	"atcr.io/pkg/appview/readme"
 	"atcr.io/pkg/atproto"
 	"github.com/distribution/distribution/v3"
@@ -326,7 +327,7 @@ func (s *ManifestStore) Put(ctx context.Context, manifest distribution.Manifest,
 	// Auto-remove untagged manifests: if the tag pointed to a different digest before,
 	// check if the old manifest is now untagged and clean it up
 	if s.ctx.AutoRemoveUntagged && oldDigest != "" && oldDigest != dgst.String() {
-		go s.cleanupUntaggedManifest(oldDigest)
+		go cleanupUntaggedManifest(s.ctx, oldDigest)
 	}
 
 	// Notify hold about manifest push (for layer tracking, Bluesky posts, and stats)
@@ -430,7 +431,13 @@ func (s *ManifestStore) Put(ctx context.Context, manifest distribution.Manifest,
 	return dgst, nil
 }
 
-// Delete removes a manifest
+// Delete removes a manifest. This is the digest-reference path of an OCI
+// `DELETE /v2/<name>/manifests/<digest>` (what `skopeo delete` uses after
+// resolving a tag). After removing the io.atcr.manifest record from the PDS it
+// asks the hold to drop the manifest's per-layer/scan/config records and free
+// any now-unreferenced blobs. The local appview DB cache is reconciled by the
+// firehose #delete handler (jetstream), consistent with how the push path
+// relies on the firehose rather than writing the cache inline.
 func (s *ManifestStore) Delete(ctx context.Context, dgst digest.Digest) error {
 	rkey := digestToRKey(dgst)
 	if err := s.ctx.ATProtoClient.DeleteRecord(ctx, atproto.ManifestCollection, rkey); err != nil {
@@ -439,9 +446,24 @@ func (s *ManifestStore) Delete(ctx context.Context, dgst digest.Digest) error {
 		}
 		return err
 	}
+	purgeDeletedManifest(s.ctx, dgst)
 	return nil
 }
 
+// purgeDeletedManifest asks the hold, in the background, to drop the layer,
+// scan, and image-config records for a manifest we just deleted from the PDS,
+// freeing any blobs no longer referenced by another manifest. Best-effort: the
+// hold's reference-counted GC is the backstop, so failures are only logged.
+// Runs on a detached context because the distribution DELETE handler returns as
+// soon as Delete/Untag return, which cancels the request context.
+func purgeDeletedManifest(rctx *RegistryContext, dgst digest.Digest) {
+	if rctx == nil || rctx.Refresher == nil || rctx.HoldDID == "" {
+		return
+	}
+	manifestURI := atproto.BuildManifestURI(rctx.DID, dgst.String())
+	go holdpurge.PurgeOnHold(context.Background(), rctx.Refresher, rctx.DID, rctx.PDSEndpoint, rctx.HoldDID, manifestURI)
+}
+
 // digestToRKey converts a digest to an ATProto record key
 // ATProto rkeys must be valid strings, so we use the digest string without the algorithm prefix
 func digestToRKey(dgst digest.Digest) string {
@@ -838,9 +860,14 @@ func detectImageMimeType(contentType, url string) string {
 	return ""
 }
 
-// cleanupUntaggedManifest checks if a manifest digest is now untagged and deletes it.
-// Runs asynchronously after a tag overwrite. Protects manifest list children.
-func (s *ManifestStore) cleanupUntaggedManifest(oldDigest string) {
+// cleanupUntaggedManifest checks if a manifest digest is now untagged and
+// deletes it, protecting manifest list children. It runs asynchronously in two
+// situations: after a tag overwrite on push (auto-remove-untagged), and as the
+// cascade after an OCI tag delete (TagStore.Untag) — deleting the manifest whose
+// last tag was just removed. On delete it also purges the manifest's records on
+// the hold. It is package-level (takes *RegistryContext) so both the manifest
+// store and the tag store can call it.
+func cleanupUntaggedManifest(rctx *RegistryContext, oldDigest string) {
 	defer func() {
 		if r := recover(); r != nil {
 			slog.Error("Panic in cleanupUntaggedManifest", "panic", r)
@@ -850,33 +877,59 @@ func (s *ManifestStore) cleanupUntaggedManifest(oldDigest string) {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
-	// Re-check: list all tags for this repository and see if any still point to the old digest
-	records, err := s.ctx.ATProtoClient.ListRecords(ctx, atproto.TagCollection, 100)
-	if err != nil {
-		slog.Warn("Auto-remove: failed to list tags", "error", err)
-		return
-	}
-
-	for _, rec := range records {
-		var tagRecord atproto.TagRecord
-		if json.Unmarshal(rec.Value, &tagRecord) != nil {
-			continue
-		}
-		if tagRecord.Repository != s.ctx.Repository {
-			continue
-		}
-		d, dErr := tagRecord.GetManifestDigest()
-		if dErr == nil && d == oldDigest {
-			// Still tagged by another tag — do not delete
-			slog.Debug("Auto-remove: manifest still tagged, skipping",
-				"digest", oldDigest, "tag", tagRecord.Tag)
+	// Re-check: page through the tag records and see if any still points at the
+	// old digest.
+	//
+	// This has to enumerate exhaustively. io.atcr.tag records for every one of
+	// the user's repositories share a single collection, so a single capped page
+	// is a per-account budget, not a per-repo one: past that many tags a live tag
+	// falls off the end, the check concludes "untagged", and a manifest that is
+	// still referenced gets deleted. Any failure to enumerate completely must
+	// therefore skip the delete rather than assume the digest is unreferenced.
+	const tagPageSize = 100
+	const maxTagPages = 100
+	cursor := ""
+	for page := 0; ; page++ {
+		if page >= maxTagPages {
+			// Refuse to guess past the budget: leaving an untagged manifest behind
+			// is recoverable, deleting a live one is not.
+			slog.Warn("Auto-remove: tag listing exceeded page budget, skipping delete",
+				"digest", oldDigest, "repository", rctx.Repository, "pages", maxTagPages)
 			return
 		}
+
+		records, next, err := rctx.ATProtoClient.ListRecordsWithCursor(ctx, atproto.TagCollection, tagPageSize, cursor)
+		if err != nil {
+			slog.Warn("Auto-remove: failed to list tags", "error", err)
+			return
+		}
+
+		for _, rec := range records {
+			var tagRecord atproto.TagRecord
+			if json.Unmarshal(rec.Value, &tagRecord) != nil {
+				continue
+			}
+			if tagRecord.Repository != rctx.Repository {
+				continue
+			}
+			d, dErr := tagRecord.GetManifestDigest()
+			if dErr == nil && d == oldDigest {
+				// Still tagged by another tag — do not delete
+				slog.Debug("Auto-remove: manifest still tagged, skipping",
+					"digest", oldDigest, "tag", tagRecord.Tag)
+				return
+			}
+		}
+
+		if next == "" || len(records) == 0 {
+			break
+		}
+		cursor = next
 	}
 
 	// Check if this digest is a child of a manifest list (multi-arch)
-	if s.ctx.ManifestRefChecker != nil {
-		referenced, err := s.ctx.ManifestRefChecker.IsManifestReferenced(s.ctx.DID, oldDigest)
+	if rctx.ManifestRefChecker != nil {
+		referenced, err := rctx.ManifestRefChecker.IsManifestReferenced(rctx.DID, oldDigest)
 		if err != nil {
 			slog.Warn("Auto-remove: failed to check manifest references", "digest", oldDigest, "error", err)
 			return
@@ -896,14 +949,17 @@ func (s *ManifestStore) cleanupUntaggedManifest(oldDigest string) {
 	}
 
 	rkey := digestToRKey(dgst)
-	if err := s.ctx.ATProtoClient.DeleteRecord(ctx, atproto.ManifestCollection, rkey); err != nil {
+	if err := rctx.ATProtoClient.DeleteRecord(ctx, atproto.ManifestCollection, rkey); err != nil {
 		slog.Warn("Auto-remove: failed to delete untagged manifest", "digest", oldDigest, "error", err)
 		return
 	}
 
+	// Free the manifest's records/blobs on the hold (best-effort, background).
+	purgeDeletedManifest(rctx, dgst)
+
 	slog.Info("Auto-removed untagged manifest",
 		"component", "manifest-store",
 		"digest", oldDigest,
-		"repository", s.ctx.Repository,
-		"did", s.ctx.DID)
+		"repository", rctx.Repository,
+		"did", rctx.DID)
 }

pkg/appview/storage/proxy_blob_store.go

@@ -252,10 +252,15 @@ func (p *ProxyBlobStore) Put(ctx context.Context, mediaType string, content []by
 	return desc, nil
 }
 
-// Delete removes a blob
+// Delete removes a blob.
+//
+// Blob deletion is not offered on the client-facing OCI path: layer bytes live
+// in the hold's S3 and are reclaimed by the hold's reference-counted GC once no
+// manifest references them (see PurgeOnHold on manifest delete). Returning the
+// distribution.ErrUnsupported sentinel makes the (always-registered) blob DELETE
+// route respond with a clean OCI UNSUPPORTED error instead of a generic 500.
 func (p *ProxyBlobStore) Delete(ctx context.Context, dgst digest.Digest) error {
-	// Not implemented - storage service would need a delete endpoint
-	return fmt.Errorf("delete not supported for proxy blob store")
+	return distribution.ErrUnsupported
 }
 
 // ServeBlob serves a blob via HTTP redirect or proxied response

pkg/appview/storage/routing_repository.go

@@ -108,5 +108,7 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore {
 // Tags returns the tag service
 // Tags are stored in ATProto as io.atcr.tag records
 func (r *RoutingRepository) Tags(ctx context.Context) distribution.TagService {
-	return NewTagStore(r.Ctx.ATProtoClient, r.Ctx.Repository)
+	ts := NewTagStore(r.Ctx.ATProtoClient, r.Ctx.Repository)
+	ts.rctx = r.Ctx // enables cascade-on-last-tag in Untag
+	return ts
 }

pkg/appview/storage/tag_store.go

@@ -16,6 +16,10 @@ import (
 type TagStore struct {
 	client     *atproto.Client
 	repository string
+	// rctx enables the cascade-on-last-tag behavior in Untag. It is nil in
+	// unit tests that construct a bare store via NewTagStore; cascade is then
+	// skipped. RoutingRepository.Tags() populates it for real requests.
+	rctx *RegistryContext
 }
 
 // NewTagStore creates a new ATProto-backed tag store
@@ -82,10 +86,37 @@ func (s *TagStore) Tag(ctx context.Context, tag string, desc distribution.Descri
 	return nil
 }
 
-// Untag removes a tag
+// Untag removes a tag. This is the tag-reference path of an OCI
+// `DELETE /v2/<name>/manifests/<tag>` (and the per-tag cleanup the distribution
+// handler runs after a digest delete). After deleting the io.atcr.tag record it
+// cascade-deletes the manifest if that was its last tag and it is not a manifest
+// list child — matching the web UI's "delete tag" behavior so `skopeo delete`
+// of an only-tagged image doesn't leave an orphaned untagged manifest behind.
 func (s *TagStore) Untag(ctx context.Context, tag string) error {
+	// Resolve the digest this tag points to before deleting it — we need it for
+	// the cascade decision and can't recover it once the record is gone. Only
+	// do this when cascade is enabled (rctx set); a missing/invalid tag record
+	// just means no cascade.
+	var manifestDigest string
+	if s.rctx != nil {
+		if desc, err := s.Get(ctx, tag); err == nil {
+			manifestDigest = desc.Digest.String()
+		}
+	}
+
 	rkey := atproto.RepositoryTagToRKey(s.repository, tag)
-	return s.client.DeleteRecord(ctx, atproto.TagCollection, rkey)
+	if err := s.client.DeleteRecord(ctx, atproto.TagCollection, rkey); err != nil {
+		return err
+	}
+
+	// Cascade: if the manifest is now untagged (and not a multi-arch child),
+	// delete it and purge its records on the hold. Best-effort in the
+	// background; cleanupUntaggedManifest re-checks for remaining tags so it is
+	// safe even when this Untag is part of a digest-delete's tag cleanup.
+	if s.rctx != nil && manifestDigest != "" {
+		go cleanupUntaggedManifest(s.rctx, manifestDigest)
+	}
+	return nil
 }
 
 // All returns all tags for this repository

test/integration/manifest_delete_test.go

@@ -0,0 +1,114 @@
+//go:build integration
+
+package integration
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/google/go-containerregistry/pkg/crane"
+	"github.com/google/go-containerregistry/pkg/v1/random"
+
+	"atcr.io/internal/testharness"
+
+	_ "github.com/distribution/distribution/v3/registry/auth/token"
+	_ "github.com/distribution/distribution/v3/registry/storage/driver/inmemory"
+)
+
+// TestManifestDelete is the end-to-end regression test for OCI manifest
+// deletion. It guards the core fix (storage.delete.enabled) — without it,
+// distribution v3.1.1's DeleteManifest handler answers every DELETE with an
+// UNSUPPORTED error before reaching our ATProto-backed stores, which is exactly
+// the `skopeo delete ... unsupported` symptom this feature addresses.
+//
+// It covers both reference forms an OCI DELETE can take:
+//   - by digest (what `skopeo delete` issues after resolving the tag) -> our
+//     ManifestStore.Delete
+//   - by tag -> our TagStore.Untag
+func TestManifestDelete(t *testing.T) {
+	h := testharness.New(t)
+	alice := h.AddSailor("alice.test")
+	creds := h.RegistryCreds(alice)
+
+	authOpts := []crane.Option{crane.WithAuth(toAuthn(creds)), crane.Insecure}
+
+	t.Run("by digest", func(t *testing.T) {
+		repoRef := fmt.Sprintf("%s/%s/del-digest", h.AppViewHostPort(), alice.Handle())
+		tagRef := repoRef + ":tag"
+
+		img, err := random.Image(1<<20, 2)
+		if err != nil {
+			t.Fatalf("build image: %v", err)
+		}
+		if err := crane.Push(img, tagRef, authOpts...); err != nil {
+			t.Fatalf("push: %v", err)
+		}
+
+		dgst, err := img.Digest()
+		if err != nil {
+			t.Fatalf("digest: %v", err)
+		}
+		digestRef := fmt.Sprintf("%s@%s", repoRef, dgst.String())
+
+		// Sanity: it is pullable before deletion.
+		if _, err := crane.Head(tagRef, authOpts...); err != nil {
+			t.Fatalf("pre-delete head: %v", err)
+		}
+
+		// Delete by digest — must succeed (202), not return UNSUPPORTED.
+		if err := crane.Delete(digestRef, authOpts...); err != nil {
+			t.Fatalf("delete by digest: %v", err)
+		}
+
+		// The manifest (and its tag) are gone.
+		if _, err := crane.Head(digestRef, authOpts...); err == nil {
+			t.Error("expected HEAD by digest to fail after delete, got nil")
+		}
+		if _, err := crane.Head(tagRef, authOpts...); err == nil {
+			t.Error("expected HEAD by tag to fail after digest delete, got nil")
+		}
+	})
+
+	t.Run("by tag", func(t *testing.T) {
+		tagRef := fmt.Sprintf("%s/%s/del-tag:tag", h.AppViewHostPort(), alice.Handle())
+
+		img, err := random.Image(1<<20, 2)
+		if err != nil {
+			t.Fatalf("build image: %v", err)
+		}
+		if err := crane.Push(img, tagRef, authOpts...); err != nil {
+			t.Fatalf("push: %v", err)
+		}
+
+		// Delete by tag — must succeed, exercising TagStore.Untag.
+		if err := crane.Delete(tagRef, authOpts...); err != nil {
+			t.Fatalf("delete by tag: %v", err)
+		}
+
+		// The tag no longer resolves.
+		if _, err := crane.Head(tagRef, authOpts...); err == nil {
+			t.Error("expected HEAD by tag to fail after tag delete, got nil")
+		}
+	})
+
+	t.Run("anonymous delete denied", func(t *testing.T) {
+		tagRef := fmt.Sprintf("%s/%s/del-anon:tag", h.AppViewHostPort(), alice.Handle())
+
+		img, err := random.Image(1<<20, 1)
+		if err != nil {
+			t.Fatalf("build image: %v", err)
+		}
+		if err := crane.Push(img, tagRef, authOpts...); err != nil {
+			t.Fatalf("push: %v", err)
+		}
+
+		// No credentials: the token endpoint must refuse a delete scope, so the
+		// DELETE fails and the image remains pullable by its owner.
+		if err := crane.Delete(tagRef, crane.Insecure); err == nil {
+			t.Error("expected anonymous delete to be denied, got nil")
+		}
+		if _, err := crane.Head(tagRef, authOpts...); err != nil {
+			t.Errorf("image should survive denied anonymous delete: %v", err)
+		}
+	})
+}