658 lines
25 KiB
Python
658 lines
25 KiB
Python
"""Deployment choreography + namespace lifecycle for Hermes Enterprise.
|
|
|
|
The Controller owns every state transition in the control plane. Drivers and
|
|
adapters consume admitted intent at their boundary; they never own resources
|
|
or select themselves. The single deploy path implemented here is:
|
|
|
|
authorize -> resolve references -> per-reference authorize
|
|
-> namespace gateway ready -> sandbox supports(ENTIRE policy)
|
|
-> immutable AgentRevision snapshot -> provision candidate
|
|
-> enforce + verify containment (BEFORE start)
|
|
-> prepare non-serving route -> retire previous revision
|
|
-> activate candidate (sole active) -> start harness -> enable route
|
|
|
|
Every failure before activation leaves the previous revision serving,
|
|
marks the candidate Failed, and tears its workload down. A failure after
|
|
activation triggers rollback to the previous revision; an unverifiable
|
|
rollback raises RollbackError and leaves the agent inactive with all
|
|
routes disabled.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
from .audit import AuditLog
|
|
from .contracts import (
|
|
AuthzRequest,
|
|
ComputeDriver,
|
|
DriverRegistry,
|
|
IAMAdapter,
|
|
SandboxDriver,
|
|
WorkloadRef,
|
|
)
|
|
from .errors import (
|
|
AuthorizationError,
|
|
DeploymentError,
|
|
NotFoundError,
|
|
RollbackError,
|
|
)
|
|
from .resources import (
|
|
Kind,
|
|
NamespacePhase,
|
|
Resource,
|
|
ResourceMeta,
|
|
RevisionPhase,
|
|
)
|
|
from .store import ResourceStore
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gateway management
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class GatewayManager(ABC):
|
|
"""Manages the per-namespace ingress gateway and per-revision routes.
|
|
|
|
Routes are prepared non-serving so activation is a flip, not a build.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def ensure_gateway(self, namespace: str) -> bool:
|
|
"""Ensure the namespace gateway exists; return True when ready."""
|
|
|
|
@abstractmethod
|
|
def prepare_route(self, namespace: str, revision_uid: str) -> None:
|
|
"""Create a non-serving route for the revision. Must not serve."""
|
|
|
|
@abstractmethod
|
|
def enable_route(self, namespace: str, revision_uid: str) -> None:
|
|
"""Begin serving traffic to the revision's route."""
|
|
|
|
@abstractmethod
|
|
def disable_route(self, namespace: str, revision_uid: str) -> None:
|
|
"""Stop serving traffic to the revision's route."""
|
|
|
|
@abstractmethod
|
|
def teardown_gateway(self, namespace: str) -> None:
|
|
"""Remove the namespace gateway and all its routes."""
|
|
|
|
|
|
class InMemoryGatewayManager(GatewayManager):
|
|
"""In-process gateway manager for tests and local development."""
|
|
|
|
def __init__(self) -> None:
|
|
self.gateways: set[str] = set()
|
|
# (namespace, revision_uid) -> serving?
|
|
self.routes: dict[tuple[str, str], bool] = {}
|
|
self.calls: list[tuple[str, ...]] = []
|
|
|
|
def ensure_gateway(self, namespace: str) -> bool:
|
|
self.calls.append(("ensure_gateway", namespace))
|
|
self.gateways.add(namespace)
|
|
return True
|
|
|
|
def prepare_route(self, namespace: str, revision_uid: str) -> None:
|
|
self.calls.append(("prepare_route", namespace, revision_uid))
|
|
self.routes[(namespace, revision_uid)] = False
|
|
|
|
def enable_route(self, namespace: str, revision_uid: str) -> None:
|
|
self.calls.append(("enable_route", namespace, revision_uid))
|
|
self.routes[(namespace, revision_uid)] = True
|
|
|
|
def disable_route(self, namespace: str, revision_uid: str) -> None:
|
|
self.calls.append(("disable_route", namespace, revision_uid))
|
|
self.routes[(namespace, revision_uid)] = False
|
|
|
|
def teardown_gateway(self, namespace: str) -> None:
|
|
self.calls.append(("teardown_gateway", namespace))
|
|
self.gateways.discard(namespace)
|
|
for key in [k for k in self.routes if k[0] == namespace]:
|
|
del self.routes[key]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Controller
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_READ_ACTIONS: dict[Kind, str] = {
|
|
Kind.CONFIGURATION: "hermes.configurations.read",
|
|
Kind.HARNESS: "hermes.harnesses.read",
|
|
Kind.CHANNEL: "hermes.channels.read",
|
|
Kind.SECRET: "hermes.secrets.read",
|
|
Kind.SANDBOX_POLICY: "hermes.sandboxpolicies.read",
|
|
}
|
|
|
|
|
|
class Controller:
|
|
"""Owns namespace lifecycle and the single deploy/rollback path."""
|
|
|
|
def __init__(
|
|
self,
|
|
store: ResourceStore,
|
|
audit: AuditLog,
|
|
registry: DriverRegistry,
|
|
gateway_manager: GatewayManager,
|
|
iam_adapter: IAMAdapter,
|
|
) -> None:
|
|
self.store = store
|
|
self.audit = audit
|
|
self.registry = registry
|
|
self.gateway = gateway_manager
|
|
self.iam = iam_adapter
|
|
self._agent_locks: dict[tuple[str, str], threading.Lock] = {}
|
|
self._locks_guard = threading.Lock()
|
|
|
|
# -- internals ---------------------------------------------------------
|
|
|
|
def _agent_lock(self, namespace: str, agent_name: str) -> threading.Lock:
|
|
key = (namespace, agent_name)
|
|
with self._locks_guard:
|
|
lock = self._agent_locks.get(key)
|
|
if lock is None:
|
|
lock = threading.Lock()
|
|
self._agent_locks[key] = lock
|
|
return lock
|
|
|
|
def _compute(self) -> ComputeDriver:
|
|
return self.registry.get("compute")
|
|
|
|
def _sandbox(self) -> SandboxDriver:
|
|
return self.registry.get("sandbox")
|
|
|
|
def _audit(self, **kw: Any) -> None:
|
|
self.audit.record(**kw)
|
|
|
|
# -- namespace lifecycle -------------------------------------------------
|
|
|
|
def ensure_namespace(self, name: str, actor: str = "controller") -> Resource:
|
|
"""Create the Namespace if missing and reconcile it toward Ready.
|
|
|
|
Ready requires the gateway to report ready; any gateway failure
|
|
moves the namespace to Failed. Each transition is audited.
|
|
"""
|
|
try:
|
|
ns = self.store.get(Kind.NAMESPACE, name)
|
|
except NotFoundError:
|
|
ns = Resource(
|
|
meta=ResourceMeta(kind=Kind.NAMESPACE.value, name=name),
|
|
status={"phase": NamespacePhase.PENDING.value},
|
|
)
|
|
self.store.create(ns)
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.namespaces.create", outcome="applied",
|
|
kind=Kind.NAMESPACE.value, resource=name,
|
|
)
|
|
|
|
prev_phase = ns.status.get("phase", NamespacePhase.PENDING.value)
|
|
try:
|
|
ready = bool(self.gateway.ensure_gateway(name))
|
|
phase = NamespacePhase.READY if ready else NamespacePhase.PENDING
|
|
reason = None if ready else "gateway not ready"
|
|
except Exception as exc: # gateway failure -> Failed, fail-closed
|
|
phase = NamespacePhase.FAILED
|
|
reason = f"gateway error: {exc}"
|
|
|
|
if phase.value != prev_phase:
|
|
ns = self.store.update_status(
|
|
Kind.NAMESPACE, name, None, {"phase": phase.value}
|
|
)
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.namespaces.reconcile",
|
|
outcome="applied" if phase is not NamespacePhase.FAILED else "error",
|
|
kind=Kind.NAMESPACE.value, resource=name,
|
|
reason=reason,
|
|
detail={"from": prev_phase, "to": phase.value},
|
|
)
|
|
return self.store.get(Kind.NAMESPACE, name)
|
|
|
|
# -- helpers -------------------------------------------------------------
|
|
|
|
def active_revision(self, agent_name: str, namespace: str) -> Resource | None:
|
|
"""The sole Active revision for the agent, or None."""
|
|
for rev in self.store.list(Kind.AGENT_REVISION, namespace):
|
|
if (
|
|
rev.spec.get("agent") == agent_name
|
|
and rev.status.get("phase") == RevisionPhase.ACTIVE.value
|
|
):
|
|
return rev
|
|
return None
|
|
|
|
def _revisions_for(self, agent_name: str, namespace: str) -> list[Resource]:
|
|
revs = [
|
|
r
|
|
for r in self.store.list(Kind.AGENT_REVISION, namespace)
|
|
if r.spec.get("agent") == agent_name
|
|
]
|
|
revs.sort(key=lambda r: int(r.spec.get("serial", 0)))
|
|
return revs
|
|
|
|
def _next_serial(self, agent_name: str, namespace: str) -> int:
|
|
revs = self._revisions_for(agent_name, namespace)
|
|
return (int(revs[-1].spec.get("serial", 0)) + 1) if revs else 1
|
|
|
|
def _authorize(self, actor: str, action: str, kind: Kind,
|
|
namespace: str | None, resource: str | None) -> None:
|
|
req = AuthzRequest(
|
|
principal=actor,
|
|
principal_kind="principal",
|
|
action=action,
|
|
kind=kind.value,
|
|
namespace=namespace,
|
|
resource=resource,
|
|
)
|
|
try:
|
|
self.iam.authorize(req)
|
|
except AuthorizationError:
|
|
self._audit(
|
|
actor=actor, actor_kind="principal", action=action,
|
|
outcome="deny", kind=kind.value, namespace=namespace,
|
|
resource=resource, reason="denied by IAM adapter",
|
|
)
|
|
raise
|
|
self._audit(
|
|
actor=actor, actor_kind="principal", action=action,
|
|
outcome="allow", kind=kind.value, namespace=namespace,
|
|
resource=resource,
|
|
)
|
|
|
|
@staticmethod
|
|
def _workload_ref(revision: Resource, driver_name: str) -> WorkloadRef:
|
|
return WorkloadRef(
|
|
revision_uid=revision.meta.uid,
|
|
namespace=revision.meta.namespace or "",
|
|
workload_identity=str(revision.spec.get("workloadIdentity", "")),
|
|
driver=driver_name,
|
|
)
|
|
|
|
# -- deploy ----------------------------------------------------------------
|
|
|
|
def deploy(self, agent_name: str, namespace: str, actor: str) -> Resource:
|
|
"""Deploy the agent's current spec as a new immutable revision."""
|
|
with self._agent_lock(namespace, agent_name):
|
|
return self._deploy_locked(agent_name, namespace, actor)
|
|
|
|
def _deploy_locked(self, agent_name: str, namespace: str, actor: str) -> Resource:
|
|
# 1. Authorize deploy on the exact Agent.
|
|
self._authorize(
|
|
actor, "hermes.agents.deploy", Kind.AGENT, namespace, agent_name
|
|
)
|
|
|
|
agent = self.store.get(Kind.AGENT, agent_name, namespace)
|
|
|
|
# 2. Resolve + per-reference authorize protected references.
|
|
config = self.store.get(
|
|
Kind.CONFIGURATION, agent.spec["configuration"], namespace
|
|
)
|
|
self._authorize(
|
|
actor, _READ_ACTIONS[Kind.CONFIGURATION], Kind.CONFIGURATION,
|
|
namespace, config.meta.name,
|
|
)
|
|
|
|
harness = self.store.get(Kind.HARNESS, agent.spec["harness"], None)
|
|
self._authorize(
|
|
actor, _READ_ACTIONS[Kind.HARNESS], Kind.HARNESS, None,
|
|
harness.meta.name,
|
|
)
|
|
|
|
for ch_name in agent.spec.get("channels", []) or []:
|
|
self.store.get(Kind.CHANNEL, ch_name, namespace)
|
|
self._authorize(
|
|
actor, _READ_ACTIONS[Kind.CHANNEL], Kind.CHANNEL,
|
|
namespace, ch_name,
|
|
)
|
|
for sec_name in agent.spec.get("secrets", []) or []:
|
|
self.store.get(Kind.SECRET, sec_name, namespace)
|
|
self._authorize(
|
|
actor, _READ_ACTIONS[Kind.SECRET], Kind.SECRET,
|
|
namespace, sec_name,
|
|
)
|
|
|
|
policy: dict[str, Any] = {}
|
|
policy_name = agent.spec.get("sandboxPolicy")
|
|
if policy_name:
|
|
policy_res = self.store.get(Kind.SANDBOX_POLICY, policy_name, namespace)
|
|
self._authorize(
|
|
actor, _READ_ACTIONS[Kind.SANDBOX_POLICY], Kind.SANDBOX_POLICY,
|
|
namespace, policy_name,
|
|
)
|
|
policy = dict(policy_res.spec)
|
|
|
|
# 3. Namespace gateway must be ready.
|
|
ns = self.ensure_namespace(namespace, actor=actor)
|
|
if ns.status.get("phase") != NamespacePhase.READY.value:
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.deploy", outcome="error",
|
|
kind=Kind.AGENT.value, namespace=namespace,
|
|
resource=agent_name,
|
|
reason=f"namespace not Ready (phase={ns.status.get('phase')})",
|
|
)
|
|
raise DeploymentError(
|
|
f"namespace {namespace!r} is not Ready; cannot deploy"
|
|
)
|
|
|
|
compute = self._compute()
|
|
sandbox = self._sandbox()
|
|
|
|
# 4. Sandbox driver must support the ENTIRE policy.
|
|
if not sandbox.supports(policy):
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.deploy", outcome="deny",
|
|
kind=Kind.AGENT.value, namespace=namespace,
|
|
resource=agent_name,
|
|
reason=(
|
|
f"sandbox driver {sandbox.name!r} cannot enforce the "
|
|
"entire SandboxPolicy; partial support is unsupported"
|
|
),
|
|
)
|
|
raise DeploymentError(
|
|
f"sandbox driver {sandbox.name!r} does not support the "
|
|
"entire SandboxPolicy"
|
|
)
|
|
|
|
# 5. Immutable AgentRevision snapshot.
|
|
previous = self.active_revision(agent_name, namespace)
|
|
serial = self._next_serial(agent_name, namespace)
|
|
revision = Resource(
|
|
meta=ResourceMeta(
|
|
kind=Kind.AGENT_REVISION.value,
|
|
name=f"{agent_name}-rev-{serial}",
|
|
namespace=namespace,
|
|
),
|
|
spec={
|
|
"agent": agent_name,
|
|
"agentUid": agent.meta.uid,
|
|
"serial": serial,
|
|
"workloadIdentity": f"wi-{agent_name}",
|
|
"harness": {
|
|
"name": harness.meta.name,
|
|
"version": harness.spec["version"],
|
|
"image": harness.spec["image"],
|
|
},
|
|
"configuration": dict(config.spec.get("config", {})),
|
|
"sandboxPolicy": policy,
|
|
"computeDriver": self.registry.selected_name("compute"),
|
|
"sandboxDriver": self.registry.selected_name("sandbox"),
|
|
},
|
|
status={"phase": RevisionPhase.CANDIDATE.value},
|
|
)
|
|
self.store.create(revision)
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agentrevisions.create", outcome="applied",
|
|
kind=Kind.AGENT_REVISION.value, namespace=namespace,
|
|
resource=revision.meta.name,
|
|
detail={"serial": serial, "agent": agent_name},
|
|
)
|
|
|
|
ref: WorkloadRef | None = None
|
|
|
|
def _fail_candidate(stage: str, exc: Exception) -> None:
|
|
"""Pre-activation failure: candidate Failed, workload torn down,
|
|
previous revision and its route untouched."""
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, revision.meta.name, namespace,
|
|
{"phase": RevisionPhase.FAILED.value, "reason": f"{stage}: {exc}"},
|
|
)
|
|
if ref is not None:
|
|
try:
|
|
compute.teardown(ref)
|
|
except Exception:
|
|
pass # best-effort teardown; candidate is already Failed
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.deploy", outcome="error",
|
|
kind=Kind.AGENT.value, namespace=namespace,
|
|
resource=agent_name,
|
|
reason=f"deploy failed at {stage}: {exc}",
|
|
detail={"revision": revision.meta.name, "stage": stage},
|
|
)
|
|
|
|
# 6. Provision candidate; enforce + verify containment BEFORE start.
|
|
try:
|
|
ref = compute.provision_candidate(revision)
|
|
except Exception as exc:
|
|
_fail_candidate("provision_candidate", exc)
|
|
raise DeploymentError(f"provisioning failed: {exc}") from exc
|
|
|
|
try:
|
|
sandbox.enforce(ref, policy)
|
|
sandbox.verify(ref, policy)
|
|
except Exception as exc:
|
|
_fail_candidate("sandbox containment", exc)
|
|
raise DeploymentError(f"containment failed: {exc}") from exc
|
|
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, revision.meta.name, namespace,
|
|
{"phase": RevisionPhase.CONTAINED.value},
|
|
)
|
|
|
|
# 7. Prepare non-serving route.
|
|
try:
|
|
self.gateway.prepare_route(namespace, revision.meta.uid)
|
|
except Exception as exc:
|
|
_fail_candidate("prepare_route", exc)
|
|
raise DeploymentError(f"route preparation failed: {exc}") from exc
|
|
|
|
# 8. Retire previous active revision (stop harness, verify stopped).
|
|
prev_ref: WorkloadRef | None = None
|
|
if previous is not None:
|
|
prev_ref = self._workload_ref(
|
|
previous, str(previous.spec.get("computeDriver", compute.name))
|
|
)
|
|
try:
|
|
compute.stop_harness(prev_ref)
|
|
self.gateway.disable_route(namespace, previous.meta.uid)
|
|
except Exception as exc:
|
|
_fail_candidate("retire previous revision", exc)
|
|
raise DeploymentError(
|
|
f"could not retire previous revision: {exc}"
|
|
) from exc
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, previous.meta.name, namespace,
|
|
{"phase": RevisionPhase.RETIRED.value},
|
|
)
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agentrevisions.retire", outcome="applied",
|
|
kind=Kind.AGENT_REVISION.value, namespace=namespace,
|
|
resource=previous.meta.name,
|
|
)
|
|
|
|
# 9. Activate candidate (sole active), start harness, enable route.
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, revision.meta.name, namespace,
|
|
{"phase": RevisionPhase.ACTIVE.value},
|
|
)
|
|
try:
|
|
compute.start_harness(ref)
|
|
self.gateway.enable_route(namespace, revision.meta.uid)
|
|
except Exception as exc:
|
|
self._rollback_after_activation(
|
|
agent_name, namespace, actor, revision, ref,
|
|
previous, prev_ref, cause=exc,
|
|
)
|
|
raise DeploymentError(
|
|
f"post-activation failure, rolled back: {exc}"
|
|
) from exc
|
|
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.deploy", outcome="applied",
|
|
kind=Kind.AGENT.value, namespace=namespace, resource=agent_name,
|
|
detail={"revision": revision.meta.name, "serial": serial},
|
|
)
|
|
return self.store.get(Kind.AGENT_REVISION, revision.meta.name, namespace)
|
|
|
|
# -- rollback ----------------------------------------------------------
|
|
|
|
def _rollback_after_activation(
|
|
self,
|
|
agent_name: str,
|
|
namespace: str,
|
|
actor: str,
|
|
candidate: Resource,
|
|
candidate_ref: WorkloadRef,
|
|
previous: Resource | None,
|
|
prev_ref: WorkloadRef | None,
|
|
cause: Exception,
|
|
) -> None:
|
|
"""Roll back to `previous` after a post-activation failure.
|
|
|
|
On unverifiable rollback the agent is left inactive with all routes
|
|
disabled and RollbackError is raised (chained to the cause).
|
|
"""
|
|
compute = self._compute()
|
|
|
|
# Retire the failed candidate first: no traffic, no harness.
|
|
try:
|
|
self.gateway.disable_route(namespace, candidate.meta.uid)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
compute.stop_harness(candidate_ref)
|
|
except Exception:
|
|
pass
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, candidate.meta.name, namespace,
|
|
{"phase": RevisionPhase.RETIRED.value,
|
|
"reason": f"rolled back: {cause}"},
|
|
)
|
|
|
|
if previous is None or prev_ref is None:
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.rollback", outcome="error",
|
|
kind=Kind.AGENT.value, namespace=namespace,
|
|
resource=agent_name,
|
|
reason=f"no previous revision to roll back to ({cause})",
|
|
)
|
|
raise RollbackError(
|
|
f"post-activation failure and no previous revision; agent "
|
|
f"{agent_name!r} is inactive: {cause}"
|
|
) from cause
|
|
|
|
try:
|
|
compute.start_harness(prev_ref)
|
|
self.gateway.enable_route(namespace, previous.meta.uid)
|
|
except Exception as exc:
|
|
# Unverifiable rollback: everything stays down.
|
|
try:
|
|
self.gateway.disable_route(namespace, previous.meta.uid)
|
|
except Exception:
|
|
pass
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.rollback", outcome="error",
|
|
kind=Kind.AGENT.value, namespace=namespace,
|
|
resource=agent_name,
|
|
reason=f"rollback unverifiable: {exc}",
|
|
detail={"candidate": candidate.meta.name,
|
|
"previous": previous.meta.name},
|
|
)
|
|
raise RollbackError(
|
|
f"rollback to {previous.meta.name!r} could not be verified; "
|
|
f"agent {agent_name!r} left inactive with routes disabled"
|
|
) from exc
|
|
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, previous.meta.name, namespace,
|
|
{"phase": RevisionPhase.ACTIVE.value},
|
|
)
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.rollback", outcome="rolled-back",
|
|
kind=Kind.AGENT.value, namespace=namespace, resource=agent_name,
|
|
reason=f"post-activation failure: {cause}",
|
|
detail={"candidate": candidate.meta.name,
|
|
"restored": previous.meta.name},
|
|
)
|
|
|
|
def rollback(self, agent_name: str, namespace: str, actor: str) -> Resource:
|
|
"""Explicit rollback to the most recently Retired revision."""
|
|
with self._agent_lock(namespace, agent_name):
|
|
return self._rollback_locked(agent_name, namespace, actor)
|
|
|
|
def _rollback_locked(self, agent_name: str, namespace: str,
|
|
actor: str) -> Resource:
|
|
self._authorize(
|
|
actor, "hermes.agents.rollback", Kind.AGENT, namespace, agent_name
|
|
)
|
|
revs = self._revisions_for(agent_name, namespace)
|
|
retired = [
|
|
r for r in revs
|
|
if r.status.get("phase") == RevisionPhase.RETIRED.value
|
|
]
|
|
if not retired:
|
|
raise DeploymentError(
|
|
f"agent {agent_name!r} has no Retired revision to roll back to"
|
|
)
|
|
target = retired[-1]
|
|
current = self.active_revision(agent_name, namespace)
|
|
compute = self._compute()
|
|
|
|
# Stop the current active revision first (fail-closed).
|
|
if current is not None:
|
|
cur_ref = self._workload_ref(
|
|
current, str(current.spec.get("computeDriver", compute.name))
|
|
)
|
|
try:
|
|
compute.stop_harness(cur_ref)
|
|
self.gateway.disable_route(namespace, current.meta.uid)
|
|
except Exception as exc:
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.rollback", outcome="error",
|
|
kind=Kind.AGENT.value, namespace=namespace,
|
|
resource=agent_name,
|
|
reason=f"could not stop current revision: {exc}",
|
|
)
|
|
raise RollbackError(
|
|
f"could not stop current revision {current.meta.name!r}: {exc}"
|
|
) from exc
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, current.meta.name, namespace,
|
|
{"phase": RevisionPhase.RETIRED.value},
|
|
)
|
|
|
|
target_ref = self._workload_ref(
|
|
target, str(target.spec.get("computeDriver", compute.name))
|
|
)
|
|
try:
|
|
compute.start_harness(target_ref)
|
|
self.gateway.enable_route(namespace, target.meta.uid)
|
|
except Exception as exc:
|
|
try:
|
|
self.gateway.disable_route(namespace, target.meta.uid)
|
|
except Exception:
|
|
pass
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.rollback", outcome="error",
|
|
kind=Kind.AGENT.value, namespace=namespace,
|
|
resource=agent_name,
|
|
reason=f"rollback unverifiable: {exc}",
|
|
)
|
|
raise RollbackError(
|
|
f"rollback to {target.meta.name!r} could not be verified; "
|
|
f"agent {agent_name!r} left inactive with routes disabled"
|
|
) from exc
|
|
|
|
self.store.update_status(
|
|
Kind.AGENT_REVISION, target.meta.name, namespace,
|
|
{"phase": RevisionPhase.ACTIVE.value},
|
|
)
|
|
self._audit(
|
|
actor=actor, actor_kind="principal",
|
|
action="hermes.agents.rollback", outcome="rolled-back",
|
|
kind=Kind.AGENT.value, namespace=namespace, resource=agent_name,
|
|
detail={"restored": target.meta.name},
|
|
)
|
|
return self.store.get(Kind.AGENT_REVISION, target.meta.name, namespace)
|