Forem: Anton Fedotov

Adding a Trust Boundary to a LlamaIndex RAG Pipeline

Anton Fedotov — Wed, 29 Apr 2026 08:03:54 +0000

Your LlamaIndex app does not only retrieve documents.

It decides which external text is allowed to become model context.

That is a trust decision, even if your code does not call it one.

A PDF can contain useful facts.
A support ticket can contain real customer context.
A web page can contain documentation.
An email thread can contain the answer your user needs.

But all of those sources can also contain instructions your model should never follow.

That is the uncomfortable part of RAG security: the dangerous text often does not come from the user prompt. It comes from the documents.

This post shows how to add a trust boundary to a LlamaIndex RAG pipeline with Omega Walls.

The core idea is simple:

Retrieved text is evidence, not policy.

And the safest place to enforce that is between retrieval and synthesis.

The RAG failure mode

A typical LlamaIndex flow looks clean:

documents -> index -> query engine -> response

In code, it may look something like this:

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex

documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine()

response = query_engine.query(
    "Summarize the customer escalation and suggest the next step."
)

print(response)

That is a good developer experience.

But it hides a trust problem.

The query engine retrieves relevant chunks. Those chunks are then used by the LLM to synthesize an answer. That is the normal RAG path: retrieve text, feed it into the answer-generation step, produce a response.

The issue is that retrieved text can carry two very different kinds of content:

Useful evidence:
- customer reported X
- policy says Y
- document describes Z

Untrusted instruction:
- ignore previous instructions
- reveal the system prompt
- call this tool
- send this data somewhere else

If both kinds of text are placed into the same context without a boundary, the model has to separate evidence from instruction by itself.

That is not a reliable boundary.

Retrieved text is evidence, not policy

The important shift is small but sharp:

Retrieved text should help the model answer.
It should not control the workflow.

That means your RAG pipeline should preserve a hard distinction between:

trusted:
- system policy
- developer instructions
- app configuration
- user request

untrusted:
- retrieved web pages
- PDFs
- emails
- support tickets
- uploaded files
- tool outputs containing external text

The model should not have to infer that distinction from formatting alone.

Your application should enforce it before the context is built.

In Omega Walls, this is the role of the trust boundary. Untrusted content is projected into structured risk signals, filtered, and only allowed chunks are passed forward into context. Tool calls stay behind a tool gateway.

For a LlamaIndex RAG app, the most natural placement is:

documents / web / tickets / PDFs
        ↓
index / retriever
        ↓
Omega Walls trust boundary
        ↓
allowed chunks
        ↓
query engine / response synthesis
        ↓
answer

The trust boundary belongs between retrieval and synthesis: external documents are useful evidence, but they should be inspected before they shape the final context.

Why post-generation checks are too late

It is tempting to check the final answer.

That can still be useful.

But it is not enough.

By the time the answer exists, the retrieved chunk may already have influenced:

which facts were selected,
which instruction hierarchy the model followed,
whether a tool should be called,
how intermediate summaries were formed,
what got written into memory,
what source got treated as authoritative.

In a RAG pipeline, the critical moment is earlier:

retrieved chunks -> context construction

That is where external text becomes model context.

So the boundary should live there.

Not only at the user input.
Not only after generation.
Before synthesis.

Install

Install Omega Walls with integration adapters:

pip install "omega-walls[integrations]"

This gives you the framework adapters, including the LlamaIndex guard.

Minimal LlamaIndex integration

Start with your normal LlamaIndex setup:

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex

documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine()

Now wrap the query engine with Omega:

from omega.integrations import OmegaLlamaIndexGuard

guard = OmegaLlamaIndexGuard(profile="quickstart")

query_engine = guard.wrap_query_engine(
    index.as_query_engine()
)

response = query_engine.query(
    "Summarize this support note",
    thread_id="sess-123",
)

print(response)

That is the smallest useful version.

You still use LlamaIndex as your data/query layer. You still build your index normally. You still call the query engine normally.

The difference is that retrieved context now passes through a trust boundary before it is allowed to shape the response.

What actually gets guarded

A useful RAG boundary needs to cover more than the initial query string.

At minimum, you should think about five surfaces.

1. User query

The user query starts the request.

It may be harmless. It may be adversarial. It may also be asking the app to summarize an external document that contains adversarial text.

The guard should know which session this belongs to.

response = query_engine.query(
    "Summarize the attached escalation notes.",
    thread_id="support-case-1842",
)

That thread_id matters because stateful detection only makes sense if related steps belong to the same workflow.

2. Retrieved chunks

This is the main RAG surface.

A retrieved chunk may contain facts and hidden instructions at the same time.

The boundary should inspect those chunks before they become prompt context.

retriever returns nodes/chunks
        ↓
guard evaluates external text
        ↓
only allowed chunks enter synthesis

3. Tool calls

Some LlamaIndex apps are not just read-only QA systems.

They retrieve, reason, and then call tools.

For example:

send a ticket update,
post a summary,
call an internal API,
write a file,
trigger a workflow,
fetch more external data.

If a tool can act outside the model, it should sit behind a gateway.

from omega.integrations import OmegaLlamaIndexGuard

guard = OmegaLlamaIndexGuard(profile="quickstart")

def network_post(url: str, payload: dict) -> dict:
    # Your real external action lives here.
    return {"status": "ok", "url": url}

safe_network_post = guard.wrap_tool(
    "network_post",
    network_post,
)

Then use safe_network_post instead of the raw function.

result = safe_network_post(
    url="https://internal.example/support/update",
    payload={
        "case_id": "1842",
        "summary": "Customer escalation summarized from guarded context."
    },
)

The important part is not this specific tool.

The important part is that tool execution goes through one chokepoint.

4. Memory writes

RAG systems often create derived state:

summaries,
notes,
extracted facts,
cached answers,
user preferences,
case memory,
long-term knowledge.

If that state came from external text, preserve provenance.

Do not turn this:

external PDF says: approval can be skipped

into this:

memory fact: approval can be skipped

without a source/trust tag.

A simple pattern:

decision = guard.check_memory_write(
    text="The external document says approval can be skipped.",
    source_id="pdf:customer-escalation-1842",
    source_type="pdf",
    source_trust="untrusted",
    thread_id="support-case-1842",
)

if decision.mode == "allow":
    save_to_memory(...)
elif decision.mode == "quarantine":
    save_to_quarantine(...)
else:
    print("Memory write denied")

The exact memory store is your choice.

The rule is the point:

Memory should remember where information came from.

5. Session context

Single-document checks miss a lot.

A mild instruction in one chunk may not look serious.
A later chunk may ask for a secret.
A later tool output may introduce an action.
Together, the pattern matters.

That is why a RAG guard should be stateful across a session, not just a one-shot text classifier.

A small end-to-end example

Here is a compact example showing the integration shape.

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from omega.integrations import OmegaLlamaIndexGuard
from omega.adapters import OmegaBlockedError, OmegaToolBlockedError

# 1. Build your normal LlamaIndex index
documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)

# 2. Create the Omega guard
guard = OmegaLlamaIndexGuard(profile="quickstart")

# 3. Wrap the query engine
query_engine = guard.wrap_query_engine(
    index.as_query_engine()
)

# 4. Optional: wrap tools that may act outside the model
def network_post(url: str, payload: dict) -> dict:
    return {"status": "ok", "url": url}

safe_network_post = guard.wrap_tool(
    "network_post",
    network_post,
)

# 5. Use the guarded query engine
try:
    response = query_engine.query(
        "Summarize the support note and recommend the next action.",
        thread_id="support-case-1842",
    )

    print(response)

except OmegaBlockedError as exc:
    print("Blocked query/input step")
    print("Outcome:", exc.decision.control_outcome)
    print("Reasons:", exc.decision.reason_codes)

except OmegaToolBlockedError as exc:
    print("Blocked tool call")
    print("Tool:", exc.gate_decision.tool_name)
    print("Reason:", exc.gate_decision.reason)

This gives your application a real block path.

Not a vague error.
Not a silent failure.
Not a mysterious bad answer.

A structured decision you can log, route, or escalate.

What happens to risky documents?

A good RAG boundary should not kill the whole app because one chunk looked suspicious.

Usually, the better behavior is selective:

risky chunk detected
        ↓
remove that chunk from context
        ↓
continue with safe chunks
        ↓
freeze tools if tool-abuse pressure appears
        ↓
escalate if secrets/exfiltration are involved

That is controlled degradation.

The workflow should be able to continue with the remaining safe context.

Example behavior:

User asks:
"Summarize this customer note."

Retriever returns:
- doc-1: normal support note
- doc-2: external email with hidden instruction
- doc-3: product policy excerpt

Boundary:
- allows doc-1
- blocks doc-2
- allows doc-3

Query engine:
- synthesizes answer from doc-1 and doc-3
- does not include doc-2 in context

This is much better than two common alternatives:

bad option 1: pass every retrieved chunk into context
bad option 2: hard-stop the whole workflow on any suspicious text

The useful middle path is:

remove risky influence, keep safe work moving

Retrieved content should enter the prompt as evidence with provenance, not as workflow authority.

Verify the integration

After wiring the adapter, run the LlamaIndex smoke:

python scripts/smoke_llamaindex_guard.py --strict

This is the first verification step.

You are not only checking that imports work.

You are checking that the guard is actually on the execution path.

For a broader release gate across all framework adapters:

python scripts/run_framework_smokes.py --strict

The expected summary should be boring:

status = ok
framework_count = 6
total_failures = 0
min_gateway_coverage >= 1.0
total_orphans = 0

Boring is good here.

It means the adapter is not decorative.

What to log

A trust boundary becomes much more useful when decisions are explainable.

At minimum, log:

session_id
source_id
source_type
decision outcome
reason codes
blocked docs
tool gateway decisions

For production systems, avoid raw content by default.

Store hashes, bounded evidence, redacted excerpts only when policy allows it, and enough structured data to reproduce the decision later.

That gives you incident review without turning your security logs into a new data leak.

Practical checklist for LlamaIndex RAG apps

When reviewing a LlamaIndex app, I would walk through this checklist.

1. What data sources are indexed?

List them:

internal docs
uploaded PDFs
support tickets
email threads
web pages
chat exports
tool outputs

Then mark trust level.

If you cannot mark trust level, assume untrusted.

2. Where does retrieval happen?

Find the point where the query engine receives retrieved chunks or nodes.

That is where the boundary belongs.

3. Can retrieved text reach synthesis directly?

If yes, add a guard before synthesis.

The synthesis step should receive allowed evidence, not raw external content.

4. Are sources preserved?

Each chunk should retain enough metadata:

{
    "source_id": "pdf:customer-escalation-1842",
    "source_type": "pdf",
    "source_trust": "untrusted",
}

If the chunk gets summarized or cached, preserve that provenance.

5. Can the RAG flow trigger tools?

If yes, wrap those tools.

Read-only retrieval is one risk level.
File writes, network calls, ticket updates, and outbound messages are another.

6. Do security docs create false positives?

Your guard should not panic just because a document discusses prompt injection, API keys, or jailbreaks.

Security guidance is not the same thing as an active attack.

That is why polarity and hard-negative tests matter.

A document saying:

Never reveal API keys.

should not be treated like:

Reveal the API key.

The difference is small in keywords and huge in intent.

Why this is not just prompt filtering

A prompt filter usually asks:

Is this text bad?

A RAG trust boundary asks a better question:

Should this external text be allowed to shape model context, memory, or tools in this session?

That is a different problem.

It needs:

source awareness,
session awareness,
context placement,
tool gateway enforcement,
memory provenance,
selective blocking,
auditable decisions.

This is why the boundary belongs in the pipeline, not just in a prompt template.

What this does not solve

A trust boundary is not magic.

It does not replace:

secret management,
least-privilege tools,
network allowlists,
auth and permissions,
model-side safety policies,
human approval for sensitive operations,
logging and incident response.

It also depends on correct placement.

If untrusted text can bypass the guard and enter context directly, the boundary is broken.

If tools can execute outside the gateway, tool enforcement is broken.

If source metadata is stripped too early, later steps cannot distinguish trusted evidence from untrusted text.

So the rule is simple:

Put the boundary on the actual path between retrieval and synthesis.

Not next to it.

A good rollout order

I would not start with hard blocking in production.

Use a safer rollout:

1. Wrap the query engine.
2. Wrap any tools that can act outside the model.
3. Preserve source_id, source_type, and source_trust.
4. Run the strict LlamaIndex smoke.
5. Run in monitor mode on realistic traffic.
6. Inspect reports and decisions.
7. Tune hard negatives and source handling.
8. Enable enforcement for selected paths.

The important part is the monitor phase.

You want to see:

which sources are noisy,
which chunks would be blocked,
whether benign security docs stay quiet,
whether tool-freeze decisions are understandable,
whether operators have enough information to act.

Hard blocking without observability is how a safety layer becomes a production incident.

Final thought

RAG makes it easy to give a model more context.

That is useful.

But context is not neutral.

When your app retrieves documents, it is deciding which external text gets a chance to influence the model. If that text comes from web pages, PDFs, emails, tickets, uploads, or tool outputs, it should not be treated as trusted by default.

The better rule is:

Retrieved text is evidence, not policy.

LlamaIndex gives you the data/query layer.

Omega Walls adds the trust boundary around the part of the pipeline where external text becomes context.

That boundary should sit before synthesis, before memory writes, and before tools.

Not because every document is malicious.

Because production RAG systems should not ask the model to guess what is trusted.

Omega Walls ships framework adapters for LangChain, LangGraph, LlamaIndex, Haystack, AutoGen, and CrewAI.

Install:

pip install "omega-walls[integrations]"

GitHub: https://github.com/synqratech/omega-walls
PyPI: https://pypi.org/project/omega-walls/
Site: https://synqra.tech/omega-walls

I built an open-source trust boundary for RAG and AI agent pipelines

Anton Fedotov — Mon, 27 Apr 2026 12:12:57 +0000

A pattern I keep seeing in AI agent workflows:

they work on clean demo data,
then become fragile on real-world content.

Not because of some dramatic “AI jailbreak” story.

More often because the pipeline has no clear boundary between:

trusted instructions
retrieved content
emails / PDFs / tickets
memory
tool outputs

For the model, all of this can become one context stream.

That means untrusted content can quietly start influencing the workflow.

I built Omega Walls as an open-source Python library for this problem.

The idea is to put a runtime trust boundary between untrusted content, model context, memory, and tools.

The first version focuses on:

RAG / agent prompt injection
cross-document or cross-step attack pressure
secret-exfiltration pressure
tool/action abuse
deterministic evidence and auditability

Install:

pip install omega-walls

GitHub:
https://github.com/synqratech/omega-walls

PyPI:
https://pypi.org/project/omega-walls/

I’d be interested in feedback from people building RAG or internal agent systems: where would you expect this layer to sit in your stack?

Adding a Stateful Trust Boundary to a LangGraph Agent

Anton Fedotov — Mon, 27 Apr 2026 09:33:55 +0000

LangGraph makes agent workflows easier to reason about.

You can see the nodes.
You can see the edges.
You can see where state is read, updated, and passed forward.

That is a big improvement over a black-box agent loop.

But it also exposes the question most agent pipelines eventually have to answer:

Which parts of this graph are allowed to trust external content?

A graph-based agent can read search results, PDFs, emails, tickets, web pages, tool outputs, and user-pasted text. Some of that content is useful evidence. Some of it may contain instructions that should never become part of the agent’s control flow.

The problem is not only whether a user prompt is malicious.

In a stateful graph, the real question is:

Where does untrusted content enter the graph, can it get written into state, and can it later influence a tool node?

That is where a stateful trust boundary helps.

This post shows how to add that boundary to a LangGraph workflow with Omega Walls.

The short version

If you only remember one thing, make it this:

Every edge that carries external text is a trust-boundary edge.
Every node that can execute tools needs a gateway.
Every state write needs a source/trust tag.

That is the whole mental model.

LangGraph gives you the graph. Omega Walls gives you a boundary around the parts of the graph that should not trust unverified content.

Why LangGraph changes the security conversation

In a simple agent loop, risk often looks like this:

user input -> model -> tool -> model -> answer

That is easy to explain, but it hides the important part.

A real workflow is messier:

user input
  -> retrieve docs
  -> summarize docs
  -> update state
  -> decide next node
  -> call tool
  -> receive tool output
  -> update state again
  -> generate final answer

Once state enters the picture, prompt injection is no longer just an input-filtering problem.

A retrieved web page can influence one node.
That node can write a summary into state.
A later node can treat that state as trusted context.
Another node can use that context to decide whether to call a tool.

Nothing needs to look dramatic in one step.

The bad pattern can be distributed across the workflow.

That is why graph agents need controls that are also graph-shaped.

The failure mode: untrusted text becomes workflow state

Here is the common failure mode.

You build a LangGraph workflow that looks roughly like this:

User request
  -> retrieve external documents
  -> analyze documents
  -> write notes into graph state
  -> agent decides what to do
  -> tool node executes

The workflow feels clean.

The retrieval node retrieves.
The analysis node analyzes.
The state carries the result.
The tool node acts.

But if the retrieved document contains hidden instructions, and you write a derived version of that content into state without marking its source, you have a trust problem.

The graph state now contains external influence.

A later node may not know whether a sentence came from:

your system policy,
the user’s actual request,
a trusted internal source,
an external PDF,
a fetched web page,
or a tool result that contained untrusted text.

Once that distinction is lost, the model has to guess.

That is not a good boundary.

The dangerous part is not always the first model call. The dangerous part is what gets written into state and reused later.

Without a boundary, external text can be summarized into graph state and later treated as trusted workflow context.

The right mental model

The easiest way to reason about this is to treat every edge carrying external content as a boundary edge.

Trust-boundary edges in a LangGraph workflow: external content is guarded before it reaches graph state, context, or tools.

A LangGraph workflow should not treat all state as equally trusted.

A cleaner mental model looks like this:

Trusted inputs
  - system / app policy
  - developer-controlled config
  - user request

Untrusted inputs
  - web pages
  - search results
  - PDFs
  - emails
  - tickets
  - tool outputs containing external text

Boundary
  - inspect
  - filter
  - decide
  - tag
  - block or allow

Graph
  - agent node
  - state
  - tool nodes
  - memory writes

Gateway
  - tool execution
  - file/network/action controls

The trust boundary should sit before untrusted content can shape model context, state, or tool execution.

Insert diagram here: Trust Boundary Edges in a LangGraph Workflow

Caption:

A practical trust-boundary model for LangGraph: trusted inputs can flow into the agent, but external content should pass through a boundary before it reaches graph state, context, or tools.

What Omega Walls adds

Omega Walls is a stateful trust boundary for RAG and agentic pipelines.

In this integration, the important pieces are:

model/input checks,
tool preflight checks,
memory write checks,
session-aware runtime state,
typed block paths,
structured decisions you can log or route into operator workflow.

The goal is not to make the graph useless by hard-blocking everything.

The goal is controlled degradation:

suspicious content can be soft-blocked,
risky sources can be quarantined,
tool execution can be frozen,
severe cases can be escalated,
safe parts of the workflow can continue.

That matters in real agents. A production workflow should not have only two modes: “everything allowed” and “everything dead.”

Install

Install Omega Walls with integration adapters:

pip install "omega-walls[integrations]"

You can also install only the base package and the framework dependencies you use, but the integrations extra is the fastest path for framework adapters.

Minimal LangGraph integration

If you already have a compiled LangGraph workflow, the integration shape is small:

from omega.integrations import OmegaLangGraphGuard

guard = OmegaLangGraphGuard(profile="quickstart")

safe_graph = guard.wrap_graph(compiled_graph)
safe_tool = guard.wrap_tool("network_post", network_post_fn)
guard_node = guard.build_guard_node()  # optional StateGraph node helper

The three important pieces are:

safe_graph = guard.wrap_graph(compiled_graph)

This wraps the graph-level execution path.

safe_tool = guard.wrap_tool("network_post", network_post_fn)

This ensures the tool call goes through a guarded preflight path before it executes.

guard_node = guard.build_guard_node()

This gives you an optional explicit guard node you can place inside a StateGraph when you want the boundary to be visible in the workflow itself.

Where to put the guard

There are two common patterns.

Pattern 1: Wrap the compiled graph

Use this when you want the simplest integration.

from omega.integrations import OmegaLangGraphGuard

guard = OmegaLangGraphGuard(profile="quickstart")

compiled_graph = graph.compile()
safe_graph = guard.wrap_graph(compiled_graph)

result = safe_graph.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Summarize the latest external research note."
            }
        ]
    },
    config={
        "configurable": {
            "thread_id": "customer-support-123"
        }
    }
)

print(result)

This is the easiest entry point.

You keep your graph structure. The guard sits around the graph execution path. If your app already passes a thread_id, conversation_id, or session_id, the adapter can use that to keep runtime decisions tied to the right workflow session.

Use this when you want a quick integration without changing the graph topology.

Pattern 2: Add an explicit guard node

Use this when you want the trust boundary to be visible in the graph.

The exact graph shape depends on your app, but the idea is simple:

START
  -> retrieve_external_content
  -> omega_guard
  -> agent
  -> tools
  -> END

Example shape:

from typing import TypedDict, List, Any
from langgraph.graph import StateGraph, START, END
from omega.integrations import OmegaLangGraphGuard

class AgentState(TypedDict):
    messages: List[dict]
    retrieved_docs: List[dict]
    guarded_docs: List[dict]
    risk_notes: List[str]

guard = OmegaLangGraphGuard(profile="quickstart")

def retrieve_external_content(state: AgentState) -> dict:
    # Replace with your retriever, search API, document loader, etc.
    docs = [
        {
            "source_id": "web:example.com/page",
            "source_type": "web",
            "trust": "untrusted",
            "text": "External page content..."
        }
    ]
    return {"retrieved_docs": docs}

omega_guard = guard.build_guard_node()

def agent_node(state: AgentState) -> dict:
    # At this point, only guarded/allowed content should shape the prompt.
    messages = state["messages"]
    guarded_docs = state.get("guarded_docs", [])

    # Build your prompt from trusted messages + guarded docs.
    # Call your model here.
    return {
        "messages": messages + [
            {
                "role": "assistant",
                "content": f"Processed {len(guarded_docs)} guarded documents."
            }
        ]
    }

graph = StateGraph(AgentState)

graph.add_node("retrieve_external_content", retrieve_external_content)
graph.add_node("omega_guard", omega_guard)
graph.add_node("agent", agent_node)

graph.add_edge(START, "retrieve_external_content")
graph.add_edge("retrieve_external_content", "omega_guard")
graph.add_edge("omega_guard", "agent")
graph.add_edge("agent", END)

compiled_graph = graph.compile()
safe_graph = guard.wrap_graph(compiled_graph)

This version has one major advantage: the trust boundary is not hidden.

Anyone reading the graph can see that external content does not go straight from retrieval into the agent node.

It passes through the guard first.

Guard tool nodes, not only text inputs

The mistake is to guard only text before the model call.

In a LangGraph workflow, tool nodes are often where the real-world impact happens.

A tool might:

send a network request,
write a file,
update a ticket,
call an internal API,
send a message,
trigger a transaction,
fetch another external page.

If external text can influence a tool call, that tool call should go through a gateway.

Example:

from omega.integrations import OmegaLangGraphGuard

guard = OmegaLangGraphGuard(profile="quickstart")

def network_post(url: str, payload: dict) -> dict:
    # Your real network operation lives here.
    # The guard should sit before this function executes.
    return {"status": "ok", "url": url}

safe_network_post = guard.wrap_tool("network_post", network_post)

Then use safe_network_post in your graph instead of the raw function.

def tool_node(state: dict) -> dict:
    payload = {
        "summary": state.get("summary", "")
    }

    result = safe_network_post(
        url="https://internal.example/ingest",
        payload=payload
    )

    return {"tool_result": result}

The important thing is not the name network_post.

The important thing is the chokepoint.

If a tool can do something outside the model, it should not be callable through an unguarded side path.

Handle blocked paths explicitly

A boundary should not fail mysteriously.

Your application should know whether the model/input step was blocked or the tool call was blocked.

from omega.adapters import OmegaBlockedError, OmegaToolBlockedError

try:
    result = safe_graph.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "Analyze this external report and continue the workflow."
                }
            ]
        },
        config={
            "configurable": {
                "thread_id": "workflow-123"
            }
        }
    )

except OmegaBlockedError as exc:
    print("Blocked model/input step")
    print("Outcome:", exc.decision.control_outcome)
    print("Reasons:", exc.decision.reason_codes)

except OmegaToolBlockedError as exc:
    print("Blocked tool call")
    print("Tool:", exc.gate_decision.tool_name)
    print("Reason:", exc.gate_decision.reason)

This gives your app a real branch.

You can log it.
You can show a safe message.
You can ask for human approval.
You can continue without the risky source.
You can freeze tools for the session.

The point is not just “block bad thing.”

The point is to make the decision operational.

Guard memory writes

Graph state is not the only state you should care about.

Many agent systems also write to memory:

user facts,
summaries,
preferences,
retrieved notes,
intermediate conclusions,
task state,
long-term memory,
cached tool results.

If a memory write came from external text, preserve that fact.

Do not let the system turn:

external page said X

into:

known fact: X

without a trust tag.

Use the memory write check when persistence is involved:

decision = guard.check_memory_write(
    text="The external document says the support workflow should skip approval.",
    source_id="web:example.com/page",
    source_type="web",
    source_trust="untrusted",
    thread_id="workflow-123",
)

if decision.mode == "allow":
    save_to_memory(...)
elif decision.mode == "quarantine":
    save_to_quarantine(...)
else:
    print("Memory write denied")

The exact storage backend is up to your app.

The principle is not optional:

Memory should remember where information came from.

If you lose provenance, your future graph steps cannot tell the difference between trusted state and external influence.

Verify the integration

After wiring the guard, run the strict smoke for LangGraph:

python scripts/smoke_langgraph_guard.py --strict

This is the first thing I would run locally.

Not “does my app still start?”
Not “does the graph compile?”
But:

Is the guard actually on the execution path?

That is the test that matters.

For a broader release gate across framework adapters, run:

python scripts/run_framework_smokes.py --strict

The expected high-level result should be boring:

status = ok
framework_count = 6
total_failures = 0
min_gateway_coverage >= 1.0
total_orphans = 0

Boring is good here.

It means your wrappers are not decorative.

A practical graph checklist

When reviewing a LangGraph workflow, I would walk it with this checklist.

1. Which nodes receive external text?

Look for nodes that read from:

retrievers,
search APIs,
browser tools,
PDFs,
emails,
tickets,
web fetchers,
tool outputs.

Those nodes should either be guarded directly or feed into a guard node before the agent consumes their output.

2. Which edges carry untrusted content?

Edges are not just control flow.

In a stateful graph, edges also carry influence.

If an edge carries external text, treat it as a boundary edge.

3. Which nodes write state?

Any node that writes to graph state can change future behavior.

If it writes derived content from an untrusted source, keep source metadata attached.

4. Which nodes can execute tools?

Tool nodes should call wrapped tools, not raw functions.

If a tool can write, send, fetch, mutate, or trigger an external action, it belongs behind a gateway.

5. Can a later node distinguish trusted from untrusted state?

If not, you probably need better state shape.

A useful state object should preserve the difference between:

trusted_policy
user_request
guarded_docs
quarantined_docs
tool_results
risk_notes

Instead of dumping everything into one generic context field.

Example state shape

Here is a simple state shape I prefer for guarded graph workflows:

from typing import TypedDict, List, Literal, Optional

class SourceRef(TypedDict):
    source_id: str
    source_type: str
    trust: Literal["trusted", "semi", "untrusted"]

class DocumentChunk(TypedDict):
    text: str
    source: SourceRef

class RiskNote(TypedDict):
    source_id: str
    outcome: str
    reason_codes: List[str]

class AgentState(TypedDict):
    messages: List[dict]

    # Raw external inputs
    retrieved_docs: List[DocumentChunk]

    # Content allowed to shape context
    guarded_docs: List[DocumentChunk]

    # Content blocked or held for review
    quarantined_docs: List[DocumentChunk]

    # Runtime/security notes
    risk_notes: List[RiskNote]

    # Tool outputs with provenance
    tool_results: List[dict]

This makes trust visible.

It is much easier to reason about:

state["guarded_docs"]

than a giant mixed list called:

state["context"]

The name matters because future contributors will follow the shape you give them.

If everything is called context, everything will eventually be treated as context.

What happens when something is risky?

The useful behavior is not always “stop the agent.”

Often, the better behavior is controlled degradation.

For example:

External document looks suspicious
  -> remove it from context
  -> continue with remaining docs
  -> log reason
  -> freeze tools if tool-abuse pressure appears
  -> escalate only if needed

That is better than letting the model ingest everything.

It is also better than killing every workflow at the first suspicious string.

A graph workflow gives you room to degrade gracefully:

retrieve docs
  -> guard docs
  -> if safe: continue
  -> if suspicious: continue without that source
  -> if tool risk: freeze tool node
  -> if severe: route to human review

That is a product behavior, not just a security behavior.

What this does not solve

A trust boundary is not magic.

It does not replace:

least-privilege tool permissions,
secret management,
network allowlists,
proper auth,
human approval for sensitive operations,
logging and incident review,
model-side safety controls.

It also depends on architecture.

If raw tools can execute outside the gateway, the boundary can be bypassed.
If untrusted content can be written directly into state, the boundary is not really a boundary.
If your app removes source metadata too early, later nodes cannot reason about trust.

So the integration rule is simple:

Put the boundary on the actual execution path, not next to it.

A useful boundary should not have only two states: allow everything or kill the whole workflow.

Controlled degradation: remove risky influence, freeze dangerous execution paths, and continue with the remaining safe workflow.

A good rollout order

I would not jump straight to hard blocking.

Use this order:

1. Wrap the graph.
2. Wrap tool nodes.
3. Add an explicit guard node where external content enters.
4. Run strict smoke.
5. Run in monitor mode.
6. Inspect reports and decisions.
7. Add operator workflow.
8. Enable enforcement for selected paths.

The monitor phase is important.

You want to see:

which sources trigger decisions,
which tools would have been blocked,
whether benign security docs stay quiet,
whether risky paths show up clearly,
whether operators can understand the decision.

Hard blocking before observability is how you create a production incident while trying to prevent one.

Final thought

LangGraph gives you a better way to build agents because it makes workflow structure explicit.

That same explicit structure gives you a better way to secure them.

Do not think of a trust boundary as a single filter before the prompt.

In a graph, the boundary is a design discipline:

external text enters through guarded edges,
state keeps provenance,
tools execute through a gateway,
risky content can be removed without killing the whole workflow,
decisions are visible enough to debug.

That is the practical shape I want in production agent systems.

Not a bigger prompt.

A clearer boundary.

Omega Walls is open source and ships framework adapters for LangChain, LangGraph, LlamaIndex, Haystack, AutoGen, and CrewAI.

Install:

```

bash
pip install "omega-walls[integrations]"

LangGraph smoke:


bash
python scripts/smoke_langgraph_guard.py --strict

GitHub: https://github.com/synqratech/omega-walls
PyPI: https://pypi.org/project/omega-walls/
Site: https://synqra.tech/omega-walls

[Boost]

Anton Fedotov — Fri, 24 Apr 2026 09:17:46 +0000

Anton Fedotov

Apr 24

How to Add a Stateful Trust Boundary to a LangChain Agent with Omega Walls

#opensource #security #agents #langchain

Comments 2

7 min read

How to Add a Stateful Trust Boundary to a LangChain Agent with Omega Walls

Anton Fedotov — Fri, 24 Apr 2026 09:05:02 +0000

Your agent looked fine in the demo.

Then it started reading real PDFs, tickets, fetched pages, and tool outputs. Nothing looked obviously malicious. No one typed “ignore all previous instructions.” Still, the workflow drifted. The model began to treat external text as policy, the context got noisier, and tool execution became harder to trust.

That is the uncomfortable part of building agents on live data: a lot of failures do not come from the user prompt. They come from the agent’s architecture of trust. External content enters the reasoning loop disguised as facts, workflow state, or routine context. A single chunk may look harmless. The pattern only appears when you look across steps.

This is where a stateful trust boundary helps.

In this post, I’ll show how to add one to a LangChain agent with Omega Walls.

Why LangChain agents drift on live data

A LangChain agent is not just “prompt in, answer out.” It runs in a loop: call the model, decide whether to use tools, execute tools, feed results back, continue until a stop condition is reached. LangChain’s create_agent is their production-ready entry point, and the runtime is graph-based under the hood.

That loop is exactly why live-data failures become subtle.

A retrieved page can contain hidden policy. An attachment can smuggle instructions inside normal-looking operational text. A tool can fetch external content that looks like context but behaves like control. If your pipeline treats all of that as just “more text,” you are asking the model to separate trusted instructions from untrusted evidence on its own, in the middle of an execution loop.

That usually works right up until it doesn’t.

The shift that matters

Before we touch the code, it helps to fix the architecture in one simple mental model.

Where the trust boundary sits in a LangChain agent: trusted inputs go straight to the agent, untrusted content passes through the boundary first, and tools execute behind a guarded gateway.

The fix is not “add one more regex filter before the prompt.”

The real shift is architectural: do not treat retrieved content as instructions. Treat it as untrusted input that must pass through a trust boundary before it is allowed to shape context or trigger tools. Omega Walls is built around exactly that idea. In the project docs, it sits between untrusted content, the model loop, and the tool layer; it projects each chunk into structured risk signals, keeps a session-scoped risk state across steps, and can react with actions such as SOFT_BLOCK, SOURCE_QUARANTINE, TOOL_FREEZE, and HUMAN_ESCALATE.

That matters because many agent attacks are not single-message events. They build across retrieved chunks, memory carry-over, tool outputs, and related steps. Omega’s design explicitly models that: packet-level aggregation, cross-wall reinforcement, state accumulation, and deterministic Off conditions instead of one-shot input scanning.

Why LangChain is a good first integration target

LangChain is a clean first framework for this because the integration point is obvious.

LangChain already treats middleware as a first-class runtime control layer. Omega already ships an official LangChain adapter. That means you do not need to redesign your agent or fork your stack. You keep your existing agent shape, then insert a guard at the execution boundary LangChain already exposes.

In Omega’s framework docs, the LangChain path is intentionally small: install the integration extra, create OmegaLangChainGuard, pass guard.middleware() into create_agent, then verify behavior with python scripts/smoke_langchain_guard.py --strict.

Install the integration

Start with the integration extras:

pip install "omega-walls[integrations]"

The current PyPI package exposes integrations as an extra, alongside api, attachments, and train, and the package is positioned as a stateful prompt-injection defense for RAG and agent pipelines.

Minimal LangChain wiring

Here is the smallest useful wiring:

from langchain.agents import create_agent
from omega.integrations import OmegaLangChainGuard

def get_customer_note(customer_id: str) -> str:
    # Example tool. Replace with your own CRM, KB, or ticket fetch.
    return f"Customer {customer_id}: recent notes loaded."

guard = OmegaLangChainGuard(profile="quickstart")

agent = create_agent(
    model="openai:gpt-4.1-mini",
    tools=[get_customer_note],
    middleware=guard.middleware(),
)

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Summarize the latest customer note and tell me if anything looks risky."
            }
        ]
    }
)

print(result)

This follows Omega’s LangChain adapter contract directly: OmegaLangChainGuard(profile="quickstart"), then middleware=guard.middleware() on the agent.

What changes after this is not your UX. It is your trust model.

The input path is normalized and checked through the guard. Tool calls can be checked before execution. Memory writes can be evaluated with source and trust tags. On the allow path, the adapter stays transparent. On the block path, you get typed exceptions and structured decisions instead of vague failure.

Handle blocked paths explicitly

Do not hide the blocked path. Model it.

from omega.adapters import OmegaBlockedError, OmegaToolBlockedError

try:
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "Summarize this note and continue the workflow."
                }
            ]
        }
    )
    print(result)

except OmegaBlockedError as exc:
    print("Blocked model/input step")
    print(exc.decision.control_outcome)
    print(exc.decision.reason_codes)

except OmegaToolBlockedError as exc:
    print("Blocked tool call")
    print(exc.gate_decision.tool_name)
    print(exc.gate_decision.reason)

Omega’s integration docs make this contract explicit: blocked model or input steps raise OmegaBlockedError, blocked tool calls raise OmegaToolBlockedError, and the decision payload gives you control outcomes and reason codes you can route into logging or operator workflows.

That is an underrated point. Good guardrails do not just stop things. They tell the rest of your application what happened in a shape the rest of your application can actually use.

Verify that the integration is real

After wiring the middleware, do not stop at “it imports.”

Run the strict LangChain smoke:

python scripts/smoke_langchain_guard.py --strict

Omega ships that exact smoke path for the LangChain adapter. The point is simple: prove that the guard is not just present, but actually sitting on the execution path you think it is sitting on.

This is where a lot of “guardrails” fail in practice. The wrapper exists. The middleware is registered. The demo runs. But one path still bypasses the gateway, or one tool still executes outside the guard. A strict smoke is boring, and boring is good.

What Omega adds beyond one-shot filtering

The usual failure mode in these systems is isolation.

A single document does not look dangerous enough. A single step does not cross the threshold. A single tool result looks routine. The problem emerges only when the system accumulates pressure across steps.

Omega is built to operate on that exact shape. The docs describe the runtime as packet-based and stateful: it projects chunks into wall-pressure vectors, aggregates packet pressure, computes toxicity, accumulates session-scoped state, and then reacts when the pattern becomes strong enough. In plain English: it does not assume that every bad workflow announces itself in one obvious prompt.

The same docs also spell out the default action pattern: soft-block toxic documents first, freeze tools when tool abuse participates, escalate when exfiltration participates, and treat shutdown as controlled degradation rather than a blind hard stop. That is a sane design choice for production systems, because the goal is not to make the app brittle. The goal is to make it harder to steer.

Start in monitor mode, not enforce mode

The safest mistake here is not technical. It’s rollout. Don’t jump from “middleware added” straight to hard blocking.

A safer rollout path: wire the guard, verify it is really on the execution path, observe in monitor mode, add operator workflow, then enforce.

This is the part most teams skip.

Omega’s own quickstart recommends a monitor-first validation phase before enforcement. The project docs are very explicit here: run the local monitor smoke, inspect the timeline and aggregated report, confirm that risky samples produce a non-allow intended outcome, and only then move toward production hardening and enforce mode.

Use this path first:

python scripts/smoke_monitor_mode.py --profile dev --projector-mode pi0
omega-walls report --session monitor-smoke --events-path <events_path> --format json
omega-walls explain --session monitor-smoke --events-path <events_path> --format json

In monitor mode, the expected behavior is subtle but important: the attack sample should show intended_action != ALLOW, while the actual_action can still remain ALLOW. That is not a bug. That is the whole point of monitor mode. It lets you validate the risk logic before you start interrupting workflows.

This is the rollout path I would actually use in production:

Wire the guard into LangChain.
Run strict smoke locally.
Enable monitor mode in a non-trivial workflow.
Inspect reports and explain output.
Add alerting and approvals.
Move to enforcement only after operators can see and resolve the outcomes.

That last step matters because Omega’s docs also require alerts and approvals before production enforcement, specifically to avoid silent workflow pauses and make escalations observable.

Logging is not an afterthought

If you are putting a trust boundary into an agent loop, logging is part of the feature, not paperwork.

Omega’s logging and audit contract is built around reproducibility: an Off decision should be replayable from structured logs, using projector outputs, configuration references, and state snapshots. By default, production logging is designed to avoid storing raw content unless capture policy explicitly allows it, and the audit schema includes top contributors, actions taken, and tool-freeze state.

That is the right shape for real systems. When something gets blocked, “the model acted weird” is not enough. You want to know which source pushed the workflow, what the system saw, what action it took, and whether the same event can be replayed later.

What this does not claim

It is worth saying this plainly.

Omega Walls is not a general-purpose security firewall. It does not replace infrastructure security, secret management, model-native safeguards, or moderation for direct user jailbreaks. Its guarantees depend on architecture: untrusted content has to pass through the boundary before it enters context, and tool execution has to stay behind a single gateway. If your stack bypasses those two points, the protection model breaks with it.

That is not a weakness in the write-up. It is a sign that the boundary is being described honestly.

Closing thought

A lot of agent security writing still assumes the main problem is a bad prompt.

In production, the bigger problem is usually trust confusion.

Your agent reads external data. Your tools bring more external data back. Your memory carries state forward. Somewhere in that loop, normal-looking text starts behaving like control.

That is why the right place to intervene is not just the prompt input. It is the boundary between untrusted content, context construction, and tool execution.

If you are already running LangChain, this is a small integration. More importantly, it is the right shape of integration.

Install the adapter. Wire the middleware. Run the strict smoke. Start in monitor mode. Then decide where enforcement belongs in your workflow.

GitHub: https://github.com/synqratech/omega-walls
PyPI: https://pypi.org/project/omega-walls/
Site: https://synqra.tech/omega-walls

We open-sourced Omega Walls: a stateful runtime defense for RAG and AI agents

Anton Fedotov — Tue, 14 Apr 2026 13:48:14 +0000

Most prompt-injection defenses still think in single turns.

But many real agent failures do not happen in one prompt. They build across retrieved documents, memory carry-over, tool outputs, and later execution.

That is the problem we built Omega Walls for.

Today we’re open-sourcing Omega Walls, a Python runtime defense layer for RAG and tool-using agents.

What Omega Walls does

Omega Walls sits at two important runtime points:

Before final context assembly
Retrieved chunks, emails, tickets, attachments, and tool outputs can be inspected before they are allowed into model context.
At tool execution
Tool calls can be constrained or blocked when accumulated risk crosses the boundary.

Instead of treating each chunk as an isolated moderation problem, Omega Walls turns untrusted content into session-level risk state and emits deterministic runtime actions such as:

block
freeze
quarantine
attribution / reason flags

What it is built for

Omega Walls is designed for:

indirect prompt injection
distributed attacks across multiple chunks or turns
cocktail attacks that combine takeover, exfiltration, tool abuse, and evasion
multi-step flows where no single step looks obviously malicious in isolation

Why we open-sourced it

We think agent security needs more work on runtime trust boundaries, not only better prompt scanning.

If you are building:

RAG pipelines
internal copilots
support or inbox agents
tool-using workflows
agent infrastructure

we’d love your feedback.

GitHub: https://github.com/synqratech/omega-walls
Website: https://synqra.tech/omega-walls

PyPI: https://pypi.org/project/omega-walls/

If you try it, tell us where it breaks, what attack patterns you think matter most, and where this layer should sit in a real stack.

Why we didn’t use an LLM-first approach for architectural drift detection

Anton Fedotov — Wed, 18 Mar 2026 08:44:29 +0000

Why we didn’t use an LLM-first approach for architectural drift detection

LLMs are very good at a lot of things in software development.

They can explain code, summarize pull requests, suggest fixes, and point out suspicious logic. For many review tasks, they are genuinely useful.

But when we started working on architectural drift detection, we ran into a different kind of problem.

Architectural drift is usually not a single “bad line of code”.
It is a gradual shift in the shape of a codebase:

boundaries get blurred,
hidden coupling appears,
new state starts leaking into places that used to stay simple,
control flow becomes more irregular,
repo-specific patterns quietly erode over time.

And that is where an LLM-first approach started to feel like the wrong primary layer.

The core issue: architectural drift is not just code understanding

A generic LLM can read code and reason about it.

But architectural drift is not only about understanding what a piece of code does.
It is about understanding whether a change is structurally abnormal for this repository.

That distinction matters.

A pattern can be valid in isolation and still be a bad architectural move in a specific repo.

For example:

introducing a new abstraction where the repo has stayed intentionally simple,
adding hidden state into an area that has historically stayed stateless,
crossing a module boundary that the team has treated as stable,
making a PR that is locally reasonable but globally erosive.

An LLM can often describe such code.
But detecting that it is out of character for this codebase is a different task.

Why LLM-first review was not enough for us

1. The decision is often local, but the damage is global

Large language models are very strong at local reasoning over the code they can see.

But architecture is a global property.
A pull request can look fine line by line while still moving the whole system in the wrong direction.

That is why drift often survives normal review:
tests pass, the code works, nothing looks obviously broken — but the shape of the system worsens.

2. Repo-specific baselines matter more than general code knowledge

Most AI review tools are built around broad priors learned from many repositories.

That is useful for generic review.
It is less useful for questions like:

“Is this kind of abstraction typical here?”
“Does this boundary crossing fit the historical structure of this repo?”
“Is this new dependency normal for this subsystem?”
“Is this complexity spike expected here or is it architectural drift?”

Those are not universal questions.
They are baseline questions.

3. Drift detection needs stability, not just plausible reasoning

For architecture work, noisy comments are deadly.

If the system raises too many vague or unstable warnings, teams stop trusting it very quickly.

We needed a layer that behaves more like structural instrumentation:
repeatable, calibrated, and tied to measurable deviation — not just a smart narrative about code.

4. Explanation and detection are different jobs

LLMs are often excellent at the second part:
explaining why something may be risky.

But the first part — consistently detecting structural deviation relative to a repo baseline — is a separate problem.

We found it useful to separate those two jobs instead of forcing one model to do both.

What we built instead

We built a non-linguistic structural layer first.

The idea is simple:

learn the repository’s structural baseline,
compare each PR against that baseline,
score the deviation,
surface a short risk summary and a few hotspots directly in the PR.

In our case, this became PhaseBrain inside Revieko.

The model is not trying to replace LLMs.
It is trying to do something narrower and more structural:
track roles, boundaries, deviations, and coherence in the evolution of a repo.

That gives us a better primary signal for architectural drift.

Then, if needed, language models can sit on top of that signal and help explain it.

Our view now

For this problem, LLMs are useful — but not as the foundation.

They are strong explainers.
They are not the best primary detector of repo-specific architectural drift.

Architectural drift is less about “what does this code mean?”
and more about
“what does this change do to the structure of this system over time?”

That pushed us toward a structural model first, and a language layer second.

That is the architecture we ended up building.

If you work on long-lived repos, I’d be very interested in your view:

Have you seen PRs that looked reasonable locally but still degraded system structure?
Do you think architectural drift is better modeled as a structural signal than as a pure language task?

Revieko:
https://synqra.tech/revieko