feat(agent): durable HITL approvals; fix duplicate gated calls and Gemini call IDs

CHANGELOG.md

@@ -2,6 +2,18 @@
 
 All notable changes to GopherAgent are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/); versions follow [Semantic Versioning](https://semver.org/) — pre-1.0, breaking API changes only require a minor bump.
 
+## [v0.44.0] — 2026-10-25
+
+### Added
+
+- **`AgentLoop.DeferApprovals`, `AgentLoop.Resume`, and `AgentLoop.PendingApprovals` — human approvals now survive a restart.** `ConfirmHITL` blocks the run's goroutine until it answers, which is fine when the reviewer clicks a button seconds later and fails as soon as the answer can take longer than the process lives: a deploy or crash in between lost every open approval, and nothing could resume the turn once the answer arrived. With `DeferApprovals` set, a call that needs approval does not block. The loop emits `ActionRequiredEvent` with the call's ID, finishes the other calls in that wave, saves the turn with the gated call left without a result, and ends the run with `ErrApprovalPending`. Calls in later waves wait, because they may depend on the gated call through `<output_of:...>`. Session history is the durable record, so every `SessionManager` backend supports it with no schema change. `Resume(ctx, sessionKey, decisions)` reads that turn back, from any process sharing the store: approved calls run, denied calls get the usual denial directive, calls without a decision stay pending (deferring again), and once every call has a result the run continues normally. The tool block is rebuilt in call order, so answers recorded before and after the suspension stay aligned with their calls. `PendingApprovals` lists what is still open, for asking again after a restart or for deciding whether an incoming reply belongs to `Resume` or `Run`. A decision for an ID that is not awaiting approval returns `ErrUnknownApproval` rather than being ignored, so passing `false` for the wrong ID can never let a call run that the caller believed was denied. `ConfirmFunc` is unchanged and blocking stays the default. `Resume` does not lock the session: route one session's answers through one goroutine, or an approved call can run twice. Sub-agent workers and the plan-mode gate still block. (`pkg/agent/resume.go`, `pkg/agent/hitl_gate.go`, `pkg/agent/loop_iteration.go`)
+- **`ActionRequiredEvent.ID`** carries the tool call's ID, so hosts no longer mint their own correlation ID and smuggle it through `Args`. Additive on the wire (`id`, omitted when empty). (`pkg/agent/event_types.go`)
+
+### Fixed
+
+- **Identical gated tool calls in one wave are confirmed once and executed once.** A model that emitted the same `RequiresConfirmation` call twice in one reply got two separate prompts, so the user was asked the same question twice, and an approved side effect ran twice. Calls with the same name and canonical arguments are now gated and executed once, and each duplicate's tool call ID receives a copy of the result, which keeps the transcript paired. Ungated tools and calls whose arguments differ are unaffected. (`pkg/agent/loop_execute.go`)
+- **Gemini: parallel calls of one tool no longer overwrite each other's results.** Most Gemini models return no call IDs, and the adapter used the tool name as the ID, so two calls of the same tool in one reply shared an ID. The loop keys results by ID, so the second result replaced the first and the model received one of them twice. The failure mode was worst for gated calls: a denial recorded for a duplicate replaced the approved result, and the model told the user an action had failed when it had run. The adapter now uses the API's ID when one is returned, keeps the bare name for a tool's first call (so stored sessions remain valid), and suffixes later ones (`name#2`). `FunctionResponse.Name` is resolved from the call that issued it. What goes over the wire is unchanged. (`pkg/llm/gemini/gemini.go`)
+
 ## [v0.43.0] — 2026-08-17
 
 ### Added
@@ -664,6 +676,7 @@ Multi-user, long-running, audit-friendly chat surface — the foundation for sid
 - README section on the permission flow — documents `RequiresConfirmation` × `ConfirmHITL` × `Permissions` interaction.
 - Enum struct tag support in `tools.SchemaFor[T]()` — emit values into JSON-Schema's `enum` array so providers reject invalid values upstream.
 
+[v0.44.0]: https://github.com/hung12ct/gopheragent/releases/tag/v0.44.0
 [v0.43.0]: https://github.com/hung12ct/gopheragent/releases/tag/v0.43.0
 [v0.42.0]: https://github.com/hung12ct/gopheragent/releases/tag/v0.42.0
 [v0.41.0]: https://github.com/hung12ct/gopheragent/releases/tag/v0.41.0

README.md

@@ -51,8 +51,9 @@ go get github.com/hung12ct/gopheragent
   load only when a skill is used.
 - **Deterministic ReAct loop** — dependency-aware parallel tool scheduling with
   `<output_of:ID>` refs, anti-loop detection, token-budget-aware pruning.
-- **Streaming & HITL** — SSE streaming, human approvals, plan mode, and
-  self-critique that keeps the best-scoring pass instead of the last one.
+- **Streaming & HITL** — SSE streaming, human approvals (durable across
+  restarts with `DeferApprovals` + `Resume`), plan mode, and self-critique
+  that keeps the best-scoring pass instead of the last one.
 - **Honest terminals** — a `context_trace` event says exactly which messages
   pruning rewrote and why; a `degraded` terminal reports "the artifact landed,
   the bookkeeping did not" instead of forcing a turn into success or failure.

docs/permissions.md

@@ -33,3 +33,37 @@ sqlTool := builtin.NewSQLAgentTool(db, "", sm, provider).
 
 See [`examples/hitl_server`](../examples/hitl_server) for approvals delivered
 over HTTP with an async bridge.
+
+## Durable approvals (answers that arrive hours later)
+
+`ConfirmHITL` blocks the run until it returns, so a restart loses every
+open approval. Where users answer much later (chat apps, email), set
+`DeferApprovals` instead:
+
+```go
+loop.DeferApprovals = true
+
+for ev := range loop.RunText(ctx, sessionKey, text) {
+    if p, ok := ev.Payload.(agent.ActionRequiredEvent); ok {
+        askUser(sessionKey, p.ID, p.Tool, p.Args) // e.g. a chat message
+    }
+    // the run ends with errors.Is(err, agent.ErrApprovalPending)
+}
+
+// Later, from any process that shares the SessionManager:
+seq, err := loop.Resume(ctx, sessionKey, map[string]bool{approvalID: true})
+```
+
+The gated call is saved in session history without a result, so every
+`SessionManager` backend keeps it across restarts. Other calls in the same
+wave still run; calls in later waves wait, since they may depend on it.
+`Resume` runs approved calls, records denials, and continues the turn. Calls
+left without a decision stay pending. `PendingApprovals(ctx, sessionKey)`
+lists what is open, for asking again after a restart or for routing a reply
+to `Resume` instead of `Run`.
+
+- A new `Run` on the session seals pending calls as cancelled, which is
+  how a user moving on reads to the model.
+- `Resume` does not lock the session. Route each session's answers
+  through one goroutine, or an approved call can run twice.
+- Sub-agent workers and the plan-mode gate still block on their callbacks.

pkg/agent/errors.go

@@ -135,6 +135,20 @@ var (
 	// the previous run ended naturally, and the caller should send a new
 	// user message instead of trying to extend it.
 	ErrNothingToContinue = errors.New("agent: session ended cleanly; nothing to continue")
+
+	// ErrApprovalPending ends a run whose tool calls are waiting on a
+	// deferred approval (AgentLoop.DeferApprovals). It is not a failure: the
+	// turn is saved, and AgentLoop.Resume picks it up once answered.
+	ErrApprovalPending = errors.New("agent: tool calls awaiting approval")
+
+	// ErrNothingToResume is returned by AgentLoop.Resume when the session's
+	// latest tool turn has no call left without a result, typically because
+	// a later Run already sealed it.
+	ErrNothingToResume = errors.New("agent: no pending tool calls to resume")
+
+	// ErrUnknownApproval is returned by AgentLoop.Resume when a decision
+	// names a call ID that is not pending in the session.
+	ErrUnknownApproval = errors.New("agent: no pending tool call with this ID")
 )
 
 // ToolNotFoundError carries the name of the missing tool.

pkg/agent/event_types.go

@@ -200,7 +200,11 @@ func (ToolProgressEvent) eventType() StreamEventType { return EventTypeToolProgr
 // Adopters typically gate this through an out-of-band channel (UI prompt,
 // Slack message, webhook) and resume the loop only after the operator
 // responds via the configured ConfirmHITL callback.
+//
+// ID is the tool call's ID. With AgentLoop.DeferApprovals it is the key to
+// answer the approval with in AgentLoop.Resume.
 type ActionRequiredEvent struct {
+	ID   string `json:"id,omitempty"`
 	Tool string `json:"tool"`
 	Args string `json:"args,omitempty"`
 }

pkg/agent/hitl_gate.go

@@ -18,8 +18,30 @@ const (
 	hitlDenied
 	hitlTimedOut
 	hitlUnconfigured
+	// hitlDeferred leaves the call pending for Resume; it gets no tool
+	// message this run.
+	hitlDeferred
 )
 
+// resolveApproval decides a gated call. A decision handed to Resume wins;
+// otherwise DeferApprovals parks the call for a later Resume, and the
+// default asks ConfirmHITL and blocks until it answers.
+func (al *AgentLoop) resolveApproval(ctx context.Context, st *iterationState, ws *waveState, tc PendingToolCall) hitlOutcome {
+	if approved, ok := st.decisions[tc.ID]; ok {
+		if approved {
+			return hitlApproved
+		}
+		al.emit(ctx, st.sessionKey, st.streamChan, Event(HITLDeniedEvent{Tool: tc.Name, Args: tc.ArgsJSON}))
+		return hitlDenied
+	}
+	if al.DeferApprovals {
+		ws.markPending(tc)
+		al.emit(ctx, st.sessionKey, st.streamChan, Event(ActionRequiredEvent{ID: tc.ID, Tool: tc.Name, Args: tc.ArgsJSON}))
+		return hitlDeferred
+	}
+	return al.runHITLGate(ctx, st.sessionKey, st.streamChan, ws, tc)
+}
+
 // runHITLGate emits the human-confirmation prompt under hitlMu and returns
 // a hitlOutcome describing how the gate resolved. The caller selects the
 // tool-message directive from the outcome:
@@ -44,7 +66,7 @@ func (al *AgentLoop) runHITLGate(ctx context.Context, sessionKey string, streamC
 
 	if al.ConfirmHITL == nil {
 		al.warnMissingConfirmHITLOnce()
-		al.emit(ctx, sessionKey, streamChan, Event(ActionRequiredEvent{Tool: tc.Name, Args: tc.ArgsJSON}))
+		al.emit(ctx, sessionKey, streamChan, Event(ActionRequiredEvent{ID: tc.ID, Tool: tc.Name, Args: tc.ArgsJSON}))
 		return hitlUnconfigured
 	}
 

pkg/agent/loop_execute.go

@@ -44,7 +44,12 @@ func (al *AgentLoop) toolCallSemaphore(waveSize int) chan struct{} {
 	return make(chan struct{}, n)
 }
 
-func (al *AgentLoop) executeToolWaves(ctx context.Context, st *iterationState, scheduled []PendingToolCall) *waveState {
+// ws may arrive seeded with results from an earlier run (see Resume); calls
+// that already have a result are skipped, and scheduling still sees the
+// whole turn so references to those results keep their order. A wave that
+// leaves any call awaiting a deferred approval ends execution: later waves
+// may depend on it, and Resume runs them once the approval is answered.
+func (al *AgentLoop) executeToolWaves(ctx context.Context, st *iterationState, ws *waveState, scheduled []PendingToolCall) {
 	waves, schedErr := scheduleToolCalls(scheduled)
 	if schedErr != nil {
 		al.emit(ctx, st.sessionKey, st.streamChan, Event(ThoughtEvent{
@@ -53,17 +58,18 @@ func (al *AgentLoop) executeToolWaves(ctx context.Context, st *iterationState, s
 		waves = [][]PendingToolCall{scheduled}
 	}
 
-	ws := newWaveState(len(scheduled))
 	for _, wave := range waves {
+		wave = ws.unfinished(wave)
 		substitutedWave := substituteWaveArgs(ws, wave, func(toolName string, subErr error) {
 			al.emit(ctx, st.sessionKey, st.streamChan, Event(ThoughtEvent{
 				Message: fmt.Sprintf("Tool scheduler: substitution for %q failed: %v — skipping execution.", toolName, subErr),
 			}))
 		})
 
-		sem := al.toolCallSemaphore(len(substitutedWave))
+		dispatch, dups := al.splitGatedDuplicates(substitutedWave)
+		sem := al.toolCallSemaphore(len(dispatch))
 		var wg sync.WaitGroup
-		for _, tc := range substitutedWave {
+		for _, tc := range dispatch {
 			wg.Add(1)
 			go func(tCall PendingToolCall) {
 				defer wg.Done()
@@ -75,12 +81,79 @@ func (al *AgentLoop) executeToolWaves(ctx context.Context, st *iterationState, s
 			}(tc)
 		}
 		wg.Wait()
+		ws.copyDuplicateResults(dups)
 
-		if ws.fatalErr != nil {
+		if ws.fatalErr != nil || len(ws.pending) > 0 {
 			break
 		}
 	}
-	return ws
+}
+
+// gatedDuplicate is a confirmation-gated call that repeats an earlier call
+// in the same wave with identical arguments; it reuses the leader's result.
+type gatedDuplicate struct {
+	call     PendingToolCall
+	leaderID string
+}
+
+// splitGatedDuplicates removes calls to RequiresConfirmation tools that
+// repeat an earlier call in the wave with the same name and canonical args.
+// Models occasionally emit the same gated call twice in one turn; gating
+// each one asks the user the same question twice and, once approved, runs
+// the side effect twice. Ungated tools are left alone so read-only fan-out
+// keeps its existing behavior.
+func (al *AgentLoop) splitGatedDuplicates(wave []PendingToolCall) ([]PendingToolCall, []gatedDuplicate) {
+	if len(wave) < 2 {
+		return wave, nil
+	}
+	var seen map[string]string
+	var dups []gatedDuplicate
+	var out []PendingToolCall
+	for i, tc := range wave {
+		tool, ok := al.Tools.Get(tc.Name)
+		if !ok || !tool.Descriptor().RequiresConfirmation {
+			if dups != nil {
+				out = append(out, tc)
+			}
+			continue
+		}
+		key := toolCacheKey(tc.Name, tc.ArgsJSON)
+		if leaderID, dup := seen[key]; dup {
+			if dups == nil {
+				out = append(make([]PendingToolCall, 0, len(wave)), wave[:i]...)
+			}
+			dups = append(dups, gatedDuplicate{call: tc, leaderID: leaderID})
+			continue
+		}
+		if seen == nil {
+			seen = make(map[string]string, 2)
+		}
+		seen[key] = tc.ID
+		if dups != nil {
+			out = append(out, tc)
+		}
+	}
+	if dups == nil {
+		return wave, nil
+	}
+	return out, dups
+}
+
+// copyDuplicateResults records each duplicate's reply as a copy of its
+// leader's, so every ToolCallID still gets a matching tool message. A
+// leader with no recorded result (wave aborted) is skipped and left to
+// synthesizeMissingToolErrors. Runs after wg.Wait, so no goroutine writes
+// concurrently.
+func (ws *waveState) copyDuplicateResults(dups []gatedDuplicate) {
+	for _, d := range dups {
+		msg, ok := ws.toolMsgs[d.leaderID]
+		if !ok {
+			continue
+		}
+		msg.ToolCallID = d.call.ID
+		_, isResult := ws.resultsByID[d.leaderID]
+		ws.recordToolMsg(d.call.ID, msg, isResult)
+	}
 }
 
 // executeToolCall is the per-tool goroutine body. The caller spawns it
@@ -144,7 +217,10 @@ func (al *AgentLoop) executeToolCall(ctx context.Context, st *iterationState, ws
 	// RequiresConfirmation covers tools that always prompt. Allow bypasses
 	// both.
 	if (desc.RequiresConfirmation || permDecision == PermissionConfirm) && permDecision != PermissionAllow {
-		outcome := al.runHITLGate(ctx, st.sessionKey, st.streamChan, ws, tCall)
+		outcome := al.resolveApproval(ctx, st, ws, tCall)
+		if outcome == hitlDeferred {
+			return
+		}
 		if outcome != hitlApproved {
 			var msg string
 			switch outcome {

pkg/agent/loop_execute_dedupe_test.go

@@ -0,0 +1,118 @@
+package agent
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	"github.com/hung12ct/gopheragent/pkg/history"
+	"github.com/hung12ct/gopheragent/pkg/tools"
+)
+
+// sideEffectTool counts executions so tests can tell a reused result from a
+// second run of the side effect.
+type sideEffectTool struct {
+	name    string
+	confirm bool
+	runs    atomic.Int32
+}
+
+func (t *sideEffectTool) Descriptor() tools.ToolDescriptor {
+	return tools.ToolDescriptor{
+		Name:                 t.name,
+		Description:          "counts runs",
+		RequiresConfirmation: t.confirm,
+		Display:              tools.DefaultDisplay(t.name, "counts runs"),
+	}
+}
+
+func (t *sideEffectTool) Execute(_ context.Context, args string) (tools.Result, error) {
+	t.runs.Add(1)
+	return tools.Text("ran:" + args), nil
+}
+
+func toolReplies(t *testing.T, sm *history.InMemSessionManager) []history.Message {
+	t.Helper()
+	msgs, err := sm.History(context.Background(), "s1")
+	if err != nil {
+		t.Fatalf("history: %v", err)
+	}
+	var out []history.Message
+	for _, m := range msgs {
+		if m.Role == "tool" {
+			out = append(out, m)
+		}
+	}
+	return out
+}
+
+func TestToolWaves_GatedDuplicatesConfirmAndRunOnce(t *testing.T) {
+	cases := map[string][]PendingToolCall{
+		"distinct IDs": {
+			{ID: "c1", Name: "wipe", ArgsJSON: `{"a":1,"b":2}`},
+			{ID: "c2", Name: "wipe", ArgsJSON: `{"b":2, "a":1}`},
+		},
+		// Gemini reuses the tool name as the call ID.
+		"shared ID": {
+			{ID: "wipe", Name: "wipe", ArgsJSON: `{}`},
+			{ID: "wipe", Name: "wipe", ArgsJSON: `{}`},
+		},
+	}
+	for name, calls := range cases {
+		t.Run(name, func(t *testing.T) {
+			tool := &sideEffectTool{name: "wipe", confirm: true}
+			provider := &scriptProvider{turns: []LLMResult{{ToolCalls: calls}, {Content: "done"}}}
+			loop, sm := setup(provider, tool)
+			var asks atomic.Int32
+			loop.ConfirmHITL = func(context.Context, string, string) bool {
+				return asks.Add(1) == 1 // a second prompt would be denied
+			}
+
+			if _, err := loop.RunIteration(context.Background(), "s1", "wipe it"); err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if n := asks.Load(); n != 1 {
+				t.Fatalf("ConfirmHITL called %d times, want 1", n)
+			}
+			if n := tool.runs.Load(); n != 1 {
+				t.Fatalf("tool ran %d times, want 1", n)
+			}
+			replies := toolReplies(t, sm)
+			if len(replies) != len(calls) {
+				t.Fatalf("got %d tool replies, want %d", len(replies), len(calls))
+			}
+			for i, m := range replies {
+				if m.IsError || m.ToolCallID != calls[i].ID {
+					t.Fatalf("reply %d = %+v, want approved result for %q", i, m, calls[i].ID)
+				}
+			}
+		})
+	}
+}
+
+func TestToolWaves_DuplicatesNotMergedWhenUngatedOrArgsDiffer(t *testing.T) {
+	gated := &sideEffectTool{name: "wipe", confirm: true}
+	plain := &sideEffectTool{name: "read"}
+	provider := &scriptProvider{turns: []LLMResult{{ToolCalls: []PendingToolCall{
+		{ID: "c1", Name: "wipe", ArgsJSON: `{"k":1}`},
+		{ID: "c2", Name: "wipe", ArgsJSON: `{"k":2}`},
+		{ID: "c3", Name: "read", ArgsJSON: `{}`},
+		{ID: "c4", Name: "read", ArgsJSON: `{}`},
+	}}, {Content: "done"}}}
+	loop, _ := setup(provider, gated, plain)
+	var asks atomic.Int32
+	loop.ConfirmHITL = func(context.Context, string, string) bool {
+		asks.Add(1)
+		return true
+	}
+
+	if _, err := loop.RunIteration(context.Background(), "s1", "go"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if asks.Load() != 2 || gated.runs.Load() != 2 {
+		t.Fatalf("gated: asks=%d runs=%d, want 2 and 2", asks.Load(), gated.runs.Load())
+	}
+	if n := plain.runs.Load(); n != 2 {
+		t.Fatalf("ungated tool ran %d times, want 2", n)
+	}
+}