Agent-System-Backend/internal/runmgr/runmgr.go

// Package runmgr 是 Run 生命周期枢纽（4.2/4.3）：一切运行皆 Run、一切增量皆事件。
// 职责：run 状态机、事件持久化+广播（seq 游标）、HITL 挂起/恢复（三分支路由）、
// 子代理派生（任务树）、工具执行策略（画像三清单 + 确认卡片 + 截断转工件）、取消级联。
package runmgr

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"agent-system-backend/internal/bus"
	"agent-system-backend/internal/cardsvc"
	"agent-system-backend/internal/ingest"
	"agent-system-backend/internal/profiles"
	"agent-system-backend/internal/runtime"
	"agent-system-backend/internal/settings"
	"agent-system-backend/internal/store"
	"agent-system-backend/internal/tools"
	core "agent-system-core"
)

type Manager struct {
	st        *store.Store
	bus       *bus.Bus
	registry  *tools.Registry
	profiles  *profiles.Service
	settings  *settings.Service
	runner    runtime.Runner // conversation runner（mock 或 LLM）
	subRunner runtime.Runner // subagent runner（LLM 轻任务）
	mock      runtime.Runner // 会话脚本回放（演示/测试与 mock 模式）
	mockSub   runtime.Runner // 子代理脚本回放
	gate      IntentGate
	cards     *cardsvc.Service     // 可选意图闸门（nil = 未启用，行为回退）
	events    *eventBuffer         // 事件批处理（写放大治理；seq 权威仍在 store）
	retriever core.Retriever       // 可选混合检索（nil = store 全文）
	indexer   core.DocumentIndexer // 可选索引入口（nil = store 整文档单块）
	// workspaceRoot fs 预留能力（Hidden 工具族）的沙箱根；空=未配置，工具执行时报错。
	workspaceRoot string

	mu     sync.Mutex
	active map[string]*runHandle

	// 会话内串行（设计并发策略）：一个会话同时至多一个在途 conversation run。
	convMu     sync.Mutex
	convActive map[string]bool // conversationID → 有在途 run
}

// ErrConversationBusy 表示会话已有在途 run（HTTP 层映射 409）。
var ErrConversationBusy = fmt.Errorf("runmgr: conversation already has an in-flight run")

type runHandle struct {
	cancel         context.CancelFunc
	hitlCh         chan hitlAnswer
	assistant      []core.Part
	ctxObs         *core.ContextObservation
	submittedCards []core.ID // submit_card 产物卡（finishRun 回填锚点）
	streamCardID   core.ID   // 活跃流式产物卡（card_begin/card_commit 三段式；D44/D45）
	started        time.Time
}

type hitlAnswer struct {
	resp core.HITLResponse
	err  error
}

func New(st *store.Store, b *bus.Bus, reg *tools.Registry, p *profiles.Service, s *settings.Service, runner, subRunner, mockRunner, mockSubRunner runtime.Runner) *Manager {
	return &Manager{
		st: st, bus: b, registry: reg, profiles: p, settings: s,
		runner: runner, subRunner: subRunner, mock: mockRunner, mockSub: mockSubRunner,
		active: map[string]*runHandle{}, convActive: map[string]bool{},
		events: newEventBuffer(st, b),
	}
}

// Shutdown 停机清空事件缓冲（在 store.Close 之前调用）。
func (m *Manager) Shutdown() { m.events.stop() }

// SetEventNotifier 注入落盘后回调（PG 档 pgbus.Notifier；SQLite 档不注入）。
func (m *Manager) SetEventNotifier(fn func(conversationID string, evs []core.StreamEvent)) {
	m.events.onPersisted = fn
}

// IntentGate 是可选意图闸门端口（intentgate.Gate 实现；nil = 未启用，行为回退）。
type IntentGate interface {
	Match(ctx context.Context, input string) (branch, topID string)
}

// SetGate 注入意图闸门（装配期一次；插件意图 overlay 经 pluginhost 回调重建）。
func (m *Manager) SetGate(g IntentGate) { m.gate = g }

// SetCards 注入收件箱卡服务（nil = 不产卡；T0 退回基线仍成立）。
func (m *Manager) SetCards(s *cardsvc.Service) { m.cards = s }

// EmitCardEvent 供服务层（cardsvc 投影）复用会话事件管道（持久化 + seq + 扇出）。
func (m *Manager) EmitCardEvent(conversationID, runID string, typ core.StreamEventType, payload any) error {
	_, err := m.emit(conversationID, runID, typ, payload)
	return err
}

// SetRetriever 注入混合检索（rag 管线；nil = store 全文检索）。
func (m *Manager) SetRetriever(r core.Retriever) { m.retriever = r }

// SetIndexer 注入检索索引入口（分块+全文+可选向量；nil = 整文档单块）。
func (m *Manager) SetIndexer(i core.DocumentIndexer) { m.indexer = i }

// SetWorkspaceRoot 注入 fs 预留能力（Hidden 工具族）的沙箱根目录。
// 实际会话沙箱为 <root>/<owner>/<conversationID>/；空值=未配置（工具执行报错）。
func (m *Manager) SetWorkspaceRoot(dir string) { m.workspaceRoot = dir }

// ---- runCtx 实现（runtime.RunCtx）----

type runCtx struct {
	mgr         *Manager
	run         core.Run
	prof        core.AgentProfile
	eff         core.EffectiveSettings
	ctx         context.Context
	h           *runHandle
	userMessage core.Message
	attachments []core.ParsedDocument
	inputPaths  []string
	subTask     string
	subHint     string
}

func (c *runCtx) Context() context.Context         { return c.ctx }
func (c *runCtx) Run() core.Run                    { return c.run }
func (c *runCtx) Profile() core.AgentProfile       { return c.prof }
func (c *runCtx) Settings() core.EffectiveSettings { return c.eff }
func (c *runCtx) History() []core.Message {
	hist, _ := c.mgr.st.LastMessages(c.run.ConversationID, 200)
	return hist
}
func (c *runCtx) Attachments() []core.ParsedDocument { return c.attachments }
func (c *runCtx) WorkspaceInputs() []string          { return c.inputPaths }
func (c *runCtx) Subtask() string                    { return c.subTask }

// Tools 模型可见工具清单：Hidden 工具（fs 预留能力）在 runtime.fsTools 未开时过滤
// （与 /api/tools 列表同门禁；registry.Execute 另有执行侧 fail-closed）。
func (c *runCtx) Tools() []core.ToolDesc {
	list := c.mgr.registry.List()
	if c.eff.GetBool("runtime.fsTools", false) {
		return list
	}
	out := make([]core.ToolDesc, 0, len(list))
	for _, t := range list {
		if !t.Hidden {
			out = append(out, t)
		}
	}
	return out
}
func (c *runCtx) SetContextObservation(o core.ContextObservation) {
	c.mgr.mu.Lock()
	c.h.ctxObs = &o
	c.mgr.mu.Unlock()
}

func (c *runCtx) Emit(typ core.StreamEventType, payload any) error {
	_, err := c.mgr.emit(c.run.ConversationID, c.run.ID, typ, payload)
	return err
}

func (c *runCtx) SetAssistantParts(parts []core.Part) error {
	c.mgr.mu.Lock()
	c.h.assistant = parts
	c.mgr.mu.Unlock()
	return nil
}

func (c *runCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) {
	card.ID = store.NewID("hitl")
	card.RunID = c.run.ID
	card.ConversationID = c.run.ConversationID
	if card.CreatedAt == "" {
		card.CreatedAt = time.Now().UTC().Format(time.RFC3339)
	}
	if err := c.mgr.st.SaveHITLCard(c.run.Owner, &card); err != nil {
		return core.HITLResponse{}, err
	}
	c.mgr.setRunStatus(&c.run, core.RunWaitingHITL, c.run.Phase)
	if _, err := c.mgr.emit(c.run.ConversationID, c.run.ID, core.EvHITLRequest, card); err != nil {
		return core.HITLResponse{}, err
	}
	// 收件箱投影卡（双现）：待确认进收件箱不沉没；ref 联动使任一侧响应即双侧销项。
	if c.mgr.cards != nil {
		proj := core.ActivityCard{
			ProjectID: c.run.ProjectID, ConversationID: c.run.ConversationID, SourceRunID: c.run.ID,
			Kind: core.CardKindConfirm, Title: card.Question, Status: core.CardActionRequired,
			Tone: map[bool]string{true: "danger", false: "warn"}[card.Blocked],
		}
		if card.Blocked {
			proj.Body = append(proj.Body, core.UINode{Type: core.UIMarkdown, Markdown: "⚠ 阻断：" + card.BlockReason})
		}
		for _, ch := range card.Choices {
			proj.Actions = append(proj.Actions, core.UIAction{ID: ch.ID, Label: ch.Label, Style: map[bool]string{true: "primary", false: "default"}[ch.Recommended]})
		}
		if card.AllowFreeInput {
			proj.Body = append(proj.Body, core.UINode{Type: core.UIMarkdown, Markdown: "_可在输入框自由输入回应。_"})
		}
		if _, err := c.mgr.cards.Create(c.run.Owner, proj, "hitl:"+card.ID); err != nil {
			return core.HITLResponse{}, err
		}
	}
	select {
	case ans := <-c.h.hitlCh:
		return ans.resp, ans.err
	case <-c.ctx.Done():
		return core.HITLResponse{}, c.ctx.Err()
	}
}

func (c *runCtx) SpawnSubagent(name, task, hint string) (core.AgentResultPayload, error) {
	return c.mgr.spawnSubagent(c.ctx, c.run, name, task, hint)
}

func (c *runCtx) UpdatePlan(p core.PlanSnapshot) error {
	p.PlanID = c.run.ConversationID
	if err := c.mgr.st.SavePlan(c.run.ConversationID, p); err != nil {
		return err
	}
	_, err := c.mgr.emit(c.run.ConversationID, c.run.ID, core.EvPlanUpdate, core.PlanUpdatePayload{Plan: p})
	return err
}

func (c *runCtx) UpdateTodos(l core.TodoList) error {
	l.ConversationID = c.run.ConversationID
	if err := c.mgr.st.SaveTodos(c.run.ConversationID, l); err != nil {
		return err
	}
	_, err := c.mgr.emit(c.run.ConversationID, c.run.ID, core.EvTodoUpdate, core.TodoUpdatePayload{List: l})
	return err
}

// CreateCard / SetCardStatus：run 内产卡入口（摄取/编排/工作流步骤；mock 演示）。
// 收件箱关闭（cards==nil）时为 no-op——T0 退回基线不受影响。
// 归属判定（D48）：调用方留空 ConversationID 且预置 ProjectID = 项目级卡
// （scope=project 工具参数），补全 ProjectID、ConversationID 保持空；其余照旧
// 归属当前会话。
func (c *runCtx) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
	if c.mgr.cards == nil {
		return card, nil
	}
	if card.ConversationID == "" && card.ProjectID == "" {
		card.ConversationID = c.run.ConversationID
	}
	card.ProjectID = c.run.ProjectID
	card.SourceRunID = c.run.ID
	return c.mgr.cards.Create(c.run.Owner, card, "")
}

func (c *runCtx) SetCardStatus(id string, status core.CardStatus, tone, note string) error {
	if c.mgr.cards == nil {
		return nil
	}
	_, err := c.mgr.cards.SetStatus(c.run.Owner, id, status, tone, note)
	return err
}

// SubmitCard：submit_card 工具入口——建卡并登记 ID，终态消息落库后回填锚点。
func (c *runCtx) SubmitCard(card core.ActivityCard) (core.ActivityCard, error) {
	created, err := c.CreateCard(card)
	if err != nil {
		return created, err
	}
	c.mgr.mu.Lock()
	c.h.submittedCards = append(c.h.submittedCards, created.ID)
	c.mgr.mu.Unlock()
	return created, nil
}

// BeginStreamCard：card_begin 工具入口——建 active 产物卡并设为活跃流式卡。
// 多次 begin 以最后一卡为活跃（提案 §4）；未收尾的旧卡由 finishRun 兜底置态。
// 收件箱关闭（cards==nil 或返回空 ID）时报错，工具层回退为对话内直写。
func (c *runCtx) BeginStreamCard(card core.ActivityCard) (core.ActivityCard, error) {
	card.Status = core.CardActive
	created, err := c.CreateCard(card)
	if err != nil {
		return created, err
	}
	if created.ID == "" {
		return created, errors.New("card: inbox disabled")
	}
	c.mgr.mu.Lock()
	c.h.streamCardID = created.ID
	c.mgr.mu.Unlock()
	return created, nil
}

// AppendStreamCard：card_append 工具入口——向活跃流式卡追加正文（事件节流）。
func (c *runCtx) AppendStreamCard(text string) error {
	c.mgr.mu.Lock()
	id := c.h.streamCardID
	c.mgr.mu.Unlock()
	if id == "" {
		return errors.New("card: no active stream card (card_begin first)")
	}
	return c.mgr.cards.Append(c.run.Owner, string(id), text)
}

// CommitStreamCard：card_commit 工具入口——活跃流式卡置 done、登记锚点回填
// 并清空活跃指针。
func (c *runCtx) CommitStreamCard() (core.ActivityCard, error) {
	c.mgr.mu.Lock()
	id := c.h.streamCardID
	c.h.streamCardID = ""
	c.mgr.mu.Unlock()
	if id == "" {
		return core.ActivityCard{}, errors.New("card: no active stream card (card_begin first)")
	}
	card, err := c.mgr.cards.SetStatus(c.run.Owner, string(id), core.CardDone, "ok", "")
	if err != nil {
		return card, err
	}
	c.mgr.mu.Lock()
	c.h.submittedCards = append(c.h.submittedCards, card.ID)
	c.mgr.mu.Unlock()
	return card, nil
}

// Card：按 ID 取卡（ctxasm 回填；无此卡或收件箱关闭返回 nil）。
func (c *runCtx) Card(id string) *core.ActivityCard {
	if c.mgr.cards == nil {
		return nil
	}
	card, err := c.mgr.cards.Get(c.run.Owner, id)
	if err != nil {
		return nil
	}
	return &card
}

// Artifact：按 ID 取工件与内容（ctxasm 回填 artifact_ref 正文用；跨租户/不存在
// 一律 false，不泄露存在性）。
func (c *runCtx) Artifact(id string) (core.Artifact, []byte, bool) {
	art, err := c.mgr.st.GetArtifact(c.run.Owner, string(id))
	if err != nil {
		return core.Artifact{}, nil, false
	}
	content, err := c.mgr.st.ArtifactContent(c.run.Owner, string(id))
	if err != nil {
		return art, nil, false
	}
	return art, content, true
}

func (c *runCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
	art := core.Artifact{ProjectID: c.run.ProjectID, Owner: c.run.Owner, Name: name, Kind: kind, MIME: mime, SourceRun: c.run.ID}
	if err := c.mgr.st.CreateArtifact(&art, content); err != nil {
		return art, err
	}
	if _, err := c.mgr.emit(c.run.ConversationID, c.run.ID, core.EvArtifactCreated, core.ArtifactPayload{Artifact: art}); err != nil {
		return art, err
	}
	// 工件文本进检索索引。二进制 Office/PDF 先经 ingest 抽出纯文本，抽到才入库；
	// 抽不到则跳过，避免把 ZIP 字节当文本索引。
	if kind == "document" && len(content) < 200000 {
		if text := indexTextOf(name, mime, content); text != "" {
			if c.mgr.indexer != nil {
				_ = c.mgr.indexer.IndexDocument(c.run.Owner, c.run.ProjectID, "artifact", name, art.ID, text)
			} else {
				_ = c.mgr.st.AddKnowledge(core.KnowledgeDoc{Owner: c.run.Owner, ProjectID: c.run.ProjectID, Source: "artifact", Title: name, Ref: art.ID, Chunk: text})
			}
		}
	}
	return art, nil
}

// binaryDocMime 判定二进制文档 mime：此类工件字节不能按 UTF-8 文本进索引。
func binaryDocMime(mime string) bool {
	switch mime {
	case "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
		"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
		"application/vnd.openxmlformats-officedocument.presentationml.presentation",
		"application/vnd.ms-excel",
		"application/vnd.ms-powerpoint",
		"application/msword",
		"application/pdf",
		"application/zip":
		return true
	}
	return false
}

// indexTextOf 决定工件进检索索引的文本。普通文本原样返回；二进制文档 mime
// 先走 ingest 抽文本，解析失败或为空则返回空串（跳过索引）。
func indexTextOf(name, mime string, content []byte) string {
	if !binaryDocMime(mime) {
		return string(content)
	}
	doc := ingest.Parse(name, mime, content)
	if len(doc.Warnings) > 0 || len(doc.Blocks) == 0 {
		return ""
	}
	return strings.TrimSpace(doc.PlainText())
}

func (c *runCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) {
	if q.TopK <= 0 {
		q.TopK = int(c.eff.GetNumber("retrieval.topK", 5))
	}
	if q.ProjectID == "" {
		q.ProjectID = c.run.ProjectID
	}
	if c.mgr.retriever != nil {
		return c.mgr.retriever.Retrieve(c.ctx, c.run.Owner, q)
	}
	return c.mgr.st.SearchKnowledge(c.run.Owner, q)
}

// ExecuteTool 执行画像策略：deny→错误；confirm→HITL confirm 卡片；执行→截断转工件。
func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
	policy := c.prof.Tools
	if !policy.ToolAllowed(call.Name) {
		return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: fmt.Sprintf("tool %q denied by profile policy", call.Name)}
	}
	if policy.NeedsConfirm(call.Name) {
		var argsPreview string
		if len(call.Arguments) > 0 {
			argsPreview = truncate(string(call.Arguments), 600)
		}
		card := core.HITLCard{
			Question: fmt.Sprintf("执行工具 %s？", call.Name), Kind: core.HITLKindConfirm,
			Choices: []core.HITLChoice{
				{ID: "yes", Label: "执行", Description: "按上述参数调用", Impact: "该工具将访问外部资源并返回结果", Recommended: true},
				{ID: "no", Label: "取消", Description: "跳过本次调用", Impact: "模型会收到调用被拒绝的说明"},
			},
			AllowFreeInput: false,
		}
		_ = argsPreview
		resp, err := c.AskHITL(card)
		if err != nil {
			return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "hitl aborted: " + err.Error()}
		}
		if resp.ChoiceID != "yes" {
			return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "用户取消了本次工具调用"}
		}
	}
	tc := &toolCtx{rc: c}
	res := c.mgr.registry.Execute(tc, call)
	// 工具产物回填（交付物卡片根因修复）：注册表工具执行期经 rc.WriteArtifact
	// 写入的工件（write_artifact / fs_write_file / register_artifact / 未来执行
	// 原语产物）挂到结果上——llm.go 依 res.Artifact 生成对话内 artifact_ref 卡片。
	// 此前只有「结果超长截断转工件」会设置它，正常产物的卡片从未出现。
	if res.Artifact == nil && len(tc.arts) > 0 {
		last := tc.arts[len(tc.arts)-1]
		res.Artifact = &last
	}
	// 超长截断并转工件引用（4.10）。
	limit := c.prof.Context.MaxToolResultChars
	if limit <= 0 {
		limit = int(c.eff.GetNumber("context.maxToolResultChars", 20000))
	}
	if res.Status == "ok" && len(res.Output) > limit {
		if art, err := c.WriteArtifact("tool-"+call.Name+"-"+call.CallID+".txt", "data", "text/plain", []byte(res.Output)); err == nil {
			res.Artifact = &art
			res.Output = res.Output[:limit] + "\n…（超长，完整结果已存为工件 " + art.Name + "）"
		} else {
			res.Output = res.Output[:limit] + "…（超长截断，工件写入失败）"
		}
		res.Truncated = true
	}
	return res
}

// toolCtx 包装 runCtx 供工具执行；arts 记录本调用期间写入的工件，
// ExecuteTool 据此回填 res.Artifact（对话内交付物卡片的来源）。
type toolCtx struct {
	rc   *runCtx
	arts []core.Artifact
}

func (t *toolCtx) ProjectID() string      { return t.rc.run.ProjectID }
func (t *toolCtx) ConversationID() string { return t.rc.run.ConversationID }
func (t *toolCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) {
	return t.rc.Retrieve(q)
}
func (t *toolCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
	art, err := t.rc.WriteArtifact(name, kind, mime, content)
	if err == nil {
		t.arts = append(t.arts, art)
	}
	return art, err
}
func (t *toolCtx) Context() context.Context              { return t.rc.Context() }
func (t *toolCtx) OwnerID() string                       { return t.rc.run.Owner }
func (t *toolCtx) WorkspaceRoot() string                 { return t.rc.mgr.workspaceRoot }
func (t *toolCtx) SettingBool(key string, def bool) bool { return t.rc.eff.GetBool(key, def) }

// ---- 跨会话读取门禁（提案 2026-09-30-cross-session-reading；fail-closed）----
//
// canRead(A→B) = crossSession.read(项目闸门) AND A.allowCrossRead(出站，仅读他人时)
// AND !B.crossPrivate(入站，仅被他人读时) AND 同项目 AND 同租户。
// 出站/入站检查对自读（A==B）豁免：私有会话仍可读自身更早历史。

var (
	// ErrCrossReadDisabled 项目闸门关或本会话关闭出站读取。
	ErrCrossReadDisabled = errors.New("跨会话读取未开启（项目设置 crossSession.read 或会话 allowCrossRead）")
	// ErrCrossReadForbidden 目标不可读：私有（被他人读）或跨项目。
	ErrCrossReadForbidden = errors.New("目标会话不可读（私有或不在本项目内）")
)

// crossReadSelf 加载当前会话并校验项目闸门；出站检查由调用方按目标是否为自身决定。
func (t *toolCtx) crossReadSelf() (core.Conversation, error) {
	if !t.rc.eff.GetBool("crossSession.read", false) {
		return core.Conversation{}, ErrCrossReadDisabled
	}
	self, err := t.rc.mgr.st.GetConversation(t.rc.run.Owner, t.rc.run.ConversationID)
	if err != nil {
		return core.Conversation{}, err
	}
	return self, nil
}

func (t *toolCtx) PeerConversations() ([]core.Conversation, error) {
	self, err := t.crossReadSelf()
	if err != nil {
		return nil, err
	}
	if !self.AllowCrossRead {
		return nil, ErrCrossReadDisabled
	}
	convs, err := t.rc.mgr.st.ListConversations(t.rc.run.Owner, t.rc.run.ProjectID)
	if err != nil {
		return nil, err
	}
	out := make([]core.Conversation, 0, len(convs))
	for _, c := range convs {
		// 只读他人：排除自身；私有会话不出现（存在性本身是信息）。
		if c.ID == self.ID || c.CrossPrivate {
			continue
		}
		out = append(out, c)
	}
	return out, nil
}

func (t *toolCtx) PeerConversationMessages(conversationID string, limit int) ([]core.Message, error) {
	if limit <= 0 {
		limit = 50
	}
	if limit > 200 {
		limit = 200
	}
	self, err := t.crossReadSelf()
	if err != nil {
		return nil, err
	}
	// 租户校验先行（GetConversation 按 owner 过滤；跨租户 = NotFound，不泄露存在性）。
	target, err := t.rc.mgr.st.GetConversation(t.rc.run.Owner, conversationID)
	if err != nil {
		return nil, err
	}
	isSelf := target.ID == self.ID
	if !isSelf && !self.AllowCrossRead {
		return nil, ErrCrossReadDisabled
	}
	if !isSelf && (target.CrossPrivate || target.ProjectID != t.rc.run.ProjectID) {
		return nil, ErrCrossReadForbidden
	}
	return t.rc.mgr.st.LastMessages(conversationID, limit)
}

// ---- run 生命周期 ----

// StartConversationRun 为一条已落库的用户消息启动 conversation run。
// attachments 为已解析的附件文档（server 层经 ingest 处理）。
// inputs 是同一批附件的原件，落到会话工作区 inputs/，供模板和仿制使用。
func (m *Manager) StartConversationRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile) (core.Run, error) {
	// 会话内串行（设计策略落地）：在途 run（含 HITL 挂起）未终态前拒绝新 run。
	m.convMu.Lock()
	if m.convActive[conv.ID] {
		m.convMu.Unlock()
		return core.Run{}, ErrConversationBusy
	}
	m.convActive[conv.ID] = true
	m.convMu.Unlock()

	prof, _ := m.profiles.Get(conv.Owner, conv.ProfileID)
	eff := m.settings.Effective(conv.Owner, conv.ProjectID)
	run := core.Run{
		ID: store.NewID("run"), ConversationID: conv.ID, ProjectID: conv.ProjectID, Owner: conv.Owner,
		Kind: core.RunConversation, ProfileID: prof.ID, Status: core.RunPending, Phase: core.PhaseThinking,
	}
	if err := m.st.CreateRun(&run); err != nil {
		m.releaseConversation(conv.ID)
		return run, err
	}
	ctx, cancel := context.WithCancel(context.Background())
	h := &runHandle{cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
	m.mu.Lock()
	m.active[run.ID] = h
	m.mu.Unlock()

	runner := m.runner
	if eff.GetBool("runtime.mock", false) {
		runner = m.mock
	}
	inputPaths, _ := m.stageInputs(conv, eff, inputs)
	rc := &runCtx{mgr: m, run: run, prof: prof, eff: eff, ctx: ctx, h: h, userMessage: userMessage, attachments: attachments, inputPaths: inputPaths}
	go m.execute(rc, runner)
	return run, nil
}

// releaseConversation 释放会话在途槽位（run 终态/启动失败时）。
func (m *Manager) releaseConversation(conversationID string) {
	m.convMu.Lock()
	delete(m.convActive, conversationID)
	m.convMu.Unlock()
}

// execute 运行至终态并收敛：状态、usage、助手消息、收尾事件。
func (m *Manager) execute(rc *runCtx, runner runtime.Runner) {
	defer func() {
		m.releaseConversation(rc.run.ConversationID)
		m.mu.Lock()
		delete(m.active, rc.run.ID)
		m.mu.Unlock()
	}()
	m.setRunStatus(&rc.run, core.RunRunning, core.PhaseThinking)
	m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseThinking})

	execErr := func() (err error) {
		defer func() {
			if r := recover(); r != nil {
				err = fmt.Errorf("runner panic: %v", r)
			}
		}()
		return runner.Execute(rc)
	}()

	usage := core.Usage{DurationMs: float64(time.Since(rc.h.started).Milliseconds())}
	if execErr != nil {
		if rc.ctx.Err() != nil {
			m.finishRun(rc, core.RunCancelled, usage, &core.ErrorPart{Code: "run.cancelled", Message: "已取消"})
			return
		}
		m.finishRun(rc, core.RunFailed, usage, &core.ErrorPart{Code: "run.failed", Message: execErr.Error()})
		return
	}
	m.finishRun(rc, core.RunCompleted, usage, nil)
}

func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage, errPart *core.ErrorPart) {
	now := time.Now().UTC()
	rc.run.EndedAt = &now
	rc.run.Status = status
	rc.run.Phase = core.PhaseDone
	rc.run.Usage = &usage
	rc.run.Error = errPart
	m.st.UpdateRun(&rc.run)

	// 终态权威消息树：落库助手消息（含过程部件标记）。
	m.mu.Lock()
	parts := append([]core.Part{}, rc.h.assistant...)
	m.mu.Unlock()
	if len(parts) > 0 {
		msg := core.Message{ConversationID: rc.run.ConversationID, RunID: rc.run.ID, Role: core.RoleAssistant, Parts: parts}
		if err := m.st.AppendMessage(&msg); err == nil {
			// 消息树权威已落库；hitl 摘要部件的卡片状态由前端对账收敛。
			// submit_card 产物卡回填锚点：卡→消息互定位（D32 方向补全）。
			m.mu.Lock()
			cardIDs := append([]core.ID{}, rc.h.submittedCards...)
			m.mu.Unlock()
			if m.cards != nil {
				for _, id := range cardIDs {
					if card, err := m.cards.Get(rc.run.Owner, string(id)); err == nil {
						card.AnchorMessageID = msg.ID
						_ = m.cards.Update(rc.run.Owner, card)
					}
				}
			}
		}
		if errPart != nil {
			m.st.AppendMessage(&core.Message{ConversationID: rc.run.ConversationID, RunID: rc.run.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartError, Err: errPart}}})
		}
	} else if errPart != nil {
		m.st.AppendMessage(&core.Message{ConversationID: rc.run.ConversationID, RunID: rc.run.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartError, Err: errPart}}})
	}

	// 流式卡兜底（D45）：run 终态时仍有未 commit 的流式卡——按 run 结果置
	// done/failed，tone 标注「未正常收尾」；SetStatus 恒发布，收件箱可见终态。
	m.mu.Lock()
	streamID := rc.h.streamCardID
	rc.h.streamCardID = ""
	m.mu.Unlock()
	if streamID != "" && m.cards != nil {
		st := core.CardDone
		if status != core.RunCompleted {
			st = core.CardFailed
		}
		if card, err := m.cards.SetStatus(rc.run.Owner, string(streamID), st, "warn", "未正常收尾"); err == nil {
			m.mu.Lock()
			rc.h.submittedCards = append(rc.h.submittedCards, card.ID)
			m.mu.Unlock()
		}
	}

	switch status {
	case core.RunCompleted:
		usagePayload := core.RunCompletedPayload{Usage: usage}
		// 上下文组装观测随终态下发（4.10/4.17：构成与裁剪可解释）。
		m.mu.Lock()
		if rc.h.ctxObs != nil {
			usagePayload.Context = rc.h.ctxObs
		}
		m.mu.Unlock()
		m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunCompleted, usagePayload)
	case core.RunFailed:
		m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: errPart.Code, Message: errPart.Message})
	}
}

// setRunStatus 更新并广播 phase（状态不单独出事件，phase 事件携带）。
func (m *Manager) setRunStatus(r *core.Run, status core.RunStatus, phase core.RunPhase) {
	r.Status = status
	r.Phase = phase
	m.st.UpdateRun(r)
}

// emit 经批处理器持久化事件并广播；返回 seq（缓冲中的增量返回 0）。
func (m *Manager) emit(conversationID, runID string, typ core.StreamEventType, payload any) (int64, error) {
	return m.events.emit(conversationID, runID, typ, payload)
}

// CancelRun 取消 run 及其子 run（ctx 级联）。
func (m *Manager) CancelRun(runID string) error {
	r, err := m.st.GetRunByID(runID)
	if err != nil {
		return err
	}
	if r.Status == core.RunCompleted || r.Status == core.RunFailed || r.Status == core.RunCancelled {
		return fmt.Errorf("runmgr: run %s already terminal (%s)", runID, r.Status)
	}
	// 取消祖先句柄（子 run 派生自父 ctx，天然级联）。
	rootID := r.ParentRunID
	if rootID == "" {
		rootID = r.ID
	}
	m.mu.Lock()
	if h, ok := m.active[rootID]; ok {
		h.cancel()
		// 唤醒可能在等 HITL 的执行。
		select {
		case h.hitlCh <- hitlAnswer{err: context.Canceled}:
		default:
		}
	}
	if h, ok := m.active[r.ID]; ok && r.ID != rootID {
		h.cancel()
		select {
		case h.hitlCh <- hitlAnswer{err: context.Canceled}:
		default:
		}
	}
	m.mu.Unlock()
	return nil
}

// ---- HITL 解析（三分支路由，4.5）----

// ResolveHITL 处理用户对卡片的响应：
//  1. 选项命中 → 回填挂起 run；
//  2. 自由输入以 "/" 开头（命令）→ 取消挂起 run，返回转轨标志由 server 层发送新消息；
//  3. 其他自由输入 → 作为澄清回答回填。
func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (pivotText string, pivoted bool, err error) {
	card, err := m.st.GetHITLCard(owner, cardID)
	if err != nil {
		return "", false, err
	}
	if card.Resolved {
		return "", false, fmt.Errorf("runmgr: card %s already resolved", cardID)
	}
	run, err := m.st.GetRunByID(card.RunID)
	if err != nil {
		return "", false, err
	}

	kind := "choice"
	answer := ""
	if resp.ChoiceID != "" {
		for _, c := range card.Choices {
			if c.ID == resp.ChoiceID {
				answer = c.Label
				break
			}
		}
	} else if resp.FreeText != "" {
		kind = "free"
		answer = resp.FreeText
		trimmed := strings.TrimSpace(resp.FreeText)
		// 转轨判定（两源）：/ 命令，或意图闸门命中其他已注册意图。
		pivot := false
		pivotReason := ""
		if strings.HasPrefix(trimmed, "/") {
			pivot, pivotReason = true, "command"
		} else if m.gate != nil {
			if branch, topID := m.gate.Match(context.Background(), trimmed); branch == "matched" {
				pivot, pivotReason = true, "intent:"+topID
			}
		}
		if pivot {
			_ = m.CancelRun(run.ID)
			m.st.ResolveHITLCard(owner, cardID, core.HITLResolvedPayload{CardID: core.ID(cardID), Kind: "cancelled", Answer: answer})
			m.emit(run.ConversationID, run.ID, core.EvHITLResolved, core.HITLResolvedPayload{CardID: core.ID(cardID), Kind: "cancelled", Answer: answer})
			if m.cards != nil {
				_ = m.cards.ResolveByRef(owner, "hitl:"+cardID, "", answer)
			}
			_ = pivotReason
			return trimmed, true, nil
		}
	}
	resolution := core.HITLResolvedPayload{CardID: core.ID(cardID), Answer: answer, Kind: kind}
	if err := m.st.ResolveHITLCard(owner, cardID, resolution); err != nil {
		return "", false, err
	}
	m.emit(run.ConversationID, run.ID, core.EvHITLResolved, resolution)
	// 投影卡同步销项（双现；幂等）。
	if m.cards != nil {
		_ = m.cards.ResolveByRef(owner, "hitl:"+cardID, resp.ChoiceID, answer)
	}

	m.mu.Lock()
	h, ok := m.active[run.ID]
	m.mu.Unlock()
	if !ok {
		return "", false, fmt.Errorf("runmgr: run %s not active (may have ended)", run.ID)
	}
	select {
	case h.hitlCh <- hitlAnswer{resp: resp}:
	default:
		return "", false, fmt.Errorf("runmgr: hitl channel busy")
	}
	// 恢复运行状态。
	if run.Status == core.RunWaitingHITL {
		run.Status = core.RunRunning
		m.st.UpdateRun(&run)
	}
	return "", false, nil
}

// ---- 子代理（任务树）----

func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, name, task, hint string) (core.AgentResultPayload, error) {
	child := core.Run{
		ID: store.NewID("run"), ParentRunID: parent.ID, ConversationID: parent.ConversationID,
		ProjectID: parent.ProjectID, Owner: parent.Owner, Kind: core.RunSubagent, ProfileID: parent.ProfileID,
		Status: core.RunPending, Phase: core.PhaseThinking, AgentName: name,
	}
	if err := m.st.CreateRun(&child); err != nil {
		return core.AgentResultPayload{}, err
	}
	m.emit(parent.ConversationID, parent.ID, core.EvAgentSpawn, core.AgentSpawnPayload{AgentRun: child.ID, Name: name, Task: task})

	ctx, cancel := context.WithCancel(parentCtx)
	h := &runHandle{cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
	m.mu.Lock()
	m.active[child.ID] = h
	m.mu.Unlock()

	prof, _ := m.profiles.Get(parent.Owner, parent.ProfileID)
	eff := m.settings.Effective(parent.Owner, parent.ProjectID)
	rc := &runCtx{mgr: m, run: child, prof: prof, eff: eff, ctx: ctx, h: h, subTask: task, subHint: hint}
	m.setRunStatus(&rc.run, core.RunRunning, core.PhaseActing)

	runner := m.subRunner
	if eff.GetBool("runtime.mock", false) || runner == nil {
		runner = m.mockSub
	}
	execErr := func() (err error) {
		defer func() {
			if r := recover(); r != nil {
				err = fmt.Errorf("subagent panic: %v", r)
			}
		}()
		return runner.Execute(rc)
	}()

	usage := core.Usage{DurationMs: float64(time.Since(h.started).Milliseconds())}
	status := core.RunCompleted
	summary := "完成"
	payload := core.AgentResultPayload{AgentRun: child.ID, Status: status, Summary: summary}
	if execErr != nil {
		if ctx.Err() != nil {
			status = core.RunCancelled
			payload.Status = status
			payload.Summary = "已取消"
		} else {
			status = core.RunFailed
			payload.Status = status
			payload.Summary = execErr.Error()
		}
	}
	now := time.Now().UTC()
	child.EndedAt = &now
	child.Status = status
	child.Phase = core.PhaseDone
	child.Usage = &usage
	m.st.UpdateRun(&child)
	m.emit(child.ConversationID, child.ID, core.EvAgentResult, payload)

	// 子代理中间过程落库（AgentTracker 弹窗的权威数据源；消息树带 run_id 归属）。
	m.mu.Lock()
	subParts := append([]core.Part{}, h.assistant...)
	m.mu.Unlock()
	fullText := partsText(subParts)
	if len(subParts) > 0 {
		_ = m.st.AppendMessage(&core.Message{ConversationID: child.ConversationID, RunID: child.ID, Role: core.RoleAssistant, Parts: subParts})
	}
	if fullText != "" {
		payload.Text = fullText
		if payload.Summary == "" || payload.Summary == "完成" {
			payload.Summary = truncate(fullText, 160)
		}
	}

	m.mu.Lock()
	delete(m.active, child.ID)
	m.mu.Unlock()
	if status != core.RunCompleted {
		return payload, fmt.Errorf("subagent %s: %s", name, payload.Summary)
	}
	return payload, nil
}

// InterruptRunning 服务停机前把在途 run 标记为 interrupted（4.2 恢复策略）。
func (m *Manager) InterruptRunning() {
	m.mu.Lock()
	ids := make([]string, 0, len(m.active))
	for id := range m.active {
		ids = append(ids, id)
	}
	handles := make([]*runHandle, 0, len(ids))
	for _, id := range ids {
		if h, ok := m.active[id]; ok {
			handles = append(handles, h)
		}
	}
	m.mu.Unlock()
	for _, h := range handles {
		h.cancel()
	}
	for _, id := range ids {
		r, err := m.st.GetRunByID(id)
		if err != nil || r.Status == core.RunCompleted || r.Status == core.RunFailed || r.Status == core.RunCancelled {
			continue
		}
		r.Status = core.RunInterrupted
		m.st.UpdateRun(&r)
	}
}

func partsText(parts []core.Part) string {
	var sb strings.Builder
	for _, p := range parts {
		if p.Type == core.PartText {
			sb.WriteString(p.Text)
			sb.WriteString("\n")
		}
	}
	return strings.TrimSpace(sb.String())
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n] + "…"
}