Add bounded MAF fan-out and isolated worker execution (#3)

* test(workers): cover bounded fanout and sandbox policy

* feat(workers): add isolated MAF task execution

* feat(workers): expose bounded MAF process adapter

* fix(ci): pin compatible agent framework orchestration

---------

Co-authored-by: root <root@Kimi.localdomain>

Author: OgeonX-Ai

maf_starter/approval_policy.py

@@ -142,6 +142,35 @@ def is_execution_approved(prompt: str, approval_word: str) -> bool:
     return normalized == approval_token or normalized.startswith("approve") or normalized.startswith("approved") or normalized.startswith("proceed")
 
 
+def classify_external_action(action: str, target: str) -> ExecutionRiskDecision:
+    normalized = str(action or "").strip().lower()
+    external = {"push", "deploy", "message", "production_mutation"}
+    destructive = {"delete"}
+    if normalized in external:
+        return _decision(
+            "externally_visible",
+            action_kind=normalized,
+            reason=f"{normalized} requires deterministic approval before execution.",
+            external_targets=[target],
+        )
+    if normalized in destructive:
+        return _decision(
+            "destructive",
+            action_kind=normalized,
+            reason=f"{normalized} requires deterministic approval before execution.",
+            external_targets=[target],
+        )
+    return _decision("blocked", action_kind=normalized or "unknown", reason="Unknown external action is blocked.")
+
+
+def require_action_approval(action: str, target: str, *, approved: bool) -> None:
+    decision = classify_external_action(action, target)
+    if decision.blocked:
+        raise PermissionError(decision.scope.reason)
+    if decision.approval_required and not approved:
+        raise PermissionError(decision.scope.reason)
+
+
 def _decision(
     classification: RiskLevel,
     *,

maf_starter/loop_sandbox.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import re
+import subprocess
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path, PurePosixPath
+
+
+_SAFE_ID = re.compile(r"^[A-Za-z0-9_-]+$")
+_BLOCKED_PARTS = {".git", ".env", "secrets", "credentials", "state"}
+
+
+@dataclass(frozen=True)
+class MutationContract:
+    goal_id: str
+    work_item_id: str
+    repo_root: Path
+    worktree_path: Path
+    base_ref: str
+    deadline: datetime
+    path_allowlist: tuple[str, ...]
+    idempotency_key: str
+    implementation_owner: str
+
+    def validate(self, *, now: datetime | None = None) -> None:
+        current = now or datetime.now(UTC)
+        if not _SAFE_ID.fullmatch(self.goal_id) or not _SAFE_ID.fullmatch(self.work_item_id):
+            raise ValueError("Goal and work-item IDs must be path-safe")
+        if not self.idempotency_key.strip() or not self.implementation_owner.strip():
+            raise ValueError("Idempotency key and implementation owner are required")
+        if not self.path_allowlist:
+            raise ValueError("At least one allowed path pattern is required")
+        if self.deadline.tzinfo is None or self.deadline <= current:
+            raise TimeoutError("Mutation work item deadline has expired")
+        repo = self.repo_root.resolve()
+        worktree = self.worktree_path.resolve()
+        if repo == worktree or repo in worktree.parents:
+            raise ValueError("Mutation worktree must be distinct from and outside the source repository")
+
+
+@dataclass(frozen=True)
+class SandboxManifest:
+    goal_id: str
+    work_item_id: str
+    worktree_path: str
+    branch: str
+    base_sha: str
+    idempotency_key: str
+    implementation_owner: str
+
+
+@dataclass(frozen=True)
+class ArtifactManifest:
+    goal_id: str
+    work_item_id: str
+    idempotency_key: str
+    changed_files: tuple[str, ...]
+    worktree_path: str
+
+
+class GitWorktreeSandbox:
+    def __init__(self, contract: MutationContract) -> None:
+        self.contract = contract
+        self._manifest: SandboxManifest | None = None
+
+    def create(self) -> SandboxManifest:
+        self.contract.validate()
+        if self.contract.worktree_path.exists():
+            raise FileExistsError(f"Worktree already exists: {self.contract.worktree_path}")
+        self.contract.worktree_path.parent.mkdir(parents=True, exist_ok=True)
+        base_sha = self._git_source("rev-parse", self.contract.base_ref).stdout.strip()
+        branch = f"codex/{self.contract.goal_id}-{self.contract.work_item_id}"
+        self._git_source("worktree", "add", "-b", branch, str(self.contract.worktree_path), base_sha)
+        self._manifest = SandboxManifest(
+            self.contract.goal_id,
+            self.contract.work_item_id,
+            str(self.contract.worktree_path.resolve()),
+            branch,
+            base_sha,
+            self.contract.idempotency_key,
+            self.contract.implementation_owner,
+        )
+        return self._manifest
+
+    def authorize_mutation(self, actor: str) -> None:
+        self.contract.validate()
+        if actor != self.contract.implementation_owner:
+            raise PermissionError(f"Mutation owner is {self.contract.implementation_owner}, not {actor}")
+
+    def resolve_write_path(self, relative_path: str) -> Path:
+        self.contract.validate()
+        if self._manifest is None:
+            raise RuntimeError("Sandbox has not been created")
+        normalized = relative_path.replace("\\", "/").lstrip("./")
+        pure = PurePosixPath(normalized)
+        if not normalized or pure.is_absolute() or ".." in pure.parts or any(part.lower() in _BLOCKED_PARTS for part in pure.parts):
+            raise ValueError(f"Write path is blocked: {relative_path}")
+        if not any(pure.match(pattern) for pattern in self.contract.path_allowlist):
+            raise ValueError(f"Write path is outside the allowlist: {relative_path}")
+        target = (self.contract.worktree_path / Path(*pure.parts)).resolve()
+        if self.contract.worktree_path.resolve() not in target.parents:
+            raise ValueError(f"Write path escapes the worktree: {relative_path}")
+        return target
+
+    def capture_artifacts(self) -> ArtifactManifest:
+        if self._manifest is None:
+            raise RuntimeError("Sandbox has not been created")
+        result = subprocess.run(
+            ["git", "-C", str(self.contract.worktree_path), "status", "--porcelain", "--untracked-files=all"],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        changed = tuple(sorted(line[3:].strip() for line in result.stdout.splitlines() if len(line) >= 4))
+        return ArtifactManifest(
+            self.contract.goal_id,
+            self.contract.work_item_id,
+            self.contract.idempotency_key,
+            changed,
+            str(self.contract.worktree_path.resolve()),
+        )
+
+    def _git_source(self, *args: str) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(["git", "-C", str(self.contract.repo_root), *args], check=True, capture_output=True, text=True)

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())

maf_starter/loop_workers.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable, Sequence
+from dataclasses import dataclass
+from typing import Any, Literal
+
+
+REQUIRED_SPECIALIST_ROLES = ("research", "architecture", "security", "test")
+TerminalStatus = Literal["succeeded", "failed", "cancelled", "timed_out"]
+
+
+@dataclass(frozen=True)
+class SpecialistSpec:
+    role: str
+    read_only: bool = True
+    capabilities: tuple[str, ...] = ("read_repo", "search_repo")
+
+    def __post_init__(self) -> None:
+        if self.role not in REQUIRED_SPECIALIST_ROLES:
+            raise ValueError(f"Unsupported specialist role: {self.role}")
+        if not self.read_only or any(capability.startswith(("write", "delete", "execute")) for capability in self.capabilities):
+            raise ValueError("Specialists must remain read-only")
+
+
+@dataclass(frozen=True)
+class SpecialistResult:
+    role: str
+    status: str
+    summary: str
+    artifacts: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class SpecialistAggregate:
+    results: tuple[SpecialistResult, ...]
+
+    @property
+    def complete(self) -> bool:
+        return all(result.status == "succeeded" for result in self.results)
+
+
+def aggregate_specialist_results(results: Sequence[SpecialistResult]) -> SpecialistAggregate:
+    by_role: dict[str, SpecialistResult] = {}
+    for result in results:
+        if result.role in by_role:
+            raise ValueError(f"Duplicate specialist result: {result.role}")
+        if result.status not in {"succeeded", "failed", "cancelled", "timed_out"}:
+            raise ValueError(f"Specialist result is not terminal: {result.role}")
+        by_role[result.role] = result
+    missing = set(REQUIRED_SPECIALIST_ROLES) - set(by_role)
+    extra = set(by_role) - set(REQUIRED_SPECIALIST_ROLES)
+    if missing or extra:
+        raise ValueError(f"Specialist fan-in mismatch; missing={sorted(missing)}, extra={sorted(extra)}")
+    return SpecialistAggregate(tuple(by_role[role] for role in REQUIRED_SPECIALIST_ROLES))
+
+
+async def run_bounded_specialists(
+    specs: Sequence[SpecialistSpec],
+    execute: Callable[[SpecialistSpec], Awaitable[SpecialistResult]],
+    *,
+    max_fan_out: int = 3,
+    timeout_seconds: float = 1800,
+) -> SpecialistAggregate:
+    if max_fan_out < 1:
+        raise ValueError("max_fan_out must be positive")
+    if timeout_seconds <= 0:
+        raise ValueError("timeout_seconds must be positive")
+    roles = tuple(spec.role for spec in specs)
+    if len(set(roles)) != len(roles) or set(roles) != set(REQUIRED_SPECIALIST_ROLES):
+        raise ValueError("Exactly one specialist for every required role is required")
+
+    semaphore = asyncio.Semaphore(max_fan_out)
+
+    async def invoke(spec: SpecialistSpec) -> SpecialistResult:
+        async with semaphore:
+            result = await execute(spec)
+            if result.role != spec.role:
+                raise ValueError(f"Specialist {spec.role} returned result for {result.role}")
+            return result
+
+    try:
+        async with asyncio.timeout(timeout_seconds):
+            results = await asyncio.gather(*(invoke(spec) for spec in specs))
+    except TimeoutError:
+        raise TimeoutError("Specialist fan-out exceeded its task-attempt deadline") from None
+    return aggregate_specialist_results(results)
+
+
+def build_maf_fanout_workflow(
+    dispatcher: Any,
+    specialists: Sequence[Any],
+    aggregator: Any,
+    *,
+    builder: Any | None = None,
+    max_iterations: int = 3,
+) -> Any:
+    if len(specialists) != len(REQUIRED_SPECIALIST_ROLES):
+        raise ValueError("MAF workflow requires all four specialist executors")
+    if builder is None:
+        from agent_framework import WorkflowBuilder
+
+        builder = WorkflowBuilder(
+            start_executor=dispatcher,
+            max_iterations=max_iterations,
+            name="cas_task_worker",
+            description="Bounded read-only specialist fan-out with typed terminal fan-in",
+        )
+    return builder.add_fan_out_edges(dispatcher, specialists).add_fan_in_edges(specialists, aggregator).build()

requirements.txt

@@ -1,5 +1,6 @@
 agent-framework==1.0.0rc5
 agent-framework-devui==1.0.0b260319
+agent-framework-orchestrations==1.0.0b260319
 autogen-agentchat>=0.7.5,<0.8.0
 autogen-core>=0.7.5,<0.8.0
 autogen-ext[anthropic,ollama,openai]>=0.7.5,<0.8.0