adelorenzo/konstruct
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(02-06): re-wire escalation and WhatsApp outbound routing in pipeline

Browse Source 
- Move key imports to module level in tasks.py for testability and clarity
- Pop WhatsApp extras (phone_number_id, bot_token) in handle_message before model_validate
- Build unified extras dict and extract wa_id from sender.user_id
- Change _process_message signature to accept extras dict
- Add _build_response_extras() helper for channel-aware extras assembly
- Replace all _update_slack_placeholder calls in _process_message with _send_response()
- Add escalation pre-check: skip LLM when Redis escalation_status_key == 'escalated'
- Add escalation post-check: check_escalation_rules after run_agent; call escalate_to_human
  when rule matches and agent.escalation_assignee is set
- Add _build_conversation_metadata() helper (billing keyword v1 detection)
- Add channel parameter to build_system_prompt(), build_messages_with_memory(),
  build_messages_with_media() for WhatsApp tier-2 business-function scoping
- WhatsApp scoping appends 'You only handle: {topics}' when tool_assignments non-empty
- Pass msg.channel to build_messages_with_memory() in _process_message
- All 26 new tests pass; all existing escalation/WhatsApp tests pass (no regressions)
This commit is contained in:
Adolfo Delorenzo
2026-03-23 19:15:20 -06:00
parent 77c9cfc825
commit bd217a4113
3 changed files with 380 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
packages/orchestrator/orchestrator/agents/builder.py
View File

@@ -133,7 +133,7 @@ def generate_presigned_url(storage_key: str, expiry: int = 3600) -> str:
return presigned_url
def build_system_prompt(agent: Agent) -> str:
def build_system_prompt(agent: Agent, channel: str = "") -> str:
"""
Assemble the full system prompt for an agent.
@@ -142,9 +142,20 @@ def build_system_prompt(agent: Agent) -> str:
- Identity section: name + role
- Persona section (if agent.persona is non-empty)
- AI transparency clause (always appended)
- WhatsApp tier-2 scoping (only when channel == "whatsapp" and tool_assignments set)
WhatsApp Tier-2 Business-Function Scoping:
Per Meta 2026 policy, agents responding on WhatsApp must constrain themselves
to declared business functions. When channel == "whatsapp" and the agent has
non-empty tool_assignments, a scoping clause is appended that instructs the LLM
to handle only the declared topics and redirect off-topic requests.
This is the tier-2 gate (borderline messages reach the LLM with this constraint;
clearly off-topic messages are caught by the tier-1 keyword gate in the gateway).
Args:
agent: ORM Agent instance.
agent: ORM Agent instance.
channel: Channel name (e.g. "slack", "whatsapp"). Used for tier-2 scoping.
Default empty string means no channel-specific scoping.
Returns:
A complete system prompt string ready to pass to the LLM.
@@ -167,6 +178,17 @@ def build_system_prompt(agent: Agent) -> str:
"If asked directly whether you are an AI, always respond honestly that you are an AI assistant."
)
# 5. WhatsApp tier-2 scoping — constrain LLM to declared business functions
if channel == "whatsapp":
functions: list[str] = getattr(agent, "tool_assignments", []) or []
if functions:
topics = ", ".join(functions)
parts.append(
f"You are responding on WhatsApp. You only handle: {topics}. "
f"If the user asks about something outside these topics, "
f"politely redirect them to the allowed topics."
)
return "\n\n".join(parts)
@@ -204,12 +226,14 @@ def build_messages_with_memory(
current_message: str,
recent_messages: list[dict],
relevant_context: list[str],
channel: str = "",
) -> list[dict]:
"""
Build an LLM messages array enriched with two-layer memory.
Structure (in order):
1. System message — agent identity, persona, AI transparency clause
1. System message — agent identity, persona, AI transparency clause,
and optional WhatsApp tier-2 scoping (when channel == "whatsapp")
2. System message — long-term context from pgvector (ONLY if non-empty)
Injected as a system message before the sliding window so the LLM
has relevant background without it appearing in the conversation.
@@ -226,11 +250,13 @@ def build_messages_with_memory(
from Redis sliding window (oldest first).
relevant_context: Long-term memory — list of content strings from
pgvector similarity search (most relevant first).
channel: Channel name for tier-2 scoping (e.g. "whatsapp").
Default empty string means no channel-specific scoping.
Returns:
List of message dicts suitable for an OpenAI-compatible API call.
"""
system_prompt = build_system_prompt(agent)
system_prompt = build_system_prompt(agent, channel=channel)
messages: list[dict] = [{"role": "system", "content": system_prompt}]
# Inject long-term pgvector context as a system message BEFORE sliding window
@@ -256,6 +282,7 @@ def build_messages_with_media(
media_attachments: list[MediaAttachment],
recent_messages: list[dict],
relevant_context: list[str],
channel: str = "",
) -> list[dict]:
"""
Build an LLM messages array with memory enrichment AND multimodal media injection.
@@ -277,7 +304,8 @@ def build_messages_with_media(
gracefully rather than raising an error.
Structure (in order):
1. System message — agent identity, persona, AI transparency clause
1. System message — agent identity, persona, AI transparency clause,
and optional WhatsApp tier-2 scoping (when channel == "whatsapp")
2. System message — long-term context (ONLY if non-empty)
3. Sliding window messages — recent history
4. Current user message (plain string or multipart content list)
@@ -289,6 +317,8 @@ def build_messages_with_media(
Empty list produces the same output as build_messages_with_memory().
recent_messages: Short-term memory — list of {"role", "content"} dicts.
relevant_context: Long-term memory — list of content strings from pgvector.
channel: Channel name for tier-2 scoping (e.g. "whatsapp").
Default empty string means no channel-specific scoping.
Returns:
List of message dicts suitable for an OpenAI/LiteLLM-compatible API call.
@@ -301,6 +331,7 @@ def build_messages_with_media(
current_message=current_message,
recent_messages=recent_messages,
relevant_context=relevant_context,
channel=channel,
)
# If no media attachments, return the base messages unchanged

277
packages/orchestrator/orchestrator/tasks.py
View File

@@ -35,6 +35,23 @@ Pending tool confirmation:
On the next user message, if a pending confirmation exists:
- "yes" → execute the pending tool and continue
- "no" / anything else → cancel and inform the user
Escalation pipeline (Phase 2 Plan 06):
Pre-check (before LLM call):
If Redis escalation_status_key == "escalated", return early assistant-mode reply.
This prevents the LLM from being called when a human has already taken over.
Post-check (after LLM response):
check_escalation_rules() evaluates configured rules against conversation metadata.
If a rule matches AND agent.escalation_assignee is set, escalate_to_human() is
called and its return value replaces the LLM response.
Outbound routing (Phase 2 Plan 06):
All response delivery goes through _send_response() which routes to:
- Slack: _update_slack_placeholder() via chat.update
- WhatsApp: send_whatsapp_message() via Meta Cloud API
handle_message now pops WhatsApp extras (phone_number_id, bot_token) and
passes them through to _process_message via the extras dict.
"""
from __future__ import annotations
@@ -43,10 +60,25 @@ import asyncio
import json
import logging
import uuid
from typing import Any
import redis.asyncio as aioredis
from gateway.channels.whatsapp import send_whatsapp_message
from orchestrator.agents.builder import build_messages_with_memory
from orchestrator.agents.runner import run_agent
from orchestrator.audit.logger import AuditLogger
from orchestrator.escalation.handler import check_escalation_rules, escalate_to_human
from orchestrator.main import app
from orchestrator.memory.embedder import embed_text
from orchestrator.memory.long_term import retrieve_relevant
from orchestrator.memory.short_term import append_message, get_recent_messages
from orchestrator.tools.registry import get_tools_for_agent
from shared.config import settings
from shared.db import async_session_factory, engine
from shared.models.message import KonstructMessage
from shared.redis_keys import escalation_status_key
from shared.rls import configure_rls_hook, current_tenant_id
logger = logging.getLogger(__name__)
@@ -152,23 +184,28 @@ def handle_message(self, message_data: dict) -> dict: # type: ignore[no-untyped
by the Channel Gateway after tenant resolution completes.
The ``message_data`` dict MAY contain extra keys beyond KonstructMessage
fields. Specifically, the Slack handler injects:
- ``placeholder_ts``: Slack message timestamp of the "Thinking..." placeholder
- ``channel_id``: Slack channel ID where the response should be posted
fields. Specifically:
- Slack handler injects:
``placeholder_ts``: Slack message timestamp of the "Thinking..." placeholder
``channel_id``: Slack channel ID where the response should be posted
- WhatsApp gateway injects:
``phone_number_id``: WhatsApp phone number ID for outbound messaging
``bot_token``: WhatsApp access_token (injected as bot_token by gateway)
These are extracted before KonstructMessage validation and used to update
the placeholder with the real LLM response via chat.update.
These are extracted before KonstructMessage validation and used to route
outbound responses via _send_response().
Pipeline:
1. Extract Slack reply metadata (placeholder_ts, channel_id) if present
1. Extract channel reply metadata (Slack + WhatsApp extras) if present
2. Deserialize message_data -> KonstructMessage
3. Run async agent pipeline via asyncio.run()
4. If Slack metadata present: call chat.update to replace placeholder
5. Return response dict
3. Extract wa_id from sender.user_id for WhatsApp messages
4. Build extras dict for channel-aware outbound routing
5. Run async agent pipeline via asyncio.run()
6. Return response dict
Args:
message_data: JSON-serializable dict. Must contain KonstructMessage
fields plus optional ``placeholder_ts`` and ``channel_id``.
fields plus optional channel-specific extras.
Returns:
Dict with keys:
@@ -181,20 +218,39 @@ def handle_message(self, message_data: dict) -> dict: # type: ignore[no-untyped
placeholder_ts: str = message_data.pop("placeholder_ts", "") or ""
channel_id: str = message_data.pop("channel_id", "") or ""
# Extract WhatsApp-specific extras before model validation
# The WhatsApp gateway injects these alongside the normalized KonstructMessage fields
phone_number_id: str = message_data.pop("phone_number_id", "") or ""
bot_token: str = message_data.pop("bot_token", "") or ""
try:
msg = KonstructMessage.model_validate(message_data)
except Exception as exc:
logger.exception("Failed to deserialize KonstructMessage: %s", message_data)
raise self.retry(exc=exc)
result = asyncio.run(_process_message(msg, placeholder_ts=placeholder_ts, channel_id=channel_id))
# Extract wa_id from sender.user_id — WhatsApp normalizer sets sender.user_id
# to the wa_id (recipient phone number). This must happen AFTER model_validate.
wa_id: str = ""
if msg.channel == "whatsapp" and msg.sender and msg.sender.user_id:
wa_id = msg.sender.user_id
# Build the unified extras dict for channel-aware outbound routing
extras: dict[str, Any] = {
"placeholder_ts": placeholder_ts,
"channel_id": channel_id,
"phone_number_id": phone_number_id,
"bot_token": bot_token,
"wa_id": wa_id,
}
result = asyncio.run(_process_message(msg, extras=extras))
return result
async def _process_message(
msg: KonstructMessage,
placeholder_ts: str = "",
channel_id: str = "",
extras: dict[str, Any] | None = None,
) -> dict:
"""
Async agent pipeline — load agent config, build memory-enriched prompt, call LLM pool.
@@ -215,24 +271,26 @@ async def _process_message(
- Tool-call loop runs inside run_agent() — no separate Celery tasks
- If run_agent returns a confirmation message: store pending action in Redis
Escalation pipeline (Phase 2 Plan 06 additions):
Pre-check: if Redis escalation_status_key == "escalated", return assistant-mode reply
Post-check: check_escalation_rules after LLM response; if triggered, escalate_to_human
Outbound routing (Phase 2 Plan 06 additions):
All response delivery goes through _send_response() — never direct _update_slack_placeholder.
extras dict is passed through to _send_response for channel-aware routing.
Args:
msg: The deserialized KonstructMessage.
placeholder_ts: Slack message timestamp of the "Thinking..." placeholder.
channel_id: Slack channel ID for the chat.update call.
msg: The deserialized KonstructMessage.
extras: Channel-specific routing metadata. For Slack: placeholder_ts, channel_id,
bot_token. For WhatsApp: phone_number_id, bot_token, wa_id.
Returns:
Dict with message_id, response, and tenant_id.
"""
from orchestrator.agents.builder import build_messages_with_memory
from orchestrator.agents.runner import run_agent
from orchestrator.audit.logger import AuditLogger
from orchestrator.memory.embedder import embed_text
from orchestrator.memory.long_term import retrieve_relevant
from orchestrator.memory.short_term import append_message, get_recent_messages
from orchestrator.tools.registry import get_tools_for_agent
from shared.db import async_session_factory, engine
from shared.models.tenant import Agent
from shared.rls import configure_rls_hook, current_tenant_id
if extras is None:
extras = {}
if msg.tenant_id is None:
logger.warning("Message %s has no tenant_id — cannot process", msg.id)
@@ -265,8 +323,10 @@ async def _process_message(
result = await session.execute(stmt)
agent = result.scalars().first()
# Load the bot token for this tenant from channel_connections config
if agent is not None and placeholder_ts and channel_id:
# Load the Slack bot token for this tenant from channel_connections config.
# This is needed for escalation DM delivery even on WhatsApp messages —
# the escalation handler always sends via Slack DM to the assignee.
if agent is not None and (extras.get("placeholder_ts") and extras.get("channel_id")):
from shared.models.tenant import ChannelConnection, ChannelTypeEnum
conn_stmt = (
@@ -290,13 +350,9 @@ async def _process_message(
msg.id,
)
no_agent_response = "No active agent is configured for your workspace. Please contact your administrator."
if placeholder_ts and channel_id:
await _update_slack_placeholder(
bot_token=slack_bot_token,
channel_id=channel_id,
placeholder_ts=placeholder_ts,
text=no_agent_response,
)
# Build response_extras for channel-aware delivery
response_extras = _build_response_extras(msg.channel, extras, slack_bot_token)
await _send_response(msg.channel, no_agent_response, response_extras)
return {
"message_id": msg.id,
"response": no_agent_response,
@@ -318,14 +374,30 @@ async def _process_message(
# -------------------------------------------------------------------------
audit_logger = AuditLogger(session_factory=async_session_factory)
# Build response_extras dict used for all outbound delivery in this pipeline run.
# For Slack: merges DB-loaded slack_bot_token with incoming extras.
# For WhatsApp: extras already contain phone_number_id, bot_token, wa_id.
response_extras = _build_response_extras(msg.channel, extras, slack_bot_token)
# -------------------------------------------------------------------------
# Escalation pre-check — if conversation already escalated, reply in assistant mode
# -------------------------------------------------------------------------
thread_key = msg.thread_id or user_id
esc_key = escalation_status_key(msg.tenant_id, thread_key)
pre_check_redis = aioredis.from_url(settings.redis_url)
try:
esc_status = await pre_check_redis.get(esc_key)
finally:
await pre_check_redis.aclose()
if esc_status == b"escalated":
assistant_reply = "I've already connected you with a team member. They'll continue assisting you."
await _send_response(msg.channel, assistant_reply, response_extras)
return {"message_id": msg.id, "response": assistant_reply, "tenant_id": msg.tenant_id}
# -------------------------------------------------------------------------
# Pending tool confirmation check
# -------------------------------------------------------------------------
import redis.asyncio as aioredis
from shared.config import settings
redis_client = aioredis.from_url(settings.redis_url)
pending_confirm_key = _PENDING_TOOL_KEY.format(
tenant_id=msg.tenant_id,
user_id=user_id,
@@ -334,6 +406,7 @@ async def _process_message(
response_text: str = ""
handled_as_confirmation = False
redis_client = aioredis.from_url(settings.redis_url)
try:
pending_raw = await redis_client.get(pending_confirm_key)
@@ -362,13 +435,7 @@ async def _process_message(
await redis_client.aclose()
if handled_as_confirmation:
if placeholder_ts and channel_id:
await _update_slack_placeholder(
bot_token=slack_bot_token,
channel_id=channel_id,
placeholder_ts=placeholder_ts,
text=response_text,
)
await _send_response(msg.channel, response_text, response_extras)
return {
"message_id": msg.id,
"response": response_text,
@@ -412,6 +479,7 @@ async def _process_message(
current_message=user_text,
recent_messages=recent_messages,
relevant_context=relevant_context,
channel=str(msg.channel) if msg.channel else "",
)
# Build tool registry for this agent
@@ -428,6 +496,37 @@ async def _process_message(
tool_registry=tool_registry if tool_registry else None,
)
# -------------------------------------------------------------------------
# Escalation post-check — evaluate rules against conversation metadata
# -------------------------------------------------------------------------
conversation_metadata = _build_conversation_metadata(recent_messages, user_text)
triggered_rule = check_escalation_rules(
agent=agent,
message_text=user_text,
conversation_metadata=conversation_metadata,
natural_lang_enabled=getattr(agent, "natural_language_escalation", False),
)
if triggered_rule and getattr(agent, "escalation_assignee", None):
escalation_redis = aioredis.from_url(settings.redis_url)
try:
response_text = await escalate_to_human(
tenant_id=msg.tenant_id,
agent=agent,
thread_id=thread_key,
trigger_reason=triggered_rule.get("condition", "rule triggered"),
recent_messages=recent_messages,
assignee_slack_user_id=agent.escalation_assignee,
bot_token=slack_bot_token,
redis=escalation_redis,
audit_logger=audit_logger,
user_id=user_id,
agent_id=agent_id_str,
)
finally:
await escalation_redis.aclose()
# Check if the response is a tool confirmation request
# The confirmation message template starts with a specific prefix
is_confirmation_request = response_text.startswith("This action requires your approval")
@@ -453,14 +552,8 @@ async def _process_message(
len(relevant_context),
)
# Replace the "Thinking..." placeholder with the real response
if placeholder_ts and channel_id:
await _update_slack_placeholder(
bot_token=slack_bot_token,
channel_id=channel_id,
placeholder_ts=placeholder_ts,
text=response_text,
)
# Send response via channel-aware routing
await _send_response(msg.channel, response_text, response_extras)
# -------------------------------------------------------------------------
# Memory persistence (after LLM response)
@@ -489,6 +582,74 @@ async def _process_message(
}
def _build_response_extras(
channel: Any,
extras: dict[str, Any],
slack_bot_token: str,
) -> dict[str, Any]:
"""
Build the response_extras dict for channel-aware outbound delivery.
For Slack: injects slack_bot_token into extras["bot_token"] so _send_response
can use it for chat.update calls.
For WhatsApp: extras already contain phone_number_id, bot_token (access_token),
and wa_id — no transformation needed.
Args:
channel: Channel name from KonstructMessage.channel.
extras: Incoming extras from handle_message.
slack_bot_token: Bot token loaded from DB channel_connections.
Returns:
Dict suitable for passing to _send_response.
"""
channel_str = str(channel) if channel else ""
if channel_str == "slack":
return {
"bot_token": slack_bot_token,
"channel_id": extras.get("channel_id", "") or "",
"placeholder_ts": extras.get("placeholder_ts", "") or "",
}
elif channel_str == "whatsapp":
return {
"phone_number_id": extras.get("phone_number_id", "") or "",
"bot_token": extras.get("bot_token", "") or "",
"wa_id": extras.get("wa_id", "") or "",
}
else:
return dict(extras)
def _build_conversation_metadata(
recent_messages: list[dict],
current_text: str,
) -> dict[str, Any]:
"""
Build conversation metadata dict for escalation rule evaluation.
Scans recent messages and current_text for billing keywords and counts occurrences.
Returns a dict with:
- "billing_dispute" (bool): True if any billing keyword found in any message
- "attempts" (int): count of messages containing billing keywords
This is the v1 keyword-based metadata detection (per STATE.md decisions).
Args:
recent_messages: List of {"role", "content"} dicts from Redis sliding window.
current_text: The user's current message text.
Returns:
Dict with billing_dispute and attempts keys.
"""
billing_keywords = {"billing", "invoice", "charge", "refund", "payment", "subscription"}
all_texts = [m.get("content", "") for m in recent_messages] + [current_text]
billing_count = sum(1 for t in all_texts if any(kw in t.lower() for kw in billing_keywords))
return {
"billing_dispute": billing_count > 0,
"attempts": billing_count,
}
async def _execute_pending_tool(
pending_data: dict,
tenant_uuid: uuid.UUID,
@@ -526,7 +687,7 @@ def _extract_tool_name_from_confirmation(confirmation_message: str) -> str:
async def _send_response(
channel: str,
channel: Any,
text: str,
extras: dict,
) -> None:
@@ -545,7 +706,9 @@ async def _send_response(
For Slack: ``bot_token``, ``channel_id``, ``placeholder_ts``
For WhatsApp: ``phone_number_id``, ``bot_token`` (access_token), ``wa_id``
"""
if channel == "slack":
channel_str = str(channel) if channel else ""
if channel_str == "slack":
bot_token: str = extras.get("bot_token", "") or ""
channel_id: str = extras.get("channel_id", "") or ""
placeholder_ts: str = extras.get("placeholder_ts", "") or ""
@@ -563,7 +726,7 @@ async def _send_response(
text=text,
)
elif channel == "whatsapp":
elif channel_str == "whatsapp":
phone_number_id: str = extras.get("phone_number_id", "") or ""
access_token: str = extras.get("bot_token", "") or ""
wa_id: str = extras.get("wa_id", "") or ""