From a4b18cd45d2d516c51506e5c94ba336db8b40601 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Fri, 30 Oct 2026 01:48:29 -0700 Subject: [PATCH 1/8] fix(strands): give the model the activity's cancellation signal Strands passes the agent's cancelSignal, an AbortSignal, in the model's stream options. The model activity got it as a plain object, so OpenAIModel, which hands it to the OpenAI client, failed every call with "signal.addEventListener is not a function" before sending a request, from @strands-agents/sdk 1.15.0 on. The model activities now give the model the activity's own cancellation signal instead. Cancelling the activity then aborts the model call, as cancelling the agent does natively. --- .../src/__tests__/test-strands.ts | 39 ++++++++++++++++++- .../strands-agents/src/model-activity.ts | 14 ++++++- 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index b9111512..c0959135 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -21,6 +21,7 @@ import { loadHistory, TestWorkflowEnvironment, } from '@temporalio/test-helpers'; +import { MockActivityEnvironment } from '@temporalio/testing'; import { bundleWorkflowCode, DefaultLogger, Worker } from '@temporalio/worker'; import { DeterminismViolationError } from '@temporalio/workflow'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; @@ -579,12 +580,14 @@ test('a history whose code depends on when a handler finishes still replays', as ); }); -test('invokeModel reconstructs Message instances from wire data', async (t) => { +test('invokeModel gives the model Message instances and an AbortSignal', async (t) => { // Strands content-block and Message classes have lossy `toJSON()` methods. // After Temporal's default payload converter serializes activity input, // the receiving side sees plain Bedrock-shape objects without the `type` // discriminator. ModelActivity must call `Message.fromMessageData` so the - // user's `Model.stream` receives real Strands types. + // user's `Model.stream` receives real Strands types. The agent's + // `cancelSignal` arrives as a plain object too, and model clients such as + // OpenAI's call its `addEventListener`. t.timeout(60_000); const { createWorker, executeWorkflow } = helpers(t); @@ -593,8 +596,10 @@ test('invokeModel reconstructs Message instances from wire data', async (t) => { // reachable from the assertion below. let receivedFirstMessage: unknown; let firstBlockIsTextBlock = false; + let receivedSignal: unknown; class AssertingStub extends StubModel { override async *stream(messages: Message[], options?: StreamOptions): AsyncIterable { + receivedSignal = options?.cancelSignal; if (receivedFirstMessage === undefined) { receivedFirstMessage = messages[0]; const first = messages[0]; @@ -618,6 +623,36 @@ test('invokeModel reconstructs Message instances from wire data', async (t) => { t.true(receivedFirstMessage instanceof Message, 'expected first message to be a Message instance'); t.true(firstBlockIsTextBlock, 'expected first content block to be a TextBlock instance'); + t.true(receivedSignal instanceof AbortSignal, 'expected cancelSignal to be an AbortSignal'); +}); + +test('cancelling the model activity aborts the model call', async (t) => { + let aborted = false; + let markStarted!: () => void; + const started = new Promise((resolve) => (markStarted = resolve)); + // Waits on its cancel signal, as a model client does while a request is in flight. + class WaitingStub extends StubModel { + override async *stream(messages: Message[], options?: StreamOptions): AsyncIterable { + await new Promise((resolve) => { + options?.cancelSignal?.addEventListener('abort', () => { + aborted = true; + resolve(); + }); + markStarted(); + }); + yield* super.stream(messages, options); + } + } + + const activity = new ModelActivity({ test: () => new WaitingStub([textTurn('late')]) }); + const env = new MockActivityEnvironment(); + // What the activity receives: the agent's AbortSignal after the payload converter. + const options = { cancelSignal: {} } as unknown as StreamOptions; + const run = env.run(activity.invokeModel, { modelName: 'test', messages: [], options }).catch(() => undefined); + await Promise.race([started, run]); + env.cancel(); + await run; + t.true(aborted, 'expected the model call to see the activity cancellation'); }); test('activityAsTool dispatches a registered activity from the agent loop', async (t) => { diff --git a/typescript/strands-agents/src/model-activity.ts b/typescript/strands-agents/src/model-activity.ts index e2ea78eb..f3ca9da8 100644 --- a/typescript/strands-agents/src/model-activity.ts +++ b/typescript/strands-agents/src/model-activity.ts @@ -1,5 +1,6 @@ import { Message, StateStore } from '@strands-agents/sdk'; import type { JSONValue, Model, ModelStreamEvent, StreamOptions } from '@strands-agents/sdk'; +import { Context } from '@temporalio/activity'; import { ApplicationFailure } from '@temporalio/common'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import type { Duration } from '@temporalio/common/lib/time'; @@ -31,6 +32,15 @@ function rebuildMessages(messages: unknown): Message[] { ); } +/** + * The stream options the model gets in the activity. Strands passes the agent's `cancelSignal`, an + * `AbortSignal`, which reaches the activity as a plain object, so give the model the activity's own + * cancellation signal instead: cancelling the activity then aborts the model call. + */ +function withActivityCancellation(options?: StreamOptions): StreamOptions { + return { ...options, cancelSignal: Context.current().cancellationSignal }; +} + /** * Input for the `invokeModel` activity. Mirrors `Model.stream(messages, options)` * with the chosen factory name carried alongside. @@ -195,7 +205,7 @@ export class ModelActivity { private async runModel(model: Model, input: InvokeModelInput): Promise { const messages = rebuildMessages(input.messages); const events: ModelStreamEvent[] = []; - for await (const event of model.stream(messages, input.options)) { + for await (const event of model.stream(messages, withActivityCancellation(input.options))) { events.push(event); } return events; @@ -209,7 +219,7 @@ export class ModelActivity { const topic = stream.topic(input.streamingTopic); const events: ModelStreamEvent[] = []; try { - for await (const event of model.stream(messages, input.options)) { + for await (const event of model.stream(messages, withActivityCancellation(input.options))) { events.push(event); topic.publish(event); } From b6093c22f139b40cd6313c04a7e282d00c879b61 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Fri, 30 Oct 2026 14:43:05 -0700 Subject: [PATCH 2/8] Address review findings --- typescript/strands-agents/README.md | 15 +++++ .../src/__tests__/test-strands.ts | 63 +++++++++++-------- .../src/__tests__/workflows/strands.ts | 31 ++++++++- .../strands-agents/src/model-activity.ts | 29 ++++++--- .../strands-agents/src/temporal-agent.ts | 4 ++ 5 files changed, 107 insertions(+), 35 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index c7176c31..8d8f24c3 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -126,6 +126,21 @@ Workflow code that catches these errors around `invoke()` by class sees the `App If your worker lists one of these errors in `workflowFailureErrorTypes`, the run used to fail with a retryable failure, so a workflow retry policy started another run. The plugin's failure is non-retryable, so that run is no longer retried. +## Cancellation + +The model gets the model activity's cancellation signal as `cancelSignal`. Cancelling the activity, through a cancellation scope or a workflow cancel, aborts the model call and ends the activity as cancelled. The worker learns of the cancellation through the activity's heartbeats, which the model activities send only when `activityOptions.heartbeatTimeout` is set: + +```ts +new TemporalAgent({ + activityOptions: { + startToCloseTimeout: '60 seconds', + heartbeatTimeout: '10 seconds', + }, +}); +``` + +Without a `heartbeatTimeout`, the model call runs to completion after its activity is cancelled. `agent.cancel()` in workflow code does not cancel the model activity. + ## Snapshots `TemporalAgent.takeSnapshot()` and `TemporalAgent.loadSnapshot()` throw. Temporal's event history already persists workflow state durably at a finer granularity than Strands snapshots, so calling either inside a workflow is redundant. diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index c0959135..7e7b655d 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -21,12 +21,10 @@ import { loadHistory, TestWorkflowEnvironment, } from '@temporalio/test-helpers'; -import { MockActivityEnvironment } from '@temporalio/testing'; import { bundleWorkflowCode, DefaultLogger, Worker } from '@temporalio/worker'; import { DeterminismViolationError } from '@temporalio/workflow'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import { STRANDS_MALFORMED_INTERRUPT_TYPE, StrandsPlugin } from '..'; -import { ModelActivity } from '../model-activity'; import { startLocalWithRetry } from './helpers/environment'; import { auditLog, @@ -43,6 +41,7 @@ import { activityToolAgent, failActivities, approveSignal, + cancellableModelCall, compressionAgent, defaultModelAgent, doneSignal, @@ -626,34 +625,44 @@ test('invokeModel gives the model Message instances and an AbortSignal', async ( t.true(receivedSignal instanceof AbortSignal, 'expected cancelSignal to be an AbortSignal'); }); -test('cancelling the model activity aborts the model call', async (t) => { - let aborted = false; - let markStarted!: () => void; - const started = new Promise((resolve) => (markStarted = resolve)); - // Waits on its cancel signal, as a model client does while a request is in flight. - class WaitingStub extends StubModel { - override async *stream(messages: Message[], options?: StreamOptions): AsyncIterable { - await new Promise((resolve) => { - options?.cancelSignal?.addEventListener('abort', () => { - aborted = true; - resolve(); +for (const streaming of [false, true]) { + const path = streaming ? 'invokeModelStreaming' : 'invokeModel'; + test(`cancelling the workflow aborts the model call in ${path} and cancels the workflow`, async (t) => { + t.timeout(60_000); + const { createWorker, startWorkflow } = helpers(t); + let receivedSignal: unknown; + let markStarted!: () => void; + const started = new Promise((resolve) => (markStarted = resolve)); + // Waits for its cancel signal, then fails as OpenAI's client does: an error named `Error`. + class AbortingStub extends StubModel { + override async *stream(messages: Message[], options?: StreamOptions): AsyncIterable { + receivedSignal = options?.cancelSignal; + const aborted = await new Promise((resolve) => { + options?.cancelSignal?.addEventListener('abort', () => resolve(true)); + markStarted(); }); - markStarted(); - }); - yield* super.stream(messages, options); + if (aborted) throw new Error('Request was aborted.'); + yield* super.stream(messages, options); + } } - } - const activity = new ModelActivity({ test: () => new WaitingStub([textTurn('late')]) }); - const env = new MockActivityEnvironment(); - // What the activity receives: the agent's AbortSignal after the payload converter. - const options = { cancelSignal: {} } as unknown as StreamOptions; - const run = env.run(activity.invokeModel, { modelName: 'test', messages: [], options }).catch(() => undefined); - await Promise.race([started, run]); - env.cancel(); - await run; - t.true(aborted, 'expected the model call to see the activity cancellation'); -}); + const worker = await createWorker({ + plugins: [new StrandsPlugin({ models: { test: () => new AbortingStub([]) } })], + }); + await worker.runUntil(async () => { + const handle = await startWorkflow(cancellableModelCall, { args: [streaming] }); + await started; + await handle.cancel(); + const err = await t.throwsAsync(handle.result(), { instanceOf: WorkflowFailedError }); + t.true(err!.cause instanceof CancelledFailure, `expected a CancelledFailure, got ${err!.cause}`); + const { events } = await handle.fetchHistory(); + t.true(events!.some((e) => e.activityTaskCanceledEventAttributes), 'expected ActivityTaskCanceled'); + t.false(events!.some((e) => e.activityTaskFailedEventAttributes), 'expected no ActivityTaskFailed'); + t.true(events!.some((e) => e.workflowExecutionCanceledEventAttributes), 'expected WorkflowExecutionCanceled'); + }); + t.true(receivedSignal instanceof AbortSignal && receivedSignal.aborted, 'expected the model signal to abort'); + }); +} test('activityAsTool dispatches a registered activity from the agent loop', async (t) => { const { createWorker, executeWorkflow } = helpers(t); diff --git a/typescript/strands-agents/src/__tests__/workflows/strands.ts b/typescript/strands-agents/src/__tests__/workflows/strands.ts index 7bbf6a79..9b37f296 100644 --- a/typescript/strands-agents/src/__tests__/workflows/strands.ts +++ b/typescript/strands-agents/src/__tests__/workflows/strands.ts @@ -12,6 +12,7 @@ import { type Message, } from '@strands-agents/sdk'; import { + ActivityCancellationType, ActivityFailure, allHandlersFinished, ApplicationFailure, @@ -25,7 +26,13 @@ import { type ActivityOptions, } from '@temporalio/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; -import { TemporalAgent, TemporalMCPClient, type TemporalAgentOptions, workflow as strandsWorkflow } from '../..'; +import { + TemporalAgent, + TemporalMCPClient, + TemporalModel, + type TemporalAgentOptions, + workflow as strandsWorkflow, +} from '../..'; export async function helloAgent(prompt: string): Promise { const agent = new TemporalAgent({ model: 'test', printer: false }); @@ -400,3 +407,25 @@ export async function streamingAgent(prompt: string): Promise { const result = await agent.invoke(prompt); return result.toString(); } + +/** + * Streams one model call whose activity heartbeats and waits for its cancellation to complete, so + * a workflow cancel reaches the model call and the workflow ends with the activity's result. + */ +export async function cancellableModelCall(streaming: boolean): Promise { + if (streaming) void new WorkflowStream(); + const model = new TemporalModel({ + modelName: 'test', + streamingTopic: streaming ? 'model' : undefined, + activityOptions: { + startToCloseTimeout: '30 seconds', + heartbeatTimeout: '5 seconds', + cancellationType: ActivityCancellationType.WAIT_CANCELLATION_COMPLETED, + }, + }); + const types: string[] = []; + for await (const event of model.stream([])) { + types.push(event.type); + } + return types; +} diff --git a/typescript/strands-agents/src/model-activity.ts b/typescript/strands-agents/src/model-activity.ts index f3ca9da8..ab28f506 100644 --- a/typescript/strands-agents/src/model-activity.ts +++ b/typescript/strands-agents/src/model-activity.ts @@ -33,12 +33,27 @@ function rebuildMessages(messages: unknown): Message[] { } /** - * The stream options the model gets in the activity. Strands passes the agent's `cancelSignal`, an - * `AbortSignal`, which reaches the activity as a plain object, so give the model the activity's own - * cancellation signal instead: cancelling the activity then aborts the model call. + * Run `model.stream` with the activity's own cancellation signal as `cancelSignal`. Strands passes + * the agent's `cancelSignal`, an `AbortSignal`, which reaches the activity as a plain object. + * + * Once the activity's cancellation reaches the worker, the signal aborts the model call. A remote + * activity learns of cancellation only through heartbeats, so this needs a `heartbeatTimeout` in + * the activity options. Model clients abort with their own errors (OpenAI's is named `Error`), + * which Temporal would record as a failure, so after cancellation the activity throws the signal's + * `CancelledFailure` instead. */ -function withActivityCancellation(options?: StreamOptions): StreamOptions { - return { ...options, cancelSignal: Context.current().cancellationSignal }; +async function* streamWithActivityCancellation( + model: Model, + messages: Message[], + options?: StreamOptions +): AsyncGenerator { + const signal = Context.current().cancellationSignal; + try { + yield* model.stream(messages, { ...options, cancelSignal: signal }); + } catch (err) { + if (signal.aborted) throw signal.reason; + throw err; + } } /** @@ -205,7 +220,7 @@ export class ModelActivity { private async runModel(model: Model, input: InvokeModelInput): Promise { const messages = rebuildMessages(input.messages); const events: ModelStreamEvent[] = []; - for await (const event of model.stream(messages, withActivityCancellation(input.options))) { + for await (const event of streamWithActivityCancellation(model, messages, input.options)) { events.push(event); } return events; @@ -219,7 +234,7 @@ export class ModelActivity { const topic = stream.topic(input.streamingTopic); const events: ModelStreamEvent[] = []; try { - for await (const event of model.stream(messages, withActivityCancellation(input.options))) { + for await (const event of streamWithActivityCancellation(model, messages, input.options)) { events.push(event); topic.publish(event); } diff --git a/typescript/strands-agents/src/temporal-agent.ts b/typescript/strands-agents/src/temporal-agent.ts index 3d520aa0..b9f9d885 100644 --- a/typescript/strands-agents/src/temporal-agent.ts +++ b/typescript/strands-agents/src/temporal-agent.ts @@ -55,6 +55,10 @@ export interface TemporalAgentOptions extends Omit Date: Fri, 30 Oct 2026 15:29:34 -0700 Subject: [PATCH 3/8] Address review findings --- typescript/strands-agents/README.md | 9 ++++++--- typescript/strands-agents/src/temporal-agent.ts | 6 +++++- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 8d8f24c3..63ed845b 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -128,19 +128,22 @@ If your worker lists one of these errors in `workflowFailureErrorTypes`, the run ## Cancellation -The model gets the model activity's cancellation signal as `cancelSignal`. Cancelling the activity, through a cancellation scope or a workflow cancel, aborts the model call and ends the activity as cancelled. The worker learns of the cancellation through the activity's heartbeats, which the model activities send only when `activityOptions.heartbeatTimeout` is set: +The model gets the model activity's cancellation signal as `cancelSignal`. Once the worker learns that the activity was cancelled, the signal aborts the model call and the activity ends as cancelled. The worker learns of the cancellation through the activity's heartbeats, which the model activities send only when `activityOptions.heartbeatTimeout` is set. Without a `heartbeatTimeout`, the model call runs to completion after its activity is cancelled. `agent.cancel()` in workflow code does not cancel the model activity. + +Through `TemporalAgent`, cancelling the workflow during a model call has a known limitation. Strands wraps the activity's cancellation in a `ModelError`, which fails the workflow task, so the workflow keeps retrying that task instead of ending as cancelled. With the default `TRY_CANCEL` cancellation type, the failed task also discards the request to cancel the activity, so the model call is never aborted. To abort the model call, also set `cancellationType` to `WAIT_CANCELLATION_COMPLETED`. The cancel request is then committed before Strands sees the result, but the workflow task still fails afterward: ```ts +import { ActivityCancellationType } from '@temporalio/workflow'; + new TemporalAgent({ activityOptions: { startToCloseTimeout: '60 seconds', heartbeatTimeout: '10 seconds', + cancellationType: ActivityCancellationType.WAIT_CANCELLATION_COMPLETED, }, }); ``` -Without a `heartbeatTimeout`, the model call runs to completion after its activity is cancelled. `agent.cancel()` in workflow code does not cancel the model activity. - ## Snapshots `TemporalAgent.takeSnapshot()` and `TemporalAgent.loadSnapshot()` throw. Temporal's event history already persists workflow state durably at a finer granularity than Strands snapshots, so calling either inside a workflow is redundant. diff --git a/typescript/strands-agents/src/temporal-agent.ts b/typescript/strands-agents/src/temporal-agent.ts index b9f9d885..6621c2fb 100644 --- a/typescript/strands-agents/src/temporal-agent.ts +++ b/typescript/strands-agents/src/temporal-agent.ts @@ -58,7 +58,11 @@ export interface TemporalAgentOptions extends Omit Date: Fri, 30 Oct 2026 15:56:28 -0700 Subject: [PATCH 4/8] Address review findings --- typescript/strands-agents/README.md | 4 ++-- typescript/strands-agents/src/model-activity.ts | 4 +++- typescript/strands-agents/src/temporal-agent.ts | 7 +++++-- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 63ed845b..e5268caf 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -128,9 +128,9 @@ If your worker lists one of these errors in `workflowFailureErrorTypes`, the run ## Cancellation -The model gets the model activity's cancellation signal as `cancelSignal`. Once the worker learns that the activity was cancelled, the signal aborts the model call and the activity ends as cancelled. The worker learns of the cancellation through the activity's heartbeats, which the model activities send only when `activityOptions.heartbeatTimeout` is set. Without a `heartbeatTimeout`, the model call runs to completion after its activity is cancelled. `agent.cancel()` in workflow code does not cancel the model activity. +The model gets the model activity's cancellation signal as `cancelSignal`. Once the worker learns that the activity was cancelled, the signal is aborted. Only a model that passes `cancelSignal` to its client stops the call; the model activity then ends as cancelled. In `@strands-agents/sdk` 1.19.0, `OpenAIModel` and `BedrockModel` do this, but `AnthropicModel`, `GoogleModel` and `VercelModel` ignore `cancelSignal`. With a model that ignores it, the provider call keeps running after cancellation, and the activity can still complete with the model's result. The worker learns of the cancellation through the activity's heartbeats, which the model activities send only when `activityOptions.heartbeatTimeout` is set. Without a `heartbeatTimeout`, the model call runs to completion after its activity is cancelled. `agent.cancel()` in workflow code does not cancel the model activity. -Through `TemporalAgent`, cancelling the workflow during a model call has a known limitation. Strands wraps the activity's cancellation in a `ModelError`, which fails the workflow task, so the workflow keeps retrying that task instead of ending as cancelled. With the default `TRY_CANCEL` cancellation type, the failed task also discards the request to cancel the activity, so the model call is never aborted. To abort the model call, also set `cancellationType` to `WAIT_CANCELLATION_COMPLETED`. The cancel request is then committed before Strands sees the result, but the workflow task still fails afterward: +Through `TemporalAgent`, cancelling the workflow during a model call has a known limitation. Strands wraps the activity's cancellation in a `ModelError`, which fails the workflow task, so the workflow keeps retrying that task instead of ending as cancelled. With the default `TRY_CANCEL` cancellation type, the failed task also discards the request to cancel the activity, so the model call is never aborted. To abort the model call (with a model that honors `cancelSignal`), also set `cancellationType` to `WAIT_CANCELLATION_COMPLETED`. The cancel request is then committed before Strands sees the result, but the workflow task still fails afterward: ```ts import { ActivityCancellationType } from '@temporalio/workflow'; diff --git a/typescript/strands-agents/src/model-activity.ts b/typescript/strands-agents/src/model-activity.ts index ab28f506..6c0d795c 100644 --- a/typescript/strands-agents/src/model-activity.ts +++ b/typescript/strands-agents/src/model-activity.ts @@ -36,7 +36,9 @@ function rebuildMessages(messages: unknown): Message[] { * Run `model.stream` with the activity's own cancellation signal as `cancelSignal`. Strands passes * the agent's `cancelSignal`, an `AbortSignal`, which reaches the activity as a plain object. * - * Once the activity's cancellation reaches the worker, the signal aborts the model call. A remote + * Once the activity's cancellation reaches the worker, the signal is aborted. That stops the model + * call only if the model passes `cancelSignal` to its client (Strands 1.19.0's `AnthropicModel` + * does not); otherwise the call runs on and the activity can complete with its result. A remote * activity learns of cancellation only through heartbeats, so this needs a `heartbeatTimeout` in * the activity options. Model clients abort with their own errors (OpenAI's is named `Error`), * which Temporal would record as a failure, so after cancellation the activity throws the signal's diff --git a/typescript/strands-agents/src/temporal-agent.ts b/typescript/strands-agents/src/temporal-agent.ts index 6621c2fb..5869f1be 100644 --- a/typescript/strands-agents/src/temporal-agent.ts +++ b/typescript/strands-agents/src/temporal-agent.ts @@ -57,8 +57,11 @@ export interface TemporalAgentOptions extends Omit Date: Fri, 30 Oct 2026 16:53:50 -0700 Subject: [PATCH 5/8] Address review findings --- typescript/strands-agents/README.md | 2 +- .../src/__tests__/test-strands.ts | 49 ++++++++++++++++++- .../src/__tests__/workflows/strands.ts | 14 ++++++ .../strands-agents/src/temporal-agent.ts | 7 +-- 4 files changed, 65 insertions(+), 7 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index e5268caf..a4313c32 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -130,7 +130,7 @@ If your worker lists one of these errors in `workflowFailureErrorTypes`, the run The model gets the model activity's cancellation signal as `cancelSignal`. Once the worker learns that the activity was cancelled, the signal is aborted. Only a model that passes `cancelSignal` to its client stops the call; the model activity then ends as cancelled. In `@strands-agents/sdk` 1.19.0, `OpenAIModel` and `BedrockModel` do this, but `AnthropicModel`, `GoogleModel` and `VercelModel` ignore `cancelSignal`. With a model that ignores it, the provider call keeps running after cancellation, and the activity can still complete with the model's result. The worker learns of the cancellation through the activity's heartbeats, which the model activities send only when `activityOptions.heartbeatTimeout` is set. Without a `heartbeatTimeout`, the model call runs to completion after its activity is cancelled. `agent.cancel()` in workflow code does not cancel the model activity. -Through `TemporalAgent`, cancelling the workflow during a model call has a known limitation. Strands wraps the activity's cancellation in a `ModelError`, which fails the workflow task, so the workflow keeps retrying that task instead of ending as cancelled. With the default `TRY_CANCEL` cancellation type, the failed task also discards the request to cancel the activity, so the model call is never aborted. To abort the model call (with a model that honors `cancelSignal`), also set `cancellationType` to `WAIT_CANCELLATION_COMPLETED`. The cancel request is then committed before Strands sees the result, but the workflow task still fails afterward: +Through `TemporalAgent`, cancelling the workflow during a model call cancels the model activity and ends the run as cancelled (see [Retries](#retries)). With the default `TRY_CANCEL` cancellation type, the workflow does not wait for the activity: the run is cancelled at once, and the model call is aborted once the worker learns of the cancellation. With `WAIT_CANCELLATION_COMPLETED`, the workflow waits until the model activity ends as cancelled, or completes with a model that ignores `cancelSignal`, before the run is cancelled: ```ts import { ActivityCancellationType } from '@temporalio/workflow'; diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index 7e7b655d..f039fd91 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -22,7 +22,7 @@ import { TestWorkflowEnvironment, } from '@temporalio/test-helpers'; import { bundleWorkflowCode, DefaultLogger, Worker } from '@temporalio/worker'; -import { DeterminismViolationError } from '@temporalio/workflow'; +import { ActivityCancellationType, DeterminismViolationError } from '@temporalio/workflow'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import { STRANDS_MALFORMED_INTERRUPT_TYPE, StrandsPlugin } from '..'; import { startLocalWithRetry } from './helpers/environment'; @@ -41,6 +41,7 @@ import { activityToolAgent, failActivities, approveSignal, + cancellableAgent, cancellableModelCall, compressionAgent, defaultModelAgent, @@ -664,6 +665,52 @@ for (const streaming of [false, true]) { }); } +for (const cancellationType of [ + ActivityCancellationType.TRY_CANCEL, + ActivityCancellationType.WAIT_CANCELLATION_COMPLETED, +]) { + test(`cancelling a TemporalAgent workflow with ${cancellationType} aborts the model call and cancels the workflow`, async (t) => { + t.timeout(60_000); + const { createWorker, startWorkflow } = helpers(t); + let markStarted!: () => void; + const started = new Promise((resolve) => (markStarted = resolve)); + let markAborted!: () => void; + const aborted = new Promise((resolve) => (markAborted = resolve)); + // Waits for its cancel signal, then fails as OpenAI's client does: an error named `Error`. + class AbortingStub extends StubModel { + // eslint-disable-next-line require-yield + override async *stream(_messages: Message[], options?: StreamOptions): AsyncIterable { + await new Promise((resolve) => { + options?.cancelSignal?.addEventListener('abort', () => resolve()); + markStarted(); + }); + markAborted(); + throw new Error('Request was aborted.'); + } + } + + const worker = await createWorker({ + plugins: [new StrandsPlugin({ models: { test: () => new AbortingStub([]) } })], + }); + await worker.runUntil(async () => { + const handle = await startWorkflow(cancellableAgent, { args: [cancellationType] }); + await started; + await handle.cancel(); + const err = await t.throwsAsync(handle.result(), { instanceOf: WorkflowFailedError }); + t.true(err!.cause instanceof CancelledFailure, `expected a CancelledFailure, got ${err!.cause}`); + // The model's signal aborts while the worker runs, not because it shuts down. + await aborted; + const { events } = await handle.fetchHistory(); + t.true(events!.some((e) => e.activityTaskCancelRequestedEventAttributes), 'expected ActivityTaskCancelRequested'); + t.false(events!.some((e) => e.workflowTaskFailedEventAttributes), 'expected no WorkflowTaskFailed'); + t.true(events!.some((e) => e.workflowExecutionCanceledEventAttributes), 'expected WorkflowExecutionCanceled'); + if (cancellationType === ActivityCancellationType.WAIT_CANCELLATION_COMPLETED) { + t.true(events!.some((e) => e.activityTaskCanceledEventAttributes), 'expected ActivityTaskCanceled'); + } + }); + }); +} + test('activityAsTool dispatches a registered activity from the agent loop', async (t) => { const { createWorker, executeWorkflow } = helpers(t); diff --git a/typescript/strands-agents/src/__tests__/workflows/strands.ts b/typescript/strands-agents/src/__tests__/workflows/strands.ts index 9b37f296..16c8494f 100644 --- a/typescript/strands-agents/src/__tests__/workflows/strands.ts +++ b/typescript/strands-agents/src/__tests__/workflows/strands.ts @@ -40,6 +40,20 @@ export async function helloAgent(prompt: string): Promise { return result.toString(); } +/** + * Runs one agent turn whose model activity heartbeats, with the given cancellation type, so a + * workflow cancel during the model call reaches the model. + */ +export async function cancellableAgent(cancellationType: ActivityCancellationType): Promise { + const agent = new TemporalAgent({ + model: 'test', + printer: false, + activityOptions: { startToCloseTimeout: '30 seconds', heartbeatTimeout: '5 seconds', cancellationType }, + }); + const result = await agent.invoke('hello'); + return result.toString(); +} + /** * Runs each prompt as a turn of one agent with Strands' proactive compression, then returns the * config the agent's model reports in workflow code. diff --git a/typescript/strands-agents/src/temporal-agent.ts b/typescript/strands-agents/src/temporal-agent.ts index 5869f1be..15c8a9f1 100644 --- a/typescript/strands-agents/src/temporal-agent.ts +++ b/typescript/strands-agents/src/temporal-agent.ts @@ -61,11 +61,8 @@ export interface TemporalAgentOptions extends Omit Date: Fri, 30 Oct 2026 17:04:02 -0700 Subject: [PATCH 6/8] Address review findings --- typescript/strands-agents/README.md | 2 +- .../src/__tests__/test-strands.ts | 43 +++++++++++++++++++ .../strands-agents/src/temporal-agent.ts | 4 +- 3 files changed, 47 insertions(+), 2 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index a4313c32..aedff096 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -130,7 +130,7 @@ If your worker lists one of these errors in `workflowFailureErrorTypes`, the run The model gets the model activity's cancellation signal as `cancelSignal`. Once the worker learns that the activity was cancelled, the signal is aborted. Only a model that passes `cancelSignal` to its client stops the call; the model activity then ends as cancelled. In `@strands-agents/sdk` 1.19.0, `OpenAIModel` and `BedrockModel` do this, but `AnthropicModel`, `GoogleModel` and `VercelModel` ignore `cancelSignal`. With a model that ignores it, the provider call keeps running after cancellation, and the activity can still complete with the model's result. The worker learns of the cancellation through the activity's heartbeats, which the model activities send only when `activityOptions.heartbeatTimeout` is set. Without a `heartbeatTimeout`, the model call runs to completion after its activity is cancelled. `agent.cancel()` in workflow code does not cancel the model activity. -Through `TemporalAgent`, cancelling the workflow during a model call cancels the model activity and ends the run as cancelled (see [Retries](#retries)). With the default `TRY_CANCEL` cancellation type, the workflow does not wait for the activity: the run is cancelled at once, and the model call is aborted once the worker learns of the cancellation. With `WAIT_CANCELLATION_COMPLETED`, the workflow waits until the model activity ends as cancelled, or completes with a model that ignores `cancelSignal`, before the run is cancelled: +Through `TemporalAgent`, cancelling the workflow during a model call cancels the model activity. With the default `TRY_CANCEL` cancellation type, the workflow does not wait for the activity: the run is cancelled at once (see [Retries](#retries)), and the model call is aborted once the worker learns of the cancellation. With `WAIT_CANCELLATION_COMPLETED`, the workflow waits for the model activity to end. If it ends as cancelled, so does the run. If it completes instead, as it can with a model that ignores `cancelSignal` or without a `heartbeatTimeout`, the agent gets the model's result and the turn carries on, so the run can complete normally rather than as cancelled: ```ts import { ActivityCancellationType } from '@temporalio/workflow'; diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index f039fd91..64311bd5 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -709,6 +709,49 @@ for (const cancellationType of [ } }); }); + + test(`cancelling a TemporalAgent workflow with ${cancellationType} while the model ignores its signal`, async (t) => { + t.timeout(60_000); + const { createWorker, startWorkflow } = helpers(t); + let markStarted!: () => void; + const started = new Promise((resolve) => (markStarted = resolve)); + let markAborted!: () => void; + const aborted = new Promise((resolve) => (markAborted = resolve)); + // Keeps going once its cancel signal aborts, as models that ignore `cancelSignal` do. + class IgnoringStub extends StubModel { + override async *stream(messages: Message[], options?: StreamOptions): AsyncIterable { + await new Promise((resolve) => { + options?.cancelSignal?.addEventListener('abort', () => resolve()); + markStarted(); + }); + markAborted(); + yield* super.stream(messages, options); + } + } + + const worker = await createWorker({ + plugins: [new StrandsPlugin({ models: { test: () => new IgnoringStub([textTurn('late result')]) } })], + }); + await worker.runUntil(async () => { + const handle = await startWorkflow(cancellableAgent, { args: [cancellationType] }); + await started; + await handle.cancel(); + if (cancellationType === ActivityCancellationType.WAIT_CANCELLATION_COMPLETED) { + // The workflow waits for the activity, which completes, so the run completes with its result. + t.is((await handle.result()).trim(), 'late result'); + const { events } = await handle.fetchHistory(); + t.true(events!.some((e) => e.activityTaskCompletedEventAttributes), 'expected ActivityTaskCompleted'); + t.true(events!.some((e) => e.workflowExecutionCompletedEventAttributes), 'expected WorkflowExecutionCompleted'); + t.false(events!.some((e) => e.workflowExecutionCanceledEventAttributes), 'expected no WorkflowExecutionCanceled'); + } else { + const err = await t.throwsAsync(handle.result(), { instanceOf: WorkflowFailedError }); + t.true(err!.cause instanceof CancelledFailure, `expected a CancelledFailure, got ${err!.cause}`); + const { events } = await handle.fetchHistory(); + t.true(events!.some((e) => e.workflowExecutionCanceledEventAttributes), 'expected WorkflowExecutionCanceled'); + } + await aborted; + }); + }); } test('activityAsTool dispatches a registered activity from the agent loop', async (t) => { diff --git a/typescript/strands-agents/src/temporal-agent.ts b/typescript/strands-agents/src/temporal-agent.ts index 15c8a9f1..197238cc 100644 --- a/typescript/strands-agents/src/temporal-agent.ts +++ b/typescript/strands-agents/src/temporal-agent.ts @@ -62,7 +62,9 @@ export interface TemporalAgentOptions extends Omit Date: Sat, 31 Oct 2026 01:50:49 -0700 Subject: [PATCH 7/8] Restore the ModelActivity import dropped by the rebase Other tests on main use ModelActivity. Rebasing the commit that removed this PR's own copy of the import also removed main's. --- typescript/strands-agents/src/__tests__/test-strands.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index 64311bd5..1fe70f38 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -25,6 +25,7 @@ import { bundleWorkflowCode, DefaultLogger, Worker } from '@temporalio/worker'; import { ActivityCancellationType, DeterminismViolationError } from '@temporalio/workflow'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import { STRANDS_MALFORMED_INTERRUPT_TYPE, StrandsPlugin } from '..'; +import { ModelActivity } from '../model-activity'; import { startLocalWithRetry } from './helpers/environment'; import { auditLog, From 4ded6619fb16356ecec1625c3e0e96635580e65d Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Sat, 31 Oct 2026 02:21:50 -0700 Subject: [PATCH 8/8] Address review findings --- typescript/strands-agents/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 4a9f6bfa..0918461c 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -95,7 +95,7 @@ Before its first model call, each `TemporalAgent` reads the model's `modelId` an A worker model that keeps the conversation server-side, such as `OpenAIModel` with `stateful: true`, gets the agent's `modelState`, such as the last response id, on each model call through `TemporalAgent`, and the agent clears its messages after each turn, as natively. As in Strands, such an agent cannot use a `conversationManager` or `contextManager`; because `TemporalAgent` learns that the model is stateful when it reads the config, it throws at the agent's first invocation instead of at construction. Catch that error, or the workflow task fails and is retried until you remove the manager and redeploy the worker. With no manager given, such an agent keeps Strands' default sliding window manager, which Strands picks before the agent reads the config, where natively it would use `NullConversationManager`. Through the plugin that manager has nothing to trim: the agent clears its messages after each turn, before the manager runs, and a context window overflow reaches the agent as a model error. A custom model that reports `stateful` but does not use `modelState` ran statelessly through earlier versions, with the agent keeping its messages; it now gets only each turn's new message, and an agent that gives it a manager now throws. Report `stateful` as false for such a model to keep the earlier behavior. -`OpenAIModel` calls through `TemporalAgent`, stateful or not, currently work only with `@strands-agents/sdk` releases before 1.15.0. From 1.15.0, Strands passes an `AbortSignal` in the model's stream options, the signal reaches the model activity as a plain object, and the OpenAI client fails with `signal.addEventListener is not a function` before it sends a request. Temporal retries that activity until its retry policy or timeout runs out. +`OpenAIModel` calls through `TemporalAgent`, stateful or not, work with current `@strands-agents/sdk` releases because the model activities give the model their own cancellation signal (see [Cancellation](#cancellation)). Earlier versions of this plugin passed the agent's signal, which reached the model activity as a plain object, so from `@strands-agents/sdk` 1.15.0 the OpenAI client failed with `signal.addEventListener is not a function` before it sent a request. During a rollout, such calls on a worker without this change fail and are retried until an upgraded worker takes them. Stateful model calls run as their own activities, `strandsInvokeStatefulModel` and `strandsInvokeStatefulModelStreaming`, chosen from the stateful flag that `strandsGetModelConfig` returns, so a run that uses a stateful model needs the workflow and the config activity from the same version. An earlier workflow that reads this version's config schedules `invokeModel`, and this version then fails to replay the run with a nondeterminism error. This version's workflow that reads an earlier config, which has no flag, keeps the stateless path for the rest of the run, and an upgraded worker fails each of its model calls with a non-retryable `StatefulModelOnStatelessPath` error. Changing a worker model between stateful and stateless, such as adding `stateful: true` to a deployed `OpenAIModel`, also applies only to new runs: an open run keeps the mode its config read recorded, and its model calls fail with `StatefulModelOnStatelessPath`, or with `StatelessModelOnStatefulPath` when the model became stateless. None of these runs recovers. The agent's `invoke()` throws an `ActivityFailure` whose cause has that type, so unless the workflow catches it, the run fails; reset a run you want to continue. Roll out with Worker Versioning, keeping each run's workflow and activity workers on one version, including any task queue set in `activityOptions`. Without it, stop the runs that use a stateful model, replace every worker on those task queues, and then start new runs.