| 1 | package recovery |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/agent" |
| 12 | ) |
| 13 | |
| 14 | // ModeProvider reports the current tool-approval mode (ask|auto|yolo). |
| 15 | type ModeProvider func() string |
| 16 | |
| 17 | // EmitPromptFunc shows a fresh Auto Guard card and returns its id. |
| 18 | // It must not grant session or persistent authorization. The gate waits until |
| 19 | // Resolve is called for that id (or ctx ends). |
| 20 | type EmitPromptFunc func(ctx context.Context, taskID string, pending PendingProposal, failure *FailureEvent) (approvalID string, err error) |
| 21 | |
| 22 | // Reviewer evaluates ambiguous failure-recovery proposals. |
| 23 | type Reviewer interface { |
| 24 | Review(ctx context.Context, failure *FailureEvent, diagnosis []string, proposal Proposal, taskSummary string) (ReviewVerdict, error) |
| 25 | } |
| 26 | |
| 27 | // Options configures a Gate. |
| 28 | type Options struct { |
| 29 | Mode ModeProvider |
| 30 | EmitPrompt EmitPromptFunc |
| 31 | Reviewer Reviewer |
| 32 | TaskSummary func() string |
| 33 | MaxReviewBlocks int // consecutive reviewer blocks before stop-and-report guidance |
| 34 | Now func() time.Time |
| 35 | // Headless, when true, never waits for a human: blocks the mutation with a |
| 36 | // structured blocker message instead. |
| 37 | Headless bool |
| 38 | // PersistenceKey is sampled synchronously when a state change is scheduled. |
| 39 | // Persist receives that captured key so an asynchronous write cannot follow |
| 40 | // a later session switch and land in the wrong sidecar. |
| 41 | PersistenceKey func() string |
| 42 | // Persist is invoked after meaningful state changes (optional). |
| 43 | // Receives the persistence projection (never active locks). |
| 44 | Persist func(key string, snapshot Snapshot) |
| 45 | } |
| 46 | |
| 47 | // Gate is the Auto Guard coordinator for one controller session. |
| 48 | // Root, foreground sub-agents, and background writer sub-agents share it. |
| 49 | // Exact-operation failure counts are isolated by TaskID; Episode totals, |
| 50 | // reviewer rejects, and hard stop are shared on episode so a new sub-agent |
| 51 | // cannot reset the hard ceiling. Pure routing lives in Decide. |
| 52 | // |
| 53 | // EpisodeID is host-owned temporary execution-round state. TaskScopeID continues |
| 54 | // to scope Goal and task grants. Episode/generation/waiters never persist. |
| 55 | type Gate struct { |
| 56 | mu sync.Mutex |
| 57 | opts Options |
| 58 | tasks map[string]*taskRuntime |
| 59 | metrics Metrics |
| 60 | waiters map[string]chan resolvePayload // keyed by approval id |
| 61 | taskOf map[string]string // approval id -> task id |
| 62 | pending map[string]PendingProposal // approval id -> transient proposal scope |
| 63 | // awaiting tracks in-flight human prompts so Phase can be derived without |
| 64 | // storing Pending on the task runtime. |
| 65 | awaiting map[string]struct{} // task ids with an open waiter |
| 66 | |
| 67 | // episodeSeq / episodeID identify the current host-owned Recovery Episode. |
| 68 | // generation invalidates in-flight tool observations across mode switches. |
| 69 | // episode holds totals and hard-stop shared by every TaskID in the Episode. |
| 70 | episodeSeq uint64 |
| 71 | episodeID string |
| 72 | generation uint64 |
| 73 | episode episodeBudget |
| 74 | lastMode string |
| 75 | haveMode bool |
| 76 | |
| 77 | // persistMu orders asynchronous snapshots. A newer state may be scheduled |
| 78 | // before an older goroutine reaches disk; sequence checks prevent that older |
| 79 | // snapshot from overwriting the newer checkpoint. |
| 80 | persistMu sync.Mutex |
| 81 | persistSeq uint64 |
| 82 | persistCond *sync.Cond |
| 83 | // persistPending and persistDone are tracked per session key so old and new |
| 84 | // sessions can drain independently without retaining keys after completion. |
| 85 | persistPending map[string]int |
| 86 | persistDone map[string]uint64 |
| 87 | } |
| 88 | |
| 89 | type resolvePayload struct { |
| 90 | action Action |
| 91 | feedback string |
| 92 | } |
| 93 | |
| 94 | // dismissedWaiter is a recovery waiter cancelled by mode switch / episode rotate. |
| 95 | type dismissedWaiter struct { |
| 96 | id string |
| 97 | taskID string |
| 98 | reply chan resolvePayload |
| 99 | payload resolvePayload |
| 100 | } |
| 101 | |
| 102 | // NewGate constructs Auto Guard. The gate is active whenever approval mode is |
| 103 | // Auto; Ask and YOLO bypass it through the mode provider. |
| 104 | func NewGate(opts Options) *Gate { |
| 105 | if opts.Mode == nil { |
| 106 | opts.Mode = func() string { return "auto" } |
| 107 | } |
| 108 | if opts.Now == nil { |
| 109 | opts.Now = time.Now |
| 110 | } |
| 111 | if opts.MaxReviewBlocks <= 0 { |
| 112 | opts.MaxReviewBlocks = MaxReviewRejects |
| 113 | } |
| 114 | g := &Gate{ |
| 115 | opts: opts, |
| 116 | tasks: map[string]*taskRuntime{}, |
| 117 | waiters: map[string]chan resolvePayload{}, |
| 118 | taskOf: map[string]string{}, |
| 119 | pending: map[string]PendingProposal{}, |
| 120 | awaiting: map[string]struct{}{}, |
| 121 | episodeSeq: 1, |
| 122 | episodeID: "ep:1", |
| 123 | generation: 1, |
| 124 | persistPending: map[string]int{}, |
| 125 | persistDone: map[string]uint64{}, |
| 126 | } |
| 127 | g.persistCond = sync.NewCond(&g.persistMu) |
| 128 | return g |
| 129 | } |
| 130 | |
| 131 | // EpisodeID returns the current host-owned Recovery Episode id. |
| 132 | func (g *Gate) EpisodeID() string { |
| 133 | if g == nil { |
| 134 | return "" |
| 135 | } |
| 136 | g.mu.Lock() |
| 137 | defer g.mu.Unlock() |
| 138 | return g.episodeID |
| 139 | } |
| 140 | |
| 141 | // Generation returns the current observation/proposal generation. |
| 142 | func (g *Gate) Generation() uint64 { |
| 143 | if g == nil { |
| 144 | return 0 |
| 145 | } |
| 146 | g.mu.Lock() |
| 147 | defer g.mu.Unlock() |
| 148 | return g.generation |
| 149 | } |
| 150 | |
| 151 | // BeginEpisode rotates into a fresh Recovery Episode. Failure, reviewer, and |
| 152 | // stop budgets clear. Explicit task grants and TaskScope authorizations are |
| 153 | // preserved. Call on: real user messages, Plan "start execution", Recovery |
| 154 | // "try another approach", real tool-approval mode changes, and new Session / |
| 155 | // Controller restore. Same-value mode replays must not call this. |
| 156 | func (g *Gate) BeginEpisode() { |
| 157 | if g == nil { |
| 158 | return |
| 159 | } |
| 160 | dismissed := g.beginEpisodeLockedCollect(true) |
| 161 | g.finishDismissed(dismissed) |
| 162 | g.persist() |
| 163 | } |
| 164 | |
| 165 | // OnModeChange rotates Episode and generation when the tool-approval mode |
| 166 | // actually changes. Same-value replays (desktop hydration/reconcile) are no-ops |
| 167 | // so in-flight Auto state is not wiped. Returns dismissed recovery approval ids |
| 168 | // so the controller can clear matching cards outside the gate lock. |
| 169 | func (g *Gate) OnModeChange(mode string) []string { |
| 170 | if g == nil { |
| 171 | return nil |
| 172 | } |
| 173 | mode = strings.ToLower(strings.TrimSpace(mode)) |
| 174 | if mode == "" { |
| 175 | return nil |
| 176 | } |
| 177 | g.mu.Lock() |
| 178 | if g.haveMode && g.lastMode == mode { |
| 179 | g.mu.Unlock() |
| 180 | return nil |
| 181 | } |
| 182 | // First observation only pins the baseline mode (desktop hydrate / initial |
| 183 | // ApplyToolApprovalMode). Same-value later replays are no-ops above; a real |
| 184 | // change rotates Episode and generation. |
| 185 | if !g.haveMode { |
| 186 | g.lastMode = mode |
| 187 | g.haveMode = true |
| 188 | g.mu.Unlock() |
| 189 | return nil |
| 190 | } |
| 191 | g.lastMode = mode |
| 192 | g.metrics.ModeResets++ |
| 193 | dismissed := g.beginEpisodeLockedCollect(false) |
| 194 | // Bump generation even when episode collection already did — mode switch |
| 195 | // must invalidate in-flight observations. |
| 196 | if g.generation == 0 { |
| 197 | g.generation = 1 |
| 198 | } |
| 199 | ids := make([]string, 0, len(dismissed)) |
| 200 | for _, d := range dismissed { |
| 201 | ids = append(ids, d.id) |
| 202 | } |
| 203 | g.mu.Unlock() |
| 204 | g.finishDismissed(dismissed) |
| 205 | g.persist() |
| 206 | return ids |
| 207 | } |
| 208 | |
| 209 | // beginEpisodeLockedCollect must be called with g.mu held when alreadyLocked is |
| 210 | // false it acquires the lock. When alreadyHeld is true, caller holds g.mu. |
| 211 | func (g *Gate) beginEpisodeLockedCollect(lock bool) []dismissedWaiter { |
| 212 | if lock { |
| 213 | g.mu.Lock() |
| 214 | } |
| 215 | g.episodeSeq++ |
| 216 | if g.episodeSeq == 0 { |
| 217 | g.episodeSeq = 1 |
| 218 | } |
| 219 | g.episodeID = fmt.Sprintf("ep:%d", g.episodeSeq) |
| 220 | g.generation++ |
| 221 | if g.generation == 0 { |
| 222 | g.generation = 1 |
| 223 | } |
| 224 | g.metrics.EpisodeRotations++ |
| 225 | // Episode-level hard-stop budgets reset for every TaskID together. |
| 226 | g.episode.clear() |
| 227 | // Clear task-local operation counters; preserve task grants. |
| 228 | for id, st := range g.tasks { |
| 229 | if st == nil { |
| 230 | delete(g.tasks, id) |
| 231 | continue |
| 232 | } |
| 233 | grants := st.taskGrants |
| 234 | grantScope := st.taskGrantScope |
| 235 | st.clearTaskRecoveryState() |
| 236 | st.episodeID = g.episodeID |
| 237 | st.taskGrants = grants |
| 238 | st.taskGrantScope = grantScope |
| 239 | if !st.hasTaskGrants() && st.empty() { |
| 240 | delete(g.tasks, id) |
| 241 | } |
| 242 | } |
| 243 | dismissed := g.collectWaitersLocked(resolvePayload{ |
| 244 | action: ActionRevise, |
| 245 | feedback: "Tool approval mode or recovery episode changed. Re-evaluate under the new mode; the previous proposal was not approved.", |
| 246 | }) |
| 247 | if lock { |
| 248 | g.mu.Unlock() |
| 249 | } |
| 250 | return dismissed |
| 251 | } |
| 252 | |
| 253 | func (g *Gate) collectWaitersLocked(payload resolvePayload) []dismissedWaiter { |
| 254 | out := make([]dismissedWaiter, 0, len(g.waiters)) |
| 255 | for id, ch := range g.waiters { |
| 256 | taskID := g.taskOf[id] |
| 257 | out = append(out, dismissedWaiter{id: id, taskID: taskID, reply: ch, payload: payload}) |
| 258 | delete(g.waiters, id) |
| 259 | delete(g.taskOf, id) |
| 260 | delete(g.pending, id) |
| 261 | delete(g.awaiting, taskID) |
| 262 | } |
| 263 | return out |
| 264 | } |
| 265 | |
| 266 | func (g *Gate) finishDismissed(dismissed []dismissedWaiter) { |
| 267 | for _, d := range dismissed { |
| 268 | if d.reply == nil { |
| 269 | continue |
| 270 | } |
| 271 | select { |
| 272 | case d.reply <- d.payload: |
| 273 | default: |
| 274 | } |
| 275 | } |
| 276 | } |
| 277 | |
| 278 | // Metrics returns a copy of content-free counters accumulated since gate |
| 279 | // construction or the most recent DrainMetrics call. |
| 280 | func (g *Gate) Metrics() Metrics { |
| 281 | if g == nil { |
| 282 | return Metrics{} |
| 283 | } |
| 284 | g.mu.Lock() |
| 285 | defer g.mu.Unlock() |
| 286 | return g.metrics |
| 287 | } |
| 288 | |
| 289 | // DrainMetrics atomically returns and clears recovery counters accumulated |
| 290 | // since the last drain. Desktop telemetry uses this delta API at TurnDone so a |
| 291 | // historical event is never counted again on later turns. |
| 292 | func (g *Gate) DrainMetrics() Metrics { |
| 293 | if g == nil { |
| 294 | return Metrics{} |
| 295 | } |
| 296 | g.mu.Lock() |
| 297 | defer g.mu.Unlock() |
| 298 | out := g.metrics |
| 299 | g.metrics = Metrics{} |
| 300 | return out |
| 301 | } |
| 302 | |
| 303 | // FlushPersistence waits until every snapshot already scheduled for key has |
| 304 | // finished. Session destruction uses this before removing sidecars so a late |
| 305 | // asynchronous write cannot resurrect an artifact that was just deleted. |
| 306 | func (g *Gate) FlushPersistence(key string) { |
| 307 | if g == nil || g.opts.Persist == nil { |
| 308 | return |
| 309 | } |
| 310 | g.persistMu.Lock() |
| 311 | for g.persistPending[key] > 0 { |
| 312 | g.persistCond.Wait() |
| 313 | } |
| 314 | g.persistMu.Unlock() |
| 315 | } |
| 316 | |
| 317 | // HasApproval reports whether a live Auto decision waiter is parked under id. |
| 318 | // Unlike Snapshot, this includes normal-execution plan transitions that have a |
| 319 | // waiter but no armed failure/taskRuntime yet. Legacy Approve paths must use |
| 320 | // this (or Resolve) instead of inferring from a persistence snapshot. |
| 321 | func (g *Gate) HasApproval(id string) bool { |
| 322 | if g == nil { |
| 323 | return false |
| 324 | } |
| 325 | id = strings.TrimSpace(id) |
| 326 | if id == "" || strings.HasPrefix(id, "pending:") { |
| 327 | return false |
| 328 | } |
| 329 | g.mu.Lock() |
| 330 | defer g.mu.Unlock() |
| 331 | if _, ok := g.waiters[id]; ok { |
| 332 | return true |
| 333 | } |
| 334 | _, ok := g.taskOf[id] |
| 335 | return ok |
| 336 | } |
| 337 | |
| 338 | // Snapshot returns a live debug copy of task state (may include budgets). |
| 339 | func (g *Gate) Snapshot() Snapshot { |
| 340 | if g == nil { |
| 341 | return Snapshot{} |
| 342 | } |
| 343 | g.mu.Lock() |
| 344 | defer g.mu.Unlock() |
| 345 | return g.snapshotLocked(false) |
| 346 | } |
| 347 | |
| 348 | // PersistenceSnapshot returns the disk projection: historical last_failure |
| 349 | // evidence only. Active locks, Episode counters, generation, and waiters never |
| 350 | // appear. |
| 351 | func (g *Gate) PersistenceSnapshot() Snapshot { |
| 352 | if g == nil { |
| 353 | return Snapshot{} |
| 354 | } |
| 355 | g.mu.Lock() |
| 356 | defer g.mu.Unlock() |
| 357 | return g.snapshotLocked(true) |
| 358 | } |
| 359 | |
| 360 | func (g *Gate) snapshotLocked(persistence bool) Snapshot { |
| 361 | // Map task id -> live approval id for observability. Restore always drops |
| 362 | // these fields so a restart never replays a transient authorization. |
| 363 | approvalByTask := map[string]string{} |
| 364 | if !persistence { |
| 365 | for approvalID, taskID := range g.taskOf { |
| 366 | if strings.HasPrefix(approvalID, "pending:") { |
| 367 | continue |
| 368 | } |
| 369 | approvalByTask[taskID] = approvalID |
| 370 | } |
| 371 | } |
| 372 | out := Snapshot{Tasks: map[string]*TaskState{}} |
| 373 | for id, st := range g.tasks { |
| 374 | var cp *TaskState |
| 375 | if persistence { |
| 376 | cp = st.toPersistenceState() |
| 377 | } else { |
| 378 | phase := PhaseDiagnosing |
| 379 | if _, waiting := g.awaiting[id]; waiting { |
| 380 | phase = PhaseAwaitingDecision |
| 381 | } |
| 382 | cp = st.toTaskState(phase) |
| 383 | } |
| 384 | if cp == nil { |
| 385 | continue |
| 386 | } |
| 387 | if !persistence { |
| 388 | if aid := approvalByTask[id]; aid != "" { |
| 389 | cp.ApprovalID = aid |
| 390 | cp.Phase = PhaseAwaitingDecision |
| 391 | } |
| 392 | // Project shared Episode budgets onto each task for live debug views. |
| 393 | cp.ReviewBlocks = int(g.episode.reviewRejects) |
| 394 | cp.EpisodeStopped = g.episode.stopped |
| 395 | cp.StopReason = string(g.episode.stopReason) |
| 396 | if cp.EpisodeID == "" { |
| 397 | cp.EpisodeID = g.episodeID |
| 398 | } |
| 399 | } |
| 400 | out.Tasks[id] = cp |
| 401 | } |
| 402 | return out |
| 403 | } |
| 404 | |
| 405 | // Restore loads persisted failure evidence after restart/controller rebuild. |
| 406 | // Live prompts, budgets, Episode counters, and task-local grants are never |
| 407 | // replayed: old consecutive_fails / review_blocks become historical evidence |
| 408 | // only and do not re-arm locks. |
| 409 | func (g *Gate) Restore(snap Snapshot) { |
| 410 | if g == nil { |
| 411 | return |
| 412 | } |
| 413 | g.mu.Lock() |
| 414 | defer g.mu.Unlock() |
| 415 | g.tasks = map[string]*taskRuntime{} |
| 416 | g.waiters = map[string]chan resolvePayload{} |
| 417 | g.taskOf = map[string]string{} |
| 418 | g.pending = map[string]PendingProposal{} |
| 419 | g.awaiting = map[string]struct{}{} |
| 420 | // Fresh Episode after restore/session switch so prior runtime budgets |
| 421 | // cannot block the user. |
| 422 | g.episodeSeq++ |
| 423 | if g.episodeSeq == 0 { |
| 424 | g.episodeSeq = 1 |
| 425 | } |
| 426 | g.episodeID = fmt.Sprintf("ep:%d", g.episodeSeq) |
| 427 | g.generation++ |
| 428 | if g.generation == 0 { |
| 429 | g.generation = 1 |
| 430 | } |
| 431 | g.episode.clear() |
| 432 | g.metrics.EpisodeRotations++ |
| 433 | for id, st := range snap.Tasks { |
| 434 | rt := taskRuntimeFromState(st) |
| 435 | if rt == nil { |
| 436 | continue |
| 437 | } |
| 438 | rt.episodeID = g.episodeID |
| 439 | // Ignore old Pending and ApprovalID — never restore as authorization. |
| 440 | g.tasks[id] = rt |
| 441 | } |
| 442 | } |
| 443 | |
| 444 | // BindApprovalID associates a prompt id with the task waiting on it so |
| 445 | // Resolve can find the waiter after EmitPrompt returns. If a provisional |
| 446 | // waiter is parked under pending:<taskID>, it is re-keyed to approvalID. |
| 447 | func (g *Gate) BindApprovalID(taskID, approvalID string) { |
| 448 | if g == nil { |
| 449 | return |
| 450 | } |
| 451 | taskID = normalizeTaskID(taskID) |
| 452 | approvalID = strings.TrimSpace(approvalID) |
| 453 | if approvalID == "" { |
| 454 | return |
| 455 | } |
| 456 | g.mu.Lock() |
| 457 | defer g.mu.Unlock() |
| 458 | provisional := "pending:" + taskID |
| 459 | if ch := g.waiters[provisional]; ch != nil { |
| 460 | delete(g.waiters, provisional) |
| 461 | delete(g.taskOf, provisional) |
| 462 | g.waiters[approvalID] = ch |
| 463 | } |
| 464 | if pending, ok := g.pending[provisional]; ok { |
| 465 | delete(g.pending, provisional) |
| 466 | g.pending[approvalID] = pending |
| 467 | } |
| 468 | g.taskOf[approvalID] = taskID |
| 469 | g.awaiting[taskID] = struct{}{} |
| 470 | } |
| 471 | |
| 472 | // Resolve applies a user decision to a pending Auto Guard approval. |
| 473 | // action is continue|continue_task|revise. For revise, feedback is returned through the |
| 474 | // blocked tool result and the current mutation is refused in the same operation. |
| 475 | func (g *Gate) Resolve(id string, action Action, feedback string) error { |
| 476 | if g == nil { |
| 477 | return fmt.Errorf("recovery gate is nil") |
| 478 | } |
| 479 | id = strings.TrimSpace(id) |
| 480 | g.mu.Lock() |
| 481 | ch := g.waiters[id] |
| 482 | taskID := g.taskOf[id] |
| 483 | pending := g.pending[id] |
| 484 | if taskID == "" { |
| 485 | g.mu.Unlock() |
| 486 | return fmt.Errorf("unknown recovery approval %q", id) |
| 487 | } |
| 488 | st := g.tasks[taskID] |
| 489 | rotateEpisode := false |
| 490 | switch action { |
| 491 | case ActionContinue, ActionContinueTask: |
| 492 | if action == ActionContinueTask { |
| 493 | if pending.TaskGrantKey == "" { |
| 494 | g.mu.Unlock() |
| 495 | return fmt.Errorf("recovery approval %q cannot grant similar actions", id) |
| 496 | } |
| 497 | if st == nil { |
| 498 | st = &taskRuntime{episodeID: g.episodeID} |
| 499 | g.tasks[taskID] = st |
| 500 | } |
| 501 | st.useTaskGrantScope(pending.TaskGrantTaskScope) |
| 502 | st.addTaskGrant(pending.TaskGrantKey) |
| 503 | g.metrics.TaskGrantContinues++ |
| 504 | } |
| 505 | // Human continue does not reset Episode reviewer rejects; only real |
| 506 | // mutation/verification progress, a new Episode, or revise does. |
| 507 | g.metrics.HumanContinues++ |
| 508 | case ActionRevise: |
| 509 | // Revise rejects the pending action and starts a fresh Recovery Episode |
| 510 | // so alternative approaches get a clean budget. |
| 511 | rotateEpisode = true |
| 512 | g.metrics.HumanRevises++ |
| 513 | if strings.TrimSpace(feedback) == "" { |
| 514 | feedback = DefaultReviseFeedback |
| 515 | } |
| 516 | default: |
| 517 | g.mu.Unlock() |
| 518 | return fmt.Errorf("unknown recovery action %q", action) |
| 519 | } |
| 520 | delete(g.waiters, id) |
| 521 | delete(g.taskOf, id) |
| 522 | delete(g.pending, id) |
| 523 | delete(g.awaiting, taskID) |
| 524 | if !rotateEpisode { |
| 525 | if st == nil || (st.empty() && !st.hasTaskGrants()) { |
| 526 | delete(g.tasks, taskID) |
| 527 | } |
| 528 | } |
| 529 | g.mu.Unlock() |
| 530 | |
| 531 | if ch != nil { |
| 532 | select { |
| 533 | case ch <- resolvePayload{action: action, feedback: feedback}: |
| 534 | default: |
| 535 | } |
| 536 | } |
| 537 | if rotateEpisode { |
| 538 | // Fresh Episode after "try another approach" so alternatives get a clean |
| 539 | // budget. BeginEpisode also dismisses any other waiters safely. |
| 540 | g.BeginEpisode() |
| 541 | } else { |
| 542 | g.persist() |
| 543 | } |
| 544 | return nil |
| 545 | } |
| 546 | |
| 547 | // ObserveResult implements agent.RecoveryGate. It returns one-shot guidance |
| 548 | // for the caller to enqueue on the exact Agent.Run that observed the failure. |
| 549 | func (g *Gate) ObserveResult(_ context.Context, obs Observation) string { |
| 550 | if g == nil || !g.activeMode() { |
| 551 | return "" |
| 552 | } |
| 553 | taskID := normalizeTaskID(obs.TaskID) |
| 554 | |
| 555 | g.mu.Lock() |
| 556 | defer g.mu.Unlock() |
| 557 | |
| 558 | // Stale observations from a previous generation (mode switch / episode |
| 559 | // rotate mid-flight) are ignored so they cannot re-arm old locks. |
| 560 | if obs.Generation != 0 && obs.Generation != g.generation { |
| 561 | g.metrics.StaleObservationsIgnored++ |
| 562 | return "" |
| 563 | } |
| 564 | |
| 565 | st := g.ensureTaskLocked(taskID) |
| 566 | |
| 567 | // Successful host-recognized verification clears Episode no-progress budgets. |
| 568 | if obs.Success && obs.Verification { |
| 569 | g.clearNoProgressLocked(taskID, st) |
| 570 | g.persistUnlocked() |
| 571 | return "" |
| 572 | } |
| 573 | // Any successful mutation ends the current no-progress budget. |
| 574 | if obs.Success && obs.Mutates { |
| 575 | g.clearNoProgressLocked(taskID, st) |
| 576 | g.persistUnlocked() |
| 577 | return "" |
| 578 | } |
| 579 | // Diagnostic read successes do not clear failure state. Preserve a bounded |
| 580 | // evidence excerpt for the isolated reviewer; otherwise it sees the failure |
| 581 | // and proposed diff but none of the investigation that connected them. |
| 582 | if obs.Success { |
| 583 | if st.lastFailure != nil && IsDiagnosticSuccess(obs) { |
| 584 | if appendDiagnosisNote(st.lastFailure, diagnosticObservationNote(obs)) { |
| 585 | g.persistUnlocked() |
| 586 | } |
| 587 | } |
| 588 | return "" |
| 589 | } |
| 590 | if !QualifyingFailure(obs) { |
| 591 | return "" |
| 592 | } |
| 593 | |
| 594 | fp := observationFingerprint(obs) |
| 595 | st.ensureMaps() |
| 596 | st.episodeID = g.episodeID |
| 597 | if st.operationFailures[fp] < 255 { |
| 598 | st.operationFailures[fp]++ |
| 599 | } |
| 600 | // Episode totals accumulate across every TaskID (root + sub-agents). |
| 601 | if g.episode.totalFailures < 255 { |
| 602 | g.episode.totalFailures++ |
| 603 | } |
| 604 | if st.operationFailures[fp] >= MaxOperationFailures { |
| 605 | st.markOperationStopped(fp) |
| 606 | g.metrics.OperationStops++ |
| 607 | } |
| 608 | if g.episode.totalFailures >= MaxEpisodeFailures { |
| 609 | g.episode.stopped = true |
| 610 | g.episode.stopReason = StopReasonEpisodeFailures |
| 611 | g.metrics.EpisodeFailureStops++ |
| 612 | } |
| 613 | |
| 614 | st.lastFailure = &activeFailure{ |
| 615 | evidence: FailureEvent{ |
| 616 | Class: ClassifyFailure(obs), |
| 617 | Tool: obs.Tool, |
| 618 | ArgsSummary: ArgsSummary(obs.Args, 200), |
| 619 | Subject: obs.Subject, |
| 620 | ErrSummary: obs.ErrSummary, |
| 621 | OutputExcerpt: clip(obs.Output, 1500), |
| 622 | SourceAgent: obs.AgentID, |
| 623 | TaskID: taskID, |
| 624 | TaskScopeID: persistentRecoveryScope(obs.TaskScopeID), |
| 625 | ReadOnly: obs.ReadOnly, |
| 626 | Verification: obs.Verification, |
| 627 | Mutates: obs.Mutates, |
| 628 | CreatedAt: g.opts.Now(), |
| 629 | Args: append(json.RawMessage(nil), obs.Args...), |
| 630 | Fingerprint: fp, |
| 631 | }, |
| 632 | safeRetryUsed: false, |
| 633 | } |
| 634 | // Keep diagnosis notes if same fingerprint; otherwise start fresh list. |
| 635 | g.metrics.FailureEvents++ |
| 636 | guidance := g.recoveryGuidanceLocked(st) |
| 637 | g.persistUnlocked() |
| 638 | return guidance |
| 639 | } |
| 640 | |
| 641 | // BeforeMutation implements agent.RecoveryGate. |
| 642 | func (g *Gate) BeforeMutation(ctx context.Context, proposal Proposal) (Decision, error) { |
| 643 | if g == nil { |
| 644 | return Decision{Allow: true}, nil |
| 645 | } |
| 646 | |
| 647 | // Host-proven read-only diagnostics always continue, including after the |
| 648 | // Episode execution budget is exhausted. Decide also encodes the non-Auto |
| 649 | // bypass so Ask and YOLO keep their existing semantics. |
| 650 | facts, failure, diagNotes, taskID, fp, gen := g.classify(proposal) |
| 651 | route := Decide(facts) |
| 652 | |
| 653 | // Escalation: re-proposing an already-stopped operation burns the |
| 654 | // stopped-op retry budget and may stop the whole turn. |
| 655 | if facts.AutoMode && facts.OperationAlreadyStopped && facts.SameFailedOperation { |
| 656 | dec, escalated := g.noteStoppedOpRetry(taskID, fp, gen, proposal) |
| 657 | if escalated { |
| 658 | return dec, nil |
| 659 | } |
| 660 | // Still under retry budget: fall through to RouteStop for this op. |
| 661 | route = DecisionResult{Route: RouteStop, StopReason: StopReasonOperationFailures} |
| 662 | } |
| 663 | |
| 664 | // Episode total failure hard stop before reviewer work. |
| 665 | if facts.AutoMode && facts.EpisodeFailureCount >= MaxEpisodeFailures && (facts.Mutates || facts.Verification) { |
| 666 | return g.stopTurnDecision(taskID, gen, StopReasonEpisodeFailures, proposal), nil |
| 667 | } |
| 668 | |
| 669 | switch route.Route { |
| 670 | case RouteBypass, RouteAllow: |
| 671 | if route.ConsumeSafeRetry { |
| 672 | g.mu.Lock() |
| 673 | if st := g.tasks[taskID]; st != nil && st.lastFailure != nil && !st.lastFailure.safeRetryUsed { |
| 674 | st.lastFailure.safeRetryUsed = true |
| 675 | g.metrics.RuleContinues++ |
| 676 | } |
| 677 | g.mu.Unlock() |
| 678 | g.persist() |
| 679 | } |
| 680 | return Decision{Allow: true, Generation: gen}, nil |
| 681 | case RouteReview: |
| 682 | return g.reviewOrEscalate(ctx, taskID, fp, gen, proposal, failure, diagNotes) |
| 683 | case RouteStop: |
| 684 | return Decision{ |
| 685 | Allow: false, |
| 686 | Blocked: true, |
| 687 | Message: repeatedFailureStopMessage(int(facts.FailureCount), proposal), |
| 688 | Generation: gen, |
| 689 | StopReason: string(StopReasonOperationFailures), |
| 690 | }, nil |
| 691 | case RouteStopTurn: |
| 692 | return g.stopTurnDecision(taskID, gen, route.StopReason, proposal), nil |
| 693 | default: |
| 694 | return Decision{Allow: true, Generation: gen}, nil |
| 695 | } |
| 696 | } |
| 697 | |
| 698 | func (g *Gate) noteStoppedOpRetry(taskID, _ string, gen uint64, proposal Proposal) (Decision, bool) { |
| 699 | g.mu.Lock() |
| 700 | _ = g.ensureTaskLocked(taskID) |
| 701 | if g.episode.stoppedOpRetries < 255 { |
| 702 | g.episode.stoppedOpRetries++ |
| 703 | } |
| 704 | retries := g.episode.stoppedOpRetries |
| 705 | if retries >= MaxStoppedOperationRetries { |
| 706 | g.episode.stopped = true |
| 707 | if g.episode.stopReason == StopReasonNone { |
| 708 | g.episode.stopReason = StopReasonStoppedOpRetries |
| 709 | } |
| 710 | g.metrics.StoppedOpRetryStops++ |
| 711 | g.mu.Unlock() |
| 712 | g.persist() |
| 713 | return g.stopTurnDecision(taskID, gen, StopReasonStoppedOpRetries, proposal), true |
| 714 | } |
| 715 | g.mu.Unlock() |
| 716 | g.persist() |
| 717 | return Decision{}, false |
| 718 | } |
| 719 | |
| 720 | func (g *Gate) stopTurnDecision(taskID string, gen uint64, reason StopReason, proposal Proposal) Decision { |
| 721 | g.mu.Lock() |
| 722 | _ = g.ensureTaskLocked(taskID) |
| 723 | g.episode.stopped = true |
| 724 | if g.episode.stopReason == StopReasonNone { |
| 725 | g.episode.stopReason = reason |
| 726 | } |
| 727 | stopReason := g.episode.stopReason |
| 728 | g.mu.Unlock() |
| 729 | g.persist() |
| 730 | msg := episodeStopMessage(stopReason, proposal) |
| 731 | return Decision{ |
| 732 | Allow: false, |
| 733 | Blocked: true, |
| 734 | Message: msg, |
| 735 | Generation: gen, |
| 736 | StopTurn: true, |
| 737 | StopReason: string(stopReason), |
| 738 | } |
| 739 | } |
| 740 | |
| 741 | // MarkFinalizationOffered records that the agent was given its one summarize-only |
| 742 | // round after an Episode stop. Subsequent tool proposals while still stopped |
| 743 | // should surface RecoveryPauseError. |
| 744 | func (g *Gate) MarkFinalizationOffered(taskID string) { |
| 745 | if g == nil { |
| 746 | return |
| 747 | } |
| 748 | _ = taskID |
| 749 | g.mu.Lock() |
| 750 | defer g.mu.Unlock() |
| 751 | if g.episode.stopped { |
| 752 | g.episode.finalizationOffered = true |
| 753 | } |
| 754 | } |
| 755 | |
| 756 | // ConsumeFinalization reports whether the finalization round already ran and |
| 757 | // the model still attempted tools. Also marks it consumed on first true check |
| 758 | // after offered. Finalization is Episode-scoped (shared by all TaskIDs). |
| 759 | func (g *Gate) ConsumeFinalization(taskID string) (offered, alreadyConsumed bool) { |
| 760 | if g == nil { |
| 761 | return false, false |
| 762 | } |
| 763 | _ = taskID |
| 764 | g.mu.Lock() |
| 765 | defer g.mu.Unlock() |
| 766 | if !g.episode.stopped { |
| 767 | return false, false |
| 768 | } |
| 769 | offered = g.episode.finalizationOffered |
| 770 | alreadyConsumed = g.episode.finalizationConsumed |
| 771 | if offered && !alreadyConsumed { |
| 772 | g.episode.finalizationConsumed = true |
| 773 | } |
| 774 | return offered, alreadyConsumed |
| 775 | } |
| 776 | |
| 777 | // EpisodeStopped reports whether the shared Recovery Episode is exhausted for |
| 778 | // any TaskID (root or sub-agent). |
| 779 | func (g *Gate) EpisodeStopped(taskID string) bool { |
| 780 | if g == nil { |
| 781 | return false |
| 782 | } |
| 783 | _ = taskID |
| 784 | g.mu.Lock() |
| 785 | defer g.mu.Unlock() |
| 786 | return g.episode.stopped |
| 787 | } |
| 788 | |
| 789 | // clearNoProgressLocked clears Episode totals and the observing task's local |
| 790 | // counters after real mutation/verification success. Caller holds g.mu. |
| 791 | func (g *Gate) clearNoProgressLocked(taskID string, st *taskRuntime) { |
| 792 | g.episode.clear() |
| 793 | if st != nil { |
| 794 | st.clearTaskRecoveryState() |
| 795 | st.episodeID = g.episodeID |
| 796 | if !st.hasTaskGrants() { |
| 797 | delete(g.tasks, taskID) |
| 798 | } |
| 799 | } |
| 800 | } |
| 801 | |
| 802 | // classify builds pure Facts for Decide. It never calls the model or UI. |
| 803 | func (g *Gate) classify(proposal Proposal) (Facts, *FailureEvent, []string, string, string, uint64) { |
| 804 | facts := Facts{ |
| 805 | AutoMode: g.activeMode(), |
| 806 | ReadOnly: proposal.ReadOnly, |
| 807 | Mutates: proposal.Mutates, |
| 808 | Verification: proposal.Verification, |
| 809 | PlanTransition: proposal.PlanTransition, |
| 810 | } |
| 811 | // Deterministic boundary checks run before the failure-recovery path. |
| 812 | boundary := riskBoundaryForProposal(proposal) |
| 813 | proposal.HighRisk = boundary.highRisk |
| 814 | facts.HighRisk = boundary.highRisk |
| 815 | |
| 816 | taskID := normalizeTaskID(proposal.TaskID) |
| 817 | // Operation failure accounting intentionally excludes Preview. Agent calls |
| 818 | // always carry a display/approval preview, while completed observations do |
| 819 | // not; mixing the two shapes would make an exact retry look like an unseen |
| 820 | // operation and bypass its three-failure stop. Keep the preview-bound |
| 821 | // fingerprint for one-shot human approval below. |
| 822 | operationFP := CallFingerprint(proposal.Tool, proposal.Subject, "", proposal.Args) |
| 823 | approvalFP := CallFingerprint(proposal.Tool, proposal.Subject, proposal.Preview, proposal.Args) |
| 824 | |
| 825 | g.mu.Lock() |
| 826 | gen := g.generation |
| 827 | // Leaving Auto does not wait for the next proposal: OnModeChange handles |
| 828 | // real mode switches. Here we still clear when mode is non-Auto so a |
| 829 | // bypass path cannot keep armed Auto locks if OnModeChange was skipped. |
| 830 | st := g.tasks[taskID] |
| 831 | var failure *FailureEvent |
| 832 | var diagNotes []string |
| 833 | stateChanged := false |
| 834 | // Shared Episode budget applies even when this TaskID has no local state. |
| 835 | facts.EpisodeStopped = g.episode.stopped |
| 836 | facts.StopReason = g.episode.stopReason |
| 837 | facts.EpisodeFailureCount = g.episode.totalFailures |
| 838 | facts.ReviewRejects = g.episode.reviewRejects |
| 839 | if st != nil && !facts.AutoMode { |
| 840 | if !st.empty() || g.episode.totalFailures > 0 || g.episode.reviewRejects > 0 || g.episode.stopped { |
| 841 | st.clearTaskRecoveryState() |
| 842 | g.episode.clear() |
| 843 | stateChanged = true |
| 844 | } |
| 845 | if !st.hasTaskGrants() && st.empty() { |
| 846 | delete(g.tasks, taskID) |
| 847 | st = nil |
| 848 | } |
| 849 | } |
| 850 | if st != nil { |
| 851 | // Align task runtime with current Episode without wiping mid-Episode. |
| 852 | if st.episodeID != "" && st.episodeID != g.episodeID { |
| 853 | grants := st.taskGrants |
| 854 | grantScope := st.taskGrantScope |
| 855 | st.clearTaskRecoveryState() |
| 856 | st.taskGrants = grants |
| 857 | st.taskGrantScope = grantScope |
| 858 | st.episodeID = g.episodeID |
| 859 | stateChanged = true |
| 860 | } else if st.episodeID == "" { |
| 861 | st.episodeID = g.episodeID |
| 862 | } |
| 863 | facts.OperationAlreadyStopped = st.isOperationStopped(operationFP) |
| 864 | facts.FailureCount = st.operationFailureCount(operationFP) |
| 865 | if st.lastFailure != nil { |
| 866 | failure = st.evidenceCopy() |
| 867 | diagNotes = st.diagnosisNotes() |
| 868 | facts.HasActiveFailure = true |
| 869 | facts.SameFailedOperation = sameFailedOperation(failure, proposal) |
| 870 | // When proposing the same op, FailureCount is the map value. |
| 871 | // When proposing a different op after failures, HasActiveFailure |
| 872 | // remains true for accounting, but Decide keeps that unrelated |
| 873 | // operation on the automatic path. |
| 874 | if facts.SameFailedOperation && facts.FailureCount == 0 { |
| 875 | // Evidence exists but count was cleared somehow — treat as 1. |
| 876 | facts.FailureCount = 1 |
| 877 | } |
| 878 | if IsSafeVerificationRetry(failure, proposal) && st.safeRetryAvailable() { |
| 879 | facts.SafeRetryAvailable = true |
| 880 | } |
| 881 | } |
| 882 | taskScope := taskGrantScopeKey(proposal) |
| 883 | st.useTaskGrantScope(taskScope) |
| 884 | if st.empty() && !st.hasTaskGrants() { |
| 885 | delete(g.tasks, taskID) |
| 886 | st = nil |
| 887 | } |
| 888 | runtimeGrantKey := taskGrantRuntimeKey(boundary.taskGrantKey, taskScope) |
| 889 | if facts.HighRisk && runtimeGrantKey != "" && st != nil && st.hasTaskGrant(runtimeGrantKey) { |
| 890 | facts.HighRisk = false |
| 891 | g.metrics.TaskGrantUses++ |
| 892 | } |
| 893 | } |
| 894 | g.mu.Unlock() |
| 895 | if stateChanged { |
| 896 | g.persist() |
| 897 | } |
| 898 | |
| 899 | if failure != nil { |
| 900 | if !proposal.ExpandedScope { |
| 901 | proposal.ExpandedScope = ScopeExpanded(failure, proposal) |
| 902 | } |
| 903 | if !proposal.StrategyChanged { |
| 904 | proposal.StrategyChanged = StrategyChanged(failure, proposal) |
| 905 | } |
| 906 | facts.ExpandedScope = proposal.ExpandedScope |
| 907 | facts.StrategyChanged = proposal.StrategyChanged |
| 908 | if facts.SafeRetryAvailable && (facts.ExpandedScope || facts.StrategyChanged || facts.HighRisk) { |
| 909 | facts.SafeRetryAvailable = false |
| 910 | } |
| 911 | } |
| 912 | return facts, failure, diagNotes, taskID, approvalFP, gen |
| 913 | } |
| 914 | |
| 915 | func (g *Gate) ensureTaskLocked(taskID string) *taskRuntime { |
| 916 | st := g.tasks[taskID] |
| 917 | if st == nil { |
| 918 | st = &taskRuntime{episodeID: g.episodeID} |
| 919 | g.tasks[taskID] = st |
| 920 | } |
| 921 | if st.episodeID == "" { |
| 922 | st.episodeID = g.episodeID |
| 923 | } |
| 924 | return st |
| 925 | } |
| 926 | |
| 927 | func (g *Gate) reviewOrEscalate(ctx context.Context, taskID, fp string, gen uint64, proposal Proposal, failure *FailureEvent, diagNotes []string) (Decision, error) { |
| 928 | // If Episode reviewer budget already exhausted, stop the turn. |
| 929 | g.mu.Lock() |
| 930 | if g.episode.reviewRejects >= uint8(g.opts.MaxReviewBlocks) { |
| 931 | g.episode.stopped = true |
| 932 | if g.episode.stopReason == StopReasonNone { |
| 933 | g.episode.stopReason = StopReasonReviewRejects |
| 934 | } |
| 935 | g.metrics.ReviewStops++ |
| 936 | g.mu.Unlock() |
| 937 | return g.stopTurnDecision(taskID, gen, StopReasonReviewRejects, proposal), nil |
| 938 | } |
| 939 | g.mu.Unlock() |
| 940 | |
| 941 | var verdict ReviewVerdict |
| 942 | if g.opts.Reviewer != nil { |
| 943 | start := g.opts.Now() |
| 944 | taskSummary := strings.TrimSpace(proposal.TaskSummary) |
| 945 | if taskSummary == "" && g.opts.TaskSummary != nil { |
| 946 | taskSummary = g.opts.TaskSummary() |
| 947 | } |
| 948 | v, err := g.opts.Reviewer.Review(ctx, failure, diagNotes, proposal, taskSummary) |
| 949 | latency := g.opts.Now().Sub(start).Milliseconds() |
| 950 | g.mu.Lock() |
| 951 | g.metrics.ReviewLatencyMsSum += latency |
| 952 | g.metrics.ReviewLatencyCount++ |
| 953 | if err != nil { |
| 954 | g.metrics.ReviewErrors++ |
| 955 | } |
| 956 | g.mu.Unlock() |
| 957 | if err != nil { |
| 958 | if proposal.PlanTransition { |
| 959 | return g.askHuman(ctx, taskID, fp, gen, proposal, failure, diagNotes, ChangeScope, |
| 960 | "The active execution plan changed, but the independent plan reviewer is unavailable.") |
| 961 | } |
| 962 | g.mu.Lock() |
| 963 | g.metrics.RuleContinues++ |
| 964 | g.mu.Unlock() |
| 965 | return Decision{Allow: true, Generation: gen}, nil |
| 966 | } |
| 967 | verdict = normalizeVerdict(v, failure, proposal, diagNotes) |
| 968 | if verdict.Outcome == ReviewContinue && reviewerContinueKind(verdict.ChangeKind) { |
| 969 | // Reviewer Continue does NOT reset cumulative rejects. Only real |
| 970 | // mutation/verification success, a new Episode, or revise does. |
| 971 | g.mu.Lock() |
| 972 | g.metrics.ReviewContinues++ |
| 973 | g.mu.Unlock() |
| 974 | return Decision{ |
| 975 | Allow: true, |
| 976 | AuthorizePlanReplacement: proposal.PlanTransition, |
| 977 | Generation: gen, |
| 978 | }, nil |
| 979 | } |
| 980 | if proposal.PlanTransition && reviewerPlanDecision(verdict) { |
| 981 | return g.askHuman(ctx, taskID, fp, gen, proposal, failure, diagNotes, verdict.ChangeKind, verdict.Rationale) |
| 982 | } |
| 983 | blocks := g.recordReviewBlock(taskID, verdict) |
| 984 | if blocks < g.opts.MaxReviewBlocks { |
| 985 | return Decision{ |
| 986 | Allow: false, |
| 987 | Blocked: true, |
| 988 | Message: reviewerBlockerMessage(verdict, blocks, g.opts.MaxReviewBlocks), |
| 989 | Generation: gen, |
| 990 | }, nil |
| 991 | } |
| 992 | g.mu.Lock() |
| 993 | g.episode.stopped = true |
| 994 | if g.episode.stopReason == StopReasonNone { |
| 995 | g.episode.stopReason = StopReasonReviewRejects |
| 996 | } |
| 997 | g.metrics.ReviewStops++ |
| 998 | g.mu.Unlock() |
| 999 | return g.stopTurnDecision(taskID, gen, StopReasonReviewRejects, proposal), nil |
| 1000 | } |
| 1001 | if proposal.PlanTransition { |
| 1002 | return g.askHuman(ctx, taskID, fp, gen, proposal, failure, diagNotes, ChangeScope, |
| 1003 | "The active execution plan changed and needs your choice because no independent plan reviewer is configured.") |
| 1004 | } |
| 1005 | g.mu.Lock() |
| 1006 | g.metrics.RuleContinues++ |
| 1007 | g.mu.Unlock() |
| 1008 | return Decision{Allow: true, Generation: gen}, nil |
| 1009 | } |
| 1010 | |
| 1011 | func (g *Gate) askHuman(ctx context.Context, taskID, fp string, gen uint64, proposal Proposal, failure *FailureEvent, diagNotes []string, kind ChangeKind, rationale string) (Decision, error) { |
| 1012 | failureSource := "" |
| 1013 | failureSummary := "" |
| 1014 | if failure != nil { |
| 1015 | failureSource = failure.SourceAgent |
| 1016 | failureSummary = failure.ErrSummary |
| 1017 | } |
| 1018 | pending := PendingProposal{ |
| 1019 | Tool: proposal.Tool, |
| 1020 | Subject: proposal.Subject, |
| 1021 | Preview: proposal.Preview, |
| 1022 | Args: append(json.RawMessage(nil), proposal.Args...), |
| 1023 | Fingerprint: fp, |
| 1024 | SourceAgent: firstNonEmpty(proposal.AgentID, failureSource), |
| 1025 | ChangeKind: kind, |
| 1026 | Rationale: firstNonEmpty(rationale, userFacingReason(kind)), |
| 1027 | Diagnosis: strings.Join(diagNotes, "\n"), |
| 1028 | Failure: failureSummary, |
| 1029 | Proposed: firstNonEmpty(proposal.Subject, proposal.Preview, proposal.Tool), |
| 1030 | PlanBefore: proposal.PlanBefore, |
| 1031 | PlanAfter: proposal.PlanAfter, |
| 1032 | } |
| 1033 | |
| 1034 | if g.opts.Headless || g.opts.EmitPrompt == nil { |
| 1035 | return Decision{ |
| 1036 | Allow: false, |
| 1037 | Blocked: true, |
| 1038 | Message: headlessBlockerMessage(pending, failure), |
| 1039 | Generation: gen, |
| 1040 | }, nil |
| 1041 | } |
| 1042 | |
| 1043 | // Create the waiter channel before EmitPrompt. Resolve may race in as soon |
| 1044 | // as the approval id is known (desktop/bot), so re-key the waiter under the |
| 1045 | // real id immediately after EmitPrompt returns. |
| 1046 | reply := make(chan resolvePayload, 1) |
| 1047 | g.mu.Lock() |
| 1048 | g.metrics.HumanPrompts++ |
| 1049 | if st := g.tasks[taskID]; st != nil && st.failureCount() > 1 { |
| 1050 | g.metrics.RepeatPrompts++ |
| 1051 | } |
| 1052 | provisional := "pending:" + taskID |
| 1053 | g.waiters[provisional] = reply |
| 1054 | g.taskOf[provisional] = taskID |
| 1055 | g.pending[provisional] = pending |
| 1056 | g.awaiting[taskID] = struct{}{} |
| 1057 | g.mu.Unlock() |
| 1058 | |
| 1059 | approvalID, err := g.opts.EmitPrompt(ctx, taskID, pending, failure) |
| 1060 | if err != nil { |
| 1061 | g.mu.Lock() |
| 1062 | delete(g.waiters, provisional) |
| 1063 | delete(g.taskOf, provisional) |
| 1064 | delete(g.pending, provisional) |
| 1065 | delete(g.awaiting, taskID) |
| 1066 | g.mu.Unlock() |
| 1067 | return Decision{Allow: false, Blocked: true, Message: "blocked: Auto Guard prompt failed: " + err.Error(), Generation: gen}, err |
| 1068 | } |
| 1069 | approvalID = strings.TrimSpace(approvalID) |
| 1070 | if approvalID == "" { |
| 1071 | g.mu.Lock() |
| 1072 | delete(g.waiters, provisional) |
| 1073 | delete(g.taskOf, provisional) |
| 1074 | delete(g.pending, provisional) |
| 1075 | delete(g.awaiting, taskID) |
| 1076 | g.mu.Unlock() |
| 1077 | return Decision{Allow: false, Blocked: true, Message: "blocked: Auto Guard prompt returned empty id", Generation: gen}, fmt.Errorf("empty Auto Guard approval id") |
| 1078 | } |
| 1079 | |
| 1080 | g.mu.Lock() |
| 1081 | // EmitPrompt implementations may bind the real id before emitting, which |
| 1082 | // lets a synchronous frontend resolve the card before EmitPrompt returns. |
| 1083 | // Only re-key a waiter that is still provisional; if both mappings are gone, |
| 1084 | // Resolve already completed and its buffered payload is waiting on reply. |
| 1085 | if provisionalReply, ok := g.waiters[provisional]; ok && provisionalReply != nil { |
| 1086 | delete(g.waiters, provisional) |
| 1087 | delete(g.taskOf, provisional) |
| 1088 | if p, exists := g.pending[provisional]; exists { |
| 1089 | delete(g.pending, provisional) |
| 1090 | g.pending[approvalID] = p |
| 1091 | } |
| 1092 | if existing, exists := g.waiters[approvalID]; exists && existing != nil { |
| 1093 | reply = existing |
| 1094 | } else { |
| 1095 | reply = provisionalReply |
| 1096 | g.waiters[approvalID] = reply |
| 1097 | g.taskOf[approvalID] = taskID |
| 1098 | } |
| 1099 | } else if existing, ok := g.waiters[approvalID]; ok && existing != nil { |
| 1100 | reply = existing |
| 1101 | } |
| 1102 | g.awaiting[taskID] = struct{}{} |
| 1103 | g.mu.Unlock() |
| 1104 | g.persist() |
| 1105 | |
| 1106 | select { |
| 1107 | case payload := <-reply: |
| 1108 | decision, err := g.decisionFromResolve(payload) |
| 1109 | if err == nil && decision.Allow && proposal.PlanTransition { |
| 1110 | decision.AuthorizePlanReplacement = true |
| 1111 | } |
| 1112 | decision.Generation = gen |
| 1113 | return decision, err |
| 1114 | case <-ctx.Done(): |
| 1115 | g.mu.Lock() |
| 1116 | delete(g.waiters, approvalID) |
| 1117 | delete(g.taskOf, approvalID) |
| 1118 | delete(g.pending, approvalID) |
| 1119 | delete(g.awaiting, taskID) |
| 1120 | g.mu.Unlock() |
| 1121 | g.persist() |
| 1122 | return Decision{Allow: false, Blocked: true, Message: "blocked: Auto Guard confirmation cancelled", Generation: gen}, ctx.Err() |
| 1123 | } |
| 1124 | } |
| 1125 | |
| 1126 | func taskGrantScopeKey(proposal Proposal) string { |
| 1127 | // Root task ids span a controller session. TaskScopeID is host-owned and |
| 1128 | // unique per ordinary turn, while goal continuations reuse their delivery |
| 1129 | // scope. Hash it so task-local runtime state never contains raw task text. |
| 1130 | taskScope := strings.TrimSpace(proposal.TaskScopeID) |
| 1131 | if taskScope == "" { |
| 1132 | taskScope = strings.TrimSpace(proposal.TaskSummary) |
| 1133 | } |
| 1134 | return CallFingerprint( |
| 1135 | "task-grant", |
| 1136 | normalizeTaskID(proposal.TaskID), |
| 1137 | taskScope, |
| 1138 | nil, |
| 1139 | ) |
| 1140 | } |
| 1141 | |
| 1142 | func taskGrantRuntimeKey(semanticKey, taskScope string) string { |
| 1143 | if semanticKey == "" || taskScope == "" { |
| 1144 | return "" |
| 1145 | } |
| 1146 | return semanticKey + "#" + taskScope |
| 1147 | } |
| 1148 | |
| 1149 | func (g *Gate) decisionFromResolve(payload resolvePayload) (Decision, error) { |
| 1150 | switch payload.action { |
| 1151 | case ActionContinue, ActionContinueTask: |
| 1152 | return Decision{Allow: true}, nil |
| 1153 | case ActionRevise: |
| 1154 | msg := "blocked: user requested a revised Auto Guard action" |
| 1155 | feedback := strings.TrimSpace(payload.feedback) |
| 1156 | if feedback == "" { |
| 1157 | feedback = DefaultReviseFeedback |
| 1158 | } |
| 1159 | msg += ": " + feedback |
| 1160 | return Decision{Allow: false, Blocked: true, Message: msg}, nil |
| 1161 | default: |
| 1162 | return Decision{Allow: false, Blocked: true, Message: "blocked: unknown Auto Guard action"}, nil |
| 1163 | } |
| 1164 | } |
| 1165 | |
| 1166 | // RecordDiagnosis appends a diagnosis note while recovering. |
| 1167 | func (g *Gate) RecordDiagnosis(taskID, note string) { |
| 1168 | if g == nil || strings.TrimSpace(note) == "" { |
| 1169 | return |
| 1170 | } |
| 1171 | g.mu.Lock() |
| 1172 | defer g.mu.Unlock() |
| 1173 | st := g.tasks[normalizeTaskID(taskID)] |
| 1174 | if st == nil || st.lastFailure == nil { |
| 1175 | return |
| 1176 | } |
| 1177 | if appendDiagnosisNote(st.lastFailure, note) { |
| 1178 | g.persistUnlocked() |
| 1179 | } |
| 1180 | } |
| 1181 | |
| 1182 | // --- internals --- |
| 1183 | |
| 1184 | func (g *Gate) activeMode() bool { |
| 1185 | mode := strings.ToLower(strings.TrimSpace(g.opts.Mode())) |
| 1186 | return mode == "auto" |
| 1187 | } |
| 1188 | |
| 1189 | func (g *Gate) recoveryGuidanceLocked(st *taskRuntime) string { |
| 1190 | if st.guidanceSent { |
| 1191 | return "" |
| 1192 | } |
| 1193 | st.guidanceSent = true |
| 1194 | if st.lastFailure != nil && st.lastFailure.evidence.Class == FailureClassTransient { |
| 1195 | return "The tool timed out or hit a transient execution limit. Inspect its current state and output before retrying so partial effects are not duplicated. " + |
| 1196 | "Read-only diagnosis and unrelated work remain available without asking the user; retry the exact operation only after ruling out partial effects." |
| 1197 | } |
| 1198 | return "A tool failed. Use read-only diagnosis as needed, continue unrelated work automatically, and do not ask the user unless a genuine product or plan choice is required. " + |
| 1199 | "Repeated retries of the exact failed operation remain bounded." |
| 1200 | } |
| 1201 | |
| 1202 | func diagnosticObservationNote(obs Observation) string { |
| 1203 | tool := clip(strings.TrimSpace(obs.Tool), 120) |
| 1204 | if tool == "" { |
| 1205 | tool = "diagnostic" |
| 1206 | } |
| 1207 | subject := clip(firstNonEmpty(obs.Subject, ArgsSummary(obs.Args, 160)), 160) |
| 1208 | header := tool |
| 1209 | if subject != "" && subject != tool { |
| 1210 | header += " (" + subject + ")" |
| 1211 | } |
| 1212 | output := strings.TrimSpace(obs.Output) |
| 1213 | if output == "" { |
| 1214 | return clipDiagnosisNote(header + ": completed successfully") |
| 1215 | } |
| 1216 | return clipDiagnosisNote(header + ": " + output) |
| 1217 | } |
| 1218 | |
| 1219 | func (g *Gate) persist() { |
| 1220 | if g == nil || g.opts.Persist == nil { |
| 1221 | return |
| 1222 | } |
| 1223 | // Disk never receives active lock state. |
| 1224 | g.schedulePersist(g.PersistenceSnapshot(), false) |
| 1225 | } |
| 1226 | |
| 1227 | func (g *Gate) persistUnlocked() { |
| 1228 | // Caller holds g.mu. |
| 1229 | if g == nil || g.opts.Persist == nil { |
| 1230 | return |
| 1231 | } |
| 1232 | g.schedulePersist(g.snapshotLocked(true), true) |
| 1233 | } |
| 1234 | |
| 1235 | func (g *Gate) schedulePersist(snap Snapshot, async bool) { |
| 1236 | if g == nil || g.opts.Persist == nil { |
| 1237 | return |
| 1238 | } |
| 1239 | key := "" |
| 1240 | if g.opts.PersistenceKey != nil { |
| 1241 | key = g.opts.PersistenceKey() |
| 1242 | } |
| 1243 | g.persistMu.Lock() |
| 1244 | g.persistSeq++ |
| 1245 | seq := g.persistSeq |
| 1246 | g.persistPending[key]++ |
| 1247 | g.persistMu.Unlock() |
| 1248 | write := func() { |
| 1249 | g.persistMu.Lock() |
| 1250 | defer g.persistMu.Unlock() |
| 1251 | defer func() { |
| 1252 | g.persistPending[key]-- |
| 1253 | if g.persistPending[key] == 0 { |
| 1254 | delete(g.persistPending, key) |
| 1255 | delete(g.persistDone, key) |
| 1256 | g.persistCond.Broadcast() |
| 1257 | } |
| 1258 | }() |
| 1259 | if seq < g.persistDone[key] { |
| 1260 | return |
| 1261 | } |
| 1262 | g.opts.Persist(key, snap) |
| 1263 | g.persistDone[key] = seq |
| 1264 | } |
| 1265 | if async { |
| 1266 | go write() |
| 1267 | return |
| 1268 | } |
| 1269 | write() |
| 1270 | } |
| 1271 | |
| 1272 | // userFacingReason is the short localized-friendly reason shown on the card. |
| 1273 | func userFacingReason(kind ChangeKind) string { |
| 1274 | switch kind { |
| 1275 | case ChangeRisk: |
| 1276 | return "This proposal is a technical execution-risk blocker, not a user-owned plan choice." |
| 1277 | case ChangeScope: |
| 1278 | return "This step would expand the change scope." |
| 1279 | case ChangeStrategy: |
| 1280 | return "Auto is about to try a different approach." |
| 1281 | default: |
| 1282 | return "Auto cannot establish how this proposal relates to the active task and plan." |
| 1283 | } |
| 1284 | } |
| 1285 | |
| 1286 | func headlessBlockerMessage(pending PendingProposal, failure *FailureEvent) string { |
| 1287 | var b strings.Builder |
| 1288 | b.WriteString("blocked: Auto Guard requires human confirmation, but this environment has no decision channel.\n") |
| 1289 | if failure != nil { |
| 1290 | b.WriteString("Failure: ") |
| 1291 | b.WriteString(firstNonEmpty(failure.ErrSummary, failure.Tool)) |
| 1292 | b.WriteString("\n") |
| 1293 | } |
| 1294 | if pending.Diagnosis != "" { |
| 1295 | b.WriteString("Diagnosis: ") |
| 1296 | b.WriteString(pending.Diagnosis) |
| 1297 | b.WriteString("\n") |
| 1298 | } |
| 1299 | b.WriteString("Proposed: ") |
| 1300 | b.WriteString(firstNonEmpty(pending.Proposed, pending.Subject, pending.Tool)) |
| 1301 | b.WriteString("\n") |
| 1302 | if pending.Rationale != "" { |
| 1303 | b.WriteString("Why confirm: ") |
| 1304 | b.WriteString(pending.Rationale) |
| 1305 | } |
| 1306 | return b.String() |
| 1307 | } |
| 1308 | |
| 1309 | func (g *Gate) recordReviewBlock(taskID string, verdict ReviewVerdict) int { |
| 1310 | g.mu.Lock() |
| 1311 | st := g.ensureTaskLocked(taskID) |
| 1312 | // Cumulative across all candidates and TaskIDs inside the Episode. |
| 1313 | if g.episode.reviewRejects < 255 { |
| 1314 | g.episode.reviewRejects++ |
| 1315 | } |
| 1316 | blocks := int(g.episode.reviewRejects) |
| 1317 | if st.lastFailure != nil { |
| 1318 | note := "Auto Guard reviewer blocked the proposal: " + firstNonEmpty(verdict.Rationale, string(verdict.ChangeKind)) |
| 1319 | appendDiagnosisNote(st.lastFailure, note) |
| 1320 | } |
| 1321 | g.mu.Unlock() |
| 1322 | g.persist() |
| 1323 | return blocks |
| 1324 | } |
| 1325 | |
| 1326 | func reviewerBlockerMessage(verdict ReviewVerdict, attempt, limit int) string { |
| 1327 | reason := firstNonEmpty(verdict.Rationale, "the proposal could not be classified as a bounded plan continuation") |
| 1328 | return fmt.Sprintf( |
| 1329 | "blocked: Auto plan reviewer could not accept this transition (attempt %d/%d): %s. Continue the current plan, propose a task-aligned plan, or ask the user about a genuine product choice.", |
| 1330 | attempt, limit, reason, |
| 1331 | ) |
| 1332 | } |
| 1333 | |
| 1334 | func repeatedFailureStopMessage(failures int, proposal Proposal) string { |
| 1335 | operation := clip(firstNonEmpty(proposal.Subject, proposal.Tool), 240) |
| 1336 | return fmt.Sprintf( |
| 1337 | "blocked: Auto stopped repeating this operation after %d consecutive failures: %s. Do not retry the same operation in this turn. Diagnose it with read-only tools, then use a different task-aligned edit or verification approach; other operations remain available. Ask the user only for a genuine product or plan choice.", |
| 1338 | failures, operation, |
| 1339 | ) |
| 1340 | } |
| 1341 | |
| 1342 | func episodeStopMessage(reason StopReason, proposal Proposal) string { |
| 1343 | operation := clip(firstNonEmpty(proposal.Subject, proposal.Tool), 240) |
| 1344 | switch reason { |
| 1345 | case StopReasonReviewRejects: |
| 1346 | return fmt.Sprintf( |
| 1347 | "blocked: Auto recovery paused this turn after %d reviewer rejections (last proposal: %s). Do not call more tools; summarize what was tried and what remains. The user can continue in the next message.", |
| 1348 | MaxReviewRejects, operation, |
| 1349 | ) |
| 1350 | case StopReasonStoppedOpRetries: |
| 1351 | return fmt.Sprintf( |
| 1352 | "blocked: Auto recovery paused this turn after repeated attempts of already-stopped operations (last: %s). Do not call more tools; summarize completed work and blockers. The user can continue in the next message.", |
| 1353 | operation, |
| 1354 | ) |
| 1355 | default: |
| 1356 | return fmt.Sprintf( |
| 1357 | "blocked: Auto recovery paused this turn after %d execution failures without progress (last: %s). Do not call more tools; summarize completed work and blockers. The user can continue in the next message.", |
| 1358 | MaxEpisodeFailures, operation, |
| 1359 | ) |
| 1360 | } |
| 1361 | } |
| 1362 | |
| 1363 | func observationFingerprint(obs Observation) string { |
| 1364 | return CallFingerprint(obs.Tool, obs.Subject, "", obs.Args) |
| 1365 | } |
| 1366 | |
| 1367 | func persistentRecoveryScope(scope string) string { |
| 1368 | scope = strings.TrimSpace(scope) |
| 1369 | if strings.HasPrefix(scope, "goal:") { |
| 1370 | return scope |
| 1371 | } |
| 1372 | return "" |
| 1373 | } |
| 1374 | |
| 1375 | func sameFailedOperation(failure *FailureEvent, proposal Proposal) bool { |
| 1376 | if failure == nil { |
| 1377 | return false |
| 1378 | } |
| 1379 | want := strings.TrimSpace(failure.Fingerprint) |
| 1380 | if want == "" { |
| 1381 | want = CallFingerprint(failure.Tool, failure.Subject, "", failure.Args) |
| 1382 | } |
| 1383 | return want == CallFingerprint(proposal.Tool, proposal.Subject, "", proposal.Args) |
| 1384 | } |
| 1385 | |
| 1386 | func normalizeVerdict(v ReviewVerdict, failure *FailureEvent, proposal Proposal, diagNotes []string) ReviewVerdict { |
| 1387 | switch strings.ToLower(strings.TrimSpace(string(v.Outcome))) { |
| 1388 | case "continue": |
| 1389 | v.Outcome = ReviewContinue |
| 1390 | case "confirm": |
| 1391 | v.Outcome = ReviewConfirm |
| 1392 | default: |
| 1393 | // Unparseable/unknown outcome fails closed. |
| 1394 | v.Outcome = ReviewConfirm |
| 1395 | if v.ChangeKind == "" { |
| 1396 | v.ChangeKind = ChangeUncertain |
| 1397 | } |
| 1398 | } |
| 1399 | switch ChangeKind(strings.ToLower(strings.TrimSpace(string(v.ChangeKind)))) { |
| 1400 | case ChangeSameStrategy, ChangeStrategy, ChangeScope, ChangeRisk, ChangeUncertain: |
| 1401 | v.ChangeKind = ChangeKind(strings.ToLower(strings.TrimSpace(string(v.ChangeKind)))) |
| 1402 | default: |
| 1403 | if v.Outcome == ReviewContinue { |
| 1404 | // Cannot silently continue without a clear bounded-recovery label. |
| 1405 | v.Outcome = ReviewConfirm |
| 1406 | } |
| 1407 | v.ChangeKind = ChangeUncertain |
| 1408 | } |
| 1409 | // Risk and uncertainty cannot silently continue, but they are technical |
| 1410 | // blockers rather than human approval requests. Strategy/scope may continue |
| 1411 | // when the reviewer established that the change remains task-aligned. |
| 1412 | if v.Outcome == ReviewContinue && !reviewerContinueKind(v.ChangeKind) { |
| 1413 | v.Outcome = ReviewConfirm |
| 1414 | } |
| 1415 | if strings.TrimSpace(v.FailureSummary) == "" && failure != nil { |
| 1416 | v.FailureSummary = failure.ErrSummary |
| 1417 | } |
| 1418 | if strings.TrimSpace(v.Diagnosis) == "" { |
| 1419 | v.Diagnosis = strings.Join(diagNotes, "\n") |
| 1420 | } |
| 1421 | if strings.TrimSpace(v.ProposedAction) == "" { |
| 1422 | v.ProposedAction = firstNonEmpty(proposal.Subject, proposal.Preview, proposal.Tool) |
| 1423 | } |
| 1424 | if strings.TrimSpace(v.Rationale) == "" { |
| 1425 | v.Rationale = userFacingReason(v.ChangeKind) |
| 1426 | } else { |
| 1427 | v.Rationale = clip(v.Rationale, 500) |
| 1428 | } |
| 1429 | return v |
| 1430 | } |
| 1431 | |
| 1432 | func reviewerContinueKind(kind ChangeKind) bool { |
| 1433 | switch kind { |
| 1434 | case ChangeSameStrategy, ChangeStrategy, ChangeScope: |
| 1435 | return true |
| 1436 | default: |
| 1437 | return false |
| 1438 | } |
| 1439 | } |
| 1440 | |
| 1441 | func reviewerPlanDecision(verdict ReviewVerdict) bool { |
| 1442 | if verdict.Outcome != ReviewConfirm { |
| 1443 | return false |
| 1444 | } |
| 1445 | switch verdict.ChangeKind { |
| 1446 | case ChangeStrategy, ChangeScope: |
| 1447 | return true |
| 1448 | default: |
| 1449 | return false |
| 1450 | } |
| 1451 | } |
| 1452 | |
| 1453 | // Ensure Gate implements agent.RecoveryGate. |
| 1454 | var _ agent.RecoveryGate = (*Gate)(nil) |
| 1455 |