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: no P0-P2 findings

Reviewed 4ec7095. Run log

Adds durable messages to the SDK. Inside a durable function, ctx.wait_for_message waits for a message and ctx.publish sends one. On the client, client.durable.messages gets send, list and subscribe; subscribe catches up from the message list and then follows a new durable realtime channel on the client's existing connection. The PR also exposes client_token on executions and moves the existing approval POST/retry code into a shared module without changing its behaviour. I read all of the hand-written source closely, including the ordering, catch-up, session-change and server-ended teardown paths, and found no defects. Risk comes from the size of the change, the new bearer-credential handling, and small changes to the realtime connection that every channel goes through.

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.

• Server contract in volcano-hosting#1826: broadcast payload with prev_seq, the session/wait/publish routes, and authorization of client_token in realtime subscribe data.
• Whether centrifuge-python 0.6 fires on_unsubscribed before ready() fails on a refused subscribe, and never for client disconnects; server-ended detection relies on both.
• The shared acceptance scenarios in volcano-hosting#1827 have not run against staging for any of the three SDKs.
• About 4,000 lines of new tests were skimmed, not read line by line, to confirm each one exercises the behavior it names.
• Confirm callback payloads, sender-scoped idempotency, and realtime authorization against Kong/volcano-hosting#1826 and the shared acceptance scenarios in #1827.

Human review: needed

• Is it intended that client_token (a bearer credential on DurableExecution) goes out as a header and as realtime subscribe data, letting anon keys and other users reach an execution?
• Does ending a subscription only on a session-lineage change (DurableChannels.session_check) reliably stop a re-signed-in user from getting the previous user's messages?
• Is reading 'unsubscribed while unpaused' as a server end in ChannelEvents.on_unsubscribed sound, given every realtime channel type goes through it?
• Should possession of DurableExecution.client_token grant other users and anonymous clients access to execution messages, including through realtime subscriptions (models.py, _realtime_durable.py)?

The AI review found no P0-P2 problems at 4ec7095, so the code needs no second review for bugs. Decide the points above: approve to accept them, or request changes.

Suggested reviewer: swkeever (wrote most of realtime.py and _transport.py, which 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).