diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index ca1746f..3692645 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -26,11 +26,20 @@ type callRecord struct { // proceeds it records the call and returns a deterministic reply. A nil err // produces a reply holding the *incoming* messages as text, so a test can assert // exactly what context the resumed turn carried. +// +// Like the real loop runner, it never executes a turn whose context is already +// canceled, and it counts those canceled returns so a test can wait for an +// interrupted turn to actually stop (waitCanceled) before releasing the next +// ticket. Without that, a turn that was canceled *while parked in the select +// below* can still win the ticket case in the same scheduler step (both cases are +// ready) and be recorded as if it had run — a test-harness artifact, not an +// orchestrator ordering guarantee. type gatedRunner struct { - mu sync.Mutex - calls []callRecord - tickets chan struct{} - err error + mu sync.Mutex + calls []callRecord + canceled int + tickets chan struct{} + err error } func newGatedRunner(cap int) *gatedRunner { @@ -38,9 +47,14 @@ func newGatedRunner(cap int) *gatedRunner { } func (g *gatedRunner) Run(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error) { + if err := ctx.Err(); err != nil { + g.recordCanceled() + return nil, err + } select { case <-g.tickets: case <-ctx.Done(): + g.recordCanceled() return nil, ctx.Err() } g.mu.Lock() @@ -78,6 +92,47 @@ func (g *gatedRunner) count() int { return len(g.calls) } +// recordCanceled notes one invocation that returned because its context was +// canceled, i.e. a turn that never executed. +func (g *gatedRunner) recordCanceled() { + g.mu.Lock() + g.canceled++ + g.mu.Unlock() +} + +// waitCanceled blocks until at least n invocations have stopped because their +// context was canceled, reporting false when that never happens within the +// deadline. It is the barrier a test uses between an Interrupt and the release of +// the next ticket: once it holds, no canceled turn can still consume a ticket. +func (g *gatedRunner) waitCanceled(n int) bool { + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + g.mu.Lock() + ok := g.canceled >= n + g.mu.Unlock() + if ok { + return true + } + time.Sleep(time.Millisecond) + } + return false +} + +// waitIdle blocks until the session reports idle with an empty pending queue, +// reporting false when it never settles within the deadline. It lets a test assert +// the settled state deterministically instead of sleeping, and turns "a dropped +// turn was resumed anyway" into a test failure rather than a silent pass. +func waitIdle(o *Orchestrator, sessionID string) bool { + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if !o.Busy(sessionID) && o.QueueDepth(sessionID) == 0 { + return true + } + time.Sleep(time.Millisecond) + } + return false +} + // eventRec is one state transition captured by the recording sink. type eventRec struct { sessionID, typ string @@ -556,7 +611,7 @@ func TestSubmitPendingBoundedRejectsOverflow(t *testing.T) { func TestInterruptCancelsTurnAndClearsPending(t *testing.T) { t.Parallel() run := newGatedRunner(2) - o, _ := seedOrchestrator(run, 5) + o, sink := seedOrchestrator(run, 5) if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate { t.Fatalf("A = %+v, want Immediate", sr) @@ -574,6 +629,14 @@ func TestInterruptCancelsTurnAndClearsPending(t *testing.T) { t.Fatalf("session not idle after interrupt: busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1")) } + // The cancel must reach the in-flight turn (design/15 §3: interrupt is a + // graceful context cancel) and it must stop before any further turn is + // released: a canceled turn never executes, so it can never consume the ticket + // meant for the next one. + if !run.waitCanceled(1) { + t.Fatal("interrupted turn never observed its context cancellation") + } + // A new message after the interrupt must start fresh (Immediate), and the // canceled chain must not have resumed B. if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("C")}, loop.OwnerEnv()); !sr.Immediate { @@ -587,6 +650,28 @@ func TestInterruptCancelsTurnAndClearsPending(t *testing.T) { if messageText(got[0].messages[0]) != "C" { t.Fatalf("only surviving call = %v, want the C turn", messageTextList(got[0].messages)) } + + // C's chain settles with nothing left to run: had the interrupt failed to + // cancel A or to clear B, the surviving chain would have resumed a turn and the + // session would never go idle again. + if !waitIdle(o, "s1") { + t.Fatalf("session never settled idle: busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1")) + } + if n := run.count(); n != 1 { + t.Fatalf("runner calls = %d, want 1 (neither the canceled A nor the dropped B may execute)", n) + } + // The canceled turn's completion is a no-op for session state (its generation is + // stale), so it must not emit the idle event of the turn now running: exactly + // one session_idle belongs to this scenario (C settling). + idles := 0 + for _, ty := range sink.types() { + if ty == EventSessionIdle { + idles++ + } + } + if idles != 1 { + t.Fatalf("session_idle events = %d, want 1 (a canceled turn must not settle the fresh chain)", idles) + } } func TestSubmitConcurrentSameSessionIsSafeAndBounded(t *testing.T) {