[kkkhs]

Fixes #1325.

What

LocalSessionManager now treats dropping a request-wise response stream as cancellation for the in-flight client request.

Fix

Wrap request-wise response streams with a drop guard that asks the session worker to send a notifications/cancelled message and clean up the channel resources when the client disconnects before the stream completes.

Test

Added stateful_sse_client_disconnect_cancels_request alongside the existing stateless disconnect cancellation coverage.

[leonardorudolfocardoso]

Thanks for picking this up, @kkkhs! First, a heads-up: I posted a correction on #1325. My original repro went through the stateless path, not LocalSessionManager, so the scope of the problem is narrower than I described.

I built this branch (e0dc4ba) and tested it end to end with a tool that watches ctx.ct:

Scenario (legacy 2025-06-18 session)	main	this PR
Client disconnects mid-call	handler completes	✅ ct fires
Client disconnects, then GET with Last-Event-ID from the priming event	✅ result replayed	❌ handler cancelled; resume logs Resume failed (Session error: Channel closed: Some(1)) and returns an empty stream

Main concern: this breaks SSE resumption. Request-wise streams send a priming event by default (SessionConfig::sse_retry = 3s) so a client can reconnect after a dropped connection. On drop, RequestWiseResponseStream sends notifications/cancelled and remove_request_wise_channel deletes the tx_router entry, so a later resume can't succeed. The test comment this PR deletes made exactly this point: on the stateful path, a disconnect may be recovered through Last-Event-ID. The stateless fix keeps the same rule. CancelOnDisconnect in tower.rs skips cancellation when the request is persisted ("Persisted requests keep running so another connection can resume them").

Suggestions:

• Only cancel on drop when resumption isn't possible, e.g. sse_retry.is_none() and no event_store. Or make it an explicit opt-in on SessionConfig.
• Add a regression test: disconnect mid-call, resume with the priming event's id, and assert the response arrives and ct did not fire. No current test covers this, which is why the suite still passes (disconnect_cancel / priming / stale_session all green locally).
• Streams returned by resume() aren't wrapped. If cancel-on-drop stays, a disconnect after a resume should behave the same way.

Smaller points:

• cancel_request_wise_channel_on_disconnect calls tokio::spawn from Drop when the channel is full. That panics if the stream is dropped outside a runtime (e.g. during shutdown). tokio::runtime::Handle::try_current() would avoid it.
• In the worker loop, context.send_to_handler(..).await? means a failed cancel notification ends the whole session worker. Logging and continuing seems safer.
• The branch is ~29 commits behind main and needs a rebase.

If this direction sounds right to maintainers, I can help with the resume test.