diff --git a/docs/heicode-manager-standalone-execution-plan.md b/docs/heicode-manager-standalone-execution-plan.md index 86cf2f87..96869b14 100644 --- a/docs/heicode-manager-standalone-execution-plan.md +++ b/docs/heicode-manager-standalone-execution-plan.md @@ -12,7 +12,7 @@ |---|---| | 可以独立做 | Manager 自己的 DB 模型、Go API、前端页面、权限校验、回调接收骨架、artifact 数据模型、幂等、模拟事件和文档口径 | | 不能独立做 | 真实 `swarm_id`、worker claim / heartbeat、真实 handoff / retry、真实 Runtime 日志指标、客户端审批弹窗、短期凭证注入 runtime、SK 工具真实调用结果 | -| 当前最大问题 | Manager 已有本地 Agnet control-plane 占位，但普通用户态部署、任务桥接、回调/artifact/幂等和持久化还没形成完整 Manager 闭环 | +| 当前最大问题 | Manager 本地闭环已基本形成；剩余主要是等待真实 Runtime/客户端产生数据后联调验证 | ## 二、排除项 @@ -33,20 +33,20 @@ | 顺序 | 任务 | 优先级 | 是否 Manager 独立 | 完成后价值 | |---:|---|---|---|---| -| 1 | 增加 sub 模式字段 | P0 | 是 | 明确任务组织方式，支撑瀑布/敏捷状态展示 | -| 2 | 打通 HeicodeTask 到 Agnet deployment 的 Manager 桥接 | P0 | 是 | 不再只能手动创建 deployment | -| 3 | 增加用户态 Agnet deployment API | P0 | 是 | 普通用户可以在自己资源范围内创建/查看/停止 deployment | -| 4 | 统一 `/api/swarms` 与 `/api/agnet/deployments` 边界 | P0 | 是 | 为后续蜂群联调留稳定 adapter | -| 5 | 建 callback 接收端骨架 | P0 | 是 | 先接住事件、artifact、审批请求、usage、status | -| 6 | 建 artifact 数据模型和 API | P0 | 是 | 先把交付物/产物摘要落库并可展示 | -| 7 | 增加回调幂等和签名/服务身份校验骨架 | P1 | 是 | 重复回调不重复写入，生产联调不乱账 | -| 8 | 持久化 SK snapshot | P1 | 是 | 容器重启后任务上下文和审计不丢 | -| 9 | 任务视角审计聚合 | P1 | 是 | 按 task/deployment/correlation_id 看完整 Manager 记录 | -| 10 | 本地模拟蜂群事件冒烟入口 | P1 | 是 | 不等 Runtime，也能自测 Manager 端完整显示链路 | -| 11 | 前端页面补齐独立闭环展示 | P1 | 是 | 用户能看到任务、deployment、事件、artifact、审批和审计关系 | -| 12 | 文档口径清理 | P2 | 是 | 避免旧 Vault/OpenBao/Secret Provider 表述误导 | -| 13 | AWS/GCP 占位提示 | P2 | 是 | 避免用户误以为 AWS/GCP 已可用 | -| 14 | 请求 body 加密策略确认 | P2 | 是 | 明确 HTTPS 传输加密、应用层 body 加密是否需要、敏感字段禁止入参和日志 | +| 1 | 增加 sub 模式字段 | P0 | 是 | 已完成：明确任务组织方式，支撑瀑布/敏捷状态展示 | +| 2 | 打通 HeicodeTask 到 Agnet deployment 的 Manager 桥接 | P0 | 是 | 已完成：任务可生成 deployment draft | +| 3 | 增加用户态 Agnet deployment API | P0 | 是 | 已完成：普通用户可以在自己资源范围内创建/查看/停止 deployment | +| 4 | 统一 `/api/swarms` 与 `/api/agnet/deployments` 边界 | P0 | 是 | 已完成：为后续蜂群联调留稳定 adapter | +| 5 | 建 callback 接收端骨架 | P0 | 是 | 已完成：先接住事件、artifact、审批请求、timeline | +| 6 | 建 artifact 数据模型和 API | P0 | 是 | 已完成：交付物/产物摘要落库并可展示 | +| 7 | 增加回调幂等和签名/服务身份校验骨架 | P1 | 是 | 已完成：重复回调不重复写入，支持 token/HMAC | +| 8 | 持久化 SK snapshot | P1 | 是 | 已完成：容器重启后任务上下文和审计不丢 | +| 9 | 任务视角审计聚合 | P1 | 是 | 已完成：按 deployment 聚合 audit/callback/artifact/SK | +| 10 | 本地模拟蜂群事件冒烟入口 | P1 | 是 | 已完成：默认模拟 task/blocked/retry/handoff/artifact/approval/timeline | +| 11 | 前端页面补齐独立闭环展示 | P1 | 是 | 已完成：展示任务流、deployment、事件、artifact、审批和审计关系 | +| 12 | 文档口径清理 | P2 | 是 | 已完成：当前口径以 Azure Key Vault / secret_ref 为准 | +| 13 | AWS/GCP 占位提示 | P2 | 是 | 已完成：避免用户误以为 AWS/GCP 已可用 | +| 14 | 请求 body 加密策略确认 | P2 | 是 | 已完成：桌面端 sub POST 复用模型调用 V2 body 加密 | ## 四、任务明细 @@ -121,16 +121,17 @@ | 项 | 内容 | |---|---| | 目标 | Manager 先具备接收蜂群平台回调的 API 和落库能力 | -| 修改文件 | 新增 `heicode/model/agnet_callback.go`、`heicode/controller/agnet_callback.go`，更新 `heicode/router/api-router.go` | -| 新增接口 | `POST /api/agnet/callbacks/swarm-events`、`/approval-requests`、`/artifacts`、`/usage`、`/status` | -| 独立能力 | 本地模拟 payload 可保存、去重、查询 | -| 测试 | controller 单测覆盖正常保存、重复 event_id 幂等、明文密钥拒绝 | +| 修改文件 | `heicode/model/agnet_callback.go`、`heicode/controller/agnet_callback.go`、`heicode/router/api-router.go` | +| 新增接口 | `POST /api/agnet/callbacks/swarm-events` | +| 独立能力 | 本地模拟 payload 可保存、去重、查询；已补普通 sub 任务事件 schema 校验 | +| 测试 | controller 单测覆盖正常保存、重复 event_id 幂等、明文密钥拒绝、handoff 缺字段拒绝 | 验收标准： - 请求体不得出现 token/password/private_key/access_key/connection_string 明文字段。 - 每个回调都有 `event_id` 或 `idempotency_key`。 - 重复回调返回成功但不重复写入。 +- `task.*` / `handoff.*` 关键事件缺少必填字段时返回 `CALLBACK_SCHEMA_INVALID`。 ### 任务 6：artifact 数据模型和 API @@ -201,14 +202,15 @@ | 目标 | 在蜂群 Runtime 未接入前，Manager 能用模拟事件自测完整链路 | | 修改文件 | `heicode/controller/agnet_callback.go`、`heicode/router/api-router.go`、可选新增 `heicode/controller/agnet_smoke.go` | | 接口建议 | admin-only `POST /api/agnet/dev/simulate-run` | -| 生成内容 | deployment accepted、task.created、task.claimed、task.completed、artifact.created、approval.requested | -| 测试 | 单测验证模拟后 timeline/artifacts/audit 可查 | +| 生成内容 | deployment accepted、task.created、task.claimed、task.heartbeat、task.blocked、handoff.requested、handoff.completed、task.retried、task.completed、artifact.created、approval.requested、timeline.updated | +| 测试 | 单测验证模拟后 timeline/artifacts/audit 可查；`scripts/agnet_sub_mode_smoke.py` 可跑生产健康和 deployment timeline 冒烟 | 验收标准： - 该接口必须 admin-only 或 dev-only。 - 响应明确 `simulated: true`。 - 线上页面不能把模拟事件显示成真实 Runtime 事件。 +- task/blocked/retry/handoff 事件能在 timeline 中看到，来源标记为 `simulated`。 ### 任务 11：前端页面补齐独立闭环展示 diff --git a/docs/heicode-manager-swarm-gap-analysis.md b/docs/heicode-manager-swarm-gap-analysis.md index bf51e9b1..6e663af9 100644 --- a/docs/heicode-manager-swarm-gap-analysis.md +++ b/docs/heicode-manager-swarm-gap-analysis.md @@ -5,7 +5,7 @@ ## 一、对照结论 -Manager 端已经具备蜂群联调需要的控制面基础：用户态 deployment、task draft、`/api/swarms` adapter、Runtime 创建/停止桥接、callback 接收、artifact、approval、approval decision 回传 adapter、timeline、SK snapshot、V2 加密请求。本地 `simulate-events` 默认链路也能写入 callback、artifact、approval、timeline 记录，便于 Manager 自测页面展示和脱敏。 +Manager 端已经具备蜂群联调需要的控制面基础：用户态 deployment、task draft、`/api/swarms` adapter、Runtime 创建/停止桥接、callback 接收、artifact、approval、approval decision 回传 adapter、timeline、SK snapshot、V2 加密请求。本地 `simulate-events` 默认链路也能写入 task、blocked、retry、handoff、artifact、approval、timeline 记录，便于 Manager 自测页面展示和脱敏。 仍然缺的是“真实蜂群 Runtime 产生的数据和状态机”。也就是说，Manager 不是完全没写；缺口主要集中在 Runtime 真正创建 Swarm Run、任务图、Agent 执行、handoff、产物和审批继续/停止闭环。 @@ -21,8 +21,8 @@ Manager 端已经具备蜂群联调需要的控制面基础：用户态 deployme | 重复回调幂等 | 已按 `event_id` / `idempotency_key` 去重 | 满足 | 需要蜂群侧稳定传唯一事件 ID | 蜂群 | | 接收 artifact 回调 | `artifact.created` 可落库并查询 | 满足接收能力 | 真实 artifact schema 和文件/分支引用需 Runtime 输出 | 蜂群 | | 展示 Swarm 状态、事件、产物 | deployment detail/events/artifacts/timeline 已有，页面已强化 runtime/simulated 来源和 artifact 类型/URI | 基本满足 | 真实展示内容仍依赖 Runtime 回调真实数据 | 蜂群 | -| 展示 task graph、claim、heartbeat | 当前没有真实 task graph 表和状态展示 | 不满足 | 需要 Runtime 产出 task graph/claim/heartbeat 事件 | 蜂群 + Manager | -| 展示 handoff、blocked、retry | callback 可接任意 event，但没有专门展示和字段约束 | 部分满足 | 需要标准事件 schema 和 Runtime 实际事件 | 蜂群 + Manager | +| 展示 task graph、claim、heartbeat | Manager 已可接收并展示普通 task flow callback；没有独立 task graph 状态表 | 部分满足 | 真实任务图仍需要 Runtime 产出 task/agent 事件 | 蜂群 | +| 展示 handoff、blocked、retry | 已补 `task.*` / `handoff.*` schema 校验和页面任务流展示 | 基本满足接收和展示 | 真实数据仍需要 Runtime 持续回调 | 蜂群 | | 高危审批请求进入 Manager | `approval.requested` callback 可转审批记录；用户 approve/reject 后可按配置 POST 回 Runtime | 部分满足 | 还需要蜂群 Runtime 提供并验证 approval decision 接收接口 | Manager + 蜂群 | | 高危审批在客户端主体验完成 | Manager 有审批 API | 部分满足 | 桌面客户端要弹窗、轮询/订阅、提交决定 | 客户端 | | 短期凭证和长期密钥隔离 | Manager 使用 `secret_ref` / `lease://` 记录，不传明文 | 基本满足 | Runtime 侧短期凭证派生/注入未验证 | 蜂群 + 基础设施 | @@ -67,7 +67,7 @@ desktop-client -> Orchestrator(FastAPI) -> Redis -> Agent Pods |---|---|---| | 真实 `POST /api/swarms` 联调 | Manager 有 adapter，蜂群当前可见接口是 `/tasks` | 蜂群侧确认是否补 `/api/swarms`；Manager 固定 adapter 文档 | | `deployment_id <-> swarm_id` 真实映射 | Manager 字段已准备，真实值需 Runtime 返回 | Runtime create response 必须返回 `swarm_id` | -| task graph / claim / heartbeat 事件 | Manager 可接 callback，但没有真实数据 | 蜂群侧定义并回调 `task.created/claimed/heartbeat/released/completed/failed` | +| task graph / claim / heartbeat 事件 | Manager 可接收/展示 task flow callback；缺真实 Runtime 数据 | 蜂群侧定义并回调 `task.created/claimed/heartbeat/released/completed/failed` | | artifact 真实产出 | Manager 能落库，缺 Runtime 产出 | 蜂群侧回调 `artifact.created`，带 Git branch/commit 或存储 URI | | 审批结果回传 Runtime | Manager adapter 已完成，真实闭环未验证 | 蜂群侧提供接收接口并验证 approved/rejected 后继续或停止 | @@ -75,7 +75,7 @@ desktop-client -> Orchestrator(FastAPI) -> Redis -> Agent Pods | 缺失项 | 当前状态 | 处理建议 | |---|---|---| -| handoff/retry/blocked 展示 | 可接通用 callback，但页面未专门展示 | 先定事件 schema，再补 timeline 展示 | +| handoff/retry/blocked 展示 | Manager 已有 schema 校验、模拟事件和页面任务流展示 | 用真实 Runtime callback 做生产联调验证 | | Runtime 日志/指标 | Manager 有 logs/metrics 位置，缺真实数据源 | 蜂群侧提供日志摘要或查询接口；指标对齐 Prometheus | | 事件来源标识 | API 有部分 runtime/simulated 状态，页面还需强化 | Manager 页面区分 `manager`、`runtime`、`simulated` | | 桌面客户端审批主流程 | Manager 有 API，客户端未完成主体验 | 客户端按 Manager approval API 接入 | @@ -85,7 +85,7 @@ desktop-client -> Orchestrator(FastAPI) -> Redis -> Agent Pods | 缺失项 | 当前状态 | 处理建议 | |---|---|---| -| 蜂群验收脚本 | Manager 默认模拟和单测已覆盖 callback/artifact/approval/timeline；生产 curl 脚本仍需整理 | 增加一套可给联调方直接执行的 curl 脚本 | +| 蜂群验收脚本 | 已新增 `scripts/agnet_sub_mode_smoke.py`，可检查生产 Manager、Agent health 和指定 deployment timeline | 后续按真实 Runtime deployment 固化执行参数 | | 页面术语统一 | sub/蜂群容易混淆 | 页面和文档统一：sub 是任务组织，swarm 是执行层 | | 兼容 `HeiCode-Swarm` demo client 的说明 | 容易误认为正式 Heicode 桌面客户端 | 文档明确 demo client 不等于 `cc-haha` 正式客户端 | diff --git a/docs/integration/heicode-desktop-sub-agile-api.md b/docs/integration/heicode-desktop-sub-agile-api.md index 95d2ba52..1d434e4c 100644 --- a/docs/integration/heicode-desktop-sub-agile-api.md +++ b/docs/integration/heicode-desktop-sub-agile-api.md @@ -1221,7 +1221,7 @@ setInterval(async () => { ## 15. 当前生产注意事项 -1. `https://code.xinghanlab.com` 的 Manager 用户态接口已上线，当前生产版本为 `1.4.8`。 +1. `https://code.xinghanlab.com` 的 Manager 用户态接口已上线，当前生产版本为 `1.4.9`。 2. Manager 本地控制面可创建 `sub_mode=agile/waterfall` deployment。 3. 生产 Manager 已配置 Agent Manager Runtime，当前直接走 `http://20.212.121.126`；域名和 HTTPS 后续单独处理，不作为客户端当前接入阻塞项。 4. V2 加密 `deployment-draft` 已在生产验证通过：真实构造 `Content-Encoding: heicode-aead-v1` 请求返回 200，`sub_mode=agile`，`user_id=22`。 diff --git a/heicode/VERSION b/heicode/VERSION index b2e46d18..4ea2b1f4 100644 --- a/heicode/VERSION +++ b/heicode/VERSION @@ -1 +1 @@ -1.4.8 +1.4.9 diff --git a/heicode/controller/agnet_callback.go b/heicode/controller/agnet_callback.go index 7cd71898..bd04f124 100644 --- a/heicode/controller/agnet_callback.go +++ b/heicode/controller/agnet_callback.go @@ -328,6 +328,63 @@ func callbackIntValue(values map[string]any, key string) int { } } +var agnetCallbackEventRequiredFields = map[string][]string{ + "task.created": {"task_id", "title"}, + "task.claimed": {"task_id", "agent_role"}, + "task.running": {"task_id", "agent_role"}, + "task.heartbeat": {"task_id", "agent_role"}, + "task.blocked": {"task_id", "reason"}, + "task.retried": {"task_id", "attempt"}, + "task.released": {"task_id", "agent_role"}, + "task.failed": {"task_id", "reason"}, + "task.completed": {"task_id"}, + "handoff.requested": {"task_id", "from_role", "to_role"}, + "handoff.completed": {"task_id", "from_role", "to_role"}, + "artifact.created": {"artifact_id"}, +} + +func callbackEnvelopeFieldValue(payload agnetCallbackEnvelope, key string) string { + switch key { + case "task_id": + return strings.TrimSpace(payload.TaskID) + case "artifact_id": + return strings.TrimSpace(payload.Artifact.ArtifactID) + default: + return "" + } +} + +func callbackHasFieldValue(payload agnetCallbackEnvelope, key string) bool { + if callbackEnvelopeFieldValue(payload, key) != "" { + return true + } + if callbackStringValue(payload.Payload, key) != "" { + return true + } + if key == "artifact_id" { + if nested := callbackMapValue(payload.Payload, "artifact"); nested != nil { + return callbackStringValue(nested, key) != "" + } + } + if key == "attempt" { + return callbackIntValue(payload.Payload, key) > 0 + } + return false +} + +func validateAgnetCallbackEventSchema(payload agnetCallbackEnvelope) error { + required, ok := agnetCallbackEventRequiredFields[payload.EventType] + if !ok { + return nil + } + for _, key := range required { + if !callbackHasFieldValue(payload, key) { + return fmt.Errorf("%s is required for %s callback", key, payload.EventType) + } + } + return nil +} + func persistAgnetApprovalFromCallback(payload agnetCallbackEnvelope, record agnetDeploymentRecord) error { if payload.EventType != "approval.requested" || model.DB == nil { return nil @@ -412,6 +469,10 @@ func AgnetReceiveSwarmEventCallback(c *gin.Context) { if payload.IdempotencyKey == "" { payload.IdempotencyKey = payload.EventID } + if err := validateAgnetCallbackEventSchema(payload); err != nil { + agnetError(c, "CALLBACK_SCHEMA_INVALID", err.Error()) + return + } if agnetCallbackHasPlaintextSecret(payload) { agnetError(c, "CALLBACK_SECRET_REJECTED", "callbacks must not contain plaintext credential fields") return @@ -520,9 +581,10 @@ func timelineEntryFromCallback(callback model.AgnetCallbackEvent) gin.H { "swarm_id": callback.SwarmID, "agent_instance_id": callback.AgentInstanceID, "occurred_at": callback.OccurredAt, + "source": callback.Source, "payload": payload, } - for _, key := range []string{"title", "summary", "stage", "checkpoint", "severity", "next_action", "agent_role"} { + for _, key := range []string{"title", "summary", "stage", "checkpoint", "severity", "next_action", "agent_role", "task_id", "from_role", "to_role", "reason", "attempt"} { if value, ok := payload[key]; ok { entry[key] = value } diff --git a/heicode/controller/agnet_control_plane.go b/heicode/controller/agnet_control_plane.go index 89110616..10a8e2d5 100644 --- a/heicode/controller/agnet_control_plane.go +++ b/heicode/controller/agnet_control_plane.go @@ -1437,17 +1437,65 @@ func persistAgnetDefaultSimulationCallbacks(record agnetDeploymentRecord) error } } callbacks := []agnetCallbackEnvelope{ + base("task.created", "task-created", map[string]any{ + "title": "Simulated task created", + "summary": "Manager simulator created a child Agnet task for ordinary sub-mode smoke testing", + "agent_role": role, + "severity": "info", + "task_id": taskID, + }), base("task.claimed", "task-claimed", map[string]any{ "title": "Simulated task claimed", "summary": "Manager simulator marked a child Agnet task as claimed", "agent_role": role, "severity": "info", + "task_id": taskID, + }), + base("task.heartbeat", "task-heartbeat", map[string]any{ + "title": "Simulated task heartbeat", + "summary": "Manager simulator received a heartbeat from the child Agnet task", + "agent_role": role, + "severity": "info", + "task_id": taskID, + }), + base("task.blocked", "task-blocked", map[string]any{ + "title": "Simulated task blocked", + "summary": "Manager simulator marked a child Agnet task as blocked before handoff", + "agent_role": role, + "severity": "warning", + "task_id": taskID, + "reason": "Waiting for downstream sub-task handoff validation", + }), + base("handoff.requested", "handoff-requested", map[string]any{ + "title": "Simulated handoff requested", + "summary": "Manager simulator requested handoff between child Agnet roles", + "severity": "info", + "task_id": taskID, + "from_role": role, + "to_role": firstNonEmpty(grant.TargetRole, role), + }), + base("handoff.completed", "handoff-completed", map[string]any{ + "title": "Simulated handoff completed", + "summary": "Manager simulator completed handoff between child Agnet roles", + "severity": "success", + "task_id": taskID, + "from_role": role, + "to_role": firstNonEmpty(grant.TargetRole, role), + }), + base("task.retried", "task-retried", map[string]any{ + "title": "Simulated task retry", + "summary": "Manager simulator recorded a retry event for ordinary sub-mode timeline testing", + "agent_role": role, + "severity": "warning", + "task_id": taskID, + "attempt": 2, }), base("task.completed", "task-completed", map[string]any{ "title": "Simulated task completed", "summary": "Manager simulator marked the child Agnet task as completed", "agent_role": role, "severity": "success", + "task_id": taskID, }), } artifact := base("artifact.created", "artifact", map[string]any{ diff --git a/heicode/controller/agnet_control_plane_test.go b/heicode/controller/agnet_control_plane_test.go index 5df5060b..767a9379 100644 --- a/heicode/controller/agnet_control_plane_test.go +++ b/heicode/controller/agnet_control_plane_test.go @@ -835,6 +835,11 @@ func TestAgnetUserDeploymentDefaultSimulationCreatesRelatedRecords(t *testing.T) require.Contains(t, timelineRecorder.Body.String(), `"event_type":"artifact.created"`) require.Contains(t, timelineRecorder.Body.String(), `"event_type":"approval.requested"`) require.Contains(t, timelineRecorder.Body.String(), `"source":"agnet-simulator"`) + require.Contains(t, timelineRecorder.Body.String(), `"event_type":"task.created"`) + require.Contains(t, timelineRecorder.Body.String(), `"event_type":"task.blocked"`) + require.Contains(t, timelineRecorder.Body.String(), `"event_type":"handoff.requested"`) + require.Contains(t, timelineRecorder.Body.String(), `"event_type":"handoff.completed"`) + require.Contains(t, timelineRecorder.Body.String(), `"event_type":"task.retried"`) } func TestAgnetUserSwarmsAdapterCreatesScopedDeployment(t *testing.T) { @@ -864,6 +869,47 @@ func TestAgnetUserSwarmsAdapterCreatesScopedDeployment(t *testing.T) { require.Contains(t, recorder.Body.String(), `"user_id":"7"`) } +func TestAgnetCallbackRejectsInvalidTaskEventSchema(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + t.Setenv("AGNET_CALLBACK_TOKEN", "callback-token") + + plan := baseAgnetResourceGrantPlan() + plan.UserContext.UserID = "7" + for idx := range plan.Agents[0].ResourceGrants { + plan.Agents[0].ResourceGrants[idx].UserID = "7" + } + createRecorder, createEnvelope := postAgnetCreateUserDeployment(t, 7, plan) + require.Equal(t, http.StatusOK, createRecorder.Code) + require.True(t, createEnvelope.Success) + var createBody map[string]any + require.NoError(t, common.Unmarshal(createRecorder.Body.Bytes(), &createBody)) + deploymentID := createBody["data"].(map[string]any)["deployment_id"].(string) + + body := fmt.Sprintf(`{ + "event_id":"evt-invalid-handoff", + "event_type":"handoff.requested", + "deployment_id":%q, + "task_id":"task-1", + "payload":{ + "from_role":"backend" + } + }`, deploymentID) + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Request = httptest.NewRequest(http.MethodPost, "/api/agnet/callbacks/swarm-events", strings.NewReader(body)) + ctx.Request.Header.Set("Content-Type", "application/json") + ctx.Request.Header.Set("X-Agnet-Service-Token", "callback-token") + AgnetReceiveSwarmEventCallback(ctx) + + var envelope agnetCreateTestEnvelope + require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &envelope)) + require.False(t, envelope.Success) + require.Equal(t, "CALLBACK_SCHEMA_INVALID", envelope.Error.Code) + require.Contains(t, envelope.Error.Message, "to_role") +} + func TestAgnetCallbackStoresEventArtifactAndIsIdempotent(t *testing.T) { db := setupAgnetControlPlaneTestDB(t) resetAgnetControlPlaneState(t) diff --git a/heicode/web/default/src/features/agnet-console/pages.tsx b/heicode/web/default/src/features/agnet-console/pages.tsx index 6e0e7e94..52bbe0ff 100644 --- a/heicode/web/default/src/features/agnet-console/pages.tsx +++ b/heicode/web/default/src/features/agnet-console/pages.tsx @@ -297,6 +297,42 @@ function artifactTypeToneClass(value: unknown): string { return 'bg-muted/40 text-muted-foreground ring-border/60' } +function isTaskFlowEvent(eventType: unknown): boolean { + const event = String(eventType || '').toLowerCase() + return event.startsWith('task.') || event.startsWith('handoff.') +} + +function recordPayloadValue( + item: Record, + key: string +): unknown { + const payload = item.payload + if (payload && typeof payload === 'object' && !Array.isArray(payload)) { + const fromPayload = (payload as Record)[key] + if (fromPayload !== undefined && fromPayload !== null) { + return fromPayload + } + } + return item[key] +} + +function taskFlowDetail(item: Record): string { + const taskId = String(recordPayloadValue(item, 'task_id') || '') + const fromRole = String(recordPayloadValue(item, 'from_role') || '') + const toRole = String(recordPayloadValue(item, 'to_role') || '') + const reason = String(recordPayloadValue(item, 'reason') || '') + const attempt = String(recordPayloadValue(item, 'attempt') || '') + const agentRole = String(recordPayloadValue(item, 'agent_role') || '') + const parts = [ + taskId && `task ${taskId}`, + agentRole && `role ${agentRole}`, + fromRole && toRole && `${fromRole} -> ${toRole}`, + reason && `reason ${reason}`, + attempt && `attempt ${attempt}`, + ].filter(Boolean) + return parts.join(' / ') || '—' +} + function collectResourceGrants( dep: AgnetDeployment ): Record[] { @@ -963,6 +999,13 @@ function RunRelatedRecordsPanel({ deploymentId }: { deploymentId: string }) { const artifacts = data?.artifacts ?? [] const snapshots = data?.sk_snapshots ?? [] const timeline = data?.timeline ?? [] + const taskFlowRecords = useMemo( + () => + timeline.filter((item) => + isTaskFlowEvent(item.event_type || item.event) + ), + [timeline] + ) return (
@@ -1003,6 +1046,48 @@ function RunRelatedRecordsPanel({ deploymentId }: { deploymentId: string }) { />
+
+
+

+ {t('Sub task flow')} +

+ + {taskFlowRecords.length} + +
+ {taskFlowRecords.length === 0 ? ( +

+ {t('No task, retry or handoff callback records yet')} +

+ ) : ( +
    + {taskFlowRecords.slice(0, 8).map((item, idx) => { + const event = String(item.event_type || item.event || 'event') + const source = formatRecordSource(item.source) + const detail = taskFlowDetail(item) + return ( +
  1. +
    +

    + {event} +

    + + {source} + +
    +

    + {detail} +

    +
    2. + ) + })} +
+ )} +
+

diff --git a/scripts/agnet_sub_mode_smoke.py b/scripts/agnet_sub_mode_smoke.py new file mode 100755 index 00000000..1736e2b3 --- /dev/null +++ b/scripts/agnet_sub_mode_smoke.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +""" +Smoke test for Heicode Manager ordinary sub-mode / Agnet integration. + +Default mode checks public health/status endpoints only. To validate an +authenticated deployment flow, provide: + + HEICODE_ACCESS_TOKEN= + HEICODE_USER_ID= + HEICODE_DEPLOYMENT_ID= + +Optional: + HEICODE_RUN_SIMULATE=true + HEICODE_MANAGER_BASE_URL=https://code.xinghanlab.com + AGENT_MANAGER_HEALTH_URL=http://20.212.121.126/api/agnet/health +""" + +from __future__ import annotations + +import json +import os +import sys +import urllib.error +import urllib.request + + +BASE = os.getenv("HEICODE_MANAGER_BASE_URL", "https://code.xinghanlab.com").rstrip("/") +AGENT_HEALTH = os.getenv( + "AGENT_MANAGER_HEALTH_URL", "http://20.212.121.126/api/agnet/health" +) +ACCESS_TOKEN = os.getenv("HEICODE_ACCESS_TOKEN", "").strip() +USER_ID = os.getenv("HEICODE_USER_ID", "").strip() +DEPLOYMENT_ID = os.getenv("HEICODE_DEPLOYMENT_ID", "").strip() +RUN_SIMULATE = os.getenv("HEICODE_RUN_SIMULATE", "").strip().lower() in { + "1", + "true", + "yes", +} +UA = "heicode-agnet-sub-mode-smoke/1.0" +FAILS: list[str] = [] + + +def request( + method: str, + url: str, + headers: dict[str, str] | None = None, + body: bytes | None = None, +) -> tuple[int, dict[str, str], str]: + req = urllib.request.Request(url, method=method, data=body) + req.add_header("User-Agent", UA) + for key, value in (headers or {}).items(): + req.add_header(key, value) + try: + resp = urllib.request.urlopen(req, timeout=20) + return resp.status, dict(resp.headers), resp.read().decode("utf-8", "replace") + except urllib.error.HTTPError as exc: + return exc.code, dict(exc.headers), exc.read().decode("utf-8", "replace") + + +def manager_request( + method: str, path: str, *, auth: bool = False, payload: dict | None = None +) -> tuple[int, dict[str, str], str]: + headers = {"Content-Type": "application/json"} + if auth: + headers["Authorization"] = f"Bearer {ACCESS_TOKEN}" + headers["X-User-ID"] = USER_ID + body = json.dumps(payload).encode("utf-8") if payload is not None else None + return request(method, BASE + path, headers=headers, body=body) + + +def check(label: str, ok: bool, detail: str = "") -> None: + print(f"[{'OK' if ok else 'FAIL'}] {label}" + (f" -- {detail}" if detail else "")) + if not ok: + FAILS.append(label) + + +def parse_json(label: str, body: str) -> dict: + try: + parsed = json.loads(body) + check(label, isinstance(parsed, dict)) + return parsed if isinstance(parsed, dict) else {} + except json.JSONDecodeError as exc: + check(label, False, str(exc)) + return {} + + +print(f"Manager base: {BASE}") +print(f"Agent health: {AGENT_HEALTH}\n") + +code, _headers, body = manager_request("GET", "/api/status") +status = parse_json("Manager /api/status JSON parseable", body) +check("Manager /api/status success", code == 200 and status.get("success") is True, f"HTTP {code}") +version = status.get("data", {}).get("version") if isinstance(status.get("data"), dict) else "" +check("Manager status carries version", bool(version), f"version={version!r}") + +code, _headers, body = request("GET", AGENT_HEALTH) +agent = parse_json("Agent Manager health JSON parseable", body) +agent_data = agent.get("data", {}) if isinstance(agent.get("data"), dict) else {} +agent_status = agent_data.get("status") or agent.get("status") +check( + "Agent Manager health reachable", + code == 200 and str(agent_status).lower() in {"healthy", "ok", "up"}, + f"HTTP {code}, status={agent_status!r}", +) + +if not (ACCESS_TOKEN and USER_ID and DEPLOYMENT_ID): + print("\nAuthenticated deployment checks skipped: set HEICODE_ACCESS_TOKEN, HEICODE_USER_ID and HEICODE_DEPLOYMENT_ID.") +else: + if RUN_SIMULATE: + code, _headers, body = manager_request( + "POST", + f"/api/agnet/user/deployments/{DEPLOYMENT_ID}/simulate-events", + auth=True, + payload={}, + ) + simulated = parse_json("simulate-events JSON parseable", body) + check( + "simulate-events accepted", + code == 200 and simulated.get("success") is True, + f"HTTP {code}", + ) + + code, _headers, body = manager_request( + "GET", f"/api/agnet/user/deployments/{DEPLOYMENT_ID}", auth=True + ) + detail = parse_json("deployment detail JSON parseable", body) + detail_data = detail.get("data", {}) if isinstance(detail.get("data"), dict) else {} + check( + "deployment detail success", + code == 200 and detail.get("success") is True and detail_data.get("deployment_id") == DEPLOYMENT_ID, + f"HTTP {code}", + ) + + code, _headers, body = manager_request( + "GET", f"/api/agnet/user/deployments/{DEPLOYMENT_ID}/artifacts", auth=True + ) + artifacts = parse_json("artifacts JSON parseable", body) + artifacts_data = artifacts.get("data", {}) if isinstance(artifacts.get("data"), dict) else {} + artifact_items = artifacts_data.get("artifacts") or artifacts_data.get("items") or [] + check( + "artifacts endpoint success", + code == 200 and artifacts.get("success") is True and isinstance(artifact_items, list), + f"HTTP {code}, count={len(artifact_items) if isinstance(artifact_items, list) else 'n/a'}", + ) + + code, _headers, body = manager_request( + "GET", f"/api/agnet/user/deployments/{DEPLOYMENT_ID}/timeline", auth=True + ) + timeline = parse_json("timeline JSON parseable", body) + timeline_data = timeline.get("data", {}) if isinstance(timeline.get("data"), dict) else {} + callbacks = timeline_data.get("callbacks") or [] + merged = timeline_data.get("timeline") or [] + event_types = { + item.get("event_type") or item.get("event") + for item in merged + if isinstance(item, dict) + } + check( + "timeline endpoint success", + code == 200 and timeline.get("success") is True and isinstance(merged, list), + f"HTTP {code}, callbacks={len(callbacks) if isinstance(callbacks, list) else 'n/a'}", + ) + check( + "timeline includes ordinary sub-mode task flow when simulated/runtime callbacks exist", + any(str(event).startswith("task.") or str(event).startswith("handoff.") for event in event_types), + "events=" + ",".join(sorted(str(event) for event in event_types if event)[:10]), + ) + check( + "timeline includes artifact or approval flow", + "artifact.created" in event_types or "approval.requested" in event_types, + "events=" + ",".join(sorted(str(event) for event in event_types if event)[:10]), + ) + +if FAILS: + print(f"\n{len(FAILS)} check(s) failed:") + for item in FAILS: + print(f" - {item}") + sys.exit(1) + +print("\nALL AGNET SUB-MODE SMOKE CHECKS PASSED")