363 lines
13 KiB
Python
363 lines
13 KiB
Python
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):
|
|
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"],
|
|
"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)
|