Deploy from GitLab c3eac333

agent/agent.py

@@ -85,6 +85,7 @@ class Agent:
         conversation: ConversationHistory,
         *,
         documents: dict[str, str] | None = None,
+        request_id: str | None = None,
     ) -> AgentResponse:
         conversation.record_user(query)
 
@@ -120,7 +121,7 @@ class Agent:
                         # the whole skill pipeline (e.g. the wiki sub-agent).
                         with timed_block(f"agent/tool/{tc.name}"):
                             record, should_return = self._dispatch_tool_call(
-                                tc, conversation, documents
+                                tc, conversation, documents, request_id
                             )
                     except Exception as exc:
                         conversation.record_tool_exchange(
@@ -184,6 +185,7 @@ class Agent:
         tc: ToolCall,
         conversation: ConversationHistory,
         documents: dict[str, str] | None = None,
+        request_id: str | None = None,
     ) -> tuple[ToolCallRecord, bool]:
         should_return = False
         try:
@@ -192,7 +194,10 @@ class Agent:
                 result = f"Instructions: {instructions}"
             elif tc.name == "execute_function":
                 result, should_return = self.skills.execute(
-                    **tc.arguments, conversation=conversation, documents=documents
+                    **tc.arguments,
+                    conversation=conversation,
+                    documents=documents,
+                    request_id=request_id,
                 )
             else:
                 result = f"Error: Unknown function: {tc.name}"

agent/agent_client.py

@@ -101,7 +101,7 @@ class AgentClient:
         self._turn_start = len(self.conversation.ordered_transcript())
         self._documents = documents or {}
         try:
-            response = self._invoke(query, lang=lang)
+            response = self._invoke(query, lang=lang, reply_id=reply_id)
         except AgentException as e:
             outcome = ChatOutcome(
                 reply="",
@@ -129,9 +129,11 @@ class AgentClient:
             self.logger_fn(outcome)
         return outcome
 
-    def _invoke(self, query: str, lang: str | None = None) -> AgentResponse:
+    def _invoke(
+        self, query: str, lang: str | None = None, reply_id: str | None = None
+    ) -> AgentResponse:
         assert self.agent is not None, "base _invoke requires self.agent"
-        return self.agent.chat(query, self.conversation)
+        return self.agent.chat(query, self.conversation, request_id=reply_id)
 
     def _inference_impacts(self, n_tokens: int) -> Any:
         """Return raw EcoLogits Impacts for this call, or None if unavailable."""
@@ -259,12 +261,16 @@ class AgentClient:
 class SkillsAgentClient(AgentClient):
     """gpt-oss-style client: post-call language-leak detection + one correction pass."""
 
-    def _invoke(self, query: str, lang: str | None = None) -> AgentResponse:
+    def _invoke(
+        self, query: str, lang: str | None = None, reply_id: str | None = None
+    ) -> AgentResponse:
         with timed_block("client/detect_language"):
             lang = lang or detect_language(query) or "en"
         docs = self._documents or None
         with timed_block("client/agent_chat"):
-            response = self.agent.chat(query, self.conversation, documents=docs)
+            response = self.agent.chat(
+                query, self.conversation, documents=docs, request_id=reply_id
+            )
         with timed_block("client/find_leakage"):
             leaks = find_leakage(response.content, target_lang=lang)
         if not leaks:
@@ -273,9 +279,12 @@ class SkillsAgentClient(AgentClient):
             language=lang, mistranslated_words=leaks
         )
         # Full second pass through the agent (and potentially the whole wiki
-        # pipeline again) — a prime latency suspect.
+        # pipeline again) — a prime latency suspect. Same reply_id: it's still
+        # one turn from the caller's perspective.
         with timed_block("client/translation_correction_chat"):
-            return self.agent.chat(prompt, self.conversation, documents=docs)
+            return self.agent.chat(
+                prompt, self.conversation, documents=docs, request_id=reply_id
+            )
 
     def _extract_context(self) -> list:
         return [

agent/champ_client.py

@@ -60,7 +60,12 @@ class ChampAgentClient(AgentClient):
         self._last_passages: list[str] = []
         self._last_triage: dict = {}
 
-    def _invoke(self, query: str, lang: str | None = None) -> AgentResponse:
+    def _invoke(
+        self, query: str, lang: str | None = None, reply_id: str | None = None
+    ) -> AgentResponse:
+        # reply_id unused: this client calls the provider directly, not
+        # through Agent/SkillsManager, so there's no draft/judge pipeline to
+        # tag with it. Accepted only to match AgentClient.call()'s signature.
         if lang not in ("en", "fr"):
             lang = "en"
         self.conversation.record_user(query)

agent/fake_client.py

@@ -21,7 +21,10 @@ class FakeAgentClient(AgentClient):
         super().__init__(agent=None, logger_fn=logger_fn)
         self.provider = provider
 
-    def _invoke(self, query: str, lang: str | None = None) -> AgentResponse:
+    def _invoke(
+        self, query: str, lang: str | None = None, reply_id: str | None = None
+    ) -> AgentResponse:
+        # reply_id unused — accepted only to match AgentClient.call()'s signature.
         self.conversation.record_user(query)
         completion = self.provider.chat(
             messages=self.conversation.to_messages(),

agent/frontier_client.py

@@ -50,7 +50,12 @@ class _SingleShotClient(AgentClient):
         self.context_window = context_window_for(self.model_id)
         self._last_provider_impacts: Any = None
 
-    def _invoke(self, query: str, lang: str | None = None) -> AgentResponse:
+    def _invoke(
+        self, query: str, lang: str | None = None, reply_id: str | None = None
+    ) -> AgentResponse:
+        # reply_id unused: this client calls the provider directly, not
+        # through Agent/SkillsManager, so there's no draft/judge pipeline to
+        # tag with it. Accepted only to match AgentClient.call()'s signature.
         if lang not in ("en", "fr"):
             lang = "en"
         self.conversation.record_user(query)

agent/skill_decorators.py

@@ -76,3 +76,31 @@ def needs_documents(func: F) -> F:
     """
     func._needs_documents = True  # type: ignore[attr-defined]
     return func
+
+
+def needs_request_id(func: F) -> F:
+    """Mark a skill function as requiring the current turn's request id.
+
+    The dispatch path (`SkillsManager._import_and_call`) sees the
+    `_needs_request_id` attribute and injects a `request_id=` kwarg — the same
+    id `AgentClient.call()` generates once per turn as `reply_id`, threaded
+    down through `Agent.chat()`. Use it to tag log lines so a call can be told
+    apart from an unrelated concurrent or abandoned one sharing the same log
+    stream, instead of a skill minting its own id internally.
+    The function must declare a matching `request_id` parameter.
+
+    Usage:
+
+        from agent.skill_decorators import needs_conversation, needs_request_id
+
+        @needs_conversation
+        @needs_request_id
+        def my_skill_func(
+            user_query: str,
+            conversation: ConversationHistory,
+            request_id: str | None = None,
+        ) -> tuple[str, bool]:
+            ...
+    """
+    func._needs_request_id = True  # type: ignore[attr-defined]
+    return func

agent/skill_manager.py

@@ -222,6 +222,7 @@ class SkillsManager:
         conversation: ConversationHistory,
         params: dict = {},
         documents: dict | None = None,
+        request_id: str | None = None,
     ) -> tuple[str, bool]:
         """Execute skill action by dynamically importing and calling Python functions."""
         if skill_name not in self.skills:
@@ -229,7 +230,7 @@ class SkillsManager:
 
         script_folder = self.skills[skill_name].path.parent / "scripts"
         result = self._import_and_call(
-            script_folder, function_name, conversation, documents, **params
+            script_folder, function_name, conversation, documents, request_id, **params
         )
         if result is not None:
             return result
@@ -245,6 +246,7 @@ class SkillsManager:
         action: str,
         conversation: ConversationHistory,
         documents: dict | None = None,
+        request_id: str | None = None,
         **params,
     ) -> tuple[str, bool]:
         folder_str = str(folder)
@@ -283,6 +285,8 @@ class SkillsManager:
                 params["conversation"] = conversation
             if getattr(func, "_needs_documents", False):
                 params["documents"] = documents
+            if getattr(func, "_needs_request_id", False):
+                params["request_id"] = request_id
 
             # Do NOT swallow exceptions raised inside the skill — surface them
             # so the caller sees what went wrong instead of the misleading

agent/skills/pediatry_wiki/scripts/generate_wiki_response.py

@@ -18,7 +18,7 @@ from agent.context_window import (
 )
 from agent.conversation_history import ConversationHistory
 from agent.documents import build_documents_block, render_documents
-from agent.skill_decorators import needs_conversation, needs_documents
+from agent.skill_decorators import needs_conversation, needs_documents, needs_request_id
 from agent.skills.pediatry_wiki.scripts.browse_category import (
     TOP_INDEX_PATH,
     browse_category,
@@ -320,6 +320,7 @@ def judge_answer(
     use_verbatim_check: bool = False,
     judge_prompt_override: str | None = None,
     echo_stream: bool = False,
+    request_id: str | None = None,
 ) -> tuple[str, str, str | None, list | None]:
     """Return (verdict, reasoning, model_reasoning, claims). verdict is
     PASS / FAIL / ERROR.
@@ -349,9 +350,14 @@ def judge_answer(
     entirely (eval-only). Must take {question}/{reference}/{answer} and
     return the same verdict/reasoning JSON shape.
 
-    `echo_stream=True` prints the judge's reasoning/content tokens to stdout
-    live as they arrive (eval-only — off by default so production calls stay
-    silent).
+    `echo_stream=True` logs the judge's reasoning/content tokens live as they
+    arrive (eval-only — off by default so production calls stay silent).
+
+    `request_id` tags every log line this call emits — including from a
+    timed-out attempt whose thread keeps running in the background and may
+    still log or complete later (HTTP requests can't be cancelled mid-flight;
+    see HFChatProvider.chat's docstring) — so it can be told apart from a
+    concurrent or later call sharing the same log stream.
     """
     rendered_docs = render_documents(documents)
     documents_section = (
@@ -367,7 +373,7 @@ def judge_answer(
         documents_section=documents_section,
     )
 
-    def _call():
+    def _call(call_id: str):
         return _provider().chat(
             messages=[Message(role="user", content=prompt)],
             model_id=JUDGE_MODEL_ID,
@@ -378,13 +384,15 @@ def judge_answer(
             max_tokens=_JUDGE_MODEL_MAX_TOKENS or 65_536,
             stream=True,  # avoids a gateway 504 on long reasoning traces
             echo_stream=echo_stream,
+            request_id=call_id,
         )
 
     completion = None
     last_timeout = False
     for attempt in range(1, JUDGE_CALL_MAX_ATTEMPTS_ON_TIMEOUT + 1):
+        call_id = f"{request_id}:judge:attempt{attempt}"
         try:
-            completion = _judge_call_pool.submit(_call).result(
+            completion = _judge_call_pool.submit(_call, call_id).result(
                 timeout=JUDGE_CALL_TIMEOUT_SECONDS
             )
             last_timeout = False
@@ -392,8 +400,10 @@ def judge_answer(
         except FutureTimeoutError:
             last_timeout = True
             logger.warning(
-                "judge_answer: call exceeded %ss, abandoning and retrying "
-                "(attempt %d/%d)",
+                "[%s] call exceeded %ss, abandoning and retrying "
+                "(attempt %d/%d) — its thread keeps running and may still "
+                "log or complete later",
+                call_id,
                 JUDGE_CALL_TIMEOUT_SECONDS,
                 attempt,
                 JUDGE_CALL_MAX_ATTEMPTS_ON_TIMEOUT,
@@ -404,7 +414,7 @@ def judge_answer(
             # the 3900-token cap); if it still exhausts its own budget and raises,
             # don't let that crash the whole chat turn — degrade to the same
             # graceful ERROR verdict the unparseable-output path below returns.
-            logger.warning("judge_answer: provider call failed: %s", e)
+            logger.warning("[%s] provider call failed: %s", call_id, e)
             return "ERROR", f"Judge provider call failed: {e}", None, None
     if last_timeout:
         # Deliberately not a graceful "ERROR" verdict: that would feed a
@@ -413,8 +423,9 @@ def judge_answer(
         # iteration for no signal, then likely timing out again on the same
         # slow request. Raise instead so the caller can fail once, locally.
         raise ProviderTimeoutExhaustedError(
-            f"judge call timed out after {JUDGE_CALL_MAX_ATTEMPTS_ON_TIMEOUT} "
-            f"attempts of {JUDGE_CALL_TIMEOUT_SECONDS}s each"
+            f"[{request_id}] judge call timed out after "
+            f"{JUDGE_CALL_MAX_ATTEMPTS_ON_TIMEOUT} attempts of "
+            f"{JUDGE_CALL_TIMEOUT_SECONDS}s each"
         )
     if usage_sink is not None:
         usage_sink["prompt_tokens"] = completion.usage.prompt_tokens
@@ -467,7 +478,8 @@ def judge_answer(
 
 
 def check_attribution(
-    answer: str, *, usage_sink: dict | None = None, echo_stream: bool = False
+    answer: str, *, usage_sink: dict | None = None, echo_stream: bool = False,
+    request_id: str | None = None,
 ) -> tuple[str, str, str | None]:
     """Return (verdict, reasoning, model_reasoning). FAIL means the reply
     attributes its answer to an external source instead of speaking directly.
@@ -478,8 +490,10 @@ def check_attribution(
     the thinking trace is worth persisting, not discarding.
 
     `usage_sink`, if given, is populated with this call's prompt_tokens /
-    total_tokens / attempts (see judge_answer). `echo_stream` — see judge_answer."""
+    total_tokens / attempts (see judge_answer). `echo_stream`/`request_id` —
+    see judge_answer."""
     prompt = ATTRIBUTION_PROMPT.format(answer=answer)
+    call_id = f"{request_id}:attribution"
     try:
         completion = _provider().chat(
             messages=[Message(role="user", content=prompt)],
@@ -491,6 +505,7 @@ def check_attribution(
             max_tokens=_JUDGE_MODEL_MAX_TOKENS or 65_536,
             stream=True,  # avoids a gateway 504 on long reasoning traces
             echo_stream=echo_stream,
+            request_id=call_id,
         )
     except ProviderError as e:
         # Same reasoning as judge_answer's wrap: don't let a provider failure
@@ -498,7 +513,7 @@ def check_attribution(
         # caller treats any non-FAIL verdict as "ship the already-judge-passed
         # answer", so degrading to ERROR is a safe, bounded fallback instead of
         # crashing the whole generate_wiki_response call.
-        logger.warning("check_attribution: provider call failed: %s", e)
+        logger.warning("[%s] provider call failed: %s", call_id, e)
         return "ERROR", f"Attribution provider call failed: {e}", None
     if usage_sink is not None:
         usage_sink["prompt_tokens"] = completion.usage.prompt_tokens
@@ -576,10 +591,12 @@ def _read_top_index() -> str:
 
 @needs_conversation
 @needs_documents
+@needs_request_id
 def generate_wiki_response(
     user_query: str,
     conversation: ConversationHistory,
     documents: dict[str, str] | None = None,
+    request_id: str | None = None,
 ) -> tuple[str, bool]:
     """Production entry point: short-answer draft prompt, reject ledger, and
     the split top-level category index — the configuration formerly split
@@ -597,6 +614,7 @@ def generate_wiki_response(
         # (or if) it ever finishes — this is prod, not eval, but nothing here
         # reaches the chat response itself.
         echo_stream=True,
+        request_id=request_id,
     )
 
 
@@ -614,6 +632,7 @@ def generate_wiki_response_with_prompt(
     judge_prompt_override: str | None = None,
     run_attribution_check: bool = True,
     echo_stream: bool = False,
+    request_id: str | None = None,
 ) -> tuple[str, bool]:
     """Restricted mini-agent: read wiki pages → answer → self-correct vs. the judge.
 
@@ -663,8 +682,17 @@ def generate_wiki_response_with_prompt(
     production wiki's category index).
 
     `echo_stream=True` streams every LLM call this makes (draft, judge,
-    attribution) and prints tokens live to stdout as they arrive — for
-    watching a long-running production call, not the chat response itself.
+    attribution) and logs tokens live as they arrive — for watching a
+    long-running production call, not the chat response itself.
+
+    `request_id` is the current turn's id (the same id AgentClient.call()
+    generates as `reply_id`, injected here via `@needs_request_id`) — tags
+    every draft/judge/attribution log line this call produces, including from
+    a timed-out attempt whose thread keeps running in the background and may
+    still log or complete later, so it can be told apart from a concurrent or
+    later call sharing the same log stream. Not minted here: a leaf pipeline
+    function inventing its own id would be disconnected from the id the rest
+    of the system already uses to identify this turn.
 
     Records every sub-step (tool call, tool result, draft, judge verdict,
     improve cycle) as `pipeline_step` events on the outer ConversationHistory.
@@ -685,6 +713,7 @@ def generate_wiki_response_with_prompt(
             "type": "pipeline_step",
             "step": "generate_wiki_response/start",
             "user_query": user_query,
+            "request_id": request_id,
         }
     )
 
@@ -726,8 +755,9 @@ def generate_wiki_response_with_prompt(
                 # would let the model repeat a discarded mistake or answer
                 # ungrounded instead of failing loudly, so bail out instead.
                 logger.warning(
-                    "generate_wiki_response exceeded the context window for "
-                    "%s (iteration %d)",
+                    "[%s] generate_wiki_response exceeded the context window "
+                    "for %s (iteration %d)",
+                    request_id,
                     DRAFT_MODEL_ID,
                     iteration,
                 )
@@ -759,6 +789,7 @@ def generate_wiki_response_with_prompt(
                         max_tokens=_DRAFT_MODEL_MAX_TOKENS or 65_536,
                         stream=True,  # avoids a gateway 504 on long generations
                         echo_stream=echo_stream,
+                        request_id=f"{request_id}:draft:iter{iteration}",
                     )
                     span["prompt_tokens"] = completion.usage.prompt_tokens
                     span["total_tokens"] = completion.usage.total_tokens
@@ -774,7 +805,8 @@ def generate_wiki_response_with_prompt(
                 # from scratch (wiping read_pages/rejected_claims and burning a
                 # full redraft loop) instead of failing once, locally, here.
                 logger.warning(
-                    "generate_wiki_response: subagent provider call failed: %s", e
+                    "[%s] generate_wiki_response: subagent provider call failed: %s",
+                    request_id, e,
                 )
                 conversation.record_event(
                     {
@@ -812,7 +844,8 @@ def generate_wiki_response_with_prompt(
                             wiki_root=wiki_root,
                         )
                     logger.info(
-                        "subagent tool call: %s(%s) -> %s", tc.name, tc.arguments, _clip(result)
+                        "[%s] subagent tool call: %s(%s) -> %s",
+                        request_id, tc.name, tc.arguments, _clip(result),
                     )
                     # Only recall_topic results ground the judge — a category
                     # browse is a navigation aid (which topics exist), not
@@ -854,7 +887,9 @@ def generate_wiki_response_with_prompt(
                     }
                 )
                 logger.info(
-                    "subagent tried to answer before recalling anything (iteration %d): %s",
+                    "[%s] subagent tried to answer before recalling anything "
+                    "(iteration %d): %s",
+                    request_id,
                     iteration,
                     _clip(answer),
                 )
@@ -877,7 +912,8 @@ def generate_wiki_response_with_prompt(
                 }
             )
             logger.info(
-                "subagent draft answer (iteration %d, %d page(s) recalled): %s",
+                "[%s] subagent draft answer (iteration %d, %d page(s) recalled): %s",
+                request_id,
                 iteration,
                 len(read_pages),
                 _clip(answer),
@@ -909,6 +945,7 @@ def generate_wiki_response_with_prompt(
                             use_verbatim_check=use_verbatim_judge,
                             judge_prompt_override=judge_prompt_override,
                             echo_stream=echo_stream,
+                            request_id=f"{request_id}:iter{iteration}",
                         )
                     )
             except ProviderTimeoutExhaustedError as e:
@@ -916,7 +953,10 @@ def generate_wiki_response_with_prompt(
                 # unlike a judge FAIL, there's nothing to feed the redraft loop.
                 # Fail once, locally, same convention as the other no-safe-draft
                 # exits above (context window, subagent provider_error).
-                logger.warning("generate_wiki_response: judge call timed out: %s", e)
+                logger.warning(
+                    "[%s] generate_wiki_response: judge call timed out: %s",
+                    request_id, e,
+                )
                 conversation.record_event(
                     {
                         "type": "pipeline_step",
@@ -940,7 +980,8 @@ def generate_wiki_response_with_prompt(
                 }
             )
             logger.info(
-                "judge verdict (iteration %d): %s — %s", iteration, verdict, _clip(judge_reasoning)
+                "[%s] judge verdict (iteration %d): %s — %s",
+                request_id, iteration, verdict, _clip(judge_reasoning),
             )
             if verdict == "PASS":
                 if not run_attribution_check:
@@ -959,7 +1000,8 @@ def generate_wiki_response_with_prompt(
                 with timed_block("subagent/attribution") as span:
                     attr_verdict, attr_reasoning, attr_model_reasoning = (
                         check_attribution(
-                            answer, usage_sink=span, echo_stream=echo_stream
+                            answer, usage_sink=span, echo_stream=echo_stream,
+                            request_id=f"{request_id}:iter{iteration}",
                         )
                     )
                 conversation.record_event(
@@ -1029,7 +1071,8 @@ def generate_wiki_response_with_prompt(
             msgs.append(Message(role="user", content=improve_prompt))
     except Exception as e:
         logger.exception(
-            "generate_wiki_response: unexpected error, degrading to fallback: %s", e
+            "[%s] generate_wiki_response: unexpected error, degrading to fallback: %s",
+            request_id, e,
         )
         conversation.record_event(
             {
@@ -1042,7 +1085,8 @@ def generate_wiki_response_with_prompt(
         return FALLBACK_MESSAGE, True
 
     logger.warning(
-        "generate_wiki_response exhausted %d iterations without a PASS",
+        "[%s] generate_wiki_response exhausted %d iterations without a PASS",
+        request_id,
         MAX_ITERATIONS,
     )
     conversation.record_event(

agent/skills/pediatry_wiki_minimal/scripts/generate_wiki_response_minimal.py

@@ -18,7 +18,7 @@ generate_wiki_response, which is what SKILL.md instructs the model to call.
 """
 
 from agent.conversation_history import ConversationHistory
-from agent.skill_decorators import needs_conversation, needs_documents
+from agent.skill_decorators import needs_conversation, needs_documents, needs_request_id
 from agent.skills.pediatry_wiki.scripts.generate_wiki_response import (
     generate_wiki_response_with_prompt,
 )
@@ -32,10 +32,12 @@ from agent.skills.pediatry_wiki.scripts.prompts_minimal_judge import (
 
 @needs_conversation
 @needs_documents
+@needs_request_id
 def generate_wiki_response(
     user_query: str,
     conversation: ConversationHistory,
     documents: dict[str, str] | None = None,
+    request_id: str | None = None,
 ) -> tuple[str, bool]:
     return generate_wiki_response_with_prompt(
         user_query,
@@ -49,4 +51,5 @@ def generate_wiki_response(
         # (or if) it ever finishes — this is prod, not eval, but nothing here
         # reaches the chat response itself.
         echo_stream=True,
+        request_id=request_id,
     )

providers/hf.py

@@ -95,14 +95,21 @@ class HFChatProvider:
         enable_thinking: bool | None = None,
         stream: bool = False,
         echo_stream: bool = False,
+        request_id: str | None = None,
     ) -> ChatCompletion:
         """stream=True avoids a gateway 504 on long (e.g. high-reasoning-effort)
         generations — the router times out a single blocking request but not a
         continuously-flowing streamed one. The stream is fully consumed and
         reassembled into the same shape a non-streaming call returns; callers
         never see a difference beyond avoiding the timeout, unless echo_stream
-        is also set — then reasoning/content tokens print to stdout live, for
-        an interactive caller that would otherwise wait in silence."""
+        is also set — then reasoning/content tokens are logged live, for an
+        interactive caller that would otherwise wait in silence.
+
+        `request_id` tags every log line this call emits, including ones from
+        a timed-out attempt whose thread the caller gave up on but which keeps
+        running in the background (HTTP requests can't be cancelled mid-flight)
+        — without it, that abandoned call's later output is indistinguishable
+        from a fresh, legitimate one sharing the same stdout/log stream."""
         kwargs: dict = {
             "model": model_id,
             "messages": [_msg_to_dict(m) for m in messages],
@@ -139,7 +146,7 @@ class HFChatProvider:
             try:
                 completion = self._create_with_transient_retry(
                     kwargs, rate_limit_info, server_error_info,
-                    stream=stream, echo_stream=echo_stream,
+                    stream=stream, echo_stream=echo_stream, request_id=request_id,
                 )
             except HfHubHTTPError as e:
                 status = e.response.status_code
@@ -157,7 +164,8 @@ class HFChatProvider:
                     server_error_info["wait_seconds"] += wait
                     server_error_info["retries"] += 1
                     logger.warning(
-                        "%d (attempt %d/%d): %s — retrying in %.1fs",
+                        "[%s] %d (attempt %d/%d): %s — retrying in %.1fs",
+                        request_id,
                         status,
                         attempt,
                         self._max_attempts,
@@ -170,7 +178,7 @@ class HFChatProvider:
                     raise
                 last_exc = e
                 logger.warning(
-                    "400 (attempt %d/%d): %s", attempt, self._max_attempts, e
+                    "[%s] 400 (attempt %d/%d): %s", request_id, attempt, self._max_attempts, e
                 )
                 try:
                     logger.warning("  raw response body: %s", e.response.text)
@@ -181,14 +189,14 @@ class HFChatProvider:
                 continue
 
             if completion is None:
-                logger.warning("empty completion (%d/%d)", attempt, self._max_attempts)
+                logger.warning("[%s] empty completion (%d/%d)", request_id, attempt, self._max_attempts)
                 continue
             if completion.choices[0]["finish_reason"] == "length":
-                logger.warning("truncated (%d/%d)", attempt, self._max_attempts)
+                logger.warning("[%s] truncated (%d/%d)", request_id, attempt, self._max_attempts)
                 continue
             msg = completion.choices[0]["message"]
             if not (msg.get("content") or "").strip() and not msg.get("tool_calls"):
-                logger.warning("empty content (%d/%d)", attempt, self._max_attempts)
+                logger.warning("[%s] empty content (%d/%d)", request_id, attempt, self._max_attempts)
                 continue
 
             result = _to_completion(completion)
@@ -199,7 +207,7 @@ class HFChatProvider:
             result.server_error_retries = server_error_info["retries"]
             return result
 
-        logger.error("All %d attempts exhausted", self._max_attempts)
+        logger.error("[%s] All %d attempts exhausted", request_id, self._max_attempts)
         # 429/502 raise directly above; only 400 and non-502 5xx land here.
         if isinstance(last_exc, HfHubHTTPError):
             if last_exc.response.status_code == 400:
@@ -214,7 +222,7 @@ class HFChatProvider:
 
     def _create_with_transient_retry(
         self, kwargs: dict, rate_limit_info: dict, server_error_info: dict,
-        stream: bool = False, echo_stream: bool = False,
+        stream: bool = False, echo_stream: bool = False, request_id: str | None = None,
     ):
         """.create(), but 429s and 502s sleep-and-retry internally (own
         backoff/budget each); other 5xx propagate for chat()'s old handling.
@@ -228,7 +236,8 @@ class HFChatProvider:
             try:
                 if stream:
                     return _accumulate_stream(
-                        self._client.chat.completions.create(**kwargs), echo=echo_stream
+                        self._client.chat.completions.create(**kwargs),
+                        echo=echo_stream, request_id=request_id,
                     )
                 return self._client.chat.completions.create(**kwargs)
             except HfHubHTTPError as e:
@@ -242,7 +251,8 @@ class HFChatProvider:
                         base_wait * (2 ** (rl_attempt - 1)), _RATE_LIMIT_MAX_WAIT_SECONDS
                     )
                     logger.warning(
-                        "rate limited (429) — retrying in %.2fs (%d/%d)",
+                        "[%s] rate limited (429) — retrying in %.2fs (%d/%d)",
+                        request_id,
                         wait,
                         rl_attempt,
                         _RATE_LIMIT_MAX_RETRIES,
@@ -265,8 +275,9 @@ class HFChatProvider:
                         remaining,
                     )
                     logger.warning(
-                        "502 — retrying in %.1fs (%.1fs/%.1fs of server-error "
+                        "[%s] 502 — retrying in %.1fs (%.1fs/%.1fs of server-error "
                         "budget used)",
+                        request_id,
                         wait,
                         server_error_info["wait_seconds"],
                         _SERVER_ERROR_MAX_WAIT_SECONDS,
@@ -287,13 +298,19 @@ class _StreamedCompletion:
         self.usage = usage
 
 
-def _accumulate_stream(chunks, echo: bool = False) -> _StreamedCompletion:
+def _accumulate_stream(
+    chunks, echo: bool = False, request_id: str | None = None
+) -> _StreamedCompletion:
     """Reassembles a stream of ChatCompletionStreamOutput chunks into the same
     shape a non-streaming response has, so downstream code (_to_completion,
     chat()'s own post-processing) doesn't need to know the call was streamed.
 
-    echo=True prints reasoning/content tokens to stdout as they arrive — for
-    an interactive caller (e.g. an eval script) that would otherwise wait in
+    echo=True logs one line via the logger announcing that reasoning/content
+    streaming has started — tagged with request_id, so a call whose caller
+    gave up waiting (but whose thread keeps running — see HFChatProvider.chat's
+    docstring) can still be told apart from a concurrent or later one sharing
+    the same log stream — then prints each token live to stdout as it arrives,
+    same as before, for an interactive caller that would otherwise wait in
     silence for a slow, high-reasoning-effort call to finish."""
     content_parts: list[str] = []
     reasoning_parts: list[str] = []
@@ -314,14 +331,16 @@ def _accumulate_stream(chunks, echo: bool = False) -> _StreamedCompletion:
         if delta["reasoning"]:
             if echo:
                 if not in_reasoning:
-                    print("\n--- reasoning (streaming) ---", flush=True)
+                    logger.info("[%s] streaming reasoning...", request_id)
                     in_reasoning = True
                 print(delta["reasoning"], end="", flush=True)
             reasoning_parts.append(delta["reasoning"])
         if delta["content"]:
             if echo:
                 if not in_content:
-                    print("\n--- content (streaming) ---", flush=True)
+                    if in_reasoning:
+                        print(flush=True)  # end the reasoning line first
+                    logger.info("[%s] streaming content...", request_id)
                     in_content = True
                 print(delta["content"], end="", flush=True)
             content_parts.append(delta["content"])
@@ -335,6 +354,8 @@ def _accumulate_stream(chunks, echo: bool = False) -> _StreamedCompletion:
                 entry["arguments"] += tc["function"]["arguments"]
         if choice["finish_reason"]:
             finish_reason = choice["finish_reason"]
+    if echo and (in_reasoning or in_content):
+        print(flush=True)  # close out the last open printed line
     message = {
         "content": "".join(content_parts) or None,
         "reasoning": "".join(reasoning_parts) or None,