agent-lab/docs/plans/todo-13-real-queue-websocket-loop.md

Todo 13 Real Queue WebSocket Loop Plan

Status: done

Goal

Make runtime queues real producer/consumer boundaries and let WebSocket run separate upstream and downstream tasks.

Why

The original queue requirement names producers and consumers explicitly:

• input queue: WebSocket upstream and EventAgent replies produce; ChatAgent consumes.
• output queue: ChatAgent produces; WebSocket downstream consumes.
• event queue: ChatAgent produces; EventAgent consumes.

The current runtime logs queue put/get calls, but DebugRuntime.run() still produces and consumes output in one method, and WebSocket reads only one initial request.

Scope

This todo covers:

• Add a runtime session entrypoint that starts ChatAgent/EventAgent work and returns RuntimeQueues.
• Keep DebugRuntime.run() as a compatibility wrapper for tests and scripts.
• Add an EventAgent queue consumer task.
• Preserve provider-valid history ordering when a user message arrives while a tool call is being handled.
• Split the WebSocket handler into downstream output sending and upstream user-message receiving tasks.

This todo does not cover:

• Multi-session persistence.
• Multiple simultaneous ChatAgent workers inside one debug session.
• UI controls for sending a second user message during an active run.
• Role validation and static asset packaging; those are separate todos.

Files

• Modify: src/agent_lab/application/queues.py
• Modify: src/agent_lab/application/runtime.py
• Modify: src/agent_lab/presentation/web.py
• Modify: tests/test_debug_runtime.py
• Modify: tests/test_websocket_api.py
• Do not modify unrelated UI styling.

Design

Add DebugRuntime.start(request) -> RuntimeQueues. It should create or reuse RuntimeQueues, seed the initial user message into queues.input, start background runtime work, and return queues immediately.

DebugRuntime.run(request) remains an async iterator by calling start() and then consuming queues.output until done or error.

Runtime work should:

• put output payloads onto queues.output only;
• put ChatAgent events onto queues.events;
• run an EventAgent worker that consumes queues.events and puts tool replies onto queues.input;
• when waiting for a tool reply, buffer unrelated input messages so the next provider history is assistant(tool_calls) -> tool -> later user.

WebSocket handling should:

• read the first JSON payload as DebugRunRequest;
• start the runtime session;
• run a downstream task that consumes queues.output and sends JSON;
• run an upstream task that accepts {"type": "user_message", "content": "..."} and puts ChatMessage(role="user", content=...) into queues.input.

Steps

1. Add failing runtime tests:
   • DebugRuntime.start() returns queues and output can be consumed directly.
   • User input enqueued during an event handoff is preserved after the matching tool reply.
2. Add a failing WebSocket test:
   • send initial request;
   • receive an event;
   • send {"type": "user_message", "content": "follow-up"};
   • assert the second ChatAgent call receives provider-valid ordering.
3. Refactor DebugRuntime to add start(), output queue consumption, EventAgent worker, and tool-reply wait logic.
4. Refactor WebSocket presentation code into separate upstream/downstream tasks for runtimes that expose queue sessions.

5. Run focused tests:

   uv run pytest tests/test_debug_runtime.py tests/test_websocket_api.py
   

6. Run the full suite:

   uv run pytest
   

7. Update docs/plans/todos.md and this plan with the result, then commit.

Verification

Result:

• Red run passed as expected: the new tests failed because DebugRuntime.start() did not exist, event-time user input was ordered before the tool reply, and WebSocket still called only run().
• uv run pytest tests/test_debug_runtime.py tests/test_websocket_api.py passed: 26 tests, 1 existing Starlette deprecation warning.
• uv run pytest passed: 31 tests, 1 existing Starlette deprecation warning.

Evaluation

• DebugRuntime.start() now starts a queue-backed session and returns RuntimeQueues for downstream consumers.
• DebugRuntime.run() remains as a compatibility wrapper over the output queue.
• EventAgent now consumes the event queue in a worker task and writes tool replies to the input queue.
• WebSocket sessions now split downstream output sending from upstream user_message receiving.
• User messages that arrive during a tool handoff are buffered until after the matching tool reply, preserving provider-valid message order.