Fix problems:

- handle progress correctly if total <= 0
- explicit 'streamed' status in run action result
- use file lock on wm port file to avoid start of multiple WM servers at the same time
- add missing WAL run id

Author: Aksem

docs/wm-er-protocol.md

@@ -88,6 +88,18 @@ The protocol is LSP-shaped with a small set of custom commands.
         "return_code": 0
       }
       ```
+    - Result (streamed): used when `partial_result_token` was provided and all
+      results were delivered via `$/progress` notifications. Following LSP convention,
+      the final response is an explicit completion signal — `result_by_format` is
+      intentionally empty. The WM treats this as a valid completion; an empty
+      `result_by_format` with any other status is a protocol error.
+      ```json
+      {
+        "status": "streamed",
+        "result_by_format": "{}",
+        "return_code": 0
+      }
+      ```
     - Result (stopped):
       ```json
       {
@@ -169,6 +181,9 @@ The protocol is LSP-shaped with a small set of custom commands.
   - Params: `{ "token": <token>, "value": "<stringified JSON partial result>" }`
   - The `token` must match `partial_result_token` from `actions/run`.
   - `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.
 
 ## Error Handling and Cancellation
 

finecode_extension_api/src/finecode_extension_api/code_action.py

@@ -191,7 +191,12 @@ def __init__(
         self._completed = 0
 
     async def __aenter__(self) -> ProgressContext:
-        percentage = 0 if self._total is not None else None
+        if self._total is None:
+            percentage = None
+        elif self._total <= 0:
+            percentage = 100
+        else:
+            percentage = 0
         await self._sender.begin(
             self._title,
             percentage=percentage,
@@ -209,7 +214,10 @@ async def advance(self, steps: int = 1, message: str | None = None) -> None:
         self._completed += steps
         percentage = None
         if self._total is not None:
-            percentage = min(int(self._completed / self._total * 100), 100)
+            if self._total <= 0:
+                percentage = 100
+            else:
+                percentage = min(int(self._completed / self._total * 100), 100)
         await self._sender.report(message=message, percentage=percentage)
 
     async def report(

finecode_extension_runner/src/finecode_extension_runner/lsp_server.py

@@ -621,6 +621,8 @@ async def run_action(
     # custom json encoder converts dict values and `convert_path_keys` is used to
     # convert dict keys
     result_by_format = response.to_dict()["result_by_format"]
+    if not result_by_format and options_schema.partial_result_token is not None:
+        status = "streamed"
     converted_result_by_format = {
         fmt: convert_path_keys(result) if isinstance(result, dict) else result
         for fmt, result in result_by_format.items()

src/finecode/wm_server/runner/runner_client.py

@@ -63,6 +63,7 @@ class ExtensionRunnerInfo(domain.ExtensionRunner):
 class RunActionResponse:
     result_by_format: dict[str, RunActionRawResult]
     return_code: int
+    status: str = "success"
 
     def json(self) -> dict[str, Any]:
         result = self.result_by_format.get("json")
@@ -142,10 +143,12 @@ async def run_action(
     except json.JSONDecodeError as exception:
         raise ActionRunFailed(f"Failed to decode result json: {exception}") from exception
 
-    if command_result["status"] == "stopped":
+    status = command_result["status"]
+
+    if status == "stopped":
         raise ActionRunStopped(message=result_by_format)
 
-    return RunActionResponse(result_by_format=result_by_format, return_code=return_code)
+    return RunActionResponse(result_by_format=result_by_format, return_code=return_code, status=status)
 
 
 async def merge_results(

src/finecode/wm_server/services/partial_results_service.py

@@ -252,6 +252,15 @@ async def _forward_progress() -> None:
 
         # Responses collected by the context manager from runner tasks
         for resp in ctx.responses:
+            if resp.status == "streamed":
+                return_codes.append(resp.return_code)
+                continue
+            if not resp.result_by_format:
+                raise runner_client.ActionRunFailed(
+                    f"ER returned empty result with status '{resp.status}' for project={project.name}; "
+                    "expected 'streamed' status when result_by_format is empty"
+                )
+
             json_result = resp.json()
             logger.trace(f"partial_results: final result for project={project.name}: return_code={resp.return_code}, keys={list(json_result.keys()) if isinstance(json_result, dict) else type(json_result)}")
             final_results.append(json_result)

src/finecode/wm_server/services/run_service/proxy_utils.py

@@ -174,11 +174,13 @@ async def run_action_and_notify(
     runner: runner_client.ExtensionRunnerInfo,
     run_trigger: runner_client.RunActionTrigger,
     dev_env: runner_client.DevEnv,
+    wal_run_id: str,
     result_formats: list[runner_client.RunResultFormat] | None = None,
     progress_token: int | str | None = None,
 ) -> runner_client.RunActionResponse:
     options: dict[str, typing.Any] = {
         "partial_result_token": partial_result_token,
+        "wal_run_id": wal_run_id,
         "meta": {"trigger": run_trigger.value, "dev_env": dev_env.value},
     }
     if progress_token is not None:
@@ -265,6 +267,7 @@ async def run_with_partial_results(
     progress_token: int | str | None = None,
 ) -> collections.abc.AsyncIterator[RunWithPartialResultsContext]:
     logger.trace(f"Run {action_name} in project {project_dir_path}")
+    wal_run_id = wal.new_wal_run_id()
 
     result: AsyncList[domain.PartialResultRawValue] = AsyncList()
     progress_result: AsyncList[domain.ProgressRawValue] | None = None
@@ -321,6 +324,7 @@ async def run_with_partial_results(
                         runner=runner,
                         run_trigger=run_trigger,
                         dev_env=dev_env,
+                        wal_run_id=wal_run_id,
                         result_formats=result_formats,
                         progress_token=progress_token,
                     )

src/finecode/wm_server/wm_lifecycle.py

@@ -13,10 +13,15 @@
 import subprocess
 import sys
 import tempfile
+import time
 
+from filelock import FileLock
 from loguru import logger
 
 NO_CLIENT_TIMEOUT_SECONDS = 30
+STARTUP_LOCK_FILENAME = "wm_start.lock"
+STARTUP_READY_TIMEOUT_SECONDS = 10.0
+STARTUP_READY_POLL_INTERVAL_SECONDS = 0.1
 
 
 def _cache_dir() -> pathlib.Path:
@@ -62,19 +67,35 @@ def is_running() -> bool:
     return running_port() is not None
 
 
+def _startup_lock() -> FileLock:
+    """Serialize shared WM startup across processes."""
+    lock_path = _cache_dir() / STARTUP_LOCK_FILENAME
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    return FileLock(str(lock_path))
+
+
 def ensure_running(workdir: pathlib.Path, log_level: str = "INFO") -> None:
     """Start the WM server as a subprocess if not already running."""
-    if is_running():
-        return
-
-    python_cmd = sys.executable
-    logger.info(f"Starting FineCode WM server subprocess in {workdir}")
-    subprocess.Popen(
-        [python_cmd, "-m", "finecode", "start-wm-server", f"--log-level={log_level}"],
-        cwd=str(workdir),
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-    )
+    with _startup_lock():
+        if is_running():
+            return
+
+        python_cmd = sys.executable
+        logger.info(f"Starting FineCode WM server subprocess in {workdir}")
+        subprocess.Popen(
+            [python_cmd, "-m", "finecode", "start-wm-server", f"--log-level={log_level}"],
+            cwd=str(workdir),
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+
+        # Keep the lock until the spawned server is observable via discovery and
+        # TCP probe so competing callers do not start a duplicate process.
+        deadline = time.monotonic() + STARTUP_READY_TIMEOUT_SECONDS
+        while time.monotonic() < deadline:
+            if is_running():
+                return
+            time.sleep(STARTUP_READY_POLL_INTERVAL_SECONDS)
 
 
 async def wait_until_ready(timeout: float = 30) -> int: