feat(query): PDF Package-2 associative log lookup via flags/metatags

Implements the source PDF's 'put(C,Z,Y,X,bytes,flags)' + 'scan_by_flag'
associative-storage API so records are discoverable by WHAT they are
(event type / doc_type metatag) rather than WHERE they live (coordinate
path) — the 'log lookup by query' the OS layers want.

cubestore:
  - CubeStore::scan_by_flag(u16) / scan_by_type(&str): predicate scans
    over decoded CubeHeader (not prefix scans).
  - CubeStore::query_by_flag / query_by_type: same, but also return the
    decoded (label, header, body) so a query tool can present results.
  - get_record's raw/un-enveloped fallback now refresh_flags() so
    synthesized headers carry SIZE_BYTES and are visible to flag queries.
  - regression tests: scan_by_flag_finds_typed_records,
    scan_by_type_event_lookup.

cubesys (daemon):
  - new 'put' verb: enveloped write with optional doc_type=/title=
    metatags (the PDF's put API). rawput remains for bulk/append payloads.
  - ConcurrentStore::query_doc_type was already the backing predicate the
    'query <doc_type>' verb uses; it now matches these typed records.

Proven end-to-end in the VM: 'cubec query klog' returns the kernel-log
index record (doc_type=klog) published by cube-os-klog.sh, and
'cubec query fn' returns the 8 pre-existing typed program records.

cubestore/src/lib.rs

@@ -411,6 +411,13 @@ impl<B: CubeBackend> CubeStore<B> {
         let raw_fallback = |bytes: &[u8]| {
             let mut h = CubeHeader::new();
             h.size_bytes = Some(bytes.len() as u64);
+            // Keep the synthesized header's flag bits consistent with its
+            // fields, so associative queries (`scan_by_flag(SIZE_BYTES)`) see
+            // raw records too. Only the derived low bits are recomputed; any
+            // out-of-band bits already on the payload-derived header (none
+            // here, since raw payloads carry no header) are preserved by
+            // `refresh_flags`.
+            h.refresh_flags();
             Some((h, bytes.to_vec()))
         };
         if raw.len() < 4 {
@@ -500,6 +507,78 @@ impl<B: CubeBackend> CubeStore<B> {
         out.sort();
         out
     }
+
+    /// PDF Package 2 API: `scan_by_flag` — associative storage lookup.
+    ///
+    /// Returns every coordinate whose decoded [`CubeHeader`] carries `flag`
+    /// set. This is the "query by metatag/flag, not by path" primitive the
+    /// source PDF describes (§"associative storage"): records are addressed
+    /// by *what they are* (a flag bit) rather than *where they live* (a
+    /// coordinate path). `cube-os-*` services that emit typed records
+    /// (e.g. a log line tagged `doc_type = "klog"`) become discoverable by
+    /// event type without knowing their CZYX address in advance.
+    ///
+    /// Records written through the RAW path (`rawput`, used by the OS layers)
+    /// carry only a synthesized `size_bytes` header, so they will NOT match
+    /// a content flag unless the writer also set one. The associative query
+    /// is therefore most powerful when records are written through
+    /// [`CubeStore::put_record`] with a populated [`CubeHeader`].
+    pub fn scan_by_flag(&self, flag: u16) -> Vec<Czyx> {
+        let mut out: Vec<Czyx> = self
+            .backend
+            .keys()
+            .into_iter()
+            .filter(|k| {
+                self.get_record(k)
+                    .map(|(h, _)| h.flags.has(flag))
+                    .unwrap_or(false)
+            })
+            .collect();
+        out.sort();
+        out
+    }
+
+    /// PDF Package 2 API: `scan_by_type` — query records by `doc_type`
+    /// (the "event type" / file-extension analogue the PDF calls Flag 2).
+    ///
+    /// This is the concrete "log lookup by query" the OS layers want: instead
+    /// of addressing `/cubefs/c200/z004/...` directly, you ask "every record
+    /// whose doc_type is `klog`" and get back all matching coordinates. The
+    /// comparison is exact-match (case-sensitive) against the header field.
+    pub fn scan_by_type(&self, doc_type: &str) -> Vec<Czyx> {
+        let mut out: Vec<Czyx> = self
+            .backend
+            .keys()
+            .into_iter()
+            .filter(|k| {
+                self.get_record(k)
+                    .map(|(h, _)| h.doc_type.as_deref() == Some(doc_type))
+                    .unwrap_or(false)
+            })
+            .collect();
+        out.sort();
+        out
+    }
+
+    /// Return `(label, header, body)` for every record matching `flag`.
+    /// Convenience wrapper over [`CubeStore::scan_by_flag`] that also pulls
+    /// the decoded payload so a caller (e.g. a `cubelog` query tool) can
+    /// present the matching records directly.
+    pub fn query_by_flag(&self, flag: u16) -> Vec<(Czyx, CubeHeader, Vec<u8>)> {
+        self.scan_by_flag(flag)
+            .into_iter()
+            .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b)))
+            .collect()
+    }
+
+    /// Return `(label, header, body)` for every record whose `doc_type`
+    /// matches. See [`CubeStore::scan_by_type`] for the matching semantics.
+    pub fn query_by_type(&self, doc_type: &str) -> Vec<(Czyx, CubeHeader, Vec<u8>)> {
+        self.scan_by_type(doc_type)
+            .into_iter()
+            .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b)))
+            .collect()
+    }
 }
 
 #[cfg(test)]
@@ -594,4 +673,64 @@ mod tests {
             "must decode the true header, not fall back to raw"
         );
     }
+
+    /// PDF Package 2: `scan_by_flag` finds records by flag bit, independent of
+    /// their coordinate address. A record tagged by `DOC_TYPE` must surface
+    /// when queried for that flag and be absent otherwise.
+    #[test]
+    fn scan_by_flag_finds_typed_records() {
+        let mut store = CubeStore::new(HashBackend::new());
+
+        // A "klog" typed event record.
+        let mut klog = CubeHeader::new();
+        klog.doc_type = Some("klog".into());
+        klog.refresh_flags();
+        store.put_record(Czyx::new(200, 4, 2, 1), &klog, b"kernel: eth0 up");
+
+        // An untyped raw record (the common OS-layer case).
+        store.put_raw(Czyx::new(200, 70, 1, 1), b"118".to_vec());
+
+        let by_doc_type = store.scan_by_flag(cubecoords::HeaderFlags::DOC_TYPE);
+        assert_eq!(by_doc_type, vec![Czyx::new(200, 4, 2, 1)]);
+
+        // The raw record carries only a synthesized size flag, so it must NOT
+        // match DOC_TYPE.
+        assert!(!by_doc_type.contains(&Czyx::new(200, 70, 1, 1)));
+
+        let by_size = store.scan_by_flag(cubecoords::HeaderFlags::SIZE_BYTES);
+        assert!(
+            !by_size.is_empty(),
+            "synthesized raw headers carry SIZE_BYTES"
+        );
+    }
+
+    /// `scan_by_type` is the concrete "log lookup by query" — pull every record
+    /// of a given event type without knowing its coordinate.
+    #[test]
+    fn scan_by_type_event_lookup() {
+        let mut store = CubeStore::new(HashBackend::new());
+        let mut a = CubeHeader::new();
+        a.doc_type = Some("klog".into());
+        let mut b = CubeHeader::new();
+        b.doc_type = Some("klog".into());
+        let mut c = CubeHeader::new();
+        c.doc_type = Some("state".into());
+        store.put_record(Czyx::new(200, 4, 2, 1), &a, b"k1");
+        store.put_record(Czyx::new(200, 4, 2, 2), &b, b"k2");
+        store.put_record(Czyx::new(200, 1, 1, 1), &c, b"s1");
+
+        let klogs = store.scan_by_type("klog");
+        assert_eq!(klogs.len(), 2);
+        assert!(klogs.contains(&Czyx::new(200, 4, 2, 1)));
+        assert!(klogs.contains(&Czyx::new(200, 4, 2, 2)));
+
+        // query_by_type returns the decoded payloads too.
+        let klogs_full = store.query_by_type("klog");
+        assert_eq!(klogs_full.len(), 2);
+        assert!(klogs_full.iter().any(|(_, _, b)| b == b"k1"));
+        assert!(klogs_full.iter().any(|(_, _, b)| b == b"k2"));
+
+        assert_eq!(store.scan_by_type("state"), vec![Czyx::new(200, 1, 1, 1)]);
+        assert!(store.scan_by_type("nonexistent").is_empty());
+    }
 }

cubesys/src/commands.rs

@@ -939,6 +939,62 @@ impl Session {
                 store.delete_raw(&coord);
                 Ok(format!("ok: deleted {}", coord.pack_u32()))
             }
+            // --- Enveloped coordinate API (PDF Package 2: `put(C,Z,Y,X,
+            //     bytes, flags)`). Unlike `rawput`, this writes a full record
+            //     with a `CubeHeader`, so the record carries metatags
+            //     (doc_type / title) that the associative `query` verb can
+            //     target by event type. `rawput` stays for bulk/append payloads
+            //     (e.g. the rolling kernel-log window) that do not need header
+            //     metadata; `put` is for addressed, discoverable records. ---
+            "put" => {
+                let c = parse_u8(it.next(), "put needs <c>")?;
+                let z = parse_u8(it.next(), "put needs <z>")?;
+                let y = parse_u8(it.next(), "put needs <y>")?;
+                let x = parse_u8(it.next(), "put needs <x>")?;
+                let hex = it
+                    .next()
+                    .ok_or_else(|| "put needs <hex-bytes>".to_string())?;
+                let val = hex_decode(hex).ok_or_else(|| "put: value must be hex".to_string())?;
+                let coord = Czyx::new(c, z, y, x);
+                // Optional metatags: `doc_type=<t>` and/or `title=<t>` may
+                // follow. These are what `query <doc_type>` matches on.
+                let mut header = CubeHeader::new();
+                while let Some(tok) = it.next() {
+                    if let Some(v) = tok.strip_prefix("doc_type=") {
+                        header.doc_type = Some(v.to_string());
+                    } else if let Some(v) = tok.strip_prefix("title=") {
+                        header.title = Some(v.to_string());
+                    } else {
+                        return Err(format!("put: unknown option {tok}"));
+                    }
+                }
+                header.created_at = Some(
+                    std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .map(|d| d.as_secs())
+                        .unwrap_or(0),
+                );
+                header.refresh_flags();
+                let doc_type_dbg = header.doc_type.clone();
+                if let Some(txn) = self.txn.as_mut() {
+                    txn.ops.push(TxnOp {
+                        coord,
+                        put: Some((val, header)),
+                    });
+                    Ok(format!(
+                        "buffered put {} (doc_type={:?}) — commit to apply",
+                        coord.pack_u32(),
+                        doc_type_dbg
+                    ))
+                } else {
+                    store.put_record(coord, &header, &val);
+                    Ok(format!(
+                        "ok: wrote {} (doc_type={:?})",
+                        coord.pack_u32(),
+                        doc_type_dbg
+                    ))
+                }
+            }
             "rawkeys" => {
                 let ks: Vec<String> = store
                     .keys()