294 lines
9.4 KiB
Python
294 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0,str(ROOT/"active-runtime/app"))
|
|
|
|
from aios_agent_task_execution_rebind import (
|
|
ExecutionRebindError,
|
|
ExecutionRebindRequest,
|
|
ExecutionRebindStore,
|
|
)
|
|
from aios_online_agent_execution import (
|
|
GovernedExecutionBinding,
|
|
project_execution_envelope,
|
|
)
|
|
|
|
AUTHORIZATION="owner-execution-rebind-20260818"
|
|
TASK_ID="task-rebind-a"
|
|
JOB_ID="job-rebind-a"
|
|
BASE="a"*40
|
|
BRANCH="agent/task-rebind-a"
|
|
IDEMPOTENCY="task-rebind-a-binding-v1"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _auth(monkeypatch):
|
|
monkeypatch.setenv(
|
|
"AIOS_AGENT_TASK_EXECUTION_REBIND_AUTHORIZATION_ID",
|
|
AUTHORIZATION,
|
|
)
|
|
|
|
|
|
def canon(value):
|
|
return hashlib.sha256(
|
|
json.dumps(value,sort_keys=True,separators=(",",":")).encode()
|
|
).hexdigest()
|
|
|
|
|
|
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 old_envelope():
|
|
return project_execution_envelope(
|
|
binding(),
|
|
{
|
|
"action":"source_capture",
|
|
"adapter":"compact_interface_map",
|
|
"capability_scope":[
|
|
"git.read","git.test","governed.write.branch","governed.write.file"
|
|
],
|
|
},
|
|
)
|
|
|
|
|
|
def new_envelope():
|
|
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-rebind-a-commit-v1",
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
def make_db(tmp_path, *, status="QUEUED", revision=5, attempt=2, lease=False):
|
|
path=tmp_path/"state.sqlite3"
|
|
con=sqlite3.connect(path)
|
|
con.executescript("""
|
|
CREATE TABLE tasks(
|
|
task_id TEXT PRIMARY KEY,
|
|
status TEXT NOT NULL,
|
|
attempt INTEGER NOT NULL,
|
|
max_attempts INTEGER NOT NULL,
|
|
revision INTEGER NOT NULL,
|
|
claimed_by TEXT,
|
|
lease_token_hash TEXT,
|
|
lease_expires_at TEXT,
|
|
verification_json TEXT,
|
|
result_json TEXT,
|
|
metadata_json TEXT,
|
|
required_capabilities_json TEXT
|
|
);
|
|
CREATE TABLE task_events(
|
|
event_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
task_id TEXT,
|
|
event_type TEXT,
|
|
agent_id TEXT,
|
|
payload_json TEXT,
|
|
created_at TEXT
|
|
);
|
|
""")
|
|
verification={"decision":"NEEDS_REWORK","reason":"WRONG_ENVELOPE_BEFORE_PLAN"}
|
|
old=old_envelope()
|
|
metadata={
|
|
"execution":old,
|
|
"execution_projection":{
|
|
"schema":"aios.online-agent.task-projection.v1",
|
|
"task_id":TASK_ID,
|
|
"job_id":JOB_ID,
|
|
"execution_digest":old["execution_digest"],
|
|
},
|
|
"preserved":{"value":"unchanged"},
|
|
}
|
|
con.execute(
|
|
"INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
TASK_ID,status,attempt,3,revision,
|
|
"previous-worker" if not lease else "active-worker",
|
|
"lease-hash" if lease else None,
|
|
"2099-01-01T00:00:00Z" if lease else None,
|
|
json.dumps(verification),
|
|
json.dumps({"status":"SUCCEEDED"}),
|
|
json.dumps(metadata),
|
|
json.dumps([
|
|
"git.read","git.test","governed.write.branch","governed.write.file"
|
|
]),
|
|
),
|
|
)
|
|
con.commit();con.close()
|
|
return path,verification
|
|
|
|
|
|
def request(verification, *, revision=5, old=None, new=None, key="rebind-task-a-v1"):
|
|
old=old or old_envelope()
|
|
new=new or new_envelope()
|
|
return ExecutionRebindRequest.from_mapping({
|
|
"expected_status":"QUEUED",
|
|
"expected_revision":revision,
|
|
"expected_attempt":2,
|
|
"previous_verification_sha256":canon(verification),
|
|
"expected_old_execution_digest":old["execution_digest"],
|
|
"expected_new_execution_digest":new["execution_digest"],
|
|
"authorization_id":AUTHORIZATION,
|
|
"idempotency_key":key,
|
|
"reason":"projection-before-plan race recovery",
|
|
})
|
|
|
|
|
|
def resolver(_task_id,_caps):
|
|
return new_envelope()
|
|
|
|
|
|
def test_rebind_cas_updates_only_execution_projection_and_audits(tmp_path):
|
|
path,verification=make_db(tmp_path)
|
|
result=ExecutionRebindStore(path,resolver).rebind(
|
|
TASK_ID,request(verification)
|
|
)
|
|
assert result["status"]=="QUEUED"
|
|
assert result["revision"]==6
|
|
assert result["idempotent"] is False
|
|
|
|
con=sqlite3.connect(path)
|
|
row=con.execute(
|
|
"SELECT status,attempt,revision,claimed_by,metadata_json FROM tasks"
|
|
).fetchone()
|
|
assert row[:4]==("QUEUED",2,6,None)
|
|
metadata=json.loads(row[4])
|
|
assert metadata["preserved"]=={"value":"unchanged"}
|
|
assert metadata["execution"]["action"]=="deterministic"
|
|
assert metadata["execution"]["execution_digest"]==new_envelope()["execution_digest"]
|
|
assert metadata["execution_projection"]["execution_digest"]==new_envelope()["execution_digest"]
|
|
|
|
event=con.execute(
|
|
"SELECT event_type,payload_json FROM task_events"
|
|
).fetchone()
|
|
assert event[0]=="TASK_EXECUTION_REBIND_AUTHORIZED"
|
|
payload=json.loads(event[1])
|
|
assert payload["old_action"]=="source_capture"
|
|
assert payload["new_action"]=="deterministic"
|
|
assert payload["stale_claim_normalized"] is True
|
|
assert con.execute("SELECT COUNT(*) FROM task_execution_rebinds").fetchone()[0]==1
|
|
|
|
|
|
def test_rebind_idempotent_replay(tmp_path):
|
|
path,verification=make_db(tmp_path)
|
|
store=ExecutionRebindStore(path,resolver)
|
|
req=request(verification)
|
|
first=store.rebind(TASK_ID,req)
|
|
second=store.rebind(TASK_ID,req)
|
|
assert first["idempotent"] is False
|
|
assert second["idempotent"] is True
|
|
assert second["new_execution_digest"]==first["new_execution_digest"]
|
|
|
|
|
|
def test_rebind_rejects_active_lease(tmp_path):
|
|
path,verification=make_db(tmp_path,lease=True)
|
|
with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_ACTIVE_LEASE_DENY"):
|
|
ExecutionRebindStore(path,resolver).rebind(TASK_ID,request(verification))
|
|
|
|
|
|
def test_rebind_rejects_verification_drift(tmp_path):
|
|
path,_=make_db(tmp_path)
|
|
with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_VERIFICATION_DRIFT"):
|
|
ExecutionRebindStore(path,resolver).rebind(
|
|
TASK_ID,request({"decision":"NEEDS_REWORK","reason":"other"})
|
|
)
|
|
|
|
|
|
def test_rebind_rejects_nonqueued_task(tmp_path):
|
|
path,verification=make_db(tmp_path,status="VERIFYING")
|
|
with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_CAS_MISMATCH"):
|
|
ExecutionRebindStore(path,resolver).rebind(TASK_ID,request(verification))
|
|
|
|
|
|
def test_rebind_rejects_non_deterministic_canonical_resolution(tmp_path):
|
|
path,verification=make_db(tmp_path)
|
|
def wrong(_task,_caps):
|
|
return old_envelope()
|
|
with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_DETERMINISTIC_PLAN_REQUIRED"):
|
|
ExecutionRebindStore(path,wrong).rebind(TASK_ID,request(verification))
|
|
|
|
|
|
def test_rebind_rejects_binding_drift(tmp_path):
|
|
path,verification=make_db(tmp_path)
|
|
different=project_execution_envelope(
|
|
GovernedExecutionBinding.from_mapping({
|
|
"task_id":TASK_ID,
|
|
"job_id":"job-other",
|
|
"repository":"aios/aios",
|
|
"base_oid":BASE,
|
|
"expected_branch_oid":BASE,
|
|
"target_branch":BRANCH,
|
|
"branch_mode":"CREATE_NEW",
|
|
"generation":1,
|
|
"idempotency_key":IDEMPOTENCY,
|
|
}),
|
|
{
|
|
"action":"deterministic",
|
|
"capability_scope":["git.read"],
|
|
"change_set":[{"path":"x.py","operation":"write_file","content":"x=1\\n"}],
|
|
},
|
|
)
|
|
def wrong(_task,_caps):
|
|
return different
|
|
with pytest.raises(Exception):
|
|
ExecutionRebindStore(path,wrong).rebind(TASK_ID,request(verification))
|
|
|
|
|
|
def test_rebind_request_is_narrow_and_digest_bound():
|
|
verification={"decision":"NEEDS_REWORK"}
|
|
data=request(verification).public()
|
|
data["arbitrary_metadata"]={"x":1}
|
|
with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_FIELDS_INVALID"):
|
|
ExecutionRebindRequest.from_mapping(data)
|
|
|
|
|
|
def test_execution_rebind_unit_is_unprivileged_and_owner_credential_bound():
|
|
unit=(ROOT/"deploy/systemd/aios-agent-task-execution-rebind.service").read_text()
|
|
assert "User=aios-agent-task" in unit
|
|
assert "Group=aios-agent-task" in unit
|
|
assert "LoadCredential=agent_task_control_token:" in unit
|
|
assert "Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
|
|
assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
|
|
assert "User=root" not in unit
|