Expose a public external-message injection API for live TUI sessions

[pritishd]

Problem Statement

I want to steer long-running pydantic-deep TUI sessions from an external collaboration surface without switching back to the terminal.

In our current prototype, that external surface is Slack: the agent can move a conversation into a Slack thread, and replies in that thread can steer or continue the live TUI session. The same pattern seems useful beyond Slack: Jira comments, Teams/Discord threads, n8n workflows, CI/build monitors, webhook services, or any human/event channel that should be able to send a follow-up into a running agent.

pydantic-deep already has the right internal behavior via MessageQueue: steering messages can be injected before the next model request, and follow-ups can be queued while the agent is running.

What would help is a documented extension point for client-owned integrations to submit a message into the active TUI session. In our prototype, we connected an external event source to the queue through TUI/app internals.

That works well as a proof of concept, but a small public injection hook would make this pattern easier to support without depending on private implementation details. The external integration would still own transport concerns like Slack/Jira/webhook polling, authentication, deduplication, and reactions; pydantic-deep would only provide the safe session injection point.

Proposed Solution

Expose a small public bridge API on the TUI/session that lets client-owned integrations submit external messages into the active agent session.

Rough API sketch:

  await app.inject_external_message(
      text="Please also check MR 123",
      source="slack",
      mode="auto",  # auto | steer | follow_up | start_if_idle
      metadata={
          "conversation_id": "C123:1784850025.868319",
          "message_id": "1784850124.802469",
      },
  )

Possible behavior:

• If an agent run is active:

  • mode="steer" injects before the next model request.
  • mode="follow_up" queues the message for when the current run would otherwise stop.
  • mode="auto" uses the same policy the TUI already uses internally.

• If the agent is idle:

  • mode="start_if_idle" or auto can start/resume a run with the external message.

• The submitted message should appear in the TUI transcript like a user message, ideally with source metadata available for logs/tracing.

An alternative shape would also work:

  bridge = app.external_message_bridge

  await bridge.submit(
      text="Please also check MR 123",
      source="jira",
      mode="auto",
      metadata={"issue": "PIPE-1234"},
  )

The key request is the stable public injection surface, not the exact naming.

Alternatives Considered

We built a Slack thread bridge outside pydantic-deep. It polls a Slack thread, reacts to user replies, and injects the reply into the running TUI session through MessageQueue.

This works well as a proof of concept, but it currently relies on private TUI details:

• accessing the app’s queue directly,
• detecting whether an agent run is active,
• deciding whether to call steer() or follow_up(),
• waking the idle TUI by reaching into screen internals.

Another option would be for pydantic-deep to implement Slack/Jira/etc. integrations directly, but I do not think that is the right layer. The client application should own the external transport and infrastructure. pydantic-deep only needs to provide the supported session injection API.

Additional Context

This is not a request for Slack support. Slack is only our current transport.

The broader use case is external human/event steering for long-running agents. For example:

• Continue a TUI coding session from a Slack thread.
• Let a Jira comment steer an agent working on that issue.
• Let an n8n workflow or webhook submit a follow-up.
• Let a CI/build monitor wake the agent when a service is ready or a test fails.
• Allow group collaboration around one agent session from a shared chat thread.

One useful side effect we observed: the external thread becomes another source of lightweight session context. After restarting a session, the agent can inspect the thread and recover what the user was trying to do, alongside pydantic-deep memory.

[DEENUU1]

Thanks for the detailed write-up — the use case is clear and I agree with your layering instinct: transport, auth, dedup and reactions belong to your app, not here. I don't want Slack code in this repo either.

Before we design a new API though, I want to lay out what already exists, because a good part of what you're describing is supported today and doesn't touch TUI internals at all. Then there's one question that decides the whole shape of the thing, and I'd rather ask it than guess.

What's already public and supported

MessageQueue is a documented public API, not a TUI detail: Message queue (steering and follow-up). It's designed exactly for "external code injects into a running agent":

from pydantic_deep import create_deep_agent, DeepAgentDeps, StateBackend
from pydantic_deep.features.message_queue import MessageQueue, run_with_queue

queue = MessageQueue()
agent = create_deep_agent(model="anthropic:claude-sonnet-4-6", message_queue=queue)
deps = DeepAgentDeps(backend=StateBackend(), message_queue=queue)   # same queue on both

async def slack_bridge() -> None:
    async for reply in poll_thread(...):          # your transport, your dedup
        if reply.urgent:
            await queue.steer(reply.text, metadata={"source": "slack", "ts": reply.ts})
        else:
            await queue.follow_up(reply.text, metadata={"source": "slack", "ts": reply.ts})

async with asyncio.TaskGroup() as tg:
    tg.create_task(slack_bridge())
    result = await run_with_queue(agent, "investigate the failing test", deps=deps, queue=queue)

The two halves of what you asked for are both there:

• steering — MessageQueueCapability injects queued steering into the message list right before each model request, so the agent sees it at the next LLM call without a cancel/restart.
• follow-up — run_with_queue is the loop: after each completed run it drains follow-ups and re-enters with them as the next prompt, repeating until the queue is empty. That's your start_if_idle, for a headless session.

And metadata is already on QueuedMessage, so {"source": "slack", "message_id": ...} round-trips today. There's precedent for it in-tree: the monitoring feature pushes process output into the same queue with metadata={"source": "monitor", ...} (pydantic_deep/features/monitoring/toolset.py:60), which is the same shape as your bridge — an external event source waking a running agent.

Also worth knowing: DeepAgentDeps.message_queue is shared with subagents by clone_for_subagent(), so an injected message reaches a session that's currently delegating, not just the top-level agent.

So: if what you need is an agent steerable from Slack, you can build that today on pydantic_deep with no changes from us, and none of it depends on apps/cli.

Where the TUI actually differs

What the library path doesn't give you is the interactive session itself — the transcript, approval modals, diffs, /fork, session resume. I read your issue as wanting to steer that, not a headless run, which is a fair thing to want.

Here's the constraint I keep hitting when I sketch it: app.inject_external_message(...) is a Python method, so it only helps if your bridge runs inside the TUI's own process and on its event loop. MessageQueue is asyncio-safe, not thread-safe (its lock binds to whichever loop first awaits it), and every visible effect — appending to the transcript, starting a run — has to happen on Textual's loop.

Which means the API only pays off if you already have a custom entrypoint constructing DeepApp and spawning your poller as a task next to it. And if you're doing that, you're an app author already, and the honest question is whether you want the TUI or just a session.

If instead your bridge is a separate process (a Slack bot on another host, an n8n webhook, a CI hook), then no Python method can help — that's IPC, and the right answer is a session endpoint on the TUI (unix socket or a loopback HTTP port) that any external process can post to. That's a genuinely useful feature and it would cover every example in your issue, including the CI/build monitor and the group-collaboration one. It's also a much bigger piece of work, and I'd want to be sure it's what's needed before starting it.

The middle option — a public in-process hook on apps/cli — is the one I'm most hesitant about. apps/ ships in the wheel, but it's deliberately outside the typecheck and coverage gates that pydantic_deep/ lives under, and chat.py is 2000+ lines that move regularly. Making a method there a supported, versioned surface is a real commitment, so I want it to be the right surface before I make it.

Questions

1. Where does your bridge run — same process as the TUI, or separate? If same: how do you launch it today, your own entrypoint around DeepApp, or a patch? If separate: how does it currently reach the queue at all?
2. Do you need the TUI specifically, or a session? Concretely: does the person in Slack care that there's a terminal with a transcript and approval modals on your machine, or would a headless agent that also posts its output back into the thread do the job? That answer decides library-vs-app for me.
3. Reliability expectations. When an injected message can't be delivered, what should happen — an error back to the caller so you can react in the thread, a retry, or silent drop? I ask because there are states today where a message goes nowhere (more on that below), and a bridge needs to know.
4. Concurrency. Is it one Slack thread ↔ one session, or can several people push into the same session at once? Group collaboration is in your issue, and if it's real then ordering and backpressure need an answer, not just injection.
5. Modals. If a message arrives while the TUI is showing an approval prompt or /model picker, what does your prototype do? I suspect "queue it as a follow-up" is right, but I'd rather hear it than assume.

Two things I'll fix regardless

Reading through this I found problems that hurt the queue whichever way we go, so they're worth doing on their own:

• A follow-up can be silently lost when a run ends. The stream worker drains follow-ups right after the run finishes (apps/cli/screens/chat.py:1251). Anything enqueued after that line still sees the run as active, so it's queued as a follow-up — but nothing drains it: the finally block only drains follow-ups when the run was cancelled (chat.py:1330), and the badge is then reset (chat.py:1342), so the message is both stranded and invisible until some later run happens to end. For a human typing that's a rare few-hundred-millisecond gap; for an event-driven bridge it's an "the agent ignored me" bug you'll hit for real. It needs a drain when the session goes idle.
• metadata is decorative. format_steering / format_follow_up (features/message_queue/capability.py:205-215) drop it, so source reaches neither the model nor the logs. Monitoring already sets it and gets nothing from it. At minimum it should land in logs and span attributes; whether the model should see [steering via slack] instead of [steering] is a separate call I'd want your opinion on, since the agent arguably should weigh a human and a CI monitor differently.

One thing I'd also add either way is a bound on the queue — it's unbounded today, and a retrying webhook or a chatty thread can pile up messages faster than the agent consumes them, each costing tokens when it lands.

Answer 1 and 2 and I think the right design falls out on its own. If it turns out you want a live TUI reachable from another process, say so plainly — that's the interesting version of this issue and I'd rather build that once than bolt an in-process hook on now and a socket on later.

[pritishd]

Thanks, this is very helpful. Yes, you read the issue correctly: the part we care about is steering the live TUI/session.

To answer your questions:

1. Where the bridge runs

Today our bridge runs in the same process as the TUI. We launch our own entrypoint around DeepApp, create the agent/deps with a shared MessageQueue, and start the external bridge as a background task from the app lifecycle. The bridge owns transport, auth, dedup, reactions/acks, etc., then injects into the queue.

2. TUI or session

We do need the TUI/session specifically for this workflow. The terminal session matters because it has the transcript, approvals, diffs, tool visibility, session resume, and the local operator view. A headless run_with_queue loop is useful, but it does not replace the live TUI for this use case.

3. Reliability

If an injected message cannot be accepted, I’d prefer an explicit failure back to the caller rather than silent drop, so the bridge can react in the external thread/system.

If the TUI is busy, queued is fine. If the TUI is idle, auto / start_if_idle should wake the session. If the TUI is in a modal such as approval or model picker, queuing as follow-up seems right. I would not expect the external message to dismiss or bypass the modal.

4. Concurrency

Our model is one external conversation/thread maps to one agent session, but several humans or systems can submit messages into that thread. The bridge should handle transport-level ordering and dedup by message timestamp/id. pydantic-deep just needs to accept messages in submitted order and apply normal queue/backpressure behavior.

5. Modals

If a message arrives while the TUI is showing an approval prompt, model picker, or similar modal, queuing it as a follow-up seems right. It should not dismiss or bypass the modal.

On metadata: I agree it should at least reach logs/spans. I also think the model should see a small source label, e.g. [follow-up via jira], [steering via slack], or [follow-up via monitor], because a human reply and a CI monitor event should be interpreted differently. The full metadata does not need to be injected into the model prompt, but source/type is useful context.

The two fixes you identified would help immediately regardless of the larger API:

• drain stranded follow-ups when the TUI goes idle,
• make queue metadata visible in logs/spans, and optionally as a compact source label to the model.

Your distinction between an in-process hook and a separate-process session endpoint is useful. Our current prototype is same-process because that was the fastest path, but the more generally useful feature may be a live TUI session endpoint. Long term, I can imagine the bridge being a separate process: Jira webhook, n8n workflow, CI monitor, chat bridge, etc. In that shape, a loopback HTTP port or local socket owned by the TUI seems cleaner than exposing apps/cli
internals.

After your explanation, my instinct is that a local session endpoint is the more durable general feature, while the queue fixes above would unblock most of our current prototype.

[OchnikBartek]

The requirements look complete to me — the title and the answers settle that this is about a live TUI session, not a headless run, and @pritishd said the exact API shape doesn't matter as long as the surface is stable and public.

What's left is a call for you rather than a question for the reporter, @DEENUU1: in-process method on the TUI, or a local session endpoint (socket / loopback HTTP)? You laid out the reasons for hesitating on a supported method in apps/cli, and the reporter's longer-term shape is a separate process anyway, so the endpoint reads like the answer to me — but it's your call on what this repo wants to maintain as a versioned surface. Tell me which and I'll pick it up.

One behavioural question worth settling either way: when draining stranded follow-ups, cancelled runs currently discard everything pending as likely-stale. Fine for a human hitting Esc, but it's the silent drop a bridge shouldn't get. Should externally-injected follow-ups survive a cancellation?

The rest — source-label format and sanitizing, the queue bound and its policy — I'll take from what's already in the thread and put up for review.