Agent-Native/runtime/codeorch/host.go

// Package codeorch 在会话目录里跑一段 Python，并按行把工具调用交回宿主。
// 工作流脚本和 run_tool_script 共用这一套进程与限额。
package codeorch

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"time"
	"unicode/utf8"

	_ "embed"
)

//go:embed agentwf.py
var agentwfSource string

const (
	defaultTimeout = 120 * time.Second
	defaultMaxWait = 600 * time.Second
	defaultCalls   = 50
	defaultResult  = 1 << 20
	defaultModel   = 16000
)

var (
	helperOnce sync.Once
	helperDir  string
	helperErr  error
)

// Call 是脚本发来的一行请求。Args 与 Choices 保持原始 JSON。
type Call struct {
	ID       string          `json:"id"`
	Op       string          `json:"op"`
	Name     string          `json:"name"`
	Args     json.RawMessage `json:"args"`
	Question string          `json:"question"`
	Choices  json.RawMessage `json:"choices"`
	Agent    string          `json:"agent"`
	Task     string          `json:"task"`
	Content  string          `json:"content"`
	Title    string          `json:"title"`
	Kind     string          `json:"kind"`
	Status   string          `json:"status"`
	StepID   string          `json:"stepId"`
}

// Limits 控制一次脚本运行。零值使用工具路径的缺省：120 秒、最长 600 秒、50 次、1 MiB、16k 字符。
type Limits struct {
	Timeout        time.Duration
	MaxTimeout     time.Duration
	MaxCalls       int
	MaxResultBytes int
	MaxModelChars  int
	FullDir        string
}

// Outcome 是给模型的文本、已成功的调用，以及超限时的全文路径。
type Outcome struct {
	Output   string
	FullPath string
	Calls    []string
	Err      error
}

// Starter 按工作目录和脚本路径准备命令。空则用 python3。
type Starter func(ctx context.Context, workDir, path string) (*exec.Cmd, error)

// Dispatch 处理一条回调。返回的错误会写回脚本并记入最终结果。
type Dispatch func(ctx context.Context, call Call) (string, error)

// ModelText 把截断后的输出、已执行调用和错误拼成给模型的文本。
func (o Outcome) ModelText() string {
	text := o.Output
	if o.Err != nil {
		msg := o.Err.Error()
		if len(o.Calls) > 0 {
			msg = "已执行：" + strings.Join(o.Calls, "、") + "\n" + msg
		}
		if text != "" {
			text += "\n"
		}
		text += msg
	}
	if o.FullPath != "" {
		if text != "" {
			text += "\n"
		}
		text += "全文在 " + o.FullPath
	}
	return text
}

// RunFile 执行已经落盘的脚本。
func RunFile(ctx context.Context, workDir, path string, lim Limits, start Starter, dispatch Dispatch) Outcome {
	if ctx == nil {
		ctx = context.Background()
	}
	timeout := lim.bounded()
	runCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	helper, err := helperPath()
	if err != nil {
		return Outcome{Err: err}
	}
	if start == nil {
		start = defaultStart
	}
	cmd, err := start(runCtx, workDir, path)
	if err != nil {
		return Outcome{Err: err}
	}
	cmd.Env = prependPythonPath(cmd.Env, helper)
	reqR, reqW, err := os.Pipe()
	if err != nil {
		return Outcome{Err: err}
	}
	respR, respW, err := os.Pipe()
	if err != nil {
		reqR.Close()
		reqW.Close()
		return Outcome{Err: err}
	}
	cmd.ExtraFiles = []*os.File{reqW, respR}
	var stdout, stderr bytes.Buffer
	cmd.Stdout = &stdout
	cmd.Stderr = &stderr
	if err := cmd.Start(); err != nil {
		reqR.Close()
		reqW.Close()
		respR.Close()
		respW.Close()
		return Outcome{Err: err}
	}
	reqW.Close()
	respR.Close()
	defer reqR.Close()
	defer respW.Close()
	calls, callErr := serve(runCtx, reqR, respW, lim, dispatch)
	if callErr != nil {
		cancel()
	}
	waitErr := cmd.Wait()
	out := Outcome{Output: clipJoined(&stdout, &stderr), Calls: calls}
	out = clipModel(out, lim)
	switch {
	case callErr != nil:
		out.Err = callErr
	case runCtx.Err() == context.DeadlineExceeded:
		out.Err = fmt.Errorf("脚本超时")
	case runCtx.Err() == context.Canceled:
		out.Err = fmt.Errorf("脚本已取消")
	case waitErr != nil:
		out.Err = fmt.Errorf("脚本退出：%v", waitErr)
	}
	return out
}

// Run 把脚本体写入工作目录后再执行。
func Run(ctx context.Context, workDir, body string, lim Limits, start Starter, dispatch Dispatch) Outcome {
	dir := filepath.Join(workDir, ".agent")
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return Outcome{Err: err}
	}
	path := filepath.Join(dir, fmt.Sprintf("orch-%d.py", time.Now().UnixNano()))
	if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
		return Outcome{Err: err}
	}
	return RunFile(ctx, workDir, path, lim, start, dispatch)
}

func (l Limits) bounded() time.Duration {
	d := l.Timeout
	if d <= 0 {
		d = defaultTimeout
	}
	max := l.MaxTimeout
	if max <= 0 {
		max = defaultMaxWait
	}
	if d > max {
		return max
	}
	return d
}

func (l Limits) calls() int {
	if l.MaxCalls <= 0 {
		return defaultCalls
	}
	return l.MaxCalls
}

func (l Limits) resultBytes() int {
	if l.MaxResultBytes <= 0 {
		return defaultResult
	}
	return l.MaxResultBytes
}

func (l Limits) modelChars() int {
	if l.MaxModelChars <= 0 {
		return defaultModel
	}
	return l.MaxModelChars
}

func serve(ctx context.Context, reqR, respW *os.File, lim Limits, dispatch Dispatch) ([]string, error) {
	sc := bufio.NewScanner(reqR)
	sc.Buffer(make([]byte, 64*1024), lim.resultBytes())
	var calls []string
	n := 0
	for sc.Scan() {
		line := strings.TrimSpace(sc.Text())
		if line == "" {
			continue
		}
		n++
		var call Call
		if err := json.Unmarshal([]byte(line), &call); err != nil {
			_ = writeResp(respW, call.ID, false, "", "请求不是合法 JSON")
			continue
		}
		if n > lim.calls() {
			_ = writeResp(respW, call.ID, false, "", "脚本回调超过 50 次")
			return calls, fmt.Errorf("脚本回调超过 50 次")
		}
		if ctx.Err() != nil {
			return calls, ctx.Err()
		}
		if dispatch == nil {
			_ = writeResp(respW, call.ID, false, "", "没有回调处理")
			continue
		}
		output, err := dispatch(ctx, call)
		if err != nil {
			_ = writeResp(respW, call.ID, false, "", err.Error())
			if ctx.Err() != nil {
				return calls, ctx.Err()
			}
			continue
		}
		if call.Op == "tool.call" && call.Name != "" {
			calls = append(calls, call.Name)
		}
		if err := writeResp(respW, call.ID, true, clipBytes(output, lim.resultBytes()), ""); err != nil {
			return calls, err
		}
	}
	if err := sc.Err(); err != nil {
		if err == bufio.ErrTooLong {
			return calls, fmt.Errorf("单行请求超过 1MB")
		}
		if ctx.Err() != nil {
			return calls, ctx.Err()
		}
		return calls, err
	}
	return calls, nil
}

func writeResp(w io.Writer, id string, ok bool, output, errText string) error {
	raw, err := json.Marshal(struct {
		ID     string `json:"id"`
		OK     bool   `json:"ok"`
		Output string `json:"output,omitempty"`
		Error  string `json:"error,omitempty"`
	}{ID: id, OK: ok, Output: output, Error: errText})
	if err != nil {
		return err
	}
	_, err = w.Write(append(raw, '\n'))
	return err
}

func clipBytes(s string, n int) string {
	if n <= 0 || len(s) <= n {
		return s
	}
	cut := s[:n]
	for !utf8.ValidString(cut) && len(cut) > 0 {
		cut = cut[:len(cut)-1]
	}
	return cut
}

func clipJoined(stdout, stderr *bytes.Buffer) string {
	text := strings.TrimSpace(stdout.String())
	errText := strings.TrimSpace(stderr.String())
	if errText != "" {
		if text != "" {
			text += "\n"
		}
		text += errText
	}
	return text
}

func clipModel(out Outcome, lim Limits) Outcome {
	limit := lim.modelChars()
	if utf8.RuneCountInString(out.Output) <= limit {
		return out
	}
	dir := lim.FullDir
	if dir == "" {
		dir = os.TempDir()
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		out.Output = clipRunes(out.Output, limit)
		return out
	}
	path := filepath.Join(dir, fmt.Sprintf("script-output-%d.txt", time.Now().UnixNano()))
	if err := os.WriteFile(path, []byte(out.Output), 0o644); err != nil {
		out.Output = clipRunes(out.Output, limit)
		return out
	}
	out.FullPath = path
	out.Output = clipRunes(out.Output, limit)
	return out
}

func clipRunes(s string, n int) string {
	if utf8.RuneCountInString(s) <= n {
		return s
	}
	i := 0
	for count := 0; count < n && i < len(s); count++ {
		_, size := utf8.DecodeRuneInString(s[i:])
		i += size
	}
	return s[:i] + "…（超长截断）"
}

func defaultStart(ctx context.Context, workDir, path string) (*exec.Cmd, error) {
	bin, err := exec.LookPath("python3")
	if err != nil {
		return nil, fmt.Errorf("找不到 python3: %w", err)
	}
	cmd := exec.CommandContext(ctx, bin, path)
	cmd.Dir = workDir
	return cmd, nil
}

func helperPath() (string, error) {
	helperOnce.Do(func() {
		dir, err := os.MkdirTemp("", "agentwf-*")
		if err != nil {
			helperErr = err
			return
		}
		if err := os.WriteFile(filepath.Join(dir, "agentwf.py"), []byte(agentwfSource), 0o644); err != nil {
			helperErr = err
			return
		}
		helperDir = dir
	})
	return helperDir, helperErr
}

func prependPythonPath(env []string, dir string) []string {
	if env == nil {
		env = os.Environ()
	}
	for i, kv := range env {
		if strings.HasPrefix(kv, "PYTHONPATH=") {
			env[i] = "PYTHONPATH=" + dir + string(os.PathListSeparator) + strings.TrimPrefix(kv, "PYTHONPATH=")
			return env
		}
	}
	return append(env, "PYTHONPATH="+dir)
}