feat(auth): add remembered local login to the frontend

frontend/src/stream/useSessionStream.test.tsx

@@ -1,13 +1,17 @@
-import { renderHook } from "@testing-library/react";
+import { renderHook, waitFor } from "@testing-library/react";
 import { act, useLayoutEffect } from "react";
 import { FakeEventSource } from "../test/fakeEventSource";
 import { createStreamEventCoalescer, useSessionStream } from "./useSessionStream";
 import { useSessionStore } from "../store/sessionStore";
+import { clearAuthState, getAuthGeneration, getAuthState, setAuthState } from "../auth/authState";
+import { http, HttpResponse } from "msw";
+import { server } from "../test/msw";
 
 beforeEach(() => {
   FakeEventSource.instances = [];
   (globalThis as any).EventSource = FakeEventSource;
   useSessionStore.getState().resetSession();
+  clearAuthState();
 });
 
 test("coalesces a burst of text deltas into bounded store updates", () => {
@@ -32,6 +36,73 @@ test("coalesces a burst of text deltas into bounded store updates", () => {
   }
 });
 
+test("disposal discards buffered deltas instead of flushing them", () => {
+  vi.useFakeTimers();
+  try {
+    const applied: Array<{ type: string; text?: string }> = [];
+    const coalescer = createStreamEventCoalescer((event) => applied.push(event), 100);
+
+    coalescer.push({ type: "text_delta", text: "a" });
+    coalescer.push({ type: "text_delta", text: "b" });
+    coalescer.dispose();
+    vi.advanceTimersByTime(100);
+
+    expect(applied).toEqual([{ type: "text_delta", text: "a" }]);
+  } finally {
+    vi.useRealTimers();
+  }
+});
+
+test("a stale coalescer timer checks source validity before applying text", () => {
+  vi.useFakeTimers();
+  try {
+    let current = true;
+    const applied: Array<{ type: string; text?: string }> = [];
+    const coalescer = createStreamEventCoalescer((event) => applied.push(event), 100, () => current);
+
+    coalescer.push({ type: "text_delta", text: "a" });
+    coalescer.push({ type: "text_delta", text: "b" });
+    current = false;
+    vi.advanceTimersByTime(100);
+
+    expect(applied).toEqual([{ type: "text_delta", text: "a" }]);
+    coalescer.dispose();
+  } finally {
+    vi.useRealTimers();
+  }
+});
+
+test("buffered user-A text cannot reappear after logout and user-B login", () => {
+  vi.useFakeTimers();
+  try {
+    const userA = {
+      issuer: "local", subject: "user-a", roles: ["user"] as const, permissions: ["session.use"], isAdmin: false,
+      csrfToken: "a".repeat(43), session: null,
+    };
+    const userB = { ...userA, subject: "user-b", csrfToken: "b".repeat(43) };
+    server.use(http.get("/api/me", () => new HttpResponse(null, { status: 401 })));
+    setAuthState(userA);
+    const { rerender } = renderHook(
+      ({ authGeneration }) => useSessionStream("s1", 0, 0, true, authGeneration),
+      { initialProps: { authGeneration: getAuthGeneration() } },
+    );
+    const sourceA = FakeEventSource.instances[0];
+    act(() => {
+      sourceA.emit({ type: "text_delta", text: "A-one" });
+      sourceA.emit({ type: "text_delta", text: "A-two" });
+      sourceA.onerror?.(new Event("error"));
+      setAuthState(userB);
+    });
+    rerender({ authGeneration: getAuthGeneration() });
+    vi.advanceTimersByTime(1000);
+
+    expect(useSessionStore.getState().transcript).toEqual([]);
+    expect(FakeEventSource.instances[0].closed).toBe(true);
+  } finally {
+    vi.useRealTimers();
+  }
+});
+
 test("flushes pending stream text before a structural event", () => {
   vi.useFakeTimers();
   try {
@@ -280,3 +351,130 @@ test("the old source is invalid before later layout effects can deliver a queued
   expect(first.closed).toBe(true);
   expect(useSessionStore.getState().transcript).toEqual([]);
 });
+
+test("auth-generation changes close the old source, reset live state, and do not reconnect after logout", () => {
+  setAuthState({
+    issuer: "local", subject: "user-1", roles: ["user"], permissions: ["session.use"], isAdmin: false,
+    csrfToken: "c".repeat(43), session: null,
+  });
+  const { rerender } = renderHook(
+    ({ authenticated, authGeneration }) => useSessionStream("s1", 0, 0, authenticated, authGeneration),
+    { initialProps: { authenticated: true, authGeneration: getAuthGeneration() } },
+  );
+  const first = FakeEventSource.instances[0];
+  act(() => first.emitNamed("text_delta", { type: "text_delta", text: "stale" }));
+  expect(useSessionStore.getState().transcript).toHaveLength(1);
+
+  clearAuthState();
+  rerender({ authenticated: false, authGeneration: getAuthGeneration() });
+
+  expect(first.closed).toBe(true);
+  expect(FakeEventSource.instances).toHaveLength(1);
+  expect(useSessionStore.getState().transcript).toEqual([]);
+});
+
+test("a stale source error cannot clear or mutate the next user's live session", async () => {
+  const userA = {
+    issuer: "local", subject: "user-a", roles: ["user"] as const, permissions: ["session.use"], isAdmin: false,
+    csrfToken: "a".repeat(43), session: null,
+  };
+  const userB = { ...userA, subject: "user-b", csrfToken: "b".repeat(43) };
+  setAuthState(userA);
+  const { rerender } = renderHook(
+    ({ authenticated, authGeneration }) => useSessionStream("s1", 0, 0, authenticated, authGeneration),
+    { initialProps: { authenticated: true, authGeneration: getAuthGeneration() } },
+  );
+  const sourceA = FakeEventSource.instances[0];
+
+  act(() => setAuthState(userB));
+  rerender({ authenticated: true, authGeneration: getAuthGeneration() });
+  const sourceB = FakeEventSource.instances[1];
+  act(() => sourceB.emitNamed("text_delta", { type: "text_delta", text: "B-data" }));
+
+  act(() => sourceA.onerror?.(new Event("error")));
+
+  expect((await import("../auth/authState")).getAuthState()).toMatchObject({ subject: "user-b" });
+  expect(sourceB.closed).toBe(false);
+  expect(FakeEventSource.instances).toHaveLength(2);
+  expect(useSessionStore.getState().transcript).toEqual([{ role: "assistant", text: "B-data" }]);
+});
+
+test("a native current-source error probes /me once and clears auth on a guarded 401", async () => {
+  let probes = 0;
+  server.use(http.get("/api/me", () => {
+    probes += 1;
+    return new HttpResponse(null, { status: 401 });
+  }));
+  setAuthState({
+    issuer: "local", subject: "user-a", roles: ["user"], permissions: ["session.use"], isAdmin: false,
+    csrfToken: "a".repeat(43), session: null,
+  });
+  renderHook(() => useSessionStream("s1", 0, 0, true, getAuthGeneration()));
+  const source = FakeEventSource.instances[0];
+
+  act(() => {
+    source.onerror?.(new Event("error"));
+    source.onerror?.(new Event("error"));
+  });
+  await waitFor(() => expect(getAuthState()).toBeNull());
+
+  expect(probes).toBe(1);
+  expect((await import("../auth/authState")).getAuthState()).toBeNull();
+});
+
+test("a native error keeps current auth connected after a 200 /me probe and does not storm", async () => {
+  let probes = 0;
+  const user = {
+    issuer: "local", subject: "user-a", roles: ["user"] as const, permissions: ["session.use"], isAdmin: false,
+    csrfToken: "a".repeat(43), session: null,
+  };
+  server.use(http.get("/api/me", () => {
+    probes += 1;
+    return HttpResponse.json(user);
+  }));
+  setAuthState(user);
+  const result = renderHook(() => useSessionStream("s1", 0, 0, true, getAuthGeneration()));
+  const source = FakeEventSource.instances[0];
+
+  act(() => source.onerror?.(new Event("error")));
+  await waitFor(() => expect(result.result.current.connected).toBe(true));
+  act(() => source.onerror?.(new Event("error")));
+
+  expect(probes).toBe(1);
+  expect(getAuthState()).toMatchObject({ subject: "user-a" });
+});
+
+test("a stale source does not probe and a user-B login during an A probe cannot mutate B", async () => {
+  let release!: () => void;
+  let probes = 0;
+  const held = new Promise<void>((resolve) => { release = resolve; });
+  const userA = {
+    issuer: "local", subject: "user-a", roles: ["user"] as const, permissions: ["session.use"], isAdmin: false,
+    csrfToken: "a".repeat(43), session: null,
+  };
+  const userB = { ...userA, subject: "user-b", csrfToken: "b".repeat(43) };
+  server.use(http.get("/api/me", async () => {
+    probes += 1;
+    await held;
+    return HttpResponse.json(userA);
+  }));
+  setAuthState(userA);
+  const { rerender } = renderHook(
+    ({ sessionId, authGeneration }) => useSessionStream(sessionId, 0, 0, true, authGeneration),
+    { initialProps: { sessionId: "s1" as string | null, authGeneration: getAuthGeneration() } },
+  );
+  const sourceA = FakeEventSource.instances[0];
+  act(() => sourceA.onerror?.(new Event("error")));
+  await waitFor(() => expect(probes).toBe(1));
+
+  act(() => setAuthState(userB));
+  rerender({ sessionId: "s2", authGeneration: getAuthGeneration() });
+  const sourceB = FakeEventSource.instances[1];
+  act(() => sourceA.onerror?.(new Event("error")));
+  release();
+  await new Promise((resolve) => setTimeout(resolve, 0));
+
+  expect(probes).toBe(1);
+  expect(getAuthState()).toMatchObject({ subject: "user-b" });
+  expect(sourceB.closed).toBe(false);
+});

frontend/src/stream/useSessionStream.ts

@@ -3,6 +3,12 @@ import { BASE } from "../api/client";
 import { joinBackendPath } from "../api/runtime-config";
 import { useSessionStore } from "../store/sessionStore";
 import type { StreamEvent } from "../api/types";
+import {
+  getAuthGeneration,
+  isAuthGenerationCurrent,
+} from "../auth/authState";
+import { getMe } from "../api/auth";
+import { captureAuthOperation, isAuthOperationCurrent } from "../auth/authOperation";
 
 const STREAM_UPDATE_INTERVAL_MS = 100;
 
@@ -15,6 +21,7 @@ function isStreamingDelta(
 export function createStreamEventCoalescer(
   applyEvent: (event: StreamEvent) => void,
   intervalMs = STREAM_UPDATE_INTERVAL_MS,
+  canApply: () => boolean = () => true,
 ) {
   let pending: Extract<StreamEvent, { type: "text_delta" | "activity_delta" }> | null = null;
   let timer: ReturnType<typeof setTimeout> | null = null;
@@ -24,6 +31,7 @@ export function createStreamEventCoalescer(
     if (!pending) return;
     const event = pending;
     pending = null;
+    if (disposed || !canApply()) return;
     applyEvent(event);
   };
 
@@ -31,14 +39,18 @@ export function createStreamEventCoalescer(
     timer = setTimeout(() => {
       timer = null;
       if (!pending || disposed) return;
+      if (!canApply()) {
+        pending = null;
+        return;
+      }
       flushPending();
-      schedule();
+      if (pending && !disposed && canApply()) schedule();
     }, intervalMs);
   };
 
   return {
     push(event: StreamEvent) {
-      if (disposed) return;
+      if (disposed || !canApply()) return;
       if (!isStreamingDelta(event)) {
         flushPending();
         if (timer) clearTimeout(timer);
@@ -60,7 +72,7 @@ export function createStreamEventCoalescer(
       if (disposed) return;
       if (timer) clearTimeout(timer);
       timer = null;
-      flushPending();
+      pending = null;
       disposed = true;
     },
   };
@@ -70,24 +82,33 @@ export function useSessionStream(
   sessionId: string | null,
   generation = 0,
   cursorResetEpoch = 0,
+  authenticated = true,
+  authGeneration?: number,
 ) {
+  const effectiveAuthGeneration = authGeneration ?? getAuthGeneration();
   const [connected, setConnected] = useState(false);
   const applyEvent = useSessionStore((s) => s.applyEvent);
+  const resetSession = useSessionStore((s) => s.resetSession);
   const cursor = useRef({
     sessionId: null as string | null,
     cursorResetEpoch,
+    authGeneration: null as number | null,
     lastEventId: "",
   });
   const activeSource = useRef<object | null>(null);
 
   useLayoutEffect(() => {
+    const authChanged = cursor.current.authGeneration !== null
+      && cursor.current.authGeneration !== effectiveAuthGeneration;
     if (
       cursor.current.sessionId !== sessionId
       || cursor.current.cursorResetEpoch !== cursorResetEpoch
+      || authChanged
     ) {
-      cursor.current = { sessionId, cursorResetEpoch, lastEventId: "" };
+      cursor.current = { sessionId, cursorResetEpoch, authGeneration: effectiveAuthGeneration, lastEventId: "" };
+      if (authChanged) resetSession();
     }
-    if (!sessionId) {
+    if (!sessionId || !authenticated) {
       activeSource.current = null;
       setConnected(false);
       return;
@@ -96,19 +117,40 @@ export function useSessionStream(
     const query = cursor.current.lastEventId
       ? `?lastEventId=${encodeURIComponent(cursor.current.lastEventId)}`
       : "";
+    const sourceAuthGeneration = effectiveAuthGeneration;
     const es = new EventSource(joinBackendPath(BASE, `/sessions/${sessionId}/events${query}`));
     const identity = { source: es, sessionId, cursorResetEpoch };
-    const coalescer = createStreamEventCoalescer(applyEvent);
+    let authProbeStarted = false;
+    const isCurrentSource = () => activeSource.current === identity
+      && isAuthGenerationCurrent(sourceAuthGeneration);
+    const probeAuth = async () => {
+      if (authProbeStarted || !isCurrentSource()) return;
+      authProbeStarted = true;
+      const operation = captureAuthOperation({ sessionId, disposalEpoch: 0 });
+      if (!operation) return;
+      try {
+        await getMe();
+        if (!isCurrentSource() || !isAuthOperationCurrent(operation, { sessionId, disposalEpoch: 0 })) return;
+        setConnected(true);
+      } catch {
+        if (!isCurrentSource() || !isAuthOperationCurrent(operation, { sessionId, disposalEpoch: 0 })) return;
+        setConnected(false);
+      }
+    };
+    const coalescer = createStreamEventCoalescer(applyEvent, STREAM_UPDATE_INTERVAL_MS, isCurrentSource);
     activeSource.current = identity;
     es.onopen = () => {
-      if (activeSource.current === identity) setConnected(true);
+      if (!isCurrentSource()) { es.close(); return; }
+      setConnected(true);
     };
     es.onerror = () => {
-      if (activeSource.current === identity) setConnected(false);
+      if (!isCurrentSource()) { es.close(); return; }
+      setConnected(false);
+      void probeAuth();
     };
 
     const handle = (ev: MessageEvent) => {
-      if (activeSource.current !== identity) return;
+      if (!isCurrentSource()) return;
       if (ev.lastEventId) cursor.current.lastEventId = ev.lastEventId;
       try {
         coalescer.push(JSON.parse(ev.data) as StreamEvent);
@@ -138,10 +180,12 @@ export function useSessionStream(
       coalescer.dispose();
       if (activeSource.current === identity) {
         activeSource.current = null;
-        setConnected(false);
+        if (getAuthGeneration() === sourceAuthGeneration) {
+          setConnected(false);
+        }
       }
     };
-  }, [sessionId, generation, cursorResetEpoch, applyEvent]);
+  }, [sessionId, generation, cursorResetEpoch, authenticated, effectiveAuthGeneration, applyEvent, resetSession]);
 
   return { connected };
 }