fix: propagate agnet runtime stop

2026-05-27 15:19:57 +08:00
parent beaf1e6611
commit 5fc0532ec1
4 changed files with 158 additions and 1 deletions
@@ -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"
@@ -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)
@@ -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"
@@ -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: