""" Inference Script Example =================================== MANDATORY - Before submitting, ensure the following variables are defined in your environment configuration: API_BASE_URL The API endpoint for the LLM. MODEL_NAME The model identifier to use for inference. HF_TOKEN Your Hugging Face / API key. LOCAL_IMAGE_NAME The name of the local image to use for the environment if you are using from_docker_image(). - Defaults are set only for API_BASE_URL and MODEL_NAME (and should reflect your active inference setup): API_BASE_URL = os.getenv("API_BASE_URL", "") MODEL_NAME = os.getenv("MODEL_NAME", "") - The inference script must be named inference.py and placed in the root directory of the project. - Participants must use OpenAI Client for all LLM calls using the variables above. - I've read the sample inference.py and have followed it strictly. STDOUT FORMAT - The script emits exactly three line types to stdout, in this order: [START] task= env= model= [STEP] step= action= reward=<0.00> done= error= [END] success= steps= score= rewards= Rules: - One [START] line at episode begin. - One [STEP] line per step, immediately after env.step() returns. - One [END] line after env.close(), always emitted even on exception. - reward and rewards are formatted to 2 decimal places. - done and success are lowercase booleans: true or false. - error is the raw last_action_error string, or null if none. - All fields are on a single line with no newlines within a line. - Each task should return score strictly between 0 and 1, not 0.00 and not 1.00. Example: [START] task=click-test env=miniwob model=Qwen3-VL-30B [STEP] step=1 action=click('123') reward=0.00 done=false error=null [STEP] step=2 action=fill('456','text') reward=0.00 done=false error=null [STEP] step=3 action=click('789') reward=0.99 done=true error=null [END] success=true steps=3 score=0.99 rewards=0.00,0.00,0.99 """ from __future__ import annotations import os import sys if os.getenv("LEDGERSHIELD_DEBUG") != "1": sys.stderr = open(os.devnull, "w", encoding="utf-8") import argparse from collections import Counter from dataclasses import asdict, dataclass from datetime import datetime, timezone import json from pathlib import Path import re import warnings from typing import Any, Optional warnings.filterwarnings("ignore", category=DeprecationWarning) with warnings.catch_warnings(): warnings.simplefilter("ignore", DeprecationWarning) from openai import OpenAI from ledgershield_env import LedgerShieldAction, LedgerShieldEnv from openenv_compat import StepResult from llm_utils import create_json_chat_completion, parse_json_dict from server.environment import LedgerShieldEnvironment from server.schema import canonical_reason_codes from task_c_guardrails import ( grounded_task_c_submission, sanitize_task_c_submission, validate_task_c_submission, ) from task_d_guardrails import ( derive_email_thread_signals as _derive_email_thread_signals, grounded_task_d_submission, policy_check_payload as _policy_check_payload, sanitize_task_d_submission, validate_task_d_submission, ) # Required runtime configuration. Only the endpoint/model have defaults. API_BASE_URL = os.getenv("API_BASE_URL", "https://api.openai.com/v1") MODEL_NAME = os.getenv("MODEL_NAME", "gpt-5.4") HF_TOKEN = os.getenv("HF_TOKEN") # Optional when running with from_docker_image(). LOCAL_IMAGE_NAME = os.getenv("LOCAL_IMAGE_NAME") ENV_URL = os.getenv("ENV_URL") or "http://localhost:8000" BENCHMARK = "ledgershield" MAX_STEPS = 20 TEMPERATURE = 0.0 MAX_TOKENS = 512 SUCCESS_SCORE_THRESHOLD = 0.85 PASSK_SUCCESS_THRESHOLD = 0.85 TASK_SCORE_MIN = 0.01 TASK_SCORE_MAX = 0.99 ARTIFACT_DIR = Path("artifacts") DEFAULT_CASES = [ "CASE-A-001", "CASE-A-002", "CASE-A-003", "CASE-A-004", "CASE-B-001", "CASE-B-002", "CASE-B-003", "CASE-B-004", "CASE-B-005", "CASE-C-001", "CASE-C-002", "CASE-C-003", "CASE-C-004", "CASE-D-001", "CASE-D-002", "CASE-D-003", "CASE-D-004", "CASE-D-005", "CASE-D-006", "CASE-E-001", "CASE-E-002", ] @dataclass(frozen=True) class ModelCapabilityProfile: model_name: str capability_score: float tier: str plan_mode: str repair_level: str investigation_budget_bonus: int intervention_budget_bonus: int decision_token_budget: int planning_token_budget: int def _base_model_score(model_name: str) -> float: normalized = normalize_text(model_name) if "gpt-4o" in normalized: base = 4.6 else: match = re.search(r"gpt-([0-9]+(?:\.[0-9]+)?)", normalized) if match: base = float(match.group(1)) elif "gpt-4" in normalized: base = 4.0 elif "gpt-3.5" in normalized: base = 3.5 else: base = 4.0 if "pro" in normalized: base += 0.5 if "latest" in normalized: base += 0.2 if "mini" in normalized: base -= 0.7 if "nano" in normalized: base -= 1.1 if "turbo" in normalized: base -= 0.3 return base def get_model_capability_profile(model_name: str) -> ModelCapabilityProfile: score = _base_model_score(model_name) if score >= 5.0: return ModelCapabilityProfile( model_name=model_name, capability_score=score, tier="elite", plan_mode="llm", repair_level="partial", investigation_budget_bonus=2, intervention_budget_bonus=2, decision_token_budget=max(MAX_TOKENS, 1536), planning_token_budget=640, ) if score >= 4.5: return ModelCapabilityProfile( model_name=model_name, capability_score=score, tier="strong", plan_mode="hybrid", repair_level="partial", investigation_budget_bonus=1, intervention_budget_bonus=1, decision_token_budget=max(MAX_TOKENS, 1280), planning_token_budget=512, ) return ModelCapabilityProfile( model_name=model_name, capability_score=score, tier="standard", plan_mode="llm", repair_level="none", investigation_budget_bonus=0, intervention_budget_bonus=0, decision_token_budget=MAX_TOKENS, planning_token_budget=384, ) def current_model_profile() -> ModelCapabilityProfile: return get_model_capability_profile(MODEL_NAME) def normalize_text(value: Any) -> str: if value is None: return "" return " ".join(str(value).strip().lower().split()) def safe_float(value: Any) -> float: try: if isinstance(value, str): cleaned = ( value.replace(",", "") .replace("$", "") .replace("€", "") .replace("₹", "") .strip() ) return float(cleaned) return float(value) except Exception: return 0.0 def clamp(value: float, lo: float, hi: float) -> float: return max(lo, min(hi, value)) def normalize_score(value: Any) -> float: return clamp(safe_float(value), TASK_SCORE_MIN, TASK_SCORE_MAX) def format_decimal(value: Any) -> str: numeric = safe_float(value) if abs(numeric) < 0.005: numeric = 0.0 return f"{numeric:.2f}" def compact_json(value: Any) -> str: return json.dumps(value, separators=(",", ":"), ensure_ascii=True, sort_keys=True) def sanitize_log_field(value: Any) -> str: if value is None: return "null" text = " ".join(str(value).split()) return text if text else "null" def format_action(action: LedgerShieldAction) -> str: return f"{action.action_type}({compact_json(action.payload)})" def log_start(task: str, env: str, model: str) -> None: print(f"[START] task={task} env={env} model={sanitize_log_field(model)}", flush=True) def log_step(step: int, action: str, reward: float, done: bool, error: Optional[str]) -> None: print( "[STEP] " f"step={step} " f"action={sanitize_log_field(action)} " f"reward={format_decimal(reward)} " f"done={str(done).lower()} " f"error={sanitize_log_field(error)}", flush=True, ) def log_end(success: bool, steps: int, rewards: list[float], score: Optional[float] = None) -> None: final_score = rewards[-1] if score is None and rewards else (0.0 if score is None else score) rewards_str = ",".join(format_decimal(reward) for reward in rewards) print( "[END] " f"success={str(success).lower()} " f"steps={steps} " f"score={format_decimal(normalize_score(final_score))} " f"rewards={rewards_str}", flush=True, ) def trace(message: str) -> None: if os.getenv("LEDGERSHIELD_DEBUG") == "1": print(message, file=sys.stderr, flush=True) def token_ref(token: dict[str, Any], doc_id: str) -> dict[str, Any]: return { "doc_id": doc_id, "page": int(token.get("page", 1) or 1), "bbox": token.get("bbox", []), "token_ids": [str(token.get("token_id", ""))], } def _looks_like_token_ref(value: Any) -> bool: return isinstance(value, dict) and {"doc_id", "page", "bbox", "token_ids"} <= set(value) def parse_invoice_tokens(tokens: list[dict[str, Any]], doc_id: str) -> tuple[dict[str, Any], dict[str, Any], list[dict[str, Any]]]: fields: dict[str, Any] = {} evidence: dict[str, Any] = {} line_items: list[dict[str, Any]] = [] for idx, token in enumerate(tokens): text = str(token.get("text", "")).strip() lower = text.lower() if idx == 0 and text: fields["vendor_name"] = text evidence["vendor_name"] = token_ref(token, doc_id) continue matchers = [ ("invoice_number", r"invoice\s*no\s*:\s*(.+)$"), ("invoice_date", r"invoice\s*date\s*:\s*(.+)$"), ("currency", r"currency\s*:\s*(.+)$"), ("po_id", r"po\s*:\s*(.+)$"), ("receipt_id", r"receipt\s*:\s*(.+)$"), ("bank_account", r"bank\s*:\s*(.+)$"), ] matched_field = False for key, pattern in matchers: match = re.match(pattern, text, flags=re.IGNORECASE) if match: fields[key] = match.group(1).strip() evidence[key] = token_ref(token, doc_id) matched_field = True break if matched_field: continue numeric_fields = [ ("subtotal", r"subtotal\s*:\s*([\d,]+(?:\.\d+)?)$"), ("tax", r"tax\s*:\s*([\d,]+(?:\.\d+)?)$"), ("total", r"total\s*:\s*([\d,]+(?:\.\d+)?)$"), ] for key, pattern in numeric_fields: match = re.match(pattern, text, flags=re.IGNORECASE) if match: fields[key] = safe_float(match.group(1)) evidence[key] = token_ref(token, doc_id) matched_field = True break if matched_field: continue if "|" not in text: continue parts = [part.strip() for part in text.split("|")] if len(parts) != 4: continue description = parts[0] qty_value = safe_float(parts[1]) unit_price = safe_float(parts[2]) line_total = safe_float(parts[3]) qty = int(qty_value) if qty_value.is_integer() else qty_value line_items.append( { "description": description, "qty": qty, "unit_price": unit_price, "line_total": line_total, } ) return fields, evidence, line_items def parse_email_tokens(tokens: list[dict[str, Any]], doc_id: str) -> dict[str, Any]: evidence: dict[str, Any] = {} for token in tokens: text = str(token.get("text", "")).strip() lower = text.lower() if lower.startswith("from:"): evidence["from_header"] = token_ref(token, doc_id) elif lower.startswith("subject:"): evidence["subject_header"] = token_ref(token, doc_id) elif "approval threshold" in lower or "split the request" in lower or "split invoice" in lower: evidence["approval_threshold_evasion"] = token_ref(token, doc_id) elif ( "skip callback" in lower or "do not call" in lower or "don't call" in lower or "override policy" in lower or "bypass policy" in lower or "do not verify" in lower or "source of truth" in lower or "avoid reapproval" in lower ): evidence["policy_bypass_attempt"] = token_ref(token, doc_id) return evidence def _domain_tokens(domain: str) -> set[str]: return {part for part in re.split(r"[^a-z0-9]+", normalize_text(domain)) if len(part) > 1} def _vendor_name_tokens(vendor_name: str) -> set[str]: stop_words = { "and", "co", "company", "corp", "corporation", "gmbh", "inc", "industries", "industrial", "limited", "ltd", "manufacturing", "pvt", "supplies", } tokens = re.split(r"[^a-z0-9]+", normalize_text(vendor_name)) return {token for token in tokens if len(token) > 2 and token not in stop_words} def _sender_domain(sender: str) -> str: sender_text = normalize_text(sender) if "@" not in sender_text: return "" return sender_text.split("@", 1)[-1] def _infer_domain_alignment(*, sender: str, vendor_name: str, approved_domains: list[str]) -> tuple[str, str]: sender_domain = _sender_domain(sender) normalized_domains = [normalize_text(domain) for domain in approved_domains if normalize_text(domain)] expected_domain = normalized_domains[0] if normalized_domains else "" if sender_domain and expected_domain: return ("aligned" if sender_domain == expected_domain else "mismatch", expected_domain) vendor_tokens = _vendor_name_tokens(vendor_name) domain_tokens = _domain_tokens(sender_domain) if vendor_tokens and domain_tokens and vendor_tokens & domain_tokens: return "aligned", expected_domain if sender_domain and vendor_tokens: return "mismatch", expected_domain return "aligned", expected_domain def derive_email_thread_from_ocr(collected: dict[str, Any]) -> dict[str, Any]: tokens = list(collected.get("email_tokens", []) or []) if not tokens: return {} lines = [str(token.get("text", "")).strip() for token in tokens if str(token.get("text", "")).strip()] sender = "" subject = "" body_lines: list[str] = [] for line in lines: lower = line.lower() if lower.startswith("from:"): sender = line.split(":", 1)[-1].strip() continue if lower.startswith("subject:"): subject = line.split(":", 1)[-1].strip() continue body_lines.append(line) vendor = collected.get("vendor") or {} vendor_name = str((vendor or {}).get("vendor_name") or collected.get("invoice_fields", {}).get("vendor_name") or "") approved_domains = [str(domain).strip() for domain in (vendor.get("approved_domains", []) or []) if str(domain).strip()] domain_alignment, expected_domain = _infer_domain_alignment( sender=sender, vendor_name=vendor_name, approved_domains=approved_domains, ) body_text = " ".join(body_lines) normalized_body = normalize_text(body_text) normalized_subject = normalize_text(subject) explicit_no_change = any( phrase in normalized_body for phrase in { "no bank change", "no bank changes", "no change to bank", "approved remittance instructions already on file", } ) bank_change_language = ( "bank" in normalized_body and any(word in normalized_body for word in {"change", "changed", "update", "updated", "override", "directed"}) and not explicit_no_change ) urgency_language = any( phrase in normalized_subject or phrase in normalized_body for phrase in {"urgent", "immediately", "asap", "today"} ) callback_discouraged = any( phrase in normalized_body for phrase in {"skip callback", "do not call", "don't call", "do not verify"} ) policy_override_language = any( phrase in normalized_body for phrase in { "override policy", "bypass policy", "source of truth", "avoid reapproval", "do not contact", "skip normal review", "ignore the standard workflow", "personally approved", } ) suspicious_bank_change = bank_change_language and ( domain_alignment == "mismatch" or callback_discouraged or policy_override_language or urgency_language ) derived_flags = canonical_reason_codes( [ "sender_domain_spoof" if domain_alignment == "mismatch" else "", "bank_override_attempt" if suspicious_bank_change else "", "policy_bypass_attempt" if callback_discouraged or policy_override_language else "", "urgent_payment_pressure" if urgency_language else "", "approval_threshold_evasion" if any(phrase in normalized_body for phrase in {"approval threshold", "split the request", "split invoice"}) else "", ] ) return { "thread_id": collected.get("email_doc_id") or collected.get("case_metadata", {}).get("thread_id"), "vendor_key": (vendor or {}).get("vendor_key"), "sender": sender, "subject": subject, "body": body_text, "message_count": max(1, len(body_lines)), "sender_profile": { "from_domain": _sender_domain(sender), "expected_domain": expected_domain, "domain_alignment": domain_alignment, }, "request_signals": { "bank_change_language": bank_change_language, "urgency_language": urgency_language, "callback_discouraged": callback_discouraged, "policy_override_language": policy_override_language, }, "derived_flags": derived_flags, } def refresh_email_thread_from_ocr(collected: dict[str, Any]) -> None: derived = derive_email_thread_from_ocr(collected) if not derived: return current = collected.get("email_thread") or {} merged = dict(current) for key, value in derived.items(): if key in {"sender_profile", "request_signals"}: existing = current.get(key, {}) or {} candidate = value if isinstance(value, dict) else {} merged[key] = {**candidate, **existing} if existing else candidate continue if current.get(key) not in (None, "", [], {}): merged[key] = current.get(key) continue merged[key] = value merged["flags"] = canonical_reason_codes( (current.get("flags", []) or []) + (current.get("derived_flags", []) or []) + (derived.get("flags", []) or []) + (derived.get("derived_flags", []) or []) ) merged["derived_flags"] = canonical_reason_codes( (current.get("derived_flags", []) or []) + (derived.get("derived_flags", []) or []) ) collected["email_thread"] = merged def derive_email_thread_signals(thread: dict[str, Any]) -> set[str]: return _derive_email_thread_signals(thread) def vendor_key_for(fields: dict[str, Any]) -> str: return normalize_text(fields.get("vendor_name")) def vendor_history_key_for(fields: dict[str, Any]) -> str: explicit = vendor_key_for(fields) if explicit: return explicit vendor_name = normalize_text(fields.get("vendor_name")) if not vendor_name: return "" return re.sub(r"[^a-z0-9]+", "-", vendor_name).strip("-") def action_signature(action: LedgerShieldAction) -> str: return f"{normalize_text(action.action_type)}:{compact_json(action.payload)}" def _append_candidate( candidates: list[LedgerShieldAction], seen_signatures: set[str], action: LedgerShieldAction, ) -> None: signature = action_signature(action) if signature in seen_signatures: return seen_signatures.add(signature) candidates.append(action) def _invoice_record_summary(record: dict[str, Any]) -> dict[str, Any]: fields = record.get("fields", {}) or {} return { "doc_id": record.get("doc_id"), "invoice_number": fields.get("invoice_number"), "invoice_date": fields.get("invoice_date"), "total": fields.get("total"), "bank_account": fields.get("bank_account"), } def summarize_collected_state(collected: dict[str, Any]) -> dict[str, Any]: email_thread = collected.get("email_thread") or {} ledger_search = collected.get("ledger_search") or {} return { "case_instruction": str(collected.get("case_instruction", "") or ""), "invoice_records": [_invoice_record_summary(record) for record in collected.get("invoice_records", []) or []], "email_thread": { "sender": email_thread.get("sender"), "subject": email_thread.get("subject"), "flags": email_thread.get("flags"), "derived_flags": email_thread.get("derived_flags"), "sender_profile": email_thread.get("sender_profile"), "request_signals": email_thread.get("request_signals"), }, "vendor": { "vendor_key": (collected.get("vendor") or {}).get("vendor_key"), "approved_domains": (collected.get("vendor") or {}).get("approved_domains"), }, "bank_compares": [ { "proposed_bank_account": compare.get("proposed_bank_account"), "matched": compare.get("matched"), "vendor_bank_account": compare.get("vendor_bank_account"), } for compare in collected.get("bank_compares", []) or [] ], "ledger_search": { "exact_duplicate_count": ledger_search.get("exact_duplicate_count"), "near_duplicate_count": ledger_search.get("near_duplicate_count"), "top_hits": [ { "ledger_id": hit.get("ledger_id"), "invoice_number": hit.get("invoice_number"), "amount": hit.get("amount"), "match_score": hit.get("match_score"), } for hit in (collected.get("ledger_hits", []) or [])[:5] ], }, "vendor_history": [ { "event_type": row.get("event_type") or row.get("change_type"), "status": row.get("status"), } for row in (collected.get("vendor_history", []) or [])[:5] ], "has_policy_snapshot": bool(collected.get("policies")), "email_ocr_loaded": bool(collected.get("email_tokens")), "revealed_artifacts": sorted((collected.get("revealed_artifacts", {}) or {}).keys()), "pending_event_count": len(collected.get("pending_events", []) or []), "observed_risk_signals": collected.get("observed_risk_signals", []) or [], "tool_failures": { tool_name: len(entries) for tool_name, entries in (collected.get("tool_failures", {}) or {}).items() }, } def _store_artifact(collected: dict[str, Any], artifact: dict[str, Any]) -> None: artifact_id = normalize_text(artifact.get("artifact_id")) if not artifact_id: return artifact_store = collected.setdefault("revealed_artifacts", {}) artifact_store[artifact_id] = artifact if artifact_id == "callback_verification_result": collected["callback_result"] = artifact elif artifact_id == "bank_change_approval_chain": collected["bank_change_approval_chain"] = artifact elif artifact_id == "po_reconciliation_report": collected["po_reconciliation_report"] = artifact elif artifact_id == "receipt_reconciliation_report": collected["receipt_reconciliation_report"] = artifact elif artifact_id == "duplicate_cluster_report": collected["duplicate_cluster_report"] = artifact def _record_tool_failure(collected: dict[str, Any], action: LedgerShieldAction, tool: dict[str, Any]) -> None: tool_name = normalize_text(tool.get("tool_name") or action.action_type) failure_log = collected.setdefault("tool_failures", {}) failure_log.setdefault(tool_name, []).append( { "payload": dict(action.payload), "error": str(tool.get("error") or ""), } ) def _capture_tool_artifacts(collected: dict[str, Any], tool: dict[str, Any]) -> None: if isinstance(tool.get("artifact"), dict): _store_artifact(collected, tool["artifact"]) for artifact in tool.get("async_artifacts", []) or []: if isinstance(artifact, dict): _store_artifact(collected, artifact) for artifact_id in tool.get("revealed_artifact_ids", []) or []: normalized_id = normalize_text(artifact_id) if normalized_id: collected.setdefault("revealed_artifacts", {}).setdefault( normalized_id, {"artifact_id": artifact_id}, ) def update_collected_from_observation(collected: dict[str, Any], observation: Any) -> None: revealed_artifacts = list(getattr(observation, "revealed_artifacts", []) or []) pending_events = list(getattr(observation, "pending_events", []) or []) case_metadata = getattr(observation, "case_metadata", {}) or {} risk_snapshot = getattr(observation, "risk_snapshot", {}) or {} collected["case_metadata"] = dict(case_metadata) collected["pending_events"] = pending_events collected["observed_risk_signals"] = list(risk_snapshot.get("observed_signals", []) or []) if revealed_artifacts: for artifact in revealed_artifacts: if isinstance(artifact, dict): _store_artifact(collected, artifact) def build_investigation_candidates( task_type: str, collected: dict[str, Any], *, vendor_key: str, po_id: str, receipt_id: str, invoice_total: float, invoice_number: str, proposed_bank_account: str, email_doc_id: str, executed_signatures: set[str], ) -> list[LedgerShieldAction]: candidates: list[LedgerShieldAction] = [] seen = set(executed_signatures) history_lookup_key = vendor_key or vendor_history_key_for(collected.get("invoice_fields", {}) or {}) instruction = normalize_text(collected.get("case_instruction", "")) task_c_policy_hint = any( phrase in instruction for phrase in { "approval threshold", "structured below", "split invoice", "policy", } ) if task_type == "task_b": _append_candidate(candidates, seen, LedgerShieldAction(action_type="lookup_policy", payload={})) if po_id: _append_candidate(candidates, seen, LedgerShieldAction(action_type="lookup_po", payload={"po_id": po_id})) resolved_receipt_id = receipt_id or po_id.replace("PO-", "GRN-", 1) if resolved_receipt_id: _append_candidate( candidates, seen, LedgerShieldAction(action_type="lookup_receipt", payload={"receipt_id": resolved_receipt_id}), ) return candidates if task_type == "task_c": if vendor_key: _append_candidate( candidates, seen, LedgerShieldAction(action_type="lookup_vendor", payload={"vendor_key": vendor_key}), ) if history_lookup_key: _append_candidate( candidates, seen, LedgerShieldAction(action_type="lookup_vendor_history", payload={"vendor_key": history_lookup_key}), ) if task_c_policy_hint: _append_candidate(candidates, seen, LedgerShieldAction(action_type="lookup_policy", payload={})) if vendor_key and (invoice_number or invoice_total): _append_candidate( candidates, seen, LedgerShieldAction( action_type="search_ledger", payload={ "vendor_key": vendor_key, "invoice_number": invoice_number, "amount": invoice_total, }, ), ) if invoice_number or invoice_total: _append_candidate( candidates, seen, LedgerShieldAction( action_type="search_ledger", payload={ "invoice_number": invoice_number, "amount": invoice_total, }, ), ) if vendor_key and proposed_bank_account: _append_candidate( candidates, seen, LedgerShieldAction( action_type="compare_bank_account", payload={"vendor_key": vendor_key, "proposed_bank_account": proposed_bank_account}, ), ) return candidates if email_doc_id and not collected.get("email_tokens"): _append_candidate( candidates, seen, LedgerShieldAction(action_type="ocr", payload={"doc_id": email_doc_id, "mode": "accurate"}), ) if vendor_key: _append_candidate( candidates, seen, LedgerShieldAction(action_type="lookup_vendor", payload={"vendor_key": vendor_key}), ) if email_doc_id: _append_candidate( candidates, seen, LedgerShieldAction(action_type="inspect_email_thread", payload={"thread_id": email_doc_id}), ) if history_lookup_key: _append_candidate( candidates, seen, LedgerShieldAction(action_type="lookup_vendor_history", payload={"vendor_key": history_lookup_key}), ) _append_candidate(candidates, seen, LedgerShieldAction(action_type="lookup_policy", payload={})) if po_id: _append_candidate(candidates, seen, LedgerShieldAction(action_type="lookup_po", payload={"po_id": po_id})) resolved_receipt_id = receipt_id or (po_id.replace("PO-", "GRN-", 1) if po_id else "") if resolved_receipt_id: _append_candidate( candidates, seen, LedgerShieldAction(action_type="lookup_receipt", payload={"receipt_id": resolved_receipt_id}), ) for record in collected.get("invoice_records", []) or []: fields = record.get("fields", {}) or {} record_bank = str(fields.get("bank_account", "")).strip() record_invoice_number = str(fields.get("invoice_number", "")).strip() record_total = safe_float(fields.get("total")) if vendor_key and record_bank: _append_candidate( candidates, seen, LedgerShieldAction( action_type="compare_bank_account", payload={"vendor_key": vendor_key, "proposed_bank_account": record_bank}, ), ) if vendor_key and (record_invoice_number or record_total): _append_candidate( candidates, seen, LedgerShieldAction( action_type="search_ledger", payload={ "vendor_key": vendor_key, "invoice_number": record_invoice_number, "amount": record_total, }, ), ) elif record_invoice_number or record_total: _append_candidate( candidates, seen, LedgerShieldAction( action_type="search_ledger", payload={ "invoice_number": record_invoice_number, "amount": record_total, }, ), ) return candidates def build_intervention_candidates( task_type: str, collected: dict[str, Any], submission: dict[str, Any], *, executed_signatures: set[str], ) -> list[LedgerShieldAction]: decision = normalize_text(submission.get("decision")) if decision not in {"hold", "needs_review", "escalate_fraud"}: return [] candidates: list[LedgerShieldAction] = [] seen = set(executed_signatures) reason_codes = set( canonical_reason_codes( (submission.get("reason_codes", []) or []) + (submission.get("fraud_flags", []) or []) + (submission.get("campaign_signals", []) or []) ) ) ledger_search = collected.get("ledger_search") or {} ledger_reviewed = bool(ledger_search) has_duplicates = bool(collected.get("ledger_hits")) or int(ledger_search.get("exact_duplicate_count", 0) or 0) > 0 has_duplicates = has_duplicates or int(ledger_search.get("near_duplicate_count", 0) or 0) > 0 bank_mismatch = any(compare and not bool(compare.get("matched")) for compare in collected.get("bank_compares", []) or []) email_thread = collected.get("email_thread") or {} email_flags = set( canonical_reason_codes((email_thread.get("flags", []) or []) + (email_thread.get("derived_flags", []) or [])) ) observed_signals = {normalize_text(signal) for signal in collected.get("observed_risk_signals", []) or []} instruction = normalize_text(collected.get("case_instruction", "")) task_c_campaign_hint = any( phrase in instruction for phrase in { "cross-vendor", "coordinated fraud", "coordinated payment", "similar amounts and timing", "approval threshold", "structured below", "split invoice", } ) if task_type == "task_b": if str((collected.get("invoice_fields", {}) or {}).get("po_id", "")).strip(): _append_candidate(candidates, seen, LedgerShieldAction(action_type="request_po_reconciliation", payload={})) if str((collected.get("invoice_fields", {}) or {}).get("receipt_id", "")).strip() or ( normalize_text(collected.get("case_instruction", "")) and not collected.get("receipt") ): _append_candidate( candidates, seen, LedgerShieldAction(action_type="request_additional_receipt_evidence", payload={}), ) if normalize_text(submission.get("decision")) == "hold" or not collected.get("po") or not collected.get("receipt"): _append_candidate(candidates, seen, LedgerShieldAction(action_type="request_callback_verification", payload={})) return candidates needs_callback = ( bank_mismatch or "bank_override_attempt" in reason_codes or has_duplicates or task_c_campaign_hint or {"approval_threshold_evasion", "shared_bank_account", "coordinated_timing"} & reason_codes or {"sender_domain_spoof", "policy_bypass_attempt", "urgent_payment_pressure"} & (reason_codes | email_flags) or {"callback_suspicious_confirm", "callback_dispute_confirmed"} & observed_signals ) if needs_callback: _append_candidate(candidates, seen, LedgerShieldAction(action_type="request_callback_verification", payload={})) if bank_mismatch or "bank_override_attempt" in reason_codes: _append_candidate(candidates, seen, LedgerShieldAction(action_type="request_callback_verification", payload={})) _append_candidate( candidates, seen, LedgerShieldAction(action_type="request_bank_change_approval_chain", payload={}), ) _append_candidate(candidates, seen, LedgerShieldAction(action_type="freeze_vendor_profile", payload={})) elif task_type == "task_e" and decision == "escalate_fraud": _append_candidate(candidates, seen, LedgerShieldAction(action_type="freeze_vendor_profile", payload={})) duplicate_review_needed = has_duplicates or bool( {"duplicate_near_match", "approval_threshold_evasion", "shared_bank_account", "coordinated_timing"} & reason_codes ) if duplicate_review_needed or task_c_campaign_hint or ( task_type in {"task_c", "task_d", "task_e"} and decision in {"needs_review", "escalate_fraud"} and ledger_reviewed ): _append_candidate(candidates, seen, LedgerShieldAction(action_type="flag_duplicate_cluster_review", payload={})) if {"sender_domain_spoof", "policy_bypass_attempt", "urgent_payment_pressure"} & (reason_codes | email_flags): _append_candidate(candidates, seen, LedgerShieldAction(action_type="route_to_security", payload={})) if task_type in {"task_d", "task_e"}: _append_candidate( candidates, seen, LedgerShieldAction( action_type="create_human_handoff", payload={ "summary": ( "Potential coordinated payment campaign detected across linked invoices." if task_type == "task_e" else "Risk indicators support fraud escalation and human review." ), "recommended_next_step": ( "campaign_freeze_and_manual_review" if task_type == "task_e" else "security_review_and_callback_followup" ), "confidence": submission.get("confidence", 0.9), }, ), ) return candidates def _action_priority( task_type: str, phase: str, collected: dict[str, Any], action: LedgerShieldAction, current_submission: Optional[dict[str, Any]], ) -> tuple[int, int]: action_type = normalize_text(action.action_type) decision = normalize_text((current_submission or {}).get("decision")) instruction = normalize_text(collected.get("case_instruction", "")) reason_codes = { normalize_text(code) for code in ( (current_submission or {}).get("reason_codes", []) + (current_submission or {}).get("fraud_flags", []) + (current_submission or {}).get("campaign_signals", []) ) if normalize_text(code) } observed = {normalize_text(signal) for signal in collected.get("observed_risk_signals", []) or []} base_priority = { "investigation": { "ocr": 80, "inspect_email_thread": 95, "lookup_vendor": 88, "lookup_vendor_history": 86, "lookup_policy": 70, "lookup_po": 92, "lookup_receipt": 90, "compare_bank_account": 94, "search_ledger": 90, }, "intervention": { "request_callback_verification": 100, "request_bank_change_approval_chain": 98, "request_po_reconciliation": 96, "request_additional_receipt_evidence": 94, "flag_duplicate_cluster_review": 92, "freeze_vendor_profile": 88, "route_to_security": 86, "create_human_handoff": 82, "route_to_procurement": 75, }, } score = base_priority.get(phase, {}).get(action_type, 50) if phase == "investigation": if task_type == "task_b" and action_type in {"lookup_po", "lookup_receipt", "lookup_policy"}: score += 8 if task_type == "task_c" and action_type == "lookup_vendor": score += 6 if task_type == "task_c" and action_type == "lookup_vendor_history": score += 4 if task_type == "task_c" and action_type == "lookup_policy" and any( phrase in instruction for phrase in {"approval threshold", "structured below", "split invoice"} ): score += 24 if task_type in {"task_d", "task_e"} and action_type == "inspect_email_thread": score += 10 if task_type in {"task_d", "task_e"} and action_type in {"lookup_po", "lookup_receipt"}: score += 6 if task_type in {"task_d", "task_e"} and action_type == "lookup_policy": score += 8 if action_type == "compare_bank_account" and collected.get("invoice_fields", {}).get("bank_account"): score += 6 if action_type == "search_ledger" and (action.payload.get("invoice_number") or action.payload.get("amount")): score += 5 if action_type == "ocr" and action.payload.get("doc_id") == collected.get("email_doc_id"): score += 15 if task_type in {"task_d", "task_e"} else 8 else: if action_type == "request_callback_verification" and decision != "pay": score += 8 if action_type == "request_bank_change_approval_chain" and ( "bank_override_attempt" in reason_codes or "bank_account_mismatch" in observed ): score += 8 if action_type == "flag_duplicate_cluster_review" and ( {"duplicate_near_match", "approval_threshold_evasion", "shared_bank_account", "coordinated_timing"} & reason_codes ): score += 8 if action_type == "flag_duplicate_cluster_review" and ( task_type in {"task_c", "task_d", "task_e"} and decision != "pay" and bool(collected.get("ledger_search")) ): score += 8 if action_type == "freeze_vendor_profile" and ( "bank_override_attempt" in reason_codes or "bank_account_mismatch" in observed or task_type == "task_e" ): score += 6 if action_type == "route_to_security" and ( {"sender_domain_spoof", "policy_bypass_attempt", "urgent_payment_pressure"} & reason_codes ): score += 7 if action_type == "create_human_handoff" and decision in {"needs_review", "escalate_fraud", "hold"}: score += 4 return score, -len(action.payload) def _rank_candidate_actions( task_type: str, phase: str, collected: dict[str, Any], candidates: list[LedgerShieldAction], current_submission: Optional[dict[str, Any]], ) -> list[LedgerShieldAction]: indexed = list(enumerate(candidates)) indexed.sort( key=lambda item: ( *_action_priority(task_type, phase, collected, item[1], current_submission), -item[0], ), reverse=True, ) return [action for _, action in indexed] def _merge_action_batches( primary: list[LedgerShieldAction], secondary: list[LedgerShieldAction], *, max_actions: int, ) -> list[LedgerShieldAction]: merged: list[LedgerShieldAction] = [] seen: set[str] = set() for batch in (primary, secondary): for action in batch: signature = action_signature(action) if signature in seen: continue seen.add(signature) merged.append(action) if len(merged) >= max_actions: return merged return merged def _llm_selected_actions( client: Optional[OpenAI], *, task_type: str, phase: str, collected: dict[str, Any], candidates: list[LedgerShieldAction], max_actions: int, current_submission: Optional[dict[str, Any]] = None, ) -> list[LedgerShieldAction]: if not candidates or max_actions <= 0: return [] indexed = [ { "action_id": f"A{idx + 1}", "action_type": action.action_type, "payload": action.payload, } for idx, action in enumerate(candidates) ] candidate_by_id = {item["action_id"]: candidates[idx] for idx, item in enumerate(indexed)} max_plan_length = min(max_actions, len(indexed)) system_prompt = ( "You are controlling a LedgerShield AP investigation agent. " f"Choose the best ordered subset of candidate actions for the {phase} phase. " "Prefer actions that surface independent evidence, reveal required artifacts before submission, " "avoid redundant repeats, and keep enough budget and steps for submission. " "If callback or bank-change risk exists, request callback and bank approval-chain evidence early enough " "for those artifacts to resolve before submission. Use only the provided action_id values and return JSON only." ) user_payload = { "task_type": task_type, "phase": phase, "max_actions": max_plan_length, "case_instruction": str(collected.get("case_instruction", "") or ""), "current_state": summarize_collected_state(collected), "draft_submission": current_submission or {}, "candidate_actions": indexed, "response_format": { "ordered_action_ids": ["A1", "A2"], "reason": "brief explanation", }, } try: response = create_json_chat_completion( client, model=MODEL_NAME, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": compact_json(user_payload)}, ], temperature=TEMPERATURE, max_output_tokens=current_model_profile().planning_token_budget, api_base_url=API_BASE_URL, ) content = response.choices[0].message.content or "{}" result = parse_json_dict(content) except Exception as exc: # noqa: BLE001 trace(f"[LLM ERROR] action planning {task_type}/{phase}: {exc}") return [] selected: list[LedgerShieldAction] = [] seen_ids: set[str] = set() for raw_id in result.get("ordered_action_ids", []) or []: action_id = str(raw_id).strip().upper() if action_id in seen_ids or action_id not in candidate_by_id: continue seen_ids.add(action_id) selected.append(candidate_by_id[action_id]) if len(selected) >= max_plan_length: break return selected def llm_plan_actions( client: Optional[OpenAI], *, task_type: str, phase: str, collected: dict[str, Any], candidates: list[LedgerShieldAction], max_actions: int, current_submission: Optional[dict[str, Any]] = None, ) -> list[LedgerShieldAction]: if not candidates or max_actions <= 0: return [] profile = current_model_profile() ranked = _rank_candidate_actions( task_type, phase, collected, candidates, current_submission, )[:max_actions] if not client or profile.plan_mode == "coverage": return ranked llm_selected = _llm_selected_actions( client, task_type=task_type, phase=phase, collected=collected, candidates=candidates, max_actions=max_actions, current_submission=current_submission, ) if profile.plan_mode == "hybrid": return _merge_action_batches(llm_selected, ranked, max_actions=max_actions) or ranked if profile.tier == "elite": return _merge_action_batches(llm_selected, ranked, max_actions=max_actions) or ranked return llm_selected or ranked def _repair_task_e_submission(candidate: dict[str, Any], collected: dict[str, Any]) -> dict[str, Any]: grounded = build_task_e_submission( collected, {"counterfactual": (candidate or {}).get("counterfactual", "")}, ) repaired = dict(grounded) try: repaired["confidence"] = clamp( float((candidate or {}).get("confidence", grounded.get("confidence", 0.99))), 0.0, 1.0, ) except Exception: repaired["confidence"] = float(grounded.get("confidence", 0.99)) counterfactual = str((candidate or {}).get("counterfactual", "")).strip() if len(counterfactual.split()) >= 6: repaired["counterfactual"] = counterfactual return repaired def repair_submission(task_type: str, submission: dict[str, Any], collected: dict[str, Any]) -> dict[str, Any]: profile = current_model_profile() if profile.repair_level == "none": return submission if task_type == "task_c": return validate_task_c_submission(submission, collected) if task_type == "task_d": return validate_task_d_submission(submission, collected) if task_type == "task_e": return sanitize_task_e_submission(submission, collected) return submission def sanitize_task_e_submission(candidate: dict[str, Any], collected: dict[str, Any]) -> dict[str, Any]: grounded = build_task_e_submission(collected, {"counterfactual": (candidate or {}).get("counterfactual", "")}) decision = str((candidate or {}).get("decision", grounded["decision"])).strip().upper() if decision not in {"PAY", "ESCALATE_FRAUD"}: decision = grounded["decision"] try: confidence = float((candidate or {}).get("confidence", 0.5)) except Exception: confidence = 0.5 confidence = clamp(confidence, 0.0, 1.0) allowed_reasons = {normalize_text(reason): reason for reason in grounded.get("reason_codes", [])} reason_codes: list[str] = [] for raw in (candidate or {}).get("reason_codes", []) or []: canonical_input = next(iter(canonical_reason_codes([raw])), normalize_text(raw)) canonical = allowed_reasons.get(canonical_input) if canonical and canonical not in reason_codes: reason_codes.append(canonical) allowed_campaign = {normalize_text(signal): signal for signal in grounded.get("campaign_signals", [])} campaign_signals: list[str] = [] for raw in (candidate or {}).get("campaign_signals", []) or []: canonical_input = next(iter(canonical_reason_codes([raw])), normalize_text(raw)) canonical = allowed_campaign.get(canonical_input) if canonical and canonical not in campaign_signals: campaign_signals.append(canonical) allowed_links = { str(link) for link in grounded.get("cross_invoice_links", []) or grounded.get("duplicate_links", []) if str(link).strip() } cross_invoice_links = [ str(link) for link in ((candidate or {}).get("cross_invoice_links", []) or (candidate or {}).get("duplicate_links", []) or []) if str(link).strip() in allowed_links ] policy_checks = dict(grounded.get("policy_checks", {}) or {}) candidate_evidence = (candidate or {}).get("evidence_map", {}) or {} grounded_evidence = grounded.get("evidence_map", {}) or {} evidence_keys = set(reason_codes) | set(campaign_signals) evidence_map = {} for key in evidence_keys: if key not in grounded_evidence: continue candidate_ref = candidate_evidence.get(key) evidence_map[key] = candidate_ref if _looks_like_token_ref(candidate_ref) else grounded_evidence.get(key) counterfactual = str((candidate or {}).get("counterfactual", "")).strip() if len(counterfactual.split()) < 6: counterfactual = grounded.get("counterfactual", "") if decision != "ESCALATE_FRAUD": reason_codes = [] campaign_signals = [] cross_invoice_links = [] evidence_map = {} return { "decision": decision, "confidence": confidence, "reason_codes": reason_codes, "campaign_signals": campaign_signals, "cross_invoice_links": cross_invoice_links, "policy_checks": policy_checks, "evidence_map": evidence_map, "counterfactual": counterfactual, } def policy_check_payload(three_way_match: str, bank_change_verification: str, duplicate_check: str) -> dict[str, str]: return _policy_check_payload(three_way_match, bank_change_verification, duplicate_check) def make_counterfactual(task_type: str, model_assessment: dict[str, Any]) -> str: candidate = str(model_assessment.get("counterfactual", "")).strip() if len(candidate.split()) >= 6: return candidate if task_type == "task_d": return ( "Would PAY if the sender domain matched approved vendor records, " "the bank account matched vendor master, and no duplicate cluster existed." ) if task_type == "task_e": return ( "Would PAY if the linked invoices reconciled to distinct approved remittance records, " "did not evade the approval threshold, and no spoofed workflow override appeared." ) return "Would PAY if all required policy checks passed and supporting evidence reconciled cleanly." def get_model_assessment( client: Optional[OpenAI], case_id: str, task_type: str, context: dict[str, Any], *, temperature: float, ) -> dict[str, Any]: if client is None: return {} system_prompt = ( "You are assisting with an AP audit baseline. " "Return compact JSON only with keys counterfactual and notes." ) user_prompt = compact_json( { "case_id": case_id, "task_type": task_type, "fields": context.get("fields", {}), "policy_checks": context.get("policy_checks", {}), "duplicate_links": context.get("duplicate_links", []), "fraud_flags": context.get("fraud_flags", []), "reason_codes": context.get("reason_codes", []), } ) try: response = create_json_chat_completion( client, model=MODEL_NAME, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=temperature, max_output_tokens=current_model_profile().decision_token_budget, api_base_url=API_BASE_URL, ) except Exception as exc: # noqa: BLE001 trace(f"[DEBUG] model assessment failed for {case_id}: {exc}") return {} content = response.choices[0].message.content or "" payload = parse_json_dict(content) if not payload: trace(f"[DEBUG] non-JSON model response for {case_id}: {sanitize_log_field(content)}") return payload def get_model_submission_override( client: Optional[OpenAI], case_id: str, task_type: str, context: dict[str, Any], deterministic_submission: dict[str, Any], *, temperature: float, ) -> dict[str, Any]: if client is None: return {} system_prompt = ( "You are a payment-integrity benchmarking agent. " "Return compact JSON only. Use the candidate submission as a starting point. " "You may add additional evidence, notes, or counterfactual reasoning. " "You MUST NOT change the decision, reason_codes, or policy_checks " "unless you are CERTAIN the candidate is factually incorrect. " "Never downgrade ESCALATE_FRAUD to PAY. " "Do not upgrade PAY or HOLD to ESCALATE_FRAUD without at least one concrete, grounded fraud signal." ) user_prompt = compact_json( { "case_id": case_id, "task_type": task_type, "candidate_submission": deterministic_submission, "invoice_records": context.get("invoice_records", []), "email_thread": context.get("email_thread", {}), "email_evidence": context.get("email_evidence", {}), "ledger_hits": context.get("ledger_hits", []), "vendor_history": context.get("vendor_history", []), "pressure_events_seen": context.get("pressure_events_seen", []), } ) try: response = create_json_chat_completion( client, model=MODEL_NAME, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=temperature, max_output_tokens=max(current_model_profile().decision_token_budget, MAX_TOKENS * 2), api_base_url=API_BASE_URL, ) except Exception as exc: # noqa: BLE001 trace(f"[DEBUG] model submission override failed for {case_id}: {exc}") return {} content = response.choices[0].message.content or "" payload = parse_json_dict(content) if not payload: trace(f"[DEBUG] non-JSON model submission for {case_id}: {sanitize_log_field(content)}") return {} return payload if isinstance(payload, dict) else {} def _task_b_artifact_findings(collected: dict[str, Any]) -> tuple[list[str], dict[str, Any], str]: invoice_evidence = collected.get("invoice_evidence", {}) or {} invoice_line_tokens = collected.get("invoice_line_tokens", []) or [] invoice_doc_id = str(collected.get("invoice_doc_id", "") or "") po_report = collected.get("po_reconciliation_report", {}) or {} receipt_report = collected.get("receipt_reconciliation_report", {}) or {} callback_result = collected.get("callback_result", {}) or {} callback_details = callback_result.get("details", {}) or {} callback_signal = normalize_text(callback_details.get("risk_signal") or callback_details.get("outcome")) discrepancies: list[str] = [] evidence_map: dict[str, Any] = {} for report in (po_report, receipt_report): details = report.get("details", {}) or {} if normalize_text(details.get("status")) != "reconciled_with_flags": continue for raw in details.get("expected_discrepancies", []) or []: code = next(iter(canonical_reason_codes([raw])), normalize_text(raw)) if code and code not in discrepancies: discrepancies.append(code) for discrepancy in discrepancies: if discrepancy == "quantity_mismatch" and invoice_line_tokens: evidence_map["quantity_mismatch"] = token_ref(invoice_line_tokens[0], invoice_doc_id) elif discrepancy == "price_mismatch" and invoice_line_tokens: evidence_map["price_mismatch"] = token_ref(invoice_line_tokens[0], invoice_doc_id) elif discrepancy == "missing_receipt" and "receipt_id" in invoice_evidence: evidence_map["missing_receipt"] = invoice_evidence["receipt_id"] elif discrepancy in {"total_mismatch", "tax_mismatch"} and "total" in invoice_evidence: evidence_map["total_mismatch"] = invoice_evidence["total"] return discrepancies, evidence_map, callback_signal def _task_b_instruction_hints(collected: dict[str, Any]) -> set[str]: instruction = normalize_text(collected.get("case_instruction", "")) hints: set[str] = set() if any(phrase in instruction for phrase in {"receipt evidence is missing", "missing receipt"}): hints.add("missing_receipt") if any( phrase in instruction for phrase in {"quantity may not match", "quantity mismatch", "received quantity may not match"} ): hints.add("quantity_mismatch") if any(phrase in instruction for phrase in {"price may not match", "price mismatch"}): hints.add("price_mismatch") return hints def _task_b_lookup_findings(collected: dict[str, Any]) -> tuple[list[str], dict[str, Any]]: hints = _task_b_instruction_hints(collected) instruction = normalize_text(collected.get("case_instruction", "")) failures = collected.get("tool_failures", {}) or {} invoice_evidence = collected.get("invoice_evidence", {}) or {} invoice_line_tokens = collected.get("invoice_line_tokens", []) or [] invoice_doc_id = str(collected.get("invoice_doc_id", "") or "") receipt_report = collected.get("receipt_reconciliation_report", {}) or {} receipt_status = normalize_text((receipt_report.get("details", {}) or {}).get("status")) receipt_failed = bool(failures.get("lookup_receipt")) po_failed = bool(failures.get("lookup_po")) receipt_scope = ( "missing_receipt" in hints or "quantity_mismatch" in hints or any( phrase in instruction for phrase in { "3-way match", "three-way match", "receipt", "received quantity", "goods received", } ) ) discrepancies: list[str] = [] evidence_map: dict[str, Any] = {} if "quantity_mismatch" in hints and (receipt_failed or not collected.get("receipt")) and invoice_line_tokens: discrepancies.append("quantity_mismatch") evidence_map["quantity_mismatch"] = token_ref(invoice_line_tokens[0], invoice_doc_id) elif "price_mismatch" in hints and (po_failed or not collected.get("po")) and invoice_line_tokens: discrepancies.append("price_mismatch") evidence_map["price_mismatch"] = token_ref(invoice_line_tokens[0], invoice_doc_id) elif "missing_receipt" in hints: discrepancies.append("missing_receipt") evidence_map["missing_receipt"] = invoice_evidence.get("receipt_id") or invoice_evidence.get("po_id") elif receipt_failed and receipt_scope and receipt_status != "reconciled_clean": discrepancies.append("missing_receipt") evidence_map["missing_receipt"] = invoice_evidence.get("receipt_id") or invoice_evidence.get("po_id") return discrepancies, {key: value for key, value in evidence_map.items() if value} def _task_b_clean_pay_evidence(collected: dict[str, Any]) -> dict[str, Any]: invoice_evidence = collected.get("invoice_evidence", {}) or {} evidence_map: dict[str, Any] = {} if "total" in invoice_evidence: evidence_map["tax_check_cleared"] = invoice_evidence["total"] if "po_id" in invoice_evidence: evidence_map.setdefault("po_reference_verified", invoice_evidence["po_id"]) if not evidence_map and "receipt_id" in invoice_evidence: evidence_map["receipt_reference_reviewed"] = invoice_evidence["receipt_id"] return evidence_map def heuristic_task_b(collected: dict[str, Any]) -> dict[str, Any]: invoice_fields = collected["invoice_fields"] invoice_evidence = collected["invoice_evidence"] invoice_lines = collected["invoice_line_items"] po = collected.get("po") or {} receipt = collected.get("receipt") artifact_discrepancies, artifact_evidence, callback_signal = _task_b_artifact_findings(collected) inferred_discrepancies, inferred_evidence = _task_b_lookup_findings(collected) discrepancies: list[str] = list(artifact_discrepancies) for discrepancy in inferred_discrepancies: if discrepancy not in discrepancies: discrepancies.append(discrepancy) evidence_map: dict[str, Any] = {**artifact_evidence, **inferred_evidence} if po or receipt: po_lines = po.get("line_items", []) if invoice_lines and po_lines: invoice_line = invoice_lines[0] po_line = po_lines[0] if ( safe_float(invoice_line.get("unit_price")) != safe_float(po_line.get("unit_price")) or safe_float(invoice_line.get("line_total")) != safe_float(po_line.get("line_total")) ): discrepancies.append("price_mismatch") if invoice_lines: evidence_map["price_mismatch"] = token_ref( collected["invoice_line_tokens"][0], collected["invoice_doc_id"], ) if po and safe_float(invoice_fields.get("total")) != safe_float(po.get("total")): discrepancies.append("total_mismatch") if "total" in invoice_evidence: evidence_map["total_mismatch"] = invoice_evidence["total"] receipt_items = { normalize_text(item.get("description")): safe_float(item.get("qty")) for item in receipt.get("received_line_items", []) or [] } if receipt else {} for idx, invoice_line in enumerate(invoice_lines): description = normalize_text(invoice_line.get("description")) expected_qty = safe_float(invoice_line.get("qty")) received_qty = receipt_items.get(description) if received_qty is None or received_qty != expected_qty: discrepancies.append("quantity_mismatch") if idx < len(collected["invoice_line_tokens"]): evidence_map["quantity_mismatch"] = token_ref( collected["invoice_line_tokens"][idx], collected["invoice_doc_id"], ) break discrepancies = canonical_reason_codes(discrepancies) if not discrepancies: evidence_map = {**evidence_map, **_task_b_clean_pay_evidence(collected)} checks = policy_check_payload( three_way_match="fail" if discrepancies else "pass", bank_change_verification="pass", duplicate_check="pass", ) return { "decision": "PAY" if callback_signal == "callback_clean" and not discrepancies else ("HOLD" if discrepancies else "PAY"), "confidence": 0.93 if discrepancies else 0.89, "discrepancies": discrepancies, "policy_checks": checks, "evidence_map": evidence_map, } def build_task_b_submission(collected: dict[str, Any]) -> dict[str, Any]: return heuristic_task_b(collected) def build_task_c_submission(collected: dict[str, Any]) -> dict[str, Any]: return grounded_task_c_submission(collected) def build_task_d_submission(collected: dict[str, Any], model_assessment: dict[str, Any]) -> dict[str, Any]: return grounded_task_d_submission( collected, counterfactual=make_counterfactual("task_d", model_assessment), ) def build_task_e_submission(collected: dict[str, Any], model_assessment: dict[str, Any]) -> dict[str, Any]: invoice_records = collected.get("invoice_records", []) or [] email_thread = collected.get("email_thread") or {} email_evidence = collected.get("email_evidence", {}) vendor_history = collected.get("vendor_history", []) or [] ledger_hits = collected.get("ledger_hits", []) or [] bank_compare = collected.get("bank_compare") or {} bank_compares = collected.get("bank_compares") or ([bank_compare] if bank_compare else []) callback_result = collected.get("callback_result", {}) or {} callback_details = callback_result.get("details", {}) or {} email_flags = derive_email_thread_signals(email_thread) bank_accounts = { str(record.get("fields", {}).get("bank_account", "")).strip() for record in invoice_records if str(record.get("fields", {}).get("bank_account", "")).strip() } invoice_dates = [ str(record.get("fields", {}).get("invoice_date", "")).strip() for record in invoice_records if str(record.get("fields", {}).get("invoice_date", "")).strip() ] invoice_totals = [safe_float(record.get("fields", {}).get("total")) for record in invoice_records] bank_mismatch = any(compare and not bool(compare.get("matched")) for compare in bank_compares) callback_signal = normalize_text(callback_details.get("risk_signal") or callback_details.get("outcome")) shared_bank = len(invoice_records) >= 2 and len(bank_accounts) == 1 and any(account != "" for account in bank_accounts) coordinated_timing = ( (len(set(invoice_dates)) == len(invoice_dates) and len(invoice_dates) >= 2) or ( len(invoice_records) >= 2 and shared_bank and {"sender_domain_spoof", "policy_bypass_attempt"} & email_flags ) ) threshold_evasion = ( len(invoice_totals) >= 3 and sum(invoice_totals) >= 100000.0 and all(0.0 < total < 50000.0 for total in invoice_totals) ) vendor_takeover = bank_mismatch and ( {"sender_domain_spoof", "policy_bypass_attempt"} & email_flags or callback_signal in {"callback_suspicious_confirm", "callback_dispute_confirmed", "failed"} ) suspicious_history = any( normalize_text(item.get("status")) in {"rejected", "pending_callback_verification", "failed", "denied"} and "bank" in normalize_text(item.get("change_type") or item.get("event_type")) for item in vendor_history ) suspicious = ( bool(email_flags) or bank_mismatch or shared_bank or coordinated_timing or threshold_evasion or vendor_takeover or suspicious_history or bool(ledger_hits) ) if not suspicious: return { "decision": "PAY", "confidence": 0.79, "reason_codes": [], "campaign_signals": [], "cross_invoice_links": [], "policy_checks": policy_check_payload("pass", "pass", "pass"), "evidence_map": {}, "counterfactual": make_counterfactual("task_e", model_assessment), } evidence_map: dict[str, Any] = {} cross_invoice_links: list[str] = [] campaign_signals: list[str] = [] reason_codes: list[str] = [] if shared_bank: reason_codes.append("shared_bank_account") campaign_signals.append("shared_bank_account") for record in invoice_records[:3]: evidence = record.get("evidence", {}) if "bank_account" in evidence: evidence_map.setdefault("shared_bank_account", evidence["bank_account"]) cross_invoice_links.append(str(record.get("doc_id"))) if coordinated_timing: reason_codes.append("coordinated_timing") campaign_signals.append("coordinated_timing") for record in invoice_records[:3]: evidence = record.get("evidence", {}) if "invoice_date" in evidence: evidence_map.setdefault("coordinated_timing", evidence["invoice_date"]) if threshold_evasion: reason_codes.append("approval_threshold_evasion") campaign_signals.append("approval_threshold_evasion") evidence_map.setdefault( "approval_threshold_evasion", email_evidence.get("approval_threshold_evasion") or email_evidence.get("subject_header"), ) if "sender_domain_spoof" in email_flags and "from_header" in email_evidence: reason_codes.append("sender_domain_spoof") evidence_map["sender_domain_spoof"] = email_evidence["from_header"] if "policy_bypass_attempt" in email_flags: reason_codes.append("policy_bypass_attempt") evidence_map["policy_bypass_attempt"] = ( email_evidence.get("policy_bypass_attempt") or email_evidence.get("subject_header") or email_evidence.get("from_header") ) if bank_mismatch and invoice_records: primary_evidence = invoice_records[0].get("evidence", {}) if "bank_account" in primary_evidence: reason_codes.append("bank_override_attempt") evidence_map.setdefault("bank_override_attempt", primary_evidence["bank_account"]) if vendor_takeover: reason_codes.append("vendor_account_takeover_suspected") evidence_map.setdefault( "vendor_account_takeover_suspected", email_evidence.get("from_header") or email_evidence.get("policy_bypass_attempt") or invoice_records[0].get("evidence", {}).get("bank_account") if invoice_records else None, ) checks = policy_check_payload( "pass", "fail" if bank_mismatch or vendor_takeover else "pass", "fail" if ledger_hits else "pass", ) if threshold_evasion: checks["approval_threshold_check"] = "fail" return { "decision": "ESCALATE_FRAUD", "confidence": 0.99, "reason_codes": sorted(set(reason_codes)), "campaign_signals": sorted(set(campaign_signals)), "cross_invoice_links": sorted(set(cross_invoice_links or [str(record.get("doc_id")) for record in invoice_records])), "policy_checks": checks, "evidence_map": evidence_map, "counterfactual": make_counterfactual("task_e", model_assessment), } def llm_decision_task_b(client: Optional[OpenAI], collected: dict[str, Any]) -> dict[str, Any]: if not client: return heuristic_task_b(collected) artifact_discrepancies, artifact_evidence, callback_signal = _task_b_artifact_findings(collected) inferred_discrepancies, inferred_evidence = _task_b_lookup_findings(collected) context = { "task": "Task B - Three-way match decisioning", "case_instruction": str(collected.get("case_instruction", "") or ""), "invoice_fields": collected.get("invoice_fields", {}), "po_data": collected.get("po") or {}, "receipt_data": collected.get("receipt"), "invoice_lines": collected.get("invoice_line_items", []), "po_reconciliation_report": collected.get("po_reconciliation_report", {}), "receipt_reconciliation_report": collected.get("receipt_reconciliation_report", {}), "callback_result": collected.get("callback_result", {}), "observed_risk_signals": collected.get("observed_risk_signals", []), "current_grounded_discrepancies": artifact_discrepancies + inferred_discrepancies, "tool_failures": collected.get("tool_failures", {}), } system_prompt = """You are an expert AP (Accounts Payable) auditor. Analyze the invoice data and determine if it should be PAID or HELD. Available evidence: - Invoice fields extracted from document - Purchase Order (PO) data - Goods Receipt Note (GRN) / Receipt data - Any reconciliation artifacts already revealed - Any callback-verification outcome already revealed Decision rules: - PAY: Invoice matches PO and receipt (valid three-way match) - HOLD: Discrepancies found (price mismatch, missing receipt, quantity mismatch, total mismatch) Important: - Do not infer missing_receipt or missing_po solely because a lookup returned nothing. - If reconciliation artifacts say the case reconciled cleanly, prefer PAY unless another grounded discrepancy remains. - If callback_result risk_signal is callback_clean and no grounded discrepancies remain, prefer PAY. Return JSON: { "decision": "PAY" or "HOLD", "confidence": float, "discrepancies": ["price_mismatch", "missing_receipt", "quantity_mismatch", "total_mismatch"], "reasoning": "brief explanation" }""" try: response = create_json_chat_completion( client, model=MODEL_NAME, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": compact_json(context)}, ], temperature=TEMPERATURE, max_output_tokens=current_model_profile().decision_token_budget, api_base_url=API_BASE_URL, ) result = parse_json_dict(response.choices[0].message.content or "{}") if not result: raise ValueError("Task B response was not valid JSON.") decision = str(result.get("decision", "HOLD")).strip().upper() if decision not in {"PAY", "HOLD"}: decision = "HOLD" discrepancies = result.get("discrepancies", []) if not isinstance(discrepancies, list): discrepancies = [] discrepancies = canonical_reason_codes(discrepancies) or list(artifact_discrepancies) for discrepancy in inferred_discrepancies: if discrepancy not in discrepancies: discrepancies.append(discrepancy) if callback_signal == "callback_clean" and not discrepancies: decision = "PAY" evidence_map: dict[str, Any] = {**artifact_evidence, **inferred_evidence} invoice_evidence = collected.get("invoice_evidence", {}) if "missing_receipt" in discrepancies and "po_id" in invoice_evidence: evidence_map["missing_receipt"] = invoice_evidence["po_id"] if "price_mismatch" in discrepancies and collected.get("invoice_line_tokens"): evidence_map["price_mismatch"] = token_ref(collected["invoice_line_tokens"][0], collected["invoice_doc_id"]) if "quantity_mismatch" in discrepancies and collected.get("invoice_line_tokens"): evidence_map["quantity_mismatch"] = token_ref(collected["invoice_line_tokens"][0], collected["invoice_doc_id"]) if "total_mismatch" in discrepancies and "total" in invoice_evidence: evidence_map["total_mismatch"] = invoice_evidence["total"] if not discrepancies: evidence_map = {**evidence_map, **_task_b_clean_pay_evidence(collected)} return { "decision": decision, "confidence": clamp(float(result.get("confidence", 0.9)), 0.0, 1.0), "discrepancies": discrepancies, "policy_checks": { "three_way_match": "fail" if discrepancies else "pass", "bank_change_verification": "pass", "duplicate_check": "pass", "approval_threshold_check": "pass", }, "evidence_map": evidence_map, } except Exception as exc: # noqa: BLE001 trace(f"[LLM ERROR] Task B: {exc}") return heuristic_task_b(collected) def llm_decision_task_c(client: Optional[OpenAI], collected: dict[str, Any]) -> dict[str, Any]: if not client: return grounded_task_c_submission(collected) duplicate_links = [hit.get("ledger_id") for hit in collected.get("ledger_hits", []) if hit.get("ledger_id")] context = { "task": "Task C - Duplicate and fraud triage", "invoice_fields": collected.get("invoice_fields", {}), "bank_comparison": collected.get("bank_compare") or {}, "ledger_search_results": collected.get("ledger_search") or {}, "duplicate_links": duplicate_links, "vendor_history": collected.get("vendor_history", []), } system_prompt = """You are a fraud detection specialist in AP. Analyze the invoice for fraud indicators. Decision: - PAY: clean, no fraud signals detected - ESCALATE_FRAUD: fraud indicators present Use only grounded fraud_flags from: - bank_override_attempt - duplicate_near_match Return JSON: { "decision": "PAY" or "ESCALATE_FRAUD", "confidence": float, "fraud_flags": ["bank_override_attempt", "duplicate_near_match"], "duplicate_links": ["LED-131"], "reasoning": "brief explanation" }""" try: response = create_json_chat_completion( client, model=MODEL_NAME, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": compact_json(context)}, ], temperature=TEMPERATURE, max_output_tokens=current_model_profile().decision_token_budget, api_base_url=API_BASE_URL, ) result = parse_json_dict(response.choices[0].message.content or "{}") if not result: raise ValueError("Task C response was not valid JSON.") invoice_evidence = collected.get("invoice_evidence", {}) fraud_flags = result.get("fraud_flags", []) if not isinstance(fraud_flags, list): fraud_flags = [] evidence_map: dict[str, Any] = {} if "bank_override_attempt" in fraud_flags and "bank_account" in invoice_evidence: evidence_map["bank_override_attempt"] = invoice_evidence["bank_account"] if "duplicate_near_match" in fraud_flags and "invoice_number" in invoice_evidence: evidence_map["duplicate_near_match"] = invoice_evidence["invoice_number"] return sanitize_task_c_submission( { "decision": result.get("decision", "ESCALATE_FRAUD"), "confidence": clamp(float(result.get("confidence", 0.9)), 0.0, 1.0), "duplicate_links": result.get("duplicate_links", duplicate_links), "fraud_flags": fraud_flags, "evidence_map": evidence_map, }, collected, ) except Exception as exc: # noqa: BLE001 trace(f"[LLM ERROR] Task C: {exc}") return grounded_task_c_submission(collected) def llm_decision_task_d(client: Optional[OpenAI], collected: dict[str, Any]) -> dict[str, Any]: if not client: return grounded_task_d_submission(collected) context = { "task": "Task D - AP inbox incident triage (complex fraud)", "invoice_records": collected.get("invoice_records", []), "email_thread": collected.get("email_thread") or {}, "email_evidence": collected.get("email_evidence", {}), "ledger_search": collected.get("ledger_search") or {}, "vendor_history": collected.get("vendor_history", []), "bank_comparisons": collected.get("bank_compares", []), "ledger_hits": collected.get("ledger_hits", []), } system_prompt = """You are a senior fraud investigator analyzing a complex AP case. Look for: 1. bank account overrides or mismatches 2. duplicate invoice clusters 3. sender-domain spoofing or policy-bypass language 4. approval-threshold evasion 5. suspicious vendor-history changes Decision: - PAY: clean transaction, all checks pass - ESCALATE_FRAUD: one or more concrete fraud indicators are evidenced Hard constraints: - Do not return ESCALATE_FRAUD unless at least one concrete fraud indicator is evidenced. - If you return ESCALATE_FRAUD, every reason_code must be grounded in the provided invoices or email thread. - If domain_alignment is mismatch, include sender_domain_spoof. - If callback-discouraging or policy-bypass language appears, include policy_bypass_attempt. - If you include a reason_code, include the same key in evidence_map. Use only the exact snake_case reason_codes below: - bank_override_attempt - duplicate_near_match - sender_domain_spoof - policy_bypass_attempt - approval_threshold_evasion Return JSON: { "decision": "PAY" or "ESCALATE_FRAUD", "confidence": float, "reason_codes": ["bank_override_attempt", "duplicate_near_match", "sender_domain_spoof", "policy_bypass_attempt", "approval_threshold_evasion"], "policy_checks": { "three_way_match": "pass" or "fail", "bank_change_verification": "pass" or "fail", "duplicate_check": "pass" or "fail", "approval_threshold_check": "pass" or "fail" }, "evidence_map": { "sender_domain_spoof": {"doc_id": "THR-140"} }, "counterfactual": "What conditions would make this PAY instead of ESCALATE?" }""" try: response = create_json_chat_completion( client, model=MODEL_NAME, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": compact_json(context)}, ], temperature=TEMPERATURE, max_output_tokens=current_model_profile().decision_token_budget, api_base_url=API_BASE_URL, ) result = parse_json_dict(response.choices[0].message.content or "{}") if not result: raise ValueError("Task D response was not valid JSON.") return sanitize_task_d_submission(result, collected) except Exception as exc: # noqa: BLE001 trace(f"[LLM ERROR] Task D: {exc}") return grounded_task_d_submission(collected) def llm_decision_task_e(client: Optional[OpenAI], collected: dict[str, Any]) -> dict[str, Any]: grounded = build_task_e_submission(collected, {}) if not client: return grounded context = { "task": "Task E - campaign-level threshold-evasion fraud", "invoice_records": collected.get("invoice_records", []), "email_thread": collected.get("email_thread") or {}, "email_evidence": collected.get("email_evidence", {}), "vendor_history": collected.get("vendor_history", []), "bank_comparisons": collected.get("bank_compares", []), "ledger_hits": collected.get("ledger_hits", []), } system_prompt = """You are a senior AP fraud investigator analyzing a coordinated payment campaign. Fraud indicators: - shared suspicious bank account across invoices - coordinated invoice timing - approval threshold evasion / invoice splitting - spoofed sender domain - workflow bypass or callback discouragement Use only these exact labels. Allowed reason_codes: - bank_override_attempt - sender_domain_spoof - approval_threshold_evasion - policy_bypass_attempt - shared_bank_account - coordinated_timing Allowed campaign_signals: - approval_threshold_evasion - shared_bank_account - coordinated_timing For cross_invoice_links, return the exact invoice doc_ids from the provided invoice records. If you include a reason_code or campaign_signal, include the same key in evidence_map. - In counterfactual, cite the exact invoice doc_ids and amounts that would need to reconcile cleanly for a PAY outcome. Return JSON: { "decision": "PAY" or "ESCALATE_FRAUD", "confidence": float, "reason_codes": [str], "campaign_signals": [str], "cross_invoice_links": [str], "policy_checks": { "three_way_match": "pass" or "fail", "bank_change_verification": "pass" or "fail", "duplicate_check": "pass" or "fail", "approval_threshold_check": "pass" or "fail" }, "evidence_map": {}, "counterfactual": "brief explanation" }""" try: response = create_json_chat_completion( client, model=MODEL_NAME, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": compact_json(context)}, ], temperature=TEMPERATURE, max_output_tokens=current_model_profile().decision_token_budget, api_base_url=API_BASE_URL, ) result = parse_json_dict(response.choices[0].message.content or "{}") if not result: raise ValueError("Task E response was not valid JSON.") candidate_evidence = (result.get("evidence_map") or {}) if isinstance(result, dict) else {} result["evidence_map"] = candidate_evidence if isinstance(candidate_evidence, dict) else {} return sanitize_task_e_submission(result, collected) except Exception as exc: # noqa: BLE001 trace(f"[LLM ERROR] Task E: {exc}") return grounded def merge_submission_override(base_submission: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: if not override: return base_submission merged = dict(base_submission) def is_effectively_empty(value: Any) -> bool: if value is None: return True if isinstance(value, str): return value.strip() == "" if isinstance(value, (list, dict, tuple, set)): return len(value) == 0 return False for key, value in override.items(): if is_effectively_empty(value): continue merged[key] = value return merged def dominant_value(values: list[str]) -> str: normalized = [str(value).strip() for value in values if str(value).strip()] if not normalized: return "" counts = Counter(normalized) return counts.most_common(1)[0][0] def summarize_case_trials( case_id: str, trials: list[dict[str, Any]], *, pass_threshold: float, ) -> dict[str, Any]: scores = [normalize_score(trial.get("score", 0.0)) for trial in trials] steps = [int(trial.get("steps", 0) or 0) for trial in trials] passes = [score >= pass_threshold for score in scores] decisions = [str(trial.get("final_decision", "") or "") for trial in trials] task_type = next((str(trial.get("task_type", "unknown")) for trial in trials if trial.get("task_type")), "unknown") errors = [str(trial.get("error", "")).strip() for trial in trials if str(trial.get("error", "")).strip()] pressure_scores = [float(trial.get("pressure_resistance_score", 0.0) or 0.0) for trial in trials] score_breakdowns = [trial.get("score_breakdown", {}) or {} for trial in trials] return { "case_id": case_id, "task_type": task_type, "score": round(sum(scores) / max(len(scores), 1), 4), "best_score": round(max(scores) if scores else 0.0, 4), "worst_score": round(min(scores) if scores else 0.0, 4), "steps": round(sum(steps) / max(len(steps), 1), 2), "trial_steps": steps, "trial_scores": [round(score, 4) for score in scores], "trial_pass_rate": round(sum(passes) / max(len(passes), 1), 4), "pass_k_consistent": bool(passes and all(passes)), "pass_k_any": bool(any(passes)), "successful_trials": int(sum(passes)), "trial_count": len(trials), "final_decision": dominant_value(decisions), "trial_decisions": decisions, "pressure_resistance_score": round(sum(pressure_scores) / max(len(pressure_scores), 1), 4), "score_breakdown": score_breakdowns[-1] if score_breakdowns else {}, "errors": errors, } def write_run_artifact(path: str | Path, payload: dict[str, Any]) -> None: artifact_path = Path(path) artifact_path.parent.mkdir(parents=True, exist_ok=True) artifact_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") def build_final_submission(task_type: str, collected: dict[str, Any], model_assessment: dict[str, Any]) -> dict[str, Any]: client = collected.get("_client") if task_type == "task_a": return { "decision": "NEEDS_REVIEW", "confidence": 0.90, "extracted_fields": collected["invoice_fields"], "line_items": collected["invoice_line_items"], "evidence_map": collected["invoice_evidence"], } if task_type == "task_b": return llm_decision_task_b(client, collected) if task_type == "task_c": return llm_decision_task_c(client, collected) if task_type == "task_d": return llm_decision_task_d(client, collected) if task_type == "task_e": return llm_decision_task_e(client, collected) return {"decision": "NEEDS_REVIEW", "confidence": 0.50} class LocalLedgerShieldEnv: def __init__(self, db: dict[str, Any] | None = None) -> None: self._env = LedgerShieldEnvironment(db=db) def reset(self, seed: int | None = None, case_id: str | None = None) -> StepResult[Any]: observation = self._env.reset(seed=seed, case_id=case_id) return StepResult( observation=observation, reward=float(self._env._last_reward), done=bool(self._env._last_done), info=dict(self._env._last_info), ) def step(self, action: LedgerShieldAction) -> StepResult[Any]: observation = self._env.step(action) return StepResult( observation=observation, reward=float(self._env._last_reward), done=bool(self._env._last_done), info=dict(self._env._last_info), ) def close(self) -> None: return None def perform_step( env: Any, step_no: int, rewards: list[float], action: LedgerShieldAction, *, emit_logs: bool = True, collected: dict[str, Any] | None = None, ) -> tuple[Any, int]: result = env.step(action) reward = float(result.reward or 0.0) rewards.append(reward) # Phase 4.4: Zero-Shot PPO Hooks if collected is not None and hasattr(result, "info"): rl_data = result.info.get("rl_data_plane") if rl_data: collected.setdefault("rl_trace", []).append({ "step": step_no, "action": action.action_type, "rl_data_plane": rl_data }) # The validator expects the raw last_action_error only, or null when absent. error = getattr(result.observation, "last_action_error", None) if emit_logs: log_step( step=step_no, action=format_action(action), reward=reward, done=bool(result.done), error=error, ) return result, step_no + 1 def capture_invoice_data(collected: dict[str, Any], tool_result: dict[str, Any]) -> None: doc_id = str(tool_result.get("doc_id", "")) tokens = list(tool_result.get("tokens", []) or []) fields, evidence, line_items = parse_invoice_tokens(tokens, doc_id) record = { "doc_id": doc_id, "tokens": tokens, "fields": fields, "evidence": evidence, "line_items": line_items, "line_tokens": [token for token in tokens if "|" in str(token.get("text", ""))], } collected.setdefault("invoice_records", []).append(record) collected["invoice_doc_id"] = doc_id collected["invoice_tokens"] = tokens collected["invoice_fields"] = fields collected["invoice_evidence"] = evidence collected["invoice_line_items"] = line_items collected["invoice_line_tokens"] = record["line_tokens"] def capture_email_data(collected: dict[str, Any], tool_result: dict[str, Any]) -> None: doc_id = str(tool_result.get("doc_id", "")) tokens = list(tool_result.get("tokens", []) or []) collected["email_doc_id"] = doc_id collected["email_tokens"] = tokens collected["email_evidence"] = parse_email_tokens(tokens, doc_id) def update_collected_from_tool_result( collected: dict[str, Any], action: LedgerShieldAction, tool: dict[str, Any], *, email_doc_id: str, ) -> None: tool_name = tool.get("tool_name") _capture_tool_artifacts(collected, tool) if not tool.get("success", False): _record_tool_failure(collected, action, tool) if tool_name == "lookup_vendor" and tool.get("success"): collected["vendor"] = tool.get("vendor") or {} refresh_email_thread_from_ocr(collected) elif tool_name == "lookup_po" and tool.get("success"): collected["po"] = tool.get("po") elif tool_name == "lookup_receipt" and tool.get("success"): collected["receipt"] = tool.get("receipt") elif tool_name == "search_ledger" and tool.get("success"): hits = list(tool.get("hits", []) or []) existing_ids = {row.get("ledger_id") for row in collected["ledger_hits"]} for hit in hits: if hit.get("ledger_id") not in existing_ids: collected["ledger_hits"].append(hit) existing_ids.add(hit.get("ledger_id")) invoice_key = normalize_text(action.payload.get("invoice_number")) if invoice_key: collected["ledger_queries"][invoice_key] = tool collected["ledger_search"] = tool elif tool_name == "lookup_vendor_history" and tool.get("success"): collected["vendor_history"] = list(tool.get("history", []) or []) elif tool_name == "inspect_email_thread" and tool.get("success"): collected["email_thread"] = tool.get("thread") or {} refresh_email_thread_from_ocr(collected) elif tool_name == "compare_bank_account" and tool.get("success"): collected["bank_compare"] = tool collected["bank_compares"].append(tool) elif tool_name == "lookup_policy" and tool.get("success"): collected["policies"] = list(tool.get("policies", []) or []) elif tool_name == "ocr" and tool.get("success") and tool.get("doc_id") == email_doc_id: capture_email_data(collected, tool) refresh_email_thread_from_ocr(collected) def run_episode_with_env( env: Any, case_id: str, client: Optional[OpenAI], *, temperature: float = TEMPERATURE, emit_logs: bool = True, ) -> dict[str, Any]: rewards: list[float] = [] steps_taken = 0 final_score = 0.0 success = False task_type = "unknown" final_decision = "" score_breakdown: dict[str, Any] = {} pressure_resistance = 0.0 if emit_logs: log_start(task=case_id, env=BENCHMARK, model=MODEL_NAME) try: reset_result = env.reset(case_id=case_id) observation = reset_result.observation task_type = observation.task_type max_steps = int(getattr(observation, "max_steps", MAX_STEPS) or MAX_STEPS) step_no = 1 profile = current_model_profile() step_limit = min(int(getattr(observation, "max_steps", MAX_STEPS) or MAX_STEPS), MAX_STEPS) collected: dict[str, Any] = { "invoice_doc_id": "", "invoice_tokens": [], "invoice_fields": {}, "invoice_evidence": {}, "invoice_line_items": [], "invoice_line_tokens": [], "invoice_records": [], "email_doc_id": "", "email_tokens": [], "email_evidence": {}, "po": None, "receipt": None, "ledger_hits": [], "ledger_queries": {}, "ledger_search": {}, "vendor_history": [], "email_thread": {}, "bank_compare": None, "bank_compares": [], "vendor": {}, "revealed_artifacts": {}, "pending_events": [], "observed_risk_signals": [], "callback_result": {}, "bank_change_approval_chain": {}, "po_reconciliation_report": {}, "receipt_reconciliation_report": {}, "duplicate_cluster_report": {}, "tool_failures": {}, "case_instruction": str(getattr(observation, "instruction", "") or ""), "case_metadata": dict(getattr(observation, "case_metadata", {}) or {}), "_client": client, } update_collected_from_observation(collected, observation) executed_signatures: set[str] = set() invoice_doc_ids: list[str] = [] email_doc_id = "" for doc in observation.visible_documents: doc_type = normalize_text(doc.get("doc_type")) if doc_type == "invoice": invoice_doc_ids.append(str(doc.get("doc_id"))) if doc_type == "email" and not email_doc_id: email_doc_id = str(doc.get("doc_id")) if not invoice_doc_ids: raise RuntimeError(f"No visible invoice document for {case_id}") for invoice_doc_id in invoice_doc_ids: invoice_ocr_action = LedgerShieldAction( action_type="ocr", payload={"doc_id": invoice_doc_id, "mode": "accurate"}, ) executed_signatures.add(action_signature(invoice_ocr_action)) ocr_invoice_result, step_no = perform_step( env, step_no, rewards, invoice_ocr_action, emit_logs=emit_logs, ) steps_taken = step_no - 1 capture_invoice_data(collected, ocr_invoice_result.observation.last_tool_result) if task_type not in {"task_d", "task_e"}: break if task_type == "task_a": invoice_doc_id = invoice_doc_ids[0] zoom_result, step_no = perform_step( env, step_no, rewards, LedgerShieldAction( action_type="zoom", payload={"doc_id": invoice_doc_id, "bbox": [0, 0, 400, 400]}, ), emit_logs=emit_logs, ) steps_taken = step_no - 1 if zoom_result.done: final_score = normalize_score(zoom_result.info.get("final_score", rewards[-1] if rewards else 0.0)) success = final_score >= SUCCESS_SCORE_THRESHOLD submit_payload = build_final_submission(task_type, collected, {}) final_result, step_no = perform_step( env, step_no, rewards, LedgerShieldAction(action_type="submit_decision", payload=submit_payload), emit_logs=emit_logs, ) steps_taken = step_no - 1 final_score = normalize_score(final_result.info.get("final_score", final_result.reward or 0.0)) success = final_score >= SUCCESS_SCORE_THRESHOLD final_decision = str((final_result.observation.last_tool_result or {}).get("decision", submit_payload.get("decision", ""))) score_breakdown = dict(final_result.info.get("score_breakdown", {}) or {}) pressure_resistance = float(final_result.info.get("pressure_resistance_score", 0.0) or 0.0) return { "case_id": case_id, "task_type": task_type, "score": final_score, "steps": steps_taken, "final_decision": final_decision, "score_breakdown": score_breakdown, "pressure_resistance_score": pressure_resistance, } invoice_fields = collected["invoice_fields"] vendor_key = vendor_key_for(invoice_fields) if not vendor_key: vendor_key = normalize_text((collected.get("email_thread") or {}).get("vendor_key")) po_id = str(invoice_fields.get("po_id", "")).strip() receipt_id = str(invoice_fields.get("receipt_id", "")).strip() invoice_total = safe_float(invoice_fields.get("total")) invoice_number = str(invoice_fields.get("invoice_number", "")).strip() proposed_bank_account = str(invoice_fields.get("bank_account", "")).strip() def execute_action_batch(actions: list[LedgerShieldAction]) -> bool: nonlocal step_no, steps_taken, final_score, success, final_decision, score_breakdown, pressure_resistance for action in actions: if step_no > step_limit: break executed_signatures.add(action_signature(action)) result, step_no = perform_step(env, step_no, rewards, action, emit_logs=emit_logs, collected=collected) steps_taken = step_no - 1 tool = result.observation.last_tool_result or {} update_collected_from_tool_result(collected, action, tool, email_doc_id=email_doc_id) update_collected_from_observation(collected, result.observation) if result.done: final_score = normalize_score(result.info.get("final_score", result.reward or 0.0)) success = final_score >= SUCCESS_SCORE_THRESHOLD final_decision = str(tool.get("decision", "")) score_breakdown = dict(result.info.get("score_breakdown", {}) or {}) pressure_resistance = float(result.info.get("pressure_resistance_score", 0.0) or 0.0) return True return False def next_pending_artifact_action() -> LedgerShieldAction | None: pending_ids = { normalize_text(event.get("artifact_id")) for event in collected.get("pending_events", []) or [] if normalize_text(event.get("artifact_id")) } high_value_pending = { "task_b": { "callback_verification_result", "po_reconciliation_report", "receipt_reconciliation_report", }, "task_c": { "callback_verification_result", "bank_change_approval_chain", "duplicate_cluster_report", }, "task_d": { "callback_verification_result", "bank_change_approval_chain", "duplicate_cluster_report", }, "task_e": { "callback_verification_result", "bank_change_approval_chain", "duplicate_cluster_report", }, }.get(task_type, set()) if not (pending_ids & high_value_pending): return None if task_type == "task_c": pending_vendor_key = normalize_text((collected.get("vendor") or {}).get("vendor_key")) or vendor_key_for( collected.get("invoice_fields", {}) or {} ) if pending_vendor_key: return LedgerShieldAction( action_type="lookup_vendor", payload={"vendor_key": pending_vendor_key}, ) payload: dict[str, Any] = {} if invoice_total: payload["amount"] = invoice_total if invoice_number: payload["invoice_number"] = invoice_number if payload: return LedgerShieldAction(action_type="search_ledger", payload=payload) return LedgerShieldAction(action_type="lookup_policy", payload={}) def drain_pending_artifacts() -> bool: nonlocal step_no, steps_taken, final_score, success, final_decision, score_breakdown, pressure_resistance wait_budget = 2 if profile.tier == "elite" else (1 if profile.tier == "strong" else 0) while wait_budget > 0 and step_no < step_limit: wait_action = next_pending_artifact_action() if wait_action is None: break executed_signatures.add(action_signature(wait_action)) result, step_no = perform_step(env, step_no, rewards, wait_action, emit_logs=emit_logs, collected=collected) steps_taken = step_no - 1 tool = result.observation.last_tool_result or {} update_collected_from_tool_result(collected, wait_action, tool, email_doc_id=email_doc_id) update_collected_from_observation(collected, result.observation) wait_budget -= 1 if result.done: final_score = normalize_score(result.info.get("final_score", result.reward or 0.0)) success = final_score >= SUCCESS_SCORE_THRESHOLD final_decision = str(tool.get("decision", "")) score_breakdown = dict(result.info.get("score_breakdown", {}) or {}) pressure_resistance = float(result.info.get("pressure_resistance_score", 0.0) or 0.0) return True return False remaining_action_slots = max(0, step_limit - step_no + 1) investigation_budget = { "task_b": 3, "task_c": 3, "task_d": 6, "task_e": 8, }.get(task_type, 0) + profile.investigation_budget_bonus investigation_candidates = build_investigation_candidates( task_type, collected, vendor_key=vendor_key, po_id=po_id, receipt_id=receipt_id, invoice_total=invoice_total, invoice_number=invoice_number, proposed_bank_account=proposed_bank_account, email_doc_id=email_doc_id, executed_signatures=executed_signatures, ) planned_investigation = llm_plan_actions( client, task_type=task_type, phase="investigation", collected=collected, candidates=investigation_candidates, max_actions=min(investigation_budget, remaining_action_slots), ) if execute_action_batch(planned_investigation): return { "case_id": case_id, "task_type": task_type, "score": final_score, "steps": steps_taken, "final_decision": final_decision, "score_breakdown": score_breakdown, "pressure_resistance_score": pressure_resistance, "rl_trace": collected.get("rl_trace", []), } submit_payload = repair_submission(task_type, build_final_submission(task_type, collected, {}), collected) remaining_action_slots = max(0, step_limit - step_no + 1) intervention_budget = { "task_b": 3, "task_c": 5, "task_d": 5, "task_e": 6, }.get(task_type, 0) + profile.intervention_budget_bonus intervention_candidates = build_intervention_candidates( task_type, collected, submit_payload, executed_signatures=executed_signatures, ) planned_interventions = llm_plan_actions( client, task_type=task_type, phase="intervention", collected=collected, candidates=intervention_candidates, max_actions=min(intervention_budget, remaining_action_slots), current_submission=submit_payload, ) if execute_action_batch(planned_interventions): return { "case_id": case_id, "task_type": task_type, "score": final_score, "steps": steps_taken, "final_decision": final_decision, "score_breakdown": score_breakdown, "pressure_resistance_score": pressure_resistance, "rl_trace": collected.get("rl_trace", []), } if drain_pending_artifacts(): return { "case_id": case_id, "task_type": task_type, "score": final_score, "steps": steps_taken, "final_decision": final_decision, "score_breakdown": score_breakdown, "pressure_resistance_score": pressure_resistance, "rl_trace": collected.get("rl_trace", []), } submit_payload = repair_submission(task_type, build_final_submission(task_type, collected, {}), collected) if step_no <= step_limit: final_result, step_no = perform_step( env, step_no, rewards, LedgerShieldAction(action_type="submit_decision", payload=submit_payload), emit_logs=emit_logs, ) steps_taken = step_no - 1 final_score = normalize_score(final_result.info.get("final_score", final_result.reward or 0.0)) success = final_score >= SUCCESS_SCORE_THRESHOLD last_tool = final_result.observation.last_tool_result or {} final_decision = str(last_tool.get("decision", submit_payload.get("decision", ""))) score_breakdown = dict(final_result.info.get("score_breakdown", {}) or {}) pressure_resistance = float(final_result.info.get("pressure_resistance_score", 0.0) or 0.0) else: final_score = normalize_score(rewards[-1] if rewards else 0.0) success = False return { "case_id": case_id, "task_type": task_type, "score": final_score, "steps": steps_taken, "final_decision": final_decision, "score_breakdown": score_breakdown, "pressure_resistance_score": pressure_resistance, "rl_trace": collected.get("rl_trace", []), } except Exception as exc: # noqa: BLE001 trace(f"[ERROR] episode failed for {case_id}: {exc}") return { "case_id": case_id, "task_type": task_type, "score": 0.0, "steps": steps_taken, "final_decision": "", "score_breakdown": {}, "pressure_resistance_score": 0.0, "error": str(exc), "rl_trace": [], } finally: try: env.close() except Exception as exc: # noqa: BLE001 trace(f"[DEBUG] env.close failed for {case_id}: {exc}") if emit_logs: log_end(success=success, steps=steps_taken, rewards=rewards, score=final_score) def run_episode( env_url: str, case_id: str, client: Optional[OpenAI], *, temperature: float = TEMPERATURE, emit_logs: bool = True, ) -> dict[str, Any]: env = LedgerShieldEnv(base_url=env_url) return run_episode_with_env( env=env, case_id=case_id, client=client, temperature=temperature, emit_logs=emit_logs, ) def build_openai_client() -> Optional[OpenAI]: if not HF_TOKEN: trace("[DEBUG] HF_TOKEN not set; running heuristic-only baseline.") return None try: return OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN) except Exception as exc: # noqa: BLE001 trace(f"[DEBUG] failed to initialize OpenAI client: {exc}") return None def run_baseline_inference( env_url: str, cases: list[str], *, temperature: float = TEMPERATURE, pass_k: int = 1, pass_threshold: float = PASSK_SUCCESS_THRESHOLD, emit_logs: bool = True, ) -> dict[str, Any]: client = build_openai_client() results: list[dict[str, Any]] = [] total_trial_successes = 0 total_trials = 0 for case_id in cases: trials = [ run_episode( env_url=env_url, case_id=case_id, client=client, temperature=temperature, emit_logs=emit_logs, ) for _ in range(max(1, int(pass_k))) ] summary = summarize_case_trials(case_id, trials, pass_threshold=pass_threshold) results.append(summary) total_trial_successes += int(summary.get("successful_trials", 0) or 0) total_trials += int(summary.get("trial_count", 0) or 0) if emit_logs and pass_k > 1: trace( "[PASSK] " f"case={case_id} " f"k={int(pass_k)} " f"trial_pass_rate={float(summary.get('trial_pass_rate', 0.0)):.3f} " f"consistent_pass={str(bool(summary.get('pass_k_consistent', False))).lower()}" ) avg_score = sum(result.get("score", 0.0) for result in results) / max(len(results), 1) consistent_pass_rate = sum(bool(result.get("pass_k_consistent", False)) for result in results) / max(len(results), 1) any_pass_rate = sum(bool(result.get("pass_k_any", False)) for result in results) / max(len(results), 1) trial_pass_rate = total_trial_successes / max(total_trials, 1) trace( "[SUMMARY] " f"cases={len(results)} " f"avg_score={avg_score:.4f} " f"scores={compact_json({result['case_id']: result.get('score', 0.0) for result in results})}" ) return { "results": results, "average_score": round(avg_score, 4), "temperature": round(float(temperature), 4), "pass_k": int(pass_k), "pass_threshold": round(float(pass_threshold), 4), "trial_pass_rate": round(trial_pass_rate, 4), "consistent_pass_rate": round(consistent_pass_rate, 4), "any_pass_rate": round(any_pass_rate, 4), "model_profile": asdict(current_model_profile()), } def run_local_baseline( cases: list[str], *, db: dict[str, Any] | None = None, client: Optional[OpenAI] = None, emit_logs: bool = False, temperature: float = TEMPERATURE, pass_k: int = 1, pass_threshold: float = PASSK_SUCCESS_THRESHOLD, ) -> dict[str, Any]: results: list[dict[str, Any]] = [] total_trial_successes = 0 total_trials = 0 for case_id in cases: trials = [ run_episode_with_env( env=LocalLedgerShieldEnv(db=db), case_id=case_id, client=client, emit_logs=emit_logs, temperature=temperature, ) for _ in range(max(1, int(pass_k))) ] summary = summarize_case_trials(case_id, trials, pass_threshold=pass_threshold) results.append(summary) total_trial_successes += int(summary.get("successful_trials", 0) or 0) total_trials += int(summary.get("trial_count", 0) or 0) avg_score = sum(result.get("score", 0.0) for result in results) / max(len(results), 1) consistent_pass_rate = sum(bool(result.get("pass_k_consistent", False)) for result in results) / max(len(results), 1) any_pass_rate = sum(bool(result.get("pass_k_any", False)) for result in results) / max(len(results), 1) return { "results": results, "average_score": round(avg_score, 4), "temperature": round(float(temperature), 4), "pass_k": int(pass_k), "pass_threshold": round(float(pass_threshold), 4), "trial_pass_rate": round(total_trial_successes / max(total_trials, 1), 4), "consistent_pass_rate": round(consistent_pass_rate, 4), "any_pass_rate": round(any_pass_rate, 4), "model_profile": asdict(current_model_profile()), } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="LedgerShield baseline inference") parser.add_argument("--api-url", default=API_BASE_URL) parser.add_argument("--model", default=MODEL_NAME) parser.add_argument("--token", default=HF_TOKEN) parser.add_argument("--env-url", default=ENV_URL) parser.add_argument("--cases", nargs="+", default=DEFAULT_CASES) parser.add_argument("--temperature", type=float, default=TEMPERATURE) parser.add_argument("--passK", type=int, default=1) parser.add_argument("--pass-threshold", type=float, default=PASSK_SUCCESS_THRESHOLD) parser.add_argument("--output-artifact", default="") parser.add_argument("--no-logs", action="store_true") return parser.parse_args() def main() -> None: global API_BASE_URL, MODEL_NAME, HF_TOKEN, ENV_URL args = parse_args() API_BASE_URL = args.api_url MODEL_NAME = args.model HF_TOKEN = args.token ENV_URL = args.env_url emit_logs = True if args.no_logs: trace("[DEBUG] Ignoring --no-logs to preserve benchmark stdout format.") payload = run_baseline_inference( env_url=args.env_url, cases=args.cases, temperature=float(args.temperature), pass_k=max(1, int(args.passK)), pass_threshold=float(args.pass_threshold), emit_logs=emit_logs, ) payload["generated_at"] = datetime.now(timezone.utc).isoformat() payload["model"] = MODEL_NAME payload["env_url"] = args.env_url payload["api_base_url"] = API_BASE_URL if args.output_artifact: write_run_artifact(args.output_artifact, payload) if __name__ == "__main__": main()