Merge pull request #242 from morluto/agent/address-shared-corpus-contracts

feat(mcp): bind offline workflows to corpus revisions

Author: morluto

docs/external-validation-receipts.md

@@ -0,0 +1,24 @@
+# External validation receipts
+
+`internal/evidence/testdata/flameox-profiler-receipt.json` is a checked-in
+Flameox-compatible example mapped to the producer-neutral
+`gitcontribute.external-validation.v1` contract. It is fixture data only: the
+repository does not depend on or execute Flameox.
+
+The field mapping is intentionally direct:
+
+| Profiler handoff | GitContribute receipt field |
+| --- | --- |
+| producer name | `producer` |
+| profiler run identity | `provider`, `external_run_id` |
+| source repository and commit | `repository`, `revision` |
+| primary output digest | `artifact_sha256` |
+| named output digests | `artifacts` |
+| invoked command and workspace label | `argv`, `working_dir` |
+| start/end and process result | `started_at`, `completed_at`, `exit_code`, `classification` |
+| partial profiler output | `incomplete`, `limitations`, `truncated` |
+
+Receipts are decoded with unknown-field rejection and validated for schema,
+provenance, SHA-256 digests, timestamps, bounded output, incomplete status, and
+round-trip preservation. Attaching a receipt records the supplied observation;
+it never invokes the named producer.

docs/mcp-composed-workflows.md

@@ -31,6 +31,18 @@ telemetry.
 
 ## Issue or pull-request research brief
 
+For a complete source audit, the canonical route is:
+
+```text
+coverage -> explicit sync -> jobs.get -> offline reread -> duplicate checks -> live verification -> receipt attachment -> evidence/draft handoff
+```
+
+Coverage reads are offline and missing coverage is unknown. Synchronization is
+always an explicit bounded operation; after `jobs.get`, perform the offline
+reread and reuse its returned `corpus_revision` for any composed duplicate
+checks. Use exact resource URIs only through MCP `resources/read` before attaching
+receipts or handing evidence to a draft workflow.
+
 Canonical MCP composition:
 
 1. Use `workflow.prepare_issue_set` for supplied exact issues. It returns

docs/mcp-scalable-workflows.md

@@ -15,7 +15,7 @@ github.sync_repository_context -> jobs.get -> corpus.get_repositories
 research.query_deepwiki
 github.sync_threads -> jobs.get -> corpus.rank_contribution_candidates
 github.sync_thread_facets -> jobs.get -> corpus.get_thread_facets
-corpus.find_precedents -> workflow.find_related_work
+corpus.find_precedents
 workflow.prepare_issue_set
 ```
 
@@ -58,6 +58,13 @@ explicit `raw_query` field; there is no deprecated alias.
 - `corpus.find_clusters` and `corpus.find_neighbors` accept up to 20 repository
   or source-thread targets respectively. Their ordered item results isolate
   missing or invalid targets instead of forcing scalar retry loops.
+- Revision-bound offline reads return a `corpus_revision`. The revision-bearing
+  search, coverage, precedent, cluster, neighbor, portfolio, code, and issue-set
+  surfaces accept that value as a pin for a composed follow-up read. A stale pin
+  is an unavailable result; reread only after an explicit, bounded
+  synchronization job. Other point reads and durable resources may omit a
+  revision, so do not infer that they share a snapshot unless their contract
+  returns and accepts the pin. Reads never refresh the corpus implicitly.
 - `corpus.rank_contribution_candidates` requires one to 50 repositories. Its derived ranking is
   intentionally non-paginated; inspect `total` and `truncated`, then raise the
   limit or narrow the repository set when more candidates are needed. Per-repo
@@ -92,8 +99,9 @@ categories over an explicit observation window. It searches the local corpus
 first, refreshes only a bounded set of finalists whose merge outcome is
 unknown, and persists a typed report. `candidate_limit`, `hydration_limit`, and
 `representative_limit` bound search, network work, and returned context
-independently. Set `hydration_limit: 0` to request a strictly offline analysis;
-otherwise the workflow performs GitHub reads and idempotent local writes.
+independently. The durable operation always creates a job and persists its
+report. Use `corpus.preview_fix_patterns` when the analysis must be strictly
+offline and must create no job, artifact, hydration, or local write.
 
 Coverage reports candidate matches, unique pull requests, unknown outcomes
 before and after hydration, hydration failures, and candidate truncation.
@@ -106,6 +114,11 @@ Relationship and proof-style labels are bounded lexical projections, so the
 report preserves their supporting phrase and states that similarity is not
 causal proof.
 
+For an analysis that must not create a job, artifact, hydration, or write, use
+`corpus.preview_fix_patterns`. It returns `persisted: false`, zero hydration,
+and the captured `corpus_revision`. The durable operation remains the path for
+persisted reports.
+
 ## Exact issue-set preparation
 
 Use `workflow.prepare_issue_set` when the contribution is already scoped by
@@ -208,6 +221,31 @@ Repository and dossier absence have different recovery paths:
 
 Reading the dossier resource again cannot resolve either state.
 
+## Canonical source audit
+
+Use this order when producing a source-backed audit or contribution handoff:
+
+```text
+coverage -> explicit sync -> jobs.get -> offline reread -> duplicate checks -> live verification -> receipt attachment -> evidence/draft handoff
+```
+
+Start with `corpus.get_coverage` and treat missing or incomplete coverage as
+unknown. If current evidence is required, choose a bounded explicit GitHub
+sync, poll it with `jobs.get`, and then perform the offline reread. Use the
+reread's returned `corpus_revision` for duplicate checks over that same state.
+Perform live verification after local evidence selection, attach a producer-neutral
+validation receipt, and hand the exact resource and any returned revision
+references to the evidence or draft workflow. Resources that do not expose a
+revision are point-in-time reads and should be reread after an explicit sync.
+Larger persisted payloads are always read with MCP `resources/read` using the
+exact opaque URI returned by the tool.
+
+Completed code-index jobs return a typed artifact containing repository, commit
+SHA, corpus revision, manifest identity and digest, file/truncation counts, and
+an exact `gitcontribute://code-index/...` resource. Consume that URI through
+`resources/read`; do not infer an artifact identity from a repository name
+alone.
+
 `corpus.get_coverage` accepts up to 100 ordered repository or exact-thread
 targets. `jobs.cancel` accepts up to 100 IDs and returns isolated item outcomes;
 repeating cancellation is safe. `jobs.get` exposes structured phase and item

internal/app/acquisition.go

@@ -64,26 +64,31 @@ func (s *Service) Acquire(ctx context.Context, repo contracts.RepoRef, remote st
 		if confirmedCommit != acq.CommitSHA {
 			return nil, fmt.Errorf("acquired checkout changed before snapshot reuse: commit %q", confirmedCommit)
 		}
+		revision, err := c.CorpusRevision(ctx)
+		if err != nil {
+			return nil, err
+		}
 		return &contracts.AcquisitionResult{
-			Repo:          repo,
-			Remote:        acq.Remote,
-			DefaultBranch: acq.DefaultBranch,
-			CommitSHA:     acq.CommitSHA,
-			Files:         existing.Manifest.IndexedFiles,
-			Bytes:         existing.TotalBytes,
-			Indexed:       true,
-			Inserted:      false,
-			AcquiredAt:    formatTime(acq.AcquiredAt),
-			Message:       "acquired; snapshot already indexed",
-			IndexManifest: existing.Manifest,
+			Repo:           repo,
+			Remote:         acq.Remote,
+			DefaultBranch:  acq.DefaultBranch,
+			CommitSHA:      acq.CommitSHA,
+			Files:          existing.Manifest.IndexedFiles,
+			Bytes:          existing.TotalBytes,
+			Indexed:        true,
+			Inserted:       false,
+			AcquiredAt:     formatTime(acq.AcquiredAt),
+			Message:        "acquired; snapshot already indexed",
+			IndexManifest:  existing.Manifest,
+			CorpusRevision: revision,
 		}, nil
 	}
 
 	snapshot, err := codeindex.Index(ctx, acq.Path, codeindex.Options{})
 	if err != nil {
 		return nil, fmt.Errorf("index acquired checkout: %w", err)
 	}
-	_, inserted, err := c.StoreCodeSnapshot(ctx, ref, snapshot)
+	_, inserted, revision, err := c.StoreCodeSnapshotWithRevision(ctx, ref, snapshot)
 	if err != nil {
 		return nil, fmt.Errorf("store code snapshot: %w", err)
 	}
@@ -94,16 +99,17 @@ func (s *Service) Acquire(ctx context.Context, repo contracts.RepoRef, remote st
 	}
 
 	return &contracts.AcquisitionResult{
-		Repo:          repo,
-		Remote:        acq.Remote,
-		DefaultBranch: acq.DefaultBranch,
-		CommitSHA:     acq.CommitSHA,
-		Files:         len(snapshot.Documents),
-		Bytes:         snapshot.TotalBytes,
-		Indexed:       true,
-		Inserted:      inserted,
-		AcquiredAt:    formatTime(acq.AcquiredAt),
-		Message:       message,
-		IndexManifest: snapshot.Manifest,
+		Repo:           repo,
+		Remote:         acq.Remote,
+		DefaultBranch:  acq.DefaultBranch,
+		CommitSHA:      acq.CommitSHA,
+		Files:          len(snapshot.Documents),
+		Bytes:          snapshot.TotalBytes,
+		Indexed:        true,
+		Inserted:       inserted,
+		AcquiredAt:     formatTime(acq.AcquiredAt),
+		Message:        message,
+		IndexManifest:  snapshot.Manifest,
+		CorpusRevision: revision,
 	}, nil
 }

internal/app/hydration_refresh.go

@@ -48,6 +48,8 @@ func (s *Service) refreshHydrationThreadHeader(ctx context.Context, repo contrac
 	writer := &syncThreadWriter{
 		ctx:          ctx,
 		corpus:       c,
+		owner:        ref.Owner,
+		repo:         ref.Repo,
 		repositoryID: repository.ID,
 		kind:         kind,
 	}

internal/app/job_executor_reconciliation_test.go

@@ -87,16 +87,19 @@ func TestLiveOwnerNotReconciledByAnotherExecutor(t *testing.T) {
 	<-blocked
 	waitForJobStatus(t, jobsA, id, corpus.JobStatusRunning, 1*time.Second)
 
-	// A second process opens the database and reconciles with a 200ms lease.
-	// Because A heartbeats every 50ms, its job must remain running.
+	// A second process opens the database and reconciles with a lease that is
+	// deliberately much wider than the heartbeat interval. This keeps the
+	// assertion about ownership independent of scheduler jitter while still
+	// allowing the test to catch an owner that stops heartbeating.
+	const leaseTimeout = 2 * time.Second
 	cB, err := corpus.Open(ctx, svc.databasePath())
 	if err != nil {
 		t.Fatalf("open second corpus: %v", err)
 	}
 	defer func() { _ = cB.Close() }()
 
 	time.Sleep(100 * time.Millisecond)
-	if err := cB.ReconcileInterruptedJobs(ctx, 200*time.Millisecond); err != nil {
+	if err := cB.ReconcileInterruptedJobs(ctx, leaseTimeout); err != nil {
 		t.Fatalf("reconcile: %v", err)
 	}
 

internal/app/manifest.go

@@ -25,8 +25,9 @@ const manifestSnapshotAttempts = 3
 
 // ManifestOptions selects optional local workspace and exact stored PR inputs.
 type ManifestOptions struct {
-	WorkspaceID string
-	PullRequest *ManifestPullRequest
+	WorkspaceID    string
+	PullRequest    *ManifestPullRequest
+	CorpusRevision *int64
 }
 
 // ManifestPullRequest identifies one exact stored pull request.
@@ -38,42 +39,64 @@ type ManifestPullRequest struct {
 
 // ContributionManifest assembles and persists one bounded local evidence statement.
 func (s *Service) ContributionManifest(ctx context.Context, opportunityID string, opts ManifestOptions) (*manifest.Statement, error) {
+	statement, _, err := s.contributionManifestWithRevision(ctx, opportunityID, opts)
+	return statement, err
+}
+
+// contributionManifestWithRevision returns the revision validated while the
+// statement was assembled. Persisting the statement advances the corpus, so a
+// later standalone revision read would identify the write rather than the
+// evidence selected by this manifest.
+func (s *Service) contributionManifestWithRevision(ctx context.Context, opportunityID string, opts ManifestOptions) (*manifest.Statement, int64, error) {
 	c, err := s.openCorpus(ctx)
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 	for attempt := 0; attempt < manifestSnapshotAttempts; attempt++ {
+		revision, err := c.CorpusRevision(ctx)
+		if err != nil {
+			return nil, 0, err
+		}
+		if opts.CorpusRevision != nil && *opts.CorpusRevision != revision {
+			return nil, 0, &corpus.StaleCorpusRevisionError{Expected: *opts.CorpusRevision, Current: revision}
+		}
 		watch, err := c.BeginChangeWatch(ctx)
 		if err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 		statement, err := s.assembleContributionManifest(ctx, c, opportunityID, opts)
 		if err != nil {
 			_ = watch.Close()
-			return nil, err
+			return nil, 0, err
 		}
 		unchanged, watchErr := watch.Unchanged(ctx)
 		closeErr := watch.Close()
 		if watchErr != nil || closeErr != nil {
-			return nil, errors.Join(watchErr, closeErr)
+			return nil, 0, errors.Join(watchErr, closeErr)
 		}
 		if !unchanged {
 			continue
 		}
+		if err := c.RequireCorpusRevision(ctx, revision); err != nil {
+			if opts.CorpusRevision == nil && corpus.IsStaleCorpusRevision(err) {
+				continue
+			}
+			return nil, 0, err
+		}
 		if err := validateManifestReferences(statement.Predicate); err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 		pullRequestRef := ""
 		if statement.Predicate.PullRequest != nil {
 			pr := statement.Predicate.PullRequest
 			pullRequestRef = fmt.Sprintf("%s/%s#%d", pr.Owner, pr.Repo, pr.Number)
 		}
 		if err := c.SaveContributionManifest(ctx, statement, opts.WorkspaceID, pullRequestRef); err != nil {
-			return nil, err
+			return nil, 0, err
 		}
-		return statement, nil
+		return statement, revision, nil
 	}
-	return nil, errors.New("corpus changed while assembling contribution manifest; retry")
+	return nil, 0, errors.New("corpus changed while assembling contribution manifest; retry")
 }
 
 func (s *Service) assembleContributionManifest(ctx context.Context, c *corpus.Corpus, opportunityID string, opts ManifestOptions) (*manifest.Statement, error) {

internal/app/mcp.go

@@ -87,6 +87,7 @@ func (r *MCPReader) Search(ctx context.Context, in mcpcontract.SearchInput) (mcp
 		Limit:  in.Limit,
 		Cursor: in.Cursor,
 		Sort:   in.Sort, MatchMode: in.MatchMode,
+		CorpusRevision: in.CorpusRevision,
 	})
 	if err != nil {
 		return mcpcontract.SearchOutput{}, err
@@ -116,6 +117,7 @@ func (r *MCPReader) Search(ctx context.Context, in mcpcontract.SearchInput) (mcp
 			MatchSource:    m.MatchSource,
 			MatchExcerpt:   m.MatchExcerpt,
 			MatchTruncated: m.MatchTruncated,
+			CorpusRevision: res.CorpusRevision,
 		}
 		if in.View == "full" {
 			matches[i].Body = m.Body
@@ -132,6 +134,7 @@ func (r *MCPReader) Search(ctx context.Context, in mcpcontract.SearchInput) (mcp
 		Query: in.Query, QueryInterpretation: strings.Join(strings.Fields(in.Query), separator),
 		MatchMode: in.MatchMode, View: in.View, Total: res.Total, Matches: matches, NextCursor: res.NextCursor,
 		UnknownMergeCount: res.UnknownMergeCount,
+		CorpusRevision:    res.CorpusRevision,
 	}
 	if out.UnknownMergeCount > 0 {
 		out.Suggestion = "Some otherwise-matching pull requests have unknown merge state. Repeat without the merged filter to identify finalists, then hydrate pr_details before inferring absence."
@@ -175,6 +178,10 @@ func (r *MCPReader) Thread(ctx context.Context, in mcpcontract.ThreadInput) (mcp
 	if err != nil {
 		return mcpcontract.ThreadOutput{}, err
 	}
+	revision, err := beginCorpusRead(ctx, c, nil)
+	if err != nil {
+		return mcpcontract.ThreadOutput{}, err
+	}
 	repo, err := c.GetRepository(ctx, in.Owner, in.Repo)
 	if err != nil {
 		return mcpcontract.ThreadOutput{}, fmt.Errorf("get repository: %w", err)
@@ -192,6 +199,10 @@ func (r *MCPReader) Thread(ctx context.Context, in mcpcontract.ThreadInput) (mcp
 	out := corpusThreadToMCPOutput(thread)
 	out.Owner = in.Owner
 	out.Repo = in.Repo
+	out.CorpusRevision = revision
+	if err := finishCorpusRead(ctx, c, revision); err != nil {
+		return mcpcontract.ThreadOutput{}, err
+	}
 	return out, nil
 }
 
@@ -603,7 +614,11 @@ func (r *MCPReader) GetCoverage(ctx context.Context, in mcpcontract.GetCoverageI
 	if err != nil {
 		return mcpcontract.GetCoverageOutput{}, err
 	}
-	out := mcpcontract.GetCoverageOutput{Status: "complete", Items: make([]mcpcontract.BatchItem[mcpcontract.CoverageOutput], len(in.Targets))}
+	revision, err := beginCorpusRead(ctx, c, in.CorpusRevision)
+	if err != nil {
+		return mcpcontract.GetCoverageOutput{}, err
+	}
+	out := mcpcontract.GetCoverageOutput{Status: "complete", Items: make([]mcpcontract.BatchItem[mcpcontract.CoverageOutput], len(in.Targets)), CorpusRevision: revision}
 	for i, target := range in.Targets {
 		if err := ctx.Err(); err != nil {
 			return out, err
@@ -635,6 +650,9 @@ func (r *MCPReader) GetCoverage(ctx context.Context, in mcpcontract.GetCoverageI
 		}
 		out.Items[i] = item
 	}
+	if err := finishCorpusRead(ctx, c, revision); err != nil {
+		return mcpcontract.GetCoverageOutput{}, err
+	}
 	return out, nil
 }
 
@@ -653,8 +671,14 @@ func readCoverageTarget(ctx context.Context, c *corpus.Corpus, target mcpcontrac
 	if err := ref.Validate(); err != nil {
 		return mcpcontract.CoverageOutput{}, "invalid_reference", fmt.Errorf("%w: %w", errInvalidCoverageTarget, err)
 	}
-	isThread := target.Kind != "" || target.Number != 0
-	if isThread && ((target.Kind != "issue" && target.Kind != "pull_request") || target.Number < 1) {
+	isThread, valid := (target.Kind != "" || target.Number != 0), false
+	switch {
+	case target.Kind == "" && target.Number == 0:
+		valid = true
+	case target.Kind == "issue" || target.Kind == "pull_request":
+		valid = target.Number > 0
+	}
+	if !valid {
 		return mcpcontract.CoverageOutput{}, "invalid_reference", errInvalidCoverageTarget
 	}
 	repo, err := c.GetRepository(ctx, ref.Owner, ref.Repo)