724 lines
25 KiB
Go
724 lines
25 KiB
Go
package controller
|
|
|
|
import (
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/heicode/manager/common"
|
|
"github.com/heicode/manager/model"
|
|
)
|
|
|
|
type agnetCallbackEnvelope struct {
|
|
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 {
|
|
ArtifactID string `json:"artifact_id"`
|
|
ArtifactType string `json:"artifact_type"`
|
|
Title string `json:"title"`
|
|
Summary string `json:"summary"`
|
|
URI string `json:"uri"`
|
|
Checksum string `json:"checksum"`
|
|
Metadata map[string]any `json:"metadata"`
|
|
}
|
|
|
|
func agnetCallbackTokenFromRequest(c *gin.Context) string {
|
|
if token := strings.TrimSpace(c.GetHeader("X-Agnet-Service-Token")); token != "" {
|
|
return token
|
|
}
|
|
auth := strings.TrimSpace(c.GetHeader("Authorization"))
|
|
if strings.HasPrefix(strings.ToLower(auth), "bearer ") {
|
|
return strings.TrimSpace(auth[len("Bearer "):])
|
|
}
|
|
return ""
|
|
}
|
|
|
|
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")
|
|
return false
|
|
}
|
|
if agnetCallbackTokenFromRequest(c) != expected {
|
|
agnetError(c, "CALLBACK_UNAUTHORIZED", "invalid callback service token")
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func agnetCallbackHasPlaintextSecret(payload agnetCallbackEnvelope) bool {
|
|
if containsPlaintextSecret(payload.Metadata) || containsPlaintextSecret(payload.Payload) || containsPlaintextSecret(payload.Artifact.Metadata) {
|
|
return true
|
|
}
|
|
raw, _ := common.Marshal(payload)
|
|
var asMap map[string]any
|
|
if err := common.Unmarshal(raw, &asMap); err != nil {
|
|
return false
|
|
}
|
|
return containsPlaintextSecret(asMap)
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
runtimeID := firstNonEmpty(swarmID, deploymentID)
|
|
if runtimeID == "" {
|
|
return agnetDeploymentRecord{}, false
|
|
}
|
|
|
|
agnetMu.RLock()
|
|
for _, record := range agnetDeployments {
|
|
if strings.TrimSpace(record.RuntimeSwarmID) == runtimeID || strings.TrimSpace(record.RuntimeDeploymentID) == runtimeID {
|
|
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 = ? OR runtime_deployment_id = ?", runtimeID, runtimeID).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 {
|
|
artifact := payload.Artifact
|
|
if strings.TrimSpace(artifact.ArtifactID) == "" {
|
|
return nil
|
|
}
|
|
metadataJSON := ""
|
|
if artifact.Metadata != nil {
|
|
if data, err := common.Marshal(artifact.Metadata); err == nil {
|
|
metadataJSON = string(data)
|
|
}
|
|
}
|
|
return model.UpsertAgnetArtifact(&model.AgnetArtifact{
|
|
ArtifactID: strings.TrimSpace(artifact.ArtifactID),
|
|
DeploymentID: strings.TrimSpace(payload.DeploymentID),
|
|
TaskID: strings.TrimSpace(payload.TaskID),
|
|
UserID: record.Plan.UserContext.UserID,
|
|
BindingScope: firstPlanBindingScope(record.Plan),
|
|
CorrelationID: strings.TrimSpace(payload.CorrelationID),
|
|
ArtifactType: strings.TrimSpace(artifact.ArtifactType),
|
|
Title: strings.TrimSpace(artifact.Title),
|
|
Summary: strings.TrimSpace(artifact.Summary),
|
|
URI: strings.TrimSpace(artifact.URI),
|
|
Checksum: strings.TrimSpace(artifact.Checksum),
|
|
MetadataJSON: metadataJSON,
|
|
CreatedAtMs: time.Now().UnixMilli(),
|
|
})
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
|
|
var agnetCallbackEventRequiredFields = map[string][]string{
|
|
"deployment.status_changed": {"status"},
|
|
"phase.changed": {"stage", "checkpoint"},
|
|
"agent.started": {"agent_role"},
|
|
"agent.completed": {"agent_role"},
|
|
"agent.crashed": {"agent_role", "reason"},
|
|
"task.created": {"task_id", "title"},
|
|
"task.claimed": {"task_id", "agent_role"},
|
|
"task.running": {"task_id", "agent_role"},
|
|
"task.heartbeat": {"task_id", "agent_role"},
|
|
"task.blocked": {"task_id", "reason"},
|
|
"task.retried": {"task_id", "attempt"},
|
|
"task.released": {"task_id", "agent_role"},
|
|
"task.failed": {"task_id", "reason"},
|
|
"task.completed": {"task_id"},
|
|
"handoff.requested": {"task_id", "from_role", "to_role"},
|
|
"handoff.completed": {"task_id", "from_role", "to_role"},
|
|
"approval.requested": {"approval_id", "operation", "risk_level"},
|
|
"artifact.created": {"artifact_id"},
|
|
"timeline.updated": {"title"},
|
|
"sk_tool.called": {"tool_name", "tool_invocation_id"},
|
|
"sk_tool.completed": {"tool_name", "tool_invocation_id"},
|
|
"sk_tool.failed": {"tool_name", "tool_invocation_id", "reason"},
|
|
"budget.alert": {"threshold_pct"},
|
|
}
|
|
|
|
var agnetCallbackEventCategories = map[string]string{
|
|
"deployment.status_changed": "deployment",
|
|
"phase.changed": "ordinary_sub",
|
|
"agent.started": "ordinary_sub",
|
|
"agent.completed": "ordinary_sub",
|
|
"agent.crashed": "ordinary_sub",
|
|
"task.created": "swarm_task_flow",
|
|
"task.claimed": "swarm_task_flow",
|
|
"task.running": "swarm_task_flow",
|
|
"task.heartbeat": "swarm_task_flow",
|
|
"task.blocked": "swarm_task_flow",
|
|
"task.retried": "swarm_task_flow",
|
|
"task.released": "swarm_task_flow",
|
|
"task.failed": "swarm_task_flow",
|
|
"task.completed": "swarm_task_flow",
|
|
"handoff.requested": "swarm_task_flow",
|
|
"handoff.completed": "swarm_task_flow",
|
|
"approval.requested": "approval",
|
|
"artifact.created": "artifact",
|
|
"timeline.updated": "timeline",
|
|
"sk_tool.called": "sk",
|
|
"sk_tool.completed": "sk",
|
|
"sk_tool.failed": "sk",
|
|
"budget.alert": "budget",
|
|
}
|
|
|
|
func AgnetGetSwarmEventCallbackSchema(c *gin.Context) {
|
|
events := make([]string, 0, len(agnetCallbackEventRequiredFields))
|
|
for eventType := range agnetCallbackEventRequiredFields {
|
|
events = append(events, eventType)
|
|
}
|
|
sort.Strings(events)
|
|
|
|
items := make([]gin.H, 0, len(events))
|
|
for _, eventType := range events {
|
|
required := append([]string(nil), agnetCallbackEventRequiredFields[eventType]...)
|
|
sort.Strings(required)
|
|
items = append(items, gin.H{
|
|
"event_type": eventType,
|
|
"category": agnetCallbackEventCategories[eventType],
|
|
"required_fields": required,
|
|
"payload_location": "top-level envelope or payload object; artifact_id may also be in artifact object",
|
|
})
|
|
}
|
|
|
|
common.ApiSuccess(c, gin.H{
|
|
"callback_path": "/api/agnet/callbacks/swarm-events",
|
|
"auth": gin.H{
|
|
"service_token_headers": []string{"X-Agnet-Service-Token", "Authorization: Bearer <token>"},
|
|
"hmac_headers": []string{"X-Agnet-Event-Id", "X-Agnet-Timestamp", "X-Agnet-Signature"},
|
|
"hmac_payload": "timestamp + \".\" + event_id + \".\" + raw_body",
|
|
},
|
|
"dedupe_keys": []string{"X-Agnet-Event-Id", "event_id", "idempotency_key"},
|
|
"events": items,
|
|
"security": gin.H{
|
|
"plaintext_secrets_allowed": false,
|
|
"secret_ref_scheme": "azkv://<vault>/secrets/<name>",
|
|
},
|
|
})
|
|
}
|
|
|
|
func callbackEnvelopeFieldValue(payload agnetCallbackEnvelope, key string) string {
|
|
switch key {
|
|
case "task_id":
|
|
return strings.TrimSpace(payload.TaskID)
|
|
case "artifact_id":
|
|
return strings.TrimSpace(payload.Artifact.ArtifactID)
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
func callbackHasFieldValue(payload agnetCallbackEnvelope, key string) bool {
|
|
if callbackEnvelopeFieldValue(payload, key) != "" {
|
|
return true
|
|
}
|
|
if callbackStringValue(payload.Payload, key) != "" {
|
|
return true
|
|
}
|
|
for _, nestedKey := range []string{"artifact", "approval"} {
|
|
if nested := callbackMapValue(payload.Payload, nestedKey); nested != nil {
|
|
if callbackStringValue(nested, key) != "" {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
if key == "attempt" {
|
|
return callbackIntValue(payload.Payload, key) > 0
|
|
}
|
|
return false
|
|
}
|
|
|
|
func validateAgnetCallbackEventSchema(payload agnetCallbackEnvelope) error {
|
|
required, ok := agnetCallbackEventRequiredFields[payload.EventType]
|
|
if !ok {
|
|
return nil
|
|
}
|
|
for _, key := range required {
|
|
if !callbackHasFieldValue(payload, key) {
|
|
return fmt.Errorf("%s is required for %s callback", key, payload.EventType)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func persistAgnetApprovalFromCallback(payload agnetCallbackEnvelope, record agnetDeploymentRecord) error {
|
|
if payload.EventType != "approval.requested" || model.DB == nil {
|
|
return nil
|
|
}
|
|
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) {
|
|
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 := 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
|
|
}
|
|
if err := validateAgnetCallbackEventSchema(payload); err != nil {
|
|
agnetError(c, "CALLBACK_SCHEMA_INVALID", err.Error())
|
|
return
|
|
}
|
|
if agnetCallbackHasPlaintextSecret(payload) {
|
|
agnetError(c, "CALLBACK_SECRET_REJECTED", "callbacks must not contain plaintext credential fields")
|
|
return
|
|
}
|
|
|
|
incomingDeploymentID := strings.TrimSpace(payload.DeploymentID)
|
|
incomingSwarmID := strings.TrimSpace(payload.SwarmID)
|
|
record, _ := agnetCallbackDeploymentContext(incomingDeploymentID, incomingSwarmID)
|
|
if strings.TrimSpace(record.DeploymentID) != "" {
|
|
payload.DeploymentID = record.DeploymentID
|
|
if strings.TrimSpace(payload.SwarmID) == "" {
|
|
payload.SwarmID = firstNonEmpty(record.RuntimeSwarmID, record.RuntimeDeploymentID, incomingDeploymentID)
|
|
}
|
|
}
|
|
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),
|
|
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())
|
|
agnetError(c, "CALLBACK_PERSIST_FAILED", "failed to persist callback")
|
|
return
|
|
}
|
|
if inserted {
|
|
if err := persistAgnetArtifactFromCallback(payload, record); err != nil {
|
|
common.SysLog("persistAgnetArtifactFromCallback: " + err.Error())
|
|
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,
|
|
SchemaVersion: 1,
|
|
UserID: record.Plan.UserContext.UserID,
|
|
ChannelID: record.Plan.UserContext.ChannelID,
|
|
BindingScope: firstPlanBindingScope(record.Plan),
|
|
DeploymentID: strings.TrimSpace(payload.DeploymentID),
|
|
CorrelationID: strings.TrimSpace(payload.CorrelationID),
|
|
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,
|
|
"deduplicated": !inserted,
|
|
"deployment_id": payload.DeploymentID,
|
|
})
|
|
}
|
|
|
|
func AgnetListUserDeploymentArtifacts(c *gin.Context) {
|
|
record, ok := requireAuthenticatedUserAgnetDeployment(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
items, err := model.ListAgnetArtifacts(model.ListAgnetArtifactsFilter{
|
|
DeploymentID: record.DeploymentID,
|
|
Limit: 500,
|
|
})
|
|
if err != nil {
|
|
common.SysLog("AgnetListUserDeploymentArtifacts: " + err.Error())
|
|
agnetError(c, "ARTIFACT_QUERY_FAILED", "failed to query artifacts")
|
|
return
|
|
}
|
|
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,
|
|
"source": callback.Source,
|
|
"payload": payload,
|
|
}
|
|
for _, key := range []string{"title", "summary", "stage", "checkpoint", "severity", "next_action", "agent_role", "task_id", "from_role", "to_role", "reason", "attempt"} {
|
|
if value, ok := payload[key]; ok {
|
|
entry[key] = value
|
|
}
|
|
}
|
|
return entry
|
|
}
|
|
|
|
func AgnetGetUserDeploymentTimeline(c *gin.Context) {
|
|
record, ok := requireAuthenticatedUserAgnetDeployment(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
events, err := model.ListAgnetAuditEventsByDeployment(record.DeploymentID)
|
|
if err != nil {
|
|
agnetError(c, "TIMELINE_QUERY_FAILED", "failed to query audit events")
|
|
return
|
|
}
|
|
callbacks, err := model.ListAgnetCallbackEvents(model.ListAgnetCallbackEventsFilter{DeploymentID: record.DeploymentID, Limit: 500})
|
|
if err != nil {
|
|
agnetError(c, "TIMELINE_QUERY_FAILED", "failed to query callbacks")
|
|
return
|
|
}
|
|
artifacts, err := model.ListAgnetArtifacts(model.ListAgnetArtifactsFilter{DeploymentID: record.DeploymentID, Limit: 500})
|
|
if err != nil {
|
|
agnetError(c, "TIMELINE_QUERY_FAILED", "failed to query artifacts")
|
|
return
|
|
}
|
|
snapshots, err := model.ListAgnetSKSnapshots(record.DeploymentID)
|
|
if err != nil {
|
|
agnetError(c, "TIMELINE_QUERY_FAILED", "failed to query sk snapshots")
|
|
return
|
|
}
|
|
timeline := make([]gin.H, 0, len(events)+len(callbacks)+len(artifacts)+len(snapshots))
|
|
for _, event := range events {
|
|
timeline = append(timeline, gin.H{"kind": "audit", "at": event.OccurredAt, "event": event.Event})
|
|
}
|
|
for _, callback := range callbacks {
|
|
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})
|
|
}
|
|
for _, snapshot := range snapshots {
|
|
timeline = append(timeline, gin.H{"kind": "sk_snapshot", "at": snapshot.ResolvedAtMs, "event": snapshot.SourceType, "snapshot_id": snapshot.SnapshotID})
|
|
}
|
|
common.ApiSuccess(c, gin.H{
|
|
"deployment_id": record.DeploymentID,
|
|
"deployment": record,
|
|
"events": events,
|
|
"callbacks": callbacks,
|
|
"artifacts": artifacts,
|
|
"sk_snapshots": snapshots,
|
|
"timeline": timeline,
|
|
})
|
|
}
|