Skip to content

[Brook S8] Added Activity Streams, SyncStream and BufferedPublisher. - #378

Closed
moetemp wants to merge 1 commit into
moe/brook/review/s7-workflow-streamsfrom
moe/brook/review/s8-activity-handles
Closed

moetemp wants to merge 1 commit into
moe/brook/review/s7-workflow-streamsfrom
moe/brook/review/s8-activity-handles

Conversation

@moetemp

@moetemp moetemp commented Oct 25, 2026 •

Copy link
Copy Markdown
Owner

What was changed

This PR adds the Stream calls in Activities, and BufferedPublisher, SyncStream and SyncSubscription in temporalio.client.

  • In an Activity, a Stream of the Worker's client, such as from activity.client().to_stream(ref), writes as the attempt, with the activity or standalone_activity epoch case. So a retry withdraws the Records of the earlier attempts, even if it writes nothing. A Stream of another client writes as that client.
  • A sync Activity has no activity.client(), so it keeps the Worker's client, such as in an Activity class, and wraps a Stream in SyncStream(stream), whose calls block. publish and subscribe of a Stream in a sync Activity raise StreamUnsupportedError, and the message points to SyncStream.
  • publish starts the write at the call. The end of the attempt waits for the attempt's writes, and a write error that no caller got fails the attempt.
  • BufferedPublisher(stream, *, max_delay=...) takes a Stream and batches its publishes, in order, and flushes at the end of an Activity attempt. Workflow code has no Stream, and a WorkflowStream passed with no type checks raises StreamUnsupportedError.
  • An Activity never closes a Stream. close, async with and SyncStream.close raise StreamNotOwnerError with no call, and async with raises at __aenter__. producer raises StreamUnsupportedError, since the attempt is the Producer.
  • A sync call runs on the Worker's loop. run_blocking makes the call's future before it schedules the call, so a cancel of the Activity thread after the schedule also cancels the call. A call when the Worker's loop is already closed closes its coroutine, so nothing warns at shutdown.

Why?

Agents stream model tokens from an Activity. The Workflow keeps the Stream, and the Activity writes to it.

Part of the Streams review series.

Checklist

  1. How was this tested:
    • poe lint and pytest tests/contrib/streams on both backends.

@moetemp moetemp added the skip-changelog Changelog entry rides another PR label Oct 25, 2026
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from b298e44 to 90966d6 Compare October 26, 2026 00:14
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch 4 times, most recently from 24f6a90 to 79cf55a Compare October 26, 2026 04:52
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from 1c5e48f to 84f1587 Compare October 26, 2026 08:44
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 79cf55a to 15eebd7 Compare October 26, 2026 08:44
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from 84f1587 to c3f30a4 Compare October 27, 2026 09:07
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 15eebd7 to b3da39a Compare October 27, 2026 09:07
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from c3f30a4 to b9572b1 Compare October 27, 2026 09:17
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from b3da39a to 212ef71 Compare October 27, 2026 09:17
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from b9572b1 to 925cf34 Compare October 28, 2026 01:33
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 212ef71 to 4a67b4e Compare October 28, 2026 01:33
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from 925cf34 to cc8e51c Compare October 28, 2026 07:54
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 4a67b4e to 0c7ec5b Compare October 28, 2026 07:54
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from cc8e51c to e7710c0 Compare October 28, 2026 08:05
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 0c7ec5b to 86d50a1 Compare October 28, 2026 08:05
@moetemp moetemp changed the title [Brook S8] Added Activity Stream handles and BufferedPublisher. [Brook S8] Added Activity Streams, SyncStream and BufferedPublisher. Nov 3, 2026
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 7fd6dcb to ee23c91 Compare November 5, 2026 17:19
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch 2 times, most recently from a912bfb to 1c9ed91 Compare November 5, 2026 23:10
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from ee23c91 to 0d2120b Compare November 5, 2026 23:10
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from 1c9ed91 to d57a2ae Compare November 6, 2026 09:06
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 0d2120b to b346382 Compare November 6, 2026 09:06
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from d57a2ae to aef29b9 Compare November 6, 2026 09:22
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from b346382 to 058106a Compare November 6, 2026 09:22
@moetemp
moetemp force-pushed the moe/brook/review/s7-workflow-streams branch from aef29b9 to ee2e261 Compare November 6, 2026 10:11
@moetemp
moetemp force-pushed the moe/brook/review/s8-activity-handles branch from 058106a to 1b71cf6 Compare November 6, 2026 10:11
@moetemp

moetemp commented Nov 10, 2026

Copy link
Copy Markdown
Owner Author

Closed in favor of #381, the combined branch, which has this change.

@moetemp moetemp closed this Nov 10, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

skip-changelog Changelog entry rides another PR

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant