bots-garden/mini-me · d7d82e6

modified agent.dmr.yaml +4 -0 Viewed

@@ -142,7 +142,11 @@ sampling:
142	142	top_p: 0.9
143	143	max_tokens: 4096
144	144
	145	+# Silence allowed between two tokens once the answer has started.
145	146	watchdogTimeout: 30s
	147	+# Silence allowed while the server reads the prompt, before the first token
	148	+# (it reports its progress meanwhile, and each report restarts this clock).
	149	+firstTokenTimeout: 3m
146	150
147	151	# Context compression
148	152	context:

    top_p: 0.9
    max_tokens: 4096
++# Silence allowed between two tokens once the answer has started.
  watchdogTimeout: 30s
++# Silence allowed while the server reads the prompt, before the first token
++# (it reports its progress meanwhile, and each report restarts this clock).
++firstTokenTimeout: 3m
  # Context compression
  context:

modified agent.llamacpp.yaml +7 -2 Viewed

@@ -159,10 +159,15 @@ sampling:
159	159	top_p: 0.9
160	160	max_tokens: 4096
161	161
162		-# llama-server processes the whole prompt before the first token;
163		-# on a laptop a 32k context can take a while. Raise it if it fires.
	162	+# Silence allowed between two tokens once the answer has started.
164	163	watchdogTimeout: 30s
165	164
	165	+# Silence allowed while llama-server reads the prompt, before the first token:
	166	+# on a laptop a 25k-token prompt can take minutes. llama-server reports its
	167	+# progress meanwhile ("Reading the prompt 12.0k/25.3k tokens"), and each report
	168	+# restarts this clock, so it only has to cover one batch of the prompt.
	169	+firstTokenTimeout: 3m
	170	+
166	171	# Context compression.
167	172	# ON here: the agent knows the window (read from /props), so `threshold` has a number to apply to.
168	173	context:

    top_p: 0.9
    max_tokens: 4096
--# llama-server processes the whole prompt before the first token;
++# Silence allowed between two tokens once the answer has started.
--# on a laptop a 32k context can take a while. Raise it if it fires.
  watchdogTimeout: 30s
++# Silence allowed while llama-server reads the prompt, before the first token:
++# on a laptop a 25k-token prompt can take minutes. llama-server reports its
++# progress meanwhile ("Reading the prompt 12.0k/25.3k tokens"), and each report
++# restarts this clock, so it only has to cover one batch of the prompt.
++firstTokenTimeout: 3m
++
  # Context compression.
  # ON here: the agent knows the window (read from /props), so `threshold` has a number to apply to.
  context:

modified docs/en/reference/configuration.md +1 -0 Viewed

@@ -23,6 +23,7 @@
23	23	\| `editTools` \| bool \| `true` \| Declare `read_file`, `write_file`, `edit_file`. `false` leaves `bash` and `read_skill` only. \|
24	24	\| `sampling` \| map \| `temperature: 0.0`, `top_p: 0.9`, `max_tokens: 4096` \| Generation settings, OpenAI API key names, passed to the server. The shipped `agent.yaml` adds `parallel_tool_calls: false`. `max_tokens` also bounds one tool call: a `write_file` of a long document (or a thinking model's reasoning plus the call) can hit it, and the call is then cut off. mm hands it back to the model twice, asking for smaller pieces, then stops with `the model hit max_tokens (N) while writing a tool call`; raise the value if that happens. With streaming, mm also asks the server for token counts (`stream_options.include_usage`). \|
25	25	\| `watchdogTimeout` \| duration \| `20s` \| Silence tolerated between two streamed tokens before the connection is declared hung. Doubled for summary requests. \|
	26	+\| `firstTokenTimeout` \| duration \| `3m` \| Silence tolerated before a model call's first token, while the server reads the prompt (every round trip of a turn has one). Never shorter than `watchdogTimeout`; must be >= 0. With a llama.cpp server (`dmr`, `llamacpp`) mm asks for `return_progress`: each progress report restarts this clock, and the spinner shows `Reading the prompt 12.0k/25.3k tokens`. On a timeout the watchdog says `no first token from the server after …` (before the first token) or `connection hang detected after …` (after). \|
26	27	\| `context` \| map \| see below \| Automatic history compression. \|
27	28	\| `sessions` \| map \| see below \| Persistent, per-project sessions. \|
28	29	\| `acp` \| map \| see below \| How `mm -acp` introduces itself to the editor. \|

  | `editTools` | bool | `true` | Declare `read_file`, `write_file`, `edit_file`. `false` leaves `bash` and `read_skill` only. |
  | `sampling` | map | `temperature: 0.0`, `top_p: 0.9`, `max_tokens: 4096` | Generation settings, OpenAI API key names, passed to the server. The shipped `agent.yaml` adds `parallel_tool_calls: false`. `max_tokens` also bounds one tool call: a `write_file` of a long document (or a thinking model's reasoning plus the call) can hit it, and the call is then cut off. mm hands it back to the model twice, asking for smaller pieces, then stops with `the model hit max_tokens (N) while writing a tool call`; raise the value if that happens. With streaming, mm also asks the server for token counts (`stream_options.include_usage`). |
  | `watchdogTimeout` | duration | `20s` | Silence tolerated between two streamed tokens before the connection is declared hung. Doubled for summary requests. |
++| `firstTokenTimeout` | duration | `3m` | Silence tolerated before a model call's first token, while the server reads the prompt (every round trip of a turn has one). Never shorter than `watchdogTimeout`; must be >= 0. With a llama.cpp server (`dmr`, `llamacpp`) mm asks for `return_progress`: each progress report restarts this clock, and the spinner shows `Reading the prompt 12.0k/25.3k tokens`. On a timeout the watchdog says `no first token from the server after …` (before the first token) or `connection hang detected after …` (after). |
  | `context` | map | see below | Automatic history compression. |
  | `sessions` | map | see below | Persistent, per-project sessions. |
  | `acp` | map | see below | How `mm -acp` introduces itself to the editor. |

modified docs/fr/explanation/petits-llms.adoc +2 -0 Viewed

@@ -236,6 +236,7 @@ Un petit modèle choisit mal une procédure sur son seul nom. `internal/skills`
236	236	* Fournisseurs : Docker Model Runner et llama-server sont deux paramétrages d'un même client OpenAI-compatible (`internal/engine/openai_compat.go`), avec une URL de repli (`host.docker.internal`) pour un agent en conteneur.
237	237	* Erreurs expliquées en une ligne : `Explain` traduit « connection refused » en « start it: llama-server … --jinja », signale l'absence de `--jinja` (nécessaire aux appels d'outils), et le 401 sur un serveur sans clé.
238	238	* Sondage de la fenêtre : `/props` de llama-server donne la taille servie ; la bannière en affiche l'origine.
	239	+* Lecture du prompt ≠ blocage : avant son premier token, le serveur lit tout le prompt sans rien envoyer — plusieurs minutes pour 25k tokens sur un 27B local. `mm` demande `return_progress` à llama.cpp, lit les chunks `prompt_progress` pour alimenter le watchdog (et afficher « Reading the prompt 12.0k/25.3k tokens »), et accorde `firstTokenTimeout` (3 min) avant le premier token de chaque appel, `watchdogTimeout` ensuite (`internal/engine/progress.go`). Observé le 26/09/2026 : cinq questions de suite, « hello » compris, tuées en « stalled » par un watchdog de 30 s.
239	240	* Flux interrompu par le serveur : llama.cpp signale une erreur en cours de génération par un événement SSE `error:`, qu'openai-go ignore (il ne restait que `unexpected end of JSON input`). `mm` récupère ce message (`internal/engine/servererr.go`) et relance une fois l'appel au modèle qui a échoué, sans rejouer les outils déjà exécutés dans le tour. Observé le 26/09/2026 : un tour perdu après vingt appels d'outils et un fichier écrit.
240	241
241	242	== Une méthode : mesurer
@@ -442,6 +443,7 @@ Sur une machine locale, un préfixe en cache fait la différence entre une répo
442	443	\| Usage en streaming \| `internal/engine/usage.go`
443	444	\| Appels d'outils coupés ou mal formés \| `internal/engine/toolargs.go`
444	445	\| Erreurs serveur en cours de flux \| `internal/engine/servererr.go`
	446	+\| Watchdog et progression de la lecture du prompt \| `internal/engine/progress.go`
445	447	\| Compression \| `internal/compact/`
446	448	\| Skills et catalogue \| `internal/skills/`, `skills/`
447	449	\| Évaluation du routage \| `internal/eval/`

  * *Fournisseurs* : Docker Model Runner et llama-server sont deux paramétrages d'un même client OpenAI-compatible (`internal/engine/openai_compat.go`), avec une URL de repli (`host.docker.internal`) pour un agent en conteneur.
  * *Erreurs expliquées en une ligne* : `Explain` traduit « connection refused » en « start it: llama-server … --jinja », signale l'absence de `--jinja` (nécessaire aux appels d'outils), et le 401 sur un serveur sans clé.
  * *Sondage de la fenêtre* : `/props` de llama-server donne la taille servie ; la bannière en affiche l'origine.
++* *Lecture du prompt ≠ blocage* : avant son premier token, le serveur lit tout le prompt sans rien envoyer — plusieurs minutes pour 25k tokens sur un 27B local. `mm` demande `return_progress` à llama.cpp, lit les chunks `prompt_progress` pour alimenter le watchdog (et afficher « Reading the prompt 12.0k/25.3k tokens »), et accorde `firstTokenTimeout` (3 min) avant le premier token de chaque appel, `watchdogTimeout` ensuite (`internal/engine/progress.go`). Observé le 26/09/2026 : cinq questions de suite, « hello » compris, tuées en « stalled » par un watchdog de 30 s.
  * *Flux interrompu par le serveur* : llama.cpp signale une erreur en cours de génération par un événement SSE `error:`, qu'openai-go ignore (il ne restait que `unexpected end of JSON input`). `mm` récupère ce message (`internal/engine/servererr.go`) et relance *une fois* l'appel au modèle qui a échoué, sans rejouer les outils déjà exécutés dans le tour. Observé le 26/09/2026 : un tour perdu après vingt appels d'outils et un fichier écrit.
  == Une méthode : mesurer
  | Usage en streaming | `internal/engine/usage.go`
  | Appels d'outils coupés ou mal formés | `internal/engine/toolargs.go`
  | Erreurs serveur en cours de flux | `internal/engine/servererr.go`
++| Watchdog et progression de la lecture du prompt | `internal/engine/progress.go`
  | Compression | `internal/compact/`
  | Skills et catalogue | `internal/skills/`, `skills/`
  | Évaluation du routage | `internal/eval/`

modified docs/fr/reference/configuration.md +1 -0 Viewed

@@ -23,6 +23,7 @@
23	23	\| `editTools` \| booléen \| `true` \| Déclare `read_file`, `write_file`, `edit_file`. `false` ne laisse que `bash` et `read_skill`. \|
24	24	\| `sampling` \| dictionnaire \| `temperature: 0.0`, `top_p: 0.9`, `max_tokens: 4096` \| Paramètres de génération, noms de l'API OpenAI, transmis au serveur. L'`agent.yaml` livré ajoute `parallel_tool_calls: false`. `max_tokens` borne aussi un appel d'outil : un `write_file` de long document (ou le raisonnement d'un modèle « thinking » plus l'appel) peut l'atteindre, et l'appel est alors coupé. mm le rend deux fois au modèle en lui demandant de procéder par morceaux, puis s'arrête avec `the model hit max_tokens (N) while writing a tool call` ; augmentez la valeur si cela arrive. En streaming, mm demande aussi au serveur le compte des tokens (`stream_options.include_usage`). \|
25	25	\| `watchdogTimeout` \| durée \| `20s` \| Silence toléré entre deux tokens diffusés avant de déclarer la connexion bloquée. Doublé pour les requêtes de résumé. \|
	26	+\| `firstTokenTimeout` \| durée \| `3m` \| Silence toléré avant le premier token d'un appel au modèle, pendant que le serveur lit le prompt (chaque aller-retour d'un tour en a un). Jamais plus court que `watchdogTimeout` ; doit être >= 0. Avec un serveur llama.cpp (`dmr`, `llamacpp`), mm demande `return_progress` : chaque rapport d'avancement relance ce délai, et le spinner affiche `Reading the prompt 12.0k/25.3k tokens`. À l'expiration, le watchdog dit `no first token from the server after …` (avant le premier token) ou `connection hang detected after …` (après). \|
26	27	\| `context` \| dictionnaire \| voir ci-dessous \| Compression automatique de l'historique. \|
27	28	\| `sessions` \| dictionnaire \| voir ci-dessous \| Sessions persistantes, par projet. \|
28	29	\| `acp` \| dictionnaire \| voir ci-dessous \| La façon dont `mm -acp` se présente à l'éditeur. \|

  | `editTools` | booléen | `true` | Déclare `read_file`, `write_file`, `edit_file`. `false` ne laisse que `bash` et `read_skill`. |
  | `sampling` | dictionnaire | `temperature: 0.0`, `top_p: 0.9`, `max_tokens: 4096` | Paramètres de génération, noms de l'API OpenAI, transmis au serveur. L'`agent.yaml` livré ajoute `parallel_tool_calls: false`. `max_tokens` borne aussi un appel d'outil : un `write_file` de long document (ou le raisonnement d'un modèle « thinking » plus l'appel) peut l'atteindre, et l'appel est alors coupé. mm le rend deux fois au modèle en lui demandant de procéder par morceaux, puis s'arrête avec `the model hit max_tokens (N) while writing a tool call` ; augmentez la valeur si cela arrive. En streaming, mm demande aussi au serveur le compte des tokens (`stream_options.include_usage`). |
  | `watchdogTimeout` | durée | `20s` | Silence toléré entre deux tokens diffusés avant de déclarer la connexion bloquée. Doublé pour les requêtes de résumé. |
++| `firstTokenTimeout` | durée | `3m` | Silence toléré avant le premier token d'un appel au modèle, pendant que le serveur lit le prompt (chaque aller-retour d'un tour en a un). Jamais plus court que `watchdogTimeout` ; doit être >= 0. Avec un serveur llama.cpp (`dmr`, `llamacpp`), mm demande `return_progress` : chaque rapport d'avancement relance ce délai, et le spinner affiche `Reading the prompt 12.0k/25.3k tokens`. À l'expiration, le watchdog dit `no first token from the server after …` (avant le premier token) ou `connection hang detected after …` (après). |
  | `context` | dictionnaire | voir ci-dessous | Compression automatique de l'historique. |
  | `sessions` | dictionnaire | voir ci-dessous | Sessions persistantes, par projet. |
  | `acp` | dictionnaire | voir ci-dessous | La façon dont `mm -acp` se présente à l'éditeur. |

modified internal/config/config.go +11 -1 Viewed

@@ -106,6 +106,13 @@ type Config struct {
106	106	// WatchdogTimeout: how long to wait for new tokens before assuming the connection has hung.
107	107	WatchdogTimeout time.Duration `yaml:"watchdogTimeout"`
108	108
	109	+ // FirstTokenTimeout: the silence allowed while the server reads the prompt,
	110	+ // before a model call's first token — much longer than WatchdogTimeout: a
	111	+ // local model reads a 25k-token prompt for minutes. A llama.cpp server
	112	+ // reports its progress meanwhile, and each report restarts this clock.
	113	+ // Never shorter than WatchdogTimeout.
	114	+ FirstTokenTimeout time.Duration `yaml:"firstTokenTimeout"`
	115	+
109	116	// Context: when and how the conversation history is compressed
110	117	// (internal/compact). Off unless `context.enabled: true`. The window it
111	118	// measures against is the top-level ContextWindow above — or, when that is
@@ -434,7 +441,8 @@ which ones you left running.`,
434	441	"top_p": 0.9,
435	442	"max_tokens": 4096,
436	443	},
437		- WatchdogTimeout: 20 * time.Second,
	444	+ WatchdogTimeout: 20 * time.Second,
	445	+ FirstTokenTimeout: 3 * time.Minute,
438	446	Sessions: SessionsConfig{
439	447	Enabled: false,
440	448	Dir: ".mm/sessions",
@@ -575,6 +583,8 @@ func (c Config) validate() error {
575	583	return errors.New("model must not be empty")
576	584	case c.ContextWindow < 0:
577	585	return errors.New("contextWindow must be >= 0")
	586	+ case c.FirstTokenTimeout < 0:
	587	+ return errors.New("firstTokenTimeout must be >= 0")
578	588	case c.MaxOutput <= 0:
579	589	return errors.New("maxOutput must be > 0")
580	590	case c.MaxTurns <= 0:

  	// WatchdogTimeout: how long to wait for new tokens before assuming the connection has hung.
  	WatchdogTimeout time.Duration `yaml:"watchdogTimeout"`
++	// FirstTokenTimeout: the silence allowed while the server reads the prompt,
++	// before a model call's first token — much longer than WatchdogTimeout: a
++	// local model reads a 25k-token prompt for minutes. A llama.cpp server
++	// reports its progress meanwhile, and each report restarts this clock.
++	// Never shorter than WatchdogTimeout.
++	FirstTokenTimeout time.Duration `yaml:"firstTokenTimeout"`
++
  	// Context: when and how the conversation history is compressed
  	// (internal/compact). Off unless `context.enabled: true`. The window it
  	// measures against is the top-level ContextWindow above — or, when that is
  		"top_p":       0.9,
  		"max_tokens":  4096,
  	},
--	WatchdogTimeout: 20 * time.Second,
++	WatchdogTimeout:   20 * time.Second,
++	FirstTokenTimeout: 3 * time.Minute,
  	Sessions: SessionsConfig{
  		Enabled: false,
  		Dir:     ".mm/sessions",
  		return errors.New("model must not be empty")
  	case c.ContextWindow < 0:
  		return errors.New("contextWindow must be >= 0")
++	case c.FirstTokenTimeout < 0:
++		return errors.New("firstTokenTimeout must be >= 0")
  	case c.MaxOutput <= 0:
  		return errors.New("maxOutput must be > 0")
  	case c.MaxTurns <= 0:

modified internal/engine/engine.go +26 -44 Viewed

@@ -19,7 +19,6 @@ import (
19	19	"strings"
20	20	"sync"
21	21	"sync/atomic"
22		- "time"
23	22
24	23	"mm/internal/config"
25	24	"mm/internal/spinner"
@@ -253,6 +252,9 @@ func (e Engine) Generate(ctx context.Context, messages []ai.Message, tools []a
253	252	// `New` is called once per `Generate`, which keeps `full` captured at the
254	253	// right level.
255	254	var full []*ai.Message
	255	+ // pl is the watchdog's pulse, set below; every model call restarts the
	256	+ // wait for a first token (progress.go).
	257	+ var pl *pulse
256	258	// argRetries counts the broken tool calls handed back to the model in
257	259	// this generation (toolargs.go).
258	260	argRetries := 0
@@ -264,6 +266,7 @@ func (e Engine) Generate(ctx context.Context, messages []ai.Message, tools []a
264	266	return &ai.Hooks{
265	267	WrapModel: func(ctx context.Context, params ai.ModelParams, next ai.ModelNext) (ai.ModelResponse, error) {
266	268	full = slices.Clone(params.Request.Messages)
	269	+ pl.expect()
267	270	resp, err := next(ctx, params)
268	271	err = withServerMessage(ctx, err)
269	272	// A stream the server broke — an error event, a cut
@@ -276,6 +279,7 @@ func (e Engine) Generate(ctx context.Context, messages []ai.Message, tools []a
276	279	if streamBroken(err) && ctx.Err() == nil {
277	280	spinner.Stop()
278	281	fmt.Fprintf(ui.Out, "\n[%s — retrying this model call once]\n", streamNotice(err))
	282	+ pl.expect()
279	283	resp, err = next(ctx, params)
280	284	err = withServerMessage(ctx, err)
281	285	}
@@ -302,6 +306,7 @@ func (e Engine) Generate(ctx context.Context, messages []ai.Message, tools []a
302	306	what, lastRunes(err.Error(), 60), argRetries, maxArgRetries)
303	307	params.Request.Messages = append(slices.Clone(params.Request.Messages), argsRetryNote(truncated))
304	308	full = slices.Clone(params.Request.Messages)
	309	+ pl.expect()
305	310	resp, err = next(ctx, params)
306	311	err = withServerMessage(ctx, err)
307	312	}
@@ -391,10 +396,15 @@ func (e Engine) Generate(ctx context.Context, messages []ai.Message, tools []a
391	396	}
392	397
393	398	// --- Watchdog ---
394		- // A sub-context the watchdog cancels when no token has arrived for
395		- // WatchdogTimeout; the streaming callback beats it on every chunk.
396		- genCtx, beat, cancelGen := watchdog(ctx, config.Cfg.WatchdogTimeout)
	399	+ // A sub-context the watchdog cancels when the server has been silent too
	400	+ // long: firstTokenTimeout while a model call reads its prompt,
	401	+ // WatchdogTimeout between tokens (progress.go). The streaming callback
	402	+ // beats it on every chunk.
	403	+ genCtx, p, cancelGen := watchdog(ctx, config.Cfg.WatchdogTimeout,
	404	+ firstTokenTimeout(config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout))
397	405	defer cancelGen()
	406	+ pl = p
	407	+ beat := pl.beat
398	408
399	409	// A thinking model streams its reasoning in a field Genkit drops, so the
400	410	// streaming callback below never runs while it thinks (reasoning.go). The
@@ -411,6 +421,14 @@ func (e Engine) Generate(ctx context.Context, messages []ai.Message, tools []a
411	421	// The error events a server streams, for the model calls to report
412	422	// (servererr.go; read back in WrapModel through the context).
413	423	genCtx, _ = withServerErrors(genCtx)
	424	+ // While the server reads the prompt it says how far it got: proof of life
	425	+ // for the watchdog, and something to show instead of "Thinking".
	426	+ genCtx = withPromptProgress(genCtx, func(pp promptProgress) {
	427	+ pl.alive()
	428	+ if pl.waiting.Load() {
	429	+ spinner.Start(progressLabel(pp))
	430	+ }
	431	+ })
414	432
415	433	spinner.Start("Thinking")
416	434	defer spinner.Stop()
@@ -484,45 +502,6 @@ func (e Engine) Generate(ctx context.Context, messages []ai.Message, tools []a
484	502	return resp, full, err
485	503	}
486	504
487		-// watchdog returns a child of ctx that is cancelled when beat() has not been
488		-// called for `timeout`. This is the "silent loop" of STALL_DETECTION_PROBLEM.md
489		-// — a liveness failure, not a behavioural one — and it is shared by Generate
490		-// and Summarize so that both fail the same way.
491		-//
492		-// The timestamp is atomic because beat() runs on the streaming goroutine while
493		-// the ticker reads it from another.
494		-func watchdog(ctx context.Context, timeout time.Duration) (genCtx context.Context, beat func(), cancel context.CancelFunc) {
495		- var last atomic.Int64 // UnixNano
496		- last.Store(time.Now().UnixNano())
497		- genCtx, cancel = context.WithCancel(ctx)
498		-
499		- go func() {
500		- ticker := time.NewTicker(1 * time.Second)
501		- defer ticker.Stop()
502		- for {
503		- select {
504		- case <-genCtx.Done():
505		- return
506		- case <-ticker.C:
507		- if time.Since(time.Unix(0, last.Load())) > timeout {
508		- // Stop() before printing, like everything else: without it
509		- // the watchdog message lands on the spinner's line and the
510		- // next frame erases it — the diagnosis would vanish at the
511		- // very moment it is needed.
512		- spinner.Stop()
513		- // ui.Out, not stdout: in ACP mode this diagnosis belongs to
514		- // the logs (stderr), never to the JSON-RPC stream.
515		- fmt.Fprintf(ui.Out, "\n[watchdog: connection hang detected after %v]\n", timeout)
516		- cancel()
517		- return
518		- }
519		- }
520		- }
521		- }()
522		-
523		- return genCtx, func() { last.Store(time.Now().UnixNano()) }, cancel
524		-}
525		-
526	505	// Summarize asks the model for a plain-text answer to `messages` — no tools,
527	506	// no screen output. It exists for the context compression (internal/compact):
528	507	// the summary must not stream onto the screen as if it were an answer, and a
@@ -545,13 +524,16 @@ func (e Engine) Summarize(ctx context.Context, messages []ai.Message, maxToken
545	524	delete(sampling, "parallel_tool_calls")
546	525	sampling["max_tokens"] = maxTokens
547	526
548		- genCtx, beat, cancel := watchdog(ctx, 2*config.Cfg.WatchdogTimeout)
	527	+ genCtx, pl, cancel := watchdog(ctx, 2*config.Cfg.WatchdogTimeout,
	528	+ firstTokenTimeout(2*config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout))
549	529	defer cancel()
	530	+ beat := pl.beat
550	531	// The same heartbeat, and deliberately no display: a summary is written
551	532	// behind the user's back, and a model that thinks about it must not spill
552	533	// that onto the screen in the middle of their conversation.
553	534	genCtx = WithReasoning(genCtx, func(string) { beat() })
554	535	genCtx, srvErrs := withServerErrors(genCtx)
	536	+ genCtx = withPromptProgress(genCtx, func(promptProgress) { pl.alive() })
555	537
556	538	resp, err := genkit.Generate(genCtx, e.G,
557	539	ai.WithModelName(e.Model),

  	"strings"
  	"sync"
  	"sync/atomic"
--	"time"
  	"mm/internal/config"
  	"mm/internal/spinner"
  	// `New` is called once per `Generate`, which keeps `full` captured at the
  	// right level.
  	var full []*ai.Message
++	// pl is the watchdog's pulse, set below; every model call restarts the
++	// wait for a first token (progress.go).
++	var pl *pulse
  	// argRetries counts the broken tool calls handed back to the model in
  	// this generation (toolargs.go).
  	argRetries := 0
  		return &ai.Hooks{
  			WrapModel: func(ctx context.Context, params *ai.ModelParams, next ai.ModelNext) (*ai.ModelResponse, error) {
  				full = slices.Clone(params.Request.Messages)
++				pl.expect()
  				resp, err := next(ctx, params)
  				err = withServerMessage(ctx, err)
  				// A stream the server broke — an error event, a cut
  				if streamBroken(err) && ctx.Err() == nil {
  					spinner.Stop()
  					fmt.Fprintf(ui.Out, "\n[%s — retrying this model call once]\n", streamNotice(err))
++					pl.expect()
  					resp, err = next(ctx, params)
  					err = withServerMessage(ctx, err)
  				}
  						what, lastRunes(err.Error(), 60), argRetries, maxArgRetries)
  					params.Request.Messages = append(slices.Clone(params.Request.Messages), argsRetryNote(truncated))
  					full = slices.Clone(params.Request.Messages)
++					pl.expect()
  					resp, err = next(ctx, params)
  					err = withServerMessage(ctx, err)
  				}
  	}
  	// --- Watchdog ---
--	// A sub-context the watchdog cancels when no token has arrived for
++	// A sub-context the watchdog cancels when the server has been silent too
--	// WatchdogTimeout; the streaming callback beats it on every chunk.
++	// long: firstTokenTimeout while a model call reads its prompt,
--	genCtx, beat, cancelGen := watchdog(ctx, config.Cfg.WatchdogTimeout)
++	// WatchdogTimeout between tokens (progress.go). The streaming callback
++	// beats it on every chunk.
++	genCtx, p, cancelGen := watchdog(ctx, config.Cfg.WatchdogTimeout,
++		firstTokenTimeout(config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout))
  	defer cancelGen()
++	pl = p
++	beat := pl.beat
  	// A thinking model streams its reasoning in a field Genkit drops, so the
  	// streaming callback below never runs while it thinks (reasoning.go). The
  	// The error events a server streams, for the model calls to report
  	// (servererr.go; read back in WrapModel through the context).
  	genCtx, _ = withServerErrors(genCtx)
++	// While the server reads the prompt it says how far it got: proof of life
++	// for the watchdog, and something to show instead of "Thinking".
++	genCtx = withPromptProgress(genCtx, func(pp promptProgress) {
++		pl.alive()
++		if pl.waiting.Load() {
++			spinner.Start(progressLabel(pp))
++		}
++	})
  	spinner.Start("Thinking")
  	defer spinner.Stop()
  	return resp, full, err
  }
--// watchdog returns a child of ctx that is cancelled when beat() has not been
--// called for `timeout`. This is the "silent loop" of STALL_DETECTION_PROBLEM.md
--// — a liveness failure, not a behavioural one — and it is shared by Generate
--// and Summarize so that both fail the same way.
--//
--// The timestamp is atomic because beat() runs on the streaming goroutine while
--// the ticker reads it from another.
--func watchdog(ctx context.Context, timeout time.Duration) (genCtx context.Context, beat func(), cancel context.CancelFunc) {
--	var last atomic.Int64 // UnixNano
--	last.Store(time.Now().UnixNano())
--	genCtx, cancel = context.WithCancel(ctx)
--
--	go func() {
--		ticker := time.NewTicker(1 * time.Second)
--		defer ticker.Stop()
--		for {
--			select {
--			case <-genCtx.Done():
--				return
--			case <-ticker.C:
--				if time.Since(time.Unix(0, last.Load())) > timeout {
--					// Stop() before printing, like everything else: without it
--					// the watchdog message lands on the spinner's line and the
--					// next frame erases it — the diagnosis would vanish at the
--					// very moment it is needed.
--					spinner.Stop()
--					// ui.Out, not stdout: in ACP mode this diagnosis belongs to
--					// the logs (stderr), never to the JSON-RPC stream.
--					fmt.Fprintf(ui.Out, "\n[watchdog: connection hang detected after %v]\n", timeout)
--					cancel()
--					return
--				}
--			}
--		}
--	}()
--
--	return genCtx, func() { last.Store(time.Now().UnixNano()) }, cancel
--}
--
  // Summarize asks the model for a plain-text answer to `messages` — no tools,
  // no screen output. It exists for the context compression (internal/compact):
  // the summary must not stream onto the screen as if it were an answer, and a
  	delete(sampling, "parallel_tool_calls")
  	sampling["max_tokens"] = maxTokens
--	genCtx, beat, cancel := watchdog(ctx, 2*config.Cfg.WatchdogTimeout)
++	genCtx, pl, cancel := watchdog(ctx, 2*config.Cfg.WatchdogTimeout,
++		firstTokenTimeout(2*config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout))
  	defer cancel()
++	beat := pl.beat
  	// The same heartbeat, and deliberately no display: a summary is written
  	// behind the user's back, and a model that thinks about it must not spill
  	// that onto the screen in the middle of their conversation.
  	genCtx = WithReasoning(genCtx, func(string) { beat() })
  	genCtx, srvErrs := withServerErrors(genCtx)
++	genCtx = withPromptProgress(genCtx, func(promptProgress) { pl.alive() })
  	resp, err := genkit.Generate(genCtx, e.G,
  		ai.WithModelName(e.Model),

modified internal/engine/openai_compat.go +7 -1 Viewed

@@ -46,6 +46,10 @@ type openaiCompat struct {
46	46
47	47	// probe is the server-specific part of Probe; nil = reachability only.
48	48	probe func(ctx context.Context, b Backend) Info
	49	+
	50	+ // progress: the server is llama.cpp and understands return_progress
	51	+ // (usage.go, progress.go). DMR's default backend is llama.cpp.
	52	+ progress bool
49	53	}
50	54
51	55	func init() {
@@ -58,6 +62,7 @@ func init() {
58	62	startHint: "is Docker Model Runner running? (docker model status)",
59	63	notFoundHint: "docker model pull %s",
60	64	probe: probeModels,
	65	+ progress: true,
61	66	})
62	67	register(&openaiCompat{
63	68	name: "llamacpp",
@@ -69,6 +74,7 @@ func init() {
69	74	startHint: "start it: llama-server -m <model.gguf> --jinja --port 8080",
70	75	notFoundHint: "llama-server serves one model; check baseUrl ends with /v1 (model: %s)",
71	76	probe: probeLlamaCpp,
	77	+ progress: true,
72	78	})
73	79	}
74	80
@@ -134,7 +140,7 @@ func (p openaiCompat) Open(ctx context.Context, b Backend) (genkit.Genkit, str
134	140	// on the HTTP client, so a thinking model's reasoning can be read off
135	141	// the wire the plugin drops it from (reasoning.go), and a streamed
136	142	// request asks for the token counts it would otherwise lack (usage.go).
137		- Opts: []option.RequestOption{option.WithMiddleware(streamUsage, reasoningTap)},
	143	+ Opts: []option.RequestOption{option.WithMiddleware(streamOptions(p.progress), reasoningTap)},
138	144	}
139	145	g := genkit.Init(ctx, genkit.WithPlugins(client))
140	146	client.DefineModel(p.name, b.Model, ai.ModelOptions{

  	// probe is the server-specific part of Probe; nil = reachability only.
  	probe func(ctx context.Context, b Backend) Info
++
++	// progress: the server is llama.cpp and understands return_progress
++	// (usage.go, progress.go). DMR's default backend is llama.cpp.
++	progress bool
  }
  func init() {
  		startHint:        "is Docker Model Runner running? (docker model status)",
  		notFoundHint:     "docker model pull %s",
  		probe:            probeModels,
++		progress:         true,
  	})
  	register(&openaiCompat{
  		name:    "llamacpp",
  		startHint:    "start it: llama-server -m <model.gguf> --jinja --port 8080",
  		notFoundHint: "llama-server serves one model; check baseUrl ends with /v1 (model: %s)",
  		probe:        probeLlamaCpp,
++		progress:     true,
  	})
  }
  		// on the HTTP client, so a thinking model's reasoning can be read off
  		// the wire the plugin drops it from (reasoning.go), and a streamed
  		// request asks for the token counts it would otherwise lack (usage.go).
--		Opts: []option.RequestOption{option.WithMiddleware(streamUsage, reasoningTap)},
++		Opts: []option.RequestOption{option.WithMiddleware(streamOptions(p.progress), reasoningTap)},
  	}
  	g := genkit.Init(ctx, genkit.WithPlugins(client))
  	client.DefineModel(p.name, b.Model, ai.ModelOptions{

added internal/engine/progress.go +164 -0 Viewed

new file mode 100644
@@ -0,0 +1,164 @@
	1	+package engine
	2	+
	3	+import (
	4	+ "context"
	5	+ "encoding/json"
	6	+ "fmt"
	7	+ "strings"
	8	+ "sync/atomic"
	9	+ "time"
	10	+
	11	+ "mm/internal/spinner"
	12	+ "mm/internal/ui"
	13	+)
	14	+
	15	+// Before its first token, a server reads the whole prompt — the history, the
	16	+// tools, the system prompt — and streams nothing while it does. On a local
	17	+// model that is the longest silence of a turn, and the one that grows with
	18	+// the session. Seen on 2026-09-26: after a failed turn the next question
	19	+// carried ≈25k tokens to a 27B model, and five questions in a row — "hello"
	20	+// included — ended as "generation timed out (stalled)" under a 30 s watchdog.
	21	+// The server was reading; the agent took it for dead.
	22	+//
	23	+// Two things keep that apart from a real hang:
	24	+//
	25	+// - a llama.cpp server asked for `return_progress` (usage.go) streams
	26	+// `prompt_progress` chunks while it reads. Genkit drops them — they carry
	27	+// no text — so the tap reads them off the wire, like the reasoning, and
	28	+// each one proves the server alive;
	29	+// - the watchdog allows `firstTokenTimeout` of silence while a model call
	30	+// has not produced anything yet, and `watchdogTimeout` once it has. A
	31	+// server without progress, or a slow machine where one batch of the
	32	+// prompt takes longer than watchdogTimeout, still gets the first.
	33	+
	34	+// promptProgress is one `prompt_progress` chunk: how much of the prompt the
	35	+// server has read (Processed) out of Total, Cache of which came from its cache.
	36	+type promptProgress struct {
	37	+ Total int `json:"total"`
	38	+ Cache int `json:"cache"`
	39	+ Processed int `json:"processed"`
	40	+}
	41	+
	42	+// progressKey is the context key carrying the listener.
	43	+type progressKey struct{}
	44	+
	45	+// withPromptProgress returns a context whose streamed requests report the
	46	+// server's prompt progress to `on`.
	47	+func withPromptProgress(ctx context.Context, on func(promptProgress)) context.Context {
	48	+ return context.WithValue(ctx, progressKey{}, on)
	49	+}
	50	+
	51	+func promptProgressFrom(ctx context.Context) func(promptProgress) {
	52	+ on, _ := ctx.Value(progressKey{}).(func(promptProgress))
	53	+ return on
	54	+}
	55	+
	56	+// promptProgressOf reads a `prompt_progress` chunk from an SSE line. The
	57	+// substring test first: this runs on every line of every answer, and all but
	58	+// a few are text.
	59	+func promptProgressOf(line string) (promptProgress, bool) {
	60	+ if !strings.Contains(line, `"prompt_progress"`) {
	61	+ return promptProgress{}, false
	62	+ }
	63	+ data, ok := strings.CutPrefix(strings.TrimSpace(line), "data:")
	64	+ if !ok {
	65	+ return promptProgress{}, false
	66	+ }
	67	+ var c struct {
	68	+ PromptProgress *promptProgress `json:"prompt_progress"`
	69	+ }
	70	+ if json.Unmarshal([]byte(strings.TrimSpace(data)), &c) != nil \|\| c.PromptProgress == nil {
	71	+ return promptProgress{}, false
	72	+ }
	73	+ return *c.PromptProgress, true
	74	+}
	75	+
	76	+// progressLabel is the spinner's wording while the server reads the prompt.
	77	+func progressLabel(p promptProgress) string {
	78	+ if p.Total <= 0 {
	79	+ return "Reading the prompt"
	80	+ }
	81	+ return fmt.Sprintf("Reading the prompt %s/%s tokens", kilo(p.Processed), kilo(p.Total))
	82	+}
	83	+
	84	+// kilo prints a token count the way the banner does: 18.2k, not 18234.
	85	+func kilo(n int) string {
	86	+ if n < 1000 {
	87	+ return fmt.Sprint(n)
	88	+ }
	89	+ return fmt.Sprintf("%.1fk", float64(n)/1000)
	90	+}
	91	+
	92	+// pulse is what the watchdog listens to. The timestamps are atomic because
	93	+// the streaming goroutine and the HTTP reader write them while the ticker
	94	+// reads.
	95	+type pulse struct {
	96	+ last atomic.Int64 // UnixNano of the last sign of life
	97	+ waiting atomic.Bool // no token yet in the current model call
	98	+}
	99	+
	100	+// beat is a token (text, reasoning, a tool call being written): the model is
	101	+// generating, so from now on the short timeout applies.
	102	+func (p *pulse) beat() {
	103	+ p.last.Store(time.Now().UnixNano())
	104	+ p.waiting.Store(false)
	105	+}
	106	+
	107	+// alive is a sign of life that is not a token — a prompt_progress chunk. The
	108	+// clock restarts, but the call is still waiting for its first token.
	109	+func (p *pulse) alive() { p.last.Store(time.Now().UnixNano()) }
	110	+
	111	+// expect starts a model call: its prompt has to be read before anything
	112	+// comes, so the long timeout applies again. A turn makes one call per
	113	+// model ↔ tools round trip, each with a prompt to read.
	114	+func (p *pulse) expect() {
	115	+ p.last.Store(time.Now().UnixNano())
	116	+ p.waiting.Store(true)
	117	+}
	118	+
	119	+// watchdog returns a child of ctx that is cancelled when the pulse has been
	120	+// silent for too long: `firstToken` while a model call has produced nothing
	121	+// yet, `between` once it has. This is the "silent loop" of
	122	+// STALL_DETECTION_PROBLEM.md — a liveness failure, not a behavioural one — and
	123	+// it is shared by Generate and Summarize so that both fail the same way.
	124	+func watchdog(ctx context.Context, between, firstToken time.Duration) (genCtx context.Context, p *pulse, cancel context.CancelFunc) {
	125	+ p = &pulse{}
	126	+ p.expect()
	127	+ genCtx, cancel = context.WithCancel(ctx)
	128	+
	129	+ go func() {
	130	+ ticker := time.NewTicker(1 * time.Second)
	131	+ defer ticker.Stop()
	132	+ for {
	133	+ select {
	134	+ case <-genCtx.Done():
	135	+ return
	136	+ case <-ticker.C:
	137	+ limit, what := between, "connection hang detected"
	138	+ if p.waiting.Load() {
	139	+ limit, what = firstToken, "no first token from the server"
	140	+ }
	141	+ if time.Since(time.Unix(0, p.last.Load())) > limit {
	142	+ // Stop() before printing, like everything else: without it
	143	+ // the watchdog message lands on the spinner's line and the
	144	+ // next frame erases it — the diagnosis would vanish at the
	145	+ // very moment it is needed.
	146	+ spinner.Stop()
	147	+ // ui.Out, not stdout: in ACP mode this diagnosis belongs to
	148	+ // the logs (stderr), never to the JSON-RPC stream.
	149	+ fmt.Fprintf(ui.Out, "\n[watchdog: %s after %v]\n", what, limit)
	150	+ cancel()
	151	+ return
	152	+ }
	153	+ }
	154	+ }
	155	+ }()
	156	+
	157	+ return genCtx, p, cancel
	158	+}
	159	+
	160	+// firstTokenTimeout is the allowance before a model call's first token: the
	161	+// configured one, never shorter than the gap allowed between tokens.
	162	+func firstTokenTimeout(between, configured time.Duration) time.Duration {
	163	+ return max(between, configured)
	164	+}

++package engine
++
++import (
++	"context"
++	"encoding/json"
++	"fmt"
++	"strings"
++	"sync/atomic"
++	"time"
++
++	"mm/internal/spinner"
++	"mm/internal/ui"
++)
++
++// Before its first token, a server reads the whole prompt — the history, the
++// tools, the system prompt — and streams nothing while it does. On a local
++// model that is the longest silence of a turn, and the one that grows with
++// the session. Seen on 2026-09-26: after a failed turn the next question
++// carried ≈25k tokens to a 27B model, and five questions in a row — "hello"
++// included — ended as "generation timed out (stalled)" under a 30 s watchdog.
++// The server was reading; the agent took it for dead.
++//
++// Two things keep that apart from a real hang:
++//
++//   - a llama.cpp server asked for `return_progress` (usage.go) streams
++//     `prompt_progress` chunks while it reads. Genkit drops them — they carry
++//     no text — so the tap reads them off the wire, like the reasoning, and
++//     each one proves the server alive;
++//   - the watchdog allows `firstTokenTimeout` of silence while a model call
++//     has not produced anything yet, and `watchdogTimeout` once it has. A
++//     server without progress, or a slow machine where one batch of the
++//     prompt takes longer than watchdogTimeout, still gets the first.
++
++// promptProgress is one `prompt_progress` chunk: how much of the prompt the
++// server has read (Processed) out of Total, Cache of which came from its cache.
++type promptProgress struct {
++	Total     int `json:"total"`
++	Cache     int `json:"cache"`
++	Processed int `json:"processed"`
++}
++
++// progressKey is the context key carrying the listener.
++type progressKey struct{}
++
++// withPromptProgress returns a context whose streamed requests report the
++// server's prompt progress to `on`.
++func withPromptProgress(ctx context.Context, on func(promptProgress)) context.Context {
++	return context.WithValue(ctx, progressKey{}, on)
++}
++
++func promptProgressFrom(ctx context.Context) func(promptProgress) {
++	on, _ := ctx.Value(progressKey{}).(func(promptProgress))
++	return on
++}
++
++// promptProgressOf reads a `prompt_progress` chunk from an SSE line. The
++// substring test first: this runs on every line of every answer, and all but
++// a few are text.
++func promptProgressOf(line string) (promptProgress, bool) {
++	if !strings.Contains(line, `"prompt_progress"`) {
++		return promptProgress{}, false
++	}
++	data, ok := strings.CutPrefix(strings.TrimSpace(line), "data:")
++	if !ok {
++		return promptProgress{}, false
++	}
++	var c struct {
++		PromptProgress *promptProgress `json:"prompt_progress"`
++	}
++	if json.Unmarshal([]byte(strings.TrimSpace(data)), &c) != nil || c.PromptProgress == nil {
++		return promptProgress{}, false
++	}
++	return *c.PromptProgress, true
++}
++
++// progressLabel is the spinner's wording while the server reads the prompt.
++func progressLabel(p promptProgress) string {
++	if p.Total <= 0 {
++		return "Reading the prompt"
++	}
++	return fmt.Sprintf("Reading the prompt %s/%s tokens", kilo(p.Processed), kilo(p.Total))
++}
++
++// kilo prints a token count the way the banner does: 18.2k, not 18234.
++func kilo(n int) string {
++	if n < 1000 {
++		return fmt.Sprint(n)
++	}
++	return fmt.Sprintf("%.1fk", float64(n)/1000)
++}
++
++// pulse is what the watchdog listens to. The timestamps are atomic because
++// the streaming goroutine and the HTTP reader write them while the ticker
++// reads.
++type pulse struct {
++	last    atomic.Int64 // UnixNano of the last sign of life
++	waiting atomic.Bool  // no token yet in the current model call
++}
++
++// beat is a token (text, reasoning, a tool call being written): the model is
++// generating, so from now on the short timeout applies.
++func (p *pulse) beat() {
++	p.last.Store(time.Now().UnixNano())
++	p.waiting.Store(false)
++}
++
++// alive is a sign of life that is not a token — a prompt_progress chunk. The
++// clock restarts, but the call is still waiting for its first token.
++func (p *pulse) alive() { p.last.Store(time.Now().UnixNano()) }
++
++// expect starts a model call: its prompt has to be read before anything
++// comes, so the long timeout applies again. A turn makes one call per
++// model ↔ tools round trip, each with a prompt to read.
++func (p *pulse) expect() {
++	p.last.Store(time.Now().UnixNano())
++	p.waiting.Store(true)
++}
++
++// watchdog returns a child of ctx that is cancelled when the pulse has been
++// silent for too long: `firstToken` while a model call has produced nothing
++// yet, `between` once it has. This is the "silent loop" of
++// STALL_DETECTION_PROBLEM.md — a liveness failure, not a behavioural one — and
++// it is shared by Generate and Summarize so that both fail the same way.
++func watchdog(ctx context.Context, between, firstToken time.Duration) (genCtx context.Context, p *pulse, cancel context.CancelFunc) {
++	p = &pulse{}
++	p.expect()
++	genCtx, cancel = context.WithCancel(ctx)
++
++	go func() {
++		ticker := time.NewTicker(1 * time.Second)
++		defer ticker.Stop()
++		for {
++			select {
++			case <-genCtx.Done():
++				return
++			case <-ticker.C:
++				limit, what := between, "connection hang detected"
++				if p.waiting.Load() {
++					limit, what = firstToken, "no first token from the server"
++				}
++				if time.Since(time.Unix(0, p.last.Load())) > limit {
++					// Stop() before printing, like everything else: without it
++					// the watchdog message lands on the spinner's line and the
++					// next frame erases it — the diagnosis would vanish at the
++					// very moment it is needed.
++					spinner.Stop()
++					// ui.Out, not stdout: in ACP mode this diagnosis belongs to
++					// the logs (stderr), never to the JSON-RPC stream.
++					fmt.Fprintf(ui.Out, "\n[watchdog: %s after %v]\n", what, limit)
++					cancel()
++					return
++				}
++			}
++		}
++	}()
++
++	return genCtx, p, cancel
++}
++
++// firstTokenTimeout is the allowance before a model call's first token: the
++// configured one, never shorter than the gap allowed between tokens.
++func firstTokenTimeout(between, configured time.Duration) time.Duration {
++	return max(between, configured)
++}

added internal/engine/progress_test.go +128 -0 Viewed

new file mode 100644
@@ -0,0 +1,128 @@
	1	+package engine
	2	+
	3	+import (
	4	+ "context"
	5	+ "fmt"
	6	+ "io"
	7	+ "net/http"
	8	+ "strings"
	9	+ "testing"
	10	+ "time"
	11	+
	12	+ "mm/internal/config"
	13	+
	14	+ "github.com/firebase/genkit/go/ai"
	15	+)
	16	+
	17	+func TestPromptProgressOf(t *testing.T) {
	18	+ p, ok := promptProgressOf(`data: {"choices":[{"index":0,"delta":{"role":"assistant","content":null}}],"prompt_progress":{"total":6013,"cache":6004,"processed":6009,"time_ms":3}}`)
	19	+ if !ok \|\| p.Total != 6013 \|\| p.Cache != 6004 \|\| p.Processed != 6009 {
	20	+ t.Errorf("progress chunk: %+v %v", p, ok)
	21	+ }
	22	+ for _, line := range []string{
	23	+ `data: {"choices":[{"delta":{"content":"prompt_progress is a word"}}]}`,
	24	+ `data: {"choices":[{"delta":{"content":"hi"}}]}`,
	25	+ `: keep-alive`,
	26	+ } {
	27	+ if _, ok := promptProgressOf(line); ok {
	28	+ t.Errorf("%s read as progress", line)
	29	+ }
	30	+ }
	31	+ if got := progressLabel(promptProgress{Total: 25300, Processed: 12000}); got != "Reading the prompt 12.0k/25.3k tokens" {
	32	+ t.Errorf("label %q", got)
	33	+ }
	34	+}
	35	+
	36	+// timeouts sets the two watchdog settings for one test.
	37	+func timeouts(t *testing.T, between, first time.Duration) {
	38	+ t.Helper()
	39	+ oldB, oldF := config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout
	40	+ config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout = between, first
	41	+ t.Cleanup(func() { config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout = oldB, oldF })
	42	+}
	43	+
	44	+// readingServer reads its prompt for `reading` before the first token,
	45	+// sending prompt_progress every `every` (0 = silent), then answers — with a
	46	+// silence of `pause` in the middle of the answer.
	47	+func readingServer(t testing.T, reading, every, pause time.Duration, sawProgress bool) string {
	48	+ srv, _ := fakeLlama(t, func(_ int, w http.ResponseWriter) {
	49	+ fl := w.(http.Flusher)
	50	+ fmt.Fprint(w, sseChunkLine(`{"role":"assistant","content":null}`, ""))
	51	+ fl.Flush()
	52	+ for start := time.Now(); time.Since(start) < reading; {
	53	+ if every == 0 {
	54	+ time.Sleep(reading)
	55	+ break
	56	+ }
	57	+ time.Sleep(every)
	58	+ fmt.Fprint(w, `data: {"choices":[{"index":0,"delta":{"content":null}}],"prompt_progress":{"total":25000,"cache":0,"processed":1000,"time_ms":10}}`+"\n\n")
	59	+ fl.Flush()
	60	+ }
	61	+ fmt.Fprint(w, sseChunkLine(`{"content":"half"}`, ""))
	62	+ fl.Flush()
	63	+ time.Sleep(pause)
	64	+ fmt.Fprint(w, sseChunkLine(`{"content":" answer"}`, ""))
	65	+ fmt.Fprint(w, sseChunkLine(`{}`, "stop"))
	66	+ fmt.Fprint(w, "data: [DONE]\n\n")
	67	+ })
	68	+ if sawProgress != nil {
	69	+ // The provider must ask for it: look at what the next request sends.
	70	+ orig := srv.Config.Handler
	71	+ srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	72	+ body, _ := io.ReadAll(r.Body)
	73	+ *sawProgress = strings.Contains(string(body), `"return_progress":true`)
	74	+ r.Body = io.NopCloser(strings.NewReader(string(body)))
	75	+ orig.ServeHTTP(w, r)
	76	+ })
	77	+ }
	78	+ return srv.URL + "/v1"
	79	+}
	80	+
	81	+func ask(t *testing.T, baseURL string) (string, error) {
	82	+ t.Helper()
	83	+ resp, _, err := engineOn(t, baseURL).Generate(context.Background(), []*ai.Message{ai.NewUserTextMessage("hi")}, nil)
	84	+ if err != nil {
	85	+ return "", err
	86	+ }
	87	+ return resp.Text(), nil
	88	+}
	89	+
	90	+// The server reads for 3 s, well past both timeouts, but says so every
	91	+// 300 ms: the turn is not killed — and the request asked for the progress.
	92	+func TestPromptProgressKeepsTheTurnAlive(t *testing.T) {
	93	+ timeouts(t, time.Second, time.Second)
	94	+ asked := false
	95	+ text, err := ask(t, readingServer(t, 3time.Second, 300time.Millisecond, 0, &asked))
	96	+ if err != nil \|\| text != "half answer" {
	97	+ t.Fatalf("answer %q, err %v", text, err)
	98	+ }
	99	+ if !asked {
	100	+ t.Error("the llamacpp provider did not send return_progress")
	101	+ }
	102	+}
	103	+
	104	+// A server that reads in silence (no progress): the first-token allowance
	105	+// covers it...
	106	+func TestFirstTokenTimeoutCoversASilentPrompt(t *testing.T) {
	107	+ timeouts(t, time.Second, 6*time.Second)
	108	+ if text, err := ask(t, readingServer(t, 3*time.Second, 0, 0, nil)); err != nil \|\| text != "half answer" {
	109	+ t.Fatalf("answer %q, err %v", text, err)
	110	+ }
	111	+}
	112	+
	113	+// ...up to its limit, then it is a hang.
	114	+func TestSilentPromptPastTheAllowanceStalls(t *testing.T) {
	115	+ timeouts(t, time.Second, time.Second)
	116	+ if _, err := ask(t, readingServer(t, 3*time.Second, 0, 0, nil)); err == nil \|\| !strings.Contains(err.Error(), "stalled") {
	117	+ t.Fatalf("err %v, want a stall", err)
	118	+ }
	119	+}
	120	+
	121	+// Once tokens flow, the short timeout is back: a silence in the middle of
	122	+// the answer is still caught quickly.
	123	+func TestSilenceAfterTheFirstTokenStalls(t *testing.T) {
	124	+ timeouts(t, time.Second, time.Minute)
	125	+ if _, err := ask(t, readingServer(t, 0, 0, 3*time.Second, nil)); err == nil \|\| !strings.Contains(err.Error(), "stalled") {
	126	+ t.Fatalf("err %v, want a stall", err)
	127	+ }
	128	+}

++package engine
++
++import (
++	"context"
++	"fmt"
++	"io"
++	"net/http"
++	"strings"
++	"testing"
++	"time"
++
++	"mm/internal/config"
++
++	"github.com/firebase/genkit/go/ai"
++)
++
++func TestPromptProgressOf(t *testing.T) {
++	p, ok := promptProgressOf(`data: {"choices":[{"index":0,"delta":{"role":"assistant","content":null}}],"prompt_progress":{"total":6013,"cache":6004,"processed":6009,"time_ms":3}}`)
++	if !ok || p.Total != 6013 || p.Cache != 6004 || p.Processed != 6009 {
++		t.Errorf("progress chunk: %+v %v", p, ok)
++	}
++	for _, line := range []string{
++		`data: {"choices":[{"delta":{"content":"prompt_progress is a word"}}]}`,
++		`data: {"choices":[{"delta":{"content":"hi"}}]}`,
++		`: keep-alive`,
++	} {
++		if _, ok := promptProgressOf(line); ok {
++			t.Errorf("%s read as progress", line)
++		}
++	}
++	if got := progressLabel(promptProgress{Total: 25300, Processed: 12000}); got != "Reading the prompt 12.0k/25.3k tokens" {
++		t.Errorf("label %q", got)
++	}
++}
++
++// timeouts sets the two watchdog settings for one test.
++func timeouts(t *testing.T, between, first time.Duration) {
++	t.Helper()
++	oldB, oldF := config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout
++	config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout = between, first
++	t.Cleanup(func() { config.Cfg.WatchdogTimeout, config.Cfg.FirstTokenTimeout = oldB, oldF })
++}
++
++// readingServer reads its prompt for `reading` before the first token,
++// sending prompt_progress every `every` (0 = silent), then answers — with a
++// silence of `pause` in the middle of the answer.
++func readingServer(t *testing.T, reading, every, pause time.Duration, sawProgress *bool) string {
++	srv, _ := fakeLlama(t, func(_ int, w http.ResponseWriter) {
++		fl := w.(http.Flusher)
++		fmt.Fprint(w, sseChunkLine(`{"role":"assistant","content":null}`, ""))
++		fl.Flush()
++		for start := time.Now(); time.Since(start) < reading; {
++			if every == 0 {
++				time.Sleep(reading)
++				break
++			}
++			time.Sleep(every)
++			fmt.Fprint(w, `data: {"choices":[{"index":0,"delta":{"content":null}}],"prompt_progress":{"total":25000,"cache":0,"processed":1000,"time_ms":10}}`+"\n\n")
++			fl.Flush()
++		}
++		fmt.Fprint(w, sseChunkLine(`{"content":"half"}`, ""))
++		fl.Flush()
++		time.Sleep(pause)
++		fmt.Fprint(w, sseChunkLine(`{"content":" answer"}`, ""))
++		fmt.Fprint(w, sseChunkLine(`{}`, "stop"))
++		fmt.Fprint(w, "data: [DONE]\n\n")
++	})
++	if sawProgress != nil {
++		// The provider must ask for it: look at what the next request sends.
++		orig := srv.Config.Handler
++		srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++			body, _ := io.ReadAll(r.Body)
++			*sawProgress = strings.Contains(string(body), `"return_progress":true`)
++			r.Body = io.NopCloser(strings.NewReader(string(body)))
++			orig.ServeHTTP(w, r)
++		})
++	}
++	return srv.URL + "/v1"
++}
++
++func ask(t *testing.T, baseURL string) (string, error) {
++	t.Helper()
++	resp, _, err := engineOn(t, baseURL).Generate(context.Background(), []*ai.Message{ai.NewUserTextMessage("hi")}, nil)
++	if err != nil {
++		return "", err
++	}
++	return resp.Text(), nil
++}
++
++// The server reads for 3 s, well past both timeouts, but says so every
++// 300 ms: the turn is not killed — and the request asked for the progress.
++func TestPromptProgressKeepsTheTurnAlive(t *testing.T) {
++	timeouts(t, time.Second, time.Second)
++	asked := false
++	text, err := ask(t, readingServer(t, 3*time.Second, 300*time.Millisecond, 0, &asked))
++	if err != nil || text != "half answer" {
++		t.Fatalf("answer %q, err %v", text, err)
++	}
++	if !asked {
++		t.Error("the llamacpp provider did not send return_progress")
++	}
++}
++
++// A server that reads in silence (no progress): the first-token allowance
++// covers it...
++func TestFirstTokenTimeoutCoversASilentPrompt(t *testing.T) {
++	timeouts(t, time.Second, 6*time.Second)
++	if text, err := ask(t, readingServer(t, 3*time.Second, 0, 0, nil)); err != nil || text != "half answer" {
++		t.Fatalf("answer %q, err %v", text, err)
++	}
++}
++
++// ...up to its limit, then it is a hang.
++func TestSilentPromptPastTheAllowanceStalls(t *testing.T) {
++	timeouts(t, time.Second, time.Second)
++	if _, err := ask(t, readingServer(t, 3*time.Second, 0, 0, nil)); err == nil || !strings.Contains(err.Error(), "stalled") {
++		t.Fatalf("err %v, want a stall", err)
++	}
++}
++
++// Once tokens flow, the short timeout is back: a silence in the middle of
++// the answer is still caught quickly.
++func TestSilenceAfterTheFirstTokenStalls(t *testing.T) {
++	timeouts(t, time.Second, time.Minute)
++	if _, err := ask(t, readingServer(t, 0, 0, 3*time.Second, nil)); err == nil || !strings.Contains(err.Error(), "stalled") {
++		t.Fatalf("err %v, want a stall", err)
++	}
++}

modified internal/engine/reasoning.go +12 -3 Viewed

@@ -59,14 +59,15 @@ func reasoningFrom(ctx context.Context) func(string) {
59	59	func reasoningTap(req http.Request, next option.MiddlewareNext) (http.Response, error) {
60	60	on := reasoningFrom(req.Context())
61	61	errs := serverErrorsFrom(req.Context())
	62	+ progress := promptProgressFrom(req.Context())
62	63	resp, err := next(req)
63		- if (on == nil && errs == nil) \|\| err != nil \|\| resp == nil \|\| resp.Body == nil {
	64	+ if (on == nil && errs == nil && progress == nil) \|\| err != nil \|\| resp == nil \|\| resp.Body == nil {
64	65	return resp, err
65	66	}
66	67	if !strings.Contains(resp.Header.Get("Content-Type"), "text/event-stream") {
67	68	return resp, nil
68	69	}
69		- tap := &sseTap{rc: resp.Body, on: on}
	70	+ tap := &sseTap{rc: resp.Body, on: on, onProgress: progress}
70	71	if errs != nil {
71	72	tap.onErr = errs.record
72	73	}
@@ -86,7 +87,9 @@ type sseTap struct {
86	87	rc io.ReadCloser
87	88	on func(string) // reasoning; nil = not listened to
88	89	onErr func(string) // a server error event; nil = not listened to
89		- buf []byte
	90	+ // onProgress: a prompt_progress chunk (progress.go); nil = not listened to
	91	+ onProgress func(promptProgress)
	92	+ buf []byte
90	93	}
91	94
92	95	func (t *sseTap) Read(p []byte) (int, error) {
@@ -118,6 +121,12 @@ func (t *sseTap) scan(b []byte) {
118	121	}
119	122	continue
120	123	}
	124	+ if p, ok := promptProgressOf(line); ok {
	125	+ if t.onProgress != nil {
	126	+ t.onProgress(p)
	127	+ }
	128	+ continue
	129	+ }
121	130	if t.on == nil {
122	131	continue
123	132	}

  func reasoningTap(req *http.Request, next option.MiddlewareNext) (*http.Response, error) {
  	on := reasoningFrom(req.Context())
  	errs := serverErrorsFrom(req.Context())
++	progress := promptProgressFrom(req.Context())
  	resp, err := next(req)
--	if (on == nil && errs == nil) || err != nil || resp == nil || resp.Body == nil {
++	if (on == nil && errs == nil && progress == nil) || err != nil || resp == nil || resp.Body == nil {
  		return resp, err
  	}
  	if !strings.Contains(resp.Header.Get("Content-Type"), "text/event-stream") {
  		return resp, nil
  	}
--	tap := &sseTap{rc: resp.Body, on: on}
++	tap := &sseTap{rc: resp.Body, on: on, onProgress: progress}
  	if errs != nil {
  		tap.onErr = errs.record
  	}
  	rc    io.ReadCloser
  	on    func(string) // reasoning; nil = not listened to
  	onErr func(string) // a server error event; nil = not listened to
--	buf   []byte
++	// onProgress: a prompt_progress chunk (progress.go); nil = not listened to
++	onProgress func(promptProgress)
++	buf        []byte
  }
  func (t *sseTap) Read(p []byte) (int, error) {
  			}
  			continue
  		}
++		if p, ok := promptProgressOf(line); ok {
++			if t.onProgress != nil {
++				t.onProgress(p)
++			}
++			continue
++		}
  		if t.on == nil {
  			continue
  		}

modified internal/engine/usage.go +37 -18 Viewed

@@ -28,27 +28,38 @@ import (
28	28	// config: that is sent on the non-streamed calls too (the retry, the
29	29	// summariser), and OpenAI rejects stream_options when stream is false.
30	30
31		-// streamUsage is the openai-go middleware that adds the option to a request
32		-// that streams and has not chosen for itself. Anything else — not JSON, not
33		-// streamed, stream_options already set — goes through untouched.
34		-func streamUsage(req http.Request, next option.MiddlewareNext) (http.Response, error) {
35		- if req.Method == http.MethodPost && req.Body != nil {
36		- if body, err := io.ReadAll(req.Body); err == nil {
37		- _ = req.Body.Close()
38		- body = withIncludeUsage(body)
39		- req.Body = io.NopCloser(bytes.NewReader(body))
40		- req.ContentLength = int64(len(body))
41		- req.GetBody = func() (io.ReadCloser, error) {
42		- return io.NopCloser(bytes.NewReader(body)), nil
	31	+// The same middleware asks a llama.cpp server for `return_progress`: while it
	32	+// reads the prompt — before the first token, for a long time on a big one —
	33	+// it then streams chunks carrying `prompt_progress` {total, cache, processed,
	34	+// time_ms} (measured on DMR: one every 3–4 s on a 6k-token prompt). They are
	35	+// what tells "the server is reading" from "the server is stuck" (progress.go).
	36	+// Only for llama.cpp providers: an unknown field is a 400 on some servers.
	37	+
	38	+// streamOptions returns the openai-go middleware that adds those options to a
	39	+// request that streams and has not chosen for itself; `progress` adds
	40	+// return_progress. Anything else — not JSON, not streamed, the options already
	41	+// set — goes through untouched.
	42	+func streamOptions(progress bool) option.Middleware {
	43	+ return func(req http.Request, next option.MiddlewareNext) (http.Response, error) {
	44	+ if req.Method == http.MethodPost && req.Body != nil {
	45	+ if body, err := io.ReadAll(req.Body); err == nil {
	46	+ _ = req.Body.Close()
	47	+ body = withStreamOptions(body, progress)
	48	+ req.Body = io.NopCloser(bytes.NewReader(body))
	49	+ req.ContentLength = int64(len(body))
	50	+ req.GetBody = func() (io.ReadCloser, error) {
	51	+ return io.NopCloser(bytes.NewReader(body)), nil
	52	+ }
43	53	}
44	54	}
	55	+ return next(req)
45	56	}
46		- return next(req)
47	57	}
48	58
49		-// withIncludeUsage returns body with stream_options.include_usage set when it
50		-// is a streamed request that says nothing about it, body unchanged otherwise.
51		-func withIncludeUsage(body []byte) []byte {
	59	+// withStreamOptions returns body with stream_options.include_usage — and
	60	+// return_progress when asked — set when it is a streamed request that says
	61	+// nothing about them, body unchanged otherwise.
	62	+func withStreamOptions(body []byte, progress bool) []byte {
52	63	var m map[string]json.RawMessage
53	64	if json.Unmarshal(body, &m) != nil {
54	65	return body
@@ -56,10 +67,18 @@ func withIncludeUsage(body []byte) []byte {
56	67	if string(m["stream"]) != "true" {
57	68	return body
58	69	}
59		- if _, set := m["stream_options"]; set {
	70	+ changed := false
	71	+ if _, set := m["stream_options"]; !set {
	72	+ m["stream_options"] = json.RawMessage(`{"include_usage":true}`)
	73	+ changed = true
	74	+ }
	75	+ if _, set := m["return_progress"]; progress && !set {
	76	+ m["return_progress"] = json.RawMessage(`true`)
	77	+ changed = true
	78	+ }
	79	+ if !changed {
60	80	return body
61	81	}
62		- m["stream_options"] = json.RawMessage(`{"include_usage":true}`)
63	82	out, err := json.Marshal(m)
64	83	if err != nil {
65	84	return body

  // config: that is sent on the non-streamed calls too (the retry, the
  // summariser), and OpenAI rejects stream_options when stream is false.
--// streamUsage is the openai-go middleware that adds the option to a request
++// The same middleware asks a llama.cpp server for `return_progress`: while it
--// that streams and has not chosen for itself. Anything else — not JSON, not
++// reads the prompt — before the first token, for a long time on a big one —
--// streamed, stream_options already set — goes through untouched.
++// it then streams chunks carrying `prompt_progress` {total, cache, processed,
--func streamUsage(req *http.Request, next option.MiddlewareNext) (*http.Response, error) {
++// time_ms} (measured on DMR: one every 3–4 s on a 6k-token prompt). They are
--	if req.Method == http.MethodPost && req.Body != nil {
++// what tells "the server is reading" from "the server is stuck" (progress.go).
--		if body, err := io.ReadAll(req.Body); err == nil {
++// Only for llama.cpp providers: an unknown field is a 400 on some servers.
--			_ = req.Body.Close()
++
--			body = withIncludeUsage(body)
++// streamOptions returns the openai-go middleware that adds those options to a
--			req.Body = io.NopCloser(bytes.NewReader(body))
++// request that streams and has not chosen for itself; `progress` adds
--			req.ContentLength = int64(len(body))
++// return_progress. Anything else — not JSON, not streamed, the options already
--			req.GetBody = func() (io.ReadCloser, error) {
++// set — goes through untouched.
--				return io.NopCloser(bytes.NewReader(body)), nil
++func streamOptions(progress bool) option.Middleware {
++	return func(req *http.Request, next option.MiddlewareNext) (*http.Response, error) {
++		if req.Method == http.MethodPost && req.Body != nil {
++			if body, err := io.ReadAll(req.Body); err == nil {
++				_ = req.Body.Close()
++				body = withStreamOptions(body, progress)
++				req.Body = io.NopCloser(bytes.NewReader(body))
++				req.ContentLength = int64(len(body))
++				req.GetBody = func() (io.ReadCloser, error) {
++					return io.NopCloser(bytes.NewReader(body)), nil
++				}
  			}
  		}
++		return next(req)
  	}
--	return next(req)
  }
--// withIncludeUsage returns body with stream_options.include_usage set when it
++// withStreamOptions returns body with stream_options.include_usage — and
--// is a streamed request that says nothing about it, body unchanged otherwise.
++// return_progress when asked — set when it is a streamed request that says
--func withIncludeUsage(body []byte) []byte {
++// nothing about them, body unchanged otherwise.
++func withStreamOptions(body []byte, progress bool) []byte {
  	var m map[string]json.RawMessage
  	if json.Unmarshal(body, &m) != nil {
  		return body
  	if string(m["stream"]) != "true" {
  		return body
  	}
--	if _, set := m["stream_options"]; set {
++	changed := false
++	if _, set := m["stream_options"]; !set {
++		m["stream_options"] = json.RawMessage(`{"include_usage":true}`)
++		changed = true
++	}
++	if _, set := m["return_progress"]; progress && !set {
++		m["return_progress"] = json.RawMessage(`true`)
++		changed = true
++	}
++	if !changed {
  		return body
  	}
--	m["stream_options"] = json.RawMessage(`{"include_usage":true}`)
  	out, err := json.Marshal(m)
  	if err != nil {
  		return body

modified internal/engine/usage_test.go +25 -3 Viewed

@@ -13,7 +13,7 @@ import (
13	13	// there), one that already chose, a body that is not JSON — is left as is.
14	14	func TestWithIncludeUsage(t *testing.T) {
15	15	var m map[string]any
16		- got := withIncludeUsage([]byte(`{"model":"m","stream":true,"messages":[]}`))
	16	+ got := withStreamOptions([]byte(`{"model":"m","stream":true,"messages":[]}`), false)
17	17	if err := json.Unmarshal(got, &m); err != nil {
18	18	t.Fatalf("not JSON any more: %s", got)
19	19	}
@@ -31,7 +31,7 @@ func TestWithIncludeUsage(t *testing.T) {
31	31	`{"stream":true,"stream_options":{"include_usage":false}}`,
32	32	`not json`,
33	33	} {
34		- if got := withIncludeUsage([]byte(body)); string(got) != body {
	34	+ if got := withStreamOptions([]byte(body), false); string(got) != body {
35	35	t.Errorf("%s was changed to %s", body, got)
36	36	}
37	37	}
@@ -42,7 +42,7 @@ func TestWithIncludeUsage(t *testing.T) {
42	42	func TestStreamUsageRewritesTheRequest(t *testing.T) {
43	43	req, _ := http.NewRequest(http.MethodPost, "http://x/v1/chat/completions", strings.NewReader(`{"stream":true}`))
44	44	var seen string
45		- _, _ = streamUsage(req, func(r http.Request) (http.Response, error) {
	45	+ _, _ = streamOptions(false)(req, func(r http.Request) (http.Response, error) {
46	46	b, _ := io.ReadAll(r.Body)
47	47	seen = string(b)
48	48	if r.ContentLength != int64(len(b)) {
@@ -58,3 +58,25 @@ func TestStreamUsageRewritesTheRequest(t *testing.T) {
58	58	t.Errorf("body sent: %s", seen)
59	59	}
60	60	}
	61	+
	62	+// A llama.cpp provider also asks for the prompt progress; the others never
	63	+// do (an unknown field is a 400 on some servers), and a request that already
	64	+// chose is left alone.
	65	+func TestWithStreamOptionsProgress(t *testing.T) {
	66	+ var m map[string]any
	67	+ if err := json.Unmarshal(withStreamOptions([]byte(`{"stream":true}`), true), &m); err != nil \|\| m["return_progress"] != true {
	68	+ t.Errorf("progress asked: %v", m)
	69	+ }
	70	+ m = nil
	71	+ _ = json.Unmarshal(withStreamOptions([]byte(`{"stream":true}`), false), &m)
	72	+ if _, set := m["return_progress"]; set {
	73	+ t.Errorf("progress not asked, but sent: %v", m)
	74	+ }
	75	+ body := `{"stream":true,"stream_options":{"include_usage":true},"return_progress":false}`
	76	+ if got := string(withStreamOptions([]byte(body), true)); got != body {
	77	+ t.Errorf("a request that chose was changed: %s", got)
	78	+ }
	79	+ if got := string(withStreamOptions([]byte(`{"stream":false}`), true)); got != `{"stream":false}` {
	80	+ t.Errorf("a non-streamed request was changed: %s", got)
	81	+ }
	82	+}

  // there), one that already chose, a body that is not JSON — is left as is.
  func TestWithIncludeUsage(t *testing.T) {
  	var m map[string]any
--	got := withIncludeUsage([]byte(`{"model":"m","stream":true,"messages":[]}`))
++	got := withStreamOptions([]byte(`{"model":"m","stream":true,"messages":[]}`), false)
  	if err := json.Unmarshal(got, &m); err != nil {
  		t.Fatalf("not JSON any more: %s", got)
  	}
  		`{"stream":true,"stream_options":{"include_usage":false}}`,
  		`not json`,
  	} {
--		if got := withIncludeUsage([]byte(body)); string(got) != body {
++		if got := withStreamOptions([]byte(body), false); string(got) != body {
  			t.Errorf("%s was changed to %s", body, got)
  		}
  	}
  func TestStreamUsageRewritesTheRequest(t *testing.T) {
  	req, _ := http.NewRequest(http.MethodPost, "http://x/v1/chat/completions", strings.NewReader(`{"stream":true}`))
  	var seen string
--	_, _ = streamUsage(req, func(r *http.Request) (*http.Response, error) {
++	_, _ = streamOptions(false)(req, func(r *http.Request) (*http.Response, error) {
  		b, _ := io.ReadAll(r.Body)
  		seen = string(b)
  		if r.ContentLength != int64(len(b)) {
  		t.Errorf("body sent: %s", seen)
  	}
  }
++
++// A llama.cpp provider also asks for the prompt progress; the others never
++// do (an unknown field is a 400 on some servers), and a request that already
++// chose is left alone.
++func TestWithStreamOptionsProgress(t *testing.T) {
++	var m map[string]any
++	if err := json.Unmarshal(withStreamOptions([]byte(`{"stream":true}`), true), &m); err != nil || m["return_progress"] != true {
++		t.Errorf("progress asked: %v", m)
++	}
++	m = nil
++	_ = json.Unmarshal(withStreamOptions([]byte(`{"stream":true}`), false), &m)
++	if _, set := m["return_progress"]; set {
++		t.Errorf("progress not asked, but sent: %v", m)
++	}
++	body := `{"stream":true,"stream_options":{"include_usage":true},"return_progress":false}`
++	if got := string(withStreamOptions([]byte(body), true)); got != body {
++		t.Errorf("a request that chose was changed: %s", got)
++	}
++	if got := string(withStreamOptions([]byte(`{"stream":false}`), true)); got != `{"stream":false}` {
++		t.Errorf("a non-streamed request was changed: %s", got)
++	}
++}

🛟 Updated.