fix(T-0896): Rust-cdylib accumulator kinds — real type in FFI metadata + macros usable in packaged crates

.metis/backlog/bugs/CLOACI-T-0896.md

@@ -169,4 +169,8 @@ initiative_id: NULL
 - `cloacina-python`: `resolve_poll_closure(name)` looks up the registered Python poll fn and wraps it (`call0` → `depythonize` → JSON bytes; None → skip). Installed from `register_authoring`, so BOTH embeddings wire it (pip wheel + server synthetic `ensure_cloaca_module`).
 - Example `python-polling-graph` + new `_graph_autofire_steps` lane asserting `poll_reactor` self-fires (no inject). **Verified live: reactor self-fired.**
 
-**T-0896 CLOSED:** batch (d2043bbc) + polling (b010dbee) + loud-WARN fallback. No plugin interface/ABI bump. All accumulator kinds now behave correctly on the packaged path (passthrough/state/batch/polling; stream = kafka/T-0898 track).
+**T-0896 CLOSED (Python path):** batch (d2043bbc) + polling (b010dbee) + loud-WARN fallback. No plugin interface/ABI bump.
+
+**2026-07-14 — RUST-CDYLIB SIDE ALSO CLOSED (commit 5b7e19fc, PR #194).** The plugin shell hardcoded `accumulator_type = "passthrough"` in get_graph/reactor metadata (Rust analog of the Python bug), AND the accumulator macros couldn't be used in a lean packaged crate (their generated `impl cloacina::…` structs need the `cloacina` umbrella; no fixture ever decorated one — the path was dead). Fixed: new `AccumulatorEntry` inventory (name/type/config); the two metadata sites report the REAL kind from it (fallback passthrough only when a name has no entry); each accumulator macro emits a cfg-gated `AccumulatorEntry` submission AND gates its `cloacina::`-bound struct/impls behind `cfg(not(packaged))` so `#[state_accumulator]` etc. compile in lean packaged crates. `reactor-only-rust` now declares `alpha` as `#[state_accumulator(capacity=5)]`; `primitive_only_packaging` asserts metadata reports alpha=`state`(cap 5), beta=`passthrough`. Verified: compiles packaged + test passes.
+
+All accumulator kinds now behave correctly on the packaged path for BOTH languages (passthrough/state/batch/polling; stream = kafka/T-0898 track).

crates/cloacina-macros/src/computation_graph/accumulator_macros.rs

@@ -23,6 +23,50 @@ use quote::{format_ident, quote};
 use syn::parse::{Parse, ParseStream};
 use syn::{Ident, ItemFn, LitStr, Token, Type};
 
+/// Emit the `AccumulatorEntry` inventory submission so a packaged Rust reactor
+/// reports this accumulator's REAL kind + config (CLOACI-T-0896, Rust-cdylib
+/// side) instead of the shell defaulting it to `passthrough`. Cfg-gated so the
+/// path resolves through `cloacina` in library mode and `cloacina-workflow-plugin`
+/// directly in packaged mode (mirrors the reactor/graph inventory submissions).
+/// `config_inserts` is a sequence of `__m.insert(...)` statements building the
+/// String-keyed config map.
+fn accumulator_entry_submission(
+    name_str: &str,
+    accumulator_type: &str,
+    config_inserts: TokenStream,
+) -> TokenStream {
+    quote! {
+        #[cfg(not(feature = "packaged"))]
+        ::cloacina::cloacina_workflow_plugin::inventory::submit! {
+            ::cloacina::cloacina_workflow_plugin::AccumulatorEntry {
+                name: #name_str,
+                accumulator_type: #accumulator_type,
+                config: || {
+                    #[allow(unused_mut)]
+                    let mut __m: ::std::collections::HashMap<String, String> =
+                        ::std::collections::HashMap::new();
+                    #config_inserts
+                    __m
+                },
+            }
+        }
+        #[cfg(feature = "packaged")]
+        ::cloacina_workflow_plugin::inventory::submit! {
+            ::cloacina_workflow_plugin::AccumulatorEntry {
+                name: #name_str,
+                accumulator_type: #accumulator_type,
+                config: || {
+                    #[allow(unused_mut)]
+                    let mut __m: ::std::collections::HashMap<String, String> =
+                        ::std::collections::HashMap::new();
+                    #config_inserts
+                    __m
+                },
+            }
+        }
+    }
+}
+
 /// Parsed args for `#[stream_accumulator(type = "...", topic = "...", ...)]`
 struct StreamAccumulatorArgs {
     backend_type: String,
@@ -105,13 +149,22 @@ pub fn passthrough_accumulator_impl(
     // Get the output type from the return type
     let output_type = extract_return_type(output)?;
 
+    let name_str = fn_name.to_string();
+    let submission = accumulator_entry_submission(&name_str, "passthrough", quote! {});
+
     Ok(quote! {
-        // Keep the original function
+        // Keep the original function (both modes).
         #func
 
-        // Generate the accumulator struct
+        // Embedded-only: the generated `Accumulator` impl uses `cloacina::` paths
+        // and is unused on the packaged path (the host provides the accumulator
+        // runtime; the kind + config travel via the `AccumulatorEntry` inventory).
+        // Gating it out lets the macro compile in lean packaged crates that don't
+        // depend on the `cloacina` umbrella (CLOACI-T-0896).
+        #[cfg(not(feature = "packaged"))]
         pub struct #struct_name;
 
+        #[cfg(not(feature = "packaged"))]
         #[async_trait::async_trait]
         impl cloacina::computation_graph::Accumulator for #struct_name {
             type Output = #output_type;
@@ -123,6 +176,8 @@ pub fn passthrough_accumulator_impl(
 
             // No run() override — socket-only (passthrough)
         }
+
+        #submission
     })
 }
 
@@ -148,6 +203,17 @@ pub fn stream_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Re
         .group
         .unwrap_or_else(|| format!("{}_group", fn_name));
 
+    let name_str = fn_name.to_string();
+    let submission = accumulator_entry_submission(
+        &name_str,
+        "stream",
+        quote! {
+            __m.insert("type".to_string(), #backend_type_str.to_string());
+            __m.insert("topic".to_string(), #topic_str.to_string());
+            __m.insert("group".to_string(), #group_str.to_string());
+        },
+    );
+
     // Check if stateful (has a state parameter)
     let has_state = parsed_args.state_type.is_some();
 
@@ -156,13 +222,16 @@ pub fn stream_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Re
         Ok(quote! {
             #func
 
+            // Embedded-only (see passthrough): unused on the packaged path.
+            #[cfg(not(feature = "packaged"))]
             pub struct #struct_name {
                 pub state: #state_type,
                 pub backend_type: String,
                 pub topic: String,
                 pub group: String,
             }
 
+            #[cfg(not(feature = "packaged"))]
             impl #struct_name {
                 pub fn new(initial_state: #state_type) -> Self {
                     Self {
@@ -174,6 +243,7 @@ pub fn stream_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Re
                 }
             }
 
+            #[cfg(not(feature = "packaged"))]
             #[async_trait::async_trait]
             impl cloacina::computation_graph::Accumulator for #struct_name {
                 type Output = #output_type;
@@ -183,17 +253,22 @@ pub fn stream_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Re
                     Some(#fn_name(parsed, &mut self.state))
                 }
             }
+
+            #submission
         })
     } else {
         Ok(quote! {
             #func
 
+            // Embedded-only (see passthrough): unused on the packaged path.
+            #[cfg(not(feature = "packaged"))]
             pub struct #struct_name {
                 pub backend_type: String,
                 pub topic: String,
                 pub group: String,
             }
 
+            #[cfg(not(feature = "packaged"))]
             impl #struct_name {
                 pub fn new() -> Self {
                     Self {
@@ -204,12 +279,14 @@ pub fn stream_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Re
                 }
             }
 
+            #[cfg(not(feature = "packaged"))]
             impl Default for #struct_name {
                 fn default() -> Self {
                     Self::new()
                 }
             }
 
+            #[cfg(not(feature = "packaged"))]
             #[async_trait::async_trait]
             impl cloacina::computation_graph::Accumulator for #struct_name {
                 type Output = #output_type;
@@ -219,6 +296,8 @@ pub fn stream_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Re
                     Some(#fn_name(parsed))
                 }
             }
+
+            #submission
         })
     }
 }
@@ -303,11 +382,24 @@ pub fn polling_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::R
 
     let interval_ms = parse_duration_ms(&parsed_args.interval_str)?;
 
+    let name_str = fn_name.to_string();
+    // Carry the ORIGINAL interval string ("5s") — the packaged config parser
+    // (`polling_interval_from_config`) re-parses it via `parse_duration_str`.
+    let interval_str = &parsed_args.interval_str;
+    let submission = accumulator_entry_submission(
+        &name_str,
+        "polling",
+        quote! { __m.insert("interval".to_string(), #interval_str.to_string()); },
+    );
+
     Ok(quote! {
         #func
 
+        // Embedded-only (see passthrough): unused on the packaged path.
+        #[cfg(not(feature = "packaged"))]
         pub struct #struct_name;
 
+        #[cfg(not(feature = "packaged"))]
         #[async_trait::async_trait]
         impl cloacina::computation_graph::PollingAccumulator for #struct_name {
             type Output = #inner_type;
@@ -320,6 +412,8 @@ pub fn polling_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::R
                 std::time::Duration::from_millis(#interval_ms)
             }
         }
+
+        #submission
     })
 }
 
@@ -398,11 +492,26 @@ pub fn batch_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Res
         None => quote! { None },
     };
 
+    let name_str = fn_name.to_string();
+    // Carry the ORIGINAL flush_interval string ("5s") + max_buffer_size — the
+    // packaged config parser (`batch_config_from_config`) re-parses them.
+    let flush_interval_str = &parsed_args.flush_interval_str;
+    let mut config_inserts =
+        quote! { __m.insert("flush_interval".to_string(), #flush_interval_str.to_string()); };
+    if let Some(max) = parsed_args.max_buffer_size {
+        config_inserts
+            .extend(quote! { __m.insert("max_buffer_size".to_string(), #max.to_string()); });
+    }
+    let submission = accumulator_entry_submission(&name_str, "batch", config_inserts);
+
     Ok(quote! {
         #func
 
+        // Embedded-only (see passthrough): unused on the packaged path.
+        #[cfg(not(feature = "packaged"))]
         pub struct #struct_name;
 
+        #[cfg(not(feature = "packaged"))]
         #[async_trait::async_trait]
         impl cloacina::computation_graph::BatchAccumulator for #struct_name {
             type Output = #inner_output_type;
@@ -416,6 +525,7 @@ pub fn batch_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Res
             }
         }
 
+        #[cfg(not(feature = "packaged"))]
         impl #struct_name {
             pub fn config() -> cloacina::computation_graph::BatchAccumulatorConfig {
                 cloacina::computation_graph::BatchAccumulatorConfig {
@@ -424,6 +534,8 @@ pub fn batch_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Res
                 }
             }
         }
+
+        #submission
     })
 }
 
@@ -535,12 +647,20 @@ pub fn state_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Res
 
     let capacity = parsed_args.capacity;
     let name_str = fn_name.to_string();
+    let submission = accumulator_entry_submission(
+        &name_str,
+        "state",
+        quote! { __m.insert("capacity".to_string(), #capacity.to_string()); },
+    );
 
     Ok(quote! {
         #func
 
+        // Embedded-only (see passthrough): unused on the packaged path.
+        #[cfg(not(feature = "packaged"))]
         pub struct #struct_name;
 
+        #[cfg(not(feature = "packaged"))]
         impl #struct_name {
             pub fn create() -> cloacina::computation_graph::accumulator::StateAccumulator<#inner_type> {
                 cloacina::computation_graph::accumulator::StateAccumulator::new(#capacity)
@@ -550,6 +670,8 @@ pub fn state_accumulator_impl(args: TokenStream, input: TokenStream) -> syn::Res
                 #name_str
             }
         }
+
+        #submission
     })
 }
 

crates/cloacina-workflow-plugin/src/inventory_entries.rs

@@ -89,6 +89,24 @@ pub struct ReactorEntry {
 }
 inventory::collect!(ReactorEntry);
 
+/// Accumulator entry emitted by the accumulator attribute macros
+/// (`#[state_accumulator]`, `#[batch_accumulator]`, `#[polling_accumulator]`,
+/// `#[stream_accumulator]`, `#[passthrough_accumulator]`). The `package!()`
+/// shell walks `inventory::iter::<AccumulatorEntry>` when building graph/reactor
+/// metadata so a packaged Rust reactor reports each accumulator's REAL kind +
+/// config — instead of the old hardcoded `"passthrough"`, which silently
+/// degraded every declared state/batch/polling accumulator (CLOACI-T-0896,
+/// Rust-cdylib side).
+pub struct AccumulatorEntry {
+    pub name: &'static str,
+    /// "passthrough" | "stream" | "polling" | "batch" | "state".
+    pub accumulator_type: &'static str,
+    /// String-keyed config (e.g. `capacity`, `flush_interval`, `interval`,
+    /// `topic`) — the same channel the manifest override + packaging bridge use.
+    pub config: fn() -> std::collections::HashMap<String, String>,
+}
+inventory::collect!(AccumulatorEntry);
+
 /// Constructor-node entry emitted by `constructor!(...)` inside a `#[workflow]`,
 /// in PACKAGED mode (CLOACI-T-0832). A packaged cdylib cannot link the WASM
 /// constructor loader, so instead of building the node it DECLARES it: the