返回 DeepSeek-Reasonix
reasoning_warn_state.go
根目录 / internal / agent / reasoning_warn_state.go
1 package agent
2
3 // reasoning_warn_state.go rate-limits missing tool-call thinking recovery
4 // attempts across sessions and processes (#7059). Legacy warning-oriented Go
5 // names and JSON fields intentionally remain unchanged for upgrade and
6 // cross-version compatibility.
7 //
8 // DeepSeek's thinking-mode contract requires provider-issued thinking content
9 // to accompany tool calls and be replayed on later requests. Missing content is
10 // therefore a compatibility incident, not a permanent provider characteristic.
11 // State is keyed by an opaque configuration fingerprint, expires after a bounded
12 // cooldown, and is cleared after three consecutive healthy tool-call turns so
13 // a future isolated regression gets one fresh silent retry without flapping.
14
15 import (
16 "context"
17 "encoding/json"
18 "math"
19 "os"
20 "path/filepath"
21 "runtime"
22 "sort"
23 "strings"
24 "sync"
25 "time"
26
27 "reasonix/internal/fileutil"
28 filelock "reasonix/internal/identitylock"
29 )
30
31 const missingReasoningWarnStateFilename = "tool-call-reasoning-warning.json"
32
33 const (
34 missingReasoningWarnStateLockFilename = "tool-call-reasoning-warning.lock"
35 missingReasoningWarnStateVersion = 2
36 missingReasoningWarnStateCooldown = 24 * time.Hour
37 missingReasoningWarnStateMaxIncidents = 256
38 missingReasoningHealthyResolveStreak = 3
39 // Recovery bookkeeping must never make a turn wait indefinitely behind
40 // another process. On contention or I/O failure, callers fail open and allow
41 // the bounded retry rather than silently losing self-healing.
42 missingReasoningWarnStateLockTimeout = 200 * time.Millisecond
43 )
44
45 type missingReasoningIncident struct {
46 Fingerprint string `json:"fingerprint"`
47 WarnedAtUnixMs int64 `json:"warnedAtUnixMs,omitempty"`
48 LastMissingUnixMs int64 `json:"lastMissingAtUnixMs,omitempty"`
49 LastMissingUnixNano int64 `json:"lastMissingAtUnixNano,omitempty"`
50 LastResolvedAtUnixNano int64 `json:"lastResolvedAtUnixNano,omitempty"`
51 // These optional fields extend the v2 document without changing its version.
52 // Older builds ignore them and may drop an in-progress streak on write, while
53 // both generations continue to share the same active-incident boundary.
54 ResolveStreak int `json:"resolveStreak,omitempty"`
55 LastHealthyAtUnixNano int64 `json:"lastHealthyAtUnixNano,omitempty"`
56 }
57
58 type missingReasoningWarnDocument struct {
59 Version int `json:"version"`
60 Incidents []missingReasoningIncident `json:"incidents,omitempty"`
61 // Providers reads the unreleased v1 preview schema. Provider names cannot
62 // safely identify endpoint/model/protocol changes and have no timestamp, so
63 // they deliberately re-arm once and are omitted on the next v2 write.
64 Providers []string `json:"providers,omitempty"`
65 }
66
67 type missingReasoningWarnState struct {
68 dir string
69 }
70
71 // missingReasoningWarnClaimFlights coalesces concurrent observations of the
72 // same incident inside one process. filelock.Acquire intentionally has a short
73 // deadline so recovery bookkeeping cannot stall a turn behind another process. On a
74 // slower filesystem (notably Windows CI), several local callers can otherwise
75 // exhaust that deadline while queued behind the first atomic write and each
76 // take the fail-open path, producing duplicate recovery requests.
77 //
78 // Followers update the latest observation timestamp and let the leader persist
79 // it before the flight is removed. This preserves resolveAt's stale-health
80 // ordering guarantee instead of merely suppressing duplicate return values.
81 type missingReasoningWarnClaimFlight struct {
82 latestObservedAt time.Time
83 }
84
85 var missingReasoningWarnClaimFlights = struct {
86 sync.Mutex
87 flights map[string]*missingReasoningWarnClaimFlight
88 }{flights: map[string]*missingReasoningWarnClaimFlight{}}
89
90 // missingReasoningWarnProcessLocks serializes transactions for one state file
91 // before they enter filelock.Acquire. The file lock's short deadline is meant
92 // to bound waits on another process, not make distinct local incidents lose
93 // persistence while queued behind a slower Windows atomic write. Reasonix uses
94 // one state path per home, so retaining these few mutexes for the process
95 // lifetime is bounded in normal operation.
96 var missingReasoningWarnProcessLocks sync.Map
97
98 func newMissingReasoningWarnState(dir string) *missingReasoningWarnState {
99 return &missingReasoningWarnState{dir: strings.TrimSpace(dir)}
100 }
101
102 func (s *missingReasoningWarnState) path() string {
103 return filepath.Join(s.dir, missingReasoningWarnStateFilename)
104 }
105
106 func (s *missingReasoningWarnState) lockPath() string {
107 return filepath.Join(s.dir, missingReasoningWarnStateLockFilename)
108 }
109
110 func (s *missingReasoningWarnState) processLockKey() string {
111 path, err := filepath.Abs(s.path())
112 if err != nil {
113 path = s.path()
114 }
115 path = filepath.Clean(path)
116 if runtime.GOOS == "windows" {
117 path = strings.ToLower(filepath.ToSlash(path))
118 }
119 return path
120 }
121
122 func (s *missingReasoningWarnState) processLock() *sync.Mutex {
123 lock, _ := missingReasoningWarnProcessLocks.LoadOrStore(s.processLockKey(), &sync.Mutex{})
124 return lock.(*sync.Mutex)
125 }
126
127 func (s *missingReasoningWarnState) claimFlightKey(fingerprint string) string {
128 return s.processLockKey() + "\x00" + fingerprint
129 }
130
131 func validMissingReasoningFingerprint(fingerprint string) bool {
132 if len(fingerprint) != 64 {
133 return false
134 }
135 for _, r := range fingerprint {
136 if (r < '0' || r > '9') && (r < 'a' || r > 'f') {
137 return false
138 }
139 }
140 return true
141 }
142
143 func missingReasoningUnixNanoFromMillis(unixMs int64) (int64, bool) {
144 if unixMs <= 0 || unixMs > math.MaxInt64/int64(time.Millisecond) {
145 return 0, false
146 }
147 return unixMs * int64(time.Millisecond), true
148 }
149
150 func validMissingReasoningIncidentFields(incident missingReasoningIncident) bool {
151 return incident.LastMissingUnixNano >= 0 && incident.LastResolvedAtUnixNano >= 0 &&
152 incident.LastHealthyAtUnixNano >= 0 &&
153 incident.ResolveStreak >= 0 && incident.ResolveStreak < missingReasoningHealthyResolveStreak
154 }
155
156 func normalizeMissingReasoningIncident(incident missingReasoningIncident, now time.Time) (missingReasoningIncident, bool) {
157 incident.Fingerprint = strings.TrimSpace(incident.Fingerprint)
158 if !validMissingReasoningFingerprint(incident.Fingerprint) ||
159 !validMissingReasoningIncidentFields(incident) {
160 return missingReasoningIncident{}, false
161 }
162 incident, warnedAtUnixNano, ok := normalizeMissingReasoningTimestamps(incident)
163 if !ok {
164 return missingReasoningIncident{}, false
165 }
166 nowUnixNano := now.UnixNano()
167 if warnedAtUnixNano > nowUnixNano || incident.LastMissingUnixNano > nowUnixNano ||
168 incident.LastResolvedAtUnixNano > nowUnixNano || incident.LastHealthyAtUnixNano > nowUnixNano {
169 return missingReasoningIncident{}, false
170 }
171 if incident.LastMissingUnixNano > incident.LastResolvedAtUnixNano {
172 return normalizeActiveMissingReasoningIncident(incident, warnedAtUnixNano, now)
173 }
174 return normalizeResolvedMissingReasoningIncident(incident, now)
175 }
176
177 func normalizeMissingReasoningTimestamps(incident missingReasoningIncident) (missingReasoningIncident, int64, bool) {
178 warnedAtUnixNano := int64(0)
179 if incident.WarnedAtUnixMs != 0 {
180 var ok bool
181 warnedAtUnixNano, ok = missingReasoningUnixNanoFromMillis(incident.WarnedAtUnixMs)
182 if !ok {
183 return missingReasoningIncident{}, 0, false
184 }
185 }
186 if incident.LastMissingUnixMs == 0 {
187 return incident, warnedAtUnixNano, incident.LastMissingUnixNano == 0
188 }
189 lastMissingFromMillis, ok := missingReasoningUnixNanoFromMillis(incident.LastMissingUnixMs)
190 if !ok {
191 return missingReasoningIncident{}, 0, false
192 }
193 if incident.LastMissingUnixNano == 0 {
194 incident.LastMissingUnixNano = lastMissingFromMillis
195 } else if incident.LastMissingUnixNano/int64(time.Millisecond) != incident.LastMissingUnixMs {
196 return missingReasoningIncident{}, 0, false
197 }
198 return incident, warnedAtUnixNano, true
199 }
200
201 func normalizeActiveMissingReasoningIncident(incident missingReasoningIncident, warnedAtUnixNano int64, now time.Time) (missingReasoningIncident, bool) {
202 if warnedAtUnixNano == 0 || incident.LastMissingUnixNano < warnedAtUnixNano {
203 return missingReasoningIncident{}, false
204 }
205 ageOrigin := time.UnixMilli(incident.WarnedAtUnixMs)
206 maxAge := missingReasoningWarnStateCooldown
207 age := now.Sub(ageOrigin)
208 if age < 0 || age >= maxAge {
209 return missingReasoningIncident{}, false
210 }
211 if incident.ResolveStreak > 0 {
212 if incident.LastHealthyAtUnixNano <= incident.LastMissingUnixNano ||
213 incident.LastHealthyAtUnixNano <= incident.LastResolvedAtUnixNano {
214 return missingReasoningIncident{}, false
215 }
216 } else if incident.LastHealthyAtUnixNano != 0 {
217 return missingReasoningIncident{}, false
218 }
219 return incident, true
220 }
221
222 func normalizeResolvedMissingReasoningIncident(incident missingReasoningIncident, now time.Time) (missingReasoningIncident, bool) {
223 if incident.LastResolvedAtUnixNano <= 0 || incident.ResolveStreak != 0 ||
224 incident.LastHealthyAtUnixNano > incident.LastResolvedAtUnixNano {
225 return missingReasoningIncident{}, false
226 }
227 resolvedAt := time.Unix(0, incident.LastResolvedAtUnixNano)
228 age := now.Sub(resolvedAt)
229 if age < 0 || age >= missingReasoningWarnStateCooldown {
230 return missingReasoningIncident{}, false
231 }
232 return incident, true
233 }
234
235 func (incident missingReasoningIncident) lastEventUnixNano() int64 {
236 return incident.lastObservedUnixNano()
237 }
238
239 func (incident missingReasoningIncident) lastObservedUnixNano() int64 {
240 return max(incident.LastHealthyAtUnixNano, max(incident.LastResolvedAtUnixNano, incident.LastMissingUnixNano))
241 }
242
243 // load returns only current v2 incidents and resolution watermarks. Missing,
244 // corrupt, legacy, expired, or future-dated entries are treated as absent and
245 // self-heal on the next write. Read failures other than a missing file remain
246 // visible to the transaction so it cannot overwrite state from a partial read.
247 func (s *missingReasoningWarnState) load(now time.Time) (map[string]missingReasoningIncident, error) {
248 incidents := map[string]missingReasoningIncident{}
249 b, err := os.ReadFile(s.path())
250 if err != nil {
251 if os.IsNotExist(err) {
252 return incidents, nil
253 }
254 return nil, err
255 }
256 var doc missingReasoningWarnDocument
257 if json.Unmarshal(b, &doc) != nil || doc.Version != missingReasoningWarnStateVersion {
258 return incidents, nil
259 }
260 for _, rawIncident := range doc.Incidents {
261 incident, ok := normalizeMissingReasoningIncident(rawIncident, now)
262 if !ok {
263 continue
264 }
265 if previous, exists := incidents[incident.Fingerprint]; !exists || previous.lastEventUnixNano() < incident.lastEventUnixNano() {
266 incidents[incident.Fingerprint] = incident
267 }
268 }
269 return incidents, nil
270 }
271
272 func (s *missingReasoningWarnState) save(incidents map[string]missingReasoningIncident) error {
273 ordered := make([]missingReasoningIncident, 0, len(incidents))
274 for _, incident := range incidents {
275 ordered = append(ordered, incident)
276 }
277 if len(ordered) > missingReasoningWarnStateMaxIncidents {
278 sort.Slice(ordered, func(i, j int) bool {
279 return ordered[i].lastEventUnixNano() > ordered[j].lastEventUnixNano()
280 })
281 ordered = ordered[:missingReasoningWarnStateMaxIncidents]
282 }
283 sort.Slice(ordered, func(i, j int) bool {
284 return ordered[i].Fingerprint < ordered[j].Fingerprint
285 })
286 b, err := json.Marshal(missingReasoningWarnDocument{
287 Version: missingReasoningWarnStateVersion,
288 Incidents: ordered,
289 })
290 if err != nil {
291 return err
292 }
293 return fileutil.AtomicWriteFile(s.path(), b, 0o600)
294 }
295
296 func (s *missingReasoningWarnState) acquire() (func(), error) {
297 if err := os.MkdirAll(s.dir, 0o700); err != nil {
298 return nil, err
299 }
300 ctx, cancel := context.WithTimeout(context.Background(), missingReasoningWarnStateLockTimeout)
301 release, err := filelock.Acquire(ctx, s.lockPath())
302 if err != nil {
303 cancel()
304 return nil, err
305 }
306 return func() {
307 release()
308 cancel()
309 }, nil
310 }
311
312 func normalizeMissingReasoningObservedAt(observedAt time.Time) time.Time {
313 if observedAt.IsZero() {
314 return time.Now()
315 }
316 return observedAt
317 }
318
319 func missingReasoningTransactionNow(observedAt time.Time) time.Time {
320 now := time.Now()
321 if observedAt.After(now) {
322 return observedAt
323 }
324 return now
325 }
326
327 // persistClaimAt performs one cross-process read-check-write transaction.
328 // Persistence failure returns true so recovery fails open rather than being
329 // disabled by local bookkeeping trouble.
330 func (s *missingReasoningWarnState) persistClaimAt(fingerprint string, observedAt time.Time) bool {
331 processLock := s.processLock()
332 processLock.Lock()
333 defer processLock.Unlock()
334
335 release, err := s.acquire()
336 if err != nil {
337 return true
338 }
339 defer release()
340
341 incidents, err := s.load(missingReasoningTransactionNow(observedAt))
342 if err != nil {
343 return true
344 }
345 incident, exists := incidents[fingerprint]
346 observedAtUnixNano := observedAt.UnixNano()
347 if exists && (observedAtUnixNano <= incident.LastResolvedAtUnixNano ||
348 observedAtUnixNano <= incident.LastHealthyAtUnixNano) {
349 return false
350 }
351 activeIncident := exists && incident.LastMissingUnixNano > incident.LastResolvedAtUnixNano
352 shouldRetry := !activeIncident
353 if shouldRetry {
354 lastResolvedAtUnixNano := incident.LastResolvedAtUnixNano
355 incident = missingReasoningIncident{
356 Fingerprint: fingerprint,
357 WarnedAtUnixMs: observedAt.UnixMilli(),
358 LastResolvedAtUnixNano: lastResolvedAtUnixNano,
359 }
360 }
361 if incident.LastMissingUnixNano < observedAtUnixNano {
362 incident.LastMissingUnixMs = observedAt.UnixMilli()
363 incident.LastMissingUnixNano = observedAtUnixNano
364 incident.ResolveStreak = 0
365 incident.LastHealthyAtUnixNano = 0
366 }
367 incidents[fingerprint] = incident
368 if err := s.save(incidents); err != nil {
369 return true
370 }
371 return shouldRetry
372 }
373
374 // claimAt records the latest missing-reasoning observation and reports whether
375 // this active incident should receive its one recovery retry. Concurrent
376 // observations of the same incident in this process share one leader; the file
377 // lock covers the leader's read-check-write transactions against other
378 // processes sharing the Reasonix home.
379 func (s *missingReasoningWarnState) claimAt(fingerprint string, observedAt time.Time) bool {
380 fingerprint = strings.TrimSpace(fingerprint)
381 if s == nil || s.dir == "" || !validMissingReasoningFingerprint(fingerprint) {
382 return true
383 }
384 observedAt = normalizeMissingReasoningObservedAt(observedAt)
385 key := s.claimFlightKey(fingerprint)
386
387 missingReasoningWarnClaimFlights.Lock()
388 if flight := missingReasoningWarnClaimFlights.flights[key]; flight != nil {
389 if observedAt.After(flight.latestObservedAt) {
390 flight.latestObservedAt = observedAt
391 }
392 missingReasoningWarnClaimFlights.Unlock()
393 return false
394 }
395 flight := &missingReasoningWarnClaimFlight{latestObservedAt: observedAt}
396 missingReasoningWarnClaimFlights.flights[key] = flight
397 missingReasoningWarnClaimFlights.Unlock()
398
399 processedAt := time.Time{}
400 shouldRetry := false
401 for {
402 missingReasoningWarnClaimFlights.Lock()
403 nextObservedAt := flight.latestObservedAt
404 missingReasoningWarnClaimFlights.Unlock()
405
406 if nextObservedAt.After(processedAt) {
407 shouldRetry = s.persistClaimAt(fingerprint, nextObservedAt) || shouldRetry
408 processedAt = nextObservedAt
409 }
410
411 missingReasoningWarnClaimFlights.Lock()
412 if !flight.latestObservedAt.After(processedAt) {
413 delete(missingReasoningWarnClaimFlights.flights, key)
414 missingReasoningWarnClaimFlights.Unlock()
415 return shouldRetry
416 }
417 missingReasoningWarnClaimFlights.Unlock()
418 }
419 }
420
421 func (s *missingReasoningWarnState) claim(fingerprint string) bool {
422 return s.claimAt(fingerprint, time.Now())
423 }
424
425 type missingReasoningResolveResult struct {
426 Recorded bool
427 Resolved bool
428 }
429
430 // resolveAt records one healthy tool-call turn. Three consecutive healthy
431 // observations resolve the incident; another missing observation resets the
432 // streak. The healthy and resolved watermarks prevent events that happened
433 // earlier but acquired the cross-process lock later from changing newer state.
434 func (s *missingReasoningWarnState) resolveAt(fingerprint string, observedAt time.Time) missingReasoningResolveResult {
435 fingerprint = strings.TrimSpace(fingerprint)
436 if s == nil || s.dir == "" || !validMissingReasoningFingerprint(fingerprint) {
437 return missingReasoningResolveResult{}
438 }
439 observedAt = normalizeMissingReasoningObservedAt(observedAt)
440 processLock := s.processLock()
441 processLock.Lock()
442 defer processLock.Unlock()
443
444 release, err := s.acquire()
445 if err != nil {
446 return missingReasoningResolveResult{}
447 }
448 defer release()
449
450 incidents, err := s.load(missingReasoningTransactionNow(observedAt))
451 if err != nil {
452 return missingReasoningResolveResult{}
453 }
454 incident, exists := incidents[fingerprint]
455 observedAtUnixNano := observedAt.UnixNano()
456 if !exists || incident.LastMissingUnixNano <= incident.LastResolvedAtUnixNano {
457 return missingReasoningResolveResult{Recorded: true, Resolved: true}
458 }
459 if incident.LastMissingUnixNano >= observedAtUnixNano || incident.LastHealthyAtUnixNano >= observedAtUnixNano {
460 return missingReasoningResolveResult{Recorded: true}
461 }
462 incident.ResolveStreak++
463 incident.LastHealthyAtUnixNano = observedAtUnixNano
464 resolved := incident.ResolveStreak >= missingReasoningHealthyResolveStreak
465 if resolved {
466 incident.ResolveStreak = 0
467 incident.LastResolvedAtUnixNano = observedAtUnixNano
468 }
469 incidents[fingerprint] = incident
470 if s.save(incidents) != nil {
471 return missingReasoningResolveResult{}
472 }
473 return missingReasoningResolveResult{Recorded: true, Resolved: resolved}
474 }
475
475 lines GO