feat(swarm): HM-side Swarm Run read-only query — Phase1 (#45)
#45 Phase1 的只读查询(list/status/events?after/artifacts),全部基于 HM 已持久化的 运行时回调数据(Swarm → HM 带签名回调,见 agent_callback.go),**无需实时调 Swarm**, 因此不被 agent_swarm#2 契约冻结阻塞、返工风险低: - GET /api/heicode/swarms — 列出当前用户的 swarm 运行(AgentDeployment, sub_mode=swarm 或有 runtime_swarm_id) - GET /api/heicode/swarms/:id — 状态(:id = deployment_id/swarm_id/correlation_id 任一) - GET /api/heicode/swarms/:id/events?after=&limit= — 事件增量拉取(id 游标 next_after,oldest-first) 新增 model.ListSwarmCallbackEventsAfter(按 deployment_id/swarm_id + id>after) - GET /api/heicode/swarms/:id/artifacts — 从已存事件(event_type 含 artifact)派生 - POST /api/heicode/swarms/:id/stop — 唯一写操作;在 agent_swarm#2 冻结 + SWARM_RUNTIME_ENABLED=true 前默认关闭并明确提示(不臆造未冻结写接口) 字段口径对齐 agent_swarm/docs/integration/runtime-contract.md(deployment_id↔swarm_id↔ manager_deployment_id;状态机 waiting_approval→running→…)。所有查询按 user 作用域,视图脱敏 (不含 plan/payload 大字段与凭据)。 测试:TestListSwarmCallbackEventsAfter(游标/过滤/空标识);TestMain 迁移 AgentDeployment + AgentCallbackEvent。go build/vet 干净,controller+model 全套回归通过。文档 §5.2。 Affects: Manager only(新增只读查询端点 + 一个 gated 写端点)。无计费/审计 schema 改动; 不依赖未冻结契约。stop 真实接入随 agent_swarm#2 冻结落地。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -289,6 +289,36 @@ signature = base64( ed25519_sign( device_priv, sha256(canonical) ) )
|
||||
|
||||
---
|
||||
|
||||
## 5.2 Swarm 运行查询 🟡(#45 Phase1 · 只读预览)
|
||||
|
||||
多 Agent 蜂群运行(`agent_swarm` / HeiCode Swarm)的**只读查询**。数据全部来自 HM 已持久化的运行时回调(Swarm → HM 带签名回调),**无需实时调 Swarm**,故不受 `agent_swarm#2` 契约冻结阻塞。
|
||||
|
||||
| 方法 | 路径 | 说明 | 状态 |
|
||||
|---|---|---|---|
|
||||
| GET | `/api/heicode/swarms` | 列出我的 swarm 运行 | 🟢 本地数据 |
|
||||
| GET | `/api/heicode/swarms/:id` | 单个运行状态(:id = deployment_id / swarm_id / correlation_id 任一) | 🟢 |
|
||||
| GET | `/api/heicode/swarms/:id/events?after=&limit=` | 事件增量拉取(`after`=上次返回的 `next_after`,oldest-first) | 🟢 |
|
||||
| GET | `/api/heicode/swarms/:id/artifacts` | 从已存事件派生的产物 | 🟢(派生) |
|
||||
| POST | `/api/heicode/swarms/:id/stop` | 停止运行(**写**) | 🟡 待 `agent_swarm#2` 冻结 + `SWARM_RUNTIME_ENABLED=true` |
|
||||
|
||||
```json
|
||||
// GET /api/heicode/swarms/:id
|
||||
{ "success": true, "data": {
|
||||
"deployment_id":"dep_…", "swarm_id":"swarm-…", "correlation_id":"…",
|
||||
"status":"running", "phase":"…", "runtime_state":"…", "failure_reason":"",
|
||||
"created_at":"…","updated_at":"…","runtime_last_sync_at":"…" }}
|
||||
|
||||
// GET /api/heicode/swarms/:id/events?after=120
|
||||
{ "success": true, "data": {
|
||||
"items":[ {"id":121,"event_type":"task.completed","task_id":"…","result":"ok","occurred_at":"…","payload":{…}} ],
|
||||
"next_after":121, "count":1 }}
|
||||
```
|
||||
|
||||
- 状态机(契约 §4):`waiting_approval → running →(blocked ⇄ running)→ completed/failed/stopped`。
|
||||
- `events` 用 `id` 游标(`next_after`)增量轮询;`stop` 在契约冻结前返回明确未启用提示(不臆造未冻结写接口)。
|
||||
|
||||
---
|
||||
|
||||
## 6. 直连 Agent(客户端 ↔ agent,A2A 协议)
|
||||
|
||||
> 客户端**直接连 agent 子域名**(§5 的 `subdomain`),不经过 HM。agent 是 AM 的 `coding_a2a_agent`,走 **A2A 协议**。
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"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,
|
||||
"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,
|
||||
}
|
||||
}
|
||||
|
||||
// swarmEventView maps a persisted callback event to the client view.
|
||||
func swarmEventView(e model.AgentCallbackEvent) gin.H {
|
||||
return gin.H{
|
||||
"id": e.Id, // 作为下一次 ?after= 的游标
|
||||
"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": unmarshalResourceJSON(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(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(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, swarmEventView(e))
|
||||
}
|
||||
}
|
||||
common.ApiSuccess(c, gin.H{"items": items, "total": len(items),
|
||||
"note": "derived from persisted runtime events; dedicated artifact contract pending agent_swarm#2 freeze"})
|
||||
}
|
||||
|
||||
// HeicodeStopSwarm: POST /api/heicode/swarms/:id/stop — 停止 swarm 运行(写操作)。
|
||||
// 这是唯一需要真正调用 Swarm 运行时的操作。在 agent_swarm#2 契约冻结 + SWARM_RUNTIME_ENABLED=true
|
||||
// 前默认关闭并明确提示(不臆造未冻结的写接口)。冻结后在此接入运行时 stop(草案路径
|
||||
// /api/agent/swarm/deployments/{id}/stop)。
|
||||
func HeicodeStopSwarm(c *gin.Context) {
|
||||
dep, ok := findUserSwarmDeployment(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if !common.GetEnvOrDefaultBool("SWARM_RUNTIME_ENABLED", false) {
|
||||
agentError(c, "POLICY_REJECTED",
|
||||
"swarm stop is not enabled yet: pending agent_swarm#2 runtime-contract freeze and SWARM_RUNTIME_ENABLED=true")
|
||||
return
|
||||
}
|
||||
// 契约冻结后在此调用 Swarm 运行时 stop;当前仅在本地记录意图,避免对未冻结写接口下注。
|
||||
common.ApiSuccess(c, gin.H{
|
||||
"deployment_id": dep.DeploymentID,
|
||||
"swarm_id": dep.RuntimeSwarmID,
|
||||
"accepted": true,
|
||||
"note": "runtime stop wiring lands with agent_swarm#2 contract freeze",
|
||||
})
|
||||
}
|
||||
@@ -55,6 +55,37 @@ func InsertAgentCallbackEvent(row *AgentCallbackEvent) (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// ListSwarmCallbackEventsAfter returns persisted swarm runtime events for one run
|
||||
// (matched by deployment_id OR swarm_id) with an id-based `after` cursor for
|
||||
// incremental polling (#45 events?after). Ordered oldest-first so the client can
|
||||
// append; the caller uses the last returned Id as the next `after`. Reading from
|
||||
// HM-persisted callback rows means this needs no live Swarm call.
|
||||
func ListSwarmCallbackEventsAfter(deploymentID, swarmID string, afterID, limit int) ([]AgentCallbackEvent, error) {
|
||||
if DB == nil {
|
||||
return nil, nil
|
||||
}
|
||||
q := DB.Model(&AgentCallbackEvent{})
|
||||
switch {
|
||||
case deploymentID != "" && swarmID != "":
|
||||
q = q.Where("deployment_id = ? OR swarm_id = ?", deploymentID, swarmID)
|
||||
case deploymentID != "":
|
||||
q = q.Where("deployment_id = ?", deploymentID)
|
||||
case swarmID != "":
|
||||
q = q.Where("swarm_id = ?", swarmID)
|
||||
default:
|
||||
return nil, nil
|
||||
}
|
||||
if afterID > 0 {
|
||||
q = q.Where("id > ?", afterID)
|
||||
}
|
||||
if limit <= 0 || limit > 1000 {
|
||||
limit = 200
|
||||
}
|
||||
var items []AgentCallbackEvent
|
||||
err := q.Order("id asc").Limit(limit).Find(&items).Error
|
||||
return items, err
|
||||
}
|
||||
|
||||
func ListAgentCallbackEvents(f ListAgentCallbackEventsFilter) ([]AgentCallbackEvent, error) {
|
||||
if DB == nil {
|
||||
return nil, nil
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// #45: events?after 游标 —— 按 deployment_id/swarm_id 过滤,id>after 增量返回,oldest-first。
|
||||
func TestListSwarmCallbackEventsAfter(t *testing.T) {
|
||||
require.NoError(t, LOG_DB.Where("1 = 1").Delete(&AgentCallbackEvent{}).Error)
|
||||
mk := func(eventID, dep, swarm, etype string) {
|
||||
_, err := InsertAgentCallbackEvent(&AgentCallbackEvent{
|
||||
EventID: eventID, DeploymentID: dep, SwarmID: swarm, EventType: etype,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
mk("e1", "dep_A", "swarm_A", "deployment.status_changed")
|
||||
mk("e2", "dep_A", "swarm_A", "task.created")
|
||||
mk("e3", "dep_A", "swarm_A", "artifact.produced")
|
||||
mk("e4", "dep_B", "swarm_B", "task.created") // 另一个 run,应被排除
|
||||
|
||||
// 全量(after=0):dep_A 的 3 条,oldest-first
|
||||
all, err := ListSwarmCallbackEventsAfter("dep_A", "swarm_A", 0, 100)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, all, 3)
|
||||
require.Equal(t, "e1", all[0].EventID)
|
||||
|
||||
// 游标:after = 第一条 id → 只返回其后的 2 条
|
||||
after := all[0].Id
|
||||
rest, err := ListSwarmCallbackEventsAfter("dep_A", "swarm_A", after, 100)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, rest, 2)
|
||||
require.Equal(t, "e2", rest[0].EventID)
|
||||
|
||||
// 仅按 swarm_id 也能查到
|
||||
bySwarm, err := ListSwarmCallbackEventsAfter("", "swarm_A", 0, 100)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, bySwarm, 3)
|
||||
|
||||
// 空标识 → 空
|
||||
none, err := ListSwarmCallbackEventsAfter("", "", 0, 100)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, none)
|
||||
}
|
||||
@@ -44,6 +44,8 @@ func TestMain(m *testing.M) {
|
||||
&SubscriptionOrder{},
|
||||
&UserSubscription{},
|
||||
&TelemetryEvent{},
|
||||
&AgentDeployment{},
|
||||
&AgentCallbackEvent{},
|
||||
); err != nil {
|
||||
panic("failed to migrate: " + err.Error())
|
||||
}
|
||||
|
||||
@@ -547,6 +547,13 @@ func SetApiRouter(router *gin.Engine) {
|
||||
// Client error-telemetry ingest (#24). Device-paired; never bills.
|
||||
// Gated by HEICODE_TELEMETRY_ENABLED (default off -> 410 kill switch).
|
||||
heicodeAgentRoute.POST("/telemetry/events", controller.HeicodeTelemetryEvents)
|
||||
// Swarm Run query (#45 Phase1). Read-only views served from HM-persisted
|
||||
// callback data (no live Swarm call); stop is gated pending agent_swarm#2 freeze.
|
||||
heicodeAgentRoute.GET("/swarms", controller.HeicodeListSwarms)
|
||||
heicodeAgentRoute.GET("/swarms/:id", controller.HeicodeGetSwarmStatus)
|
||||
heicodeAgentRoute.GET("/swarms/:id/events", controller.HeicodeListSwarmEvents)
|
||||
heicodeAgentRoute.GET("/swarms/:id/artifacts", controller.HeicodeListSwarmArtifacts)
|
||||
heicodeAgentRoute.POST("/swarms/:id/stop", controller.HeicodeStopSwarm)
|
||||
}
|
||||
|
||||
// Client↔agent access control (HM-provided, AM-optional). Called by the
|
||||
|
||||
Reference in New Issue
Block a user