The off-fleet authority witnesses take their executor as a supplied value

dag/gunbc/ci/ci_layer_roots.dag

@@ -382,10 +382,6 @@ 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")
@@ -977,16 +973,6 @@ data witness_exclusion_frontier: List<WitnessExclusionRow> = [
     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/test/claim/spark/fabric_capacity_standing_witness_test.dag

@@ -1,6 +1,6 @@
 module test.claim.spark.fabric_capacity_standing_witness
 
-import std.types { Bool, String, NonEmptyStr, List }
+import std.types { Bool, String, NonEmptyStr, List, EpochSecs }
 import std.process { ProcessExit, ExitSuccess, ExitFailure }
 import v2.std.optional { Present, Absent }
 import gunbc.spark.fabric_switch_observed { fabric_group_hosts }
@@ -21,7 +21,7 @@ import gunbc.instruments.fabric_capacity_standing {
   GroupStanding, capacity_standing_verdict, group_report, fabric_capacity_standing_current,
   standing_for_authority,
 }
-import gunbc.spark.pair_serving_authority_log { CurrentAuthority, CurrentAuthorityRead, AuthorityTransitionRecord }
+import gunbc.spark.pair_serving_authority_log { CurrentAuthority, CurrentAuthorityRead, CurrentAuthorityUnread, AuthorityTransitionRecord, current_pair_serving_authorities }
 import product.capacity.event_chain { HeadAbsent }
 import std.temporal_effect { held_lease, LeaseEpoch, LeaseRunningExpected }
 import std.content_hash { ContentHash, Sha256Hash, sha256_hex_digest }
@@ -315,8 +315,7 @@ test fn one_refusing_member_refuses_the_whole_standing() -> Bool {
 // THE ROSTER IS SUPPLIED AT GENERATION 0 -- every claimed group's resting row, which is what the
 // authority log folds to when no transition is recorded -- so the claim stays hermetic while the
 // real group_report producer and the binding that maps it over the roster both execute. The
-// live entry, which reads that roster from the event log first, is executed by
-// test.claim.spark.fabric_capacity_standing_wet_witness on the local-repo wet lane.
+// roster READ is exercised below with a supplied executor.
 fn w_resting_current() -> List<CurrentAuthority> {
   map(spark_pair_serving_authorities, a => CurrentAuthorityRead { authority: a, head: HeadAbsent, generation: 0, grant: none, entry_state: none, previous: none, admitted_by: none, transitions: [] as List<AuthorityTransitionRecord> })
 }
@@ -331,6 +330,38 @@ test fn the_verdict_over_the_resting_roster_runs_its_real_producer_and_refuses_t
   }
 }
 
+// THE ROSTER IS READ BEFORE ANY STANDING IS JUDGED, WITH THE EXECUTOR SUPPLIED. The live entry
+// differs from the claim above only in where the executor's name and the instant come from (the
+// host's `hostname` and clock); both are supplied here, so the verdict does not depend on which
+// runner executes it. The executor is under the RFC 2606 reserved `.invalid` TLD, so no declared
+// dashboard instance can answer for it: the real read route refuses at its store step, once per
+// claimed group, and the verdict names that read for every group. Letting an unreadable log fall
+// back to the resting row, or dropping a group from the unread population, turns this red.
+data w_off_fleet_executor: String = "off-fleet.invalid"
+data w_supplied_instant: EpochSecs = 1788000000
+
+fn w_unread_at_store(c: CurrentAuthority) -> Bool {
+  match c {
+    CurrentAuthorityUnread { group: _, step: step, reason: why } => step == "store" && string_contains(s: why, pattern: w_off_fleet_executor)
+    CurrentAuthorityRead { authority: _, head: _, generation: _, grant: _, entry_state: _, previous: _, admitted_by: _, transitions: _ } => false
+  }
+}
+
+fn w_verdict_names_the_store_read(current: List<CurrentAuthority>) -> Bool {
+  match fabric_capacity_standing_current(current: current) {
+    ExitSuccess => false
+    ExitFailure { code: _, reason: reason } =>
+      string_contains(s: reason, pattern: "group-a: serving authority could not be read at store: ")
+  }
+}
+
+test fn the_verdict_reads_the_roster_first_and_names_an_unreadable_store_for_every_group() -> Bool {
+  let current = current_pair_serving_authorities(short_hostname: w_off_fleet_executor, at: w_supplied_instant)
+  length(current) > 0 && length(current) == length(spark_pair_serving_authorities)
+    && all(current, c => w_unread_at_store(c: c))
+    && w_verdict_names_the_store_read(current: current)
+}
+
 // A SUSPENDED GROUP PRODUCES A STANDING THAT DOES NOT ADMIT, AND A RELEASED ONE TOO -- but neither
 // disappears, which is the property the empty-roster guard depends on. If suspension had been modeled
 // by dropping the group from the population instead, the guard would see a shorter list rather than a

dag/test/claim/spark/spark_pair_serving_apply_witness_test.dag

@@ -1,20 +1,20 @@
 module test.claim.spark.spark_pair_serving_apply_witness
 
-import std.types { Bool, List, NonEmptyStr, String }
+import std.types { Bool, List, NonEmptyStr, String, EpochSecs }
 import product.placement_supply { host_identity_eq }
 import gunbc.fleet_intent_network { operator_host_srv5, operator_host_srv6, operator_host_srv7, operator_host_srv8 }
 import gunbc.spark.pair_serving_realization { spark_pair_realizations }
 import gunbc.spark.pair_serving_apply {
   SparkPairApplyPlan, SparkPairApplyPlanned, SparkPairApplyRefused, SparkPairHostApplyPlan,
-  spark_pair_apply_plan_current, spark_pair_apply_plans_current, spark_pair_apply_plan_refusals,
+  spark_pair_apply_plan_current, spark_pair_apply_plans_current, spark_pair_apply_plan_refusals, spark_pair_apply_plans,
 }
 import gunbc.spark.pair_serving_authority { spark_pair_serving_authorities, pair_serving_authority_group }
 import gunbc.spark.pair_serving_authority_log { CurrentAuthority, CurrentAuthorityRead, CurrentAuthorityUnread, AuthorityTransitionRecord }
 import product.capacity.event_chain { HeadAbsent }
 
 // THE CURRENT AUTHORITY IS SUPPLIED: the resting row of every claimed group at generation 0, which
-// is exactly what the log folds to when no transition has been recorded. The live read of the log
-// is executed by test.claim.spark.spark_pair_serving_apply_wet_witness on the local-repo wet lane.
+// is exactly what the log folds to when no transition has been recorded. The log READ is
+// exercised by the supplied-executor claim at the end of this file.
 fn w_current() -> List<CurrentAuthority> {
   map(spark_pair_serving_authorities, a => CurrentAuthorityRead { authority: a, head: HeadAbsent, generation: 0, grant: none, entry_state: none, previous: none, admitted_by: none, transitions: [] as List<AuthorityTransitionRecord> })
 }
@@ -90,3 +90,22 @@ test fn an_unread_authority_refuses_the_plan_with_the_reads_cause() -> Bool {
     && length(refusals) == length(plans)
     && all(refusals, r => string_contains(s: r, pattern: "the serving authority could not be read") && string_contains(s: r, pattern: "witness: the partition could not be fetched"))
 }
+
+// THE APPLY SEAM READS THE AUTHORITY BEFORE PLANNING, WITH THE EXECUTOR SUPPLIED. The live seam
+// takes the executor's name from the host's `hostname` and the instant from its clock; both are
+// supplied here so the verdict does not depend on which runner executes it. The executor is under
+// the RFC 2606 reserved `.invalid` TLD, so no declared dashboard instance answers for it and the
+// real read route refuses at its store step: every claimed group's plan is a refusal that names the
+// read, and nothing is planned from the resting row. Deleting the log read, or letting an
+// unreadable log fall back to the row, turns these refusals into plans and this claim red.
+data w_off_fleet_executor: String = "off-fleet.invalid"
+data w_supplied_instant: EpochSecs = 1788000000
+
+test fn the_apply_seam_reads_the_authority_before_planning_and_refuses_an_unreadable_store() -> Bool {
+  let plans = spark_pair_apply_plans(executor: w_off_fleet_executor, at: w_supplied_instant)
+  let refusals = spark_pair_apply_plan_refusals(plans: plans)
+  length(plans) > 0
+    && length(plans) == length(spark_pair_serving_authorities)
+    && length(refusals) == length(plans)
+    && all(refusals, r => string_contains(s: r, pattern: "the serving authority could not be read") && string_contains(s: r, pattern: w_off_fleet_executor))
+}

src/v2/workflow/local_repo_wet_terminal.dag

@@ -1045,18 +1045,6 @@ fn local_repo_wet_schedule() -> List<WetScheduledClaim> {
       function: "a_rerun_resumes_its_own_pending_state_and_restores_the_exact_active_by_real_execution",
       expectation: ExpectedToHold {}
     },
-    WetScheduledClaim {
-      identity: WitnessIdentity { module_path: "test.claim.spark.fabric_capacity_standing_wet_witness", function: "the_live_entry_reads_the_authority_before_judging_and_refuses_off_fleet" },
-      entry: "dag/test/claim/spark/fabric_capacity_standing_wet_witness_test.dag",
-      function: "the_live_entry_reads_the_authority_before_judging_and_refuses_off_fleet",
-      expectation: ExpectedToHold {}
-    },
-    WetScheduledClaim {
-      identity: WitnessIdentity { module_path: "test.claim.spark.spark_pair_serving_apply_wet_witness", function: "the_apply_seam_reads_the_authority_before_planning_and_refuses_off_fleet" },
-      entry: "dag/test/claim/spark/spark_pair_serving_apply_wet_witness_test.dag",
-      function: "the_apply_seam_reads_the_authority_before_planning_and_refuses_off_fleet",
-      expectation: ExpectedToHold {}
-    },
     WetScheduledClaim {
       identity: WitnessIdentity { module_path: "test.claim.spark.pair_serving_d0_real_execution", function: "a_fence_and_a_restore_each_reach_their_terminal_by_real_execution" },
       entry: "dag/test/claim/spark/pair_serving_d0_real_execution_witness_test.dag",