Cancel active agent work during shutdown
This commit is contained in:
parent
2599ccb4ac
commit
16bf2115fa
|
|
@ -33,6 +33,8 @@ type Agent struct {
|
|||
newToolServer func() *toolclient.ToolsrvConn
|
||||
newBackend func(string) (backend.Backend, error)
|
||||
currentAction atomic.Pointer[actionHandle]
|
||||
closed atomic.Bool
|
||||
actionMu sync.Mutex // serializes action registration with shutdown
|
||||
warnedContext bool
|
||||
resultCache sync.Map
|
||||
|
||||
|
|
@ -379,6 +381,8 @@ func (ag *Agent) IsRunning() bool {
|
|||
// Interrupt cancels the current in-progress agent turn.
|
||||
// Returns true if an action was running and was cancelled.
|
||||
func (ag *Agent) Interrupt(cause error) bool {
|
||||
ag.actionMu.Lock()
|
||||
defer ag.actionMu.Unlock()
|
||||
if h := ag.currentAction.Load(); h != nil {
|
||||
h.cancel(cause)
|
||||
return true
|
||||
|
|
@ -505,6 +509,7 @@ type EventHandler func(Event)
|
|||
// actionHandle holds the cancel function for the current agent turn.
|
||||
type actionHandle struct {
|
||||
cancel context.CancelCauseFunc
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// WatchField names supported by Agent.WaitChange.
|
||||
|
|
@ -532,9 +537,23 @@ func (ag *Agent) SetEnv(key, value string) {
|
|||
ag.runtime.ToolServer.SetEnv(key, value)
|
||||
}
|
||||
|
||||
// Close releases agent resources (dispatcher, execute server).
|
||||
// Close releases agent resources (dispatcher, execute server). It first prevents
|
||||
// new work, cancels the active turn, and waits for that turn to release its
|
||||
// references to the runtime before closing the tools connection.
|
||||
func (ag *Agent) Close() {
|
||||
if ag.runtime.ToolServer != nil {
|
||||
ag.closed.Store(true)
|
||||
|
||||
ag.actionMu.Lock()
|
||||
h := ag.currentAction.Load()
|
||||
if h != nil {
|
||||
h.cancel(context.Canceled)
|
||||
}
|
||||
ag.actionMu.Unlock()
|
||||
if h != nil {
|
||||
<-h.done
|
||||
}
|
||||
|
||||
if ag.runtime != nil && ag.runtime.ToolServer != nil {
|
||||
ag.runtime.ToolServer.Close()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -39,6 +39,9 @@ func (ag *Agent) Submit(ctx context.Context, input string) {
|
|||
}
|
||||
}()
|
||||
ag.log.Debug("Agent.Submit() input_len=%d running=%v", len(input), ag.IsRunning())
|
||||
if ag.closed.Load() {
|
||||
return
|
||||
}
|
||||
if input == "" {
|
||||
return
|
||||
}
|
||||
|
|
@ -56,6 +59,9 @@ func (ag *Agent) Submit(ctx context.Context, input string) {
|
|||
// with an executeTurn arriving via prompt.
|
||||
ag.submitMu.Lock()
|
||||
defer ag.submitMu.Unlock()
|
||||
if ag.closed.Load() {
|
||||
return
|
||||
}
|
||||
|
||||
if ag.HandleCommand(ctx, input) {
|
||||
return
|
||||
|
|
@ -115,11 +121,20 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
|
|||
}
|
||||
|
||||
actCtx, actCancel := context.WithCancelCause(ctx)
|
||||
handle := &actionHandle{cancel: actCancel}
|
||||
turnDone := make(chan struct{})
|
||||
handle := &actionHandle{cancel: actCancel, done: turnDone}
|
||||
ag.actionMu.Lock()
|
||||
if ag.closed.Load() {
|
||||
ag.actionMu.Unlock()
|
||||
actCancel(context.Canceled)
|
||||
return ""
|
||||
}
|
||||
ag.currentAction.Store(handle)
|
||||
ag.actionMu.Unlock()
|
||||
defer func() {
|
||||
if ag.currentAction.CompareAndSwap(handle, nil) {
|
||||
actCancel(context.Canceled)
|
||||
close(handle.done)
|
||||
}
|
||||
}()
|
||||
ag.SetState("thinking")
|
||||
|
|
@ -197,7 +212,7 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
|
|||
if compactLimit := ag.autoCompactLimit(actCtx); compactLimit > 0 && tokens >= compactLimit {
|
||||
ag.emit(Event{Role: "info", Content: "auto-compacting context...\n"})
|
||||
ag.SetState("compacting")
|
||||
if _, err := ag.runCompact(context.WithoutCancel(actCtx), "auto"); err != nil {
|
||||
if _, err := ag.runCompact(actCtx, "auto"); err != nil {
|
||||
ag.log.Error("auto-compact failed: %v", err)
|
||||
}
|
||||
ag.SetState("thinking")
|
||||
|
|
@ -246,8 +261,10 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
|
|||
)
|
||||
for {
|
||||
err = ag.run(actCtx)
|
||||
actCancel(nil)
|
||||
ag.currentAction.CompareAndSwap(handle, nil)
|
||||
if ag.currentAction.CompareAndSwap(handle, nil) {
|
||||
actCancel(nil)
|
||||
close(handle.done)
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
break
|
||||
|
|
@ -273,8 +290,15 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
|
|||
ag.history.resetTurnAccumulators()
|
||||
replyBuf.Reset()
|
||||
actCtx, actCancel = context.WithCancelCause(ctx)
|
||||
handle = &actionHandle{cancel: actCancel}
|
||||
handle = &actionHandle{cancel: actCancel, done: make(chan struct{})}
|
||||
ag.actionMu.Lock()
|
||||
if ag.closed.Load() {
|
||||
ag.actionMu.Unlock()
|
||||
actCancel(context.Canceled)
|
||||
break
|
||||
}
|
||||
ag.currentAction.Store(handle)
|
||||
ag.actionMu.Unlock()
|
||||
// Re-inject preamble — compaction strips it from history.
|
||||
ag.history.injectPreamble(ag.runtime.PreambleString())
|
||||
continue
|
||||
|
|
|
|||
Loading…
Reference in New Issue