From 323b615b002fe3c9820948674465c81d8fc6d489 Mon Sep 17 00:00:00 2001 From: AIOS Governed Continuation Date: Mon, 17 Aug 2026 11:54:03 +0800 Subject: [PATCH] fix(online-agent): close automatic execution loop --- .../app/aios_agent_task_retry_extension.py | 427 ++++++++++++++++++ active-runtime/app/aios_chatgpt_bridge_v2.py | 15 + .../app/aios_chatgpt_bridge_v2_server.py | 69 ++- .../app/aios_continuation_executor.py | 33 +- .../app/aios_gitops_job_controller.py | 187 +++++++- .../app/aios_governed_task_bootstrap.py | 47 +- active-runtime/app/aios_isolated_executor.py | 306 +++++++++++++ .../app/aios_online_agent_execution.py | 28 +- .../app/aios_online_agent_plan_store.py | 187 ++++++++ .../app/aios_online_agent_worker.py | 326 ++++++++++++- .../aios-agent-task-retry-extension.service | 36 ++ ...ios-online-agent-isolated-executor.service | 33 ++ .../systemd/aios-online-agent-worker.service | 3 +- .../aios-online-engineering-consumer.service | 3 +- .../aios-online-agent-executor.conf | 4 + .../aios-online-agent-executor.conf | 5 + .../2026-08-17-p0-claim-scheme-a-followup.md | 56 +++ tests/test_agent_task_retry_extension.py | 117 +++++ .../test_bridge_v2_governed_http_bootstrap.py | 2 +- tests/test_isolated_executor.py | 67 +++ tests/test_online_agent_auto_claim.py | 41 +- tests/test_online_agent_plan_store.py | 60 +++ tests/test_online_agent_scheme_a.py | 245 ++++++++++ 23 files changed, 2244 insertions(+), 53 deletions(-) create mode 100644 active-runtime/app/aios_agent_task_retry_extension.py create mode 100644 active-runtime/app/aios_isolated_executor.py create mode 100644 active-runtime/app/aios_online_agent_plan_store.py create mode 100644 deploy/systemd/aios-agent-task-retry-extension.service create mode 100644 deploy/systemd/aios-online-agent-isolated-executor.service create mode 100644 deploy/sysusers.d/aios-online-agent-executor.conf create mode 100644 deploy/tmpfiles.d/aios-online-agent-executor.conf create mode 100644 docs/superpowers/specs/2026-08-17-p0-claim-scheme-a-followup.md create mode 100644 tests/test_agent_task_retry_extension.py create mode 100644 tests/test_isolated_executor.py create mode 100644 tests/test_online_agent_plan_store.py create mode 100644 tests/test_online_agent_scheme_a.py diff --git a/active-runtime/app/aios_agent_task_retry_extension.py b/active-runtime/app/aios_agent_task_retry_extension.py new file mode 100644 index 0000000..c8d5abf --- /dev/null +++ b/active-runtime/app/aios_agent_task_retry_extension.py @@ -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() diff --git a/active-runtime/app/aios_chatgpt_bridge_v2.py b/active-runtime/app/aios_chatgpt_bridge_v2.py index 67920b8..2e6bdeb 100644 --- a/active-runtime/app/aios_chatgpt_bridge_v2.py +++ b/active-runtime/app/aios_chatgpt_bridge_v2.py @@ -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) diff --git a/active-runtime/app/aios_chatgpt_bridge_v2_server.py b/active-runtime/app/aios_chatgpt_bridge_v2_server.py index 3c0b11d..a1e7d62 100644 --- a/active-runtime/app/aios_chatgpt_bridge_v2_server.py +++ b/active-runtime/app/aios_chatgpt_bridge_v2_server.py @@ -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,7 +150,14 @@ class GovernedRuntime: while True: for path in sorted(self.projection_outbox_root.glob("*.json")): try: - self._deliver_projection_outbox(path) + 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 threading.Event().wait(5) @@ -521,8 +542,9 @@ class GovernedRuntime: expected_job_id=str(result.get("job_id") or "") or None, ) durable_binding = self.bootstrap.get_binding(result["task_id"]) - path = self._write_projection_outbox(result["task_id"], projection, durable_binding) - created = self._deliver_projection_outbox(path) + 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 { **result, "online_task_projection": { @@ -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) diff --git a/active-runtime/app/aios_continuation_executor.py b/active-runtime/app/aios_continuation_executor.py index d8aebf8..376f3c3 100644 --- a/active-runtime/app/aios_continuation_executor.py +++ b/active-runtime/app/aios_continuation_executor.py @@ -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() diff --git a/active-runtime/app/aios_gitops_job_controller.py b/active-runtime/app/aios_gitops_job_controller.py index cd13585..92f8510 100644 --- a/active-runtime/app/aios_gitops_job_controller.py +++ b/active-runtime/app/aios_gitops_job_controller.py @@ -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, diff --git a/active-runtime/app/aios_governed_task_bootstrap.py b/active-runtime/app/aios_governed_task_bootstrap.py index 981dcea..5c4ba80 100644 --- a/active-runtime/app/aios_governed_task_bootstrap.py +++ b/active-runtime/app/aios_governed_task_bootstrap.py @@ -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,8 +182,11 @@ 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): - return self._create_locked(request) + with self._branch_lock(repository, target_branch): + return self._create_locked(request) def _create_locked(self, request: Mapping[str, Any]) -> dict[str, Any]: values = {key: str(request[key]) for key in ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key")} diff --git a/active-runtime/app/aios_isolated_executor.py b/active-runtime/app/aios_isolated_executor.py new file mode 100644 index 0000000..126e8c9 --- /dev/null +++ b/active-runtime/app/aios_isolated_executor.py @@ -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)) diff --git a/active-runtime/app/aios_online_agent_execution.py b/active-runtime/app/aios_online_agent_execution.py index 0d978d0..bba945b 100644 --- a/active-runtime/app/aios_online_agent_execution.py +++ b/active-runtime/app/aios_online_agent_execution.py @@ -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", diff --git a/active-runtime/app/aios_online_agent_plan_store.py b/active-runtime/app/aios_online_agent_plan_store.py new file mode 100644 index 0000000..07aaf67 --- /dev/null +++ b/active-runtime/app/aios_online_agent_plan_store.py @@ -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"], + } diff --git a/active-runtime/app/aios_online_agent_worker.py b/active-runtime/app/aios_online_agent_worker.py index 46ffaa5..d469cfe 100644 --- a/active-runtime/app/aios_online_agent_worker.py +++ b/active-runtime/app/aios_online_agent_worker.py @@ -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", - { - "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}, - }, - ) + request: dict[str, Any] = { + "task_id": str(task.get("task_id") or ""), + "required_capabilities": list(task.get("required_capabilities") or []), + } + 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) diff --git a/deploy/systemd/aios-agent-task-retry-extension.service b/deploy/systemd/aios-agent-task-retry-extension.service new file mode 100644 index 0000000..9e3b9c3 --- /dev/null +++ b/deploy/systemd/aios-agent-task-retry-extension.service @@ -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 diff --git a/deploy/systemd/aios-online-agent-isolated-executor.service b/deploy/systemd/aios-online-agent-isolated-executor.service new file mode 100644 index 0000000..2330485 --- /dev/null +++ b/deploy/systemd/aios-online-agent-isolated-executor.service @@ -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 diff --git a/deploy/systemd/aios-online-agent-worker.service b/deploy/systemd/aios-online-agent-worker.service index 5be9430..8fe237d 100644 --- a/deploy/systemd/aios-online-agent-worker.service +++ b/deploy/systemd/aios-online-agent-worker.service @@ -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 diff --git a/deploy/systemd/aios-online-engineering-consumer.service b/deploy/systemd/aios-online-engineering-consumer.service index f7859c8..b7b6598 100644 --- a/deploy/systemd/aios-online-engineering-consumer.service +++ b/deploy/systemd/aios-online-engineering-consumer.service @@ -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 diff --git a/deploy/sysusers.d/aios-online-agent-executor.conf b/deploy/sysusers.d/aios-online-agent-executor.conf new file mode 100644 index 0000000..0544aa2 --- /dev/null +++ b/deploy/sysusers.d/aios-online-agent-executor.conf @@ -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 diff --git a/deploy/tmpfiles.d/aios-online-agent-executor.conf b/deploy/tmpfiles.d/aios-online-agent-executor.conf new file mode 100644 index 0000000..e8b5097 --- /dev/null +++ b/deploy/tmpfiles.d/aios-online-agent-executor.conf @@ -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 - diff --git a/docs/superpowers/specs/2026-08-17-p0-claim-scheme-a-followup.md b/docs/superpowers/specs/2026-08-17-p0-claim-scheme-a-followup.md new file mode 100644 index 0000000..cbda351 --- /dev/null +++ b/docs/superpowers/specs/2026-08-17-p0-claim-scheme-a-followup.md @@ -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. diff --git a/tests/test_agent_task_retry_extension.py b/tests/test_agent_task_retry_extension.py new file mode 100644 index 0000000..f20edae --- /dev/null +++ b/tests/test_agent_task_retry_extension.py @@ -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 diff --git a/tests/test_bridge_v2_governed_http_bootstrap.py b/tests/test_bridge_v2_governed_http_bootstrap.py index 77ca6c1..90124b2 100644 --- a/tests/test_bridge_v2_governed_http_bootstrap.py +++ b/tests/test_bridge_v2_governed_http_bootstrap.py @@ -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", diff --git a/tests/test_isolated_executor.py b/tests/test_isolated_executor.py new file mode 100644 index 0000000..252addc --- /dev/null +++ b/tests/test_isolated_executor.py @@ -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 diff --git a/tests/test_online_agent_auto_claim.py b/tests/test_online_agent_auto_claim.py index a76ba48..e25e53f 100644 --- a/tests/test_online_agent_auto_claim.py +++ b/tests/test_online_agent_auto_claim.py @@ -465,8 +465,11 @@ 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) - assert body["task_id"] == TASK - assert body["job_id"] == JOB + 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": return 200, { "task_id": TASK, @@ -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 diff --git a/tests/test_online_agent_plan_store.py b/tests/test_online_agent_plan_store.py new file mode 100644 index 0000000..d7878c7 --- /dev/null +++ b/tests/test_online_agent_plan_store.py @@ -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"] diff --git a/tests/test_online_agent_scheme_a.py b/tests/test_online_agent_scheme_a.py new file mode 100644 index 0000000..11b58f4 --- /dev/null +++ b/tests/test_online_agent_scheme_a.py @@ -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"})