Compare commits

..
29 changed files with 5451 additions and 95 deletions
+410 -67
View File
@@ -16,6 +16,11 @@ import subprocess
import tempfile
import time
from typing import Any, Mapping, Sequence
from aios_model_governance import (
ModelGovernanceError,
ModelGovernanceStore,
RouteRequest,
)
REQUEST_SCHEMA = "aios.external-agent-runtime.request.v1"
RESULT_SCHEMA = "aios.external-agent-runtime.result.v1"
@@ -62,6 +67,15 @@ MAX_PROFILE_BYTES = 64 * 1024
MAX_CREDENTIAL_BYTES = 64 * 1024
DEFAULT_PROFILE_ROOT = Path("/etc/aios/external-agent-runtime.d")
DEFAULT_EXCHANGE_ROOT = Path("/var/lib/aios-external-agent-executor")
DEFAULT_MODEL_GOVERNANCE_DB = Path(
os.environ.get(
"AIOS_MODEL_GOVERNANCE_DB",
"/var/lib/aios-model-governance/state.sqlite3",
)
)
MODEL_GOVERNANCE_ENABLED = os.environ.get(
"AIOS_MODEL_GOVERNANCE_ENABLED", "0"
).strip().lower() in {"1", "true", "yes", "on"}
ALLOWED_EXECUTABLE_ROOTS = (
Path("/usr/bin"),
Path("/usr/local/bin"),
@@ -208,6 +222,172 @@ def _profile_credential_environment(
return tuple(sorted(result))
def _profile_model_governance(
value: object,
) -> tuple[bool, tuple[str, ...], tuple[str, ...], int, int]:
if value is None:
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",
"input_reservation_floor_tokens",
}
if set(value) - allowed:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_GOVERNANCE_PROFILE_FIELDS_INVALID")
required = bool(value.get("required", False))
providers = value.get("provider_ids") or []
capabilities = value.get("model_capabilities") or []
if (
not isinstance(providers, Sequence)
or isinstance(providers, (str, bytes))
or not all(isinstance(item, str) and item.strip() for item in providers)
):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_PROVIDER_IDS_INVALID")
if (
not isinstance(capabilities, Sequence)
or isinstance(capabilities, (str, bytes))
or not all(isinstance(item, str) and item.strip() for item in capabilities)
):
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:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_OUTPUT_LIMIT_REQUIRED")
return (
required,
tuple(sorted({item.strip() for item in providers})),
tuple(sorted({item.strip() for item in capabilities})),
max_output_tokens,
input_reservation_floor_tokens,
)
def _estimate_tokens(text: str) -> int:
# Conservative byte-based estimate; runtime-reported usage supersedes this at settlement.
data = text.encode("utf-8", "replace")
return max(1, (len(data) + 3) // 4)
def _extract_runtime_usage(
output_json: object,
) -> tuple[int | None, int | None, int | None]:
if not isinstance(output_json, Mapping):
return None, None, None
candidates: list[Mapping[str, Any]] = []
direct = output_json.get("usage")
if isinstance(direct, Mapping):
candidates.append(direct)
meta = output_json.get("meta")
if isinstance(meta, Mapping):
agent_meta = meta.get("agentMeta")
if isinstance(agent_meta, Mapping):
nested = agent_meta.get("usage")
if isinstance(nested, Mapping):
candidates.append(nested)
last_call = agent_meta.get("lastCallUsage")
if isinstance(last_call, Mapping):
candidates.append(last_call)
def number(mapping: Mapping[str, Any], *names: str) -> int | None:
for name in names:
value = mapping.get(name)
if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
return value
return None
input_tokens = None
output_tokens = None
cost_micros = None
for usage in candidates:
if input_tokens is None:
input_tokens = number(usage, "input_tokens", "prompt_tokens", "input")
if output_tokens is None:
output_tokens = number(
usage, "output_tokens", "completion_tokens", "output"
)
if cost_micros is None:
cost_micros = number(usage, "cost_micros", "total_cost_micros")
if cost_micros is None and isinstance(meta, Mapping):
agent_meta = meta.get("agentMeta")
if isinstance(agent_meta, Mapping):
cost_usd = agent_meta.get("costUsd")
if (
isinstance(cost_usd, (int, float))
and not isinstance(cost_usd, bool)
and cost_usd >= 0
):
cost_micros = int(round(float(cost_usd) * 1_000_000))
return input_tokens, output_tokens, cost_micros
def _scope_from_payload(payload: Mapping[str, Any]) -> tuple[str, str | None, str | None, str]:
scope = payload.get("scope")
if not isinstance(scope, Mapping):
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_SCOPE_MISSING")
scope_type = str(scope.get("scope_type") or "legacy")
tenant_id = scope.get("tenant_id")
project_id = scope.get("project_id")
trace_id = str(payload.get("trace_id") or "").strip()
if not trace_id:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_TRACE_ID_MISSING")
return (
scope_type,
str(tenant_id) if tenant_id not in (None, "") else None,
str(project_id) if project_id not in (None, "") else None,
trace_id,
)
def _model_replacements(decision: object | None) -> dict[str, str]:
if decision is None:
return {}
return {
"{provider_id}": str(getattr(decision, "provider_id")),
"{model_id}": str(getattr(decision, "model_id")),
"{model_version}": str(getattr(decision, "model_version")),
}
def _bind_model_values(
values: Sequence[tuple[str, str]],
decision: object | None,
) -> tuple[tuple[str, str], ...]:
replacements = _model_replacements(decision)
result: list[tuple[str, str]] = []
for key, value in values:
bound = value
for marker, replacement in replacements.items():
bound = bound.replace(marker, replacement)
result.append((key, bound))
return tuple(result)
def _runtime_model_binding_present(profile: "RuntimeProfile") -> bool:
markers = ("{model_id}", "{model_version}", "{provider_id}")
return any(marker in item for item in (*profile.argv, *(value for _, value in profile.environment)) for marker in markers)
@dataclass(frozen=True)
class RuntimeProfile:
runtime: str
@@ -218,6 +398,11 @@ class RuntimeProfile:
max_output_bytes: int
environment: tuple[tuple[str, str], ...]
credential_env: tuple[tuple[str, str, str, str], ...]
model_governance_required: bool
model_provider_ids: tuple[str, ...]
model_capabilities: tuple[str, ...]
model_max_output_tokens: int
model_input_reservation_floor_tokens: int
@classmethod
def load(
@@ -275,7 +460,14 @@ class RuntimeProfile:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_LIMIT_INVALID")
environment = _profile_environment(value.get("environment"))
credential_env = _profile_credential_environment(value.get("credential_env"))
return cls(
(
model_governance_required,
model_provider_ids,
model_capabilities,
model_max_output_tokens,
model_input_reservation_floor_tokens,
) = _profile_model_governance(value.get("model_governance"))
profile = cls(
runtime,
tuple(normalized),
input_mode,
@@ -284,7 +476,15 @@ class RuntimeProfile:
max_output,
environment,
credential_env,
model_governance_required,
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")
return profile
def validate_request(value: Mapping[str, Any]) -> dict[str, Any]:
@@ -295,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")
@@ -317,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),
}
@@ -351,11 +559,13 @@ def _argv_with_files(
input_path: Path,
output_path: Path,
prompt: str,
model_decision: object | None = None,
) -> list[str]:
replacements = {
"{input_json}": str(input_path),
"{output_json}": str(output_path),
"{prompt}": prompt,
**_model_replacements(model_decision),
}
result = []
for item in profile.argv:
@@ -471,7 +681,12 @@ def execute_request(
request = validate_request(json.loads(raw_request.decode("utf-8")))
except json.JSONDecodeError as exc:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_JSON_INVALID") from exc
profile = RuntimeProfile.load(request["runtime"], profile_root)
governance_store: ModelGovernanceStore | None = None
route_request: RouteRequest | None = None
decision = None
governance_attempts: list[dict[str, Any]] = []
with tempfile.TemporaryDirectory(prefix=f"aios-{request['runtime']}-") as temporary:
temp_root = Path(temporary)
@@ -480,78 +695,203 @@ def execute_request(
output_path = temp_root / "output.json"
input_path.write_bytes(canonical(request["payload"]))
prompt = render_prompt(request["payload"])
argv = _argv_with_files(profile, input_path, output_path, prompt)
stdin_data = None
if profile.input_mode == "stdin_json":
stdin_data = canonical(request["payload"]).decode("utf-8")
elif profile.input_mode == "stdin_text":
stdin_data = prompt
child_env = {
"PATH": "/usr/bin:/usr/local/bin:/bin",
"LC_ALL": "C.UTF-8",
"LANG": "C.UTF-8",
"HOME": str(temp_root / "home"),
"NO_COLOR": "1",
}
child_env.update(dict(profile.environment))
child_env.update(_credential_environment(profile))
started = time.monotonic()
try:
completed = subprocess.run(
argv,
input=stdin_data,
capture_output=True,
text=True,
check=False,
shell=False,
timeout=profile.timeout_seconds,
cwd=str(temp_root),
env=child_env,
if profile.model_governance_required:
if not MODEL_GOVERNANCE_ENABLED:
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,
)
except subprocess.TimeoutExpired as exc:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_TIMEOUT") from exc
if completed.returncode != 0:
raise ExternalAgentRuntimeError(
f"EXTERNAL_AGENT_PROCESS_FAILED:{completed.returncode}:"
+ _bounded_text(completed.stderr, 2048)
route_request = RouteRequest(
request_id=request["request_id"] + ":model",
task_id=request["task_id"],
trace_id=trace_id,
scope_type=scope_type,
tenant_id=tenant_id,
project_id=project_id,
required_capabilities=tuple(
sorted(
{
f"agent.runtime.{request['runtime']}",
*profile.model_capabilities,
}
)
),
estimated_input_tokens=reserved_estimated_input_tokens,
max_output_tokens=profile.model_max_output_tokens,
)
try:
decision = governance_store.route(route_request)
except ModelGovernanceError as exc:
raise ExternalAgentRuntimeError(exc.code) from exc
if profile.model_provider_ids and decision.provider_id not in profile.model_provider_ids:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_PROVIDER_PROFILE_DENY")
governance_attempts.append(
{
"request_id": decision.request_id,
"provider_id": decision.provider_id,
"model_id": decision.model_id,
"model_version": decision.model_version,
"fallback_index": decision.fallback_index,
}
)
output_text, output_json = _parse_output(profile, completed.stdout, output_path)
result: dict[str, Any] = {
"schema": RESULT_SCHEMA,
"request_id": request["request_id"],
"request_sha256": sha256(canonical(request)),
"runtime": request["runtime"],
"task_id": request["task_id"],
"execution_digest": request["execution_digest"],
"exit_code": completed.returncode,
"duration_ms": int((time.monotonic() - started) * 1000),
"output_text": output_text,
"output_sha256": sha256(output_text.encode("utf-8", "replace")),
"stderr_sha256": sha256(completed.stderr.encode("utf-8", "replace")),
"credential_material_exposed": False,
"control_plane_access_granted": False,
}
if output_json is not None:
result["output_json"] = output_json
result["result_sha256"] = sha256(canonical(result))
result_path.parent.mkdir(parents=True, exist_ok=True)
fd, name = tempfile.mkstemp(prefix=f".{result_path.name}.", dir=result_path.parent)
try:
os.fchmod(fd, 0o640)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(result, handle, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(name, result_path)
while True:
argv = _argv_with_files(
profile,
input_path,
output_path,
prompt,
decision,
)
stdin_data = None
if profile.input_mode == "stdin_json":
stdin_data = canonical(request["payload"]).decode("utf-8")
elif profile.input_mode == "stdin_text":
stdin_data = prompt
child_env = {
"PATH": "/usr/bin:/usr/local/bin:/bin",
"LC_ALL": "C.UTF-8",
"LANG": "C.UTF-8",
"HOME": str(temp_root / "home"),
"NO_COLOR": "1",
}
child_env.update(dict(_bind_model_values(profile.environment, decision)))
child_env.update(_credential_environment(profile))
started = time.monotonic()
failure_class = None
try:
completed = subprocess.run(
argv,
input=stdin_data,
capture_output=True,
text=True,
check=False,
shell=False,
timeout=profile.timeout_seconds,
cwd=str(temp_root),
env=child_env,
)
if completed.returncode != 0:
failure_class = f"process_exit_{completed.returncode}"
except subprocess.TimeoutExpired:
completed = None
failure_class = "timeout"
if failure_class is None and completed is not None:
break
if governance_store is None or decision is None or route_request is None:
if failure_class == "timeout":
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_TIMEOUT")
raise ExternalAgentRuntimeError(
f"EXTERNAL_AGENT_PROCESS_FAILED:{completed.returncode}:"
+ _bounded_text(completed.stderr, 2048)
)
try:
governance_store.record_failure(decision.request_id, failure_class)
decision = governance_store.fallback(route_request, decision.request_id)
except ModelGovernanceError as exc:
raise ExternalAgentRuntimeError(exc.code) from exc
if profile.model_provider_ids and decision.provider_id not in profile.model_provider_ids:
raise ExternalAgentRuntimeError("EXTERNAL_AGENT_MODEL_PROVIDER_PROFILE_DENY")
governance_attempts.append(
{
"request_id": decision.request_id,
"provider_id": decision.provider_id,
"model_id": decision.model_id,
"model_version": decision.model_version,
"fallback_index": decision.fallback_index,
"previous_failure_class": failure_class,
}
)
output_text, output_json = _parse_output(profile, completed.stdout, output_path)
model_governance: dict[str, Any] | None = None
if governance_store is not None and decision is not None:
input_tokens, output_tokens, actual_cost = _extract_runtime_usage(output_json)
try:
settlement = governance_store.record_success(
decision.request_id,
input_tokens=input_tokens,
output_tokens=output_tokens,
actual_cost_micros=actual_cost,
usage_source="runtime",
)
except ModelGovernanceError as exc:
raise ExternalAgentRuntimeError(exc.code) from exc
model_governance = {
"decision_id": decision.decision_id,
"request_id": decision.request_id,
"provider_id": decision.provider_id,
"model_id": decision.model_id,
"model_version": decision.model_version,
"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(),
}
result: dict[str, Any] = {
"schema": RESULT_SCHEMA,
"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": completed.returncode,
"duration_ms": int((time.monotonic() - started) * 1000),
"output_text": output_text,
"output_sha256": sha256(output_text.encode("utf-8", "replace")),
"stderr_sha256": sha256(completed.stderr.encode("utf-8", "replace")),
"credential_material_exposed": False,
"control_plane_access_granted": False,
}
if output_json is not None:
result["output_json"] = output_json
if model_governance is not None:
result["model_governance"] = model_governance
result["result_sha256"] = sha256(canonical(result))
result_path.parent.mkdir(parents=True, exist_ok=True)
fd, name = tempfile.mkstemp(prefix=f".{result_path.name}.", dir=result_path.parent)
try:
os.fchmod(fd, 0o640)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(
result,
handle,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(name, result_path)
finally:
if os.path.exists(name):
os.unlink(name)
finally:
if os.path.exists(name):
os.unlink(name)
if governance_store is not None:
governance_store.close()
class ExternalAgentExchange:
@@ -566,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,
@@ -576,6 +917,7 @@ class ExternalAgentExchange:
"request_id": request_id,
"runtime": runtime,
"task_id": task_id,
"attempt": attempt,
"execution_digest": execution_digest,
"payload": dict(payload),
}
@@ -651,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")
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,232 @@
#!/usr/bin/env python3
"""Root-owned Model Governance configuration reconciler.
The config contains provider/model/policy metadata only. Raw credentials are rejected by
aios_model_governance and never stored in the governance database.
"""
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
from typing import Any, Mapping
from aios_model_governance import (
ModelGovernanceError,
ModelGovernanceStore,
ModelRecord,
ProviderRecord,
ScopePolicy,
)
SCHEMA = "aios.model-governance.bootstrap.v1"
MAX_CONFIG_BYTES = 512 * 1024
def canonical(value: object) -> bytes:
return json.dumps(
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode("utf-8")
def fail(code: str) -> None:
raise SystemExit(code)
def _load(path: Path) -> dict[str, Any]:
if path.is_symlink() or not path.is_file() or path.stat().st_size > MAX_CONFIG_BYTES:
fail("MODEL_GOVERNANCE_CONFIG_INVALID")
try:
value = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
fail("MODEL_GOVERNANCE_CONFIG_INVALID")
if not isinstance(value, Mapping) or value.get("schema") != SCHEMA:
fail("MODEL_GOVERNANCE_CONFIG_SCHEMA_INVALID")
return dict(value)
def _provider_desired(value: Mapping[str, Any]) -> ProviderRecord:
return ProviderRecord(
provider_id=str(value.get("provider_id") or ""),
credential_ref=value.get("credential_ref"),
enabled=bool(value.get("enabled", True)),
failure_domain=str(value.get("failure_domain") or "default"),
max_failures=int(value.get("max_failures") or 3),
cooldown_seconds=int(value.get("cooldown_seconds") or 60),
)
def _model_desired(value: Mapping[str, Any]) -> ModelRecord:
return ModelRecord(
model_id=str(value.get("model_id") or ""),
provider_id=str(value.get("provider_id") or ""),
version=str(value.get("version") or ""),
capabilities=tuple(value.get("capabilities") or ()),
input_cost_micros_per_1k=int(value.get("input_cost_micros_per_1k") or 0),
output_cost_micros_per_1k=int(value.get("output_cost_micros_per_1k") or 0),
max_input_tokens=int(value.get("max_input_tokens") or 0),
max_output_tokens=int(value.get("max_output_tokens") or 0),
evaluation_state=str(value.get("evaluation_state") or "APPROVED"),
evaluation_sha256=str(value.get("evaluation_sha256") or ""),
enabled=bool(value.get("enabled", True)),
priority=int(value.get("priority") or 100),
)
def _policy_desired(value: Mapping[str, Any]) -> ScopePolicy:
return ScopePolicy(
scope_type=str(value.get("scope_type") or ""),
tenant_id=value.get("tenant_id"),
project_id=value.get("project_id"),
allowed_models=tuple(value.get("allowed_models") or ()),
denied_providers=tuple(value.get("denied_providers") or ()),
required_capabilities=tuple(value.get("required_capabilities") or ()),
pinned_model_id=value.get("pinned_model_id"),
pinned_model_version=value.get("pinned_model_version"),
max_input_tokens_per_request=int(value.get("max_input_tokens_per_request") or 0),
max_output_tokens_per_request=int(value.get("max_output_tokens_per_request") or 0),
daily_token_budget=int(value.get("daily_token_budget") or 0),
daily_cost_budget_micros=int(value.get("daily_cost_budget_micros") or 0),
fallback_limit=int(value.get("fallback_limit") or 0),
)
def _row_dict(row: object) -> dict[str, Any] | None:
return dict(row) if row is not None else None
def _provider_matches(row: Mapping[str, Any], desired: ProviderRecord) -> bool:
return (
row["credential_ref"] == desired.credential_ref
and bool(row["enabled"]) == desired.enabled
and row["failure_domain"] == desired.failure_domain
and int(row["max_failures"]) == desired.max_failures
and int(row["cooldown_seconds"]) == desired.cooldown_seconds
)
def _model_matches(row: Mapping[str, Any], desired: ModelRecord) -> bool:
return (
row["provider_id"] == desired.provider_id
and row["version"] == desired.version
and tuple(json.loads(row["capabilities_json"])) == tuple(sorted(set(desired.capabilities)))
and int(row["input_cost_micros_per_1k"]) == desired.input_cost_micros_per_1k
and int(row["output_cost_micros_per_1k"]) == desired.output_cost_micros_per_1k
and int(row["max_input_tokens"]) == desired.max_input_tokens
and int(row["max_output_tokens"]) == desired.max_output_tokens
and row["evaluation_state"] == desired.evaluation_state
and row["evaluation_sha256"] == desired.evaluation_sha256
and bool(row["enabled"]) == desired.enabled
and int(row["priority"]) == desired.priority
)
def _policy_matches(row: Mapping[str, Any], desired: ScopePolicy) -> bool:
return (
row["scope_type"] == desired.scope_type
and row["tenant_id"] == desired.tenant_id
and row["project_id"] == desired.project_id
and tuple(json.loads(row["allowed_models_json"])) == tuple(sorted(set(desired.allowed_models)))
and tuple(json.loads(row["denied_providers_json"])) == tuple(sorted(set(desired.denied_providers)))
and tuple(json.loads(row["required_capabilities_json"])) == tuple(sorted(set(desired.required_capabilities)))
and row["pinned_model_id"] == desired.pinned_model_id
and row["pinned_model_version"] == desired.pinned_model_version
and int(row["max_input_tokens_per_request"]) == desired.max_input_tokens_per_request
and int(row["max_output_tokens_per_request"]) == desired.max_output_tokens_per_request
and int(row["daily_token_budget"]) == desired.daily_token_budget
and int(row["daily_cost_budget_micros"]) == desired.daily_cost_budget_micros
and int(row["fallback_limit"]) == desired.fallback_limit
)
def apply_config(db_path: Path, config_path: Path) -> dict[str, Any]:
config = _load(config_path)
config_sha = hashlib.sha256(canonical(config)).hexdigest()
providers = config.get("providers") or []
models = config.get("models") or []
policies = config.get("policies") or []
if not all(isinstance(value, list) for value in (providers, models, policies)):
fail("MODEL_GOVERNANCE_CONFIG_COLLECTION_INVALID")
store = ModelGovernanceStore(db_path)
changes = {"providers": 0, "models": 0, "policies": 0, "unchanged": 0}
try:
for item in providers:
if not isinstance(item, Mapping):
fail("MODEL_GOVERNANCE_PROVIDER_CONFIG_INVALID")
desired = _provider_desired(item)
row = store.conn.execute(
"SELECT * FROM model_governance_providers WHERE provider_id=?",
(desired.provider_id,),
).fetchone()
if row is not None and _provider_matches(row, desired):
changes["unchanged"] += 1
continue
expected = int(row["revision"]) if row is not None else 0
store.register_provider(desired, expected_revision=expected)
changes["providers"] += 1
for item in models:
if not isinstance(item, Mapping):
fail("MODEL_GOVERNANCE_MODEL_CONFIG_INVALID")
desired = _model_desired(item)
row = store.conn.execute(
"SELECT * FROM model_governance_models WHERE model_id=?",
(desired.model_id,),
).fetchone()
if row is not None and _model_matches(row, desired):
changes["unchanged"] += 1
continue
expected = int(row["revision"]) if row is not None else 0
store.register_model(desired, expected_revision=expected)
changes["models"] += 1
for item in policies:
if not isinstance(item, Mapping):
fail("MODEL_GOVERNANCE_POLICY_CONFIG_INVALID")
desired = _policy_desired(item)
row = store.conn.execute(
"SELECT * FROM model_governance_scope_policies WHERE scope_key=?",
(desired.scope_key,),
).fetchone()
if row is not None and _policy_matches(row, desired):
changes["unchanged"] += 1
continue
expected = int(row["revision"]) if row is not None else 0
store.set_scope_policy(desired, expected_revision=expected)
changes["policies"] += 1
if not store.verify_audit_chain():
fail("MODEL_GOVERNANCE_AUDIT_CHAIN_INVALID")
return {
"schema": "aios.model-governance.bootstrap-result.v1",
"database": str(db_path),
"config_sha256": config_sha,
"changes": changes,
"audit_chain_valid": True,
}
finally:
store.close()
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--db", required=True)
parser.add_argument("--config", required=True)
parser.add_argument("--output")
args = parser.parse_args()
try:
result = apply_config(Path(args.db), Path(args.config))
except ModelGovernanceError as exc:
print(json.dumps({"status": "FAILED", "error": exc.code}, sort_keys=True))
return 20
data = json.dumps(result, indent=2, sort_keys=True) + "\n"
if args.output:
Path(args.output).write_text(data, encoding="utf-8")
print(data, end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -179,7 +179,10 @@ def _agent_runtime_input(value: object) -> dict[str, Any]:
if not isinstance(value, Mapping):
_error("AGENT_RUNTIME_INPUT_REQUIRED")
raw = dict(value)
allowed = {"goal", "instructions", "success_criteria", "constraints"}
allowed = {
"goal", "instructions", "success_criteria", "constraints",
"scope", "trace_id",
}
if set(raw) - allowed:
_error("AGENT_RUNTIME_INPUT_FIELDS_INVALID")
goal = _string(raw.get("goal"), "AGENT_RUNTIME_GOAL_REQUIRED")
@@ -208,11 +211,34 @@ def _agent_runtime_input(value: object) -> dict[str, Any]:
lowered = str(key).lower()
if any(marker in lowered for marker in sensitive_markers):
_error("AGENT_RUNTIME_SECRET_FIELD_DENY")
scope = raw.get("scope", {})
if not isinstance(scope, Mapping):
_error("AGENT_RUNTIME_SCOPE_INVALID")
scope = {
"scope_type": str(scope.get("scope_type") or "legacy"),
"tenant_id": scope.get("tenant_id"),
"project_id": scope.get("project_id"),
}
if scope["scope_type"] not in {"legacy", "system", "tenant"}:
_error("AGENT_RUNTIME_SCOPE_INVALID")
if scope["scope_type"] == "tenant":
if not isinstance(scope["tenant_id"], str) or not scope["tenant_id"].strip():
_error("AGENT_RUNTIME_SCOPE_INVALID")
elif scope["tenant_id"] not in (None, "") or scope["project_id"] not in (None, ""):
_error("AGENT_RUNTIME_SCOPE_INVALID")
trace_id = str(raw.get("trace_id") or "").strip()
if not trace_id or len(trace_id) > 255:
_error("AGENT_RUNTIME_TRACE_ID_INVALID")
normalized = {
"goal": goal[:12000],
"instructions": instructions[:32000],
"success_criteria": normalized_criteria,
"constraints": dict(constraints),
"scope": scope,
"trace_id": trace_id,
}
if len(_canonical(normalized).encode("utf-8")) > MAX_AGENT_RUNTIME_INPUT_BYTES:
_error("AGENT_RUNTIME_INPUT_TOO_LARGE")
@@ -425,6 +451,12 @@ def project_online_task(
if adapter not in AGENT_RUNTIME_ADAPTERS:
_error("AGENT_RUNTIME_ADAPTER_INVALID")
runtime_input = execution_value.get("runtime_input")
tenant_id = request.get("tenant_id")
project_id = request.get("project_id")
declared_scope = str(
request.get("scope_type")
or ("tenant" if isinstance(tenant_id, str) and tenant_id.strip() else "legacy")
)
if runtime_input is None:
runtime_input = {
"goal": request.get("goal"),
@@ -432,6 +464,20 @@ def project_online_task(
"success_criteria": request.get("success_criteria") or [],
"constraints": request.get("constraints") or {},
}
elif not isinstance(runtime_input, Mapping):
_error("AGENT_RUNTIME_INPUT_REQUIRED")
runtime_input = dict(runtime_input)
# Scope/trace are authoritative Task identity, never caller-selected runtime hints.
runtime_input["scope"] = {
"scope_type": declared_scope,
"tenant_id": tenant_id,
"project_id": project_id,
}
runtime_input["trace_id"] = str(
request.get("trace_id")
or request.get("run_id")
or binding.task_id
)
execution_value["runtime_input"] = runtime_input
execution = project_execution_envelope(
binding,
+189 -24
View File
@@ -236,6 +236,64 @@ def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o640) -> None
os.unlink(temporary)
WORKER_LIVENESS_PATH = Path(
os.environ.get(
"AIOS_WORKER_LIVENESS_PATH",
str(STATE_ROOT / "worker-liveness.json"),
)
)
_LIVENESS_LOCK = threading.Lock()
_LIVENESS_SEQUENCE = 0
ENGINEERING_WRITE_CAPABILITIES = frozenset({
"governed.write.branch",
"governed.write.file",
})
def write_worker_liveness(
state: str,
*,
visible_count: int | None = None,
task_id: str | None = None,
stage: str | None = None,
error: str | None = None,
) -> None:
"""Write idle/busy liveness without touching Task/Job state."""
global _LIVENESS_SEQUENCE
with _LIVENESS_LOCK:
_LIVENESS_SEQUENCE += 1
epoch = time.time()
payload = {
"schema": "aios.online-agent.worker-liveness.v1",
"agent_id": AGENT_ID,
"agent_type": AGENT_TYPE,
"agent_version": AGENT_VERSION,
"pid": os.getpid(),
"state": state,
"sequence": _LIVENESS_SEQUENCE,
"last_seen_epoch": epoch,
"last_seen_at": datetime.fromtimestamp(epoch, timezone.utc).isoformat(),
"poll_seconds": POLL_SECONDS,
"heartbeat_seconds": HEARTBEAT_SECONDS,
"api_base": API_BASE,
"bridge_base": BRIDGE_BASE,
"visible_count": visible_count,
"task_id": task_id,
"stage": stage,
"error": error,
}
atomic_json(WORKER_LIVENESS_PATH, payload)
def source_capture_requires_deterministic_plan(envelope: "ExecutionEnvelope") -> bool:
"""Treat write-capable source_capture as a binding placeholder, never executable work."""
return (
envelope.action == "source_capture"
and bool(ENGINEERING_WRITE_CAPABILITIES.intersection(envelope.capability_scope))
)
def safe_relative_path(value: Any, *, field: str = "path") -> str:
if not isinstance(value, str) or not value or len(value) > 240:
raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
@@ -525,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
@@ -550,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"])
@@ -605,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))
@@ -674,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:
@@ -704,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",
@@ -715,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",
@@ -729,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:
@@ -746,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,
@@ -761,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,
@@ -890,6 +1007,11 @@ class LeaseHeartbeat:
self.failed = str(response.get("error") or status)
raise WorkerError("TASK_HEARTBEAT_FAILED", phase="heartbeat", details={"api_status": status, "error": self.failed})
audit("TASK_HEARTBEAT", task_id=self.task_id, stage=self.stage, api_status=status)
write_worker_liveness(
"BUSY",
task_id=self.task_id,
stage=self.stage,
)
def start(self) -> None:
self.thread.start()
@@ -909,6 +1031,11 @@ class LeaseHeartbeat:
audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, api_status=status, error=self.failed)
else:
audit("TASK_HEARTBEAT", task_id=self.task_id, stage=stage, api_status=status)
write_worker_liveness(
"BUSY",
task_id=self.task_id,
stage=stage,
)
except Exception as exc: # pragma: no cover - defensive runtime guard
self.failed = type(exc).__name__
audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, error=self.failed)
@@ -1286,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
@@ -1772,12 +1905,25 @@ class WorkerSupervisor:
if not isinstance(execution, Mapping):
raise WorkerError("EXECUTION_ENVELOPE_SCHEMA_VALIDATION", phase="validation")
persisted = ExecutionEnvelope.from_task(task, worker_capabilities=WORKER_CAPABILITIES)
# Deterministic tasks must re-resolve through the plan registry. A
# read-only source-capture task has no plan to replace its payload: replay
# the already validated durable envelope so source signals remain part of
# the canonical digest and the worker does not invent a different capture.
projection_input = execution if persisted.action in {"source_capture", "agent_runtime"} else None
# Read-only source_capture/agent_runtime tasks replay their persisted
# payload. A write-capable source_capture is only a Bridge-created
# engineering placeholder: never pass that stale payload back to the
# resolver, and never claim it until a deterministic plan exists.
pending_plan = source_capture_requires_deterministic_plan(persisted)
projection_input = (
execution
if persisted.action in {"source_capture", "agent_runtime"} and not pending_plan
else None
)
status, authoritative = self.bridge.project_execution(task, projection_input)
if pending_plan and (
status == 404
or authoritative.get("error") == "TASK_EXECUTION_PROJECTION_MISSING"
):
raise WorkerError(
"TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN",
phase="projection",
)
if not 200 <= status < 300:
raise WorkerError("TASK_PERSISTED_BINDING_UNVERIFIED", phase="projection", details={"api_status": status, "error": authoritative.get("error")})
canonical = ExecutionEnvelope.from_task(task, authoritative, WORKER_CAPABILITIES)
@@ -1806,6 +1952,11 @@ class WorkerSupervisor:
execution_digest=canonical.digest,
)
return canonical, "plan_registry_verified"
if pending_plan:
raise WorkerError(
"TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN",
phase="projection",
)
return persisted, "persisted_verified"
status, projected = self.bridge.project_execution(task)
if status == 404 or projected.get("error") == "TASK_EXECUTION_PROJECTION_MISSING":
@@ -2107,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",
@@ -2290,8 +2445,13 @@ class WorkerSupervisor:
def once(self) -> int:
try:
tasks = self.agent.poll()
write_worker_liveness("POLL_OK", visible_count=len(tasks))
audit("TASK_POLL", visible_count=len(tasks))
except WorkerError as exc:
write_worker_liveness(
"POLL_FAILED",
error=exc.code,
)
audit("TASK_POLL_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
return 10
dynamic_tasks: list[tuple[dict[str, Any], ExecutionEnvelope]] = []
@@ -2300,7 +2460,11 @@ class WorkerSupervisor:
envelope, source = self.resolve_envelope(task)
except WorkerError as exc:
event = "TASK_SKIPPED_ENVELOPE_INVALID"
if exc.code in {"TASK_SKIPPED_PROJECTION_MISSING", "TASK_SKIPPED_POLICY_DENY"}:
if exc.code in {
"TASK_SKIPPED_PROJECTION_MISSING",
"TASK_SKIPPED_POLICY_DENY",
"TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN",
}:
event = exc.code
elif exc.code == "CAPABILITY_MISMATCH_DENY":
event = "TASK_SKIPPED_CAPABILITY_MISMATCH"
@@ -2334,6 +2498,7 @@ class WorkerSupervisor:
ensure_roots()
recover_stale_workspaces()
checkpoint_store()
write_worker_liveness("STARTING")
audit("WORKER_STARTED", task_agnostic=True, api_base=API_BASE, bridge_base=BRIDGE_BASE, adapter_names=["isolated-deterministic", "source_capture", "codex", "hermes"], runtime_only_credentials=True, isolated_executor=True, durable_checkpoint=True)
while True:
self.once()
@@ -0,0 +1,462 @@
"""Self-healing availability guard for persistent Online Agent workers.
The guard never claims Tasks and never writes Task/Job/Git databases. It only:
- ensures required worker/executor systemd units are enabled + active;
- checks worker-owned liveness JSON written by the real poll/heartbeat loop;
- restarts an active worker only when its liveness is stale;
- honors an explicit maintenance inhibit file;
- writes a bounded status JSON under /run for observability.
Idle liveness is the successful poll loop itself. Busy liveness is refreshed by the
Task lease-heartbeat thread, so long-running Tasks are not mistaken for hung workers.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass
import argparse
import json
import os
from pathlib import Path
import shlex
import subprocess
import tempfile
import time
from typing import Any, Callable, Mapping, Sequence
SCHEMA = "aios.worker-availability.status.v1"
DEFAULT_WORKER_UNITS = (
"aios-online-agent-worker-20260815.service",
"aios-hermes-online-agent-worker.service",
"aios-openclaw-online-agent-worker.service",
)
DEFAULT_DEPENDENCY_UNITS = (
"aios-external-agent-executor.service",
)
DEFAULT_RUN_ROOT = Path("/run/aios-worker-availability")
DEFAULT_STATUS = DEFAULT_RUN_ROOT / "status.json"
DEFAULT_RECOVERY = DEFAULT_RUN_ROOT / "recovery.json"
DEFAULT_INHIBIT = DEFAULT_RUN_ROOT / "maintenance.lock"
DEFAULT_MAX_STALE_SECONDS = 90
DEFAULT_RESTART_COOLDOWN_SECONDS = 60
DEFAULT_STARTUP_WAIT_SECONDS = 10
ACCEPTED_ENABLED_STATES = frozenset({"enabled", "enabled-runtime", "static", "indirect", "generated"})
class AvailabilityGuardError(RuntimeError):
pass
@dataclass(frozen=True)
class UnitSnapshot:
unit: str
role: str
load_state: str
active_state: str
unit_file_state: str
enabled_ok: bool
liveness_path: str | None = None
liveness_state: str | None = None
last_seen_epoch: float | None = None
liveness_age_seconds: float | None = None
fresh: bool | None = None
action: str = "NONE"
detail: str | None = None
class Systemd:
def __init__(
self,
runner: Callable[..., subprocess.CompletedProcess[str]] | None = None,
) -> None:
self.runner = runner or subprocess.run
def _run(self, args: Sequence[str], *, check: bool = False) -> subprocess.CompletedProcess[str]:
result = self.runner(
["/usr/bin/systemctl", *args],
text=True,
capture_output=True,
check=False,
timeout=20,
)
if check and result.returncode != 0:
raise AvailabilityGuardError(
"SYSTEMD_COMMAND_FAILED:"
+ ":".join(args)
+ ":"
+ str(result.returncode)
)
return result
def property(self, unit: str, name: str) -> str:
result = self._run(["show", unit, f"--property={name}", "--value"])
if result.returncode != 0:
return ""
return result.stdout.strip()
def environment(self, unit: str) -> dict[str, str]:
text = self.property(unit, "Environment")
result: dict[str, str] = {}
for item in shlex.split(text):
if "=" not in item:
continue
key, value = item.split("=", 1)
result[key] = value
return result
def enable(self, unit: str) -> None:
self._run(["enable", unit], check=True)
def start(self, unit: str) -> None:
self._run(["start", unit], check=True)
def restart(self, unit: str) -> None:
self._run(["restart", unit], check=True)
def _atomic_json(path: Path, value: Mapping[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True, mode=0o755)
if path.parent.is_symlink() or path.is_symlink():
raise AvailabilityGuardError("STATUS_PATH_SYMLINK_REJECTED")
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
try:
os.fchmod(fd, 0o644)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(value, handle, ensure_ascii=False, sort_keys=True, indent=2)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
os.chmod(path, 0o644)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
def _load_json(path: Path) -> dict[str, Any] | None:
if path.is_symlink() or not path.is_file():
return None
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
return dict(value) if isinstance(value, Mapping) else None
def _state_root(environment: Mapping[str, str], unit: str) -> Path:
explicit = environment.get("AIOS_WORKER_STATE_ROOT")
if explicit:
return Path(explicit)
fallbacks = {
"aios-online-agent-worker-20260815.service": Path(
"/var/lib/aios-online-agent-worker-20260815"
),
"aios-hermes-online-agent-worker.service": Path(
"/var/lib/aios-online-agent-hermes"
),
"aios-openclaw-online-agent-worker.service": Path(
"/var/lib/aios-online-agent-openclaw"
),
}
if unit not in fallbacks:
raise AvailabilityGuardError("WORKER_STATE_ROOT_UNAVAILABLE:" + unit)
return fallbacks[unit]
def _liveness_path(environment: Mapping[str, str], unit: str) -> Path:
explicit = environment.get("AIOS_WORKER_LIVENESS_PATH")
if explicit:
return Path(explicit)
return _state_root(environment, unit) / "worker-liveness.json"
class AvailabilityGuard:
def __init__(
self,
*,
systemd: Systemd | None = None,
worker_units: Sequence[str] = DEFAULT_WORKER_UNITS,
dependency_units: Sequence[str] = DEFAULT_DEPENDENCY_UNITS,
status_path: Path = DEFAULT_STATUS,
recovery_path: Path = DEFAULT_RECOVERY,
inhibit_path: Path = DEFAULT_INHIBIT,
max_stale_seconds: int = DEFAULT_MAX_STALE_SECONDS,
restart_cooldown_seconds: int = DEFAULT_RESTART_COOLDOWN_SECONDS,
startup_wait_seconds: int = DEFAULT_STARTUP_WAIT_SECONDS,
now_fn: Callable[[], float] = time.time,
sleep_fn: Callable[[float], None] = time.sleep,
) -> None:
self.systemd = systemd or Systemd()
self.worker_units = tuple(worker_units)
self.dependency_units = tuple(dependency_units)
self.status_path = Path(status_path)
self.recovery_path = Path(recovery_path)
self.inhibit_path = Path(inhibit_path)
self.max_stale_seconds = int(max_stale_seconds)
self.restart_cooldown_seconds = int(restart_cooldown_seconds)
self.startup_wait_seconds = int(startup_wait_seconds)
self.now_fn = now_fn
self.sleep_fn = sleep_fn
if self.max_stale_seconds < 15:
raise AvailabilityGuardError("MAX_STALE_SECONDS_TOO_SMALL")
if self.restart_cooldown_seconds < 15:
raise AvailabilityGuardError("RESTART_COOLDOWN_TOO_SMALL")
if self.startup_wait_seconds < 0 or self.startup_wait_seconds > 60:
raise AvailabilityGuardError("STARTUP_WAIT_INVALID")
def _recovery_state(self) -> dict[str, float]:
value = _load_json(self.recovery_path) or {}
raw = value.get("last_restart_epoch")
if not isinstance(raw, Mapping):
return {}
result: dict[str, float] = {}
for unit, epoch in raw.items():
try:
result[str(unit)] = float(epoch)
except (TypeError, ValueError):
continue
return result
def _save_recovery_state(self, state: Mapping[str, float]) -> None:
_atomic_json(
self.recovery_path,
{
"schema": "aios.worker-availability.recovery.v1",
"last_restart_epoch": dict(state),
},
)
def _basic_snapshot(self, unit: str, role: str) -> tuple[str, str, str, bool]:
load = self.systemd.property(unit, "LoadState")
active = self.systemd.property(unit, "ActiveState")
enabled = self.systemd.property(unit, "UnitFileState")
return load, active, enabled, enabled in ACCEPTED_ENABLED_STATES
def _ensure_enabled_active(self, unit: str, role: str) -> tuple[str, bool]:
load, active, enabled, enabled_ok = self._basic_snapshot(unit, role)
if load != "loaded":
raise AvailabilityGuardError("REQUIRED_UNIT_NOT_LOADED:" + unit)
action = "NONE"
if not enabled_ok:
self.systemd.enable(unit)
action = "ENABLE"
if active != "active":
self.systemd.start(unit)
action = "START" if action == "NONE" else "ENABLE_START"
return action, active == "active"
def _wait_for_liveness(self, path: Path) -> dict[str, Any] | None:
deadline = self.now_fn() + self.startup_wait_seconds
while True:
value = _load_json(path)
if value is not None:
try:
epoch = float(value.get("last_seen_epoch"))
except (TypeError, ValueError):
epoch = 0.0
if epoch > 0 and self.now_fn() - epoch <= self.max_stale_seconds:
return value
if self.now_fn() >= deadline:
return value
self.sleep_fn(0.5)
def run(self, *, heal: bool) -> dict[str, Any]:
now = self.now_fn()
if self.inhibit_path.exists():
result = {
"schema": SCHEMA,
"status": "MAINTENANCE_INHIBITED",
"generated_at_epoch": now,
"maintenance_inhibit": str(self.inhibit_path),
"workers": [],
"dependencies": [],
"all_fresh": False,
"actions": [],
}
_atomic_json(self.status_path, result)
return result
actions: list[dict[str, str]] = []
dependencies: list[UnitSnapshot] = []
for unit in self.dependency_units:
load, active, enabled, enabled_ok = self._basic_snapshot(unit, "dependency")
action = "NONE"
if heal:
if load != "loaded":
raise AvailabilityGuardError("REQUIRED_UNIT_NOT_LOADED:" + unit)
if not enabled_ok:
self.systemd.enable(unit)
action = "ENABLE"
if active != "active":
self.systemd.start(unit)
action = "START" if action == "NONE" else "ENABLE_START"
if action != "NONE":
actions.append({"unit": unit, "action": action})
load, active, enabled, enabled_ok = self._basic_snapshot(unit, "dependency")
dependencies.append(
UnitSnapshot(
unit=unit,
role="dependency",
load_state=load,
active_state=active,
unit_file_state=enabled,
enabled_ok=enabled_ok,
fresh=(active == "active"),
action=action,
)
)
recovery = self._recovery_state()
workers: list[UnitSnapshot] = []
for unit in self.worker_units:
load, active, enabled, enabled_ok = self._basic_snapshot(unit, "worker")
environment = self.systemd.environment(unit) if load == "loaded" else {}
path = _liveness_path(environment, unit) if load == "loaded" else None
action = "NONE"
if heal:
if load != "loaded":
raise AvailabilityGuardError("REQUIRED_UNIT_NOT_LOADED:" + unit)
if not enabled_ok:
self.systemd.enable(unit)
action = "ENABLE"
if active != "active":
self.systemd.start(unit)
action = "START" if action == "NONE" else "ENABLE_START"
actions.append({"unit": unit, "action": action})
if path is not None:
self._wait_for_liveness(path)
load, active, enabled, enabled_ok = self._basic_snapshot(unit, "worker")
live = _load_json(path) if path is not None else None
last_seen: float | None = None
state: str | None = None
age: float | None = None
fresh = False
detail: str | None = None
if live is not None:
state = str(live.get("state") or "")
try:
last_seen = float(live.get("last_seen_epoch"))
except (TypeError, ValueError):
last_seen = None
if last_seen is not None:
age = max(0.0, now - last_seen)
fresh = age <= self.max_stale_seconds
if live.get("agent_id") in (None, ""):
fresh = False
detail = "LIVENESS_AGENT_ID_MISSING"
else:
detail = "LIVENESS_MISSING"
if heal and active == "active" and not fresh:
previous = float(recovery.get(unit, 0.0))
if now - previous >= self.restart_cooldown_seconds:
self.systemd.restart(unit)
recovery[unit] = now
action = "RESTART_STALE"
actions.append({"unit": unit, "action": action})
if path is not None:
live = self._wait_for_liveness(path)
load, active, enabled, enabled_ok = self._basic_snapshot(unit, "worker")
if live is not None:
state = str(live.get("state") or "")
try:
last_seen = float(live.get("last_seen_epoch"))
except (TypeError, ValueError):
last_seen = None
if last_seen is not None:
age = max(0.0, self.now_fn() - last_seen)
fresh = age <= self.max_stale_seconds
detail = None if fresh else "LIVENESS_STALE_AFTER_RESTART"
else:
detail = "RESTART_COOLDOWN_ACTIVE"
workers.append(
UnitSnapshot(
unit=unit,
role="worker",
load_state=load,
active_state=active,
unit_file_state=enabled,
enabled_ok=enabled_ok,
liveness_path=str(path) if path is not None else None,
liveness_state=state,
last_seen_epoch=last_seen,
liveness_age_seconds=age,
fresh=bool(active == "active" and enabled_ok and fresh),
action=action,
detail=detail,
)
)
if heal:
self._save_recovery_state(recovery)
all_fresh = (
all(item.active_state == "active" and item.enabled_ok and item.fresh for item in workers)
and all(item.active_state == "active" for item in dependencies)
)
result = {
"schema": SCHEMA,
"status": "HEALTHY" if all_fresh else "DEGRADED",
"generated_at_epoch": self.now_fn(),
"max_stale_seconds": self.max_stale_seconds,
"maintenance_inhibit": str(self.inhibit_path),
"workers": [asdict(item) for item in workers],
"dependencies": [asdict(item) for item in dependencies],
"all_fresh": all_fresh,
"actions": actions,
}
_atomic_json(self.status_path, result)
return result
def main() -> int:
parser = argparse.ArgumentParser()
mode = parser.add_mutually_exclusive_group(required=True)
mode.add_argument("--heal", action="store_true")
mode.add_argument("--check", action="store_true")
parser.add_argument("--status-file", default=str(DEFAULT_STATUS))
parser.add_argument("--recovery-file", default=str(DEFAULT_RECOVERY))
parser.add_argument("--inhibit-file", default=str(DEFAULT_INHIBIT))
parser.add_argument(
"--max-stale-seconds",
type=int,
default=int(
os.environ.get(
"AIOS_WORKER_AVAILABILITY_MAX_STALE_SECONDS",
str(DEFAULT_MAX_STALE_SECONDS),
)
),
)
parser.add_argument(
"--restart-cooldown-seconds",
type=int,
default=int(
os.environ.get(
"AIOS_WORKER_AVAILABILITY_RESTART_COOLDOWN_SECONDS",
str(DEFAULT_RESTART_COOLDOWN_SECONDS),
)
),
)
args = parser.parse_args()
guard = AvailabilityGuard(
status_path=Path(args.status_file),
recovery_path=Path(args.recovery_file),
inhibit_path=Path(args.inhibit_file),
max_stale_seconds=args.max_stale_seconds,
restart_cooldown_seconds=args.restart_cooldown_seconds,
)
try:
result = guard.run(heal=args.heal)
except AvailabilityGuardError as exc:
print(json.dumps({"status": "FAILED", "error": str(exc)}, sort_keys=True))
return 20
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
if args.check and not result.get("all_fresh"):
return 10
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -6,7 +6,7 @@
"--provider",
"oneapi",
"-m",
"deepseek-v4-flash",
"{model_id}",
"--no-restore-cwd"
],
"credential_env": {
@@ -21,6 +21,16 @@
},
"input_mode": "stdin_text",
"max_output_bytes": 131072,
"model_governance": {
"max_output_tokens": 512,
"model_capabilities": [
"text"
],
"provider_ids": [
"oneapi"
],
"required": true
},
"output_mode": "stdout_text",
"runtime": "hermes",
"schema": "aios.external-agent-runtime.profile.v1",
@@ -12,7 +12,7 @@
"--message",
"{prompt}",
"--model",
"openai/deepseek-v4-flash",
"{model_id}",
"--timeout",
"180"
],
@@ -30,6 +30,17 @@
},
"input_mode": "stdin_text",
"max_output_bytes": 131072,
"model_governance": {
"input_reservation_floor_tokens": 16384,
"max_output_tokens": 256,
"model_capabilities": [
"text"
],
"provider_ids": [
"openai"
],
"required": true
},
"output_mode": "stdout_json",
"runtime": "openclaw",
"schema": "aios.external-agent-runtime.profile.v1",
@@ -0,0 +1,95 @@
{
"models": [
{
"capabilities": [
"agent.runtime.hermes",
"text"
],
"enabled": true,
"evaluation_sha256": "43ae0db1640daeb6d33e2787769732844f366d36070065d3a2239905bd9a823b",
"evaluation_state": "APPROVED",
"input_cost_micros_per_1k": 0,
"max_input_tokens": 32768,
"max_output_tokens": 512,
"model_id": "deepseek-v4-flash",
"output_cost_micros_per_1k": 0,
"priority": 10,
"provider_id": "oneapi",
"version": "profile-sha:54615e533bfdb6a2e8b870ae74bb8cd7fc45177495bcf6e8c45a50837cefb85a"
},
{
"capabilities": [
"agent.runtime.openclaw",
"text"
],
"enabled": true,
"evaluation_sha256": "7b1c89634eb9c0e7cb5a3d75323d904ebbd91b3ae5494ce938bd8316a4cf5b12",
"evaluation_state": "APPROVED",
"input_cost_micros_per_1k": 0,
"max_input_tokens": 32768,
"max_output_tokens": 256,
"model_id": "openai/deepseek-v4-flash",
"output_cost_micros_per_1k": 0,
"priority": 10,
"provider_id": "openai",
"version": "profile-sha:2a31c480226aa78d44ed9692bddcedc99c3aa03c23c0dd5f031a5cbbb8bbd7ba"
}
],
"policies": [
{
"allowed_models": [
"deepseek-v4-flash",
"openai/deepseek-v4-flash"
],
"daily_cost_budget_micros": 0,
"daily_token_budget": 250000,
"denied_providers": [],
"fallback_limit": 1,
"max_input_tokens_per_request": 16384,
"max_output_tokens_per_request": 512,
"project_id": null,
"required_capabilities": [
"text"
],
"scope_type": "legacy",
"tenant_id": null
},
{
"allowed_models": [
"deepseek-v4-flash",
"openai/deepseek-v4-flash"
],
"daily_cost_budget_micros": 0,
"daily_token_budget": 250000,
"denied_providers": [],
"fallback_limit": 1,
"max_input_tokens_per_request": 16384,
"max_output_tokens_per_request": 512,
"project_id": "staging",
"required_capabilities": [
"text"
],
"scope_type": "tenant",
"tenant_id": "model-governance-live"
}
],
"providers": [
{
"cooldown_seconds": 60,
"credential_ref": "runtime-profile:hermes",
"enabled": true,
"failure_domain": "oneapi",
"max_failures": 2,
"provider_id": "oneapi"
},
{
"cooldown_seconds": 60,
"credential_ref": "runtime-profile:openclaw",
"enabled": true,
"failure_domain": "oneapi",
"max_failures": 2,
"provider_id": "openai"
}
],
"schema": "aios.model-governance.bootstrap.v1"
}
@@ -33,7 +33,10 @@ InaccessiblePaths=/data/agent/secrets
ReadOnlyPaths=/etc/aios/external-agent-runtime.d
ReadOnlyPaths=/opt/aios-hermes-agent
ReadOnlyPaths=/opt/aios/openclaw-engineering
Environment=AIOS_MODEL_GOVERNANCE_DB=/var/lib/aios-model-governance/state.sqlite3
Environment=AIOS_MODEL_GOVERNANCE_ENABLED=1
ReadWritePaths=/var/lib/aios-external-agent-executor
ReadWritePaths=/var/lib/aios-model-governance
UMask=0027
[Install]
@@ -0,0 +1,27 @@
[Unit]
Description=AIOS persistent Online Agent worker availability guard
After=local-fs.target
[Service]
Type=oneshot
User=root
Group=root
Environment=PYTHONUNBUFFERED=1
Environment=AIOS_WORKER_AVAILABILITY_MAX_STALE_SECONDS=90
Environment=AIOS_WORKER_AVAILABILITY_RESTART_COOLDOWN_SECONDS=60
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_worker_availability_guard.py --heal
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictSUIDSGID=true
RestrictRealtime=true
RestrictNamespaces=true
MemoryDenyWriteExecute=true
RestrictAddressFamilies=AF_UNIX
ReadWritePaths=/run/aios-worker-availability
UMask=0022
@@ -0,0 +1,13 @@
[Unit]
Description=Periodically ensure AIOS Online Agent workers remain available
[Timer]
OnBootSec=15s
OnUnitActiveSec=30s
AccuracySec=5s
RandomizedDelaySec=3s
Persistent=true
Unit=aios-worker-availability-guard.service
[Install]
WantedBy=timers.target
@@ -0,0 +1 @@
d /var/lib/aios-model-governance 0750 aios-engineering-executor aios-executor-exchange -
@@ -0,0 +1 @@
d /run/aios-worker-availability 0755 root root -
@@ -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,211 @@
# AIOS Model Governance Runtime
## Status and purpose
This component turns the previously `PARTIAL` Model Governance interface map into an enforceable
runtime policy for the already verified Hermes/OpenClaw Online Agent path.
It is intentionally a **single SQLite library**, not a new network service. The existing external
Agent executor calls the library immediately before launching a model-backed runtime and settles the
same reservation immediately after the result.
The authoritative execution chain remains:
```text
Online Task scope / trace
→ dedicated worker claim + heartbeat
→ external Agent executor
→ ModelGovernanceStore.route()
→ root-owned runtime profile binds provider/model/version
→ Hermes / OpenClaw process
→ ModelGovernanceStore.record_success() / record_failure()
→ durable Online Agent checkpoint/result
```
P0 scope/trace/credential, P1 checkpoint/retry, P2 fleet scheduling and the existing external Agent
credential boundary remain authoritative.
## Ten Model Governance contracts
### MODEL_PROVIDER_REGISTRY
`model_governance_providers` stores provider identity, failure domain, circuit policy and an opaque
credential reference. Raw credentials are rejected.
Allowed reference prefixes:
- `systemd-credential:`
- `runtime-profile:`
- `credential-broker:`
### MODEL_SELECTION_POLICY
`model_governance_scope_policies` constrains eligible models/providers/capabilities. Routing is
deterministic: explicit preferred models, then configured priority, then stable provider/model IDs.
### MODEL_VERSION_PINNING
A registry row always includes a model version. A scope can additionally pin a model ID + exact
version. Every route decision persists both model ID and version plus model/policy revisions.
### MODEL_BUDGET_ACCOUNTING
A route first reserves estimated input/output tokens and estimated cost inside `BEGIN IMMEDIATE`.
Settled usage and outstanding reservations both count against the daily scope budget, preventing two
concurrent requests from oversubscribing the same budget.
Runtime-reported usage is preferred. When telemetry is absent, the reserved upper bound is charged
rather than silently under-counting.
OpenClaw's verified result shape is supported directly:
```text
meta.agentMeta.usage.input
meta.agentMeta.usage.output
meta.agentMeta.costUsd
```
Hermes currently has no structured usage in the verified output, so it uses conservative
reservation settlement.
### MODEL_FALLBACK_POLICY
Provider failures are durable. Repeated failures open a circuit for the configured cooldown. A
fallback request excludes the failed provider and increments `fallback_index`; the scope policy
bounds the number of fallback attempts.
### MODEL_SCOPE_ISOLATION
The signed `agent_runtime` execution envelope now carries authoritative Task scope + trace identity.
The projection layer always overwrites `runtime_input.scope` and `runtime_input.trace_id` from the
Online Task, even if an execution request attempts to supply different values.
Tenant/project lookup may fall back only to the same tenant's policy. It never falls back to a
system or legacy policy. `system` and `legacy` require their own explicit policies, avoiding fake
tenant identities.
### MODEL_CREDENTIAL_BOUNDARY
Model governance stores credential references only. The root-owned external-runtime profile still
owns the actual provider credential projection. Task payloads cannot select executable, argv,
environment, provider credential, or credential path.
The external child process remains unable to reach the AIOS control plane.
### MODEL_TRACE_AUDIT
Every route, failure, provider/model/policy update and settlement appends to a SHA-256 hash-chained
audit table with task ID, trace ID and scope key. `verify_audit_chain()` detects tampering.
### MODEL_EVALUATION_GATE
A model cannot enter the registry unless its state is `APPROVED` and it has a 64-hex evaluation
evidence digest.
The staging bootstrap uses the already verified Hermes/OpenClaw E2E external-result SHA256 values as
evaluation evidence, rather than a configuration-file hash.
### MODEL_ROUTING_RESILIENCE
Route request IDs are idempotent and bound to a canonical request hash. Conflicting replay is
rejected. Provider health, reservations, usage settlement and policy/provider/model updates use
transaction/CAS semantics.
## Runtime profile contract
A root-owned Hermes/OpenClaw profile may declare:
```json
{
"model_governance": {
"required": true,
"provider_ids": ["provider-a"],
"model_capabilities": ["text"],
"max_output_tokens": 512
}
}
```
When `required=true`:
- `provider_ids` must be non-empty;
- `max_output_tokens` must be positive;
- argv/environment must contain at least one governance-controlled marker:
`{provider_id}`, `{model_id}`, or `{model_version}`;
- `AIOS_MODEL_GOVERNANCE_ENABLED` must be enabled;
- a matching scope policy/model/provider must exist;
- budget reservation must succeed before the child process is launched.
The task cannot provide these markers or their values.
## SQLite authority
Default database:
```text
/var/lib/aios-model-governance/state.sqlite3
```
The executor process already owns the only runtime mutation path. The directory is created by
tmpfiles and is the only new writable path added to that executor.
No direct Task/Job/Git database writes are introduced.
## Bootstrap/reconciliation
`aios_model_governance_admin.py` applies a root-owned JSON bootstrap idempotently. It compares desired
state to current rows and uses the current revision as the CAS input only when a change is actually
required.
The repository staging bootstrap is not a Production policy. It exists only to reproduce and
govern the already verified non-production Hermes/OpenClaw runtimes.
## Cost caveat
The current verified OneAPI-backed runtime evidence reports zero authoritative provider cost for
OpenClaw and no structured cost for Hermes. The staging bootstrap therefore does **not** invent
pricing and sets cost rates/budget to zero/unbounded.
Token budgets remain enforced. Real monetary budget enforcement becomes effective as soon as
authoritative per-model pricing is entered into the root-owned registry.
## OpenClaw JIT/WASM compatibility
The previous live acceptance proved that OpenClaw Node/WASM requires
`MemoryDenyWriteExecute=no`. Model Governance must preserve that already-evidenced runtime exception;
it must not further weaken `NoNewPrivileges`, `ProtectSystem=strict`, `ProtectHome`, loopback deny,
credential isolation or repository isolation.
## Failure/recovery invariants
- Budget denial occurs before provider execution.
- Missing scope policy fails closed.
- Missing/disabled governance fails closed when a profile requires it.
- Actual usage cannot exceed the reservation.
- Provider failure releases that reservation into `FAILED` state before fallback.
- Settlement is idempotent.
- Route replay is idempotent only for the exact canonical request.
- Audit corruption is detectable.
- Online Agent checkpoint/result retry remains unchanged and remains the outer recovery authority.
## Verification
Required unit/integration coverage includes:
- registry/evaluation/CAS;
- scope isolation and no system→tenant fallback;
- model/version pinning;
- atomic budget reservation;
- settlement/replay;
- missing telemetry conservative settlement;
- provider circuit/fallback;
- trace/audit tamper detection;
- authoritative scope/trace projection;
- task runtime-input scope override attack;
- root-owned provider/model binding;
- OpenClaw nested usage extraction;
- real Hermes/OpenClaw staging live tasks with Model Governance evidence.
The governed implementation is not complete without the unified all-modified-files bundle for the
candidate delta and the cumulative lineage bundle.
@@ -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,212 @@
# AIOS Worker Availability Runtime
## Priority
This runtime exists to make the already verified Online Agent execution path continuously usable
without repeated human intervention.
The system already proved that generic, Hermes and OpenClaw workers can claim and complete real
Online Tasks. The remaining availability defect was operational: evidence-freeze workflows left the
worker services stopped, while registry `ACTIVE` records remained stale.
The availability contract therefore separates:
```text
AUTO_CLAIM_MECHANISM=VERIFIED
CURRENT_WORKER_AVAILABILITY=<fresh runtime state>
```
A historical successful claim is not current liveness.
## Source of truth for idle liveness
An idle worker may have no Task lease heartbeat. Therefore Task heartbeat alone cannot prove that
the worker is continuously polling.
`aios_online_agent_worker.py` now writes:
```text
<AIOS_WORKER_STATE_ROOT>/worker-liveness.json
```
on:
- process start: `STARTING`;
- every successful poll: `POLL_OK`;
- poll failure: `POLL_FAILED`;
- every successful Task lease heartbeat while executing: `BUSY`.
The file contains:
- agent ID/type/version;
- process PID;
- monotonically increasing local sequence;
- `last_seen_epoch`;
- poll/heartbeat intervals;
- current Task/stage while busy.
The liveness file is worker-owned observability state only. It does not write or mutate Task, Job,
Git, WriterLease or registry databases.
## Self-healing guard
`aios-worker-availability-guard.timer` invokes a root-owned oneshot guard every 30 seconds.
The guard manages:
```text
aios-online-agent-worker-20260815.service
aios-hermes-online-agent-worker.service
aios-openclaw-online-agent-worker.service
```
and the required external runtime dependency:
```text
aios-external-agent-executor.service
```
For each worker:
1. missing systemd unit → fail closed;
2. disabled → `systemctl enable`;
3. inactive → `systemctl start`;
4. active but liveness older than 90 seconds → restart;
5. repeated stale restart within 60 seconds → do not restart again; report degraded.
Long-running Tasks are safe because the Task lease-heartbeat thread refreshes `BUSY` liveness.
The guard never claims a Task and never touches Task/Job/Git databases.
## Maintenance inhibit
Planned maintenance creates:
```text
/run/aios-worker-availability/maintenance.lock
```
While this file exists the guard performs no start/restart actions and reports:
```text
MAINTENANCE_INHIBITED
```
Maintenance tooling must remove the file when done. This prevents the self-healer from fighting
intentional service changes.
## Status
The guard writes:
```text
/run/aios-worker-availability/status.json
```
with current worker/dependency states, liveness ages and recovery actions.
A worker is currently available only when:
```text
systemd load_state=loaded
systemd active_state=active
unit enabled/static
worker-liveness.json present
last_seen_age <= 90s
```
Registry `ACTIVE` without fresh poll/busy liveness is not sufficient.
## Boot/restart behavior
The timer is:
```text
OnBootSec=15s
OnUnitActiveSec=30s
Persistent=true
```
Therefore reboot, manual stop left behind by another workflow, process crash, and hung poll loop are
all repaired without requiring a human to notice and intervene.
Normal worker units retain their own `Restart=always`; the availability guard handles the cases that
`Restart=always` cannot, especially a deliberate `systemctl stop` left after an evidence workflow.
## Acceptance
The live acceptance requires all of the following:
```text
guard timer active/enabled
external executor active/enabled
generic worker active/enabled + fresh liveness
Hermes worker active/enabled + fresh liveness
OpenClaw worker active/enabled + fresh liveness
```
Then a new read-only Online Task is projected through canonical Bridge `18892`.
Success requires:
```text
TASK_CREATED
→ generic worker auto claim
→ TASK_HEARTBEAT
→ RESULT_SUBMITTED
→ official TASK_VERIFIED(SUCCEEDED)
```
After the Task, the services must remain active and liveness must continue advancing. The acceptance
also stops a worker intentionally once and proves the guard starts it again with a new fresh
liveness sequence.
Only after these gates may current status be upgraded from:
```text
OVERALL=PARTIAL
```
to:
```text
PERSISTENT_WORKER_LIVENESS=VERIFIED
NEW_TASK_AUTO_CONSUMPTION=VERIFIED
WORKER_AVAILABILITY_RUNTIME=VERIFIED_CANDIDATE
```
Protected main and Production remain out of scope.
## Forced source-capture projection compatibility
The deployed canonical Bridge currently projects a default `source_capture` envelope even when a
governed binding request asks for binding-only behavior. This must not be allowed to make always-on
workers consume engineering Tasks before their deterministic plan is registered.
The worker therefore distinguishes a real read-only source capture from an engineering placeholder:
```text
source_capture
+ capability_scope contains governed.write.branch or governed.write.file
= PENDING_DETERMINISTIC_PLAN
```
For such a placeholder the worker:
1. re-resolves through the canonical plan registry **without replaying the stale source-capture
envelope**;
2. if no deterministic plan exists, emits `TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN` and does not
claim;
3. if the registered canonical envelope is deterministic and the governed binding still matches,
uses that deterministic envelope and claims normally.
A real read-only source-capture Task with only `git.read`/`git.test` still replays its exact persisted
payload and remains claimable.
This makes persistent workers safe with the currently deployed Bridge while preserving the same
Task/Job binding. No Task DB rewrite, cancellation, replacement Task, fake NEEDS_REWORK, or
pre-claim metadata mutation is required.
A later Bridge cleanup may implement true binding-only projection, but Worker Availability does not
depend on that cleanup.
@@ -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"
+325
View File
@@ -0,0 +1,325 @@
from __future__ import annotations
from pathlib import Path
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
from aios_model_governance import (
APPROVED,
ModelGovernanceError,
ModelGovernanceStore,
ModelRecord,
ProviderRecord,
RouteRequest,
ScopePolicy,
)
def setup_store(tmp_path, now=1_800_000_000.0):
store=ModelGovernanceStore(tmp_path/"model-governance.sqlite3",now_fn=lambda:now)
store.register_provider(
ProviderRecord(
"provider-a",
"runtime-profile:provider-a",
max_failures=1,
cooldown_seconds=120,
)
)
store.register_provider(
ProviderRecord(
"provider-b",
"runtime-profile:provider-b",
max_failures=2,
cooldown_seconds=60,
)
)
store.register_model(
ModelRecord(
"model-a",
"provider-a",
"2026-08-18",
("agent.runtime.hermes","text"),
input_cost_micros_per_1k=100,
output_cost_micros_per_1k=200,
max_input_tokens=10000,
max_output_tokens=4000,
evaluation_state=APPROVED,
evaluation_sha256="a"*64,
priority=10,
)
)
store.register_model(
ModelRecord(
"model-b",
"provider-b",
"2026-08-17",
("agent.runtime.hermes","text"),
input_cost_micros_per_1k=120,
output_cost_micros_per_1k=250,
max_input_tokens=10000,
max_output_tokens=4000,
evaluation_state=APPROVED,
evaluation_sha256="b"*64,
priority=20,
)
)
store.set_scope_policy(
ScopePolicy(
"tenant",
"tenant-a",
"project-a",
allowed_models=("model-a","model-b"),
required_capabilities=("text",),
max_input_tokens_per_request=5000,
max_output_tokens_per_request=2000,
daily_token_budget=12000,
daily_cost_budget_micros=4000,
fallback_limit=2,
)
)
return store
def req(request_id="request-001", *, exclude=()):
return RouteRequest(
request_id=request_id,
task_id="task-a",
trace_id="trace-a",
scope_type="tenant",
tenant_id="tenant-a",
project_id="project-a",
required_capabilities=("agent.runtime.hermes",),
estimated_input_tokens=1000,
max_output_tokens=500,
preferred_models=(),
exclude_providers=exclude,
)
def test_registry_rejects_raw_secret(tmp_path):
store=ModelGovernanceStore(tmp_path/"db.sqlite3")
with pytest.raises(ModelGovernanceError,match="MODEL_CREDENTIAL"):
store.register_provider(
ProviderRecord("provider-a","Bearer raw-secret-value")
)
def test_model_registration_requires_evaluation_gate(tmp_path):
store=ModelGovernanceStore(tmp_path/"db.sqlite3")
store.register_provider(ProviderRecord("provider-a"))
with pytest.raises(ModelGovernanceError,match="MODEL_EVALUATION_GATE_DENY"):
store.register_model(
ModelRecord(
"model-a","provider-a","v1",("text",),
evaluation_state="PENDING",evaluation_sha256="a"*64
)
)
def test_scope_policy_requires_version_with_pin(tmp_path):
store=ModelGovernanceStore(tmp_path/"db.sqlite3")
with pytest.raises(ModelGovernanceError,match="MODEL_POLICY_PIN_VERSION_REQUIRED"):
store.set_scope_policy(
ScopePolicy("tenant","t","p",pinned_model_id="model-a")
)
def test_route_is_deterministic_version_pinned_and_idempotent(tmp_path):
store=setup_store(tmp_path)
first=store.route(req())
second=store.route(req())
assert first==second
assert first.model_id=="model-a"
assert first.model_version=="2026-08-18"
assert first.provider_credential_ref=="runtime-profile:provider-a"
assert first.policy_revision==1
assert first.model_revision==1
snapshot=store.usage_snapshot("tenant","tenant-a","project-a")
assert snapshot["reserved_input_tokens"]==1000
assert snapshot["reserved_output_tokens"]==500
assert snapshot["reserved_cost_micros"]>0
def test_route_idempotency_payload_conflict_fails(tmp_path):
store=setup_store(tmp_path)
store.route(req())
changed=RouteRequest(
**{**req().__dict__,"estimated_input_tokens":2000}
)
with pytest.raises(ModelGovernanceError,match="MODEL_ROUTE_IDEMPOTENCY_CONFLICT"):
store.route(changed)
def test_budget_reservation_is_atomic(tmp_path):
store=setup_store(tmp_path)
store.set_scope_policy(
ScopePolicy(
"tenant","tenant-a","project-a",
allowed_models=("model-a",),
required_capabilities=("text",),
daily_token_budget=1600,
daily_cost_budget_micros=10000,
),
expected_revision=1,
)
store.route(req("request-1"))
with pytest.raises(ModelGovernanceError,match="MODEL_BUDGET_DENY"):
store.route(req("request-2"))
def test_usage_settlement_is_idempotent_and_releases_reservation(tmp_path):
store=setup_store(tmp_path)
decision=store.route(req())
first=store.record_success(
decision.request_id,
input_tokens=900,
output_tokens=400,
usage_source="runtime",
)
second=store.record_success(
decision.request_id,
input_tokens=900,
output_tokens=400,
usage_source="runtime",
)
assert first["idempotent"] is False
assert second["idempotent"] is True
snapshot=store.usage_snapshot("tenant","tenant-a","project-a")
assert snapshot["settled_input_tokens"]==900
assert snapshot["settled_output_tokens"]==400
assert snapshot["reserved_input_tokens"]==0
assert snapshot["reserved_output_tokens"]==0
def test_missing_runtime_usage_falls_back_to_reserved_upper_bound(tmp_path):
store=setup_store(tmp_path)
decision=store.route(req())
result=store.record_success(
decision.request_id,
input_tokens=None,
output_tokens=None,
)
assert result["usage_source"]=="reserved_upper_bound"
assert result["input_tokens"]==1000
assert result["output_tokens"]==500
def test_usage_cannot_exceed_reserved_budget(tmp_path):
store=setup_store(tmp_path)
decision=store.route(req())
with pytest.raises(ModelGovernanceError,match="MODEL_USAGE_EXCEEDS_RESERVATION"):
store.record_success(
decision.request_id,
input_tokens=1001,
output_tokens=500,
)
def test_provider_failure_opens_circuit_and_fallback_uses_other_provider(tmp_path):
store=setup_store(tmp_path)
original=req()
first=store.route(original)
assert first.provider_id=="provider-a"
failure=store.record_failure(first.request_id,"provider_timeout")
assert failure["open_until"]>0
fallback=store.fallback(original,first.request_id)
assert fallback.provider_id=="provider-b"
assert fallback.fallback_index==1
def test_circuit_excludes_provider_for_new_request(tmp_path):
store=setup_store(tmp_path)
first=store.route(req("request-first"))
store.record_failure(first.request_id,"provider_timeout")
next_decision=store.route(req("request-next"))
assert next_decision.provider_id=="provider-b"
def test_model_version_pin_is_enforced(tmp_path):
store=setup_store(tmp_path)
store.set_scope_policy(
ScopePolicy(
"tenant","tenant-a","project-a",
allowed_models=("model-a","model-b"),
pinned_model_id="model-b",
pinned_model_version="2026-08-17",
daily_token_budget=12000,
daily_cost_budget_micros=4000,
),
expected_revision=1,
)
decision=store.route(req())
assert decision.model_id=="model-b"
assert decision.model_version=="2026-08-17"
def test_scope_isolation_denies_missing_cross_scope_policy(tmp_path):
store=setup_store(tmp_path)
cross=RouteRequest(
request_id="request-cross",
task_id="task-a",
trace_id="trace-a",
scope_type="tenant",
tenant_id="tenant-b",
project_id="project-a",
required_capabilities=("agent.runtime.hermes",),
estimated_input_tokens=100,
max_output_tokens=50,
)
with pytest.raises(ModelGovernanceError,match="MODEL_SCOPE_POLICY_MISSING"):
store.route(cross)
def test_audit_chain_detects_tampering(tmp_path):
store=setup_store(tmp_path)
decision=store.route(req())
store.record_success(decision.request_id,input_tokens=100,output_tokens=50)
assert store.verify_audit_chain() is True
store.conn.execute(
"UPDATE model_governance_audit SET payload_json='{}' WHERE sequence=1"
)
assert store.verify_audit_chain() is False
def test_registry_and_policy_use_cas(tmp_path):
store=ModelGovernanceStore(tmp_path/"db.sqlite3")
store.register_provider(ProviderRecord("p"))
with pytest.raises(ModelGovernanceError,match="MODEL_PROVIDER_CAS_MISMATCH"):
store.register_provider(ProviderRecord("p"),expected_revision=7)
store.set_scope_policy(ScopePolicy("tenant","t","p"))
with pytest.raises(ModelGovernanceError,match="MODEL_POLICY_CAS_MISMATCH"):
store.set_scope_policy(
ScopePolicy("tenant","t","p"),
expected_revision=7,
)
def test_system_policy_never_authorizes_missing_tenant_policy(tmp_path):
store=ModelGovernanceStore(tmp_path/"db.sqlite3")
store.register_provider(ProviderRecord("provider-a"))
store.register_model(
ModelRecord(
"model-a","provider-a","v1",("text",),
evaluation_sha256="a"*64,
)
)
store.set_scope_policy(
ScopePolicy("system",None,None,allowed_models=("model-a",))
)
request=RouteRequest(
request_id="request-tenant-no-policy",
task_id="task-a",
trace_id="trace-a",
scope_type="tenant",
tenant_id="tenant-a",
project_id="project-a",
required_capabilities=("text",),
estimated_input_tokens=10,
max_output_tokens=10,
)
with pytest.raises(ModelGovernanceError,match="MODEL_SCOPE_POLICY_MISSING"):
store.route(request)
+82
View File
@@ -0,0 +1,82 @@
from __future__ import annotations
import json
from pathlib import Path
import sys
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
from aios_model_governance_admin import apply_config
from aios_model_governance import ModelGovernanceStore
def config():
return {
"schema":"aios.model-governance.bootstrap.v1",
"providers":[{
"provider_id":"provider-a",
"credential_ref":"runtime-profile:provider-a",
"max_failures":2,
"cooldown_seconds":60,
}],
"models":[{
"model_id":"model-a",
"provider_id":"provider-a",
"version":"v1",
"capabilities":["text"],
"input_cost_micros_per_1k":1,
"output_cost_micros_per_1k":2,
"max_input_tokens":1000,
"max_output_tokens":100,
"evaluation_state":"APPROVED",
"evaluation_sha256":"a"*64,
"priority":10,
}],
"policies":[{
"scope_type":"legacy",
"tenant_id":None,
"project_id":None,
"allowed_models":["model-a"],
"required_capabilities":["text"],
"daily_token_budget":10000,
"daily_cost_budget_micros":1000,
"fallback_limit":0,
}],
}
def test_admin_apply_is_idempotent(tmp_path):
cfg=tmp_path/"config.json"
cfg.write_text(json.dumps(config()),encoding="utf-8")
db=tmp_path/"db.sqlite3"
first=apply_config(db,cfg)
second=apply_config(db,cfg)
assert first["changes"]["providers"]==1
assert first["changes"]["models"]==1
assert first["changes"]["policies"]==1
assert second["changes"]["providers"]==0
assert second["changes"]["models"]==0
assert second["changes"]["policies"]==0
assert second["changes"]["unchanged"]==3
store=ModelGovernanceStore(db)
assert store.verify_audit_chain() is True
store.close()
def test_admin_config_change_uses_cas_revision(tmp_path):
value=config()
cfg=tmp_path/"config.json"
cfg.write_text(json.dumps(value),encoding="utf-8")
db=tmp_path/"db.sqlite3"
apply_config(db,cfg)
value["providers"][0]["cooldown_seconds"]=120
cfg.write_text(json.dumps(value),encoding="utf-8")
result=apply_config(db,cfg)
assert result["changes"]["providers"]==1
store=ModelGovernanceStore(db)
row=store.conn.execute(
"select revision,cooldown_seconds from model_governance_providers where provider_id='provider-a'"
).fetchone()
assert tuple(row)==(2,120)
store.close()
@@ -0,0 +1,554 @@
from __future__ import annotations
import json
import os
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))
from aios_model_governance import (
ModelGovernanceStore,
ModelRecord,
ProviderRecord,
ScopePolicy,
)
from aios_online_agent_execution import GovernedExecutionBinding, project_online_task
import aios_external_agent_runtime as runtime
BASE="a"*40
def binding(task_id):
return GovernedExecutionBinding.from_mapping({
"task_id":task_id,
"job_id":"job-"+task_id,
"repository":"aios/aios",
"base_oid":BASE,
"expected_branch_oid":BASE,
"target_branch":"agent/"+task_id,
"branch_mode":"CREATE_NEW",
"generation":1,
"idempotency_key":task_id+"-binding-v1",
})
def projected(task_id="task-model-governed", *, scope_type="tenant", tenant_id="tenant-a", project_id="project-a"):
request={
"task_id":task_id,
"goal":"Return the requested sentinel.",
"instructions":"No repository or control-plane side effects.",
"success_criteria":[{"id":"done","description":"Return a result."}],
"constraints":{"repository_mutation":"forbidden"},
"scope_type":scope_type,
"tenant_id":tenant_id,
"project_id":project_id,
"trace_id":"trace-model-a",
"required_capabilities":["agent.runtime.hermes"],
}
return project_online_task(
binding(task_id),
request,
{
"action":"agent_runtime",
"adapter":"hermes",
"capability_scope":["agent.runtime.hermes"],
},
)
def helper(tmp_path):
path=tmp_path/"model_runtime.py"
path.write_text(
"import json,sys\n"
"model=sys.argv[1]\n"
"provider=sys.argv[2]\n"
"payload=json.load(sys.stdin)\n"
"if model=='model-a' and payload.get('constraints',{}).get('force_fail_a'):\n"
" raise SystemExit(7)\n"
"print(json.dumps({'answer':'ok','model':model,'provider':provider,"
"'usage':{'input_tokens':20,'output_tokens':10,'cost_micros':2}},sort_keys=True))\n",
encoding="utf-8",
)
return path
def write_profile(tmp_path, helper_path, *, providers=("provider-a","provider-b")):
root=tmp_path/"profiles";root.mkdir(exist_ok=True)
value={
"schema":"aios.external-agent-runtime.profile.v1",
"runtime":"hermes",
"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":list(providers),
"model_capabilities":["text"],
"max_output_tokens":100,
},
}
(root/"hermes.json").write_text(json.dumps(value),encoding="utf-8")
return root
def seed(db_path, *, legacy=False, tiny_budget=False):
store=ModelGovernanceStore(db_path,now_fn=lambda:1_800_000_000.0)
for provider in ("provider-a","provider-b"):
store.register_provider(
ProviderRecord(
provider,
f"runtime-profile:{provider}",
max_failures=1 if provider=="provider-a" else 2,
cooldown_seconds=120,
)
)
store.register_model(
ModelRecord(
"model-a","provider-a","v1",
("agent.runtime.hermes","text"),
input_cost_micros_per_1k=10,
output_cost_micros_per_1k=20,
max_input_tokens=5000,max_output_tokens=1000,
evaluation_sha256="a"*64,priority=1,
)
)
store.register_model(
ModelRecord(
"model-b","provider-b","v2",
("agent.runtime.hermes","text"),
input_cost_micros_per_1k=12,
output_cost_micros_per_1k=22,
max_input_tokens=5000,max_output_tokens=1000,
evaluation_sha256="b"*64,priority=2,
)
)
if legacy:
policy=ScopePolicy(
"legacy",None,None,
allowed_models=("model-a","model-b"),
required_capabilities=("text",),
daily_token_budget=10000,
daily_cost_budget_micros=10000,
fallback_limit=2,
)
else:
policy=ScopePolicy(
"tenant","tenant-a","project-a",
allowed_models=("model-a","model-b"),
required_capabilities=("text",),
daily_token_budget=1 if tiny_budget else 10000,
daily_cost_budget_micros=10000,
fallback_limit=2,
)
store.set_scope_policy(policy)
store.close()
def make_external_request(projected_task, *, force_fail=False, attempt=1):
execution=projected_task["metadata"]["execution"]
payload=dict(execution["runtime_input"])
if force_fail:
payload["constraints"]=dict(payload["constraints"])
payload["constraints"]["force_fail_a"]=True
return {
"schema":"aios.external-agent-runtime.request.v1",
"request_id":"external-model-governance-0123456789",
"runtime":"hermes",
"task_id":projected_task["task_id"],
"attempt":attempt,
"execution_digest":execution["execution_digest"],
"payload":payload,
}
@pytest.fixture(autouse=True)
def clean_env(monkeypatch):
for key in list(os.environ):
upper=key.upper()
if any(marker in upper for marker in ("TOKEN","SECRET","PASSWORD","CREDENTIAL","AUTHORIZATION","PRIVATE_KEY","API_KEY")):
monkeypatch.delenv(key,raising=False)
monkeypatch.setenv("CREDENTIALS_DIRECTORY","/definitely-not-present-model-governance")
def test_execution_projection_binds_trusted_scope_and_trace():
task=projected()
runtime_input=task["metadata"]["execution"]["runtime_input"]
assert runtime_input["scope"]=={
"scope_type":"tenant",
"tenant_id":"tenant-a",
"project_id":"project-a",
}
assert runtime_input["trace_id"]=="trace-model-a"
def test_legacy_scope_remains_explicit_not_fake_tenant():
task=projected(scope_type="legacy",tenant_id=None,project_id=None)
scope=task["metadata"]["execution"]["runtime_input"]["scope"]
assert scope=={"scope_type":"legacy","tenant_id":None,"project_id":None}
def test_profile_requires_model_binding_when_governance_required(tmp_path):
path=tmp_path/"helper.py";path.write_text("print('ok')\n")
root=tmp_path/"profiles";root.mkdir()
(root/"hermes.json").write_text(json.dumps({
"schema":"aios.external-agent-runtime.profile.v1",
"runtime":"hermes",
"argv":["/usr/bin/python3",str(path)],
"input_mode":"stdin_text",
"output_mode":"stdout_text",
"model_governance":{"required":True,"provider_ids":["provider-a"],"max_output_tokens":100},
}))
with pytest.raises(runtime.ExternalAgentRuntimeError,match="EXTERNAL_AGENT_MODEL_BINDING_REQUIRED"):
runtime.RuntimeProfile.load("hermes",root)
def test_governed_runtime_selects_model_and_settles_usage(tmp_path,monkeypatch):
helper_path=helper(tmp_path)
profile_root=write_profile(tmp_path,helper_path)
db=tmp_path/"governance.sqlite3";seed(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(make_external_request(projected())),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["model_id"]=="model-a"
assert mg["model_version"]=="v1"
assert mg["provider_id"]=="provider-a"
assert mg["settlement"]["usage_source"]=="runtime"
assert mg["settlement"]["input_tokens"]==20
assert mg["settlement"]["output_tokens"]==10
assert mg["audit_chain_valid"] is True
assert result["output_json"]["model"]=="model-a"
def test_budget_denies_before_runtime_process(tmp_path,monkeypatch):
helper_path=helper(tmp_path)
profile_root=write_profile(tmp_path,helper_path)
db=tmp_path/"governance.sqlite3";seed(db,tiny_budget=True)
monkeypatch.setattr(runtime,"MODEL_GOVERNANCE_ENABLED",True)
monkeypatch.setattr(runtime,"DEFAULT_MODEL_GOVERNANCE_DB",db)
req=tmp_path/"request.json";res=tmp_path/"result.json"
req.write_text(json.dumps(make_external_request(projected())),encoding="utf-8")
with pytest.raises(runtime.ExternalAgentRuntimeError,match="MODEL_BUDGET_DENY"):
runtime.execute_request(req,res,profile_root=profile_root)
assert not res.exists()
def test_provider_failure_routes_to_fallback_provider(tmp_path,monkeypatch):
helper_path=helper(tmp_path)
profile_root=write_profile(tmp_path,helper_path)
db=tmp_path/"governance.sqlite3";seed(db)
monkeypatch.setattr(runtime,"MODEL_GOVERNANCE_ENABLED",True)
monkeypatch.setattr(runtime,"DEFAULT_MODEL_GOVERNANCE_DB",db)
req=tmp_path/"request.json";res=tmp_path/"result.json"
req.write_text(json.dumps(make_external_request(projected(),force_fail=True)),encoding="utf-8")
runtime.execute_request(req,res,profile_root=profile_root)
result=json.loads(res.read_text())
mg=result["model_governance"]
assert len(mg["attempts"])==2
assert mg["attempts"][0]["provider_id"]=="provider-a"
assert mg["attempts"][1]["provider_id"]=="provider-b"
assert mg["provider_id"]=="provider-b"
assert result["output_json"]["provider"]=="provider-b"
def test_profile_provider_allowlist_fails_closed(tmp_path,monkeypatch):
helper_path=helper(tmp_path)
profile_root=write_profile(tmp_path,helper_path,providers=("provider-b",))
db=tmp_path/"governance.sqlite3";seed(db)
monkeypatch.setattr(runtime,"MODEL_GOVERNANCE_ENABLED",True)
monkeypatch.setattr(runtime,"DEFAULT_MODEL_GOVERNANCE_DB",db)
req=tmp_path/"request.json";res=tmp_path/"result.json"
req.write_text(json.dumps(make_external_request(projected())),encoding="utf-8")
with pytest.raises(runtime.ExternalAgentRuntimeError,match="EXTERNAL_AGENT_MODEL_PROVIDER_PROFILE_DENY"):
runtime.execute_request(req,res,profile_root=profile_root)
def test_governance_required_but_runtime_disabled_fails_closed(tmp_path,monkeypatch):
helper_path=helper(tmp_path)
profile_root=write_profile(tmp_path,helper_path)
monkeypatch.setattr(runtime,"MODEL_GOVERNANCE_ENABLED",False)
req=tmp_path/"request.json";res=tmp_path/"result.json"
req.write_text(json.dumps(make_external_request(projected())),encoding="utf-8")
with pytest.raises(runtime.ExternalAgentRuntimeError,match="EXTERNAL_AGENT_MODEL_GOVERNANCE_DISABLED"):
runtime.execute_request(req,res,profile_root=profile_root)
def test_execution_request_cannot_override_authoritative_task_scope():
task_id="task-scope-override"
request={
"task_id":task_id,
"goal":"Return result.",
"instructions":"",
"success_criteria":[],
"constraints":{},
"scope_type":"tenant",
"tenant_id":"tenant-a",
"project_id":"project-a",
"trace_id":"trace-authoritative",
"required_capabilities":["agent.runtime.hermes"],
}
task=project_online_task(
binding(task_id),
request,
{
"action":"agent_runtime",
"adapter":"hermes",
"capability_scope":["agent.runtime.hermes"],
"runtime_input":{
"goal":"Return result.",
"instructions":"",
"success_criteria":[],
"constraints":{},
"scope":{
"scope_type":"tenant",
"tenant_id":"tenant-b",
"project_id":"project-b",
},
"trace_id":"trace-forged",
},
},
)
runtime_input=task["metadata"]["execution"]["runtime_input"]
assert runtime_input["scope"]["tenant_id"]=="tenant-a"
assert runtime_input["scope"]["project_id"]=="project-a"
assert runtime_input["trace_id"]=="trace-authoritative"
def test_openclaw_nested_usage_is_extracted_exactly():
value={
"meta":{
"agentMeta":{
"usage":{"input":7770,"output":13,"total":7783},
"lastCallUsage":{"input":7770,"output":13,"total":7783},
"costUsd":0,
}
}
}
assert runtime._extract_runtime_usage(value)==(7770,13,0)
def test_governance_required_profile_needs_provider_allowlist_and_output_limit(tmp_path):
path=tmp_path/"helper.py";path.write_text("print('ok')\\n")
root=tmp_path/"profiles";root.mkdir()
(root/"hermes.json").write_text(json.dumps({
"schema":"aios.external-agent-runtime.profile.v1",
"runtime":"hermes",
"argv":["/usr/bin/python3",str(path),"{model_id}"],
"input_mode":"stdin_text",
"output_mode":"stdout_text",
"model_governance":{"required":True},
}))
with pytest.raises(
runtime.ExternalAgentRuntimeError,
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)
@@ -0,0 +1,99 @@
from __future__ import annotations
import json
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
def test_executor_has_single_model_governance_sqlite_write_path():
unit=(ROOT/"deploy/systemd/aios-external-agent-executor.service").read_text()
assert "AIOS_MODEL_GOVERNANCE_DB=/var/lib/aios-model-governance/state.sqlite3" in unit
assert "AIOS_MODEL_GOVERNANCE_ENABLED=1" in unit
assert "ReadWritePaths=/var/lib/aios-model-governance" in unit
assert "ReadWritePaths=/var/lib/aios-external-agent-executor" in unit
assert "InaccessiblePaths=/etc/aios-gitops" in unit
assert "InaccessiblePaths=/data/aios-gitops" in unit
def test_tmpfiles_owner_matches_external_executor():
line=(ROOT/"deploy/tmpfiles.d/aios-model-governance.conf").read_text().strip()
assert line == (
"d /var/lib/aios-model-governance 0750 "
"aios-engineering-executor aios-executor-exchange -"
)
def test_staging_models_use_real_e2e_evaluation_evidence():
value=json.loads(
(ROOT/"config/model-governance/model-governance.staging.json").read_text()
)
assert value["schema"]=="aios.model-governance.bootstrap.v1"
models={item["model_id"]:item for item in value["models"]}
assert models["deepseek-v4-flash"]["evaluation_sha256"] == (
"43ae0db1640daeb6d33e2787769732844f366d36070065d3a2239905bd9a823b"
)
assert models["openai/deepseek-v4-flash"]["evaluation_sha256"] == (
"7b1c89634eb9c0e7cb5a3d75323d904ebbd91b3ae5494ce938bd8316a4cf5b12"
)
def test_runtime_profiles_bind_model_from_governance_not_task():
for name in ("hermes","openclaw"):
value=json.loads(
(ROOT/"config/external-agent-runtime.d"/f"{name}.profile.json").read_text()
)
governance=value["model_governance"]
assert governance["required"] is True
assert governance["provider_ids"]
assert governance["max_output_tokens"] > 0
serialized=json.dumps(value["argv"])+json.dumps(value.get("environment",{}))
assert "{model_id}" in serialized or "{provider_id}" in serialized
def test_staging_bootstrap_has_explicit_tenant_live_policy():
value=json.loads(
(ROOT/"config/model-governance/model-governance.staging.json").read_text()
)
keys={
(
item["scope_type"],
item.get("tenant_id"),
item.get("project_id"),
)
for item in value["policies"]
}
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,209 @@
from __future__ import annotations
from pathlib import Path
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
import aios_online_agent_worker as worker
from aios_online_agent_execution import (
GovernedExecutionBinding,
project_execution_envelope,
)
@pytest.fixture(autouse=True)
def _state_paths(monkeypatch, tmp_path):
root=tmp_path/"worker-state"
monkeypatch.setattr(worker,"STATE_ROOT",root)
monkeypatch.setattr(worker,"WORKER_LIVENESS_PATH",root/"worker-liveness.json")
BASE="a"*40
TASK_ID="task-worker-availability-placeholder"
JOB_ID="job-worker-availability-placeholder"
BRANCH="agent/task-worker-availability-placeholder"
IDEMPOTENCY="task-worker-availability-placeholder-binding-v1"
def binding():
return GovernedExecutionBinding.from_mapping({
"task_id":TASK_ID,
"job_id":JOB_ID,
"repository":"aios/aios",
"base_oid":BASE,
"expected_branch_oid":BASE,
"target_branch":BRANCH,
"branch_mode":"CREATE_NEW",
"generation":1,
"idempotency_key":IDEMPOTENCY,
})
def placeholder():
return project_execution_envelope(
binding(),
{
"action":"source_capture",
"adapter":"source_capture",
"capability_scope":[
"git.read",
"git.test",
"governed.write.branch",
"governed.write.file",
],
},
)
def readonly_source_capture():
return project_execution_envelope(
binding(),
{
"action":"source_capture",
"adapter":"source_capture",
"capability_scope":["git.read","git.test"],
"source_signals":["worker availability"],
},
)
def deterministic():
return project_execution_envelope(
binding(),
{
"action":"deterministic",
"adapter":"deterministic",
"capability_scope":[
"git.read",
"git.test",
"governed.write.branch",
"governed.write.file",
],
"change_set":[{
"path":"active-runtime/app/example.py",
"operation":"write_file",
"content":"VALUE=1\\n",
}],
"verification":[{
"label":"unit",
"argv":["/usr/bin/python3","-m","pytest","-q"],
"cwd":".",
"timeout_seconds":120,
}],
"governance":{
"commit_idempotency_key":"task-worker-availability-placeholder-commit-v1",
"allowed_files":["active-runtime/app/example.py"],
},
},
)
def task(execution):
return {
"task_id":TASK_ID,
"required_capabilities":list(execution["capability_scope"]),
"metadata":{
"execution":execution,
"execution_projection":{
"schema":"aios.online-agent.task-projection.v1",
"task_id":TASK_ID,
"job_id":JOB_ID,
"execution_digest":execution["execution_digest"],
},
},
}
class FakeBridge:
def __init__(self,status,execution):
self.status=status
self.execution=execution
self.calls=[]
def project_execution(self,task_value,execution=None):
self.calls.append(execution)
return self.status,self.execution
class FakeAgent:
def __init__(self,tasks):
self.tasks=tasks
self.claim_calls=[]
def poll(self):
return list(self.tasks)
def claim(self,task_id):
self.claim_calls.append(task_id)
return 500,{"error":"SHOULD_NOT_CLAIM"}
def supervisor(bridge, agent=None):
value=object.__new__(worker.WorkerSupervisor)
value.bridge=bridge
value.agent=agent
return value
def test_write_capable_source_capture_is_classified_as_placeholder():
envelope=worker.ExecutionEnvelope.from_task(
task(placeholder()),
worker_capabilities=tuple(placeholder()["capability_scope"]),
)
assert worker.source_capture_requires_deterministic_plan(envelope) is True
def test_readonly_source_capture_is_not_placeholder():
envelope=worker.ExecutionEnvelope.from_task(
task(readonly_source_capture()),
worker_capabilities=tuple(readonly_source_capture()["capability_scope"]),
)
assert worker.source_capture_requires_deterministic_plan(envelope) is False
def test_placeholder_without_registered_plan_is_skipped_before_claim():
persisted=placeholder()
bridge=FakeBridge(200,persisted)
agent=FakeAgent([task(persisted)])
sup=supervisor(bridge,agent)
assert sup.once()==0
assert bridge.calls==[None]
assert agent.claim_calls==[]
def test_placeholder_with_missing_projection_is_skipped_before_claim():
persisted=placeholder()
bridge=FakeBridge(404,{"error":"TASK_EXECUTION_PROJECTION_MISSING"})
agent=FakeAgent([task(persisted)])
sup=supervisor(bridge,agent)
assert sup.once()==0
assert bridge.calls==[None]
assert agent.claim_calls==[]
def test_registered_deterministic_plan_replaces_placeholder():
persisted=placeholder()
resolved=deterministic()
bridge=FakeBridge(200,resolved)
sup=supervisor(bridge)
envelope,source=sup.resolve_envelope(task(persisted))
assert source=="plan_registry_verified"
assert envelope.action=="deterministic"
assert envelope.digest==resolved["execution_digest"]
assert bridge.calls==[None]
def test_real_readonly_source_capture_replays_exact_payload():
persisted=readonly_source_capture()
bridge=FakeBridge(200,persisted)
sup=supervisor(bridge)
envelope,source=sup.resolve_envelope(task(persisted))
assert source=="persisted_verified"
assert envelope.action=="source_capture"
assert envelope.digest==persisted["execution_digest"]
assert bridge.calls==[persisted]
@@ -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
+199
View File
@@ -0,0 +1,199 @@
from __future__ import annotations
import json
from pathlib import Path
import sys
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
from aios_worker_availability_guard import AvailabilityGuard
class FakeSystemd:
def __init__(self, roots, now=1000.0):
self.now=now
self.actions=[]
self.units={}
for unit,root in roots.items():
self.units[unit]={
"LoadState":"loaded",
"ActiveState":"active",
"UnitFileState":"enabled",
"Environment":f"AIOS_WORKER_STATE_ROOT={root}",
}
self.units["aios-external-agent-executor.service"]={
"LoadState":"loaded",
"ActiveState":"active",
"UnitFileState":"enabled",
"Environment":"",
}
def property(self, unit, name):
return self.units.get(unit,{}).get(name,"")
def environment(self, unit):
raw=self.property(unit,"Environment")
result={}
for item in raw.split():
if "=" in item:
k,v=item.split("=",1)
result[k]=v
return result
def enable(self, unit):
self.actions.append((unit,"enable"))
self.units[unit]["UnitFileState"]="enabled"
def start(self, unit):
self.actions.append((unit,"start"))
self.units[unit]["ActiveState"]="active"
env=self.environment(unit)
if "AIOS_WORKER_STATE_ROOT" not in env:
return
root=Path(env["AIOS_WORKER_STATE_ROOT"])
root.mkdir(parents=True,exist_ok=True)
(root/"worker-liveness.json").write_text(json.dumps({
"schema":"aios.online-agent.worker-liveness.v1",
"agent_id":"agent-"+unit,
"state":"POLL_OK",
"last_seen_epoch":self.now,
}),encoding="utf-8")
def restart(self, unit):
self.actions.append((unit,"restart"))
self.units[unit]["ActiveState"]="active"
env=self.environment(unit)
root=Path(env["AIOS_WORKER_STATE_ROOT"])
root.mkdir(parents=True,exist_ok=True)
(root/"worker-liveness.json").write_text(json.dumps({
"schema":"aios.online-agent.worker-liveness.v1",
"agent_id":"agent-"+unit,
"state":"POLL_OK",
"last_seen_epoch":self.now,
}),encoding="utf-8")
def make_guard(tmp_path, fake):
worker_units=tuple(
unit for unit in fake.units
if unit!="aios-external-agent-executor.service"
)
return AvailabilityGuard(
systemd=fake,
worker_units=worker_units,
dependency_units=("aios-external-agent-executor.service",),
status_path=tmp_path/"run/status.json",
recovery_path=tmp_path/"run/recovery.json",
inhibit_path=tmp_path/"run/maintenance.lock",
max_stale_seconds=90,
restart_cooldown_seconds=60,
startup_wait_seconds=0,
now_fn=lambda:fake.now,
sleep_fn=lambda _seconds:None,
)
def write_liveness(root, now, *, state="POLL_OK"):
root.mkdir(parents=True,exist_ok=True)
(root/"worker-liveness.json").write_text(json.dumps({
"schema":"aios.online-agent.worker-liveness.v1",
"agent_id":"agent",
"state":state,
"last_seen_epoch":now,
}),encoding="utf-8")
def test_healthy_workers_need_no_action(tmp_path):
roots={
"generic.service":tmp_path/"generic",
"hermes.service":tmp_path/"hermes",
"openclaw.service":tmp_path/"openclaw",
}
fake=FakeSystemd(roots)
for root in roots.values():
write_liveness(root,fake.now)
result=make_guard(tmp_path,fake).run(heal=True)
assert result["status"]=="HEALTHY"
assert result["all_fresh"] is True
assert fake.actions==[]
def test_inactive_worker_is_started_and_becomes_fresh(tmp_path):
roots={"generic.service":tmp_path/"generic"}
fake=FakeSystemd(roots)
fake.units["generic.service"]["ActiveState"]="inactive"
result=make_guard(tmp_path,fake).run(heal=True)
assert ("generic.service","start") in fake.actions
assert result["all_fresh"] is True
def test_disabled_worker_is_enabled_and_started(tmp_path):
roots={"generic.service":tmp_path/"generic"}
fake=FakeSystemd(roots)
fake.units["generic.service"]["UnitFileState"]="disabled"
fake.units["generic.service"]["ActiveState"]="inactive"
result=make_guard(tmp_path,fake).run(heal=True)
assert ("generic.service","enable") in fake.actions
assert ("generic.service","start") in fake.actions
assert result["all_fresh"] is True
def test_active_but_stale_worker_is_restarted(tmp_path):
roots={"generic.service":tmp_path/"generic"}
fake=FakeSystemd(roots)
write_liveness(roots["generic.service"],fake.now-500)
result=make_guard(tmp_path,fake).run(heal=True)
assert ("generic.service","restart") in fake.actions
assert result["all_fresh"] is True
assert result["workers"][0]["action"]=="RESTART_STALE"
def test_restart_cooldown_prevents_restart_storm(tmp_path):
roots={"generic.service":tmp_path/"generic"}
fake=FakeSystemd(roots)
write_liveness(roots["generic.service"],fake.now-500)
guard=make_guard(tmp_path,fake)
guard.recovery_path.parent.mkdir(parents=True,exist_ok=True)
guard.recovery_path.write_text(json.dumps({
"schema":"aios.worker-availability.recovery.v1",
"last_restart_epoch":{"generic.service":fake.now-10},
}),encoding="utf-8")
result=guard.run(heal=True)
assert ("generic.service","restart") not in fake.actions
assert result["status"]=="DEGRADED"
assert result["workers"][0]["detail"]=="RESTART_COOLDOWN_ACTIVE"
def test_maintenance_inhibit_disables_healing(tmp_path):
roots={"generic.service":tmp_path/"generic"}
fake=FakeSystemd(roots)
fake.units["generic.service"]["ActiveState"]="inactive"
guard=make_guard(tmp_path,fake)
guard.inhibit_path.parent.mkdir(parents=True,exist_ok=True)
guard.inhibit_path.write_text("planned maintenance\n")
result=guard.run(heal=True)
assert result["status"]=="MAINTENANCE_INHIBITED"
assert fake.actions==[]
def test_dependency_is_enabled_and_started(tmp_path):
roots={"generic.service":tmp_path/"generic"}
fake=FakeSystemd(roots)
write_liveness(roots["generic.service"],fake.now)
fake.units["aios-external-agent-executor.service"]["UnitFileState"]="disabled"
fake.units["aios-external-agent-executor.service"]["ActiveState"]="inactive"
result=make_guard(tmp_path,fake).run(heal=True)
assert ("aios-external-agent-executor.service","enable") in fake.actions
assert ("aios-external-agent-executor.service","start") in fake.actions
assert result["all_fresh"] is True
def test_check_reports_stale_without_mutation(tmp_path):
roots={"generic.service":tmp_path/"generic"}
fake=FakeSystemd(roots)
write_liveness(roots["generic.service"],fake.now-500)
result=make_guard(tmp_path,fake).run(heal=False)
assert result["status"]=="DEGRADED"
assert result["all_fresh"] is False
assert fake.actions==[]
@@ -0,0 +1,56 @@
from __future__ import annotations
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
def test_worker_source_has_idle_and_busy_liveness():
source=(ROOT/"active-runtime/app/aios_online_agent_worker.py").read_text()
assert "aios.online-agent.worker-liveness.v1" in source
assert 'write_worker_liveness("POLL_OK"' in source
assert '"POLL_FAILED"' in source
assert 'write_worker_liveness("STARTING")' in source
assert '"BUSY"' in source
assert "last_seen_epoch" in source
def test_guard_timer_is_persistent_and_frequent():
timer=(ROOT/"deploy/systemd/aios-worker-availability-guard.timer").read_text()
assert "OnBootSec=15s" in timer
assert "OnUnitActiveSec=30s" in timer
assert "Persistent=true" in timer
assert "WantedBy=timers.target" in timer
def test_guard_service_has_no_network_or_credentials():
unit=(ROOT/"deploy/systemd/aios-worker-availability-guard.service").read_text()
assert "User=root" in unit
assert "LoadCredential" not in unit
assert "RestrictAddressFamilies=AF_UNIX" in unit
assert "NoNewPrivileges=true" in unit
assert "ProtectSystem=strict" in unit
assert "ReadWritePaths=/run/aios-worker-availability" in unit
assert "AIOS_WORKER_AVAILABILITY_MAX_STALE_SECONDS=90" in unit
def test_guard_manages_all_three_workers_and_external_executor():
source=(ROOT/"active-runtime/app/aios_worker_availability_guard.py").read_text()
for unit in (
"aios-online-agent-worker-20260815.service",
"aios-hermes-online-agent-worker.service",
"aios-openclaw-online-agent-worker.service",
"aios-external-agent-executor.service",
):
assert unit in source
assert "maintenance.lock" in source
assert "RESTART_STALE" in source
def test_worker_defers_write_capable_source_capture_until_plan_exists():
source=(ROOT/"active-runtime/app/aios_online_agent_worker.py").read_text()
assert "ENGINEERING_WRITE_CAPABILITIES" in source
assert "source_capture_requires_deterministic_plan" in source
assert "TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN" in source
assert '"governed.write.branch"' in source
assert '"governed.write.file"' in source