Consolidate runtime execution, producer and projection ownership

benchmark/control_service.py

@@ -64,7 +64,7 @@ def finalize_measured_run(
             request.job_id
         )
         observation_path = ZMQAuxiliaryExecutionParams.from_transport(
-            completed.submission.config_params
+            completed.request.config_params
         ).runtime_observation_export_path
         if observation_path is None:
             raise ValueError("Completed job has no runtime observation export.")
@@ -91,12 +91,12 @@ def finalize_measured_run(
             seconds=execution_seconds,
         )
         return retain_measured_openhcs_completion(
-            submission=completed.submission,
+            request=completed.request,
             execution_id=record.execution_id,
             results_summary=record.results_summary,
             endpoint_provenance=measured_endpoint_provenance(completed.endpoint),
             phase_timing=phase_timing,
-            compile_artifact_id=completed.submission.compile_artifact_id,
+            compile_artifact_id=completed.request.compile_control.compile_artifact_id,
         ).receipt
 
     def discover_cases(

benchmark/openhcs_measured_run.py

@@ -16,7 +16,7 @@
 from typing import Any
 
 from zmqruntime import DataControlPortPairAuthority
-from zmqruntime.messages import ExecutionStatusSnapshot, PongResponse
+from zmqruntime.messages import PongResponse
 
 from benchmark.contracts.measured_run_receipt import (
     MEASURED_PIPELINE_RUN_RECEIPT_SCHEMA_VERSION,
@@ -29,15 +29,14 @@
 )
 from benchmark.contracts.tool_adapter import ToolExecutionError
 from benchmark.file_digest import sha256_file
-from openhcs.core.config import GlobalPipelineConfig
-from openhcs.core.config_document import ConfigDocumentAuthority
 from openhcs.core.execution_state import ExecutionOutputPlateSummary
 from openhcs.core.runtime_execution_validation import (
     RuntimeArtifactExecutionObservation,
 )
 from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION
 from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG
 from openhcs.runtime.zmq_execution_client import (
+    ZMQExecutionRequestBuilder,
     OpenHCSExecutionSubmission,
     ZMQExecutionClient,
     ZMQPipelineRunPhase,
@@ -230,25 +229,6 @@ def progress_description(self) -> str:
         return f"{self.last_progress_phase}/{self.last_progress_status}"
 
 
-def _completed_server_job_seconds(
-    client: ZMQExecutionClient, execution_id: str
-) -> float:
-    """Read an ordinary completed job's authoritative server time bounds."""
-
-    snapshot = ExecutionStatusSnapshot.from_dict(client.poll_status(execution_id))
-    record = snapshot.execution
-    if record is None:
-        raise ToolExecutionError(
-            f"Completed OpenHCS job {execution_id!r} has no valid server time bounds."
-        )
-    try:
-        return completed_server_execution_seconds(
-            record, expected_execution_id=execution_id
-        )
-    except ValueError as exc:
-        raise ToolExecutionError(str(exc)) from exc
-
-
 def _require_measured_submission(
     submission: OpenHCSExecutionSubmission,
     expected_axis_count: int | None,
@@ -342,32 +322,39 @@ def execute_measured_openhcs_pipeline_on_client(
         )
     except RuntimeError as exc:
         raise ToolExecutionError(str(exc)) from exc
-    phase_timing.record(
-        BenchmarkPhase.SERVER_COMPILATION_JOB,
-        seconds=_completed_server_job_seconds(client, run.compile_artifact_id),
-    )
-    phase_timing.record(
-        BenchmarkPhase.SERVER_PIPELINE_JOB,
-        seconds=_completed_server_job_seconds(client, run.execution_id),
-    )
+    try:
+        phase_timing.record(
+            BenchmarkPhase.SERVER_COMPILATION_JOB,
+            seconds=completed_server_execution_seconds(
+                run.compile_record, expected_execution_id=run.compile_artifact_id
+            ),
+        )
+        phase_timing.record(
+            BenchmarkPhase.SERVER_PIPELINE_JOB,
+            seconds=completed_server_execution_seconds(
+                run.execution_record, expected_execution_id=run.execution_id
+            ),
+        )
+    except ValueError as exc:
+        raise ToolExecutionError(str(exc)) from exc
     timing_observer.record_phase_timings(phase_timing)
     return (
         retain_measured_openhcs_completion(
-            submission=submission,
+            request=run.request,
             execution_id=run.execution_id,
             results_summary=run.results_summary,
             endpoint_provenance=endpoint_provenance,
             phase_timing=phase_timing,
             compile_artifact_id=run.compile_artifact_id,
             expected_axis_count=expected_axis_count,
         ),
-        submission.pipeline_code(),
+        run.request.pipeline_code,
     )
 
 
 def retain_measured_openhcs_completion(
     *,
-    submission: OpenHCSExecutionSubmission,
+    request: ZMQExecutionRequestBuilder,
     execution_id: str,
     results_summary: Mapping[str, Any],
     endpoint_provenance: MeasuredEndpointProvenance,
@@ -380,7 +367,7 @@ def retain_measured_openhcs_completion(
     if expected_axis_count is not None and expected_axis_count < 1:
         raise ValueError("Expected axis count must be positive when declared.")
     auxiliary_params = ZMQAuxiliaryExecutionParams.from_transport(
-        submission.config_params
+        request.config_params
     )
     observation_export_path = auxiliary_params.runtime_observation_export_path
     if observation_export_path is None:
@@ -443,11 +430,8 @@ def retain_measured_openhcs_completion(
         )
     artifact_root = observation_export_path.parent
     output_roots = tuple(Path(root) for root in observation_export.output_roots)
-    pipeline_source = submission.pipeline_code()
-    global_config_source = ConfigDocumentAuthority.render(
-        submission.global_pipeline_config,
-        expected_config_type=GlobalPipelineConfig,
-    )
+    pipeline_source = request.pipeline_code
+    global_config_source = request.global_config_code
     results_summary_path = MeasuredPipelineRunArtifact.RESULTS_SUMMARY.path_in(
         artifact_root
     )
@@ -465,9 +449,9 @@ def retain_measured_openhcs_completion(
         schema_version=MEASURED_PIPELINE_RUN_RECEIPT_SCHEMA_VERSION,
         run_id=phase_timing.run_id,
         pipeline_name=phase_timing.pipeline_name,
-        plate_id=submission.plate_id,
-        execution_plate_id=submission.execution_plate_id,
-        selected_pipeline_path=submission.selected_pipeline_path,
+        plate_id=request.identity.plate_id,
+        execution_plate_id=request.identity.execution_plate_id,
+        selected_pipeline_path=request.identity.selected_pipeline_path,
         execution_id=execution_id,
         pipeline_source_sha256=hashlib.sha256(
             pipeline_source.encode("utf-8")

benchmark/reports/cppipe_figures.py

@@ -1192,7 +1192,7 @@ def _plot_average_speedup_points(
             alpha=0.86,
         )
         axis.annotate(
-            "4x target",
+            f"{SPEEDUP_TARGET:g}x target",
             xy=(0.995, SPEEDUP_TARGET),
             xycoords=("axes fraction", "data"),
             xytext=(-2, 3),
@@ -1423,7 +1423,7 @@ def plot_cdf(self, *, log_x: bool) -> tuple[Path, ...]:
                 alpha=0.86,
             )
             axis.annotate(
-                "4x target",
+                f"{SPEEDUP_TARGET:g}x target",
                 xy=(SPEEDUP_TARGET, 99.0),
                 xycoords=("data", "data"),
                 xytext=(3, -2),
@@ -1626,7 +1626,10 @@ def _draw_reference_lines(axis, *, metric: FigureMetricSpec, log_y: bool) -> Non
         linestyle="--",
         alpha=0.86,
     )
-    label = "4x target" if metric.key == SPEEDUP_METRIC_KEY else "target"
+    label = (
+        f"{metric.target_line:g}x target"
+        if metric.key == SPEEDUP_METRIC_KEY else "target"
+    )
     axis.annotate(
         label,
         xy=(0.995, metric.target_line),
@@ -1689,7 +1692,7 @@ def _write_benchmark_figure_index(path: Path, outputs: Sequence[Path]) -> None:
         "- `cppipe_accuracy_zoom.*`: broken-axis parity view for tiny numeric drift near 100%.",
         "- `cppipe_raw_seconds.*`: single-thread execution runtime in seconds.",
         "- `cppipe_raw_seconds_log.*`: runtime on a log scale for mixed short and long pipelines.",
-        "- `cppipe_speedup.*`: execution speedup versus native CellProfiler with the 4x target line.",
+        f"- `cppipe_speedup.*`: execution speedup versus native CellProfiler with the {SPEEDUP_TARGET:g}x target line.",
         "- `cppipe_speedup_log.*`: speedup on a log scale for wide dynamic range.",
         "- `cppipe_speedup_summary_statistics.*`: min, max, median, and mean speedup tables.",
         "- `cppipe_speedup_cumulative_distribution.*`: percent of datasets at or above each speedup threshold.",

benchmark/results/perf_canonical_output_policy_paired_20261003/README.md

@@ -0,0 +1,10 @@
+The output-policy correctness fix is merged in [PR #531](https://github.com/OpenHCSDev/openhcs/pull/531), closing #530. These fresh observations use performance source `ee8f97552b349499655bb3f961860b34dc784071`, synced with main `ba7b26b82`. Existing environments and the public ordinary/native drivers were reused; both repetitions are retained. Full strict saved-output comparisons pass for all eight ordinary/native pairings, including exact 3D images.
+
+| Pipeline | Compile | OH execution | OH total | Native CP invocation | Execution speedup | Total speedup |
+| --- | ---: | ---: | ---: | ---: | ---: | ---: |
+| 3D Monolayer | 1.176 s | 8.994 s | 10.974 s | 14.317 s | 1.592× | 1.305× |
+| Speckles | 0.498 s | 1.261 s | 2.091 s | 2.081 s | 1.650× | 0.995× |
+
+Means use both measured observations. Ordinary 3D execution spans 8.562–9.425 s; the earlier qualified mean was 9.127 s. These separate two-repeat observations do not establish a causal speedup. The remaining 2× execution gap is 1.835 s, so the larger generic source/context/publication work remains necessary. Seven corrected CP groups preserve their input instead of producing anonymous image replacements; this is an architectural dependency, not completion of the performance target.
+
+Server startup, mandatory library/kernel readiness, shutdown and native imports/JVM/loading/full warmup precede or follow pipeline clocks. Ordinary total includes compilation and default OUTCOMES/RSS completion. This is a two-case preview; full30 and scaling figures remain unfinished. Figures use the existing `benchmark.reports.cppipe_figures` owner with a 2× reference. `timing_summary.json`, raw CSVs, native reports, strict science and `qualification.json` retain determining evidence.