diff --git a/docs/orchestrator_mapreduce.md b/docs/orchestrator_mapreduce.md
index 0e758ba..c58d953 100644
--- a/docs/orchestrator_mapreduce.md
+++ b/docs/orchestrator_mapreduce.md
@@ -12,13 +12,13 @@
 
 The `orchestrator_mapreduce` module introduces a formal **Plan → Map → Reduce** orchestration layer to the self-correcting executor. It enables parallel execution of protocol-based subtasks with bounded concurrency, a configurable Verification Gateway, automatic self-correction via protocol mutation on failure, and graceful human escalation when all retries are exhausted.
 
-The module is additive — it imports from existing project modules without modifying them.
+This guide describes the current `HierarchicalOrchestrator` in this module, not the separate `OrchestrationEngine` or BuildPlan runtime. The class invokes synchronous protocol functions in a shared thread pool. It does not provide process isolation, hard cancellation, or an authorization boundary. Read the return-value, retry, and timeout limits below before using it for side effects.
 
 ---
 
 ## Architecture
 
-```
+```text
                     ┌──────────────────────────────┐
                     │     orchestrator.run()        │
                     │     (Single Entry Point)      │
@@ -51,7 +51,10 @@ The module is additive — it imports from existing project modules without modi
 ### Exports
 
 ```python
-__all__ = ["HierarchicalOrchestrator", "OrchestratedJob", "SubTask", "TaskStatus"]
+__all__ = [
+    "HierarchicalOrchestrator", "OrchestratedJob", "SubTask", "TaskStatus",
+    "SemanticGate", "StateStore",
+]
 ```
 
 ---
@@ -66,7 +69,7 @@ An enumeration of possible states for subtasks and jobs.
 | :--- | :--- |
 | `PENDING` | Created but not yet started |
 | `RUNNING` | Currently executing |
-| `COMPLETED` | Finished successfully |
+| `COMPLETED` | Protocol returned without raising, or job passed its configured gate; not proof of business success |
 | `FAILED` | Finished with an error or timeout |
 | `RETRYING` | Previously failed, now being re-executed after mutation |
 
@@ -87,7 +90,7 @@ A dataclass representing a single unit of work dispatched to a worker.
 | `attempts` | `int` | `0` | Number of execution attempts made |
 | `max_attempts` | `int` | `3` | Maximum allowed attempts before giving up |
 | `started_at` | `Optional[float]` | `None` | Unix timestamp when execution began |
-| `completed_at` | `Optional[float]` | `None` | Unix timestamp when execution ended |
+| `completed_at` | `Optional[float]` | `None` | Time the await finished or timed out; a timed-out thread may still be running |
 
 ---
 
@@ -104,6 +107,7 @@ A dataclass representing a top-level job composed of parallel subtasks.
 | `reduced_result` | `Optional[Dict]` | `None` | Aggregated output from the reducer |
 | `verification_passed` | `bool` | `False` | Whether the Verification Gateway passed |
 | `created_at` | `float` | `time.time()` | Unix timestamp of job creation |
+| `attempt_count` | `int` | `0` | Number of verification rounds |
 
 ---
 
@@ -115,23 +119,31 @@ The main orchestration engine. Manages the full Plan → Map → Reduce lifecycl
 
 ```python
 HierarchicalOrchestrator(
-    max_concurrency: int = 10,
-    verification_gate: Optional[Callable] = None,
-    reducer: Optional[Callable] = None,
-    state_file: Optional[str] = None,
-    subtask_timeout: float = 300.0,
-    max_job_history: int = 1000,
+    max_concurrency=10,
+    verification_gate=None,
+    reducer=None,
+    state_file=None,
+    subtask_timeout=300.0,
+    max_job_history=1000,
+    slack_channel="#agent-alerts",
+    use_semantic_gate=True,
+    semantic_model="gpt-4o-mini",
+    db_path=None,
 )
 ```
 
 | Parameter | Type | Default | Description |
 | :--- | :--- | :--- | :--- |
-| `max_concurrency` | `int` | `10` | Maximum number of subtasks executing simultaneously (semaphore size) |
-| `verification_gate` | `Callable` | `_default_gate` | Async function that evaluates whether a job passes verification |
+| `max_concurrency` | `int` | `10` | Initial map-phase semaphore size; does not cap retry fan-out or already-timed-out threads |
+| `verification_gate` | `Callable` | `SemanticGate.verify` | Custom async gate takes precedence; otherwise semantic verification is enabled by default |
 | `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 |
-| `max_job_history` | `int` | `1000` | Maximum number of completed jobs retained in memory |
+| `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 in-memory jobs of any status; oldest entries are pruned during planning |
+| `slack_channel` | `str` | `#agent-alerts` | Channel passed to the Slack MCP CLI on exhausted verification failure |
+| `use_semantic_gate` | `bool` | `True` | Use `SemanticGate` unless a custom gate is supplied; `False` selects mechanical verification |
+| `semantic_model` | `str` | `gpt-4o-mini` | Model for the optional OpenAI client |
+| `db_path` | `str` | `<module_dir>/loop_state.db` | SQLite persistence path |
 
 ---
 
@@ -143,10 +155,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 +170,10 @@ job = await orchestrator.run(
 
 **Returns:** `OrchestratedJob` with final status, reduced results, and verification outcome.
 
+**Example paths:** The `/data/users` and `/data/orders` directories above must already exist and contain the intended files. `data_processor` rejects an invalid explicit path on current master. The self-contained local example below creates real temporary fixtures.
+
+**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`
@@ -175,14 +191,15 @@ job = await orchestrator.reduce_verify(job)
 
 ### `map_execute(job) → OrchestratedJob`
 
-Phase 2. Executes all subtasks in parallel with bounded concurrency. Each subtask runs in isolation — no shared mutable state between workers.
+Phase 2. Executes all subtasks using a semaphore and the event loop's shared thread pool. Inputs and protocol module state are not cloned or isolated. The loader runs on the event-loop thread before the task is submitted; blocking import work can delay the loop and timeout delivery.
 
 **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`)
-- Sets `subtask.status` to `COMPLETED` or `FAILED`
-- Never raises — failures are captured on individual subtasks
+- 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` for any normal return, including `{"success": False}`; exceptions and timeouts set `FAILED`
+- Captures ordinary per-subtask exceptions; caller cancellation and setup/configuration errors are not a guarantee of successful completion
+- Does not await coroutine results returned by async protocols. Use synchronous `task()` functions here; the separate `OrchestrationEngine` has a different dispatch contract
 
 ---
 
@@ -194,13 +211,15 @@ Phase 3. Aggregates results, runs the Verification Gateway, and triggers self-co
 1. Calls the `reducer` to aggregate all results
 2. Calls the `verification_gate` to evaluate pass/fail
 3. If failed: enters retry loop
-   - Identifies subtasks with `attempts < max_attempts`
-   - Deduplicates protocols and calls `mutate_protocol()` on each (non-blocking)
-   - Re-executes failed subtasks in parallel
+   - Identifies `FAILED` subtasks with `attempts < max_attempts`
+   - Calls `SafeMutator.mutate()` for each retryable subtask, without deduplicating protocol names
+   - Re-executes those subtasks in parallel, without the initial map semaphore or `track_outcome()` calls
    - Re-verifies after each retry round
    - Loops until verification passes or all retries exhausted
-4. If all retries exhausted: calls `_escalate_to_human()`
-5. Persists final state to `STATE.md`
+4. If verification still fails with no retryable subtasks: calls `SlackEscalation.escalate()`
+5. Persists the final snapshot to SQLite and appends a summary to `STATE.md`
+
+Only subtasks whose status is `FAILED` are retried. A normal return with `success: false` is still `COMPLETED`, so a rejecting custom gate does not make it retryable. Likewise, a semantic rejection with no failed subtasks proceeds to escalation. Reducers receive truthy stored results and the failed-subtask list; the default reducer counts results, not their `success` fields. Errors from a custom reducer/gate or persistence can propagate and prevent the normal final-state write.
 
 ---
 
@@ -227,8 +246,13 @@ async def my_gate(job: OrchestratedJob) -> Dict[str, Any]:
 
 ```python
 async def strict_gate(job):
-    failed = [st for st in job.subtasks if st.status == TaskStatus.FAILED]
-    if not failed:
+    failed = [
+        st for st in job.subtasks
+        if st.status != TaskStatus.COMPLETED
+        or not isinstance(st.result, dict)
+        or st.result.get("success") is not True
+    ]
+    if job.subtasks and not failed:
         return {"passed": True}
     return {
         "passed": False,
@@ -242,7 +266,7 @@ orchestrator = HierarchicalOrchestrator(verification_gate=strict_gate)
 
 ### Custom Reducer
 
-A reducer is an async function that receives the list of successful results and the list of failed subtasks, and returns an aggregated dict:
+A reducer is an async function that receives truthy stored results and the list of subtasks marked `FAILED`, and returns an aggregated dict. A stored result is not necessarily a successful domain outcome; inspect its fields as needed:
 
 ```python
 async def my_reducer(
@@ -278,11 +302,15 @@ 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 |
 
-**Example — New-style protocol (`protocols/api_health_checker.py`):**
+**Current built-in signatures:** `data_processor.task(data_path=None, source=None)` accepts an existing directory or the field-capture source contract. `api_health_checker.task(service=None, url=None)` now requires an artifact bundle and an explicit full health URL; it no longer uses the legacy fixed endpoint list. `log_analyzer.task(source=None)` is synchronous only when `source` is omitted. Its source-based branch returns a coroutine and belongs in `OrchestrationEngine`, not this class.
+
+**Illustrative new protocol, `protocols/example_endpoint_check.py` (not a shipped module):**
+
+Create and review this module before naming `example_endpoint_check` in a task list. Replace `https://api.example.com` with an authorized service. Do not pass `endpoint` to the existing `api_health_checker` expecting this different contract:
 
 ```python
 import requests
@@ -299,18 +327,24 @@ def task(endpoint: str, timeout: int = 10) -> dict:
 **Example — Legacy protocol (still works unchanged):**
 
 ```python
-from random import randint
-
 def task():
-    """Legacy protocol with no parameters."""
-    return {"success": bool(randint(0, 1)), "action": "default_execution"}
+    """Illustrative zero-argument fixture, not a real service check."""
+    return {"success": True, "action": "fixture_execution"}
 ```
 
 ---
 
+## Verification and Persistence Helpers
+
+`SemanticGate(model="gpt-4o-mini", threshold=0.8)` first checks the fraction of subtasks with `COMPLETED` status. If it meets the threshold, it asks an OpenAI-compatible chat model to evaluate up to 20 result summaries (each result is truncated). It accepts a response beginning with `PASS`. If the client/import/provider call fails, it returns a **mechanical fallback pass**; semantic verification is not guaranteed. The default mechanical gate also passes at **at least 80%**, even if some subtasks failed or returned `success: false`. Supply a domain-specific gate when those outcomes must fail.
+
+The LLM client is constructed as `OpenAI()` and therefore uses that installed SDK's configuration. The optional `openai` dependency and credentials are required for live semantic evaluation. Set `use_semantic_gate=False` or supply a custom gate to avoid this provider call. Local examples below use a custom gate.
+
+`StateStore(db_path=None, md_path=None)` creates SQLite `jobs` and `subtask_runs` tables. `persist_job(job)` replaces each job/subtask row by ID; it is a **final snapshot**, not per-attempt history. `persist_markdown(job)` appends the human-readable summary. `query_recent_failures(limit=10)` and `get_protocol_success_rate(protocol, days=7)` query stored final rows; the latter returns `1.0` when no rows exist, which is not measured success. These methods perform synchronous SQLite/file I/O when called directly.
+
 ## State Persistence
 
-After every job completes (pass or fail), the orchestrator appends a summary to `STATE.md`:
+On the normal return path after verification (pass or fail), the orchestrator appends a summary to `STATE.md`:
 
 ```markdown
 ## Job: job_1719082800_a1b2c3d4
@@ -321,7 +355,7 @@ After every job completes (pass or fail), the orchestrator appends a summary to
 - **Timestamp:** 2026-06-22T19:00:00.000000+00:00
 ```
 
-This file serves as a persistent audit log across runs. It is append-only and can be read by other tools, skills, or agents for cross-run continuity.
+This file records final summaries across runs, not every failed attempt or worker side effect. It is append-only and can be read by other tools, skills, or agents for cross-run continuity.
 
 ---
 
@@ -331,158 +365,90 @@ The module implements a three-tier failure strategy:
 
 | Tier | Condition | Action |
 | :--- | :--- | :--- |
-| **Retry** | Subtask failed, `attempts < max_attempts` | Mutate protocol, re-execute |
+| **Retry** | Subtask status is `FAILED`, `attempts < max_attempts` | Apply the attempt-based mutation tier and 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.
+`SafeMutator` adds `_retry`, `_previous_error`, and `_attempt` on the first retry; on a timeout-like error it also adds/doubles `timeout`. The second retry adds strategy-hint keys. Protocols without compatible parameters or `**kwargs` will reject these added keys; hints are not proof of a correction. With the default three total attempts, code mutation is never reached. Raising a subtask's `max_attempts` beyond three permits the code tier, which invokes the existing mutator and checks that the loaded module exposes a callable task. Despite its method name, this is **not a process sandbox, transactional rollback, or successful dry-run**; the legacy mutator can rewrite source files and Python imports can remain cached.
+
+Escalation is already active: `SlackEscalation` runs `manus-mcp-cli tool call slack_post_message --server slack --input ...` with the configured channel and a 30-second subprocess timeout. It catches CLI errors and logs locally; logging is not delivery proof. There is no `_escalate_to_human()` method to uncomment. A failed full `run()` can therefore attempt an external Slack call. The local demonstration uses successful controlled inputs; tests replace escalation with a failing assertion to detect any unexpected external attempt.
+
+`asyncio.run()` normally waits for its default executor to finish during shutdown. A protocol that times out from the orchestrator's perspective may therefore still delay process exit. The timeout also excludes tracking, reduction, verification and persistence; this is not a deadline for the whole job.
 
 ---
 
 ## Memory Management
 
-The orchestrator prevents unbounded memory growth through the `max_job_history` parameter. When the number of stored jobs exceeds this limit, the oldest jobs are pruned automatically during the `plan()` phase.
+The `max_job_history` parameter bounds the orchestrator's in-memory `self.jobs` history. During `plan()`, the oldest entries are pruned when this limit is exceeded, regardless of their status. Pruning does not delete rows from SQLite or entries from `STATE.md`; persistent-storage retention must be managed separately.
 
 ---
 
 ## Usage Examples
 
+All five patterns below are implemented in `examples/mapreduce_local.py` using the actual `data_processor` and synchronous `log_analyzer` protocols. Run the complete demonstration from the repository root:
+
+```sh
+python -m examples.mapreduce_local
+```
+
+The runner creates JSON files with one and two records and one log line in a temporary directory. It uses a strict gate that requires every subtask to return `success: true`, writes real SQLite/Markdown/tracker state, then removes the fixtures. It replaces Slack escalation with a visible local exception and never invokes the LLM gate. The example is local I/O verification, not a production health check.
+
+For the callable recipes below, `data_path`/`data_paths` must be existing directories containing the intended files and `output_dir` must be writable. `LOG_DIR` must name an existing directory for the two log-analysis recipes. The standalone runner supplies all these inputs. Helpers are ordinary application examples, not additional orchestrator API methods.
+
 ### Minimal Example
 
 ```python
-import asyncio
-from orchestrator_mapreduce import HierarchicalOrchestrator
-
-async def main():
-    orchestrator = HierarchicalOrchestrator(max_concurrency=5)
-
-    job = await orchestrator.run(
-        intent="Run smoke tests",
-        task_list=[
-            {"protocol": "smoke_test", "inputs": {"service": "auth"}},
-            {"protocol": "smoke_test", "inputs": {"service": "payments"}},
-            {"protocol": "smoke_test", "inputs": {"service": "notifications"}},
-        ],
-    )
-
-    print(f"Status: {job.status.value}")
-    print(f"Passed: {job.verification_passed}")
-    print(f"Results: {job.reduced_result}")
-
-asyncio.run(main())
+from examples.mapreduce_local import minimal
+
+job = await minimal(data_path, output_dir)
+print(job.status.value, job.verification_passed, job.reduced_result)
 ```
 
+`minimal` supplies one `data_processor` task with `inputs={"data_path": ...}`. It does not use an absent `smoke_test` protocol: the loader's fallback would create a random placeholder, not run a real smoke test.
+
 ### Fan-Out Same Protocol Across Many Inputs
 
 ```python
-async def validate_endpoints():
-    orchestrator = HierarchicalOrchestrator(
-        max_concurrency=20,
-        subtask_timeout=30.0,
-    )
-
-    endpoints = ["/users", "/orders", "/payments", "/auth", "/products"]
-    task_list = [
-        {"protocol": "api_health_checker", "inputs": {"endpoint": ep}}
-        for ep in endpoints
-    ]
+from examples.mapreduce_local import fan_out
 
-    return await orchestrator.run(
-        intent="Post-deployment endpoint validation",
-        task_list=task_list,
-    )
+job = await fan_out(data_paths, output_dir)
 ```
 
+`fan_out` creates one `data_processor` task per existing directory and uses `max_concurrency=2`. The supplied directory paths are actually consumed; they are not ignored zero-argument inputs.
+
 ### Fan-Out Multiple Protocols (Heterogeneous)
 
 ```python
-async def research_pipeline():
-    orchestrator = HierarchicalOrchestrator(max_concurrency=4)
-
-    keywords = ["ai-agents", "mcp-protocol", "loop-engineering"]
+from examples.mapreduce_local import heterogeneous
 
-    task_list = [
-        {"protocol": "keyword_research", "inputs": {"keyword": kw}}
-        for kw in keywords
-    ] + [
-        {"protocol": "competitor_scan", "inputs": {"keyword": kw}}
-        for kw in keywords
-    ]
-
-    return await orchestrator.run(
-        intent="Research and scan competitors for target keywords",
-        task_list=task_list,
-    )
+job = await heterogeneous(data_path, output_dir)
 ```
 
+This runs `data_processor(data_path=...)` alongside `log_analyzer()` configured by `LOG_DIR`. To adapt the pattern to keyword research and competitor scans, first implement and review those protocol modules with the intended `keyword` contract. `keyword_research` and `competitor_scan` are not shipped protocols; naming them alone does not provide research functionality.
+
 ### Custom Gate + Custom Reducer
 
 ```python
-async def ci_triage():
-    async def triage_gate(job):
-        results = [st.result for st in job.subtasks if st.result]
-        unknowns = [r for r in results if r.get("classification") == "unknown"]
-        if not unknowns:
-            return {"passed": True}
-        return {
-            "passed": False,
-            "reason": f"{len(unknowns)} failures could not be classified",
-        }
-
-    async def triage_reducer(results, failures):
-        categories = {}
-        for r in results:
-            cat = r.get("classification", "unknown")
-            categories.setdefault(cat, []).append(r)
-        return {
-            "total_classified": len(results),
-            "categories": {k: len(v) for k, v in categories.items()},
-        }
-
-    orchestrator = HierarchicalOrchestrator(
-        max_concurrency=5,
-        verification_gate=triage_gate,
-        reducer=triage_reducer,
-    )
-
-    return await orchestrator.run(
-        intent="Classify overnight CI failures",
-        task_list=[
-            {"protocol": "log_analyzer", "inputs": {"run_id": rid}}
-            for rid in failed_run_ids
-        ],
-    )
+from examples.mapreduce_local import custom_gate_and_reducer
+
+job = await custom_gate_and_reducer(data_path, output_dir)
+print(job.reduced_result["categories"])
 ```
 
+The example's strict gate checks both execution status and `result["success"]`. Its reducer groups actual results by `action`, producing one `data_processing` and one `log_analysis` result for the fixture. A CI-classification application can supply its own classifier and classification-aware gate/reducer; the existing `log_analyzer` does not accept `run_id` or produce a `classification` field.
+
 ### Step-by-Step Execution (Plan → Inspect → Execute)
 
 ```python
-async def controlled_execution():
-    orchestrator = HierarchicalOrchestrator()
-
-    # Phase 1: Plan only
-    job = await orchestrator.plan(
-        intent="Validate all endpoints",
-        task_list=task_list,
-    )
-
-    # Inspect the plan
-    print(f"Will execute {len(job.subtasks)} subtasks")
-    for st in job.subtasks:
-        print(f"  - {st.protocol}({st.inputs})")
-
-    # Phase 2: Execute
-    job = await orchestrator.map_execute(job)
-
-    # Inspect intermediate results
-    completed = [st for st in job.subtasks if st.status == TaskStatus.COMPLETED]
-    print(f"{len(completed)} completed before verification")
-
-    # Phase 3: Verify and self-correct
-    job = await orchestrator.reduce_verify(job)
-    return job
+from examples.mapreduce_local import step_by_step
+
+job = await step_by_step(data_path, output_dir)
 ```
 
+`step_by_step` calls `plan()`, checks the pending plan, calls `map_execute()`, inspects returned statuses, then calls `reduce_verify()`. Inspecting `COMPLETED` alone is insufficient for business verification, so it uses the same result-aware custom gate.
+
 ---
 
 ## Integration with Existing Modules
@@ -494,7 +460,7 @@ async def controlled_execution():
 | `utils.tracker.track_outcome(name, result)` | Appends execution outcomes to `memory/<name>.json` |
 | `utils.logger.log(message)` | Structured logging throughout the orchestration lifecycle |
 
-No modifications to these modules are required. The orchestrator wraps all synchronous calls in `asyncio.get_running_loop().run_in_executor()` to maintain non-blocking behavior.
+Task execution, initial outcome tracking and final persistence use `run_in_executor()`. Protocol loading/imports and constructor database initialization remain synchronous; the module does not make every synchronous dependency non-blocking. The loader defaults to creating a random placeholder for missing protocol modules, so use only reviewed, existing protocol names. That placeholder is not the requested protocol behavior.
 
 ---
 
@@ -511,13 +477,13 @@ No modifications to these modules are required. The orchestrator wraps all synch
 - For CPU-bound protocols: set `max_concurrency` to CPU core count
 - For I/O-bound protocols (API calls, file reads): set `max_concurrency` to 20–50
 - For long-running protocols: increase `subtask_timeout` accordingly
-- For high-throughput systems: increase `max_job_history` or implement external persistence
+- For high-throughput systems: size `max_job_history` for the desired in-memory history window and manage persistent-storage retention separately
 
 ---
 
 ## Lifecycle Diagram
 
-```
+```text
 User Intent
     │
     ▼
@@ -525,7 +491,7 @@ User Intent
 │ PLAN                                                     │
 │  • Generate job ID                                       │
 │  • Create SubTask for each item in task_list             │
-│  • Prune old jobs if history exceeds max_job_history     │
+│  • Prune in-memory jobs beyond max_job_history           │
 └────────────────────────────┬────────────────────────────┘
                              │
                              ▼
@@ -537,7 +503,8 @@ User Intent
 │    ├─ Inspect signature → pass inputs or call bare       │
 │    ├─ Execute in thread pool (run_in_executor)           │
 │    ├─ Apply timeout (asyncio.wait_for)                   │
-│    ├─ Track outcome (non-blocking)                       │
+│    ├─ Track returns / non-timeout exceptions             │
+│    ├─ Timeout: fail await; no tracker; thread may run    │
 │    └─ Set status: COMPLETED or FAILED                    │
 └────────────────────────────┬────────────────────────────┘
                              │
@@ -551,14 +518,13 @@ User Intent
 │  •                                                       │
 │  • If FAILED → enter retry loop:                         │
 │    ├─ Find retryable subtasks (attempts < max_attempts)  │
-│    ├─ Deduplicate protocols                              │
-│    ├─ mutate_protocol() on each (non-blocking)           │
-│    ├─ Re-execute failed subtasks in parallel             │
+│    ├─ SafeMutator per failed subtask (no deduplication)  │
+│    ├─ Retry in parallel (no semaphore / outcome tracker) │
 │    ├─ Re-reduce and re-verify                            │
 │    └─ Loop until PASSED or no retryable subtasks remain  │
 │  •                                                       │
-│  • If exhausted → escalate_to_human()                    │
-│  • Persist state to STATE.md                             │
+│  • If no retries remain → SlackEscalation.escalate()     │
+│  • Persist final SQLite snapshot + STATE.md summary      │
 └─────────────────────────────────────────────────────────┘
 ```
 
@@ -568,4 +534,5 @@ User Intent
 
 | Version | Date | Changes |
 | :--- | :--- | :--- |
-| 1.0.0 | 2026-06-22 | Initial release with all 8 review fixes applied |
+| 1.0.0 | 2026-06-22 | Initial module guide |
+| Documentation correction | 2026-10-29 | Reconciled invocation, examples, timeout/tracking, current signatures, verification and persistence guarantees against current master |
diff --git a/examples/mapreduce_local.py b/examples/mapreduce_local.py
new file mode 100644
index 0000000..b0c27e8
--- /dev/null
+++ b/examples/mapreduce_local.py
@@ -0,0 +1,145 @@
+"""Run the documented Map/Reduce patterns against real, temporary local files.
+
+Run from the repository root: python -m examples.mapreduce_local
+No provider, service, source mutation or Slack operation is needed.
+"""
+
+import asyncio
+import json
+import os
+from pathlib import Path
+import tempfile
+
+from orchestrator_mapreduce import HierarchicalOrchestrator, TaskStatus
+
+
+async def strict_gate(job):
+    failures = [
+        subtask
+        for subtask in job.subtasks
+        if subtask.status != TaskStatus.COMPLETED
+        or not isinstance(subtask.result, dict)
+        or subtask.result.get("success") is not True
+    ]
+    return {
+        "passed": bool(job.subtasks) and not failures,
+        "reason": f"{len(failures)} unsuccessful local subtasks",
+    }
+
+
+async def category_reducer(results, failures):
+    categories = {}
+    for result in results:
+        action = result.get("action", "unknown")
+        categories[action] = categories.get(action, 0) + 1
+    return {"categories": categories, "failures": len(failures), "results": results}
+
+
+class LocalEscalation:
+    async def escalate(self, job, reason):
+        raise RuntimeError(f"Local demo failed: {job.id}: {reason}")
+
+
+def make_orchestrator(output_dir, **kwargs):
+    output_dir = Path(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    orchestrator = HierarchicalOrchestrator(
+        verification_gate=strict_gate,
+        state_file=str(output_dir / "STATE.md"),
+        db_path=str(output_dir / "state.sqlite3"),
+        **kwargs,
+    )
+    # This local demonstration fails visibly instead of attempting Slack.
+    orchestrator.escalation = LocalEscalation()
+    return orchestrator
+
+
+def data_tasks(data_paths):
+    return [
+        {"protocol": "data_processor", "inputs": {"data_path": str(path)}}
+        for path in data_paths
+    ]
+
+
+async def minimal(data_path, output_dir):
+    return await make_orchestrator(output_dir).run(
+        "Count records in the supplied local directory", data_tasks([data_path])
+    )
+
+
+async def fan_out(data_paths, output_dir):
+    return await make_orchestrator(output_dir, max_concurrency=2).run(
+        "Count records separately in each supplied directory", data_tasks(data_paths)
+    )
+
+
+async def heterogeneous(data_path, output_dir):
+    # The existing log_analyzer reads LOG_DIR; it does not accept run_id/keyword.
+    return await make_orchestrator(output_dir).run(
+        "Count local data records and summarize the configured local logs",
+        data_tasks([data_path]) + [{"protocol": "log_analyzer", "inputs": {}}],
+    )
+
+
+async def custom_gate_and_reducer(data_path, output_dir):
+    return await make_orchestrator(output_dir, reducer=category_reducer).run(
+        "Group actual local data and log results by their action",
+        data_tasks([data_path]) + [{"protocol": "log_analyzer", "inputs": {}}],
+    )
+
+
+async def step_by_step(data_path, output_dir):
+    orchestrator = make_orchestrator(output_dir)
+    job = await orchestrator.plan(
+        "Inspect then count local records", data_tasks([data_path])
+    )
+    assert all(subtask.status == TaskStatus.PENDING for subtask in job.subtasks)
+    job = await orchestrator.map_execute(job)
+    assert all(subtask.status == TaskStatus.COMPLETED for subtask in job.subtasks)
+    return await orchestrator.reduce_verify(job)
+
+
+async def demonstrate(directory):
+    root = Path(directory)
+    data_paths = [root / "users", root / "orders"]
+    for path, rows in zip(data_paths, [[{"id": 1}], [{"id": 2}, {"id": 3}]]):
+        path.mkdir(parents=True)
+        (path / "records.json").write_text(json.dumps(rows))
+    log_dir = root / "input-logs"
+    log_dir.mkdir()
+    (log_dir / "service.log").write_text("[2026-10-29 09:00] INFO fixture loaded\n")
+    previous = os.environ.get("LOG_DIR")
+    os.environ["LOG_DIR"] = str(log_dir)
+    try:
+        return {
+            "minimal": await minimal(data_paths[0], root / "minimal"),
+            "fan_out": await fan_out(data_paths, root / "fan-out"),
+            "heterogeneous": await heterogeneous(data_paths[0], root / "mixed"),
+            "custom": await custom_gate_and_reducer(data_paths[0], root / "custom"),
+            "step_by_step": await step_by_step(data_paths[0], root / "steps"),
+        }
+    finally:
+        if previous is None:
+            os.environ.pop("LOG_DIR", None)
+        else:
+            os.environ["LOG_DIR"] = previous
+
+
+def main():
+    previous = Path.cwd()
+    with tempfile.TemporaryDirectory(prefix="mapreduce-demo-") as directory:
+        try:
+            # Existing tracker/logger write relative paths; contain these too.
+            os.chdir(directory)
+            jobs = asyncio.run(demonstrate(Path(directory) / "fixtures"))
+            print(
+                json.dumps(
+                    {name: job.reduced_result for name, job in jobs.items()}, indent=2
+                )
+            )
+        finally:
+            os.chdir(previous)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/orchestrator_mapreduce.py b/orchestrator_mapreduce.py
index d67093b..8f02900 100644
--- a/orchestrator_mapreduce.py
+++ b/orchestrator_mapreduce.py
@@ -6,17 +6,24 @@
 Extends the existing OrchestrationEngine with parallel fan-out,
 verification gateway, self-correcting mutation on failure,
 semantic LLM verification, structured state persistence,
-safe mutation hierarchy, and live Slack escalation.
+attempt-based mutation, and Slack CLI escalation.
+
+Worker threads share process state. Timeouts bound the wait, not thread lifetime;
+semantic verification falls back to mechanical approval on provider failure.
+See docs/orchestrator_mapreduce.md for the actual guarantees and limitations.
 
 Usage:
     from orchestrator_mapreduce import HierarchicalOrchestrator
 
-    orchestrator = HierarchicalOrchestrator(max_concurrency=10)
+    # These data directories must exist. Disable optional provider verification.
+    orchestrator = HierarchicalOrchestrator(
+        max_concurrency=10, use_semantic_gate=False
+    )
     job = await orchestrator.run(
-        intent="Validate all API endpoints after deployment",
+        intent="Process existing 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"}},
         ],
     )
 """
@@ -111,7 +118,8 @@ async def verify(self, job: OrchestratedJob) -> Dict[str, Any]:
         1. Mechanical: Check execution success rate (fast, cheap)
         2. Semantic: Ask LLM to evaluate output quality (deeper)
 
-        Both must pass for the job to be considered verified.
+        A provider/client failure falls back to mechanical approval; in that
+        case no semantic verdict is obtained.
         """
         # Layer 1: Mechanical gate (fast pre-filter)
         total = len(job.subtasks)
@@ -202,7 +210,8 @@ class StateStore:
     SQLite-backed structured state persistence.
     Provides queryable run history alongside the human-readable STATE.md.
 
-    Every job run is recorded with full metadata, enabling:
+    Final job/subtask snapshots replace rows by ID; attempts are not retained
+    as separate history. The stored final rows enable:
     - Querying past runs by protocol, status, or time range
     - Computing success rates over time
     - Identifying recurring failure patterns
@@ -369,10 +378,11 @@ class SafeMutator:
     Three-tier mutation hierarchy that escalates risk gradually:
       1. Mutate INPUTS (safest) — change what goes into the protocol
       2. Mutate STRATEGY (medium) — change how the protocol approaches the task
-      3. Mutate CODE (last resort) — rewrite the protocol, sandboxed
+      3. Mutate CODE (last resort) — invoke the legacy source mutator
 
-    Only escalates to the next tier if the previous tier has been
-    attempted and failed.
+    Selects a tier from the execution attempt count. Added keyword arguments
+    require a compatible protocol signature. The code tier does not provide
+    a process sandbox or rollback.
     """
 
     def __init__(self):
@@ -463,7 +473,8 @@ async def _mutate_code_sandboxed(
         """
         Tier 3: Mutate the protocol's source code.
         This is the most dangerous tier — only used as a last resort.
-        The mutation is tested in a sandboxed dry-run before being applied.
+        The existing mutator runs before a callable-task check. This is not a
+        sandbox or dry-run, and an imported module can remain cached.
         """
         # Perform the mutation via the existing mutator
         mutated = await loop.run_in_executor(None, mutate_protocol, protocol_name)
@@ -567,7 +578,7 @@ class HierarchicalOrchestrator:
     """
     Three-phase orchestration:
       1. PLAN  — Decompose intent into subtasks
-      2. MAP   — Execute subtasks in parallel (isolated state)
+      2. MAP   — Execute subtasks in parallel (shared process state)
       3. REDUCE — Collect, verify, and self-correct
 
     Now with:
@@ -658,8 +669,8 @@ async def plan(
 
     async def map_execute(self, job: OrchestratedJob) -> OrchestratedJob:
         """
-        Execute all subtasks in parallel with bounded concurrency.
-        Each subtask runs in isolation — no shared mutable state.
+        Execute subtasks with an initial-map semaphore and shared worker threads.
+        Inputs/module state are not isolated; timed-out threads may keep running.
         """
         job.status = TaskStatus.RUNNING
         semaphore = asyncio.Semaphore(self.max_concurrency)
@@ -834,8 +845,9 @@ async def _run_protocol_isolated(
         self, protocol_name: str, inputs: Dict[str, Any]
     ) -> Dict[str, Any]:
         """
-        Execute a protocol in isolation.
-        Passes inputs to the protocol task function.
+        Dispatch a synchronous protocol in the shared executor.
+        Pass all inputs as kwargs when the task has parameters; otherwise
+        ignore them. Coroutine return values are not awaited here.
         """
         from protocols.loader import load_protocol
 
@@ -897,7 +909,8 @@ def _prune_job_history(self):
 
     async def _default_gate(self, job: OrchestratedJob) -> Dict[str, Any]:
         """
-        Fallback mechanical gate: passes if >80% of subtasks succeeded.
+        Fallback mechanical gate: passes if at least 80% returned normally.
+        It does not inspect result success fields.
         Used only when use_semantic_gate=False and no custom gate provided.
         """
         total = len(job.subtasks)
diff --git a/tests/test_mapreduce_documentation.py b/tests/test_mapreduce_documentation.py
new file mode 100644
index 0000000..3d6cd28
--- /dev/null
+++ b/tests/test_mapreduce_documentation.py
@@ -0,0 +1,233 @@
+"""Execute the guide's local examples and its documented dispatch/timeout limits."""
+
+import asyncio
+import inspect
+import json
+import sqlite3
+import threading
+from pathlib import Path
+
+import pytest
+
+from examples.mapreduce_local import demonstrate, make_orchestrator
+import orchestrator_mapreduce as module
+from protocols import api_health_checker, data_processor, loader
+
+
+@pytest.fixture(autouse=True)
+def isolated_working_directory(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+
+    async def no_provider(*args, **kwargs):
+        pytest.fail("The local guide checks must not call a provider")
+
+    async def no_slack(*args, **kwargs):
+        pytest.fail("The local guide checks must not attempt Slack")
+
+    monkeypatch.setattr(module.SemanticGate, "_llm_evaluate", no_provider)
+    monkeypatch.setattr(module.SlackEscalation, "escalate", no_slack)
+
+
+@pytest.mark.asyncio
+async def test_all_five_local_patterns_process_actual_files_and_persist(tmp_path):
+    jobs = await demonstrate(tmp_path / "fixtures")
+    assert set(jobs) == {
+        "minimal",
+        "fan_out",
+        "heterogeneous",
+        "custom",
+        "step_by_step",
+    }
+    assert all(job.verification_passed for job in jobs.values())
+    assert all(job.status == module.TaskStatus.COMPLETED for job in jobs.values())
+    assert jobs["minimal"].subtasks[0].result["total_records"] == 1
+    assert [
+        subtask.result["total_records"] for subtask in jobs["fan_out"].subtasks
+    ] == [1, 2]
+    mixed = jobs["heterogeneous"].subtasks
+    assert mixed[0].result["action"] == "data_processing"
+    assert mixed[1].result["total_lines_analyzed"] == 1
+    assert jobs["custom"].reduced_result["categories"] == {
+        "data_processing": 1,
+        "log_analysis": 1,
+    }
+    directories = ["minimal", "fan-out", "mixed", "custom", "steps"]
+    for job, directory in zip(jobs.values(), directories):
+        saved = tmp_path / "fixtures" / directory
+        with sqlite3.connect(saved / "state.sqlite3") as conn:
+            row = conn.execute(
+                "SELECT status,verification_passed,subtask_count FROM jobs WHERE id=?",
+                (job.id,),
+            ).fetchone()
+        assert row == ("completed", 1, len(job.subtasks))
+        assert f"Job: {job.id}" in (saved / "STATE.md").read_text()
+
+
+@pytest.mark.asyncio
+async def test_signature_dispatch_defaults_full_kwargs_and_tracking(
+    tmp_path, monkeypatch
+):
+    def zero():
+        return {"success": True, "source": "zero"}
+
+    def required(value, scale=2):
+        return {"success": True, "value": value * scale}
+
+    def kwargs(**inputs):
+        return {"success": True, "received": inputs}
+
+    monkeypatch.setattr(
+        loader,
+        "load_protocol",
+        lambda name: {
+            "task": {"zero": zero, "required": required, "kwargs": kwargs}[name]
+        },
+    )
+    orchestrator = make_orchestrator(tmp_path / "state")
+    job = await orchestrator.plan(
+        "Dispatch supplied arguments",
+        [
+            {"protocol": "zero", "inputs": {"ignored": "yes"}},
+            {"protocol": "required", "inputs": {"value": 3}},
+            {"protocol": "required", "inputs": {"value": 3, "unexpected": True}},
+            {"protocol": "required", "inputs": {}},
+            {"protocol": "kwargs", "inputs": {"value": 4, "extra": True}},
+        ],
+    )
+    await orchestrator.map_execute(job)
+    assert job.subtasks[0].result == {"success": True, "source": "zero"}
+    assert job.subtasks[1].result["value"] == 6
+    assert job.subtasks[2].status == module.TaskStatus.FAILED
+    assert "unexpected" in job.subtasks[2].error
+    assert job.subtasks[3].status == module.TaskStatus.FAILED
+    assert "value" in job.subtasks[3].error
+    assert job.subtasks[4].result["received"] == {"value": 4, "extra": True}
+    tracked = [
+        json.loads(line)
+        for line in Path("memory/required.json").read_text().splitlines()
+    ]
+    assert len(tracked) == 3
+    assert sum(result["success"] is True for result in tracked) == 1
+
+
+@pytest.mark.asyncio
+async def test_timeout_does_not_track_or_terminate_running_worker(
+    tmp_path, monkeypatch
+):
+    started, release, finished = threading.Event(), threading.Event(), threading.Event()
+
+    def blocking():
+        started.set()
+        release.wait(2)
+        finished.set()
+        return {"success": True}
+
+    monkeypatch.setattr(loader, "load_protocol", lambda name: {"task": blocking})
+    orchestrator = make_orchestrator(tmp_path / "state", subtask_timeout=0.03)
+    job = await orchestrator.plan("Observe timeout", [{"protocol": "blocking"}])
+    try:
+        await orchestrator.map_execute(job)
+        assert started.is_set()
+        assert job.subtasks[0].status == module.TaskStatus.FAILED
+        assert "timed out" in job.subtasks[0].error
+        assert not finished.is_set()
+        assert not Path("memory/blocking.json").exists()
+    finally:
+        release.set()
+        assert await asyncio.to_thread(finished.wait, 2)
+    assert job.subtasks[0].status == module.TaskStatus.FAILED
+    assert job.subtasks[0].result is None
+    assert not Path("memory/blocking.json").exists()
+
+
+@pytest.mark.asyncio
+async def test_normal_failure_return_is_completed_and_mechanical_gate_passes(
+    tmp_path, monkeypatch
+):
+    monkeypatch.setattr(
+        loader, "load_protocol", lambda name: {"task": lambda: {"success": False}}
+    )
+    orchestrator = make_orchestrator(tmp_path / "state")
+    job = await orchestrator.plan(
+        "Observe return status", [{"protocol": "failure_return"}]
+    )
+    await orchestrator.map_execute(job)
+    assert job.subtasks[0].status == module.TaskStatus.COMPLETED
+    assert (await orchestrator._default_gate(job))["passed"] is True
+    assert (await orchestrator.verification_gate(job))["passed"] is False
+
+
+@pytest.mark.asyncio
+async def test_retry_dispatch_is_not_outcome_tracked(tmp_path, monkeypatch):
+    def task(**kwargs):
+        return {"success": True, "received": kwargs}
+
+    monkeypatch.setattr(loader, "load_protocol", lambda name: {"task": task})
+    orchestrator = make_orchestrator(tmp_path / "state")
+    subtask = module.SubTask(
+        "retry", "retry_fixture", {}, status=module.TaskStatus.FAILED, attempts=1
+    )
+    await orchestrator.mutator.mutate("retry_fixture", subtask, "fixture failure")
+    await orchestrator._retry_subtask(subtask)
+    assert subtask.status == module.TaskStatus.COMPLETED
+    assert subtask.result["received"]["_retry"] is True
+    assert subtask.attempts == 2
+    assert not Path("memory/retry_fixture.json").exists()
+
+
+@pytest.mark.asyncio
+async def test_async_protocol_result_is_not_awaited(tmp_path, monkeypatch):
+    invoked = []
+
+    async def task():
+        invoked.append(True)
+        return {"success": True}
+
+    monkeypatch.setattr(loader, "load_protocol", lambda name: {"task": task})
+    orchestrator = make_orchestrator(tmp_path / "state")
+    result = await orchestrator._run_protocol_isolated("async_fixture", {})
+    try:
+        assert inspect.iscoroutine(result)
+        assert invoked == []
+    finally:
+        result.close()
+
+
+def test_current_builtin_signatures_and_missing_data_directory():
+    assert list(inspect.signature(api_health_checker.task).parameters) == [
+        "service",
+        "url",
+    ]
+    assert list(inspect.signature(data_processor.task).parameters) == [
+        "data_path",
+        "source",
+    ]
+    with pytest.raises(ValueError, match="does not exist"):
+        data_processor.task(data_path="/does-not-exist-mapreduce-guide")
+
+
+@pytest.mark.asyncio
+async def test_semantic_provider_failure_returns_documented_mechanical_fallback(
+    monkeypatch,
+):
+    calls = []
+
+    async def unavailable(self, job):
+        calls.append(job.id)
+        raise RuntimeError("controlled unavailable provider")
+
+    monkeypatch.setattr(module.SemanticGate, "_llm_evaluate", unavailable)
+    job = module.OrchestratedJob(
+        "fallback",
+        "Observe fallback",
+        subtasks=[
+            module.SubTask("one", "fixture", {}, status=module.TaskStatus.COMPLETED)
+        ],
+    )
+    result = await module.SemanticGate().verify(job)
+    assert result == {
+        "passed": True,
+        "layer": "mechanical_fallback",
+        "success_rate": 1.0,
+    }
+    assert calls == [job.id]