cloudwego--eino/internal/core/address.go

/*
 * Copyright 2025 CloudWeGo Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package core

import (
	"context"
	"fmt"
	"strings"
	"sync"

	"github.com/cloudwego/eino/internal/generic"
)

// AddressSegmentType defines the type of a segment in an execution address.
type AddressSegmentType string

// Address represents a full, hierarchical address to a point in the execution structure.
type Address []AddressSegment

// String converts an Address into its unique string representation.
func (p Address) String() string {
	if p == nil {
		return ""
	}
	var sb strings.Builder
	for i, s := range p {
		sb.WriteString(string(s.Type))
		sb.WriteString(":")
		sb.WriteString(s.ID)
		if s.SubID != "" {
			sb.WriteString(":")
			sb.WriteString(s.SubID)
		}
		if i != len(p)-1 {
			sb.WriteString(";")
		}
	}
	return sb.String()
}

func (p Address) Equals(other Address) bool {
	if len(p) != len(other) {
		return false
	}
	for i := range p {
		if p[i].Type != other[i].Type || p[i].ID != other[i].ID || p[i].SubID != other[i].SubID {
			return false
		}
	}
	return true
}

// AddressSegment represents a single segment in the hierarchical address of an execution point.
// A sequence of AddressSegments uniquely identifies a location within a potentially nested structure.
type AddressSegment struct {
	// ID is the unique identifier for this segment, e.g., the node's key or the tool's name.
	ID string
	// Type indicates whether this address segment is a graph node, a tool call, an agent, etc.
	Type AddressSegmentType
	// In some cases, ID alone are not unique enough, we need this SubID to guarantee uniqueness.
	// e.g. parallel tool calls with the same name but different tool call IDs.
	SubID string
}

type addrCtxKey struct{}

type addrCtx struct {
	addr           Address
	interruptState *InterruptState
	isResumeTarget bool
	resumeData     any
}

type globalResumeInfoKey struct{}

type globalResumeInfo struct {
	mu                sync.RWMutex
	id2ResumeData     map[string]any
	id2ResumeDataUsed map[string]bool
	id2State          map[string]InterruptState
	id2StateUsed      map[string]bool
	id2Addr           map[string]Address
}

// GetCurrentAddress returns the hierarchical address of the currently executing component.
// The address is a sequence of segments, each identifying a structural part of the execution
// like an agent, a graph node, or a tool call. This can be useful for logging or debugging.
func GetCurrentAddress(ctx context.Context) Address {
	if p, ok := ctx.Value(addrCtxKey{}).(*addrCtx); ok {
		return p.addr
	}

	return nil
}

// AppendAddressSegment creates a new execution context for a sub-component (e.g., a graph node or a tool call).
//
// It extends the current context's address with a new segment and populates the new context with the
// appropriate interrupt state and resume data for that specific sub-address.
//
//   - ctx: The parent context, typically the one passed into the component's Invoke/Stream method.
//   - segType: The type of the new address segment (e.g., "node", "tool").
//   - segID: The unique ID for the new address segment.
func AppendAddressSegment(ctx context.Context, segType AddressSegmentType, segID string,
	subID string) context.Context {
	// get current address
	currentAddress := GetCurrentAddress(ctx)
	if len(currentAddress) == 0 {
		currentAddress = []AddressSegment{
			{
				Type:  segType,
				ID:    segID,
				SubID: subID,
			},
		}
	} else {
		newAddress := make([]AddressSegment, len(currentAddress)+1)
		copy(newAddress, currentAddress)
		newAddress[len(newAddress)-1] = AddressSegment{
			Type:  segType,
			ID:    segID,
			SubID: subID,
		}
		currentAddress = newAddress
	}

	runCtx := &addrCtx{
		addr: currentAddress,
	}

	rInfo, hasRInfo := getResumeInfo(ctx)
	if !hasRInfo {
		return context.WithValue(ctx, addrCtxKey{}, runCtx)
	}

	rInfo.mu.Lock()
	defer rInfo.mu.Unlock()

	var id string
	for id_, addr := range rInfo.id2Addr {
		if addr.Equals(currentAddress) {
			if used, ok := rInfo.id2StateUsed[id_]; !ok || !used {
				runCtx.interruptState = generic.PtrOf(rInfo.id2State[id_])
				rInfo.id2StateUsed[id_] = true
				id = id_
				break
			}
		}
	}

	used := rInfo.id2ResumeDataUsed[id]
	if !used {
		rData, existed := rInfo.id2ResumeData[id]
		if existed {
			rInfo.id2ResumeDataUsed[id] = true
			runCtx.resumeData = rData
			runCtx.isResumeTarget = true
		}
	}

	if !runCtx.isResumeTarget {
		for id_, addr := range rInfo.id2Addr {
			if len(addr) > len(currentAddress) && addr[:len(currentAddress)].Equals(currentAddress) {
				if !rInfo.id2ResumeDataUsed[id_] {
					runCtx.isResumeTarget = true
					break
				}
			}
		}
	}

	return context.WithValue(ctx, addrCtxKey{}, runCtx)
}

// GetNextResumptionPoints finds the immediate child resumption points for a given parent address.
func GetNextResumptionPoints(ctx context.Context) (map[string]bool, error) {
	parentAddr := GetCurrentAddress(ctx)

	rInfo, exists := getResumeInfo(ctx)
	if !exists {
		return nil, fmt.Errorf("GetNextResumptionPoints: failed to get resume info from context")
	}

	rInfo.mu.RLock()
	defer rInfo.mu.RUnlock()

	nextPoints := make(map[string]bool)
	parentAddrLen := len(parentAddr)

	for _, addr := range rInfo.id2Addr {
		// Check if addr is a potential child (must be longer than parent)
		if len(addr) <= parentAddrLen {
			continue
		}

		// Check if it has the parent address as a prefix
		var isPrefix bool
		if parentAddrLen == 0 {
			isPrefix = true
		} else {
			isPrefix = addr[:parentAddrLen].Equals(parentAddr)
		}

		if !isPrefix {
			continue
		}

		// We are looking for immediate children.
		// The address of an immediate child should be one segment longer.
		childAddr := addr[parentAddrLen : parentAddrLen+1]
		childID := childAddr[0].ID

		// Avoid adding duplicates.
		if _, ok := nextPoints[childID]; !ok {
			nextPoints[childID] = true
		}
	}

	return nextPoints, nil
}

// BatchResumeWithData is the core function for preparing a resume context. It injects a map
// of resume targets and their corresponding data into the context.
//
// The `resumeData` map should contain the interrupt IDs (which are the string form of addresses) of the
// components to be resumed as keys. The value can be the resume data for that component, or `nil`
// if no data is needed (equivalent to using `Resume`).
//
// This function is the foundation for the "Explicit Targeted Resume" strategy. Components whose interrupt IDs
// are present as keys in the map will receive `isResumeFlow = true` when they call `GetResumeContext`.
func BatchResumeWithData(ctx context.Context, resumeData map[string]any) context.Context {
	rInfo, ok := ctx.Value(globalResumeInfoKey{}).(*globalResumeInfo)
	if !ok {
		// Create a new globalResumeInfo and copy the map to prevent external mutation.
		newMap := make(map[string]any, len(resumeData))
		for k, v := range resumeData {
			newMap[k] = v
		}
		return context.WithValue(ctx, globalResumeInfoKey{}, &globalResumeInfo{
			id2ResumeData:     newMap,
			id2ResumeDataUsed: make(map[string]bool),
			id2StateUsed:      make(map[string]bool),
		})
	}

	rInfo.mu.Lock()
	defer rInfo.mu.Unlock()
	if rInfo.id2ResumeData == nil {
		rInfo.id2ResumeData = make(map[string]any)
	}
	for id, data := range resumeData {
		rInfo.id2ResumeData[id] = data
	}
	return ctx
}

func PopulateInterruptState(ctx context.Context, id2Addr map[string]Address,
	id2State map[string]InterruptState) context.Context {
	rInfo, ok := ctx.Value(globalResumeInfoKey{}).(*globalResumeInfo)
	if ok {
		rInfo.mu.Lock()
		defer rInfo.mu.Unlock()

		if rInfo.id2Addr == nil {
			rInfo.id2Addr = make(map[string]Address)
		}
		for id, addr := range id2Addr {
			rInfo.id2Addr[id] = addr
		}
		if rInfo.id2State == nil {
			rInfo.id2State = make(map[string]InterruptState)
		}
		for id, state := range id2State {
			rInfo.id2State[id] = state
		}
	} else {
		rInfo = &globalResumeInfo{
			id2Addr:           id2Addr,
			id2State:          id2State,
			id2StateUsed:      make(map[string]bool),
			id2ResumeDataUsed: make(map[string]bool),
		}
		ctx = context.WithValue(ctx, globalResumeInfoKey{}, rInfo)
	}

	runCtx, ok := getRunCtx(ctx)
	if ok {
		for id_, addr := range id2Addr {
			if addr.Equals(runCtx.addr) {
				if used, ok := rInfo.id2StateUsed[id_]; !ok || !used {
					runCtx.interruptState = generic.PtrOf(rInfo.id2State[id_])
					rInfo.id2StateUsed[id_] = true
				}

				if used, ok := rInfo.id2ResumeDataUsed[id_]; !ok || !used {
					runCtx.isResumeTarget = true
					runCtx.resumeData = rInfo.id2ResumeData[id_]
					rInfo.id2ResumeDataUsed[id_] = true
				}

				break
			}
		}
	}

	return ctx
}

func getResumeInfo(ctx context.Context) (*globalResumeInfo, bool) {
	info, ok := ctx.Value(globalResumeInfoKey{}).(*globalResumeInfo)
	return info, ok
}

type InterruptInfo struct {
	Info        any
	IsRootCause bool
}

func (i *InterruptInfo) String() string {
	if i == nil {
		return ""
	}
	return fmt.Sprintf("interrupt info: Info=%v, IsRootCause=%v", i.Info, i.IsRootCause)
}