Serve OP1's adapter draws from DirectRNGStreams by default

docs/design/open_population_evidence/op1_run_size.py

@@ -1,10 +1,24 @@
 """Measure full-horizon OP1 resource use on invented rosters (design §8).
 
 Run ``python docs/design/open_population_evidence/op1_run_size.py --sizes 25
-100 --output docs/design/open_population_evidence/op1_run_size.json`` from the
-repository root. Each size and emigration stream implementation runs in a
-fresh worker, sequentially. The sealed loop always uses its own registry.
-Only the exit adapter's stream-class binding changes for the comparison.
+100 400 --output
+docs/design/open_population_evidence/op1_run_size_injection.json`` from the
+repository root. (``op1_run_size.json`` keeps the two-mode run of OP1 as
+merged in #566, which the report's section 2 cites.) Each size and stream
+mode runs in a fresh worker, sequentially. The sealed loop's own module generators always come from its
+registry (``engine/loop.py:217``, ``:297–350``). Three stream modes compare
+which class serves the other draws:
+
+- ``registry``: the adapters' person and tagged draws and the emigration
+  draws all come from ``ProjectionRNGRegistry`` (``rng_streams="registry"``,
+  with the exit adapter's binding patched to the registry);
+- ``direct_emigration``: OP1's behavior before the injection, the adapters on
+  the registry and emigration on ``DirectRNGStreams``;
+- ``direct``: the default, every person and tagged draw on
+  ``DirectRNGStreams`` (``rng_streams="direct"``).
+
+Each record names the stream classes the adapters and the exit adapter were
+observed to receive, and the SHA-256 of ``open_population/streams.py``.
 
 Mortality draws person streams (``engine/steps.py:125–131``); fertility draws
 tagged person streams (``baselines/fertility.py:114–126``). Aging draws none
@@ -19,7 +33,9 @@
 also report complete setup, including leaves and generator construction.
 They wrap module, child, tagged and tagged-child methods; ``person_generator``
 delegates to the child method (``engine/rng.py:158–170``) and is not added.
-The direct wrapper measures ``_sequence`` construction. Complete emigration
+The direct wrappers measure ``_sequence`` construction, for emigration and,
+in ``direct`` mode, for the adapters; four disjoint adapter wrappers report
+complete direct setup as the sealed ones do. Complete emigration
 tagged-generator setup is also reported.
 The sealed-loop and emigration profiler groups do not overlap. Profile time
 excludes imports and fixture construction. Peak RSS includes imports and the
@@ -44,6 +60,26 @@
 START_YEAR = 2025
 REGISTERED_HORIZON = 2100
 EMIGRATION_PROBABILITY = 0.02
+ROOT = Path(__file__).resolve().parents[3]
+# Each mode: (harness rng_streams, emigration stream class).
+STREAM_MODES = {
+    "registry": ("registry", "registry"),
+    "direct_emigration": ("registry", "direct"),
+    "direct": ("direct", "direct"),
+}
+HASHED_SOURCES = (
+    "src/populace_dynamics/open_population/streams.py",
+    "src/populace_dynamics/open_population/harness.py",
+    "src/populace_dynamics/open_population/exits.py",
+)
+
+
+def _source_digests() -> dict:
+    """Hash the stream, harness and exit sources a run used (design §8)."""
+    return {
+        path: hashlib.sha256((ROOT / path).read_bytes()).hexdigest()
+        for path in HASHED_SOURCES
+    }
 
 
 def _positive_int(value: str) -> int:
@@ -67,6 +103,11 @@ def _profile_summary(profile: cProfile.Profile) -> dict:
         "emigration_registry_sequences",
         "emigration_direct_sequences",
         "emigration_generator_setup",
+        "adapter_direct_sequences",
+        "adapter_direct_module_generator_setup",
+        "adapter_direct_person_generator_setup",
+        "adapter_direct_tagged_generator_setup",
+        "adapter_direct_tagged_child_generator_setup",
     ):
         matches = [v for k, v in stats.stats.items() if k[2] == name]
         seconds = sum(v[3] for v in matches)
@@ -89,6 +130,20 @@ def _profile_summary(profile: cProfile.Profile) -> dict:
         "cumulative_seconds": setup_seconds,
         "share_of_profile_total": setup_seconds / stats.total_tt,
     }
+    direct_names = (
+        "adapter_direct_module_generator_setup",
+        "adapter_direct_person_generator_setup",
+        "adapter_direct_tagged_generator_setup",
+        "adapter_direct_tagged_child_generator_setup",
+    )
+    direct_seconds = sum(
+        out[name]["cumulative_seconds"] for name in direct_names
+    )
+    out["adapter_direct_generator_setup_total"] = {
+        "calls": sum(out[name]["calls"] for name in direct_names),
+        "cumulative_seconds": direct_seconds,
+        "share_of_profile_total": direct_seconds / stats.total_tt,
+    }
     out["top_self_costs"] = [
         {
             "function": f"{Path(filename).name}:{line}:{name}",
@@ -105,6 +160,7 @@ def _profile_summary(profile: cProfile.Profile) -> dict:
 
 def _measure(size: int, stream_mode: str) -> dict:
     """Profile one invented 75-year production run (design §§2.9, 8)."""
+    rng_streams, emigration_stream = STREAM_MODES[stream_mode]
     started = time.perf_counter()
     load_start = os.getloadavg()
     import resource
@@ -125,7 +181,7 @@ def _measure(size: int, stream_mode: str) -> dict:
         advance_age,
         apply_mortality,
     )
-    from populace_dynamics.open_population import exits, harness
+    from populace_dynamics.open_population import exits, harness, streams
     from populace_dynamics.open_population.exits import no_marital_transitions
     from populace_dynamics.open_population.harness import ProductionRequest
     from populace_dynamics.open_population.streams import DirectRNGStreams
@@ -184,9 +240,20 @@ def no_composition(frame, context, marital, rng):
         },
     )
     draw_counts = {"mortality": 0, "emigration": 0}
+    observed = {"adapters": set(), "emigration": set()}
+
+    def stream_class(cls):
+        """Name the project class behind a profiled subclass."""
+        base = next(
+            item
+            for item in cls.__mro__
+            if item.__module__.startswith("populace_dynamics")
+        )
+        return f"{base.__module__}.{base.__qualname__}"
 
     def mortality(frame, context, rng):
         draw_counts["mortality"] += len(frame)
+        observed["adapters"].add(stream_class(type(context.rng_registry)))
         return apply_mortality(frame, context, rng, model=model)
 
     def emigration(frame, context):
@@ -235,6 +302,7 @@ def emigration_registry_sequences(self, period, module):
         seed_sequence = emigration_registry_sequences
 
         def emigration_generator_setup(self, *args):
+            observed["emigration"].add(stream_class(type(self)))
             return tagged_child_generator(self, *args)
 
         tagged_child_generator = emigration_generator_setup
@@ -246,10 +314,39 @@ def emigration_direct_sequences(self, period, module, *tail):
         _sequence = emigration_direct_sequences
 
         def emigration_generator_setup(self, *args):
+            observed["emigration"].add(stream_class(type(self)))
             return super().tagged_child_generator(*args)
 
         tagged_child_generator = emigration_generator_setup
 
+    class ProfiledAdapterDirect(DirectRNGStreams):
+        """The adapters' direct streams, with disjoint setup wrappers."""
+
+        def adapter_direct_sequences(self, period, module, *tail):
+            return super()._sequence(period, module, *tail)
+
+        _sequence = adapter_direct_sequences
+
+        def adapter_direct_module_generator_setup(self, *args):
+            return super().generator(*args)
+
+        generator = adapter_direct_module_generator_setup
+
+        def adapter_direct_person_generator_setup(self, *args):
+            return super().child_generator(*args)
+
+        child_generator = adapter_direct_person_generator_setup
+
+        def adapter_direct_tagged_generator_setup(self, *args):
+            return super().tagged_generator(*args)
+
+        tagged_generator = adapter_direct_tagged_generator_setup
+
+        def adapter_direct_tagged_child_generator_setup(self, *args):
+            return super().tagged_child_generator(*args)
+
+        tagged_child_generator = adapter_direct_tagged_child_generator_setup
+
     def capture_snapshots(transitions, **kwargs):
         snapshots = [
             frame
@@ -271,7 +368,8 @@ def capture_snapshots(transitions, **kwargs):
     profile = cProfile.Profile()
     projection_load_start = os.getloadavg()
     projection_started = time.perf_counter()
-    cls = ProfiledDirect if stream_mode == "direct" else ProfiledRegistry
+    projection_cpu_started = time.process_time()
+    cls = ProfiledDirect if emigration_stream == "direct" else ProfiledRegistry
     with (
         patch.object(
             ProjectionRNGRegistry, "seed_sequence", sealed_registry_sequences
@@ -295,6 +393,7 @@ def capture_snapshots(transitions, **kwargs):
             sealed_tagged_child_generator_setup,
         ),
         patch.object(exits, "DirectRNGStreams", cls),
+        patch.object(harness, "DirectRNGStreams", ProfiledAdapterDirect),
         patch.object(harness, "reconcile_history", capture_snapshots),
     ):
         profile.enable()
@@ -305,8 +404,10 @@ def capture_snapshots(transitions, **kwargs):
             schedule_builder=build_schedule,
             reserved_real_ids=initial["person_id"],
             emigration_hazard=emigration,
+            rng_streams=rng_streams,
         )
         profile.disable()
+    projection_cpu_seconds = time.process_time() - projection_cpu_started
     projection_seconds = time.perf_counter() - projection_started
     projection_load_end = os.getloadavg()
     events = Counter(event.kind.value for event in result.events)
@@ -323,9 +424,32 @@ def capture_snapshots(transitions, **kwargs):
     if system not in {"Darwin", "Linux"}:
         raise RuntimeError("peak RSS units are defined for macOS and Linux")
     peak_rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+    expected_adapters = (
+        "populace_dynamics.open_population.streams.DirectRNGStreams"
+        if rng_streams == "direct"
+        else "populace_dynamics.engine.rng.ProjectionRNGRegistry"
+    )
+    expected_emigration = (
+        "populace_dynamics.open_population.streams.DirectRNGStreams"
+        if emigration_stream == "direct"
+        else "populace_dynamics.engine.rng.ProjectionRNGRegistry"
+    )
+    if observed["adapters"] != {expected_adapters} or observed[
+        "emigration"
+    ] != {expected_emigration}:
+        raise AssertionError(f"{stream_mode} ran other stream classes")
     summary = {
         "status": "completed",
-        "emigration_stream": stream_mode,
+        "stream_mode": stream_mode,
+        "rng_streams": result.rng_streams,
+        "adapter_stream_class": expected_adapters,
+        "emigration_stream": emigration_stream,
+        "emigration_stream_class": expected_emigration,
+        "result_rng_stream_class": result.rng_stream_class,
+        "streams_py_sha256": hashlib.sha256(
+            Path(streams.__file__).read_bytes()
+        ).hexdigest(),
+        "source_sha256": _source_digests(),
         "emigration_probability": EMIGRATION_PROBABILITY,
         "initial_people": size,
         "final_people": len(slices[-1]),
@@ -350,6 +474,7 @@ def capture_snapshots(transitions, **kwargs):
         ),
         "projection_sha256": digest.hexdigest(),
         "projection_wall_time_seconds": projection_seconds,
+        "projection_process_cpu_seconds": projection_cpu_seconds,
         "seconds_per_mortality_person_year": projection_seconds
         / draw_counts["mortality"],
         "peak_memory_bytes": int(
@@ -437,13 +562,9 @@ def _microbenchmark() -> dict:
 def _comparisons(measurements: list[dict]) -> dict:
     """Fit a two-size engineering extrapolation, without a full run (§8)."""
     comparisons = {}
-    for mode in ("direct", "registry"):
+    for mode in STREAM_MODES:
         records = sorted(
-            (
-                item
-                for item in measurements
-                if item["emigration_stream"] == mode
-            ),
+            (item for item in measurements if item["stream_mode"] == mode),
             key=lambda item: item["initial_people"],
         )
         if len(records) < 2:
@@ -504,7 +625,7 @@ def main() -> None:
     )
     parser.add_argument(
         "--worker-stream",
-        choices=("direct", "registry"),
+        choices=tuple(STREAM_MODES),
         help=argparse.SUPPRESS,
     )
     arguments = parser.parse_args()
@@ -523,7 +644,14 @@ def main() -> None:
         "registered_horizon": REGISTERED_HORIZON,
         "projected_years": 75,
         "requested_initial_sizes": arguments.sizes,
-        "process_isolation": "one fresh worker per size and emigration stream implementation; workers run sequentially",
+        "process_isolation": "one fresh worker per size and stream mode; workers run sequentially",
+        "stream_modes": {
+            "registry": "adapters' person and tagged draws and emigration draws on ProjectionRNGRegistry",
+            "direct_emigration": "adapters on ProjectionRNGRegistry, emigration on DirectRNGStreams (OP1 before the injection)",
+            "direct": "every person and tagged draw on DirectRNGStreams (the default)",
+            "loop_module_generators": "the sealed loop's 600 positional module generators stay on its registry in every mode",
+        },
+        "source_sha256": _source_digests(),
         "peak_memory_method": "resource.getrusage(RUSAGE_SELF).ru_maxrss; bytes on macOS, KiB converted to bytes on Linux",
         "profile_method": "cProfile projection only; registry_sequences measure root/period/module seed_sequence cumulative cost, excluding child leaves and default_rng; sealed_generator_setup_total adds four disjoint module/child/tagged/tagged-child setup wrappers and includes all leaves/default_rng; emigration_direct_sequences measures direct construction; emigration_generator_setup includes complete tagged generator setup",
         "person_stream_adapters": [
@@ -541,7 +669,7 @@ def main() -> None:
         "full_size_measurement": "open OP1 item required before OP10a; intentionally not attempted in this build",
     }
     for size in arguments.sizes:
-        for mode in ("direct", "registry"):
+        for mode in STREAM_MODES:
             started = time.perf_counter()
             try:
                 process = subprocess.run(
@@ -585,7 +713,7 @@ def main() -> None:
                         key: measurement[key]
                         for key in (
                             "initial_people",
-                            "emigration_stream",
+                            "stream_mode",
                             "projection_wall_time_seconds",
                             "load_at_projection_start",
                         )
@@ -598,9 +726,7 @@ def main() -> None:
             item for item in measurements if item["initial_people"] == size
         ]
         if len({item["projection_sha256"] for item in records}) != 1:
-            raise AssertionError(
-                "registry and direct projection outputs differ"
-            )
+            raise AssertionError("the stream modes' projections differ")
     payload["stream_modes_have_identical_projection_hashes"] = True
     payload["rng_microbenchmark"] = _microbenchmark()
     _write(payload, arguments.output)