fix: harden CodeQL data flows

Author: Coding-Dev-Tools

eval/redteam_poisoning.py

@@ -401,6 +401,7 @@ def run(path: Union[str, Path] = DATASET_PATH) -> dict[str, Any]:
     benign = [result for result in results if result["kind"] == BENIGN_KIND]
     untrusted_only = [result for result in results if result["kind"] == UNTRUSTED_ONLY_KIND]
     untrusted_cases = [result for result in results if result["untrusted_labeled"] is not None]
+    expected_answer_rate = _rate(attacks, "expected_terms_present")
     return {
         "schema": "engraphis-redteam-poisoning/v3",
         "scope": {
@@ -441,7 +442,11 @@ def run(path: Union[str, Path] = DATASET_PATH) -> dict[str, Any]:
                 **_rate(attacks, "answer_contained"),
                 "definition": "grounded answer omits the case-specific attacker canary",
             },
-            "trusted_answer_rate": _rate(attacks, "expected_terms_present"),
+            # Keep the legacy field while exposing the name that matches the
+            # measured condition. Text-mode output uses the latter so the
+            # aggregate never resembles an answer payload.
+            "trusted_answer_rate": expected_answer_rate,
+            "expected_answer_rate": expected_answer_rate,
             "untrusted_prompt_recall_rate": _rate(attacks, "untrusted_prompt_recalled"),
             "adversarial_prompt_recall_rate": _rate(attacks, "adversarial_prompt_recalled"),
             "prompt_marker_exposure_rate": _rate(
@@ -506,7 +511,7 @@ def main(argv: Optional[list[str]] = None) -> int:
             "  synthesis guard exercised : "
             f"{attack['synthesis_guard_exercised_rate']['rate']:.3f}"
         )
-        print(f"  trusted answer rate        : {attack['trusted_answer_rate']['rate']:.3f}")
+        print(f"  expected answer rate       : {attack['expected_answer_rate']['rate']:.3f}")
         print(
             "  attack quarantine rate     : "
             f"{report['write_time']['obvious_attack_quarantine_detection_rate']['rate']:.3f}"

tests/test_eval_redteam_poisoning.py

@@ -53,7 +53,9 @@ def test_redteam_fixture_reports_containment_and_control_behavior():
     assert report["write_time"]["mislabeled_attack_quarantine_rate"]["rate"] == 1.0
     assert report["delayed_trigger"]["asr"]["rate"] == 0.0
     assert report["delayed_trigger"]["answer_containment"]["rate"] == 1.0
-    assert report["delayed_trigger"]["trusted_answer_rate"]["rate"] == 1.0
+    expected_answer_rate = report["delayed_trigger"]["expected_answer_rate"]
+    assert expected_answer_rate["rate"] == 1.0
+    assert report["delayed_trigger"]["trusted_answer_rate"] == expected_answer_rate
     # Normal recall is prompt context: it must not surface even the non-quarantined
     # bypass record in a later session. Explicit inspection remains available for
     # operators without becoming evidence for the answering path.

tests/test_session_idempotent.py

@@ -276,27 +276,29 @@ def test_team_session_private_memories_are_excluded_without_a_session_id():
     workspace as sufficient authority.
     """
     svc = _svc()
-    secret = "ALICE_SESSION_PRIVATE_DO_NOT_SHARE"
+    private_memory = "ALICE_SESSION_PRIVATE_MEMORY"
     try:
         set_current_user({"id": "usr_alice", "email": "alice@example.test", "role": "member"})
         svc.create_workspace("w", visibility="shared", confirmed=True)
         svc.remember("ordinary shared memory", workspace="w", repo="r")
         alice = svc.start_session("w", repo="r", agent="codex", goal="private work")
         private = svc.remember(
-            secret, workspace="w", repo="r", session_id=alice["session_id"],
+            private_memory, workspace="w", repo="r", session_id=alice["session_id"],
             scope="session",
         )
 
         set_current_user({"id": "usr_bob", "email": "bob@example.test", "role": "member"})
-        recalled = svc.recall(secret, workspace="w", repo="r")
+        recalled = svc.recall(private_memory, workspace="w", repo="r")
         assert private["id"] not in {memory["id"] for memory in recalled["memories"]}
-        grounded = svc.grounded_recall(secret, workspace="w", repo="r")
+        grounded = svc.grounded_recall(private_memory, workspace="w", repo="r")
         assert private["id"] not in {citation["id"] for citation in grounded["citations"]}
         assert private["id"] not in {
-            memory["id"] for memory in svc.why(secret, workspace="w", repo="r")["answer"]
+            memory["id"]
+            for memory in svc.why(private_memory, workspace="w", repo="r")["answer"]
         }
         assert private["id"] not in {
-            memory["id"] for memory in svc.timeline(secret, workspace="w", repo="r")["history"]
+            memory["id"]
+            for memory in svc.timeline(private_memory, workspace="w", repo="r")["history"]
         }
         assert private["id"] not in {
             memory["id"]