[moetemp]

What changed?

This PR adds the Activity handles ActivityStream and SyncActivityStream, their Topics, and BufferedPublisher.

An Activity gets a Stream only as an argument, such as a field of its one input object. await topic.publish(*values) writes the Records with one WriteRecords call and returns the offset of the last one. The sync handle blocks the Activity thread and runs the call on the Worker's loop. Both handles also read, give the next offset, and close. A handle of another backend raises StreamUnsupportedError.

streams.BufferedPublisher(topic, max_delay=timedelta(milliseconds=100)) is the optional helper that batches. publish returns at once, a batch goes out at most max_delay after its first value, flush() writes the rest, and async with flushes at the end of the block.

After a batch fails, the publisher writes nothing more, so a reader sees no gap, and each later call raises the error. The end of an Activity attempt flushes the attempt's open publishers, and a write error there fails the attempt unless the Activity already saw it.

A cancel of a sync Activity also cancels the call that it waits for on the Worker's loop.

Before a Local Activity runs, the plugin calls the crate's FlushRunBuffer with the dispatching Workflow Task's facts, so the run's Records come before the Local Activity's. A failure fails the attempt. A retried attempt opens a writer on each Stream it received, which withdraws the earlier attempt's Records also when the new attempt writes nothing. It finds the Streams in arguments, fields, slots, lists and dicts.

Why?

S47, S67, S70, S71 and S79: an awaited publish has published when it returns, with no hidden timer, and a Local Activity sees the Records that its run wrote before it.

How did you test it?

Live tests in tests/contrib/streams/test_activity_handles.py on the memory Store and on a scratch redis-server: offsets from async and sync publishes, a read in an Activity, a buffered publisher, a retry that withdraws a received Stream's Records, a close from an Activity, the backend check, and the order of Workflow and Local Activity Records. A publisher that is never flushed, Streams in a list and an optional field of a retried Activity's input, and a sync read. Unit tests cover the buffered publisher's batching and sticky errors, and the cancel of a blocked sync call. pyright, mypy, basedpyright, pydocstyle, ruff.

• Unit Tests
• Staging
• End to End Tests

[moetemp]

Closing as stale. This layer is folded into the v5 tip, #365, and the upstream review uses a fresh series instead. Plan: https://www.notion.so/4048fc567738814990b7c796ac28fe3f