chenchenandClaude Opus 4.8
56a9b02a25
按 agent_swarm#14/#15 冻结契约做 HM 读侧适配(客户端 #28 消费): - 注册 6 类新事件(event-schema v1):swarm.completed/failed/stopped、approval.approved/rejected、 handoff.created(categories + requiredFields 两表;必填先最小集,待 agent_swarm PR #28 §4 精校)。 - AgentCallbackEvent 增 Sequence(per-swarm 严格递增序号),回调入库透传 envelope.sequence; 事件视图暴露 sequence 供客户端去重/排序。游标仍用稳定 id(next_after)避免 sequence 未全量 上线时回归。 - 脱敏键补 credential_ref/signing_secret_ref(envelope 按设计透传 azkv:// 引用,客户端视图剔除)。 - artifact.created → 扁平视图 {uri,checksum,task_id,size_bytes?,created_at}(无 secret_ref; size 未知省略不伪造)。 - 状态展示映射 §4.1:display_status(blocked→degraded;不臆造 preparing/verifying)。 测试:状态映射、artifact 视图(脱敏 + size 省略)、6 类事件注册;controller+model 全回归通过, go build/vet 干净。文档 §5.2 更新。 未含(下一 PR):stop 真实运行时接入(写路径,复用 agentRuntimeClientConfigForMode("swarm"))。 Affects: Manager only(只读查询契约适配)。AgentCallbackEvent 加列(AutoMigrate);无计费改动。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
921 lines
32 KiB
Go
921 lines
32 KiB
Go
package controller
|
|
|
|
import (
|
|
"context"
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/heicode/manager/common"
|
|
"github.com/heicode/manager/model"
|
|
)
|
|
|
|
type agentCallbackEnvelope 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"`
|
|
Sequence int `json:"sequence"`
|
|
Source string `json:"source"`
|
|
Metadata map[string]any `json:"metadata"`
|
|
Payload map[string]any `json:"payload"`
|
|
Artifact agentArtifactPayload `json:"artifact"`
|
|
}
|
|
|
|
type agentArtifactPayload 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 agentCallbackTokenFromRequest(c *gin.Context) string {
|
|
if token := strings.TrimSpace(c.GetHeader("X-Agent-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 agentCallbackSigningSecret() string {
|
|
if secret := strings.TrimSpace(common.GetEnvOrDefaultString("AGENT_CALLBACK_SIGNING_SECRET", "")); secret != "" {
|
|
return secret
|
|
}
|
|
secretRef := firstNonEmpty(
|
|
common.GetEnvOrDefaultString("AGENT_CALLBACK_SIGNING_SECRET_REF", ""),
|
|
common.GetEnvOrDefaultString("AGENT_RUNTIME_CALLBACK_SIGNING_SECRET_REF", ""),
|
|
)
|
|
if secretRef == "" {
|
|
return ""
|
|
}
|
|
client, err := newSecretStoreClientFromEnv()
|
|
if err != nil {
|
|
common.SysLog("agentCallbackSigningSecret: " + err.Error())
|
|
return ""
|
|
}
|
|
data, err := client.getJSONSecret(secretRef)
|
|
if err != nil {
|
|
common.SysLog("agentCallbackSigningSecret: " + err.Error())
|
|
return ""
|
|
}
|
|
for _, key := range []string{"callback_signing_secret", "signing_secret", "secret", "value"} {
|
|
if value := callbackStringValue(data, key); value != "" {
|
|
return value
|
|
}
|
|
}
|
|
common.SysLog("agentCallbackSigningSecret: signing secret is missing from Azure Key Vault payload")
|
|
return ""
|
|
}
|
|
|
|
func agentCallbackSignatureTolerance() time.Duration {
|
|
seconds := common.GetEnvOrDefault("AGENT_CALLBACK_SIGNATURE_TOLERANCE_SECONDS", 300)
|
|
if seconds <= 0 {
|
|
seconds = 300
|
|
}
|
|
return time.Duration(seconds) * time.Second
|
|
}
|
|
|
|
func validateAgentCallbackHMAC(c *gin.Context, rawBody []byte, eventID string) (bool, bool) {
|
|
secret := agentCallbackSigningSecret()
|
|
if secret == "" {
|
|
return false, false
|
|
}
|
|
timestamp := strings.TrimSpace(c.GetHeader("X-Agent-Timestamp"))
|
|
signature := strings.TrimSpace(c.GetHeader("X-Agent-Signature"))
|
|
if timestamp == "" || signature == "" || eventID == "" {
|
|
return false, false
|
|
}
|
|
tsMs, err := strconv.ParseInt(timestamp, 10, 64)
|
|
if err != nil {
|
|
agentError(c, "CALLBACK_UNAUTHORIZED", "invalid callback timestamp")
|
|
return true, false
|
|
}
|
|
now := time.Now()
|
|
eventTime := time.UnixMilli(tsMs)
|
|
tolerance := agentCallbackSignatureTolerance()
|
|
if eventTime.Before(now.Add(-tolerance)) || eventTime.After(now.Add(tolerance)) {
|
|
agentError(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)) {
|
|
agentError(c, "CALLBACK_UNAUTHORIZED", "invalid callback signature")
|
|
return true, false
|
|
}
|
|
return true, true
|
|
}
|
|
|
|
func validateAgentCallbackAuth(c *gin.Context, rawBody []byte, eventID string) bool {
|
|
if common.GetEnvOrDefaultBool("AGENT_CALLBACK_AUTH_DISABLED", false) {
|
|
return true
|
|
}
|
|
if attempted, ok := validateAgentCallbackHMAC(c, rawBody, eventID); attempted {
|
|
return ok
|
|
}
|
|
expected := strings.TrimSpace(os.Getenv("AGENT_CALLBACK_TOKEN"))
|
|
if expected == "" {
|
|
agentError(c, "CALLBACK_UNAUTHORIZED", "callback token is not configured")
|
|
return false
|
|
}
|
|
if agentCallbackTokenFromRequest(c) != expected {
|
|
agentError(c, "CALLBACK_UNAUTHORIZED", "invalid callback service token")
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func agentCallbackHasPlaintextSecret(payload agentCallbackEnvelope) 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 agentCallbackDeploymentContext(deploymentID string, swarmID string) (agentDeploymentRecord, bool) {
|
|
deploymentID = strings.TrimSpace(deploymentID)
|
|
swarmID = strings.TrimSpace(swarmID)
|
|
if deploymentID != "" {
|
|
if record, ok := findAgentDeploymentRecord(deploymentID); ok {
|
|
return record, true
|
|
}
|
|
}
|
|
runtimeID := firstNonEmpty(swarmID, deploymentID)
|
|
if runtimeID == "" {
|
|
return agentDeploymentRecord{}, false
|
|
}
|
|
|
|
agentMu.RLock()
|
|
for _, record := range agentDeployments {
|
|
if strings.TrimSpace(record.RuntimeSwarmID) == runtimeID || strings.TrimSpace(record.RuntimeDeploymentID) == runtimeID {
|
|
agentMu.RUnlock()
|
|
return record, true
|
|
}
|
|
}
|
|
agentMu.RUnlock()
|
|
|
|
if model.DB == nil {
|
|
return agentDeploymentRecord{}, false
|
|
}
|
|
var row model.AgentDeployment
|
|
if err := model.DB.Where("runtime_swarm_id = ? OR runtime_deployment_id = ?", runtimeID, runtimeID).First(&row).Error; err != nil {
|
|
return agentDeploymentRecord{}, false
|
|
}
|
|
record, err := agentDeploymentModelToRecord(row)
|
|
if err != nil {
|
|
common.SysLog("agentCallbackDeploymentContext: " + err.Error())
|
|
return agentDeploymentRecord{}, false
|
|
}
|
|
agentMu.Lock()
|
|
agentDeployments[record.DeploymentID] = record
|
|
agentMu.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 agentCallbackEventRuntimeState(eventType string, payload map[string]any) string {
|
|
if state := callbackStringValue(payload, "status"); state != "" {
|
|
return state
|
|
}
|
|
switch eventType {
|
|
case "agent.started":
|
|
return "running"
|
|
case "agent.completed":
|
|
return "completed"
|
|
case "agent.crashed", "sk_tool.failed", "task.failed":
|
|
return "failed"
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
func upsertAgentCallbackAgentInstance(record *agentDeploymentRecord, payload agentCallbackEnvelope, phase string, runtimeState string) bool {
|
|
if record == nil {
|
|
return false
|
|
}
|
|
instanceID := strings.TrimSpace(payload.AgentInstanceID)
|
|
source := payload.Payload
|
|
role := firstNonEmpty(callbackStringValue(source, "agent_role"), callbackStringValue(source, "role"))
|
|
if role == "" && instanceID == "" {
|
|
return false
|
|
}
|
|
failureReason := firstNonEmpty(callbackStringValue(source, "reason"), callbackStringValue(source, "error"))
|
|
for idx := range record.AgentInstances {
|
|
instance := &record.AgentInstances[idx]
|
|
if instanceID != "" && strings.TrimSpace(instance.InstanceID) == instanceID {
|
|
if role != "" {
|
|
instance.Role = role
|
|
}
|
|
if phase != "" {
|
|
instance.Phase = phase
|
|
}
|
|
if runtimeState != "" {
|
|
instance.RuntimeState = runtimeState
|
|
}
|
|
if failureReason != "" {
|
|
instance.FailureReason = failureReason
|
|
}
|
|
return true
|
|
}
|
|
if instanceID != "" && role != "" && strings.TrimSpace(instance.Role) == role && (strings.TrimSpace(instance.RuntimeState) == "queued" || strings.TrimSpace(instance.RuntimeState) == "pending") {
|
|
instance.InstanceID = instanceID
|
|
instance.Phase = firstNonEmpty(phase, instance.Phase)
|
|
instance.RuntimeState = firstNonEmpty(runtimeState, instance.RuntimeState)
|
|
instance.FailureReason = failureReason
|
|
return true
|
|
}
|
|
}
|
|
record.AgentInstances = append(record.AgentInstances, agentAgentInstance{
|
|
InstanceID: firstNonEmpty(instanceID, "agi_"+common.GetUUID()[:12]),
|
|
Role: role,
|
|
Phase: firstNonEmpty(phase, record.Phase),
|
|
RuntimeState: firstNonEmpty(runtimeState, record.RuntimeState),
|
|
FailureReason: failureReason,
|
|
})
|
|
return true
|
|
}
|
|
|
|
func applyAgentCallbackDeploymentState(payload agentCallbackEnvelope, record agentDeploymentRecord) (agentDeploymentRecord, bool) {
|
|
if strings.TrimSpace(record.DeploymentID) == "" {
|
|
return record, false
|
|
}
|
|
source := payload.Payload
|
|
changed := false
|
|
switch payload.EventType {
|
|
case "deployment.status_changed":
|
|
status := callbackStringValue(source, "status")
|
|
if status != "" {
|
|
record.Status = status
|
|
record.RuntimeState = status
|
|
if containsString([]string{"completed", "failed", "stopped"}, status) {
|
|
for idx := range record.AgentInstances {
|
|
if record.AgentInstances[idx].RuntimeState == "" || !containsString([]string{"failed", "stopped"}, record.AgentInstances[idx].RuntimeState) {
|
|
record.AgentInstances[idx].RuntimeState = status
|
|
}
|
|
if record.Phase != "" {
|
|
record.AgentInstances[idx].Phase = record.Phase
|
|
}
|
|
}
|
|
}
|
|
changed = true
|
|
}
|
|
case "phase.changed":
|
|
phase := firstNonEmpty(callbackStringValue(source, "stage"), callbackStringValue(source, "phase"))
|
|
checkpoint := callbackStringValue(source, "checkpoint")
|
|
if phase != "" {
|
|
record.Phase = phase
|
|
changed = true
|
|
}
|
|
if checkpoint != "" {
|
|
record.RuntimeState = checkpoint
|
|
changed = true
|
|
}
|
|
case "timeline.updated":
|
|
phase := firstNonEmpty(callbackStringValue(source, "stage"), callbackStringValue(source, "phase"))
|
|
checkpoint := callbackStringValue(source, "checkpoint")
|
|
if phase != "" {
|
|
record.Phase = phase
|
|
changed = true
|
|
}
|
|
if checkpoint != "" && !containsString([]string{"completed", "failed", "stopped"}, record.RuntimeState) {
|
|
record.RuntimeState = checkpoint
|
|
changed = true
|
|
}
|
|
case "agent.started", "agent.completed", "agent.crashed":
|
|
phase := firstNonEmpty(callbackStringValue(source, "stage"), callbackStringValue(source, "phase"), record.Phase)
|
|
runtimeState := agentCallbackEventRuntimeState(payload.EventType, source)
|
|
if upsertAgentCallbackAgentInstance(&record, payload, phase, runtimeState) {
|
|
changed = true
|
|
}
|
|
}
|
|
if !changed {
|
|
return record, false
|
|
}
|
|
record.UpdatedAt = firstNonEmpty(strings.TrimSpace(payload.OccurredAt), agentNow())
|
|
agentMu.Lock()
|
|
agentDeployments[record.DeploymentID] = record
|
|
agentMu.Unlock()
|
|
if err := updateAgentDeploymentRecord(record); err != nil {
|
|
common.SysLog("applyAgentCallbackDeploymentState: " + err.Error())
|
|
}
|
|
return record, true
|
|
}
|
|
|
|
func normalizeCallbackArtifact(payload *agentCallbackEnvelope) {
|
|
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 = agentArtifactPayload{
|
|
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 persistAgentArtifactFromCallback(payload agentCallbackEnvelope, record agentDeploymentRecord) 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.UpsertAgentArtifact(&model.AgentArtifact{
|
|
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 agentCallbackEventRequiredFields = 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"},
|
|
// agent_swarm event-schema FROZEN v1 新增 6 类(#15)。必填字段先按最小集登记
|
|
// (避免误拒);agent_swarm PR #28 合并后按其 §4 表精校。
|
|
"handoff.created": {"task_id"},
|
|
"approval.approved": {"approval_id"},
|
|
"approval.rejected": {"approval_id"},
|
|
"swarm.completed": {},
|
|
"swarm.failed": {"reason"},
|
|
"swarm.stopped": {},
|
|
"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 agentCallbackEventCategories = 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",
|
|
"handoff.created": "swarm_task_flow",
|
|
"approval.requested": "approval",
|
|
"approval.approved": "approval",
|
|
"approval.rejected": "approval",
|
|
"swarm.completed": "swarm_lifecycle",
|
|
"swarm.failed": "swarm_lifecycle",
|
|
"swarm.stopped": "swarm_lifecycle",
|
|
"artifact.created": "artifact",
|
|
"timeline.updated": "timeline",
|
|
"sk_tool.called": "sk",
|
|
"sk_tool.completed": "sk",
|
|
"sk_tool.failed": "sk",
|
|
"budget.alert": "budget",
|
|
}
|
|
|
|
func AgentGetRuntimeEventCallbackSchema(c *gin.Context) {
|
|
events := make([]string, 0, len(agentCallbackEventRequiredFields))
|
|
for eventType := range agentCallbackEventRequiredFields {
|
|
events = append(events, eventType)
|
|
}
|
|
sort.Strings(events)
|
|
|
|
items := make([]gin.H, 0, len(events))
|
|
for _, eventType := range events {
|
|
required := append([]string(nil), agentCallbackEventRequiredFields[eventType]...)
|
|
sort.Strings(required)
|
|
items = append(items, gin.H{
|
|
"event_type": eventType,
|
|
"category": agentCallbackEventCategories[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/agent/callbacks/runtime-events",
|
|
"auth": gin.H{
|
|
"service_token_headers": []string{"X-Agent-Service-Token", "Authorization: Bearer <token>"},
|
|
"hmac_headers": []string{"X-Agent-Event-Id", "X-Agent-Timestamp", "X-Agent-Signature"},
|
|
"hmac_payload": "timestamp + \".\" + event_id + \".\" + raw_body",
|
|
},
|
|
"dedupe_keys": []string{"X-Agent-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 agentCallbackEnvelope, 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 agentCallbackEnvelope, 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 validateAgentCallbackEventSchema(payload agentCallbackEnvelope) error {
|
|
required, ok := agentCallbackEventRequiredFields[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 persistAgentApprovalFromCallback(payload agentCallbackEnvelope, record agentDeploymentRecord) 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.AgentApprovalRequest{
|
|
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: agentApprovalStatusPending,
|
|
RequestedBy: firstNonEmpty(callbackStringValue(source, "requested_by"), "agent-runtime"),
|
|
RequestReason: firstNonEmpty(callbackStringValue(source, "reason"), callbackStringValue(source, "summary"), "Runtime requested approval"),
|
|
TTLSeconds: callbackIntValue(source, "ttl_seconds"),
|
|
}
|
|
if approval.TTLSeconds <= 0 {
|
|
approval.TTLSeconds = defaultAgentApprovalTTLSeconds
|
|
}
|
|
if approval.TTLSeconds > maxAgentApprovalTTLSeconds {
|
|
approval.TTLSeconds = maxAgentApprovalTTLSeconds
|
|
}
|
|
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.AgentApprovalRequest
|
|
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
|
|
}
|
|
recordAgentApprovalAudit("approval.requested", &approval, nil, "ok", "")
|
|
return nil
|
|
}
|
|
|
|
func AgentReceiveRuntimeEventCallback(c *gin.Context) {
|
|
rawBody, err := io.ReadAll(io.LimitReader(c.Request.Body, 1<<20))
|
|
if err != nil {
|
|
agentError(c, "CALLBACK_INVALID", "failed to read callback body")
|
|
return
|
|
}
|
|
var payload agentCallbackEnvelope
|
|
if err := common.Unmarshal(rawBody, &payload); err != nil {
|
|
agentError(c, "CALLBACK_INVALID", err.Error())
|
|
return
|
|
}
|
|
if payload.EventID == "" {
|
|
payload.EventID = strings.TrimSpace(c.GetHeader("X-Agent-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 !validateAgentCallbackAuth(c, rawBody, payload.EventID) {
|
|
return
|
|
}
|
|
if payload.EventID == "" || payload.EventType == "" {
|
|
agentError(c, "CALLBACK_INVALID", "event_id and event_type are required")
|
|
return
|
|
}
|
|
normalizeCallbackArtifact(&payload)
|
|
if payload.IdempotencyKey == "" {
|
|
payload.IdempotencyKey = payload.EventID
|
|
}
|
|
if err := validateAgentCallbackEventSchema(payload); err != nil {
|
|
agentError(c, "CALLBACK_SCHEMA_INVALID", err.Error())
|
|
return
|
|
}
|
|
if agentCallbackHasPlaintextSecret(payload) {
|
|
agentError(c, "CALLBACK_SECRET_REJECTED", "callbacks must not contain plaintext credential fields")
|
|
return
|
|
}
|
|
|
|
incomingDeploymentID := strings.TrimSpace(payload.DeploymentID)
|
|
incomingSwarmID := strings.TrimSpace(payload.SwarmID)
|
|
record, _ := agentCallbackDeploymentContext(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.InsertAgentCallbackEvent(&model.AgentCallbackEvent{
|
|
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),
|
|
Sequence: payload.Sequence,
|
|
Source: strings.TrimSpace(payload.Source),
|
|
Result: "ok",
|
|
PayloadJSON: string(payloadJSON),
|
|
OccurredAt: strings.TrimSpace(payload.OccurredAt),
|
|
CreatedAtMs: time.Now().UnixMilli(),
|
|
})
|
|
if err != nil {
|
|
common.SysLog("AgentReceiveRuntimeEventCallback: " + err.Error())
|
|
agentError(c, "CALLBACK_PERSIST_FAILED", "failed to persist callback")
|
|
return
|
|
}
|
|
if inserted {
|
|
if err := persistAgentArtifactFromCallback(payload, record); err != nil {
|
|
common.SysLog("persistAgentArtifactFromCallback: " + err.Error())
|
|
agentError(c, "CALLBACK_PERSIST_FAILED", "failed to persist artifact")
|
|
return
|
|
}
|
|
if err := persistAgentApprovalFromCallback(payload, record); err != nil {
|
|
common.SysLog("persistAgentApprovalFromCallback: " + err.Error())
|
|
agentError(c, "CALLBACK_PERSIST_FAILED", "failed to persist approval request")
|
|
return
|
|
}
|
|
record, _ = applyAgentCallbackDeploymentState(payload, record)
|
|
recordAgentAuditEvent(agentEvent{
|
|
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), agentNow()),
|
|
}, "agent_callback", strings.TrimSpace(payload.DeploymentID), agentRequestID(c), "ok")
|
|
}
|
|
common.ApiSuccess(c, gin.H{
|
|
"event_id": payload.EventID,
|
|
"inserted": inserted,
|
|
"idempotent": !inserted,
|
|
"deduplicated": !inserted,
|
|
"deployment_id": payload.DeploymentID,
|
|
})
|
|
}
|
|
|
|
func AgentListUserDeploymentArtifacts(c *gin.Context) {
|
|
record, ok := requireAuthenticatedUserAgentDeployment(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
record = reconcileDeploymentFromRuntime(c.Request.Context(), record)
|
|
items, err := model.ListAgentArtifacts(model.ListAgentArtifactsFilter{
|
|
DeploymentID: record.DeploymentID,
|
|
Limit: 500,
|
|
})
|
|
if err != nil {
|
|
common.SysLog("AgentListUserDeploymentArtifacts: " + err.Error())
|
|
agentError(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 AgentGetUserDeploymentArtifactContent(c *gin.Context) {
|
|
record, ok := requireAuthenticatedUserAgentDeployment(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
artifactID := strings.TrimSpace(c.Param("artifact_id"))
|
|
if artifactID == "" {
|
|
agentError(c, "ARTIFACT_ID_REQUIRED", "artifact_id is required")
|
|
return
|
|
}
|
|
artifact, found, err := model.GetAgentArtifactByDeployment(record.DeploymentID, artifactID)
|
|
if err != nil {
|
|
common.SysLog("AgentGetUserDeploymentArtifactContent: " + err.Error())
|
|
agentError(c, "ARTIFACT_QUERY_FAILED", "failed to query artifact")
|
|
return
|
|
}
|
|
if !found {
|
|
agentError(c, "ARTIFACT_NOT_FOUND", "artifact not found")
|
|
return
|
|
}
|
|
cfg := agentRuntimeClientConfigForMode(agentRuntimeModeForRecord(record))
|
|
if !cfg.Enabled || strings.TrimSpace(cfg.BaseURL) == "" {
|
|
agentError(c, "RUNTIME_NOT_CONFIGURED", "runtime is not configured")
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(c.Request.Context(), cfg.Timeout)
|
|
defer cancel()
|
|
resp, err := callAgentRuntimeArtifactContent(ctx, cfg, record, artifact.ArtifactID)
|
|
if err != nil {
|
|
common.SysLog("AgentGetUserDeploymentArtifactContent: " + err.Error())
|
|
agentError(c, "ARTIFACT_CONTENT_FETCH_FAILED", "failed to fetch artifact content")
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
headers := map[string]string{}
|
|
for _, key := range []string{"Content-Disposition", "ETag", "Last-Modified", "Cache-Control"} {
|
|
if value := strings.TrimSpace(resp.Header.Get(key)); value != "" {
|
|
headers[key] = value
|
|
}
|
|
}
|
|
contentType := strings.TrimSpace(resp.Header.Get("Content-Type"))
|
|
if contentType == "" {
|
|
contentType = "application/octet-stream"
|
|
}
|
|
c.DataFromReader(http.StatusOK, resp.ContentLength, contentType, resp.Body, headers)
|
|
}
|
|
|
|
func callbackPayloadMap(row model.AgentCallbackEvent) map[string]any {
|
|
var payload agentCallbackEnvelope
|
|
if err := common.UnmarshalJsonStr(row.PayloadJSON, &payload); err != nil {
|
|
return nil
|
|
}
|
|
if payload.Payload != nil {
|
|
return payload.Payload
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func timelineEntryFromCallback(callback model.AgentCallbackEvent) 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 AgentGetUserDeploymentTimeline(c *gin.Context) {
|
|
record, ok := requireAuthenticatedUserAgentDeployment(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
record = reconcileDeploymentFromRuntime(c.Request.Context(), record)
|
|
events, err := model.ListAgentAuditEventsByDeployment(record.DeploymentID)
|
|
if err != nil {
|
|
agentError(c, "TIMELINE_QUERY_FAILED", "failed to query audit events")
|
|
return
|
|
}
|
|
callbacks, err := model.ListAgentCallbackEvents(model.ListAgentCallbackEventsFilter{DeploymentID: record.DeploymentID, Limit: 500})
|
|
if err != nil {
|
|
agentError(c, "TIMELINE_QUERY_FAILED", "failed to query callbacks")
|
|
return
|
|
}
|
|
artifacts, err := model.ListAgentArtifacts(model.ListAgentArtifactsFilter{DeploymentID: record.DeploymentID, Limit: 500})
|
|
if err != nil {
|
|
agentError(c, "TIMELINE_QUERY_FAILED", "failed to query artifacts")
|
|
return
|
|
}
|
|
snapshots, err := model.ListAgentSKSnapshots(record.DeploymentID)
|
|
if err != nil {
|
|
agentError(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": withDisplayStatus(record),
|
|
"events": events,
|
|
"callbacks": callbacks,
|
|
"artifacts": artifacts,
|
|
"sk_snapshots": snapshots,
|
|
"timeline": timeline,
|
|
})
|
|
}
|