Compare commits

...
20 changed files with 3090 additions and 26 deletions
@@ -224,12 +224,18 @@ def _profile_credential_environment(
def _profile_model_governance(
value: object,
) -> tuple[bool, tuple[str, ...], tuple[str, ...], int]:
) -> tuple[bool, tuple[str, ...], tuple[str, ...], int, int]:
if value is None:
return False, (), (), 0
return False, (), (), 0, 0
if not isinstance(value, Mapping):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_GOVERNANCE_PROFILE_INVALID")
allowed = {"required", "provider_ids", "model_capabilities", "max_output_tokens"}
allowed = {
"required",
"provider_ids",
"model_capabilities",
"max_output_tokens",
"input_reservation_floor_tokens",
}
if set(value) - allowed:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_GOVERNANCE_PROFILE_FIELDS_INVALID")
required = bool(value.get("required", False))
@@ -248,8 +254,18 @@ def _profile_model_governance(
):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_CAPABILITIES_INVALID")
max_output_tokens = int(value.get("max_output_tokens") or 0)
input_reservation_floor_tokens = int(
value.get("input_reservation_floor_tokens") or 0
)
if max_output_tokens < 0 or max_output_tokens > 262144:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_OUTPUT_LIMIT_INVALID")
if (
input_reservation_floor_tokens < 0
or input_reservation_floor_tokens > 262144
):
raise ExternalAgentRuntimeError(
"EXTERNAL_AGENT_MODEL_INPUT_RESERVATION_FLOOR_INVALID"
)
if required and not providers:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_PROVIDER_IDS_REQUIRED")
if required and max_output_tokens < 1:
@@ -259,6 +275,7 @@ def _profile_model_governance(
tuple(sorted({item.strip() for item in providers})),
tuple(sorted({item.strip() for item in capabilities})),
max_output_tokens,
input_reservation_floor_tokens,
)
@@ -385,6 +402,7 @@ class RuntimeProfile:
model_provider_ids: tuple[str, ...]
model_capabilities: tuple[str, ...]
model_max_output_tokens: int
model_input_reservation_floor_tokens: int
@classmethod
def load(
@@ -447,6 +465,7 @@ class RuntimeProfile:
model_provider_ids,
model_capabilities,
model_max_output_tokens,
model_input_reservation_floor_tokens,
) = _profile_model_governance(value.get("model_governance"))
profile = cls(
runtime,
@@ -461,6 +480,7 @@ class RuntimeProfile:
model_provider_ids,
model_capabilities,
model_max_output_tokens,
model_input_reservation_floor_tokens,
)
if profile.model_governance_required and not _runtime_model_binding_present(profile):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_BINDING_REQUIRED")
@@ -475,9 +495,16 @@ def validate_request(value: Mapping[str, Any]) -> dict[str, Any]:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_ID_INVALID")
runtime = _safe_runtime(value.get("runtime"))
task_id = str(value.get("task_id") or "")
attempt = value.get("attempt")
execution_digest = str(value.get("execution_digest") or "")
if not task_id or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{2,179}", task_id):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_TASK_ID_INVALID")
if (
not isinstance(attempt, int)
or isinstance(attempt, bool)
or not 1 <= attempt <= 1000
):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_ATTEMPT_INVALID")
if re.fullmatch(r"[0-9a-f]{64}", execution_digest) is None:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_EXECUTION_DIGEST_INVALID")
payload = value.get("payload")
@@ -497,6 +524,7 @@ def validate_request(value: Mapping[str, Any]) -> dict[str, Any]:
"request_id": request_id,
"runtime": runtime,
"task_id": task_id,
"attempt": attempt,
"execution_digest": execution_digest,
"payload": dict(payload),
}
@@ -673,6 +701,11 @@ def execute_request(
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_GOVERNANCE_DISABLED")
scope_type, tenant_id, project_id, trace_id = _scope_from_payload(request["payload"])
governance_store = ModelGovernanceStore(DEFAULT_MODEL_GOVERNANCE_DB)
prompt_estimated_input_tokens = _estimate_tokens(prompt)
reserved_estimated_input_tokens = max(
prompt_estimated_input_tokens,
profile.model_input_reservation_floor_tokens,
)
route_request = RouteRequest(
request_id=request["request_id"] + ":model",
task_id=request["task_id"],
@@ -688,7 +721,7 @@ def execute_request(
}
)
),
estimated_input_tokens=_estimate_tokens(prompt),
estimated_input_tokens=reserved_estimated_input_tokens,
max_output_tokens=profile.model_max_output_tokens,
)
try:
@@ -804,6 +837,13 @@ def execute_request(
"policy_revision": decision.policy_revision,
"model_revision": decision.model_revision,
"fallback_index": decision.fallback_index,
"prompt_estimated_input_tokens": prompt_estimated_input_tokens,
"input_reservation_floor_tokens": (
profile.model_input_reservation_floor_tokens
),
"reserved_estimated_input_tokens": (
reserved_estimated_input_tokens
),
"settlement": settlement,
"attempts": governance_attempts,
"audit_chain_valid": governance_store.verify_audit_chain(),
@@ -815,6 +855,7 @@ def execute_request(
"request_sha256": sha256(canonical(request)),
"runtime": request["runtime"],
"task_id": request["task_id"],
"attempt": request["attempt"],
"execution_digest": request["execution_digest"],
"exit_code": completed.returncode,
"duration_ms": int((time.monotonic() - started) * 1000),
@@ -865,6 +906,7 @@ class ExternalAgentExchange:
request_id: str,
runtime: str,
task_id: str,
attempt: int,
execution_digest: str,
payload: Mapping[str, Any],
timeout_seconds: int = 1800,
@@ -875,6 +917,7 @@ class ExternalAgentExchange:
"request_id": request_id,
"runtime": runtime,
"task_id": task_id,
"attempt": attempt,
"execution_digest": execution_digest,
"payload": dict(payload),
}
@@ -950,6 +993,7 @@ class ExternalAgentExchange:
or value.get("request_sha256") != request_sha
or value.get("runtime") != request["runtime"]
or value.get("task_id") != request["task_id"]
or value.get("attempt") != request["attempt"]
or value.get("execution_digest") != request["execution_digest"]
):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_RESULT_BINDING_INVALID")
+456
View File
@@ -0,0 +1,456 @@
#!/usr/bin/env python3
"""Disabled-by-default TLS edge for the read-only GPT Online Task ingress.
The internal AIOS ingress remains loopback-only. This service is a separate edge
that may bind non-loopback only when the owner explicitly approves it. It
forwards only the tiny read-only ingress API and never proxies the canonical
Bridge or Task API.
"""
from __future__ import annotations
from collections import defaultdict, deque
from dataclasses import dataclass
import hashlib
import hmac
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import re
import secrets
import ssl
import sys
import threading
import time
import urllib.error
import urllib.request
from typing import Any, Mapping
from urllib.parse import urlsplit
UPSTREAM = "http://127.0.0.1:18945"
MAX_BODY = 64 * 1024
MAX_RESPONSE = 256 * 1024
MAX_CONCURRENT_REQUESTS = 32
REQUEST_TIMEOUT_SECONDS = 20
RATE_WINDOW_SECONDS = 60
RATE_LIMIT_REQUESTS = 60
RATE_BURST_REQUESTS = 15
RATE_MAX_CLIENTS = 4096
_AUDIT_SALT = secrets.token_bytes(32)
PUBLIC_APPROVAL_ENV = "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED"
PUBLIC_DOMAIN_ENV = "AIOS_GPT_EDGE_PUBLIC_DOMAIN"
TASK_PATH_RE = re.compile(r"^/v1/read-only-tasks(?:/[A-Za-z0-9][A-Za-z0-9._:-]{7,127})?$")
def _client_pseudonym(value: str) -> str:
return hashlib.sha256(_AUDIT_SALT + value.encode("utf-8", "replace")).hexdigest()[:16]
def audit(event: str, **fields: object) -> None:
safe = {
"event": event,
"ts_unix_ms": int(time.time() * 1000),
}
for key, value in fields.items():
if key in {"authorization", "body", "token", "cookie", "client_ip"}:
continue
safe[key] = value
print(
json.dumps(safe, sort_keys=True, separators=(",", ":")),
file=sys.stderr,
flush=True,
)
class EdgeError(RuntimeError):
def __init__(self, code: str, status: int = 400) -> None:
super().__init__(code)
self.code = code
self.status = status
def canonical(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def public_bind_allowed(host: str, approval: str) -> str:
value = host.strip()
if value in {"127.0.0.1", "::1", "localhost"}:
return value
if approval != "1":
raise RuntimeError("GPT_EDGE_PUBLIC_BIND_OWNER_APPROVAL_REQUIRED")
if value not in {"0.0.0.0", "::"}:
raise RuntimeError("GPT_EDGE_BIND_ADDRESS_INVALID")
return value
def validated_public_domain(value: str) -> str:
domain = value.strip().lower()
if not domain:
return ""
if len(domain) > 253 or re.fullmatch(
r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?",
domain,
) is None:
raise RuntimeError("GPT_EDGE_PUBLIC_DOMAIN_INVALID")
return domain
def validated_edge_identity(host: str, approval: str, domain: str) -> tuple[str, str]:
bind_host = public_bind_allowed(host, approval)
public_domain = validated_public_domain(domain)
if bind_host not in {"127.0.0.1", "::1", "localhost"} and not public_domain:
raise RuntimeError("GPT_EDGE_PUBLIC_DOMAIN_REQUIRED")
return bind_host, public_domain
def validate_tls_file(path: Path, *, key: bool) -> Path:
if path.is_symlink() or not path.is_file():
raise RuntimeError("GPT_EDGE_TLS_FILE_INVALID")
size = path.stat().st_size
if size < 16 or size > 256 * 1024:
raise RuntimeError("GPT_EDGE_TLS_FILE_INVALID")
if key and (path.stat().st_mode & 0o077):
raise RuntimeError("GPT_EDGE_TLS_KEY_MODE_INVALID")
return path
def build_tls_context(cert_path: Path, key_path: Path) -> ssl.SSLContext:
cert = validate_tls_file(cert_path, key=False)
key = validate_tls_file(key_path, key=True)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.options |= ssl.OP_NO_COMPRESSION
context.load_cert_chain(certfile=str(cert), keyfile=str(key))
return context
def allowed_route(method: str, raw_path: str) -> str:
parsed = urlsplit(raw_path)
if parsed.query or parsed.fragment:
raise EdgeError("QUERY_OR_FRAGMENT_DENY", 400)
path = parsed.path
if method == "GET" and path in {"/health", "/openapi.json"}:
return path
if method == "POST" and path == "/v1/read-only-tasks":
return path
if method == "GET" and TASK_PATH_RE.fullmatch(path):
return path
raise EdgeError("ROUTE_DENY", 404)
def route_label(method: str, path: str) -> str:
if method == "GET" and path == "/health":
return "health"
if method == "GET" and path == "/openapi.json":
return "openapi"
if method == "POST" and path == "/v1/read-only-tasks":
return "task_create"
if method == "GET" and path.startswith("/v1/read-only-tasks/"):
return "task_status"
return "unknown"
def forwarded_headers(headers: Mapping[str, str], *, body: bool) -> dict[str, str]:
result = {"Accept": "application/json"}
authorization = headers.get("Authorization")
if not isinstance(authorization, str) or not authorization.startswith("Bearer "):
raise EdgeError("AUTHORIZATION_REQUIRED", 401)
token = authorization[7:]
if len(token) < 32 or len(token) > 4096 or any(ch.isspace() for ch in token):
raise EdgeError("AUTHORIZATION_INVALID", 401)
result["Authorization"] = authorization
if body:
content_type = headers.get("Content-Type", "")
if content_type.split(";", 1)[0].strip().lower() != "application/json":
raise EdgeError("CONTENT_TYPE_INVALID", 415)
result["Content-Type"] = "application/json"
return result
class RateLimiter:
def __init__(
self,
*,
window_seconds: int = RATE_WINDOW_SECONDS,
limit: int = RATE_LIMIT_REQUESTS,
burst: int = RATE_BURST_REQUESTS,
max_clients: int = RATE_MAX_CLIENTS,
now_fn=time.monotonic,
) -> None:
self.window_seconds = window_seconds
self.limit = limit
self.burst = burst
self.max_clients = max_clients
self.now_fn = now_fn
self._lock = threading.Lock()
self._events: dict[str, deque[float]] = defaultdict(deque)
def _prune_stale(self, cutoff: float) -> None:
stale = [
key
for key, events in self._events.items()
if not events or events[-1] <= cutoff
]
for key in stale:
self._events.pop(key, None)
def allowed(self, key: str) -> bool:
now = self.now_fn()
cutoff = now - self.window_seconds
with self._lock:
if key not in self._events and len(self._events) >= self.max_clients:
self._prune_stale(cutoff)
if len(self._events) >= self.max_clients:
return False
events = self._events[key]
while events and events[0] <= cutoff:
events.popleft()
if len(events) >= self.limit + self.burst:
return False
events.append(now)
return True
@dataclass(frozen=True)
class Proxy:
upstream: str = UPSTREAM
timeout: int = REQUEST_TIMEOUT_SECONDS
def request(
self,
method: str,
path: str,
headers: Mapping[str, str],
body: bytes | None,
) -> tuple[int, bytes, str]:
if self.upstream != UPSTREAM:
raise EdgeError("UPSTREAM_DRIFT_DENY", 503)
req = urllib.request.Request(
self.upstream + path,
method=method,
data=body,
headers=dict(headers),
)
try:
with urllib.request.urlopen(req, timeout=self.timeout) as response:
raw = response.read(MAX_RESPONSE + 1)
status = response.status
content_type = response.headers.get("Content-Type", "application/json")
except urllib.error.HTTPError as exc:
raw = exc.read(MAX_RESPONSE + 1)
status = exc.code
content_type = exc.headers.get("Content-Type", "application/json")
except Exception as exc:
raise EdgeError("UPSTREAM_UNAVAILABLE", 502) from exc
if len(raw) > MAX_RESPONSE:
raise EdgeError("UPSTREAM_RESPONSE_TOO_LARGE", 502)
if content_type.split(";", 1)[0].strip().lower() != "application/json":
raise EdgeError("UPSTREAM_CONTENT_TYPE_INVALID", 502)
return status, raw, "application/json"
class EdgeServer(ThreadingHTTPServer):
daemon_threads = True
request_queue_size = 64
allow_reuse_address = True
class Handler(BaseHTTPRequestHandler):
proxy = Proxy()
limiter = RateLimiter()
concurrency = threading.BoundedSemaphore(MAX_CONCURRENT_REQUESTS)
expected_domain = ""
server_version = "AIOSGPTIngressEdge/1.0"
sys_version = ""
def log_message(self, *_args: Any) -> None:
return
def respond_json(self, value: Mapping[str, Any], status: int) -> None:
raw = canonical(value)
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("Content-Security-Policy", "default-src 'none'")
self.send_header("Strict-Transport-Security", "max-age=31536000")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def send_proxy_response(self, status: int, raw: bytes, content_type: str) -> None:
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("Content-Security-Policy", "default-src 'none'")
self.send_header("Strict-Transport-Security", "max-age=31536000")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def client_key(self) -> str:
return str(self.client_address[0])
def validate_host(self) -> None:
if not self.expected_domain:
return
host = self.headers.get("Host", "").split(":", 1)[0].lower()
if not hmac.compare_digest(host, self.expected_domain):
raise EdgeError("HOST_HEADER_DENY", 421)
def body(self) -> bytes:
try:
length = int(self.headers.get("Content-Length", "-1"))
except ValueError as exc:
raise EdgeError("CONTENT_LENGTH_INVALID") from exc
if length < 1 or length > MAX_BODY:
raise EdgeError("REQUEST_SIZE_INVALID", 413)
raw = self.rfile.read(length)
if len(raw) != length:
raise EdgeError("REQUEST_BODY_TRUNCATED", 400)
# Validate JSON at the edge so malformed bodies do not consume upstream work.
try:
value = json.loads(raw.decode("utf-8"))
except Exception as exc:
raise EdgeError("REQUEST_JSON_INVALID", 400) from exc
if not isinstance(value, Mapping):
raise EdgeError("REQUEST_JSON_OBJECT_REQUIRED", 400)
return canonical(value)
def handle_proxy(self, method: str) -> None:
client_id = _client_pseudonym(self.client_key())
started = time.monotonic()
acquired = self.concurrency.acquire(blocking=False)
if not acquired:
audit("EDGE_BUSY", client_id=client_id, method=method)
self.respond_json({"error": "EDGE_BUSY"}, 503)
return
label = "unknown"
try:
self.validate_host()
path = allowed_route(method, self.path)
label = route_label(method, path)
if path == "/health" and method == "GET":
self.respond_json(
{
"status": "ok",
"service": "aios-gpt-https-edge",
"upstream": "loopback-readonly-ingress",
},
200,
)
audit(
"EDGE_REQUEST",
client_id=client_id,
method=method,
route=label,
status=200,
duration_ms=int((time.monotonic() - started) * 1000),
)
return
if not self.limiter.allowed(self.client_key()):
audit(
"EDGE_RATE_LIMIT",
client_id=client_id,
method=method,
route=label,
)
self.respond_json({"error": "RATE_LIMITED"}, 429)
return
has_body = method == "POST"
headers = forwarded_headers(self.headers, body=has_body)
body = self.body() if has_body else None
status, raw, content_type = self.proxy.request(
method,
path,
headers,
body,
)
self.send_proxy_response(status, raw, content_type)
audit(
"EDGE_REQUEST",
client_id=client_id,
method=method,
route=label,
status=status,
duration_ms=int((time.monotonic() - started) * 1000),
)
except EdgeError as exc:
audit(
"EDGE_REJECT",
client_id=client_id,
method=method,
route=label,
status=exc.status,
error=exc.code,
)
self.respond_json({"error": exc.code}, exc.status)
finally:
self.concurrency.release()
def do_GET(self) -> None:
self.handle_proxy("GET")
def do_POST(self) -> None:
self.handle_proxy("POST")
def do_PUT(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_PATCH(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_DELETE(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_CONNECT(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_OPTIONS(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def serve() -> None:
host, public_domain = validated_edge_identity(
os.environ.get("AIOS_GPT_EDGE_HOST", "127.0.0.1"),
os.environ.get(PUBLIC_APPROVAL_ENV, "0"),
os.environ.get(PUBLIC_DOMAIN_ENV, ""),
)
port = int(os.environ.get("AIOS_GPT_EDGE_PORT", "18443"))
if not 1024 <= port <= 65535:
raise RuntimeError("GPT_EDGE_PORT_INVALID")
Handler.expected_domain = public_domain
credentials = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
cert_path = Path(
os.environ.get(
"AIOS_GPT_EDGE_TLS_CERT_FILE",
str(credentials / "tls_cert"),
)
)
key_path = Path(
os.environ.get(
"AIOS_GPT_EDGE_TLS_KEY_FILE",
str(credentials / "tls_key"),
)
)
context = build_tls_context(cert_path, key_path)
server = EdgeServer((host, port), Handler)
server.socket = context.wrap_socket(server.socket, server_side=True)
server.serve_forever()
if __name__ == "__main__":
serve()
@@ -0,0 +1,816 @@
#!/usr/bin/env python3
"""Credential-isolated host-side ingress for externally scheduled read-only Online Tasks.
This service is intentionally not a public AIOS control plane. It accepts a tiny,
authenticated external intent and translates it to the canonical loopback Bridge.
Callers cannot select repository coordinates, capabilities, execution action,
worker, branch policy, credentials, verification decisions, or write permissions.
"""
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
import threading
import urllib.error
import urllib.request
from typing import Any, Mapping
SCHEMA = "aios.gpt-online-task-ingress.request.v1"
RESPONSE_SCHEMA = "aios.gpt-online-task-ingress.response.v1"
OPENAPI_VERSION = "3.1.0"
REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
JOB_ID_RE = re.compile(r"^job-[0-9a-f]{32}$")
OID_RE = re.compile(r"^[0-9a-f]{40}$")
MAX_BODY = 64 * 1024
MAX_GOAL = 12_000
MAX_INSTRUCTIONS = 24_000
MAX_CRITERIA = 16
MAX_CRITERION_TEXT = 1_000
READONLY_CAPABILITIES = ("git.read", "git.test")
REPOSITORY = "aios/aios"
class IngressError(RuntimeError):
def __init__(self, code: str, status: int = 400) -> None:
super().__init__(code)
self.code = code
self.status = status
def canonical(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def digest(value: object) -> str:
return hashlib.sha256(canonical(value)).hexdigest()
def now() -> str:
return datetime.now(timezone.utc).isoformat()
def load_secret(path: Path, *, code: str) -> str:
if path.is_symlink() or not path.is_file():
raise RuntimeError(code + "_UNAVAILABLE")
value = path.read_text(encoding="utf-8").strip()
if len(value) < 32 or "\n" in value or "\r" in value:
raise RuntimeError(code + "_INVALID")
return value
def normalized_request(value: Mapping[str, Any]) -> dict[str, Any]:
allowed = {
"schema",
"request_id",
"goal",
"instructions",
"success_criteria",
"constraints",
}
if set(value) - allowed:
raise IngressError("UNKNOWN_FIELDS")
if value.get("schema") != SCHEMA:
raise IngressError("SCHEMA_INVALID")
request_id = value.get("request_id")
if not isinstance(request_id, str) or REQUEST_ID_RE.fullmatch(request_id) is None:
raise IngressError("REQUEST_ID_INVALID")
goal = value.get("goal")
if not isinstance(goal, str) or not goal.strip() or len(goal) > MAX_GOAL:
raise IngressError("GOAL_INVALID")
instructions = value.get("instructions", "")
if not isinstance(instructions, str) or len(instructions) > MAX_INSTRUCTIONS:
raise IngressError("INSTRUCTIONS_INVALID")
criteria = value.get("success_criteria") or []
if not isinstance(criteria, list) or len(criteria) > MAX_CRITERIA:
raise IngressError("SUCCESS_CRITERIA_INVALID")
normalized_criteria = []
for item in criteria:
if not isinstance(item, Mapping) or set(item) != {"id", "description"}:
raise IngressError("SUCCESS_CRITERIA_INVALID")
criterion_id = item.get("id")
description = item.get("description")
if (
not isinstance(criterion_id, str)
or not criterion_id.strip()
or len(criterion_id) > 120
or not isinstance(description, str)
or not description.strip()
or len(description) > MAX_CRITERION_TEXT
):
raise IngressError("SUCCESS_CRITERIA_INVALID")
normalized_criteria.append(
{"id": criterion_id.strip(), "description": description.strip()}
)
constraints = value.get("constraints") or {}
if not isinstance(constraints, Mapping):
raise IngressError("CONSTRAINTS_INVALID")
allowed_constraints = {
"max_files",
"focus_paths",
"notes",
}
if set(constraints) - allowed_constraints:
raise IngressError("CONSTRAINTS_FIELDS_INVALID")
normalized_constraints: dict[str, Any] = {}
if "max_files" in constraints:
max_files = constraints["max_files"]
if (
not isinstance(max_files, int)
or isinstance(max_files, bool)
or not 1 <= max_files <= 500
):
raise IngressError("MAX_FILES_INVALID")
normalized_constraints["max_files"] = max_files
if "focus_paths" in constraints:
paths = constraints["focus_paths"]
if (
not isinstance(paths, list)
or len(paths) > 32
or not all(
isinstance(path, str)
and path
and len(path) <= 512
and not path.startswith("/")
and ".." not in Path(path).parts
for path in paths
)
):
raise IngressError("FOCUS_PATHS_INVALID")
normalized_constraints["focus_paths"] = paths
if "notes" in constraints:
notes = constraints["notes"]
if not isinstance(notes, str) or len(notes) > 4_000:
raise IngressError("CONSTRAINT_NOTES_INVALID")
normalized_constraints["notes"] = notes
return {
"schema": SCHEMA,
"request_id": request_id,
"goal": goal.strip(),
"instructions": instructions,
"success_criteria": normalized_criteria,
"constraints": normalized_constraints,
}
def validated_bind_host(value: str) -> str:
host = value.strip()
if host not in {"127.0.0.1", "::1", "localhost"}:
raise RuntimeError("GPT_INGRESS_NON_LOOPBACK_BIND_DENY")
return host
def task_id_for(request_id: str) -> str:
suffix = hashlib.sha256(request_id.encode("utf-8")).hexdigest()[:24]
return "gpt-readonly-" + suffix
def branch_for(task_id: str) -> str:
return "agent/external-gpt-readonly/" + task_id[-24:]
@dataclass(frozen=True)
class Upstream:
bridge_base: str
task_base: str
control_token: str
timeout: int = 20
def _request(
self,
base: str,
method: str,
path: str,
body: Mapping[str, Any] | None = None,
) -> tuple[int, dict[str, Any]]:
raw_body = None if body is None else canonical(body)
headers = {
"Authorization": "Bearer " + self.control_token,
"Accept": "application/json",
}
if raw_body is not None:
headers["Content-Type"] = "application/json"
req = urllib.request.Request(
base.rstrip("/") + path,
method=method,
data=raw_body,
headers=headers,
)
try:
with urllib.request.urlopen(req, timeout=self.timeout) as response:
raw = response.read()
status = response.status
except urllib.error.HTTPError as exc:
raw = exc.read()
status = exc.code
except Exception as exc:
raise IngressError("UPSTREAM_TRANSPORT_FAILED", 503) from exc
try:
value = json.loads(raw.decode("utf-8"))
except Exception as exc:
raise IngressError("UPSTREAM_RESPONSE_INVALID", 502) from exc
if not isinstance(value, Mapping):
raise IngressError("UPSTREAM_RESPONSE_INVALID", 502)
return status, dict(value)
def control_snapshot(self) -> dict[str, Any]:
status, value = self._request(
self.bridge_base,
"GET",
"/v2/control-plane-snapshot",
)
if status != 200:
raise IngressError("CONTROL_SNAPSHOT_UNAVAILABLE", 503)
return value
def create_governed_task(
self,
body: Mapping[str, Any],
) -> tuple[int, dict[str, Any]]:
return self._request(
self.bridge_base,
"POST",
"/v2/governed-task",
body,
)
def execution_envelope(
self,
body: Mapping[str, Any],
) -> tuple[int, dict[str, Any]]:
return self._request(
self.bridge_base,
"POST",
"/v2/execution-envelope",
body,
)
def get_task(self, task_id: str) -> tuple[int, dict[str, Any]]:
return self._request(
self.task_base,
"GET",
"/v1/tasks/" + task_id,
)
class Store:
def __init__(self, path: Path) -> None:
self.path = Path(path)
self._lock = threading.RLock()
self.path.parent.mkdir(parents=True, exist_ok=True)
with self._connect() as con:
con.execute(
"""
CREATE TABLE IF NOT EXISTS external_task_requests(
request_id TEXT PRIMARY KEY,
request_sha256 TEXT NOT NULL,
task_id TEXT NOT NULL UNIQUE,
job_id TEXT,
base_oid TEXT,
state TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
response_json TEXT
)
"""
)
con.commit()
def _connect(self) -> sqlite3.Connection:
con = sqlite3.connect(self.path, timeout=30)
con.row_factory = sqlite3.Row
con.execute("PRAGMA busy_timeout=30000")
return con
def get(self, request_id: str) -> sqlite3.Row | None:
with self._connect() as con:
return con.execute(
"SELECT * FROM external_task_requests WHERE request_id=?",
(request_id,),
).fetchone()
def reserve(self, request: Mapping[str, Any], task_id: str) -> sqlite3.Row:
request_sha = digest(request)
timestamp = now()
with self._lock, self._connect() as con:
con.execute("BEGIN IMMEDIATE")
row = con.execute(
"SELECT * FROM external_task_requests WHERE request_id=?",
(request["request_id"],),
).fetchone()
if row is not None:
if row["request_sha256"] != request_sha:
raise IngressError("IDEMPOTENCY_CONFLICT", 409)
con.commit()
return row
con.execute(
"""
INSERT INTO external_task_requests(
request_id,request_sha256,task_id,state,created_at,updated_at
) VALUES(?,?,?,?,?,?)
""",
(
request["request_id"],
request_sha,
task_id,
"PENDING",
timestamp,
timestamp,
),
)
con.commit()
row = self.get(request["request_id"])
assert row is not None
return row
def bound(
self,
request_id: str,
*,
job_id: str,
base_oid: str,
response: Mapping[str, Any],
) -> None:
with self._lock, self._connect() as con:
con.execute(
"""
UPDATE external_task_requests
SET job_id=?,base_oid=?,state='BOUND',response_json=?,updated_at=?
WHERE request_id=?
""",
(
job_id,
base_oid,
json.dumps(response, sort_keys=True, separators=(",", ":")),
now(),
request_id,
),
)
con.commit()
class IngressService:
def __init__(self, upstream: Upstream, store: Store) -> None:
self.upstream = upstream
self.store = store
@staticmethod
def _main_oid(snapshot: Mapping[str, Any]) -> str:
candidates = [
snapshot.get("authoritative_head_oid"),
snapshot.get("head"),
(snapshot.get("aggregate") or {}).get("authoritative_head_oid")
if isinstance(snapshot.get("aggregate"), Mapping)
else None,
]
for candidate in candidates:
if isinstance(candidate, str) and OID_RE.fullmatch(candidate):
return candidate
raise IngressError("AUTHORITATIVE_HEAD_UNAVAILABLE", 503)
@staticmethod
def _public_response(
request_id: str,
task_id: str,
job_id: str,
state: str,
) -> dict[str, Any]:
return {
"schema": RESPONSE_SCHEMA,
"request_id": request_id,
"task_id": task_id,
"job_id": job_id,
"state": state,
"execution_mode": "source_capture",
"capabilities": list(READONLY_CAPABILITIES),
"write_access": False,
"official_verification_required": True,
}
def _recover_existing(
self,
request: Mapping[str, Any],
row: sqlite3.Row,
) -> dict[str, Any] | None:
task_id = str(row["task_id"])
status, task = self.upstream.get_task(task_id)
if status == 404:
return None
if status != 200:
raise IngressError("TASK_RECOVERY_FAILED", 503)
metadata = task.get("metadata")
execution = (
metadata.get("execution")
if isinstance(metadata, Mapping)
and isinstance(metadata.get("execution"), Mapping)
else {}
)
job_id = str(execution.get("job_id") or row["job_id"] or "")
base_oid = str(execution.get("base_oid") or row["base_oid"] or "")
if (
task.get("task_id") != task_id
or not JOB_ID_RE.fullmatch(job_id)
or not OID_RE.fullmatch(base_oid)
or execution.get("action") != "source_capture"
or sorted(execution.get("capability_scope") or [])
!= sorted(READONLY_CAPABILITIES)
or (execution.get("governance") or {}).get("writer_lease") is not False
):
raise IngressError("TASK_RECOVERY_BINDING_INVALID", 409)
response = self._public_response(
request["request_id"],
task_id,
job_id,
str(task.get("status") or "BOUND"),
)
self.store.bound(
request["request_id"],
job_id=job_id,
base_oid=base_oid,
response=response,
)
return response
def create(self, raw: Mapping[str, Any]) -> dict[str, Any]:
request = normalized_request(raw)
task_id = task_id_for(request["request_id"])
row = self.store.reserve(request, task_id)
if row["state"] == "BOUND" and row["response_json"]:
return json.loads(row["response_json"])
recovered = self._recover_existing(request, row)
if recovered is not None:
return recovered
snapshot = self.upstream.control_snapshot()
base_oid = self._main_oid(snapshot)
target_branch = branch_for(task_id)
binding_key = "gpt-readonly-binding-" + digest(
{"request_id": request["request_id"], "base_oid": base_oid}
)[:24]
constraints = {
**request["constraints"],
"repository_mutation": "forbidden",
"control_plane_side_effects": "forbidden",
"credential_access": "forbidden",
}
body = {
"task_id": task_id,
"repository": REPOSITORY,
"base_oid": base_oid,
"expected_branch_oid": base_oid,
"target_branch": target_branch,
"branch_mode": "CREATE_NEW",
"generation": 1,
"risk": "low",
"description": "Externally requested read-only governed Online Task.",
"idempotency_key": binding_key,
"priority": 0,
"max_attempts": 3,
"require_online_task_projection": True,
"online_task": {
"goal": request["goal"],
"instructions": request["instructions"],
"success_criteria": request["success_criteria"],
"constraints": constraints,
"required_capabilities": list(READONLY_CAPABILITIES),
"priority": 0,
"max_attempts": 3,
},
"execution": {
"action": "source_capture",
"adapter": "source_capture",
"capability_scope": list(READONLY_CAPABILITIES),
"source_signals": [
"external read-only source analysis request",
"no repository mutation no credential access no control-plane side effects",
],
},
}
status, value = self.upstream.create_governed_task(body)
if status not in {200, 201, 202}:
if status == 409:
recovered = self._recover_existing(request, row)
if recovered is not None:
return recovered
raise IngressError(
"GOVERNED_TASK_CREATE_FAILED",
409 if status == 409 else 503,
)
job_id = str(value.get("job_id") or "")
if value.get("task_id") != task_id or JOB_ID_RE.fullmatch(job_id) is None:
raise IngressError("GOVERNED_TASK_RESPONSE_INVALID", 502)
envelope_status, envelope = self.upstream.execution_envelope(
{
"task_id": task_id,
"job_id": job_id,
"required_capabilities": list(READONLY_CAPABILITIES),
"execution": body["execution"],
}
)
if envelope_status != 200:
raise IngressError("EXECUTION_PROJECTION_FAILED", 503)
if (
envelope.get("task_id") != task_id
or envelope.get("job_id") != job_id
or envelope.get("base_oid") != base_oid
or envelope.get("action") != "source_capture"
or sorted(envelope.get("capability_scope") or [])
!= sorted(READONLY_CAPABILITIES)
or (envelope.get("governance") or {}).get("writer_lease") is not False
):
raise IngressError("EXECUTION_PROJECTION_INVALID", 502)
response = self._public_response(
request["request_id"],
task_id,
job_id,
"BOUND",
)
self.store.bound(
request["request_id"],
job_id=job_id,
base_oid=base_oid,
response=response,
)
return response
def status(self, request_id: str) -> dict[str, Any]:
if REQUEST_ID_RE.fullmatch(request_id) is None:
raise IngressError("REQUEST_ID_INVALID")
row = self.store.get(request_id)
if row is None:
raise IngressError("REQUEST_NOT_FOUND", 404)
task_id = str(row["task_id"])
status, task = self.upstream.get_task(task_id)
if status == 200 and not row["job_id"]:
metadata = task.get("metadata")
execution = (
metadata.get("execution")
if isinstance(metadata, Mapping)
and isinstance(metadata.get("execution"), Mapping)
else {}
)
job_id = str(execution.get("job_id") or "")
base_oid = str(execution.get("base_oid") or "")
if (
JOB_ID_RE.fullmatch(job_id)
and OID_RE.fullmatch(base_oid)
and execution.get("action") == "source_capture"
and sorted(execution.get("capability_scope") or [])
== sorted(READONLY_CAPABILITIES)
and (execution.get("governance") or {}).get("writer_lease") is False
):
recovered = self._public_response(
request_id,
task_id,
job_id,
str(task.get("status") or "BOUND"),
)
self.store.bound(
request_id,
job_id=job_id,
base_oid=base_oid,
response=recovered,
)
row = self.store.get(request_id) or row
if status == 404:
return {
"schema": RESPONSE_SCHEMA,
"request_id": request_id,
"task_id": task_id,
"state": "PENDING",
"official_verification_required": True,
}
if status != 200:
raise IngressError("TASK_STATUS_UNAVAILABLE", 503)
result = task.get("result") if isinstance(task.get("result"), Mapping) else {}
verification = (
task.get("verification")
if isinstance(task.get("verification"), Mapping)
else {}
)
safe = {
"schema": RESPONSE_SCHEMA,
"request_id": request_id,
"task_id": task_id,
"job_id": str(row["job_id"] or ""),
"state": str(task.get("status") or "UNKNOWN"),
"attempt": int(task.get("attempt") or 0),
"claimed": bool(task.get("claimed_by")),
"result_status": result.get("status"),
"result_classification": result.get("classification"),
"official_verification": verification.get("decision"),
"official_classification": verification.get("classification"),
"write_access": False,
}
if result.get("execution_mode") == "source_capture":
snapshot = result.get("source_snapshot")
if isinstance(snapshot, Mapping):
safe["source_snapshot_sha256"] = snapshot.get("payload_sha256")
safe["source_snapshot_file_count"] = snapshot.get("file_count")
return safe
def openapi_document() -> dict[str, Any]:
return {
"openapi": OPENAPI_VERSION,
"info": {
"title": "AIOS Read-Only Online Task Ingress",
"version": "1.0.0",
},
"paths": {
"/v1/read-only-tasks": {
"post": {
"operationId": "createReadOnlyOnlineTask",
"security": [{"bearerAuth": []}],
"requestBody": {"required": True},
"responses": {"200": {"description": "Task accepted"}},
}
},
"/v1/read-only-tasks/{request_id}": {
"get": {
"operationId": "getReadOnlyOnlineTaskStatus",
"security": [{"bearerAuth": []}],
"responses": {"200": {"description": "Sanitized task status"}},
}
},
},
"components": {
"securitySchemes": {
"bearerAuth": {
"type": "http",
"scheme": "bearer",
}
}
},
}
class Handler(BaseHTTPRequestHandler):
service: IngressService | None = None
ingress_token: str = ""
def log_message(self, *_args: Any) -> None:
return
def respond(self, value: Mapping[str, Any], status: int) -> None:
raw = canonical(value)
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Cache-Control", "no-store")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def authorized(self) -> bool:
header = self.headers.get("Authorization", "")
return (
header.startswith("Bearer ")
and bool(self.ingress_token)
and hmac.compare_digest(header[7:], self.ingress_token)
)
def body(self) -> dict[str, Any]:
try:
length = int(self.headers.get("Content-Length", "-1"))
except ValueError as exc:
raise IngressError("CONTENT_LENGTH_INVALID") from exc
if length < 1 or length > MAX_BODY:
raise IngressError("REQUEST_SIZE_INVALID", 413)
try:
value = json.loads(self.rfile.read(length).decode("utf-8"))
except Exception as exc:
raise IngressError("REQUEST_JSON_INVALID") from exc
if not isinstance(value, Mapping):
raise IngressError("REQUEST_OBJECT_REQUIRED")
return dict(value)
def do_GET(self) -> None:
if self.path == "/health":
self.respond(
{
"status": "ok",
"service": "aios-gpt-online-task-ingress",
"mode": "read-only",
},
200,
)
return
if self.path == "/openapi.json":
self.respond(openapi_document(), 200)
return
if not self.authorized():
self.respond({"error": "UNAUTHORIZED"}, 401)
return
match = re.fullmatch(r"/v1/read-only-tasks/([^/?#]+)", self.path)
if match is None:
self.respond({"error": "NOT_FOUND"}, 404)
return
try:
if self.service is None:
raise IngressError("RUNTIME_UNAVAILABLE", 503)
self.respond(self.service.status(match.group(1)), 200)
except IngressError as exc:
self.respond({"error": exc.code}, exc.status)
def do_POST(self) -> None:
if self.path != "/v1/read-only-tasks":
self.respond({"error": "NOT_FOUND"}, 404)
return
if not self.authorized():
self.respond({"error": "UNAUTHORIZED"}, 401)
return
try:
if self.service is None:
raise IngressError("RUNTIME_UNAVAILABLE", 503)
value = self.service.create(self.body())
self.respond(value, 200)
except IngressError as exc:
self.respond({"error": exc.code}, exc.status)
def serve() -> None:
ingress_token_path = Path(
os.environ.get(
"AIOS_GPT_INGRESS_TOKEN_FILE",
str(
Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
/ "gpt_ingress_token"
),
)
)
control_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"
),
)
)
ingress_token = load_secret(
ingress_token_path,
code="GPT_INGRESS_CREDENTIAL",
)
control_token = load_secret(
control_token_path,
code="CONTROL_PLANE_CREDENTIAL",
)
state_root = Path(
os.environ.get(
"AIOS_GPT_INGRESS_STATE_ROOT",
"/var/lib/aios-gpt-online-task-ingress",
)
)
host = validated_bind_host(
os.environ.get("AIOS_GPT_INGRESS_HOST", "127.0.0.1")
)
port = int(os.environ.get("AIOS_GPT_INGRESS_PORT", "18945"))
if not 1024 <= port <= 65535:
raise RuntimeError("GPT_INGRESS_PORT_INVALID")
Handler.service = IngressService(
Upstream(
bridge_base=os.environ.get(
"AIOS_GOVERNED_WRITE_API_URL",
"http://127.0.0.1:18892",
),
task_base=os.environ.get(
"AIOS_AGENT_TASK_API_URL",
"http://127.0.0.1:18940",
),
control_token=control_token,
),
Store(state_root / "state.sqlite3"),
)
Handler.ingress_token = ingress_token
ThreadingHTTPServer((host, port), Handler).serve_forever()
if __name__ == "__main__":
serve()
+87 -18
View File
@@ -583,11 +583,44 @@ class ExecutionEnvelope:
)
def external_agent_request_id(
envelope: ExecutionEnvelope,
runtime_name: str,
attempt: int,
) -> str:
if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
raise WorkerError("TASK_ATTEMPT_INVALID", phase="external_agent")
if runtime_name not in {"hermes", "openclaw"}:
raise WorkerError(
"EXTERNAL_AGENT_RUNTIME_UNSUPPORTED",
phase="external_agent",
details={"runtime": runtime_name},
)
return (
f"external-{runtime_name}-"
+ digest_json(
{
"task_id": envelope.task_id,
"job_id": envelope.job_id,
"execution_digest": envelope.digest,
"runtime": runtime_name,
"attempt": attempt,
}
)[:32]
)
class RuntimeAdapter(ABC):
name: str
@abstractmethod
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
def execute(
self,
envelope: ExecutionEnvelope,
worktree: Path,
*,
attempt: int,
) -> dict[str, Any]:
raise NotImplementedError
@@ -608,7 +641,13 @@ class DeterministicExecutor(RuntimeAdapter):
if os.path.exists(temporary):
os.unlink(temporary)
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
def execute(
self,
envelope: ExecutionEnvelope,
worktree: Path,
*,
attempt: int,
) -> dict[str, Any]:
changed: set[str] = set()
for change in envelope.change_set:
relative = str(change["path"])
@@ -663,7 +702,13 @@ class SourceCaptureExecutor(RuntimeAdapter):
"governed", "trace", "tool", "credential", "scope", "message", "handoff",
}
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
def execute(
self,
envelope: ExecutionEnvelope,
worktree: Path,
*,
attempt: int,
) -> dict[str, Any]:
terms = set(self.defaults)
for signal in envelope.source_signals:
terms.update(token.lower() for token in re.findall(r"[A-Za-z][A-Za-z0-9_.-]{2,63}", signal))
@@ -732,7 +777,13 @@ class CompactInterfaceMapExecutor(RuntimeAdapter):
name = "compact_interface_map"
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
def execute(
self,
envelope: ExecutionEnvelope,
worktree: Path,
*,
attempt: int,
) -> dict[str, Any]:
if envelope.action != "source_capture":
raise WorkerError("COMPACT_INTERFACE_MAP_ACTION_INVALID", phase="interface_map")
try:
@@ -762,7 +813,13 @@ class CompactInterfaceMapExecutor(RuntimeAdapter):
class CodexExecutor(RuntimeAdapter):
name = "codex"
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
def execute(
self,
envelope: ExecutionEnvelope,
worktree: Path,
*,
attempt: int,
) -> dict[str, Any]:
raise WorkerError(
"CODEX_ADAPTER_REQUIRES_EXPLICIT_POLICY",
phase="adapter",
@@ -773,7 +830,13 @@ class CodexExecutor(RuntimeAdapter):
class ExternalAgentExecutor(RuntimeAdapter):
runtime_name: str
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
def execute(
self,
envelope: ExecutionEnvelope,
worktree: Path,
*,
attempt: int,
) -> dict[str, Any]:
if envelope.action != "agent_runtime":
raise WorkerError(
"EXTERNAL_AGENT_ACTION_INVALID",
@@ -787,16 +850,10 @@ class ExternalAgentExecutor(RuntimeAdapter):
)
if not envelope.runtime_input:
raise WorkerError("EXTERNAL_AGENT_RUNTIME_INPUT_MISSING", phase="adapter")
request_id = (
f"external-{self.runtime_name}-"
+ digest_json(
{
"task_id": envelope.task_id,
"job_id": envelope.job_id,
"execution_digest": envelope.digest,
"runtime": self.runtime_name,
}
)[:32]
request_id = external_agent_request_id(
envelope,
self.runtime_name,
attempt,
)
exchange = ExternalAgentExchange(EXTERNAL_AGENT_EXECUTOR_ROOT)
try:
@@ -804,6 +861,7 @@ class ExternalAgentExecutor(RuntimeAdapter):
request_id=request_id,
runtime=self.runtime_name,
task_id=envelope.task_id,
attempt=attempt,
execution_digest=envelope.digest,
payload=envelope.runtime_input,
timeout_seconds=EXTERNAL_AGENT_EXECUTOR_TIMEOUT_SECONDS,
@@ -819,6 +877,7 @@ class ExternalAgentExecutor(RuntimeAdapter):
task_id=envelope.task_id,
adapter=self.runtime_name,
request_id=request_id,
attempt=attempt,
output_sha256=external.get("output_sha256"),
duration_ms=external.get("duration_ms"),
credential_material_exposed=False,
@@ -1354,7 +1413,13 @@ class IsolatedDeterministicExecutor(RuntimeAdapter):
if os.path.exists(temporary):
os.unlink(temporary)
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
def execute(
self,
envelope: ExecutionEnvelope,
worktree: Path,
*,
attempt: int,
) -> dict[str, Any]:
if envelope.action != "deterministic":
raise WorkerError("ISOLATED_EXECUTOR_ACTION_INVALID", phase="adapter")
root = ISOLATED_EXECUTOR_ROOT
@@ -2193,7 +2258,11 @@ class WorkerSupervisor:
save_task_state(task_id, state)
save_checkpoint(task_id, state, envelope, attempt)
adapter = adapter_for(envelope.adapter)
execution = adapter.execute(envelope, worktree)
execution = adapter.execute(
envelope,
worktree,
attempt=attempt,
)
state.update(
{
"phase": "verification_passed",
@@ -31,6 +31,7 @@
"input_mode": "stdin_text",
"max_output_bytes": 131072,
"model_governance": {
"input_reservation_floor_tokens": 16384,
"max_output_tokens": 256,
"model_capabilities": [
"text"
@@ -45,7 +45,7 @@
"daily_token_budget": 250000,
"denied_providers": [],
"fallback_limit": 1,
"max_input_tokens_per_request": 16000,
"max_input_tokens_per_request": 16384,
"max_output_tokens_per_request": 512,
"project_id": null,
"required_capabilities": [
@@ -63,7 +63,7 @@
"daily_token_budget": 250000,
"denied_providers": [],
"fallback_limit": 1,
"max_input_tokens_per_request": 16000,
"max_input_tokens_per_request": 16384,
"max_output_tokens_per_request": 512,
"project_id": "staging",
"required_capabilities": [
@@ -0,0 +1,41 @@
[Unit]
Description=AIOS disabled-by-default HTTPS edge for GPT read-only Online Task ingress
After=network-online.target aios-gpt-online-task-ingress.service
Wants=network-online.target aios-gpt-online-task-ingress.service
ConditionPathExists=/etc/aios-gitops/gpt-https-edge.approved
[Service]
Type=simple
DynamicUser=yes
LoadCredential=tls_cert:/etc/aios-gitops/gpt-https-edge-tls.crt
LoadCredential=tls_key:/etc/aios-gitops/gpt-https-edge-tls.key
Environment=AIOS_GPT_EDGE_TLS_CERT_FILE=%d/tls_cert
Environment=AIOS_GPT_EDGE_TLS_KEY_FILE=%d/tls_key
Environment=AIOS_GPT_EDGE_HOST=127.0.0.1
Environment=AIOS_GPT_EDGE_PORT=18443
Environment=AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0
Environment=AIOS_GPT_EDGE_PUBLIC_DOMAIN=
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_gpt_https_edge.py
Restart=on-failure
RestartSec=2
NoNewPrivileges=yes
PrivateTmp=yes
PrivateDevices=yes
ProtectSystem=strict
ProtectHome=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectControlGroups=yes
ProtectKernelLogs=yes
ProtectClock=yes
LockPersonality=yes
RestrictSUIDSGID=yes
RestrictRealtime=yes
RestrictNamespaces=yes
MemoryDenyWriteExecute=yes
RestrictAddressFamilies=AF_INET AF_INET6
InaccessiblePaths=/data/aios-gitops /etc/aios-gitops /home /var/lib/aios-agent-task
UMask=0077
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,45 @@
[Unit]
Description=AIOS external GPT read-only Online Task ingress
After=network.target aios-chatgpt-bridge-v2-staging.service aios-agent-task-api.service
Wants=aios-chatgpt-bridge-v2-staging.service aios-agent-task-api.service
[Service]
Type=simple
DynamicUser=yes
StateDirectory=aios-gpt-online-task-ingress
StateDirectoryMode=0750
LoadCredential=gpt_ingress_token:/etc/aios-gitops/gpt-online-task-ingress.token
LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
Environment=AIOS_GPT_INGRESS_TOKEN_FILE=%d/gpt_ingress_token
Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
Environment=AIOS_GPT_INGRESS_STATE_ROOT=%S/aios-gpt-online-task-ingress
Environment=AIOS_GPT_INGRESS_HOST=127.0.0.1
Environment=AIOS_GPT_INGRESS_PORT=18945
Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_gpt_online_task_ingress.py
Restart=on-failure
RestartSec=2
NoNewPrivileges=yes
PrivateTmp=yes
PrivateDevices=yes
ProtectSystem=strict
ProtectHome=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectControlGroups=yes
ProtectKernelLogs=yes
ProtectClock=yes
LockPersonality=yes
RestrictSUIDSGID=yes
RestrictRealtime=yes
RestrictNamespaces=yes
MemoryDenyWriteExecute=yes
RestrictAddressFamilies=AF_UNIX AF_INET
IPAddressDeny=any
IPAddressAllow=127.0.0.0/8
InaccessiblePaths=/data/aios-gitops /etc/aios-gitops /home
UMask=0077
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,123 @@
# External Agent Attempt-Scoped Request Identity
## Problem
External Agent execution previously derived its durable request ID from:
```text
task_id
job_id
execution_digest
runtime
```
The Online Task attempt was missing.
That is incorrect for retryable external side effects. A failed Model Governance route reservation is
durable and terminal for that request ID. Reusing the same external request ID on a later official
Task attempt therefore reuses the old route decision/reservation and can fail settlement with:
```text
MODEL_RESERVATION_STATE_INVALID
```
Archiving an exchange failed marker does not fix the root cause because the Model Governance
reservation remains intentionally durable.
## Contract
External request identity is now:
```text
hash(
task_id,
job_id,
execution_digest,
runtime,
attempt
)
```
The attempt is a **transient execution coordinate**. It is not added to the signed governed execution
envelope and does not alter the Task/Job binding or execution digest.
The worker gets the authoritative attempt from the successful claim response and passes it through:
```text
process_claim
→ RuntimeAdapter.execute(..., attempt=N)
→ ExternalAgentExecutor
→ ExternalAgentExchange
→ external request JSON
→ external result JSON
```
The external result must bind the same attempt.
## Idempotency
This gives the intended two-level behavior:
```text
same Task/Job/execution/runtime + same attempt
→ same external request ID
→ same external result/reservation can be replayed safely
same Task/Job/execution/runtime + next official attempt
→ different external request ID
→ different Model Governance request ID
→ new reservation
```
Old FAILED reservations are retained for audit. No Model Governance DB cleanup is required or
permitted.
## Model Governance
Model Governance already derives its route request ID from:
```text
external_request_id + ":model"
```
Therefore making the external request ID attempt-scoped automatically makes route decisions and
reservations attempt-scoped without creating another state authority.
## Recovery
For the current Hermes live Task:
```text
attempt 6
→ old request/reservation remains FAILED
official NEEDS_REWORK
→ bounded retry extension 6 → 7
attempt 7
→ new external request ID
→ new route request ID
→ new reservation
→ provider execution
→ settlement
```
No replacement Task/Job, direct DB rewrite, failed-reservation deletion or marker deletion is
required.
## Worker Availability
The repair must preserve the already verified Worker Availability runtime. Runtime switches use the
maintenance inhibit and must finish with the guard healthy and all three workers fresh.
## Verification
Required regression coverage:
- request ID stable within one attempt;
- request ID different across attempts;
- request schema requires a valid attempt;
- external result binds attempt;
- old failed Model Governance reservation remains FAILED;
- next attempt gets a new reservation and can settle;
- existing external-agent and Model Governance runtime tests still pass.
@@ -0,0 +1,109 @@
# Model Governance Policy / Profile Floor Compatibility
## Problem
The root-owned OpenClaw runtime profile reserves a minimum of 16,384 input tokens so Model
Governance can cover OpenClaw-injected workspace, skills and system context before provider
execution.
The staging scope policies still limited input to 16,000 tokens per request. Route selection
therefore failed before reservation with:
```text
MODEL_INPUT_TOKEN_LIMIT_DENY
```
This was a configuration invariant violation:
```text
scope max_input_tokens_per_request
<
root-owned profile input_reservation_floor_tokens
```
The registered OpenClaw model itself supports 32,768 input tokens, so the model capability was not
the blocker.
## Invariant
For every Model Governance scope that allows a model/runtime whose root-owned profile declares an
input reservation floor:
```text
scope.max_input_tokens_per_request
>=
profile.model_governance.input_reservation_floor_tokens
```
The profile floor remains root-owned. Task payloads cannot alter it.
## Staging policy
The two staging policies that allow `openai/deepseek-v4-flash` are updated from:
```text
16000
```
to:
```text
16384
```
This is the minimum compatible value. It intentionally remains below the model's 32,768-token
technical input ceiling.
The daily token budget is unchanged.
## Governance path
The live SQLite policy must be changed only through the repository bootstrap and
`aios_model_governance_admin.py`.
The admin reconciler:
- reads the governed repository config;
- uses current policy revision as CAS;
- calls `ModelGovernanceStore.set_scope_policy`;
- appends normal Model Governance audit evidence;
- verifies the audit chain.
Direct SQL updates remain forbidden.
## Exhausted OpenClaw recovery
The fixed OpenClaw live Task has exhausted its normal three attempts.
After the governed policy candidate is officially verified and the live policy is reconciled:
```text
attempt 3 VERIFYING/FAILED result
→ official NEEDS_REWORK
→ Task becomes FAILED 3/3
→ official bounded retry extension 3 → 4
→ same Task / same Job returns QUEUED
→ persistent OpenClaw worker auto-claims attempt 4
```
AR1 attempt-scoped external request identity ensures attempt 4 receives a new external request and a
new Model Governance reservation.
No historical reservation is deleted or rewritten.
## Acceptance
Attempt 4 must prove:
```text
scope max_input_tokens_per_request=16384
profile input_reservation_floor_tokens=16384
reserved_estimated_input_tokens>=16384
actual input <= reserved input
settlement=SETTLED
usage_source=runtime
audit_chain_valid=true
```
If attempt 4 exposes a new failure, stop. No additional retry extension is authorized by this
repair.
@@ -0,0 +1,80 @@
# OpenClaw Input Reservation Floor
## Problem
Model Governance originally estimated input reservation from the user-visible prompt only.
The verified OpenClaw runtime injects additional root-owned/system-controlled context including
workspace, skills and system instructions. In live acceptance the prompt estimate reserved only
153 input tokens while OpenClaw reported approximately 7,770 actual input tokens.
The settlement correctly failed closed with:
```text
MODEL_USAGE_EXCEEDS_RESERVATION
```
The provider call itself had succeeded. The defect was therefore pre-execution reservation
underestimation, not provider connectivity.
## Contract
A root-owned external Agent profile may now declare:
```json
{
"model_governance": {
"input_reservation_floor_tokens": 16384
}
}
```
The effective input reservation is:
```text
max(
byte-based prompt estimate,
profile input_reservation_floor_tokens
)
```
Only the root-owned runtime profile can choose this floor. Online Task payloads cannot lower or
override it.
The OpenClaw staging profile uses 16,384 tokens. This is greater than twice the observed 7,770-token
live input high-water mark while remaining below the registered 32,768-token model input limit.
Hermes has no floor unless explicitly configured and therefore preserves its existing behavior.
## Fail-closed behavior
The floor is not an unlimited post-execution top-up.
If actual runtime usage exceeds the reserved amount, settlement still fails with
`MODEL_USAGE_EXCEEDS_RESERVATION`.
This keeps budget enforcement pre-execution and avoids silently authorizing usage after it has
already happened.
## Evidence
External results expose:
```text
prompt_estimated_input_tokens
input_reservation_floor_tokens
reserved_estimated_input_tokens
```
alongside the normal settlement and audit evidence.
## Retry
The current OpenClaw attempt 1 failure remains preserved. After this code repair is governed and
activated, official `NEEDS_REWORK` returns the same fixed Task to the normal retry path. Because AR1
already made external request identity attempt-scoped, attempt 2 receives a new external request ID
and a new reservation.
No retry-extension is required because the Task is currently attempt 1 of 3.
No Model Governance reservation is deleted or rewritten.
@@ -0,0 +1,120 @@
# Disabled-by-default HTTPS edge for external GPT Online Task access
## Purpose
The internal ingress (`127.0.0.1:18945`) remains loopback-only. This edge is the only component
allowed to become Internet-facing, and only after a separate owner approval.
It forwards only:
```text
POST /v1/read-only-tasks
GET /v1/read-only-tasks/{request_id}
GET /openapi.json
GET /health
```
It never proxies:
```text
18892 canonical Bridge
18940 Task API
WriterLease
commit
verification
release
tag
baseline
arbitrary URLs
```
## Default state
The systemd unit is disabled by policy unless the owner creates:
```text
/etc/aios-gitops/gpt-https-edge.approved
```
The base unit still binds only to:
```text
127.0.0.1:18443
AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0
```
Non-loopback bind is rejected by code unless an owner-controlled drop-in sets:
```text
AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1
AIOS_GPT_EDGE_HOST=0.0.0.0
AIOS_GPT_EDGE_PUBLIC_DOMAIN=<approved hostname>
```
This repository candidate does not create that drop-in or approval marker.
## TLS
TLS certificate/key are owner-managed systemd credentials:
```text
/etc/aios-gitops/gpt-https-edge-tls.crt
/etc/aios-gitops/gpt-https-edge-tls.key
```
The service requires TLS 1.2 or newer, disables TLS compression, and never logs or returns key
material.
Certificate/key issuance and renewal are external owner responsibilities until a separately
governed ACME workflow is implemented.
## Request protections
The edge provides:
- 64 KiB request cap;
- 256 KiB upstream response cap;
- JSON validation before proxying;
- fixed upstream `http://127.0.0.1:18945`;
- Authorization-only header forwarding;
- no Cookie/X-Forwarded-* trust;
- route/method allowlist;
- per-source-IP in-memory rate limiting;
- bounded concurrent request handling;
- optional exact Host-header enforcement;
- HSTS / no-store / nosniff / no-referrer response headers.
The rate limiter is intentionally simple and local. Multi-node/global abuse protection is outside
the current one-host architecture.
## Authentication
The HTTPS edge does not create a new authentication scheme. It forwards the external Bearer
credential to the loopback GPT ingress, which verifies `gpt_ingress_token`.
The edge never receives the AIOS control-plane token.
## Deployment boundary
Public activation is a high-risk network operation and requires explicit owner approval. Before
activation verify:
```text
host-side GPT ingress E2E=VERIFIED
owner ingress token provisioned
TLS certificate/key provisioned
public hostname owned and DNS configured
firewall/network exposure reviewed
explicit public-bind approval recorded
rollback plan available
```
Only after those gates may an owner-authorized runtime drop-in bind this service publicly.
Until then:
```text
GPT_HTTPS_EDGE_IMPLEMENTATION=candidate
PUBLIC_NETWORK_EXPOSURE=NONE
PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=NOT_VERIFIED
```
@@ -0,0 +1,138 @@
# External GPT → Online Task ingress
## Scope
This component closes the current architectural gap between an external Custom GPT and AIOS
without exposing the internal Task API (`18940`) or canonical Bridge (`18892`) directly.
The first supported public-facing capability is intentionally read-only:
```text
authenticated external intent
→ host-side ingress
→ canonical /v2/governed-task
→ canonical /v2/execution-envelope
→ persistent generic worker
→ source_capture result
→ official control-plane verification
```
The ingress itself never claims, heartbeats, verifies, commits, acquires WriterLease, or writes
protected refs.
## Security boundary
The service binds only to loopback (`127.0.0.1:18945`). A future owner-approved TLS reverse proxy
or tunnel may expose this service externally. Direct non-loopback binding is denied by code.
Two independent credentials are projected by systemd:
- `gpt_ingress_token`: authenticates the external caller to this ingress.
- `agent_task_control_token`: lets the ingress call the loopback AIOS control plane.
Neither credential is stored in Git, SQLite, task payloads, results, logs, or API responses.
The ingress has no repository filesystem access.
## Caller-controlled fields
Allowed:
```text
request_id
goal
instructions
success_criteria
constraints.max_files
constraints.focus_paths
constraints.notes
```
Not caller-controlled:
```text
repository
base_oid
expected_branch_oid
target_branch
branch_mode
generation
risk
priority
max_attempts
capabilities
action/adapter
WriterLease
worker identity
credentials
verification decision
main/production/tag/baseline permissions
```
All external tasks are hard-bound to:
```text
repository=aios/aios
capabilities=[git.read, git.test]
action=source_capture
writer_lease=false
repository_mutation=forbidden
control_plane_side_effects=forbidden
credential_access=forbidden
```
## Idempotency
`request_id` is the durable external idempotency boundary.
The ingress stores only:
```text
request hash
derived AIOS task id
canonical job id
base oid
sanitized receipt
state
```
An exact retry returns the same receipt. Reusing a `request_id` with different content returns
`IDEMPOTENCY_CONFLICT`.
A crash after canonical task creation can recover the same binding from the Task API rather than
creating a replacement Task.
## Status
`GET /v1/read-only-tasks/{request_id}` returns only sanitized state:
```text
task/job ids
status
attempt
whether claimed
result classification
official verification decision/classification
source snapshot hash/count when available
```
It does not proxy raw Task metadata, execution envelopes, result text, credentials, or source files.
## Public exposure
This candidate does **not** make AIOS publicly reachable. Public exposure requires a separately
approved HTTPS ingress/reverse proxy with:
- TLS;
- external authentication lifecycle;
- rate limiting / abuse controls;
- request size limits;
- observability;
- explicit owner approval for the network exposure.
Until that step is verified:
```text
HOST_SIDE_EXTERNAL_GPT_ADAPTER=implemented/staging-candidate
PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=not verified
```
@@ -0,0 +1,232 @@
from __future__ import annotations
import json
from pathlib import Path
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_online_agent_worker as worker
from aios_external_agent_runtime import (
ExternalAgentExchange,
ExternalAgentRuntimeError,
validate_request,
)
def envelope():
value=object.__new__(worker.ExecutionEnvelope)
object.__setattr__(value,"task_id","task-attempt-identity")
object.__setattr__(value,"job_id","job-attempt-identity")
object.__setattr__(value,"digest","d"*64)
return value
def test_external_request_id_is_stable_within_attempt_and_changes_across_attempts():
env=envelope()
a1=worker.external_agent_request_id(env,"hermes",6)
a1_replay=worker.external_agent_request_id(env,"hermes",6)
a2=worker.external_agent_request_id(env,"hermes",7)
assert a1==a1_replay
assert a1!=a2
assert a1.startswith("external-hermes-")
assert a2.startswith("external-hermes-")
def test_external_request_id_rejects_invalid_attempt():
env=envelope()
with pytest.raises(worker.WorkerError,match="TASK_ATTEMPT_INVALID"):
worker.external_agent_request_id(env,"hermes",0)
def test_external_request_schema_binds_attempt():
value=validate_request({
"schema":"aios.external-agent-runtime.request.v1",
"request_id":"external-hermes-0123456789abcdef",
"runtime":"hermes",
"task_id":"task-attempt-identity",
"attempt":7,
"execution_digest":"d"*64,
"payload":{
"goal":"Return ok.",
"instructions":"",
"success_criteria":[],
"constraints":{},
},
})
assert value["attempt"]==7
def test_external_request_schema_rejects_missing_attempt():
with pytest.raises(ExternalAgentRuntimeError,match="EXTERNAL_AGENT_ATTEMPT_INVALID"):
validate_request({
"schema":"aios.external-agent-runtime.request.v1",
"request_id":"external-hermes-0123456789abcdef",
"runtime":"hermes",
"task_id":"task-attempt-identity",
"execution_digest":"d"*64,
"payload":{
"goal":"Return ok.",
"instructions":"",
"success_criteria":[],
"constraints":{},
},
})
def _write_bound_result(path, request):
import hashlib
from aios_external_agent_runtime import canonical, sha256
material={
"schema":"aios.external-agent-runtime.result.v1",
"request_id":request["request_id"],
"request_sha256":sha256(canonical(request)),
"runtime":request["runtime"],
"task_id":request["task_id"],
"attempt":request["attempt"],
"execution_digest":request["execution_digest"],
"exit_code":0,
"duration_ms":1,
"output_text":"ok",
"output_sha256":hashlib.sha256(b"ok").hexdigest(),
"stderr_sha256":hashlib.sha256(b"").hexdigest(),
"credential_material_exposed":False,
"control_plane_access_granted":False,
}
material["result_sha256"]=sha256(canonical(material))
path.write_text(json.dumps(material,sort_keys=True),encoding="utf-8")
def test_exchange_result_binding_rejects_wrong_attempt(tmp_path):
exchange=ExternalAgentExchange(tmp_path)
request={
"schema":"aios.external-agent-runtime.request.v1",
"request_id":"external-hermes-0123456789abcdef",
"runtime":"hermes",
"task_id":"task-attempt-identity",
"attempt":7,
"execution_digest":"d"*64,
"payload":{
"goal":"Return ok.",
"instructions":"",
"success_criteria":[],
"constraints":{},
},
}
request=validate_request(request)
result_path=tmp_path/"result.json"
_write_bound_result(result_path,request)
value=json.loads(result_path.read_text())
value["attempt"]=6
# Recompute digest so the binding check, not the digest check, is what fails.
from aios_external_agent_runtime import canonical,sha256
material=dict(value); material.pop("result_sha256")
value["result_sha256"]=sha256(canonical(material))
result_path.write_text(json.dumps(value),encoding="utf-8")
with pytest.raises(ExternalAgentRuntimeError,match="EXTERNAL_AGENT_RESULT_BINDING_INVALID"):
exchange._validated_result(
result_path,
request,
sha256(canonical(request)),
)
def test_model_governance_request_id_is_attempt_scoped():
env=envelope()
attempt6=worker.external_agent_request_id(env,"hermes",6)+":model"
attempt7=worker.external_agent_request_id(env,"hermes",7)+":model"
assert attempt6!=attempt7
def test_failed_previous_attempt_does_not_poison_next_model_governance_reservation(tmp_path):
from aios_model_governance import (
ModelGovernanceStore,
ModelRecord,
ProviderRecord,
RouteRequest,
ScopePolicy,
)
store=ModelGovernanceStore(tmp_path/"model-governance.sqlite3",now_fn=lambda:1_800_000_000.0)
store.register_provider(
ProviderRecord(
"oneapi",
"runtime-profile:hermes",
max_failures=5,
cooldown_seconds=60,
)
)
store.register_model(
ModelRecord(
"deepseek-v4-flash",
"oneapi",
"v1",
("agent.runtime.hermes","text"),
input_cost_micros_per_1k=1,
output_cost_micros_per_1k=1,
max_input_tokens=10000,
max_output_tokens=1000,
evaluation_sha256="a"*64,
)
)
store.set_scope_policy(
ScopePolicy(
"tenant",
"model-governance-live",
"staging",
allowed_models=("deepseek-v4-flash",),
required_capabilities=("text",),
daily_token_budget=100000,
daily_cost_budget_micros=100000,
fallback_limit=1,
)
)
def route(request_id):
return store.route(
RouteRequest(
request_id=request_id,
task_id="r42-model-governance-hermes-live-20260818",
trace_id="trace-hermes-model-governance-live",
scope_type="tenant",
tenant_id="model-governance-live",
project_id="staging",
required_capabilities=("agent.runtime.hermes",),
estimated_input_tokens=100,
max_output_tokens=50,
)
)
attempt6=route("external-hermes-attempt6:model")
store.record_failure(attempt6.request_id,"provider_process_failed")
row6=store.conn.execute(
"select state from model_governance_reservations where request_id=?",
(attempt6.request_id,),
).fetchone()
assert row6["state"]=="FAILED"
attempt7=route("external-hermes-attempt7:model")
assert attempt7.request_id!=attempt6.request_id
row7=store.conn.execute(
"select state from model_governance_reservations where request_id=?",
(attempt7.request_id,),
).fetchone()
assert row7["state"]=="RESERVED"
settled=store.record_success(
attempt7.request_id,
input_tokens=90,
output_tokens=40,
usage_source="runtime",
)
assert settled["state"]=="SETTLED"
assert store.conn.execute(
"select state from model_governance_reservations where request_id=?",
(attempt6.request_id,),
).fetchone()["state"]=="FAILED"
assert store.verify_audit_chain() is True
store.close()
+4 -1
View File
@@ -110,6 +110,7 @@ def make_text_request(tmp_path, runtime, payload=None):
"request_id":f"external-{runtime}-0123456789abcdef",
"runtime":runtime,
"task_id":f"task-{runtime}-runtime",
"attempt":1,
"execution_digest":"d"*64,
"payload":payload or {
"goal":"Return ok.",
@@ -193,6 +194,7 @@ def test_external_executor_runs_root_owned_profile_without_shell(tmp_path,monkey
"request_id":f"external-{runtime}-0123456789abcdef",
"runtime":runtime,
"task_id":f"task-{runtime}-runtime",
"attempt":1,
"execution_digest":"d"*64,
"payload":{
"goal":"Return ok.",
@@ -326,6 +328,7 @@ def test_worker_external_adapter_builds_readonly_result(monkeypatch,tmp_path,run
"request_sha256":"e"*64,
"runtime":runtime,
"task_id":kwargs["task_id"],
"attempt":kwargs["attempt"],
"execution_digest":kwargs["execution_digest"],
"exit_code":0,
"duration_ms":10,
@@ -339,7 +342,7 @@ def test_worker_external_adapter_builds_readonly_result(monkeypatch,tmp_path,run
monkeypatch.setattr(worker,"ExternalAgentExchange",FakeExchange)
monkeypatch.setattr(worker,"audit",lambda *a,**k:None)
executor=worker.HermesExecutor() if runtime=="hermes" else worker.OpenClawExecutor()
execution=executor.execute(envelope,tmp_path)
execution=executor.execute(envelope,tmp_path,attempt=1)
result=worker.build_agent_runtime_result(envelope.task_id,envelope,execution)
assert result["status"]=="SUCCEEDED"
assert result["classification"]=="EXTERNAL_AGENT_RUNTIME_SUCCEEDED"
+205
View File
@@ -0,0 +1,205 @@
from __future__ import annotations
from email.message import Message
import json
from pathlib import Path
import stat
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_gpt_https_edge as edge
class Headers(dict):
def get(self,key,default=None):
return super().get(key,default)
def test_default_and_public_bind_require_explicit_owner_approval():
assert edge.public_bind_allowed("127.0.0.1","0")=="127.0.0.1"
assert edge.public_bind_allowed("::1","0")=="::1"
with pytest.raises(RuntimeError,match="OWNER_APPROVAL_REQUIRED"):
edge.public_bind_allowed("0.0.0.0","0")
with pytest.raises(RuntimeError,match="OWNER_APPROVAL_REQUIRED"):
edge.public_bind_allowed("::","")
assert edge.public_bind_allowed("0.0.0.0","1")=="0.0.0.0"
def test_arbitrary_public_bind_address_is_denied_even_with_approval():
with pytest.raises(RuntimeError,match="BIND_ADDRESS_INVALID"):
edge.public_bind_allowed("192.0.2.10","1")
def test_public_domain_validation():
assert edge.validated_public_domain("tasks.example.com")=="tasks.example.com"
assert edge.validated_public_domain("")==""
with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_INVALID"):
edge.validated_public_domain("https://example.com")
with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_INVALID"):
edge.validated_public_domain("localhost")
def test_route_allowlist_accepts_only_readonly_api():
assert edge.allowed_route("POST","/v1/read-only-tasks")=="/v1/read-only-tasks"
assert edge.allowed_route("GET","/v1/read-only-tasks/external-request-0001")==(
"/v1/read-only-tasks/external-request-0001"
)
assert edge.allowed_route("GET","/health")=="/health"
assert edge.allowed_route("GET","/openapi.json")=="/openapi.json"
for method,path in [
("POST","/v2/governed-task"),
("GET","/v1/tasks/x"),
("POST","/v1/read-only-tasks/x"),
("DELETE","/v1/read-only-tasks/x"),
]:
with pytest.raises(edge.EdgeError,match="ROUTE_DENY"):
edge.allowed_route(method,path)
def test_query_and_fragment_are_denied():
with pytest.raises(edge.EdgeError,match="QUERY_OR_FRAGMENT_DENY"):
edge.allowed_route("GET","/health?x=1")
with pytest.raises(edge.EdgeError,match="QUERY_OR_FRAGMENT_DENY"):
edge.allowed_route("GET","/openapi.json#x")
def test_forwarded_headers_are_minimal_and_require_bearer():
incoming=Headers({
"Authorization":"Bearer "+"a"*40,
"Content-Type":"application/json; charset=utf-8",
"Cookie":"secret=1",
"X-Forwarded-For":"203.0.113.10",
"X-Api-Key":"do-not-forward",
})
forwarded=edge.forwarded_headers(incoming,body=True)
assert forwarded=={
"Accept":"application/json",
"Authorization":"Bearer "+"a"*40,
"Content-Type":"application/json",
}
with pytest.raises(edge.EdgeError,match="AUTHORIZATION_REQUIRED"):
edge.forwarded_headers(Headers({}),body=False)
def test_invalid_content_type_is_denied():
with pytest.raises(edge.EdgeError,match="CONTENT_TYPE_INVALID"):
edge.forwarded_headers(
Headers({
"Authorization":"Bearer "+"a"*40,
"Content-Type":"text/plain",
}),
body=True,
)
def test_rate_limiter_caps_window_and_allows_expiry():
now=[100.0]
limiter=edge.RateLimiter(window_seconds=60,limit=2,burst=1,now_fn=lambda:now[0])
assert limiter.allowed("client")
assert limiter.allowed("client")
assert limiter.allowed("client")
assert not limiter.allowed("client")
now[0]=161.0
assert limiter.allowed("client")
def test_rate_limiter_client_table_is_bounded():
now=[100.0]
limiter=edge.RateLimiter(
window_seconds=60,
limit=10,
burst=0,
max_clients=2,
now_fn=lambda:now[0],
)
assert limiter.allowed("a")
assert limiter.allowed("b")
assert not limiter.allowed("c")
now[0]=161.0
assert limiter.allowed("c")
assert len(limiter._events) <= 2
def test_audit_source_does_not_log_sensitive_request_material():
source=(ROOT/"active-runtime/app/aios_gpt_https_edge.py").read_text()
assert '"authorization", "body", "token", "cookie", "client_ip"' in source
assert "_client_pseudonym" in source
def test_proxy_upstream_is_fixed():
proxy=edge.Proxy()
assert proxy.upstream=="http://127.0.0.1:18945"
bad=edge.Proxy(upstream="http://127.0.0.1:18892")
with pytest.raises(edge.EdgeError,match="UPSTREAM_DRIFT_DENY"):
bad.request("GET","/health",{"Authorization":"Bearer "+"a"*40},None)
def test_public_bind_requires_approved_domain():
assert edge.validated_edge_identity("127.0.0.1","0","")==("127.0.0.1","")
with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_REQUIRED"):
edge.validated_edge_identity("0.0.0.0","1","")
assert edge.validated_edge_identity(
"0.0.0.0","1","tasks.example.com"
)==("0.0.0.0","tasks.example.com")
def test_tls_key_requires_no_group_or_other_permissions(tmp_path):
key=tmp_path/"key.pem"
key.write_text("x"*64,encoding="utf-8")
key.chmod(0o600)
assert edge.validate_tls_file(key,key=True)==key
key.chmod(0o644)
with pytest.raises(RuntimeError,match="TLS_KEY_MODE_INVALID"):
edge.validate_tls_file(key,key=True)
def test_tls_symlink_is_denied(tmp_path):
real=tmp_path/"cert.pem"
real.write_text("x"*64,encoding="utf-8")
link=tmp_path/"link.pem"
link.symlink_to(real)
with pytest.raises(RuntimeError,match="TLS_FILE_INVALID"):
edge.validate_tls_file(link,key=False)
def test_systemd_unit_is_disabled_by_owner_approval_condition_and_loopback_default():
unit=(ROOT/"deploy/systemd/aios-gpt-https-edge.service").read_text()
assert "ConditionPathExists=/etc/aios-gitops/gpt-https-edge.approved" in unit
assert "AIOS_GPT_EDGE_HOST=127.0.0.1" in unit
assert "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0" in unit
assert "AIOS_GPT_EDGE_PORT=18443" in unit
assert "LoadCredential=tls_cert:" in unit
assert "LoadCredential=tls_key:" in unit
assert "InaccessiblePaths=/data/aios-gitops" in unit
assert "InaccessiblePaths=" in unit and "/var/lib/aios-agent-task" in unit
assert "NoNewPrivileges=yes" in unit
assert "ProtectSystem=strict" in unit
def test_systemd_unit_does_not_reference_internal_control_credentials():
unit=(ROOT/"deploy/systemd/aios-gpt-https-edge.service").read_text()
assert "agent_task_control" not in unit
assert "gitea" not in unit.lower()
assert "18892" not in unit
assert "18940" not in unit
def test_edge_source_does_not_proxy_internal_bridge_or_task_api():
source=(ROOT/"active-runtime/app/aios_gpt_https_edge.py").read_text()
assert 'UPSTREAM = "http://127.0.0.1:18945"' in source
assert "18892" not in source
assert "18940" not in source
assert "/v2/governed-task" not in source
assert "/v1/tasks/" not in source
def test_docs_keep_public_activation_separately_approved():
doc=(ROOT/"docs/superpowers/specs/2026-08-19-gpt-https-edge.md").read_text()
assert "explicit owner approval" in doc.lower()
assert "PUBLIC_NETWORK_EXPOSURE=NONE" in doc
assert "PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=NOT_VERIFIED" in doc
+342
View File
@@ -0,0 +1,342 @@
from __future__ import annotations
import json
from pathlib import Path
import sqlite3
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
from aios_gpt_online_task_ingress import (
IngressError,
IngressService,
Store,
READONLY_CAPABILITIES,
branch_for,
normalized_request,
openapi_document,
task_id_for,
validated_bind_host,
)
MAIN="3244f0f17f8845592ac94d585f5ebd65693c9f29"
JOB="job-0123456789abcdef0123456789abcdef"
def request(request_id="external-request-0001", **overrides):
value={
"schema":"aios.gpt-online-task-ingress.request.v1",
"request_id":request_id,
"goal":"Inspect the repository for durable worker state.",
"instructions":"Read only. Return a concise source analysis.",
"success_criteria":[
{"id":"analysis","description":"Return a source-backed analysis."}
],
"constraints":{
"max_files":50,
"focus_paths":["active-runtime/app","tests"],
"notes":"Do not mutate repository state.",
},
}
value.update(overrides)
return value
class FakeUpstream:
def __init__(self):
self.created=[]
self.projected=[]
self.tasks={}
self.snapshot={"authoritative_head_oid":MAIN}
def control_snapshot(self):
return dict(self.snapshot)
def create_governed_task(self, body):
self.created.append(dict(body))
task_id=body["task_id"]
job_id=JOB
self.tasks[task_id]={
"task_id":task_id,
"status":"PROJECTED",
"attempt":0,
"metadata":{
"execution":{
"task_id":task_id,
"job_id":job_id,
"base_oid":body["base_oid"],
"action":"source_capture",
"capability_scope":list(READONLY_CAPABILITIES),
"governance":{"writer_lease":False},
}
},
}
return 201,{"task_id":task_id,"job_id":job_id,"state":"BOUND"}
def execution_envelope(self, body):
self.projected.append(dict(body))
task_id=body["task_id"]
return 200,{
"schema":"aios.online-agent.execution.v2",
"task_id":task_id,
"job_id":body["job_id"],
"repository":"aios/aios",
"base_oid":MAIN,
"action":"source_capture",
"capability_scope":list(READONLY_CAPABILITIES),
"governance":{"writer_lease":False},
}
def get_task(self, task_id):
value=self.tasks.get(task_id)
return (404,{}) if value is None else (200,dict(value))
def service(tmp_path):
upstream=FakeUpstream()
store=Store(tmp_path/"state.sqlite3")
return IngressService(upstream,store),upstream,store
def test_request_schema_is_narrow_and_rejects_capability_or_execution_control():
value=normalized_request(request())
assert value["goal"].startswith("Inspect")
for field in ("capability_scope","required_capabilities","action","adapter","repository","base_oid"):
bad=request()
bad[field]="evil"
with pytest.raises(IngressError,match="UNKNOWN_FIELDS"):
normalized_request(bad)
def test_constraints_are_bounded_and_path_traversal_is_denied():
bad=request(constraints={"focus_paths":["../etc/passwd"]})
with pytest.raises(IngressError,match="FOCUS_PATHS_INVALID"):
normalized_request(bad)
bad=request(constraints={"max_files":1000})
with pytest.raises(IngressError,match="MAX_FILES_INVALID"):
normalized_request(bad)
def test_non_loopback_bind_is_denied():
assert validated_bind_host("127.0.0.1")=="127.0.0.1"
with pytest.raises(RuntimeError,match="GPT_INGRESS_NON_LOOPBACK_BIND_DENY"):
validated_bind_host("0.0.0.0")
with pytest.raises(RuntimeError,match="GPT_INGRESS_NON_LOOPBACK_BIND_DENY"):
validated_bind_host("192.0.2.10")
def test_task_and_branch_ids_are_deterministic_and_bounded():
a=task_id_for("external-request-0001")
b=task_id_for("external-request-0001")
c=task_id_for("external-request-0002")
assert a==b
assert a!=c
assert a.startswith("gpt-readonly-")
branch=branch_for(a)
assert branch.startswith("agent/external-gpt-readonly/")
assert len(a)<180
def test_create_translates_to_canonical_readonly_governed_task(tmp_path):
svc,upstream,store=service(tmp_path)
response=svc.create(request())
assert response["state"]=="BOUND"
assert response["write_access"] is False
assert response["official_verification_required"] is True
assert response["capabilities"]==list(READONLY_CAPABILITIES)
assert len(upstream.created)==1
body=upstream.created[0]
assert body["repository"]=="aios/aios"
assert body["base_oid"]==MAIN
assert body["expected_branch_oid"]==MAIN
assert body["branch_mode"]=="CREATE_NEW"
assert body["risk"]=="low"
assert body["priority"]==0
assert body["max_attempts"]==3
assert body["require_online_task_projection"] is True
assert body["online_task"]["required_capabilities"]==list(READONLY_CAPABILITIES)
assert body["online_task"]["constraints"]["repository_mutation"]=="forbidden"
assert body["online_task"]["constraints"]["credential_access"]=="forbidden"
assert body["execution"]["action"]=="source_capture"
assert body["execution"]["adapter"]=="source_capture"
assert body["execution"]["capability_scope"]==list(READONLY_CAPABILITIES)
assert "writer_lease" not in body["execution"]
assert len(upstream.projected)==1
projection=upstream.projected[0]
assert projection["task_id"]==response["task_id"]
assert projection["job_id"]==JOB
assert projection["required_capabilities"]==list(READONLY_CAPABILITIES)
row=store.get("external-request-0001")
assert row is not None
assert row["state"]=="BOUND"
assert row["job_id"]==JOB
assert row["base_oid"]==MAIN
def test_exact_retry_is_idempotent_without_duplicate_bridge_call(tmp_path):
svc,upstream,_=service(tmp_path)
first=svc.create(request())
second=svc.create(request())
assert second==first
assert len(upstream.created)==1
assert len(upstream.projected)==1
def test_same_request_id_with_different_content_is_conflict(tmp_path):
svc,upstream,_=service(tmp_path)
svc.create(request())
with pytest.raises(IngressError,match="IDEMPOTENCY_CONFLICT"):
svc.create(request(goal="Different goal."))
assert len(upstream.created)==1
def test_pending_request_recovers_existing_canonical_task_after_crash(tmp_path):
svc,upstream,store=service(tmp_path)
raw=normalized_request(request())
task_id=task_id_for(raw["request_id"])
store.reserve(raw,task_id)
upstream.tasks[task_id]={
"task_id":task_id,
"status":"PROJECTED",
"attempt":0,
"metadata":{
"execution":{
"task_id":task_id,
"job_id":JOB,
"base_oid":MAIN,
"action":"source_capture",
"capability_scope":list(READONLY_CAPABILITIES),
"governance":{"writer_lease":False},
}
},
}
result=svc.create(request())
assert result["job_id"]==JOB
assert result["task_id"]==task_id
assert len(upstream.created)==0
row=store.get(raw["request_id"])
assert row["state"]=="BOUND"
def test_status_is_sanitized_and_never_proxies_raw_execution_or_result_text(tmp_path):
svc,upstream,store=service(tmp_path)
created=svc.create(request())
task_id=created["task_id"]
upstream.tasks[task_id]={
"task_id":task_id,
"status":"SUCCEEDED",
"attempt":1,
"claimed_by":"online-engineering-worker-cn-network-mkt-generic-20260815",
"metadata":{
"execution":{
"job_id":JOB,
"base_oid":MAIN,
"action":"source_capture",
"capability_scope":list(READONLY_CAPABILITIES),
"governance":{"writer_lease":False},
"secret_marker":"do-not-return",
}
},
"result":{
"status":"SUCCEEDED",
"classification":"NEEDS_REWORK_SOURCE_CAPTURE",
"execution_mode":"source_capture",
"output_text":"raw source text that must not be proxied",
"source_snapshot":{
"payload_sha256":"a"*64,
"file_count":42,
"files":["secret/path"],
},
},
"verification":{
"decision":"SUCCEEDED",
"classification":"GENERIC_ONLINE_TASK_AUTO_CONSUMPTION_VERIFIED",
"private":"do-not-return",
},
}
status=svc.status("external-request-0001")
assert status["state"]=="SUCCEEDED"
assert status["attempt"]==1
assert status["claimed"] is True
assert status["official_verification"]=="SUCCEEDED"
assert status["source_snapshot_sha256"]=="a"*64
assert status["source_snapshot_file_count"]==42
serialized=json.dumps(status)
assert "raw source text" not in serialized
assert "secret_marker" not in serialized
assert "secret/path" not in serialized
assert "do-not-return" not in serialized
def test_pending_status_can_recover_job_binding(tmp_path):
svc,upstream,store=service(tmp_path)
raw=normalized_request(request())
task_id=task_id_for(raw["request_id"])
store.reserve(raw,task_id)
upstream.tasks[task_id]={
"task_id":task_id,
"status":"PROJECTED",
"attempt":0,
"metadata":{
"execution":{
"job_id":JOB,
"base_oid":MAIN,
"action":"source_capture",
"capability_scope":list(READONLY_CAPABILITIES),
"governance":{"writer_lease":False},
}
},
}
value=svc.status(raw["request_id"])
assert value["job_id"]==JOB
assert store.get(raw["request_id"])["state"]=="BOUND"
def test_openapi_exposes_only_create_readonly_and_status_operations():
doc=openapi_document()
assert doc["openapi"]=="3.1.0"
paths=doc["paths"]
assert set(paths)=={
"/v1/read-only-tasks",
"/v1/read-only-tasks/{request_id}",
}
rendered=json.dumps(doc)
for forbidden in ("writer-lease","commit","verify","production","tag","baseline"):
assert forbidden not in rendered.lower()
def test_systemd_unit_is_loopback_credential_isolated_and_has_no_repository_access():
unit=(ROOT/"deploy/systemd/aios-gpt-online-task-ingress.service").read_text()
assert "DynamicUser=yes" in unit
assert "LoadCredential=gpt_ingress_token:" in unit
assert "LoadCredential=agent_task_control_token:" in unit
assert "AIOS_GPT_INGRESS_HOST=127.0.0.1" in unit
assert "AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
assert "AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940" in unit
assert "IPAddressDeny=any" in unit
assert "IPAddressAllow=127.0.0.0/8" in unit
assert "InaccessiblePaths=/data/aios-gitops /etc/aios-gitops /home" in unit
assert "ProtectSystem=strict" in unit
assert "NoNewPrivileges=yes" in unit
assert "MemoryDenyWriteExecute=yes" in unit
assert "0.0.0.0" not in unit
def test_ingress_sqlite_contains_no_request_body_or_credentials(tmp_path):
svc,upstream,store=service(tmp_path)
svc.create(request(instructions="sensitive-looking but non-secret user text"))
con=sqlite3.connect(store.path)
columns=[row[1] for row in con.execute("pragma table_info(external_task_requests)")]
assert "request_json" not in columns
assert "goal" not in columns
assert "instructions" not in columns
assert "token" not in " ".join(columns).lower()
+193 -1
View File
@@ -154,7 +154,7 @@ def seed(db_path, *, legacy=False, tiny_budget=False):
store.close()
def make_external_request(projected_task, *, force_fail=False):
def make_external_request(projected_task, *, force_fail=False, attempt=1):
execution=projected_task["metadata"]["execution"]
payload=dict(execution["runtime_input"])
if force_fail:
@@ -165,6 +165,7 @@ def make_external_request(projected_task, *, force_fail=False):
"request_id":"external-model-governance-0123456789",
"runtime":"hermes",
"task_id":projected_task["task_id"],
"attempt":attempt,
"execution_digest":execution["execution_digest"],
"payload":payload,
}
@@ -360,3 +361,194 @@ def test_governance_required_profile_needs_provider_allowlist_and_output_limit(t
match="EXTERNAL_AGENT_MODEL_PROVIDER_IDS_REQUIRED",
):
runtime.RuntimeProfile.load("hermes",root)
def _openclaw_context_helper(tmp_path, *, input_tokens=7770, output_tokens=13):
path=tmp_path/"openclaw_context_helper.py"
path.write_text(
"import json,sys\n"
"model=sys.argv[1]\n"
"provider=sys.argv[2]\n"
"json.load(sys.stdin)\n"
f"print(json.dumps({{'answer':'ok','model':model,'provider':provider,"
f"'meta':{{'agentMeta':{{'usage':{{'input':{input_tokens},'output':{output_tokens},"
f"'total':{input_tokens+output_tokens}}},'costUsd':0}}}}}},sort_keys=True))\n",
encoding="utf-8",
)
return path
def _openclaw_floor_profile(tmp_path, helper_path, *, floor=16384):
root=tmp_path/"openclaw-profiles"
root.mkdir(exist_ok=True)
value={
"schema":"aios.external-agent-runtime.profile.v1",
"runtime":"openclaw",
"argv":["/usr/bin/python3",str(helper_path),"{model_id}","{provider_id}"],
"input_mode":"stdin_json",
"output_mode":"stdout_json",
"timeout_seconds":30,
"max_output_bytes":65536,
"environment":{},
"credential_env":{},
"model_governance":{
"required":True,
"provider_ids":["openai"],
"model_capabilities":["text"],
"max_output_tokens":256,
"input_reservation_floor_tokens":floor,
},
}
(root/"openclaw.json").write_text(json.dumps(value),encoding="utf-8")
return root
def _seed_openclaw_floor_db(db_path):
store=ModelGovernanceStore(db_path,now_fn=lambda:1_800_000_000.0)
store.register_provider(
ProviderRecord(
"openai",
"runtime-profile:openclaw",
max_failures=2,
cooldown_seconds=60,
)
)
store.register_model(
ModelRecord(
"openai/deepseek-v4-flash",
"openai",
"v1",
("agent.runtime.openclaw","text"),
input_cost_micros_per_1k=0,
output_cost_micros_per_1k=0,
max_input_tokens=32768,
max_output_tokens=256,
evaluation_sha256="c"*64,
priority=10,
)
)
store.set_scope_policy(
ScopePolicy(
"tenant",
"model-governance-live",
"staging",
allowed_models=("openai/deepseek-v4-flash",),
required_capabilities=("text",),
max_input_tokens_per_request=32768,
max_output_tokens_per_request=256,
daily_token_budget=250000,
daily_cost_budget_micros=0,
fallback_limit=1,
)
)
store.close()
def _openclaw_floor_request(*, attempt=2):
return {
"schema":"aios.external-agent-runtime.request.v1",
"request_id":f"external-openclaw-floor-attempt-{attempt:02d}",
"runtime":"openclaw",
"task_id":"r42-model-governance-openclaw-live-20260818",
"attempt":attempt,
"execution_digest":"d"*64,
"payload":{
"goal":"Return AIOS_MODEL_GOVERNANCE_OPENCLAW_OK.",
"instructions":"No side effects.",
"success_criteria":[],
"constraints":{},
"scope":{
"scope_type":"tenant",
"tenant_id":"model-governance-live",
"project_id":"staging",
},
"trace_id":"trace-openclaw-model-governance-live",
},
}
def test_openclaw_profile_reservation_floor_covers_injected_context(
tmp_path,monkeypatch
):
helper_path=_openclaw_context_helper(tmp_path,input_tokens=7770,output_tokens=13)
profile_root=_openclaw_floor_profile(tmp_path,helper_path,floor=16384)
db=tmp_path/"openclaw-governance.sqlite3"
_seed_openclaw_floor_db(db)
monkeypatch.setattr(runtime,"MODEL_GOVERNANCE_ENABLED",True)
monkeypatch.setattr(runtime,"DEFAULT_MODEL_GOVERNANCE_DB",db)
request_path=tmp_path/"request.json"
result_path=tmp_path/"result.json"
request_path.write_text(json.dumps(_openclaw_floor_request()),encoding="utf-8")
runtime.execute_request(request_path,result_path,profile_root=profile_root)
result=json.loads(result_path.read_text())
mg=result["model_governance"]
assert mg["input_reservation_floor_tokens"]==16384
assert mg["prompt_estimated_input_tokens"] < 16384
assert mg["reserved_estimated_input_tokens"]==16384
assert mg["settlement"]["state"]=="SETTLED"
assert mg["settlement"]["usage_source"]=="runtime"
assert mg["settlement"]["input_tokens"]==7770
assert mg["settlement"]["output_tokens"]==13
store=ModelGovernanceStore(db)
row=store.conn.execute(
"""
select reserved_input_tokens,state,actual_input_tokens,actual_output_tokens
from model_governance_reservations
where request_id=?
""",
(result["request_id"]+":model",),
).fetchone()
assert int(row["reserved_input_tokens"])==16384
assert row["state"]=="SETTLED"
assert int(row["actual_input_tokens"])==7770
assert int(row["actual_output_tokens"])==13
store.close()
def test_openclaw_reservation_floor_still_fails_closed_above_floor(
tmp_path,monkeypatch
):
helper_path=_openclaw_context_helper(tmp_path,input_tokens=17000,output_tokens=13)
profile_root=_openclaw_floor_profile(tmp_path,helper_path,floor=16384)
db=tmp_path/"openclaw-governance.sqlite3"
_seed_openclaw_floor_db(db)
monkeypatch.setattr(runtime,"MODEL_GOVERNANCE_ENABLED",True)
monkeypatch.setattr(runtime,"DEFAULT_MODEL_GOVERNANCE_DB",db)
request_path=tmp_path/"request.json"
result_path=tmp_path/"result.json"
request_path.write_text(json.dumps(_openclaw_floor_request()),encoding="utf-8")
with pytest.raises(
runtime.ExternalAgentRuntimeError,
match="MODEL_USAGE_EXCEEDS_RESERVATION",
):
runtime.execute_request(request_path,result_path,profile_root=profile_root)
assert not result_path.exists()
def test_model_governance_profile_rejects_invalid_input_reservation_floor(tmp_path):
path=tmp_path/"helper.py"
path.write_text("print('ok')\n",encoding="utf-8")
root=tmp_path/"profiles"
root.mkdir()
(root/"openclaw.json").write_text(json.dumps({
"schema":"aios.external-agent-runtime.profile.v1",
"runtime":"openclaw",
"argv":["/usr/bin/python3",str(path),"{model_id}"],
"input_mode":"stdin_text",
"output_mode":"stdout_text",
"model_governance":{
"required":True,
"provider_ids":["openai"],
"max_output_tokens":256,
"input_reservation_floor_tokens":262145,
},
}))
with pytest.raises(
runtime.ExternalAgentRuntimeError,
match="EXTERNAL_AGENT_MODEL_INPUT_RESERVATION_FLOOR_INVALID",
):
runtime.RuntimeProfile.load("openclaw",root)
@@ -65,3 +65,35 @@ def test_staging_bootstrap_has_explicit_tenant_live_policy():
}
assert ("legacy",None,None) in keys
assert ("tenant","model-governance-live","staging") in keys
def test_staging_policy_input_limit_covers_root_owned_openclaw_reservation_floor():
config=json.loads(
(ROOT/"config/model-governance/model-governance.staging.json").read_text()
)
profile=json.loads(
(ROOT/"config/external-agent-runtime.d/openclaw.profile.json").read_text()
)
floor=int(profile["model_governance"]["input_reservation_floor_tokens"])
assert floor==16384
openclaw_model="openai/deepseek-v4-flash"
relevant=[
policy for policy in config["policies"]
if openclaw_model in policy["allowed_models"]
]
assert relevant
for policy in relevant:
assert int(policy["max_input_tokens_per_request"]) >= floor
def test_staging_openclaw_scope_limit_is_least_privilege_floor_not_model_ceiling():
config=json.loads(
(ROOT/"config/model-governance/model-governance.staging.json").read_text()
)
models={item["model_id"]:item for item in config["models"]}
model=models["openai/deepseek-v4-flash"]
assert int(model["max_input_tokens"])==32768
for policy in config["policies"]:
if "openai/deepseek-v4-flash" in policy["allowed_models"]:
assert int(policy["max_input_tokens_per_request"])==16384
assert int(policy["max_input_tokens_per_request"]) < int(model["max_input_tokens"])
@@ -0,0 +1,16 @@
from __future__ import annotations
import json
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
def test_openclaw_profile_uses_conservative_input_reservation_floor():
value=json.loads(
(ROOT/"config/external-agent-runtime.d/openclaw.profile.json").read_text()
)
mg=value["model_governance"]
assert mg["required"] is True
assert mg["input_reservation_floor_tokens"]==16384
assert mg["max_output_tokens"]==256