import json import tempfile import unittest from unittest.mock import patch from cyber_agent.application import ( AgentApplication, AgentRole, ApplicationRegistry, ApplicationRuntime, ApplicationServices, ProviderRef, QualityRuleSpec, ) from cyber_agent.application.candidate import CandidateLedger from cyber_agent.application.quality import QualityCheckOutcome from cyber_agent.core.artifacts import ( ArtifactRef, ValidationMaterial, material_content_hash, ) from cyber_agent.tools.registry import ToolRegistry from cyber_agent.trace.store import FileSystemTraceStore class CandidateRepository: def __init__(self): self.requests = {} async def put_version(self, request): self.requests.setdefault(request.operation_id, request) saved = self.requests[request.operation_id] return ArtifactRef( artifact_id=f"candidate:{saved.candidate_id}:{saved.revision}", version=str(saved.revision), content_hash=material_content_hash(saved.content), kind="candidate.output", mime_type="application/json", ) async def merge(self, request): return await self.put_version(request) async def load_version(self, candidate_ref): return next( item.content for item in self.requests.values() if item.candidate_id == candidate_ref.candidate_id and item.revision == candidate_ref.revision ) async def resolve(self, ref, root_trace_id, uid): request = next( item for item in self.requests.values() if ref.artifact_id == f"candidate:{item.candidate_id}:{item.revision}" ) return ValidationMaterial( **ref.model_dump(mode="json"), root_trace_id=root_trace_id, uid=uid, content=request.content, ) async def commit_adoption(self, request): raise NotImplementedError class QualityProvider: def __init__(self): self.mode = "normal" self.calls = 0 async def check(self, request): self.calls += 1 outcomes = [] for rule in request.rules: failed = "{{placeholder}}" in request.material.content["text"] outcomes.append(QualityCheckOutcome( rule_id=rule.rule_id, rule_version=rule.version, status="failed" if failed else "passed", issue="placeholder remains" if failed else None, )) if self.mode == "missing": return [] if self.mode == "extra": outcomes.append(QualityCheckOutcome( rule_id="unexpected", rule_version="1", status="passed", )) if self.mode == "error": raise RuntimeError("provider unavailable") return outcomes class CandidateQualityTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.temp = tempfile.TemporaryDirectory() self.store = FileSystemTraceStore(self.temp.name) self.repository = CandidateRepository() self.quality = QualityProvider() application = AgentApplication( application_id="quality.reference", application_version="1", root_role="writer", roles=(AgentRole( role_id="writer", model="validator-model", system_prompt="write", ),), artifact_resolver_ref=ProviderRef( provider_id="artifacts", provider_version="1", ), candidate_repository_ref=ProviderRef( provider_id="candidates", provider_version="1", ), quality_provider_ref=ProviderRef( provider_id="quality", provider_version="1", ), quality_rules=(QualityRuleSpec( rule_id="no_placeholder", version="1", criterion="Candidate contains no placeholder text", ),), ) registry = ApplicationRegistry() self.binding = registry.register( application, ApplicationServices( tool_registry=ToolRegistry(), artifact_resolver=self.repository, candidate_repository=self.repository, quality_provider=self.quality, ), ) self.llm_calls = 0 async def llm_call(**kwargs): self.llm_calls += 1 packet = json.loads(kwargs["messages"][-1]["content"]) scope = packet["validation_scope"] return { "content": json.dumps({ "scope": scope, "outcome": "passed", "checks": [ { "check_id": item["check_id"], "status": "passed", "evidence_refs": [], "issue": None, } for item in packet["validation_plan"]["checks"] ], "reason": "candidate output is complete", "retry_from": None, }), "tool_calls": [], } runtime = ApplicationRuntime( registry=registry, trace_store=self.store, llm_call=llm_call, ) self.runner, config = runtime.new_run( "quality.reference", "1", uid="user-1", root_task_anchor={ "objective": "write", "completion_criteria": ["complete"], "constraints": [], }, ) with patch.dict("os.environ", {"AGENT_MODE": "recursive"}, clear=False): self.trace, _goal, _sequence = await self.runner._prepare_new_trace( [{"role": "user", "content": "start"}], config, ) async def asyncTearDown(self): self.temp.cleanup() async def _candidate(self, text, sequence): return await self.runner.candidate_service.manage( self.trace.trace_id, operation="create", content={"text": text}, parent_refs=[], effective_at_sequence=sequence, ) async def test_deterministic_and_llm_checks_are_separate_and_cached(self): candidate = await self._candidate("finished", 2) first = await self.runner.validate_recursive_trace( self.trace.trace_id, candidate_ref=candidate, ) checks = first.result.scope_results[0].checks self.assertEqual("passed", first.result.outcome) self.assertIn("quality.no_placeholder", [item.check_id for item in checks]) self.assertTrue(any(item.check_id.startswith("output.policy") for item in checks)) self.assertEqual(1, self.quality.calls) self.assertEqual(1, self.llm_calls) second = await self.runner.validate_recursive_trace( self.trace.trace_id, candidate_ref=candidate, ) self.assertTrue(second.cached) self.assertEqual(1, self.quality.calls) self.assertEqual(1, self.llm_calls) ledger = CandidateLedger.model_validate( await self.store.get_candidate_ledger(self.trace.trace_id) ) self.assertEqual(1, len(ledger.validations)) async def test_failed_candidate_does_not_call_llm_or_pollute_peer(self): good = await self._candidate("finished", 2) bad = await self._candidate("{{placeholder}}", 3) good_run = await self.runner.validate_recursive_trace( self.trace.trace_id, candidate_ref=good, ) bad_run = await self.runner.validate_recursive_trace( self.trace.trace_id, candidate_ref=bad, ) self.assertEqual("passed", good_run.result.outcome) self.assertEqual("failed", bad_run.result.outcome) self.assertEqual(1, self.llm_calls) ledger = CandidateLedger.model_validate( await self.store.get_candidate_ledger(self.trace.trace_id) ) self.assertEqual(2, len(ledger.validations)) async def test_provider_shape_and_exception_fail_closed_with_complete_plan(self): for index, mode in enumerate(("missing", "extra", "error"), start=2): with self.subTest(mode=mode): self.quality.mode = mode candidate = await self._candidate(f"finished-{mode}", index) run = await self.runner.validate_recursive_trace( self.trace.trace_id, candidate_ref=candidate, ) result = run.result.scope_results[0] self.assertEqual("error", result.outcome) self.assertEqual( { item.check_id for item in result.checks }, { "quality.no_placeholder", *{ item.check_id for item in result.checks if item.check_id.startswith("output.policy") }, }, ) self.assertTrue(all( item.status == "unknown" for item in result.checks )) self.assertEqual(0, self.llm_calls)