diff --git a/maf_starter/approval_policy.py b/maf_starter/approval_policy.py index cdd9c2c..c17db24 100644 --- a/maf_starter/approval_policy.py +++ b/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, *, diff --git a/maf_starter/loop_sandbox.py b/maf_starter/loop_sandbox.py new file mode 100644 index 0000000..1776971 --- /dev/null +++ b/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) diff --git a/maf_starter/loop_worker_cli.py b/maf_starter/loop_worker_cli.py new file mode 100644 index 0000000..0d2e30a --- /dev/null +++ b/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()) diff --git a/maf_starter/loop_workers.py b/maf_starter/loop_workers.py new file mode 100644 index 0000000..74cce82 --- /dev/null +++ b/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() diff --git a/requirements.txt b/requirements.txt index 7df677e..a22d802 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/test_loop_sandbox.py b/tests/test_loop_sandbox.py new file mode 100644 index 0000000..09ef566 --- /dev/null +++ b/tests/test_loop_sandbox.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import shutil +import subprocess +import tempfile +import unittest +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from maf_starter.approval_policy import classify_external_action, require_action_approval +from maf_starter.loop_sandbox import GitWorktreeSandbox, MutationContract + + +class LoopSandboxTests(unittest.TestCase): + def setUp(self) -> None: + self.root = Path(tempfile.mkdtemp(prefix="cas-sandbox-")) + self.repo = self.root / "repo" + self.repo.mkdir() + self._git("init", "-b", "main") + self._git("config", "user.email", "test@example.invalid") + self._git("config", "user.name", "CAS Test") + (self.repo / "README.md").write_text("baseline\n", encoding="utf-8") + self._git("add", "README.md") + self._git("commit", "-m", "baseline") + + def tearDown(self) -> None: + shutil.rmtree(self.root, ignore_errors=True) + + def test_mutating_item_gets_distinct_worktree_manifest_and_preserves_default_sha(self) -> None: + before = self._git("rev-parse", "main").stdout.strip() + contract = self._contract() + sandbox = GitWorktreeSandbox(contract) + + manifest = sandbox.create() + sandbox.authorize_mutation("implementer") + target = sandbox.resolve_write_path("src/change.txt") + target.parent.mkdir(parents=True) + target.write_text("change\n", encoding="utf-8") + artifact = sandbox.capture_artifacts() + + self.assertTrue(contract.worktree_path.is_dir()) + self.assertNotEqual(contract.worktree_path.resolve(), self.repo.resolve()) + self.assertEqual(manifest.base_sha, before) + self.assertEqual(artifact.changed_files, ("src/change.txt",)) + self.assertEqual(self._git("rev-parse", "main").stdout.strip(), before) + self.assertEqual(artifact.idempotency_key, "goal-1:work-1") + + def test_owner_deadline_and_path_allowlist_are_enforced(self) -> None: + sandbox = GitWorktreeSandbox(self._contract()) + sandbox.create() + with self.assertRaises(PermissionError): + sandbox.authorize_mutation("research") + with self.assertRaises(ValueError): + sandbox.resolve_write_path("../escape.txt") + with self.assertRaises(ValueError): + sandbox.resolve_write_path("secrets/token.txt") + with self.assertRaises(TimeoutError): + GitWorktreeSandbox(self._contract(deadline=datetime.now(UTC) - timedelta(seconds=1))).create() + + def test_external_and_destructive_actions_cannot_run_before_approval(self) -> None: + for action in ("push", "deploy", "delete", "message", "production_mutation"): + decision = classify_external_action(action, "target") + self.assertTrue(decision.approval_required, action) + with self.assertRaises(PermissionError): + require_action_approval(action, "target", approved=False) + require_action_approval(action, "target", approved=True) + + def _contract(self, *, deadline: datetime | None = None) -> MutationContract: + return MutationContract( + goal_id="goal-1", + work_item_id="work-1", + repo_root=self.repo, + worktree_path=self.root / "worktrees" / "goal-1" / "work-1", + base_ref="main", + deadline=deadline or datetime.now(UTC) + timedelta(minutes=5), + path_allowlist=("src/**",), + idempotency_key="goal-1:work-1", + implementation_owner="implementer", + ) + + def _git(self, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run(["git", "-C", str(self.repo), *args], check=True, capture_output=True, text=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_loop_worker_cli.py b/tests/test_loop_worker_cli.py new file mode 100644 index 0000000..cfd92ec --- /dev/null +++ b/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)) diff --git a/tests/test_loop_workers.py b/tests/test_loop_workers.py new file mode 100644 index 0000000..3e7b89b --- /dev/null +++ b/tests/test_loop_workers.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import asyncio +import unittest + +from maf_starter.loop_workers import ( + REQUIRED_SPECIALIST_ROLES, + SpecialistResult, + SpecialistSpec, + aggregate_specialist_results, + build_maf_fanout_workflow, + run_bounded_specialists, +) + + +class LoopWorkerTests(unittest.IsolatedAsyncioTestCase): + async def test_four_roles_run_with_peak_concurrency_three_and_complete_fan_in(self) -> None: + 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) + await asyncio.sleep(0.05) + async with lock: + active -= 1 + return SpecialistResult(spec.role, "succeeded", f"{spec.role} complete", ()) + + aggregate = await run_bounded_specialists( + [SpecialistSpec(role) for role in REQUIRED_SPECIALIST_ROLES], + execute, + max_fan_out=3, + timeout_seconds=2, + ) + + self.assertEqual(peak, 3) + self.assertEqual(tuple(result.role for result in aggregate.results), REQUIRED_SPECIALIST_ROLES) + self.assertTrue(aggregate.complete) + + async def test_timeout_is_terminal_failure_not_partial_success(self) -> None: + async def execute(spec: SpecialistSpec) -> SpecialistResult: + await asyncio.sleep(1) + return SpecialistResult(spec.role, "succeeded", "late", ()) + + with self.assertRaises(TimeoutError): + await run_bounded_specialists( + [SpecialistSpec(role) for role in REQUIRED_SPECIALIST_ROLES], + execute, + max_fan_out=3, + timeout_seconds=0.01, + ) + + def test_aggregator_rejects_missing_duplicate_or_nonterminal_results(self) -> None: + complete = [SpecialistResult(role, "succeeded", "ok", ()) for role in REQUIRED_SPECIALIST_ROLES] + with self.assertRaises(ValueError): + aggregate_specialist_results(complete[:-1]) + with self.assertRaises(ValueError): + aggregate_specialist_results([*complete, complete[0]]) + with self.assertRaises(ValueError): + aggregate_specialist_results([*complete[:-1], SpecialistResult("test", "running", "", ())]) + + def test_specialist_specs_are_read_only_and_maf_builder_uses_native_fan_edges(self) -> None: + specs = [SpecialistSpec(role) for role in REQUIRED_SPECIALIST_ROLES] + self.assertTrue(all(spec.read_only for spec in specs)) + self.assertTrue(all("write" not in spec.capabilities for spec in specs)) + + class BuilderSpy: + def __init__(self) -> None: + self.fan_out = None + self.fan_in = None + def add_fan_out_edges(self, source, targets): + self.fan_out = (source, tuple(targets)); return self + def add_fan_in_edges(self, sources, target): + self.fan_in = (tuple(sources), target); return self + def build(self): return self + + builder = BuilderSpy() + result = build_maf_fanout_workflow("dispatcher", [1, 2, 3, 4], "aggregator", builder=builder) + self.assertIs(result, builder) + self.assertEqual(builder.fan_out, ("dispatcher", (1, 2, 3, 4))) + self.assertEqual(builder.fan_in, ((1, 2, 3, 4), "aggregator")) + + +if __name__ == "__main__": + unittest.main()