feat: add exact owner-directed thread purges

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+## 0.15.0 - 2026-10-01
+
+**Highlights:** Exact owner-directed archive removal with durable collection exclusions.
+
+- Add `purge-threads` with exact owner-selected plans, atomic local removal, and durable collection exclusions. Thanks @hannesrudolph.
+- Update CrawlKit to v0.16.7. Thanks @vincentkoc.
+
 ## 0.14.0 - 2026-09-30
 
 **Highlights:** New `gitcrawl analytics` commands for publication repair, actor evidence, and continuous collection with durable review-state recovery.

docs/governance.md

@@ -137,3 +137,28 @@ The thread stays open on GitHub; only your local triage view hides it.
 - It does not edit, label, comment on, or close GitHub issues. Use `gh` for that.
 - It does not retrain embeddings or reshape the underlying graph — it overlays decisions on top of the algorithm output.
 - It does not propagate to other gitcrawl installations unless you publish your database via a [portable store](/portable-stores/).
+
+## Permanent owner removal
+
+For a native local archive, preview an exact selection, then apply its `plan_id`:
+
+```sh
+gitcrawl purge-threads owner/repo --numbers 123,456 --json
+gitcrawl purge-threads owner/repo --numbers 123,456 --apply PLAN_ID --json
+```
+
+The read-only plan lists thread identities and affected row counts. Apply rechecks
+its content hash in one transaction, removes selected bodies, history, derived
+records and retry work, and installs permanent collection exclusions. Changed
+plans require a new preview. Shared actor profiles and unrelated threads stay
+intact. Owner exclusions are reported separately from provider deletion or
+successful recovery; core collection coverage remains independent.
+
+Selections must contain 1–100 distinct existing numbers in one unambiguous local
+repository. The analytics collector must be stopped. Portable/cloud archives,
+blob-backed targets, shared cluster state, retained repository workflow snapshots,
+linked workflow reservations, and deletions allowing native integer ID reuse are
+refused. Exclusions follow the explicit `owner/repo` collection namespace; carry
+that policy when changing targets. Portable publication is refused while local
+exclusions exist. This removes logical archive data, not GitHub content, backups,
+or forensic disk remnants. There is no undo or automatic backup.

internal/cli/analytics_integrity.go

@@ -379,6 +379,11 @@ func analyticsReviewBudget(limits []gh.RateLimitSnapshot, now time.Time) (int, g
 }
 
 func (a *App) analyticsNumbers(ctx context.Context, s *store.Store, owner, repo string, numbers []int, discovery bool, operation string) (resultErr error) {
+	var err error
+	numbers, err = s.FilterExcludedNumbers(ctx, owner+"/"+repo, numbers)
+	if err != nil {
+		return err
+	}
 	var wg sync.WaitGroup
 	var failures []error
 	// Every return, including cancelled admission, waits for durable receipts.

internal/cli/app.go

@@ -141,6 +141,8 @@ func (a *App) Run(ctx context.Context, args []string) error {
 		return a.runThreads(ctx, rest[1:])
 	case "capture":
 		return a.runCapture(ctx, rest[1:])
+	case "purge-threads":
+		return a.runPurgeThreads(ctx, rest[1:])
 	case "close-thread":
 		return a.runCloseThread(ctx, rest[1:])
 	case "reopen-thread":

internal/cli/app_test.go

@@ -4344,10 +4344,10 @@ func TestDoctorJSONReportsCurrentSchemaDiagnosticsWithoutMutation(t *testing.T)
 	if got := schema["state"]; got != "current" {
 		t.Fatalf("db_schema.state = %#v, payload=%#v", got, schema)
 	}
-	if got := schema["current_version"]; got != float64(15) {
+	if got := schema["current_version"]; got != float64(16) {
 		t.Fatalf("db_schema.current_version = %#v, payload=%#v", got, schema)
 	}
-	if got := schema["supported_version"]; got != float64(15) {
+	if got := schema["supported_version"]; got != float64(16) {
 		t.Fatalf("db_schema.supported_version = %#v, payload=%#v", got, schema)
 	}
 	if got := schema["child_observation_reservations"]; got != true {
@@ -4609,7 +4609,7 @@ func TestDoctorJSONReportsLegacyPendingSchemaWithoutMutation(t *testing.T) {
 		t.Fatalf("pr_details.duplicate_path_files_supported = %#v, payload=%#v", got, prDetails)
 	}
 	pending := doctorStringList(t, schema, "pending_migrations")
-	if !doctorListContains(pending, "schema_version_3_to_15") ||
+	if !doctorListContains(pending, "schema_version_3_to_16") ||
 		!doctorListContains(pending, "pull_request_files_position_key") ||
 		!doctorListContains(pending, "thread_child_observation_reservations_table") {
 		t.Fatalf("pending_migrations = %#v", pending)

internal/cli/gh_search_test.go

@@ -291,8 +291,8 @@ func TestGHSearchSyncIfStaleMigratesFreshPortableRuntime(t *testing.T) {
 	if err := rt.Store.DB().QueryRowContext(ctx, `pragma user_version`).Scan(&schemaVersion); err != nil {
 		t.Fatalf("read runtime schema version: %v", err)
 	}
-	if schemaVersion != 15 {
-		t.Fatalf("runtime schema version = %d, want 15", schemaVersion)
+	if schemaVersion != 16 {
+		t.Fatalf("runtime schema version = %d, want 16", schemaVersion)
 	}
 	var tableName string
 	if err := rt.Store.DB().QueryRowContext(ctx, `select name from sqlite_schema where type = 'table' and name = 'sync_runs'`).Scan(&tableName); err != nil {

internal/cli/help.go

@@ -65,6 +65,7 @@ Core commands:
   capture              export a stable code-free conversation snapshot
   code index           index tracked text files from a local Git checkout
   cluster              build durable clusters from local thread vectors
+  purge-threads        permanently remove selected local threads and exclude collection
   close-thread         locally hide one issue or pull request row
   reopen-thread        clear a local hide for one issue or pull request row
   close-cluster        locally hide one durable cluster
@@ -264,6 +265,17 @@ Usage:
 Usage:
   gitcrawl runs owner/repo [--kind sync|summary|embedding|cluster] [--limit N] [--json]
 `,
+	"purge-threads": `gitcrawl purge-threads plans owner-directed local removal, never a GitHub deletion.
+
+Usage:
+  gitcrawl purge-threads owner/repo --numbers 1,2 [--apply PLAN_ID] [--json]
+
+Preview the exact selection, then pass its plan_id to --apply. Changed plans
+are refused. Requires a native local archive and an idle analytics collector.
+Blob-backed targets, workflow dependencies, shared clusters and reusable native
+IDs are refused. Exclusions are permanent; portable publication is unavailable.
+`,
+
 	"close-thread": `gitcrawl close-thread locally hides one issue or pull request row.
 
 Usage:

internal/cli/thread_purge.go

@@ -0,0 +1,96 @@
+package cli
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"github.com/openclaw/gitcrawl/internal/config"
+	"github.com/openclaw/gitcrawl/internal/store"
+)
+
+func (a *App) runPurgeThreads(ctx context.Context, args []string) error {
+	fs := flag.NewFlagSet("purge-threads", flag.ContinueOnError)
+	fs.SetOutput(io.Discard)
+	raw := fs.String("numbers", "", "explicit issue/PR numbers")
+	apply := fs.String("apply", "", "apply the exact previewed plan ID")
+	jsonOut := fs.Bool("json", false, "JSON plan/result")
+	if err := fs.Parse(normalizeCommandArgs(args, map[string]bool{"numbers": true, "apply": true})); err != nil {
+		return usageErr(err)
+	}
+	if fs.NArg() != 1 {
+		return usageErr(fmt.Errorf("purge-threads requires owner/repo"))
+	}
+	owner, repo, err := parseOwnerRepo(fs.Arg(0))
+	if err != nil {
+		return usageErr(err)
+	}
+	repository := owner + "/" + repo
+	numbers, err := parseOptionalThreadNumberList(*raw, repository)
+	if err != nil {
+		return usageErr(err)
+	}
+	if len(numbers) == 0 || len(numbers) > 100 {
+		return usageErr(fmt.Errorf("purge-threads requires 1..100 explicit --numbers"))
+	}
+	if flagWasSet(fs, "apply") && len(*apply) != 64 {
+		return usageErr(fmt.Errorf("--apply requires the plan_id from a preview"))
+	}
+	a.applyCommandJSON(*jsonOut)
+	cfg, err := config.LoadRuntime(a.configPath)
+	if err != nil {
+		return err
+	}
+	if cfg.Remote.Enabled() {
+		return fmt.Errorf("purge-threads requires a native local archive")
+	}
+	_, portable, err := portableStoreRoot(ctx, cfg.DBPath)
+	if err != nil {
+		return err
+	}
+	if portable {
+		return fmt.Errorf("purge-threads refuses portable stores and mirrors")
+	}
+	// Lock before opening a writer: an analytics collector owns this same file.
+	if *apply != "" {
+		lock, err := os.OpenFile(filepath.Join(filepath.Dir(cfg.DBPath), "runner.lock"), os.O_CREATE|os.O_RDWR, 0600)
+		if err != nil {
+			return err
+		}
+		defer lock.Close()
+		if err = lockPortableFile(lock); err != nil {
+			return fmt.Errorf("collector ownership lock busy: %w", err)
+		}
+	}
+	probe, err := store.OpenReadOnly(ctx, cfg.DBPath)
+	if err != nil {
+		return err
+	}
+	plan, err := probe.PlanThreadPurge(ctx, repository, numbers)
+	closeErr := probe.Close()
+	if err != nil {
+		return err
+	}
+	if closeErr != nil {
+		return closeErr
+	}
+	if *apply == "" {
+		return a.writeOutput("thread_purge_plan", plan, true)
+	}
+	if *apply != plan.PlanID {
+		return fmt.Errorf("purge plan changed; preview again before applying")
+	}
+	st, err := store.Open(ctx, cfg.DBPath)
+	if err != nil {
+		return err
+	}
+	defer st.Close()
+	result, err := st.PurgeThreads(ctx, repository, numbers, *apply)
+	if err != nil {
+		return err
+	}
+	return a.writeOutput("thread_purge", result, true)
+}

internal/cli/thread_purge_test.go

@@ -0,0 +1,93 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/openclaw/gitcrawl/internal/store"
+)
+
+func TestThreadPurgeCLIPlansThenRequiresIdleOwner(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+	db := filepath.Join(dir, "archive.db")
+	cfg := filepath.Join(dir, "config.toml")
+	if err := os.WriteFile(cfg, []byte(fmt.Sprintf("db_path=%q\n", db)), 0600); err != nil {
+		t.Fatal(err)
+	}
+	s, err := store.Open(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = s.DB().Exec(`INSERT INTO repositories(id,owner,name,full_name,raw_json,updated_at) VALUES(1,'fixture','repo','fixture/repo','{}','2026-01-01');
+ INSERT INTO threads(id,repo_id,github_id,number,kind,state,title,html_url,labels_json,assignees_json,raw_json,content_hash,updated_at) VALUES
+ (1,1,'one',10,'pull_request','open','PRIVATE_SENTINEL','','[]','[]','{}','h','2026-01-01'),
+ (2,1,'two',20,'pull_request','open','keeper','','[]','[]','{}','h','2026-01-01')`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.Close()
+	run := func(args ...string) (string, error) {
+		a := New()
+		var out, stderr bytes.Buffer
+		a.Stdout = &out
+		a.Stderr = &stderr
+		err := a.Run(ctx, append([]string{"--config", cfg, "purge-threads", "fixture/repo", "--numbers", "10", "--json"}, args...))
+		return out.String(), err
+	}
+	out, err := run()
+	if err != nil || !strings.Contains(out, `"applied": false`) || strings.Contains(out, "PRIVATE_SENTINEL") {
+		t.Fatalf("plan=%s err=%v", out, err)
+	}
+	var plan store.ThreadPurgePlan
+	if err := json.Unmarshal([]byte(out), &plan); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = run("--apply"); err == nil {
+		t.Fatal("missing owner request accepted")
+	}
+	for _, args := range [][]string{{"--apply", ""}, {"--apply", strings.Repeat("0", 64)}, {"--numbers", ""}, {"--numbers", "10,10"}, {"--numbers", "other/repo#10"}, {"--numbers", "10,"}, {"--numbers", "999"}} {
+		if _, err := run(args...); err == nil {
+			t.Fatal("unsafe arguments accepted", args)
+		}
+	}
+	lock, err := os.OpenFile(filepath.Join(dir, "runner.lock"), os.O_CREATE|os.O_RDWR, 0600)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = lockPortableFile(lock); err != nil {
+		t.Fatal(err)
+	}
+	_, err = run("--apply", plan.PlanID)
+	if err == nil || !strings.Contains(err.Error(), "ownership lock busy") {
+		t.Fatalf("owner bypass: %v", err)
+	}
+	lock.Close()
+	out, err = run("--apply", plan.PlanID)
+	if err != nil || !strings.Contains(out, `"applied": true`) {
+		t.Fatalf("apply=%s %v", out, err)
+	}
+	out, err = run()
+	if err == nil {
+		t.Fatalf("repeat plan=%s %v", out, err)
+	}
+}
+
+func TestThreadPurgeRefusesRemoteWithoutAccess(t *testing.T) {
+	cfg := filepath.Join(t.TempDir(), "config.toml")
+	if err := os.WriteFile(cfg, []byte(fmt.Sprintf("db_path=%q\n[remote]\nmode=\"cloud\"\nendpoint=\"https://example.invalid\"\narchive=\"fixture\"\n", filepath.Join(filepath.Dir(cfg), "synthetic.db"))), 0600); err != nil {
+		t.Fatal(err)
+	}
+	a := New()
+	a.Stdout = &bytes.Buffer{}
+	a.Stderr = &bytes.Buffer{}
+	if err := a.Run(context.Background(), []string{"--config", cfg, "purge-threads", "fixture/repo", "--numbers", "1"}); err == nil || !strings.Contains(err.Error(), "native local") {
+		t.Fatal("remote target admitted", err)
+	}
+}

internal/store/analytics_integrity.go

@@ -57,6 +57,13 @@ func (s *Store) RecordAnalyticsAttempt(ctx context.Context, a AnalyticsAttempt)
 		return err
 	}
 	return s.WithTx(ctx, func(tx *Store) error {
+		excluded, err := tx.ThreadExcluded(ctx, a.Repository, a.Number)
+		if err != nil {
+			return err
+		}
+		if excluded {
+			return nil
+		}
 		status, class, message := a.Status, a.ErrorClass, a.ErrorText
 		knownReview := true
 		if status == "success" && (a.Operation == "graphql_history" || a.Operation == "review_state") {
@@ -359,12 +366,23 @@ func (s *Store) SaveReviewStateCoverage(ctx context.Context, repository string,
 	if err := s.q().QueryRowContext(ctx, `SELECT count(*) FROM analytics_retries WHERE repository=? AND operation='review_state' AND resolved_at IS NULL`, repository).Scan(&pending); err != nil {
 		return err
 	}
-	_, err := s.q().ExecContext(ctx, `INSERT INTO analytics_review_state_coverage VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(repository) DO UPDATE SET cursor=excluded.cursor,ceiling=excluded.ceiling,scanned=excluded.scanned,queued=excluded.queued,pending_items=excluded.pending_items,scan_complete=excluded.scan_complete,complete=excluded.complete,observed_at=excluded.observed_at`, repository, progress.Cursor, progress.Ceiling, progress.Scanned, progress.Queued, pending, boolInt(progress.Done), boolInt(progress.Done && pending == 0), time.Now().UTC().Format(time.RFC3339Nano))
+	excluded, err := s.OwnerExcludedCount(ctx, repository, true)
+	if err != nil {
+		return err
+	}
+	_, err = s.q().ExecContext(ctx, `INSERT INTO analytics_review_state_coverage(repository,cursor,ceiling,scanned,queued,pending_items,scan_complete,complete,observed_at,owner_excluded_items) VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(repository) DO UPDATE SET cursor=excluded.cursor,ceiling=excluded.ceiling,scanned=excluded.scanned,queued=excluded.queued,pending_items=excluded.pending_items,scan_complete=excluded.scan_complete,complete=excluded.complete,observed_at=excluded.observed_at,owner_excluded_items=excluded.owner_excluded_items`, repository, progress.Cursor, progress.Ceiling, progress.Scanned, progress.Queued, pending, boolInt(progress.Done), boolInt(progress.Done && pending == 0 && excluded == 0), time.Now().UTC().Format(time.RFC3339Nano), excluded)
 	return err
 }
 
 func (s *Store) AnalyticsIntegrityStatus(ctx context.Context, repository string) (map[string]any, error) {
 	out := map[string]any{"repository": repository, "checked_at": time.Now().UTC().Format(time.RFC3339Nano)}
+	if s.hasTable(ctx, "thread_exclusions") {
+		excluded, err := s.OwnerExcludedCount(ctx, repository, false)
+		if err != nil {
+			return nil, err
+		}
+		out["owner_excluded_items"] = excluded
+	}
 	var through, observed string
 	var issues, prs, complete int
 	coverageErr := s.q().QueryRowContext(ctx, "SELECT through,issues,pull_requests,complete,observed_at FROM analytics_coverage WHERE repository=?", repository).Scan(&through, &issues, &prs, &complete, &observed)

internal/store/analytics_integrity_test.go

@@ -152,7 +152,7 @@ func TestAnalyticsV14MigrationPreservesReceiptsAndCoverageWatermark(t *testing.T
 	s.DB().QueryRow("PRAGMA user_version").Scan(&version)
 	s.DB().QueryRow("SELECT through,complete FROM analytics_coverage WHERE repository='fixture/repo'").Scan(&through, &complete)
 	cp, err := s.AnalyticsState(ctx, "updates:fixture/repo")
-	if err != nil || version != 15 || through != "2026-01-01T00:00:00Z" || complete != 1 || cp != `{"kind":1,"cursor":"provider-cursor"}` {
+	if err != nil || version != schemaVersion || through != "2026-01-01T00:00:00Z" || complete != 1 || cp != `{"kind":1,"cursor":"provider-cursor"}` {
 		t.Fatalf("migration changed evidence: %d %s %d %s %v", version, through, complete, cp, err)
 	}
 }

internal/store/analytics_source.go

@@ -93,6 +93,11 @@ func (s *Store) queueAnalyticsActor(ctx context.Context, raw string) error {
 	if id == "" {
 		return nil
 	}
+	if excluded, err := s.NodeExcluded(ctx, id); err != nil {
+		return err
+	} else if excluded {
+		return nil
+	}
 	_, err := s.q().ExecContext(ctx, "INSERT OR IGNORE INTO analytics_pending_nodes(node_id,kind) VALUES(?,?)", id, kind)
 	return err
 }
@@ -236,6 +241,11 @@ func (s *Store) SaveActorEvidence(ctx context.Context, nodes []map[string]any, a
 			if id == "" {
 				continue
 			}
+			if excluded, err := tx.NodeExcluded(ctx, id); err != nil {
+				return err
+			} else if excluded {
+				continue
+			}
 			a, _ := n["author"].(map[string]any)
 			actor, _ := a["id"].(string)
 			if actor != "" {