From a36e816161e4852fe13e6c5e5a73993fd5bb90cb Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sat, 15 Aug 2026 01:26:36 +0100 Subject: [PATCH] Recover interrupted Patrol investigations --- .../v6/internal/subsystems/ai-runtime.md | 13 +- internal/ai/chat/agentic.go | 50 ++++++- internal/ai/chat/agentic_additional_test.go | 127 ++++++++++++++++++ .../ai/chat/agentic_investigation_budget.go | 16 +++ internal/ai/cost/pricing.go | 1 + internal/ai/cost/store_extended_test.go | 11 +- 6 files changed, 209 insertions(+), 9 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 2edb6068c..cdccacfc6 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -6359,6 +6359,16 @@ short default stall bound. A compatible provider pausing between streamed reasoning and its next tool-call chunk must not invalidate an otherwise live Patrol run, while a genuinely stalled stream must still terminate within the Patrol run budget. +Non-interactive Patrol turns may replay one provider stream that closes before +its completion marker, even when the provider emitted a partial fragment, +because no tool call or durable turn exists until the terminal marker arrives. +Interactive Assistant turns retain the no-replay rule after visible output. +When a Patrol investigation has already gathered its evidence but its final +tool-free response reaches the provider output limit, core must discard the +partial conclusion and allow one bounded, tool-free, low-reasoning synthesis +turn over the existing transcript. That recovery may only produce the required +investigation sections; it cannot gather more evidence or reopen mutation +authority, and a second truncation fails the investigation closed. Qualification scoring must keep synthetic Patrol runtime findings on the `ai-service` resource separate from model-authored infrastructure findings. Provider/runtime failure remains an unconditional qualification hard failure, @@ -6374,7 +6384,8 @@ OpenRouter `:free` route may be recorded at zero only after that route's public catalog entry is reviewed; the suffix alone never makes other free aliases known. The reviewed zero-priced qualification routes are currently `nvidia/nemotron-3.5-lightning:free` and -`nvidia/nemotron-3-super-120b-a12b:free`. Live scoring +`nvidia/nemotron-3-super-120b-a12b:free` (reviewed 2026-08-14), plus +`nvidia/nemotron-3-ultra-550b-a55b:free` (reviewed 2026-08-15). Live scoring uses the provider resolved by Patrol readiness, even when the configured OpenRouter model is an unprefixed slash route such as `anthropic/claude-sonnet-5`; scorer replay persists and reuses that resolved diff --git a/internal/ai/chat/agentic.go b/internal/ai/chat/agentic.go index 74337b9aa..5f51453f1 100644 --- a/internal/ai/chat/agentic.go +++ b/internal/ai/chat/agentic.go @@ -61,6 +61,7 @@ func isRetryableProviderStreamError(err error) bool { "temporarily unavailable", "http2: client connection lost", "stream error", + "stream ended before completion marker", "upstream", "502", "503", @@ -900,6 +901,8 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me patrolFindingRepairAttempted := false // The repair-only extension is bounded to one provider turn patrolOutputLimitRecoveryPending := false // A truncated Watch decision needs one decision-only retry patrolOutputLimitRecoveryAttempted := false + investigationOutputLimitRecoveryPending := false // A truncated investigation conclusion needs one evidence-only retry + investigationOutputLimitRecoveryAttempted := false toolBlockedLastTurn := false // When true, request final text after budget/loop block investigationProposalCompleted := false acceptedFindingReports := 0 @@ -940,7 +943,8 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me for turn < maxTurns || (patrolFindingRepairPending && !patrolFindingRepairAttempted) || - (patrolOutputLimitRecoveryPending && !patrolOutputLimitRecoveryAttempted) { + (patrolOutputLimitRecoveryPending && !patrolOutputLimitRecoveryAttempted) || + (investigationOutputLimitRecoveryPending && !investigationOutputLimitRecoveryAttempted) { // === CONTEXT COMPACTION: Compact old tool results to prevent context blowout === if turn > 0 { compactOldToolResults(providerMessages, currentTurnStartIndex, compactionKeepTurns, compactionMinChars, a.knowledgeAccumulator) @@ -1025,6 +1029,7 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me patrolFindingRepairTurn := false patrolFindingContinuationTurn := false patrolOutputLimitRecoveryTurn := false + investigationOutputLimitRecoveryTurn := false if patrolFindingRepairPending && !patrolFindingRepairAttempted { if applyPatrolFindingLifecycleRepairRequest(&req, a.currentExecutionProfile(), tools) { patrolFindingRepairTurn = true @@ -1051,6 +1056,18 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me Str("session_id", sessionID). Msg("[AgenticLoop] Watch output limit reached — retrying one finding-decision-only turn") } + if !patrolFindingRepairTurn && !patrolOutputLimitRecoveryTurn && investigationOutputLimitRecoveryPending && !investigationOutputLimitRecoveryAttempted { + if !applyInvestigationOutputLimitRecoveryRequest(&req, a.currentExecutionProfile()) { + return resultMessages, fmt.Errorf("Patrol investigation exhausted its output budget before a conclusion, and its evidence-only recovery profile is unavailable") + } + investigationOutputLimitRecoveryTurn = true + investigationOutputLimitRecoveryPending = false + investigationOutputLimitRecoveryAttempted = true + log.Warn(). + Int("turn", turn). + Str("session_id", sessionID). + Msg("[AgenticLoop] Investigation output limit reached — retrying one evidence-only conclusion turn") + } if !patrolFindingRepairTurn && !patrolOutputLimitRecoveryTurn && shouldOfferPatrolFindingLifecycleContinuation(patrolFindingContinuationPending, writeCompletedLastTurn, toolBlockedLastTurn) { if applyPatrolFindingLifecycleContinuationRequest(&req, a.currentExecutionProfile(), tools) { patrolFindingContinuationTurn = true @@ -1112,7 +1129,7 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me Str("session_id", sessionID). Msg("[AgenticLoop] Tool calls blocked last turn — omitting tools for final response") } - if isPatrolInvestigationExecution(a.currentExecutionProfile()) { + if isPatrolInvestigationExecution(a.currentExecutionProfile()) && !investigationOutputLimitRecoveryTurn { switch { case investigationProposalCompleted: req.Tools = nil @@ -1417,7 +1434,14 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me err = nil break } - if attempt < maxProviderAttempts && !attemptEmittedVisibleEvents && isRetryableProviderStreamError(effectiveErr) && ctx.Err() == nil { + // Interactive output cannot be retracted once shown, so preserve the + // historical no-replay rule after visible events there. A Patrol run is + // non-interactive and persists nothing until a terminal marker arrives; + // replaying one incomplete stream is therefore safe even if the provider + // leaked a partial fragment before closing. + canReplayIncompleteNonInteractiveTurn := a.currentExecutionProfile().NonInteractive() && + strings.Contains(strings.ToLower(effectiveErr.Error()), "stream ended before completion marker") + if attempt < maxProviderAttempts && (!attemptEmittedVisibleEvents || canReplayIncompleteNonInteractiveTurn) && isRetryableProviderStreamError(effectiveErr) && ctx.Err() == nil { backoff := time.Duration(200*attempt) * time.Millisecond log.Warn(). Int("attempt", attempt). @@ -1595,6 +1619,26 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me continue } + if isPatrolInvestigationExecution(a.currentExecutionProfile()) && isProviderOutputLimitStopReason(stopReason) { + // Tool results already in provider context are the authoritative + // investigation evidence. Do not persist or replay a truncated + // conclusion as if it were evidence, and do not restart diagnosis; + // give the model one tool-free synthesis turn. + if len(resultMessages) > 0 { + resultMessages[len(resultMessages)-1].Content = "" + } + if len(providerMessages) > 0 { + providerMessages[len(providerMessages)-1].Content = "" + providerMessages[len(providerMessages)-1].ReasoningContent = "" + } + if investigationOutputLimitRecoveryAttempted { + return resultMessages, fmt.Errorf("Patrol investigation exhausted its output budget twice before completing its conclusion (last stop reason %q)", stopReason) + } + investigationOutputLimitRecoveryPending = true + turn++ + continue + } + // === FSM ENFORCEMENT GATE 2: Check if final answer is allowed === a.mu.Lock() fsm := a.sessionFSM diff --git a/internal/ai/chat/agentic_additional_test.go b/internal/ai/chat/agentic_additional_test.go index 9c262dc1e..87b9c6771 100644 --- a/internal/ai/chat/agentic_additional_test.go +++ b/internal/ai/chat/agentic_additional_test.go @@ -1876,6 +1876,133 @@ func TestAgenticLoop_DoesNotRetryAfterPartialEvents(t *testing.T) { } } +func TestAgenticLoop_RetriesIncompleteNonInteractiveStreamAfterPartialEvent(t *testing.T) { + provider := &stubStreamingProvider{} + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + executor.ApplyExecutionProfile(tools.ProfilePatrolInvestigation) + loop := NewAgenticLoop(provider, executor, "investigation prompt") + loop.SetExecutionProfile(tools.ProfilePatrolInvestigation) + + callCount := 0 + provider.chatStream = func(_ context.Context, _ providers.ChatRequest, callback providers.StreamCallback) error { + callCount++ + if callCount == 1 { + callback(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "partial provider fragment"}}) + return errors.New("stream ended before completion marker") + } + callback(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "complete investigation"}}) + callback(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{StopReason: "end_turn"}}) + return nil + } + + var retries int + result, err := loop.ExecuteWithTools(context.Background(), "retry-incomplete-investigation", []Message{{Role: "user", Content: "investigate"}}, nil, func(event StreamEvent) { + if event.Type == "workflow_state" { + var state WorkflowStateData + if json.Unmarshal(event.Data, &state) == nil && state.Phase == "provider_retry" { + retries++ + } + } + }) + if err != nil { + t.Fatalf("expected incomplete non-interactive stream replay to recover, got %v", err) + } + if callCount != 2 || retries != 1 { + t.Fatalf("provider calls=%d retries=%d, want 2 calls and one retry", callCount, retries) + } + if len(result) != 1 || result[0].Content != "complete investigation" { + t.Fatalf("durable result = %+v, want only completed replay", result) + } +} + +func TestAgenticLoop_RecoversTruncatedInvestigationConclusionFromExistingEvidence(t *testing.T) { + provider := &stubStreamingProvider{} + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + executor.ApplyExecutionProfile(tools.ProfilePatrolInvestigation) + loop := NewAgenticLoop(provider, executor, "full investigation prompt") + loop.SetExecutionProfile(tools.ProfilePatrolInvestigation) + loop.SetMaxTurns(1) + + providerCalls := 0 + var recoveryRequest providers.ChatRequest + provider.chatStream = func(_ context.Context, req providers.ChatRequest, callback providers.StreamCallback) error { + providerCalls++ + switch providerCalls { + case 1: + callback(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "unfinished reasoning"}}) + callback(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{StopReason: "length", OutputTokens: 4096}}) + case 2: + recoveryRequest = req + callback(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "### Investigation Summary\nClient is unhealthy.\n\n### Root Cause\nDependency is stopped.\n\n### Affected Resources\nClient.\n\n### Recommendation\nStart dependency.\n\n### Conclusion\nNEEDS_ATTENTION: approval required"}}) + callback(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{StopReason: "end_turn", OutputTokens: 120}}) + default: + t.Fatalf("unexpected provider call %d", providerCalls) + } + return nil + } + + result, err := loop.ExecuteWithTools(context.Background(), "truncated-investigation", []Message{{Role: "user", Content: "investigate the unhealthy client"}}, []providers.Tool{{Name: agentcapabilities.PulseQueryToolName}}, func(StreamEvent) {}) + if err != nil { + t.Fatalf("investigation recovery failed: %v", err) + } + if providerCalls != 2 { + t.Fatalf("provider calls=%d, want one bounded recovery", providerCalls) + } + if len(recoveryRequest.Tools) != 0 || recoveryRequest.ToolChoice != nil { + t.Fatalf("recovery request retained tools: %+v", recoveryRequest) + } + if recoveryRequest.System != investigationOutputLimitRecoverySystemPrompt { + t.Fatalf("recovery system prompt = %q", recoveryRequest.System) + } + if recoveryRequest.MaxTokens != investigationOutputLimitRecoveryAllowance || recoveryRequest.ReasoningEffort != providers.ReasoningEffortLow { + t.Fatalf("recovery allowance = %+v", recoveryRequest) + } + for _, message := range recoveryRequest.Messages { + if strings.Contains(message.Content, "unfinished reasoning") || strings.Contains(message.ReasoningContent, "unfinished reasoning") { + t.Fatalf("recovery request retained truncated conclusion: %+v", recoveryRequest.Messages) + } + } + var persisted strings.Builder + for _, message := range result { + if message.Role == "assistant" { + persisted.WriteString(message.Content) + } + } + if strings.Contains(persisted.String(), "unfinished reasoning") || !strings.Contains(persisted.String(), "Dependency is stopped") { + t.Fatalf("persisted investigation = %q", persisted.String()) + } +} + +func TestAgenticLoop_FailsClosedWhenInvestigationRecoveryAlsoTruncates(t *testing.T) { + provider := &stubStreamingProvider{} + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + executor.ApplyExecutionProfile(tools.ProfilePatrolInvestigation) + loop := NewAgenticLoop(provider, executor, "full investigation prompt") + loop.SetExecutionProfile(tools.ProfilePatrolInvestigation) + loop.SetMaxTurns(1) + + providerCalls := 0 + provider.chatStream = func(_ context.Context, _ providers.ChatRequest, callback providers.StreamCallback) error { + providerCalls++ + callback(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "still incomplete"}}) + callback(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{StopReason: "length", OutputTokens: 4096}}) + return nil + } + + result, err := loop.ExecuteWithTools(context.Background(), "twice-truncated-investigation", []Message{{Role: "user", Content: "investigate"}}, nil, func(StreamEvent) {}) + if err == nil || !strings.Contains(err.Error(), "exhausted its output budget twice") { + t.Fatalf("error = %v, want bounded fail-closed error", err) + } + if providerCalls != 2 { + t.Fatalf("provider calls=%d, want one original and one recovery turn", providerCalls) + } + for _, message := range result { + if strings.TrimSpace(message.Content) != "" { + t.Fatalf("truncated conclusion leaked into durable result: %+v", result) + } + } +} + func TestAgenticLoop_EmitsFallbackErrorEventOnTransportFailure(t *testing.T) { provider := &stubStreamingProvider{} executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) diff --git a/internal/ai/chat/agentic_investigation_budget.go b/internal/ai/chat/agentic_investigation_budget.go index e2ae6ab0d..19d482aa9 100644 --- a/internal/ai/chat/agentic_investigation_budget.go +++ b/internal/ai/chat/agentic_investigation_budget.go @@ -22,6 +22,22 @@ const investigationEvidenceBudgetExhaustedSystemPrompt = ` INVESTIGATION COMPLETION: The evidence-call budget is exhausted. No more evidence tools are available. Use the evidence already collected. If it supports a safe advertised remediation and no proposal has been submitted, you may call patrol_propose_action once; otherwise produce the required final summary and state any remaining uncertainty.` +const investigationOutputLimitRecoverySystemPrompt = `You are Pulse Patrol completing an investigation after the previous final response exhausted its output budget. Do not call tools, repeat the investigation, or narrate your reasoning. Synthesize only the evidence already present in the conversation into the required five sections: Investigation Summary, Root Cause, Affected Resources, Recommendation, and Conclusion. Name causal and affected resources with their exact observed canonical name or ID. If the evidence does not establish root cause, say exactly what remains uncertain. Never invent evidence, actions, verification, or remediation.` + +const investigationOutputLimitRecoveryAllowance = 4_096 + +func applyInvestigationOutputLimitRecoveryRequest(req *providers.ChatRequest, profile aitools.ExecutionProfile) bool { + if req == nil || !isPatrolInvestigationExecution(profile) { + return false + } + req.Tools = nil + req.ToolChoice = nil + req.System = investigationOutputLimitRecoverySystemPrompt + req.MaxTokens = investigationOutputLimitRecoveryAllowance + req.ReasoningEffort = providers.ReasoningEffortLow + return true +} + func isInvestigationEvidenceTool(name string) bool { return strings.TrimSpace(name) != agentcapabilities.PatrolProposeActionToolName } diff --git a/internal/ai/cost/pricing.go b/internal/ai/cost/pricing.go index f15814c12..a6ba85602 100644 --- a/internal/ai/cost/pricing.go +++ b/internal/ai/cost/pricing.go @@ -71,6 +71,7 @@ var providerPrices = map[string][]modelPrice{ flatPriceAsOf("deepseek/deepseek-v4-flash", 0.09, 0.18, "2026-07-14"), flatPriceAsOf("nvidia/nemotron-3.5-lightning:free", 0, 0, "2026-08-14"), flatPriceAsOf("nvidia/nemotron-3-super-120b-a12b:free", 0, 0, "2026-08-14"), + flatPriceAsOf("nvidia/nemotron-3-ultra-550b-a55b:free", 0, 0, "2026-08-15"), }, "gemini": { // Gemini Developer API standard paid-tier pricing, checked from diff --git a/internal/ai/cost/store_extended_test.go b/internal/ai/cost/store_extended_test.go index db7e54d21..937c23358 100644 --- a/internal/ai/cost/store_extended_test.go +++ b/internal/ai/cost/store_extended_test.go @@ -138,9 +138,10 @@ func TestEstimateUSD_OllamaFree(t *testing.T) { } func TestEstimateUSD_ReviewedOpenRouterFreeRoute(t *testing.T) { - for _, model := range []string{ - "nvidia/nemotron-3.5-lightning:free", - "nvidia/nemotron-3-super-120b-a12b:free", + for model, wantAsOf := range map[string]string{ + "nvidia/nemotron-3.5-lightning:free": "2026-08-14", + "nvidia/nemotron-3-super-120b-a12b:free": "2026-08-14", + "nvidia/nemotron-3-ultra-550b-a55b:free": "2026-08-15", } { usd, ok, price := EstimateUSD("openrouter", model, 1_000_000, 1_000_000) if !ok { @@ -149,8 +150,8 @@ func TestEstimateUSD_ReviewedOpenRouterFreeRoute(t *testing.T) { if usd != 0 || price.InputUSDPerMTok != 0 || price.OutputUSDPerMTok != 0 { t.Fatalf("reviewed OpenRouter free route %s should estimate $0, got usd=%f price=%+v", model, usd, price) } - if price.AsOf != "2026-08-14" { - t.Fatalf("reviewed OpenRouter free-route date for %s = %q, want 2026-08-14", model, price.AsOf) + if price.AsOf != wantAsOf { + t.Fatalf("reviewed OpenRouter free-route date for %s = %q, want %s", model, price.AsOf, wantAsOf) } }