实时状态流 深度解析

  30    const stream = new ReadableStream({
  31      async start(controller) {
  32        let closed = false;
  33        const startedAt = Date.now();
  34        let lastSnapshot = '';
  35        let lastKeepalive = Date.now();
  36  
  37        const send = (data: string) => {
  38          controller.enqueue(encoder.encode(`data: ${data}\n\n`));
  39        };
  40  
  41        const tick = async (): Promise<boolean> => {
  42          // Also tail the Project (title/desc/logo) + its assets so product edits
  43          // made by the worker's qb-designer tools surface live — the worker bumps
  44          // Project.updatedAt / inserts Asset rows, and this snapshot picks it up.
  45          // Also tail the harness tables + harnessRunning flag so pass-3 verdicts and
  46          // the running indicator stream over this same connection (browsers cap ~6
  47          // per origin, so the harness UI shares the chat stream instead of opening
  48          // its own). The worker runs the harness right after the turn settles, so
  49          // the in-flight condition below also keys off harnessRunning to keep this
  50          // connection open across that hand-off.

  51          const [messages, project, assets, requirements] = await Promise.all([
  52            prisma.chatMessage.findMany({
  53              where: { projectId: id },
  54              orderBy: { createdAt: 'asc' },
  55            }),
  56            prisma.project.findUnique({
  57              where: { id },
  58              select: {
  59                title: true,
  60                desc: true,
  61                logo: true,
  62                status: true,
  63                updatedAt: true,
  64                harnessRunning: true,
  65              },
  66            }),
  67            prisma.asset.findMany({
  68              where: { projectId: id },
  69              orderBy: { createdAt: 'asc' },
  70            }),
  71            prisma.harnessRequirement.findMany({
  72              where: { projectId: id },
  73              include: { result: true },
  74              orderBy: { createdAt: 'asc' },
  75            }),
  76          ]);
  77          const harness = {
  78            requirements: requirements.map(serializeRequirement),
  79            results: requirements
  80              .map((r) => r.result)
  81              .filter((r) => r !== null)
  82              .map(serializeResult),
  83            is_running: project?.harnessRunning ?? false,
  84          };
  85          const payload = { messages, project, assets, harness };
  86          const snapshot = JSON.stringify(payload);
  87          if (snapshot !== lastSnapshot) {
  88            lastSnapshot = snapshot;
  89            send(snapshot);
  90            lastKeepalive = Date.now();
  91          } else if (Date.now() - lastKeepalive > KEEPALIVE_MS) {
  92            controller.enqueue(encoder.encode(': keepalive\n\n'));
  93            lastKeepalive = Date.now();
  94          }
  95  
  96          const inFlight =
  97            messages.some((m) => m.status === 'PENDING' || m.status === 'PROCESSING') ||
  98            (project?.harnessRunning ?? false);
  99          return inFlight;
 100        };

  83    /**
  84     * Tail the SSE stream. Uses fetch + a reader (not EventSource) so we control the
  85     * lifecycle — the server self-closes once neither a turn nor the harness is in
  86     * flight, and we don't auto-reconnect while idle.
  87     */
  88    const startStream = useCallback(async () => {
  89      if (abortRef.current) return; // already tailing
  90      const ac = new AbortController();
  91      abortRef.current = ac;
  92      try {
  93        const res = await fetch(`/api/projects/${projectId}/chat/stream`, {
  94          signal: ac.signal,
  95          headers: { Accept: 'text/event-stream' },
  96        });
  97        if (!res.ok || !res.body) return;
  98        const reader = res.body.getReader();
  99        const decoder = new TextDecoder();
 100        let buffer = '';
 101        for (;;) {
 102          const { value, done } = await reader.read();
 103          if (done) break;
 104          buffer += decoder.decode(value, { stream: true });
 105          const events = buffer.split('\n\n');
 106          buffer = events.pop() ?? '';
 107          for (const evt of events) {
 108            const line = evt.split('\n').find((l) => l.startsWith('data:'));
 109            if (!line) continue;
 110            try {
 111              const payload = JSON.parse(line.slice(5).trim());
 112              if (payload.messages) setRows(payload.messages);
 113              if (payload.project) setProject(payload.project);
 114              if (payload.harness) setHarness(payload.harness);
 115            } catch {
 116              /* ignore partial/non-json */
 117            }
 118          }
 119        }
 120      } catch {
 121        /* aborted or network — ignore */
 122      } finally {
 123        if (abortRef.current === ac) abortRef.current = null;
 124        // Settle after the stream closes; if the harness kicked off (or is still
 125        // running) the effect below reopens the stream.
 126        refetch();
 127      }
 128    }, [projectId, refetch]);