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/mcp"
	"agent-native/runtime/promptx"
	"agent-native/runtime/session"
	"agent-native/runtime/skills"
	"agent-native/runtime/subagent"
	"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
	// Search 可选：提供时注册 tavily_search / web_search（密钥由组合根注入）。
	Search tools.SearchProvider
	// Exec 可选：提供时注册 run_command / run_python（解释器留空项不注册）。
	// nil = 不注册执行类工具（fail-closed，与 BashPermission 默认 deny 对齐）。
	Exec *tools.ExecConfig
	// Fetch 可选：提供时注册 fetch_url；nil = 不注册（fail-closed）。
	Fetch *tools.FetchConfig
	// MCP 可选：提供时把各 server 工具以 mcp__<server>__<tool> 注册进工作区。
	// nil = 不注册（fail-closed：server 不可达时整个步骤失败，不静默跳过）。
	MCP *mcp.Manager
	// SkillsRoots 可选：全局技能根目录；执行时自动并入工作区 .agent/skills/。
	SkillsRoots []string
	// Compact 可选：转录超过阈值时把头部历史压成摘要（长会话上下文治理）。
	Compact *CompactConfig
	// Subagent 可选：提供时注册 delegate_task（委派由宿主实现，内核只开端口）。
	Subagent subagent.Factory
	// Log 可选。nil 时丢弃日志。
	Log *logsys.Logger
	// ExecutionID 写入日志，不参与执行控制。
	ExecutionID string
}

// CompactConfig 声明长会话压缩策略（优化性质：压缩失败不阻断执行，但留日志）。
type CompactConfig struct {
	// ThresholdChars 触发阈值（按字节数近似），默认 300_000。
	ThresholdChars int
	// KeepLastN 压缩后保留的最近消息条数，默认 8。
	KeepLastN int
	// Summarizer 可选注入摘要函数；缺省用后端 stream 做一次无工具调用。
	Summarizer func(ctx context.Context, messages []llm.Message) (string, error)
}

// 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" }

// Close 释放后端持有的长连接资源（MCP server 子进程等）。
func (b *Backend) Close() {
	if b.opts.MCP != nil {
		b.opts.MCP.Close()
	}
}

// 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)
	}

	// 技能：全局根 + 工作区 .agent/skills；声明的技能必须存在（fail-closed）。
	roots := make([]string, 0, len(b.opts.SkillsRoots)+1)
	roots = append(roots, b.opts.SkillsRoots...)
	roots = append(roots, filepath.Join(workDir, ".agent", "skills"))
	skillStore := skills.NewStore(roots...)
	skillIndex := skillStore.List()
	indexSet := make(map[string]bool, len(skillIndex))
	for _, s := range skillIndex {
		indexSet[s.Name] = true
	}
	for _, name := range req.Skills {
		if !indexSet[name] {
			return &exec.Result{Success: false, ErrorReason: "declared skill not installed: " + name}, nil
		}
	}

	var reg *tools.Registry
	if profile == exec.ToolProfileWorkspace {
		reg = tools.WorkspaceTools(workDir)
		skills.RegisterLoader(reg, skillStore)
		if b.opts.Search != nil {
			tools.RegisterSearch(reg, b.opts.Search)
		}
		if b.opts.Fetch != nil {
			tools.RegisterFetch(reg, *b.opts.Fetch)
		}
		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 b.opts.MCP != nil {
			if err := b.opts.MCP.RegisterInto(ctx, reg); err != nil {
				return &exec.Result{Success: false, ErrorReason: err.Error()}, nil
			}
		}
		subagent.Register(reg, b.opts.Subagent)
		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
	}
	_ = store.SaveMeta(map[string]string{
		"model":   model,
		"scope":   req.SessionScope,
		"exec_id": req.ExecutionID,
		"updated": time.Now().UTC().Format(time.RFC3339),
	})
	b.maybeCompact(ctx, store, stream, model)

	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, workDir, skillIndex, &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, workDir string, skillIndex []skills.Skill,
	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: b.systemPrompt(req, model, profile, workDir, skillIndex),
		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
}

// maybeCompact 在转录超过阈值时压缩头部历史。压缩是优化性质：
// 失败不阻断执行（按原文继续），但必须留日志，不静默。
func (b *Backend) maybeCompact(ctx context.Context, store *session.Store, stream llm.StreamFunc, model string) {
	cfg := b.opts.Compact
	if cfg == nil {
		return
	}
	threshold := int64(cfg.ThresholdChars)
	if threshold <= 0 {
		threshold = 300_000
	}
	keep := cfg.KeepLastN
	if keep <= 0 {
		keep = 8
	}
	if store.Stats().Bytes < threshold {
		return
	}
	summarize := cfg.Summarizer
	if summarize == nil {
		summarize = defaultSummarizer(stream, model)
	}
	done, err := store.Compact(keep, func(messages []llm.Message) (string, error) {
		return summarize(ctx, messages)
	})
	if err != nil {
		b.logger().Warn("session compaction failed; continuing with full transcript", "error", err.Error())
		return
	}
	if done {
		b.logger().Info("session compacted", "keep_last", keep)
	}
}

// defaultSummarizer 用后端 stream 做一次无工具调用，把历史压成摘要。
func defaultSummarizer(stream llm.StreamFunc, model string) func(ctx context.Context, messages []llm.Message) (string, error) {
	return func(ctx context.Context, messages []llm.Message) (string, error) {
		var b strings.Builder
		for _, m := range messages {
			b.WriteString(string(m.Role))
			b.WriteString(": ")
			b.WriteString(m.Content)
			b.WriteString("\n")
			if b.Len() > 120_000 {
				break
			}
		}
		prompt := "把以下智能体对话历史压缩成摘要：保留全部关键事实、已做决定、文件变更与未决问题；中文；不超过 800 字；只输出摘要正文。\n\n" + b.String()
		var out strings.Builder
		for ev := range stream(ctx, model, []llm.Message{{Role: llm.RoleUser, Content: prompt}}, nil, llm.StreamOptions{}) {
			switch ev.Kind {
			case llm.KindTextDelta:
				out.WriteString(ev.Text)
			case llm.KindError:
				return "", ev.Err
			}
		}
		if strings.TrimSpace(out.String()) == "" {
			return "", errString("summarizer produced empty output")
		}
		return out.String(), nil
	}
}

type errString string

func (e errString) Error() string { return string(e) }

// 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 经 promptx 组装：角色指令 + 环境事实 + 技能索引 + 交付物约束。
func (b *Backend) systemPrompt(req exec.Request, model, profile, workDir string, skillIndex []skills.Skill) string {
	var index []promptx.IndexEntry
	for _, s := range skillIndex {
		index = append(index, promptx.IndexEntry{Name: s.Name, Description: s.Description})
	}
	now := time.Now()
	weekday := map[string]string{
		"Sunday": "周日", "Monday": "周一", "Tuesday": "周二", "Wednesday": "周三",
		"Thursday": "周四", "Friday": "周五", "Saturday": "周六",
	}[now.Weekday().String()]
	return promptx.Compose(promptx.Parts{
		Instructions:   req.Instructions,
		SkillsDeclared: req.Skills,
		SkillsIndex:    index,
		Deliverables:   req.Targets,
		Env: promptx.EnvFacts{
			Date:          now.Format("2006-01-02") + "（" + weekday + "）",
			Model:         model,
			WorkspaceRoot: workDir,
			Profile:       profile,
		},
	})
}