| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "time" |
| 8 | |
| 9 | "reasonix/internal/agent" |
| 10 | "reasonix/internal/event" |
| 11 | "reasonix/internal/extension" |
| 12 | "reasonix/internal/provider" |
| 13 | "reasonix/internal/session" |
| 14 | ) |
| 15 | |
| 16 | type queuedTurnKind int |
| 17 | |
| 18 | const ( |
| 19 | queuedUser queuedTurnKind = iota |
| 20 | queuedGoal |
| 21 | ) |
| 22 | |
| 23 | type queuedTurn struct { |
| 24 | kind queuedTurnKind |
| 25 | body func(ctx context.Context) error |
| 26 | onStart func() |
| 27 | goalRound *goalRoundReservation |
| 28 | // admissionCtx is only used for the synchronous durability boundary. The |
| 29 | // model turn has its own controller-owned context and survives RPC return. |
| 30 | admissionCtx context.Context |
| 31 | } |
| 32 | |
| 33 | // turnLoop is the session-scoped execution authority. Controller.mu guards it. |
| 34 | type turnLoop struct { |
| 35 | phase session.RuntimePhase |
| 36 | cancel context.CancelFunc |
| 37 | done chan struct{} |
| 38 | recoveryFanout bool // watchdog terminal publication still owns the stores |
| 39 | turnID string |
| 40 | token uint64 |
| 41 | lastToken uint64 |
| 42 | pending []queuedTurn |
| 43 | wake bool |
| 44 | generation uint64 |
| 45 | runtime *session.Runtime |
| 46 | cancelRequested bool |
| 47 | finishingBound turnFinishingBoundary |
| 48 | } |
| 49 | |
| 50 | type controllerExecution struct { |
| 51 | c *Controller |
| 52 | } |
| 53 | |
| 54 | func (e controllerExecution) Snapshot() session.RuntimeSnapshot { |
| 55 | if e.c == nil { |
| 56 | return session.RuntimeSnapshot{Phase: session.RuntimeIdle} |
| 57 | } |
| 58 | e.c.mu.Lock() |
| 59 | defer e.c.mu.Unlock() |
| 60 | return session.RuntimeSnapshot{Phase: e.c.turns.phase, Activity: e.c.turns.activityNameLocked()} |
| 61 | } |
| 62 | |
| 63 | func (e controllerExecution) Cancel() bool { |
| 64 | if e.c == nil { |
| 65 | return false |
| 66 | } |
| 67 | return e.c.signalTurnCancel() |
| 68 | } |
| 69 | |
| 70 | func (t *turnLoop) activityNameLocked() string { |
| 71 | switch t.phase { |
| 72 | case session.RuntimeCancelling: |
| 73 | return "cancelling" |
| 74 | case session.RuntimeRecoveryRequired: |
| 75 | return "recovery_required" |
| 76 | case session.RuntimeRunning, session.RuntimeFinalizing: |
| 77 | if t.cancelRequested { |
| 78 | return "cancelling" |
| 79 | } |
| 80 | return "turn" |
| 81 | default: |
| 82 | return "" |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | func (c *Controller) bodyActiveLocked() bool { |
| 87 | switch c.turns.phase { |
| 88 | case session.RuntimeRunning, session.RuntimeCancelling: |
| 89 | return true |
| 90 | default: |
| 91 | return false |
| 92 | } |
| 93 | } |
| 94 | |
| 95 | func (c *Controller) finalizingLocked() bool { |
| 96 | return c.turns.phase == session.RuntimeFinalizing |
| 97 | } |
| 98 | |
| 99 | func (c *Controller) cancelRequestedLocked() bool { |
| 100 | if c.closed { |
| 101 | return false |
| 102 | } |
| 103 | return c.turns.cancelRequested || c.turns.phase == session.RuntimeCancelling |
| 104 | } |
| 105 | |
| 106 | func (c *Controller) recoveryRequiredLocked() bool { |
| 107 | return c.turns.phase == session.RuntimeRecoveryRequired |
| 108 | } |
| 109 | |
| 110 | func (c *Controller) bindExecutionControl() { |
| 111 | _, runtime, exclusive := c.v3Binding() |
| 112 | if !exclusive || runtime == nil { |
| 113 | return |
| 114 | } |
| 115 | snap := runtime.StateSnapshot() |
| 116 | gen := runtime.BindExecution(controllerExecution{c: c}) |
| 117 | c.executionGeneration.Store(gen) |
| 118 | c.mu.Lock() |
| 119 | c.turns.generation = gen |
| 120 | c.turns.runtime = runtime |
| 121 | if snap.Phase == session.RuntimeRecoveryRequired { |
| 122 | c.turns.phase = session.RuntimeRecoveryRequired |
| 123 | } |
| 124 | c.mu.Unlock() |
| 125 | } |
| 126 | |
| 127 | // ExecutionGeneration returns the session-runtime execution generation owned |
| 128 | // by this controller. Zero means the controller is a prepared replacement that |
| 129 | // has not been published as the execution owner. |
| 130 | func (c *Controller) ExecutionGeneration() uint64 { |
| 131 | if c == nil { |
| 132 | return 0 |
| 133 | } |
| 134 | return c.executionGeneration.Load() |
| 135 | } |
| 136 | |
| 137 | // ActivateSessionExecution publishes this controller as the exact execution |
| 138 | // owner. expectedGeneration is zero for a previously unbound runtime and the |
| 139 | // outgoing controller generation for a fail-atomic replacement. The method is |
| 140 | // deliberately callback- and I/O-free so hosts may invoke it in their final |
| 141 | // pointer-swap critical section. |
| 142 | func (c *Controller) ActivateSessionExecution(expectedGeneration uint64) error { |
| 143 | if c == nil { |
| 144 | return session.ErrSessionNotRunning |
| 145 | } |
| 146 | c.turnEvents.commitMu.Lock() |
| 147 | defer c.turnEvents.commitMu.Unlock() |
| 148 | c.mu.Lock() |
| 149 | defer c.mu.Unlock() |
| 150 | runtime := c.turns.runtime |
| 151 | if runtime == nil { |
| 152 | return nil |
| 153 | } |
| 154 | if generation := c.turns.generation; generation != 0 && runtime.OwnsExecution(generation) { |
| 155 | return nil |
| 156 | } |
| 157 | var generation uint64 |
| 158 | if expectedGeneration == 0 { |
| 159 | generation = runtime.BindExecution(controllerExecution{c: c}) |
| 160 | } else if pending := c.turnEvents.pendingExecutionCommit; pending != nil { |
| 161 | c.turnEvents.pendingExecutionCommit = nil |
| 162 | var err error |
| 163 | generation, _, err = runtime.ReplaceExecutionAndCommit(expectedGeneration, controllerExecution{c: c}, *pending) |
| 164 | if err != nil { |
| 165 | return err |
| 166 | } |
| 167 | } else { |
| 168 | generation = runtime.ReplaceExecution(expectedGeneration, controllerExecution{c: c}) |
| 169 | } |
| 170 | if generation == 0 { |
| 171 | return session.ErrRuntimeBusy |
| 172 | } |
| 173 | c.turns.generation = generation |
| 174 | c.executionGeneration.Store(generation) |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | // ActivateControllerReplacement transfers session execution ownership when a |
| 179 | // host commits a controller pointer swap. Controllers without a shared Runtime |
| 180 | // need no additional activation. |
| 181 | func ActivateControllerReplacement(old, next *Controller) error { |
| 182 | if next == nil { |
| 183 | return session.ErrSessionNotRunning |
| 184 | } |
| 185 | _, nextRuntime, nextExclusive := next.v3Binding() |
| 186 | if !nextExclusive || nextRuntime == nil { |
| 187 | return nil |
| 188 | } |
| 189 | expected := uint64(0) |
| 190 | if old != nil { |
| 191 | _, oldRuntime, oldExclusive := old.v3Binding() |
| 192 | if oldExclusive && oldRuntime == nextRuntime { |
| 193 | expected = old.ExecutionGeneration() |
| 194 | } |
| 195 | } |
| 196 | return next.ActivateSessionExecution(expected) |
| 197 | } |
| 198 | |
| 199 | // ActivateSessionAPIReplacement is the host-facing form used at a final |
| 200 | // controller pointer swap. Non-Controller implementations have no exclusive |
| 201 | // Runtime ownership to transfer and are left unchanged. |
| 202 | func ActivateSessionAPIReplacement(old, next SessionAPI) error { |
| 203 | concreteNext, ok := next.(*Controller) |
| 204 | if !ok || concreteNext == nil { |
| 205 | return nil |
| 206 | } |
| 207 | concreteOld, _ := old.(*Controller) |
| 208 | if err := ActivateControllerReplacement(concreteOld, concreteNext); err != nil { |
| 209 | return err |
| 210 | } |
| 211 | if concreteOld != nil && concreteOld.attachmentScope() != "" && concreteOld.attachmentScope() == concreteNext.attachmentScope() && concreteOld.workspaceRoot == concreteNext.workspaceRoot { |
| 212 | concreteOld.attachmentService().Drafts().CopyScopeTo(concreteOld.attachmentScope(), concreteNext.attachmentService().Drafts()) |
| 213 | } |
| 214 | return nil |
| 215 | } |
| 216 | |
| 217 | func (c *Controller) unbindExecutionControl(runtime *session.Runtime) { |
| 218 | if runtime == nil { |
| 219 | return |
| 220 | } |
| 221 | c.mu.Lock() |
| 222 | gen := c.turns.generation |
| 223 | c.mu.Unlock() |
| 224 | runtime.UnbindExecution(gen) |
| 225 | } |
| 226 | |
| 227 | func (c *Controller) noteExecutionLocked(phase session.RuntimePhase, activity string) { |
| 228 | if c.turns.runtime == nil || c.turns.generation == 0 { |
| 229 | return |
| 230 | } |
| 231 | c.turns.runtime.NoteExecution(c.turns.generation, phase, activity) |
| 232 | } |
| 233 | |
| 234 | func (c *Controller) currentTurnToken() (token uint64, turnID string, active bool) { |
| 235 | c.mu.Lock() |
| 236 | defer c.mu.Unlock() |
| 237 | switch c.turns.phase { |
| 238 | case session.RuntimeIdle, session.RuntimeClosed: |
| 239 | return c.turns.lastToken, "", false |
| 240 | default: |
| 241 | return c.turns.token, c.turns.turnID, true |
| 242 | } |
| 243 | } |
| 244 | |
| 245 | func (c *Controller) discardLateTurnEvent(e event.Event) bool { |
| 246 | if e.TurnID == "" || !lateBusinessEvent(e.Kind) { |
| 247 | return false |
| 248 | } |
| 249 | _, turnID, active := c.currentTurnToken() |
| 250 | if active { |
| 251 | return e.TurnID != turnID |
| 252 | } |
| 253 | return true |
| 254 | } |
| 255 | |
| 256 | func (c *Controller) startTurnLocked(parent context.Context, next queuedTurn) (ctx context.Context, cancel context.CancelFunc, admitted bool) { |
| 257 | if parent == nil { |
| 258 | parent = context.Background() |
| 259 | } |
| 260 | if c.turns.runtime != nil && !c.turns.runtime.BeginExecution(c.turns.generation, "turn") { |
| 261 | return nil, nil, false |
| 262 | } |
| 263 | ctx, cancel = context.WithCancel(extension.ContextWithRuntimeOwner(c.withAuthentication(parent), c.runtimeOwner)) |
| 264 | c.turns.cancel = cancel |
| 265 | c.turns.done = make(chan struct{}) |
| 266 | c.turns.finishingBound.beginIdle() |
| 267 | c.turns.phase = session.RuntimeRunning |
| 268 | c.turns.cancelRequested = false |
| 269 | c.turns.token++ |
| 270 | ctx = context.WithValue(ctx, executionTokenKey{}, c.turns.token) |
| 271 | c.turns.turnID = "" |
| 272 | return ctx, cancel, true |
| 273 | } |
| 274 | |
| 275 | func (c *Controller) popNextPendingLocked() (queuedTurn, bool) { |
| 276 | if len(c.turns.pending) == 0 { |
| 277 | c.turns.wake = false |
| 278 | return queuedTurn{}, false |
| 279 | } |
| 280 | next := c.turns.pending[0] |
| 281 | c.turns.pending = c.turns.pending[1:] |
| 282 | c.turns.wake = len(c.turns.pending) > 0 |
| 283 | return next, true |
| 284 | } |
| 285 | |
| 286 | func (c *Controller) queueTurnLocked(item queuedTurn) { |
| 287 | // Harness wakeRequested: input that cannot join the current activity is |
| 288 | // claimed once when the body converges. Close clears this queue so a |
| 289 | // disposed session never starts a latched turn. |
| 290 | c.turns.pending = append(c.turns.pending, item) |
| 291 | c.turns.wake = true |
| 292 | } |
| 293 | |
| 294 | func (c *Controller) signalTurnCancel() bool { |
| 295 | _, _, cancelled := c.signalTurnCancelIdentity() |
| 296 | return cancelled |
| 297 | } |
| 298 | |
| 299 | func (c *Controller) signalTurnCancelIdentity() (uint64, string, bool) { |
| 300 | c.mu.Lock() |
| 301 | token, turnID := c.turns.token, c.turns.turnID |
| 302 | cancel := c.turns.cancel |
| 303 | first := cancel != nil && c.turns.phase == session.RuntimeRunning |
| 304 | if cancel != nil && (c.turns.phase == session.RuntimeRunning || c.turns.phase == session.RuntimeCancelling) { |
| 305 | c.turns.phase = session.RuntimeCancelling |
| 306 | c.turns.cancelRequested = true |
| 307 | } |
| 308 | done := c.turns.done |
| 309 | c.mu.Unlock() |
| 310 | if cancel == nil { |
| 311 | return token, turnID, false |
| 312 | } |
| 313 | cancel() |
| 314 | if first { |
| 315 | c.startCancellationWatchdog(done) |
| 316 | } |
| 317 | return token, turnID, true |
| 318 | } |
| 319 | |
| 320 | func (c *Controller) enterRecoveryLocked(reason string) { |
| 321 | c.turns.phase = session.RuntimeRecoveryRequired |
| 322 | c.noteExecutionLocked(session.RuntimeRecoveryRequired, reason) |
| 323 | if c.turns.runtime != nil { |
| 324 | c.turns.runtime.RequireRecovery(reason) |
| 325 | } |
| 326 | } |
| 327 | |
| 328 | func (c *Controller) spawnGuardedTurn(ctx context.Context, cancel context.CancelFunc, item queuedTurn) { |
| 329 | ctx, completion := withGuardedTurnCompletion(ctx) |
| 330 | admissionCtx := item.admissionCtx |
| 331 | if admissionCtx == nil { |
| 332 | admissionCtx = context.Background() |
| 333 | } |
| 334 | body := c.prepareTurnAdmissionWithGoalRound(admissionCtx, item.body, item.goalRound) |
| 335 | if ledger := c.turnEventLedger(); ledger != nil { |
| 336 | c.mu.Lock() |
| 337 | c.turns.turnID = ledger.ActiveTurnID() |
| 338 | c.mu.Unlock() |
| 339 | } |
| 340 | c.liveness.reset(time.Now()) |
| 341 | c.autosaveWG.Go(func() { |
| 342 | c.autosaveWhileRunning(ctx) |
| 343 | }) |
| 344 | go func() { |
| 345 | defer cancel() |
| 346 | defer func() { |
| 347 | c.finishGoalRoundActivity(item.goalRound) |
| 348 | c.kickGoalDriver() |
| 349 | }() |
| 350 | defer func() { |
| 351 | if r := recover(); r != nil { |
| 352 | err := fmt.Errorf("internal error: %v", r) |
| 353 | item.goalRound.setResult(err, false) |
| 354 | c.finishGuardedTurn(err, completion) |
| 355 | } |
| 356 | }() |
| 357 | err := body(ctx) |
| 358 | if item.goalRound != nil { |
| 359 | item.goalRound.setResult(err, errors.Is(ctx.Err(), context.Canceled) && c.CancelRequested()) |
| 360 | } |
| 361 | c.finishGuardedTurn(explainError(err), completion) |
| 362 | }() |
| 363 | } |
| 364 | |
| 365 | func (c *Controller) cancellationGrace() time.Duration { |
| 366 | if c != nil && c.testCancelGrace > 0 { |
| 367 | return c.testCancelGrace |
| 368 | } |
| 369 | return 15 * time.Second |
| 370 | } |
| 371 | |
| 372 | func (c *Controller) finishGuardedTurn(err error, completion *guardedTurnCompletion) { |
| 373 | c.authentication.recordFailure(err, c.ModelRef()) |
| 374 | c.memory.clearAutoRemember() |
| 375 | c.mu.Lock() |
| 376 | cancelRequested := c.turns.cancelRequested |
| 377 | if c.turns.phase == session.RuntimeRecoveryRequired { |
| 378 | c.turns.cancel = nil |
| 379 | closing := c.closed |
| 380 | c.mu.Unlock() |
| 381 | // The cancellation watchdog already committed the recovery terminal. |
| 382 | // A closing controller must not emit another terminal after its ledger |
| 383 | // and session binding have been finalized. |
| 384 | if !closing { |
| 385 | c.emitTurnDoneEvent(err, cancelRequested, completion) |
| 386 | } |
| 387 | c.mu.Lock() |
| 388 | // Keep the owned turn live through terminal fanout. Close must not |
| 389 | // release stores while that fanout can still publish durable events. |
| 390 | if c.turns.done != nil { |
| 391 | close(c.turns.done) |
| 392 | c.turns.done = nil |
| 393 | } |
| 394 | closing = c.closed && !c.turns.recoveryFanout |
| 395 | c.turns.finishingBound.endIdle() |
| 396 | c.mu.Unlock() |
| 397 | if closing { |
| 398 | c.finalizeControllerClose() |
| 399 | } |
| 400 | c.refreshRuntimeState(event.Event{}) |
| 401 | return |
| 402 | } |
| 403 | if c.turns.done != nil { |
| 404 | close(c.turns.done) |
| 405 | c.turns.done = nil |
| 406 | } |
| 407 | c.turns.phase = session.RuntimeFinalizing |
| 408 | c.turns.finishingBound.begin(true) |
| 409 | c.turns.cancel = nil |
| 410 | c.noteExecutionLocked(session.RuntimeFinalizing, "turn") |
| 411 | c.mu.Unlock() |
| 412 | |
| 413 | c.refreshRuntimeState(event.Event{}) |
| 414 | defer func() { |
| 415 | c.mu.Lock() |
| 416 | c.turns.finishingBound.end() |
| 417 | c.turns.cancelRequested = false |
| 418 | if c.turns.phase == session.RuntimeRecoveryRequired { |
| 419 | closing := c.closed |
| 420 | c.turns.finishingBound.endIdle() |
| 421 | c.mu.Unlock() |
| 422 | if closing { |
| 423 | c.finalizeControllerClose() |
| 424 | } |
| 425 | c.refreshRuntimeState(event.Event{}) |
| 426 | return |
| 427 | } |
| 428 | if ledger := c.turnEventLedger(); ledger != nil && ledger.CurrentStatus() == event.TurnRecoveryRequired { |
| 429 | c.enterRecoveryLocked("terminal") |
| 430 | c.turns.finishingBound.endIdle() |
| 431 | c.mu.Unlock() |
| 432 | c.refreshRuntimeState(event.Event{}) |
| 433 | return |
| 434 | } |
| 435 | if c.closed { |
| 436 | c.turns.lastToken = c.turns.token |
| 437 | c.turns.phase = session.RuntimeClosed |
| 438 | c.turns.turnID = "" |
| 439 | c.noteExecutionLocked(session.RuntimeIdle, "") |
| 440 | c.turns.finishingBound.endIdle() |
| 441 | c.mu.Unlock() |
| 442 | c.finalizeControllerClose() |
| 443 | c.refreshRuntimeState(event.Event{}) |
| 444 | return |
| 445 | } |
| 446 | if authErr := c.authentication.admissionError(); authErr != nil { |
| 447 | for _, pending := range c.turns.pending { |
| 448 | if pending.goalRound != nil { |
| 449 | pending.goalRound.setResult(authErr, false) |
| 450 | } |
| 451 | } |
| 452 | c.turns.pending = nil |
| 453 | c.turns.wake = false |
| 454 | c.turns.lastToken = c.turns.token |
| 455 | c.turns.phase = session.RuntimeIdle |
| 456 | c.turns.turnID = "" |
| 457 | c.noteExecutionLocked(session.RuntimeIdle, "") |
| 458 | c.turns.finishingBound.endIdle() |
| 459 | c.mu.Unlock() |
| 460 | c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Code: "authentication_not_ready", Text: authErr.Error()}) |
| 461 | c.refreshRuntimeState(event.Event{}) |
| 462 | return |
| 463 | } |
| 464 | next, ok := c.popNextPendingLocked() |
| 465 | if !ok { |
| 466 | c.turns.lastToken = c.turns.token |
| 467 | c.turns.phase = session.RuntimeIdle |
| 468 | c.turns.turnID = "" |
| 469 | c.noteExecutionLocked(session.RuntimeIdle, "") |
| 470 | c.turns.finishingBound.endIdle() |
| 471 | c.mu.Unlock() |
| 472 | c.maybeDispatchInbox() |
| 473 | c.refreshRuntimeState(event.Event{}) |
| 474 | return |
| 475 | } |
| 476 | ctx, cancel, admitted := c.startTurnLocked(context.Background(), next) |
| 477 | if !admitted { |
| 478 | c.enterRecoveryLocked("execution_owner_lost") |
| 479 | c.turns.finishingBound.endIdle() |
| 480 | c.mu.Unlock() |
| 481 | c.refreshRuntimeState(event.Event{}) |
| 482 | return |
| 483 | } |
| 484 | c.mu.Unlock() |
| 485 | if next.onStart != nil { |
| 486 | next.onStart() |
| 487 | } |
| 488 | c.spawnGuardedTurn(ctx, cancel, next) |
| 489 | c.refreshRuntimeState(event.Event{}) |
| 490 | }() |
| 491 | c.emitTurnDoneEvent(err, cancelRequested, completion) |
| 492 | } |
| 493 | |
| 494 | func (c *Controller) emitTurnDoneEvent(err error, cancelRequested bool, completion *guardedTurnCompletion) { |
| 495 | c.inbox.mu.Lock() |
| 496 | activeInboxID := "" |
| 497 | for id := range c.inbox.activeItemIDs { |
| 498 | activeInboxID = id |
| 499 | break |
| 500 | } |
| 501 | c.inbox.mu.Unlock() |
| 502 | done := event.Event{ |
| 503 | Kind: event.TurnDone, |
| 504 | Err: err, |
| 505 | Cancelled: cancelRequested, |
| 506 | Outcome: turnOutcome(err), |
| 507 | CheckpointTurn: c.validatedCheckpointTurn(completion), |
| 508 | Receipt: c.executor.CompletionReceipt(), |
| 509 | ItemID: activeInboxID, |
| 510 | } |
| 511 | if done.CheckpointTurn != nil { |
| 512 | changes := completion.checkpoint.store.FreezeTurnChanges(*done.CheckpointTurn) |
| 513 | if done.Receipt == nil && (len(changes.Files) > 0 || len(changes.Reasons) > 0) { |
| 514 | done.Receipt = &event.CompletionReceipt{AssessmentKind: "facts", Verdict: "unknown"} |
| 515 | } |
| 516 | if done.Receipt != nil { |
| 517 | receipt := *done.Receipt |
| 518 | receipt.Diff = changes.Summary() |
| 519 | receipt.Interrupted = cancelRequested |
| 520 | done.Receipt = &receipt |
| 521 | } |
| 522 | } |
| 523 | done.Receipt = bindCompletionLogSources(done.Receipt, c.History()) |
| 524 | done = c.applyTurnDoneProtocol(done, cancelRequested) |
| 525 | c.applyToolRecoveryTurnStatus(&done, completion) |
| 526 | var readErr *agent.IncompleteReadError |
| 527 | if errors.As(err, &readErr) { |
| 528 | done.ReadPause = readErr.Pause |
| 529 | } |
| 530 | done.Diagnostic = provider.DiagnoseFailure(err) |
| 531 | done.Detail = provider.FailureDiagnosticDetail(done.Diagnostic) |
| 532 | if !cancelRequested { |
| 533 | done.ProtocolRecovery = c.executor.PendingProtocolRecovery() |
| 534 | } |
| 535 | var readinessErr *agent.FinalReadinessError |
| 536 | if errors.As(err, &readinessErr) { |
| 537 | done.Readiness = &event.FinalReadiness{Attempts: readinessErr.Attempts, Missing: append([]string(nil), readinessErr.Missing...)} |
| 538 | } |
| 539 | c.onInboxTurnDone() |
| 540 | c.sink.Emit(done) |
| 541 | } |
| 542 | |
| 543 | func (c *Controller) startCancellationWatchdog(done chan struct{}) { |
| 544 | if c == nil || done == nil { |
| 545 | return |
| 546 | } |
| 547 | go func() { |
| 548 | timer := time.NewTimer(c.cancellationGrace()) |
| 549 | defer timer.Stop() |
| 550 | select { |
| 551 | case <-timer.C: |
| 552 | case <-done: |
| 553 | return |
| 554 | } |
| 555 | |
| 556 | c.mu.Lock() |
| 557 | stillRunning := c.turns.done == done && (c.turns.phase == session.RuntimeRunning || c.turns.phase == session.RuntimeCancelling) |
| 558 | turnID := c.turns.turnID |
| 559 | if stillRunning { |
| 560 | c.turns.recoveryFanout = true |
| 561 | c.enterRecoveryLocked("cancellation_grace_expired") |
| 562 | } |
| 563 | c.mu.Unlock() |
| 564 | if !stillRunning { |
| 565 | return |
| 566 | } |
| 567 | if turnID == "" { |
| 568 | if ledger := c.turnEventLedger(); ledger != nil { |
| 569 | turnID = ledger.ActiveTurnID() |
| 570 | } |
| 571 | } |
| 572 | recovery := &event.RecoveryStatus{ |
| 573 | State: "recovery_required", |
| 574 | Phase: "cancellation_grace_expired", |
| 575 | Reason: "cancellation_grace_expired", |
| 576 | RequiresUserDecision: true, |
| 577 | } |
| 578 | _ = c.emitTurnEventChecked(event.Event{ |
| 579 | Kind: event.TurnDone, |
| 580 | TurnID: turnID, |
| 581 | Status: event.TurnRecoveryRequired, |
| 582 | Cancelled: true, |
| 583 | Outcome: "unknown", |
| 584 | Recovery: recovery, |
| 585 | }) |
| 586 | c.mu.Lock() |
| 587 | c.turns.recoveryFanout = false |
| 588 | closing := c.closed && c.turns.done == nil && !c.finalizingLocked() |
| 589 | c.mu.Unlock() |
| 590 | if closing { |
| 591 | c.finalizeControllerClose() |
| 592 | } |
| 593 | c.refreshRuntimeState(event.Event{}) |
| 594 | }() |
| 595 | } |
| 596 |