"""Strictly audit current raw-history image-only traces.""" from __future__ import annotations import argparse import json import math import re import sys from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Iterable, Mapping, Sequence from urllib.parse import parse_qsl, unquote, urlsplit REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from eviscene.orchestrator.source_access import ( # noqa: E402 SourceAccessPolicy, url_variants, ) from eviscene.orchestrator.source_provenance import domain_matches # noqa: E402 from eviscene.orchestrator.react_runtime import ( # noqa: E402 REACT_RUNTIME_SCHEMA_VERSION, REACT_RUNTIME_TOOLS, is_unified_react_runtime_budget_action, ) from eviscene.orchestrator.tool_result import parse_tool_result # noqa: E402 HARD = "hard" SCHEDULER = "scheduler" PROTOCOL = "protocol" ROUTE_CONTROL = "route_control" CORRECTION = "correction" BOUNDED_FALLBACK = "bounded_fallback" FORMAT = "format" REJECTION_ACTIONS = frozenset({"format_error", "output_rejected", "policy_replan"}) UNIFIED_REACT_POLICY_VERSION = "unified-react-v1" @dataclass(frozen=True) class Issue: code: str message: str category: str = HARD location: str = "" @dataclass class TraceReport: path: str image_id: str = "" termination: str = "" issues: list[Issue] = field(default_factory=list) stats: dict[str, int] = field(default_factory=dict) def failures(self, *, strict_scheduler: bool) -> list[Issue]: return [ issue for issue in self.issues if issue.category == HARD or (strict_scheduler and issue.category in {SCHEDULER, PROTOCOL}) ] def warnings(self, *, strict_scheduler: bool) -> list[Issue]: failed = set(self.failures(strict_scheduler=strict_scheduler)) return [issue for issue in self.issues if issue not in failed] def to_dict(self, *, strict_scheduler: bool) -> dict[str, Any]: failures = self.failures(strict_scheduler=strict_scheduler) warnings = self.warnings(strict_scheduler=strict_scheduler) return { "path": self.path, "image_id": self.image_id, "termination": self.termination, "passed": not failures, "failure_count": len(failures), "warning_count": len(warnings), "stats": self.stats, "issues": [ { **asdict(issue), "severity": "error" if issue in failures else "warning", } for issue in self.issues ], } def _mapping(value: Any) -> Mapping[str, Any]: return value if isinstance(value, Mapping) else {} def _rows(value: Any) -> list[Mapping[str, Any]]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, Mapping)] def _state(trace: Mapping[str, Any]) -> Mapping[str, Any]: state = trace.get("state") return state if isinstance(state, Mapping) else trace def _issue( report: TraceReport, code: str, message: str, *, category: str = HARD, location: str = "", ) -> None: report.issues.append( Issue(code=code, message=message, category=category, location=location) ) def _step_label(index: int, step: Mapping[str, Any]) -> str: call_id = str(_mapping(step.get("metadata")).get("function_call_id", "")).strip() suffix = f",call={call_id}" if call_id else "" return f"state.all_steps[{index}]({step.get('stage', '')}:{step.get('round', '')}{suffix})" def _audit_termination( trace: Mapping[str, Any], state: Mapping[str, Any], report: TraceReport ) -> None: outer = str(trace.get("termination", "")).strip() inner = str(state.get("termination", "")).strip() report.termination = outer or inner if outer != "success" or inner != "success": _issue( report, "TERMINATION_NOT_SUCCESS", f"termination must be success at trace and state levels (trace={outer!r}, state={inner!r})", location="termination", ) def _numeric_token(value: Any) -> float | None: if isinstance(value, bool): return None if isinstance(value, (int, float)): numeric = float(value) elif isinstance(value, str): try: numeric = float(value.strip()) except ValueError: return None else: return None return numeric if math.isfinite(numeric) else None def _audit_thought_tokens( trace: Mapping[str, Any], state: Mapping[str, Any], steps: Sequence[Mapping[str, Any]], report: TraceReport ) -> None: locations: list[tuple[str, Any, bool]] = [] for prefix, usage in ( ("token_usage", trace.get("token_usage")), ("state.token_usage", state.get("token_usage")), ): row = _mapping(usage) if "thought" in row: locations.append((f"{prefix}.thought", row.get("thought"), True)) if "thought_tokens" in row: locations.append( (f"{prefix}.thought_tokens", row.get("thought_tokens"), True) ) if "total_thought_tokens" in row: locations.append( ( f"{prefix}.total_thought_tokens", row.get("total_thought_tokens"), True, ) ) native_steps = 0 for index, step in enumerate(steps): metadata = _mapping(step.get("metadata")) tokens = _mapping(step.get("tokens")) if metadata.get("native_interactions"): native_steps += 1 if not any( key in tokens for key in ("thought", "thought_tokens", "total_thought_tokens") ): _issue( report, "THOUGHT_TOKENS_UNRECORDED", "Gemini Interactions step does not record a thought-token field", location=_step_label(index, step), ) for key in ("thought", "thought_tokens", "total_thought_tokens"): if key in tokens: locations.append( ( f"state.all_steps[{index}].tokens.{key}", tokens[key], True, ) ) recorded_paths = {path for path, _, _ in locations} for key, raw_value, path in _iter_named_values(trace): token_key = key.casefold() parent_segments = { segment.casefold() for segment in re.split(r"[.\[\]<>]+", path.rsplit(".", 1)[0]) if segment } is_token_field = token_key in {"thought_tokens", "total_thought_tokens"} is_usage_total = ( token_key == "thought" and bool(parent_segments & {"usage", "tokens", "token_usage"}) ) if (is_token_field or is_usage_total) and path not in recorded_paths: locations.append((path, raw_value, True)) recorded_paths.add(path) report.stats["native_interaction_steps"] = native_steps report.stats["thought_token_fields"] = len(locations) for path, value, nonzero_allowed in locations: numeric = _numeric_token(value) if numeric is None: _issue( report, "THOUGHT_TOKENS_INVALID", f"thought-token value must be numeric, got {value!r}", location=path, ) def _fact_check_domain( value: str, *, source_access_policy: SourceAccessPolicy | None = None, ) -> str: if source_access_policy is not None: for variant in url_variants(value): hostname = (urlsplit(variant).hostname or "").lower().rstrip(".") for domain in sorted( source_access_policy.excluded_domains, key=len, reverse=True, ): if domain_matches(hostname, domain): return domain return "" return "" def _fact_check_query_domain( value: str, *, source_access_policy: SourceAccessPolicy | None = None, ) -> str: text = unquote(str(value or "")) candidates = re.findall( r"(?:https?://)?(?:[a-z0-9-]+\.)+[a-z]{2,}(?:/[^\s\"'<>]*)?", text, flags=re.IGNORECASE, ) for candidate in candidates: if domain := _fact_check_domain( candidate.rstrip(".,;:!?)]}"), source_access_policy=source_access_policy, ): return domain return "" def _fact_check_embedded_url( value: str, *, source_access_policy: SourceAccessPolicy | None = None, ) -> str: urls = re.findall( r"https?://[^\s\"'<>]+", str(value or ""), flags=re.IGNORECASE, ) for url in urls: if domain := _fact_check_domain( url.rstrip(".,;:!?)]}"), source_access_policy=source_access_policy, ): return domain return "" def _fact_check_url_query( value: str, *, source_access_policy: SourceAccessPolicy | None = None, ) -> str: for variant in url_variants(value): try: query = " ".join( f"{key} {item}" for key, item in parse_qsl(urlsplit(variant).query) ) except ValueError: continue if domain := _fact_check_query_reference( query, source_access_policy=source_access_policy, ): return f"known fact-check domain {domain!r} in URL query" return "" def _fact_check_query_reference( value: str, *, source_access_policy: SourceAccessPolicy | None = None, ) -> str: if source_access_policy is None: return "" return ( source_access_policy.blocked_query_reference(value) or _fact_check_query_domain( value, source_access_policy=source_access_policy, ) ) def _iter_named_values( value: Any, path: str = "", inherited_key: str = "", ) -> Iterable[tuple[str, str, str]]: if isinstance(value, Mapping): for raw_key, child in value.items(): key = str(raw_key) child_path = f"{path}.{key}" if path else key if isinstance(child, (str, int, float)) and not isinstance(child, bool): if key == "tool_result" and isinstance(child, str): try: parsed = json.loads(child) except (TypeError, json.JSONDecodeError): parsed = None if isinstance(parsed, (Mapping, list)): yield from _iter_named_values( parsed, f"{child_path}", ) continue yield key, str(child), child_path else: yield from _iter_named_values(child, child_path, key) elif isinstance(value, list): for index, child in enumerate(value): child_path = f"{path}[{index}]" if isinstance(child, (str, int, float)) and not isinstance(child, bool): yield inherited_key, str(child), child_path else: yield from _iter_named_values(child, child_path, inherited_key) def _looks_like_url_key(key: str) -> bool: lowered = key.casefold() return lowered in { "source", "hostname", "registered_domain", "domain", "link", "href", "reference_image_candidates", } or "url" in lowered def _looks_like_query_key(key: str) -> bool: lowered = key.casefold() return lowered in {"query", "queries", "suggested_query", "suggested_queries", "crop_query", "vlm_query"} def _policy_rejected_step_for_path( trace: Mapping[str, Any], path: str, ) -> Mapping[str, Any] | None: """Return the step when a forbidden proposal was blocked before execution.""" match = re.search(r"(?:^|\.)all_steps\[(\d+)\](?:\.|$)", path) if match is None: return None steps = _rows(_state(trace).get("all_steps")) index = int(match.group(1)) if index >= len(steps): return None step = steps[index] metadata = _mapping(step.get("metadata")) if ( str(step.get("action_type", "")) in REJECTION_ACTIONS or metadata.get("search_policy_rejection") is True or int(metadata.get("policy_filtered_query_count", 0) or 0) > 0 ): return step return None def _audit_leaks( trace: Mapping[str, Any], report: TraceReport, *, enforce_source_access_policy: bool, source_access_policy: SourceAccessPolicy | None = None, ) -> None: seen: set[tuple[str, str]] = set() url_count = 0 query_count = 0 rejected_query_count = 0 for key, raw_value, path in _iter_named_values(trace): query_key = _looks_like_query_key(key) if enforce_source_access_policy and _looks_like_url_key(key): domain = _fact_check_domain( raw_value, source_access_policy=source_access_policy, ) or _fact_check_query_domain( raw_value, source_access_policy=source_access_policy, ) elif enforce_source_access_policy and not query_key: domain = _fact_check_embedded_url( raw_value, source_access_policy=source_access_policy, ) else: domain = "" if domain: signature = ("url", path) if signature not in seen: seen.add(signature) url_count += 1 _issue( report, "FACT_CHECK_URL_LEAK", f"known fact-check domain {domain!r} appears in URL-bearing trace data", location=path, ) if enforce_source_access_policy and _looks_like_url_key(key) and ( url_query_leak := _fact_check_url_query( raw_value, source_access_policy=source_access_policy, ) ): signature = ("query", path) if signature not in seen: seen.add(signature) if _policy_rejected_step_for_path(trace, path) is not None: rejected_query_count += 1 _issue( report, "FACT_CHECK_QUERY_REJECTED", f"{url_query_leak} was rejected before execution: {raw_value!r}", category=CORRECTION, location=path, ) else: query_count += 1 _issue( report, "FACT_CHECK_QUERY_LEAK", f"{url_query_leak} leaked into the trace: {raw_value!r}", location=path, ) if query_key: query_domain = ( _fact_check_query_reference( raw_value, source_access_policy=source_access_policy, ) if enforce_source_access_policy else "" ) if query_domain: signature = ("query", path) if signature not in seen: seen.add(signature) detail = f"excluded source {query_domain!r} is named in query" if _policy_rejected_step_for_path(trace, path) is not None: rejected_query_count += 1 _issue( report, "FACT_CHECK_QUERY_REJECTED", f"{detail} was rejected before execution: {raw_value!r}", category=CORRECTION, location=path, ) else: query_count += 1 _issue( report, "FACT_CHECK_QUERY_LEAK", f"{detail} leaked into the trace: {raw_value!r}", location=path, ) report.stats["fact_check_url_leaks"] = url_count report.stats["fact_check_query_leaks"] = query_count report.stats["fact_check_query_rejections"] = rejected_query_count def _rejection_category(step: Mapping[str, Any]) -> str: metadata = _mapping(step.get("metadata")) if metadata.get("search_query_format_error"): return FORMAT if metadata.get("search_policy_rejection"): return CORRECTION if _empty_text_search_query(step): return FORMAT if metadata.get("unbalanced_priority_coverage") or metadata.get("tool_budget_reached"): return SCHEDULER reason = str(metadata.get("rejection_reason", "")).casefold() if not reason: try: reason = str( _mapping(json.loads(str(step.get("tool_result", "")))).get("error", "") ).casefold() except (TypeError, json.JSONDecodeError): reason = "" if "source policy" in reason or "excluded source reference" in reason: return CORRECTION scheduler_markers = ( "coverage requires", "required investigation questions", "tool budget", "budget ended", "before resampling", "untouched p1", "untouched p2", "at least 1 tool call", ) if any(marker in reason for marker in scheduler_markers): return SCHEDULER return PROTOCOL def _empty_text_search_query(step: Mapping[str, Any]) -> bool: if str(step.get("tool_name", "")).strip() != "text_search": return False tool_args = _mapping(step.get("tool_args")) queries = tool_args.get("queries", tool_args.get("query", [])) if isinstance(queries, str): return not queries.strip() if not isinstance(queries, list): return True return not any(str(query).strip() for query in queries) def _audit_rejections( steps: Sequence[Mapping[str, Any]], report: TraceReport ) -> None: scheduler_count = 0 protocol_count = 0 route_control_count = 0 corrected_count = 0 bounded_fallback_count = 0 format_count = 0 def rejected(candidate: Mapping[str, Any]) -> bool: candidate_metadata = _mapping(candidate.get("metadata")) return ( str(candidate.get("action_type", "")) in REJECTION_ACTIONS or bool(candidate_metadata.get("rejection_reason")) ) def attempt_id(candidate: Mapping[str, Any]) -> str: candidate_metadata = _mapping(candidate.get("metadata")) if candidate_metadata.get("native_chat_completions"): return str( candidate_metadata.get("context_request_id", "") or candidate_metadata.get("interaction_id", "") ).strip() return str( candidate_metadata.get("interaction_id", "") or candidate_metadata.get("context_request_id", "") ).strip() def correction_parent_id(candidate: Mapping[str, Any]) -> str: candidate_metadata = _mapping(candidate.get("metadata")) if candidate_metadata.get("native_chat_completions"): return str( candidate_metadata.get("parent_context_request_id", "") or candidate_metadata.get("previous_interaction_id", "") ).strip() return str( candidate_metadata.get("previous_interaction_id", "") or candidate_metadata.get("parent_context_request_id", "") ).strip() def eventually_corrected( rejected_index: int, rejected_step: Mapping[str, Any], ) -> bool: root_id = attempt_id(rejected_step) if not root_id: return False reachable_rejected_ids = {root_id} stage = str(rejected_step.get("stage", "")) for candidate in steps[rejected_index + 1 :]: if str(candidate.get("stage", "")) != stage: continue candidate_metadata = _mapping(candidate.get("metadata")) lifecycle = str( candidate_metadata.get("interaction_lifecycle_kind", "") ).strip() is_unified_outer_correction = ( stage == "unified_react" and lifecycle == "tool_roundtrip" and str(candidate.get("action_type", "")) == "tool_call" and bool( _mapping( candidate_metadata.get("unified_react_delta") ).get("state_update", {}) ) ) if lifecycle != "protocol_correction" and not is_unified_outer_correction: continue if correction_parent_id(candidate) not in reachable_rejected_ids: continue if not rejected(candidate): return True candidate_id = attempt_id(candidate) if candidate_id: reachable_rejected_ids.add(candidate_id) return False def eventually_reaches_bounded_fallback( rejected_index: int, rejected_step: Mapping[str, Any], ) -> bool: root_id = attempt_id(rejected_step) if not root_id: return False stage = str(rejected_step.get("stage", "")) for candidate in steps[rejected_index + 1 :]: if str(candidate.get("stage", "")) != stage: continue candidate_metadata = _mapping(candidate.get("metadata")) if not candidate_metadata.get( "protocol_correction_exhaustion_boundary" ): continue resolved = { str(item).strip() for item in candidate_metadata.get( "resolved_rejection_request_ids", [] ) or [] if str(item).strip() } if root_id in resolved: return True return False for index, step in enumerate(steps): metadata = _mapping(step.get("metadata")) if not rejected(step): continue corrected = eventually_corrected(index, step) bounded_fallback = eventually_reaches_bounded_fallback(index, step) reason = str(metadata.get("rejection_reason", "")).strip() if not reason: try: payload = json.loads(str(step.get("tool_result", ""))) reason = str(_mapping(payload).get("error", "")).strip() except (TypeError, json.JSONDecodeError): reason = "" route_control = bool(metadata.get("duplicate_tool_call")) or ( bool(metadata.get("invalid_question_id")) and "already resolved" in reason.casefold() ) category = ( CORRECTION if corrected else BOUNDED_FALLBACK if bounded_fallback else ROUTE_CONTROL if route_control else _rejection_category(step) ) if category == CORRECTION: corrected_count += 1 elif category == BOUNDED_FALLBACK: bounded_fallback_count += 1 elif category == ROUTE_CONTROL: route_control_count += 1 elif category == FORMAT: format_count += 1 elif category == SCHEDULER: scheduler_count += 1 else: protocol_count += 1 _issue( report, ( "PROTOCOL_CORRECTION" if category == CORRECTION else "PROTOCOL_EXHAUSTION_BOUNDARY" if category == BOUNDED_FALLBACK else "ROUTE_CONTROL_REJECTION" if category == ROUTE_CONTROL else "TOOL_ARGUMENT_FORMAT_ERROR" if category == FORMAT else "SCHEDULER_REJECTION" if category == SCHEDULER else "PROTOCOL_REJECTION" ), reason or f"{step.get('action_type', 'rejected')} step", category=category, location=_step_label(index, step), ) report.stats["scheduler_rejections"] = scheduler_count report.stats["protocol_rejections"] = protocol_count report.stats["route_control_rejections"] = route_control_count report.stats["successful_protocol_corrections"] = corrected_count report.stats["bounded_protocol_fallbacks"] = bounded_fallback_count report.stats["tool_argument_format_errors"] = format_count def _audit_unified_react_trace( trace: Mapping[str, Any], state: Mapping[str, Any], steps: Sequence[Mapping[str, Any]], report: TraceReport, ) -> None: """Audit the one supported raw-history ReAct contract.""" if str( trace.get("decision_policy_version") or state.get("decision_policy_version") or "" ).strip() != UNIFIED_REACT_POLICY_VERSION: _issue( report, "REACT_RUNTIME_POLICY_MISMATCH", "current trace must use decision_policy_version=unified-react-v1", location="decision_policy_version", ) return investigation = _mapping(state.get("investigation_state")) if str(investigation.get("schema_version", "")).strip() != ( REACT_RUNTIME_SCHEMA_VERSION ): _issue( report, "REACT_RUNTIME_SCHEMA_INVALID", "current trace must use the raw-history runtime schema", location="state.investigation_state.schema_version", ) return _audit_current_react_runtime_trace(trace, state, steps, report) def _audit_current_react_runtime_trace( trace: Mapping[str, Any], state: Mapping[str, Any], steps: Sequence[Mapping[str, Any]], report: TraceReport, ) -> None: """Audit the active raw-history runtime without reducer semantics.""" investigation = _mapping(state.get("investigation_state")) allowed_state_fields = { "schema_version", "case_id", "image_sha256", "objective", "action_count", "stop_reason", "finish_rationale", } unexpected_state_fields = sorted(set(investigation) - allowed_state_fields) if unexpected_state_fields: _issue( report, "REACT_RUNTIME_STATE_FIELDS_INVALID", "raw-history ReAct state contains non-mechanical fields: " + ", ".join(unexpected_state_fields), location="state.investigation_state", ) actions = [ (index, step) for index, step in enumerate(steps) if str(step.get("stage", "")).strip() == "unified_react" and str(step.get("action_type", "")).strip() == "tool_call" ] if not actions: _issue( report, "REACT_RUNTIME_ACTIONS_MISSING", "current ReAct trace requires at least one accepted tool action", location="state.all_steps", ) accepted_tool_names = set(REACT_RUNTIME_TOOLS) | {"finish_investigation"} call_ids: set[str] = set() observation_rows: list[dict[str, str]] = [] successful_observation_ids: list[str] = [] successful_results = 0 error_results = 0 malformed_results = 0 native_interactions: list[tuple[int, Mapping[str, Any]]] = [] for index, step in actions: location = _step_label(index, step) tool_name = str(step.get("tool_name", "")).strip() metadata = _mapping(step.get("metadata")) call_id = str(metadata.get("function_call_id", "")).strip() if tool_name not in accepted_tool_names: _issue( report, "REACT_RUNTIME_TOOL_UNKNOWN", f"unknown current ReAct tool {tool_name!r}", location=location, ) if not call_id: _issue( report, "REACT_RUNTIME_CALL_ID_MISSING", "accepted current ReAct action lacks function_call_id", location=location, ) elif call_id in call_ids: _issue( report, "REACT_RUNTIME_CALL_ID_DUPLICATE", f"duplicate current ReAct function_call_id {call_id!r}", location=location, ) call_ids.add(call_id) policy_action = _mapping(metadata.get("policy_action")) if ( str(policy_action.get("type", "")).strip() != "tool_call" or str(policy_action.get("name", "")).strip() != tool_name ): _issue( report, "REACT_RUNTIME_POLICY_ACTION_MISMATCH", "policy_action must record the same current ReAct tool call", location=location, ) try: _payload, succeeded = parse_tool_result(str(step.get("tool_result", ""))) result_status = "success" if succeeded else "error" except Exception as exc: result_status = "malformed" _issue( report, "REACT_RUNTIME_TOOL_RESULT_INVALID", f"cannot parse tool result: {exc}", location=location, ) if result_status == "success": successful_results += 1 elif result_status == "error": error_results += 1 else: malformed_results += 1 if tool_name == "finish_investigation" or not call_id: continue observation_rows.append( { "observation_id": call_id, "function_call_id": call_id, "tool_name": tool_name, "status": result_status, } ) if result_status == "success": successful_observation_ids.append(call_id) # Include protocol-correction requests as well as accepted tool actions. # They are real provider interactions and must participate in the same # parent chain; otherwise a correction can make the next accepted action # look like a broken root. native_interactions = [ (index, _mapping(step.get("metadata"))) for index, step in enumerate(steps) if str(step.get("stage", "")).strip() == "unified_react" and _mapping(step.get("metadata")).get("native_interactions") ] budget_actions = [ step for _, step in actions if is_unified_react_runtime_budget_action(step) ] action_count = int(investigation.get("action_count", 0) or 0) if action_count != len(budget_actions): _issue( report, "REACT_RUNTIME_ACTION_COUNT_MISMATCH", ( f"action_count={action_count}, but trace records " f"{len(budget_actions)} budgeted ReAct actions" ), location="state.investigation_state.action_count", ) if action_count < 0 or action_count > 24: _issue( report, "REACT_RUNTIME_ACTION_BUDGET_INVALID", f"current ReAct action_count must be between 0 and 24, got {action_count}", location="state.investigation_state.action_count", ) interaction_rows: dict[str, tuple[int, Mapping[str, Any], Mapping[str, Any]]] = {} previous_interaction_id: str | None = None for index, metadata in native_interactions: interaction_id = str(metadata.get("interaction_id", "")).strip() parent_id = str(metadata.get("previous_interaction_id", "") or "").strip() lifecycle = str( metadata.get("interaction_lifecycle_kind", "") ).strip() location = _step_label(index, steps[index]) if not interaction_id: _issue( report, "REACT_RUNTIME_INTERACTION_ID_MISSING", "native current ReAct action lacks interaction_id", location=location, ) if lifecycle not in {"tool_roundtrip", "protocol_correction"}: _issue( report, "REACT_RUNTIME_INTERACTION_LIFECYCLE_INVALID", "native current ReAct interaction has an invalid lifecycle", location=location, ) if previous_interaction_id is None: if parent_id: _issue( report, "REACT_RUNTIME_INTERACTION_ROOT_INVALID", ( "the first unified ReAct interaction must have an empty " f"parent, got {parent_id!r}" ), location=location, ) elif parent_id != previous_interaction_id: _issue( report, "REACT_RUNTIME_INTERACTION_CHAIN_BROKEN", ( "expected previous_interaction_id " f"{previous_interaction_id!r}, got {parent_id!r}" ), location=location, ) previous_step = ( interaction_rows.get(parent_id, (None, None, None))[2] if parent_id else None ) if previous_step is not None and str( previous_step.get("action_type", "") ).strip() == "tool_call": previous_call_id = str( _mapping(previous_step.get("metadata")).get( "function_call_id", "", ) ).strip() policy_input = _mapping(metadata.get("policy_input")) input_payload = policy_input.get("input_payload") function_result_ids: list[str] = [] def collect_function_result_ids(value: Any) -> None: if isinstance(value, Mapping): if str(value.get("type", "")).strip() == "function_result": call_id = str(value.get("call_id", "")).strip() if call_id: function_result_ids.append(call_id) for child in value.values(): collect_function_result_ids(child) elif isinstance(value, list): for child in value: collect_function_result_ids(child) collect_function_result_ids(input_payload) if previous_call_id and previous_call_id not in function_result_ids: _issue( report, "REACT_RUNTIME_TOOL_RESULT_NOT_REINJECTED", ( "the request following a completed tool action does not " f"contain its function_result call_id {previous_call_id!r}" ), location=location, ) if interaction_id: interaction_rows[interaction_id] = ( index, metadata, steps[index], ) previous_interaction_id = interaction_id judgment_outputs = [ (index, step) for index, step in enumerate(steps) if str(step.get("stage", "")).strip() == "unified_judgment" and str(step.get("action_type", "")).strip() == "output" ] if len(judgment_outputs) != 1: _issue( report, "REACT_RUNTIME_JUDGMENT_COUNT_INVALID", f"current ReAct trace must contain exactly one final Judgment, got {len(judgment_outputs)}", location="state.all_steps", ) basis = _mapping(trace.get("verdict_basis")) if str(basis.get("schema_version", "")).strip() != ( "ifv-raw-history-judgment-basis-v1" ): _issue( report, "REACT_RUNTIME_BASIS_INVALID", "raw-history ReAct verdict basis has an invalid schema_version", location="verdict_basis.schema_version", ) if str(basis.get("decision_mode", "")).strip() != "bounded_binary_judgment": _issue( report, "REACT_RUNTIME_DECISION_MODE_INVALID", "current ReAct verdict basis must use bounded_binary_judgment", location="verdict_basis.decision_mode", ) basis_observation_ids = [ str(item).strip() for item in basis.get("observation_ids", []) or [] if str(item).strip() ] if basis_observation_ids != successful_observation_ids: _issue( report, "REACT_RUNTIME_BASIS_OBSERVATIONS_MISMATCH", "verdict basis observation_ids must equal successful raw observations", location="verdict_basis.observation_ids", ) basis_observation_rows = [ { "observation_id": str(item.get("observation_id", "")).strip(), "function_call_id": str(item.get("function_call_id", "")).strip(), "tool_name": str(item.get("tool_name", "")).strip(), "status": str(item.get("status", "")).strip(), } for item in _rows(basis.get("observations")) ] if basis_observation_rows != observation_rows: _issue( report, "REACT_RUNTIME_OBSERVATION_LOCATOR_MISMATCH", "verdict basis observations must match raw tool calls in order", location="verdict_basis.observations", ) if int(basis.get("action_count", 0) or 0) != action_count: _issue( report, "REACT_RUNTIME_BASIS_ACTION_COUNT_MISMATCH", "verdict basis action_count must equal investigation action_count", location="verdict_basis.action_count", ) judgment = _mapping(trace.get("judgment") or state.get("judgment")) verdict = str(trace.get("verdict", "")).strip() if verdict not in {"real", "fake"}: _issue( report, "REACT_RUNTIME_VERDICT_INVALID", f"current ReAct verdict must be real or fake, got {verdict!r}", location="verdict", ) if str(judgment.get("policy_rule_id", "")).strip() != ( UNIFIED_REACT_POLICY_VERSION ): _issue( report, "REACT_RUNTIME_JUDGMENT_POLICY_INVALID", "current ReAct Judgment must use unified-react-v1", location="judgment.policy_rule_id", ) if str(judgment.get("verdict", "")).strip() != verdict: _issue( report, "REACT_RUNTIME_JUDGMENT_VERDICT_MISMATCH", "trace verdict and current ReAct Judgment verdict must match", location="judgment.verdict", ) judgment_selected_observations = [ str(item).strip() for item in judgment.get("selected_observation_ids", []) or [] if str(item).strip() ] if judgment_selected_observations != basis_observation_ids: _issue( report, "REACT_RUNTIME_JUDGMENT_BASIS_MISMATCH", ( "Judgment selected_observation_ids must equal " "verdict_basis.observation_ids" ), location="judgment.selected_observation_ids", ) verdict_observation_ids = [ str(item).strip() for item in judgment.get("verdict_observation_ids", []) or [] if str(item).strip() ] if len(verdict_observation_ids) != len(set(verdict_observation_ids)): _issue( report, "REACT_RUNTIME_VERDICT_OBSERVATIONS_DUPLICATE", "Judgment verdict_observation_ids must be unique", location="judgment.verdict_observation_ids", ) unknown_verdict_observations = sorted( set(verdict_observation_ids) - set(successful_observation_ids) ) if unknown_verdict_observations: _issue( report, "REACT_RUNTIME_VERDICT_OBSERVATION_UNKNOWN", "Judgment cites unsuccessful or unknown observations: " + ", ".join(unknown_verdict_observations), location="judgment.verdict_observation_ids", ) if not isinstance(judgment.get("fact_check_report"), Mapping): _issue( report, "REACT_RUNTIME_REPORT_MISSING", "current ReAct Judgment must contain fact_check_report", location="judgment.fact_check_report", ) report.stats.update( { "unified_react_actions": action_count, "react_runtime_observations": len(observation_rows), "react_runtime_successful_observations": len( successful_observation_ids ), "react_runtime_success_results": successful_results, "react_runtime_error_results": error_results, "react_runtime_malformed_results": malformed_results, } ) def audit_trace( path: Path, *, enforce_source_access_policy: bool = True, source_access_policy: SourceAccessPolicy | None = None, payload: Mapping[str, Any] | None = None, ) -> TraceReport: report = TraceReport(path=str(path)) if payload is None: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: _issue(report, "TRACE_JSON_INVALID", f"cannot read trace JSON: {exc}") return report if not isinstance(payload, Mapping): _issue(report, "TRACE_ROOT_INVALID", "trace JSON root must be an object") return report state = _state(payload) report.image_id = str(payload.get("image_id", state.get("image_id", path.stem))) steps = _rows(state.get("all_steps")) is_image_only = str( payload.get("input_mode") or state.get("input_mode") or "" ) == "image_only" report.stats.update( { "steps": len(steps), } ) if "state" not in payload or not isinstance(payload.get("state"), Mapping): _issue(report, "CANONICAL_STATE_MISSING", "canonical trace must contain a state object") if not isinstance(state.get("all_steps"), list): _issue(report, "STEPS_MISSING", "canonical state must contain all_steps") if not is_image_only: _issue( report, "UNSUPPORTED_TRACE_MODE", "v3 strict audit accepts input_mode=image_only only", location="input_mode", ) return report _audit_termination(payload, state, report) _audit_thought_tokens(payload, state, steps, report) policy_version = str( payload.get("decision_policy_version") or state.get("decision_policy_version") or "" ) if policy_version == UNIFIED_REACT_POLICY_VERSION: _audit_unified_react_trace(payload, state, steps, report) else: _issue( report, "UNSUPPORTED_POLICY_VERSION", ( "current strict audit accepts only " f"{UNIFIED_REACT_POLICY_VERSION}; legacy policy traces are archived" ), location="decision_policy_version", ) _audit_leaks( payload, report, enforce_source_access_policy=enforce_source_access_policy, source_access_policy=source_access_policy, ) _audit_rejections(steps, report) return report def discover_trace_files(input_path: Path) -> list[Path]: if input_path.is_file(): return [ input_path ] if input_path.suffix.casefold() == ".json" and not _looks_like_benchmark_artifact( input_path ) else [] if not input_path.is_dir(): return [] if input_path.name.casefold() == "traces": return sorted( path for path in input_path.glob("*.json") if path.is_file() and not _looks_like_benchmark_artifact(path) ) traces_dirs = sorted( path for path in input_path.rglob("traces") if path.is_dir() ) if traces_dirs: return sorted( { trace for directory in traces_dirs for trace in directory.glob("*.json") if trace.is_file() and not _looks_like_benchmark_artifact(trace) } ) return sorted( path for path in input_path.glob("*.json") if path.is_file() and not _looks_like_benchmark_artifact(path) ) def _looks_like_benchmark_artifact(path: Path) -> bool: stem = path.stem.casefold() if stem in {"run_manifest", "summary", "predictions", "manifest"}: return True return bool(re.search(r"(?:^|[-_.])(benchmark|gold)(?:$|[-_.])", stem)) def _print_human(reports: Sequence[TraceReport], *, strict_scheduler: bool) -> None: for report in reports: failures = report.failures(strict_scheduler=strict_scheduler) warnings = report.warnings(strict_scheduler=strict_scheduler) status = "PASS" if not failures else "FAIL" print( f"[{status}] {report.path} " f"(termination={report.termination or 'unknown'}, failures={len(failures)}, warnings={len(warnings)})" ) print( " " f"steps={report.stats.get('steps', 0)} " f"claims={report.stats.get('claims', 0)} " f"evidence={report.stats.get('evidence', 0)} " f"scheduler_rejections={report.stats.get('scheduler_rejections', 0)} " f"protocol_rejections={report.stats.get('protocol_rejections', 0)} " f"route_control_rejections={report.stats.get('route_control_rejections', 0)}" ) failure_set = set(failures) for issue in report.issues: severity = "ERROR" if issue in failure_set else "WARN" where = f" at {issue.location}" if issue.location else "" print(f" {severity} {issue.code}{where}: {issue.message}") failed = sum( bool(report.failures(strict_scheduler=strict_scheduler)) for report in reports ) warnings = sum( len(report.warnings(strict_scheduler=strict_scheduler)) for report in reports ) scheduler_rejections = sum( report.stats.get("scheduler_rejections", 0) for report in reports ) protocol_rejections = sum( report.stats.get("protocol_rejections", 0) for report in reports ) route_control_rejections = sum( report.stats.get("route_control_rejections", 0) for report in reports ) print( f"Audited {len(reports)} trace(s): {len(reports) - failed} passed, " f"{failed} failed, {warnings} warning(s), " f"{scheduler_rejections} scheduler rejection(s), " f"{protocol_rejections} protocol rejection(s), " f"{route_control_rejections} route-control rejection(s)." ) def _json_summary( reports: Sequence[TraceReport], *, strict_scheduler: bool ) -> dict[str, Any]: failed = sum( bool(report.failures(strict_scheduler=strict_scheduler)) for report in reports ) return { "passed": failed == 0, "strict_scheduler": strict_scheduler, "trace_count": len(reports), "passed_count": len(reports) - failed, "failed_count": failed, "warning_count": sum( len(report.warnings(strict_scheduler=strict_scheduler)) for report in reports ), "scheduler_rejections": sum( report.stats.get("scheduler_rejections", 0) for report in reports ), "protocol_rejections": sum( report.stats.get("protocol_rejections", 0) for report in reports ), "route_control_rejections": sum( report.stats.get("route_control_rejections", 0) for report in reports ), "traces": [ report.to_dict(strict_scheduler=strict_scheduler) for report in reports ], } class _ArgumentParser(argparse.ArgumentParser): def error(self, message: str) -> None: self.print_usage(sys.stderr) self.exit(1, f"{self.prog}: error: {message}\n") def build_parser() -> argparse.ArgumentParser: parser = _ArgumentParser( description="Strictly audit canonical v3 image-only traces." ) parser.add_argument( "input_path", type=Path, help="A canonical trace JSON file or a directory containing traces.", ) parser.add_argument( "--json", action="store_true", dest="json_output", help="Emit one machine-readable JSON summary.", ) parser.add_argument( "--strict-scheduler", action="store_true", help="Treat runtime action/output rejections as failures instead of warnings.", ) parser.add_argument( "--source-access-policy", type=Path, help=( "Optional active source-access policy. Only sources listed in this " "policy are checked for query or URL leaks." ), ) return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) files = discover_trace_files(args.input_path) if not files: message = f"No trace JSON files found at: {args.input_path}" if args.json_output: print( json.dumps( { "passed": False, "strict_scheduler": args.strict_scheduler, "trace_count": 0, "passed_count": 0, "failed_count": 0, "warning_count": 0, "error": message, "traces": [], }, ensure_ascii=False, indent=2, ) ) else: print(message, file=sys.stderr) return 1 source_access_policy = ( SourceAccessPolicy.load(args.source_access_policy) if args.source_access_policy is not None else None ) reports = [ audit_trace(path, source_access_policy=source_access_policy) for path in files ] if args.json_output: print( json.dumps( _json_summary(reports, strict_scheduler=args.strict_scheduler), ensure_ascii=False, indent=2, ) ) else: _print_human(reports, strict_scheduler=args.strict_scheduler) return int( any( report.failures(strict_scheduler=args.strict_scheduler) for report in reports ) ) if __name__ == "__main__": raise SystemExit(main())