How Redis Manages Task State and Broadcasts WebSocket Messages in MathModelAgent

MathModelAgent uses Redis as an in-memory broker to persist task identifiers with TTL expiration and leverages Redis Pub/Sub channels to push real-time status updates to WebSocket clients.

The open-source MathModelAgent repository (jihe520/mathmodelagent) implements a decoupled, asynchronous architecture for mathematical modeling workflows. By combining Redis key-value storage for task state management with Redis Pub/Sub for message broadcasting, the system achieves scalable real-time communication between backend processing and frontend clients.

Storing Task State with TTL

When a new modeling request arrives at the /example or /modeling endpoints, the system immediately registers the task in Redis. The RedisManager.set method in backend/app/services/redis_manager.py (lines 33-38) creates a key using the pattern task_id:{task_id} and sets a 10-hour TTL to ensure automatic cleanup of stale entries.

In backend/app/routers/modeling_router.py, the implementation stores the task identifier after creating the working directory:

await redis_manager.set(f"task_id:{task_id}", task_id)

This key serves as the source of truth for task existence checks throughout the system lifecycle.

Broadcasting Progress via Redis Pub/Sub

Throughout the workflow, the backend publishes progress messages—such as "任务开始处理" (task started) and "任务处理完成" (task completed)—to a Redis Pub/Sub channel unique to each task. The channel follows the pattern task:{task_id}:messages.

The RedisManager.publish_message method in backend/app/services/redis_manager.py (lines 67-74) serializes the Message model to JSON and publishes it to the channel:

await redis_manager.publish_message(
    task_id,
    SystemMessage(content="任务开始处理")
)

Simultaneously, the system persists these messages to a local log file via RedisManager._save_message_to_file (lines 39-61) for later inspection and debugging.

WebSocket Subscription and Message Delivery

When a client connects to the /task/{task_id} WebSocket endpoint, backend/app/routers/ws_router.py first validates task existence by checking the Redis key. If the key does not exist, the connection is rejected immediately (lines 15-19).

Upon validation, the endpoint subscribes to the task-specific channel:

pubsub = await redis_manager.subscribe_to_task(task_id)

The subscribe_to_task helper in backend/app/services/redis_manager.py (lines 83-88) creates a pubsub object and subscribes to task:{task_id}:messages.

Inside an infinite loop, the endpoint retrieves messages from Redis and forwards them via WebSocketManager.send_personal_message_json:

# ws_router.py message loop

while True:
    msg = await pubsub.get_message(ignore_subscribe_messages=True)
    if msg:
        payload = json.loads(msg["data"])
        await ws_manager.send_personal_message_json(payload, websocket)
    await asyncio.sleep(0.1)

The WebSocketManager.send_personal_message_json method in backend/app/services/ws_manager.py (lines 18-20) handles the actual JSON transmission to the specific client connection.

Graceful Teardown and Resource Cleanup

When a WebSocket client disconnects, the ws_router.py endpoint (lines 64-66) ensures proper cleanup by unsubscribing from the Redis Pub/Sub channel and removing the socket from the WebSocketManager connection pool. This prevents memory leaks and orphaned subscriptions.

Summary

• Task state persistence: MathModelAgent stores task identifiers in Redis using the task_id:{task_id} pattern with a 10-hour TTL via RedisManager.set.
• Real-time broadcasting: Progress updates flow through task-specific Pub/Sub channels (task:{task_id}:messages) using RedisManager.publish_message, which also archives messages to local files.
• WebSocket integration: The /task/{task_id} endpoint validates task existence against Redis before establishing subscriptions, then relays Pub/Sub messages to clients via WebSocketManager.send_personal_message_json.
• Resource management: Automatic cleanup of Redis subscriptions and WebSocket connections prevents resource exhaustion when clients disconnect.

Frequently Asked Questions

How long does MathModelAgent retain task state in Redis?

Task state keys expire after 10 hours (36,000 seconds) by default. The RedisManager.set method in backend/app/services/redis_manager.py automatically applies this TTL, ensuring that completed or abandoned tasks do not consume memory indefinitely.

What Redis channel pattern does the system use for task messages?

The system uses the pattern task:{task_id}:messages for Pub/Sub channels. When a task begins, publishers and subscribers both reference this specific channel identifier to isolate communication between individual workflow instances.

How does the WebSocket endpoint verify task existence before connecting?

The endpoint in backend/app/routers/ws_router.py checks for the existence of the task_id:{task_id} key in Redis before accepting the WebSocket upgrade. If the key is missing, the connection is rejected immediately, preventing subscriptions to non-existent workflows.

What happens to published messages if the WebSocket client disconnects unexpectedly?

Redis Pub/Sub operates as a fire-and-forget messaging system, so messages published during a disconnection are not queued for that client. However, RedisManager.publish_message persists every message to a local log file via _save_message_to_file, enabling developers to inspect the message history for debugging and auditing purposes.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}