| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "io" |
| 6 | "path/filepath" |
| 7 | "sync" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/agent" |
| 12 | "reasonix/internal/event" |
| 13 | "reasonix/internal/jobs" |
| 14 | "reasonix/internal/provider" |
| 15 | ) |
| 16 | |
| 17 | func TestCancelSessionTreatsIndependentBackgroundJobAsIdle(t *testing.T) { |
| 18 | manager := jobs.NewManager(event.Discard) |
| 19 | t.Cleanup(manager.Close) |
| 20 | path := filepath.Join(t.TempDir(), "session.jsonl") |
| 21 | c := newOwnedTestController(t, Options{Jobs: manager, SessionPath: path}) |
| 22 | t.Cleanup(c.Close) |
| 23 | started := make(chan struct{}) |
| 24 | manager.StartForSession(agent.BranchID(path), "bash", "background", func(ctx context.Context, _ io.Writer) (string, error) { |
| 25 | close(started) |
| 26 | <-ctx.Done() |
| 27 | return "", ctx.Err() |
| 28 | }) |
| 29 | <-started |
| 30 | |
| 31 | receipt := c.CancelSession() |
| 32 | if !receipt.Accepted || !receipt.AlreadyIdle { |
| 33 | t.Fatalf("idle receipt with background job = %+v", receipt) |
| 34 | } |
| 35 | if running := manager.RunningForSession(agent.BranchID(path)); len(running) != 1 { |
| 36 | t.Fatalf("session Stop changed independent background jobs: %+v", running) |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | // Stop must acknowledge after signalling the turn without waiting for the |
| 41 | // cancelling status to cross a synchronous event barrier. |
| 42 | func TestCancelSessionAcknowledgesBeforeStatusBarrier(t *testing.T) { |
| 43 | releaseStatus := make(chan struct{}) |
| 44 | statusEntered := make(chan struct{}, 1) |
| 45 | c := newOwnedTestController(t, Options{Sink: event.FuncSink(func(e event.Event) { |
| 46 | if e.Kind == event.TurnStatusChanged && e.Status == event.TurnCancelling { |
| 47 | statusEntered <- struct{}{} |
| 48 | <-releaseStatus |
| 49 | } |
| 50 | })}) |
| 51 | t.Cleanup(c.Close) |
| 52 | |
| 53 | turnCtxDone := make(chan struct{}) |
| 54 | releaseTurn := make(chan struct{}) |
| 55 | started := make(chan struct{}) |
| 56 | c.runGuarded(func(ctx context.Context) error { |
| 57 | close(started) |
| 58 | <-ctx.Done() |
| 59 | close(turnCtxDone) |
| 60 | // Hold the turn open so TurnDone cannot race ahead of the cancelling |
| 61 | // status; the assertion is about ordering inside Cancel itself. |
| 62 | <-releaseTurn |
| 63 | return ctx.Err() |
| 64 | }) |
| 65 | <-started |
| 66 | defer close(releaseTurn) |
| 67 | |
| 68 | cancelReturned := make(chan struct{}) |
| 69 | go func() { |
| 70 | c.CancelSession() |
| 71 | close(cancelReturned) |
| 72 | }() |
| 73 | select { |
| 74 | case <-turnCtxDone: |
| 75 | case <-time.After(5 * time.Second): |
| 76 | close(releaseStatus) |
| 77 | t.Fatal("turn context was not cancelled before the status barrier") |
| 78 | } |
| 79 | select { |
| 80 | case <-statusEntered: |
| 81 | case <-time.After(5 * time.Second): |
| 82 | close(releaseStatus) |
| 83 | t.Fatal("cancel never emitted the cancelling status") |
| 84 | } |
| 85 | select { |
| 86 | case <-cancelReturned: |
| 87 | case <-time.After(5 * time.Second): |
| 88 | close(releaseStatus) |
| 89 | t.Fatal("CancelSession receipt waited for the status barrier") |
| 90 | } |
| 91 | close(releaseStatus) |
| 92 | } |
| 93 | |
| 94 | // A cancelling status stamped for a turn that already terminated must not turn |
| 95 | // the next admitted turn into a permanently "cancelling" one. |
| 96 | func TestStaleCancellingStatusDoesNotStickToNextTurn(t *testing.T) { |
| 97 | dir := t.TempDir() |
| 98 | done := make(chan event.Event, 4) |
| 99 | c := newOwnedTestController(t, Options{SessionDir: dir, SessionPath: dir + "/session.jsonl", Sink: event.FuncSink(func(e event.Event) { |
| 100 | if e.Kind == event.TurnDone { |
| 101 | done <- e |
| 102 | } |
| 103 | })}) |
| 104 | t.Cleanup(c.Close) |
| 105 | |
| 106 | c.runGuarded(func(context.Context) error { return nil }) |
| 107 | first := waitTurnDoneEvent(t, done) |
| 108 | if first.TurnID == "" { |
| 109 | t.Fatal("first turn has no ledger id") |
| 110 | } |
| 111 | |
| 112 | started := make(chan struct{}) |
| 113 | c.runGuarded(func(ctx context.Context) error { |
| 114 | close(started) |
| 115 | <-ctx.Done() |
| 116 | return ctx.Err() |
| 117 | }) |
| 118 | <-started |
| 119 | c.emitTurnStatus(event.TurnCancelling, first.TurnID) |
| 120 | if st := c.RuntimeStatus(); st.Status == event.TurnCancelling || st.CancelRequested { |
| 121 | t.Fatalf("stale cancelling status leaked into the next turn: %+v", st) |
| 122 | } |
| 123 | c.Cancel() |
| 124 | if second := waitTurnDoneEvent(t, done); second.Status != event.TurnInterrupted { |
| 125 | t.Fatalf("second turn terminal = %q, want interrupted", second.Status) |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | func TestCancellationGraceSealsUncooperativeTurnAndPreservesQueue(t *testing.T) { |
| 130 | dir := t.TempDir() |
| 131 | states := make(chan event.RuntimeStateSnapshot, 32) |
| 132 | sink := &runtimeStateTestSink{Sink: event.Discard, states: states} |
| 133 | exec := agent.New(nil, nil, agent.NewSession("system"), agent.Options{}, event.Discard) |
| 134 | c := newOwnedTestController(t, Options{Executor: exec, SessionDir: dir, SessionPath: dir + "/session.jsonl", Sink: sink}) |
| 135 | c.testCancelGrace = 20 * time.Millisecond |
| 136 | release := make(chan struct{}) |
| 137 | var releaseOnce sync.Once |
| 138 | t.Cleanup(func() { |
| 139 | releaseOnce.Do(func() { close(release) }) |
| 140 | for deadline := time.Now().Add(5 * time.Second); c.Running() && time.Now().Before(deadline); { |
| 141 | time.Sleep(time.Millisecond) |
| 142 | } |
| 143 | c.Close() |
| 144 | }) |
| 145 | |
| 146 | started := make(chan struct{}) |
| 147 | if got := c.runGuarded(func(context.Context) error { |
| 148 | close(started) |
| 149 | <-release // deliberately ignores cancellation |
| 150 | return nil |
| 151 | }); got != turnStarted { |
| 152 | t.Fatalf("admission = %v", got) |
| 153 | } |
| 154 | <-started |
| 155 | c.Cancel() |
| 156 | queuedStarted := make(chan struct{}) |
| 157 | if got := c.runGuarded(func(context.Context) error { close(queuedStarted); return nil }); got != turnParked { |
| 158 | t.Fatalf("cancelling admission = %v, want queued", got) |
| 159 | } |
| 160 | |
| 161 | recovery := runtimeStateAwait(t, states, func(state event.RuntimeStateSnapshot) bool { |
| 162 | return state.Phase == "recovery_required" |
| 163 | }) |
| 164 | if recovery.Recovery == nil || recovery.Recovery.State != "recovery_required" || recovery.Cancellable { |
| 165 | t.Fatalf("recovery snapshot = %+v", recovery) |
| 166 | } |
| 167 | if !c.Running() { |
| 168 | t.Fatal("uncooperative worker ownership was released before exit") |
| 169 | } |
| 170 | |
| 171 | if got := c.runGuarded(func(context.Context) error { return nil }); got != turnDroppedWriteAuthority { |
| 172 | t.Fatalf("recovery admission = %v, want rejected", got) |
| 173 | } |
| 174 | select { |
| 175 | case <-queuedStarted: |
| 176 | t.Fatal("turn started while recovery was required") |
| 177 | default: |
| 178 | } |
| 179 | |
| 180 | // A late semantic result from the sealed worker cannot mutate the durable |
| 181 | // todo projection. |
| 182 | c.sink.Emit(event.Event{Kind: event.ToolResult, Tool: event.Tool{ |
| 183 | Name: "todo_write", TodoWritten: true, |
| 184 | Todos: []event.Todo{{Content: "late", Status: "completed"}}, |
| 185 | }}) |
| 186 | if todos, written := c.turnEventLedger().TodoState(); written || len(todos) != 0 { |
| 187 | t.Fatalf("late todo committed after recovery: written=%v todos=%+v", written, todos) |
| 188 | } |
| 189 | v3 := c.sessionEventStore().Snapshot().Projection |
| 190 | if v3.TodoWritten || len(v3.Todos) != 0 { |
| 191 | t.Fatalf("late todo committed to v3 after recovery: written=%v todos=%+v", v3.TodoWritten, v3.Todos) |
| 192 | } |
| 193 | if v3.Recovery == nil || v3.Recovery.State != "recovery_required" || v3.Recovery.Reason != "cancellation_grace_expired" { |
| 194 | t.Fatalf("v3 recovery seal = %+v", v3.Recovery) |
| 195 | } |
| 196 | |
| 197 | releaseOnce.Do(func() { close(release) }) |
| 198 | deadline := time.Now().Add(5 * time.Second) |
| 199 | for c.Running() && time.Now().Before(deadline) { |
| 200 | time.Sleep(time.Millisecond) |
| 201 | } |
| 202 | if c.Running() || c.RuntimeStateSnapshot().Phase != "recovery_required" { |
| 203 | t.Fatalf("worker exit changed recovery boundary: %+v", c.RuntimeStateSnapshot()) |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | func TestRecoverySealDropsLateTranscriptOutput(t *testing.T) { |
| 208 | dir := t.TempDir() |
| 209 | path := dir + "/session.jsonl" |
| 210 | session := agent.NewSession("system") |
| 211 | session.Add(provider.Message{Role: provider.RoleUser, ID: "user", Content: "do work"}) |
| 212 | session.Add(provider.Message{Role: provider.RoleAssistant, ID: "late-message", Content: "must stay diagnostic-only"}) |
| 213 | exec := agent.New(nil, nil, session, agent.Options{}, event.Discard) |
| 214 | c := newOwnedTestController(t, Options{Executor: exec, SessionDir: dir, SessionPath: path, Sink: event.Discard}) |
| 215 | t.Cleanup(c.Close) |
| 216 | |
| 217 | ledger := c.turnEventLedger() |
| 218 | if _, err := ledger.Begin(); err != nil { |
| 219 | t.Fatal(err) |
| 220 | } |
| 221 | if _, ok, err := ledger.Append(event.Event{Kind: event.TurnStarted}, event.TurnInProgress); err != nil || !ok { |
| 222 | t.Fatalf("append turn start: ok=%v err=%v", ok, err) |
| 223 | } |
| 224 | if _, ok, err := ledger.Append(event.Event{Kind: event.TurnDone, Recovery: &event.RecoveryStatus{State: "recovery_required"}}, event.TurnRecoveryRequired); err != nil || !ok { |
| 225 | t.Fatalf("append recovery seal: ok=%v err=%v", ok, err) |
| 226 | } |
| 227 | c.finishInFlightTurn(1, agent.InFlightTurnMeta{ID: "turn", StartMessageIndex: 1, PreserveUser: true}) |
| 228 | history := c.History() |
| 229 | if len(history) != 2 || history[1].ID != "user" { |
| 230 | t.Fatalf("recovery-sealed history = %+v", history) |
| 231 | } |
| 232 | for _, message := range history { |
| 233 | if message.ID == "late-message" { |
| 234 | t.Fatal("late sealed-worker message entered the durable transcript") |
| 235 | } |
| 236 | } |
| 237 | } |
| 238 |