aios/aios
1
0
Fork 0
Code Issues Pull Requests 5 Actions Packages Projects Releases Wiki Activity

fix(online-agent): close automatic execution loop

Browse Source
This commit is contained in:
AIOS Governed Continuation
2026-08-17 11:54:03 +08:00
parent 71523aef6e
commit 323b615b00
23 changed files with 2244 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
active-runtime/app/aios_agent_task_retry_extension.py
+427
View File
@@ -0,0 +1,427 @@
"""Owner-controlled, CAS-bound retry extension for exhausted Online Agent Tasks."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import hmac
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import re
import sqlite3
from typing import Any, Mapping
TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,159}$")
MAX_BODY = 64 * 1024
AUTHORIZE_EXTENSION_ID_ENV = "AIOS_AGENT_TASK_AUTHORIZATION_ID"
class RetryExtensionError(RuntimeError):
def __init__(self, code: str, status: int = 409) -> None:
super().__init__(code)
self.code = code
self.status = status
def _canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def _sha(value: object) -> str:
return hashlib.sha256(_canonical(value)).hexdigest()
def _required_authorization_id() -> str:
value = os.environ.get(AUTHORIZE_EXTENSION_ID_ENV, "").strip()
if not value:
raise RuntimeError("TASK_RETRY_AUTHORIZATION_ID_MISSING")
if KEY_RE.fullmatch(value) is None:
raise RuntimeError("TASK_RETRY_AUTHORIZATION_ID_INVALID")
return value
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
def _require_string(value: object, code: str, pattern: re.Pattern[str] | None = None) -> str:
if not isinstance(value, str) or not value.strip():
raise RetryExtensionError(code, 400)
value = value.strip()
if pattern is not None and pattern.fullmatch(value) is None:
raise RetryExtensionError(code, 400)
return value
@dataclass(frozen=True)
class RetryExtensionRequest:
expected_status: str
expected_revision: int
expected_attempt: int
expected_max_attempts: int
next_max_attempts: int
previous_verification_sha256: str
authorization_id: str
idempotency_key: str
@classmethod
def from_mapping(cls, value: Mapping[str, Any]) -> "RetryExtensionRequest":
allowed = {
"expected_status", "expected_revision", "expected_attempt", "expected_max_attempts",
"next_max_attempts", "previous_verification_sha256", "authorization_id", "idempotency_key",
}
if set(value) != allowed:
raise RetryExtensionError("RETRY_EXTENSION_FIELDS_INVALID", 400)
status = _require_string(value["expected_status"], "EXPECTED_STATUS_INVALID")
if status != "FAILED":
raise RetryExtensionError("EXPECTED_STATUS_INVALID", 400)
ints: dict[str, int] = {}
for field in ("expected_revision", "expected_attempt", "expected_max_attempts", "next_max_attempts"):
item = value[field]
if not isinstance(item, int) or isinstance(item, bool) or item < 0:
raise RetryExtensionError("RETRY_EXTENSION_INTEGER_INVALID", 400)
ints[field] = item
if ints["expected_attempt"] != ints["expected_max_attempts"]:
raise RetryExtensionError("RETRY_EXTENSION_NOT_EXHAUSTED", 409)
if ints["next_max_attempts"] != ints["expected_max_attempts"] + 1:
raise RetryExtensionError("RETRY_EXTENSION_MUST_BE_BOUNDED_ONE_STEP", 400)
if ints["next_max_attempts"] > 50:
raise RetryExtensionError("RETRY_EXTENSION_LIMIT_EXCEEDED", 400)
verification = _require_string(value["previous_verification_sha256"], "VERIFICATION_SHA256_INVALID")
if re.fullmatch(r"[0-9a-f]{64}", verification) is None:
raise RetryExtensionError("VERIFICATION_SHA256_INVALID", 400)
authorization = _require_string(value["authorization_id"], "AUTHORIZATION_ID_INVALID", KEY_RE)
key = _require_string(value["idempotency_key"], "IDEMPOTENCY_KEY_INVALID", KEY_RE)
return cls(
status,
ints["expected_revision"],
ints["expected_attempt"],
ints["expected_max_attempts"],
ints["next_max_attempts"],
verification,
authorization,
key,
)
def public(self) -> dict[str, Any]:
return {
"expected_status": self.expected_status,
"expected_revision": self.expected_revision,
"expected_attempt": self.expected_attempt,
"expected_max_attempts": self.expected_max_attempts,
"next_max_attempts": self.next_max_attempts,
"previous_verification_sha256": self.previous_verification_sha256,
"authorization_id": self.authorization_id,
"idempotency_key": self.idempotency_key,
}
class RetryExtensionStore:
"""Owns the official mutation. Direct/manual DB edits remain forbidden."""
def __init__(self, db_path: Path) -> None:
self.db_path = Path(db_path)
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self.db_path, timeout=30.0)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys=ON")
connection.execute("PRAGMA busy_timeout=30000")
return connection
@staticmethod
def _columns(connection: sqlite3.Connection, table: str) -> dict[str, sqlite3.Row]:
return {str(row["name"]): row for row in connection.execute(f'PRAGMA table_info("{table}")')}
@staticmethod
def _event_table(connection: sqlite3.Connection) -> tuple[str, dict[str, sqlite3.Row]]:
tables = {
row["name"]
for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
for name in ("task_events", "events"):
if name not in tables:
continue
columns = RetryExtensionStore._columns(connection, name)
if "task_id" in columns and ("event_type" in columns or "type" in columns):
return name, columns
raise RetryExtensionError("TASK_EVENT_TABLE_UNSUPPORTED", 503)
@staticmethod
def _verification_hash(raw: object) -> str:
if raw in (None, ""):
value: object = {}
elif isinstance(raw, str):
try:
value = json.loads(raw)
except json.JSONDecodeError as exc:
raise RetryExtensionError("TASK_VERIFICATION_INVALID", 409) from exc
else:
value = raw
return _sha(value)
@staticmethod
def _ensure_extension_table(connection: sqlite3.Connection) -> None:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS task_retry_extensions (
task_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
request_sha256 TEXT NOT NULL,
authorization_id TEXT NOT NULL,
previous_verification_sha256 TEXT NOT NULL,
previous_max_attempts INTEGER NOT NULL,
next_max_attempts INTEGER NOT NULL,
expected_revision INTEGER NOT NULL,
applied_revision INTEGER NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY(task_id, idempotency_key)
)
"""
)
@staticmethod
def _insert_event(
connection: sqlite3.Connection,
task_id: str,
payload: Mapping[str, Any],
) -> None:
table, columns = RetryExtensionStore._event_table(connection)
values: dict[str, Any] = {"task_id": task_id}
if "event_type" in columns:
values["event_type"] = "TASK_RETRY_EXTENSION_AUTHORIZED"
else:
values["type"] = "TASK_RETRY_EXTENSION_AUTHORIZED"
payload_text = json.dumps(payload, sort_keys=True, separators=(",", ":"))
if "payload_json" in columns:
values["payload_json"] = payload_text
elif "payload" in columns:
values["payload"] = payload_text
else:
raise RetryExtensionError("TASK_EVENT_PAYLOAD_UNSUPPORTED", 503)
if "created_at" in columns:
values["created_at"] = _now()
if "agent_id" in columns:
values["agent_id"] = "owner-control"
if "actor_id" in columns:
values["actor_id"] = "owner-control"
names = list(values)
sql = f'INSERT INTO "{table}" ({",".join(f"""\"{name}\"""" for name in names)}) VALUES ({",".join("?" for _ in names)})'
connection.execute(sql, tuple(values[name] for name in names))
def extend(self, task_id: str, request: RetryExtensionRequest) -> dict[str, Any]:
task_id = _require_string(task_id, "TASK_ID_INVALID", TASK_RE)
required_authorization_id = _required_authorization_id()
if request.authorization_id != required_authorization_id:
raise RetryExtensionError("RETRY_EXTENSION_AUTHORIZATION_ID_MISMATCH", 403)
request_payload = request.public()
request_sha = _sha(request_payload)
with self._connect() as connection:
connection.execute("BEGIN IMMEDIATE")
self._ensure_extension_table(connection)
replay = connection.execute(
"SELECT * FROM task_retry_extensions WHERE task_id=? AND idempotency_key=?",
(task_id, request.idempotency_key),
).fetchone()
if replay is not None:
if replay["request_sha256"] != request_sha:
raise RetryExtensionError("RETRY_EXTENSION_IDEMPOTENCY_CONFLICT", 409)
connection.commit()
return {
"task_id": task_id,
"status": "QUEUED",
"attempt": request.expected_attempt,
"max_attempts": int(replay["next_max_attempts"]),
"revision": int(replay["applied_revision"]),
"idempotent": True,
"authorization_id": replay["authorization_id"],
"previous_verification_sha256": replay["previous_verification_sha256"],
}
columns = self._columns(connection, "tasks")
required = {
"task_id", "status", "attempt", "max_attempts", "revision", "claimed_by",
"lease_token_hash", "lease_expires_at", "verification_json",
}
if not required.issubset(columns):
raise RetryExtensionError("TASK_SCHEMA_RETRY_EXTENSION_UNSUPPORTED", 503)
task = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
if task is None:
raise RetryExtensionError("TASK_NOT_FOUND", 404)
actual_verification_sha = self._verification_hash(task["verification_json"])
if actual_verification_sha != request.previous_verification_sha256:
raise RetryExtensionError("RETRY_EXTENSION_VERIFICATION_DRIFT", 409)
if (
task["status"] != request.expected_status
or int(task["revision"]) != request.expected_revision
or int(task["attempt"]) != request.expected_attempt
or int(task["max_attempts"]) != request.expected_max_attempts
):
raise RetryExtensionError("RETRY_EXTENSION_CAS_MISMATCH", 409)
for lease_field in ("lease_token_hash", "lease_expires_at"):
if lease_field in columns and task[lease_field] not in (None, ""):
raise RetryExtensionError("RETRY_EXTENSION_ACTIVE_LEASE_DENY", 409)
previous_claimed_by = task["claimed_by"]
applied_revision = request.expected_revision + 1
cursor = connection.execute(
"""
UPDATE tasks
SET status='QUEUED', max_attempts=?, revision=?, claimed_by=NULL,
lease_token_hash=NULL, lease_expires_at=NULL
WHERE task_id=? AND status=? AND revision=? AND attempt=? AND max_attempts=?
AND claimed_by IS ?
AND lease_token_hash IS NULL AND lease_expires_at IS NULL
""",
(
request.next_max_attempts,
applied_revision,
task_id,
request.expected_status,
request.expected_revision,
request.expected_attempt,
request.expected_max_attempts,
previous_claimed_by,
),
)
if cursor.rowcount != 1:
raise RetryExtensionError("RETRY_EXTENSION_CAS_MISMATCH", 409)
event_payload = {
"schema": "aios.task-retry-extension.event.v1",
"task_id": task_id,
"authorization_id": request.authorization_id,
"idempotency_key": request.idempotency_key,
"previous_verification_sha256": request.previous_verification_sha256,
"previous_max_attempts": request.expected_max_attempts,
"next_max_attempts": request.next_max_attempts,
"previous_revision": request.expected_revision,
"applied_revision": applied_revision,
"request_sha256": request_sha,
"previous_claimed_by": previous_claimed_by,
"stale_terminal_claim_normalized": previous_claimed_by not in (None, ""),
}
self._insert_event(connection, task_id, event_payload)
connection.execute(
"""
INSERT INTO task_retry_extensions(
task_id,idempotency_key,request_sha256,authorization_id,previous_verification_sha256,
previous_max_attempts,next_max_attempts,expected_revision,applied_revision,created_at
) VALUES (?,?,?,?,?,?,?,?,?,?)
""",
(
task_id,
request.idempotency_key,
request_sha,
request.authorization_id,
request.previous_verification_sha256,
request.expected_max_attempts,
request.next_max_attempts,
request.expected_revision,
applied_revision,
_now(),
),
)
connection.commit()
return {
"task_id": task_id,
"status": "QUEUED",
"attempt": request.expected_attempt,
"max_attempts": request.next_max_attempts,
"revision": applied_revision,
"idempotent": False,
"authorization_id": request.authorization_id,
"previous_verification_sha256": request.previous_verification_sha256,
}
class RetryExtensionHandler(BaseHTTPRequestHandler):
store: RetryExtensionStore | None = None
control_token: str = ""
def log_message(self, *_args: Any) -> None:
return
def _json(self, value: Mapping[str, Any], status: int) -> None:
data = json.dumps(dict(value), sort_keys=True, separators=(",", ":")).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def _authorized(self) -> bool:
header = self.headers.get("Authorization", "")
return (
header.startswith("Bearer ")
and bool(self.control_token)
and hmac.compare_digest(header[7:], self.control_token)
)
def do_GET(self) -> None:
if self.path == "/health":
self._json({"status": "ok", "service": "aios-agent-task-retry-extension"}, 200)
else:
self._json({"error": "NOT_FOUND"}, 404)
def do_POST(self) -> None:
match = re.fullmatch(r"/v1/tasks/([^/]+)/retry-extension", self.path)
if match is None:
self._json({"error": "NOT_FOUND"}, 404)
return
if not self._authorized():
self._json({"error": "UNAUTHORIZED"}, 401)
return
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
self._json({"error": "CONTENT_LENGTH_INVALID"}, 400)
return
if length < 1 or length > MAX_BODY:
self._json({"error": "REQUEST_SIZE_INVALID"}, 413)
return
try:
value = json.loads(self.rfile.read(length).decode("utf-8"))
if not isinstance(value, Mapping):
raise ValueError
request = RetryExtensionRequest.from_mapping(value)
if self.store is None:
raise RetryExtensionError("RUNTIME_UNAVAILABLE", 503)
result = self.store.extend(match.group(1), request)
self._json(result, 200 if result["idempotent"] else 201)
except RetryExtensionError as exc:
self._json({"error": exc.code}, exc.status)
except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
self._json({"error": "REQUEST_INVALID"}, 400)
def load_control_token(path: Path) -> str:
if path.is_symlink() or not path.is_file():
raise RuntimeError("TASK_RETRY_CONTROL_CREDENTIAL_UNAVAILABLE")
value = path.read_text(encoding="utf-8").strip()
if len(value) < 32:
raise RuntimeError("TASK_RETRY_CONTROL_CREDENTIAL_INVALID")
return value
def serve() -> None:
db = Path(os.environ.get("AIOS_AGENT_TASK_DB", "/var/lib/aios-agent-task/state.sqlite3"))
_required_authorization_id()
token_path = Path(
os.environ.get(
"AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
str(Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials")) / "agent_task_control_token"),
)
)
host = os.environ.get("AIOS_TASK_RETRY_HOST", "127.0.0.1")
port = int(os.environ.get("AIOS_TASK_RETRY_PORT", "18941"))
RetryExtensionHandler.store = RetryExtensionStore(db)
RetryExtensionHandler.control_token = load_control_token(token_path)
ThreadingHTTPServer((host, port), RetryExtensionHandler).serve_forever()
if __name__ == "__main__":
serve()
active-runtime/app/aios_chatgpt_bridge_v2.py
+15
View File
@@ -13,6 +13,7 @@ BRIDGE_OPERATIONS = (
"getGovernedWriteCapabilities",
"createGovernedTask",
"projectOnlineAgentExecution",
"registerOnlineAgentExecutionPlan",
"acquireWriterLease",
"bootstrapWriterLease",
"renewWriterLease",
@@ -34,6 +35,10 @@ BRIDGE_OPERATIONS = (
OPERATION_REQUIRED_FIELDS = {
"createGovernedTask": ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key"),
"projectOnlineAgentExecution": ("task_id",),
"registerOnlineAgentExecutionPlan": (
"schema", "task_id", "job_id", "source_head_oid", "generation",
"idempotency_key", "execution",
),
"acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
"bootstrapWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
"renewWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
@@ -63,6 +68,15 @@ BRIDGE_OPERATION_CONTRACTS = {
"idempotency_key", "action", "execution_digest",
),
},
"registerOnlineAgentExecutionPlan": {
"method": "POST",
"path": "/v2/execution-plan",
"required_fields": OPERATION_REQUIRED_FIELDS["registerOnlineAgentExecutionPlan"],
"response_fields": (
"schema", "task_id", "job_id", "source_head_oid", "generation",
"idempotency_key", "plan_digest", "execution", "execution_digest",
),
},
"acquireWriterLease": {
"method": "POST",
"path": "/v2/writer-lease",
@@ -218,6 +232,7 @@ class BridgeV2Adapter:
def createGovernedTask(self, request: Mapping[str, Any]) -> Any: return self._call("createGovernedTask", request)
def projectOnlineAgentExecution(self, request: Mapping[str, Any]) -> Any: return self._call("projectOnlineAgentExecution", request)
def registerOnlineAgentExecutionPlan(self, request: Mapping[str, Any]) -> Any: return self._call("registerOnlineAgentExecutionPlan", request)
def acquireWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("acquireWriterLease", request)
def bootstrapWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("bootstrapWriterLease", request)
def renewWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("renewWriterLease", request)
active-runtime/app/aios_chatgpt_bridge_v2_server.py
+58 -5
View File
@@ -24,6 +24,7 @@ from aios_control_plane_snapshot import build_control_plane_snapshot
from aios_gitops_job_controller import GitOpsJobController, JobError, JobState, JobStore
from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBinding, GovernedTaskBootstrap, _legacy_safe, _task_key
from aios_online_agent_execution import ExecutionProjectionError, OnlineTaskProjectionClient
from aios_online_agent_plan_store import ExecutionPlanError, ExecutionPlanStore
from governed_production_release import build_executor_config
MAX_BODY = 2 * 1024 * 1024
@@ -32,6 +33,7 @@ ROUTES = {
"/v2/capabilities": "getGovernedWriteCapabilities",
"/v2/governed-task": "createGovernedTask",
"/v2/execution-envelope": "projectOnlineAgentExecution",
"/v2/execution-plan": "registerOnlineAgentExecutionPlan",
"/v2/writer-lease": "acquireWriterLease",
"/v2/writer-lease/bootstrap": "bootstrapWriterLease",
"/v2/writer-lease/renew": "renewWriterLease",
@@ -89,6 +91,9 @@ class GovernedRuntime:
)
self.projection_outbox_root = config.state_root / "online-task-projection-outbox"
self.projection_outbox_root.mkdir(parents=True, exist_ok=True)
self.execution_plan_store = ExecutionPlanStore(config.state_root / "online-agent-execution-plans")
self._projection_locks_guard = threading.Lock()
self._projection_locks: dict[str, threading.RLock] = {}
self.runtimes: dict[tuple[str, int], tuple[Any, GitOpsJobController, ContinuationExecutor]] = {}
self.controllers: dict[str, GitOpsJobController] = {}
self._recovery_lock = threading.RLock()
@@ -111,12 +116,21 @@ class GovernedRuntime:
if self.online_task_projector is not None:
threading.Thread(target=self._projection_reconciler, name="online-task-projection-reconciler", daemon=True).start()
def _projection_lock(self, task_id: str) -> threading.RLock:
guard = getattr(self, "_projection_locks_guard", None)
if guard is None:
guard = threading.Lock()
self._projection_locks_guard = guard
self._projection_locks = {}
with guard:
return self._projection_locks.setdefault(task_id, threading.RLock())
def _projection_outbox_path(self, task_id: str) -> Path:
return self.projection_outbox_root / (_task_key(task_id) + ".json")
def _write_projection_outbox(self, task_id: str, projection: Mapping[str, Any], binding: Any) -> Path:
path = self._projection_outbox_path(task_id)
temporary = path.with_suffix(".tmp")
temporary = path.with_name(path.name + f".{os.getpid()}.{threading.get_ident()}.tmp")
temporary.write_text(json.dumps({"task": dict(projection), "binding": asdict(binding)}, sort_keys=True, separators=(",", ":")), encoding="utf-8")
temporary.replace(path)
return path
@@ -136,6 +150,13 @@ class GovernedRuntime:
while True:
for path in sorted(self.projection_outbox_root.glob("*.json")):
try:
value = json.loads(path.read_text(encoding="utf-8"))
binding = value.get("binding") if isinstance(value, Mapping) else None
task_id = str(binding.get("task_id") or "") if isinstance(binding, Mapping) else ""
if not task_id:
continue
with self._projection_lock(task_id):
if path.exists():
self._deliver_projection_outbox(path)
except Exception:
continue
@@ -521,6 +542,7 @@ class GovernedRuntime:
expected_job_id=str(result.get("job_id") or "") or None,
)
durable_binding = self.bootstrap.get_binding(result["task_id"])
with self._projection_lock(result["task_id"]):
path = self._write_projection_outbox(result["task_id"], projection, durable_binding)
created = self._deliver_projection_outbox(path)
return {
@@ -541,13 +563,29 @@ class GovernedRuntime:
execution = request.get("execution")
if execution is not None and not isinstance(execution, Mapping):
raise ValueError("EXECUTION_REQUEST_INVALID")
task_id = str(request["task_id"])
durable = self.bootstrap.get_binding(task_id)
binding = self.bootstrap.execution_binding(durable)
expected_job_id = str(request.get("job_id") or "") or None
if expected_job_id and expected_job_id != binding.job_id:
raise BootstrapError("JOB_BINDING_CONFLICT")
if execution is None:
planned = self.execution_plan_store.resolve(binding)
if planned is not None:
execution = planned["execution"]
return self.bootstrap.execution_envelope(
str(request["task_id"]),
task_id,
execution,
required_capabilities=tuple(capabilities),
expected_job_id=str(request.get("job_id") or "") or None,
expected_job_id=expected_job_id,
)
def register_online_agent_execution_plan(self, request: Mapping[str, Any]) -> dict[str, Any]:
task_id = str(request["task_id"])
durable = self.bootstrap.get_binding(task_id)
binding = self.bootstrap.execution_binding(durable)
return self.execution_plan_store.register(binding, request)
def acquire_lease(self, request: Mapping[str, Any]) -> dict[str, Any]:
binding, controller, _ = self._runtime(request)
req = binding.request
@@ -1443,7 +1481,19 @@ class GovernedRuntime:
controller = self._controller_for_job(job_id)
row = controller.store.read(job_id)
if action == "get":
return controller.store._receipt(row, duplicate=True).as_dict()
response = controller.store._receipt(row, duplicate=True).as_dict()
try:
with controller.store._connect() as connection:
connection.row_factory = sqlite3.Row
leases = connection.execute(
"SELECT lease_id,job_id,branch,lineage_anchor_oid,current_head_oid,generation,"
"lease_expires_at,state,revision FROM branch_leases WHERE job_id=? ORDER BY revision DESC",
(job_id,),
).fetchall()
except sqlite3.Error as exc:
raise RuntimeError("JOB_DB_UNAVAILABLE") from exc
response["branch_leases"] = [dict(item) for item in leases]
return response
if action == "resume":
return controller.resume(job_id).as_dict()
return self._executor_from_row(row, controller).reconcile(job_id)
@@ -1509,6 +1559,7 @@ class GovernedRuntime:
return {
"createGovernedTask": self.create_task,
"projectOnlineAgentExecution": self.project_online_agent_execution,
"registerOnlineAgentExecutionPlan": self.register_online_agent_execution_plan,
"acquireWriterLease": self.acquire_lease,
"bootstrapWriterLease": self.bootstrap_writer_lease,
"renewWriterLease": self.renew_writer_lease,
@@ -1559,6 +1610,8 @@ class BridgeHandler(BaseHTTPRequestHandler):
allowed.update({"owner_approval", "execute"})
if operation == "projectOnlineAgentExecution":
allowed.update({"execution", "required_capabilities"})
if operation == "registerOnlineAgentExecutionPlan":
allowed.update({"execution", "source_head_oid", "expected_previous_plan_digest"})
if operation == "createGovernedTask":
allowed.update({"online_task", "execution", "require_online_task_projection", "priority", "max_attempts", "parent_task_id", "parent_job_id"})
if set(body) - allowed: raise _HTTPError("UNKNOWN_FIELDS")
@@ -1615,7 +1668,7 @@ class BridgeHandler(BaseHTTPRequestHandler):
try:
self._allowed(operation, body); self._json(getattr(self.adapter, operation)(body))
except _HTTPError as exc: self._json({"error": str(exc)}, 413 if str(exc) == "REQUEST_TOO_LARGE" else 400)
except (BootstrapError, ExecutionProjectionError, ValueError, BridgeV2Error) as exc: self._json({"error": str(exc)}, 409 if "CONFLICT" in str(exc) else 400)
except (BootstrapError, ExecutionProjectionError, ExecutionPlanError, ValueError, BridgeV2Error) as exc: self._json({"error": str(exc)}, 409 if any(token in str(exc) for token in ("CONFLICT", "DRIFT", "CAS")) else 400)
except Exception as exc: self._json({"error": str(exc) or type(exc).__name__}, 409 if any(token in str(exc) for token in ("CONFLICT", "EXPECTATION", "MONOTONIC")) else 503)
active-runtime/app/aios_continuation_executor.py
+29 -4
View File
@@ -157,7 +157,7 @@ class DetachedGitWorkspace:
def materialize_files(self, files: Mapping[str, bytes]) -> None:
dirty = self._dirty_paths()
expected_paths = set(files)
if dirty and dirty != expected_paths:
if dirty and not dirty.issubset(expected_paths):
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
for relative, content in files.items():
target = (self.path / relative).resolve()
@@ -499,7 +499,19 @@ class ContinuationExecutor:
current = self.controller.store.read(binding.job_id)
current_state = JobState(current["state"])
if current_state == JobState.COMPLETED:
return self.controller.store._receipt(current, duplicate=True).as_dict()
cycle = self.controller.begin_continuation_cycle(
binding.job_id,
idempotency_key=commit.idempotency_key,
expected_head_oid=binding.request.expected_branch_oid,
generation=binding.request.generation,
expected_revision=int(current["revision"]),
)
if cycle.get("completed"):
return self.controller.store._receipt(
self.controller.store.read(binding.job_id), duplicate=True
).as_dict()
current = self.controller.store.read(binding.job_id)
current_state = JobState(current["state"])
if current_state == JobState.REMOTE_VERIFIED:
receipt = json.loads(str(current["receipt_json"] or "{}"))
commit_oid = str(receipt.get("authority_oid") or "")
@@ -507,12 +519,18 @@ class ContinuationExecutor:
raise ContinuationExecutionError("GOVERNED_COMMIT_OID_INVALID")
if before_complete is not None:
before_complete(commit_oid)
return self.controller.advance(
result = self.controller.advance(
binding.job_id,
JobState.COMPLETED,
side_effect_state="REMOTE_VERIFIED",
receipt=receipt,
).as_dict()
self.controller.complete_continuation_cycle(
binding.job_id,
idempotency_key=commit.idempotency_key,
commit_oid=commit_oid,
)
return result
self._validate_preflight(binding, job_id=binding.job_id)
request = binding.request
projected = False
@@ -576,12 +594,19 @@ class ContinuationExecutor:
self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
except Exception:
return self._remote_pending(binding.job_id, authority_oid, new_oid)
return self._finish_remote_verification(
result = self._finish_remote_verification(
binding.job_id,
authority_oid,
new_oid,
before_complete=before_complete,
)
if str(result.get("state") or "") == JobState.COMPLETED.value:
self.controller.complete_continuation_cycle(
binding.job_id,
idempotency_key=commit.idempotency_key,
commit_oid=new_oid,
)
return result
finally:
if projected:
self.credential.revoke()
active-runtime/app/aios_gitops_job_controller.py
+186 -1
View File
@@ -230,7 +230,15 @@ class JobStore:
kind TEXT NOT NULL, digest TEXT NOT NULL, payload_json TEXT NOT NULL,
created_at TEXT NOT NULL, FOREIGN KEY(job_id) REFERENCES jobs(job_id)
);
INSERT OR REPLACE INTO metadata(key, value) VALUES('schema_version', '1');
CREATE TABLE IF NOT EXISTS continuation_cycles (
job_id TEXT NOT NULL, idempotency_key TEXT NOT NULL,
sequence INTEGER NOT NULL, expected_head_oid TEXT NOT NULL,
generation INTEGER NOT NULL, state TEXT NOT NULL,
commit_oid TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
PRIMARY KEY(job_id,idempotency_key), UNIQUE(job_id,sequence),
FOREIGN KEY(job_id) REFERENCES jobs(job_id)
);
INSERT OR REPLACE INTO metadata(key, value) VALUES('schema_version', '2');
"""
)
except sqlite3.Error as exc:
@@ -531,6 +539,183 @@ class GitOpsJobController:
except sqlite3.Error as exc:
raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
def begin_continuation_cycle(
self,
job_id: str,
*,
idempotency_key: str,
expected_head_oid: str,
generation: int,
expected_revision: int,
) -> dict[str, Any]:
if (
not idempotency_key
or len(idempotency_key) > 128
or not all(character.isalnum() or character in "._:-" for character in idempotency_key)
or len(expected_head_oid) != 40
or any(character not in "0123456789abcdef" for character in expected_head_oid)
or generation < 1
):
raise JobError("CONTINUATION_CYCLE_BINDING_INVALID")
try:
with self.store._connect() as connection:
connection.execute("BEGIN IMMEDIATE")
row = connection.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
if row is None:
raise JobError("JOB_NOT_FOUND")
existing = connection.execute(
"SELECT * FROM continuation_cycles WHERE job_id=? AND idempotency_key=?",
(job_id, idempotency_key),
).fetchone()
if existing is not None:
if str(existing["expected_head_oid"]) != expected_head_oid or int(existing["generation"]) != generation:
raise JobError("CONTINUATION_CYCLE_BINDING_CONFLICT")
if str(existing["state"]) == "COMPLETED":
connection.commit()
return {"completed": True, "commit_oid": str(existing["commit_oid"] or ""), "idempotent": True}
if str(row["state"]) == JobState.COMPLETED.value:
receipt = json.loads(str(row["receipt_json"] or "{}"))
commit_oid = str(receipt.get("authority_oid") or "")
lease = connection.execute(
"SELECT current_head_oid FROM branch_leases WHERE job_id=? AND generation=? AND state='ACTIVE'",
(job_id, generation),
).fetchall()
if (
len(lease) == 1
and len(commit_oid) == 40
and str(lease[0]["current_head_oid"] or "") == commit_oid
and receipt.get("remote_verified") is True
):
now = utc_now()
connection.execute(
"UPDATE continuation_cycles SET state='COMPLETED',commit_oid=?,updated_at=? "
"WHERE job_id=? AND idempotency_key=? AND state='ACTIVE'",
(commit_oid, now, job_id, idempotency_key),
)
connection.commit()
return {"completed": True, "commit_oid": commit_oid, "idempotent": True}
raise JobError("CONTINUATION_CYCLE_RECOVERY_REQUIRED")
connection.commit()
return {"completed": False, "commit_oid": None, "idempotent": True}
if int(row["revision"]) != expected_revision:
raise JobError("REVISION_CAS_CONFLICT")
if str(row["state"]) != JobState.COMPLETED.value:
raise JobError("CONTINUATION_CYCLE_JOB_STATE_INVALID")
if str(row["branch"]) in {"main", "master"} or str(row["branch"]).startswith(("refs/", "tags/")):
raise JobError("PROTECTED_BRANCH_REJECTED")
receipt = json.loads(str(row["receipt_json"] or "{}"))
if receipt.get("remote_verified") is not True or any(
str(receipt.get(field) or "") != expected_head_oid
for field in ("authority_oid", "api_branch_oid", "git_remote_oid")
):
raise JobError("CONTINUATION_CYCLE_RECEIPT_INVALID")
leases = connection.execute(
"SELECT * FROM branch_leases WHERE job_id=? AND branch=? AND generation=? AND state='ACTIVE'",
(job_id, str(row["branch"]), generation),
).fetchall()
if len(leases) != 1 or str(leases[0]["current_head_oid"] or "") != expected_head_oid:
raise JobError("CONTINUATION_CYCLE_LEASE_INVALID")
if connection.execute(
"SELECT 1 FROM continuation_cycles WHERE job_id=? AND state='ACTIVE' LIMIT 1",
(job_id,),
).fetchone() is not None:
raise JobError("CONTINUATION_CYCLE_ACTIVE_CONFLICT")
sequence = int(connection.execute(
"SELECT COALESCE(MAX(sequence),0)+1 FROM continuation_cycles WHERE job_id=?",
(job_id,),
).fetchone()[0])
now = utc_now()
connection.execute(
"INSERT INTO continuation_cycles(job_id,idempotency_key,sequence,expected_head_oid,generation,state,commit_oid,created_at,updated_at) "
"VALUES(?,?,?,?,?,'ACTIVE',NULL,?,?)",
(job_id, idempotency_key, sequence, expected_head_oid, generation, now, now),
)
next_revision = expected_revision + 1
updated = connection.execute(
"UPDATE jobs SET state=?,side_effect_state=?,revision=?,updated_at=? WHERE job_id=? AND revision=?",
(JobState.LEASED.value, "CONTINUATION_LEASED", next_revision, now, job_id, expected_revision),
)
if updated.rowcount != 1:
raise JobError("REVISION_CAS_CONFLICT")
connection.execute(
"INSERT INTO state_transitions(job_id,from_state,to_state,revision,side_effect_state,created_at) VALUES(?,?,?,?,?,?)",
(job_id, JobState.COMPLETED.value, JobState.LEASED.value, next_revision, "CONTINUATION_LEASED", now),
)
cycle_receipt = {
"idempotency_key": idempotency_key,
"expected_head_oid": expected_head_oid,
"generation": generation,
"sequence": sequence,
}
connection.execute(
"INSERT INTO receipts(receipt_id,job_id,kind,digest,payload_json,created_at) VALUES(?,?,?,?,?,?)",
(
f"receipt-{job_id}-{next_revision}",
job_id,
"CONTINUATION_LEASED",
hashlib.sha256(canonical_json(cycle_receipt)).hexdigest(),
json.dumps(cycle_receipt, sort_keys=True),
now,
),
)
result = dict(connection.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone())
connection.commit()
except (JobError, BindingConflict):
raise
except sqlite3.Error as exc:
raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
result["request"] = json.loads(result["request_json"])
self.store._mirror(result)
return {"completed": False, "commit_oid": None, "idempotent": False}
def complete_continuation_cycle(self, job_id: str, *, idempotency_key: str, commit_oid: str) -> bool:
if len(commit_oid) != 40 or any(character not in "0123456789abcdef" for character in commit_oid):
raise JobError("CONTINUATION_CYCLE_COMMIT_OID_INVALID")
try:
with self.store._connect() as connection:
connection.execute("BEGIN IMMEDIATE")
cycle = connection.execute(
"SELECT * FROM continuation_cycles WHERE job_id=? AND idempotency_key=?",
(job_id, idempotency_key),
).fetchone()
if cycle is None:
connection.commit()
return False
if str(cycle["state"]) == "COMPLETED":
if str(cycle["commit_oid"] or "") != commit_oid:
raise JobError("CONTINUATION_CYCLE_COMMIT_CONFLICT")
connection.commit()
return True
row = connection.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
leases = connection.execute(
"SELECT current_head_oid FROM branch_leases WHERE job_id=? AND generation=? AND state='ACTIVE'",
(job_id, int(cycle["generation"])),
).fetchall()
receipt = json.loads(str(row["receipt_json"] or "{}")) if row is not None else {}
if (
row is None
or str(row["state"]) != JobState.COMPLETED.value
or receipt.get("remote_verified") is not True
or str(receipt.get("authority_oid") or "") != commit_oid
or len(leases) != 1
or str(leases[0]["current_head_oid"] or "") != commit_oid
):
raise JobError("CONTINUATION_CYCLE_COMPLETION_EVIDENCE_INVALID")
now = utc_now()
updated = connection.execute(
"UPDATE continuation_cycles SET state='COMPLETED',commit_oid=?,updated_at=? "
"WHERE job_id=? AND idempotency_key=? AND state='ACTIVE'",
(commit_oid, now, job_id, idempotency_key),
)
if updated.rowcount != 1:
raise JobError("CONTINUATION_CYCLE_CAS_CONFLICT")
connection.commit()
return True
except (JobError, BindingConflict):
raise
except sqlite3.Error as exc:
raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
def advance(self, job_id: str, state: JobState, *, side_effect_state: str, receipt: Mapping[str, Any] | None = None, expected_revision: int | None = None) -> JobReceipt:
return self._advance(
job_id,
active-runtime/app/aios_governed_task_bootstrap.py
+41 -4
View File
@@ -78,6 +78,17 @@ class GovernedTaskBootstrap:
finally:
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
@contextmanager
def _branch_lock(self, repository: str, target_branch: str):
key = hashlib.sha256((repository + "\0" + target_branch).encode("utf-8")).hexdigest()
path = self.lock_root / ("branch-" + key + ".lock")
with path.open("a+b") as handle:
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
def _read_existing(self, task_id: str) -> GovernedTaskBinding | None:
path = self._metadata(task_id)
if not path.exists():
@@ -95,17 +106,40 @@ class GovernedTaskBootstrap:
raise BootstrapError("TASK_EXECUTION_PROJECTION_MISSING")
return existing
@staticmethod
def execution_binding(binding: GovernedTaskBinding) -> GovernedExecutionBinding:
def execution_binding(self, binding: GovernedTaskBinding) -> GovernedExecutionBinding:
expected = binding.expected_branch_oid
mode = binding.branch_mode
if mode == "CREATE_NEW":
current = self.authority.read_ref("refs/heads/" + binding.target_branch)
if current is not None:
store, _ = self._store(binding.task_id)
with store._connect() as connection:
rows = connection.execute(
"SELECT current_head_oid,generation,state,lineage_anchor_oid "
"FROM branch_leases WHERE job_id=? AND branch=? AND state='ACTIVE' "
"ORDER BY revision DESC",
(binding.job_id, binding.target_branch),
).fetchall()
if len(rows) != 1:
raise BootstrapError("SAME_JOB_CONTINUATION_LEASE_INVALID")
row = rows[0]
if (
str(row["current_head_oid"] or "") != current
or int(row["generation"] or 0) != binding.generation
or str(row["lineage_anchor_oid"] or "") != binding.base_oid
):
raise BootstrapError("SAME_JOB_CONTINUATION_BINDING_INVALID")
expected = current
mode = "CONTINUE_EXISTING"
return GovernedExecutionBinding.from_mapping(
{
"task_id": binding.task_id,
"job_id": binding.job_id,
"repository": binding.repository,
"base_oid": binding.base_oid,
"expected_branch_oid": binding.expected_branch_oid,
"expected_branch_oid": expected,
"target_branch": binding.target_branch,
"branch_mode": binding.branch_mode,
"branch_mode": mode,
"generation": binding.generation,
"idempotency_key": binding.idempotency_key,
}
@@ -148,7 +182,10 @@ class GovernedTaskBootstrap:
task_id = str(request.get("task_id") or "")
if not task_id:
raise BootstrapError("TASK_BINDING_INVALID")
repository = str(request.get("repository") or "")
target_branch = str(request.get("target_branch") or "")
with self._task_lock(task_id):
with self._branch_lock(repository, target_branch):
return self._create_locked(request)
def _create_locked(self, request: Mapping[str, Any]) -> dict[str, Any]:
active-runtime/app/aios_isolated_executor.py
+306
View File
@@ -0,0 +1,306 @@
"""Uncredentialed deterministic executor for governed Online Agent changes.
This process never receives Agent/Owner/Bridge credentials and never writes Git refs.
It operates only on an exchange bundle and returns hash-bound changed files + test evidence.
"""
from __future__ import annotations
import base64
from dataclasses import dataclass
import errno
import hashlib
import io
import json
import os
from pathlib import Path
import re
import shutil
import socket
import subprocess
import tarfile
import tempfile
import time
from typing import Any, Mapping, Sequence
ALLOWED_PROGRAMS = {"/usr/bin/python3", "/usr/bin/git", "/bin/true", "/usr/bin/true"}
MAX_ARCHIVE_BYTES = 64 * 1024 * 1024
MAX_FILE_BYTES = 4 * 1024 * 1024
SECRET_ENV_TOKENS = ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "AUTHORIZATION", "PRIVATE_KEY")
REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,254}$")
SENSITIVE_VALUE_RE = re.compile(
r"(?i)(authorization\s*[:=]\s*bearer\s+\S+|authorization\s+bearer\s+\S+|"
r"(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)"
)
class IsolatedExecutorError(RuntimeError):
pass
def sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def safe_relative(value: object) -> str:
if not isinstance(value, str) or not value or value.startswith("/"):
raise IsolatedExecutorError("PATH_INVALID")
path = Path(value)
if ".." in path.parts or "." in path.parts or any(part in {"", ".git"} for part in path.parts):
raise IsolatedExecutorError("PATH_INVALID")
return path.as_posix()
def resolve_under(root: Path, relative: str) -> Path:
root = root.resolve()
path = (root / safe_relative(relative)).resolve(strict=False)
if path == root or root not in path.parents:
raise IsolatedExecutorError("PATH_ESCAPE")
return path
def _credential_exposure(directory: Path) -> str | None:
"""Return an exposure code; EACCES is the intended credential isolation proof."""
try:
if not directory.exists():
return None
if directory.is_symlink():
return "CREDENTIAL_MOUNT_SYMLINK"
if os.access(directory, os.R_OK):
return "CREDENTIAL_MOUNT_ACCESSIBLE"
if directory.is_dir():
for child in directory.iterdir():
if child.is_file() and os.access(child, os.R_OK):
return "CREDENTIAL_FILE_ACCESSIBLE"
except PermissionError:
return None
return None
def assert_uncredentialed() -> None:
credentials = Path("/run/credentials")
candidates = {credentials}
credentials_directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
candidates.add(credentials_directory)
leaked = sorted(
key for key, value in os.environ.items()
if value
and (any(token in key.upper() for token in SECRET_ENV_TOKENS) or SENSITIVE_VALUE_RE.search(str(value)))
)
if leaked:
raise IsolatedExecutorError("CREDENTIAL_ENV_EXPOSED:" + ",".join(leaked))
for directory in candidates:
exposure = _credential_exposure(directory)
if exposure is not None:
raise IsolatedExecutorError(exposure)
# In the intended PrivateNetwork namespace host loopback services are unreachable.
try:
probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
except OSError as exc:
if exc.errno in {errno.EACCES, errno.EPERM, errno.EAFNOSUPPORT}:
return
raise IsolatedExecutorError("NETWORK_ISOLATION_PROBE_FAILED") from exc
probe.settimeout(0.15)
try:
if probe.connect_ex(("127.0.0.1", 18892)) == 0:
raise IsolatedExecutorError("GOVERNED_BRIDGE_REACHABLE")
finally:
probe.close()
def extract_source(archive: Path, root: Path) -> None:
if archive.is_symlink() or not archive.is_file() or archive.stat().st_size > MAX_ARCHIVE_BYTES:
raise IsolatedExecutorError("SOURCE_ARCHIVE_INVALID")
with tarfile.open(archive, "r:gz") as tf:
for member in tf.getmembers():
path = Path(member.name)
if path.is_absolute() or ".." in path.parts or ".git" in path.parts:
raise IsolatedExecutorError("SOURCE_ARCHIVE_PATH_INVALID")
if member.issym() or member.islnk():
raise IsolatedExecutorError("SOURCE_ARCHIVE_LINK_DENY")
tf.extractall(root, filter="data")
def run_command(item: Mapping[str, Any], root: Path) -> dict[str, Any]:
argv = item.get("argv")
if not isinstance(argv, Sequence) or isinstance(argv, (str, bytes)) or not argv:
raise IsolatedExecutorError("COMMAND_INVALID")
argv = [str(part) for part in argv]
if argv[0] not in ALLOWED_PROGRAMS:
raise IsolatedExecutorError("COMMAND_PROGRAM_DENY")
if argv[0] == "/usr/bin/python3" and "-m" in argv:
index = argv.index("-m")
if index + 1 >= len(argv) or argv[index + 1] not in {"pytest", "py_compile"}:
raise IsolatedExecutorError("PYTHON_MODULE_DENY")
if argv[0] == "/usr/bin/git" and any(
token in {"push", "update-ref", "commit", "merge", "rebase", "reset", "clean", "branch"}
for token in argv[1:]
):
raise IsolatedExecutorError("GIT_MUTATION_DENY")
timeout = int(item.get("timeout_seconds") or 300)
if not 1 <= timeout <= 1800:
raise IsolatedExecutorError("COMMAND_TIMEOUT_INVALID")
cwd = root if item.get("cwd", ".") == "." else resolve_under(root, str(item.get("cwd")))
started = time.monotonic()
completed = subprocess.run(
argv,
cwd=cwd,
capture_output=True,
text=True,
check=False,
timeout=timeout,
env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8"},
)
result = {
"label": str(item.get("label") or "command"),
"exit_code": completed.returncode,
"stdout_sha256": sha256(completed.stdout.encode("utf-8", "replace")),
"stderr_sha256": sha256(completed.stderr.encode("utf-8", "replace")),
"duration_ms": int((time.monotonic() - started) * 1000),
}
if completed.returncode != 0:
raise IsolatedExecutorError(
"COMMAND_FAILED:" + result["label"] + ":" + str(completed.returncode)
)
return result
def apply_changes(request: Mapping[str, Any], root: Path) -> list[str]:
changes = request.get("change_set")
if not isinstance(changes, list) or not changes:
raise IsolatedExecutorError("CHANGE_SET_INVALID")
changed: list[str] = []
for change in changes:
if not isinstance(change, Mapping):
raise IsolatedExecutorError("CHANGE_INVALID")
relative = safe_relative(change.get("path"))
path = resolve_under(root, relative)
current = path.read_bytes() if path.is_file() else None
expected = change.get("expected_sha256")
if expected is not None and (current is None or sha256(current) != expected):
raise IsolatedExecutorError("SOURCE_DRIFT:" + relative)
operation = change.get("operation")
if operation == "write_file":
content = change.get("content")
if not isinstance(content, str):
raise IsolatedExecutorError("WRITE_CONTENT_INVALID")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
elif operation == "replace_text":
if current is None:
raise IsolatedExecutorError("REPLACE_SOURCE_MISSING")
old = str(change.get("old"))
new = str(change.get("new"))
count = int(change.get("count", 1))
text = current.decode("utf-8")
if text.count(old) != count:
raise IsolatedExecutorError("REPLACE_COUNT_INVALID:" + relative)
path.write_text(text.replace(old, new), encoding="utf-8")
elif operation == "delete_file":
if path.exists():
path.unlink()
else:
raise IsolatedExecutorError("CHANGE_OPERATION_INVALID")
changed.append(relative)
return sorted(set(changed))
def changed_bundle(root: Path, changed: list[str]) -> list[dict[str, Any]]:
files = []
for relative in changed:
path = resolve_under(root, relative)
if path.exists():
data = path.read_bytes()
if len(data) > MAX_FILE_BYTES:
raise IsolatedExecutorError("CHANGED_FILE_TOO_LARGE:" + relative)
files.append({
"path": relative,
"deleted": False,
"sha256": sha256(data),
"content_b64": base64.b64encode(data).decode("ascii"),
})
else:
files.append({"path": relative, "deleted": True, "sha256": None, "content_b64": None})
return files
def execute_request(request_path: Path, result_path: Path) -> None:
assert_uncredentialed()
request = json.loads(request_path.read_text(encoding="utf-8"))
if request.get("schema") != "aios.isolated-execution.request.v1":
raise IsolatedExecutorError("REQUEST_SCHEMA_INVALID")
request_id = request.get("request_id")
if not isinstance(request_id, str) or not REQUEST_ID_RE.fullmatch(request_id):
raise IsolatedExecutorError("REQUEST_ID_INVALID")
source_archive = Path(str(request.get("source_archive") or ""))
if not source_archive.is_absolute():
raise IsolatedExecutorError("SOURCE_ARCHIVE_PATH_INVALID")
expected_archive_sha = str(request.get("source_archive_sha256") or "")
if sha256(source_archive.read_bytes()) != expected_archive_sha:
raise IsolatedExecutorError("SOURCE_ARCHIVE_SHA_MISMATCH")
with tempfile.TemporaryDirectory(prefix="aios-isolated-executor-") as temporary:
root = Path(temporary) / "workspace"
root.mkdir()
extract_source(source_archive, root)
changed = apply_changes(request, root)
commands = [
run_command(item, root)
for item in request.get("verification", [])
if isinstance(item, Mapping)
]
result = {
"schema": "aios.isolated-execution.result.v1",
"request_id": request_id,
"request_sha256": sha256(canonical(request)),
"changed_files": changed_bundle(root, changed),
"verification": commands,
"credential_isolation": {
"credential_env_absent": True,
"credential_mount_inaccessible": True,
"governed_bridge_unreachable": True,
},
}
result["result_sha256"] = sha256(canonical(result))
temporary_result = result_path.with_suffix(".tmp")
temporary_result.write_text(json.dumps(result, sort_keys=True, separators=(",", ":")), encoding="utf-8")
os.replace(temporary_result, result_path)
def serve(root: Path) -> None:
assert_uncredentialed()
requests = root / "requests"
results = root / "results"
failed = root / "failed"
for path in (requests, results, failed):
path.mkdir(parents=True, exist_ok=True)
while True:
for request in sorted(requests.glob("*.json")):
if request.is_symlink():
continue
result = results / request.name
try:
execute_request(request, result)
request.unlink()
except Exception as exc:
(failed / request.name).write_text(
json.dumps({"error": str(exc) or type(exc).__name__}, sort_keys=True),
encoding="utf-8",
)
request.unlink(missing_ok=True)
time.sleep(0.5)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--root", default="/var/lib/aios-online-agent-executor")
parser.add_argument("--request")
parser.add_argument("--result")
args = parser.parse_args()
if args.request and args.result:
execute_request(Path(args.request), Path(args.result))
else:
serve(Path(args.root))
active-runtime/app/aios_online_agent_execution.py
+25 -3
View File
@@ -150,6 +150,28 @@ def _digest(value: object) -> str:
return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
def _source_signals(value: object) -> tuple[str, ...]:
"""Normalize untrusted natural-language hints before they enter a signed envelope."""
if value is None:
return ()
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
_error("SOURCE_CAPTURE_SIGNALS_INVALID")
result: list[str] = []
for item in value:
if not isinstance(item, str):
_error("SOURCE_CAPTURE_SIGNALS_INVALID")
text = " ".join(item.split()).strip()
if not text:
continue
# Projection may accept long task prose, but execution envelopes stay bounded.
text = text[:512].rstrip()
if text not in result:
result.append(text)
if len(result) == 16:
break
return tuple(result)
def _capabilities(value: object) -> tuple[str, ...]:
if value is None:
return ()
@@ -263,8 +285,8 @@ def validate_execution_envelope(
if action == "source_capture":
if changes or verification:
_error("SOURCE_CAPTURE_MUTATION_DENY")
signals = envelope.get("source_signals") or []
if not isinstance(signals, list) or len(signals) > 16 or not all(isinstance(item, str) and len(item) <= 512 for item in signals):
signals = list(_source_signals(envelope.get("source_signals") or []))
if envelope.get("source_signals") != signals:
_error("SOURCE_CAPTURE_SIGNALS_INVALID")
elif not changes:
_error("CHANGE_SET_INVALID")
@@ -311,7 +333,7 @@ def project_execution_envelope(
"capability_scope": list(capabilities),
"change_set": list(request.get("change_set") or []),
"verification": list(request.get("verification") or []),
"source_signals": list(request.get("source_signals") or []) if action == "source_capture" else [],
"source_signals": list(_source_signals(request.get("source_signals") or [])) if action == "source_capture" else [],
"governance": governance,
"workspace": {
"authority_git_dir": "/data/aios-gitops/repo.git",
active-runtime/app/aios_online_agent_plan_store.py
+187
View File
@@ -0,0 +1,187 @@
"""Durable, authority-bound deterministic execution plans for Online Agent Tasks."""
from __future__ import annotations
from contextlib import contextmanager
from dataclasses import dataclass
import fcntl
import hashlib
import json
from pathlib import Path
from typing import Any, Mapping
from aios_online_agent_execution import (
GovernedExecutionBinding,
project_execution_envelope,
)
class ExecutionPlanError(RuntimeError):
pass
def _canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def _digest(value: object) -> str:
return hashlib.sha256(_canonical(value)).hexdigest()
def _task_key(task_id: str) -> str:
readable = "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in task_id).strip("-")[:72] or "task"
return readable + "-" + hashlib.sha256(task_id.encode("utf-8")).hexdigest()[:24]
@dataclass(frozen=True)
class ExecutionPlanRecord:
task_id: str
job_id: str
source_head_oid: str
generation: int
idempotency_key: str
plan_digest: str
execution: dict[str, Any]
def public(self) -> dict[str, Any]:
return {
"schema": "aios.online-agent.plan-record.v1",
"task_id": self.task_id,
"job_id": self.job_id,
"source_head_oid": self.source_head_oid,
"generation": self.generation,
"idempotency_key": self.idempotency_key,
"plan_digest": self.plan_digest,
"execution": dict(self.execution),
}
class ExecutionPlanStore:
def __init__(self, root: Path) -> None:
self.root = Path(root)
self.root.mkdir(parents=True, exist_ok=True)
def _path(self, task_id: str) -> Path:
return self.root / (_task_key(task_id) + ".json")
@contextmanager
def _lock(self, task_id: str):
path = self.root / (_task_key(task_id) + ".lock")
with path.open("a+b") as handle:
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
@staticmethod
def _source_head(binding: GovernedExecutionBinding) -> str:
return binding.expected_branch_oid if binding.branch_mode == "CONTINUE_EXISTING" else binding.base_oid
def _read(self, task_id: str) -> ExecutionPlanRecord | None:
path = self._path(task_id)
if not path.is_file() or path.is_symlink():
return None
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, Mapping) or value.get("schema") != "aios.online-agent.plan-record.v1":
raise ExecutionPlanError("EXECUTION_PLAN_RECORD_INVALID")
execution = value.get("execution")
if not isinstance(execution, Mapping):
raise ExecutionPlanError("EXECUTION_PLAN_RECORD_INVALID")
return ExecutionPlanRecord(
task_id=str(value.get("task_id") or ""),
job_id=str(value.get("job_id") or ""),
source_head_oid=str(value.get("source_head_oid") or ""),
generation=int(value.get("generation") or 0),
idempotency_key=str(value.get("idempotency_key") or ""),
plan_digest=str(value.get("plan_digest") or ""),
execution=dict(execution),
)
def register(
self,
binding: GovernedExecutionBinding,
request: Mapping[str, Any],
) -> dict[str, Any]:
with self._lock(binding.task_id):
return self._register_locked(binding, request)
def _register_locked(
self,
binding: GovernedExecutionBinding,
request: Mapping[str, Any],
) -> dict[str, Any]:
if request.get("schema") != "aios.online-agent.plan.v1":
raise ExecutionPlanError("EXECUTION_PLAN_SCHEMA_INVALID")
if request.get("task_id") != binding.task_id or request.get("job_id") != binding.job_id:
raise ExecutionPlanError("EXECUTION_PLAN_BINDING_CONFLICT")
if int(request.get("generation") or 0) != binding.generation:
raise ExecutionPlanError("EXECUTION_PLAN_GENERATION_CONFLICT")
source_head = str(request.get("source_head_oid") or "")
if source_head != self._source_head(binding):
raise ExecutionPlanError("EXECUTION_PLAN_SOURCE_HEAD_CONFLICT")
key = str(request.get("idempotency_key") or "")
if len(key) < 8:
raise ExecutionPlanError("EXECUTION_PLAN_IDEMPOTENCY_INVALID")
execution_request = request.get("execution")
if not isinstance(execution_request, Mapping) or execution_request.get("action") != "deterministic":
raise ExecutionPlanError("EXECUTION_PLAN_DETERMINISTIC_REQUIRED")
envelope = project_execution_envelope(binding, execution_request)
material = {
"task_id": binding.task_id,
"job_id": binding.job_id,
"source_head_oid": source_head,
"generation": binding.generation,
"idempotency_key": key,
"execution": dict(execution_request),
}
plan_digest = _digest(material)
existing = self._read(binding.task_id)
expected_previous = request.get("expected_previous_plan_digest")
if existing is not None:
if existing.idempotency_key == key:
if existing.plan_digest != plan_digest:
raise ExecutionPlanError("EXECUTION_PLAN_IDEMPOTENCY_CONFLICT")
return {**existing.public(), "idempotent": True, "execution_digest": envelope["execution_digest"]}
if expected_previous != existing.plan_digest:
raise ExecutionPlanError("EXECUTION_PLAN_REPLACEMENT_CAS_MISMATCH")
if existing.source_head_oid != source_head:
raise ExecutionPlanError("EXECUTION_PLAN_STALE_REPLACEMENT_DENY")
elif expected_previous not in (None, ""):
raise ExecutionPlanError("EXECUTION_PLAN_REPLACEMENT_CAS_MISMATCH")
record = ExecutionPlanRecord(
task_id=binding.task_id,
job_id=binding.job_id,
source_head_oid=source_head,
generation=binding.generation,
idempotency_key=key,
plan_digest=plan_digest,
execution=dict(execution_request),
)
path = self._path(binding.task_id)
temporary = path.with_suffix(".tmp")
temporary.write_text(
json.dumps(record.public(), sort_keys=True, separators=(",", ":")),
encoding="utf-8",
)
temporary.replace(path)
return {**record.public(), "idempotent": False, "execution_digest": envelope["execution_digest"]}
def resolve(self, binding: GovernedExecutionBinding) -> dict[str, Any] | None:
record = self._read(binding.task_id)
if record is None:
return None
if (
record.task_id != binding.task_id
or record.job_id != binding.job_id
or record.generation != binding.generation
):
raise ExecutionPlanError("EXECUTION_PLAN_BINDING_CONFLICT")
if record.source_head_oid != self._source_head(binding):
raise ExecutionPlanError("EXECUTION_PLAN_SOURCE_HEAD_DRIFT")
envelope = project_execution_envelope(binding, record.execution)
return {
"execution": dict(record.execution),
"plan_digest": record.plan_digest,
"execution_digest": envelope["execution_digest"],
}
active-runtime/app/aios_online_agent_worker.py
+299 -23
View File
@@ -18,6 +18,7 @@ from dataclasses import dataclass
from datetime import datetime, timezone
from abc import ABC, abstractmethod
import hashlib
import secrets
import io
import json
import os
@@ -31,7 +32,7 @@ import threading
import time
import urllib.error
import urllib.request
from typing import Any, Mapping
from typing import Any, Mapping, Sequence
from urllib.parse import unquote
from aios_online_agent_execution import (
@@ -74,9 +75,20 @@ WORKER_CAPABILITIES = tuple(
)
MAX_TASK_BYTES = 512 * 1024
MAX_FILE_BYTES = 2 * 1024 * 1024
ISOLATED_EXECUTOR_ROOT = Path(
os.environ.get("AIOS_ISOLATED_EXECUTOR_ROOT", "/var/lib/aios-online-agent-executor")
)
ISOLATED_EXECUTOR_TIMEOUT_SECONDS = int(
os.environ.get("AIOS_ISOLATED_EXECUTOR_TIMEOUT_SECONDS", "1800")
)
ISOLATED_EXECUTOR_TIMEOUT_HARD_MAX_SECONDS = 7200
GIT_VISIBILITY_TIMEOUT_SECONDS = int(
os.environ.get("AIOS_GIT_VISIBILITY_TIMEOUT_SECONDS", "45")
)
OID_RE = re.compile(r"^[0-9a-f]{40}$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,254}$")
BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
PATH_RE = re.compile(r"^[^\\\x00-\x1f\x7f]+$")
SENSITIVE_RE = re.compile(
@@ -92,6 +104,33 @@ class WorkerError(RuntimeError):
self.details = dict(details or {})
def _isolated_executor_wait_seconds(verification: Sequence[Mapping[str, Any]]) -> int:
verification_budget = sum(int(item["timeout_seconds"]) for item in verification)
required_wait_seconds = max(30, verification_budget + 30)
if not 30 <= ISOLATED_EXECUTOR_TIMEOUT_SECONDS <= ISOLATED_EXECUTOR_TIMEOUT_HARD_MAX_SECONDS:
raise WorkerError("ISOLATED_EXECUTOR_TIMEOUT_CONFIG_INVALID", phase="adapter")
if required_wait_seconds > ISOLATED_EXECUTOR_TIMEOUT_SECONDS:
raise WorkerError(
"ISOLATED_EXECUTOR_TIMEOUT_BUDGET_EXCEEDED",
phase="adapter",
details={
"required_seconds": required_wait_seconds,
"configured_seconds": ISOLATED_EXECUTOR_TIMEOUT_SECONDS,
},
)
return required_wait_seconds
def _claim_order(item: tuple[Mapping[str, Any], "ExecutionEnvelope"]) -> tuple[int, int, str, str]:
task, envelope = item
try:
priority = int(task.get("priority") or 0)
except (TypeError, ValueError):
priority = 0
action_rank = 0 if envelope.action == "deterministic" else 1
return (-priority, action_rank, str(task.get("created_at") or ""), envelope.task_id)
def now() -> str:
return datetime.now(timezone.utc).isoformat()
@@ -621,7 +660,7 @@ class HermesExecutor(RuntimeAdapter):
def adapter_for(name: str) -> RuntimeAdapter:
return {
"deterministic": DeterministicExecutor,
"deterministic": IsolatedDeterministicExecutor,
"source_capture": SourceCaptureExecutor,
"codex": CodexExecutor,
"hermes": HermesExecutor,
@@ -697,15 +736,13 @@ class BridgeClient:
return HttpClient.call(BRIDGE_BASE, "POST", route, self.token, body, timeout=60)
def project_execution(self, task: Mapping[str, Any], execution: Mapping[str, Any] | None = None) -> tuple[int, dict[str, Any]]:
signals = [str(task.get("goal") or ""), str(task.get("instructions") or "")]
return self.call(
"/v2/execution-envelope",
{
request: dict[str, Any] = {
"task_id": str(task.get("task_id") or ""),
"required_capabilities": list(task.get("required_capabilities") or []),
"execution": dict(execution) if execution is not None else {"action": "source_capture", "source_signals": signals},
},
)
}
if execution is not None:
request["execution"] = dict(execution)
return self.call("/v2/execution-envelope", request)
class LeaseHeartbeat:
@@ -792,7 +829,15 @@ def task_state_path(task_id: str) -> Path:
return STATE_ROOT / "tasks" / f"{task_id}.json"
def load_task_state(task_id: str, execution_digest: str) -> dict[str, Any] | None:
def load_task_state(
task_id: str,
execution_digest: str,
*,
attempt: int | None = None,
continuation_head_oid: str | None = None,
continuation_job_id: str | None = None,
commit_idempotency_key: str | None = None,
) -> dict[str, Any] | None:
path = task_state_path(task_id)
if not path.is_file() or path.is_symlink():
return None
@@ -803,7 +848,54 @@ def load_task_state(task_id: str, execution_digest: str) -> dict[str, Any] | Non
if not isinstance(value, dict):
raise WorkerError("TASK_STATE_INVALID", phase="state")
if value.get("execution_digest") not in {None, execution_digest}:
if value.get("phase") == "source_capture_done" and not value.get("governed"):
if value.get("phase") in {"claimed", "workspace_prepared", "source_capture_done"} and not value.get("governed"):
return None
if (
value.get("phase") == "verification_passed"
and not value.get("governed")
and attempt is not None
and int(value.get("attempt") or 0) < int(attempt)
):
# The prior official attempt completed deterministic validation but failed before
# a WriterLease or governed commit. A later CAS-authorized retry may use its new
# digest; the same attempt can never erase or replace this state.
return None
if value.get("phase") == "reconciliation_required" and not value.get("governed"):
stored_attempt = value.get("attempt")
if (
stored_attempt is not None
and attempt is not None
and int(stored_attempt) < int(attempt)
and value.get("job_id") == continuation_job_id
and isinstance(continuation_head_oid, str)
and OID_RE.fullmatch(continuation_head_oid)
):
return None
details = value.get("details")
if (
stored_attempt is None
and attempt is not None
and isinstance(details, Mapping)
and details.get("api_status") == 503
and details.get("error") == "WORKTREE_DIRTY_SCOPE_MISMATCH"
and value.get("job_id") == continuation_job_id
and isinstance(continuation_head_oid, str)
and OID_RE.fullmatch(continuation_head_oid)
):
# Compatibility for the pre-fix recovery record written by attempt 18,
# which omitted its attempt number. This exact pre-commit failure has no
# governed side effect; plan-registry CAS and the continuation head still
# bind the later official claim before the stale state can be discarded.
return None
governed = value.get("governed")
if (
continuation_head_oid
and isinstance(governed, Mapping)
and governed.get("commit_oid") == continuation_head_oid
and governed.get("commit_idempotency_key") != commit_idempotency_key
):
# A previous governed commit is the exact continuation head; a new digest/key is a
# legitimate next deterministic plan on the same Task/Job lineage.
return None
raise WorkerError("TASK_PAYLOAD_DRIFT", phase="recovery")
return value
@@ -829,15 +921,180 @@ def cleanup_run_root(path: Path) -> None:
shutil.rmtree(candidate)
def prepare_worktree(run_root: Path, base_oid: str) -> Path:
def prepare_worktree(run_root: Path, envelope: ExecutionEnvelope) -> Path:
clone = run_root / "repository"
worktree = run_root / "worktree"
worktree_oid = (
envelope.expected_branch_oid
if envelope.branch_mode == "CONTINUE_EXISTING"
else envelope.base_oid
)
run_command("workspace_clone", ["/usr/bin/git", "clone", "--no-local", "--no-hardlinks", str(AUTHORITY_GIT_DIR), str(clone)], cwd=run_root, timeout=300, internal_git=True)
run_command("workspace_base_probe", ["/usr/bin/git", "-C", str(clone), "rev-parse", "--verify", base_oid + "^{commit}"], cwd=run_root, timeout=60)
run_command("workspace_worktree_create", ["/usr/bin/git", "-C", str(clone), "worktree", "add", "--detach", str(worktree), base_oid], cwd=run_root, timeout=300, internal_git=True)
run_command("workspace_base_probe", ["/usr/bin/git", "-C", str(clone), "rev-parse", "--verify", worktree_oid + "^{commit}"], cwd=run_root, timeout=60)
run_command("workspace_worktree_create", ["/usr/bin/git", "-C", str(clone), "worktree", "add", "--detach", str(worktree), worktree_oid], cwd=run_root, timeout=300, internal_git=True)
return worktree
def _archive_worktree(worktree: Path, archive_path: Path) -> str:
with tarfile.open(archive_path, mode="w:gz") as archive:
for path in sorted(worktree.rglob("*")):
relative = path.relative_to(worktree)
if ".git" in relative.parts or path.is_symlink() or not path.is_file():
continue
data = path.read_bytes()
info = tarfile.TarInfo(relative.as_posix())
info.size = len(data)
info.mode = 0o640
info.mtime = 0
archive.addfile(info, io.BytesIO(data))
archive_path.chmod(0o640)
return digest_bytes(archive_path.read_bytes())
class IsolatedDeterministicExecutor(RuntimeAdapter):
"""Coordinator-side client for the uncredentialed executor exchange."""
name = "isolated-deterministic"
@staticmethod
def _atomic(path: Path, data: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
try:
with os.fdopen(fd, "wb") as handle:
handle.write(data)
handle.flush()
os.fsync(handle.fileno())
os.chmod(temporary, 0o640)
os.replace(temporary, path)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
if envelope.action != "deterministic":
raise WorkerError("ISOLATED_EXECUTOR_ACTION_INVALID", phase="adapter")
root = ISOLATED_EXECUTOR_ROOT
for child in ("requests", "results", "failed", "sources"):
candidate = root / child
if candidate.is_symlink() or not candidate.is_dir():
raise WorkerError("ISOLATED_EXECUTOR_UNAVAILABLE", phase="adapter", details={"path": str(candidate)})
request_id = f"{envelope.task_id}.{secrets.token_hex(16)}"
if not REQUEST_ID_RE.fullmatch(request_id):
raise WorkerError("ISOLATED_EXECUTOR_REQUEST_ID_INVALID", phase="adapter")
archive = root / "sources" / f"{request_id}.tgz"
request_path = root / "requests" / f"{request_id}.json"
result_path = root / "results" / f"{request_id}.json"
failed_path = root / "failed" / f"{request_id}.json"
for stale in (request_path, result_path, failed_path, archive):
if stale.exists() or stale.is_symlink():
raise WorkerError(
"ISOLATED_EXECUTOR_COLLISION_DETECTED", phase="adapter", details={"request_id": request_id, "path": str(stale)}
)
archive_sha = _archive_worktree(worktree, archive)
request = {
"schema": "aios.isolated-execution.request.v1",
"request_id": request_id,
"task_id": envelope.task_id,
"job_id": envelope.job_id,
"execution_digest": envelope.digest,
"source_archive": str(archive),
"source_archive_sha256": archive_sha,
"change_set": list(envelope.change_set),
"verification": list(envelope.verification),
}
request_sha = digest_json(request)
self._atomic(request_path, json.dumps(request, sort_keys=True, separators=(",", ":")).encode("utf-8"))
wait_seconds = _isolated_executor_wait_seconds(envelope.verification)
deadline = time.monotonic() + wait_seconds
try:
while time.monotonic() < deadline:
if failed_path.is_file():
details = json.loads(failed_path.read_text(encoding="utf-8"))
raise WorkerError("ISOLATED_EXECUTOR_FAILED", phase="adapter", details={"error": details.get("error")})
if result_path.is_file():
result = json.loads(result_path.read_text(encoding="utf-8"))
if result.get("request_id") != request_id:
raise WorkerError("ISOLATED_EXECUTOR_REQUEST_ID_MISMATCH", phase="adapter")
if result.get("schema") != "aios.isolated-execution.result.v1":
raise WorkerError("ISOLATED_EXECUTOR_RESULT_INVALID", phase="adapter")
if result.get("request_sha256") != request_sha:
raise WorkerError("ISOLATED_EXECUTOR_REQUEST_BINDING_INVALID", phase="adapter")
material = {key: value for key, value in result.items() if key != "result_sha256"}
if result.get("result_sha256") != digest_json(material):
raise WorkerError("ISOLATED_EXECUTOR_RESULT_DIGEST_INVALID", phase="adapter")
isolation = result.get("credential_isolation")
if not isinstance(isolation, Mapping) or not all(isolation.get(k) is True for k in (
"credential_env_absent", "credential_mount_inaccessible", "governed_bridge_unreachable"
)):
raise WorkerError("CREDENTIAL_ISOLATION_VERIFY", phase="adapter")
expected_paths = sorted({str(item["path"]) for item in envelope.change_set})
returned = result.get("changed_files")
if not isinstance(returned, list) or sorted(str(item.get("path")) for item in returned) != expected_paths:
raise WorkerError("ISOLATED_EXECUTOR_CHANGED_FILESET_INVALID", phase="adapter")
for item in returned:
relative = safe_relative_path(item.get("path"), field="isolated_changed_file")
path = resolve_under(worktree, relative, write=True)
if item.get("deleted") is True:
path.unlink(missing_ok=True)
continue
raw = base64.b64decode(str(item.get("content_b64") or ""), validate=True)
if digest_bytes(raw) != item.get("sha256"):
raise WorkerError("ISOLATED_EXECUTOR_FILE_DIGEST_INVALID", phase="adapter")
self._atomic(path, raw)
commands = [
run_command("git_diff_check", ["/usr/bin/git", "-C", str(worktree), "diff", "--check"], cwd=worktree, timeout=120),
run_command("git_diff_stat", ["/usr/bin/git", "-C", str(worktree), "diff", "--no-ext-diff", "--stat"], cwd=worktree, timeout=120),
]
return {
"changed_files": expected_paths,
"commands": commands,
"tests": list(result.get("verification") or []),
"adapter": self.name,
"credential_isolation": dict(isolation),
}
time.sleep(0.25)
raise WorkerError("ISOLATED_EXECUTOR_TIMEOUT", phase="adapter")
finally:
request_path.unlink(missing_ok=True)
result_path.unlink(missing_ok=True)
failed_path.unlink(missing_ok=True)
archive.unlink(missing_ok=True)
def _verify_converged_job(
response: Mapping[str, Any],
envelope: ExecutionEnvelope,
commit_oid: str,
) -> None:
if response.get("state") != "COMPLETED":
raise WorkerError("GOVERNED_RECEIPT_STATE_INVALID", phase="verification")
receipt = response.get("receipt")
if not isinstance(receipt, Mapping) or receipt.get("remote_verified") is not True:
raise WorkerError("GOVERNED_RECEIPT_REQUIRED", phase="verification")
for field in ("authority_oid", "api_branch_oid", "git_remote_oid"):
if receipt.get(field) != commit_oid:
raise WorkerError("GOVERNED_RECEIPT_OID_MISMATCH", phase="verification", details={"field": field})
leases = response.get("branch_leases")
if not isinstance(leases, list):
raise WorkerError("WRITER_LEASE_RECOVERY_EVIDENCE_MISSING", phase="verification")
active = [
item for item in leases
if isinstance(item, Mapping)
and item.get("state") == "ACTIVE"
and item.get("job_id") == envelope.job_id
and item.get("branch") == envelope.target_branch
]
if len(active) != 1:
raise WorkerError("WRITER_LEASE_RECOVERY_EVIDENCE_INVALID", phase="verification")
lease = active[0]
if (
int(lease.get("generation") or 0) != envelope.generation
or lease.get("current_head_oid") != commit_oid
):
raise WorkerError("WRITER_LEASE_CURRENT_HEAD_MISMATCH", phase="verification")
def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeClient, heartbeat: LeaseHeartbeat) -> dict[str, Any]:
governance = envelope.governance
if governance.get("task_id") != envelope.task_id or governance.get("job_id") != envelope.job_id:
@@ -904,8 +1161,13 @@ def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeC
commit_oid = commit_response.get("commit_oid")
if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
raise WorkerError("GOVERNED_COMMIT_OID_INVALID", phase="committing")
receipt = commit_response.get("receipt")
if not isinstance(receipt, Mapping) or receipt.get("remote_verified") is not True:
raise WorkerError("GOVERNED_COMMIT_RECEIPT_REQUIRED", phase="verification")
visible_oid: str | None = None
for _ in range(5):
deadline = time.monotonic() + max(5, min(GIT_VISIBILITY_TIMEOUT_SECONDS, 120))
delay = 0.5
while time.monotonic() < deadline:
probe = subprocess.run(
["/usr/bin/git", "--git-dir", str(AUTHORITY_GIT_DIR), "rev-parse", "--verify", f"refs/heads/{envelope.target_branch}"],
text=True,
@@ -918,9 +1180,14 @@ def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeC
visible_oid = probe.stdout.strip()
if visible_oid == commit_oid:
break
time.sleep(1)
time.sleep(delay)
delay = min(delay * 1.7, 5.0)
if visible_oid != commit_oid:
raise WorkerError("GIT_VISIBILITY_MISMATCH", phase="verification", details={"commit_oid": commit_oid, "visible_oid": visible_oid})
raise WorkerError("RECONCILIATION_REQUIRED", phase="verification", details={"commit_oid": commit_oid, "visible_oid": visible_oid, "job_id": envelope.job_id})
job_status, job_response = bridge.call("/v2/job", {"job_id": envelope.job_id})
if not 200 <= job_status < 300:
raise WorkerError("RECONCILIATION_REQUIRED", phase="verification", details={"api_status": job_status, "job_id": envelope.job_id})
_verify_converged_job(job_response, envelope, commit_oid)
return {
"governed_task_id": envelope.task_id,
"job_id": envelope.job_id,
@@ -1060,14 +1327,22 @@ class WorkerSupervisor:
current_envelope, _ = self.resolve_envelope(current_task)
if current_envelope.digest != envelope.digest or current_envelope.job_id != envelope.job_id:
raise WorkerError("TASK_PAYLOAD_DRIFT", phase="claim")
state = load_task_state(task_id, envelope.digest) or {"task_id": task_id, "execution_digest": envelope.digest, "phase": "claimed"}
state = load_task_state(
task_id,
envelope.digest,
attempt=int(claim.get("attempt") or 0),
continuation_head_oid=envelope.expected_branch_oid if envelope.branch_mode == "CONTINUE_EXISTING" else None,
continuation_job_id=envelope.job_id,
commit_idempotency_key=str(envelope.governance.get("commit_idempotency_key") or ""),
) or {"task_id": task_id, "execution_digest": envelope.digest, "phase": "claimed"}
governed = state.get("governed") if isinstance(state.get("governed"), Mapping) else None
if envelope.action == "deterministic" and isinstance(governed, Mapping) and isinstance(governed.get("commit_oid"), str) and OID_RE.fullmatch(governed["commit_oid"]):
if governed.get("job_id") != envelope.job_id or governed.get("target_branch") != envelope.target_branch or int(governed.get("generation") or 0) != envelope.generation or not isinstance(governed.get("receipt"), Mapping):
raise WorkerError("RECOVERY_EVIDENCE_BINDING_INVALID", phase="recovery")
job_status, job_receipt = self.bridge.call("/v2/job", {"job_id": envelope.job_id})
if not 200 <= job_status < 300 or job_receipt.get("state") != "COMPLETED":
raise WorkerError("RECOVERY_RECEIPT_UNVERIFIED", phase="recovery", details={"api_status": job_status, "state": job_receipt.get("state")})
if not 200 <= job_status < 300:
raise WorkerError("RECOVERY_RECEIPT_UNVERIFIED", phase="recovery", details={"api_status": job_status})
_verify_converged_job(job_receipt, envelope, str(governed["commit_oid"]))
probe = subprocess.run(["/usr/bin/git", "--git-dir", str(AUTHORITY_GIT_DIR), "rev-parse", "--verify", f"refs/heads/{envelope.target_branch}"], text=True, capture_output=True, check=False, env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "GIT_CONFIG_GLOBAL": str(GIT_CONFIG_GLOBAL)}, timeout=30)
if probe.returncode != 0 or probe.stdout.strip() != governed["commit_oid"]:
raise WorkerError("RECOVERY_HEAD_UNVERIFIED", phase="recovery")
@@ -1078,7 +1353,7 @@ class WorkerSupervisor:
else:
run_root = new_run_root()
try:
worktree = prepare_worktree(run_root, envelope.base_oid)
worktree = prepare_worktree(run_root, envelope)
state.update({"phase": "workspace_prepared", "execution_digest": envelope.digest, "attempt": claim.get("attempt")})
save_task_state(task_id, state)
adapter = adapter_for(envelope.adapter)
@@ -1108,7 +1383,7 @@ class WorkerSupervisor:
except WorkerError as exc:
audit("TASK_FAILED", task_id=task_id, error=exc.code, phase=exc.phase, details=exc.details)
if exc.code == "RECONCILIATION_REQUIRED":
save_task_state(task_id, {"task_id": task_id, "execution_digest": envelope.digest, "phase": "reconciliation_required", "job_id": envelope.job_id, "details": exc.details})
save_task_state(task_id, {"task_id": task_id, "execution_digest": envelope.digest, "phase": "reconciliation_required", "attempt": claim.get("attempt"), "job_id": envelope.job_id, "details": exc.details})
audit("TASK_RECONCILIATION_REQUIRED", task_id=task_id, job_id=envelope.job_id, details=exc.details)
return
failure_key = f"{task_id}-attempt-{claim.get('attempt', 0)}-failure"
@@ -1155,6 +1430,7 @@ class WorkerSupervisor:
continue
audit("TASK_VISIBLE_TO_WORKER_CLAIMABLE", task_id=task.get("task_id"), envelope_source=source, action=envelope.action)
dynamic_tasks.append((task, envelope))
dynamic_tasks.sort(key=_claim_order)
for task, envelope in dynamic_tasks:
task_id = str(task.get("task_id") or "")
status, claim = self.agent.claim(task_id)
@@ -1170,7 +1446,7 @@ class WorkerSupervisor:
def run_forever(self) -> int:
ensure_roots()
audit("WORKER_STARTED", task_agnostic=True, api_base=API_BASE, bridge_base=BRIDGE_BASE, adapter_names=["deterministic", "source_capture", "codex", "hermes"], runtime_only_credentials=True)
audit("WORKER_STARTED", task_agnostic=True, api_base=API_BASE, bridge_base=BRIDGE_BASE, adapter_names=["isolated-deterministic", "source_capture", "codex", "hermes"], runtime_only_credentials=True, isolated_executor=True)
while True:
self.once()
time.sleep(POLL_SECONDS)
deploy/systemd/aios-agent-task-retry-extension.service
+36
View File
@@ -0,0 +1,36 @@
[Unit]
Description=AIOS owner-controlled Task retry-extension API
After=aios-agent-task-api.service
Wants=aios-agent-task-api.service
[Service]
Type=simple
User=aios-agent-task
Group=aios-agent-task
LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
EnvironmentFile=/etc/aios-gitops/agent-task-retry-extension.env
Environment=AIOS_AGENT_TASK_DB=/var/lib/aios-agent-task/state.sqlite3
Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
Environment=AIOS_TASK_RETRY_HOST=127.0.0.1
Environment=AIOS_TASK_RETRY_PORT=18941
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_agent_task_retry_extension.py
Restart=on-failure
RestartSec=2
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictSUIDSGID=true
RestrictRealtime=true
RestrictNamespaces=true
MemoryDenyWriteExecute=true
RestrictAddressFamilies=AF_UNIX AF_INET
ReadWritePaths=/var/lib/aios-agent-task
UMask=0077
[Install]
WantedBy=multi-user.target
deploy/systemd/aios-online-agent-isolated-executor.service
+33
View File
@@ -0,0 +1,33 @@
[Unit]
Description=AIOS uncredentialed deterministic Planner/Executor
After=local-fs.target
[Service]
Type=simple
User=aios-engineering-executor
Group=aios-executor-exchange
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_isolated_executor.py --root /var/lib/aios-online-agent-executor
Restart=always
RestartSec=2
NoNewPrivileges=true
PrivateTmp=true
PrivateNetwork=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictSUIDSGID=true
RestrictRealtime=true
RestrictNamespaces=true
MemoryDenyWriteExecute=true
RestrictAddressFamilies=AF_UNIX AF_INET
IPAddressDeny=any
IPAddressAllow=localhost
InaccessiblePaths=/run/credentials
ReadWritePaths=/var/lib/aios-online-agent-executor
UMask=0027
[Install]
WantedBy=multi-user.target
deploy/systemd/aios-online-agent-worker.service
+2 -1
View File
@@ -7,7 +7,7 @@ Wants=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
Type=simple
User=aios-engineering-worker
Group=aios-engineering-worker
SupplementaryGroups=gitops
SupplementaryGroups=gitops aios-executor-exchange
LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
Environment=PYTHONUNBUFFERED=1
Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
@@ -17,6 +17,7 @@ Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
Environment=AIOS_AGENT_VERSION=2026-08-17.1-dynamic
Environment=AIOS_WORKER_CAPABILITIES=git.read,git.test,governed.write.branch,governed.write.file
Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
Environment=AIOS_ISOLATED_EXECUTOR_ROOT=/var/lib/aios-online-agent-executor
Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-worker
Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-agent-worker/workspaces
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_online_agent_worker.py
deploy/systemd/aios-online-engineering-consumer.service
+2 -1
View File
@@ -7,7 +7,7 @@ Wants=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
Type=simple
User=aios-engineering-worker
Group=aios-engineering-worker
SupplementaryGroups=gitops
SupplementaryGroups=gitops aios-executor-exchange
LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-online-engineering-20260815/agent_token
Environment=PYTHONUNBUFFERED=1
Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
@@ -17,6 +17,7 @@ Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
Environment=AIOS_AGENT_VERSION=2026-08-17.1-dynamic
Environment=AIOS_WORKER_CAPABILITIES=git.read,git.test,governed.write.branch,governed.write.file
Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
Environment=AIOS_ISOLATED_EXECUTOR_ROOT=/var/lib/aios-online-agent-executor
Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-engineering-consumer
Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-engineering-consumer/workspaces
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_online_agent_worker.py
deploy/sysusers.d/aios-online-agent-executor.conf
+4
View File
@@ -0,0 +1,4 @@
g aios-executor-exchange -
u aios-engineering-executor - "AIOS isolated engineering executor" /var/lib/aios-online-agent-executor /usr/sbin/nologin
m aios-engineering-executor aios-executor-exchange
m aios-engineering-worker aios-executor-exchange
deploy/tmpfiles.d/aios-online-agent-executor.conf
+5
View File
@@ -0,0 +1,5 @@
d /var/lib/aios-online-agent-executor 2770 aios-engineering-executor aios-executor-exchange -
d /var/lib/aios-online-agent-executor/requests 2770 aios-engineering-worker aios-executor-exchange -
d /var/lib/aios-online-agent-executor/results 2770 aios-engineering-executor aios-executor-exchange -
d /var/lib/aios-online-agent-executor/failed 2770 aios-engineering-executor aios-executor-exchange -
d /var/lib/aios-online-agent-executor/sources 2770 aios-engineering-worker aios-executor-exchange -
docs/superpowers/specs/2026-08-17-p0-claim-scheme-a-followup.md
+56
View File
@@ -0,0 +1,56 @@
# P0-CLAIM Scheme A follow-up
This follow-up continues the existing candidate `71523aef6e48b1e2fa516c456ca189b7a039104b`.
It does not create a replacement Task, Job, branch or generation.
## Retry extension
A versioned Owner-controlled loopback API provides
`POST /v1/tasks/{task_id}/retry-extension`.
The request binds FAILED status, revision, exhausted attempt/max_attempts, next max attempts,
previous verification SHA256, Owner authorization ID and idempotency key. The service performs one
SQLite `BEGIN IMMEDIATE` transaction containing the CAS update, durable retry-extension record and
Task event. Each authorization extends the boundary by exactly one attempt.
## Same-Job continuation
The durable GovernedTask binding remains the lineage anchor. After the first CREATE_NEW commit,
`execution_binding()` may normalize that SAME Job to `CONTINUE_EXISTING` only when exactly one
ACTIVE WriterLease for the Job owns the current authority branch head, generation and lineage
anchor. The lineage `base_oid` stays unchanged; only `expected_branch_oid` becomes the current head.
GovernedTask creation is also serialized by a branch-digest lock, preventing sibling continuation
tasks from racing generation admission.
## Projection and recovery
Natural-language source signals are normalized server-side to at most 16 unique 512-byte values.
Malformed persisted envelopes still fail closed.
Fresh commits and recovery require a COMPLETED Job, `remote_verified=true`, exact
authority/API/remote receipt OIDs, and exactly one ACTIVE WriterLease whose durable current head
matches the commit. Delayed local ref visibility becomes `RECONCILIATION_REQUIRED`, not a terminal
canonical failure.
## Credential isolation
The credentialed Coordinator performs only Task/Bridge coordination and governed commit. All
deterministic repository modification and test commands run in
`aios-online-agent-isolated-executor` under a separate system user with:
- no Agent/Owner credential projection;
- no gitops supplementary group;
- `PrivateNetwork=true`;
- `RestrictAddressFamilies=AF_UNIX AF_INET`, with `IPAddressDeny=any` and
`IPAddressAllow=localhost`, so regression tests may use an isolated loopback
HTTP server without gaining host or external network reachability;
- `InaccessiblePaths=/run/credentials`.
The executor receives a hash-bound source archive and deterministic change set through a filesystem
exchange and returns only changed file bytes/hashes plus command evidence. The Coordinator validates
that response before materializing exact allowed files and calling the governed Bridge.
Codex/Hermes mutation adapters remain disabled until a separately audited uncredentialed planner
backend exists. Under the current operating model, implementation code is authored before Online
Agent execution and is carried as a deterministic plan.
tests/test_agent_task_retry_extension.py
+117
View File
@@ -0,0 +1,117 @@
from pathlib import Path
import hashlib, json, sqlite3, sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
from aios_agent_task_retry_extension import RetryExtensionRequest, RetryExtensionStore, RetryExtensionError
AUTHORIZATION_ID="owner-scheme-a-20260817"
@pytest.fixture(autouse=True)
def _authorization_id(monkeypatch):
monkeypatch.setenv("AIOS_AGENT_TASK_AUTHORIZATION_ID", AUTHORIZATION_ID)
def canon(v):
return hashlib.sha256(json.dumps(v,sort_keys=True,separators=(",",":")).encode()).hexdigest()
def db(tmp_path):
p=tmp_path/"state.sqlite3"
c=sqlite3.connect(p)
c.executescript("""
CREATE TABLE tasks(
task_id TEXT PRIMARY KEY,status TEXT,attempt INTEGER,max_attempts INTEGER,revision INTEGER,
claimed_by TEXT,lease_token_hash TEXT,lease_expires_at TEXT,verification_json TEXT,result_json TEXT
);
CREATE TABLE task_events(
event_id INTEGER PRIMARY KEY AUTOINCREMENT,task_id TEXT,event_type TEXT,agent_id TEXT,
payload_json TEXT,created_at TEXT
);
""")
verification={"decision":"NEEDS_REWORK","reason":"BASE_OID_BINDING_INVALID"}
c.execute("INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?)",
("task-a","FAILED",10,10,41,None,None,None,json.dumps(verification),json.dumps({"status":"FAILED"})))
c.commit(); c.close()
return p,verification
def request(v):
return RetryExtensionRequest.from_mapping({
"expected_status":"FAILED","expected_revision":41,"expected_attempt":10,"expected_max_attempts":10,
"next_max_attempts":11,"previous_verification_sha256":canon(v),
"authorization_id":AUTHORIZATION_ID,"idempotency_key":"retry-task-a-attempt11",
})
def test_retry_extension_cas_and_audit(tmp_path):
p,v=db(tmp_path)
out=RetryExtensionStore(p).extend("task-a",request(v))
assert out["status"]=="QUEUED" and out["max_attempts"]==11 and out["revision"]==42
c=sqlite3.connect(p)
assert c.execute("select status,attempt,max_attempts,revision from tasks").fetchone()==("QUEUED",10,11,42)
event=c.execute("select event_type,payload_json from task_events").fetchone()
assert event[0]=="TASK_RETRY_EXTENSION_AUTHORIZED"
assert json.loads(event[1])["authorization_id"]=="owner-scheme-a-20260817"
def test_retry_extension_idempotent_replay(tmp_path):
p,v=db(tmp_path); store=RetryExtensionStore(p); req=request(v)
first=store.extend("task-a",req); second=store.extend("task-a",req)
assert first["idempotent"] is False and second["idempotent"] is True
c=sqlite3.connect(p)
assert c.execute("select count(*) from task_retry_extensions").fetchone()[0]==1
def test_retry_extension_normalizes_stale_terminal_claim_without_active_lease(tmp_path):
p,v=db(tmp_path)
c=sqlite3.connect(p)
c.execute("update tasks set claimed_by='previous-worker'")
c.commit(); c.close()
out=RetryExtensionStore(p).extend("task-a",request(v))
assert out["status"]=="QUEUED"
c=sqlite3.connect(p)
assert c.execute(
"select claimed_by,lease_token_hash,lease_expires_at from tasks"
).fetchone()==(None,None,None)
payload=json.loads(c.execute("select payload_json from task_events").fetchone()[0])
assert payload["previous_claimed_by"]=="previous-worker"
assert payload["stale_terminal_claim_normalized"] is True
assert c.execute("select count(*) from task_events").fetchone()[0]==1
def test_retry_extension_verification_drift_denied(tmp_path):
p,v=db(tmp_path)
bad=request({"decision":"NEEDS_REWORK","reason":"other"})
with pytest.raises(RetryExtensionError,match="RETRY_EXTENSION_VERIFICATION_DRIFT"):
RetryExtensionStore(p).extend("task-a",bad)
def test_retry_extension_authorization_id_mismatch_rejected(tmp_path, monkeypatch):
p,v=db(tmp_path)
monkeypatch.setenv("AIOS_AGENT_TASK_AUTHORIZATION_ID", "other-owner-id")
with pytest.raises(RetryExtensionError,match="RETRY_EXTENSION_AUTHORIZATION_ID_MISMATCH"):
RetryExtensionStore(p).extend("task-a",request(v))
def test_retry_extension_bounded_one_step():
v={"decision":"NEEDS_REWORK"}
data={
"expected_status":"FAILED","expected_revision":41,"expected_attempt":10,"expected_max_attempts":10,
"next_max_attempts":12,"previous_verification_sha256":canon(v),
"authorization_id":"owner-scheme-a-20260817","idempotency_key":"retry-task-a-attempt11",
}
with pytest.raises(RetryExtensionError,match="RETRY_EXTENSION_MUST_BE_BOUNDED_ONE_STEP"):
RetryExtensionRequest.from_mapping(data)
def test_retry_extension_unit_is_unprivileged_and_requires_owner_binding():
unit=(ROOT/"deploy/systemd/aios-agent-task-retry-extension.service").read_text()
assert "User=aios-agent-task" in unit
assert "Group=aios-agent-task" in unit
assert "LoadCredential=agent_task_control_token:" in unit
assert "EnvironmentFile=/etc/aios-gitops/agent-task-retry-extension.env" in unit
assert "User=root" not in unit
tests/test_bridge_v2_governed_http_bootstrap.py
+1 -1
View File
@@ -77,7 +77,7 @@ def test_slash_branch_uses_paginated_collection():
def test_http_route_contract_is_complete():
expected = {
"/v2/runtime", "/v2/capabilities", "/v2/governed-task", "/v2/execution-envelope",
"/v2/runtime", "/v2/capabilities", "/v2/governed-task", "/v2/execution-envelope", "/v2/execution-plan",
"/v2/writer-lease", "/v2/writer-lease/bootstrap", "/v2/writer-lease/renew", "/v2/commit-file", "/v2/commit-files",
"/v2/job", "/v2/job/resume", "/v2/job/reconcile",
"/v2/tag/approval-request",
tests/test_isolated_executor.py
+67
View File
@@ -0,0 +1,67 @@
from pathlib import Path
import hashlib,json,os,tarfile,sys,tempfile
import pytest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
import aios_isolated_executor as mod
def make_archive(tmp_path):
root=tmp_path/"src"; root.mkdir()
(root/"hello.py").write_text("VALUE = 1\n")
archive=tmp_path/"source.tgz"
with tarfile.open(archive,"w:gz") as tf: tf.add(root/"hello.py",arcname="hello.py")
return archive
def test_isolated_executor_applies_hash_bound_change(tmp_path,monkeypatch):
archive=make_archive(tmp_path)
monkeypatch.delenv("CREDENTIALS_DIRECTORY",raising=False)
monkeypatch.setattr(mod,"assert_uncredentialed",lambda:None)
request={
"schema":"aios.isolated-execution.request.v1",
"request_id":"task-req-0000000000000000",
"source_archive":str(archive),
"source_archive_sha256":mod.sha256(archive.read_bytes()),
"change_set":[{
"path":"hello.py","operation":"replace_text","old":"VALUE = 1","new":"VALUE = 2","count":1,
"expected_sha256":mod.sha256(b"VALUE = 1\n")
}],
"verification":[{"label":"compile","argv":["/usr/bin/python3","-m","py_compile","hello.py"],"cwd":".","timeout_seconds":30}]
}
req=tmp_path/"req.json"; res=tmp_path/"res.json"
req.write_text(json.dumps(request))
mod.execute_request(req,res)
out=json.loads(res.read_text())
assert out["request_id"]=="task-req-0000000000000000"
assert out["credential_isolation"]["credential_env_absent"] is True
changed=out["changed_files"][0]
assert changed["path"]=="hello.py" and changed["deleted"] is False
def test_isolated_executor_source_drift_denied(tmp_path,monkeypatch):
archive=make_archive(tmp_path); monkeypatch.setattr(mod,"assert_uncredentialed",lambda:None)
request={
"schema":"aios.isolated-execution.request.v1","request_id":"task-req-0000000000000001",
"source_archive":str(archive),
"source_archive_sha256":mod.sha256(archive.read_bytes()),
"change_set":[{"path":"hello.py","operation":"write_file","content":"x\n","expected_sha256":"0"*64}],
"verification":[]
}
req=tmp_path/"req.json"; res=tmp_path/"res.json"; req.write_text(json.dumps(request))
with pytest.raises(mod.IsolatedExecutorError,match="SOURCE_DRIFT"):
mod.execute_request(req,res)
def test_credential_boundary_eacces_is_accepted(monkeypatch):
class DeniedDirectory:
def exists(self): return True
def is_symlink(self): return False
def is_dir(self): return True
def iterdir(self): raise PermissionError(13,"permission denied")
monkeypatch.setattr(mod.os,"access",lambda *_args,**_kwargs:False)
assert mod._credential_exposure(DeniedDirectory()) is None
def test_systemd_unit_has_no_credentials_and_private_network():
unit=(ROOT/"deploy/systemd/aios-online-agent-isolated-executor.service").read_text()
assert "PrivateNetwork=true" in unit
assert "InaccessiblePaths=/run/credentials" in unit
assert "LoadCredential" not in unit
assert "SupplementaryGroups=gitops" not in unit
tests/test_online_agent_auto_claim.py
+35 -2
View File
@@ -465,6 +465,9 @@ def test_governed_commit_uses_existing_exact_job_without_recreating(monkeypatch:
def call(self, route: str, body: dict) -> tuple[int, dict]:
self.routes.append(route)
if route == "/v2/job":
assert body == {"job_id": JOB}
else:
assert body["task_id"] == TASK
assert body["job_id"] == JOB
if route == "/v2/writer-lease":
@@ -479,13 +482,43 @@ def test_governed_commit_uses_existing_exact_job_without_recreating(monkeypatch:
"target_branch": BRANCH,
}
if route == "/v2/commit-files":
return 200, {"job_id": JOB, "target_branch": BRANCH, "generation": 1, "branch_mode": "CONTINUE_EXISTING", "commit_oid": commit_oid, "commit_visible": True}
return 200, {
"job_id": JOB,
"target_branch": BRANCH,
"generation": 1,
"branch_mode": "CONTINUE_EXISTING",
"commit_oid": commit_oid,
"commit_visible": True,
"receipt": {
"remote_verified": True,
"authority_oid": commit_oid,
"api_branch_oid": commit_oid,
"git_remote_oid": commit_oid,
},
}
if route == "/v2/job":
return 200, {
"state": "COMPLETED",
"receipt": {
"remote_verified": True,
"authority_oid": commit_oid,
"api_branch_oid": commit_oid,
"git_remote_oid": commit_oid,
},
"branch_leases": [{
"job_id": JOB,
"branch": BRANCH,
"state": "ACTIVE",
"generation": 1,
"current_head_oid": commit_oid,
}],
}
raise AssertionError(route)
bridge = Bridge()
monkeypatch.setattr(worker_module.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout=commit_oid + "\n", stderr=""))
heartbeat = SimpleNamespace(ensure=lambda: None)
result = governed_commit(envelope, tmp_path, bridge, heartbeat)
assert bridge.routes == ["/v2/writer-lease", "/v2/commit-files"]
assert bridge.routes == ["/v2/writer-lease", "/v2/commit-files", "/v2/job"]
assert result["job_id"] == JOB
assert result["commit_oid"] == commit_oid
tests/test_online_agent_plan_store.py
+60
View File
@@ -0,0 +1,60 @@
from pathlib import Path
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
from aios_online_agent_execution import GovernedExecutionBinding
from aios_online_agent_plan_store import ExecutionPlanStore, ExecutionPlanError
BASE="a642b045775b9154a00ae78543037fbef4fb447b"
HEAD="71523aef6e48b1e2fa516c456ca189b7a039104b"
def binding(head=HEAD):
return GovernedExecutionBinding.from_mapping({
"task_id":"task-plan-a","job_id":"job-plan-a","repository":"aios/aios","base_oid":BASE,
"expected_branch_oid":head,"target_branch":"agent/task-plan-a","branch_mode":"CONTINUE_EXISTING",
"generation":1,"idempotency_key":"governed-plan-a",
})
def request():
return {
"schema":"aios.online-agent.plan.v1","task_id":"task-plan-a","job_id":"job-plan-a",
"source_head_oid":HEAD,"generation":1,"idempotency_key":"plan-task-a-v1",
"execution":{
"action":"deterministic",
"change_set":[{"path":"x.py","operation":"write_file","content":"x=1\n"}],
"verification":[{"label":"compile","argv":["/usr/bin/python3","-m","py_compile","x.py"],"cwd":".","timeout_seconds":30}],
"capability_scope":["git.read","git.test"],
}
}
def test_plan_register_and_resolve(tmp_path):
store=ExecutionPlanStore(tmp_path)
created=store.register(binding(),request())
assert created["idempotent"] is False
resolved=store.resolve(binding())
assert resolved["plan_digest"]==created["plan_digest"]
assert resolved["execution"]["action"]=="deterministic"
def test_plan_is_idempotent(tmp_path):
store=ExecutionPlanStore(tmp_path); req=request()
first=store.register(binding(),req); second=store.register(binding(),req)
assert second["idempotent"] is True
assert first["plan_digest"]==second["plan_digest"]
def test_plan_source_head_drift_denied(tmp_path):
store=ExecutionPlanStore(tmp_path); store.register(binding(),request())
with pytest.raises(ExecutionPlanError,match="SOURCE_HEAD_DRIFT"):
store.resolve(binding("b"*40))
def test_plan_replacement_requires_cas(tmp_path):
store=ExecutionPlanStore(tmp_path); first=store.register(binding(),request())
changed=request(); changed["idempotency_key"]="plan-task-a-v2"
changed["execution"]["change_set"][0]["content"]="x=2\n"
with pytest.raises(ExecutionPlanError,match="REPLACEMENT_CAS_MISMATCH"):
store.register(binding(),changed)
changed["expected_previous_plan_digest"]=first["plan_digest"]
second=store.register(binding(),changed)
assert second["plan_digest"]!=first["plan_digest"]
tests/test_online_agent_scheme_a.py
+245
View File
@@ -0,0 +1,245 @@
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
import json
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
from aios_online_agent_execution import GovernedExecutionBinding, project_execution_envelope
from aios_governed_task_bootstrap import GovernedTaskBinding, GovernedTaskBootstrap
from aios_gitops_job_controller import GitOpsJobController, JobState, JobStore
from aios_continuation_executor import ContinuationExecutionError, DetachedGitWorkspace
from aios_online_agent_worker import (
BridgeClient,
IsolatedDeterministicExecutor,
load_task_state,
_claim_order,
_isolated_executor_wait_seconds,
_verify_converged_job,
WorkerError,
)
BASE="a642b045775b9154a00ae78543037fbef4fb447b"
HEAD="71523aef6e48b1e2fa516c456ca189b7a039104b"
TASK="p0-claim-online-agent-auto-execution-20260817"
JOB="job-2031a9e5682a4238f98e3ac7aee8a593"
BRANCH="agent/p0-claim-online-agent-auto-execution-20260817"
def test_source_signals_are_bounded_by_projection():
binding=GovernedExecutionBinding.from_mapping({
"task_id":TASK,"job_id":JOB,"repository":"aios/aios","base_oid":BASE,
"expected_branch_oid":BASE,"target_branch":BRANCH,"branch_mode":"CREATE_NEW",
"generation":1,"idempotency_key":"p0-claim-online-agent-auto-execution-a642b04-v1",
})
execution=project_execution_envelope(binding,{
"action":"source_capture",
"source_signals":["x"*5000,"x"*5000,"trace credential tool "*100],
})
assert len(execution["source_signals"]) <= 16
assert all(0 < len(item) <= 512 for item in execution["source_signals"])
def test_same_job_create_new_binding_normalizes_to_current_head():
binding=GovernedTaskBinding(
task_id=TASK,job_id=JOB,repository="aios/aios",base_oid=BASE,
expected_branch_oid=BASE,target_branch=BRANCH,branch_mode="CREATE_NEW",
generation=1,idempotency_key="p0-claim-online-agent-auto-execution-a642b04-v1",
risk="engineering",description="automatic claim",
)
bootstrap=GovernedTaskBootstrap.__new__(GovernedTaskBootstrap)
bootstrap.authority=SimpleNamespace(read_ref=lambda ref: HEAD)
class Connection:
def __enter__(self): return self
def __exit__(self,*a): return False
def execute(self,*a,**k):
return SimpleNamespace(fetchall=lambda:[{
"current_head_oid":HEAD,"generation":1,"state":"ACTIVE","lineage_anchor_oid":BASE
}])
bootstrap._store=lambda _task:(SimpleNamespace(_connect=lambda:Connection()),None)
normalized=bootstrap.execution_binding(binding)
assert normalized.base_oid == BASE
assert normalized.expected_branch_oid == HEAD
assert normalized.branch_mode == "CONTINUE_EXISTING"
assert normalized.generation == 1
def test_recovery_receipt_requires_exact_oid_and_active_lease():
envelope=SimpleNamespace(job_id=JOB,target_branch=BRANCH,generation=1)
response={
"state":"COMPLETED",
"receipt":{"remote_verified":True,"authority_oid":HEAD,"api_branch_oid":HEAD,"git_remote_oid":HEAD},
"branch_leases":[{
"job_id":JOB,"branch":BRANCH,"state":"ACTIVE","generation":1,"current_head_oid":HEAD
}],
}
_verify_converged_job(response,envelope,HEAD)
bad={**response,"receipt":{**response["receipt"],"git_remote_oid":BASE}}
with pytest.raises(WorkerError,match="GOVERNED_RECEIPT_OID_MISMATCH"):
_verify_converged_job(bad,envelope,HEAD)
def test_executor_unit_is_uncredentialed():
unit=(ROOT/"deploy/systemd/aios-online-agent-isolated-executor.service").read_text()
assert "PrivateNetwork=true" in unit
assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
assert "IPAddressDeny=any" in unit
assert "IPAddressAllow=localhost" in unit
assert "InaccessiblePaths=/run/credentials" in unit
assert "LoadCredential" not in unit
assert "SupplementaryGroups=gitops" not in unit
def test_isolated_executor_wait_budget_includes_safety_margin():
assert _isolated_executor_wait_seconds(({"timeout_seconds":1200},)) == 1230
def test_prepared_deterministic_plan_precedes_source_capture_at_same_priority():
deterministic=SimpleNamespace(action="deterministic",task_id="task-deterministic")
source_capture=SimpleNamespace(action="source_capture",task_id="task-source-capture")
task={"priority":100,"created_at":"2026-08-17T00:00:00Z"}
ordered=sorted(((task,source_capture),(task,deterministic)),key=_claim_order)
assert ordered[0][1] is deterministic
def test_plan_registry_gets_precedence_when_task_has_no_execution(monkeypatch):
client=BridgeClient("agent-token-placeholder")
seen=[]
monkeypatch.setattr(client,"call",lambda route,body:(seen.append((route,body)) or (200,{})))
task={"task_id":TASK,"required_capabilities":["git.read"]}
client.project_execution(task)
assert seen == [("/v2/execution-envelope",{"task_id":TASK,"required_capabilities":["git.read"]})]
client.project_execution(task,{"action":"source_capture","source_signals":["trace"]})
assert seen[-1][1]["execution"]["action"] == "source_capture"
def test_executor_exchange_files_are_group_readable_and_directories_setgid(tmp_path):
request=tmp_path/"request.json"
IsolatedDeterministicExecutor._atomic(request,b"{}")
assert request.stat().st_mode & 0o777 == 0o640
tmpfiles=(ROOT/"deploy/tmpfiles.d/aios-online-agent-executor.conf").read_text()
assert tmpfiles.count(" 2770 ") == 5
def test_precommit_state_recovery_requires_a_later_official_attempt(monkeypatch,tmp_path):
import aios_online_agent_worker as worker_module
monkeypatch.setattr(worker_module,"STATE_ROOT",tmp_path)
state_path=tmp_path/"tasks"/f"{TASK}.json"
state_path.parent.mkdir(parents=True)
state_path.write_text(json.dumps({"task_id":TASK,"execution_digest":"a"*64,"phase":"workspace_prepared"}))
assert load_task_state(TASK,"b"*64) is None
state_path.write_text(json.dumps({"task_id":TASK,"execution_digest":"a"*64,"phase":"verification_passed","attempt":16}))
with pytest.raises(WorkerError,match="TASK_PAYLOAD_DRIFT"):
load_task_state(TASK,"b"*64,attempt=16)
assert load_task_state(TASK,"b"*64,attempt=17) is None
def test_reconciliation_state_recovery_requires_later_attempt_and_same_job(monkeypatch,tmp_path):
import aios_online_agent_worker as worker_module
monkeypatch.setattr(worker_module,"STATE_ROOT",tmp_path)
state_path=tmp_path/"tasks"/f"{TASK}.json"
state_path.parent.mkdir(parents=True)
state_path.write_text(json.dumps({
"task_id":TASK,
"execution_digest":"a"*64,
"phase":"reconciliation_required",
"attempt":18,
"job_id":JOB,
"details":{"api_status":503,"error":"WORKTREE_DIRTY_SCOPE_MISMATCH"},
}))
with pytest.raises(WorkerError,match="TASK_PAYLOAD_DRIFT"):
load_task_state(TASK,"b"*64,attempt=18,continuation_job_id=JOB,continuation_head_oid=HEAD)
assert load_task_state(TASK,"b"*64,attempt=19,continuation_job_id=JOB,continuation_head_oid=HEAD) is None
with pytest.raises(WorkerError,match="TASK_PAYLOAD_DRIFT"):
load_task_state(TASK,"b"*64,attempt=19,continuation_job_id="job-other",continuation_head_oid=HEAD)
def test_legacy_dirty_scope_reconciliation_state_is_exactly_bounded(monkeypatch,tmp_path):
import aios_online_agent_worker as worker_module
monkeypatch.setattr(worker_module,"STATE_ROOT",tmp_path)
state_path=tmp_path/"tasks"/f"{TASK}.json"
state_path.parent.mkdir(parents=True)
state_path.write_text(json.dumps({
"task_id":TASK,
"execution_digest":"a"*64,
"phase":"reconciliation_required",
"job_id":JOB,
"details":{"api_status":503,"error":"WORKTREE_DIRTY_SCOPE_MISMATCH"},
}))
assert load_task_state(TASK,"b"*64,attempt=19,continuation_job_id=JOB,continuation_head_oid=HEAD) is None
with pytest.raises(WorkerError,match="TASK_PAYLOAD_DRIFT"):
load_task_state(TASK,"b"*64,attempt=19,continuation_job_id=JOB,continuation_head_oid="not-an-oid")
def test_completed_same_job_begins_one_idempotent_continuation_cycle(tmp_path):
controller=GitOpsJobController(JobStore(tmp_path/"jobs",payload_root=tmp_path/"payloads"))
submitted=controller.submit({
"idempotency_key":"same-job-continuation-test",
"campaign_id":TASK,
"task_id":TASK,
"repository":"aios/aios",
"operation":"GOVERNED_TASK_CREATE",
"branch":BRANCH,
"expected_branch_oid":HEAD,
"payload_digest":"a"*64,
})
job_id=submitted.job_id
controller.advance(job_id,JobState.AUTHORIZED,side_effect_state="AUTHORIZED",expected_revision=0)
controller.lease_job(
job_id,
lease_id="lease-"+job_id,
lineage_anchor_oid=BASE,
current_head_oid=HEAD,
generation=1,
lease_expires_at="2099-01-01T00:00:00+00:00",
expected_revision=1,
)
controller.advance(job_id,JobState.WORKTREE_READY,side_effect_state="WORKTREE_READY",expected_revision=2)
controller.advance(job_id,JobState.EXECUTING,side_effect_state="IN_PROGRESS",expected_revision=3)
controller.advance(job_id,JobState.LOCAL_VERIFIED,side_effect_state="LOCAL_VERIFIED",receipt={"authority_oid":HEAD},expected_revision=4)
controller.advance(job_id,JobState.REMOTE_SYNCING,side_effect_state="IN_PROGRESS",expected_revision=5)
receipt={
"remote_verified":True,
"commit_visible":True,
"authority_oid":HEAD,
"api_branch_oid":HEAD,
"git_remote_oid":HEAD,
}
controller.advance(job_id,JobState.REMOTE_VERIFIED,side_effect_state="REMOTE_VERIFIED",receipt=receipt,expected_revision=6)
controller.advance(job_id,JobState.COMPLETED,side_effect_state="REMOTE_VERIFIED",receipt=receipt,expected_revision=7)
first=controller.begin_continuation_cycle(
job_id,
idempotency_key="same-job-continuation-commit-v1",
expected_head_oid=HEAD,
generation=1,
expected_revision=8,
)
assert first == {"completed":False,"commit_oid":None,"idempotent":False}
assert controller.store.read(job_id)["state"] == "LEASED"
replay=controller.begin_continuation_cycle(
job_id,
idempotency_key="same-job-continuation-commit-v1",
expected_head_oid=HEAD,
generation=1,
expected_revision=8,
)
assert replay == {"completed":False,"commit_oid":None,"idempotent":True}
def test_deterministic_materialization_recovers_only_an_in_scope_dirty_subset(monkeypatch,tmp_path):
workspace=object.__new__(DetachedGitWorkspace)
workspace.path=tmp_path
monkeypatch.setattr(DetachedGitWorkspace,"_dirty_paths",lambda _self:{"old.py"})
workspace.materialize_files({"old.py":b"new-old\n","new.py":b"new\n"})
assert (tmp_path/"old.py").read_bytes() == b"new-old\n"
assert (tmp_path/"new.py").read_bytes() == b"new\n"
monkeypatch.setattr(DetachedGitWorkspace,"_dirty_paths",lambda _self:{"outside.py"})
with pytest.raises(ContinuationExecutionError,match="WORKTREE_DIRTY_SCOPE_MISMATCH"):
workspace.materialize_files({"old.py":b"again\n"})