Add multi-agent self-loop orchestrator

active-runtime/app/self_loop_orchestrator.py

@@ -0,0 +1,142 @@
+"""Multi-agent self-loop orchestration.
+
+Coordinates planning, implementation proposals, and validation feedback from
+Hermes/OpenClaw/CodexBridge-style agents without storing credentials or bypassing
+GitOps governance.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+from .self_loop_executor import SelfLoopAgentExecutor
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+@dataclass(frozen=True)
+class AgentProfile:
+    name: str
+    role: str
+    weight: int
+    capabilities: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class AgentTask:
+    task_id: str
+    goal: str
+    task_type: str
+    risk: str = "low"
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class AgentResponse:
+    agent: str
+    status: str
+    proposal: dict[str, Any]
+    confidence: float
+
+
+@dataclass(frozen=True)
+class AgentVote:
+    agent: str
+    approve: bool
+    weight: int
+    reason: str
+
+
+@dataclass
+class PatchManifest:
+    branch_policy: str
+    files: list[str]
+    tests: list[str]
+    approvals_required: bool
+    generated_at: str = field(default_factory=utc_now)
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+class SelfLoopAgentOrchestrator:
+    AGENTS = (
+        AgentProfile("Hermes", "planner", 2, ("analysis", "planning", "risk")),
+        AgentProfile("OpenClaw", "implementation", 2, ("code", "patch", "test")),
+        AgentProfile("CodexBridge", "validation", 2, ("gitops", "validation", "acceptance")),
+    )
+
+    def __init__(self, executor: SelfLoopAgentExecutor | None = None):
+        self.executor = executor or SelfLoopAgentExecutor()
+
+    def dispatch(self, task: AgentTask) -> list[AgentResponse]:
+        responses = []
+        for agent in self.AGENTS:
+            if agent.role == "planner":
+                proposal = {"risk": task.risk, "plan": "classify issue and produce remediation steps"}
+            elif agent.role == "implementation":
+                proposal = {"patch": True, "branch_policy": "patch-only", "goal": task.goal}
+            else:
+                proposal = {"validation": ["tests", "gitops_policy", "acceptance_report"]}
+            responses.append(AgentResponse(agent.name, "ready", proposal, 0.9))
+        return responses
+
+    def vote(self, responses: list[AgentResponse]) -> dict[str, Any]:
+        votes = []
+        total = 0
+        approved = 0
+        for response in responses:
+            profile = next(a for a in self.AGENTS if a.name == response.agent)
+            approve = response.status == "ready"
+            votes.append(AgentVote(response.agent, approve, profile.weight, "ready" if approve else "rejected"))
+            total += profile.weight
+            approved += profile.weight if approve else 0
+        return {
+            "approved": approved >= total / 2,
+            "approval_weight": approved,
+            "total_weight": total,
+            "votes": [asdict(v) for v in votes],
+        }
+
+    def create_patch_manifest(self, task: AgentTask, requires_approval: bool) -> PatchManifest:
+        return PatchManifest(
+            branch_policy="patch-only",
+            files=[
+                "active-runtime/app",
+                "openapi",
+                "tests",
+            ],
+            tests=[
+                "self_loop_executor",
+                "portal_self_loop",
+                "acceptance_pipeline",
+            ],
+            approvals_required=requires_approval or task.risk != "low",
+        )
+
+    def run_cycle(self, runtime: dict[str, Any], baseline: dict[str, Any], goal: str) -> dict[str, Any]:
+        task = AgentTask("self-loop-cycle", goal, "remediation")
+        executor_result = self.executor.run_once(runtime, baseline)
+        responses = self.dispatch(task)
+        decision = self.vote(responses)
+        manifest = self.create_patch_manifest(task, executor_result["owner_approval_required"])
+        return {
+            "status": "ready_for_patch" if decision["approved"] else "blocked_agent_vote",
+            "executor": executor_result,
+            "agents": [asdict(r) for r in responses],
+            "vote": decision,
+            "patch_manifest": manifest.to_dict(),
+        }
+
+
+__all__ = [
+    "AgentProfile",
+    "AgentTask",
+    "AgentResponse",
+    "SelfLoopAgentOrchestrator",
+    "PatchManifest",
+]