diff --git a/heicode/controller/agnet_control_plane.go b/heicode/controller/agnet_control_plane.go index a384731..0195489 100644 --- a/heicode/controller/agnet_control_plane.go +++ b/heicode/controller/agnet_control_plane.go @@ -1202,9 +1202,20 @@ func AgnetStopDeployment(c *gin.Context) { agnetError(c, "DEPLOYMENT_CONFLICT", "deployment not found") return } + var stopPayload struct { + Reason string `json:"reason"` + } + if c.Request != nil && c.Request.Body != nil { + _ = common.UnmarshalBodyReusable(c, &stopPayload) + } + var runtimeOK bool + record, runtimeOK = syncAgnetRuntimeStop(c, record, stopPayload.Reason) + if !runtimeOK { + return + } record.Status = "stopped" record.Phase = "stopped" - record.RuntimeState = "stopped" + record.RuntimeState = firstNonEmpty(record.RuntimeState, "stopped") record.FailureReason = "" for i := range record.AgentInstances { record.AgentInstances[i].Phase = "stopped" diff --git a/heicode/controller/agnet_control_plane_test.go b/heicode/controller/agnet_control_plane_test.go index fc86126..ef29154 100644 --- a/heicode/controller/agnet_control_plane_test.go +++ b/heicode/controller/agnet_control_plane_test.go @@ -1193,6 +1193,55 @@ func TestAgnetStopDeploymentPersistsState(t *testing.T) { require.Contains(t, getRecorder.Body.String(), `"runtime_state":"stopped"`) } +func TestAgnetStopDeploymentPropagatesToRuntime(t *testing.T) { + setupAgnetControlPlaneTestDB(t) + resetAgnetControlPlaneState(t) + + stopCalled := false + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/api/agnet/deployments": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"deployment_id":"runtime-dep-stop","status":"pending"}}`)) + case r.Method == http.MethodPost && r.URL.Path == "/api/agnet/deployments/runtime-dep-stop/stop": + stopCalled = true + require.Equal(t, "Bearer service-token", r.Header.Get("Authorization")) + body, err := io.ReadAll(r.Body) + require.NoError(t, err) + require.Contains(t, string(body), `"reason":"runtime stop smoke"`) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"deployment_id":"runtime-dep-stop","status":"stopped"}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + t.Setenv("AGNET_RUNTIME_ENABLED", "true") + t.Setenv("AGNET_RUNTIME_ASYNC", "false") + t.Setenv("AGNET_RUNTIME_BASE_URL", server.URL) + t.Setenv("AGNET_RUNTIME_SERVICE_TOKEN", "service-token") + + recorder, envelope := postAgnetCreateDeployment(t, baseAgnetResourceGrantPlan()) + require.Equal(t, http.StatusOK, recorder.Code) + require.True(t, envelope.Success) + var createBody map[string]any + require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &createBody)) + deploymentID := createBody["data"].(map[string]any)["deployment_id"].(string) + + stopRecorder := httptest.NewRecorder() + stopCtx, _ := gin.CreateTestContext(stopRecorder) + stopCtx.Params = gin.Params{{Key: "deployment_id", Value: deploymentID}} + stopCtx.Request = httptest.NewRequest(http.MethodPost, "/api/agnet/deployments/"+deploymentID+"/stop", strings.NewReader(`{"reason":"runtime stop smoke"}`)) + stopCtx.Request.Header.Set("Content-Type", "application/json") + AgnetStopDeployment(stopCtx) + + require.Equal(t, http.StatusOK, stopRecorder.Code) + require.Contains(t, stopRecorder.Body.String(), `"success":true`) + require.Contains(t, stopRecorder.Body.String(), `"runtime_state":"stopped"`) + require.True(t, stopCalled) +} + func TestManagerOnlyAgnetSmokeFlow(t *testing.T) { setupAgnetControlPlaneTestDB(t) resetAgnetControlPlaneState(t) diff --git a/heicode/controller/agnet_runtime_client.go b/heicode/controller/agnet_runtime_client.go index bcc6d27..eaf4539 100644 --- a/heicode/controller/agnet_runtime_client.go +++ b/heicode/controller/agnet_runtime_client.go @@ -28,6 +28,7 @@ type agnetRuntimeConfig struct { Token string CreatePath string HealthPath string + StopPath string Timeout time.Duration } @@ -50,6 +51,7 @@ func agnetRuntimeClientConfig() agnetRuntimeConfig { Token: strings.TrimSpace(common.GetEnvOrDefaultString("AGNET_RUNTIME_SERVICE_TOKEN", "")), CreatePath: common.GetEnvOrDefaultString("AGNET_RUNTIME_CREATE_PATH", "/api/agnet/deployments"), HealthPath: common.GetEnvOrDefaultString("AGNET_RUNTIME_HEALTH_PATH", "/api/agnet/health"), + StopPath: common.GetEnvOrDefaultString("AGNET_RUNTIME_STOP_PATH", "/api/agnet/deployments/{deployment_id}/stop"), Timeout: time.Duration(timeoutSec) * time.Second, } } @@ -349,6 +351,99 @@ func callAgnetRuntimeCreate(ctx context.Context, cfg agnetRuntimeConfig, record return result, nil } +func agnetRuntimeStopPath(cfg agnetRuntimeConfig, runtimeDeploymentID string) string { + path := strings.TrimSpace(cfg.StopPath) + if path == "" { + path = "/api/agnet/deployments/{deployment_id}/stop" + } + return strings.ReplaceAll(path, "{deployment_id}", url.PathEscape(runtimeDeploymentID)) +} + +func callAgnetRuntimeStop(ctx context.Context, cfg agnetRuntimeConfig, record agnetDeploymentRecord, reason string) (agnetRuntimeSyncResult, error) { + runtimeDeploymentID := strings.TrimSpace(record.RuntimeDeploymentID) + if runtimeDeploymentID == "" { + return agnetRuntimeSyncResult{}, nil + } + endpoint, err := agnetRuntimeURL(cfg.BaseURL, agnetRuntimeStopPath(cfg, runtimeDeploymentID)) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + payload, err := common.Marshal(gin.H{ + "reason": firstNonEmpty(strings.TrimSpace(reason), "Heicode Manager requested stop"), + "manager_deployment_id": record.DeploymentID, + }) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload)) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + agnetRuntimeHeaders(req, cfg, record) + client := &http.Client{Timeout: cfg.Timeout} + resp, err := client.Do(req) + if err != nil { + return agnetRuntimeSyncResult{}, err + } + defer resp.Body.Close() + body, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if readErr != nil { + return agnetRuntimeSyncResult{}, readErr + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return agnetRuntimeSyncResult{RawStatusCode: resp.StatusCode}, fmt.Errorf("runtime stop returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + var envelope map[string]any + if len(body) > 0 { + if err := common.Unmarshal(body, &envelope); err != nil { + return agnetRuntimeSyncResult{RawStatusCode: resp.StatusCode}, err + } + } + if message := agnetRuntimeEnvelopeError(envelope); message != "" { + return agnetRuntimeSyncResult{RawStatusCode: resp.StatusCode}, errors.New(message) + } + data := extractAgnetRuntimeData(envelope) + return agnetRuntimeSyncResult{ + RuntimeDeploymentID: firstNonEmpty(stringFromMap(data, "runtime_deployment_id", "deployment_id", "id"), runtimeDeploymentID), + RuntimeSwarmID: stringFromMap(data, "swarm_id", "runtime_swarm_id"), + RuntimeStatus: stringFromMap(data, "runtime_status", "status"), + RawStatusCode: resp.StatusCode, + }, nil +} + +func syncAgnetRuntimeStop(c *gin.Context, record agnetDeploymentRecord, reason string) (agnetDeploymentRecord, bool) { + cfg := agnetRuntimeClientConfig() + if !cfg.Enabled || strings.TrimSpace(record.RuntimeDeploymentID) == "" { + return record, true + } + ctx, cancel := context.WithTimeout(c.Request.Context(), cfg.Timeout) + defer cancel() + result, err := callAgnetRuntimeStop(ctx, cfg, record, reason) + record.RuntimeLastSyncAt = agnetNow() + if err != nil { + record.RuntimeState = agnetRuntimeStateFailed + record.FailureReason = truncateAgnetFailureReason(err.Error()) + record.UpdatedAt = agnetNow() + _ = updateAgnetDeploymentRecord(record) + agnetMu.Lock() + agnetDeployments[record.DeploymentID] = record + agnetMu.Unlock() + recordAgnetRuntimeSyncAudit(record, "runtime.stop.failed", "failed") + agnetError(c, "RUNTIME_STOP_FAILED", record.FailureReason) + return record, false + } + record.RuntimeState = firstNonEmpty(result.RuntimeStatus, "stopped") + if result.RuntimeDeploymentID != "" { + record.RuntimeDeploymentID = result.RuntimeDeploymentID + } + if result.RuntimeSwarmID != "" { + record.RuntimeSwarmID = result.RuntimeSwarmID + } + record.FailureReason = "" + recordAgnetRuntimeSyncAudit(record, "runtime.stop.accepted", "ok") + return record, true +} + func updateAgnetRuntimeSyncState(record agnetDeploymentRecord, result agnetRuntimeSyncResult, syncErr error) agnetDeploymentRecord { record.RuntimeLastSyncAt = agnetNow() if syncErr != nil { @@ -438,6 +533,7 @@ func AgnetRuntimeHealth(c *gin.Context) { "configured": cfg.BaseURL != "", "create_path": cfg.CreatePath, "health_path": cfg.HealthPath, + "stop_path": cfg.StopPath, } if cfg.BaseURL == "" { data["status"] = "not_configured" diff --git a/heicode/docker-compose.azure-vm.yml b/heicode/docker-compose.azure-vm.yml index da720d7..ed93201 100644 --- a/heicode/docker-compose.azure-vm.yml +++ b/heicode/docker-compose.azure-vm.yml @@ -58,6 +58,7 @@ services: - AGNET_RUNTIME_BASE_URL=${AGNET_RUNTIME_BASE_URL:-http://20.212.121.126} - AGNET_RUNTIME_CREATE_PATH=${AGNET_RUNTIME_CREATE_PATH:-/api/agnet/deployments} - AGNET_RUNTIME_HEALTH_PATH=${AGNET_RUNTIME_HEALTH_PATH:-/api/agnet/health} + - AGNET_RUNTIME_STOP_PATH=${AGNET_RUNTIME_STOP_PATH:-/api/agnet/deployments/{deployment_id}/stop} - AGNET_RUNTIME_SERVICE_TOKEN=${AGNET_RUNTIME_SERVICE_TOKEN:-} - AGNET_RUNTIME_CALLBACK_SIGNING_SECRET_REF=${AGNET_RUNTIME_CALLBACK_SIGNING_SECRET_REF:-} networks: