diff --git a/client/dive-common/components/TrackSettingsPanel.vue b/client/dive-common/components/TrackSettingsPanel.vue
index 03111b4df..45467923b 100644
--- a/client/dive-common/components/TrackSettingsPanel.vue
+++ b/client/dive-common/components/TrackSettingsPanel.vue
@@ -58,7 +58,7 @@ export default defineComponent({
       help,
       modes,
       typeList,
-      stereoMatchMethods: stereoMatchMethodsFor(isDesktopRuntime()),
+      stereoMatchMethods: computed(() => stereoMatchMethodsFor(isDesktopRuntime())),
     };
   },
 });
diff --git a/client/dive-common/use/stereo/StereoServerMatcher.spec.ts b/client/dive-common/use/stereo/StereoServerMatcher.spec.ts
new file mode 100644
index 000000000..fc7c8f012
--- /dev/null
+++ b/client/dive-common/use/stereo/StereoServerMatcher.spec.ts
@@ -0,0 +1,37 @@
+import {
+  describe, expect, it, vi,
+} from 'vitest';
+import StereoServerMatcher, { parseFrameKey } from './StereoServerMatcher';
+
+const RANGE = { min: 0, max: 0 } as never;
+
+describe('StereoServerMatcher', () => {
+  it('reads the frame pair out of the transfer key', () => {
+    expect(parseFrameKey('left>right@12')).toEqual({ source: 'left', target: 'right', frame: 12 });
+    expect(() => parseFrameKey(undefined)).toThrow('which frame');
+  });
+
+  it('asks the server for the points of that frame and marks the misses', async () => {
+    const api = {
+      transferPoints: vi.fn(async () => ({
+        success: true, transferredPoints: [[10, 11], [NaN, NaN]] as [number, number][], validMatches: [true, false],
+      })),
+      setFrame: vi.fn(async () => ({})),
+    };
+    const matcher = new StereoServerMatcher(api, 'foundation');
+    const results = await matcher.warpPoints([[1, 2], [3, 4]], null, null, null, { range: RANGE, frameKey: 'left>right@3' });
+    expect(api.transferPoints).toHaveBeenCalledWith({
+      frame: 3, sourceCamera: 'left', points: [[1, 2], [3, 4]], method: 'foundation',
+    });
+    expect(results.map((r) => r.accepted)).toEqual([true, false]);
+    expect([results[0].x, results[0].y]).toEqual([10, 11]);
+    await matcher.prepare('right>left@7', null, null, null);
+    expect(api.setFrame).toHaveBeenCalledWith(7, 'foundation');
+  });
+
+  it('surfaces the server reason when the warp fails', async () => {
+    const api = { transferPoints: vi.fn(async () => ({ success: false, error: 'no calibration' })), setFrame: vi.fn() };
+    const matcher = new StereoServerMatcher(api, 'ncc');
+    await expect(matcher.warpPoints([[1, 2]], null, null, null, { range: RANGE, frameKey: 'a>b@1' })).rejects.toThrow('no calibration');
+  });
+});
diff --git a/client/dive-common/use/stereo/StereoServerMatcher.ts b/client/dive-common/use/stereo/StereoServerMatcher.ts
new file mode 100644
index 000000000..6a7685cd5
--- /dev/null
+++ b/client/dive-common/use/stereo/StereoServerMatcher.ts
@@ -0,0 +1,77 @@
+import type { StereoMatcher, StereoMatchMethod } from './stereoMatcher';
+import type { WarpOptions, WarpResult } from './StereoOnnxMatcher';
+
+/** Identity of the frame pair a warp belongs to, from the transfer's frameKey. */
+export interface FramePair {
+  source: string;
+  target: string;
+  frame: number;
+}
+
+/** The transfer keys frames as `source>target@frame`. */
+export function parseFrameKey(key: string | undefined): FramePair {
+  const match = key?.match(/^(.+?)>(.+?)@(\d+)$/);
+  if (!match) throw new Error('A server-side warp needs to know which frame it is for.');
+  return { source: match[1], target: match[2], frame: Number(match[3]) };
+}
+
+export interface ServerStereoApi {
+  transferPoints(request: {
+    frame: number; sourceCamera: string; points: [number, number][]; method: StereoMatchMethod;
+  }): Promise<{
+    success: boolean; error?: string; transferredPoints?: [number, number][]; validMatches?: boolean[];
+  }>;
+  setFrame(frame: number, method: StereoMatchMethod): Promise<unknown>;
+}
+
+/**
+ * A matcher whose correspondence search runs in VIAME's interactive service
+ * on the server. The frame pixels and rig the transfer hands over are unused:
+ * the server reads the same frames from its own copy of the dataset.
+ */
+export default class StereoServerMatcher implements StereoMatcher {
+  private api: ServerStereoApi;
+
+  private method: StereoMatchMethod;
+
+  constructor(api: ServerStereoApi, method: StereoMatchMethod) {
+    this.api = api;
+    this.method = method;
+  }
+
+  async warpPoints(
+    points: [number, number][],
+    _source: unknown,
+    _target: unknown,
+    _rig: unknown,
+    opts: WarpOptions,
+  ): Promise<WarpResult[]> {
+    if (!points.length) return [];
+    const pair = parseFrameKey(opts.frameKey);
+    const response = await this.api.transferPoints({
+      frame: pair.frame, sourceCamera: pair.source, points, method: this.method,
+    });
+    if (!response.success) {
+      throw new Error(response.error || 'The server could not map the points to the other camera.');
+    }
+    const mapped = response.transferredPoints ?? [];
+    return points.map((_, i) => {
+      const point = mapped[i];
+      const valid = response.validMatches?.[i] !== false
+        && !!point && Number.isFinite(point[0]) && Number.isFinite(point[1]);
+      return {
+        x: valid ? point[0] : NaN,
+        y: valid ? point[1] : NaN,
+        score: valid ? 1 : 0,
+        secondScore: 0,
+        accepted: valid,
+      };
+    });
+  }
+
+  async prepare(frameKey: string, _s: unknown, _t: unknown, _r: unknown, stillWanted?: () => boolean) {
+    const pair = parseFrameKey(frameKey);
+    if (stillWanted && !stillWanted()) return;
+    await this.api.setFrame(pair.frame, this.method);
+  }
+}
diff --git a/client/dive-common/use/stereo/stereoMatcher.ts b/client/dive-common/use/stereo/stereoMatcher.ts
index 1a0881eb5..cf4df6b47 100644
--- a/client/dive-common/use/stereo/stereoMatcher.ts
+++ b/client/dive-common/use/stereo/stereoMatcher.ts
@@ -1,3 +1,4 @@
+import { ref } from 'vue';
 /**
  * The contract both correspondence methods satisfy, so the transfer composable
  * and its callers never branch on which one is selected.
@@ -45,8 +46,13 @@ export const STEREO_MATCH_METHODS: { value: StereoMatchMethod; text: string; des
   { value: 'ncc', text: 'Lower Quality, Faster' },
 ];
 
+/** Methods a server-side interactive service can run for the web client. */
+export const serverStereoMethods = ref<StereoMatchMethod[]>([]);
+
 export function stereoMatchMethodsFor(desktop: boolean) {
-  return STEREO_MATCH_METHODS.filter((m) => desktop || !m.desktopOnly);
+  return STEREO_MATCH_METHODS.filter(
+    (m) => desktop || !m.desktopOnly || serverStereoMethods.value.includes(m.value),
+  );
 }
 
 export function isStereoMatchMethod(value: unknown, desktop: boolean): value is StereoMatchMethod {
diff --git a/client/dive-common/use/stereo/useStereoOnnxTransfer.ts b/client/dive-common/use/stereo/useStereoOnnxTransfer.ts
index c142fa735..37287e262 100644
--- a/client/dive-common/use/stereo/useStereoOnnxTransfer.ts
+++ b/client/dive-common/use/stereo/useStereoOnnxTransfer.ts
@@ -603,6 +603,7 @@ export default function useStereoOnnxTransfer(config: StereoOnnxTransferConfig)
     warpAllFromCamera,
     warpPoint,
     measureAtFrame,
+    measureAndReport,
     precomputeFrame,
   };
 }
diff --git a/client/platform/web-girder/api/configuration.service.ts b/client/platform/web-girder/api/configuration.service.ts
index 78cef19c5..1224511f7 100644
--- a/client/platform/web-girder/api/configuration.service.ts
+++ b/client/platform/web-girder/api/configuration.service.ts
@@ -65,6 +65,11 @@ export interface DiveConfiguration {
   trainingEnabled?: boolean;
   jobsDisabled?: boolean;
   jobsDisabledMessage?: string;
+  /** Server-side interactive segmentation and stereo (the `interactive` service). */
+  interactiveEnabled?: boolean;
+  interactiveMessage?: string;
+  interactiveStereoMethods?: string[];
+  interactiveTextQuery?: boolean;
 }
 
 export interface JobsDisabledConfig {
diff --git a/client/platform/web-girder/api/index.ts b/client/platform/web-girder/api/index.ts
index 9bcca84e4..7118469aa 100644
--- a/client/platform/web-girder/api/index.ts
+++ b/client/platform/web-girder/api/index.ts
@@ -10,6 +10,7 @@ export * from './configuration.service';
 export * from './dataset.service';
 export { default as loadFrameMetadata } from './frameMetadata.service';
 export * from './girder.service';
+export * from './interactive.service';
 export * from './multicamResolve';
 export * from './rpc.service';
 export * from './scoring.service';
diff --git a/client/platform/web-girder/api/interactive.service.ts b/client/platform/web-girder/api/interactive.service.ts
new file mode 100644
index 000000000..af9631db8
--- /dev/null
+++ b/client/platform/web-girder/api/interactive.service.ts
@@ -0,0 +1,174 @@
+/**
+ * Server-side interactive segmentation and stereo: the GPU host runs VIAME's
+ * interactive service and girder resolves dataset frames to its inputs.
+ */
+import type {
+  SegmentationPolygon, SegmentationPolygonKeypointsResponse, SegmentationPredictResponse,
+  SegmentationStereoSegmentResponse, TextQueryResponse,
+} from 'dive-common/apispec';
+import type { StereoMatchMethod } from 'dive-common/use/stereo/stereoMatcher';
+import girderRest from 'platform/web-girder/plugins/girder';
+
+export interface InteractiveStatus {
+  interactiveEnabled: boolean;
+  interactiveMessage: string;
+  interactiveStereoMethods?: StereoMatchMethod[];
+  interactiveTextQuery?: boolean;
+}
+
+interface FrameRequest {
+  datasetId: string;
+  /** Camera name of a multi-camera dataset; ignored for single-camera ones. */
+  camera?: string;
+  frame: number;
+}
+
+export interface InteractivePredictRequest extends FrameRequest {
+  points: [number, number][];
+  pointLabels: number[];
+  box?: [number, number, number, number];
+  line?: [number, number][];
+  multimaskOutput?: boolean;
+}
+
+export interface InteractiveStereoSegmentRequest {
+  datasetId: string;
+  frame: number;
+  sourceCamera: string;
+  points: [number, number][];
+  pointLabels: number[];
+  polygon?: [number, number][];
+  polygons?: SegmentationPolygon[];
+  method?: StereoMatchMethod;
+}
+
+export interface InteractiveTransferPointsRequest {
+  datasetId: string;
+  frame: number;
+  sourceCamera: string;
+  points: [number, number][];
+  strict?: boolean;
+  method?: StereoMatchMethod;
+}
+
+export interface InteractiveTransferPointsResponse {
+  success: boolean;
+  error?: string;
+  transferredPoints?: [number, number][];
+  validMatches?: boolean[];
+  disparityValues?: number[];
+}
+
+export interface InteractiveMeasurement {
+  length: number;
+  midpoint_x: number;
+  midpoint_y: number;
+  midpoint_z: number;
+  midpoint_range: number;
+  stereo_rms: number;
+}
+
+export interface InteractiveTransferLineResponse {
+  success: boolean;
+  error?: string;
+  transferredLine?: [number, number][];
+  length?: number;
+  measurement?: InteractiveMeasurement;
+}
+
+const BASE = 'dive_interactive';
+
+async function post<T>(path: string, body: Record<string, unknown>): Promise<T> {
+  return (await girderRest.post<T>(`${BASE}/${path}`, body)).data;
+}
+
+export async function getInteractiveStatus(): Promise<InteractiveStatus> {
+  return (await girderRest.get<InteractiveStatus>(`${BASE}/status`)).data;
+}
+
+export function interactiveSegmentationPredict(
+  request: InteractivePredictRequest,
+): Promise<SegmentationPredictResponse> {
+  return post<SegmentationPredictResponse>('segmentation/predict', { ...request });
+}
+
+export function interactiveSegmentationKeypoints(datasetId: string, polygon: [number, number][], polygons?: SegmentationPolygon[]): Promise<SegmentationPolygonKeypointsResponse> {
+  return post<SegmentationPolygonKeypointsResponse>('segmentation/keypoints', { datasetId, polygon, polygons });
+}
+
+export async function interactiveStereoSegment(
+  request: InteractiveStereoSegmentRequest,
+): Promise<SegmentationStereoSegmentResponse> {
+  const raw = await post<Record<string, unknown>>('segmentation/stereo_segment', { ...request });
+  return {
+    id: '',
+    success: !!raw.success,
+    error: raw.error as string | undefined,
+    polygon: raw.polygon as SegmentationStereoSegmentResponse['polygon'],
+    polygons: raw.polygons as SegmentationPolygon[] | undefined,
+    bounds: raw.bounds as SegmentationStereoSegmentResponse['bounds'],
+    score: raw.score as number | undefined,
+    seedPoints: raw.seed_points as [number, number][] | undefined,
+    seedLabels: raw.seed_labels as number[] | undefined,
+    generateLine: raw.generate_line as boolean | undefined,
+    lineSource: raw.line_source as SegmentationStereoSegmentResponse['lineSource'],
+    lineOther: raw.line_other as SegmentationStereoSegmentResponse['lineOther'],
+    measurement: raw.measurement as InteractiveMeasurement | undefined,
+  };
+}
+
+export function interactiveTextQuery(
+  request: FrameRequest & { text: string },
+): Promise<TextQueryResponse> {
+  return post<TextQueryResponse>('text_query', { ...request });
+}
+
+export function interactiveStereoSetFrame(datasetId: string, frame: number, method?: StereoMatchMethod): Promise<{ success: boolean; error?: string; disparity_ready?: boolean }> {
+  return post('stereo/set_frame', { datasetId, frame, method });
+}
+
+export async function interactiveStereoTransferPoints(
+  request: InteractiveTransferPointsRequest,
+): Promise<InteractiveTransferPointsResponse> {
+  const raw = await post<Record<string, unknown>>('stereo/transfer_points', { ...request });
+  return {
+    success: !!raw.success,
+    error: raw.error as string | undefined,
+    transferredPoints: raw.transferred_points as [number, number][] | undefined,
+    validMatches: raw.valid_matches as boolean[] | undefined,
+    disparityValues: raw.disparity_values as number[] | undefined,
+  };
+}
+
+export async function interactiveStereoTransferLine(datasetId: string, frame: number, line: [number, number][], method?: StereoMatchMethod): Promise<InteractiveTransferLineResponse> {
+  const raw = await post<Record<string, unknown>>('stereo/transfer_line', {
+    datasetId, frame, line, method,
+  });
+  return {
+    success: !!raw.success,
+    error: raw.error as string | undefined,
+    transferredLine: raw.transferred_line as [number, number][] | undefined,
+    length: raw.length as number | undefined,
+    measurement: raw.measurement as InteractiveMeasurement | undefined,
+  };
+}
+
+export function interactiveStereoMeasureLine(
+  datasetId: string,
+  frame: number,
+  leftLine: [number, number][],
+  rightLine: [number, number][],
+  method?: StereoMatchMethod,
+): Promise<{ success: boolean; error?: string; length?: number; measurement?: InteractiveMeasurement }> {
+  return post('stereo/measure_line', {
+    datasetId, frame, leftLine, rightLine, method,
+  });
+}
+
+export function interactiveStereoAggregateLengths(datasetId: string, lengths: number[], method?: string): Promise<{ success: boolean; error?: string; avg_length?: number }> {
+  return post('stereo/aggregate_lengths', { datasetId, lengths, method });
+}
+
+export async function endInteractiveSession(datasetId: string): Promise<void> {
+  await girderRest.delete(`${BASE}/session`, { params: { datasetId } });
+}
diff --git a/client/platform/web-girder/store/useConfig.ts b/client/platform/web-girder/store/useConfig.ts
index c80ee0135..5a69064b8 100644
--- a/client/platform/web-girder/store/useConfig.ts
+++ b/client/platform/web-girder/store/useConfig.ts
@@ -4,6 +4,8 @@ import {
   DEFAULT_JOBS_DISABLED_MESSAGE,
   getConfig,
 } from 'platform/web-girder/api/configuration.service';
+import { getInteractiveStatus } from 'platform/web-girder/api/interactive.service';
+import { serverStereoMethods, type StereoMatchMethod } from 'dive-common/use/stereo/stereoMatcher';
 
 export interface ConfigState {
   distributedWorkerEnabled: boolean;
@@ -11,6 +13,10 @@ export interface ConfigState {
   trainingEnabled: boolean;
   jobsDisabled: boolean;
   jobsDisabledMessage: string;
+  interactiveEnabled: boolean;
+  interactiveMessage: string;
+  interactiveStereoMethods: StereoMatchMethod[];
+  interactiveTextQuery: boolean;
 }
 
 const distributedWorkerEnabled = ref(false);
@@ -18,6 +24,9 @@ const pipelinesEnabled = ref(false);
 const trainingEnabled = ref(false);
 const jobsDisabled = ref(false);
 const jobsDisabledMessage = ref(DEFAULT_JOBS_DISABLED_MESSAGE);
+const interactiveEnabled = ref(false);
+const interactiveMessage = ref('');
+const interactiveTextQuery = ref(false);
 
 export function useConfig() {
   function getDistributedWorkerEnabled(): boolean {
@@ -60,12 +69,27 @@ export function useConfig() {
     jobsDisabledMessage.value = value || DEFAULT_JOBS_DISABLED_MESSAGE;
   }
 
+  function setInteractive(payload: {
+    interactiveEnabled?: boolean;
+    interactiveMessage?: string;
+    interactiveStereoMethods?: string[];
+    interactiveTextQuery?: boolean;
+  }): void {
+    interactiveEnabled.value = !!payload.interactiveEnabled;
+    interactiveMessage.value = payload.interactiveMessage || '';
+    interactiveTextQuery.value = !!payload.interactiveTextQuery;
+    // Methods the server can run are offered in the settings even when the
+    // browser could not run them itself.
+    serverStereoMethods.value = (payload.interactiveStereoMethods || []) as StereoMatchMethod[];
+  }
+
   function setCapabilities(payload: Partial<ConfigState>): void {
     distributedWorkerEnabled.value = payload.distributedWorkerEnabled ?? false;
     pipelinesEnabled.value = payload.pipelinesEnabled ?? false;
     trainingEnabled.value = payload.trainingEnabled ?? false;
     jobsDisabled.value = payload.jobsDisabled ?? false;
     jobsDisabledMessage.value = payload.jobsDisabledMessage || DEFAULT_JOBS_DISABLED_MESSAGE;
+    setInteractive(payload);
   }
 
   async function loadConfig(): Promise<void> {
@@ -76,9 +100,23 @@ export function useConfig() {
       trainingEnabled: !!data.trainingEnabled,
       jobsDisabled: !!data.jobsDisabled,
       jobsDisabledMessage: data.jobsDisabledMessage || DEFAULT_JOBS_DISABLED_MESSAGE,
+      interactiveEnabled: !!data.interactiveEnabled,
+      interactiveMessage: data.interactiveMessage || '',
+      interactiveStereoMethods: (data.interactiveStereoMethods || []) as StereoMatchMethod[],
+      interactiveTextQuery: !!data.interactiveTextQuery,
     });
   }
 
+  /** Re-check the interactive service, whose availability follows the job queue. */
+  async function refreshInteractive(): Promise<boolean> {
+    try {
+      setInteractive(await getInteractiveStatus());
+    } catch {
+      setInteractive({ interactiveEnabled: false, interactiveMessage: 'The interactive service could not be reached.' });
+    }
+    return interactiveEnabled.value;
+  }
+
   return {
     distributedWorkerEnabled,
     pipelinesEnabled,
@@ -97,5 +135,9 @@ export function useConfig() {
     setJobsDisabledMessage,
     setCapabilities,
     loadConfig,
+    interactiveEnabled,
+    interactiveMessage,
+    interactiveTextQuery,
+    refreshInteractive,
   };
 }
diff --git a/client/platform/web-girder/useServerSegmentation.ts b/client/platform/web-girder/useServerSegmentation.ts
new file mode 100644
index 000000000..b3d8e7d0f
--- /dev/null
+++ b/client/platform/web-girder/useServerSegmentation.ts
@@ -0,0 +1,136 @@
+/**
+ * Web point segmentation backed by the server's interactive service: the
+ * recipe's clicks become `dive_interactive` requests, and a finished mask is
+ * carried to the other stereo camera by the service's own `stereo_segment`.
+ */
+import type { SegmentationPolygon, SegmentationPredictRequest } from 'dive-common/apispec';
+import { SegmentationPolygonKey } from 'dive-common/recipes/segmentationpointclick';
+import {
+  componentsBounds, isSegmentationPolygonKey, segmentationPolygonFeatures,
+} from 'dive-common/recipes/segmentationPolygons';
+import { clientSettings } from 'dive-common/store/settings';
+import type { StereoAnnotationCompleteParams } from 'dive-common/use/useModeManager';
+import { headTailFeatures } from 'vue-media-annotator/headTail';
+import type Track from 'vue-media-annotator/track';
+import {
+  interactiveSegmentationPredict, interactiveStereoSegment,
+} from 'platform/web-girder/api/interactive.service';
+import { fromViewer } from 'platform/web-girder/useStereoOnnxWeb';
+
+interface ServerSegmentationOptions {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  getViewer: () => any;
+  getDatasetId: () => string;
+  /** Whether the server can take requests right now, refreshed by the host. */
+  isEnabled: () => boolean;
+  /** Why it cannot, shown when the user reaches for the tool anyway. */
+  disabledMessage: () => string;
+  refreshMeasurement: (trackId: number, frameNum: number) => Promise<unknown>;
+  onError: (message: string) => void;
+}
+
+export default function useServerSegmentation(opts: ServerSegmentationOptions) {
+  function trackAt(camera: string, trackId: number): Track | undefined {
+    return opts.getViewer()?.cameraStore?.getPossibleTrack(trackId, camera);
+  }
+
+  function polygonsAt(camera: string, trackId: number, frameNum: number): SegmentationPolygon[] {
+    return (trackAt(camera, trackId)?.getFeature(frameNum)[0]?.geometry?.features ?? [])
+      .filter((f): f is GeoJSON.Feature<GeoJSON.Polygon> => f.geometry.type === 'Polygon')
+      .map((f) => ({
+        exterior: f.geometry.coordinates[0] as [number, number][],
+        holes: f.geometry.coordinates.slice(1) as [number, number][][],
+      }));
+  }
+
+  /** Hook the shared recipe up once the Viewer is mounted. */
+  function initialize(): boolean {
+    const viewer = opts.getViewer();
+    if (!viewer?.segmentationRecipe) return false;
+    viewer.segmentationRecipe.initialize({
+      predictFn: (request: SegmentationPredictRequest, frameNum: number) => interactiveSegmentationPredict({
+        datasetId: opts.getDatasetId(),
+        // The recipe passes getImagePath's value back: the camera name here.
+        camera: request.imagePath || undefined,
+        frame: frameNum,
+        points: request.points,
+        pointLabels: request.pointLabels,
+        box: request.box,
+        line: request.line,
+        multimaskOutput: request.multimaskOutput,
+      }),
+      getImagePath: () => fromViewer<string>(opts.getViewer()?.selectedCamera) ?? '',
+      initializeServiceFn: async () => {
+        if (!opts.isEnabled()) {
+          throw new Error(opts.disabledMessage() || 'Interactive segmentation is not available right now.');
+        }
+      },
+    });
+    return true;
+  }
+
+  /**
+   * A click-segmented mask on one camera, carried to the other by the
+   * service. Returns false when the event is not one this handles, so the
+   * host can pass it to the geometry transfer instead.
+   */
+  async function handleStereoSegmentation(params: StereoAnnotationCompleteParams): Promise<boolean> {
+    if (params.type !== 'segmentation') return false;
+    const viewer = opts.getViewer();
+    const cameras: string[] = fromViewer<string[]>(viewer?.multiCamList) ?? [];
+    const otherCamera = cameras.find((c) => c !== params.camera);
+    if (!otherCamera || !clientSettings.stereoSettings.autoComputeOtherCamera) return true;
+    if (!opts.isEnabled()) return true;
+    const polygons = polygonsAt(params.camera, params.trackId, params.frameNum);
+    if (!polygons.length) return true;
+    try {
+      const response = await interactiveStereoSegment({
+        datasetId: opts.getDatasetId(),
+        frame: params.frameNum,
+        sourceCamera: params.camera,
+        points: params.points,
+        pointLabels: params.labels,
+        polygon: polygons[0].exterior,
+        polygons,
+        method: clientSettings.stereoSettings.matchMethod,
+      });
+      if (!response.success) throw new Error(response.error || 'No mask found on the other camera.');
+      let mask = response.polygons ?? [];
+      if (!mask.length && response.polygon) mask = [{ exterior: response.polygon, holes: [] }];
+      if (!mask.length) throw new Error('No mask found on the other camera.');
+      if (viewer !== opts.getViewer()) return true;
+      let track = trackAt(otherCamera, params.trackId);
+      const source = trackAt(params.camera, params.trackId);
+      if (!track) {
+        track = viewer.cameraStore.camMap.value.get(otherCamera)?.trackStore
+          .add(params.frameNum, source?.confidencePairs?.[0]?.[0] || 'unknown', undefined, params.trackId);
+        if (track && source) track.set = source.set;
+      }
+      if (!track) throw new Error('Could not create the other-camera detection.');
+      const bounds = response.bounds ?? componentsBounds(mask);
+      track.setFeature({
+        frame: params.frameNum, keyframe: true, interpolate: false, bounds,
+      });
+      const geometries = segmentationPolygonFeatures(mask, SegmentationPolygonKey);
+      const keys = new Set(geometries.map((g) => g.properties?.key));
+      track.getPolygonFeatures(params.frameNum).forEach((p) => {
+        if (isSegmentationPolygonKey(p.key, SegmentationPolygonKey) && !keys.has(p.key)) {
+          track!.removeFeatureGeometry(params.frameNum, { key: p.key, type: 'Polygon' });
+        }
+      });
+      track.setFeature({ frame: params.frameNum, keyframe: true }, geometries);
+      if (response.generateLine && response.lineOther) {
+        track.setFeature({ frame: params.frameNum, keyframe: true, bounds }, headTailFeatures(response.lineOther));
+        if (response.lineSource && source) {
+          source.setFeature({ frame: params.frameNum, keyframe: true }, headTailFeatures(response.lineSource));
+        }
+      }
+      await opts.refreshMeasurement(params.trackId, params.frameNum);
+    } catch (err) {
+      opts.onError(`Stereo segmentation: ${(err as Error).message}`);
+    }
+    return true;
+  }
+
+  return { initialize, handleStereoSegmentation };
+}
diff --git a/client/platform/web-girder/useStereoOnnxWeb.ts b/client/platform/web-girder/useStereoOnnxWeb.ts
index 92a598eb6..f024576f4 100644
--- a/client/platform/web-girder/useStereoOnnxWeb.ts
+++ b/client/platform/web-girder/useStereoOnnxWeb.ts
@@ -72,6 +72,11 @@ export interface StereoOnnxWebOptions {
   foundationModelSpec?: FoundationModelSpec;
   /** Overrides the user's dropdown choice; mainly for tests. */
   getMatchMethod?: () => StereoMatchMethod;
+  /**
+   * A matcher run by the server's interactive service, used for every method
+   * while `enabled()` holds; the browser models are the fallback.
+   */
+  server?: { enabled: () => boolean; matcher: (method: StereoMatchMethod) => StereoMatcher };
   range?: SearchRange;
   /**
    * Progress message; null clears it. `progress` accompanies the messages whose
@@ -213,6 +218,11 @@ export default function useStereoOnnxWeb(opts: StereoOnnxWebOptions) {
 
   function getMatcher(imagery?: ImagerySize): Promise<StereoMatcher | null> {
     const method = currentMethod();
+    if (opts.server?.enabled()) {
+      const serverKey = `server:${method}`;
+      if (!matchers[serverKey]) matchers[serverKey] = Promise.resolve(opts.server.matcher(method));
+      return matchers[serverKey];
+    }
     const key = method === 'foundation' && imagery
       ? `${method}:${imagery.width}x${imagery.height}`
       : method;
@@ -438,5 +448,8 @@ export default function useStereoOnnxWeb(opts: StereoOnnxWebOptions) {
     stereoViewLink,
     precomputeCurrentFrame,
     invalidateCalibration,
+    refreshMeasurement: async (trackId: number, frameNum: number) => (
+      getTransfer()?.measureAndReport(trackId, frameNum)
+    ),
   };
 }
diff --git a/client/platform/web-girder/views/ViewerLoader.vue b/client/platform/web-girder/views/ViewerLoader.vue
index eedb65de1..cb6d4b1fa 100644
--- a/client/platform/web-girder/views/ViewerLoader.vue
+++ b/client/platform/web-girder/views/ViewerLoader.vue
@@ -1,6 +1,6 @@
 <script lang="ts">
 import {
-  computed, defineComponent, onBeforeUnmount, onMounted, ref, watch, Ref, PropType,
+  computed, defineComponent, nextTick, onBeforeUnmount, onMounted, ref, watch, Ref, PropType,
 } from 'vue';
 
 import Viewer from 'dive-common/components/Viewer.vue';
@@ -19,6 +19,7 @@ import { useJobs } from 'platform/web-girder/store/useJobs';
 import { usePrompt } from 'dive-common/vue-utilities/prompt-service';
 import type { DatasetCalibrationResult, DatasetType, SubType } from 'dive-common/apispec';
 import { useApi } from 'dive-common/apispec';
+import type { StereoAnnotationCompleteParams } from 'dive-common/use/useModeManager';
 import { parentDatasetId } from 'dive-common/compositeDatasetId';
 import { getMultiCamCameraCount } from 'dive-common/pipelineMenuFilters';
 import { webExcludedPipelineTerms } from 'dive-common/constants';
@@ -27,6 +28,12 @@ import { useRouter, useRoute } from 'vue-router/composables';
 import { ANNOTATION_SOURCE_QUERY } from 'dive-common/scoring/viewerNavigation';
 import { parseViewerFocus } from 'dive-common/review/viewerNavigation';
 import useStereoOnnxWeb from 'platform/web-girder/useStereoOnnxWeb';
+import useServerSegmentation from 'platform/web-girder/useServerSegmentation';
+import StereoServerMatcher from 'dive-common/use/stereo/StereoServerMatcher';
+import type { StereoMatchMethod } from 'dive-common/use/stereo/stereoMatcher';
+import {
+  interactiveStereoSetFrame, interactiveStereoTransferPoints,
+} from 'platform/web-girder/api/interactive.service';
 import type { StereoModelProgress } from 'platform/web-girder/useStereoOnnxWeb';
 import {
   STEREO_LENGTH_METHOD_ATTR, STEREO_MEASUREMENT_ATTRS,
@@ -176,11 +183,29 @@ export default defineComponent({
     }
 
     const {
-      handleStereoAnnotationComplete, handleStereoTrackLinked, warpAllFromCamera,
-      invalidateCalibration, stereoViewLink,
+      interactiveEnabled, interactiveMessage, refreshInteractive,
+    } = useConfig();
+
+    // Stereo matching runs in the server's interactive service when it is up;
+    // otherwise the browser models do the work as before.
+    const serverStereo = {
+      enabled: () => interactiveEnabled.value,
+      matcher: (method: StereoMatchMethod) => new StereoServerMatcher({
+        transferPoints: (request) => interactiveStereoTransferPoints({
+          datasetId: parentDatasetId(props.id), ...request, strict: true,
+        }),
+        setFrame: (frame, method2) => interactiveStereoSetFrame(parentDatasetId(props.id), frame, method2),
+      }, method),
+    };
+
+    const {
+      handleStereoAnnotationComplete: transferStereoAnnotation,
+      handleStereoTrackLinked, warpAllFromCamera,
+      invalidateCalibration, stereoViewLink, refreshMeasurement: refreshStereoMeasurement,
     } = useStereoOnnxWeb({
       getViewer: () => viewerRef.value,
       getDatasetId: () => parentDatasetId(props.id),
+      server: serverStereo,
       ensureMeasurementAttributes,
       onStatus: (message, progress) => {
         stereoBusyMessage.value = message;
@@ -205,6 +230,27 @@ export default defineComponent({
       stereoError.value = '';
     }
 
+    const serverSegmentation = useServerSegmentation({
+      getViewer: () => viewerRef.value,
+      getDatasetId: () => parentDatasetId(props.id),
+      isEnabled: () => interactiveEnabled.value,
+      disabledMessage: () => interactiveMessage.value,
+      refreshMeasurement: refreshStereoMeasurement,
+      onError: (message) => { stereoError.value = message; },
+    });
+
+    /** Masks go through the service; boxes, lines and points through the transfer. */
+    async function handleStereoAnnotationComplete(params: StereoAnnotationCompleteParams) {
+      if (await serverSegmentation.handleStereoSegmentation(params)) return 'transferred';
+      return transferStereoAnnotation(params);
+    }
+
+    async function connectInteractive() {
+      await refreshInteractive();
+      if (interactiveEnabled.value) await nextTick();
+      serverSegmentation.initialize();
+    }
+
     const stereoDownloadPercent = computed(() => {
       const progress = stereoDownloadProgress.value;
       if (!progress) return 0;
@@ -294,6 +340,8 @@ export default defineComponent({
       loadDataset(datasetId).catch((reason) => {
         reportHandledPromiseRejection('ViewerLoader: loadDataset', reason);
       });
+      // <Viewer :key="id"> remounts with a fresh recipe to hook up.
+      nextTick().then(() => serverSegmentation.initialize());
     }, { immediate: true });
 
     // Seed as soon as parent meta arrives so pipeline menus don't wait on Viewer emit.
@@ -415,6 +463,7 @@ export default defineComponent({
     });
 
     onMounted(() => {
+      connectInteractive().catch(() => {});
       window.addEventListener('beforeunload', viewerRef.value.warnBrowserExit);
     });
 
diff --git a/docker-compose.override.yml b/docker-compose.override.yml
index f1afcb61c..430aa8228 100644
--- a/docker-compose.override.yml
+++ b/docker-compose.override.yml
@@ -24,3 +24,7 @@ services:
   localworker:
     volumes:
       - ./server:/opt/dive/src
+
+  interactive:
+    volumes:
+      - ./server:/opt/dive/src
diff --git a/docker-compose.yml b/docker-compose.yml
index 293b0c5a2..378f0f65f 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -112,6 +112,8 @@ services:
       - "GIRDER_STATIC_ROOT_DIR=/opt/dive/clients/girder"
       - "GIRDER_NOTIFICATION_REDIS_URL=redis://redis:6379"
       - "DIVE_MODEL_CACHE_DIR=/tmp/dive_models"
+      # Server-side interactive segmentation and stereo (the `interactive` service)
+      - "DIVE_INTERACTIVE_URL=${DIVE_INTERACTIVE_URL:-http://interactive:8800}"
 
       # Rabbitmq management variables
       - "RABBITMQ_MANAGEMENT_USERNAME=${RABBITMQ_MANAGEMENT_USERNAME:-guest}"
@@ -186,6 +188,38 @@ services:
     labels:
       - autoheal=true
 
+  # Interactive segmentation and stereo for the web client: keeps VIAME's
+  # interactive service resident on the GPU and answers girder's requests.
+  # Sessions are capped and retired when idle; girder pauses the tools while
+  # pipeline or training jobs hold the GPU.
+  interactive:
+    << : *base-worker
+    profiles:
+      - gpu
+    build:
+      context: .
+      dockerfile: docker/girder_worker_gpu.Dockerfile
+    image: kitware/viame-worker:gpu
+    restart: always
+    entrypoint: ["uv", "run", "--no-sync", "python", "-m", "dive_interactive"]
+    volumes:
+      - addons:/tmp/addons:ro
+      - girder_assetstore:/home/assetstore:ro
+    deploy:
+      resources:
+        reservations:
+          devices:
+          - driver: nvidia
+            device_ids: ['0']
+            capabilities: [gpu]
+    environment:
+      - "WORKER_GPU_UUID=${INTERACTIVE_GPU_UUID:-${PIPELINE_GPU_UUID}}"
+      - "DIVE_INTERACTIVE_PORT=8800"
+      - "DIVE_INTERACTIVE_MAX_SESSIONS=${DIVE_INTERACTIVE_MAX_SESSIONS:-2}"
+      - "DIVE_INTERACTIVE_IDLE_SECONDS=${DIVE_INTERACTIVE_IDLE_SECONDS:-600}"
+      - "DIVE_INTERACTIVE_QUEUE_TIMEOUT=${DIVE_INTERACTIVE_QUEUE_TIMEOUT:-30}"
+      - "DIVE_INTERACTIVE_MAX_INFLIGHT=${DIVE_INTERACTIVE_MAX_INFLIGHT:-1}"
+
   girder_worker_training:
     # Merge base-worker object with this config
     << : *base-worker
diff --git a/docs/Deployment-Docker-Compose.md b/docs/Deployment-Docker-Compose.md
index 18b330ee2..84894ca47 100644
--- a/docs/Deployment-Docker-Compose.md
+++ b/docs/Deployment-Docker-Compose.md
@@ -92,6 +92,12 @@ When developing with `docker-compose.override.yml`, the same service mounts your
 
 ![Login Page](images/General/login.png)
 
+### `interactive`
+
+The `interactive` service (GPU profile) runs VIAME's interactive segmentation and stereo service on the GPU host so the web client's Segment tool, mask keypoints, text query and stereo point mapping run server-side. It mounts the assetstore read-only and answers girder over `DIVE_INTERACTIVE_URL`.
+
+It keeps at most `DIVE_INTERACTIVE_MAX_SESSIONS` service processes resident (one per user and dataset), retires a process idle for `DIVE_INTERACTIVE_IDLE_SECONDS`, runs `DIVE_INTERACTIVE_MAX_INFLIGHT` inferences at a time, and refuses a request that cannot get a turn within `DIVE_INTERACTIVE_QUEUE_TIMEOUT` seconds instead of queueing it. Girder pauses the tools entirely while any pipeline or training job is queued or running, since they share the GPU; `dive_configuration` reports this as `interactiveEnabled` with a message. Set `INTERACTIVE_GPU_UUID` to pin it to a different GPU than the job workers.
+
 ## Production deployment
 
 If you have a server with a **public-facing IP address** and a **domain name** that points to it, you should be able to use our production deployment configuration.  This is the way we deploy viame.kitware.com.
diff --git a/server/dive_interactive/__init__.py b/server/dive_interactive/__init__.py
new file mode 100644
index 000000000..a9aa27c1c
--- /dev/null
+++ b/server/dive_interactive/__init__.py
@@ -0,0 +1,5 @@
+"""Server-side interactive segmentation and stereo for the web client.
+
+A broker keeps VIAME `interactive_service` processes on the GPU host and
+relays JSON requests to them; girder resolves media to paths and forwards.
+"""
diff --git a/server/dive_interactive/__main__.py b/server/dive_interactive/__main__.py
new file mode 100644
index 000000000..333d27842
--- /dev/null
+++ b/server/dive_interactive/__main__.py
@@ -0,0 +1,3 @@
+from .broker import serve
+
+serve()
diff --git a/server/dive_interactive/broker.py b/server/dive_interactive/broker.py
new file mode 100644
index 000000000..96d6af1de
--- /dev/null
+++ b/server/dive_interactive/broker.py
@@ -0,0 +1,250 @@
+"""HTTP front for the interactive sessions, run on the GPU host.
+
+    POST   /sessions/<key>/request   {"command", "params", "timeout"?}
+    GET    /sessions/<key>/events?since=N
+    DELETE /sessions/<key>
+    GET    /health
+
+Configuration comes from the environment:
+    VIAME_INSTALL_PATH                  VIAME install (default /opt/noaa/viame)
+    ADDON_ROOT_DIR                      extracted add-ons (default /tmp/addons/extracted)
+    DIVE_INTERACTIVE_PORT               listen port (default 8800)
+    DIVE_INTERACTIVE_MAX_SESSIONS       resident service processes (default 2)
+    DIVE_INTERACTIVE_IDLE_SECONDS       retire a session idle this long (default 600)
+    DIVE_INTERACTIVE_QUEUE_TIMEOUT      seconds a request may wait for a turn (default 30)
+    DIVE_INTERACTIVE_MAX_INFLIGHT       inferences running at once (default 1)
+    DIVE_INTERACTIVE_DEVICE             cuda | cpu | auto (default cuda)
+    DIVE_INTERACTIVE_SEGMENTATION_CONFIG  explicit segmenter config path (optional)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import shlex
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+from urllib.parse import parse_qs, urlparse
+
+from .manager import BusyError, ServiceError, SessionManager
+from .service import InteractiveProcess
+
+logger = logging.getLogger('dive_interactive')
+
+# Stereo configs per client method, the desktop's STEREO_METHOD_FILES.
+STEREO_CONFIGS = {
+    'foundation': 'interactive_stereo_fast_fdn_stereo.conf',
+    'dino': 'interactive_stereo_ncc_dino.conf',
+    'ncc': 'interactive_stereo_template.conf',
+}
+SEGMENTATION_CONFIGS = [
+    'interactive_segmenter_sam2.conf',
+    'interactive_segmenter_sam3.conf',
+    'interactive_segmenter_default.conf',
+]
+TEXT_QUERY_CONFIG = 'interactive_text_query_sam3.conf'
+
+
+def _env_float(name: str, default: float) -> float:
+    try:
+        return float(os.environ.get(name, default))
+    except ValueError:
+        return default
+
+
+def _env_int(name: str, default: int) -> int:
+    try:
+        return int(os.environ.get(name, default))
+    except ValueError:
+        return default
+
+
+class Launcher:
+    """Builds the command for one service process from the install layout."""
+
+    def __init__(self) -> None:
+        self.viame = Path(os.environ.get('VIAME_INSTALL_PATH', '/opt/noaa/viame'))
+        addon_root = Path(os.environ.get('ADDON_ROOT_DIR', '/tmp/addons/extracted'))
+        # Add-ons unpack into their own tree, which the service's own
+        # auto-discovery never looks at, so configs are found here explicitly.
+        self.pipeline_dirs = [addon_root / 'configs' / 'pipelines', self.viame / 'configs' / 'pipelines']
+        self.device = os.environ.get('DIVE_INTERACTIVE_DEVICE', 'cuda')
+
+    def find_config(self, name: str) -> Optional[Path]:
+        for directory in self.pipeline_dirs:
+            candidate = directory / name
+            if candidate.is_file():
+                return candidate
+        return None
+
+    def segmentation_configs(self) -> List[Path]:
+        explicit = os.environ.get('DIVE_INTERACTIVE_SEGMENTATION_CONFIG')
+        configs: List[Path] = []
+        if explicit:
+            configs.append(Path(explicit))
+        else:
+            for name in SEGMENTATION_CONFIGS:
+                found = self.find_config(name)
+                if found:
+                    configs.append(found)
+                    break
+        text_query = self.find_config(TEXT_QUERY_CONFIG)
+        if text_query:
+            configs.append(text_query)
+        return configs
+
+    def stereo_config(self, method: str) -> Optional[Path]:
+        name = STEREO_CONFIGS.get(method, STEREO_CONFIGS['ncc'])
+        return self.find_config(name) or self.find_config('interactive_stereo_default.conf')
+
+    def capabilities(self) -> Dict[str, Any]:
+        return {
+            'segmentation': [str(c) for c in self.segmentation_configs()],
+            'stereoMethods': [m for m in STEREO_CONFIGS if self.find_config(STEREO_CONFIGS[m])],
+            'textQuery': self.find_config(TEXT_QUERY_CONFIG) is not None,
+        }
+
+    def spawn(self, info: Dict[str, Any]) -> InteractiveProcess:
+        parts = [
+            'python', '-s', '-m', 'viame.core.interactive_service',
+            '--viame-path', str(self.viame),
+            '--device', self.device,
+        ]
+        for config in self.segmentation_configs():
+            parts += ['--segmentation-config', str(config)]
+        stereo = self.stereo_config(str(info.get('stereoMethod', 'ncc')))
+        if stereo:
+            parts += ['--stereo-config', str(stereo)]
+        setup = self.viame / 'setup_viame.sh'
+        command = f'. {shlex.quote(str(setup))} && exec {" ".join(shlex.quote(p) for p in parts)}'
+        env = dict(os.environ)
+        gpu = os.environ.get('WORKER_GPU_UUID')
+        if gpu:
+            env['CUDA_VISIBLE_DEVICES'] = gpu
+        return InteractiveProcess(
+            ['bash', '-c', command],
+            cwd=str(self.viame),
+            env=env,
+            ready_timeout=_env_float('DIVE_INTERACTIVE_READY_TIMEOUT', 300.0),
+        )
+
+
+def build_manager(launcher: Optional[Launcher] = None) -> SessionManager:
+    launcher = launcher or Launcher()
+    return SessionManager(
+        spawn=launcher.spawn,
+        max_sessions=_env_int('DIVE_INTERACTIVE_MAX_SESSIONS', 2),
+        idle_seconds=_env_float('DIVE_INTERACTIVE_IDLE_SECONDS', 600.0),
+        queue_timeout=_env_float('DIVE_INTERACTIVE_QUEUE_TIMEOUT', 30.0),
+        max_inflight=_env_int('DIVE_INTERACTIVE_MAX_INFLIGHT', 1),
+        request_timeout=_env_float('DIVE_INTERACTIVE_REQUEST_TIMEOUT', 300.0),
+    )
+
+
+def make_handler(manager: SessionManager, launcher: Launcher):
+    class Handler(BaseHTTPRequestHandler):
+        server_version = 'DiveInteractive/1'
+
+        def log_message(self, fmt: str, *args: Any) -> None:  # quieter than the default
+            logger.debug(fmt, *args)
+
+        def _send(self, status: int, body: Dict[str, Any]) -> None:
+            data = json.dumps(body).encode()
+            self.send_response(status)
+            self.send_header('Content-Type', 'application/json')
+            self.send_header('Content-Length', str(len(data)))
+            self.end_headers()
+            self.wfile.write(data)
+
+        def _read_json(self) -> Dict[str, Any]:
+            length = int(self.headers.get('Content-Length') or 0)
+            raw = self.rfile.read(length) if length else b''
+            return json.loads(raw or b'{}')
+
+        def _session_key(self) -> Optional[str]:
+            parts = urlparse(self.path).path.strip('/').split('/')
+            if len(parts) >= 2 and parts[0] == 'sessions' and parts[1]:
+                return parts[1]
+            return None
+
+        def do_GET(self) -> None:  # noqa: N802
+            url = urlparse(self.path)
+            if url.path == '/health':
+                self._send(200, {'ok': True, **manager.status(), **launcher.capabilities()})
+                return
+            key = self._session_key()
+            if key and url.path.endswith('/events'):
+                since = int(parse_qs(url.query).get('since', ['0'])[0] or 0)
+                self._send(200, manager.events(key, since))
+                return
+            self._send(404, {'error': 'not found'})
+
+        def do_POST(self) -> None:  # noqa: N802
+            key = self._session_key()
+            if not key or not urlparse(self.path).path.endswith('/request'):
+                self._send(404, {'error': 'not found'})
+                return
+            try:
+                body = self._read_json()
+                command = str(body.get('command') or '')
+                if not command:
+                    self._send(400, {'error': 'command is required'})
+                    return
+                response = manager.request(
+                    key,
+                    command,
+                    dict(body.get('params') or {}),
+                    info=dict(body.get('session') or {}),
+                    timeout=body.get('timeout'),
+                )
+                self._send(200, response)
+            except BusyError as err:
+                self._send(503, {'error': str(err), 'busy': True})
+            except ServiceError as err:
+                self._send(502, {'error': str(err)})
+            except Exception as err:  # noqa: BLE001
+                logger.exception('interactive request failed')
+                self._send(500, {'error': str(err)})
+
+        def do_DELETE(self) -> None:  # noqa: N802
+            key = self._session_key()
+            if not key:
+                self._send(404, {'error': 'not found'})
+                return
+            manager.drop(key)
+            self._send(200, {'ok': True})
+
+    return Handler
+
+
+def serve(port: Optional[int] = None) -> None:
+    logging.basicConfig(level=os.environ.get('DIVE_INTERACTIVE_LOG_LEVEL', 'INFO'))
+    launcher = Launcher()
+    manager = build_manager(launcher)
+
+    def reaper() -> None:
+        while True:
+            time.sleep(15)
+            try:
+                manager.reap_idle()
+            except Exception:  # noqa: BLE001
+                logger.exception('reaper failed')
+
+    threading.Thread(target=reaper, daemon=True).start()
+    server = ThreadingHTTPServer(
+        ('0.0.0.0', port or _env_int('DIVE_INTERACTIVE_PORT', 8800)),
+        make_handler(manager, launcher),
+    )
+    logger.info('interactive broker listening on %s', server.server_address)
+    try:
+        server.serve_forever()
+    finally:
+        manager.stop_all()
+
+
+if __name__ == '__main__':
+    serve()
diff --git a/server/dive_interactive/manager.py b/server/dive_interactive/manager.py
new file mode 100644
index 000000000..3027c2751
--- /dev/null
+++ b/server/dive_interactive/manager.py
@@ -0,0 +1,189 @@
+"""Sessions of the interactive service, bounded so they cannot crowd the GPU.
+
+Each session is one service process holding its models resident. The manager
+caps how many exist, retires idle ones, runs one inference at a time per
+process, and bounds how many inferences run at once across all of them; a
+request that cannot get a turn quickly is refused rather than queued forever.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, Optional
+
+from .service import InteractiveProcess, ServiceError
+
+
+class BusyError(RuntimeError):
+    """The server is at its interactive capacity right now."""
+
+
+@dataclass
+class Session:
+    key: str
+    process: InteractiveProcess
+    generation: int
+    created: float = field(default_factory=time.monotonic)
+    last_used: float = field(default_factory=time.monotonic)
+    lock: threading.Lock = field(default_factory=threading.Lock)
+    inflight: int = 0
+    info: Dict[str, Any] = field(default_factory=dict)
+
+    def touch(self) -> None:
+        self.last_used = time.monotonic()
+
+
+class SessionManager:
+    def __init__(
+        self,
+        spawn: Callable[[Dict[str, Any]], InteractiveProcess],
+        max_sessions: int = 2,
+        idle_seconds: float = 600.0,
+        queue_timeout: float = 30.0,
+        max_inflight: int = 1,
+        request_timeout: float = 300.0,
+    ):
+        self._spawn = spawn
+        self.max_sessions = max_sessions
+        self.idle_seconds = idle_seconds
+        self.queue_timeout = queue_timeout
+        self.request_timeout = request_timeout
+        self._sessions: Dict[str, Session] = {}
+        self._lock = threading.Lock()
+        self._inflight = threading.BoundedSemaphore(max_inflight)
+        self._generation = 0
+
+    # ---- lifecycle ---------------------------------------------------------
+
+    def ensure(self, key: str, info: Optional[Dict[str, Any]] = None) -> Session:
+        """The session for `key`, started if needed; evicts an idle one for room."""
+        with self._lock:
+            session = self._sessions.get(key)
+            if session is not None and session.process.alive:
+                session.touch()
+                return session
+            if session is not None:
+                self._sessions.pop(key, None)
+            if len(self._sessions) >= self.max_sessions:
+                victim = self._idle_victim()
+                if victim is None:
+                    raise BusyError(
+                        'All interactive sessions are in use; try again in a moment.'
+                    )
+                self._sessions.pop(victim.key, None)
+                self._stop_later(victim)
+            self._generation += 1
+            generation = self._generation
+        process = self._spawn(info or {})
+        process.start()
+        session = Session(key=key, process=process, generation=generation, info=dict(info or {}))
+        with self._lock:
+            existing = self._sessions.get(key)
+            if existing is not None and existing.process.alive:
+                process.stop()
+                existing.touch()
+                return existing
+            self._sessions[key] = session
+        return session
+
+    def _idle_victim(self) -> Optional[Session]:
+        idle = [s for s in self._sessions.values() if s.inflight == 0]
+        if not idle:
+            return None
+        return min(idle, key=lambda s: s.last_used)
+
+    @staticmethod
+    def _stop_later(session: Session) -> None:
+        threading.Thread(target=session.process.stop, daemon=True).start()
+
+    def drop(self, key: str) -> None:
+        with self._lock:
+            session = self._sessions.pop(key, None)
+        if session is not None:
+            self._stop_later(session)
+
+    def reap_idle(self, now: Optional[float] = None) -> int:
+        now = time.monotonic() if now is None else now
+        with self._lock:
+            stale = [
+                s for s in self._sessions.values()
+                if s.inflight == 0 and now - s.last_used > self.idle_seconds
+            ]
+            for session in stale:
+                self._sessions.pop(session.key, None)
+        for session in stale:
+            self._stop_later(session)
+        return len(stale)
+
+    def stop_all(self) -> None:
+        with self._lock:
+            sessions = list(self._sessions.values())
+            self._sessions.clear()
+        for session in sessions:
+            session.process.stop()
+
+    # ---- requests ----------------------------------------------------------
+
+    def request(
+        self,
+        key: str,
+        command: str,
+        params: Dict[str, Any],
+        info: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+    ) -> Dict[str, Any]:
+        session = self.ensure(key, info)
+        if not self._inflight.acquire(timeout=self.queue_timeout):
+            raise BusyError(
+                'The interactive service is busy with other requests; try again in a moment.'
+            )
+        try:
+            with self._lock:
+                session.inflight += 1
+            try:
+                with session.lock:
+                    session.touch()
+                    response = session.process.request(
+                        command, params, timeout or self.request_timeout
+                    )
+            finally:
+                with self._lock:
+                    session.inflight -= 1
+                    session.touch()
+        finally:
+            self._inflight.release()
+        if not session.process.alive:
+            self.drop(key)
+        response['_session'] = {'generation': session.generation}
+        return response
+
+    def events(self, key: str, since: int) -> Dict[str, Any]:
+        with self._lock:
+            session = self._sessions.get(key)
+        if session is None:
+            return {'events': [], 'next': since, 'generation': None}
+        events = session.process.events_since(since)
+        return {'events': events, 'next': since + len(events), 'generation': session.generation}
+
+    def status(self) -> Dict[str, Any]:
+        with self._lock:
+            sessions = [
+                {
+                    'key': s.key,
+                    'generation': s.generation,
+                    'inflight': s.inflight,
+                    'idleSeconds': round(time.monotonic() - s.last_used, 1),
+                    'alive': s.process.alive,
+                }
+                for s in self._sessions.values()
+            ]
+        return {
+            'sessions': sessions,
+            'maxSessions': self.max_sessions,
+            'idleSeconds': self.idle_seconds,
+        }
+
+
+__all__ = ['BusyError', 'Session', 'SessionManager', 'ServiceError']
diff --git a/server/dive_interactive/service.py b/server/dive_interactive/service.py
new file mode 100644
index 000000000..d8d92682d
--- /dev/null
+++ b/server/dive_interactive/service.py
@@ -0,0 +1,169 @@
+"""One VIAME interactive_service process, spoken to over stdin/stdout."""
+
+from __future__ import annotations
+
+import json
+import logging
+import subprocess
+import threading
+import time
+from collections import deque
+from typing import Any, Deque, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+READY_MARKER = 'Service started, waiting for requests'
+
+
+class ServiceError(RuntimeError):
+    pass
+
+
+class InteractiveProcess:
+    """Spawns the service and matches its responses to requests by id.
+
+    The service answers one line per request but not in order: some stereo
+    replies arrive later from its background threads, and `set_frame` later
+    emits events that reuse the request's id. A reply for an id nobody is
+    waiting on is kept as an event.
+    """
+
+    def __init__(
+        self,
+        command: List[str] | str,
+        cwd: Optional[str] = None,
+        env: Optional[Dict[str, str]] = None,
+        ready_timeout: float = 300.0,
+        shell: bool = False,
+    ):
+        self.command = command
+        self.cwd = cwd
+        self.env = env
+        self.ready_timeout = ready_timeout
+        self.shell = shell
+        self.process: Optional[subprocess.Popen[str]] = None
+        self.ready = threading.Event()
+        self.exited = threading.Event()
+        self.stderr_tail: Deque[str] = deque(maxlen=40)
+        self.events: List[Dict[str, Any]] = []
+        self._pending: Dict[str, Dict[str, Any]] = {}
+        self._waiters: Dict[str, threading.Event] = {}
+        self._lock = threading.Lock()
+        self._write_lock = threading.Lock()
+        self._counter = 0
+
+    def start(self) -> None:
+        self.process = subprocess.Popen(
+            self.command,
+            cwd=self.cwd,
+            env=self.env,
+            shell=self.shell,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1,
+        )
+        threading.Thread(target=self._read_stdout, daemon=True).start()
+        threading.Thread(target=self._read_stderr, daemon=True).start()
+        if not self.ready.wait(self.ready_timeout):
+            self.stop()
+            raise ServiceError(
+                'The interactive service did not start in time. '
+                + ' '.join(list(self.stderr_tail)[-5:])
+            )
+        if self.exited.is_set():
+            raise ServiceError(
+                'The interactive service exited on startup. '
+                + ' '.join(list(self.stderr_tail)[-5:])
+            )
+
+    @property
+    def alive(self) -> bool:
+        return self.process is not None and self.process.poll() is None and not self.exited.is_set()
+
+    def request(self, command: str, params: Dict[str, Any], timeout: float = 300.0) -> Dict[str, Any]:
+        if not self.alive:
+            raise ServiceError('The interactive service is not running.')
+        with self._lock:
+            self._counter += 1
+            request_id = f'req_{int(time.time() * 1000)}_{self._counter}'
+            waiter = threading.Event()
+            self._waiters[request_id] = waiter
+        payload = {'id': request_id, 'command': command, **params}
+        line = json.dumps(payload) + '\n'
+        with self._write_lock:
+            assert self.process is not None and self.process.stdin is not None
+            self.process.stdin.write(line)
+            self.process.stdin.flush()
+        if not waiter.wait(timeout):
+            with self._lock:
+                self._waiters.pop(request_id, None)
+            raise ServiceError(f'The interactive service did not answer "{command}" in time.')
+        with self._lock:
+            response = self._pending.pop(request_id, None)
+            self._waiters.pop(request_id, None)
+        if response is None:
+            raise ServiceError('The interactive service exited before answering.')
+        return response
+
+    def events_since(self, index: int) -> List[Dict[str, Any]]:
+        with self._lock:
+            return list(self.events[index:])
+
+    def stop(self) -> None:
+        process = self.process
+        if process is None:
+            return
+        try:
+            if process.poll() is None and process.stdin:
+                with self._write_lock:
+                    process.stdin.write(json.dumps({'id': 'shutdown', 'command': 'shutdown'}) + '\n')
+                    process.stdin.flush()
+                process.wait(timeout=5)
+        except Exception:
+            pass
+        if process.poll() is None:
+            process.terminate()
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                process.kill()
+        self._release_waiters()
+
+    def _release_waiters(self) -> None:
+        self.exited.set()
+        with self._lock:
+            waiters = list(self._waiters.values())
+        for waiter in waiters:
+            waiter.set()
+
+    def _read_stdout(self) -> None:
+        assert self.process is not None and self.process.stdout is not None
+        for raw in self.process.stdout:
+            raw = raw.strip()
+            if not raw:
+                continue
+            try:
+                message = json.loads(raw)
+            except json.JSONDecodeError:
+                logger.warning('interactive service wrote non-JSON: %s', raw[:200])
+                continue
+            with self._lock:
+                request_id = message.get('id')
+                waiter = self._waiters.get(request_id) if request_id else None
+                if waiter is not None and request_id not in self._pending:
+                    self._pending[request_id] = message
+                    waiter.set()
+                else:
+                    self.events.append(message)
+        self._release_waiters()
+
+    def _read_stderr(self) -> None:
+        assert self.process is not None and self.process.stderr is not None
+        for raw in self.process.stderr:
+            line = raw.rstrip()
+            self.stderr_tail.append(line)
+            if READY_MARKER in line:
+                self.ready.set()
+        self._release_waiters()
diff --git a/server/dive_server/__init__.py b/server/dive_server/__init__.py
index 0cda82a9c..fa7c5bdf3 100644
--- a/server/dive_server/__init__.py
+++ b/server/dive_server/__init__.py
@@ -17,6 +17,7 @@
 from .views_annotation import AnnotationResource
 from .views_configuration import ConfigurationResource
 from .views_dataset import DatasetResource
+from .views_interactive import InteractiveResource
 from .views_override import (
     countJobs,
     get_root_path_or_relative,
@@ -38,6 +39,7 @@ def load(self, info):
         info["apiRoot"].dive_annotation = AnnotationResource("dive_annotation")
         info["apiRoot"].dive_configuration = ConfigurationResource("dive_configuration")
         info["apiRoot"].dive_dataset = DatasetResource("dive_dataset")
+        info["apiRoot"].dive_interactive = InteractiveResource("dive_interactive")
         info["apiRoot"].dive_rpc = RpcResource("dive_rpc")
         info["apiRoot"].dive_scoring = ScoringResource("dive_scoring")
         # required because girder doesn't load plugins in order so we need to manually load first.
diff --git a/server/dive_server/interactive_gate.py b/server/dive_server/interactive_gate.py
new file mode 100644
index 000000000..7ce684899
--- /dev/null
+++ b/server/dive_server/interactive_gate.py
@@ -0,0 +1,91 @@
+"""Whether the server-side interactive tools may be used right now."""
+
+from __future__ import annotations
+
+import os
+import time
+from typing import Any, Dict, Optional
+
+from girder.exceptions import RestException
+from girder_jobs.models.job import Job, JobStatus
+from girder_plugin_worker.status import CustomJobStatus
+import requests
+
+BUSY_MESSAGE = 'Interactive segmentation and stereo pause while pipeline or training jobs run.'
+OFFLINE_MESSAGE = 'The interactive service is not available on this server.'
+
+# Job types that hold the GPU the interactive service shares.
+GPU_JOB_TYPES = ['pipelines', 'training', 'private']
+ACTIVE_STATUSES = [
+    JobStatus.INACTIVE,
+    JobStatus.QUEUED,
+    JobStatus.RUNNING,
+    CustomJobStatus.CANCELING,
+    CustomJobStatus.CONVERTING_OUTPUT,
+    CustomJobStatus.CONVERTING_INPUT,
+    CustomJobStatus.FETCHING_INPUT,
+    CustomJobStatus.PUSHING_OUTPUT,
+]
+
+_health_cache: Dict[str, Any] = {'at': 0.0, 'value': None}
+HEALTH_TTL = 30.0
+
+
+def broker_url() -> Optional[str]:
+    url = os.environ.get('DIVE_INTERACTIVE_URL', '').strip()
+    return url.rstrip('/') or None
+
+
+def broker_health(force: bool = False) -> Optional[Dict[str, Any]]:
+    """The broker's /health, cached briefly; None when unreachable or unset."""
+    url = broker_url()
+    if not url:
+        return None
+    now = time.monotonic()
+    if not force and now - _health_cache['at'] < HEALTH_TTL:
+        return _health_cache['value']
+    value: Optional[Dict[str, Any]] = None
+    try:
+        response = requests.get(f'{url}/health', timeout=3)
+        response.raise_for_status()
+        value = response.json()
+    except (requests.RequestException, ValueError):
+        value = None
+    _health_cache.update(at=now, value=value)
+    return value
+
+
+def gpu_jobs_active() -> bool:
+    return (
+        Job().findOne({'type': {'$in': GPU_JOB_TYPES}, 'status': {'$in': ACTIVE_STATUSES}})
+        is not None
+    )
+
+
+def get_interactive_capability() -> Dict[str, Any]:
+    """What `dive_configuration` reports to the client."""
+    health = broker_health()
+    if health is None:
+        return {'interactiveEnabled': False, 'interactiveMessage': OFFLINE_MESSAGE}
+    if gpu_jobs_active():
+        return {
+            'interactiveEnabled': False,
+            'interactiveMessage': BUSY_MESSAGE,
+            'interactiveStereoMethods': health.get('stereoMethods', []),
+        }
+    return {
+        'interactiveEnabled': True,
+        'interactiveMessage': '',
+        'interactiveStereoMethods': health.get('stereoMethods', []),
+        'interactiveTextQuery': bool(health.get('textQuery')),
+    }
+
+
+def require_interactive_available() -> str:
+    """The broker URL to use, or a 503 saying why not."""
+    url = broker_url()
+    if not url or broker_health() is None:
+        raise RestException(OFFLINE_MESSAGE, code=503)
+    if gpu_jobs_active():
+        raise RestException(BUSY_MESSAGE, code=503)
+    return url
diff --git a/server/dive_server/interactive_media.py b/server/dive_server/interactive_media.py
new file mode 100644
index 000000000..b640b224f
--- /dev/null
+++ b/server/dive_server/interactive_media.py
@@ -0,0 +1,93 @@
+"""Resolve a dataset frame or calibration to a path the interactive service
+can open. The service runs on a host that mounts the same assetstore girder
+writes to, so the assetstore's own file paths are used as they are."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from girder.constants import AccessType
+from girder.exceptions import RestException
+from girder.models.file import File
+from girder.models.folder import Folder
+from girder.models.item import Item
+
+from dive_server import crud, crud_dataset
+from dive_utils import constants, fromMeta, types
+
+
+def item_path(item: types.GirderModel) -> str:
+    file = File().findOne({'itemId': item['_id']})
+    if file is None:
+        raise RestException(f'Item {item["name"]} has no file', code=404)
+    adapter = File().getAssetstoreAdapter(file)
+    if not hasattr(adapter, 'fullPath'):
+        raise RestException('Interactive tools need a filesystem assetstore', code=501)
+    return adapter.fullPath(file)
+
+
+def camera_order(dsFolder: types.GirderModel) -> List[str]:
+    multi_cam = fromMeta(dsFolder, constants.MultiCamMarker) or {}
+    if not multi_cam:
+        return []
+    # The calibration's left camera is first, as the client also assumes.
+    return list(crud_dataset._multicam_camera_order(multi_cam))  # noqa: SLF001
+
+
+def camera_folder(
+    dsFolder: types.GirderModel, camera: Optional[str], user: types.GirderUserModel
+) -> types.GirderModel:
+    if fromMeta(dsFolder, constants.TypeMarker) != constants.MultiType:
+        return dsFolder
+    multi_cam = fromMeta(dsFolder, constants.MultiCamMarker) or {}
+    cameras = multi_cam.get('cameras') or {}
+    name = camera or multi_cam.get('defaultDisplay')
+    info = cameras.get(name) if name else None
+    if not info or not info.get('folderId'):
+        raise RestException(f'Unknown camera "{camera}"', code=400)
+    folder = Folder().load(info['folderId'], level=AccessType.READ, user=user)
+    if folder is None:
+        raise RestException(f'Camera folder for "{name}" was not found', code=404)
+    return folder
+
+
+def frame_media(
+    folder: types.GirderModel, frame: int, user: types.GirderUserModel
+) -> Dict[str, Any]:
+    """`{'path', 'frameTime'}` for one frame of a single-camera folder."""
+    source_type = fromMeta(folder, constants.TypeMarker)
+    if source_type == constants.VideoType:
+        root = crud.getCloneRoot(user, folder)
+        video = Item().findOne(
+            {'folderId': root['_id'], 'meta.codec': 'h264', 'meta.source_video': {'$in': [None, False]}}
+        )
+        if video is None:
+            raise RestException('The dataset has no transcoded video', code=404)
+        fps = fromMeta(folder, constants.FPSMarker) or fromMeta(root, constants.FPSMarker)
+        if not fps:
+            raise RestException('The dataset has no frame rate', code=400)
+        return {'path': item_path(video), 'frameTime': frame / float(fps)}
+    if source_type == constants.ImageSequenceType:
+        images = crud.valid_images(folder, user)
+        if frame < 0 or frame >= len(images):
+            raise RestException(f'Frame {frame} is outside the dataset', code=400)
+        return {'path': item_path(images[frame]), 'frameTime': None}
+    raise RestException(
+        f'Interactive tools do not support {source_type} datasets', code=501
+    )
+
+
+def resolve_frame(
+    dsFolder: types.GirderModel, camera: Optional[str], frame: int, user: types.GirderUserModel
+) -> Dict[str, Any]:
+    return frame_media(camera_folder(dsFolder, camera, user), frame, user)
+
+
+def calibration_path(dsFolder: types.GirderModel, user: types.GirderUserModel) -> Optional[str]:
+    item_id = crud_dataset.find_calibration_item_id(str(dsFolder['_id']), dsFolder)
+    if not item_id:
+        return None
+    item = Item().load(item_id, level=AccessType.READ, user=user)
+    if item is None:
+        return None
+    return item_path(item)
diff --git a/server/dive_server/views_configuration.py b/server/dive_server/views_configuration.py
index 7c56a7e1f..4e3bb40c1 100644
--- a/server/dive_server/views_configuration.py
+++ b/server/dive_server/views_configuration.py
@@ -17,7 +17,7 @@
 from girder_jobs.models.job import Job
 import requests
 
-from dive_server import crud, crud_rpc, worker_capabilities
+from dive_server import crud, crud_rpc, interactive_gate, worker_capabilities
 from dive_tasks import tasks
 from dive_utils import TRUTHY_META_VALUES, constants, models, stereo_models, types
 
@@ -92,6 +92,7 @@ def get_config(self):
             **capabilities,
             'jobsDisabled': jobs_disabled['disabled'],
             'jobsDisabledMessage': jobs_disabled['message'],
+            **interactive_gate.get_interactive_capability(),
         }
 
     @access.public
diff --git a/server/dive_server/views_interactive.py b/server/dive_server/views_interactive.py
new file mode 100644
index 000000000..f10f2bafe
--- /dev/null
+++ b/server/dive_server/views_interactive.py
@@ -0,0 +1,335 @@
+"""Server-side interactive segmentation and stereo for the web client.
+
+Each request names a dataset, camera and frame; girder resolves them to the
+assetstore paths the VIAME interactive service reads, checks that the GPU is
+not held by a job, and forwards the command to the broker on the GPU host.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any, Dict, List, Optional, Tuple
+
+from girder.api import access
+from girder.api.describe import Description, autoDescribeRoute
+from girder.api.rest import Resource
+from girder.constants import AccessType
+from girder.exceptions import RestException
+from girder.models.folder import Folder
+import requests
+
+from dive_server import crud, interactive_gate, interactive_media
+from dive_utils import constants, fromMeta, types
+
+STEREO_METHODS = ('ncc', 'dino', 'foundation')
+REQUEST_TIMEOUT = 320
+
+
+class InteractiveClient:
+    """Talks to the broker; one session per user and dataset."""
+
+    def __init__(self) -> None:
+        # Stereo backends are enabled once per session generation and method.
+        self._stereo_ready: Dict[str, Tuple[int, str, Optional[str]]] = {}
+        self._lock = threading.Lock()
+
+    @staticmethod
+    def session_key(user: types.GirderUserModel, dsFolder: types.GirderModel) -> str:
+        return f"{user['_id']}-{dsFolder['_id']}"
+
+    def request(
+        self,
+        key: str,
+        command: str,
+        params: Dict[str, Any],
+        session: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        url = interactive_gate.require_interactive_available()
+        try:
+            response = requests.post(
+                f'{url}/sessions/{key}/request',
+                json={'command': command, 'params': params, 'session': session or {}},
+                timeout=REQUEST_TIMEOUT,
+            )
+        except requests.RequestException as err:
+            raise RestException(f'The interactive service is unreachable: {err}', code=503)
+        body: Dict[str, Any] = {}
+        try:
+            body = response.json()
+        except ValueError:
+            pass
+        if response.status_code == 503:
+            raise RestException(body.get('error') or 'The interactive service is busy.', code=503)
+        if response.status_code >= 400:
+            raise RestException(body.get('error') or 'The interactive service failed.', code=502)
+        return body
+
+    def ensure_stereo(
+        self,
+        key: str,
+        method: str,
+        calibration: Optional[str],
+        generation_hint: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        """Enable the session's stereo backend for `method`, once per process."""
+        with self._lock:
+            ready = self._stereo_ready.get(key)
+        if ready and generation_hint is not None and ready == (generation_hint, method, calibration):
+            return {'_session': {'generation': generation_hint}}
+        params: Dict[str, Any] = {}
+        if calibration:
+            params['calibration_file'] = calibration
+        response = self.request(key, 'enable', params, session={'stereoMethod': method})
+        generation = int((response.get('_session') or {}).get('generation') or 0)
+        with self._lock:
+            self._stereo_ready[key] = (generation, method, calibration)
+        return response
+
+    def generation(self, key: str) -> Optional[int]:
+        with self._lock:
+            ready = self._stereo_ready.get(key)
+        return ready[0] if ready else None
+
+    def forget(self, key: str) -> None:
+        with self._lock:
+            self._stereo_ready.pop(key, None)
+
+
+client = InteractiveClient()
+
+
+def _method(value: Optional[str]) -> str:
+    return value if value in STEREO_METHODS else 'ncc'
+
+
+def _bad(response: Dict[str, Any], fallback: str) -> RestException:
+    return RestException(response.get('error') or fallback, code=400)
+
+
+class InteractiveResource(Resource):
+    def __init__(self, resourceName: str):
+        super().__init__()
+        self.resourceName = resourceName
+        self.route('GET', ('status',), self.status)
+        self.route('POST', ('segmentation', 'predict'), self.segmentation_predict)
+        self.route('POST', ('segmentation', 'keypoints'), self.segmentation_keypoints)
+        self.route('POST', ('segmentation', 'stereo_segment'), self.segmentation_stereo_segment)
+        self.route('POST', ('text_query',), self.text_query)
+        self.route('POST', ('stereo', 'set_frame'), self.stereo_set_frame)
+        self.route('POST', ('stereo', 'transfer_points'), self.stereo_transfer_points)
+        self.route('POST', ('stereo', 'transfer_line'), self.stereo_transfer_line)
+        self.route('POST', ('stereo', 'measure_line'), self.stereo_measure_line)
+        self.route('POST', ('stereo', 'aggregate_lengths'), self.stereo_aggregate_lengths)
+        self.route('DELETE', ('session',), self.end_session)
+
+    # ---- helpers -----------------------------------------------------------
+
+    def _dataset(self, dataset_id: str) -> Tuple[types.GirderModel, types.GirderUserModel]:
+        user = self.getCurrentUser()
+        folder = Folder().load(dataset_id, level=AccessType.READ, user=user)
+        if folder is None:
+            raise RestException('Dataset not found', code=404)
+        crud.verify_dataset(folder)
+        return folder, user
+
+    def _stereo_pair(
+        self, folder: types.GirderModel, user: types.GirderUserModel, frame: int
+    ) -> Tuple[List[str], Dict[str, Any], Dict[str, Any]]:
+        cameras = interactive_media.camera_order(folder)
+        if len(cameras) < 2:
+            raise RestException('Stereo tools need a two-camera dataset', code=400)
+        left = interactive_media.resolve_frame(folder, cameras[0], frame, user)
+        right = interactive_media.resolve_frame(folder, cameras[1], frame, user)
+        return cameras, left, right
+
+    def _stereo_session(
+        self, folder: types.GirderModel, user: types.GirderUserModel, method: str
+    ) -> str:
+        key = client.session_key(user, folder)
+        calibration = interactive_media.calibration_path(folder, user)
+        client.ensure_stereo(key, method, calibration, client.generation(key))
+        return key
+
+    def _stereo_request(
+        self, key: str, method: str, command: str, params: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        response = client.request(key, command, params, session={'stereoMethod': method})
+        # A restarted process lost its stereo backend: enable it again and retry once.
+        generation = int((response.get('_session') or {}).get('generation') or 0)
+        if not response.get('success', True) and client.generation(key) not in (None, generation):
+            client.forget(key)
+            raise RestException('The interactive session restarted; retry the request.', code=503)
+        return response
+
+    # ---- routes ------------------------------------------------------------
+
+    @access.user
+    @autoDescribeRoute(Description('Whether server-side interactive tools can be used now'))
+    def status(self):
+        return interactive_gate.get_interactive_capability()
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Segment one frame from point and box prompts')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def segmentation_predict(self, data):
+        folder, user = self._dataset(data['datasetId'])
+        media = interactive_media.resolve_frame(folder, data.get('camera'), int(data['frame']), user)
+        params: Dict[str, Any] = {
+            'image_path': media['path'],
+            'points': data.get('points') or [],
+            'point_labels': data.get('pointLabels') or [],
+        }
+        if media['frameTime'] is not None:
+            params['frame_time'] = media['frameTime']
+        for name, key in (('box', 'box'), ('line', 'line'), ('multimaskOutput', 'multimask_output')):
+            if data.get(name) is not None:
+                params[key] = data[name]
+        return client.request(client.session_key(user, folder), 'predict', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Head and tail points of a mask polygon')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def segmentation_keypoints(self, data):
+        folder, user = self._dataset(data['datasetId'])
+        params = {k: data[k] for k in ('polygon', 'polygons') if data.get(k) is not None}
+        return client.request(client.session_key(user, folder), 'polygon_keypoints', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Carry a source-camera mask to the other stereo camera')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def segmentation_stereo_segment(self, data):
+        folder, user = self._dataset(data['datasetId'])
+        frame = int(data['frame'])
+        method = _method(data.get('method'))
+        cameras, left, right = self._stereo_pair(folder, user, frame)
+        source = data.get('sourceCamera') or cameras[0]
+        if source not in cameras:
+            raise RestException(f'Unknown camera "{source}"', code=400)
+        source_is_left = source == cameras[0]
+        key = self._stereo_session(folder, user, method)
+        params: Dict[str, Any] = {
+            'points': data.get('points') or [],
+            'point_labels': data.get('pointLabels') or [],
+            'source_camera': 'left' if source_is_left else 'right',
+            'source_image_path': (left if source_is_left else right)['path'],
+            'other_image_path': (right if source_is_left else left)['path'],
+        }
+        calibration = interactive_media.calibration_path(folder, user)
+        if calibration:
+            params['calibration_file'] = calibration
+        if left['frameTime'] is not None:
+            params['frame_time'] = left['frameTime']
+        for name in ('polygon', 'polygons'):
+            if data.get(name) is not None:
+                params[name] = data[name]
+        return self._stereo_request(key, method, 'stereo_segment', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Find every instance of a text prompt in one frame')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def text_query(self, data):
+        folder, user = self._dataset(data['datasetId'])
+        media = interactive_media.resolve_frame(folder, data.get('camera'), int(data['frame']), user)
+        params: Dict[str, Any] = {'image_path': media['path'], 'text': str(data.get('text') or '')}
+        if media['frameTime'] is not None:
+            params['frame_time'] = media['frameTime']
+        return client.request(client.session_key(user, folder), 'text_query', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Prepare a stereo frame pair (dense disparity ahead of warps)')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def stereo_set_frame(self, data):
+        folder, user = self._dataset(data['datasetId'])
+        method = _method(data.get('method'))
+        _cameras, left, right = self._stereo_pair(folder, user, int(data['frame']))
+        key = self._stereo_session(folder, user, method)
+        params: Dict[str, Any] = {'left_image_path': left['path'], 'right_image_path': right['path']}
+        if left['frameTime'] is not None:
+            params['frame_time'] = left['frameTime']
+        return self._stereo_request(key, method, 'set_frame', params)
+
+    def _stereo_frame_params(self, data) -> Tuple[str, str, Dict[str, Any], List[str]]:
+        folder, user = self._dataset(data['datasetId'])
+        method = _method(data.get('method'))
+        cameras, left, right = self._stereo_pair(folder, user, int(data['frame']))
+        key = self._stereo_session(folder, user, method)
+        params: Dict[str, Any] = {'left_image_path': left['path'], 'right_image_path': right['path']}
+        if left['frameTime'] is not None:
+            params['frame_time'] = left['frameTime']
+        return key, method, params, cameras
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Warp points from one stereo camera to the other')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def stereo_transfer_points(self, data):
+        key, method, params, cameras = self._stereo_frame_params(data)
+        source = data.get('sourceCamera') or cameras[0]
+        if source not in cameras:
+            raise RestException(f'Unknown camera "{source}"', code=400)
+        params.update({
+            'points': data.get('points') or [],
+            'source_camera': 'left' if source == cameras[0] else 'right',
+            'strict': bool(data.get('strict', True)),
+        })
+        return self._stereo_request(key, method, 'transfer_points', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Warp a head/tail line to the other camera and measure it')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def stereo_transfer_line(self, data):
+        key, method, params, _cameras = self._stereo_frame_params(data)
+        params['line'] = data.get('line') or []
+        return self._stereo_request(key, method, 'transfer_line', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Measure a line drawn on both cameras')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def stereo_measure_line(self, data):
+        key, method, params, _cameras = self._stereo_frame_params(data)
+        params.update({'left_line': data.get('leftLine') or [], 'right_line': data.get('rightLine') or []})
+        return self._stereo_request(key, method, 'measure_line', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('Combine per-frame lengths into one track length')
+        .jsonParam('data', 'Request body', paramType='body', requireObject=True)
+    )
+    def stereo_aggregate_lengths(self, data):
+        folder, user = self._dataset(data['datasetId'])
+        params = {'lengths': data.get('lengths') or []}
+        if data.get('method'):
+            params['method'] = data['method']
+        return client.request(client.session_key(user, folder), 'aggregate_lengths', params)
+
+    @access.user
+    @autoDescribeRoute(
+        Description('End this user\'s interactive session for a dataset')
+        .param('datasetId', 'Dataset id', required=True)
+    )
+    def end_session(self, datasetId):
+        folder, user = self._dataset(datasetId)
+        key = client.session_key(user, folder)
+        client.forget(key)
+        url = interactive_gate.broker_url()
+        if url:
+            try:
+                requests.delete(f'{url}/sessions/{key}', timeout=5)
+            except requests.RequestException:
+                pass
+        return {'ok': True}
diff --git a/server/pyproject.toml b/server/pyproject.toml
index d04ceb8f0..e5201d7e2 100644
--- a/server/pyproject.toml
+++ b/server/pyproject.toml
@@ -102,7 +102,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["bucket_notifications", "dive_server", "dive_tasks", "dive_utils", "rabbitmq_user_queues"]
+packages = ["bucket_notifications", "dive_interactive", "dive_server", "dive_tasks", "dive_utils", "rabbitmq_user_queues"]
 include = ["**/*.mako"]
 
 [tool.hatch.build.targets.sdist]
diff --git a/server/tests/fixtures/fake_interactive_service.py b/server/tests/fixtures/fake_interactive_service.py
new file mode 100644
index 000000000..c2c3e5bea
--- /dev/null
+++ b/server/tests/fixtures/fake_interactive_service.py
@@ -0,0 +1,43 @@
+"""Stands in for viame.core.interactive_service in tests: JSON lines in, JSON lines out."""
+import json
+import sys
+import threading
+import time
+
+sys.stderr.write('Service started, waiting for requests...\n')
+sys.stderr.flush()
+lock = threading.Lock()
+
+
+def send(message):
+    with lock:
+        sys.stdout.write(json.dumps(message) + '\n')
+        sys.stdout.flush()
+
+
+def later(delay, message):
+    def run():
+        time.sleep(delay)
+        send(message)
+    threading.Thread(target=run, daemon=True).start()
+
+
+for raw in sys.stdin:
+    request = json.loads(raw)
+    command = request.get('command')
+    rid = request.get('id')
+    if command == 'shutdown':
+        send({'id': rid, 'success': True})
+        break
+    if command == 'sleep':
+        time.sleep(float(request.get('seconds', 0.5)))
+        send({'id': rid, 'success': True, 'slept': True})
+    elif command == 'set_frame':
+        send({'id': rid, 'success': True, 'disparity_ready': False})
+        later(0.1, {'id': rid, 'type': 'disparity_ready', 'success': True})
+    elif command == 'deferred':
+        later(0.2, {'id': rid, 'success': True, 'deferred': True})
+    elif command == 'crash':
+        sys.exit(3)
+    else:
+        send({'id': rid, 'success': True, 'echo': {k: v for k, v in request.items() if k not in ('id', 'command')}})
diff --git a/server/tests/test_interactive_gate.py b/server/tests/test_interactive_gate.py
new file mode 100644
index 000000000..1423f7668
--- /dev/null
+++ b/server/tests/test_interactive_gate.py
@@ -0,0 +1,57 @@
+from unittest.mock import MagicMock, patch
+
+from girder.exceptions import RestException
+import pytest
+
+from dive_server import interactive_gate
+
+
+@pytest.fixture(autouse=True)
+def reset_cache():
+    interactive_gate._health_cache.update(at=0.0, value=None)
+    yield
+    interactive_gate._health_cache.update(at=0.0, value=None)
+
+
+def test_disabled_without_a_broker(monkeypatch):
+    monkeypatch.delenv('DIVE_INTERACTIVE_URL', raising=False)
+    capability = interactive_gate.get_interactive_capability()
+    assert capability['interactiveEnabled'] is False
+    with pytest.raises(RestException) as err:
+        interactive_gate.require_interactive_available()
+    assert err.value.code == 503
+
+
+@patch('dive_server.interactive_gate.Job')
+@patch('dive_server.interactive_gate.requests.get')
+def test_paused_while_gpu_jobs_run(get, job_cls, monkeypatch):
+    monkeypatch.setenv('DIVE_INTERACTIVE_URL', 'http://interactive:8800/')
+    get.return_value = MagicMock(json=lambda: {'ok': True, 'stereoMethods': ['ncc']})
+    job_cls.return_value.findOne.return_value = {'_id': 'running'}
+    capability = interactive_gate.get_interactive_capability()
+    assert capability['interactiveEnabled'] is False
+    assert capability['interactiveMessage'] == interactive_gate.BUSY_MESSAGE
+    with pytest.raises(RestException) as err:
+        interactive_gate.require_interactive_available()
+    assert err.value.code == 503
+    query = job_cls.return_value.findOne.call_args[0][0]
+    assert set(query['type']['$in']) == {'pipelines', 'training', 'private'}
+
+
+@patch('dive_server.interactive_gate.Job')
+@patch('dive_server.interactive_gate.requests.get')
+def test_enabled_when_the_broker_answers_and_the_gpu_is_free(get, job_cls, monkeypatch):
+    monkeypatch.setenv('DIVE_INTERACTIVE_URL', 'http://interactive:8800')
+    get.return_value = MagicMock(json=lambda: {'ok': True, 'stereoMethods': ['ncc', 'foundation'], 'textQuery': True})
+    job_cls.return_value.findOne.return_value = None
+    capability = interactive_gate.get_interactive_capability()
+    assert capability == {
+        'interactiveEnabled': True,
+        'interactiveMessage': '',
+        'interactiveStereoMethods': ['ncc', 'foundation'],
+        'interactiveTextQuery': True,
+    }
+    assert interactive_gate.require_interactive_available() == 'http://interactive:8800'
+    # The health probe is cached, not repeated per call.
+    interactive_gate.get_interactive_capability()
+    assert get.call_count == 1
diff --git a/server/tests/test_interactive_manager.py b/server/tests/test_interactive_manager.py
new file mode 100644
index 000000000..7144b1482
--- /dev/null
+++ b/server/tests/test_interactive_manager.py
@@ -0,0 +1,78 @@
+import sys
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+from dive_interactive.manager import BusyError, SessionManager
+from dive_interactive.service import InteractiveProcess, ServiceError
+
+FAKE = Path(__file__).parent / 'fixtures' / 'fake_interactive_service.py'
+
+
+def spawn(_info):
+    return InteractiveProcess([sys.executable, '-u', str(FAKE)], ready_timeout=10)
+
+
+@pytest.fixture
+def manager():
+    manager = SessionManager(spawn, max_sessions=2, idle_seconds=60, queue_timeout=0.5, max_inflight=1)
+    yield manager
+    manager.stop_all()
+
+
+def test_requests_are_answered_by_id_even_when_deferred(manager):
+    first = manager.request('a', 'deferred', {})
+    second = manager.request('a', 'predict', {'points': [[1, 2]]})
+    assert first['deferred'] is True
+    assert second['echo'] == {'points': [[1, 2]]}
+    assert first['_session']['generation'] == second['_session']['generation']
+
+
+def test_events_reusing_a_request_id_are_kept_for_polling(manager):
+    manager.request('a', 'set_frame', {})
+    time.sleep(0.3)
+    events = manager.events('a', 0)
+    assert [e.get('type') for e in events['events']] == ['disparity_ready']
+    assert manager.events('a', events['next'])['events'] == []
+
+
+def test_one_inference_at_a_time_and_a_refusal_instead_of_a_long_wait(manager):
+    started = threading.Event()
+
+    def slow():
+        started.set()
+        manager.request('a', 'sleep', {'seconds': 1.5})
+
+    thread = threading.Thread(target=slow)
+    thread.start()
+    started.wait()
+    time.sleep(0.1)
+    with pytest.raises(BusyError):
+        manager.request('b', 'predict', {})
+    thread.join()
+    assert manager.request('b', 'predict', {})['success'] is True
+
+
+def test_session_cap_evicts_the_idlest_session(manager):
+    manager.request('a', 'predict', {})
+    manager.request('b', 'predict', {})
+    a_generation = manager.request('a', 'predict', {})['_session']['generation']
+    manager.request('c', 'predict', {})
+    keys = {s['key'] for s in manager.status()['sessions']}
+    assert keys == {'a', 'c'}
+    assert manager.request('a', 'predict', {})['_session']['generation'] == a_generation
+
+
+def test_idle_sessions_are_reaped_and_restarted_with_a_new_generation(manager):
+    generation = manager.request('a', 'predict', {})['_session']['generation']
+    assert manager.reap_idle(now=time.monotonic() + 120) == 1
+    assert manager.status()['sessions'] == []
+    assert manager.request('a', 'predict', {})['_session']['generation'] > generation
+
+
+def test_a_crashed_session_is_reported_and_replaced(manager):
+    with pytest.raises(ServiceError):
+        manager.request('a', 'crash', {})
+    assert manager.request('a', 'predict', {})['success'] is True