Introduction: The Hidden Failure Mode of Inbound Evaluation Agents

B2B revenue teams increasingly build custom inbound evaluation agents on orchestration frameworks such as LangChain (via LangGraph) and LlamaIndex (via event-driven Workflows). These agents qualify enterprise leads, parse technical RFPs, inspect architecture diagrams, and answer product compliance questions before a discovery call is booked.

In short interactions, these agents work well. Enterprise evaluations are rarely short. A prospect uploads a SOC 2 questionnaire, pastes a large JSON schema, and negotiates SLA terms over many turns. At that point an autonomous agent faces two problems: it can run into a hard token or rate limit, and its answer quality can degrade well before that limit.

The second problem is documented. In July 2025, Chroma published a technical report called *Context Rot*, which tested 18 LLMs. It found that models do not use their context uniformly, and that performance becomes less reliable as input length grows, even on simple tasks. Large context windows therefore don’t make it safe to let an agent run until it hits the ceiling. Windows now reach 1M tokens or more, but reliability can fall off long before the window is full.

Cost also grows with conversation length. Most chat APIs are stateless, so every turn re-sends the accumulated history. Cumulative input tokens across a conversation therefore grow roughly quadratically with the number of turns, unless provider-side prompt caching reduces the bill. And if the agent errors out, loops, or improvises a pricing commitment late in a long session, you may lose a high-value deal.

The fix is proactive, telemetry-driven escalation. Intercept the agent *before* it exhausts its budget or starts to degrade. Checkpoint its state, compress the qualification context into a short brief, and post an interactive claiming card to a Slack channel. A human Account Executive (AE) or Solutions Engineer (SE) can then claim the lead and continue the conversation.

This guide covers the architecture and working patterns for LangGraph, LlamaIndex Workflows, Redis locking, and Slack Bolt. The code is a starting point rather than a drop-in product. Thresholds, model names, and storage choices should be tuned to your own traffic.

---

1. Token Exhaustion vs. Context Rot in Multi-Turn Evaluations

Two distinct failure thresholds apply to multi-turn agent loops.

Hard token exhaustion (the infrastructure ceiling)

The prompt sent on turn *n* contains the system prompt, the full history of prior user, assistant, and tool messages, the new user message, and any retrieved RAG chunks:

```

prompt_tokens(n) = system + Σ(i < n) [user_i + assistant_i + tool_output_i] + user_n + rag_chunks_n

```

Trouble starts when this approaches the model’s maximum context window, or when sustained usage trips an API tokens-per-minute (TPM) rate limit. A single turn with a large document upload and a wide retrieval step can add tens of thousands of tokens. How fast you reach the limit depends on your documents, chunk sizes, and tools, so measure it on real transcripts instead of assuming a number.

Soft exhaustion: context rot and reasoning loops

Context rot is the decline in output quality as input grows, even when the task stays equally hard. Chroma’s work showed it varies by model, by task, and by how similar the distractor content is to the question. Retrieval quality also affects it. In inbound sales evaluations it tends to show up as:

• Circular clarification loops: the agent re-asks for deployment constraints the prospect already gave.
• Tool-calling thrash: the agent calls the same CRM lookup or documentation retriever repeatedly with slightly altered arguments and makes no progress.
• Policy drift: the agent stops following negative constraints in the system prompt (for example, “never quote custom discounts without SE approval”).

The escalation threshold matrix

Because degradation is gradual and model-specific, define an operational token budget well below the provider’s maximum window. Set it from evals on your own transcripts, for example the point where qualification accuracy or instruction-following starts to slip. The thresholds below are illustrative starting points, not industry standards. The token rows assume an operational budget of 64k tokens.

A simple composite rule works well. Any critical signal escalates immediately. Two soft signals, or one soft signal combined with high intent, also escalate.

---

2. System Architecture: Decoupling Agent Execution from Human Claiming

Don’t hold an HTTP request or a Python worker open while a human decides to answer in Slack. A closer may claim a lead in seconds, or the lead may sit unclaimed for minutes or longer. The agent’s execution state therefore has to be checkpointed to durable storage, suspended, and resumed through events.

```

┌──────────────────────┐ ┌──────────────────────────┐ ┌───────────────────────┐

│ Inbound Prospect │◄─────►│ API Gateway / Router │◄─────►│ Redis (locks, maps) │

│ (Web Chat / Widget) │ │ (FastAPI + WebSockets) │ │ + durable checkpoints│

└──────────────────────┘ └────────────┬─────────────┘ └───────────┬───────────┘

│ │

┌──────────────────┴──────────────────┐ │

▼ ▼ │

┌─────────────────────────────────┐ ┌─────────────────────────────────┐

│ LangGraph Execution Engine │ │ LlamaIndex Workflow Engine │

│ (token guard + interrupt()) │ │ (token guard + InputRequired) │

└────────────────┬────────────────┘ └────────────────┬────────────────┘

└──────────────────┬──────────────────┘

▼ (escalation payload)

┌─────────────────────────┐

│ Handoff Brief Generator│

└────────────┬────────────┘

▼

┌─────────────────────────┐

│ Slack Bolt App / API │

│ (#inbound-hot-queue) │

└────────────┬────────────┘

┌───────────────────────┴───────────────────────┐

▼ ▼

┌─────────────────────────┐ ┌─────────────────────────┐

│ AE #1 clicks "Claim" │ │ AE #2 clicks "Claim" │

│ (acquires Redis lock) │ │ (rejected: already won)│

└────────────┬────────────┘ └─────────────────────────┘

▼

┌─────────────────────────┐

│ Bidirectional Slack │

│ thread <-> web widget │

└─────────────────────────┘

```

Core components

1. Token and telemetry interceptor: logic that runs after every model step. It reads provider-reported usage (LangChain exposes `usage_metadata` on `AIMessage`, with `input_tokens`, `output_tokens`, and `total_tokens`), or counts with a tokenizer when metadata is unavailable. It also tracks tool-call repetition and qualification state.

2. State checkpointer: LangGraph persists graph state through a checkpointer. Postgres (`langgraph-checkpoint-postgres`) is the usual production choice, and a Redis checkpointer (`langgraph-checkpoint-redis`) is also available. LlamaIndex Workflows instead serialize their `Context` with `ctx.to_dict()`, which you store yourself.

3. Handoff brief generator: a small, fast model that runs only on escalation. It turns structured qualification state plus the recent transcript into a scannable brief for the Slack card.

4. Atomic claiming queue: a Slack channel driven by Slack Bolt, with a Redis `SET key value NX EX ttl` lock so two closers can’t both win the same lead.

---

3. Token-Aware Interrupts in LangChain (LangGraph)

LangGraph’s `interrupt()` pauses a graph at a node and saves state through the checkpointer. The graph waits indefinitely until you resume it with `Command(resume=...)`. Two requirements follow from the LangGraph docs:

• You need a checkpointer, and a durable one in production.
• You must resume with the same `thread_id` that was used when the interrupt fired.

There is a third rule that is easy to miss. When a graph resumes, the interrupted node re-runs from its beginning. Any side effect placed before `interrupt()` in the same node, such as an LLM summary call or a Slack post, will therefore execute twice. The pattern below puts side effects in a separate node that completes and is checkpointed before the interrupting node runs. It also makes the Slack post idempotent.

Step 1: State with separate context and billing counters

`total_tokens` on a single response covers that call’s input plus output. The input already includes the full history, so summing it across turns measures billed usage, not context size. Track both numbers. Use the latest call’s total as the proxy for how large the next prompt will be.

```

import hashlib

import json

import os

from collections import Counter

from typing import Annotated, Literal, Optional, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage

from langchain_openai import ChatOpenAI

from langgraph.graph import END, START, StateGraph

from langgraph.graph.message import add_messages

from langgraph.prebuilt import ToolNode

from langgraph.types import Command, interrupt

class InboundEvalState(TypedDict, total=False):

messages: Annotated[list[BaseMessage], add_messages]

session_id: str

lead_email: Optional[str]

company_domain: Optional[str]

high_intent: bool # set by an upstream classifier

user_requested_human: bool # set by an upstream classifier

context_tokens: int # size of the latest model call (next-prompt proxy)

billed_tokens: int # cumulative usage, for cost and TPM tracking

tool_call_history: list[str]

escalation_reason: Optional[str]

handoff_brief: Optional[str]

human_assigned_id: Optional[str]

```

Step 2: Agent node and escalation guard

```

OPERATIONAL_BUDGET = 64_000

SOFT_TOKENS = int(OPERATIONAL_BUDGET * 0.55)

HARD_TOKENS = int(OPERATIONAL_BUDGET * 0.75)

SOFT_TURNS, HARD_TURNS = 6, 9

Model names are placeholders; use whichever models you've validated.

llm = ChatOpenAI(model=os.environ["AGENT_MODEL"], temperature=0.1)

summarizer_llm = ChatOpenAI(model=os.environ["HANDOFF_MODEL"], temperature=0)

TOOLS = [] # your CRM lookup, docs retriever, etc.

llm_with_tools = llm.bind_tools(TOOLS) if TOOLS else llm

def tool_signature(name: str, args: dict) -> str:

payload = json.dumps({"name": name, "args": args}, sort_keys=True, default=str)

return hashlib.sha256(payload.encode()).hexdigest()

def agent_node(state: InboundEvalState):

response: AIMessage = llm_with_tools.invoke(state["messages"])

usage = response.usage_metadata or {}

call_tokens = usage.get("total_tokens", 0)

history = list(state.get("tool_call_history", []))

for tc in response.tool_calls:

history.append(tool_signature(tc["name"], tc["args"]))

return {

"messages": [response],

"context_tokens": call_tokens,

"billed_tokens": state.get("billed_tokens", 0) + call_tokens,

"tool_call_history": history,

}

def guard_node(state: InboundEvalState):

tokens = state.get("context_tokens", 0)

user_turns = sum(isinstance(m, HumanMessage) for m in state["messages"])

max_repeat = max(Counter(state.get("tool_call_history", [])).values(), default=0)

Critical signals: escalate immediately.

if state.get("user_requested_human"):

return {"escalation_reason": "Prospect asked for a human"}

if tokens >= HARD_TOKENS:

return {"escalation_reason": f"Context {tokens:,}/{OPERATIONAL_BUDGET:,} tokens"}

if user_turns >= HARD_TURNS:

return {"escalation_reason": f"{user_turns} user turns without a booking"}

if max_repeat >= 3:

return {"escalation_reason": "Repeated identical tool calls"}

Soft signals: two together, or one plus high intent.

soft = sum([tokens >= SOFT_TOKENS, user_turns >= SOFT_TURNS, max_repeat >= 2])

if soft >= 2 or (soft >= 1 and state.get("high_intent")):

return {"escalation_reason": "Composite soft-signal threshold"}

return {"escalation_reason": None}

def route_after_guard(state: InboundEvalState) -> Literal["dispatch_handoff", "tools", "__end__"]:

if state.get("escalation_reason"):

return "dispatch_handoff"

last = state["messages"][-1]

if isinstance(last, AIMessage) and last.tool_calls:

return "tools"

return END

```

Step 3: Dispatch node (side effects) and interrupt node (pause only)

```

def dispatch_handoff_node(state: InboundEvalState):

Compress structured state plus the most recent turns into a short brief.

recent = "\n".join(

f"{m.type}: {str(m.content)[:1500]}" for m in state["messages"][-10:]

)

brief = summarizer_llm.invoke([

SystemMessage(content=(

"Summarize this inbound technical evaluation in four bullets: "

"1) Prospect and use case, 2) Technical requirements, "

"3) Unresolved blocker, 4) Recommended next step. "

"Do not invent commitments, prices, or SLAs."

)),

HumanMessage(content=(

f"Company: {state.get('company_domain')}\n"

f"Escalation reason: {state.get('escalation_reason')}\n\n{recent}"

)),

]).content

Must be idempotent (see the Redis SET NX guard in section 5).

publish_to_slack_queue(

session_id=state["session_id"],

lead_email=state.get("lead_email") or "Unknown",

reason=state["escalation_reason"],

context_tokens=state.get("context_tokens", 0),

budget=OPERATIONAL_BUDGET,

brief=brief,

)

return {"handoff_brief": brief}

def await_closer_node(state: InboundEvalState):

Nothing else may run before interrupt(): this node re-executes on resume.

reply = interrupt({

"status": "awaiting_human_closer",

"session_id": state["session_id"],

"brief": state["handoff_brief"],

})

return {

"human_assigned_id": reply["closer_slack_id"],

"messages": [AIMessage(content=reply["closer_message"])],

}

```

Step 4: Wire the graph and drive it

```

builder = StateGraph(InboundEvalState)

builder.add_node("agent", agent_node)

builder.add_node("guard", guard_node)

builder.add_node("tools", ToolNode(TOOLS))

builder.add_node("dispatch_handoff", dispatch_handoff_node)

builder.add_node("await_closer", await_closer_node)

builder.add_edge(START, "agent")

builder.add_edge("agent", "guard")

builder.add_conditional_edges("guard", route_after_guard)

builder.add_edge("tools", "agent")

builder.add_edge("dispatch_handoff", "await_closer")

builder.add_edge("await_closer", END)

`checkpointer` is a durable saver, e.g. PostgresSaver or RedisSaver.

graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": session_id}}

Normal turn. If the guard escalates, the result carries an "__interrupt__" entry.

result = graph.invoke({"messages": [HumanMessage(content=user_text)], "session_id": session_id}, config)

Later, when the closer first replies from Slack:

graph.invoke(

Command(resume={"closer_slack_id": "U123", "closer_message": "Hi, I'm Priya from solutions..."}),

config,

)

```

If your service is async, use the async checkpointer variants (for example `AsyncPostgresSaver`) and `ainvoke`.

---

4. Event-Driven Escalation in LlamaIndex Workflows

LlamaIndex Workflows are asynchronous and event-driven. Their human-in-the-loop support is built on `InputRequiredEvent` and `HumanResponseEvent`, and both can be subclassed to carry custom payloads. The docs also describe how to pause a workflow and resume it in a different request, which is the scenario in a web app. Snapshot the context with `handler.ctx.to_dict()` and store it. Cancel the run, then later rebuild it with `Context.from_dict(workflow, ...)`, start a new run with that context, and send the human response event.

In current releases these classes live in the `workflows` package (`from workflows import Workflow, Context, step` and `from workflows.events import ...`). Older tutorials import them from `llama_index.core.workflow`, so check the version you have installed. Context state is read and written through `ctx.store`.

Two implementation notes apply here. Keep chat history as plain JSON-safe dicts, so that serializing the context stays simple. And `tiktoken` only matches OpenAI tokenizers, so treat its counts as an estimate for other providers and add headroom.

```

import json

import tiktoken

from llama_index.core.llms import ChatMessage

from llama_index.llms.openai import OpenAI

from workflows import Context, Workflow, step

from workflows.events import HumanResponseEvent, InputRequiredEvent, StartEvent, StopEvent

encoding = tiktoken.get_encoding("o200k_base") # estimate for non-OpenAI models

class EscalationTriggeredEvent(InputRequiredEvent):

session_id: str

reason: str

token_count: int

executive_summary: str

class CloserReplyEvent(HumanResponseEvent):

closer_id: str

reply_text: str

class InboundEvaluationWorkflow(Workflow):

def __init__(self, token_threshold: int = 48_000, **kwargs):

super().__init__(**kwargs)

self.token_threshold = token_threshold

self.llm = OpenAI(model=os.environ["AGENT_MODEL"])

self.fast_llm = OpenAI(model=os.environ["HANDOFF_MODEL"])

@staticmethod

def _count_tokens(history: list[dict]) -> int:

Excludes system prompt and retrieved chunks; add them for a fuller estimate.

return sum(len(encoding.encode(m["content"] or "")) for m in history)

@step

async def evaluate_turn(

self, ctx: Context, ev: StartEvent

) -> StopEvent | EscalationTriggeredEvent:

history: list[dict] = await ctx.store.get("chat_history", default=[])

history.append({"role": "user", "content": ev.query})

tokens = self._count_tokens(history)

Check the budget BEFORE calling the primary reasoning model.

if tokens >= self.token_threshold:

recent = json.dumps(history[-8:])

summary = await self.fast_llm.acomplete(

"Create a concise B2B sales handoff summary from this log. "

f"Do not invent commitments or prices:\n{recent}"

)

await ctx.store.set("chat_history", history)

return EscalationTriggeredEvent(

prefix="Token threshold reached. Routing to a human closer.",

session_id=ev.session_id,

reason=f"Token budget reached ({tokens}/{self.token_threshold})",

token_count=tokens,

executive_summary=summary.text,

)

response = await self.llm.achat([ChatMessage(**m) for m in history])

history.append({"role": "assistant", "content": response.message.content})

await ctx.store.set("chat_history", history)

return StopEvent(result=response.message.content)

@step

async def handle_closer_takeover(self, ctx: Context, ev: CloserReplyEvent) -> StopEvent:

history: list[dict] = await ctx.store.get("chat_history", default=[])

history.append({"role": "assistant", "content": ev.reply_text})

await ctx.store.set("chat_history", history)

await ctx.store.set("assigned_closer", ev.closer_id)

return StopEvent(result=ev.reply_text)

```

Suspending and restoring the context across processes

The worker that handles the prospect’s request shouldn’t wait on Slack. When the stream emits the escalation event, snapshot the context, post the Slack card, and cancel the run.

```

async def run_llama_turn(workflow, session_id: str, user_msg: str, redis_client):

key = f"llama_ctx:{session_id}"

raw = await redis_client.get(key)

if raw:

ctx = Context.from_dict(workflow, json.loads(raw))

handler = workflow.run(ctx=ctx, query=user_msg, session_id=session_id)

else:

handler = workflow.run(query=user_msg, session_id=session_id)

async for event in handler.stream_events():

if isinstance(event, EscalationTriggeredEvent):

await redis_client.setex(key, 86_400, json.dumps(handler.ctx.to_dict()))

await push_slack_claiming_card(event)

await handler.cancel_run()

return {"status": "escalated",

"message": "Connecting you with a solutions specialist now..."}

final_answer = await handler

await redis_client.setex(key, 86_400, json.dumps(handler.ctx.to_dict()))

return {"status": "autonomous", "message": final_answer}

async def resume_with_closer_reply(workflow, session_id, closer_id, text, redis_client):

ctx = Context.from_dict(workflow, json.loads(await redis_client.get(f"llama_ctx:{session_id}")))

handler = workflow.run(ctx=ctx)

await handler.send_event(CloserReplyEvent(closer_id=closer_id, reply_text=text))

return await handler

```

Check the exact `send_event` call form against the Workflows version you have installed. Older releases used `handler.ctx.send_event(...)`.

---

5. The Live Slack Claiming Queue: Slack Bolt and Redis Locks

If an AE receives a raw dump of 40 messages, the handoff will probably be ignored. If two AEs claim the same lead and both message the prospect, the buyer experience falls apart. The card has to be scannable, and claiming has to be atomic.

Designing the Block Kit card

A good claiming card includes:

1. A short header with telemetry: the escalation reason and how much of the token budget was used.

2. Lead metadata: email or company domain, and a CRM link.

3. The generated brief: the four-bullet summary from the dispatch step.

4. Two buttons: a primary claim button and a secondary button that opens the full transcript in a modal, so the channel stays clean.

Mind Slack’s Block Kit limits. A message can hold up to 50 blocks, a `header` is capped at 150 characters, and a `section` text field at 3,000 characters. Always include a plain-text `text` fallback for notifications.

Preventing double-claims with an atomic lock

When someone clicks a button, Slack sends a `block_actions` interaction payload to your app. Your app has 3 seconds to acknowledge it, so call `await ack()` first. Then take the lock with a single atomic Redis `SET ... NX EX` command. Only the first request that sets the key wins. Everyone else gets an ephemeral message saying who has the lead.

```

import os

import redis.asyncio as redis

from slack_bolt.async_app import AsyncApp

slack_app = AsyncApp(

token=os.environ["SLACK_BOT_TOKEN"],

signing_secret=os.environ["SLACK_SIGNING_SECRET"],

)

redis_pool = redis.from_url(os.environ["REDIS_URL"], decode_responses=True)

def build_claiming_blocks(session_id, lead_email, reason, context_tokens, budget, brief):

return [

{"type": "header",

"text": {"type": "plain_text", "text": "🔥 Inbound evaluation ready for a closer"}},

{"type": "section", "fields": [

{"type": "mrkdwn", "text": f"*Prospect:*\n{lead_email}"},

{"type": "mrkdwn",

"text": f"*Trigger:*\n{reason} ({context_tokens / budget:.0%} of budget)"},

]},

{"type": "section",

"text": {"type": "mrkdwn", "text": f"*Handoff brief:*\n{brief[:2800]}"}},

{"type": "actions", "block_id": f"claim_actions_{session_id}", "elements": [

{"type": "button", "style": "primary", "action_id": "claim_inbound_lead",

"text": {"type": "plain_text", "text": "Claim lead"}, "value": session_id},

{"type": "button", "action_id": "view_raw_transcript",

"text": {"type": "plain_text", "text": "View transcript"}, "value": session_id},

]},

]

@slack_app.action("claim_inbound_lead")

async def handle_claim_lead(ack, body, client):

await ack() # within 3 seconds

user_id = body["user"]["id"]

session_id = body["actions"][0]["value"]

channel_id = body["channel"]["id"]

message_ts = body["message"]["ts"]

Atomic: only the first caller sets the key.

lock_key = f"lead_claim_lock:{session_id}"

acquired = await redis_pool.set(lock_key, user_id, nx=True, ex=86_400)

if not acquired:

owner = await redis_pool.get(lock_key)

await client.chat_postEphemeral(

channel=channel_id, user=user_id,

text=f"This evaluation was already claimed by <@{owner}>.",

)

return

Replace the actions block so nobody else can click.

blocks = body["message"]["blocks"][:-1]

blocks.append({"type": "context", "elements": [

{"type": "mrkdwn", "text": f"✅ *Claimed by <@{user_id}>*. Reply in this thread to message the prospect."}

]})

await client.chat_update(

channel=channel_id, ts=message_ts, blocks=blocks,

text=f"Lead {session_id} claimed by <@{user_id}>",

)

Routing maps for bidirectional messaging.

await redis_pool.setex(f"slack_thread_map:{channel_id}:{message_ts}", 86_400, session_id)

await redis_pool.setex(f"session_slack_route:{session_id}", 86_400,

f"{channel_id}:{message_ts}:{user_id}")

await client.chat_postMessage(

channel=channel_id, thread_ts=message_ts,

text=f"<@{user_id}> You're connected to session `{session_id}`. "

"Messages you post in this thread are delivered to the buyer's chat.",

)

```

The same `SET NX` pattern makes the Slack post idempotent in the LangGraph dispatch node. Use something like `handoff_posted:{session_id}` as the key, and skip the post if the key already exists.

Opening the transcript modal

A modal’s `trigger_id` expires 3 seconds after Slack sends it. Don’t fetch a large transcript before calling `views_open`. Open a placeholder first, then update it.

```

@slack_app.action("view_raw_transcript")

async def handle_view_transcript(ack, body, client):

await ack()

opened = await client.views_open(

trigger_id=body["trigger_id"],

view={"type": "modal", "title": {"type": "plain_text", "text": "Transcript"},

"blocks": [{"type": "section", "text": {"type": "mrkdwn", "text": "Loading..."}}]},

)

transcript = await load_transcript(body["actions"][0]["value"]) # your storage

await client.views_update(

view_id=opened["view"]["id"],

view={"type": "modal", "title": {"type": "plain_text", "text": "Transcript"},

"blocks": [{"type": "section",

"text": {"type": "mrkdwn", "text": transcript[:2900]}}]},

)

```

Transcripts can contain confidential customer data, such as questionnaires and schemas. Restrict the queue channel to the people who need it, and consider redacting sensitive fields before posting.

---

6. Closing the Loop: Bidirectional Messaging and Unclaimed-Lead SLAs

After a closer claims the lead, the agent stops being the responder. It becomes a bridge for state and routing.

1. Closer to prospect: a Bolt `message` event listener looks up `slack_thread_map:{channel}:{thread_ts}`. It ignores bot messages and anyone other than the claimant, then pushes the text to the web widget over WebSocket or SSE. It also appends the human turn to the stored history, so the audit trail stays in one place. Reading channel messages requires the matching history scope (such as `channels:history` or `groups:history`).

2. Prospect to closer: while `lead_claim_lock:{session_id}` exists, your API router skips LLM inference entirely and forwards the prospect’s message to the Slack thread with `chat_postMessage(..., thread_ts=...)`. This also removes model costs for those turns.

3. Unclaimed-lead timeout: pick an SLA, for example 120 seconds. When the card is posted, schedule a delayed job with a durable scheduler such as Celery with an ETA, Temporal timers, or a Redis sorted set polled by a worker. When it fires, it checks whether the claim lock exists. If not, it pages on-call through PagerDuty or an `@channel` alert. It can also offer the prospect a priority calendar link in the widget, so they aren’t left on a frozen spinner.

```

@slack_app.event("message")

async def relay_closer_reply(event, logger):

if event.get("bot_id") or event.get("subtype") or "thread_ts" not in event:

return

key = f"slack_thread_map:{event['channel']}:{event['thread_ts']}"

session_id = await redis_pool.get(key)

if not session_id:

return

route = await redis_pool.get(f"session_slack_route:{session_id}")

claimant = route.split(":")[-1] if route else None

if event["user"] != claimant:

return

await push_to_widget(session_id, event["text"]) # WebSocket/SSE to the buyer

```

Don’t build the unclaimed-lead timeout on Redis keyspace expiration events. Redis documents keyspace notifications as Pub/Sub, which is fire-and-forget: if your subscriber disconnects, the events sent while it was away are lost. Expired-key events also aren’t emitted at the exact moment a TTL reaches zero. A missed SLA alert is the kind of silent failure this architecture exists to prevent, so use a durable scheduler.

---

7. Production Checklist

Before shipping a token-aware human routing pipeline, verify these points:

• Set the operational budget empirically. Run evals on real long transcripts to find where accuracy and instruction-following start to slip. Context rot shows up well before the window is full. Set escalation thresholds below that point, and leave headroom for the summarization call.
• Track context size and billed tokens separately. Cumulative `total_tokens` measures spend. The most recent call’s usage approximates how big the next prompt will be.
• Hash tool name and normalized arguments. Counting turns alone misses an agent that burns tens of thousands of tokens in a few turns by looping on a retriever. Hash `(tool_name, canonical_json_args)` per session, with sorted keys.
• Prefer provider usage metadata, then an exact tokenizer. Character heuristics such as `len(text) / 4` come from English prose. They tend to undercount tokens for JSON, UUIDs, and code.
• Make side effects idempotent. LangGraph re-runs an interrupted node on resume. Keep Slack posts and CRM writes out of the node that calls `interrupt()`, and guard them with an idempotency key.
• Use atomic claim locks and update the card. Pair `SET NX` with `chat_update` so the button disappears once someone claims the lead. Give locks a TTL, and provide a way to reassign or release a lead.
• Keep the queue channel high-signal. Put full transcripts behind a modal and restrict channel access.
• Use durable timers for SLAs. Don’t rely on Pub/Sub or keyspace notifications.
• Test the failure paths: two simultaneous claims, a Slack retry of the same interaction, a worker crash between dispatch and interrupt, a Redis restart, and a prospect who messages while the lead is unclaimed.

---

Sources and Further Reading

• LangGraph interrupts and human-in-the-loop: <https://docs.langchain.com/oss/python/langgraph/interrupts>
• LangChain messages and token usage metadata: <https://docs.langchain.com/oss/python/langchain-messages>
• LlamaIndex Workflows, human in the loop: <https://developers.llamaindex.ai/python/llamaagents/workflows/human_in_the_loop/>
• Chroma, *Context Rot: How Increasing Input Tokens Impacts LLM Performance* (July 2025): <https://research.trychroma.com/context-rot>
• Slack modals and `trigger_id` expiry: <https://docs.slack.dev/surfaces/modals>
• Slack Bolt for Python: <https://docs.slack.dev/tools/bolt-python/>
• Redis keyspace notifications: <https://redis.io/docs/latest/develop/pubsub/keyspace-notifications/>