diff --git a/typescript/strands-agents/README.md b/typescript/strands-agents/README.md index 0c5d0f67..c7176c31 100644 --- a/typescript/strands-agents/README.md +++ b/typescript/strands-agents/README.md @@ -93,6 +93,12 @@ Each `TemporalAgent` carries its own activity options (timeouts, retry policy, t Before its first model call, each `TemporalAgent` reads the model's `modelId` and `contextWindowLimit` from its `getConfig()` through the `strandsGetModelConfig` activity, with the agent's activity options. Strands' traces and [proactive compression](https://strandsagents.com/docs/user-guide/sdk/agents/conversation-management/#proactive-context-compression) then use the real model's values, as they do outside Temporal. If you scaled `compressionThreshold` by the model's window divided by 200,000 to make up for the missing limit, remove that scaling. +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, 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. + ## 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`: @@ -242,24 +248,26 @@ The plugin keeps the failure converter your client and worker set with `dataConv ## 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[] = []; @@ -272,7 +280,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__/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 1e5af41b..b9111512 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 { 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, @@ -52,6 +53,8 @@ import { inWorkflowToolAgent, mcpAgent, retryConfigAgent, + statefulAgent, + statefulChatAgent, streamingAgent, structuredOutputAgent, turnsAgent, @@ -352,6 +355,212 @@ 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 }; + +/** 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) }, + }); + // 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`; + + 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 (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)); + } + }); + 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("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) } }); + // 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; + } + } + + 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(); + }); + + 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) => { + 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('a history whose code depends on when a handler finishes still replays', async (t) => { // Recorded with `handlerTiming` on main at 63d7aec: one microtask after its Signal handler ran, // every handler had finished, so the run scheduled `finished`. The plugin wraps no handlers, so diff --git a/typescript/strands-agents/src/__tests__/workflows/strands.ts b/typescript/strands-agents/src/__tests__/workflows/strands.ts index b47132b5..7bbf6a79 100644 --- a/typescript/strands-agents/src/__tests__/workflows/strands.ts +++ b/typescript/strands-agents/src/__tests__/workflows/strands.ts @@ -8,16 +8,21 @@ import { type BaseModelConfig, type InterruptResponseContent, type InterruptResponseContentData, + type JSONValue, + type Message, } from '@strands-agents/sdk'; import { ActivityFailure, allHandlersFinished, ApplicationFailure, + continueAsNew, defineSignal, defineUpdate, proxyActivities, setHandler, condition, + workflowInfo, + type ActivityOptions, } from '@temporalio/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; import { TemporalAgent, TemporalMCPClient, type TemporalAgentOptions, workflow as strandsWorkflow } from '../..'; @@ -61,6 +66,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..e2ea78eb 100644 --- a/typescript/strands-agents/src/model-activity.ts +++ b/typescript/strands-agents/src/model-activity.ts @@ -1,5 +1,6 @@ -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 { ApplicationFailure } from '@temporalio/common'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import type { Duration } from '@temporalio/common/lib/time'; import { autoHeartbeat } from './heartbeat'; @@ -49,6 +50,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 +77,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 +96,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. */ @@ -87,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 { @@ -111,12 +135,64 @@ 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 { - 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 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); + 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, 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; + } + + 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)) { @@ -125,8 +201,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, @@ -143,4 +218,25 @@ 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 model = this.getStatefulModel(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 model = this.getStatefulModel(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/plugin.ts b/typescript/strands-agents/src/plugin.ts index 005774a0..9b153a8a 100644 --- a/typescript/strands-agents/src/plugin.ts +++ b/typescript/strands-agents/src/plugin.ts @@ -7,8 +7,18 @@ import type { ActivityInboundCallsInterceptor, BundleOptions } from '@temporalio import { SimplePlugin } from '@temporalio/plugin'; import { rewriteError } from './strands-errors'; 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, @@ -98,6 +108,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 d39e71e4..3d520aa0 100644 --- a/typescript/strands-agents/src/temporal-agent.ts +++ b/typescript/strands-agents/src/temporal-agent.ts @@ -12,6 +12,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 ' + @@ -52,6 +57,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(); } /** diff --git a/typescript/strands-agents/src/temporal-model.ts b/typescript/strands-agents/src/temporal-model.ts index 932e2c2c..0ae0871f 100644 --- a/typescript/strands-agents/src/temporal-model.ts +++ b/typescript/strands-agents/src/temporal-model.ts @@ -8,12 +8,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}. * @@ -37,6 +44,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; @@ -44,6 +55,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 = {}) { @@ -66,10 +78,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. @@ -82,8 +99,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; }); } @@ -92,40 +111,56 @@ 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, }); - let events: ModelStreamEvent[]; - try { - 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, - }); - } catch (err) { - // Give the agent Strands' own error for a context overflow, so its conversation manager can - // trim the conversation and call the model again. Keep the activity failure as its cause. - if ( - err instanceof workflow.ActivityFailure && - applicationFailureType(err.cause) === CONTEXT_WINDOW_OVERFLOW_TYPE - ) { - throw Object.assign(new ContextWindowOverflowError(err.cause?.message ?? ''), { cause: err }); + 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) { + let events: ModelStreamEvent[]; + try { + events = streaming + ? await proxiedActivities.invokeModelStreaming({ ...input, ...streaming }) + : await proxiedActivities.invokeModel(input); + } catch (err) { + // Give the agent Strands' own error for a context overflow, so its conversation manager can + // trim the conversation and call the model again. Keep the activity failure as its cause. + if ( + err instanceof workflow.ActivityFailure && + applicationFailureType(err.cause) === CONTEXT_WINDOW_OVERFLOW_TYPE + ) { + throw Object.assign(new ContextWindowOverflowError(err.cause?.message ?? ''), { cause: err }); + } + throw err; } - throw err; + yield* events; + return; } - for (const event of events) { - yield event; + // A stateful model's overflow is not rethrown as Strands' error: its conversation is on the + // server, and the default manager the agent keeps would trim and retry where Strands, which + // gives a stateful agent no manager, fails the turn. + + // 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; } }