Wave-1 review gate: validation coverage fixes (unschedule, handle-level consumer-local entry points), ADR-008 §4 annotations (whitespace-exclusion, class scope), task check-line staleness fix (ADR-008/009/021, core-contract)

alkstore-contract-suite/src/properties.rs

@@ -9,10 +9,20 @@
 
 use serde_json::json;
 
-use alkstore::{BoxedFuture, Error, RESERVED_PREFIX, Result, validate_shared_name};
+use alkstore::{
+    BoxedFuture, Delivery, Error, JobHandle, RESERVED_PREFIX, Result, validate_shared_name,
+};
 
 use crate::factory::StoreFactory;
 
+struct NoDelivery;
+
+impl Delivery for NoDelivery {
+    fn deliver<'a>(&'a mut self, _job: Box<dyn JobHandle>) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(std::future::ready(Ok(())))
+    }
+}
+
 fn expect_invalid<T>(label: &str, out: alkstore::Result<T>) {
     match out {
         Err(Error::InvalidName { .. }) => {}
@@ -49,9 +59,13 @@ fn expect_ok<T>(label: &str, out: alkstore::Result<T>) {
 /// [`Error::InvalidName`], and the reserved-prefix name
 /// (`__alkstore_...`) with [`Error::ReservedName`], on the
 /// shared-namespace kinds — channels, streams, queues, outboxes,
-/// locks, schedule names, and `schedule()`'s queue argument.
+/// locks, schedule names (including `unschedule`), and `schedule()`'s
+/// queue argument.
 /// Consumer-local identifiers take the non-empty rule only: `owner`
-/// reserved-prefixed is legal, `owner` empty is not. The tx path is
+/// reserved-prefixed is legal, `owner` empty is not — likewise
+/// `worker_id` on the claim/pull entry points (`claim_one`,
+/// `claim_batch`, `run_once`) and the consumer arguments on the
+/// stream handle's offset/read forms. The tx path is
 /// exercised through `begin_tx` and `with_tx` (the provided wrapper):
 /// validation must not be bypassable by the commit-atomic path.
 ///
@@ -96,6 +110,11 @@ pub async fn name_validation_rejects_empty_and_reserved(factory: &dyn StoreFacto
             .schedule("tick", "@every 1s", "", json!(null), Default::default())
             .await,
     );
+    expect_invalid("unschedule(empty name)", store.unschedule("").await);
+    expect_reserved(
+        "unschedule(reserved name)",
+        store.unschedule(RESERVED_PREFIX).await,
+    );
 
     // Auto-commit paths — reserved-prefix names.
     expect_reserved(
@@ -155,6 +174,40 @@ pub async fn name_validation_rejects_empty_and_reserved(factory: &dyn StoreFacto
         store.try_lock("lock", RESERVED_PREFIX, 60).await,
     );
 
+    // Handle-level name-bearing entry points: the claim/pull ops'
+    // `worker_id` and the stream handle's consumer arguments are
+    // consumer-local identifiers (non-empty only, ADR-008 §4's
+    // third-round annotation) — validated at the handle's entry
+    // points identically to the store-level ones.
+    let queue = store
+        .queue("work", alkstore::QueueOpts::default())
+        .await
+        .expect("queue handle");
+    expect_invalid("claim_one(empty worker_id)", queue.claim_one("").await);
+    expect_invalid(
+        "claim_batch(empty worker_id)",
+        queue.claim_batch("", 1).await,
+    );
+    let stream = store.stream("work").await.expect("stream handle");
+    expect_invalid(
+        "save_offset(empty consumer)",
+        stream.save_offset("", 0).await,
+    );
+    expect_invalid("get_offset(empty consumer)", stream.get_offset("").await);
+    expect_invalid(
+        "read_from_consumer(empty consumer)",
+        stream.read_from_consumer("", 10).await,
+    );
+    expect_invalid(
+        "get_offset(whitespace-only consumer)",
+        stream.get_offset(" \t ").await,
+    );
+    let mut outbox = store.outbox("work").await.expect("outbox handle");
+    expect_invalid(
+        "run_once(empty worker_id)",
+        outbox.run_once("", &mut NoDelivery).await,
+    );
+
     // Tx paths — the twins validate identically (begin_tx route).
     let mut tx = store.begin_tx().await.expect("begin_tx opens");
     expect_invalid(
@@ -247,6 +300,10 @@ pub async fn name_validation_rejects_empty_and_reserved(factory: &dyn StoreFacto
         "read_from_consumer_tx(empty consumer)",
         tx.read_from_consumer_tx("work", "", 10).await,
     );
+    expect_invalid(
+        "get_offset_tx(empty consumer)",
+        tx.get_offset_tx("work", "").await,
+    );
     expect_invalid(
         "outbox_enqueue_tx(empty outbox)",
         tx.outbox_enqueue_tx("", Default::default(), json!(null))

alkstore-contract-suite/tests/suite_harness.rs

@@ -79,9 +79,11 @@ impl TxHandle for MockTx {
     fn get_offset_tx<'a>(
         &'a mut self,
         stream: &str,
-        _consumer: &str,
+        consumer: &str,
     ) -> BoxedFuture<'a, Result<i64>> {
-        let r = alkstore::validate_shared_name(stream).map(|_| 0);
+        let r = alkstore::validate_shared_name(stream)
+            .and_then(|_| alkstore::validate_local_name(consumer))
+            .map(|_| 0);
         Box::pin(std::future::ready(r))
     }
     fn read_since_tx<'a>(
@@ -133,16 +135,18 @@ impl Queue for MockQueue {
     }
     fn claim_one<'a>(
         &'a self,
-        _worker_id: &str,
+        worker_id: &str,
     ) -> BoxedFuture<'a, Result<Option<Box<dyn JobHandle>>>> {
-        Box::pin(std::future::ready(Ok(None)))
+        let r = alkstore::validate_local_name(worker_id).map(|_| None);
+        Box::pin(std::future::ready(r))
     }
     fn claim_batch<'a>(
         &'a self,
-        _worker_id: &str,
+        worker_id: &str,
         _n: i64,
     ) -> BoxedFuture<'a, Result<Vec<Box<dyn JobHandle>>>> {
-        Box::pin(std::future::ready(Ok(Vec::new())))
+        let r = alkstore::validate_local_name(worker_id).map(|_| Vec::new());
+        Box::pin(std::future::ready(r))
     }
     fn ack_batch<'a>(&'a self, _ids: &[i64]) -> BoxedFuture<'a, Result<i64>> {
         Box::pin(std::future::ready(Ok(0)))
@@ -183,16 +187,18 @@ impl StreamHandle for MockStream {
     }
     fn read_from_consumer<'a>(
         &'a self,
-        _consumer: &str,
+        consumer: &str,
         _limit: i64,
     ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
-        Box::pin(std::future::ready(Ok(Vec::new())))
+        let r = alkstore::validate_local_name(consumer).map(|_| Vec::new());
+        Box::pin(std::future::ready(r))
     }
-    fn save_offset<'a>(&'a self, _consumer: &str, _offset: i64) -> BoxedFuture<'a, Result<()>> {
-        Box::pin(std::future::ready(Ok(())))
+    fn save_offset<'a>(&'a self, consumer: &str, _offset: i64) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(std::future::ready(alkstore::validate_local_name(consumer)))
     }
-    fn get_offset<'a>(&'a self, _consumer: &str) -> BoxedFuture<'a, Result<i64>> {
-        Box::pin(std::future::ready(Ok(0)))
+    fn get_offset<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<i64>> {
+        let r = alkstore::validate_local_name(consumer).map(|_| 0);
+        Box::pin(std::future::ready(r))
     }
     fn trim_to<'a>(&'a self, _horizon: i64) -> BoxedFuture<'a, Result<i64>> {
         Box::pin(std::future::ready(Ok(0)))
@@ -287,8 +293,9 @@ impl Store for MockStore {
             });
         Box::pin(std::future::ready(r))
     }
-    fn unschedule<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result<bool>> {
-        Box::pin(std::future::ready(Ok(false)))
+    fn unschedule<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result<bool>> {
+        let r = alkstore::validate_shared_name(name).map(|_| false);
+        Box::pin(std::future::ready(r))
     }
     fn run_schedules<'a>(&'a self, _stop: StopToken) -> BoxedFuture<'a, Result<()>> {
         Box::pin(std::future::ready(Ok(())))
@@ -310,10 +317,11 @@ impl Outbox for MockOutbox {
     }
     fn run_once<'a>(
         &'a mut self,
-        _worker_id: &str,
+        worker_id: &str,
         _delivery: &'a mut dyn Delivery,
     ) -> BoxedFuture<'a, Result<bool>> {
-        Box::pin(std::future::ready(Ok(false)))
+        let r = alkstore::validate_local_name(worker_id).map(|_| false);
+        Box::pin(std::future::ready(r))
     }
 }