store: held-session op + O(1) named hold-slot reads

dag/extdeps/realization/materialization_store_local.dag

@@ -91,10 +91,10 @@ import std.durable_exclusive_hold {
 import gunbc.durable_exclusive_hold_file_store {
   file_hold_acquire_windowed, FileHoldAcquireOutcome, FileHoldAcquired, FileHoldOccupied, FileHoldAcquireLost,
   FileHoldAcquireStoreUnavailable, FileHoldSlotUndecodable, FileHoldKeyNotSlotAddressable,
-  file_hold_release_assess_windowed, FileHoldReleasePlanned, FileHoldReleaseNotEligible,
+  file_hold_release_assess_windowed_at, FileHoldReleasePlanned, FileHoldReleaseNotEligible,
   file_hold_release_commit, FileHoldReleased, FileHoldReleaseLost, FileHoldReleaseStoreUnavailable, FileHoldReleaseKeyNotSlotAddressable,
   file_hold_recovery_assess_windowed, FileHoldRecoveryPlanned, FileHoldRecoveryNotEligible,
-  observe_file_hold_windowed,
+  observe_file_hold_windowed_at,
 }
 import gunbc.process_hold_identity {
   process_hold_text, process_hold_marker, process_hold_decode, ProcessHoldHolder,
@@ -1033,6 +1033,20 @@ fn local_store_write_index_settled(held: LocalStoreFamilyHeld, expected: CasExpe
 // on this host, and only a holder observed dead is recovered through file_hold_recovery_assess. An
 // unobservable holder stays held -- a family that cannot be shown free is not written. The hold slot
 // is windowed like the occupancy index (k = 2), so its bookkeeping does not grow with commits.
+//
+// HOLD CHECK AND RELEASE READ THE SLOT BY NAME. The windowed verify step is: generation G present
+// and G+1 absent. LocalStoreFamilyHeld already names G, so those two reads are cas_file_slot_path
+// Filesystem.Read -- O(1) in the number of objects in the store root. That two-read proof is sound
+// only while a surviving eligible prefix cannot have holes: cas_reclaim_below deletes in generation
+// order and stops at the first host refusal, so it cannot remove G+1 while G stays. A successor
+// present still means the slot moved; an absent G is not this hold. Unknown-head observation
+// (acquire, index) still lists, as DESIGN §3b states for a window whose generation is not named.
+//
+// ACQUIRE STILL LISTS. Windowed presence is not a prefix: generations at or below head-k are
+// deleted, so a named gallop from generation 1 would read NotFound and treat a live head at 100 as
+// an empty slot, then create-new generation 1 beside it. Listing the root to discover this key's
+// generation files is the unknown-head observation cas_observe_window already uses. A held session
+// pays that list once, not once per commit.
 
 data local_store_hold_owner_prefix: String = "materialization-store"
 
@@ -1196,7 +1210,7 @@ fn local_store_hold_check(held: LocalStoreFamilyHeld) -> LocalStoreHoldCheck
     decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_commit_verified"),
   ]
 {
-  let observed = observe_file_hold_windowed(root: materialization_store_root_path(root: held.binding.root) as NonEmptyStr, key: local_store_hold_key(family: held.family), window: held.window)
+  let observed = observe_file_hold_windowed_at(root: materialization_store_root_path(root: held.binding.root) as NonEmptyStr, key: local_store_hold_key(family: held.family), generation: held.generation)
   match observed {
     HoldSlotHeld { generation: g, owner: o } =>
       if cas_generation_count(g: g) == cas_generation_count(g: held.generation) && durable_hold_owner_equal(left: o, right: held.owner) { LocalStoreHoldCurrent } else { LocalStoreHoldNotCurrent { observed: observed } }
@@ -1223,6 +1237,7 @@ fn local_store_with_family_of<T>(binding: LocalStoreBinding, family: ArtifactKin
   admit_callers: [
     decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_with_family"),
     decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_record_recency"),
+    decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_held_session"),
   ]
 {
   match local_store_acquire_family_hold(binding: binding, family: family, recovered: false) {
@@ -1234,6 +1249,25 @@ fn local_store_with_family_of<T>(binding: LocalStoreBinding, family: ArtifactKin
   }
 }
 
+// A HELD SESSION: acquire the family once, run many lookups and commits under LocalStoreFamilyHeld,
+// release once. Distinct from local_store_commit, which still brackets every write; this operation
+// does not change that signature. Sealed to gunbc.typecheck_module_store (and the inhabitance
+// witness that proves N commits acquire once).
+//
+// DEADLINE AND RENEWAL follow std.durable_exclusive_hold, not a clock. That protocol recovers a
+// hold only when its holder is OBSERVED dead -- there is no timestamp, duration or silent expiry.
+// This process is the holder (gunbc.process_hold_identity). Renewal is the named re-observation
+// local_store_hold_check performs before every mutation: the slot must still be held at the
+// generation and by the owner the session recorded, or the mutation refuses as stale. A session
+// whose process has died is recovered by the next writer's liveness observation, never by age.
+fn local_store_held_session<T>(capability: LocalStoreCapability, family: ArtifactKindId, run: fn(LocalStoreFamilyHeld) -> T) -> LocalStoreBracket<T>
+  admit_callers: [
+    decl_ref(module_path: "gunbc.typecheck_module_store", decl_name: "typecheck_module_store_session"),
+  ]
+{
+  local_store_with_family_of(binding: capability.binding, family: family, run: run)
+}
+
 // The release is reported, never fatal to the write it closes: a hold that could not be freed stays
 // held by this process, and once this process is gone the next writer recovers it by observation.
 type LocalStoreHoldRelease
@@ -1250,7 +1284,7 @@ fn local_store_release_family_hold(held: LocalStoreFamilyHeld) -> LocalStoreHold
     decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_acquire_all"),
   ]
 {
-  match file_hold_release_assess_windowed(root: materialization_store_root_path(root: held.binding.root) as NonEmptyStr, slot_key: local_store_hold_key(family: held.family), owner: held.owner, acquired_generation: held.generation, released_by: local_store_hold_release_ref, window: held.window) {
+  match file_hold_release_assess_windowed_at(root: materialization_store_root_path(root: held.binding.root) as NonEmptyStr, slot_key: local_store_hold_key(family: held.family), owner: held.owner, acquired_generation: held.generation, released_by: local_store_hold_release_ref, window: held.window) {
     FileHoldReleaseNotEligible { slot_key: _, refusal: r } => LocalStoreHoldReleaseNotEligible { refusal: r }
     FileHoldReleasePlanned { plan: pl } =>
       match file_hold_release_commit(plan: pl) {

dag/gunbc/durable_cas_file_store.dag

@@ -491,7 +491,7 @@ fn cas_key_generation_of_name(key: NonEmptyStr, name: String) -> CasGeneration?
 fn cas_window_generations(root: NonEmptyStr, key: NonEmptyStr) -> CasWindowGenerations
   admit_callers: [
     decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_settle_slot"),
-    decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "cas_observe_window"),
+    decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "cas_observe_window_once"),
     decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "cas_reclaim_below"),
   ]
 {
@@ -516,37 +516,52 @@ fn cas_window_max_generation(generations: List<CasGeneration>) -> CasGeneration?
   })
 }
 
-fn cas_observe_window(root: NonEmptyStr, key: NonEmptyStr, attempt: Int) -> CasSlotProbe
+data cas_window_observation_attempts: List<Int> = [1, 2, 3, 4, 5, 6, 7, 8]
+
+// ONE LIST-AND-VERIFY PASS. Recursion here was the unbounded arm: a window whose listed max is
+// strictly below a generation that still reads present (stale listing, or a TCO lowering that does
+// not increment `attempt`) re-entered cas_observe_window forever, re-reading G and G+1 with no
+// further List. A fold over a finite attempt roster cannot fail to stop.
+fn cas_observe_window_once(root: NonEmptyStr, key: NonEmptyStr) -> CasSlotProbe?
   admit_callers: [
-    decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "cas_observe_retained_slot"),
     decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "cas_observe_window"),
   ]
 {
-  if attempt > cas_window_observation_attempt_bound {
-    ProbedWindowHeadUnsettled { attempts: cas_window_observation_attempt_bound }
-  } else {
-    match cas_window_generations(root: root, key: key) {
-      CasWindowGenerationsUnlisted { cause: c } => ProbedAbsenceUnestablished { cause: c }
-      CasWindowGenerationsListed { generations: gs } =>
-        match cas_window_max_generation(generations: gs) {
-          Absent => ProbedAbsent
-          Present { value: head } =>
-            match cas_read_generation(root: root, key: key, generation: head) {
-              GenerationRefused { probe: p } => p
-              GenerationAbsent => cas_observe_window(root: root, key: key, attempt: attempt + 1)
-              GenerationPresent { content: c } =>
-                match cas_generation_successor(g: head) {
-                  CasSuccessorExhausted { head: _ } => ProbedHead { generation: head, value: c }
-                  CasSuccessorGeneration { generation: next } =>
-                    match cas_read_generation(root: root, key: key, generation: next) {
-                      GenerationRefused { probe: p } => p
-                      GenerationAbsent => ProbedHead { generation: head, value: c }
-                      GenerationPresent { content: _ } => cas_observe_window(root: root, key: key, attempt: attempt + 1)
-                    }
-                }
-            }
-        }
-    }
+  match cas_window_generations(root: root, key: key) {
+    CasWindowGenerationsUnlisted { cause: c } => Present { value: ProbedAbsenceUnestablished { cause: c } }
+    CasWindowGenerationsListed { generations: gs } =>
+      match cas_window_max_generation(generations: gs) {
+        Absent => Present { value: ProbedAbsent }
+        Present { value: head } =>
+          match cas_read_generation(root: root, key: key, generation: head) {
+            GenerationRefused { probe: p } => Present { value: p }
+            GenerationAbsent => none
+            GenerationPresent { content: c } =>
+              match cas_generation_successor(g: head) {
+                CasSuccessorExhausted { head: _ } => Present { value: ProbedHead { generation: head, value: c } }
+                CasSuccessorGeneration { generation: next } =>
+                  match cas_read_generation(root: root, key: key, generation: next) {
+                    GenerationRefused { probe: p } => Present { value: p }
+                    GenerationAbsent => Present { value: ProbedHead { generation: head, value: c } }
+                    GenerationPresent { content: _ } => none
+                  }
+              }
+          }
+      }
+  }
+}
+
+fn cas_observe_window(root: NonEmptyStr, key: NonEmptyStr) -> CasSlotProbe
+  admit_callers: [
+    decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "cas_observe_retained_slot"),
+  ]
+{
+  match fold(cas_window_observation_attempts, init: none, f: (acc, _) => match acc {
+    Present { value: p } => Present { value: p }
+    Absent => cas_observe_window_once(root: root, key: key)
+  }) {
+    Present { value: p } => p
+    Absent => ProbedWindowHeadUnsettled { attempts: cas_window_observation_attempt_bound }
   }
 }
 
@@ -560,7 +575,7 @@ fn cas_observe_retained_slot(root: NonEmptyStr, key: NonEmptyStr, retention: Cas
 {
   match retention {
     KeepAllGenerations => cas_observe_slot(root: root, key: key)
-    KeepGenerationWindow { window: _ } => cas_observe_window(root: root, key: key, attempt: 1)
+    KeepGenerationWindow { window: _ } => cas_observe_window(root: root, key: key)
   }
 }
 
@@ -677,11 +692,59 @@ fn observe_cas_slot_state_windowed(root: NonEmptyStr, key: NonEmptyStr, window:
   admit_callers: [
     decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_read_index"),
     decl_ref(module_path: "test.claim.durable_cas_file_store_wet_witness", decl_name: "a_window_slot_holds_at_most_k_plus_one_generations_by_real_execution"),
+    decl_ref(module_path: "test.claim.durable_cas_file_store_wet_witness", decl_name: "a_window_observation_at_generation_1500_terminates_by_real_execution"),
   ]
 {
   observe_cas_slot_state_retained(root: root, key: key, retention: KeepGenerationWindow { window: window })
 }
 
+// A WINDOWED HEAD READ BY CONSTRUCTED NAME, NOT BY LISTING THE SLOT ROOT.
+//
+// cas_observe_window finds the head by listing every name under the store root and keeping this
+// key's generation files. That is the correct discovery when the generation is unknown, because a
+// windowed slot is NOT a presence prefix: generations at or below head-k are deleted, so galloping
+// from generation 1 would read NotFound and report an empty slot while the live head sat at 100.
+// Listing is therefore retained for unknown-head observation (cas_observe_window).
+//
+// When the caller already names the generation it believes is head -- a hold it just acquired, a
+// release at the generation the protocol recorded -- the existing correctness argument does not
+// need the listing. The windowed contract's verify step is: the named generation reads present and
+// its successor reads absent. Those two reads are Filesystem.Read of cas_file_slot_path, O(1) in
+// the store's object count. A successor that is present means the head moved; an absent named
+// generation is not this head. The observation is never guessed.
+fn cas_verify_named_window_head(root: NonEmptyStr, key: NonEmptyStr, generation: CasGeneration) -> CasSlotProbe
+  admit_callers: [
+    decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "observe_cas_slot_state_windowed_at"),
+  ]
+{
+  match cas_read_generation(root: root, key: key, generation: generation) {
+    GenerationRefused { probe: p } => p
+    GenerationAbsent => ProbedAbsent
+    GenerationPresent { content: c } =>
+      match cas_generation_successor(g: generation) {
+        CasSuccessorExhausted { head: _ } => ProbedHead { generation: generation, value: c }
+        CasSuccessorGeneration { generation: next } =>
+          match cas_read_generation(root: root, key: key, generation: next) {
+            GenerationRefused { probe: p } => p
+            GenerationAbsent => ProbedHead { generation: generation, value: c }
+            GenerationPresent { content: _ } => ProbedWindowHeadUnsettled { attempts: 1 }
+          }
+      }
+  }
+}
+
+fn observe_cas_slot_state_windowed_at(root: NonEmptyStr, key: NonEmptyStr, generation: CasGeneration) -> CasSlotObservation<NonEmptyStr>
+  admit_callers: [
+    decl_ref(module_path: "gunbc.durable_exclusive_hold_file_store", decl_name: "observe_file_hold_windowed_at"),
+  ]
+{
+  if !cas_key_is_slot_addressable(key: key) {
+    CasObservedUnreadable { cause: CasUnreadableReadRefused { detail: cas_key_not_slot_addressable_detail } }
+  } else {
+    cas_probe_as_observation(probe: cas_verify_named_window_head(root: root, key: key, generation: generation))
+  }
+}
+
 data cas_key_not_slot_addressable_detail: NonEmptyStr = "key is not slot-addressable: it carries a path separator"
 
 
@@ -1035,23 +1098,44 @@ type CasFileWrite {
   reclamation: CasWindowReclamation
 }
 
+type CasReclaimProgress {
+  removed: List<CasGeneration>
+  failed: List<CasGenerationReclaimFailure>
+  stopped: Bool
+}
+
+// ASCENDING, AND STOP AT THE FIRST HOST REFUSAL. Independent deletes of every generation at or
+// below the floor could remove G+1 while G survived, which punches a hole: named window verify
+// (G present and G+1 absent) would then report G as head while a later generation was live.
+// One sort_by of the eligible set, then one walk that leaves every later generation in place when
+// a delete fails, keeps that prefix contiguous without a min-and-filter per step.
 fn cas_reclaim_below(root: NonEmptyStr, key: NonEmptyStr, window: CasGenerationWindow, head: CasGeneration) -> CasWindowReclamation
   admit_callers: [
     decl_ref(module_path: "gunbc.durable_cas_file_store", decl_name: "file_compare_and_set_retained"),
     decl_ref(module_path: "extdeps.realization.materialization_store_local", decl_name: "local_store_settle_slot"),
+    decl_ref(module_path: "test.claim.durable_cas_file_store_wet_witness", decl_name: "reclaim_stops_before_a_later_generation_when_an_earlier_delete_fails_by_real_execution"),
   ]
 {
   match cas_window_generations(root: root, key: key) {
     CasWindowGenerationsUnlisted { cause: c } => CasReclamationUnlisted { cause: c }
     CasWindowGenerationsListed { generations: gs } => {
       let floor = cas_generation_count(g: head) - window.k
-      let attempts = gs |> filter(g => cas_generation_count(g: g) <= floor) |> map(g => {
-        let d = Filesystem.Delete(path: cas_file_slot_path(root: root, key: key, generation: g))
-        CasGenerationDeleteAttempt { generation: g, success: d.success, error: d.error }
-      })
-      let removed = attempts |> filter(t => t.success) |> map(t => t.generation)
-      let failed = attempts |> filter(t => !t.success) |> map(t => CasGenerationReclaimFailure { generation: t.generation, host_error: t.error })
-      if count(failed) == 0 { CasReclaimed { removed: removed } } else { CasReclamationIncomplete { removed: removed, failed: failed } }
+      let ordered = gs |> filter(g => cas_generation_count(g: g) <= floor) |> sort_by(g => cas_generation_count(g: g))
+      let walk = fold(ordered, init: CasReclaimProgress { removed: [], failed: [], stopped: false }, f: (acc, g) =>
+        if acc.stopped { acc } else {
+          let d = Filesystem.Delete(path: cas_file_slot_path(root: root, key: key, generation: g))
+          if d.success {
+            CasReclaimProgress { removed: acc.removed |> list_push(g), failed: acc.failed, stopped: false }
+          } else {
+            CasReclaimProgress {
+              removed: acc.removed,
+              failed: [CasGenerationReclaimFailure { generation: g, host_error: d.error }],
+              stopped: true,
+            }
+          }
+        }
+      )
+      if count(walk.failed) == 0 { CasReclaimed { removed: walk.removed } } else { CasReclamationIncomplete { removed: walk.removed, failed: walk.failed } }
     }
   }
 }