Agent-Native/adapters/native/backend.go

// Package native 把 agent-native 自研执行内核（runtime）适配为 exec.Backend，
// 规范后端名 "agent-native"。来源：主仓 internal/agent/pinative.go 的通用化
// 副本（去除产品 config/workspace 依赖）；主仓文件仍是产品路径权威，
// 本文件为模块原生新增，语义变更需两侧同步评审。
package native

import (
	"context"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"agent-native/exec"
	"agent-native/runtime"
	"agent-native/runtime/llm"
	"agent-native/runtime/session"
	"agent-native/runtime/tools"

	"logging-system/logsys"
)

// Options 构造后端所需的模型接入与执行参数（由组合根注入，无产品依赖）。
type Options struct {
	APIBase  string
	APIKey   string
	Model    string
	MaxTurns int
	Timeout  time.Duration
	// Stream 测试注入的流函数；缺省按 APIBase/APIKey 构造 llm.Client。
	Stream llm.StreamFunc
	// Exec 可选：提供时注册 run_command / run_python（解释器留空项不注册）。
	// nil = 不注册执行类工具（fail-closed，与 BashPermission 默认 deny 对齐）。
	Exec *tools.ExecConfig
	// Log 可选。nil 时丢弃日志。
	Log *logsys.Logger
	// ExecutionID 写入日志，不参与执行控制。
	ExecutionID string
}

// Backend 以自研内核实现 exec.Backend。
type Backend struct {
	opts Options
	// lastFailure 记录各工作目录上次失败原因，用于会话毒化检测
	// （context_overflow 等错误续跑同一会话只会复现失败，须换新会话）。
	mu          sync.Mutex
	lastFailure map[string]string
}

// New 创建 native 后端。
func New(opts Options) *Backend {
	return &Backend{opts: opts, lastFailure: map[string]string{}}
}

// Name 返回规范后端名。
func (b *Backend) Name() string { return "agent-native" }

// Execute 运行一轮带工具的自研智能体循环，产出目标交付物。
func (b *Backend) Execute(ctx context.Context, req exec.Request) (*exec.Result, error) {
	if len(req.Targets) == 0 {
		req.Targets = []string{"output.md"}
	}
	profile := strings.TrimSpace(req.ToolProfile)
	if profile == "" {
		profile = exec.ToolProfileWorkspace
	}
	if profile != exec.ToolProfileWorkspace && profile != exec.ToolProfileReasoning {
		return &exec.Result{Success: false, ErrorReason: "unknown tool profile: " + profile}, nil
	}
	if b.opts.Stream == nil && (strings.TrimSpace(b.opts.APIBase) == "" || strings.TrimSpace(b.opts.APIKey) == "") {
		return &exec.Result{Success: false, ErrorReason: "agent-native backend requires api base/key (or injected stream)"}, nil
	}
	model := strings.TrimSpace(b.opts.Model)
	if strings.TrimSpace(req.Model) != "" {
		model = strings.TrimSpace(req.Model)
	}
	if model == "" {
		return &exec.Result{Success: false, ErrorReason: "agent-native backend requires model"}, nil
	}

	workDir := req.WorkDir
	if !filepath.IsAbs(workDir) {
		wd, _ := os.Getwd()
		workDir = filepath.Join(wd, workDir)
	}
	if err := os.MkdirAll(workDir, 0o755); err != nil {
		return &exec.Result{Success: false, ErrorReason: err.Error()}, nil
	}
	if strings.TrimSpace(req.Instructions) != "" {
		_ = os.WriteFile(filepath.Join(workDir, "INSTRUCTIONS.md"), []byte(req.Instructions), 0o644)
		_ = os.WriteFile(filepath.Join(workDir, "AGENTS.md"), []byte(req.Instructions), 0o644)
	}
	if strings.TrimSpace(req.Prompt) != "" {
		_ = os.WriteFile(filepath.Join(workDir, "TASK.md"), []byte(req.Prompt), 0o644)
	}

	var reg *tools.Registry
	if profile == exec.ToolProfileWorkspace {
		reg = tools.WorkspaceTools(workDir)
		if b.opts.Exec != nil {
			execCfg := *b.opts.Exec
			for k, v := range req.Env {
				execCfg.Env = append(execCfg.Env, k+"="+v)
			}
			tools.RegisterExecTools(reg, workDir, execCfg)
		}
		if err := reg.Restrict(req.ToolAllowlist); err != nil {
			return &exec.Result{Success: false, ErrorReason: err.Error()}, nil
		}
	}
	// 会话毒化轮转：上次失败属 resume-unsafe 时，把旧 transcript 归档，
	// 本次以全新会话开始（续跑毒化会话只会复现同样的失败）。
	b.rotateIfPoisoned(workDir)
	store, err := session.Open(filepath.Join(workDir, ".agentx", "transcript.jsonl"))
	if err != nil {
		return &exec.Result{Success: false, ErrorReason: err.Error()}, nil
	}
	stream := b.opts.Stream
	if stream == nil {
		client := &llm.Client{BaseURL: b.opts.APIBase, APIKey: b.opts.APIKey}
		stream = client.Stream
	}

	timeout := req.Timeout
	if timeout <= 0 {
		timeout = b.opts.Timeout
	}
	var runBudget *runtime.Budget
	if req.Budget != nil {
		runBudget = &runtime.Budget{
			MaxPromptTokens:     req.Budget.MaxPromptTokens,
			MaxCompletionTokens: req.Budget.MaxCompletionTokens,
			MaxToolCalls:        req.Budget.MaxToolCalls,
			Deadline:            req.Budget.Deadline,
		}
		if !req.Budget.Deadline.IsZero() {
			// Deadline 与 Timeout 先到者生效：把更短者作为 ctx 超时。
			if until := time.Until(req.Budget.Deadline); timeout <= 0 || until < timeout {
				timeout = until
			}
		}
	}
	runCtx := ctx
	cancel := func() {}
	if timeout > 0 {
		runCtx, cancel = context.WithTimeout(ctx, timeout)
	}
	defer cancel()

	execID := req.ExecutionID
	if strings.TrimSpace(execID) == "" {
		execID = b.opts.ExecutionID
	}
	var logs strings.Builder
	if strings.TrimSpace(execID) != "" || strings.TrimSpace(req.SessionScope) != "" {
		logs.WriteString("execution_id=" + execID + " session=" + req.SessionScope + "\n")
	}
	onProcess := req.OnEvent
	runRes, err := b.run(runCtx, model, runBudget, reg, store, stream, profile, req, &logs, onProcess)
	if err != nil {
		return &exec.Result{Success: false, ErrorReason: err.Error(), Logs: logs.String()}, nil
	}
	_ = os.WriteFile(filepath.Join(workDir, "execution.log"), []byte(logs.String()), 0o644)

	// reasoning 档无工具：把最终文本写入仍为空的目标文件。
	if profile == exec.ToolProfileReasoning && strings.TrimSpace(runRes.Text) != "" {
		for _, target := range req.Targets {
			path := filepath.Join(workDir, filepath.FromSlash(target))
			if b2, err := os.ReadFile(path); err == nil && len(strings.TrimSpace(string(b2))) > 0 {
				continue
			}
			_ = os.MkdirAll(filepath.Dir(path), 0o755)
			_ = os.WriteFile(path, []byte(runRes.Text), 0o644)
		}
	}

	if runRes.Err != nil {
		reason := runRes.Err.Error()
		b.mu.Lock()
		b.lastFailure[workDir] = reason
		b.mu.Unlock()
		assessment := exec.AssessStepFailure(exec.StepFailureSignals{
			Err:     runRes.Err,
			WorkDir: workDir,
			Result:  &exec.Result{ErrorReason: reason, Retryable: runRes.Retryable, Logs: logs.String()},
		})
		b.logger().Error("native step failure classified", runRes.Err,
			"failure_class", assessment.Class.String(),
			"work_dir", workDir)
		res := &exec.Result{Success: false, ErrorReason: reason, Logs: logs.String(), Retryable: runRes.Retryable}
		if runRes.StopReason == llm.StopAborted {
			res.Retryable = true
		}
		return res, nil
	}
	b.mu.Lock()
	delete(b.lastFailure, workDir)
	b.mu.Unlock()

	patches, matErr := exec.MaterializeTargets(workDir, req.Targets, nil)
	if matErr != nil {
		return &exec.Result{Success: false, ErrorReason: matErr.Error(), Patches: patches, Logs: logs.String()}, nil
	}
	return &exec.Result{
		Success:          true,
		Patches:          patches,
		Logs:             logs.String(),
		PromptTokens:     runRes.Usage.PromptTokens,
		CompletionTokens: runRes.Usage.CompletionTokens,
	}, nil
}

// run 驱动 runtime 循环并桥接过程事件（语义对齐主仓 pinative）。
func (b *Backend) run(ctx context.Context, model string, budget *runtime.Budget, reg *tools.Registry, store *session.Store,
	stream llm.StreamFunc, profile string, req exec.Request, logs *strings.Builder,
	onProcess func(exec.ProcessEvent)) (runtime.Result, error) {

	execID := req.ExecutionID
	if strings.TrimSpace(execID) == "" {
		execID = b.opts.ExecutionID
	}
	runRes, err := runtime.Run(ctx, runtime.Config{
		Model:        model,
		SystemPrompt: systemPrompt(req, profile),
		Tools:        reg,
		Stream:       stream,
		Session:      store,
		MaxTurns:     b.opts.MaxTurns,
		Log:          b.logger(),
		ExecutionID:  execID,
		Budget:       budget,
		OnEvent: func(ev runtime.Event) {
			logs.WriteString(ev.Type)
			if ev.Tool != "" {
				logs.WriteString(" ")
				logs.WriteString(ev.Tool)
			}
			if ev.Text != "" {
				logs.WriteString(" ")
				logs.WriteString(ev.Text)
			}
			logs.WriteByte('\n')
			if onProcess != nil {
				onProcess(exec.ProcessEvent{
					Type:   ev.Type,
					Text:   ev.Text,
					Tool:   ev.Tool,
					Detail: ev.Detail,
				})
			}
		},
	}, llm.Message{Role: llm.RoleUser, Content: req.Prompt})
	return runRes, err
}

// resumeUnsafeMarkers 续跑不安全的关键词（会话已被污染：上下文溢出/
// 迭代上限/非法请求——重试必须换新会话而非续跑）。
var resumeUnsafeMarkers = []string{
	"context_overflow", "context length", "context window", "too long",
	"iteration_limit", "max turns", "turn cap",
	"invalid_request", "invalid request", "bad request", "400",
}

// rotateIfPoisoned 上次失败属 resume-unsafe 且存在旧会话时归档旧会话。
func (b *Backend) rotateIfPoisoned(workDir string) {
	b.mu.Lock()
	reason := strings.ToLower(b.lastFailure[workDir])
	b.mu.Unlock()
	if reason == "" {
		return
	}
	unsafe := false
	for _, m := range resumeUnsafeMarkers {
		if strings.Contains(reason, m) {
			unsafe = true
			break
		}
	}
	if !unsafe {
		return
	}
	path := filepath.Join(workDir, ".agentx", "transcript.jsonl")
	if _, err := os.Stat(path); err != nil {
		return
	}
	b.logger().Warn("rotating poisoned session", "work_dir", workDir)
	_ = os.Rename(path, filepath.Join(workDir, ".agentx",
		"transcript.poisoned."+time.Now().UTC().Format("20060102T150405.000000000")+".jsonl"))
}

func (b *Backend) logger() *logsys.Logger {
	if b == nil || b.opts.Log == nil {
		return logsys.Nop()
	}
	log := b.opts.Log.Module("native")
	if b.opts.ExecutionID == "" {
		return log
	}
	return log.With("execution_id", b.opts.ExecutionID)
}

// systemPrompt 组装系统提示：角色指令 + 交付物与工具档位约束 + 声明的技能。
func systemPrompt(req exec.Request, profile string) string {
	var b strings.Builder
	if strings.TrimSpace(req.Instructions) != "" {
		b.WriteString(req.Instructions)
		b.WriteString("\n\n")
	}
	if len(req.Skills) > 0 {
		b.WriteString("Declared skills for this task: ")
		b.WriteString(strings.Join(req.Skills, ", "))
		b.WriteString(".\n\n")
	}
	b.WriteString("Deliverable files: ")
	b.WriteString(strings.Join(req.Targets, ", "))
	b.WriteString(".")
	if profile == exec.ToolProfileWorkspace {
		b.WriteString(" Use workspace tools to write those files. Do not invent paths outside the workspace.")
	} else {
		b.WriteString(" Reply with the deliverable text only. Tools are disabled.")
	}
	return b.String()
}