Add opt-in bound-intent fence to the ordered-effect engine

docs/roadmaps/utel-release-successor.md

@@ -209,6 +209,8 @@ The first implementation slice, [PR #3570](https://github.com/FS-GG/.github/pull
 
 `scripts/release_successor_execution.py` is the effect-order/recovery policy. Its fake-provider controls require a fresh admission for journal intent, remote dispatch and settlement; a response alone does not mark an effect verified, and a 404 after an uncertain write does not authorize blind replay. The scoped publisher supplies a protected Git commit/ref journal, non-forced compare-and-swap, native Actions and main-head admission rereads, and provider-specific package and release readback. Journal genesis itself requires the admitted exact candidate and verifies the protected ref was previously absent. The writer must fail closed on unreadable or conflicting evidence. The ambient `GITHUB_TOKEN` is a provider credential, not the admission source; the ordinary App has only the authority-repository contents scope. The five retired workflows remain sealed.
 
+The engine also provides an opt-in bound-intent fence (`BoundIntentFence`, `fence=` on `advance`, `advance_effects` and `full_transition`; [Coordination#1026](https://github.com/FS-GG/FS.GG.Coordination/issues/1026)). One fence belongs to one caller invocation. It is armed only by the intent compare-and-swap that invocation just won, bound to the exact journal state that write produced, and spent immediately before `provider.dispatch` after a fresh journal read finds exactly that state; an unreadable, changed or merely claimed write refuses the send. The permit is single-use, one may be outstanding, and it names the content and effect. The provider the engine calls is wrapped so that `dispatch` itself refuses unless that release happened, which also stops a regressed engine that tries to resend an inherited intent. An `intent` found in the journal by any later invocation, or by the same one in a later call, never arms a fence and is settled by readback only. Without a fence behavior is byte-for-byte unchanged, and the journal schema, protocol and every existing caller are untouched. The fence records only invocation-local evidence (`written`, `sends`); it grants no authority, admission or retry. `tests/release-successor-execution/` proves each part with causal controls and source mutants.
+
 `scripts/release-successor-artifact.py` is an offline guard for an artifact, run and archive supplied by a trusted GitHub API reader. It verifies the successful first-attempt exact-main candidate run, repository and artifact identity, the outer artifact SHA-256, exact safe members, the source-bound release manifest, all package archive digests, runtime qualification and stable predecessor before extraction. Reruns are refused because artifact metadata does not bind a run attempt. Its JSON arguments are evidence inputs, not authenticated API responses by themselves; a live writer must fetch them through its own scoped GitHub API client and recheck the current main SHA and version uniqueness before the first effect.
 
 Live observations on 2026-09-19 found the operating-v1 cutover genesis in `FS-GG/FS.GG.Coordination.Authority`, but the canonical `fleet-v1-admission:fs-gg-production` operation journal ref `refs/heads/fsgg/v2/journal/operation/79` returned HTTP 404. The imported registry deliberately has no public initializer, so this scoped release cannot use that operation without a separate authority migration. The `.github` `release-successor` environment is main-only with no required reviewers, matching the user's single-operator decision. The ordinary ledger App `4882140` has active FS-GG installation `160261608`; [read-only proof run #35454913240](https://github.com/FS-GG/.github/actions/runs/35454913240) verified its key, contents permission and a token scoped to the protected authority repository. The nuget.org owner registered Trusted Publishing for the distinct `release-successor-publish.yml` filename, that environment and exactly the three package IDs, with push-only-new-versions scope. These are credential prerequisites, not publication evidence.

scripts/release_successor_execution.py

@@ -77,6 +77,99 @@ def observe(self, effect: Effect) -> Observation: ...
     def dispatch(self, effect: Effect) -> Dispatch: ...
 
 
+_INVOCATION = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}")
+BOUND_INTENT_FENCE = 1  # capability marker: the engine accepts `fence=` and enforces BoundIntentFence
+
+
+class BoundIntentFence:
+    """Opt-in binding of a journal `intent` to the invocation that wrote it.
+
+    One fence belongs to exactly one invocation of a caller. The engine arms it only
+    from its own won intent compare-and-swap and spends it only immediately before
+    `provider.dispatch`, after re-reading the journal and finding exactly the state
+    that write produced. An intent found in the journal at the start of a transition
+    (written by any earlier invocation, or by this one in an earlier call) never
+    arms the fence and is therefore never a send permit. A permit is single-use and
+    at most one is outstanding; once a release is refused the fence arms nothing more.
+    Callers that do not pass a fence behave as before.
+
+    Trust root: the release compares the journal's state, not a writer identity, so
+    it relies on the journal's compare-and-swap being linearizable (the protected
+    journal's non-forced fast-forward plus matching readback is) and on `journal.read()`
+    being a fresh read, not a cached one. `invocation` names the owner in refusals; the
+    binding itself is the identity of this object and of the state its own write produced.
+    """
+
+    def __init__(self, invocation: str):
+        if type(invocation) is not str or not _INVOCATION.fullmatch(invocation):
+            raise Refused("bound-intent fence requires a closed invocation identifier")
+        self.invocation = invocation
+        self._armed: tuple[str, str, JournalState] | None = None
+        self._released: str | None = None
+        self._refused = False
+        self._written: list[tuple[str, int]] = []
+        self._sent: list[str] = []
+
+    @property
+    def written(self) -> tuple[tuple[str, int], ...]:
+        """(effect, journal generation) of each intent this invocation wrote."""
+        return tuple(self._written)
+
+    @property
+    def sends(self) -> tuple[str, ...]:
+        """Effects whose single provider send this invocation was permitted."""
+        return tuple(self._sent)
+
+    def arm(self, content_id: str, effect: str, written: JournalState) -> None:
+        """Bind the intent this invocation just won to its exact journal state."""
+        if self._armed is not None or self._released is not None or self._refused:
+            raise Refused(f"{effect}: a prior intent in invocation {self.invocation} was never dispatched")
+        if effect in self._sent or any(name == effect for name, _ in self._written):
+            raise Refused(f"{effect}: this invocation already wrote an intent for the effect")
+        if written.effects.get(effect) != "intent":
+            raise Refused(f"{effect}: bound journal state carries no intent")
+        self._armed = (content_id, effect, written)
+        self._written.append((effect, written.generation))
+
+    def release(self, content_id: str, effect: str, journal: Journal) -> None:
+        """Spend the permit for exactly this effect; refuse unless the journal still shows our write."""
+        armed, self._armed = self._armed, None  # spent on any outcome: never retried
+        if armed is None or armed[0] != content_id or armed[1] != effect:
+            self._refused = True
+            raise Refused(f"{effect}: send fence {self.invocation}: no intent written by this invocation permits a send")
+        try:
+            current = journal.read()
+        except Exception as error:
+            self._refused = True
+            raise Refused(f"{effect}: send fence {self.invocation}: journal unreadable before send: {error}") from error
+        if current != armed[2]:
+            self._refused = True
+            raise Refused(f"{effect}: send fence {self.invocation}: journal no longer holds the intent this invocation wrote")
+        self._released = effect
+        self._sent.append(effect)
+
+    def guard(self, provider: Provider) -> Provider:
+        """A provider whose `dispatch` itself refuses unless this fence just released that effect."""
+        return _GuardedProvider(provider, self)
+
+    def _consume(self, effect: str) -> None:
+        released, self._released = self._released, None
+        if released != effect:
+            raise Refused(f"{effect}: send fence {self.invocation}: no intent written by this invocation permits a send")
+
+
+class _GuardedProvider:
+    def __init__(self, provider: Provider, fence: BoundIntentFence):
+        self._provider, self._fence = provider, fence
+
+    def observe(self, effect: Effect) -> Observation:
+        return self._provider.observe(effect)
+
+    def dispatch(self, effect: Effect) -> Dispatch:
+        self._fence._consume(effect.identity)
+        return self._provider.dispatch(effect)
+
+
 def ordered_effects(manifest: dict) -> tuple[Effect, ...]:
     descriptor = manifest["descriptor"]
     content_id = "sha256:" + hashlib.sha256(
@@ -142,23 +235,29 @@ def _observe(provider: Provider, effect: Effect) -> Observation:
     return observation
 
 
-def advance(manifest: dict, journal: Journal, admission: Admission, provider: Provider) -> str:
+def advance(manifest: dict, journal: Journal, admission: Admission, provider: Provider, *, fence: BoundIntentFence | None = None) -> str:
     """Attempt at most one provider effect; return `verified`, `waiting`, or `complete`.
 
     A live caller first verifies the manifest and archives through release-saga.py.
     The protected operation journal must already exist and bind the content id.
     """
-    return advance_effects(manifest["contentId"], ordered_effects(manifest), journal, admission, provider)
+    return advance_effects(manifest["contentId"], ordered_effects(manifest), journal, admission, provider, fence=fence)
 
 
-def advance_effects(content_id: str, effects: tuple[Effect, ...], journal: Journal, admission: Admission, provider: Provider) -> str:
+def advance_effects(content_id: str, effects: tuple[Effect, ...], journal: Journal, admission: Admission, provider: Provider, *, fence: BoundIntentFence | None = None) -> str:
     """Apply the same admitted, readback-first policy to an exact ordered effect set."""
-    result = full_transition(content_id, effects, journal, admission, provider)
+    result = full_transition(content_id, effects, journal, admission, provider, fence=fence)
     return "waiting" if result.state == "unknown" else result.state
 
 
-def full_transition(content_id: str, effects: tuple[Effect, ...], journal: Journal, admission: Admission, provider: Provider, *, prepare_dispatch=None) -> Transition:
-    """One full-prefix transition, never a cached authorization for polling."""
+def full_transition(content_id: str, effects: tuple[Effect, ...], journal: Journal, admission: Admission, provider: Provider, *, prepare_dispatch=None, fence: BoundIntentFence | None = None) -> Transition:
+    """One full-prefix transition, never a cached authorization for polling.
+
+    With a `fence`, a provider send additionally requires the intent that this very
+    invocation just wrote (see BoundIntentFence); without one the behavior is unchanged.
+    """
+    if fence is not None:
+        provider = fence.guard(provider)
     current = journal.read()
     if current.content_id != content_id or current.generation < 1:
         raise Refused("protected journal does not bind the release candidate")
@@ -196,11 +295,16 @@ def full_transition(content_id: str, effects: tuple[Effect, ...], journal: Journ
             raise Refused(f"{effect.identity}: intent admission denied")
         if not journal.compare_and_swap(current, effect.identity, "intent"):
             raise Refused(f"{effect.identity}: journal intent CAS conflict")
+        written = JournalState(current.generation + 1, content_id, {**current.effects, effect.identity: "intent"})
+        if fence is not None:
+            fence.arm(content_id, effect.identity, written)
         if prepare_dispatch is not None:
             prepare_dispatch(effect)
         # Re-read authority after all potentially slow preparation at the remote boundary.
         if not admission.authorize(content_id, effect.identity, "dispatch", effect.request_digest):
             raise Refused(f"{effect.identity}: dispatch admission denied")
+        if fence is not None:
+            fence.release(content_id, effect.identity, journal)
         outcome = provider.dispatch(effect)
         if outcome.state not in {"applied", "unknown", "refused"}:
             raise Refused(f"{effect.identity}: unrecognized dispatch outcome")
@@ -209,6 +313,5 @@ def full_transition(content_id: str, effects: tuple[Effect, ...], journal: Journ
         # Even a success response is not completion. A later independently observed
         # advance must read back; an admitted caller may wait within its original window.
         return Transition("unknown" if outcome.state == "unknown" else "waiting",
-                          PendingObservation(effect, JournalState(current.generation + 1, content_id,
-                              {**current.effects, effect.identity: "intent"})))
+                          PendingObservation(effect, written))
     return Transition("complete")