feat(release): add dedicated release preparation worker
This commit is contained in:
AIOS Governed Continuation
@@ -0,0 +1,636 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Persistent release-preparation Online Agent worker.
|
||||
|
||||
The worker is intentionally narrower than the generic engineering worker:
|
||||
it can only consume tasks whose required capability set is exactly
|
||||
{"release:candidate"} and whose payload asks for immutable release-candidate
|
||||
creation. It never receives Owner signing, deployment, tag, baseline or Git
|
||||
write credentials.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import Any, Mapping
|
||||
|
||||
AGENT_ID = os.environ.get(
|
||||
"AIOS_RELEASE_WORKER_AGENT_ID",
|
||||
"online-release-worker-cn-network-mkt-20260819",
|
||||
).strip()
|
||||
AGENT_VERSION = os.environ.get(
|
||||
"AIOS_RELEASE_WORKER_AGENT_VERSION",
|
||||
"2026-08-19.1-release-candidate",
|
||||
).strip()
|
||||
TASK_API = os.environ.get(
|
||||
"AIOS_AGENT_TASK_API_URL",
|
||||
"http://127.0.0.1:18940",
|
||||
).rstrip("/")
|
||||
RELEASE_API = os.environ.get("AIOS_RELEASE_API_URL", "").rstrip("/")
|
||||
POLL_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_POLL_SECONDS", "3"))
|
||||
LEASE_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_LEASE_SECONDS", "60"))
|
||||
HEARTBEAT_SECONDS = int(
|
||||
os.environ.get("AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS", "15")
|
||||
)
|
||||
STATE_ROOT = Path(
|
||||
os.environ.get(
|
||||
"AIOS_RELEASE_WORKER_STATE_ROOT",
|
||||
"/var/lib/aios-online-agent-release",
|
||||
)
|
||||
)
|
||||
EXPECTED_CAPABILITIES = frozenset({"release:candidate"})
|
||||
LIVENESS_PATH = Path(
|
||||
os.environ.get(
|
||||
"AIOS_WORKER_LIVENESS_PATH",
|
||||
str(STATE_ROOT / "worker-liveness.json"),
|
||||
)
|
||||
)
|
||||
CANDIDATE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,255}$")
|
||||
OID_RE = re.compile(r"^[0-9a-f]{40}$")
|
||||
VERSION_RE = re.compile(r"^V[0-9]+R[0-9]+C[0-9]+$")
|
||||
TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
|
||||
SECRET_MARKERS = (
|
||||
"authorization",
|
||||
"token",
|
||||
"secret",
|
||||
"credential",
|
||||
"password",
|
||||
"private_key",
|
||||
"signature",
|
||||
)
|
||||
|
||||
|
||||
class ReleaseWorkerError(RuntimeError):
|
||||
def __init__(self, code: str, *, phase: str) -> None:
|
||||
super().__init__(code)
|
||||
self.code = code
|
||||
self.phase = phase
|
||||
|
||||
|
||||
def now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def canonical(value: object) -> bytes:
|
||||
return json.dumps(
|
||||
value,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def sha256(value: bytes) -> str:
|
||||
return hashlib.sha256(value).hexdigest()
|
||||
|
||||
|
||||
def sanitized(value: Any) -> Any:
|
||||
if isinstance(value, Mapping):
|
||||
result: dict[str, Any] = {}
|
||||
for key, inner in value.items():
|
||||
lowered = str(key).lower()
|
||||
if any(marker in lowered for marker in SECRET_MARKERS):
|
||||
result[str(key)] = "[REDACTED]"
|
||||
else:
|
||||
result[str(key)] = sanitized(inner)
|
||||
return result
|
||||
if isinstance(value, list):
|
||||
return [sanitized(item) for item in value]
|
||||
return value
|
||||
|
||||
|
||||
def audit(event: str, **fields: Any) -> None:
|
||||
STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
|
||||
path = STATE_ROOT / "events.jsonl"
|
||||
record = {
|
||||
"schema": "aios.release-worker.event.v1",
|
||||
"event": event,
|
||||
"created_at": now(),
|
||||
"agent_id": AGENT_ID,
|
||||
**sanitized(fields),
|
||||
}
|
||||
with path.open("a", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps(record, sort_keys=True) + "\n")
|
||||
os.chmod(path, 0o640)
|
||||
|
||||
|
||||
def sd_notify(message: str) -> None:
|
||||
target = os.environ.get("NOTIFY_SOCKET", "")
|
||||
if not target:
|
||||
return
|
||||
if target.startswith("@"):
|
||||
address = "\0" + target[1:]
|
||||
else:
|
||||
address = target
|
||||
sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
|
||||
try:
|
||||
sock.connect(address)
|
||||
sock.sendall(message.encode("utf-8"))
|
||||
except OSError:
|
||||
# Liveness JSON remains the diagnostic source; systemd will enforce
|
||||
# WatchdogSec if notify transport is persistently broken.
|
||||
pass
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
|
||||
def write_liveness(
|
||||
state: str,
|
||||
*,
|
||||
task_id: str | None = None,
|
||||
stage: str | None = None,
|
||||
error: str | None = None,
|
||||
visible_count: int | None = None,
|
||||
) -> None:
|
||||
STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
|
||||
value = {
|
||||
"schema": "aios.online-agent.worker-liveness.v1",
|
||||
"agent_id": AGENT_ID,
|
||||
"agent_type": "release-preparation-worker",
|
||||
"agent_version": AGENT_VERSION,
|
||||
"pid": os.getpid(),
|
||||
"state": state,
|
||||
"last_seen_epoch": time.time(),
|
||||
"last_seen_at": now(),
|
||||
"poll_seconds": POLL_SECONDS,
|
||||
"heartbeat_seconds": HEARTBEAT_SECONDS,
|
||||
"api_base": TASK_API,
|
||||
"capabilities": sorted(EXPECTED_CAPABILITIES),
|
||||
"task_id": task_id,
|
||||
"stage": stage,
|
||||
"error": error,
|
||||
"visible_count": visible_count,
|
||||
}
|
||||
LIVENESS_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
|
||||
temporary = LIVENESS_PATH.with_name(
|
||||
"." + LIVENESS_PATH.name + f".{os.getpid()}.tmp"
|
||||
)
|
||||
temporary.write_text(
|
||||
json.dumps(value, ensure_ascii=False, sort_keys=True) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
os.chmod(temporary, 0o640)
|
||||
os.replace(temporary, LIVENESS_PATH)
|
||||
|
||||
|
||||
def load_credential(name: str) -> str:
|
||||
directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
|
||||
path = directory / name
|
||||
if path.is_symlink() or not path.is_file():
|
||||
raise ReleaseWorkerError(
|
||||
"CREDENTIAL_UNAVAILABLE:" + name,
|
||||
phase="startup",
|
||||
)
|
||||
value = path.read_text(encoding="utf-8").strip()
|
||||
if len(value) < 24 or any(ch.isspace() for ch in value):
|
||||
raise ReleaseWorkerError(
|
||||
"CREDENTIAL_INVALID:" + name,
|
||||
phase="startup",
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def http_json(
|
||||
base: str,
|
||||
method: str,
|
||||
route: str,
|
||||
token: str,
|
||||
body: Mapping[str, Any] | None = None,
|
||||
*,
|
||||
timeout: int = 30,
|
||||
) -> tuple[int, dict[str, Any]]:
|
||||
data = None if body is None else canonical(body)
|
||||
headers = {"Accept": "application/json"}
|
||||
if token:
|
||||
headers["Authorization"] = "Bearer " + token
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
request = urllib.request.Request(
|
||||
base + route,
|
||||
method=method,
|
||||
data=data,
|
||||
headers=headers,
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||||
raw, status = response.read(), response.status
|
||||
except urllib.error.HTTPError as exc:
|
||||
raw, status = exc.read(), exc.code
|
||||
except (urllib.error.URLError, TimeoutError) as exc:
|
||||
raise ReleaseWorkerError(
|
||||
"HTTP_TRANSPORT_UNAVAILABLE",
|
||||
phase="network",
|
||||
) from exc
|
||||
try:
|
||||
value = json.loads(raw.decode())
|
||||
except Exception as exc:
|
||||
raise ReleaseWorkerError(
|
||||
"HTTP_RESPONSE_INVALID",
|
||||
phase="network",
|
||||
) from exc
|
||||
return status, value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReleaseRequest:
|
||||
candidate_id: str
|
||||
source_commit: str
|
||||
version: str
|
||||
projection_id: str | None
|
||||
|
||||
@classmethod
|
||||
def from_task(cls, task: Mapping[str, Any]) -> "ReleaseRequest":
|
||||
if not TASK_RE.fullmatch(str(task.get("task_id") or "")):
|
||||
raise ReleaseWorkerError("TASK_ID_INVALID", phase="validation")
|
||||
caps = task.get("required_capabilities")
|
||||
if (
|
||||
not isinstance(caps, list)
|
||||
or frozenset(caps) != EXPECTED_CAPABILITIES
|
||||
or len(caps) != 1
|
||||
):
|
||||
raise ReleaseWorkerError(
|
||||
"TASK_CAPABILITY_SCOPE_INVALID",
|
||||
phase="validation",
|
||||
)
|
||||
metadata = task.get("metadata")
|
||||
if not isinstance(metadata, Mapping):
|
||||
metadata = {}
|
||||
constraints = task.get("constraints")
|
||||
if not isinstance(constraints, Mapping):
|
||||
constraints = {}
|
||||
release = constraints.get("release_candidate")
|
||||
if not isinstance(release, Mapping):
|
||||
release = metadata.get("release_candidate")
|
||||
if not isinstance(release, Mapping):
|
||||
release = task.get("release_candidate")
|
||||
if not isinstance(release, Mapping):
|
||||
raise ReleaseWorkerError(
|
||||
"RELEASE_CANDIDATE_PAYLOAD_MISSING",
|
||||
phase="validation",
|
||||
)
|
||||
allowed = {
|
||||
"candidate_id",
|
||||
"source_commit",
|
||||
"version",
|
||||
"projection_id",
|
||||
}
|
||||
if set(release) - allowed:
|
||||
raise ReleaseWorkerError(
|
||||
"RELEASE_CANDIDATE_PAYLOAD_FIELDS_INVALID",
|
||||
phase="validation",
|
||||
)
|
||||
candidate_id = str(release.get("candidate_id") or "")
|
||||
source_commit = str(release.get("source_commit") or "")
|
||||
version = str(release.get("version") or "")
|
||||
projection_id = release.get("projection_id")
|
||||
if CANDIDATE_RE.fullmatch(candidate_id) is None:
|
||||
raise ReleaseWorkerError(
|
||||
"RELEASE_CANDIDATE_ID_INVALID",
|
||||
phase="validation",
|
||||
)
|
||||
if OID_RE.fullmatch(source_commit) is None:
|
||||
raise ReleaseWorkerError(
|
||||
"RELEASE_SOURCE_COMMIT_INVALID",
|
||||
phase="validation",
|
||||
)
|
||||
if VERSION_RE.fullmatch(version) is None:
|
||||
raise ReleaseWorkerError(
|
||||
"RELEASE_VERSION_INVALID",
|
||||
phase="validation",
|
||||
)
|
||||
if projection_id is not None:
|
||||
projection_id = str(projection_id)
|
||||
if not re.fullmatch(r"projection-[0-9a-f]{32}", projection_id):
|
||||
raise ReleaseWorkerError(
|
||||
"RELEASE_PROJECTION_ID_INVALID",
|
||||
phase="validation",
|
||||
)
|
||||
return cls(
|
||||
candidate_id=candidate_id,
|
||||
source_commit=source_commit,
|
||||
version=version,
|
||||
projection_id=projection_id,
|
||||
)
|
||||
|
||||
def body(self) -> dict[str, str]:
|
||||
return {
|
||||
"candidate_id": self.candidate_id,
|
||||
"source_commit": self.source_commit,
|
||||
"version": self.version,
|
||||
}
|
||||
|
||||
|
||||
class Heartbeat:
|
||||
def __init__(
|
||||
self,
|
||||
agent_token: str,
|
||||
task_id: str,
|
||||
lease_token: str,
|
||||
) -> None:
|
||||
self.agent_token = agent_token
|
||||
self.task_id = task_id
|
||||
self.lease_token = lease_token
|
||||
self.stage = "claimed"
|
||||
self.failure: str | None = None
|
||||
self.stop = threading.Event()
|
||||
self.thread = threading.Thread(
|
||||
target=self._run,
|
||||
name="release-worker-heartbeat",
|
||||
daemon=True,
|
||||
)
|
||||
|
||||
def first(self) -> None:
|
||||
status, value = http_json(
|
||||
TASK_API,
|
||||
"POST",
|
||||
f"/v1/tasks/{self.task_id}/heartbeat",
|
||||
self.agent_token,
|
||||
{
|
||||
"lease_token": self.lease_token,
|
||||
"lease_seconds": LEASE_SECONDS,
|
||||
"progress": {
|
||||
"stage": self.stage,
|
||||
"worker_version": AGENT_VERSION,
|
||||
},
|
||||
},
|
||||
)
|
||||
if status != 200:
|
||||
raise ReleaseWorkerError(
|
||||
"TASK_HEARTBEAT_FAILED",
|
||||
phase="heartbeat",
|
||||
)
|
||||
audit(
|
||||
"TASK_HEARTBEAT",
|
||||
task_id=self.task_id,
|
||||
stage=self.stage,
|
||||
api_status=status,
|
||||
)
|
||||
write_liveness(
|
||||
"BUSY",
|
||||
task_id=self.task_id,
|
||||
stage=self.stage,
|
||||
)
|
||||
sd_notify("WATCHDOG=1")
|
||||
|
||||
def _run(self) -> None:
|
||||
while not self.stop.wait(HEARTBEAT_SECONDS):
|
||||
try:
|
||||
status, value = http_json(
|
||||
TASK_API,
|
||||
"POST",
|
||||
f"/v1/tasks/{self.task_id}/heartbeat",
|
||||
self.agent_token,
|
||||
{
|
||||
"lease_token": self.lease_token,
|
||||
"lease_seconds": LEASE_SECONDS,
|
||||
"progress": {
|
||||
"stage": self.stage,
|
||||
"worker_version": AGENT_VERSION,
|
||||
},
|
||||
},
|
||||
)
|
||||
if status != 200:
|
||||
self.failure = str(value.get("error") or status)
|
||||
return
|
||||
audit(
|
||||
"TASK_HEARTBEAT",
|
||||
task_id=self.task_id,
|
||||
stage=self.stage,
|
||||
api_status=status,
|
||||
)
|
||||
write_liveness(
|
||||
"BUSY",
|
||||
task_id=self.task_id,
|
||||
stage=self.stage,
|
||||
)
|
||||
sd_notify("WATCHDOG=1")
|
||||
except Exception as exc:
|
||||
self.failure = type(exc).__name__
|
||||
return
|
||||
|
||||
def ensure(self) -> None:
|
||||
if self.failure:
|
||||
raise ReleaseWorkerError(
|
||||
"TASK_LEASE_LOST",
|
||||
phase="heartbeat",
|
||||
)
|
||||
|
||||
def close(self) -> None:
|
||||
self.stop.set()
|
||||
if self.thread.ident is not None:
|
||||
self.thread.join(timeout=2)
|
||||
|
||||
|
||||
def result_payload(
|
||||
task_id: str,
|
||||
release: ReleaseRequest,
|
||||
builder: Mapping[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
builder_candidate = builder.get("candidate_id")
|
||||
if builder_candidate not in {None, release.candidate_id}:
|
||||
raise ReleaseWorkerError(
|
||||
"BUILDER_CANDIDATE_BINDING_INVALID",
|
||||
phase="release",
|
||||
)
|
||||
immutable = {
|
||||
"candidate_id": release.candidate_id,
|
||||
"source_commit": release.source_commit,
|
||||
"version": release.version,
|
||||
}
|
||||
return {
|
||||
"status": "SUCCEEDED",
|
||||
"classification": "RELEASE_CANDIDATE_CREATED",
|
||||
"execution_mode": "release_candidate",
|
||||
"adapter": "release_candidate",
|
||||
"task_id": task_id,
|
||||
"agent_id": AGENT_ID,
|
||||
"release_candidate": immutable,
|
||||
"release_candidate_request_sha256": sha256(canonical(immutable)),
|
||||
"builder_response": sanitized(dict(builder)),
|
||||
"projection_id": release.projection_id,
|
||||
"writer_lease": "NONE",
|
||||
"changed_files": [],
|
||||
"direct_main_write": False,
|
||||
"production_write": False,
|
||||
"tag_write": False,
|
||||
"baseline_write": False,
|
||||
"runtime_switch": False,
|
||||
"owner_approval_submitted": False,
|
||||
"credential_material_exposed": False,
|
||||
}
|
||||
|
||||
|
||||
def execute_task(
|
||||
task: Mapping[str, Any],
|
||||
*,
|
||||
agent_token: str,
|
||||
release_token: str,
|
||||
) -> bool:
|
||||
task_id = str(task.get("task_id") or "")
|
||||
release = ReleaseRequest.from_task(task)
|
||||
status, claimed = http_json(
|
||||
TASK_API,
|
||||
"POST",
|
||||
f"/v1/tasks/{task_id}/claim",
|
||||
agent_token,
|
||||
{"lease_seconds": LEASE_SECONDS},
|
||||
)
|
||||
if status in {409, 423}:
|
||||
return False
|
||||
if status != 200:
|
||||
raise ReleaseWorkerError(
|
||||
"TASK_CLAIM_FAILED",
|
||||
phase="claim",
|
||||
)
|
||||
lease_token = str(claimed.get("lease_token") or "")
|
||||
if len(lease_token) < 16:
|
||||
raise ReleaseWorkerError(
|
||||
"TASK_LEASE_TOKEN_INVALID",
|
||||
phase="claim",
|
||||
)
|
||||
attempt = int(claimed.get("attempt") or 0)
|
||||
audit(
|
||||
"TASK_CLAIMED",
|
||||
task_id=task_id,
|
||||
attempt=attempt,
|
||||
)
|
||||
|
||||
heartbeat = Heartbeat(agent_token, task_id, lease_token)
|
||||
heartbeat.first()
|
||||
heartbeat.thread.start()
|
||||
try:
|
||||
heartbeat.stage = "release_candidate"
|
||||
heartbeat.ensure()
|
||||
status, builder = http_json(
|
||||
RELEASE_API,
|
||||
"POST",
|
||||
"/createReleaseCandidate",
|
||||
release_token,
|
||||
release.body(),
|
||||
timeout=120,
|
||||
)
|
||||
if status not in {200, 201, 202}:
|
||||
raise ReleaseWorkerError(
|
||||
"RELEASE_CANDIDATE_CREATE_FAILED:"
|
||||
+ str(builder.get("error") or status),
|
||||
phase="release",
|
||||
)
|
||||
heartbeat.ensure()
|
||||
result = result_payload(task_id, release, builder)
|
||||
idempotency_key = (
|
||||
f"{task_id}-attempt-{attempt}-release-result-v1"
|
||||
)
|
||||
status, submitted = http_json(
|
||||
TASK_API,
|
||||
"POST",
|
||||
f"/v1/tasks/{task_id}/result",
|
||||
agent_token,
|
||||
{
|
||||
"lease_token": lease_token,
|
||||
"idempotency_key": idempotency_key,
|
||||
"result": result,
|
||||
},
|
||||
)
|
||||
if status not in {200, 201, 202}:
|
||||
raise ReleaseWorkerError(
|
||||
"TASK_RESULT_SUBMIT_FAILED",
|
||||
phase="result",
|
||||
)
|
||||
audit(
|
||||
"RESULT_SUBMITTED",
|
||||
task_id=task_id,
|
||||
attempt=attempt,
|
||||
release_candidate_id=release.candidate_id,
|
||||
)
|
||||
return True
|
||||
finally:
|
||||
heartbeat.close()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if not RELEASE_API:
|
||||
raise SystemExit("FATAL:AIOS_RELEASE_API_URL_REQUIRED")
|
||||
agent_token = load_credential("agent_token")
|
||||
release_token = load_credential("release_candidate_token")
|
||||
STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
|
||||
audit(
|
||||
"WORKER_STARTED",
|
||||
capabilities=sorted(EXPECTED_CAPABILITIES),
|
||||
release_api=RELEASE_API,
|
||||
)
|
||||
write_liveness("STARTING", stage="startup")
|
||||
sd_notify("READY=1\nSTATUS=release worker ready")
|
||||
while True:
|
||||
try:
|
||||
status, response = http_json(
|
||||
TASK_API,
|
||||
"POST",
|
||||
"/v1/tasks/poll",
|
||||
agent_token,
|
||||
{"limit": 100},
|
||||
)
|
||||
if status != 200:
|
||||
raise ReleaseWorkerError(
|
||||
"TASK_POLL_FAILED",
|
||||
phase="poll",
|
||||
)
|
||||
tasks = response.get("tasks")
|
||||
if not isinstance(tasks, list):
|
||||
tasks = []
|
||||
write_liveness(
|
||||
"IDLE",
|
||||
stage="poll",
|
||||
visible_count=len(tasks),
|
||||
)
|
||||
sd_notify("WATCHDOG=1\nSTATUS=release worker poll ok")
|
||||
handled = False
|
||||
for raw in tasks:
|
||||
if not isinstance(raw, Mapping):
|
||||
continue
|
||||
caps = raw.get("required_capabilities")
|
||||
if (
|
||||
not isinstance(caps, list)
|
||||
or frozenset(caps) != EXPECTED_CAPABILITIES
|
||||
):
|
||||
continue
|
||||
try:
|
||||
handled = execute_task(
|
||||
raw,
|
||||
agent_token=agent_token,
|
||||
release_token=release_token,
|
||||
) or handled
|
||||
except ReleaseWorkerError as exc:
|
||||
audit(
|
||||
"TASK_EXECUTION_FAILED",
|
||||
task_id=raw.get("task_id"),
|
||||
phase=exc.phase,
|
||||
error=exc.code,
|
||||
)
|
||||
time.sleep(0 if handled else POLL_SECONDS)
|
||||
except KeyboardInterrupt:
|
||||
return 0
|
||||
except Exception as exc:
|
||||
audit(
|
||||
"WORKER_LOOP_ERROR",
|
||||
error=type(exc).__name__,
|
||||
)
|
||||
write_liveness(
|
||||
"DEGRADED",
|
||||
stage="loop",
|
||||
error=type(exc).__name__,
|
||||
)
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,50 @@
|
||||
[Unit]
|
||||
Description=AIOS release-candidate Online Agent worker
|
||||
After=network.target aios-agent-task-api.service
|
||||
Wants=aios-agent-task-api.service
|
||||
|
||||
[Service]
|
||||
Type=notify
|
||||
NotifyAccess=main
|
||||
WatchdogSec=90
|
||||
User=root
|
||||
Group=root
|
||||
StateDirectory=aios-online-agent-release
|
||||
StateDirectoryMode=0750
|
||||
LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-release-online-agent-worker/agent_token
|
||||
LoadCredentialEncrypted=release_candidate_token:/etc/credstore.encrypted/aios-release-online-agent-worker/release_candidate_token
|
||||
Environment=AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819
|
||||
Environment=AIOS_RELEASE_WORKER_AGENT_VERSION=2026-08-19.1-release-candidate
|
||||
Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
|
||||
Environment=AIOS_RELEASE_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
|
||||
Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
|
||||
Environment=AIOS_RELEASE_WORKER_POLL_SECONDS=3
|
||||
Environment=AIOS_RELEASE_WORKER_LEASE_SECONDS=60
|
||||
Environment=AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS=15
|
||||
Environment=AIOS_RELEASE_API_URL=http://127.0.0.1:18890
|
||||
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_release_online_agent_worker.py
|
||||
Restart=always
|
||||
RestartSec=2
|
||||
NoNewPrivileges=yes
|
||||
PrivateTmp=yes
|
||||
PrivateDevices=yes
|
||||
ProtectSystem=strict
|
||||
ProtectHome=yes
|
||||
ProtectKernelTunables=yes
|
||||
ProtectKernelModules=yes
|
||||
ProtectControlGroups=yes
|
||||
ProtectKernelLogs=yes
|
||||
ProtectClock=yes
|
||||
LockPersonality=yes
|
||||
RestrictSUIDSGID=yes
|
||||
RestrictRealtime=yes
|
||||
RestrictNamespaces=yes
|
||||
MemoryDenyWriteExecute=yes
|
||||
RestrictAddressFamilies=AF_UNIX AF_INET
|
||||
IPAddressDeny=any
|
||||
IPAddressAllow=127.0.0.0/8
|
||||
InaccessiblePaths=/data/aios-gitops /home
|
||||
UMask=0077
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -0,0 +1,142 @@
|
||||
# Release Preparation Online Agent
|
||||
|
||||
## Purpose
|
||||
|
||||
This worker closes the release-preparation execution gap without widening the ChatGPT connector
|
||||
identity.
|
||||
|
||||
The dedicated Agent identity is:
|
||||
|
||||
```text
|
||||
online-release-worker-cn-network-mkt-20260819
|
||||
```
|
||||
|
||||
Its only advertised task capability is:
|
||||
|
||||
```text
|
||||
release:candidate
|
||||
```
|
||||
|
||||
The worker may create an immutable release candidate, but it never submits Owner approval and never
|
||||
starts deployment, publishes a tag, finalizes a baseline, switches runtime or writes Git.
|
||||
|
||||
## Credentials
|
||||
|
||||
The worker uses two independently scoped credentials:
|
||||
|
||||
```text
|
||||
agent_token
|
||||
release_candidate_token
|
||||
```
|
||||
|
||||
`agent_token` is the Online Agent registration credential. It must be encrypted into systemd's
|
||||
credential store and must never be written to Git, Task metadata, evidence or logs.
|
||||
|
||||
`release_candidate_token` is provisioned during Owner-authorized runtime activation. Only its SHA-256 record is stored in the release API token registry; the plaintext is immediately encrypted into systemd credstore and the temporary plaintext is removed. Its only allowed scope is:
|
||||
|
||||
```text
|
||||
release:candidate
|
||||
```
|
||||
|
||||
It is explicitly not an Owner signing key and must not grant:
|
||||
|
||||
```text
|
||||
submitOwnerApproval
|
||||
startDeployment
|
||||
tag_publish
|
||||
baseline_registry_finalize
|
||||
runtime_switch
|
||||
```
|
||||
|
||||
The unit reads both credentials through systemd `LoadCredentialEncrypted`; neither plaintext token is kept in the unit, repository, Task metadata, evidence, or logs.
|
||||
|
||||
## Task contract
|
||||
|
||||
The worker will only claim Tasks whose exact required capability set is:
|
||||
|
||||
```json
|
||||
["release:candidate"]
|
||||
```
|
||||
|
||||
The Task must provide a bounded `constraints.release_candidate` payload (metadata is accepted only for compatibility):
|
||||
|
||||
```json
|
||||
{
|
||||
"candidate_id": "...",
|
||||
"source_commit": "<40 hex>",
|
||||
"version": "V2R2C2",
|
||||
"projection_id": "projection-..."
|
||||
}
|
||||
```
|
||||
|
||||
No caller-supplied URL, method, headers, command, shell, environment, credential path, deployment
|
||||
operation or arbitrary release route is accepted.
|
||||
|
||||
The release API base is host configuration, not Task input. The only release mutation invoked by the
|
||||
worker is:
|
||||
|
||||
```text
|
||||
POST /createReleaseCandidate
|
||||
```
|
||||
|
||||
## Lifecycle
|
||||
|
||||
```text
|
||||
poll
|
||||
→ claim
|
||||
→ heartbeat
|
||||
→ createReleaseCandidate
|
||||
→ result submit
|
||||
→ official verifier
|
||||
```
|
||||
|
||||
The result declares:
|
||||
|
||||
```text
|
||||
writer_lease=NONE
|
||||
changed_files=[]
|
||||
direct_main_write=false
|
||||
production_write=false
|
||||
tag_write=false
|
||||
baseline_write=false
|
||||
runtime_switch=false
|
||||
owner_approval_submitted=false
|
||||
```
|
||||
|
||||
## Remaining high-risk gate
|
||||
|
||||
Successful release-candidate creation does not authorize Production.
|
||||
|
||||
The next stage remains:
|
||||
|
||||
```text
|
||||
release candidate
|
||||
→ immutable approval envelope
|
||||
→ Owner signature
|
||||
→ startDeployment
|
||||
```
|
||||
|
||||
That separation is mandatory even when release preparation is fully unattended.
|
||||
|
||||
|
||||
## Unattended availability
|
||||
|
||||
The release worker has its own systemd watchdog rather than modifying the shared worker-availability
|
||||
guard contract.
|
||||
|
||||
Every successful Task poll and Task heartbeat emits:
|
||||
|
||||
```text
|
||||
WATCHDOG=1
|
||||
```
|
||||
|
||||
and the service uses:
|
||||
|
||||
```text
|
||||
Type=notify
|
||||
WatchdogSec=90
|
||||
Restart=always
|
||||
```
|
||||
|
||||
A process that is technically alive but no longer polling/heartbeating therefore gets restarted by
|
||||
systemd. The worker also writes `worker-liveness.json` for operator diagnostics.
|
||||
@@ -0,0 +1,216 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT=Path(__file__).resolve().parents[1]
|
||||
APP=ROOT/"active-runtime/app"
|
||||
sys.path.insert(0,str(APP))
|
||||
|
||||
import aios_release_online_agent_worker as worker
|
||||
|
||||
|
||||
def task(**overrides):
|
||||
value={
|
||||
"task_id":"r42-release-candidate-live-acceptance-20260819",
|
||||
"required_capabilities":["release:candidate"],
|
||||
"constraints":{
|
||||
"release_candidate":{
|
||||
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
|
||||
"source_commit":"5"*40,
|
||||
"version":"V2R2C2",
|
||||
"projection_id":"projection-"+"a"*32,
|
||||
}
|
||||
}
|
||||
}
|
||||
value.update(overrides)
|
||||
return value
|
||||
|
||||
|
||||
def test_release_request_accepts_only_exact_release_capability():
|
||||
req=worker.ReleaseRequest.from_task(task())
|
||||
assert req.candidate_id=="r42-release-candidate-v2r2c2-20260819"
|
||||
assert req.source_commit=="5"*40
|
||||
assert req.version=="V2R2C2"
|
||||
assert req.projection_id=="projection-"+"a"*32
|
||||
|
||||
with pytest.raises(worker.ReleaseWorkerError,match="TASK_CAPABILITY_SCOPE_INVALID"):
|
||||
worker.ReleaseRequest.from_task(task(required_capabilities=["release:candidate","git.read"]))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"payload",
|
||||
[
|
||||
{
|
||||
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
|
||||
"source_commit":"short",
|
||||
"version":"V2R2C2",
|
||||
"projection_id":"projection-"+"a"*32,
|
||||
},
|
||||
{
|
||||
"candidate_id":"bad",
|
||||
"source_commit":"5"*40,
|
||||
"version":"V2R2C2",
|
||||
"projection_id":"projection-"+"a"*32,
|
||||
},
|
||||
{
|
||||
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
|
||||
"source_commit":"5"*40,
|
||||
"version":"bad",
|
||||
"projection_id":"projection-"+"a"*32,
|
||||
},
|
||||
{
|
||||
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
|
||||
"source_commit":"5"*40,
|
||||
"version":"V2R2C2",
|
||||
"projection_id":"bad",
|
||||
},
|
||||
{
|
||||
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
|
||||
"source_commit":"5"*40,
|
||||
"version":"V2R2C2",
|
||||
"projection_id":"projection-"+"a"*32,
|
||||
"execute":True,
|
||||
},
|
||||
],
|
||||
)
|
||||
def test_release_request_rejects_invalid_or_extra_fields(payload):
|
||||
value=task()
|
||||
value["constraints"]={"release_candidate":payload}
|
||||
with pytest.raises(worker.ReleaseWorkerError):
|
||||
worker.ReleaseRequest.from_task(value)
|
||||
|
||||
|
||||
def test_result_payload_preserves_high_risk_denials():
|
||||
release=worker.ReleaseRequest.from_task(task())
|
||||
result=worker.result_payload(
|
||||
"r42-release-candidate-live-acceptance-20260819",
|
||||
release,
|
||||
{"candidate_id":release.candidate_id,"status":"BUILT"},
|
||||
)
|
||||
assert result["status"]=="SUCCEEDED"
|
||||
assert result["classification"]=="RELEASE_CANDIDATE_CREATED"
|
||||
assert result["writer_lease"]=="NONE"
|
||||
assert result["changed_files"]==[]
|
||||
assert result["direct_main_write"] is False
|
||||
assert result["production_write"] is False
|
||||
assert result["tag_write"] is False
|
||||
assert result["baseline_write"] is False
|
||||
assert result["runtime_switch"] is False
|
||||
assert result["owner_approval_submitted"] is False
|
||||
assert result["credential_material_exposed"] is False
|
||||
|
||||
|
||||
def test_result_payload_rejects_builder_candidate_mismatch():
|
||||
release=worker.ReleaseRequest.from_task(task())
|
||||
with pytest.raises(worker.ReleaseWorkerError,match="BUILDER_CANDIDATE_BINDING_INVALID"):
|
||||
worker.result_payload(
|
||||
"r42-release-candidate-live-acceptance-20260819",
|
||||
release,
|
||||
{"candidate_id":"other-release-candidate"},
|
||||
)
|
||||
|
||||
|
||||
def test_worker_calls_only_fixed_release_candidate_route(monkeypatch):
|
||||
release=worker.ReleaseRequest.from_task(task())
|
||||
calls=[]
|
||||
|
||||
def fake_http(base,method,route,token,body=None,timeout=30):
|
||||
calls.append((base,method,route,token,body))
|
||||
if route.endswith("/claim"):
|
||||
return 200,{"lease_token":"l"*32,"attempt":1}
|
||||
if route.endswith("/heartbeat"):
|
||||
return 200,{}
|
||||
if route=="/createReleaseCandidate":
|
||||
return 200,{
|
||||
"candidate_id":release.candidate_id,
|
||||
"artifact_sha256":"f"*64,
|
||||
}
|
||||
if route.endswith("/result"):
|
||||
return 200,{"status":"VERIFYING"}
|
||||
raise AssertionError(route)
|
||||
|
||||
monkeypatch.setattr(worker,"http_json",fake_http)
|
||||
monkeypatch.setattr(worker,"audit",lambda *args,**kwargs:None)
|
||||
monkeypatch.setattr(worker.Heartbeat,"first",lambda self:None)
|
||||
monkeypatch.setattr(worker.Heartbeat,"close",lambda self:None)
|
||||
monkeypatch.setattr(worker.Heartbeat,"ensure",lambda self:None)
|
||||
monkeypatch.setattr(worker.threading.Thread,"start",lambda self:None)
|
||||
|
||||
assert worker.execute_task(
|
||||
task(),
|
||||
agent_token="a"*32,
|
||||
release_token="r"*32,
|
||||
) is True
|
||||
|
||||
release_calls=[item for item in calls if item[0]==worker.RELEASE_API]
|
||||
assert len(release_calls)==1
|
||||
assert release_calls[0][1]=="POST"
|
||||
assert release_calls[0][2]=="/createReleaseCandidate"
|
||||
assert release_calls[0][4]=={
|
||||
"candidate_id":release.candidate_id,
|
||||
"source_commit":release.source_commit,
|
||||
"version":release.version,
|
||||
}
|
||||
|
||||
|
||||
|
||||
def test_write_liveness_records_release_capability(tmp_path,monkeypatch):
|
||||
path=tmp_path/"worker-liveness.json"
|
||||
monkeypatch.setattr(worker,"STATE_ROOT",tmp_path)
|
||||
monkeypatch.setattr(worker,"LIVENESS_PATH",path)
|
||||
worker.write_liveness(
|
||||
"IDLE",
|
||||
stage="poll",
|
||||
visible_count=2,
|
||||
)
|
||||
value=json.loads(path.read_text())
|
||||
assert value["schema"]=="aios.online-agent.worker-liveness.v1"
|
||||
assert value["agent_type"]=="release-preparation-worker"
|
||||
assert value["capabilities"]==["release:candidate"]
|
||||
assert value["state"]=="IDLE"
|
||||
assert value["visible_count"]==2
|
||||
|
||||
|
||||
def test_sd_notify_is_noop_without_systemd_socket(monkeypatch):
|
||||
monkeypatch.delenv("NOTIFY_SOCKET",raising=False)
|
||||
worker.sd_notify("WATCHDOG=1")
|
||||
|
||||
def test_systemd_unit_uses_two_encrypted_credentials_and_loopback_release_api():
|
||||
unit=(ROOT/"deploy/systemd/aios-release-online-agent-worker.service").read_text()
|
||||
assert "LoadCredentialEncrypted=agent_token:" in unit
|
||||
assert "LoadCredentialEncrypted=release_candidate_token:" in unit
|
||||
assert "AIOS_RELEASE_API_URL=http://127.0.0.1:18890" in unit
|
||||
assert "AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819" in unit
|
||||
assert "Type=notify" in unit
|
||||
assert "WatchdogSec=90" in unit
|
||||
assert "Restart=always" in unit
|
||||
assert "IPAddressDeny=any" in unit
|
||||
assert "IPAddressAllow=127.0.0.0/8" in unit
|
||||
assert "InaccessiblePaths=/data/aios-gitops /home" in unit
|
||||
assert "submitOwnerApproval" not in unit
|
||||
assert "startDeployment" not in unit
|
||||
|
||||
|
||||
def test_source_contains_no_high_risk_release_routes():
|
||||
source=(ROOT/"active-runtime/app/aios_release_online_agent_worker.py").read_text()
|
||||
assert '"/createReleaseCandidate"' in source
|
||||
for forbidden in (
|
||||
"/submitOwnerApproval",
|
||||
"/startDeployment",
|
||||
"/requestOrCreateTag",
|
||||
"/buildProductionArtifact",
|
||||
"/executeApprovedProductionRelease",
|
||||
):
|
||||
assert forbidden not in source
|
||||
|
||||
|
||||
def test_docs_keep_release_candidate_distinct_from_production():
|
||||
doc=(ROOT/"docs/superpowers/specs/2026-08-19-release-preparation-online-agent.md").read_text()
|
||||
assert "release:candidate" in doc
|
||||
assert "Owner signature" in doc
|
||||
assert "never\nstarts deployment" in doc or "never starts deployment" in doc
|
||||
assert "LoadCredentialEncrypted" in doc
|
||||
Reference in New Issue
Block a user