Let a seal-only population observer skip the executor's snapshot detachment

changelog.d/graph-observer-detach.added.md

@@ -0,0 +1 @@
+The graph executor accepts an opt-in flag that lets a seal-only population observer receive the live admitted population instead of a detached pickle snapshot per reached node; the default is unchanged and the flag enters no key, receipt or cache record.

docs/graph-acceptance.md

@@ -512,6 +512,20 @@ lock unchanged:
     cost before scaling. Extracted with the independently reviewed observer
     isolation repair on 2026-09-12.
 
+25. **A caller may decline the observer snapshot.** `run_graph(...,
+    _population_observer_detach=False)` passes the live admitted population
+    to the private observer instead of the detached snapshot amendment 24
+    describes. The flag is the caller's declaration that its observer will
+    neither retain nor mutate what it is given; in this mode the executor no
+    longer enforces the execution or persistence guarantee of amendment 24,
+    and a mutating observer can leave the content store holding bytes the
+    node key does not name, served to every later run sharing that store as
+    a cache hit. The default is unchanged (detached); the flag is passed at
+    run time only and enters no node key, receipt or cache record; an
+    observer's exception still refuses the run. Runtime-only, the interface
+    lock is unchanged. Adopted 2026-09-18 for the retention seal's verifier,
+    which reads the live population and seals its content (#950, #951).
+
 Adding a normative field with a default changes the canonical projection
 of every node that carries it, so node keys moved with amendments 11 and
 13's sibling field `entrants`; no released artifact pins a graph key yet.

packages/microcosm-graph/src/microcosm/graph/executor.py

@@ -2538,6 +2538,7 @@ def run_graph(
     resume: ResumePolicy = "auto",
     decisions: tuple[Decision, ...] = (),
     _population_observer: Callable[[str, Population], None] | None = None,
+    _population_observer_detach: bool = True,
     _verification_epoch: Mapping[str, object] | None = None,
 ) -> RunManifest:
     """Execute a compiled graph with content-addressed reuse and receipts.
@@ -2550,6 +2551,28 @@ def run_graph(
     kernel capability, enters no key or receipt, and an unreached node has no
     population to observe.
 
+    The detachment costs one full independent population and a temporary
+    serialized table buffer per reached node, which is the whole cost for an
+    observer that only reads. ``_population_observer_detach=False`` is that
+    observer's declaration that it will neither retain nor mutate what it is
+    given: no snapshot is allocated and the live admitted population is passed
+    instead. The declaration is the caller's, not the executor's.
+
+    The execution and persistence guarantees stated above are withdrawn in
+    this mode, and the second matters more than the first. The executor no longer enforces
+    that an observer cannot alter **execution**: a mutating observer corrupts
+    the run it is in. It no longer enforces that an observer cannot alter
+    **persistence** either, and that damage outlives the run -- a mutation
+    before the node is persisted can leave the store holding bytes that are
+    not the content the node key names, with the payload digest rewritten to
+    match, so every later run sharing that store serves them as a cache hit
+    under an unchanged node key and nothing afterwards can detect it. Pass
+    this keyword only for an observer whose whole body is a read.
+
+    The default is unchanged and still enforces both. The keyword enters no
+    key, no receipt and no cache record, and is meaningless without an
+    observer.
+
     The private verification-epoch record is a caller's own counts mapping --
     a country runtime that scopes source verification around the whole run
     hands in the record that scope yields. It is attached to the manifest
@@ -2584,6 +2607,7 @@ def run_graph(
                 written=written,
                 run_sources=run_sources,
                 _population_observer=_population_observer,
+                _population_observer_detach=_population_observer_detach,
                 _verification_epoch=_verification_epoch,
             )
         except BaseException as error:
@@ -2602,6 +2626,7 @@ def _execute_graph(
     written: set[str],
     run_sources: _RunSources,
     _population_observer: Callable[[str, Population], None] | None,
+    _population_observer_detach: bool,
     _verification_epoch: Mapping[str, object] | None,
 ) -> RunManifest:
     """One run, with ``written`` collecting every key it publishes.
@@ -2613,6 +2638,8 @@ def _execute_graph(
 
     if resume not in ("auto", "require", "forbid"):
         raise ValueError("resume must be 'auto', 'require', or 'forbid'.")
+    if type(_population_observer_detach) is not bool:
+        raise TypeError("_population_observer_detach must be a bool.")
     normalized_decisions: list[Decision] = []
     for decision in decisions:
         if isinstance(decision, Decision):
@@ -2931,7 +2958,10 @@ def _execute_graph(
             populations[node.id] = updated
 
         if _population_observer is not None:
-            _population_observer(node_id, _observer_snapshot(updated))
+            _population_observer(
+                node_id,
+                _observer_snapshot(updated) if _population_observer_detach else updated,
+            )
 
         if not hit:
             manifest_artifacts, record = _write_node(

packages/microcosm-graph/tests/test_graph_executor.py

@@ -4374,6 +4374,158 @@ def forbidden(population):
     _run(_graph(), source, ContentStore(tmp_path / "store"), _registry())
 
 
+def test_a_seal_only_observer_allocates_no_snapshot(
+    tmp_path: Path, monkeypatch
+) -> None:
+    """``_population_observer_detach=False`` skips the pickle round trip.
+
+    The detachment is one full independent population plus a temporary
+    serialized table buffer per reached node. An observer that only reads --
+    one that seals each population and keeps the seal -- needs none of it, and
+    this is the shape of ``test_absent_observer_allocates_no_snapshot`` with an
+    observer actually present.
+    """
+
+    def forbidden(population):
+        raise AssertionError("snapshot allocated for a seal-only observer")
+
+    monkeypatch.setattr(graph_executor, "_observer_snapshot", forbidden)
+    source = _source_path(tmp_path / "source")
+    seen: list[tuple[str, int]] = []
+    manifest = run_graph(
+        compile_graph(_graph()),
+        sources={"survey": source},
+        store=ContentStore(tmp_path / "store"),
+        kernels=_registry(),
+        _population_observer=lambda node_id, population: seen.append(
+            (node_id, population.frame.n("person"))
+        ),
+        _population_observer_detach=False,
+    )
+    assert [node_id for node_id, _ in seen] == list(manifest.nodes)
+    assert {n for _, n in seen} == {3}
+
+
+@pytest.mark.parametrize("warm", (False, True))
+def test_a_seal_only_observer_moves_no_key_receipt_or_store_object(
+    tmp_path: Path, warm: bool
+) -> None:
+    """A read-only observer in either mode is invisible to what a run exports."""
+    source = _source_path(tmp_path / "source")
+    compiled = compile_graph(_graph(leaf=False))
+    plain_store = ContentStore(tmp_path / "plain")
+    plain = run_graph(
+        compiled,
+        sources={"survey": source},
+        store=plain_store,
+        kernels=_registry(),
+    )
+    plain_objects = _object_bytes(plain_store)
+    store = ContentStore(tmp_path / "sealed")
+    if warm:
+        run_graph(
+            compiled, sources={"survey": source}, store=store, kernels=_registry()
+        )
+    seals: list[tuple[str, tuple]] = []
+
+    def observe(node_id, population):
+        seals.append(
+            (
+                node_id,
+                tuple(
+                    (entity, population.frame.n(entity))
+                    for entity in population.frame.entities
+                ),
+            )
+        )
+
+    sealed = run_graph(
+        compiled,
+        sources={"survey": source},
+        store=store,
+        kernels=_registry(),
+        _population_observer=observe,
+        _population_observer_detach=False,
+    )
+    assert [node_id for node_id, _ in seals] == list(sealed.nodes)
+    assert sealed.key == plain.key
+    assert {name: item.key for name, item in sealed.nodes.items()} == {
+        name: item.key for name, item in plain.nodes.items()
+    }
+    assert _object_bytes(store) == plain_objects
+
+
+def test_a_seal_only_observer_receives_the_live_population(tmp_path: Path) -> None:
+    """The mode's whole point, and the contract the caller is declaring.
+
+    In the default mode the observer receives an independent copy. Here it
+    receives the object the executor holds -- which is why the keyword is a
+    declaration that the observer will not retain or mutate it, and why the
+    default is unchanged.
+
+    Both arms compare WITHIN their own run. An earlier version of this test
+    compared the default run's snapshots against the seal-only run's frames,
+    and no object from one ``run_graph`` call can ever be ``is``-identical to
+    one from another, so that arm passed whatever the default did. An
+    adversarial pass found it; it is a real assertion now, and reducing
+    ``_observer_snapshot`` to ``lambda population: population`` turns it red.
+    """
+    source = _source_path(tmp_path / "source")
+    compiled = compile_graph(_graph(leaf=False))
+    detached: list[Population] = []
+    detached_manifest = run_graph(
+        compiled,
+        sources={"survey": source},
+        store=ContentStore(tmp_path / "detached"),
+        kernels=_registry(),
+        _population_observer=lambda node_id, population: detached.append(population),
+    )
+    live: list[Population] = []
+    manifest = run_graph(
+        compiled,
+        sources={"survey": source},
+        store=ContentStore(tmp_path / "live"),
+        kernels=_registry(),
+        _population_observer=lambda node_id, population: live.append(population),
+        _population_observer_detach=False,
+    )
+    attached = manifest.populations["survey"]
+    assert any(
+        seen.frame.table(entity) is attached.table(entity)
+        for seen in live
+        for entity in attached.entities
+    )
+    default_attached = detached_manifest.populations["survey"]
+    assert not any(
+        seen.frame.table(entity) is default_attached.table(entity)
+        for seen in detached
+        for entity in default_attached.entities
+    )
+
+
+def test_the_detach_keyword_is_a_bool_and_does_nothing_without_an_observer(
+    tmp_path: Path,
+) -> None:
+    source = _source_path(tmp_path / "source")
+    with pytest.raises(TypeError, match="_population_observer_detach must be a bool"):
+        run_graph(
+            compile_graph(_graph()),
+            sources={"survey": source},
+            store=ContentStore(tmp_path / "typed"),
+            kernels=_registry(),
+            _population_observer_detach=0,
+        )
+    plain = _run(_graph(), source, ContentStore(tmp_path / "plain"), _registry())
+    alone = run_graph(
+        compile_graph(_graph()),
+        sources={"survey": source},
+        store=ContentStore(tmp_path / "alone"),
+        kernels=_registry(),
+        _population_observer_detach=False,
+    )
+    assert alone.key == plain.key
+
+
 def test_a_gate_reached_only_through_bytes_still_derives_the_tier(
     tmp_path: Path,
 ) -> None: