Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 34 additions & 2 deletions docs/llm-task-artifacts.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
51 changes: 34 additions & 17 deletions internal/pipeline/pipeline.go
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,7 @@ type Result struct {
CurrentHeadSHA string
ReviewBaseSHA string
ReviewHeadSHA string
ThreadAnalysisFailures []threadanalysis.Failure
ReviewerFailures []ReviewerFailure
ReviewerCoverage []reviewplan.ReviewerCoverageSummary
RelocationAssessments []ReviewerRelocationAssessments
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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 {
Expand Down
37 changes: 36 additions & 1 deletion internal/pipeline/prompts.go
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand All @@ -832,13 +832,48 @@ 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)
}
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"`
Expand Down
Loading
Loading