Add running of single handlers (compared to whole actions) to WM-ER protocol and implement it

Author: Aksem

docs/wm-er-protocol.md

@@ -25,12 +25,17 @@ method names.
 3. WM sends `initialized`.
 4. WM sends `finecodeRunner/updateConfig` to bootstrap handlers and services.
    - ER processes it and returns `{}`.
-5. WM sends `finecodeRunner/resolveActionSources` to get canonical action source paths.
-   - ER returns a sparse map of `configSource → canonicalSource` for actions whose
-     declared config path differs from the fully qualified runtime path.
+5. WM sends `finecodeRunner/resolveActionMeta` to get action meta info.
+   - ER returns a complete map of `configSource → { canonical_source, runs_concurrently }` for every action
+     whose class can be imported in this env.  Actions that fail to import are omitted.
    - WM stores these on its `Action` domain objects before the runner is considered ready.
 6. On shutdown: WM sends `shutdown` then `exit`.
 
+> **Multi-env runs:** when an action's handlers span more than one env, the WM
+> orchestrates execution segment-by-segment using `actions/runHandlers` so that
+> the serialized run context (`previousResult`) crosses the wire only at actual
+> env boundaries. See [Multi-Env Action Orchestration](#multi-env-action-orchestration).
+
 ## Message Catalog
 
 ### WM -> ER
@@ -110,6 +115,58 @@ method names.
   - Note: `result_by_format` is a JSON-encoded string (not a nested object) —
     the WM decodes it with `json.loads` after receiving the response.
 
+- `actions/runHandlers`
+  - Runs a named subset of an action's handlers sequentially within this ER,
+    seeding `context.current_result` from a prior segment's serialized result.
+    Used by the WM to orchestrate multi-env action runs; not used for single-env
+    actions (those still use `actions/run`).
+  - Params:
+    - `actionName` (string): action name as registered via `finecodeRunner/updateConfig`
+    - `handlerNames` (list of string): ordered list of handler names to execute;
+      all must belong to this ER's env
+    - `previousResult` (object | null): serialized `RunActionResult`
+      (`dataclasses.asdict`) from the last handler of the preceding segment, or
+      `null` for the first segment. Reconstructed as `context.current_result`
+      before the first handler in `handlerNames` is invoked.
+    - `options` (object | null): same keys as `actions/run`. `resultFormats`
+      should be omitted (or `[]`) for intermediate segments and non-empty only
+      for the final segment of a run.
+  - Result (success):
+    ```json
+    {
+      "status": "success",
+      "result": {"<resultField>": "..."},
+      "resultByFormat": {"json": {"...": "..."}, "string": "..."},
+      "returnCode": 0
+    }
+    ```
+    - `result`: serialized `RunActionResult` after all specified handlers ran
+      (`dataclasses.asdict`). Pass as `previousResult` to the next segment's
+      `actions/runHandlers` call.
+    - `resultByFormat`: formatted results in the requested formats; `{}` when
+      `resultFormats` was empty in options.
+  - Result (streamed): used when `partialResultToken` was provided and all
+    results were delivered via `$/progress`. `result` is still populated for
+    context chaining.
+    ```json
+    {
+      "status": "streamed",
+      "result": {"<resultField>": "..."},
+      "resultByFormat": {},
+      "returnCode": 0
+    }
+    ```
+  - Result (stopped):
+    ```json
+    {
+      "status": "stopped",
+      "result": {"<resultField>": "..."},
+      "resultByFormat": {"json": {"...": "..."}},
+      "returnCode": 1
+    }
+    ```
+  - Result (error): `{"error": "message"}`
+
 - `actions/getPayloadSchemas`
   - Params: `{}`
   - Result: `{ action_name: JSON Schema fragment | null }`
@@ -120,33 +177,34 @@ method names.
 
 - `actions/mergeResults`
   - Params: `{ "actionName": string, "results": list }`
-  - Result: `{ "merged": ... }` or `{ "error": "..." }`
+  - `results`: list of serialized `RunActionResult` objects (`dataclasses.asdict`),
+    one per concurrent segment or handler. Used by the WM after a concurrent
+    multi-env run to merge the per-env results into a single final result.
+  - Result: `{ "merged": <serialized RunActionResult> }` or `{ "error": "..." }`
 
 - `actions/reload`
   - Params: `{ "actionName": string }`
   - Result: `{}`
 
-- `finecodeRunner/resolveActionSources`
+- `finecodeRunner/resolveActionMeta`
   - Params: `{}` (no params)
-  - Result: sparse map of `{ "<configSource>": "<canonicalSource>", ... }` for actions
-    whose declared config path differs from the fully qualified runtime path.
-    Only entries where the two differ are included.
-    Example: `{ "myext.LintAction": "myext.actions.lint.LintAction" }`
+  - Result: complete map of `{ "<configSource>": { "canonical_source": string, "runs_concurrently": bool }, ... }` for every
+    action whose class can be imported in this env.  Actions that fail to import are
+    omitted entirely.
+    Example: `{ "myext.LintAction": { "canonical_source": "myext.actions.lint.LintAction", "runs_concurrently": true } }`
   - Called by the WM after `finecodeRunner/updateConfig` completes to store canonical
-    sources on its `Action` domain objects before the runner is considered ready.
-    The WM uses `canonical_source` as the primary identifier in all subsequent action
-    lookups; `source` (from config) is the fallback for actions whose class could not
-    be imported in this env.
-  - Actions where `cls.__module__ + "." + cls.__qualname__ == config source` are
-    omitted (no mapping needed — the config source is already canonical).
+    sources and execution modes on its `Action` domain objects before the runner is
+    considered ready.  The WM uses `canonical_source` as the primary identifier in all
+    subsequent action lookups.  Actions absent from the response (import failure) keep
+    `canonical_source = None` until another runner for the same project resolves them.
 
 - `actions/resolveSource`
   - Params: `{ "source": string }` — an arbitrary import-path alias to resolve.
   - Result: `{ "canonicalSource": string }` — the fully qualified class path
     (`cls.__module__ + "." + cls.__qualname__`).
   - Raises a JSON-RPC error if the alias cannot be imported or resolved.
   - Used during action lookup when a caller provides an alias not already known
-    from `finecodeRunner/resolveActionSources` (full ADR-0019 support).
+    from `finecodeRunner/resolveActionMeta` (full ADR-0019 support).
 
 - `packages/resolvePath`
   - Params: `{ "packageName": string }`
@@ -180,7 +238,7 @@ method names.
       `f"{cls.__module__}.{cls.__qualname__}"` (e.g. `"myext.actions.lint.LintAction"`).
       Must not be a re-exported alias such as `"myext.LintAction"`. The WM resolves
       the action name by matching against the canonical source reported by
-      `finecodeRunner/resolveActionSources`; a re-exported path will not match and the
+      `finecodeRunner/resolveActionMeta`; a re-exported path will not match and the
       request will fail.
     - `payload` (object): serialized action payload (`dataclasses.asdict`)
     - `meta` (object): `{ "trigger": string, "devEnv": string, "orchestrationDepth": int }`
@@ -202,11 +260,76 @@ method names.
 
 - `$/progress`
   - Params: `{ "token": <token>, "value": "<stringified JSON partial result>" }`
-  - The `token` must match `partial_result_token` from `actions/run`.
+  - The `token` must match `partialResultToken` from `actions/run` or
+    `actions/runHandlers`.
   - `value` is a JSON string produced by the ER from a partial run result.
-  - When `$/progress` is used to deliver results, the final `actions/run` response
-    must have `status: "streamed"` and empty `result_by_format`. See `actions/run`
-    result (streamed) above.
+  - When `$/progress` is used to deliver results, the final `actions/run` or
+    `actions/runHandlers` response must have `status: "streamed"` and empty
+    `result_by_format`. See result (streamed) entries above.
+
+## Multi-Env Action Orchestration
+
+When an action's handlers span more than one env, the WM cannot delegate the
+whole run to a single ER via `actions/run`. Instead the WM becomes the
+orchestrator and drives execution using `actions/runHandlers`.
+
+### Sequential mode (default)
+
+The WM groups the action's handlers into **consecutive same-env segments**:
+
+```text
+handlers:  [h1/env1, h2/env1, h3/env1, h4/env2]
+segments:  [(env1, [h1, h2, h3]), (env2, [h4])]
+
+handlers:  [h1/env1, h2/env2, h3/env1]
+segments:  [(env1, [h1]), (env2, [h2]), (env1, [h3])]
+```
+
+Execution:
+
+1. WM calls `actions/runHandlers` for segment 1 with `previousResult: null`.
+2. For each subsequent segment, WM calls `actions/runHandlers` on that segment's
+   ER with `previousResult` set to the `result` returned by the previous call.
+   The ER reconstructs this as `context.current_result` before the first handler
+   in the segment runs.
+3. If any call returns `status: "stopped"`, WM stops the chain and returns that
+   result to the caller.
+4. `resultFormats` is passed only in the final segment's options — earlier
+   segments return `resultByFormat: {}` to avoid unnecessary serialization.
+5. WM assembles the final response from the last segment's `result` and
+   `resultByFormat`.
+
+### Concurrent mode
+
+The WM groups handlers by env (order within an env does not matter for
+concurrent execution):
+
+```text
+handlers:  [h1/env1, h2/env2, h3/env1]
+groups:    [(env1, [h1, h3]), (env2, [h2])]
+```
+
+Execution:
+
+1. WM dispatches `actions/runHandlers` to all env groups in parallel, all with
+   `previousResult: null`.
+2. WM collects all `result` objects from the parallel calls.
+3. WM calls `actions/mergeResults` on any available ER for the action, passing
+   the collected `result` objects.
+4. The merged result and its formatted representation form the final response.
+
+### Single-env actions
+
+When all handlers are in the same env, the WM uses `actions/run` — a single
+delegated call where the ER manages handler sequencing internally.
+`actions/runHandlers` is only used when handlers span multiple envs.
+
+### `walRunId` continuity
+
+The WM generates a single `walRunId` for the whole logical action run and
+passes it in every `actions/runHandlers` call's options. Each ER emits WAL
+events tagged with that ID for the handler(s) it executes, so traces can be
+correlated across envs for the same logical run.
 
 ## Error Handling and Cancellation
 

finecode_extension_runner/src/finecode_extension_runner/_services/run_action.py

@@ -162,6 +162,7 @@ async def run_action(
     run_id: int | None = None,
     partial_result_queue: asyncio.Queue | None = None,
     caller_kwargs: code_action.CallerRunContextKwargs | None = None,
+    initial_result: code_action.RunActionResult | None = None,
 ) -> code_action.RunActionResult | None:
     # design decisions:
     # - keep payload unchanged between all subaction runs.
@@ -259,7 +260,9 @@ async def run_action(
         # TODO: check run_context below, whether AsyncPlaceholder can really be used
         run_context = AsyncPlaceholderContext()
 
-    action_result: code_action.RunActionResult | None = None
+    action_result: code_action.RunActionResult | None = initial_result
+    if initial_result is not None:
+        run_context_info.update(initial_result)
     runner_context = global_state.runner_context
 
     # to be able to catch source of exceptions in user-accessible code more precisely,
@@ -518,6 +521,15 @@ async def run_action(
                                 action_result.update(handler_result)
 
                             run_context_info.update(action_result)
+
+                # Surface results sent via run_context.partial_result_sender.send()
+                # (accumulated in accumulating_sender) — these are not captured by
+                # handler return values but must contribute to the final result.
+                if accumulating_sender is not None and accumulating_sender.has_sent:
+                    if action_result is None:
+                        action_result = accumulating_sender.accumulated
+                    elif accumulating_sender.accumulated is not None:
+                        action_result.update(accumulating_sender.accumulated)
     finally:
         # exit run context
         try:
@@ -673,6 +685,111 @@ def action_result_to_run_action_response(
     )
 
 
+async def run_handlers_raw(
+    request: schemas.RunHandlersRequest, options: schemas.RunActionOptions
+) -> schemas.RunHandlersResponse:
+    """Execute a named subset of an action's handlers for multi-env orchestration.
+
+    Seeds context.current_result from request.previous_result before the first
+    handler runs, so sequential handlers across env boundaries see a continuous
+    result chain.  Returns both the raw serialized result (for chaining) and
+    formatted output (populated only when options.result_formats is non-empty).
+    """
+    global last_run_id
+    run_id = last_run_id
+    last_run_id += 1
+
+    if global_state.runner_context is None:
+        raise ActionFailedException(
+            "run_handlers called before extension runner is initialized"
+        )
+
+    project_def = global_state.runner_context.project
+
+    try:
+        action = project_def.actions[request.action_name]
+    except KeyError as exc:
+        raise ActionFailedException(
+            f"R{run_id} | Action {request.action_name} not found"
+        ) from exc
+
+    try:
+        action_cache = global_state.runner_context.action_cache_by_name[request.action_name]
+    except KeyError:
+        action_cache = domain.ActionCache()
+        global_state.runner_context.action_cache_by_name[request.action_name] = action_cache
+
+    if action_cache.exec_info is None:
+        action_cache.exec_info = create_action_exec_info(action)
+    action_exec_info = action_cache.exec_info
+
+    # Build a filtered ActionDeclaration containing only the requested handlers,
+    # in the order specified by handler_names (preserves WM segment ordering).
+    handler_names_ordered = request.handler_names
+    handlers_by_name = {h.name: h for h in action.handlers}
+    filtered_handlers = [
+        handlers_by_name[name]
+        for name in handler_names_ordered
+        if name in handlers_by_name
+    ]
+    filtered_action = domain.ActionDeclaration(
+        name=action.name,
+        config=action.config,
+        handlers=filtered_handlers,
+        source=action.source,
+    )
+
+    # Reconstruct the previous segment's result so handlers see a continuous
+    # context.current_result across the env boundary.
+    initial_result: code_action.RunActionResult | None = None
+    if request.previous_result is not None and action_exec_info.result_type is not None:
+        try:
+            initial_result = action_exec_info.result_type(**request.previous_result)
+        except Exception as exc:
+            logger.warning(
+                f"R{run_id} | Could not reconstruct previous_result for "
+                f"{request.action_name}: {exc}. Handlers will see no prior context."
+            )
+
+    # Build payload from params.
+    payload: code_action.RunActionPayload | None = None
+    if action_exec_info.payload_type is not None and request.params:
+        payload_type_with_validation = pydantic_dataclass(action_exec_info.payload_type)
+        payload = typing.cast(
+            code_action.RunActionPayload,
+            payload_type_with_validation(**request.params),
+        )
+
+    wal_run_id = getattr(options, "wal_run_id", None)
+    if not isinstance(wal_run_id, str) or wal_run_id.strip() == "":
+        raise ActionFailedException("Missing required wal_run_id in run options")
+
+    meta = dataclasses.replace(options.meta, wal_run_id=wal_run_id)
+
+    action_result = await run_action(
+        action_def=filtered_action,
+        payload=payload,
+        meta=meta,
+        partial_result_token=options.partial_result_token,
+        progress_token=options.progress_token,
+        run_id=run_id,
+        initial_result=initial_result,
+    )
+
+    # Raw serialized result for chaining to the next segment.
+    raw_result: dict = dataclasses.asdict(action_result) if action_result is not None else {}
+
+    # Formatted result — only populated when the caller requests formats.
+    formatted = action_result_to_run_action_response(action_result, options.result_formats)
+    result_by_format: dict = formatted.result_by_format or {}
+
+    return schemas.RunHandlersResponse(
+        return_code=formatted.return_code,
+        result=raw_result,
+        result_by_format=result_by_format,
+    )
+
+
 def create_action_exec_info(action: domain.ActionDeclaration) -> domain.ActionExecInfo:
     try:
         action_type_def = run_utils.import_module_member_by_source_str(action.source)