Update a roadside pickup location without sending two jobs

A driver asks for roadside assistance, then corrects the pickup location. The assistant must use the new location and check whether a job was already sent.

In this fictional example, you will inspect the corrected location, the question the driver answered and the dispatch record. Those checks distinguish a fresh customer answer from an earlier confirmation.

{"fresh_memory": true, "fresh_question": false, "jobs_sent": 0, "reason": "wrong_incident_location", "status": "blocked"}

This is real offline Python output from 8 October 2026. The draft says exit 16; the caller answered a question about exit 14. Updating memory did not make that answer apply to the corrected location. The guard sent no job.

Build that distinction into the function that sends the request. Then treat a repeat of an already sent draft as a receipt lookup. Otherwise a retry can become a second dispatch, or ask for approval of work already done.

Bind the question to the revised draft

git clone https://github.com/RasaHQ/rasa-community-resources.git
cd rasa-community-resources
git checkout 4aa0c4419dc193fef7a969c12d59edcf720f2606
cd examples/mantle-voice-insurance-roadside-claude

The full clone also keeps the neighbouring case contract used by a unit test. Stay in this example directory for every command below. Python’s standard library is enough for the offline path; it needs no licence, API key or package install.

The recorded offline suite completed without skipped tests:

Ran 30 tests in 0.004s

OK

Your timing can differ. The tests include stale confirmation, a completed request replay and a declined provider retry. See the pinned guard tests. Their output checks Python behaviour, rather than whether a model chooses the right tool.

The project separates dialogue from dispatch. Its Rasa skill calls a wrapper; the wrapper passes memory and conversation events to the dispatch function. After a draft change, it writes the current draft values back to memory. The wrapper also sends outcome wording through the tool context.

These are exact excerpts from skills/roadside_dispatch/tools.py:

def _memory(context: Optional[ToolContext]) -> dict:
    return {key: (context.memory.get(key) if context is not None else None) or None for key in hc.MEMORY_KEYS}


def _conversation(context: Optional[ToolContext]) -> hc.Conversation:
    return hc.conversation_from_events(context.events) if context is not None else hc.Conversation()


def _write_draft(context: Optional[ToolContext], service: hc.RoadsideService, draft: Optional[hc.Draft]) -> None:
    if context is None or draft is None:
        return
    for key, value in hc.memory_values(service, draft).items():
        context.memory.set(key, value)


async def _send_receipt(context: Optional[ToolContext], tool_name: str, result: dict) -> None:
    text = hc.customer_receipt(tool_name, result) if hc.TOOL_SENDS_RECEIPT else None
    if text and context is not None:
        await context.send(text)

The dispatch wrapper then calls the guard:

    result = hc.request_dispatch(_service(), draft_id, _memory(context), _conversation(context))
    await _send_receipt(context, "request_dispatch", result)
    return ToolResult(llm_response=result)

Caller words enter the draft through the start or update tool. The guard resolves a fixture place and checks provider suitability there. The model does not supply the provider’s acceptance or arrival estimate.

The pinned skill.md declares the confirmation question on the request tool:

tool_constraints:
  - request_dispatch:
      requires: session.roadside_dispatch.pending_draft_ref
      requires_confirmation:
        enabled: true
        utter_for_confirmation: utter_confirm_location

This is project configuration, not a fresh engine test. Read the skill alongside responses.yml: the question reads back the vehicle, location and service from draft memory. The skill asks the engine to handle confirmation. A correction calls the update tool before another request.

In lib/roadside.py, draft updates increment the version. The execution check requires the current tag and the current location and service in an answered question:

def _location_confirmed(service: RoadsideService, draft: Draft, memory: dict, conversation: Conversation) -> bool:
    labels = service.labels(draft)
    question = conversation.confirmation_question or ""
    return (
        (memory.get("pending_draft_ref") or "") == draft.tag
        and labels["location"] in question
        and labels["service"] in question
        and conversation.confirmation_answered
    )

The cost of this choice is carrying draft identity through both memory and event handling. A bare yes/no value is insufficient. Keep the location text that was read back, rather than replacing it with the latest location when constructing evidence.

Cross the old question with new memory

The confirmation and replay rows start with an exit-14 draft, then change it to exit 16. The provider-retry rows use a separate incident at Kestrel Plaza.

Run the exercise below from this example directory. It adapts the recorded experiment to your working directory; no private script is required. It renders the question from the real response file. Setting confirmation_answered=True is deliberate fixture evidence, not a test of spoken consent interpretation.

python3 - <<'PY'
"""Offline exercise: run from the selected example directory."""
import json
import sys
from pathlib import Path
sys.dont_write_bytecode = True
project = Path.cwd()
from lib import roadside as hc

def question(svc, draft):
    template = (project / 'skills/roadside_dispatch/responses.yml').read_text()
    text = ' '.join(template.split('- text: >', 1)[1].split())
    for key, value in hc.memory_values(svc, draft).items():
        text = text.replace('{' + key + '}', value)
    return hc.Conversation(confirmation_question=text, confirmation_answered=True)

for fresh_memory in (False, True):
    for fresh_question in (False, True):
        svc = hc.RoadsideService('independent-revision-matrix')
        _, draft = hc.start_dispatch_draft(svc, 'HC-AU-517306', 'VEH-5173-1', 'tow', 'exit 14')
        old_memory, old_question = hc.memory_values(svc, draft), question(svc, draft)
        hc.update_dispatch_draft(svc, draft.draft_id, location='exit 16')
        result = hc.request_dispatch(svc, draft.draft_id,
                                     hc.memory_values(svc, draft) if fresh_memory else old_memory,
                                     question(svc, draft) if fresh_question else old_question)
        print(json.dumps({'fresh_memory': fresh_memory, 'fresh_question': fresh_question,
                          'status': result['status'], 'reason': result['reason'],
                          'jobs_sent': len(svc.dispatches)}, sort_keys=True))
        assert len(svc.dispatches) == int(fresh_memory and fresh_question)
        if fresh_memory and fresh_question:
            replay = hc.request_dispatch(svc, draft.draft_id, old_memory, old_question)
            assert replay['replay'] is True and len(svc.dispatches) == 1
            print(json.dumps({'retry_with_old_evidence': replay['status'], 'replay': replay['replay'],
                              'jobs_sent': len(svc.dispatches), 'location_ref': replay['location_ref']}, sort_keys=True))

svc = hc.RoadsideService('independent-provider-retry')
_, draft = hc.start_dispatch_draft(svc, 'HC-AU-440218', 'VEH-4402-1', 'tow', 'Kestrel Plaza')
first = hc.request_dispatch(svc, draft.draft_id, hc.memory_values(svc, draft), question(svc, draft))
second = hc.request_next_provider(svc, first['assistance_ref'])
third = hc.request_next_provider(svc, first['assistance_ref'])
assert first['status'] == 'pending' and second['status'] == 'succeeded'
assert first['assistance_ref'] == second['assistance_ref'] == third['assistance_ref']
assert len(svc.dispatches) == 2 and third['replay']
print(json.dumps({'initial_acceptance': first['provider_acceptance'], 'next_acceptance': second['provider_acceptance'],
                  'same_assistance_ref': first['assistance_ref'] == second['assistance_ref'],
                  'repeat_next_provider_replay': third['replay'], 'jobs_sent': len(svc.dispatches),
                  'places': [job['place_ref'] for job in svc.dispatches]}, sort_keys=True))
PY

Recorded stdout from the same exercise logic:

{"fresh_memory": false, "fresh_question": false, "jobs_sent": 0, "reason": "wrong_incident_location", "status": "blocked"}
{"fresh_memory": false, "fresh_question": true, "jobs_sent": 0, "reason": "wrong_incident_location", "status": "blocked"}
{"fresh_memory": true, "fresh_question": false, "jobs_sent": 0, "reason": "wrong_incident_location", "status": "blocked"}
{"fresh_memory": true, "fresh_question": true, "jobs_sent": 1, "reason": "verified_fixture_receipt", "status": "succeeded"}
{"jobs_sent": 1, "location_ref": "R12-N-EXIT16", "replay": true, "retry_with_old_evidence": "succeeded"}
{"initial_acceptance": "declined", "jobs_sent": 2, "next_acceptance": "accepted", "places": ["KESTREL-PLAZA", "KESTREL-PLAZA"], "repeat_next_provider_replay": true, "same_assistance_ref": true}

Change just the corrected location back to exit 14 and predict which assertions will fail. The question checks location and service, while memory checks the version tag. Even unchanged display text cannot repair an old memory tag.

Replay a sent draft before considering another effect

The request function looks for an incident belonging to this draft before it checks confirmation:

    existing = next((i for i in service.incidents.values() if i["draft_id"] == draft.draft_id), None)
    if existing is not None:
        return _incident_result(service, existing, replay=True)

That order explains the fifth output line. Old evidence cannot create a new job, but it does not prevent retrieving the existing result. Check both the replay flag and the number of jobs. A success status alone cannot distinguish an original effect from a replay.

A correction after dispatch returns already_dispatched; it needs the desk route rather than a silent rewrite of an existing job. The replay mechanism shown here is local to the in-memory service instance. Durable storage and atomic request handling remain integration work before using a real provider. These receipts establish neither behaviour across restarts nor concurrent dispatch safety.

Retry a declined provider under one incident

The final exercise starts an electric-car tow at fictional Kestrel Plaza. The first suitable provider declines. request_next_provider uses the existing incident location and assistance reference, excluding declined providers from the next search. Once a provider accepts, another call replays the incident instead of sending a third job.

Keep request, acceptance and estimate separate in the caller’s answer. pending with a decline does not mean someone is coming. succeeded does not invent an estimate; only the provider fixture can supply one. The tests exercise pending, accepted and accepted-without-estimate receipts.

Show what to inspect when the result seems wrong

For a blocked first dispatch, inspect the draft tag, the exact question and whether the conversation records an answer. For a successful replay, inspect replay and the unchanged dispatch count. For a provider retry, compare the assistance reference and place across both jobs. Changing either means you are no longer retrying the same incident.

Check the project with the pinned SDK

A separate offline check on 8 October 2026 used Rasa 3.21.0.dev5. Its full suite passed without skips, and project validation loaded the tools and skills. The test summary was:

Ran 30 tests in 0.004s

OK

Validation also printed:

validate_project: ok

After installing this selected project, repeat those checks with:

uv run --locked python -m unittest discover -s tests -v
make validate

This adds evidence for SDK imports and project validation. It does not run training, a model or speech providers. The independent property exercise above still tests pure Python rather than engine confirmation or spoken behaviour.

Try the configured voice path separately

The public README and Makefile provide the live setup. With Python 3.12 and uv available, install only this selected example:

uv sync --locked --python 3.12
make env

Fill the local .env with RASA_LICENSE, ANTHROPIC_API_KEY and DEEPGRAM_API_KEY. The project pins Rasa Pro 3.21.0.dev5. Its integrations.yml selects Anthropic claude-sonnet-5-5, Deepgram flux-general-en input and aura-2-andromeda-en output, with the Inspector enabled.

make validate
make train
make inspect

Allow microphone access. This optional path uses billed provider APIs; it was not run for the offline evidence above. Use a short manual acceptance check: “I am James Whitfield, policy five one seven three zero six. I need a tow at exit fourteen.” At the location question, say “Actually exit sixteen.” These are proposed caller lines, not recorded dialogue. Inspect the correction tool result and the next confirmation before agreeing. Check that the eventual receipt names exit 16 and that a status check retains the assistance reference.

For the refusal path, leave confirmation unanswered and inspect whether any job was sent. Finish this small check before designing a wider voice evaluation. The quickstart covers the general Rasa setup.