multi-agent.md

← Back to README

Multi-Agent Enforcement

A single Fence works for one agent. Production systems often have multiple agents — specialists that search, a synthesizer that writes, an auditor that verifies. fence.link(upstream) declares "this fence can cite evidence from that fence's history."

Hierarchical — manager cites workers

from audit_fence import Fence

worker_a = Fence(name="worker_a")
worker_b = Fence(name="worker_b")
manager = Fence(name="manager")

manager.link(worker_a, worker_b)

# Workers search independently
tools_a = worker_a.wrap_tools(worker_a_tools, search=["search_*"])
tools_b = worker_b.wrap_tools(worker_b_tools, search=["search_*"])

# Manager validates against its own history PLUS both workers'
manager_tools = manager.wrap_tools(
    [search_summary, write_report],
    search=["search_*"],
    submit=["write_*"],
)

Pipeline — transitive evidence flow

researcher = Fence(name="researcher")
enricher = Fence(name="enricher")
reporter = Fence(name="reporter")

enricher.link(researcher)    # enricher can cite researcher
reporter.link(enricher)      # reporter can cite enricher AND researcher (transitive)

Links are transitive. Each stage validates against the accumulated history of all preceding stages.

Production — multi-specialist audit (Firn)

A real-world topology from Firn, the financial analysis system audit-fence was extracted from:

from audit_fence import Fence, FenceGroup

group = FenceGroup()

# Specialist agents (each searches independently)
fund = group.create("fundamental")
tech = group.create("technical")
value = group.create("value")
macro = group.create("macro")

# R1 audit: each auditor is restricted to one specialist
r1_fund = group.create("r1_fundamental")
r1_fund.link(fund)     # can only cite fundamental's searches

r1_tech = group.create("r1_technical")
r1_tech.link(tech)     # can only cite technical's searches

# R2 audit: cross-specialist verification
r2 = group.create("r2_specialist")
r2.link(fund, tech, value, macro)   # can cite all four

# Unified view
group.save_log("audit/enforcement_log.jsonl")

Each R1 auditor is isolated to its specialist. The R2 auditor intentionally sees all four. The topology encodes the audit policy.

Path sandboxing

Different agents should be restricted to different data sources. SandboxedSearch wraps a search backend with path restrictions:

from audit_fence import SandboxedSearch

search_a = SandboxedSearch(
    backend=grep_backend,
    allowed_dirs=["trace/specialist_outputs/"],
)

search_b = SandboxedSearch(
    backend=grep_backend,
    allowed_dirs=["tools/"],
    allowed_files=["report.md"],
)

search_a("revenue", "trace/specialist_outputs/fund.md")   # allowed
search_a("revenue", "tools/data.json")                     # => ERROR: outside sandbox

Path traversal (e.g., tools/../secret/data.json) is automatically blocked.

RipgrepBackend

A ready-to-use search backend wrapping the rg CLI via subprocess:

from audit_fence import RipgrepBackend, SandboxedSearch, Fence

fence = Fence()

grep = RipgrepBackend(root="./trace/")
search = fence.wrap_tool(grep, role="search")

# With sandbox
sandboxed = SandboxedSearch(backend=grep, allowed_dirs=["tools/"])
search = fence.wrap_tool(sandboxed, role="search")

Requires rg (ripgrep) installed on the system — no Python dependencies added.

FenceGroup

Optional convenience for managing multiple fences:

group = FenceGroup()
fund = group.create("fundamental", min_evidence_length=20)
tech = group.create("technical", min_evidence_length=20)

group["fundamental"].rejections       # named access
group.all_rejections                  # sorted across all fences
group.all_claims                      # all ClaimRecords
group.save_log("audit.jsonl")         # combined log (write mode, aggregates all fences)
group.reset()                         # reset all

chain = group.trace_chain(some_claim) # evidence chain traversal

FenceGroup.from_snapshot_manifest()

Build an entire audit topology automatically from a Snapshot manifest — no manual fence creation or linking needed:

from audit_fence import FenceGroup, Snapshot

snap = Snapshot("trace/")
# ... production pipeline runs ...
snap.finalize()

manifest = snap.load_manifest()
group = FenceGroup.from_snapshot_manifest(
    manifest,
    document=open("report.md").read(),
    trace_dir="trace/",
    per_agent_documents={                    # optional overrides
        "writer": open("final_report.md").read(),
    },
)

This creates one Fence per agent in the manifest, with:

• source restricted to that agent's trace subdirectory
• document resolved from: per_agent_documents override > first artifact content > fallback document
• link topology matching the manifest's declared dependencies

link() naming requirement

link() requires upstream fences to have a name (needed for serialization):

unnamed = Fence()          # no name
named = Fence(name="data") # has name

manager = Fence(name="manager")
manager.link(named)        # OK
manager.link(unnamed)      # raises ValueError

Always create fences with Fence(name=...) or use group.create("name") when linking is involved.

Cascade audit

audit_cascade() verifies that data flowed through the entire DAG, not just that evidence exists somewhere. For each edge in the dependency graph, it runs an independent audit agent that checks whether the downstream node's output is supported by the upstream node's trace:

from audit_fence import FenceGroup

manifest = snap.load_manifest()
cascade = await FenceGroup.audit_cascade(
    manifest,
    llm=ChatOpenAI(model="gpt-4o"),
    trace_dir="trace/",
    final_document=open("report.md").read(),
)

if cascade.all_passed:
    print("Full evidence chain verified")
else:
    for src, dst in cascade.failed_edges:
        print(f"Evidence gap: {src} -> {dst}")

Tracing a claim back to its source:

After the cascade completes, claims are automatically linked across edges. Use trace_claim() to walk any claim back to the original data source — zero extra LLM calls:

# Pick a claim from the final report edge
final_edge = cascade.edge_results[-1]
claim = final_edge.result.claims[0]

chain = cascade.trace_claim(claim)
# [report_claim, analysis_claim, research_claim]
for c in chain:
    print(f"  {c.claim[:60]}  ← evidence: {c.evidence[:60]}")

Cross-edge linking works via deterministic substring matching: because LangGraph state passes through verbatim between nodes, a downstream claim's evidence (grep'd from the upstream trace) will appear as a substring of the upstream claim's text. No semantic similarity needed.

How it works:

1. For each edge A→B in the manifest dependencies, a Fence is created where:
   • document = B's output trace (or final_document for terminal nodes)
   • source = A's output only (not the full trace directory) — this isolates what B actually received from what A internally consumed
2. When a downstream node has multiple upstreams (fan-in, e.g. A→C, B→C), a single merged Fence is created instead of N separate ones. The merged source directory contains one file per upstream ({upstream}_output.txt), and trace_claim() uses search_file to determine which upstream each claim came from
3. All edge audits run in parallel via asyncio.gather
4. The cascade passes only if every edge audit finds evidence for all claims
5. After all edge audits complete, claims are linked across adjacent edges via substring matching, enabling trace_claim() to walk the full provenance chain

On-disk artifacts:

• .audit_cache/{node}/output_summary.txt — permanent output-only extracts generated by Snapshot.finalize(). Used as the source for cascade audits (avoids re-extracting on each run)
• .audit_sources/{downstream}/ — permanent directories containing symlinks to upstream output files. Rebuilt on each audit_cascade() call. These ensure that search_file references in claim records remain valid after the audit completes

Distinction from from_snapshot_manifest:

• from_snapshot_manifest creates per-agent fences (each agent audited against its own traces)
• audit_cascade creates per-edge fences (each edge verifies data provenance between two agents)

Terminal nodes (those with edges to __end__ in the graph topology) use final_document as their document, connecting the last pipeline stage to the actual user-facing report.

Cross-subgraph tracing

When your LangGraph graph contains compiled subgraphs (nested graphs as nodes), snap.config() automatically captures checkpoint_ns metadata to infer hierarchical topology. During finalize(), subgraph nodes are grouped by scope, edges are inferred within each scope, and cross-scope edges connect subgraph terminal nodes back to their parent nodes.

trace_claim() works through these subgraph boundaries — if a claim in the final report traces back through a parent node that delegates to a subgraph, the chain follows through the subgraph's internal edges all the way to the original data source:

cascade = await FenceGroup.audit_cascade(
    manifest, llm=llm, trace_dir="trace/",
    final_document=report,
)

# Walk through subgraph boundaries
chain = cascade.trace_claim(final_claim)
# [report_claim, supervisor_claim, subgraph_compress_claim, subgraph_search_claim]

Limitation: When two subgraph instances run concurrently via asyncio.gather and contain nodes with the same name, LangGraph's checkpoint_ns differentiates them by UUID, but the agent name prefix is identical. If both subgraph instances produce a node named worker, they both map to supervisor_worker — trace data interleaves. Use unique node names within subgraphs that may run in parallel, or use prefix in snap.config() to disambiguate.

Thread safety

Fence instances are thread-safe. In wrap_tools() scenarios where multiple tools execute concurrently, the internal _history, _rejections, and _claims lists are protected by threading.Lock. The BFS traversal in _collect_history acquires per-fence locks without holding multiple simultaneously (no deadlock risk).