fix(learn): consume /learn?resume= deep links (#164) + ADR-0020 interrupted-turn Retry + promoted streaming journeys (#356)

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")

backend/agents/_providers.py

@@ -280,6 +280,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 ]
 _FUNCTION_HANDLERS: dict[str, FunctionModelHandler] = {}
 
+# Streamed-replay pacing (#356). 0 = off: text parts replay as one whole
+# delta each, exactly the pre-pacing behavior. When a handlers module opts
+# in (set_function_stream_delay_ms), _stream_dispatch re-chunks text into
+# _FUNCTION_STREAM_CHUNK_CHARS-sized deltas with this delay between them, so
+# browser journeys have a real mid-stream window to act in (Stop a turn,
+# switch sessions). Only the streamed lane paces; the JSON lane and the
+# joined stream text stay byte-identical to the handler's constant.
+_FUNCTION_STREAM_DELAY_MS = 0.0
+_FUNCTION_STREAM_CHUNK_CHARS = 24
+
+
+def set_function_stream_delay_ms(ms: float) -> None:
+ """Opt the function-mode STREAMED replay into pacing (see the comment on
+ _FUNCTION_STREAM_DELAY_MS above). Called by boot-time handler modules
+ (agents/function_handlers_e2e.py) rather than wired to another lane env
+ var; `clear_function_handlers()` resets it so in-process tests always
+ start unpaced."""
+ global _FUNCTION_STREAM_DELAY_MS
+ _FUNCTION_STREAM_DELAY_MS = max(0.0, float(ms))
+
+
+def function_stream_delay_ms() -> float:
+ """Public read of the streamed-replay pacing knob (tests pin it)."""
+ return _FUNCTION_STREAM_DELAY_MS
+
 
 class UnregisteredHandlerError(LookupError):
  """Raised by the function-mode dispatch when a task has no registered
@@ -326,9 +351,12 @@ def register_function_handler(task: AgentTask, handler: FunctionModelHandler) ->
 
 
 def clear_function_handlers() -> None:
- """Drop all registered handlers. Tests call this around each case so
- process-global registrations never leak between them."""
+ """Drop all registered handlers and reset the streamed-replay pacing
+ knob. Tests call this around each case so process-global registrations
+ (and a handlers module's pacing opt-in) never leak between them."""
+ global _FUNCTION_STREAM_DELAY_MS
  _FUNCTION_HANDLERS.clear()
+ _FUNCTION_STREAM_DELAY_MS = 0.0
 
 
 # ── Boot-time handler registration for out-of-process runs (#392) ──────────
@@ -401,12 +429,32 @@ async def _stream_dispatch(messages, info):
  # as one delta, each tool call as a DeltaToolCall — so a task's handler
  # constants stay byte-identical between the JSON and streamed lanes
  # (spec assertions compare against the same E2E_* constants either way).
+ #
+ # Pacing (#356): when set_function_stream_delay_ms opted this process
+ # in, text is re-chunked into small deltas with a sleep between them —
+ # a real mid-stream window for browser journeys. The delay is read
+ # AFTER _resolve_handler(), which is what imports a boot-time handlers
+ # module that sets the knob, so the very first paced stream paces.
  response = _resolve_handler()(messages, info)
+ delay_s = _FUNCTION_STREAM_DELAY_MS / 1000.0
+ emitted_text = False
  for index, part in enumerate(response.parts):
  kind = getattr(part, "part_kind", "")
  if kind == "text":
- if part.content:
- yield part.content
+ if not part.content:
+ continue
+ if delay_s > 0:
+ chunks = [
+ part.content[i : i + _FUNCTION_STREAM_CHUNK_CHARS]
+ for i in range(0, len(part.content), _FUNCTION_STREAM_CHUNK_CHARS)
+ ]
+ else:
+ chunks = [part.content]
+ for chunk in chunks:
+ if emitted_text and delay_s > 0:
+ await asyncio.sleep(delay_s)
+ emitted_text = True
+ yield chunk
  elif kind == "tool-call":
  args = part.args
  json_args = args if isinstance(args, str) else json.dumps(args or {})

backend/agents/function_handlers_e2e.py

@@ -31,7 +31,21 @@
 
 from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart
 
-from agents._providers import FunctionModelHandler, register_function_handler
+from agents._providers import (
+ FunctionModelHandler,
+ register_function_handler,
+ set_function_stream_delay_ms,
+)
+
+# Streamed-replay pacing (#356): re-chunk streamed text into small deltas with
+# 150ms between them, giving the mid-stream journeys (Stop a turn, switch
+# sessions while streaming — frontend/e2e/streaming.spec.ts) a real window to
+# act in. Import-time is the right moment: this module only loads in the E2E
+# lane, and the seam reads the knob after resolving the handler, so even the
+# first stream paces. Replies are unchanged byte-for-byte — pacing only slices
+# HOW the same constant streams. The seam tests' clear_function_handlers()
+# resets the knob, so in-process pytest runs stay unpaced.
+set_function_stream_delay_ms(150)
 
 # Asserted verbatim by frontend/e2e/tutor.spec.ts (rendered reply + decrypted
 # messages.content readback). Keep the two literals in sync.
@@ -40,8 +54,53 @@
  "needs a base case so it can stop calling itself."
 )
 
+# Slow lane for mid-stream journeys (#356). A tutor message carrying the
+# trigger substring streams this LONG reply instead — ~1000 chars ≈ 40+ paced
+# chunks ≈ a 6-second window to press Stop or switch sessions inside.
+# Asserted verbatim by frontend/e2e/streaming.spec.ts (including the final
+# sentence as the completion sentinel). Keep the literals in sync.
+E2E_SLOW_STREAM_TRIGGER = "E2E_SLOW_STREAM"
+E2E_TUTOR_SLOW_REPLY = (
+ "[e2e-function-model] Deterministic SLOW tutor reply for mid-stream "
+ "journeys. Recursion solves a problem by reducing it to a smaller copy "
+ "of itself, and every recursive function needs two ingredients: a base "
+ "case that stops the descent, and a recursive step that makes real "
+ "progress toward that base case on every call. Picture the call stack "
+ "as a tower of postponed promises: each frame waits for the smaller "
+ "problem beneath it to resolve before it can finish its own work. When "
+ "the base case finally answers, the tower unwinds in reverse order and "
+ "every waiting frame completes with the value it was promised. If the "
+ "recursive step ever fails to shrink the problem, the tower grows "
+ "without bound until the runtime refuses to add another frame and the "
+ "program crashes with a stack overflow. That is the whole discipline in "
+ "one sentence: shrink toward a base case you are certain to reach. This "
+ "is the final sentence of the slow deterministic reply."
+)
+
+
+def _last_user_prompt_text(messages) -> str:
+ """The most recent user-prompt text in a pydantic-ai message history.
+
+ Used only for trigger sniffing, so it is deliberately tolerant: content
+ may be a plain string or a sequence mixing strings with binary parts
+ (pydantic-ai allows both); non-string members are ignored."""
+ for message in reversed(messages):
+ for part in reversed(getattr(message, "parts", None) or []):
+ if getattr(part, "part_kind", "") != "user-prompt":
+ continue
+ content = getattr(part, "content", "")
+ if isinstance(content, str):
+ return content
+ try:
+ return " ".join(c for c in content if isinstance(c, str))
+ except TypeError:
+ return ""
+ return ""
+
 
 def _chat_tutor_handler(messages, info) -> ModelResponse:
+ if E2E_SLOW_STREAM_TRIGGER in _last_user_prompt_text(messages):
+ return ModelResponse(parts=[TextPart(content=E2E_TUTOR_SLOW_REPLY)])
  return ModelResponse(parts=[TextPart(content=E2E_TUTOR_REPLY)])
 
 

backend/routes/documents.py

@@ -39,7 +39,7 @@
 from services.graph_service import apply_graph_update
 from services.course_context_service import update_course_context
 from services.achievement_service import check_achievements
-from services.agent_events import SaplingEvent, sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
@@ -1053,7 +1053,9 @@ async def event_stream():
  ):
  yield sse_event
 
- return EventSourceResponse(event_stream())
+ return EventSourceResponse(
+ event_stream(), headers={"Cache-Control": SSE_CACHE_CONTROL}
+ )
 
 
 async def _stream_legacy_fallback(

backend/routes/learn.py

@@ -19,7 +19,7 @@
 from db.connection import table
 from services.academics import offering_course_id, resolve_offering
 from models import StartSessionBody, ChatBody, EndSessionBody, ActionBody, ModeSwitchBody, RenameSessionBody
-from services.agent_events import sapling_event_to_sse
+from services.agent_events import SSE_CACHE_CONTROL, sapling_event_to_sse
 from services.auth_guard import require_self, get_session_user_id
 from services.chat_stream import merge_graph_updates, stream_agent_turn
 from services.encryption import encrypt_if_present, encrypt_json, decrypt_if_present, decrypt_json
@@ -869,7 +869,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 
@@ -975,7 +976,8 @@ async def event_stream():
  yield sapling_event_to_sse(ev)
 
  return EventSourceResponse(
- event_stream(), headers={"X-Request-ID": request_id}
+ event_stream(),
+ headers={"X-Request-ID": request_id, "Cache-Control": SSE_CACHE_CONTROL},
  )
 
 

backend/services/agent_events.py

@@ -110,3 +110,17 @@ def sapling_event_to_sse(event: SaplingEvent) -> dict[str, str]:
  JSON-encoded full payload, so the frontend can switch on type and
  still read the full structured event."""
  return {"event": event.type, "data": event.model_dump_json()}
+
+
+# Every EventSourceResponse must pass this as its Cache-Control (#356).
+# `no-transform` opts the stream out of intermediary compression: the
+# frontend proxies /api/* through Next's node server (`next start` — the e2e
+# stack and any self-hosted deploy), which wraps responses in the
+# `compression` middleware unless config.compress is false. gzip BUFFERS
+# small SSE frames, so a paced token stream reaches the browser as one
+# burst at end-of-response — progressive rendering silently broken. The
+# middleware's standard filter skips responses whose Cache-Control contains
+# `no-transform`. `no-store` preserves sse_starlette's own default caching
+# posture (it only ever `setdefault`s Cache-Control, so a route-supplied
+# value replaces it entirely).
+SSE_CACHE_CONTROL = "no-store, no-transform"

backend/tests/test_e2e_function_handlers.py

@@ -325,3 +325,65 @@ def test_concept_describe_handler_passes_real_output_schema(monkeypatch):
 
  assert result.output.description == E2E_CONCEPT_DESCRIPTION
  assert len(E2E_CONCEPT_DESCRIPTION) <= 400
+
+
+# ── Slow-stream trigger + lane pacing (#356) ──────────────────────────────
+#
+# frontend/e2e/streaming.spec.ts needs a mid-stream window to press Stop /
+# switch sessions inside. The env module (a) sets the streamed-replay pacing
+# knob at import, and (b) serves a LONG deterministic reply when the user
+# message carries E2E_SLOW_STREAM_TRIGGER — giving those journeys several
+# seconds of real streaming. The default (no trigger) reply must stay
+# E2E_TUTOR_REPLY byte-for-byte: tutor.spec.ts asserts it verbatim.
+
+
+def test_env_module_slow_trigger_returns_slow_reply(monkeypatch):
+ """A tutor turn whose message carries the trigger gets the long slow-lane
+ constant — through the real agent wiring via the env-autoloaded module."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync(
+ "Walk me through this E2E_SLOW_STREAM please", deps=_deps()
+ )
+
+ from agents.function_handlers_e2e import E2E_TUTOR_SLOW_REPLY
+
+ assert result.output == E2E_TUTOR_SLOW_REPLY
+
+
+def test_env_module_default_reply_unchanged_by_trigger_support(monkeypatch):
+ """Regression guard: a normal message (no trigger) still gets the fixed
+ E2E_TUTOR_REPLY — the slow lane must never hijack the default journey."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ result = socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents.function_handlers_e2e import E2E_TUTOR_REPLY
+
+ assert result.output == E2E_TUTOR_REPLY
+
+
+def test_env_module_import_sets_stream_pacing(monkeypatch):
+ """Importing the module opts the lane into streamed-replay pacing (150ms
+ between chunked deltas) so mid-stream journeys have a window to act in.
+ In-process tests stay unpaced: the autouse registry reset
+ (clear_function_handlers) zeroes the knob again after each case."""
+ monkeypatch.setenv("SAPLING_MODEL_MODE", "function")
+ monkeypatch.setenv(
+ "SAPLING_FUNCTION_HANDLERS", "agents.function_handlers_e2e"
+ )
+
+ with socratic_agent.override(model=model_for("chat_tutor")):
+ socratic_agent.run_sync("What is recursion?", deps=_deps())
+
+ from agents._providers import function_stream_delay_ms
+
+ assert function_stream_delay_ms() == 150

backend/tests/test_learn_stream_routes.py

@@ -196,3 +196,53 @@ async def fake_stream(**kwargs):
  assert stashed["assistant_reply"] == "Legacy greeting"
  assert stashed["topic"] == "Eigenvalues"
  assert "Legacy greeting" in r.text, "the legacy reply must reach the client as a token/done"
+
+
+class TestSseCompressionOptOut:
+ """SSE responses must carry `Cache-Control: no-transform` (#356 journeys).
+
+ The e2e stack (and any self-hosted `next start`) proxies /api/* through
+ Next's production server, which wraps responses in the `compression`
+ middleware unless config.compress is false. gzip BUFFERS small SSE
+ frames: a paced token stream produced nothing client-side for its whole
+ duration and arrived as one burst at `done` — progressive rendering
+ silently broken behind that proxy. `no-transform` is the standard
+ opt-out the middleware honors; sse_starlette only `setdefault`s its
+ own Cache-Control, so the route's value must win.
+ """
+
+ def test_chat_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hi", "graph_update": {}, "mastery_changes": []})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._consume_pending"), \
+ patch("routes.learn._get_session_offering_id", return_value="off-1"), \
+ patch("routes.learn.offering_course_id", return_value="c1"), \
+ patch("routes.learn._load_message_history", return_value=[]):
+ r = client.post("/api/learn/chat/stream", json={
+ "session_id": "s1", "user_id": "u1", "message": "hello", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")
+
+ def test_start_session_stream_sets_no_transform(self):
+ async def fake_stream(**kwargs):
+ from services.agent_events import SaplingEvent
+ yield SaplingEvent(type="done", step="reply", message="Complete.",
+ data={"reply": "Hello", "session_id": "s-new"})
+
+ with patch("routes.learn.stream_agent_turn", fake_stream), \
+ patch("routes.learn._prepare_chat_run",
+ return_value=(MagicMock(), "msg", {}, MagicMock())), \
+ patch("routes.learn._get_course_id_for_topic", return_value=""), \
+ patch("routes.learn.resolve_offering", return_value=""):
+ r = client.post("/api/learn/start-session/stream", json={
+ "user_id": "u1", "topic": "Recursion", "mode": "socratic",
+ })
+ assert r.status_code == 200
+ assert "no-transform" in r.headers.get("cache-control", "")