Replace elicitation origin ContextVar with an explicit parameter

The ContextVar-based originating_client() let the connection an ER
should ask questions of leak into ambient state, which is invisible
at call sites and silently defaults to "nobody" if a hop forgets to
set it — exactly the failure the two progressToken handlers in
_streaming.py had (a connected client whose run could never be
asked). Passing RunDispatchOrigin explicitly, with no default at any
hop, turns a dropped connection into a signature the type checker
can catch instead of a question nobody receives (ADR-0082).

- elicitation_bridge.py: RunDispatchOrigin(connection) replaces the
  originating_client() context manager; bind_run() now takes it as a
  required argument instead of reading the ContextVar.
- er_dispatch.py, in_flight_runs.py, proxy_utils.py,
  matrix_streaming.py, project_executor.py, execution_scopes.py,
  workspace_executor.py, partial_results_service.py: thread `origin`
  explicitly from each request handler down through every dispatch
  seam to the choke point (in_flight_runs.track) that binds it.
- _actions.py, _streaming.py, cli_app/utils.py,
  prepare_envs_service.py: each call site states its origin
  explicitly, including the plain request/response and internal
  dispatch paths that have none.
- developing-finecode.md: documents the "ambient state" rule this
  refactor follows (when a ContextVar is justified vs. when a
  parameter is required with no default), plus the boolean-flags,
  comments, knowledge-package-split and local-observability-stack
  sections accumulated in this file since the last commit touching
  it.

Author: Aksem

docs/guides/developing-finecode.md

@@ -124,6 +124,9 @@ async def run_actions_in_projects_and_concat_results(
         run_trigger=run_trigger,
         dev_env=dev_env,
         payload_overrides_by_project=payload_overrides_by_project or {},
+        # The CLI drives the WM in-process; there is no client connection an ER
+        # could be pointed at.
+        origin=None,
     )
 
     result_output: str = ""

src/finecode/cli_app/utils.py

@@ -75,6 +75,10 @@ async def _handle_run_action(
                 result_formats=parsed.result_formats,
                 initialize_all_handlers=True,
                 selected_interpreters=selected_interpreters,
+                # Plain request/response: this handler never receives the
+                # caller's writer, so there is no connection to put a question
+                # to. Elicitation is available on the streamed paths only.
+                origin=None,
             )
             return {
                 "resultByFormat": result.result_by_format,
@@ -240,6 +244,8 @@ async def _handle_run_batch(
             concurrently=parsed.concurrently,
             result_formats=parsed.result_formats,
             payload_overrides_by_project=parsed.params_by_project,
+            # No writer here either — see `_handle_run_action` above.
+            origin=None,
         )
 
         results, overall_return_code = _build_batch_result(

src/finecode/wm_server/_api_handlers/_actions.py

@@ -123,96 +123,97 @@ async def _handle_run_action_with_partial_results(
 
         # From here to the final result, this connection is the origin of the
         # run: an ER that asks a question mid-run is asking the client that
-        # started it and nobody else (ADR-0082 rule 1). Set on the streamed
-        # paths, which are the ones that still hold their caller's connection;
-        # the dispatch below binds it to the run id it mints, which is what the
-        # ER names when it asks.
-        with elicitation_bridge.originating_client(writer):
-            stream = await partial_results_service.run_action_with_partial_results(
-                action_name=action_name,
-                project_path=project_path,
-                params=params.get("params", {}),
-                partial_result_token=token,
-                run_trigger=trigger,
-                dev_env=dev_env,
-                ws_context=ws_context,
-                result_formats=result_formats,
-                progress_token=progress_token,
-                selected_interpreters=selected_interpreters,
-            )
-
-            # Opt-in (collect-style callers like MCP): accumulate the `json` format of
-            # each partial per project so it can be type-safely merged into the response.
-            merge_results_enabled = options.get("mergeResults", False)
-            json_by_project: dict[str, list[dict]] = {}
-
-            async def _forward_partials() -> int:
-                count = 0
-                async for value in stream:
-                    count += 1
-                    if merge_results_enabled and isinstance(value, dict):
-                        project_str = value.get("project", "")
-                        result_by_format = value.get("resultByFormat") or {}
-                        json_by_project.setdefault(project_str, []).append(
-                            result_by_format.get("json")
-                        )
-                    logger.trace(
-                        f"run+partialResults: sending partial #{count} for token={token}, keys={list(value.keys()) if isinstance(value, dict) else type(value)}"
-                    )
-                    _notify_client(
-                        writer,
-                        "actions/partialResult",
-                        {"token": token, "value": value},
+        # started it and nobody else (ADR-0082 rule 1). Constructed on the
+        # streamed paths, which are the ones that still hold their caller's
+        # connection; the dispatch below binds it to the run id it mints,
+        # which is what the ER names when it asks.
+        origin = elicitation_bridge.RunDispatchOrigin(connection=writer)
+        stream = await partial_results_service.run_action_with_partial_results(
+            action_name=action_name,
+            project_path=project_path,
+            params=params.get("params", {}),
+            partial_result_token=token,
+            run_trigger=trigger,
+            dev_env=dev_env,
+            ws_context=ws_context,
+            result_formats=result_formats,
+            progress_token=progress_token,
+            selected_interpreters=selected_interpreters,
+            origin=origin,
+        )
+
+        # Opt-in (collect-style callers like MCP): accumulate the `json` format of
+        # each partial per project so it can be type-safely merged into the response.
+        merge_results_enabled = options.get("mergeResults", False)
+        json_by_project: dict[str, list[dict]] = {}
+
+        async def _forward_partials() -> int:
+            count = 0
+            async for value in stream:
+                count += 1
+                if merge_results_enabled and isinstance(value, dict):
+                    project_str = value.get("project", "")
+                    result_by_format = value.get("resultByFormat") or {}
+                    json_by_project.setdefault(project_str, []).append(
+                        result_by_format.get("json")
                     )
-                    await writer.drain()
-                return count
+                logger.trace(
+                    f"run+partialResults: sending partial #{count} for token={token}, keys={list(value.keys()) if isinstance(value, dict) else type(value)}"
+                )
+                _notify_client(
+                    writer,
+                    "actions/partialResult",
+                    {"token": token, "value": value},
+                )
+                await writer.drain()
+            return count
 
-            async def _forward_progress() -> None:
-                if stream.progress_stream is None or progress_token is None:
-                    return
-                async for value in stream.progress_stream:
-                    logger.trace(
-                        f"run+partialResults: sending progress type={value.get('type')} for token={progress_token}"
-                    )
-                    _notify_client(
-                        writer,
-                        "actions/progress",
-                        {"token": progress_token, "value": value},
-                    )
-                    await writer.drain()
+        async def _forward_progress() -> None:
+            if stream.progress_stream is None or progress_token is None:
+                return
+            async for value in stream.progress_stream:
+                logger.trace(
+                    f"run+partialResults: sending progress type={value.get('type')} for token={progress_token}"
+                )
+                _notify_client(
+                    writer,
+                    "actions/progress",
+                    {"token": progress_token, "value": value},
+                )
+                await writer.drain()
 
-            partial_count = 0
-            async with asyncio.TaskGroup() as forward_tg:
-                partials_task = forward_tg.create_task(_forward_partials())
-                forward_tg.create_task(_forward_progress())
-            partial_count = partials_task.result()
-
-            final = await stream.final_result()
-
-            if merge_results_enabled and json_by_project:
-                return_code = final.get("returnCode", 0) if isinstance(final, dict) else 0
-                results: dict[str, dict] = {}
-                for project_str, payloads in json_by_project.items():
-                    merged_json = await merge_partial_results_for_action(
-                        project_path=pathlib.Path(project_str),
-                        action_name=action_name,
-                        json_payloads=payloads,
-                        ws_context=ws_context,
-                    )
-                    if merged_json is not None:
-                        results[project_str] = {
-                            action_source: {
-                                "resultByFormat": {"json": merged_json},
-                                "returnCode": return_code,
-                            }
+        partial_count = 0
+        async with asyncio.TaskGroup() as forward_tg:
+            partials_task = forward_tg.create_task(_forward_partials())
+            forward_tg.create_task(_forward_progress())
+        partial_count = partials_task.result()
+
+        final = await stream.final_result()
+
+        if merge_results_enabled and json_by_project:
+            return_code = final.get("returnCode", 0) if isinstance(final, dict) else 0
+            results: dict[str, dict] = {}
+            for project_str, payloads in json_by_project.items():
+                merged_json = await merge_partial_results_for_action(
+                    project_path=pathlib.Path(project_str),
+                    action_name=action_name,
+                    json_payloads=payloads,
+                    ws_context=ws_context,
+                )
+                if merged_json is not None:
+                    results[project_str] = {
+                        action_source: {
+                            "resultByFormat": {"json": merged_json},
+                            "returnCode": return_code,
                         }
-                if results:
-                    final = {**final, "results": results}
+                    }
+            if results:
+                final = {**final, "results": results}
 
-            logger.trace(
-                f"run+partialResults: done, sent {partial_count} partials, final keys={list(final.keys()) if isinstance(final, dict) else type(final)}"
-            )
-            return final
+        logger.trace(
+            f"run+partialResults: done, sent {partial_count} partials, final keys={list(final.keys()) if isinstance(final, dict) else type(final)}"
+        )
+        return final
 
 
 async def _handle_run_action_with_partial_results_task(
@@ -301,13 +302,11 @@ async def _handle_run_batch_with_partial_results(
 
     params = params or {}
     # The connection that asked for the batch is the origin of every run in it,
-    # including the ones the per-project tasks below dispatch: a task copies the
-    # context it is created in, so setting this before they exist is what makes
-    # it reach them (ADR-0082 rule 1).
-    with (
-        telemetry.attach_incoming_traceparent(params),
-        elicitation_bridge.originating_client(writer),
-    ):
+    # including the ones the per-project tasks below dispatch: `origin` is
+    # captured in `_stream_action`'s closure, so a task created from it carries
+    # the same descriptor to every dispatch it makes (ADR-0082 rule 1).
+    origin = elicitation_bridge.RunDispatchOrigin(connection=writer)
+    with telemetry.attach_incoming_traceparent(params):
         parsed = _parse_run_batch_params(params)
         token = params["partialResultToken"]
 
@@ -438,6 +437,7 @@ async def _on_partial(
                     merge_results=parsed.merge_results,
                     on_partial=_on_partial,
                     selected_interpreters=selected_interpreters,
+                    origin=origin,
                 )
                 if parsed.merge_results:
                     merged_results.setdefault(str(project_path), {})[action_source] = {
@@ -461,6 +461,7 @@ async def _on_partial(
                 ws_context=ws_context,
                 initialize_all_handlers=True,
                 result_formats=parsed.result_formats,
+                origin=origin,
             ) as ctx:
                 async for value in ctx:
                     partial_count += 1
@@ -727,6 +728,10 @@ async def _forward_progress() -> None:
                 result_formats=parsed.result_formats,
                 initialize_all_handlers=True,
                 progress_token=progress_token,
+                # Progress is forwarded to this connection for the whole run, so
+                # it is just as much the run's origin as on the partial-results
+                # paths: an ER that elicits mid-run has a client to ask.
+                origin=elicitation_bridge.RunDispatchOrigin(connection=writer),
             )
             return {
                 "resultByFormat": result.result_by_format,
@@ -906,6 +911,9 @@ async def _forward_to_client() -> None:
                 result_formats=parsed.result_formats,
                 payload_overrides_by_project=parsed.params_by_project or None,
                 progress_token_by_project=progress_token_by_project,
+                # Aggregated progress goes to this connection until the batch
+                # ends, so it is the origin of every run in the batch.
+                origin=elicitation_bridge.RunDispatchOrigin(connection=writer),
             )
         finally:
             # Cancel get_progress tasks first, then drain slot lists through aggregator,

src/finecode/wm_server/_api_handlers/_streaming.py

@@ -28,34 +28,50 @@
 implementation notes park exactly this until such an identifier exists — it does
 now, independently of the WAL, since ADR-0079 keys in-flight runs by it).
 
-The two halves are separate on purpose: :func:`originating_client` marks *this
-task* as belonging to a connection, and :func:`bind_run` records the run id the
-dispatch minted under that connection. The first is a
-:class:`contextvars.ContextVar`, so it reaches the dispatch through call chains
-and into the tasks it spawns without every layer in between having to carry it;
-the second is what turns it into a lookup an ER on another connection can
-resolve.
+:class:`RunDispatchOrigin` carries the originating connection down to
+:func:`bind_run`, the one choke point every dispatch passes through
+(``in_flight_runs.track``): the request handler that still holds its caller's
+connection constructs it, threads it explicitly through the dispatch, and
+:func:`bind_run` records the run id the dispatch minted under it. Passed
+explicitly rather than read from ambient state, per "Ambient state: when a
+``ContextVar`` is allowed" in developing-finecode.md — a value many
+intermediate signatures carry is a cost, not a structural blocker.
 """
 
 from __future__ import annotations
 
 import collections.abc
 import contextlib
-import contextvars
+import dataclasses
 import typing
 
 __all__ = [
     "ElicitationBridge",
+    "RunDispatchOrigin",
     "bind_run",
     "handlers",
     "install",
-    "originating_client",
     "originating_client_for_run",
     "reset",
     "reset_origins",
 ]
 
 
+@dataclasses.dataclass(frozen=True, slots=True)
+class RunDispatchOrigin:
+    """Which client connection, if any, is asking for a run.
+
+    Constructed at the request handler that still holds its caller's
+    connection — or, for a nested ER→WM→ER dispatch, derived from the calling
+    run's connection via :func:`originating_client_for_run` — and threaded
+    explicitly down to :func:`bind_run`. ``connection=None`` is the honest
+    answer for anything the WM started on its own behalf, or a dispatch that
+    never had a client to begin with, and binds nothing.
+    """
+
+    connection: object | None
+
+
 class ElicitationBridge(typing.Protocol):
     """What the runner needs from whoever owns client connections."""
 
@@ -113,49 +129,33 @@ def handlers() -> ElicitationBridge | None:
 # Addressing: which connection started a given run
 # ---------------------------------------------------------------------------
 
-# The connection whose request this task is executing. A ContextVar rather than
-# a parameter because the dispatch that mints a run id sits many layers below
-# the handler that knows the connection, and every layer in between would
-# otherwise have to carry something it has no use for. Tasks copy the context
-# they are created in, so a fan-out inherits it without being told.
-_origin: contextvars.ContextVar[object | None] = contextvars.ContextVar(
-    "finecode_elicitation_origin", default=None
-)
-
 # Run id → the connection that started it. Written for the life of the dispatch
 # and read by an ER on a different connection entirely, which is why this is a
-# registry and not just the ContextVar above.
+# registry rather than a value carried on the run's own call stack.
 _runs: dict[str, object] = {}
 
 
 @contextlib.contextmanager
-def originating_client(connection: object | None) -> collections.abc.Iterator[None]:
-    """Mark this task, and what it starts, as running for *connection*.
-
-    Entered by the request handlers that still hold their caller's connection,
-    and again by nested dispatch on behalf of the run that asked for it. Passing
-    ``None`` is meaningful: it states that the work in the block has no
-    identifiable origin, which is the honest answer for anything the WM started
-    on its own behalf.
-    """
-    token = _origin.set(connection)
-    try:
-        yield
-    finally:
-        _origin.reset(token)
-
-
-@contextlib.contextmanager
-def bind_run(run_id: str) -> collections.abc.Iterator[None]:
-    """Record *run_id* as belonging to the connection this task is running for.
+def bind_run(
+    run_id: str, origin: RunDispatchOrigin | None
+) -> collections.abc.Iterator[None]:
+    """Record *run_id* as belonging to *origin*'s connection.
 
     Entered where the run identifier is minted, so the binding lasts exactly as
     long as the run does: a question can only be addressed while the run that
     asks it is in flight, and an entry outliving its run would address a later
-    question to a client that has moved on. A dispatch with no originating
-    connection records nothing rather than an empty entry.
+    question to a client that has moved on. ``origin=None``, or an origin whose
+    ``connection`` is ``None``, records nothing rather than an empty entry —
+    that is the honest state for a dispatch with no identifiable origin.
+
+    *origin* has no default, here and at every hop that forwards it. Passing an
+    explicit ``None`` is cheap and states a real fact; a default would let the
+    argument be dropped silently at one hop of a long chain, and the result —
+    a connected client that is never asked — looks exactly like a run that
+    genuinely had nobody to ask. That is the failure this parameter replaced a
+    ``ContextVar`` to avoid, so it must not be reintroduced as a default.
     """
-    connection = _origin.get()
+    connection = origin.connection if origin is not None else None
     if connection is None:
         yield
         return