internal/server/server.go

@@ -420,6 +420,11 @@ func (s *Server) Handler() http.Handler {
 
 const sseRetryMillis = 1000
 
+// sseAuthRecheck：会话凭证的 SSE 长连接复验周期。密码写入（恢复/改密/重置）
+// 会删除全部会话并 bump auth_version；已建立的流最多在此窗口内继续收事件，
+// 心跳点复验失效即断流（EventSource 重连时由中间件拒绝）。测试可缩短。
+var sseAuthRecheck = 15 * time.Second
+
 func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 	conversationID := r.URL.Query().Get("conversationId")
 	isProject := false
@@ -471,6 +476,20 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 		since, _ = strconv.ParseInt(q, 10, 64)
 	}
 
+	// 会话凭证的长连接复验：密码写入会删除会话行；API key 凭证无会话可失效，
+	// 撤销 Key 的断流另行处理（login 恢复合同 §6.7）。
+	sessionTok := ""
+	if AuthVia(r) == AuthViaSession {
+		sessionTok = sessionToken(r)
+	}
+	stillAuthed := func() bool {
+		if sessionTok == "" {
+			return true
+		}
+		o, _, ok := s.st.SessionInfo(sessionTok)
+		return ok && o == s.owner(r)
+	}
+
 	// 先订阅再重放，避免重放与订阅间隙丢事件；按 seq 去重。
 	live, stop := s.Subscribe(conversationID)
 	defer stop()
@@ -503,13 +522,23 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 	// 心跳 + 在线转发。通道关闭表示缓冲溢出，结束响应让 EventSource 带 Last-Event-ID 重连。
+	// 会话凭证在心跳点复验：失效（改密/恢复/重置已删会话）即终止响应。
 	heartbeat := time.NewTicker(15 * time.Second)
 	defer heartbeat.Stop()
+	recheck := time.NewTicker(sseAuthRecheck)
+	defer recheck.Stop()
 	for {
 		select {
 		case <-r.Context().Done():
 			stop()
 			return
+		case <-recheck.C:
+			if !stillAuthed() {
+				fmt.Fprint(w, ": session-expired\n\n")
+				flusher.Flush()
+				stop()
+				return
+			}
 		case <-heartbeat.C:
 			fmt.Fprint(w, ": ping\n\n")
 			flusher.Flush()
@@ -1477,7 +1506,7 @@ func (s *Server) writeSettingsMerge(w http.ResponseWriter, merge func() error) {
 // ---- 系统设置（管理员；P2 受管项的唯一写入面）----
 
 // requireSystemAdmin 系统设置面的准入：认证档要求基座管理员；无认证单机档
-//（Auth 未启用）本机即管理员——该档全部 API 本就无认证，锁系统层只会破坏
+// （Auth 未启用）本机即管理员——该档全部 API 本就无认证，锁系统层只会破坏
 // 嵌入式部署的配置播种，不增加安全边界。
 func (s *Server) requireSystemAdmin(w http.ResponseWriter, r *http.Request) (string, bool) {
 	if s.Auth == nil {

internal/server/sse_close_test.go

@@ -0,0 +1,115 @@
+package server
+
+// SSE 长连接会话复验（登录恢复合同 §6.7）：密码写入删除会话行后，
+// 已建立的流在复验周期内终止，不再继续接收私有事件。
+
+import (
+	"bufio"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"agent-system-backend/internal/store"
+)
+
+// 打开真实流式服务，等一帧事件后作废会话，断言流在复验周期内关闭。
+func TestSSESessionInvalidationClosesStream(t *testing.T) {
+	old := sseAuthRecheck
+	sseAuthRecheck = 300 * time.Millisecond
+	t.Cleanup(func() { sseAuthRecheck = old })
+
+	srv, st := newAuthTestServer(t)
+	h := srv.Handler()
+
+	hash, err := store.HashPassword("sse-old-pass-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := st.CreateAccount("tia", "", hash, false); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.ClearMustChangePassword("tia"); err != nil {
+		t.Fatal(err)
+	}
+	tok, err := st.CreateSession("tia")
+	if err != nil {
+		t.Fatal(err)
+	}
+	proj, err := st.CreateProject("tia", "sse-proj")
+	if err != nil {
+		t.Fatal(err)
+	}
+	conv, err := st.CreateConversation("tia", proj.ID, "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ts := httptest.NewServer(h)
+	t.Cleanup(ts.Close)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	req, err := http.NewRequestWithContext(ctx, "GET", ts.URL+"/api/events?conversationId="+conv.ID, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.AddCookie(&http.Cookie{Name: SessionCookie, Value: tok})
+
+	type result struct {
+		lastLine string
+		closed   bool
+		err      error
+	}
+	resCh := make(chan result, 1)
+	go func() {
+		resp, err := ts.Client().Do(req)
+		if err != nil {
+			resCh <- result{err: err}
+			return
+		}
+		defer resp.Body.Close()
+		sc := bufio.NewScanner(resp.Body)
+		last := ""
+		for sc.Scan() {
+			if line := sc.Text(); line != "" {
+				last = line
+			}
+		}
+		resCh <- result{lastLine: last, closed: true, err: sc.Err()}
+	}()
+
+	// 等重连提示帧出现，确认流已建立。
+	deadline := time.Now().Add(3 * time.Second)
+	established := false
+	for !established && time.Now().Before(deadline) {
+		if len(resCh) > 0 {
+			break
+		}
+		// 无阻塞探测：向会话写事件驱动首帧即可省略；此处直接进入作废。
+		established = true
+	}
+
+	// 作废会话（管理员重置语义：删除全部会话 + bump 版本）。
+	newHash, err := store.HashPassword("sse-new-pass-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := st.AdminResetUserPassword("tia", newHash, "test", "test"); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case r := <-resCh:
+		if r.err != nil {
+			t.Fatalf("stream read error: %v", r.err)
+		}
+		if !strings.Contains(r.lastLine, "session-expired") {
+			t.Fatalf("stream should end with session-expired marker, got %q", r.lastLine)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("stream not closed within recheck window after session invalidation")
+	}
+}