223 lines
8.0 KiB
Python
223 lines
8.0 KiB
Python
"""Autonomous self-loop executor.
|
|
|
|
This layer turns probe/plan reports into safe execution outcomes. It does not
|
|
bypass owner governance: high-risk or restricted actions are blocked at the owner
|
|
approval gate and emitted as explicit pending work.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable
|
|
|
|
from .autonomous_self_loop import AutonomousSelfLoop, RemediationStep, SelfLoopReport
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExecutionResult:
|
|
step_id: str
|
|
action: str
|
|
status: str
|
|
reason: str
|
|
risk: str
|
|
requires_owner_approval: bool
|
|
output: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TestResult:
|
|
name: str
|
|
status: str
|
|
detail: str = ""
|
|
|
|
def passed(self) -> bool:
|
|
return self.status == "pass"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass
|
|
class ExecutionReport:
|
|
status: str
|
|
report_id: str
|
|
actions: list[ExecutionResult]
|
|
tests: list[TestResult]
|
|
owner_approval_required: bool
|
|
created_at: str = field(default_factory=utc_now)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"status": self.status,
|
|
"report_id": self.report_id,
|
|
"actions": [a.to_dict() for a in self.actions],
|
|
"tests": [t.to_dict() for t in self.tests],
|
|
"owner_approval_required": self.owner_approval_required,
|
|
"created_at": self.created_at,
|
|
}
|
|
|
|
|
|
class SelfLoopTestRunner:
|
|
"""Deterministic self-loop checks that avoid shell execution and secrets."""
|
|
|
|
def run(self, runtime: dict[str, Any], baseline: dict[str, Any]) -> list[TestResult]:
|
|
return [
|
|
self._runtime_active(runtime),
|
|
self._patch_only_policy(baseline),
|
|
self._owner_gate_policy(baseline),
|
|
self._portal_domain_smoke(),
|
|
]
|
|
|
|
def _runtime_active(self, runtime: dict[str, Any]) -> TestResult:
|
|
if runtime.get("status") == "active":
|
|
return TestResult("runtime_active", "pass", "runtime reports active")
|
|
return TestResult("runtime_active", "fail", "runtime is not active")
|
|
|
|
def _patch_only_policy(self, baseline: dict[str, Any]) -> TestResult:
|
|
if baseline.get("update_mode") == "git_patch_only":
|
|
return TestResult("git_patch_only", "pass", "patch-only updates enforced")
|
|
return TestResult("git_patch_only", "fail", "update mode is not git_patch_only")
|
|
|
|
def _owner_gate_policy(self, baseline: dict[str, Any]) -> TestResult:
|
|
codex_gitops = baseline.get("codex_gitops", {})
|
|
if codex_gitops.get("direct_main_push") is False:
|
|
return TestResult("owner_gate", "pass", "direct main push disabled")
|
|
return TestResult("owner_gate", "fail", "direct main push must remain disabled")
|
|
|
|
def _portal_domain_smoke(self) -> TestResult:
|
|
try:
|
|
from .portal_rbac_ops import ROLE_OWNER, PortalOpsService, build_context
|
|
|
|
service = PortalOpsService()
|
|
result = service.owner_cockpit(build_context("owner", ROLE_OWNER))
|
|
if result.get("status") == "ok":
|
|
return TestResult("portal_domain_smoke", "pass", "owner cockpit domain service responds")
|
|
return TestResult("portal_domain_smoke", "fail", "owner cockpit returned non-ok")
|
|
except Exception as exc: # pragma: no cover - defensive smoke output
|
|
return TestResult("portal_domain_smoke", "fail", f"portal smoke failed: {type(exc).__name__}")
|
|
|
|
|
|
class SelfLoopAgentExecutor:
|
|
"""Executes safe self-loop steps and gates restricted ones."""
|
|
|
|
SAFE_ACTIONS = {
|
|
"health_probe",
|
|
"test_execution",
|
|
"failure_classification",
|
|
"patch_plan_generation",
|
|
"report_generation",
|
|
"create-low-risk-patch",
|
|
}
|
|
|
|
RESTRICTED_ACTIONS = {
|
|
"request-runtime-recovery",
|
|
"commit-policy-patch",
|
|
"commit-security-policy",
|
|
"commit-runtime-config",
|
|
"commit-portal-config",
|
|
"production-promotion",
|
|
"permission-escalation",
|
|
"secret-rotation",
|
|
}
|
|
|
|
def __init__(self, loop: AutonomousSelfLoop | None = None, test_runner: SelfLoopTestRunner | None = None):
|
|
self.loop = loop or AutonomousSelfLoop(owner_approval_required=True)
|
|
self.test_runner = test_runner or SelfLoopTestRunner()
|
|
self.handlers: dict[str, Callable[[RemediationStep], dict[str, Any]]] = {
|
|
"create-low-risk-patch": self._create_low_risk_patch_manifest,
|
|
"patch_plan_generation": self._patch_plan_manifest,
|
|
"report_generation": self._report_manifest,
|
|
}
|
|
|
|
def execute_step(self, step: RemediationStep) -> ExecutionResult:
|
|
if step.requires_owner_approval or step.action in self.RESTRICTED_ACTIONS:
|
|
return ExecutionResult(
|
|
step.id,
|
|
step.action,
|
|
"blocked_owner_gate",
|
|
"restricted action requires owner approval",
|
|
step.risk,
|
|
True,
|
|
{"title": step.title},
|
|
)
|
|
if step.risk not in {"low"}:
|
|
return ExecutionResult(
|
|
step.id,
|
|
step.action,
|
|
"blocked_risk_gate",
|
|
"only low-risk autonomous actions are allowed",
|
|
step.risk,
|
|
True,
|
|
{"title": step.title},
|
|
)
|
|
handler = self.handlers.get(step.action, self._safe_noop_manifest)
|
|
return ExecutionResult(
|
|
step.id,
|
|
step.action,
|
|
"executed",
|
|
"safe autonomous action executed",
|
|
step.risk,
|
|
False,
|
|
handler(step),
|
|
)
|
|
|
|
def execute_report(self, report: SelfLoopReport, runtime: dict[str, Any], baseline: dict[str, Any]) -> ExecutionReport:
|
|
actions = [self.execute_step(step) for step in report.remediation]
|
|
tests = self.test_runner.run(runtime, baseline)
|
|
owner_approval_required = self.loop.owner_approval_required or any(a.requires_owner_approval for a in actions)
|
|
failed_tests = [t for t in tests if not t.passed()]
|
|
if failed_tests:
|
|
status = "failed_tests"
|
|
elif owner_approval_required:
|
|
status = "waiting_owner_gate"
|
|
else:
|
|
status = "completed"
|
|
return ExecutionReport(status, report.id, actions, tests, owner_approval_required)
|
|
|
|
def run_once(self, runtime: dict[str, Any], baseline: dict[str, Any]) -> dict[str, Any]:
|
|
report = self.loop.evaluate(runtime, baseline)
|
|
execution = self.execute_report(report, runtime, baseline)
|
|
acceptance = self.loop.acceptance(report)
|
|
owner_approval_required = execution.owner_approval_required or acceptance["owner_approval_required"]
|
|
return {
|
|
"status": execution.status,
|
|
"can_unattended_continue": execution.status == "completed" and not owner_approval_required,
|
|
"owner_approval_required": owner_approval_required,
|
|
"plan": report.to_dict(),
|
|
"execution": execution.to_dict(),
|
|
}
|
|
|
|
def _create_low_risk_patch_manifest(self, step: RemediationStep) -> dict[str, Any]:
|
|
return {
|
|
"patch_intent": step.title,
|
|
"branch_policy": "patch-only",
|
|
"direct_main_push": False,
|
|
"secrets_in_git": False,
|
|
}
|
|
|
|
def _patch_plan_manifest(self, step: RemediationStep) -> dict[str, Any]:
|
|
return {"plan": step.title, "action": step.action}
|
|
|
|
def _report_manifest(self, step: RemediationStep) -> dict[str, Any]:
|
|
return {"report": step.title, "generated": True}
|
|
|
|
def _safe_noop_manifest(self, step: RemediationStep) -> dict[str, Any]:
|
|
return {"noop": True, "title": step.title}
|
|
|
|
|
|
__all__ = [
|
|
"ExecutionResult",
|
|
"TestResult",
|
|
"ExecutionReport",
|
|
"SelfLoopTestRunner",
|
|
"SelfLoopAgentExecutor",
|
|
]
|