Talk Over the Bot: Steering an LLM Mid-Reply

Chat on a phone is not turn-taking. Nobody composes one complete message, sends it, and sits still until the reply lands. They fire off a thought, remember a detail, add it. They attach a photo and the caption arrives as a separate event. They read the first line of the answer coming in and type “no, in Hindi” before the second line is out.

Every LLM API, on the other hand, is a request and a response. You send a prompt, you get a stream, and the only thing you can do to a stream you don’t like is close the socket. Bridging those two worlds is the job of the bot runner in gooey-server, and until last month it did the naive thing: one inbound event, one run, one reply. Send three messages in five seconds and you got three overlapping runs, three partial answers, and whichever finished last won.

What it looks like now

Three things happen when a WhatsApp event arrives on a thread that is already busy:

1. If the model hasn’t said anything yet, the new message is merged into the pending run’s inputs. Text is concatenated, images and documents are appended, the newest audio wins. A photo album that arrives as four webhook calls becomes one prompt with four attachments instead of four prompts with one.
2. If the model is mid-sentence, the run is cancelled and its partial output is preserved as a real assistant turn in the conversation history. The replacement run starts with that turn plus the new message, so the model reads it as “you were saying X, the user interjected with Y”.
3. If the event is a duplicate delivery of a message we’ve already started a run for, it’s dropped on the floor. Meta retries webhooks aggressively and used to earn a second reply for its trouble.

Why this works for every model

None of this touches a provider SDK. OpenAI, Anthropic, Gemini, Llama via Fireworks, Sarvam, the self-hosted models on our own GPUs: they all arrive at the same run loop with the samemessages list, and they all stream back through the same state-saving step. The steer is expressed entirely in that shared layer, as a cancelled row and a rewritten history.

That’s the whole trick. A provider can’t be told to “change course”, but every provider will happily continue a transcript in which the assistant was cut off. The merge function is short enough to read in full:

def _merge_run_inputs(previous: dict, current: dict) -> dict:
    merged = previous | current
    previous_reply = (previous.get("raw_output_text") or [""])[0]

    if previous_reply:
        # the model already said something: keep it as a turn, so the
        # new message reads as a steer rather than a fresh question
        merged["messages"] = (current.get("messages") or []) + [
            format_chat_entry(
                role=CHATML_ROLE_USER,
                content_text=previous.get("raw_input_text") or "",
                input_images=previous.get("input_images"),
                input_documents=previous.get("input_documents"),
            ),
            format_chat_entry(role=CHATML_ROLE_ASSISTANT, content_text=previous_reply),
        ]
        return merged

    # nothing said yet: fold the inputs together into one prompt
    merged["input_prompt"] = "\n".join(
        filter(None, [previous.get("input_prompt"), current.get("input_prompt")])
    )
    for field in ("input_images", "input_documents"):
        merged[field] = (previous.get(field) or []) + (current.get(field) or []) or None
    merged["input_audio"] = current.get("input_audio") or previous.get("input_audio")
    return merged

The part that took three tries

Merging is easy. Deciding who gets to merge is the hard part, because the two events are being handled by two different web workers at the same time, and the thing they’re fighting over is a Celery task on a third machine.

The first version took a per-thread lock in Redis. It was correct on paper and wrong in practice: the lock lived in a different system from the rows it was guarding, and the “is there an active run?” check read a JSON blob that never actually holds the run status. Batching silently never fired. The fix was to stop inventing a lock and use the one Postgres already gives you:

with transaction.atomic():
    thread = MessageThread.objects.select_for_update().get(pk=thread.pk)

    # duplicate webhook delivery for a message that already has a run
    if SavedRun.objects.filter(
        platform=bot.platform, user_message_id=bot.user_msg_id
    ).exists():
        return None

    body = _cancel_active_run_and_merge_inputs(bot, thread, body)

    # create the run while still holding the lock, so thread.last_run
    # already points at it when the next event takes the lock
    page, sr = create_new_run(..., message_thread=thread, request_body=body)

# dispatch only after commit: a Celery worker on another connection
# can't see a row that hasn't been committed yet
result = page.call_runner_task(sr)

SELECT … FOR UPDATE on the thread row serialises the two events. The second one blocks until the first has committed its run, then sees that run as last_run, cancels it, and inherits its inputs. Creating the run inside the lock and dispatching the task after commit closes the two races that the Redis version had opened.

How the cancel actually reaches the model

Flipping is_cancelled on the superseded run is the only signal anyone sends. A post_save hook revokes the Celery task with a SIGUSR1, which surfaces inside the worker as a soft time limit. Independently, the runner re-reads the flag on every streaming step, right after it has persisted whatever the model has produced so far:

def save_on_step(yield_val=None, *, done=False):
    sr.refresh_from_db(fields=["is_cancelled"])
    ...
    gui.realtime_push(channel, output)   # live update to whoever is watching
    page.dump_state_to_sr(saved_state, sr)  # partial output lands in the db

    if not done and sr.is_cancelled:
        raise SoftTimeLimitExceeded

Because the partial output is written to the row before the check, the replacement run can read it back and hand it to the model as history. The WhatsApp handler streaming that run notices the same flag, stops editing the outgoing message, and writes the user turn plus the truncated reply into the conversation store, marked to skip the usual post-reply analysis since the sentence was cut off.

Bugs we found on the way

• The turn survived exactly one hop. The merge handed the cancelled run’s turn to its replacement, but the run after that rebuilt history from the database, where nothing had been saved. The user’s own message vanished along with the reply. Fix: the cancelled run’s handler saves the pair itself when it observes the cancellation.
• Raw vs. display text. Runs keep both the raw model output and the post-processed version shown to the user. The first partial-save collapsed them into one field, so translated replies were stored in the wrong language. Each now lands in its own field.
• A swallowed IntegrityError poisons the transaction. The duplicate-message guard used to catch the unique-constraint error and carry on. Inside atomic() that leaves the connection in an aborted state and every later query fails. The check is now an explicit exists() before the insert.
• Title generation raced the commit. The conversation title task was dispatched while the run row was still uncommitted, so the worker couldn’t find it. It now runs from transaction.on_commit.
• Albums are “unsupported”. WhatsApp delivers a multi-photo album as the individual images plus one callback of type unsupported, which we used to answer with an error message. It’s now ignored.

Scope, honestly

This is live for WhatsApp today, which is where almost all of the overlapping traffic comes from: album uploads, voice notes followed by text, and people who type the way they talk. Slack, Telegram, Messenger, Instagram, Twilio voice and the web widget go through the same runner and the same run-state machinery, so extending the row lock and merge to them is a gating change rather than a redesign. They keep the old one-event-one-run behaviour until then.

If you build bots on Gooey.AI, there’s nothing to switch on. Pick any model in the workflow, deploy it to WhatsApp, and talk over it.