| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "reflect" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/event" |
| 13 | ) |
| 14 | |
| 15 | type runtimeStateTestSink struct { |
| 16 | event.Sink |
| 17 | states chan event.RuntimeStateSnapshot |
| 18 | } |
| 19 | |
| 20 | func (s *runtimeStateTestSink) RuntimeStateChanged(state event.RuntimeStateSnapshot) { |
| 21 | s.states <- state |
| 22 | } |
| 23 | |
| 24 | func runtimeStateAwait(t *testing.T, states <-chan event.RuntimeStateSnapshot, accept func(event.RuntimeStateSnapshot) bool) event.RuntimeStateSnapshot { |
| 25 | t.Helper() |
| 26 | timer := time.NewTimer(5 * time.Second) |
| 27 | defer timer.Stop() |
| 28 | var last event.RuntimeStateSnapshot |
| 29 | for { |
| 30 | select { |
| 31 | case last = <-states: |
| 32 | if accept(last) { |
| 33 | return last |
| 34 | } |
| 35 | case <-timer.C: |
| 36 | t.Fatalf("runtime publication did not reach expected state; last=%+v", last) |
| 37 | } |
| 38 | } |
| 39 | } |
| 40 | |
| 41 | func runtimeStateSignal(t *testing.T, signal <-chan struct{}, description string) { |
| 42 | t.Helper() |
| 43 | select { |
| 44 | case <-signal: |
| 45 | case <-time.After(5 * time.Second): |
| 46 | t.Fatal(description) |
| 47 | } |
| 48 | } |
| 49 | |
| 50 | func assertRuntimeStateSameVersion(t *testing.T, published, current event.RuntimeStateSnapshot) { |
| 51 | t.Helper() |
| 52 | if published.RuntimeEpoch == current.RuntimeEpoch && published.Revision == current.Revision && !reflect.DeepEqual(published, current) { |
| 53 | t.Fatalf("same runtime version has different contents: published=%+v current=%+v", published, current) |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | func TestRuntimeStateSnapshotFinishingAndFinalPublication(t *testing.T) { |
| 58 | isolateControlConfigHome(t) |
| 59 | entered, release := make(chan struct{}, 1), make(chan struct{}) |
| 60 | releaseDone := sync.OnceFunc(func() { close(release) }) |
| 61 | sink := &runtimeStateTestSink{Sink: holdFinishingWindow(release, entered, nil), states: make(chan event.RuntimeStateSnapshot, 32)} |
| 62 | c := newOwnedTestController(t, Options{SessionDir: t.TempDir(), Sink: sink}) |
| 63 | defer c.Close() |
| 64 | defer releaseDone() |
| 65 | initial := c.RuntimeStateSnapshot() |
| 66 | if initial.SchemaVersion != 1 || initial.RuntimeEpoch == "" || initial.Revision == 0 || initial.Phase != "idle" { |
| 67 | t.Fatalf("invalid initial contract: %+v", initial) |
| 68 | } |
| 69 | if second := c.RuntimeStateSnapshot(); !reflect.DeepEqual(second, initial) { |
| 70 | t.Fatalf("reading runtime state changed the snapshot: first=%+v second=%+v", initial, second) |
| 71 | } |
| 72 | runRelease := make(chan struct{}) |
| 73 | releaseBody := sync.OnceFunc(func() { close(runRelease) }) |
| 74 | defer releaseBody() |
| 75 | c.runGuarded(func(context.Context) error { <-runRelease; return nil }) |
| 76 | executing := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "executing" && s.TurnID != "" }) |
| 77 | if !executing.Running || !executing.Cancellable || executing.Activity != "thinking" || executing.Revision <= initial.Revision { |
| 78 | t.Fatalf("invalid execution state: %+v", executing) |
| 79 | } |
| 80 | releaseBody() |
| 81 | runtimeStateSignal(t, entered, "TurnDone did not enter finishing window") |
| 82 | finishing := c.RuntimeStateSnapshot() |
| 83 | if finishing.Phase != "finishing" || !finishing.Running || finishing.Activity != "" || finishing.Cancellable { |
| 84 | t.Fatalf("finishing must keep admission closed without presenting thought/cancel: %+v", finishing) |
| 85 | } |
| 86 | if !c.RuntimeStatus().Running { |
| 87 | t.Fatal("legacy running guard opened during TurnDone fan-out") |
| 88 | } |
| 89 | publishedFinishing := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "finishing" }) |
| 90 | assertRuntimeStateSameVersion(t, publishedFinishing, c.RuntimeStateSnapshot()) |
| 91 | releaseDone() |
| 92 | completed := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "idle" && s.Revision > executing.Revision }) |
| 93 | if completed.Running || completed.Cancellable || completed.PendingPrompt || completed.Activity != "" || completed.BackgroundJobs != 0 { |
| 94 | t.Fatalf("completed publication retained active work: %+v", completed) |
| 95 | } |
| 96 | if completed.RuntimeEpoch != initial.RuntimeEpoch || completed.Revision <= publishedFinishing.Revision { |
| 97 | t.Fatalf("completion lost instance identity or ordering: finishing=%+v completed=%+v", publishedFinishing, completed) |
| 98 | } |
| 99 | assertRuntimeStateSameVersion(t, completed, c.RuntimeStateSnapshot()) |
| 100 | } |
| 101 | |
| 102 | func TestRuntimeStateQueuedTurnNeverPublishesPreviousIdleOverNext(t *testing.T) { |
| 103 | isolateControlConfigHome(t) |
| 104 | entered, release := make(chan struct{}, 1), make(chan struct{}) |
| 105 | releaseDone := sync.OnceFunc(func() { close(release) }) |
| 106 | sink := &runtimeStateTestSink{Sink: holdFinishingWindow(release, entered, nil), states: make(chan event.RuntimeStateSnapshot, 64)} |
| 107 | c := newOwnedTestController(t, Options{SessionDir: t.TempDir(), Sink: sink}) |
| 108 | defer c.Close() |
| 109 | defer releaseDone() |
| 110 | c.runGuarded(func(context.Context) error { return nil }) |
| 111 | runtimeStateSignal(t, entered, "first turn did not enter finishing window") |
| 112 | first := c.RuntimeStateSnapshot() |
| 113 | secondStarted, secondRelease := make(chan struct{}), make(chan struct{}) |
| 114 | releaseSecond := sync.OnceFunc(func() { close(secondRelease) }) |
| 115 | defer releaseSecond() |
| 116 | if admission := c.runGuarded(func(context.Context) error { close(secondStarted); <-secondRelease; return nil }); admission != turnParked { |
| 117 | t.Fatalf("follow-up admission=%v; expected parked", admission) |
| 118 | } |
| 119 | releaseDone() |
| 120 | runtimeStateSignal(t, secondStarted, "parked second turn did not start") |
| 121 | second := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { |
| 122 | if s.Revision > first.Revision && s.Phase == "idle" { |
| 123 | t.Fatalf("old completion published idle while queued turn owns admission: %+v", s) |
| 124 | } |
| 125 | return s.Revision > first.Revision && s.Phase == "executing" && s.TurnID != "" && s.TurnID != first.TurnID |
| 126 | }) |
| 127 | if second.RuntimeEpoch != first.RuntimeEpoch || second.Revision <= first.Revision { |
| 128 | t.Fatalf("queued turn lost runtime ordering: first=%+v second=%+v", first, second) |
| 129 | } |
| 130 | assertRuntimeStateSameVersion(t, second, c.RuntimeStateSnapshot()) |
| 131 | releaseSecond() |
| 132 | final := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "idle" && s.Revision > second.Revision }) |
| 133 | if final.TurnID != second.TurnID && final.TurnID != "" { |
| 134 | t.Fatalf("old turn replaced the completed next turn: second=%+v final=%+v", second, final) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | func TestRuntimeStatePromptCancellationAndClosed(t *testing.T) { |
| 139 | for _, kind := range []string{"ask", "approval"} { |
| 140 | t.Run(kind, func(t *testing.T) { |
| 141 | isolateControlConfigHome(t) |
| 142 | requests := make(chan struct{}, 1) |
| 143 | sink := &runtimeStateTestSink{Sink: event.FuncSink(func(e event.Event) { |
| 144 | if e.Kind == event.AskRequest || e.Kind == event.ApprovalRequest { |
| 145 | requests <- struct{}{} |
| 146 | } |
| 147 | }), states: make(chan event.RuntimeStateSnapshot, 64)} |
| 148 | c := newOwnedTestController(t, Options{SessionDir: t.TempDir(), Sink: sink}) |
| 149 | defer c.Close() |
| 150 | if kind == "ask" { |
| 151 | c.runner = &askBlockingRunner{c: c} |
| 152 | } else { |
| 153 | c.runner = &approvalBlockingRunner{c: c} |
| 154 | } |
| 155 | c.Send("isolated interaction") |
| 156 | runtimeStateSignal(t, requests, "prompt was not requested") |
| 157 | pending := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.PendingPrompt }) |
| 158 | if !pending.Running || !pending.Cancellable { |
| 159 | t.Fatalf("pending prompt is not actionable: %+v", pending) |
| 160 | } |
| 161 | c.Cancel() |
| 162 | c.Cancel() |
| 163 | idle := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "idle" && s.Revision > pending.Revision }) |
| 164 | if idle.PendingPrompt || idle.CancelRequested || idle.Cancellable || idle.Running { |
| 165 | t.Fatalf("cancel left stale prompt state: %+v", idle) |
| 166 | } |
| 167 | c.Close() |
| 168 | closed := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "closed" }) |
| 169 | if closed.Running || closed.PendingPrompt || closed.Cancellable || closed.Activity != "" || closed.Revision <= idle.Revision { |
| 170 | t.Fatalf("invalid closed snapshot: %+v", closed) |
| 171 | } |
| 172 | assertRuntimeStateSameVersion(t, closed, c.RuntimeStateSnapshot()) |
| 173 | }) |
| 174 | } |
| 175 | } |
| 176 | |
| 177 | func TestRuntimeStateCloseDuringFinishingCannotResurrectActivity(t *testing.T) { |
| 178 | isolateControlConfigHome(t) |
| 179 | entered, release := make(chan struct{}, 1), make(chan struct{}) |
| 180 | releaseDone := sync.OnceFunc(func() { close(release) }) |
| 181 | sink := &runtimeStateTestSink{Sink: holdFinishingWindow(release, entered, nil), states: make(chan event.RuntimeStateSnapshot, 32)} |
| 182 | c := newOwnedTestController(t, Options{SessionDir: t.TempDir(), Sink: sink}) |
| 183 | defer c.Close() |
| 184 | defer releaseDone() |
| 185 | c.runGuarded(func(context.Context) error { return nil }) |
| 186 | runtimeStateSignal(t, entered, "turn did not enter finishing window") |
| 187 | c.Close() |
| 188 | closed := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "closed" }) |
| 189 | if closed.Running || closed.Cancellable || closed.PendingPrompt || closed.Activity != "" { |
| 190 | t.Fatalf("closed runtime retained finishing activity: %+v", closed) |
| 191 | } |
| 192 | releaseDone() |
| 193 | // A rejected new admission is a synchronous observation after Close. The |
| 194 | // released old completion may still run, but it must never reopen the gate. |
| 195 | if got := c.runGuarded(func(context.Context) error { t.Error("closed runtime admitted a new turn"); return nil }); got != turnDroppedClosed { |
| 196 | t.Fatalf("admission after close=%v, want closed", got) |
| 197 | } |
| 198 | current := c.RuntimeStateSnapshot() |
| 199 | if current.Phase != "closed" || current.Running || current.Activity != "" { |
| 200 | t.Fatalf("old completion resurrected closed runtime: %+v", current) |
| 201 | } |
| 202 | assertRuntimeStateSameVersion(t, closed, current) |
| 203 | } |
| 204 | |
| 205 | func TestRuntimeStateStreamingActivityClearsOnCompletion(t *testing.T) { |
| 206 | isolateControlConfigHome(t) |
| 207 | sink := &runtimeStateTestSink{Sink: event.Discard, states: make(chan event.RuntimeStateSnapshot, 32)} |
| 208 | c := newOwnedTestController(t, Options{SessionDir: t.TempDir(), Sink: sink}) |
| 209 | defer c.Close() |
| 210 | release := make(chan struct{}) |
| 211 | releaseBody := sync.OnceFunc(func() { close(release) }) |
| 212 | defer releaseBody() |
| 213 | c.runGuarded(func(context.Context) error { |
| 214 | c.sink.Emit(event.Event{Kind: event.Text, Text: "isolated visible output"}) |
| 215 | <-release |
| 216 | return nil |
| 217 | }) |
| 218 | streaming := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Activity == "streaming" }) |
| 219 | if streaming.Phase != "executing" || !streaming.Running { |
| 220 | t.Fatalf("streaming state is not executing: %+v", streaming) |
| 221 | } |
| 222 | releaseBody() |
| 223 | idle := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { return s.Phase == "idle" && s.Revision > streaming.Revision }) |
| 224 | if idle.Activity != "" || idle.Running { |
| 225 | t.Fatalf("completed runtime retained streaming activity: %+v", idle) |
| 226 | } |
| 227 | assertRuntimeStateSameVersion(t, idle, c.RuntimeStateSnapshot()) |
| 228 | } |
| 229 | |
| 230 | type runtimeStateSlowTestSink struct { |
| 231 | entered chan struct{} |
| 232 | release chan struct{} |
| 233 | done chan struct{} |
| 234 | states chan event.RuntimeStateSnapshot |
| 235 | once sync.Once |
| 236 | } |
| 237 | |
| 238 | func (s *runtimeStateSlowTestSink) Emit(e event.Event) { |
| 239 | if e.Kind == event.TurnDone { |
| 240 | s.done <- struct{}{} |
| 241 | } |
| 242 | } |
| 243 | func (s *runtimeStateSlowTestSink) RuntimeStateChanged(state event.RuntimeStateSnapshot) { |
| 244 | s.once.Do(func() { close(s.entered); <-s.release }) |
| 245 | s.states <- state |
| 246 | } |
| 247 | |
| 248 | func TestRuntimeStateSlowObserverDoesNotBlockTurnAndRetainsFinalSnapshot(t *testing.T) { |
| 249 | isolateControlConfigHome(t) |
| 250 | sink := &runtimeStateSlowTestSink{entered: make(chan struct{}), release: make(chan struct{}), done: make(chan struct{}, 1), states: make(chan event.RuntimeStateSnapshot, 32)} |
| 251 | releaseObserver := sync.OnceFunc(func() { close(sink.release) }) |
| 252 | c := newOwnedTestController(t, Options{SessionDir: t.TempDir(), Sink: sink}) |
| 253 | defer c.Close() |
| 254 | defer releaseObserver() |
| 255 | bodyRelease := make(chan struct{}) |
| 256 | releaseBody := sync.OnceFunc(func() { close(bodyRelease) }) |
| 257 | defer releaseBody() |
| 258 | c.runGuarded(func(context.Context) error { <-bodyRelease; return nil }) |
| 259 | runtimeStateSignal(t, sink.entered, "runtime observer did not enter") |
| 260 | releaseBody() |
| 261 | runtimeStateSignal(t, sink.done, "slow runtime observer blocked turn completion") |
| 262 | releaseObserver() |
| 263 | final := runtimeStateAwait(t, sink.states, func(state event.RuntimeStateSnapshot) bool { |
| 264 | return state.Phase == "idle" && state.TurnStatus == event.TurnCompleted |
| 265 | }) |
| 266 | if final.Running || final.Activity != "" { |
| 267 | t.Fatalf("slow observer lost final idle: %+v", final) |
| 268 | } |
| 269 | assertRuntimeStateSameVersion(t, final, c.RuntimeStateSnapshot()) |
| 270 | } |
| 271 | |
| 272 | type runtimeStateFailureRunner func(context.Context, string) error |
| 273 | |
| 274 | func (r runtimeStateFailureRunner) Run(ctx context.Context, input string) error { |
| 275 | return r(ctx, input) |
| 276 | } |
| 277 | |
| 278 | func TestRuntimeStateRunnerFailuresPublishIdleAndPreserveFailure(t *testing.T) { |
| 279 | for _, failure := range []string{"error", "panic"} { |
| 280 | t.Run(failure, func(t *testing.T) { |
| 281 | isolateControlConfigHome(t) |
| 282 | started, release := make(chan struct{}), make(chan struct{}) |
| 283 | releaseRunner := sync.OnceFunc(func() { close(release) }) |
| 284 | done := make(chan event.Event, 2) |
| 285 | sink := &runtimeStateTestSink{states: make(chan event.RuntimeStateSnapshot, 64), Sink: event.FuncSink(func(e event.Event) { |
| 286 | if e.Kind == event.TurnDone { |
| 287 | done <- e |
| 288 | } |
| 289 | })} |
| 290 | message := "runtime-state " + failure + " fixture" |
| 291 | runner := runtimeStateFailureRunner(func(context.Context, string) error { |
| 292 | close(started) |
| 293 | <-release |
| 294 | if failure == "panic" { |
| 295 | panic(message) |
| 296 | } |
| 297 | return errors.New(message) |
| 298 | }) |
| 299 | c := newOwnedTestController(t, Options{SessionDir: t.TempDir(), Sink: sink, Runner: runner}) |
| 300 | defer c.Close() |
| 301 | defer releaseRunner() |
| 302 | c.Send("exercise isolated runner failure") |
| 303 | runtimeStateSignal(t, started, "runner did not start") |
| 304 | executing := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { |
| 305 | return s.Phase == "executing" && s.Running && s.Cancellable && s.TurnID != "" |
| 306 | }) |
| 307 | releaseRunner() |
| 308 | idle := runtimeStateAwait(t, sink.states, func(s event.RuntimeStateSnapshot) bool { |
| 309 | return s.Phase == "idle" && s.Revision > executing.Revision |
| 310 | }) |
| 311 | if idle.Running || idle.PendingPrompt || idle.CancelRequested || idle.Cancellable || idle.Activity != "" || idle.BackgroundJobs != 0 { |
| 312 | t.Fatalf("%s retained active state in final publication: %+v", failure, idle) |
| 313 | } |
| 314 | if idle.RuntimeEpoch != executing.RuntimeEpoch || idle.TurnStatus != event.TurnFailed { |
| 315 | t.Fatalf("%s lost failure identity/status: executing=%+v idle=%+v", failure, executing, idle) |
| 316 | } |
| 317 | terminal := waitTurnDoneEvent(t, done) |
| 318 | if terminal.Err == nil || !strings.Contains(terminal.Err.Error(), message) || terminal.Status != event.TurnFailed || terminal.Cancelled { |
| 319 | t.Fatalf("%s failure notification was lost or reclassified: %+v", failure, terminal) |
| 320 | } |
| 321 | // The final idle publication follows synchronous TurnDone delivery. |
| 322 | // Therefore another queued terminal notification would be a duplicate. |
| 323 | select { |
| 324 | case extra := <-done: |
| 325 | t.Fatalf("%s emitted duplicate terminal notification: %+v", failure, extra) |
| 326 | default: |
| 327 | } |
| 328 | assertRuntimeStateSameVersion(t, idle, c.RuntimeStateSnapshot()) |
| 329 | }) |
| 330 | } |
| 331 | } |
| 332 |