dtoro
39e9227fdb
The agent could run two turns for one session at once (a reconnect resumed
while the live turn was still going), and their interleaved tool calls
corrupted the activity panel, fabricated a confusing "parallel/nested"
sequence, and made tasks feel stuck/never-ending. Several UX gaps compounded it.
Turn serialization (F1):
- turnGate: at most one in-flight turn per session. Background resume paths
(continuation worker, idle sweep, answer-question, /resume, reconnect)
skip non-blocking when busy; the live chat path waits briefly then bails
cleanly instead of stacking a second turn.
- resumeSession returns whether it ran; continueSession marks an execution
"continued" only after a real run (review P0) so a busy-skip can't lose a
finished-execution result. Idle nudge bumps only after delivery (P1).
Connection state (F2/F3, web):
- humanize/bucket raw errors ("model connection dropped..."); one surface
per drop; a terminal task.status event clears stuck streaming/disconnected
state and dismisses the connection toast. Reconnect no longer spawns turns.
Streaming where you look (F4, web):
- live command output in the global activity timeline and in the inline
tool card (auto-opened, tail-pinned) -- not just the per-window rail.
Other (web): artifact/knowledge deep links (F5); step-first stable
"thinking" headline (F6); stable chat layout, no empty->content reflow (F7);
lazy event sync (P2.2); reconnect skips a terminal session (P2.1).
VERSION: 0.14.2 -> 0.15.0
115 lines
2.8 KiB
Go
115 lines
2.8 KiB
Go
package main
|
|
|
|
import (
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestTurnGate_NonBlockingSkipsWhenBusy(t *testing.T) {
|
|
g := newTurnGate()
|
|
if !g.acquire("s1", 0) {
|
|
t.Fatal("first non-blocking acquire should succeed on a free session")
|
|
}
|
|
// A second non-blocking acquire (a background resume) must skip, not queue.
|
|
if g.acquire("s1", 0) {
|
|
t.Fatal("second non-blocking acquire should fail while a turn is active")
|
|
}
|
|
// A different session is independent.
|
|
if !g.acquire("s2", 0) {
|
|
t.Fatal("acquire on a different session should succeed")
|
|
}
|
|
g.release("s2")
|
|
g.release("s1")
|
|
// After release, the session is free again.
|
|
if !g.acquire("s1", 0) {
|
|
t.Fatal("acquire should succeed again after release")
|
|
}
|
|
g.release("s1")
|
|
}
|
|
|
|
func TestTurnGate_BlockingAcquireWaitsForRelease(t *testing.T) {
|
|
g := newTurnGate()
|
|
if !g.acquire("s1", 0) {
|
|
t.Fatal("first acquire should succeed")
|
|
}
|
|
|
|
got := make(chan bool, 1)
|
|
go func() { got <- g.acquire("s1", 2*time.Second) }()
|
|
|
|
select {
|
|
case <-got:
|
|
t.Fatal("blocking acquire should wait, not return before release")
|
|
case <-time.After(50 * time.Millisecond):
|
|
// expected: still waiting
|
|
}
|
|
|
|
g.release("s1")
|
|
select {
|
|
case ok := <-got:
|
|
if !ok {
|
|
t.Fatal("blocking acquire should succeed after release")
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("blocking acquire did not return after release")
|
|
}
|
|
g.release("s1")
|
|
}
|
|
|
|
func TestTurnGate_BlockingAcquireTimesOut(t *testing.T) {
|
|
g := newTurnGate()
|
|
g.acquire("s1", 0) // hold the permit
|
|
|
|
start := time.Now()
|
|
if g.acquire("s1", 60*time.Millisecond) {
|
|
t.Fatal("acquire should time out while permit is held")
|
|
}
|
|
if elapsed := time.Since(start); elapsed < 50*time.Millisecond {
|
|
t.Fatalf("acquire returned too fast (%v); expected to wait ~60ms", elapsed)
|
|
}
|
|
g.release("s1")
|
|
}
|
|
|
|
// TestTurnGate_SingleFlightConcurrent is the core F1 guarantee: many concurrent
|
|
// background acquirers on the SAME session, exactly one runs at a time. This is
|
|
// the property that prevents two turns interleaving tool calls.
|
|
func TestTurnGate_SingleFlightConcurrent(t *testing.T) {
|
|
g := newTurnGate()
|
|
const n = 50
|
|
var inFlight, maxInFlight int64
|
|
var runs int64
|
|
var wg sync.WaitGroup
|
|
wg.Add(n)
|
|
start := make(chan struct{})
|
|
for i := 0; i < n; i++ {
|
|
go func() {
|
|
defer wg.Done()
|
|
<-start
|
|
if !g.acquire("shared", 0) { // background-style: skip if busy
|
|
return
|
|
}
|
|
defer g.release("shared")
|
|
cur := atomic.AddInt64(&inFlight, 1)
|
|
for {
|
|
m := atomic.LoadInt64(&maxInFlight)
|
|
if cur <= m || atomic.CompareAndSwapInt64(&maxInFlight, m, cur) {
|
|
break
|
|
}
|
|
}
|
|
atomic.AddInt64(&runs, 1)
|
|
time.Sleep(2 * time.Millisecond)
|
|
atomic.AddInt64(&inFlight, -1)
|
|
}()
|
|
}
|
|
close(start)
|
|
wg.Wait()
|
|
|
|
if maxInFlight != 1 {
|
|
t.Fatalf("max in-flight turns = %d, want 1 (turns must not overlap)", maxInFlight)
|
|
}
|
|
if runs == 0 {
|
|
t.Fatal("expected at least one turn to run")
|
|
}
|
|
}
|