Live Streaming to the Org UI

The feature/issues-streaming flag turns the issues phase from a "wait for the whole run to settle" experience into a live one: a stale organisation's research starts the moment its page loads, and the scorecard, Analyst, and Spotlight views update as individual issues land rather than only once the run finishes. This page covers the on-load trigger, the realtime channel, client subscription, query invalidation, the live issue count, and the end-to-end sequence.

What the user sees

• Existing issues render immediately. A previously-enriched organisation's Analyst/Spotlight reported-issue lists and scorecard count show their last-known-good data straight away — nothing here blocks on a fresh run.
• Opening a stale or never-gathered organisation kicks off a fresh run on load. No manual "re-run" click is needed — the layout detects staleness and enqueues the issues job as a side effect of the page render.
• The scorecard count ticks up, and the reported-issue lists stream in, while research is still running. Rather than a single jump from 0 to N at settle time, the Modern Slavery flag count climbs incrementally, and Analyst/Spotlight cards appear as each source (BHRRC, the seven sanctions profiles, then Phase 2's deep-research cells) persists its findings.
• All of the above is off by default. With feature/issues-streaming OFF, the organisation view is unchanged from before this work: no live count, no on-load trigger, no mid-run list updates.

See Issue Research Pipeline for how the two engines and their flags fit together, and Two-Phase Native Brain for what Phase 1 and Phase 2 actually do. See Configuration & Operations for the full flag reference.

On-load trigger

OrganisationLayout (apps/web/src/app/(main)/organisations/[id]/layout.tsx) resolves the on-load trigger once, at the top of the layout, so every client surface underneath sees the same decision:

const willTriggerIssues = issuesStreamingEnabled && shouldEnqueueIssues(organisation, Date.now());
if (willTriggerIssues) {
  const useNative =
    readIssuesResearchHarnessLocalOverride() ??
    (await issuesResearchHarness().catch(() => resolveIssuesResearchHarness()));

  after(() =>
    new EnqueueIssuesIfStale()
      .execute({ org: organisation, flags: { useNative, streamingEnabled: issuesStreamingEnabled } })
      .catch((error) => logError('[OrganisationLayout] EnqueueIssuesIfStale failed', error, { organisationId })),
  );
}

shouldEnqueueIssues is the same pure gate EnqueueIssuesIfStale.execute calls internally (via enrichmentNeeds, mode 'if-stale'), so the layout's prediction and the use case's actual enqueue decision can never disagree about the rule — only, harmlessly, about timing (e.g. a concurrent tab already claimed the trigger).

The useNative/streamingEnabled flags are resolved in request scope (inside the render body, not inside after()), because the job itself is cookieless and this is the only way a per-session Vercel toolbar override reaches it. The actual enqueue runs inside after() so it never blocks the response, and EnqueueIssuesIfStale claims a Redis dedup key before sending the event:

const claimed = await getRedisClient().set(`issues-trigger:${org.id}`, '1', 'EX', 60, 'NX');
if (claimed !== 'OK') return { enqueued: false }; // another render/tab already claimed it

issues-trigger:{organisationId} is a SET NX EX 60 claim — a 60-second window that absorbs the several renders a single page load can produce (router.refresh(), the profile-complete/settle double refresh, <Link> prefetch) without enqueueing the job more than once. generate-issues.ts's own freshness self-gate (enrichmentNeeds again) is a second, independent backstop once the job actually runs.

IssuesStreamingProvider (_components/issues-streaming-provider.tsx) broadcasts the resolved issuesStreamingEnabled boolean to every client component under the layout via context (useIssuesStreaming()), defaulting to false for any consumer rendered outside the provider — a missing wrap fails safe into today's behaviour rather than crashing.

Realtime channel and events

Progress is published on a per-organisation Inngest realtime channel, enrichmentChannel (infrastructure/inngest/channels.ts):

export const enrichmentChannel = channel({
  name: ({ organisationId }: { organisationId: string }) => `enrichment:${organisationId}`,
  topics: { progress: { schema: enrichmentProgressSchema } },
});

The channel name is enrichment:{organisationId}, with a single topic, progress. Four event types (domain/constants/enrichment.ts, ENRICHMENT_PROGRESS_TYPES) can appear on that topic:

issues-landed only fires on the native path, and only when streamingEnabled was resolved true for that run:

const twoPhaseDeps = withOnProgress(
  streamingEnabled ? (input: { organisationId: string }) => publishIssuesLanded(input.organisationId) : undefined,
);

publishIssuesLanded is deliberately payload-free — it carries no issue data, only a wake signal ({ type: 'issues-landed' }) — and is best-effort: a publish failure is caught and logged, never allowed to fail the research run.

Because Phase 1 fans out eight sources (BHRRC plus seven sanctions profiles) in parallel but persists them through one serial chain as each resolves, onProgress — and therefore issues-landed — can fire well before the phase as a whole completes. BHRRC, typically the fastest source, can land only a few seconds into a run rather than only at the end of the phase. See Two-Phase Native Brain for the persist-chain mechanics, and Data Model & Persistence for why persists are serialised.

Client subscription

EnrichmentRefresher (_components/enrichment-refresher.tsx) mounts a client-only RealtimeRefreshEngine, deferred until after the first client render so useRealtime (SSR-unsafe) never runs during server render. RealtimeRefreshEngine hosts the shared useEnrichmentAutoRefresh hook, which opens the subscription:

const { messages } = useRealtime({
  channel: enrichmentChannel({ organisationId }),
  topics: ['progress'] as const,
  token: () => fetchRealtimeToken(organisationId),
  enabled: realtimeEnabled,
  autoCloseOnTerminal: true,
});

fetchRealtimeToken calls the account-scoped realtime-token route, GET /api/identify/sidebar/[organisationId]/realtime-token. The route authenticates via requireAccountOr401, then re-reads the organisation with OrganisationRepository.findByIdForAccount — an unscoped read here would let any authenticated user mint a token for, and watch, another tenant's organisation (IDOR) — before minting the token with getClientSubscriptionToken.

realtimeEnabled is not simply "is a phase pending". It also ORs in an optimistic bridge:

const waitingForIssuesStart = willTriggerIssues && !observedPending && !timedOut;
const realtimeEnabled = isPending || waitingForIssuesStart;

The gap the bridge closes: after() plus the async on-load trigger means the client's first hydrated status read shows "not pending" — the enqueue hasn't landed yet — so a subscription gated on isPending alone would never turn on to observe the PENDING transition. While willTriggerIssues is true and neither terminal condition has fired, the bridge forces the subscription on and actively polls the enrichment-status query every 5 seconds (BRIDGE_POLL_INTERVAL_MS), bounded by a 75-second timeout (BRIDGE_TIMEOUT_MS) that silently falls back to passive rendering — it never fabricates a settle. waitingForIssuesStart is only ever added to realtimeEnabled, never to isPending itself, so a bridge timeout can't trigger a bogus { source: 'settle' } refresh before the job has actually run.

Whatever triggers a refresh — a realtime message or the pending→settled edge — the client always re-fetches authoritative truth (React Query invalidation, or router.refresh()); it never trusts data riding the event itself, since every event on this channel (including issues-landed) is payload-free by design.

Invalidation routing

enrichmentInvalidationFilters (_components/enrichment-invalidation.ts) is the pure routing table between "what changed" and "which queries to invalidate":

The 750ms debounce (ISSUES_LANDED_DEBOUNCE_MS) exists because Phase 1's up-to-eight-source burst can fire several issues-landed events in quick succession; debouncing collapses that burst to one or two refetches instead of one per source. Every other trigger invalidates immediately. router.refresh() (the whole-route RSC re-render) is likewise skipped specifically for issues-landed (shouldSkipRouterRefresh) — nothing server-rendered changes per landed issue, so re-running the whole route on every tick of a run would reopen the #1368 refetch storm this invalidation split originally fixed.

On the read side, useReportedIssues (hooks/use-organisation-risk.ts) is a plain, non-suspense useQuery with placeholderData: keepPreviousData — not a Suspense boundary — so a reportedIssues invalidation mid-run swaps in the new list in place, without unmounting and re-suspending the component the way a Suspense-driven query would.

Live issues count

useLiveIssuesCount (hooks/use-live-issues-count.ts) polls a dedicated live count for the scorecard badge while research is running:

const LIVE_ISSUES_COUNT_REFETCH_INTERVAL_MS = 3000;

return useQuery<GetLiveIssuesCountOutput>({
  queryKey: queryKeys.organisations.liveIssuesCount(organisationId),
  queryFn: () => queryAction(fetchLiveIssuesCountAction({ organisationId, issuesStatus })),
  enabled,          // streamingEnabled && issuesPending
  staleTime: 0,
  refetchInterval: LIVE_ISSUES_COUNT_REFETCH_INTERVAL_MS,
  refetchIntervalInBackground: false,
});

It polls every 3 seconds, but only while streamingEnabled && issuesPending — issuesPending itself comes from getEnrichmentGates, fed by the separate 5-second useEnrichmentStatusQuery poll. The count is always re-fetched from the database, never client-incremented, so a retry-driven replay of an already-persisted issue can't double-count it.

fetchLiveIssuesCountAction calls GetLiveIssuesCount, which is deliberately cheap outside its one active window:

if (issuesStatus !== EnrichmentPhaseStatus.Pending) return { issuesCount: 0 };
const useNative = await resolveIssuesResearchHarness();
if (!useNative) return { issuesCount: 0 };
return { issuesCount: await IssueRepository.countByOrganisationAndModule(organisationId, SatelliteMode.ModernSlavery) };

It returns 0 immediately — no database read — unless issuesStatus === Pending and the native harness is on; only the native path persists incrementally, so a climbing count is only meaningful there. Once the phase settles, the scorecard switches back to the authoritative flagCounts.issuesCount (also countByOrganisationAndModule under the hood — see Data Model & Persistence) rather than continuing to poll the live count.

A FAILED run and a healthy 0-issue run both settle at issuesCount === 0 — otherwise indistinguishable. isIssuesPhaseFailed(enrichmentStatus) (lib/enrichment-status.ts) distinguishes the two: when settled with issuesStatus === Failed, the scorecard cell and the Analyst/Spotlight lists render a distinct "couldn't complete" affordance next to the (still-real, however zero) count, rather than a bare "0" that reads as "nothing found".

Sequence