Probe loop

A probe is the framework pattern for trying several candidates (often several cluster counts K) without baking “who wins” into the platform. Framework owns the loop, nested invoke, parallelism budget, and I/O. Domain owns prepare, decide, and how the winner is published.

for each candidate:
    prepare(candidate)     → opaque bindings (e.g. fitted centroids)
    evaluate(bindings)     → metrics from a nested downstream entry
domain decide(metrics)     → one winner
publish @ winner           → streaming write / assign

There is no winner field on the DTO: selection always happens in domain code reading ProbeRow metrics (and domain.probe.* params that framework never interprets).

Contracts

contracts/probe.py:

• ProbeBindings — candidate_id plus an opaque payload map;
• ProbeRow — candidate_id plus metrics for decide.

Production parent: processing/probe/parent_run.run_probe_parent. A thinner loop.run_probe_loop exists for callback-style tests.

Streaming parent (no full matrix)

1. Open a storage session and build a ProbeStreamRef (session + upstream runs + optional row cap) — vectors stay as a re-openable stream, not a loaded matrix.
2. Prepare candidates (can share that stream ref).
3. Fan out evaluate jobs (ProbeEvaluateJob) across workers via probe/parallel.py → job_map.map_jobs.
4. Domain decide must pick a candidate that appeared in the probe rows.
5. Domain publish returns row maps and/or an AssignPublishRequest; the parent executes streaming assign / writes.

Interleaved prepare→evaluate (phase C) uses probe/interleave.py → job_map.pipeline_jobs.

Probe subset (compute.probe.subset.max_input_rows) applies to prepare/evaluate sampling only. Final publish uses the full upstream unless the domain says otherwise. Multi-shard subset (shard_count > 1) is rejected on the streaming parent.

Parallelism and nesting

Probe does not invent a second parallel backend. HOW is always runtime.parallelism.*. Fan-out width is min(compute.probe.max_parallel, workers).

Nested evaluate steps increment depth (default max 2). Beyond the nesting budget, or when allow_nested_parallel is false, the child is forced to inprocess / workers=1. When nesting is allowed, child workers come from compute.probe.child_workers when set; otherwise parent.workers // fanout — so a probe of eight candidates does not spawn eight full Ray clusters.

With sticky Ray fit, prepare is exclusive: all candidates first, then evaluate. processing/probe/barrier.py releases fit actors before evaluate (KubeRay stays inflated) so nested evaluate can use Ray. Idle restore runs only when the outermost probe parent closes.

Evaluate invokes the same downstream entry the DAG would use, typically with persist_outputs=false and bindings injected into params. Jobs stay picklable (no open sessions on the wire).

Row assembly helper

Nested evaluate often needs “load inputs → domain build rows/metrics → maybe write”. processing/row_assembly.run_row_assembly owns that glue: compile, open session, load by AssemblyInputSpec, apply optional subset, call domain build, and write only when persist is enabled. Domain code stays free of storage and metrics emit. The helper is not probe-owned: probe and other nested invokes call it.

Large inputs take the spill-parallel path: shard by parquet row group, map, then SUM over shuffle buckets (row_assembly/spill.py → reduce_merge workers).

Nested evaluate caches and bindings

• row_feed_cache (compute.probe.cache_eval_inputs, default on): process-local reuse of identical dataset loads across candidates; cleared when the probe parent ends.
• Assignment spill: probe evaluate jobs do not pickle assignment_map. The parent writes unique vote-key → label rows to parquet (processing/probe/assignment_spill.py) and passes assignment_run_id ({parent}__probe_{candidate}). Nested evaluate mapping reads that run via mapping.assignment_upstream_runs.
• Bindings trim: after evaluate, heavy payload keys (default assignment_map) are dropped from ProbeBindings so RAM does not grow with probe width; centroids needed for publish stay.

Live phase status (observability)

processing/live_status tracks probe phase codes (inflating, prepare_load, prepare_fit, prepare_assign, evaluate, decide, publish), candidate id, and fit iteration / inertia. Framework emits optional Prometheus gauges (probe_phase, probe_candidate_k, probe_fit_iter, …) with an immediate flush on each note_* so Grafana does not wait for the heartbeat. Domain does not instrument.

Phase 7 (inflating) is sticky Ray waiting for KubeRay workers before vector load. After inflate, the progress clock restarts so warmup time is not counted as throughput.

Progress ticks: one unit per prepare or evaluate candidate, plus publish; progress_total is set at begin.

Typical knobs (compute.probe.*)

Param	Role
candidate_mode	How to build the candidate list: list, linspace, logspace, geom
candidates	Explicit list (when mode=list)
k_min / k_max / k_n / k_ratio	Grid / geometric generators
max_parallel	Evaluate fan-out width
downstream_*	Nested evaluate manifest / entry
allow_nested_parallel	Whether children may keep workers
child_workers	Nested evaluate actor quota (else workers // fanout)
subset.*	Sample size for probe prepare
persist_outputs	Whether nested evaluate may write
cache_eval_inputs	Reuse nested evaluate dataset loads (default on)

domain.probe.* is decide policy only. Cluster sticky / warm-start knobs live under compute.cluster.* — see Streaming cluster.

Read next

• Streaming cluster fit and assign
• Parallelism
• Offline evaluation
• framework/contracts