fix: restore harness interrupt resume and events (#22050)

internal/harness/core/callback.go

@@ -199,19 +199,23 @@ func DeleteRunLocalValue(ctx context.Context, key string) error {
 
 func SendEvent(ctx context.Context, event *AgentEvent) error {
 	ec := getChatModelExecCtx(ctx)
-	if ec == nil || ec.generator == nil {
+	if !ec.canSend() {
 		return errNotInAgentExec
 	}
-	ec.send(event)
+	if !ec.send(event) {
+		return errAgentEventTypeMismatch
+	}
 	return nil
 }
 
 func TypedSendEvent[M MessageType](ctx context.Context, event *TypedAgentEvent[M]) error {
 	ec := getReActExecCtx[M](ctx)
-	if ec == nil || ec.generator == nil {
+	if !ec.canSend() {
 		return errNotInAgentExec
 	}
-	ec.send(event)
+	if !ec.send(event) {
+		return errAgentEventTypeMismatch
+	}
 	return nil
 }
 
@@ -220,6 +224,7 @@ type AgentExecError struct{ Message string }
 func (e *AgentExecError) Error() string { return e.Message }
 
 var errNotInAgentExec = &AgentExecError{Message: "must be called within ReActAgent Run/Resume"}
+var errAgentEventTypeMismatch = &AgentExecError{Message: "event type does not match the active ReAct agent"}
 
 // checkGobEncodability probes whether the value can be gob-encoded as part of
 // a map[string]any, which is exactly how session values are serialized during

internal/harness/core/callback_test.go

@@ -194,6 +194,54 @@ func TestSendEvent_NotInAgentExec(t *testing.T) {
 	}
 }
 
+type sendEventBeforeAgentMiddleware struct {
+	BaseMiddleware[*schema.Message]
+	err      error
+	typedErr error
+}
+
+func (m *sendEventBeforeAgentMiddleware) BeforeAgent(ctx context.Context, rc *ReActAgentContext) (context.Context, *ReActAgentContext, error) {
+	m.err = SendEvent(ctx, EventFromMessage(schema.AssistantMessage("middleware event"), nil, schema.RoleAssistant, ""))
+	m.typedErr = TypedSendEvent[*schema.Message](ctx, EventFromMessage(schema.AssistantMessage("typed middleware event"), nil, schema.RoleAssistant, ""))
+	return ctx, rc, nil
+}
+
+func TestSendEventInsideReActExecution(t *testing.T) {
+	model := &mockModel{}
+	model.addResp("model response")
+	mw := &sendEventBeforeAgentMiddleware{}
+	agent := NewReActAgent(&ReActConfig[*schema.Message]{
+		Model:       model,
+		Middlewares: []TypedReActMiddleware[*schema.Message]{mw, NewEventSenderModelWrapper[*schema.Message]()},
+	})
+	runner := NewTypedRunner(RunnerConfig[*schema.Message]{Agent: agent})
+	iter := runner.Run(t.Context(), []*schema.Message{schema.UserMessage("run")})
+	events := drainAgentEvents(t, iter)
+	if mw.err != nil {
+		t.Fatalf("SendEvent inside ReAct middleware: %v", mw.err)
+	}
+	if mw.typedErr != nil {
+		t.Fatalf("TypedSendEvent inside ReAct middleware: %v", mw.typedErr)
+	}
+	wanted := map[string]bool{
+		"middleware event":       false,
+		"typed middleware event": false,
+		"model response":         false,
+	}
+	for _, event := range events {
+		if event.Output != nil && event.Output.MessageOutput != nil && event.Output.MessageOutput.Message != nil {
+			if _, ok := wanted[event.Output.MessageOutput.Message.Content]; ok {
+				wanted[event.Output.MessageOutput.Message.Content] = true
+			}
+		}
+	}
+	for content, found := range wanted {
+		if !found {
+			t.Errorf("event %q was not sent to the run", content)
+		}
+	}
+}
+
 func TestCheckGobEncodability_StringValue(t *testing.T) {
 	err := checkGobEncodability("key", "string value")
 	if err != nil {

internal/harness/core/event_sender.go

@@ -20,7 +20,7 @@ type eventSenderModelHandler[M MessageType] struct{}
 
 func (h *eventSenderModelHandler[M]) WrapModel(ctx context.Context, m Model[M], mc *TypedModelContext[M]) (Model[M], error) {
 	ec := getReActExecCtx[M](ctx)
-	if ec == nil {
+	if !ec.canSend() {
 		return m, nil
 	}
 	return wrapModelWithEventSender(m, ec), nil

internal/harness/core/interface.go

@@ -11,6 +11,8 @@ import (
 
 func init() {
 	gob.Register(&RunStep{})
+	gob.Register(&TypedAgentInput[*schema.Message]{})
+	gob.Register(&TypedAgentInput[*schema.AgenticMessage]{})
 }
 
 // MessageType is the sealed type constraint for agent message types.

internal/harness/core/interrupt.go

@@ -173,6 +173,8 @@ type checkpointPayload struct {
 	RunCtx              *runContext
 	Info                *InterruptInfo
 	EnableStreaming     bool
+	WasInterrupted      bool
+	InterruptState      any
 	InterruptID2Address map[string]Address
 	InterruptID2State   map[string]InterruptState
 	TenantID            string
@@ -299,6 +301,8 @@ func loadCheckpoint(store CheckPointStore, ctx context.Context, cid string) (con
 	return ctx, p.RunCtx, &ResumeInfo{
 		EnableStreaming: p.EnableStreaming,
 		InterruptInfo:   p.Info,
+		WasInterrupted:  p.WasInterrupted,
+		InterruptState:  p.InterruptState,
 	}, nil
 }
 
@@ -309,12 +313,16 @@ func saveCheckpoint(store CheckPointStore, ctx context.Context, key string, enab
 	rc := getRunCtx(ctx)
 	id2addr, id2state := signalToMaps(is)
 	tenantID := extractCheckpointTenant(ctx)
+	var interruptState any
+	if is != nil {
+		interruptState = is.State
+	}
 
 	// Encode payload with tenant ID
 	p := checkpointPayload{
 		RunCtx: rc, Info: info, EnableStreaming: enableStreaming,
 		InterruptID2Address: id2addr, InterruptID2State: id2state,
-		TenantID: tenantID,
+		TenantID: tenantID, WasInterrupted: is != nil, InterruptState: interruptState,
 	}
 	var buf bytes.Buffer
 	if err := gob.NewEncoder(&buf).Encode(p); err != nil {
@@ -420,6 +428,7 @@ func buildResumeInfo(ctx context.Context, nextID string, info *ResumeInfo) (cont
 		EnableStreaming: info.EnableStreaming,
 		InterruptInfo:   info.InterruptInfo,
 		WasInterrupted:  info.WasInterrupted,
+		InterruptState:  info.InterruptState,
 		IsResumeTarget:  info.IsResumeTarget,
 		ResumeData:      info.ResumeData,
 	}

internal/harness/core/model_chain.go

@@ -26,7 +26,7 @@ func (w *eventSenderModelWrapper[M]) Generate(ctx context.Context, msgs []M, opt
 	if err != nil {
 		return resp, err
 	}
-	if w.execCtx != nil && w.execCtx.generator != nil && !isNilMessage(resp) {
+	if w.execCtx != nil && w.execCtx.canSend() && !isNilMessage(resp) {
 		w.execCtx.send(typedModelOutputEvent(resp, nil))
 	}
 	return resp, nil
@@ -201,7 +201,10 @@ func (w *callbackModelWrapper[M]) BindTools(tools []*schema.ToolInfo) error {
 //
 // The chain is built from innermost (closest to model) to outermost.
 // allToolInfos contains the merged tool infos from config.Tools + ToolContributor middlewares.
-func BuildModelWrapperChain[M MessageType](base Model[M], ec *reActExecCtx, cfg *ReActConfig[M], allToolInfos []*schema.ToolInfo) Model[M] {
+func BuildModelWrapperChain[M MessageType](ctx context.Context, base Model[M], ec *reActExecCtx, cfg *ReActConfig[M], allToolInfos []*schema.ToolInfo) Model[M] {
+	if ec != nil {
+		ctx = withReActExecCtx(ctx, ec)
+	}
 	model := base
 
 	// 1. Event sender (skip if user middlewares provide their own to avoid duplicates)
@@ -230,7 +233,7 @@ func BuildModelWrapperChain[M MessageType](base Model[M], ec *reActExecCtx, cfg
 			ModelRetryConfig:    cfg.RetryConfig,
 			ModelFailoverConfig: cfg.FailoverConfig,
 		}
-		wrapped, err := mw.WrapModel(context.Background(), model, mc)
+		wrapped, err := mw.WrapModel(ctx, model, mc)
 		if err == nil && wrapped != nil {
 			model = wrapped
 		}

internal/harness/core/model_chain_retry_failover_test.go

@@ -285,7 +285,7 @@ func TestBuildModelWrapperChain_RetryThenFailover_Integration(t *testing.T) {
 		FailoverConfig: &FailoverConfig[Message]{Models: []Model[Message]{m2}},
 	}
 
-	wrapped := BuildModelWrapperChain(m1, nil, cfg, nil)
+	wrapped := BuildModelWrapperChain(context.Background(), m1, nil, cfg, nil)
 
 	ctx := t.Context()
 	resp, err := wrapped.Generate(ctx, []Message{schema.UserMessage("test")})

internal/harness/core/model_chain_test.go

@@ -13,7 +13,7 @@ import (
 func TestBuildModelWrapperChain_NoConfig(t *testing.T) {
 	base := &mockModel{}
 	base.addResp("raw")
-	model := BuildModelWrapperChain(base, nil, DefaultReActConfig[*schema.Message](), nil)
+	model := BuildModelWrapperChain(context.Background(), base, nil, DefaultReActConfig[*schema.Message](), nil)
 	if model == nil {
 		t.Fatal("nil model")
 	}
@@ -31,7 +31,7 @@ func TestBuildModelWrapperChain_WithRetry(t *testing.T) {
 	base.addResp("retry-ok")
 	cfg := DefaultReActConfig[*schema.Message]()
 	cfg.RetryConfig = &ModelRetryConfig{MaxRetries: 2}
-	model := BuildModelWrapperChain(base, nil, cfg, nil)
+	model := BuildModelWrapperChain(context.Background(), base, nil, cfg, nil)
 	if model == nil {
 		t.Fatal("nil model")
 	}
@@ -52,7 +52,7 @@ func TestBuildModelWrapperChain_WithFailover(t *testing.T) {
 
 	cfg := DefaultReActConfig[*schema.Message]()
 	cfg.FailoverConfig = &FailoverConfigMsg{Models: []Model[*schema.Message]{fallback}}
-	model := BuildModelWrapperChain(primary, nil, cfg, nil)
+	model := BuildModelWrapperChain(context.Background(), primary, nil, cfg, nil)
 	resp, err := model.Generate(t.Context(), []Message{schema.UserMessage("hi")})
 	if err != nil {
 		t.Fatalf("Generate: %v", err)
@@ -73,7 +73,7 @@ func TestBuildModelWrapperChain_WithMiddleware(t *testing.T) {
 	base.addResp("mw-ok")
 	cfg := DefaultReActConfig[*schema.Message]()
 	cfg.Middlewares = []ReActMiddleware{mw}
-	model := BuildModelWrapperChain(base, nil, cfg, nil)
+	model := BuildModelWrapperChain(context.Background(), base, nil, cfg, nil)
 	resp, err := model.Generate(t.Context(), []Message{schema.UserMessage("hi")})
 	if err != nil {
 		t.Fatalf("Generate: %v", err)
@@ -102,7 +102,7 @@ func TestBuildModelWrapperChain_WithFullChain(t *testing.T) {
 	cfg.FailoverConfig = &FailoverConfigMsg{Models: []Model[*schema.Message]{fallback}}
 	cfg.Middlewares = []ReActMiddleware{mw}
 
-	model := BuildModelWrapperChain(primary, nil, cfg, nil)
+	model := BuildModelWrapperChain(context.Background(), primary, nil, cfg, nil)
 	resp, err := model.Generate(t.Context(), []Message{schema.UserMessage("hi")})
 	if err != nil {
 		t.Fatalf("Generate: %v", err)
@@ -118,7 +118,7 @@ func TestBuildModelWrapperChain_NilMiddleware(t *testing.T) {
 	base.addResp("nil-mw")
 	cfg := DefaultReActConfig[*schema.Message]()
 	cfg.Middlewares = []ReActMiddleware{nil, nil}
-	model := BuildModelWrapperChain(base, nil, cfg, nil)
+	model := BuildModelWrapperChain(context.Background(), base, nil, cfg, nil)
 	_, err := model.Generate(t.Context(), []Message{schema.UserMessage("hi")})
 	if err != nil {
 		t.Fatalf("Generate: %v", err)

internal/harness/core/react_agent.go

@@ -190,6 +190,7 @@ type typedRunParams[M MessageType] struct {
 // retry signal propagation, and after-tool-calls hooks.
 type reActExecCtx struct {
 	generator          *AsyncGenerator[*TypedAgentEvent[*schema.Message]]
+	sendTypedEvent     func(any) bool
 	cancelCtx          *cancelContext
 	suppressEventSend  bool
 	retrySignal        *retrySignal
@@ -197,12 +198,54 @@ type reActExecCtx struct {
 	afterToolCallsHook func(ctx context.Context) error
 }
 
-func (ec *reActExecCtx) send(ev any) {
-	if ec != nil && ec.generator != nil {
-		if te, ok := ev.(*TypedAgentEvent[*schema.Message]); ok {
-			ec.generator.Send(te)
-		}
+func newReActExecCtx[M MessageType](gen *AsyncGenerator[*TypedAgentEvent[M]]) *reActExecCtx {
+	ec := &reActExecCtx{}
+	if gen == nil {
+		return ec
 	}
+	ec.sendTypedEvent = func(ev any) bool {
+		event, ok := ev.(*TypedAgentEvent[M])
+		if !ok {
+			return false
+		}
+		gen.Send(event)
+		return true
+	}
+	if messageGen, ok := any(gen).(*AsyncGenerator[*TypedAgentEvent[*schema.Message]]); ok {
+		ec.generator = messageGen
+	}
+	return ec
+}
+
+func (ec *reActExecCtx) canSend() bool {
+	return ec != nil && (ec.generator != nil || ec.sendTypedEvent != nil)
+}
+
+func (ec *reActExecCtx) send(ev any) bool {
+	if ec == nil {
+		return false
+	}
+	if ec.sendTypedEvent != nil {
+		return ec.sendTypedEvent(ev)
+	}
+	if ec.generator == nil {
+		return false
+	}
+	event, ok := ev.(*TypedAgentEvent[*schema.Message])
+	if !ok {
+		return false
+	}
+	ec.generator.Send(event)
+	return true
+}
+
+type reActExecCtxKey struct{}
+
+func withReActExecCtx(ctx context.Context, ec *reActExecCtx) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return context.WithValue(ctx, reActExecCtxKey{}, ec)
 }
 
 type execContext struct {
@@ -299,6 +342,11 @@ func (a *ReActAgent[M]) prepareExecContext(ctx context.Context) (*execContext, e
 
 func (a *ReActAgent[M]) buildNoToolsRunFunc() typedRunFunc[M] {
 	return func(ctx context.Context, p *typedRunParams[M]) {
+		ec := newReActExecCtx(p.generator)
+		ec.cancelCtx = getCancelContext(ctx)
+		ec.afterToolCallsHook = p.afterToolCallsHook
+		ctx = withReActExecCtx(ctx, ec)
+
 		// Build allTools: config.Tools + contribTools
 		allTools := make([]Tool, 0, len(a.config.Tools)+len(a.exeCtx.contribTools))
 		allTools = append(allTools, a.config.Tools...)
@@ -310,7 +358,7 @@ func (a *ReActAgent[M]) buildNoToolsRunFunc() typedRunFunc[M] {
 			return
 		}
 
-		model := BuildModelWrapperChain(a.config.Model, nil, a.config, a.exeCtx.toolInfos)
+		model := BuildModelWrapperChain(ctx, a.config.Model, ec, a.config, a.exeCtx.toolInfos)
 		state := NewReActAgentState(p.input.Messages, a.exeCtx.toolInfos, a.config.MaxIterations)
 
 		// BeforeModelRewrite middleware
@@ -334,7 +382,6 @@ func (a *ReActAgent[M]) buildNoToolsRunFunc() typedRunFunc[M] {
 			p.generator.Send(&TypedAgentEvent[M]{Err: err})
 			return
 		}
-		p.generator.Send(typedModelOutputEvent(resp, nil))
 		state.Messages = append(state.Messages, resp)
 
 		// AfterModelRewrite middleware
@@ -568,17 +615,11 @@ func streamWithCancel[M MessageType](s *schema.StreamReader[M], cc *cancelContex
 
 // getChatModelExecCtx retrieves the chat model execution context from context.
 func getChatModelExecCtx(ctx context.Context) *reActExecCtx {
-	rc := getRunCtx(ctx)
-	if rc == nil {
+	if ctx == nil {
 		return nil
 	}
-	// The exec ctx is stored on the run session or passed via context value
-	if value, ok := rc.Session.getValue("__exec_ctx"); ok {
-		if ec, ok := value.(*reActExecCtx); ok {
-			return ec
-		}
-	}
-	return nil
+	ec, _ := ctx.Value(reActExecCtxKey{}).(*reActExecCtx)
+	return ec
 }
 
 // getReActExecCtx retrieves the typed execution context from context.
@@ -635,6 +676,11 @@ func (a *ReActAgent[M]) buildGraphReActRunFunc() typedRunFunc[M] {
 			return
 		}
 
+		ec := newReActExecCtx(p.generator)
+		ec.cancelCtx = getCancelContext(ctx)
+		ec.afterToolCallsHook = p.afterToolCallsHook
+		ctx = withReActExecCtx(ctx, ec)
+
 		// Build graph config from agent config.
 		graphCfg := &ReActGraphConfig{
 			Checkpointer:    a.config.GraphReActCheckpointer,
@@ -659,20 +705,12 @@ func (a *ReActAgent[M]) buildGraphReActRunFunc() typedRunFunc[M] {
 		input := &AgentInput{Messages: messageSliceToAny2(p.input.Messages)}
 
 		// Run the graph (synchronous invoke or streaming).
-		state, err := rg.Invoke(ctx, input, nil)
+		_, err = rg.Invoke(ctx, input, nil)
 		if err != nil {
 			p.generator.Send(&TypedAgentEvent[M]{Err: err})
 			return
 		}
 
-		// Emit the final model response as an event.
-		if len(state.Messages) > 0 {
-			last := state.Messages[len(state.Messages)-1]
-			if !isNilMessage(last) {
-				p.generator.Send(any(typedModelOutputEvent(last, nil)).(*TypedAgentEvent[M]))
-			}
-		}
-
 		// Emit afterToolCallsHook if configured.
 		if p.afterToolCallsHook != nil {
 			if err := p.afterToolCallsHook(ctx); err != nil {

internal/harness/core/react_graph.go

@@ -154,7 +154,8 @@ func NewReActGraph(agent *ReActAgent[*schema.Message], cfg *ReActGraphConfig, al
 		// Clear tool cache at start of each iteration.
 		s.ToolExecutedCache = nil
 
-		model := BuildModelWrapperChain(agentCfg.Model, nil, agentCfg, allToolInfos)
+		ec := getReActExecCtx[*schema.Message](ctx)
+		model := BuildModelWrapperChain(ctx, agentCfg.Model, ec, agentCfg, allToolInfos)
 
 		agentState := NewReActAgentState(
 			messageSliceToAny(s.Messages),

internal/harness/core/react_loop.go

@@ -11,6 +11,11 @@ import (
 
 func (a *ReActAgent[M]) buildReActRunFunc() typedRunFunc[M] {
 	return func(ctx context.Context, p *typedRunParams[M]) {
+		ec := newReActExecCtx(p.generator)
+		ec.cancelCtx = getCancelContext(ctx)
+		ec.afterToolCallsHook = p.afterToolCallsHook
+		ctx = withReActExecCtx(ctx, ec)
+
 		maxIter := a.config.MaxIterations
 		if maxIter <= 0 {
 			maxIter = 10
@@ -84,7 +89,7 @@ func (a *ReActAgent[M]) buildReActRunFunc() typedRunFunc[M] {
 		allTools = rc.Tools
 		allRD = rc.ReturnDirectly
 
-		model := BuildModelWrapperChain(a.config.Model, nil, a.config, a.exeCtx.toolInfos)
+		model := BuildModelWrapperChain(ctx, a.config.Model, ec, a.config, a.exeCtx.toolInfos)
 
 		// Build ToolsNode from merged tools, preserving ToolInvokeMiddlewares from config.
 		var tn *ToolsNode[M]
@@ -141,7 +146,6 @@ func (a *ReActAgent[M]) buildReActRunFunc() typedRunFunc[M] {
 				p.generator.Send(&TypedAgentEvent[M]{Err: fmt.Errorf("model: %w", err)})
 				return
 			}
-			p.generator.Send(typedModelOutputEvent(resp, nil))
 			state.Messages = append(state.Messages, resp)
 
 			if err := a.runAfterModelRewrite(&ctx, &state, mc, p.generator); err != nil {

internal/harness/core/retry.go

@@ -180,7 +180,7 @@ func (r *typedRetryModelWrapper[M]) generateWithShouldRetry(ctx context.Context,
 			if err != nil {
 				return zero, err
 			}
-			if execCtx != nil && execCtx.generator != nil && !isNilMessage(out) {
+			if execCtx != nil && execCtx.canSend() && !isNilMessage(out) {
 				execCtx.send(typedModelOutputEvent(out, nil))
 			}
 			return out, nil
@@ -195,7 +195,7 @@ func (r *typedRetryModelWrapper[M]) generateWithShouldRetry(ctx context.Context,
 		}
 
 		// Emit WillRetryError event before sleeping
-		if execCtx != nil && execCtx.generator != nil {
+		if execCtx != nil && execCtx.canSend() {
 			willRetry := &WillRetryError{ErrStr: lastErr.Error(), RetryAttempt: attempt + 1, rejectReason: decision.RejectReason, err: lastErr}
 			execCtx.send(&TypedAgentEvent[M]{Err: any(willRetry).(error)})
 		}
@@ -326,7 +326,7 @@ func (r *typedRetryModelWrapper[M]) streamWithShouldRetry(ctx context.Context, i
 			}
 			lastErr = err
 			if attempt < r.config.MaxRetries {
-				if execCtx != nil && execCtx.generator != nil {
+				if execCtx != nil && execCtx.canSend() {
 					execCtx.send(&TypedAgentEvent[M]{Err: &WillRetryError{ErrStr: lastErr.Error(), RetryAttempt: attempt + 1, rejectReason: decision.RejectReason, err: lastErr}})
 				}
 				applyRetryDecision(&currentInput, &currentOpts, decision)
@@ -364,7 +364,7 @@ func (r *typedRetryModelWrapper[M]) streamWithShouldRetry(ctx context.Context, i
 			default:
 			}
 			if attempt < r.config.MaxRetries {
-				if execCtx != nil && execCtx.generator != nil {
+				if execCtx != nil && execCtx.canSend() {
 					execCtx.send(&TypedAgentEvent[M]{Err: &WillRetryError{ErrStr: lastErr.Error(), RetryAttempt: attempt + 1, rejectReason: decision.RejectReason, err: lastErr}})
 				}
 				applyRetryDecision(&currentInput, &currentOpts, decision)
@@ -402,7 +402,7 @@ func (r *typedRetryModelWrapper[M]) streamWithShouldRetry(ctx context.Context, i
 				callerCh.Send(c, nil)
 			}
 			// Send output event with merged message
-			if execCtx != nil && execCtx.generator != nil && len(allChunks) > 0 {
+			if execCtx != nil && execCtx.canSend() && len(allChunks) > 0 {
 				if merged, err := mergeChunks(allChunks); err == nil {
 					execCtx.send(typedModelOutputEvent(merged, nil))
 				}

internal/harness/core/session.go

@@ -110,6 +110,7 @@ type branchEvents struct {
 // runSession holds per-execution mutable state for an agent run.
 type runSession struct {
 	mu           sync.Mutex
+	valuesInitMu sync.Mutex
 	Values       map[string]any
 	valuesMx     *sync.RWMutex
 	events       []*eventWrapEntry
@@ -125,6 +126,8 @@ func (s *runSession) shareValuesFrom(parent *runSession) {
 	if s == nil || parent == nil || s == parent {
 		return
 	}
+	s.ensureValues()
+	parent.ensureValues()
 	parent.valuesMx.RLock()
 	values := parent.Values
 	valuesMx := parent.valuesMx
@@ -138,8 +141,11 @@ func (s *runSession) ensureValues() {
 	if s == nil {
 		return
 	}
-	s.valuesMx.Lock()
-	defer s.valuesMx.Unlock()
+	s.valuesInitMu.Lock()
+	defer s.valuesInitMu.Unlock()
+	if s.valuesMx == nil {
+		s.valuesMx = &sync.RWMutex{}
+	}
 	if s.Values == nil {
 		s.Values = make(map[string]any)
 	}

internal/harness/core/stability_integration_test.go

@@ -212,6 +212,31 @@ func TestStability_CheckpointCorruption(t *testing.T) {
 		t.Logf("checkpoint roundtrip: info=%v", info)
 	})
 
+	t.Run("checkpoint_restores_resume_state", func(t *testing.T) {
+		store := newCancelTestStore()
+		cid := "resume-state"
+		wantState := "waiting-for-approval"
+		signal := &InterruptSignal{
+			Address: Address{{Type: AddressSegmentAgent, ID: "agent"}},
+			State:   wantState,
+		}
+		if err := saveCheckpoint(store, t.Context(), cid, false, &InterruptInfo{Data: "approve"}, signal); err != nil {
+			t.Fatalf("saveCheckpoint: %v", err)
+		}
+
+		_, _, info, err := loadCheckpoint(store, t.Context(), cid)
+		if err != nil {
+			t.Fatalf("loadCheckpoint: %v", err)
+		}
+		if !info.WasInterrupted {
+			t.Fatal("WasInterrupted = false, want true")
+		}
+		gotState, ok := info.InterruptState.(string)
+		if !ok || gotState != wantState {
+			t.Fatalf("InterruptState = %#v, want %#v", info.InterruptState, wantState)
+		}
+	})
+
 	t.Run("partial_checkpoint_after_cancel", func(t *testing.T) {
 		m := newCancelTestChatModel(nil)
 		m.addResp("cancel response")
@@ -260,6 +285,58 @@ func TestStability_CheckpointCorruption(t *testing.T) {
 	})
 }
 
+type checkpointResumeTestAgent struct {
+	resumeInfo *ResumeInfo
+}
+
+func (a *checkpointResumeTestAgent) Name(context.Context) string { return "checkpoint_resume" }
+func (a *checkpointResumeTestAgent) Description(context.Context) string {
+	return "checkpoint resume test agent"
+}
+
+func (a *checkpointResumeTestAgent) Run(ctx context.Context, _ *AgentInput, _ ...RunOption) *AsyncIterator[*AgentEvent] {
+	iter, gen := NewAsyncIteratorPair[*AgentEvent]()
+	go func() {
+		defer gen.Close()
+		gen.Send(StatefulInterrupt(ctx, "approval required", "saved-state"))
+	}()
+	return iter
+}
+
+func (a *checkpointResumeTestAgent) Resume(_ context.Context, info *ResumeInfo, _ ...RunOption) *AsyncIterator[*AgentEvent] {
+	a.resumeInfo = info
+	iter, gen := NewAsyncIteratorPair[*AgentEvent]()
+	gen.Send(EventFromMessage(schema.AssistantMessage("resumed"), nil, schema.RoleAssistant, ""))
+	gen.Close()
+	return iter
+}
+
+func TestRunner_ResumeRestoresInterruptState(t *testing.T) {
+	agent := &checkpointResumeTestAgent{}
+	store := newCancelTestStore()
+	runner := NewRunner(t.Context(), RunnerConfig[*schema.Message]{
+		Agent:           agent,
+		CheckPointStore: store,
+	})
+	cid := "resume-stateful-agent"
+	initial := drainAgentEvents(t, runner.Run(t.Context(), []*schema.Message{schema.UserMessage("start")}, WithCheckPointID(cid)))
+	if len(initial) == 0 || initial[0].Action == nil || initial[0].Action.Interrupted == nil {
+		t.Fatal("initial run did not emit an interrupt")
+	}
+
+	resumed, err := runner.Resume(t.Context(), cid)
+	if err != nil {
+		t.Fatalf("Resume: %v", err)
+	}
+	resumedEvents := drainAgentEvents(t, resumed)
+	if len(resumedEvents) == 0 || resumedEvents[0].Err != nil {
+		t.Fatalf("resume events = %#v", resumedEvents)
+	}
+	if agent.resumeInfo == nil || !agent.resumeInfo.WasInterrupted || agent.resumeInfo.InterruptState != "saved-state" {
+		t.Fatalf("resume info = %#v, want interrupted with saved state", agent.resumeInfo)
+	}
+}
+
 // ============================================================
 // P0-5: 10-layer nested agent cancel propagation
 // ============================================================

internal/harness/core/tool_invoke.go

@@ -227,7 +227,7 @@ func NewEventSenderToolMiddleware[M MessageType]() ToolInvokeMiddleware {
 				return nil, err
 			}
 			ec := getReActExecCtx[M](ctx)
-			if ec != nil && ec.generator != nil && result != nil {
+			if ec != nil && ec.canSend() && result != nil {
 				content := result.Content
 				if content == "" {
 					content = result.Error

internal/harness/graph/constants/constants.go

@@ -63,7 +63,8 @@ const (
 	ConfigKeyDurability = "__pregel_durability"
 	// ConfigKeyRuntime holds a Runtime instance with context, store, stream writer, etc.
 	ConfigKeyRuntime = "__pregel_runtime"
-	// ConfigKeyResumeMap holds a mapping of task ns -> resume value for resuming tasks.
+	// ConfigKeyResumeMap maps a task namespace to its resume value(s).
+	// A slice supplies values to successive Interrupt calls within that task.
 	ConfigKeyResumeMap = "__pregel_resume_map"
 )
 

internal/harness/graph/graph/loop.go

@@ -10,12 +10,12 @@
 //   - Fresh run: iterate from iteration 1 using the outer input.
 //   - Sub-graph interrupt: the closure captures its own state
 //     (iteration, current input, sub-checkpoint ID), serialises it to
-//     JSON, and re-throws via interrupt.Interrupt(ctx, encodedState).
+//     JSON, and re-throws it via interrupt.Interrupt.
 //     The outer Pregel engine saves a checkpoint and surfaces the
 //     GraphInterrupt error.
 //   - Resume: the closure is re-invoked. It reads the encoded state
-//     from interrupt.GetResumeValues(ctx), decodes iteration/input/
-//     sub-checkpoint ID, and continues from where it left off.
+//     from the context populated by checkpoint restore, decodes iteration/
+//     input/sub-checkpoint ID, and continues from where it left off.
 //     The sub-graph is invoked with a RunnableConfig whose ThreadID
 //     incorporates the sub-checkpoint namespace, so the sub-graph's
 //     own checkpointer can resume the interrupted sub-iteration.
@@ -242,9 +242,7 @@ func runLoop(
 	}
 }
 
-// loadLoopSnapshot reads the loop state from context (set by the engine
-// during checkpoint restore). Falls back to interrupt resume values for
-// backward compatibility.
+// loadLoopSnapshot reads the loop state from context set by checkpoint restore.
 func loadLoopSnapshot(ctx context.Context) (loopInterruptState, bool) {
 	// Check context key first (set by engine's checkpoint restore).
 	if ctx != nil {
@@ -258,22 +256,5 @@ func loadLoopSnapshot(ctx context.Context) (loopInterruptState, bool) {
 			}
 		}
 	}
-	// Fallback: interrupt resume values (deprecated, remove when
-	// engine.go and graph.go are migrated to loopStateCtxKey).
-	values := interrupt.GetResumeValues(ctx)
-	for _, v := range values {
-		switch tv := v.(type) {
-		case []byte:
-			var st loopInterruptState
-			if err := json.Unmarshal(tv, &st); err == nil && st.Iteration > 0 {
-				return st, true
-			}
-		case string:
-			var st loopInterruptState
-			if err := json.Unmarshal([]byte(tv), &st); err == nil && st.Iteration > 0 {
-				return st, true
-			}
-		}
-	}
 	return loopInterruptState{}, false
 }

internal/harness/graph/graph/parallel.go

@@ -296,18 +296,18 @@ func fanOutItems(
 	return resultCh
 }
 
-// loadParallelSnapshot reads the parallel state from interrupt resume values.
+// loadParallelSnapshot reads the parallel state from context set by checkpoint restore.
 func loadParallelSnapshot(ctx context.Context) (*parallelInterruptState, bool) {
-	values := interrupt.GetResumeValues(ctx)
-	for _, v := range values {
+	if ctx != nil {
 		var data []byte
-		switch tv := v.(type) {
+		switch value := ctx.Value(interrupt.SubGraphStateCtxKey).(type) {
 		case []byte:
-			data = tv
+			data = value
 		case string:
-			data = []byte(tv)
-		default:
-			continue
+			data = []byte(value)
+		}
+		if data == nil {
+			return nil, false
 		}
 		var st parallelInterruptState
 		if err := json.Unmarshal(data, &st); err == nil && st.TotalCount > 0 {

internal/harness/graph/interrupt/interrupt.go

@@ -4,6 +4,7 @@ package interrupt
 import (
 	"context"
 	"crypto/sha256"
+	stderrors "errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -15,6 +16,9 @@ import (
 // contextKey is the key for interrupt context in context.Context.
 type contextKey struct{}
 
+// ErrMissingInterruptContext is returned when Interrupt is called outside a graph task.
+var ErrMissingInterruptContext = stderrors.New("interrupt requires a graph execution context")
+
 // SubGraphStateCtxKey is the context key for sub-graph checkpoint state
 // (e.g. Loop iteration, currentInput). Defined here so that both the
 // engine (pregel) and the sub-graph node (graph/loop.go) can access it
@@ -25,6 +29,9 @@ var SubGraphStateCtxKey = SubGraphStateCtxKeyType{}
 
 // WithInterruptContext creates a new context with interrupt support.
 func WithInterruptContext(ctx context.Context) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	return context.WithValue(ctx, contextKey{}, &interruptContext{
 		resumeValues: make([]interface{}, 0),
 		index:        0,
@@ -33,6 +40,9 @@ func WithInterruptContext(ctx context.Context) context.Context {
 
 // GetInterruptContext retrieves the interrupt context from the context.
 func GetInterruptContext(ctx context.Context) *interruptContext {
+	if ctx == nil {
+		return nil
+	}
 	if ic, ok := ctx.Value(contextKey{}).(*interruptContext); ok {
 		return ic
 	}
@@ -51,9 +61,9 @@ func IsInterruptContext(ctx context.Context) bool {
 // exception, halting execution. The provided value is included with the exception
 // and sent to the client executing the graph.
 //
-// A client resuming the graph must use the Command primitive to specify a value
-// for the interrupt and continue execution.
-// The graph resumes from the start of the node, re-executing all logic.
+// A caller resumes the graph by supplying task-scoped resume values through
+// the graph runtime. The graph resumes from the start of the node, re-executing
+// all logic.
 //
 // If a node contains multiple interrupt calls, LangGraph matches resume values
 // to interrupts based on their order in the node.
@@ -63,8 +73,7 @@ func IsInterruptContext(ctx context.Context) bool {
 func Interrupt(ctx context.Context, value interface{}) (interface{}, error) {
 	ic := GetInterruptContext(ctx)
 	if ic == nil {
-		// Fall back to global context for backward compatibility
-		ic = globalContext
+		return nil, ErrMissingInterruptContext
 	}
 
 	// Try to consume the next pending resume value under a single lock
@@ -73,14 +82,6 @@ func Interrupt(ctx context.Context, value interface{}) (interface{}, error) {
 		return v, nil
 	}
 
-	// Check for current resume value
-	v := ic.getNullResume()
-	if v != nil {
-		ic.setNullResume(nil) // consume it before appending
-		ic.appendResumeValue(v)
-		return v, nil
-	}
-
 	// No resume value found, raise interrupt
 	return nil, &errors.GraphInterrupt{
 		Interrupts: []interface{}{
@@ -97,18 +98,11 @@ type interruptContext struct {
 	mu           sync.Mutex
 	resumeValues []interface{}
 	index        int
-	nullResume   interface{}
 }
 
-// Global context for backward compatibility
 // interruptIDCounter provides unique IDs across all interrupt points in the process.
 var interruptIDCounter int64
 
-var globalContext = &interruptContext{
-	resumeValues: make([]interface{}, 0),
-	index:        0,
-}
-
 // consumeNextResumeValue atomically reads the next resume value and advances
 // the index under a single lock (avoids TOCTOU between separate lock acquisitions).
 func (ic *interruptContext) consumeNextResumeValue() (interface{}, bool) {
@@ -147,16 +141,6 @@ func (ic *interruptContext) getInterruptIndex() int {
 	return ic.index
 }
 
-// getNullResume checks for a null resume value.
-func (ic *interruptContext) getNullResume() interface{} {
-	if ic == nil {
-		return nil
-	}
-	ic.mu.Lock()
-	defer ic.mu.Unlock()
-	return ic.nullResume
-}
-
 // appendResumeValue appends a resume value.
 func (ic *interruptContext) appendResumeValue(v interface{}) {
 	if ic == nil {
@@ -167,26 +151,6 @@ func (ic *interruptContext) appendResumeValue(v interface{}) {
 	ic.resumeValues = append(ic.resumeValues, v)
 }
 
-// setNullResume sets the null resume value.
-func (ic *interruptContext) setNullResume(v interface{}) {
-	if ic == nil {
-		return
-	}
-	ic.mu.Lock()
-	defer ic.mu.Unlock()
-	ic.nullResume = v
-}
-
-// setResumeValues replaces the resume values.
-func (ic *interruptContext) setResumeValues(values []interface{}) {
-	if ic == nil {
-		return
-	}
-	ic.mu.Lock()
-	defer ic.mu.Unlock()
-	ic.resumeValues = values
-}
-
 // reset clears all interrupt context fields.
 func (ic *interruptContext) reset() {
 	if ic == nil {
@@ -196,67 +160,30 @@ func (ic *interruptContext) reset() {
 	defer ic.mu.Unlock()
 	ic.resumeValues = make([]interface{}, 0)
 	ic.index = 0
-	ic.nullResume = nil
 }
 
 // GetResumeValues returns the current resume values from context.
 func GetResumeValues(ctx context.Context) []interface{} {
-	var ic *interruptContext
-	if ctx != nil {
-		ic = GetInterruptContext(ctx)
-	}
-	if ic == nil {
-		ic = globalContext
-	}
+	ic := GetInterruptContext(ctx)
 	return ic.getResumeValues()
 }
 
 // GetInterruptIndex returns the current interrupt index from context.
 func GetInterruptIndex(ctx context.Context) int {
 	ic := GetInterruptContext(ctx)
-	if ic == nil {
-		ic = globalContext
-	}
 	return ic.getInterruptIndex()
 }
 
 // AppendResumeValue appends a resume value to the context.
 func AppendResumeValue(ctx context.Context, value interface{}) {
-	var ic *interruptContext
-	if ctx != nil {
-		ic = GetInterruptContext(ctx)
-	}
-	if ic == nil {
-		ic = globalContext
-	}
+	ic := GetInterruptContext(ctx)
 	ic.appendResumeValue(value)
 }
 
-// GetNullResume gets the null resume value from context.
-// If consume is true, the value is cleared after retrieval.
-func GetNullResume(ctx context.Context, consume bool) interface{} {
-	ic := GetInterruptContext(ctx)
-	if ic == nil {
-		ic = globalContext
-	}
-	v := ic.getNullResume()
-	if consume {
-		ic.setNullResume(nil)
-	}
-	return v
-}
-
-// Reset clears the interrupt context.
-// When a per-request context is found, only that context is reset.
-// The global fallback context is only reset when no per-request context
-// exists, preventing concurrent requests from corrupting each other.
+// Reset clears the interrupt context attached to ctx.
 func Reset(ctx context.Context) {
 	ic := GetInterruptContext(ctx)
-	if ic != nil {
-		ic.reset()
-		return
-	}
-	globalContext.reset()
+	ic.reset()
 }
 
 // generateInterruptID generates a unique ID for an interrupt.
@@ -291,12 +218,3 @@ func GetInterruptValue(err error) (interface{}, bool) {
 
 	return nil, false
 }
-
-// SetResumeValues sets the resume values for testing.
-func SetResumeValues(ctx context.Context, values []interface{}) {
-	ic := GetInterruptContext(ctx)
-	if ic == nil {
-		ic = globalContext
-	}
-	ic.setResumeValues(values)
-}

internal/harness/graph/interrupt/interrupt_test.go

@@ -148,11 +148,12 @@ func TestIsInterruptContext(t *testing.T) {
 	}
 }
 
-func TestSetResumeValues(t *testing.T) {
+func TestAppendResumeValues(t *testing.T) {
 	ctx := WithInterruptContext(context.Background())
 
-	// Set multiple resume values at once
-	SetResumeValues(ctx, []interface{}{"a", "b", "c"})
+	AppendResumeValue(ctx, "a")
+	AppendResumeValue(ctx, "b")
+	AppendResumeValue(ctx, "c")
 
 	// Check values
 	values := GetResumeValues(ctx)
@@ -161,15 +162,13 @@ func TestSetResumeValues(t *testing.T) {
 	}
 }
 
-func TestGlobalContext(t *testing.T) {
-	// Reset global context
-	Reset(context.Background())
-
-	// Test with nil context (should fall back to global)
-	AppendResumeValue(nil, "global_value")
-
-	values := GetResumeValues(nil)
-	if len(values) != 1 || values[0] != "global_value" {
-		t.Error("Should use global context when ctx is nil")
+func TestInterruptRequiresContext(t *testing.T) {
+	ctx := context.Background()
+	AppendResumeValue(ctx, "orphaned")
+	if values := GetResumeValues(ctx); len(values) != 0 {
+		t.Fatalf("resume values without an interrupt context = %v, want none", values)
+	}
+	if _, err := Interrupt(ctx, "input"); err != ErrMissingInterruptContext {
+		t.Fatalf("Interrupt error = %v, want %v", err, ErrMissingInterruptContext)
 	}
 }

internal/harness/graph/pregel/engine.go

@@ -52,6 +52,70 @@ type Engine struct {
 	deferredCheckpoints []deferredCheckpoint // for DurabilityExit mode
 }
 
+type resumeMapContextKey struct{}
+
+func withResumeMap(ctx context.Context, config *types.RunnableConfig) context.Context {
+	if config == nil || config.Configurable == nil {
+		return ctx
+	}
+	configured, ok := config.Configurable[constants.ConfigKeyResumeMap].(map[string]any)
+	if !ok || len(configured) == 0 {
+		return ctx
+	}
+	merged := make(map[string]any)
+	if inherited, ok := ctx.Value(resumeMapContextKey{}).(map[string]any); ok {
+		for key, value := range inherited {
+			merged[key] = value
+		}
+	}
+	for key, value := range configured {
+		merged[key] = value
+	}
+	return context.WithValue(ctx, resumeMapContextKey{}, merged)
+}
+
+func resumeValuesForTask(ctx context.Context, namespace string, task *Task) []any {
+	if task == nil {
+		return nil
+	}
+	resumeMap, ok := ctx.Value(resumeMapContextKey{}).(map[string]any)
+	if !ok {
+		return nil
+	}
+	keys := make([]string, 0, 4)
+	if namespace != "" {
+		if path := TaskPathStr(task.Path); path != "" {
+			keys = append(keys, namespace+string(constants.NSSep)+path)
+		}
+		keys = append(keys, namespace+string(constants.NSSep)+task.Name)
+	}
+	if path := TaskPathStr(task.Path); path != "" {
+		keys = append(keys, path)
+	}
+	keys = append(keys, task.Name, task.ID)
+	for _, key := range keys {
+		if value, ok := resumeMap[key]; ok {
+			if values, ok := value.([]any); ok {
+				return values
+			}
+			return []any{value}
+		}
+	}
+	return nil
+}
+
+func (e *Engine) contextForTask(ctx context.Context, task *Task) context.Context {
+	taskCtx := interrupt.WithInterruptContext(ctx)
+	var namespace string
+	if e.config != nil {
+		namespace = e.config.GetOrEmpty(constants.ConfigKeyCheckpointNS)
+	}
+	for _, value := range resumeValuesForTask(ctx, namespace, task) {
+		interrupt.AppendResumeValue(taskCtx, value)
+	}
+	return taskCtx
+}
+
 // deferredCheckpoint stores checkpoint data for deferred saving (DurabilityExit mode)
 type deferredCheckpoint struct {
 	ThreadID     string
@@ -203,6 +267,7 @@ type ExecuteResult struct {
 func (e *Engine) Run(ctx context.Context, input any, mode types.StreamMode) (<-chan any, <-chan error) {
 	outputCh := make(chan any, 100)
 	errCh := make(chan error, 1)
+	ctx = withResumeMap(ctx, e.config)
 
 	go func() {
 		defer close(errCh)
@@ -1164,7 +1229,8 @@ func (e *Engine) executeTasksAsync(
 			retryPolicy := e.resolveRetryPolicy(t)
 
 			// Execute with async pipeline
-			resultCh := asyncPipeline.ExecuteNode(ctx, t.Name, executeFn, &RetryConfig{Policy: retryPolicy})
+			taskCtx := e.contextForTask(ctx, t)
+			resultCh := asyncPipeline.ExecuteNode(taskCtx, t.Name, executeFn, &RetryConfig{Policy: retryPolicy})
 
 			// Wait for result
 			select {
@@ -1221,6 +1287,8 @@ func (e *Engine) executeTask(
 	task *Task,
 	registry *channels.Registry,
 ) *TaskResult {
+	ctx = e.contextForTask(ctx, task)
+
 	// Read input for this task
 	input, err := e.readTaskInput(registry, task)
 	if err != nil {

internal/harness/graph/pregel/pregel_interrupt_nested_test.go

@@ -4,12 +4,16 @@
 package pregel
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"testing"
 
+	"ragflow/internal/harness/graph/channels"
 	"ragflow/internal/harness/graph/checkpoint"
 	"ragflow/internal/harness/graph/constants"
+	"ragflow/internal/harness/graph/graph"
+	graphinterrupt "ragflow/internal/harness/graph/interrupt"
 	"ragflow/internal/harness/graph/types"
 )
 
@@ -139,6 +143,47 @@ func TestInterrupt_WithCheckpointer(t *testing.T) {
 	}
 }
 
+func TestInterrupt_ResumeValueFromConfig(t *testing.T) {
+	saver := checkpoint.NewMemorySaver()
+	cfg := &types.RunnableConfig{Configurable: map[string]interface{}{
+		constants.ConfigKeyThreadID: "dynamic-interrupt-resume",
+	}}
+	sg := graph.NewStateGraph(map[string]any{"value": ""})
+	sg.AddChannel("value", channels.NewLastValue(""))
+	sg.AddNode("request_approval", func(ctx context.Context, state any) (any, error) {
+		value, err := graphinterrupt.Interrupt(ctx, "approval required")
+		if err != nil {
+			return nil, err
+		}
+		state.(map[string]any)["value"] = value
+		return state, nil
+	})
+	if err := sg.AddEdge(constants.Start, "request_approval"); err != nil {
+		t.Fatal(err)
+	}
+	if err := sg.AddEdge("request_approval", constants.End); err != nil {
+		t.Fatal(err)
+	}
+	engine := NewEngine(sg,
+		WithRecursionLimit(10),
+		WithCheckpointer(saver),
+		WithConfig(cfg),
+	)
+
+	if _, err := engine.RunSync(t.Context(), map[string]any{"value": ""}); err == nil {
+		t.Fatal("expected the first execution to interrupt")
+	}
+
+	cfg.Configurable[constants.ConfigKeyResumeMap] = map[string]interface{}{"request_approval": "approved"}
+	result, err := engine.RunSync(t.Context(), map[string]any{"value": ""})
+	if err != nil {
+		t.Fatalf("resume execution: %v", err)
+	}
+	if got := result.(map[string]any)["value"]; got != "approved" {
+		t.Fatalf("resumed value = %v, want approved", got)
+	}
+}
+
 // ============================================================
 // P1: No-checkpointer interrupt
 // ============================================================