heicode-mananger/heicode/controller/agent_swarm_query.go

package controller

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/heicode/manager/common"
	"github.com/heicode/manager/model"
	"gorm.io/gorm"
)

// HM-side Swarm Run query (#45 Phase1: list / status / events?after / artifacts / stop).
//
// 设计要点(关键):**只读查询全部基于 HM 本地已持久化的数据**——Swarm 运行时通过带签名回调
// (`/api/agent/callbacks/runtime-events`,见 agent_callback.go)把生命周期事件推给 HM,HM 落
// 到 AgentDeployment + AgentCallbackEvent。因此 list/status/events/artifacts **无需**调用 Swarm
// 运行时,也就**不依赖 `agent_swarm#2` 尚未冻结的拉取契约**,返工风险低。
//
// 仅 `stop`(写操作)需要真正调用 Swarm 运行时;在契约冻结 + `SWARM_RUNTIME_ENABLED=true` 前
// 默认关闭并明确提示(不 mock,不臆造未冻结的写接口)。
//
// 字段口径对齐 agent_swarm/docs/integration/runtime-contract.md(草案):
// deployment_id ↔ swarm_id ↔ manager_deployment_id 三者映射;事件按 swarm_id/deployment_id 持久化。

// findUserSwarmDeployment 按 :id(匹配 deployment_id / runtime_swarm_id / correlation_id)加载
// 当前用户的 swarm 部署。非本人或非 swarm 模式 → 失败。
func findUserSwarmDeployment(c *gin.Context) (model.AgentDeployment, bool) {
	userID := c.GetInt("id")
	id := strings.TrimSpace(c.Param("id"))
	var dep model.AgentDeployment
	err := model.DB.Where(
		"user_id = ? AND (deployment_id = ? OR runtime_swarm_id = ? OR correlation_id = ?)",
		strconv.Itoa(userID), id, id, id,
	).First(&dep).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			agentError(c, "POLICY_REJECTED", "swarm run not found")
		} else {
			agentError(c, "DEPLOYMENT_CONFLICT", "failed to load swarm run")
		}
		return model.AgentDeployment{}, false
	}
	if !isSwarmDeployment(dep) {
		agentError(c, "POLICY_REJECTED", "deployment is not a swarm run")
		return model.AgentDeployment{}, false
	}
	return dep, true
}

func isSwarmDeployment(dep model.AgentDeployment) bool {
	return strings.EqualFold(dep.SubMode, "swarm") || strings.TrimSpace(dep.RuntimeSwarmID) != ""
}

// swarmDeploymentView 是 swarm 运行的状态摘要视图(脱敏:不含 plan/payload 等大字段与凭据)。
func swarmDeploymentView(dep model.AgentDeployment) gin.H {
	return gin.H{
		"deployment_id":         dep.DeploymentID,
		"swarm_id":              dep.RuntimeSwarmID,
		"runtime_deployment_id": dep.RuntimeDeploymentID,
		"correlation_id":        dep.CorrelationID,
		"status":                dep.Status,                     // 运行时真实状态(契约 §4)
		"display_status":        swarmDisplayStatus(dep.Status), // 客户端展示态(契约 §4.1 映射)
		"goal_summary":          swarmGoalSummary(dep.PlanJSON), // 单行目标(客户端 Run 列表展示;@Mem0ried #28 消费需求)
		"phase":                 dep.Phase,
		"runtime_state":         dep.RuntimeState,
		"failure_reason":        dep.FailureReason,
		"created_at":            dep.CreatedAtText,
		"updated_at":            dep.UpdatedAtText,
		"runtime_last_sync_at":  dep.RuntimeLastSyncAtText,
	}
}

// swarmGoalSummary 从持久化的 plan_json 提取顶层 objective,折叠为单行供客户端 Run 列表展示
// (#28 消费需求:列表不止 deployment_id/status)。空白折叠 + 截断 + RedactText 兜底(objective
// 是用户目标文本而非密钥,但仍按统一红线剥离误入的 sk-/token);取不到则返回空串(不臆造)。
func swarmGoalSummary(planJSON string) string {
	if strings.TrimSpace(planJSON) == "" {
		return ""
	}
	var plan struct {
		Objective string `json:"objective"`
	}
	if err := common.UnmarshalJsonStr(planJSON, &plan); err != nil {
		return ""
	}
	s := strings.TrimSpace(plan.Objective)
	if s == "" {
		return ""
	}
	s = strings.Join(strings.Fields(s), " ") // 折叠所有空白(含换行)为单空格
	const maxLen = 200
	if len([]rune(s)) > maxLen {
		s = strings.TrimSpace(string([]rune(s)[:maxLen])) + "…"
	}
	return model.RedactText(s)
}

// swarmDisplayStatus 把运行时真实状态映射到客户端展示态(agent_swarm runtime-contract §4.1)。
// 运行时只有 waiting_approval/running/blocked/completed/failed/stopped;`blocked` 展示为
// `degraded`,其余直通。preparing/verifying 是 running 的子态、由运行时阶段决定,HM 未单独存,
// 不臆造(规则:无信号不造态)。
func swarmDisplayStatus(status string) string {
	if strings.EqualFold(strings.TrimSpace(status), "blocked") {
		return "degraded"
	}
	return status
}

// swarmSensitivePayloadKeys 是事件 payload 中**绝不下发**给客户端的键(凭据/大字段)。
// 递归剔除(#45 复审 #1:回调 envelope 可能含 secret_ref,如 approval.requested)。
var swarmSensitivePayloadKeys = map[string]bool{
	"secret_ref": true, "secretref": true, "credentials": true, "credential": true,
	// agent_swarm event-schema v1(#15):envelope 按设计透传 azkv:// 引用,客户端可见视图须剔除。
	"credential_ref": true, "signing_secret_ref": true,
	"secret": true, "token": true, "access_token": true, "refresh_token": true,
	"api_key": true, "apikey": true, "private_key": true, "access_key": true,
	"password": true, "passwd": true,
	// 大字段/内部结构,避免顺带泄漏
	"plan": true, "payload": true, "permission_manifest": true, "env": true,
}

// stripSensitiveKeys 递归删除敏感键(键名小写匹配 swarmSensitivePayloadKeys)。
func stripSensitiveKeys(v any) any {
	switch t := v.(type) {
	case map[string]any:
		out := make(map[string]any, len(t))
		for k, val := range t {
			if swarmSensitivePayloadKeys[strings.ToLower(strings.TrimSpace(k))] {
				continue
			}
			out[k] = stripSensitiveKeys(val)
		}
		return out
	case []any:
		out := make([]any, 0, len(t))
		for _, item := range t {
			out = append(out, stripSensitiveKeys(item))
		}
		return out
	default:
		return v
	}
}

// sanitizeSwarmPayload 递归剔除敏感/大字段键,再对序列化结果跑一次 RedactText 兜底
// (剥离 sk-/Bearer/URL token/JSON 密钥字段)。
func sanitizeSwarmPayload(raw string) map[string]any {
	m := unmarshalResourceJSON(raw)
	if m == nil {
		return nil
	}
	cleaned, _ := stripSensitiveKeys(m).(map[string]any)
	if b, err := common.Marshal(cleaned); err == nil {
		var out map[string]any
		if err := common.UnmarshalJsonStr(model.RedactText(string(b)), &out); err == nil {
			return out
		}
	}
	return cleaned
}

// swarmEventView maps a persisted callback event to the client view (payload 脱敏)。
func swarmEventView(e model.AgentCallbackEvent) gin.H {
	return gin.H{
		"id":                e.Id,       // HM 不透明分页游标(next_after);单调,兼容 sequence 未上线
		"sequence":          e.Sequence, // agent_swarm v1 per-swarm 序号(客户端去重/排序;0=envelope 未带)
		"event_id":          e.EventID,
		"event_type":        e.EventType,
		"task_id":           e.TaskID,
		"agent_instance_id": e.AgentInstanceID,
		"result":            e.Result,
		"occurred_at":       e.OccurredAt,
		"created_at_ms":     e.CreatedAtMs,
		"payload":           sanitizeSwarmPayload(e.PayloadJSON),
	}
}

// HeicodeListSwarms: GET /api/heicode/swarms — 列出当前用户的 swarm 运行(读 HM 本地部署表)。
func HeicodeListSwarms(c *gin.Context) {
	userID := c.GetInt("id")
	if userID <= 0 {
		agentError(c, "POLICY_REJECTED", "authentication required")
		return
	}
	if model.DB == nil {
		agentError(c, "DEPLOYMENT_PERSIST_FAILED", "database not initialised")
		return
	}
	var rows []model.AgentDeployment
	err := model.DB.Where(
		"user_id = ? AND (LOWER(sub_mode) = ? OR runtime_swarm_id <> '')",
		strconv.Itoa(userID), "swarm",
	).Order("created_at_ms desc, id desc").Limit(200).Find(&rows).Error
	if err != nil {
		agentError(c, "DEPLOYMENT_CONFLICT", "failed to list swarm runs")
		return
	}
	items := make([]gin.H, 0, len(rows))
	for _, r := range rows {
		items = append(items, swarmDeploymentView(r))
	}
	common.ApiSuccess(c, gin.H{"items": items, "total": len(items)})
}

// HeicodeGetSwarmStatus: GET /api/heicode/swarms/:id — 单个 swarm 运行状态。
func HeicodeGetSwarmStatus(c *gin.Context) {
	dep, ok := findUserSwarmDeployment(c)
	if !ok {
		return
	}
	common.ApiSuccess(c, swarmDeploymentView(dep))
}

// HeicodeListSwarmEvents: GET /api/heicode/swarms/:id/events?after=&limit= — 事件增量拉取。
func HeicodeListSwarmEvents(c *gin.Context) {
	dep, ok := findUserSwarmDeployment(c)
	if !ok {
		return
	}
	after, _ := strconv.Atoi(strings.TrimSpace(c.Query("after")))
	limit, _ := strconv.Atoi(strings.TrimSpace(c.Query("limit")))
	events, err := model.ListSwarmCallbackEventsAfter(strconv.Itoa(c.GetInt("id")), dep.DeploymentID, dep.RuntimeSwarmID, after, limit)
	if err != nil {
		agentError(c, "DEPLOYMENT_CONFLICT", "failed to list swarm events")
		return
	}
	items := make([]gin.H, 0, len(events))
	nextAfter := after
	for _, e := range events {
		items = append(items, swarmEventView(e))
		if e.Id > nextAfter {
			nextAfter = e.Id
		}
	}
	common.ApiSuccess(c, gin.H{"items": items, "next_after": nextAfter, "count": len(items)})
}

// HeicodeListSwarmArtifacts: GET /api/heicode/swarms/:id/artifacts — 从已持久化事件中筛产物。
// 当前从 event_type 含 "artifact" 的回调事件派生(真实数据);专用 artifact 端点待 agent_swarm#2 冻结后补。
func HeicodeListSwarmArtifacts(c *gin.Context) {
	dep, ok := findUserSwarmDeployment(c)
	if !ok {
		return
	}
	events, err := model.ListSwarmCallbackEventsAfter(strconv.Itoa(c.GetInt("id")), dep.DeploymentID, dep.RuntimeSwarmID, 0, 1000)
	if err != nil {
		agentError(c, "DEPLOYMENT_CONFLICT", "failed to list swarm artifacts")
		return
	}
	items := make([]gin.H, 0)
	for _, e := range events {
		if strings.Contains(strings.ToLower(e.EventType), "artifact") {
			items = append(items, swarmArtifactView(e))
		}
	}
	common.ApiSuccess(c, gin.H{"items": items, "total": len(items)})
}

// swarmArtifactView 从 artifact.created 事件产出客户端扁平视图(agent_swarm event-schema v1):
// {uri, checksum, task_id, size_bytes?, created_at}。值取自(已脱敏的)payload —— uri/checksum
// 非敏感保留;**绝不含 secret_ref**。size 未知则省略(不伪造,契约要求)。
func swarmArtifactView(e model.AgentCallbackEvent) gin.H {
	p := sanitizeSwarmPayload(e.PayloadJSON)
	get := func(k string) any {
		if p == nil {
			return nil
		}
		return p[k]
	}
	createdAt := get("created_at")
	if createdAt == nil || createdAt == "" {
		createdAt = e.OccurredAt
	}
	out := gin.H{
		"event_id":   e.EventID,
		"sequence":   e.Sequence,
		"task_id":    firstNonNil(get("task_id"), e.TaskID),
		"uri":        get("uri"),
		"checksum":   get("checksum"),
		"created_at": createdAt,
	}
	if sz := get("size_bytes"); sz != nil { // 未知则省略,不伪造
		out["size_bytes"] = sz
	}
	return out
}

// firstNonNil 返回第一个非 nil/非空的值。
func firstNonNil(vals ...any) any {
	for _, v := range vals {
		if v != nil && v != "" {
			return v
		}
	}
	return nil
}

// HeicodeStopSwarm: POST /api/heicode/swarms/:id/stop — 停止 swarm 运行(写操作)。
// 这是唯一需要真正调用 Swarm 运行时的操作。契约已随 agent_swarm runtime-contract v1 冻结
// (#45 复审 + agent_swarm#14):POST /api/agent/swarm/deployments/{deployment_id}/stop,
// Bearer SWARM_RUNTIME_SERVICE_TOKEN,X-Idempotency-Key 幂等;运行时受理后异步停机并回推
// swarm.stopped 事件。开关 SWARM_RUNTIME_ENABLED=false(默认)或缺 base_url/token 时一律拒绝,
// **绝不伪造 accepted**(#45 复审 #2)。
func HeicodeStopSwarm(c *gin.Context) {
	dep, ok := findUserSwarmDeployment(c)
	if !ok {
		return
	}
	cfg := agentRuntimeClientConfigForMode(agentRuntimeModeSwarm)
	// 开关 + base_url + 服务令牌缺一不可:任一缺失都不调用、不伪造受理。
	if !cfg.Enabled || strings.TrimSpace(cfg.BaseURL) == "" || strings.TrimSpace(cfg.Token) == "" {
		agentError(c, "POLICY_REJECTED",
			"swarm stop not enabled: requires SWARM_RUNTIME_ENABLED=true with SWARM_RUNTIME_BASE_URL and SWARM_RUNTIME_SERVICE_TOKEN")
		return
	}
	var body struct {
		Reason string `json:"reason"`
	}
	_ = c.ShouldBindJSON(&body)
	result, err := callSwarmRuntimeStop(c.Request.Context(), cfg, dep, body.Reason)
	if err != nil {
		common.SysLog("HeicodeStopSwarm: " + err.Error())
		agentError(c, "RUNTIME_UNAVAILABLE", "failed to stop swarm run at runtime: "+err.Error())
		return
	}
	// 运行时已受理停机;真正终态由运行时异步回推 swarm.stopped(经 /api/agent/callbacks)写回,
	// HM 此处不抢先改写 dep.Status 以免与回调竞态。只落审计。
	recordAgentAuditEvent(agentEvent{
		EventID:       "evt_" + common.GetUUID()[:12],
		Event:         "swarm.stop_requested",
		SchemaVersion: 1,
		UserID:        dep.UserID,
		ChannelID:     dep.ChannelID,
		BindingScope:  dep.BindingScope,
		DeploymentID:  dep.DeploymentID,
		CorrelationID: dep.CorrelationID,
		OccurredAt:    agentNow(),
	}, "agent_swarm_query", dep.DeploymentID, agentRequestID(c), "ok")
	common.ApiSuccess(c, gin.H{
		"deployment_id":  dep.DeploymentID,
		"swarm_id":       dep.RuntimeSwarmID,
		"accepted":       true,
		"runtime_status": firstNonEmpty(result.RuntimeStatus, "stopping"),
		"note":           "stop accepted by runtime; final state arrives via swarm.stopped callback",
	})
}

// swarmStopResult 是运行时 stop 调用的归一化结果。
type swarmStopResult struct {
	RuntimeStatus string
	HTTPStatus    int
}

// callSwarmRuntimeStop 向 Swarm 运行时发起真实 stop(runtime-contract v1)。复用 agent_runtime_client
// 的配置/URL/信封解析 helper,但按 model.AgentDeployment 直接构造请求(无需 agentDeploymentRecord)。
// runtime id 取 runtime_deployment_id,缺失时回退 runtime_swarm_id(契约三映射)。
func callSwarmRuntimeStop(ctx context.Context, cfg agentRuntimeConfig, dep model.AgentDeployment, reason string) (swarmStopResult, error) {
	runtimeID := firstNonEmpty(strings.TrimSpace(dep.RuntimeDeploymentID), strings.TrimSpace(dep.RuntimeSwarmID))
	if runtimeID == "" {
		return swarmStopResult{}, errors.New("swarm run has no runtime deployment id yet")
	}
	endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeStopPath(cfg, runtimeID))
	if err != nil {
		return swarmStopResult{}, err
	}
	payload, err := common.Marshal(gin.H{
		"reason":                firstNonEmpty(strings.TrimSpace(reason), "Heicode Manager requested stop"),
		"manager_deployment_id": dep.DeploymentID,
	})
	if err != nil {
		return swarmStopResult{}, err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
	if err != nil {
		return swarmStopResult{}, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(cfg.Token))
	// 幂等键固定按 manager deployment_id 派生:同一 stop 重试不会在运行时侧重复执行(契约要求)。
	req.Header.Set("X-Idempotency-Key", "manager-stop-"+dep.DeploymentID)
	if cid := strings.TrimSpace(dep.CorrelationID); cid != "" {
		req.Header.Set("X-Correlation-ID", cid)
	}
	client := &http.Client{Timeout: cfg.Timeout}
	resp, err := client.Do(req)
	if err != nil {
		return swarmStopResult{}, err
	}
	defer resp.Body.Close()
	respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if readErr != nil {
		return swarmStopResult{HTTPStatus: resp.StatusCode}, readErr
	}
	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		return swarmStopResult{HTTPStatus: resp.StatusCode}, fmt.Errorf("runtime stop returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
	}
	var envelope map[string]any
	if len(respBody) > 0 {
		if err := common.Unmarshal(respBody, &envelope); err != nil {
			return swarmStopResult{HTTPStatus: resp.StatusCode}, err
		}
	}
	if message := agentRuntimeEnvelopeError(envelope); message != "" {
		return swarmStopResult{HTTPStatus: resp.StatusCode}, errors.New(message)
	}
	data := extractAgentRuntimeData(envelope)
	return swarmStopResult{
		RuntimeStatus: stringFromMap(data, "runtime_status", "status"),
		HTTPStatus:    resp.StatusCode,
	}, nil
}

// swarmTerminalEventTypes 是 agent_swarm event-schema v1(#15)的运行终态事件。命中即可结束 SSE 流。
var swarmTerminalEventTypes = map[string]bool{
	"swarm.completed": true,
	"swarm.failed":    true,
	"swarm.stopped":   true,
}

func swarmEventIsTerminal(eventType string) bool {
	return swarmTerminalEventTypes[strings.ToLower(strings.TrimSpace(eventType))]
}

// HeicodeStreamSwarmEvents: GET /api/heicode/swarms/:id/events/stream?after= — SSE 实时事件流。
//
// 与 §events?after 同源:**全部来自 HM 已持久化的回调事件**(model.ListSwarmCallbackEventsAfter,
// 按 user_id 收口防跨用户泄漏),服务端轮询新事件后以 SSE 帧推送,**无需调用 Swarm 运行时**——
// 因此与 stop 不同,不受 SWARM_RUNTIME_ENABLED 限制。事件 payload 经 swarmEventView 脱敏。
// 进入即回放 after 之后的历史(免去客户端先调 events?after 再订阅);命中终态事件
// (swarm.completed/failed/stopped)、客户端断开或超过 maxLifetime 即结束。
func HeicodeStreamSwarmEvents(c *gin.Context) {
	dep, ok := findUserSwarmDeployment(c)
	if !ok {
		return
	}
	userID := strconv.Itoa(c.GetInt("id"))
	after, _ := strconv.Atoi(strings.TrimSpace(c.Query("after")))

	c.Writer.Header().Set("Content-Type", "text/event-stream")
	c.Writer.Header().Set("Cache-Control", "no-cache")
	c.Writer.Header().Set("Connection", "keep-alive")
	c.Writer.Header().Set("X-Accel-Buffering", "no") // 关掉反代缓冲,逐帧下发

	const (
		pollInterval = 1500 * time.Millisecond
		maxLifetime  = 30 * time.Minute
	)
	deadline := time.Now().Add(maxLifetime)
	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()
	ctx := c.Request.Context()

	// flush 推送 after 之后的新事件,返回 false 表示已命中终态(应结束流)或读失败。
	flush := func() bool {
		events, err := model.ListSwarmCallbackEventsAfter(userID, dep.DeploymentID, dep.RuntimeSwarmID, after, 200)
		if err != nil {
			c.SSEvent("error", gin.H{"message": "failed to read swarm events"})
			c.Writer.Flush()
			return false
		}
		terminal := false
		for _, e := range events {
			c.SSEvent("message", swarmEventView(e))
			if e.Id > after {
				after = e.Id
			}
			if swarmEventIsTerminal(e.EventType) {
				terminal = true
			}
		}
		c.Writer.Flush()
		return !terminal
	}
	if !flush() {
		c.SSEvent("done", gin.H{"next_after": after, "reason": "terminal"})
		c.Writer.Flush()
		return
	}
	for {
		select {
		case <-ctx.Done(): // 客户端断开
			return
		case <-ticker.C:
			if time.Now().After(deadline) {
				c.SSEvent("done", gin.H{"next_after": after, "reason": "timeout"})
				c.Writer.Flush()
				return
			}
			if !flush() {
				c.SSEvent("done", gin.H{"next_after": after, "reason": "terminal"})
				c.Writer.Flush()
				return
			}
		}
	}
}