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

package runmgr

import (
	"context"
	"encoding/json"
	"log"
	"time"

	core "agent-system-core"
)

// 协作存储 U5（KTD5）：持久命令消费。任意 API 实例把取消/HITL 写入
// run_commands 收件箱；命令由「目标 run 的当前执行实例」消费（路由依据
// conversation_executions 的当前 owner；本地不在执行的实例跳过，命令保持
// pending 直到执行实例领取或超时由恢复路径收口）。
//
// 本文件是执行侧消费循环；server 侧的投递入口见 server 包 EnqueueRunCommand
// 的两个薄封装（CancelRunRemote / ResolveHITLRemote）。

// commandPollEvery 消费轮询间隔。
const commandPollEvery = 300 * time.Millisecond

// HITLCommandPayload HITL 命令载荷（gate 身份由 cardID+gateVersion 承载：
// 新 owner 消费时重验 gate 版本，过期审批明确拒绝而不静默丢弃——KTD5）。
type HITLCommandPayload struct {
	Owner       string          `json:"owner"`
	CardID      string          `json:"cardId"`
	GateVersion int64           `json:"gateVersion"`
	Kind        string          `json:"kind"`
	Answer      json.RawMessage `json:"answer,omitempty"`
}

// CancelCommandPayload 取消命令载荷。
type CancelCommandPayload struct {
	Owner  string `json:"owner"`
	RunID  string `json:"runId"`
	Reason string `json:"reason,omitempty"`
}

// StartCommandConsumer 启动命令消费循环（仅 PG 档；SQLite 单实例无跨实例
// 命令）。返回停止函数。消费的命令只处理「本实例当前持有执行权」的会话；
// 其余跳过（不认领）——避免非执行实例吃掉命令。
func (m *Manager) StartCommandConsumer(ctx context.Context) (stop func()) {
	if m.st.Dialect() != "postgres" {
		return func() {}
	}
	ctx, cancel := context.WithCancel(ctx)
	go func() {
		t := time.NewTicker(commandPollEvery)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				m.consumeCommandsOnce()
			}
		}
	}()
	return cancel
}

// consumeCommandsOnce 单轮消费：先按执行权过滤（不认领非本实例会话的命令），
// 再逐条原子认领并执行。认领即消费（at-least-once；命令处理幂等：取消对
// 终态 run 报已终态、HITL 重验 gate）。
func (m *Manager) consumeCommandsOnce() {
	pending, err := m.pendingCommandsForOwned()
	if err != nil || len(pending) == 0 {
		return
	}
	for _, c := range pending {
		claimed, err := m.claimCommand(c.CommandID)
		if err != nil || !claimed {
			continue
		}
		switch c.Kind {
		case "cancel":
			m.applyCancelCommand(c)
		case "hitl":
			m.applyHITLCommand(c)
		default:
			log.Printf("[runmgr] command %s: unknown kind %q", c.CommandID, c.Kind)
		}
	}
}

// pendingCommandsForOwned 列出 pending 且其会话执行权在本实例的命令。
func (m *Manager) pendingCommandsForOwned() ([]ownedCommand, error) {
	rows, err := m.st.DB().Query(
		`SELECT c.command_id, c.kind, c.run_id, c.conversation_id, c.epoch, c.payload
		 FROM run_commands c JOIN conversation_executions e ON e.conversation_id = c.conversation_id
		 WHERE c.state='pending' AND e.instance_id = $1
		 ORDER BY c.created_at LIMIT 16`, m.InstanceID())
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ownedCommand
	for rows.Next() {
		var c ownedCommand
		if err := rows.Scan(&c.CommandID, &c.Kind, &c.RunID, &c.ConversationID, &c.Epoch, &c.Payload); err != nil {
			return nil, err
		}
		out = append(out, c)
	}
	return out, rows.Err()
}

type ownedCommand struct {
	CommandID      string
	Kind           string
	RunID          string
	ConversationID string
	Epoch          int64
	Payload        string
}

// claimCommand 条件认领（pending→consumed，consumer=本实例）。
func (m *Manager) claimCommand(commandID string) (bool, error) {
	res, err := m.st.DB().Exec(
		`UPDATE run_commands SET state='consumed', consumed_at=$3, consumer=$1 WHERE command_id=$2 AND state='pending'`,
		m.InstanceID(), commandID, time.Now().UTC().Format(time.RFC3339Nano))
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	return n > 0, nil
}

// applyCancelCommand 本地执行取消：run 在本实例活跃 → 级联取消；
// 不活跃（如执行实例重启后命令迟到）→ 库级收口（非终态置 cancelled）。
func (m *Manager) applyCancelCommand(c ownedCommand) {
	var p CancelCommandPayload
	if err := json.Unmarshal([]byte(c.Payload), &p); err != nil {
		log.Printf("[runmgr] cancel %s payload: %v", c.CommandID, err)
		return
	}
	runID := c.RunID
	if runID == "" {
		runID = p.RunID
	}
	// 本地活跃 → 级联取消（含 HITL 唤醒）；否则库级收口。CancelRun 对
	// 非本地 run 不落库（只动内存句柄），不能作为成败判据。
	if m.RunActiveLocally(runID) {
		if err := m.CancelRun(runID); err == nil {
			return
		}
	}
	// 库级取消（幂等——仅非终态可置）。
	if _, err := m.st.DB().Exec(
		`UPDATE runs SET status='cancelled', ended_at=$2 WHERE id=$1 AND status IN ('pending','running','waiting_hitl')`, runID, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
		log.Printf("[runmgr] remote cancel %s: %v", runID, err)
	}
}

// applyHITLCommand 本地执行 HITL 应答：重验 gate（cardID + gateVersion），
// 过期/未知明确拒绝（卡片置 resolved+stale，事件可见），不静默丢弃（KTD5）。
func (m *Manager) applyHITLCommand(c ownedCommand) {
	var p HITLCommandPayload
	if err := json.Unmarshal([]byte(c.Payload), &p); err != nil {
		log.Printf("[runmgr] hitl %s payload: %v", c.CommandID, err)
		return
	}
	resp := core.HITLResponse{CardID: core.ID(p.CardID)}
	if p.Kind == "free" {
		resp.FreeText = string(p.Answer)
	} else {
		resp.ChoiceID = p.Kind // choiceId 即按钮语义（与 cardAction 契约一致）
	}
	_, _, err := m.ResolveHITL(p.Owner, p.CardID, resp)
	if err != nil {
		// 过期/未知 gate：明确销项（resolved + 注记），发出可见结论。
		_ = m.resolveHITLCardForEffectiveAccount(p.Owner, p.CardID, core.HITLResolvedPayload{
			CardID: core.ID(p.CardID), Kind: "expired", Answer: "跨实例审批迟到：gate 已过期或执行已移交",
		})
		log.Printf("[runmgr] hitl %s (card %s) rejected: %v", c.CommandID, p.CardID, err)
	}
}