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..be2249aa 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, 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 + 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 [--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 [...]` | 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..f2b745bd --- /dev/null +++ b/internal/cli/cluster_candidates.go @@ -0,0 +1,113 @@ +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 + } + // 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 + } + 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 + } + 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, + "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..2dada0d8 --- /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, PublishedClusterIDs: map[string]int64{"b": 7}}) + 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..eebb3f7e --- /dev/null +++ b/internal/store/cluster_generations.go @@ -0,0 +1,528 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "sort" + "strconv" + "time" +) + +// ClusterCandidateFormatV3 identifies an immutable, run-scoped computed +// candidate generation. Candidate IDs (":") 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") + +// 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"} +} + +// 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( + 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: 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" { + 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 { + 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 + // 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 { + 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 { + 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 { + 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(¤t); 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 id0 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..c127a950 --- /dev/null +++ b/internal/store/cluster_generations_test.go @@ -0,0 +1,286 @@ +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() + 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)) + 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, PublishedClusterIDs: published}) + 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) + } + 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) + } + 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] + }, + "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) { + 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) + } +} + +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_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/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/internal/store/thread_exclusions.go b/internal/store/thread_exclusions.go index f440ee1a..35550d43 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 } @@ -487,6 +488,13 @@ func (s *Store) PurgeThreads(ctx context.Context, repository string, numbers []i 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; 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 + } + } } // Never turn removal of unavailable work into claimed provider completeness. if tx.hasTable(ctx, "analytics_review_state_coverage") { 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..bf066199 100644 --- a/scripts/native-cluster/native_cluster.go +++ b/scripts/native-cluster/native_cluster.go @@ -425,6 +425,30 @@ 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. + // 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, + PublishedClusterIDs: publishedIDs}); 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..9ad65230 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,23 @@ 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) + } 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) + } if path := os.Getenv("GITCRAWL_QUARANTINE_FINAL_DB"); path != "" { if _, err := s.DB().Exec("VACUUM INTO ?", path); err != nil { t.Fatal(err)