spark wet witnesses: supply the off-fleet premise; real-read route claims (unblocks srv1-placed floors)

dag/gunbc/instruments/fabric_capacity_standing.dag

@@ -576,6 +576,8 @@ fn capacity_standing_verdict(standings: List<GroupStanding>) -> ProcessExit {
   }
 }
 
+data fabric_capacity_unread_marker: String = ": serving authority could not be read at "
+
 // 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
@@ -585,7 +587,7 @@ fn standing_for_current(c: CurrentAuthority) -> GroupStanding {
   match c {
     CurrentAuthorityRead { authority: a, head: _, generation: _, grant: _, entry_state: _, previous: _, admitted_by: _, transitions: _ } => 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], ""))
+      refused_standing(text: join([fabric_group_wire(g: g) as String, fabric_capacity_unread_marker, s, ": ", why], ""))
   }
 }
 
@@ -603,7 +605,15 @@ fn fabric_capacity_standing() -> ProcessExit
     ExecutorReachKnown { short_hostname: executor, path: _ } =>
       match now_epoch_seconds() {
         Absent => exit_failure(reason: "fabric capacity standing: the clock could not be read as epoch seconds, so the lease standing of the authority cannot be derived")
-        Present { value: at } => fabric_capacity_standing_current(current: current_pair_serving_authorities(short_hostname: executor, at: at))
+        Present { value: at } => fabric_capacity_standing_for(executor: executor, at: at)
       }
   }
 }
+
+// THE ENTRY BELOW ITS TWO OBSERVATIONS: the executor and the instant are the only facts the live
+// entry reads from the host before the log, so this is the seam at which a witness supplies them.
+// A supplied host that declares no dashboard instance reaches the real store resolution and the
+// real per-group refusal on every runner, instead of depending on which host the claim landed on.
+fn fabric_capacity_standing_for(executor: String, at: EpochSecs) -> ProcessExit {
+  fabric_capacity_standing_current(current: current_pair_serving_authorities(short_hostname: executor, at: at))
+}

dag/gunbc/spark/pair_serving_apply.dag

@@ -197,6 +197,8 @@ fn spark_pair_apply_plan_under(r: SparkPairRealization, authority: PairServingGr
   }
 }
 
+data spark_pair_apply_unread_refusal_prefix: String = "the serving authority could not be read, so nothing is converged: "
+
 // THE LIVE AUTHORITY IS THE LOG'S, NOT THE SOURCE ROW'S. gunbc.spark.pair_serving_authority_log
 // folds the row through every recorded transition, so a group a transaction has suspended is
 // suspended HERE, from the transaction's own tree. A group whose authority could not be read is
@@ -208,7 +210,7 @@ fn spark_pair_apply_plan_current(r: SparkPairRealization, current: List<CurrentA
     Present { value: c } =>
       match c {
         CurrentAuthorityUnread { group: _, step: _, reason: _ } =>
-          SparkPairApplyRefused { cause: join(["the serving authority could not be read, so nothing is converged: ", current_authority_wire(c: c)], "") }
+          SparkPairApplyRefused { cause: join([spark_pair_apply_unread_refusal_prefix, current_authority_wire(c: c)], "") }
         CurrentAuthorityRead { authority: a, head: _, generation: _, grant: _, entry_state: _, previous: _, admitted_by: _, transitions: _ } =>
           spark_pair_apply_plan_under(r: r, authority: Present { value: a })
       }

dag/gunbc/spark/pair_serving_authority_log.dag

@@ -2258,10 +2258,12 @@ fn pair_serving_authority_record_entry_state(store: FabricStorageBinding, group:
   }
 }
 
+data current_authority_unread_marker: String = " authority unread at "
+
 fn current_authority_wire(c: CurrentAuthority) -> String {
   match c {
     CurrentAuthorityRead { authority: a, head: _, generation: g, grant: _, entry_state: _, previous: _, admitted_by: _, transitions: _ } => join(["generation ", to_string(g), ": ", authority_wire(a: a)], "")
-    CurrentAuthorityUnread { group: g, step: s, reason: why } => join([fabric_group_wire(g: g) as String, " authority unread at ", s, ": ", why], "")
+    CurrentAuthorityUnread { group: g, step: s, reason: why } => join([fabric_group_wire(g: g) as String, current_authority_unread_marker, s, ": ", why], "")
   }
 }
 

dag/test/claim/spark/fabric_capacity_standing_wet_witness_test.dag

@@ -1,27 +1,78 @@
 module test.claim.spark.fabric_capacity_standing_wet_witness
 
 import std.logic { Bool }
-import std.types { String }
+import std.types { String, List, EpochSecs }
+import v2.std.optional { Present, Absent }
 import v2.std.live_tree { LiveTreeDisposition, SubstrateInputsOnly }
-import std.process { ExitSuccess, ExitFailure }
-import gunbc.instruments.fabric_capacity_standing { fabric_capacity_standing }
+import std.process { ProcessExit, ExitSuccess, ExitFailure }
+import gunbc.fabric_event_log_host { now_epoch_seconds }
+import gunbc.spark.fabric_reach { ExecutorReachKnown, ExecutorReachUnknown, observe_executor_reach }
+import gunbc.spark.fabric_switch_observed { fabric_group_wire }
+import gunbc.spark.pair_serving_authority { spark_pair_serving_authorities }
+import gunbc.spark.pair_serving_authority_log {
+  CurrentAuthorityRead, CurrentAuthorityUnread,
+  current_pair_serving_authorities, current_authority_group,
+}
+import gunbc.instruments.fabric_capacity_standing { fabric_capacity_standing, fabric_capacity_standing_for, fabric_capacity_standing_current, fabric_capacity_unread_marker }
 
 data live_tree_disposition: LiveTreeDisposition = SubstrateInputsOnly
 
-// THE LIVE ENTRY, EXECUTED. test.claim.spark.fabric_capacity_standing_witness drives the verdict
-// over a SUPPLIED roster so the real group_report producer and its binding run hermetically; this
-// file is the other half of DESIGN §3's pairing obligation, the one that cannot be hermetic: the
-// entry reads the executor's hostname (shell.Exec.RunArgv, no mock_response), resolves its event
-// log layout and reads every claimed group's authority from the log before any standing is
-// judged. On the wet lane's runner that read does not reach a log -- the runner is no dashboard
-// host, or its remote is unreachable -- so the honest verdict is a refusal that NAMES THE READ:
-// every claimed group is reported "serving authority could not be read at <step>", and no
-// group's capacity is stood for. That is the route: deleting the live binding, or letting an
-// unreadable log fall back to the source row, changes this text.
+// THE OFF-FLEET PREMISE IS SUPPLIED, NOT READ FROM THE RUNNER. This claim used to take the executor
+// from the runner's own hostname and assume that host declared no dashboard instance. That is true
+// on srv3/srv4 runners and false on srv1 ones, where the log is readable, so the claim went red by
+// runner placement (#12444 floor runs 36355617646 and 36361351081, both on srv1). The premise is
+// now a supplied value at the entry's seam, fabric_capacity_standing_for. Everything beneath it is
+// real on every runner: store resolution for that host, the per-group CurrentAuthorityUnread, and
+// the verdict text. test.claim.spark.fabric_capacity_standing_witness drives the READ arm over a
+// supplied roster; the real read on the actual runner is the next claim's subject.
+data witness_host_declaring_no_dashboard_instance: String = "gunbc-witness-host-without-dashboard-instance"
+data witness_instant: EpochSecs = 1000
+
 test fn the_live_entry_reads_the_authority_before_judging_and_refuses_off_fleet() -> Bool {
-  match fabric_capacity_standing() {
+  match fabric_capacity_standing_for(executor: witness_host_declaring_no_dashboard_instance, at: witness_instant) {
     ExitSuccess => false
     ExitFailure { code: _, reason: reason } =>
-      string_contains(s: reason, pattern: "group-a: serving authority could not be read at ")
+      string_contains(s: reason, pattern: "group-a: serving authority could not be read at store: no dashboard instance is declared for host gunbc-witness-host-without-dashboard-instance")
+  }
+}
+
+fn exit_reason(e: ProcessExit) -> String {
+  match e {
+    ExitSuccess => ""
+    ExitFailure { code: _, reason: r } => r as String
+  }
+}
+
+// THE REAL READ, ON WHATEVER RUNNER THIS LANDS ON: DESIGN §3's pairing obligation for the seam
+// above. The live entry is run end to end (hostname, clock, store, log), and the log is read again
+// for the same host. Two things are asserted.
+// - The entry's verdict EQUALS the verdict over the readings the log returned. An entry that judged
+//   a group from its resting row, or skipped the read, disagrees whenever the log's state for that
+//   group differs from the row.
+// - Per group, each unread reading appears in the verdict with its step, and no read group is
+//   reported unread.
+// ITS CEILING, stated rather than implied (DESIGN §4d): on a runner that can read the log, a group
+// whose logged authority still equals its resting row (generation 0, no transition) cannot tell a
+// row fallback from a real read, because both give the same answer. The supplied-roster witness
+// test.claim.spark.fabric_capacity_standing_witness drives the read arm with differing authorities.
+test fn the_live_entry_judges_exactly_what_the_runners_log_read_returned() -> Bool {
+  match observe_executor_reach() {
+    ExecutorReachUnknown { cause: _ } => false
+    ExecutorReachKnown { short_hostname: executor, path: _ } =>
+      match now_epoch_seconds() {
+        Absent => false
+        Present { value: at } => {
+          let current = current_pair_serving_authorities(short_hostname: executor, at: at)
+          let reason = exit_reason(e: fabric_capacity_standing())
+          length(current) == length(spark_pair_serving_authorities)
+            && reason == exit_reason(e: fabric_capacity_standing_current(current: current))
+            && all(current, c => match c {
+              CurrentAuthorityUnread { group: g, step: s, reason: _ } =>
+                string_contains(s: reason, pattern: join([fabric_group_wire(g: g) as String, fabric_capacity_unread_marker, s], ""))
+              CurrentAuthorityRead { authority: _, head: _, generation: _, grant: _, entry_state: _, previous: _, admitted_by: _, transitions: _ } =>
+                !string_contains(s: reason, pattern: join([fabric_group_wire(g: current_authority_group(c: c)) as String, fabric_capacity_unread_marker], ""))
+            })
+        }
+      }
   }
 }

dag/test/claim/spark/spark_pair_serving_apply_wet_witness_test.dag

@@ -1,36 +1,67 @@
 module test.claim.spark.spark_pair_serving_apply_wet_witness
 
 import std.logic { Bool }
-import std.types { String, List }
+import std.types { String, List, EpochSecs }
 import v2.std.optional { Present, Absent }
 import v2.std.live_tree { LiveTreeDisposition, SubstrateInputsOnly }
 import gunbc.fabric_event_log_host { now_epoch_seconds }
 import gunbc.spark.fabric_reach { ExecutorReachKnown, ExecutorReachUnknown, observe_executor_reach }
-import gunbc.spark.pair_serving_apply { spark_pair_apply_plans, spark_pair_apply_plan_refusals }
+import gunbc.spark.pair_serving_apply { spark_pair_apply_plans, spark_pair_apply_plan_refusals, spark_pair_apply_unread_refusal_prefix }
 import gunbc.spark.pair_serving_authority { spark_pair_serving_authorities }
+import gunbc.spark.pair_serving_authority_log {
+  CurrentAuthorityRead, CurrentAuthorityUnread,
+  current_pair_serving_authorities, current_authority_group, current_authority_wire, current_authority_unread_marker,
+}
+import gunbc.spark.fabric_switch_observed { fabric_group_wire }
 
 data live_tree_disposition: LiveTreeDisposition = SubstrateInputsOnly
 
-// THE APPLY SEAM'S LIVE BINDING, EXECUTED. test.claim.spark.spark_pair_serving_apply_witness drives
-// the plan fold over a SUPPLIED roster; this is the other half of DESIGN §3's pairing obligation:
-// spark_pair_apply_plans reads the executor's hostname, resolves its event log layout and reads
-// every claimed group's authority from the log before planning anything. On the wet lane's
-// runner that read does not reach a log, so every claimed group's plan is a refusal that NAMES
-// THE READ -- "the serving authority could not be 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.
+// THE OFF-FLEET PREMISE IS SUPPLIED, NOT READ FROM THE RUNNER. test.claim.spark.spark_pair_serving_apply_witness
+// drives the plan fold over a SUPPLIED roster. This file is the other half of DESIGN §3's pairing
+// obligation: spark_pair_apply_plans resolves the executor's event log layout and reads every
+// claimed group's authority from the log before planning anything. This claim once took the
+// executor from the runner's hostname and assumed it declared no dashboard instance. That is false
+// on srv1 runners, so the claim went red by placement (#12444 floor runs 36355617646 and
+// 36361351081). The host is now supplied. The store resolution and the refusal that NAMES THE READ
+// run for real on every runner, 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 witness_host_declaring_no_dashboard_instance: String = "gunbc-witness-host-without-dashboard-instance"
+data witness_instant: EpochSecs = 1000
+
 test fn the_apply_seam_reads_the_authority_before_planning_and_refuses_off_fleet() -> Bool {
+  let plans = spark_pair_apply_plans(executor: witness_host_declaring_no_dashboard_instance, at: witness_instant)
+  let refusals = spark_pair_apply_plan_refusals(plans: plans)
+  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"))
+}
+
+// THE REAL READ, ON WHATEVER RUNNER THIS LANDS ON: the pairing claim for the supplied host above.
+// The executor is the runner's own hostname and the log is really read. The ROUTE holds whether or
+// not the runner can read the log. It is an IDENTITY JOIN, not a count (DESIGN section 5: completeness is an
+// identity join): each unread reading is joined to the one refusal that carries exactly that
+// reading's wire (group, step, cause), and no group the log did answer for appears in any unread
+// refusal. So an unreadable authority never becomes a plan, a readable one is never reported
+// unreadable, and one group's refusal cannot stand in for another's.
+test fn the_apply_seam_refuses_exactly_what_the_runners_log_read_could_not_answer() -> Bool {
   match observe_executor_reach() {
     ExecutorReachUnknown { cause: _ } => false
     ExecutorReachKnown { short_hostname: executor, path: _ } =>
       match now_epoch_seconds() {
         Absent => false
         Present { value: at } => {
+          let current = current_pair_serving_authorities(short_hostname: executor, at: at)
           let plans = spark_pair_apply_plans(executor: executor, at: at)
           let refusals = spark_pair_apply_plan_refusals(plans: plans)
           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"))
+            && length(current) == length(spark_pair_serving_authorities)
+            && all(current, c => match c {
+              CurrentAuthorityUnread { group: _, step: _, reason: _ } =>
+                any(refusals, r => r == join([spark_pair_apply_unread_refusal_prefix, current_authority_wire(c: c)], ""))
+              CurrentAuthorityRead { authority: _, head: _, generation: _, grant: _, entry_state: _, previous: _, admitted_by: _, transitions: _ } =>
+                !any(refusals, r => string_contains(s: r, pattern: join([fabric_group_wire(g: current_authority_group(c: c)) as String, current_authority_unread_marker], "")))
+            })
         }
       }
   }

src/v2/workflow/floor_route_gap.dag

@@ -1686,6 +1686,10 @@ fn floor_route_gap_expectation_chunk_24() -> List<FloorRouteGapExpectation> {
 // expectations, their WetScheduledClaim rows in local_repo_wet_schedule, and the gunbc.ci_layer_roots
 // LocalRepoWetLane rows for approval_device_enrolment_code_wet_witness_test.dag and
 // approval_ntfy_access_readback_wet_witness_test.dag.
+// The last two rows are the spark real-read route claims from #12478: each reads the runner's
+// hostname through shell.Exec.RunArgv, which has no published mock case. Floor run 36377059685
+// measured both gaps. They are enrolled on the same three terms: these rows, their WetScheduledClaim rows,
+// and the existing LocalRepoWetLane rows for their two files.
 fn floor_route_gap_expectation_chunk_25() -> List<FloorRouteGapExpectation> {
   Cons {
     head: FloorRouteGapExpectation { identity: "test.claim.approval_device_enrolment_code_wet_witness_test.the_code_slot_is_written_before_delivery_by_real_execution", operation: "DirWithTemplate", ground: NoMockResponse {} },
@@ -1695,7 +1699,11 @@ fn floor_route_gap_expectation_chunk_25() -> List<FloorRouteGapExpectation> {
     head: FloorRouteGapExpectation { identity: "test.claim.approval_ntfy_access_readback_wet_witness_test.the_root_refuses_before_minting_on_this_host_by_real_execution", operation: "shell.Exec.RunArgv", ground: UnpublishedMockCase {} },
     tail: Cons {
     head: FloorRouteGapExpectation { identity: "test.claim.approval_ntfy_access_readback_wet_witness_test.the_helper_output_decodes_as_an_outcome_by_real_execution", operation: "shell.Exec.RunArgv", ground: UnpublishedMockCase {} },
-    tail: Empty {} } } } }
+    tail: Cons {
+    head: FloorRouteGapExpectation { identity: "test.claim.spark.fabric_capacity_standing_wet_witness.the_live_entry_judges_exactly_what_the_runners_log_read_returned", operation: "shell.Exec.RunArgv", ground: UnpublishedMockCase {} },
+    tail: Cons {
+    head: FloorRouteGapExpectation { identity: "test.claim.spark.spark_pair_serving_apply_wet_witness.the_apply_seam_refuses_exactly_what_the_runners_log_read_could_not_answer", operation: "shell.Exec.RunArgv", ground: UnpublishedMockCase {} },
+    tail: Empty {} } } } } } }
 }
 
 fn floor_route_gap_expectation_chunks() -> List<List<FloorRouteGapExpectation>> {

src/v2/workflow/local_repo_wet_terminal.dag

@@ -1057,6 +1057,18 @@ fn local_repo_wet_schedule() -> List<WetScheduledClaim> {
       function: "the_apply_seam_reads_the_authority_before_planning_and_refuses_off_fleet",
       expectation: ExpectedToHold {}
     },
+    WetScheduledClaim {
+      identity: WitnessIdentity { module_path: "test.claim.spark.fabric_capacity_standing_wet_witness", function: "the_live_entry_judges_exactly_what_the_runners_log_read_returned" },
+      entry: "dag/test/claim/spark/fabric_capacity_standing_wet_witness_test.dag",
+      function: "the_live_entry_judges_exactly_what_the_runners_log_read_returned",
+      expectation: ExpectedToHold {}
+    },
+    WetScheduledClaim {
+      identity: WitnessIdentity { module_path: "test.claim.spark.spark_pair_serving_apply_wet_witness", function: "the_apply_seam_refuses_exactly_what_the_runners_log_read_could_not_answer" },
+      entry: "dag/test/claim/spark/spark_pair_serving_apply_wet_witness_test.dag",
+      function: "the_apply_seam_refuses_exactly_what_the_runners_log_read_could_not_answer",
+      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",