fix: detach local sessions from request context

cmd/hub/main.go

@@ -70,6 +70,7 @@ func main() {
 		AdminToken:    adminToken,
 		ExecLimiter:   make(chan struct{}, 32),
 		SearchLimiter: make(chan struct{}, 8),
+		LocalSessions: executor.NewSessionManager(0),
 	}
 	if err := s.LoadSecurityPolicySnapshot(); err != nil {
 		log.Fatalf("load security policy snapshot: %v", err)
@@ -121,6 +122,13 @@ func main() {
 		IdleTimeout:       90 * time.Second,
 	}
 	log.Printf("vps-commander-hub listening on %s as %s", *addr, local)
+	go func() {
+		ticker := time.NewTicker(time.Minute)
+		defer ticker.Stop()
+		for range ticker.C {
+			s.LocalSessions.Cleanup()
+		}
+	}()
 	go func() {
 		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 			log.Fatal(err)

docs/KNOWN_ISSUE_STREAMABLE_HTTP_20261002.md

@@ -0,0 +1,49 @@
+# 已知问题：MCP Streamable HTTP 传输不受支持（待排期）
+
+- **发现日期**：2026-10-02
+- **发现人**：审核专家 B（生产问题分析）
+- **状态**：待排期，未修复
+
+---
+
+## 现象
+
+Android 客户端使用 MCP Kotlin SDK 的 `StreamableHttpClientTransport` 连接 vps-commander，
+发送第一条消息（`initialize`）时失败：
+
+```
+io.modelcontextprotocol.kotlin.sdk.types.McpException: Error while sending message: Streamable HTTP error:
+Caused by: io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpError: Streamable HTTP error:
+    at ...StreamableHttpClientTransport.performSend(...)
+```
+
+## 根因
+
+传输协议代际不匹配：
+
+| | 服务端 (vps-commander) | 客户端 (Kotlin SDK) |
+|---|---|---|
+| 协议 | SSE 传输（MCP 2024-11-05） | Streamable HTTP（MCP 2025-03-26） |
+| 端点 | `GET /mcp/sse` + `POST /mcp/message` | `POST /mcp`（单端点） |
+| 行为 | `HandleSSE` 非 GET 直接 405（`internal/mcp/mcp.go:120`） | POST 收到非 2xx 即抛 `StreamableHttpError` |
+
+服务端没有 `POST /mcp` 路由。客户端 POST 打到 `/mcp/sse` 得 405，打到 `/mcp` 得 404，
+两种情况都会触发该异常。SDK 的错误信息为空（未携带 HTTP 状态码），需抓包/curl 确认。
+
+## 验证
+
+```bash
+curl -i -X POST https://vpstool.kory.kdns.fr/mcp/sse \
+  -H "Content-Type: application/json" -d '{}'
+# 预期：405 method not allowed（坐实协议不匹配）
+```
+
+## 修复选项（待决策）
+
+1. **客户端改（快）**：Android 端改用 `SseClientTransport`，URL 指向 `/mcp/sse`，与服务端现状匹配。
+2. **服务端加（彻底）**：实现 Streamable HTTP（`POST /mcp`，MCP 2025-03-26），属新功能开发。
+
+## 备注
+
+- 与 2026-10-02 bug hunt 的 4 个 bug 无关，是独立的协议兼容性问题。
+- 决策前不要动生产端点。

internal/api/api.go

@@ -49,6 +49,7 @@ type Server struct {
 	AdminToken    string
 	ExecLimiter   chan struct{}
 	SearchLimiter chan struct{}
+	LocalSessions *executor.SessionManager
 }
 
 const (
@@ -782,9 +783,13 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	var req struct {
-		Device, SessionID, Command, Workdir, Data string
-		Force                                     bool
-		RingCapacity                              int `json:"ring_capacity"`
+		Device       string `json:"device"`
+		SessionID    string `json:"session_id"`
+		Command      string `json:"command"`
+		Workdir      string `json:"workdir"`
+		Data         string `json:"data"`
+		Force        bool   `json:"force"`
+		RingCapacity int    `json:"ring_capacity"`
 	}
 	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 256*1024)).Decode(&req); err != nil {
 		http.Error(w, "bad json", 400)
@@ -796,38 +801,106 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) {
 	if !s.requireSecurity(w, r, req.Device, "exec", req.SessionID) {
 		return
 	}
-	if s.Cluster == nil {
-		http.Error(w, "cluster unavailable", 503)
-		return
-	}
 	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
 	defer cancel()
 	action := r.URL.Query().Get("action")
+	owner := sessionOwner(r)
 	var result any
 	var err error
-	switch action {
-	case "start":
-		if req.SessionID == "" {
-			req.SessionID, err = newSessionID()
-			if err != nil {
-				http.Error(w, "session id generation failed", 500)
-				return
+	if s.isRemote(req.Device) {
+		if s.Cluster == nil || !s.Cluster.Online(req.Device) {
+			http.Error(w, "device offline", 409)
+			return
+		}
+		switch action {
+		case "start":
+			if req.SessionID == "" {
+				req.SessionID, err = newSessionID()
+				if err != nil {
+					http.Error(w, "session id generation failed", 500)
+					return
+				}
 			}
+			result, err = s.Cluster.StartProcess(ctx, req.Device, cluster.StartProcessPayload{Command: req.Command, Workdir: req.Workdir, SessionID: req.SessionID, Owner: owner, RingCapacity: req.RingCapacity})
+		case "read":
+			result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, owner)
+		case "input":
+			err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, owner)
+			result = map[string]any{"ok": true}
+		case "stop":
+			err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, owner)
+			result = map[string]any{"ok": true}
+		case "list":
+			result, err = s.Cluster.ListSessions(ctx, req.Device)
+		default:
+			http.Error(w, "unknown action", 400)
+			return
+		}
+	} else {
+		if s.LocalSessions == nil {
+			http.Error(w, "local session manager unavailable", 503)
+			return
+		}
+		switch action {
+		case "start":
+			if req.SessionID == "" {
+				req.SessionID, err = newSessionID()
+				if err != nil {
+					http.Error(w, "session id generation failed", 500)
+					return
+				}
+			}
+			var sess *executor.Session
+			sess, err = s.LocalSessions.Start(context.Background(), req.SessionID, req.Command, req.Workdir, owner, req.RingCapacity)
+			if err == nil {
+				result = map[string]any{"session_id": sess.ID, "started_at": sess.StartedAt.Unix()}
+			}
+		case "read":
+			var sess *executor.Session
+			sess, ok := s.LocalSessions.Get(req.SessionID)
+			if !ok {
+				err = fmt.Errorf("session not found")
+				break
+			}
+			if !sess.Authorized(owner) {
+				err = fmt.Errorf("session ownership denied")
+				break
+			}
+			code, done, _ := sess.Status()
+			stdout, stderr := sess.Output()
+			result = cluster.ProcessOutputPayload{SessionID: req.SessionID, Stdout: stdout, Stderr: stderr, ExitCode: code, Running: !done}
+		case "input":
+			var sess *executor.Session
+			sess, ok := s.LocalSessions.Get(req.SessionID)
+			if !ok {
+				err = fmt.Errorf("session not found")
+				break
+			}
+			if !sess.Authorized(owner) {
+				err = fmt.Errorf("session ownership denied")
+				break
+			}
+			err = sess.Stdin([]byte(req.Data))
+			result = map[string]any{"ok": true}
+		case "stop":
+			var sess *executor.Session
+			sess, ok := s.LocalSessions.Get(req.SessionID)
+			if !ok {
+				err = fmt.Errorf("session not found")
+				break
+			}
+			if !sess.Authorized(owner) {
+				err = fmt.Errorf("session ownership denied")
+				break
+			}
+			err = s.LocalSessions.Stop(req.SessionID, true)
+			result = map[string]any{"ok": true}
+		case "list":
+			result = s.LocalSessions.List()
+		default:
+			http.Error(w, "unknown action", 400)
+			return
 		}
-		result, err = s.Cluster.StartProcess(ctx, req.Device, cluster.StartProcessPayload{Command: req.Command, Workdir: req.Workdir, SessionID: req.SessionID, Owner: sessionOwner(r), RingCapacity: req.RingCapacity})
-	case "read":
-		result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, sessionOwner(r))
-	case "input":
-		err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, sessionOwner(r))
-		result = map[string]any{"ok": true}
-	case "stop":
-		err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, sessionOwner(r))
-		result = map[string]any{"ok": true}
-	case "list":
-		result, err = s.Cluster.ListSessions(ctx, req.Device)
-	default:
-		http.Error(w, "unknown action", 400)
-		return
 	}
 	if err != nil {
 		http.Error(w, err.Error(), 409)

internal/api/api_test.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/wjyhk/vps-commander/internal/executor"
 	"github.com/wjyhk/vps-commander/internal/storage"
@@ -92,6 +93,38 @@ func TestSecurityEffectiveModeAndExecGate(t *testing.T) {
 	}
 }
 
+func TestProcessSessionLocalSnakeCaseAndLifecycle(t *testing.T) {
+	dir := t.TempDir()
+	store, err := storage.Open(filepath.Join(dir, "db.sqlite"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.DB.Close()
+	if err := store.SetGlobalSecurityMode("high"); err != nil {
+		t.Fatal(err)
+	}
+	s := &Server{Exec: executor.Local{MaxOutput: 1024 * 1024}, Store: store, LocalName: "local", LocalSessions: executor.NewSessionManager(0)}
+	if err := s.LoadSecurityPolicySnapshot(); err != nil {
+		t.Fatal(err)
+	}
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/process/session?action=start", strings.NewReader(`{"device":"local","session_id":"session-test-001","command":"sleep 0.2; printf local-ok"}`))
+	w := httptest.NewRecorder()
+	s.processSession(w, req)
+	if w.Code != http.StatusOK {
+		t.Fatalf("start status=%d body=%s", w.Code, w.Body.String())
+	}
+	time.Sleep(400 * time.Millisecond)
+	req = httptest.NewRequest(http.MethodPost, "/api/v1/process/session?action=read", strings.NewReader(`{"device":"local","session_id":"session-test-001"}`))
+	w = httptest.NewRecorder()
+	s.processSession(w, req)
+	if w.Code != http.StatusOK {
+		t.Fatalf("read status=%d body=%s", w.Code, w.Body.String())
+	}
+	if !strings.Contains(w.Body.String(), "local-ok") {
+		t.Fatalf("missing session output: %s", w.Body.String())
+	}
+}
+
 func TestEditBlockSecurityAndReplacement(t *testing.T) {
 	dir := t.TempDir()
 	store, err := storage.Open(filepath.Join(dir, "db.sqlite"))