chenchenandClaude Opus 4.8
2c60af0216
agent_swarm PR#41 已合并 main(契约冻结),HM 补齐客户端+swarm 双方点名需要的
两条透传端点(均设备签名,挂 /api/heicode/swarms/:id 下):
- POST /swarms/:id/input → 代理 Swarm POST …/{deployment_id}/input(追加用户输入,
注入 source=user_append;终态 run 自动 reopen,stopped 拒绝)。写路,需
SWARM_RUNTIME_ENABLED+base+token,缺一即拒不伪造受理。**指令文本仅转发,绝不落
HM 日志/审计/事件**(#40 原文不进事件流 + #46 勿落日志);审计只记发生过一次 append。
- GET /swarms/:id/result → 代理 Swarm GET …/{deployment_id}/result,返回
{summary, deliverable, artifacts[], termination_reason, status};产物按 uri 取非内联。
读路,需 base+token(与派发开关解耦);透出前 stripSensitiveKeys 递归剔除 secret_ref 等兜底。
新增:callSwarmRuntimeJSON(通用 JSON 调用 + envelope 归一,错误信息不含请求体)、
agentRuntimeSwarm{Input,Result}Path(默认 create-path 基路径,SWARM_RUNTIME_{INPUT,RESULT}_PATH 可覆盖)。
测试:swarm_io_proxy_test.go 覆盖路径构造(默认/空回退/env 覆盖)、input 转发(方法/路径/鉴权/body)、
result 解析 + deliverable/artifacts 脱敏、运行时非 2xx 报错。go build ./... + controller/router 全绿。
影响面:Manager ✅(透传端点)/ agent_swarm ✅(消费 PR#41 契约)/ Client ✅(驾驶舱 I/O 闭环)/
Agent ❌ / CodeGW ❌ / 计费 ❌ / 密钥 ❌(不经手 secret)/ 审计 ✅(input_appended,不含原文)/ 发布链路 ❌
Refs #46 #28。Depends-on-contract: agent_swarm PR#41 (已合并 main)。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1400 lines
50 KiB
Go
1400 lines
50 KiB
Go
package controller
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/heicode/manager/common"
|
|
"github.com/heicode/manager/model"
|
|
)
|
|
|
|
const (
|
|
agentRuntimeStateSyncing = "runtime_syncing"
|
|
agentRuntimeStateSynced = "runtime_accepted"
|
|
agentRuntimeStateFailed = "runtime_sync_failed"
|
|
|
|
agentRuntimeModeAgent = "agent"
|
|
agentRuntimeModeSwarm = "swarm"
|
|
)
|
|
|
|
type agentRuntimeConfig struct {
|
|
Enabled bool
|
|
Async bool
|
|
BaseURL string
|
|
Token string
|
|
CreatePath string
|
|
HealthPath string
|
|
StatusPath string
|
|
ArtifactContentPath string
|
|
StopPath string
|
|
ApprovalDecisionPath string
|
|
Timeout time.Duration
|
|
}
|
|
|
|
type agentRuntimeSyncResult struct {
|
|
RuntimeDeploymentID string
|
|
RuntimeSwarmID string
|
|
RuntimeStatus string
|
|
RawStatusCode int
|
|
}
|
|
|
|
type agentRuntimeDiagnostics struct {
|
|
DeploymentID string `json:"deployment_id"`
|
|
RuntimeMode string `json:"runtime_mode"`
|
|
SubMode string `json:"sub_mode"`
|
|
RuntimeDeploymentID string `json:"runtime_deployment_id,omitempty"`
|
|
RuntimeSwarmID string `json:"runtime_swarm_id,omitempty"`
|
|
DataSource string `json:"data_source"`
|
|
HTTPStatus int `json:"http_status,omitempty"`
|
|
Status string `json:"status,omitempty"`
|
|
Phase string `json:"phase,omitempty"`
|
|
Progress any `json:"progress,omitempty"`
|
|
ErrorMessage string `json:"error_message,omitempty"`
|
|
Agents []gin.H `json:"agents"`
|
|
Artifacts []gin.H `json:"artifacts"`
|
|
Metrics map[string]any `json:"metrics,omitempty"`
|
|
Warnings []string `json:"warnings"`
|
|
CheckedAt string `json:"checked_at"`
|
|
}
|
|
|
|
func normalizeAgentRuntimeMode(value string) string {
|
|
switch strings.ToLower(strings.TrimSpace(value)) {
|
|
case agentRuntimeModeSwarm:
|
|
return agentRuntimeModeSwarm
|
|
default:
|
|
return agentRuntimeModeAgent
|
|
}
|
|
}
|
|
|
|
func agentRuntimeModeForSource(source string) string {
|
|
if strings.TrimSpace(source) == "api_swarms_adapter" {
|
|
return agentRuntimeModeSwarm
|
|
}
|
|
return agentRuntimeModeAgent
|
|
}
|
|
|
|
func agentRuntimeModeForRecord(record agentDeploymentRecord) string {
|
|
return normalizeAgentRuntimeMode(record.Plan.Metadata.RuntimeMode)
|
|
}
|
|
|
|
func agentRuntimeClientConfig() agentRuntimeConfig {
|
|
return agentRuntimeClientConfigForMode(agentRuntimeModeAgent)
|
|
}
|
|
|
|
func agentRuntimeClientConfigForMode(mode string) agentRuntimeConfig {
|
|
timeoutSec := common.GetEnvOrDefault("AGENT_RUNTIME_TIMEOUT_SECONDS", 5)
|
|
if timeoutSec <= 0 {
|
|
timeoutSec = 5
|
|
}
|
|
mode = normalizeAgentRuntimeMode(mode)
|
|
prefix := "AGENT_RUNTIME_"
|
|
// Sub Agile primary route per agent_management Sub Mode Runtime §2.2.
|
|
defaultCreatePath := "/api/agent/sub-agile/deployments"
|
|
defaultStopPath := "/api/agent/sub-agile/deployments/{deployment_id}/stop"
|
|
defaultApprovalPath := "/api/swarms/{swarm_id}/approvals/{approval_id}"
|
|
defaultStatusPath := "/api/swarms/{swarm_id}/status"
|
|
defaultArtifactContentPath := "/api/swarms/{swarm_id}/artifacts/{artifact_id}/content"
|
|
if mode == agentRuntimeModeSwarm {
|
|
prefix = "SWARM_RUNTIME_"
|
|
// HeiCode-Swarm standard routes (蜂群对接文档 §3.1). The runtime has no
|
|
// /status subpath — deployment status is read from the deployment detail
|
|
// endpoint. {deployment_id} resolves to the runtime deployment id.
|
|
// Legacy /api/swarms/* still works on the runtime but is no longer the
|
|
// Manager default. (sub-agile defaults above are untouched.)
|
|
defaultCreatePath = "/api/agent/swarm/deployments"
|
|
defaultStopPath = "/api/agent/swarm/deployments/{deployment_id}/stop"
|
|
defaultApprovalPath = "/api/agent/swarm/deployments/{deployment_id}/approvals/{approval_id}"
|
|
defaultStatusPath = "/api/agent/swarm/deployments/{deployment_id}"
|
|
if swarmTimeout := common.GetEnvOrDefault("SWARM_RUNTIME_TIMEOUT_SECONDS", timeoutSec); swarmTimeout > 0 {
|
|
timeoutSec = swarmTimeout
|
|
}
|
|
}
|
|
baseURL := strings.TrimRight(strings.TrimSpace(common.GetEnvOrDefaultString(prefix+"BASE_URL", "")), "/")
|
|
enabledDefault := false
|
|
if mode == agentRuntimeModeAgent {
|
|
enabledDefault = common.GetEnvOrDefaultBool("AGENT_RUNTIME_ENABLED", false)
|
|
} else {
|
|
enabledDefault = baseURL != ""
|
|
}
|
|
return agentRuntimeConfig{
|
|
Enabled: common.GetEnvOrDefaultBool(prefix+"ENABLED", enabledDefault),
|
|
Async: common.GetEnvOrDefaultBool(prefix+"ASYNC", common.GetEnvOrDefaultBool("AGENT_RUNTIME_ASYNC", true)),
|
|
BaseURL: baseURL,
|
|
Token: strings.TrimSpace(common.GetEnvOrDefaultString(prefix+"SERVICE_TOKEN", "")),
|
|
CreatePath: common.GetEnvOrDefaultString(prefix+"CREATE_PATH", defaultCreatePath),
|
|
HealthPath: common.GetEnvOrDefaultString(prefix+"HEALTH_PATH", "/api/agent/health"),
|
|
StatusPath: common.GetEnvOrDefaultString(prefix+"STATUS_PATH", defaultStatusPath),
|
|
ArtifactContentPath: common.GetEnvOrDefaultString(prefix+"ARTIFACT_CONTENT_PATH", defaultArtifactContentPath),
|
|
StopPath: common.GetEnvOrDefaultString(prefix+"STOP_PATH", defaultStopPath),
|
|
ApprovalDecisionPath: common.GetEnvOrDefaultString(prefix+"APPROVAL_DECISION_PATH", defaultApprovalPath),
|
|
Timeout: time.Duration(timeoutSec) * time.Second,
|
|
}
|
|
}
|
|
|
|
func agentRuntimeCallbackURL() string {
|
|
if value := strings.TrimSpace(common.GetEnvOrDefaultString("AGENT_RUNTIME_CALLBACK_URL", "")); value != "" {
|
|
return value
|
|
}
|
|
baseURL := strings.TrimRight(strings.TrimSpace(common.GetEnvOrDefaultString("HEICODE_PUBLIC_BASE_URL", "https://code.xinghanlab.com")), "/")
|
|
return baseURL + "/api/agent/callbacks/runtime-events"
|
|
}
|
|
|
|
func agentRuntimeCallbackSigningSecretRef() string {
|
|
return strings.TrimSpace(common.GetEnvOrDefaultString("AGENT_RUNTIME_CALLBACK_SIGNING_SECRET_REF", ""))
|
|
}
|
|
|
|
func agentRuntimeSubscribedEvents() []string {
|
|
return []string{
|
|
"deployment.status_changed",
|
|
"phase.changed",
|
|
"agent.started",
|
|
"agent.completed",
|
|
"agent.crashed",
|
|
"sk_tool.called",
|
|
"sk_tool.completed",
|
|
"sk_tool.failed",
|
|
"approval.requested",
|
|
"budget.alert",
|
|
"artifact.created",
|
|
"timeline.updated",
|
|
"task.created",
|
|
"task.claimed",
|
|
"task.running",
|
|
"task.heartbeat",
|
|
"task.blocked",
|
|
"task.retried",
|
|
"task.failed",
|
|
"task.completed",
|
|
"handoff.requested",
|
|
"handoff.completed",
|
|
}
|
|
}
|
|
|
|
func firstNonEmpty(values ...string) string {
|
|
for _, value := range values {
|
|
if strings.TrimSpace(value) != "" {
|
|
return strings.TrimSpace(value)
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func truncateAgentFailureReason(value string) string {
|
|
value = strings.TrimSpace(value)
|
|
if len(value) <= 480 {
|
|
return value
|
|
}
|
|
return value[:480]
|
|
}
|
|
|
|
func agentRuntimeURL(baseURL string, path string) (string, error) {
|
|
if strings.TrimSpace(baseURL) == "" {
|
|
return "", errors.New("AGENT_RUNTIME_BASE_URL is not configured")
|
|
}
|
|
parsed, err := url.Parse(baseURL)
|
|
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
|
|
return "", errors.New("AGENT_RUNTIME_BASE_URL must be an absolute http(s) URL")
|
|
}
|
|
if strings.TrimSpace(path) == "" {
|
|
path = "/"
|
|
}
|
|
if !strings.HasPrefix(path, "/") {
|
|
path = "/" + path
|
|
}
|
|
return strings.TrimRight(baseURL, "/") + path, nil
|
|
}
|
|
|
|
func agentRuntimeHeaders(req *http.Request, cfg agentRuntimeConfig, record agentDeploymentRecord) {
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("X-User-ID", record.Plan.UserContext.UserID)
|
|
req.Header.Set("X-Binding-Scope", firstPlanBindingScope(record.Plan))
|
|
req.Header.Set("X-Correlation-ID", record.Plan.Metadata.CorrelationID)
|
|
req.Header.Set("X-Idempotency-Key", "manager-"+record.DeploymentID)
|
|
if cfg.Token != "" {
|
|
req.Header.Set("Authorization", "Bearer "+cfg.Token)
|
|
}
|
|
}
|
|
|
|
func agentRuntimeRequestAgents(plan agentOrchestrationPlan) []gin.H {
|
|
byRole := make(map[string]agentAgentPlan, len(plan.Agents))
|
|
for _, agent := range plan.Agents {
|
|
role := strings.TrimSpace(agent.RoleTemplate)
|
|
if role != "" {
|
|
byRole[role] = agent
|
|
}
|
|
}
|
|
|
|
items := make([]gin.H, 0, len(plan.AgentRuntime.Agents))
|
|
for _, runtimeAgent := range plan.AgentRuntime.Agents {
|
|
role := strings.TrimSpace(runtimeAgent.Role)
|
|
if role == "" {
|
|
continue
|
|
}
|
|
item := gin.H{"role": role}
|
|
if agent, ok := byRole[role]; ok {
|
|
if len(agent.SKSources) > 0 {
|
|
item["sk_sources"] = agent.SKSources
|
|
}
|
|
if len(agent.ResourceGrants) > 0 {
|
|
item["resource_grants"] = agentRuntimeResourceGrantPayloads(agent.ResourceGrants)
|
|
}
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
|
|
if len(items) > 0 {
|
|
return items
|
|
}
|
|
for _, agent := range plan.Agents {
|
|
role := strings.TrimSpace(agent.RoleTemplate)
|
|
if role == "" {
|
|
continue
|
|
}
|
|
item := gin.H{"role": role}
|
|
if len(agent.SKSources) > 0 {
|
|
item["sk_sources"] = agent.SKSources
|
|
}
|
|
if len(agent.ResourceGrants) > 0 {
|
|
item["resource_grants"] = agentRuntimeResourceGrantPayloads(agent.ResourceGrants)
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items
|
|
}
|
|
|
|
func agentRuntimeRequestSwarmAgents(plan agentOrchestrationPlan) []gin.H {
|
|
runtimeModels := make(map[string]string, len(plan.AgentRuntime.Agents))
|
|
for _, runtimeAgent := range plan.AgentRuntime.Agents {
|
|
role := strings.TrimSpace(runtimeAgent.Role)
|
|
if role != "" && strings.TrimSpace(runtimeAgent.ModelRef) != "" {
|
|
runtimeModels[role] = strings.TrimSpace(runtimeAgent.ModelRef)
|
|
}
|
|
}
|
|
|
|
items := make([]gin.H, 0, len(plan.Agents))
|
|
for index, agent := range plan.Agents {
|
|
role := strings.TrimSpace(agent.RoleTemplate)
|
|
if role == "" {
|
|
continue
|
|
}
|
|
taskID := fmt.Sprintf("%s-%d", sanitizeAgentRef(role), index+1)
|
|
item := gin.H{
|
|
"task_id": taskID,
|
|
"role": role,
|
|
"title": role + " task",
|
|
"description": firstNonEmpty(agent.Goal, "Execute the Heicode swarm task as "+role),
|
|
"depends_on": []string{},
|
|
}
|
|
if modelRef := firstNonEmpty(runtimeModels[role], agent.DefaultModelID); modelRef != "" {
|
|
item["model_ref"] = modelRef
|
|
}
|
|
if len(agent.SKSources) > 0 {
|
|
item["sk_sources"] = agent.SKSources
|
|
}
|
|
if len(agent.ResourceGrants) > 0 {
|
|
item["resource_grants"] = agentRuntimeResourceGrantPayloads(agent.ResourceGrants)
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items
|
|
}
|
|
|
|
func agentRuntimeResourceGrantPayloads(grants []agentResourceGrant) []gin.H {
|
|
items := make([]gin.H, 0, len(grants))
|
|
for _, grant := range grants {
|
|
secretRef := strings.TrimSpace(grant.SecretRef)
|
|
if secretRef == "" {
|
|
continue
|
|
}
|
|
item := gin.H{
|
|
"grant_id": grant.GrantID,
|
|
"resource_id": grant.ResourceID,
|
|
"resource_type": grant.ResourceType,
|
|
"type": grant.ResourceType,
|
|
"user_id": grant.UserID,
|
|
"binding_scope": grant.BindingScope,
|
|
"target_role": grant.TargetRole,
|
|
"target_agent_ref": grant.TargetAgentRef,
|
|
"permission_scope": grant.PermissionScope,
|
|
"permissions": grant.PermissionScope,
|
|
"status": grant.Status,
|
|
"ref": secretRef,
|
|
"secret_ref": secretRef,
|
|
"allowed_ref": agentGrantResourceRef(grant),
|
|
"resource_ref_hint": agentGrantResourceRef(grant),
|
|
}
|
|
if grant.TenantID != "" {
|
|
item["tenant_id"] = grant.TenantID
|
|
}
|
|
if grant.ProjectID != "" {
|
|
item["project_id"] = grant.ProjectID
|
|
}
|
|
if len(grant.Constraints) > 0 {
|
|
item["constraints"] = grant.Constraints
|
|
}
|
|
if len(grant.Metadata) > 0 {
|
|
item["metadata"] = grant.Metadata
|
|
}
|
|
if len(grant.Audit) > 0 {
|
|
item["audit"] = grant.Audit
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items
|
|
}
|
|
|
|
func agentRuntimeRequestResourceGrants(plan agentOrchestrationPlan) []gin.H {
|
|
items := make([]agentResourceGrant, 0, len(plan.ResourceGrants))
|
|
items = append(items, plan.ResourceGrants...)
|
|
for _, agent := range plan.Agents {
|
|
items = append(items, agent.ResourceGrants...)
|
|
}
|
|
return agentRuntimeResourceGrantPayloads(items)
|
|
}
|
|
|
|
func agentRuntimeRequestMetadata(record agentDeploymentRecord, source string) gin.H {
|
|
metadata := gin.H{
|
|
"correlation_id": record.Plan.Metadata.CorrelationID,
|
|
"manager_deployment_id": record.DeploymentID,
|
|
"source": source,
|
|
"heicode_deployment_id": record.DeploymentID,
|
|
"heicode_runtime_bridge": true,
|
|
"runtime_mode": agentRuntimeModeForRecord(record),
|
|
}
|
|
if record.Plan.Metadata.TenantID != "" {
|
|
metadata["tenant_id"] = record.Plan.Metadata.TenantID
|
|
}
|
|
if record.Plan.Metadata.ProjectID != "" {
|
|
metadata["project_id"] = record.Plan.Metadata.ProjectID
|
|
}
|
|
return metadata
|
|
}
|
|
|
|
func agentRuntimeBudgetPayload(budget agentBudget) gin.H {
|
|
return gin.H{
|
|
"max_tokens": budget.MaxTokens,
|
|
"token_limit": budget.MaxTokens,
|
|
"max_cost_usd": budget.MaxCostUSD,
|
|
"max_duration_sec": budget.MaxDurationSec,
|
|
"duration_seconds": budget.MaxDurationSec,
|
|
"max_duration_seconds": budget.MaxDurationSec,
|
|
}
|
|
}
|
|
|
|
func agentRuntimeOrchestrationPlanPayload(record agentDeploymentRecord) any {
|
|
if agentRuntimeModeForRecord(record) != agentRuntimeModeSwarm {
|
|
return record.Plan
|
|
}
|
|
plan := record.Plan
|
|
return gin.H{
|
|
"intent_id": plan.IntentID,
|
|
"template_hint": plan.TemplateHint,
|
|
"objective": plan.Objective,
|
|
"sub_mode": firstNonEmpty(plan.SubMode, "goal_driven_swarm"),
|
|
"risk_level": plan.RiskLevel,
|
|
"budget": agentRuntimeBudgetPayload(plan.Budget),
|
|
"user_context": plan.UserContext,
|
|
"billing_context": plan.BillingContext,
|
|
"agile_context": plan.AgileContext,
|
|
"agents": agentRuntimeRequestSwarmAgents(plan),
|
|
"resource_grants": agentRuntimeRequestResourceGrants(plan),
|
|
"constraints": plan.Constraints,
|
|
"metadata": agentRuntimeRequestMetadata(record, "orchestration_plan"),
|
|
"agent_runtime": plan.AgentRuntime,
|
|
"acceptance": plan.AgileContext.AcceptanceCriteria,
|
|
"acceptance_tests": plan.AgileContext.AcceptanceCriteria,
|
|
}
|
|
}
|
|
|
|
func agentRuntimeCreatePayload(record agentDeploymentRecord, source string) gin.H {
|
|
callback := gin.H{
|
|
"url": agentRuntimeCallbackURL(),
|
|
"subscribed_events": agentRuntimeSubscribedEvents(),
|
|
}
|
|
if ref := agentRuntimeCallbackSigningSecretRef(); ref != "" {
|
|
callback["signing_secret_ref"] = ref
|
|
}
|
|
return gin.H{
|
|
"orchestration_plan": agentRuntimeOrchestrationPlanPayload(record),
|
|
"agents": agentRuntimeRequestAgents(record.Plan),
|
|
"risk_level": record.Plan.RiskLevel,
|
|
"budget": agentRuntimeBudgetPayload(record.Plan.Budget),
|
|
"billing_context": record.Plan.BillingContext,
|
|
"resource_grants": agentRuntimeRequestResourceGrants(record.Plan),
|
|
"callback": callback,
|
|
"agile_context": record.Plan.AgileContext,
|
|
"sub_mode": record.Plan.SubMode,
|
|
"metadata": agentRuntimeRequestMetadata(record, source),
|
|
}
|
|
}
|
|
|
|
func extractAgentRuntimeData(payload map[string]any) map[string]any {
|
|
if data, ok := payload["data"].(map[string]any); ok {
|
|
return data
|
|
}
|
|
return payload
|
|
}
|
|
|
|
func agentRuntimeEnvelopeError(payload map[string]any) string {
|
|
success, hasSuccess := payload["success"].(bool)
|
|
if !hasSuccess || success {
|
|
return ""
|
|
}
|
|
if errPayload, ok := payload["error"].(map[string]any); ok {
|
|
return firstNonEmpty(stringFromMap(errPayload, "message"), stringFromMap(errPayload, "code"), "runtime returned success=false")
|
|
}
|
|
return "runtime returned success=false"
|
|
}
|
|
|
|
func stringFromMap(values map[string]any, keys ...string) string {
|
|
for _, key := range keys {
|
|
if value, ok := values[key]; ok {
|
|
switch typed := value.(type) {
|
|
case string:
|
|
if strings.TrimSpace(typed) != "" {
|
|
return strings.TrimSpace(typed)
|
|
}
|
|
case fmt.Stringer:
|
|
if strings.TrimSpace(typed.String()) != "" {
|
|
return strings.TrimSpace(typed.String())
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func callAgentRuntimeCreate(ctx context.Context, cfg agentRuntimeConfig, record agentDeploymentRecord, source string) (agentRuntimeSyncResult, error) {
|
|
endpoint, err := agentRuntimeURL(cfg.BaseURL, cfg.CreatePath)
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
payload, err := common.Marshal(agentRuntimeCreatePayload(record, source))
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
agentRuntimeHeaders(req, cfg, record)
|
|
client := &http.Client{Timeout: cfg.Timeout}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
defer resp.Body.Close()
|
|
body, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
if readErr != nil {
|
|
return agentRuntimeSyncResult{}, readErr
|
|
}
|
|
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
|
|
return agentRuntimeSyncResult{RawStatusCode: resp.StatusCode}, fmt.Errorf("runtime create returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
|
|
}
|
|
var envelope map[string]any
|
|
if len(body) > 0 {
|
|
if err := common.Unmarshal(body, &envelope); err != nil {
|
|
return agentRuntimeSyncResult{RawStatusCode: resp.StatusCode}, err
|
|
}
|
|
}
|
|
if message := agentRuntimeEnvelopeError(envelope); message != "" {
|
|
return agentRuntimeSyncResult{RawStatusCode: resp.StatusCode}, errors.New(message)
|
|
}
|
|
data := extractAgentRuntimeData(envelope)
|
|
result := agentRuntimeSyncResult{
|
|
RuntimeDeploymentID: stringFromMap(data, "runtime_deployment_id", "deployment_id", "id"),
|
|
RuntimeSwarmID: stringFromMap(data, "swarm_id", "runtime_swarm_id"),
|
|
RuntimeStatus: stringFromMap(data, "runtime_status", "status"),
|
|
RawStatusCode: resp.StatusCode,
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func agentRuntimeStopPath(cfg agentRuntimeConfig, runtimeDeploymentID string) string {
|
|
path := strings.TrimSpace(cfg.StopPath)
|
|
if path == "" {
|
|
path = "/api/agent/deployments/{deployment_id}/stop"
|
|
}
|
|
return strings.ReplaceAll(path, "{deployment_id}", url.PathEscape(runtimeDeploymentID))
|
|
}
|
|
|
|
// agentRuntimeSwarmInputPath / agentRuntimeSwarmResultPath 构造 swarm 运行时的追加输入 / 结果
|
|
// 端点(agent_swarm PR#41 runtime-contract:POST/GET …/{deployment_id}/{input,result})。默认沿用
|
|
// create-path 的 /api/agent/swarm/deployments 基路径,可经 SWARM_RUNTIME_{INPUT,RESULT}_PATH 覆盖。
|
|
func agentRuntimeSwarmInputPath(cfg agentRuntimeConfig, runtimeDeploymentID string) string {
|
|
return swarmRuntimeDeploymentSubPath(cfg, runtimeDeploymentID, "input", "SWARM_RUNTIME_INPUT_PATH")
|
|
}
|
|
|
|
func agentRuntimeSwarmResultPath(cfg agentRuntimeConfig, runtimeDeploymentID string) string {
|
|
return swarmRuntimeDeploymentSubPath(cfg, runtimeDeploymentID, "result", "SWARM_RUNTIME_RESULT_PATH")
|
|
}
|
|
|
|
func swarmRuntimeDeploymentSubPath(cfg agentRuntimeConfig, runtimeDeploymentID, sub, envKey string) string {
|
|
path := strings.TrimSpace(common.GetEnvOrDefaultString(envKey, ""))
|
|
if path == "" {
|
|
base := strings.TrimRight(strings.TrimSpace(cfg.CreatePath), "/")
|
|
if base == "" {
|
|
base = "/api/agent/swarm/deployments"
|
|
}
|
|
path = base + "/{deployment_id}/" + sub
|
|
}
|
|
return strings.ReplaceAll(path, "{deployment_id}", url.PathEscape(strings.TrimSpace(runtimeDeploymentID)))
|
|
}
|
|
|
|
func agentRuntimeStopPathForRecord(cfg agentRuntimeConfig, record agentDeploymentRecord) string {
|
|
// {deployment_id} falls back to the swarm id so sub-agile-style paths still
|
|
// resolve for records that only persisted a runtime swarm id.
|
|
path := agentRuntimeStopPath(cfg, firstNonEmpty(record.RuntimeDeploymentID, record.RuntimeSwarmID))
|
|
replacer := strings.NewReplacer(
|
|
"{swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{runtime_swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{runtime_deployment_id}", url.PathEscape(strings.TrimSpace(record.RuntimeDeploymentID)),
|
|
"{manager_deployment_id}", url.PathEscape(strings.TrimSpace(record.DeploymentID)),
|
|
)
|
|
return replacer.Replace(path)
|
|
}
|
|
|
|
func agentRuntimeStatusPathForRecord(cfg agentRuntimeConfig, record agentDeploymentRecord) string {
|
|
path := strings.TrimSpace(cfg.StatusPath)
|
|
if path == "" {
|
|
path = "/api/swarms/{swarm_id}/status"
|
|
}
|
|
replacer := strings.NewReplacer(
|
|
"{swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{runtime_swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{deployment_id}", url.PathEscape(strings.TrimSpace(record.RuntimeDeploymentID)),
|
|
"{runtime_deployment_id}", url.PathEscape(strings.TrimSpace(record.RuntimeDeploymentID)),
|
|
"{manager_deployment_id}", url.PathEscape(strings.TrimSpace(record.DeploymentID)),
|
|
)
|
|
return replacer.Replace(path)
|
|
}
|
|
|
|
func agentRuntimeArtifactContentPathForRecord(cfg agentRuntimeConfig, record agentDeploymentRecord, artifactID string) string {
|
|
path := strings.TrimSpace(cfg.ArtifactContentPath)
|
|
if path == "" {
|
|
path = "/api/swarms/{swarm_id}/artifacts/{artifact_id}/content"
|
|
}
|
|
replacer := strings.NewReplacer(
|
|
"{swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{runtime_swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{deployment_id}", url.PathEscape(strings.TrimSpace(record.RuntimeDeploymentID)),
|
|
"{runtime_deployment_id}", url.PathEscape(strings.TrimSpace(record.RuntimeDeploymentID)),
|
|
"{manager_deployment_id}", url.PathEscape(strings.TrimSpace(record.DeploymentID)),
|
|
"{artifact_id}", url.PathEscape(strings.TrimSpace(artifactID)),
|
|
)
|
|
return replacer.Replace(path)
|
|
}
|
|
|
|
func callAgentRuntimeStatus(ctx context.Context, cfg agentRuntimeConfig, record agentDeploymentRecord) (map[string]any, int, error) {
|
|
if strings.TrimSpace(record.RuntimeSwarmID) == "" && strings.TrimSpace(record.RuntimeDeploymentID) == "" {
|
|
return nil, 0, errors.New("runtime identifiers missing")
|
|
}
|
|
endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeStatusPathForRecord(cfg, record))
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
agentRuntimeHeaders(req, cfg, record)
|
|
client := &http.Client{Timeout: cfg.Timeout}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer resp.Body.Close()
|
|
body, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
if readErr != nil {
|
|
return nil, resp.StatusCode, readErr
|
|
}
|
|
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
|
|
return nil, resp.StatusCode, fmt.Errorf("runtime status returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
|
|
}
|
|
var envelope map[string]any
|
|
if len(body) > 0 {
|
|
if err := common.Unmarshal(body, &envelope); err != nil {
|
|
return nil, resp.StatusCode, err
|
|
}
|
|
}
|
|
if message := agentRuntimeEnvelopeError(envelope); message != "" {
|
|
return nil, resp.StatusCode, errors.New(message)
|
|
}
|
|
return extractAgentRuntimeData(envelope), resp.StatusCode, nil
|
|
}
|
|
|
|
func callAgentRuntimeArtifactContent(ctx context.Context, cfg agentRuntimeConfig, record agentDeploymentRecord, artifactID string) (*http.Response, error) {
|
|
if strings.TrimSpace(artifactID) == "" {
|
|
return nil, errors.New("artifact_id is required")
|
|
}
|
|
if strings.TrimSpace(record.RuntimeSwarmID) == "" && strings.TrimSpace(record.RuntimeDeploymentID) == "" {
|
|
return nil, errors.New("runtime identifiers missing")
|
|
}
|
|
endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeArtifactContentPathForRecord(cfg, record, artifactID))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
agentRuntimeHeaders(req, cfg, record)
|
|
req.Header.Set("Accept", "*/*")
|
|
client := &http.Client{Timeout: cfg.Timeout}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices {
|
|
return resp, nil
|
|
}
|
|
defer resp.Body.Close()
|
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
|
return nil, fmt.Errorf("runtime artifact content returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
|
|
}
|
|
|
|
func mapSliceFromAny(value any) []gin.H {
|
|
raw, ok := value.([]any)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
items := make([]gin.H, 0, len(raw))
|
|
for _, entry := range raw {
|
|
if item, ok := entry.(map[string]any); ok {
|
|
items = append(items, gin.H(item))
|
|
}
|
|
}
|
|
return items
|
|
}
|
|
|
|
func mapFromAny(value any) map[string]any {
|
|
if item, ok := value.(map[string]any); ok {
|
|
return item
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func runtimeAgentHasFailed(agents []gin.H) bool {
|
|
for _, agent := range agents {
|
|
status := strings.ToLower(strings.TrimSpace(fmt.Sprint(agent["status"])))
|
|
if strings.Contains(status, "fail") ||
|
|
strings.Contains(status, "crash") ||
|
|
strings.Contains(status, "error") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// deliverableArtifactTypes are Runtime artifact_type values that represent a
|
|
// real business deliverable (code, tests, deploy manifest). Anything outside
|
|
// this set with no file-change signal is treated as a non-deliverable summary.
|
|
var deliverableArtifactTypes = map[string]bool{
|
|
"code_patch": true,
|
|
"code_bundle": true,
|
|
"code_document": true,
|
|
"deployment_manifest": true,
|
|
"test_report": true,
|
|
"diff": true,
|
|
"patch": true,
|
|
}
|
|
|
|
// artifactHasFileChanges reports whether the artifact carries a structured
|
|
// signal that real files were produced (count > 0 or a non-empty list), at the
|
|
// top level or under metadata. This is the authoritative deliverable signal,
|
|
// preferred over title/summary/uri string heuristics.
|
|
func artifactHasFileChanges(artifact gin.H) bool {
|
|
if anyPositiveFileSignal(artifact) {
|
|
return true
|
|
}
|
|
if meta, ok := artifact["metadata"].(map[string]any); ok {
|
|
if anyPositiveFileSignal(meta) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func anyPositiveFileSignal(values map[string]any) bool {
|
|
for _, key := range []string{"files_modified", "files_deleted", "files_changed", "files"} {
|
|
switch v := values[key].(type) {
|
|
case float64:
|
|
if v > 0 {
|
|
return true
|
|
}
|
|
case int:
|
|
if v > 0 {
|
|
return true
|
|
}
|
|
case []any:
|
|
if len(v) > 0 {
|
|
return true
|
|
}
|
|
case string:
|
|
if trimmed := strings.TrimSpace(v); trimmed != "" && trimmed != "0" {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// artifactIsSummaryOnly judges a single artifact by structured fields first
|
|
// (artifact_type + file-change signal), then falls back to the legacy title/
|
|
// summary/uri markers. Reading artifact_type fixes the case where Runtime emits
|
|
// artifact_type="document" (its no-files fallback) under a uri scheme that the
|
|
// old "/artifacts/summary" heuristic no longer matched.
|
|
func artifactIsSummaryOnly(artifact gin.H) bool {
|
|
// Runtime marks fallback artifacts (no real agent output) with
|
|
// metadata.synthesized=true — the authoritative non-deliverable signal
|
|
// (agent_management Sub Mode Runtime §7.2).
|
|
if meta, ok := artifact["metadata"].(map[string]any); ok {
|
|
if synth, ok := meta["synthesized"].(bool); ok && synth {
|
|
return true
|
|
}
|
|
}
|
|
atype := strings.ToLower(strings.TrimSpace(fmt.Sprint(artifact["artifact_type"])))
|
|
if deliverableArtifactTypes[atype] || artifactHasFileChanges(artifact) {
|
|
return false
|
|
}
|
|
title := strings.ToLower(strings.TrimSpace(fmt.Sprint(artifact["title"])))
|
|
summary := strings.ToLower(strings.TrimSpace(fmt.Sprint(artifact["summary"])))
|
|
uri := strings.ToLower(strings.TrimSpace(fmt.Sprint(artifact["uri"])))
|
|
if strings.Contains(summary, "without per-agent artifacts") ||
|
|
strings.Contains(title, "runtime execution summary") ||
|
|
strings.Contains(title, "runtime execution failed") ||
|
|
strings.Contains(uri, "/artifacts/summary") {
|
|
return true
|
|
}
|
|
// Runtime emits artifact_type="document" (or "summary") for the no-deliverable
|
|
// fallback; with no file-change signal that is a summary, not a code delivery.
|
|
return atype == "document" || atype == "summary"
|
|
}
|
|
|
|
func runtimeArtifactsAreSummaryOnly(artifacts []gin.H) bool {
|
|
if len(artifacts) == 0 {
|
|
return false
|
|
}
|
|
for _, artifact := range artifacts {
|
|
if !artifactIsSummaryOnly(artifact) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// persistedArtifactToGin adapts a stored artifact to the gin.H shape consumed
|
|
// by artifactIsSummaryOnly.
|
|
func persistedArtifactToGin(a model.AgentArtifact) gin.H {
|
|
g := gin.H{
|
|
"artifact_id": a.ArtifactID,
|
|
"artifact_type": a.ArtifactType,
|
|
"title": a.Title,
|
|
"summary": a.Summary,
|
|
"uri": a.URI,
|
|
}
|
|
if strings.TrimSpace(a.MetadataJSON) != "" {
|
|
var meta map[string]any
|
|
if err := common.UnmarshalJsonStr(a.MetadataJSON, &meta); err == nil && len(meta) > 0 {
|
|
g["metadata"] = meta
|
|
}
|
|
}
|
|
return g
|
|
}
|
|
|
|
// agentDeploymentDisplayStatus is the single status the client should show.
|
|
// Manager is the sole judge (unified spec §10.6): a `completed` runtime status
|
|
// is only surfaced as `completed` when there is a real (non-summary)
|
|
// deliverable; otherwise it is downgraded so an empty result is not shown as
|
|
// success — `needs_codegen` when only a plan/summary exists, or
|
|
// `completed_without_deliverable` when no artifact exists at all.
|
|
func agentDeploymentDisplayStatus(record agentDeploymentRecord) string {
|
|
if strings.ToLower(strings.TrimSpace(record.Status)) != "completed" {
|
|
return record.Status
|
|
}
|
|
artifacts, err := model.ListAgentArtifacts(model.ListAgentArtifactsFilter{
|
|
DeploymentID: record.DeploymentID,
|
|
Limit: 500,
|
|
})
|
|
if err != nil {
|
|
common.SysLog("agentDeploymentDisplayStatus: " + err.Error())
|
|
return record.Status
|
|
}
|
|
for _, a := range artifacts {
|
|
if !artifactIsSummaryOnly(persistedArtifactToGin(a)) {
|
|
return "completed"
|
|
}
|
|
}
|
|
if len(artifacts) > 0 {
|
|
return "needs_codegen"
|
|
}
|
|
return "completed_without_deliverable"
|
|
}
|
|
|
|
// heicodeClientMode maps the internal runtime mode to the client-facing mode
|
|
// string the desktop uses everywhere (capabilities/list/detail/workflow), so a
|
|
// task recovered from the list after restart routes to the right mode instead
|
|
// of defaulting to sub-agile (gap D/F). Internal `agent` => `sub_agile`.
|
|
func heicodeClientMode(record agentDeploymentRecord) string {
|
|
if normalizeAgentRuntimeMode(record.Plan.Metadata.RuntimeMode) == agentRuntimeModeSwarm {
|
|
return "swarm"
|
|
}
|
|
return "sub_agile"
|
|
}
|
|
|
|
// withDisplayStatus returns the record with DisplayStatus and the client-facing
|
|
// Mode computed for response (gap B/D/F).
|
|
func withDisplayStatus(record agentDeploymentRecord) agentDeploymentRecord {
|
|
record.DisplayStatus = agentDeploymentDisplayStatus(record)
|
|
record.Mode = heicodeClientMode(record)
|
|
return record
|
|
}
|
|
|
|
func ginString(g gin.H, key string) string {
|
|
v, ok := g[key]
|
|
if !ok || v == nil {
|
|
return ""
|
|
}
|
|
return strings.TrimSpace(fmt.Sprint(v))
|
|
}
|
|
|
|
// persistRuntimeStatusArtifact upserts an artifact surfaced by a runtime status
|
|
// query into the local store, so /artifacts and deliverable judgement converge
|
|
// even when the artifact.created callback was lost or raced.
|
|
func persistRuntimeStatusArtifact(record agentDeploymentRecord, a gin.H) {
|
|
artifactID := ginString(a, "artifact_id")
|
|
if artifactID == "" {
|
|
return
|
|
}
|
|
metaJSON := ""
|
|
if meta, ok := a["metadata"].(map[string]any); ok && len(meta) > 0 {
|
|
if raw, err := common.Marshal(meta); err == nil {
|
|
metaJSON = string(raw)
|
|
}
|
|
}
|
|
if err := model.UpsertAgentArtifact(&model.AgentArtifact{
|
|
ArtifactID: artifactID,
|
|
DeploymentID: record.DeploymentID,
|
|
UserID: record.Plan.UserContext.UserID,
|
|
BindingScope: firstPlanBindingScope(record.Plan),
|
|
ArtifactType: ginString(a, "artifact_type"),
|
|
Title: ginString(a, "title"),
|
|
Summary: ginString(a, "summary"),
|
|
URI: ginString(a, "uri"),
|
|
Checksum: ginString(a, "checksum"),
|
|
MetadataJSON: metaJSON,
|
|
CreatedAtMs: time.Now().UnixMilli(),
|
|
}); err != nil {
|
|
common.SysLog("persistRuntimeStatusArtifact: " + err.Error())
|
|
}
|
|
}
|
|
|
|
// reconcileDeploymentFromRuntime converges a non-terminal deployment from the
|
|
// authoritative runtime status. Manager must not rely solely on callbacks: they
|
|
// can race the runtime-id mapping (callback arrives before runtime_swarm_id is
|
|
// persisted) and orphan, leaving the deployment stuck at `accepted` while the
|
|
// runtime already completed. This pulls the live status on read and reflects
|
|
// terminal state + artifacts into the record.
|
|
func reconcileDeploymentFromRuntime(ctx context.Context, record agentDeploymentRecord) agentDeploymentRecord {
|
|
status := strings.ToLower(strings.TrimSpace(record.Status))
|
|
if containsString([]string{"completed", "failed", "stopped"}, status) {
|
|
return record
|
|
}
|
|
if strings.TrimSpace(record.RuntimeSwarmID) == "" && strings.TrimSpace(record.RuntimeDeploymentID) == "" {
|
|
return record
|
|
}
|
|
diag := agentRuntimeDiagnosticsForRecord(ctx, record)
|
|
if diag.DataSource != "runtime_status" || diag.HTTPStatus < 200 || diag.HTTPStatus >= 300 {
|
|
return record
|
|
}
|
|
rs := strings.ToLower(strings.TrimSpace(diag.Status))
|
|
if rs == "" {
|
|
return record
|
|
}
|
|
for _, a := range diag.Artifacts {
|
|
persistRuntimeStatusArtifact(record, a)
|
|
}
|
|
changed := false
|
|
if rs != strings.ToLower(strings.TrimSpace(record.Status)) {
|
|
record.Status = diag.Status
|
|
record.RuntimeState = diag.Status
|
|
changed = true
|
|
}
|
|
if strings.TrimSpace(diag.Phase) != "" && diag.Phase != record.Phase {
|
|
record.Phase = diag.Phase
|
|
changed = true
|
|
}
|
|
if containsString([]string{"completed", "failed", "stopped"}, rs) {
|
|
for idx := range record.AgentInstances {
|
|
cur := strings.ToLower(strings.TrimSpace(record.AgentInstances[idx].RuntimeState))
|
|
if !containsString([]string{"failed", "stopped"}, cur) {
|
|
record.AgentInstances[idx].RuntimeState = diag.Status
|
|
if record.Phase != "" {
|
|
record.AgentInstances[idx].Phase = record.Phase
|
|
}
|
|
}
|
|
}
|
|
changed = true
|
|
}
|
|
if changed {
|
|
record.UpdatedAt = agentNow()
|
|
agentMu.Lock()
|
|
agentDeployments[record.DeploymentID] = record
|
|
agentMu.Unlock()
|
|
if err := updateAgentDeploymentRecord(record); err != nil {
|
|
common.SysLog("reconcileDeploymentFromRuntime: " + err.Error())
|
|
}
|
|
}
|
|
return record
|
|
}
|
|
|
|
func buildAgentRuntimeDiagnostics(record agentDeploymentRecord, data map[string]any, httpStatus int, source string) agentRuntimeDiagnostics {
|
|
agents := mapSliceFromAny(data["agents"])
|
|
artifacts := mapSliceFromAny(data["artifacts"])
|
|
status := stringFromMap(data, "runtime_status", "status")
|
|
metrics := mapFromAny(data["metrics"])
|
|
warnings := make([]string, 0, 4)
|
|
agentFailed := runtimeAgentHasFailed(agents)
|
|
if agentFailed {
|
|
warnings = append(warnings, "runtime_agent_failed")
|
|
}
|
|
if strings.Contains(strings.ToLower(status), "completed") && agentFailed {
|
|
warnings = append(warnings, "runtime_completed_with_failed_agents")
|
|
}
|
|
if runtimeArtifactsAreSummaryOnly(artifacts) {
|
|
warnings = append(warnings, "runtime_summary_artifact_only")
|
|
}
|
|
if metrics != nil {
|
|
tokens := fmt.Sprint(metrics["tokens_used"])
|
|
if tokens == "0" || tokens == "0.0" {
|
|
warnings = append(warnings, "runtime_zero_model_usage")
|
|
}
|
|
}
|
|
return agentRuntimeDiagnostics{
|
|
DeploymentID: record.DeploymentID,
|
|
RuntimeMode: agentRuntimeModeForRecord(record),
|
|
SubMode: record.SubMode,
|
|
RuntimeDeploymentID: record.RuntimeDeploymentID,
|
|
RuntimeSwarmID: record.RuntimeSwarmID,
|
|
DataSource: source,
|
|
HTTPStatus: httpStatus,
|
|
Status: status,
|
|
Phase: stringFromMap(data, "phase", "stage"),
|
|
Progress: data["progress"],
|
|
ErrorMessage: stringFromMap(data, "error_message", "failure_reason", "error"),
|
|
Agents: agents,
|
|
Artifacts: artifacts,
|
|
Metrics: metrics,
|
|
Warnings: warnings,
|
|
CheckedAt: agentNow(),
|
|
}
|
|
}
|
|
|
|
func agentRuntimeDiagnosticsForRecord(ctx context.Context, record agentDeploymentRecord) agentRuntimeDiagnostics {
|
|
mode := agentRuntimeModeForRecord(record)
|
|
cfg := agentRuntimeClientConfigForMode(mode)
|
|
if !cfg.Enabled || strings.TrimSpace(cfg.BaseURL) == "" {
|
|
return agentRuntimeDiagnostics{
|
|
DeploymentID: record.DeploymentID,
|
|
RuntimeMode: mode,
|
|
SubMode: record.SubMode,
|
|
DataSource: "not_configured",
|
|
Warnings: []string{"runtime_not_configured"},
|
|
CheckedAt: agentNow(),
|
|
}
|
|
}
|
|
if strings.TrimSpace(record.RuntimeSwarmID) == "" && strings.TrimSpace(record.RuntimeDeploymentID) == "" {
|
|
return agentRuntimeDiagnostics{
|
|
DeploymentID: record.DeploymentID,
|
|
RuntimeMode: mode,
|
|
SubMode: record.SubMode,
|
|
DataSource: "missing_runtime_id",
|
|
Warnings: []string{"runtime_identifiers_missing"},
|
|
CheckedAt: agentNow(),
|
|
}
|
|
}
|
|
data, status, err := callAgentRuntimeStatus(ctx, cfg, record)
|
|
if err != nil {
|
|
return agentRuntimeDiagnostics{
|
|
DeploymentID: record.DeploymentID,
|
|
RuntimeMode: mode,
|
|
SubMode: record.SubMode,
|
|
RuntimeDeploymentID: record.RuntimeDeploymentID,
|
|
RuntimeSwarmID: record.RuntimeSwarmID,
|
|
DataSource: "runtime_status_error",
|
|
HTTPStatus: status,
|
|
ErrorMessage: truncateAgentFailureReason(err.Error()),
|
|
Warnings: []string{"runtime_status_query_failed"},
|
|
CheckedAt: agentNow(),
|
|
}
|
|
}
|
|
return buildAgentRuntimeDiagnostics(record, data, status, "runtime_status")
|
|
}
|
|
|
|
func AgentGetUserDeploymentRuntimeDiagnostics(c *gin.Context) {
|
|
record, ok := requireAuthenticatedUserAgentDeployment(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(c.Request.Context(), agentRuntimeClientConfigForMode(agentRuntimeModeForRecord(record)).Timeout)
|
|
defer cancel()
|
|
common.ApiSuccess(c, agentRuntimeDiagnosticsForRecord(ctx, record))
|
|
}
|
|
|
|
func agentRuntimeApprovalDecisionPath(cfg agentRuntimeConfig, record agentDeploymentRecord, approvalID string) string {
|
|
path := strings.TrimSpace(cfg.ApprovalDecisionPath)
|
|
if path == "" {
|
|
path = "/api/swarms/{swarm_id}/approvals/{approval_id}"
|
|
}
|
|
replacer := strings.NewReplacer(
|
|
"{swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{runtime_swarm_id}", url.PathEscape(strings.TrimSpace(record.RuntimeSwarmID)),
|
|
"{deployment_id}", url.PathEscape(strings.TrimSpace(record.RuntimeDeploymentID)),
|
|
"{runtime_deployment_id}", url.PathEscape(strings.TrimSpace(record.RuntimeDeploymentID)),
|
|
"{manager_deployment_id}", url.PathEscape(strings.TrimSpace(record.DeploymentID)),
|
|
"{approval_id}", url.PathEscape(strings.TrimSpace(approvalID)),
|
|
)
|
|
return replacer.Replace(path)
|
|
}
|
|
|
|
func callAgentRuntimeApprovalDecision(ctx context.Context, cfg agentRuntimeConfig, record agentDeploymentRecord, approval model.AgentApprovalRequest, lease *model.AgentCredentialLease, decision string) error {
|
|
endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeApprovalDecisionPath(cfg, record, approval.ApprovalID))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
payload := gin.H{
|
|
"approval_id": approval.ApprovalID,
|
|
"decision": decision,
|
|
"manager_deployment_id": record.DeploymentID,
|
|
"runtime_deployment_id": record.RuntimeDeploymentID,
|
|
"swarm_id": record.RuntimeSwarmID,
|
|
"operation": approval.Operation,
|
|
"resource_id": approval.ResourceID,
|
|
"resource_type": approval.ResourceType,
|
|
"resource_scope": approval.ResourceScope,
|
|
"target_role": approval.TargetRole,
|
|
"risk_level": approval.RiskLevel,
|
|
"requires_credential": approval.RequiresCredential,
|
|
"decided_by": approval.DecidedBy,
|
|
"reason": approval.DecisionReason,
|
|
"decided_at": approval.DecidedAt,
|
|
}
|
|
if approval.UserId > 0 {
|
|
payload["user_id"] = fmt.Sprintf("%d", approval.UserId)
|
|
}
|
|
if lease != nil && strings.TrimSpace(lease.CredentialRef) != "" {
|
|
payload["credential_ref"] = strings.TrimSpace(lease.CredentialRef)
|
|
payload["lease_id"] = strings.TrimSpace(lease.LeaseID)
|
|
payload["lease_expires_at"] = lease.ExpiresAt
|
|
}
|
|
body, err := common.Marshal(payload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("X-Correlation-ID", firstNonEmpty(record.Plan.Metadata.CorrelationID, approval.ApprovalID))
|
|
req.Header.Set("X-Idempotency-Key", "approval-decision-"+approval.ApprovalID+"-"+decision)
|
|
if cfg.Token != "" {
|
|
req.Header.Set("Authorization", "Bearer "+cfg.Token)
|
|
}
|
|
client := &http.Client{Timeout: cfg.Timeout}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer resp.Body.Close()
|
|
respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
if readErr != nil {
|
|
return readErr
|
|
}
|
|
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
|
|
return fmt.Errorf("runtime approval decision returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
|
|
}
|
|
if len(respBody) == 0 {
|
|
return nil
|
|
}
|
|
var envelope map[string]any
|
|
if err := common.Unmarshal(respBody, &envelope); err != nil {
|
|
return err
|
|
}
|
|
if message := agentRuntimeEnvelopeError(envelope); message != "" {
|
|
return errors.New(message)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func syncAgentRuntimeApprovalDecision(c *gin.Context, approval *model.AgentApprovalRequest, lease *model.AgentCredentialLease, decision string) {
|
|
if approval == nil || strings.TrimSpace(approval.DeploymentID) == "" {
|
|
return
|
|
}
|
|
if !agentRuntimeClientConfigForMode(agentRuntimeModeAgent).Enabled && !agentRuntimeClientConfigForMode(agentRuntimeModeSwarm).Enabled {
|
|
return
|
|
}
|
|
record, ok := findAgentDeploymentRecord(approval.DeploymentID)
|
|
if !ok {
|
|
recordAgentApprovalAudit("runtime.approval_decision.skipped", approval, lease, "skipped", "deployment not found")
|
|
return
|
|
}
|
|
cfg := agentRuntimeClientConfigForMode(agentRuntimeModeForRecord(record))
|
|
if !cfg.Enabled {
|
|
return
|
|
}
|
|
if strings.TrimSpace(record.RuntimeSwarmID) == "" && strings.TrimSpace(record.RuntimeDeploymentID) == "" {
|
|
recordAgentApprovalAudit("runtime.approval_decision.skipped", approval, lease, "skipped", "runtime identifiers missing")
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(c.Request.Context(), cfg.Timeout)
|
|
defer cancel()
|
|
if err := callAgentRuntimeApprovalDecision(ctx, cfg, record, *approval, lease, decision); err != nil {
|
|
recordAgentApprovalAudit("runtime.approval_decision.failed", approval, lease, "failed", truncateAgentFailureReason(err.Error()))
|
|
common.SysLog("Agent runtime approval decision failed for " + approval.ApprovalID + ": " + err.Error())
|
|
return
|
|
}
|
|
recordAgentApprovalAudit("runtime.approval_decision.accepted", approval, lease, "ok", "")
|
|
}
|
|
|
|
func callAgentRuntimeStop(ctx context.Context, cfg agentRuntimeConfig, record agentDeploymentRecord, reason string) (agentRuntimeSyncResult, error) {
|
|
runtimeDeploymentID := strings.TrimSpace(record.RuntimeDeploymentID)
|
|
if runtimeDeploymentID == "" {
|
|
return agentRuntimeSyncResult{}, nil
|
|
}
|
|
endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeStopPathForRecord(cfg, record))
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
payload, err := common.Marshal(gin.H{
|
|
"reason": firstNonEmpty(strings.TrimSpace(reason), "Heicode Manager requested stop"),
|
|
"manager_deployment_id": record.DeploymentID,
|
|
})
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
agentRuntimeHeaders(req, cfg, record)
|
|
client := &http.Client{Timeout: cfg.Timeout}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return agentRuntimeSyncResult{}, err
|
|
}
|
|
defer resp.Body.Close()
|
|
body, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
if readErr != nil {
|
|
return agentRuntimeSyncResult{}, readErr
|
|
}
|
|
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
|
|
return agentRuntimeSyncResult{RawStatusCode: resp.StatusCode}, fmt.Errorf("runtime stop returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
|
|
}
|
|
var envelope map[string]any
|
|
if len(body) > 0 {
|
|
if err := common.Unmarshal(body, &envelope); err != nil {
|
|
return agentRuntimeSyncResult{RawStatusCode: resp.StatusCode}, err
|
|
}
|
|
}
|
|
if message := agentRuntimeEnvelopeError(envelope); message != "" {
|
|
return agentRuntimeSyncResult{RawStatusCode: resp.StatusCode}, errors.New(message)
|
|
}
|
|
data := extractAgentRuntimeData(envelope)
|
|
return agentRuntimeSyncResult{
|
|
RuntimeDeploymentID: firstNonEmpty(stringFromMap(data, "runtime_deployment_id", "deployment_id", "id"), runtimeDeploymentID),
|
|
RuntimeSwarmID: stringFromMap(data, "swarm_id", "runtime_swarm_id"),
|
|
RuntimeStatus: stringFromMap(data, "runtime_status", "status"),
|
|
RawStatusCode: resp.StatusCode,
|
|
}, nil
|
|
}
|
|
|
|
func syncAgentRuntimeStop(c *gin.Context, record agentDeploymentRecord, reason string) (agentDeploymentRecord, bool) {
|
|
cfg := agentRuntimeClientConfigForMode(agentRuntimeModeForRecord(record))
|
|
if !cfg.Enabled || strings.TrimSpace(record.RuntimeDeploymentID) == "" {
|
|
return record, true
|
|
}
|
|
ctx, cancel := context.WithTimeout(c.Request.Context(), cfg.Timeout)
|
|
defer cancel()
|
|
result, err := callAgentRuntimeStop(ctx, cfg, record, reason)
|
|
record.RuntimeLastSyncAt = agentNow()
|
|
if err != nil {
|
|
record.RuntimeState = agentRuntimeStateFailed
|
|
record.FailureReason = truncateAgentFailureReason(err.Error())
|
|
record.UpdatedAt = agentNow()
|
|
_ = updateAgentDeploymentRecord(record)
|
|
agentMu.Lock()
|
|
agentDeployments[record.DeploymentID] = record
|
|
agentMu.Unlock()
|
|
recordAgentRuntimeSyncAudit(record, "runtime.stop.failed", "failed")
|
|
agentError(c, "RUNTIME_STOP_FAILED", record.FailureReason)
|
|
return record, false
|
|
}
|
|
record.RuntimeState = firstNonEmpty(result.RuntimeStatus, "stopped")
|
|
if result.RuntimeDeploymentID != "" {
|
|
record.RuntimeDeploymentID = result.RuntimeDeploymentID
|
|
}
|
|
if result.RuntimeSwarmID != "" {
|
|
record.RuntimeSwarmID = result.RuntimeSwarmID
|
|
}
|
|
record.FailureReason = ""
|
|
recordAgentRuntimeSyncAudit(record, "runtime.stop.accepted", "ok")
|
|
return record, true
|
|
}
|
|
|
|
func updateAgentRuntimeSyncState(record agentDeploymentRecord, result agentRuntimeSyncResult, syncErr error) agentDeploymentRecord {
|
|
record.RuntimeLastSyncAt = agentNow()
|
|
if syncErr != nil {
|
|
record.RuntimeState = agentRuntimeStateFailed
|
|
record.FailureReason = truncateAgentFailureReason(syncErr.Error())
|
|
} else {
|
|
record.RuntimeState = firstNonEmpty(result.RuntimeStatus, agentRuntimeStateSynced)
|
|
record.RuntimeDeploymentID = result.RuntimeDeploymentID
|
|
record.RuntimeSwarmID = result.RuntimeSwarmID
|
|
record.FailureReason = ""
|
|
}
|
|
record.UpdatedAt = agentNow()
|
|
if err := updateAgentDeploymentRecord(record); err != nil {
|
|
common.SysLog("updateAgentRuntimeSyncState: " + err.Error())
|
|
}
|
|
agentMu.Lock()
|
|
agentDeployments[record.DeploymentID] = record
|
|
agentMu.Unlock()
|
|
return record
|
|
}
|
|
|
|
func recordAgentRuntimeSyncAudit(record agentDeploymentRecord, event string, result string) {
|
|
recordAgentAuditEvent(agentEvent{
|
|
EventID: "evt_" + common.GetUUID()[:12],
|
|
Event: event,
|
|
SchemaVersion: 1,
|
|
UserID: record.Plan.UserContext.UserID,
|
|
ChannelID: record.Plan.UserContext.ChannelID,
|
|
BindingScope: firstPlanBindingScope(record.Plan),
|
|
DeploymentID: record.DeploymentID,
|
|
CorrelationID: record.Plan.Metadata.CorrelationID,
|
|
OccurredAt: agentNow(),
|
|
}, "agent_runtime_bridge", record.DeploymentID, "", result)
|
|
}
|
|
|
|
func dispatchAgentRuntimeCreate(record agentDeploymentRecord, source string, cfg agentRuntimeConfig) agentDeploymentRecord {
|
|
recordAgentRuntimeSyncAudit(record, "runtime.sync.started", "started")
|
|
ctx, cancel := context.WithTimeout(context.Background(), cfg.Timeout)
|
|
defer cancel()
|
|
result, err := callAgentRuntimeCreate(ctx, cfg, record, source)
|
|
record = updateAgentRuntimeSyncState(record, result, err)
|
|
if err != nil {
|
|
common.SysLog("Agent runtime shadow create failed for " + record.DeploymentID + ": " + err.Error())
|
|
recordAgentRuntimeSyncAudit(record, "runtime.sync.failed", "failed")
|
|
return record
|
|
}
|
|
recordAgentRuntimeSyncAudit(record, "runtime.sync.accepted", "ok")
|
|
return record
|
|
}
|
|
|
|
func maybeDispatchAgentRuntimeCreate(c *gin.Context, record agentDeploymentRecord, source string) agentDeploymentRecord {
|
|
mode := agentRuntimeModeForSource(source)
|
|
if strings.TrimSpace(record.Plan.Metadata.RuntimeMode) == "" {
|
|
record.Plan.Metadata.RuntimeMode = mode
|
|
} else {
|
|
mode = agentRuntimeModeForRecord(record)
|
|
}
|
|
cfg := agentRuntimeClientConfigForMode(mode)
|
|
if !cfg.Enabled {
|
|
return record
|
|
}
|
|
if _, err := agentRuntimeURL(cfg.BaseURL, cfg.CreatePath); err != nil {
|
|
record.RuntimeState = agentRuntimeStateFailed
|
|
record.RuntimeLastSyncAt = agentNow()
|
|
record.FailureReason = truncateAgentFailureReason(err.Error())
|
|
record.UpdatedAt = agentNow()
|
|
_ = updateAgentDeploymentRecord(record)
|
|
recordAgentRuntimeSyncAudit(record, "runtime.sync.failed", "failed")
|
|
return record
|
|
}
|
|
record.RuntimeState = agentRuntimeStateSyncing
|
|
record.RuntimeLastSyncAt = agentNow()
|
|
record.UpdatedAt = agentNow()
|
|
if err := updateAgentDeploymentRecord(record); err != nil {
|
|
common.SysLog("maybeDispatchAgentRuntimeCreate: " + err.Error())
|
|
}
|
|
agentMu.Lock()
|
|
agentDeployments[record.DeploymentID] = record
|
|
agentMu.Unlock()
|
|
if cfg.Async {
|
|
sourceCopy := source
|
|
recordCopy := record
|
|
go dispatchAgentRuntimeCreate(recordCopy, sourceCopy, cfg)
|
|
return record
|
|
}
|
|
return dispatchAgentRuntimeCreate(record, source, cfg)
|
|
}
|
|
|
|
func AgentRuntimeHealth(c *gin.Context) {
|
|
cfg := agentRuntimeClientConfigForMode(c.Query("mode"))
|
|
data := gin.H{
|
|
"enabled": cfg.Enabled,
|
|
"configured": cfg.BaseURL != "",
|
|
"create_path": cfg.CreatePath,
|
|
"health_path": cfg.HealthPath,
|
|
"stop_path": cfg.StopPath,
|
|
"mode": normalizeAgentRuntimeMode(c.Query("mode")),
|
|
}
|
|
if cfg.BaseURL == "" {
|
|
data["status"] = "not_configured"
|
|
common.ApiSuccess(c, data)
|
|
return
|
|
}
|
|
endpoint, err := agentRuntimeURL(cfg.BaseURL, cfg.HealthPath)
|
|
if err != nil {
|
|
data["status"] = "invalid_config"
|
|
data["message"] = err.Error()
|
|
common.ApiSuccess(c, data)
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(c.Request.Context(), cfg.Timeout)
|
|
defer cancel()
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
|
if err != nil {
|
|
data["status"] = "request_failed"
|
|
data["message"] = err.Error()
|
|
common.ApiSuccess(c, data)
|
|
return
|
|
}
|
|
if cfg.Token != "" {
|
|
req.Header.Set("Authorization", "Bearer "+cfg.Token)
|
|
}
|
|
resp, err := (&http.Client{Timeout: cfg.Timeout}).Do(req)
|
|
if err != nil {
|
|
data["status"] = "unreachable"
|
|
data["message"] = err.Error()
|
|
common.ApiSuccess(c, data)
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
data["http_status"] = resp.StatusCode
|
|
if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices {
|
|
data["status"] = "healthy"
|
|
} else {
|
|
data["status"] = "unhealthy"
|
|
}
|
|
if len(body) > 0 {
|
|
var remote map[string]any
|
|
if err := common.Unmarshal(body, &remote); err == nil {
|
|
data["remote"] = remote
|
|
} else {
|
|
data["body"] = strings.TrimSpace(string(body))
|
|
}
|
|
}
|
|
common.ApiSuccess(c, data)
|
|
}
|