Extract bounded activity fetch utilities (codex)

docs/architecture/code-quality-performance-contracts.md

@@ -12,7 +12,7 @@ Use this page when changing hot paths that affect `/api/live`, `/api/execution`,
 | Scoped execution caches | Execution snapshot enrichment must deduplicate task, sprint-run, dispatch, and invocation IDs before rollups and must not introduce per-task follow-up queries. Cache scope and TTL changes belong in the lifecycle cache policy. | `src/repositories/execution/project-execution-snapshot-query.ts`, `src/repositories/execution/execution-usage-query.ts`, `src/repositories/execution/execution-wall-time-query.ts`, `src/repositories/execution/execution-invocations-query.ts`, `src/app/lifecycle/dashboard-snapshot-cache-policy.ts` | `pnpm exec vitest run tests/backend/repositories/execution/project-execution-snapshot-query.test.ts tests/backend/repositories/execution/execution-usage-query.test.ts tests/backend/repositories/execution/execution-wall-time-query.test.ts` |
 | Indexed projection slices | Execution snapshots must load bounded SQL slices for sprint runs, dispatches, runtime events, invocations, attention, usage, and wall time, then merge by stable identifiers in memory. Avoid all-history scans, unbounded `ORDER BY`, JSON-expression indexes, and repeated array filters in task loops. | `src/repositories/execution/execution-sprint-runs-query.ts`, `src/repositories/execution/execution-task-dispatches-query.ts`, `src/repositories/execution/execution-runtime-events-query.ts`, `src/repositories/execution/execution-invocations-query.ts`, `src/repositories/execution/execution-human-intervention-query.ts`, `src/repositories/db/app-db-schema.ts`, `src/repositories/db/app-db-migrations.ts` | `pnpm exec vitest run tests/backend/repositories/execution/execution-snapshot-slice-queries.test.ts tests/backend/repositories/execution/execution-task-dispatches-query.test.ts tests/backend/repositories/execution/project-execution-snapshot-query.test.ts` |
 | Metadata-first provider telemetry | Live provider telemetry must use cheap metadata and source fingerprints before reading full transcripts or provider databases. Final post-process usage collection remains authoritative, and cumulative-session providers must subtract baselines so resumed runs do not re-report earlier usage. | `src/infrastructure/providers/cli/provider-telemetry-watcher.ts`, `src/infrastructure/providers/cli/provider-runner.ts`, `src/infrastructure/providers/cli/provider-usage.ts`, `src/infrastructure/providers/cli/provider-logs/codex-log-parser.ts`, `src/infrastructure/providers/cli/provider-logs/opencode-log-parser.ts`, `src/infrastructure/providers/cli/provider-logs/antigravity-log-parser.ts`, `src/infrastructure/providers/cli/provider-logs/usage-parse-utils.ts`, `src/repositories/execution/provider-invocation-usage-writes.ts`, `src/repositories/execution/execution-invocations-query.ts` | `pnpm exec vitest run tests/backend/infrastructure/providers/cli/provider-telemetry-watcher.test.ts tests/backend/infrastructure/providers/cli/provider-usage.test.ts tests/backend/infrastructure/providers/cli/usage-parse-utils.test.ts` |
-| Bounded activity fetches | Session sync must plan the smallest set of active sessions, skip foreign or locally terminal work where possible, cap per-session activity pages, limit concurrency, and timeout provider reads without failing the whole sync loop. | `src/domain/sprint/session-sync/activity-fetch-plan.ts`, `src/domain/sprint/session-sync/bounded-activity-fetch.ts`, `src/sprint/steps/session-sync-step.ts`, `src/server/activity-cache-service.ts` | `pnpm exec vitest run tests/backend/domain/sprint/session-sync/activity-fetch-plan.test.ts tests/backend/domain/sprint/session-sync/bounded-activity-fetch.test.ts tests/backend/sprint/session-sync-step.test.ts tests/backend/server/activity-cache-service.test.ts` |
+| Bounded activity fetches | Session sync must plan the smallest set of active sessions, skip foreign or locally terminal work where possible, cap per-session activity pages, limit concurrency, and timeout provider reads without failing the whole sync loop. | `src/domain/sprint/session-sync/activity-fetch-plan.ts`, `src/domain/sprint/session-sync/activity-fetch-utils.ts`, `src/domain/sprint/session-sync/bounded-activity-fetch.ts`, `src/sprint/steps/session-sync-step.ts`, `src/server/activity-cache-service.ts` | `pnpm exec vitest run tests/backend/domain/sprint/session-sync/activity-fetch-plan.test.ts tests/backend/domain/sprint/session-sync/bounded-activity-fetch.test.ts tests/backend/sprint/session-sync-step.test.ts tests/backend/server/activity-cache-service.test.ts` |
 | Pure dashboard view models | v2 dashboard pages must build task, live runtime, and stats render models through pure helpers before JSX composition. Components should memoize scoped inputs and avoid rebuilding indexes, filter counts, task-card invocation feeds, or board columns inline during render. | `dashboard/src/v2/lib/live-session-view-model.ts`, `dashboard/src/v2/lib/tasks/task-board-view-model.ts`, `dashboard/src/v2/lib/task-board-state.ts`, `dashboard/src/v2/pages/stats/use-stats-page-data.ts`, `dashboard/src/v2/pages/stats/chart-view-models.ts`, `dashboard/src/v2/LiveSessionPage.tsx`, `dashboard/src/v2/TasksPage.tsx` | `pnpm exec vitest run tests/dashboard/v2/lib/live-session-view-model.test.ts tests/dashboard/lib/task-board-view-model.test.ts tests/dashboard/lib/task-board-state.test.ts tests/dashboard/v2/use-stats-page-data.test.tsx` |
 | Guardrail-backed regressions | Hot-path regressions must be enforced by typed tests and repository guardrails, not reviewer memory. Keep guardrail checks focused on durable risks: stale artifacts, broad `any`, unsafe dependency placeholders, realtime snapshot persistence, duplicate optimistic insertion, and large duplicate implementation blocks. | `scripts/check-quality-guardrails.mjs`, `tests/backend/scripts/quality-guardrails.test.ts`, `src/shared/late-bound-dependency.ts`, `src/app/dependency-factory/dashboard-factory.ts` | `pnpm run quality:guardrails`, `pnpm exec vitest run tests/backend/scripts/quality-guardrails.test.ts` |
 
@@ -55,6 +55,7 @@ Related docs: [Usage Telemetry And Stats](./usage-telemetry-and-stats.md), [Exec
 
 - Session sync should fetch activity only for sessions that can still change visible task state or runtime messages.
 - Keep page size, concurrency, and timeout limits at the fetch helper boundary so provider stalls degrade individual sessions rather than the entire sprint loop.
+- Keep generic timeout, error metadata normalization, and ordered bounded mapping semantics in `activity-fetch-utils.ts`; Jules-specific activity shaping belongs in caller wrappers such as `bounded-activity-fetch.ts`.
 - Activity sync must preserve ordering after bounded concurrent fetches so task updates stay deterministic.
 - Dashboard activity cache changes must maintain bounded fetch concurrency and short negative caching for repeated empty or failing reads.
 

src/domain/sprint/session-sync/activity-fetch-utils.ts

@@ -0,0 +1,79 @@
+export interface ActivityFetchErrorMetadata {
+  error: unknown;
+  errorName: string;
+  errorMessage: string;
+}
+
+export interface ActivityFetchTimeoutOptions {
+  timeoutMs: number;
+  createTimeoutError: () => Error;
+}
+
+export interface BoundedOrderedMapOptions<T, R> {
+  items: readonly T[];
+  concurrency: number;
+  mapper: (item: T, index: number) => Promise<R>;
+}
+
+export const normalizeActivityFetchError = (error: unknown): ActivityFetchErrorMetadata => {
+  if (error instanceof Error) {
+    return {
+      error,
+      errorName: error.name,
+      errorMessage: error.message,
+    };
+  }
+  return {
+    error,
+    errorName: typeof error,
+    errorMessage: String(error),
+  };
+};
+
+export const withActivityFetchTimeout = async <T>(
+  promise: Promise<T>,
+  options: ActivityFetchTimeoutOptions,
+): Promise<T> => {
+  if (!Number.isFinite(options.timeoutMs) || options.timeoutMs <= 0) {
+    return promise;
+  }
+
+  let timeout: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<T>((_, reject) => {
+        timeout = setTimeout(() => {
+          reject(options.createTimeoutError());
+        }, options.timeoutMs);
+      }),
+    ]);
+  } finally {
+    if (timeout) {
+      clearTimeout(timeout);
+    }
+  }
+};
+
+export const mapBoundedOrdered = async <T, R>({
+  items,
+  concurrency,
+  mapper,
+}: BoundedOrderedMapOptions<T, R>): Promise<R[]> => {
+  const results: R[] = new Array(items.length);
+  let currentIndex = 0;
+  const workerCount = Math.max(0, Math.min(Math.floor(concurrency), items.length));
+
+  const worker = async (): Promise<void> => {
+    while (currentIndex < items.length) {
+      const index = currentIndex++;
+      results[index] = await mapper(items[index], index);
+    }
+  };
+
+  await Promise.all(
+    Array.from({ length: workerCount }, () => worker()),
+  );
+
+  return results;
+};

src/domain/sprint/session-sync/bounded-activity-fetch.ts

@@ -1,49 +1,13 @@
 import type { JulesActivity } from "../../../contracts/app-types.js";
 import type { Logger } from "../../../shared/logging/logger.js";
+import {
+  mapBoundedOrdered,
+  normalizeActivityFetchError,
+  withActivityFetchTimeout,
+} from "./activity-fetch-utils.js";
 
 const DEFAULT_ACTIVITY_FETCH_TIMEOUT_MS = 30_000;
 
-const describeFetchError = (error: unknown): Record<string, unknown> => {
-  if (error instanceof Error) {
-    return {
-      error,
-      errorName: error.name,
-      errorMessage: error.message,
-    };
-  }
-  return {
-    error,
-    errorName: typeof error,
-    errorMessage: String(error),
-  };
-};
-
-const withTimeout = async <T>(
-  promise: Promise<T>,
-  timeoutMs: number,
-  sessionName: string,
-): Promise<T> => {
-  if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
-    return promise;
-  }
-
-  let timeout: NodeJS.Timeout | undefined;
-  try {
-    return await Promise.race([
-      promise,
-      new Promise<T>((_, reject) => {
-        timeout = setTimeout(() => {
-          reject(new Error(`Timed out fetching activities for ${sessionName} after ${timeoutMs}ms`));
-        }, timeoutMs);
-      }),
-    ]);
-  } finally {
-    if (timeout) {
-      clearTimeout(timeout);
-    }
-  }
-};
-
 export const fetchActivitiesBounded = async (
   sessionNames: string[],
   concurrency: number,
@@ -52,42 +16,37 @@ export const fetchActivitiesBounded = async (
   logger: Logger,
   timeoutMs: number = DEFAULT_ACTIVITY_FETCH_TIMEOUT_MS,
 ): Promise<Map<string, JulesActivity[]>> => {
-  const results = new Map<string, JulesActivity[]>();
-  let currentIndex = 0;
-
-  const worker = async () => {
-    while (currentIndex < sessionNames.length) {
-      const index = currentIndex++;
-      const sessionName = sessionNames[index];
+  const fetchResults = await mapBoundedOrdered({
+    items: sessionNames,
+    concurrency,
+    mapper: async (sessionName) => {
       const startedAt = Date.now();
       try {
-        const activities = await withTimeout(fetchRecentActivities(sessionName, pageSize), timeoutMs, sessionName);
-        results.set(sessionName, activities);
+        const activities = await withActivityFetchTimeout(
+          fetchRecentActivities(sessionName, pageSize),
+          {
+            timeoutMs,
+            createTimeoutError: () => new Error(`Timed out fetching activities for ${sessionName} after ${timeoutMs}ms`),
+          },
+        );
+        return activities;
       } catch (err) {
         logger.warn("Could not fetch activities for session", {
           sessionName,
           pageSize,
           concurrency,
           timeoutMs,
           elapsedMs: Date.now() - startedAt,
-          ...describeFetchError(err),
+          ...normalizeActivityFetchError(err),
         });
-        results.set(sessionName, []);
+        return [];
       }
-    }
-  };
-
-  const workers = [];
-  for (let i = 0; i < Math.min(concurrency, sessionNames.length); i++) {
-    workers.push(worker());
-  }
-
-  await Promise.all(workers);
+    },
+  });
 
-  // Preserve ordering of results matching input sessionNames array
   const orderedResults = new Map<string, JulesActivity[]>();
-  for (const sessionName of sessionNames) {
-    orderedResults.set(sessionName, results.get(sessionName) || []);
+  for (const [index, sessionName] of sessionNames.entries()) {
+    orderedResults.set(sessionName, fetchResults[index] || []);
   }
 
   return orderedResults;

tests/backend/domain/sprint/session-sync/bounded-activity-fetch.test.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 import { fetchActivitiesBounded } from "../../../../../src/domain/sprint/session-sync/bounded-activity-fetch.js";
+import { withActivityFetchTimeout } from "../../../../../src/domain/sprint/session-sync/activity-fetch-utils.js";
 import type { JulesActivity } from "../../../../../src/contracts/app-types.js";
 import type { Logger } from "../../../../../src/shared/logging/logger.js";
 
@@ -143,6 +144,21 @@ describe("fetchActivitiesBounded", () => {
     expect(mockLogger.warn).not.toHaveBeenCalled();
   });
 
+  it("passes non-positive timeout values through without scheduling timeout fallback", async () => {
+    vi.useFakeTimers();
+
+    const promise = new Promise<JulesActivity[]>((resolve) => {
+      setTimeout(() => resolve([{ id: "late-activity" }]), 5_000);
+    });
+    const resultPromise = withActivityFetchTimeout(promise, {
+      timeoutMs: 0,
+      createTimeoutError: () => new Error("should not timeout"),
+    });
+
+    await vi.advanceTimersByTimeAsync(5_000);
+    await expect(resultPromise).resolves.toEqual([{ id: "late-activity" }]);
+  });
+
   it("handles empty session list without error", async () => {
     const mockFetch = vi.fn();
     const result = await fetchActivitiesBounded([], 5, 5, mockFetch, mockLogger);