From 50b76dd4e6098ec2de28c4e496d1f2f4585827d2 Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Wed, 27 May 2026 11:04:08 +0800 Subject: [PATCH] feat: align agnet runtime v2.1 integration --- heicode/VERSION | 2 +- heicode/controller/agnet_callback.go | 419 ++++++++++++++++-- heicode/controller/agnet_control_plane.go | 170 +++++-- .../controller/agnet_control_plane_test.go | 378 ++++++++++++++++ heicode/controller/agnet_runtime_client.go | 368 +++++++++++++++ heicode/model/agnet_callback.go | 31 +- heicode/model/agnet_deployment.go | 3 + .../router/agnet_runtime_http_smoke_test.go | 178 ++++++++ heicode/router/api-router.go | 1 + heicode/router/api_router_test.go | 15 + 10 files changed, 1459 insertions(+), 106 deletions(-) create mode 100644 heicode/controller/agnet_runtime_client.go create mode 100644 heicode/router/agnet_runtime_http_smoke_test.go create mode 100644 heicode/router/api_router_test.go diff --git a/heicode/VERSION b/heicode/VERSION index 1c99cf0e..e516bb9d 100644 --- a/heicode/VERSION +++ b/heicode/VERSION @@ -1 +1 @@ -1.4.4 +1.4.5 diff --git a/heicode/controller/agnet_callback.go b/heicode/controller/agnet_callback.go index c51dc3a3..7cd71898 100644 --- a/heicode/controller/agnet_callback.go +++ b/heicode/controller/agnet_callback.go @@ -1,7 +1,13 @@ package controller import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" "os" + "strconv" "strings" "time" @@ -11,15 +17,19 @@ import ( ) type agnetCallbackEnvelope struct { - EventID string `json:"event_id"` - IdempotencyKey string `json:"idempotency_key"` - EventType string `json:"event_type"` - DeploymentID string `json:"deployment_id"` - TaskID string `json:"task_id"` - CorrelationID string `json:"correlation_id"` - Source string `json:"source"` - Metadata map[string]any `json:"metadata"` - Artifact agnetArtifactPayload `json:"artifact"` + EventID string `json:"event_id"` + IdempotencyKey string `json:"idempotency_key"` + EventType string `json:"event_type"` + DeploymentID string `json:"deployment_id"` + SwarmID string `json:"swarm_id"` + AgentInstanceID string `json:"agent_instance_id"` + TaskID string `json:"task_id"` + OccurredAt string `json:"occurred_at"` + CorrelationID string `json:"correlation_id"` + Source string `json:"source"` + Metadata map[string]any `json:"metadata"` + Payload map[string]any `json:"payload"` + Artifact agnetArtifactPayload `json:"artifact"` } type agnetArtifactPayload struct { @@ -43,10 +53,84 @@ func agnetCallbackTokenFromRequest(c *gin.Context) string { return "" } -func validateAgnetCallbackAuth(c *gin.Context) bool { +func agnetCallbackSigningSecret() string { + if secret := strings.TrimSpace(common.GetEnvOrDefaultString("AGNET_CALLBACK_SIGNING_SECRET", "")); secret != "" { + return secret + } + secretRef := firstNonEmpty( + common.GetEnvOrDefaultString("AGNET_CALLBACK_SIGNING_SECRET_REF", ""), + common.GetEnvOrDefaultString("AGNET_RUNTIME_CALLBACK_SIGNING_SECRET_REF", ""), + ) + if secretRef == "" { + return "" + } + client, err := newSecretStoreClientFromEnv() + if err != nil { + common.SysLog("agnetCallbackSigningSecret: " + err.Error()) + return "" + } + data, err := client.getJSONSecret(secretRef) + if err != nil { + common.SysLog("agnetCallbackSigningSecret: " + err.Error()) + return "" + } + for _, key := range []string{"callback_signing_secret", "signing_secret", "secret", "value"} { + if value := callbackStringValue(data, key); value != "" { + return value + } + } + common.SysLog("agnetCallbackSigningSecret: signing secret is missing from Azure Key Vault payload") + return "" +} + +func agnetCallbackSignatureTolerance() time.Duration { + seconds := common.GetEnvOrDefault("AGNET_CALLBACK_SIGNATURE_TOLERANCE_SECONDS", 300) + if seconds <= 0 { + seconds = 300 + } + return time.Duration(seconds) * time.Second +} + +func validateAgnetCallbackHMAC(c *gin.Context, rawBody []byte, eventID string) (bool, bool) { + secret := agnetCallbackSigningSecret() + if secret == "" { + return false, false + } + timestamp := strings.TrimSpace(c.GetHeader("X-Agnet-Timestamp")) + signature := strings.TrimSpace(c.GetHeader("X-Agnet-Signature")) + if timestamp == "" || signature == "" || eventID == "" { + return false, false + } + tsMs, err := strconv.ParseInt(timestamp, 10, 64) + if err != nil { + agnetError(c, "CALLBACK_UNAUTHORIZED", "invalid callback timestamp") + return true, false + } + now := time.Now() + eventTime := time.UnixMilli(tsMs) + tolerance := agnetCallbackSignatureTolerance() + if eventTime.Before(now.Add(-tolerance)) || eventTime.After(now.Add(tolerance)) { + agnetError(c, "CALLBACK_UNAUTHORIZED", "callback timestamp outside allowed window") + return true, false + } + payload := timestamp + "." + eventID + "." + string(rawBody) + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write([]byte(payload)) + expected := "sha256=" + hex.EncodeToString(mac.Sum(nil)) + if !hmac.Equal([]byte(expected), []byte(signature)) { + agnetError(c, "CALLBACK_UNAUTHORIZED", "invalid callback signature") + return true, false + } + return true, true +} + +func validateAgnetCallbackAuth(c *gin.Context, rawBody []byte, eventID string) bool { if common.GetEnvOrDefaultBool("AGNET_CALLBACK_AUTH_DISABLED", false) { return true } + if attempted, ok := validateAgnetCallbackHMAC(c, rawBody, eventID); attempted { + return ok + } expected := strings.TrimSpace(os.Getenv("AGNET_CALLBACK_TOKEN")) if expected == "" { agnetError(c, "CALLBACK_UNAUTHORIZED", "callback token is not configured") @@ -60,7 +144,7 @@ func validateAgnetCallbackAuth(c *gin.Context) bool { } func agnetCallbackHasPlaintextSecret(payload agnetCallbackEnvelope) bool { - if containsPlaintextSecret(payload.Metadata) || containsPlaintextSecret(payload.Artifact.Metadata) { + if containsPlaintextSecret(payload.Metadata) || containsPlaintextSecret(payload.Payload) || containsPlaintextSecret(payload.Artifact.Metadata) { return true } raw, _ := common.Marshal(payload) @@ -71,11 +155,116 @@ func agnetCallbackHasPlaintextSecret(payload agnetCallbackEnvelope) bool { return containsPlaintextSecret(asMap) } -func agnetCallbackDeploymentContext(deploymentID string) (agnetDeploymentRecord, bool) { - if deploymentID == "" { +func agnetCallbackDeploymentContext(deploymentID string, swarmID string) (agnetDeploymentRecord, bool) { + deploymentID = strings.TrimSpace(deploymentID) + swarmID = strings.TrimSpace(swarmID) + if deploymentID != "" { + if record, ok := findAgnetDeploymentRecord(deploymentID); ok { + return record, true + } + } + if swarmID == "" { return agnetDeploymentRecord{}, false } - return findAgnetDeploymentRecord(deploymentID) + + agnetMu.RLock() + for _, record := range agnetDeployments { + if strings.TrimSpace(record.RuntimeSwarmID) == swarmID { + agnetMu.RUnlock() + return record, true + } + } + agnetMu.RUnlock() + + if model.DB == nil { + return agnetDeploymentRecord{}, false + } + var row model.AgnetDeployment + if err := model.DB.Where("runtime_swarm_id = ?", swarmID).First(&row).Error; err != nil { + return agnetDeploymentRecord{}, false + } + record, err := agnetDeploymentModelToRecord(row) + if err != nil { + common.SysLog("agnetCallbackDeploymentContext: " + err.Error()) + return agnetDeploymentRecord{}, false + } + agnetMu.Lock() + agnetDeployments[record.DeploymentID] = record + agnetMu.Unlock() + return record, true +} + +func callbackStringValue(values map[string]any, key string) string { + if values == nil { + return "" + } + value, ok := values[key] + if !ok { + return "" + } + switch typed := value.(type) { + case string: + return strings.TrimSpace(typed) + case fmt.Stringer: + return strings.TrimSpace(typed.String()) + case float64: + return strconv.FormatFloat(typed, 'f', -1, 64) + case int: + return strconv.Itoa(typed) + case int64: + return strconv.FormatInt(typed, 10) + default: + return "" + } +} + +func callbackMapValue(values map[string]any, key string) map[string]any { + if values == nil { + return nil + } + if nested, ok := values[key].(map[string]any); ok { + return nested + } + return nil +} + +func normalizeCallbackArtifact(payload *agnetCallbackEnvelope) { + if payload == nil || strings.TrimSpace(payload.Artifact.ArtifactID) != "" { + return + } + source := payload.Payload + if nested := callbackMapValue(source, "artifact"); nested != nil { + source = nested + } + if len(source) == 0 { + return + } + payload.Artifact = agnetArtifactPayload{ + ArtifactID: callbackStringValue(source, "artifact_id"), + ArtifactType: callbackStringValue(source, "artifact_type"), + Title: callbackStringValue(source, "title"), + Summary: callbackStringValue(source, "summary"), + URI: callbackStringValue(source, "uri"), + Checksum: callbackStringValue(source, "checksum"), + Metadata: map[string]any{}, + } + for key, value := range source { + switch key { + case "artifact_id", "artifact_type", "title", "summary", "uri", "checksum": + continue + case "metadata": + if nested, ok := value.(map[string]any); ok { + for nestedKey, nestedValue := range nested { + payload.Artifact.Metadata[nestedKey] = nestedValue + } + } + default: + payload.Artifact.Metadata[key] = value + } + } + if len(payload.Artifact.Metadata) == 0 { + payload.Artifact.Metadata = nil + } } func persistAgnetArtifactFromCallback(payload agnetCallbackEnvelope, record agnetDeploymentRecord) error { @@ -106,21 +295,120 @@ func persistAgnetArtifactFromCallback(payload agnetCallbackEnvelope, record agne }) } +func callbackBoolValue(values map[string]any, key string) bool { + if values == nil { + return false + } + switch typed := values[key].(type) { + case bool: + return typed + case string: + return strings.EqualFold(strings.TrimSpace(typed), "true") + default: + return false + } +} + +func callbackIntValue(values map[string]any, key string) int { + if values == nil { + return 0 + } + switch typed := values[key].(type) { + case float64: + return int(typed) + case int: + return typed + case int64: + return int(typed) + case string: + value, _ := strconv.Atoi(strings.TrimSpace(typed)) + return value + default: + return 0 + } +} + +func persistAgnetApprovalFromCallback(payload agnetCallbackEnvelope, record agnetDeploymentRecord) error { + if payload.EventType != "approval.requested" || model.DB == nil { + return nil + } + source := payload.Payload + if nested := callbackMapValue(source, "approval"); nested != nil { + source = nested + } + approvalID := firstNonEmpty(callbackStringValue(source, "approval_id"), "appr_"+common.GetUUID()) + userID, _ := strconv.Atoi(strings.TrimSpace(record.Plan.UserContext.UserID)) + approval := model.AgnetApprovalRequest{ + ApprovalID: approvalID, + UserId: userID, + DeploymentID: strings.TrimSpace(payload.DeploymentID), + BindingScope: firstNonEmpty(callbackStringValue(source, "binding_scope"), firstPlanBindingScope(record.Plan)), + Operation: firstNonEmpty(callbackStringValue(source, "operation"), "runtime.approval"), + ResourceID: firstNonEmpty(callbackStringValue(source, "resource_id"), strings.TrimSpace(payload.AgentInstanceID), strings.TrimSpace(payload.DeploymentID)), + ResourceType: firstNonEmpty(callbackStringValue(source, "resource_type"), "custom"), + ResourceScope: callbackStringValue(source, "resource_scope"), + TargetRole: firstNonEmpty(callbackStringValue(source, "target_role"), callbackStringValue(source, "agent_role"), strings.TrimSpace(payload.AgentInstanceID), "runtime"), + RiskLevel: firstNonEmpty(callbackStringValue(source, "risk_level"), "high"), + RequiresCredential: callbackBoolValue(source, "requires_credential"), + SecretRef: strings.TrimSpace(callbackStringValue(source, "secret_ref")), + Status: agnetApprovalStatusPending, + RequestedBy: firstNonEmpty(callbackStringValue(source, "requested_by"), "agnet-runtime"), + RequestReason: firstNonEmpty(callbackStringValue(source, "reason"), callbackStringValue(source, "summary"), "Runtime requested approval"), + TTLSeconds: callbackIntValue(source, "ttl_seconds"), + } + if approval.TTLSeconds <= 0 { + approval.TTLSeconds = defaultAgnetApprovalTTLSeconds + } + if approval.TTLSeconds > maxAgnetApprovalTTLSeconds { + approval.TTLSeconds = maxAgnetApprovalTTLSeconds + } + if approval.SecretRef != "" && !strings.HasPrefix(approval.SecretRef, "azkv://") { + return fmt.Errorf("approval secret_ref must use azkv:// Azure Key Vault reference") + } + if approval.RequiresCredential && approval.SecretRef == "" { + return fmt.Errorf("approval secret_ref is required when requires_credential is true") + } + now := time.Now().UnixMilli() + approval.ExpiresAt = now + int64(approval.TTLSeconds)*1000 + + var existing model.AgnetApprovalRequest + if err := model.DB.Where("approval_id = ?", approval.ApprovalID).First(&existing).Error; err == nil { + return nil + } + if err := model.DB.Create(&approval).Error; err != nil { + return err + } + recordAgnetApprovalAudit("approval.requested", &approval, nil, "ok", "") + return nil +} + func AgnetReceiveSwarmEventCallback(c *gin.Context) { - if !validateAgnetCallbackAuth(c) { + rawBody, err := io.ReadAll(io.LimitReader(c.Request.Body, 1<<20)) + if err != nil { + agnetError(c, "CALLBACK_INVALID", "failed to read callback body") return } var payload agnetCallbackEnvelope - if err := c.ShouldBindJSON(&payload); err != nil { + if err := common.Unmarshal(rawBody, &payload); err != nil { agnetError(c, "CALLBACK_INVALID", err.Error()) return } + if payload.EventID == "" { + payload.EventID = strings.TrimSpace(c.GetHeader("X-Agnet-Event-Id")) + } + if payload.CorrelationID == "" { + payload.CorrelationID = strings.TrimSpace(c.GetHeader("X-Correlation-ID")) + } payload.EventID = strings.TrimSpace(payload.EventID) payload.EventType = strings.TrimSpace(payload.EventType) + if !validateAgnetCallbackAuth(c, rawBody, payload.EventID) { + return + } if payload.EventID == "" || payload.EventType == "" { agnetError(c, "CALLBACK_INVALID", "event_id and event_type are required") return } + normalizeCallbackArtifact(&payload) if payload.IdempotencyKey == "" { payload.IdempotencyKey = payload.EventID } @@ -129,25 +417,31 @@ func AgnetReceiveSwarmEventCallback(c *gin.Context) { return } - record, _ := agnetCallbackDeploymentContext(strings.TrimSpace(payload.DeploymentID)) + record, _ := agnetCallbackDeploymentContext(strings.TrimSpace(payload.DeploymentID), strings.TrimSpace(payload.SwarmID)) + if strings.TrimSpace(payload.DeploymentID) == "" && strings.TrimSpace(record.DeploymentID) != "" { + payload.DeploymentID = record.DeploymentID + } if payload.CorrelationID == "" { payload.CorrelationID = record.Plan.Metadata.CorrelationID } payloadJSON, _ := common.Marshal(payload) inserted, err := model.InsertAgnetCallbackEvent(&model.AgnetCallbackEvent{ - EventID: payload.EventID, - IdempotencyKey: strings.TrimSpace(payload.IdempotencyKey), - CallbackType: "swarm-event", - EventType: payload.EventType, - DeploymentID: strings.TrimSpace(payload.DeploymentID), - TaskID: strings.TrimSpace(payload.TaskID), - UserID: record.Plan.UserContext.UserID, - BindingScope: firstPlanBindingScope(record.Plan), - CorrelationID: strings.TrimSpace(payload.CorrelationID), - Source: strings.TrimSpace(payload.Source), - Result: "ok", - PayloadJSON: string(payloadJSON), - CreatedAtMs: time.Now().UnixMilli(), + EventID: payload.EventID, + IdempotencyKey: strings.TrimSpace(payload.IdempotencyKey), + CallbackType: "swarm-event", + EventType: payload.EventType, + DeploymentID: strings.TrimSpace(payload.DeploymentID), + SwarmID: strings.TrimSpace(payload.SwarmID), + AgentInstanceID: strings.TrimSpace(payload.AgentInstanceID), + TaskID: strings.TrimSpace(payload.TaskID), + UserID: record.Plan.UserContext.UserID, + BindingScope: firstPlanBindingScope(record.Plan), + CorrelationID: strings.TrimSpace(payload.CorrelationID), + Source: strings.TrimSpace(payload.Source), + Result: "ok", + PayloadJSON: string(payloadJSON), + OccurredAt: strings.TrimSpace(payload.OccurredAt), + CreatedAtMs: time.Now().UnixMilli(), }) if err != nil { common.SysLog("AgnetReceiveSwarmEventCallback: " + err.Error()) @@ -160,6 +454,11 @@ func AgnetReceiveSwarmEventCallback(c *gin.Context) { agnetError(c, "CALLBACK_PERSIST_FAILED", "failed to persist artifact") return } + if err := persistAgnetApprovalFromCallback(payload, record); err != nil { + common.SysLog("persistAgnetApprovalFromCallback: " + err.Error()) + agnetError(c, "CALLBACK_PERSIST_FAILED", "failed to persist approval request") + return + } recordAgnetAuditEvent(agnetEvent{ EventID: "evt_" + common.GetUUID()[:12], Event: "callback." + payload.EventType, @@ -169,13 +468,15 @@ func AgnetReceiveSwarmEventCallback(c *gin.Context) { BindingScope: firstPlanBindingScope(record.Plan), DeploymentID: strings.TrimSpace(payload.DeploymentID), CorrelationID: strings.TrimSpace(payload.CorrelationID), - OccurredAt: agnetNow(), + OccurredAt: firstNonEmpty(strings.TrimSpace(payload.OccurredAt), agnetNow()), }, "agnet_callback", strings.TrimSpace(payload.DeploymentID), agnetRequestID(c), "ok") } common.ApiSuccess(c, gin.H{ - "event_id": payload.EventID, - "inserted": inserted, - "idempotent": !inserted, + "event_id": payload.EventID, + "inserted": inserted, + "idempotent": !inserted, + "deduplicated": !inserted, + "deployment_id": payload.DeploymentID, }) } @@ -193,7 +494,40 @@ func AgnetListUserDeploymentArtifacts(c *gin.Context) { agnetError(c, "ARTIFACT_QUERY_FAILED", "failed to query artifacts") return } - common.ApiSuccess(c, gin.H{"items": items, "total": len(items)}) + common.ApiSuccess(c, gin.H{"deployment_id": record.DeploymentID, "artifacts": items, "items": items, "total": len(items)}) +} + +func callbackPayloadMap(row model.AgnetCallbackEvent) map[string]any { + var payload agnetCallbackEnvelope + if err := common.UnmarshalJsonStr(row.PayloadJSON, &payload); err != nil { + return nil + } + if payload.Payload != nil { + return payload.Payload + } + return nil +} + +func timelineEntryFromCallback(callback model.AgnetCallbackEvent) gin.H { + payload := callbackPayloadMap(callback) + entry := gin.H{ + "kind": "callback", + "at": callback.CreatedAtMs, + "event": callback.EventType, + "event_id": callback.EventID, + "event_type": callback.EventType, + "deployment_id": callback.DeploymentID, + "swarm_id": callback.SwarmID, + "agent_instance_id": callback.AgentInstanceID, + "occurred_at": callback.OccurredAt, + "payload": payload, + } + for _, key := range []string{"title", "summary", "stage", "checkpoint", "severity", "next_action", "agent_role"} { + if value, ok := payload[key]; ok { + entry[key] = value + } + } + return entry } func AgnetGetUserDeploymentTimeline(c *gin.Context) { @@ -226,7 +560,7 @@ func AgnetGetUserDeploymentTimeline(c *gin.Context) { timeline = append(timeline, gin.H{"kind": "audit", "at": event.OccurredAt, "event": event.Event}) } for _, callback := range callbacks { - timeline = append(timeline, gin.H{"kind": "callback", "at": callback.CreatedAtMs, "event": callback.EventType}) + timeline = append(timeline, timelineEntryFromCallback(callback)) } for _, artifact := range artifacts { timeline = append(timeline, gin.H{"kind": "artifact", "at": artifact.CreatedAtMs, "event": artifact.ArtifactType, "artifact_id": artifact.ArtifactID}) @@ -235,11 +569,12 @@ func AgnetGetUserDeploymentTimeline(c *gin.Context) { timeline = append(timeline, gin.H{"kind": "sk_snapshot", "at": snapshot.ResolvedAtMs, "event": snapshot.SourceType, "snapshot_id": snapshot.SnapshotID}) } common.ApiSuccess(c, gin.H{ - "deployment": record, - "events": events, - "callbacks": callbacks, - "artifacts": artifacts, - "sk_snapshots": snapshots, - "timeline": timeline, + "deployment_id": record.DeploymentID, + "deployment": record, + "events": events, + "callbacks": callbacks, + "artifacts": artifacts, + "sk_snapshots": snapshots, + "timeline": timeline, }) } diff --git a/heicode/controller/agnet_control_plane.go b/heicode/controller/agnet_control_plane.go index d8ec870b..a3847317 100644 --- a/heicode/controller/agnet_control_plane.go +++ b/heicode/controller/agnet_control_plane.go @@ -54,6 +54,15 @@ type agnetBillingContext struct { QuotaRef string `json:"quota_ref"` } +type agnetAgileContext struct { + Iteration string `json:"iteration"` + Stage string `json:"stage"` + Checkpoint string `json:"checkpoint"` + AcceptanceCriteria []string `json:"acceptance_criteria"` + NextAction string `json:"next_action"` + RequiresUserApproval bool `json:"requires_user_approval"` +} + type agnetRuntimeAgent struct { Role string `json:"role"` ModelRef string `json:"model_ref"` @@ -135,18 +144,20 @@ type agnetMetadata struct { } type agnetOrchestrationPlan struct { - IntentID string `json:"intent_id"` - TemplateHint string `json:"template_hint"` - Objective string `json:"objective"` - SubMode string `json:"sub_mode"` - RiskLevel string `json:"risk_level"` - Budget agnetBudget `json:"budget"` - UserContext agnetUserContext `json:"user_context"` - BillingContext agnetBillingContext `json:"billing_context"` - AgentRuntime agnetAgentRuntime `json:"agent_runtime"` - Agents []agnetAgentPlan `json:"agents"` - Constraints agnetConstraints `json:"constraints"` - Metadata agnetMetadata `json:"metadata"` + IntentID string `json:"intent_id"` + TemplateHint string `json:"template_hint"` + Objective string `json:"objective"` + SubMode string `json:"sub_mode"` + RiskLevel string `json:"risk_level"` + Budget agnetBudget `json:"budget"` + UserContext agnetUserContext `json:"user_context"` + BillingContext agnetBillingContext `json:"billing_context"` + AgileContext agnetAgileContext `json:"agile_context"` + AgentRuntime agnetAgentRuntime `json:"agent_runtime"` + Agents []agnetAgentPlan `json:"agents"` + ResourceGrants []agnetResourceGrant `json:"resource_grants"` + Constraints agnetConstraints `json:"constraints"` + Metadata agnetMetadata `json:"metadata"` } type agnetDeploymentRequest struct { @@ -159,6 +170,9 @@ type agnetDeploymentRecord struct { Status string `json:"status"` Phase string `json:"phase"` RuntimeState string `json:"runtime_state"` + RuntimeDeploymentID string `json:"runtime_deployment_id,omitempty"` + RuntimeSwarmID string `json:"runtime_swarm_id,omitempty"` + RuntimeLastSyncAt string `json:"runtime_last_sync_at,omitempty"` FailureReason string `json:"failure_reason"` AgentInstances []agnetAgentInstance `json:"agent_instances"` ResourceGrantManifest agnetPermissionManifest `json:"permission_manifest"` @@ -303,6 +317,9 @@ func agnetDeploymentModelToRecord(row model.AgnetDeployment) (agnetDeploymentRec record.Status = row.Status record.Phase = row.Phase record.RuntimeState = row.RuntimeState + record.RuntimeDeploymentID = row.RuntimeDeploymentID + record.RuntimeSwarmID = row.RuntimeSwarmID + record.RuntimeLastSyncAt = row.RuntimeLastSyncAtText record.FailureReason = row.FailureReason record.CreatedAt = row.CreatedAtText record.UpdatedAt = row.UpdatedAtText @@ -362,6 +379,9 @@ func persistAgnetDeploymentRecord(record agnetDeploymentRecord, req agnetDeploym Status: record.Status, Phase: record.Phase, RuntimeState: record.RuntimeState, + RuntimeDeploymentID: record.RuntimeDeploymentID, + RuntimeSwarmID: record.RuntimeSwarmID, + RuntimeLastSyncAtText: record.RuntimeLastSyncAt, FailureReason: record.FailureReason, CreatedAtText: record.CreatedAt, UpdatedAtText: record.UpdatedAt, @@ -390,15 +410,18 @@ func updateAgnetDeploymentRecord(record agnetDeploymentRecord) error { return model.DB.Model(&model.AgnetDeployment{}). Where("deployment_id = ?", record.DeploymentID). Updates(map[string]any{ - "status": record.Status, - "sub_mode": normalizeAgnetSubMode(record.SubMode), - "phase": record.Phase, - "runtime_state": record.RuntimeState, - "failure_reason": record.FailureReason, - "updated_at_text": record.UpdatedAt, - "updated_at_ms": agnetTimestampMs(record.UpdatedAt), - "agent_instances_json": instancesJSON, - "permission_manifest_json": manifestJSON, + "status": record.Status, + "sub_mode": normalizeAgnetSubMode(record.SubMode), + "phase": record.Phase, + "runtime_state": record.RuntimeState, + "runtime_deployment_id": record.RuntimeDeploymentID, + "runtime_swarm_id": record.RuntimeSwarmID, + "runtime_last_sync_at_text": record.RuntimeLastSyncAt, + "failure_reason": record.FailureReason, + "updated_at_text": record.UpdatedAt, + "updated_at_ms": agnetTimestampMs(record.UpdatedAt), + "agent_instances_json": instancesJSON, + "permission_manifest_json": manifestJSON, }).Error } @@ -714,6 +737,10 @@ func validateResourceGrant(c *gin.Context, plan agnetOrchestrationPlan, agent ag agnetError(c, "RESOURCE_GRANT_SECRET_REF_REQUIRED", "resource_grants.secret_ref is required for credential-backed resources") return false } + if strings.TrimSpace(grant.SecretRef) != "" && !strings.HasPrefix(strings.TrimSpace(grant.SecretRef), "azkv://") { + agnetError(c, "SECRET_REF_INVALID", "resource_grants.secret_ref must use azkv:// Azure Key Vault reference") + return false + } if containsSensitiveGrantField(grant.Metadata) || containsSensitiveGrantField(grant.Constraints) || containsSensitiveGrantField(grant.Audit) { agnetError(c, "RESOURCE_GRANT_SECRET_REJECTED", "resource_grants metadata/constraints/audit must not contain plaintext credential fields") return false @@ -793,11 +820,30 @@ func validateOrchestrationPlan(c *gin.Context, plan agnetOrchestrationPlan) bool } } } + agentsByRole := make(map[string]agnetAgentPlan, len(plan.Agents)) + for _, agent := range plan.Agents { + agentsByRole[strings.TrimSpace(agent.RoleTemplate)] = agent + } + for _, grant := range plan.ResourceGrants { + agent, ok := agentsByRole[strings.TrimSpace(grant.TargetRole)] + if !ok { + agnetError(c, "RESOURCE_GRANT_INVALID", "top-level resource_grants.target_role must match an orchestration agent role") + return false + } + if !validateResourceGrant(c, plan, agent, grant) { + return false + } + } return true } func firstPlanBindingScope(plan agnetOrchestrationPlan) string { + for _, grant := range plan.ResourceGrants { + if scope := strings.TrimSpace(grant.BindingScope); scope != "" { + return scope + } + } for _, agent := range plan.Agents { for _, grant := range agent.ResourceGrants { if scope := strings.TrimSpace(grant.BindingScope); scope != "" { @@ -817,32 +863,39 @@ func agnetGrantResourceRef(grant agnetResourceGrant) string { return grant.GrantID } +func appendAgnetManifestGrant(manifest *agnetPermissionManifest, grant agnetResourceGrant) { + if manifest == nil || strings.TrimSpace(grant.Status) != agnetGrantStatusActive { + return + } + if manifest.AgentRole == "" { + manifest.AgentRole = grant.TargetRole + } + if manifest.TargetAgentRef == "" { + manifest.TargetAgentRef = grant.TargetAgentRef + } + manifest.ResourceGrants = append(manifest.ResourceGrants, agnetManifestGrant{ + GrantID: grant.GrantID, + ResourceID: grant.ResourceID, + ResourceType: grant.ResourceType, + ResourceRef: agnetGrantResourceRef(grant), + AllowedActions: append([]string{}, grant.PermissionScope...), + Constraints: grant.Constraints, + SecretRef: grant.SecretRef, + Status: grant.Status, + }) +} + func buildAgnetPermissionManifest(plan agnetOrchestrationPlan) agnetPermissionManifest { manifest := agnetPermissionManifest{ UserID: plan.UserContext.UserID, BindingScope: firstPlanBindingScope(plan), } + for _, grant := range plan.ResourceGrants { + appendAgnetManifestGrant(&manifest, grant) + } for _, agent := range plan.Agents { for _, grant := range agent.ResourceGrants { - if strings.TrimSpace(grant.Status) != agnetGrantStatusActive { - continue - } - if manifest.AgentRole == "" { - manifest.AgentRole = grant.TargetRole - } - if manifest.TargetAgentRef == "" { - manifest.TargetAgentRef = grant.TargetAgentRef - } - manifest.ResourceGrants = append(manifest.ResourceGrants, agnetManifestGrant{ - GrantID: grant.GrantID, - ResourceID: grant.ResourceID, - ResourceType: grant.ResourceType, - ResourceRef: agnetGrantResourceRef(grant), - AllowedActions: append([]string{}, grant.PermissionScope...), - Constraints: grant.Constraints, - SecretRef: grant.SecretRef, - Status: grant.Status, - }) + appendAgnetManifestGrant(&manifest, grant) } } if manifest.ResourceGrants == nil { @@ -894,6 +947,11 @@ func planHasBindingScope(plan agnetOrchestrationPlan, bindingScope string) bool } } } + for _, grant := range plan.ResourceGrants { + if strings.TrimSpace(grant.BindingScope) == bindingScope { + return true + } + } return false } @@ -947,6 +1005,14 @@ func enforceAuthenticatedAgnetUserContext(c *gin.Context, plan *agnetOrchestrati plan.Agents[agentIdx].ResourceGrants[grantIdx].UserID = userID } } + for grantIdx := range plan.ResourceGrants { + grantUserID := strings.TrimSpace(plan.ResourceGrants[grantIdx].UserID) + if grantUserID != "" && grantUserID != userID { + agnetError(c, "RESOURCE_GRANT_FORBIDDEN", "resource_grants.user_id must match authenticated user") + return false + } + plan.ResourceGrants[grantIdx].UserID = userID + } return true } @@ -1033,14 +1099,17 @@ func createAgnetDeploymentRecord(c *gin.Context, enforceUserScope bool) (agnetDe func writeAgnetDeploymentCreateSuccess(c *gin.Context, record agnetDeploymentRecord, extra gin.H) { data := gin.H{ - "deployment_id": record.DeploymentID, - "sub_mode": record.SubMode, - "status": record.Status, - "phase": record.Phase, - "runtime_state": record.RuntimeState, - "failure_reason": record.FailureReason, - "agent_instances": record.AgentInstances, - "permission_manifest": record.ResourceGrantManifest, + "deployment_id": record.DeploymentID, + "sub_mode": record.SubMode, + "status": record.Status, + "phase": record.Phase, + "runtime_state": record.RuntimeState, + "runtime_deployment_id": record.RuntimeDeploymentID, + "runtime_swarm_id": record.RuntimeSwarmID, + "runtime_last_sync_at": record.RuntimeLastSyncAt, + "failure_reason": record.FailureReason, + "agent_instances": record.AgentInstances, + "permission_manifest": record.ResourceGrantManifest, } for key, value := range extra { data[key] = value @@ -1053,6 +1122,7 @@ func createAgnetDeployment(c *gin.Context, enforceUserScope bool) { if !ok { return } + record = maybeDispatchAgnetRuntimeCreate(c, record, "agnet_deployments") writeAgnetDeploymentCreateSuccess(c, record, nil) } @@ -1061,8 +1131,9 @@ func AgnetCreateUserSwarm(c *gin.Context) { if !ok { return } + record = maybeDispatchAgnetRuntimeCreate(c, record, "api_swarms_adapter") writeAgnetDeploymentCreateSuccess(c, record, gin.H{ - "swarm_id": record.DeploymentID, + "swarm_id": firstNonEmpty(record.RuntimeSwarmID, record.DeploymentID), "adapter": "manager-local-control-plane", "source": "api_swarms_adapter", }) @@ -1540,6 +1611,7 @@ func AgnetListSKSnapshots(c *gin.Context) { } common.ApiSuccess(c, gin.H{ "deployment_id": deploymentID, + "sk_snapshots": items, "items": items, "total": len(items), }) diff --git a/heicode/controller/agnet_control_plane_test.go b/heicode/controller/agnet_control_plane_test.go index 2dbd978e..67ac1532 100644 --- a/heicode/controller/agnet_control_plane_test.go +++ b/heicode/controller/agnet_control_plane_test.go @@ -1,11 +1,17 @@ package controller import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" "fmt" + "io" "net/http" "net/http/httptest" + "strconv" "strings" "testing" + "time" "github.com/gin-gonic/gin" "github.com/glebarez/sqlite" @@ -59,6 +65,7 @@ func setupAgnetControlPlaneTestDB(t *testing.T) *gorm.DB { &model.AgnetCallbackEvent{}, &model.AgnetArtifact{}, &model.AgnetSKSnapshot{}, + &model.AgnetApprovalRequest{}, )) t.Cleanup(func() { if model.DB == db { @@ -296,6 +303,147 @@ func TestAgnetDeploymentDefaultsSubModeToAgile(t *testing.T) { require.Contains(t, recorder.Body.String(), `"sub_mode":"agile"`) } +func TestAgnetRuntimeHealthNotConfigured(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + t.Setenv("AGNET_RUNTIME_BASE_URL", "") + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Request = httptest.NewRequest(http.MethodGet, "/api/agnet/runtime/health", nil) + + AgnetRuntimeHealth(ctx) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Contains(t, recorder.Body.String(), `"status":"not_configured"`) + require.Contains(t, recorder.Body.String(), `"configured":false`) +} + +func TestAgnetRuntimeHealthProxy(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "/api/agnet/health", r.URL.Path) + require.Equal(t, "Bearer service-token", r.Header.Get("Authorization")) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"status":"healthy"}}`)) + })) + defer server.Close() + + t.Setenv("AGNET_RUNTIME_BASE_URL", server.URL) + t.Setenv("AGNET_RUNTIME_SERVICE_TOKEN", "service-token") + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Request = httptest.NewRequest(http.MethodGet, "/api/agnet/runtime/health", nil) + + AgnetRuntimeHealth(ctx) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Contains(t, recorder.Body.String(), `"status":"healthy"`) + require.Contains(t, recorder.Body.String(), `"http_status":200`) + require.Contains(t, recorder.Body.String(), `"remote"`) +} + +func TestAgnetRuntimeShadowCreateStoresRuntimeMapping(t *testing.T) { + db := setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "/api/agnet/deployments", r.URL.Path) + require.Equal(t, http.MethodPost, r.Method) + require.Equal(t, "Bearer service-token", r.Header.Get("Authorization")) + require.NotEmpty(t, r.Header.Get("X-Correlation-ID")) + body, err := io.ReadAll(r.Body) + require.NoError(t, err) + require.Contains(t, string(body), `"sub_mode":"agile"`) + require.Contains(t, string(body), `"manager_deployment_id"`) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"deployment_id":"runtime-dep-1","swarm_id":"swarm-1","runtime_status":"accepted"}}`)) + })) + defer server.Close() + + t.Setenv("AGNET_RUNTIME_ENABLED", "true") + t.Setenv("AGNET_RUNTIME_ASYNC", "false") + t.Setenv("AGNET_RUNTIME_BASE_URL", server.URL) + t.Setenv("AGNET_RUNTIME_SERVICE_TOKEN", "service-token") + + plan := baseAgnetResourceGrantPlan() + plan.SubMode = "agile" + recorder, envelope := postAgnetCreateDeployment(t, plan) + require.Equal(t, http.StatusOK, recorder.Code) + require.True(t, envelope.Success) + require.Contains(t, recorder.Body.String(), `"runtime_deployment_id":"runtime-dep-1"`) + require.Contains(t, recorder.Body.String(), `"runtime_swarm_id":"swarm-1"`) + + var createBody map[string]any + require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &createBody)) + deploymentID := createBody["data"].(map[string]any)["deployment_id"].(string) + + var stored model.AgnetDeployment + require.NoError(t, db.Where("deployment_id = ?", deploymentID).First(&stored).Error) + require.Equal(t, "runtime-dep-1", stored.RuntimeDeploymentID) + require.Equal(t, "swarm-1", stored.RuntimeSwarmID) + require.Equal(t, "accepted", stored.RuntimeState) +} + +func TestAgnetRuntimeShadowCreateFailureDoesNotFailLocalDeployment(t *testing.T) { + db := setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "runtime unavailable", http.StatusBadGateway) + })) + defer server.Close() + + t.Setenv("AGNET_RUNTIME_ENABLED", "true") + t.Setenv("AGNET_RUNTIME_ASYNC", "false") + t.Setenv("AGNET_RUNTIME_BASE_URL", server.URL) + + recorder, envelope := postAgnetCreateDeployment(t, baseAgnetResourceGrantPlan()) + require.Equal(t, http.StatusOK, recorder.Code) + require.True(t, envelope.Success) + require.Contains(t, recorder.Body.String(), `"runtime_state":"runtime_sync_failed"`) + + var createBody map[string]any + require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &createBody)) + deploymentID := createBody["data"].(map[string]any)["deployment_id"].(string) + + var stored model.AgnetDeployment + require.NoError(t, db.Where("deployment_id = ?", deploymentID).First(&stored).Error) + require.Equal(t, agnetRuntimeStateFailed, stored.RuntimeState) + require.Contains(t, stored.FailureReason, "HTTP 502") +} + +func TestAgnetRuntimeShadowCreateTreatsSuccessFalseAsFailure(t *testing.T) { + db := setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":false,"error":{"code":"POLICY_REJECTED","message":"sub mode rejected"}}`)) + })) + defer server.Close() + + t.Setenv("AGNET_RUNTIME_ENABLED", "true") + t.Setenv("AGNET_RUNTIME_ASYNC", "false") + t.Setenv("AGNET_RUNTIME_BASE_URL", server.URL) + + recorder, envelope := postAgnetCreateDeployment(t, baseAgnetResourceGrantPlan()) + require.Equal(t, http.StatusOK, recorder.Code) + require.True(t, envelope.Success) + + var createBody map[string]any + require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &createBody)) + deploymentID := createBody["data"].(map[string]any)["deployment_id"].(string) + + var stored model.AgnetDeployment + require.NoError(t, db.Where("deployment_id = ?", deploymentID).First(&stored).Error) + require.Equal(t, agnetRuntimeStateFailed, stored.RuntimeState) + require.Contains(t, stored.FailureReason, "sub mode rejected") +} + func TestAgnetCreateDeploymentRejectsInvalidSubMode(t *testing.T) { setupAgnetControlPlaneTestDB(t) resetAgnetControlPlaneState(t) @@ -309,6 +457,37 @@ func TestAgnetCreateDeploymentRejectsInvalidSubMode(t *testing.T) { require.Empty(t, agnetDeployments) } +func TestAgnetCreateDeploymentRejectsNonAzureSecretRef(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + plan := baseAgnetResourceGrantPlan() + plan.Agents[0].ResourceGrants[0].SecretRef = "vault:heicode/repo-main" + + recorder, envelope := postAgnetCreateDeployment(t, plan) + + require.Equal(t, http.StatusOK, recorder.Code) + require.False(t, envelope.Success) + require.Equal(t, "SECRET_REF_INVALID", envelope.Error.Code) +} + +func TestAgnetCreateDeploymentAcceptsTopLevelResourceGrants(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + plan := baseAgnetResourceGrantPlan() + plan.ResourceGrants = plan.Agents[0].ResourceGrants + plan.Agents[0].ResourceGrants = nil + + recorder, envelope := postAgnetCreateDeployment(t, plan) + + require.Equal(t, http.StatusOK, recorder.Code) + require.True(t, envelope.Success) + require.Contains(t, recorder.Body.String(), `"resource_grants":[`) + require.Contains(t, recorder.Body.String(), `"grant_id":"grant-git-builder"`) + require.Contains(t, recorder.Body.String(), `"secret_ref":"azkv://heicode-kv.vault.azure.net/secrets/user-p1-project-main-res-git-main"`) +} + func TestAgnetUserDeploymentForcesAuthenticatedUserScope(t *testing.T) { db := setupAgnetControlPlaneTestDB(t) resetAgnetControlPlaneState(t) @@ -668,6 +847,199 @@ func TestAgnetCallbackStoresEventArtifactAndIsIdempotent(t *testing.T) { require.Equal(t, int64(1), artifactCount) } +func TestAgnetCallbackAcceptsDocumentedHMACPayloadArtifact(t *testing.T) { + db := setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + t.Setenv("AGNET_CALLBACK_SIGNING_SECRET", "callback-signing-secret") + + 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.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_type":"artifact.created", + "deployment_id":%q, + "swarm_id":"swarm-doc-1", + "agent_instance_id":"agi-backend-1", + "occurred_at":"2026-05-26T10:40:00Z", + "payload":{ + "artifact_id":"art-doc-payload-1", + "artifact_type":"code_patch", + "title":"Backend API patch", + "summary":"Patch created", + "uri":"azblob://heicode-artifacts/task-123/backend.patch", + "checksum":"sha256:abc", + "stage":"development", + "checkpoint":"artifact_ready" + } + }`, deploymentID) + eventID := "evt-doc-hmac-1" + timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10) + mac := hmac.New(sha256.New, []byte("callback-signing-secret")) + mac.Write([]byte(timestamp + "." + eventID + "." + body)) + signature := "sha256=" + hex.EncodeToString(mac.Sum(nil)) + + 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-Event-Id", eventID) + ctx.Request.Header.Set("X-Agnet-Timestamp", timestamp) + ctx.Request.Header.Set("X-Agnet-Signature", signature) + ctx.Request.Header.Set("X-Correlation-ID", "corr-doc-hmac") + + AgnetReceiveSwarmEventCallback(ctx) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Contains(t, recorder.Body.String(), `"success":true`) + + var callback model.AgnetCallbackEvent + require.NoError(t, db.Where("event_id = ?", eventID).First(&callback).Error) + require.Equal(t, "swarm-doc-1", callback.SwarmID) + require.Equal(t, "agi-backend-1", callback.AgentInstanceID) + require.Equal(t, "2026-05-26T10:40:00Z", callback.OccurredAt) + + var artifact model.AgnetArtifact + require.NoError(t, db.Where("artifact_id = ?", "art-doc-payload-1").First(&artifact).Error) + require.Equal(t, "code_patch", artifact.ArtifactType) + require.Contains(t, artifact.MetadataJSON, "artifact_ready") +} + +func TestAgnetCallbackHMACCanUseAzureKeyVaultSigningSecretRef(t *testing.T) { + db := setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + tokenServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "true", r.Header.Get("Metadata")) + _, _ = w.Write([]byte(`{"access_token":"manager-token","expires_in":"3600"}`)) + })) + defer tokenServer.Close() + + vaultServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, http.MethodGet, r.Method) + require.Equal(t, "Bearer manager-token", r.Header.Get("Authorization")) + require.Equal(t, "/secrets/callback-signing", r.URL.Path) + _, _ = w.Write([]byte(`{"value":"{\"callback_signing_secret\":\"from-kv-secret\"}"}`)) + })) + defer vaultServer.Close() + t.Setenv("AZURE_KEY_VAULT_URL", vaultServer.URL) + t.Setenv("AZURE_MANAGED_IDENTITY_TOKEN_URL", tokenServer.URL) + t.Setenv("AGNET_CALLBACK_SIGNING_SECRET_REF", fmt.Sprintf("azkv://%s/secrets/callback-signing", strings.TrimPrefix(vaultServer.URL, "http://"))) + + 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.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_type":"timeline.updated","deployment_id":%q,"payload":{"title":"Ready"}}`, deploymentID) + eventID := "evt-kv-hmac-1" + timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10) + mac := hmac.New(sha256.New, []byte("from-kv-secret")) + mac.Write([]byte(timestamp + "." + eventID + "." + body)) + signature := "sha256=" + hex.EncodeToString(mac.Sum(nil)) + + 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-Event-Id", eventID) + ctx.Request.Header.Set("X-Agnet-Timestamp", timestamp) + ctx.Request.Header.Set("X-Agnet-Signature", signature) + + AgnetReceiveSwarmEventCallback(ctx) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Contains(t, recorder.Body.String(), `"success":true`) + var callback model.AgnetCallbackEvent + require.NoError(t, db.Where("event_id = ?", eventID).First(&callback).Error) +} + +func TestAgnetCallbackUsesSwarmIDFallbackAndCreatesApproval(t *testing.T) { + db := 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.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) + swarmID := "runtime-swarm-only-1" + require.NoError(t, db.Model(&model.AgnetDeployment{}). + Where("deployment_id = ?", deploymentID). + Update("runtime_swarm_id", swarmID).Error) + resetAgnetControlPlaneState(t) + + body := `{ + "event_id":"evt-approval-swarm-only", + "event_type":"approval.requested", + "swarm_id":"runtime-swarm-only-1", + "agent_instance_id":"agi-backend-approval", + "occurred_at":"2026-05-27T10:41:00Z", + "payload":{ + "approval_id":"runtime-approval-1", + "operation":"repo.write", + "resource_id":"res-git-main", + "resource_type":"git", + "resource_scope":"heicode/", + "target_role":"builder", + "risk_level":"high", + "requires_credential":true, + "secret_ref":"azkv://heicode-kv.vault.azure.net/secrets/repo-main", + "ttl_seconds":600, + "reason":"Runtime needs write access for backend patch" + } + }` + + for i := 0; i < 2; i++ { + 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) + require.Equal(t, http.StatusOK, recorder.Code) + require.Contains(t, recorder.Body.String(), `"success":true`) + if i == 0 { + require.Contains(t, recorder.Body.String(), `"deduplicated":false`) + } else { + require.Contains(t, recorder.Body.String(), `"deduplicated":true`) + } + } + + var callback model.AgnetCallbackEvent + require.NoError(t, db.Where("event_id = ?", "evt-approval-swarm-only").First(&callback).Error) + require.Equal(t, deploymentID, callback.DeploymentID) + require.Equal(t, swarmID, callback.SwarmID) + require.Equal(t, "agi-backend-approval", callback.AgentInstanceID) + + var approval model.AgnetApprovalRequest + require.NoError(t, db.Where("approval_id = ?", "runtime-approval-1").First(&approval).Error) + require.Equal(t, 7, approval.UserId) + require.Equal(t, deploymentID, approval.DeploymentID) + require.Equal(t, "pending", approval.Status) + require.Equal(t, "repo.write", approval.Operation) + require.Equal(t, "azkv://heicode-kv.vault.azure.net/secrets/repo-main", approval.SecretRef) +} + func TestAgnetCallbackRejectsPlaintextSecretsAndMissingToken(t *testing.T) { setupAgnetControlPlaneTestDB(t) resetAgnetControlPlaneState(t) @@ -751,6 +1123,8 @@ func TestAgnetArtifactsAndTimelineReturnPersistedRelatedRecords(t *testing.T) { AgnetListUserDeploymentArtifacts(artifactCtx) require.Equal(t, http.StatusOK, artifactRecorder.Code) require.Contains(t, artifactRecorder.Body.String(), `"art-timeline"`) + require.Contains(t, artifactRecorder.Body.String(), `"deployment_id":"`+deploymentID+`"`) + require.Contains(t, artifactRecorder.Body.String(), `"artifacts":[`) snapshotListRecorder := httptest.NewRecorder() snapshotListCtx, _ := gin.CreateTestContext(snapshotListRecorder) @@ -760,6 +1134,7 @@ func TestAgnetArtifactsAndTimelineReturnPersistedRelatedRecords(t *testing.T) { AgnetListUserSKSnapshots(snapshotListCtx) require.Equal(t, http.StatusOK, snapshotListRecorder.Code) require.Contains(t, snapshotListRecorder.Body.String(), `"deployment_id":"`+deploymentID+`"`) + require.Contains(t, snapshotListRecorder.Body.String(), `"sk_snapshots":[`) timelineRecorder := httptest.NewRecorder() timelineCtx, _ := gin.CreateTestContext(timelineRecorder) @@ -772,6 +1147,9 @@ func TestAgnetArtifactsAndTimelineReturnPersistedRelatedRecords(t *testing.T) { require.Contains(t, timelineRecorder.Body.String(), `"artifacts"`) require.Contains(t, timelineRecorder.Body.String(), `"sk_snapshots"`) require.Contains(t, timelineRecorder.Body.String(), `"timeline"`) + require.Contains(t, timelineRecorder.Body.String(), `"deployment_id":"`+deploymentID+`"`) + require.Contains(t, timelineRecorder.Body.String(), `"event_id":"evt-artifact-timeline"`) + require.Contains(t, timelineRecorder.Body.String(), `"event_type":"artifact.created"`) } func TestAgnetStopDeploymentPersistsState(t *testing.T) { diff --git a/heicode/controller/agnet_runtime_client.go b/heicode/controller/agnet_runtime_client.go new file mode 100644 index 00000000..d6a58b1d --- /dev/null +++ b/heicode/controller/agnet_runtime_client.go @@ -0,0 +1,368 @@ +package controller + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/gin-gonic/gin" + "github.com/heicode/manager/common" +) + +const ( + agnetRuntimeStateSyncing = "runtime_syncing" + agnetRuntimeStateSynced = "runtime_accepted" + agnetRuntimeStateFailed = "runtime_sync_failed" +) + +type agnetRuntimeConfig struct { + Enabled bool + Async bool + BaseURL string + Token string + CreatePath string + HealthPath string + Timeout time.Duration +} + +type agnetRuntimeSyncResult struct { + RuntimeDeploymentID string + RuntimeSwarmID string + RuntimeStatus string + RawStatusCode int +} + +func agnetRuntimeClientConfig() agnetRuntimeConfig { + timeoutSec := common.GetEnvOrDefault("AGNET_RUNTIME_TIMEOUT_SECONDS", 5) + if timeoutSec <= 0 { + timeoutSec = 5 + } + return agnetRuntimeConfig{ + Enabled: common.GetEnvOrDefaultBool("AGNET_RUNTIME_ENABLED", false), + Async: common.GetEnvOrDefaultBool("AGNET_RUNTIME_ASYNC", true), + BaseURL: strings.TrimRight(strings.TrimSpace(common.GetEnvOrDefaultString("AGNET_RUNTIME_BASE_URL", "")), "/"), + Token: strings.TrimSpace(common.GetEnvOrDefaultString("AGNET_RUNTIME_SERVICE_TOKEN", "")), + CreatePath: common.GetEnvOrDefaultString("AGNET_RUNTIME_CREATE_PATH", "/api/agnet/deployments"), + HealthPath: common.GetEnvOrDefaultString("AGNET_RUNTIME_HEALTH_PATH", "/api/agnet/health"), + Timeout: time.Duration(timeoutSec) * time.Second, + } +} + +func agnetRuntimeCallbackURL() string { + if value := strings.TrimSpace(common.GetEnvOrDefaultString("AGNET_RUNTIME_CALLBACK_URL", "")); value != "" { + return value + } + baseURL := strings.TrimRight(strings.TrimSpace(common.GetEnvOrDefaultString("HEICODE_PUBLIC_BASE_URL", "https://code.xinghanlab.com")), "/") + return baseURL + "/api/agnet/callbacks/swarm-events" +} + +func agnetRuntimeCallbackSigningSecretRef() string { + return strings.TrimSpace(common.GetEnvOrDefaultString("AGNET_RUNTIME_CALLBACK_SIGNING_SECRET_REF", "")) +} + +func agnetRuntimeSubscribedEvents() []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", + } +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return strings.TrimSpace(value) + } + } + return "" +} + +func truncateAgnetFailureReason(value string) string { + value = strings.TrimSpace(value) + if len(value) <= 480 { + return value + } + return value[:480] +} + +func agnetRuntimeURL(baseURL string, path string) (string, error) { + if strings.TrimSpace(baseURL) == "" { + return "", errors.New("AGNET_RUNTIME_BASE_URL is not configured") + } + parsed, err := url.Parse(baseURL) + if err != nil || parsed.Scheme == "" || parsed.Host == "" { + return "", errors.New("AGNET_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 agnetRuntimeHeaders(req *http.Request, cfg agnetRuntimeConfig, record agnetDeploymentRecord) { + 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 agnetRuntimeCreatePayload(record agnetDeploymentRecord, source string) gin.H { + callback := gin.H{ + "url": agnetRuntimeCallbackURL(), + "subscribed_events": agnetRuntimeSubscribedEvents(), + } + if ref := agnetRuntimeCallbackSigningSecretRef(); ref != "" { + callback["signing_secret_ref"] = ref + } + return gin.H{ + "deployment_id": record.DeploymentID, + "manager_deployment_id": record.DeploymentID, + "source": source, + "callback": callback, + "orchestration_plan": record.Plan, + } +} + +func extractAgnetRuntimeData(payload map[string]any) map[string]any { + if data, ok := payload["data"].(map[string]any); ok { + return data + } + return payload +} + +func agnetRuntimeEnvelopeError(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 callAgnetRuntimeCreate(ctx context.Context, cfg agnetRuntimeConfig, record agnetDeploymentRecord, source string) (agnetRuntimeSyncResult, error) { + endpoint, err := agnetRuntimeURL(cfg.BaseURL, cfg.CreatePath) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + payload, err := common.Marshal(agnetRuntimeCreatePayload(record, source)) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload)) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + agnetRuntimeHeaders(req, cfg, record) + client := &http.Client{Timeout: cfg.Timeout} + resp, err := client.Do(req) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + defer resp.Body.Close() + body, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if readErr != nil { + return agnetRuntimeSyncResult{}, readErr + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return agnetRuntimeSyncResult{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 agnetRuntimeSyncResult{RawStatusCode: resp.StatusCode}, err + } + } + if message := agnetRuntimeEnvelopeError(envelope); message != "" { + return agnetRuntimeSyncResult{RawStatusCode: resp.StatusCode}, errors.New(message) + } + data := extractAgnetRuntimeData(envelope) + result := agnetRuntimeSyncResult{ + 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 updateAgnetRuntimeSyncState(record agnetDeploymentRecord, result agnetRuntimeSyncResult, syncErr error) agnetDeploymentRecord { + record.RuntimeLastSyncAt = agnetNow() + if syncErr != nil { + record.RuntimeState = agnetRuntimeStateFailed + record.FailureReason = truncateAgnetFailureReason(syncErr.Error()) + } else { + record.RuntimeState = firstNonEmpty(result.RuntimeStatus, agnetRuntimeStateSynced) + record.RuntimeDeploymentID = result.RuntimeDeploymentID + record.RuntimeSwarmID = result.RuntimeSwarmID + record.FailureReason = "" + } + record.UpdatedAt = agnetNow() + if err := updateAgnetDeploymentRecord(record); err != nil { + common.SysLog("updateAgnetRuntimeSyncState: " + err.Error()) + } + agnetMu.Lock() + agnetDeployments[record.DeploymentID] = record + agnetMu.Unlock() + return record +} + +func recordAgnetRuntimeSyncAudit(record agnetDeploymentRecord, event string, result string) { + recordAgnetAuditEvent(agnetEvent{ + 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: agnetNow(), + }, "agnet_runtime_bridge", record.DeploymentID, "", result) +} + +func dispatchAgnetRuntimeCreate(record agnetDeploymentRecord, source string, cfg agnetRuntimeConfig) agnetDeploymentRecord { + recordAgnetRuntimeSyncAudit(record, "runtime.sync.started", "started") + ctx, cancel := context.WithTimeout(context.Background(), cfg.Timeout) + defer cancel() + result, err := callAgnetRuntimeCreate(ctx, cfg, record, source) + record = updateAgnetRuntimeSyncState(record, result, err) + if err != nil { + common.SysLog("Agnet runtime shadow create failed for " + record.DeploymentID + ": " + err.Error()) + recordAgnetRuntimeSyncAudit(record, "runtime.sync.failed", "failed") + return record + } + recordAgnetRuntimeSyncAudit(record, "runtime.sync.accepted", "ok") + return record +} + +func maybeDispatchAgnetRuntimeCreate(c *gin.Context, record agnetDeploymentRecord, source string) agnetDeploymentRecord { + cfg := agnetRuntimeClientConfig() + if !cfg.Enabled { + return record + } + if _, err := agnetRuntimeURL(cfg.BaseURL, cfg.CreatePath); err != nil { + record.RuntimeState = agnetRuntimeStateFailed + record.RuntimeLastSyncAt = agnetNow() + record.FailureReason = truncateAgnetFailureReason(err.Error()) + record.UpdatedAt = agnetNow() + _ = updateAgnetDeploymentRecord(record) + recordAgnetRuntimeSyncAudit(record, "runtime.sync.failed", "failed") + return record + } + record.RuntimeState = agnetRuntimeStateSyncing + record.RuntimeLastSyncAt = agnetNow() + record.UpdatedAt = agnetNow() + if err := updateAgnetDeploymentRecord(record); err != nil { + common.SysLog("maybeDispatchAgnetRuntimeCreate: " + err.Error()) + } + agnetMu.Lock() + agnetDeployments[record.DeploymentID] = record + agnetMu.Unlock() + if cfg.Async { + sourceCopy := source + recordCopy := record + go dispatchAgnetRuntimeCreate(recordCopy, sourceCopy, cfg) + return record + } + return dispatchAgnetRuntimeCreate(record, source, cfg) +} + +func AgnetRuntimeHealth(c *gin.Context) { + cfg := agnetRuntimeClientConfig() + data := gin.H{ + "enabled": cfg.Enabled, + "configured": cfg.BaseURL != "", + "create_path": cfg.CreatePath, + "health_path": cfg.HealthPath, + } + if cfg.BaseURL == "" { + data["status"] = "not_configured" + common.ApiSuccess(c, data) + return + } + endpoint, err := agnetRuntimeURL(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) +} diff --git a/heicode/model/agnet_callback.go b/heicode/model/agnet_callback.go index f7caf0ec..27ec8bf3 100644 --- a/heicode/model/agnet_callback.go +++ b/heicode/model/agnet_callback.go @@ -3,20 +3,23 @@ package model import "errors" type AgnetCallbackEvent struct { - Id int `gorm:"primaryKey" json:"id"` - EventID string `gorm:"type:varchar(128);uniqueIndex" json:"event_id"` - IdempotencyKey string `gorm:"type:varchar(128);index" json:"idempotency_key"` - CallbackType string `gorm:"type:varchar(64);index" json:"callback_type"` - EventType string `gorm:"type:varchar(64);index" json:"event_type"` - DeploymentID string `gorm:"type:varchar(64);index" json:"deployment_id"` - TaskID string `gorm:"type:varchar(128);index" json:"task_id"` - UserID string `gorm:"type:varchar(64);index" json:"user_id"` - BindingScope string `gorm:"type:varchar(512);index" json:"binding_scope"` - CorrelationID string `gorm:"type:varchar(64);index" json:"correlation_id"` - Source string `gorm:"type:varchar(64)" json:"source"` - Result string `gorm:"type:varchar(32)" json:"result"` - PayloadJSON string `gorm:"type:text" json:"payload_json"` - CreatedAtMs int64 `gorm:"bigint;index" json:"created_at_ms"` + Id int `gorm:"primaryKey" json:"id"` + EventID string `gorm:"type:varchar(128);uniqueIndex" json:"event_id"` + IdempotencyKey string `gorm:"type:varchar(128);index" json:"idempotency_key"` + CallbackType string `gorm:"type:varchar(64);index" json:"callback_type"` + EventType string `gorm:"type:varchar(64);index" json:"event_type"` + DeploymentID string `gorm:"type:varchar(64);index" json:"deployment_id"` + SwarmID string `gorm:"type:varchar(128);index" json:"swarm_id"` + AgentInstanceID string `gorm:"type:varchar(128);index" json:"agent_instance_id"` + TaskID string `gorm:"type:varchar(128);index" json:"task_id"` + UserID string `gorm:"type:varchar(64);index" json:"user_id"` + BindingScope string `gorm:"type:varchar(512);index" json:"binding_scope"` + CorrelationID string `gorm:"type:varchar(64);index" json:"correlation_id"` + Source string `gorm:"type:varchar(64)" json:"source"` + Result string `gorm:"type:varchar(32)" json:"result"` + PayloadJSON string `gorm:"type:text" json:"payload_json"` + OccurredAt string `gorm:"type:varchar(32)" json:"occurred_at"` + CreatedAtMs int64 `gorm:"bigint;index" json:"created_at_ms"` } func (AgnetCallbackEvent) TableName() string { diff --git a/heicode/model/agnet_deployment.go b/heicode/model/agnet_deployment.go index 870d3066..1615b24d 100644 --- a/heicode/model/agnet_deployment.go +++ b/heicode/model/agnet_deployment.go @@ -15,6 +15,9 @@ type AgnetDeployment struct { Status string `gorm:"type:varchar(32);index" json:"status"` Phase string `gorm:"type:varchar(32);index" json:"phase"` RuntimeState string `gorm:"type:varchar(32)" json:"runtime_state"` + RuntimeDeploymentID string `gorm:"type:varchar(128);index" json:"runtime_deployment_id"` + RuntimeSwarmID string `gorm:"type:varchar(128);index" json:"runtime_swarm_id"` + RuntimeLastSyncAtText string `gorm:"type:varchar(32)" json:"runtime_last_sync_at"` FailureReason string `gorm:"type:varchar(512)" json:"failure_reason"` CreatedAtText string `gorm:"type:varchar(32)" json:"created_at"` UpdatedAtText string `gorm:"type:varchar(32)" json:"updated_at"` diff --git a/heicode/router/agnet_runtime_http_smoke_test.go b/heicode/router/agnet_runtime_http_smoke_test.go new file mode 100644 index 00000000..93b6ad51 --- /dev/null +++ b/heicode/router/agnet_runtime_http_smoke_test.go @@ -0,0 +1,178 @@ +package router + +import ( + "bytes" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-contrib/sessions" + "github.com/gin-contrib/sessions/cookie" + "github.com/gin-gonic/gin" + "github.com/glebarez/sqlite" + "github.com/heicode/manager/common" + "github.com/heicode/manager/model" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +func setupAgnetRuntimeHTTPSmokeDB(t *testing.T) *gorm.DB { + t.Helper() + common.UsingSQLite = true + common.UsingMySQL = false + common.UsingPostgreSQL = false + common.RedisEnabled = false + + dsn := fmt.Sprintf("file:%s?mode=memory&cache=shared", strings.ReplaceAll(t.Name(), "/", "_")) + db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{}) + require.NoError(t, err) + model.DB = db + model.LOG_DB = db + require.NoError(t, db.AutoMigrate( + &model.User{}, + &model.AgnetDeployment{}, + &model.AgnetAuditEvent{}, + )) + adminToken := "runtime-smoke-admin-token" + require.NoError(t, db.Create(&model.User{ + Id: 101, + Username: "runtime-smoke-admin", + Password: "not-used", + Role: common.RoleAdminUser, + Status: common.UserStatusEnabled, + DisplayName: "Runtime Smoke Admin", + AccessToken: &adminToken, + Group: "default", + }).Error) + t.Cleanup(func() { + if model.DB == db { + model.DB = nil + } + if model.LOG_DB == db { + model.LOG_DB = nil + } + sqlDB, err := db.DB() + if err == nil { + _ = sqlDB.Close() + } + }) + return db +} + +func startAgnetRuntimeManagerSmokeServer(t *testing.T) string { + t.Helper() + gin.SetMode(gin.TestMode) + engine := gin.New() + engine.Use(sessions.Sessions("runtime-smoke", cookie.NewStore([]byte("runtime-smoke-secret")))) + SetApiRouter(engine) + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + server := &http.Server{Handler: engine} + go func() { + _ = server.Serve(listener) + }() + t.Cleanup(func() { + _ = server.Close() + }) + return "http://" + listener.Addr().String() +} + +func agnetRuntimeAdminRequest(t *testing.T, method string, url string, body string) *http.Response { + t.Helper() + req, err := http.NewRequest(method, url, bytes.NewBufferString(body)) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer runtime-smoke-admin-token") + req.Header.Set("New-Api-User", "101") + if body != "" { + req.Header.Set("Content-Type", "application/json") + } + resp, err := (&http.Client{Timeout: 5 * time.Second}).Do(req) + require.NoError(t, err) + return resp +} + +func readAgnetRuntimeSmokeBody(t *testing.T, resp *http.Response) string { + t.Helper() + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + return string(body) +} + +func TestAgnetRuntimeRealHTTPHealthAndShadowCreateSmoke(t *testing.T) { + db := setupAgnetRuntimeHTTPSmokeDB(t) + runtimeCreateCalled := false + runtime := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodGet && r.URL.Path == "/api/agnet/health": + require.Equal(t, "Bearer runtime-service-token", r.Header.Get("Authorization")) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"status":"healthy"}}`)) + case r.Method == http.MethodPost && r.URL.Path == "/api/agnet/deployments": + runtimeCreateCalled = true + require.Equal(t, "Bearer runtime-service-token", r.Header.Get("Authorization")) + require.Equal(t, "corr-http-smoke", r.Header.Get("X-Correlation-ID")) + body, err := io.ReadAll(r.Body) + require.NoError(t, err) + require.Contains(t, string(body), `"sub_mode":"agile"`) + require.Contains(t, string(body), `"agile_context"`) + require.Contains(t, string(body), `"checkpoint":"ready_for_test"`) + require.Contains(t, string(body), `"manager_deployment_id"`) + require.Contains(t, string(body), `"signing_secret_ref":"azkv://heicode-kv.vault.azure.net/secrets/callback-signing"`) + require.Contains(t, string(body), `"subscribed_events"`) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"deployment_id":"runtime-http-dep","swarm_id":"runtime-http-swarm","runtime_status":"accepted"}}`)) + default: + http.NotFound(w, r) + } + })) + defer runtime.Close() + + t.Setenv("AGNET_RUNTIME_ENABLED", "true") + t.Setenv("AGNET_RUNTIME_ASYNC", "false") + t.Setenv("AGNET_RUNTIME_BASE_URL", runtime.URL) + t.Setenv("AGNET_RUNTIME_SERVICE_TOKEN", "runtime-service-token") + t.Setenv("AGNET_RUNTIME_CALLBACK_SIGNING_SECRET_REF", "azkv://heicode-kv.vault.azure.net/secrets/callback-signing") + + managerURL := startAgnetRuntimeManagerSmokeServer(t) + + healthResp := agnetRuntimeAdminRequest(t, http.MethodGet, managerURL+"/api/agnet/runtime/health", "") + healthBody := readAgnetRuntimeSmokeBody(t, healthResp) + require.Equal(t, http.StatusOK, healthResp.StatusCode) + require.Contains(t, healthBody, `"success":true`) + require.Contains(t, healthBody, `"status":"healthy"`) + + createBody := `{ + "orchestration_plan":{ + "intent_id":"intent-http-smoke", + "template_hint":"heicode-task", + "objective":"real HTTP runtime smoke", + "sub_mode":"agile", + "risk_level":"low", + "budget":{"max_tokens":10000,"max_cost_usd":1,"max_duration_sec":600}, + "user_context":{"user_id":"101","channel_id":"default"}, + "billing_context":{"provider":"newapi","newapi_user_ref":"newapi-http-smoke"}, + "agile_context":{"iteration":"2026-05-27~2026-05-28","stage":"development","checkpoint":"ready_for_test","acceptance_criteria":["接口返回成功"],"next_action":"submit_test_result","requires_user_approval":false}, + "agent_runtime":{"platform":"agnet","agents":[{"role":"builder","model_ref":"model-http-smoke","instance_count":1}]}, + "agents":[{"role_template":"builder","goal":"smoke","default_model_id":"model-http-smoke","resource_grants":[{"grant_id":"grant-http-doc","resource_id":"doc-http","resource_type":"project_doc","user_id":"101","binding_scope":"task-http-smoke","target_role":"builder","target_agent_ref":"agent-builder-1","permission_scope":["doc:read"],"status":"active"}]}], + "constraints":{"allowed_model_ids":["model-http-smoke"]}, + "metadata":{"correlation_id":"corr-http-smoke"} + } + }` + createResp := agnetRuntimeAdminRequest(t, http.MethodPost, managerURL+"/api/agnet/deployments", createBody) + responseBody := readAgnetRuntimeSmokeBody(t, createResp) + require.Equal(t, http.StatusOK, createResp.StatusCode) + require.Contains(t, responseBody, `"success":true`) + require.Contains(t, responseBody, `"runtime_deployment_id":"runtime-http-dep"`) + require.Contains(t, responseBody, `"runtime_swarm_id":"runtime-http-swarm"`) + require.True(t, runtimeCreateCalled) + + var stored model.AgnetDeployment + require.NoError(t, db.Where("runtime_swarm_id = ?", "runtime-http-swarm").First(&stored).Error) + require.Equal(t, "accepted", stored.RuntimeState) +} diff --git a/heicode/router/api-router.go b/heicode/router/api-router.go index e624a67c..3b5e44f6 100644 --- a/heicode/router/api-router.go +++ b/heicode/router/api-router.go @@ -529,6 +529,7 @@ func SetApiRouter(router *gin.Engine) { agnetRoute.GET("/deployments/:deployment_id/events", controller.AgnetListDeploymentEvents) agnetRoute.GET("/deployments/:deployment_id/sk-snapshots", controller.AgnetListSKSnapshots) agnetRoute.POST("/sk-snapshots/resolve", controller.AgnetResolveSKSnapshots) + agnetRoute.GET("/runtime/health", controller.AgnetRuntimeHealth) agnetRoute.GET("/projects/:project_id/dashboard-snapshot", controller.AgnetProjectDashboardSnapshot) agnetRoute.GET("/audit-logs", controller.AgnetListAuditLogs) } diff --git a/heicode/router/api_router_test.go b/heicode/router/api_router_test.go new file mode 100644 index 00000000..88e37264 --- /dev/null +++ b/heicode/router/api_router_test.go @@ -0,0 +1,15 @@ +package router + +import ( + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestSetApiRouterDoesNotRegisterDuplicateRoutes(t *testing.T) { + gin.SetMode(gin.TestMode) + require.NotPanics(t, func() { + SetApiRouter(gin.New()) + }) +}