fix: close resumed streams when the DONE pub/sub message is lost

[mdnanocom]

Problem

The DONE control message is delivered to resumed streams over Redis pub/sub, which is fire-and-forget. If that single message is lost — subscriber connection dropped and reconnected at the wrong moment, or the producer died between writing the DONE sentinel and publishing — a resumed stream stays open forever, even though:

• every content chunk was already delivered, and
• the durable sentinel key already says DONE.

We hit this in production miltiple times behind an AI SDK chat UI: the answer rendered fully, but the SSE response never closed, so the client stayed in streaming state indefinitely (the AI SDK only leaves streaming when the connection closes). A page refresh recovered — resumeExistingStream reads the sentinel and returns null — which shows the durable state was correct; only the live consumer never re-checks it.

Related but distinct from #42 / #43, which cover the initial ack phase. This is about a consumer that attached successfully and then never learns the stream ended.

Fix

Add a watchdog to resumeStream: periodically re-read the durable sentinel and close the stream once the producer is finished or the sentinel expired.

• Closes only after two consecutive DONE/missing observations, so in-flight tail messages get a full interval to drain before we give up on them.
• Interval configurable via a new doneWatchdogIntervalMs context option (default 10 000 ms) — one GET per interval per attached consumer.
• Watchdog is cleared on every existing cleanup path (DONE received, ack timeout, enqueue failure).
• A failed sentinel read (Redis briefly unavailable) is swallowed and retried next tick — it must not kill a healthy stream.

Test

New test in generic.test.ts using the in-memory pub/sub: a wrapper publisher drops the DONE control message (durable SET still goes through), a resumed consumer reads the stream. Without the watchdog the test hangs forever; with it, the consumer receives all chunks and closes within two intervals.

[mdnanocom]

@cramforce hi sorry to bother you, any chance this pr gets merged or should i close it?

[cramforce]

I'm slightly sus of this PR but open to it. Is polling really the only option?

[cramforce]

(And thanks for the ping!)

[mdnanocom]

@cramforce
are you suspicious of the problem or of the implem?
some form of durable-state recheck is necessary with the current primitives.
if the DONE publish is lost and the producer disappears after writing the sentinel, no further event is guaranteed
the sentinel is the only surviving evidence, so the consumer must eventually read it again.
that does not require polling every attached stream at a fixed interval. I could reset an idle timer on each received chunk and check the sentinel only after the stream has been quiet for the configured interval, continuing periodically only while it remains idle.
that would keep the recovery behavior while avoiding reads for active streams.
a durable blocking system such as Redis Streams could avoid polling entirely, but the current publisher/subscriber interface does not expose one.

[cramforce]

The fix makes sense, but the watchdog currently leaks if subscription setup fails: after resumeStream() rejects, the interval keeps polling Redis indefinitely. Async setInterval can also overlap when Redis is slow. Could you clean it up in the outer catch/stream cancellation path and use non-overlapping polling (e.g. recursive setTimeout), with regression tests?

[mdnanocom]

@cramforce
addressed in my latest commit
cleanup is now idempotent and shared by setup failures, stream cancellation, timeouts, and normal completion.
replaced the async interval with recursive setTimeout
the watchdog now starts only after the initial ACK/data message, preventing it from interfering with subscription setup

[cramforce]

Alright, thanks so much and so for this being a slow process. One last thing: I need you to sign the commits