refactor(execution): share node worker pool

docs/execution-graph.md

@@ -575,6 +575,7 @@ metis_engine:
               - index
               - memory
           private_policy:
+            max_concurrency: 50
             inputs:
               review: simple_llm_review
           finding_dedup:
@@ -592,7 +593,13 @@ metis_engine:
 
 A handler receives validated inputs and a small context containing its granted
 capabilities together with the repository lookup contract, CodeGraph
-materialize/load API, model runner, runtime limits, and callbacks. A node
+materialize/load API, model runner, runtime limits, shared job scheduler, and
+callbacks. `max_concurrency` limits that node's active jobs without reserving
+threads; omitted values use the global `metis_engine.max_workers` limit. All
+nodes on one engine share that worker pool, so their combined active work never
+exceeds the global limit; larger node values are capped by it. A node submits
+bounded work through `invocation.context.jobs.run(...)`; it does not construct
+its own executor. A node
 accesses only the names in `invocation.context.capabilities`. A node declares
 `request: ReviewCommand` when it needs the review-stage request; the runner
 supplies that typed stage input without extra YAML. The context does not expose

src/metis/configuration.py

@@ -71,7 +71,12 @@ def load_runtime_config(config_path=None, enable_psql=False):
 
     engine_cfg = cfg.get("metis_engine", {})
     runtime["max_token_length"] = engine_cfg.get("max_token_length", 100000)
-    runtime["max_workers"] = engine_cfg.get("max_workers", 8)
+    runtime["max_workers"] = _required_positive_int(
+        engine_cfg,
+        "max_workers",
+        section="metis_engine",
+        default=8,
+    )
     runtime["embed_dim"] = engine_cfg.get("embed_dim", 1536)
     runtime["doc_chunk_size"] = engine_cfg.get("doc_chunk_size", 1024)
     runtime["doc_chunk_overlap"] = engine_cfg.get("doc_chunk_overlap", 200)
@@ -195,8 +200,14 @@ def _positive_int(value: object, *, fallback: int) -> int:
     return parsed
 
 
-def _required_positive_int(values: dict[str, Any], key: str, *, section: str) -> int:
-    value = values.get(key)
+def _required_positive_int(
+    values: dict[str, Any],
+    key: str,
+    *,
+    section: str,
+    default: int | None = None,
+) -> int:
+    value = values.get(key, default)
     if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
         raise ValueError(f"{section}.{key} must be a positive integer")
     return value

src/metis/engine/capabilities/navigation.py

@@ -8,6 +8,7 @@
 import re
 import shutil
 import subprocess
+from threading import Lock
 from typing import Sequence
 
 from pydantic import BaseModel
@@ -43,6 +44,9 @@ def __init__(
         self.max_chars = max_chars
         self._has_grep = shutil.which("grep") is not None
         self._has_find = shutil.which("find") is not None
+        # Serialize subprocess tools; add a measured FD-aware pool if tool
+        # latency becomes material.
+        self._subprocess_lock = Lock()
 
     def _resolve_path(self, raw_path: str) -> Path:
         return resolve_path_within_root(self.codebase_path, raw_path)
@@ -53,13 +57,14 @@ def _run(
         *,
         ok_returncodes: tuple[int, ...] = (0,),
     ) -> str:
-        proc = subprocess.run(
-            list(argv),
-            cwd=str(self.codebase_path),
-            capture_output=True,
-            text=True,
-            timeout=self.timeout_seconds,
-        )
+        with self._subprocess_lock:
+            proc = subprocess.run(
+                list(argv),
+                cwd=str(self.codebase_path),
+                capture_output=True,
+                text=True,
+                timeout=self.timeout_seconds,
+            )
         stdout = (proc.stdout or "").strip()
         stderr = (proc.stderr or "").strip()
         if proc.returncode not in ok_returncodes:

src/metis/engine/concurrency.py

@@ -6,29 +6,121 @@
 import threading
 from collections.abc import Callable
 from collections.abc import Sequence
+from concurrent.futures import Executor
 from concurrent.futures import FIRST_COMPLETED
 from concurrent.futures import Future
 from concurrent.futures import ThreadPoolExecutor
 from concurrent.futures import wait
-from functools import partial
+from typing import TYPE_CHECKING
 
 from metis import runlog
 from metis.usage import submit_with_current_context
 
+if TYPE_CHECKING:
+    from metis.engine.execution.contracts import NodeJobs
+
+
+class JobScheduler:
+    def __init__(self, max_workers: int) -> None:
+        self._max_workers = max_workers
+        self._executor = ThreadPoolExecutor(
+            max_workers=max_workers,
+            thread_name_prefix="metis",
+        )
+
+    def limit(self, max_concurrency: int) -> "NodeJobs":
+        return _NodeJobs(
+            self._executor,
+            self._max_workers,
+        ).limit(max_concurrency)
+
+    def close(self) -> None:
+        self._executor.shutdown()
+
+
+class _NodeJobs:
+    def __init__(self, executor: Executor, max_concurrency: int) -> None:
+        self._executor = executor
+        self._max_concurrency = max_concurrency
+
+    def limit(self, max_concurrency: int) -> "_NodeJobs":
+        if max_concurrency < 1:
+            raise ValueError("max_concurrency must be positive")
+        return _NodeJobs(
+            self._executor,
+            min(max_concurrency, self._max_concurrency),
+        )
+
+    def run[JobT, ResultT](
+        self,
+        jobs: Sequence[JobT],
+        worker: Callable[[JobT], ResultT],
+        *,
+        label: str | None,
+        result_key: Callable[[JobT], object],
+        on_complete: Callable[[JobT, int, int], None] | None = None,
+    ) -> list[ResultT]:
+        if not jobs:
+            return []
+
+        total = len(jobs)
+        worker_count = min(self._max_concurrency, total)
+        results: list[ResultT] = []
+
+        def invoke_worker(job: JobT) -> ResultT:
+            if label is None:
+                return worker(job)
+            key = result_key(job)
+            with runlog.span(
+                "task",
+                label,
+                {"kind": "concurrent_job", "key": key},
+            ) as task_span:
+                runlog.bump("tasks")
+                result = worker(job)
+                task_span.end(attributes={"result": result})
+                return result
+
+        def collect(
+            job: JobT,
+            completed: int,
+            result_func: Callable[[], ResultT],
+        ) -> None:
+            results.append(result_func())
+            if on_complete:
+                on_complete(job, completed, total)
 
-def coerce_worker_count(
-    max_workers: int | str | None,
-    *,
-    default: int = 1,
-) -> int:
-    value = default if max_workers is None or max_workers == "" else max_workers
-    return max(1, int(value))
-
+        job_iterator = iter(jobs)
+        futures: dict[Future[ResultT], JobT] = {}
+        completed = 0
+        try:
+            for job in job_iterator:
+                futures[
+                    submit_with_current_context(self._executor, invoke_worker, job)
+                ] = job
+                if len(futures) >= worker_count:
+                    break
 
-def bounded_worker_count(max_workers: int | str | None, item_count: int) -> int:
-    if item_count <= 1:
-        return 1
-    return min(coerce_worker_count(max_workers), item_count)
+            while futures:
+                completed_futures, _pending = wait(
+                    futures,
+                    return_when=FIRST_COMPLETED,
+                )
+                for future in completed_futures:
+                    completed += 1
+                    collect(futures.pop(future), completed, future.result)
+                for job in job_iterator:
+                    futures[
+                        submit_with_current_context(self._executor, invoke_worker, job)
+                    ] = job
+                    if len(futures) >= worker_count:
+                        break
+        except BaseException:
+            for future in futures:
+                future.cancel()
+            wait(futures)
+            raise
+        return results
 
 
 def serialized_progress_callback(callback):
@@ -45,70 +137,3 @@ def _serialized(event):
 
     _serialized._metis_serialized_progress_callback = True
     return _serialized
-
-
-def run_jobs[JobT, ResultT](
-    jobs: Sequence[JobT],
-    worker: Callable[[JobT], ResultT],
-    *,
-    max_workers: int | str | None,
-    label: str,
-    result_key: Callable[[JobT], object],
-    on_complete: Callable[[JobT, int, int], None] | None = None,
-) -> list[ResultT]:
-    if not jobs:
-        return []
-
-    total = len(jobs)
-    worker_count = bounded_worker_count(max_workers, total)
-    results: list[ResultT] = []
-
-    def invoke_worker(job: JobT) -> ResultT:
-        key = result_key(job)
-        with runlog.span(
-            "task",
-            label,
-            {"kind": "concurrent_job", "key": key},
-        ) as task_span:
-            runlog.bump("tasks")
-            result = worker(job)
-            task_span.end(attributes={"result": result})
-            return result
-
-    def collect(
-        job: JobT,
-        completed: int,
-        result_func: Callable[[], ResultT],
-    ) -> None:
-        results.append(result_func())
-        if on_complete:
-            on_complete(job, completed, total)
-
-    if worker_count == 1:
-        for completed, job in enumerate(jobs, start=1):
-            collect(job, completed, partial(invoke_worker, job))
-        return results
-
-    with ThreadPoolExecutor(max_workers=worker_count) as executor:
-        job_iterator = iter(jobs)
-        futures: dict[Future[ResultT], JobT] = {}
-        completed = 0
-
-        for job in job_iterator:
-            futures[submit_with_current_context(executor, invoke_worker, job)] = job
-            if len(futures) >= worker_count:
-                break
-
-        while futures:
-            completed_futures, _pending = wait(
-                futures,
-                return_when=FIRST_COMPLETED,
-            )
-            for future in completed_futures:
-                completed += 1
-                collect(futures.pop(future), completed, future.result)
-            for job in job_iterator:
-                futures[submit_with_current_context(executor, invoke_worker, job)] = job
-                if len(futures) >= worker_count:
-                    break
-    return results

src/metis/engine/core.py

@@ -12,6 +12,7 @@
 
 from metis import runlog
 from metis.chat_model_options import merge_chat_model_kwargs
+from metis.configuration import _required_positive_int
 from metis.configuration import load_execution_config
 from metis.configuration import load_plugin_config
 from metis.plugins.c_family.codegraph import CFamilyCodeGraphProvider
@@ -75,7 +76,11 @@ def __init__(
         if missing:
             raise ValueError(f"Missing required config: {', '.join(missing)}")
 
-        max_workers = cast(int, kwargs["max_workers"])
+        max_workers = _required_positive_int(
+            kwargs,
+            "max_workers",
+            section="MetisEngine",
+        )
         max_token_length = cast(int, kwargs["max_token_length"])
         llama_query_model = cast(str, kwargs["llama_query_model"])
         similarity_top_k = cast(int, kwargs["similarity_top_k"])
@@ -105,7 +110,6 @@ def __init__(
             kwargs.get("reachability_config") or {}
         )
         reachability_settings = reachability_config.as_review_settings()
-        reachability_settings["max_workers"] = max_workers
         reasoning_effort = chat_model_kwargs.get("reasoning_effort")
         if reasoning_effort is not None:
             reachability_settings["reasoning_effort"] = reasoning_effort
@@ -394,6 +398,7 @@ def execute_triage(
             return _execution_value(triage)
 
     def close(self):
+        self.execution.close()
         if self._triage_classifier is not None:
             self._triage_classifier.close()
         self.capabilities.close()

src/metis/engine/execution/contracts.py

@@ -5,6 +5,7 @@
 
 from collections.abc import Callable
 from collections.abc import Mapping
+from collections.abc import Sequence
 from dataclasses import dataclass
 from dataclasses import field
 from enum import Enum
@@ -65,6 +66,20 @@ def materialize(
     def load(self, reference: CodeGraphReference) -> CodeGraph: ...
 
 
+class NodeJobs(Protocol):
+    def limit(self, max_concurrency: int) -> "NodeJobs": ...
+
+    def run[JobT, ResultT](
+        self,
+        jobs: Sequence[JobT],
+        worker: Callable[[JobT], ResultT],
+        *,
+        label: str | None,
+        result_key: Callable[[JobT], object],
+        on_complete: Callable[[JobT, int, int], None] | None = None,
+    ) -> list[ResultT]: ...
+
+
 class EmptyNodeConfiguration(BaseModel):
     model_config = ConfigDict(extra="forbid", frozen=True)
 
@@ -116,6 +131,7 @@ class NodeRuntime:
     chat_model_kwargs: Mapping[str, object]
     model_tool_max_rounds: int = 0
     token_counter: Callable[[str], int] = count_tokens
+    jobs: NodeJobs | None = None
 
     def __post_init__(self) -> None:
         object.__setattr__(
@@ -144,6 +160,12 @@ def __post_init__(self) -> None:
             MappingProxyType(dict(self.capabilities)),
         )
 
+    @property
+    def jobs(self) -> NodeJobs:
+        if self.runtime.jobs is None:
+            raise RuntimeError("Node job scheduler is unavailable")
+        return self.runtime.jobs
+
 
 @dataclass(frozen=True, slots=True)
 class NodeInvocation:

src/metis/engine/execution/graph.py

@@ -20,6 +20,7 @@
 class ConfiguredNode(BaseModel):
     inputs: dict[str, InputBinding] = Field(default_factory=dict)
     depends_on: tuple[str, ...] = ()
+    max_concurrency: int | None = Field(default=None, strict=True, ge=1)
     capabilities: tuple[str, ...] = ()
     formats: tuple[ResultFormat, ...] | None = None
     filename: str | None = Field(default=None, min_length=1)