perf(sessions): faster task open, less render churn and memory

apps/code/src/main/di/container.ts

@@ -711,6 +711,8 @@ container.bind(LOGS_SERVICE).toDynamicValue((ctx) => {
     },
     readLocalLogs: (taskRunId: string) =>
       ws.localLogs.read.query({ taskRunId }),
+    readLocalLogsTail: (taskRunId: string, maxBytes: number) =>
+      ws.localLogs.readTail.query({ taskRunId, maxBytes }),
     writeLocalLogs: (taskRunId: string, content: string) =>
       ws.localLogs.write.mutate({ taskRunId, content }),
   };

packages/core/src/sessions/sessionEventBatching.test.ts

@@ -0,0 +1,167 @@
+import type { AcpMessage, AgentSession } from "@posthog/shared";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { SessionService, type SessionServiceDeps } from "./sessionService";
+
+const TASK_ID = "task-1";
+const RUN_ID = "run-1";
+const FLUSH_MS = 16;
+
+/** A plain streamed agent-message chunk — the common per-token event that just
+ * gets appended to the transcript. */
+function chunk(text: string): AcpMessage {
+  return {
+    ts: 1,
+    message: {
+      jsonrpc: "2.0",
+      method: "session/update",
+      params: {
+        sessionId: RUN_ID,
+        update: {
+          sessionUpdate: "agent_message_chunk",
+          content: { type: "text", text },
+        },
+      },
+    },
+  } as unknown as AcpMessage;
+}
+
+function chunkText(event: AcpMessage): string {
+  const params = (event.message as { params?: unknown }).params as {
+    update: { content: { text: string } };
+  };
+  return params.update.content.text;
+}
+
+function createHarness() {
+  const sessions: Record<string, AgentSession> = {
+    [RUN_ID]: {
+      taskRunId: RUN_ID,
+      taskId: TASK_ID,
+      events: [],
+      messageQueue: [],
+      pendingPermissions: new Map(),
+      status: "connected",
+    } as unknown as AgentSession,
+  };
+
+  const appendEvents = vi.fn(
+    (taskRunId: string, events: AcpMessage[], newLineCount?: number) => {
+      const session = sessions[taskRunId];
+      if (!session) return;
+      session.events = [...session.events, ...events];
+      if (newLineCount !== undefined) session.processedLineCount = newLineCount;
+    },
+  );
+
+  const store = {
+    getSessions: () => sessions,
+    getSessionByTaskId: (taskId: string) =>
+      Object.values(sessions).find((s) => s.taskId === taskId),
+    setSession: (session: AgentSession) => {
+      sessions[session.taskRunId] = session;
+    },
+    updateSession: (taskRunId: string, updates: Partial<AgentSession>) => {
+      const session = sessions[taskRunId];
+      if (session) Object.assign(session, updates);
+    },
+    appendEvents,
+    replaceOptimisticWithEvent: vi.fn(),
+    setPendingPermissions: vi.fn(),
+    clearMessageQueue: vi.fn(),
+    clearTailOptimisticItems: vi.fn(),
+    appendOptimisticItem: vi.fn(),
+  };
+
+  let onEvent: ((payload: unknown) => void) | undefined;
+  const noopLog = {
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    debug: vi.fn(),
+  };
+
+  const deps = {
+    store,
+    log: noopLog,
+    notifyPromptComplete: vi.fn(),
+    notifyPermissionRequest: vi.fn(),
+    taskViewedApi: { markActivity: vi.fn() },
+    getPersistedConfigOptions: () => undefined,
+    setPersistedConfigOptions: vi.fn(),
+    trpc: {
+      agent: {
+        onSessionEvent: {
+          subscribe: (
+            _input: unknown,
+            handlers: { onData: (payload: unknown) => void },
+          ) => {
+            onEvent = handlers.onData;
+            return { unsubscribe: vi.fn() };
+          },
+        },
+        onPermissionRequest: {
+          subscribe: () => ({ unsubscribe: vi.fn() }),
+        },
+        onSessionIdleKilled: {
+          subscribe: () => ({ unsubscribe: vi.fn() }),
+        },
+      },
+    },
+  } as unknown as SessionServiceDeps;
+
+  const service = new SessionService(deps);
+  // Register the streamed-event subscription (captures onData).
+  (
+    service as unknown as { subscribeToChannel(id: string): void }
+  ).subscribeToChannel(RUN_ID);
+  if (!onEvent)
+    throw new Error("subscribeToChannel did not subscribe to events");
+
+  return {
+    service,
+    appendEvents,
+    emit: (event: AcpMessage) => onEvent?.(event),
+    events: () => sessions[RUN_ID].events,
+  };
+}
+
+describe("streamed event batching", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("defers a burst and applies it on one flush tick, in order", () => {
+    const h = createHarness();
+
+    h.emit(chunk("a"));
+    h.emit(chunk("b"));
+    h.emit(chunk("c"));
+
+    // Nothing is applied synchronously — the burst is buffered.
+    expect(h.appendEvents).not.toHaveBeenCalled();
+    expect(h.events()).toHaveLength(0);
+
+    // A single flush tick drains the whole burst, in arrival order.
+    vi.advanceTimersByTime(FLUSH_MS);
+    expect(h.events().map(chunkText)).toEqual(["a", "b", "c"]);
+  });
+
+  it("flushes buffered events synchronously on teardown", () => {
+    const h = createHarness();
+
+    h.emit(chunk("a"));
+    h.emit(chunk("b"));
+    expect(h.events()).toHaveLength(0);
+
+    // reset() tears down subscriptions and must not drop buffered events.
+    h.service.reset();
+    expect(h.events().map(chunkText)).toEqual(["a", "b"]);
+
+    // The flush timer was cleared, so advancing does not re-apply anything.
+    vi.advanceTimersByTime(FLUSH_MS);
+    expect(h.events()).toHaveLength(2);
+  });
+});

packages/core/src/sessions/sessionEventResidency.test.ts

@@ -0,0 +1,141 @@
+import type { AgentSession, SessionStatus } from "@posthog/shared";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { SessionService, type SessionServiceDeps } from "./sessionService";
+import { sessionStore, sessionStoreSetters } from "./sessionStore";
+
+const RUN = "run-res";
+const TASK = "task-res";
+const GRACE_MS = 20_000;
+
+const LOG_LINE = JSON.stringify({
+  type: "notification",
+  notification: {
+    method: "session/update",
+    params: {
+      sessionId: RUN,
+      update: {
+        sessionUpdate: "agent_message_chunk",
+        content: { type: "text", text: "restored" },
+      },
+    },
+  },
+});
+
+function makeService(readLocalLogs = vi.fn().mockResolvedValue("")) {
+  const deps = {
+    store: sessionStoreSetters,
+    log: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
+    notifyPromptComplete: vi.fn(),
+    notifyPermissionRequest: vi.fn(),
+    taskViewedApi: { markActivity: vi.fn() },
+    getPersistedConfigOptions: () => undefined,
+    setPersistedConfigOptions: vi.fn(),
+    trpc: {
+      agent: {
+        onSessionEvent: { subscribe: () => ({ unsubscribe: vi.fn() }) },
+        onPermissionRequest: { subscribe: () => ({ unsubscribe: vi.fn() }) },
+        onSessionIdleKilled: { subscribe: () => ({ unsubscribe: vi.fn() }) },
+      },
+      logs: { readLocalLogs: { query: readLocalLogs } },
+    },
+  } as unknown as SessionServiceDeps;
+  return new SessionService(deps);
+}
+
+function seed(status: SessionStatus, isPromptPending = false) {
+  sessionStoreSetters.setSession({
+    taskRunId: RUN,
+    taskId: TASK,
+    events: [],
+    messageQueue: [],
+    pendingPermissions: new Map(),
+    status,
+    isPromptPending,
+  } as unknown as AgentSession);
+  sessionStoreSetters.appendEvents(RUN, [{ ts: 1, message: {} } as never]);
+}
+
+const events = () => sessionStore.getState().sessions[RUN]?.events ?? [];
+
+describe("session transcript residency", () => {
+  beforeEach(() => vi.useFakeTimers());
+  afterEach(() => {
+    vi.useRealTimers();
+    sessionStoreSetters.removeSession(RUN);
+  });
+
+  it("evicts a disconnected, idle session after the grace window", () => {
+    const service = makeService();
+    seed("disconnected");
+    service.scheduleEventEviction(TASK);
+
+    expect(events()).toHaveLength(1);
+    vi.advanceTimersByTime(GRACE_MS);
+    expect(events()).toHaveLength(0);
+  });
+
+  it("never evicts a connected session", () => {
+    const service = makeService();
+    seed("connected");
+    service.scheduleEventEviction(TASK);
+
+    vi.advanceTimersByTime(GRACE_MS);
+    expect(events()).toHaveLength(1);
+  });
+
+  it("never evicts a session with a prompt in flight", () => {
+    const service = makeService();
+    seed("disconnected", true);
+    service.scheduleEventEviction(TASK);
+
+    vi.advanceTimersByTime(GRACE_MS);
+    expect(events()).toHaveLength(1);
+  });
+
+  it("ensureEventsLoaded cancels a pending eviction", () => {
+    const service = makeService();
+    seed("disconnected");
+    service.scheduleEventEviction(TASK);
+
+    void service.ensureEventsLoaded(TASK); // return to the view before grace
+    vi.advanceTimersByTime(GRACE_MS);
+    expect(events()).toHaveLength(1);
+  });
+
+  it("rehydrates an evicted transcript from disk on return", async () => {
+    const readLocalLogs = vi.fn().mockResolvedValue(LOG_LINE);
+    const service = makeService(readLocalLogs);
+    seed("disconnected");
+
+    service.scheduleEventEviction(TASK);
+    vi.advanceTimersByTime(GRACE_MS);
+    expect(events()).toHaveLength(0);
+
+    await service.ensureEventsLoaded(TASK);
+    expect(readLocalLogs).toHaveBeenCalledWith({ taskRunId: RUN });
+    expect(events()).toHaveLength(1);
+  });
+
+  it("retries rehydration after a failed log read instead of stranding it", async () => {
+    const readLocalLogs = vi
+      .fn()
+      .mockRejectedValueOnce(new Error("transient"))
+      .mockResolvedValueOnce(LOG_LINE);
+    const service = makeService(readLocalLogs);
+    seed("disconnected");
+
+    service.scheduleEventEviction(TASK);
+    vi.advanceTimersByTime(GRACE_MS);
+    expect(events()).toHaveLength(0);
+
+    // First visit: the log read throws — the transcript stays empty but the
+    // run is re-marked evicted so a later visit can retry.
+    await service.ensureEventsLoaded(TASK);
+    expect(events()).toHaveLength(0);
+
+    // Second visit: the read succeeds and the transcript is restored.
+    await service.ensureEventsLoaded(TASK);
+    expect(events()).toHaveLength(1);
+    expect(readLocalLogs).toHaveBeenCalledTimes(2);
+  });
+});

packages/core/src/sessions/sessionEvents.test.ts

@@ -237,6 +237,14 @@ describe("convertStoredEntriesToEvents — imported user prompts", () => {
     const msg = events[0].message;
     expect("method" in msg && msg.method).toBe("session/update");
   });
+
+  it("freezes converted events on both the promoted and raw branches", () => {
+    const events = convertStoredEntriesToEvents([
+      userChunkEntry("promoted", { importedUserPrompt: true }),
+      userChunkEntry("raw"),
+    ]);
+    expect(events.every((event) => Object.isFrozen(event))).toBe(true);
+  });
 });
 
 describe("isAbsoluteFolderPath", () => {

packages/core/src/sessions/sessionEvents.ts

@@ -29,12 +29,14 @@ import { extractPromptDisplayContent } from "./promptContent";
 function storedEntryToAcpMessage(entry: StoredLogEntry): AcpMessage {
   const ts = entry.timestamp ? new Date(entry.timestamp).getTime() : Date.now();
   const promoted = promoteImportedUserPrompt(entry, ts);
-  if (promoted) return promoted;
-  return {
+  // Freeze at creation: events assigned via setSession bypass the store's
+  // per-append freeze, so this keeps them read-only once stored.
+  if (promoted) return Object.freeze(promoted);
+  return Object.freeze({
     type: "acp_message",
     ts,
     message: (entry.notification ?? {}) as JsonRpcMessage,
-  };
+  });
 }
 
 /**