fix(runtime): hold taskrun fanout lock, guard goroutines, and escape math XSS

Serialize appendEvent sends under the subscriber mutex, wrap worker/memory goroutines with recover, lock synchronous chat SendMessage, and escape KaTeX error HTML. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-20 10:54:18 +00:00
parent 3a3a79769e
commit b4388934a0
9 changed files with 126 additions and 31 deletions
@@ -16,8 +16,8 @@
 |-------|------|--------|----------|-------|--------|
 | Cursor-Auto | 2026-09-20-p0-hotfixes（Wave 0 文档） | docs/p0-triage-2026-09-20 | - | docs | done |
 | Cursor-Auto | 2026-09-20-p0-capability-00058 | fix/capability-merge-00058-2026-09-20 | - | infra+chat | done |
-| Cursor-Auto | 2026-09-20-p0-idor | fix/authz-idor-hotfixes-2026-09-20 | - | authz | in_progress |
-| Cursor-Auto | 2026-09-20-p0-runtime-stability | fix/runtime-panic-races-2026-09-20 | - | taskrun+chat+web | planned |
+| Cursor-Auto | 2026-09-20-p0-idor | fix/authz-idor-hotfixes-2026-09-20 | - | authz | done |
+| Cursor-Auto | 2026-09-20-p0-runtime-stability | fix/runtime-panic-races-2026-09-20 | - | taskrun+chat+web | in_progress |
 | Cursor-Auto | 2026-09-20-p0-credentials | fix/credential-hardening-2026-09-20 | - | ops+infra | planned |
 | Cursor-Auto | 2026-09-20-proposed-authoritative-reports | feature/capability-enablement-impact-plan | - | docgov | done |
 | Cursor-Auto | 2026-09-20-pi-narrative-clipped-diff | feature/capability-enablement-impact-plan | - | docgov | done |
@@ -165,7 +165,7 @@

 | holder | phase | since | scope | queue |
 |--------|-------|-------|-------|-------|
-| Cursor-Auto | merging | 2026-09-20 | fix/authz-idor-hotfixes | fix/runtime-panic-races → fix/credential-hardening |
+| - | idle | 2026-09-20 | authz-idor rebuild+curl PASS；槽位释放 | fix/runtime-panic-races → fix/credential-hardening |

 **phase**: `idle` | `merging` | `smoke` | `e2e`

@@ -3,6 +3,7 @@ package chat
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"strings"
 	"sync"
 	"time"
@@ -602,6 +603,11 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str
 		iName := string(coordRes.Intent)
 		bName := string(recRes.Branch)
 		go func() {
+			defer func() {
+				if rec := recover(); rec != nil {
+					slog.Error("chat_turn_memory_panic", "session_id", sessID, "recover", rec)
+				}
+			}()
 			bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 			defer cancel()
 			_, _ = s.CtxMgr.RecordTurnMemory(bgCtx, sessID, turnNo, uid, pid, q, ans, iName, bName, isMilestone)
@@ -15,6 +15,7 @@ import (
 	"chat-agent-system/internal/app/docflow"
 	"chat-agent-system/internal/app/uow"
 	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/chat"
 	"chat-agent-system/internal/ctxmgr"
 	"chat-agent-system/internal/docgov"
 	"chat-agent-system/internal/httpx"
@@ -252,7 +253,12 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
 	}
 	s.bindSessionDocumentSummaryReader(sess, user)

-	resp, err := sess.SendMessage(r.Context(), req.Message)
+	var resp *chat.ChatResponse
+	err = sess.WithExecLock(func() error {
+		var sendErr error
+		resp, sendErr = sess.SendMessage(r.Context(), req.Message)
+		return sendErr
+	})
 	if err != nil {
 		writeSafeError(w, r, http.StatusInternalServerError, "process chat message failed: "+err.Error())
 		return
@@ -262,16 +262,25 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen
 				for _, s := range parallelGroup {
 					sCopy := s
 					wg.Add(1)
-					go func() {
-						defer wg.Done()
-						if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
-							mu.Lock()
-							if groupErr == nil {
-								groupErr = err
-							}
-							mu.Unlock()
-						}
-					}()
+		go func() {
+			defer wg.Done()
+			defer func() {
+				if rec := recover(); rec != nil {
+					mu.Lock()
+					if groupErr == nil {
+						groupErr = fmt.Errorf("parallel step panic: %v", rec)
+					}
+					mu.Unlock()
+				}
+			}()
+			if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
+				mu.Lock()
+				if groupErr == nil {
+					groupErr = err
+				}
+				mu.Unlock()
+			}
+		}()
 				}
 				wg.Wait()
 				if groupErr != nil {
@@ -287,17 +287,32 @@ func (m *Manager) StartClaimedRun(parent context.Context, record *RunRecord, tas
 		})
 	}

-	go m.runWorker(workerCtx, ar, taskReq, stageReq, resumeWorkDir)
+	m.runGuarded("taskrun_worker", func() {
+		m.runWorker(workerCtx, ar, taskReq, stageReq, resumeWorkDir)
+	})
 	cpy := *record
 	return &cpy, nil
 }

+func (m *Manager) runGuarded(name string, fn func()) {
+	go func() {
+		defer func() {
+			if rec := recover(); rec != nil {
+				slog.Error("taskrun_goroutine_panic", "name", name, "recover", rec)
+			}
+		}()
+		fn()
+	}()
+}
+
 func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) {
 	defer close(ar.done)
 	hbCtx, hbCancel := context.WithCancel(ctx)
 	defer hbCancel()
 	if m.store != nil && m.store.Available() {
-		go m.heartbeatLoop(hbCtx, ar.record.RunID)
+		m.runGuarded("taskrun_heartbeat", func() {
+			m.heartbeatLoop(hbCtx, ar.record.RunID)
+		})
 	}

 	taskReq.RunID = ar.record.RunID
@@ -478,9 +493,12 @@ func (m *Manager) appendEvent(ar *activeRun, ev stream.StreamEvent) {
 	ar.events = append(ar.events, stored)
 	runID := ar.record.RunID
 	seq := ar.nextSeq
-	subs := make([]chan StoredEvent, 0, len(ar.subs))
+	// 持锁发送：与 Unsubscribe 的 close 互斥，避免向已关闭 channel 发送 panic。
 	for _, ch := range ar.subs {
-		subs = append(subs, ch)
+		select {
+		case ch <- stored:
+		default:
+		}
 	}
 	ar.mu.Unlock()

@@ -489,13 +507,6 @@ func (m *Manager) appendEvent(ar *activeRun, ev stream.StreamEvent) {
 			slog.Warn("taskrun_event_persist_failed", "run_id", runID, "seq", seq, "error", err)
 		}
 	}
-
-	for _, ch := range subs {
-		select {
-		case ch <- stored:
-		default:
-		}
-	}
 }

 func (m *Manager) GetRun(runID string) (*RunRecord, error) {
@@ -651,7 +662,9 @@ func (m *Manager) ResumeInterrupted(parent context.Context, record *RunRecord, r
 	m.runs[cpy.RunID] = ar
 	m.mu.Unlock()

-	go m.runWorker(workerCtx, ar, taskReq, stageReq, resumeWorkDir)
+	m.runGuarded("taskrun_worker_resume", func() {
+		m.runWorker(workerCtx, ar, taskReq, stageReq, resumeWorkDir)
+	})
 	return &cpy, nil
 }

@@ -0,0 +1,48 @@
+package taskrun
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/stream"
+)
+
+func TestSubscribeUnsubscribeRace(t *testing.T) {
+	m := NewManager(nil, nil)
+	ar := &activeRun{
+		record: &RunRecord{RunID: "run-race-1", Status: StatusRunning},
+		subs:   make(map[int]chan StoredEvent),
+		done:   make(chan struct{}),
+	}
+	m.mu.Lock()
+	m.runs[ar.record.RunID] = ar
+	m.mu.Unlock()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 50; i++ {
+		wg.Add(3)
+		go func() {
+			defer wg.Done()
+			ch, _, err := m.Subscribe(ar.record.RunID, 0)
+			if err != nil {
+				return
+			}
+			time.Sleep(time.Millisecond)
+			m.Unsubscribe(ar.record.RunID, ch)
+		}()
+		go func() {
+			defer wg.Done()
+			m.appendEvent(ar, stream.StreamEvent{Type: stream.EventDelta, Delta: "x"})
+		}()
+		go func() {
+			defer wg.Done()
+			ch, _, err := m.Subscribe(ar.record.RunID, 0)
+			if err != nil {
+				return
+			}
+			m.Unsubscribe(ar.record.RunID, ch)
+		}()
+	}
+	wg.Wait()
+}
@@ -24,6 +24,7 @@ import {
 import katex from 'katex';
 import 'katex/dist/katex.min.css';
 import { renderInlineMarkdown } from '../../engine/renderMarkdown';
+import { escapeHtml } from '../../engine/replyDocument/sanitize';
 import { BLOCK_EDITOR_DIALOG_LABEL, BLOCK_EDITOR_SAVE_NAME } from '../../lib/blockEditorDialog';

 interface BlockEditorModalProps {
@@ -202,8 +203,9 @@ export const BlockEditorModal: React.FC<BlockEditorModalProps> = ({ block, onSav
          throwOnError: true,
        });
        return `<div style="padding:24px;display:flex;justifyContent:center;align-items:center;background:var(--bg-card);border-radius:var(--r-md);overflow-x:auto;">${rendered}</div>`;
-      } catch (err: any) {
-        return `<div style="padding:14px;color:var(--danger);background:var(--danger-bg);border:1px solid rgba(239, 68, 68, 0.2);border-radius:var(--r-md);font-size:12px;"><strong>LaTeX 语法解析异常:</strong> ${err?.message || '公式排版错误'}<pre style="margin-top:8px;padding:8px;background:rgba(0,0,0,0.2);border-radius:4px;font-family:var(--font-mono);font-size:11px;overflow-x:auto;">${cleanMath}</pre></div>`;
+      } catch (err: unknown) {
+        const message = err instanceof Error ? err.message : '公式排版错误';
+        return `<div style="padding:14px;color:var(--danger);background:var(--danger-bg);border:1px solid rgba(239, 68, 68, 0.2);border-radius:var(--r-md);font-size:12px;"><strong>LaTeX 语法解析异常:</strong> ${escapeHtml(message)}<pre style="margin-top:8px;padding:8px;background:rgba(0,0,0,0.2);border-radius:4px;font-family:var(--font-mono);font-size:11px;overflow-x:auto;">${escapeHtml(cleanMath)}</pre></div>`;
      }
    }

@@ -3,6 +3,7 @@ import { DocumentBlock } from '../../../types/block';
 import katex from 'katex';
 import 'katex/dist/katex.min.css';
 import { Calculator } from 'lucide-react';
+import { escapeHtml } from '../../../engine/replyDocument/sanitize';

 interface MathBlockViewProps {
  block: DocumentBlock;
@@ -16,8 +17,9 @@ export const MathBlockView: React.FC<MathBlockViewProps> = ({ block }) => {
        displayMode: true,
        throwOnError: false,
      });
-    } catch (err: any) {
-      return `<span style="color:var(--danger)">LaTeX 解析错误: ${err.message}</span>`;
+    } catch (err: unknown) {
+      const message = err instanceof Error ? err.message : 'unknown';
+      return `<span style="color:var(--danger)">LaTeX 解析错误: ${escapeHtml(message)}</span>`;
    }
  }, [block.content]);

@@ -4,7 +4,7 @@ import { parseReplyMarkdown } from './parseReplyMarkdown';
 import { renderClipboardHtml } from './renderClipboardHtml';
 import { renderPlainText } from './renderPlainText';
 import { serializeMarkdown } from './serializeMarkdown';
-import { stripUnsafeHtml, sanitizeLinkHref } from './sanitize';
+import { stripUnsafeHtml, sanitizeLinkHref, escapeHtml } from './sanitize';
 import { deriveReplyTitle, sanitizeFilename } from '../../features/replyExport/title';

 describe('Reply Document IR & Export Engine', () => {
@@ -137,4 +137,13 @@ $$
    assert.ok(clean.includes('Normal'));
    assert.equal(sanitizeLinkHref('javascript:alert(1)'), '#');
  });
+
+  test('escapeHtml escapes math error payloads used by BlockEditorModal catch path', () => {
+    const dirty = `<img src=x onerror=alert(1)><script>alert(2)</script>`;
+    const escaped = escapeHtml(dirty);
+    assert.ok(!escaped.includes('<script>'));
+    assert.ok(!escaped.includes('<img'));
+    assert.ok(escaped.includes('&lt;script&gt;'));
+    assert.ok(escaped.includes('&lt;img'));
+  });
 });