diff --git a/docs/llm-task-artifacts.md b/docs/llm-task-artifacts.md index ff7fc9b3..409e783a 100644 --- a/docs/llm-task-artifacts.md +++ b/docs/llm-task-artifacts.md @@ -112,8 +112,9 @@ differ. reuse the output only when the metadata schema and input fingerprint still match the current task. -`failed_isolated` is for reviewer-local LLM failures while the caller context is -still valid. This includes structured validation failures and provider failures +`failed_isolated` is for reviewer-local LLM failures and review-time thread +analysis failures while the caller context is still valid. This includes +structured validation failures and provider failures after a task provider session has started. Provider start failures with no session are treated as blocking because they can indicate auth, quota, or other systemic adapter problems. The failed reviewer is treated as @@ -122,6 +123,37 @@ diagnostic. Sibling reviewers continue to run. A review with any isolated reviewer failure must not approve; the final event is clamped to at least `comment`. +Review-time thread analysis uses the same isolation boundary. An isolated thread +failure leaves that thread untouched, emits a warning, and appears under +**Unanalyzed Threads** in the posted rollup. Public diagnostics report total +counts and at most five deterministic examples, with each ID and diagnostic +sampled to 500 Unicode runes. The complete `thread_analysis_failures` collection +remains in the dry-run JSON summary, and every failure remains in the local +per-thread task artifacts. Successful sibling analyses still produce replies and +resolutions, and reviewer execution continues. Proposed selection actions for a +failed thread are suppressed. Any unanalyzed thread withholds an otherwise +approving review, even when the reviewers inspected every changed file. + +Matching isolated thread failures are reused when resuming an interrupted run; +resume does not spend another provider call on them. Use `--rerun` to retry in a +fresh run. Existing `failed_blocking` thread tasks still retry on resume, and +matching successful analyses remain cached. Public diagnostics do not echo +rejected model values or provider error text; detailed errors remain in the local +task artifacts. Cancellation, provider failures without execution evidence, +stale or corrupt artifacts, and persistence or checkpoint errors remain +blocking. Response-only `cr respond` retains its blocking thread-analysis +behavior. + +The synthesis prompt includes bounded failure counts and examples plus a digest +of the complete typed thread failure evidence when present. Its fingerprint +therefore changes even when an omitted failure changes. The existing coverage +artifact, relocation manifest, context contract, and pinned symlink metadata +inspection dependencies remain intact. Prompts without thread failures remain +unchanged, preserving reuse of existing successful synthesis tasks. This change +requires no additional task-metadata schema bump: existing blocking tasks retain +their retry semantics, and new isolated thread tasks use the existing status +contract. + `failed_blocking` means the task prevents dependent phases from safely running. Selection and rollup failures are blocking. `reviewrun` is the sole owner of live planner-error outcomes: blocking LLM failures leave fresh and resumed runs diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 4c27c498..231b8794 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -292,6 +292,7 @@ type Result struct { CurrentHeadSHA string ReviewBaseSHA string ReviewHeadSHA string + ThreadAnalysisFailures []threadanalysis.Failure ReviewerFailures []ReviewerFailure ReviewerCoverage []reviewplan.ReviewerCoverageSummary RelocationAssessments []ReviewerRelocationAssessments @@ -895,12 +896,13 @@ func executeLLMPhases(ctx context.Context, opts Options, req Request, mode execu if reusedCohort && !selectionInRun { selectionTaskIDs = nil } - threadResponses, err := analyzeReviewThreads(ctx, opts, req, run, prepared.artifacts, prepared.threadContext, namedSession.resumeID(), func(sessionID string) error { + threadResponses, threadFailures, err := analyzeReviewThreads(ctx, opts, req, run, prepared.artifacts, prepared.threadContext, namedSession.resumeID(), func(sessionID string) error { return namedSession.checkpointProviderSessionID(ctx, sessionID, opts.now()) }) if err != nil { return executionPhaseFailure(err) } + result.ThreadAnalysisFailures = threadFailures checkpointActions, err := checkpointThreadResponses(ctx, opts, req, mode, run, result.EffectiveCaps, threadResponses) if err != nil { return nil, false, err @@ -929,7 +931,7 @@ func executeLLMPhases(ctx context.Context, opts Options, req Request, mode execu return nil, false, Failure(FailureTerminal, err) } rollupModel, rollupEffort := rollupRuntimeConfig.model, rollupRuntimeConfig.effort - rollupPrompt, err := buildRollupPrompt(prepared.reviewPR, reviewerRun.findings, reviewerRun.failures, reviewerCoverage) + rollupPrompt, err := buildRollupPrompt(prepared.reviewPR, reviewerRun.findings, reviewerRun.failures, reviewerCoverage, threadFailureSummaries(threadFailures)) if err != nil { return nil, false, Failure(FailureTerminal, err) } @@ -979,6 +981,7 @@ func executeLLMPhases(ctx context.Context, opts Options, req Request, mode execu plan, err := opts.buildPlan(req, prepared.reviewPR, mode.planPostMode, result.EffectiveCaps, prepared.parsed.PlanDiff, reviewerRun.findings, rollup, selection.ThreadActions, false, result.AgentDefsChanged, planRunInputs{ threadResponses: threadResponses, + threadFailures: threadFailures, repoSources: repoSources, hasRun: true, selection: selectionSession, @@ -1393,16 +1396,16 @@ func runSelectionPhase(ctx context.Context, opts Options, req selectionPhaseRequ return selection, selectionSession, ledgerSession, nil } -func analyzeReviewThreads(ctx context.Context, opts Options, req Request, run ledger.Run, artifacts ArtifactPaths, threads []threadcontext.Thread, resumeSessionID string, onSessionID func(string) error) ([]review.ThreadResponseAction, error) { +func analyzeReviewThreads(ctx context.Context, opts Options, req Request, run ledger.Run, artifacts ArtifactPaths, threads []threadcontext.Thread, resumeSessionID string, onSessionID func(string) error) ([]review.ThreadResponseAction, []threadanalysis.Failure, error) { eligible := threadcontext.PendingCRAuthoredFindingThreads(threads) if len(eligible) == 0 { - return nil, nil + return nil, nil, nil } runtimeConfig, err := resolveThreadAnalysisRuntimeConfig(req.Profile) if err != nil { - return nil, err + return nil, nil, err } - results, err := threadanalysis.AnalyzeThreads(ctx, threadanalysis.Options{ + results, failures, err := threadanalysis.AnalyzeThreads(ctx, threadanalysis.Options{ Store: opts.Store, RunID: run.RunID, Adapter: opts.Adapter, @@ -1414,13 +1417,17 @@ func analyzeReviewThreads(ctx context.Context, opts Options, req Request, run le NewStepID: opts.newSessionRowID, ResumeSessionID: resumeSessionID, OnSessionID: onSessionID, + IsolateFailures: true, }, eligible, func(thread threadcontext.Thread) (string, error) { return artifacts.ThreadAnalysisLog(string(thread.ID)) }) if err != nil { - return nil, pipelineTaskError(err) + return nil, nil, pipelineTaskError(err) + } + for _, failure := range failures { + opts.emitWarning(fmt.Sprintf("thread %s was not analyzed: %s; isolated failures are cached on resume, use --rerun to retry", failure.ThreadID, failure.Error)) } - return threadanalysis.ResponseActions(results), nil + return threadanalysis.ResponseActions(results), failures, nil } func checkpointThreadResponses(ctx context.Context, opts Options, req Request, mode executionMode, run ledger.Run, caps reviewplan.ProviderCaps, responses []review.ThreadResponseAction) ([]ledger.PlannedAction, error) { @@ -2775,6 +2782,7 @@ type planRunInputs struct { rollup sessionDraft selectedAgents []llm.SelectedAgent threadResponses []review.ThreadResponseAction + threadFailures []threadanalysis.Failure repoSources []agents.SourceInfo findingSessions map[review.FindingID]string reviewerFailures []ReviewerFailure @@ -2845,15 +2853,16 @@ func (opts Options) buildRunSummary(req Request, inputs planRunInputs) (reviewpl wallMS := opts.now().Sub(inputs.startedAt).Milliseconds() summary := reviewplan.RunSummary{ - ToolVersion: req.ToolVersion, - Adapter: runAdapter(inputs), - Model: sharedWorkstreamModel(workstreams), - PostingIdentity: runlifecycle.PostingKey(req.PostingIdentity), - SelectedReviewers: selectedIDs, - ReviewerFailures: reviewerFailureSummaries(inputs.reviewerFailures), - ReviewerCoverage: inputs.reviewerCoverage, - WallDurationMS: &wallMS, - Workstreams: workstreams, + ToolVersion: req.ToolVersion, + Adapter: runAdapter(inputs), + Model: sharedWorkstreamModel(workstreams), + PostingIdentity: runlifecycle.PostingKey(req.PostingIdentity), + SelectedReviewers: selectedIDs, + ReviewerFailures: reviewerFailureSummaries(inputs.reviewerFailures), + ThreadAnalysisFailures: threadFailureSummaries(inputs.threadFailures), + ReviewerCoverage: inputs.reviewerCoverage, + WallDurationMS: &wallMS, + Workstreams: workstreams, } findingReviewers := make(map[review.FindingID]string, len(inputs.findingSessions)) @@ -2991,6 +3000,14 @@ func runAdapter(inputs planRunInputs) string { return strings.TrimSpace(inputs.rollup.Adapter) } +func threadFailureSummaries(failures []threadanalysis.Failure) []reviewplan.ThreadAnalysisFailureSummary { + out := make([]reviewplan.ThreadAnalysisFailureSummary, 0, len(failures)) + for _, failure := range failures { + out = append(out, reviewplan.ThreadAnalysisFailureSummary{ThreadID: failure.ThreadID, Error: failure.Error}) + } + return out +} + func reviewerFailureSummaries(failures []ReviewerFailure) []reviewplan.ReviewerFailureSummary { out := make([]reviewplan.ReviewerFailureSummary, 0, len(failures)) for _, failure := range failures { diff --git a/internal/pipeline/prompts.go b/internal/pipeline/prompts.go index f58e6e0b..a6001273 100644 --- a/internal/pipeline/prompts.go +++ b/internal/pipeline/prompts.go @@ -822,7 +822,7 @@ func selectionThreadStatus(thread threadcontext.Thread) string { } } -func buildRollupPrompt(pr gitprovider.PR, findings []review.Finding, reviewerFailures []ReviewerFailure, reviewerCoverage []reviewplan.ReviewerCoverageSummary) (string, error) { +func buildRollupPrompt(pr gitprovider.PR, findings []review.Finding, reviewerFailures []ReviewerFailure, reviewerCoverage []reviewplan.ReviewerCoverageSummary, threadFailures ...[]reviewplan.ThreadAnalysisFailureSummary) (string, error) { payload := map[string]any{ "task": "dedupe findings and return rollup JSON only", "output_contract": rollupOutputContract(findings), @@ -832,6 +832,9 @@ func buildRollupPrompt(pr gitprovider.PR, findings []review.Finding, reviewerFai "reviewer_failures": reviewerFailures, "reviewer_coverage": rollupCoveragePrompt(reviewerCoverage), } + if len(threadFailures) > 0 && len(threadFailures[0]) > 0 { + payload["thread_analysis_failures"] = rollupThreadAnalysisFailurePrompt(threadFailures[0]) + } body, err := json.Marshal(payload) if err != nil { return "", fmt.Errorf("pipeline: build rollup prompt: %w", err) @@ -839,6 +842,38 @@ func buildRollupPrompt(pr gitprovider.PR, findings []review.Finding, reviewerFai return string(body), nil } +type rollupThreadAnalysisFailureSummary struct { + Count int `json:"count"` + Examples []reviewplan.ThreadAnalysisFailureSummary `json:"examples"` + Omitted int `json:"omitted"` + EvidenceDigest string `json:"evidence_digest"` +} + +func rollupThreadAnalysisFailurePrompt(failures []reviewplan.ThreadAnalysisFailureSummary) rollupThreadAnalysisFailureSummary { + // Bind the complete evidence independently of the sampled prompt. A changed + // omitted failure must invalidate cached synthesis just like a visible one. + complete := append([]reviewplan.ThreadAnalysisFailureSummary(nil), failures...) + slices.SortFunc(complete, func(a, b reviewplan.ThreadAnalysisFailureSummary) int { + if order := strings.Compare(a.ThreadID, b.ThreadID); order != 0 { + return order + } + return strings.Compare(a.Error, b.Error) + }) + digest := sha256.New() + for _, failure := range complete { + // Length-prefix raw strings so even invalid UTF-8 repaired for display + // cannot make distinct complete evidence share a reuse fingerprint. + _, _ = fmt.Fprintf(digest, "%d:%s%d:%s", len(failure.ThreadID), failure.ThreadID, len(failure.Error), failure.Error) + } + examples := reviewplan.ThreadAnalysisFailureExamples(complete) + return rollupThreadAnalysisFailureSummary{ + Count: len(complete), + Examples: examples, + Omitted: len(complete) - len(examples), + EvidenceDigest: hex.EncodeToString(digest.Sum(nil)), + } +} + type rollupCoveragePromptEntry struct { AgentID string `json:"agent_id"` Status string `json:"status"` diff --git a/internal/pipeline/thread_analysis_failure_test.go b/internal/pipeline/thread_analysis_failure_test.go new file mode 100644 index 00000000..e9fca2c5 --- /dev/null +++ b/internal/pipeline/thread_analysis_failure_test.go @@ -0,0 +1,159 @@ +package pipeline + +import ( + "bytes" + "context" + "errors" + "fmt" + "strings" + "testing" + + "github.com/open-cli-collective/codereview-cli/internal/gitprovider" + "github.com/open-cli-collective/codereview-cli/internal/ledger" + "github.com/open-cli-collective/codereview-cli/internal/llm" + "github.com/open-cli-collective/codereview-cli/internal/llmlifecycle" + "github.com/open-cli-collective/codereview-cli/internal/outbox" + "github.com/open-cli-collective/codereview-cli/internal/review" + "github.com/open-cli-collective/codereview-cli/internal/reviewplan" + "github.com/open-cli-collective/codereview-cli/internal/statepaths" +) + +func TestLiveThreadAnalysisFailureContinuesAndPosts(t *testing.T) { + for _, invalidOutput := range []bool{false, true} { + t.Run(fmt.Sprintf("invalid_output=%t", invalidOutput), func(t *testing.T) { + ctx := context.Background() + store := openPipelineStore(t) + defer closeStore(t, store) + provider, req := dryRunHarness(t) + provider.caps.ThreadResolution = true + for i := 1; i <= 3; i++ { + provider.threads = append(provider.threads, markedReviewThread(t, fmt.Sprintf("thread-%d", i), "main.go", 2, req.PostingIdentity, gitprovider.Identity{Login: "human", ID: "human-id"})) + } + run := allocateLiveRun(t, store, provider, req, "run-partial-thread-analysis") + adapter := &llm.FakeAdapter{NameValue: "fake-llm"} + queuePartialThreadReview(t, adapter, invalidOutput) + var warnings bytes.Buffer + result, err := liveForTest(ctx, Options{ + Provider: provider, Adapter: adapter, Store: store, + Layout: statepaths.NewLayout(t.TempDir(), t.TempDir()), Now: fixedNow, + NewSessionRowID: sequence("session"), NewFindingID: findingSequence("finding"), NewActionID: actionSequence(), + MaxConcurrency: 1, Warnings: &warnings, + }, req, run) + if err != nil { + t.Fatalf("Live: %v", err) + } + if result.Plan.Outcome != reviewplan.OutcomeComment || len(result.ThreadAnalysisFailures) != 1 || result.ThreadAnalysisFailures[0].ThreadID != "thread-2" { + t.Fatalf("plan/failures = %s/%#v, want completed non-approving partial review", result.Plan.Outcome, result.ThreadAnalysisFailures) + } + if len(result.ReviewerCoverage) != 1 || result.ReviewerCoverage[0].Status != reviewerCoverageCompleteBroad { + t.Fatalf("reviewer coverage = %#v, want reviewer phase to run", result.ReviewerCoverage) + } + for _, want := range []string{"thread-2", "not analyzed", "cached on resume", "--rerun"} { + if !strings.Contains(warnings.String(), want) || !strings.Contains(result.Plan.RollupMarkdown, want) { + t.Fatalf("missing diagnostic %q: warnings=%s rollup=%s", want, warnings.String(), result.Plan.RollupMarkdown) + } + } + requests := adapter.Requests() + if !strings.Contains(requests[len(requests)-1].Prompt, `"thread_analysis_failures"`) || !strings.Contains(requests[len(requests)-1].Prompt, `"thread_id":"thread-2"`) { + t.Fatalf("rollup prompt missing failure: %s", requests[len(requests)-1].Prompt) + } + for _, action := range result.PlannedActions { + if action.ThreadID == "thread-2" { + t.Fatalf("failed thread has planned action: %#v", action) + } + } + for i := 1; i <= 3; i++ { + meta, ok, err := llmlifecycle.ReadMetadata(lifecyclePaths(result.Artifacts), fmt.Sprintf("thread-analysis-thread-%d", i)) + want := llmlifecycle.StatusSucceeded + if i == 2 { + want = llmlifecycle.StatusFailedIsolated + } + if err != nil || !ok || meta.Status != want { + t.Fatalf("thread %d metadata = %#v ok=%t err=%v", i, meta, ok, err) + } + } + posting := &gitprovider.Fake{} + posting.SetCapabilities(provider.caps) + if err := posting.SetPR(req.PRRef, provider.pr); err != nil { + t.Fatal(err) + } + if err := posting.SetInlineThreads(req.PRRef, provider.threads); err != nil { + t.Fatal(err) + } + posted, err := outbox.Post(ctx, outbox.Options{Store: store, Provider: posting, Limiter: threadIsolationLimiter{}, Now: fixedNow}, outbox.Request{Run: run, PRRef: req.PRRef, PostingIdentity: req.PostingIdentity, DesiredOutcome: ledger.OutcomeComment}) + if err != nil || posted.ExitCode != 0 || posted.Outcome != ledger.OutcomeComment { + t.Fatalf("Post = %#v err=%v, want completed review", posted, err) + } + if replies := posting.RecordedThreadReplies(req.PRRef); len(replies) != 2 { + t.Fatalf("replies = %#v, want both successful siblings", replies) + } + resolved := posting.RecordedResolvedThreads(req.PRRef) + if len(resolved) != 2 { + t.Fatalf("resolutions = %#v, want successful siblings only", resolved) + } + for _, id := range resolved { + if id == "thread-2" { + t.Fatal("unanalyzed thread was resolved") + } + } + if reviews := posting.RecordedReviews(req.PRRef); len(reviews) != 1 || reviews[0].Event != review.ReviewEventComment || !strings.Contains(reviews[0].Body, "Unanalyzed Threads") { + t.Fatalf("posted reviews = %#v, want reported partial coverage", reviews) + } + }) + } +} + +func TestLiveResumeCachesIsolatedThreadAnalysisFailure(t *testing.T) { + ctx := context.Background() + store := openPipelineStore(t) + defer closeStore(t, store) + provider, req := dryRunHarness(t) + provider.threads = []gitprovider.InlineThread{markedReviewThread(t, "thread-2", "main.go", 2, req.PostingIdentity, gitprovider.Identity{Login: "human", ID: "human-id"})} + run := allocateLiveRun(t, store, provider, req, "run-resume-partial-thread-analysis") + adapter := &llm.FakeAdapter{NameValue: "fake-llm"} + adapter.Queue(fakeLLMResult("dossier", discussionSummaryJSON(nil, nil), 1, 1)) + adapter.Queue(fakeLLMResult("selection", selectionJSON("harness:reviewer", "main.go"), 1, 1)) + adapter.Queue(llm.FakeResult{SessionID: "failed-thread", WaitErr: errors.New("analysis provider failed")}) + adapter.Queue(fakeLLMResult("reviewer", findingsWithCoverageJSON("harness:reviewer", []string{"main.go"}, nil, nil, nil), 1, 1)) + adapter.Queue(llm.FakeResult{SessionID: "failed-rollup", WaitErr: errors.New("rollup provider failed")}) + opts := Options{Provider: provider, Adapter: adapter, Store: store, Layout: statepaths.NewLayout(t.TempDir(), t.TempDir()), Now: fixedNow, NewSessionRowID: sequence("session"), NewFindingID: findingSequence("finding"), NewActionID: actionSequence(), MaxConcurrency: 1} + if _, err := liveForTest(ctx, opts, req, run); err == nil || ClassifyFailure(err) != FailureDurableBlocking { + t.Fatalf("first Live error = %v, want blocking rollup failure", err) + } + resumed := &llm.FakeAdapter{NameValue: "fake-llm"} + resumed.Queue(fakeLLMResult("recovered-rollup", rollupJSON("approve", nil), 1, 1)) + opts.Adapter = resumed + opts.NewSessionRowID = sequence("resumed-session") + result, err := liveForTest(ctx, opts, req, run) + if err != nil { + t.Fatalf("resumed Live: %v", err) + } + if len(resumed.Requests()) != 1 || !strings.Contains(resumed.Requests()[0].Prompt, `"schema":"rollup"`) { + t.Fatalf("resume requests = %#v, want only blocking rollup retry", resumed.Requests()) + } + if len(result.ThreadAnalysisFailures) != 1 || result.Plan.Outcome != reviewplan.OutcomeComment || !strings.Contains(result.Plan.RollupMarkdown, "thread-2") { + t.Fatalf("resumed partial outcome = %#v/%s", result.ThreadAnalysisFailures, result.Plan.Outcome) + } +} + +func queuePartialThreadReview(t *testing.T, adapter *llm.FakeAdapter, invalidOutput bool) { + t.Helper() + adapter.Queue(fakeLLMResult("dossier", discussionSummaryJSON(nil, nil), 1, 1)) + selection := strings.Replace(selectionJSON("harness:reviewer", "main.go"), `"thread_actions": []`, `"thread_actions": [{"thread_id":"thread-2","decision":"summarize_and_resolve","summary":"selector proposed closure","safe_to_resolve_rationale":"settled"}]`, 1) + adapter.Queue(fakeLLMResult("selection", selection, 1, 1)) + adapter.Queue(fakeLLMResult("thread-1", `{"thread_id":"thread-1","decision":"summarize","summary":"First thread settled","resolve":true,"rationale":"settled"}`, 1, 1)) + if invalidOutput { + for i := 0; i < 2; i++ { + adapter.Queue(fakeLLMResult(fmt.Sprintf("thread-2-attempt-%d", i), `{"thread_id":"thread-2","decision":"invalid","resolve":false}`, 1, 1)) + } + } else { + adapter.Queue(llm.FakeResult{SessionID: "thread-2-failed", WaitErr: errors.New("thread provider failed")}) + } + adapter.Queue(fakeLLMResult("thread-3", `{"thread_id":"thread-3","decision":"summarize","summary":"Third thread settled","resolve":true,"rationale":"settled"}`, 1, 1)) + adapter.Queue(fakeLLMResult("reviewer", findingsWithCoverageJSON("harness:reviewer", []string{"main.go"}, nil, nil, nil), 1, 1)) + adapter.Queue(fakeLLMResult("rollup", rollupJSON("approve", nil), 1, 1)) +} + +type threadIsolationLimiter struct{} + +func (threadIsolationLimiter) Wait(context.Context, string) error { return nil } diff --git a/internal/pipeline/thread_analysis_prompt_test.go b/internal/pipeline/thread_analysis_prompt_test.go new file mode 100644 index 00000000..3157eae2 --- /dev/null +++ b/internal/pipeline/thread_analysis_prompt_test.go @@ -0,0 +1,126 @@ +package pipeline + +import ( + "encoding/json" + "fmt" + "reflect" + "unicode/utf8" + + "strings" + "testing" + + "github.com/open-cli-collective/codereview-cli/internal/gitprovider" + "github.com/open-cli-collective/codereview-cli/internal/llmlifecycle" + "github.com/open-cli-collective/codereview-cli/internal/reviewplan" +) + +func TestThreadAnalysisFailuresChangeSynthesisFingerprint(t *testing.T) { + pr := gitprovider.PR{Title: "fixture"} + base, err := buildRollupPrompt(pr, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + empty, err := buildRollupPrompt(pr, nil, nil, nil, nil) + if err != nil || empty != base || strings.Contains(base, "thread_analysis_failures") { + t.Fatalf("no-failure prompt changed: base=%s empty=%s err=%v", base, empty, err) + } + fingerprint := func(prompt string) string { + return llmlifecycle.Fingerprint("fake", orchestratorRollupStage, "rollup", "model", "effort", prompt, nil) + } + seen := map[string]bool{fingerprint(base): true} + for _, failure := range []reviewplan.ThreadAnalysisFailureSummary{ + {ThreadID: "thread-1", Error: "structured output remained invalid after retry"}, + {ThreadID: "thread-2", Error: "structured output remained invalid after retry"}, + {ThreadID: "thread-1", Error: "provider execution failed; see local task artifacts for details"}, + } { + prompt, err := buildRollupPrompt(pr, nil, nil, nil, []reviewplan.ThreadAnalysisFailureSummary{failure}) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(prompt, failure.ThreadID) || !strings.Contains(prompt, failure.Error) || !strings.Contains(prompt, "thread_analysis_failures") { + t.Fatalf("prompt missing failure evidence: %s", prompt) + } + fp := fingerprint(prompt) + if seen[fp] { + t.Fatalf("failure evidence did not change fingerprint: %#v", failure) + } + seen[fp] = true + } +} + +func TestThreadFailureSynthesisPromptBoundsExamplesAndBindsOmittedEvidence(t *testing.T) { + failures := make([]reviewplan.ThreadAnalysisFailureSummary, 2500) + for i := range failures { + failures[i] = reviewplan.ThreadAnalysisFailureSummary{ThreadID: fmt.Sprintf("thread-%04d", i), Error: strings.Repeat("界", 800)} + } + original := append([]reviewplan.ThreadAnalysisFailureSummary(nil), failures...) + prompt, err := buildRollupPrompt(gitprovider.PR{}, nil, nil, nil, failures) + if err != nil { + t.Fatal(err) + } + var payload struct { + Failures rollupThreadAnalysisFailureSummary `json:"thread_analysis_failures"` + } + if err := json.Unmarshal([]byte(prompt), &payload); err != nil { + t.Fatal(err) + } + if payload.Failures.Count != 2500 || payload.Failures.Omitted != 2495 || len(payload.Failures.Examples) != 5 || len(payload.Failures.EvidenceDigest) != 64 { + t.Fatalf("failure prompt summary = %#v", payload.Failures) + } + if len(prompt) > 12000 || strings.Contains(prompt, "thread-0005") || strings.Contains(prompt, "thread-2499") || !utf8.ValidString(prompt) { + t.Fatalf("failure prompt bytes=%d, want bounded examples", len(prompt)) + } + if !reflect.DeepEqual(failures, original) { + t.Fatal("prompt sampling mutated complete failure collection") + } + changed := append([]reviewplan.ThreadAnalysisFailureSummary(nil), failures...) + changed[5].ThreadID = "zz-omitted-thread-changed" + changedPrompt, err := buildRollupPrompt(gitprovider.PR{}, nil, nil, nil, changed) + if err != nil { + t.Fatal(err) + } + var changedPayload struct { + Failures rollupThreadAnalysisFailureSummary `json:"thread_analysis_failures"` + } + if err := json.Unmarshal([]byte(changedPrompt), &changedPayload); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(payload.Failures.Examples, changedPayload.Failures.Examples) { + t.Fatal("omitted evidence mutation unexpectedly changed visible examples") + } + fingerprint := func(value string) string { + return llmlifecycle.Fingerprint("fake", orchestratorRollupStage, "rollup", "model", "effort", value, []string{"coverage-artifact=current", "relocation-manifest=current", "context-contract=current"}) + } + if payload.Failures.EvidenceDigest == changedPayload.Failures.EvidenceDigest || fingerprint(prompt) == fingerprint(changedPrompt) { + t.Fatal("changed omitted failure did not invalidate complete-evidence fingerprint") + } + changed = append([]reviewplan.ThreadAnalysisFailureSummary(nil), failures...) + changed[5].Error = "provider execution failed; see local task artifacts for details" + changedPrompt, err = buildRollupPrompt(gitprovider.PR{}, nil, nil, nil, changed) + if err != nil { + t.Fatal(err) + } + if fingerprint(prompt) == fingerprint(changedPrompt) { + t.Fatal("changed omitted diagnostic did not invalidate fingerprint") + } + reordered := append([]reviewplan.ThreadAnalysisFailureSummary(nil), failures...) + for i, j := 0, len(reordered)-1; i < j; i, j = i+1, j-1 { + reordered[i], reordered[j] = reordered[j], reordered[i] + } + reorderedPrompt, err := buildRollupPrompt(gitprovider.PR{}, nil, nil, nil, reordered) + if err != nil || reorderedPrompt != prompt { + t.Fatalf("failure evidence digest is not deterministic across order, err=%v", err) + } +} + +func TestThreadFailureDigestBindsRawEvidenceBeforeDisplayRepair(t *testing.T) { + first := []reviewplan.ThreadAnalysisFailureSummary{{ThreadID: "thread-" + string([]byte{0xff}), Error: "provider execution failed"}} + second := []reviewplan.ThreadAnalysisFailureSummary{{ThreadID: "thread-�", Error: "provider execution failed"}} + firstSummary, secondSummary := rollupThreadAnalysisFailurePrompt(first), rollupThreadAnalysisFailurePrompt(second) + if !reflect.DeepEqual(firstSummary.Examples, secondSummary.Examples) { + t.Fatal("display repair should produce identical safe examples") + } + if firstSummary.EvidenceDigest == secondSummary.EvidenceDigest { + t.Fatal("distinct raw evidence shared a complete-evidence digest after display repair") + } +} diff --git a/internal/reviewplan/relocation_coverage_test.go b/internal/reviewplan/relocation_coverage_test.go index a440add6..22290ec6 100644 --- a/internal/reviewplan/relocation_coverage_test.go +++ b/internal/reviewplan/relocation_coverage_test.go @@ -32,7 +32,7 @@ func TestApprovalWithheldDoesNotCallValidRelocationUnread(t *testing.T) { RelocationReviewedFiles: []string{"old/module.go"}, }} var out strings.Builder - writeApprovalWithheld(&out, nil, coverage) + writeApprovalWithheld(&out, nil, coverage, nil) text := out.String() if !strings.Contains(text, "Every changed file was either body-inspected or received relocation-impact review") { t.Fatalf("withheld text does not describe relocation coverage accurately: %s", text) diff --git a/internal/reviewplan/reviewplan.go b/internal/reviewplan/reviewplan.go index fa454035..b84d3cb3 100644 --- a/internal/reviewplan/reviewplan.go +++ b/internal/reviewplan/reviewplan.go @@ -250,6 +250,27 @@ type builder struct { // Build turns validated review-domain values into a deterministic action plan. func Build(req Request) (Plan, error) { + // Selection runs before analysis. Its proposed thread actions cannot make + // an unanalyzed discussion safe to reply to or resolve. + if len(req.RunSummary.ThreadAnalysisFailures) > 0 { + failed := make(map[string]bool, len(req.RunSummary.ThreadAnalysisFailures)) + for _, failure := range req.RunSummary.ThreadAnalysisFailures { + failed[strings.TrimSpace(failure.ThreadID)] = true + } + var actions []review.ThreadAction + for _, action := range req.ThreadActions { + if !failed[strings.TrimSpace(action.ThreadID)] { + actions = append(actions, action) + } + } + var responses []review.ThreadResponseAction + for _, response := range req.ThreadResponses { + if !failed[strings.TrimSpace(response.ThreadID)] { + responses = append(responses, response) + } + } + req.ThreadActions, req.ThreadResponses = actions, responses + } b, err := newBuilder(req) if err != nil { return Plan{}, err @@ -463,7 +484,7 @@ func (b *builder) buildReview() (Plan, error) { } event = applySelfApprovalPolicy(event, b.req.EventOptions) approvalWithheld := false - if event == review.ReviewEventApprove && (len(b.req.RunSummary.ReviewerFailures) > 0 || hasIncompleteReviewerCoverage(b.req.RunSummary.ReviewerCoverage)) { + if event == review.ReviewEventApprove && (len(b.req.RunSummary.ReviewerFailures) > 0 || len(b.req.RunSummary.ThreadAnalysisFailures) > 0 || hasIncompleteReviewerCoverage(b.req.RunSummary.ReviewerCoverage)) { event = review.ReviewEventComment approvalWithheld = true } @@ -965,7 +986,7 @@ func (b *builder) renderRollup(ordered []review.Finding, anchored []AnchoredFind // misleads, so the explanation has to be the next thing read. withheld := func() { if approvalWithheld { - writeApprovalWithheld(&out, summary.Run.ReviewerFailures, summary.Run.ReviewerCoverage) + writeApprovalWithheld(&out, summary.Run.ReviewerFailures, summary.Run.ReviewerCoverage, summary.Run.ThreadAnalysisFailures) } } if len(summary.Reviewers) > 0 { @@ -995,6 +1016,7 @@ func (b *builder) renderRollup(ordered []review.Finding, anchored []AnchoredFind writeReviewerCoverageDiagnostics(&out, summary.Run.ReviewerCoverage) writeReviewerFailureDiagnostics(&out, summary.Run.ReviewerFailures) } + writeThreadAnalysisDiagnostics(&out, summary.Run.ThreadAnalysisFailures) writeThreadCountsLine(&out, summary.Threads) if b.req.AgentDefinitionsChanged { out.WriteString("\n---\n\n") diff --git a/internal/reviewplan/summary.go b/internal/reviewplan/summary.go index d69d01ae..18da144a 100644 --- a/internal/reviewplan/summary.go +++ b/internal/reviewplan/summary.go @@ -60,14 +60,15 @@ type ThreadCounts struct { type RunSummary struct { // ToolVersion is the raw version string (e.g. "0.3.63"); the renderer // adds the "cr " prefix. - ToolVersion string - Adapter string - Model string - PostingIdentity string - SelectedReviewers []string - ReviewerFailures []ReviewerFailureSummary - ReviewerCoverage []ReviewerCoverageSummary - WallDurationMS *int64 + ToolVersion string + Adapter string + Model string + PostingIdentity string + SelectedReviewers []string + ReviewerFailures []ReviewerFailureSummary + ThreadAnalysisFailures []ThreadAnalysisFailureSummary + ReviewerCoverage []ReviewerCoverageSummary + WallDurationMS *int64 // Workstreams holds the stages that ran in this round; a stage skipped by // reuse (e.g. selection under a reused reviewer cohort) is absent rather // than present with empty usage, which is what lets AggregateUsage's @@ -81,6 +82,13 @@ type ReviewerFailureSummary struct { Error string } +// ThreadAnalysisFailureSummary identifies an existing discussion that could not +// be analyzed. Its absence from thread responses does not imply it was settled. +type ThreadAnalysisFailureSummary struct { + ThreadID string `json:"thread_id"` + Error string `json:"error"` +} + // WorkstreamUsage is adapter-reported usage for one workstream: the reserved // orchestrator stage names or a reviewer agent ID. type WorkstreamUsage struct { @@ -119,7 +127,7 @@ type AggregateUsage struct { func (r RunSummary) hasData() bool { return r.ToolVersion != "" || r.Adapter != "" || r.Model != "" || r.PostingIdentity != "" || len(r.SelectedReviewers) > 0 || - len(r.ReviewerFailures) > 0 || len(r.ReviewerCoverage) > 0 || r.WallDurationMS != nil || + len(r.ReviewerFailures) > 0 || len(r.ThreadAnalysisFailures) > 0 || len(r.ReviewerCoverage) > 0 || r.WallDurationMS != nil || len(r.Workstreams) > 0 } @@ -311,6 +319,53 @@ func writeReviewerFailureDiagnostics(out *strings.Builder, failures []ReviewerFa out.WriteString("\n") } +func writeThreadAnalysisDiagnostics(out *strings.Builder, failures []ThreadAnalysisFailureSummary) { + if len(failures) == 0 { + return + } + out.WriteString("### Unanalyzed Threads\n\n") + examples := ThreadAnalysisFailureExamples(failures) + out.WriteString(threadAnalysisFailureCount(len(failures))) + if omitted := len(failures) - len(examples); omitted > 0 { + fmt.Fprintf(out, "; showing %d examples; %d omitted", len(examples), omitted) + } + out.WriteString(".\n\n") + for _, failure := range examples { + fmt.Fprintf(out, "- %s — not analyzed: %s\n", codeSpan(failure.ThreadID), escapeCell(failure.Error)) + } + out.WriteString("\nThese threads were left untouched. Complete failure evidence remains in the local task artifacts. Isolated analysis failures are cached on resume; use `--rerun` to retry with a fresh run.\n\n") +} + +const maxPublicThreadExamples = 5 + +// ThreadAnalysisFailureExamples returns deterministic, bounded display samples. +// The complete typed failure collection is left unchanged for local evidence, +// JSON presentation and safe cache fingerprinting. +func ThreadAnalysisFailureExamples(failures []ThreadAnalysisFailureSummary) []ThreadAnalysisFailureSummary { + examples := append([]ThreadAnalysisFailureSummary(nil), failures...) + sort.Slice(examples, func(i, j int) bool { + if examples[i].ThreadID != examples[j].ThreadID { + return examples[i].ThreadID < examples[j].ThreadID + } + return examples[i].Error < examples[j].Error + }) + if len(examples) > maxPublicThreadExamples { + examples = examples[:maxPublicThreadExamples] + } + for i := range examples { + examples[i].ThreadID = samplePublicDiagnostic(examples[i].ThreadID) + examples[i].Error = samplePublicDiagnostic(examples[i].Error) + } + return examples +} + +func threadAnalysisFailureCount(count int) string { + if count == 1 { + return "1 thread was not analyzed" + } + return fmt.Sprintf("%d threads were not analyzed", count) +} + // GitHub tables offer no column-width control, so variable-length values // (paths, constraint prose) wrap mid-word and repeat per row. Coverage is // rendered as a list instead, with bounded path examples and unknown fields @@ -526,9 +581,30 @@ func filesNotIn(files []string, exclude map[string]bool) []string { // already knows that incomplete coverage downgrades the event. Re-running does // not clear it either, since a reviewer that declined a file declines it again, // so a reader given no cause has no next step. -func writeApprovalWithheld(out *strings.Builder, failures []ReviewerFailureSummary, coverage []ReviewerCoverageSummary) { +func writeApprovalWithheld(out *strings.Builder, failures []ReviewerFailureSummary, coverage []ReviewerCoverageSummary, threadFailures []ThreadAnalysisFailureSummary) { out.WriteString("### Approval Withheld\n\n") - out.WriteString("No blocking or major findings were reported. Approval is withheld because at least one reviewer did not complete its assigned review obligations:\n\n") + threadOnly := len(failures) == 0 && !hasIncompleteReviewerCoverage(coverage) && len(threadFailures) > 0 + if threadOnly { + out.WriteString("No blocking or major findings were reported. Approval is withheld because existing discussions could not be analyzed:\n\n") + } else { + out.WriteString("No blocking or major findings were reported. Approval is withheld because at least one reviewer did not complete its assigned review obligations:\n\n") + } + if len(threadFailures) > 0 { + examples := ThreadAnalysisFailureExamples(threadFailures) + ids := make([]string, 0, len(examples)) + for _, failure := range examples { + ids = append(ids, codeSpan(failure.ThreadID)) + } + fmt.Fprintf(out, "- %s; examples: %s", threadAnalysisFailureCount(len(threadFailures)), strings.Join(ids, ", ")) + if omitted := len(threadFailures) - len(examples); omitted > 0 { + fmt.Fprintf(out, "; %d omitted", omitted) + } + out.WriteString(".\n") + } + if threadOnly { + out.WriteString("\nSee Unanalyzed Threads below for diagnostics and retry guidance.\n\n") + return + } for _, failure := range failures { fmt.Fprintf(out, "- %s did not produce a result: %s\n", codeSpan(failure.AgentID), escapeCell(samplePublicDiagnostic(failure.Error))) } diff --git a/internal/reviewplan/thread_analysis_failure_test.go b/internal/reviewplan/thread_analysis_failure_test.go new file mode 100644 index 00000000..97fa669b --- /dev/null +++ b/internal/reviewplan/thread_analysis_failure_test.go @@ -0,0 +1,139 @@ +package reviewplan + +import ( + "fmt" + "reflect" + "unicode/utf8" + + "strings" + "testing" + + "github.com/open-cli-collective/codereview-cli/internal/review" +) + +func TestBuildUnanalyzedThreadsWithholdApprovalAndSuppressActions(t *testing.T) { + for _, event := range []review.ReviewEvent{review.ReviewEventApprove, review.ReviewEventComment, review.ReviewEventRequestChanges} { + t.Run(string(event), func(t *testing.T) { + req := baseRequest() + req.Findings = nil + req.Rollup = review.Rollup{ReviewEvent: event} + if event != review.ReviewEventApprove { + f := finding("blocking", "main.go", review.Anchor{Kind: review.AnchorKindLine, Side: review.DiffSideRight, Line: 12}) + if event == review.ReviewEventRequestChanges { + f.Severity = review.SeverityBlocking + } + req.Findings = []review.Finding{f} + req.Rollup.OrderedFindings = []review.FindingID{f.ID} + } + req.ProviderCaps.ThreadResolution = true + req.RunSummary.ThreadAnalysisFailures = []ThreadAnalysisFailureSummary{{ThreadID: "failed-thread", Error: "invalid output"}} + req.ThreadActions = []review.ThreadAction{{ThreadID: "failed-thread", Decision: review.ThreadDecisionSummarizeAndResolve, Summary: "selector proposed closure"}} + req.ThreadResponses = []review.ThreadResponseAction{ + {ThreadID: "failed-thread", Kind: review.ThreadResponseSummaryReply, Body: "unsafe response", Resolve: true}, + {ThreadID: "successful-thread", Kind: review.ThreadResponseSummaryReply, Body: "settled", Resolve: true}, + } + plan, err := Build(req) + if err != nil { + t.Fatal(err) + } + for _, action := range plan.Actions { + if action.ThreadID == "failed-thread" { + t.Fatalf("action for unanalyzed thread = %#v", action) + } + } + if plan.Summary.Threads.Considered != 1 || plan.Summary.Threads.Resolved != 1 { + t.Fatalf("thread counts = %#v, want successful response only", plan.Summary.Threads) + } + want := OutcomeComment + if event == review.ReviewEventRequestChanges { + want = OutcomeRequestChanges + } + if plan.Outcome != want { + t.Fatalf("outcome = %q, want %q", plan.Outcome, want) + } + for _, want := range []string{"### Unanalyzed Threads", "failed-thread", "invalid output", "cached on resume", "--rerun"} { + if !strings.Contains(plan.RollupMarkdown, want) { + t.Fatalf("rollup missing %q: %s", want, plan.RollupMarkdown) + } + } + if got := strings.Contains(plan.RollupMarkdown, "### Approval Withheld"); got != (event == review.ReviewEventApprove) { + t.Fatalf("withheld section = %t for event %s", got, event) + } + }) + } +} + +func TestUnanalyzedThreadDiagnosticsBoundLargeCollections(t *testing.T) { + failures := make([]ThreadAnalysisFailureSummary, 2500) + for i := range failures { + failures[i] = ThreadAnalysisFailureSummary{ThreadID: fmt.Sprintf("thread-%04d", len(failures)-i-1), Error: strings.Repeat("界", 800)} + } + original := append([]ThreadAnalysisFailureSummary(nil), failures...) + for _, mixed := range []bool{false, true} { + t.Run(fmt.Sprintf("mixed_coverage=%t", mixed), func(t *testing.T) { + req := baseRequest() + req.Findings = nil + req.Rollup = review.Rollup{ReviewEvent: review.ReviewEventApprove} + req.RunSummary.ThreadAnalysisFailures = failures + if mixed { + req.RunSummary.ReviewerCoverage = []ReviewerCoverageSummary{{AgentID: "reviewer", Status: "incomplete_skipped", Scope: []string{"relocated.go", "missing.go"}, MissingFiles: []string{"missing.go"}, RelocationReviewedFiles: []string{"relocated.go"}}} + } + plan, err := Build(req) + if err != nil { + t.Fatal(err) + } + if plan.Outcome != OutcomeComment { + t.Fatalf("outcome = %s, want withheld approval", plan.Outcome) + } + for _, want := range []string{"2500 threads were not analyzed", "2495 omitted", "thread-0000", "thread-0004", "### Approval Withheld", "### Unanalyzed Threads"} { + if !strings.Contains(plan.RollupMarkdown, want) { + t.Errorf("rollup missing %q", want) + } + } + if strings.Contains(plan.RollupMarkdown, "thread-0005") || strings.Contains(plan.RollupMarkdown, "thread-2499") { + t.Fatal("public body rendered omitted thread IDs") + } + if len(plan.RollupMarkdown) > 15000 || !utf8.ValidString(plan.RollupMarkdown) { + t.Fatalf("public body bytes=%d utf8=%t, want bounded valid body", len(plan.RollupMarkdown), utf8.ValidString(plan.RollupMarkdown)) + } + if strings.Contains(plan.RollupMarkdown, strings.Repeat("界", 501)) { + t.Fatal("public diagnostic exceeded 500-rune limit") + } + if !reflect.DeepEqual(failures, original) || !reflect.DeepEqual(plan.Summary.Run.ThreadAnalysisFailures, original) { + t.Fatal("bounded rendering lost or mutated complete typed evidence") + } + if mixed { + for _, want := range []string{"assigned review obligations", "no body inspection or relocation-impact review", "missing.go", "relocation-impact-reviewed (not body-inspected)"} { + if !strings.Contains(plan.RollupMarkdown, want) { + t.Errorf("mixed coverage rollup lost %q", want) + } + } + } + var forward, reversed strings.Builder + writeThreadAnalysisDiagnostics(&forward, failures) + reordered := append([]ThreadAnalysisFailureSummary(nil), failures...) + for i, j := 0, len(reordered)-1; i < j; i, j = i+1, j-1 { + reordered[i], reordered[j] = reordered[j], reordered[i] + } + writeThreadAnalysisDiagnostics(&reversed, reordered) + if forward.String() != reversed.String() { + t.Fatal("thread diagnostic examples depend on source order") + } + }) + } +} + +func TestThreadFailureExamplesBoundLongIDsWithoutMutatingEvidence(t *testing.T) { + failures := []ThreadAnalysisFailureSummary{{ThreadID: strings.Repeat("🙂", 800) + string([]byte{0xff}), Error: strings.Repeat("界", 800) + ""}} + original := append([]ThreadAnalysisFailureSummary(nil), failures...) + examples := ThreadAnalysisFailureExamples(failures) + if len(examples) != 1 || utf8.RuneCountInString(examples[0].ThreadID) != 500 || utf8.RuneCountInString(examples[0].Error) != 500 { + t.Fatalf("example sizes = %#v, want 500-rune samples", examples) + } + if !utf8.ValidString(examples[0].ThreadID+examples[0].Error) || strings.Contains(examples[0].Error, "