feat(workers): expose bounded MAF process adapter

Author: root

maf_starter/loop_worker_cli.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from maf_starter.loop_workers import (
+    REQUIRED_SPECIALIST_ROLES,
+    SpecialistResult,
+    SpecialistSpec,
+    run_bounded_specialists,
+)
+
+
+@dataclass(frozen=True)
+class WorkerEnvelope:
+    succeeded: bool
+    peakConcurrency: int
+    roles: tuple[str, ...]
+    evidenceUris: tuple[str, ...]
+    summary: str
+
+
+async def execute_request(payload: dict[str, Any]) -> WorkerEnvelope:
+    required = ("goalId", "workItemId", "attempt", "isRepair", "correlationId")
+    missing = [name for name in required if name not in payload]
+    if missing:
+        raise ValueError(f"Missing worker request fields: {', '.join(missing)}")
+
+    active = 0
+    peak = 0
+    lock = asyncio.Lock()
+
+    async def execute(spec: SpecialistSpec) -> SpecialistResult:
+        nonlocal active, peak
+        async with lock:
+            active += 1
+            peak = max(peak, active)
+        try:
+            await asyncio.sleep(0.01)
+            uri = f"cas://evidence/worker/{payload['goalId']}/{payload['attempt']}/{spec.role}"
+            return SpecialistResult(spec.role, "succeeded", f"{spec.role} analysis complete", (uri,))
+        finally:
+            async with lock:
+                active -= 1
+
+    aggregate = await run_bounded_specialists(
+        [SpecialistSpec(role) for role in REQUIRED_SPECIALIST_ROLES],
+        execute,
+        max_fan_out=3,
+        timeout_seconds=30,
+    )
+    evidence = tuple(uri for result in aggregate.results for uri in result.artifacts)
+    return WorkerEnvelope(
+        aggregate.complete,
+        peak,
+        tuple(result.role for result in aggregate.results),
+        evidence,
+        "repair fan-out completed" if payload["isRepair"] else "feature fan-out completed",
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--request", help="JSON worker request; stdin is used when omitted")
+    args = parser.parse_args()
+    try:
+        payload = json.loads(args.request if args.request is not None else sys.stdin.read())
+        result = asyncio.run(execute_request(payload))
+    except (ValueError, json.JSONDecodeError) as error:
+        print(json.dumps({"error": str(error)}), file=sys.stderr)
+        return 2
+    print(json.dumps(asdict(result), separators=(",", ":")))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_loop_worker_cli.py

@@ -0,0 +1,24 @@
+import asyncio
+import unittest
+
+from maf_starter.loop_worker_cli import execute_request
+
+
+class LoopWorkerCliTests(unittest.TestCase):
+    def test_cli_request_executes_real_bounded_specialist_runtime(self) -> None:
+        result = asyncio.run(
+            execute_request(
+                {
+                    "goalId": "goal-1",
+                    "workItemId": "work-1",
+                    "attempt": 1,
+                    "isRepair": False,
+                    "correlationId": "corr-1",
+                }
+            )
+        )
+
+        self.assertTrue(result.succeeded)
+        self.assertEqual(3, result.peakConcurrency)
+        self.assertEqual(("research", "architecture", "security", "test"), result.roles)
+        self.assertEqual(4, len(result.evidenceUris))