13Runner 与 flowAgent：运行骨架

type TypedRunner[M MessageType] struct {
    a               TypedAgent[M]
    enableStreaming bool
    store           CheckPointStore
}

// 你在 Part I 一直用的 Runner,其实是它的一个具体实例化。
type Runner = TypedRunner[*schema.Message]

// NewRunner creates a new Runner with the given config.
func NewRunner(_ context.Context, conf RunnerConfig) *Runner {
  return NewTypedRunner(conf)
}

type TypedRunnerConfig[M MessageType] struct {
  Agent           TypedAgent[M]
  EnableStreaming bool

  CheckPointStore CheckPointStore
}

// Query is a convenience method that starts a new execution with a single user query string.
func (r *TypedRunner[M]) Query(ctx context.Context,
  query string, opts ...AgentRunOption) *AsyncIterator[*TypedAgentEvent[M]] {
  msgs, err := newUserMessage[M](query)
  if err != nil {
    return errorIterator[M](err)
  }
  return r.Run(ctx, []M{msgs}, opts...)
}

func (r *TypedRunner[M]) Run(ctx context.Context, messages []M,
  opts ...AgentRunOption) *AsyncIterator[*TypedAgentEvent[M]] {
  return typedRunnerRunImpl(r.a, r.enableStreaming, r.store, ctx, messages, opts...)
}

func typedRunnerRunImpl[M MessageType](a TypedAgent[M], enableStreaming bool, store CheckPointStore, ctx context.Context, messages []M, opts ...AgentRunOption) *AsyncIterator[*TypedAgentEvent[M]] {
  o := getCommonOptions(nil, opts...)

  input := &TypedAgentInput[M]{
    Messages:        messages,
    EnableStreaming: enableStreaming,
  }

  var zero M
  if _, ok := any(zero).(*schema.Message); ok {
    concreteAgent, _ := any(a).(Agent)
    fa := toFlowAgent(ctx, concreteAgent)
    if store != nil {
      fa.checkPointStore = store
    }
    concreteInput := any(input).(*AgentInput)
    ctx = ctxWithNewTypedRunCtx(ctx, input, o.sharedParentSession)
    AddSessionValues(ctx, o.sessionValues)

    iter := fa.Run(ctx, concreteInput, opts...)

    if store == nil && o.cancelCtx == nil {
      return any(iter).(*AsyncIterator[*TypedAgentEvent[M]])
    }

    niter, gen := NewAsyncIteratorPair[*TypedAgentEvent[M]]()
    go typedRunnerHandleIterImpl(enableStreaming, store, ctx, any(iter).(*AsyncIterator[*TypedAgentEvent[M]]), gen, o.checkPointID, o.cancelCtx)
    return niter
  }

  fa := toTypedFlowAgent(a)
  if store != nil {
    fa.checkPointStore = store
  }

  ctx = ctxWithNewTypedRunCtx(ctx, input, o.sharedParentSession)
  AddSessionValues(ctx, o.sessionValues)

  iter := fa.Run(ctx, input, opts...)

  if store == nil && o.cancelCtx == nil {
    return iter
  }

  niter, gen := NewAsyncIteratorPair[*TypedAgentEvent[M]]()
  go typedRunnerHandleIterImpl(enableStreaming, store, ctx, iter, gen, o.checkPointID, o.cancelCtx)
  return niter
}

func toFlowAgent(ctx context.Context, agent Agent, opts ...AgentOption) *flowAgent {
    var fa *flowAgent
    var ok bool
    if fa, ok = agent.(*flowAgent); !ok {
        fa = &flowAgent{Agent: agent}   // 普通 Agent:包一层
    } else {
        fa = fa.deepCopy()              // 已经是 flowAgent:深拷贝一份
    }
    // ...补上默认的 historyRewriter
    return fa
}

type flowAgent struct {
    Agent                              // 嵌入:原 Agent 的能力原样保留

    subAgents   []*flowAgent           // 它的下属
    parentAgent *flowAgent             // 它的上级

    disallowTransferToParent bool
    historyRewriter          HistoryRewriter
    checkPointStore          compose.CheckPointStore
}

func (a *flowAgent) Run(ctx context.Context, input *AgentInput, opts ...AgentRunOption) *AsyncIterator[*AgentEvent] {
  agentName := a.Name(ctx)

  var runCtx *runContext
  ctx, runCtx = initRunCtx(ctx, agentName, input)
  ctx = AppendAddressSegment(ctx, AddressSegmentAgent, agentName)

  o := getCommonOptions(nil, opts...)
  cancelCtx := o.cancelCtx

  processedInput, err := a.genAgentInput(ctx, runCtx, o.skipTransferMessages)
  if err != nil {
    if cancelCtx != nil {
      cancelCtx.markDone()
    }
    cbInput := &AgentCallbackInput{Input: input}
    ctx = callbacks.OnStart(ctx, cbInput)
    return wrapIterWithOnEnd(ctx, genErrorIter(err))
  }

  ctxForSubAgents := ctx

  agentType := getAgentType(a.Agent)
  ctx = initAgentCallbacks(ctx, agentName, agentType, filterOptions(agentName, opts)...)
  cbInput := &AgentCallbackInput{Input: processedInput}
  ctx = callbacks.OnStart(ctx, cbInput)

  input = processedInput

  if wf, ok := a.Agent.(*workflowAgent); ok {
    ctx = withCancelContext(ctx, cancelCtx)
    filteredOpts := filterCancelOption(filterCallbackHandlersForNestedAgents(agentName, opts))
    iter := wf.Run(ctx, input, filteredOpts...)
    iter = wrapIterWithCancelCtx(iter, cancelCtx)
    return wrapIterWithOnEnd(ctx, iter)
  }

  aIter := a.Agent.Run(withCancelContext(ctx, cancelCtx), input, filterOptions(agentName, opts)...)

  iterator, generator := NewAsyncIteratorPair[*AgentEvent]()

  go a.run(withCancelContext(ctx, cancelCtx), withCancelContext(ctxForSubAgents, cancelCtx), runCtx, aIter, generator, filterCancelOption(opts)...)

  return wrapIterWithCancelCtx(iterator, cancelCtx)
}

func (a *flowAgent) run(
  ctx context.Context,
  ctxForSubAgents context.Context,
  runCtx *runContext,
  aIter *AsyncIterator[*AgentEvent],
  generator *AsyncGenerator[*AgentEvent],
  opts ...AgentRunOption) {

  cbIter, cbGen := NewAsyncIteratorPair[*AgentEvent]()

  cbOutput := &AgentCallbackOutput{Events: cbIter}
  icb.On(ctx, cbOutput, icb.BuildOnEndHandleWithCopy(copyAgentCallbackOutput), callbacks.TimingOnEnd, false)

  defer func() {
    panicErr := recover()
    if panicErr != nil {
      e := safe.NewPanicErr(panicErr, debug.Stack())
      generator.Send(&AgentEvent{Err: e})
    }

    cbGen.Close()
    generator.Close()
  }()

  var lastAction *AgentAction
  for {
    event, ok := aIter.Next()
    if !ok {
      break
    }

    // RunPath ownership: the eino framework sets RunPath exactly once.
    // If event.RunPath is already set (e.g., by agentTool), we don't modify it.
    // If event.RunPath is nil/empty, we set it to the current runCtx.RunPath.
    // This ensures RunPath is set exactly once and not duplicated.
    if len(event.RunPath) == 0 {
      event.AgentName = a.Name(ctx)
      event.RunPath = runCtx.RunPath
    }
    // Recording policy: exact RunPath match (non-interrupt) indicates events belonging to this agent execution.
    // This prevents parent recording of child/tool-internal emissions.
    if (event.Action == nil || event.Action.Interrupted == nil) && exactRunPathMatch(runCtx.RunPath, event.RunPath) {
      // copy the event so that the copied event's stream is exclusive for any potential consumer
      // copy before adding to session because once added to session it's stream could be consumed by genAgentInput at any time
      // interrupt action are not added to session, because ALL information contained in it
      // is either presented to end-user, or made available to agents through other means
      copied := copyTypedAgentEvent(event)
      setAutomaticClose(copied)
// … 省略 52 行；完整声明 L478–577，点击上方「浏览完整文件」

if len(event.RunPath) == 0 {
    event.AgentName = a.Name(ctx)
    event.RunPath = runCtx.RunPath
}

if (event.Action == nil || event.Action.Interrupted == nil) && exactRunPathMatch(runCtx.RunPath, event.RunPath) {
    copied := copyTypedAgentEvent(event)
    // ...
    runCtx.Session.addEvent(copied)
}

if destName != "" {
    agentToRun := a.getAgent(ctxForSubAgents, destName)
    // ...
    subAIter := agentToRun.Run(ctxForSubAgents, nil /* 子 Agent 从 runCtx 取输入 */, opts...)
    for {
        subEvent, ok_ := subAIter.Next()
        if !ok_ { break }
        generator.Send(subEvent)
    }
}

type runContext struct {
    RootInput        *AgentInput   // 最初的用户输入
    RunPath          []RunStep     // 当前的 Agent 运行路径
    AgenticRootInput any
    Session          *runSession   // 跨 Agent 共享的会话
}

// runSession CheckpointSchema: persisted via serialization.RunCtx (gob).
type runSession struct {
  Values    map[string]any
  valuesMtx *sync.Mutex

  Events     []*agentEventWrapper
  LaneEvents *laneEvents
  mtx        sync.Mutex

  // TypedEvents stores *[]*typedAgentEventWrapper[M] for M != *schema.Message.
  // For M = *schema.Message, the existing Events field is used instead.
  // The any type is required because Go does not support generic fields in non-generic structs.
  TypedEvents any
}

func initRunCtx(ctx context.Context, agentName string, input *AgentInput) (context.Context, *runContext) {
  runCtx := getRunCtx(ctx)
  if runCtx != nil {
    runCtx = runCtx.deepCopy()
  } else {
    runCtx = &runContext{Session: newRunSession()}
  }

  runCtx.RunPath = append(runCtx.RunPath, RunStep{agentName: agentName})
  if runCtx.isRoot() && input != nil {
    runCtx.RootInput = input
  }

  return setRunCtx(ctx, runCtx), runCtx
}

func typedRunnerHandleIterImpl[M MessageType](enableStreaming bool, store CheckPointStore, ctx context.Context, aIter *AsyncIterator[*TypedAgentEvent[M]], //nolint:revive // argument-limit
  gen *AsyncGenerator[*TypedAgentEvent[M]], checkPointID *string, cancelCtx *cancelContext) {
  defer func() {
    panicErr := recover()
    if panicErr != nil {
      e := safe.NewPanicErr(panicErr, debug.Stack())
      gen.Send(&TypedAgentEvent[M]{Err: e})
    }

    gen.Close()
  }()
  var (
    interruptSignal *core.InterruptSignal
    legacyData      any
  )
  for {
    event, ok := aIter.Next()
    if !ok {
      break
    }

    if event.Err != nil {
      var cancelErr *CancelError
      if errors.As(event.Err, &cancelErr) {
        if cancelCtx != nil && cancelCtx.isRoot() && cancelCtx.shouldCancel() {
          cancelCtx.markCancelHandled()
        }
        if cancelErr.interruptSignal != nil && checkPointID != nil {
          cancelErr.InterruptContexts = core.ToInterruptContexts(cancelErr.interruptSignal, allowedAddressSegmentTypes)
          err := runnerSaveCheckPointImpl(enableStreaming, store, ctx, *checkPointID, &InterruptInfo{}, cancelErr.interruptSignal)
          if err != nil {
            gen.Send(&TypedAgentEvent[M]{Err: fmt.Errorf("failed to save checkpoint on cancel: %w", err)})
          }
        }
        gen.Send(event)
        break
      }
    }

    if event.Action != nil && event.Action.internalInterrupted != nil {
      if interruptSignal != nil {
        panic("multiple interrupt actions should not happen in Runner")
      }
      interruptSignal = event.Action.internalInterrupted
      interruptContexts := core.ToInterruptContexts(interruptSignal, allowedAddressSegmentTypes)
      event = &TypedAgentEvent[M]{
        AgentName: event.AgentName,
        RunPath:   event.RunPath,
// … 省略 24 行；完整声明 L271–342，点击上方「浏览完整文件」

err := gob.NewEncoder(buf).Encode(&serialization{
    RunCtx:              runCtx,        // 整个运行环境(含 session、RunPath)
    Info:                info,
    InterruptID2Address: id2Addr,       // 中断点的地址映射
    InterruptID2State:   id2State,
    EnableStreaming:     enableStreaming,
})
// ...
return store.Set(ctx, key, buf.Bytes())

// Resume continues an interrupted execution from a checkpoint, using an "Implicit Resume All" strategy.
// This method is best for simpler use cases where the act of resuming implies that all previously
// interrupted points should proceed without specific data.
//
// When using this method, all interrupted agents will receive `isResumeFlow = false` when they
// call `GetResumeContext`, as no specific agent was targeted. This is suitable for the "Simple Confirmation"
// pattern where an agent only needs to know `wasInterrupted` is true to continue.
func (r *TypedRunner[M]) Resume(ctx context.Context, checkPointID string, opts ...AgentRunOption) (
  *AsyncIterator[*TypedAgentEvent[M]], error) {
  return r.resumeInternal(ctx, checkPointID, nil, opts...)
}

// ResumeWithParams continues an interrupted execution from a checkpoint with specific parameters.
// This is the most common and powerful way to resume, allowing you to target specific interrupt points
// (identified by their address/ID) and provide them with data.
//
// The params.Targets map should contain the addresses of the components to be resumed as keys. These addresses
// can point to any interruptible component in the entire execution graph, including ADK agents, compose
// graph nodes, or tools. The value can be the resume data for that component, or `nil` if no data is needed.
//
// When using this method:
//   - Components whose addresses are in the params.Targets map will receive `isResumeFlow = true` when they
//     call `GetResumeContext`.
//   - Interrupted components whose addresses are NOT in the params.Targets map must decide how to proceed:
//     -- "Leaf" components (the actual root causes of the original interrupt) MUST re-interrupt themselves
//     to preserve their state.
//     -- "Composite" agents (like SequentialAgent or ChatModelAgent) should generally proceed with their
//     execution. They act as conduits, allowing the resume signal to flow to their children. They will
//     naturally re-interrupt if one of their interrupted children re-interrupts, as they receive the
//     new `CompositeInterrupt` signal from them.
func (r *TypedRunner[M]) ResumeWithParams(ctx context.Context, checkPointID string, params *ResumeParams, opts ...AgentRunOption) (*AsyncIterator[*TypedAgentEvent[M]], error) {
  return r.resumeInternal(ctx, checkPointID, params.Targets, opts...)
}