feat(durable): wait for, publish, send, and subscribe to durable messages

[subnetmarco]

Change

Durable functions can wait for messages and publish their own, and applications can send, list and follow them. Server support is in volcano-hosting#1826.

• Authoring. ctx.wait_for_message(name, timeout=...) suspends until a message with that name arrives, and returns None on timeout. ctx.publish(name, data) is a checkpointed step returning the message's id and seq. The first publish in a context opens a messages session through a callback; each publish sends sha256(session token):position as its idempotency key, where the position counts publishes in that context, so a rerun after a crash returns the original message.
• Client. client.durable.messages.send(function, execution_id, name, data, client_token=..., idempotency_key=...), list(...) with direction, name, after and limit, and await subscribe(..., handler). Subscribe catches up from the list and follows the execution's realtime channel on the client's existing connection. Passing project_id uses the project route.
• Executions expose the client_token their creating start returned.
• Internal realtime types stay out of public signatures: the client builds a ClientRealtime subclass, and the durable code depends on a small DurableChannel protocol.
• docs/functions.md and the README are updated. The OpenAPI client is regenerated.
• Purely additive. Release Please should cut 0.15.0, the minimum the Hosting docs name for wait_for_message and publish.

Verification

• uv run --locked poe checks: 2,865 tests, 100% coverage, and 100% type completeness in package-types
• Mutation testing (scripts/mutation.sh) for every changed module, including _durable_messages, _realtime_durable, _durable_message_stream, _durable_message_response, _transport_durable_messages, _realtime_connection, realtime, client and durable_authoring: no surviving mutants

Cross-language impact

Classify this change: public facade / wire contract / shared behavior

• Requirement IDs and affected behavior: SDK-DURABLE-011 to 020 (message waits, timeout, publish order, gap-free subscribe, user isolation, client-token anon sends, idempotent sends, ended executions, read-only tokens).
• JavaScript / Python / Ruby companion PRs, or why each is unaffected: volcano-sdk-js#319 and volcano-sdk-ruby#343 (client only, since Ruby has no durable authoring API).
• Canonical scenarios, bindings and active/staged copies: tests/sdk-contract/features/durable-messages.feature, with bindings for all three languages, in volcano-hosting#1827.
• Public examples updated in each affected language: docs/functions.md and the README here; JavaScript and Ruby in their PRs; docs/public/functions/durable-messages.md in Hosting.
• Native checks and shared acceptance evidence (with tested revisions): native checks above on 990f833. Shared acceptance hasn't run yet; it runs from volcano-hosting#1827 once the three SDK PRs merge and volcano-hosting#1826 is on staging.

[subnetmarco]

Matches JS closely: same publish key, same stream algorithm, same retry rules. Comments:

• Subscribe and the client token. subscribe accepts client_token but its docstring says the realtime connection needs a signed-in session. The public docs and the volcano-durable skill both show a Python subscribe that passes only client_token, which reads as though an anon client can follow an execution the way it can in JS. Either support it, or have the docs and skill say that Python subscribe needs a signed-in user (and whether a service key works), and raise a clear error instead of failing at the realtime layer.
• Same two issues as JS: the stream drops seq <= last_seq, so a lost or reordered broadcast is skipped for good (see #1826), and durable_authoring opens one session per context against the server's cap of 20, so ctx.map over more than 20 publishing items fails.
• The OpenAPI sync pulls in unrelated models (variable environments, sandbox fields). Same suggestion as JS: a separate chore: commit so the release notes don't credit them to this feature.

[subnetmarco]

Addressed in 58b4f2c.

• Subscribe and the client token. subscribe now checks for a signed-in session before touching the channel. Without one it raises RuntimeError: subscribe() needs a signed-in session to open the realtime connection; a client token or service key alone cannot, with no subscription made and no list read. A service key doesn't work either; the test covers both cases. The README, docs/functions.md, the hosting page (hosting#1826) and the volcano-durable skill (volcano-skills#59) now say that a client holding only the client token or a service key can poll list with after.
• Gaps. Each broadcast now carries prev_seq. When it's ahead of the last delivered seq, the stream holds the message, reads the gap from the list in the background, then releases it in order. A failed read is reported to the loop's exception handler and the held message is still delivered. A resubscription catches up the same way. Reads also stop at a page that delivers nothing new, so a server repeating a page can't loop it. New tests cover a lost broadcast, a late arrival, a non-integer prev_seq, and a first broadcast without one.
• Sessions. Hosting raised the cap to 1,000 per execution over its life, the same as the publish cap. The docs and the skill give the per-item cost and say to publish from the parent across a ctx.map. We kept one session per context: reusing the parent's would need publish keys built from the context path and more replay handling in each SDK.
• Mutation. _durable_message_stream, _durable_message_response and _durable_messages score 100% with no timeouts. Two test-harness fixes made that possible. dispatched() waits with asyncio.wait, so a handler running on the dispatcher can't deadlock it by cancelling it. The fake transport's canned list answers are each given once, so a read that keeps going falls through to the published pages instead of spinning.
• OpenAPI sync. It's part of 990f833 here. The models it adds only touch volcano_sdk/_generated, which isn't public API. Splitting it into a chore: commit means rewriting that commit and force-pushing, and it only shows up in the release notes if the PR is rebase-merged rather than squashed. Say if you want that and I'll do it.

[github-actions]

AI review: 1 P0-P2 finding(s)

Reviewed 30111f5. Run log

Adds durable message authoring, REST operations, and realtime subscriptions. A startup race can return a subscription that has already stopped receiving messages.

🟡 P2

• src/volcano_sdk/_durable_messages.py:365 Register server-end handling before awaiting subscription readiness

⚪ P3

• src/volcano_sdk/_durable_messages.py:329 A project-route subscribe still sends client_token, though a test name says it sends none

Not checked: The AI review read the diff, the code, and the PR's text; it ran no tests or builds and read nothing outside this repository.

• Whether volcano-hosting#1826 broadcasts the full message plus an integer prev_seq on durable:<id>. Broadcasts the SDK can't parse are silently ignored.
• Whether realtime accepts client_token as subscribe data and ends a hidden execution with a permanent unsubscribe, not a resubscribe code.
• Whether real centrifuge-python 0.6 fires on_unsubscribed only when the server ends a subscription, not on dropped connections. The tests use a fake.
• Whether publish and wait work on the real runtime and deployed Volcano: per-context sessions, token lifetime, replay idempotency keys. volcano-hosting#1827 hasn't run.
• Verify callback credentials, message payloads, and durable realtime authorization against Kong/volcano-hosting#1826 and the shared scenarios in #1827 at this SDK revision.

Push a fix, or have a collaborator with write access (including the PR author) add the ai-review-override label to accept 30111f5 as is. An override also needs a human review.

Human review: needed

• Are the new client-token and workflow-session capabilities appropriately scoped to one execution across REST and realtime access (src/volcano_sdk/models.py, src/volcano_sdk/_durable_message.py, src/v…

Decide the points above: approve to accept them, or request changes.

Suggested reviewer: swkeever (wrote most of the realtime and transport code this change extends). Any collaborator with write access other than the author can approve.

[subnetmarco]

Second pass at 2cfdab0. The earlier points are resolved:

• subscribe refuses up front without a signed-in session.
• The stream catches up when prev_seq shows a gap.
• Reads stop at a page that delivers nothing new.
• Mutation is at 100% with no timeouts.

I re-read _durable_message.py, _durable_message_stream.py, _durable_messages.py, _realtime_durable.py and the DurableContext changes against hosting#1826 and the JS SDK. Publish keys and positions, session reuse after a failed open, the callback name and the wait result all match JS. One thing worth fixing:

• A subscription the server ends goes silent. The AI review's P3 is real:

  • When the server ends the subscription, ChannelEvents.on_unsubscribed only calls transport_lost(). That covers an unsubscribe push, and a resubscribe after a reconnect that's refused because the function's visibility changed.
  • channel.ended only runs from reset(), which only removal and disconnect reach.
  • So the stream stays open and the handler never hears anything again.
  • The channel also stays in state.channels, so a new subscribe for that execution raises "already subscribes" until the caller unsubscribes.

  JS handles this in leaveIfServerEnded, which logs and removes the channel. Doing the same here would keep the SDKs aligned: close the stream, report through the loop's exception handler, and remove the channel. Add a test that ends the subscription from the server side.

Smaller:

• The description still says each publish sends its step's operation id as the idempotency key. The code uses sha256(session token):position, the same as JS.
• On the AI review's questions:
  • The checkpointed session token isn't readable through the API. The trace returns operation kind and timing, not callback results, and hosting stores only the token's hash.
  • Postgres delivery already ends a subscription on the next read or delivery after a session change, so binding durable messages the same way is consistent. _handle_if_current checks the session before it calls the handler, so a changed session never gets a message. That looks fine to me.

[subnetmarco]

Done in 30111f5. A subscription the server ends now closes, leaves its channel, and reports once, like JS's leaveIfServerEnded.

• ChannelEvents.on_unsubscribed uses the same test as JS. The SDK pauses or detaches a channel before it unsubscribes one, so an unsubscribe that reaches a current, unpaused channel came from the server. That covers both an unsubscribe push and a resubscribe refused after a reconnect. It calls the channel's new server_ended hook with the server's reason. Other channel types don't set the hook and behave as before.
• The durable subscription sets the hook only after its first subscribe succeeds; before that, subscribe() still raises the refusal. The hook closes the stream at once. It then leaves the channel in a task of its own and reports RuntimeError(<reason>) to the loop's exception handler as "The server ended the durable message subscription; subscribe again to resume". The leave isn't awaited inside the event: centrifuge runs that event on the connection's reader, and leaving takes the connection lock, which a publish waiting on that reader may hold. The report comes after the leave, so a handler can subscribe again right away.
• Tests cover an unsubscribe push (2000) and a refused resubscribe (103): nothing more is delivered, the channel is gone, there's one report, and a new subscribe gets the backlog. Another test checks that an unsubscribe the client makes itself isn't taken for the server's. Both fail against the previous code.
• docs/functions.md and the README describe it.
• The description now gives the publish key as sha256(session token):position.

Ruff, mypy, basedpyright, the policy check, and the full suite at 100% line and branch coverage pass. Mutation on _durable_messages and _realtime_channel leaves no survivors (144 and 250 killed).

[github-actions]

🟡 P2: Register server-end handling before awaiting subscription readiness

RealtimeState._wait_subscription_readiness() awaits a separate task. If the server unsubscribes after wait_subscription() checks channel.subscribed but before its caller resumes, ChannelEvents.on_unsubscribed() finds server_ended still unset and loses the termination. If the subsequent HTTP catch-up succeeds, subscribe() returns a permanently inactive subscription without reporting an error, and another subscription is refused until this one is removed. Install the termination handler before awaiting channel.subscribe(), or retain termination events until it is installed.

_{Found by gpt-6.1-sol}