| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "sync" |
| 7 | "testing" |
| 8 | ) |
| 9 | |
| 10 | type testExecution struct { |
| 11 | mu sync.Mutex |
| 12 | phase RuntimePhase |
| 13 | cancel context.CancelFunc |
| 14 | ctx context.Context |
| 15 | gen uint64 |
| 16 | runtime *Runtime |
| 17 | } |
| 18 | |
| 19 | type blockingRejectExecution struct { |
| 20 | entered chan<- struct{} |
| 21 | release <-chan struct{} |
| 22 | } |
| 23 | |
| 24 | func (e *blockingRejectExecution) Snapshot() RuntimeSnapshot { |
| 25 | return RuntimeSnapshot{Phase: RuntimeIdle} |
| 26 | } |
| 27 | func (e *blockingRejectExecution) Cancel() bool { |
| 28 | e.entered <- struct{}{} |
| 29 | <-e.release |
| 30 | return false |
| 31 | } |
| 32 | |
| 33 | func bindTestExecution(t *testing.T, runtime *Runtime, name string) (context.Context, *testExecution) { |
| 34 | t.Helper() |
| 35 | ctx, cancel := context.WithCancel(context.Background()) |
| 36 | exec := &testExecution{phase: RuntimeRunning, cancel: cancel, ctx: ctx, gen: 1, runtime: runtime} |
| 37 | exec.gen = runtime.BindExecution(exec) |
| 38 | runtime.NoteExecution(exec.gen, RuntimeRunning, name) |
| 39 | t.Cleanup(exec.Finish) |
| 40 | return ctx, exec |
| 41 | } |
| 42 | |
| 43 | func (e *testExecution) Snapshot() RuntimeSnapshot { |
| 44 | e.mu.Lock() |
| 45 | defer e.mu.Unlock() |
| 46 | return RuntimeSnapshot{Phase: e.phase} |
| 47 | } |
| 48 | |
| 49 | func (e *testExecution) Cancel() bool { |
| 50 | e.mu.Lock() |
| 51 | defer e.mu.Unlock() |
| 52 | if e.phase != RuntimeRunning && e.phase != RuntimeCancelling { |
| 53 | return false |
| 54 | } |
| 55 | e.phase = RuntimeCancelling |
| 56 | if e.cancel != nil { |
| 57 | e.cancel() |
| 58 | } |
| 59 | return true |
| 60 | } |
| 61 | |
| 62 | func (e *testExecution) Finish() { |
| 63 | e.mu.Lock() |
| 64 | if e.phase == RuntimeIdle || e.phase == RuntimeClosed { |
| 65 | e.mu.Unlock() |
| 66 | return |
| 67 | } |
| 68 | e.phase = RuntimeIdle |
| 69 | e.mu.Unlock() |
| 70 | if e.runtime != nil { |
| 71 | e.runtime.NoteExecution(e.gen, RuntimeIdle, "") |
| 72 | } |
| 73 | } |
| 74 | |
| 75 | func TestBindExecutionDoesNotSilentlyReplaceExistingOwner(t *testing.T) { |
| 76 | _, runtime := reviewRuntime(t) |
| 77 | first := &testExecution{phase: RuntimeRunning, runtime: runtime} |
| 78 | first.gen = runtime.BindExecution(first) |
| 79 | runtime.NoteExecution(first.gen, RuntimeRunning, "first") |
| 80 | |
| 81 | second := &testExecution{phase: RuntimeRunning, runtime: runtime} |
| 82 | if gen := runtime.BindExecution(second); gen != 0 { |
| 83 | t.Fatalf("second bind generation = %d, want rejection", gen) |
| 84 | } |
| 85 | if !runtime.Cancel() { |
| 86 | t.Fatal("cancel did not reach the original execution owner") |
| 87 | } |
| 88 | if got := first.phase; got != RuntimeCancelling { |
| 89 | t.Fatalf("original owner phase = %s, want cancelling", got) |
| 90 | } |
| 91 | // A runtime with a live execution refuses to close, which would strand its |
| 92 | // writer lease past the test. |
| 93 | first.Finish() |
| 94 | } |
| 95 | |
| 96 | func TestUnbindDoesNotClearNewerExecutionGeneration(t *testing.T) { |
| 97 | _, runtime := reviewRuntime(t) |
| 98 | first := &testExecution{phase: RuntimeRunning, runtime: runtime} |
| 99 | second := &testExecution{phase: RuntimeRunning, runtime: runtime} |
| 100 | first.gen = runtime.BindExecution(first) |
| 101 | second.gen = runtime.ReplaceExecution(first.gen, second) |
| 102 | if second.gen == 0 { |
| 103 | t.Fatal("replace idle execution") |
| 104 | } |
| 105 | runtime.NoteExecution(second.gen, RuntimeRunning, "new") |
| 106 | runtime.UnbindExecution(first.gen) |
| 107 | if !runtime.Cancel() { |
| 108 | t.Fatal("new generation lost cancel after old unbind") |
| 109 | } |
| 110 | if got := runtime.StateSnapshot().Phase; got != RuntimeCancelling { |
| 111 | t.Fatalf("phase = %s, want cancelling", got) |
| 112 | } |
| 113 | runtime.NoteExecution(first.gen, RuntimeIdle, "") |
| 114 | if got := runtime.StateSnapshot().Phase; got != RuntimeCancelling { |
| 115 | t.Fatalf("old generation cleared new phase: %s", got) |
| 116 | } |
| 117 | second.Finish() |
| 118 | } |
| 119 | |
| 120 | func TestSessionAcceptsCancelHistoryWhileCancelling(t *testing.T) { |
| 121 | _, runtime := reviewRuntime(t) |
| 122 | _, exec := bindTestExecution(t, runtime, "turn") |
| 123 | if !runtime.Cancel() { |
| 124 | t.Fatal("cancel") |
| 125 | } |
| 126 | payload := []byte(`{"messages":[],"reason":"cancel-or-recovery-rewrite"}`) |
| 127 | if _, err := runtime.Session().Append(t.Context(), Batch{ |
| 128 | OperationID: "history-replace", |
| 129 | TurnID: "turn-1", |
| 130 | Events: []Event{{Kind: "history/replace", Payload: payload}}, |
| 131 | }); err != nil { |
| 132 | t.Fatalf("history/replace during cancel: %v", err) |
| 133 | } |
| 134 | if _, err := runtime.Session().Append(t.Context(), Batch{ |
| 135 | OperationID: "turn-end", |
| 136 | TurnID: "turn-1", |
| 137 | Events: []Event{{Kind: "turn/end", Payload: []byte(`{"status":"interrupted"}`)}}, |
| 138 | }); err != nil { |
| 139 | t.Fatalf("turn/end during cancel: %v", err) |
| 140 | } |
| 141 | exec.Finish() |
| 142 | if got := runtime.StateSnapshot().Phase; got != RuntimeIdle { |
| 143 | t.Fatalf("phase after finish = %s", got) |
| 144 | } |
| 145 | } |
| 146 | |
| 147 | func TestOldControllerCannotIdleNewGeneration(t *testing.T) { |
| 148 | _, runtime := reviewRuntime(t) |
| 149 | old := &testExecution{phase: RuntimeRunning, runtime: runtime} |
| 150 | old.gen = runtime.BindExecution(old) |
| 151 | next := &testExecution{phase: RuntimeIdle, runtime: runtime} |
| 152 | next.gen = runtime.ReplaceExecution(old.gen, next) |
| 153 | if next.gen == 0 { |
| 154 | t.Fatal("replace idle execution") |
| 155 | } |
| 156 | runtime.NoteExecution(next.gen, RuntimeRunning, "new") |
| 157 | runtime.NoteExecution(old.gen, RuntimeIdle, "") |
| 158 | if got := runtime.StateSnapshot().Phase; got != RuntimeRunning { |
| 159 | t.Fatalf("old finish cleared new turn: %s", got) |
| 160 | } |
| 161 | // next was built idle, so Finish is a no-op for it; idle the generation the |
| 162 | // runtime actually observes or the runtime stays busy and cannot close. |
| 163 | runtime.NoteExecution(next.gen, RuntimeIdle, "") |
| 164 | } |
| 165 | |
| 166 | func TestReplaceExecutionRejectsBusyOwner(t *testing.T) { |
| 167 | _, runtime := reviewRuntime(t) |
| 168 | old := &testExecution{phase: RuntimeRunning, runtime: runtime} |
| 169 | old.gen = runtime.BindExecution(old) |
| 170 | runtime.NoteExecution(old.gen, RuntimeRunning, "old") |
| 171 | if gen := runtime.ReplaceExecution(old.gen, &testExecution{}); gen != 0 { |
| 172 | t.Fatalf("busy replacement generation = %d, want rejection", gen) |
| 173 | } |
| 174 | if got := runtime.StateSnapshot().Phase; got != RuntimeRunning { |
| 175 | t.Fatalf("phase after rejected replacement = %s, want running", got) |
| 176 | } |
| 177 | old.Finish() |
| 178 | } |
| 179 | |
| 180 | func TestStaleExecutionGenerationCannotCommitPreparedBatch(t *testing.T) { |
| 181 | _, runtime := reviewRuntime(t) |
| 182 | old := &testExecution{phase: RuntimeIdle, runtime: runtime} |
| 183 | old.gen = runtime.BindExecution(old) |
| 184 | prepared, err := runtime.Session().PrepareBatchContext(t.Context(), "old-config", Batch{ |
| 185 | Events: []Event{{Kind: "session/config", Payload: []byte(`{"modelRef":"old"}`)}}, |
| 186 | }) |
| 187 | if err != nil { |
| 188 | t.Fatal(err) |
| 189 | } |
| 190 | next := &testExecution{phase: RuntimeIdle, runtime: runtime} |
| 191 | next.gen = runtime.ReplaceExecution(old.gen, next) |
| 192 | if next.gen == 0 { |
| 193 | t.Fatal("replace idle execution") |
| 194 | } |
| 195 | if _, err := runtime.CommitPreparedForExecution(old.gen, prepared); !errors.Is(err, ErrStaleExecution) { |
| 196 | t.Fatalf("stale commit error = %v, want %v", err, ErrStaleExecution) |
| 197 | } |
| 198 | if got := runtime.Session().ExecutionSnapshot().EventSequence; got != 0 { |
| 199 | t.Fatalf("stale generation committed sequence %d", got) |
| 200 | } |
| 201 | current, err := runtime.Session().PrepareBatchContext(t.Context(), "new-config", Batch{ |
| 202 | Events: []Event{{Kind: "session/config", Payload: []byte(`{"modelRef":"new"}`)}}, |
| 203 | }) |
| 204 | if err != nil { |
| 205 | t.Fatal(err) |
| 206 | } |
| 207 | if _, err := runtime.CommitPreparedForExecution(next.gen, current); err != nil { |
| 208 | t.Fatalf("current generation commit: %v", err) |
| 209 | } |
| 210 | } |
| 211 | |
| 212 | func TestCancelRetriesAcrossExecutionCutover(t *testing.T) { |
| 213 | _, runtime := reviewRuntime(t) |
| 214 | entered := make(chan struct{}, 1) |
| 215 | release := make(chan struct{}) |
| 216 | old := &blockingRejectExecution{entered: entered, release: release} |
| 217 | oldGeneration := runtime.BindExecution(old) |
| 218 | if oldGeneration == 0 { |
| 219 | t.Fatal("bind outgoing execution") |
| 220 | } |
| 221 | result := make(chan bool, 1) |
| 222 | go func() { result <- runtime.Cancel() }() |
| 223 | <-entered |
| 224 | next := &testExecution{phase: RuntimeRunning, runtime: runtime} |
| 225 | next.gen = runtime.ReplaceExecution(oldGeneration, next) |
| 226 | if next.gen == 0 { |
| 227 | t.Fatal("replace execution while cancel is in flight") |
| 228 | } |
| 229 | close(release) |
| 230 | if accepted := <-result; !accepted { |
| 231 | t.Fatal("cancel was lost across execution cutover") |
| 232 | } |
| 233 | if got := next.phase; got != RuntimeCancelling { |
| 234 | t.Fatalf("replacement phase = %s, want cancelling", got) |
| 235 | } |
| 236 | } |
| 237 | |
| 238 | func TestRuntimeFinalizingIsBusy(t *testing.T) { |
| 239 | if !RuntimeFinalizing.Busy() { |
| 240 | t.Fatal("finalizing phase must remain busy until terminal commit finishes") |
| 241 | } |
| 242 | } |
| 243 |