merge: sync origin/main and preserve learning stream fix

electron/services/learning-agent-client.ts

@@ -4,7 +4,15 @@ import { proxyAwareFetch } from '../utils/proxy-fetch';
 import { getValidWorksSquareAccessToken } from './works-square-session';
 
 const COURSE_ID_PATTERN = /^[A-Za-z0-9_-]{1,64}$/;
+const MODULE_ID_PATTERN = /^(?!\.{1,2}$)[A-Za-z0-9._-]{1,128}$/;
 const SHA256_PATTERN = /^[0-9a-f]{64}$/;
+const REMOTE_ID_PATTERN = /^[A-Za-z0-9_.:-]{1,128}$/;
+const MAX_MESSAGE_LENGTH = 4_000;
+const MAX_HISTORY_ITEMS = 8;
+const MAX_HISTORY_CONTENT_LENGTH = 4_000;
+const MAX_SCENE_TEXT_LENGTH = 256;
+const MAX_ANSWER_LENGTH = 64_000;
+const MAX_SSE_BUFFER_LENGTH = 256_000;
 
 export type LearningAgentRequest = {
   courseId: string;
@@ -38,11 +46,20 @@ function record(value: unknown): Record<string, unknown> {
   return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
 }
 
+function isLearningAccountChanged(error: unknown): boolean {
+  return record(error).code === 'LEARNING_ACCOUNT_CHANGED';
+}
+
 function createSseDataParser() {
   let line = '';
   let pendingCarriageReturn = false;
   let dataLines: string[] = [];
 
+  const assertBufferSize = () => {
+    const size = line.length + dataLines.reduce((total, data) => total + data.length, 0);
+    if (size > MAX_SSE_BUFFER_LENGTH) throw new Error('助教服务暂时不可用');
+  };
+
   const commitLine = (events: string[]) => {
     if (line === '') {
       if (dataLines.length > 0) events.push(dataLines.join('\n'));
@@ -54,7 +71,10 @@ function createSseDataParser() {
       const field = separator === -1 ? line : line.slice(0, separator);
       let value = separator === -1 ? '' : line.slice(separator + 1);
       if (value.startsWith(' ')) value = value.slice(1);
-      if (field === 'data') dataLines.push(value);
+      if (field === 'data') {
+        dataLines.push(value);
+        assertBufferSize();
+      }
     }
     line = '';
   };
@@ -73,6 +93,7 @@ function createSseDataParser() {
         commitLine(events);
       } else {
         line += character;
+        assertBufferSize();
       }
     }
     return events;
@@ -101,32 +122,53 @@ export function createLearningAgentClient(dependencies: Dependencies = {}) {
   const apiBaseUrl = (dependencies.apiBaseUrl ?? WORKS_SQUARE_CONFIG.apiBaseUrl).replace(/\/+$/, '');
   const sessions = new Map<string, string>();
   const sessionFlights = new Map<string, Promise<string>>();
+  const turnTails = new Map<string, Promise<void>>();
 
-  async function authorizedJson(path: string, init: RequestInit): Promise<Record<string, unknown>> {
-    const token = await getAccessToken({ fetchImpl });
-    if (!token) throw new Error('请先登录');
-    const request = (accessToken: string) => fetchImpl(`${apiBaseUrl}${path}`, {
-      ...init,
-      headers: { Accept: 'application/json', 'Content-Type': 'application/json', Authorization: `Bearer ${accessToken}` },
-    });
-    let response = await request(token);
-    if (response.status === 401) {
-      await response.body?.cancel().catch(() => undefined);
-      const refreshed = await getAccessToken({ fetchImpl, forceRefresh: true });
-      if (refreshed) response = await request(refreshed);
+  async function authorizedJson(
+    path: string,
+    init: RequestInit,
+    assertCurrentAccount: () => void,
+  ): Promise<Record<string, unknown>> {
+    assertCurrentAccount();
+    try {
+      const token = await getAccessToken({ fetchImpl });
+      assertCurrentAccount();
+      if (!token) throw new Error('请先登录');
+      const request = (accessToken: string) => {
+        assertCurrentAccount();
+        return fetchImpl(`${apiBaseUrl}${path}`, {
+          ...init,
+          headers: { Accept: 'application/json', 'Content-Type': 'application/json', Authorization: `Bearer ${accessToken}` },
+        });
+      };
+      let response = await request(token);
+      assertCurrentAccount();
+      if (response.status === 401) {
+        await response.body?.cancel().catch(() => undefined);
+        assertCurrentAccount();
+        const refreshed = await getAccessToken({ fetchImpl, forceRefresh: true });
+        assertCurrentAccount();
+        if (refreshed) {
+          response = await request(refreshed);
+          assertCurrentAccount();
+        }
+      }
+      const payload = await response.json().catch(() => null);
+      if (!response.ok) throw new LearningAgentHttpError(response.status, '助教服务暂时不可用');
+      return record(payload);
+    } catch (error) {
+      if (isLearningAccountChanged(error)) throw error;
+      if (error instanceof LearningAgentHttpError || (error instanceof Error && error.message === '请先登录')) throw error;
+      throw new Error('助教服务暂时不可用', { cause: error });
     }
-    const payload = await response.json().catch(() => null);
-    if (!response.ok) {
-      const detail = record(record(payload).detail);
-      throw new LearningAgentHttpError(
-        response.status,
-        typeof detail.message === 'string' ? detail.message : '助教服务暂时不可用',
-      );
-    }
-    return record(payload);
   }
 
-  async function createSession(bindingKey: string, courseId: string): Promise<string> {
+  async function createSession(
+    bindingKey: string,
+    courseId: string,
+    assertCurrentAccount: () => void,
+  ): Promise<string> {
+    assertCurrentAccount();
     const existing = sessions.get(bindingKey);
     if (existing) return existing;
     const inFlight = sessionFlights.get(bindingKey);
@@ -139,9 +181,12 @@ export function createLearningAgentClient(dependencies: Dependencies = {}) {
         runtime_version: 'v1',
         binding: { kind: 'learning-course', key: bindingKey },
       }),
-    }).then((session) => {
-      const sessionId = String(session.session_id || '');
-      if (!sessionId) throw new Error('助教会话创建失败');
+    }, assertCurrentAccount).then((session) => {
+      assertCurrentAccount();
+      const sessionId = typeof session.session_id === 'string' && REMOTE_ID_PATTERN.test(session.session_id)
+        ? session.session_id
+        : '';
+      if (!sessionId) throw new Error('助教服务暂时不可用');
       sessions.set(bindingKey, sessionId);
       return sessionId;
     }).finally(() => sessionFlights.delete(bindingKey));
@@ -149,36 +194,88 @@ export function createLearningAgentClient(dependencies: Dependencies = {}) {
     return pending;
   }
 
-  async function runTurn(sessionId: string, input: LearningAgentRequest): Promise<{ text: string }> {
+  function safeTurnInput(input: LearningAgentRequest) {
+    if (!input || typeof input.message !== 'string') throw new Error('助教请求无效');
+    const message = input.message.trim();
+    const history = Array.isArray(input.history) ? input.history.slice(-MAX_HISTORY_ITEMS).map((item) => {
+      if ((item?.role !== 'user' && item?.role !== 'assistant')
+        || typeof item.content !== 'string'
+        || !item.content.trim()
+        || item.content.length > MAX_HISTORY_CONTENT_LENGTH) throw new Error('助教请求无效');
+      return { role: item.role, content: item.content };
+    }) : [];
+    const anchor = input.anchor ?? {};
+    if ((anchor.sceneId !== undefined && (typeof anchor.sceneId !== 'string' || !anchor.sceneId || anchor.sceneId.length > MAX_SCENE_TEXT_LENGTH))
+      || (anchor.sceneTitle !== undefined && (typeof anchor.sceneTitle !== 'string' || anchor.sceneTitle.length > MAX_SCENE_TEXT_LENGTH))
+      || (anchor.sceneOrder !== undefined && (!Number.isSafeInteger(anchor.sceneOrder) || anchor.sceneOrder < 0 || anchor.sceneOrder > 100_000))
+      || (anchor.actionIndex !== undefined && (!Number.isSafeInteger(anchor.actionIndex) || anchor.actionIndex < 0 || anchor.actionIndex > 100_000))
+      || (anchor.moduleId !== undefined && anchor.moduleId !== null && !MODULE_ID_PATTERN.test(anchor.moduleId))
+      || (anchor.moduleContentHash !== undefined && !SHA256_PATTERN.test(anchor.moduleContentHash))) {
+      throw new Error('助教请求无效');
+    }
+    return {
+      message,
+      history,
+      anchor: {
+        ...(anchor.sceneId === undefined ? {} : { sceneId: anchor.sceneId }),
+        ...(anchor.sceneOrder === undefined ? {} : { sceneOrder: anchor.sceneOrder }),
+        ...(anchor.sceneTitle === undefined ? {} : { sceneTitle: anchor.sceneTitle }),
+        ...(anchor.actionIndex === undefined ? {} : { actionIndex: anchor.actionIndex }),
+        ...(anchor.moduleId === undefined ? {} : { moduleId: anchor.moduleId }),
+        ...(anchor.moduleContentHash === undefined ? {} : { moduleContentHash: anchor.moduleContentHash }),
+      },
+    };
+  }
+
+  async function runTurn(
+    sessionId: string,
+    input: ReturnType<typeof safeTurnInput>,
+    assertCurrentAccount: () => void,
+  ): Promise<{ text: string }> {
+    assertCurrentAccount();
     const command = await authorizedJson(`/api/agents/sessions/${encodeURIComponent(sessionId)}/commands`, {
       method: 'POST',
       body: JSON.stringify({
         client_command_id: `turn-${randomUUID()}`,
         name: 'turn.submit',
         input: {
-          message: input.message.trim().slice(0, 4000),
+          message: input.message,
           // Compatibility context for an older runtime. The durable Works
           // Agent session remains authoritative across turns.
-          history: (input.history ?? []).slice(-8),
-          anchor: input.anchor ?? {},
+          history: input.history,
+          anchor: input.anchor,
         },
       }),
-    });
-    const runId = String(command.run_id || '');
+    }, assertCurrentAccount);
+    const runId = typeof command.run_id === 'string' && REMOTE_ID_PATTERN.test(command.run_id) ? command.run_id : '';
     const ticket = await authorizedJson(`/api/agents/sessions/${encodeURIComponent(sessionId)}/stream-tickets`, {
       method: 'POST',
       body: JSON.stringify({ transport: 'sse' }),
-    });
-    const streamUrl = String(ticket.stream_url || '');
-    if (!runId || !streamUrl.startsWith('/api/agents/')) throw new Error('助教事件通道创建失败');
+    }, assertCurrentAccount);
+    const streamUrl = typeof ticket.stream_url === 'string' && ticket.stream_url.length <= 2_048
+      ? ticket.stream_url
+      : '';
+    const expectedStreamPath = `/api/agents/sessions/${encodeURIComponent(sessionId)}/events`;
+    let streamTarget: URL | null = null;
+    try {
+      const candidate = new URL(streamUrl, `${apiBaseUrl}/`);
+      const base = new URL(apiBaseUrl);
+      if (candidate.origin === base.origin && candidate.pathname === expectedStreamPath) streamTarget = candidate;
+    } catch {
+      streamTarget = null;
+    }
+    if (!runId || !streamTarget) throw new Error('助教服务暂时不可用');
     const controller = new AbortController();
     const timeout = setTimeout(() => controller.abort(), 120_000);
     let response: Response;
     try {
-      response = await fetchImpl(`${apiBaseUrl}${streamUrl}`, { headers: { Accept: 'text/event-stream' }, signal: controller.signal });
+      assertCurrentAccount();
+      response = await fetchImpl(streamTarget, { headers: { Accept: 'text/event-stream' }, signal: controller.signal });
+      assertCurrentAccount();
     } catch (error) {
       clearTimeout(timeout);
-      throw error;
+      if (isLearningAccountChanged(error)) throw error;
+      throw new Error('助教服务暂时不可用', { cause: error });
     }
     if (!response.ok || !response.body) {
       clearTimeout(timeout);
@@ -190,23 +287,32 @@ export function createLearningAgentClient(dependencies: Dependencies = {}) {
     let text = '';
     const consume = (data: string): { text?: string; completed?: true } => {
       if (!data) return {};
-      const envelope = record(JSON.parse(data));
+      let envelope: Record<string, unknown>;
+      try {
+        envelope = record(JSON.parse(data));
+      } catch {
+        throw new Error('助教服务暂时不可用');
+      }
       if (envelope.run_id !== runId) return {};
       const payload = record(envelope.payload);
       if (envelope.type === 'learning.assistant.delta' && typeof payload.delta === 'string') {
+        if (text.length + payload.delta.length > MAX_ANSWER_LENGTH) throw new Error('助教服务暂时不可用');
         text += payload.delta;
       }
       if (envelope.type === 'learning.assistant.failed') {
-        throw new Error(typeof payload.message === 'string' ? payload.message : '助教回答失败');
+        throw new Error('助教回答失败');
       }
       if (envelope.type === 'learning.assistant.completed') {
-        return { text: text.trim(), completed: true };
+        const answer = text.trim();
+        if (!answer) throw new Error('助教回答失败');
+        return { text: answer, completed: true };
       }
       return {};
     };
     try {
       while (true) {
         const chunk = await reader.read();
+        assertCurrentAccount();
         const events = chunk.done
           ? [...parser.push(decoder.decode()), ...parser.finish()]
           : parser.push(decoder.decode(chunk.value, { stream: true }));
@@ -226,30 +332,59 @@ export function createLearningAgentClient(dependencies: Dependencies = {}) {
     throw new Error('助教回答意外中断');
   }
 
-  async function ask(input: LearningAgentRequest): Promise<{ text: string }> {
+  async function ask(
+    input: LearningAgentRequest,
+    assertCurrentAccount: () => void = () => undefined,
+  ): Promise<{ text: string }> {
+    assertCurrentAccount();
+    const safeInput = safeTurnInput(input);
     if (!COURSE_ID_PATTERN.test(input.courseId)
       || !SHA256_PATTERN.test(input.contentHash)
-      || !input.message.trim()) throw new Error('助教请求无效');
+      || !safeInput.message
+      || safeInput.message.length > MAX_MESSAGE_LENGTH) throw new Error('助教请求无效');
     const bindingKey = `${input.courseId}:${input.contentHash}`;
-    let sessionId = await createSession(bindingKey, input.courseId);
+    const previous = turnTails.get(bindingKey) ?? Promise.resolve();
+    const turn = previous.catch(() => undefined).then(async () => {
+      assertCurrentAccount();
+      let sessionId = await createSession(bindingKey, input.courseId, assertCurrentAccount);
+      try {
+        return await runTurn(sessionId, safeInput, assertCurrentAccount);
+      } catch (error) {
+        if (!(error instanceof LearningAgentHttpError) || error.status !== 404) throw error;
+        assertCurrentAccount();
+        sessions.delete(bindingKey);
+        sessionId = await createSession(bindingKey, input.courseId, assertCurrentAccount);
+        assertCurrentAccount();
+        return runTurn(sessionId, safeInput, assertCurrentAccount);
+      }
+    });
+    const tail = turn.then(() => undefined, () => undefined);
+    turnTails.set(bindingKey, tail);
     try {
-      return await runTurn(sessionId, input);
-    } catch (error) {
-      if (!(error instanceof LearningAgentHttpError) || error.status !== 404) throw error;
-      sessions.delete(bindingKey);
-      sessionId = await createSession(bindingKey, input.courseId);
-      return runTurn(sessionId, input);
+      return await turn;
+    } finally {
+      if (turnTails.get(bindingKey) === tail) turnTails.delete(bindingKey);
     }
   }
 
-  async function reset(courseId?: string, contentHash?: string): Promise<void> {
+  async function reset(
+    courseId?: string,
+    contentHash?: string,
+    assertCurrentAccount: () => void = () => undefined,
+  ): Promise<void> {
+    assertCurrentAccount();
     const exactKey = courseId && contentHash ? `${courseId}:${contentHash}` : null;
     const entries = [...sessions.entries()].filter(([key]) => exactKey
       ? key === exactKey
       : courseId ? key.startsWith(`${courseId}:`) : true);
     for (const [key, sessionId] of entries) {
+      assertCurrentAccount();
       sessions.delete(key);
-      void authorizedJson(`/api/agents/sessions/${encodeURIComponent(sessionId)}`, { method: 'DELETE' }).catch(() => undefined);
+      void authorizedJson(
+        `/api/agents/sessions/${encodeURIComponent(sessionId)}`,
+        { method: 'DELETE' },
+        assertCurrentAccount,
+      ).catch(() => undefined);
     }
   }