fix(agentex): end SSE subscriptions on terminal task / vanished topic; stop deleting shared stream

[deepthi-rao-scale]

What was broken

When you open a task in the UI, the backend keeps a live stream open to push updates. That stream had no way to end itself — it only stopped if the browser disconnected.

So when a task finished, the backend just kept listening forever, holding one Redis connection each time. These pile up until the pool is empty, at which point the pod can't serve requests (and its health check fails, so it sits stuck until a manual restart).

A second bug: when one viewer left, the code deleted the task's shared event stream — yanking it out from under everyone else watching the same task.

The fix

• End the stream when the task is done. The backend already gets a task_updated event when a task finishes, so the stream now closes as soon as it sees a terminal status.
• Fallback for the rare cases the event can't cover (viewer connects after the task already finished, or the producer dies silently): a lightweight check on the existing keepalive interval closes the stream then too.
• Stop deleting the shared stream on one viewer's exit — the Redis TTL already cleans it up on its own.

Testing

Two integration regression tests (stream self-ends on terminal task; one viewer disconnecting doesn't delete the shared stream).

Verified live (A/B against main) — same steps both times: create a RUNNING task → open GET /tasks/{id}/stream → complete the task, while watching Redis blocked_clients.

	on task completion	Redis connection
main	stream keeps XREAD-looping; kept sending :ping 15s+ after completion and only ended when the client was force-disconnected	blocked_clients stayed 1 — released only on client disconnect (zombie)
this branch	stream delivers the terminal task_updated and returns immediately (Ending SSE stream …: received a terminal task_updated event)	blocked_clients 1 → 0 instantly, no client disconnect needed

On main the reader logged Reading messages from Redis stream … every ~2s indefinitely after the task was done; on this branch it stops the moment the terminal event arrives. Same reproduction the incident described, and the connection is released instead of pinned.

Note: the unit/integration suite runs in CI; the tutorial-agent integration jobs were red due to a pre-existing missing-pytest-asyncio issue unrelated to this change (fixed separately in scale-agentex-python).

Related: the UI-side half of this leak is #383.

🤖 Generated with Claude Code

Greptile Summary

This PR updates task SSE subscription lifecycle handling.

• Ends subscriptions when a terminal update is received, authoritative task state becomes terminal, or the task disappears.
• Replays buffered events and closes when connecting to an already-terminal task.
• Publishes failure updates and centralizes terminal/non-terminal status classification.
• Preserves shared Redis streams when individual subscribers disconnect.
• Adds integration coverage for terminal shutdown, late connections, reclaimed keys, transient lookups, and shared-stream retention.

Confidence Score: 4/5

The PR should not merge until recoverable FAILED tasks no longer terminate subscriptions that must observe their subsequent RUNNING updates.

The stream returns upon receiving any FAILED task update, while the task service intentionally supports forwarding that same FAILED task again and publishing a RUNNING recovery update; viewers whose subscriptions closed on FAILED cannot receive that recovery or subsequent events.

Files Needing Attention: agentex/src/domain/entities/tasks.py, agentex/src/domain/use_cases/streams_use_case.py, agentex/src/domain/services/task_service.py

Important Files Changed

Filename	Overview
agentex/src/domain/entities/tasks.py	Introduces canonical terminal and non-terminal status sets shared by state transitions and SSE termination.
agentex/src/domain/services/task_service.py	Publishes FAILED updates through the normal update path, while existing retry behavior can still restore FAILED tasks to RUNNING.
agentex/src/domain/use_cases/streams_use_case.py	Adds event-driven and authoritative fallback termination and stops deleting the shared Redis topic, but the previously reported recoverable-FAILED lifecycle conflict remains.
agentex/src/domain/use_cases/tasks_use_case.py	Reuses the canonical non-terminal status set for transitions to terminal states.
agentex/tests/integration/test_task_stream.py	Adds regression coverage for terminal shutdown, late subscribers, vanished tasks, transient lookups, reclaimed keys, and shared-stream preservation.

Sequence Diagram

sequenceDiagram
    participant Client
    participant SSE as StreamsUseCase
    participant DB as Task Store
    participant Redis
    Client->>SSE: Subscribe to task stream
    SSE->>Redis: Snapshot stream cursor
    SSE->>DB: Read authoritative task state
    alt Task already terminal
        SSE->>Redis: Replay buffered events
        SSE-->>Client: Events, then close
    else Task is non-terminal
        loop Until terminal, missing, or disconnect
            SSE->>Redis: Read new events
            alt Terminal task_updated received
                SSE-->>Client: Terminal event
                SSE-->>Client: Close stream
            else Status-check interval elapsed
                SSE->>DB: Recheck task
                alt Terminal or missing
                    SSE-->>Client: Close stream
                end
            end
        end
    end
    Note over SSE,Redis: Subscriber exit does not delete the shared stream

Loading

_{Reviews (14): Last reviewed commit: "fix(agentex): terminate SSE task streams..." | Re-trigger Greptile}