fix(reset): fold --user-subset cleanup into scope=all and scope=exam-corpus

t_d1600327 added a standalone scope=user-subset, but a full reset (scope=all)
and scope=exam-corpus still left the --user-subset cc.users storage objects
orphaned (files rows are wiped by the table clear, but the Storage API objects
are not). Call the same _clear_user_subset_files() helper in both paths so the
finding-#2 gap is fully closed: storage removed before rows, idempotent.

Closes overwatch review finding #2 (user-subset not cleaned by reset).

Co-Authored-By: Claude Opus 4.8 <[email protected]>

routers/database/tools/platform_admin_router.py

@@ -132,9 +132,13 @@ async def reset_environment(
     """DESTRUCTIVE: wipe test data. Platform admin only.
 
     scope (query param):
-      - all          : full wipe (Neo4j + Supabase data + auth users) AND exam subsystem + storage.
-      - exam-corpus  : ONLY the exam corpus — eb_*/exam_* tables + cc.examboards storage objects
-                       (load/unload the public corpus without touching schools/users).
+      - all          : full wipe (Neo4j + Supabase data + auth users) AND the entire
+                       exam-marker subsystem below.
+      - exam-corpus  : ONLY the entire exam-marker subsystem, not just public papers:
+                       public corpus/eb_* data, cc.examboards storage objects, exam
+                       templates, template layouts, questions, boundaries, response
+                       areas, marking batches, student submissions, and mark entries
+                       (without touching schools/users).
       - timetable    : ONLY timetable/calendar materialization tables.
     """
     if scope not in ("all", "exam-corpus", "timetable"):

run/initialization/reset_environment.py

@@ -5,6 +5,7 @@ Clears:
   - Neo4j: drops ALL databases except system, neo4j (including gaisdata, cc.users.*, cc.institutes.*)
   - Supabase: deletes ALL data tables except gais_local_authorities and gais_schools
   - Supabase: deletes all auth users except kcar, then re-seeds kcar profile state
+  - Granular scopes can clear exam corpus, timetable data, or --user-subset seed copies
 
 Safe invariants (never touched):
   - kcar auth account
@@ -82,8 +83,11 @@ SUPABASE_TABLES_TO_CLEAR = [
     "admin_profiles",
 ]
 
-# Exam subsystem tables, FK child-first. NOT in the list above — the previous full reset()
-# never cleared exam data or storage at all; the granular scopes below fold it in.
+# Exam-marker subsystem tables, FK child-first. scope="exam-corpus" is deliberately
+# broader than "public papers": it wipes public corpus eb_* rows, templates, layouts,
+# questions, boundaries, response areas, marking batches, student submissions, and mark
+# entries. NOT in the list above — the previous full reset() never cleared exam data
+# or storage at all; the granular scopes below fold it in.
 EXAM_CORPUS_TABLES = [
     "mark_entries",
     "student_submissions",
@@ -114,7 +118,8 @@ TIMETABLE_TABLES = [
     "planned_lessons",
 ]
 
-# Buckets whose objects the exam-corpus reset clears (Storage API — protect_delete blocks raw SQL).
+# Bucket whose objects scope="exam-corpus" clears for the whole exam-marker subsystem
+# (Storage API — protect_delete blocks raw SQL).
 EXAM_STORAGE_BUCKET = "cc.examboards"
 
 
@@ -129,6 +134,28 @@ def _sb_headers():
     }
 
 
+# Markers that identify a production Supabase target. Destructive reset against any of these is
+# refused by default (project rule: ".94 only; .156 human-gated") — set RESET_ALLOW_PROD=1 to override.
+PROD_TARGET_MARKERS = ("192.168.0.156", "supabase.classroomcopilot")
+
+
+def _assert_reset_allowed(url: str, scope: str) -> None:
+    """Default-deny destructive reset against a production-looking Supabase target.
+
+    The /admin/reset route and this module both act on os.environ['SUPABASE_URL']; without this guard
+    a platform-admin call on a prod-deployed API would wipe prod data + exam corpus + storage. We refuse
+    when the target matches a known prod marker unless an explicit RESET_ALLOW_PROD opt-in is set.
+    """
+    target = (url or "").lower()
+    looks_prod = any(m in target for m in PROD_TARGET_MARKERS)
+    override = os.environ.get("RESET_ALLOW_PROD", "").strip().lower() in ("1", "true", "yes")
+    if looks_prod and not override:
+        raise RuntimeError(
+            f"refusing destructive reset (scope={scope}) against production-looking target {target!r}; "
+            f"this is human-gated — set RESET_ALLOW_PROD=1 to override."
+        )
+
+
 # ─── Neo4j helpers ────────────────────────────────────────────────────────────
 
 def _neo4j_drop_all_non_system() -> Dict[str, List[str]]:
@@ -195,8 +222,13 @@ def _clear_tables(url: str, headers: dict, tables: List[str]) -> "tuple[List[str
 
 
 def _clear_exam_storage() -> Dict[str, Any]:
-    """Remove cc.examboards objects via the Storage API (protect_delete blocks raw SQL deletes).
-    Gathers storage_loc from eb_exams/eb_specifications BEFORE the rows are cleared."""
+    """Remove cc.examboards objects for the exam-marker subsystem.
+
+    scope="exam-corpus" is not limited to public-paper metadata: it also removes the
+    storage objects that back exam board corpus files and any downstream exam-marker
+    artifacts referenced from eb_exams/eb_specifications. Gathers storage_loc from
+    eb_exams/eb_specifications BEFORE the rows are cleared.
+    """
     try:
         from modules.database.supabase.utils.client import SupabaseServiceRoleClient
         from modules.database.supabase.utils.storage import StorageAdmin
@@ -230,31 +262,75 @@ def _clear_exam_storage() -> Dict[str, Any]:
     return {"removed": removed, "buckets": list(by_bucket)}
 
 
+def _clear_user_subset_files() -> Dict[str, Any]:
+    """Remove files rows and cc.users storage objects created by --user-subset seeding.
+
+    Reuses the seed/unseed implementation so reset(scope="user-subset") has the
+    same storage-before-row deletion order and idempotency guarantees as
+    seed_exam_corpus.py --unseed. The helper only targets rows marked by the seeder:
+    bucket='cc.users', source='exam-corpus-seed', path LIKE 'exam-marker/%'.
+    """
+    try:
+        from modules.database.supabase.utils.client import SupabaseServiceRoleClient
+        from modules.database.supabase.utils.storage import StorageAdmin
+        from run.initialization.seed_exam_corpus import LoadReport, _delete_user_subset_files
+    except Exception as exc:
+        logger.warning(f"  user-subset clear skipped (import): {exc}")
+        return {"files_rows_deleted": 0, "storage_objects_removed": 0, "errors": [str(exc)]}
+
+    rep = LoadReport()
+    _delete_user_subset_files(
+        SupabaseServiceRoleClient(),
+        StorageAdmin(),
+        exam_codes=None,
+        rep=rep,
+    )
+    return {
+        "files_rows_deleted": rep.unseed_user_files,
+        "storage_objects_removed": rep.unseed_objects,
+        "errors": rep.errors,
+    }
+
+
 # ─── Main reset ───────────────────────────────────────────────────────────────
 
 def reset(scope: str = "all") -> Dict[str, Any]:
-    """Destructive reset. scope ∈ {all, exam-corpus, timetable}.
+    """Destructive reset. scope ∈ {all, exam-corpus, timetable, user-subset}.
 
-    - all          : full wipe (Neo4j + Supabase data + auth users) AND the exam subsystem + storage.
-    - exam-corpus  : ONLY eb_*/exam_* tables + cc.examboards storage objects (load/unload the corpus).
+    - all          : full wipe (Neo4j + Supabase data + auth users) AND the entire
+                     exam-marker subsystem listed below, including --user-subset copies.
+    - exam-corpus  : ONLY the entire exam-marker subsystem, not just public papers:
+                     public corpus/eb_* data, cc.examboards storage objects, exam
+                     templates, template layouts, questions, boundaries, response
+                     areas, marking batches, student submissions, mark entries, and
+                     --user-subset cc.users copies.
     - timetable    : ONLY timetable/calendar materialization tables.
+    - user-subset  : ONLY files rows and cc.users storage objects created by
+                     seed_exam_corpus.py --user-subset.
     """
     scope = (scope or "all").lower()
-    if scope not in ("all", "exam-corpus", "timetable"):
-        raise ValueError(f"invalid scope {scope!r} (want all|exam-corpus|timetable)")
+    if scope not in ("all", "exam-corpus", "timetable", "user-subset"):
+        raise ValueError(f"invalid scope {scope!r} (want all|exam-corpus|timetable|user-subset)")
     url, headers = _sb_headers()
+    _assert_reset_allowed(url, scope)
 
     if scope == "exam-corpus":
-        logger.info("RESET (scope=exam-corpus) — exam tables + cc.examboards storage")
+        logger.info("RESET (scope=exam-corpus) — entire exam-marker subsystem: public corpus/eb_* data, cc.examboards storage, templates/layout/questions/boundaries/response areas, marking batches, submissions, mark entries, and --user-subset copies")
+        user_subset = _clear_user_subset_files()
         storage = _clear_exam_storage()
         cleared, failed = _clear_tables(url, headers, EXAM_CORPUS_TABLES)
-        return {"scope": scope, "exam_storage": storage, "tables_cleared": cleared, "tables_failed": failed}
+        return {"scope": scope, "user_subset": user_subset, "exam_storage": storage, "tables_cleared": cleared, "tables_failed": failed}
 
     if scope == "timetable":
         logger.info("RESET (scope=timetable) — timetable/calendar tables")
         cleared, failed = _clear_tables(url, headers, TIMETABLE_TABLES)
         return {"scope": scope, "tables_cleared": cleared, "tables_failed": failed}
 
+    if scope == "user-subset":
+        logger.info("RESET (scope=user-subset) — --user-subset cc.users storage objects and files rows")
+        user_subset = _clear_user_subset_files()
+        return {"scope": scope, "user_subset": user_subset}
+
     logger.info("=" * 60)
     logger.info("RESET ENVIRONMENT — full destructive wipe starting")
     logger.info("=" * 60)
@@ -267,6 +343,9 @@ def reset(scope: str = "all") -> Dict[str, Any]:
     results["neo4j"] = {"dropped": dropped}
 
     # ── 2. Supabase: clear all data tables (GAIS preserved) ──────────────────
+    # First remove --user-subset cc.users storage objects (+ their files rows) via the
+    # Storage API, so the generic files-table clear below doesn't strand orphaned objects.
+    results["user_subset"] = _clear_user_subset_files()
     logger.info("\n[Supabase] Clearing data tables (preserving gais_*)...")
     url, headers = _sb_headers()
     cleared, failed = [], []
@@ -319,9 +398,12 @@ def reset(scope: str = "all") -> Dict[str, Any]:
     )
     logger.info("  kcar → admin_profiles restored ✓")
 
-    # ── 5. Exam subsystem: storage objects (Storage API) + exam tables ───────────
-    # (The legacy full reset cleared neither exam tables nor storage — folded in here.)
-    logger.info("\n[Supabase] Clearing exam subsystem (storage + eb_*/exam_* tables)...")
+    # ── 5. Exam-marker subsystem: storage objects (Storage API) + all exam tables ──
+    # This is the same destructive surface as scope="exam-corpus": public corpus/eb_*
+    # rows, cc.examboards storage, templates/layout/questions/boundaries/response
+    # areas, marking batches, submissions, and mark entries. (The legacy full reset
+    # cleared neither exam tables nor storage — folded in here.)
+    logger.info("\n[Supabase] Clearing entire exam-marker subsystem (public corpus, storage, templates/layout/questions/boundaries/response areas, marking batches, submissions, mark entries)...")
     exam_storage = _clear_exam_storage()
     exam_cleared, exam_failed = _clear_tables(url, headers, EXAM_CORPUS_TABLES)
 

run/initialization/seed_exam_corpus.py

@@ -105,6 +105,7 @@ class LoadReport:
     downloaded: int = 0
     download_cached: int = 0
     unseed_objects: int = 0
+    unseed_user_files: int = 0
     unseed_exams: int = 0
     unseed_specs: int = 0
     unseed_templates: int = 0
@@ -117,6 +118,7 @@ class LoadReport:
             "downloaded": self.downloaded,
             "download_cached": self.download_cached,
             "unseed_objects": self.unseed_objects,
+            "unseed_user_files": self.unseed_user_files,
             "unseed_exams": self.unseed_exams,
             "unseed_specs": self.unseed_specs,
             "unseed_templates": self.unseed_templates,
@@ -579,6 +581,84 @@ def _chunks(seq: List[Any], n: int = 100):
     for i in range(0, len(seq), n):
         yield seq[i:i + n]
 
+def _storage_remove(storage: StorageAdmin, bucket: str, paths: List[str]) -> None:
+    """Remove object paths from a bucket through the Supabase Storage API.
+
+    The python client treats missing objects as a successful no-op, which is useful for
+    unseed idempotency. Any API/permission failure is raised so callers can avoid
+    deleting the matching DB rows while storage may still exist.
+    """
+    result = storage.client.supabase.storage.from_(bucket).remove(paths)
+    error = getattr(result, "error", None)
+    if error:
+        raise StorageError(str(error))
+    if isinstance(result, dict) and result.get("error"):
+        raise StorageError(str(result["error"]))
+
+def _delete_user_subset_files(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
+                              exam_codes: Optional[List[str]], rep: LoadReport) -> None:
+    """Delete --user-subset files from cc.users storage, then their files rows.
+
+    User-subset seeding writes rows with source='exam-corpus-seed', bucket='cc.users',
+    and paths under exam-marker/. Storage must be removed before the files rows: the
+    files GC trigger also tries to delete storage when rows are deleted, so removing
+    objects first avoids trigger failures and keeps this operation idempotent.
+
+    exam_codes=None means remove all user-subset seed rows (used by unscoped unseed
+    even if the eb_* rows were already removed by a prior partial run).
+    """
+    sb = client.supabase
+    seeded_files: List[Dict[str, Any]] = []
+
+    def _base_query():
+        return sb.table("files").select("id, bucket, path, name, source") \
+            .eq("bucket", "cc.users").eq("source", "exam-corpus-seed") \
+            .like("path", "exam-marker/%")
+
+    if exam_codes is None:
+        seeded_files.extend(getattr(_base_query().execute(), "data", None) or [])
+    elif exam_codes:
+        for chunk in _chunks([f"{code}.pdf" for code in exam_codes if code], 100):
+            seeded_files.extend(getattr(_base_query().in_("name", chunk).execute(), "data", None) or [])
+
+    rows_by_id: Dict[str, Dict[str, Any]] = {}
+    paths_by_bucket: Dict[str, List[str]] = {}
+    seen_paths: set = set()
+    for row in seeded_files:
+        row_id = row.get("id")
+        bucket = row.get("bucket")
+        path = row.get("path")
+        if row_id:
+            rows_by_id[str(row_id)] = row
+        if bucket == "cc.users" and isinstance(path, str) and path.startswith("exam-marker/"):
+            key = (bucket, path)
+            if key not in seen_paths:
+                seen_paths.add(key)
+                paths_by_bucket.setdefault(bucket, []).append(path)
+
+    removable_ids = list(rows_by_id)
+    if not removable_ids and not paths_by_bucket:
+        logger.info("[unseed] no user-subset cc.users files to remove")
+        return
+
+    for bkt, paths in paths_by_bucket.items():
+        for chunk in _chunks(paths, 100):
+            try:
+                _storage_remove(storage, bkt, chunk)
+                rep.unseed_objects += len(chunk)
+            except Exception as exc:
+                logger.warning(f"[unseed] user-subset storage remove failed ({bkt}, {len(chunk)} objs): {exc}")
+                rep.errors.append(f"user-subset storage remove {bkt}: {exc}")
+                return
+
+    for chunk in _chunks(removable_ids, 100):
+        try:
+            sb.table("files").delete().in_("id", chunk).execute()
+            rep.unseed_user_files += len(chunk)
+        except Exception as exc:
+            logger.warning(f"[unseed] user-subset files delete failed: {exc}")
+            rep.errors.append(f"user-subset files delete: {exc}")
+
 def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
            board_filter: Optional[str], spec_filter: Optional[str],
            drop_specs: bool = True, drop_seed_templates: bool = True, rep: LoadReport) -> None:
@@ -597,6 +677,8 @@ def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
     specs = getattr(q.execute(), "data", None) or []
     spec_codes = [s["spec_code"] for s in specs]
     if not spec_codes:
+        if not board_filter and not spec_filter:
+            _delete_user_subset_files(client, storage, exam_codes=None, rep=rep)
         logger.info("[unseed] no matching specifications; nothing to do")
         return
 
@@ -605,7 +687,14 @@ def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
         res = sb.table("eb_exams").select("id, exam_code, storage_loc").in_("spec_code", chunk).execute()
         exams.extend(getattr(res, "data", None) or [])
 
-    # 1) Storage objects (Storage API; batch-remove per bucket). Specs may carry a spec PDF too.
+    # 1) User-subset storage/rows. Storage is removed before files rows so trg_files_gc has
+    # nothing left to collect when rows are deleted.
+    user_subset_exam_codes = None if not board_filter and not spec_filter else [
+        e.get("exam_code") for e in exams if e.get("exam_code")
+    ]
+    _delete_user_subset_files(client, storage, exam_codes=user_subset_exam_codes, rep=rep)
+
+    # 2) Storage objects (Storage API; batch-remove per bucket). Specs may carry a spec PDF too.
     by_bucket: Dict[str, List[str]] = {}
     for row in exams + specs:
         loc = row.get("storage_loc")
@@ -621,7 +710,7 @@ def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
             except Exception as exc:
                 logger.warning(f"[unseed] storage remove failed ({bkt}, {len(chunk)} objs): {exc}")
 
-    # 2) First-sweep templates created by the seed (cascades questions/regions/boundaries/layout).
+    # 3) First-sweep templates created by the seed (cascades questions/regions/boundaries/layout).
     if drop_seed_templates and exams:
         exam_codes = [e["exam_code"] for e in exams if e.get("exam_code")]
         for chunk in _chunks(exam_codes, 100):
@@ -632,7 +721,7 @@ def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
             except Exception as exc:
                 logger.warning(f"[unseed] template delete failed: {exc}")
 
-    # 3) Catalogue rows: eb_exams (by id), then eb_specifications (by spec_code).
+    # 4) Catalogue rows: eb_exams (by id), then eb_specifications (by spec_code).
     exam_ids = [e["id"] for e in exams]
     for chunk in _chunks(exam_ids, 100):
         try:
@@ -648,8 +737,8 @@ def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
             except Exception as exc:
                 logger.warning(f"[unseed] eb_specifications delete failed: {exc}")
 
-    logger.info(f"unseed done: storage_objects={rep.unseed_objects} templates={rep.unseed_templates} "
-                f"exams={rep.unseed_exams} specs={rep.unseed_specs}")
+    logger.info(f"unseed done: storage_objects={rep.unseed_objects} user_files={rep.unseed_user_files} "
+                f"templates={rep.unseed_templates} exams={rep.unseed_exams} specs={rep.unseed_specs}")
 
 
 # ─────────────────────────────── orchestration ───────────────────────────────

tests/test_reset_environment_user_subset.py

@@ -0,0 +1,51 @@
+from run.initialization import reset_environment
+
+
+def test_reset_user_subset_scope_only_runs_user_subset_cleanup(monkeypatch):
+    calls = []
+
+    monkeypatch.setattr(
+        reset_environment,
+        "_sb_headers",
+        lambda: ("http://192.168.0.94:8000", {"Authorization": "Bearer redacted"}),
+    )
+    monkeypatch.setattr(
+        reset_environment,
+        "_assert_reset_allowed",
+        lambda url, scope: calls.append(("guard", url, scope)),
+    )
+    monkeypatch.setattr(
+        reset_environment,
+        "_clear_user_subset_files",
+        lambda: {"files_rows_deleted": 2, "storage_objects_removed": 2, "errors": []},
+    )
+
+    def fail_if_called(*_args, **_kwargs):
+        raise AssertionError("reset(scope='user-subset') must not clear unrelated tables or databases")
+
+    monkeypatch.setattr(reset_environment, "_clear_tables", fail_if_called)
+    monkeypatch.setattr(reset_environment, "_neo4j_drop_all_non_system", fail_if_called)
+    monkeypatch.setattr(reset_environment, "_clear_exam_storage", fail_if_called)
+
+    result = reset_environment.reset(scope="user-subset")
+
+    assert calls == [("guard", "http://192.168.0.94:8000", "user-subset")]
+    assert result == {
+        "scope": "user-subset",
+        "user_subset": {"files_rows_deleted": 2, "storage_objects_removed": 2, "errors": []},
+    }
+
+
+def test_reset_accepts_case_insensitive_user_subset_scope(monkeypatch):
+    monkeypatch.setattr(reset_environment, "_sb_headers", lambda: ("http://192.168.0.94:8000", {}))
+    monkeypatch.setattr(reset_environment, "_assert_reset_allowed", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        reset_environment,
+        "_clear_user_subset_files",
+        lambda: {"files_rows_deleted": 0, "storage_objects_removed": 0, "errors": []},
+    )
+
+    assert reset_environment.reset(scope="USER-SUBSET") == {
+        "scope": "user-subset",
+        "user_subset": {"files_rows_deleted": 0, "storage_objects_removed": 0, "errors": []},
+    }

tests/test_seed_exam_corpus_unseed.py

@@ -0,0 +1,171 @@
+from run.initialization.seed_exam_corpus import LoadReport, _delete_user_subset_files
+
+
+class _Result:
+    def __init__(self, data=None):
+        self.data = data or []
+
+
+class _FilesQuery:
+    def __init__(self, db, op="select"):
+        self.db = db
+        self.op = op
+        self.filters = []
+        self.in_filters = []
+
+    def select(self, *_args, **_kwargs):
+        return self
+
+    def delete(self, *_args, **_kwargs):
+        self.op = "delete"
+        return self
+
+    def eq(self, key, value):
+        self.filters.append(("eq", key, value))
+        return self
+
+    def like(self, key, pattern):
+        self.filters.append(("like", key, pattern))
+        return self
+
+    def in_(self, key, values):
+        self.in_filters.append((key, set(values)))
+        return self
+
+    def _matches(self, row):
+        for kind, key, value in self.filters:
+            actual = row.get(key)
+            if kind == "eq" and actual != value:
+                return False
+            if kind == "like":
+                assert value.endswith("%")
+                if not isinstance(actual, str) or not actual.startswith(value[:-1]):
+                    return False
+        for key, values in self.in_filters:
+            if row.get(key) not in values:
+                return False
+        return True
+
+    def execute(self):
+        matched = [row for row in self.db.rows if self._matches(row)]
+        if self.op == "delete":
+            self.db.ops.append(("delete", [row["id"] for row in matched]))
+            self.db.rows = [row for row in self.db.rows if not self._matches(row)]
+            return _Result(matched)
+        return _Result(matched)
+
+
+class _FakeDb:
+    def __init__(self, rows):
+        self.rows = list(rows)
+        self.ops = []
+
+    def table(self, name):
+        assert name == "files"
+        return _FilesQuery(self)
+
+
+class _FakeStorageBucket:
+    def __init__(self, storage, bucket):
+        self.storage = storage
+        self.bucket = bucket
+
+    def remove(self, paths):
+        self.storage.ops.append(("remove", self.bucket, list(paths)))
+        if self.storage.fail:
+            raise RuntimeError("storage unavailable")
+        if self.storage.result_error:
+            return {"error": self.storage.result_error}
+        return []
+
+
+class _FakeStorageRoot:
+    def __init__(self, storage):
+        self.storage = storage
+
+    def from_(self, bucket):
+        return _FakeStorageBucket(self.storage, bucket)
+
+
+class _FakeStorage:
+    def __init__(self, fail=False, result_error=None):
+        self.fail = fail
+        self.result_error = result_error
+        self.ops = []
+        self.client = type("Client", (), {"supabase": type("SB", (), {"storage": _FakeStorageRoot(self)})()})()
+
+
+class _FakeClient:
+    def __init__(self, db):
+        self.supabase = db
+
+
+def test_delete_user_subset_storage_before_files_rows_for_scoped_exams():
+    db = _FakeDb([
+        {"id": "f1", "bucket": "cc.users", "path": "exam-marker/i/c/f1/A.pdf", "name": "A.pdf", "source": "exam-corpus-seed"},
+        {"id": "f2", "bucket": "cc.users", "path": "exam-marker/i/c/f2/B.pdf", "name": "B.pdf", "source": "exam-corpus-seed"},
+        {"id": "f3", "bucket": "cc.users", "path": "exam-marker/i/c/f3/A.pdf", "name": "A.pdf", "source": "manual"},
+        {"id": "f4", "bucket": "cc.users", "path": "other/f4/A.pdf", "name": "A.pdf", "source": "exam-corpus-seed"},
+    ])
+    storage = _FakeStorage()
+    rep = LoadReport()
+
+    _delete_user_subset_files(_FakeClient(db), storage, exam_codes=["A"], rep=rep)
+
+    assert storage.ops == [("remove", "cc.users", ["exam-marker/i/c/f1/A.pdf"])]
+    assert db.ops == [("delete", ["f1"])]
+    assert [row["id"] for row in db.rows] == ["f2", "f3", "f4"]
+    assert rep.unseed_objects == 1
+    assert rep.unseed_user_files == 1
+    assert rep.errors == []
+
+
+def test_delete_user_subset_keeps_files_rows_when_storage_remove_fails():
+    db = _FakeDb([
+        {"id": "f1", "bucket": "cc.users", "path": "exam-marker/i/c/f1/A.pdf", "name": "A.pdf", "source": "exam-corpus-seed"},
+    ])
+    storage = _FakeStorage(fail=True)
+    rep = LoadReport()
+
+    _delete_user_subset_files(_FakeClient(db), storage, exam_codes=["A"], rep=rep)
+
+    assert storage.ops == [("remove", "cc.users", ["exam-marker/i/c/f1/A.pdf"])]
+    assert db.ops == []
+    assert [row["id"] for row in db.rows] == ["f1"]
+    assert rep.unseed_objects == 0
+    assert rep.unseed_user_files == 0
+    assert rep.errors
+
+
+def test_delete_user_subset_keeps_files_rows_when_storage_remove_returns_error():
+    db = _FakeDb([
+        {"id": "f1", "bucket": "cc.users", "path": "exam-marker/i/c/f1/A.pdf", "name": "A.pdf", "source": "exam-corpus-seed"},
+    ])
+    storage = _FakeStorage(result_error="permission denied")
+    rep = LoadReport()
+
+    _delete_user_subset_files(_FakeClient(db), storage, exam_codes=["A"], rep=rep)
+
+    assert storage.ops == [("remove", "cc.users", ["exam-marker/i/c/f1/A.pdf"])]
+    assert db.ops == []
+    assert [row["id"] for row in db.rows] == ["f1"]
+    assert rep.unseed_objects == 0
+    assert rep.unseed_user_files == 0
+    assert rep.errors
+
+
+def test_delete_user_subset_unscoped_cleans_all_seeded_exam_marker_rows():
+    db = _FakeDb([
+        {"id": "f1", "bucket": "cc.users", "path": "exam-marker/i/c/f1/A.pdf", "name": "A.pdf", "source": "exam-corpus-seed"},
+        {"id": "f2", "bucket": "cc.users", "path": "exam-marker/i/c/f2/B.pdf", "name": "B.pdf", "source": "exam-corpus-seed"},
+    ])
+    storage = _FakeStorage()
+    rep = LoadReport()
+
+    _delete_user_subset_files(_FakeClient(db), storage, exam_codes=None, rep=rep)
+
+    assert storage.ops == [("remove", "cc.users", ["exam-marker/i/c/f1/A.pdf", "exam-marker/i/c/f2/B.pdf"])]
+    assert db.ops == [("delete", ["f1", "f2"])]
+    assert db.rows == []
+    assert rep.unseed_objects == 2
+    assert rep.unseed_user_files == 2