Add, move or change a guard in the inbound-message dispatch chain (app/services/dispatch_guards.py GUARD_CHAIN) or a step in the customer pipeline it falls through to. Use whenever a task adds a new keyword, a new UserStates holding state, a new pre-gate, or changes which branch answers a message first.
Install
npx skillscat add netanel152/proli/dispatch-guard Install via the SkillsCat registry.
Adding or moving a dispatch guard
The chain is data: GUARD_CHAIN in app/services/dispatch_guards.py is an ordered tuple
of (name, guard) pairs, and its order is the product — PRO-121's emergency hoist,
for one, is defined by what it runs after and before. Read the module docstring first; it
says why a tuple and not a state-keyed dict. Nothing below is a value to trust over the
code: every name, count and position is owned by dispatch_guards.py and the tests that
pin it.
What a guard is
async def guard_<name>(ctx: DispatchContext, deps: GuardDeps):
if <not my case>:
return None # fall through to the next guard
... # answer via deps.whatsapp / deps.state_manager
return HANDLED # dispatch stops hereDispatchContext(chat_id,user_text,normalized_text,media_url,is_emergency_detected,current_state,is_exempt,emergency_inbound_logged) is the
shared mutable state. A guard that clears state re-readsctx.current_statefor the
guards after it — see how the zero-touch and reset guards do it.GuardDepscarries the injected collaborators (whatsapp,state_manager,context_manager,users_collection,security,settings). Use them, never module
globals, so the guard is testable against fakes without monkeypatching.- Helpers that still live in
workflow_serviceare resolved per call through the module
(from app.services import workflow_service as wf; wf._handle_status_query(...)), never
imported at module top — late monkeypatching in the suite depends on it. - Return exactly
HANDLEDorNone.run_guard_chainraises on anything else.
Procedure
- Decide the position by what must outrank it. Write down which existing guards may
not be bypassed by the new one (a live human handoff, consent, the rate limiter, the
admin wizard usually outrank everything) and which holding state would swallow its
trigger if it ran later. That sentence becomes the guard's docstring. - Define the function directly above the guard that follows it in the chain.
tests/test_agent_pack_drift.py::test_guard_chain_runs_in_source_definition_order
pinsGUARD_CHAIN's order to source definition order, so chain order and file order
must agree. - Insert the
(name, guard)pair at that position inGUARD_CHAIN. - Update the three order pins, in the same commit:
tests/test_dispatch_guards.py::test_guard_chain_order_is_pinned— the full list of
names; plus any relative-position test that names your neighbours
(test_pro_bypass_runs_immediately_before_pro_mode_routing, the emergency-hoist
position test, the holding-states membership test).tests/test_agent_pack_drift.py::_DISPATCH_SEQUENCE— a(bold label, unique source anchor)pair for the new branch; the anchor must occur exactly once acrossdispatch_guards.py+workflow_service.py+conversation_pipeline.py..claude/agents/flow-tracer.md## Dispatch Order— a numbered line whose bold label
matches the sequence entry exactly, and the entry count in the intro paragraph.
- Write the guard's own tests in
tests/test_dispatch_guards.pyagainst adeps
fixture built from fakes: the case it handles (returnsHANDLED, sends the rightMessages.*constant), the case it must ignore (returnsNone, touches nothing), and
the state it must not break (a message from PRO_MODE, from an ADMIN_ state). - If it introduces a holding state, give it a TTL in
WorkerConstantswith a comment
saying what expiry means for the customer (seeCANCEL_CONFIRM_TTL_SECONDS), add the
state toUserStates, and updateflow-tracer.md's## UserStateslist — drift tests
fail otherwise. - Any message it sends follows the
whatsapp-copyskill; any keyword it accepts is aMessages.Keywordslist, never a literal.
Run before committing
pytest tests/test_dispatch_guards.py tests/test_agent_pack_drift.py tests/e2e/test_e2e_state_matrix.py -qThe PRO-83 state × input matrix (tests/e2e/test_e2e_state_matrix.py) is the wide net:
a new guard that steals a message from an existing state shows up there as a changed cell.