fix(forwarder): reset bidi append seq when Cursor reuses request_id

Cursor may reuse the same request_id for a later turn and restart
append_seqno from 1. Accept seq=1 after progress as a sequence restart
when idle so tool results are not discarded as stale forever.
This commit is contained in:
lixingcheng
2026-07-27 18:09:09 +08:00
parent 9e05739969
commit 3a3be0c9e9
2 changed files with 195 additions and 3 deletions
+28 -3
View File
@@ -2,6 +2,7 @@ package forwarder
import (
"context"
"log"
"strings"
"sync"
"time"
@@ -37,8 +38,9 @@ func (tracker *appendSequenceTracker) Acquire(ctx context.Context, requestID str
if tracker == nil || strings.TrimSpace(requestID) == "" || appendSeq <= 0 {
return appendSequenceTicket{}, false, nil
}
state := tracker.state(strings.TrimSpace(requestID))
stale, err := state.acquire(ctx, appendSeq)
requestID = strings.TrimSpace(requestID)
state := tracker.state(requestID)
stale, err := state.acquire(ctx, requestID, appendSeq)
if err != nil || stale {
return appendSequenceTicket{}, stale, err
}
@@ -72,7 +74,7 @@ func (tracker *appendSequenceTracker) state(requestID string) *appendSequenceSta
return state
}
func (state *appendSequenceState) acquire(ctx context.Context, appendSeq int64) (bool, error) {
func (state *appendSequenceState) acquire(ctx context.Context, requestID string, appendSeq int64) (bool, error) {
for {
state.mu.Lock()
now := time.Now().UTC()
@@ -83,6 +85,29 @@ func (state *appendSequenceState) acquire(ctx context.Context, appendSeq int64)
state.ready = make(chan struct{})
}
state.updatedAt = now
// Cursor may reuse the same request_id for a later turn and restart
// append_seqno from 1. Accept that as a sequence restart when idle so
// tool results are not discarded as stale forever.
if appendSeq == 1 && state.next > 1 {
if state.processing {
ready := state.ready
state.mu.Unlock()
select {
case <-ctx.Done():
return false, ctx.Err()
case <-ready:
}
continue
}
prevNext := state.next
state.next = 1
state.processing = true
state.mu.Unlock()
log.Printf("forwarder reset append sequence request_id=%s previous_next=%d append_seqno=1", requestID, prevNext)
return false, nil
}
switch {
case appendSeq < state.next:
state.mu.Unlock()
@@ -0,0 +1,167 @@
package forwarder
import (
"context"
"errors"
"testing"
"time"
)
func TestAppendSequenceTrackerOrderedAcquire(t *testing.T) {
tracker := newAppendSequenceTracker()
ctx := context.Background()
ticket1, stale, err := tracker.Acquire(ctx, "req-1", 1)
if err != nil || stale {
t.Fatalf("seq 1: stale=%v err=%v", stale, err)
}
ticket1.Release()
ticket2, stale, err := tracker.Acquire(ctx, "req-1", 2)
if err != nil || stale {
t.Fatalf("seq 2: stale=%v err=%v", stale, err)
}
ticket2.Release()
}
func TestAppendSequenceTrackerRestartFromOne(t *testing.T) {
tracker := newAppendSequenceTracker()
ctx := context.Background()
for seq := int64(1); seq <= 5; seq++ {
ticket, stale, err := tracker.Acquire(ctx, "req-restart", seq)
if err != nil || stale {
t.Fatalf("seq %d: stale=%v err=%v", seq, stale, err)
}
ticket.Release()
}
// Same request_id, Cursor restarts append_seqno from 1 for a new turn.
ticket, stale, err := tracker.Acquire(ctx, "req-restart", 1)
if err != nil {
t.Fatalf("restart seq 1 err=%v", err)
}
if stale {
t.Fatal("expected sequence restart on append_seqno=1 after progress")
}
ticket.Release()
ticket2, stale, err := tracker.Acquire(ctx, "req-restart", 2)
if err != nil || stale {
t.Fatalf("post-restart seq 2: stale=%v err=%v", stale, err)
}
ticket2.Release()
}
func TestAppendSequenceTrackerMidGapStillStale(t *testing.T) {
tracker := newAppendSequenceTracker()
ctx := context.Background()
for seq := int64(1); seq <= 3; seq++ {
ticket, stale, err := tracker.Acquire(ctx, "req-gap", seq)
if err != nil || stale {
t.Fatalf("seq %d: stale=%v err=%v", seq, stale, err)
}
ticket.Release()
}
_, stale, err := tracker.Acquire(ctx, "req-gap", 2)
if err != nil {
t.Fatalf("gap retransmit err=%v", err)
}
if !stale {
t.Fatal("expected mid-sequence retransmit below next (but not 1) to stay stale")
}
}
func TestAppendSequenceTrackerWaitsForInOrder(t *testing.T) {
tracker := newAppendSequenceTracker()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
ticket1, stale, err := tracker.Acquire(ctx, "req-wait", 1)
if err != nil || stale {
t.Fatalf("seq 1: stale=%v err=%v", stale, err)
}
done := make(chan error, 1)
go func() {
ticket2, stale, err := tracker.Acquire(ctx, "req-wait", 2)
if err != nil {
done <- err
return
}
if stale {
done <- errors.New("seq 2 unexpectedly stale")
return
}
ticket2.Release()
done <- nil
}()
time.Sleep(50 * time.Millisecond)
ticket1.Release()
select {
case err := <-done:
if err != nil {
t.Fatalf("waiting seq 2 failed: %v", err)
}
case <-ctx.Done():
t.Fatal("timed out waiting for ordered seq 2")
}
}
func TestAppendSequenceTrackerRestartWaitsUntilIdle(t *testing.T) {
tracker := newAppendSequenceTracker()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
// Advance to next=3.
for seq := int64(1); seq <= 2; seq++ {
ticket, stale, err := tracker.Acquire(ctx, "req-idle", seq)
if err != nil || stale {
t.Fatalf("seq %d: stale=%v err=%v", seq, stale, err)
}
ticket.Release()
}
// Hold seq 3 so a concurrent restart must wait.
hold, stale, err := tracker.Acquire(ctx, "req-idle", 3)
if err != nil || stale {
t.Fatalf("seq 3 hold: stale=%v err=%v", stale, err)
}
done := make(chan error, 1)
go func() {
ticket, stale, err := tracker.Acquire(ctx, "req-idle", 1)
if err != nil {
done <- err
return
}
if stale {
done <- errors.New("restart seq 1 unexpectedly stale")
return
}
ticket.Release()
done <- nil
}()
time.Sleep(50 * time.Millisecond)
hold.Release()
select {
case err := <-done:
if err != nil {
t.Fatalf("restart while busy failed: %v", err)
}
case <-ctx.Done():
t.Fatal("timed out waiting for restart after idle")
}
ticket2, stale, err := tracker.Acquire(ctx, "req-idle", 2)
if err != nil || stale {
t.Fatalf("post-restart seq 2: stale=%v err=%v", stale, err)
}
ticket2.Release()
}