diff --git a/docs/orchestrator_mapreduce.md b/docs/orchestrator_mapreduce.md index a8a2852..40b6770 100644 --- a/docs/orchestrator_mapreduce.md +++ b/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` | `/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 │ ▼ diff --git a/orchestrator_mapreduce.py b/orchestrator_mapreduce.py index f5f6111..d67093b 100644 --- a/orchestrator_mapreduce.py +++ b/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: