功能：完善云管助手与GPU故障恢复核对

2026-11-13 19:08:40 +08:00
1 parent b6026f9759
commit 9f7b01e4db
@@ -14,10 +14,11 @@ type Config struct {
 	PVEURL, PVETokenID, PVETokenSecret, PVECAFile                       string
 	ExclusiveGPUNode                                                    string
 	ExclusiveGPUVMIDs                                                   []int
+	GPUAgentToken                                                       string
 }

 func Load() (Config, error) {
-	c := Config{Mode: env("VDI_MODE", "production"), Address: env("VDI_LISTEN", "127.0.0.1:8080"), DatabaseURL: os.Getenv("DATABASE_URL"), JWTSecret: os.Getenv("VDI_JWT_SECRET"), AdminUsername: os.Getenv("VDI_ADMIN_USERNAME"), AdminPassword: os.Getenv("VDI_ADMIN_PASSWORD"), PVEURL: os.Getenv("PVE_URL"), PVETokenID: os.Getenv("PVE_TOKEN_ID"), PVETokenSecret: os.Getenv("PVE_TOKEN_SECRET"), PVECAFile: os.Getenv("PVE_CA_FILE")}
+	c := Config{Mode: env("VDI_MODE", "production"), Address: env("VDI_LISTEN", "127.0.0.1:8080"), DatabaseURL: os.Getenv("DATABASE_URL"), JWTSecret: os.Getenv("VDI_JWT_SECRET"), AdminUsername: os.Getenv("VDI_ADMIN_USERNAME"), AdminPassword: os.Getenv("VDI_ADMIN_PASSWORD"), PVEURL: os.Getenv("PVE_URL"), PVETokenID: os.Getenv("PVE_TOKEN_ID"), PVETokenSecret: os.Getenv("PVE_TOKEN_SECRET"), PVECAFile: os.Getenv("PVE_CA_FILE"), GPUAgentToken: os.Getenv("VDI_GPU_AGENT_TOKEN")}
 	c.ExclusiveGPUNode = strings.TrimSpace(os.Getenv("VDI_EXCLUSIVE_GPU_NODE"))
 	ids := strings.TrimSpace(os.Getenv("VDI_EXCLUSIVE_GPU_VMIDS"))
 	if (c.ExclusiveGPUNode == "") != (ids == "") {
@@ -31,15 +31,16 @@ type Desktop struct {
 }

 type ImageTemplate struct {
-	ID          string    `json:"id" db:"id"`
-	Name        string    `json:"name" db:"name"`
-	Version     string    `json:"version" db:"version"`
-	OS          string    `json:"os" db:"os"`
-	Node        string    `json:"node" db:"node"`
-	VMID        int       `json:"vmid" db:"vmid"`
-	Digest      string    `json:"-" db:"digest"`
-	Enabled     bool      `json:"enabled" db:"enabled"`
-	ValidatedAt time.Time `json:"validatedAt" db:"validated_at"`
+	ID                       string    `json:"id" db:"id"`
+	Name                     string    `json:"name" db:"name"`
+	Version                  string    `json:"version" db:"version"`
+	OS                       string    `json:"os" db:"os"`
+	Node                     string    `json:"node" db:"node"`
+	VMID                     int       `json:"vmid" db:"vmid"`
+	Digest                   string    `json:"-" db:"digest"`
+	Enabled                  bool      `json:"enabled" db:"enabled"`
+	HardwareEncodingVerified bool      `json:"hardwareEncodingVerified" db:"hardware_encoding_verified"`
+	ValidatedAt              time.Time `json:"validatedAt" db:"validated_at"`
 }
 type DesktopPool struct {
 	ID            string    `json:"id" db:"id"`
@@ -43,7 +43,7 @@ func (s *Server) settings(c *gin.Context) {
 	if pve != nil {
 		provider = "proxmox"
 	}
-	c.JSON(200, gin.H{"mode": s.Config.Mode, "computeProvider": provider, "pveConfigured": pve != nil, "terminalTransport": "https-poll", "capabilities": []string{"diagnostics-center", "admin-mfa", "administrator-sessions", "task-history", "audit-query-export", "terminal-metric-alerts", "terminal-system-metrics", "gpu-inventory", "desktop-batches", "desktop-snapshots", "pve-connection", "desktop-hardware", "login-rate-limit", "csv-imports", "session-history", "monitor-alerts", "administrators", "password", "auth", "resources", "desktops", "templates", "pools", "clone", "desktop-connection", "terminals", "terminal-lifecycle", "terminal-policies", "users", "batch-bindings", "bindings", "tasks", "audit"}})
+	c.JSON(200, gin.H{"mode": s.Config.Mode, "computeProvider": provider, "pveConfigured": pve != nil, "terminalTransport": "https-poll", "capabilities": []string{"common-assistant", "diagnostics-center", "admin-mfa", "administrator-sessions", "task-history", "audit-query-export", "terminal-metric-alerts", "terminal-system-metrics", "gpu-inventory", "gpu-health-agent", "desktop-batches", "desktop-snapshots", "pve-connection", "desktop-hardware", "login-rate-limit", "csv-imports", "session-history", "monitor-alerts", "administrators", "password", "auth", "resources", "desktops", "templates", "pools", "clone", "desktop-connection", "terminals", "terminal-lifecycle", "terminal-policies", "users", "batch-bindings", "bindings", "tasks", "audit"}})
 }
 func (s *Server) resources(c *gin.Context) {
 	items := []model.Resource{}
@@ -514,6 +514,7 @@ func (s *Server) RunWorker(ctx context.Context) {
 	workerCtx, stop := context.WithCancel(ctx)
 	provisionDone := make(chan struct{}, 1)
 	provisionBusy := false
+	var nextGPUCheck time.Time
 	defer func() {
 		stop()
 		if provisionBusy {
@@ -532,7 +533,8 @@ func (s *Server) RunWorker(ctx context.Context) {
 			if err = conn.PingContext(ctx); err != nil {
 				return
 			}
-			// Unknown outcomes are never automatically retried after a worker restart.
+			// Unknown commands are never replayed. Known GPU tasks are observed
+			// separately below so a transient PVE read failure can recover.
 			var expired []model.Task
 			if s.DB.SelectContext(ctx, &expired, "SELECT * FROM tasks WHERE status IN ('queued','running') AND ((kind NOT IN ('provision','hardware','snapshot','batch-desktop','gpu-prepare') AND created_at<now()-interval '2 minutes') OR (kind='gpu-prepare' AND created_at<now()-interval '5 minutes') OR (kind='batch-desktop' AND ((status='queued' AND created_at<now()-interval '2 hours') OR (status='running' AND updated_at<now()-interval '2 minutes'))) OR (kind IN ('provision','hardware','snapshot') AND updated_at<now()-interval '35 minutes'))") == nil {
 				for _, t := range expired {
@@ -563,6 +565,14 @@ func (s *Server) RunWorker(ctx context.Context) {
 					_ = s.complete(ctx, t, "timeout", "结果未确认，请核对设备或 PVE 实际状态；不会自动重放")
 				}
 			}
+			if !provisionBusy && !time.Now().Before(nextGPUCheck) {
+				nextGPUCheck = time.Now().Add(15 * time.Second)
+				if recovery, readErr := s.nextGPURecovery(ctx); readErr == nil {
+					provisionBusy = true
+					go func(task model.Task) { s.recoverGPUPrepare(workerCtx, task); provisionDone <- struct{}{} }(recovery)
+					continue
+				}
+			}
 			var t model.Task
 			err := s.DB.GetContext(ctx, &t, `UPDATE tasks SET status='running',updated_at=now(),message='正在执行，等待最终回执' WHERE id=(SELECT id FROM tasks WHERE ((status='queued' AND kind!='terminal') OR (kind='provision' AND status='running' AND provider_ref!='') OR (kind IN ('hardware','snapshot','batch-desktop') AND status='running')) AND ($1 OR kind NOT IN ('provision','hardware','snapshot','batch-desktop','gpu-prepare')) ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *`, !provisionBusy)
 			if err != nil {
@@ -0,0 +1,302 @@
+package server
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"vdi-cloud/platform/internal/model"
+	"vdi-cloud/platform/internal/store"
+)
+
+type assistantCandidate struct {
+	ID    string `json:"id"`
+	Label string `json:"label"`
+	Kind  string `json:"kind"`
+}
+type assistantPlan struct {
+	Action               string               `json:"action"`
+	Kind                 string               `json:"kind"`
+	Title                string               `json:"title"`
+	Summary              string               `json:"summary"`
+	UserID               string               `json:"userId,omitempty"`
+	UserName             string               `json:"userName,omitempty"`
+	DesktopID            string               `json:"desktopId,omitempty"`
+	DesktopName          string               `json:"desktopName,omitempty"`
+	DesktopVMID          int                  `json:"desktopVmid,omitempty"`
+	Missing              []string             `json:"missing"`
+	Candidates           []assistantCandidate `json:"candidates"`
+	Risk                 string               `json:"risk"`
+	ConfirmationRequired bool                 `json:"confirmationRequired"`
+	Path                 string               `json:"path,omitempty"`
+	Reason               string               `json:"reason,omitempty"`
+}
+type assistantRun struct {
+	ID         string         `json:"id" db:"id"`
+	Input      string         `json:"input" db:"input_text"`
+	Action     string         `json:"action" db:"action"`
+	Status     string         `json:"status" db:"status"`
+	Plan       assistantPlan  `json:"plan"`
+	CreatedAt  time.Time      `json:"createdAt" db:"created_at"`
+	ExpiresAt  time.Time      `json:"expiresAt" db:"expires_at"`
+	ExecutedAt *time.Time     `json:"executedAt,omitempty" db:"executed_at"`
+	Result     map[string]any `json:"result,omitempty" db:"-"`
+}
+
+func assistantTextMatch(text, value string) bool {
+	text, value = strings.ToLower(strings.TrimSpace(text)), strings.ToLower(strings.TrimSpace(value))
+	return value != "" && strings.Contains(text, value)
+}
+
+func (s *Server) assistantPlanHandler(c *gin.Context) {
+	var in struct {
+		Text string `json:"text" binding:"required,max=500"`
+	}
+	if !s.decode(c, &in) {
+		return
+	}
+	text := strings.TrimSpace(in.Text)
+	plan := assistantPlan{Kind: "blocked", Risk: "low", Missing: []string{}, Candidates: []assistantCandidate{}, ConfirmationRequired: false}
+	revoke := strings.Contains(text, "撤销") || strings.Contains(text, "收回") || strings.Contains(text, "取消授权")
+	grant := !revoke && (strings.Contains(text, "分配") || strings.Contains(text, "授权") || strings.Contains(text, "安排") || strings.Contains(text, "开通"))
+	if !grant && !revoke {
+		if strings.Contains(text, "打不开") || strings.Contains(text, "无法连接") || strings.Contains(text, "连不上") || strings.Contains(text, "故障") {
+			plan = assistantPlan{Kind: "route", Action: "diagnose_desktop", Title: "进入桌面问题处理", Summary: "先检查终端在线状态、员工关联、桌面授权、电源和连接服务。", Path: "/help", Risk: "low", Missing: []string{}, Candidates: []assistantCandidate{}}
+		} else if strings.Contains(text, "状态") || strings.Contains(text, "在线") || strings.Contains(text, "离线") {
+			plan = assistantPlan{Kind: "route", Action: "query_status", Title: "查看当前状态", Summary: "打开设备与桌面状态页面，查看在线设备、桌面电源和最近刷新时间。", Path: "/devices", Risk: "low", Missing: []string{}, Candidates: []assistantCandidate{}}
+		} else {
+			plan.Reason = "暂不支持这个固定场景"
+			plan.Summary = "请描述分配、撤销、桌面连接问题或设备状态；复杂运维操作请使用高级版。"
+		}
+	} else {
+		ctx := c.Request.Context()
+		var users []model.User
+		var desktops []model.Desktop
+		if err := s.DB.SelectContext(ctx, &users, "SELECT * FROM users WHERE enabled ORDER BY display_name,id"); err != nil {
+			s.dbError(c, err)
+			return
+		}
+		if err := s.DB.SelectContext(ctx, &desktops, "SELECT * FROM desktops WHERE provider IN ('demo','proxmox') ORDER BY name,id"); err != nil {
+			s.dbError(c, err)
+			return
+		}
+		userMatches := []model.User{}
+		for _, user := range users {
+			if assistantTextMatch(text, user.DisplayName) || assistantTextMatch(text, user.Username) {
+				userMatches = append(userMatches, user)
+			}
+		}
+		if len(userMatches) != 1 {
+			plan.Missing = append(plan.Missing, "user")
+			for _, user := range userMatches {
+				plan.Candidates = append(plan.Candidates, assistantCandidate{ID: user.ID, Label: user.DisplayName + " · " + user.Username, Kind: "user"})
+			}
+			if len(userMatches) == 0 {
+				for i, user := range users {
+					if i >= 20 {
+						break
+					}
+					plan.Candidates = append(plan.Candidates, assistantCandidate{ID: user.ID, Label: user.DisplayName + " · " + user.Username, Kind: "user"})
+				}
+			}
+			plan.Title, plan.Summary, plan.Reason = "需要确认员工", "请写出员工姓名或账号；如果有同名员工，请选择准确对象。", "没有唯一匹配到员工"
+		} else {
+			user := userMatches[0]
+			desktopMatches := []model.Desktop{}
+			for _, desktop := range desktops {
+				if assistantTextMatch(text, desktop.Name) || strings.Contains(text, strings.TrimSpace(stringID(desktop.VMID))) {
+					desktopMatches = append(desktopMatches, desktop)
+				}
+			}
+			if len(desktopMatches) != 1 {
+				plan.Missing = append(plan.Missing, "desktop")
+				for _, desktop := range desktopMatches {
+					plan.Candidates = append(plan.Candidates, assistantCandidate{ID: desktop.ID, Label: desktop.Name + " · VM " + stringID(desktop.VMID), Kind: "desktop"})
+				}
+				if len(desktopMatches) == 0 {
+					for i, desktop := range desktops {
+						if i >= 20 {
+							break
+						}
+						plan.Candidates = append(plan.Candidates, assistantCandidate{ID: desktop.ID, Label: desktop.Name + " · VM " + stringID(desktop.VMID), Kind: "desktop"})
+					}
+				}
+				plan.Title, plan.Summary, plan.Reason = "需要确认桌面", "请写出桌面名称或 VMID，再继续生成方案。", "没有唯一匹配到桌面"
+			} else {
+				desktop := desktopMatches[0]
+				plan.UserID, plan.UserName, plan.DesktopID, plan.DesktopName, plan.DesktopVMID = user.ID, user.DisplayName, desktop.ID, desktop.Name, desktop.VMID
+				plan.Action, plan.Kind, plan.Risk, plan.ConfirmationRequired = map[bool]string{true: "grant_desktop", false: "revoke_desktop"}[grant], "action", "medium", true
+				if grant {
+					plan.Title, plan.Summary = "分配已有桌面", "为 "+user.DisplayName+" 追加“"+desktop.Name+"”使用权限。不会创建 VM、不会自动开机。"
+					var exists bool
+					if err := s.DB.GetContext(ctx, &exists, "SELECT EXISTS(SELECT 1 FROM bindings WHERE user_id=$1 AND desktop_id=$2)", user.ID, desktop.ID); err != nil {
+						s.dbError(c, err)
+						return
+					}
+					if exists {
+						plan.Kind, plan.Reason, plan.Summary = "blocked", "授权已经存在", user.DisplayName+" 已经拥有“"+desktop.Name+"”的使用权限。"
+					}
+				} else {
+					plan.Title, plan.Summary = "撤销桌面授权", "撤销 "+user.DisplayName+" 对“"+desktop.Name+"”的使用权限，不会关机或中断已建立的连接。"
+					var exists bool
+					if err := s.DB.GetContext(ctx, &exists, "SELECT EXISTS(SELECT 1 FROM bindings WHERE user_id=$1 AND desktop_id=$2)", user.ID, desktop.ID); err != nil {
+						s.dbError(c, err)
+						return
+					}
+					if !exists {
+						plan.Kind, plan.Reason, plan.Summary = "blocked", "授权不存在", user.DisplayName+" 当前没有“"+desktop.Name+"”的使用权限。"
+					}
+				}
+			}
+		}
+	}
+	planJSON, _ := json.Marshal(plan)
+	id := store.ID("assist_")
+	status := "planned"
+	if plan.Kind == "blocked" {
+		status = "blocked"
+	}
+	expires := time.Now().UTC().Add(5 * time.Minute)
+	fingerprint := sha256.Sum256(append([]byte(text+"|"+c.MustGet("admin").(model.Administrator).Username+"|"), planJSON...))
+	if _, err := s.DB.ExecContext(c.Request.Context(), `INSERT INTO assistant_runs(id,actor,input_text,action,status,plan,fingerprint,expires_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8)`, id, s.admin(c).Username, text, plan.Action, status, string(planJSON), hex.EncodeToString(fingerprint[:]), expires); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	c.JSON(200, gin.H{"runId": id, "status": status, "input": text, "plan": plan, "createdAt": time.Now().UTC(), "expiresAt": expires})
+}
+
+func stringID(value int) string { return strconv.Itoa(value) }
+
+func (s *Server) assistantExecute(c *gin.Context) {
+	var in struct {
+		RunID string `json:"runId" binding:"required,max=100"`
+	}
+	if !s.decode(c, &in) {
+		return
+	}
+	ctx := c.Request.Context()
+	tx, err := s.DB.BeginTxx(ctx, nil)
+	if err != nil {
+		s.dbError(c, err)
+		return
+	}
+	defer tx.Rollback()
+	var row struct {
+		ID        string    `db:"id"`
+		Input     string    `db:"input_text"`
+		Action    string    `db:"action"`
+		Status    string    `db:"status"`
+		Plan      string    `db:"plan"`
+		ExpiresAt time.Time `db:"expires_at"`
+		Result    []byte    `db:"result"`
+	}
+	if err = tx.GetContext(ctx, &row, "SELECT id,input_text,action,status,plan,expires_at,result FROM assistant_runs WHERE id=$1 AND actor=$2 FOR UPDATE", in.RunID, s.admin(c).Username); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	if row.Status == "executed" {
+		result := map[string]any{}
+		_ = json.Unmarshal(row.Result, &result)
+		c.JSON(200, gin.H{"runId": in.RunID, "status": "executed", "result": result, "idempotent": true})
+		return
+	}
+	if row.Status != "planned" {
+		s.fail(c, 409, "assistant_not_executable", "方案已经执行、阻止或失效")
+		return
+	}
+	if time.Now().UTC().After(row.ExpiresAt) {
+		_, _ = tx.ExecContext(ctx, "UPDATE assistant_runs SET status='expired' WHERE id=$1", in.RunID)
+		s.fail(c, 409, "assistant_plan_expired", "方案已过期，请重新生成")
+		return
+	}
+	var plan assistantPlan
+	if json.Unmarshal([]byte(row.Plan), &plan) != nil || plan.Kind != "action" || (plan.Action != "grant_desktop" && plan.Action != "revoke_desktop") {
+		s.fail(c, 409, "assistant_invalid_plan", "方案内容无效，请重新生成")
+		return
+	}
+	if !s.lockAuthorization(c, tx) {
+		return
+	}
+	var user model.User
+	var desktop model.Desktop
+	if err = tx.GetContext(ctx, &user, "SELECT * FROM users WHERE id=$1 FOR UPDATE", plan.UserID); err != nil {
+		s.fail(c, 409, "assistant_state_changed", "员工已不存在或状态已变化，请重新生成方案")
+		return
+	}
+	if err = tx.GetContext(ctx, &desktop, "SELECT * FROM desktops WHERE id=$1 FOR UPDATE", plan.DesktopID); err != nil {
+		s.fail(c, 409, "assistant_state_changed", "桌面已不存在或状态已变化，请重新生成方案")
+		return
+	}
+	var exists bool
+	if err = tx.GetContext(ctx, &exists, "SELECT EXISTS(SELECT 1 FROM bindings WHERE user_id=$1 AND desktop_id=$2)", user.ID, desktop.ID); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	if plan.Action == "grant" || plan.Action == "grant_desktop" {
+		if exists {
+			s.fail(c, 409, "assistant_state_changed", "授权已经被其他操作创建，请重新生成方案")
+			return
+		}
+		_, err = tx.ExecContext(ctx, "INSERT INTO bindings(user_id,desktop_id) VALUES($1,$2)", user.ID, desktop.ID)
+	} else {
+		if !exists {
+			s.fail(c, 409, "assistant_state_changed", "授权已经被其他操作撤销，请重新生成方案")
+			return
+		}
+		_, err = tx.ExecContext(ctx, "DELETE FROM bindings WHERE user_id=$1 AND desktop_id=$2", user.ID, desktop.ID)
+	}
+	if err == nil {
+		_, err = tx.ExecContext(ctx, "DELETE FROM connection_tickets WHERE consumed_at IS NULL AND terminal_id=(SELECT terminal_id FROM users WHERE id=$1)", user.ID)
+	}
+	if err == nil {
+		err = accountAudit(ctx, tx, s.admin(c).Username, "assistant_"+plan.Action, user.ID, c.GetString("requestId"))
+	}
+	if err != nil {
+		s.dbError(c, err)
+		return
+	}
+	result := map[string]any{"userId": user.ID, "desktopId": desktop.ID, "action": plan.Action, "message": "操作已保存"}
+	encoded, _ := json.Marshal(result)
+	if _, err = tx.ExecContext(ctx, "UPDATE assistant_runs SET status='executed',executed_at=now(),result=$1 WHERE id=$2", string(encoded), in.RunID); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	if err = tx.Commit(); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	c.JSON(200, gin.H{"runId": in.RunID, "status": "executed", "result": result})
+}
+
+func (s *Server) assistantRun(c *gin.Context) {
+	var row struct {
+		ID         string     `db:"id"`
+		Input      string     `db:"input_text"`
+		Action     string     `db:"action"`
+		Status     string     `db:"status"`
+		Plan       string     `db:"plan"`
+		CreatedAt  time.Time  `db:"created_at"`
+		ExpiresAt  time.Time  `db:"expires_at"`
+		ExecutedAt *time.Time `db:"executed_at"`
+		Result     []byte     `db:"result"`
+	}
+	query := "SELECT id,input_text,action,status,plan,created_at,expires_at,executed_at,result FROM assistant_runs WHERE id=$1"
+	args := []any{c.Param("id")}
+	if s.admin(c).Role != "super_admin" {
+		query += " AND actor=$2"
+		args = append(args, s.admin(c).Username)
+	}
+	if err := s.DB.GetContext(c.Request.Context(), &row, query, args...); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	var plan assistantPlan
+	_ = json.Unmarshal([]byte(row.Plan), &plan)
+	result := map[string]any{}
+	_ = json.Unmarshal(row.Result, &result)
+	c.JSON(200, gin.H{"runId": row.ID, "input": row.Input, "action": row.Action, "status": row.Status, "plan": plan, "createdAt": row.CreatedAt, "expiresAt": row.ExpiresAt, "executedAt": row.ExecutedAt, "result": result})
+}
@@ -0,0 +1,44 @@
+package server
+
+import "testing"
+
+func TestAssistantPlanRequiresExplicitDesktopAndExecutesWithRecheck(t *testing.T) {
+	f := newFixtureWorker(t, false)
+	admin := f.login("admin")
+	code, out := f.request("POST", "/assistant/plan", admin, map[string]any{"text": "给王工分配一台桌面"}, "")
+	expect(t, 200, code, out)
+	if out["status"] != "blocked" {
+		t.Fatalf("unexpected plan status: %#v", out)
+	}
+	plan := out["plan"].(map[string]any)
+	if plan["kind"] != "blocked" || plan["reason"] != "没有唯一匹配到桌面" {
+		t.Fatalf("assistant selected a desktop without confirmation: %#v", plan)
+	}
+	code, out = f.request("POST", "/assistant/plan", admin, map[string]any{"text": "给王工分配国产办公桌面"}, "")
+	expect(t, 200, code, out)
+	if out["status"] != "planned" || out["plan"].(map[string]any)["kind"] != "action" {
+		t.Fatalf("unexpected executable plan: %#v", out)
+	}
+	runID := out["runId"].(string)
+	code, out = f.request("POST", "/assistant/execute", admin, map[string]any{"runId": runID}, "assistant-test-1")
+	expect(t, 200, code, out)
+	var exists bool
+	if err := f.db.Get(&exists, `SELECT EXISTS(SELECT 1 FROM bindings b JOIN users u ON u.id=b.user_id JOIN desktops d ON d.id=b.desktop_id WHERE u.display_name='王工' AND d.name='国产办公桌面')`); err != nil || !exists {
+		t.Fatalf("assistant did not create binding: %v", err)
+	}
+	code, out = f.request("POST", "/assistant/execute", admin, map[string]any{"runId": runID}, "assistant-test-2")
+	if code != 200 || out["idempotent"] != true {
+		t.Fatalf("assistant execution was not idempotent: %d %#v", code, out)
+	}
+}
+
+func TestAssistantPlanBlocksAmbiguousUser(t *testing.T) {
+	f := newFixtureWorker(t, false)
+	admin := f.login("admin")
+	code, out := f.request("POST", "/assistant/plan", admin, map[string]any{"text": "请分配研发办公桌面"}, "")
+	expect(t, 200, code, out)
+	plan := out["plan"].(map[string]any)
+	if plan["kind"] != "blocked" || plan["reason"] != "没有唯一匹配到员工" {
+		t.Fatalf("missing employee was not blocked: %#v", plan)
+	}
+}
@@ -48,11 +48,12 @@ func (s *Server) templateList(c *gin.Context) {
 func (s *Server) registerTemplate(c *gin.Context) {
 	pve := s.currentPVE()
 	var in struct {
-		Name    string `json:"name" binding:"required,max=80"`
-		Version string `json:"version" binding:"required,max=40"`
-		OS      string `json:"os" binding:"required,oneof=Windows Ubuntu 麒麟"`
-		Node    string `json:"node" binding:"required,max=100"`
-		VMID    int    `json:"vmid" binding:"min=100,max=999999999"`
+		Name                     string `json:"name" binding:"required,max=80"`
+		Version                  string `json:"version" binding:"required,max=40"`
+		OS                       string `json:"os" binding:"required,oneof=Windows Ubuntu 麒麟"`
+		Node                     string `json:"node" binding:"required,max=100"`
+		VMID                     int    `json:"vmid" binding:"min=100,max=999999999"`
+		HardwareEncodingVerified bool   `json:"hardwareEncodingVerified"`
 	}
 	if !s.decode(c, &in) {
 		return
@@ -82,7 +83,7 @@ func (s *Server) registerTemplate(c *gin.Context) {
 		return
 	}
 	var item model.ImageTemplate
-	err = s.DB.GetContext(c.Request.Context(), &item, `INSERT INTO image_templates(id,name,version,os,node,vmid,digest) VALUES($1,$2,$3,$4,$5,$6,$7) RETURNING *`, store.ID("tpl_"), in.Name, in.Version, in.OS, in.Node, in.VMID, digest)
+	err = s.DB.GetContext(c.Request.Context(), &item, `INSERT INTO image_templates(id,name,version,os,node,vmid,digest,hardware_encoding_verified) VALUES($1,$2,$3,$4,$5,$6,$7,$8) RETURNING *`, store.ID("tpl_"), in.Name, in.Version, in.OS, in.Node, in.VMID, digest, in.HardwareEncodingVerified)
 	if conflict(err) {
 		s.fail(c, 409, "template_conflict", "该模板或名称版本已经登记")
 		return
@@ -96,13 +97,18 @@ func (s *Server) registerTemplate(c *gin.Context) {
 }
 func (s *Server) updateTemplate(c *gin.Context) {
 	var in struct {
-		Enabled *bool `json:"enabled" binding:"required"`
+		Enabled                  *bool `json:"enabled"`
+		HardwareEncodingVerified *bool `json:"hardwareEncodingVerified"`
 	}
 	if !s.decode(c, &in) {
 		return
 	}
+	if in.Enabled == nil && in.HardwareEncodingVerified == nil {
+		s.fail(c, 400, "invalid_request", "至少提供一个模板状态字段")
+		return
+	}
 	var item model.ImageTemplate
-	if err := s.DB.GetContext(c.Request.Context(), &item, "UPDATE image_templates SET enabled=$1 WHERE id=$2 RETURNING *", *in.Enabled, c.Param("id")); err != nil {
+	if err := s.DB.GetContext(c.Request.Context(), &item, `UPDATE image_templates SET enabled=COALESCE($1,enabled), hardware_encoding_verified=COALESCE($2,hardware_encoding_verified) WHERE id=$3 RETURNING *`, in.Enabled, in.HardwareEncodingVerified, c.Param("id")); err != nil {
 		s.dbError(c, err)
 		return
 	}
@@ -303,6 +309,9 @@ func (s *Server) clonePreflight(ctx context.Context, in cloneInput) (model.Image
 	if err := s.DB.GetContext(ctx, &template, "SELECT * FROM image_templates WHERE id=$1 AND enabled", in.TemplateID); err != nil {
 		return template, errors.New("模板不存在或已停用")
 	}
+	if !template.HardwareEncodingVerified {
+		return template, errors.New("模板尚未确认硬件编码，请先在镜像模板中完成硬件编码验收")
+	}
 	config, err := pve.VMConfig(ctx, template.Node, template.VMID)
 	if err != nil {
 		return template, errors.New("无法读取源模板配置")
@@ -97,7 +97,7 @@ func fakeCatalogPVE(t *testing.T, f *fixture, outcome string) *catalogPVE {
 }
 func registerCatalog(t *testing.T, f *fixture, admin string) (string, string) {
 	t.Helper()
-	status, out := f.request("POST", "/templates", admin, map[string]any{"name": "Ubuntu base", "version": "1.0", "os": "Ubuntu", "node": "node-a", "vmid": 8000}, "")
+	status, out := f.request("POST", "/templates", admin, map[string]any{"name": "Ubuntu base", "version": "1.0", "os": "Ubuntu", "node": "node-a", "vmid": 8000, "hardwareEncodingVerified": true}, "")
 	expect(t, 201, status, out)
 	tpl := out["id"].(string)
 	status, out = f.request("POST", "/pools", admin, map[string]any{"name": "Engineering", "type": "dedicated", "maxCount": 1, "templateId": tpl, "enabled": true}, "")
@@ -218,6 +218,36 @@ func TestClonePreflightAndCatalogRBAC(t *testing.T) {
 	status, out = f.request("POST", "/desktops/clone", admin, body, "disabled")
 	expect(t, 409, status, out)
 }
+
+func TestTemplateValidationRequiresHardwareReacceptanceAfterSourceChange(t *testing.T) {
+	f := newFixture(t)
+	p := fakeCatalogPVE(t, f, "success")
+	admin := f.login("admin")
+	tpl, _ := registerCatalog(t, f, admin)
+
+	status, out := f.request("POST", "/templates/"+tpl+"/validate", admin, nil, "")
+	expect(t, 200, status, out)
+	if out["hardwareEncodingVerified"] != true {
+		t.Fatal("unchanged template lost hardware encoding acceptance", out)
+	}
+
+	p.changed.Store(true)
+	status, out = f.request("POST", "/templates/"+tpl+"/validate", admin, nil, "")
+	expect(t, 200, status, out)
+	var digest string
+	if err := f.db.Get(&digest, "SELECT digest FROM image_templates WHERE id=$1", tpl); err != nil {
+		t.Fatal(err)
+	}
+	if out["hardwareEncodingVerified"] != false || digest != "v2" {
+		t.Fatal("changed template remained accepted", out)
+	}
+
+	status, out = f.request("POST", "/desktops/clone/check", admin, cloneBody(tpl, ""), "")
+	expect(t, 200, status, out)
+	if out["available"] != false {
+		t.Fatal("unverified changed template passed clone preflight", out)
+	}
+}
 func TestAmbiguousAndFailedCloneRetainPoolCapacity(t *testing.T) {
 	for _, outcome := range []string{"ambiguous", "failed"} {
 		t.Run(outcome, func(t *testing.T) {
@@ -0,0 +1,121 @@
+package server
+
+import (
+	"context"
+	"database/sql"
+	"net"
+	"strings"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"vdi-cloud/platform/internal/model"
+)
+
+type desktopReadinessCheck struct {
+	Name    string `json:"name"`
+	Status  string `json:"status"`
+	Message string `json:"message"`
+}
+
+type desktopReadiness struct {
+	DesktopID     string                  `json:"desktopId"`
+	ObservedAt    time.Time               `json:"observedAt"`
+	Ready         bool                    `json:"ready"`
+	PowerState    string                  `json:"powerState"`
+	ServiceState  string                  `json:"serviceState"`
+	SetupRequired bool                    `json:"setupRequired"`
+	Checks        []desktopReadinessCheck `json:"checks"`
+}
+
+// checkDesktopReadiness verifies the parts of the provisioning contract that
+// the platform can observe without logging into the guest OS. It never starts
+// or stops a VM and never changes guest configuration.
+func (s *Server) checkDesktopReadiness(c *gin.Context) {
+	ctx := c.Request.Context()
+	var desktop model.Desktop
+	if err := s.DB.GetContext(ctx, &desktop, "SELECT * FROM desktops WHERE id=$1", c.Param("id")); err != nil {
+		if err == sql.ErrNoRows {
+			s.fail(c, 404, "desktop_not_found", "桌面不存在")
+		} else {
+			s.dbError(c, err)
+		}
+		return
+	}
+
+	result := desktopReadiness{
+		DesktopID: desktop.ID, ObservedAt: time.Now().UTC(), Ready: false,
+		PowerState: desktop.PowerState, ServiceState: desktop.ServiceState,
+		SetupRequired: desktop.SetupRequired, Checks: []desktopReadinessCheck{},
+	}
+	add := func(name, status, message string) {
+		result.Checks = append(result.Checks, desktopReadinessCheck{Name: name, Status: status, Message: message})
+	}
+
+	if desktop.Provider != "proxmox" {
+		add("provider", "fail", "桌面不是 Proxmox 资源，无法执行真实连接探测")
+		c.JSON(200, result)
+		return
+	}
+	if desktop.Endpoint == "" {
+		add("endpoint", "fail", "尚未配置桌面 IP 或主机名")
+	} else {
+		add("endpoint", "pass", "已配置桌面连接地址")
+	}
+	if desktop.SetupRequired {
+		add("initialization", "fail", "尚未确认系统身份、硬件编码、Sunshine 和终端配对")
+	} else {
+		add("initialization", "pass", "已完成平台初始化确认")
+	}
+
+	pve := s.currentPVE()
+	if pve == nil {
+		add("power", "unknown", "尚未配置 Proxmox，无法读取真实电源状态")
+	} else {
+		power, err := pve.Power(ctx, desktop.Node, desktop.VMID)
+		if err != nil {
+			result.PowerState = "unknown"
+			add("power", "unknown", "无法读取 Proxmox 实际电源状态")
+		} else {
+			result.PowerState = power
+			if power == "running" {
+				add("power", "pass", "虚拟机正在运行")
+			} else {
+				add("power", "fail", "虚拟机未运行，请先开机")
+			}
+		}
+	}
+
+	serviceReady := false
+	if desktop.Endpoint != "" && result.PowerState == "running" {
+		serviceReady = sunshineReady(ctx, desktop.Endpoint)
+		if serviceReady {
+			result.ServiceState = "ready"
+			add("sunshine", "pass", "Sunshine serverinfo 已响应")
+		} else {
+			result.ServiceState = "unknown"
+			add("sunshine", "fail", "Sunshine 未响应，请检查服务、网络和硬件编码配置")
+		}
+	} else {
+		add("sunshine", "unknown", "虚拟机未运行或尚未配置连接地址，暂不探测 Sunshine")
+	}
+
+	result.Ready = !desktop.SetupRequired && result.PowerState == "running" && serviceReady
+	if _, err := s.DB.ExecContext(ctx, "UPDATE desktops SET power_state=$1,service_state=$2,updated_at=now() WHERE id=$3", result.PowerState, result.ServiceState, desktop.ID); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	c.JSON(200, result)
+}
+
+// sunshineReadyAt is split out so the real probe remains fixed to Sunshine's
+// serverinfo port while tests can use an isolated local listener.
+func sunshineReadyAt(ctx context.Context, endpoint, port string) bool {
+	host := endpoint
+	if parsed, _, err := net.SplitHostPort(endpoint); err == nil {
+		host = parsed
+	}
+	if strings.ContainsAny(host, "/\\@?#") || host == "" {
+		return false
+	}
+	return sunshineHTTPProbe(ctx, host, port)
+}
@@ -0,0 +1,38 @@
+package server
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+)
+
+func TestSunshineReadyAt(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/serverinfo" {
+			t.Fatalf("unexpected path %s", r.URL.Path)
+		}
+		w.Header().Set("Content-Type", "application/xml")
+		_, _ = w.Write([]byte(`<root status_code="200"><hostname>desktop-test</hostname></root>`))
+	}))
+	t.Cleanup(server.Close)
+	host, port, ok := splitTestEndpoint(t, server.URL)
+	if !ok || !sunshineReadyAt(context.Background(), host, port) {
+		t.Fatal("expected Sunshine probe to pass")
+	}
+	if sunshineReadyAt(context.Background(), host, "1") {
+		t.Fatal("unexpected Sunshine probe success on wrong port")
+	}
+}
+
+func splitTestEndpoint(t *testing.T, raw string) (string, string, bool) {
+	t.Helper()
+	parsed, err := url.Parse(raw)
+	if err != nil {
+		return "", "", false
+	}
+	host, port, err := net.SplitHostPort(parsed.Host)
+	return host, port, err == nil && host != "" && port != ""
+}
@@ -63,6 +63,7 @@ type terminalLogArchive struct {
 	AgentVersion string    `json:"agentVersion" db:"agent_version"`
 	KioskState   string    `json:"kioskState" db:"kiosk_state"`
 	LogText      string    `json:"logText" db:"log_text"`
+	Source       string    `json:"source" db:"source"`
 }
 type terminalLogArchivePage struct {
 	Items         []terminalLogArchive `json:"items"`
@@ -87,13 +88,19 @@ func (s *Server) terminalLogArchives(c *gin.Context) {
 	page, size := pageParams(c)
 	args := []any{from, to, terminalID, keyword}
 	var total int
-	if err := s.DB.GetContext(c.Request.Context(), &total, `SELECT count(*) FROM terminal_log_archives WHERE captured_at>=$1 AND captured_at<=$2 AND ($3='' OR terminal_id=$3) AND ($4='' OR log_text ILIKE '%'||$4||'%')`, args...); err != nil {
+	if err := s.DB.GetContext(c.Request.Context(), &total, `SELECT count(*) FROM (
+ SELECT captured_at,terminal_id,log_text FROM terminal_log_archives
+ UNION ALL SELECT captured_at,terminal_id,message FROM terminal_log_entries
+ ) logs WHERE captured_at>=$1 AND captured_at<=$2 AND ($3='' OR terminal_id=$3) AND ($4='' OR log_text ILIKE '%'||$4||'%')`, args...); err != nil {
 		s.dbError(c, err)
 		return
 	}
 	items := []terminalLogArchive{}
 	args = append(args, size, (page-1)*size)
-	if err := s.DB.SelectContext(c.Request.Context(), &items, `SELECT a.id,a.task_id,a.terminal_id,t.name terminal_name,a.captured_at,a.boot_id,a.agent_version,a.kiosk_state,a.log_text FROM terminal_log_archives a JOIN terminals t ON t.id=a.terminal_id WHERE a.captured_at>=$1 AND a.captured_at<=$2 AND ($3='' OR a.terminal_id=$3) AND ($4='' OR a.log_text ILIKE '%'||$4||'%') ORDER BY a.captured_at DESC,a.id DESC LIMIT $5 OFFSET $6`, args...); err != nil {
+	if err := s.DB.SelectContext(c.Request.Context(), &items, `SELECT logs.id,logs.task_id,logs.terminal_id,t.name terminal_name,logs.captured_at,logs.boot_id,logs.agent_version,logs.kiosk_state,logs.log_text,logs.source FROM (
+ SELECT a.id,a.task_id,a.terminal_id,a.captured_at,a.boot_id,a.agent_version,a.kiosk_state,a.log_text,'snapshot' source FROM terminal_log_archives a
+ UNION ALL SELECT e.id,'' task_id,e.terminal_id,e.captured_at,e.boot_id,'' agent_version,e.level kiosk_state,e.message log_text,'stream' source FROM terminal_log_entries e
+ ) logs JOIN terminals t ON t.id=logs.terminal_id WHERE logs.captured_at>=$1 AND logs.captured_at<=$2 AND ($3='' OR logs.terminal_id=$3) AND ($4='' OR logs.log_text ILIKE '%'||$4||'%') ORDER BY logs.captured_at DESC,logs.id DESC LIMIT $5 OFFSET $6`, args...); err != nil {
 		s.dbError(c, err)
 		return
 	}
@@ -291,8 +291,12 @@ func (s *Server) prepareGPUDesktop(c *gin.Context) {
 	var previous model.Task
 	err = tx.GetContext(ctx, &previous, `SELECT * FROM tasks WHERE kind='gpu-prepare' AND status IN ('queued','running','timeout') ORDER BY created_at DESC LIMIT 1`)
 	if err == nil {
-		if previous.Actor == "device:"+terminal && previous.TargetID == d.ID && previous.Status != "timeout" {
-			s.deviceJSON(c, 202, gin.H{"taskId": previous.ID, "status": previous.Status, "message": previous.Message})
+		if previous.Actor == "device:"+terminal && previous.TargetID == d.ID && (previous.Status != "timeout" || s.gpuRecoveryPending(previous)) {
+			status := previous.Status
+			if status == "timeout" {
+				status = "running"
+			}
+			s.deviceJSON(c, 202, gin.H{"taskId": previous.ID, "status": status, "message": previous.Message})
 			return
 		}
 		s.fail(c, 409, "gpu_switch_busy", "已有显卡切换任务或未确认结果，请稍后重试或联系管理员")
@@ -343,7 +347,11 @@ func (s *Server) gpuPrepareStatus(c *gin.Context) {
 		s.dbError(c, err)
 		return
 	}
-	s.deviceJSON(c, 200, gin.H{"taskId": task.ID, "status": task.Status, "message": task.Message})
+	status := task.Status
+	if s.gpuRecoveryPending(task) {
+		status = "running"
+	}
+	s.deviceJSON(c, 200, gin.H{"taskId": task.ID, "status": status, "message": task.Message})
 }

 func (s *Server) saveGPUStage(ctx context.Context, t model.Task, state gpuPrepareState, message string) error {
@@ -359,15 +367,11 @@ func (s *Server) saveGPUStage(ctx context.Context, t model.Task, state gpuPrepar
 }

 func sunshineReady(ctx context.Context, endpoint string) bool {
-	host := endpoint
-	if parsed, _, err := net.SplitHostPort(endpoint); err == nil {
-		host = parsed
-	}
-	// Platform endpoints are hostnames/IPs, never arbitrary URLs or paths.
-	if strings.ContainsAny(host, "/\\@?#") || host == "" {
-		return false
-	}
-	req, err := http.NewRequestWithContext(ctx, "GET", "http://"+net.JoinHostPort(host, "47989")+"/serverinfo", nil)
+	return sunshineReadyAt(ctx, endpoint, "47989")
+}
+
+func sunshineHTTPProbe(ctx context.Context, host, port string) bool {
+	req, err := http.NewRequestWithContext(ctx, "GET", "http://"+net.JoinHostPort(host, port)+"/serverinfo", nil)
 	if err != nil {
 		return false
 	}
@@ -532,6 +536,7 @@ func (s *Server) executeGPUPrepare(parent context.Context, t model.Task) {
 				expected = "running"
 			}
 			if e != nil || power != expected {
+				uncertain = true
 				err = errors.New("final GPU ownership check failed")
 				break
 			}
@@ -546,9 +551,12 @@ func (s *Server) executeGPUPrepare(parent context.Context, t model.Task) {
 		log.Printf("GPU prepare task=%s phase=%s error=%v", t.ID, state.Phase, err)
 		status = "failed"
 		message = "桌面准备失败，请检查桌面授权、占用状态及 PVE"
-		if uncertain {
+		if uncertain || (state.Phase == "service-waiting" && s.knownGPURecovery(state)) {
 			status = "timeout"
-			message = "电源操作或服务状态尚未确认，已阻止再次切换；请管理员核对 PVE 原任务"
+			message = "电源操作尚未确认，平台正在自动核对原任务；不会重复发送电源命令"
+			if state.UPID == "" {
+				message = "未收到 PVE 原任务编号，请管理员核查；不会重复发送电源命令"
+			}
 		}
 	}
 	finalCtx, finish := context.WithTimeout(context.Background(), 5*time.Second)
@@ -17,11 +17,12 @@ import (
 )

 type exclusiveTestPVE struct {
-	mu        sync.Mutex
-	power     map[int]string
-	actions   []string
-	mode      string
-	confirmed bool
+	mu           sync.Mutex
+	power        map[int]string
+	actions      []string
+	mode         string
+	confirmed    bool
+	readFailures int
 }

 func exclusiveFixture(t *testing.T, mode string) (*fixture, *exclusiveTestPVE) {
@@ -49,7 +50,10 @@ func exclusiveFixture(t *testing.T, mode string) (*fixture, *exclusiveTestPVE) {
 		var id int
 		switch {
 		case strings.Contains(path, "/tasks/"):
-			if p.mode == "poll-failure" {
+			if p.mode == "poll-failure" || p.readFailures > 0 {
+				if p.readFailures > 0 {
+					p.readFailures--
+				}
 				w.WriteHeader(503)
 				return
 			}
@@ -62,7 +66,11 @@ func exclusiveFixture(t *testing.T, mode string) (*fixture, *exclusiveTestPVE) {
 					p.power[102] = "running"
 				}
 			}
-			data = map[string]string{"status": "stopped", "exitstatus": "OK"}
+			exitStatus := "OK"
+			if p.mode == "failed-task" {
+				exitStatus = "ERROR"
+			}
+			data = map[string]string{"status": "stopped", "exitstatus": exitStatus}
 		case strings.HasSuffix(path, "/status/current"):
 			fmt.Sscanf(path, "/nodes/gpu-node/qemu/%d/status/current", &id)
 			data = map[string]string{"status": p.power[id]}
@@ -202,7 +210,14 @@ func TestExclusiveGPUUnknownSubmissionNeverReplayed(t *testing.T) {
 				t.Fatal("replayed uncertain power action", p.actions)
 			}
 			code, out := f.request("POST", "/terminal/desktop-prepare", "gpu-device", map[string]any{"desktopId": "desk_demo_ubuntu"}, "")
-			expect(t, 409, code, out)
+			if mode == "poll-failure" {
+				expect(t, 202, code, out)
+				if out["taskId"] != task.ID || out["status"] != "running" {
+					t.Fatal("recovering task was not reused", out)
+				}
+			} else {
+				expect(t, 409, code, out)
+			}
 			code, out = f.request("POST", "/tasks/"+task.ID+"/gpu-reconcile", f.login("admin"), nil, "")
 			if mode == "ambiguous" {
 				expect(t, 409, code, out)
@@ -103,6 +103,11 @@ func (s *Server) gpuInventory(c *gin.Context) {
 		s.fail(c, 502, "pci_unavailable", "无法读取节点 PCI 清单，请检查 PVE 连通性及根路径 Sys.Audit 权限")
 		return
 	}
+	healthSamples, err := s.latestGPUHealth(ctx, node)
+	if err != nil {
+		s.dbError(c, err)
+		return
+	}
 	items := []gpuItem{}
 	observedAt := time.Now().UTC()
 	for _, d := range devices {
@@ -122,7 +127,23 @@ func (s *Server) gpuInventory(c *gin.Context) {
 		if strings.TrimSpace(d.Driver) != "" {
 			driverBinding = "bound"
 		}
-		items = append(items, gpuItem{ID: node + ":" + address, Address: address, Name: d.DeviceName, Vendor: d.VendorName, VendorID: d.Vendor, DeviceID: d.Device, Class: d.Class, IOMMUGroup: group, Health: gpuHealth{Status: "not_collected", Source: "pve-pci", ObservedAt: observedAt, Driver: d.Driver, DriverBinding: driverBinding}, References: []gpuReference{}})
+		health := gpuHealth{Status: "not_collected", Source: "pve-pci", ObservedAt: observedAt, Driver: d.Driver, DriverBinding: driverBinding}
+		if sample, ok := healthSamples[address]; ok {
+			health.Status = "collected"
+			health.Source = sample.Source
+			health.ObservedAt = sample.ObservedAt
+			health.TemperatureC = sample.TemperatureC
+			health.UtilizationPercent = sample.UtilizationPercent
+			health.MemoryUsedBytes = sample.MemoryUsedBytes
+			health.MemoryTotalBytes = sample.MemoryTotalBytes
+			if sample.Driver != "" {
+				health.Driver = sample.Driver
+			}
+			if sample.DriverBinding != "" {
+				health.DriverBinding = sample.DriverBinding
+			}
+		}
+		items = append(items, gpuItem{ID: node + ":" + address, Address: address, Name: d.DeviceName, Vendor: d.VendorName, VendorID: d.Vendor, DeviceID: d.Device, Class: d.Class, IOMMUGroup: group, Health: health, References: []gpuReference{}})
 	}
 	fullVisibility, err := pve.HasPermission(ctx, "/", "VM.Audit")
 	if err != nil || !fullVisibility {
@@ -244,7 +265,13 @@ func (s *Server) gpuInventory(c *gin.Context) {
 		}
 	}
 	sort.Slice(items, func(i, j int) bool { return items[i].Address < items[j].Address })
-	c.JSON(200, gin.H{"node": node, "observedAt": observedAt, "complete": len(issues) == 0, "visibleVmCount": visibleCount, "scannedVmCount": scannedCount, "items": items, "issues": issues, "health": gin.H{"status": "partial", "source": "pve-pci", "temperature": "not_collected", "utilization": "not_collected", "memory": "not_collected", "driverBinding": "reported_when_pve_exposes_driver"}})
+	healthStatus := "not_collected"
+	healthSource := "pve-pci"
+	if len(healthSamples) > 0 {
+		healthStatus = "collected"
+		healthSource = "node-agent"
+	}
+	c.JSON(200, gin.H{"node": node, "observedAt": observedAt, "complete": len(issues) == 0, "visibleVmCount": visibleCount, "scannedVmCount": scannedCount, "items": items, "issues": issues, "health": gin.H{"status": healthStatus, "source": healthSource, "temperature": healthStatus, "utilization": healthStatus, "memory": healthStatus, "driverBinding": "reported_when_collected"}})
 }

 func rawOptionText(value json.RawMessage) string {
@@ -0,0 +1,141 @@
+package server
+
+import (
+	"context"
+	"crypto/subtle"
+	"strings"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"vdi-cloud/platform/pkg/proxmox"
+)
+
+// gpuAgentSample is the small, vendor-neutral contract implemented by a node
+// collector. The collector may use nvidia-smi, rocm-smi, intel_gpu_top or a
+// vendor API, while the platform keeps the same health model for every GPU.
+type gpuAgentSample struct {
+	Address            string   `json:"address" binding:"required"`
+	ObservedAt         string   `json:"observedAt"`
+	TemperatureC       *float64 `json:"temperatureC"`
+	UtilizationPercent *float64 `json:"utilizationPercent"`
+	MemoryUsedBytes    *int64   `json:"memoryUsedBytes"`
+	MemoryTotalBytes   *int64   `json:"memoryTotalBytes"`
+	Driver             string   `json:"driver"`
+	DriverBinding      string   `json:"driverBinding"`
+}
+
+type gpuAgentReport struct {
+	Node    string           `json:"node" binding:"required,max=100"`
+	Samples []gpuAgentSample `json:"samples" binding:"required,min=1,max=256"`
+}
+
+func (s *Server) gpuAgentAuth(c *gin.Context) {
+	if strings.TrimSpace(s.Config.GPUAgentToken) == "" {
+		s.fail(c, 503, "gpu_agent_unconfigured", "尚未配置 GPU 节点采集令牌")
+		return
+	}
+	provided := c.GetHeader("X-GPU-Agent-Token")
+	if len(provided) != len(s.Config.GPUAgentToken) || subtle.ConstantTimeCompare([]byte(provided), []byte(s.Config.GPUAgentToken)) != 1 {
+		s.fail(c, 401, "gpu_agent_unauthorized", "GPU 节点采集令牌无效")
+		return
+	}
+	c.Next()
+}
+
+func validGPUHealth(sample gpuAgentSample) bool {
+	if sample.TemperatureC != nil && (*sample.TemperatureC < -40 || *sample.TemperatureC > 150) {
+		return false
+	}
+	if sample.UtilizationPercent != nil && (*sample.UtilizationPercent < 0 || *sample.UtilizationPercent > 100) {
+		return false
+	}
+	if sample.MemoryUsedBytes != nil && *sample.MemoryUsedBytes < 0 {
+		return false
+	}
+	if sample.MemoryTotalBytes != nil && *sample.MemoryTotalBytes < 0 {
+		return false
+	}
+	if sample.MemoryUsedBytes != nil && sample.MemoryTotalBytes != nil && *sample.MemoryUsedBytes > *sample.MemoryTotalBytes {
+		return false
+	}
+	_, ok := proxmox.NormalizePCIAddress(sample.Address)
+	return ok
+}
+
+func (s *Server) ingestGPUHealth(c *gin.Context) {
+	var in gpuAgentReport
+	if !s.decode(c, &in) {
+		return
+	}
+	in.Node = strings.TrimSpace(in.Node)
+	if in.Node == "" || strings.ContainsAny(in.Node, "/\\?# ") {
+		s.fail(c, 400, "invalid_node", "节点名称无效")
+		return
+	}
+	ctx := c.Request.Context()
+	tx, err := s.DB.BeginTxx(ctx, nil)
+	if err != nil {
+		s.dbError(c, err)
+		return
+	}
+	defer tx.Rollback()
+	for _, sample := range in.Samples {
+		if !validGPUHealth(sample) {
+			s.fail(c, 400, "invalid_gpu_health", "GPU 健康数据超出范围或 PCI 地址无效")
+			return
+		}
+		address, _ := proxmox.NormalizePCIAddress(sample.Address)
+		observedAt := time.Now().UTC()
+		if strings.TrimSpace(sample.ObservedAt) != "" {
+			observedAt, err = time.Parse(time.RFC3339Nano, sample.ObservedAt)
+			if err != nil || observedAt.After(time.Now().Add(2*time.Minute)) || observedAt.Before(time.Now().Add(-10*time.Minute)) {
+				s.fail(c, 400, "invalid_gpu_health", "采集时间必须是最近 10 分钟内的 RFC3339 时间")
+				return
+			}
+			observedAt = observedAt.UTC()
+		}
+		_, err = tx.ExecContext(ctx, `INSERT INTO gpu_health_samples(node,address,observed_at,temperature_c,utilization_percent,memory_used_bytes,memory_total_bytes,driver,driver_binding,source)
+VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,'node-agent') ON CONFLICT DO NOTHING`, in.Node, address, observedAt, sample.TemperatureC, sample.UtilizationPercent, sample.MemoryUsedBytes, sample.MemoryTotalBytes, strings.TrimSpace(sample.Driver), strings.TrimSpace(sample.DriverBinding))
+		if err != nil {
+			s.dbError(c, err)
+			return
+		}
+	}
+	if _, err = tx.ExecContext(ctx, "DELETE FROM gpu_health_samples WHERE observed_at < now() - interval '7 days'"); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	if err = tx.Commit(); err != nil {
+		s.dbError(c, err)
+		return
+	}
+	c.JSON(202, gin.H{"accepted": len(in.Samples), "node": in.Node})
+}
+
+type latestGPUHealth struct {
+	Address            string    `db:"address"`
+	ObservedAt         time.Time `db:"observed_at"`
+	TemperatureC       *float64  `db:"temperature_c"`
+	UtilizationPercent *float64  `db:"utilization_percent"`
+	MemoryUsedBytes    *int64    `db:"memory_used_bytes"`
+	MemoryTotalBytes   *int64    `db:"memory_total_bytes"`
+	Driver             string    `db:"driver"`
+	DriverBinding      string    `db:"driver_binding"`
+	Source             string    `db:"source"`
+}
+
+func (s *Server) latestGPUHealth(ctx context.Context, node string) (map[string]latestGPUHealth, error) {
+	// Kept separate from inventory assembly so a missing collector is an
+	// ordinary empty result and remains visibly "not_collected" in the API.
+	items := []latestGPUHealth{}
+	query := `SELECT DISTINCT ON(address) address,observed_at,temperature_c,utilization_percent,memory_used_bytes,memory_total_bytes,driver,driver_binding,source
+FROM gpu_health_samples WHERE node=$1 AND observed_at > now()-interval '10 minutes' ORDER BY address,observed_at DESC`
+	if err := s.DB.SelectContext(ctx, &items, query, node); err != nil {
+		return nil, err
+	}
+	result := make(map[string]latestGPUHealth, len(items))
+	for _, item := range items {
+		result[item.Address] = item
+	}
+	return result, nil
+}
@@ -0,0 +1,50 @@
+package server
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestGPUAgentHealthIsAuthenticatedAndVisibleInInventory(t *testing.T) {
+	f := newFixture(t)
+	f.server.Config.GPUAgentToken = "agent-token"
+	fakeGPUPVE(t, f, "normal")
+
+	status, out := f.request("POST", "/gpu-agent/health", "", map[string]any{
+		"node": "node-a",
+		"samples": []map[string]any{{
+			"address": "0000:01:00.0", "temperatureC": 61.5, "utilizationPercent": 42.0,
+			"memoryUsedBytes": 1024, "memoryTotalBytes": 4096, "driver": "nvidia", "driverBinding": "bound",
+		}},
+	}, "")
+	// f.request cannot attach the collector header, so the route must reject it.
+	expect(t, 401, status, out)
+
+	body, _ := json.Marshal(map[string]any{
+		"node": "node-a", "samples": []map[string]any{{"address": "0000:01:00.0", "temperatureC": 61.5, "driver": "nvidia", "driverBinding": "bound"}},
+	})
+	r := httptest.NewRequest(http.MethodPost, "/api/v1/gpu-agent/health", bytes.NewReader(body))
+	r.Header.Set("Content-Type", "application/json")
+	r.Header.Set("X-GPU-Agent-Token", "agent-token")
+	w := httptest.NewRecorder()
+	f.server.Router.ServeHTTP(w, r)
+	expect(t, 202, w.Code, w.Body.String())
+
+	status, out = f.request("GET", "/resources/gpus?node=node-a", f.login("admin"), nil, "")
+	expect(t, 200, status, out)
+	items := out["items"].([]any)
+	var found bool
+	for _, raw := range items {
+		item := raw.(map[string]any)
+		if item["address"] == "0000:01:00.0" {
+			health := item["health"].(map[string]any)
+			found = health["status"] == "collected" && health["driverBinding"] == "bound" && health["temperatureC"] == 61.5 && health["driver"] == "nvidia"
+		}
+	}
+	if !found {
+		t.Fatal("ingested GPU health was not visible in inventory", out)
+	}
+}
@@ -0,0 +1,145 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"vdi-cloud/platform/internal/model"
+	"vdi-cloud/platform/internal/store"
+)
+
+func (s *Server) knownGPURecovery(state gpuPrepareState) bool {
+	return state.UPID != "" && state.Node == s.Config.ExclusiveGPUNode &&
+		fmt.Sprint(state.Members) == fmt.Sprint(s.Config.ExclusiveGPUVMIDs) &&
+		(state.Phase == "start-waiting" || state.Phase == "shutdown-waiting" || state.Phase == "service-waiting")
+}
+
+func (s *Server) gpuRecoveryPending(task model.Task) bool {
+	var state gpuPrepareState
+	return task.Status == "timeout" && time.Since(task.CreatedAt) < 5*time.Minute &&
+		json.Unmarshal([]byte(task.ProviderRef), &state) == nil && s.knownGPURecovery(state)
+}
+
+// Recovery observes the original UPID and every member under the same group
+// lock as preparation. It never sends a power command, even after an API
+// restart or a lost submission response.
+func (s *Server) recoverGPUPrepare(parent context.Context, task model.Task) {
+	ctx, cancel := context.WithTimeout(parent, 60*time.Second)
+	defer cancel()
+	if s.RefreshCompute(ctx) != nil {
+		return
+	}
+	pve := s.currentPVE()
+	if pve == nil {
+		return
+	}
+	tx, err := s.DB.BeginTxx(ctx, nil)
+	if err != nil {
+		return
+	}
+	defer tx.Rollback()
+	locked, err := tryGPULock(ctx, tx)
+	if err != nil || !locked {
+		return
+	}
+	if err = tx.GetContext(ctx, &task, "SELECT * FROM tasks WHERE id=$1 AND kind='gpu-prepare' AND status IN ('running','timeout') FOR UPDATE", task.ID); err != nil {
+		return
+	}
+	finish := func(status, message string, power map[string]string) {
+		for id, value := range power {
+			if _, err = tx.ExecContext(ctx, "UPDATE desktops SET power_state=$1,service_state=CASE WHEN $2=$3 AND $4='success' THEN 'ready' ELSE 'unknown' END,updated_at=now() WHERE id=$2", value, id, task.TargetID, status); err != nil {
+				return
+			}
+		}
+		if _, err = tx.ExecContext(ctx, "UPDATE tasks SET status=$1,message=$2,updated_at=now() WHERE id=$3", status, message, task.ID); err != nil {
+			return
+		}
+		if status != "timeout" {
+			if _, err = tx.ExecContext(ctx, "INSERT INTO audit_logs VALUES($1,'system','auto-reconcile-gpu-switch',$2,$3,$4,now())", store.ID("audit_"), task.TargetID, status, task.ID); err != nil {
+				return
+			}
+		}
+		_ = tx.Commit()
+	}
+	var state gpuPrepareState
+	if json.Unmarshal([]byte(task.ProviderRef), &state) == nil && state.Phase == "queued" && state.UPID == "" {
+		finish("failed", "服务重启前尚未提交电源操作，可重新选择桌面", nil)
+		return
+	}
+	if json.Unmarshal([]byte(task.ProviderRef), &state) != nil || !s.knownGPURecovery(state) {
+		finish("timeout", "未记录可核对的原任务或独占组配置已变化，请管理员核查；不会重复发送电源命令", nil)
+		return
+	}
+	original, err := pve.Task(ctx, state.Node, state.UPID)
+	if err != nil || original.Status != "stopped" || original.ExitStatus == "" {
+		finish("timeout", "平台正在自动核对原 PVE 任务，暂不重复切换；计算节点恢复后将继续检查", nil)
+		return
+	}
+	members, err := s.gpuMembers(ctx, tx)
+	if err != nil {
+		finish("timeout", "独占显卡组成员不完整，请管理员核查", nil)
+		return
+	}
+	powers := map[string]string{}
+	running, matches, submittedMember := 0, true, false
+	var target model.Desktop
+	for _, d := range members {
+		power, readErr := pve.Power(ctx, d.Node, d.VMID)
+		if readErr != nil || (power != "running" && power != "stopped") {
+			finish("timeout", "平台正在自动核对虚拟机实际电源状态，暂不重复切换", nil)
+			return
+		}
+		powers[d.ID] = power
+		if d.VMID == state.VMID {
+			submittedMember = true
+		}
+		if power == "running" {
+			running++
+		}
+		if d.ID == task.TargetID {
+			target = d
+			matches = matches && power == "running"
+		} else {
+			matches = matches && power == "stopped"
+		}
+	}
+	if running > 1 || target.ID == "" || !submittedMember {
+		finish("timeout", "实际电源或显卡归属存在冲突，请管理员核查；平台不会自动关机", powers)
+		return
+	}
+	if original.ExitStatus != "OK" || !matches {
+		finish("failed", "已自动核对原任务和电源状态，本次准备未完成；可重新选择桌面", powers)
+		return
+	}
+	var allowed bool
+	err = tx.GetContext(ctx, &allowed, `SELECT EXISTS(SELECT 1 FROM bindings b JOIN users u ON u.id=b.user_id JOIN terminals t ON t.id=u.terminal_id WHERE b.desktop_id=$1 AND t.id=$2 AND u.enabled AND t.enabled)`, target.ID, state.Terminal)
+	if err != nil {
+		return
+	}
+	if !allowed || target.SetupRequired || target.Endpoint == "" {
+		finish("failed", "原任务已核对，但桌面授权或连接配置已变化，请检查配置后重新选择", powers)
+		return
+	}
+	if !sunshineReady(ctx, target.Endpoint) {
+		if time.Since(task.CreatedAt) < 5*time.Minute {
+			finish("timeout", "原任务已确认成功，正在等待桌面连接服务恢复", powers)
+		} else {
+			finish("failed", "原任务及显卡归属已自动确认，但 Sunshine 仍未响应，请检查桌面服务", powers)
+		}
+		return
+	}
+	finish("success", "已自动核对原任务、独占显卡归属和连接服务，桌面可连接", powers)
+}
+
+// Only called by the worker while no GPU/provisioning operation is running.
+// A restart resumes observations, never the command sequence.
+func (s *Server) nextGPURecovery(ctx context.Context) (model.Task, error) {
+	var task model.Task
+	err := s.DB.GetContext(ctx, &task, `SELECT * FROM tasks WHERE kind='gpu-prepare'
+AND ((status='running')
+ OR (status='timeout' AND provider_ref::jsonb->>'upid' IS NOT NULL AND provider_ref::jsonb->>'upid' != '' AND created_at>now()-interval '24 hours'))
+ORDER BY updated_at,id LIMIT 1`)
+	return task, err
+}
@@ -0,0 +1,161 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"vdi-cloud/platform/internal/model"
+)
+
+func serveRecoverySunshine(t *testing.T) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:47989")
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, `<root status_code="200"><hostname>test-sunshine</hostname></root>`)
+	})}
+	go srv.Serve(listener)
+	t.Cleanup(func() { srv.Close() })
+}
+
+func seedGPURecovery(t *testing.T, f *fixture, status string) model.Task {
+	t.Helper()
+	task := prepareExclusive(t, f)
+	state := gpuPrepareState{Phase: "start-waiting", Node: "gpu-node", Members: []int{100, 101, 102}, Terminal: "gpu-box", VMID: 102, UPID: "UPID:gpu-test"}
+	data, _ := json.Marshal(state)
+	if err := f.db.Get(&task, "UPDATE tasks SET status=$1,provider_ref=$2 WHERE id=$3 RETURNING *", status, string(data), task.ID); err != nil {
+		t.Fatal(err)
+	}
+	return task
+}
+
+func TestGPURecoveryRequiresActualOwnershipAndSunshine(t *testing.T) {
+	for _, scenario := range []string{"success", "original-failed", "pending-shutdown", "multiple-running", "pve-unreachable", "sunshine-unreachable", "authorization-revoked", "config-changed", "unknown-submission"} {
+		t.Run(scenario, func(t *testing.T) {
+			f, p := exclusiveFixture(t, "normal")
+			task := seedGPURecovery(t, f, "timeout")
+			p.power[100], p.power[102] = "stopped", "running"
+			want := "timeout"
+			switch scenario {
+			case "success":
+				serveRecoverySunshine(t)
+				want = "success"
+			case "original-failed":
+				p.mode = "failed-task"
+				want = "failed"
+			case "pending-shutdown":
+				p.power[102] = "stopped"
+				want = "failed"
+			case "multiple-running":
+				p.power[100] = "running"
+			case "pve-unreachable":
+				p.mode = "poll-failure"
+			case "authorization-revoked":
+				if _, err := f.db.Exec("UPDATE users SET enabled=false"); err != nil {
+					t.Fatal(err)
+				}
+				want = "failed"
+			case "config-changed":
+				f.server.Config.ExclusiveGPUVMIDs = []int{100, 102}
+			case "unknown-submission":
+				if _, err := f.db.Exec("UPDATE tasks SET provider_ref=$1 WHERE id=$2", `{"phase":"start-submitting","node":"gpu-node","members":[100,101,102],"terminal":"gpu-box","vmid":102}`, task.ID); err != nil {
+					t.Fatal(err)
+				}
+			}
+			f.server.recoverGPUPrepare(context.Background(), task)
+			var result model.Task
+			if err := f.db.Get(&result, "SELECT * FROM tasks WHERE id=$1", task.ID); err != nil {
+				t.Fatal(err)
+			}
+			if result.Status != want || len(p.actions) != 0 {
+				t.Fatal("incorrect read-only recovery", result.Status, want, p.actions)
+			}
+			if want == "success" {
+				var power, service string
+				f.db.QueryRow("SELECT power_state,service_state FROM desktops WHERE id=$1", task.TargetID).Scan(&power, &service)
+				if power != "running" || service != "ready" {
+					t.Fatal("desktop cache did not converge", power, service)
+				}
+				f.server.recoverGPUPrepare(context.Background(), task)
+				var n int
+				f.db.Get(&n, "SELECT count(*) FROM audit_logs WHERE action='auto-reconcile-gpu-switch' AND request_id=$1", task.ID)
+				if n != 1 {
+					t.Fatal("recovery audit was duplicated", n)
+				}
+			}
+		})
+	}
+}
+
+func TestGPUWorkerRecoversOriginalTaskAfterRestartWithoutCommands(t *testing.T) {
+	f, p := exclusiveFixture(t, "normal")
+	task := seedGPURecovery(t, f, "running")
+	p.power[100], p.power[102] = "stopped", "running"
+	serveRecoverySunshine(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { f.server.RunWorker(ctx); close(done) }()
+	defer func() { cancel(); <-done }()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		var status string
+		if err := f.db.Get(&status, "SELECT status FROM tasks WHERE id=$1", task.ID); err != nil {
+			t.Fatal(err)
+		}
+		if status == "success" {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("worker did not recover orphaned task", status)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if len(p.actions) != 0 {
+		t.Fatal("restart replayed a power command", p.actions)
+	}
+}
+
+func TestGPUTransientReadFailureKeepsSingleCommandSequence(t *testing.T) {
+	f, p := exclusiveFixture(t, "normal")
+	p.readFailures = 1
+	serveRecoverySunshine(t)
+	task := prepareExclusive(t, f)
+	f.server.executeGPUPrepare(context.Background(), task)
+	var status string
+	f.db.Get(&status, "SELECT status FROM tasks WHERE id=$1", task.ID)
+	if status != "success" || fmt.Sprint(p.actions) != "[100:shutdown 102:start]" {
+		t.Fatal("transient poll failure interrupted or replayed commands", status, p.actions)
+	}
+}
+
+func TestGPURecoveryPollingRetainsTaskAndDeviceIsolation(t *testing.T) {
+	f, _ := exclusiveFixture(t, "normal")
+	task := seedGPURecovery(t, f, "timeout")
+	code, out := f.request("GET", "/terminal/desktop-prepare/"+task.ID, "gpu-device", nil, "")
+	expect(t, 200, code, out)
+	if out["status"] != "running" {
+		t.Fatal("terminal stopped polling recoverable task", out)
+	}
+	code, out = f.request("POST", "/terminal/desktop-prepare", "gpu-device", map[string]any{"desktopId": task.TargetID}, "")
+	expect(t, 202, code, out)
+	if out["taskId"] != task.ID {
+		t.Fatal("new task submitted during recovery", out)
+	}
+	if _, err := f.db.Exec("UPDATE tasks SET created_at=now()-interval '6 minutes' WHERE id=$1", task.ID); err != nil {
+		t.Fatal(err)
+	}
+	code, out = f.request("GET", "/terminal/desktop-prepare/"+task.ID, "gpu-device", nil, "")
+	expect(t, 200, code, out)
+	if out["status"] != "timeout" {
+		t.Fatal("terminal waiting exceeded recovery window", out)
+	}
+}
@@ -299,7 +299,7 @@ func (s *Server) observe(ctx context.Context) error {
 		}
 	}
 	// Bounded retention. Current incidents and active session records are preserved.
-	for _, query := range []string{"DELETE FROM login_limits WHERE window_started<now()-interval '1 day'", "DELETE FROM resource_samples WHERE sampled_at<now()-interval '30 days'", "DELETE FROM terminal_metric_samples WHERE sampled_at<now()-interval '30 days'", "DELETE FROM terminal_log_archives WHERE captured_at<now()-interval '30 days'", "DELETE FROM session_history WHERE ended_at<now()-interval '90 days'", "DELETE FROM alerts WHERE resolved_at<now()-interval '90 days'"} {
+	for _, query := range []string{"DELETE FROM login_limits WHERE window_started<now()-interval '1 day'", "DELETE FROM resource_samples WHERE sampled_at<now()-interval '30 days'", "DELETE FROM terminal_metric_samples WHERE sampled_at<now()-interval '30 days'", "DELETE FROM terminal_log_archives WHERE captured_at<now()-interval '30 days'", "DELETE FROM terminal_log_entries WHERE captured_at<now()-interval '30 days'", "DELETE FROM session_history WHERE ended_at<now()-interval '90 days'", "DELETE FROM alerts WHERE resolved_at<now()-interval '90 days'"} {
 		if _, err = tx.ExecContext(ctx, query); err != nil {
 			return err
 		}
@@ -74,7 +74,14 @@ func (s *Server) validateTemplate(c *gin.Context) {
 		s.fail(c, 502, "invalid_template", "模板配置摘要缺失")
 		return
 	}
-	if err = s.DB.GetContext(ctx, &template, "UPDATE image_templates SET digest=$1,validated_at=now() WHERE id=$2 RETURNING *", digest, template.ID); err != nil {
+	// A source template change invalidates the previous hardware-encoding
+	// acceptance. The administrator must verify the updated guest image again
+	// before it can be used for a new desktop, regardless of its OS.
+	hardwareVerified := template.HardwareEncodingVerified
+	if template.Digest != digest {
+		hardwareVerified = false
+	}
+	if err = s.DB.GetContext(ctx, &template, "UPDATE image_templates SET digest=$1,validated_at=now(),hardware_encoding_verified=$2 WHERE id=$3 RETURNING *", digest, hardwareVerified, template.ID); err != nil {
 		s.dbError(c, err)
 		return
 	}
@@ -59,8 +59,10 @@ func New(cfg config.Config, db *sqlx.DB, pve *proxmox.Client) *Server {
 	})
 	r.POST("/api/v1/auth/login", s.login)
 	r.POST("/api/v1/terminal/register", s.register)
+	r.POST("/api/v1/gpu-agent/health", s.gpuAgentAuth, s.ingestGPUHealth)
 	t := r.Group("/api/v1/terminal", s.deviceAuth)
 	t.POST("/heartbeat", s.heartbeat)
+	t.POST("/logs", s.appendTerminalLogs)
 	t.GET("/config", s.deviceConfig)
 	t.POST("/config/ack", s.policyAck)
 	t.GET("/desktops", s.deviceDesktops)
@@ -87,6 +89,9 @@ func New(cfg config.Config, db *sqlx.DB, pve *proxmox.Client) *Server {
 	a.GET("/administrators/:id/sessions", s.allow("super_admin"), s.administratorSessions)
 	a.DELETE("/administrators/:id/sessions/:sessionKey", s.allow("super_admin"), s.revokeAdministratorSession)
 	a.GET("/overview", s.overview)
+	a.POST("/assistant/plan", s.assistantPlanHandler)
+	a.POST("/assistant/execute", s.allow("super_admin"), s.assistantExecute)
+	a.GET("/assistant/runs/:id", s.assistantRun)
 	a.GET("/sessions/history", s.sessionHistory)
 	a.GET("/monitor/series", s.resourceSeries)
 	a.GET("/monitor/rules", s.alertRules)
@@ -134,6 +139,7 @@ func New(cfg config.Config, db *sqlx.DB, pve *proxmox.Client) *Server {
 	a.POST("/desktops/import", s.allow("super_admin"), s.importDesktop)
 	a.GET("/desktops/:id", s.desktop)
 	a.PUT("/desktops/:id/connection", s.allow("super_admin"), s.configureDesktop)
+	a.POST("/desktops/:id/readiness/check", s.allow("super_admin", "operator"), s.checkDesktopReadiness)
 	a.POST("/desktops/:id/actions", s.allow("super_admin", "operator"), s.desktopAction)
 	a.GET("/terminals", s.terminals)
 	a.GET("/terminals/:id/metrics", s.terminalMetrics)
@@ -36,7 +36,7 @@ func newFixtureWorker(t *testing.T, worker bool) *fixture {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if _, err = db.Exec(`TRUNCATE terminal_metric_samples,terminal_metrics_current,power_batch_items,power_batches,snapshot_jobs,hardware_jobs,pve_connection,login_limits,alert_states,alerts,alert_rules,resource_samples,session_history,import_batches,terminal_policy_deliveries,terminal_policy_deployments,terminal_policy_templates,provisioning_jobs,desktop_pools,image_templates,connection_tickets,audit_logs,tasks,bindings,users,terminals,desktops,resources,admin_sessions,administrators CASCADE`); err != nil {
+	if _, err = db.Exec(`TRUNCATE assistant_runs,gpu_health_samples,terminal_log_entries,terminal_log_archives,terminal_metric_samples,terminal_metrics_current,power_batch_items,power_batches,snapshot_jobs,hardware_jobs,pve_connection,login_limits,alert_states,alerts,alert_rules,resource_samples,session_history,import_batches,terminal_policy_deliveries,terminal_policy_deployments,terminal_policy_templates,provisioning_jobs,desktop_pools,image_templates,connection_tickets,audit_logs,tasks,bindings,users,terminals,desktops,resources,admin_sessions,administrators CASCADE`); err != nil {
 		t.Fatal(err)
 	}
 	if err = store.Bootstrap(ctx, db, "demo", "", ""); err != nil {
@@ -258,6 +258,13 @@ func TestDeviceIdentityBindingTicketAndAcknowledgement(t *testing.T) {
 	if out["total"] != float64(1) || len(out["items"].([]any)) != 1 || out["items"].([]any)[0].(map[string]any)["logText"] != "private diagnostic" {
 		t.Fatalf("terminal log archive not searchable: %#v", out)
 	}
+	status, out = f.request("POST", "/terminal/logs", device, map[string]any{"bootId": "test-boot", "entries": []map[string]any{{"level": "error", "message": "stream error password=hidden"}}}, "")
+	expect(t, 202, status, out)
+	status, out = f.request("GET", "/terminal-logs?terminalId="+id+"&keyword=stream", admin, nil, "")
+	expect(t, 200, status, out)
+	if out["total"] != float64(1) || out["items"].([]any)[0].(map[string]any)["source"] != "stream" || strings.Contains(out["items"].([]any)[0].(map[string]any)["logText"].(string), "hidden") {
+		t.Fatalf("continuous terminal log was not safely archived: %#v", out)
+	}
 	_ = otherID
 }

@@ -140,6 +140,58 @@ func (s *Server) heartbeat(c *gin.Context) {
 	}
 	s.deviceJSON(c, 200, gin.H{"accepted": true, "serverTime": time.Now().UTC()})
 }
+
+// appendTerminalLogs accepts small log batches from an online terminal. It is
+// deliberately separate from collect-logs: agents can stream useful events
+// during normal operation without creating a management task for every batch.
+func (s *Server) appendTerminalLogs(c *gin.Context) {
+	var in struct {
+		BootID  string `json:"bootId" binding:"max=120"`
+		Entries []struct {
+			CapturedAt string `json:"capturedAt"`
+			Level      string `json:"level" binding:"omitempty,oneof=debug info warn error"`
+			Message    string `json:"message" binding:"required,max=16000"`
+		} `json:"entries" binding:"required,min=1,max=100"`
+	}
+	if !s.decode(c, &in) {
+		return
+	}
+	if len(in.Entries) == 0 {
+		s.fail(c, 400, "invalid_logs", "日志批次不能为空")
+		return
+	}
+	if len(in.BootID) > 120 || strings.IndexByte(in.BootID, 0) >= 0 {
+		s.fail(c, 400, "invalid_logs", "日志启动标识无效")
+		return
+	}
+	tx := deviceTx(c)
+	for _, entry := range in.Entries {
+		message := redactDiagnostic(strings.TrimSpace(entry.Message))
+		if message == "" || len(message) > 16000 {
+			s.fail(c, 400, "invalid_logs", "日志内容不能为空或超出长度限制")
+			return
+		}
+		capturedAt := time.Now().UTC()
+		if strings.TrimSpace(entry.CapturedAt) != "" {
+			parsed, err := time.Parse(time.RFC3339Nano, entry.CapturedAt)
+			if err != nil || parsed.After(time.Now().Add(2*time.Minute)) || parsed.Before(time.Now().Add(-10*time.Minute)) {
+				s.fail(c, 400, "invalid_logs", "日志时间必须是最近 10 分钟内的 RFC3339 时间")
+				return
+			}
+			capturedAt = parsed.UTC()
+		}
+		level := strings.TrimSpace(entry.Level)
+		if level == "" {
+			level = "info"
+		}
+		if _, err := tx.ExecContext(c.Request.Context(), `INSERT INTO terminal_log_entries(id,terminal_id,captured_at,boot_id,level,message) VALUES($1,$2,$3,$4,$5,$6)`, store.ID("log_"), c.GetString("terminalId"), capturedAt, redactDiagnostic(in.BootID), level, message); err != nil {
+			s.dbError(c, err)
+			return
+		}
+	}
+	_, _ = tx.ExecContext(c.Request.Context(), "UPDATE terminals SET last_seen=now(),status='online' WHERE id=$1", c.GetString("terminalId"))
+	s.deviceJSON(c, 202, gin.H{"accepted": len(in.Entries), "serverTime": time.Now().UTC()})
+}
 func (s *Server) deviceConfig(c *gin.Context) {
 	var version int
 	if err := deviceTx(c).GetContext(c.Request.Context(), &version, "SELECT config_version FROM terminals WHERE id=$1", c.GetString("terminalId")); err != nil {
@@ -82,12 +82,27 @@ CREATE TABLE IF NOT EXISTS terminal_log_archives (
 );
 CREATE INDEX IF NOT EXISTS terminal_log_archives_terminal_time ON terminal_log_archives(terminal_id,captured_at DESC,id);
 CREATE INDEX IF NOT EXISTS terminal_log_archives_time ON terminal_log_archives(captured_at DESC);
+CREATE TABLE IF NOT EXISTS terminal_log_entries (
+ id text PRIMARY KEY, terminal_id text NOT NULL REFERENCES terminals(id) ON DELETE CASCADE,
+ captured_at timestamptz NOT NULL DEFAULT now(), boot_id text NOT NULL DEFAULT '', level text NOT NULL DEFAULT 'info',
+ message text NOT NULL, metadata jsonb NOT NULL DEFAULT '{}'::jsonb
+);
+CREATE INDEX IF NOT EXISTS terminal_log_entries_terminal_time ON terminal_log_entries(terminal_id,captured_at DESC,id);
+CREATE INDEX IF NOT EXISTS terminal_log_entries_time ON terminal_log_entries(captured_at DESC);
+CREATE TABLE IF NOT EXISTS assistant_runs (
+ id text PRIMARY KEY, actor text NOT NULL REFERENCES administrators(username), input_text text NOT NULL,
+ action text NOT NULL, status text NOT NULL CHECK(status IN ('planned','blocked','executed','expired','failed')),
+ plan jsonb NOT NULL, fingerprint text NOT NULL, created_at timestamptz NOT NULL DEFAULT now(),
+ expires_at timestamptz NOT NULL, executed_at timestamptz, result jsonb NOT NULL DEFAULT '{}'::jsonb
+);
+CREATE INDEX IF NOT EXISTS assistant_runs_actor_time ON assistant_runs(actor,created_at DESC);
 CREATE TABLE IF NOT EXISTS image_templates (
 id text PRIMARY KEY, name text NOT NULL, version text NOT NULL, os text NOT NULL CHECK(os IN ('Windows','Ubuntu','麒麟')),
 node text NOT NULL, vmid integer NOT NULL CHECK(vmid>=100), digest text NOT NULL,
 enabled boolean NOT NULL DEFAULT true, validated_at timestamptz NOT NULL DEFAULT now(),
 UNIQUE(node,vmid), UNIQUE(name,version)
 );
+ALTER TABLE image_templates ADD COLUMN IF NOT EXISTS hardware_encoding_verified boolean NOT NULL DEFAULT false;
 CREATE TABLE IF NOT EXISTS desktop_pools (
 id text PRIMARY KEY, name text NOT NULL UNIQUE, type text NOT NULL CHECK(type IN ('dedicated','shared')),
 max_count integer NOT NULL CHECK(max_count BETWEEN 1 AND 10000),
@@ -124,6 +139,17 @@ CREATE TABLE IF NOT EXISTS resource_samples (
 memory double precision, storage double precision, PRIMARY KEY(resource_id,sampled_at)
 );
 CREATE INDEX IF NOT EXISTS resource_samples_time ON resource_samples(sampled_at);
+-- GPU health is reported by an explicitly configured node collector. PVE's
+-- PCI inventory does not expose live temperature, utilization or VRAM usage.
+CREATE TABLE IF NOT EXISTS gpu_health_samples (
+ node text NOT NULL, address text NOT NULL, observed_at timestamptz NOT NULL DEFAULT now(),
+ temperature_c double precision, utilization_percent double precision,
+ memory_used_bytes bigint, memory_total_bytes bigint,
+ driver text NOT NULL DEFAULT '', driver_binding text NOT NULL DEFAULT '', source text NOT NULL,
+ PRIMARY KEY(node,address,observed_at)
+);
+CREATE INDEX IF NOT EXISTS gpu_health_latest ON gpu_health_samples(node,address,observed_at DESC);
+DELETE FROM gpu_health_samples WHERE observed_at < now() - interval '7 days';
 CREATE TABLE IF NOT EXISTS alert_rules (
 id text PRIMARY KEY, name text NOT NULL, metric text NOT NULL CHECK(metric IN ('node_cpu','node_memory','node_storage','node_unavailable','terminal_offline')),
 threshold double precision NOT NULL, hold_seconds integer NOT NULL DEFAULT 0, enabled boolean NOT NULL DEFAULT true
@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
@@ -123,16 +124,48 @@ func (c *Client) Action(ctx context.Context, node string, vmid int, action strin
 }
 func (c *Client) Task(ctx context.Context, node, upid string) (TaskStatus, error) {
 	var status TaskStatus
-	err := c.request(ctx, "GET", "/nodes/"+url.PathEscape(node)+"/tasks/"+url.PathEscape(upid)+"/status", nil, &status)
+	err := c.readWithRetry(ctx, "/nodes/"+url.PathEscape(node)+"/tasks/"+url.PathEscape(upid)+"/status", &status)
 	return status, err
 }
 func (c *Client) Power(ctx context.Context, node string, vmid int) (string, error) {
 	var status struct {
 		Status string `json:"status"`
 	}
-	err := c.request(ctx, "GET", fmt.Sprintf("/nodes/%s/qemu/%d/status/current", url.PathEscape(node), vmid), nil, &status)
+	err := c.readWithRetry(ctx, fmt.Sprintf("/nodes/%s/qemu/%d/status/current", url.PathEscape(node), vmid), &status)
 	return status.Status, err
 }
+
+// Retry only observations. A lost POST response may already have changed the
+// VM, so commands must never use this path.
+func (c *Client) readWithRetry(ctx context.Context, path string, out any) error {
+	var err error
+	for attempt := 0; attempt < 3; attempt++ {
+		if err = ctx.Err(); err != nil {
+			return err
+		}
+		err = c.request(ctx, "GET", path, nil, out)
+		if err == nil || !retryableRead(err) || attempt == 2 {
+			return err
+		}
+		timer := time.NewTimer(time.Duration(attempt+1) * 250 * time.Millisecond)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return ctx.Err()
+		case <-timer.C:
+		}
+	}
+	return err
+}
+
+func retryableRead(err error) bool {
+	var response *HTTPError
+	if errors.As(err, &response) {
+		return response.Status == 408 || response.Status == 429 || response.Status == 500 || response.Status == 502 || response.Status == 503 || response.Status == 504
+	}
+	var network net.Error
+	return errors.As(err, &network) && (network.Timeout() || network.Temporary()) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF)
+}
 func (c *Client) Wait(ctx context.Context, node, upid string, interval time.Duration) error {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
@@ -2,6 +2,7 @@ package proxmox

 import (
 	"context"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -58,6 +59,70 @@ func TestTLSAndTaskLifecycle(t *testing.T) {
 		t.Fatal("restore must never map to hard reset")
 	}
 }
+
+func TestObservationRetriesDoNotReplayPowerCommands(t *testing.T) {
+	for _, endpoint := range []string{"task", "power"} {
+		t.Run(endpoint, func(t *testing.T) {
+			reads, writes := 0, 0
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.Method == "POST" {
+					writes++
+					w.WriteHeader(http.StatusServiceUnavailable)
+					return
+				}
+				reads++
+				if reads < 3 {
+					w.WriteHeader(http.StatusServiceUnavailable)
+					return
+				}
+				fmtBody := `{"data":{"status":"running"}}`
+				if endpoint == "task" {
+					fmtBody = `{"data":{"status":"stopped","exitstatus":"OK"}}`
+				}
+				w.Write([]byte(fmtBody))
+			}))
+			defer ts.Close()
+			c := &Client{BaseURL: ts.URL, HTTP: ts.Client()}
+			var err error
+			if endpoint == "task" {
+				err = c.Wait(context.Background(), "node", "UPID:one", time.Millisecond)
+			} else {
+				_, err = c.Power(context.Background(), "node", 100)
+			}
+			if err != nil || reads != 3 {
+				t.Fatal("observation did not recover", reads, err)
+			}
+			if _, err = c.Action(context.Background(), "node", 100, "start"); err == nil || writes != 1 {
+				t.Fatal("command response loss must not replay POST", writes, err)
+			}
+		})
+	}
+}
+
+func TestObservationRetriesAreBoundedAndRespectCancellation(t *testing.T) {
+	for _, status := range []int{http.StatusForbidden, http.StatusServiceUnavailable} {
+		calls := 0
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			calls++
+			w.WriteHeader(status)
+		}))
+		c := &Client{BaseURL: ts.URL, HTTP: ts.Client()}
+		_, err := c.Power(context.Background(), "node", 100)
+		want := 1
+		if status == http.StatusServiceUnavailable {
+			want = 3
+		}
+		if err == nil || calls != want {
+			t.Fatal("incorrect retry limit", calls, err)
+		}
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		if _, err = c.Power(ctx, "node", 100); !errors.Is(err, context.Canceled) || calls != want {
+			t.Fatal("cancellation ignored", calls, err)
+		}
+		ts.Close()
+	}
+}
 func TestFailedPVEJobIsNotSuccess(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Write([]byte(`{"data":{"status":"stopped","exitstatus":"ERROR: disk full"}}`))