P8a: the experiment harness (conditions, TerraPreserve loader, trials, records)

core/application/baseline.py

@@ -95,7 +95,9 @@ def __init__(self, reasons: Sequence[str]) -> None:
 @dataclass(frozen=True)
 class BaselineRequest:
     """What to establish. ``eligible`` is the caller's declaration that the fixture may serve as a
-    baseline (Doc 13 §13); ``now`` is the one instant used for everything that is hashed."""
+    baseline (Doc 13 §13); ``now`` is the one instant used for everything that is hashed.
+    ``allow_no_invariants`` (default off) accepts an empty ``definitions``: only the experiment
+    harness sets it, for condition A on a change request (C-206); the gates still run."""
 
     baseline_id: str
     lineage_id: str
@@ -108,6 +110,7 @@ class BaselineRequest:
     eligible: bool
     verifiers: VerifierTable = DEFAULT_VERIFIERS
     registry_version: int = 1
+    allow_no_invariants: bool = False
 
     def __post_init__(self) -> None:
         require_uuid(self.baseline_id, "BaselineRequest.baseline_id")
@@ -118,8 +121,12 @@ def __post_init__(self) -> None:
             raise DomainValidationError("BaselineRequest.eligible must be a bool")
         if not isinstance(self.context, PromotionContext):
             raise DomainValidationError("BaselineRequest.context must be a PromotionContext")
+        if not isinstance(self.allow_no_invariants, bool):
+            raise DomainValidationError("BaselineRequest.allow_no_invariants must be a bool")
         definitions = tuple(self.definitions)
-        if not definitions or not all(isinstance(d, Invariant) for d in definitions):
+        if (not definitions and not self.allow_no_invariants) or not all(
+            isinstance(d, Invariant) for d in definitions
+        ):
             raise DomainValidationError("BaselineRequest.definitions must hold Invariant objects")
         bindings = tuple(self.bindings)
         if not all(isinstance(b, SemanticBinding) for b in bindings):
@@ -353,6 +360,7 @@ def _write(
         invariant_registry_version=request.registry_version,
         now=request.now,
         state_id=prepared.state_id,
+        allow_no_invariants=request.allow_no_invariants,
     )
     if existing is not None and (
         existing.state_hash != state.state_hash or existing.lineage_id != state.lineage_id

core/application/orchestrator.py

@@ -31,6 +31,13 @@
 back with ``duplicate=True``.
 
 ``checkpoint`` is a test-only hook, called with the name of a stage after its records committed.
+
+**Experiment seams (P8a; C-196, C-198).** ``identity_strategy`` is the identity strategy given to
+``compare``; its default is ``DEFAULT_STRATEGY`` (``EvidenceIdentity``), the only one production
+code uses, and only the experiment harness passes the address-only ablation (C-127, condition C').
+``EvaluationRequest.open_targets`` (default none) is handed to ``decide``; ``check_open_targets``
+refuses a bad one against the candidate's parent before anything is stored. The result carries the
+identity report and the work the scope policy selected, for the experiment records.
 """
 
 from __future__ import annotations
@@ -52,8 +59,9 @@
 from core.domain.enums import CandidateSource, CandidateStatus, DecisionType
 from core.domain.errors import DomainValidationError, StaleParentError
 from core.domain.hashing import content_hash
+from core.domain.identity import IdentityReport
 from core.domain.ids import require_uuid
-from core.domain.impact import ImpactReport
+from core.domain.impact import ImpactReport, VerificationWorkItem
 from core.domain.invariant import Invariant
 from core.domain.patch import Patch
 from core.domain.promotion import (
@@ -75,6 +83,7 @@
 )
 from core.domain.verification import ConstructionStatus, VerificationReport
 from core.identity.engine import compare
+from core.identity.strategy import DEFAULT_STRATEGY, IdentityStrategy
 from core.impact.engine import compute_impact
 from core.impact.inputs import PriorProof
 from core.promotion import ids as pids
@@ -85,7 +94,7 @@
     commit_promotion,
     record_decision,
 )
-from core.promotion.decide import decide
+from core.promotion.decide import check_open_targets, decide
 from core.promotion.policy import dail_policy
 from core.terraform_model.normalization import normalize_directory
 from core.verification.dispatch import DEFAULT_VERIFIERS, VerifierTable
@@ -141,6 +150,7 @@ class EvaluationRequest:
     bindings: tuple[SemanticBinding, ...]
     context: PromotionContext
     oracle_verdicts: tuple[OracleVerdict, ...] = ()
+    open_targets: tuple[str, ...] = ()
     _PATCH_SOURCE: ClassVar[CandidateSource] = CandidateSource.FIXED_PATCH
 
     def __post_init__(self) -> None:
@@ -167,6 +177,12 @@ def __post_init__(self) -> None:
         verdicts = tuple(self.oracle_verdicts)
         if not all(isinstance(v, OracleVerdict) for v in verdicts):
             raise EvaluationInputError("EvaluationRequest.oracle_verdicts must hold OracleVerdict")
+        if isinstance(self.open_targets, str):
+            raise EvaluationInputError("EvaluationRequest.open_targets must be a list of ids")
+        opened = tuple(self.open_targets)
+        if not all(isinstance(t, str) and t for t in opened) or len(set(opened)) != len(opened):
+            raise EvaluationInputError("EvaluationRequest.open_targets: invariant ids, each once")
+        object.__setattr__(self, "open_targets", opened)
         object.__setattr__(self, "definitions", definitions)
         object.__setattr__(self, "bindings", bindings)
         object.__setattr__(self, "oracle_verdicts", verdicts)
@@ -222,6 +238,8 @@ class EvaluationResult:
     verification: VerificationReport | None = None
     evidence_ids: Mapping[str, str] | None = None
     duplicate: bool = False
+    identity: IdentityReport | None = None
+    work: tuple[VerificationWorkItem, ...] | None = None
 
 
 class EvaluationOrchestrator:
@@ -238,6 +256,7 @@ def __init__(
         verifiers: VerifierTable = DEFAULT_VERIFIERS,
         policy_factory: Callable[..., PromotionPolicy] = dail_policy,
         checkpoint: Callable[[str], None] | None = None,
+        identity_strategy: IdentityStrategy = DEFAULT_STRATEGY,
     ) -> None:
         if not callable(factory):
             raise EvaluationInputError("EvaluationOrchestrator: factory must open a unit of work")
@@ -249,6 +268,12 @@ def __init__(
             raise EvaluationInputError("EvaluationOrchestrator: policy_version must be text")
         if checkpoint is not None and not callable(checkpoint):
             raise EvaluationInputError("EvaluationOrchestrator: checkpoint must be callable")
+        if not callable(getattr(identity_strategy, "decide", None)) or not isinstance(
+            getattr(identity_strategy, "algorithm_version", None), str
+        ):
+            raise EvaluationInputError(
+                "EvaluationOrchestrator: identity_strategy must be an IdentityStrategy"
+            )
         try:
             flag = config.get(_FLAG)
         except (KeyError, AttributeError) as missing:
@@ -263,6 +288,14 @@ def __init__(
         self._policy_factory = policy_factory
         self._checkpoint = checkpoint
         self._automatic = flag
+        self._identity = identity_strategy
+
+    def _policy(self, request: EvaluationRequest) -> PromotionPolicy:
+        return self._policy_factory(
+            policy_version=self._policy_version,
+            invariants=request.definitions,
+            automatic_enabled=self._automatic,
+        )
 
     def evaluate(self, request: EvaluationRequest) -> EvaluationResult:
         """Evaluate ``request`` (see the module docstring). Raises ``EvaluationInputError`` for a
@@ -296,6 +329,8 @@ def __init__(self, orchestrator: EvaluationOrchestrator, request: EvaluationRequ
         self.stage = "load"
         self.candidate: CandidateState | None = None
         self.failed_now = False  # this call is the one that moved the candidate to FAILED
+        self.identity: IdentityReport | None = None
+        self.work: tuple[VerificationWorkItem, ...] | None = None
 
     # --- small helpers -------------------------------------------------------------------------
 
@@ -334,6 +369,7 @@ def execute(self) -> EvaluationResult:
                 f"the patch was written against state {r.patch.parent_state_id}, which is not a "
                 f"state of lineage {r.lineage_id}"
             )
+        check_open_targets(parent, o._policy(r), r.open_targets)  # C-196, before any write
         self.stage = "candidate"
         candidate = self._create(parent)
         at = candidate.created_at
@@ -353,7 +389,9 @@ def execute(self) -> EvaluationResult:
             parent_state_id=parent.state_id,
             candidate_id=self._cid,
             now=at,
+            strategy=o._identity,
         )
+        self.identity = identity
         self._record(
             lambda u: self.recorder.record_identity(
                 u, candidate, identity, attempt_number=r.attempt_number, at=at, when=self._when()
@@ -398,7 +436,8 @@ def execute(self) -> EvaluationResult:
         )
         self._passed("after_impact")
         self.stage = "verification"
-        work = o._scope_policy.select(impact, r.definitions)
+        work = o._scope_policy.select(impact, r.definitions, parent_refs=parent.invariant_refs)
+        self.work = tuple(work)
         inputs = VerificationInputs(
             candidate_resources=outcome.resources,
             parent_resources=parent.resources,
@@ -634,6 +673,7 @@ def _decision(
             now=at,
             attempt_number=r.attempt_number,
             oracle_verdicts=r.oracle_verdicts,
+            open_targets=() if stale else r.open_targets,
         )
         return outcome, stale
 
@@ -648,11 +688,7 @@ def _decide(
         candidate = self._stored()
         if candidate.status in _TERMINAL:
             return self._stored_result(candidate, impact, report, evidence_ids)
-        policy = o._policy_factory(
-            policy_version=o._policy_version,
-            invariants=r.definitions,
-            automatic_enabled=o._automatic,
-        )
+        policy = o._policy(r)
         outcome, stale = self._decision(candidate, impact, report, policy, at)
         try:
             with o._factory() as uow:
@@ -675,6 +711,8 @@ def _decide(
                 impact=impact,
                 verification=report,
                 evidence_ids=evidence_ids,
+                identity=self.identity,
+                work=self.work,
             )
         self.stage = "commit"
         with o._factory() as uow:
@@ -735,6 +773,8 @@ def _after_commit(
             verification=report,
             evidence_ids=evidence_ids,
             duplicate=committed.duplicate,
+            identity=self.identity,
+            work=self.work,
         )
 
     def _stored_result(
@@ -772,6 +812,8 @@ def _stored_result(
             verification=report,
             evidence_ids=evidence_ids,
             duplicate=True,
+            identity=self.identity,
+            work=self.work,
         )
 
     # --- failures that are not decisions -------------------------------------------------------