Fix continue-as-new handling in workflow_streams subscribe

[brianstrauch]

What

Two fixes to how WorkflowStreamClient.subscribe() behaves when the host workflow continues-as-new.

1. Continue-as-new detection (_follow_continue_as_new)

_follow_continue_as_new described the workflow with no run id, which returns the current run. After a continue-as-new that is the new RUNNING run — never CONTINUED_AS_NEW, which only sits on the old, closed run — so the check never fired and subscribe() stopped (or surfaced an error) during a rollover instead of following the stream.

Now we capture the run id each poll's update is admitted to (start_update with WaitForStage.ACCEPTED, read workflow_run_id, then await result()) and describe that specific run on failure. A rolled-over run reports CONTINUED_AS_NEW (→ retry), a terminal run reports a terminal status (→ stop), a still-RUNNING run is a transient error (→ surface). This also avoids mistaking an unrelated new execution that reused the workflow id for a successor.

2. Retry polls rejected while draining

The poll update's validator rejects new polls during detach-for-CAN. It raised an untyped RuntimeError, which subscribe() did not classify and re-raised — ending the subscription with an error during a routine rollover. The validator now raises the well-known StreamDraining ApplicationError type, and subscribe() backs off and retries on it.

Cross-language note

"StreamDraining" is a new well-known error type in the cross-language workflow-streams protocol (alongside TruncatedOffset). The matching changes land in sdk-go and sdk-typescript.

Tests

• test_follow_continue_as_new_describes_polled_run — against a real post-CAN run, the polled (old) run reports CONTINUED_AS_NEW and the latest reports RUNNING; the helper follows only when describing the polled run.
• test_subscribe_retries_while_draining — a workflow that detaches and holds open until released; a live subscribe() survives the draining window and resumes on the successor run.

Full tests/contrib/workflow_streams/ suite passes; mypy clean.

🤖 Generated with Claude Code