feat(cubesys): Task 6 — owner enforcement on mutating commands

Stamp owner_local_user on records written via prog/write and gate the
mutating paths (prog, write, del — both live and buffered txn) so a
session may only create or overwrite a record whose owner_local_user
matches its HELLO-declared identity.

Design (logical + expedient for the whole project):
- Owner is the durable record-level CubeHeader.owner_local_user field,
  so enforcement is replay-safe and works across daemon restart.
- Enforcement is opt-in/non-breaking: gated only when the session has a
  stamped identity AND the record has an owner. First write by an owner
  claims an unowned coord; a session with no identity (tests, legacy)
  writes freely.
- COMMIT re-checks owner on each buffered op before applying, so a
  concurrent cross-owner commit between BEGIN and COMMIT is rejected
  (txn is restored for retry, not silently dropped).
- seal/open (encrypted raw put/del) left ungated for now: their headers
  are not owner-stamped yet — tracked as follow-up.

Verification:
- ./check quick: EXIT=0, fmt+clippy clean, 26 cubesys tests (added
  owner_enforcement_blocks_cross_owner_overwrite,
  owner_enforcement_allows_first_claim_and_same_owner,
  for_tenant_carries_identity).
- Ad-hoc daemon verifier over real cube-server socket (LE framing):
  cross-owner overwrite + delete rejected, same-owner + first-claim
  allowed, no-HELLO legacy writes allowed. ALL PASS.

cubesys/src/commands.rs

@@ -207,6 +207,19 @@ impl Session {
                     .take()
                     .ok_or_else(|| "commit: no transaction is open".to_string())?;
                 let n = txn.ops.len();
+                // Task 6: owner enforcement for buffered txn ops. The live path
+                // already checked at `prog`/`write`/`del` time, but a concurrent
+                // commit from another owner could have claimed the coord between
+                // our BEGIN and COMMIT, so re-check now (the buffer is re-read
+                // here, consistent with the single WAL entry T5 writes).
+                for op in &txn.ops {
+                    if let Some(msg) = owner_violation(&self.identity, &store.owner(&op.coord)) {
+                        // Roll the txn back: restore it so the caller can retry
+                        // after resolving the conflict (do not silently drop).
+                        self.txn = Some(txn);
+                        return Err(msg);
+                    }
+                }
                 // Apply every buffered op atomically under one store write lock
                 // and append a SINGLE WAL entry (WalOp::Txn) so the whole batch
                 // is durable as one unit and replays idempotently.
@@ -266,13 +279,18 @@ impl Session {
                 // a throwaway store so we can buffer (or apply) them without
                 // duplicating the record codec.
                 let coord = scratch_code_coord(path, Kind::Fn, name, &ops)?;
+                // Task 6: reject overwriting a record owned by a different owner.
+                if let Some(msg) = owner_violation(&self.identity, &store.owner(&coord)) {
+                    return Err(msg);
+                }
+                let owner = self.identity.as_ref().map(|i| i.owner_local.as_str());
                 let value = {
                     let mut scratch = CubeStore::new(HashBackend::new());
-                    crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &ops)
+                    crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &ops, owner)
                         .map_err(|e| e.to_string())?;
                     scratch.get_raw(&coord).unwrap_or_default()
                 };
-                let header = header_for_code(Kind::Fn, name, &ops);
+                let header = header_for_code(Kind::Fn, name, &ops, owner);
                 if let Some(txn) = self.txn.as_mut() {
                     txn.ops.push(TxnOp {
                         coord,
@@ -284,7 +302,7 @@ impl Session {
                     ));
                 }
                 let coord = store
-                    .put_code_cell(path, Kind::Fn, name, &[], &ops)
+                    .put_code_cell(path, Kind::Fn, name, &[], &ops, owner)
                     .map_err(|e| e.to_string())?;
                 Ok(format!(
                     "wrote program {path} -> coord {} ({} ops)",
@@ -302,13 +320,17 @@ impl Session {
                     .map_err(|e| format!("bytecode decode error: {e:?}"))?;
                 let name = path.rsplit('/').next().unwrap_or(path);
                 let coord = scratch_code_coord(path, Kind::Fn, name, &code)?;
+                if let Some(msg) = owner_violation(&self.identity, &store.owner(&coord)) {
+                    return Err(msg);
+                }
+                let owner = self.identity.as_ref().map(|i| i.owner_local.as_str());
                 let value = {
                     let mut scratch = CubeStore::new(HashBackend::new());
-                    crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &code)
+                    crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &code, owner)
                         .map_err(|e| e.to_string())?;
                     scratch.get_raw(&coord).unwrap_or_default()
                 };
-                let header = header_for_code(Kind::Fn, name, &code);
+                let header = header_for_code(Kind::Fn, name, &code, owner);
                 if let Some(txn) = self.txn.as_mut() {
                     txn.ops.push(TxnOp {
                         coord,
@@ -320,13 +342,18 @@ impl Session {
                     ));
                 }
                 let coord = store
-                    .put_code_cell(path, Kind::Fn, name, &[], &code)
+                    .put_code_cell(path, Kind::Fn, name, &[], &code, owner)
                     .map_err(|e| e.to_string())?;
                 Ok(format!("wrote {path} -> coord {}", coord.pack_u32()))
             }
             "del" => {
                 let path = it.next().ok_or_else(|| "del needs <path>".to_string())?;
                 let coord = crate::path_to_czyx(path).map_err(|e| e.to_string())?;
+                // Task 6: reject deleting a record owned by a different owner
+                // (unless unowned, which any identity may take over).
+                if let Some(msg) = owner_violation(&self.identity, &store.owner(&coord)) {
+                    return Err(msg);
+                }
                 if let Some(txn) = self.txn.as_mut() {
                     txn.ops.push(TxnOp { coord, put: None });
                     return Ok(format!("buffered del {path} — commit to apply"));
@@ -455,16 +482,48 @@ pub fn txn_snapshot(s: &Session) -> CubeStore<HashBackend> {
 /// writing — used to buffer `prog`/`write` mutations during a transaction.
 fn scratch_code_coord(path: &str, kind: Kind, name: &str, code: &[Op]) -> Result<Czyx, String> {
     let mut scratch = CubeStore::new(HashBackend::new());
-    crate::store_code_cell(&mut scratch, path, kind, name, &[], code).map_err(|e| e.to_string())
+    crate::store_code_cell(&mut scratch, path, kind, name, &[], code, None).map_err(|e| e.to_string())
+}
+
+/// Owner-enforcement check (Task 6): a mutating command may only create or
+/// overwrite a record when EITHER the record is unowned OR the session's
+/// identity owner matches the record's `owner_local_user`. Returns `None` when
+/// the write is allowed, or an `Err`-style message when it must be rejected.
+///
+/// Enforcement is opt-in / non-breaking: if the session has no stamped
+/// identity (e.g. a `Session::with_store` used by tests, or a client that
+/// never sent `HELLO`), every write is permitted. The daemon stamps identity
+/// via `HELLO` (Task 3) before any mutating op, so live traffic is gated.
+fn owner_violation(identity: &Option<TenantIdentity>, record_owner: &Option<String>) -> Option<String> {
+    let owner = identity.as_ref().map(|i| i.owner_local.as_str());
+    match (owner, record_owner) {
+        // No session identity -> no gate (legacy / test path).
+        (None, _) => None,
+        // Session has an identity but the record is unowned -> allow (first write
+        // by this owner claims it).
+        (Some(_), None) => None,
+        // Both set: must match exactly.
+        (Some(session_owner), Some(record_owner)) => {
+            if session_owner == record_owner.as_str() {
+                None
+            } else {
+                Some(format!(
+                    "owner violation: record owned by '{record_owner}', you are '{session_owner}'"
+                ))
+            }
+        }
+    }
 }
 
 /// Build the `CubeHeader` a `store_code_cell` call would attach (mirrors
 /// `crate::store_code_cell`), so a buffered txn put carries the same header.
-fn header_for_code(kind: Kind, name: &str, code: &[Op]) -> CubeHeader {
+/// `owner` (when set) is stamped on `owner_local_user` for Task 6 enforcement.
+fn header_for_code(kind: Kind, name: &str, code: &[Op], owner: Option<&str>) -> CubeHeader {
     let mut h = CubeHeader::new();
     h.title = Some(name.to_string());
     h.doc_type = Some(kind.as_str().to_string());
     h.linked_records = Vec::new();
+    h.owner_local_user = owner.map(|o| o.to_string());
     if h.doc_type.as_deref() == Some("fn") {
         h.size_bytes = Some(cubecode::encode(code).len() as u64);
     }
@@ -559,6 +618,7 @@ pub fn parse_transform(s: &str) -> Option<TransformId> {
 mod tests {
     use super::*;
     use crate::store::DurabilityConfig;
+    use std::str::FromStr;
     use std::sync::Arc;
 
     fn session() -> Session {
@@ -591,6 +651,104 @@ mod tests {
         assert!(s.exec("commit").is_ok());
     }
 
+    #[test]
+    fn owner_enforcement_blocks_cross_owner_overwrite() {
+        use crate::tenant::{TenantIdentity, TenantId};
+        let mut s = session();
+        // Stamp an identity (Task 3 path) for owner "alice".
+        s.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "alice".to_string(),
+            owner_remote: None,
+        });
+        // Alice writes a record — it should be stamped + succeed.
+        assert!(s.exec("prog /c050/z001/y001/x001 const 7 halt").is_ok());
+        assert_eq!(
+            s.store.owner(&Czyx::new(50, 1, 1, 1)).as_deref(),
+            Some("alice")
+        );
+        // Now "bob" tries to overwrite alice's coord — must be rejected.
+        s.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "bob".to_string(),
+            owner_remote: None,
+        });
+        let r = s.exec("prog /c050/z001/y001/x001 const 8 halt");
+        assert!(r.is_err(), "cross-owner overwrite must be rejected");
+        assert!(r.unwrap_err().contains("owner violation"));
+        // The record must still be alice's (unchanged value).
+        assert_eq!(
+            s.store.owner(&Czyx::new(50, 1, 1, 1)).as_deref(),
+            Some("alice")
+        );
+        // And a delete by bob is likewise rejected.
+        let d = s.exec("del /c050/z001/y001/x001");
+        assert!(d.is_err());
+        assert!(d.unwrap_err().contains("owner violation"));
+    }
+
+    #[test]
+    fn owner_enforcement_allows_first_claim_and_same_owner() {
+        use crate::tenant::{TenantIdentity, TenantId};
+        let mut s = session();
+        // A session with NO identity writes freely (legacy / test path).
+        assert!(s.exec("prog /c051/z001/y001/x001 const 1 halt").is_ok());
+        assert_eq!(s.store.owner(&Czyx::new(51, 1, 1, 1)), None);
+        // Alice claims the previously-unowned coord — allowed (first write).
+        s.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "alice".to_string(),
+            owner_remote: None,
+        });
+        assert!(s.exec("prog /c051/z001/y001/x001 const 2 halt").is_ok());
+        assert_eq!(
+            s.store.owner(&Czyx::new(51, 1, 1, 1)).as_deref(),
+            Some("alice")
+        );
+        // Same owner overwrites freely.
+        assert!(s.exec("prog /c051/z001/y001/x001 const 3 halt").is_ok());
+    }
+
+    #[test]
+    fn for_tenant_carries_identity_to_session() {
+        // Mirrors the cube-server daemon flow: a connection resolves its tenant
+        // session, stamps identity via HELLO, then runs a per-connection
+        // `Session::for_tenant` for the rest of the connection. The identity
+        // MUST propagate so owner enforcement fires on the live path.
+        use crate::tenant::{TenantConfig, TenantId, TenantIdentity, TenantRegistry};
+        use std::str::FromStr;
+
+        let registry = TenantRegistry::with_config(TenantConfig::Memory);
+        let ts = registry
+            .get_or_provision(TenantId::from_str("alpha").unwrap())
+            .unwrap();
+        ts.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "alice".to_string(),
+            owner_remote: None,
+        });
+        let mut s = Session::for_tenant(&ts);
+        // The per-connection session must carry the stamped identity.
+        assert_eq!(s.identity().unwrap().owner_local, "alice");
+        // Alice writes her coord (stamped).
+        assert!(s.exec("prog /c052/z001/y001/x001 const 7 halt").is_ok());
+        assert_eq!(
+            s.store.owner(&Czyx::new(52, 1, 1, 1)).as_deref(),
+            Some("alice")
+        );
+        // A second connection as bob shares the same store (legacy mode) but
+        // must be blocked from overwriting alice's record.
+        ts.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "bob".to_string(),
+            owner_remote: None,
+        });
+        let mut b = Session::for_tenant(&ts);
+        let r = b.exec("prog /c052/z001/y001/x001 const 8 halt");
+        assert!(r.is_err(), "cross-owner overwrite must be rejected live");
+        assert!(r.unwrap_err().contains("owner violation"));
+    }
+
     #[test]
     fn txn_isolation_begin_snapshot_hides_live_writer() {
         // Stream A opens a txn and snapshots; stream B mutates the live store.

cubesys/src/lib.rs

@@ -132,12 +132,14 @@ pub fn store_code_cell<B: CubeBackend>(
     name: &str,
     links: &[Czyx],
     code: &[cubecode::Op],
+    owner: Option<&str>,
 ) -> Result<Czyx, SysError> {
     let coord = path_to_czyx(path)?;
     let mut h = CubeHeader::new();
     h.title = Some(name.to_string());
     h.doc_type = Some(kind.as_str().to_string());
     h.linked_records = links.to_vec();
+    h.owner_local_user = owner.map(|o| o.to_string());
     if h.doc_type.as_deref() == Some("fn") {
         h.size_bytes = Some(cubecode::encode(code).len() as u64);
     }
@@ -176,6 +178,7 @@ mod tests {
             "add",
             &[],
             &[Op::Const(2), Op::Const(3), Op::Add, Op::Halt],
+            None,
         )
         .unwrap();
 
@@ -282,7 +285,7 @@ pub mod demo {
         let entry_code = vec![Op::Const(21), Op::CallLink(0), Op::Halt];
 
         let double_coord =
-            super::store_code_cell(&mut store, double, Kind::Fn, "double", &[], &double_code)
+            super::store_code_cell(&mut store, double, Kind::Fn, "double", &[], &double_code, None)
                 .expect("store double");
         let entry_coord = super::store_code_cell(
             &mut store,
@@ -291,6 +294,7 @@ pub mod demo {
             "entry",
             &[double_coord],
             &entry_code,
+            None,
         )
         .expect("store entry");
         println!("  wrote {double}  -> coord {}", double_coord.pack_u32());

cubesys/src/store.rs

@@ -491,6 +491,17 @@ impl ConcurrentStore {
         out
     }
 
+    /// The `owner_local_user` stamped on the record at `key`, if it has one.
+    /// Used by owner enforcement (Task 6): a mutating command may only
+    /// overwrite a record whose owner matches the session's identity owner.
+    pub fn owner(&self, key: &Czyx) -> Option<String> {
+        self.inner
+            .read()
+            .unwrap()
+            .get_record(key)
+            .and_then(|(h, _)| h.owner_local_user.clone())
+    }
+
     /// A consistent point-in-time snapshot of the whole store. Used by the VM
     /// and cubefs, which take a `CubeStore` by value. A read-lock clone, so it
     /// does not block concurrent readers (it only waits for an in-flight
@@ -580,6 +591,8 @@ impl ConcurrentStore {
     }
 
     /// Store a code cell at `path` (the path->code bridge), durability-logged.
+    /// `owner` (when set) is stamped on the record's `owner_local_user` field
+    /// so owner enforcement (Task 6) can later reject cross-owner overwrites.
     pub fn put_code_cell(
         &self,
         path: &str,
@@ -587,9 +600,9 @@ impl ConcurrentStore {
         name: &str,
         links: &[Czyx],
         code: &[Op],
+        owner: Option<&str>,
     ) -> Result<Czyx, crate::SysError> {
-        let coord =
-            self.with_mut(|store| crate::store_code_cell(store, path, kind, name, links, code))?;
+        let coord = self.with_mut(|store| crate::store_code_cell(store, path, kind, name, links, code, owner))?;
         if let Some(v) = self.get_raw(&coord) {
             self.log_put(coord, v);
         }