From e2d1dd402c2740e162d5dcc0997886b77b9f512b Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 10 Sep 2026 01:22:20 +0800 Subject: [PATCH] feat(cloud): publish archives with explicit admission evidence --- CHANGELOG.md | 1 + docs/cloud-archives.md | 48 ++- go.mod | 2 +- go.sum | 4 +- internal/cli/app.go | 2 +- internal/cli/app_test.go | 77 +++- internal/cli/cloud_admission.go | 354 +++++++++++++++++++ internal/cli/cloud_admission_test.go | 483 ++++++++++++++++++++++++++ internal/cli/cloud_commands.go | 79 ++++- internal/cli/cloud_commands_test.go | 28 ++ internal/cli/cloud_snapshot.go | 71 +++- internal/cli/cloud_snapshot_test.go | 6 +- internal/store/archive_source.go | 208 +++++++++++ internal/store/archive_source_test.go | 172 +++++++++ 14 files changed, 1485 insertions(+), 50 deletions(-) create mode 100644 internal/cli/cloud_admission.go create mode 100644 internal/cli/cloud_admission_test.go create mode 100644 internal/store/archive_source.go create mode 100644 internal/store/archive_source_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c71e7cd4..5888d30b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Add explicit `cloud publish --admission-policy=archive-v1 --observation-order` with frozen source, integrity, and enrichment evidence and immutable warnings. Preserve strict publication defaults, canonical archive contents, privacy, and existing size limits. - Preserve newly fetched GitHub numeric IDs exactly across REST and GraphQL syncs, including IDs above the floating-point precision limit. Previously stored IDs are not repaired. ## 0.9.5 - 2026-09-09 diff --git a/docs/cloud-archives.md b/docs/cloud-archives.md index 53b580e3..a10a0d87 100644 --- a/docs/cloud-archives.md +++ b/docs/cloud-archives.md @@ -91,12 +91,58 @@ gitcrawl cloud publish \ --json ``` -A later publish verifies the candidate through the publisher-only status projection. It skips repeated ingest only when the digest, source sync, schema, resolved publication profile, generation timestamp, and coverage match. +A later publish verifies the candidate through the publisher-only status projection. It skips repeated ingest only when the digest, source sync, schema, resolved publication profile, immutable warnings, generation timestamp, and coverage match. Cutover requires reader-authenticated `GET /sqlite`. Gitcrawl validates the cutover acknowledgement, polls the scoped reader projection until its digest, profile, generation, and dataset coverage match, rechecks the publisher metadata, downloads the bound SQLite image, and verifies its hash before reporting success. Without `--stage-only`, a successful publish moves unpinned reads to the complete snapshot. Incomplete local enrichment fails before remote mutation. `--allow-incomplete` is the explicit override. `--observation-order` publishes durable fetch ordering only after the remote operator fence is enabled. +## Archive admission + +To publish a raw archive with explicitly reported source and enrichment gaps: + +```bash +gitcrawl cloud publish \ + --remote URL \ + --archive gitcrawl/openclaw__openclaw \ + --admission-policy=archive-v1 \ + --observation-order \ + --stage-only \ + --json +``` + +This opt-in policy requires the remote capabilities +`gitcrawl.archive-admission.v1` and `gitcrawl.observation-order.v1`. +An older remote or a disabled observation fence fails preflight before upload. +Do not combine the policy with `--allow-incomplete`. Omitting the policy retains +the strict default. + +Archive admission still requires SQLite integrity, compatible canonical tables, +repositories, referential closure, full bodies, and the existing privacy scrub. +Use the full runtime archive, not a lossy portable export. Native portable +profiles that declare excerpts or excluded patches/history are rejected even +when current bodies fit the excerpt limit. Empty patch rows in a full runtime +are not proof of loss. The existing +4 GiB SQLite, 512 MiB gzip, eight-part, staging, and cutover limits still apply. +Admission does not request summaries, embeddings, or other model work. + +The JSON result includes typed `admission` evidence and stable `warnings`. +Repository inventory observations distinguish unsupported, unknown, missing, +partial, complete, and empty observations. Child observations also report stale +reservations relative to the archived thread. Complete observations describe +the recorded scope, not current GitHub state. Workflow freshness and current +remote freshness remain unknown; export clocks never establish either. +Source gaps, PR detail/file gaps, and all six enrichment metrics remain visible. +Incomplete current revisions retain their actual coverage counts and +`complete: false`; admission does not make stale enrichment current for readers. + +The sanitized SQLite copy stores the same deterministic evidence in +`portable_metadata.cloud_admission_v1` before hashing. Existing source markers +are not trusted, and strict exports remove them. The original database is not +changed. Canonically sorted warnings are part of immutable snapshot identity, +including stage-only replay and reader verification. Warnings alone never +enable admission. + ## Privacy and retention Digest-scoped bundles can contain private issue and pull request text. Bundle metadata declares both message-body and source-code sensitivity because patch text is retained. diff --git a/go.mod b/go.mod index f8e1ee21..16550af1 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/charmbracelet/lipgloss v1.1.0 github.com/charmbracelet/x/ansi v0.11.8 github.com/mattn/go-isatty v0.0.24 - github.com/openclaw/crawlkit v0.15.0 + github.com/openclaw/crawlkit v0.15.1 github.com/zalando/go-keyring v0.2.8 golang.org/x/sys v0.47.0 modernc.org/sqlite v1.58.0 diff --git a/go.sum b/go.sum index eedb839c..06f1cbb7 100644 --- a/go.sum +++ b/go.sum @@ -62,8 +62,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= -github.com/openclaw/crawlkit v0.15.0 h1:KAmdew2UQPZm/gOfqiLw/WhhDY7Y2UZj0EcgQ5s7dl4= -github.com/openclaw/crawlkit v0.15.0/go.mod h1:JhEsXnoxozd2qp1p4Dw8ZhKzQoYm3WbrlvO0ILVx8cs= +github.com/openclaw/crawlkit v0.15.1 h1:B281AyJgVCTj7bMuONW57DXNV4sQyj9F5LLEaUc71l0= +github.com/openclaw/crawlkit v0.15.1/go.mod h1:JhEsXnoxozd2qp1p4Dw8ZhKzQoYm3WbrlvO0ILVx8cs= github.com/pelletier/go-toml/v2 v2.4.3 h1:GTRvJQutkOSftxIFD5xw9aepkYNuPWmVJpffdDPYVpY= github.com/pelletier/go-toml/v2 v2.4.3/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= diff --git a/internal/cli/app.go b/internal/cli/app.go index 00fb5c80..96ebf04e 100644 --- a/internal/cli/app.go +++ b/internal/cli/app.go @@ -5406,7 +5406,7 @@ Usage: "cloud": `gitcrawl cloud manages Worker-backed remote archives. Usage: - gitcrawl cloud publish --remote URL --archive id [--allow-incomplete] [--observation-order] [--stage-only] [--json] + gitcrawl cloud publish --remote URL --archive id [--allow-incomplete | --admission-policy=archive-v1] [--observation-order] [--stage-only] [--json] `, "whoami": `gitcrawl whoami prints the configured remote archive identity. diff --git a/internal/cli/app_test.go b/internal/cli/app_test.go index 29791b8b..7549d27c 100644 --- a/internal/cli/app_test.go +++ b/internal/cli/app_test.go @@ -437,7 +437,7 @@ func TestCloudSQLiteSnapshotDropsLocalCodeCorpus(t *testing.T) { `, repoID, repoID, repoID, repoID, repoID, repoID); err != nil { t.Fatalf("seed private cloud payloads: %v", err) } - snapshotPath, cleanup, err := cloudSQLiteSnapshotPath(ctx, st.DB(), dbPath) + snapshotPath, _, cleanup, err := cloudSQLiteSnapshotPath(ctx, st.DB(), dbPath, gitcrawlCloudPublishOptions{}) if err != nil { t.Fatalf("cloud snapshot: %v", err) } @@ -1083,6 +1083,11 @@ func TestGitcrawlPublisherStatusMatchesExactMetadata(t *testing.T) { } func TestCloudPublishSendsLocalRows(t *testing.T) { + t.Run("legacy allow incomplete", func(t *testing.T) { testCloudPublishSendsLocalRows(t, false) }) + t.Run("archive admission", func(t *testing.T) { testCloudPublishSendsLocalRows(t, true) }) +} + +func testCloudPublishSendsLocalRows(t *testing.T, archiveAdmission bool) { ctx := context.Background() dir := t.TempDir() cfgPath := filepath.Join(dir, "config.toml") @@ -1107,10 +1112,21 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { var publishedDatasets []crawlremote.DatasetCoverage var publisherStatusSnapshotIDs []string mutationCounter := 0 + uploadRequests := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodGet && r.URL.EscapedPath() == "/v1/contract" { w.Header().Set("content-type", "application/json") - _ = json.NewEncoder(w).Encode(testSnapshotPublishContract()) + contract := testSnapshotPublishContract() + if archiveAdmission { + contract.Apps[0].Capabilities = append(contract.Apps[0].Capabilities, gitcrawlArchiveAdmissionCapability, gitcrawlObservationOrderCapability) + for index := range contract.Apps[0].IngestTables { + table := &contract.Apps[0].IngestTables[index] + if table.Name == "threads" || table.Name == "thread_revisions" { + table.Columns = append(table.Columns, "observation_sequence") + } + } + } + _ = json.NewEncoder(w).Encode(contract) return } if got := r.Header.Get("authorization"); got != "Bearer publish-token" { @@ -1127,6 +1143,7 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { return } if r.Method == http.MethodPut && r.URL.EscapedPath() == "/v1/apps/gitcrawl/archives/gitcrawl%2Fopenclaw__openclaw/sqlite" { + uploadRequests++ uploadKind := r.Header.Get("x-crawl-sqlite-upload") payload, err := io.ReadAll(r.Body) if err != nil { @@ -1338,6 +1355,7 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { SourceSyncAt: body.Manifest.SourceSyncAt, DatasetGeneratedAt: fmt.Sprint(body.Rows[0][5]), CoverageComplete: true, + Warnings: slices.Clone(body.Manifest.Warnings), } } seenTables[body.Table] = body @@ -1360,15 +1378,20 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { app := New() var out bytes.Buffer app.Stdout = &out - if err := app.Run(ctx, []string{ + args := []string{ "--config", cfgPath, "cloud", "publish", "--remote", server.URL, "--archive", "gitcrawl/openclaw__openclaw", "--token-env", tokenEnv, - "--allow-incomplete", "--json", - }); err != nil { + } + if archiveAdmission { + args = append(args, "--admission-policy=archive-v1", "--observation-order") + } else { + args = append(args, "--allow-incomplete") + } + if err := app.Run(ctx, args); err != nil { t.Fatalf("cloud publish: %v", err) } @@ -1425,6 +1448,24 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { privacy["includes_source_code"] != true { t.Fatalf("missing sqlite bundle privacy output: %#v", payload) } + if archiveAdmission { + if len(publishedSnapshot.Warnings) == 0 || payload["admission"] == nil { + t.Fatal("archive publication lost persistent admission evidence") + } + for _, dataset := range publishedDatasets { + if dataset.Dataset == "thread_revisions" && (dataset.EligibleCount != 3 || dataset.Complete) { + t.Fatalf("archive admission hid incomplete revision coverage: %+v", dataset) + } + } + } + beforeMutations, beforeUploads := mutationCounter, uploadRequests + out.Reset() + if err := app.Run(ctx, append(args, "--stage-only")); err != nil { + t.Fatalf("stage-only replay: %v", err) + } + if mutationCounter != beforeMutations || uploadRequests != beforeUploads { + t.Fatal("identical staged snapshot was uploaded or ingested again") + } } func TestCloudPublishRejectsMissingSnapshotCapabilityBeforeUpload(t *testing.T) { @@ -1480,7 +1521,20 @@ func TestCloudPublishRejectsMissingRequestedCapabilityBeforeUpload(t *testing.T) name string args []string missingCapability string + archiveAdmission bool }{ + { + name: "archive admission", + args: []string{"--admission-policy=archive-v1", "--observation-order", "--stage-only"}, + missingCapability: gitcrawlArchiveAdmissionCapability, + archiveAdmission: true, + }, + { + name: "archive observation fence", + args: []string{"--admission-policy=archive-v1", "--observation-order", "--stage-only"}, + missingCapability: gitcrawlObservationOrderCapability, + archiveAdmission: true, + }, { name: "observation order", args: []string{"--observation-order", "--stage-only"}, @@ -1515,6 +1569,15 @@ func TestCloudPublishRejectsMissingRequestedCapabilityBeforeUpload(t *testing.T) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodGet && r.URL.EscapedPath() == "/v1/contract" { contract := testSnapshotPublishContract() + if test.archiveAdmission { + contract.Apps[0].Capabilities = append(contract.Apps[0].Capabilities, gitcrawlArchiveAdmissionCapability, gitcrawlObservationOrderCapability) + for index := range contract.Apps[0].IngestTables { + table := &contract.Apps[0].IngestTables[index] + if table.Name == "threads" || table.Name == "thread_revisions" { + table.Columns = append(table.Columns, "observation_sequence") + } + } + } capabilities := make([]string, 0, len(contract.Apps[0].Capabilities)) for _, capability := range contract.Apps[0].Capabilities { if capability != test.missingCapability { @@ -1537,9 +1600,11 @@ func TestCloudPublishRejectsMissingRequestedCapabilityBeforeUpload(t *testing.T) "--remote", server.URL, "--archive", "gitcrawl/openclaw__openclaw", "--token-env", tokenEnv, - "--allow-incomplete", "--json", } + if !test.archiveAdmission { + args = append(args, "--allow-incomplete") + } args = append(args, test.args...) err := New().Run(ctx, args) if err == nil || !strings.Contains(err.Error(), test.missingCapability) { diff --git a/internal/cli/cloud_admission.go b/internal/cli/cloud_admission.go new file mode 100644 index 00000000..c2517ef2 --- /dev/null +++ b/internal/cli/cloud_admission.go @@ -0,0 +1,354 @@ +package cli + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "slices" + "strconv" + "strings" + "time" + + portableexport "github.com/openclaw/gitcrawl/internal/portable" + crawlstore "github.com/openclaw/gitcrawl/internal/store" +) + +const ( + gitcrawlArchiveAdmissionPolicy = "archive-v1" + gitcrawlArchiveAdmissionCapability = "gitcrawl.archive-admission.v1" + gitcrawlCloudAdmissionMetadataKey = "cloud_admission_v1" + gitcrawlCloudWarningsMaxElements = 32 + gitcrawlCloudWarningsMaxBytes = 8192 +) + +type gitcrawlCloudPublishOptions struct { + AllowIncomplete bool + ObservationOrder bool + AdmissionPolicy string +} + +type gitcrawlCloudIntegrity struct { + SQLite bool `json:"sqlite"` + CompatibleSchema bool `json:"compatible_schema"` + RequiredData bool `json:"required_data"` + ReferentialClosure bool `json:"referential_closure"` + FullBodies bool `json:"full_bodies"` + Privacy bool `json:"privacy"` +} + +type gitcrawlCloudAdmission struct { + Policy string `json:"policy"` + Integrity gitcrawlCloudIntegrity `json:"integrity"` + Source crawlstore.ArchiveSourceAssessment `json:"source"` + Enrichment crawlstore.EnrichmentCoverage `json:"enrichment"` + Warnings []string `json:"warnings"` + datasets []gitcrawlCloudDataset +} + +func assessGitcrawlCloudArchive(ctx context.Context, path string) (*gitcrawlCloudAdmission, error) { + st, err := crawlstore.OpenReadOnly(ctx, path) + if err != nil { + return nil, fmt.Errorf("open frozen archive for admission: %w", err) + } + defer st.Close() + if err := requireLosslessGitcrawlCloudSource(ctx, st.DB()); err != nil { + return nil, err + } + coverage, err := st.ArchiveCoverage(ctx, crawlstore.ArchiveCoverageOptions{}) + if err != nil { + return nil, err + } + source, err := st.ArchiveSourceObservations(ctx, coverage) + if err != nil { + return nil, err + } + admission := &gitcrawlCloudAdmission{ + Policy: gitcrawlArchiveAdmissionPolicy, Source: source, Enrichment: coverage.Totals.Enrichment, + } + admission.Warnings, err = gitcrawlArchiveWarnings(*admission) + return admission, err +} + +func requireLosslessGitcrawlCloudSource(ctx context.Context, db *sql.DB) error { + if exists, err := sqliteTableExists(ctx, db, "portable_metadata"); err != nil { + return err + } else if !exists { + return nil + } + rows, err := db.QueryContext(ctx, `select key, value from portable_metadata + where key in ('profile', 'body_chars', 'capabilities', 'excluded')`) + if err != nil { + return fmt.Errorf("read archive source profile: %w", err) + } + defer rows.Close() + for rows.Next() { + var key, value string + if err := rows.Scan(&key, &value); err != nil { + return fmt.Errorf("scan archive source profile: %w", err) + } + // Current bodies can fit the excerpt limit even after history/review + // bodies or patches were lost. Native profile declarations retain that + // evidence; empty patch rows alone do not establish loss. + lossy := false + switch key { + case "profile": + lossy = value == portableexport.CurrentStateV1 + case "body_chars": + limit, err := strconv.Atoi(value) + lossy = err == nil && limit > 0 + case "capabilities": + values := strings.Split(value, ",") + lossy = slices.Contains(values, "body_excerpts") || slices.Contains(values, "comment_excerpts") + case "excluded": + values := strings.Split(value, ",") + lossy = slices.Contains(values, "pull_request_file_patches") || slices.Contains(values, "comment_revision_history") + } + if lossy { + return fmt.Errorf("archive source declares a lossy portable profile; use the full runtime archive") + } + } + return rows.Err() +} + +func gitcrawlArchiveWarnings(admission gitcrawlCloudAdmission) ([]string, error) { + unknown := admission.Source.RemoteFreshness != crawlstore.ArchiveObservationComplete + incomplete := false + for _, repo := range admission.Source.Repositories { + switch repo.Inventory.State { + case crawlstore.ArchiveObservationUnsupported, crawlstore.ArchiveObservationUnknown: + unknown = true + case crawlstore.ArchiveObservationMissing, crawlstore.ArchiveObservationStale, crawlstore.ArchiveObservationPartial: + incomplete = true + } + unknown = unknown || repo.WorkflowFreshness == crawlstore.ArchiveObservationUnknown || + repo.WorkflowFreshness == crawlstore.ArchiveObservationUnsupported || repo.FailedHydrations == nil + incomplete = incomplete || (repo.FailedHydrations != nil && *repo.FailedHydrations > 0) + for _, child := range repo.Children { + unknown = unknown || child.State == crawlstore.ArchiveObservationUnsupported || child.Unknown > 0 + incomplete = incomplete || child.Missing > 0 || child.Stale > 0 + } + for _, metric := range []crawlstore.EnrichmentCoverageMetric{repo.PRDetails, repo.PRFiles} { + unknown = unknown || !metric.Supported + incomplete = incomplete || (metric.Supported && !metric.Complete) + } + } + warnings := []string{} + if unknown { + warnings = append(warnings, "gitcrawl.archive.source.unknown") + } + if incomplete { + warnings = append(warnings, "gitcrawl.archive.source.incomplete") + } + for name, metric := range map[string]crawlstore.EnrichmentCoverageMetric{ + "revisions": admission.Enrichment.Revisions, "fingerprints": admission.Enrichment.Fingerprints, + "summaries": admission.Enrichment.Summaries, "clusters": admission.Enrichment.Clusters, + "pr_details": admission.Enrichment.PRDetails, "pr_files": admission.Enrichment.PRFiles, + } { + if !metric.Supported || !metric.Complete { + warnings = append(warnings, "gitcrawl.archive.enrichment."+name+".incomplete") + } + } + return canonicalGitcrawlCloudWarnings(warnings) +} + +// Use the oldest repository's actual last observation, or unknown if any is +// missing. Export clocks and another repository's newer run are not evidence. +func (admission gitcrawlCloudAdmission) sourceSyncAt() string { + var oldest time.Time + for _, repo := range admission.Source.Repositories { + if !repo.Inventory.Successful { + return "" + } + at, err := time.Parse(time.RFC3339Nano, repo.Inventory.FinishedAt) + if err != nil { + return "" + } + if oldest.IsZero() || at.Before(oldest) { + oldest = at + } + } + if oldest.IsZero() { + return "" + } + return oldest.UTC().Format(time.RFC3339Nano) +} + +func validateGitcrawlArchiveIntegrity(ctx context.Context, db *sql.DB, admission *gitcrawlCloudAdmission) error { + var result string + if err := db.QueryRowContext(ctx, `pragma integrity_check(1)`).Scan(&result); err != nil { + return fmt.Errorf("check archive SQLite integrity: %w", err) + } + if result != "ok" { + return fmt.Errorf("archive SQLite integrity check failed") + } + admission.Integrity.SQLite = true + if _, err := gitcrawlCloudCapabilities(ctx, db, true); err != nil { + return err + } + for table, columns := range map[string][]string{ + "comments": {"id", "thread_id", "body"}, + "comment_revisions": {"id", "comment_id", "body"}, + "pull_request_files": {"thread_id", "patch"}, + "pull_request_review_threads": {"thread_id", "review_thread_id", "first_comment_body"}, + "pull_request_review_thread_revisions": {"thread_id", "review_thread_id", "first_comment_body"}, + } { + if ok, err := sqliteTableHasColumns(ctx, db, table, columns...); err != nil { + return err + } else if !ok { + return fmt.Errorf("archive requires canonical %s schema", table) + } + } + datasets, err := loadGitcrawlCloudDatasets(ctx, db, true, admission.Enrichment) + if err != nil { + return err + } + admission.Integrity.CompatibleSchema = true + admission.datasets = datasets + if len(datasets) == 0 || datasets[0].RowCount == 0 { + return fmt.Errorf("cloud snapshot has no repositories") + } + admission.Integrity.RequiredData = true + // Check declared FKs and semantic repository ownership independently. + // Permissive imported schemas must not hide orphan or cross-repo rows. + for _, query := range []string{ + `select exists(select 1 from pragma_foreign_key_check)`, + `select exists(select 1 from threads t left join repositories r on r.id = t.repo_id where r.id is null)`, + `select exists(select 1 from comments c left join threads t on t.id = c.thread_id where t.id is null)`, + `select exists(select 1 from comment_revisions r left join comments c on c.id = r.comment_id where c.id is null)`, + `select exists(select 1 from pull_request_review_threads r left join threads t on t.id = r.thread_id where t.id is null or t.kind != 'pull_request')`, + `select exists(select 1 from pull_request_review_thread_revisions r left join threads t on t.id = r.thread_id where t.id is null or t.kind != 'pull_request')`, + `select exists(select 1 from thread_revisions r left join threads t on t.id = r.thread_id where t.id is null)`, + `select exists(select 1 from thread_fingerprints f left join thread_revisions r on r.id = f.thread_revision_id where r.id is null)`, + `select exists(select 1 from thread_key_summaries s left join thread_revisions r on r.id = s.thread_revision_id where r.id is null)`, + `select exists(select 1 from cluster_groups g left join repositories r on r.id = g.repo_id + left join threads t on t.id = g.representative_thread_id + where r.id is null or (g.representative_thread_id is not null and (t.id is null or t.repo_id != g.repo_id)))`, + `select exists(select 1 from cluster_memberships m left join cluster_groups g on g.id = m.cluster_id + left join threads t on t.id = m.thread_id where g.id is null or t.id is null or t.repo_id != g.repo_id)`, + `select exists(select 1 from pull_request_details d left join threads t on t.id = d.thread_id + where t.id is null or t.repo_id != d.repo_id or t.number != d.number or t.kind != 'pull_request')`, + `select exists(select 1 from pull_request_files f left join threads t on t.id = f.thread_id + where t.id is null or t.kind != 'pull_request')`, + } { + var invalid bool + if err := db.QueryRowContext(ctx, query).Scan(&invalid); err != nil { + return fmt.Errorf("check archive referential closure: %w", err) + } + if invalid { + return fmt.Errorf("archive referential closure check failed") + } + } + admission.Integrity.ReferentialClosure = true + for _, table := range []string{"threads", "comments"} { + ok, err := sqliteTableHasColumns(ctx, db, table, "body") + if err != nil { + return err + } + if !ok { + return fmt.Errorf("archive requires canonical %s bodies", table) + } + if metadata, err := sqliteColumnExists(ctx, db, table, "body_length"); err != nil { + return err + } else if metadata { + var truncated bool + if err := db.QueryRowContext(ctx, `select exists(select 1 from `+table+ + ` where body_length > length(coalesce(body, '')))`).Scan(&truncated); err != nil { + return err + } + if truncated { + return fmt.Errorf("archive contains truncated %s bodies; use the full runtime archive", table) + } + } + } + admission.Integrity.FullBodies = true + return nil +} + +func writeGitcrawlCloudAdmission(ctx context.Context, db *sql.DB, admission *gitcrawlCloudAdmission) error { + if admission == nil { + if exists, err := sqliteTableExists(ctx, db, "portable_metadata"); err != nil { + return err + } else if exists { + _, err := db.ExecContext(ctx, `delete from portable_metadata where key = ?`, gitcrawlCloudAdmissionMetadataKey) + return err + } + return nil + } + encoded, err := json.Marshal(admission) + if err != nil { + return err + } + // This is the existing portable artifact metadata table, only in the copy. + if _, err := db.ExecContext(ctx, `create table if not exists portable_metadata ( + key text primary key, value text not null + )`); err != nil { + return err + } + _, err = db.ExecContext(ctx, `insert into portable_metadata(key, value) values(?, ?) + on conflict(key) do update set value = excluded.value`, gitcrawlCloudAdmissionMetadataKey, string(encoded)) + return err +} + +func (options gitcrawlCloudPublishOptions) validate() error { + switch options.AdmissionPolicy { + case "": + return nil + case gitcrawlArchiveAdmissionPolicy: + if options.AllowIncomplete { + return fmt.Errorf("--admission-policy cannot be combined with --allow-incomplete") + } + if !options.ObservationOrder { + return fmt.Errorf("--admission-policy=archive-v1 requires --observation-order") + } + return nil + default: + return fmt.Errorf("unsupported admission policy; use archive-v1 or omit --admission-policy") + } +} + +func gitcrawlCloudWarningAllowed(value string) bool { + switch value { + case "gitcrawl.archive.source.incomplete", + "gitcrawl.archive.source.unknown", + "gitcrawl.archive.enrichment.revisions.incomplete", + "gitcrawl.archive.enrichment.fingerprints.incomplete", + "gitcrawl.archive.enrichment.summaries.incomplete", + "gitcrawl.archive.enrichment.clusters.incomplete", + "gitcrawl.archive.enrichment.pr_details.incomplete", + "gitcrawl.archive.enrichment.pr_files.incomplete": + return true + default: + return false + } +} + +func canonicalGitcrawlCloudWarnings(values []string) ([]string, error) { + if len(values) > gitcrawlCloudWarningsMaxElements { + return nil, fmt.Errorf("cloud warnings exceed %d elements", gitcrawlCloudWarningsMaxElements) + } + canonical := append([]string{}, values...) + encoded, err := json.Marshal(canonical) + if err != nil || len(encoded) > gitcrawlCloudWarningsMaxBytes { + return nil, fmt.Errorf("cloud warnings exceed %d JSON bytes", gitcrawlCloudWarningsMaxBytes) + } + seen := make(map[string]struct{}, len(canonical)) + for _, value := range canonical { + if !gitcrawlCloudWarningAllowed(value) { + return nil, fmt.Errorf("cloud warnings contain an unsupported code") + } + if _, duplicate := seen[value]; duplicate { + return nil, fmt.Errorf("cloud warnings contain a duplicate code") + } + seen[value] = struct{}{} + } + // Warning order is immutable snapshot identity, not display order. + slices.Sort(canonical) + return canonical, nil +} + +func gitcrawlCloudWarningsMatch(left, right []string) bool { + leftCanonical, leftErr := canonicalGitcrawlCloudWarnings(left) + rightCanonical, rightErr := canonicalGitcrawlCloudWarnings(right) + return leftErr == nil && rightErr == nil && slices.Equal(leftCanonical, rightCanonical) +} diff --git a/internal/cli/cloud_admission_test.go b/internal/cli/cloud_admission_test.go new file mode 100644 index 00000000..4153b0bb --- /dev/null +++ b/internal/cli/cloud_admission_test.go @@ -0,0 +1,483 @@ +package cli + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "slices" + "strings" + "sync/atomic" + "testing" + + "github.com/openclaw/gitcrawl/internal/config" + portableexport "github.com/openclaw/gitcrawl/internal/portable" + crawlstore "github.com/openclaw/gitcrawl/internal/store" +) + +func TestGitcrawlCloudAdmissionPolicy(t *testing.T) { + for _, options := range []gitcrawlCloudPublishOptions{ + {}, + {AllowIncomplete: true}, + {ObservationOrder: true}, + {AdmissionPolicy: gitcrawlArchiveAdmissionPolicy, ObservationOrder: true}, + } { + if err := options.validate(); err != nil { + t.Errorf("valid options %+v: %v", options, err) + } + } + for _, options := range []gitcrawlCloudPublishOptions{ + {AdmissionPolicy: "unknown", ObservationOrder: true}, + {AdmissionPolicy: gitcrawlArchiveAdmissionPolicy}, + {AdmissionPolicy: gitcrawlArchiveAdmissionPolicy, ObservationOrder: true, AllowIncomplete: true}, + } { + if err := options.validate(); err == nil { + t.Errorf("invalid options accepted: %+v", options) + } + } +} + +func TestCanonicalGitcrawlCloudWarnings(t *testing.T) { + values := []string{ + "gitcrawl.archive.source.unknown", + "gitcrawl.archive.enrichment.revisions.incomplete", + } + before := slices.Clone(values) + got, err := canonicalGitcrawlCloudWarnings(values) + if err != nil { + t.Fatal(err) + } + if !slices.Equal(values, before) || !slices.Equal(got, []string{before[1], before[0]}) { + t.Fatalf("canonical=%v input=%v", got, values) + } + for _, empty := range [][]string{nil, {}} { + got, err := canonicalGitcrawlCloudWarnings(empty) + encoded, _ := json.Marshal(got) + if err != nil || string(encoded) != "[]" { + t.Fatalf("empty warnings = %s, %v", encoded, err) + } + } + for name, invalid := range map[string][]string{ + "unknown": {"arbitrary provider text"}, + "duplicate": {values[0], values[0]}, + "count": make([]string, gitcrawlCloudWarningsMaxElements+1), + "bytes": {strings.Repeat("x", gitcrawlCloudWarningsMaxBytes)}, + "json escaping": {strings.Repeat("\n", gitcrawlCloudWarningsMaxBytes/2)}, + "UTF-8 bytes": {strings.Repeat("\u754c", gitcrawlCloudWarningsMaxBytes/3)}, + } { + t.Run(name, func(t *testing.T) { + if _, err := canonicalGitcrawlCloudWarnings(invalid); err == nil { + t.Fatal("invalid warnings accepted") + } + }) + } +} + +func TestArchiveAdmissionFlagsRejectBeforeOpeningRuntime(t *testing.T) { + for _, flags := range [][]string{ + {"--admission-policy=unknown"}, + {"--admission-policy=archive-v1"}, + {"--admission-policy=archive-v1", "--observation-order", "--allow-incomplete"}, + } { + args := append([]string{"--config", filepath.Join(t.TempDir(), "missing.toml"), "cloud", "publish"}, flags...) + if err := New().Run(context.Background(), args); err == nil || !strings.Contains(err.Error(), "admission") { + t.Fatalf("flags %v did not reject policy first: %v", flags, err) + } + } +} + +func seedArchiveAdmissionFixture(t *testing.T) string { + t.Helper() + path := filepath.Join(t.TempDir(), "source.db") + st, err := crawlstore.Open(context.Background(), path) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if _, err := st.DB().Exec(` + insert into repositories(id, owner, name, full_name, raw_json, updated_at) + values(1,'example','repo','example/repo','{"private":"private-admission-marker"}','2026-09-01T00:00:00Z'); + insert into threads(id, repo_id, github_id, number, kind, state, title, body, html_url, + labels_json, assignees_json, raw_json, content_hash, updated_at_gh, observation_sequence, updated_at) + values(1,1,'9007199254740993',1,'pull_request','open','Archive','Full canonical body', + 'https://github.com/example/repo/pull/1','[]','[]','{}','body-hash','2026-09-01T00:00:00Z',1,'2026-09-01T00:00:00Z'); + insert into comments(id, thread_id, github_id, comment_type, body, raw_json) + values(1,1,'9007199254740995','issue','Full canonical comment','{"private":"private-admission-marker"}'); + insert into comment_revisions(id, comment_id, body, raw_json, recorded_at) + values(1,1,'Historical canonical body','{"private":"private-admission-marker"}','2026-09-01T00:00:00Z'); + insert into pull_request_files(thread_id, position, path, patch, raw_json, fetched_at) + values(1,0,'example.go','@@ -1 +1 @@ original patch','{}','2026-09-01T00:00:00Z'); + create table portable_metadata(key text primary key, value text not null); + insert into portable_metadata(key,value) values('cloud_admission_v1','untrusted-old-marker'), + ('exported_at','2099-01-01T00:00:00Z'), ('source_path','private-admission-marker'); + `); err != nil { + t.Fatal(err) + } + return path +} + +func TestArchiveAdmissionFreezesDeterministicEvidence(t *testing.T) { + ctx := context.Background() + sourcePath := seedArchiveAdmissionFixture(t) + db, err := sql.Open("sqlite", sourcePath) + if err != nil { + t.Fatal(err) + } + defer db.Close() + sourceHash, err := cloudFileSHA256(sourcePath) + if err != nil { + t.Fatal(err) + } + options := gitcrawlCloudPublishOptions{AdmissionPolicy: gitcrawlArchiveAdmissionPolicy, ObservationOrder: true} + var firstHash string + for pass := 0; pass < 2; pass++ { + path, admission, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, "", options) + if err != nil { + t.Fatal(err) + } + defer cleanup() + frozen, err := crawlstore.OpenReadOnlyImmutable(ctx, path) + if err != nil { + t.Fatal(err) + } + defer frozen.Close() + snapshot, err := buildGitcrawlCloudSnapshot(ctx, frozen.DB(), path, options, admission) + if err != nil { + t.Fatal(err) + } + if pass == 0 { + firstHash = snapshot.ID + } else if snapshot.ID != firstHash { + t.Fatal("identical input changed snapshot digest") + } + if len(snapshot.Datasets) != 9 || snapshot.SourceSyncAt != "" { + t.Fatalf("snapshot source/datasets = %+v", snapshot) + } + if !slices.Contains(snapshot.Capabilities, gitcrawlArchiveAdmissionCapability) || + !slices.Contains(snapshot.Warnings, "gitcrawl.archive.source.unknown") || + !slices.Contains(snapshot.Warnings, "gitcrawl.archive.source.incomplete") { + t.Fatalf("missing admission identity: %+v", snapshot) + } + if admission.Source.Repositories[0].Inventory.State != crawlstore.ArchiveObservationMissing { + t.Fatal("export clock was treated as source evidence") + } + if admission.Enrichment.Revisions.Complete || admission.Enrichment.Summaries.Complete { + t.Fatal("admission masked missing enrichment") + } + revision := snapshot.Datasets[2] + if revision.Complete || revision.EligibleCount != 1 || revision.CoveredCount != 0 { + t.Fatalf("revision coverage = %+v", revision) + } + encoded, _ := json.Marshal(admission) + var evidence string + if err := frozen.DB().QueryRow(`select value from portable_metadata where key = ?`, gitcrawlCloudAdmissionMetadataKey).Scan(&evidence); err != nil { + t.Fatal(err) + } + if evidence != string(encoded) || strings.Contains(evidence, snapshot.ID) { + t.Fatal("metadata is not deterministic assessment without self-reference") + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if bytes.Contains(data, []byte("private-admission-marker")) || bytes.Contains(data, []byte("untrusted-old-marker")) { + t.Fatal("snapshot leaked private or stale admission evidence") + } + for _, canonical := range []string{"Full canonical body", "Full canonical comment", "Historical canonical body", "@@ -1 +1 @@ original patch"} { + if !bytes.Contains(data, []byte(canonical)) { + t.Fatalf("lost canonical content %q", canonical) + } + } + manifest := gitcrawlCloudManifest("example/archive", snapshot) + if !gitcrawlCloudWarningsMatch(manifest.Warnings, admission.Warnings) { + t.Fatal("manifest lost warnings") + } + admission.Source.RemoteFreshness = crawlstore.ArchiveObservationComplete + if _, err := buildGitcrawlCloudSnapshot(ctx, frozen.DB(), path, options, admission); err == nil { + t.Fatal("mutated assessment was accepted") + } + } + after, err := cloudFileSHA256(sourcePath) + if err != nil || after != sourceHash { + t.Fatalf("source changed: %s %v", after, err) + } + + strictPath, admission, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, "", gitcrawlCloudPublishOptions{}) + if err != nil { + t.Fatal(err) + } + defer cleanup() + if admission != nil { + t.Fatal("strict path inherited source admission") + } + strict, err := crawlstore.OpenReadOnlyImmutable(ctx, strictPath) + if err != nil { + t.Fatal(err) + } + defer strict.Close() + var markers int + if err := strict.DB().QueryRow(`select count(*) from portable_metadata where key = ?`, gitcrawlCloudAdmissionMetadataKey).Scan(&markers); err != nil || markers != 0 { + t.Fatal("strict path retained stale policy marker") + } + if _, err := buildGitcrawlCloudSnapshot(ctx, strict.DB(), strictPath, gitcrawlCloudPublishOptions{}, nil); err == nil || !strings.Contains(err.Error(), "enrichment is incomplete") { + t.Fatalf("strict default changed: %v", err) + } + if _, err := buildGitcrawlCloudSnapshot(ctx, strict.DB(), strictPath, gitcrawlCloudPublishOptions{AllowIncomplete: true}, nil); err != nil { + t.Fatalf("allow-incomplete changed: %v", err) + } +} + +func TestArchiveAdmissionRejectsIntegrityFailures(t *testing.T) { + for _, test := range []struct{ name, sql, want string }{ + {"missing dataset", `drop table thread_fingerprints`, "thread_fingerprints"}, + {"missing canonical history", `drop table comment_revisions`, "canonical comment_revisions"}, + {"missing repositories", `delete from repositories`, "no repositories"}, + {"orphan revision", `pragma foreign_keys=off; insert into thread_revisions(id,thread_id,content_hash,title_hash,body_hash,labels_hash,created_at) values(1,99,'h','h','h','h','2026-09-01T00:00:00Z')`, "referential closure"}, + {"cross repo details", `pragma foreign_keys=off; insert into pull_request_details(thread_id,repo_id,number,raw_json,fetched_at,updated_at) values(1,99,1,'{}','2026-09-01T00:00:00Z','2026-09-01T00:00:00Z')`, "referential closure"}, + {"truncated body", `alter table threads add column body_length integer not null default 999`, "truncated threads"}, + } { + t.Run(test.name, func(t *testing.T) { + ctx := context.Background() + path := seedArchiveAdmissionFixture(t) + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(test.sql); err != nil { + t.Fatal(err) + } + _, _, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, "", gitcrawlCloudPublishOptions{AdmissionPolicy: gitcrawlArchiveAdmissionPolicy, ObservationOrder: true}) + defer cleanup() + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("error = %v, want %q", err, test.want) + } + }) + } +} + +func TestArchiveAdmissionEvidenceChangesDigest(t *testing.T) { + ctx := context.Background() + path := seedArchiveAdmissionFixture(t) + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + options := gitcrawlCloudPublishOptions{AdmissionPolicy: gitcrawlArchiveAdmissionPolicy, ObservationOrder: true} + digest := func() string { + t.Helper() + frozen, _, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, "", options) + if err != nil { + t.Fatal(err) + } + defer cleanup() + hash, err := cloudFileSHA256(frozen) + if err != nil { + t.Fatal(err) + } + return hash + } + if _, err := db.Exec(`insert into sync_runs(repo_id, scope, status, started_at, finished_at, stats_json) values( + 1,'all','success','2026-09-01T00:00:00Z','2026-09-01T00:01:00Z', + '{"started_at":"2026-09-01T00:00:00Z","finished_at":"2026-09-01T00:01:00Z","metadata_only":false,"threads_synced":1}')`); err != nil { + t.Fatal(err) + } + before := digest() + // Stats are scrubbed from the artifact. Only derived admission evidence can + // make this bounds change visible in the final sanitized bytes. + if _, err := db.Exec(`update sync_runs set stats_json = json_set(stats_json, '$.limit', 1)`); err != nil { + t.Fatal(err) + } + if after := digest(); before == after { + t.Fatal("source assessment change did not change artifact digest") + } +} + +func TestArchiveAdmissionSourceSyncRequiresEveryRepositoryObservation(t *testing.T) { + admission := gitcrawlCloudAdmission{Source: crawlstore.ArchiveSourceAssessment{ + Repositories: []crawlstore.ArchiveRepositoryObservations{ + {RepoID: 1, Inventory: crawlstore.ArchiveInventoryObservation{Successful: true, FinishedAt: "2026-09-01T00:00:00Z"}}, + {RepoID: 2, Inventory: crawlstore.ArchiveInventoryObservation{Successful: true, FinishedAt: "2026-09-02T00:00:00Z"}}, + }, + }} + if got := admission.sourceSyncAt(); got != "2026-09-01T00:00:00Z" { + t.Fatalf("source sync = %q, want oldest repository observation", got) + } + admission.Source.Repositories[0].Inventory.Successful = false + if got := admission.sourceSyncAt(); got != "" { + t.Fatalf("failed repository became a source sync: %q", got) + } + admission.Source.Repositories[0].Inventory.Successful = true + admission.Source.Repositories[0].Inventory.FinishedAt = "" + if got := admission.sourceSyncAt(); got != "" { + t.Fatalf("missing observation inherited another repository clock: %q", got) + } +} + +func TestArchiveAdmissionRejectsLossyPortableExportBeforeHTTP(t *testing.T) { + ctx := context.Background() + sourcePath := seedArchiveAdmissionFixture(t) + source, err := sql.Open("sqlite", sourcePath) + if err != nil { + t.Fatal(err) + } + longBody := strings.Repeat("canonical history and review body ", 16) + if _, err := source.Exec(`update threads set body = 'short'; + update comments set body = 'short'; + update comment_revisions set body = ?; + insert into pull_request_review_threads(thread_id, review_thread_id, first_comment_body, comments_json, raw_json, fetched_at) + values(1, 'review-1', ?, '[]', '{}', '2026-09-01T00:00:00Z'); + insert into pull_request_review_thread_revisions(thread_id, review_thread_id, first_comment_body, + comments_json, raw_json, fetched_at, recorded_at) + values(1, 'review-1', ?, '[]', '{}', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z')`, + longBody, longBody, longBody); err != nil { + _ = source.Close() + t.Fatal(err) + } + if err := source.Close(); err != nil { + t.Fatal(err) + } + sourceHash, err := cloudFileSHA256(sourcePath) + if err != nil { + t.Fatal(err) + } + exported, err := portableexport.Export(ctx, portableexport.ExportOptions{ + SourceDBPath: sourcePath, OutputDir: filepath.Join(t.TempDir(), "export"), + DatabaseName: "portable.db", PublicPath: "db/portable.db", + Profile: portableexport.CurrentStateV1, BodyChars: 32, + }) + if err != nil { + t.Fatal(err) + } + portable, err := crawlstore.OpenReadOnlyImmutable(ctx, exported.DatabasePath) + if err != nil { + t.Fatal(err) + } + defer portable.Close() + var truncatedCurrent, historyRows, reviewLength, reviewHistoryLength, patches int + if err := portable.DB().QueryRow(`select + (select count(*) from threads where body_length > length(coalesce(body, ''))) + + (select count(*) from comments where body_length > length(coalesce(body, ''))), + (select count(*) from comment_revisions), + (select length(first_comment_body) from pull_request_review_threads limit 1), + (select length(first_comment_body) from pull_request_review_thread_revisions limit 1), + (select count(*) from pull_request_files where patch is not null and patch != '')`).Scan( + &truncatedCurrent, &historyRows, &reviewLength, &reviewHistoryLength, &patches); err != nil { + t.Fatal(err) + } + if truncatedCurrent != 0 || historyRows != 0 || reviewLength != 32 || reviewHistoryLength != 32 || patches != 0 { + t.Fatalf("native portable loss fixture: current=%d history=%d review=%d review_history=%d patches=%d", + truncatedCurrent, historyRows, reviewLength, reviewHistoryLength, patches) + } + + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + http.Error(w, "unexpected request", http.StatusInternalServerError) + })) + defer server.Close() + cfg := config.Default() + cfg.DBPath = exported.DatabasePath + configPath := filepath.Join(t.TempDir(), "config.toml") + if err := config.Save(configPath, cfg); err != nil { + t.Fatal(err) + } + t.Setenv("GITCRAWL_TEST_LOSSY_ARCHIVE_TOKEN", "fixture-token") + err = New().Run(ctx, []string{ + "--config", configPath, "cloud", "publish", "--remote", server.URL, + "--archive", "example/archive", "--token-env", "GITCRAWL_TEST_LOSSY_ARCHIVE_TOKEN", + "--admission-policy=archive-v1", "--observation-order", "--stage-only", "--json", + }) + if err == nil || !strings.Contains(err.Error(), "lossy portable profile") { + t.Fatalf("lossy native export admission error = %v", err) + } + if requests.Load() != 0 { + t.Fatalf("lossy archive made %d HTTP requests", requests.Load()) + } + after, err := cloudFileSHA256(sourcePath) + if err != nil || after != sourceHash { + t.Fatalf("original archive changed: %s %v", after, err) + } +} + +func TestArchiveAdmissionAcceptsFullRuntimeWithoutPatches(t *testing.T) { + for _, statement := range []string{ + `update pull_request_files set patch = null`, + `delete from pull_request_files`, + } { + t.Run(statement, func(t *testing.T) { + ctx := context.Background() + path := seedArchiveAdmissionFixture(t) + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(statement); err != nil { + t.Fatal(err) + } + _, admission, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, "", gitcrawlCloudPublishOptions{ + AdmissionPolicy: gitcrawlArchiveAdmissionPolicy, ObservationOrder: true, + }) + defer cleanup() + if err != nil || admission == nil || !admission.Integrity.FullBodies { + t.Fatalf("full runtime with legitimate absent patches rejected: admission=%+v error=%v", admission, err) + } + }) + } +} + +func TestArchiveAdmissionLossyProfileDeclarations(t *testing.T) { + for _, test := range []struct { + key, value string + lossy bool + }{ + {"profile", portableexport.CurrentStateV1, true}, + {"body_chars", "32", true}, + {"capabilities", "raw_json_stripped,body_excerpts", true}, + {"capabilities", "comment_excerpts,author_association", true}, + {"excluded", "raw_json,pull_request_file_patches,documents", true}, + {"excluded", "comment_revision_history", true}, + {"capabilities", "raw_json_stripped", false}, + {"excluded", "raw_json,documents", false}, + {gitcrawlCloudAdmissionMetadataKey, `{"policy":"archive-v1"}`, false}, + } { + t.Run(test.key+"="+test.value, func(t *testing.T) { + ctx := context.Background() + db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "metadata.db")) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := requireLosslessGitcrawlCloudSource(ctx, db); err != nil { + t.Fatalf("no portable metadata: %v", err) + } + if _, err := db.Exec(`create table portable_metadata(key text primary key, value text not null); + insert into portable_metadata(key,value) values(?,?)`, test.key, test.value); err != nil { + t.Fatal(err) + } + if err := requireLosslessGitcrawlCloudSource(ctx, db); (err != nil) != test.lossy { + t.Fatalf("profile rejection = %v, want lossy=%v", err, test.lossy) + } + }) + } +} + +func TestGitcrawlCloudWarningIdentity(t *testing.T) { + warnings := []string{"gitcrawl.archive.source.unknown"} + if !gitcrawlCloudWarningsMatch(nil, []string{}) || + !gitcrawlCloudWarningsMatch(warnings, slices.Clone(warnings)) { + t.Fatal("equal canonical warnings did not match") + } + if gitcrawlCloudWarningsMatch(nil, warnings) || + gitcrawlCloudWarningsMatch(warnings, nil) || + gitcrawlCloudWarningsMatch([]string{"unknown"}, []string{"unknown"}) || + gitcrawlCloudWarningsMatch(append(slices.Clone(warnings), warnings[0]), warnings) { + t.Fatal("invalid or changed warnings matched") + } +} diff --git a/internal/cli/cloud_commands.go b/internal/cli/cloud_commands.go index 2f7400d0..91d62be0 100644 --- a/internal/cli/cloud_commands.go +++ b/internal/cli/cloud_commands.go @@ -20,6 +20,7 @@ import ( crawlremote "github.com/openclaw/crawlkit/remote" "github.com/openclaw/gitcrawl/internal/config" + crawlstore "github.com/openclaw/gitcrawl/internal/store" ) const ( @@ -82,15 +83,22 @@ func (a *App) runCloudPublish(ctx context.Context, args []string) error { tokenEnv := fs.String("token-env", "", "remote token environment variable") allowIncomplete := fs.Bool("allow-incomplete", false, "publish even when local enrichment coverage is incomplete") observationOrder := fs.Bool("observation-order", false, "publish durable observation ordering when the remote fence is enabled") + admissionPolicy := fs.String("admission-policy", "", "explicit archive admission policy (archive-v1)") stageOnly := fs.Bool("stage-only", false, "stage the immutable snapshot without moving unpinned reads") jsonOut := fs.Bool("json", false, "write JSON output") - if err := fs.Parse(normalizeCommandArgs(args, map[string]bool{"remote": true, "archive": true, "token-env": true})); err != nil { + if err := fs.Parse(normalizeCommandArgs(args, map[string]bool{"remote": true, "archive": true, "token-env": true, "admission-policy": true})); err != nil { return usageErr(err) } a.applyCommandJSON(*jsonOut) if fs.NArg() != 0 { return usageErr(fmt.Errorf("cloud publish takes flags only")) } + options := gitcrawlCloudPublishOptions{ + AllowIncomplete: *allowIncomplete, ObservationOrder: *observationOrder, AdmissionPolicy: *admissionPolicy, + } + if err := options.validate(); err != nil { + return usageErr(err) + } cutover := !*stageOnly cfg, err := config.LoadRuntime(a.configPath) @@ -128,22 +136,23 @@ func (a *App) runCloudPublish(ctx context.Context, args []string) error { if err != nil { return err } - snapshotPath, cleanupSnapshot, err := cloudSQLiteSnapshotPath(ctx, rt.Store.DB(), rt.Store.Path()) + snapshotPath, admission, cleanupSnapshot, err := cloudSQLiteSnapshotPath(ctx, rt.Store.DB(), rt.Store.Path(), options) if err != nil { return err } defer cleanupSnapshot() - snapshotDB, err := sql.Open("sqlite", snapshotPath) + frozen, err := crawlstore.OpenReadOnlyImmutable(ctx, snapshotPath) if err != nil { return fmt.Errorf("open frozen cloud snapshot: %w", err) } - defer snapshotDB.Close() + defer frozen.Close() + snapshotDB := frozen.DB() snapshot, err := buildGitcrawlCloudSnapshot( ctx, snapshotDB, snapshotPath, - *allowIncomplete, - *observationOrder, + options, + admission, ) if err != nil { return err @@ -348,6 +357,8 @@ func (a *App) runCloudPublish(ctx context.Context, args []string) error { "capabilities": manifest.Capabilities, "datasets": counts, "hydration": snapshot.Hydration, + "admission": snapshot.Admission, + "warnings": snapshot.Warnings, "already_staged": alreadyStaged, "already_cut_over": alreadyCutOver, "mutation_token": mutationToken, @@ -918,7 +929,7 @@ func cursorFor(start int64) string { } func uploadSQLiteArchive(ctx context.Context, client *crawlremote.Client, app, archive string, db *sql.DB, dbPath string, manifest crawlremote.IngestManifest, counts map[string]int64) (*crawlremote.SQLiteBundle, error) { - snapshotPath, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, dbPath) + snapshotPath, _, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, dbPath, gitcrawlCloudPublishOptions{}) if err != nil { return nil, err } @@ -1336,6 +1347,9 @@ func gitcrawlPublisherStatusMatches( if !equalUniqueStringSet(snapshot.Capabilities, publicationCapabilities) { return false } + if !gitcrawlCloudWarningsMatch(snapshot.Warnings, manifest.Warnings) { + return false + } return true } @@ -1668,50 +1682,79 @@ func gitcrawlCloudSQLiteBundlePrivacy() map[string]any { } } -func cloudSQLiteSnapshotPath(ctx context.Context, db *sql.DB, dbPath string) (string, func(), error) { +func cloudSQLiteSnapshotPath( + ctx context.Context, db *sql.DB, dbPath string, options gitcrawlCloudPublishOptions, +) (string, *gitcrawlCloudAdmission, func(), error) { + if err := options.validate(); err != nil { + return "", nil, func() {}, err + } snapshotPath, cleanup, err := sqliteSnapshotPath(ctx, db, "") if err != nil { source := strings.TrimSpace(dbPath) if source == "" { - return "", func() {}, err + return "", nil, func() {}, err } if _, statErr := os.Stat(source); statErr != nil { - return "", func() {}, fmt.Errorf("stat cloud SQLite source: %w", statErr) + return "", nil, func() {}, fmt.Errorf("stat cloud SQLite source: %w", statErr) } reopened, openErr := sql.Open("sqlite", source) if openErr != nil { - return "", func() {}, fmt.Errorf("reopen cloud SQLite source: %w", openErr) + return "", nil, func() {}, fmt.Errorf("reopen cloud SQLite source: %w", openErr) } snapshotPath, cleanup, err = sqliteSnapshotPath(ctx, reopened, "") closeErr := reopened.Close() if err != nil { - return "", func() {}, err + return "", nil, func() {}, err } if closeErr != nil { cleanup() - return "", func() {}, fmt.Errorf("close reopened cloud SQLite source: %w", closeErr) + return "", nil, func() {}, fmt.Errorf("close reopened cloud SQLite source: %w", closeErr) } } snapshotDB, err := sql.Open("sqlite", snapshotPath) if err != nil { cleanup() - return "", func() {}, fmt.Errorf("open cloud SQLite snapshot: %w", err) + return "", nil, func() {}, fmt.Errorf("open cloud SQLite snapshot: %w", err) + } + // Derive evidence from the frozen copy before the sanitizer removes run + // diagnostics. A source's previous admission marker is never consulted. + var admission *gitcrawlCloudAdmission + if options.AdmissionPolicy == gitcrawlArchiveAdmissionPolicy { + admission, err = assessGitcrawlCloudArchive(ctx, snapshotPath) + if err != nil { + _ = snapshotDB.Close() + cleanup() + return "", nil, func() {}, err + } } if err := sanitizeCloudSQLiteSnapshot(ctx, snapshotDB); err != nil { _ = snapshotDB.Close() cleanup() - return "", func() {}, err + return "", nil, func() {}, err + } + if admission != nil { + if err := validateGitcrawlArchiveIntegrity(ctx, snapshotDB, admission); err != nil { + _ = snapshotDB.Close() + cleanup() + return "", nil, func() {}, err + } + admission.Integrity.Privacy = true + } + if err := writeGitcrawlCloudAdmission(ctx, snapshotDB, admission); err != nil { + _ = snapshotDB.Close() + cleanup() + return "", nil, func() {}, fmt.Errorf("write cloud admission evidence: %w", err) } if _, err := snapshotDB.ExecContext(ctx, `vacuum`); err != nil { _ = snapshotDB.Close() cleanup() - return "", func() {}, fmt.Errorf("compact cloud SQLite snapshot: %w", err) + return "", nil, func() {}, fmt.Errorf("compact cloud SQLite snapshot: %w", err) } if err := snapshotDB.Close(); err != nil { cleanup() - return "", func() {}, fmt.Errorf("close cloud SQLite snapshot: %w", err) + return "", nil, func() {}, fmt.Errorf("close cloud SQLite snapshot: %w", err) } - return snapshotPath, cleanup, nil + return snapshotPath, admission, cleanup, nil } func sanitizeCloudSQLiteSnapshot(ctx context.Context, db *sql.DB) error { diff --git a/internal/cli/cloud_commands_test.go b/internal/cli/cloud_commands_test.go index 61ffd581..3f66bb3a 100644 --- a/internal/cli/cloud_commands_test.go +++ b/internal/cli/cloud_commands_test.go @@ -994,6 +994,20 @@ func TestGitcrawlReaderStatusMatchesCompleteServingSnapshot(t *testing.T) { if !gitcrawlReaderStatusMatches(status, snapshot, manifest, capabilities, cutoverAt) { t.Fatal("complete serving snapshot did not match") } + manifest.Warnings = []string{"gitcrawl.archive.source.unknown"} + status.Warnings = slices.Clone(manifest.Warnings) + if gitcrawlReaderStatusMatches(status, snapshot, manifest, capabilities, cutoverAt) { + t.Fatal("transient status warnings replaced immutable snapshot warnings") + } + status.Snapshot.Warnings = slices.Clone(manifest.Warnings) + if !gitcrawlReaderStatusMatches(status, snapshot, manifest, capabilities, cutoverAt) { + t.Fatal("matching snapshot warnings did not match") + } + manifest.Warnings = nil + if gitcrawlReaderStatusMatches(status, snapshot, manifest, capabilities, cutoverAt) { + t.Fatal("nonempty-to-empty snapshot warning drift matched") + } + status.Snapshot.Warnings = nil status.CoverageComplete = false if gitcrawlReaderStatusMatches(status, snapshot, manifest, capabilities, cutoverAt) { @@ -1322,6 +1336,7 @@ func TestRecoverConcurrentGitcrawlSnapshotAdoptsOnlyMatchingCompletedSnapshot(t ID: snapshotID, SourceSyncAt: "2026-07-12T12:00:00Z", DatasetGeneratedAt: "2026-07-12T12:01:00Z", + Warnings: []string{"gitcrawl.archive.source.unknown"}, } manifest := gitcrawlCloudManifest("gitcrawl/openclaw__openclaw", snapshot) publicationCapabilities := gitcrawlCloudPublicationCapabilities(manifest.Capabilities) @@ -1333,6 +1348,7 @@ func TestRecoverConcurrentGitcrawlSnapshotAdoptsOnlyMatchingCompletedSnapshot(t coverageComplete bool wantGeneration string want string + omitWarnings bool }{ { name: "independent winner generation", @@ -1340,6 +1356,13 @@ func TestRecoverConcurrentGitcrawlSnapshotAdoptsOnlyMatchingCompletedSnapshot(t coverageComplete: true, wantGeneration: winnerGeneration, }, + { + name: "missing immutable warnings", + activeSnapshotID: snapshotID, + coverageComplete: true, + omitWarnings: true, + want: "does not match the requested digest, profile, and coverage", + }, { name: "unrelated active candidate", activeSnapshotID: strings.Repeat("b", 64), @@ -1358,6 +1381,10 @@ func TestRecoverConcurrentGitcrawlSnapshotAdoptsOnlyMatchingCompletedSnapshot(t http.Error(w, fmt.Sprintf("snapshot_id = %q, want %q", got, snapshotID), http.StatusBadRequest) return } + warnings := manifest.Warnings + if test.omitWarnings { + warnings = nil + } _ = json.NewEncoder(w).Encode(crawlremote.PublisherStatus{ App: manifest.App, Archive: manifest.Archive, @@ -1373,6 +1400,7 @@ func TestRecoverConcurrentGitcrawlSnapshotAdoptsOnlyMatchingCompletedSnapshot(t SchemaHash: manifest.SchemaHash, Capabilities: publicationCapabilities, CoverageComplete: test.coverageComplete, + Warnings: warnings, }, }) })) diff --git a/internal/cli/cloud_snapshot.go b/internal/cli/cloud_snapshot.go index 4ece1b7c..607b5ee0 100644 --- a/internal/cli/cloud_snapshot.go +++ b/internal/cli/cloud_snapshot.go @@ -3,6 +3,7 @@ package cli import ( "context" "database/sql" + "encoding/json" "fmt" "slices" "strings" @@ -38,45 +39,76 @@ type gitcrawlCloudSnapshot struct { Capabilities []string Datasets []gitcrawlCloudDataset Hydration crawlstore.EnrichmentCoverage + Admission *gitcrawlCloudAdmission + Warnings []string } func buildGitcrawlCloudSnapshot( ctx context.Context, db *sql.DB, snapshotPath string, - allowIncomplete bool, - observationOrder bool, + options gitcrawlCloudPublishOptions, + admission *gitcrawlCloudAdmission, ) (gitcrawlCloudSnapshot, error) { - snapshotID, err := cloudFileSHA256(snapshotPath) - if err != nil { + if err := options.validate(); err != nil { return gitcrawlCloudSnapshot{}, err } - sourceSyncAt, err := gitcrawlCloudSourceSyncAt(ctx, db) - if err != nil { - return gitcrawlCloudSnapshot{}, err + if (options.AdmissionPolicy != "") != (admission != nil) { + return gitcrawlCloudSnapshot{}, fmt.Errorf("cloud admission policy does not match frozen assessment") } - capabilities, err := gitcrawlCloudCapabilities(ctx, db, observationOrder) + snapshotID, err := cloudFileSHA256(snapshotPath) if err != nil { return gitcrawlCloudSnapshot{}, err } - hydration, err := gitcrawlCloudHydration(ctx, snapshotPath) + capabilities, err := gitcrawlCloudCapabilities(ctx, db, options.ObservationOrder) if err != nil { return gitcrawlCloudSnapshot{}, err } - datasets, err := loadGitcrawlCloudDatasets( - ctx, - db, - slices.Contains(capabilities, gitcrawlObservationOrderCapability), - hydration, - ) - if err != nil { - return gitcrawlCloudSnapshot{}, err + var sourceSyncAt string + var hydration crawlstore.EnrichmentCoverage + var datasets []gitcrawlCloudDataset + var warnings []string + if admission != nil { + encoded, err := json.Marshal(admission) + if err != nil { + return gitcrawlCloudSnapshot{}, err + } + var frozenEvidence string + if err := db.QueryRowContext(ctx, `select value from portable_metadata where key = ?`, + gitcrawlCloudAdmissionMetadataKey).Scan(&frozenEvidence); err != nil || + admission.Policy != options.AdmissionPolicy || frozenEvidence != string(encoded) { + return gitcrawlCloudSnapshot{}, fmt.Errorf("cloud admission assessment does not match frozen evidence") + } + if !admission.Integrity.SQLite || !admission.Integrity.CompatibleSchema || + !admission.Integrity.RequiredData || !admission.Integrity.ReferentialClosure || + !admission.Integrity.FullBodies || !admission.Integrity.Privacy { + return gitcrawlCloudSnapshot{}, fmt.Errorf("frozen archive admission integrity is incomplete") + } + sourceSyncAt, hydration, datasets = admission.sourceSyncAt(), admission.Enrichment, admission.datasets + warnings, err = gitcrawlArchiveWarnings(*admission) + if err != nil || !gitcrawlCloudWarningsMatch(warnings, admission.Warnings) { + return gitcrawlCloudSnapshot{}, fmt.Errorf("frozen archive admission warnings do not match assessment") + } + capabilities = append(capabilities, gitcrawlArchiveAdmissionCapability) + } else { + sourceSyncAt, err = gitcrawlCloudSourceSyncAt(ctx, db) + if err != nil { + return gitcrawlCloudSnapshot{}, err + } + hydration, err = gitcrawlCloudHydration(ctx, snapshotPath) + if err != nil { + return gitcrawlCloudSnapshot{}, err + } + datasets, err = loadGitcrawlCloudDatasets(ctx, db, options.ObservationOrder, hydration) + if err != nil { + return gitcrawlCloudSnapshot{}, err + } } if len(datasets) == 0 || datasets[0].RowCount == 0 { return gitcrawlCloudSnapshot{}, fmt.Errorf("cloud snapshot has no repositories") } missing := incompleteGitcrawlCloudHydration(hydration) - if len(missing) > 0 && !allowIncomplete { + if len(missing) > 0 && !options.AllowIncomplete && admission == nil { return gitcrawlCloudSnapshot{}, fmt.Errorf( "cloud snapshot enrichment is incomplete (%s); hydrate the archive or pass --allow-incomplete", strings.Join(missing, ", "), @@ -89,6 +121,8 @@ func buildGitcrawlCloudSnapshot( Capabilities: capabilities, Datasets: datasets, Hydration: hydration, + Admission: admission, + Warnings: warnings, }, nil } @@ -109,6 +143,7 @@ func gitcrawlCloudManifest(archive string, snapshot gitcrawlCloudSnapshot) crawl SnapshotID: snapshot.ID, SourceSHA256: snapshot.ID, Capabilities: capabilities, + Warnings: slices.Clone(snapshot.Warnings), } } diff --git a/internal/cli/cloud_snapshot_test.go b/internal/cli/cloud_snapshot_test.go index e85c39bb..5858309f 100644 --- a/internal/cli/cloud_snapshot_test.go +++ b/internal/cli/cloud_snapshot_test.go @@ -121,7 +121,7 @@ with a second line', 'https://github.com/example/repo/pull/1', checks[i].excludedBefore = readValues(db, checks[i].table, checks[i].excluded) } - snapshotPath, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, dbPath) + snapshotPath, _, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, dbPath, gitcrawlCloudPublishOptions{}) if err != nil { t.Fatalf("cloud snapshot: %v", err) } @@ -240,7 +240,7 @@ func TestCloudSQLiteSnapshotLegacyPrivacyColumns(t *testing.T) { if _, err := db.ExecContext(ctx, tc.schema); err != nil { t.Fatalf("seed legacy source: %v", err) } - snapshotPath, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, "") + snapshotPath, _, cleanup, err := cloudSQLiteSnapshotPath(ctx, db, "", gitcrawlCloudPublishOptions{}) if err != nil { t.Fatalf("legacy cloud snapshot: %v", err) } @@ -385,7 +385,7 @@ func TestGitcrawlCloudSourceSyncAtUsesPortableMetadataWithoutSyncRuns(t *testing t.Fatalf("reopen portable sqlite: %v", err) } defer db.Close() - snapshot, err := buildGitcrawlCloudSnapshot(ctx, db, path, true, false) + snapshot, err := buildGitcrawlCloudSnapshot(ctx, db, path, gitcrawlCloudPublishOptions{AllowIncomplete: true}, nil) if err != nil { t.Fatalf("build portable cloud snapshot: %v", err) } diff --git a/internal/store/archive_source.go b/internal/store/archive_source.go new file mode 100644 index 00000000..dceb66c9 --- /dev/null +++ b/internal/store/archive_source.go @@ -0,0 +1,208 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "strings" +) + +type ArchiveObservationState string + +const ( + ArchiveObservationUnsupported ArchiveObservationState = "unsupported" + ArchiveObservationUnknown ArchiveObservationState = "unknown" + ArchiveObservationMissing ArchiveObservationState = "missing" + ArchiveObservationStale ArchiveObservationState = "stale" + ArchiveObservationPartial ArchiveObservationState = "partial" + ArchiveObservationComplete ArchiveObservationState = "complete" + ArchiveObservationEmpty ArchiveObservationState = "empty" +) + +type ArchiveInventoryObservation struct { + State ArchiveObservationState `json:"state"` + Successful bool `json:"successful"` + RunID int64 `json:"run_id,omitempty"` + FinishedAt string `json:"finished_at,omitempty"` +} + +type ArchiveChildObservations struct { + State ArchiveObservationState `json:"state"` + Eligible int `json:"eligible"` + Observed int `json:"observed"` + Missing int `json:"missing"` + Stale int `json:"stale"` + Unknown int `json:"unknown"` +} + +type ArchiveRepositoryObservations struct { + RepoID int64 `json:"repo_id"` + Inventory ArchiveInventoryObservation `json:"inventory"` + Children map[string]ArchiveChildObservations `json:"children"` + PRDetails EnrichmentCoverageMetric `json:"pr_details"` + PRFiles EnrichmentCoverageMetric `json:"pr_files"` + WorkflowFreshness ArchiveObservationState `json:"workflow_freshness"` + FailedHydrations *int `json:"failed_hydrations"` +} + +type ArchiveSourceAssessment struct { + RemoteFreshness ArchiveObservationState `json:"remote_freshness"` + Repositories []ArchiveRepositoryObservations `json:"repositories"` +} + +// ArchiveSourceObservations describes persisted observations, not current GitHub +// state. The caller must use the same frozen database for coverage and this read. +func (s *Store) ArchiveSourceObservations(ctx context.Context, coverage ArchiveCoverage) (ArchiveSourceAssessment, error) { + result := ArchiveSourceAssessment{ + RemoteFreshness: ArchiveObservationUnknown, + Repositories: make([]ArchiveRepositoryObservations, 0, len(coverage.Rows)), + } + for _, row := range coverage.Rows { + inventory, err := s.archiveInventoryObservation(ctx, row.RepoID) + if err != nil { + return ArchiveSourceAssessment{}, err + } + repo := ArchiveRepositoryObservations{ + RepoID: row.RepoID, Inventory: inventory, + Children: make(map[string]ArchiveChildObservations), + PRDetails: row.Enrichment.PRDetails, PRFiles: row.Enrichment.PRFiles, + WorkflowFreshness: ArchiveObservationUnknown, + FailedHydrations: row.KnownFailedHydrations, + } + // Workflow observations are per head SHA, not a repository-wide list or + // a thread clock. Their presence cannot prove repository freshness. + if !s.archiveCoverageHasColumns(ctx, "workflow_run_observation_reservations", "repo_id", "head_sha", "observation_sequence") { + repo.WorkflowFreshness = ArchiveObservationUnsupported + } + for _, family := range threadChildObservationFamilies { + metric, err := s.archiveChildObservations(ctx, row.RepoID, family) + if err != nil { + return ArchiveSourceAssessment{}, err + } + repo.Children[string(family)] = metric + } + result.Repositories = append(result.Repositories, repo) + } + return result, nil +} + +func (s *Store) archiveInventoryObservation(ctx context.Context, repoID int64) (ArchiveInventoryObservation, error) { + result := ArchiveInventoryObservation{State: ArchiveObservationUnsupported} + if !s.archiveCoverageHasColumns(ctx, "sync_runs", "id", "repo_id", "scope", "status", "started_at", "finished_at", "stats_json") { + return result, nil + } + var scope, status, started, finished, encoded string + err := s.q().QueryRowContext(ctx, ` + select id, scope, status, started_at, coalesce(finished_at, ''), coalesce(stats_json, '') + from sync_runs where repo_id = ? order by id desc limit 1 + `, repoID).Scan(&result.RunID, &scope, &status, &started, &finished, &encoded) + if err == sql.ErrNoRows { + result.State = ArchiveObservationMissing + return result, nil + } + if err != nil { + return result, fmt.Errorf("read archive inventory observation: %w", err) + } + result.State = ArchiveObservationUnknown + if value, ok := parseArchiveCoverageTimestamp(finished); ok { + result.FinishedAt = formatArchiveCoverageTimestamp(value) + } + if status != "success" && status != "completed" { + result.State = ArchiveObservationPartial + return result, nil + } + result.Successful = true + // Syncer.Stats records bounds and these mandatory fields at commit time. + // Missing/sanitized stats are unknown, never an implicit unbounded run. + var stats struct { + StartedAt string `json:"started_at"` + FinishedAt string `json:"finished_at"` + MetadataOnly *bool `json:"metadata_only"` + ThreadsSynced *int `json:"threads_synced"` + ThreadsSkippedStale int `json:"threads_skipped_stale"` + RequestedSince string `json:"requested_since"` + Limit int `json:"limit"` + Numbers []int `json:"numbers"` + } + if json.Unmarshal([]byte(encoded), &stats) != nil || + stats.MetadataOnly == nil || stats.ThreadsSynced == nil || *stats.ThreadsSynced < 0 || + stats.StartedAt != started || stats.FinishedAt != finished || result.FinishedAt == "" { + return result, nil + } + if _, ok := parseArchiveCoverageTimestamp(started); !ok { + return result, nil + } + result.State = ArchiveObservationPartial + if scope == "all" && stats.Limit <= 0 && len(stats.Numbers) == 0 && stats.RequestedSince == "" { + result.State = ArchiveObservationComplete + if *stats.ThreadsSynced == 0 && stats.ThreadsSkippedStale == 0 { + result.State = ArchiveObservationEmpty + } + } + return result, nil +} + +func (s *Store) archiveChildObservations(ctx context.Context, repoID int64, family ThreadChildObservationFamily) (ArchiveChildObservations, error) { + metric := ArchiveChildObservations{State: ArchiveObservationUnsupported} + if !s.archiveCoverageHasColumns(ctx, "thread_child_observation_reservations", + "thread_id", "family", "source_updated_at", "observation_sequence") || + !s.archiveCoverageHasColumns(ctx, "threads", "updated_at_gh", "observation_sequence") { + return metric, nil + } + filter := "" + if family != ThreadChildComments { + filter = " and t.kind = 'pull_request'" + } + rows, err := s.q().QueryContext(ctx, ` + select coalesce(t.updated_at_gh, ''), t.observation_sequence, + r.thread_id is not null, coalesce(r.source_updated_at, ''), coalesce(r.observation_sequence, 0) + from threads t + left join thread_child_observation_reservations r on r.thread_id = t.id and r.family = ? + where t.repo_id = ?`+filter, string(family), repoID) + if err != nil { + return metric, fmt.Errorf("read archive child observations: %w", err) + } + defer rows.Close() + for rows.Next() { + var source, observedSource string + var sequence, observedSequence int64 + var found bool + if err := rows.Scan(&source, &sequence, &found, &observedSource, &observedSequence); err != nil { + return metric, fmt.Errorf("scan archive child observations: %w", err) + } + metric.Eligible++ + if !found { + metric.Missing++ + continue + } + metric.Observed++ + _, sourceValid := parseArchiveCoverageTimestamp(source) + _, observedValid := parseArchiveCoverageTimestamp(observedSource) + if (!sourceValid && strings.TrimSpace(source) != "") || + (!observedValid && strings.TrimSpace(observedSource) != "") || + observedSequence <= 0 || sequence == 0 { + metric.Unknown++ + } else if !archiveObservationAtOrAfter(observedSource, observedSequence, source, observationSequenceOrderValue(sequence)) { + metric.Stale++ + } + } + if err := rows.Err(); err != nil { + return metric, fmt.Errorf("iterate archive child observations: %w", err) + } + switch { + case metric.Eligible == 0: + metric.State = ArchiveObservationEmpty + case metric.Missing == metric.Eligible: + metric.State = ArchiveObservationMissing + case metric.Stale == metric.Eligible: + metric.State = ArchiveObservationStale + case metric.Unknown == metric.Eligible: + metric.State = ArchiveObservationUnknown + case metric.Missing+metric.Stale+metric.Unknown > 0: + metric.State = ArchiveObservationPartial + default: + metric.State = ArchiveObservationComplete + } + return metric, nil +} diff --git a/internal/store/archive_source_test.go b/internal/store/archive_source_test.go new file mode 100644 index 00000000..1a75ee89 --- /dev/null +++ b/internal/store/archive_source_test.go @@ -0,0 +1,172 @@ +package store + +import ( + "context" + "encoding/json" + "path/filepath" + "testing" +) + +func TestArchiveInventoryObservation(t *testing.T) { + ctx := context.Background() + start := "2026-09-01T00:00:00Z" + finish := "2026-09-01T00:01:00Z" + for _, test := range []struct { + name, scope, status, stats string + drop, noRun bool + want ArchiveObservationState + }{ + {name: "unsupported", drop: true, want: ArchiveObservationUnsupported}, + {name: "missing", noRun: true, want: ArchiveObservationMissing}, + {name: "unknown sanitized stats", scope: "all", status: "success", want: ArchiveObservationUnknown}, + {name: "unknown malformed stats", scope: "all", status: "success", stats: "{", want: ArchiveObservationUnknown}, + {name: "partial state", scope: "open", status: "success", stats: `"threads_synced":1`, want: ArchiveObservationPartial}, + {name: "partial bounded", scope: "all", status: "success", stats: `"threads_synced":1,"limit":1`, want: ArchiveObservationPartial}, + {name: "partial since", scope: "all", status: "success", stats: `"threads_synced":1,"requested_since":"2026-08-01T00:00:00Z"`, want: ArchiveObservationPartial}, + {name: "partial targeted", scope: "all", status: "success", stats: `"threads_synced":1,"numbers":[1]`, want: ArchiveObservationPartial}, + {name: "failed", scope: "all", status: "failed", want: ArchiveObservationPartial}, + {name: "complete", scope: "all", status: "success", stats: `"threads_synced":1`, want: ArchiveObservationComplete}, + {name: "stale skipped is not empty", scope: "all", status: "success", stats: `"threads_synced":0,"threads_skipped_stale":1`, want: ArchiveObservationComplete}, + {name: "empty", scope: "all", status: "success", stats: `"threads_synced":0`, want: ArchiveObservationEmpty}, + } { + t.Run(test.name, func(t *testing.T) { + st, err := Open(ctx, filepath.Join(t.TempDir(), "source.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + repoID, err := st.UpsertRepository(ctx, Repository{Owner: "example", Name: "repo", FullName: "example/repo", UpdatedAt: start}) + if err != nil { + t.Fatal(err) + } + if test.drop { + if _, err := st.DB().ExecContext(ctx, `drop table sync_runs`); err != nil { + t.Fatal(err) + } + } else if !test.noRun { + stats := test.stats + if stats != "" && stats != "{" { + stats = `{"started_at":"` + start + `","finished_at":"` + finish + `","metadata_only":false,` + stats + `}` + } + if _, err := st.RecordRun(ctx, RunRecord{RepoID: repoID, Kind: "sync", Scope: test.scope, + Status: test.status, StartedAt: start, FinishedAt: finish, StatsJSON: stats}); err != nil { + t.Fatal(err) + } + } + got, err := st.archiveInventoryObservation(ctx, repoID) + if err != nil { + t.Fatal(err) + } + if got.State != test.want { + t.Fatalf("state = %q, want %q (%+v)", got.State, test.want, got) + } + }) + } +} + +func TestArchiveChildObservations(t *testing.T) { + ctx := context.Background() + for _, test := range []struct { + name string + sequence, observedSequence int64 + source, observedSource string + empty, missing, drop bool + want ArchiveObservationState + }{ + {name: "unsupported", drop: true, want: ArchiveObservationUnsupported}, + {name: "empty", empty: true, want: ArchiveObservationEmpty}, + {name: "missing", missing: true, want: ArchiveObservationMissing}, + {name: "stale sequence", sequence: 2, observedSequence: 1, want: ArchiveObservationStale}, + {name: "clockless complete", sequence: 2, observedSequence: 2, want: ArchiveObservationComplete}, + {name: "negative metadata sequence", sequence: -3, observedSequence: 2, want: ArchiveObservationStale}, + {name: "unknown legacy sequence", sequence: 0, observedSequence: 2, want: ArchiveObservationUnknown}, + {name: "unknown invalid clock", sequence: 2, observedSequence: 2, source: "invalid", want: ArchiveObservationUnknown}, + {name: "stale source", sequence: 2, observedSequence: 3, source: "2026-09-02T00:00:00Z", observedSource: "2026-09-01T00:00:00Z", want: ArchiveObservationStale}, + } { + t.Run(test.name, func(t *testing.T) { + st, err := Open(ctx, filepath.Join(t.TempDir(), "source.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + repoID, err := st.UpsertRepository(ctx, Repository{Owner: "example", Name: "repo", FullName: "example/repo", UpdatedAt: "2026-09-01T00:00:00Z"}) + if err != nil { + t.Fatal(err) + } + if !test.empty { + thread := archiveCoverageThread(repoID, 1, "issue") + id, err := st.UpsertThread(ctx, thread) + if err != nil { + t.Fatal(err) + } + if _, err := st.DB().ExecContext(ctx, `update threads set updated_at_gh = ?, observation_sequence = ? where id = ?`, test.source, test.sequence, id); err != nil { + t.Fatal(err) + } + if !test.missing && !test.drop { + if _, err := st.DB().ExecContext(ctx, `insert into thread_child_observation_reservations(thread_id, family, source_updated_at, observation_sequence) values(?, 'comments', ?, ?)`, id, test.observedSource, test.observedSequence); err != nil { + t.Fatal(err) + } + } + } + if test.drop { + if _, err := st.DB().ExecContext(ctx, `drop table thread_child_observation_reservations`); err != nil { + t.Fatal(err) + } + } + got, err := st.archiveChildObservations(ctx, repoID, ThreadChildComments) + if err != nil { + t.Fatal(err) + } + if got.State != test.want { + t.Fatalf("metric = %+v, want %q", got, test.want) + } + }) + } +} + +func TestArchiveSourceObservationsKeepRepositoryScopeAndUnknownFreshness(t *testing.T) { + ctx := context.Background() + st, err := Open(ctx, filepath.Join(t.TempDir(), "source.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + first, second := seedArchiveCoverageRows(t, ctx, st) + if _, err := st.RecordRun(ctx, RunRecord{RepoID: second, Kind: "sync", Scope: "all", Status: "success", + StartedAt: "2026-09-01T00:00:00Z", FinishedAt: "2026-09-01T00:01:00Z", + StatsJSON: `{"started_at":"2026-09-01T00:00:00Z","finished_at":"2026-09-01T00:01:00Z","metadata_only":false,"threads_synced":1}`}); err != nil { + t.Fatal(err) + } + if err := st.ensurePortableMetadata(ctx); err != nil { + t.Fatal(err) + } + if _, err := st.DB().ExecContext(ctx, `insert into portable_metadata(key,value) values('exported_at','2099-01-01T00:00:00Z')`); err != nil { + t.Fatal(err) + } + coverage, err := st.ArchiveCoverage(ctx, ArchiveCoverageOptions{}) + if err != nil { + t.Fatal(err) + } + got, err := st.ArchiveSourceObservations(ctx, coverage) + if err != nil { + t.Fatal(err) + } + if got.RemoteFreshness != ArchiveObservationUnknown || len(got.Repositories) != 2 { + t.Fatalf("source = %+v", got) + } + for _, repo := range got.Repositories { + if repo.RepoID == first && repo.Inventory.State == ArchiveObservationComplete { + t.Fatal("another repo established inventory coverage") + } + if repo.RepoID == second && repo.Inventory.State != ArchiveObservationComplete { + t.Fatalf("second repository: %+v", repo) + } + } + encoded, err := json.Marshal(got) + if err != nil { + t.Fatal(err) + } + if !json.Valid(encoded) { + t.Fatal("invalid source assessment") + } +}