diff --git a/docs/changes/unreleased/964-midstream-teaching.md b/docs/changes/unreleased/964-midstream-teaching.md new file mode 100644 index 0000000000..0ff18dec0f --- /dev/null +++ b/docs/changes/unreleased/964-midstream-teaching.md @@ -0,0 +1,18 @@ +--- +kind: fixed +title: a running stream teaches the ledger mid-stream, so the next pick is not working from this morning +pr: 964 +surface: [engine] +invalidates: + - "Every `lane.Sighting` was built at settlement — `Client.noteVelocity` from the stream epilogue, and `streamWatch.sighting` for a losing arm — so while one endpoint spent minutes writing, the belief the chooser ranks on still described that machine's morning. The 14:32 retry of 2026-09-11 picked the same collapsed Morph endpoint the 14:29 call was at that moment still being throttled by. A live stream now teaches the ledger as it runs: the read loop folds a partial sighting in at most once a minute, each window covering only the span since the one before it, and the settlement sighting is rebased to cover only the tail no partial claimed — so one stream is one measurement told in chapters, never the same span twice." +--- + +This is the mid-stream teaching seam of #891 §1 that #924 measured and +deliberately did not take (it took the live-wire rescue and the +watch-every-attempt half). `streamWatch` already held the live figures — +`tokens`, `first`, `last`, `gap` — to drive the hazard controller; nothing +folded a partial reading into `lane.Ledger` until the call settled, so +`beyondThePatience` → `choice.Ignore` → `provider.ignore` read a belief that +had not been told. The windows are non-overlapping because successive readings +of one stream are not independent observations, which is why the change sits at +the `streamWatch`/`sighting` seam and not at a call site. diff --git a/internal/provider/armwatch.go b/internal/provider/armwatch.go index b6e7dfc148..0ad8a08f57 100644 --- a/internal/provider/armwatch.go +++ b/internal/provider/armwatch.go @@ -97,6 +97,15 @@ type streamWatch struct { // Recovery suspends the controller; recovered calls cannot teach lane timing. recovering bool recovered bool + // taught is the moment up to which this attempt has already folded a + // partial sighting into the ledger, zero before the first one, and + // taughtTokens the count that partial was rated on. They are what make one + // stream a chain of NON-OVERLAPPING windows rather than a series of + // cumulative re-measurements: a partial covers the span since the one + // before it, and the settlement sighting is rebased to cover only what no + // partial claimed. See part below. + taught time.Time + taughtTokens int // guard is the silence watch over this same request, and free says the // ceiling has decided to use it. See THE CEILING PICKS ONE OR THE OTHER, // NEVER NEITHER below. @@ -228,6 +237,10 @@ func (w *streamWatch) attempt(now time.Time) { w.served = "" w.acted, w.silence, w.fault = control.Act{}, 0, false w.applied, w.appliedWord = 0, "" + // THE TEACHING WINDOWS START OVER WITH THE SEND. A partial filed under the + // previous attempt's clock is a measurement of a dead stream, and a window + // left open across two sends would time one machine and charge another. + w.taught, w.taughtTokens = time.Time{}, 0 w.deadline = w.control.Deadline() if !w.deadline.IsZero() { w.armed = w.deadline.Sub(now) @@ -271,7 +284,15 @@ func (w *streamWatch) note(reading control.Reading) { // so that nothing outside this package ever takes this lock. Hidden is the // remainder because [streamWatch.tokens] is both channels together. visible, hidden := w.visible, w.tokens-w.visible + // AND THE LEDGER LEARNS THE STREAM WHILE IT IS STILL RUNNING. A live reading + // is folded in under the same non-overlapping-window law the settlement + // sighting keeps (teach below), so the belief the NEXT pick ranks on is not + // still describing this morning while this machine writes for minutes. + lesson, learn := w.part() w.mu.Unlock() + if learn { + w.noteLesson(lesson) + } // AND WHOEVER ASKED TO WATCH THIS CALL HEARS THE SAME MOMENT. It is the one // forwarding this seam needs: every streamed delta in the process reaches // this method, so there is no second decoder and no second count @@ -469,16 +490,115 @@ func (w *streamWatch) sighting(model string, tokens int) (lanes.Sighting, bool) if tokens <= 0 { tokens = w.tokens } + // REBASE THE SETTLEMENT ONTO WHAT NO PARTIAL CLAIMED. One stream is one + // chain of non-overlapping windows (teach below): the partials have already + // folded the spans up to w.taught, so the final sighting covers only the + // tail from there to the last token, rated on the tokens that tail carried. + // A stream that taught nothing yet settles exactly as it always did — the + // whole first-to-last window — and a stream that taught everything settles + // with an empty Gen, which the belief reads as "nothing new about the rate". + gen := w.last.Sub(w.first) + tokens = tailTokens(tokens, w.tokens, w.taughtTokens) + if !w.taught.IsZero() { + gen = w.last.Sub(w.taught) + } return lanes.Sighting{ ID: lanes.ID{Model: model, Lane: w.served}, TTFT: w.first.Sub(w.began), - Gen: w.last.Sub(w.first), + Gen: gen, Gap: w.gap, Tokens: tokens, At: w.last, }, true } +// partialSightingFloor is the shortest span of one stream a partial sighting +// may claim. The read loop runs per delta, but a sighting is one timed answer — +// not a running average — so a live stream teaches at most one window a minute. +// That is often enough that the NEXT pick is not working from this morning, and +// rare enough that successive windows of one stream stay the independent +// observations the ledger's arithmetic assumes they are. +const partialSightingFloor = time.Minute + +// part is the live window this stream has measured since the last teaching and +// whether it is worth folding in now. It is [sighting]'s mid-stream half: the +// same fields, the same refusals, and the same attribution law — but covering +// only the span since the previous partial (or since the first token), so the +// chain of windows of one stream never overlaps. +// +// Called with the lock held, from the read loop and from nowhere else. +func (w *streamWatch) part() (lanes.Sighting, bool) { + if w.served == "" || w.first.IsZero() || w.recovered { + return lanes.Sighting{}, false + } + // THE CADENCE GATE. One stream teaches at most one window a minute, and the + // first window may not open until the stream has been writing at least that + // long — a lane that answers in nine seconds is measured once, at settle, + // exactly as it always was. + base := w.first + if !w.taught.IsZero() { + base = w.taught + } + if w.last.Sub(base) < partialSightingFloor { + return lanes.Sighting{}, false + } + // AND ONLY A RATED WINDOW TEACHES THE RATE. The span since the last partial + // must have carried enough new tokens to rate, or the reading is the + // handshake and the warm-up, which [ratedFloor] already refuses at settle. + if w.tokens-w.taughtTokens < ratedFloor { + return lanes.Sighting{}, false + } + model := "" + if w.race != nil { + model = w.race.model + } + return lanes.Sighting{ + ID: lanes.ID{Model: model, Lane: w.served}, + TTFT: w.first.Sub(w.began), + Gen: w.last.Sub(base), + Gap: w.gap, + Tokens: w.tokens - w.taughtTokens, + At: w.last, + }, true +} + +// noteLesson files one live window into the ledger and advances the stream's +// teaching frontier past it. It runs OUTSIDE the lock, from the read loop, so +// the ledger's own gate never sits inside the per-delta critical section — the +// same reason [streamWatch.note] forwards to the controller after unlocking. +func (w *streamWatch) noteLesson(s lanes.Sighting) { + if w == nil || w.race == nil || s.Gen <= 0 { + return + } + model := laneModel(s.ID.Model) + if model == "" { + return + } + s.ID.Model = model + lanes.Default().Ledger().Note(s) + // Advance the frontier to the moment this window closed, so the next + // partial and the settlement rebase both measure from here. The lock is + // taken for the two-field write only. + w.mu.Lock() + w.taught, w.taughtTokens = s.At, w.tokens + w.mu.Unlock() +} + +// tailTokens is how many of the settlement's tokens belong to the window no +// partial claimed. The count handed in is the wire's own completion figure and +// the partials were rated on this process's count, so the tail is taken against +// the smaller of the two and never below zero — a frame that under-counts must +// not turn the last window negative. +func tailTokens(settled, streamed, taught int) int { + if settled > streamed { + settled = streamed + } + if tail := settled - taught; tail > 0 { + return tail + } + return 0 +} + // consequence is the moment this arm will be acted on and the lane the act // would go to, both empty on a request with nowhere to go. // diff --git a/internal/provider/armwatch_test.go b/internal/provider/armwatch_test.go index 0328754c6e..1e28cab945 100644 --- a/internal/provider/armwatch_test.go +++ b/internal/provider/armwatch_test.go @@ -1,9 +1,11 @@ package provider import ( + "sync" "testing" "time" + lanes "github.com/Agent-Field/aforge-v2/internal/lane" "github.com/Agent-Field/aforge-v2/internal/lane/control" ) @@ -223,3 +225,160 @@ func TestALateFirstTokenStillTeachesTheLedgerAboutItsMachine(t *testing.T) { t.Fatalf("a %s first token is not over the lag line of %s — this test is asking nothing", sighting.TTFT, LagTTFT) } } + +// ── A STREAM TEACHES THE LEDGER WHILE IT IS STILL RUNNING ────────────────── + +// partialLedger records every sighting the ledger is handed, so a test can see +// what a running stream taught and when. The belief reads it does not answer — +// this seam only ever writes. +type partialLedger struct { + mu sync.Mutex + sighted []lanes.Sighting +} + +func (l *partialLedger) Note(s lanes.Sighting) { + l.mu.Lock() + defer l.mu.Unlock() + l.sighted = append(l.sighted, s) +} +func (l *partialLedger) NoteOutcome(lanes.Outcome) {} +func (l *partialLedger) Belief(lanes.ID) (lanes.Belief, bool) { return lanes.Belief{}, false } +func (l *partialLedger) Beliefs(string) []lanes.Belief { return nil } +func (l *partialLedger) Prime(lanes.Row, float64) {} +func (l *partialLedger) noted() []lanes.Sighting { + l.mu.Lock() + defer l.mu.Unlock() + return append([]lanes.Sighting(nil), l.sighted...) +} + +// TestASlowStreamTeachesTheLedgerMidStream is the 14:29–14:32 incident of +// 2026-09-11: one Morph endpoint wrote for 86 seconds and then 363, and the +// belief the NEXT pick ranked on still described that machine's morning, so the +// retry chose the same collapsed endpoint the first call was at that moment +// still being throttled by. +// +// THE DEFECT WAS ONE DOOR OPENING ONLY AT THE END. Every lane.Sighting was +// built at settlement — [Client.noteVelocity] from the epilogue, or +// [streamWatch.sighting] for a losing arm — so a stream producing tokens +// throughout taught the ledger nothing until it closed. The chooser's refusal +// (beyondThePatience → provider.ignore) reads the belief, and the belief had +// not been told. +// +// THE FIX IS A CHAIN OF NON-OVERLAPPING WINDOWS. The read loop already holds +// the live figures; it now folds a partial sighting in at most once a minute, +// each window covering only the span since the one before it, and the +// settlement sighting is rebased to cover only what no partial claimed — so one +// stream is one measurement, told in chapters, never the same span twice. +func TestASlowStreamTeachesTheLedgerMidStream(t *testing.T) { + ledger := &partialLedger{} + previous := lanes.Default().Ledger() + lanes.Default().SetLedger(ledger) + t.Cleanup(func() { lanes.Default().SetLedger(previous) }) + + began := time.Now() + race := &hedgeRace{model: "deepseek/deepseek-v4.1-flash"} + watch := &streamWatch{race: race, began: began} + // A running arm always has its controller; install the scripted one rather + // than driving a whole race for what is a read-loop measurement. + watch.control = scriptedController{} + watch.served = "Morph" + watch.first = began.Add(4 * time.Second) // ttft 3993ms in the incident + + // THE STREAM WRITES FOR MINUTES. Eighty-six seconds of steady output — + // sixty rated tokens a minute apart, so both the cadence and the rate floor + // are crossed — with no settle anywhere in it. + at := watch.first + feed := func(tokens int, span time.Duration) { + at = at.Add(span) + watch.note(control.Reading{Visible: tokens, At: at}) + } + feed(ratedFloor, 30*time.Second) // 0:30 — under the cadence, teaches nothing + feed(ratedFloor, 31*time.Second) // 1:01 — a minute of stream: first window + feed(ratedFloor, 61*time.Second) // 2:02 — second minute: second window + + taught := ledger.noted() + if len(taught) != 2 { + t.Fatalf("a stream writing for two minutes taught the ledger %d times, want the two minute-windows", len(taught)) + } + + // EACH WINDOW IS ITS OWN SPAN, NOT A CUMULATIVE RE-MEASUREMENT. The first + // covers first-token to one minute in; the second covers the next minute. + // Overlapping windows would tell the belief one slow minute two and three + // times — which is exactly the double-count settlement is rebased to avoid. + first, second := taught[0], taught[1] + if first.ID.Lane != "Morph" { + t.Fatalf("the partial is filed against %q, want the machine that is writing", first.ID.Lane) + } + if first.Gen <= 0 || second.Gen <= 0 { + t.Fatalf("a window with no span teaches nothing: %s and %s", first.Gen, second.Gen) + } + if first.Gen > 61*time.Second { + t.Fatalf("the first window claims %s — it may cover only its own minute, not the whole stream", first.Gen) + } + if !second.At.After(first.At) { + t.Fatalf("the windows are not in stream order: %s then %s", first.At, second.At) + } + // The two windows together account for the whole generation window and no + // more: disjoint spans of one stream. + if gap := second.At.Sub(first.At); gap < time.Minute { + t.Fatalf("the second window opened %s after the first — under the one-minute cadence", gap) + } + + // AND SETTLEMENT DOES NOT TEACH THE SAME SPAN TWICE. The stream now closes, + // and its final sighting must cover only the tail no partial claimed. + watch.last = at + settled, ok := watch.sighting("deepseek/deepseek-v4.1-flash", watch.tokens) + if !ok { + t.Fatal("a stream that wrote for two minutes produced no settlement sighting") + } + // The whole stream ran ~2 minutes past its first token; the partials already + // claimed that span, so the settlement's generation window is the empty tail. + if settled.Gen > 2*time.Second { + t.Fatalf("settlement re-taught %s the partials already covered — one stream counted twice", settled.Gen) + } + if settled.Tokens > watch.tokens { + t.Fatalf("settlement claims %d tokens of a %d-token stream", settled.Tokens, watch.tokens) + } + // And it is still one honest sighting: the wait to the first token is told + // once, here, because no partial could measure it before it happened. + if settled.TTFT != 4*time.Second { + t.Fatalf("the settlement lost the first-token wait: %s", settled.TTFT) + } +} + +// TestAHealthyFastStreamTeachesNothingEarly is the case the cadence exists to +// protect: a lane that answers in nine seconds is measured once, at settle, +// exactly as it always was. The mid-stream door is for the stream that runs +// long enough to be worth interrupting the belief over — never for the ordinary +// quick answer, which would only add noise. +func TestAHealthyFastStreamTeachesNothingEarly(t *testing.T) { + ledger := &partialLedger{} + previous := lanes.Default().Ledger() + lanes.Default().SetLedger(ledger) + t.Cleanup(func() { lanes.Default().SetLedger(previous) }) + + began := time.Now() + race := &hedgeRace{model: "sim/model"} + watch := &streamWatch{race: race, began: began} + watch.control = scriptedController{} + watch.served = "quickmachine" + watch.first = began.Add(800 * time.Millisecond) + + // A normal fast stream: eighty tokens over nine seconds, well under the + // one-minute cadence. + at := watch.first + for i := 0; i < 8; i++ { + at = at.Add(time.Second) + watch.note(control.Reading{Visible: 10, At: at}) + } + + if got := ledger.noted(); len(got) != 0 { + t.Fatalf("a nine-second answer taught the ledger %d times mid-stream — the cadence gate is off", len(got)) + } + // And settlement measures it whole, unchanged. + watch.last = at + settled, ok := watch.sighting("sim/model", 80) + if !ok || settled.Gen != 8*time.Second { + t.Fatalf("the fast stream's settlement is wrong: ok=%v gen=%s", ok, settled.Gen) + } +}