diff --git a/heicode/controller/agnet_callback.go b/heicode/controller/agnet_callback.go new file mode 100644 index 00000000..c51dc3a3 --- /dev/null +++ b/heicode/controller/agnet_callback.go @@ -0,0 +1,245 @@ +package controller + +import ( + "os" + "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"` + TaskID string `json:"task_id"` + CorrelationID string `json:"correlation_id"` + Source string `json:"source"` + Metadata map[string]any `json:"metadata"` + 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 validateAgnetCallbackAuth(c *gin.Context) bool { + if common.GetEnvOrDefaultBool("AGNET_CALLBACK_AUTH_DISABLED", false) { + return true + } + 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.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) (agnetDeploymentRecord, bool) { + if deploymentID == "" { + return agnetDeploymentRecord{}, false + } + return findAgnetDeploymentRecord(deploymentID) +} + +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 AgnetReceiveSwarmEventCallback(c *gin.Context) { + if !validateAgnetCallbackAuth(c) { + return + } + var payload agnetCallbackEnvelope + if err := c.ShouldBindJSON(&payload); err != nil { + agnetError(c, "CALLBACK_INVALID", err.Error()) + return + } + payload.EventID = strings.TrimSpace(payload.EventID) + payload.EventType = strings.TrimSpace(payload.EventType) + if payload.EventID == "" || payload.EventType == "" { + agnetError(c, "CALLBACK_INVALID", "event_id and event_type are required") + return + } + if payload.IdempotencyKey == "" { + payload.IdempotencyKey = payload.EventID + } + if agnetCallbackHasPlaintextSecret(payload) { + agnetError(c, "CALLBACK_SECRET_REJECTED", "callbacks must not contain plaintext credential fields") + return + } + + record, _ := agnetCallbackDeploymentContext(strings.TrimSpace(payload.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(), + }) + 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 + } + 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: agnetNow(), + }, "agnet_callback", strings.TrimSpace(payload.DeploymentID), agnetRequestID(c), "ok") + } + common.ApiSuccess(c, gin.H{ + "event_id": payload.EventID, + "inserted": inserted, + "idempotent": !inserted, + }) +} + +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{"items": items, "total": len(items)}) +} + +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, gin.H{"kind": "callback", "at": callback.CreatedAtMs, "event": callback.EventType}) + } + 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": 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 e6bce7d8..5ed8f576 100644 --- a/heicode/controller/agnet_control_plane.go +++ b/heicode/controller/agnet_control_plane.go @@ -973,22 +973,22 @@ func requireAuthenticatedUserAgnetDeployment(c *gin.Context) (agnetDeploymentRec return record, true } -func createAgnetDeployment(c *gin.Context, enforceUserScope bool) { +func createAgnetDeploymentRecord(c *gin.Context, enforceUserScope bool) (agnetDeploymentRecord, bool) { var req agnetDeploymentRequest if err := c.ShouldBindJSON(&req); err != nil { agnetError(c, "POLICY_REJECTED", err.Error()) - return + return agnetDeploymentRecord{}, false } plan := req.Plan if enforceUserScope { if !enforceAuthenticatedAgnetUserContext(c, &plan) { - return + return agnetDeploymentRecord{}, false } } else { applyAuthenticatedManagerUserContext(c, &plan) } if !validateOrchestrationPlan(c, plan) { - return + return agnetDeploymentRecord{}, false } plan.SubMode = normalizeAgnetSubMode(plan.SubMode) @@ -1022,15 +1022,18 @@ func createAgnetDeployment(c *gin.Context, enforceUserScope bool) { if err := persistAgnetDeploymentRecord(record, agnetDeploymentRequest{Plan: plan}); err != nil { common.SysLog("AgnetCreateDeployment persist: " + err.Error()) agnetError(c, "DEPLOYMENT_PERSIST_FAILED", "failed to persist Agnet deployment") - return + return agnetDeploymentRecord{}, false } agnetMu.Lock() agnetDeployments[deploymentID] = record agnetMu.Unlock() recordAgnetAuditEvent(event, "agnet_control_plane", deploymentID, agnetRequestID(c), "ok") + return record, true +} - common.ApiSuccess(c, gin.H{ - "deployment_id": deploymentID, +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, @@ -1038,6 +1041,30 @@ func createAgnetDeployment(c *gin.Context, enforceUserScope bool) { "failure_reason": record.FailureReason, "agent_instances": record.AgentInstances, "permission_manifest": record.ResourceGrantManifest, + } + for key, value := range extra { + data[key] = value + } + common.ApiSuccess(c, data) +} + +func createAgnetDeployment(c *gin.Context, enforceUserScope bool) { + record, ok := createAgnetDeploymentRecord(c, enforceUserScope) + if !ok { + return + } + writeAgnetDeploymentCreateSuccess(c, record, nil) +} + +func AgnetCreateUserSwarm(c *gin.Context) { + record, ok := createAgnetDeploymentRecord(c, true) + if !ok { + return + } + writeAgnetDeploymentCreateSuccess(c, record, gin.H{ + "swarm_id": record.DeploymentID, + "adapter": "manager-local-control-plane", + "source": "api_swarms_adapter", }) } @@ -1440,6 +1467,24 @@ func AgnetResolveSKSnapshots(c *gin.Context) { agnetMu.Lock() agnetSnapshots[deploymentID] = snapshots agnetMu.Unlock() + rows := make([]model.AgnetSKSnapshot, 0, len(snapshots)) + for _, snapshot := range snapshots { + rows = append(rows, model.AgnetSKSnapshot{ + SnapshotID: snapshot.SnapshotID, + DeploymentID: snapshot.DeploymentID, + UserID: snapshot.UserID, + BindingScope: snapshot.BindingScope, + SourceType: snapshot.SourceType, + SourceRef: snapshot.SourceRef, + ResolvedAt: snapshot.ResolvedAt, + ResolvedAtMs: agnetTimestampMs(snapshot.ResolvedAt), + }) + } + if err := model.InsertAgnetSKSnapshots(rows); err != nil { + common.SysLog("AgnetResolveSKSnapshots persist: " + err.Error()) + agnetError(c, "SK_SNAPSHOT_PERSIST_FAILED", "failed to persist sk snapshots") + return + } snapEvent := agnetEvent{ EventID: "evt_" + common.GetUUID()[:12], Event: "sk_snapshot_refreshed", @@ -1469,6 +1514,26 @@ func AgnetListSKSnapshots(c *gin.Context) { agnetMu.RLock() items := agnetSnapshots[deploymentID] agnetMu.RUnlock() + if len(items) == 0 { + rows, err := model.ListAgnetSKSnapshots(deploymentID) + if err != nil { + common.SysLog("AgnetListSKSnapshots: " + err.Error()) + agnetError(c, "SK_SOURCE_UNRESOLVABLE", "failed to query sk snapshots") + return + } + items = make([]agnetSKSnapshot, 0, len(rows)) + for _, row := range rows { + items = append(items, agnetSKSnapshot{ + SnapshotID: row.SnapshotID, + DeploymentID: row.DeploymentID, + UserID: row.UserID, + BindingScope: row.BindingScope, + SourceType: row.SourceType, + SourceRef: row.SourceRef, + ResolvedAt: row.ResolvedAt, + }) + } + } common.ApiSuccess(c, gin.H{ "deployment_id": deploymentID, "items": items, @@ -1476,6 +1541,13 @@ func AgnetListSKSnapshots(c *gin.Context) { }) } +func AgnetListUserSKSnapshots(c *gin.Context) { + if _, ok := requireAuthenticatedUserAgnetDeployment(c); !ok { + return + } + AgnetListSKSnapshots(c) +} + func AgnetListAuditLogs(c *gin.Context) { // Persistent audit query — survives container restarts. Filters // are optional; the dashboard call usually narrows by user_id + diff --git a/heicode/controller/agnet_control_plane_test.go b/heicode/controller/agnet_control_plane_test.go index c7f4be57..d93d444c 100644 --- a/heicode/controller/agnet_control_plane_test.go +++ b/heicode/controller/agnet_control_plane_test.go @@ -56,6 +56,9 @@ func setupAgnetControlPlaneTestDB(t *testing.T) *gorm.DB { &model.ResourceGrant{}, &model.AgnetAuditEvent{}, &model.AgnetDeployment{}, + &model.AgnetCallbackEvent{}, + &model.AgnetArtifact{}, + &model.AgnetSKSnapshot{}, )) t.Cleanup(func() { if model.DB == db { @@ -554,6 +557,190 @@ func TestAgnetUserDeploymentSimulationRejectsOtherUsersDeployment(t *testing.T) require.Equal(t, "DEPLOYMENT_FORBIDDEN", envelope.Error.Code) } +func TestAgnetUserSwarmsAdapterCreatesScopedDeployment(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + plan := baseAgnetResourceGrantPlan() + plan.UserContext.UserID = "" + for idx := range plan.Agents[0].ResourceGrants { + plan.Agents[0].ResourceGrants[idx].UserID = "" + } + body, err := common.Marshal(agnetDeploymentRequest{Plan: plan}) + require.NoError(t, err) + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Set("id", 7) + ctx.Set("group", "development") + ctx.Request = httptest.NewRequest(http.MethodPost, "/api/swarms", strings.NewReader(string(body))) + ctx.Request.Header.Set("Content-Type", "application/json") + + AgnetCreateUserSwarm(ctx) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Contains(t, recorder.Body.String(), `"swarm_id"`) + require.Contains(t, recorder.Body.String(), `"deployment_id"`) + require.Contains(t, recorder.Body.String(), `"user_id":"7"`) +} + +func TestAgnetCallbackStoresEventArtifactAndIsIdempotent(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) + + body := fmt.Sprintf(`{ + "event_id":"evt-callback-1", + "idempotency_key":"idem-callback-1", + "event_type":"artifact.created", + "deployment_id":%q, + "task_id":"task-1", + "correlation_id":"corr-p1-resource-grant", + "artifact":{ + "artifact_id":"art-1", + "artifact_type":"summary", + "title":"Build summary", + "summary":"Created backend scaffold", + "uri":"artifact://task-1/summary" + } + }`, deploymentID) + + 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`) + } + + var callbackCount int64 + require.NoError(t, db.Model(&model.AgnetCallbackEvent{}).Where("event_id = ?", "evt-callback-1").Count(&callbackCount).Error) + require.Equal(t, int64(1), callbackCount) + var artifactCount int64 + require.NoError(t, db.Model(&model.AgnetArtifact{}).Where("artifact_id = ?", "art-1").Count(&artifactCount).Error) + require.Equal(t, int64(1), artifactCount) +} + +func TestAgnetCallbackRejectsPlaintextSecretsAndMissingToken(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + t.Setenv("AGNET_CALLBACK_TOKEN", "callback-token") + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Request = httptest.NewRequest(http.MethodPost, "/api/agnet/callbacks/swarm-events", strings.NewReader(`{"event_id":"evt-no-token","event_type":"status.updated"}`)) + ctx.Request.Header.Set("Content-Type", "application/json") + AgnetReceiveSwarmEventCallback(ctx) + var missingToken agnetCreateTestEnvelope + require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &missingToken)) + require.False(t, missingToken.Success) + require.Equal(t, "CALLBACK_UNAUTHORIZED", missingToken.Error.Code) + + secretRecorder := httptest.NewRecorder() + secretCtx, _ := gin.CreateTestContext(secretRecorder) + secretCtx.Request = httptest.NewRequest(http.MethodPost, "/api/agnet/callbacks/swarm-events", strings.NewReader(`{ + "event_id":"evt-secret", + "event_type":"status.updated", + "metadata":{"access_token":"must-not-store"} + }`)) + secretCtx.Request.Header.Set("Content-Type", "application/json") + secretCtx.Request.Header.Set("X-Agnet-Service-Token", "callback-token") + AgnetReceiveSwarmEventCallback(secretCtx) + var secretEnvelope agnetCreateTestEnvelope + require.NoError(t, common.Unmarshal(secretRecorder.Body.Bytes(), &secretEnvelope)) + require.False(t, secretEnvelope.Success) + require.Equal(t, "CALLBACK_SECRET_REJECTED", secretEnvelope.Error.Code) +} + +func TestAgnetArtifactsAndTimelineReturnPersistedRelatedRecords(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + t.Setenv("AGNET_CALLBACK_TOKEN", "callback-token") + + plan := baseAgnetResourceGrantPlan() + plan.UserContext.UserID = "7" + plan.SubMode = "waterfall" + 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) + + callbackBody := fmt.Sprintf(`{ + "event_id":"evt-artifact-timeline", + "idempotency_key":"idem-artifact-timeline", + "event_type":"artifact.created", + "deployment_id":%q, + "task_id":"task-timeline", + "correlation_id":"corr-p1-resource-grant", + "artifact":{"artifact_id":"art-timeline","artifact_type":"summary","title":"Timeline artifact","summary":"Done","uri":"artifact://task-timeline/summary"} + }`, deploymentID) + callbackRecorder := httptest.NewRecorder() + callbackCtx, _ := gin.CreateTestContext(callbackRecorder) + callbackCtx.Request = httptest.NewRequest(http.MethodPost, "/api/agnet/callbacks/swarm-events", strings.NewReader(callbackBody)) + callbackCtx.Request.Header.Set("Content-Type", "application/json") + callbackCtx.Request.Header.Set("X-Agnet-Service-Token", "callback-token") + AgnetReceiveSwarmEventCallback(callbackCtx) + require.Contains(t, callbackRecorder.Body.String(), `"success":true`) + + snapshotRecorder := httptest.NewRecorder() + snapshotCtx, _ := gin.CreateTestContext(snapshotRecorder) + snapshotCtx.Set("id", 7) + snapshotCtx.Request = httptest.NewRequest(http.MethodPost, "/api/agnet/sk-snapshots/resolve", strings.NewReader(fmt.Sprintf(`{"deployment_id":%q}`, deploymentID))) + snapshotCtx.Request.Header.Set("Content-Type", "application/json") + AgnetResolveSKSnapshots(snapshotCtx) + require.Equal(t, http.StatusOK, snapshotRecorder.Code) + + resetAgnetControlPlaneState(t) + + artifactRecorder := httptest.NewRecorder() + artifactCtx, _ := gin.CreateTestContext(artifactRecorder) + artifactCtx.Set("id", 7) + artifactCtx.Params = gin.Params{{Key: "deployment_id", Value: deploymentID}} + artifactCtx.Request = httptest.NewRequest(http.MethodGet, "/api/agnet/user/deployments/"+deploymentID+"/artifacts", nil) + AgnetListUserDeploymentArtifacts(artifactCtx) + require.Equal(t, http.StatusOK, artifactRecorder.Code) + require.Contains(t, artifactRecorder.Body.String(), `"art-timeline"`) + + snapshotListRecorder := httptest.NewRecorder() + snapshotListCtx, _ := gin.CreateTestContext(snapshotListRecorder) + snapshotListCtx.Set("id", 7) + snapshotListCtx.Params = gin.Params{{Key: "deployment_id", Value: deploymentID}} + snapshotListCtx.Request = httptest.NewRequest(http.MethodGet, "/api/agnet/user/deployments/"+deploymentID+"/sk-snapshots", nil) + AgnetListUserSKSnapshots(snapshotListCtx) + require.Equal(t, http.StatusOK, snapshotListRecorder.Code) + require.Contains(t, snapshotListRecorder.Body.String(), `"deployment_id":"`+deploymentID+`"`) + + timelineRecorder := httptest.NewRecorder() + timelineCtx, _ := gin.CreateTestContext(timelineRecorder) + timelineCtx.Set("id", 7) + timelineCtx.Params = gin.Params{{Key: "deployment_id", Value: deploymentID}} + timelineCtx.Request = httptest.NewRequest(http.MethodGet, "/api/agnet/user/deployments/"+deploymentID+"/timeline", nil) + AgnetGetUserDeploymentTimeline(timelineCtx) + require.Equal(t, http.StatusOK, timelineRecorder.Code) + require.Contains(t, timelineRecorder.Body.String(), `"callbacks"`) + require.Contains(t, timelineRecorder.Body.String(), `"artifacts"`) + require.Contains(t, timelineRecorder.Body.String(), `"sk_snapshots"`) + require.Contains(t, timelineRecorder.Body.String(), `"timeline"`) +} + func TestAgnetStopDeploymentPersistsState(t *testing.T) { setupAgnetControlPlaneTestDB(t) resetAgnetControlPlaneState(t) diff --git a/heicode/model/agnet_artifact.go b/heicode/model/agnet_artifact.go new file mode 100644 index 00000000..540a1ea2 --- /dev/null +++ b/heicode/model/agnet_artifact.go @@ -0,0 +1,66 @@ +package model + +type AgnetArtifact struct { + Id int `gorm:"primaryKey" json:"id"` + ArtifactID string `gorm:"type:varchar(128);uniqueIndex" json:"artifact_id"` + 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"` + ArtifactType string `gorm:"type:varchar(64);index" json:"artifact_type"` + Title string `gorm:"type:varchar(256)" json:"title"` + Summary string `gorm:"type:text" json:"summary"` + URI string `gorm:"type:varchar(1024)" json:"uri"` + Checksum string `gorm:"type:varchar(128)" json:"checksum"` + MetadataJSON string `gorm:"type:text" json:"metadata_json"` + CreatedAtMs int64 `gorm:"bigint;index" json:"created_at_ms"` +} + +func (AgnetArtifact) TableName() string { + return "agnet_artifacts" +} + +type ListAgnetArtifactsFilter struct { + DeploymentID string + TaskID string + CorrelationID string + Limit int +} + +func UpsertAgnetArtifact(row *AgnetArtifact) error { + if DB == nil || row == nil { + return nil + } + var existing AgnetArtifact + if row.ArtifactID != "" { + if err := DB.Where("artifact_id = ?", row.ArtifactID).First(&existing).Error; err == nil { + row.Id = existing.Id + return DB.Model(&existing).Updates(row).Error + } + } + return DB.Create(row).Error +} + +func ListAgnetArtifacts(f ListAgnetArtifactsFilter) ([]AgnetArtifact, error) { + if DB == nil { + return nil, nil + } + q := DB.Model(&AgnetArtifact{}) + if f.DeploymentID != "" { + q = q.Where("deployment_id = ?", f.DeploymentID) + } + if f.TaskID != "" { + q = q.Where("task_id = ?", f.TaskID) + } + if f.CorrelationID != "" { + q = q.Where("correlation_id = ?", f.CorrelationID) + } + limit := f.Limit + if limit <= 0 || limit > 1000 { + limit = 200 + } + var items []AgnetArtifact + err := q.Order("created_at_ms asc, id asc").Limit(limit).Find(&items).Error + return items, err +} diff --git a/heicode/model/agnet_callback.go b/heicode/model/agnet_callback.go new file mode 100644 index 00000000..f7caf0ec --- /dev/null +++ b/heicode/model/agnet_callback.go @@ -0,0 +1,76 @@ +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"` +} + +func (AgnetCallbackEvent) TableName() string { + return "agnet_callback_events" +} + +type ListAgnetCallbackEventsFilter struct { + DeploymentID string + TaskID string + CorrelationID string + Limit int +} + +func InsertAgnetCallbackEvent(row *AgnetCallbackEvent) (bool, error) { + if DB == nil || row == nil { + return false, nil + } + if row.EventID == "" { + return false, errors.New("event_id is required") + } + var existing AgnetCallbackEvent + if err := DB.Where("event_id = ?", row.EventID).First(&existing).Error; err == nil { + return false, nil + } + if row.IdempotencyKey != "" { + if err := DB.Where("idempotency_key = ?", row.IdempotencyKey).First(&existing).Error; err == nil { + return false, nil + } + } + if err := DB.Create(row).Error; err != nil { + return false, err + } + return true, nil +} + +func ListAgnetCallbackEvents(f ListAgnetCallbackEventsFilter) ([]AgnetCallbackEvent, error) { + if DB == nil { + return nil, nil + } + q := DB.Model(&AgnetCallbackEvent{}) + if f.DeploymentID != "" { + q = q.Where("deployment_id = ?", f.DeploymentID) + } + if f.TaskID != "" { + q = q.Where("task_id = ?", f.TaskID) + } + if f.CorrelationID != "" { + q = q.Where("correlation_id = ?", f.CorrelationID) + } + limit := f.Limit + if limit <= 0 || limit > 1000 { + limit = 200 + } + var items []AgnetCallbackEvent + err := q.Order("created_at_ms asc, id asc").Limit(limit).Find(&items).Error + return items, err +} diff --git a/heicode/model/agnet_sk_snapshot.go b/heicode/model/agnet_sk_snapshot.go new file mode 100644 index 00000000..7bfcdaa0 --- /dev/null +++ b/heicode/model/agnet_sk_snapshot.go @@ -0,0 +1,35 @@ +package model + +type AgnetSKSnapshot struct { + Id int `gorm:"primaryKey" json:"id"` + SnapshotID string `gorm:"type:varchar(128);uniqueIndex" json:"snapshot_id"` + DeploymentID string `gorm:"type:varchar(64);index" json:"deployment_id"` + UserID string `gorm:"type:varchar(64);index" json:"user_id"` + BindingScope string `gorm:"type:varchar(512);index" json:"binding_scope"` + SourceType string `gorm:"type:varchar(64)" json:"source_type"` + SourceRef string `gorm:"type:varchar(1024)" json:"source_ref"` + ResolvedAt string `gorm:"type:varchar(32)" json:"resolved_at"` + ResolvedAtMs int64 `gorm:"bigint;index" json:"resolved_at_ms"` +} + +func (AgnetSKSnapshot) TableName() string { + return "agnet_sk_snapshots" +} + +func InsertAgnetSKSnapshots(items []AgnetSKSnapshot) error { + if DB == nil || len(items) == 0 { + return nil + } + return DB.Create(&items).Error +} + +func ListAgnetSKSnapshots(deploymentID string) ([]AgnetSKSnapshot, error) { + if DB == nil || deploymentID == "" { + return nil, nil + } + var items []AgnetSKSnapshot + err := DB.Where("deployment_id = ?", deploymentID). + Order("resolved_at_ms asc, id asc"). + Find(&items).Error + return items, err +} diff --git a/heicode/model/main.go b/heicode/model/main.go index 12a31d73..e382c778 100644 --- a/heicode/model/main.go +++ b/heicode/model/main.go @@ -286,6 +286,9 @@ func migrateDB() error { &AgnetApprovalRequest{}, &AgnetCredentialLease{}, &AgnetDeployment{}, + &AgnetCallbackEvent{}, + &AgnetArtifact{}, + &AgnetSKSnapshot{}, // V2 device-binding: X25519 keypair the Manager uses for ECDH // body decryption. See model/server_key.go. &ServerKey{}, @@ -355,6 +358,9 @@ func migrateDBFast() error { {&AgnetApprovalRequest{}, "AgnetApprovalRequest"}, {&AgnetCredentialLease{}, "AgnetCredentialLease"}, {&AgnetDeployment{}, "AgnetDeployment"}, + {&AgnetCallbackEvent{}, "AgnetCallbackEvent"}, + {&AgnetArtifact{}, "AgnetArtifact"}, + {&AgnetSKSnapshot{}, "AgnetSKSnapshot"}, {&AgnetAuditEvent{}, "AgnetAuditEvent"}, } // 动态计算migration数量，确保errChan缓冲区足够大 diff --git a/heicode/router/api-router.go b/heicode/router/api-router.go index 8637cf60..e624a67c 100644 --- a/heicode/router/api-router.go +++ b/heicode/router/api-router.go @@ -59,6 +59,8 @@ func SetApiRouter(router *gin.Engine) { apiRouter.POST("/stripe/webhook", controller.StripeWebhook) apiRouter.POST("/creem/webhook", controller.CreemWebhook) apiRouter.POST("/waffo/webhook", controller.WaffoWebhook) + apiRouter.POST("/agnet/callbacks/swarm-events", controller.AgnetReceiveSwarmEventCallback) + apiRouter.POST("/swarms", middleware.UserAuth(), controller.AgnetCreateUserSwarm) //apiRouter.POST("/waffo-pancake/webhook", controller.WaffoPancakeWebhook) // Universal secure verification routes @@ -508,6 +510,9 @@ func SetApiRouter(router *gin.Engine) { agnetApprovalRoute.GET("/user/deployments/:deployment_id/metrics", controller.AgnetGetUserDeploymentMetrics) agnetApprovalRoute.GET("/user/deployments/:deployment_id/events", controller.AgnetListUserDeploymentEvents) agnetApprovalRoute.POST("/user/deployments/:deployment_id/simulate-events", controller.AgnetSimulateUserDeploymentEvents) + agnetApprovalRoute.GET("/user/deployments/:deployment_id/artifacts", controller.AgnetListUserDeploymentArtifacts) + agnetApprovalRoute.GET("/user/deployments/:deployment_id/sk-snapshots", controller.AgnetListUserSKSnapshots) + agnetApprovalRoute.GET("/user/deployments/:deployment_id/timeline", controller.AgnetGetUserDeploymentTimeline) agnetApprovalRoute.POST("/user/tasks/:task_id/deployment-draft", controller.AgnetCreateTaskDeploymentDraft) } diff --git a/heicode/web/default/src/features/agnet-console/api.ts b/heicode/web/default/src/features/agnet-console/api.ts index a67579b0..f12a0361 100644 --- a/heicode/web/default/src/features/agnet-console/api.ts +++ b/heicode/web/default/src/features/agnet-console/api.ts @@ -324,6 +324,39 @@ export async function getAgnetDeploymentEvents(deploymentId: string) { return res.data?.data?.items ?? [] } +export async function getAgnetDeploymentArtifacts(deploymentId: string) { + const res = await api.get< + ApiEnvelope<{ items?: Array> }> + >(`/api/agnet/user/deployments/${deploymentId}/artifacts`) + return res.data?.data?.items ?? [] +} + +export async function getAgnetDeploymentSKSnapshots(deploymentId: string) { + const res = await api.get< + ApiEnvelope<{ items?: Array> }> + >(`/api/agnet/user/deployments/${deploymentId}/sk-snapshots`) + return res.data?.data?.items ?? [] +} + +export async function getAgnetDeploymentTimeline(deploymentId: string) { + const res = await api.get< + ApiEnvelope<{ + callbacks?: Array> + artifacts?: Array> + sk_snapshots?: Array> + timeline?: Array> + }> + >(`/api/agnet/user/deployments/${deploymentId}/timeline`) + return ( + res.data?.data ?? { + callbacks: [], + artifacts: [], + sk_snapshots: [], + timeline: [], + } + ) +} + export async function simulateAgnetDeploymentEvents( deploymentId: string, events?: string[] diff --git a/heicode/web/default/src/features/agnet-console/pages.tsx b/heicode/web/default/src/features/agnet-console/pages.tsx index d93c5815..c9ae4463 100644 --- a/heicode/web/default/src/features/agnet-console/pages.tsx +++ b/heicode/web/default/src/features/agnet-console/pages.tsx @@ -61,6 +61,7 @@ import { QueryState } from '@/components/query-state' import { approveAgnetApproval, getAgnetDeploymentEvents, + getAgnetDeploymentTimeline, listAgnetApprovals, listAgnetCredentialLeases, listAgnetDeployments, @@ -784,6 +785,9 @@ export function AgnetDeploymentsPage() {
+
)} @@ -891,6 +895,117 @@ function RunAuditTimeline({ deploymentId }: { deploymentId: string }) { ) } +function RunRelatedRecordsPanel({ deploymentId }: { deploymentId: string }) { + const { t } = useTranslation() + const { data, isLoading } = useQuery({ + queryKey: ['agnet', 'deployment-timeline', deploymentId], + queryFn: () => getAgnetDeploymentTimeline(deploymentId), + enabled: Boolean(deploymentId), + refetchInterval: 15_000, + }) + const callbacks = data?.callbacks ?? [] + const artifacts = data?.artifacts ?? [] + const snapshots = data?.sk_snapshots ?? [] + const timeline = data?.timeline ?? [] + + return ( +
+
+
+

+ {t('Related records')} +

+

+ {t('Callbacks, artifacts, SK snapshots and merged timeline')} +

+
+ {isLoading ? ( + + ) : null} +
+ +
+ + + + +
+ +
+
+

+ {t('Artifacts')} +

+ {artifacts.length === 0 ? ( +

+ {t('No artifacts yet')} +

+ ) : ( +
    + {artifacts.slice(0, 5).map((item, idx) => ( +
  • +

    + {String(item.title || item.artifact_id || t('Artifact'))} +

    +

    + {String(item.summary || item.uri || '—')} +

    +
    • + ))} +
+ )} +
+
+

+ {t('SK snapshots')} +

+ {snapshots.length === 0 ? ( +

+ {t('No SK snapshots yet')} +

+ ) : ( +
    + {snapshots.slice(0, 5).map((item, idx) => ( +
  • +

    + {String(item.snapshot_id || 'snapshot')} +

    +

    + {String(item.source_type || 'source')} ·{' '} + {String(item.source_ref || '—')} +

    +
    • + ))} +
+ )} +
+
+
+ ) +} + // ============================================================================= // Events page // =============================================================================