| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "math/rand" |
| 8 | "strings" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/event" |
| 12 | "reasonix/internal/evidence" |
| 13 | "reasonix/internal/jobs" |
| 14 | "reasonix/internal/provider" |
| 15 | "reasonix/internal/tool" |
| 16 | ) |
| 17 | |
| 18 | // runLoopState holds per-Run loop counters and flags. It is package-private and |
| 19 | // not shared across goroutines; the first extraction keeps the existing lock |
| 20 | // model and only structures the sequential turn state machine. |
| 21 | type runLoopState struct { |
| 22 | runMaxSteps int |
| 23 | runMaxStepsKey string |
| 24 | runLimitHostOwned bool |
| 25 | |
| 26 | emptyFinalBlocks int |
| 27 | handoffNudges int |
| 28 | usedAnyTool bool |
| 29 | goalToolRepairs int |
| 30 | graceRound bool |
| 31 | recoveryGraceRound bool |
| 32 | |
| 33 | todoProgress int |
| 34 | trackingTodoProgress bool |
| 35 | todoStallRounds int |
| 36 | seenTodoProgress map[string]struct{} |
| 37 | |
| 38 | executorHandoff bool |
| 39 | // input is the user turn text after withTurnPreferences (used by handoff |
| 40 | // nudges that inspect the original request wording). |
| 41 | input string |
| 42 | |
| 43 | workDurationMs func() int64 |
| 44 | } |
| 45 | |
| 46 | // streamedTurn is one provider completion collected by stream. Keeping the |
| 47 | // result together makes the missing-reasoning recovery path explicit: the |
| 48 | // first, malformed completion is never committed before a safe replacement is |
| 49 | // available, and a failed recovery can still fall back to the complete first |
| 50 | // response without re-running any tool. |
| 51 | type streamedTurn struct { |
| 52 | text string |
| 53 | reasoning string |
| 54 | signature string |
| 55 | reasoningID string |
| 56 | reasoningStatus string |
| 57 | calls []provider.ToolCall |
| 58 | responsesItems []json.RawMessage |
| 59 | usage *provider.Usage |
| 60 | interrupted bool |
| 61 | partialToolStarted bool |
| 62 | partialCalls []provider.ToolCall |
| 63 | maxArgChars int // peak streaming tool-arg size for failed-attempt estimates |
| 64 | err error |
| 65 | } |
| 66 | |
| 67 | // deferredStreamSink keeps selected stream events local until the caller |
| 68 | // chooses which provider response to adopt. On an ordinary healthy DeepSeek |
| 69 | // turn, reasoning arrives before tool calls and unlocks live tool-card events. |
| 70 | // On the rare malformed turn with no reasoning, only the speculative partial |
| 71 | // tool cards remain buffered, so retrying does not flash duplicate cards in the |
| 72 | // UI. A recovery attempt buffers everything because it may be discarded. |
| 73 | type deferredStreamSink struct { |
| 74 | inner event.Sink |
| 75 | deferAll bool |
| 76 | waitingForReasoning bool |
| 77 | sawReasoning bool |
| 78 | events []event.Event |
| 79 | } |
| 80 | |
| 81 | func newReasoningAwareStreamSink(inner event.Sink) *deferredStreamSink { |
| 82 | return &deferredStreamSink{inner: inner, waitingForReasoning: true} |
| 83 | } |
| 84 | |
| 85 | func newDeferredStreamSink(inner event.Sink) *deferredStreamSink { |
| 86 | return &deferredStreamSink{inner: inner, deferAll: true} |
| 87 | } |
| 88 | |
| 89 | func (s *deferredStreamSink) Emit(e event.Event) { |
| 90 | if s == nil { |
| 91 | return |
| 92 | } |
| 93 | if s.deferAll { |
| 94 | s.events = append(s.events, e) |
| 95 | return |
| 96 | } |
| 97 | if s.waitingForReasoning && e.Kind == event.Reasoning && strings.TrimSpace(e.Text) != "" { |
| 98 | s.sawReasoning = true |
| 99 | s.inner.Emit(e) |
| 100 | s.flushBuffered() |
| 101 | return |
| 102 | } |
| 103 | if s.waitingForReasoning && !s.sawReasoning && e.Kind == event.ToolDispatch { |
| 104 | s.events = append(s.events, e) |
| 105 | return |
| 106 | } |
| 107 | s.inner.Emit(e) |
| 108 | } |
| 109 | |
| 110 | func (s *deferredStreamSink) flushBuffered() { |
| 111 | if s == nil { |
| 112 | return |
| 113 | } |
| 114 | for _, e := range s.events { |
| 115 | s.inner.Emit(e) |
| 116 | } |
| 117 | s.events = nil |
| 118 | } |
| 119 | |
| 120 | func (s *deferredStreamSink) Flush() { |
| 121 | if s == nil { |
| 122 | return |
| 123 | } |
| 124 | s.flushBuffered() |
| 125 | } |
| 126 | |
| 127 | func (s *deferredStreamSink) Discard() { |
| 128 | if s != nil { |
| 129 | s.events = nil |
| 130 | } |
| 131 | } |
| 132 | |
| 133 | // beginRunTurn handles evidence scope, delivery classification, background-job |
| 134 | // evidence re-lease, and the initial user-turn persistence. Callers still own |
| 135 | // all Run-level defers (workspace lease, evidence commit, delivery checkpoint, |
| 136 | // steer queue, active-turn timestamp). |
| 137 | func (a *Agent) beginRunTurn(ctx context.Context, input string) (rawInput string, state *runLoopState) { |
| 138 | rawInput = RawUserInput(ctx, input) |
| 139 | providerInput := input |
| 140 | scope, scoped := DeliveryExecutionScopeFromContext(ctx) |
| 141 | preserveEvidence := a.preserveEvidenceOnce |
| 142 | // A run that starts with a pending readiness recovery (or an explicit |
| 143 | // evidence-preserving continuation) and then passes readiness counts as a |
| 144 | // recovery in the final audit. |
| 145 | a.readinessRecovered = preserveEvidence || a.deliveryRecoveryPending |
| 146 | if a.evidence != nil { |
| 147 | switch { |
| 148 | case preserveEvidence: |
| 149 | a.evidence.ResetBackgroundLeases() |
| 150 | case scoped && a.deliveryScopeID == scope.ID: |
| 151 | a.evidence.ResetBackgroundLeases() |
| 152 | default: |
| 153 | a.evidence.Reset() |
| 154 | } |
| 155 | } |
| 156 | a.preserveEvidenceOnce = false |
| 157 | if !preserveEvidence { |
| 158 | a.deliveryRecoveryPending = false |
| 159 | } |
| 160 | if scoped { |
| 161 | a.deliveryScopeID = scope.ID |
| 162 | } else if !preserveEvidence { |
| 163 | a.deliveryScopeID = "" |
| 164 | } |
| 165 | a.deliveryScopeActive = scoped |
| 166 | if scoped && a.deliveryCheckpoint.ScopeID != scope.ID { |
| 167 | a.deliveryCheckpoint = evidence.DeliveryCheckpoint{ScopeID: scope.ID} |
| 168 | } |
| 169 | // Re-lease this session's background-job mutations that no turn has |
| 170 | // committed yet. The Reset above just wiped any lease a failed or |
| 171 | // cancelled turn held (its ledger is gone), and a process restart starts |
| 172 | // from an empty ledger too — in both cases the job manager still marks the |
| 173 | // job's evidence uncommitted. Without re-injecting it here, a turn that |
| 174 | // never re-issues wait/bash_output (the model has no reason to if it |
| 175 | // doesn't know a mutation is still pending) would ship the background |
| 176 | // change without the final-readiness gate ever seeing it. Plan turns defer |
| 177 | // this lease like collectBackgroundEvidence does so execution evidence is |
| 178 | // consumed and audited only after plan approval. |
| 179 | if a.evidence != nil && a.jobs != nil && !a.planMode.Load() { |
| 180 | session := jobs.SessionFromContext(ctx) |
| 181 | for _, jobID := range a.jobs.PendingEvidenceJobIDsForSession(session) { |
| 182 | summary, ready := a.jobs.TryLeaseEvidenceForSession(session, jobID) |
| 183 | if !ready { |
| 184 | continue |
| 185 | } |
| 186 | if !a.evidence.NoteBackgroundLease(session, jobID) { |
| 187 | continue |
| 188 | } |
| 189 | a.evidence.MergeChild(summary) |
| 190 | } |
| 191 | } |
| 192 | a.deliveryCriteriaEstablished = a.hasIncompleteCanonicalCriteria() || |
| 193 | (a.evidence != nil && a.evidence.HasSuccessfulTodoWrite()) || |
| 194 | (scoped && a.deliveryCheckpoint.CriteriaEstablished) |
| 195 | // Classify delivery expectations from the task text. Sub-agent spawners |
| 196 | // pass the pristine task through Options.ClassifierTaskText (a trusted |
| 197 | // host channel) because their Run input carries host framing whose |
| 198 | // incidental verbs — "file tools resolve relative paths" — once classified |
| 199 | // every workspace-wrapped subagent prompt as a mutation request and |
| 200 | // deadlocked read-only subagents. Without the override the raw input is |
| 201 | // classified verbatim: stripping user-controllable markup here would let |
| 202 | // input dressed up as host framing disarm the delivery gates. |
| 203 | classifierInput := a.classifierTaskText |
| 204 | if scoped && strings.TrimSpace(scope.TaskText) != "" { |
| 205 | classifierInput = scope.TaskText |
| 206 | } else if strings.TrimSpace(classifierInput) == "" { |
| 207 | classifierInput = rawInput |
| 208 | } |
| 209 | intent := classifyDeliveryTaskIntent(classifierInput) |
| 210 | a.deliveryTaskExpected = intent == deliveryIntentObservableRead || intent == deliveryIntentMutation || intent == deliveryIntentPersistentAction |
| 211 | a.deliveryMutationExpected = intent == deliveryIntentMutation && registryHasWriterTools(a.tools) |
| 212 | a.deliveryPersistentExpected = deliveryTaskNeedsPersistentAction(classifierInput) |
| 213 | a.recoveryTaskSummary = boundedRecoveryTaskSummary(classifierInput) |
| 214 | // A cancelled/error turn leaves a provider-excluded recovery record at the |
| 215 | // transcript tail. Fold its bounded facts into this new user turn exactly |
| 216 | // once; the user's raw text remains the classifier source above. |
| 217 | providerInput = withInterruptedRecovery(providerInput, a.pendingInterruptedRecovery()) |
| 218 | a.repeatSuccessCounts = nil |
| 219 | if !scoped || a.repeatFailureScope != scope.ID { |
| 220 | a.repeatFailureCounts = nil |
| 221 | } else { |
| 222 | // Only stale-anchor failures have a side-effect-free state recheck. |
| 223 | // Ordinary write failures may recover between Runs after user action or |
| 224 | // an external state change, so do not carry their retry budget forward. |
| 225 | for sig, failure := range a.repeatFailureCounts { |
| 226 | if !failure.stateRecheck { |
| 227 | delete(a.repeatFailureCounts, sig) |
| 228 | } |
| 229 | } |
| 230 | } |
| 231 | if scoped { |
| 232 | a.repeatFailureScope = scope.ID |
| 233 | } else { |
| 234 | a.repeatFailureScope = "" |
| 235 | } |
| 236 | a.blockedTurnStreak = 0 |
| 237 | a.loopGuardArmed = false |
| 238 | a.loopGuardReceiptMark = 0 |
| 239 | a.sink.Emit(event.Event{Kind: event.TurnStarted}) |
| 240 | input = a.withTurnPreferences(providerInput) |
| 241 | userCreatedAt := time.Now().UnixMilli() |
| 242 | a.activeTurnCreatedAt.Store(userCreatedAt) |
| 243 | rawContent := "" |
| 244 | if input != rawInput { |
| 245 | rawContent = rawInput |
| 246 | } |
| 247 | a.session.Add(provider.Message{ |
| 248 | Role: provider.RoleUser, Content: input, RawContent: rawContent, |
| 249 | Images: userImages(ctx), CreatedAt: userCreatedAt, |
| 250 | }) |
| 251 | |
| 252 | state = &runLoopState{ |
| 253 | emptyFinalBlocks: 0, |
| 254 | handoffNudges: 0, |
| 255 | usedAnyTool: false, |
| 256 | graceRound: false, |
| 257 | recoveryGraceRound: false, |
| 258 | todoStallRounds: 0, |
| 259 | seenTodoProgress: make(map[string]struct{}), |
| 260 | executorHandoff: a.executorHandoffGuard && strings.Contains(input, executorHandoffMarker), |
| 261 | input: input, |
| 262 | } |
| 263 | state.todoProgress, state.trackingTodoProgress = a.canonicalTodoProgress() |
| 264 | if a.evidence != nil { |
| 265 | for _, sig := range a.evidence.SuccessfulProgressSignaturesSince(0) { |
| 266 | state.seenTodoProgress[sig] = struct{}{} |
| 267 | } |
| 268 | } |
| 269 | return rawInput, state |
| 270 | } |
| 271 | |
| 272 | // runToolLoop owns the main tool-round budget and dispatches each streamed |
| 273 | // assistant turn into final-response or tool-round handling. |
| 274 | func (a *Agent) runToolLoop(ctx context.Context, state *runLoopState) error { |
| 275 | for step := 0; state.runMaxSteps <= 0 || step < state.runMaxSteps || state.graceRound || state.recoveryGraceRound; step++ { |
| 276 | // Consume a queued steer and persist it to the session so it |
| 277 | // survives tab switches and history replay. The model sees it as |
| 278 | // guidance (with a prefix), not a new task. One cache miss per |
| 279 | // steer is unavoidable — the model must see the new instruction. |
| 280 | if text, ok := a.consumeSteer(); ok { |
| 281 | a.session.Add(provider.Message{Role: provider.RoleUser, Content: a.withTurnPreferences(midTurnSteerMessage(text))}) |
| 282 | a.sink.Emit(event.Event{Kind: event.Steer, Text: text}) |
| 283 | } |
| 284 | schemas := a.tools.Schemas() |
| 285 | prefixShape := a.capturePrefixShape(schemas) |
| 286 | prevPrefixShape := a.lastPrefixShape |
| 287 | if !a.haveLastPrefixShape { |
| 288 | prevPrefixShape = prefixShape |
| 289 | } |
| 290 | |
| 291 | // Drain reasons queued since the previous capture (compaction, |
| 292 | // snip/prune, rewind, guardian merge) so CompareShape can attribute |
| 293 | // any prefix change to the operation that actually caused it, instead |
| 294 | // of a generic rewrite signal that also fires on local-only metadata |
| 295 | // edits. |
| 296 | contentReasons := a.session.DrainContentRewriteReasons() |
| 297 | |
| 298 | // Prefix shape is captured once before sampling and frozen for the |
| 299 | // whole attempt lifecycle — stream retries must not rewrite session |
| 300 | // history mid-round, so the shape stays stable across body replays. |
| 301 | streamed := a.streamWithSamplingRecovery(ctx, step+1) |
| 302 | text, reasoning, signature, calls, responsesItems, usage := streamed.text, streamed.reasoning, streamed.signature, streamed.calls, streamed.responsesItems, streamed.usage |
| 303 | partialCalls, err := streamed.partialCalls, streamed.err |
| 304 | cacheDiagnostics := CompareShape(prevPrefixShape, prefixShape, usage, contentReasons) |
| 305 | if err != nil { |
| 306 | a.emitTurnUsage(usage, &cacheDiagnostics) |
| 307 | if msg, ok := finishReasonMessage(usage); ok { |
| 308 | a.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: msg}) |
| 309 | } |
| 310 | // Exhausted stream retries (or a non-retryable error): persist one |
| 311 | // bounded LocalOnly recovery record for the next real user message. |
| 312 | // Intermediate failed attempts never wrote session state. |
| 313 | a.recordInterruptedDisplay(text, reasoning, partialCalls, true, state.workDurationMs()) |
| 314 | return err |
| 315 | } |
| 316 | a.lastPrefixShape = prefixShape |
| 317 | a.haveLastPrefixShape = true |
| 318 | a.emitTurnUsage(usage, &cacheDiagnostics) |
| 319 | if msg, ok := finishReasonMessage(usage); ok { |
| 320 | a.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: msg}) |
| 321 | } |
| 322 | |
| 323 | // Commit boundary: only a clean terminal attempt reaches here. |
| 324 | // Keep reasoning_content on the assistant turn for display and session |
| 325 | // archive. Most OpenAI-compatible backends do not replay it; providers |
| 326 | // with an explicit round-trip contract retain the raw provider text. |
| 327 | calls = a.withPreviewFileDiffs(calls) |
| 328 | a.session.Add(provider.Message{ |
| 329 | Role: provider.RoleAssistant, |
| 330 | Content: text, |
| 331 | ReasoningContent: reasoning, |
| 332 | ReasoningSignature: signature, |
| 333 | ReasoningID: streamed.reasoningID, |
| 334 | ReasoningStatus: streamed.reasoningStatus, |
| 335 | ToolCalls: calls, |
| 336 | ResponsesItems: responsesItems, |
| 337 | WorkDurationMs: state.workDurationMs(), |
| 338 | }) |
| 339 | |
| 340 | if len(calls) == 0 { |
| 341 | cont, ferr := a.handleFinalResponse(ctx, state, text, reasoning, usage) |
| 342 | if !cont { |
| 343 | return ferr |
| 344 | } |
| 345 | continue |
| 346 | } |
| 347 | |
| 348 | // Invariant: executeBatch only ever receives tool calls from a |
| 349 | // committed sampling attempt (clean terminal + response intercept). |
| 350 | cont, terr := a.handleToolRound(ctx, state, step, text, reasoning, calls, usage) |
| 351 | if !cont { |
| 352 | return terr |
| 353 | } |
| 354 | } |
| 355 | // Only reached when a positive maxSteps guard is configured. The work so far |
| 356 | // is already in the session, so the user can just send another message to pick |
| 357 | // up where it left off. |
| 358 | return &maxStepsPause{steps: state.runMaxSteps, key: state.runMaxStepsKey} |
| 359 | } |
| 360 | |
| 361 | // streamWithSamplingRecovery coordinates Codex-style original-request replay |
| 362 | // for one model round: prepare once, freeze the provider request, run up to |
| 363 | // maxSamplingAttempts body attempts, and only commit after a clean terminal. |
| 364 | // Failed attempts never write Session state or execute tools. missing-reasoning |
| 365 | // repair shares this lifecycle (at most one extra exact replay). |
| 366 | func (a *Agent) streamWithSamplingRecovery(ctx context.Context, turn int) streamedTurn { |
| 367 | frozen, err := a.prepareSamplingRequest(ctx) |
| 368 | if err != nil { |
| 369 | return streamedTurn{err: err} |
| 370 | } |
| 371 | // One request counter spans every body attempt; each attempt records only |
| 372 | // its delta so RequestCount equals real HTTP POSTs (no triangular growth). |
| 373 | ctx = provider.WithRequestAttemptCounter(ctx) |
| 374 | |
| 375 | var billable *provider.Usage |
| 376 | var last streamedTurn |
| 377 | |
| 378 | runAttempt := func(attemptID string, sink event.Sink) streamedTurn { |
| 379 | before := provider.RequestAttemptCount(ctx) |
| 380 | result := a.streamTurnFrozen(ctx, turn, sink, &frozen, attemptID) |
| 381 | after := provider.RequestAttemptCount(ctx) |
| 382 | delta := after - before |
| 383 | if delta < 0 { |
| 384 | delta = 0 |
| 385 | } |
| 386 | // httpRequests=0 means the provider does not use SendWithRetry |
| 387 | // (extension/custom), or it failed before issuing an HTTP request. |
| 388 | // Only overwrite RequestCount when the built-in counter observed POSTs; |
| 389 | // otherwise keep the provider-reported count (zero still means one via |
| 390 | // usageRequestCount compatibility). estimateFailedAttemptUsage returns nil |
| 391 | // for zero-output local failures so no invented request appears. |
| 392 | result.usage = estimateFailedAttemptUsage(result.usage, frozen, result, delta) |
| 393 | if result.usage != nil { |
| 394 | if delta > 0 { |
| 395 | result.usage.RequestCount = delta |
| 396 | } |
| 397 | } else if delta > 0 { |
| 398 | result.usage = &provider.Usage{RequestCount: delta} |
| 399 | } |
| 400 | return result |
| 401 | } |
| 402 | |
| 403 | for attempt := 1; attempt <= maxSamplingAttempts; attempt++ { |
| 404 | attemptID := newStreamAttemptID(attempt) |
| 405 | a.emitStreamAttempt(attemptID, event.StreamAttemptBegin, attempt, "", nil) |
| 406 | |
| 407 | var streamSink *deferredStreamSink |
| 408 | attemptSink := a.sink |
| 409 | if provider.WarnOnMissingToolCallReasoning(a.prov) { |
| 410 | streamSink = newReasoningAwareStreamSink(a.sink) |
| 411 | attemptSink = streamSink |
| 412 | } |
| 413 | |
| 414 | result := runAttempt(attemptID, attemptSink) |
| 415 | billable = mergeSamplingUsage(billable, result.usage) |
| 416 | // lastUsage is the latest single-request shape (prompt+completion+cache |
| 417 | // for that attempt only). Never the multi-attempt billable aggregate — |
| 418 | // that would inflate ContextSnapshot and compaction decisions. |
| 419 | a.storeLatestRequestUsage(result.usage) |
| 420 | last = result |
| 421 | last.usage = finalizeSamplingUsage(billable, result.usage) |
| 422 | |
| 423 | if result.err != nil { |
| 424 | if provider.IsStreamInterrupted(result.err) && attempt < maxSamplingAttempts { |
| 425 | streamSink.Discard() |
| 426 | reason := provider.StreamInterruptReason(result.err) |
| 427 | a.emitStreamAttempt(attemptID, event.StreamAttemptDiscard, attempt, reason, result.err) |
| 428 | a.sink.Emit(event.Event{ |
| 429 | Kind: event.Retrying, RetryAttempt: attempt, RetryMax: maxStreamRecoveries, |
| 430 | RetryScope: event.RetryScopeStream, |
| 431 | }) |
| 432 | if !streamRetrySleep(ctx, attempt) { |
| 433 | return streamedTurn{usage: finalizeSamplingUsage(billable, result.usage), interrupted: true, err: ctx.Err()} |
| 434 | } |
| 435 | continue |
| 436 | } |
| 437 | // Exhausted retries or non-retryable error: leave the last |
| 438 | // speculative UI visible (no discard) so LocalOnly can mirror it. |
| 439 | streamSink.Flush() |
| 440 | last.usage = finalizeSamplingUsage(billable, result.usage) |
| 441 | return last |
| 442 | } |
| 443 | |
| 444 | // Clean terminal. Optionally repair missing reasoning with one extra |
| 445 | // exact replay of the same frozen request (no synthetic prompt). |
| 446 | missing, shouldRetry := a.observeMissingToolCallReasoning(result.calls, result.reasoning) |
| 447 | if missing { |
| 448 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningDetected}) |
| 449 | if shouldRetry && strings.TrimSpace(result.text) == "" { |
| 450 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningRetryAttempted}) |
| 451 | retrySink := newDeferredStreamSink(a.sink) |
| 452 | retry := runAttempt(attemptID, retrySink) |
| 453 | billable = mergeSamplingUsage(billable, retry.usage) |
| 454 | if retry.err != nil { |
| 455 | retrySink.Discard() |
| 456 | if ctx.Err() != nil { |
| 457 | streamSink.Discard() |
| 458 | a.emitStreamAttempt(attemptID, event.StreamAttemptDiscard, attempt, provider.StreamInterruptReason(retry.err), retry.err) |
| 459 | // Use the cancelled retry as the "latest" shape so |
| 460 | // FinishReason=interrupted is preserved for accounting. |
| 461 | return streamedTurn{usage: finalizeSamplingUsage(billable, retry.usage), err: retry.err} |
| 462 | } |
| 463 | // Fall back to the first complete response; no tool ran. |
| 464 | streamSink.Flush() |
| 465 | a.storeLatestRequestUsage(result.usage) |
| 466 | result.usage = finalizeSamplingUsage(billable, result.usage) |
| 467 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningFallback}) |
| 468 | a.emitStreamAttempt(attemptID, event.StreamAttemptCommit, attempt, "", nil) |
| 469 | return result |
| 470 | } |
| 471 | streamSink.Discard() |
| 472 | retrySink.Flush() |
| 473 | a.storeLatestRequestUsage(retry.usage) |
| 474 | retry.usage = finalizeSamplingUsage(billable, retry.usage) |
| 475 | retryMissing, _ := a.observeMissingToolCallReasoning(retry.calls, retry.reasoning) |
| 476 | if retryMissing { |
| 477 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningDetected}) |
| 478 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningFallback}) |
| 479 | } else if len(retry.calls) == 0 { |
| 480 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningRetryReplaced}) |
| 481 | } else { |
| 482 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningRetryRecovered}) |
| 483 | } |
| 484 | a.emitStreamAttempt(attemptID, event.StreamAttemptCommit, attempt, "", nil) |
| 485 | return retry |
| 486 | } |
| 487 | if !shouldRetry || strings.TrimSpace(result.text) != "" { |
| 488 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningRetrySuppressed}) |
| 489 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningFallback}) |
| 490 | } else { |
| 491 | event.RecordProtocolRecovery(a.sink, event.ProtocolRecoveryAudit{Kind: event.ProtocolRecoveryMissingReasoningFallback}) |
| 492 | } |
| 493 | } |
| 494 | |
| 495 | streamSink.Flush() |
| 496 | a.emitStreamAttempt(attemptID, event.StreamAttemptCommit, attempt, "", nil) |
| 497 | result.usage = finalizeSamplingUsage(billable, result.usage) |
| 498 | return result |
| 499 | } |
| 500 | return last |
| 501 | } |
| 502 | |
| 503 | func (a *Agent) streamTurnFrozen(ctx context.Context, turn int, sink event.Sink, frozen *samplingRequest, attemptID string) streamedTurn { |
| 504 | text, reasoning, signature, reasoningID, reasoningStatus, calls, responsesItems, usage, interrupted, partialToolStarted, partialCalls, maxArgChars, err := a.streamWithFrozen(ctx, turn, sink, frozen, attemptID) |
| 505 | return streamedTurn{ |
| 506 | text: text, reasoning: reasoning, signature: signature, |
| 507 | reasoningID: reasoningID, reasoningStatus: reasoningStatus, |
| 508 | calls: calls, responsesItems: responsesItems, usage: usage, |
| 509 | interrupted: interrupted, partialToolStarted: partialToolStarted, partialCalls: partialCalls, |
| 510 | maxArgChars: maxArgChars, err: err, |
| 511 | } |
| 512 | } |
| 513 | |
| 514 | func (a *Agent) emitStreamAttempt(id string, action event.StreamAttemptAction, attempt int, reason string, err error) { |
| 515 | if reason == "" && err != nil { |
| 516 | reason = provider.StreamInterruptReason(err) |
| 517 | } |
| 518 | a.sink.Emit(event.Event{ |
| 519 | Kind: event.StreamAttempt, |
| 520 | StreamAttempt: event.StreamAttemptInfo{ |
| 521 | ID: id, Action: action, Attempt: attempt, Max: maxSamplingAttempts, Reason: reason, |
| 522 | }, |
| 523 | }) |
| 524 | } |
| 525 | |
| 526 | func newStreamAttemptID(attempt int) string { |
| 527 | // Host-local only: never persisted, never sent to the model. |
| 528 | return fmt.Sprintf("sa-%d-%d", attempt, time.Now().UnixNano()) |
| 529 | } |
| 530 | |
| 531 | // streamRetrySleep is the body-retry backoff. Tests replace it with a no-op so |
| 532 | // recovery suites stay fast while production keeps the Codex-shaped delays. |
| 533 | var streamRetrySleep = sleepStreamRetryBackoff |
| 534 | |
| 535 | // sleepStreamRetryBackoff waits ~0.5s, 1s, 2s, 4s, 8s with small jitter. |
| 536 | // Returns false when ctx is cancelled during the wait. |
| 537 | func sleepStreamRetryBackoff(ctx context.Context, attempt int) bool { |
| 538 | // attempt is 1-based for the failed attempt about to be retried. |
| 539 | shift := attempt - 1 |
| 540 | if shift < 0 { |
| 541 | shift = 0 |
| 542 | } |
| 543 | if shift > 4 { |
| 544 | shift = 4 |
| 545 | } |
| 546 | base := time.Duration(1<<shift) * 500 * time.Millisecond |
| 547 | jitter := time.Duration(rand.Intn(250)) * time.Millisecond |
| 548 | timer := time.NewTimer(base + jitter) |
| 549 | defer timer.Stop() |
| 550 | select { |
| 551 | case <-ctx.Done(): |
| 552 | return false |
| 553 | case <-timer.C: |
| 554 | return true |
| 555 | } |
| 556 | } |
| 557 | |
| 558 | // estimateFailedAttemptUsage fills Estimated usage when a body attempt ends |
| 559 | // without a terminal provider usage record, so billing and observational Goal |
| 560 | // usage still include the issued request plus any observed speculative output. |
| 561 | // Non-interrupt failures that already carry usage (e.g. client reasoning limit) |
| 562 | // are left intact. |
| 563 | // |
| 564 | // httpRequests is the SendWithRetry attempt-counter delta for this body attempt. |
| 565 | // When it is 0 and there was no speculative output, the failure was local or |
| 566 | // came from a provider without observable transport accounting; return nil or |
| 567 | // its existing usage rather than inventing billable tokens. |
| 568 | func estimateFailedAttemptUsage(usage *provider.Usage, frozen samplingRequest, result streamedTurn, httpRequests int) *provider.Usage { |
| 569 | if result.err == nil { |
| 570 | return usage |
| 571 | } |
| 572 | // Preserve exact client-side finish reasons that already computed usage. |
| 573 | if usage != nil && usage.FinishReason != "" && usage.FinishReason != "interrupted" { |
| 574 | return usage |
| 575 | } |
| 576 | // A zero-output, non-interrupted failure with no observed HTTP request is a |
| 577 | // local/provider validation failure. It is not a billable sampling attempt. |
| 578 | preBodyLocal := httpRequests <= 0 && !result.interrupted && |
| 579 | !provider.IsStreamInterrupted(result.err) && !sawSpeculativeSamplingOutput(result) |
| 580 | if preBodyLocal { |
| 581 | if usage != nil && usageTotalTokens(usage) > 0 { |
| 582 | return usage |
| 583 | } |
| 584 | return nil |
| 585 | } |
| 586 | if !provider.IsStreamInterrupted(result.err) && !result.interrupted { |
| 587 | // Auth/cancel/decode/limit paths keep their own accounting. |
| 588 | if usage != nil { |
| 589 | return usage |
| 590 | } |
| 591 | if httpRequests <= 0 { |
| 592 | return nil |
| 593 | } |
| 594 | } |
| 595 | textBytes := len(result.text) |
| 596 | reasoningBytes := len(result.reasoning) |
| 597 | maxArg := result.maxArgChars |
| 598 | for _, call := range result.partialCalls { |
| 599 | if n := len(call.Arguments); n > maxArg { |
| 600 | maxArg = n |
| 601 | } |
| 602 | } |
| 603 | for _, call := range result.calls { |
| 604 | if n := len(call.Arguments); n > maxArg { |
| 605 | maxArg = n |
| 606 | } |
| 607 | } |
| 608 | if usage != nil && !usage.Estimated && usage.TotalTokens > 0 { |
| 609 | return usage |
| 610 | } |
| 611 | finish := "interrupted" |
| 612 | if usage != nil && usage.FinishReason != "" { |
| 613 | finish = usage.FinishReason |
| 614 | } |
| 615 | est := bestEffortStreamUsage(usage, textBytes, reasoningBytes, finish) |
| 616 | if est == nil { |
| 617 | est = &provider.Usage{Estimated: true, FinishReason: finish} |
| 618 | } |
| 619 | if est.PromptTokens <= 0 { |
| 620 | est.PromptTokens = estimateSamplingRequestInputTokens(frozen.req) |
| 621 | est.Estimated = true |
| 622 | } |
| 623 | // Estimated failed attempts without cache split still need Cost() to see |
| 624 | // billable input — Price falls back to PromptTokens only when hit+miss=0. |
| 625 | if est.CacheHitTokens+est.CacheMissTokens == 0 && est.PromptTokens > 0 { |
| 626 | est.CacheMissTokens = est.PromptTokens |
| 627 | } |
| 628 | if maxArg > 0 { |
| 629 | argTokens := (maxArg + 3) / 4 |
| 630 | if est.CompletionTokens < argTokens+estimateTokensFromBytes(textBytes)+estimateTokensFromBytes(reasoningBytes) { |
| 631 | est.CompletionTokens = argTokens + estimateTokensFromBytes(textBytes) + estimateTokensFromBytes(reasoningBytes) |
| 632 | est.Estimated = true |
| 633 | } |
| 634 | } |
| 635 | if minTotal := est.PromptTokens + est.CompletionTokens; est.TotalTokens < minTotal { |
| 636 | est.TotalTokens = minTotal |
| 637 | est.Estimated = true |
| 638 | } |
| 639 | return est |
| 640 | } |
| 641 | |
| 642 | func sawSpeculativeSamplingOutput(result streamedTurn) bool { |
| 643 | return result.text != "" || result.reasoning != "" || result.maxArgChars > 0 || |
| 644 | result.partialToolStarted || len(result.calls) > 0 || len(result.partialCalls) > 0 |
| 645 | } |
| 646 | |
| 647 | // estimateSamplingRequestInputTokens reconstructs a conservative input count |
| 648 | // only when an interrupted attempt closed before terminal provider usage. It is |
| 649 | // accounting telemetry, not request admission: the estimate never changes the |
| 650 | // frozen provider request or imposes a token ceiling. |
| 651 | func estimateSamplingRequestInputTokens(req provider.Request) int { |
| 652 | total := 3 |
| 653 | for _, msg := range provider.ModelMessages(req.Messages) { |
| 654 | total += 4 |
| 655 | total += estimateTextTokens(msg.Content) |
| 656 | total += estimateTextTokens(msg.ReasoningContent) |
| 657 | total += estimateTextTokens(msg.ReasoningSignature) |
| 658 | total += estimateTextTokens(msg.Name) |
| 659 | total += estimateTextTokens(msg.ToolCallID) |
| 660 | for _, image := range msg.Images { |
| 661 | total += estimateTextTokens(image) |
| 662 | } |
| 663 | for _, call := range msg.ToolCalls { |
| 664 | total += 8 + estimateTextTokens(call.ID) + estimateTextTokens(call.Name) + estimateTextTokens(call.Arguments) |
| 665 | } |
| 666 | for _, item := range msg.ResponsesItems { |
| 667 | total += estimateTextTokens(string(item)) |
| 668 | } |
| 669 | } |
| 670 | for _, schema := range req.Tools { |
| 671 | encoded, _ := json.Marshal(schema) |
| 672 | total += 8 + estimateTextTokens(string(encoded)) |
| 673 | } |
| 674 | return max(total, 1) |
| 675 | } |
| 676 | |
| 677 | // mergeSamplingUsage accumulates billable counters across body attempts. |
| 678 | // PromptTokens is the billable input total (aligned with cache hit+miss). |
| 679 | // ContextPromptTokens is set later by finalizeSamplingUsage from the latest attempt. |
| 680 | func mergeSamplingUsage(acc, attempt *provider.Usage) *provider.Usage { |
| 681 | if attempt == nil { |
| 682 | return acc |
| 683 | } |
| 684 | billableHitMiss := func(u *provider.Usage) (hit, miss int) { |
| 685 | if u == nil { |
| 686 | return 0, 0 |
| 687 | } |
| 688 | if u.CacheHitTokens+u.CacheMissTokens > 0 { |
| 689 | return u.CacheHitTokens, u.CacheMissTokens |
| 690 | } |
| 691 | // No cache split: treat PromptTokens as uncached billable input. |
| 692 | return 0, u.PromptTokens |
| 693 | } |
| 694 | billablePrompt := func(hit, miss, prompt int) int { |
| 695 | if hit+miss > 0 { |
| 696 | return hit + miss |
| 697 | } |
| 698 | return prompt |
| 699 | } |
| 700 | if acc == nil { |
| 701 | merged := *attempt |
| 702 | if merged.RequestCount <= 0 { |
| 703 | merged.RequestCount = 1 |
| 704 | } |
| 705 | hit, miss := billableHitMiss(attempt) |
| 706 | merged.CacheHitTokens = hit |
| 707 | merged.CacheMissTokens = miss |
| 708 | merged.PromptTokens = billablePrompt(hit, miss, attempt.PromptTokens) |
| 709 | return &merged |
| 710 | } |
| 711 | merged := *acc |
| 712 | // Billable input for Cost: sum hit/miss (prompt when no cache split). |
| 713 | ah, am := billableHitMiss(acc) |
| 714 | bh, bm := billableHitMiss(attempt) |
| 715 | // If acc was previously merged, CacheHit+Miss already holds the sum and |
| 716 | // PromptTokens may still be the first attempt's value — prefer stored sums. |
| 717 | if acc.CacheHitTokens+acc.CacheMissTokens > 0 { |
| 718 | ah, am = acc.CacheHitTokens, acc.CacheMissTokens |
| 719 | } |
| 720 | merged.CacheHitTokens = ah + bh |
| 721 | merged.CacheMissTokens = am + bm |
| 722 | merged.CacheWriteTokens += attempt.CacheWriteTokens |
| 723 | merged.CacheWriteBilledTokens += attempt.CacheWriteBilledTokens |
| 724 | merged.PromptTokens = billablePrompt(merged.CacheHitTokens, merged.CacheMissTokens, 0) |
| 725 | if merged.PromptTokens == 0 { |
| 726 | merged.PromptTokens = acc.PromptTokens + attempt.PromptTokens |
| 727 | } |
| 728 | merged.CompletionTokens += attempt.CompletionTokens |
| 729 | merged.ReasoningTokens += attempt.ReasoningTokens |
| 730 | merged.TotalTokens += usageTotalTokens(attempt) |
| 731 | merged.RequestCount = usageRequestCount(acc) + usageRequestCount(attempt) |
| 732 | if attempt.Estimated { |
| 733 | merged.Estimated = true |
| 734 | } |
| 735 | if attempt.FinishReason != "" { |
| 736 | merged.FinishReason = attempt.FinishReason |
| 737 | } |
| 738 | return &merged |
| 739 | } |
| 740 | |
| 741 | // storeLatestRequestUsage records the most recent single-request usage for |
| 742 | // ContextSnapshot and compaction. It must never receive a multi-attempt |
| 743 | // billable aggregate. |
| 744 | func (a *Agent) storeLatestRequestUsage(attempt *provider.Usage) { |
| 745 | if a == nil || attempt == nil { |
| 746 | return |
| 747 | } |
| 748 | // Skip request-only shells with no token shape. |
| 749 | if attempt.PromptTokens <= 0 && attempt.CompletionTokens <= 0 && attempt.TotalTokens <= 0 { |
| 750 | return |
| 751 | } |
| 752 | clone := *attempt |
| 753 | // RequestCount on lastUsage is not used for context; keep per-attempt value. |
| 754 | a.lastUsage.Store(&clone) |
| 755 | } |
| 756 | |
| 757 | // finalizeSamplingUsage builds the Usage event payload for consumers that |
| 758 | // expect one coherent billable record: |
| 759 | // - PromptTokens / cache hit+miss / Completion / Total / RequestCount: billable aggregate |
| 760 | // - Context* fields: latest attempt only (context gauges + rebind telemetry) |
| 761 | func finalizeSamplingUsage(billable, latest *provider.Usage) *provider.Usage { |
| 762 | if billable == nil && latest == nil { |
| 763 | return nil |
| 764 | } |
| 765 | if billable == nil { |
| 766 | out := *latest |
| 767 | applyLatestContextShape(&out, latest) |
| 768 | return &out |
| 769 | } |
| 770 | out := *billable |
| 771 | if latest != nil { |
| 772 | applyLatestContextShape(&out, latest) |
| 773 | out.FinishReason = latest.FinishReason |
| 774 | } |
| 775 | // Ensure PromptTokens matches billable input (hit+miss) for CLI/ACP/Desktop |
| 776 | // telemetry that requires cache totals to align with PromptTokens. |
| 777 | if hitMiss := out.CacheHitTokens + out.CacheMissTokens; hitMiss > 0 { |
| 778 | out.PromptTokens = hitMiss |
| 779 | } |
| 780 | if out.TotalTokens < out.PromptTokens+out.CompletionTokens { |
| 781 | out.TotalTokens = out.PromptTokens + out.CompletionTokens |
| 782 | } |
| 783 | return &out |
| 784 | } |
| 785 | |
| 786 | // applyLatestContextShape copies the latest single-request shape into Context* |
| 787 | // fields for gauges and Desktop rebind telemetry. |
| 788 | func applyLatestContextShape(dst, latest *provider.Usage) { |
| 789 | if dst == nil || latest == nil { |
| 790 | return |
| 791 | } |
| 792 | dst.ContextPromptTokens = latest.PromptTokens |
| 793 | dst.ContextCompletionTokens = latest.CompletionTokens |
| 794 | dst.ContextReasoningTokens = latest.ReasoningTokens |
| 795 | dst.ContextCacheHitTokens = latest.CacheHitTokens |
| 796 | dst.ContextCacheMissTokens = latest.CacheMissTokens |
| 797 | } |
| 798 | |
| 799 | // mergeStreamUsage remains for missing-reasoning style single-repair merges that |
| 800 | // need a simple sum. Sampling recovery uses mergeSamplingUsage instead. |
| 801 | func mergeStreamUsage(first, retry *provider.Usage) *provider.Usage { |
| 802 | return mergeSamplingUsage(first, retry) |
| 803 | } |
| 804 | |
| 805 | func usageTotalTokens(u *provider.Usage) int { |
| 806 | if u == nil { |
| 807 | return 0 |
| 808 | } |
| 809 | if u.TotalTokens > 0 { |
| 810 | return u.TotalTokens |
| 811 | } |
| 812 | return u.PromptTokens + u.CompletionTokens |
| 813 | } |
| 814 | |
| 815 | func usageRequestCount(usage *provider.Usage) int { |
| 816 | if usage == nil { |
| 817 | return 0 |
| 818 | } |
| 819 | if usage.RequestCount > 0 { |
| 820 | return usage.RequestCount |
| 821 | } |
| 822 | return 1 |
| 823 | } |
| 824 | |
| 825 | func (a *Agent) emitTurnUsage(usage *provider.Usage, cacheDiagnostics *CacheDiagnostics) { |
| 826 | if usage == nil || (usage.TotalTokens <= 0 && usage.RequestCount <= 0) { |
| 827 | return |
| 828 | } |
| 829 | // lastUsage must stay as the latest single-request shape (set during |
| 830 | // sampling recovery). Never overwrite it with a multi-attempt billable |
| 831 | // aggregate — that would inflate ContextSnapshot and compaction decisions. |
| 832 | if a.lastUsage.Load() == nil && usage.PromptTokens > 0 { |
| 833 | a.storeLatestRequestUsage(usage) |
| 834 | } |
| 835 | a.sink.Emit(event.Event{Kind: event.Usage, ModelRef: a.modelRef, Usage: usage, Pricing: a.pricing, |
| 836 | UsageSource: a.usageSource, |
| 837 | CacheDiagnostics: cacheDiagnostics, |
| 838 | SessionHit: int(a.sessCacheHit.Load()), SessionMiss: int(a.sessCacheMiss.Load())}) |
| 839 | } |
| 840 | |
| 841 | // handleFinalResponse processes a no-tool assistant turn: recovery pause, |
| 842 | // readiness retry, empty final retry, executor handoff nudge, steer drain, and |
| 843 | // final compaction. cont=true continues the tool loop; cont=false returns err |
| 844 | // from Run (err may be nil for a clean final answer). |
| 845 | func (a *Agent) handleFinalResponse(ctx context.Context, state *runLoopState, text, reasoning string, usage *provider.Usage) (cont bool, err error) { |
| 846 | // Recovery finalization produced a summary. Keep it in the session, |
| 847 | // but still pause so Goal auto-continue cannot open another Run with |
| 848 | // a fresh finalization round. turn_done reports recovery_paused. |
| 849 | if state.recoveryGraceRound { |
| 850 | a.maybeCompact(ctx, usage) |
| 851 | reason := "" |
| 852 | if ctrl := a.recoveryEpisodeControl(); ctrl != nil { |
| 853 | _, _ = ctrl.ConsumeFinalization(a.recoveryTaskID) |
| 854 | } |
| 855 | return false, &RecoveryPauseError{ |
| 856 | Message: "Automatic retries paused. Reasonix stopped repeated attempts and kept completed work. Send \"continue\" to start a fresh attempt, or add instructions to change direction.", |
| 857 | StopReason: reason, |
| 858 | } |
| 859 | } |
| 860 | readiness := a.finalReadinessCheckFor() |
| 861 | if state.graceRound && (readiness.reason != "" || !hasVisibleFinalAnswer(text)) { |
| 862 | a.maybeCompact(ctx, usage) |
| 863 | return false, &maxStepsPause{steps: state.runMaxSteps, key: state.runMaxStepsKey} |
| 864 | } |
| 865 | if readiness.reason != "" { |
| 866 | // Delivery no longer retries readiness with hidden model messages: the |
| 867 | // run ends immediately with the missing requirements, and the host owns |
| 868 | // what happens next. In Goal mode the FSM auto-continues under budget |
| 869 | // with the missing list as the next turn; plain Delivery turns surface |
| 870 | // the recovery card for an explicit user continuation. |
| 871 | event.RecordReadinessAudit(a.sink, readiness.audit(evidence.ReadinessErrored, false)) |
| 872 | a.deliveryRecoveryPending = true |
| 873 | return false, &FinalReadinessError{Attempts: 1, Reason: readiness.reason, Missing: readiness.missingIDs()} |
| 874 | } |
| 875 | if !hasVisibleFinalAnswer(text) { |
| 876 | // DeepSeek thinking mode can stream a long reasoning_content and |
| 877 | // then finish with finish_reason="stop" but an empty content |
| 878 | // block: the model has explicitly signalled completion and its |
| 879 | // reasoning was already streamed to the user. Retrying here overrides |
| 880 | // that stop signal and forces another expensive thinking round (the |
| 881 | // "still thinking after the task is done" symptom), so honour the |
| 882 | // stop when reasoning carried the substance of the answer and treat |
| 883 | // the turn as a final answer instead of retrying. |
| 884 | if !reasoningOnlyFinishHonoured(a.prov, usage, reasoning) { |
| 885 | state.emptyFinalBlocks++ |
| 886 | if state.emptyFinalBlocks >= maxEmptyFinalBlocks { |
| 887 | return false, fmt.Errorf("model finished without a visible final answer %d times", state.emptyFinalBlocks) |
| 888 | } |
| 889 | a.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo, Code: event.NoticeCodeEmptyFinal, Text: emptyFinalNotice(), Detail: emptyFinalNoticeDetail(a.prov.Name(), usage, len(reasoning))}) |
| 890 | a.session.Add(provider.Message{Role: provider.RoleUser, Content: a.withTurnPreferences(emptyFinalRetryMessage())}) |
| 891 | a.maybeCompact(ctx, usage) |
| 892 | return true, nil |
| 893 | } |
| 894 | } |
| 895 | if state.executorHandoff && !state.usedAnyTool && state.handoffNudges < maxExecutorHandoffNudges && shouldNudgeExecutorHandoff(state.input, text) { |
| 896 | state.handoffNudges++ |
| 897 | a.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo, Code: event.NoticeCodeExecutorHandoff, Text: executorHandoffNoticeText(), Detail: "executor answered without taking any action; nudging it to use its tools"}) |
| 898 | a.session.Add(provider.Message{Role: provider.RoleUser, Content: a.withTurnPreferences(executorHandoffRetryMessage())}) |
| 899 | a.maybeCompact(ctx, usage) |
| 900 | return true, nil |
| 901 | } |
| 902 | if readiness.applies { |
| 903 | event.RecordReadinessAudit(a.sink, readiness.audit(evidence.ReadinessAllowed, a.readinessRecovered)) |
| 904 | } |
| 905 | if !a.closeSteerIntakeIfIdle() { |
| 906 | return true, nil |
| 907 | } |
| 908 | // A final-answer turn otherwise skips compaction, so a large context |
| 909 | // carries into the next turn un-folded and can overflow the model window. |
| 910 | // No-op below the trigger, so normal turns keep their warm cache. |
| 911 | a.maybeCompact(ctx, usage) |
| 912 | return false, nil // model gave a final answer |
| 913 | } |
| 914 | |
| 915 | // handleToolRound executes a tool batch, persists tool messages, handles |
| 916 | // cancellation, todo stall tracking, recovery finalization pause, and the |
| 917 | // max-steps grace round. cont=true continues the tool loop; cont=false returns |
| 918 | // err from Run. |
| 919 | func (a *Agent) handleToolRound(ctx context.Context, state *runLoopState, step int, text, reasoning string, calls []provider.ToolCall, usage *provider.Usage) (cont bool, err error) { |
| 920 | state.emptyFinalBlocks = 0 |
| 921 | state.usedAnyTool = true |
| 922 | outOfContextGoalOnly := toolCallsAreOutOfContextGoalReports(ctx, calls) |
| 923 | |
| 924 | // Grace round guard: if we already gave the model one extra response |
| 925 | // and it still wants to call tools, stop here. |
| 926 | if state.graceRound { |
| 927 | return false, &maxStepsPause{steps: state.runMaxSteps, key: state.runMaxStepsKey} |
| 928 | } |
| 929 | // Recovery Episode exhausted: one finalization round only. Further tool |
| 930 | // calls are not executed; return a typed pause so the host can surface |
| 931 | // recovery_paused without treating it as a send failure. |
| 932 | if state.recoveryGraceRound { |
| 933 | reason := "" |
| 934 | if ctrl := a.recoveryEpisodeControl(); ctrl != nil { |
| 935 | _, _ = ctrl.ConsumeFinalization(a.recoveryTaskID) |
| 936 | } |
| 937 | // Pair tool-call / tool-result without executing. |
| 938 | msg := "blocked: Auto recovery already paused this turn. Do not call tools; the user will continue in the next message." |
| 939 | for _, call := range calls { |
| 940 | a.session.Add(provider.Message{ |
| 941 | Role: provider.RoleTool, |
| 942 | Content: msg, |
| 943 | ToolCallID: call.ID, |
| 944 | Name: call.Name, |
| 945 | }) |
| 946 | } |
| 947 | a.maybeCompact(ctx, usage) |
| 948 | return false, &RecoveryPauseError{ |
| 949 | Message: "Automatic retries paused. Reasonix stopped repeated attempts and kept completed work. Send \"continue\" to start a fresh attempt, or add instructions to change direction.", |
| 950 | StopReason: reason, |
| 951 | } |
| 952 | } |
| 953 | |
| 954 | receiptMark := 0 |
| 955 | if a.evidence != nil { |
| 956 | receiptMark = a.evidence.Len() |
| 957 | } |
| 958 | batch := a.executeBatch(ctx, calls) |
| 959 | results, images := batch.results, batch.images |
| 960 | for i, call := range calls { |
| 961 | msg := provider.Message{ |
| 962 | Role: provider.RoleTool, |
| 963 | Content: results[i], |
| 964 | Images: images[i], |
| 965 | ToolCallID: call.ID, |
| 966 | Name: call.Name, |
| 967 | } |
| 968 | if i < len(batch.executions) { |
| 969 | msg.ToolExecution = toProviderToolExecution(batch.executions[i]) |
| 970 | } |
| 971 | a.session.Add(msg) |
| 972 | } |
| 973 | // If the context was cancelled during tool execution, return after storing |
| 974 | // the batch results so the session keeps paired tool-call history. |
| 975 | if ctx.Err() != nil { |
| 976 | a.recordInterruptedDisplay("", "", nil, true, state.workDurationMs()) |
| 977 | return false, ctx.Err() |
| 978 | } |
| 979 | if outOfContextGoalOnly { |
| 980 | if hasVisibleFinalAnswer(text) { |
| 981 | // Keep the assistant tool call and host error paired in the transcript, |
| 982 | // but accept the co-streamed answer instead of spending another model |
| 983 | // request repairing harmless Goal bookkeeping outside Goal mode. |
| 984 | return a.handleFinalResponse(ctx, state, text, reasoning, usage) |
| 985 | } |
| 986 | state.goalToolRepairs++ |
| 987 | if state.goalToolRepairs > 1 { |
| 988 | return false, fmt.Errorf("model repeatedly called update_goal outside Goal mode without a visible answer") |
| 989 | } |
| 990 | } |
| 991 | if !a.planMode.Load() { |
| 992 | nextProgress, nextTracking := a.canonicalTodoProgress() |
| 993 | hostProgress := false |
| 994 | if a.evidence != nil { |
| 995 | for _, sig := range a.evidence.SuccessfulProgressSignaturesSince(receiptMark) { |
| 996 | if _, seen := state.seenTodoProgress[sig]; !seen { |
| 997 | hostProgress = true |
| 998 | state.seenTodoProgress[sig] = struct{}{} |
| 999 | } |
| 1000 | } |
| 1001 | } |
| 1002 | switch { |
| 1003 | case !nextTracking: |
| 1004 | state.todoStallRounds = 0 |
| 1005 | case !state.trackingTodoProgress || nextProgress > state.todoProgress || hostProgress: |
| 1006 | state.todoStallRounds = 0 |
| 1007 | default: |
| 1008 | state.todoStallRounds++ |
| 1009 | } |
| 1010 | state.todoProgress, state.trackingTodoProgress = nextProgress, nextTracking |
| 1011 | if state.todoStallRounds == todoProgressNudgeRounds { |
| 1012 | nudge := todoProgressNudgeMessage(state.todoStallRounds) |
| 1013 | a.session.Add(provider.Message{Role: provider.RoleUser, Content: a.withTurnPreferences(nudge)}) |
| 1014 | a.sink.Emit(event.Event{ |
| 1015 | Kind: event.Notice, Level: event.LevelInfo, Code: event.NoticeCodeLoopGuard, |
| 1016 | Text: loopGuardNoticeText(), |
| 1017 | Detail: fmt.Sprintf("the current todo has no new completion, unique read, command, or mutation for %d consecutive tool-call rounds; asking the assistant to reassess", state.todoStallRounds), |
| 1018 | }) |
| 1019 | } |
| 1020 | if state.todoStallRounds >= maxTodoStallRounds { |
| 1021 | a.sink.Emit(event.Event{ |
| 1022 | Kind: event.Notice, Level: event.LevelInfo, Code: event.NoticeCodeLoopGuard, |
| 1023 | Text: "Task progress stalled; pausing before more tools are called.", |
| 1024 | Detail: fmt.Sprintf("the current todo has no new completion, unique read, command, or mutation for %d consecutive tool-call rounds after a host reassessment; work is saved and can be resumed", state.todoStallRounds), |
| 1025 | }) |
| 1026 | return false, &todoStallPause{rounds: state.todoStallRounds} |
| 1027 | } |
| 1028 | } |
| 1029 | |
| 1030 | // The prompt only grows from here; compact before the next turn so it |
| 1031 | // stays within the model's window. |
| 1032 | a.maybeCompact(ctx, usage) |
| 1033 | |
| 1034 | // When Auto recovery exhausts its Episode budget, offer exactly one |
| 1035 | // summarize-only finalization round. Successful summary ends cleanly; |
| 1036 | // further tool calls surface RecoveryPauseError. |
| 1037 | if batch.recoveryStopTurn && !state.recoveryGraceRound { |
| 1038 | state.recoveryGraceRound = true |
| 1039 | if ctrl := a.recoveryEpisodeControl(); ctrl != nil { |
| 1040 | ctrl.MarkFinalizationOffered(a.recoveryTaskID) |
| 1041 | } |
| 1042 | nudge := "Auto recovery has reached its limit for this turn. Do not call any more tools. Summarize what was completed, what failed, and what the user should do next. The user can continue in the next message." |
| 1043 | a.session.Add(provider.Message{Role: provider.RoleUser, Content: a.withTurnPreferences(nudge)}) |
| 1044 | return true, nil |
| 1045 | } |
| 1046 | |
| 1047 | // When the tool-call budget runs out this round, give the model |
| 1048 | // one grace round to produce a final answer from completed work. |
| 1049 | if state.runMaxSteps > 0 && step+1 >= state.runMaxSteps { |
| 1050 | state.graceRound = true |
| 1051 | nextStep := fmt.Sprintf("The user can increase %s or continue in the next turn if more work is needed.", state.runMaxStepsKey) |
| 1052 | if state.runLimitHostOwned { |
| 1053 | nextStep = "Use the evidence already collected, label remaining uncertainty, and keep the final answer actionable." |
| 1054 | } |
| 1055 | nudge := fmt.Sprintf("Do not call any more tools — your tool-call round limit (%s) has been reached. Instead, synthesize a final answer from all the work already completed: summarize what was accomplished, what remains to be done, and any decisions the user should make. %s", state.runMaxStepsKey, nextStep) |
| 1056 | a.session.Add(provider.Message{Role: provider.RoleUser, Content: a.withTurnPreferences(nudge)}) |
| 1057 | a.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo, Code: event.NoticeCodeToolBudget, Text: toolBudgetNoticeText(), Detail: fmt.Sprintf("budget (%s=%d) exhausted: one grace round to finalize", state.runMaxStepsKey, state.runMaxSteps)}) |
| 1058 | } |
| 1059 | return true, nil |
| 1060 | } |
| 1061 | |
| 1062 | func toolCallsAreOutOfContextGoalReports(ctx context.Context, calls []provider.ToolCall) bool { |
| 1063 | if len(calls) == 0 { |
| 1064 | return false |
| 1065 | } |
| 1066 | if _, ok := tool.GoalTurnRecorderFromContext(ctx); ok { |
| 1067 | return false |
| 1068 | } |
| 1069 | for _, call := range calls { |
| 1070 | if call.Name != "update_goal" { |
| 1071 | return false |
| 1072 | } |
| 1073 | } |
| 1074 | return true |
| 1075 | } |
| 1076 |