docs: correct orchestrator invocation and timeout guarantees

docs/orchestrator_mapreduce.md

@@ -18,7 +18,7 @@ The module is additive — it imports from existing project modules without modi
 
 ## Architecture
 
-```
+```text
                     ┌──────────────────────────────┐
                     │     orchestrator.run()        │
                     │     (Single Entry Point)      │
@@ -130,7 +130,7 @@ HierarchicalOrchestrator(
 | `verification_gate` | `Callable` | `_default_gate` | Async function that evaluates whether a job passes verification |
 | `reducer` | `Callable` | `_default_reducer` | Async function that aggregates subtask results into a single output |
 | `state_file` | `str` | `<module_dir>/STATE.md` | Path to the Markdown file where job state is persisted |
-| `subtask_timeout` | `float` | `300.0` | Maximum seconds a single subtask may run before being killed |
+| `subtask_timeout` | `float` | `300.0` | Maximum seconds to await a subtask before marking it failed; a running protocol thread may continue |
 | `max_job_history` | `int` | `1000` | Maximum number of completed jobs retained in memory |
 
 ---
@@ -143,10 +143,10 @@ The single entry point. Executes the full Plan → Map → Reduce cycle.
 
 ```python
 job = await orchestrator.run(
-    intent="Validate all API endpoints after deployment",
+    intent="Process multiple data directories",
     task_list=[
-        {"protocol": "api_health_checker", "inputs": {"endpoint": "/users"}},
-        {"protocol": "api_health_checker", "inputs": {"endpoint": "/orders"}},
+        {"protocol": "data_processor", "inputs": {"data_path": "/data/users"}},
+        {"protocol": "data_processor", "inputs": {"data_path": "/data/orders"}},
     ],
 )
 ```
@@ -158,6 +158,8 @@ job = await orchestrator.run(
 
 **Returns:** `OrchestratedJob` with final status, reduced results, and verification outcome.
 
+**Inputs:** Zero-argument `task()` functions ignore `inputs`. Parameterized tasks receive the full dict via `task(**inputs)` without key filtering, so callers must supply compatible keys or use a protocol accepting `**kwargs`.
+
 ---
 
 ### `plan(intent, task_list) → OrchestratedJob`
@@ -180,7 +182,7 @@ Phase 2. Executes all subtasks in parallel with bounded concurrency. Each subtas
 **Behavior:**
 - Acquires a semaphore slot before executing each subtask
 - Applies `subtask_timeout` via `asyncio.wait_for()`
-- Tracks outcomes via `track_outcome()` (non-blocking, via `run_in_executor`)
+- Tracks successful returns and non-timeout exceptions via `track_outcome()` (non-blocking, via `run_in_executor`); timeout failures are not sent to this tracker
 - Sets `subtask.status` to `COMPLETED` or `FAILED`
 - Never raises — failures are captured on individual subtasks
 
@@ -278,11 +280,13 @@ The module uses `inspect.signature()` to determine how to call each protocol's `
 
 | Protocol Signature | Behavior |
 | :--- | :--- |
-| `def task():` | Called with no arguments (legacy compatibility) |
+| `def task():` | Called with no arguments; `inputs` is ignored (legacy compatibility) |
 | `def task(**kwargs):` | Receives the full `inputs` dict as keyword arguments |
-| `def task(endpoint, timeout=30):` | Receives matching keys from `inputs` as named arguments |
+| `def task(endpoint, timeout=30):` | Receives the full `inputs` dict as keyword arguments; extra keys or missing required keys raise `TypeError` and fail the subtask |
+
+**Illustrative new-style protocol (not the current `protocols/api_health_checker.py`):**
 
-**Example — New-style protocol (`protocols/api_health_checker.py`):**
+The repository's existing `api_health_checker.task()` takes no arguments and checks its own endpoint list. To use the parameterized examples, implement a separate protocol like the following and use its name in `task_list`:
 
 ```python
 import requests
@@ -333,7 +337,9 @@ The module implements a three-tier failure strategy:
 | :--- | :--- | :--- |
 | **Retry** | Subtask failed, `attempts < max_attempts` | Mutate protocol, re-execute |
 | **Escalate** | All retries exhausted, verification still failing | Log escalation, notify human (via MCP/Slack when configured) |
-| **Timeout** | Subtask exceeds `subtask_timeout` seconds | Kill subtask, mark as FAILED, enter retry tier |
+| **Timeout** | Awaiting the subtask exceeds `subtask_timeout` seconds | Mark as FAILED and enter retry tier; the underlying protocol thread may continue |
+
+`asyncio.wait_for()` bounds the wait for the executor result; it does not terminate an already-running thread. A timed-out protocol may still produce side effects, and a retry may overlap that work. Set timeouts inside blocking operations and make retryable protocols idempotent.
 
 To enable Slack escalation in production, uncomment the MCP connector call in `_escalate_to_human()` and configure your Slack channel.
 
@@ -375,20 +381,20 @@ asyncio.run(main())
 ### Fan-Out Same Protocol Across Many Inputs
 
 ```python
-async def validate_endpoints():
+async def process_data_directories():
     orchestrator = HierarchicalOrchestrator(
         max_concurrency=20,
         subtask_timeout=30.0,
     )
 
-    endpoints = ["/users", "/orders", "/payments", "/auth", "/products"]
+    data_paths = ["/data/users", "/data/orders", "/data/payments"]
     task_list = [
-        {"protocol": "api_health_checker", "inputs": {"endpoint": ep}}
-        for ep in endpoints
+        {"protocol": "data_processor", "inputs": {"data_path": data_path}}
+        for data_path in data_paths
     ]
 
     return await orchestrator.run(
-        intent="Post-deployment endpoint validation",
+        intent="Process multiple data directories",
         task_list=task_list,
     )
 ```
@@ -517,7 +523,7 @@ No modifications to these modules are required. The orchestrator wraps all synch
 
 ## Lifecycle Diagram
 
-```
+```text
 User Intent
     │
     ▼

orchestrator_mapreduce.py

@@ -48,7 +48,6 @@
 from agents.mutator import mutate_protocol
 from utils.tracker import track_outcome, get_protocol_stats
 
-
 # ─── Data Models ───────────────────────────────────────────────
 
 
@@ -116,9 +115,7 @@ async def verify(self, job: OrchestratedJob) -> Dict[str, Any]:
         """
         # Layer 1: Mechanical gate (fast pre-filter)
         total = len(job.subtasks)
-        succeeded = sum(
-            1 for st in job.subtasks if st.status == TaskStatus.COMPLETED
-        )
+        succeeded = sum(1 for st in job.subtasks if st.status == TaskStatus.COMPLETED)
         rate = succeeded / total if total > 0 else 0
 
         if rate < self.threshold:
@@ -127,8 +124,7 @@ async def verify(self, job: OrchestratedJob) -> Dict[str, Any]:
                 "layer": "mechanical",
                 "success_rate": rate,
                 "reason": (
-                    f"Success rate {rate:.0%} below "
-                    f"{self.threshold:.0%} threshold"
+                    f"Success rate {rate:.0%} below " f"{self.threshold:.0%} threshold"
                 ),
             }
 
@@ -139,7 +135,11 @@ async def verify(self, job: OrchestratedJob) -> Dict[str, Any]:
         except Exception as e:
             # If LLM is unavailable, fall back to mechanical pass
             log(f"⚠️  Semantic gate unavailable ({e}), using mechanical only")
-            return {"passed": True, "layer": "mechanical_fallback", "success_rate": rate}
+            return {
+                "passed": True,
+                "layer": "mechanical_fallback",
+                "success_rate": rate,
+            }
 
     async def _llm_evaluate(self, job: OrchestratedJob) -> Dict[str, Any]:
         """Call the LLM to semantically evaluate the job output."""
@@ -209,12 +209,8 @@ class StateStore:
     """
 
     def __init__(self, db_path: Optional[str] = None, md_path: Optional[str] = None):
-        self.db_path = db_path or str(
-            Path(__file__).parent / "loop_state.db"
-        )
-        self.md_path = md_path or str(
-            Path(__file__).parent / "STATE.md"
-        )
+        self.db_path = db_path or str(Path(__file__).parent / "loop_state.db")
+        self.md_path = md_path or str(Path(__file__).parent / "STATE.md")
         self._init_db()
 
     def _init_db(self):
@@ -261,12 +257,8 @@ def persist_job(self, job: OrchestratedJob):
         conn = sqlite3.connect(self.db_path)
         now = datetime.now(timezone.utc).isoformat()
 
-        succeeded = sum(
-            1 for st in job.subtasks if st.status == TaskStatus.COMPLETED
-        )
-        failed = sum(
-            1 for st in job.subtasks if st.status == TaskStatus.FAILED
-        )
+        succeeded = sum(1 for st in job.subtasks if st.status == TaskStatus.COMPLETED)
+        failed = sum(1 for st in job.subtasks if st.status == TaskStatus.FAILED)
 
         conn.execute(
             """INSERT OR REPLACE INTO jobs
@@ -305,12 +297,16 @@ def persist_job(self, job: OrchestratedJob):
                     st.error,
                     st.attempts,
                     (
-                        datetime.fromtimestamp(st.started_at, tz=timezone.utc).isoformat()
+                        datetime.fromtimestamp(
+                            st.started_at, tz=timezone.utc
+                        ).isoformat()
                         if st.started_at
                         else None
                     ),
                     (
-                        datetime.fromtimestamp(st.completed_at, tz=timezone.utc).isoformat()
+                        datetime.fromtimestamp(
+                            st.completed_at, tz=timezone.utc
+                        ).isoformat()
                         if st.completed_at
                         else None
                     ),
@@ -322,9 +318,7 @@ def persist_job(self, job: OrchestratedJob):
 
     def persist_markdown(self, job: OrchestratedJob):
         """Append a human-readable summary to STATE.md."""
-        succeeded = sum(
-            1 for st in job.subtasks if st.status == TaskStatus.COMPLETED
-        )
+        succeeded = sum(1 for st in job.subtasks if st.status == TaskStatus.COMPLETED)
         entry = (
             f"\n## Job: {job.id}\n"
             f"- **Intent:** {job.intent}\n"
@@ -458,7 +452,9 @@ async def _mutate_strategy(
 
         subtask.inputs.update(strategy_hints)
 
-        log(f"   → Tier 2 (Strategy Mutation): Injected strategy hints for '{protocol_name}'")
+        log(
+            f"   → Tier 2 (Strategy Mutation): Injected strategy hints for '{protocol_name}'"
+        )
         return {"tier": 2, "strategy_note": strategy_hints["_instruction"]}
 
     async def _mutate_code_sandboxed(
@@ -470,12 +466,12 @@ async def _mutate_code_sandboxed(
         The mutation is tested in a sandboxed dry-run before being applied.
         """
         # Perform the mutation via the existing mutator
-        mutated = await loop.run_in_executor(
-            None, mutate_protocol, protocol_name
-        )
+        mutated = await loop.run_in_executor(None, mutate_protocol, protocol_name)
 
         if not mutated:
-            log(f"   → Tier 3 (Code Mutation): mutate_protocol returned None for '{protocol_name}'")
+            log(
+                f"   → Tier 3 (Code Mutation): mutate_protocol returned None for '{protocol_name}'"
+            )
             return {"tier": 3, "code_mutated": False, "reason": "Mutator returned None"}
 
         # Sandbox test: attempt to load and validate the mutated protocol
@@ -486,7 +482,9 @@ async def _mutate_code_sandboxed(
                 None, load_protocol, protocol_name
             )
             if test_protocol and callable(test_protocol.get("task")):
-                log(f"   → Tier 3 (Code Mutation): Mutation applied and validated for '{protocol_name}'")
+                log(
+                    f"   → Tier 3 (Code Mutation): Mutation applied and validated for '{protocol_name}'"
+                )
                 return {"tier": 3, "code_mutated": True}
             else:
                 log(f"   → Tier 3 (Code Mutation): Mutated protocol failed validation")
@@ -514,9 +512,7 @@ async def escalate(self, job: OrchestratedJob, reason: str):
         Post a structured failure alert to Slack.
         Includes job ID, intent, failure reason, and failed subtask details.
         """
-        failed_subtasks = [
-            st for st in job.subtasks if st.status == TaskStatus.FAILED
-        ]
+        failed_subtasks = [st for st in job.subtasks if st.status == TaskStatus.FAILED]
         failed_details = "\n".join(
             f"  • `{st.protocol}` — {st.error or 'unknown error'}"
             for st in failed_subtasks[:5]
@@ -535,9 +531,7 @@ async def escalate(self, job: OrchestratedJob, reason: str):
         loop = asyncio.get_running_loop()
 
         try:
-            await loop.run_in_executor(
-                None, self._post_to_slack, message
-            )
+            await loop.run_in_executor(None, self._post_to_slack, message)
             log(f"📨 Escalation posted to Slack {self.channel}")
         except Exception as e:
             # If Slack fails, log locally — never let escalation failure crash the system
@@ -684,9 +678,7 @@ async def _execute_one(subtask: SubTask):
 
                 try:
                     result = await asyncio.wait_for(
-                        self._run_protocol_isolated(
-                            subtask.protocol, subtask.inputs
-                        ),
+                        self._run_protocol_isolated(subtask.protocol, subtask.inputs),
                         timeout=self.subtask_timeout,
                     )
                     subtask.result = result
@@ -727,9 +719,7 @@ async def _execute_one(subtask: SubTask):
             return_exceptions=True,
         )
 
-        completed = sum(
-            1 for st in job.subtasks if st.status == TaskStatus.COMPLETED
-        )
+        completed = sum(1 for st in job.subtasks if st.status == TaskStatus.COMPLETED)
         failed = sum(1 for st in job.subtasks if st.status == TaskStatus.FAILED)
         log(f"📊 MAP complete: {completed} succeeded, {failed} failed")
 
@@ -760,15 +750,16 @@ async def reduce_verify(self, job: OrchestratedJob) -> OrchestratedJob:
             log(f"✅ REDUCE: Verification PASSED for job '{job.intent}'")
         else:
             job.verification_passed = False
-            log(f"❌ REDUCE: Verification FAILED — {verification.get('reason', 'unknown')}")
+            log(
+                f"❌ REDUCE: Verification FAILED — {verification.get('reason', 'unknown')}"
+            )
 
             # Step 3: Self-Correct with safe mutation hierarchy
             while True:
                 retryable = [
                     st
                     for st in job.subtasks
-                    if st.status == TaskStatus.FAILED
-                    and st.attempts < st.max_attempts
+                    if st.status == TaskStatus.FAILED and st.attempts < st.max_attempts
                 ]
 
                 if not retryable:
@@ -797,11 +788,7 @@ async def reduce_verify(self, job: OrchestratedJob) -> OrchestratedJob:
                 # Re-verify after retry
                 job.reduced_result = await self.reducer(
                     [st.result for st in job.subtasks if st.result],
-                    [
-                        st
-                        for st in job.subtasks
-                        if st.status == TaskStatus.FAILED
-                    ],
+                    [st for st in job.subtasks if st.status == TaskStatus.FAILED],
                 )
                 job.attempt_count += 1
                 re_verification = await self.verification_gate(job)
@@ -875,9 +862,7 @@ async def _retry_subtask(self, subtask: SubTask):
         subtask.started_at = time.time()
         try:
             result = await asyncio.wait_for(
-                self._run_protocol_isolated(
-                    subtask.protocol, subtask.inputs
-                ),
+                self._run_protocol_isolated(subtask.protocol, subtask.inputs),
                 timeout=self.subtask_timeout,
             )
             subtask.result = result
@@ -903,9 +888,7 @@ async def _persist_state(self, job: OrchestratedJob):
     def _prune_job_history(self):
         """Prevent unbounded memory growth from stored jobs."""
         if len(self.jobs) > self.max_job_history:
-            sorted_jobs = sorted(
-                self.jobs.items(), key=lambda x: x[1].created_at
-            )
+            sorted_jobs = sorted(self.jobs.items(), key=lambda x: x[1].created_at)
             excess = len(self.jobs) - self.max_job_history
             for job_id, _ in sorted_jobs[:excess]:
                 del self.jobs[job_id]
@@ -918,9 +901,7 @@ async def _default_gate(self, job: OrchestratedJob) -> Dict[str, Any]:
         Used only when use_semantic_gate=False and no custom gate provided.
         """
         total = len(job.subtasks)
-        succeeded = sum(
-            1 for st in job.subtasks if st.status == TaskStatus.COMPLETED
-        )
+        succeeded = sum(1 for st in job.subtasks if st.status == TaskStatus.COMPLETED)
         rate = succeeded / total if total > 0 else 0
 
         if rate >= 0.8: