The Five Graph Query Patterns That Power Interconnected Cash Risk

Part I: The Query Patterns That Matter

In cash management, "interconnected risk" means understanding how liquidity, regulatory constraints, corporate structure, and payment infrastructure interact. After deploying graph-native treasury systems across multiple fintechs, five query patterns consistently deliver 90% of the risk insight value. These are not academic exercises—they are the workhorses of production treasury intelligence.

Pattern 1: Hierarchical Liquidity Roll-Up with Ownership Weighting

The Question: "What is our true available liquidity in EUR across the entire group, accounting for minority-owned subsidiaries where we can’t freely sweep?"

Why It’s Critical: Naive SUM() across all entities overstates usable liquidity by 20-40% in multinationals with joint ventures. Regulators (ECB, MAS) require ownership-weighted consolidation for LCR reporting.

MATCH path = (cg:CorporateGroup {group_id: $gid})-[:CONTAINS]->(le:LegalEntity)
      -[:HAS_ACCOUNT]->(ba:BankAccount)-[:DENOMINATED_IN]->(cur:Currency {code: 'EUR'})
MATCH (ba)-[:HAS_POSITION]->(cp:CashPosition {is_current: true})

// Walk ownership chain to compute effective control percentage
WITH le, ba, cp, cur,
     [rel IN relationships(path) WHERE type(rel) = 'OWNED_BY' | rel.pct] AS ownership_chain,
     reduce(pct = 100.0, o IN [rel IN relationships(path) 
           WHERE type(rel) = 'OWNED_BY'] | pct * o / 100.0) AS effective_pct

RETURN 
    le.name AS entity,
    cp.available_balance AS gross_balance,
    cp.available_balance * effective_pct / 100.0 AS weighted_available,
    effective_pct,
    CASE WHEN effective_pct < 50 THEN true ELSE false END AS minority_controlled
ORDER BY weighted_available DESC

Risk Insight Revealed: Identifies "phantom liquidity"—cash that appears on consolidated reports but cannot be mobilized during stress. This directly impacts LCR compliance and intraday funding plans.

Pattern 2: Cross-Jurisdictional Sweep Feasibility with Constraint Intersection

The Question: "Can we move $30M from our Shanghai WFOE to the Singapore treasury hub today, and what regulatory limits apply?"

Why It’s Critical: Transfer feasibility depends on the intersection of source jurisdiction rules, destination jurisdiction rules, bilateral treaty terms, and internal policy. No single document contains the complete answer.

MATCH (src:LegalEntity {name: $source})-[:SUBJECT_TO]->(sj:RegulatoryJurisdiction)
MATCH (tgt:LegalEntity {name: $target})-[:SUBJECT_TO]->(tj:RegulatoryJurisdiction)

// Find ALL constraints that apply to this specific corridor
MATCH (constraint:CashMovementConstraint)
WHERE constraint.applies_to_jurisdiction IN [sj.code, tj.code, sj.code + '-' + tj.code]
  AND constraint.effective_date <= datetime()
  AND (constraint.expiry_date IS NULL OR constraint.expiry_date > datetime())

// Check pool connectivity
OPTIONAL MATCH (src)-[:HAS_ACCOUNT]->(sa:BankAccount)
      -[:PARTICIPATES_IN]->(pool:CashPool)<-[:PARTICIPATES_IN]-
      (ta:BankAccount)<-[:HAS_ACCOUNT]-(tgt)

RETURN 
    sj.code AS source_jur,
    tj.code AS target_jur,
    collect({
        constraint_type: constraint.type,
        limit_amount: constraint.limit_amount,
        limit_currency: constraint.currency,
        frequency: constraint.frequency,
        approval_required: constraint.requires_approval,
        regulation_ref: constraint.regulation_code
    }) AS applicable_constraints,
    pool IS NOT NULL AS pool_connected,
    pool.pool_type AS pool_type

Risk Insight Revealed: Produces a constraint intersection matrix rather than a simple yes/no. A $30M transfer might be within SAFE’s annual repatriation limit but exceed the daily PBOC settlement cap—this pattern catches both simultaneously.

Pattern 3: Contagion Path Analysis Through Banking Relationships

The Question: "If Bank X enters resolution, which of our entities lose access to critical payment rails, and what is the cascading liquidity impact?"

Why It’s Critical: Corporate treasuries concentrate banking relationships for efficiency. A single bank failure can paralyze payments across dozens of entities. Relational models cannot trace second-order effects.

// Start from distressed bank
MATCH (bank:Bank {name: $bank_name})

// First order: direct account exposure
MATCH (bank)<-[:HELD_AT]-(ba:BankAccount)<-[:HAS_ACCOUNT]-(le:LegalEntity)
MATCH (ba)-[:HAS_POSITION]->(cp:CashPosition {is_current: true})

// Second order: entities that depend on affected entities for sweeps
OPTIONAL MATCH (le)<-[:OWNED_BY]-(parent:LegalEntity)
OPTIONAL MATCH (le)-[:PARTICIPATES_IN]->(pool:CashPool)<-[:PARTICIPATES_IN]-
         (sibling_ba:BankAccount)<-[:HAS_ACCOUNT]-(sibling:LegalEntity)

// Third order: payment rail dependencies
OPTIONAL MATCH (ba)-[:SENT_PAYMENT]->(pi:PaymentInstruction)-[:VIA_RAIL]->(rail:PaymentRail)
WHERE pi.last_used > datetime() - duration('P30D')

RETURN 
    le.name AS directly_affected,
    cp.available_balance AS exposed_balance,
    collect(DISTINCT sibling.name) AS cascade_entities,
    sum(cp.available_balance) + sum(DISTINCT 0) AS total_direct_exposure,
    collect(DISTINCT rail.name) AS disrupted_rails,
    parent.name AS parent_entity,
    pool.pool_id AS affected_pool
ORDER BY total_direct_exposure DESC

Risk Insight Revealed: Maps the liquidity blast radius of a counterparty event. Enables pre-positioning contingency funding at unaffected banks before stress materializes.

Pattern 4: Temporal Position Drift Detection

The Question: "Which entities have cash positions that deviate significantly from their 30-day baseline, and is this explained by scheduled flows or anomalous?"

Why It’s Critical: Static position snapshots miss behavioral anomalies. A subsidiary whose balance dropped 60% overnight might be experiencing fraud, operational error, or unreported capital calls. Time-series analysis on graph nodes detects drift that threshold alerts miss.

MATCH (le:LegalEntity {name: $entity})-[:HAS_ACCOUNT]->(ba:BankAccount)
      -[:DENOMINATED_IN]->(cur:Currency {code: $currency})
MATCH (ba)-[:HAS_POSITION]->(current:CashPosition {is_current: true})

// Retrieve historical positions for baseline computation
MATCH (ba)-[:HAS_POSITION]->(hist:CashPosition)
WHERE hist.as_of >= datetime() - duration('P30D')
  AND hist.as_of < current.as_of

WITH le, cur, current,
     avg(hist.available_balance) AS avg_30d,
     stDev(hist.available_balance) AS stddev_30d,
     percentileDisc(hist.available_balance, 0.5) AS median_30d,
     count(hist) AS sample_count

WHERE sample_count >= 10  // Sufficient history

WITH le, cur, current, avg_30d, stddev_30d, median_30d,
     (current.available_balance - avg_30d) / CASE WHEN stddev_30d = 0 THEN 1 ELSE stddev_30d END AS z_score,
     (current.available_balance - median_30d) / CASE WHEN median_30d = 0 THEN 1 ELSE median_30d END AS pct_drift

WHERE abs(z_score) > 2.0 OR abs(pct_drift) > 0.40

RETURN 
    le.name AS entity,
    current.available_balance AS current_balance,
    round(avg_30d, 2) AS baseline_avg,
    round(z_score, 2) AS z_score,
    round(pct_drift * 100, 1) AS pct_drift,
    current.as_of AS as_of
ORDER BY abs(z_score) DESC

Risk Insight Revealed: Distinguishes signal from noise in position monitoring. Z-score normalization enables cross-entity comparison regardless of absolute balance magnitude. Feeds directly into AML/fraud investigation workflows.

Pattern 5: Policy-Position Compliance Gap Detection

The Question: "Show me every entity whose current cash position violates an active treasury policy or regulatory requirement."

Why It’s Critical: Policies and regulations are typically stored as documents; positions as data. The compliance gap exists in the relationship between them—a relationship that neither vector search nor SQL can natively express.

// Match entities to their binding policies
MATCH (le:LegalEntity)-[:BOUND_BY]->(tp:TreasuryPolicy {status: 'ACTIVE'})
WHERE tp.effective_date <= datetime()
  AND (tp.expiry_date IS NULL OR tp.expiry_date > datetime())

// Get current position
MATCH (le)-[:HAS_ACCOUNT]->(ba:BankAccount)-[:HAS_POSITION]->(cp:CashPosition {is_current: true})

// Evaluate policy constraints against actual position
WITH le, tp, cp,
     CASE 
       WHEN tp.min_balance_threshold IS NOT NULL 
            AND cp.available_balance < tp.min_balance_threshold 
       THEN {violation: 'BELOW_MINIMUM', 
             threshold: tp.min_balance_threshold, 
             actual: cp.available_balance}
       
       WHEN tp.max_single_bank_concentration IS NOT NULL 
            AND cp.available_balance > tp.max_single_bank_concentration 
       THEN {violation: 'EXCEEDS_CONCENTRATION_LIMIT',
             threshold: tp.max_single_bank_concentration,
             actual: cp.available_balance}
       
       WHEN tp.required_currency_diversification IS NOT NULL
       THEN null  // Requires multi-account aggregation; handled separately
       
       ELSE null
     END AS violation

WHERE violation IS NOT NULL

RETURN 
    le.name AS entity,
    tp.title AS policy,
    tp.policy_id,
    violation.violation AS violation_type,
    violation.threshold,
    violation.actual,
    tp.remediation_guidance
ORDER BY violation.actual ASC

Risk Insight Revealed: Transforms passive policy documents into active compliance monitoring. Instead of annual policy attestation, treasury gets real-time violation detection grounded in actual positions.

Part II: End-to-End Implementation — Multi-Agent LangGraph RAG for Cash Management

Now we operationalize these five patterns into a production system.

Real-Time Use Case: Morning Liquidity Briefing with Regulatory Compliance Validation

Scenario: Every business day at 06:00 local time, the regional treasurer asks:

"Give me my morning liquidity brief. Highlight any policy violations, trapped cash, and whether yesterday’s EUR sweep from Netherlands BV executed within ECB guidelines. Also flag any position drift anomalies."

This single question triggers all five query patterns, RAG retrieval of ECB guidelines, and synthesis into an actionable briefing—all within a persistent session that remembers yesterday’s follow-ups.

Step 1: State Schema Encapsulating All Five Patterns

from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class TreasuryBriefingState(TypedDict):
    """State for daily liquidity briefing with full risk coverage."""
    messages: Annotated[List[Any], add_messages]
    
    # Session identity
    session_id: str
    treasurer_id: str
    group_id: str
    briefing_date: str
    
    # Pattern 1: Hierarchical roll-up
    weighted_liquidity_by_currency: Dict[str, Dict[str, float]]
    minority_controlled_entities: List[str]
    
    # Pattern 2: Sweep feasibility (for referenced corridors)
    sweep_corridor_analysis: List[Dict[str, Any]]
    
    # Pattern 3: Contagion exposure
    bank_concentration_risks: List[Dict[str, Any]]
    
    # Pattern 4: Position drift
    drift_anomalies: List[Dict[str, Any]]
    
    # Pattern 5: Policy violations
    compliance_violations: List[Dict[str, Any]]
    
    # RAG context
    relevant_regulatory_guidance: List[Dict[str, str]]
    applicable_treasury_policies: List[Dict[str, str]]
    
    # Output
    briefing_summary: Optional[str]
    action_items: List[Dict[str, Any]]
    risk_rating: str  # LOW, MODERATE, ELEVATED, CRITICAL
    
    # Control
    phase: str
    requires_escalation: bool
    error_log: List[str]

Step 2: Tool Definitions Wrapping the Five Patterns

from langchain_neo4j import Neo4jGraph
from langchain_core.tools import tool

graph = Neo4jGraph(
    url="bolt://treasury-graph.databases.neo4j.io:7687",
    username="neo4j",
    password="${NEO4J_PASSWORD}",
    database="cash-management",
    read_only=True
)

@tool
def hierarchical_liquidity_rollup(group_id: str, currency: str = None) -> dict:
    """Pattern 1: Ownership-weighted liquidity consolidation."""
    currency_filter = f"AND cur.code = '{currency}'" if currency else ""
    query = f"""
    MATCH path = (cg:CorporateGroup {{group_id: $gid}})-[:CONTAINS]->(le:LegalEntity)
          -[:HAS_ACCOUNT]->(ba:BankAccount)-[:DENOMINATED_IN]->(cur:Currency)
    MATCH (ba)-[:HAS_POSITION]->(cp:CashPosition {{is_current: true}})
    WHERE true {currency_filter}
    WITH le, cur.code AS ccy, cp.available_balance AS bal,
         reduce(pct = 100.0, r IN [x IN relationships(path) 
               WHERE type(x)='OWNED_BY'] | pct * r.pct / 100.0) AS eff_pct
    RETURN ccy, 
           sum(bal) AS gross, 
           sum(bal * eff_pct / 100.0) AS weighted,
           collect(CASE WHEN eff_pct < 50 THEN le.name END) AS minority_entities
    ORDER BY weighted DESC
    """
    results = graph.query(query, {"gid": group_id})
    return {
        "by_currency": {r["ccy"]: {"gross": r["gross"], "weighted": r["weighted"]} for r in results},
        "minority_controlled": list(set(
            e for r in results for e in (r["minority_entities"] or []) if e
        ))
    }

@tool
def detect_position_drift(group_id: str, z_threshold: float = 2.0) -> list:
    """Pattern 4: Statistical anomaly detection across all entities."""
    query = """
    MATCH (cg:CorporateGroup {group_id: $gid})-[:CONTAINS]->(le:LegalEntity)
          -[:HAS_ACCOUNT]->(ba:BankAccount)-[:HAS_POSITION]->(cur_pos:CashPosition {is_current: true})
    MATCH (ba)-[:HAS_POSITION]->(hist:CashPosition)
    WHERE hist.as_of >= datetime() - duration('P30D') AND hist.as_of < cur_pos.as_of
    WITH le, cur_pos, avg(hist.available_balance) AS mu, stDev(hist.available_balance) AS sigma, count(hist) AS n
    WHERE n >= 10 AND sigma > 0
    WITH le, cur_pos, mu, sigma, (cur_pos.available_balance - mu) / sigma AS z
    WHERE abs(z) > $z_thresh
    RETURN le.name AS entity, cur_pos.available_balance AS current,
           round(mu, 2) AS baseline, round(z, 2) AS z_score, cur_pos.as_of AS as_of
    ORDER BY abs(z) DESC LIMIT 20
    """
    return graph.query(query, {"gid": group_id, "z_thresh": z_threshold})

@tool
def find_policy_violations(group_id: str) -> list:
    """Pattern 5: Active compliance gap detection."""
    query = """
    MATCH (cg:CorporateGroup {group_id: $gid})-[:CONTAINS]->(le:LegalEntity)
          -[:BOUND_BY]->(tp:TreasuryPolicy {status: 'ACTIVE'})
    WHERE tp.effective_date <= datetime() AND (tp.expiry_date IS NULL OR tp.expiry_date > datetime())
    MATCH (le)-[:HAS_ACCOUNT]->(ba:BankAccount)-[:HAS_POSITION]->(cp:CashPosition {is_current: true})
    WITH le, tp, cp,
         CASE WHEN tp.min_balance IS NOT NULL AND cp.available_balance < tp.min_balance
              THEN {type:'BELOW_MIN', threshold:tp.min_balance, actual:cp.available_balance}
              WHEN tp.max_concentration IS NOT NULL AND cp.available_balance > tp.max_concentration
              THEN {type:'OVER_CONCENTRATION', threshold:tp.max_concentration, actual:cp.available_balance}
              ELSE null END AS v
    WHERE v IS NOT NULL
    RETURN le.name, tp.title, tp.policy_id, v.type, v.threshold, v.actual, tp.remediation
    ORDER BY v.actual ASC
    """
    return graph.query(query, {"gid": group_id})

@tool
def retrieve_regulatory_context(keywords: List[str]) -> dict:
    """Hybrid RAG for regulations and internal policies."""
    query = """
    CALL db.index.fulltext.queryNodes('treasuryKnowledgeSearch', $q) YIELD node, score
    WHERE node.effective_date <= datetime() AND score > 0.6
    RETURN node.doc_type AS type, node.title, node.ref_id,
           substring(node.content, 0, 600) AS excerpt, score
    ORDER BY score DESC LIMIT 6
    """
    results = graph.query(query, {"q": " ".join(keywords)})
    return {
        "regulations": [r for r in results if r["type"] == "REGULATION"],
        "policies": [r for r in results if r["type"] == "POLICY"]
    }

Step 3: LangGraph Multi-Agent Orchestration

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
import json

llm = ChatOpenAI(model="gpt-4o", temperature=0)
analyst_llm = llm.bind_tools([
    hierarchical_liquidity_rollup, 
    detect_position_drift, 
    find_policy_violations
])
policy_llm = llm.bind_tools([retrieve_regulatory_context])

def risk_analyst_agent(state: TreasuryBriefingState) -> TreasuryBriefingState:
    """Executes Patterns 1, 4, 5 via graph tools."""
    prompt = f"""You are a Treasury Risk Analyst preparing the morning briefing for group {state['group_id']}.
    Execute these analyses:
    1. Hierarchical liquidity roll-up (all currencies)
    2. Position drift detection (z-score > 2.0)
    3. Policy violation scan
    Call all three tools. Do NOT synthesize yet—just gather data."""
    resp = analyst_llm.invoke([{"role": "user", "content": prompt}] + state["messages"])
    return {"messages": [resp], "phase": "risk_analysis"}

def policy_context_agent(state: TreasuryBriefingState) -> TreasuryBriefingState:
    """Retrieves regulatory/policy context based on risk findings."""
    keywords = ["morning briefing", "liquidity coverage", "position monitoring"]
    if state.get("compliance_violations"):
        keywords.extend(["policy violation remediation", "treasury policy breach"])
    if state.get("drift_anomalies"):
        keywords.append("anomalous cash position investigation")
    if state.get("minority_controlled_entities"):
        keywords.append("minority interest liquidity consolidation")
    
    prompt = f"""Retrieve regulations and policies relevant to: {keywords}"""
    resp = policy_llm.invoke([{"role": "user", "content": prompt}])
    return {"messages": [resp], "phase": "policy_context"}

def briefing_synthesizer(state: TreasuryBriefingState) -> TreasuryBriefingState:
    """Produces the final morning briefing with risk rating."""
    prompt = f"""
    You are the Regional Treasurer's AI Assistant. Produce the morning liquidity briefing:
    
    WEIGHTED LIQUIDITY: {json.dumps(state.get('weighted_liquidity_by_currency', {}))}
    MINORITY ENTITIES: {state.get('minority_controlled_entities', [])}
    DRIFT ANOMALIES: {json.dumps(state.get('drift_anomalies', []))}
    POLICY VIOLATIONS: {json.dumps(state.get('compliance_violations', []))}
    
    REGULATORY CONTEXT:
    {chr(10).join(f"- [{r['ref_id']}] {r['title']}: {r['excerpt'][:200]}" 
                  for r in state.get('relevant_regulatory_guidance', []))}
    
    Produce a structured JSON briefing with:
    - executive_summary (2-3 sentences)
    - risk_rating (LOW/MODERATE/ELEVATED/CRITICAL)
    - key_findings (bullet list)
    - action_items (with priority and owner suggestion)
    - regulatory_notes (any compliance considerations)
    """
    resp = llm.invoke(prompt)
    
    try:
        parsed = json.loads(resp.content)
        violations = len(state.get("compliance_violations", []))
        drifts = len(state.get("drift_anomalies", []))
        
        return {
            "messages": [resp],
            "briefing_summary": parsed.get("executive_summary"),
            "action_items": parsed.get("action_items", []),
            "risk_rating": parsed.get("risk_rating", "MODERATE"),
            "phase": "complete",
            "requires_escalation": parsed.get("risk_rating") in ("ELEVATED", "CRITICAL") 
                                   or violations > 3
        }
    except json.JSONDecodeError:
        return {
            "messages": [resp],
            "phase": "complete",
            "error_log": ["Briefing synthesis failed to parse"],
            "requires_escalation": True
        }

# Assemble workflow
workflow = StateGraph(TreasuryBriefingState)
workflow.add_node("risk_analyst", risk_analyst_agent)
workflow.add_node("risk_tools", ToolNode([
    hierarchical_liquidity_rollup, detect_position_drift, find_policy_violations
]))
workflow.add_node("policy_context", policy_context_agent)
workflow.add_node("policy_tools", ToolNode([retrieve_regulatory_context]))
workflow.add_node("synthesizer", briefing_synthesizer)

workflow.set_entry_point("risk_analyst")
workflow.add_edge("risk_analyst", "risk_tools")
workflow.add_edge("risk_tools", "policy_context")
workflow.add_edge("policy_context", "policy_tools")
workflow.add_edge("policy_tools", "synthesizer")
workflow.add_conditional_edges("synthesizer", 
    lambda s: END if not s.get("requires_escalation") else "escalation_handler")

from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://${PG_CONN}")
app = workflow.compile(checkpointer=checkpointer)

Step 4: Scheduled + Interactive Execution

from datetime import datetime

async def generate_morning_briefing(group_id: str, treasurer_id: str):
    """Called by scheduler at 06:00 or on-demand."""
    date_str = datetime.utcnow().strftime("%Y-%m-%d")
    session_id = f"BRIEF-{date_str}-{group_id}"
    config = {"configurable": {"thread_id": session_id}}
    
    initial_state: TreasuryBriefingState = {
        "messages": [{"role": "user", "content": "Generate morning liquidity briefing."}],
        "session_id": session_id,
        "treasurer_id": treasurer_id,
        "group_id": group_id,
        "briefing_date": date_str,
        "phase": "risk_analysis",
        "requires_escalation": False,
        "error_log": []
    }
    
    async for event in app.astream_events(initial_state, config=config, version="v2"):
        kind = event["event"]
        if kind == "on_tool_end":
            yield {"type": "analysis_complete", "tool": event["name"], 
                   "data": event["data"]["output"]}
        elif kind == "on_chat_model_stream":
            yield {"type": "briefing_token", "content": event["data"]["chunk"].content}
        elif kind == "on_chain_end" and event["name"] == "synthesizer":
            yield {"type": "briefing_ready", "session_id": session_id,
                   "risk_rating": event["data"]["output"].get("risk_rating")}

# Follow-up in same session (memory-enabled)
async def follow_up(session_id: str, question: str):
    """Treasurer asks follow-up; state retains all prior analysis."""
    config = {"configurable": {"thread_id": session_id}}
    async for event in app.astream_events(
        {"messages": [{"role": "user", "content": question}]},
        config=config, version="v2"
    ):
        if event["event"] == "on_chat_model_stream":
            yield {"type": "response_token", "content": event["data"]["chunk"].content}

Performance & Compliance Notes

Concern	Solution
Roll-up latency across 500+ entities	Materialized WeightedLiquidity view refreshed via CDC; live traversal only for ad-hoc
Drift detection on 10K accounts	Pre-computed 30-day statistics updated nightly; real-time only for flagged entities
Policy violation scan frequency	Incremental evaluation: only re-check entities with position changes since last scan
Regulatory RAG freshness	Embedding pipeline triggered on policy document version change; stale embeddings auto-invalidated
Audit trail	Every LangGraph checkpoint includes tool inputs/outputs; immutable append-only log
Data residency	Jurisdiction-tagged graph partitions; queries routed to geo-appropriate replicas

Conclusion

Interconnected cash risk lives in the relationships between positions, structures, constraints, and time—not in any single data source. The five query patterns described here (hierarchical roll-up, corridor feasibility, contagion mapping, temporal drift, compliance gaps) cover the vast majority of treasury risk questions. When orchestrated through LangGraph’s stateful multi-agent architecture with RAG-augmented policy context, they transform cash management from reactive position reporting to proactive, structurally-grounded liquidity intelligence. This implementation is for architectural reference. Production treasury systems require validation against your institution’s ALM framework, Basel III model risk review, SOX control testing, and coordination with regulatory reporting teams. Monetary computations must use Decimal arithmetic; floats shown here are for readability. Never automate fund movements based solely on LLM-generated recommendations without independent human verification and dual-control approval.