Spaces:
Sleeping
Sleeping
File size: 3,451 Bytes
3b835f5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | 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()) |