Tutorial 17 — Advanced Trust & Behavior Monitoring¶

pip install agentmesh-platform

from agentmesh.services.reward_engine import RewardService
from agentmesh.services.behavior_monitor import AgentBehaviorMonitor
from agentmesh.services.audit import AuditService

AGENT = "did:mesh:analyst-001"

# --- Behavior monitor: detects anomalous runtime patterns ---
monitor = AgentBehaviorMonitor(
    burst_threshold=100,           # max calls per window
    burst_window_seconds=60,       # sliding window size
    consecutive_failure_threshold=20,
    capability_denial_threshold=10,
)

# --- Reward service: multi-dimension trust scoring ---
reward = RewardService()

# --- Audit service: tamper-proof event log ---
audit = AuditService()

# Record a successful tool invocation
monitor.record_tool_call(AGENT, "file_read", success=True)
reward.record_task_success(AGENT, task_id="task-42")
audit.log_action(AGENT, action="file_read", outcome="success")

score = reward.get_score(AGENT)
print(score.total_score)  # 500+ (default boosted by success signals)
print(score.tier)         # "standard"

# Record a failure — score decreases
reward.record_task_failure(AGENT, reason="timeout")
monitor.record_tool_call(AGENT, "db_write", success=False)
audit.log_action(AGENT, action="db_write", outcome="failure")

print(reward.get_score(AGENT).total_score)  # slightly lower

from agentmesh.services.reward_engine import RewardService
from agentmesh.reward.engine import RewardConfig

config = RewardConfig(
    revocation_threshold=300,        # auto-revoke below this score
    warning_threshold=500,           # emit warnings below this score
    policy_compliance_weight=0.25,
    resource_efficiency_weight=0.15,
    output_quality_weight=0.20,
    security_posture_weight=0.25,
    collaboration_health_weight=0.15,
)

reward = RewardService(config=config)

AGENT = "did:mesh:worker-007"

reward.record_task_success(AGENT, task_id="task-99")        # policy + quality ↑
reward.record_task_failure(AGENT, reason="Rate limit exceeded")  # quality ↓
reward.record_policy_violation(AGENT, policy_name="no-external-api")
reward.record_handshake(AGENT, peer_did="did:mesh:peer-003", success=True)
reward.record_security_event(AGENT, within_boundary=True, event_type="data_access")
reward.record_security_event(AGENT, within_boundary=False, event_type="egress_attempt")

score = reward.get_score(AGENT)

print(score.total_score)     # 0–1000
print(score.tier)            # "verified_partner" | "trusted" | "standard"
                             # | "probationary" | "untrusted"
print(score.score_change)    # delta since last calculation

# Detailed breakdown
explanation = reward.engine.get_score_explanation(AGENT)
for dim_name, dim_info in explanation["dimensions"].items():
    print(f"  {dim_name}: score={dim_info['score']:.1f}  "
          f"signals={dim_info['signal_count']}  "
          f"contribution={dim_info['contribution']:.1f}")

print(f"Trend: {explanation['trend']}")  # "improving" | "stable" | "degrading"

def on_agent_revoked(agent_did: str, reason: str):
    print(f"REVOKED: {agent_did} — {reason}")

reward.on_revocation(on_agent_revoked)

# Agents approaching revocation
at_risk = reward.engine.get_agents_at_risk()
print(at_risk)  # ["did:mesh:risky-agent"]

# Agents already below a custom threshold
below = reward.agents_below_threshold(threshold=400)

# Full summary across all tracked agents
summary = reward.summary()
print(summary)
# {"total_agents": 12, "avg_score": 623.4, "min_score": 310, "max_score": 890}

report = reward.engine.get_health_report(days=7)

print(report["total_agents"])
print(report["revoked_agents"])
print(report["at_risk_agents"])

for did, info in report["agents"].items():
    print(f"{did}: current={info['current_score']}  "
          f"trend={info['trend']}  revoked={info['revoked']}")

from datetime import timedelta
from agentmesh.services.behavior_monitor import AgentBehaviorMonitor

monitor = AgentBehaviorMonitor(
    burst_window_seconds=60,
    burst_threshold=100,
    consecutive_failure_threshold=20,
    capability_denial_threshold=10,
    quarantine_duration=timedelta(minutes=15),
    max_tracked_agents=50_000,  # LRU eviction when exceeded
)

AGENT = "did:mesh:worker-007"

# Successful tool call — resets consecutive-failure counter
monitor.record_tool_call(AGENT, "file_read", success=True)

# Failed tool call — increments consecutive-failure counter
monitor.record_tool_call(AGENT, "db_write", success=False)

# Capability denial — agent tried to access something it shouldn't
monitor.record_capability_denial(AGENT, capability="write:secrets")

# Simulate 21 consecutive failures
for _ in range(21):
    monitor.record_tool_call(AGENT, "flaky_api", success=False)

print(monitor.is_quarantined(AGENT))  # True

metrics = monitor.get_metrics(AGENT)
print(metrics.quarantine_reason)
# "Consecutive failure threshold breached (21 >= 20)"
print(metrics.quarantined_at)
# 2025-01-15T10:32:00+00:00

monitor.release_quarantine(AGENT)
print(monitor.is_quarantined(AGENT))  # False

# If an agent fires 101 calls in 60 seconds:
for i in range(101):
    monitor.record_tool_call(AGENT, "search", success=True)

print(monitor.is_quarantined(AGENT))  # True
metrics = monitor.get_metrics(AGENT)
print(metrics.quarantine_reason)
# "Burst threshold breached (101 calls in 60s)"

quarantined = monitor.get_quarantined_agents()
for m in quarantined:
    print(f"{m.agent_did}: {m.quarantine_reason}")

def on_quarantine(agent_did: str, reason: str):
    """When the monitor quarantines an agent, notify the reward engine."""
    reward.record_policy_violation(agent_did, policy_name=f"quarantine:{reason}")
    audit.log_action(agent_did, action="quarantine", outcome="failure",
                     data={"reason": reason})

# Example: check after each tool call
def handle_tool_call(agent_did: str, tool: str, success: bool):
    monitor.record_tool_call(agent_did, tool, success=success)

    if monitor.is_quarantined(agent_did):
        metrics = monitor.get_metrics(agent_did)
        on_quarantine(agent_did, metrics.quarantine_reason)
    elif success:
        reward.record_task_success(agent_did)
    else:
        reward.record_task_failure(agent_did, reason=f"{tool} failed")

# policies/high-risk-operations.yaml
name: high-risk-operations
version: "1.0"
description: "Restrict high-risk operations to trusted agents"

defaults:
  min_trust_score: 500
  max_delegation_depth: 3
  allowed_namespaces: ["*"]
  require_handshake: true

rules:
  - name: block-untrusted-writes
    description: "Deny write operations for agents below 700 trust"
    condition:
      field: trust_score
      operator: lt
      value: 700
    action: deny
    priority: 10        # lower = evaluated first

  - name: warn-on-delegation
    description: "Warn when delegation depth exceeds 2"
    condition:
      field: delegation_depth
      operator: gt
      value: 2
    action: warn
    priority: 50

  - name: allow-verified-partners
    description: "Allow verified partners unconditionally"
    condition:
      field: trust_tier
      operator: eq
      value: verified_partner
    action: allow
    priority: 1         # highest priority

from agentmesh.governance.trust_policy import TrustPolicy, load_policies

# Load a single policy
policy = TrustPolicy.from_yaml("policies/high-risk-operations.yaml")

# Load all policies from a directory (sorted alphabetically)
policies = load_policies("policies/")

# Evaluate a rule's condition against a runtime context
context = {
    "trust_score": 650,
    "trust_tier": "standard",
    "delegation_depth": 1,
    "agent": {"namespace": "analytics"},
}

for rule in sorted(policy.rules, key=lambda r: r.priority):
    if rule.condition.evaluate(context):
        print(f"Rule '{rule.name}' matched → action: {rule.action}")
        break
else:
    print("No rule matched — applying defaults")
    print(f"  min_trust_score: {policy.defaults.min_trust_score}")
    print(f"  require_handshake: {policy.defaults.require_handshake}")

rules:
  - name: restrict-external-namespace
    condition:
      field: agent.namespace       # resolves context["agent"]["namespace"]
      operator: not_in
      value: ["internal", "core"]
    action: deny
    priority: 20

# Save to YAML
policy.to_yaml("policies/exported.yaml")

# Re-load and verify
reloaded = TrustPolicy.from_yaml("policies/exported.yaml")
assert reloaded.name == policy.name
assert len(reloaded.rules) == len(policy.rules)

Agent Request ──► ShadowModeExecutor
                      │
                      ├─ _validate_request()   → SimulationOutcome
                      ├─ _simulate_execution()  → Fake result per action type
                      ├─ _analyze_impact()       → Side-effect analysis
                      │
                      └─ SimulationResult (logged, never executed)

from agent_control_plane.shadow_mode import (
    ShadowModeConfig,
    ShadowModeExecutor,
    SimulationOutcome,
    add_reasoning_step,
)

config = ShadowModeConfig(
    enabled=True,
    log_reasoning=True,          # capture reasoning chains
    simulate_results=True,       # generate fake outputs per action type
    validate_constraints=True,   # check against constraint graphs
    intercept_all=True,          # full shadow mode — nothing executes
    allow_safe_actions=False,    # even reads are simulated
)

shadow = ShadowModeExecutor(config)

from agent_control_plane.agent_kernel import ExecutionRequest, ActionType

# Build a request (normally done by the agent kernel)
request = ExecutionRequest(
    request_id="req-001",
    agent_context=agent_ctx,
    action_type=ActionType.FILE_WRITE,
    parameters={"path": "/data/report.csv", "content": "..."},
    risk_score=0.3,
)

# Build a reasoning chain (optional)
chain = []
add_reasoning_step(chain, "Analyze data sources", ActionType.FILE_READ,
                   {"path": "/data/input.csv"}, "Need raw data first")
add_reasoning_step(chain, "Write output report", ActionType.FILE_WRITE,
                   {"path": "/data/report.csv"}, "Deliver results")

# Execute in shadow mode
result = shadow.execute_in_shadow(request, reasoning_chain=chain)

print(result.outcome)           # SimulationOutcome.WOULD_SUCCEED
print(result.simulated_result)
# {"action": "file_write", "path": "/data/report.csv",
#  "bytes_written": 3, "note": "... shadow mode simulation ..."}

print(result.actual_impact)
# {"action_type": "file_write", "side_effects": [
#    {"type": "file_system_modification", "path": "/data/report.csv",
#     "reversible": true}]}

stats = shadow.get_statistics()

print(stats["total_simulations"])   # 42
print(stats["success_rate"])        # 0.85
print(stats["policy_violations"])   # 3
print(stats["risk_denials"])        # 2
print(stats["outcome_distribution"])
# {"would_succeed": 36, "policy_violation": 3, "risk_too_high": 2, ...}

violations = shadow.get_policy_violations()
for v in violations:
    print(f"{v.request_id}: {v.outcome.value}")
    for note in v.validation_notes:
        print(f"  - {note}")

PENDING → SHADOW → CANARY → PROMOTING → COMPLETE
                                    ↘ ROLLED_BACK / FAILED

from agentmesh.governance._shadow_impl import (
    DeploymentStrategy,
    RolloutStep,
    AnalysisCriterion,
)

steps = [
    RolloutStep(
        name="shadow-phase",
        weight=0.0,                  # 0% real traffic
        duration_seconds=3600,
        analysis=[
            AnalysisCriterion(metric="error_rate", threshold=0.01, comparator="lte"),
            AnalysisCriterion(metric="latency_p99", threshold=500, comparator="lte"),
        ],
    ),
    RolloutStep(
        name="canary-phase",
        weight=0.05,                 # 5% real traffic
        duration_seconds=1800,
        analysis=[
            AnalysisCriterion(metric="error_rate", threshold=0.02, comparator="lte"),
        ],
    ),
    RolloutStep(
        name="full-rollout",
        weight=1.0,                  # 100% traffic
        manual_gate=True,            # requires human approval
    ),
]

reward.record_task_success(agent_did, task_id="task-77")

reward.record_task_failure(agent_did, reason="Output rejected by reviewer")

reward.record_task_success(AGENT, task_id="t-100")
audit.log_trust_change(
    AGENT,
    old_score=reward.get_score_value(AGENT) - 5,
    new_score=reward.get_score_value(AGENT),
    reason="task-100 completed successfully",
)

# Agent accessed only authorized resources
reward.record_security_event(
    AGENT,
    within_boundary=True,
    event_type="data_read",
)

# Agent attempted to access an external endpoint (boundary violation)
reward.record_security_event(
    AGENT,
    within_boundary=False,
    event_type="egress_attempt",
)

from agentmesh.reward.scoring import DimensionType

reward.engine.record_signal(
    agent_did=AGENT,
    dimension=DimensionType.SECURITY_POSTURE,
    value=0.5,              # partial violation
    source="network_monitor",
    details="Attempted DNS resolution for unapproved domain",
)

import asyncio
from datetime import datetime, timedelta
from agentmesh.services.registry import AgentRegistry, AgentRegistryEntry

registry = AgentRegistry()

entry = AgentRegistryEntry(
    did="did:mesh:analyst-001",
    name="DataAnalyst",
    description="Runs SQL queries and generates reports",
    organization="Analytics",
    sponsor_email="alice@company.com",
    sponsor_verified=True,
    capabilities=["read:data", "write:reports"],
    supported_protocols=["agentmesh/v1"],
    trust_score=500,                                  # default
    trust_tier="standard",
    public_key_fingerprint="sha256:abc123...",
    svid_serial_number="svid-00001",
    current_credential_expires_at=datetime.utcnow() + timedelta(minutes=15),
    delegation_depth=0,
)

asyncio.run(registry.register(entry))

async def update_trust():
    await registry.update_trust_score("did:mesh:analyst-001", new_score=720)

    agent = await registry.get("did:mesh:analyst-001")
    print(agent.trust_score)  # 720
    print(agent.trust_tier)   # "trusted" (>= 700)

asyncio.run(update_trust())

async def suspend_agent():
    await registry.update_status(
        "did:mesh:analyst-001",
        status="suspended",
        reason="Under investigation for data exfiltration",
    )

    # Heartbeat tracking
    await registry.record_activity("did:mesh:analyst-001")

asyncio.run(suspend_agent())

async def query():
    # List all active agents with trust score >= 700
    trusted = await registry.list_agents(status="active", min_trust_score=700)
    for a in trusted:
        print(f"{a.did}: {a.trust_tier} ({a.trust_score})")

    # Count agents by status
    total = await registry.count_agents()
    suspended = await registry.count_agents(status="suspended")
    print(f"{suspended}/{total} agents suspended")

asyncio.run(query())

async def stats():
    stats = await registry.get_trust_statistics()
    print(stats)
    # {
    #     "total_agents": 42,
    #     "average_trust_score": 623.4,
    #     "min_trust_score": 210,
    #     "max_trust_score": 950,
    #     "tier_distribution": {
    #         "verified_partner": 3,
    #         "trusted": 12,
    #         "standard": 20,
    #         "probationary": 5,
    #         "untrusted": 2,
    #     }
    # }

asyncio.run(stats())

summary = reward.summary()
print(summary)
# {"total_agents": 42, "avg_score": 623.4,
#  "min_score": 210, "max_score": 950}

report = reward.engine.get_health_report(days=7)

print(f"Revoked: {report['revoked_agents']}")
print(f"At risk: {report['at_risk_agents']}")

# Per-agent breakdown
for did, info in report["agents"].items():
    trend = info["trend"]  # "improving" | "stable" | "degrading"
    print(f"  {did}: {info['current_score']}  [{trend}]")

  AuditService            AuditLog              MerkleAuditChain
  ┌──────────┐       ┌──────────────┐       ┌──────────────────┐
  │ log_*()  │──────►│ Indexes:     │──────►│ Entries[]        │
  │ query_*()│       │   by_agent   │       │ Merkle Tree      │
  │ verify() │       │   by_type    │       │ Root Hash        │
  │ summary()│       │ Sink (opt.)  │       │ Inclusion Proofs │
  └──────────┘       └──────────────┘       └──────────────────┘

from agentmesh.services.audit import AuditService

audit = AuditService()

# Log an action
entry = audit.log_action(
    agent_did="did:mesh:worker-007",
    action="file_write",
    outcome="success",
    resource="/data/report.csv",
    data={"bytes_written": 4096},
    trace_id="trace-abc-123",
)
print(entry.entry_id)     # "audit_a1b2c3d4..."
print(entry.entry_hash)   # SHA-256 hash
print(entry.previous_hash)  # hash of previous entry (chain link)

# Log a policy decision
audit.log_policy_decision(
    agent_did="did:mesh:worker-007",
    action="db_write",
    decision="deny",
    policy_name="read-only-agents",
)

# Log a trust handshake
audit.log_handshake(
    initiator_did="did:mesh:worker-007",
    responder_did="did:mesh:peer-003",
    success=True,
)

# Log a trust score change
audit.log_trust_change(
    agent_did="did:mesh:worker-007",
    old_score=500.0,
    new_score=650.0,
    reason="Completed 10 tasks without violations",
)

# Verify the entire chain
is_valid = audit.verify_chain()
print(is_valid)  # True

# Get a summary
summary = audit.summary()
print(summary)
# {"total_entries": 128, "chain_valid": True,
#  "root_hash": "a3f8b2c1d4e5..."}

# Get an inclusion proof for a specific entry
proof_data = audit.chain._log.get_proof(entry.entry_id)
# Returns: {"entry": {...}, "merkle_proof": [...], "merkle_root": "...",
#           "verified": True}

# Verify the proof independently
chain = audit.chain
proof = chain.get_proof(entry.entry_id)
root = chain.get_root_hash()
is_included = chain.verify_proof(entry.entry_hash, proof, root)
print(is_included)  # True

# By agent
entries = audit.query_by_agent("did:mesh:worker-007")

# By event type
policy_entries = audit.query_by_type("policy_decision")

# Advanced: use the underlying AuditLog for complex queries
from datetime import datetime, timezone

results = audit._log.query(
    agent_did="did:mesh:worker-007",
    event_type="agent_action",
    start_time=datetime(2025, 1, 1, tzinfo=timezone.utc),
    outcome="failure",
    limit=50,
)

from agentmesh.governance.audit_backends import FileAuditSink, HashChainVerifier

sink = FileAuditSink(
    path="audit/events.jsonl",
    secret_key=b"your-256-bit-secret-key",
    max_file_size=10 * 1024 * 1024,  # 10 MB — auto-rotates
)

# Create an AuditLog with the sink attached
from agentmesh.governance.audit import AuditLog

log = AuditLog(sink=sink)
log.log("agent_action", "did:mesh:worker-007", "file_read",
        resource="/data/input.csv")

# Each line in audit/events.jsonl includes:
# - content_hash (SHA-256 of canonical payload)
# - previous_hash (chain link to prior entry)
# - signature (HMAC-SHA256 using your secret key)

# Verify the entire file
verifier = HashChainVerifier()
is_valid, errors = verifier.verify_file("audit/events.jsonl",
                                         secret_key=b"your-256-bit-secret-key")
print(is_valid)  # True
print(errors)    # [] (empty if valid)

# Read entries back
entries = sink.read_entries()
for e in entries:
    assert e.verify(b"your-256-bit-secret-key")

# Close when done
sink.close()

events = audit._log.export_cloudevents()
for ce in events:
    print(ce["type"])    # e.g., "ai.agentmesh.tool.invoked"
    print(ce["source"])  # "agentmesh"
    print(ce["data"])    # entry payload

from agentmesh.services.rate_limiter import RateLimiter, RateLimitConfig

config = RateLimitConfig(
    global_rate=100.0,            # tokens/second (global)
    global_capacity=200,          # max burst (global)
    per_agent_rate=10.0,          # tokens/second (per agent)
    per_agent_capacity=20,        # max burst (per agent)
    backpressure_threshold=0.8,   # signal backpressure at 80% usage
)

limiter = RateLimiter(
    global_rate=config.global_rate,
    global_capacity=config.global_capacity,
    per_agent_rate=config.per_agent_rate,
    per_agent_capacity=config.per_agent_capacity,
    backpressure_threshold=config.backpressure_threshold,
)

AGENT = "did:mesh:worker-007"

# Simple allow/deny check
if limiter.allow(AGENT):
    print("Request allowed")
else:
    print("Rate limited")

# Detailed check with retry information
result = limiter.check(AGENT)
print(result.allowed)              # True / False
print(result.remaining_tokens)     # tokens left
print(result.retry_after_seconds)  # seconds to wait (if denied)
print(result.backpressure)         # True if approaching limit

# Global + per-agent status
status = limiter.get_status(agent_did=AGENT)
print(status)
# {"global_tokens": 195.3, "global_capacity": 200,
#  "agent_tokens": 18.7, "agent_capacity": 20}

# Reset a single agent's bucket
limiter.reset(agent_did=AGENT)

# Reset everything
limiter.reset()

from agentmesh.reward.trust_decay import NetworkTrustEngine, TrustEvent

network = NetworkTrustEngine(
    decay_rate=2.0,           # score points lost per hour of inactivity
    propagation_factor=0.3,   # fraction of impact propagated to neighbors
    propagation_depth=2,      # hops
    regime_threshold=0.5,     # KL divergence threshold for alerts
)

AGENT_A = "did:mesh:agent-a"
AGENT_B = "did:mesh:agent-b"

network.set_score(AGENT_A, 800)
network.set_score(AGENT_B, 750)

# Record interactions to build the agent graph
network.record_interaction(AGENT_A, AGENT_B)
network.record_interaction(AGENT_B, AGENT_A)

event = TrustEvent(
    agent_did=AGENT_A,
    event_type="policy_violation",
    severity_weight=0.5,         # 0.0 = minor, 1.0 = critical
    details="Attempted unauthorized API call",
)

deltas = network.process_trust_event(event)
print(deltas)
# {"did:mesh:agent-a": -50.0, "did:mesh:agent-b": -7.5}
# Agent A lost 50 points; Agent B (neighbor) lost 7.5 (propagated)

# Apply decay — agents without recent positive signals lose trust
deltas = network.apply_temporal_decay()
for did, loss in deltas.items():
    print(f"{did}: {loss:.1f}")  # e.g., "did:mesh:agent-a: -4.0"

# Positive signals reset the decay clock
network.record_positive_signal(AGENT_A, bonus=5.0)

# Record a variety of actions to build a baseline
for _ in range(50):
    network.record_action(AGENT_A, "file_read")
    network.record_action(AGENT_A, "db_query")

# Suddenly the agent shifts behavior
for _ in range(20):
    network.record_action(AGENT_A, "api_call")  # new dominant action

alert = network.detect_regime_change(AGENT_A)
if alert:
    print(f"Regime change detected!")
    print(f"  KL divergence: {alert.kl_divergence:.3f}")
    print(f"  Threshold: {alert.threshold}")
    print(f"  Recent: {alert.recent_distribution}")
    print(f"  Historical: {alert.historical_distribution}")

import asyncio
from datetime import datetime, timedelta
from agentmesh.services.reward_engine import RewardService
from agentmesh.services.behavior_monitor import AgentBehaviorMonitor
from agentmesh.services.audit import AuditService
from agentmesh.services.rate_limiter import RateLimiter
from agentmesh.services.registry import AgentRegistry, AgentRegistryEntry
from agentmesh.governance.trust_policy import TrustPolicy


class GovernanceOrchestrator:
    """Wires together all trust and behavior components."""

    def __init__(self):
        self.reward = RewardService()
        self.monitor = AgentBehaviorMonitor(
            burst_threshold=50,
            consecutive_failure_threshold=10,
        )
        self.audit = AuditService()
        self.limiter = RateLimiter()
        self.registry = AgentRegistry()
        self.policy = TrustPolicy.from_yaml("policies/default.yaml")

    async def handle_request(self, agent_did: str, tool: str) -> bool:
        """Process an agent tool-call request through the governance stack."""

        # 1. Rate limit
        limit_result = self.limiter.check(agent_did)
        if not limit_result.allowed:
            self.audit.log_action(agent_did, tool, outcome="denied",
                                  data={"reason": "rate_limited"})
            return False

        # 2. Check quarantine
        if self.monitor.is_quarantined(agent_did):
            self.audit.log_action(agent_did, tool, outcome="denied",
                                  data={"reason": "quarantined"})
            return False

        # 3. Check trust score against policy
        score = self.reward.get_score(agent_did)
        context = {
            "trust_score": score.total_score,
            "trust_tier": score.tier,
        }
        for rule in sorted(self.policy.rules, key=lambda r: r.priority):
            if rule.condition.evaluate(context):
                if rule.action == "deny":
                    self.audit.log_policy_decision(
                        agent_did, tool, decision="deny",
                        policy_name=rule.name)
                    return False
                break

        # 4. Execute (simulated here)
        success = True  # replace with real execution

        # 5. Record outcome
        self.monitor.record_tool_call(agent_did, tool, success=success)
        if success:
            self.reward.record_task_success(agent_did)
        else:
            self.reward.record_task_failure(agent_did, reason=f"{tool} failed")

        # 6. Update registry
        new_score = self.reward.get_score(agent_did).total_score
        await self.registry.update_trust_score(agent_did, new_score)

        # 7. Audit
        self.audit.log_action(agent_did, tool,
                              outcome="success" if success else "failure")
        return success