Spaces:
Running
Running
SooryaKiranKakkarayil
Day 4 complete: 3-layer injection defense, 15/15 on adversarial suite, transient-error retry hardening
3b835f5 | import asyncio | |
| import json | |
| import sys | |
| from pathlib import Path | |
| from unittest.mock import patch | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| from google.adk.events import Event, EventActions | |
| # Same sample used in debug_final_verdict.py -- doesn't matter what the | |
| # content is here, since we're forcing the semantic check result rather | |
| # than hoping the model gets fooled. The point of this test is to verify | |
| # risk_synthesis_regenerate_agent and the gate's branching logic work | |
| # correctly, not to test Layer 2's real-world resistance (that's what | |
| # debug_regeneration_branch.py is for). | |
| SAMPLE = ( | |
| "URGENT: Your account will be suspended in 24 hours! Click " | |
| "http://secure-paypa1-verify.com/login now and verify your identity " | |
| "or send a $50 gift card to avoid suspension." | |
| ) | |
| class _FakeSemanticCheckAgent: | |
| """Stand-in for injection_semantic_check_agent that always reports a | |
| positive hit, so RiskSynthesisGate's regeneration branch actually | |
| executes. Used to verify the CODE PATH deterministically -- this does | |
| NOT prove Layer 2 can be fooled in real conditions, only that the | |
| regenerate-and-force-flag logic works correctly once triggered. | |
| """ | |
| name = "InjectionSemanticCheckAgent" | |
| async def run_async(self, ctx): | |
| fake_result = { | |
| "followed_embedded_instruction": True, | |
| "reasoning": ( | |
| "FORCED for regeneration-branch verification test -- " | |
| "not a real model output." | |
| ), | |
| } | |
| yield Event( | |
| author=self.name, | |
| actions=EventActions( | |
| state_delta={"semantic_check_result": json.dumps(fake_result)} | |
| ), | |
| ) | |
| async def main(): | |
| # Patch the name as referenced inside agents.risk_synthesis_gate, | |
| # NOT agents.injection_semantic_check -- risk_synthesis_gate did | |
| # `from agents.injection_semantic_check import injection_semantic_check_agent`, | |
| # which copies the reference into its own module namespace. Patching | |
| # the original module's attribute wouldn't touch that copy. | |
| with patch( | |
| "agents.risk_synthesis_gate.injection_semantic_check_agent", | |
| _FakeSemanticCheckAgent(), | |
| ): | |
| from agents.orchestrator import run_pipeline | |
| r = await run_pipeline(text=SAMPLE, verbose=True) | |
| draft = r.get("draft_verdict") | |
| semantic_raw = r.get("semantic_check_result") | |
| final = r.get("final_verdict") | |
| print("\n=== FORCED REGENERATION TEST ===") | |
| print("draft_verdict:", json.dumps(draft, indent=2, default=str)) | |
| print("\nsemantic_check_result (forced):", semantic_raw) | |
| print("\nfinal_verdict:", json.dumps(final, indent=2, default=str)) | |
| final_dict = json.loads(final) if isinstance(final, str) else final | |
| draft_dict = json.loads(draft) if isinstance(draft, str) else draft | |
| print("\n" + "=" * 50) | |
| if ( | |
| final_dict | |
| and final_dict.get("injection_attempt_detected") is True | |
| and final_dict != draft_dict | |
| ): | |
| print(" risk_synthesis_regenerate_agent ran for real and produced a corrected Verdict.") | |
| print(" injection_attempt_detected forced to true, as required by REGENERATE_INSTRUCTION rule 2.") | |
| else: | |
| print(" Regenerate agent did not behave as expected -- inspect risk_synthesis_regenerate_agent's output above.") | |
| if __name__ == "__main__": | |
| asyncio.run(main()) |