Closed
Report actual sync HTTP attempt evidence #173
urandonAdd on 27 days agoClosed at 27 days ago
Report actual sync HTTP attempt evidence #173
Closed
urandonAdd on 27 days agoClosed at 27 days ago
Total: 25 files changed+1757-77
@@ -0,0 +1,141 @@
1+package main
2+ 
3+import (
4+ "bytes"
5+ "context"
6+ "errors"
7+ "io"
8+ "net/http"
9+ "path/filepath"
10+ "strings"
11+ "testing"
12+ "time"
13+ 
14+ "gitcode-mcp/internal/config"
15+)
16+ 
17+type rejectUnexpectedSyncTransport struct{ calls int }
18+ 
19+func (p *rejectUnexpectedSyncTransport) RoundTrip(*http.Request) (*http.Response, error) {
20+ p.calls++
21+ return nil, errors.New("fixture transport must not be entered")
22+}
23+ 
24+func TestCLISyncPreDispatchCancellationHasNoHTTPAttempt(t *testing.T) {
25+ for _, expired := range []bool{false, true} {
26+ name := "cancelled"
27+ if expired {
28+ name = "expired"
29+ }
30+ t.Run(name, func(t *testing.T) {
31+ src := newTestSource(t)
32+ src.env[config.EnvToken] = "fixture-credential"
33+ cfg := filepath.Join(t.TempDir(), "config.json")
34+ src.env[config.EnvConfigPath] = cfg
35+ src.files[cfg] = []byte(`{"max_retries":0}`)
36+ cachePath := filepath.Join(t.TempDir(), "cache.db")
37+ addRepoForStartupTest(t, cachePath, "https://provider.invalid/api/v5")
38+ transport := &rejectUnexpectedSyncTransport{}
39+ previousTransport := http.DefaultTransport
40+ http.DefaultTransport = transport
41+ defer func() { http.DefaultTransport = previousTransport }()
42+ previousRoute := cliRoute
43+ defer func() { cliRoute = previousRoute }()
44+ cliRoute = func(ctx context.Context, args []string, stdout, stderr io.Writer, deps StartupDeps) int {
45+ var bounded context.Context
46+ var cancel context.CancelFunc
47+ if expired {
48+ bounded, cancel = context.WithDeadline(ctx, time.Now().Add(-time.Second))
49+ } else {
50+ bounded, cancel = context.WithCancel(ctx)
51+ cancel()
52+ }
53+ defer cancel()
54+ return previousRoute(bounded, args, stdout, stderr, deps)
55+ }
56+ var stdout, stderr bytes.Buffer
57+ code := run([]string{"sync", "--live", "--issues", "--cache-path", cachePath, "--repo", "fixture-a"}, strings.NewReader(""), &stdout, &stderr, src)
58+ out := stdout.String() + stderr.String()
59+ if code != 1 || transport.calls != 0 {
60+ t.Fatal("pre-dispatch cancellation did not preserve zero transport entry")
61+ }
62+ if strings.Contains(out, "http_attempted: true") {
63+ t.Fatal("zero transport entry was reported as HTTP attempted")
64+ }
65+ if !strings.Contains(out, "http_attempted: false") {
66+ t.Fatal("missing explicit no-attempt diagnostic")
67+ }
68+ want := "operation_cancelled"
69+ if expired {
70+ want = "operation_timeout"
71+ }
72+ if !strings.Contains(out, "failure_class: "+want) {
73+ t.Fatal("pre-provider interruption must not be classified as a transport failure")
74+ }
75+ })
76+ }
77+}
78+ 
79+func TestCLISyncKeepsEmptyListHTTPBeforeLaterCancellation(t *testing.T) {
80+ src := newTestSource(t)
81+ src.env[config.EnvToken] = "fixture-credential"
82+ cfg := filepath.Join(t.TempDir(), "config.json")
83+ src.env[config.EnvConfigPath] = cfg
84+ src.files[cfg] = []byte(`{"max_retries":0}`)
85+ cachePath := filepath.Join(t.TempDir(), "cache.db")
86+ addRepoForStartupTest(t, cachePath, "https://provider.invalid")
87+ previousRoute := cliRoute
88+ t.Cleanup(func() { cliRoute = previousRoute })
89+ var cancel context.CancelFunc
90+ cliRoute = func(ctx context.Context, args []string, stdout, stderr io.Writer, deps StartupDeps) int {
91+ ctx, cancel = context.WithCancel(ctx)
92+ defer cancel()
93+ return previousRoute(ctx, args, stdout, stderr, deps)
94+ }
95+ calls := 0
96+ withStartupTransport(t, startupRoundTripper(func(req *http.Request) (*http.Response, error) {
97+ calls++
98+ cancel()
99+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(`[]`)), Request: req}, nil
100+ }))
101+ var stdout, stderr bytes.Buffer
102+ code := run([]string{"sync", "--live", "--issues", "--wiki", "--cache-path", cachePath, "--repo", "fixture-a"}, strings.NewReader(""), &stdout, &stderr, src)
103+ out := stdout.String() + stderr.String()
104+ if code != 1 || calls != 1 || !strings.Contains(out, "http_attempted: true") || !strings.Contains(out, "failure_class: operation_cancelled") {
105+ t.Fatal("late cancellation lost actual HTTP evidence or invented a transport failure")
106+ }
107+}
108+ 
109+func TestCLISyncParentCancellationKeepsEnteredFailure(t *testing.T) {
110+ for _, collection := range []string{"--issues", "--wiki"} {
111+ t.Run(collection, func(t *testing.T) {
112+ src := newTestSource(t)
113+ src.env[config.EnvToken] = "fixture-credential"
114+ cfg := filepath.Join(t.TempDir(), "config.json")
115+ src.env[config.EnvConfigPath] = cfg
116+ src.files[cfg] = []byte(`{"max_retries":0}`)
117+ cachePath := filepath.Join(t.TempDir(), "cache.db")
118+ addRepoForStartupTest(t, cachePath, "https://provider.invalid")
119+ previousRoute := cliRoute
120+ t.Cleanup(func() { cliRoute = previousRoute })
121+ var cancel context.CancelFunc
122+ cliRoute = func(ctx context.Context, args []string, stdout, stderr io.Writer, deps StartupDeps) int {
123+ ctx, cancel = context.WithCancel(ctx)
124+ defer cancel()
125+ return previousRoute(ctx, args, stdout, stderr, deps)
126+ }
127+ calls := 0
128+ withStartupTransport(t, startupRoundTripper(func(*http.Request) (*http.Response, error) {
129+ calls++
130+ cancel()
131+ return nil, context.Canceled
132+ }))
133+ var stdout, stderr bytes.Buffer
134+ code := run([]string{"sync", "--live", collection, "--max-pages", "1", "--cache-path", cachePath, "--repo", "fixture-a"}, strings.NewReader(""), &stdout, &stderr, src)
135+ out := stdout.String() + stderr.String()
136+ if code != 1 || calls != 1 || !strings.Contains(out, "http_attempted: true") || !strings.Contains(out, "failure_class: live_transport_failure") {
137+ t.Fatal("parent cancellation discarded entered transport failure")
138+ }
139+ })
140+ }
141+}
@@ -99,6 +99,15 @@ gitcode-mcp repo-docs rebind --repo owner/repo --registration-id REG \
99 --source-registration-generation 1 --repository-path /new/worktree99 --source-registration-generation 1 --repository-path /new/worktree
100```100```
101 101 
102+Rebind fences the previous source generation. A late reconcile completion or
103+scheduling failure from that old registration/source/path cannot change the
104+replacement's state, inspection history, retry/error state, active-job links or
105+polling timestamps. It returns the current sanitized authority without writing
106+the registry. Only a matching active job intent can mark the current source
107+`indexing`; current-generation failures still apply normally. Registry restart
108+preserves that authority through the existing exact source-id redirects. This
109+is shared by CLI, MCP and Admin status, not an observation-side repair.
110+ 
102Public status exposes only opaque Git111Public status exposes only opaque Git
103store/worktree refs. The coordinator then polls committed `HEAD` and policy112store/worktree refs. The coordinator then polls committed `HEAD` and policy
104identity every minute and coalesces a new immutable-set job when either113identity every minute and coalesces a new immutable-set job when either
@@ -167,10 +167,14 @@ survive strict replay; paths, source ids, content and raw safe-looking error
167classes remain excluded. Registration and source captures share one bounded167classes remain excluded. Registration and source captures share one bounded
168operation-local buffer and publication boundary.168operation-local buffer and publication boundary.
169Actual queued-job/rebind/fence and reconcile-snapshot/late-scheduling-failure169Actual queued-job/rebind/fence and reconcile-snapshot/late-scheduling-failure
170-fixtures pair obsolete owners with current-generation controls. They preserve170+fixtures pair obsolete owners with current-generation controls. Obsolete
171-existing execution results while requiring exactly one count-only loss for an171+completion and failure callbacks must preserve the entire replacement authority
172-obsolete changed pair, one confirmed source capture for a current pair, and no172+and scheduling state, perform zero registry writes, and manufacture neither a
173-extra boundary on unchanged replay. The held generation-fence test waits for an173+transition nor diagnostic loss. Current controls still apply normally. Restart
174+readback qualifies the exact existing source redirect and retains every other
175+source field. A deterministic registration/cache/repository/source/path/profile
176+mismatch matrix and active-job intent controls fence unrelated work. The held
177+generation-fence test waits for an
174accepted FIFO collector marker after registry save/unlock before asserting no178accepted FIFO collector marker after registry save/unlock before asserting no
175early maintenance publication; releasing the real fence conserves the transition.179early maintenance publication; releasing the real fence conserves the transition.
176Identity-resolution fixtures apply real legacy registry conflicts and physical180Identity-resolution fixtures apply real legacy registry conflicts and physical
@@ -210,6 +214,27 @@ construction failure: discovery and observation do not create/migrate the cache
210or resolve credentials. Later explicit live writes use offline HTTP fixtures to214or resolve credentials. Later explicit live writes use offline HTTP fixtures to
211prove provider confirmation, audit, and one POST on idempotent replay; absent215prove provider confirmation, audit, and one POST on idempotent replay; absent
212credentials/offline requests fail closed rather than selecting fixture writes.216credentials/offline requests fail closed rather than selecting fixture writes.
217+Sync attempt-evidence fixtures cancel or expire contexts before dispatch,
218+cancel rate-limit waits from their explicit callback, and inject failures at
219+`RoundTrip` entry rather than racing a whole-command millisecond deadline.
220+They distinguish zero-entry operation interruption from entered-transport
221+failure, retain an empty successful list before later cancellation, freeze
222+operation-local evidence independently of telemetry, and preserve typed error
223+unwraps and partial aggregation. CLI text/JSON and MCP structured diagnostics
224+must agree. Adapter tests preserve configured timeout and redirect policy.
225+Actual shared-service multi-collection controls succeed on the first issues
226+list, cancel the later PR request from its local rate-limit callback, and retain
227+overall HTTP evidence without classifying the unentered request as a transport
228+failure. Failed-request receipts have separate mixed-cause and unknown-branch
229+controls; prior confirmed provider/schema failures retain precedence.
230+Target-batch controls also confirm that already-cancelled tail targets carry
231+explicit skipped-before-dispatch evidence, while genuinely unknown failures
232+remain unknown. Earlier successful targets and their HTTP receipts are retained.
233+Actual parent-context cancellation at transport entry is exercised separately
234+for bounded issues, wiki, and PR synchronization, with successful-empty-read
235+controls. CLI/MCP exercise the same cancellation boundary. Private typed causes
236+survive queue-summary failures and nested collection aggregation without adding
237+public failure counts, serializing cause text, or retaining aggregate cycles.
213Safe-copy fixtures accept only returned catalog/incident evidence, reject raw238Safe-copy fixtures accept only returned catalog/incident evidence, reject raw
214free-form reasons, and preserve omitted unknown authority references. The239free-form reasons, and preserve omitted unknown authority references. The
215dedicated UI and integrated rollout remain separately qualified cohorts; see240dedicated UI and integrated rollout remain separately qualified cohorts; see
@@ -141,6 +141,31 @@ adapter_unavailable
141 141 
142**Fix:** Reads work offline from cache. Writes require network and token.142**Fix:** Reads work offline from cache. Writes require network and token.
143 143 
144+### Sync cancellation and HTTP attempt evidence
145+ 
146+CLI error diagnostics and MCP `sync_live` diagnostics distinguish
147+`operation_cancelled` / `operation_timeout` from `live_transport_failure`.
148+`http_attempted` reports entry into the configured HTTP transport, not selection
149+of live mode, request construction, time spent waiting for a rate limit, or
150+successful delivery to GitCode. It is not a packet count or write confirmation.
151+ 
152+A cancelled or expired context before transport entry reports
153+`http_attempted: false`. A timeout/cancellation returned by an entered transport
154+reports `live_transport_failure` with `http_attempted: true`. Cancellation later
155+in a multi-collection sync can report an operation interruption with
156+`http_attempted: true`: even an empty successful list already contacted the
157+provider. Partial results and retries retain that fact independently of metrics.
158+This also applies when a later request is cancelled in its local rate-limit
159+wait: an earlier successful request is not evidence that the failing request
160+entered transport. Confirmed provider/schema failures or an actually entered
161+failed request still retain their failure classification in mixed results.
162+CLI compact partial summaries expose the known bit; MCP lifecycle diagnostics
163+keep their existing `code` and add canonical `error_class` and `http_attempted`.
164+Startup configuration-read cancellation remains a configuration failure.
165+ 
166+Retry only after inspecting partial progress and the operation's idempotency
167+state. An HTTP attempt alone never proves that a remote write succeeded.
168+ 
144### Alias collision169### Alias collision
145 170 
146```text171```text
@@ -2629,6 +2629,9 @@ func dispatch(ctx context.Context, svc queryService, command string, args []stri
2629 }2629 }
2630 return attachSyncJob(ctx, client, job.ID, opts, stdout, stderr)2630 return attachSyncJob(ctx, client, job.ID, opts, stdout, stderr)
2631 }2631 }
2632+ // Collections in this action share attempt evidence, including an empty
2633+ // successful list before cancellation in a later collection.
2634+ ctx, _ = diagnostics.WithHTTPAttemptWitness(ctx)
2632 if syncSingleRecordRequested(opts) {2635 if syncSingleRecordRequested(opts) {
2633 if opts.input != "" && (opts.prComments || opts.comments) && syncRemoteAliasSurface(opts.input) == "pull_request" {2636 if opts.input != "" && (opts.prComments || opts.comments) && syncRemoteAliasSurface(opts.input) == "pull_request" {
2634 started := time.Now().UTC()2637 started := time.Now().UTC()
@@ -3722,10 +3725,13 @@ func mergeSyncError(existing error, result *service.SyncResourcesResult, err err
3722 if result == nil {3725 if result == nil {
3723 return existing3726 return existing
3724 }3727 }
3725- return &service.PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount}3728+ partial := &service.PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount}
3729+ partial.PreserveHTTPAttemptEvidence(existing, err)
3730+ return partial
3726}3731}
3727 3732 
3728type syncResourcesCompactSummary struct {3733type syncResourcesCompactSummary struct {
3734+ HTTPAttempted *bool `json:"http_attempted,omitempty"`
3729 Status string `json:"status"`3735 Status string `json:"status"`
3730 FailureClass string `json:"failure_class,omitempty"`3736 FailureClass string `json:"failure_class,omitempty"`
3731 SuccessCount int `json:"success_count"`3737 SuccessCount int `json:"success_count"`
@@ -3913,6 +3919,9 @@ func syncResourcesSummary(result *service.SyncResourcesResult, partial *service.
3913 summary.FailureGroups = syncFailureGroups(result.Failures)3919 summary.FailureGroups = syncFailureGroups(result.Failures)
3914 }3920 }
3915 if partial != nil {3921 if partial != nil {
3922+ if attempted, known := diagnostics.HTTPAttemptEvidence(partial); known {
3923+ summary.HTTPAttempted = &attempted
3924+ }
3916 summary.Status = "partial"3925 summary.Status = "partial"
3917 summary.Diagnostic = partial.Diagnostic3926 summary.Diagnostic = partial.Diagnostic
3918 summary.TotalRequested = partial.TotalRequested3927 summary.TotalRequested = partial.TotalRequested
@@ -3987,6 +3996,9 @@ func renderSyncResourcesSummaryText(w io.Writer, summary syncResourcesCompactSum
3987 summary.ZeroDeltaCount,3996 summary.ZeroDeltaCount,
3988 summary.Elapsed,3997 summary.Elapsed,
3989 )3998 )
3999+ if summary.HTTPAttempted != nil {
4000+ fmt.Fprintf(w, " http_attempted=%t", *summary.HTTPAttempted)
4001+ }
3990 if summary.PagesListed > 0 || summary.RecordsListed > 0 || summary.SkippedByWatermark > 0 {4002 if summary.PagesListed > 0 || summary.RecordsListed > 0 || summary.SkippedByWatermark > 0 {
3991 fmt.Fprintf(w, " pages_listed=%d records_listed=%d skipped_by_watermark=%d", summary.PagesListed, summary.RecordsListed, summary.SkippedByWatermark)4003 fmt.Fprintf(w, " pages_listed=%d records_listed=%d skipped_by_watermark=%d", summary.PagesListed, summary.RecordsListed, summary.SkippedByWatermark)
3992 }4004 }
@@ -4447,6 +4459,9 @@ func writeCommandError(stderr io.Writer, format string, plan startupPlan, err er
4447 if errors.As(err, &lockErr) {4459 if errors.As(err, &lockErr) {
4448 publicErr = lockErr4460 publicErr = lockErr
4449 }4461 }
4462+ if attempted, known := diagnostics.HTTPAttemptEvidence(err); known {
4463+ publicErr = diagnostics.RetainHTTPAttempt(publicErr, attempted)
4464+ }
4450 message := config.RedactDiagnostic(publicErr.Error(), config.OSSource{})4465 message := config.RedactDiagnostic(publicErr.Error(), config.OSSource{})
4451 var diagnostic diagnostics.Diagnostic4466 var diagnostic diagnostics.Diagnostic
4452 if plan.ProviderMode == "live-http" {4467 if plan.ProviderMode == "live-http" {
@@ -4595,8 +4610,11 @@ func diagnosticContext(plan startupPlan, err error) diagnostics.CommandContext {
4595 }4610 }
4596 var partialSync *service.PartialSyncError4611 var partialSync *service.PartialSyncError
4597 if errors.As(err, &partialSync) {4612 if errors.As(err, &partialSync) {
4613+ _, attemptKnown := diagnostics.HTTPAttemptEvidence(err)
4598 for _, failure := range partialSync.Errors {4614 for _, failure := range partialSync.Errors {
4599 switch failure.FailureClass {4615 switch failure.FailureClass {
4616+ case "network_timeout":
4617+ ctx.HTTPAttempted, ctx.TransportFailure = true, true
4600 case "schema_decode", "partial_response", "malformed_response", "unexpected_content_type":4618 case "schema_decode", "partial_response", "malformed_response", "unexpected_content_type":
4601 ctx.HTTPAttempted = true4619 ctx.HTTPAttempted = true
4602 ctx.SchemaDecodeFailure = true4620 ctx.SchemaDecodeFailure = true
@@ -4606,7 +4624,7 @@ func diagnosticContext(plan startupPlan, err error) diagnostics.CommandContext {
4606 switch partialSync.Diagnostic {4624 switch partialSync.Diagnostic {
4607 case service.SyncDiagnosticTimeout, service.SyncDiagnosticCancelled:4625 case service.SyncDiagnosticTimeout, service.SyncDiagnosticCancelled:
4608 ctx.HTTPAttempted = true4626 ctx.HTTPAttempted = true
4609- ctx.TransportFailure = true4627+ ctx.TransportFailure = ctx.TransportFailure || !attemptKnown
4610 ctx.FailureSource = string(partialSync.Diagnostic)4628 ctx.FailureSource = string(partialSync.Diagnostic)
4611 case service.SyncDiagnosticEmptyWiki:4629 case service.SyncDiagnosticEmptyWiki:
4612 ctx.FailureSource = string(partialSync.Diagnostic)4630 ctx.FailureSource = string(partialSync.Diagnostic)
@@ -0,0 +1,96 @@
1+package cli
2+ 
3+import (
4+ "bytes"
5+ "context"
6+ "encoding/json"
7+ "testing"
8+ "time"
9+ 
10+ "gitcode-mcp/internal/cache"
11+ "gitcode-mcp/internal/diagnostics"
12+ "gitcode-mcp/internal/gitcode"
13+ "gitcode-mcp/internal/service"
14+)
15+ 
16+func TestCLISyncAggregateKeepsKnownHTTPAndInterruption(t *testing.T) {
17+ negative := diagnostics.RetainHTTPAttempt(&service.PartialSyncError{Diagnostic: service.SyncDiagnosticCancelled}, false)
18+ positive := diagnostics.RetainHTTPAttempt(&service.PartialSyncError{SuccessCount: 1}, true)
19+ result := &service.SyncResourcesResult{SuccessCount: 1}
20+ merged := mergeSyncError(negative, result, positive)
21+ partial, ok := merged.(*service.PartialSyncError)
22+ if !ok || partial.Diagnostic != service.SyncDiagnosticCancelled {
23+ t.Fatal("aggregate lost partial pointer or interruption")
24+ }
25+ summary := syncResourcesSummary(result, partial, time.Now())
26+ if summary.HTTPAttempted == nil || !*summary.HTTPAttempted {
27+ t.Fatal("partial success summary lost positive HTTP evidence")
28+ }
29+ d := diagnostics.Classify(merged, diagnosticContext(startupPlan{ProviderMode: "live-http"}, merged))
30+ if d.Code != diagnostics.CodeOperationCancelled || !d.HTTPAttempted {
31+ t.Fatal("late cancellation must retain prior HTTP without inventing a transport failure")
32+ }
33+}
34+ 
35+func TestCLISyncLaterPreDispatchInterruptionKeepsPriorHTTP(t *testing.T) {
36+ for _, cause := range []error{context.Canceled, context.DeadlineExceeded} {
37+ for _, entered := range []int{0, 1} {
38+ network := gitcode.ErrNetworkUnavailable{Attempts: entered, Cause: cause}
39+ failure := service.ErrSyncFailure{Mode: "network_timeout", Cause: network}
40+ partial := &service.PartialSyncError{Errors: []service.ResourceError{{Err: failure, FailureClass: "network_timeout"}}, SuccessCount: 1, FailureCount: 1}
41+ partial.Errors = append(partial.Errors, service.ResourceError{Err: diagnostics.RetainPreDispatchInterruption(cause)})
42+ partial.FailureCount++
43+ partial.PreserveHTTPAttemptEvidence(diagnostics.RetainHTTPAttempt(failure, true))
44+ d := diagnostics.Classify(partial, diagnosticContext(startupPlan{ProviderMode: "live-http"}, partial))
45+ want := diagnostics.CodeOperationCancelled
46+ if cause == context.DeadlineExceeded {
47+ want = diagnostics.CodeOperationTimeout
48+ }
49+ if entered != 0 {
50+ want = diagnostics.CodeLiveTransportFailure
51+ }
52+ if d.Code != want || !d.HTTPAttempted {
53+ t.Fatalf("partial interruption mismatch: entered=%d class=%s attempted=%t", entered, d.Code, d.HTTPAttempted)
54+ }
55+ }
56+ }
57+}
58+ 
59+func TestCLISyncKnownNoHTTPJSONDiagnostic(t *testing.T) {
60+ for _, cause := range []error{context.Canceled, context.DeadlineExceeded} {
61+ var out bytes.Buffer
62+ err := diagnostics.RetainHTTPAttempt(cause, false)
63+ if writeCommandError(&out, "json", startupPlan{ProviderMode: "live-http"}, err) != 1 {
64+ t.Fatal("interruption must fail")
65+ }
66+ var d struct {
67+ FailureClass string `json:"failure_class"`
68+ HTTPAttempted *bool `json:"http_attempted"`
69+ }
70+ if json.Unmarshal(out.Bytes(), &d) != nil || d.HTTPAttempted == nil || *d.HTTPAttempted {
71+ t.Fatal("JSON diagnostic must explicitly report no HTTP attempt")
72+ }
73+ want := "operation_cancelled"
74+ if cause == context.DeadlineExceeded {
75+ want = "operation_timeout"
76+ }
77+ if d.FailureClass != want {
78+ t.Fatal("JSON diagnostic mislabeled pre-provider interruption")
79+ }
80+ }
81+}
82+ 
83+func TestCLIContentionSanitizationKeepsHTTPReceipt(t *testing.T) {
84+ for _, attempted := range []bool{false, true} {
85+ var out bytes.Buffer
86+ err := diagnostics.RetainHTTPAttempt(cache.ErrLockContention{}, attempted)
87+ writeCommandError(&out, "json", startupPlan{ProviderMode: "live-http"}, err)
88+ var d struct {
89+ FailureClass string `json:"failure_class"`
90+ HTTPAttempted *bool `json:"http_attempted"`
91+ }
92+ if json.Unmarshal(out.Bytes(), &d) != nil || d.FailureClass != "cache_busy" || d.HTTPAttempted == nil || *d.HTTPAttempted != attempted {
93+ t.Fatal("public-safe contention projection discarded actual attempt evidence")
94+ }
95+ }
96+}
@@ -1,7 +1,9 @@
1package diagnostics1package diagnostics
2 2 
3import (3import (
4+ "context"
4 "encoding/json"5 "encoding/json"
6+ "errors"
5 "fmt"7 "fmt"
6 "net/http"8 "net/http"
7 "sort"9 "sort"
@@ -35,6 +37,8 @@ const (
35 CodeWriteAmbiguousRemote Code = "write_ambiguous_remote"37 CodeWriteAmbiguousRemote Code = "write_ambiguous_remote"
36 CodeWriteAmbiguousReadback Code = "write_ambiguous_readback_failed"38 CodeWriteAmbiguousReadback Code = "write_ambiguous_readback_failed"
37 CodeWriteConflict Code = "write_conflict"39 CodeWriteConflict Code = "write_conflict"
40+ CodeOperationCancelled Code = "operation_cancelled"
41+ CodeOperationTimeout Code = "operation_timeout"
38)42)
39 43 
40type Diagnostic struct {44type Diagnostic struct {
@@ -95,8 +99,15 @@ func (c Classifier) Classify(err error, ctx CommandContext) Diagnostic {
95 if len(ctx.SensitiveTerms) > 0 {99 if len(ctx.SensitiveTerms) > 0 {
96 filter.SensitiveTerms = append(filter.SensitiveTerms, ctx.SensitiveTerms...)100 filter.SensitiveTerms = append(filter.SensitiveTerms, ctx.SensitiveTerms...)
97 }101 }
102+ attempted, known := HTTPAttemptEvidence(err)
103+ if known {
104+ ctx.HTTPAttempted = attempted
105+ }
98 code := classifyCode(err, ctx)106 code := classifyCode(err, ctx)
99 d := Diagnostic{Code: code, Message: messageFor(code, err), ExitClass: exitClassFor(code), HTTPAttempted: httpAttemptedFor(code, ctx), Retryable: retryableFor(code) || ctx.RetryableProviderFailure}107 d := Diagnostic{Code: code, Message: messageFor(code, err), ExitClass: exitClassFor(code), HTTPAttempted: httpAttemptedFor(code, ctx), Retryable: retryableFor(code) || ctx.RetryableProviderFailure}
108+ if known {
109+ d.HTTPAttempted = attempted
110+ }
100 d.Message = filter.RedactText(d.Message)111 d.Message = filter.RedactText(d.Message)
101 d.Context = redactedContext(filter, ctx)112 d.Context = redactedContext(filter, ctx)
102 return d113 return d
@@ -156,6 +167,11 @@ func classifyCode(err error, ctx CommandContext) Code {
156 if ctx.InvalidSelectedAPIBaseURL || hasCode(err, "invalid_api_base_url") {167 if ctx.InvalidSelectedAPIBaseURL || hasCode(err, "invalid_api_base_url") {
157 return CodeConfigCredential168 return CodeConfigCredential
158 }169 }
170+ if attempted, known := HTTPAttemptEvidence(err); known && !attempted {
171+ if code := operationInterruptionCode(err); code != "" {
172+ return code
173+ }
174+ }
159 if hasCode(err, "push_mirror_sync_in_progress") && ctx.HTTPAttempted {175 if hasCode(err, "push_mirror_sync_in_progress") && ctx.HTTPAttempted {
160 return CodePushMirrorSyncInProgress176 return CodePushMirrorSyncInProgress
161 }177 }
@@ -189,12 +205,27 @@ func classifyCode(err error, ctx CommandContext) Code {
189 if hasCode(err, "schema_decode") || hasCode(err, "malformed_response") || hasCode(err, "unexpected_content_type") {205 if hasCode(err, "schema_decode") || hasCode(err, "malformed_response") || hasCode(err, "unexpected_content_type") {
190 return CodeSchemaDecode206 return CodeSchemaDecode
191 }207 }
208+ // Earlier successful HTTP must not turn a later pre-dispatch interruption
209+ // into a transport failure. Preserve confirmed provider/schema failures
210+ // above, and any actually entered failed request in a mixed aggregate.
211+ if attempted, known := HTTPTransportFailureEvidence(err); known && !attempted {
212+ if code := operationInterruptionCode(err); code != "" {
213+ return code
214+ }
215+ }
192 if (ctx.TransportFailure || hasCode(err, "live_transport_failure") || hasCode(err, "network_unavailable") || hasCode(err, "write_network_unavailable")) && ctx.HTTPAttempted {216 if (ctx.TransportFailure || hasCode(err, "live_transport_failure") || hasCode(err, "network_unavailable") || hasCode(err, "write_network_unavailable")) && ctx.HTTPAttempted {
193 return CodeLiveTransportFailure217 return CodeLiveTransportFailure
194 }218 }
195 if ctx.HTTPStatus >= 500 && ctx.HTTPAttempted {219 if ctx.HTTPStatus >= 500 && ctx.HTTPAttempted {
196 return CodeLiveTransportFailure220 return CodeLiveTransportFailure
197 }221 }
222+ // Preserve confirmed provider/schema failures in mixed partial results.
223+ // A later local interruption is not itself an HTTP transport failure.
224+ if attempted, known := HTTPAttemptEvidence(err); known && attempted {
225+ if code := operationInterruptionCode(err); code != "" {
226+ return code
227+ }
228+ }
198 if ctx.FixtureFallbackSentinel || hasCode(err, "fixture_fallback_detected") || hasCode(err, "write_fixture_fallback_detected") || hasCode(err, "fixture_read_only") || hasCode(err, "write_fixture_read_only") {229 if ctx.FixtureFallbackSentinel || hasCode(err, "fixture_fallback_detected") || hasCode(err, "write_fixture_fallback_detected") || hasCode(err, "fixture_read_only") || hasCode(err, "write_fixture_read_only") {
199 return CodeFixtureFallbackDetected230 return CodeFixtureFallbackDetected
200 }231 }
@@ -207,6 +238,21 @@ func classifyCode(err error, ctx CommandContext) Code {
207 return CodeConfigurationError238 return CodeConfigurationError
208}239}
209 240 
241+func operationInterruptionCode(err error) Code {
242+ var interrupted interface{ OperationInterruption() string }
243+ interruption := ""
244+ if errors.As(err, &interrupted) {
245+ interruption = interrupted.OperationInterruption()
246+ }
247+ if errors.Is(err, context.Canceled) || interruption == "cancelled" {
248+ return CodeOperationCancelled
249+ }
250+ if errors.Is(err, context.DeadlineExceeded) || interruption == "timeout" {
251+ return CodeOperationTimeout
252+ }
253+ return ""
254+}
255+ 
210func codeFromError(err error) Code {256func codeFromError(err error) Code {
211 for _, code := range diagnosticCodes(err) {257 for _, code := range diagnosticCodes(err) {
212 switch code {258 switch code {
@@ -330,6 +376,8 @@ func exitClassFor(code Code) string {
330 return "configuration"376 return "configuration"
331 case CodeLiveTransportFailure:377 case CodeLiveTransportFailure:
332 return "transport"378 return "transport"
379+ case CodeOperationCancelled, CodeOperationTimeout:
380+ return "operation"
333 case CodeLiveAPIFailure, CodeAPIFailure, CodePushMirrorSyncInProgress:381 case CodeLiveAPIFailure, CodeAPIFailure, CodePushMirrorSyncInProgress:
334 return "provider"382 return "provider"
335 case CodePRReviewAnchorMismatch, CodeWriteConfirmationMissing, CodeWriteAmbiguousRemote, CodeWriteAmbiguousReadback, CodeWriteConflict:383 case CodePRReviewAnchorMismatch, CodeWriteConfirmationMissing, CodeWriteAmbiguousRemote, CodeWriteAmbiguousReadback, CodeWriteConflict:
@@ -366,6 +414,10 @@ func messageFor(code Code, err error) string {
366 base += ": live provider rejected credentials"414 base += ": live provider rejected credentials"
367 case CodeLiveTransportFailure:415 case CodeLiveTransportFailure:
368 base += ": live provider request failed before a valid API response"416 base += ": live provider request failed before a valid API response"
417+ case CodeOperationCancelled:
418+ base += ": operation cancelled"
419+ case CodeOperationTimeout:
420+ base += ": operation deadline expired"
369 case CodeLiveAPIFailure:421 case CodeLiveAPIFailure:
370 base += ": live provider returned an API failure"422 base += ": live provider returned an API failure"
371 case CodeAPIFailure:423 case CodeAPIFailure:
@@ -0,0 +1,130 @@
1+package diagnostics
2+ 
3+import (
4+ "context"
5+ "sync/atomic"
6+)
7+ 
8+type httpAttemptKey struct{}
9+ 
10+// HTTPAttemptWitness is one bounded operation-local bit, independent of
11+// telemetry. Nested service calls share the enclosing operation's witness.
12+// It retains no request, URL, identity, payload, or error text.
13+type HTTPAttemptWitness struct{ attempted atomic.Bool }
14+ 
15+func WithHTTPAttemptWitness(ctx context.Context) (context.Context, *HTTPAttemptWitness) {
16+ if witness, ok := ctx.Value(httpAttemptKey{}).(*HTTPAttemptWitness); ok {
17+ return ctx, witness
18+ }
19+ witness := &HTTPAttemptWitness{}
20+ return context.WithValue(ctx, httpAttemptKey{}, witness), witness
21+}
22+ 
23+func RecordHTTPAttempt(ctx context.Context) {
24+ if witness, ok := ctx.Value(httpAttemptKey{}).(*HTTPAttemptWitness); ok {
25+ witness.attempted.Store(true)
26+ }
27+}
28+ 
29+func (w *HTTPAttemptWitness) Attempted() bool { return w != nil && w.attempted.Load() }
30+ 
31+type httpAttemptError struct {
32+ cause error
33+ attempted bool
34+}
35+ 
36+func (e httpAttemptError) Error() string { return e.cause.Error() }
37+func (e httpAttemptError) Unwrap() error { return e.cause }
38+func (e httpAttemptError) HTTPAttemptEvidence() (bool, bool) { return e.attempted, true }
39+ 
40+func (e httpAttemptError) DiagnosticCode() string {
41+ if codes := diagnosticCodes(e.cause); len(codes) != 0 {
42+ return codes[0]
43+ }
44+ return ""
45+}
46+ 
47+// RetainHTTPAttempt freezes the bit at return, preserving the underlying error.
48+func RetainHTTPAttempt(err error, attempted bool) error {
49+ if err == nil {
50+ return nil
51+ }
52+ return httpAttemptError{cause: err, attempted: attempted}
53+}
54+ 
55+type preDispatchInterruption struct{ cause error }
56+ 
57+func (e preDispatchInterruption) Error() string { return e.cause.Error() }
58+func (e preDispatchInterruption) Unwrap() error { return e.cause }
59+func (e preDispatchInterruption) HTTPAttemptEvidence() (bool, bool) { return false, true }
60+func (e preDispatchInterruption) HTTPTransportFailureEvidence() (bool, bool) { return false, true }
61+ 
62+// RetainPreDispatchInterruption is used only by an owner's context guard that
63+// skips dispatch. It must not be applied to errors returned by a provider.
64+// Unknown/wrapped causes are left unchanged rather than inferred as local.
65+func RetainPreDispatchInterruption(err error) error {
66+ if err != context.Canceled && err != context.DeadlineExceeded {
67+ return err
68+ }
69+ return preDispatchInterruption{cause: err}
70+}
71+ 
72+// HTTPAttemptEvidence returns known evidence, not an inference from error text.
73+// An aggregate is definitely attempted if any child is; definitely not only
74+// when every child has known negative evidence. Outer operation receipts win.
75+func HTTPAttemptEvidence(err error) (attempted, known bool) {
76+ if err == nil {
77+ return false, false
78+ }
79+ if evidence, ok := err.(interface{ HTTPAttemptEvidence() (bool, bool) }); ok {
80+ if attempted, known = evidence.HTTPAttemptEvidence(); known {
81+ return attempted, true
82+ }
83+ }
84+ switch e := err.(type) {
85+ case interface{ Unwrap() []error }:
86+ children := e.Unwrap()
87+ allKnown := len(children) != 0
88+ for _, child := range children {
89+ attempted, known := HTTPAttemptEvidence(child)
90+ if attempted && known {
91+ return true, true
92+ }
93+ allKnown = allKnown && known
94+ }
95+ return false, allKnown
96+ case interface{ Unwrap() error }:
97+ return HTTPAttemptEvidence(e.Unwrap())
98+ }
99+ return false, false
100+}
101+ 
102+// HTTPTransportFailureEvidence describes failed requests, not the enclosing
103+// operation's HTTP receipt. A positive failed-request receipt in any branch
104+// wins; a negative is authoritative only when every branch is known negative.
105+func HTTPTransportFailureEvidence(err error) (attempted, known bool) {
106+ if err == nil {
107+ return false, false
108+ }
109+ if evidence, ok := err.(interface{ HTTPTransportFailureEvidence() (bool, bool) }); ok {
110+ if attempted, known = evidence.HTTPTransportFailureEvidence(); known {
111+ return attempted, true
112+ }
113+ }
114+ switch e := err.(type) {
115+ case interface{ Unwrap() []error }:
116+ children := e.Unwrap()
117+ allKnown := len(children) != 0
118+ for _, child := range children {
119+ attempted, known := HTTPTransportFailureEvidence(child)
120+ if attempted && known {
121+ return true, true
122+ }
123+ allKnown = allKnown && known
124+ }
125+ return false, allKnown
126+ case interface{ Unwrap() error }:
127+ return HTTPTransportFailureEvidence(e.Unwrap())
128+ }
129+ return false, false
130+}
@@ -0,0 +1,145 @@
1+package diagnostics
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "sync"
8+ "testing"
9+)
10+ 
11+func TestHTTPAttemptWitnessIsBoundedSharedAndFrozen(t *testing.T) {
12+ ctx, witness := WithHTTPAttemptWitness(context.Background())
13+ nested, shared := WithHTTPAttemptWitness(ctx)
14+ if witness != shared || witness.Attempted() {
15+ t.Fatal("nested witness must share one initially false bit")
16+ }
17+ before := RetainHTTPAttempt(context.Canceled, witness.Attempted())
18+ var wg sync.WaitGroup
19+ for i := 0; i < 8; i++ {
20+ wg.Add(1)
21+ go func() { defer wg.Done(); RecordHTTPAttempt(nested) }()
22+ }
23+ wg.Wait()
24+ if !witness.Attempted() {
25+ t.Fatal("transport entry not retained")
26+ }
27+ if attempted, known := HTTPAttemptEvidence(before); !known || attempted || !errors.Is(before, context.Canceled) {
28+ t.Fatal("receipt changed after return or lost cause")
29+ }
30+ positive := RetainHTTPAttempt(context.DeadlineExceeded, true)
31+ for _, err := range []error{positive, fmt.Errorf("outer: %w", positive), errors.Join(before, positive), errors.Join(errors.New("unknown"), positive)} {
32+ if attempted, known := HTTPAttemptEvidence(err); !known || !attempted {
33+ t.Fatal("positive aggregate evidence lost")
34+ }
35+ }
36+ if _, known := HTTPAttemptEvidence(errors.Join(before, errors.New("unknown"))); known {
37+ t.Fatal("unknown child is not negative evidence")
38+ }
39+ if attempted, known := HTTPAttemptEvidence(errors.Join(before, before)); !known || attempted {
40+ t.Fatal("known negative aggregate lost")
41+ }
42+}
43+ 
44+func TestHTTPAttemptEvidenceOverridesInference(t *testing.T) {
45+ for _, tc := range []struct {
46+ cause error
47+ attempted bool
48+ code Code
49+ }{
50+ {context.Canceled, false, CodeOperationCancelled},
51+ {context.DeadlineExceeded, false, CodeOperationTimeout},
52+ {context.Canceled, true, CodeLiveTransportFailure},
53+ {context.DeadlineExceeded, true, CodeLiveTransportFailure},
54+ } {
55+ d := Classify(RetainHTTPAttempt(tc.cause, tc.attempted), CommandContext{ProviderMode: "live-http", HTTPAttempted: !tc.attempted, TransportFailure: true})
56+ if d.Code != tc.code || d.HTTPAttempted != tc.attempted {
57+ t.Fatal("classification contradicts transport evidence")
58+ }
59+ }
60+ // Configuration failures remain configuration failures even with a
61+ // cancellation cause; startup never becomes a network incident.
62+ d := Classify(RetainHTTPAttempt(context.Canceled, false), CommandContext{ProviderMode: "live-http", BroaderConfigurationInvalid: true})
63+ if d.Code != CodeConfigCredential || d.HTTPAttempted {
64+ t.Fatal("configuration precedence changed")
65+ }
66+ for _, ctx := range []CommandContext{
67+ {ProviderMode: "live-http", SchemaDecodeFailure: true, HTTPStatus: 200},
68+ {ProviderMode: "live-http", APIFailure: true, HTTPStatus: 400},
69+ } {
70+ ctx.HTTPAttempted = true
71+ want := Classify(errors.New("fixture provider failure"), ctx).Code
72+ d := Classify(RetainHTTPAttempt(context.Canceled, true), ctx)
73+ if d.Code != want || !d.HTTPAttempted {
74+ t.Fatal("late interruption masked prior provider/schema failure evidence")
75+ }
76+ }
77+}
78+ 
79+type failedRequestReceipt struct {
80+ cause error
81+ entered bool
82+}
83+ 
84+func (e failedRequestReceipt) Error() string { return "fixture failed request" }
85+func (e failedRequestReceipt) Unwrap() error { return e.cause }
86+func (e failedRequestReceipt) DiagnosticCode() string { return "network_unavailable" }
87+func (e failedRequestReceipt) HTTPTransportFailureEvidence() (bool, bool) { return e.entered, true }
88+ 
89+func TestHTTPTransportFailureEvidenceIsNotOperationReceipt(t *testing.T) {
90+ local := failedRequestReceipt{cause: context.Canceled}
91+ entered := failedRequestReceipt{cause: context.DeadlineExceeded, entered: true}
92+ for _, tc := range []struct {
93+ err error
94+ attempted, known bool
95+ code Code
96+ }{
97+ {local, false, true, CodeOperationCancelled},
98+ {fmt.Errorf("outer: %w", local), false, true, CodeOperationCancelled},
99+ {errors.Join(local, local), false, true, CodeOperationCancelled},
100+ {errors.Join(local, RetainPreDispatchInterruption(context.Canceled)), false, true, CodeOperationCancelled},
101+ {errors.Join(local, RetainPreDispatchInterruption(context.DeadlineExceeded)), false, true, CodeOperationCancelled},
102+ {errors.Join(local, entered), true, true, CodeLiveTransportFailure},
103+ {errors.Join(entered, local), true, true, CodeLiveTransportFailure},
104+ {errors.Join(local, errors.New("unknown failure")), false, false, CodeLiveTransportFailure},
105+ } {
106+ err := RetainHTTPAttempt(tc.err, true)
107+ attempted, known := HTTPTransportFailureEvidence(err)
108+ if attempted != tc.attempted || known != tc.known {
109+ t.Fatal("outer operation receipt or unknown branch masked failed-request evidence")
110+ }
111+ d := Classify(err, CommandContext{ProviderMode: "live-http", TransportFailure: true})
112+ if d.Code != tc.code || !d.HTTPAttempted {
113+ t.Fatalf("failed-request evidence mismatch: class=%s attempted=%t", d.Code, d.HTTPAttempted)
114+ }
115+ }
116+ for _, ctx := range []CommandContext{
117+ {ProviderMode: "live-http", SchemaDecodeFailure: true, HTTPStatus: 200},
118+ {ProviderMode: "live-http", APIFailure: true, HTTPStatus: 400},
119+ } {
120+ ctx.HTTPAttempted = true
121+ want := Classify(errors.New("fixture provider failure"), ctx).Code
122+ d := Classify(RetainHTTPAttempt(local, true), ctx)
123+ if d.Code != want || !d.HTTPAttempted {
124+ t.Fatal("local cancellation masked confirmed provider/schema evidence")
125+ }
126+ }
127+}
128+ 
129+func TestPreDispatchInterruptionRequiresOwnerGuardEvidence(t *testing.T) {
130+ for _, cause := range []error{context.Canceled, context.DeadlineExceeded} {
131+ err := RetainPreDispatchInterruption(cause)
132+ attempted, known := HTTPTransportFailureEvidence(err)
133+ if !known || attempted || !errors.Is(err, cause) {
134+ t.Fatal("owner guard lost known pre-dispatch evidence or typed cause")
135+ }
136+ if attempted, known := HTTPAttemptEvidence(err); !known || attempted {
137+ t.Fatal("skipped request is not HTTP")
138+ }
139+ }
140+ for _, err := range []error{nil, errors.New("unknown"), fmt.Errorf("unknown boundary: %w", context.Canceled)} {
141+ if RetainPreDispatchInterruption(err) != err {
142+ t.Fatal("unknown boundary was inferred as skipped dispatch")
143+ }
144+ }
145+}
@@ -22,6 +22,16 @@ func (e ErrNetworkUnavailable) Error() string {
22 22 
23func (e ErrNetworkUnavailable) Unwrap() error { return e.Cause }23func (e ErrNetworkUnavailable) Unwrap() error { return e.Cause }
24 24 
25+func (e ErrNetworkUnavailable) HTTPAttemptEvidence() (bool, bool) {
26+ return e.Attempts > 0, true
27+}
28+ 
29+// The failing request has its own transport evidence. An enclosing operation
30+// may already have made successful requests before this request was cancelled.
31+func (e ErrNetworkUnavailable) HTTPTransportFailureEvidence() (bool, bool) {
32+ return e.Attempts > 0, true
33+}
34+ 
25func (e ErrNetworkUnavailable) DiagnosticCode() string { return "network_unavailable" }35func (e ErrNetworkUnavailable) DiagnosticCode() string { return "network_unavailable" }
26 36 
27type ErrRateLimited struct {37type ErrRateLimited struct {
@@ -0,0 +1,164 @@
1+package gitcode
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "io"
7+ "net/http"
8+ "strings"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+ 
13+ "gitcode-mcp/internal/diagnostics"
14+)
15+ 
16+type attemptRoundTripper func(*http.Request) (*http.Response, error)
17+ 
18+func (f attemptRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
19+ 
20+type legacyAttemptTransport struct{ cancelled *http.Request }
21+ 
22+func (t *legacyAttemptTransport) RoundTrip(*http.Request) (*http.Response, error) {
23+ return nil, context.Canceled
24+}
25+func (t *legacyAttemptTransport) CancelRequest(req *http.Request) { t.cancelled = req }
26+ 
27+func TestHTTPAttemptWrapperPreservesLegacyCancellation(t *testing.T) {
28+ inner := &legacyAttemptTransport{}
29+ req, _ := http.NewRequest(http.MethodGet, "https://provider.invalid/fixture", nil)
30+ wrapper := httpAttemptTransport{inner: inner}
31+ wrapper.CancelRequest(req)
32+ if inner.cancelled != req {
33+ t.Fatal("legacy client cancellation hook lost")
34+ }
35+}
36+ 
37+func TestHTTPAttemptBoundaryBeforeAndInsideTransport(t *testing.T) {
38+ for _, tc := range []struct {
39+ name string
40+ before bool
41+ cause error
42+ }{
43+ {"cancelled_before", true, context.Canceled},
44+ {"deadline_before", true, context.DeadlineExceeded},
45+ {"cancelled_inside", false, context.Canceled},
46+ {"deadline_inside", false, context.DeadlineExceeded},
47+ } {
48+ t.Run(tc.name, func(t *testing.T) {
49+ client, err := NewHTTPClient(Config{BaseURL: "https://provider.invalid", MaxRetries: 0})
50+ if err != nil {
51+ t.Fatal("client setup failed")
52+ }
53+ var calls atomic.Int32
54+ client.client.Transport = attemptRoundTripper(func(*http.Request) (*http.Response, error) { calls.Add(1); return nil, tc.cause })
55+ ctx := context.Background()
56+ if tc.before {
57+ if tc.cause == context.Canceled {
58+ var cancel context.CancelFunc
59+ ctx, cancel = context.WithCancel(ctx)
60+ cancel()
61+ } else {
62+ var cancel context.CancelFunc
63+ ctx, cancel = context.WithDeadline(ctx, time.Now().Add(-time.Second))
64+ defer cancel()
65+ }
66+ }
67+ ctx, witness := diagnostics.WithHTTPAttemptWitness(ctx)
68+ _, err = client.GetIssue(ctx, IssueRequest{Owner: "example-owner", Repo: "example-repo", Number: 1})
69+ var network ErrNetworkUnavailable
70+ if !errors.As(err, &network) || !errors.Is(err, tc.cause) {
71+ t.Fatal("network wrapper lost interruption cause")
72+ }
73+ want := 1
74+ if tc.before {
75+ want = 0
76+ }
77+ if int(calls.Load()) != want || network.Attempts != want || witness.Attempted() != (want != 0) {
78+ t.Fatal("HTTP count must match actual transport entry independently of telemetry")
79+ }
80+ })
81+ }
82+}
83+ 
84+func TestHTTPAttemptBoundaryRequestConstructionAndRateWait(t *testing.T) {
85+ client, _ := NewHTTPClient(Config{BaseURL: "https://provider.invalid", RateLimitRPS: 0.001, RateLimitBurst: 1})
86+ var calls atomic.Int32
87+ client.client.Transport = attemptRoundTripper(func(*http.Request) (*http.Response, error) {
88+ calls.Add(1)
89+ return nil, errors.New("unexpected transport")
90+ })
91+ ctx, witness := diagnostics.WithHTTPAttemptWitness(context.Background())
92+ observation := newProviderMetricAttempt(ctx)
93+ _, err := client.do(ctx, "invalid method", "/fixture", nil, nil, requestOptions{metricAttempt: observation})
94+ if err == nil || witness.Attempted() || observation.httpEntered.Load() || calls.Load() != 0 {
95+ t.Fatal("request construction is not an HTTP attempt")
96+ }
97+ // Consume the initial token locally; cancel deterministically from the
98+ // throttle callback before entering transport, without a timing sleep.
99+ client.rateLimiter.reserve(time.Now())
100+ ctx, cancel := context.WithCancel(ctx)
101+ defer cancel()
102+ ctx = WithRateLimitObserver(ctx, func(event RateLimitEvent) {
103+ if event.Type == RateLimitEventThrottleWaitStarted {
104+ cancel()
105+ }
106+ })
107+ _, err = client.GetIssue(ctx, IssueRequest{Owner: "example-owner", Repo: "example-repo", Number: 1})
108+ var network ErrNetworkUnavailable
109+ if !errors.As(err, &network) || network.Attempts != 0 || witness.Attempted() || calls.Load() != 0 {
110+ t.Fatal("rate wait is not an HTTP attempt")
111+ }
112+}
113+ 
114+func TestHTTPAttemptWrapperPreservesClientRedirectPolicy(t *testing.T) {
115+ client, _ := NewHTTPClient(Config{BaseURL: "https://provider.invalid", Timeout: time.Second})
116+ ctx, witness := diagnostics.WithHTTPAttemptWitness(context.Background())
117+ var calls, redirects int
118+ transport := attemptRoundTripper(func(req *http.Request) (*http.Response, error) {
119+ calls++
120+ if _, ok := req.Context().Deadline(); !ok {
121+ t.Error("configured client timeout lost")
122+ }
123+ if calls == 1 {
124+ return &http.Response{StatusCode: 302, Header: http.Header{"Location": {"/redirect"}}, Body: io.NopCloser(strings.NewReader("")), Request: req}, nil
125+ }
126+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(`{"id":1,"number":1}`)), Request: req}, nil
127+ })
128+ client.client.Transport = transport
129+ client.client.CheckRedirect = func(req *http.Request, via []*http.Request) error { redirects++; return nil }
130+ original := client.client
131+ _, _, err := client.getBytes(ctx, "/fixture", nil)
132+ if err != nil || calls != 2 || redirects != 1 || !witness.Attempted() || client.client != original {
133+ t.Fatalf("client behavior changed: error type=%T calls=%d redirects=%d attempted=%t", err, calls, redirects, witness.Attempted())
134+ }
135+}
136+ 
137+func TestHTTPAttemptPriorSuccessThenLocalRateCancellation(t *testing.T) {
138+ client, _ := NewHTTPClient(Config{BaseURL: "https://provider.invalid", MaxRetries: 0, RateLimitRPS: 0.001, RateLimitBurst: 1})
139+ var calls atomic.Int32
140+ client.client.Transport = attemptRoundTripper(func(req *http.Request) (*http.Response, error) {
141+ calls.Add(1)
142+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(`{"id":1,"number":1}`)), Request: req}, nil
143+ })
144+ ctx, cancel := context.WithCancel(context.Background())
145+ defer cancel()
146+ ctx, witness := diagnostics.WithHTTPAttemptWitness(ctx)
147+ if _, _, err := client.getBytes(ctx, "/fixture", nil); err != nil {
148+ t.Fatal("initial request failed")
149+ }
150+ ctx = WithRateLimitObserver(ctx, func(event RateLimitEvent) {
151+ if event.Type == RateLimitEventThrottleWaitStarted {
152+ cancel()
153+ }
154+ })
155+ _, err := client.GetIssue(ctx, IssueRequest{Owner: "example-owner", Repo: "example-repo", Number: 2})
156+ var network ErrNetworkUnavailable
157+ if !errors.As(err, &network) || network.Attempts != 0 || !errors.Is(err, context.Canceled) || calls.Load() != 1 || !witness.Attempted() {
158+ t.Fatal("later request must stop before transport without erasing prior HTTP")
159+ }
160+ d := diagnostics.Classify(diagnostics.RetainHTTPAttempt(err, witness.Attempted()), diagnostics.CommandContext{ProviderMode: "live-http", TransportFailure: true})
161+ if d.Code != diagnostics.CodeOperationCancelled || !d.HTTPAttempted {
162+ t.Fatalf("prior successful HTTP is not a transport failure: class=%s attempted=%t", d.Code, d.HTTPAttempted)
163+ }
164+}
@@ -1264,7 +1264,11 @@ func (c *HTTPClient) listWikiEntries(ctx context.Context, owner, repo, dir strin
1264 resp, err := c.do(ctx, http.MethodGet, endpoint, nil, nil, requestOptions{metricAttempt: observation})1264 resp, err := c.do(ctx, http.MethodGet, endpoint, nil, nil, requestOptions{metricAttempt: observation})
1265 if err != nil {1265 if err != nil {
1266 observation.finish(0, err)1266 observation.finish(0, err)
1267- return nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: 1, Cause: err}1267+ attempts := 0
1268+ if observation.httpEntered.Load() {
1269+ attempts = 1
1270+ }
1271+ return nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: attempts, Cause: err}
1268 }1272 }
1269 1273 
1270 body, readErr := c.readBounded(resp, endpoint)1274 body, readErr := c.readBounded(resp, endpoint)
@@ -2284,12 +2288,13 @@ func (c *HTTPClient) bytesWithOptions(ctx context.Context, method, endpoint stri
2284 }2288 }
2285 var lastRetryAfter time.Duration2289 var lastRetryAfter time.Duration
2286 var rawRetryAfter string2290 var rawRetryAfter string
2291+ httpAttempts := 0
2287 for attempt := 1; attempt <= attempts; attempt++ {2292 for attempt := 1; attempt <= attempts; attempt++ {
2288 if err := ctx.Err(); err != nil {2293 if err := ctx.Err(); err != nil {
2289- return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: attempt - 1, Cause: err}2294+ return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: httpAttempts, Cause: err}
2290 }2295 }
2291 if err := c.waitForRateLimit(ctx, method, endpoint, attempt); err != nil {2296 if err := c.waitForRateLimit(ctx, method, endpoint, attempt); err != nil {
2292- return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: attempt - 1, Cause: err}2297+ return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: httpAttempts, Cause: err}
2293 }2298 }
2294 observation := newProviderMetricAttempt(ctx)2299 observation := newProviderMetricAttempt(ctx)
2295 if attempt <= observability.MaxExecutionAttempt {2300 if attempt <= observability.MaxExecutionAttempt {
@@ -2297,16 +2302,19 @@ func (c *HTTPClient) bytesWithOptions(ctx context.Context, method, endpoint stri
2297 }2302 }
2298 opts.metricAttempt = observation2303 opts.metricAttempt = observation
2299 resp, err := c.do(ctx, method, endpoint, values, bytes.NewReader(requestBody), opts)2304 resp, err := c.do(ctx, method, endpoint, values, bytes.NewReader(requestBody), opts)
2305+ if observation.httpEntered.Load() {
2306+ httpAttempts++
2307+ }
2300 if err != nil {2308 if err != nil {
2301 if ctx.Err() != nil {2309 if ctx.Err() != nil {
2302 observation.finish(0, ctx.Err())2310 observation.finish(0, ctx.Err())
2303- return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: attempt, Cause: ctx.Err()}2311+ return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: httpAttempts, Cause: ctx.Err()}
2304 }2312 }
2305 observation.finish(0, err)2313 observation.finish(0, err)
2306 if isRetryableTransport(err) && attempt < attempts {2314 if isRetryableTransport(err) && attempt < attempts {
2307 continue2315 continue
2308 }2316 }
2309- return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: attempt, Cause: err}2317+ return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: httpAttempts, Cause: err}
2310 }2318 }
2311 if resp.StatusCode == http.StatusRequestEntityTooLarge {2319 if resp.StatusCode == http.StatusRequestEntityTooLarge {
2312 resp.Body.Close()2320 resp.Body.Close()
@@ -2340,7 +2348,7 @@ func (c *HTTPClient) bytesWithOptions(ctx context.Context, method, endpoint stri
2340 resumeAt := time.Now().Add(lastRetryAfter)2348 resumeAt := time.Now().Add(lastRetryAfter)
2341 emitRateLimitEvent(ctx, RateLimitEvent{Type: RateLimitEventRetryAfterWaitStarted, Method: method, Endpoint: endpoint, Attempt: attempt, Wait: lastRetryAfter, ResumeAt: resumeAt, RawRetryAfter: rawRetryAfter})2349 emitRateLimitEvent(ctx, RateLimitEvent{Type: RateLimitEventRetryAfterWaitStarted, Method: method, Endpoint: endpoint, Attempt: attempt, Wait: lastRetryAfter, ResumeAt: resumeAt, RawRetryAfter: rawRetryAfter})
2342 if err := sleepContext(ctx, lastRetryAfter); err != nil {2350 if err := sleepContext(ctx, lastRetryAfter); err != nil {
2343- return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: attempt, Cause: err}2351+ return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: httpAttempts, Cause: err}
2344 }2352 }
2345 emitRateLimitEvent(ctx, RateLimitEvent{Type: RateLimitEventRetryAfterWaitCompleted, Method: method, Endpoint: endpoint, Attempt: attempt, Wait: lastRetryAfter, ResumeAt: resumeAt, RawRetryAfter: rawRetryAfter})2353 emitRateLimitEvent(ctx, RateLimitEvent{Type: RateLimitEventRetryAfterWaitCompleted, Method: method, Endpoint: endpoint, Attempt: attempt, Wait: lastRetryAfter, ResumeAt: resumeAt, RawRetryAfter: rawRetryAfter})
2346 continue2354 continue
@@ -2352,7 +2360,7 @@ func (c *HTTPClient) bytesWithOptions(ctx context.Context, method, endpoint stri
2352 if attempt < attempts {2360 if attempt < attempts {
2353 continue2361 continue
2354 }2362 }
2355- return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Status: resp.StatusCode, Attempts: attempt}2363+ return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Status: resp.StatusCode, Attempts: httpAttempts}
2356 case strings.Contains(endpoint, ".wiki/") && isWikiEmptyResponse(resp.StatusCode, body):2364 case strings.Contains(endpoint, ".wiki/") && isWikiEmptyResponse(resp.StatusCode, body):
2357 owner, repo := parseWikiEndpointOwnerRepo(endpoint)2365 owner, repo := parseWikiEndpointOwnerRepo(endpoint)
2358 return nil, nil, ErrEmptyWiki{Owner: owner, Repo: repo}2366 return nil, nil, ErrEmptyWiki{Owner: owner, Repo: repo}
@@ -2365,7 +2373,7 @@ func (c *HTTPClient) bytesWithOptions(ctx context.Context, method, endpoint stri
2365 return nil, nil, c.statusError(resp.StatusCode, endpoint, body, opts)2373 return nil, nil, c.statusError(resp.StatusCode, endpoint, body, opts)
2366 }2374 }
2367 }2375 }
2368- return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: attempts}2376+ return nil, nil, ErrNetworkUnavailable{Endpoint: endpoint, Attempts: httpAttempts}
2369}2377}
2370 2378 
2371func effectiveRateLimitRetryAfter(raw string, now time.Time, attempt int) time.Duration {2379func effectiveRateLimitRetryAfter(raw string, now time.Time, attempt int) time.Duration {
@@ -2457,7 +2465,40 @@ func (c *HTTPClient) do(ctx context.Context, method, endpoint string, values url
2457 opts.beforeAttempt()2465 opts.beforeAttempt()
2458 }2466 }
2459 opts.metricAttempt.start()2467 opts.metricAttempt.start()
2460- return c.client.Do(req)2468+ // Copy only the client configuration; reuse the original transport, jar,
2469+ // redirect policy and timeout. Never mutate the shared client's transport.
2470+ client := *c.client
2471+ client.Transport = httpAttemptTransport{inner: c.client.Transport, attempt: opts.metricAttempt}
2472+ return client.Do(req)
2473+}
2474+ 
2475+type httpAttemptTransport struct {
2476+ inner http.RoundTripper
2477+ attempt *providerMetricAttempt
2478+}
2479+ 
2480+// Preserve http.Client's legacy timeout cancellation hook for transports that
2481+// implement it. Modern transports continue to use the request context.
2482+func (t httpAttemptTransport) CancelRequest(req *http.Request) {
2483+ inner := t.inner
2484+ if inner == nil {
2485+ inner = http.DefaultTransport
2486+ }
2487+ if canceller, ok := inner.(interface{ CancelRequest(*http.Request) }); ok {
2488+ canceller.CancelRequest(req)
2489+ }
2490+}
2491+ 
2492+func (t httpAttemptTransport) RoundTrip(req *http.Request) (*http.Response, error) {
2493+ inner := t.inner
2494+ if inner == nil {
2495+ inner = http.DefaultTransport
2496+ }
2497+ if t.attempt != nil {
2498+ t.attempt.httpEntered.Store(true)
2499+ }
2500+ diagnostics.RecordHTTPAttempt(req.Context())
2501+ return inner.RoundTrip(req)
2461}2502}
2462 2503 
2463func (c *HTTPClient) v4BaseURL() *url.URL {2504func (c *HTTPClient) v4BaseURL() *url.URL {
@@ -5,6 +5,7 @@ import (
5 "errors"5 "errors"
6 "net"6 "net"
7 "net/http"7 "net/http"
8+ "sync/atomic"
8 "time"9 "time"
9 10 
10 "gitcode-mcp/internal/observability"11 "gitcode-mcp/internal/observability"
@@ -14,11 +15,12 @@ import (
14// bounded response transfer. It is not a TCP packet count or durable receipt.15// bounded response transfer. It is not a TCP packet count or durable receipt.
15// No endpoint, header, coordinate, body, or error string crosses this boundary.16// No endpoint, header, coordinate, body, or error string crosses this boundary.
16type providerMetricAttempt struct {17type providerMetricAttempt struct {
17- ctx context.Context18+ ctx context.Context
18- ordinal uint3219+ ordinal uint32
19- metrics *observability.Metrics20+ metrics *observability.Metrics
20- started time.Time21+ started time.Time
21- finished bool22+ finished bool
23+ httpEntered atomic.Bool
22}24}
23 25 
24func newProviderMetricAttempt(ctx context.Context) *providerMetricAttempt {26func newProviderMetricAttempt(ctx context.Context) *providerMetricAttempt {
@@ -11,6 +11,7 @@ import (
11 11 
12 "gitcode-mcp/internal/cache"12 "gitcode-mcp/internal/cache"
13 "gitcode-mcp/internal/config"13 "gitcode-mcp/internal/config"
14+ "gitcode-mcp/internal/diagnostics"
14 "gitcode-mcp/internal/gitcode"15 "gitcode-mcp/internal/gitcode"
15 "gitcode-mcp/internal/service"16 "gitcode-mcp/internal/service"
16 "gitcode-mcp/internal/servicectl"17 "gitcode-mcp/internal/servicectl"
@@ -32,10 +33,20 @@ type repoStatusResult struct {
32}33}
33 34 
34type lifecycleDiagnostic struct {35type lifecycleDiagnostic struct {
35- Code string `json:"code"`36+ Code string `json:"code"`
36- ErrorClass string `json:"error_class,omitempty"`37+ ErrorClass string `json:"error_class,omitempty"`
37- Message string `json:"message"`38+ Message string `json:"message"`
38- Remediation string `json:"remediation,omitempty"`39+ Remediation string `json:"remediation,omitempty"`
40+ HTTPAttempted *bool `json:"http_attempted,omitempty"`
41+}
42+ 
43+func syncLifecycleDiagnostic(err error, code, message string) lifecycleDiagnostic {
44+ d := lifecycleDiagnostic{Code: code, Message: message}
45+ if classified, ok := mcpDiagnostic(err); ok {
46+ d.ErrorClass = string(classified.Code)
47+ d.HTTPAttempted = &classified.HTTPAttempted
48+ }
49+ return d
39}50}
40 51 
41func (s *Server) callRepoStatus(ctx context.Context, id *json.RawMessage, args json.RawMessage) {52func (s *Server) callRepoStatus(ctx context.Context, id *json.RawMessage, args json.RawMessage) {
@@ -212,6 +223,7 @@ func (s *Server) callSyncLive(ctx context.Context, id *json.RawMessage, args jso
212 s.writeToolResult(id, toolCallResult{Content: []toolContentItem{{Type: "text", Text: text}}, StructuredContent: result})223 s.writeToolResult(id, toolCallResult{Content: []toolContentItem{{Type: "text", Text: text}}, StructuredContent: result})
213 return224 return
214 }225 }
226+ ctx, _ = diagnostics.WithHTTPAttemptWitness(ctx)
215 if strings.TrimSpace(a.RemoteAlias) != "" {227 if strings.TrimSpace(a.RemoteAlias) != "" {
216 if commentSelection.PR && syncLiveRemoteAliasSurface(a.RemoteAlias) == "pull_request" {228 if commentSelection.PR && syncLiveRemoteAliasSurface(a.RemoteAlias) == "pull_request" {
217 part, err := s.svc.BulkSyncPRComments(ctx, service.BulkSyncRequest{RepoID: a.RepoID, RemoteAlias: strings.TrimSpace(a.RemoteAlias), IdempotencyKey: strings.TrimSpace(a.IdempotencyKey)})229 part, err := s.svc.BulkSyncPRComments(ctx, service.BulkSyncRequest{RepoID: a.RepoID, RemoteAlias: strings.TrimSpace(a.RemoteAlias), IdempotencyKey: strings.TrimSpace(a.IdempotencyKey)})
@@ -224,9 +236,9 @@ func (s *Server) callSyncLive(ctx context.Context, id *json.RawMessage, args jso
224 return236 return
225 }237 }
226 if partial, ok := extractLifecyclePartial(err); ok && partial.Diagnostic != "" {238 if partial, ok := extractLifecyclePartial(err); ok && partial.Diagnostic != "" {
227- result.Diagnostics = append(result.Diagnostics, lifecycleDiagnostic{Code: string(partial.Diagnostic), Message: "pr_comments sync returned a diagnostic"})239+ result.Diagnostics = append(result.Diagnostics, syncLifecycleDiagnostic(err, string(partial.Diagnostic), "pr_comments sync returned a diagnostic"))
228 } else {240 } else {
229- result.Diagnostics = append(result.Diagnostics, lifecycleDiagnostic{Code: "partial_sync", Message: err.Error()})241+ result.Diagnostics = append(result.Diagnostics, syncLifecycleDiagnostic(err, "partial_sync", "pr_comments sync returned a partial failure"))
230 }242 }
231 }243 }
232 text := fmt.Sprintf("fresh_count=%d collections=pr_comments", result.FreshCount)244 text := fmt.Sprintf("fresh_count=%d collections=pr_comments", result.FreshCount)
@@ -274,7 +286,7 @@ func (s *Server) callSyncLive(ctx context.Context, id *json.RawMessage, args jso
274 }286 }
275 syncErr = mergeLifecycleSyncError(syncErr, &result, err)287 syncErr = mergeLifecycleSyncError(syncErr, &result, err)
276 if partial, ok := extractLifecyclePartial(err); ok && partial.Diagnostic != "" {288 if partial, ok := extractLifecyclePartial(err); ok && partial.Diagnostic != "" {
277- result.Diagnostics = append(result.Diagnostics, lifecycleDiagnostic{Code: string(partial.Diagnostic), Message: collection + " sync returned a diagnostic"})289+ result.Diagnostics = append(result.Diagnostics, syncLifecycleDiagnostic(err, string(partial.Diagnostic), collection+" sync returned a diagnostic"))
278 }290 }
279 }291 }
280 }292 }
@@ -304,9 +316,9 @@ func (s *Server) callSyncLive(ctx context.Context, id *json.RawMessage, args jso
304 }316 }
305 if syncErr != nil {317 if syncErr != nil {
306 if partial, ok := extractLifecyclePartial(syncErr); ok && partial.Diagnostic != "" {318 if partial, ok := extractLifecyclePartial(syncErr); ok && partial.Diagnostic != "" {
307- result.Diagnostics = append(result.Diagnostics, lifecycleDiagnostic{Code: string(partial.Diagnostic), Message: "sync_live completed with a diagnostic"})319+ result.Diagnostics = append(result.Diagnostics, syncLifecycleDiagnostic(syncErr, string(partial.Diagnostic), "sync_live completed with a diagnostic"))
308 } else {320 } else {
309- result.Diagnostics = append(result.Diagnostics, lifecycleDiagnostic{Code: "partial_sync", Message: syncErr.Error()})321+ result.Diagnostics = append(result.Diagnostics, syncLifecycleDiagnostic(syncErr, "partial_sync", "sync_live completed with a partial failure"))
310 }322 }
311 }323 }
312 text := fmt.Sprintf("fresh_count=%d collections=%s", result.FreshCount, strings.Join(result.Collections, ","))324 text := fmt.Sprintf("fresh_count=%d collections=%s", result.FreshCount, strings.Join(result.Collections, ","))
@@ -370,7 +382,9 @@ func mergeLifecycleSyncError(existing error, result *syncLiveResult, err error)
370 if result == nil {382 if result == nil {
371 return existing383 return existing
372 }384 }
373- return &service.PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount}385+ partial := &service.PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount}
386+ partial.PreserveHTTPAttemptEvidence(existing, err)
387+ return partial
374}388}
375 389 
376func extractLifecyclePartial(err error) (*service.PartialSyncError, bool) {390func extractLifecyclePartial(err error) (*service.PartialSyncError, bool) {
@@ -296,6 +296,7 @@ type errorData struct {
296 Code string `json:"code"`296 Code string `json:"code"`
297 Message string `json:"message"`297 Message string `json:"message"`
298 FailureClass string `json:"failure_class,omitempty"`298 FailureClass string `json:"failure_class,omitempty"`
299+ HTTPAttempted *bool `json:"http_attempted,omitempty"`
299 Operation string `json:"operation,omitempty"`300 Operation string `json:"operation,omitempty"`
300 RepoID string `json:"repo_id,omitempty"`301 RepoID string `json:"repo_id,omitempty"`
301 StartedAt string `json:"started_at,omitempty"`302 StartedAt string `json:"started_at,omitempty"`
@@ -2521,6 +2522,7 @@ func mcpDiagnostic(err error) (diagnostics.Diagnostic, bool) {
2521 return diagnostics.Classify(err, ctx), true2522 return diagnostics.Classify(err, ctx), true
2522 }2523 }
2523 if errors.As(err, &syncErr) {2524 if errors.As(err, &syncErr) {
2525+ ctx.TransportFailure = syncErr.Mode == "network_timeout"
2524 ctx.HTTPAttempted = syncErr.Mode == "live_auth_failure" || syncErr.Mode == "network_timeout" || syncErr.Mode == "rate_limited" || syncErr.Mode == "partial_response" || syncErr.Mode == "live_graph_invalid" || syncErr.Mode == "remote_identity_mismatch" || syncErr.Mode == "payload_too_large" || syncErr.Mode == "remote_not_found" || syncErr.Mode == "conflict" || syncErr.Mode == "remote_collision"2526 ctx.HTTPAttempted = syncErr.Mode == "live_auth_failure" || syncErr.Mode == "network_timeout" || syncErr.Mode == "rate_limited" || syncErr.Mode == "partial_response" || syncErr.Mode == "live_graph_invalid" || syncErr.Mode == "remote_identity_mismatch" || syncErr.Mode == "payload_too_large" || syncErr.Mode == "remote_not_found" || syncErr.Mode == "conflict" || syncErr.Mode == "remote_collision"
2525 ctx.FailureSource = syncErr.PayloadSource2527 ctx.FailureSource = syncErr.PayloadSource
2526 ctx.LocalPayloadTooLarge = syncErr.Mode == "payload_too_large" && syncErr.PayloadSource == "local_body_limit"2528 ctx.LocalPayloadTooLarge = syncErr.Mode == "payload_too_large" && syncErr.PayloadSource == "local_body_limit"
@@ -2599,6 +2601,25 @@ func mcpDiagnostic(err error) (diagnostics.Diagnostic, bool) {
2599 ctx.HTTPStatus = forbiddenErr.Status2601 ctx.HTTPStatus = forbiddenErr.Status
2600 return diagnostics.Classify(err, ctx), true2602 return diagnostics.Classify(err, ctx), true
2601 }2603 }
2604+ var partialSync *service.PartialSyncError
2605+ if errors.As(err, &partialSync) {
2606+ _, attemptKnown := diagnostics.HTTPAttemptEvidence(err)
2607+ ctx.TransportFailure = !attemptKnown && (partialSync.Diagnostic == service.SyncDiagnosticCancelled || partialSync.Diagnostic == service.SyncDiagnosticTimeout)
2608+ ctx.HTTPAttempted = ctx.TransportFailure
2609+ for _, failure := range partialSync.Errors {
2610+ switch failure.FailureClass {
2611+ case "network_timeout":
2612+ ctx.HTTPAttempted, ctx.TransportFailure = true, true
2613+ case "schema_decode", "partial_response", "malformed_response", "unexpected_content_type":
2614+ ctx.HTTPAttempted, ctx.SchemaDecodeFailure = true, true
2615+ ctx.FailureSource = failure.FailureClass
2616+ }
2617+ }
2618+ return diagnostics.Classify(err, ctx), true
2619+ }
2620+ if _, known := diagnostics.HTTPAttemptEvidence(err); known {
2621+ return diagnostics.Classify(err, ctx), true
2622+ }
2602 return diagnostics.Diagnostic{}, false2623 return diagnostics.Diagnostic{}, false
2603}2624}
2604 2625 
@@ -2623,6 +2644,7 @@ func (s *Server) writeOperationalError(id *json.RawMessage, err error, ctx domai
2623 diagnostic, hasDiagnostic := mcpDiagnostic(err)2644 diagnostic, hasDiagnostic := mcpDiagnostic(err)
2624 if hasDiagnostic {2645 if hasDiagnostic {
2625 data.FailureClass = string(diagnostic.Code)2646 data.FailureClass = string(diagnostic.Code)
2647+ data.HTTPAttempted = &diagnostic.HTTPAttempted
2626 if data.Code == "internal_error" {2648 if data.Code == "internal_error" {
2627 data.Code = string(diagnostic.Code)2649 data.Code = string(diagnostic.Code)
2628 data.Message = diagnostic.Message2650 data.Message = diagnostic.Message
@@ -0,0 +1,139 @@
1+package mcp
2+ 
3+import (
4+ "bytes"
5+ "context"
6+ "encoding/json"
7+ "io"
8+ "net/http"
9+ "strings"
10+ "testing"
11+ "time"
12+ 
13+ "gitcode-mcp/internal/cache"
14+ "gitcode-mcp/internal/diagnostics"
15+ "gitcode-mcp/internal/gitcode"
16+ "gitcode-mcp/internal/service"
17+)
18+ 
19+type syncAttemptRoundTripper func(*http.Request) (*http.Response, error)
20+ 
21+func (f syncAttemptRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) }
22+ 
23+func TestMCPSyncLiveHTTPAttemptParity(t *testing.T) {
24+ for _, tc := range []struct {
25+ name string
26+ before, expired, empty, parentCancel, wikiOnly bool
27+ wantClass string
28+ wantCalls int
29+ }{
30+ {name: "before_cancel", before: true, wantClass: "operation_cancelled"},
31+ {name: "before_deadline", before: true, expired: true, wantClass: "operation_timeout"},
32+ {name: "inside_cancel", wantClass: "live_transport_failure", wantCalls: 1},
33+ {name: "inside_deadline", expired: true, wantClass: "live_transport_failure", wantCalls: 1},
34+ {name: "after_empty_list", empty: true, wantClass: "operation_cancelled", wantCalls: 1},
35+ {name: "issues_parent_cancel", parentCancel: true, wantClass: "live_transport_failure", wantCalls: 1},
36+ {name: "wiki_parent_cancel", parentCancel: true, wikiOnly: true, wantClass: "live_transport_failure", wantCalls: 1},
37+ } {
38+ t.Run(tc.name, func(t *testing.T) {
39+ store, err := cache.NewInMemorySQLiteStore(context.Background())
40+ if err != nil {
41+ t.Fatal("cache setup failed")
42+ }
43+ defer store.Close()
44+ if err := store.AddRepository(context.Background(), cache.RepositoryBinding{RepoID: "fixture-a", Owner: "example-owner", Name: "example-repo", APIBaseURL: "https://provider.invalid", Scopes: []cache.RepositoryScope{cache.RepositoryScopeIssues, cache.RepositoryScopeWiki}}); err != nil {
45+ t.Fatal("binding setup failed")
46+ }
47+ svc, err := service.NewWithMode(store, gitcode.ProviderModeLive, "fixture-credential", service.ServiceConfig{BaseURL: "https://provider.invalid"})
48+ if err != nil {
49+ t.Fatal("provider setup failed")
50+ }
51+ ctx, cancel := context.WithCancel(context.Background())
52+ defer cancel()
53+ if tc.before {
54+ cancel()
55+ if tc.expired {
56+ ctx, cancel = context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
57+ defer cancel()
58+ }
59+ }
60+ calls := 0
61+ previous := http.DefaultTransport
62+ t.Cleanup(func() { http.DefaultTransport = previous })
63+ http.DefaultTransport = syncAttemptRoundTripper(func(req *http.Request) (*http.Response, error) {
64+ calls++
65+ if tc.parentCancel {
66+ cancel()
67+ }
68+ if tc.empty {
69+ cancel()
70+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(`[]`)), Request: req}, nil
71+ }
72+ if tc.expired {
73+ return nil, context.DeadlineExceeded
74+ }
75+ return nil, context.Canceled
76+ })
77+ var out bytes.Buffer
78+ srv := NewWithToolAccess(nil, &out, io.Discard, svc, nil, ToolAccessWrite)
79+ id := json.RawMessage(`"attempt-parity"`)
80+ args, _ := json.Marshal(map[string]any{"repo_id": "fixture-a", "issues": !tc.wikiOnly, "wiki": tc.empty || tc.wikiOnly, "max_pages": 1})
81+ srv.callSyncLive(ctx, &id, args)
82+ var envelope struct {
83+ Result struct {
84+ StructuredContent syncLiveResult `json:"structuredContent"`
85+ } `json:"result"`
86+ }
87+ if err := json.Unmarshal(out.Bytes(), &envelope); err != nil {
88+ t.Fatal("response decode failed")
89+ }
90+ ds := envelope.Result.StructuredContent.Diagnostics
91+ if calls != tc.wantCalls || len(ds) == 0 {
92+ t.Fatal("missing diagnostic or unexpected provider traffic")
93+ }
94+ last := ds[len(ds)-1]
95+ if last.HTTPAttempted == nil || *last.HTTPAttempted != (calls != 0) || last.ErrorClass != tc.wantClass {
96+ t.Fatalf("MCP evidence mismatch: class=%s attempted_present=%t calls=%d", last.ErrorClass, last.HTTPAttempted != nil, calls)
97+ }
98+ })
99+ }
100+}
101+ 
102+func TestMCPSyncLaterPreDispatchInterruptionKeepsPriorHTTP(t *testing.T) {
103+ for _, cause := range []error{context.Canceled, context.DeadlineExceeded} {
104+ for _, entered := range []int{0, 1} {
105+ network := gitcode.ErrNetworkUnavailable{Attempts: entered, Cause: cause}
106+ failure := service.ErrSyncFailure{Mode: "network_timeout", Cause: network}
107+ partial := &service.PartialSyncError{Errors: []service.ResourceError{{Err: failure, FailureClass: "network_timeout"}}, SuccessCount: 1, FailureCount: 1}
108+ partial.Errors = append(partial.Errors, service.ResourceError{Err: diagnostics.RetainPreDispatchInterruption(cause)})
109+ partial.FailureCount++
110+ partial.PreserveHTTPAttemptEvidence(diagnostics.RetainHTTPAttempt(failure, true))
111+ d, ok := mcpDiagnostic(partial)
112+ want := diagnostics.CodeOperationCancelled
113+ if cause == context.DeadlineExceeded {
114+ want = diagnostics.CodeOperationTimeout
115+ }
116+ if entered != 0 {
117+ want = diagnostics.CodeLiveTransportFailure
118+ }
119+ if !ok || d.Code != want || !d.HTTPAttempted {
120+ t.Fatalf("partial interruption mismatch: entered=%d class=%s attempted=%t", entered, d.Code, d.HTTPAttempted)
121+ }
122+ }
123+ }
124+}
125+ 
126+func TestMCPOperationalErrorIncludesKnownNoHTTPAttempt(t *testing.T) {
127+ var out bytes.Buffer
128+ srv := &Server{writer: &out, stderr: io.Discard}
129+ id := json.RawMessage(`"attempt-error"`)
130+ srv.writeOperationalError(&id, gitcode.ErrNetworkUnavailable{Attempts: 0, Cause: context.Canceled}, domainErrorContext{Operation: "sync_live"})
131+ var resp response
132+ if err := json.Unmarshal(out.Bytes(), &resp); err != nil || resp.Error == nil || resp.Error.Data == nil {
133+ t.Fatal("missing operational error")
134+ }
135+ d := resp.Error.Data
136+ if d.FailureClass != "operation_cancelled" || d.HTTPAttempted == nil || *d.HTTPAttempted {
137+ t.Fatal("operational envelope contradicted zero transport entry")
138+ }
139+}
@@ -21,6 +21,7 @@ import (
21 "gitcode-mcp/internal/buildinfo"21 "gitcode-mcp/internal/buildinfo"
22 "gitcode-mcp/internal/cache"22 "gitcode-mcp/internal/cache"
23 "gitcode-mcp/internal/config"23 "gitcode-mcp/internal/config"
24+ "gitcode-mcp/internal/diagnostics"
24 "gitcode-mcp/internal/feedback"25 "gitcode-mcp/internal/feedback"
25 "gitcode-mcp/internal/gitcode"26 "gitcode-mcp/internal/gitcode"
26 "gitcode-mcp/internal/index"27 "gitcode-mcp/internal/index"
@@ -1578,9 +1579,11 @@ func (s *Service) Index(ctx context.Context, req OperationRequest) (OperationRes
1578 return OperationResult{Command: "index", Status: "ok", ProcessedCount: processed, Evidence: "snapshot_id=" + stored.ID, GeneratedAt: s.now().UTC()}, nil1579 return OperationResult{Command: "index", Status: "ok", ProcessedCount: processed, Evidence: "snapshot_id=" + stored.ID, GeneratedAt: s.now().UTC()}, nil
1579}1580}
1580 1581 
1581-func (s *Service) SyncToCache(ctx context.Context, req SyncRequest) (SyncResult, error) {1582+func (s *Service) SyncToCache(ctx context.Context, req SyncRequest) (_ SyncResult, retErr error) {
1583+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
1584+ defer func() { retErr = retainAttempt(retErr) }()
1582 if err := ctx.Err(); err != nil {1585 if err := ctx.Err(); err != nil {
1583- return SyncResult{}, err1586+ return SyncResult{}, diagnostics.RetainPreDispatchInterruption(err)
1584 }1587 }
1585 repoID, err := s.requireRepo(ctx, req.RepoID, "sync")1588 repoID, err := s.requireRepo(ctx, req.RepoID, "sync")
1586 if err != nil {1589 if err != nil {
@@ -1691,7 +1694,9 @@ func (s *Service) SyncToCache(ctx context.Context, req SyncRequest) (SyncResult,
1691// with structured per-resource failure details. Results always contains entries from1694// with structured per-resource failure details. Results always contains entries from
1692// successful SyncToCache calls; Failures contains entries from failed calls.1695// successful SyncToCache calls; Failures contains entries from failed calls.
1693// Callers should check PartialSyncError before using the result.1696// Callers should check PartialSyncError before using the result.
1694-func (s *Service) SyncResources(ctx context.Context, reqs []SyncRequest) (*SyncResourcesResult, error) {1697+func (s *Service) SyncResources(ctx context.Context, reqs []SyncRequest) (_ *SyncResourcesResult, retErr error) {
1698+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
1699+ defer func() { retErr = retainAttempt(retErr) }()
1695 result := &SyncResourcesResult{1700 result := &SyncResourcesResult{
1696 Results: make([]SyncResult, 0, len(reqs)),1701 Results: make([]SyncResult, 0, len(reqs)),
1697 Failures: make([]ResourceError, 0),1702 Failures: make([]ResourceError, 0),
@@ -1699,7 +1704,7 @@ func (s *Service) SyncResources(ctx context.Context, reqs []SyncRequest) (*SyncR
1699 var partial *PartialSyncError1704 var partial *PartialSyncError
1700 for i, req := range reqs {1705 for i, req := range reqs {
1701 if err := ctx.Err(); err != nil {1706 if err := ctx.Err(); err != nil {
1702- re := newResourceErrorWithMessage(req.StableID, "", err, fmt.Sprintf("sync resources: context cancelled at request %d", i))1707+ re := newResourceErrorWithMessage(req.StableID, "", diagnostics.RetainPreDispatchInterruption(err), fmt.Sprintf("sync resources: context cancelled at request %d", i))
1703 result.Failures = append(result.Failures, re)1708 result.Failures = append(result.Failures, re)
1704 continue1709 continue
1705 }1710 }
@@ -1734,7 +1739,9 @@ func (s *Service) SyncResources(ctx context.Context, reqs []SyncRequest) (*SyncR
1734 return result, nil1739 return result, nil
1735}1740}
1736 1741 
1737-func (s *Service) BulkSyncIssues(ctx context.Context, req BulkSyncRequest) (*SyncResourcesResult, error) {1742+func (s *Service) BulkSyncIssues(ctx context.Context, req BulkSyncRequest) (_ *SyncResourcesResult, retErr error) {
1743+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
1744+ defer func() { retErr = retainAttempt(retErr) }()
1738 if err := ctx.Err(); err != nil {1745 if err := ctx.Err(); err != nil {
1739 if req.Bounds != nil {1746 if req.Bounds != nil {
1740 diag := SyncDiagnosticCancelled1747 diag := SyncDiagnosticCancelled
@@ -1787,6 +1794,9 @@ func (s *Service) BulkSyncIssues(ctx context.Context, req BulkSyncRequest) (*Syn
1787 result.SuccessCount = len(result.Results)1794 result.SuccessCount = len(result.Results)
1788 result.FailureCount = len(result.Failures)1795 result.FailureCount = len(result.Failures)
1789 if summaryErr := s.attachIssueCommentQueueSummary(ctx, result, repoID, "parent_backfill"); summaryErr != nil {1796 if summaryErr := s.attachIssueCommentQueueSummary(ctx, result, repoID, "parent_backfill"); summaryErr != nil {
1797+ if result.FailureCount > 0 {
1798+ return result, partialSyncWithLocalCause(result, summaryErr)
1799+ }
1790 return result, summaryErr1800 return result, summaryErr
1791 }1801 }
1792 if result.FailureCount > 0 {1802 if result.FailureCount > 0 {
@@ -1870,7 +1880,7 @@ func (s *Service) bulkSyncIssuesBounded(ctx context.Context, req BulkSyncRequest
1870 result.Failures = append(result.Failures, re)1880 result.Failures = append(result.Failures, re)
1871 result.FailureCount = len(result.Failures)1881 result.FailureCount = len(result.Failures)
1872 if summaryErr := s.attachIssueCommentQueueSummary(ctx, result, req.RepoID, "parent_backfill"); summaryErr != nil {1882 if summaryErr := s.attachIssueCommentQueueSummary(ctx, result, req.RepoID, "parent_backfill"); summaryErr != nil {
1873- return result, summaryErr1883+ return result, partialSyncWithLocalCause(result, summaryErr)
1874 }1884 }
1875 s.recordSyncFrontierBestEffort(ctx, req.RepoID, "issue", result, observed)1885 s.recordSyncFrontierBestEffort(ctx, req.RepoID, "issue", result, observed)
1876 return result, &PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount, TotalRequested: totalRequested}1886 return result, &PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount, TotalRequested: totalRequested}
@@ -2036,7 +2046,9 @@ func (s *Service) prepareIssueCommentSync(ctx context.Context, graph cache.Sourc
2036 return item, issue.Comments == 0, item.Status != "complete", nil2046 return item, issue.Comments == 0, item.Status != "complete", nil
2037}2047}
2038 2048 
2039-func (s *Service) BulkSyncIssueComments(ctx context.Context, req BulkSyncRequest) (*SyncResourcesResult, error) {2049+func (s *Service) BulkSyncIssueComments(ctx context.Context, req BulkSyncRequest) (_ *SyncResourcesResult, retErr error) {
2050+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
2051+ defer func() { retErr = retainAttempt(retErr) }()
2040 if err := ctx.Err(); err != nil {2052 if err := ctx.Err(); err != nil {
2041 return nil, err2053 return nil, err
2042 }2054 }
@@ -2707,7 +2719,9 @@ func syncResultsDeferredCount(results []SyncResult) int {
2707 return total2719 return total
2708}2720}
2709 2721 
2710-func (s *Service) BulkSyncPullRequests(ctx context.Context, req BulkSyncRequest) (*SyncResourcesResult, error) {2722+func (s *Service) BulkSyncPullRequests(ctx context.Context, req BulkSyncRequest) (_ *SyncResourcesResult, retErr error) {
2723+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
2724+ defer func() { retErr = retainAttempt(retErr) }()
2711 if err := ctx.Err(); err != nil {2725 if err := ctx.Err(); err != nil {
2712 if req.Bounds != nil {2726 if req.Bounds != nil {
2713 diag := SyncDiagnosticCancelled2727 diag := SyncDiagnosticCancelled
@@ -3031,7 +3045,9 @@ func (s *Service) recordSyncFrontierBestEffort(ctx context.Context, repoID, remo
3031 _ = s.recordSyncFrontier(ctx, repoID, remoteType, result, high)3045 _ = s.recordSyncFrontier(ctx, repoID, remoteType, result, high)
3032}3046}
3033 3047 
3034-func (s *Service) BulkSyncPRComments(ctx context.Context, req BulkSyncRequest) (*SyncResourcesResult, error) {3048+func (s *Service) BulkSyncPRComments(ctx context.Context, req BulkSyncRequest) (_ *SyncResourcesResult, retErr error) {
3049+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
3050+ defer func() { retErr = retainAttempt(retErr) }()
3035 remoteAlias := strings.TrimSpace(req.RemoteAlias)3051 remoteAlias := strings.TrimSpace(req.RemoteAlias)
3036 targetedPRNumber := 03052 targetedPRNumber := 0
3037 if remoteAlias != "" {3053 if remoteAlias != "" {
@@ -3402,7 +3418,9 @@ func (s *Service) stagePRCommentPage(ctx context.Context, req BulkSyncRequest, p
3402 }3418 }
3403}3419}
3404 3420 
3405-func (s *Service) BulkSyncWiki(ctx context.Context, req BulkSyncRequest) (*SyncResourcesResult, error) {3421+func (s *Service) BulkSyncWiki(ctx context.Context, req BulkSyncRequest) (_ *SyncResourcesResult, retErr error) {
3422+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
3423+ defer func() { retErr = retainAttempt(retErr) }()
3406 if err := ctx.Err(); err != nil {3424 if err := ctx.Err(); err != nil {
3407 if req.Bounds != nil {3425 if req.Bounds != nil {
3408 diag := SyncDiagnosticCancelled3426 diag := SyncDiagnosticCancelled
@@ -3507,9 +3525,9 @@ func (s *Service) bulkSyncWikiBounded(ctx context.Context, req BulkSyncRequest,
3507 result.SuccessCount = len(result.Results)3525 result.SuccessCount = len(result.Results)
3508 result.FailureCount = len(result.Failures)3526 result.FailureCount = len(result.Failures)
3509 if result.SuccessCount == 0 && result.FailureCount == 0 {3527 if result.SuccessCount == 0 && result.FailureCount == 0 {
3510- return nil, &PartialSyncError{Errors: nil, SuccessCount: 0, FailureCount: 0, Diagnostic: diag, TotalRequested: totalRequested}3528+ return nil, &PartialSyncError{Errors: nil, SuccessCount: 0, FailureCount: 0, Diagnostic: diag, TotalRequested: totalRequested, retainedCause: err}
3511 }3529 }
3512- return result, &PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount, Diagnostic: diag, TotalRequested: totalRequested}3530+ return result, &PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount, Diagnostic: diag, TotalRequested: totalRequested, retainedCause: err}
3513 }3531 }
3514 result.SuccessCount = len(result.Results)3532 result.SuccessCount = len(result.Results)
3515 result.FailureCount = len(result.Failures)3533 result.FailureCount = len(result.Failures)
@@ -3693,7 +3711,9 @@ func (s *Service) skipWikiListItemByRevision(ctx context.Context, req SyncReques
3693 return SyncResult{IdempotencyKey: req.IdempotencyKey, Status: "succeeded", Counts: counts, SyncEventID: eventID, Freshness: string(FreshnessFresh), Record: sourceSummary(stored), GeneratedAt: completedAt, StartedAt: completedAt, CompletedAt: completedAt, ZeroDelta: true}, true, nil3711 return SyncResult{IdempotencyKey: req.IdempotencyKey, Status: "succeeded", Counts: counts, SyncEventID: eventID, Freshness: string(FreshnessFresh), Record: sourceSummary(stored), GeneratedAt: completedAt, StartedAt: completedAt, CompletedAt: completedAt, ZeroDelta: true}, true, nil
3694}3712}
3695 3713 
3696-func (s *Service) BulkSyncAll(ctx context.Context, req BulkSyncRequest) (*SyncResourcesResult, error) {3714+func (s *Service) BulkSyncAll(ctx context.Context, req BulkSyncRequest) (_ *SyncResourcesResult, retErr error) {
3715+ ctx, retainAttempt := beginSyncHTTPAttempt(ctx)
3716+ defer func() { retErr = retainAttempt(retErr) }()
3697 if err := ctx.Err(); err != nil {3717 if err := ctx.Err(); err != nil {
3698 return nil, err3718 return nil, err
3699 }3719 }
@@ -3746,12 +3766,14 @@ func (s *Service) BulkSyncAll(ctx context.Context, req BulkSyncRequest) (*SyncRe
3746 aggregated.FailureCount = len(aggregated.Failures)3766 aggregated.FailureCount = len(aggregated.Failures)
3747 3767 
3748 if issuesErr != nil || wikiErr != nil {3768 if issuesErr != nil || wikiErr != nil {
3749- return aggregated, &PartialSyncError{3769+ partial := &PartialSyncError{
3750 Errors: aggregated.Failures,3770 Errors: aggregated.Failures,
3751 SuccessCount: aggregated.SuccessCount,3771 SuccessCount: aggregated.SuccessCount,
3752 FailureCount: aggregated.FailureCount,3772 FailureCount: aggregated.FailureCount,
3753 Diagnostic: firstDiagnostic,3773 Diagnostic: firstDiagnostic,
3754 }3774 }
3775+ partial.PreserveHTTPAttemptEvidence(issuesErr, wikiErr)
3776+ return aggregated, partial
3755 }3777 }
3756 if aggregated.FailureCount > 0 {3778 if aggregated.FailureCount > 0 {
3757 return aggregated, &PartialSyncError{3779 return aggregated, &PartialSyncError{
@@ -0,0 +1,73 @@
1+package service
2+ 
3+import (
4+ "context"
5+ "errors"
6+ 
7+ "gitcode-mcp/internal/diagnostics"
8+)
9+ 
10+func beginSyncHTTPAttempt(ctx context.Context) (context.Context, func(error) error) {
11+ ctx, witness := diagnostics.WithHTTPAttemptWitness(ctx)
12+ return ctx, func(err error) error {
13+ if err == nil {
14+ return nil
15+ }
16+ if partial, ok := err.(*PartialSyncError); ok {
17+ partial.httpAttempted, partial.httpAttemptKnown = witness.Attempted(), true
18+ return partial
19+ }
20+ return diagnostics.RetainHTTPAttempt(err, witness.Attempted())
21+ }
22+}
23+ 
24+func (e PartialSyncError) HTTPAttemptEvidence() (bool, bool) {
25+ return e.httpAttempted, e.httpAttemptKnown
26+}
27+ 
28+func (e PartialSyncError) OperationInterruption() string {
29+ switch e.Diagnostic {
30+ case SyncDiagnosticCancelled:
31+ return "cancelled"
32+ case SyncDiagnosticTimeout:
33+ return "timeout"
34+ }
35+ return ""
36+}
37+ 
38+// PreserveHTTPAttemptEvidence keeps receipts and the first interruption when
39+// transport surfaces combine independent collection errors.
40+func (e *PartialSyncError) PreserveHTTPAttemptEvidence(errs ...error) {
41+ e.httpAttempted, e.httpAttemptKnown = diagnostics.HTTPAttemptEvidence(errors.Join(errs...))
42+ // Keep typed causes that are not represented in public failure counts.
43+ // Never retain this aggregate itself, directly or through a wrapper.
44+ var causes []error
45+ for _, err := range errs {
46+ if err != nil && !errors.Is(err, e) {
47+ causes = append(causes, err)
48+ }
49+ var partial *PartialSyncError
50+ if e.Diagnostic == "" && errors.As(err, &partial) {
51+ e.Diagnostic = partial.Diagnostic
52+ }
53+ if e.Diagnostic == "" {
54+ switch {
55+ case errors.Is(err, context.Canceled):
56+ e.Diagnostic = SyncDiagnosticCancelled
57+ case errors.Is(err, context.DeadlineExceeded):
58+ e.Diagnostic = SyncDiagnosticTimeout
59+ }
60+ }
61+ }
62+ e.retainedCause = errors.Join(causes...)
63+}
64+ 
65+func partialSyncWithLocalCause(result *SyncResourcesResult, cause error) *PartialSyncError {
66+ partial := &PartialSyncError{Errors: result.Failures, SuccessCount: result.SuccessCount, FailureCount: result.FailureCount}
67+ // A duplicate context cause is already retained by the provider failure.
68+ // Do not turn that known request boundary into an unknown extra branch.
69+ if cause != nil && !errors.Is(partial, cause) {
70+ partial.retainedCause = cause
71+ }
72+ return partial
73+}
@@ -0,0 +1,302 @@
1+package service
2+ 
3+import (
4+ "context"
5+ "encoding/json"
6+ "errors"
7+ "fmt"
8+ "io"
9+ "net/http"
10+ "strings"
11+ "testing"
12+ "time"
13+ 
14+ "gitcode-mcp/internal/cache"
15+ "gitcode-mcp/internal/diagnostics"
16+ "gitcode-mcp/internal/gitcode"
17+)
18+ 
19+type syncAttemptTransport func(*http.Request) (*http.Response, error)
20+ 
21+func (f syncAttemptTransport) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) }
22+ 
23+func TestSyncOwnersRetainPreProviderInterruption(t *testing.T) {
24+ store, err := cache.NewInMemorySQLiteStore(context.Background())
25+ if err != nil {
26+ t.Fatal("cache setup failed")
27+ }
28+ defer store.Close()
29+ svc := New(store)
30+ for _, expired := range []bool{false, true} {
31+ ctx, cancel := context.WithCancel(context.Background())
32+ cancel()
33+ want := diagnostics.CodeOperationCancelled
34+ if expired {
35+ ctx, cancel = context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
36+ defer cancel()
37+ want = diagnostics.CodeOperationTimeout
38+ }
39+ req := BulkSyncRequest{RepoID: "fixture-a", Bounds: &SyncBounds{MaxPages: 1}}
40+ for _, fn := range []func(context.Context, BulkSyncRequest) (*SyncResourcesResult, error){svc.BulkSyncIssues, svc.BulkSyncIssueComments, svc.BulkSyncWiki, svc.BulkSyncPullRequests, svc.BulkSyncPRComments, svc.BulkSyncAll} {
41+ _, err := fn(ctx, req)
42+ d := diagnostics.Classify(err, diagnostics.CommandContext{ProviderMode: "live-http", HTTPAttempted: true, TransportFailure: true})
43+ if err == nil || d.Code != want || d.HTTPAttempted {
44+ t.Fatal("sync owner inferred HTTP from pre-provider interruption")
45+ }
46+ }
47+ _, err := svc.SyncToCache(ctx, SyncRequest{RepoID: "fixture-a", RemoteAlias: "issue:1"})
48+ if !errors.Is(err, ctx.Err()) {
49+ t.Fatal("targeted sync lost cause")
50+ }
51+ _, err = svc.SyncResources(ctx, []SyncRequest{{RepoID: "fixture-a", RemoteAlias: "issue:1"}})
52+ if attempted, known := diagnostics.HTTPAttemptEvidence(err); !known || attempted {
53+ t.Fatal("multi-target sync lost known negative evidence")
54+ }
55+ }
56+}
57+ 
58+func TestSyncRetainsEmptyListHTTPBeforeLaterCancellation(t *testing.T) {
59+ ctx, cancel := context.WithCancel(context.Background())
60+ defer cancel()
61+ store, err := cache.NewInMemorySQLiteStore(ctx)
62+ if err != nil {
63+ t.Fatal("cache setup failed")
64+ }
65+ defer store.Close()
66+ if err := store.AddRepository(ctx, cache.RepositoryBinding{RepoID: "fixture-a", Owner: "example-owner", Name: "example-repo", APIBaseURL: "https://provider.invalid", Scopes: []cache.RepositoryScope{cache.RepositoryScopeIssues, cache.RepositoryScopeWiki}}); err != nil {
67+ t.Fatal("binding setup failed")
68+ }
69+ client, _ := gitcode.NewHTTPClient(gitcode.Config{BaseURL: "https://provider.invalid"})
70+ svc := NewWithClient(store, client)
71+ previous := http.DefaultTransport
72+ defer func() { http.DefaultTransport = previous }()
73+ calls := 0
74+ http.DefaultTransport = syncAttemptTransport(func(req *http.Request) (*http.Response, error) {
75+ calls++
76+ cancel()
77+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(`[]`)), Request: req}, nil
78+ })
79+ _, err = svc.BulkSyncAll(ctx, BulkSyncRequest{RepoID: "fixture-a", Bounds: &SyncBounds{MaxPages: 1}})
80+ if attempted, known := diagnostics.HTTPAttemptEvidence(err); err == nil || !known || !attempted || calls != 1 {
81+ t.Fatal("empty list HTTP lost after later cancellation")
82+ }
83+}
84+ 
85+func TestSyncPartialAggregatePreservesAttemptAndDiagnostic(t *testing.T) {
86+ _, retain := beginSyncHTTPAttempt(context.Background())
87+ negative := retain(&PartialSyncError{Diagnostic: SyncDiagnosticCancelled})
88+ ctx, witness := diagnostics.WithHTTPAttemptWitness(context.Background())
89+ diagnostics.RecordHTTPAttempt(ctx)
90+ _, retainPositive := beginSyncHTTPAttempt(ctx)
91+ positive := retainPositive(&PartialSyncError{SuccessCount: 1, Diagnostic: SyncDiagnosticTimeout})
92+ if !witness.Attempted() {
93+ t.Fatal("positive witness absent")
94+ }
95+ merged := &PartialSyncError{SuccessCount: 1}
96+ merged.PreserveHTTPAttemptEvidence(negative, positive)
97+ if attempted, known := merged.HTTPAttemptEvidence(); !known || !attempted || merged.Diagnostic != SyncDiagnosticCancelled {
98+ t.Fatal("aggregate lost first diagnostic or positive attempt")
99+ }
100+ if _, ok := positive.(*PartialSyncError); !ok {
101+ t.Fatal("partial error pointer contract changed")
102+ }
103+}
104+ 
105+func TestSyncPartialAggregateRetainsPrivateCauseWithoutInventedCounts(t *testing.T) {
106+ cause := gitcode.ErrNetworkUnavailable{Attempts: 1, Cause: context.Canceled}
107+ child := &PartialSyncError{Diagnostic: SyncDiagnosticCancelled, retainedCause: cause}
108+ child.httpAttempted, child.httpAttemptKnown = true, true
109+ merged := &PartialSyncError{}
110+ merged.PreserveHTTPAttemptEvidence(child)
111+ outer := &PartialSyncError{}
112+ outer.PreserveHTTPAttemptEvidence(merged)
113+ for _, partial := range []*PartialSyncError{child, merged, outer} {
114+ d := diagnostics.Classify(partial, diagnostics.CommandContext{ProviderMode: "live-http"})
115+ if d.Code != diagnostics.CodeLiveTransportFailure || !d.HTTPAttempted || !errors.Is(partial, context.Canceled) {
116+ t.Fatal("private typed cause was lost through aggregation")
117+ }
118+ data, err := json.Marshal(partial)
119+ if err != nil || string(data) != `{"errors":null,"success_count":0,"failure_count":0,"diagnostic":"sync_cancelled"}` {
120+ t.Fatal("private evidence changed public partial counts or JSON")
121+ }
122+ }
123+ // Self references must not create an error-tree cycle, even through wrappers.
124+ outer.PreserveHTTPAttemptEvidence(outer, fmt.Errorf("wrapped: %w", outer), child)
125+ if !errors.Is(outer, context.Canceled) || len(outer.Unwrap()) != 1 {
126+ t.Fatal("aggregate retained itself or lost the independent child")
127+ }
128+}
129+ 
130+func TestSyncCollectionsPriorHTTPThenLocalRateCancellation(t *testing.T) {
131+ ctx, cancel := context.WithCancel(context.Background())
132+ defer cancel()
133+ store, err := cache.NewInMemorySQLiteStore(ctx)
134+ if err != nil {
135+ t.Fatal("cache setup failed")
136+ }
137+ defer store.Close()
138+ if err := store.AddRepository(ctx, cache.RepositoryBinding{RepoID: "fixture-a", Owner: "example-owner", Name: "example-repo", APIBaseURL: "https://provider.invalid", Scopes: []cache.RepositoryScope{cache.RepositoryScopeIssues, cache.RepositoryScopeWiki}}); err != nil {
139+ t.Fatal("binding setup failed")
140+ }
141+ svc, err := NewWithMode(store, gitcode.ProviderModeLive, "fixture-credential", ServiceConfig{BaseURL: "https://provider.invalid", RateLimitRPS: 0.001, RateLimitBurst: 1})
142+ if err != nil {
143+ t.Fatal("provider setup failed")
144+ }
145+ calls := 0
146+ previous := http.DefaultTransport
147+ defer func() { http.DefaultTransport = previous }()
148+ http.DefaultTransport = syncAttemptTransport(func(req *http.Request) (*http.Response, error) {
149+ calls++
150+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(`[]`)), Request: req}, nil
151+ })
152+ ctx, _ = diagnostics.WithHTTPAttemptWitness(ctx)
153+ _, err = svc.BulkSyncIssues(ctx, BulkSyncRequest{RepoID: "fixture-a", Bounds: &SyncBounds{MaxPages: 1}})
154+ if err != nil || calls != 1 {
155+ t.Fatal("first collection must succeed with exactly one HTTP request")
156+ }
157+ ctx = gitcode.WithRateLimitObserver(ctx, func(event gitcode.RateLimitEvent) {
158+ if event.Type == gitcode.RateLimitEventThrottleWaitStarted {
159+ cancel()
160+ }
161+ })
162+ _, err = svc.BulkSyncPullRequests(ctx, BulkSyncRequest{RepoID: "fixture-a", Bounds: &SyncBounds{MaxPages: 1}})
163+ var network gitcode.ErrNetworkUnavailable
164+ if !errors.As(err, &network) || network.Attempts != 0 || calls != 1 || !errors.Is(err, context.Canceled) {
165+ t.Fatal("later collection must stop locally before another HTTP request")
166+ }
167+ d := diagnostics.Classify(err, diagnostics.CommandContext{ProviderMode: "live-http", TransportFailure: true})
168+ if d.Code != diagnostics.CodeOperationCancelled || !d.HTTPAttempted {
169+ t.Fatalf("shared owner evidence mismatch: class=%s attempted=%t", d.Code, d.HTTPAttempted)
170+ }
171+}
172+ 
173+func TestSyncResourcesPriorHTTPThenCancelledAndSkippedTargets(t *testing.T) {
174+ testSyncResourcesCancelledTail(t, true)
175+}
176+ 
177+func TestSyncResourcesBatchSuccessThenCancelledAndSkippedTargets(t *testing.T) {
178+ testSyncResourcesCancelledTail(t, false)
179+}
180+ 
181+func testSyncResourcesCancelledTail(t *testing.T, priorCollection bool) {
182+ t.Helper()
183+ ctx, cancel := context.WithCancel(context.Background())
184+ defer cancel()
185+ store, err := cache.NewInMemorySQLiteStore(ctx)
186+ if err != nil {
187+ t.Fatal("cache setup failed")
188+ }
189+ defer store.Close()
190+ if err := store.AddRepository(ctx, cache.RepositoryBinding{RepoID: "fixture-a", Owner: "example-owner", Name: "example-repo", APIBaseURL: "https://provider.invalid", Scopes: []cache.RepositoryScope{cache.RepositoryScopeIssues}}); err != nil {
191+ t.Fatal("binding setup failed")
192+ }
193+ burst := 2
194+ if priorCollection {
195+ burst = 1
196+ }
197+ svc, err := NewWithMode(store, gitcode.ProviderModeLive, "fixture-credential", ServiceConfig{BaseURL: "https://provider.invalid", RateLimitRPS: 0.001, RateLimitBurst: burst})
198+ if err != nil {
199+ t.Fatal("provider setup failed")
200+ }
201+ calls := 0
202+ previous := http.DefaultTransport
203+ defer func() { http.DefaultTransport = previous }()
204+ http.DefaultTransport = syncAttemptTransport(func(req *http.Request) (*http.Response, error) {
205+ calls++
206+ body := `[]`
207+ if !priorCollection && !strings.HasSuffix(req.URL.Path, "/comments") {
208+ body = `{"id":1,"number":1,"title":"fixture","body":"","state":"open","comments":0,"labels":[]}`
209+ }
210+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(body)), Request: req}, nil
211+ })
212+ ctx, _ = diagnostics.WithHTTPAttemptWitness(ctx)
213+ if priorCollection {
214+ if _, err := svc.BulkSyncIssues(ctx, BulkSyncRequest{RepoID: "fixture-a", Bounds: &SyncBounds{MaxPages: 1}}); err != nil || calls != 1 {
215+ t.Fatal("prior collection must succeed with one HTTP request")
216+ }
217+ }
218+ ctx = gitcode.WithRateLimitObserver(ctx, func(event gitcode.RateLimitEvent) {
219+ if event.Type == gitcode.RateLimitEventThrottleWaitStarted {
220+ cancel()
221+ }
222+ })
223+ reqs := []SyncRequest{{RepoID: "fixture-a", RemoteAlias: "issue:1"}, {RepoID: "fixture-a", RemoteAlias: "issue:2"}}
224+ wantSuccess, wantCalls := 0, 1
225+ if !priorCollection {
226+ reqs = append(reqs, SyncRequest{RepoID: "fixture-a", RemoteAlias: "issue:3"})
227+ wantSuccess, wantCalls = 1, 2
228+ }
229+ result, err := svc.SyncResources(ctx, reqs)
230+ if result == nil || result.SuccessCount != wantSuccess || result.FailureCount != 2 {
231+ t.Fatal("batch must retain prior success and both cancellation failures")
232+ }
233+ var network gitcode.ErrNetworkUnavailable
234+ if !errors.As(err, &network) || network.Attempts != 0 || calls != wantCalls || !errors.Is(err, context.Canceled) {
235+ t.Fatal("later target must stop locally and the final target must be skipped")
236+ }
237+ d := diagnostics.Classify(err, diagnostics.CommandContext{ProviderMode: "live-http", TransportFailure: true})
238+ if d.Code != diagnostics.CodeOperationCancelled || !d.HTTPAttempted {
239+ t.Fatalf("skipped-target evidence mismatch: class=%s attempted=%t", d.Code, d.HTTPAttempted)
240+ }
241+}
242+ 
243+func TestBoundedSyncParentCancellationRetainsActualFailure(t *testing.T) {
244+ for _, collection := range []string{"issues", "wiki", "pulls"} {
245+ for _, failed := range []bool{false, true} {
246+ name := collection + "/successful_then_local"
247+ if failed {
248+ name = collection + "/entered_failure"
249+ }
250+ t.Run(name, func(t *testing.T) {
251+ ctx, cancel := context.WithCancel(context.Background())
252+ defer cancel()
253+ ctx, _ = diagnostics.WithHTTPAttemptWitness(ctx)
254+ store, err := cache.NewInMemorySQLiteStore(ctx)
255+ if err != nil {
256+ t.Fatal("cache setup failed")
257+ }
258+ defer store.Close()
259+ if err := store.AddRepository(ctx, cache.RepositoryBinding{RepoID: "fixture-a", Owner: "example-owner", Name: "example-repo", APIBaseURL: "https://provider.invalid", Scopes: []cache.RepositoryScope{cache.RepositoryScopeIssues, cache.RepositoryScopeWiki}}); err != nil {
260+ t.Fatal("binding setup failed")
261+ }
262+ svc, err := NewWithMode(store, gitcode.ProviderModeLive, "fixture-credential", ServiceConfig{BaseURL: "https://provider.invalid"})
263+ if err != nil {
264+ t.Fatal("provider setup failed")
265+ }
266+ calls := 0
267+ previous := http.DefaultTransport
268+ defer func() { http.DefaultTransport = previous }()
269+ http.DefaultTransport = syncAttemptTransport(func(req *http.Request) (*http.Response, error) {
270+ calls++
271+ cancel()
272+ if failed {
273+ return nil, context.Canceled
274+ }
275+ return &http.Response{StatusCode: 200, Header: http.Header{}, ContentLength: -1, Body: io.NopCloser(strings.NewReader(`[]`)), Request: req}, nil
276+ })
277+ fn := svc.BulkSyncIssues
278+ switch collection {
279+ case "wiki":
280+ fn = svc.BulkSyncWiki
281+ case "pulls":
282+ fn = svc.BulkSyncPullRequests
283+ }
284+ _, err = fn(ctx, BulkSyncRequest{RepoID: "fixture-a", Bounds: &SyncBounds{MaxPages: 1}})
285+ if !failed && err == nil {
286+ _, err = svc.BulkSyncIssues(ctx, BulkSyncRequest{RepoID: "fixture-a", Bounds: &SyncBounds{MaxPages: 1}})
287+ }
288+ if err == nil || calls != 1 {
289+ t.Fatal("control must reach exactly one actual HTTP request")
290+ }
291+ d := diagnostics.Classify(err, diagnostics.CommandContext{ProviderMode: "live-http"})
292+ want := diagnostics.CodeOperationCancelled
293+ if failed {
294+ want = diagnostics.CodeLiveTransportFailure
295+ }
296+ if d.Code != want || !d.HTTPAttempted {
297+ t.Fatalf("parent cancellation evidence mismatch: class=%s attempted=%t", d.Code, d.HTTPAttempted)
298+ }
299+ })
300+ }
301+ }
302+}
@@ -668,11 +668,14 @@ func resourceStatusCode(err error) int {
668}668}
669 669 
670type PartialSyncError struct {670type PartialSyncError struct {
671- Errors []ResourceError `json:"errors"`671+ httpAttempted bool
672- SuccessCount int `json:"success_count"`672+ httpAttemptKnown bool
673- FailureCount int `json:"failure_count"`673+ retainedCause error
674- Diagnostic SyncDiagnostic `json:"diagnostic,omitempty"`674+ Errors []ResourceError `json:"errors"`
675- TotalRequested int `json:"total_requested,omitempty"`675+ SuccessCount int `json:"success_count"`
676+ FailureCount int `json:"failure_count"`
677+ Diagnostic SyncDiagnostic `json:"diagnostic,omitempty"`
678+ TotalRequested int `json:"total_requested,omitempty"`
676}679}
677 680 
678func (e PartialSyncError) Error() string {681func (e PartialSyncError) Error() string {
@@ -686,6 +689,9 @@ func (e PartialSyncError) Unwrap() []error {
686 out = append(out, resourceErr.Err)689 out = append(out, resourceErr.Err)
687 }690 }
688 }691 }
692+ if e.retainedCause != nil {
693+ out = append(out, e.retainedCause)
694+ }
689 return out695 return out
690}696}
691 697 
@@ -2826,16 +2826,27 @@ func repositoryDocsRetryReady(state *RepositoryDocsMaintenanceState, now time.Ti
2826 2826 
2827func (m *MaintenanceManager) finishReconcileEntry(ctx context.Context, registrationID string, snapshot MaintenanceEntry, contentGeneration, covered int64, ragStatus, namespaceID string, frontiers []cache.MaintenanceFrontier, activeSync, activeRAG, activeRepositoryDocs Job, now time.Time) MaintenanceEntry {2827func (m *MaintenanceManager) finishReconcileEntry(ctx context.Context, registrationID string, snapshot MaintenanceEntry, contentGeneration, covered int64, ragStatus, namespaceID string, frontiers []cache.MaintenanceFrontier, activeSync, activeRAG, activeRepositoryDocs Job, now time.Time) MaintenanceEntry {
2828 m.mu.Lock()2828 m.mu.Lock()
2829+ defer m.mu.Unlock()
2829 entry := m.entries[registrationID]2830 entry := m.entries[registrationID]
2831+ if entry == nil {
2832+ return MaintenanceEntry{}
2833+ }
2834+ _, currentEntry := m.reconcileAuthorityLocked(&snapshot)
2835+ _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&snapshot)
2836+ if registrationID != snapshot.RegistrationID || !currentEntry || snapshot.RepositoryDocs != nil && !currentSourceAuthority {
2837+ return cloneMaintenanceEntry(entry)
2838+ }
2839+ if activeRepositoryDocs.ID != "" && !repositoryDocsJobMatchesSnapshot(activeRepositoryDocs, snapshot) {
2840+ activeRepositoryDocs = Job{}
2841+ }
2830 beforeObservation := maintenanceObservedState(entry)2842 beforeObservation := maintenanceObservedState(entry)
2831- observedSource := m.sources[registrationID][snapshot.repositorySourceID]2843+ observedSource := authoritySource
2832 beforeSourceObservation := maintenanceObservedSourceState(observedSource)2844 beforeSourceObservation := maintenanceObservedSourceState(observedSource)
2833 entry.ContentGeneration = contentGeneration2845 entry.ContentGeneration = contentGeneration
2834 entry.CoveredGeneration = covered2846 entry.CoveredGeneration = covered
2835 entry.RAGStatus = ragStatus2847 entry.RAGStatus = ragStatus
2836 entry.SyncStage = snapshot.SyncStage2848 entry.SyncStage = snapshot.SyncStage
2837 entry.RAGStage = snapshot.RAGStage2849 entry.RAGStage = snapshot.RAGStage
2838- _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&snapshot)
2839 if currentSourceAuthority {2850 if currentSourceAuthority {
2840 // Inspection is separately confirmed at its actual completion. A later2851 // Inspection is separately confirmed at its actual completion. A later
2841 // stale finish must not overwrite that transaction or bypass its failure.2852 // stale finish must not overwrite that transaction or bypass its failure.
@@ -2875,13 +2886,16 @@ func (m *MaintenanceManager) finishReconcileEntry(ctx context.Context, registrat
2875 entry.NextReconcileAt = now.Add(time.Minute)2886 entry.NextReconcileAt = now.Add(time.Minute)
2876 persisted := m.saveLocked() == nil2887 persisted := m.saveLocked() == nil
2877 captureMaintenanceObservation(ctx, beforeObservation, entry, persisted)2888 captureMaintenanceObservation(ctx, beforeObservation, entry, persisted)
2878- // An obsolete callback may retain an existing execution-state mutation. Do2889+ captureMaintenanceSourceObservation(ctx, beforeSourceObservation, entry, observedSource, persisted && currentSourceAuthority)
2879- // not certify it as a current source observation: conserve the changed pair2890+ return cloneMaintenanceEntry(entry)
2880- // as unconfirmed loss without altering execution or persistence behavior.2891+}
2881- captureMaintenanceSourceObservation(ctx, beforeSourceObservation, entry, observedSource, persisted && currentSourceAuthority && observedSource == authoritySource)2892+ 
2882- updated := cloneMaintenanceEntry(entry)2893+func repositoryDocsJobMatchesSnapshot(job Job, snapshot MaintenanceEntry) bool {
2883- m.mu.Unlock()2894+ state := snapshot.RepositoryDocs
2884- return updated2895+ return state != nil && job.Type == RepositoryDocsIndexJobType && jobActiveStatus(job.Status) &&
2896+ job.CacheUUID == snapshot.CacheUUID && job.RepoID == snapshot.RepoID &&
2897+ job.RegistrationID == snapshot.RegistrationID && job.SourceRegistrationID == state.SourceRegistrationID &&
2898+ job.SourceRegistrationGeneration == state.SourceRegistrationGeneration
2885}2899}
2886 2900 
2887func (m *MaintenanceManager) reconcileRepositoryDocsState(ctx context.Context, store *cache.SQLiteStore, entry *MaintenanceEntry, now time.Time) bool {2901func (m *MaintenanceManager) reconcileRepositoryDocsState(ctx context.Context, store *cache.SQLiteStore, entry *MaintenanceEntry, now time.Time) bool {
@@ -2962,11 +2976,11 @@ func (m *MaintenanceManager) updateRepositoryDocsFailure(ctx context.Context, id
2962 return MaintenanceEntry{}2976 return MaintenanceEntry{}
2963 }2977 }
2964 source := m.sources[id][strings.TrimSpace(sourceID)]2978 source := m.sources[id][strings.TrimSpace(sourceID)]
2965- if source == nil {2979+ _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&expected)
2980+ if id != expected.RegistrationID || !currentSourceAuthority || source == nil || source != authoritySource {
2966 return cloneMaintenanceEntry(entry)2981 return cloneMaintenanceEntry(entry)
2967 }2982 }
2968 beforeObservation := maintenanceObservedSourceState(source)2983 beforeObservation := maintenanceObservedSourceState(source)
2969- _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&expected)
2970 source.State.State = "degraded"2984 source.State.State = "degraded"
2971 source.State.LastErrorClass = class2985 source.State.LastErrorClass = class
2972 source.State.LastError = message2986 source.State.LastError = message
@@ -2976,7 +2990,7 @@ func (m *MaintenanceManager) updateRepositoryDocsFailure(ctx context.Context, id
2976 entry.LastReconciledAt = m.now()2990 entry.LastReconciledAt = m.now()
2977 entry.NextReconcileAt = entry.LastReconciledAt.Add(time.Minute)2991 entry.NextReconcileAt = entry.LastReconciledAt.Add(time.Minute)
2978 persisted := m.saveLocked() == nil2992 persisted := m.saveLocked() == nil
2979- captureMaintenanceSourceObservation(ctx, beforeObservation, entry, source, persisted && currentSourceAuthority && source == authoritySource)2993+ captureMaintenanceSourceObservation(ctx, beforeObservation, entry, source, persisted)
2980 return cloneMaintenanceEntry(entry)2994 return cloneMaintenanceEntry(entry)
2981}2995}
2982 2996 
@@ -112,7 +112,8 @@ func captureMaintenanceObservation(ctx context.Context, before maintenanceObserv
112}112}
113 113 
114// Source confirmation requires both the existing save and current originating114// Source confirmation requires both the existing save and current originating
115-// authority. Retained stale-owner changes are count-only loss, not certification.115+// authority. Failed retained writes are count-only loss, not certification;
116+// rejected obsolete callbacks mutate nothing and contribute neither.
116func captureMaintenanceSourceObservation(ctx context.Context, before maintenanceObservationState, entry *MaintenanceEntry, source *repositoryDocsRegisteredSource, confirmed bool) {117func captureMaintenanceSourceObservation(ctx context.Context, before maintenanceObservationState, entry *MaintenanceEntry, source *repositoryDocsRegisteredSource, confirmed bool) {
117 if source == nil {118 if source == nil {
118 return119 return
@@ -6,6 +6,7 @@ import (
6 "errors"6 "errors"
7 "os"7 "os"
8 "path/filepath"8 "path/filepath"
9+ "reflect"
9 "strings"10 "strings"
10 "sync"11 "sync"
11 "sync/atomic"12 "sync/atomic"
@@ -107,7 +108,9 @@ func TestMaintenanceSourceObservationActualInspectionFailureRecovery(t *testing.
107 at = at.Add(time.Minute)108 at = at.Add(time.Minute)
108 run(Job{})109 run(Job{})
109 run(Job{}) // repeated successful inspection has no finite pair change110 run(Job{}) // repeated successful inspection has no finite pair change
110- run(Job{ID: "fixture-active-source"})111+ run(Job{ID: "fixture-active-source", Type: RepositoryDocsIndexJobType, Status: JobStatusQueued,
112+ CacheUUID: m.entries[reg].CacheUUID, RepoID: m.entries[reg].RepoID, RegistrationID: reg,
113+ SourceRegistrationID: id, SourceRegistrationGeneration: m.sources[reg][id].State.SourceRegistrationGeneration})
111 run(Job{})114 run(Job{})
112 if m.sources[reg][id].State.State != "not_indexed" || m.sources[reg][id].State.Inspection.Result != "succeeded" {115 if m.sources[reg][id].State.State != "not_indexed" || m.sources[reg][id].State.Inspection.Result != "succeeded" {
113 t.Fatal("actual inspection/indexing owner changed authority")116 t.Fatal("actual inspection/indexing owner changed authority")
@@ -312,10 +315,14 @@ func TestMaintenanceSourceObservationStaleSnapshotAndSharedOverflow(t *testing.T
312 t.Fatal("stale completion changed current source authority")315 t.Fatal("stale completion changed current source authority")
313 }316 }
314 buffer := ctx.Value(maintenanceObservationKey{}).(*maintenanceObservationBuffer)317 buffer := ctx.Value(maintenanceObservationKey{}).(*maintenanceObservationBuffer)
315- // Parent's actual enrolled->indexing is the only transition above. The stale318+ if buffer.count != 0 || buffer.lost != 0 {
316- // source's ready projection was never the current authority owner.319+ t.Fatal("rejected stale source callback manufactured an event or loss")
320+ }
321+ // A qualified current finish supplies the real enrolled->indexing parent
322+ // transition; obsolete work can no longer mutate even the parent state.
323+ m.finishReconcileEntry(ctx, reg, cloneMaintenanceEntryPrivate(m.entries[reg]), 0, 0, "", "", nil, Job{}, Job{}, Job{}, time.Now().UTC())
317 if buffer.count != 1 || buffer.captures[0].scope != "registration" {324 if buffer.count != 1 || buffer.captures[0].scope != "registration" {
318- t.Fatal("stale source completion manufactured an event")325+ t.Fatal("current completion lost parent observation")
319 }326 }
320 entry, source := m.entries[reg], m.sources[reg][id]327 entry, source := m.entries[reg], m.sources[reg][id]
321 for i := 0; i < maxMaintenanceObservations+5; i++ {328 for i := 0; i < maxMaintenanceObservations+5; i++ {
@@ -366,17 +373,32 @@ func TestMaintenanceSourceObservationActualObsoleteJobAndCurrentControl(t *testi
366 }373 }
367 c := maintenanceEventCollector(t)374 c := maintenanceEventCollector(t)
368 m.jobs.SetObservation(c)375 m.jobs.SetObservation(c)
376+ before := cloneMaintenanceEntryPrivate(m.entries[reg])
377+ beforeSource := *m.sources[reg][id]
378+ writes := 0
379+ m.writeFile = func(name string, data []byte, permissions os.FileMode) error {
380+ writes++
381+ return durableAtomicWriteFile(name, data, permissions)
382+ }
369 ctx, flush := beginMaintenanceObservation(context.Background(), m.jobs)383 ctx, flush := beginMaintenanceObservation(context.Background(), m.jobs)
370 defer flush()384 defer flush()
371 updated := m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC())385 updated := m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC())
372- // Execution semantics predate instrumentation and remain unchanged.386+ if stale {
373- if updated.State != "indexing" || m.sources[reg][id].State.State != "indexing" || m.sources[reg][id].State.SourceRegistrationGeneration != 2 {387+ if !reflect.DeepEqual(updated, cloneMaintenanceEntry(&before)) {
374- t.Fatal("observation changed retained execution behavior")388+ t.Fatal("obsolete completion returned stale state")
389+ }
390+ assertObsoleteSourceCallbackUnchanged(t, m, reg, id, before, beforeSource, writes)
391+ } else if updated.State != "indexing" || m.sources[reg][id].State.State != "indexing" || m.sources[reg][id].State.SourceRegistrationGeneration != 2 || writes == 0 {
392+ t.Fatal("current source completion did not apply")
375 }393 }
376 assertMaintenanceSourceAuthorityCaptures(t, ctx, stale)394 assertMaintenanceSourceAuthorityCaptures(t, ctx, stale)
377 // Unchanged replay must not retroactively certify or recount the pair.395 // Unchanged replay must not retroactively certify or recount the pair.
378 m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC())396 m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC())
379 assertMaintenanceSourceAuthorityCaptures(t, ctx, stale)397 assertMaintenanceSourceAuthorityCaptures(t, ctx, stale)
398+ if stale {
399+ assertObsoleteSourceCallbackUnchanged(t, m, reg, id, before, beforeSource, writes)
400+ }
401+ assertSourceAuthorityRestart(t, m, reg, id, *m.sources[reg][id])
380 })402 })
381 }403 }
382}404}
@@ -420,19 +442,33 @@ func TestMaintenanceSourceObservationActualLateFailureAndCurrentControl(t *testi
420 t.Fatal("actual source rebind unavailable")442 t.Fatal("actual source rebind unavailable")
421 }443 }
422 }444 }
445+ before := cloneMaintenanceEntryPrivate(m.entries[reg])
446+ beforeSource := *m.sources[reg][id]
447+ var writes atomic.Int64
448+ m.writeFile = func(name string, data []byte, permissions os.FileMode) error {
449+ writes.Add(1)
450+ return durableAtomicWriteFile(name, data, permissions)
451+ }
423 unblock()452 unblock()
424 select {453 select {
425 case updated := <-done:454 case updated := <-done:
426- if updated.RepositoryDocs == nil || updated.RepositoryDocs.State != "degraded" {455+ if stale {
427- t.Fatal("observation changed existing failure return")456+ if !reflect.DeepEqual(updated, cloneMaintenanceEntry(&before)) {
457+ t.Fatal("obsolete failure returned stale state")
458+ }
459+ } else if updated.RepositoryDocs == nil || updated.RepositoryDocs.State != "degraded" {
460+ t.Fatal("current source failure did not apply")
428 }461 }
429 case <-time.After(5 * time.Second):462 case <-time.After(5 * time.Second):
430 t.Fatal("actual source scheduling failure did not unwind")463 t.Fatal("actual source scheduling failure did not unwind")
431 }464 }
432- if m.sources[reg][id].State.State != "degraded" || m.sources[reg][id].State.LastErrorClass != "repository_docs_provider_boundary_blocked" {465+ if stale {
466+ assertObsoleteSourceCallbackUnchanged(t, m, reg, id, before, beforeSource, int(writes.Load()))
467+ } else if m.sources[reg][id].State.State != "degraded" || m.sources[reg][id].State.LastErrorClass != "repository_docs_provider_boundary_blocked" || writes.Load() == 0 {
433 t.Fatal("actual local-only scheduling boundary unavailable")468 t.Fatal("actual local-only scheduling boundary unavailable")
434 }469 }
435 assertMaintenanceSourceAuthorityCaptures(t, ctx, stale)470 assertMaintenanceSourceAuthorityCaptures(t, ctx, stale)
471+ assertSourceAuthorityRestart(t, m, reg, id, *m.sources[reg][id])
436 })472 })
437 }473 }
438}474}
@@ -448,10 +484,54 @@ func assertMaintenanceSourceAuthorityCaptures(t *testing.T, ctx context.Context,
448 }484 }
449 expected, lost := 1, uint64(0)485 expected, lost := 1, uint64(0)
450 if stale {486 if stale {
451- expected, lost = 0, 1487+ expected = 0
452 }488 }
453 if sourceCaptures != expected || buffer.lost != lost {489 if sourceCaptures != expected || buffer.lost != lost {
454- t.Fatal("current confirmation and obsolete count-only loss not conserved")490+ t.Fatal("source confirmation or unchanged obsolete callback accounting incorrect")
491+ }
492+}
493+ 
494+func assertObsoleteSourceCallbackUnchanged(t *testing.T, m *MaintenanceManager, reg, id string, before MaintenanceEntry, beforeSource repositoryDocsRegisteredSource, writes int) {
495+ t.Helper()
496+ if writes != 0 || !reflect.DeepEqual(cloneMaintenanceEntryPrivate(m.entries[reg]), before) || *m.sources[reg][id] != beforeSource {
497+ t.Fatal("obsolete callback changed replacement authority or persisted state")
498+ }
499+ list, err := m.List(context.Background())
500+ if err != nil || len(list.Entries) != 1 || !reflect.DeepEqual(list.Entries[0], cloneMaintenanceEntry(&before)) {
501+ t.Fatal("public maintenance projection changed after obsolete callback")
502+ }
503+}
504+ 
505+func assertSourceAuthorityRestart(t *testing.T, m *MaintenanceManager, reg, id string, want repositoryDocsRegisteredSource) {
506+ t.Helper()
507+ reloaded := NewMaintenanceManager(m.manager, NewJobManager(""), m.path)
508+ if reloaded.Load() != nil {
509+ t.Fatal("durable source authority changed across restart")
510+ }
511+ // Existing registry canonicalization may redirect a rebound source's
512+ // original id to the id of its replacement Git/worktree/profile authority.
513+ // Require that exact redirect, not an arbitrary substitute source.
514+ canonicalID := reloaded.resolveSourceRegistrationIDLocked(id)
515+ if canonicalID != repositoryDocsSourceRegistrationID(reg, want.State.GitStoreRef, want.State.WorktreeRef, want.Profile) || reloaded.sources[reg][canonicalID] == nil {
516+ t.Fatal("durable source redirect does not match replacement authority")
517+ }
518+ want.State.SourceRegistrationID = canonicalID
519+ got := *reloaded.sources[reg][canonicalID]
520+ if got != want {
521+ changed := []string{}
522+ if got.RepositoryPath != want.RepositoryPath {
523+ changed = append(changed, "path")
524+ }
525+ if got.Profile != want.Profile {
526+ changed = append(changed, "profile")
527+ }
528+ a, b := reflect.ValueOf(got.State), reflect.ValueOf(want.State)
529+ for i := 0; i < a.NumField(); i++ {
530+ if !reflect.DeepEqual(a.Field(i).Interface(), b.Field(i).Interface()) {
531+ changed = append(changed, a.Type().Field(i).Name)
532+ }
533+ }
534+ t.Fatalf("durable source authority changed across restart: fields %s", strings.Join(changed, ","))
455 }535 }
456}536}
457 537 
@@ -0,0 +1,136 @@
1+package servicectl
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "os"
7+ "reflect"
8+ "testing"
9+ "time"
10+)
11+ 
12+func TestRepositoryDocsCallbacksRejectMismatchedAuthority(t *testing.T) {
13+ for _, owner := range []string{"finish", "failure"} {
14+ for _, mismatch := range []string{"registration_generation", "registration_id", "cache_uuid", "repo_id", "physical_cache", "source_generation", "source_path", "source_profile", "selected_source", "state_source", "other_source_target", "missing_source", "missing_registration"} {
15+ t.Run(owner+"/"+mismatch, func(t *testing.T) {
16+ m, _, repo, reg, id := inspectionBoundaryFixture(t)
17+ targetID := id
18+ if mismatch == "other_source_target" {
19+ otherRepo := createRepositoryDocsRegistrationRepo(t, repo+"-other", "fixture-other-source")
20+ other, ok, err := m.RegisterRepositoryDocsSource(context.Background(), m.entries[reg].CacheUUID, m.entries[reg].RepoID, otherRepo, "")
21+ if err != nil || !ok || other.RepositoryDocs == nil {
22+ t.Fatal("second source fixture unavailable")
23+ }
24+ targetID = other.RepositoryDocs.SourceRegistrationID
25+ }
26+ snapshot := cloneMaintenanceEntryPrivate(m.entries[reg])
27+ state := m.sources[reg][id].State
28+ snapshot.RepositoryDocs = &state
29+ snapshot.repositorySourceID = id
30+ snapshot.repositoryPath = m.sources[reg][id].RepositoryPath
31+ snapshot.repositoryProfile = m.sources[reg][id].Profile
32+ snapshot.RepositoryDocs.State = "ready"
33+ snapshot.RepositoryDocs.RevisionSetID = "fixture-stale-set"
34+ switch mismatch {
35+ case "registration_generation":
36+ snapshot.Generation--
37+ case "registration_id":
38+ snapshot.RegistrationID = "fixture-other-registration"
39+ case "cache_uuid":
40+ snapshot.CacheUUID = "fixture-other-cache"
41+ case "repo_id":
42+ snapshot.RepoID = "fixture/other-repo"
43+ case "physical_cache":
44+ snapshot.PathFingerprint = "fixture-other-physical-cache"
45+ case "source_generation":
46+ snapshot.RepositoryDocs.SourceRegistrationGeneration--
47+ case "source_path":
48+ snapshot.repositoryPath += "-obsolete"
49+ case "source_profile":
50+ snapshot.repositoryProfile = "fixture-other-profile"
51+ case "selected_source":
52+ snapshot.repositorySourceID = "fixture-other-source"
53+ case "state_source":
54+ snapshot.RepositoryDocs.SourceRegistrationID = "fixture-other-source"
55+ case "missing_source":
56+ delete(m.sources[reg], id)
57+ case "missing_registration":
58+ delete(m.entries, reg)
59+ }
60+ before := cloneMaintenanceEntryPrivate(m.entries[reg])
61+ beforeSources := cloneRepositoryDocsSources(m.sources[reg])
62+ beforeList, err := m.List(context.Background())
63+ if err != nil {
64+ t.Fatal("public source fixture unavailable")
65+ }
66+ writes := 0
67+ m.writeFile = func(string, []byte, os.FileMode) error {
68+ writes++
69+ return errors.New("fixture unexpected authority write")
70+ }
71+ m.jobs.SetObservation(maintenanceEventCollector(t))
72+ ctx, flush := beginMaintenanceObservation(context.Background(), m.jobs)
73+ defer flush()
74+ if owner == "finish" {
75+ // A different failure target is not part of finish's contract;
76+ // model the corresponding mismatched selected-source callback.
77+ if mismatch == "other_source_target" {
78+ snapshot.repositorySourceID = targetID
79+ }
80+ m.finishReconcileEntry(ctx, reg, snapshot, 10, 9, "ready", "fixture-namespace", nil, Job{}, Job{}, Job{}, time.Now().UTC())
81+ } else {
82+ m.updateRepositoryDocsFailure(ctx, reg, targetID, snapshot, "git_store_unavailable", "fixture failure")
83+ }
84+ afterList, err := m.List(context.Background())
85+ if err != nil || writes != 0 || !reflect.DeepEqual(beforeList, afterList) || !reflect.DeepEqual(before, cloneMaintenanceEntryPrivate(m.entries[reg])) || !reflect.DeepEqual(beforeSources, m.sources[reg]) {
86+ t.Fatal("mismatched callback changed retained source or scheduling authority")
87+ }
88+ buffer := ctx.Value(maintenanceObservationKey{}).(*maintenanceObservationBuffer)
89+ if buffer.count != 0 || buffer.lost != 0 {
90+ t.Fatal("unchanged rejected callback manufactured observation evidence or loss")
91+ }
92+ })
93+ }
94+ }
95+}
96+ 
97+func TestRepositoryDocsCompletionQualifiesActiveJobIntent(t *testing.T) {
98+ for _, mismatch := range []string{"current", "type", "status", "cache", "repository", "registration", "source", "generation"} {
99+ t.Run(mismatch, func(t *testing.T) {
100+ m, _, _, reg, id := inspectionBoundaryFixture(t)
101+ snapshot := cloneMaintenanceEntryPrivate(m.entries[reg])
102+ before := m.sources[reg][id].State
103+ job := Job{ID: "fixture-active-job", Type: RepositoryDocsIndexJobType, Status: JobStatusQueued,
104+ CacheUUID: snapshot.CacheUUID, RepoID: snapshot.RepoID, RegistrationID: reg,
105+ SourceRegistrationID: id, SourceRegistrationGeneration: before.SourceRegistrationGeneration}
106+ switch mismatch {
107+ case "type":
108+ job.Type = RAGIndexJobType
109+ case "status":
110+ job.Status = JobStatusSuperseded
111+ case "cache":
112+ job.CacheUUID = "fixture-other-cache"
113+ case "repository":
114+ job.RepoID = "fixture/other-repo"
115+ case "registration":
116+ job.RegistrationID = "fixture-other-registration"
117+ case "source":
118+ job.SourceRegistrationID = "fixture-other-source"
119+ case "generation":
120+ job.SourceRegistrationGeneration--
121+ }
122+ // A registered but not-yet-indexed source normally makes the parent
123+ // registration indexing even without an active job. Compare that
124+ // existing derivation, not a fabricated ready parent status.
125+ baseline := m.finishReconcileEntry(context.Background(), reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, Job{}, time.Now().UTC())
126+ got := m.finishReconcileEntry(context.Background(), reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, job, time.Now().UTC())
127+ if mismatch == "current" {
128+ if m.sources[reg][id].State.State != "indexing" || got.State != "indexing" || len(got.ActiveJobs) != 1 || got.ActiveJobs[0] != job.ID {
129+ t.Fatal("current active source job did not apply")
130+ }
131+ } else if m.sources[reg][id].State != before || len(got.ActiveJobs) != 0 || got.State != baseline.State {
132+ t.Fatal("unrelated or terminal job copy acquired current source authority")
133+ }
134+ })
135+ }
136+}
@@ -56,16 +56,29 @@ func (m *MaintenanceManager) retainRepositoryDocsInspection(snapshot *Maintenanc
56}56}
57 57 
58func (m *MaintenanceManager) inspectionAuthorityLocked(snapshot *MaintenanceEntry) (*MaintenanceEntry, *repositoryDocsRegisteredSource, bool) {58func (m *MaintenanceManager) inspectionAuthorityLocked(snapshot *MaintenanceEntry) (*MaintenanceEntry, *repositoryDocsRegisteredSource, bool) {
59- entry := m.entries[snapshot.RegistrationID]59+ entry, currentEntry := m.reconcileAuthorityLocked(snapshot)
60 state := snapshot.RepositoryDocs60 state := snapshot.RepositoryDocs
61- if entry == nil || state == nil {61+ if !currentEntry || state == nil {
62 return entry, nil, false62 return entry, nil, false
63 }63 }
64 source := m.sources[snapshot.RegistrationID][state.SourceRegistrationID]64 source := m.sources[snapshot.RegistrationID][state.SourceRegistrationID]
65- current := source != nil && entry.Generation == snapshot.Generation && entry.CacheUUID == snapshot.CacheUUID && entry.RepoID == snapshot.RepoID && entry.PathFingerprint == snapshot.PathFingerprint && source.RepositoryPath == snapshot.repositoryPath && source.State.SourceRegistrationGeneration == state.SourceRegistrationGeneration65+ current := source != nil && state.SourceRegistrationID == snapshot.repositorySourceID &&
66+ source.State.SourceRegistrationID == state.SourceRegistrationID &&
67+ source.RepositoryPath == snapshot.repositoryPath && source.Profile == snapshot.repositoryProfile &&
68+ source.State.SourceRegistrationGeneration == state.SourceRegistrationGeneration
66 return entry, source, current69 return entry, source, current
67}70}
68 71 
72+// Completion authority belongs to the snapshot that started the work, not a
73+// replacement registration found later under the same map key. Call under mu
74+// before mutating either scheduling state or source state.
75+func (m *MaintenanceManager) reconcileAuthorityLocked(snapshot *MaintenanceEntry) (*MaintenanceEntry, bool) {
76+ entry := m.entries[snapshot.RegistrationID]
77+ return entry, entry != nil && entry.RegistrationID == snapshot.RegistrationID &&
78+ entry.Generation == snapshot.Generation && entry.CacheUUID == snapshot.CacheUUID &&
79+ entry.RepoID == snapshot.RepoID && entry.PathFingerprint == snapshot.PathFingerprint
80+}
81+ 
69func inspectionHealthReason(class string) string {82func inspectionHealthReason(class string) string {
70 switch class {83 switch class {
71 case "git_store":84 case "git_store":