From 930fc6351352330245d2b73b59de4a343963406b Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Thu, 29 Oct 2026 15:43:20 -0700 Subject: [PATCH 1/5] fix(strands): run stateful models with their server-side state (breaking) A worker model that keeps the conversation server-side, such as OpenAIModel with stateful: true, failed every model call through TemporalAgent. The agent's modelState store reached the model activity as a plain object, so the model could not read the last response id, and TemporalModel.stateful was always false. - The model config activity also returns the model's stateful flag, and TemporalModel.stateful reports it. - Stateful calls run as their own activities, strandsInvokeStatefulModel and strandsInvokeStatefulModelStreaming. They take a copy of the agent's model state and return the updated state with the events. Strands then clears the agent's messages after each turn, as natively. - TemporalAgent refuses a conversationManager or contextManager for a stateful model at its first invocation, as Strands does at construction. - README: stateful models, their rollout, and the model state in the continue-as-new example. Breaking: a run that used a stateful model on this version does not replay on an earlier version. --- typescript/strands-agents/README.md | 12 +- .../src/__tests__/test-strands.ts | 122 ++++++++++++++++++ .../src/__tests__/workflows/strands.ts | 82 +++++++++++- .../strands-agents/src/model-activity.ts | 57 ++++++-- typescript/strands-agents/src/plugin.ts | 17 ++- .../strands-agents/src/temporal-agent.ts | 24 +++- .../strands-agents/src/temporal-model.ts | 47 ++++++- 7 files changed, 340 insertions(+), 21 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index b994c147..8965fad2 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -95,6 +95,10 @@ Before its first model call, each `TemporalAgent` reads the model's `modelId` an The read is a new activity, so a run whose agent ran before you deploy this version fails to replay on it with a nondeterminism error. Let those runs finish first, or use Worker Versioning. +A worker model that keeps the conversation server-side, such as `OpenAIModel` with `stateful: true`, works as it does natively: each model call gets the agent's `modelState`, such as the last response id, and the agent clears its messages after each turn. 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. + +Stateful model calls run as their own activities, `strandsInvokeStatefulModel` and `strandsInvokeStatefulModelStreaming`. While you roll out this version, a worker running an earlier version does not register them, so it fails such a call and Temporal retries it until an upgraded worker takes it. Upgrade every worker on the task queue before a retry policy with a low `maximumAttempts` runs out, or route upgraded workflows to upgraded workers with Worker Versioning. + ## Retries `TemporalAgent` disables Strands' built-in `ModelRetryStrategy` so retries are handled exclusively by Temporal. Configure retries via `activityOptions.retry` on `TemporalAgent`, and on the activity options accepted by `workflow.activityAsTool`, `workflow.activityAsHook`, and `TemporalMCPClient`: @@ -234,24 +238,26 @@ const client = new Client({ /* ... */, plugins: [new StrandsPlugin({ models })] ## Continue-as-new -A chat-style workflow accumulates history with every turn and will eventually hit Temporal's per-workflow history limit. `workflowInfo().continueAsNewSuggested` flips true once the server decides history has grown large enough; check it after each turn and hand off to a fresh run, carrying `agent.messages` as input: +A chat-style workflow accumulates history with every turn and will eventually hit Temporal's per-workflow history limit. `workflowInfo().continueAsNewSuggested` flips true once the server decides history has grown large enough; check it after each turn and hand off to a fresh run, carrying `agent.messages` and `agent.modelState` as input. A stateful model clears the messages after each turn and keeps its context in the model state, such as the last response id; a run that carries only the messages starts the model's conversation over: ```ts import { defineSignal, setHandler, condition, continueAsNew, workflowInfo } from '@temporalio/workflow'; import { TemporalAgent } from '@temporalio/strands-agents'; -import type { Message } from '@strands-agents/sdk'; +import type { JSONValue, Message } from '@strands-agents/sdk'; export const userSays = defineSignal<[string]>('userSays'); export const endChat = defineSignal('endChat'); export interface ChatInput { messages?: Message[]; + modelState?: Record; } export async function chatWorkflow(input: ChatInput = {}): Promise { const agent = new TemporalAgent({ activityOptions: { startToCloseTimeout: '60 seconds' }, messages: input.messages, + modelState: input.modelState, }); const pending: string[] = []; @@ -264,7 +270,7 @@ export async function chatWorkflow(input: ChatInput = {}): Promise { if (done) return; await agent.invoke(pending.shift()!); if (workflowInfo().continueAsNewSuggested) { - await continueAsNew({ messages: agent.messages }); + await continueAsNew({ messages: agent.messages, modelState: agent.modelState.getAll() }); } } } diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index 4dcebdc3..e1de7412 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -4,6 +4,7 @@ import { McpClient, Message, Model, TextBlock } from '@strands-agents/sdk'; import type { BaseModelConfig, JSONValue, ModelStreamEvent, StreamOptions, Tool } from '@strands-agents/sdk'; import type { Transport } from '@modelcontextprotocol/sdk/shared/transport.js'; import { ApplicationFailure } from '@temporalio/common'; +import type { temporal } from '@temporalio/proto'; import type { BaseContext } from '@temporalio/test-helpers'; import { test as anyTest, @@ -16,6 +17,7 @@ import { bundleWorkflowCode, DefaultLogger, Worker } from '@temporalio/worker'; import { DeterminismViolationError } from '@temporalio/workflow'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import { StrandsPlugin } from '..'; +import { ModelActivity } from '../model-activity'; import { startLocalWithRetry } from './helpers/environment'; import { auditLog, auditTool, deleteState, deleteThing, getWeather, getWeatherCalls } from './activities/strands'; import { @@ -31,6 +33,8 @@ import { inWorkflowToolAgent, mcpAgent, retryConfigAgent, + statefulAgent, + statefulChatAgent, streamingAgent, structuredOutputAgent, } from './workflows/strands'; @@ -287,6 +291,124 @@ test('a history recorded before the model config read fails to replay', async (t t.regex(err!.message, /\[TMPRL1100\].*'invokeModel'.*'strandsGetModelConfig'/); }); +/** What each call of a {@link StatefulStub} got: the response id it chained to, and its messages. */ +type StatefulCall = { previousResponseId: JSONValue | undefined; messages: number }; + +/** Keeps the conversation "server-side", as `OpenAIModel` with `stateful: true` does. */ +class StatefulStub extends StubModel { + constructor( + turns: ModelStreamEvent[][], + private readonly calls: StatefulCall[] + ) { + super(turns); + } + + override get stateful(): boolean { + return true; + } + + override async *stream(messages: Message[], options?: StreamOptions): AsyncIterable { + this.calls.push({ previousResponseId: options?.modelState?.get('responseId'), messages: messages.length }); + options?.modelState?.set('responseId', `resp_${this.calls.length}`); + yield* super.stream(messages, options); + } +} + +const CHAINED_CALLS = [ + { previousResponseId: undefined, messages: 1 }, + { previousResponseId: 'resp_1', messages: 1 }, +]; + +function scheduledActivities(history: temporal.api.history.v1.IHistory): string[] { + return (history.events ?? []).flatMap((e) => e.activityTaskScheduledEventAttributes?.activityType?.name ?? []); +} + +test('a stateful worker model keeps the conversation server-side', async (t) => { + t.timeout(60_000); + const { createWorker, executeWorkflow, startWorkflow } = helpers(t); + const calls: StatefulCall[] = []; + const turns = [textTurn('one'), textTurn('two'), textTurn('three'), textTurn('four')]; + const plugin = (): StrandsPlugin => new StrandsPlugin({ models: { stateful: () => new StatefulStub(turns, calls) } }); + + const worker = await createWorker({ plugins: [plugin()], maxCachedWorkflows: 0 }); + await worker.runUntil(async () => { + for (const [mode, activityType] of [ + ['plain', 'strandsInvokeStatefulModel'], + ['streaming', 'strandsInvokeStatefulModelStreaming'], + ] as const) { + calls.length = 0; + const handle = await startWorkflow(statefulAgent, { args: [mode] }); + t.deepEqual(await handle.result(), [0, 0]); + t.deepEqual(calls, CHAINED_CALLS); + const history = await handle.fetchHistory(); + t.deepEqual(scheduledActivities(history), ['strandsGetModelConfig', activityType, activityType]); + await Worker.runReplayHistory({ workflowBundle: t.context.workflowBundle, plugins: [plugin()] }, history); + } + + t.is( + await executeWorkflow(statefulAgent, { args: ['manager'] }), + 'contextManager and conversationManager cannot be used with a stateful model. ' + + 'The model manages conversation state server-side.' + ); + }); +}); + +test('a stateful model call waits for a worker with stateful support', async (t) => { + t.timeout(60_000); + const { createWorker, startWorkflow, taskQueue } = helpers(t); + const calls: StatefulCall[] = []; + const plugin = new StrandsPlugin({ + models: { stateful: () => new StatefulStub([textTurn('one'), textTurn('two')], calls) }, + }); + // During a rolling upgrade, the config can come from an upgraded worker while a worker without + // stateful support still polls the queue: it has the config and the stateless model activities only. + const legacy = new ModelActivity({ stateful: () => new StatefulStub([], calls) }); + const modelsQueue = `${taskQueue}-models`; + + const worker = await createWorker({ plugins: [plugin] }); + await worker.runUntil(async () => { + const handle = await startWorkflow(statefulAgent, { args: ['models-queue'] }); + const legacyWorker = await createWorker({ + taskQueue: modelsQueue, + activities: { + strandsGetModelConfig: legacy.getModelConfig.bind(legacy), + invokeModel: legacy.invokeModel, + invokeModelStreaming: legacy.invokeModelStreaming, + }, + }); + const pending = await legacyWorker.runUntil(async () => { + for (;;) { + const [activity] = (await handle.describe()).raw.pendingActivities ?? []; + if (activity && (activity.attempt ?? 0) > 2) return activity; + await new Promise((resolve) => setTimeout(resolve, 100)); + } + }); + t.is(pending.activityType?.name, 'strandsInvokeStatefulModel'); + t.regex(pending.lastFailure?.message ?? '', /not registered/); + t.deepEqual(calls, []); + + const upgradedWorker = await createWorker({ taskQueue: modelsQueue, plugins: [plugin] }); + t.deepEqual(await upgradedWorker.runUntil(handle.result()), [0, 0]); + }); + t.deepEqual(calls, CHAINED_CALLS); +}); + +test("the README's continue-as-new handoff carries a stateful model's state", async (t) => { + t.timeout(60_000); + const { createWorker, executeWorkflow } = helpers(t); + const calls: StatefulCall[] = []; + const worker = await createWorker({ + plugins: [ + new StrandsPlugin({ models: { stateful: () => new StatefulStub([textTurn('one'), textTurn('two')], calls) } }), + ], + }); + + const messages = await worker.runUntil( + executeWorkflow(statefulChatAgent, { args: [{ prompts: ['turn one', 'turn two'] }] }) + ); + t.deepEqual({ messages, calls }, { messages: 0, calls: CHAINED_CALLS }); +}); + test('invokeModel reconstructs Message instances from wire data', async (t) => { // Strands content-block and Message classes have lossy `toJSON()` methods. // After Temporal's default payload converter serializes activity input, diff --git a/typescript/strands-agents/src/__tests__/workflows/strands.ts b/typescript/strands-agents/src/__tests__/workflows/strands.ts index b2730f4d..07d405ac 100644 --- a/typescript/strands-agents/src/__tests__/workflows/strands.ts +++ b/typescript/strands-agents/src/__tests__/workflows/strands.ts @@ -8,8 +8,18 @@ import { type BaseModelConfig, type InterruptResponseContent, type InterruptResponseContentData, + type JSONValue, + type Message, } from '@strands-agents/sdk'; -import { defineSignal, setHandler, condition, type ActivityFailure } from '@temporalio/workflow'; +import { + defineSignal, + setHandler, + condition, + continueAsNew, + workflowInfo, + type ActivityFailure, + type ActivityOptions, +} from '@temporalio/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; import { TemporalAgent, TemporalMCPClient, type TemporalAgentOptions, workflow as strandsWorkflow } from '../..'; @@ -52,6 +62,76 @@ export async function retryConfigAgent(prompt: string): Promise<{ turns: string[ return { turns, config: agent.model.getConfig() }; } +// Retry a failed model activity quickly, and not forever. +const STATEFUL_ACTIVITY_OPTIONS: ActivityOptions = { + startToCloseTimeout: '15 seconds', + retry: { initialInterval: '100 milliseconds', backoffCoefficient: 1, maximumAttempts: 50 }, +}; + +/** + * Runs two turns on one agent with the `stateful` worker model and returns the agent's message + * count after each turn, or the error that ended the run. `models-queue` runs the model activities + * on the `-models` task queue. + */ +export async function statefulAgent( + mode: 'plain' | 'streaming' | 'manager' | 'models-queue' +): Promise { + void new WorkflowStream(); + const agent = new TemporalAgent({ + model: 'stateful', + printer: false, + activityOptions: { + ...STATEFUL_ACTIVITY_OPTIONS, + ...(mode === 'models-queue' ? { taskQueue: `${workflowInfo().taskQueue}-models` } : {}), + }, + ...(mode === 'streaming' ? { streamingTopic: 'model' } : {}), + ...(mode === 'manager' ? { conversationManager: new SlidingWindowConversationManager() } : {}), + }); + const messagesAfterTurn: number[] = []; + try { + for (const prompt of ['turn one', 'turn two']) { + await agent.invoke(prompt); + messagesAfterTurn.push(agent.messages.length); + } + } catch (err) { + return (err as Error).message; + } + return messagesAfterTurn; +} + +export interface StatefulChatInput { + prompts: string[]; + messages?: Message[]; + modelState?: Record; +} + +/** + * The README's continue-as-new handoff with the `stateful` worker model: one turn per run. Returns + * the last run's message count, or the error that ended the turn. + */ +export async function statefulChatAgent(input: StatefulChatInput): Promise { + const agent = new TemporalAgent({ + model: 'stateful', + printer: false, + activityOptions: STATEFUL_ACTIVITY_OPTIONS, + messages: input.messages, + modelState: input.modelState, + }); + try { + await agent.invoke(input.prompts[0]!); + } catch (err) { + return (err as Error).message; + } + if (input.prompts.length > 1) { + await continueAsNew({ + prompts: input.prompts.slice(1), + messages: agent.messages, + modelState: agent.modelState.getAll(), + }); + } + return agent.messages.length; +} + /** * Same as {@link helloAgent} but omits `model:` so the plugin's single-factory * default kicks in. Exercises the fallback that picks the only registered diff --git a/typescript/strands-agents/src/model-activity.ts b/typescript/strands-agents/src/model-activity.ts index 10322119..103eab5b 100644 --- a/typescript/strands-agents/src/model-activity.ts +++ b/typescript/strands-agents/src/model-activity.ts @@ -1,5 +1,5 @@ -import { Message } from '@strands-agents/sdk'; -import type { Model, ModelStreamEvent, StreamOptions } from '@strands-agents/sdk'; +import { Message, StateStore } from '@strands-agents/sdk'; +import type { JSONValue, Model, ModelStreamEvent, StreamOptions } from '@strands-agents/sdk'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import type { Duration } from '@temporalio/common/lib/time'; import { autoHeartbeat } from './heartbeat'; @@ -49,6 +49,25 @@ export interface InvokeModelStreamingInput extends InvokeModelInput { streamingBatchInterval: Duration; } +/** + * Input for the stateful model activities: a copy of the agent's model state, such as the last + * response id, which the model reads and updates. + */ +export interface InvokeStatefulModelInput extends InvokeModelInput { + modelState: Record; +} + +/** Input for the streaming stateful model activity. */ +export interface InvokeStatefulModelStreamingInput extends InvokeModelStreamingInput { + modelState: Record; +} + +/** A stateful model's events and its updated model state. */ +export interface InvokeStatefulModelResult { + events: ModelStreamEvent[]; + modelState: Record; +} + /** * Input for the model config activity: the factory name, as in {@link InvokeModelInput}. */ @@ -57,13 +76,15 @@ export interface GetModelConfigInput { } /** - * The parts of the worker model's config that Strands reads in workflow code: the model id for - * traces, and the context window for proactive compression. Nothing else is returned, so provider - * settings such as credentials never reach the workflow history. + * The parts of the worker model that Strands reads in workflow code: the model id for traces, the + * context window for proactive compression, and whether the model keeps the conversation + * server-side. Nothing else is returned, so provider settings such as credentials never reach the + * workflow history. */ export interface ModelConfigInfo { modelId?: string; contextWindowLimit?: number; + stateful?: boolean; } /** @@ -74,7 +95,7 @@ export interface ModelConfigInfo { * default-Bedrock fallback; user-supplied factories force `model: 'name'` * on every {@link TemporalAgent}. * - * Both activities heartbeat at half the configured `heartbeatTimeout` via + * The model activities heartbeat at half the configured `heartbeatTimeout` via * {@link autoHeartbeat} so long model calls don't stall the worker's view * of activity liveness. */ @@ -111,8 +132,9 @@ export class ModelActivity { } async getModelConfig(input: GetModelConfigInput): Promise { - const { modelId, contextWindowLimit } = this.getModel(input.modelName).getConfig(); - return { modelId, contextWindowLimit }; + const model = this.getModel(input.modelName); + const { modelId, contextWindowLimit } = model.getConfig(); + return { modelId, contextWindowLimit, stateful: model.stateful }; } async invokeModel(input: InvokeModelInput): Promise { @@ -143,4 +165,23 @@ export class ModelActivity { } return events; } + + // Stateful calls use their own activity types. A worker without stateful support does not + // register them, so it cannot take such a call, drop the model state and return no updated state + // while the agent clears its messages: the call fails there and is retried until a worker with + // stateful support takes it. + + /** Run a stateful model with the agent's model state; return its events and the updated state. */ + async invokeStatefulModel(input: InvokeStatefulModelInput): Promise { + const modelState = new StateStore(input.modelState); + const events = await this.invokeModel({ ...input, options: { ...input.options, modelState } }); + return { events, modelState: modelState.getAll() }; + } + + /** As {@link invokeStatefulModel}, publishing each event as {@link invokeModelStreaming} does. */ + async invokeStatefulModelStreaming(input: InvokeStatefulModelStreamingInput): Promise { + const modelState = new StateStore(input.modelState); + const events = await this.invokeModelStreaming({ ...input, options: { ...input.options, modelState } }); + return { events, modelState: modelState.getAll() }; + } } diff --git a/typescript/strands-agents/src/plugin.ts b/typescript/strands-agents/src/plugin.ts index c05546ca..dc12eb7d 100644 --- a/typescript/strands-agents/src/plugin.ts +++ b/typescript/strands-agents/src/plugin.ts @@ -5,8 +5,18 @@ import { msOptionalToNumber } from '@temporalio/common/lib/time'; import type { BundleOptions } from '@temporalio/worker'; import { SimplePlugin } from '@temporalio/plugin'; import { ModelActivity } from './model-activity'; -import type { GetModelConfigInput, InvokeModelInput, InvokeModelStreamingInput } from './model-activity'; -import { GET_MODEL_CONFIG_ACTIVITY } from './temporal-model'; +import type { + GetModelConfigInput, + InvokeModelInput, + InvokeModelStreamingInput, + InvokeStatefulModelInput, + InvokeStatefulModelStreamingInput, +} from './model-activity'; +import { + GET_MODEL_CONFIG_ACTIVITY, + INVOKE_STATEFUL_MODEL_ACTIVITY, + INVOKE_STATEFUL_MODEL_STREAMING_ACTIVITY, +} from './temporal-model'; import { buildCallToolActivity, buildListToolsActivity, @@ -79,6 +89,9 @@ export class StrandsPlugin extends SimplePlugin { const activities: Record unknown> = { invokeModel: (input: InvokeModelInput) => modelActivity.invokeModel(input), invokeModelStreaming: (input: InvokeModelStreamingInput) => modelActivity.invokeModelStreaming(input), + [INVOKE_STATEFUL_MODEL_ACTIVITY]: (input: InvokeStatefulModelInput) => modelActivity.invokeStatefulModel(input), + [INVOKE_STATEFUL_MODEL_STREAMING_ACTIVITY]: (input: InvokeStatefulModelStreamingInput) => + modelActivity.invokeStatefulModelStreaming(input), [GET_MODEL_CONFIG_ACTIVITY]: (input: GetModelConfigInput) => modelActivity.getModelConfig(input), }; diff --git a/typescript/strands-agents/src/temporal-agent.ts b/typescript/strands-agents/src/temporal-agent.ts index a5f848bf..559d2579 100644 --- a/typescript/strands-agents/src/temporal-agent.ts +++ b/typescript/strands-agents/src/temporal-agent.ts @@ -11,6 +11,11 @@ const SNAPSHOT_DISABLED = 'granularity than Strands snapshots. Remove the snapshot call and rely on ' + "Temporal's durable execution instead."; +// Strands' own message, which its Agent throws at construction for a stateful model. +const STATEFUL_MANAGER_REFUSED = + 'contextManager and conversationManager cannot be used with a stateful model. ' + + 'The model manages conversation state server-side.'; + const RETRY_STRATEGY_DISABLED = 'TemporalAgent disables Strands retries; configure retries via retryPolicy ' + 'in the model activity options on TemporalAgent, and on the activity options ' + @@ -51,6 +56,8 @@ export interface TemporalAgentOptions extends Omit { const loading = this.model instanceof TemporalModel ? this.model.loadConfig() : undefined; - return loading ? loading.then(() => super.initialize()) : super.initialize(); + return loading ? loading.then(() => this.initializeWithConfig()) : this.initializeWithConfig(); + } + + private initializeWithConfig(): Promise { + // Strands' Agent refuses these managers for a stateful model at construction, before + // TemporalAgent can read the worker model's stateful flag, so refuse them here. + if (this.model.stateful && this.managerGiven) { + return Promise.reject(new Error(STATEFUL_MANAGER_REFUSED)); + } + return super.initialize(); } override takeSnapshot(): never { diff --git a/typescript/strands-agents/src/temporal-model.ts b/typescript/strands-agents/src/temporal-model.ts index 8c55374e..f3df143b 100644 --- a/typescript/strands-agents/src/temporal-model.ts +++ b/typescript/strands-agents/src/temporal-model.ts @@ -7,12 +7,19 @@ import type { GetModelConfigInput, InvokeModelInput, InvokeModelStreamingInput, + InvokeStatefulModelInput, + InvokeStatefulModelResult, + InvokeStatefulModelStreamingInput, ModelConfigInfo, } from './model-activity'; /** Name of the activity that returns the worker model's {@link ModelConfigInfo}. */ export const GET_MODEL_CONFIG_ACTIVITY = 'strandsGetModelConfig'; +/** Names of the activities that run a stateful model with the agent's model state. */ +export const INVOKE_STATEFUL_MODEL_ACTIVITY = 'strandsInvokeStatefulModel'; +export const INVOKE_STATEFUL_MODEL_STREAMING_ACTIVITY = 'strandsInvokeStatefulModelStreaming'; + /** * Options for {@link TemporalModel}. * @@ -43,6 +50,7 @@ export class TemporalModel extends Model { private readonly streamingTopic?: string; private readonly streamingBatchInterval: Duration; private config: ModelConfigInfo = {}; + private isStateful = false; private configLoaded = false; constructor(options: TemporalModelOptions = {}) { @@ -65,10 +73,15 @@ export class TemporalModel extends Model { return { ...this.config }; } + /** Whether the worker model keeps the conversation server-side, once {@link loadConfig} has read it. */ + override get stateful(): boolean { + return this.isStateful; + } + /** - * Read the worker model's id and context window through an activity, once per instance, so that - * Strands' traces and proactive compression use the real model's values. {@link TemporalAgent} - * calls this before Strands initializes the agent. + * Read the worker model's id, context window and stateful flag through an activity, once per + * instance, so that Strands' traces, proactive compression and stateful handling use the real + * model's values. {@link TemporalAgent} calls this before Strands initializes the agent. * * Returns `undefined`, without awaiting anything, when the config was already read, so callers * can go on synchronously. @@ -81,8 +94,10 @@ export class TemporalModel extends Model { startToCloseTimeout: '10 minutes', ...this.activityOptions, }); - return activities[GET_MODEL_CONFIG_ACTIVITY]({ modelName: this.modelName }).then((config) => { + return activities[GET_MODEL_CONFIG_ACTIVITY]({ modelName: this.modelName }).then(({ stateful, ...config }) => { this.config = config; + // A result recorded before the flag existed has none: the model ran as stateless then. + this.isStateful = stateful === true; this.configLoaded = true; }); } @@ -91,11 +106,35 @@ export class TemporalModel extends Model { const proxiedActivities = workflow.proxyActivities<{ invokeModel(input: InvokeModelInput): Promise; invokeModelStreaming(input: InvokeModelStreamingInput): Promise; + [INVOKE_STATEFUL_MODEL_ACTIVITY](input: InvokeStatefulModelInput): Promise; + [INVOKE_STATEFUL_MODEL_STREAMING_ACTIVITY]( + input: InvokeStatefulModelStreamingInput + ): Promise; }>({ startToCloseTimeout: '10 minutes', ...this.activityOptions, }); + if (this.isStateful) { + // The agent's model state is a live store; send a copy of its values, and copy the updated + // values back, as the model would have written them natively. + const { modelState, ...rest } = options ?? {}; + const input = { modelName: this.modelName, messages, options: rest, modelState: modelState?.getAll() ?? {} }; + const result = this.streamingTopic + ? await proxiedActivities[INVOKE_STATEFUL_MODEL_STREAMING_ACTIVITY]({ + ...input, + streamingTopic: this.streamingTopic, + streamingBatchInterval: this.streamingBatchInterval, + }) + : await proxiedActivities[INVOKE_STATEFUL_MODEL_ACTIVITY](input); + if (modelState) { + modelState.clear(); + for (const [key, value] of Object.entries(result.modelState)) modelState.set(key, value); + } + yield* result.events; + return; + } + const events = this.streamingTopic ? await proxiedActivities.invokeModelStreaming({ modelName: this.modelName, From f6573c5058273fcb1ec00ec16943808a9af469ef Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Thu, 29 Oct 2026 19:12:36 -0700 Subject: [PATCH 2/5] Address review findings --- typescript/strands-agents/README.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 8965fad2..6874d7aa 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -95,7 +95,9 @@ Before its first model call, each `TemporalAgent` reads the model's `modelId` an The read is a new activity, so a run whose agent ran before you deploy this version fails to replay on it with a nondeterminism error. Let those runs finish first, or use Worker Versioning. -A worker model that keeps the conversation server-side, such as `OpenAIModel` with `stateful: true`, works as it does natively: each model call gets the agent's `modelState`, such as the last response id, and the agent clears its messages after each turn. 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. +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, 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. + +`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. Stateful model calls run as their own activities, `strandsInvokeStatefulModel` and `strandsInvokeStatefulModelStreaming`. While you roll out this version, a worker running an earlier version does not register them, so it fails such a call and Temporal retries it until an upgraded worker takes it. Upgrade every worker on the task queue before a retry policy with a low `maximumAttempts` runs out, or route upgraded workflows to upgraded workers with Worker Versioning. From a69c4d8430b6c7f3bb69c0b5233522b67b86f108 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Fri, 30 Oct 2026 00:33:01 -0700 Subject: [PATCH 3/5] docs(strands): require one version per run for stateful models The stateful path is chosen from the flag the config activity returns, so a run needs its workflow and config activity from the same version. Say so in the README, with the rollout that keeps them together, and what changes for a custom model that reports stateful but ignores modelState. Correct the comment on the test whose activity worker lacks the stateful activities. --- typescript/strands-agents/README.md | 4 ++-- typescript/strands-agents/src/__tests__/test-strands.ts | 5 +++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 6874d7aa..96a83cf7 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -95,11 +95,11 @@ Before its first model call, each `TemporalAgent` reads the model's `modelId` an The read is a new activity, so a run whose agent ran before you deploy this version fails to replay on it with a nondeterminism error. Let those runs finish first, or use Worker Versioning. -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, 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. +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, 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. 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. -Stateful model calls run as their own activities, `strandsInvokeStatefulModel` and `strandsInvokeStatefulModelStreaming`. While you roll out this version, a worker running an earlier version does not register them, so it fails such a call and Temporal retries it until an upgraded worker takes it. Upgrade every worker on the task queue before a retry policy with a low `maximumAttempts` runs out, or route upgraded workflows to upgraded workers with Worker Versioning. +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: a model that reads `modelState`, such as `OpenAIModel`, keeps failing after every worker is upgraded. Retries repair neither case. 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. ## Retries diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index e1de7412..47878a6e 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -360,8 +360,9 @@ test('a stateful model call waits for a worker with stateful support', async (t) const plugin = new StrandsPlugin({ models: { stateful: () => new StatefulStub([textTurn('one'), textTurn('two')], calls) }, }); - // During a rolling upgrade, the config can come from an upgraded worker while a worker without - // stateful support still polls the queue: it has the config and the stateless model activities only. + // An activity worker that registers only the stateless model activities, behind this version's + // workflow and config. This is not a mixed-version rollout: an earlier workflow or config activity + // breaks the run, as the README says. const legacy = new ModelActivity({ stateful: () => new StatefulStub([], calls) }); const modelsQueue = `${taskQueue}-models`; From f6fac5113e5678fddc57681eee0a8ace47d64eb5 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Fri, 30 Oct 2026 01:10:13 -0700 Subject: [PATCH 4/5] fix(strands): fail a stateful model's call that lost its model state A run whose config read had no stateful flag, from a worker without stateful support, stays on the stateless path, and a model that reads modelState then failed every attempt with a retryable error, forever. invokeModel and invokeModelStreaming now fail such a call at once with a non-retryable StatefulModelOnStatelessPath error that names the cause. The stateful activities run the model through private helpers instead, and all four model activities are bound and heartbeat once. Also from review: - stream() builds the activity input once. - README: stateful models work through TemporalAgent; catch the manager refusal or the workflow task retries; the default manager stays installed and has nothing to trim; the rollout paragraph names the new error. - TemporalModel's doc says to call loadConfig() with a plain Agent. - A history recorded on main, replayed here, pins that a config without the flag reads as stateless. - The rollout test's poll loop stops with a message when the run ends. --- typescript/strands-agents/README.md | 4 +- .../histories/stateful-before-flag.json | 250 ++++++++++++++++++ .../src/__tests__/test-strands.ts | 61 ++++- .../strands-agents/src/model-activity.ts | 43 ++- .../strands-agents/src/temporal-model.ts | 57 ++-- 5 files changed, 373 insertions(+), 42 deletions(-) create mode 100644 typescript/strands-agents/src/__tests__/histories/stateful-before-flag.json diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 96a83cf7..00973f90 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -95,11 +95,11 @@ Before its first model call, each `TemporalAgent` reads the model's `modelId` an The read is a new activity, so a run whose agent ran before you deploy this version fails to replay on it with a nondeterminism error. Let those runs finish first, or use Worker Versioning. -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, 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. 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. +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. -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: a model that reads `modelState`, such as `OpenAIModel`, keeps failing after every worker is upgraded. Retries repair neither case. 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. +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. Neither run recovers: reset or terminate it. 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. ## Retries diff --git a/typescript/strands-agents/src/__tests__/histories/stateful-before-flag.json b/typescript/strands-agents/src/__tests__/histories/stateful-before-flag.json new file mode 100644 index 00000000..ec908fca --- /dev/null +++ b/typescript/strands-agents/src/__tests__/histories/stateful-before-flag.json @@ -0,0 +1,250 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-10-30T07:43:24.163131175Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048587", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "statefulAgent" + }, + "taskQueue": { + "name": "record-a-stateful-run-on-main", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "InBsYWluIg==" + } + ] + }, + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "01a18bc3-7203-71fa-8179-be473da5e3cb", + "identity": "119@a529f6a35480", + "firstExecutionRunId": "01a18bc3-7203-71fa-8179-be473da5e3cb", + "attempt": 1, + "firstWorkflowTaskBackoff": "0s", + "header": {}, + "workflowId": "ead9880d-031a-4bb5-a558-fe6da5afedeb" + } + }, + { + "eventId": "2", + "eventTime": "2026-10-30T07:43:24.165304134Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048588", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "record-a-stateful-run-on-main", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-10-30T07:43:24.244065967Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048593", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "119@a529f6a35480", + "requestId": "aa9e4dbb-6a2b-47d5-b117-2a477445069b", + "historySizeBytes": "329", + "workerVersion": { + "buildId": "@temporalio/worker@1.24.0+935bcefedc3349fb2c2daaa8be302fc89c585e57f7f410c833f142f5b7ed6955" + } + } + }, + { + "eventId": "4", + "eventTime": "2026-10-30T07:43:25.214269551Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048598", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "3", + "identity": "119@a529f6a35480", + "workerVersion": { + "buildId": "@temporalio/worker@1.24.0+935bcefedc3349fb2c2daaa8be302fc89c585e57f7f410c833f142f5b7ed6955" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 2, + 3, + 1 + ], + "sdkName": "temporal-typescript", + "sdkVersion": "1.24.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "5", + "eventTime": "2026-10-30T07:43:25.215057134Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED", + "taskId": "1048599", + "activityTaskScheduledEventAttributes": { + "activityId": "1", + "activityType": { + "name": "strandsGetModelConfig" + }, + "taskQueue": { + "name": "record-a-stateful-run-on-main", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "header": {}, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "eyJtb2RlbE5hbWUiOiJzdGF0ZWZ1bCJ9" + } + ] + }, + "scheduleToCloseTimeout": "0s", + "scheduleToStartTimeout": "0s", + "startToCloseTimeout": "15s", + "heartbeatTimeout": "0s", + "workflowTaskCompletedEventId": "4", + "retryPolicy": { + "initialInterval": "0.100s", + "backoffCoefficient": 1, + "maximumInterval": "10s", + "maximumAttempts": 50 + }, + "useWorkflowBuildId": true + } + }, + { + "eventId": "6", + "eventTime": "2026-10-30T07:43:25.215569968Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_STARTED", + "taskId": "1048603", + "activityTaskStartedEventAttributes": { + "scheduledEventId": "5", + "identity": "119@a529f6a35480", + "requestId": "66ea76a3-d045-44e5-86a8-ca7b0eb02891", + "attempt": 1, + "workerVersion": { + "buildId": "@temporalio/worker@1.24.0+935bcefedc3349fb2c2daaa8be302fc89c585e57f7f410c833f142f5b7ed6955" + } + } + }, + { + "eventId": "7", + "eventTime": "2026-10-30T07:43:25.246124134Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_COMPLETED", + "taskId": "1048604", + "activityTaskCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "e30=" + } + ] + }, + "scheduledEventId": "5", + "startedEventId": "6", + "identity": "119@a529f6a35480" + } + }, + { + "eventId": "8", + "eventTime": "2026-10-30T07:43:25.246205759Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048605", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "119@a529f6a35480-d1063acad8154749a5ad79f09925f0d4", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "record-a-stateful-run-on-main" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "9", + "eventTime": "2026-10-30T07:43:25.249141176Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048609", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "8", + "identity": "119@a529f6a35480", + "requestId": "855b5a9d-8e00-4ace-8dd1-acd30f9b6a3d", + "historySizeBytes": "1255", + "workerVersion": { + "buildId": "@temporalio/worker@1.24.0+935bcefedc3349fb2c2daaa8be302fc89c585e57f7f410c833f142f5b7ed6955" + } + } + }, + { + "eventId": "10", + "eventTime": "2026-10-30T07:43:25.285266968Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048614", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "8", + "startedEventId": "9", + "identity": "119@a529f6a35480", + "workerVersion": { + "buildId": "@temporalio/worker@1.24.0+935bcefedc3349fb2c2daaa8be302fc89c585e57f7f410c833f142f5b7ed6955" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "11", + "eventTime": "2026-10-30T07:43:25.285404093Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED", + "taskId": "1048615", + "activityTaskScheduledEventAttributes": { + "activityId": "2", + "activityType": { + "name": "invokeModel" + }, + "taskQueue": { + "name": "record-a-stateful-run-on-main", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "header": {}, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "eyJtb2RlbE5hbWUiOiJzdGF0ZWZ1bCIsIm1lc3NhZ2VzIjpbeyJyb2xlIjoidXNlciIsImNvbnRlbnQiOlt7InRleHQiOiJ0dXJuIG9uZSJ9XSwidHJhY2tpbmdJZCI6IjY5NWY3Y2MxLTk1NzctNDJhNC1iNTNhLWIwNDY1ZWU0ZjJiZiJ9XSwib3B0aW9ucyI6eyJjYW5jZWxTaWduYWwiOnt9LCJ0b29sU3BlY3MiOltdLCJtb2RlbFN0YXRlIjp7Il9zdGF0ZSI6e319fX0=" + } + ] + }, + "scheduleToCloseTimeout": "0s", + "scheduleToStartTimeout": "0s", + "startToCloseTimeout": "15s", + "heartbeatTimeout": "0s", + "workflowTaskCompletedEventId": "10", + "retryPolicy": { + "initialInterval": "0.100s", + "backoffCoefficient": 1, + "maximumInterval": "10s", + "maximumAttempts": 50 + }, + "useWorkflowBuildId": true + } + } + ] +} \ No newline at end of file diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index 47878a6e..0dc36d6e 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -291,6 +291,21 @@ test('a history recorded before the model config read fails to replay', async (t t.regex(err!.message, /\[TMPRL1100\].*'invokeModel'.*'strandsGetModelConfig'/); }); +test('a stateful run recorded before the stateful flag replays on the stateless path', async (t) => { + // Recorded with `statefulAgent` on 3af00d9, whose config activity returns no `stateful` flag: the + // run scheduled `invokeModel`, where a config with the flag schedules `strandsInvokeStatefulModel`. + const history = await loadHistory(path.resolve(__dirname, '../../src/__tests__/histories/stateful-before-flag.json')); + await t.notThrowsAsync( + Worker.runReplayHistory( + { + workflowBundle: t.context.workflowBundle, + plugins: [new StrandsPlugin({ models: { stateful: () => new StatefulStub([], []) } })], + }, + history + ) + ); +}); + /** What each call of a {@link StatefulStub} got: the response id it chained to, and its messages. */ type StatefulCall = { previousResponseId: JSONValue | undefined; messages: number }; @@ -378,9 +393,16 @@ test('a stateful model call waits for a worker with stateful support', async (t) }, }); const pending = await legacyWorker.runUntil(async () => { - for (;;) { - const [activity] = (await handle.describe()).raw.pendingActivities ?? []; + for (let polls = 0; ; polls++) { + const description = await handle.describe(); + const [activity] = description.raw.pendingActivities ?? []; if (activity && (activity.attempt ?? 0) > 2) return activity; + if (description.status.name !== 'RUNNING' || polls === 300) { + throw new Error( + `no retried model activity: run ${description.status.name}, pending ` + + JSON.stringify(description.raw.pendingActivities) + ); + } await new Promise((resolve) => setTimeout(resolve, 100)); } }); @@ -394,6 +416,41 @@ test('a stateful model call waits for a worker with stateful support', async (t) t.deepEqual(calls, CHAINED_CALLS); }); +test('a stateful model call fails at once in a run whose config has no stateful flag', async (t) => { + t.timeout(60_000); + const { createWorker, startWorkflow, taskQueue } = helpers(t); + const calls: StatefulCall[] = []; + const plugin = new StrandsPlugin({ models: { stateful: () => new StatefulStub([], calls) } }); + const upgraded = new ModelActivity({ stateful: () => new StatefulStub([textTurn('one')], calls) }); + + const worker = await createWorker({ plugins: [plugin] }); + const history = await worker.runUntil(async () => { + const handle = await startWorkflow(statefulAgent, { args: ['models-queue'] }); + // The config comes from an earlier version, without the flag, so the run takes the stateless + // path, and its model call reaches this version's stateless activity. + const modelsWorker = await createWorker({ + taskQueue: `${taskQueue}-models`, + activities: { + strandsGetModelConfig: async () => ({}), + invokeModel: upgraded.invokeModel, + invokeModelStreaming: upgraded.invokeModelStreaming, + }, + }); + await modelsWorker.runUntil(handle.result()); + return handle.fetchHistory(); + }); + + t.deepEqual(scheduledActivities(history), ['strandsGetModelConfig', 'invokeModel']); + const events = history.events ?? []; + const attempts = events.flatMap((e) => e.activityTaskStartedEventAttributes?.attempt ?? []); + const failures = events.flatMap((e) => e.activityTaskFailedEventAttributes?.failure ?? []); + t.deepEqual(attempts, [1, 1]); + t.is(failures.length, 1); + t.like(failures[0], { applicationFailureInfo: { type: 'StatefulModelOnStatelessPath', nonRetryable: true } }); + t.regex(failures[0]!.message ?? '', /is stateful, but the workflow called it without its model state/); + t.deepEqual(calls, []); +}); + test("the README's continue-as-new handoff carries a stateful model's state", async (t) => { t.timeout(60_000); const { createWorker, executeWorkflow } = helpers(t); diff --git a/typescript/strands-agents/src/model-activity.ts b/typescript/strands-agents/src/model-activity.ts index 103eab5b..cbbab88f 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 { ApplicationFailure } from '@temporalio/common'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import type { Duration } from '@temporalio/common/lib/time'; import { autoHeartbeat } from './heartbeat'; @@ -108,6 +109,8 @@ export class ModelActivity { ) { this.invokeModel = autoHeartbeat(this.invokeModel.bind(this)); this.invokeModelStreaming = autoHeartbeat(this.invokeModelStreaming.bind(this)); + this.invokeStatefulModel = autoHeartbeat(this.invokeStatefulModel.bind(this)); + this.invokeStatefulModelStreaming = autoHeartbeat(this.invokeStatefulModelStreaming.bind(this)); } private getModel(name?: string): Model { @@ -138,7 +141,36 @@ export class ModelActivity { } async invokeModel(input: InvokeModelInput): Promise { - const model = this.getModel(input.modelName); + return this.runModel(this.getStatelessModel(input.modelName), input); + } + + async invokeModelStreaming(input: InvokeModelStreamingInput): Promise { + return this.runModelStreaming(this.getStatelessModel(input.modelName), input); + } + + /** + * The model for a stateless call. A stateful model cannot run here without the agent's model + * state, and retrying does not bring it: the workflow chose this activity because its config + * read had no stateful flag, from a worker without stateful support, or because `TemporalModel` + * runs without `TemporalAgent` and never read the config. + */ + private getStatelessModel(name?: string): Model { + const model = this.getModel(name); + if (model.stateful) { + throw ApplicationFailure.create({ + type: 'StatefulModelOnStatelessPath', + nonRetryable: true, + message: + `The worker model ${JSON.stringify(name ?? this.defaultName)} is stateful, but the workflow ` + + `called it without its model state. Its config was read from a worker without stateful ` + + `support, or TemporalModel runs without TemporalAgent and did not read it. Reset or ` + + `terminate the run.`, + }); + } + return model; + } + + 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)) { @@ -147,8 +179,7 @@ export class ModelActivity { return events; } - async invokeModelStreaming(input: InvokeModelStreamingInput): Promise { - const model = this.getModel(input.modelName); + private async runModelStreaming(model: Model, input: InvokeModelStreamingInput): Promise { const messages = rebuildMessages(input.messages); const stream = WorkflowStreamClient.fromWithinActivity({ batchInterval: input.streamingBatchInterval, @@ -174,14 +205,16 @@ export class ModelActivity { /** Run a stateful model with the agent's model state; return its events and the updated state. */ async invokeStatefulModel(input: InvokeStatefulModelInput): Promise { const modelState = new StateStore(input.modelState); - const events = await this.invokeModel({ ...input, options: { ...input.options, modelState } }); + const model = this.getModel(input.modelName); + const events = await this.runModel(model, { ...input, options: { ...input.options, modelState } }); return { events, modelState: modelState.getAll() }; } /** As {@link invokeStatefulModel}, publishing each event as {@link invokeModelStreaming} does. */ async invokeStatefulModelStreaming(input: InvokeStatefulModelStreamingInput): Promise { const modelState = new StateStore(input.modelState); - const events = await this.invokeModelStreaming({ ...input, options: { ...input.options, modelState } }); + const model = this.getModel(input.modelName); + const events = await this.runModelStreaming(model, { ...input, options: { ...input.options, modelState } }); return { events, modelState: modelState.getAll() }; } } diff --git a/typescript/strands-agents/src/temporal-model.ts b/typescript/strands-agents/src/temporal-model.ts index f3df143b..e9c03a19 100644 --- a/typescript/strands-agents/src/temporal-model.ts +++ b/typescript/strands-agents/src/temporal-model.ts @@ -43,6 +43,10 @@ export interface TemporalModelOptions { * The `modelName` selects which factory the plugin will invoke worker-side; * it must match a key in {@link StrandsPlugin}'s `models` map. A name not * present in `models` raises an error inside the model activity at first call. + * + * `TemporalAgent` reads the worker model's config before its first call. With a plain Strands + * `Agent`, call `await model.loadConfig()` before building the agent: without it, the model reports + * itself as stateless, and a stateful worker model fails every call. */ export class TemporalModel extends Model { private readonly modelName?: string; @@ -115,42 +119,29 @@ export class TemporalModel extends Model { ...this.activityOptions, }); - if (this.isStateful) { - // The agent's model state is a live store; send a copy of its values, and copy the updated - // values back, as the model would have written them natively. - const { modelState, ...rest } = options ?? {}; - const input = { modelName: this.modelName, messages, options: rest, modelState: modelState?.getAll() ?? {} }; - const result = this.streamingTopic - ? await proxiedActivities[INVOKE_STATEFUL_MODEL_STREAMING_ACTIVITY]({ - ...input, - streamingTopic: this.streamingTopic, - streamingBatchInterval: this.streamingBatchInterval, - }) - : await proxiedActivities[INVOKE_STATEFUL_MODEL_ACTIVITY](input); - if (modelState) { - modelState.clear(); - for (const [key, value] of Object.entries(result.modelState)) modelState.set(key, value); - } - yield* result.events; + const { modelState, ...rest } = options ?? {}; + const input = { modelName: this.modelName, messages, options: this.isStateful ? rest : options }; + const streaming = this.streamingTopic + ? { streamingTopic: this.streamingTopic, streamingBatchInterval: this.streamingBatchInterval } + : undefined; + + if (!this.isStateful) { + yield* streaming + ? await proxiedActivities.invokeModelStreaming({ ...input, ...streaming }) + : await proxiedActivities.invokeModel(input); return; } - const events = this.streamingTopic - ? await proxiedActivities.invokeModelStreaming({ - modelName: this.modelName, - messages, - options, - streamingTopic: this.streamingTopic, - streamingBatchInterval: this.streamingBatchInterval, - }) - : await proxiedActivities.invokeModel({ - modelName: this.modelName, - messages, - options, - }); - - for (const event of events) { - yield event; + // The agent's model state is a live store; send a copy of its values, and copy the updated + // values back, as the model would have written them natively. + const statefulInput = { ...input, modelState: modelState?.getAll() ?? {} }; + const result = streaming + ? await proxiedActivities[INVOKE_STATEFUL_MODEL_STREAMING_ACTIVITY]({ ...statefulInput, ...streaming }) + : await proxiedActivities[INVOKE_STATEFUL_MODEL_ACTIVITY](statefulInput); + if (modelState) { + modelState.clear(); + for (const [key, value] of Object.entries(result.modelState)) modelState.set(key, value); } + yield* result.events; } } From 6c7d8d62df43cfe7980415fff68a9981377010c6 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Fri, 30 Oct 2026 14:21:14 -0700 Subject: [PATCH 5/5] fix(strands): fail a stateful call whose worker model is no longer stateful A run whose config read recorded a stateful model keeps the stateful path, and its agent clears its messages after each turn. If the worker model became stateless since, for example an OpenAIModel redeployed with stateful: false, the stateful activities ran it with only each turn's new message and nothing failed. - The stateful activities now fail such a call at once with a non-retryable StatelessModelOnStatefulPath error. - The StatefulModelOnStatelessPath message also names a model that became stateful after the run read its config. - README: changing a model between stateful and stateless applies only to new runs, and an uncaught error leaves the run open with a failing workflow task, with the type on the ModelError's cause.cause. - The pinned-run test covers both directions. --- typescript/strands-agents/README.md | 2 +- .../src/__tests__/test-strands.ts | 82 +++++++++++++------ .../strands-agents/src/model-activity.ts | 34 ++++++-- 3 files changed, 85 insertions(+), 33 deletions(-) diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 00973f90..e3a97505 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -99,7 +99,7 @@ A worker model that keeps the conversation server-side, such as `OpenAIModel` wi `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. -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. Neither run recovers: reset or terminate it. 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. +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: reset or terminate it. Until then, unless the workflow catches the agent's error, the run stays open: the error reaches the workflow as a `ModelError` with the type on its `cause.cause`, and the workflow task fails on each attempt. 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. ## Retries diff --git a/typescript/strands-agents/src/__tests__/test-strands.ts b/typescript/strands-agents/src/__tests__/test-strands.ts index 0dc36d6e..f0f554b5 100644 --- a/typescript/strands-agents/src/__tests__/test-strands.ts +++ b/typescript/strands-agents/src/__tests__/test-strands.ts @@ -416,39 +416,69 @@ test('a stateful model call waits for a worker with stateful support', async (t) t.deepEqual(calls, CHAINED_CALLS); }); -test('a stateful model call fails at once in a run whose config has no stateful flag', async (t) => { +test("a model call fails at once when the worker model's stateful flag differs from the run's config", async (t) => { t.timeout(60_000); const { createWorker, startWorkflow, taskQueue } = helpers(t); const calls: StatefulCall[] = []; const plugin = new StrandsPlugin({ models: { stateful: () => new StatefulStub([], calls) } }); - const upgraded = new ModelActivity({ stateful: () => new StatefulStub([textTurn('one')], calls) }); + // The worker model now in one mode, as after a model change or an upgrade. + class SwitchedStub extends StatefulStub { + constructor(private readonly isStateful: boolean) { + super([textTurn('one')], calls); + } + override get stateful(): boolean { + return this.isStateful; + } + } - const worker = await createWorker({ plugins: [plugin] }); - const history = await worker.runUntil(async () => { - const handle = await startWorkflow(statefulAgent, { args: ['models-queue'] }); - // The config comes from an earlier version, without the flag, so the run takes the stateless - // path, and its model call reaches this version's stateless activity. - const modelsWorker = await createWorker({ - taskQueue: `${taskQueue}-models`, - activities: { - strandsGetModelConfig: async () => ({}), - invokeModel: upgraded.invokeModel, - invokeModelStreaming: upgraded.invokeModelStreaming, - }, + for (const c of [ + // The run's config is stateless (here an earlier version's, with no flag; a model that became + // stateful later gives the same path), so its call to the stateful model reaches the stateless + // activity. + { + config: {}, + stateful: true, + activity: 'invokeModel', + type: 'StatefulModelOnStatelessPath', + message: /is stateful, .* became stateful after the run read its config/, + }, + // The config says stateful, but the model became stateless. + { + config: { stateful: true }, + stateful: false, + activity: 'strandsInvokeStatefulModel', + type: 'StatelessModelOnStatefulPath', + message: /is not stateful, .* became stateless after the run read its config/, + }, + ]) { + const current = new ModelActivity({ stateful: () => new SwitchedStub(c.stateful) }); + const worker = await createWorker({ plugins: [plugin] }); + const history = await worker.runUntil(async () => { + const handle = await startWorkflow(statefulAgent, { args: ['models-queue'] }); + const modelsWorker = await createWorker({ + taskQueue: `${taskQueue}-models`, + activities: { + strandsGetModelConfig: async () => c.config, + invokeModel: current.invokeModel, + invokeModelStreaming: current.invokeModelStreaming, + strandsInvokeStatefulModel: current.invokeStatefulModel, + strandsInvokeStatefulModelStreaming: current.invokeStatefulModelStreaming, + }, + }); + await modelsWorker.runUntil(handle.result()); + return handle.fetchHistory(); }); - await modelsWorker.runUntil(handle.result()); - return handle.fetchHistory(); - }); - t.deepEqual(scheduledActivities(history), ['strandsGetModelConfig', 'invokeModel']); - const events = history.events ?? []; - const attempts = events.flatMap((e) => e.activityTaskStartedEventAttributes?.attempt ?? []); - const failures = events.flatMap((e) => e.activityTaskFailedEventAttributes?.failure ?? []); - t.deepEqual(attempts, [1, 1]); - t.is(failures.length, 1); - t.like(failures[0], { applicationFailureInfo: { type: 'StatefulModelOnStatelessPath', nonRetryable: true } }); - t.regex(failures[0]!.message ?? '', /is stateful, but the workflow called it without its model state/); - t.deepEqual(calls, []); + t.deepEqual(scheduledActivities(history), ['strandsGetModelConfig', c.activity], c.type); + const events = history.events ?? []; + const attempts = events.flatMap((e) => e.activityTaskStartedEventAttributes?.attempt ?? []); + const failures = events.flatMap((e) => e.activityTaskFailedEventAttributes?.failure ?? []); + t.deepEqual(attempts, [1, 1], c.type); + t.is(failures.length, 1, c.type); + t.like(failures[0], { applicationFailureInfo: { type: c.type, nonRetryable: true } }); + t.regex(failures[0]?.message ?? '', c.message); + t.deepEqual(calls, [], c.type); + } }); test("the README's continue-as-new handoff carries a stateful model's state", async (t) => { diff --git a/typescript/strands-agents/src/model-activity.ts b/typescript/strands-agents/src/model-activity.ts index cbbab88f..e2ea78eb 100644 --- a/typescript/strands-agents/src/model-activity.ts +++ b/typescript/strands-agents/src/model-activity.ts @@ -151,8 +151,9 @@ export class ModelActivity { /** * The model for a stateless call. A stateful model cannot run here without the agent's model * state, and retrying does not bring it: the workflow chose this activity because its config - * read had no stateful flag, from a worker without stateful support, or because `TemporalModel` - * runs without `TemporalAgent` and never read the config. + * read had no stateful flag, from a worker without stateful support or from a model that became + * stateful later, or because `TemporalModel` runs without `TemporalAgent` and never read the + * config. */ private getStatelessModel(name?: string): Model { const model = this.getModel(name); @@ -163,8 +164,29 @@ export class ModelActivity { message: `The worker model ${JSON.stringify(name ?? this.defaultName)} is stateful, but the workflow ` + `called it without its model state. Its config was read from a worker without stateful ` + - `support, or TemporalModel runs without TemporalAgent and did not read it. Reset or ` + - `terminate the run.`, + `support, the worker model became stateful after the run read its config, or ` + + `TemporalModel runs without TemporalAgent and did not read it. Reset or terminate the run.`, + }); + } + return model; + } + + /** + * The model for a stateful call. A model that is no longer stateful ignores the model state, and + * the agent, which read a stateful config, clears its messages after each turn, so the model + * would see only each turn's new message: the worker model became stateless after the run read + * its config. + */ + private getStatefulModel(name?: string): Model { + const model = this.getModel(name); + if (!model.stateful) { + throw ApplicationFailure.create({ + type: 'StatelessModelOnStatefulPath', + nonRetryable: true, + message: + `The worker model ${JSON.stringify(name ?? this.defaultName)} is not stateful, but the run ` + + `read a stateful config, so its agent does not keep the messages this model needs. The ` + + `worker model became stateless after the run read its config. Reset or terminate the run.`, }); } return model; @@ -205,7 +227,7 @@ export class ModelActivity { /** Run a stateful model with the agent's model state; return its events and the updated state. */ async invokeStatefulModel(input: InvokeStatefulModelInput): Promise { const modelState = new StateStore(input.modelState); - const model = this.getModel(input.modelName); + const model = this.getStatefulModel(input.modelName); const events = await this.runModel(model, { ...input, options: { ...input.options, modelState } }); return { events, modelState: modelState.getAll() }; } @@ -213,7 +235,7 @@ export class ModelActivity { /** As {@link invokeStatefulModel}, publishing each event as {@link invokeModelStreaming} does. */ async invokeStatefulModelStreaming(input: InvokeStatefulModelStreamingInput): Promise { const modelState = new StateStore(input.modelState); - const model = this.getModel(input.modelName); + const model = this.getStatefulModel(input.modelName); const events = await this.runModelStreaming(model, { ...input, options: { ...input.options, modelState } }); return { events, modelState: modelState.getAll() }; }