fix(scheduling): close seven ledger and authoring follow-ups from the MVP review

crates/registry-scheduling-client/src/client.rs

@@ -11,13 +11,14 @@ use registry_platform_httputil::{
     read_bounded, url::append_path_segments, validate_response_headers,
 };
 use registry_scheduling_core::{
-    type_uri, AdmissionRequest, AppointmentDocument, AppointmentHistoryEntryDocument,
-    AvailabilityEntry, CancelAppointmentRequest, CreateAppointmentRequest, ExplainDocument,
-    HoldDocument, LocationDocument, OfferingDocument, PageDocument, ProblemCode,
-    RescheduleAppointmentRequest, ResourceDocument, SchedulingServiceDocument, ServiceDocument,
-    APPOINTMENTS_PATH, AVAILABILITY_EXPLAIN_PATH, AVAILABILITY_PATH, CURSOR_QUERY_PARAMETER,
-    HOLDS_PATH, IDEMPOTENCY_KEY_HEADER, LIMIT_QUERY_PARAMETER, LOCATIONS_PATH,
-    MAXIMUM_IDEMPOTENCY_KEY_BYTES, OFFERINGS_PATH, RESOURCES_PATH, SCHEDULING_PATH, SERVICES_PATH,
+    type_uri, valid_identifier, AdmissionRequest, AppointmentDocument,
+    AppointmentHistoryEntryDocument, AvailabilityEntry, CancelAppointmentRequest,
+    CreateAppointmentRequest, ExplainDocument, HoldDocument, LocationDocument, OfferingDocument,
+    PageDocument, ProblemCode, RescheduleAppointmentRequest, ResourceDocument,
+    SchedulingServiceDocument, ServiceDocument, APPOINTMENTS_PATH, AVAILABILITY_EXPLAIN_PATH,
+    AVAILABILITY_PATH, CURSOR_QUERY_PARAMETER, HOLDS_PATH, IDEMPOTENCY_KEY_HEADER,
+    LIMIT_QUERY_PARAMETER, LOCATIONS_PATH, MAXIMUM_IDEMPOTENCY_KEY_BYTES, OFFERINGS_PATH,
+    RESOURCES_PATH, SCHEDULING_PATH, SERVICES_PATH,
 };
 use reqwest::header::{HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE};
 use reqwest::{RequestBuilder, Response, StatusCode, Url};
@@ -142,11 +143,7 @@ impl SchedulingClient {
         offering: &str,
         start: DateTime<Utc>,
     ) -> Result<SchedulingComplete<ExplainDocument>, SchedulingClientError> {
-        if offering.is_empty() {
-            return Err(SchedulingClientError::invalid_request(
-                "the offering selector is invalid",
-            ));
-        }
+        validate_offering(offering)?;
         let query = [
             (OFFERING_QUERY_PARAMETER, offering.to_owned()),
             (START_QUERY_PARAMETER, rfc3339(start)),
@@ -564,18 +561,30 @@ fn validate_identifier(
     Ok(())
 }
 
+/// The offering selector both availability routes carry. It travels as a
+/// query parameter rather than a path segment, so an out-of-grammar value
+/// cannot reach a different route; what it would otherwise buy is a round
+/// trip spent sending a URL the runtime's own grammar already refuses. An
+/// offering is authored in a policy rather than minted into a document, so
+/// the grammar it is held to here is the one the runtime publishes offerings
+/// under, not the wider alphabet an opaque route identifier travels in.
+fn validate_offering(offering: &str) -> Result<(), SchedulingClientError> {
+    if valid_identifier(offering) {
+        return Ok(());
+    }
+    Err(SchedulingClientError::invalid_request(
+        "the offering selector is invalid",
+    ))
+}
+
 fn validate_availability(
     offering: &str,
     _start: Option<DateTime<Utc>>,
     _end: Option<DateTime<Utc>>,
     cursor: Option<&str>,
     limit: Option<u32>,
 ) -> Result<(), SchedulingClientError> {
-    if offering.is_empty() {
-        return Err(SchedulingClientError::invalid_request(
-            "the offering selector is invalid",
-        ));
-    }
+    validate_offering(offering)?;
     validate_cursor(cursor)?;
     if limit.is_some_and(|value| value == 0) {
         return Err(SchedulingClientError::invalid_request(
@@ -648,6 +657,38 @@ mod tests {
         assert!(validate_availability("registry-update-30", None, None, None, Some(0)).is_err());
     }
 
+    /// Both availability entry points refuse an out-of-grammar selector
+    /// before any I/O, against the grammar the runtime publishes offerings
+    /// under rather than the looser one a route segment carries: a selector
+    /// the runtime would refuse on arrival is refused here instead.
+    #[test]
+    fn offering_selectors_are_bounded_before_io() {
+        assert!(validate_offering("registry-update-30").is_ok());
+        for foreign in [
+            "",
+            "registry/update",
+            "registry update",
+            "registry?update",
+            // Each of these is a legal opaque route segment and no
+            // identifier the policy grammar admits.
+            "Registry_Update",
+            "registry_update",
+            "registry.update",
+            "30-minute-update",
+            &"x".repeat(65),
+            &"x".repeat(MAXIMUM_IDENTIFIER_BYTES + 1),
+        ] {
+            assert!(
+                validate_offering(foreign).is_err(),
+                "{foreign:?} must be refused"
+            );
+            assert!(
+                validate_availability(foreign, None, None, None, None).is_err(),
+                "{foreign:?} must be refused by availability"
+            );
+        }
+    }
+
     #[test]
     fn availability_ranges_preserve_runtime_normalization() {
         let start = Utc.with_ymd_and_hms(2026, 10, 5, 2, 0, 0).unwrap();

crates/registry-scheduling-core/src/diagnostics.rs

@@ -56,6 +56,10 @@ pub enum PolicyCheckReason {
     /// Two declarations draw on the same supply without a partition the
     /// ledger can enforce.
     SharedSupplyUnpartitioned,
+    /// A resource pool and a published window claim the same identifier.
+    /// Both anchor their capacity transactions on one row keyed by that
+    /// identifier, so the two kinds of supply must not share one.
+    SupplyIdentifierCollision,
     /// A local hook omits or changes the shared handler ABI.
     UnsupportedHookAbi,
     /// A hook phase and handler kind cannot run together.
@@ -101,6 +105,7 @@ impl PolicyCheckReason {
             Self::InvalidBands => "invalid-bands",
             Self::SubquotaOverdrawn => "subquota-overdrawn",
             Self::SharedSupplyUnpartitioned => "shared-supply-unpartitioned",
+            Self::SupplyIdentifierCollision => "supply-identifier-collision",
             Self::UnsupportedHookAbi => "unsupported-hook-abi",
             Self::UnsupportedHookPhase => "unsupported-hook-phase",
             Self::UnsupportedHookTrigger => "unsupported-hook-trigger",

crates/registry-scheduling-core/src/model.rs

@@ -158,6 +158,11 @@ impl LedgerSnapshot {
 
     /// Whether a consuming claim already holds `duplicate_key` at `now`,
     /// ignoring `exclude`.
+    ///
+    /// The key admits one live claim per party for an offering, so a claim
+    /// holds it only until the time it occupies has passed. A booking keeps
+    /// consuming capacity for overlap forever, but a party whose appointment
+    /// is behind them is free to book the offering again.
     pub fn duplicate_active(
         &self,
         offering: &str,
@@ -169,6 +174,7 @@ impl LedgerSnapshot {
             claim.offering == offering
                 && claim.duplicate_key.as_deref() == Some(duplicate_key)
                 && Some(claim.id.as_str()) != exclude
+                && claim.end > now
         })
     }
 }
@@ -574,6 +580,7 @@ mod tests {
             id: "claim-2".to_owned(),
             offering: "offering-b".to_owned(),
             kind: LedgerKind::Booking,
+            end: utc(5, 0),
             expires_at: None,
             ..expired_hold.clone()
         };
@@ -584,6 +591,30 @@ mod tests {
         assert!(snapshot.duplicate_active("offering-b", "subject:one", None, now));
     }
 
+    #[test]
+    fn an_elapsed_booking_stops_holding_the_duplicate_key() {
+        let booking = LedgerClaim {
+            id: "claim-1".to_owned(),
+            offering: "offering-a".to_owned(),
+            supply_id: "morning-window".to_owned(),
+            kind: LedgerKind::Booking,
+            channel: None,
+            start: utc(1, 0),
+            end: utc(2, 0),
+            units: 1,
+            duplicate_key: Some("subject:one".to_owned()),
+            expires_at: None,
+        };
+        let snapshot = LedgerSnapshot {
+            claims: vec![booking],
+        };
+        // The key names one live booking per party, so it is held while the
+        // appointment stands and released once the appointment has passed.
+        assert!(snapshot.duplicate_active("offering-a", "subject:one", None, utc(1, 30)));
+        assert!(!snapshot.duplicate_active("offering-a", "subject:one", None, utc(2, 0)));
+        assert!(!snapshot.duplicate_active("offering-a", "subject:one", None, utc(3, 0)));
+    }
+
     #[test]
     fn the_same_request_always_hashes_identically() {
         assert_eq!(

crates/registry-scheduling-core/src/policy.rs

@@ -544,6 +544,16 @@ impl SchedulingPolicy {
         for (index, offering) in self.offerings.iter().enumerate() {
             self.check_offering(offering, index, &mut findings);
         }
+        for (index, offering) in self.offerings.iter().enumerate() {
+            if let Some(arrival) = &offering.arrival {
+                if self.sells_pool(&arrival.window) {
+                    findings.push(SchedulingDiagnostic::new(
+                        format!("offerings[{index}].arrival.window"),
+                        PolicyCheckReason::SupplyIdentifierCollision,
+                    ));
+                }
+            }
+        }
 
         check_because(
             &self.hold_policy.because,
@@ -691,6 +701,21 @@ impl SchedulingPolicy {
                 findings,
             );
         }
+        // A request carries the capabilities and prerequisites that satisfy
+        // these, and the request edge bounds its own lists at the same
+        // maximum. An offering requiring more than that bound publishes
+        // cleanly and refuses every admissible request, so it is refused
+        // here instead.
+        check_collection_bound(
+            &offering.requires_capabilities,
+            &format!("{path}.requiresCapabilities"),
+            findings,
+        );
+        check_collection_bound(
+            &offering.prerequisites,
+            &format!("{path}.prerequisites"),
+            findings,
+        );
         for capability in &offering.requires_capabilities {
             if !valid_identifier(capability) {
                 findings.push(SchedulingDiagnostic::new(
@@ -778,6 +803,19 @@ impl SchedulingPolicy {
         }
         findings.extend(window.units_policy.check(&format!("{path}.unitsPolicy")));
 
+        // A pool and a window anchor their capacity transactions on one row
+        // keyed by the identifier, so the two supply namespaces must stay
+        // disjoint. Sharing an identifier makes the two publish paths write
+        // the same row: one aborts on the key, the other leaves the wrong
+        // kind standing and the next records swap deletes the anchor the
+        // pool still depends on.
+        if self.sells_pool(&window.id) {
+            findings.push(SchedulingDiagnostic::new(
+                format!("{path}.id"),
+                PolicyCheckReason::SupplyIdentifierCollision,
+            ));
+        }
+
         if window.leftover.is_some() {
             findings.push(SchedulingDiagnostic::new(
                 format!("{path}.leftover"),
@@ -893,6 +931,18 @@ impl SchedulingPolicy {
         }
     }
 
+    /// Whether any exact-time offering draws on the pool named `id`. Those
+    /// are the pools publication anchors, so they are the identifiers a
+    /// window may not take.
+    fn sells_pool(&self, id: &str) -> bool {
+        self.offerings.iter().any(|offering| {
+            offering
+                .exact_time
+                .as_ref()
+                .is_some_and(|exact| exact.pool == id)
+        })
+    }
+
     fn check_unique_id(
         &self,
         id: &str,
@@ -1715,6 +1765,69 @@ windows:
         );
     }
 
+    /// A resource pool and a published window anchor their capacity
+    /// transactions on one row keyed by the identifier, so the two supply
+    /// namespaces have to stay disjoint. The policy states one side of the
+    /// collision by itself, and the records state the other, so each
+    /// authored document is refused naming its own field.
+    #[test]
+    fn a_pool_and_a_window_may_not_share_one_supply_identifier() {
+        let mut policy = household_window_policy();
+        let windows = household_windows();
+        assert!(policy.check().is_empty(), "{:?}", policy.check());
+        assert!(
+            policy.check_window_records(&windows).is_empty(),
+            "{:?}",
+            policy.check_window_records(&windows)
+        );
+
+        policy.offerings.push(OfferingPolicy {
+            id: "urgent-five".to_owned(),
+            service: "household-day".to_owned(),
+            label: "Urgent five-minute slot".to_owned(),
+            mode: SchedulingMode::ExactTime,
+            location: "civic-hall".to_owned(),
+            because: "Urgent matters get exact five-minute slots.".to_owned(),
+            exact_time: Some(ExactTimeOffering {
+                duration_minutes: 5,
+                buffer_before_minutes: 0,
+                buffer_after_minutes: 0,
+                lead_time_minutes: 30,
+                horizon_days: 14,
+                pool: "household-morning-window".to_owned(),
+                start_increment_minutes: 5,
+                max_recipients: 1,
+            }),
+            arrival: None,
+            cancellation_cutoff_minutes: 60,
+            reminders: Vec::new(),
+            duplicate_active_key: None,
+            requires_capabilities: Vec::new(),
+            prerequisites: Vec::new(),
+        });
+
+        // The policy alone already names both sides: the arrival offering
+        // points at a window identifier its own exact-time offering sells.
+        let rendered: Vec<String> = policy.check().iter().map(ToString::to_string).collect();
+        assert!(
+            rendered
+                .contains(&"offerings[0].arrival.window: supply-identifier-collision".to_owned()),
+            "{rendered:?}"
+        );
+
+        // The published record is refused in its own vocabulary, which is
+        // the finding the two database writes carry.
+        let rendered: Vec<String> = policy
+            .check_window_records(&windows)
+            .iter()
+            .map(ToString::to_string)
+            .collect();
+        assert!(
+            rendered.contains(&"windows[0].id: supply-identifier-collision".to_owned()),
+            "{rendered:?}"
+        );
+    }
+
     /// AT-22, static half: an unpartitioned staffing claim over a pool that an
     /// exact-time offering also sells is rejected at publication.
     ///
@@ -2435,4 +2548,31 @@ surprise: true
         let at_bound = minimal_exact_time_policy();
         assert!(at_bound.check().is_empty());
     }
+
+    /// An offering may require no more capabilities or prerequisites than a
+    /// request is allowed to carry. Past that bound the offering publishes
+    /// cleanly and no admissible request can ever satisfy it, so the refusal
+    /// belongs at authoring time where the operator can still name it.
+    #[test]
+    fn offering_requirement_collections_are_bounded() {
+        let mut policy = minimal_exact_time_policy();
+        policy.offerings[0].requires_capabilities = (0..=MAXIMUM_COLLECTION_ENTRIES)
+            .map(|index| format!("capability-{index}"))
+            .collect();
+        policy.offerings[0].prerequisites = (0..=MAXIMUM_COLLECTION_ENTRIES)
+            .map(|index| format!("urn:evidence:residency-{index}"))
+            .collect();
+        let rendered: Vec<String> = policy.check().iter().map(|f| f.to_string()).collect();
+        assert!(rendered.contains(&"offerings[0].requiresCapabilities: invalid-bound".to_owned()));
+        assert!(rendered.contains(&"offerings[0].prerequisites: invalid-bound".to_owned()));
+
+        let mut at_bound = minimal_exact_time_policy();
+        at_bound.offerings[0].requires_capabilities = (1..=MAXIMUM_COLLECTION_ENTRIES)
+            .map(|index| format!("capability-{index}"))
+            .collect();
+        at_bound.offerings[0].prerequisites = (1..=MAXIMUM_COLLECTION_ENTRIES)
+            .map(|index| format!("urn:evidence:residency-{index}"))
+            .collect();
+        assert!(at_bound.check().is_empty());
+    }
 }

crates/registry-scheduling-core/src/problem.rs

@@ -266,7 +266,7 @@ impl ProblemCode {
                 "The bearer credential is missing, invalid, or expired. Sign in again."
             }
             Self::BookingDuplicateActive => {
-                "An active booking already holds this party's duplicate key. Cancel or complete it before booking again."
+                "An active booking already holds this party's duplicate key. Cancel it, or wait until it has passed, before booking again."
             }
             Self::CancellationCutoffPassed => {
                 "The cancellation cutoff for this appointment has passed, so it can no longer be cancelled."

crates/registry-scheduling/migrations/0007_duplicate_lookup_index.sql

@@ -0,0 +1,13 @@
+-- The offering-wide duplicate guard filters on three columns: the key the
+-- caller sent, the offering the claim was made under, and whether the booking
+-- it names is still ahead. Schema version 1 indexed the key alone, leaving the
+-- other two to a recheck over every active claim carrying that key. A booking
+-- stays 'active' once its time has passed, because the ledger publishes no
+-- completion transition and nothing sweeps elapsed bookings closed, so the
+-- rechecked set grows for the life of the deployment rather than with the
+-- bookings a party currently holds. Index the whole predicate instead.
+CREATE INDEX IF NOT EXISTS scheduling_claims_duplicate_lookup_idx
+    ON scheduling_claims(duplicate_key, offering, occupied_end)
+    WHERE state = 'active' AND duplicate_key IS NOT NULL;
+
+DROP INDEX IF EXISTS scheduling_claims_duplicate_idx;