AI Agent Incident Reconstruction

Walkthrough of a multi-agent hedge fund workflow where a risk limit violation triggers a human-in-the-loop review and downstream forensic replay.

Workflow Pattern

Analyst → Risk → Executor

Autonomous multi-agent chain

Incident Trigger

Risk Limit Violation

Value exceeds $50,000 threshold

Resolution Path

HITL Review + Replay

Causal timeline & root cause

Step 01

Multi-Agent Instrumentation Setup

An analyst agent produces a trading signal, a risk agent checks portfolio limits, and an executor fills the trade. A single WORKFLOW_ID binds the execution context across the workflow so every decision and tool call can be reconstructed later.

multi_agent_workflow.py

python

import os
import random
from uuid import uuid4
import blocklog

# Initialize SDK
blocklog.init(api_key=os.environ.get("BLOCKLOG_API_KEY", "blk_demo_key"))
WORKFLOW_ID = str(uuid4())

@blocklog.tool(name="fetch-price")
def fetch_price(ticker: str) -> float:
    return {"TSLA": 412.50, "AAPL": 189.30}.get(ticker, 250.0)

@blocklog.tool(name="check-risk-limits")
def check_risk_limits(ticker: str, qty: int, price: float) -> dict:
    trade_value = qty * price
    return {
        "approved": trade_value < 50_000,
        "trade_value": trade_value,
        "limit": 50_000,
    }

@blocklog.agent(name="market-analyst", version="2.1")
def analyst_agent(ticker: str) -> dict:
    price = fetch_price(ticker)
    score = 0.85
    signal = "BUY"

    with blocklog.decision(
        type="SIGNAL",
        asset=ticker,
        confidence=score,
        metadata={"workflow_id": WORKFLOW_ID},
    ) as d:
        d.record_input(price=price)
        d.record_output(signal=signal, score=score)

    return {"ticker": ticker, "price": price, "signal": signal, "decision_id": d.id}

@blocklog.agent(name="risk-manager", version="1.5")
def risk_agent(ticker: str, price: float, qty: int, analyst_dec_id: str) -> dict:
    risk = check_risk_limits(ticker, qty, price)

    with blocklog.decision(
        type="RISK_APPROVAL",
        asset=ticker,
        confidence=1.0 if risk["approved"] else 0.0,
        metadata={"workflow_id": WORKFLOW_ID},
    ) as d:
        d.record_input(qty=qty, price=price, analyst_decision_id=analyst_dec_id)
        d.record_output(approved=risk["approved"])

        if not risk["approved"]:
            d.request_approval(
                reason=f"Value ${risk['trade_value']:.0f} exceeds limit ${risk['limit']:.0f}",
                reviewer="cro@fund.com",
            )

    return {**risk, "risk_decision_id": d.id}

if __name__ == "__main__":
    TICKER = "TSLA"
    QTY = 150  # 150 * 412.50 = $61,875

    analysis = analyst_agent(TICKER)
    risk_status = risk_agent(TICKER, analysis["price"], QTY, analysis["decision_id"])

Step 02

Query Forensics and Automated Root Cause

Once an exception or human review request is raised, the forensic replay API reconstructs the timeline, surfaces the causal chain, and helps identify the root cause behind the incident.

forensic_replay.py

python

import blocklog

# 1. Initialize forensic replay session for the trace
session = blocklog.replay(trace_id="trc_8f92a10b4c7e")

# 2. Iterate through causal timeline in sequence
for event in session.timeline():
    print(f"[{event.get('at')}] {event.get('item_type')}: {event.get('summary')}")

# 3. Request automated root cause inference
cause = session.root_cause()
if cause["detected"]:
    print(f"Incident: {cause['root_cause_type']}")
    print(f"Explanation: {cause['description']}")
    print(f"Remediation: {cause['remediation']}")

What this example demonstrates

Deterministic Context Binding: A shared workflow identifier links the signal, risk evaluation, and review path into one auditable trail.
Explicit Decision Objects: Risk approval becomes an explicit, verifiable decision entity rather than an opaque internal boolean check.
Unified Lifecycle: Human escalation and post-incident replay investigation belong to the exact same cryptographic audit trail.