From b09c0f11bef27bf8e1b065389651f0b41c0f8a06 Mon Sep 17 00:00:00 2001
From: Hannes Rudolph <49103247+hannesrudolph@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:31:15 -0600
Subject: [PATCH 1/5] feat(clusters): record run-scoped computed candidate
 generations

Every accepted native cluster run now records its complete computed
partition as an immutable gitcrawl.cluster-candidates.v3 generation in the
same transaction, with each candidate's members, v2 durable status and
observed overlaps with active durable groups, and selects it. Durable IDs
and the six history tables are untouched and the quarantine postimage is
rechecked after the write, so quarantined candidates become visible in an
explicit run-scoped domain without asserting identity equivalence.

Generations are immutable except for owner removal. Rollback moves the
per-repository selection to the previous generation. Portable exports drop
the new tables. Add cluster-candidates to the CLI and the native reader.
---
 CHANGELOG.md                               |   1 +
 docs/clustering.md                         |  38 ++
 docs/commands.md                           |   1 +
 internal/cli/app.go                        |   2 +
 internal/cli/cluster_candidates.go         | 104 +++++
 internal/cli/cluster_candidates_test.go    | 122 +++++
 internal/cli/cluster_reader.go             |   6 +-
 internal/cli/help.go                       |  14 +
 internal/store/cluster_generations.go      | 512 +++++++++++++++++++++
 internal/store/cluster_generations_test.go | 246 ++++++++++
 internal/store/portable_schema.go          |  10 +
 internal/store/store.go                    |   3 +
 internal/store/thread_exclusions.go        |   8 +
 scripts/native-cluster/README.md           |  14 +
 scripts/native-cluster/native_cluster.go   |  13 +
 scripts/native-cluster/quarantine_test.go  |  20 +
 16 files changed, 1112 insertions(+), 2 deletions(-)
 create mode 100644 internal/cli/cluster_candidates.go
 create mode 100644 internal/cli/cluster_candidates_test.go
 create mode 100644 internal/store/cluster_generations.go
 create mode 100644 internal/store/cluster_generations_test.go

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9e9d4ad1..d27fca49 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@
 - Request `synchronous=FULL` on every SQLite connection and, on macOS, `fullfsync` for commits and checkpoints. Go test binaries skip the macOS full flush.
 - Let the native code producer publish from a captured upstream commit when core progress is fresh, even while provider coverage is incomplete. Retries and coverage are not changed.
 - Add `portable export --compression split`, which publishes the uncompressed portable database as verified 32 MiB parts so Git stores each new generation as a small delta instead of a new 68 MB gzip blob. Refresh, init, status and recovery read split stores; older readers refuse them and keep their last good mirror. After advancing to a split generation, strict refresh geometrically repacks recent objects so subscribers also keep each generation as a delta, skipping a loose backlog of more than 512 objects. Raw and gzip stores are still never repacked.
+- Record every accepted native cluster run's complete computed partition as an immutable, run-scoped `gitcrawl.cluster-candidates.v3` generation, with observed overlaps to durable groups, and select it atomically. Quarantined candidates are visible in this domain without changing durable IDs or their history. Add `cluster-candidates` (also in the native reader) and `--rollback`.
 
 ## 0.16.0 - 2026-10-20
 
diff --git a/docs/clustering.md b/docs/clustering.md
index a8afe12b..df7b57d4 100644
--- a/docs/clustering.md
+++ b/docs/clustering.md
@@ -206,6 +206,44 @@ Durable cluster IDs are derived from the representative thread, so they survive
 
 Cluster runs are recorded in `run_records` and visible via `gitcrawl runs --kind cluster`.
 
+## Computed candidate generations
+
+The native available-subset producer keeps durable IDs stable by quarantining
+any computed candidate whose identity is ambiguous: merges, splits and
+protected targets. Those threads keep their retained durable membership, but
+the current computation would otherwise be invisible. Every accepted native run
+now also records its complete computed partition as an immutable
+`gitcrawl.cluster-candidates.v3` generation, and selects it, in the same
+transaction:
+
+- Candidate IDs are `GENERATION:ORDINAL`. They are scoped to that run and are
+  not claimed stable across topology changes. Durable IDs remain the stable,
+  curated domain.
+- `observed_overlaps` lists durable groups that share active members with a
+  candidate. These are lineage observations, not aliases or equivalence. The
+  six durable history tables (`cluster_groups`, `cluster_memberships`,
+  `cluster_overrides`, `cluster_closures`, `cluster_aliases`,
+  `cluster_events`) are never written by generations.
+- `durable_status` says whether the v2 run published the candidate under a
+  durable ID (`published`, with `durable_cluster_id`) or quarantined it
+  (`identity_quarantined`). Both are present in the computed generation.
+- Generation rows cannot be updated or deleted, with one exception: owner
+  removal (`purge`) deletes a removed thread's member rows, and readers derive
+  counts and representatives from the remaining members.
+- `cluster-candidates --rollback` selects the previous generation. The next
+  accepted run selects its own generation again. A ref no-op run records
+  nothing.
+
+```bash
+gitcrawl cluster-candidates openclaw/openclaw --limit 20 --json
+gitcrawl cluster-candidates openclaw/openclaw --generations --json
+```
+
+The dedicated native cluster reader also permits `cluster-candidates`, but not
+`--rollback`. Strict v2 scope metadata is unchanged; consumers that need
+current computed clusters read this explicit, versioned domain rather than
+reinterpreting v2 scope fields.
+
 ## See also
 
 - [Governance](/governance/) — close clusters, exclude members, set canonical
diff --git a/docs/commands.md b/docs/commands.md
index b5acc14e..2588ab68 100644
--- a/docs/commands.md
+++ b/docs/commands.md
@@ -115,6 +115,7 @@ This applies to `sync --numbers`, `threads --numbers`, `summarize --number`, `em
 | `gitcrawl durable-clusters owner/repo [--include-closed --sort --min-size --limit --json]` | Strict durable-cluster audit view | [Clustering](/clustering/#list-clusters) |
 | `gitcrawl cluster-detail owner/repo --id <n> [--source auto\|run\|durable --member-limit --body-chars --hide-closed --json]` | Cluster + members detail | [Clustering](/clustering/#inspect-a-cluster) |
 | `gitcrawl cluster-explain owner/repo --id <n> [...]` | Alias for `cluster-detail` | [Clustering](/clustering/#inspect-a-cluster) |
+| `gitcrawl cluster-candidates owner/repo [--generation N --generations --limit N --members --rollback --json]` | Run-scoped computed candidate generations | [Clustering](/clustering/#computed-candidate-generations) |
 
 ## Governance
 
diff --git a/internal/cli/app.go b/internal/cli/app.go
index 89cbefea..14b49ead 100644
--- a/internal/cli/app.go
+++ b/internal/cli/app.go
@@ -187,6 +187,8 @@ func (a *App) Run(ctx context.Context, args []string) error {
 		return a.runClusterDetail(ctx, rest[1:])
 	case "cluster-explain":
 		return a.runClusterDetail(ctx, rest[1:])
+	case "cluster-candidates":
+		return a.runClusterCandidates(ctx, rest[1:], false)
 	case "neighbors":
 		return a.runNeighbors(ctx, rest[1:])
 	case "cluster":
diff --git a/internal/cli/cluster_candidates.go b/internal/cli/cluster_candidates.go
new file mode 100644
index 00000000..e6ea46a0
--- /dev/null
+++ b/internal/cli/cluster_candidates.go
@@ -0,0 +1,104 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+
+	"github.com/openclaw/gitcrawl/internal/store"
+)
+
+// runClusterCandidates reads the selected (or an explicit) computed candidate
+// generation. Candidate IDs are run-scoped and never durable identities.
+// Rollback is a local owner action and is unavailable in the cluster reader.
+func (a *App) runClusterCandidates(ctx context.Context, args []string, readerOnly bool) error {
+	fs := flag.NewFlagSet("cluster-candidates", flag.ContinueOnError)
+	fs.SetOutput(io.Discard)
+	generationRaw := fs.String("generation", "", "generation id (default: selected)")
+	limitRaw := fs.String("limit", "", "maximum candidates")
+	members := fs.Bool("members", false, "include member thread ids")
+	listGenerations := fs.Bool("generations", false, "list recent generations instead of candidates")
+	rollback := fs.Bool("rollback", false, "select the previous generation")
+	jsonOut := fs.Bool("json", false, "write JSON output")
+	if err := fs.Parse(normalizeCommandArgs(args, map[string]bool{"generation": true, "limit": true})); err != nil {
+		return usageErr(err)
+	}
+	a.applyCommandJSON(*jsonOut)
+	if fs.NArg() != 1 {
+		return usageErr(fmt.Errorf("cluster-candidates requires owner/repo"))
+	}
+	owner, repoName, err := parseOwnerRepo(fs.Arg(0))
+	if err != nil {
+		return usageErr(err)
+	}
+	generationID, err := parseOptionalPositiveInt(*generationRaw)
+	if err != nil {
+		return usageErr(err)
+	}
+	limit, err := parseOptionalPositiveInt(*limitRaw)
+	if err != nil {
+		return usageErr(err)
+	}
+	if *rollback && (readerOnly || generationID != 0 || *listGenerations) {
+		return usageErr(fmt.Errorf("--rollback is a local owner action without --generation or --generations"))
+	}
+	if *rollback {
+		rt, err := a.openLocalRuntime(ctx)
+		if err != nil {
+			return err
+		}
+		defer rt.Store.Close()
+		repo, err := rt.repository(ctx, owner, repoName)
+		if err != nil {
+			return err
+		}
+		generation, err := rt.Store.RollbackClusterCandidateGeneration(ctx, repo.ID)
+		if err != nil {
+			return err
+		}
+		return a.writeOutput("cluster-candidates", map[string]any{"repository": repo.FullName, "selected_generation": generation}, false)
+	}
+	rt, err := a.openLocalRuntimeReadOnly(ctx)
+	if err != nil {
+		return err
+	}
+	defer rt.Store.Close()
+	repo, err := rt.repository(ctx, owner, repoName)
+	if err != nil {
+		return err
+	}
+	if *listGenerations {
+		generations, err := rt.Store.ListClusterCandidateGenerations(ctx, repo.ID, limit)
+		if err != nil {
+			return err
+		}
+		return a.writeOutput("cluster-candidates", map[string]any{"repository": repo.FullName, "format": store.ClusterCandidateFormatV3, "generations": generations}, false)
+	}
+	var generation store.ClusterCandidateGeneration
+	if generationID != 0 {
+		generation, err = rt.Store.ClusterCandidateGenerationByID(ctx, repo.ID, int64(generationID))
+	} else {
+		generation, err = rt.Store.SelectedClusterCandidateGeneration(ctx, repo.ID)
+	}
+	if errors.Is(err, store.ErrNoClusterCandidateGeneration) {
+		return fmt.Errorf("%w for %s", err, repo.FullName)
+	}
+	if err != nil {
+		return err
+	}
+	candidates, err := rt.Store.ClusterCandidates(ctx, repo.ID, generation.ID, limit, *members)
+	if err != nil {
+		return err
+	}
+	return a.writeOutput("cluster-candidates", map[string]any{
+		"repository": repo.FullName,
+		"generation": generation,
+		// Computed IDs are scoped to this generation. Overlaps are observed
+		// shared members with durable groups, not equivalence.
+		"identity_domain":                  "computed-run-scoped",
+		"identity_reconciliation_complete": false,
+		"candidates":                       candidates,
+	}, false)
+}
diff --git a/internal/cli/cluster_candidates_test.go b/internal/cli/cluster_candidates_test.go
new file mode 100644
index 00000000..b5406399
--- /dev/null
+++ b/internal/cli/cluster_candidates_test.go
@@ -0,0 +1,122 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/openclaw/gitcrawl/internal/store"
+)
+
+func TestClusterCandidatesCommandReadsSelectsAndRollsBack(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+	configPath := filepath.Join(dir, "config.toml")
+	dbPath := filepath.Join(dir, "gitcrawl.db")
+	initApp := New()
+	initApp.Stdout, initApp.Stderr = &bytes.Buffer{}, &bytes.Buffer{}
+	if err := initApp.Run(ctx, []string{"--config", configPath, "init", "--db", dbPath, "--json"}); err != nil {
+		t.Fatal(err)
+	}
+	st, err := store.Open(ctx, dbPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now().UTC().Format(time.RFC3339Nano)
+	repoID, err := st.UpsertRepository(ctx, store.Repository{Owner: "openclaw", Name: "openclaw", FullName: "openclaw/openclaw", RawJSON: "{}", UpdatedAt: now})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var ids []int64
+	for i := 1; i <= 3; i++ {
+		id, err := st.UpsertThread(ctx, store.Thread{RepoID: repoID, GitHubID: fmt.Sprint(i), Number: i, Kind: "issue", State: "open", Title: "t", HTMLURL: "https://github.com/openclaw/openclaw/issues/1", LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: fmt.Sprint(i), UpdatedAt: now})
+		if err != nil {
+			t.Fatal(err)
+		}
+		ids = append(ids, id)
+	}
+	for generation := 0; generation < 2; generation++ {
+		plan := store.ClusterQuarantinePlan{RepositoryID: repoID, SelectedIDs: ids,
+			Candidates: []store.DurableClusterInput{
+				{StableKey: "a", RepresentativeThreadID: ids[0], Members: []store.DurableClusterMemberInput{{ThreadID: ids[0]}, {ThreadID: ids[1]}}},
+				{StableKey: "b", RepresentativeThreadID: ids[2], Members: []store.DurableClusterMemberInput{{ThreadID: ids[2]}}}},
+			Identities: []store.ClusterIdentityMetadata{{ID: 7, Active: true, ActiveMembers: []int64{ids[0]}}},
+			Targets:    []store.ClusterPublicationTarget{{CandidateKey: "b", ClusterID: 7}}}
+		result, err := st.DB().ExecContext(ctx, "insert into cluster_runs(repo_id, scope, status, started_at) values(?, 'durable', 'success', ?)", repoID, now)
+		if err != nil {
+			t.Fatal(err)
+		}
+		runID, err := result.LastInsertId()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := st.WithTx(ctx, func(tx *store.Store) error {
+			_, err := tx.SaveClusterCandidateGeneration(ctx, store.ClusterCandidateGenerationInput{RepoID: repoID, RunID: runID, Algorithm: "test",
+				InputSHA256: strings.Repeat(fmt.Sprint(generation), 64), QuarantinePlanSHA256: strings.Repeat("b", 64), Plan: plan})
+			return err
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := st.Close(); err != nil {
+		t.Fatal(err)
+	}
+	run := func(reader bool, args ...string) (map[string]any, error) {
+		app := New()
+		var stdout bytes.Buffer
+		app.Stdout, app.Stderr = &stdout, &bytes.Buffer{}
+		full := append([]string{"--config", configPath, "--json", "cluster-candidates", "openclaw/openclaw"}, args...)
+		if reader {
+			err = app.RunClusterReader(ctx, full)
+		} else {
+			err = app.Run(ctx, full)
+		}
+		if err != nil {
+			return nil, err
+		}
+		var payload map[string]any
+		if err := json.Unmarshal(stdout.Bytes(), &payload); err != nil {
+			t.Fatalf("decode %s: %v", stdout.String(), err)
+		}
+		return payload, nil
+	}
+	current, err := run(true, "--members")
+	if err != nil {
+		t.Fatal(err)
+	}
+	generation := current["generation"].(map[string]any)
+	candidates := current["candidates"].([]any)
+	if generation["generation_id"].(float64) != 2 || current["identity_domain"] != "computed-run-scoped" || current["identity_reconciliation_complete"] != false || len(candidates) != 2 {
+		t.Fatalf("selected generation output: %+v", current)
+	}
+	first := candidates[0].(map[string]any)
+	if first["computed_candidate_id"] != "2:1" || first["durable_status"] != "identity_quarantined" || len(first["observed_overlaps"].([]any)) != 1 || len(first["members"].([]any)) != 2 {
+		t.Fatalf("candidate output: %+v", first)
+	}
+	if _, err := run(true, "--rollback"); err == nil {
+		t.Fatal("cluster reader allowed rollback")
+	}
+	rolled, err := run(false, "--rollback")
+	if err != nil || rolled["selected_generation"].(map[string]any)["generation_id"].(float64) != 1 {
+		t.Fatalf("rollback output=%+v err=%v", rolled, err)
+	}
+	listed, err := run(false, "--generations")
+	if err != nil || len(listed["generations"].([]any)) != 2 || listed["generations"].([]any)[1].(map[string]any)["selected"] != true {
+		t.Fatalf("generation list=%+v err=%v", listed, err)
+	}
+	explicit, err := run(true, "--generation", "2", "--limit", "1")
+	if err != nil || len(explicit["candidates"].([]any)) != 1 || explicit["generation"].(map[string]any)["selected"] != false {
+		t.Fatalf("explicit generation=%+v err=%v", explicit, err)
+	}
+	if _, err := run(true, "--generation", "9"); err == nil {
+		t.Fatal("unknown generation accepted")
+	}
+	if _, err := run(false, "--rollback", "--generation", "1"); err == nil {
+		t.Fatal("rollback with explicit generation accepted")
+	}
+}
diff --git a/internal/cli/cluster_reader.go b/internal/cli/cluster_reader.go
index 231069df..2bb4f792 100644
--- a/internal/cli/cluster_reader.go
+++ b/internal/cli/cluster_reader.go
@@ -33,7 +33,7 @@ func (a *App) RunClusterReader(ctx context.Context, args []string) error {
 		command = rest[1]
 	}
 	switch command {
-	case "clusters", "durable-clusters", "clusters-report", "cluster-detail":
+	case "clusters", "durable-clusters", "clusters-report", "cluster-detail", "cluster-candidates":
 	default:
 		return usageErr(fmt.Errorf("cluster reader supports local cluster reads only"))
 	}
@@ -62,10 +62,12 @@ func (a *App) RunClusterReader(ctx context.Context, args []string) error {
 		return a.runClustersReport(ctx, rest[1:])
 	case "cluster-detail":
 		return a.runClusterDetail(ctx, rest[1:])
+	case "cluster-candidates":
+		return a.runClusterCandidates(ctx, rest[1:], true)
 	}
 	return usageErr(fmt.Errorf("unsupported cluster read"))
 }
 func (a *App) clusterReaderUsage() error {
-	_, err := fmt.Fprintln(a.Stdout, "Usage: gitcrawl-cluster-reader --config SOURCE_CONFIG clusters|durable-clusters|clusters-report|cluster-detail [options]\nLocal read-only cluster reader; no collection, embedding or provider commands.")
+	_, err := fmt.Fprintln(a.Stdout, "Usage: gitcrawl-cluster-reader --config SOURCE_CONFIG clusters|durable-clusters|clusters-report|cluster-detail|cluster-candidates [options]\nLocal read-only cluster reader; no collection, embedding or provider commands.")
 	return err
 }
diff --git a/internal/cli/help.go b/internal/cli/help.go
index ff76fad0..8ee6a011 100644
--- a/internal/cli/help.go
+++ b/internal/cli/help.go
@@ -81,6 +81,7 @@ Core commands:
   durable-clusters     list durable cluster groups
   cluster-detail       dump one latest run cluster, with durable fallback
   cluster-explain      alias for cluster-detail
+  cluster-candidates   read run-scoped computed candidate generations
   neighbors            list vector-nearest local issue and pull request rows
   search               search local thread and source documents; also supports search issues|prs gh syntax
   gh                   moved to Octopool; prints migration note
@@ -254,6 +255,19 @@ Usage:
 
 Usage:
   gitcrawl cluster-detail owner/repo --id N [--source auto|run|durable] [--member-limit N] [--body-chars N] [--hide-closed] [--json]
+`,
+	"cluster-candidates": `gitcrawl cluster-candidates reads computed candidate generations.
+
+Each accepted native cluster run records its complete computed partition as an
+immutable generation and selects it. Candidate IDs (GENERATION:ORDINAL) are
+run-scoped, never durable cluster IDs; observed_overlaps list shared members
+with durable groups and do not claim equivalence. --rollback selects the
+previous generation; the next accepted run selects its own again.
+
+Usage:
+  gitcrawl cluster-candidates owner/repo [--generation N] [--limit N] [--members] [--json]
+  gitcrawl cluster-candidates owner/repo --generations [--limit N] [--json]
+  gitcrawl cluster-candidates owner/repo --rollback [--json]
 `,
 	"neighbors": `gitcrawl neighbors lists vector-nearest local issue and pull request rows.
 
diff --git a/internal/store/cluster_generations.go b/internal/store/cluster_generations.go
new file mode 100644
index 00000000..02562bdc
--- /dev/null
+++ b/internal/store/cluster_generations.go
@@ -0,0 +1,512 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strconv"
+	"time"
+)
+
+// ClusterCandidateFormatV3 identifies an immutable, run-scoped computed
+// candidate generation. Candidate IDs ("<generation>:<ordinal>") are only
+// meaningful inside their generation; they never alias or merge durable
+// cluster_groups identities, whose six history tables this file never writes.
+const ClusterCandidateFormatV3 = "gitcrawl.cluster-candidates.v3"
+
+var ErrNoClusterCandidateGeneration = errors.New("no computed cluster candidate generation")
+
+func (s *Store) ensureClusterCandidateGenerationSchema(ctx context.Context) error {
+	_, err := s.q().ExecContext(ctx, `
+CREATE TABLE IF NOT EXISTS cluster_candidate_generations(
+ id INTEGER PRIMARY KEY,
+ repo_id INTEGER NOT NULL REFERENCES repositories(id),
+ run_id INTEGER NOT NULL UNIQUE REFERENCES cluster_runs(id),
+ format TEXT NOT NULL CHECK(format='gitcrawl.cluster-candidates.v3'),
+ algorithm TEXT NOT NULL,
+ input_sha256 TEXT NOT NULL,
+ quarantine_plan_sha256 TEXT NOT NULL,
+ partition_sha256 TEXT NOT NULL,
+ candidate_count INTEGER NOT NULL CHECK(candidate_count>=0),
+ member_count INTEGER NOT NULL CHECK(member_count>=0),
+ overlap_count INTEGER NOT NULL CHECK(overlap_count>=0),
+ durable_published_candidates INTEGER NOT NULL CHECK(durable_published_candidates>=0),
+ durable_quarantined_candidates INTEGER NOT NULL CHECK(durable_quarantined_candidates>=0),
+ created_at TEXT NOT NULL);
+CREATE INDEX IF NOT EXISTS cluster_candidate_generations_repo ON cluster_candidate_generations(repo_id,id);
+CREATE TABLE IF NOT EXISTS cluster_candidates(
+ generation_id INTEGER NOT NULL REFERENCES cluster_candidate_generations(id),
+ ordinal INTEGER NOT NULL CHECK(ordinal>0),
+ computed_key TEXT NOT NULL,
+ representative_thread_id INTEGER,
+ member_count INTEGER NOT NULL CHECK(member_count>0),
+ durable_status TEXT NOT NULL CHECK(durable_status IN ('published','identity_quarantined')),
+ durable_cluster_id INTEGER,
+ PRIMARY KEY(generation_id,ordinal),
+ UNIQUE(generation_id,computed_key));
+CREATE TABLE IF NOT EXISTS cluster_candidate_members(
+ generation_id INTEGER NOT NULL,
+ ordinal INTEGER NOT NULL,
+ thread_id INTEGER NOT NULL,
+ role TEXT NOT NULL,
+ PRIMARY KEY(generation_id,thread_id),
+ FOREIGN KEY(generation_id,ordinal) REFERENCES cluster_candidates(generation_id,ordinal));
+CREATE INDEX IF NOT EXISTS cluster_candidate_members_candidate ON cluster_candidate_members(generation_id,ordinal);
+CREATE TABLE IF NOT EXISTS cluster_candidate_overlaps(
+ generation_id INTEGER NOT NULL,
+ ordinal INTEGER NOT NULL,
+ cluster_id INTEGER NOT NULL,
+ shared_members INTEGER NOT NULL CHECK(shared_members>0),
+ PRIMARY KEY(generation_id,ordinal,cluster_id),
+ FOREIGN KEY(generation_id,ordinal) REFERENCES cluster_candidates(generation_id,ordinal));
+CREATE TABLE IF NOT EXISTS cluster_candidate_selection(
+ repo_id INTEGER PRIMARY KEY REFERENCES repositories(id),
+ generation_id INTEGER NOT NULL REFERENCES cluster_candidate_generations(id),
+ selected_at TEXT NOT NULL,
+ reason TEXT NOT NULL CHECK(reason IN ('accepted','rollback')));
+`)
+	if err != nil {
+		return err
+	}
+	// Generations are immutable history; only the selection pointer moves.
+	// The single exception is owner removal: PurgeThreads records the thread
+	// exclusion first and may then delete that thread's member rows.
+	for _, table := range []string{"cluster_candidate_generations", "cluster_candidates", "cluster_candidate_members", "cluster_candidate_overlaps"} {
+		for _, event := range []string{"UPDATE", "DELETE"} {
+			condition := ""
+			if table == "cluster_candidate_members" && event == "DELETE" {
+				if err = s.ensureThreadExclusionsSchema(ctx); err != nil {
+					return err
+				}
+				condition = " WHEN NOT EXISTS(SELECT 1 FROM thread_exclusions WHERE original_thread_id=OLD.thread_id)"
+			}
+			if _, err = s.q().ExecContext(ctx, "CREATE TRIGGER IF NOT EXISTS "+table+"_immutable_"+event+" BEFORE "+event+" ON "+table+condition+
+				" BEGIN SELECT RAISE(ABORT,'cluster candidate generations are immutable'); END"); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// ClusterCandidateGenerationInput is the complete computed partition and its
+// v2 quarantine plan for one accepted native run.
+type ClusterCandidateGenerationInput struct {
+	RepoID               int64
+	RunID                int64
+	Algorithm            string
+	InputSHA256          string
+	QuarantinePlanSHA256 string
+	Plan                 ClusterQuarantinePlan
+}
+
+type ClusterCandidateGeneration struct {
+	ID                           int64  `json:"generation_id"`
+	Format                       string `json:"format"`
+	RepoID                       int64  `json:"repo_id"`
+	RunID                        int64  `json:"run_id"`
+	Algorithm                    string `json:"algorithm"`
+	InputSHA256                  string `json:"input_sha256"`
+	QuarantinePlanSHA256         string `json:"quarantine_plan_sha256"`
+	PartitionSHA256              string `json:"partition_sha256"`
+	CandidateCount               int    `json:"candidate_count"`
+	MemberCount                  int    `json:"member_count"`
+	OverlapCount                 int    `json:"overlap_count"`
+	DurablePublishedCandidates   int    `json:"durable_published_candidates"`
+	DurableQuarantinedCandidates int    `json:"durable_quarantined_candidates"`
+	CreatedAt                    string `json:"created_at"`
+	Selected                     bool   `json:"selected"`
+	SelectedAt                   string `json:"selected_at,omitempty"`
+	SelectionReason              string `json:"selection_reason,omitempty"`
+}
+
+type ClusterCandidateOverlap struct {
+	ClusterID     int64 `json:"durable_cluster_id"`
+	SharedMembers int   `json:"shared_members"`
+}
+
+type ClusterCandidateMember struct {
+	ThreadID int64  `json:"thread_id"`
+	Role     string `json:"role"`
+}
+
+type ClusterCandidate struct {
+	ID                     string                    `json:"computed_candidate_id"`
+	Ordinal                int                       `json:"ordinal"`
+	ComputedKey            string                    `json:"computed_key"`
+	RepresentativeThreadID int64                     `json:"representative_thread_id,omitempty"`
+	MemberCount            int                       `json:"member_count"`
+	DurableStatus          string                    `json:"durable_status"`
+	DurableClusterID       int64                     `json:"durable_cluster_id,omitempty"`
+	Overlaps               []ClusterCandidateOverlap `json:"observed_overlaps"`
+	Members                []ClusterCandidateMember  `json:"members,omitempty"`
+}
+
+func clusterCandidateID(generation int64, ordinal int) string {
+	return strconv.FormatInt(generation, 10) + ":" + strconv.Itoa(ordinal)
+}
+
+// SaveClusterCandidateGeneration records the complete computed partition of
+// an accepted native run and selects it, inside the caller's transaction.
+// Overlaps are observed shared active members with prior durable groups; they
+// are lineage evidence, never equivalence, aliases or merges.
+func (s *Store) SaveClusterCandidateGeneration(ctx context.Context, in ClusterCandidateGenerationInput) (ClusterCandidateGeneration, error) {
+	var out ClusterCandidateGeneration
+	if s.queries == nil {
+		return out, fmt.Errorf("cluster candidate generation requires the native run transaction")
+	}
+	if in.RepoID <= 0 || in.RunID <= 0 || in.Algorithm == "" || len(in.InputSHA256) != 64 || len(in.QuarantinePlanSHA256) != 64 {
+		return out, fmt.Errorf("invalid cluster candidate generation binding")
+	}
+	if in.Plan.RepositoryID != in.RepoID {
+		return out, fmt.Errorf("cluster candidate generation repository mismatch")
+	}
+	if err := s.ensureClusterCandidateGenerationSchema(ctx); err != nil {
+		return out, err
+	}
+	var runRepo int64
+	if err := s.q().QueryRowContext(ctx, "SELECT repo_id FROM cluster_runs WHERE id=?", in.RunID).Scan(&runRepo); err != nil {
+		return out, fmt.Errorf("cluster candidate generation run: %w", err)
+	}
+	if runRepo != in.RepoID {
+		return out, fmt.Errorf("cluster candidate generation run belongs to another repository")
+	}
+	candidates := append([]DurableClusterInput(nil), in.Plan.Candidates...)
+	sort.Slice(candidates, func(i, j int) bool { return candidates[i].StableKey < candidates[j].StableKey })
+	published := map[string]int64{}
+	for _, target := range in.Plan.Targets {
+		published[target.CandidateKey] = target.ClusterID
+	}
+	activeOwner := map[int64][]int64{}
+	for _, identity := range in.Plan.Identities {
+		if !identity.Active {
+			continue
+		}
+		for _, thread := range identity.ActiveMembers {
+			activeOwner[thread] = append(activeOwner[thread], identity.ID)
+		}
+	}
+	selected := map[int64]bool{}
+	for _, id := range in.Plan.SelectedIDs {
+		selected[id] = true
+	}
+	type partitionRow struct {
+		Key     string  `json:"key"`
+		Members []int64 `json:"members"`
+	}
+	partition := make([]partitionRow, 0, len(candidates))
+	seen := map[int64]bool{}
+	members, overlaps, durablePublished := 0, 0, 0
+	type candidateRow struct {
+		input    DurableClusterInput
+		threads  []int64
+		overlaps map[int64]int
+		durable  int64
+		status   string
+	}
+	rows := make([]candidateRow, 0, len(candidates))
+	for _, candidate := range candidates {
+		if candidate.StableKey == "" || len(candidate.Members) == 0 {
+			return out, fmt.Errorf("cluster candidate generation requires keyed, non-empty candidates")
+		}
+		row := candidateRow{input: candidate, overlaps: map[int64]int{}, status: "identity_quarantined"}
+		for _, member := range candidate.Members {
+			if member.ThreadID <= 0 || seen[member.ThreadID] || !selected[member.ThreadID] {
+				return out, fmt.Errorf("cluster candidate generation membership is not a partition of the selected cohort")
+			}
+			seen[member.ThreadID] = true
+			row.threads = append(row.threads, member.ThreadID)
+			for _, cluster := range activeOwner[member.ThreadID] {
+				row.overlaps[cluster]++
+			}
+		}
+		sort.Slice(row.threads, func(i, j int) bool { return row.threads[i] < row.threads[j] })
+		if cluster, ok := published[candidate.StableKey]; ok {
+			row.status, row.durable = "published", cluster
+			durablePublished++
+		}
+		members += len(row.threads)
+		overlaps += len(row.overlaps)
+		partition = append(partition, partitionRow{Key: candidate.StableKey, Members: row.threads})
+		rows = append(rows, row)
+	}
+	if len(seen) != len(selected) {
+		return out, fmt.Errorf("cluster candidate generation does not cover the selected cohort")
+	}
+	// Owner-removed threads never enter a generation, even if a caller's input
+	// is stale.
+	if s.hasTable(ctx, "thread_exclusions") {
+		for thread := range seen {
+			var excluded bool
+			if err := s.q().QueryRowContext(ctx, "SELECT EXISTS(SELECT 1 FROM thread_exclusions WHERE original_thread_id=?)", thread).Scan(&excluded); err != nil {
+				return out, err
+			}
+			if excluded {
+				return out, fmt.Errorf("cluster candidate generation contains an owner-excluded thread")
+			}
+		}
+	}
+	encoded, err := json.Marshal(partition)
+	if err != nil {
+		return out, err
+	}
+	out = ClusterCandidateGeneration{Format: ClusterCandidateFormatV3, RepoID: in.RepoID, RunID: in.RunID, Algorithm: in.Algorithm,
+		InputSHA256: in.InputSHA256, QuarantinePlanSHA256: in.QuarantinePlanSHA256, PartitionSHA256: clusterHash(encoded),
+		CandidateCount: len(rows), MemberCount: members, OverlapCount: overlaps,
+		DurablePublishedCandidates: durablePublished, DurableQuarantinedCandidates: len(rows) - durablePublished,
+		CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), Selected: true, SelectionReason: "accepted"}
+	result, err := s.q().ExecContext(ctx, `INSERT INTO cluster_candidate_generations(repo_id,run_id,format,algorithm,input_sha256,quarantine_plan_sha256,partition_sha256,candidate_count,member_count,overlap_count,durable_published_candidates,durable_quarantined_candidates,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`,
+		out.RepoID, out.RunID, out.Format, out.Algorithm, out.InputSHA256, out.QuarantinePlanSHA256, out.PartitionSHA256,
+		out.CandidateCount, out.MemberCount, out.OverlapCount, out.DurablePublishedCandidates, out.DurableQuarantinedCandidates, out.CreatedAt)
+	if err != nil {
+		return out, err
+	}
+	if out.ID, err = result.LastInsertId(); err != nil {
+		return out, err
+	}
+	for index, row := range rows {
+		ordinal := index + 1
+		var representative any
+		if row.input.RepresentativeThreadID > 0 {
+			index := sort.Search(len(row.threads), func(i int) bool { return row.threads[i] >= row.input.RepresentativeThreadID })
+			if index == len(row.threads) || row.threads[index] != row.input.RepresentativeThreadID {
+				return out, fmt.Errorf("cluster candidate representative is outside its candidate")
+			}
+			representative = row.input.RepresentativeThreadID
+		}
+		var durable any
+		if row.durable > 0 {
+			durable = row.durable
+		}
+		if _, err = s.q().ExecContext(ctx, "INSERT INTO cluster_candidates(generation_id,ordinal,computed_key,representative_thread_id,member_count,durable_status,durable_cluster_id) VALUES(?,?,?,?,?,?,?)",
+			out.ID, ordinal, row.input.StableKey, representative, len(row.threads), row.status, durable); err != nil {
+			return out, err
+		}
+		for _, member := range row.input.Members {
+			role := member.Role
+			if role == "" {
+				role = "member"
+			}
+			if _, err = s.q().ExecContext(ctx, "INSERT INTO cluster_candidate_members(generation_id,ordinal,thread_id,role) VALUES(?,?,?,?)", out.ID, ordinal, member.ThreadID, role); err != nil {
+				return out, err
+			}
+		}
+		clusters := make([]int64, 0, len(row.overlaps))
+		for cluster := range row.overlaps {
+			clusters = append(clusters, cluster)
+		}
+		sort.Slice(clusters, func(i, j int) bool { return clusters[i] < clusters[j] })
+		for _, cluster := range clusters {
+			if _, err = s.q().ExecContext(ctx, "INSERT INTO cluster_candidate_overlaps(generation_id,ordinal,cluster_id,shared_members) VALUES(?,?,?,?)", out.ID, ordinal, cluster, row.overlaps[cluster]); err != nil {
+				return out, err
+			}
+		}
+	}
+	if err = s.selectClusterCandidateGeneration(ctx, in.RepoID, out.ID, "accepted", out.CreatedAt); err != nil {
+		return out, err
+	}
+	out.SelectedAt = out.CreatedAt
+	return out, nil
+}
+
+func (s *Store) selectClusterCandidateGeneration(ctx context.Context, repoID, generationID int64, reason, at string) error {
+	_, err := s.q().ExecContext(ctx, `INSERT INTO cluster_candidate_selection(repo_id,generation_id,selected_at,reason) VALUES(?,?,?,?)
+ON CONFLICT(repo_id) DO UPDATE SET generation_id=excluded.generation_id,selected_at=excluded.selected_at,reason=excluded.reason`, repoID, generationID, at, reason)
+	return err
+}
+
+// RollbackClusterCandidateGeneration atomically selects the newest generation
+// older than the current selection. Generations themselves are unchanged; the
+// next accepted native run selects its own newer generation again.
+func (s *Store) RollbackClusterCandidateGeneration(ctx context.Context, repoID int64) (ClusterCandidateGeneration, error) {
+	var out ClusterCandidateGeneration
+	err := s.WithTx(ctx, func(tx *Store) error {
+		if !tx.hasTable(ctx, "cluster_candidate_selection") {
+			return ErrNoClusterCandidateGeneration
+		}
+		var current int64
+		if err := tx.q().QueryRowContext(ctx, "SELECT generation_id FROM cluster_candidate_selection WHERE repo_id=?", repoID).Scan(&current); err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return ErrNoClusterCandidateGeneration
+			}
+			return err
+		}
+		var previous int64
+		if err := tx.q().QueryRowContext(ctx, "SELECT id FROM cluster_candidate_generations WHERE repo_id=? AND id<? ORDER BY id DESC LIMIT 1", repoID, current).Scan(&previous); err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return fmt.Errorf("no earlier computed cluster candidate generation to roll back to")
+			}
+			return err
+		}
+		if err := tx.selectClusterCandidateGeneration(ctx, repoID, previous, "rollback", time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
+			return err
+		}
+		var err error
+		out, err = tx.clusterCandidateGeneration(ctx, repoID, previous)
+		return err
+	})
+	return out, err
+}
+
+// SelectedClusterCandidateGeneration returns the selected generation.
+func (s *Store) SelectedClusterCandidateGeneration(ctx context.Context, repoID int64) (ClusterCandidateGeneration, error) {
+	if !s.hasTable(ctx, "cluster_candidate_selection") {
+		return ClusterCandidateGeneration{}, ErrNoClusterCandidateGeneration
+	}
+	var id int64
+	if err := s.q().QueryRowContext(ctx, "SELECT generation_id FROM cluster_candidate_selection WHERE repo_id=?", repoID).Scan(&id); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ClusterCandidateGeneration{}, ErrNoClusterCandidateGeneration
+		}
+		return ClusterCandidateGeneration{}, err
+	}
+	return s.clusterCandidateGeneration(ctx, repoID, id)
+}
+
+// ClusterCandidateGenerationByID returns one generation of the repository.
+func (s *Store) ClusterCandidateGenerationByID(ctx context.Context, repoID, id int64) (ClusterCandidateGeneration, error) {
+	if !s.hasTable(ctx, "cluster_candidate_generations") {
+		return ClusterCandidateGeneration{}, ErrNoClusterCandidateGeneration
+	}
+	return s.clusterCandidateGeneration(ctx, repoID, id)
+}
+
+func (s *Store) clusterCandidateGeneration(ctx context.Context, repoID, id int64) (ClusterCandidateGeneration, error) {
+	var g ClusterCandidateGeneration
+	var selectedAt, reason sql.NullString
+	err := s.q().QueryRowContext(ctx, `SELECT g.id,g.format,g.repo_id,g.run_id,g.algorithm,g.input_sha256,g.quarantine_plan_sha256,g.partition_sha256,g.candidate_count,g.member_count,g.overlap_count,g.durable_published_candidates,g.durable_quarantined_candidates,g.created_at,s.selected_at,s.reason
+FROM cluster_candidate_generations g LEFT JOIN cluster_candidate_selection s ON s.repo_id=g.repo_id AND s.generation_id=g.id
+WHERE g.repo_id=? AND g.id=?`, repoID, id).Scan(&g.ID, &g.Format, &g.RepoID, &g.RunID, &g.Algorithm, &g.InputSHA256, &g.QuarantinePlanSHA256, &g.PartitionSHA256,
+		&g.CandidateCount, &g.MemberCount, &g.OverlapCount, &g.DurablePublishedCandidates, &g.DurableQuarantinedCandidates, &g.CreatedAt, &selectedAt, &reason)
+	if errors.Is(err, sql.ErrNoRows) {
+		return g, ErrNoClusterCandidateGeneration
+	}
+	if err != nil {
+		return g, err
+	}
+	if g.Format != ClusterCandidateFormatV3 {
+		return g, fmt.Errorf("unsupported cluster candidate generation format %q", g.Format)
+	}
+	g.Selected, g.SelectedAt, g.SelectionReason = selectedAt.Valid, selectedAt.String, reason.String
+	return g, nil
+}
+
+// ListClusterCandidateGenerations returns the newest generations first.
+func (s *Store) ListClusterCandidateGenerations(ctx context.Context, repoID int64, limit int) ([]ClusterCandidateGeneration, error) {
+	if !s.hasTable(ctx, "cluster_candidate_generations") {
+		return []ClusterCandidateGeneration{}, nil
+	}
+	if limit <= 0 {
+		limit = 20
+	}
+	rows, err := s.q().QueryContext(ctx, "SELECT id FROM cluster_candidate_generations WHERE repo_id=? ORDER BY id DESC LIMIT ?", repoID, limit)
+	if err != nil {
+		return nil, err
+	}
+	var ids []int64
+	for rows.Next() {
+		var id int64
+		if err = rows.Scan(&id); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	err = rows.Err()
+	rows.Close()
+	if err != nil {
+		return nil, err
+	}
+	out := make([]ClusterCandidateGeneration, 0, len(ids))
+	for _, id := range ids {
+		g, err := s.clusterCandidateGeneration(ctx, repoID, id)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, g)
+	}
+	return out, nil
+}
+
+// ClusterCandidates returns a page of candidates of one generation, largest
+// first, with observed overlaps and optionally members.
+func (s *Store) ClusterCandidates(ctx context.Context, repoID, generationID int64, limit int, withMembers bool) ([]ClusterCandidate, error) {
+	if _, err := s.clusterCandidateGeneration(ctx, repoID, generationID); err != nil {
+		return nil, err
+	}
+	if limit <= 0 {
+		limit = 50
+	}
+	// Member counts and representatives are read from current member rows,
+	// so owner-removed threads are withheld everywhere.
+	rows, err := s.q().QueryContext(ctx, `SELECT c.ordinal,c.computed_key,
+ CASE WHEN EXISTS(SELECT 1 FROM cluster_candidate_members m WHERE m.generation_id=c.generation_id AND m.thread_id=c.representative_thread_id) THEN c.representative_thread_id ELSE 0 END,
+ (SELECT count(*) FROM cluster_candidate_members m WHERE m.generation_id=c.generation_id AND m.ordinal=c.ordinal) AS members,
+ c.durable_status,coalesce(c.durable_cluster_id,0)
+FROM cluster_candidates c WHERE c.generation_id=? AND members>0 ORDER BY members DESC,c.ordinal LIMIT ?`, generationID, limit)
+	if err != nil {
+		return nil, err
+	}
+	var out []ClusterCandidate
+	for rows.Next() {
+		var c ClusterCandidate
+		if err = rows.Scan(&c.Ordinal, &c.ComputedKey, &c.RepresentativeThreadID, &c.MemberCount, &c.DurableStatus, &c.DurableClusterID); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		c.ID = clusterCandidateID(generationID, c.Ordinal)
+		c.Overlaps = []ClusterCandidateOverlap{}
+		out = append(out, c)
+	}
+	err = rows.Err()
+	rows.Close()
+	if err != nil {
+		return nil, err
+	}
+	for i := range out {
+		overlapRows, err := s.q().QueryContext(ctx, "SELECT cluster_id,shared_members FROM cluster_candidate_overlaps WHERE generation_id=? AND ordinal=? ORDER BY shared_members DESC,cluster_id", generationID, out[i].Ordinal)
+		if err != nil {
+			return nil, err
+		}
+		for overlapRows.Next() {
+			var o ClusterCandidateOverlap
+			if err = overlapRows.Scan(&o.ClusterID, &o.SharedMembers); err != nil {
+				overlapRows.Close()
+				return nil, err
+			}
+			out[i].Overlaps = append(out[i].Overlaps, o)
+		}
+		err = overlapRows.Err()
+		overlapRows.Close()
+		if err != nil {
+			return nil, err
+		}
+		if !withMembers {
+			continue
+		}
+		memberRows, err := s.q().QueryContext(ctx, "SELECT thread_id,role FROM cluster_candidate_members WHERE generation_id=? AND ordinal=? ORDER BY thread_id", generationID, out[i].Ordinal)
+		if err != nil {
+			return nil, err
+		}
+		for memberRows.Next() {
+			var m ClusterCandidateMember
+			if err = memberRows.Scan(&m.ThreadID, &m.Role); err != nil {
+				memberRows.Close()
+				return nil, err
+			}
+			out[i].Members = append(out[i].Members, m)
+		}
+		err = memberRows.Err()
+		memberRows.Close()
+		if err != nil {
+			return nil, err
+		}
+	}
+	if out == nil {
+		out = []ClusterCandidate{}
+	}
+	return out, nil
+}
diff --git a/internal/store/cluster_generations_test.go b/internal/store/cluster_generations_test.go
new file mode 100644
index 00000000..25267ec8
--- /dev/null
+++ b/internal/store/cluster_generations_test.go
@@ -0,0 +1,246 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func seedCandidateThreads(t *testing.T, ctx context.Context, st *Store, count int) (int64, []int64) {
+	t.Helper()
+	now := time.Now().UTC().Format(time.RFC3339Nano)
+	repoID, err := st.UpsertRepository(ctx, Repository{Owner: "openclaw", Name: "openclaw", FullName: "openclaw/openclaw", RawJSON: "{}", UpdatedAt: now})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ids := make([]int64, 0, count)
+	for i := 1; i <= count; i++ {
+		id, err := st.UpsertThread(ctx, Thread{RepoID: repoID, GitHubID: fmt.Sprint(900 + i), Number: 900 + i, Kind: "issue", State: "open", Title: fmt.Sprintf("candidate thread %d", i),
+			HTMLURL: fmt.Sprintf("https://github.com/openclaw/openclaw/issues/%d", 900+i), LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: fmt.Sprint("h", i), UpdatedAt: now})
+		if err != nil {
+			t.Fatal(err)
+		}
+		ids = append(ids, id)
+	}
+	return repoID, ids
+}
+
+func candidateInput(key string, members ...int64) DurableClusterInput {
+	in := DurableClusterInput{StableKey: key, RepresentativeThreadID: members[0]}
+	for _, id := range members {
+		in.Members = append(in.Members, DurableClusterMemberInput{ThreadID: id, Role: "member"})
+	}
+	return in
+}
+
+// durableHistoryDump captures every row of the six durable history tables.
+func durableHistoryDump(t *testing.T, ctx context.Context, st *Store) string {
+	t.Helper()
+	var out strings.Builder
+	for _, table := range []string{"cluster_groups", "cluster_memberships", "cluster_overrides", "cluster_closures", "cluster_aliases", "cluster_events"} {
+		if !st.hasTable(ctx, table) {
+			fmt.Fprintf(&out, "%s:absent\n", table)
+			continue
+		}
+		rows, err := st.DB().QueryContext(ctx, "SELECT * FROM "+table+" ORDER BY rowid")
+		if err != nil {
+			t.Fatal(err)
+		}
+		columns, _ := rows.Columns()
+		var dump strings.Builder
+		for rows.Next() {
+			values := make([]any, len(columns))
+			pointers := make([]any, len(columns))
+			for i := range values {
+				pointers[i] = &values[i]
+			}
+			if err := rows.Scan(pointers...); err != nil {
+				t.Fatal(err)
+			}
+			fmt.Fprintf(&dump, "%v;", values)
+		}
+		if err := errors.Join(rows.Err(), rows.Close()); err != nil {
+			t.Fatal(err)
+		}
+		fmt.Fprintf(&out, "%s:%s\n", table, dump.String())
+	}
+	return out.String()
+}
+
+func saveGeneration(t *testing.T, ctx context.Context, st *Store, plan ClusterQuarantinePlan, input string) (ClusterCandidateGeneration, error) {
+	t.Helper()
+	var generation ClusterCandidateGeneration
+	err := st.WithTx(ctx, func(tx *Store) error {
+		runID, err := tx.insertClusterRun(ctx, plan.RepositoryID, time.Now().UTC().Format(time.RFC3339Nano))
+		if err != nil {
+			return err
+		}
+		generation, err = tx.SaveClusterCandidateGeneration(ctx, ClusterCandidateGenerationInput{RepoID: plan.RepositoryID, RunID: runID,
+			Algorithm: "test-algorithm", InputSHA256: strings.Repeat(input, 64), QuarantinePlanSHA256: strings.Repeat("b", 64), Plan: plan})
+		return err
+	})
+	return generation, err
+}
+
+func TestClusterCandidateGenerationsPublishSelectAndRollBack(t *testing.T) {
+	ctx := context.Background()
+	st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	repoID, ids := seedCandidateThreads(t, ctx, st, 6)
+	before := durableHistoryDump(t, ctx, st)
+	if _, err := st.SelectedClusterCandidateGeneration(ctx, repoID); !errors.Is(err, ErrNoClusterCandidateGeneration) {
+		t.Fatalf("empty selection error = %v", err)
+	}
+	// A two-group merge and a split: both are durable quarantine, but every
+	// computed candidate is published in the run-scoped domain.
+	plan := ClusterQuarantinePlan{RepositoryID: repoID, SelectedIDs: ids,
+		Candidates: []DurableClusterInput{candidateInput("b-merge", ids[0], ids[1], ids[2]), candidateInput("a-split", ids[3]), candidateInput("c-new", ids[4], ids[5])},
+		Identities: []ClusterIdentityMetadata{{ID: 11, Active: true, ActiveMembers: []int64{ids[0], ids[1]}}, {ID: 12, Active: true, ActiveMembers: []int64{ids[2], ids[3]}}, {ID: 13, Active: false, ActiveMembers: []int64{ids[4]}}},
+		Targets:    []ClusterPublicationTarget{{CandidateKey: "c-new", ClusterID: 0}}}
+	first, err := saveGeneration(t, ctx, st, plan, "1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.CandidateCount != 3 || first.MemberCount != 6 || first.OverlapCount != 3 || first.DurablePublishedCandidates != 1 || first.DurableQuarantinedCandidates != 2 || !first.Selected || first.Format != ClusterCandidateFormatV3 || len(first.PartitionSHA256) != 64 {
+		t.Fatalf("first generation = %+v", first)
+	}
+	candidates, err := st.ClusterCandidates(ctx, repoID, first.ID, 10, true)
+	if err != nil || len(candidates) != 3 {
+		t.Fatalf("candidates=%+v err=%v", candidates, err)
+	}
+	merge := candidates[0]
+	if merge.ComputedKey != "b-merge" || merge.ID != fmt.Sprintf("%d:2", first.ID) || merge.MemberCount != 3 || merge.DurableStatus != "identity_quarantined" ||
+		len(merge.Overlaps) != 2 || merge.Overlaps[0] != (ClusterCandidateOverlap{ClusterID: 11, SharedMembers: 2}) || len(merge.Members) != 3 {
+		t.Fatalf("merge candidate = %+v", merge)
+	}
+	if after := durableHistoryDump(t, ctx, st); after != before {
+		t.Fatalf("generation changed durable history:\n%s\n%s", before, after)
+	}
+	plan.Candidates = []DurableClusterInput{candidateInput("x", ids[0], ids[1], ids[2], ids[3]), candidateInput("y", ids[4], ids[5])}
+	second, err := saveGeneration(t, ctx, st, plan, "2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if selected, err := st.SelectedClusterCandidateGeneration(ctx, repoID); err != nil || selected.ID != second.ID || selected.SelectionReason != "accepted" {
+		t.Fatalf("selected=%+v err=%v", selected, err)
+	}
+	rolled, err := st.RollbackClusterCandidateGeneration(ctx, repoID)
+	if err != nil || rolled.ID != first.ID || rolled.SelectionReason != "rollback" || !rolled.Selected {
+		t.Fatalf("rollback=%+v err=%v", rolled, err)
+	}
+	if _, err := st.RollbackClusterCandidateGeneration(ctx, repoID); err == nil || !strings.Contains(err.Error(), "no earlier") {
+		t.Fatalf("rollback past the first generation = %v", err)
+	}
+	listed, err := st.ListClusterCandidateGenerations(ctx, repoID, 10)
+	if err != nil || len(listed) != 2 || listed[0].ID != second.ID || listed[0].Selected || !listed[1].Selected {
+		t.Fatalf("listed=%+v err=%v", listed, err)
+	}
+	// Generations are immutable history.
+	for _, statement := range []string{"UPDATE cluster_candidates SET member_count=99", "DELETE FROM cluster_candidate_overlaps", "UPDATE cluster_candidate_generations SET run_id=0", "DELETE FROM cluster_candidate_members"} {
+		if _, err := st.DB().ExecContext(ctx, statement); err == nil || !strings.Contains(err.Error(), "immutable") {
+			t.Fatalf("%s error = %v", statement, err)
+		}
+	}
+	// A third accepted run selects its own newer generation again.
+	third, err := saveGeneration(t, ctx, st, plan, "3")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if selected, _ := st.SelectedClusterCandidateGeneration(ctx, repoID); selected.ID != third.ID {
+		t.Fatalf("new run did not select its generation: %+v", selected)
+	}
+	if after := durableHistoryDump(t, ctx, st); after != before {
+		t.Fatal("generations changed durable history")
+	}
+}
+
+func TestClusterCandidateGenerationRefusesInvalidPartitions(t *testing.T) {
+	ctx := context.Background()
+	st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	repoID, ids := seedCandidateThreads(t, ctx, st, 3)
+	valid := ClusterQuarantinePlan{RepositoryID: repoID, SelectedIDs: ids, Candidates: []DurableClusterInput{candidateInput("a", ids[0], ids[1]), candidateInput("b", ids[2])}}
+	cases := map[string]func(*ClusterQuarantinePlan){
+		"duplicate member":   func(p *ClusterQuarantinePlan) { p.Candidates[1] = candidateInput("b", ids[1], ids[2]) },
+		"uncovered selected": func(p *ClusterQuarantinePlan) { p.Candidates = p.Candidates[:1] },
+		"outside selected":   func(p *ClusterQuarantinePlan) { p.SelectedIDs = ids[:2] },
+		"empty candidate":    func(p *ClusterQuarantinePlan) { p.Candidates[1].Members = nil },
+		"other repository":   func(p *ClusterQuarantinePlan) { p.RepositoryID = repoID + 1 },
+		"outside representative": func(p *ClusterQuarantinePlan) {
+			p.Candidates[1].RepresentativeThreadID = ids[0]
+		},
+	}
+	for name, mutate := range cases {
+		t.Run(name, func(t *testing.T) {
+			plan := valid
+			plan.Candidates = append([]DurableClusterInput(nil), valid.Candidates...)
+			mutate(&plan)
+			if _, err := saveGeneration(t, ctx, st, plan, "a"); err == nil {
+				t.Fatal("invalid partition accepted")
+			}
+			if _, err := st.SelectedClusterCandidateGeneration(ctx, repoID); !errors.Is(err, ErrNoClusterCandidateGeneration) {
+				t.Fatalf("failed save left a selection: %v", err)
+			}
+		})
+	}
+	if _, err := st.SaveClusterCandidateGeneration(ctx, ClusterCandidateGenerationInput{RepoID: repoID, RunID: 1, Algorithm: "a", InputSHA256: strings.Repeat("a", 64), QuarantinePlanSHA256: strings.Repeat("b", 64), Plan: valid}); err == nil || !strings.Contains(err.Error(), "transaction") {
+		t.Fatalf("save outside a transaction = %v", err)
+	}
+	if _, err := st.RollbackClusterCandidateGeneration(ctx, repoID); !errors.Is(err, ErrNoClusterCandidateGeneration) {
+		t.Fatalf("rollback without a generation = %v", err)
+	}
+}
+
+func TestClusterCandidateGenerationsHonorOwnerRemoval(t *testing.T) {
+	ctx := context.Background()
+	st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	repoID, ids := seedCandidateThreads(t, ctx, st, 3)
+	plan := ClusterQuarantinePlan{RepositoryID: repoID, SelectedIDs: ids, Candidates: []DurableClusterInput{candidateInput("a", ids[0], ids[1]), candidateInput("b", ids[2])}}
+	generation, err := saveGeneration(t, ctx, st, plan, "a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	preview, err := st.PlanThreadPurge(ctx, "openclaw/openclaw", []int{901})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if preview.Counts["cluster_candidate_members"] != 1 {
+		t.Fatalf("purge plan did not inventory candidate members: %+v", preview.Counts)
+	}
+	if _, err := st.PurgeThreads(ctx, "openclaw/openclaw", []int{901}, preview.PlanID); err != nil {
+		t.Fatal(err)
+	}
+	candidates, err := st.ClusterCandidates(ctx, repoID, generation.ID, 10, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, candidate := range candidates {
+		for _, member := range candidate.Members {
+			if member.ThreadID == ids[0] {
+				t.Fatal("owner-removed thread remains a candidate member")
+			}
+		}
+		if candidate.RepresentativeThreadID == ids[0] || candidate.ComputedKey == "a" && candidate.MemberCount != 1 {
+			t.Fatalf("owner-removed thread still counted: %+v", candidate)
+		}
+	}
+	// A later generation containing an owner-excluded thread is refused.
+	var stale ClusterQuarantinePlan = plan
+	if _, err := saveGeneration(t, ctx, st, stale, "c"); err == nil || !strings.Contains(err.Error(), "owner-excluded") {
+		t.Fatalf("stale owner-excluded input = %v", err)
+	}
+}
diff --git a/internal/store/portable_schema.go b/internal/store/portable_schema.go
index 46a06eff..5cae121a 100644
--- a/internal/store/portable_schema.go
+++ b/internal/store/portable_schema.go
@@ -442,6 +442,11 @@ func canonicalPortableDroppedTables() []string {
 		"thread_code_snapshots",
 		"thread_changed_files",
 		"thread_hunk_signatures",
+		"cluster_candidate_selection",
+		"cluster_candidate_overlaps",
+		"cluster_candidate_members",
+		"cluster_candidates",
+		"cluster_candidate_generations",
 		"cluster_events",
 		"cluster_members",
 		"clusters",
@@ -476,6 +481,11 @@ func canonicalPortableBulkDropOrder() []string {
 		"thread_changed_files",
 		"thread_hunk_signatures",
 		"thread_code_snapshots",
+		"cluster_candidate_selection",
+		"cluster_candidate_overlaps",
+		"cluster_candidate_members",
+		"cluster_candidates",
+		"cluster_candidate_generations",
 		"cluster_events",
 		"cluster_members",
 		"clusters",
diff --git a/internal/store/store.go b/internal/store/store.go
index 75d2d602..b225294f 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -355,6 +355,9 @@ func (s *Store) migrate(ctx context.Context) error {
 	if err := s.ensureThreadExclusionsSchema(ctx); err != nil {
 		return err
 	}
+	if err := s.ensureClusterCandidateGenerationSchema(ctx); err != nil {
+		return err
+	}
 	if err := s.ensureCanonicalObservationTables(ctx); err != nil {
 		return err
 	}
diff --git a/internal/store/thread_exclusions.go b/internal/store/thread_exclusions.go
index f440ee1a..2311fdfe 100644
--- a/internal/store/thread_exclusions.go
+++ b/internal/store/thread_exclusions.go
@@ -135,6 +135,7 @@ func purgeRelations(ids string) map[string]string {
 		out[table] = "snapshot_id in (" + snapshots + ")"
 	}
 	out["similarity_edges"] = "left_thread_id in (" + ids + ") or right_thread_id in (" + ids + ")"
+	out["cluster_candidate_members"] = "thread_id in (" + ids + ")"
 	return out
 }
 
@@ -484,6 +485,13 @@ func (s *Store) PurgeThreads(ctx context.Context, repository string, numbers []i
 					return err
 				}
 			}
+			// Computed candidate generations are immutable except for owner
+			// removal; the exclusion recorded above admits this deletion.
+			if tx.hasTable(ctx, "cluster_candidate_members") {
+				if _, err = tx.q().ExecContext(ctx, "DELETE FROM cluster_candidate_members WHERE thread_id=?", t.ThreadID); err != nil {
+					return err
+				}
+			}
 			if _, err = tx.q().ExecContext(ctx, "DELETE FROM threads WHERE id=?", t.ThreadID); err != nil {
 				return err
 			}
diff --git a/scripts/native-cluster/README.md b/scripts/native-cluster/README.md
index 510ad6ec..6e5cfbc6 100644
--- a/scripts/native-cluster/README.md
+++ b/scripts/native-cluster/README.md
@@ -117,3 +117,17 @@ The strict wire schema is in docs/native-cluster-scope-v2.schema.json. Synthetic
 writer fixtures cover partial publication, provider-complete identity quarantine,
 blocked analysis/no-op, indirect-write rollback and full recovery. Native model
 work remains exclusively in the authorized processor broker.
+
+## Computed candidate generations (v3)
+
+After v2 publication and scope validation, the same native transaction records
+the complete computed candidate partition of the quarantine plan as an
+immutable `gitcrawl.cluster-candidates.v3` generation. The generation is bound
+to its run, algorithm, input hash and quarantine proof hash. It stores each
+candidate's members, its v2 durable status and target, and observed overlaps
+with active durable groups, then moves the per-repository selection pointer to
+it. The quarantine postimage hash is checked again after this write. A
+generation never writes the six durable history tables, so v2 scope, strict
+schema and reader semantics are unchanged. Blocked-identity runs still record
+their generation. No-op runs record none. See `docs/clustering.md`
+(Computed candidate generations) and `gitcrawl cluster-candidates`.
diff --git a/scripts/native-cluster/native_cluster.go b/scripts/native-cluster/native_cluster.go
index 068b31f1..6bbd2d42 100644
--- a/scripts/native-cluster/native_cluster.go
+++ b/scripts/native-cluster/native_cluster.go
@@ -425,6 +425,19 @@ func (s *Store) NativePublishClusters(ctx context.Context, want NativeClusterSna
 		if e = ValidateClusterScope(scope, result.RunID, scope.Repository); e != nil {
 			return e
 		}
+		// Record the complete computed partition as this run's immutable v3
+		// generation and select it in the same transaction. It never writes the
+		// durable identity tables, so the quarantine postimage stays unchanged.
+		if _, e = tx.SaveClusterCandidateGeneration(ctx, ClusterCandidateGenerationInput{RepoID: want.RepoID, RunID: result.RunID,
+			Algorithm: NativeClusterAlgorithm, InputSHA256: want.InputSHA256, QuarantinePlanSHA256: proofHash, Plan: plan}); e != nil {
+			return e
+		}
+		if after, e = tx.ClusterQuarantineRowHash(ctx, want.RepoID, plan.PreservedGroupIDs); e != nil {
+			return e
+		}
+		if after != plan.ProtectedPreimageSHA256 {
+			return fmt.Errorf("computed candidate generation changed quarantined identities; rollback")
+		}
 		if _, e = tx.q().ExecContext(ctx, "update cluster_runs set started_at=?,finished_at=? where id=?", got.Availability.AdmissionAt, time.Now().UTC().Format(time.RFC3339Nano), result.RunID); e != nil {
 			return e
 		}
diff --git a/scripts/native-cluster/quarantine_test.go b/scripts/native-cluster/quarantine_test.go
index 189d8595..dcaa8164 100644
--- a/scripts/native-cluster/quarantine_test.go
+++ b/scripts/native-cluster/quarantine_test.go
@@ -218,6 +218,17 @@ func TestNativeQuarantineWriterRollbackAnalysisAndRecovery(t *testing.T) {
 		t.Fatal("blocked analysis replaced success", read, e)
 	}
 	save("retained_only", blocked.RunID)
+	// v3: the durably blocked run still publishes its complete computed
+	// partition as the selected run-scoped generation.
+	computed, e := s.SelectedClusterCandidateGeneration(ctx, repo)
+	if e != nil || computed.RunID != blocked.RunID || computed.CandidateCount != 1 || computed.MemberCount != 5 ||
+		computed.DurableQuarantinedCandidates != 1 || computed.DurablePublishedCandidates != 0 || computed.OverlapCount == 0 {
+		t.Fatal("blocked run did not select its computed generation", computed, e)
+	}
+	generations, e := s.ListClusterCandidateGenerations(ctx, repo, 100)
+	if e != nil {
+		t.Fatal(e)
+	}
 	snap, e = s.NativeClusterSnapshot(ctx, policy)
 	if e != nil {
 		t.Fatal(e)
@@ -226,6 +237,9 @@ func TestNativeQuarantineWriterRollbackAnalysisAndRecovery(t *testing.T) {
 	if e != nil || same.RunID != blocked.RunID {
 		t.Fatal("blocked no-op churn", e)
 	}
+	if after, e := s.ListClusterCandidateGenerations(ctx, repo, 100); e != nil || len(after) != len(generations) {
+		t.Fatal("no-op run recorded another computed generation", e)
+	}
 	// A different genuine graph with one-to-one identity relations can recover.
 	recovery := []DurableClusterInput{quarantineInput("a", 2, unknown), quarantineInput("b", 3), quarantineInput("c", 4), quarantineInput("new-five", 5)}
 	// Reflect a changed native graph input, as a real algorithm change in source
@@ -257,6 +271,12 @@ func TestNativeQuarantineWriterRollbackAnalysisAndRecovery(t *testing.T) {
 		t.Fatal("full recovery", read, e)
 	}
 	save("full_recovery", full.RunID)
+	if computed, e := s.SelectedClusterCandidateGeneration(ctx, repo); e != nil || computed.RunID != full.RunID || computed.CandidateCount != 4 || computed.DurableQuarantinedCandidates != 0 {
+		t.Fatal("full recovery generation", computed, e)
+	}
+	if rolled, e := s.RollbackClusterCandidateGeneration(ctx, repo); e != nil || rolled.RunID != blocked.RunID {
+		t.Fatal("computed generation rollback", rolled, e)
+	}
 	if path := os.Getenv("GITCRAWL_QUARANTINE_FINAL_DB"); path != "" {
 		if _, err := s.DB().Exec("VACUUM INTO ?", path); err != nil {
 			t.Fatal(err)

From c84e136ed76a1083a57bac9b9a643b2424836d1b Mon Sep 17 00:00:00 2001
From: Hannes Rudolph <49103247+hannesrudolph@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:36:04 -0600
Subject: [PATCH 2/5] fix(clusters): admit candidate member cleanup only after
 thread removal

Key the immutability exception on the thread no longer existing instead of
on thread_exclusions, so schema changes to the exclusion tables cannot
break the trigger; purge deletes the thread before its member rows.
---
 docs/clustering.md                    | 2 +-
 internal/store/cluster_generations.go | 9 +++------
 internal/store/thread_exclusions.go   | 8 ++++----
 3 files changed, 8 insertions(+), 11 deletions(-)

diff --git a/docs/clustering.md b/docs/clustering.md
index df7b57d4..be2249aa 100644
--- a/docs/clustering.md
+++ b/docs/clustering.md
@@ -228,7 +228,7 @@ transaction:
   durable ID (`published`, with `durable_cluster_id`) or quarantined it
   (`identity_quarantined`). Both are present in the computed generation.
 - Generation rows cannot be updated or deleted, with one exception: owner
-  removal (`purge`) deletes a removed thread's member rows, and readers derive
+  removal (`purge`) deletes a removed thread, then its member rows, and readers derive
   counts and representatives from the remaining members.
 - `cluster-candidates --rollback` selects the previous generation. The next
   accepted run selects its own generation again. A ref no-op run records
diff --git a/internal/store/cluster_generations.go b/internal/store/cluster_generations.go
index 02562bdc..0c65ec7c 100644
--- a/internal/store/cluster_generations.go
+++ b/internal/store/cluster_generations.go
@@ -72,16 +72,13 @@ CREATE TABLE IF NOT EXISTS cluster_candidate_selection(
 		return err
 	}
 	// Generations are immutable history; only the selection pointer moves.
-	// The single exception is owner removal: PurgeThreads records the thread
-	// exclusion first and may then delete that thread's member rows.
+	// The single exception is owner removal: once PurgeThreads has deleted a
+	// thread, that thread's member rows may be deleted too.
 	for _, table := range []string{"cluster_candidate_generations", "cluster_candidates", "cluster_candidate_members", "cluster_candidate_overlaps"} {
 		for _, event := range []string{"UPDATE", "DELETE"} {
 			condition := ""
 			if table == "cluster_candidate_members" && event == "DELETE" {
-				if err = s.ensureThreadExclusionsSchema(ctx); err != nil {
-					return err
-				}
-				condition = " WHEN NOT EXISTS(SELECT 1 FROM thread_exclusions WHERE original_thread_id=OLD.thread_id)"
+				condition = " WHEN EXISTS(SELECT 1 FROM threads WHERE id=OLD.thread_id)"
 			}
 			if _, err = s.q().ExecContext(ctx, "CREATE TRIGGER IF NOT EXISTS "+table+"_immutable_"+event+" BEFORE "+event+" ON "+table+condition+
 				" BEGIN SELECT RAISE(ABORT,'cluster candidate generations are immutable'); END"); err != nil {
diff --git a/internal/store/thread_exclusions.go b/internal/store/thread_exclusions.go
index 2311fdfe..35550d43 100644
--- a/internal/store/thread_exclusions.go
+++ b/internal/store/thread_exclusions.go
@@ -485,16 +485,16 @@ func (s *Store) PurgeThreads(ctx context.Context, repository string, numbers []i
 					return err
 				}
 			}
+			if _, err = tx.q().ExecContext(ctx, "DELETE FROM threads WHERE id=?", t.ThreadID); err != nil {
+				return err
+			}
 			// Computed candidate generations are immutable except for owner
-			// removal; the exclusion recorded above admits this deletion.
+			// removal; deleting the thread above admits this deletion.
 			if tx.hasTable(ctx, "cluster_candidate_members") {
 				if _, err = tx.q().ExecContext(ctx, "DELETE FROM cluster_candidate_members WHERE thread_id=?", t.ThreadID); err != nil {
 					return err
 				}
 			}
-			if _, err = tx.q().ExecContext(ctx, "DELETE FROM threads WHERE id=?", t.ThreadID); err != nil {
-				return err
-			}
 		}
 		// Never turn removal of unavailable work into claimed provider completeness.
 		if tx.hasTable(ctx, "analytics_review_state_coverage") {

From 201a062dc27deba89dd3d2aa3f8c8eb3a20babe8 Mon Sep 17 00:00:00 2001
From: Hannes Rudolph <49103247+hannesrudolph@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:40:49 -0600
Subject: [PATCH 3/5] fix(clusters): record new durable IDs and drop
 generations from portable scope

Resolve each published candidate's durable ID after the v2 write, so
identities created by the run are recorded instead of a zero target.
Drop the derived candidate tables from a disposable portable snapshot
before repository filtering, which would otherwise hit their immutability
triggers.
---
 internal/cli/cluster_candidates_test.go    |  2 +-
 internal/store/cluster_generations.go      | 18 +++++++++-
 internal/store/cluster_generations_test.go | 42 +++++++++++++++++++++-
 internal/store/portable_scope.go           |  8 +++++
 scripts/native-cluster/native_cluster.go   | 13 ++++++-
 scripts/native-cluster/quarantine_test.go  | 11 ++++++
 6 files changed, 90 insertions(+), 4 deletions(-)

diff --git a/internal/cli/cluster_candidates_test.go b/internal/cli/cluster_candidates_test.go
index b5406399..2dada0d8 100644
--- a/internal/cli/cluster_candidates_test.go
+++ b/internal/cli/cluster_candidates_test.go
@@ -57,7 +57,7 @@ func TestClusterCandidatesCommandReadsSelectsAndRollsBack(t *testing.T) {
 		}
 		if err := st.WithTx(ctx, func(tx *store.Store) error {
 			_, err := tx.SaveClusterCandidateGeneration(ctx, store.ClusterCandidateGenerationInput{RepoID: repoID, RunID: runID, Algorithm: "test",
-				InputSHA256: strings.Repeat(fmt.Sprint(generation), 64), QuarantinePlanSHA256: strings.Repeat("b", 64), Plan: plan})
+				InputSHA256: strings.Repeat(fmt.Sprint(generation), 64), QuarantinePlanSHA256: strings.Repeat("b", 64), Plan: plan, PublishedClusterIDs: map[string]int64{"b": 7}})
 			return err
 		}); err != nil {
 			t.Fatal(err)
diff --git a/internal/store/cluster_generations.go b/internal/store/cluster_generations.go
index 0c65ec7c..3fbb037a 100644
--- a/internal/store/cluster_generations.go
+++ b/internal/store/cluster_generations.go
@@ -19,6 +19,11 @@ const ClusterCandidateFormatV3 = "gitcrawl.cluster-candidates.v3"
 
 var ErrNoClusterCandidateGeneration = errors.New("no computed cluster candidate generation")
 
+// clusterCandidateGenerationTables lists the v3 tables, children first.
+func clusterCandidateGenerationTables() []string {
+	return []string{"cluster_candidate_selection", "cluster_candidate_overlaps", "cluster_candidate_members", "cluster_candidates", "cluster_candidate_generations"}
+}
+
 func (s *Store) ensureClusterCandidateGenerationSchema(ctx context.Context) error {
 	_, err := s.q().ExecContext(ctx, `
 CREATE TABLE IF NOT EXISTS cluster_candidate_generations(
@@ -98,6 +103,10 @@ type ClusterCandidateGenerationInput struct {
 	InputSHA256          string
 	QuarantinePlanSHA256 string
 	Plan                 ClusterQuarantinePlan
+	// PublishedClusterIDs maps each published candidate key to the durable
+	// cluster_groups ID it was written to in this run, including identities
+	// newly created by this publication (plan targets record those as 0).
+	PublishedClusterIDs map[string]int64
 }
 
 type ClusterCandidateGeneration struct {
@@ -175,7 +184,14 @@ func (s *Store) SaveClusterCandidateGeneration(ctx context.Context, in ClusterCa
 	sort.Slice(candidates, func(i, j int) bool { return candidates[i].StableKey < candidates[j].StableKey })
 	published := map[string]int64{}
 	for _, target := range in.Plan.Targets {
-		published[target.CandidateKey] = target.ClusterID
+		id := in.PublishedClusterIDs[target.CandidateKey]
+		if id <= 0 || target.ClusterID != 0 && target.ClusterID != id {
+			return out, fmt.Errorf("cluster candidate generation lacks the published durable identity for a target")
+		}
+		published[target.CandidateKey] = id
+	}
+	if len(in.PublishedClusterIDs) != len(published) {
+		return out, fmt.Errorf("cluster candidate generation published identities do not match the plan targets")
 	}
 	activeOwner := map[int64][]int64{}
 	for _, identity := range in.Plan.Identities {
diff --git a/internal/store/cluster_generations_test.go b/internal/store/cluster_generations_test.go
index 25267ec8..c127a950 100644
--- a/internal/store/cluster_generations_test.go
+++ b/internal/store/cluster_generations_test.go
@@ -73,6 +73,13 @@ func durableHistoryDump(t *testing.T, ctx context.Context, st *Store) string {
 
 func saveGeneration(t *testing.T, ctx context.Context, st *Store, plan ClusterQuarantinePlan, input string) (ClusterCandidateGeneration, error) {
 	t.Helper()
+	published := map[string]int64{}
+	for i, target := range plan.Targets {
+		published[target.CandidateKey] = target.ClusterID
+		if target.ClusterID == 0 {
+			published[target.CandidateKey] = int64(900 + i)
+		}
+	}
 	var generation ClusterCandidateGeneration
 	err := st.WithTx(ctx, func(tx *Store) error {
 		runID, err := tx.insertClusterRun(ctx, plan.RepositoryID, time.Now().UTC().Format(time.RFC3339Nano))
@@ -80,7 +87,7 @@ func saveGeneration(t *testing.T, ctx context.Context, st *Store, plan ClusterQu
 			return err
 		}
 		generation, err = tx.SaveClusterCandidateGeneration(ctx, ClusterCandidateGenerationInput{RepoID: plan.RepositoryID, RunID: runID,
-			Algorithm: "test-algorithm", InputSHA256: strings.Repeat(input, 64), QuarantinePlanSHA256: strings.Repeat("b", 64), Plan: plan})
+			Algorithm: "test-algorithm", InputSHA256: strings.Repeat(input, 64), QuarantinePlanSHA256: strings.Repeat("b", 64), Plan: plan, PublishedClusterIDs: published})
 		return err
 	})
 	return generation, err
@@ -120,6 +127,11 @@ func TestClusterCandidateGenerationsPublishSelectAndRollBack(t *testing.T) {
 		len(merge.Overlaps) != 2 || merge.Overlaps[0] != (ClusterCandidateOverlap{ClusterID: 11, SharedMembers: 2}) || len(merge.Members) != 3 {
 		t.Fatalf("merge candidate = %+v", merge)
 	}
+	for _, candidate := range candidates {
+		if candidate.ComputedKey == "c-new" && (candidate.DurableStatus != "published" || candidate.DurableClusterID != 900) {
+			t.Fatalf("newly published identity not recorded: %+v", candidate)
+		}
+	}
 	if after := durableHistoryDump(t, ctx, st); after != before {
 		t.Fatalf("generation changed durable history:\n%s\n%s", before, after)
 	}
@@ -179,6 +191,9 @@ func TestClusterCandidateGenerationRefusesInvalidPartitions(t *testing.T) {
 		"outside representative": func(p *ClusterQuarantinePlan) {
 			p.Candidates[1].RepresentativeThreadID = ids[0]
 		},
+		"conflicting published identity": func(p *ClusterQuarantinePlan) {
+			p.Targets = []ClusterPublicationTarget{{CandidateKey: "a", ClusterID: -1}}
+		},
 	}
 	for name, mutate := range cases {
 		t.Run(name, func(t *testing.T) {
@@ -244,3 +259,28 @@ func TestClusterCandidateGenerationsHonorOwnerRemoval(t *testing.T) {
 		t.Fatalf("stale owner-excluded input = %v", err)
 	}
 }
+
+func TestPortableRepositoryScopeDropsCandidateGenerations(t *testing.T) {
+	ctx := context.Background()
+	st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	repoID, ids := seedCandidateThreads(t, ctx, st, 2)
+	plan := ClusterQuarantinePlan{RepositoryID: repoID, SelectedIDs: ids, Candidates: []DurableClusterInput{candidateInput("a", ids[0], ids[1])}}
+	if _, err := saveGeneration(t, ctx, st, plan, "a"); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now().UTC().Format(time.RFC3339Nano)
+	if _, err := st.UpsertRepository(ctx, Repository{Owner: "openclaw", Name: "other", FullName: "openclaw/other", RawJSON: "{}", UpdatedAt: now}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.RestrictPortableRepository(ctx, "openclaw/other"); err != nil {
+		t.Fatalf("restrict a snapshot holding another repository's generations: %v", err)
+	}
+	var remaining int
+	if err := st.DB().QueryRowContext(ctx, "SELECT count(*) FROM sqlite_schema WHERE type='table' AND name LIKE 'cluster_candidate%'").Scan(&remaining); err != nil || remaining != 0 {
+		t.Fatalf("portable snapshot retained %d candidate tables: %v", remaining, err)
+	}
+}
diff --git a/internal/store/portable_scope.go b/internal/store/portable_scope.go
index f52f22ba..fd001903 100644
--- a/internal/store/portable_scope.go
+++ b/internal/store/portable_scope.go
@@ -40,6 +40,14 @@ func (s *Store) RestrictPortableRepositoryWithOptions(ctx context.Context, fullN
 		return result, fmt.Errorf("begin portable repository scope: %w", err)
 	}
 	defer func() { _ = tx.Rollback() }()
+	// Computed candidate generations are immutable derived history that no
+	// portable profile publishes. Drop them from this disposable snapshot
+	// before repository filtering, which would otherwise hit their triggers.
+	for _, table := range clusterCandidateGenerationTables() {
+		if _, err := tx.ExecContext(ctx, `drop table if exists `+sqliteIdentifier(table)); err != nil {
+			return result, fmt.Errorf("drop portable %s: %w", table, err)
+		}
+	}
 	var targetCount int64
 	if err := tx.QueryRowContext(ctx, `select count(*) from repositories where full_name = ?`, fullName).Scan(&targetCount); err != nil {
 		return result, fmt.Errorf("count portable target repository: %w", err)
diff --git a/scripts/native-cluster/native_cluster.go b/scripts/native-cluster/native_cluster.go
index 6bbd2d42..bf066199 100644
--- a/scripts/native-cluster/native_cluster.go
+++ b/scripts/native-cluster/native_cluster.go
@@ -428,8 +428,19 @@ func (s *Store) NativePublishClusters(ctx context.Context, want NativeClusterSna
 		// Record the complete computed partition as this run's immutable v3
 		// generation and select it in the same transaction. It never writes the
 		// durable identity tables, so the quarantine postimage stays unchanged.
+		// Resolve every published target to the durable ID actually written in
+		// this run, including identities this publication just created.
+		publishedIDs := map[string]int64{}
+		for i, target := range plan.Targets {
+			var id int64
+			if e = tx.q().QueryRowContext(ctx, "select id from cluster_groups where repo_id=? and stable_key=?", want.RepoID, publish[i].StableKey).Scan(&id); e != nil {
+				return fmt.Errorf("resolve published durable identity: %w", e)
+			}
+			publishedIDs[target.CandidateKey] = id
+		}
 		if _, e = tx.SaveClusterCandidateGeneration(ctx, ClusterCandidateGenerationInput{RepoID: want.RepoID, RunID: result.RunID,
-			Algorithm: NativeClusterAlgorithm, InputSHA256: want.InputSHA256, QuarantinePlanSHA256: proofHash, Plan: plan}); e != nil {
+			Algorithm: NativeClusterAlgorithm, InputSHA256: want.InputSHA256, QuarantinePlanSHA256: proofHash, Plan: plan,
+			PublishedClusterIDs: publishedIDs}); e != nil {
 			return e
 		}
 		if after, e = tx.ClusterQuarantineRowHash(ctx, want.RepoID, plan.PreservedGroupIDs); e != nil {
diff --git a/scripts/native-cluster/quarantine_test.go b/scripts/native-cluster/quarantine_test.go
index dcaa8164..9ad65230 100644
--- a/scripts/native-cluster/quarantine_test.go
+++ b/scripts/native-cluster/quarantine_test.go
@@ -273,6 +273,17 @@ func TestNativeQuarantineWriterRollbackAnalysisAndRecovery(t *testing.T) {
 	save("full_recovery", full.RunID)
 	if computed, e := s.SelectedClusterCandidateGeneration(ctx, repo); e != nil || computed.RunID != full.RunID || computed.CandidateCount != 4 || computed.DurableQuarantinedCandidates != 0 {
 		t.Fatal("full recovery generation", computed, e)
+	} else {
+		candidates, e := s.ClusterCandidates(ctx, repo, computed.ID, 10, false)
+		if e != nil || len(candidates) != 4 {
+			t.Fatal("full recovery candidates", candidates, e)
+		}
+		for _, c := range candidates {
+			var key string
+			if e = s.DB().QueryRow("select stable_key from cluster_groups where id=?", c.DurableClusterID).Scan(&key); e != nil || c.DurableStatus != "published" {
+				t.Fatal("published candidate lacks its durable identity", c, e)
+			}
+		}
 	}
 	if rolled, e := s.RollbackClusterCandidateGeneration(ctx, repo); e != nil || rolled.RunID != blocked.RunID {
 		t.Fatal("computed generation rollback", rolled, e)

From 410d6e60533df36b05e3d36373ac2af63958d349 Mon Sep 17 00:00:00 2001
From: Hannes Rudolph <49103247+hannesrudolph@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:49:00 -0600
Subject: [PATCH 4/5] perf(clusters): create candidate tables on first producer
 write

Keep ordinary store opens free of the new DDL; readers, purge and portable
shaping already treat absent tables as no generations.
---
 internal/store/cluster_generations.go | 3 +++
 internal/store/store.go               | 3 ---
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/internal/store/cluster_generations.go b/internal/store/cluster_generations.go
index 3fbb037a..eebb3f7e 100644
--- a/internal/store/cluster_generations.go
+++ b/internal/store/cluster_generations.go
@@ -24,6 +24,9 @@ func clusterCandidateGenerationTables() []string {
 	return []string{"cluster_candidate_selection", "cluster_candidate_overlaps", "cluster_candidate_members", "cluster_candidates", "cluster_candidate_generations"}
 }
 
+// ensureClusterCandidateGenerationSchema creates the v3 tables on first use by
+// the native producer's transaction. Ordinary opens never pay for this DDL;
+// readers, purge and portable shaping treat absent tables as no generations.
 func (s *Store) ensureClusterCandidateGenerationSchema(ctx context.Context) error {
 	_, err := s.q().ExecContext(ctx, `
 CREATE TABLE IF NOT EXISTS cluster_candidate_generations(
diff --git a/internal/store/store.go b/internal/store/store.go
index b225294f..75d2d602 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -355,9 +355,6 @@ func (s *Store) migrate(ctx context.Context) error {
 	if err := s.ensureThreadExclusionsSchema(ctx); err != nil {
 		return err
 	}
-	if err := s.ensureClusterCandidateGenerationSchema(ctx); err != nil {
-		return err
-	}
 	if err := s.ensureCanonicalObservationTables(ctx); err != nil {
 		return err
 	}

From d7c43db8af3166ea791555439043133f4ecc1453 Mon Sep 17 00:00:00 2001
From: Hannes Rudolph <49103247+hannesrudolph@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:55:34 -0600
Subject: [PATCH 5/5] fix(clusters): read each candidate response in one
 snapshot

Wrap generation listing, selection lookup and candidate reads in one read
transaction so a concurrent rollback or owner purge cannot produce two
selected generations or counts that disagree with members.
---
 internal/cli/cluster_candidates.go | 35 +++++++++++++++++++-----------
 1 file changed, 22 insertions(+), 13 deletions(-)

diff --git a/internal/cli/cluster_candidates.go b/internal/cli/cluster_candidates.go
index e6ea46a0..f2b745bd 100644
--- a/internal/cli/cluster_candidates.go
+++ b/internal/cli/cluster_candidates.go
@@ -69,28 +69,37 @@ func (a *App) runClusterCandidates(ctx context.Context, args []string, readerOnl
 	if err != nil {
 		return err
 	}
-	if *listGenerations {
-		generations, err := rt.Store.ListClusterCandidateGenerations(ctx, repo.ID, limit)
+	// One read transaction gives each response a single database snapshot,
+	// so selection flags, counts and members cannot straddle a rollback or an
+	// owner purge.
+	var generations []store.ClusterCandidateGeneration
+	var generation store.ClusterCandidateGeneration
+	var candidates []store.ClusterCandidate
+	err = rt.Store.WithTx(ctx, func(tx *store.Store) error {
+		var err error
+		if *listGenerations {
+			generations, err = tx.ListClusterCandidateGenerations(ctx, repo.ID, limit)
+			return err
+		}
+		if generationID != 0 {
+			generation, err = tx.ClusterCandidateGenerationByID(ctx, repo.ID, int64(generationID))
+		} else {
+			generation, err = tx.SelectedClusterCandidateGeneration(ctx, repo.ID)
+		}
 		if err != nil {
 			return err
 		}
-		return a.writeOutput("cluster-candidates", map[string]any{"repository": repo.FullName, "format": store.ClusterCandidateFormatV3, "generations": generations}, false)
-	}
-	var generation store.ClusterCandidateGeneration
-	if generationID != 0 {
-		generation, err = rt.Store.ClusterCandidateGenerationByID(ctx, repo.ID, int64(generationID))
-	} else {
-		generation, err = rt.Store.SelectedClusterCandidateGeneration(ctx, repo.ID)
-	}
+		candidates, err = tx.ClusterCandidates(ctx, repo.ID, generation.ID, limit, *members)
+		return err
+	})
 	if errors.Is(err, store.ErrNoClusterCandidateGeneration) {
 		return fmt.Errorf("%w for %s", err, repo.FullName)
 	}
 	if err != nil {
 		return err
 	}
-	candidates, err := rt.Store.ClusterCandidates(ctx, repo.ID, generation.ID, limit, *members)
-	if err != nil {
-		return err
+	if *listGenerations {
+		return a.writeOutput("cluster-candidates", map[string]any{"repository": repo.FullName, "format": store.ClusterCandidateFormatV3, "generations": generations}, false)
 	}
 	return a.writeOutput("cluster-candidates", map[string]any{
 		"repository": repo.FullName,