diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index daf54a7..a3e37cd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -77,14 +77,16 @@ 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. The Agent +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- +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..551eb03 100644 --- a/docs/adr/0016-agent-owned-inactivity-expiration.md +++ b/docs/adr/0016-agent-owned-inactivity-expiration.md @@ -6,32 +6,33 @@ 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. - -Expiration commits the `expiring` status and one `inactivity_expired` activity -event before invoking the constructor-injected `InactivityExpirationHandler`. +`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 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. + +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..947eb50 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,14 @@ 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. 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 @@ -233,7 +237,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..62e04a6 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,58 @@ 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( + activityAt := time.Now().UTC() + resetTarget := activityAt.Add( time.Duration(config.InactivityTimeoutSeconds) * time.Second, ) - if err := inactivityDeadlineAttribute.Set(ctx, deadline); err != nil { - return nil, err + return flow.resetInactivityTimer(ctx, activityAt, resetTarget) +} + +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 } - remaining := time.Until(deadline) - if remaining <= 0 { - remaining = time.Nanosecond + if !deadline.After(activityAt.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 +774,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 +1945,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 +2083,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 +2136,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 +2181,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 +2192,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 +2208,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 +2225,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 +2245,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 +2255,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 +2268,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 +2275,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 +2897,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 +2914,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 +2928,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 +3466,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 +3492,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 +3522,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 +3698,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, ctx.FirstAttemptAt(), 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 +3803,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 +3876,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..6127123 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 = 5 + 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) }},