Cut D PR 2a: pair-serving authority on its own event-log partition, read by every effectful consumer

dag/gunbc/ci/ci_layer_roots.dag

@@ -374,6 +374,10 @@ data excl_local_repo_wet_bash_materialized_reason: String = "real host-effect ex
 
 data excl_local_repo_wet_bash_materialized_dissolve: DissolutionCondition = unbound_dissolution(description: "mock_response coverage lands for shell.Exec.Run over a sealed transport script and the emitted-bytes assertions are re-checked under a fixed mock -- which the transport_script_stdin_byte_fidelity note already refuses as vacuous for a byte-fidelity claim -- OR the bin-witness wet class regains an executing per-PR consumer; then these fns re-enroll there, drop off local_repo_wet_schedule, and this row deletes")
 
+data excl_local_repo_wet_executor_identity_reason: String = "real host-effect execution witness whose effects are the EXECUTOR'S OWN IDENTITY AND EVENT-LOG READ: the instrument entry reads the executing host's hostname (shell.Exec.RunArgv, no mock_response), resolves that host's event log layout, and reads each claimed group's serving authority from the log before judging any standing (gunbc.instruments.fabric_capacity_standing fabric_capacity_standing over gunbc.spark.pair_serving_authority_log). No temporary repository is built and nothing is written; on a runner that is no dashboard host the read refuses at the layout step and the entry reports every group unread, which is the route the claim asserts. It is the wet half of the pairing obligation whose hermetic half is test.claim.spark.fabric_capacity_standing_witness over a supplied roster."
+
+data excl_local_repo_wet_executor_identity_dissolve: DissolutionCondition = unbound_dissolution(description: "mock_response coverage lands for shell.Exec.RunArgv and the event-log head read, so the instrument entry can be driven hermetically with a fixed executor and a fixed partition head; then the claim re-enrolls as an ordinary hermetic discovery row, drops off local_repo_wet_schedule, and this row deletes")
+
 data excl_local_repo_wet_host_probe_reason: String = "real host-effect execution witness whose ONLY effect is a READ-ONLY HOST PATH PROBE (shell.PosixCommandV.Check, i.e. command -v) -- refused by the hermetic envelope (no mock_response for operation Check), so excluded from the discovery corpus and executed by the required floor's local-repo wet lane. THIS IS A DISTINCT ADMITTED EFFECT, NOT THE THROWAWAY-REPOSITORY PROPERTY excl_local_repo_wet_reason NAMES: these fns build no temporary repository and write nothing. What they share with that class, and what admits them to the same lane, is the negative half -- no network, no cargo, no remote host, no install media. THE VERDICT IS HOST-INDEPENDENT ON THE TOOL-PRESENCE AXIS, MEASURED RATHER THAN ASSUMED (2026-09-02, PR #10055): every arm of both fns returns the literal true, and the npm fn was executed on one host in both conditions -- probe found, and probe exit=127 with sh still resolvable -- returning the same verdict in each. So admitting them imports no host-conditional red. The residual red these fns CAN produce is route-unreachability (no shell at all: TypeError failed to execute sh), which is the same reachability risk every member of this lane already carries and is not specific to the probed tool. PROVENANCE CARRIED FORWARD from the bin-wet row this replaces: falsifier Codex materialization reuses the same PosixCommandV.Check.exists path (migrated off shell.Which.Check, crisp-wren-896, PR #8590)."
 
 data excl_local_repo_wet_host_probe_dissolve: DissolutionCondition = unbound_dissolution(description: "mock_response coverage lands for shell.PosixCommandV.Check, so observe_host_cli_dependency can be asserted under a fixed mock; then these fns re-enroll as ordinary hermetic discovery rows, drop off local_repo_wet_schedule, and this row deletes")
@@ -913,6 +917,21 @@ data witness_exclusion_frontier: List<WitnessExclusionRow> = [
     classification: LocalRepoWetLane,
     reason: excl_local_repo_wet_reason,
     dissolution: excl_local_repo_wet_dissolve},
+  WitnessExclusionRow {
+    pattern: "pair_serving_authority_log_real_execution_witness_test.dag",
+    classification: LocalRepoWetLane,
+    reason: excl_local_repo_wet_reason,
+    dissolution: excl_local_repo_wet_dissolve},
+  WitnessExclusionRow {
+    pattern: "fabric_capacity_standing_wet_witness_test.dag",
+    classification: LocalRepoWetLane,
+    reason: excl_local_repo_wet_executor_identity_reason,
+    dissolution: excl_local_repo_wet_executor_identity_dissolve},
+  WitnessExclusionRow {
+    pattern: "spark_pair_serving_apply_wet_witness_test.dag",
+    classification: LocalRepoWetLane,
+    reason: excl_local_repo_wet_executor_identity_reason,
+    dissolution: excl_local_repo_wet_executor_identity_dissolve},
   WitnessExclusionRow {
     pattern: "devboot_text_blob_real_execution_witness_test.dag",
     classification: LocalRepoWetLane,

dag/gunbc/fabric/fabric_event_log.dag

@@ -18,9 +18,10 @@ import product.capacity.pool { Pool, PoolReading, PoolRead, PoolReadingRefused,
 import product.capacity.event_chain {
   PartitionId, EventId, ChainEvent, ChainEnvelope, HeadExpectation, HeadAbsent, HeadAt, head_expectation_eq,
   event_parent_expectation, AppendDecision, AppendAdmitted, AppendStale, ChainWalk, ChainWalked, ChainIncomplete, ChainBudgetExhausted, chain_from_head,
+  EventDecode, EventDecoded, EventUndecodable,
 }
 import product.capacity.pool_events {
-  PoolEvent, pool_event_wire_text, pool_event_decode, PoolEventDecode, PoolEventDecoded, PoolEventUndecodable,
+  PoolEvent, pool_event_wire_text, pool_event_decode,
   PoolFold, PoolFolded, PoolFoldRefused, pool_fold, SeatRequest, SeatProposal, SeatProposed, SeatRefused, propose_acquire, grant_from_admission,
 }
 import product.capacity.lease { LeasePolicy, LeaseGrant, release_law_eq, release_law_wire }
@@ -172,17 +173,25 @@ fn event_log_observe_head_from_mirror(layout: EventLogLayout, partition: Partiti
 // READING A PARTITION: fetch the head's objects into the mirror, then read each event by its
 // address following parents, bounded, and hand the collected envelopes to the pure walk so the
 // chain's shape is decided by one authority.
-type PartitionRead
-  = PartitionReadOk { head: HeadExpectation, walk: ChainWalk<PoolEvent> }
+// THE LOG IS GENERIC IN ITS PAYLOAD, AND THE CODEC IS THE PAYLOAD'S. One object per event, one ref
+// per partition, one compare-and-set on the head -- none of that is about seats. The pool events
+// were the first partition kind and the read and append were written against their codec, which
+// made a second partition kind (the pair-serving authority, gunbc.spark.pair_serving_authority_log)
+// need a second read and a second append: the §2 duplication, with the linearization law copied
+// twice. So the read walks and the append publishes for ANY payload, taking the payload's encode
+// and decode, and event_log_read_partition / event_log_append below are those two with the pool
+// codec supplied -- specializations of one root, not a second spelling of it.
+type PartitionRead<P>
+  = PartitionReadOk { head: HeadExpectation, walk: ChainWalk<P> }
   | PartitionReadRefused { step: String, reason: String }
 
-type CollectState {
+type CollectState<P> {
   current: HeadExpectation
-  collected: List<ChainEnvelope<PoolEvent>>
+  collected: List<ChainEnvelope<P>>
   refused: String?
 }
 
-fn collect_step(layout: EventLogLayout, st: CollectState) -> CollectState {
+fn collect_step<P>(layout: EventLogLayout, st: CollectState<P>, decode: fn(String) -> EventDecode<P>) -> CollectState<P> {
   match st.refused {
     Present { value: _ } => st
     Absent =>
@@ -193,9 +202,9 @@ fn collect_step(layout: EventLogLayout, st: CollectState) -> CollectState {
           if !r.success {
             CollectState { current: st.current, collected: st.collected, refused: Present { value: join(["cat-file ", h as String, " refused"], "") } }
           } else {
-            match pool_event_decode(text: r.content) {
-              PoolEventUndecodable { reason: why } => CollectState { current: st.current, collected: st.collected, refused: Present { value: join(["event ", h as String, " undecodable: ", why], "") } }
-              PoolEventDecoded { event: e } => CollectState {
+            match decode(r.content) {
+              EventUndecodable { reason: why } => CollectState { current: st.current, collected: st.collected, refused: Present { value: join(["event ", h as String, " undecodable: ", why], "") } }
+              EventDecoded { event: e } => CollectState {
                 current: event_parent_expectation(e: e),
                 collected: concat(st.collected, [ChainEnvelope { id: h, event: e }]),
                 refused: none,
@@ -207,7 +216,7 @@ fn collect_step(layout: EventLogLayout, st: CollectState) -> CollectState {
   }
 }
 
-fn event_log_read_partition(layout: EventLogLayout, partition: PartitionId, budget: Nat) -> PartitionRead {
+fn event_log_read_partition_with<P>(layout: EventLogLayout, partition: PartitionId, budget: Nat, decode: fn(String) -> EventDecode<P>) -> PartitionRead<P> {
   match event_log_ensure_mirror(layout: layout) {
     MirrorRefused { step: s, reason: why } => PartitionReadRefused { step: s, reason: why }
     MirrorReady =>
@@ -222,7 +231,7 @@ fn event_log_read_partition(layout: EventLogLayout, partition: PartitionId, budg
               if f.exit_code != 0 {
                 PartitionReadRefused { step: "fetch", reason: join(["exit ", to_string(f.exit_code), ": ", trim(s: f.stderr)], "") }
               } else {
-                let st = fold(nat_range_inclusive(lo: 1, hi: budget), init: CollectState { current: HeadAt { id: h }, collected: [], refused: none }, f: (acc, _i) => collect_step(layout: layout, st: acc))
+                let st = fold(nat_range_inclusive(lo: 1, hi: budget), init: CollectState { current: HeadAt { id: h }, collected: [], refused: none }, f: (acc, _i) => collect_step(layout: layout, st: acc, decode: decode))
                 match st.refused {
                   Present { value: why } => PartitionReadRefused { step: "read", reason: why }
                   Absent => PartitionReadOk { head: HeadAt { id: h }, walk: chain_from_head(envs: st.collected, head: HeadAt { id: h }, budget: budget) }
@@ -234,6 +243,10 @@ fn event_log_read_partition(layout: EventLogLayout, partition: PartitionId, budg
   }
 }
 
+fn event_log_read_partition(layout: EventLogLayout, partition: PartitionId, budget: Nat) -> PartitionRead<PoolEvent> {
+  event_log_read_partition_with(layout: layout, partition: partition, budget: budget, decode: fn(text) { pool_event_decode(text: text) })
+}
+
 // APPEND: publish the bytes as an object in the mirror, then advance the partition ref at the
 // placement with a lease on the expected head. A refused push is classified by re-observing the
 // head, because git's exit code cannot tell a lost race from an unreachable placement: the head
@@ -249,7 +262,7 @@ type EventAppend
 // and a completed foreign write published another writer's event under this writer's admitted
 // append. The lease guards the head, not the object's content. The enrolled evidence is
 // test.claim.fabric_event_log_append_real_execution.
-fn event_log_append(layout: EventLogLayout, partition: PartitionId, event: ChainEvent<PoolEvent>, expected: HeadExpectation) -> EventAppend {
+fn event_log_append_with<P>(layout: EventLogLayout, partition: PartitionId, event: ChainEvent<P>, expected: HeadExpectation, encode: fn(ChainEvent<P>) -> String) -> EventAppend {
   if !head_expectation_eq(a: event_parent_expectation(e: event), b: expected) {
     EventAppendRefused { step: "parent", reason: "the event's parent is not the head the append expects" }
   } else {
@@ -259,7 +272,7 @@ fn event_log_append(layout: EventLogLayout, partition: PartitionId, event: Chain
         match event_log_remote_word(remote: layout.remote) {
           Absent => EventAppendRefused { step: "remote", reason: "the event log is unplaced" }
           Present { value: remote } => {
-            let h = git.Plumbing.HashObjectWriteStdin(address: mirror_address(layout: layout), content: pool_event_wire_text(event: event))
+            let h = git.Plumbing.HashObjectWriteStdin(address: mirror_address(layout: layout), content: encode(event))
             if h.exit_code != 0 {
               EventAppendRefused { step: "hash-object", reason: join(["exit ", to_string(h.exit_code), ": ", trim(s: h.stderr)], "") }
             } else {
@@ -285,6 +298,10 @@ fn event_log_append(layout: EventLogLayout, partition: PartitionId, event: Chain
   }
 }
 
+fn event_log_append(layout: EventLogLayout, partition: PartitionId, event: ChainEvent<PoolEvent>, expected: HeadExpectation) -> EventAppend {
+  event_log_append_with(layout: layout, partition: partition, event: event, expected: expected, encode: fn(e) { pool_event_wire_text(event: e) })
+}
+
 // THE SEAT TRANSACTION, WET AND BOUNDED: read, fold, propose, append; a stale append re-reads and
 // decides again, at most `attempts` times, and exhaustion is its own typed outcome rather than a
 // seat. The grant is minted only from an admitted append.

dag/gunbc/instruments/fabric_capacity_standing.dag

@@ -11,10 +11,13 @@ import gunbc.spark.fabric_switch_observed { FabricGroup, fabric_group_wire }
 import std.content_hash { ContentHash, Sha256Hash, sha256_hex_digest }
 import product.placement_supply { HostIdentity }
 import gunbc.spark.pair_serving_authority {
-  PairServingGroupAuthority, spark_pair_serving_authorities, pair_serving_capacity_subject,
+  PairServingGroupAuthority, pair_serving_capacity_subject,
   CapacityOrdinarySubject, CapacitySuspendedSubject, CapacityReleasedSubject,
 }
 import gunbc.spark.vllm_serving_launch { VllmServingLaunch, vllm_serving_launch }
+import gunbc.spark.pair_serving_authority_log { CurrentAuthority, CurrentAuthorityRead, CurrentAuthorityUnread, current_pair_serving_authorities }
+import gunbc.spark.fabric_reach { ExecutorReachKnown, ExecutorReachUnknown, observe_executor_reach }
+import std.resources { Network }
 import gunbc.spark.vllm_kv_layout_observe {
   VllmKvLayoutReceipt, admit_kv_layout,
 }
@@ -552,6 +555,32 @@ fn capacity_standing_verdict(standings: List<GroupStanding>) -> ProcessExit {
   }
 }
 
-fn fabric_capacity_standing() -> ProcessExit {
-  capacity_standing_verdict(standings: map(spark_pair_serving_authorities, a => standing_for_authority(a: a)))
+// THE AUTHORITY IS READ FROM THE LOG, NOT THE SOURCE ROW. gunbc.spark.pair_serving_authority_log
+// folds the resting row through every recorded transition, so a group a transaction has suspended
+// is reported suspended here without a commit. An authority that could not be read is a refused
+// standing naming the read's step -- the instrument does not stand for a group whose state it
+// could not establish, and it does not fall back to the row.
+fn standing_for_current(c: CurrentAuthority) -> GroupStanding {
+  match c {
+    CurrentAuthorityRead { authority: a, head: _, generation: _ } => standing_for_authority(a: a)
+    CurrentAuthorityUnread { group: g, step: s, reason: why } =>
+      refused_standing(text: join([fabric_group_wire(g: g) as String, ": serving authority could not be read at ", s, ": ", why], ""))
+  }
+}
+
+// The verdict over a SUPPLIED current roster: the real group_report route for every read
+// authority, so a witness can hand in the resting row at generation 0 and still execute the
+// producer and the binding, while the live entry below hands in what the executor's log says.
+fn fabric_capacity_standing_current(current: List<CurrentAuthority>) -> ProcessExit {
+  capacity_standing_verdict(standings: map(current, c => standing_for_current(c: c)))
+}
+
+fn fabric_capacity_standing() -> ProcessExit
+  uses net: Network
+{
+  match observe_executor_reach() {
+    ExecutorReachUnknown { cause: c } => exit_failure(reason: join(["fabric capacity standing: ", c], ""))
+    ExecutorReachKnown { short_hostname: executor, path: _ } =>
+      fabric_capacity_standing_current(current: current_pair_serving_authorities(short_hostname: executor))
+  }
 }