diff --git a/PERF.md b/PERF.md index f1375a92e3..e78b8cbda5 100644 --- a/PERF.md +++ b/PERF.md @@ -1824,6 +1824,7 @@ and the run met none of them. | Law | The bound, and where it comes from | Where it is pinned | | --- | --- | --- | +| **A connection nobody is using is kept for as long as the ROUTER itself keeps one, and pinged so neither end races the other at the edge of that window.** The pool's idle timeout was `http.DefaultTransport`'s inherited 90 seconds, which is shorter than a person reads for: someone who took two minutes over an answer and then typed again paid DNS, TCP and TLS inside the next request's own `Do`, invisibly, because `ttft_ms` folds a handshake into the model's wait. | `idleConnTimeout`, 6m, is the FAR END'S OWN figure, measured against the live router on 2026-09-11 by holding connections idle and asking `httptrace` whether the next request rode the same socket: 30s, 2m, 5m and 6m reused; 7m, 8m and 9m opened fresh. Adopting the router's answer means neither end is guessing. `http2KeepAlive`, 2m, is what stops the far end closing first INSIDE that window, which is known to lie between six and seven minutes and not to the second — at least two pings must land inside the shortest window the measurement proves, so the period is at most half of it; 15m and 20m of idle both reused with a ping every 2m. The same ping is the health check that closes a socket whose answer never comes, which is what makes a pool held for minutes safe on a laptop that slept. `tlsHandshakeTimeout` 10s, `expectContinueTimeout` 1s and `pingAnswerTimeout` 30s are stated rather than inherited for the same reason, each with its own line in the table at the top of `transport.go`. The row that separates a cold pool from a slow machine is `conn_reused` / `dns_ms` / `connect_ms` / `tls_ms` on the call log. | `internal/provider/conntrace_test.go` (`TestTheSharedTransportStatesEveryValueItRelieson` asserts every value and the two-pings-inside-the-window inequality between the named constants) | | **A completion is asked for as a STREAM whether or not anybody is watching it.** The observer decides who is TOLD; it never decided whether the call is guarded, and until this it silently did. | No bound of its own. It is what arms the three below on a headless call. | `internal/provider/unwatched_test.go` | | **A request that has produced no token is cut.** | `firstDeltaBound`, 90s — past every healthy first token this adapter has measured, and under the streaming transport's `responseHeaderTimeout` so a stall is named rather than surfacing as a torn connection. | `internal/provider/streamguard_test.go` | | **A stream that has gone quiet is cut**, with keepalives buying bounded patience and no more. | `midStreamGapBound`, 45s, is what a lane nothing is known about gets — half the first bound, because a model that has started writing has finished deciding. A lane whose RATE this process has measured is cut at `gapFor(rate)` instead: the time that lane takes to write `streamGapLumpTokens` (3,500 — the fourteen-kilobyte server-side lump of 2026-08-24, at the estimator's four bytes to the token), clamped to `LagGap`…`midStreamGapBound`. So 250 tok/s waits 15s where a stranger waits 45. The keepalive extension window stays the FLAT bound at every rate, and `bufferedQuietBound`, 150s, still caps the total quiet. | `internal/provider/streamguard_test.go`, `internal/provider/patience_measured_test.go` | diff --git a/cmd/aforge-census/report.go b/cmd/aforge-census/report.go index 36c9ded161..f42d151561 100644 --- a/cmd/aforge-census/report.go +++ b/cmd/aforge-census/report.go @@ -331,6 +331,17 @@ type health struct { clean int millis []int64 failures map[statusClass]int + // traced is how many of these finishes said anything about their + // connection, and warm how many of those rode one the pool already had. + // The pair is kept rather than a rate because a log written by a build + // older than the tracing carries neither, and a share computed over rows + // that could not answer would read as a cold pool rather than as silence. + traced int + warm int + // opening is how long the ones that were NOT warm spent resolving, + // connecting and shaking hands — the part of `ms` that was never the + // model's. + opening []int64 } // healthSection is the clean-answer rate and median latency of each (model, @@ -376,6 +387,24 @@ func healthSection(out io.Writer, finishes []row, look settings) { } held.n++ counted++ + // WHAT THE CONNECTION COST, kept beside the answer's own time. A cold + // pool and a slow model are the same number in `ms` and in `ttft_ms`, + // and these two columns are the whole of the difference + // (internal/provider's conntrace.go). + if r.ConnReused != nil { + held.traced++ + if *r.ConnReused { + held.warm++ + } else { + // A CONNECTION OPENED FRESH COUNTS EVEN WHEN IT COST NOTHING + // MEASURABLE. On a fast link the three parts together round to + // zero milliseconds, and dropping those rows would leave the + // median describing only the expensive handshakes — the + // emptiness law is about a figure nobody measured, and this one + // was measured and came out small. + held.opening = append(held.opening, r.DNSms+r.ConnectMs+r.TLSms) + } + } class := r.statusClass() if class == classClean { held.clean++ @@ -399,20 +428,34 @@ func healthSection(out io.Writer, finishes []row, look settings) { fmt.Fprint(out, "No pair has been asked enough times to say anything about.\n\n") return } - fmt.Fprintln(out, "| model | served | n | ok % | median ms | top failures |") - fmt.Fprintln(out, "| --- | --- | ---: | ---: | ---: | --- |") + fmt.Fprintln(out, "| model | served | n | ok % | median ms | warm % | opening ms | top failures |") + fmt.Fprintln(out, "| --- | --- | ---: | ---: | ---: | ---: | ---: | --- |") for _, held := range ordered { spelledFailures := map[string]int{} for class, n := range held.failures { spelledFailures[string(class)] = n } - fmt.Fprintf(out, "| %s | %s | %d | %s | %s | %s |\n", + fmt.Fprintf(out, "| %s | %s | %d | %s | %s | %s | %s | %s |\n", cell(held.model), cell(held.machine), held.n, - trimShare(held.clean, held.n), median(held.millis), topOf(spelledFailures, 3)) + trimShare(held.clean, held.n), median(held.millis), + warmth(held.warm, held.traced), median(held.opening), + topOf(spelledFailures, 3)) } fmt.Fprintln(out) } +// warmth is how often this pair rode a connection the pool already had, and +// nothing at all for rows written by a build that did not measure it. THE +// EMPTINESS LAW MATTERS HERE MORE THAN ANYWHERE ELSE IN THIS TABLE: a zero +// would read as a pool that is always cold, which is exactly the finding this +// column exists to report, and reporting it from silence would be a lie. +func warmth(warm, traced int) string { + if traced == 0 { + return "—" + } + return trimShare(warm, traced) +} + // median is the middle of what was measured, and nothing at all when nothing // was — the emptiness law, in a table cell. func median(values []int64) string { diff --git a/docs/changes/unreleased/872-warm-wire-and-the-socket.md b/docs/changes/unreleased/872-warm-wire-and-the-socket.md new file mode 100644 index 0000000000..3674706af6 --- /dev/null +++ b/docs/changes/unreleased/872-warm-wire-and-the-socket.md @@ -0,0 +1,76 @@ +--- +kind: changed +title: A warm wire and a socket that never waits — the reader reads, and a keystroke crosses +pr: 872 +surface: [engine, remote, chat] +invalidates: + - "\"The manual describes the typing measurement, so a person can find out + about it.\" The page said what it sends and what it costs, and left out the + one thing somebody checking a bill needs: it appears in NO money figure + aforge draws. /cost, the status line, the spend place and the end-of-run + total are all sums over call-log rows, and the probe deliberately writes + none — it hangs up at the first word, so the provider's own token count + never arrives. models-and-cost.md now says so under `Spend that /cost does + not show`, and the /cost section's claim that it counts EVERY request is + corrected. #903 is whether that absence should end." + - "\"Typing pre-warms the lane.\" It did not — anywhere, on any road, since + the completer wrapper landed — and it was broken in TWO places, both mended + here. The surface reaches for the capability through an optional interface + that *remote.Agent did not satisfy, so on every launch that is not + --no-host, --debug, first-run setup or a hostless --once the keystroke never + left the window; it crosses now, as MethodTyping. And internal/session's + probeClientLanes asserts an optional prober on a completer that + installSessionClient always wraps in sessionCompleter, which forwarded + CompleteWithMessages and FallbackModels and not ProbeLanes — so even in + process, with the keystroke arriving, no probe had ever been bought. The + test that covered it built an Agent literal and never met the wrapper." + - "\"An ordered call runs on the reader goroutine, and that is what makes it + ordered.\" Two separate things wearing one name. Ordered calls now run on + the connection's orderedLane — one goroutine, arrival order — and NOTHING + runs on the reader. staysOnReader is deleted; callClass.road() has two + values and neither is the reader. Anything that said Compact or Submit + \"holds the reader\" is stale: they hold the lane, which is the thing that + owes the order." + - "\"A pause between turns is free.\" It cost a handshake. The shared + transport inherited http.DefaultTransport's 90-second IdleConnTimeout, + which is shorter than a person reads for, so the next turn paid DNS, TCP + and TLS inside its own httpClient.Do. It was invisible because ttft_ms + folds the handshake into the model's wait. The four transport values are + now stated with a reason each, and the idle window is the router's OWN, + measured against it from the Spark on 2026-09-11: a connection held idle + for six minutes still rode the same socket and one held for seven did not. + An HTTP/2 PING every two minutes keeps it open through fifteen and twenty + minutes of silence, so the pool also carries a keep-alive rather than + racing the far end at the edge of its window." + - "\"A slow ttft_ms means a slow model.\" Not on its own. The finish row now + carries conn_reused, and dns_ms / connect_ms / tls_ms where the connection + was opened fresh, so a cold pool and a slow machine are separable. `make + census` grows a `warm %` and an `opening ms` column on the lane-health + table. Rows written before this carry none of the four, and the census + draws nothing rather than a zero." +--- + +Two laws land with them, and both are the same shape one layer apart: an +optional door asserted on a value somebody wraps or replaces looks exactly like +a capability that was never meant to be there, and nothing tells you when it +stops being reachable. internal/remote's +TestEverySurfaceDoorTheEngineHasCrossesTheWire holds the surface-to-wire seam +(with a ratcheted ledger of the twenty-three doors that do not cross yet), and +internal/session's TestTheCompleterWrapperForwardsEveryDoorTheAdapterOffers +holds the agent-to-adapter one. + +The three findings are one sentence each and they are all about waiting for +something that was not the model. A send held the whole socket shut while the +engine assembled a prompt; a keystroke that was supposed to open the connection +early never reached the engine at all; and the pool threw the connection away +while the person was still reading. `internal/remote/orderedlane.go` is the one +mechanism the first needed, and +`TestEverySurfaceDoorTheEngineHasCrossesTheWire` is the ledger that stops the +second class coming back — twenty-three other doors the local engine answers +and the wire does not are written down in it, and the count may only fall. +Each entry says what a person loses, and the eight that do not merely go quiet +but print a REASON THAT IS NOT THE REASON — `/remember` saying memory is +switched off, `/land` saying nothing is waiting, `/autonomy` saying the +conversation has no project — carry the sentence verbatim, checked against the +surface's own source so it cannot drift into fiction. #901 is the parent issue +for landing them, grouped into families rather than doors. diff --git a/internal/calllog/calllog.go b/internal/calllog/calllog.go index 22644d088d..96861a14d2 100644 --- a/internal/calllog/calllog.go +++ b/internal/calllog/calllog.go @@ -267,6 +267,37 @@ type Record struct { // any row, because nothing wrote it. RetryAfterS float64 `json:"retry_after,omitempty"` + // ── what the connection itself cost + // + // A COLD POOL AND A SLOW MODEL ARE INDISTINGUISHABLE IN `ttft_ms`, and + // separating them is what these four are for. The wait before the first + // token includes everything the transport had to do before a byte of the + // request left this machine: resolve the name, open the socket, finish the + // handshake. On a warm pool all three are zero and the whole of `ttft_ms` + // belongs to the endpoint; on a cold one a hundred to four hundred + // milliseconds of it is ours, and a census that could not see the + // difference read every think-pause as the model getting slower. + // + // They are written from `net/http/httptrace`, which costs nothing: the + // hooks fire on the transport's own goroutine and record instants. + + // ConnReused is whether this attempt rode a connection the pool already + // had. It is the single most useful field of the four, and it is spelled + // as a bool rather than inferred from a zero handshake because "the pool + // was warm" and "nothing was measured" are different facts. + // + // FALSE IS WRITTEN OUT. The emptiness law leaves an unknown blank, and a + // cold connection is not unknown — it is the finding. So the row carries + // `conn_reused:false` where a fresh connection was opened, and carries the + // field not at all where no trace was taken (a probe, a document post). + ConnReused *bool `json:"conn_reused,omitempty"` + // DNSms, ConnectMs and TLSms are the three parts of opening one, in + // milliseconds, and every one of them is absent on a reused connection + // because none of them happened. + DNSms int64 `json:"dns_ms,omitempty"` + ConnectMs int64 `json:"connect_ms,omitempty"` + TLSms int64 `json:"tls_ms,omitempty"` + // ── why it waited, and what was done about it // // The four fields above say what was BELIEVED before the request went out. diff --git a/internal/manual/chat/models-and-cost.md b/internal/manual/chat/models-and-cost.md index b2cda6f91b..531c12f5d3 100644 --- a/internal/manual/chat/models-and-cost.md +++ b/internal/manual/chat/models-and-cost.md @@ -1776,13 +1776,17 @@ function, and if you ever see them differ, that is a bug worth reporting. the provider — every step of a turn is its own request — so this figure is normally larger than the number of times you have spoken. -It counts **every** request, not only the ones in your turns: naming the session, a judge -deciding where something should be routed, looking at a picture, every request a task's -own agent made on its own lane, and every request a harness run made while it walked its -program. That is deliberate, because the `spend` line above it is the +It counts every request that is written down, not only the ones in your turns: naming the +session, a judge deciding where something should be routed, looking at a picture, every +request a task's own agent made on its own lane, and every request a harness run made +while it walked its program. That is deliberate, because the `spend` line above it is the sum over exactly those requests — a smaller count beside it would be a bill divided by the wrong number. +**One request is not written down and so is not in either figure**: the one-token +measurement sent while you are typing. Your provider bills it and aforge does not count +it — there is a section on that below, `Spend that /cost does not show`. + `empty reflex answers` appears only when that failure happened. Those requests remain in the token, call and spend totals because the provider billed them; the separate count says that the money bought no memory decision. aforge retries one such answer with more room, @@ -2740,7 +2744,7 @@ Turn it off if you are paying for every token and never mind waiting. With it of are one row because they are one promise: aforge may spend a little extra to keep an answer moving. -## Why aforge sends something when you start typing — the one-token measurement +## Does aforge send anything while I am typing — the one-token measurement it sends before you press enter While you are typing, and before you press enter, aforge sends **one token** to each of the two lanes your next message would most likely go to, and times how long the first @@ -2754,12 +2758,47 @@ token out, twice. **How often.** At most one pair every **twenty seconds** per model, however fast you type — so a long message buys one, not one per keystroke. None at all when the **speed -guard** is off, when `routing` is `off`, when the lane row says `openrouter`, or when -the connection pool is already being rate-limited. +guard** is off, when `routing` is `off`, when the lane row says `openrouter`, when the +pool is already backing off a rate limit, when aforge is still recovering a dropped +connection, or when **nobody is waiting on that model** — a task working on its own and +an errand buy none, because the measurement exists to shorten a wait somebody is sitting +through. **Nothing ever waits for it.** It is sent and forgotten; a message you send a moment later does not wait on it, and a probe that fails teaches nothing and changes nothing. +**It is not in `/cost`.** It is a real request to a real provider and your provider bills +you for it, and aforge's own figures do not include it. The next section says why, what +it adds up to, and where to see it. + +## Spend that /cost does not show — why the typing measurement is missing from the figures + +There is exactly one request aforge makes that its own money figures do not count: the +**one-token measurement** it sends while you are typing, to warm the connection and time +the machine your next message is heading for. Your provider bills you for it. `/cost`, +the status line, the spend place (`alt+3`) and the total at the end of `aforge do` all +leave it out, and so do the call-log rows and `aforge-census`. + +**Why it is missing.** Those figures are all counts of the **call log**, and the +measurement deliberately writes no row there — it skips the shaping, the retries and the +record on purpose, so that what it times is one clean request to one named machine and +not a retry of one. And it hangs up **at the first word**, so the token count a provider +sends at the end of a reply never arrives: there is no measured figure to add up, only +the fixed shape of the request. + +**What it comes to.** Ten tokens in and one token out, twice — about **two hundredths of +a cent** a pair, at most one pair every twenty seconds per model, and only while somebody +is actually sitting there waiting. An unbroken hour of typing is a few cents. It cannot +run in the background, and a task working on its own buys none. + +**How to see it anyway.** Every measurement it buys is appended to +`~/.aforge/v3/lanes.log`, one line of JSON each, with the ones bought this way marked as +probes. That file is the record of what was sent. + +**How to make it zero.** Settings → Providers → **speed guard**, off. The same row governs +asking a second machine when an answer is slow to start, so turning it off stops both. +`routing off` and a lane row set to `openrouter` also stop it. + ## The lane row in settings — auto, pinned, pinned but borrowable, openrouter Settings → Providers has two rows under **routing**: diff --git a/internal/manual/chat/running-on-another-machine.md b/internal/manual/chat/running-on-another-machine.md index 4cfe8c0a09..b25eac0838 100644 --- a/internal/manual/chat/running-on-another-machine.md +++ b/internal/manual/chat/running-on-another-machine.md @@ -611,12 +611,20 @@ all work over `--host`. What to know is **whose machine they are on**: - Pausing or stopping one writes to the far machine's store, and a write that store refuses is shown as its own refusal rather than redrawn as done. -**Home and the standing place both work, and both are about the far machine.** Home lists -that machine's projects with each one's `◦` item band under it and the `p`/`s` keys live on -them; the standing place lists both what stands on this conversation and what stands anywhere -else on that machine. The `◦ 2 standing orders` count at the foot of the task column counts -the far machine's items for the workspace this window is on, because over `--host` that path -is the far machine's own. +**Home works and is about the far machine; the standing place works in half.** Home lists +that machine's projects with each one's `◦` item band under it, and the `p`/`s` keys live on +them. + +The standing place over a connection draws ONE of its shelves: **what stands anywhere else +on that machine**. The shelf for *what stands on this conversation*, and the one for +*exceptions*, are both empty — not because nothing stands, but because the reading behind +them is a door the connection does not carry. For the same reason the `◦ 2 standing orders` +count at the foot of the task column is **absent** over `--host`: there is no line rather +than a wrong number. + +To see and change what stands on the conversation itself, open it on the machine it runs on. +This is a gap and not a design: the doors exist on that machine and nothing carries them +across yet. Two readings are absent over a connection, and each says nothing rather than guessing: diff --git a/internal/manual/chat_test.go b/internal/manual/chat_test.go index 19eb142819..9350ae475d 100644 --- a/internal/manual/chat_test.go +++ b/internal/manual/chat_test.go @@ -174,6 +174,14 @@ func TestTheChatManualAnswersTheQuestionsPeopleAsk(t *testing.T) { // meet it: reading the figure, and finding it beside the call log. {"what does the total at the end of aforge do include", "models-and-cost"}, {"why is the printed cost different from the call log", "models-and-cost"}, + // THE ONE SEND THAT IS NOT IN THE MONEY FIGURES, asked the three ways + // somebody meets it: noticing the sending, doubting the total, and + // looking for the switch. A capability that spends without appearing in + // /cost has to be findable from a person's own suspicion, or the only + // place it exists is the code. + {"does aforge send anything while I am typing", "models-and-cost"}, + {"my provider bill is higher than what /cost says", "models-and-cost"}, + {"how do I stop aforge sending requests I did not ask for", "models-and-cost"}, // A refused headless run names the project it worked in, asked in the two // ways somebody meets an empty file list after spending time and money. {"the run finished and my directory is empty where did the work go", "running-from-the-terminal"}, diff --git a/internal/provider/calllog.go b/internal/provider/calllog.go index 68ef2b393f..d4ae3cab48 100644 --- a/internal/provider/calllog.go +++ b/internal/provider/calllog.go @@ -137,6 +137,16 @@ type callTrace struct { // enters on its own context and gets its own — and the transport under it is // sequential. open *openAttempt + // conn is what the transport had to do to get the CURRENT attempt onto the + // wire — whether the pool already had a connection, and where the time went + // when it did not. It is replaced per attempt for [callTrace.attemptID]'s + // reason: a row about attempt two may not carry attempt one's handshake. + // + // THE POINTER NEEDS NO LOCK FOR `open`'s REASON ABOVE — a trace belongs to + // one call and the transport under it is sequential. The mutex inside + // [connFacts] is a different question and a real one: those fields are + // written by `httptrace` hooks on the transport's own dial goroutines. + conn *connFacts // body is the request as it was last encoded, kept ONLY when somebody // asked for it: the old bodies pin, which puts it on the line of the // model-call log, or the debug record, which is where bodies live now @@ -476,6 +486,13 @@ func (c *Client) record(facts recordFacts) { if facts.id != "" { record.ID = facts.id } + // WHAT THE CONNECTION ITSELF COST, on the row that ends the attempt it was + // measured on. It is stamped after the id is settled because the facts are + // filed under the attempt they belong to and a closing row is written about + // an attempt that has already been overtaken (see [recordFacts.id]). + if facts.phase != calllog.PhaseStart && facts.knobs.trace != nil { + facts.knobs.trace.stampConn(&record) + } if facts.attempt == 0 && facts.knobs.trace != nil { // A row about the whole call says how many times it went out; a row // about one attempt already said which attempt it was. diff --git a/internal/provider/conntrace.go b/internal/provider/conntrace.go new file mode 100644 index 0000000000..223ca6e214 --- /dev/null +++ b/internal/provider/conntrace.go @@ -0,0 +1,167 @@ +package provider + +import ( + "crypto/tls" + "net/http/httptrace" + "sync" + "sync/atomic" + "time" + + "github.com/Agent-Field/aforge-v2/internal/calllog" +) + +// ── WHAT THE CONNECTION ITSELF COST ───────────────────────────────────────── +// +// A COLD POOL AND A SLOW MODEL LOOK THE SAME IN `ttft_ms`, and this file is +// what tells them apart. The wait a person sees before the first word includes +// everything the transport had to do before a byte of the request left this +// machine: resolve openrouter.ai, open the socket, finish the handshake. On a +// warm pool that is nothing at all. On a cold one it is a hundred to four +// hundred milliseconds, every attempt, and for ten days the log recorded it as +// the endpoint being slow. +// +// IT IS FREE. `net/http/httptrace` is a set of function pointers the transport +// already carries a nil check for; the hooks below record instants and do no +// work. The one cost is a mutex, and it is there because the hooks do NOT all +// run on the caller's goroutine — a dial and its handshake run on the +// transport's own, and Happy Eyeballs may run two at once — while the row that +// reads them is often written by the stream loop on a third. + +// connFacts is one attempt's connection, filed under the attempt it belongs to. +// +// It is per-ATTEMPT rather than per-call for [callTrace.attemptID]'s reason: a +// call that was paced and sent again opened its second connection under +// different conditions, and a row about attempt two carrying attempt one's +// handshake would be a measurement of the wrong socket. +type connFacts struct { + mu sync.Mutex + // id is the attempt these facts are about. A closing row is written about + // an attempt that has already been overtaken, so the row and the facts are + // paired by this rather than by whichever attempt is current. + id string + // seen is whether the transport ever reached a connection at all. A request + // refused before the dial — a context already expired, a limiter that would + // not let go — has nothing to say and says nothing. + seen bool + reused bool + + dnsBegan, connectBegan, tlsBegan time.Time + dns, connect, handshake time.Duration +} + +// newConnFacts opens the measurement for one attempt. +func newConnFacts(attemptID string) *connFacts { return &connFacts{id: attemptID} } + +// hooks is the trace this attempt rides. sent is the caller's own flag for +// whether the request bytes ever left — it is a separate fact from the +// connection and is kept where it was, on the one hook that answers it. +func (f *connFacts) hooks(sent *atomic.Bool) *httptrace.ClientTrace { + return &httptrace.ClientTrace{ + WroteRequest: func(httptrace.WroteRequestInfo) { sent.Store(true) }, + DNSStart: func(httptrace.DNSStartInfo) { + f.mu.Lock() + defer f.mu.Unlock() + f.dnsBegan = time.Now() + }, + DNSDone: func(httptrace.DNSDoneInfo) { + f.mu.Lock() + defer f.mu.Unlock() + if !f.dnsBegan.IsZero() { + f.dns = time.Since(f.dnsBegan) + } + }, + ConnectStart: func(string, string) { + f.mu.Lock() + defer f.mu.Unlock() + // THE FIRST DIAL IS THE ONE THAT IS TIMED. Happy Eyeballs opens a + // second one to the other address family a fraction of a second + // later, and restarting the clock on it would report the loser's + // head start as the whole cost of connecting. + if f.connectBegan.IsZero() { + f.connectBegan = time.Now() + } + }, + ConnectDone: func(_, _ string, err error) { + f.mu.Lock() + defer f.mu.Unlock() + // A dial that failed is not a cost anybody paid for an answer: the + // winning dial writes the figure, and a row where every dial failed + // carries no connect time because no connection was made. + if err != nil || f.connectBegan.IsZero() || f.connect != 0 { + return + } + f.connect = time.Since(f.connectBegan) + }, + TLSHandshakeStart: func() { + f.mu.Lock() + defer f.mu.Unlock() + f.tlsBegan = time.Now() + }, + TLSHandshakeDone: func(_ tls.ConnectionState, err error) { + f.mu.Lock() + defer f.mu.Unlock() + if err != nil || f.tlsBegan.IsZero() { + return + } + f.handshake = time.Since(f.tlsBegan) + }, + GotConn: func(info httptrace.GotConnInfo) { + f.mu.Lock() + defer f.mu.Unlock() + f.seen, f.reused = true, info.Reused + }, + } +} + +// stamp writes what was measured onto a row. +// +// FALSE IS WRITTEN OUT AND ZERO IS NOT. A connection that was opened fresh is +// the finding rather than the absence of one, so the row says `conn_reused` is +// false; the three parts of opening it are left off a reused connection, +// because on a reused connection none of them happened. A row whose attempt +// never reached a connection carries none of the four. +func (f *connFacts) stamp(record *calllog.Record) { + if f == nil || record == nil { + return + } + f.mu.Lock() + defer f.mu.Unlock() + if !f.seen { + return + } + reused := f.reused + record.ConnReused = &reused + if reused { + return + } + record.DNSms = f.dns.Milliseconds() + record.ConnectMs = f.connect.Milliseconds() + record.TLSms = f.handshake.Milliseconds() +} + +// stampConn puts the CURRENT attempt's connection on a row, and only when the +// row is about that attempt. The pairing is the id, for the reason the trace +// keeps one at all ([recordFacts.id]). +func (t *callTrace) stampConn(record *calllog.Record) { + if t == nil || t.conn == nil || record == nil { + return + } + if t.conn.id != record.ID { + return + } + t.conn.stamp(record) +} + +// openConn starts the measurement for the attempt that is about to go out, and +// hands back the trace it rides. A trace-less caller — a probe, which is +// deliberately the leanest request this package makes — gets a trace that still +// reports whether the bytes were written, and nothing is filed. +func (t *callTrace) openConn(sent *atomic.Bool) *httptrace.ClientTrace { + if t == nil { + return &httptrace.ClientTrace{ + WroteRequest: func(httptrace.WroteRequestInfo) { sent.Store(true) }, + } + } + t.conn = newConnFacts(t.attemptID) + return t.conn.hooks(sent) +} diff --git a/internal/provider/conntrace_test.go b/internal/provider/conntrace_test.go new file mode 100644 index 0000000000..126faa6dc3 --- /dev/null +++ b/internal/provider/conntrace_test.go @@ -0,0 +1,163 @@ +package provider + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// TestTheRowSaysWhetherThePoolWasWarm is the fix for the blind spot this wave +// was written about: `ttft_ms` folds the handshake into the model's wait, so a +// person who paused to read and then typed again looked like a provider that +// had slowed down. The first call opens a connection and says so; the second +// rides the one the pool kept and says that. +func TestTheRowSaysWhetherThePoolWasWarm(t *testing.T) { + read := loggingTo(t) + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { + writer.Header().Set("Content-Type", "application/json") + _, _ = writer.Write([]byte(`{"model":"sim/model","choices":[{"index":0,"finish_reason":"stop","message":{"role":"assistant","content":"ok"}}],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`)) + })) + t.Cleanup(server.Close) + + // A POOL OF ITS OWN, so the two calls below are the only two things that + // have ever used it and the first one is certainly cold. + transport := &http.Transport{IdleConnTimeout: idleConnTimeout} + t.Cleanup(transport.CloseIdleConnections) + client, err := NewClient(Config{ + APIKey: "test-key", + BaseURL: server.URL, + Model: "sim/model", + HTTPClient: &http.Client{Transport: transport, Timeout: 10 * time.Second}, + }) + if err != nil { + t.Fatal(err) + } + forgetLanes(t) + + for range 2 { + if _, err := client.CompleteWithMessages(context.Background(), userMessages("hello")); err != nil { + t.Fatal(err) + } + } + + done := ended(read()) + if len(done) != 2 { + t.Fatalf("two calls should leave two finished rows; got %d", len(done)) + } + cold, warm := done[0], done[1] + if cold.ConnReused == nil { + t.Fatal("the first row says nothing about its connection") + } + if *cold.ConnReused { + t.Error("the first call of a fresh pool cannot have ridden a connection the pool already had") + } + if cold.ConnectMs < 0 { + t.Errorf("connect_ms = %d", cold.ConnectMs) + } + if warm.ConnReused == nil || !*warm.ConnReused { + t.Fatalf("the second call should have ridden the pool: %+v", warm) + } + // A REUSED CONNECTION PAID FOR NONE OF THE THREE, so the row carries none + // of them rather than three honest zeroes (the emptiness law). + if warm.DNSms != 0 || warm.ConnectMs != 0 || warm.TLSms != 0 { + t.Errorf("a warm row should carry no handshake: %+v", warm) + } +} + +// TestAThinkPauseKeepsThePool is the whole point of [idleConnTimeout], stated +// as the thing a person does: read an answer, think, type again. The pause here +// is short because the law under test is that the pool's own timeout is not +// what closes the connection — the transport's figure is minutes, and a pool +// that dropped the socket in between would fail this at any pause at all. +func TestAThinkPauseKeepsThePool(t *testing.T) { + read := loggingTo(t) + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { + writer.Header().Set("Content-Type", "application/json") + _, _ = writer.Write([]byte(`{"model":"sim/model","choices":[{"index":0,"finish_reason":"stop","message":{"role":"assistant","content":"ok"}}],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`)) + })) + t.Cleanup(server.Close) + + transport := SharedTransport().Clone() + t.Cleanup(transport.CloseIdleConnections) + client, err := NewClient(Config{ + APIKey: "test-key", + BaseURL: server.URL, + Model: "sim/model", + HTTPClient: &http.Client{Transport: transport, Timeout: 10 * time.Second}, + }) + if err != nil { + t.Fatal(err) + } + forgetLanes(t) + + if _, err := client.CompleteWithMessages(context.Background(), userMessages("hello")); err != nil { + t.Fatal(err) + } + // A pause, short enough to be free and long enough that a transport with + // no idle timeout of its own would already have reaped the connection. + time.Sleep(150 * time.Millisecond) + if _, err := client.CompleteWithMessages(context.Background(), userMessages("again")); err != nil { + t.Fatal(err) + } + + done := ended(read()) + if len(done) != 2 { + t.Fatalf("two calls should leave two finished rows; got %d", len(done)) + } + if done[1].ConnReused == nil || !*done[1].ConnReused { + t.Fatalf("the turn after a pause paid a fresh handshake: %+v", done[1]) + } +} + +// TestTheSharedTransportStatesEveryValueItRelieson refuses an inherited +// default. Each of these is a number this build has a reason for +// (transport.go's table), and a zero here is the standard library's answer to a +// question nobody asked. +func TestTheSharedTransportStatesEveryValueItRelieson(t *testing.T) { + shared := SharedTransport() + if shared.IdleConnTimeout != idleConnTimeout { + t.Errorf("IdleConnTimeout = %v, want %v", shared.IdleConnTimeout, idleConnTimeout) + } + if !shared.ForceAttemptHTTP2 { + t.Error("ForceAttemptHTTP2 is off: one multiplexed connection is the whole reason a warm pool is cheap here") + } + if shared.TLSHandshakeTimeout != tlsHandshakeTimeout { + t.Errorf("TLSHandshakeTimeout = %v, want %v", shared.TLSHandshakeTimeout, tlsHandshakeTimeout) + } + if shared.ExpectContinueTimeout != expectContinueTimeout { + t.Errorf("ExpectContinueTimeout = %v, want %v", shared.ExpectContinueTimeout, expectContinueTimeout) + } + if shared.HTTP2 == nil { + t.Fatal("no keep-alive: the far end closes an idle connection between six and seven minutes, and a pool that only waited out its own would hand a request a socket the router had already dropped") + } + if shared.HTTP2.SendPingTimeout != http2KeepAlive { + t.Errorf("SendPingTimeout = %v, want %v", shared.HTTP2.SendPingTimeout, http2KeepAlive) + } + if shared.HTTP2.PingTimeout != pingAnswerTimeout { + t.Errorf("PingTimeout = %v, want %v", shared.HTTP2.PingTimeout, pingAnswerTimeout) + } + // AT LEAST TWO PINGS MUST LAND INSIDE THE WINDOW THE MEASUREMENT PROVES, + // which is what makes one lost ping survivable and is the whole derivation + // of the period (transport.go's table). The pool's own timeout is the far + // end's measured figure, so the same inequality states both. + if http2KeepAlive*2 > idleConnTimeout { + t.Errorf("a keep-alive of %v inside a window of %v leaves fewer than two pings", http2KeepAlive, idleConnTimeout) + } + if shared.MaxIdleConnsPerHost != limiterCeiling { + t.Errorf("MaxIdleConnsPerHost = %d, want the limiter's ceiling %d", shared.MaxIdleConnsPerHost, limiterCeiling) + } + // AND THE STREAMING POOL IS THE SAME POOL PLUS ONE DEADLINE. A second set + // of values would drift from this one the first time somebody changed one. + streaming := streamTransport() + if streaming.IdleConnTimeout != shared.IdleConnTimeout || streaming.ForceAttemptHTTP2 != shared.ForceAttemptHTTP2 { + t.Error("the streaming pool no longer inherits the shared pool's settings") + } + if streaming.ResponseHeaderTimeout != responseHeaderTimeout { + t.Errorf("ResponseHeaderTimeout = %v, want %v", streaming.ResponseHeaderTimeout, responseHeaderTimeout) + } + if shared.ResponseHeaderTimeout != 0 { + t.Error("a non-streamed completion must not inherit a header deadline: its headers do not arrive until the answer is generated") + } +} diff --git a/internal/provider/dispatch.go b/internal/provider/dispatch.go index bceaecc939..cbe375acd3 100644 --- a/internal/provider/dispatch.go +++ b/internal/provider/dispatch.go @@ -465,9 +465,13 @@ func (c *Client) send(ctx context.Context, request *ai.Request, knobs callKnobs, // watchdog, which fires it on stall or on Close. attemptCtx, cancelAttempt := attemptContext(ctx, stream) var sent atomic.Bool - attemptCtx = httptrace.WithClientTrace(attemptCtx, &httptrace.ClientTrace{ - WroteRequest: func(httptrace.WroteRequestInfo) { sent.Store(true) }, - }) + // AND THE CONNECTION IS MEASURED AS WELL AS THE ANSWER. The hooks cost + // nothing and they are the only thing that can separate a cold pool from + // a slow model in `ttft_ms` — a person who read an answer for two minutes + // and then typed again used to pay a whole handshake inside this Do, and + // the log filed every millisecond of it as the endpoint being slow + // (conntrace.go). + attemptCtx = httptrace.WithClientTrace(attemptCtx, knobs.trace.openConn(&sent)) // AND THE BYTES ARE WRITTEN HERE, immediately before the send, which is // the same timing [Client.providerPreferences] exists for: the machine diff --git a/internal/provider/transport.go b/internal/provider/transport.go index 7aace05ac0..968219c478 100644 --- a/internal/provider/transport.go +++ b/internal/provider/transport.go @@ -35,6 +35,89 @@ const ( streamIdleTimeout = 2 * time.Minute ) +// ── THE POOL'S OWN TABLE: ONE VALUE, ONE REASON, ONE MEASUREMENT ──────────── +// +// EVERY NUMBER HERE IS THIS BUILD'S ANSWER AND NOT THE STANDARD LIBRARY'S. +// `http.DefaultTransport` is a sensible default for a program nobody measured; +// it was measured here, and one of its answers was costing a person a TLS +// handshake every time they stopped to read. A value inherited silently is a +// decision nobody took, so each of the five below is set on purpose and says +// why, and [TestTheSharedTransportStatesEveryValueItRelieson] fails the build +// if one of them goes back to being inherited. +// +// ── THE MEASUREMENT, 2026-09-11, against the live router from the Spark ───── +// +// A connection was opened, held idle for a fixed interval, and used again with +// `httptrace` watching `GotConn`. Each hold had a pool of its own, so the only +// thing under test was the far end. +// +// hold keep-alive rode the same socket +// 30s none yes +// 2m none yes +// 5m none yes +// 6m none yes +// 7m none NO — DNS 3ms, connect 3ms, TLS 9ms +// 8m none NO +// 9m none NO +// 15m PING every 2m yes +// 20m PING every 2m yes +// +// TWO FACTS COME OUT OF IT AND BOTH ARE USED BELOW. The router's own idle +// window is between six and seven minutes — so ninety seconds was throwing +// away a connection the far end was still holding, four times over. And an +// HTTP/2 PING keeps it open for as long as we care to: fifteen and twenty +// minutes of silence both rode the same socket. +// +// - idleConnTimeout is how long THIS END keeps a connection nobody is using, +// and it is the far end's own measured figure. Adopting the router's answer +// rather than inventing one means neither end is guessing: inside the window +// the router itself considers a connection live, a pause costs nothing; +// past it, a pause costs one handshake, which is what a connection nobody +// believes in any more is worth. The default was 90 seconds, which is +// shorter than a person reads for — so someone who took two minutes over an +// answer and then typed again paid DNS, TCP and TLS inside the next +// request's own `httpClient.Do`, invisibly, because `ttft_ms` cannot tell a +// handshake from a slow model (conntrace.go is what tells them apart now). +// - http2KeepAlive is what stops the far end closing first INSIDE that +// window. The window is known to be between six and seven minutes and not +// to the second, so a pool that simply waited out its own six would +// sometimes hand a request a socket the router had already dropped — a +// failure, which is the expensive way to learn the same thing. At least two +// pings must therefore land inside the shortest window the measurement +// proves, which puts the period at no more than three minutes; two is the +// figure the fifteen- and twenty-minute readings confirm. It is also the +// health check that closes a socket whose ping goes unanswered, which is +// what makes a pool held for minutes safe on a laptop that slept. +// - forceAttemptHTTP2 is why a warm pool is cheap at all against the router: +// one connection carries every request this process makes, so the pool that +// matters is one socket rather than [limiterCeiling] of them — and it is +// what makes the ping above possible, since HTTP/1.1 has no such frame. It +// is on in the standard library's default and off in a transport somebody +// built by hand, which is exactly the accident this states out loud. +// - tlsHandshakeTimeout bounds the part of a cold connection this file exists +// to stop paying for. Ten seconds is the standard library's figure and it +// is far too long for a handshake that is going to work — the measured ones +// above are nine to eleven MILLISECONDS. A lane that has not finished one +// in this long is a path to somewhere else, and the ladder above has +// somewhere else to go (dispatch.go). +// - expectContinueTimeout is stated to say that it never applies: nothing in +// this package sends `Expect: 100-continue`, so the wait it bounds is a +// wait no request of ours takes. It is kept at a second rather than +// disabled because a zero here means "wait forever" for the one request +// that might one day carry the header through a proxy at AFORGE_BASE_URL. +const ( + idleConnTimeout = 6 * time.Minute + http2KeepAlive = 2 * time.Minute + tlsHandshakeTimeout = 10 * time.Second + expectContinueTimeout = time.Second + // pingAnswerTimeout is how long a health check waits for its answer before + // the connection is closed. It is generous beside the handshake it protects + // — a machine that cannot answer a PING in this long has nothing to do with + // a slow model — and it exists so that the number is ours rather than the + // standard library's fifteen seconds arrived at by default. + pingAnswerTimeout = 30 * time.Second +) + var ( transportOnce sync.Once shared *http.Transport @@ -59,6 +142,18 @@ func buildTransports() { if shared.MaxIdleConns < 2*limiterCeiling { shared.MaxIdleConns = 2 * limiterCeiling } + // And the five the table above states. They are assigned here rather than + // left to the clone so that a base transport somebody replaces — the + // process-global this function is careful about above — cannot quietly + // change any of them. + shared.IdleConnTimeout = idleConnTimeout + shared.ForceAttemptHTTP2 = true + shared.TLSHandshakeTimeout = tlsHandshakeTimeout + shared.ExpectContinueTimeout = expectContinueTimeout + shared.HTTP2 = &http.HTTP2Config{ + SendPingTimeout: http2KeepAlive, + PingTimeout: pingAnswerTimeout, + } streaming = shared.Clone() streaming.ResponseHeaderTimeout = responseHeaderTimeout diff --git a/internal/remote/callclass.go b/internal/remote/callclass.go index a5cb08281e..cf1afd15d7 100644 --- a/internal/remote/callclass.go +++ b/internal/remote/callclass.go @@ -2,32 +2,30 @@ package remote // ── THE ROAD IS CLASSED, NOT FREE-FOR-ALL ─────────────────────────────────── // -// callClass is which side of the road a method travels on, and it is THE ONE -// PREDICATE that decides it: the engine's reader asks it whether [server.invoke] -// may leave the reader goroutine, and nothing else in this package asks that -// question anywhere else. A list sprinkled through the switch in -// [server.invoke] would be a second answer waiting to disagree with this one, -// and one method in the wrong place there is a keystroke queued behind a getter -// — which is what this file was written for. +// callClass is WHAT A METHOD OWES THE ONES AROUND IT, and it is THE ONE +// PREDICATE that decides where the method runs: [server.serve] asks it and +// nothing else in this package asks that question anywhere else. A list +// sprinkled through the switch in [server.invoke] would be a second answer +// waiting to disagree with this one, and one method in the wrong place there is +// a keystroke queued behind a getter — which is what this file was written for. // -// ORDERED stays on the reader, in the order the surface sent it: anything that -// opens a stream or changes the conversation's shape. A stream's first event -// cannot overtake the result that names it ([server.release] is one slot for -// that reason), and a surface that sets a model and then submits means those -// two things in that order (server.go's file header). Compact is the one that -// pays for this — it does the work itself and holds the reader for as long as -// the summarizer takes. +// ORDERED is anything that opens a stream or changes the conversation's shape. +// What it owes is an ORDER, twice over: a stream's first event may not overtake +// the result that names it ([server.release] is one slot for that reason), and +// a surface that sets a model and then submits means those two things in that +// order (server.go's file header). // // A GETTER is a read. The surface asks half of them from its update loop, so -// they cannot hold the reader (a listing that blocked a keystroke was the -// measured defect) and they cannot wait longer than [callDeadline] (a terminal -// that has stopped repainting). +// they may not queue behind anything (a listing that blocked a keystroke was +// the measured defect) and they may not wait longer than [callDeadline] (a +// terminal that has stopped repainting). // // AN ACT is a person's small write — answering a card, taking the keyboard, -// interrupting. It runs off the reader for the getter's reason, said the other -// way round: a key is not queued behind a listing. It waits the same -// [callDeadline] a getter does, and the note below the classes says why the -// longer window it was given first had to come back out. +// interrupting, saying they have started typing. It owes nothing to anything +// for the getter's reason said the other way round: a key is not queued behind +// a listing. It waits the same [callDeadline] a getter does, and the note below +// the classes says why the longer window it was given first had to come back +// out. // // AND LETTING EITHER OF THEM OVERTAKE AN ORDERED CALL COSTS NOTHING, which is // the whole licence for this split and is worth stating plainly: EVERY CALL ON @@ -36,13 +34,37 @@ package remote // answer in hand before the second frame is written, so two calls whose order // matters were already ordered by the caller and cannot be reordered here; two // made from different goroutines were never ordered by anything, before this -// change or after it. What the reader still owes is the order WITHIN one call — -// the result frame before the first event of the stream it names — and that is -// exactly what stays on it. +// change or after it. // // A METHOD THIS FUNCTION HAS NOT HEARD OF IS ORDERED. A door that lands next -// year stays on the reader until someone decides it is a getter or a small -// act, which is the same allow-list shape [watcherMay] uses. +// year keeps its place in the queue until someone decides it is a getter or a +// small act, which is the same allow-list shape [watcherMay] uses. +// +// ── AND ORDERED NEVER MEANT "ON THE READER" ───────────────────────────────── +// +// That was the confusion this file carried until 2026-09-11, and it is why a +// keystroke could sit behind a send. Running the ordered bodies on the +// goroutine that reads the socket gave the order for free and took the READING +// away with it: [MethodSubmit] alone rebinds the client, re-reads the person's +// standing orders, reassembles the system prompt, journals the message and +// starts the naming errand before it returns, and for all of that the engine +// was not reading its own pipe. A person who pressed Enter and then Esc was +// queued behind their own send. [MethodCompact] was the extreme case — it does +// the summarizer's work itself — and it was described here as the price of +// being ordered, which it never was. +// +// SO THE READER READS, AND NOTHING ELSE. Every class runs its body off it; what +// a class chooses is WHICH goroutine ([callClass.road]). Ordered calls go to +// the connection's one [orderedLane] — one goroutine, arrival order, which is +// exactly the guarantee the reader was giving and is now given by the thing +// that actually needs it. Getters and small acts owe each other nothing, so +// each takes a goroutine of its own. +// +// AND THE ORDERED LANE IS THE ONLY OWNER OF [server.pending]. Every method that +// names a stream is ordered, so the one slot is written and read by one +// goroutine and needs no lock. A method that opened a stream from any other +// class would be racing it, which is what [TestEveryStreamOpenerIsOrdered] +// refuses. type callClass int @@ -52,6 +74,31 @@ const ( classAct ) +// road is WHERE a class of call runs, and this is the one place a class is +// turned into a goroutine. [server.serve] switches on it and nothing else +// decides it, so a method's road is a property of its class rather than a +// branch at the site that happens to read the socket. +// +// NEITHER ROAD IS THE READER. That is the law of this file said as a type: the +// goroutine that reads the socket has no third option to be given. +type road int + +const ( + // inOrder is the connection's one ordered lane: off the reader, and in the + // order the frames arrived (orderedlane.go). + inOrder road = iota + // onItsOwn is a goroutine per call. A getter and a small act owe nothing to + // each other, so neither owes a queue. + onItsOwn +) + +func (c callClass) road() road { + if c == classOrdered { + return inOrder + } + return onItsOwn +} + // ── AND AN ACT DOES NOT GET A LONGER WINDOW THAN A GETTER ─────────────────── // // It is the obvious second half of this fix and it was written, measured and @@ -103,13 +150,16 @@ func classify(method string) callClass { MethodConsent, MethodConsentRemember, MethodStandingResolve, MethodHarness, MethodConnect, MethodConnectKey, MethodNoteConnected, - MethodTake, MethodAnswerLaneOffer, MethodInterrupt: + MethodTake, MethodAnswerLaneOffer, MethodInterrupt, + MethodTyping: return classAct default: return classOrdered } } -// staysOnReader is [classify] as the engine's reader asks it: stream-opening -// and shape-changing calls stay here; getters and small acts run off it. -func staysOnReader(method string) bool { return classify(method) == classOrdered } +// opensAStream reports whether calls of this class may name a stream, which is +// the same question as "may they write [server.pending]". Exactly one class +// may, one goroutine runs that class, and [server.dispatch] starts the pump for +// exactly that one — which is the whole of why the slot needs no lock. +func (c callClass) opensAStream() bool { return c == classOrdered } diff --git a/internal/remote/client.go b/internal/remote/client.go index 65c1bc40bb..43ff1ab924 100644 --- a/internal/remote/client.go +++ b/internal/remote/client.go @@ -146,6 +146,13 @@ type Client struct { // collide even though both are uint64. seq atomic.Uint64 + // typing is the hush on the one frame nobody waits for (typing.go), and + // hinting is whether one of those frames is on the wire right now — the + // whole of what keeps [Client.notify] off the caller's goroutine without + // letting goroutines pile up behind a wedged link. + typing typingBeat + hinting atomic.Bool + // newsHeard is set the first time a "phase" or "lane" frame arrives, and it // is half of how [Client.NewsSilent] answers: an engine that has sent one // has the news whether or not its welcome said so ([Welcome.News]). @@ -959,6 +966,61 @@ func (c *Client) call(ctx context.Context, method string, args any) (json.RawMes return c.callWithin(ctx, method, args, callDeadline) } +// notify is a frame with NOBODY WAITING FOR IT: it is written and forgotten, +// and the engine's answer is dropped by [Client.deliver] the way it drops the +// answer to any call that has given up. +// +// IT IS THE SHAPE FOR A HINT AND NOT FOR AN INSTRUCTION, and the difference is +// worth stating because everything else on this wire is a round trip on +// purpose. A call that is answered is how one end learns that the other end +// agreed; a hint has nothing to agree to. [MethodTyping] is the whole of it +// today — somebody is writing, which is true whether the engine acts on it or +// not — and a door that needs to know it was heard must use [Client.call]. +// +// AND IT NEVER BLOCKS THE CALLER, WHICH IS A LAW AND NOT AN ASPIRATION. +// [Client.write] holds the writer's lock across a `Write` and a flush, and over +// `--host` that pipe is an ssh process's stdin — so writing here would put a +// blocking socket write on whatever goroutine sent the hint. The one caller is +// a keystroke, on internal/tui3's update loop, which is the goroutine that +// draws: a wedged-but-not-dead link would stop the terminal echoing what the +// person types. That is the defect family this wave exists to remove, and it is +// not being reintroduced one layer down (the owner's ruling: the only wait a +// person feels is the main model generating). +// +// SO THE WRITE HAPPENS ON A GOROUTINE, AND AT MOST ONE IS EVER IN FLIGHT. A +// hint is free to lose — the person will send another with their next character +// — so a second one arriving while the first is still on the wire is DROPPED +// rather than queued. That is what keeps this unbounded in neither goroutines +// nor memory while [orderedLane], whose frames are not free to lose, is +// unbounded in the one and bounded by the far end's patience in the other. +// [TestEveryNotifiedMethodOwesNobodyAnOrder] is the law that keeps a call which +// DOES need an answer off this road. +func (c *Client) notify(method string, args any) { + var payload json.RawMessage + if args != nil { + encoded, err := json.Marshal(args) + if err != nil { + return + } + payload = encoded + } + c.mu.Lock() + unusable := c.dead != nil || c.reconnecting + c.mu.Unlock() + if unusable { + return + } + // A hint already on the wire is a hint this one has nothing to add to. + if !c.hinting.CompareAndSwap(false, true) { + return + } + frame := Frame{Kind: "call", ID: c.seq.Add(1), Method: method, Payload: payload} + go func() { + defer c.hinting.Store(false) + _ = c.write(frame) + }() +} + func (c *Client) callWithin(ctx context.Context, method string, args any, deadline time.Duration) (json.RawMessage, error) { payload, _, err := c.callAnswered(ctx, method, args, deadline) return payload, err diff --git a/internal/remote/notifylaw_test.go b/internal/remote/notifylaw_test.go new file mode 100644 index 0000000000..a8d5229adc --- /dev/null +++ b/internal/remote/notifylaw_test.go @@ -0,0 +1,143 @@ +package remote + +import ( + "go/ast" + "path/filepath" + "sync" + "testing" + "time" +) + +// TestEveryNotifiedMethodOwesNobodyAnOrder is the law [orderedLane]'s doc +// comment depends on, and it is written down because that comment is currently +// the only thing holding it up. +// +// THE UNCAPPED QUEUE IS SAFE FOR EXACTLY ONE REASON: every frame on it is a call +// the far end is waiting for synchronously under [callDeadline], so its depth is +// the number of calls one surface has outstanding. [Client.notify] is the one +// mechanism in this package that puts a frame on the wire with NOBODY waiting +// for it — and `c.notify(MethodSubmit, …)` is one line, compiles, and turns that +// bound from "a surface's outstanding calls" into unbounded. +// +// So a method sent through notify must be one that owes nobody an order: its +// class must run on its own goroutine, never on the ordered lane. That also +// keeps the two mechanisms' promises apart — a hint may be dropped +// ([Client.notify] drops one while another is on the wire), and a frame on the +// ordered lane may not be. +func TestEveryNotifiedMethodOwesNobodyAnOrder(t *testing.T) { + root := repoRoot(t) + found := 0 + walkGo(t, filepath.Join(root, "internal", "remote"), func(path string, file *ast.File) { + ast.Inspect(file, func(node ast.Node) bool { + call, ok := node.(*ast.CallExpr) + if !ok || len(call.Args) == 0 { + return true + } + selected, ok := call.Fun.(*ast.SelectorExpr) + if !ok || selected.Sel.Name != "notify" { + return true + } + found++ + name, ok := call.Args[0].(*ast.Ident) + if !ok { + t.Errorf("%s: a notify whose method is not a named constant cannot be checked; name it", + filepath.Base(path)) + return true + } + method, known := wireMethod(name.Name) + if !known { + t.Errorf("%s: notify sends %s, which is not a method constant this law can read", + filepath.Base(path), name.Name) + return true + } + if classify(method).road() != onItsOwn { + t.Errorf("%s: notify sends %s, which owes an order — a frame nobody waits for may not ride the ordered lane", + filepath.Base(path), name.Name) + } + return true + }) + }) + if found == 0 { + t.Error("this law found no notify call at all: either the mechanism was renamed or the walk stopped reaching it") + } +} + +// wireMethod resolves a method constant's NAME to its value, so the law above +// reads the same table [classify] does rather than a copy of it. Every constant +// in wire.go's block is a plain string and the compiler has them all here; this +// is the one place they are looked up by their Go identifier. +func wireMethod(identifier string) (string, bool) { + method, ok := wireMethodsByName[identifier] + return method, ok +} + +// wireMethodsByName is every method a notify may name. It is deliberately SHORT +// rather than the whole wire: a method added here is somebody deciding that a +// frame of that kind may be sent with nobody waiting for it, which is exactly +// the decision this law exists to make visible. +var wireMethodsByName = map[string]string{ + "MethodTyping": MethodTyping, + "MethodPing": MethodPing, + "MethodSubmit": MethodSubmit, + "MethodSteer": MethodSteer, + "MethodCompact": MethodCompact, +} + +// blockingPipe is a link that has accepted the connection and stopped reading: +// not dead, not reconnecting, just wedged. It is the case [Client.notify]'s law +// is about and the one `c.dead != nil || c.reconnecting` cannot see. +type blockingPipe struct { + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (p *blockingPipe) Write(b []byte) (int, error) { + p.once.Do(func() { close(p.entered) }) + <-p.release + return len(b), nil +} + +func (p *blockingPipe) Read([]byte) (int, error) { <-p.release; return 0, nil } +func (p *blockingPipe) Close() error { return nil } + +// TestAHintDoesNotWaitForAWedgedLink is [Client.notify]'s stated law, driven. +// +// THE CALLER IS THE GOROUTINE THAT DRAWS. internal/tui3's app.key calls +// [Agent.Typing] from Update, on every character, so a hint that waited on +// [Client.write] would stop the terminal echoing what the person is typing for +// as long as the far end was wedged — the ≤50 ms ruling, broken by the very +// mechanism meant to make a turn faster. +func TestAHintDoesNotWaitForAWedgedLink(t *testing.T) { + pipe := &blockingPipe{entered: make(chan struct{}), release: make(chan struct{})} + client := &Client{conn: pipe} + t.Cleanup(func() { close(pipe.release) }) + + // The first hint takes the wedged write and stays there. + client.notify(MethodTyping, nil) + select { + case <-pipe.entered: + case <-time.After(2 * time.Second): + t.Fatal("the hint never reached the pipe") + } + + // AND EVERY KEYSTROKE AFTER IT RETURNS AT ONCE. They are dropped rather + // than queued, because a hint is free to lose and the person is about to + // send another one anyway. + done := make(chan time.Duration, 1) + go func() { + began := time.Now() + for range 200 { + client.notify(MethodTyping, nil) + } + done <- time.Since(began) + }() + select { + case took := <-done: + if took > 50*time.Millisecond { + t.Fatalf("two hundred keystrokes behind a wedged link took %v", took) + } + case <-time.After(2 * time.Second): + t.Fatal("a keystroke is waiting on the socket: the draw goroutine is blocked") + } +} diff --git a/internal/remote/orderedlane.go b/internal/remote/orderedlane.go new file mode 100644 index 0000000000..5591d00a13 --- /dev/null +++ b/internal/remote/orderedlane.go @@ -0,0 +1,111 @@ +package remote + +import "sync" + +// ── ONE MECHANISM: WORK THAT KEEPS ITS ORDER AND LEAVES THE READER ────────── +// +// orderedLane runs handed-in work ON ONE GOROUTINE, IN THE ORDER IT ARRIVED, +// and never makes the hand-in wait. Those three properties together are the +// whole of what the engine's reader goroutine was providing by running calls +// itself — and the fourth thing it was providing, accidentally, was a socket +// that stopped being read for as long as the work took. +// +// THE HAND-IN NEVER BLOCKS AND THE QUEUE IS NOT CAPPED, which is a deliberate +// choice and not an oversight. Every frame on this queue is a call the far end +// is waiting for synchronously, under its own [callDeadline]; a surface that is +// waiting for an answer does not send another. So the depth of this queue is +// the number of calls ONE surface has outstanding — a handful — and it is +// bounded by that surface's own patience rather than by a number invented here. +// A cap would have to be such a number, and the first thing it would do on the +// day it was reached is stop reading the socket, which is the defect this type +// exists to remove. +// +// THAT INVARIANT IS A LAW AND NOT A HOPE. [Client.notify] is the one mechanism +// on this wire that writes a frame nobody waits for, and one line of it naming +// an ordered method would turn the bound above into nothing; +// [TestEveryNotifiedMethodOwesNobodyAnOrder] refuses exactly that. Two smaller +// things the depth argument rests on, said out loud because a reader will ask: +// a surface whose call reaches [callDeadline] is free to send another, so the +// true bound is its outstanding calls plus one per deadline per goroutine — which +// is what "bounded by that surface's own patience" means; and after a long +// [MethodCompact] the lane replays ordered calls nobody is waiting for any more, +// which is what the socket buffer did before this type existed and is therefore +// the same behaviour rather than a new one. +// +// It is a type rather than a channel and a goroutine written inline because +// "run these in the order they came, off the goroutine that received them" is a +// shape, and a shape spelled out twice is two shapes that will disagree. +type orderedLane struct { + mu sync.Mutex + wake chan struct{} + queue []Frame + closed bool + done chan struct{} +} + +func newOrderedLane() *orderedLane { + return &orderedLane{wake: make(chan struct{}, 1), done: make(chan struct{})} +} + +// hand puts one frame at the back of the queue and returns at once. A frame +// handed to a lane that has already been closed is dropped: the connection it +// belonged to is over, and running it would write into a session that has been +// told nobody is there. +func (l *orderedLane) hand(frame Frame) { + l.mu.Lock() + if l.closed { + l.mu.Unlock() + return + } + l.queue = append(l.queue, frame) + l.mu.Unlock() + select { + case l.wake <- struct{}{}: + default: + } +} + +// run drains the lane until it is closed and everything handed in before the +// close has been run. It is the goroutine the work happens on, and there is +// exactly one of it per lane — which is what makes "in order" and "one at a +// time" the same statement. +func (l *orderedLane) run(work func(Frame)) { + defer close(l.done) + for { + l.mu.Lock() + if len(l.queue) == 0 { + if l.closed { + l.mu.Unlock() + return + } + l.mu.Unlock() + <-l.wake + continue + } + next := l.queue[0] + l.queue = l.queue[1:] + l.mu.Unlock() + work(next) + } +} + +// close says there is no more work. Everything already handed in still runs, in +// order, because a call whose frame crossed the wire is a call the far end is +// waiting for an answer to. +func (l *orderedLane) close() { + l.mu.Lock() + if l.closed { + l.mu.Unlock() + return + } + l.closed = true + l.mu.Unlock() + select { + case l.wake <- struct{}{}: + default: + } +} + +// wait blocks until the lane has run everything it was given. A lane that was +// never started returns at once, because nothing was ever taken from it. +func (l *orderedLane) wait() { <-l.done } diff --git a/internal/remote/ratchet_test.go b/internal/remote/ratchet_test.go new file mode 100644 index 0000000000..4be934548e --- /dev/null +++ b/internal/remote/ratchet_test.go @@ -0,0 +1,29 @@ +package remote + +import "strconv" + +// ratchetComplaint is what a ledger that may only shrink says when it has not. +// +// IT IS internal/ci's SENTENCE, SAID ONCE FOR THE SECOND LEDGER. That package's +// `knownRedEntries` and this package's [surfaceDoorLedger] are the same shape — +// a list of things that are wrong, a number beside it, and a rule that the +// number only comes down — and a rule spelled twice is two rules that will one +// day disagree about which direction is the red one. The second branch is the +// half that is easy to forget and is the reason both exist: A RATCHET LEFT SLACK +// WOULD LET THE NEXT CHANGE PUT AN ENTRY BACK WITH THE GATE GREEN THROUGHOUT. +// +// It lives in a test file because both callers are laws, and a law's vocabulary +// is not something the shipped binary should carry (the size budget, SIZE-BUDGET). +func ratchetComplaint(ledger string, entries, ratchet int) string { + switch { + case entries > ratchet: + return ledger + " holds " + strconv.Itoa(entries) + " entries and the ratchet is at " + + strconv.Itoa(ratchet) + ". There is no road to a new entry: the ledger only shrinks. " + + "Land the door across the wire and delete its line instead." + case entries < ratchet: + return ledger + " is down to " + strconv.Itoa(entries) + " entries and the ratchet still says " + + strconv.Itoa(ratchet) + " — lower it to " + strconv.Itoa(entries) + " in this change. A ratchet " + + "left slack would let the next change put an entry back with the gate green throughout." + } + return "" +} diff --git a/internal/remote/road_test.go b/internal/remote/road_test.go index 36f89e122f..305c93a501 100644 --- a/internal/remote/road_test.go +++ b/internal/remote/road_test.go @@ -140,11 +140,19 @@ func TestADeadlineOnALiveConnectionDoesNotSayTheConnectionIsGone(t *testing.T) { } func TestTheCallClassKeepsAKeystrokeOffTheReader(t *testing.T) { - if staysOnReader(MethodQuestionResolve) || staysOnReader(MethodTranscript) || staysOnReader(MethodPing) { - t.Fatal("a getter or a small act stayed on the reader") + for _, method := range []string{MethodQuestionResolve, MethodTranscript, MethodPing, MethodTyping} { + if classify(method).road() != onItsOwn { + t.Fatalf("%s is a getter or a small act and owes nobody a queue", method) + } } - if !staysOnReader(MethodSubmit) || !staysOnReader(MethodCompact) || !staysOnReader(MethodSetModel) { - t.Fatal("a stream or a shape change left the reader") + // AND NOTHING AT ALL RUNS ON THE READER NOW. Everything the engine does + // between Submit and the first request leaving used to run there and + // everything behind it on this socket waited (callclass.go); what an + // ordered call owes is an order, and the lane is what gives it. + for _, method := range []string{MethodSubmit, MethodCompact, MethodSetModel} { + if classify(method).road() != inOrder { + t.Fatalf("%s owes an order and is not on the ordered lane", method) + } } // AND AN ACT IS NOT A CLASS OF ITS OWN ON THE CLOCK, which is a law with a // measurement behind it (callclass.go): a longer window for a keystroke buys diff --git a/internal/remote/server.go b/internal/remote/server.go index 7cc9b6aa64..a39679490d 100644 --- a/internal/remote/server.go +++ b/internal/remote/server.go @@ -4,28 +4,31 @@ package remote // answers frames about it. It is what `aforge engine` runs after it has changed // into the workspace and assembled an agent exactly the way `aforge chat` does. // -// The shape is one reader and one writer, and everything else follows from it. -// Calls that open a stream or change the conversation's shape still arrive and -// are answered in the order the surface made them, on this goroutine -// ([classify] is the one predicate). A surface that sets a model and then -// submits a message means those two things in that order and nothing here may -// reorder them. The one exception is a turn's events, which arrive on a +// The shape is one reader, one writer, and one ORDERED LANE, and everything +// else follows from it. Calls that open a stream or change the conversation's +// shape still arrive and are answered in the order the surface made them — on +// the lane, which is one goroutine that is not this one ([classify] is the one +// predicate, orderedlane.go is the mechanism). A surface that sets a model and +// then submits a message means those two things in that order and nothing here +// may reorder them. The one exception is a turn's events, which arrive on a // channel the session owns and are pumped by a goroutine of their own — that // is the whole reason Submit answers with a stream id instead of a transcript. // -// GETTERS AND SMALL ACTS RUN OFF THIS GOROUTINE. A listing that held the -// reader used to queue a person's keystroke behind it until [callDeadline] -// fired and the surface declared the connection gone, while the engine went -// on to apply the key. That is the defect: a person's act is not queued -// behind an unrelated getter. +// NOTHING AT ALL RUNS ON THE READER. A listing that held it used to queue a +// person's keystroke behind it until [callDeadline] fired and the surface +// declared the connection gone, while the engine went on to apply the key; a +// SEND held it for the whole of the engine's preamble and did the same thing to +// every frame behind it. Both are the same defect — a person's act queued +// behind work that had nothing to do with it — and the reader now does the one +// job its name claims. // -// ORDER IS WORTH MORE THAN OVERLAP for the calls that stay here, and the one -// that pays for it is Compact: it is the only method that does the work itself -// rather than starting it, so a compaction pass holds the reader for as long -// as the summarizer takes and the ordered calls behind it wait. Handing Compact -// a goroutine would buy a live status line during a compaction and cost the -// guarantee that a /model followed by a message is a message on the new model -// — a bad trade, and a bug nobody would reproduce twice. +// ORDER IS WORTH MORE THAN OVERLAP for the calls on the lane, and the one that +// pays for it is Compact: it is the only method that does the work itself +// rather than starting it, so a compaction pass holds the lane for as long as +// the summarizer takes and the ordered calls behind it wait. Handing Compact a +// goroutine would buy nothing the reader does not already give — the status +// line stays live because the socket stays read — and would cost the guarantee +// that a /model followed by a message is a message on the new model. // // ── VERSION 2: THE CONVERSATION IS NOT THE CONNECTION ──────────────────────── // @@ -59,6 +62,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/Agent-Field/aforge-v2/internal/buildinfo" @@ -1520,10 +1524,21 @@ type server struct { joined string // pending is the stream a call has just opened and dispatch has not yet let - // speak. It is one slot rather than a queue because only an ordered call - // opens a stream, and those stay on the reader, one at a time ([classify]). + // speak. It is one slot rather than a queue, and it needs no lock, because + // only an ORDERED call opens a stream and every ordered call on this + // connection runs on the one lane below, one at a time + // ([callClass.opensAStream]). pending *pending - // side is every getter or small act this connection has handed off the + // ordered is that lane: every call that owes an order, in the order the + // surface sent it, on one goroutine that is not the reader. + // + // IT IS WHY A KEYSTROKE NO LONGER WAITS FOR A SEND. Everything the engine + // does between Submit and the first request leaving — the client rebind, + // the standing orders, the system prompt, the journal write, the naming + // errand — used to run on the reader, and every frame behind it on this + // socket waited for all of it (callclass.go's header). + ordered *orderedLane + // side is every getter and small act this connection has handed off the // reader. The serve loop waits for it before leave, so a listing still // running does not write into a session that has already been closed. side sync.WaitGroup @@ -1537,8 +1552,13 @@ type server struct { // conversation itself is over. Neither set, and a reader that stops, is a // TORN PIPE, whose meaning depends on whether anything is holding the // conversation (see the bottom of [server.serve]). - detached bool - goodbye bool + // They are atomics because they are decided on the ordered lane and read by + // the reader loop, which is the one place in this file where two goroutines + // look at the same fact. The reader's read is a HINT that lets it stop a + // frame early; the reading that matters is [server.leave]'s, and that one + // happens after the lane has drained. + detached atomic.Bool + goodbye atomic.Bool } func (s *server) serve(in io.Reader) (err error) { @@ -1573,25 +1593,51 @@ func (s *server) serve(in io.Reader) (err error) { return nil } + // THE ORDERED LANE IS OPENED WITH THE CONNECTION AND DRAINED BEFORE IT + // LEAVES. The drain is deferred here rather than at the top so that it runs + // BEFORE the leave the first defer registered — deferred calls run in + // reverse — which is what makes [server.leave] read a `detached` or a + // `goodbye` that was decided on the lane a moment ago. + s.ordered = newOrderedLane() + go s.ordered.run(s.dispatch) + defer func() { + s.ordered.close() + s.ordered.wait() + }() + for scan.Scan() { frame, err := readCall(scan.Bytes()) if err != nil { s.fatal(err.Error()) return err } - if staysOnReader(frame.Method) { - s.dispatch(frame) - } else { + // WHERE A CALL RUNS IS ITS CLASS'S PROPERTY AND NOT A DECISION TAKEN + // HERE (callclass.go's [callClass.road]). This switch is the whole of + // the reader's part in it, so a method added next year travels the road + // its class names without anybody touching the loop that reads the + // socket. + switch classify(frame.Method).road() { + case inOrder: + s.ordered.hand(frame) + default: s.side.Add(1) go func(call Frame) { defer s.side.Done() s.dispatch(call) }(frame) } - if s.detached { + if s.detached.Load() { // THE SURFACE SAID IT WAS GOING, and said so before it went, which // is the fact version 1 could not express. The turn keeps running; // this connection is simply over. + // + // THIS IS A SHORTCUT AND NO LONGER THE ROAD. [MethodDetach] is + // ordered, so the flag is stored on the lane long after the reader + // has looped back into Scan, and this test will essentially never + // be the thing that ends the loop. What ends it is the pipe: the + // surface closes it the moment the call returns, Scan reaches EOF, + // the lane drains, and [server.leave] reads a settled flag. Kept + // because it costs one atomic load and is honest when it does win. return nil } if s.hungUp() { @@ -1640,7 +1686,7 @@ func (s *server) leave() { return } sess.detach(s) - if s.goodbye { + if s.goodbye.Load() { _ = sess.closeLeaving() return } @@ -1805,10 +1851,10 @@ func (s *server) dispatch(call Frame) { } _ = s.send(result) // And only now does a turn opened by that call begin to speak. - // GETTERS AND SMALL ACTS NEVER OPEN A STREAM, so they must not touch - // pending: two of them running at once would race the one slot, and one - // of them could steal a stream an ordered call had just named. - if staysOnReader(call.Method) { + // ONLY A TURN EVER OPENS A STREAM, so only a turn releases one: two calls + // of any other class running at once would race the one slot, and one of + // them could steal a stream a turn had just named ([callClass.opensAStream]). + if classify(call.Method).opensAStream() { s.release() } } @@ -2168,6 +2214,28 @@ func (s *server) invoke(call Frame) (out json.RawMessage, err error) { // clock, so the engine contributes no timestamp and no machine-clock skew. return nil, nil + case MethodTyping: + // SOMEBODY IS WRITING, WHICH IS THE CHEAPEST THING THIS ENGINE IS EVER + // TOLD. It buys a measurement of the machines the next turn will use and + // a connection already open when that turn goes out; it returns before + // anything is sent, spends nothing when the speed guard is off, and has + // its own budget inside (internal/session's lanenews.go). + // + // THE RESULT FRAME STILL GOES BACK AND IS DROPPED AT THE OTHER END. + // [server.dispatch] answers every call it runs, and this road does not + // know that nobody is waiting — [Client.deliver] drops a result whose + // id has no waiter, which is the same path a call that reached its + // deadline takes. It is an empty frame once per [typingHush] and buying + // a second frame KIND to avoid it would be a wire change for nothing. + // + // AN ENGINE WHOSE AGENT CANNOT HEAR IT DOES NOTHING, rather than + // refusing: a door with nothing behind it is a capability that is ABSENT + // (the design law), and a task node's agent is deliberately silent here. + if door, ok := agent.(typingDoor); ok { + door.Typing() + } + return nil, nil + case MethodSubmit: args, err := arg[SubmitArgs](call) if err != nil { @@ -2287,7 +2355,7 @@ func (s *server) invoke(call Frame) (out json.RawMessage, err error) { // here rather than at the hang-up that follows, which is the same work // done a moment earlier and with somebody still listening for the // failure. - s.goodbye = true + s.goodbye.Store(true) return nil, sess.closeLeaving() case MethodDetach: @@ -2295,7 +2363,7 @@ func (s *server) invoke(call Frame) (out json.RawMessage, err error) { // Nothing is interrupted and nothing is closed; the reader loop sees // this flag on its next pass and ends the connection, leaving the turn // to finish (wire.go's MethodDetach states the whole case). - s.detached = true + s.detached.Store(true) return nil, nil case MethodHeldQuestions: @@ -2659,8 +2727,8 @@ type pending struct { events <-chan session.Event } -// release starts whatever the call just opened. It runs on the reader -// goroutine, after the result is on the wire, and it starts the pump even when +// release starts whatever the call just opened. It runs on the ordered lane, +// after the result is on the wire, and it starts the pump even when // that write failed: the channel has a session writing into it, and a channel // nobody drains is a turn that never finishes. // diff --git a/internal/remote/surfacedoors_law_test.go b/internal/remote/surfacedoors_law_test.go new file mode 100644 index 0000000000..5df378e4c4 --- /dev/null +++ b/internal/remote/surfacedoors_law_test.go @@ -0,0 +1,563 @@ +package remote + +import ( + "go/ast" + "go/parser" + "go/token" + "io/fs" + "os" + "path/filepath" + "runtime" + "sort" + "strconv" + "strings" + "testing" +) + +// ── THE LAW: A DOOR THE ENGINE HAS EITHER CROSSES THE WIRE OR IS WRITTEN DOWN ─ +// +// internal/tui3 reaches for almost everything through an OPTIONAL interface — +// `a.agent.(typingAgent)`, `a.agent.(memoryAgent)` — and that is the right +// shape: a capability with nothing behind it must be ABSENT rather than present +// and failing (the design law in CLAUDE.md). What it cannot do is tell you when +// the thing behind it went missing, because a failed assertion looks exactly +// like a capability that was never meant to be there. +// +// AND THE DEFAULT ROAD IS THE REMOTE ONE. A bare `aforge` holds a +// `*remote.Agent` talking to a detached engine — `v3TakeHostRoad` in +// cmd/aforge/chatv3_local.go is true for every launch but `--no-host`, +// `--debug`, first-run setup and a hostless `--once`. So a door that +// `*session.Agent` has and `*remote.Agent` does not is a feature that works in +// every test and in nobody's terminal. [MethodTyping] was one: the probe that +// exists so a think-pause does not cost a TLS handshake was wired to a keystroke +// the default road could not deliver, and the whole mechanism was dead where +// people actually run. +// +// SO THE GAPS ARE A LEDGER AND THE LEDGER ONLY SHRINKS. Every door the local +// engine has and the wire does not must be named below with the reason it has +// not crossed yet; a door that DOES cross must be deleted from the list; and +// the count is ratcheted, so the next gap cannot be added silently. This is the +// known-red discipline (internal/ci), applied to the one asymmetry that has +// twice shipped a feature nobody could reach. + +// absentDoor is one door the wire does not carry, and WHAT A PERSON MEETS +// BECAUSE OF IT — which is the half the first version of this ledger left out +// and the half that decides which of these gets fixed first. +// +// A capability that cannot work is ABSENT, NOT BROKEN (CLAUDE.md's design law), +// and about half of these obey it: the page, the section or the count is simply +// not drawn, and a person meets nothing. The other half do the inverse. The +// surface's `a.hosted()` guard is `a.host != ""` and is set only on the +// `--host ` road, so on a BARE `aforge` — the default launch, and the one +// that holds a `*remote.Agent` — every one of those guards is skipped, the local +// branch runs, and the person is told a reason that is not the reason. `/remember` +// says memory is switched off. `/land` says nothing is waiting. `/autonomy` says +// the conversation has no project. All three are false, and all three send a +// person to a settings page that will not help them. +type absentDoor struct { + // says is the sentence a person meets, quoted as the code spells it, and + // EMPTY WHERE THE DOOR IS SILENTLY ABSENT. Empty is the law being obeyed; + // a sentence is the law being inverted, and those are the ones to land + // first. + says string + // loses is what does not work. + loses string +} + +// doorsThatHaveNotCrossed is every optional door internal/tui3 asserts, that +// `*session.Agent` answers and `*remote.Agent` does not, with what a person on +// the default road meets instead. NEVER ADD A LINE. Landing a door removes one, +// and [surfaceDoorLedger] comes down in the same commit. +// +// THEY ARE A FINDING AND NOT A DESIGN. Nothing on this list was decided to be +// local-only; each is a door somebody wired to `*session.Agent` and nobody +// carried across. Writing them down is what turns a silence into a number +// somebody can watch fall, and #901 is the public account of it — the same +// split, grouped into the five families they would actually be built in, with +// the ruling it asks for first: a family at a time, not a door at a time. +var doorsThatHaveNotCrossed = map[string]absentDoor{ + // ── SAYS SOMETHING THAT IS NOT TRUE. Land these first. + "memoryAgent": { + says: "memory is off for this session · turn it on under /settings", + loses: "/remember, /forget, /memories, /memory and the memory place — memory is not off, it is unreachable", + }, + "folderLander": { + says: "nothing is waiting · what this conversation writes in the folder it is standing in is already there", + loses: "/land; and the `changes for … · /land` row above the box goes quiet too", + }, + "autonomyAgent": { + says: "this conversation has no project to keep question rules in", + loses: "reading the autonomy a project has set — the write half (SetAutonomy) crosses, so the sheet can be changed and not shown", + }, + "subharnessAgent": { + says: "no subharnesses here yet — a subharness is a saved program for work that comes round again.", + loses: "the subharness list, its intake form and running one", + }, + "orchAgent": { + says: "adaptive runs unavailable — this session has no orchestrator", + loses: "the orchestration room: its snapshot, a node's journal, steering it, answering it", + }, + "harnessRunner": { + says: "harnesses are unavailable here", + loses: "running a harness the picker offered", + }, + "standingHereAgent": { + says: "this window cannot change it", + loses: "the standing page's four doors — what holds here, an exception, standing one down, pausing one; the page draws only the elsewhere shelf", + }, + "taskRoomAgent": { + says: "this session has no task rooms", + loses: "watching one task and reading its journal; the room's steer door crosses and its two reading doors do not, so a room falls back to a read-only far reading, and a room the roster does not know meets this sentence with `· say it to main` on the end of it", + }, + + // ── SILENTLY ABSENT, which is the law obeyed. Still missing, still owed. + "abandonAgent": {loses: "abandoning a turn's second stage; the surface falls back to an ordinary stop"}, + "elsewhereAgent": {loses: "the work this conversation started that is running somewhere else"}, + "leavableRunner": {loses: "watching orchestration runs, which leaves the run page dark"}, + "leavableWaker": {loses: "watching wakes, which leaves a woken turn unannounced"}, + "promoteAgent": {loses: "promoting a call out of the background"}, + "runAgent": {loses: "listing orchestration runs"}, + "spellOutAgent": {loses: "spelling a reply out again in longer form"}, + "standingCountAgent": {loses: "the `◦ n standing orders` count the margin draws; there is no section at all"}, + "taskMentionAgent": {loses: "the task index an @-mention completes from, where the far reading is nil too"}, + "taskWeightDoor": {loses: "one task's context tokens (the conversation's own ContextTokens crosses; the task's does not)"}, + "turnResumer": {loses: "resuming a turn that was stopped"}, + "wakeAgent": {loses: "reading wakes"}, + "workingNowAgent": {loses: "what is working right now, which the margin draws"}, + "interface{ LandingFor/1/2 }": {loses: "the landing a folder already has, beside folderLander"}, + "interface{ PendingConsent/0/1 }": {loses: "which approvals are still open when a surface detaches"}, +} + +// surfaceDoorLedger is the ratchet: the ledger above may shrink and may never +// grow, and shrinking it without lowering this number in the same commit is a +// red as well ([ratchetComplaint]). +const surfaceDoorLedger = 23 + +// TestEverySurfaceDoorTheEngineHasCrossesTheWire is the law above. +func TestEverySurfaceDoorTheEngineHasCrossesTheWire(t *testing.T) { + root := repoRoot(t) + set := token.NewFileSet() + + surfaceInterfaces := interfacesIn(t, set, filepath.Join(root, "internal", "tui3")) + sessionInterfaces := interfacesIn(t, set, filepath.Join(root, "internal", "session")) + known := map[string]map[string]bool{} + for name, methods := range sessionInterfaces { + known[name] = methods + } + for name, methods := range surfaceInterfaces { + // The surface's own spelling wins where both packages declare a name: + // the assertion sites being read are its, so its declarations are the + // ones they mean. + known[name] = methods + } + known = resolveEmbedded(known) + + sessionDoors := methodsOfAgent(t, set, filepath.Join(root, "internal", "session")) + remoteDoors := methodsOfAgent(t, set, filepath.Join(root, "internal", "remote")) + + gaps := map[string][]string{} + for _, asserted := range agentAssertions(t, set, filepath.Join(root, "internal", "tui3"), known) { + if len(asserted.methods) == 0 { + continue + } + if !holds(sessionDoors, asserted.methods) { + // A DOOR THE LOCAL ENGINE DOES NOT HAVE EITHER IS NOT AN ASYMMETRY. + // The surface is reaching for something nothing implements — a test + // double, an engine that has not been written — and the wire owes it + // nothing. + continue + } + if holds(remoteDoors, asserted.methods) { + continue + } + gaps[asserted.name] = missing(remoteDoors, asserted.methods) + } + + for name, absent := range gaps { + if _, written := doorsThatHaveNotCrossed[name]; !written { + t.Errorf("internal/tui3 asserts %s, the engine answers it, and *remote.Agent does not (missing %s).\n"+ + "A door the engine has must cross the wire, or be written into doorsThatHaveNotCrossed with the reason it cannot yet.", + name, strings.Join(absent, ", ")) + } + } + for name := range doorsThatHaveNotCrossed { + if _, still := gaps[name]; !still { + t.Errorf("doorsThatHaveNotCrossed still lists %s, which now crosses the wire. Delete the line and lower surfaceDoorLedger.", name) + } + } + // THE RATCHET FAILS IN BOTH DIRECTIONS, which is the whole of what makes it + // one. internal/ci's own says the second half out loud and it is the half + // that is easy to leave off: a ratchet left slack would let the next change + // put an entry back with the gate green throughout. So a ledger that has + // SHRUNK is a red too, until the number under it comes down in the same + // commit. + if got := len(doorsThatHaveNotCrossed); got != surfaceDoorLedger { + t.Error(ratchetComplaint("doorsThatHaveNotCrossed", got, surfaceDoorLedger)) + } + if _, listed := doorsThatHaveNotCrossed["typingAgent"]; listed { + t.Error("typingAgent is the door this law was written for and it crosses the wire now (typing.go)") + } + // AND EVERY ENTRY SAYS WHAT IT COSTS. A line with no `loses` is a name + // somebody wrote down and did not look at, which is the ledger becoming the + // silence it was written to end. + surface := surfaceProse(t, filepath.Join(root, "internal", "tui3")) + for name, door := range doorsThatHaveNotCrossed { + if strings.TrimSpace(door.loses) == "" { + t.Errorf("%s is on the ledger with nothing said about what a person loses by it", name) + } + // AND A QUOTED SENTENCE IS ONE THE SURFACE ACTUALLY SPELLS. This is what + // keeps the silent/false split a fact rather than a note: respell the + // string, or delete the branch that prints it, and the entry that claims + // a person meets it fails here instead of quietly becoming fiction. + if door.says != "" && !strings.Contains(surface, door.says) { + t.Errorf("%s is on the ledger as saying %q and internal/tui3 no longer spells that.\n"+ + "Either quote what it says now, or — if the sentence is gone — this door may have become silently absent, which is a different entry.", + name, door.says) + } + } +} + +// surfaceProse is internal/tui3's own source as one blob, for the one question +// the ledger asks of it: does the surface still spell this sentence. +func surfaceProse(t *testing.T, dir string) string { + t.Helper() + var prose strings.Builder + listing, err := filepath.Glob(filepath.Join(dir, "*.go")) + if err != nil { + t.Fatal(err) + } + for _, path := range listing { + if strings.HasSuffix(path, "_test.go") { + continue + } + body, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatal(readErr) + } + prose.Write(body) + } + return prose.String() +} + +// TestEveryStreamOpenerIsOrdered refuses a method that names a stream and is +// not ordered. [server.pending] is one slot with no lock, and the whole of what +// makes that safe is that one goroutine owns it ([callClass.opensAStream]); a +// stream opened from any other class would be racing it. +func TestEveryStreamOpenerIsOrdered(t *testing.T) { + root := repoRoot(t) + opened := map[string]bool{} + walkGo(t, filepath.Join(root, "internal", "remote"), func(path string, file *ast.File) { + for _, decl := range file.Decls { + function, ok := decl.(*ast.FuncDecl) + if !ok { + continue + } + ast.Inspect(function, func(node ast.Node) bool { + // A method is a stream opener when its body mentions the one + // slot or the one helper that writes it. + switch shape := node.(type) { + case *ast.SelectorExpr: + if shape.Sel.Name == "pending" || shape.Sel.Name == "stream" { + for _, name := range methodNamesIn(function) { + opened[name] = true + } + } + } + return true + }) + } + }) + for method := range opened { + if !classify(method).opensAStream() { + t.Errorf("%s names a stream from a class that may not: it would race server.pending", method) + } + } + // And the classification itself must not have drifted empty. + for _, method := range []string{MethodSubmit, MethodSubmitImage, MethodSubmitFiles, MethodFollowUp, MethodSteer, MethodObserve} { + if !classify(method).opensAStream() { + t.Errorf("%s opens a stream and must be ordered", method) + } + } + // AND NO CLASS AT ALL RUNS ON THE GOROUTINE THAT READS THE SOCKET. The road + // type has no third value, so this is a statement about the two it has + // rather than a check somebody could forget to extend. + for _, class := range []callClass{classOrdered, classGetter, classAct} { + if class.road() != inOrder && class.road() != onItsOwn { + t.Errorf("a class found a road that is not one of the two") + } + } +} + +// ── the tree readers ──────────────────────────────────────────────────────── + +func repoRoot(t *testing.T) string { + t.Helper() + _, here, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot locate the surface-door law") + } + return filepath.Clean(filepath.Join(filepath.Dir(here), "..", "..")) +} + +func walkGo(t *testing.T, dir string, visit func(string, *ast.File)) { + t.Helper() + set := token.NewFileSet() + err := filepath.WalkDir(dir, func(path string, entry fs.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr + } + if entry.IsDir() { + if path != dir { + // THE WALK IS ONE PACKAGE DEEP AND SAYS SO WHEN THAT STOPS + // BEING ENOUGH. internal/tui3 is flat today; a sub-package + // appearing under it would take its assertions off this law + // silently, so the day one lands is a red rather than a + // coverage hole nobody meets. + if sub, _ := filepath.Glob(filepath.Join(path, "*.go")); len(sub) > 0 { + t.Errorf("%s has grown a sub-package with Go in it; this law walks one package deep", path) + } + return fs.SkipDir + } + return nil + } + if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return nil + } + file, err := parser.ParseFile(set, path, nil, 0) + if err != nil { + return err + } + visit(path, file) + return nil + }) + if err != nil && !os.IsNotExist(err) { + t.Fatal(err) + } +} + +// interfacesIn is every interface declared in one package, as its method names +// plus the names of the interfaces it embeds (resolved afterwards). +func interfacesIn(t *testing.T, _ *token.FileSet, dir string) map[string]map[string]bool { + found := map[string]map[string]bool{} + walkGo(t, dir, func(_ string, file *ast.File) { + for _, decl := range file.Decls { + general, ok := decl.(*ast.GenDecl) + if !ok || general.Tok != token.TYPE { + continue + } + for _, spec := range general.Specs { + typed, ok := spec.(*ast.TypeSpec) + if !ok { + continue + } + shape, ok := typed.Type.(*ast.InterfaceType) + if !ok { + continue + } + found[typed.Name.Name] = interfaceMethods(shape) + } + } + }) + return found +} + +// interfaceMethods is a method set keyed by the name and shape of each method, +// with an embedded interface carried through as its own name for +// [resolveEmbedded] to expand. +func interfaceMethods(shape *ast.InterfaceType) map[string]bool { + methods := map[string]bool{} + if shape.Methods == nil { + return methods + } + for _, field := range shape.Methods.List { + if len(field.Names) == 0 { + if name, ok := field.Type.(*ast.Ident); ok { + methods["embed:"+name.Name] = true + } + if selected, ok := field.Type.(*ast.SelectorExpr); ok { + methods["embed:"+selected.Sel.Name] = true + } + continue + } + function, ok := field.Type.(*ast.FuncType) + if !ok { + continue + } + for _, name := range field.Names { + methods[signature(name.Name, function)] = true + } + } + return methods +} + +// signature is the one spelling a method is compared by: its name and how many +// things go in and come out. +// +// IT IS DELIBERATELY NOT THE TYPES. The same method is written `Answer` in +// internal/session and `session.Answer` in internal/remote, and a comparison +// that read the spelling would report every door as missing. What this catches +// is a door that is absent or has the wrong shape, which is the whole of the +// class. +func signature(name string, function *ast.FuncType) string { + in, out := 0, 0 + if function.Params != nil { + for _, field := range function.Params.List { + in += max(1, len(field.Names)) + } + } + if function.Results != nil { + for _, field := range function.Results.List { + out += max(1, len(field.Names)) + } + } + return name + "/" + strconv.Itoa(in) + "/" + strconv.Itoa(out) +} + +// resolveEmbedded expands every `embed:Name` into the method set of the +// interface it names, repeatedly, so an interface built out of two others +// carries both. +func resolveEmbedded(known map[string]map[string]bool) map[string]map[string]bool { + for range 8 { + settled := true + for name, methods := range known { + for entry := range methods { + embedded, is := strings.CutPrefix(entry, "embed:") + if !is { + continue + } + delete(methods, entry) + settled = false + for inherited := range known[embedded] { + methods[inherited] = true + } + } + known[name] = methods + } + if settled { + break + } + } + return known +} + +type assertion struct { + name string + methods map[string]bool +} + +// agentAssertions is every `…agent.(T)` in the surface, as the method set the +// surface is asking for. +func agentAssertions(t *testing.T, _ *token.FileSet, dir string, known map[string]map[string]bool) []assertion { + var found []assertion + seen := map[string]bool{} + walkGo(t, dir, func(where string, file *ast.File) { + ast.Inspect(file, func(node ast.Node) bool { + asserted, ok := node.(*ast.TypeAssertExpr) + if !ok || asserted.Type == nil { + return true + } + selected, ok := asserted.X.(*ast.SelectorExpr) + if !ok || selected.Sel.Name != "agent" { + return true + } + switch shape := asserted.Type.(type) { + case *ast.Ident: + methods, declared := known[shape.Name] + if !declared { + // AN ASSERTION THIS LAW CANNOT RESOLVE IS NOT ONE IT MAY + // SKIP. Moving a door's interface into a sub-package would + // otherwise take it off the law with the gate green + // throughout, which is the evasion the ledger exists to + // stop. + t.Errorf("%s asserts %s on the agent and this law cannot find its declaration: "+ + "an optional door must be declared where the law can read it", filepath.Base(where), shape.Name) + return true + } + if !seen[shape.Name] { + seen[shape.Name] = true + found = append(found, assertion{name: shape.Name, methods: methods}) + } + case *ast.InterfaceType: + methods := resolveEmbedded(map[string]map[string]bool{"inline": interfaceMethods(shape)})["inline"] + name := "interface{ " + strings.Join(sorted(methods), "; ") + " }" + if !seen[name] { + seen[name] = true + found = append(found, assertion{name: name, methods: methods}) + } + } + return true + }) + }) + sort.Slice(found, func(i, j int) bool { return found[i].name < found[j].name }) + return found +} + +// methodsOfAgent is every method on `*Agent` in one package. +func methodsOfAgent(t *testing.T, _ *token.FileSet, dir string) map[string]bool { + doors := map[string]bool{} + walkGo(t, dir, func(_ string, file *ast.File) { + for _, decl := range file.Decls { + function, ok := decl.(*ast.FuncDecl) + if !ok || function.Recv == nil || len(function.Recv.List) != 1 { + continue + } + pointer, ok := function.Recv.List[0].Type.(*ast.StarExpr) + if !ok { + continue + } + name, ok := pointer.X.(*ast.Ident) + if !ok || name.Name != "Agent" { + continue + } + doors[signature(function.Name.Name, function.Type)] = true + } + }) + return doors +} + +// methodNamesIn is every method name a function declaration belongs to, which +// for a method is its own name and for a plain function is nothing. +func methodNamesIn(function *ast.FuncDecl) []string { + if function.Recv == nil { + return nil + } + // The engine's doors are named by the wire constant the switch matches, not + // by the Go method, so what this returns is the method's own name for the + // few helpers that carry it (server.observe → MethodObserve). + switch function.Name.Name { + case "observe": + return []string{MethodObserve} + case "stream", "release", "dispatch", "serve": + return nil + } + return nil +} + +func holds(doors map[string]bool, wanted map[string]bool) bool { + for method := range wanted { + if !doors[method] { + return false + } + } + return true +} + +func missing(doors map[string]bool, wanted map[string]bool) []string { + var absent []string + for method := range wanted { + if !doors[method] { + absent = append(absent, method) + } + } + sort.Strings(absent) + return absent +} + +func sorted(methods map[string]bool) []string { + names := make([]string, 0, len(methods)) + for method := range methods { + names = append(names, method) + } + sort.Strings(names) + return names +} diff --git a/internal/remote/turnlane_test.go b/internal/remote/turnlane_test.go new file mode 100644 index 0000000000..518334c011 --- /dev/null +++ b/internal/remote/turnlane_test.go @@ -0,0 +1,257 @@ +package remote + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/Agent-Field/aforge-v2/internal/session" +) + +// slowSender is an engine whose Submit takes as long as a real one's preamble: +// the client rebind, the standing orders, the system prompt, the journal write +// and the naming errand all happen before [session.Agent.Submit] returns, and +// on a cold disk that is not free. Held here on purpose, at two seconds, so the +// question is unambiguous. +type slowSender struct { + *fakeAgent + entered chan struct{} + release chan struct{} + once sync.Once + typed chan struct{} + typedOn sync.Once +} + +func (s *slowSender) Submit(ctx context.Context, text string) (<-chan session.Event, error) { + s.once.Do(func() { close(s.entered) }) + <-s.release + return s.fakeAgent.Submit(ctx, text) +} + +func (s *slowSender) Typing() { s.typedOn.Do(func() { close(s.typed) }) } + +// TestASendDoesNotHoldTheSocketShut is the measured defect of this wave, on the +// real protocol. +// +// EVERY FRAME ON THE CONNECTION USED TO WAIT FOR THE SEND. [MethodSubmit] was +// classed ordered, ordered stayed on the reader, and the reader does not read +// the next line until the call it is running returns — so a person who pressed +// Enter and then Esc was queued behind their own message, and so was every +// getter their surface asks from its update loop. +func TestASendDoesNotHoldTheSocketShut(t *testing.T) { + far := &slowSender{ + fakeAgent: &fakeAgent{model: "m"}, + entered: make(chan struct{}), + release: make(chan struct{}), + typed: make(chan struct{}), + } + loop, err := Loopback(Hello{Version: Version}, Options{Boot: func(Hello) (*Engine, error) { + return &Engine{Agent: far, Workspace: "/srv/app", SessionFile: "/srv/app/j.jsonl"}, nil + }}) + if err != nil { + t.Fatalf("dial the loopback: %v", err) + } + released := make(chan struct{}) + var releaseOnce sync.Once + letGo := func() { releaseOnce.Do(func() { close(far.release); close(released) }) } + t.Cleanup(func() { letGo(); _ = loop.Close() }) + + sent := make(chan error, 1) + go func() { + _, err := loop.Client.Agent().Submit(context.Background(), "hello") + sent <- err + }() + select { + case <-far.entered: + case <-time.After(2 * time.Second): + t.Fatal("the send never reached the engine") + } + // The preamble is held for two seconds from here, exactly as the brief for + // this lane stated it, and let go early only if the assertions below fail. + timer := time.AfterFunc(2*time.Second, letGo) + t.Cleanup(func() { timer.Stop() }) + + began := time.Now() + if _, err := loop.Client.Ping(); err != nil { + t.Fatalf("a beat on the same socket was refused while a send was in flight: %v", err) + } + took := time.Since(began) + select { + case <-released: + t.Fatal("the beat only came back after the send let go: the reader is still holding the socket") + default: + } + // TEN MILLISECONDS IS THE BAR AND THE MEASUREMENT IS FAR UNDER IT — a + // loopback round trip is tens of microseconds — so the generous figure here + // is about a shared box under load rather than about the mechanism. + if took > 250*time.Millisecond { + t.Fatalf("a beat behind an in-flight send took %v", took) + } + + letGo() + select { + case err := <-sent: + if err != nil { + t.Fatalf("the send failed: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("the send never finished") + } +} + +// TestTheOrderedLaneRunsWhatItWasGivenInOrder is the guarantee a turn owes, +// tested on the thing that gives it. +// +// IT IS DELIBERATELY NOT DRIVEN THROUGH THE PROTOCOL. Every call on this wire is +// a synchronous round trip, so three ordinary Submits in a row put ONE frame on +// the lane at a time — a protocol-level ordering test would pass against a +// goroutine per call and would be proving nothing. The queue is only ever more +// than one deep when frames arrive faster than they run, which is what this +// hands it, and the mechanism is small enough to state exactly. +func TestTheOrderedLaneRunsWhatItWasGivenInOrder(t *testing.T) { + lane := newOrderedLane() + began := make(chan struct{}) + var ran []uint64 + go lane.run(func(frame Frame) { + if frame.ID == 1 { + // The first piece of work is slow, so everything behind it waits in + // the queue rather than arriving one at a time. + <-began + } + ran = append(ran, frame.ID) + }) + + handed := make(chan time.Duration, 1) + go func() { + start := time.Now() + for id := uint64(1); id <= 200; id++ { + lane.hand(Frame{Kind: "call", ID: id, Method: MethodSubmit}) + } + handed <- time.Since(start) + }() + + // THE HAND-IN NEVER WAITS FOR THE WORK. Two hundred frames go in while the + // first one is still running, which is the property that keeps the reader + // reading (orderedlane.go). + select { + case took := <-handed: + if took > time.Second { + t.Fatalf("handing two hundred frames to a busy lane took %v", took) + } + case <-time.After(5 * time.Second): + t.Fatal("the hand-in is waiting for the work: the reader would be stopped") + } + + close(began) + lane.close() + lane.wait() + + if len(ran) != 200 { + t.Fatalf("the lane ran %d of 200 frames", len(ran)) + } + for i, id := range ran { + if id != uint64(i+1) { + t.Fatalf("frame %d ran in position %d: the lane is not a queue", id, i+1) + } + } +} + +// TestALaneClosedWithWorkOnItStillRunsThatWork is the other half of the +// contract, and the reason [orderedLane.close] does not drop what it holds: a +// frame that crossed the wire is a call the far end is waiting for an answer to, +// and a connection going away is not a reason to leave it unanswered. +func TestALaneClosedWithWorkOnItStillRunsThatWork(t *testing.T) { + lane := newOrderedLane() + for id := uint64(1); id <= 5; id++ { + lane.hand(Frame{ID: id}) + } + lane.close() + ran := 0 + go lane.run(func(Frame) { ran++ }) + lane.wait() + if ran != 5 { + t.Fatalf("a closed lane ran %d of the 5 frames already handed to it", ran) + } + // AND NOTHING HANDED IN AFTER THE CLOSE RUNS. The connection is over; work + // accepted now would write into a session that has been told nobody is here. + lane.hand(Frame{ID: 6}) + if ran != 5 { + t.Fatal("a lane that was closed took new work") + } +} + +// TestAKeystrokeReachesTheEngineOnTheDefaultRoad is the typing door end to end: +// the surface says somebody is writing, the frame crosses, and the engine's own +// door is called — which is what buys the probe, and with it a connection that +// is already open when the turn goes out (typing.go). +func TestAKeystrokeReachesTheEngineOnTheDefaultRoad(t *testing.T) { + far := &slowSender{ + fakeAgent: &fakeAgent{model: "m"}, + entered: make(chan struct{}), + release: make(chan struct{}), + typed: make(chan struct{}), + } + close(far.release) + loop, err := Loopback(Hello{Version: Version}, Options{Boot: func(Hello) (*Engine, error) { + return &Engine{Agent: far, Workspace: "/srv/app", SessionFile: "/srv/app/j.jsonl"}, nil + }}) + if err != nil { + t.Fatalf("dial the loopback: %v", err) + } + t.Cleanup(func() { _ = loop.Close() }) + + agent := loop.Client.Agent() + // THE SURFACE CALLS THIS ON EVERY CHARACTER (internal/tui3's app.key), so + // the door is driven the same way here. + for range 40 { + agent.Typing() + } + select { + case <-far.typed: + case <-time.After(2 * time.Second): + t.Fatal("the engine was never told somebody was writing") + } +} + +// TestTypingIsHushedToOneFrameAWireIsWorth refuses a frame per character. The +// engine's prober is the budget that matters — one pair every twenty seconds +// per model — and this is only about what goes onto the pipe. +func TestTypingIsHushedToOneFrameAWireIsWorth(t *testing.T) { + var beat typingBeat + at := time.Date(2026, 9, 11, 10, 0, 0, 0, time.UTC) + if !beat.due(at) { + t.Fatal("THE FIRST KEYSTROKE MUST GO AT ONCE: the whole value of the signal is how early it arrives") + } + for step := time.Millisecond; step < typingHush; step *= 4 { + if beat.due(at.Add(step)) { + t.Fatalf("a second frame went out %v after the first", step) + } + } + if !beat.due(at.Add(typingHush)) { + t.Fatal("the hush never lifts") + } +} + +// TestATypingFrameIsNeverWaitedFor states the one property that makes the door +// safe to call from a surface's update loop: it is a hint with nobody holding +// the other end, so it registers no call and cannot be made to wait out a +// deadline ([Client.notify]). +func TestATypingFrameIsNeverWaitedFor(t *testing.T) { + far := &fakeAgent{model: "m"} + loop, err := Loopback(Hello{Version: Version}, Options{Boot: func(Hello) (*Engine, error) { + return &Engine{Agent: far, Workspace: "/srv/app", SessionFile: "/srv/app/j.jsonl"}, nil + }}) + if err != nil { + t.Fatalf("dial the loopback: %v", err) + } + t.Cleanup(func() { _ = loop.Close() }) + + loop.Client.Agent().Typing() + loop.Client.mu.Lock() + waiting := len(loop.Client.calls) + loop.Client.mu.Unlock() + if waiting != 0 { + t.Fatalf("a hint left %d calls waiting for an answer", waiting) + } +} diff --git a/internal/remote/typing.go b/internal/remote/typing.go new file mode 100644 index 0000000000..e421f6a67b --- /dev/null +++ b/internal/remote/typing.go @@ -0,0 +1,90 @@ +package remote + +import ( + "sync" + "time" + + "github.com/Agent-Field/aforge-v2/internal/session" +) + +// ── THE KEYSTROKE, OVER A CONNECTION ──────────────────────────────────────── +// +// A PERSON WHO HAS STARTED WRITING IS THE ONLY WARNING THIS BUILD EVER GETS +// that a request is about to go out, and internal/session's [session.Agent] +// spends it on two things: a fresh measurement of the two machines the next +// turn is most likely to use, and a connection already open when it does. The +// second of those is what stops a think-pause costing a handshake — +// internal/provider's probe.go says so in its own header, and it is the half +// nobody could see missing. +// +// IT WAS BROKEN IN TWO PLACES AND BOTH ARE MENDED IN THE SAME CHANGE, because +// mending one of them would have delivered nothing. +// +// THE SURFACE COULD NOT SEND IT. internal/tui3 reaches for the capability +// through an optional interface ([typingAgent]), which a `*session.Agent` +// satisfies and a `*remote.Agent` did not — and the default road holds a +// `*remote.Agent` talking to a detached engine (cmd/aforge/chatv3_local.go's +// v3TakeHostRoad is true for every launch but `--no-host`, `--debug`, first-run +// setup and a hostless `--once`). So the assertion failed silently, on the road +// every person is on. [TestEverySurfaceDoorTheEngineHasCrossesTheWire] is what +// stops that class of gap coming back quietly. +// +// AND THE ENGINE COULD NOT ACT ON IT EITHER, WHICH IS THE OLDER HALF. +// internal/session's [session.Agent.probeClientLanes] asserts an optional +// prober on its completer, and every construction path wraps that completer in +// `sessionCompleter` — which forwarded two methods and not this one. So even +// in process, with the keystroke arriving perfectly, the assertion failed and +// no probe had ever been bought. internal/session's +// [TestTheCompleterWrapperForwardsEveryDoorTheAdapterOffers] is that layer's +// version of the same law. + +// typingDoor is an agent that can be told a person has started writing. It is +// an optional door on the engine's side for the reason it is one on the +// surface's: a task node's agent is deliberately silent here, and a capability +// with nothing behind it is ABSENT rather than present and refusing. +type typingDoor interface{ Typing() } + +var _ typingDoor = (*session.Agent)(nil) + +// typingHush is the shortest gap between two of these frames on one +// connection. +// +// THE BUDGET THAT MATTERS IS THE ENGINE'S, not this one: the prober buys at +// most one pair every twenty seconds per model and asks every gate it has +// inside the call (internal/lane's probe.go). This exists only so that a fast +// typist does not put a frame on the wire per character to be refused eighty +// times a second at the other end. It is leading-edge — the FIRST keystroke +// goes at once, because the whole value of the signal is how early it arrives +// — and a second under it is the honest floor for a hint that is free to lose. +const typingHush = time.Second + +// typingBeat is the hush, kept per connection. +type typingBeat struct { + mu sync.Mutex + last time.Time +} + +// due reports whether a frame may go now, and records that it did. +func (b *typingBeat) due(now time.Time) bool { + b.mu.Lock() + defer b.mu.Unlock() + if !b.last.IsZero() && now.Sub(b.last) < typingHush { + return false + } + b.last = now + return true +} + +// Typing tells the engine somebody is writing. IT DOES NOT WAIT, it reports +// nothing, and a connection that cannot carry it swallows it — which is the +// same contract [session.Agent.Typing] has and the reason the surface may call +// it on every character (internal/tui3's [app.laneTyping]). +func (a *Agent) Typing() { + if a == nil || a.c == nil { + return + } + if !a.c.typing.due(time.Now()) { + return + } + a.c.notify(MethodTyping, nil) +} diff --git a/internal/remote/wire.go b/internal/remote/wire.go index d98ec3be96..acaf4e8f1d 100644 --- a/internal/remote/wire.go +++ b/internal/remote/wire.go @@ -372,11 +372,32 @@ type Frame struct { const ( // Agent — payloads are the method's own argument struct below; results are // the return values likewise. - MethodSubmit = "Submit" // SubmitArgs → StreamRef, then "event" frames - MethodSubmitImage = "SubmitImage" // SubmitImageArgs → StreamRef, then "event" frames - MethodSubmitFiles = "SubmitFiles" // SubmitFilesArgs → StreamRef, then "event" frames - MethodFollowUp = "FollowUp" // SubmitArgs → StreamRef, then "event" frames - MethodSteer = "Steer" // SubmitArgs → StreamRef, then "event" frames + MethodSubmit = "Submit" // SubmitArgs → StreamRef, then "event" frames + MethodSubmitImage = "SubmitImage" // SubmitImageArgs → StreamRef, then "event" frames + MethodSubmitFiles = "SubmitFiles" // SubmitFilesArgs → StreamRef, then "event" frames + MethodFollowUp = "FollowUp" // SubmitArgs → StreamRef, then "event" frames + MethodSteer = "Steer" // SubmitArgs → StreamRef, then "event" frames + // MethodTyping is a person having started writing, and it is the only frame + // on this wire that nobody waits for ([Agent.Typing]). + // + // IT IS ONE OF THE TWO HALVES THAT WERE MISSING FROM THE PROBE. + // internal/session's [session.Agent.Typing] buys a measurement of the two + // machines the next turn is most likely to use, and the second thing it + // buys is a WARM CONNECTION, so the real request's first token is not also + // paying for a handshake (internal/provider's probe.go says so in its own + // header). The surface asks for it through an optional interface, and until + // this door existed the assertion simply failed on the default road — which + // is every launch that is not `--no-host`. The other half was inside the + // engine and is mended in the same change (typing.go's header): the + // completer wrapper was swallowing the probing door, so the measurement had + // never been bought in process either. + // + // IT RIDES VERSION 14 RATHER THAN MOVING THE NUMBER, under the rule stated + // on [Version]: an engine that does not know it answers "no such method", + // the surface drops the answer it was never waiting for, and what is lost is + // a pre-warm nobody can see. Nothing goes dark, so nothing is refused at the + // door. + MethodTyping = "Typing" // nothing → nothing, and nothing waits MethodStopWork = "StopWork" // nothing → nothing; stop this conversation, retaining history MethodInterrupt = "Interrupt" // InterruptArgs, or nothing → nothing MethodCompact = "Compact" // nothing → nothing (error carries the failure) diff --git a/internal/session/agent.go b/internal/session/agent.go index a0a98ccf83..fef2eaea8b 100644 --- a/internal/session/agent.go +++ b/internal/session/agent.go @@ -1993,6 +1993,29 @@ func (f sessionCompleter) CompleteWithMessages(ctx context.Context, messages []a return f.inner.CompleteWithMessages(ctx, messages, options...) } +// ProbeLanes passes the keystroke's pre-warm through, and does nothing at all +// for an inner completer that cannot buy one ([laneProber]). +// +// IT IS THE DOOR THIS WRAPPER SWALLOWED. [FallbackModels] below was written +// against exactly this hazard and the sentence it carries is the whole of why +// this one is here: the wrapper is the ONE thing every request an agent makes +// goes through, so an optional door it does not forward is a door the +// conversation does not have. Nobody forwarded this one, and because +// [Agent.installSessionClient] wraps on every construction path, the assertion +// in [Agent.probeClientLanes] had never once succeeded in a built agent — on the +// default road or in process. The probe that exists so a turn's first token is +// not also paying for a TLS handshake had therefore never fired at all, and the +// test that covered it built an Agent literal and skipped the wrapper. +// [TestTheCompleterWrapperForwardsEveryDoorTheAdapterOffers] is what stops the +// next one going the same way. +func (f sessionCompleter) ProbeLanes(ctx context.Context, model string) { + prober, ok := f.inner.(laneProber) + if !ok { + return + } + prober.ProbeLanes(ctx, model) +} + // FallbackModels passes the adapter's chain through, and answers nil for an // inner completer that has none ([modelChain]). // diff --git a/internal/session/clientdoor.go b/internal/session/clientdoor.go index 57de604716..8571995f92 100644 --- a/internal/session/clientdoor.go +++ b/internal/session/clientdoor.go @@ -273,7 +273,14 @@ func (a *Agent) fallbackModels(model string) []string { } // probeClientLanes asks the optional prober without exposing the conversation -// completer. False means this client has no lane-probing capability. +// completer. False means this completer has no probing DOOR at all — a test +// double, a refusing completer, a build wired to no router. +// +// TRUE IS NOT "A PROBE WAS BOUGHT" AND NEVER WAS. Every gate that could refuse +// one is inside the call — the speed guard, the routing row, the pool's own +// pacing, the prober's budget, a frontier with nothing on it — and this returns +// before any of them is asked. What it reports is whether there was anything to +// ask (sessionCompleter forwards, so a built agent always has one). func (a *Agent) probeClientLanes(ctx context.Context, model string) bool { if a == nil { return false diff --git a/internal/session/completerdoors_law_test.go b/internal/session/completerdoors_law_test.go new file mode 100644 index 0000000000..cc74ae4714 --- /dev/null +++ b/internal/session/completerdoors_law_test.go @@ -0,0 +1,328 @@ +package session + +import ( + "context" + "go/ast" + "go/parser" + "go/token" + "path/filepath" + "runtime" + "sort" + "strings" + "testing" + + "github.com/Agent-Field/aforge-v2/internal/provider" +) + +// ── THE LAW: THE WRAPPER HIDES NO DOOR THE ADAPTER OFFERS ─────────────────── +// +// [sessionCompleter] is the one thing every request an agent makes goes +// through, and it is a WRAPPER: [Agent.installSessionClient] writes it on every +// construction path, so `a.client` is never the adapter itself. Everything this +// package asks the adapter through an optional interface — `client.(modelChain)`, +// `client.(laneProber)` — is therefore asking the WRAPPER, and a door the +// wrapper does not forward is a door the conversation does not have, however +// completely the adapter implements it. +// +// THAT IS NOT A HYPOTHETICAL. `ProbeLanes` was swallowed exactly this way: the +// probe bought by a person starting to type — the one that exists so a turn's +// first token is not also paying for a TLS handshake — had never fired in a +// built agent, on any road, since the wrapper landed. It looked covered, because +// the test that covered it constructed `&Agent{client: adapter}` directly and so +// never met the wrapper at all. A green test over a door nobody could reach is +// the same shape internal/remote's surface-door law was written for, one layer +// down. +// +// So: every optional interface this package type-asserts on a completer, that +// the real adapter satisfies, must be forwarded by [sessionCompleter]. The +// comparison is by name and arity for internal/remote's stated reason — the same +// method is spelled `Answer` in one package and `session.Answer` in another — +// and a door that is absent or the wrong shape is what this catches. + +// TestTheCompleterWrapperForwardsEveryDoorTheAdapterOffers is the law above. +func TestTheCompleterWrapperForwardsEveryDoorTheAdapterOffers(t *testing.T) { + root := completerLawRoot(t) + here := filepath.Join(root, "internal", "session") + + declared := interfacesDeclaredIn(t, here) + asserted := completerAssertionsIn(t, here, declared) + if len(asserted) == 0 { + t.Fatal("this law found no optional door asserted on a completer: the walk has stopped reaching them") + } + adapter := methodsOfType(t, filepath.Join(root, "internal", "provider"), "Client") + wrapper := methodsOfType(t, here, "sessionCompleter") + + for _, door := range asserted { + if !completerHolds(adapter, door.methods) { + // A DOOR THE ADAPTER DOES NOT HAVE EITHER IS NOT AN ASYMMETRY. The + // package is reaching for something only a test double or a future + // completer implements, and the wrapper owes it nothing. + continue + } + if !door.onCompleter { + // AN ASSERTION THIS LAW CANNOT PLACE IS NOT ONE IT MAY SKIP, which + // is internal/remote's surface-door law's rule said on this seam. + // The name list below ([completerNamed]) is the weak point of the + // whole thing: rename the field, or put the completer in a local + // first, and every door this law holds would quietly leave it. So a + // door the ADAPTER answers, asserted on something this law does not + // recognise as the completer, is a red — add the spelling, or say + // in the message why that value is not one. + t.Errorf("internal/session asserts %s on %q, *provider.Client answers it, and this law does not recognise that as the completer.\n"+ + "Either the completer is now spelled something else — teach completerNamed — or this is a different value that happens to share a door's shape.", + door.name, door.receiver) + continue + } + if completerHolds(wrapper, door.methods) { + continue + } + t.Errorf("internal/session asserts %s on a completer, *provider.Client answers it, and sessionCompleter does not forward it (missing %s).\n"+ + "Every request goes through the wrapper, so a door it swallows is a door the conversation does not have.", + door.name, strings.Join(completerMissing(wrapper, door.methods), ", ")) + } +} + +// TestAnAgentBuiltTheOrdinaryWayCanProbe is the behaviour under that law, +// driven through the REAL constructor rather than an Agent literal. +// +// THE LITERAL IS WHY THIS WAS NOT CAUGHT. [TestTypingReachesTheTransportAndOnlyWhereThereIsOne] +// builds `&Agent{client: …}`, which puts the adapter straight into the field so +// the wrapper never exists — and the assertion that failed in every shipped +// binary succeeded there. This one goes through [newAgent], the constructor +// every road opens, and asks whether the probing seam survived it. +func TestAnAgentBuiltTheOrdinaryWayCanProbe(t *testing.T) { + adapter := &probingCompleter{} + agent, err := newAgent(Config{Workspace: t.TempDir(), Model: "sim/model"}, adapter) + if err != nil { + t.Fatalf("build an agent the ordinary way: %v", err) + } + t.Cleanup(func() { _ = agent.Close() }) + + if _, wrapped := agent.client.(sessionCompleter); !wrapped { + t.Fatal("this test no longer goes through the wrapper, which is the only thing it is about") + } + if !agent.probeClientLanes(context.Background(), "sim/model") { + t.Fatal("AN AGENT BUILT THE ORDINARY WAY HAS NO PROBING DOOR: sessionCompleter is swallowing it again") + } + if len(adapter.probed) != 1 || adapter.probed[0] != "sim/model" { + t.Fatalf("the door answered and the adapter underneath was asked for %v", adapter.probed) + } +} + +// A compile-time statement of what this file is about: the adapter this build +// ships has the door, so the wrapper is the only thing that can lose it. +var _ laneProber = (*provider.Client)(nil) + +// ── the tree readers ──────────────────────────────────────────────────────── + +func completerLawRoot(t *testing.T) string { + t.Helper() + _, here, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot locate the completer-door law") + } + return filepath.Clean(filepath.Join(filepath.Dir(here), "..", "..")) +} + +func walkPackage(t *testing.T, dir string, visit func(string, *ast.File)) { + t.Helper() + set := token.NewFileSet() + listing, readErr := filepath.Glob(filepath.Join(dir, "*.go")) + if readErr != nil { + t.Fatal(readErr) + } + for _, path := range listing { + if strings.HasSuffix(path, "_test.go") { + continue + } + file, err := parser.ParseFile(set, path, nil, 0) + if err != nil { + t.Fatal(err) + } + visit(path, file) + } +} + +type completerDoor struct { + name string + methods map[string]bool + // receiver is what the assertion was written on, and onCompleter is whether + // [completerNamed] recognised it. They are carried rather than filtered on + // so that an assertion this law cannot place becomes a MESSAGE instead of a + // silence — see the loop in the law above. + receiver string + onCompleter bool +} + +// interfacesDeclaredIn is every interface a package declares, as a method set +// keyed by name and arity. +func interfacesDeclaredIn(t *testing.T, dir string) map[string]map[string]bool { + found := map[string]map[string]bool{} + walkPackage(t, dir, func(_ string, file *ast.File) { + for _, decl := range file.Decls { + general, ok := decl.(*ast.GenDecl) + if !ok || general.Tok != token.TYPE { + continue + } + for _, spec := range general.Specs { + typed, ok := spec.(*ast.TypeSpec) + if !ok { + continue + } + shape, ok := typed.Type.(*ast.InterfaceType) + if !ok { + continue + } + found[typed.Name.Name] = completerMethodSet(shape) + } + } + }) + return found +} + +func completerMethodSet(shape *ast.InterfaceType) map[string]bool { + methods := map[string]bool{} + if shape.Methods == nil { + return methods + } + for _, field := range shape.Methods.List { + function, ok := field.Type.(*ast.FuncType) + if !ok { + continue + } + for _, name := range field.Names { + methods[completerSignature(name.Name, function)] = true + } + } + return methods +} + +// completerSignature is name and arity, for internal/remote's stated reason: the +// same method is spelled `Answer` in one package and `session.Answer` in another, +// and a comparison that read the spelling would report every door as missing. +func completerSignature(name string, function *ast.FuncType) string { + in, out := 0, 0 + if function.Params != nil { + for _, field := range function.Params.List { + in += max(1, len(field.Names)) + } + } + if function.Results != nil { + for _, field := range function.Results.List { + out += max(1, len(field.Names)) + } + } + return name + "/" + itoaSmall(in) + "/" + itoaSmall(out) +} + +func itoaSmall(n int) string { + if n == 0 { + return "0" + } + digits := "" + for ; n > 0; n /= 10 { + digits = string(rune('0'+n%10)) + digits + } + return digits +} + +// completerAssertionsIn is every `x.(T)` in the package where T is an interface +// the package declares, carrying what x was spelled and whether that spelling is +// one this law recognises as the completer. +// +// IT DOES NOT FILTER BY RECEIVER, and that is the change internal/remote's +// surface-door law asked for. Filtering here would mean a renamed field or a +// local alias took a door off the law with the gate green throughout; carrying +// the spelling instead lets the law say so out loud, for the doors the adapter +// actually answers. +func completerAssertionsIn(t *testing.T, dir string, declared map[string]map[string]bool) []completerDoor { + best := map[string]completerDoor{} + walkPackage(t, dir, func(_ string, file *ast.File) { + ast.Inspect(file, func(node ast.Node) bool { + asserted, ok := node.(*ast.TypeAssertExpr) + if !ok || asserted.Type == nil { + return true + } + name, ok := asserted.Type.(*ast.Ident) + if !ok { + return true + } + methods, known := declared[name.Name] + if !known || len(methods) == 0 { + return true + } + on, named := completerReceiver(asserted.X) + // A RECOGNISED SPELLING WINS wherever one door is asserted on both. + // The door IS on this law when any site asks it of the completer, + // and one local alias elsewhere must not turn that into a complaint + // that it is not. + if prior, already := best[name.Name]; already && (prior.onCompleter || !named) { + return true + } + best[name.Name] = completerDoor{name: name.Name, methods: methods, receiver: on, onCompleter: named} + return true + }) + }) + found := make([]completerDoor, 0, len(best)) + for _, door := range best { + found = append(found, door) + } + sort.Slice(found, func(i, j int) bool { return found[i].name < found[j].name }) + return found +} + +// completerReceiver is what an assertion was written on, and whether that is one +// of the two names this package gives the completer underneath. +func completerReceiver(expr ast.Expr) (string, bool) { + switch shape := expr.(type) { + case *ast.Ident: + return shape.Name, shape.Name == "client" || shape.Name == "inner" + case *ast.SelectorExpr: + return shape.Sel.Name, shape.Sel.Name == "client" || shape.Sel.Name == "inner" + } + return "an expression with no name", false +} + +// methodsOfType is every method on a named type in one package, by either +// receiver form. +func methodsOfType(t *testing.T, dir, typeName string) map[string]bool { + doors := map[string]bool{} + walkPackage(t, dir, func(_ string, file *ast.File) { + for _, decl := range file.Decls { + function, ok := decl.(*ast.FuncDecl) + if !ok || function.Recv == nil || len(function.Recv.List) != 1 { + continue + } + receiver := function.Recv.List[0].Type + if pointer, isPointer := receiver.(*ast.StarExpr); isPointer { + receiver = pointer.X + } + name, ok := receiver.(*ast.Ident) + if !ok || name.Name != typeName { + continue + } + doors[completerSignature(function.Name.Name, function.Type)] = true + } + }) + return doors +} + +func completerHolds(doors, wanted map[string]bool) bool { + for method := range wanted { + if !doors[method] { + return false + } + } + return true +} + +func completerMissing(doors, wanted map[string]bool) []string { + var absent []string + for method := range wanted { + if !doors[method] { + absent = append(absent, method) + } + } + sort.Strings(absent) + return absent +}