feat(audit): a run manifest should answer questions about its own run

replicant/core/models.py

@@ -222,6 +222,19 @@ class RunManifest(BaseModel):
     # ones that raise, so a short manifest and a failed one have to be
     # distinguishable: without this a run that died after two events looks
     # exactly like a run that was only ever meant to emit two.
+    # Which vendor dialect rendered these events. ScenarioManifest has always
+    # recorded it; RunManifest did not, so a manifest could not answer "which
+    # profile produced this?" without re-reading the settings that made it.
+    vendor: str = ""
+    #: The --duration the operator asked for, verbatim, or None for the default.
+    duration: str | None = None
+    #: The events-per-second ceiling actually in force for this run.
+    rate: int | None = None
+    #: What the socket really did: sends, bytes, errors, oversize. None when the
+    #: run had no collector. `event_count` counts events rendered; this counts
+    #: datagrams handed to the kernel, and the two differing is the interesting
+    #: case rather than an inconsistency.
+    send_stats: dict[str, int] | None = None
     status: RunStatus = "done"
     #: Bounded description of the failure, or None. Type and message only, never
     #: a traceback: this is an operator record, not a debugger.

replicant/core/orchestrator.py

@@ -231,6 +231,9 @@ def __init__(
         self.engine = engine or ScenarioEngine()
         self.profile = profile or build_profile(self.settings)
         self._stop = threading.Event()
+        #: What the last run's socket actually did, or None when it had no
+        #: collector. Set by _emit on every exit path and read into the manifest.
+        self.last_send_stats: dict[str, int] | None = None
 
     # -- vendor identity -------------------------------------------------------
 
@@ -357,6 +360,7 @@ def run(
         file_path = request.to_file
 
         self.reset()
+        self.last_send_stats = None
         plan = self.build_plan(request)
 
         target, transport = self._describe_target(request, send)
@@ -412,6 +416,10 @@ def run(
             warmup_note=warmup,
             pace=pace,
             speed=request.speed,
+            vendor=self.settings.vendor,
+            duration=request.duration,
+            rate=eps_cap,
+            send_stats=self.last_send_stats,
             status=_run_status(failure, stopped),
             error=describe_error(failure) if failure is not None else None,
         )
@@ -641,6 +649,17 @@ def _emit(
             if sink is not None:
                 sink.close()
             if emitter is not None:
+                # Captured before close, and on every exit path, so a run that
+                # failed part-way still records what its socket actually did.
+                # `event_count` counts events rendered; this counts datagrams the
+                # kernel accepted, and the two differing is the interesting case.
+                # getattr, not emitter.stats: the emitter is an injection point
+                # and several test doubles implement only send/close. Requiring a
+                # stats attribute would make this record cost every future fake a
+                # field it does not otherwise need.
+                stats = getattr(emitter, "stats", None)
+                if stats is not None:
+                    self.last_send_stats = stats.as_dict()
                 emitter.close()
         return count, stopped
 

tests/test_run_stream_fanout.py

@@ -198,3 +198,88 @@ def test_the_replay_is_bounded(self) -> None:
             handle.publish({"type": "line", "data": str(i)})
 
         assert len(handle.history) == MAX_HISTORY_ITEMS
+
+
+class TestManifestCompleteness:
+    """The manifest should answer questions about its own run without help.
+
+    Review finding #1: ScenarioManifest recorded vendor and duration, RunManifest
+    did not, and neither recorded the rate in force or what the socket actually
+    did. A manifest that cannot say which profile rendered it, or how many
+    datagrams left, is a weaker audit record than safety rule 5 implies.
+    """
+
+    def test_a_run_records_its_vendor_duration_and_rate(self, tmp_path: Path) -> None:
+        orch = Orchestrator(CATALOG, Settings(manifest_dir=str(tmp_path)))
+        result = orch.run(
+            RunRequest(
+                technique_id="REP-001",
+                intensity="low",
+                duration="90s",
+                seed=1337,
+                no_send=True,
+            )
+        )
+
+        assert result.manifest.vendor == "fortigate"
+        assert result.manifest.duration == "90s"
+        assert result.manifest.rate is not None and result.manifest.rate > 0
+
+    def test_a_sending_run_records_what_the_socket_did(self, tmp_path: Path) -> None:
+        import socket as _socket
+
+        listener = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
+        listener.bind(("127.0.0.1", 0))
+        port = int(listener.getsockname()[1])
+        try:
+            orch = Orchestrator(CATALOG, Settings(manifest_dir=str(tmp_path)))
+            result = orch.run(
+                RunRequest(
+                    technique_id="REP-001",
+                    intensity="low",
+                    duration="60s",
+                    no_send=False,
+                    pace="burst",
+                    collector=CollectorProfile(
+                        name="t", host="127.0.0.1", port=port, transport="udp"
+                    ),
+                )
+            )
+        finally:
+            listener.close()
+
+        stats = result.manifest.send_stats
+        assert stats is not None
+        assert stats["sends"] == result.event_count
+        assert stats["errors"] == 0
+
+    def test_a_run_with_no_collector_records_no_send_stats(self, tmp_path: Path) -> None:
+        """None is the honest answer, not a row of zeroes that reads like a send."""
+        orch = Orchestrator(CATALOG, Settings(manifest_dir=str(tmp_path)))
+        result = orch.run(RunRequest(technique_id="REP-001", intensity="low", no_send=True))
+
+        assert result.manifest.send_stats is None
+
+    def test_an_older_manifest_without_the_new_fields_still_loads(self) -> None:
+        """Every new field is defaulted, as pace and speed were before them."""
+        from replicant.core.models import RunManifest
+
+        manifest = RunManifest(
+            replicant_version="0.1.0",
+            technique_id="REP-001",
+            technique_name="x",
+            ndr_uc="UC-001",
+            intensity="low",
+            seed=1,
+            params={},
+            entities={},
+            target="none",
+            transport="none",
+            event_count=1,
+            started_at="t",
+            ended_at="t",
+            anchor_epoch=0,
+        )
+
+        assert manifest.vendor == ""
+        assert manifest.send_stats is None