From fc42a9a02fbf5c2c518b049188090d0b199bdd93 Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Wed, 23 Sep 2026 22:04:22 -0700 Subject: [PATCH 1/4] Simplify agent inactivity tracking --- ARCHITECTURE.md | 16 +- Makefile | 2 +- README.md | 12 +- .../0016-agent-owned-inactivity-expiration.md | 28 +- docs/flow-model.md | 28 +- internal/agent/flow.go | 411 +++++++++++------- internal/agent/flow_integration_test.go | 64 ++- internal/agent/tool_recovery_test.go | 2 +- internal/agent/types.go | 13 +- internal/agent/types_test.go | 1 + 10 files changed, 374 insertions(+), 203 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index daf54a7..dc45d3a 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -77,14 +77,14 @@ defines no placeholder API or compatibility path for it. An embedding application may configure `AgentConfig.InactivityTimeoutSeconds` and inject `InactivityExpirationHandler`. The Agent owns the inactivity -decision because only its durable state distinguishes a user wait from active -model/tool work and `durable_wait`. `AwaitUser`, `AwaitToolApproval`, and -`AwaitManualToolRecovery` persist `InactivityDeadline` and race their existing -Channels with one Dex Timer. Accepted input wins when both are ready. All other -Steps expose a null deadline. Expiration first commits `expiring` and one -`inactivity_expired` event, then retries the stable Flow ID, Run ID, deadline, -and runtime-metadata callback for up to seven days. Only callback success closes -the Agent Flow; cross-resource cleanup remains the embedding application's job. +decision through one `InactivityTimeout` branch started beside the Agent loop. +Accepted user RPCs atomically advance `InactivityDeadline` and publish +`ResetInactivityTimer` only when the extension exceeds one minute. Model and +ordinary tool work do not pause the deadline. `DurableWait` extends it through +the wait target. Expiration retries the stable Flow ID, Run ID, deadline, and +runtime-metadata callback for up to seven days. Callback success commits +`expiring`, emits `inactivity_expired`, and force-completes the Agent; cross- +resource cleanup remains the embedding application's job. `CurrentMessages` and `ArchivedMessages` are the typed application history; they are not Dex execution history. `AgentState` owns the retained sequence diff --git a/Makefile b/Makefile index fa17a67..748a701 100644 --- a/Makefile +++ b/Makefile @@ -87,7 +87,7 @@ check-flow-definition: install-dexcli echo "Flow definition must contain the canonical ExecuteTool Step" >&2; \ exit 1; \ fi; \ - for channel in answeredUserInputsChannel queuedUserMessagesChannel steeredUserMessagesChannel toolApprovalsChannel toolRecoveryDecisionsChannel parallelToolResultsChannel planExecutionsChannel; do \ + for channel in resetInactivityTimerChannel answeredUserInputsChannel queuedUserMessagesChannel steeredUserMessagesChannel toolApprovalsChannel toolRecoveryDecisionsChannel parallelToolResultsChannel planExecutionsChannel; do \ if ! grep -Fq "\"id\": \"resource:channel:$${channel}\"" "$${flow_definition}" || \ ! grep -Fq "\"resourceId\": \"resource:channel:$${channel}\"" "$${flow_definition}"; then \ echo "Flow definition must render Channel $${channel} and its WaitFor edge" >&2; \ diff --git a/README.md b/README.md index ed05bad..b671003 100644 --- a/README.md +++ b/README.md @@ -70,14 +70,14 @@ typed `StartRequest`. `RuntimeMetadata` is an optional JSON object of at most passed only to tool implementations, never to models, browser Snapshots, or Streams. Do not put secrets in it. -Set `AgentConfig.InactivityTimeoutSeconds` to arm inactivity expiration while -the Agent waits for a message, question answer, approval, or recovery decision; -zero disables it. Embedders enabling the timeout must pass +Set `AgentConfig.InactivityTimeoutSeconds` to arm inactivity expiration across +the Agent lifecycle; zero disables it and positive values start at two minutes. +Embedders enabling the timeout must pass `agent.WithInactivityExpirationHandler(...)` to `agent.NewFlow`. The handler receives a stable Flow ID, Run ID, deadline, and runtime metadata for an -idempotent application cleanup request. Snapshot exposes `inactivityDeadline` -only while the Timer is armed; model work, tool work, and `durable_wait` pause -the timeout. +idempotent application cleanup request. Accepted user mutations coalesce Timer +resets within one minute. A `durable_wait` extends the deadline through its +target; model and ordinary tool work do not pause it. External tool retries belong to Dex. `ToolDefinition` controls attempt timeout, maximum attempts, and total duration. A registry performs exactly one call for diff --git a/docs/adr/0016-agent-owned-inactivity-expiration.md b/docs/adr/0016-agent-owned-inactivity-expiration.md index 034e6c2..b4597c5 100644 --- a/docs/adr/0016-agent-owned-inactivity-expiration.md +++ b/docs/adr/0016-agent-owned-inactivity-expiration.md @@ -6,32 +6,32 @@ Accepted on 2026-09-23. ## Context -Session infrastructure cannot infer inactivity from API traffic. Snapshot and -Stream reads are observation, model and tool calls are active work, and a -`durable_wait` intentionally suspends longer than an inactivity window. Copying +Session infrastructure cannot infer user inactivity from API traffic. Snapshot +and Stream reads are observation, while a `durable_wait` may intentionally +suspend longer than an inactivity window. Copying deadlines into multiple lifecycle Flows also requires publishing every activity to every resource owner and creates competing shutdown decisions. ## Decision `AIAgentFlow` owns the inactivity decision. A positive -`inactivity_timeout_seconds` arms a Dex Timer only in `AwaitUser`, -`AwaitToolApproval`, and `AwaitManualToolRecovery`. The same atomic wait writes -`InactivityDeadline`; every active-work or durable-wait path leaves it absent. -Accepted Channel input takes precedence if it becomes ready with the Timer. +`inactivity_timeout_seconds` starts one parallel `InactivityTimeout` Step and +persists `InactivityDeadline`. Accepted user RPCs advance the Attribute and +publish `ResetInactivityTimer` only when the extension exceeds one minute. +Model and ordinary tool work do not pause the deadline. `DurableWait` advances +it to the wait target plus the configured timeout. -Expiration commits the `expiring` status and one `inactivity_expired` activity -event before invoking the constructor-injected `InactivityExpirationHandler`. +Expiration invokes the constructor-injected `InactivityExpirationHandler`. The handler receives Flow ID, Run ID, deadline, and trusted runtime metadata. Those values form a stable idempotency identity across retries and Worker replacement. Dex retries the callback for a bounded seven-day window, and the -Agent completes only after the callback succeeds. +Agent commits `expiring`, emits `inactivity_expired`, and force-completes only +after the callback succeeds. ## Consequences Embedders configure the timeout and own cross-resource cleanup behind the handler. They do not report each activity back to lifecycle infrastructure. -Snapshots expose the exact deadline only while inactivity is armed, allowing a -UI to distinguish an active countdown from paused model, tool, and durable-wait -work. A failed callback leaves the Agent durably `expiring` until retry succeeds -or the explicit retry policy is exhausted. +Snapshots expose the exact deadline for the configured Agent lifecycle. A +failed callback retains the expired deadline until retry succeeds or the +explicit retry policy is exhausted. diff --git a/docs/flow-model.md b/docs/flow-model.md index ce9366d..937df40 100644 --- a/docs/flow-model.md +++ b/docs/flow-model.md @@ -105,8 +105,12 @@ DurableWait -> next tool or CompactContext (timer fired) -> CompactContext (steered) -AwaitUser / AwaitToolApproval / AwaitManualToolRecovery - -> BeginInactivityExpiration -> ExpireInactivity (inactivity Timer fired) +Init + -> AwaitUser + InactivityTimeout (timeout configured) + +InactivityTimeout + -> InactivityTimeout (reset deadline) + -> ForceComplete (handler succeeded) ``` ### Tool-call routing @@ -168,8 +172,7 @@ history, and makes the model replan. | `PrepareManualToolRecovery` | none | Capture a serial returned-unknown or exhausted call and its redacted error type | | `AwaitManualToolRecovery` | exact recovery decision or steering | Persist recovery state; retry selected calls, continue unknowns, stop the sequence, or replan | | `DurableWait` | Timer or steering | Persist waiting status; record completion or interruption and continue | -| `BeginInactivityExpiration` | none | Clear the deadline, persist `expiring`, emit one expiration event, and schedule the callback | -| `ExpireInactivity` | none | Invoke the application handler with stable identity under bounded retry; complete only after success | +| `InactivityTimeout` | reset timestamp or deadline Timer | Coalesce reset deadlines, invoke the application handler under bounded retry, and force-complete only after success | ### Tool StepOptions resolution @@ -209,13 +212,13 @@ Approval and timer waits never change the round. The next value must remain within JavaScript's safe integer range or the WaitFor fails explicitly. -When `inactivity_timeout_seconds` is positive, the three user-controlled waits -append a Dex Timer and persist its exact `InactivityDeadline`. Ready answers, -messages, plan execution, steering, approvals, and recovery decisions are read -before the Timer outcome, so accepted input wins a simultaneous wake. Leaving a -user wait deletes the deadline. Model calls, serial or parallel tools, and -`DurableWait` never carry an inactivity Timer. After a durable wait completes or -is steered, the next user wait receives a complete new timeout window. +When `inactivity_timeout_seconds` is positive, `Init` starts one parallel +`InactivityTimeout` Step and persists its exact `InactivityDeadline`. Accepted +messages, answers, plan execution, steering, queue deletion, approvals, and +recovery decisions advance the deadline transactionally. Extensions of one +minute or less do not publish another reset. Snapshot and Stream reads do not +reset it. Model calls and serial or parallel tools continue under the same +deadline. `DurableWait` advances it to the wait target plus the timeout. ## Durable resources @@ -233,7 +236,8 @@ is steered, the next user wait receives a complete new timeout window. | `PendingToolRecovery` | Attribute | Exact recovery revision and redacted failed-call presentation | | `PendingTimer` | Attribute | Current durable wait presentation | | `PendingUserInput` | Attribute | Current structured question batch | -| `InactivityDeadline` | Attribute | Exact deadline only while one user-controlled wait is armed | +| `InactivityDeadline` | Attribute | Current global deadline while inactivity expiration is configured | +| `ResetInactivityTimer` | Channel | Coalesced UTC deadline updates for the timeout branch | | `AnsweredUserInputs` | Channel | Validated answers awaiting immediate consumption | | `QueuedUserMessages` | Channel | FIFO `PendingUserMessage` payloads with stable application IDs | | `SteeredUserMessages` | Channel | Safe-boundary steering payloads preserving the same application IDs | diff --git a/internal/agent/flow.go b/internal/agent/flow.go index ee3fbe3..2301297 100644 --- a/internal/agent/flow.go +++ b/internal/agent/flow.go @@ -43,6 +43,7 @@ var ( pendingTimerAttribute = dex.DefineAttribute[PendingTimer]("PendingTimer") pendingUserInputAttribute = dex.DefineAttribute[PendingUserInput]("PendingUserInput") inactivityDeadlineAttribute = dex.DefineAttribute[time.Time]("InactivityDeadline") + resetInactivityTimerChannel = dex.DefineChannel[time.Time]("ResetInactivityTimer") answeredUserInputsChannel = dex.DefineChannel[AnsweredUserInput]("AnsweredUserInputs") queuedUserMessagesChannel = dex.DefineChannel[PendingUserMessage]("QueuedUserMessages") steeredUserMessagesChannel = dex.DefineChannel[PendingUserMessage]("SteeredUserMessages") @@ -57,10 +58,12 @@ var ( // Flow is the durable AI Agent state machine. type Flow struct { - modelClient ModelClient - tools ToolRegistry - inactivityExpiration InactivityExpirationHandler - rpcDefinitionsForTestOnly []dex.RPCDef + modelClient ModelClient + tools ToolRegistry + inactivityExpiration InactivityExpirationHandler + rpcDefinitionsForTestOnly []dex.RPCDef + minimumInactivityTimeoutSecondsForTestOnly int64 + inactivityResetMinimumExtensionForTestOnly time.Duration } var _ dex.Flow = (*Flow)(nil) @@ -137,8 +140,7 @@ func (flow *Flow) GetSteps() []dex.StepDef { dex.DefineStep(prepareManualToolRecoveryStep{flow: flow}), dex.DefineStep(awaitManualToolRecoveryStep{flow: flow}), dex.DefineStep(durableWaitStep{flow: flow}), - dex.DefineStep(beginInactivityExpirationStep{flow: flow}), - dex.DefineStep(expireInactivityStep{flow: flow}), + dex.DefineStep(inactivityTimeoutStep{flow: flow}), } } @@ -146,16 +148,25 @@ func (flow *Flow) GetSteps() []dex.StepDef { func (flow *Flow) GetRPCs() []dex.RPCDef { definitions := []dex.RPCDef{ dex.DefineRPC(flow.SendMessage, &dex.RPCOptions{ - Timeout: defaultCommandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingUserInputAttribute)}, + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{ + dex.LockAttribute(pendingUserInputAttribute), + dex.LockAttribute(inactivityDeadlineAttribute), + }, }), dex.DefineRPC(flow.AnswerQuestions, &dex.RPCOptions{ - Timeout: defaultCommandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingUserInputAttribute)}, + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{ + dex.LockAttribute(pendingUserInputAttribute), + dex.LockAttribute(inactivityDeadlineAttribute), + }, }), dex.DefineRPC(flow.SteerMessage, &dex.RPCOptions{ - Timeout: defaultCommandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingToolRecoveryAttribute)}, + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{ + dex.LockAttribute(pendingToolRecoveryAttribute), + dex.LockAttribute(inactivityDeadlineAttribute), + }, IsTransactional: true, LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, }), @@ -173,21 +184,31 @@ func (flow *Flow) GetRPCs() []dex.RPCDef { }), dex.DefineRPC(flow.DeleteQueuedMessage, &dex.RPCOptions{ Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{dex.LockAttribute(inactivityDeadlineAttribute)}, IsTransactional: true, LoadChannels: []dex.ChannelDef{queuedUserMessagesChannel}, }), dex.DefineRPC(flow.ApproveTool, &dex.RPCOptions{ - Timeout: defaultCommandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingApprovalAttribute)}, + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{ + dex.LockAttribute(pendingApprovalAttribute), + dex.LockAttribute(inactivityDeadlineAttribute), + }, IsTransactional: true, }), dex.DefineRPC(flow.ResolveToolRecovery, &dex.RPCOptions{ - Timeout: defaultCommandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(pendingToolRecoveryAttribute)}, + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{ + dex.LockAttribute(pendingToolRecoveryAttribute), + dex.LockAttribute(inactivityDeadlineAttribute), + }, }), dex.DefineRPC(flow.ExecutePlan, &dex.RPCOptions{ - Timeout: defaultCommandTimeout, - LockAttributes: []dex.AttributeLock{dex.LockAttribute(agentStateAttribute)}, + Timeout: defaultCommandTimeout, + LockAttributes: []dex.AttributeLock{ + dex.LockAttribute(agentStateAttribute), + dex.LockAttribute(inactivityDeadlineAttribute), + }, IsTransactional: true, }), } @@ -213,6 +234,7 @@ func (*Flow) GetPersistenceSchema() dex.PersistenceSchema { inactivityDeadlineAttribute, }, Channels: []dex.ChannelDef{ + resetInactivityTimerChannel, answeredUserInputsChannel, queuedUserMessagesChannel, steeredUserMessagesChannel, @@ -230,7 +252,7 @@ func (*Flow) GetPersistenceSchema() dex.PersistenceSchema { } // SendMessage queues one non-empty user message when no question is pending. -func (*Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.RPCResult[bool], error) { +func (flow *Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.RPCResult[bool], error) { if strings.TrimSpace(input.Value.Content) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -244,6 +266,13 @@ func (*Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.RPCRes if pending != nil { return &dex.RPCResult[bool]{Output: false}, nil } + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + if err != nil { + return nil, err + } + if !accepted { + return &dex.RPCResult[bool]{Output: false}, nil + } if err := queuedUserMessagesChannel.Publish(ctx, input); err != nil { return nil, err } @@ -251,7 +280,7 @@ func (*Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.RPCRes } // AnswerQuestions validates and publishes one exact pending input batch atomically. -func (*Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) (*dex.RPCResult[bool], error) { if strings.TrimSpace(string(input.CallID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -266,6 +295,13 @@ func (*Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) (*de if !isValid { return &dex.RPCResult[bool]{Output: false}, nil } + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + if err != nil { + return nil, err + } + if !accepted { + return &dex.RPCResult[bool]{Output: false}, nil + } if err := pendingUserInputAttribute.Delete(ctx); err != nil { return nil, err } @@ -280,7 +316,7 @@ func (*Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) (*de } // SteerMessage atomically moves a queued message into the Steer queue. -func (*Flow) SteerMessage(ctx dex.Context, input SteerMessageRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) SteerMessage(ctx dex.Context, input SteerMessageRequest) (*dex.RPCResult[bool], error) { if strings.TrimSpace(string(input.MessageID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -296,6 +332,7 @@ func (*Flow) SteerMessage(ctx dex.Context, input SteerMessageRequest) (*dex.RPCR if err != nil { return nil, err } + shouldDeletePendingToolRecovery := false if state.Status == AgentStatusWaitingForToolRecovery { pending, pendingErr := getPendingToolRecovery(ctx) if pendingErr != nil { @@ -304,6 +341,16 @@ func (*Flow) SteerMessage(ctx dex.Context, input SteerMessageRequest) (*dex.RPCR if pending == nil { return &dex.RPCResult[bool]{Output: false}, nil } + shouldDeletePendingToolRecovery = true + } + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + if err != nil { + return nil, err + } + if !accepted { + return &dex.RPCResult[bool]{Output: false}, nil + } + if shouldDeletePendingToolRecovery { if deleteErr := pendingToolRecoveryAttribute.Delete(ctx); deleteErr != nil { return nil, deleteErr } @@ -403,7 +450,7 @@ func (*Flow) GetArchivedMessages( } // DeleteQueuedMessage removes one exact pending user message. -func (*Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*dex.RPCResult[bool], error) { +func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*dex.RPCResult[bool], error) { if strings.TrimSpace(string(messageID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -415,6 +462,13 @@ func (*Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*dex.RPC if !found { return &dex.RPCResult[bool]{Output: false}, nil } + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + if err != nil { + return nil, err + } + if !accepted { + return &dex.RPCResult[bool]{Output: false}, nil + } if err := queuedUserMessagesChannel.Delete(ctx, message.MessageID); err != nil { return nil, err } @@ -422,7 +476,7 @@ func (*Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*dex.RPC } // ApproveTool publishes an approval only for the current exact call ID. -func (*Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex.RPCResult[bool], error) { pending, err := getPendingApproval(ctx) if err != nil { return nil, err @@ -430,6 +484,13 @@ func (*Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex.RPCRe if pending == nil || pending.CallID != input.CallID { return &dex.RPCResult[bool]{Output: false}, nil } + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + if err != nil { + return nil, err + } + if !accepted { + return &dex.RPCResult[bool]{Output: false}, nil + } if err := pendingApprovalAttribute.Delete(ctx); err != nil { return nil, err } @@ -440,7 +501,7 @@ func (*Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex.RPCRe } // ResolveToolRecovery publishes one complete decision for the exact pending recovery revision. -func (*Flow) ResolveToolRecovery( +func (flow *Flow) ResolveToolRecovery( ctx dex.Context, input ResolveToolRecoveryRequest, ) (*dex.RPCResult[bool], error) { @@ -454,6 +515,13 @@ func (*Flow) ResolveToolRecovery( if !isValidToolRecoveryResolution(*pending, input) { return &dex.RPCResult[bool]{Output: false}, nil } + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + if err != nil { + return nil, err + } + if !accepted { + return &dex.RPCResult[bool]{Output: false}, nil + } if err := pendingToolRecoveryAttribute.Delete(ctx); err != nil { return nil, err } @@ -464,7 +532,7 @@ func (*Flow) ResolveToolRecovery( } // ExecutePlan schedules an exact waiting draft or active plan revision. -func (*Flow) ExecutePlan(ctx dex.Context, input PlanExecutionRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) ExecutePlan(ctx dex.Context, input PlanExecutionRequest) (*dex.RPCResult[bool], error) { state, err := agentStateAttribute.Get(ctx) if err != nil { return nil, err @@ -503,6 +571,13 @@ func (*Flow) ExecutePlan(ctx dex.Context, input PlanExecutionRequest) (*dex.RPCR if !canExecute { return &dex.RPCResult[bool]{Output: false}, nil } + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + if err != nil { + return nil, err + } + if !accepted { + return &dex.RPCResult[bool]{Output: false}, nil + } revision := plan.Revision state.PendingPlanExecutionRevision = &revision state.PlanNoProgressAttempts = 0 @@ -598,32 +673,53 @@ func clearInactivityDeadline(ctx dex.Context) error { return err } -func (flow *Flow) waitForUserActivity( - ctx dex.Context, - conditions ...dex.Condition, -) (*dex.Wait, error) { +func (flow *Flow) resetInactivityTimerForUserOperation(ctx dex.Context) (bool, error) { config, err := agentConfigAttribute.Get(ctx) if err != nil { - return nil, err + return false, err } if config.InactivityTimeoutSeconds == 0 { - if clearErr := clearInactivityDeadline(ctx); clearErr != nil { - return nil, clearErr - } - return dex.AnyOf(conditions...), nil + return true, nil } - deadline := ctx.FirstAttemptAt().UTC().Add( + resetTarget := ctx.FirstAttemptAt().UTC().Add( time.Duration(config.InactivityTimeoutSeconds) * time.Second, ) - if err := inactivityDeadlineAttribute.Set(ctx, deadline); err != nil { - return nil, err + return flow.resetInactivityTimer(ctx, resetTarget) +} + +func (flow *Flow) resetInactivityTimer(ctx dex.Context, resetTarget time.Time) (bool, error) { + deadline, err := inactivityDeadlineAttribute.Get(ctx) + if err != nil { + return false, err } - remaining := time.Until(deadline) - if remaining <= 0 { - remaining = time.Nanosecond + if !deadline.After(ctx.FirstAttemptAt().UTC()) { + return false, nil + } + resetTarget = resetTarget.UTC() + if resetTarget.Sub(deadline) <= flow.inactivityResetMinimumExtension() { + return true, nil + } + if err := inactivityDeadlineAttribute.Set(ctx, resetTarget); err != nil { + return false, err + } + if err := resetInactivityTimerChannel.Publish(ctx, resetTarget); err != nil { + return false, err + } + return true, nil +} + +func (flow *Flow) inactivityResetMinimumExtension() time.Duration { + if flow.inactivityResetMinimumExtensionForTestOnly > 0 { + return flow.inactivityResetMinimumExtensionForTestOnly } - conditions = append(conditions, dex.Timer(remaining)) - return dex.AnyOf(conditions...), nil + return time.Minute +} + +func (flow *Flow) minimumInactivityTimeoutSeconds() int64 { + if flow.minimumInactivityTimeoutSecondsForTestOnly > 0 { + return flow.minimumInactivityTimeoutSecondsForTestOnly + } + return int64((2 * time.Minute) / time.Second) } func (flow *Flow) initializingSnapshot( @@ -673,7 +769,7 @@ func findPendingUserMessage( } func (flow *Flow) validateConfig(config AgentConfig) error { - if err := config.Validate(); err != nil { + if err := config.validate(flow.minimumInactivityTimeoutSeconds()); err != nil { return err } if config.InactivityTimeoutSeconds > 0 && flow.inactivityExpiration == nil { @@ -1844,8 +1940,7 @@ const ( stepTypePrepareManualRecovery stepType = "PrepareManualToolRecovery" stepTypeAwaitManualRecovery stepType = "AwaitManualToolRecovery" stepTypeDurableWait stepType = "DurableWait" - stepTypeBeginExpiration stepType = "BeginInactivityExpiration" - stepTypeExpireInactivity stepType = "ExpireInactivity" + stepTypeInactivityTimeout stepType = "InactivityTimeout" maximumSteeringMessageCount = 2_147_483_647 maximumAutomaticPlanRecoveryCount = 1 @@ -1983,9 +2078,25 @@ var ( dex.LockAttribute(pendingToolRecoveryAttribute), }, } + durableWaitStepOptions = &dex.StepOptions{ + WaitForMethodTimeout: messageMutationStepOptions.WaitForMethodTimeout, + ExecuteMethodTimeout: messageMutationStepOptions.ExecuteMethodTimeout, + ExecuteLoadAttributeMaps: messageMutationStepOptions.ExecuteLoadAttributeMaps, + WaitForLockAttributes: []dex.AttributeLock{ + dex.LockAttribute(inactivityDeadlineAttribute), + }, + ExecuteLockAttributes: messageMutationStepOptions.ExecuteLockAttributes, + } inactivityExpirationStepOptions = &dex.StepOptions{ + WaitForMethodTimeout: time.Minute, ExecuteMethodTimeout: time.Minute, ExecuteDurability: dex.StepDurabilitySync, + WaitForLockAttributes: []dex.AttributeLock{ + dex.LockAttribute(inactivityDeadlineAttribute), + }, + ExecuteLockAttributes: []dex.AttributeLock{ + dex.LockAttribute(inactivityDeadlineAttribute), + }, ExecuteRetry: &dex.RetryPolicy{ InitialInterval: time.Second, BackoffCoefficient: 2, @@ -2020,7 +2131,19 @@ func (step initStep) Execute(ctx dex.Context, input AgentConfig) (*dex.StepDecis if err := waitingInputRoundAttribute.Set(ctx, 0); err != nil { return nil, err } - return dex.GoTo(awaitUserStep{flow: step.flow}, nil), nil + if input.InactivityTimeoutSeconds == 0 { + return dex.GoTo(awaitUserStep{flow: step.flow}, nil), nil + } + deadline := ctx.FirstAttemptAt().UTC().Add( + time.Duration(input.InactivityTimeoutSeconds) * time.Second, + ) + if err := inactivityDeadlineAttribute.Set(ctx, deadline); err != nil { + return nil, err + } + return dex.GoToMany( + dex.MovementOf(awaitUserStep{flow: step.flow}, nil), + dex.MovementOf(inactivityTimeoutStep{flow: step.flow}, deadline), + ), nil } type awaitUserStep struct { @@ -2053,7 +2176,7 @@ func (step awaitUserStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wait, error if err := incrementWaitingInputRound(ctx); err != nil { return nil, err } - return step.flow.waitForUserActivity(ctx, answeredUserInputsChannel.ForOne()) + return dex.Until(answeredUserInputsChannel.ForOne()), nil } if plan != nil && plan.Status != PlanStatusCompleted { planKey := planRevisionKey(plan.Revision) @@ -2064,12 +2187,11 @@ func (step awaitUserStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wait, error if err := incrementWaitingInputRound(ctx); err != nil { return nil, err } - return step.flow.waitForUserActivity( - ctx, + return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), queuedUserMessagesChannel.ForOne(), planExecutionsChannel.ForOne(planKey), - ) + ), nil } return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), @@ -2081,11 +2203,10 @@ func (step awaitUserStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wait, error if err := incrementWaitingInputRound(ctx); err != nil { return nil, err } - return step.flow.waitForUserActivity( - ctx, + return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), queuedUserMessagesChannel.ForOne(), - ) + ), nil } return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), @@ -2099,9 +2220,6 @@ func (step awaitUserStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecisio return nil, err } if len(answeredInputs) > 0 { - if clearErr := clearInactivityDeadline(ctx); clearErr != nil { - return nil, clearErr - } answeredInput := answeredInputs[0] sequence, beginErr := step.flow.beginUserTurn(ctx, answeredInput.Message) if beginErr != nil { @@ -2122,9 +2240,6 @@ func (step awaitUserStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecisio return nil, err } if len(steered) > 0 { - if clearErr := clearInactivityDeadline(ctx); clearErr != nil { - return nil, clearErr - } if beginErr := step.flow.beginSteeredTurn(ctx, steered); beginErr != nil { return nil, beginErr } @@ -2135,9 +2250,6 @@ func (step awaitUserStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecisio return nil, err } if len(queued) > 0 { - if clearErr := clearInactivityDeadline(ctx); clearErr != nil { - return nil, clearErr - } if _, beginErr := step.flow.beginUserTurn(ctx, queued[0].Value); beginErr != nil { return nil, beginErr } @@ -2151,13 +2263,6 @@ func (step awaitUserStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecisio return nil, err } if plan == nil { - if ctx.HasTimerFired() { - deadline, deadlineErr := inactivityDeadlineAttribute.Get(ctx) - if deadlineErr != nil { - return nil, deadlineErr - } - return dex.GoTo(beginInactivityExpirationStep{flow: step.flow}, deadline.UTC()), nil - } return nil, errors.New("the Agent wait completed without an input") } executions, err := planExecutionsChannel.GetConditionResults(ctx, planRevisionKey(plan.Revision)) @@ -2165,19 +2270,10 @@ func (step awaitUserStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecisio return nil, err } if len(executions) > 0 { - if clearErr := clearInactivityDeadline(ctx); clearErr != nil { - return nil, clearErr - } revision := executions[0].Revision if consumptionErr := step.flow.writeInputConsumption(ctx, nil, nil, &revision); consumptionErr != nil { return nil, consumptionErr } - } else if ctx.HasTimerFired() { - deadline, deadlineErr := inactivityDeadlineAttribute.Get(ctx) - if deadlineErr != nil { - return nil, deadlineErr - } - return dex.GoTo(beginInactivityExpirationStep{flow: step.flow}, deadline.UTC()), nil } state, err := agentStateAttribute.Get(ctx) if err != nil { @@ -2796,11 +2892,10 @@ func (step awaitToolApprovalStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wai return nil, err } if steeredUserMessagesChannel.Size(ctx) == 0 && toolApprovalsChannel.Size(ctx, string(call.ID)) == 0 { - return step.flow.waitForUserActivity( - ctx, + return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), toolApprovalsChannel.ForOne(string(call.ID)), - ) + ), nil } return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), @@ -2814,9 +2909,6 @@ func (step awaitToolApprovalStep) Execute(ctx dex.Context, _ dex.None) (*dex.Ste return nil, err } if len(steered) > 0 { - if clearErr := clearInactivityDeadline(ctx); clearErr != nil { - return nil, clearErr - } if beginErr := step.flow.beginSteeredTurn(ctx, steered); beginErr != nil { return nil, beginErr } @@ -2831,18 +2923,8 @@ func (step awaitToolApprovalStep) Execute(ctx dex.Context, _ dex.None) (*dex.Ste return nil, err } if len(approvals) == 0 { - if ctx.HasTimerFired() { - deadline, deadlineErr := inactivityDeadlineAttribute.Get(ctx) - if deadlineErr != nil { - return nil, deadlineErr - } - return dex.GoTo(beginInactivityExpirationStep{flow: step.flow}, deadline.UTC()), nil - } return nil, errors.New("the approval wait completed without a decision") } - if clearErr := clearInactivityDeadline(ctx); clearErr != nil { - return nil, clearErr - } callID := call.ID toolName := call.Name resolutionMessage := "Tool approval rejected." @@ -3379,11 +3461,10 @@ func (step awaitManualToolRecoveryStep) WaitFor( } if steeredUserMessagesChannel.Size(ctx) == 0 && toolRecoveryDecisionsChannel.Size(ctx, string(pending.RecoveryID)) == 0 { - return step.flow.waitForUserActivity( - ctx, + return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), toolRecoveryDecisionsChannel.ForOne(string(pending.RecoveryID)), - ) + ), nil } return dex.AnyOf( steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), @@ -3406,9 +3487,6 @@ func (step awaitManualToolRecoveryStep) Execute( return nil, conditionErr } if len(steered) > 0 { - if err := clearInactivityDeadline(ctx); err != nil { - return nil, err - } if err := step.flow.writeActivity(ctx, AgentEvent{ Kind: EventKindToolRecoveryResolved, Message: "Manual tool recovery was interrupted by steering.", CallID: &callID, ToolName: &toolName, @@ -3439,21 +3517,9 @@ func (step awaitManualToolRecoveryStep) Execute( if decisionErr != nil { return nil, decisionErr } - if len(requests) == 0 { - if ctx.HasTimerFired() { - deadline, deadlineErr := inactivityDeadlineAttribute.Get(ctx) - if deadlineErr != nil { - return nil, deadlineErr - } - return dex.GoTo(beginInactivityExpirationStep{flow: step.flow}, deadline.UTC()), nil - } - } if len(requests) != 1 { return nil, errors.New("manual recovery wait completed without one decision") } - if err := clearInactivityDeadline(ctx); err != nil { - return nil, err - } request := requests[0] if err := validateToolRecoveryResolution(pendingToolRecovery(batch), request); err != nil { return nil, err @@ -3627,22 +3693,37 @@ var _ dex.Step[dex.None] = durableWaitStep{} func (durableWaitStep) GetStepType() string { return string(stepTypeDurableWait) } -func (durableWaitStep) GetStepOptions() *dex.StepOptions { return messageMutationStepOptions } +func (durableWaitStep) GetStepOptions() *dex.StepOptions { return durableWaitStepOptions } func (step durableWaitStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wait, error) { - if err := clearInactivityDeadline(ctx); err != nil { - return nil, err - } timer, err := pendingTimerAttribute.Get(ctx) if err != nil { return nil, err } - if err := step.flow.updateStatus(ctx, AgentStatusWaitingForTimer); err != nil { - return nil, err + if statusErr := step.flow.updateStatus(ctx, AgentStatusWaitingForTimer); statusErr != nil { + return nil, statusErr } - if err := step.flow.writeSnapshotRequired(ctx); err != nil { + if snapshotErr := step.flow.writeSnapshotRequired(ctx); snapshotErr != nil { + return nil, snapshotErr + } + config, err := agentConfigAttribute.Get(ctx) + if err != nil { return nil, err } + if config.InactivityTimeoutSeconds > 0 { + if timer.StartedAt == nil { + return nil, errors.New("durable timer start is missing") + } + waitTarget := timer.StartedAt.UTC().Add(time.Duration(timer.DurationSeconds) * time.Second) + resetTarget := waitTarget.Add(time.Duration(config.InactivityTimeoutSeconds) * time.Second) + accepted, resetErr := step.flow.resetInactivityTimer(ctx, resetTarget) + if resetErr != nil { + return nil, resetErr + } + if !accepted { + return nil, errors.New("durable wait started after inactivity expiration") + } + } return dex.AnyOf( dex.Timer(time.Duration(timer.DurationSeconds)*time.Second), steeredUserMessagesChannel.AtLeastAtMost(1, maximumSteeringMessageCount), @@ -3717,59 +3798,69 @@ func (step durableWaitStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecis return step.flow.continueAfterTool(ctx) } -type beginInactivityExpirationStep struct { - dex.StepDefaultsNoWaitFor[time.Time] +type inactivityTimeoutStep struct { + dex.StepDefaults flow *Flow } -var _ dex.Step[time.Time] = beginInactivityExpirationStep{} +var _ dex.Step[time.Time] = inactivityTimeoutStep{} -func (beginInactivityExpirationStep) GetStepType() string { - return string(stepTypeBeginExpiration) +func (inactivityTimeoutStep) GetStepType() string { + return string(stepTypeInactivityTimeout) } -func (beginInactivityExpirationStep) GetStepOptions() *dex.StepOptions { - return defaultStepOptions +func (inactivityTimeoutStep) GetStepOptions() *dex.StepOptions { + return inactivityExpirationStepOptions } -func (step beginInactivityExpirationStep) Execute( +func (inactivityTimeoutStep) WaitFor( ctx dex.Context, deadline time.Time, -) (*dex.StepDecision, error) { - if err := clearInactivityDeadline(ctx); err != nil { +) (*dex.Wait, error) { + currentDeadline, err := inactivityDeadlineAttribute.Get(ctx) + if err != nil { return nil, err } - if err := step.flow.updateStatus(ctx, AgentStatusExpiring); err != nil { - return nil, err + if currentDeadline.After(deadline) || resetInactivityTimerChannel.Size(ctx) > 0 { + return dex.Until(resetInactivityTimerChannel.AtMost(maximumSteeringMessageCount)), nil } - if err := step.flow.writeActivity(ctx, AgentEvent{ - Kind: EventKindInactivityExpired, - Message: "Agent inactivity timeout expired.", - }); err != nil { - return nil, err + remaining := deadline.UTC().Sub(ctx.FirstAttemptAt().UTC()) + if remaining <= 0 { + remaining = time.Nanosecond } - return dex.GoTo(expireInactivityStep{flow: step.flow}, deadline.UTC()), nil -} - -type expireInactivityStep struct { - dex.StepDefaultsNoWaitFor[time.Time] - flow *Flow -} - -var _ dex.Step[time.Time] = expireInactivityStep{} - -func (expireInactivityStep) GetStepType() string { - return string(stepTypeExpireInactivity) -} - -func (expireInactivityStep) GetStepOptions() *dex.StepOptions { - return inactivityExpirationStepOptions + return dex.AnyOf( + dex.Timer(remaining), + resetInactivityTimerChannel.AtLeastAtMost(1, maximumSteeringMessageCount), + ), nil } -func (step expireInactivityStep) Execute( +func (step inactivityTimeoutStep) Execute( ctx dex.Context, deadline time.Time, ) (*dex.StepDecision, error) { + resetTargets, err := resetInactivityTimerChannel.GetConditionResults(ctx) + if err != nil { + return nil, err + } + currentDeadline, err := inactivityDeadlineAttribute.Get(ctx) + if err != nil { + return nil, err + } + latestDeadline := currentDeadline.UTC() + for _, resetTarget := range resetTargets { + if resetTarget.After(latestDeadline) { + latestDeadline = resetTarget.UTC() + } + } + if latestDeadline.After(deadline.UTC()) { + if setErr := inactivityDeadlineAttribute.Set(ctx, latestDeadline); setErr != nil { + return nil, setErr + } + return dex.GoTo(inactivityTimeoutStep{flow: step.flow}, latestDeadline), nil + } + if !ctx.HasTimerFired() { + return nil, errors.New("inactivity wait completed without a reset or deadline") + } if step.flow.inactivityExpiration == nil { return nil, errors.New("inactivity expiration handler is not configured") } @@ -3780,12 +3871,24 @@ func (step expireInactivityStep) Execute( expiration := InactivityExpiration{ FlowID: FlowID(ctx.FlowID()), RunID: RunID(ctx.RunID()), - Deadline: deadline.UTC(), + Deadline: latestDeadline, RuntimeMetadata: metadata, } if err := step.flow.inactivityExpiration.HandleInactivityExpiration(ctx, expiration); err != nil { return nil, err } + if err := clearInactivityDeadline(ctx); err != nil { + return nil, err + } + if err := step.flow.updateStatus(ctx, AgentStatusExpiring); err != nil { + return nil, err + } + if err := step.flow.writeActivity(ctx, AgentEvent{ + Kind: EventKindInactivityExpired, + Message: "Agent inactivity timeout expired.", + }); err != nil { + return nil, err + } return dex.ForceComplete("inactivity_expired"), nil } diff --git a/internal/agent/flow_integration_test.go b/internal/agent/flow_integration_test.go index 7ec6b1e..394aefc 100644 --- a/internal/agent/flow_integration_test.go +++ b/internal/agent/flow_integration_test.go @@ -44,6 +44,57 @@ const integrationToolName ToolName = "integration_tool" const integrationWaitTimeout = 30 * time.Second func TestAgentInactivityExpirationIntegration(t *testing.T) { + t.Run("disabled timeout has no deadline", func(t *testing.T) { + environment := newAgentIntegrationEnvironment( + t, integrationModel{}, newIntegrationToolRegistry(), + ) + flowID := FlowID("agent-inactivity-disabled-" + randomLocalID(t)) + if _, err := environment.agent.Start(t.Context(), flowID, StartRequest{Config: NewAgentConfig()}); err != nil { + t.Fatal(err) + } + snapshot := readSnapshot(t, environment, flowID) + if snapshot.Description.InactivityDeadline != nil { + t.Fatalf("disabled Snapshot = %#v", snapshot.Description) + } + }) + + t.Run("accepted user operation advances the global deadline", func(t *testing.T) { + handler := newIntegrationExpirationHandler(0) + environment := newAgentIntegrationEnvironment( + t, + integrationModel{}, + newIntegrationToolRegistry(), + WithInactivityExpirationHandler(handler), + ) + flowID := FlowID("agent-inactivity-reset-" + randomLocalID(t)) + config := NewAgentConfig() + config.InactivityTimeoutSeconds = 2 + if _, err := environment.agent.Start(t.Context(), flowID, StartRequest{Config: config}); err != nil { + t.Fatal(err) + } + initial := readSnapshot(t, environment, flowID) + if initial.Description.InactivityDeadline == nil { + t.Fatalf("initial Snapshot = %#v", initial.Description) + } + delay := time.NewTimer(200 * time.Millisecond) + defer delay.Stop() + select { + case <-delay.C: + case <-t.Context().Done(): + t.Fatal(t.Context().Err()) + } + if err := environment.agent.SendMessage(t.Context(), flowID, UserMessage{Content: "reset deadline"}); err != nil { + t.Fatal(err) + } + advanced := waitForSnapshot(t, environment, flowID, func(snapshot AgentSnapshot) bool { + return snapshot.Description.InactivityDeadline != nil && + snapshot.Description.InactivityDeadline.After(*initial.Description.InactivityDeadline) + }) + if advanced.Description.InactivityDeadline == nil { + t.Fatalf("advanced Snapshot = %#v", advanced.Description) + } + }) + t.Run("message wait expires once and survives Worker replacement", func(t *testing.T) { handler := newIntegrationExpirationHandler(0) environment := newAgentIntegrationEnvironment( @@ -147,14 +198,14 @@ func TestAgentInactivityExpirationIntegration(t *testing.T) { } }) - t.Run("durable wait pauses and restarts the full window", func(t *testing.T) { + t.Run("durable wait extends the global deadline", func(t *testing.T) { handler := newIntegrationExpirationHandler(0) environment := newAgentIntegrationEnvironment( t, integrationModel{}, newIntegrationToolRegistry(), WithInactivityExpirationHandler(handler), ) flowID := FlowID("agent-inactivity-durable-wait-" + randomLocalID(t)) config := NewAgentConfig() - config.InactivityTimeoutSeconds = 1 + config.InactivityTimeoutSeconds = 3 if _, err := environment.agent.Start(t.Context(), flowID, StartRequest{Config: config}); err != nil { t.Fatal(err) } @@ -163,7 +214,7 @@ func TestAgentInactivityExpirationIntegration(t *testing.T) { } waitForSnapshot(t, environment, flowID, func(snapshot AgentSnapshot) bool { return snapshot.Description.Status == AgentStatusWaitingForTimer && - snapshot.Description.PendingTimer != nil && snapshot.Description.InactivityDeadline == nil + snapshot.Description.PendingTimer != nil && snapshot.Description.InactivityDeadline != nil }) observation := time.NewTimer(1500 * time.Millisecond) defer observation.Stop() @@ -183,11 +234,10 @@ func TestAgentInactivityExpirationIntegration(t *testing.T) { return snapshot.Description.Status == AgentStatusWaitingForMessage && snapshot.Description.InactivityDeadline != nil }) - if rearmed.Description.InactivityDeadline == nil || time.Until(*rearmed.Description.InactivityDeadline) <= 0 { + if rearmed.Description.InactivityDeadline == nil || time.Until(*rearmed.Description.InactivityDeadline) < 20*time.Second { t.Fatalf("rearmed Snapshot = %#v", rearmed.Description) } - _ = handler.waitForSuccess(t) - assertInactivityCompleted(t, environment, flowID) + handler.assertAttempts(t, 0) }) t.Run("callback retries with one stable expiration", func(t *testing.T) { @@ -1949,6 +1999,8 @@ func newAgentIntegrationEnvironment( ) *agentIntegrationEnvironment { t.Helper() flow := NewFlow(modelClient, tools, options...) + flow.minimumInactivityTimeoutSecondsForTestOnly = 1 + flow.inactivityResetMinimumExtensionForTestOnly = 100 * time.Millisecond registerRPCDefinitionsForTestOnly(flow) environment := &agentIntegrationEnvironment{ flow: flow, diff --git a/internal/agent/tool_recovery_test.go b/internal/agent/tool_recovery_test.go index 0d1a117..585aa80 100644 --- a/internal/agent/tool_recovery_test.go +++ b/internal/agent/tool_recovery_test.go @@ -100,7 +100,7 @@ func TestToolDefinitionsExposeFailureSimulationOnlyToLocalMock(t *testing.T) { func TestValidateConfigRequiresExpirationHandlerWhenTimeoutEnabled(t *testing.T) { config := NewAgentConfig() - config.InactivityTimeoutSeconds = 60 + config.InactivityTimeoutSeconds = 120 flow := &Flow{tools: staticToolRegistryForTestOnly{}} if err := flow.validateConfig(config); err == nil { t.Fatal("enabled inactivity timeout without handler validated") diff --git a/internal/agent/types.go b/internal/agent/types.go index ccad3af..991ebee 100644 --- a/internal/agent/types.go +++ b/internal/agent/types.go @@ -576,7 +576,8 @@ type AgentConfig struct { EnabledMCPServers []string `json:"enabled_mcp_servers"` // EnabledTools restricts tools; empty selects all tools on enabled servers. EnabledTools []ToolName `json:"enabled_tools"` - // InactivityTimeoutSeconds closes the Agent after an idle user wait; zero disables it. + // InactivityTimeoutSeconds closes the Agent after user inactivity; zero disables it. + // Positive values must be between two minutes and one year. InactivityTimeoutSeconds int64 `json:"inactivity_timeout_seconds"` } @@ -604,6 +605,10 @@ func NewAgentConfig() AgentConfig { // Validate rejects configurations that cannot execute safely. func (config AgentConfig) Validate() error { + return config.validate(int64((2 * time.Minute) / time.Second)) +} + +func (config AgentConfig) validate(minimumInactivityTimeoutSeconds int64) error { if _, err := config.Model.Provider(); err != nil { return fmt.Errorf("model: %w", err) } @@ -629,6 +634,12 @@ func (config AgentConfig) Validate() error { return fmt.Errorf("max_parallel_tool_calls must be between 1 and %d when set", MaximumParallelToolCalls) case config.InactivityTimeoutSeconds < 0: return errors.New("inactivity_timeout_seconds must not be negative") + case config.InactivityTimeoutSeconds > 0 && + config.InactivityTimeoutSeconds < minimumInactivityTimeoutSeconds: + return fmt.Errorf( + "inactivity_timeout_seconds must be zero or at least %d", + minimumInactivityTimeoutSeconds, + ) case config.InactivityTimeoutSeconds > int64((365*24*time.Hour)/time.Second): return errors.New("inactivity_timeout_seconds must not exceed one year") default: diff --git a/internal/agent/types_test.go b/internal/agent/types_test.go index b36b013..8b2e425 100644 --- a/internal/agent/types_test.go +++ b/internal/agent/types_test.go @@ -56,6 +56,7 @@ func TestAgentConfigRejectsInvalidValues(t *testing.T) { {"negative parallel calls", func(config *AgentConfig) { config.MaxParallelToolCalls = -1 }}, {"too many parallel calls", func(config *AgentConfig) { config.MaxParallelToolCalls = 33 }}, {"negative inactivity timeout", func(config *AgentConfig) { config.InactivityTimeoutSeconds = -1 }}, + {"short inactivity timeout", func(config *AgentConfig) { config.InactivityTimeoutSeconds = 60 }}, {"excessive inactivity timeout", func(config *AgentConfig) { config.InactivityTimeoutSeconds = int64((366 * 24 * time.Hour) / time.Second) }}, From 68b647511fdd78d17854ece70df3b4deed1ef2b4 Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Wed, 23 Sep 2026 22:21:53 -0700 Subject: [PATCH 2/4] Preserve inactivity reset timestamps --- ARCHITECTURE.md | 7 +-- .../0016-agent-owned-inactivity-expiration.md | 2 + docs/flow-model.md | 7 +-- internal/agent/client.go | 21 +++++--- internal/agent/flow.go | 54 ++++++++++++------- internal/agent/flow_integration_test.go | 6 ++- internal/agent/types.go | 35 ++++++++++++ 7 files changed, 97 insertions(+), 35 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index dc45d3a..bf3ba08 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -79,9 +79,10 @@ An embedding application may configure `AgentConfig.InactivityTimeoutSeconds` and inject `InactivityExpirationHandler`. The Agent owns the inactivity decision through one `InactivityTimeout` branch started beside the Agent loop. Accepted user RPCs atomically advance `InactivityDeadline` and publish -`ResetInactivityTimer` only when the extension exceeds one minute. Model and -ordinary tool work do not pause the deadline. `DurableWait` extends it through -the wait target. Expiration retries the stable Flow ID, Run ID, deadline, and +`ResetInactivityTimer` only when the extension exceeds one minute. The Agent +Client records the operation timestamp once before any RPC lock retry. Model +and ordinary tool work do not pause the deadline. `DurableWait` extends it +through the wait target. Expiration retries the stable Flow ID, Run ID, deadline, and runtime-metadata callback for up to seven days. Callback success commits `expiring`, emits `inactivity_expired`, and force-completes the Agent; cross- resource cleanup remains the embedding application's job. diff --git a/docs/adr/0016-agent-owned-inactivity-expiration.md b/docs/adr/0016-agent-owned-inactivity-expiration.md index b4597c5..6481cd0 100644 --- a/docs/adr/0016-agent-owned-inactivity-expiration.md +++ b/docs/adr/0016-agent-owned-inactivity-expiration.md @@ -18,6 +18,8 @@ to every resource owner and creates competing shutdown decisions. `inactivity_timeout_seconds` starts one parallel `InactivityTimeout` Step and persists `InactivityDeadline`. Accepted user RPCs advance the Attribute and publish `ResetInactivityTimer` only when the extension exceeds one minute. +The Agent Client captures one operation timestamp before any RPC lock retry and +keeps it in the command envelope so every attempt computes the same deadline. Model and ordinary tool work do not pause the deadline. `DurableWait` advances it to the wait target plus the configured timeout. diff --git a/docs/flow-model.md b/docs/flow-model.md index 937df40..641edd9 100644 --- a/docs/flow-model.md +++ b/docs/flow-model.md @@ -216,9 +216,10 @@ When `inactivity_timeout_seconds` is positive, `Init` starts one parallel `InactivityTimeout` Step and persists its exact `InactivityDeadline`. Accepted messages, answers, plan execution, steering, queue deletion, approvals, and recovery decisions advance the deadline transactionally. Extensions of one -minute or less do not publish another reset. Snapshot and Stream reads do not -reset it. Model calls and serial or parallel tools continue under the same -deadline. `DurableWait` advances it to the wait target plus the timeout. +minute or less do not publish another reset. The Agent Client captures the +operation timestamp once before any RPC lock retry. Snapshot and Stream reads +do not reset it. Model calls and serial or parallel tools continue under the +same deadline. `DurableWait` advances it to the wait target plus the timeout. ## Durable resources diff --git a/internal/agent/client.go b/internal/agent/client.go index c801feb..3d2fc07 100644 --- a/internal/agent/client.go +++ b/internal/agent/client.go @@ -115,8 +115,9 @@ func (client *Client) SendMessage(ctx context.Context, flowID FlowID, message Us return err } pending := PendingUserMessage{MessageID: MessageID(uuid.NewString()), Value: message} + request := sendMessageRPCInput{FirstAttemptAt: time.Now().UTC(), Value: pending} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SendMessage, pending, accepted) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SendMessage, request, accepted) }) if err != nil { return err @@ -136,8 +137,9 @@ func (client *Client) AnswerQuestions( if err := validateAnswerQuestionsRequest(request); err != nil { return err } + rpcRequest := answerQuestionsRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.AnswerQuestions, request, accepted) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.AnswerQuestions, rpcRequest, accepted) }) if err != nil { return err @@ -184,8 +186,9 @@ func (client *Client) SteerMessage(ctx context.Context, flowID FlowID, request S if err := validateMessageID(request.MessageID); err != nil { return err } + rpcRequest := steerMessageRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SteerMessage, request, accepted) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SteerMessage, rpcRequest, accepted) }) if err != nil { return err @@ -284,11 +287,12 @@ func (client *Client) DeleteQueuedMessage(ctx context.Context, flowID FlowID, me return err } var deleted bool + request := deleteQueuedMessageRPCInput{FirstAttemptAt: time.Now().UTC(), Value: messageID} if err := client.sdk.InvokeRPC( ctx, string(flowID), client.flow.DeleteQueuedMessage, - messageID, + request, &deleted, ); err != nil { return err @@ -308,7 +312,8 @@ func (client *Client) ApproveTool(ctx context.Context, flowID FlowID, request To return errors.New("call ID must not be empty") } var accepted bool - if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ApproveTool, request, &accepted); err != nil { + rpcRequest := approveToolRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} + if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ApproveTool, rpcRequest, &accepted); err != nil { return err } return ensureAccepted(accepted, CommandApproveTool) @@ -337,12 +342,13 @@ func (client *Client) ResolveToolRecovery( return err } } + rpcRequest := resolveToolRecoveryRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { return client.sdk.InvokeRPC( ctx, string(flowID), client.flow.ResolveToolRecovery, - request, + rpcRequest, accepted, ) }) @@ -361,7 +367,8 @@ func (client *Client) ExecutePlan(ctx context.Context, flowID FlowID, request Pl return errors.New("plan revision must be positive") } var accepted bool - if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ExecutePlan, request, &accepted); err != nil { + rpcRequest := executePlanRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} + if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ExecutePlan, rpcRequest, &accepted); err != nil { return err } return ensureAccepted(accepted, CommandExecutePlan) diff --git a/internal/agent/flow.go b/internal/agent/flow.go index 2301297..a412984 100644 --- a/internal/agent/flow.go +++ b/internal/agent/flow.go @@ -252,7 +252,8 @@ func (*Flow) GetPersistenceSchema() dex.PersistenceSchema { } // SendMessage queues one non-empty user message when no question is pending. -func (flow *Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.RPCResult[bool], error) { +func (flow *Flow) SendMessage(ctx dex.Context, request sendMessageRPCInput) (*dex.RPCResult[bool], error) { + input := request.Value if strings.TrimSpace(input.Value.Content) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -266,7 +267,7 @@ func (flow *Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.R if pending != nil { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) if err != nil { return nil, err } @@ -280,7 +281,8 @@ func (flow *Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.R } // AnswerQuestions validates and publishes one exact pending input batch atomically. -func (flow *Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) AnswerQuestions(ctx dex.Context, request answerQuestionsRPCInput) (*dex.RPCResult[bool], error) { + input := request.Value if strings.TrimSpace(string(input.CallID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -295,7 +297,7 @@ func (flow *Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) if !isValid { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) if err != nil { return nil, err } @@ -316,7 +318,8 @@ func (flow *Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) } // SteerMessage atomically moves a queued message into the Steer queue. -func (flow *Flow) SteerMessage(ctx dex.Context, input SteerMessageRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) SteerMessage(ctx dex.Context, request steerMessageRPCInput) (*dex.RPCResult[bool], error) { + input := request.Value if strings.TrimSpace(string(input.MessageID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -343,7 +346,7 @@ func (flow *Flow) SteerMessage(ctx dex.Context, input SteerMessageRequest) (*dex } shouldDeletePendingToolRecovery = true } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) if err != nil { return nil, err } @@ -450,7 +453,8 @@ func (*Flow) GetArchivedMessages( } // DeleteQueuedMessage removes one exact pending user message. -func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*dex.RPCResult[bool], error) { +func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, request deleteQueuedMessageRPCInput) (*dex.RPCResult[bool], error) { + messageID := request.Value if strings.TrimSpace(string(messageID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -462,7 +466,7 @@ func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*de if !found { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) if err != nil { return nil, err } @@ -476,7 +480,8 @@ func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*de } // ApproveTool publishes an approval only for the current exact call ID. -func (flow *Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) ApproveTool(ctx dex.Context, request approveToolRPCInput) (*dex.RPCResult[bool], error) { + input := request.Value pending, err := getPendingApproval(ctx) if err != nil { return nil, err @@ -484,7 +489,7 @@ func (flow *Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex. if pending == nil || pending.CallID != input.CallID { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) if err != nil { return nil, err } @@ -503,8 +508,9 @@ func (flow *Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex. // ResolveToolRecovery publishes one complete decision for the exact pending recovery revision. func (flow *Flow) ResolveToolRecovery( ctx dex.Context, - input ResolveToolRecoveryRequest, + request resolveToolRecoveryRPCInput, ) (*dex.RPCResult[bool], error) { + input := request.Value pending, err := getPendingToolRecovery(ctx) if err != nil { return nil, err @@ -515,7 +521,7 @@ func (flow *Flow) ResolveToolRecovery( if !isValidToolRecoveryResolution(*pending, input) { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) if err != nil { return nil, err } @@ -532,7 +538,8 @@ func (flow *Flow) ResolveToolRecovery( } // ExecutePlan schedules an exact waiting draft or active plan revision. -func (flow *Flow) ExecutePlan(ctx dex.Context, input PlanExecutionRequest) (*dex.RPCResult[bool], error) { +func (flow *Flow) ExecutePlan(ctx dex.Context, request executePlanRPCInput) (*dex.RPCResult[bool], error) { + input := request.Value state, err := agentStateAttribute.Get(ctx) if err != nil { return nil, err @@ -571,7 +578,7 @@ func (flow *Flow) ExecutePlan(ctx dex.Context, input PlanExecutionRequest) (*dex if !canExecute { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) if err != nil { return nil, err } @@ -673,7 +680,10 @@ func clearInactivityDeadline(ctx dex.Context) error { return err } -func (flow *Flow) resetInactivityTimerForUserOperation(ctx dex.Context) (bool, error) { +func (flow *Flow) resetInactivityTimerForUserOperation( + ctx dex.Context, + firstAttemptAt time.Time, +) (bool, error) { config, err := agentConfigAttribute.Get(ctx) if err != nil { return false, err @@ -681,18 +691,22 @@ func (flow *Flow) resetInactivityTimerForUserOperation(ctx dex.Context) (bool, e if config.InactivityTimeoutSeconds == 0 { return true, nil } - resetTarget := ctx.FirstAttemptAt().UTC().Add( + resetTarget := firstAttemptAt.UTC().Add( time.Duration(config.InactivityTimeoutSeconds) * time.Second, ) - return flow.resetInactivityTimer(ctx, resetTarget) + return flow.resetInactivityTimer(ctx, firstAttemptAt, resetTarget) } -func (flow *Flow) resetInactivityTimer(ctx dex.Context, resetTarget time.Time) (bool, error) { +func (flow *Flow) resetInactivityTimer( + ctx dex.Context, + activityAt time.Time, + resetTarget time.Time, +) (bool, error) { deadline, err := inactivityDeadlineAttribute.Get(ctx) if err != nil { return false, err } - if !deadline.After(ctx.FirstAttemptAt().UTC()) { + if !deadline.After(activityAt.UTC()) { return false, nil } resetTarget = resetTarget.UTC() @@ -3716,7 +3730,7 @@ func (step durableWaitStep) WaitFor(ctx dex.Context, _ dex.None) (*dex.Wait, err } waitTarget := timer.StartedAt.UTC().Add(time.Duration(timer.DurationSeconds) * time.Second) resetTarget := waitTarget.Add(time.Duration(config.InactivityTimeoutSeconds) * time.Second) - accepted, resetErr := step.flow.resetInactivityTimer(ctx, resetTarget) + accepted, resetErr := step.flow.resetInactivityTimer(ctx, ctx.FirstAttemptAt(), resetTarget) if resetErr != nil { return nil, resetErr } diff --git a/internal/agent/flow_integration_test.go b/internal/agent/flow_integration_test.go index 394aefc..47f3094 100644 --- a/internal/agent/flow_integration_test.go +++ b/internal/agent/flow_integration_test.go @@ -68,7 +68,7 @@ func TestAgentInactivityExpirationIntegration(t *testing.T) { ) flowID := FlowID("agent-inactivity-reset-" + randomLocalID(t)) config := NewAgentConfig() - config.InactivityTimeoutSeconds = 2 + config.InactivityTimeoutSeconds = 5 if _, err := environment.agent.Start(t.Context(), flowID, StartRequest{Config: config}); err != nil { t.Fatal(err) } @@ -76,7 +76,9 @@ func TestAgentInactivityExpirationIntegration(t *testing.T) { if initial.Description.InactivityDeadline == nil { t.Fatalf("initial Snapshot = %#v", initial.Description) } - delay := time.NewTimer(200 * time.Millisecond) + // Cross the durable clock's one-second precision before testing the + // 100ms reset-coalescing threshold. + delay := time.NewTimer(1500 * time.Millisecond) defer delay.Stop() select { case <-delay.C: diff --git a/internal/agent/types.go b/internal/agent/types.go index 991ebee..94a984f 100644 --- a/internal/agent/types.go +++ b/internal/agent/types.go @@ -753,6 +753,41 @@ type PendingUserMessage struct { Value UserMessage `json:"value"` } +type sendMessageRPCInput struct { + FirstAttemptAt time.Time `json:"first_attempt_at"` + Value PendingUserMessage `json:"value"` +} + +type answerQuestionsRPCInput struct { + FirstAttemptAt time.Time `json:"first_attempt_at"` + Value AnswerQuestionsRequest `json:"value"` +} + +type steerMessageRPCInput struct { + FirstAttemptAt time.Time `json:"first_attempt_at"` + Value SteerMessageRequest `json:"value"` +} + +type deleteQueuedMessageRPCInput struct { + FirstAttemptAt time.Time `json:"first_attempt_at"` + Value MessageID `json:"value"` +} + +type approveToolRPCInput struct { + FirstAttemptAt time.Time `json:"first_attempt_at"` + Value ToolApprovalRequest `json:"value"` +} + +type resolveToolRecoveryRPCInput struct { + FirstAttemptAt time.Time `json:"first_attempt_at"` + Value ResolveToolRecoveryRequest `json:"value"` +} + +type executePlanRPCInput struct { + FirstAttemptAt time.Time `json:"first_attempt_at"` + Value PlanExecutionRequest `json:"value"` +} + // AgentDescription is the durable application state needed to render a conversation. type AgentDescription struct { Status AgentStatus `json:"status"` From aceb5e891ff6c176f1b8f69a04d3c0b776768519 Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Wed, 23 Sep 2026 22:26:59 -0700 Subject: [PATCH 3/4] Clarify durable wait deadline --- ARCHITECTURE.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index bf3ba08..b5d2f8f 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -82,8 +82,9 @@ Accepted user RPCs atomically advance `InactivityDeadline` and publish `ResetInactivityTimer` only when the extension exceeds one minute. The Agent Client records the operation timestamp once before any RPC lock retry. Model and ordinary tool work do not pause the deadline. `DurableWait` extends it -through the wait target. Expiration retries the stable Flow ID, Run ID, deadline, and -runtime-metadata callback for up to seven days. Callback success commits +to the wait target plus the configured timeout. Expiration retries the stable +Flow ID, Run ID, deadline, and runtime-metadata callback for up to seven days. +Callback success commits `expiring`, emits `inactivity_expired`, and force-completes the Agent; cross- resource cleanup remains the embedding application's job. From 83324563357e58bc70c70e0fa3c34ab74e7513c0 Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Wed, 23 Sep 2026 22:40:52 -0700 Subject: [PATCH 4/4] Simplify inactivity RPC resets --- ARCHITECTURE.md | 6 +-- .../0016-agent-owned-inactivity-expiration.md | 3 +- docs/flow-model.md | 8 ++-- internal/agent/client.go | 21 +++------ internal/agent/flow.go | 45 ++++++++----------- internal/agent/flow_integration_test.go | 4 +- internal/agent/types.go | 35 --------------- 7 files changed, 34 insertions(+), 88 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index b5d2f8f..a3e37cd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -80,9 +80,9 @@ and inject `InactivityExpirationHandler`. The Agent owns the inactivity decision through one `InactivityTimeout` branch started beside the Agent loop. Accepted user RPCs atomically advance `InactivityDeadline` and publish `ResetInactivityTimer` only when the extension exceeds one minute. The Agent -Client records the operation timestamp once before any RPC lock retry. Model -and ordinary tool work do not pause the deadline. `DurableWait` extends it -to the wait target plus the configured timeout. Expiration retries the stable +RPC handler samples the current time after validating the accepted operation. +Model and ordinary tool work do not pause the deadline. `DurableWait` extends +it to the wait target plus the configured timeout. Expiration retries the stable Flow ID, Run ID, deadline, and runtime-metadata callback for up to seven days. Callback success commits `expiring`, emits `inactivity_expired`, and force-completes the Agent; cross- diff --git a/docs/adr/0016-agent-owned-inactivity-expiration.md b/docs/adr/0016-agent-owned-inactivity-expiration.md index 6481cd0..551eb03 100644 --- a/docs/adr/0016-agent-owned-inactivity-expiration.md +++ b/docs/adr/0016-agent-owned-inactivity-expiration.md @@ -18,8 +18,7 @@ to every resource owner and creates competing shutdown decisions. `inactivity_timeout_seconds` starts one parallel `InactivityTimeout` Step and persists `InactivityDeadline`. Accepted user RPCs advance the Attribute and publish `ResetInactivityTimer` only when the extension exceeds one minute. -The Agent Client captures one operation timestamp before any RPC lock retry and -keeps it in the command envelope so every attempt computes the same deadline. +The accepted RPC handler samples the current time after business validation. Model and ordinary tool work do not pause the deadline. `DurableWait` advances it to the wait target plus the configured timeout. diff --git a/docs/flow-model.md b/docs/flow-model.md index 641edd9..947eb50 100644 --- a/docs/flow-model.md +++ b/docs/flow-model.md @@ -216,10 +216,10 @@ When `inactivity_timeout_seconds` is positive, `Init` starts one parallel `InactivityTimeout` Step and persists its exact `InactivityDeadline`. Accepted messages, answers, plan execution, steering, queue deletion, approvals, and recovery decisions advance the deadline transactionally. Extensions of one -minute or less do not publish another reset. The Agent Client captures the -operation timestamp once before any RPC lock retry. Snapshot and Stream reads -do not reset it. Model calls and serial or parallel tools continue under the -same deadline. `DurableWait` advances it to the wait target plus the timeout. +minute or less do not publish another reset. The accepted RPC handler samples +the current time after business validation. Snapshot and Stream reads do not +reset it. Model calls and serial or parallel tools continue under the same +deadline. `DurableWait` advances it to the wait target plus the timeout. ## Durable resources diff --git a/internal/agent/client.go b/internal/agent/client.go index 3d2fc07..c801feb 100644 --- a/internal/agent/client.go +++ b/internal/agent/client.go @@ -115,9 +115,8 @@ func (client *Client) SendMessage(ctx context.Context, flowID FlowID, message Us return err } pending := PendingUserMessage{MessageID: MessageID(uuid.NewString()), Value: message} - request := sendMessageRPCInput{FirstAttemptAt: time.Now().UTC(), Value: pending} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SendMessage, request, accepted) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SendMessage, pending, accepted) }) if err != nil { return err @@ -137,9 +136,8 @@ func (client *Client) AnswerQuestions( if err := validateAnswerQuestionsRequest(request); err != nil { return err } - rpcRequest := answerQuestionsRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.AnswerQuestions, rpcRequest, accepted) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.AnswerQuestions, request, accepted) }) if err != nil { return err @@ -186,9 +184,8 @@ func (client *Client) SteerMessage(ctx context.Context, flowID FlowID, request S if err := validateMessageID(request.MessageID); err != nil { return err } - rpcRequest := steerMessageRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { - return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SteerMessage, rpcRequest, accepted) + return client.sdk.InvokeRPC(ctx, string(flowID), client.flow.SteerMessage, request, accepted) }) if err != nil { return err @@ -287,12 +284,11 @@ func (client *Client) DeleteQueuedMessage(ctx context.Context, flowID FlowID, me return err } var deleted bool - request := deleteQueuedMessageRPCInput{FirstAttemptAt: time.Now().UTC(), Value: messageID} if err := client.sdk.InvokeRPC( ctx, string(flowID), client.flow.DeleteQueuedMessage, - request, + messageID, &deleted, ); err != nil { return err @@ -312,8 +308,7 @@ func (client *Client) ApproveTool(ctx context.Context, flowID FlowID, request To return errors.New("call ID must not be empty") } var accepted bool - rpcRequest := approveToolRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} - if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ApproveTool, rpcRequest, &accepted); err != nil { + if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ApproveTool, request, &accepted); err != nil { return err } return ensureAccepted(accepted, CommandApproveTool) @@ -342,13 +337,12 @@ func (client *Client) ResolveToolRecovery( return err } } - rpcRequest := resolveToolRecoveryRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} accepted, err := invokeLockedCommand(ctx, client.commandTimeout, func(ctx context.Context, accepted *bool) error { return client.sdk.InvokeRPC( ctx, string(flowID), client.flow.ResolveToolRecovery, - rpcRequest, + request, accepted, ) }) @@ -367,8 +361,7 @@ func (client *Client) ExecutePlan(ctx context.Context, flowID FlowID, request Pl return errors.New("plan revision must be positive") } var accepted bool - rpcRequest := executePlanRPCInput{FirstAttemptAt: time.Now().UTC(), Value: request} - if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ExecutePlan, rpcRequest, &accepted); err != nil { + if err := client.sdk.InvokeRPC(ctx, string(flowID), client.flow.ExecutePlan, request, &accepted); err != nil { return err } return ensureAccepted(accepted, CommandExecutePlan) diff --git a/internal/agent/flow.go b/internal/agent/flow.go index a412984..62e04a6 100644 --- a/internal/agent/flow.go +++ b/internal/agent/flow.go @@ -252,8 +252,7 @@ func (*Flow) GetPersistenceSchema() dex.PersistenceSchema { } // SendMessage queues one non-empty user message when no question is pending. -func (flow *Flow) SendMessage(ctx dex.Context, request sendMessageRPCInput) (*dex.RPCResult[bool], error) { - input := request.Value +func (flow *Flow) SendMessage(ctx dex.Context, input PendingUserMessage) (*dex.RPCResult[bool], error) { if strings.TrimSpace(input.Value.Content) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -267,7 +266,7 @@ func (flow *Flow) SendMessage(ctx dex.Context, request sendMessageRPCInput) (*de if pending != nil { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) if err != nil { return nil, err } @@ -281,8 +280,7 @@ func (flow *Flow) SendMessage(ctx dex.Context, request sendMessageRPCInput) (*de } // AnswerQuestions validates and publishes one exact pending input batch atomically. -func (flow *Flow) AnswerQuestions(ctx dex.Context, request answerQuestionsRPCInput) (*dex.RPCResult[bool], error) { - input := request.Value +func (flow *Flow) AnswerQuestions(ctx dex.Context, input AnswerQuestionsRequest) (*dex.RPCResult[bool], error) { if strings.TrimSpace(string(input.CallID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -297,7 +295,7 @@ func (flow *Flow) AnswerQuestions(ctx dex.Context, request answerQuestionsRPCInp if !isValid { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) if err != nil { return nil, err } @@ -318,8 +316,7 @@ func (flow *Flow) AnswerQuestions(ctx dex.Context, request answerQuestionsRPCInp } // SteerMessage atomically moves a queued message into the Steer queue. -func (flow *Flow) SteerMessage(ctx dex.Context, request steerMessageRPCInput) (*dex.RPCResult[bool], error) { - input := request.Value +func (flow *Flow) SteerMessage(ctx dex.Context, input SteerMessageRequest) (*dex.RPCResult[bool], error) { if strings.TrimSpace(string(input.MessageID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -346,7 +343,7 @@ func (flow *Flow) SteerMessage(ctx dex.Context, request steerMessageRPCInput) (* } shouldDeletePendingToolRecovery = true } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) if err != nil { return nil, err } @@ -453,8 +450,7 @@ func (*Flow) GetArchivedMessages( } // DeleteQueuedMessage removes one exact pending user message. -func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, request deleteQueuedMessageRPCInput) (*dex.RPCResult[bool], error) { - messageID := request.Value +func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, messageID MessageID) (*dex.RPCResult[bool], error) { if strings.TrimSpace(string(messageID)) == "" { return &dex.RPCResult[bool]{Output: false}, nil } @@ -466,7 +462,7 @@ func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, request deleteQueuedMessa if !found { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) if err != nil { return nil, err } @@ -480,8 +476,7 @@ func (flow *Flow) DeleteQueuedMessage(ctx dex.Context, request deleteQueuedMessa } // ApproveTool publishes an approval only for the current exact call ID. -func (flow *Flow) ApproveTool(ctx dex.Context, request approveToolRPCInput) (*dex.RPCResult[bool], error) { - input := request.Value +func (flow *Flow) ApproveTool(ctx dex.Context, input ToolApprovalRequest) (*dex.RPCResult[bool], error) { pending, err := getPendingApproval(ctx) if err != nil { return nil, err @@ -489,7 +484,7 @@ func (flow *Flow) ApproveTool(ctx dex.Context, request approveToolRPCInput) (*de if pending == nil || pending.CallID != input.CallID { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) if err != nil { return nil, err } @@ -508,9 +503,8 @@ func (flow *Flow) ApproveTool(ctx dex.Context, request approveToolRPCInput) (*de // ResolveToolRecovery publishes one complete decision for the exact pending recovery revision. func (flow *Flow) ResolveToolRecovery( ctx dex.Context, - request resolveToolRecoveryRPCInput, + input ResolveToolRecoveryRequest, ) (*dex.RPCResult[bool], error) { - input := request.Value pending, err := getPendingToolRecovery(ctx) if err != nil { return nil, err @@ -521,7 +515,7 @@ func (flow *Flow) ResolveToolRecovery( if !isValidToolRecoveryResolution(*pending, input) { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) if err != nil { return nil, err } @@ -538,8 +532,7 @@ func (flow *Flow) ResolveToolRecovery( } // ExecutePlan schedules an exact waiting draft or active plan revision. -func (flow *Flow) ExecutePlan(ctx dex.Context, request executePlanRPCInput) (*dex.RPCResult[bool], error) { - input := request.Value +func (flow *Flow) ExecutePlan(ctx dex.Context, input PlanExecutionRequest) (*dex.RPCResult[bool], error) { state, err := agentStateAttribute.Get(ctx) if err != nil { return nil, err @@ -578,7 +571,7 @@ func (flow *Flow) ExecutePlan(ctx dex.Context, request executePlanRPCInput) (*de if !canExecute { return &dex.RPCResult[bool]{Output: false}, nil } - accepted, err := flow.resetInactivityTimerForUserOperation(ctx, request.FirstAttemptAt) + accepted, err := flow.resetInactivityTimerForUserOperation(ctx) if err != nil { return nil, err } @@ -680,10 +673,7 @@ func clearInactivityDeadline(ctx dex.Context) error { return err } -func (flow *Flow) resetInactivityTimerForUserOperation( - ctx dex.Context, - firstAttemptAt time.Time, -) (bool, error) { +func (flow *Flow) resetInactivityTimerForUserOperation(ctx dex.Context) (bool, error) { config, err := agentConfigAttribute.Get(ctx) if err != nil { return false, err @@ -691,10 +681,11 @@ func (flow *Flow) resetInactivityTimerForUserOperation( if config.InactivityTimeoutSeconds == 0 { return true, nil } - resetTarget := firstAttemptAt.UTC().Add( + activityAt := time.Now().UTC() + resetTarget := activityAt.Add( time.Duration(config.InactivityTimeoutSeconds) * time.Second, ) - return flow.resetInactivityTimer(ctx, firstAttemptAt, resetTarget) + return flow.resetInactivityTimer(ctx, activityAt, resetTarget) } func (flow *Flow) resetInactivityTimer( diff --git a/internal/agent/flow_integration_test.go b/internal/agent/flow_integration_test.go index 47f3094..6127123 100644 --- a/internal/agent/flow_integration_test.go +++ b/internal/agent/flow_integration_test.go @@ -76,9 +76,7 @@ func TestAgentInactivityExpirationIntegration(t *testing.T) { if initial.Description.InactivityDeadline == nil { t.Fatalf("initial Snapshot = %#v", initial.Description) } - // Cross the durable clock's one-second precision before testing the - // 100ms reset-coalescing threshold. - delay := time.NewTimer(1500 * time.Millisecond) + delay := time.NewTimer(200 * time.Millisecond) defer delay.Stop() select { case <-delay.C: diff --git a/internal/agent/types.go b/internal/agent/types.go index 94a984f..991ebee 100644 --- a/internal/agent/types.go +++ b/internal/agent/types.go @@ -753,41 +753,6 @@ type PendingUserMessage struct { Value UserMessage `json:"value"` } -type sendMessageRPCInput struct { - FirstAttemptAt time.Time `json:"first_attempt_at"` - Value PendingUserMessage `json:"value"` -} - -type answerQuestionsRPCInput struct { - FirstAttemptAt time.Time `json:"first_attempt_at"` - Value AnswerQuestionsRequest `json:"value"` -} - -type steerMessageRPCInput struct { - FirstAttemptAt time.Time `json:"first_attempt_at"` - Value SteerMessageRequest `json:"value"` -} - -type deleteQueuedMessageRPCInput struct { - FirstAttemptAt time.Time `json:"first_attempt_at"` - Value MessageID `json:"value"` -} - -type approveToolRPCInput struct { - FirstAttemptAt time.Time `json:"first_attempt_at"` - Value ToolApprovalRequest `json:"value"` -} - -type resolveToolRecoveryRPCInput struct { - FirstAttemptAt time.Time `json:"first_attempt_at"` - Value ResolveToolRecoveryRequest `json:"value"` -} - -type executePlanRPCInput struct { - FirstAttemptAt time.Time `json:"first_attempt_at"` - Value PlanExecutionRequest `json:"value"` -} - // AgentDescription is the durable application state needed to render a conversation. type AgentDescription struct { Status AgentStatus `json:"status"`