from __future__ import annotations from pathlib import Path import json import sys import pytest ROOT = Path(__file__).resolve().parents[1] APP = ROOT / "active-runtime/app" sys.path.insert(0, str(APP)) from aios_online_agent_compact_map import ( MAX_RESULT_BYTES, build_compact_interface_map, ) from aios_online_agent_execution import ( ExecutionProjectionError, GovernedExecutionBinding, project_execution_envelope, validate_execution_envelope, ) import aios_online_agent_worker as worker from aios_online_agent_worker import ( CompactInterfaceMapExecutor, ExecutionEnvelope, build_compact_interface_map_result, ) BASE = "68b95fea0d9af3b52269e79b7cf878254f191b61" def make_repo(tmp_path: Path) -> Path: root = tmp_path / "repo" (root / "active-runtime/app").mkdir(parents=True) (root / "tests").mkdir(parents=True) (root / "db/migrations").mkdir(parents=True) (root / "deploy/systemd").mkdir(parents=True) (root / "active-runtime/app/fleet.py").write_text( """ class WorkerRegistry: def register_worker(self, worker_id, capabilities): return {"worker_id": worker_id, "capabilities": capabilities} class TaskScheduler: def claim_next(self, tenant_id, capacity): return None def recover_assignment(task_id, worker_id): return task_id, worker_id TOKEN = "fixturetok" """.strip() + "\n", encoding="utf-8", ) (root / "active-runtime/app/checkpoint.py").write_text( """ class CheckpointStore: def resume(self, task_id, attempt): return None """.strip() + "\n", encoding="utf-8", ) (root / "tests/test_fleet.py").write_text( """ def test_scheduler_fairness(): assert True def test_worker_reassignment(): assert True """.strip() + "\n", encoding="utf-8", ) (root / "db/migrations/20260817_fleet.sql").write_text( "CREATE TABLE IF NOT EXISTS worker_slots (worker_id TEXT, capacity INTEGER);\n", encoding="utf-8", ) (root / "deploy/systemd/aios-fleet-worker.service").write_text( "[Service]\nExecStart=/usr/bin/true\n", encoding="utf-8", ) return root def binding() -> GovernedExecutionBinding: return GovernedExecutionBinding.from_mapping( { "task_id": "r42-p2-agent-fleet-interface-map-20260817", "job_id": "job-2c451360f78ef595e41663e3b444a8fb", "repository": "aios/aios", "base_oid": BASE, "expected_branch_oid": BASE, "target_branch": "agent/r42-p2-agent-fleet-interface-map-20260817", "branch_mode": "CREATE_NEW", "generation": 1, "idempotency_key": "r42-p2-agent-fleet-interface-map-68b95fe-v1", } ) def envelope_mapping(): return project_execution_envelope( binding(), { "action": "source_capture", "adapter": "compact_interface_map", "capability_scope": ["git.read", "git.test"], "source_signals": [ "worker registry scheduler capacity fairness fault isolation", "checkpoint resume retry scope trace message", ], }, required_capabilities=["git.read", "git.test"], ) def task_for(envelope): return { "task_id": envelope["task_id"], "metadata": {"execution": envelope}, "required_capabilities": ["git.read", "git.test"], } def test_compact_map_bounds_and_surfaces(tmp_path): root = make_repo(tmp_path) result = build_compact_interface_map( root, ["worker scheduler capacity fairness checkpoint recovery"], source_commit=BASE, ) encoded = json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() assert result["schema"] == "aios.engineering.interface-map.v1" assert len(encoded) <= MAX_RESULT_BYTES assert len(result["implementation_map"]["modules"]) <= 40 assert len(result["bounded_excerpts"]) <= 25 assert len(result["implementation_map"]["tests"]) <= 40 assert any(item["symbol"] == "WorkerRegistry" for item in result["implementation_map"]["classes"]) assert any(item["table"] == "worker_slots" for item in result["implementation_map"]["tables"]) assert any(item["unit"] == "aios-fleet-worker.service" for item in result["implementation_map"]["systemd_units"]) def test_compact_map_redacts_sensitive_source_patterns(tmp_path): root = make_repo(tmp_path) result = build_compact_interface_map(root, ["token fleet"], source_commit=BASE) excerpts = json.dumps(result["bounded_excerpts"], ensure_ascii=False) assert "fixturetok" not in excerpts def test_compact_execution_projection_is_read_only(): envelope = envelope_mapping() assert envelope["action"] == "source_capture" assert envelope["adapter"] == "compact_interface_map" assert envelope["governance"]["writer_lease"] is False assert envelope["change_set"] == [] assert envelope["verification"] == [] validate_execution_envelope(envelope, binding=binding(), worker_capabilities=["git.read", "git.test"]) def test_compact_execution_rejects_mutation(): with pytest.raises(ExecutionProjectionError, match="SOURCE_CAPTURE_MUTATION_DENY"): project_execution_envelope( binding(), { "action": "source_capture", "adapter": "compact_interface_map", "change_set": [{"path": "x.py", "operation": "write_file", "content": "x=1\n"}], }, ) def test_worker_accepts_compact_adapter_and_builds_result(tmp_path, monkeypatch): root = make_repo(tmp_path) monkeypatch.setattr(worker, "audit", lambda *args, **kwargs: None) mapped = envelope_mapping() envelope = ExecutionEnvelope.from_task( task_for(mapped), worker_capabilities=("git.read", "git.test"), ) execution = CompactInterfaceMapExecutor().execute(envelope, root) result = build_compact_interface_map_result(envelope.task_id, envelope, execution) assert result["status"] == "SUCCEEDED" assert result["classification"] == "P2_INTERFACE_MAP_CAPTURED" assert result["execution_mode"] == "compact_interface_map" assert result["writer_lease"] == "NONE" assert "source_snapshot" not in result assert result["result_bytes"] <= 120 * 1024 assert result["source_commit"] == BASE def test_compact_contract_keys_are_complete(tmp_path): root = make_repo(tmp_path) result = build_compact_interface_map(root, ["fleet worker scheduler retry"], source_commit=BASE) assert set(result["contracts"]) == { "FLEET_AGENT_REGISTRATION", "FLEET_CAPACITY_SCHEDULING", "FLEET_FAIRNESS", "FLEET_FAULT_ISOLATION", "FLEET_WORKER_REASSIGNMENT", "FLEET_SCOPE_ISOLATION", "FLEET_IDEMPOTENT_DISPATCH", "FLEET_FAILURE_RECOVERY", } def test_compact_result_has_no_source_archive(tmp_path): root = make_repo(tmp_path) result = build_compact_interface_map(root, ["fleet"], source_commit=BASE) serialized = json.dumps(result) assert "payload_b64" not in serialized assert "tar.gz+base64" not in serialized def test_create_new_retries_discard_known_policy_reconciliation_marker(tmp_path, monkeypatch): task_id = "r42-p2-compact-interface-map-runtime-20260817" job_id = "job-74d0982f616a7f2653ef346144139044" monkeypatch.setattr(worker, "STATE_ROOT", tmp_path / "worker-state") state_path = worker.task_state_path(task_id) state_path.parent.mkdir(parents=True) state_path.write_text( json.dumps( { "task_id": task_id, "execution_digest": "old-digest", "phase": "reconciliation_required", "attempt": 16, "job_id": job_id, "details": { "api_status": 503, "error": "SECRET_LIKE_PAYLOAD_REJECTED", }, } ), encoding="utf-8", ) assert worker.load_task_state( task_id, "new-digest", attempt=18, continuation_head_oid=None, continuation_job_id=job_id, commit_idempotency_key="r42-p2-compact-interface-map-runtime-commit-v1", ) is None