Recover interrupted Patrol investigations

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

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

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{})

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
 }

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

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)
 		}
 	}