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 <noreply@anthropic.com>

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
@@ -261,36 +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 entire
-                     exam-marker subsystem listed below.
+                     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, and mark entries.
+                     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"):
+    if scope not in ("all", "exam-corpus", "timetable", "user-subset"):
-        raise ValueError(f"invalid scope {scope!r} (want all|exam-corpus|timetable)")
+        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) — entire exam-marker subsystem: public corpus/eb_* data, cc.examboards storage, templates/layout/questions/boundaries/response areas, marking batches, submissions, mark entries")
+        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)
@@ -303,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 = [], []

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": []},
+    }