[campoy]

What was changed

On the streaming route, a parallel input guardrail that trips now stops the streaming model activity, so stream topic subscribers stop getting the answer.

• When a parallel input guardrail trips while the streaming model activity is running, _TemporalModelStub.stream_response cancels the activity. The run raises InputGuardrailTripwireTriggered, as before.
• When the guardrail has already tripped as the model is called, the activity is not started.
• The stub acts on the trip as soon as the framework records it. It does not wait for the other parallel guardrails to finish their cleanup.
• When the workflow cancels it, invoke_model_activity_streaming publishes an error event (ResponseErrorEvent, code cancelled) to the topic and stops. This also happens when the cancel arrives after the model finished, while the activity is still publishing its events, stopping the stream's background flusher, or stopping its heartbeats. Then the error event comes after the buffered events. Other kinds of cancel (Worker shutdown, heartbeat timeout) publish no error event, because the activity's retry can still answer.
• To cover the heartbeat cleanup, the streaming activity no longer uses auto_heartbeater. It uses heartbeating(), a context manager taken out of the decorator, inside its cancellation handling. auto_heartbeater uses the same context manager, so other activities are unchanged.
• To watch the run's guardrails, the runner hands the streamed run to the model stub through a context variable. The stub reads RunResultStreaming._input_guardrails_task and _triggered_input_guardrail_result.
• The plugin now requires temporalio>=1.35.0,<2, and the lock moves to 1.35.0. With 1.34.0, a workflow that cancels an activity in WAIT_CANCELLATION_COMPLETED mode and is then evicted never finishes the eviction (Eviction processed, but 2 tasks remain), and Worker shutdown hangs. 1.35.0 fixes this SDK bug, and this change makes that cancel automatic.
• The README's streaming section says what subscribers now get and that the cancel needs heartbeat_timeout. It says the error event comes after response.completed when the model had already finished, and that there is no error event when the activity had already returned. It also says that with the default TRY_CANCEL, a workflow that completes right after the trip can close before the error event is published, and that WAIT_CANCELLATION_COMPLETED makes the run raise only after it is published. The README's version line now says 1.35.

No public name or option changes.

Breaking change: the cancel request is a new command, and not starting the activity drops one, with no version gate. We are not adding version gates to these fixes: breaking old histories is accepted with the plugins' 1.0 releases. A streamed run in which a parallel input guardrail tripped before this version recorded no cancel, so its replay now fails. When the guardrail tripped before the model call, it fails with [TMPRL1100]. When it tripped while the model streamed, it fails with [TMPRL1100] or with the SDK core's Fatal error in workflow machines: ActivityMachine in state ScheduledActivityCancelCommandCreated says the transition is invalid during event ... ActivityTaskStarted, depending on the History. Streamed runs whose guardrails did not trip, and runs that don't stream, are not affected.

Upgrade: let open streamed runs in which a parallel input guardrail tripped finish on the earlier version before you upgrade, or keep them there with Worker Versioning. Upgrade temporalio to 1.35.0 or later.

Review the combined diff against main.

Why?

See #227. The streaming activity publishes each event to the topic as the model sends it, but the workflow only yields the events after the activity returns, so the framework's per-event guardrail check never gated the topic. A subscriber got the whole answer to a rejected request, and response.completed.

Checklist

1. Closes [Bug] openai_agents: stream topic subscribers get the rest of the answer after a parallel input guardrail trips #227

2. How was this tested:

   • tests/test_openai_streaming.py:
     • test_streaming_input_guardrail_trip_stops_the_topic[alone|sibling-cleaning-up]: a parallel guardrail trips after one second while the fake model holds after its first delta. In the second case, another guardrail takes 12 s to clean up. The subscriber's last event is the cancelled error, with no response.completed, and the model's stream was cancelled.
     • test_streaming_input_guardrail_known_trip_skips_the_model[alone|sibling-cleaning-up]: a guardrail trips while an agent hook delays the model call. No streaming activity is scheduled.
     • test_streaming_input_guardrail_trip_waits_for_the_error_event[streaming|publishing|stopping-the-flusher|stopping-the-heartbeats]-[cached|uncached]: with WAIT_CANCELLATION_COMPLETED, a workflow returns right after the trip, and the Worker shuts down, with or without a Workflow cache. The stream's last event is the error event. In streaming, the cancel arrives while the model holds after its first delta, so the stream has that delta and the error event. In publishing, the model finishes at once but the stream's publish signals are held for 8 s, so the cancel arrives during the activity's final flush. In stopping-the-flusher, the background flusher's first publish hangs until it is cancelled and then takes 8 s to clean up, so the cancel arrives while the stream's context manager stops the flusher. In stopping-the-heartbeats, the heartbeat task takes 8 s to stop and keeps heartbeating meanwhile, so the cancel arrives while the activity stops its heartbeats. In those three, the stream then has two deltas, response.completed and the error event.
     • test_streaming_worker_shutdown_publishes_no_error_event: the Worker shuts down while the model streams, and the retry on a new Worker answers. The topic gets no error event.
   • Run against the plugin source of 68c494b (on temporalio 1.34.0), these fail: the sibling-cleaning-up cases, the Worker shutdown test, and the uncached waits_for_the_error_event case (Eviction processed, but 2 tasks remain, pytest timeout). The other new cases pass.
   • With this change but temporalio 1.34.0, only the uncached waits_for_the_error_event case fails, with the same eviction hang. On 1.35.0 it passes.
   • Run against the plugin source of ecff357, both publishing cases fail: the stream has no error event. The streaming cases pass.
   • Run against the plugin source of b8a943e, both stopping-the-flusher cases fail: the stream's last event is response.completed. The streaming and publishing cases pass.
   • Run against the plugin source of fb5db8c, both stopping-the-heartbeats cases fail: the stream's last event is response.completed.
   • A throwaway run with TRY_CANCEL and a workflow that returns right after the trip had only the delta in its stream, with and without a Workflow cache. That is the caveat the README describes.
   • Locally, make lint and make test pass (458 tests). These ran in Docker with Python 3.13 and uv 0.12.24.
   • On main at c664a34, the trip test failed: assert 'response.completed' not in [...]. This was run on an earlier version of the test.
   • A History of the trip test recorded on main (the guardrail tripped mid-stream, the activity ran to completion, no cancel) fails replay on 68c494b: [TMPRL1100] Nondeterminism error: Activity machine does not handle this event: HistoryEvent(id: 28, WorkflowExecutionUpdateCompleted). That is the breaking change above, so it is not in the tests.
   • [Bug] openai_agents: stream topic subscribers get the rest of the answer after a parallel input guardrail trips #227's reproduction exits 1 on 1.1.0 and on main at c664a34. At 68c494b it exits 0: subscriber got: ['part 0 ', 'part 1 ', 'part 2 ', 'part 3 ', 'error'], the same four deltas a native consumer got.

3. Any docs updates needed?
   The README is updated here. Release notes should mention the cancel, the error event on the topic, the new temporalio>=1.35.0 floor, and which runs stop replaying (see "Breaking change").

Not covered:

• Without heartbeat_timeout, the activity is not told about the cancel and keeps publishing until the model finishes.
• A cancel that reaches the Worker after the activity has returned leaves the stream ending at response.completed. This race is inherent; the README states it.
• Only Worker shutdown was tested among the cancels that publish no error event; pause and reset were not.
• The replay check was not rerun without the gate, and histories recorded with this change were not replayed on main.
• temporalio 1.34.x still bundles the old code as temporalio.contrib.openai_agents. No backport here.

Open questions:

• Should the streaming activity heartbeat even without a heartbeat timeout, given the 30 s throttle?
• Is reading the two private RunResultStreaming fields fine with openai-agents pinned to <0.23, or should the plugin wrap each parallel input guardrail instead to learn about the trip?
• Should temporalio 1.34.x get a backport?

Draft while the breaking change and these questions are open.

[campoy]

Pushed addebd2 to fix a flaky run of the new test (CI on 6dc7a2a failed it on Ubuntu with Python 3.14).

The test raced the fix: the fake model streamed five deltas 300 ms apart, so the cancel had about 0.5 s to arrive with the next heartbeat, and on a slow runner the model sometimes finished first. Now the fake model sends its first delta and then waits up to 10 s, and the workflow stays open until the test has read the topic's last event instead of sleeping. The replay test's History is recorded again on main with the new workflow. The plugin code is unchanged.

Locally the test now passes five runs of five on Python 3.13 and on 3.14, and still fails on main on assert 'response.completed' not in [...]. The fix without its gate now fails the replay test with [TMPRL1100] Nondeterminism error. CI on addebd2 passes in every cell.