Skip to main content

uni_plugin/traits/
algorithm.rs

1//! Graph algorithm plugins.
2//!
3//! Two surfaces: [`AlgorithmProvider`] for black-box algorithms (the
4//! existing `uni-algo` library style), and [`GraphView`] — the stable,
5//! read-only topology API a provider obtains from its [`AlgorithmHost`]
6//! via [`AlgorithmHost::project`] to walk the graph without depending on
7//! `uni-store` / `uni-algo` types.
8
9use std::sync::Arc;
10
11use datafusion::execution::SendableRecordBatchStream;
12use futures::future::BoxFuture;
13use uni_common::core::id::Vid;
14
15use crate::errors::FnError;
16
17/// Static signature of an algorithm.
18///
19/// `args` and `slices` are additive, defaulted fields (construct with
20/// `..Default::default()`): existing providers that leave them empty keep the
21/// legacy untyped `config_json` contract unchanged. A provider that declares
22/// `args` opts into host-side arity/type validation before it runs (proposal
23/// §4.6 / decision D7); a provider that declares `slices` opts into load-time
24/// capability-slice version negotiation (proposal §4.3 / decision D6).
25#[derive(Clone, Debug, Default)]
26pub struct AlgorithmSignature {
27    /// Output column schema.
28    pub output_fields: Vec<arrow_schema::Field>,
29    /// Markdown docs.
30    pub docs: String,
31    /// Declared positional arguments, in call order.
32    ///
33    /// Empty (the default) preserves the legacy behavior: arguments arrive as a
34    /// raw positional `config_json` array the provider parses itself. When
35    /// non-empty, the host validates arity and coerces each positional argument
36    /// against the declared [`NamedArgType`](crate::traits::procedure::NamedArgType) before the provider runs, filling
37    /// omitted trailing arguments from their declared defaults.
38    pub args: Vec<crate::traits::procedure::NamedArgType>,
39    /// Required capability slices, checked at load time.
40    ///
41    /// Empty (the default) means the algorithm targets only the always-present
42    /// `graph-compute@1` surface. A declared [`SliceReq`] whose version the host
43    /// does not provide fails the load with a clear error (`0x86A`) rather than a
44    /// mysterious runtime "unknown kernel op" trap.
45    pub slices: Vec<SliceReq>,
46    /// Whether a `CALL` of this algorithm may be planned as a first-class
47    /// DataFusion `ExecutionPlan` node (the vectorized path), rather than only
48    /// through the row-based interpreter (proposal §6, DF-3).
49    ///
50    /// `false` (the default) preserves the row path. A provider sets this `true`
51    /// to *declare* it composes correctly as a leaf/source plan node — its `run`
52    /// returns a well-formed `RecordBatch` stream matching `output_fields`, and it
53    /// consumes MATCH-bound arguments via `outer_values` rather than a child plan.
54    /// This replaces the previous name-prefix allowlist (`uni.algo.*`): eligibility
55    /// is now **registration-driven**, so a third-party `myco.algo.*` provider that
56    /// declares `df_composable` is a first-class plan node like the first-party
57    /// ones, and a `uni.algo.`-named provider that does *not* declare it no longer
58    /// gets the DF path by prefix alone.
59    pub df_composable: bool,
60}
61
62/// A required capability-slice version an algorithm declares in its signature.
63///
64/// The host checks each requirement against the slices it actually implements
65/// (today only `graph-compute@1`) when the algorithm loads, refusing a mismatch
66/// up front (proposal §4.3 / decision D6). Adding a slice or bumping a version is
67/// a forward-compatible, additive change: an algorithm that declares no slices is
68/// grandfathered onto the base surface.
69#[derive(Clone, Debug, PartialEq, Eq)]
70pub struct SliceReq {
71    /// The capability-slice name, e.g. `"graph-compute"`.
72    pub slice: smol_str::SmolStr,
73    /// The minimum slice version the algorithm requires.
74    pub version: u16,
75}
76
77/// The capability slices this host implements, for load-time negotiation.
78///
79/// A guest algorithm's declared [`SliceReq`]s are checked against this table
80/// when it loads. The host implements `graph-compute@1` (the coarse read-only
81/// kernels over a projection) and `graph-arena@1` (mutable session-local
82/// structure, proposal §5.1). A future slice is added here in lockstep with its
83/// kernels so negotiation stays a pure lookup (proposal §4.3 / §10).
84pub const HOST_CAPABILITY_SLICES: &[(&str, u16)] = &[("graph-compute", 1), ("graph-arena", 1)];
85
86impl AlgorithmSignature {
87    /// Validates the declared capability slices against `host_slices`.
88    ///
89    /// Each requirement must be met by a host slice of the same name whose
90    /// version is at least the requested one. Pass [`HOST_CAPABILITY_SLICES`] for
91    /// the production surface.
92    ///
93    /// # Errors
94    /// Returns `0x86A` (`SliceVersionMismatch`) naming the first requirement the
95    /// host cannot satisfy (proposal §4.3 / §12, decision D6).
96    pub fn check_slices(&self, host_slices: &[(&str, u16)]) -> Result<(), FnError> {
97        for req in &self.slices {
98            let satisfied = host_slices
99                .iter()
100                .any(|(name, ver)| *name == req.slice.as_str() && *ver >= req.version);
101            if !satisfied {
102                return Err(FnError::new(
103                    0x86A,
104                    format!(
105                        "algorithm requires capability slice `{}@{}` the host does not provide",
106                        req.slice, req.version
107                    ),
108                ));
109            }
110        }
111        Ok(())
112    }
113
114    /// Validates and normalizes a positional `config_json` array against `args`.
115    ///
116    /// When `args` is empty this is a no-op returning `config_json` unchanged, so
117    /// providers on the legacy untyped contract are unaffected. Otherwise it
118    /// parses the positional JSON array and, per declared argument: rejects a
119    /// present value whose JSON kind is incompatible with the declared
120    /// [`ArgType`](crate::traits::scalar::ArgType), errors on a missing argument
121    /// that has no default, and appends the declared default for an omitted
122    /// trailing argument. Extra positional arguments beyond the declared arity
123    /// are rejected. The returned JSON array is what the provider then parses, so
124    /// it observes defaults already filled in.
125    ///
126    /// # Errors
127    /// Returns `0x86E` (argument arity/type violation) with a message naming the
128    /// offending argument (proposal §4.6, decision D7).
129    pub fn coerce_config_json(&self, config_json: &str) -> Result<String, FnError> {
130        use crate::traits::scalar::ArgType;
131
132        if self.args.is_empty() {
133            return Ok(config_json.to_owned());
134        }
135        let mut provided: Vec<serde_json::Value> = if config_json.trim().is_empty() {
136            Vec::new()
137        } else {
138            serde_json::from_str(config_json)
139                .map_err(|e| FnError::new(0x86E, format!("bad positional config json: {e}")))?
140        };
141        if provided.len() > self.args.len() {
142            return Err(FnError::new(
143                0x86E,
144                format!(
145                    "too many arguments: got {}, expected at most {}",
146                    provided.len(),
147                    self.args.len()
148                ),
149            ));
150        }
151        let mut out = Vec::with_capacity(self.args.len());
152        for (i, arg) in self.args.iter().enumerate() {
153            match provided.get_mut(i) {
154                Some(value) => {
155                    let value = std::mem::replace(value, serde_json::Value::Null);
156                    // A `CypherValue` argument is opaque and accepts any JSON.
157                    if !matches!(arg.ty, ArgType::CypherValue)
158                        && !json_matches_argtype(&value, &arg.ty)
159                    {
160                        return Err(FnError::new(
161                            0x86E,
162                            format!("argument `{}` (position {i}) has the wrong type", arg.name),
163                        ));
164                    }
165                    out.push(value);
166                }
167                None => match &arg.default {
168                    Some(default) => out.push(scalar_default_to_json(default)),
169                    None => {
170                        return Err(FnError::new(
171                            0x86E,
172                            format!("missing required argument `{}` (position {i})", arg.name),
173                        ));
174                    }
175                },
176            }
177        }
178        serde_json::to_string(&out)
179            .map_err(|e| FnError::new(0x86E, format!("re-encoding coerced config: {e}")))
180    }
181}
182
183/// Whether a JSON value is compatible with a declared primitive/vector arg type.
184fn json_matches_argtype(value: &serde_json::Value, ty: &crate::traits::scalar::ArgType) -> bool {
185    use arrow_schema::DataType;
186
187    use crate::traits::scalar::ArgType;
188    match ty {
189        ArgType::CypherValue => true,
190        ArgType::Vector { .. } => value.is_array(),
191        ArgType::Variadic(inner) => json_matches_argtype(value, inner),
192        ArgType::Primitive(dt) => match dt {
193            DataType::Boolean => value.is_boolean(),
194            DataType::Utf8 | DataType::LargeUtf8 => value.is_string(),
195            DataType::Float16 | DataType::Float32 | DataType::Float64 => value.is_number(),
196            d if d.is_integer() => value.is_i64() || value.is_u64(),
197            // Unknown/opaque primitive: don't reject, defer to the provider.
198            _ => true,
199        },
200    }
201}
202
203/// Renders a declared [`ScalarValue`](datafusion::scalar::ScalarValue) default as
204/// JSON to append for an omitted trailing argument.
205fn scalar_default_to_json(default: &datafusion::scalar::ScalarValue) -> serde_json::Value {
206    use datafusion::scalar::ScalarValue;
207
208    match default {
209        ScalarValue::Null => serde_json::Value::Null,
210        ScalarValue::Boolean(Some(b)) => serde_json::Value::Bool(*b),
211        ScalarValue::Float32(Some(x)) => serde_json::json!(*x),
212        ScalarValue::Float64(Some(x)) => serde_json::json!(*x),
213        ScalarValue::Int8(Some(x)) => serde_json::json!(*x),
214        ScalarValue::Int16(Some(x)) => serde_json::json!(*x),
215        ScalarValue::Int32(Some(x)) => serde_json::json!(*x),
216        ScalarValue::Int64(Some(x)) => serde_json::json!(*x),
217        ScalarValue::UInt8(Some(x)) => serde_json::json!(*x),
218        ScalarValue::UInt16(Some(x)) => serde_json::json!(*x),
219        ScalarValue::UInt32(Some(x)) => serde_json::json!(*x),
220        ScalarValue::UInt64(Some(x)) => serde_json::json!(*x),
221        ScalarValue::Utf8(Some(s)) | ScalarValue::LargeUtf8(Some(s)) => {
222            serde_json::Value::String(s.clone())
223        }
224        // Any other/None scalar defaults to JSON null.
225        _ => serde_json::Value::Null,
226    }
227}
228
229/// Per-invocation context passed to an [`AlgorithmProvider`].
230///
231/// `host` is an opaque [`AlgorithmHost`] callback the host populates
232/// when invoking the algorithm. Algorithms that need a concrete
233/// graph-projection / storage handle downcast through `host` rather
234/// than depend on `uni-store` / `uni-algo` types directly — this keeps
235/// `uni-plugin` free of upward dependencies.
236#[non_exhaustive]
237pub struct AlgorithmContext<'a> {
238    /// JSON-serialized algorithm configuration.
239    pub config_json: &'a str,
240    /// Optional opaque host handle. `None` when no host is bound — the
241    /// algorithm may fall back to a config-only path or surface an
242    /// `Unbound` error.
243    pub host: Option<&'a dyn AlgorithmHost>,
244}
245
246impl std::fmt::Debug for AlgorithmContext<'_> {
247    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
248        f.debug_struct("AlgorithmContext")
249            .field("config_json", &self.config_json)
250            .field("host_bound", &self.host.is_some())
251            .finish()
252    }
253}
254
255impl<'a> AlgorithmContext<'a> {
256    /// Construct an `AlgorithmContext` with no host bound.
257    #[must_use]
258    pub fn new(config_json: &'a str) -> Self {
259        Self {
260            config_json,
261            host: None,
262        }
263    }
264
265    /// Attach a host handle.
266    #[must_use]
267    pub fn with_host(mut self, host: &'a dyn AlgorithmHost) -> Self {
268        self.host = Some(host);
269        self
270    }
271}
272
273/// Host callback surfacing graph access to plugin algorithms.
274///
275/// A provider's [`AlgorithmProvider::run`] receives an [`AlgorithmHost`]
276/// through its [`AlgorithmContext`] and calls [`AlgorithmHost::project`]
277/// to materialize a [`GraphView`] over the requested subgraph. Hosts
278/// (e.g. `uni-plugin-builtin`) implement `project` by building a
279/// projection from their `StorageManager` / `L0Manager`; the
280/// [`AlgorithmHost::as_any`] downcast hook remains for hosts that expose
281/// additional concrete state. This keeps `uni-plugin` free of upward
282/// dependencies on `uni-store` / `uni-algo`.
283pub trait AlgorithmHost: Send + Sync {
284    /// Downcast hook — bridges implement this to expose the concrete
285    /// host type.
286    fn as_any(&self) -> &dyn std::any::Any;
287
288    /// Materialize a read-only [`GraphView`] over the subgraph named by
289    /// `spec`.
290    ///
291    /// The returned future is `'static` (owns its inputs) so a provider
292    /// can move it into the stream it returns from the synchronous
293    /// [`AlgorithmProvider::run`] and `.await` it there. The default
294    /// implementation reports that the host offers no graph access;
295    /// graph-capable hosts override it.
296    ///
297    /// # Errors
298    ///
299    /// Returns [`FnError`] if the host offers no graph access, the
300    /// caller lacks the required capability (e.g. `HostQuery`), or the
301    /// projection cannot be built.
302    fn project(
303        &self,
304        spec: &GraphProjectionSpec,
305    ) -> BoxFuture<'static, Result<Arc<dyn GraphView>, FnError>> {
306        let _ = spec;
307        Box::pin(async {
308            Err(FnError::new(
309                0x805,
310                "AlgorithmHost: project() is not supported by this host",
311            ))
312        })
313    }
314}
315
316/// Selects which subgraph an [`AlgorithmHost::project`] call materializes.
317///
318/// Naming `node_labels` / `edge_types` scopes the projection to exactly those.
319/// Leaving BOTH empty means "the whole graph", but that is only honored when
320/// [`Self::project_all`] is `true` — otherwise the projection fails loud (G9),
321/// so an unscoped projection can never *silently* pull in unrelated labels (e.g.
322/// a coexisting search tree) and corrupt an index-keyed kernel. `weight_property`
323/// names an edge property to expose through [`GraphView::out_weight`];
324/// `include_reverse` requests inbound adjacency ([`GraphView::in_neighbors`]).
325#[derive(Clone, Debug, Default)]
326pub struct GraphProjectionSpec {
327    /// Vertex labels to include; empty (with [`Self::project_all`]) selects every label.
328    pub node_labels: Vec<String>,
329    /// Edge types to include; empty (with [`Self::project_all`]) selects every type.
330    pub edge_types: Vec<String>,
331    /// Edge property surfaced as the traversal weight, if any.
332    pub weight_property: Option<String>,
333    /// Whether to also build inbound adjacency.
334    pub include_reverse: bool,
335    /// Vertex properties to materialize into per-vertex `[V]` tensors the guest
336    /// reads by name via `gc.node_property` (issue #151).
337    pub node_properties: Vec<String>,
338    /// Edge properties to materialize into per-edge `[E]` tensors the guest
339    /// reads by name via `gc.edge_property` (issue #151).
340    pub edge_properties: Vec<String>,
341    /// Deliberate opt-in to projecting the *whole* graph when neither
342    /// `node_labels` nor `edge_types` is named (G9). With both empty and this
343    /// `false`, [`AlgorithmHost::project`] fails loud instead of silently
344    /// pulling in every schema label/edge-type. First-party providers set this
345    /// to preserve the ergonomic whole-graph default; guest loaders must pass
346    /// `projectAll: true` in their config to project everything on purpose.
347    pub project_all: bool,
348}
349
350impl GraphProjectionSpec {
351    /// Parses the Native-mode projection knobs from a guest/procedure config
352    /// object. This is the single source of truth for `nodeLabels` /
353    /// `edgeTypes` (accepting the `relationshipTypes` alias) / `weightProperty`
354    /// / `includeReverse` — the native providers and all four guest loaders
355    /// funnel through it so the knob names cannot drift.
356    ///
357    /// `includeReverse` defaults to `true` (inbound adjacency is built unless
358    /// the caller opts out), matching the graphRef contract in `uni-algo` and
359    /// keeping In-direction kernels (WCC / k-core / HITS) working. Unknown keys
360    /// are ignored and malformed values fall back to the field default: this is
361    /// a best-effort projection hint, not a strict schema.
362    #[must_use]
363    pub fn from_config_object(cfg: &serde_json::Map<String, serde_json::Value>) -> Self {
364        fn string_array(v: &serde_json::Value) -> Vec<String> {
365            v.as_array()
366                .map(|arr| {
367                    arr.iter()
368                        .filter_map(|s| s.as_str().map(str::to_owned))
369                        .collect()
370                })
371                .unwrap_or_default()
372        }
373
374        let node_labels = cfg.get("nodeLabels").map(string_array).unwrap_or_default();
375        // `relationshipTypes` is the openCypher-flavored alias the native
376        // procedures already accept as a synonym for `edgeTypes`.
377        let edge_types = cfg
378            .get("edgeTypes")
379            .or_else(|| cfg.get("relationshipTypes"))
380            .map(string_array)
381            .unwrap_or_default();
382        let weight_property = cfg
383            .get("weightProperty")
384            .and_then(serde_json::Value::as_str)
385            .map(str::to_owned);
386        let include_reverse = cfg
387            .get("includeReverse")
388            .and_then(serde_json::Value::as_bool)
389            .unwrap_or(true);
390        let node_properties = cfg
391            .get("nodeProperties")
392            .map(string_array)
393            .unwrap_or_default();
394        let edge_properties = cfg
395            .get("edgeProperties")
396            .map(string_array)
397            .unwrap_or_default();
398        // G9: explicit opt-in to a whole-graph projection. Absent/false means an
399        // unscoped projection is rejected fail-loud by the bridge.
400        let project_all = cfg
401            .get("projectAll")
402            .and_then(serde_json::Value::as_bool)
403            .unwrap_or(false);
404
405        Self {
406            node_labels,
407            edge_types,
408            weight_property,
409            include_reverse,
410            node_properties,
411            edge_properties,
412            project_all,
413        }
414    }
415
416    /// Keys that mark a trailing CALL argument as a projection-config object
417    /// (a "graphRef") rather than a guest algorithm argument. Covers the
418    /// Native knobs, the P2 property-tensor knobs, and the P3 Cypher/Named
419    /// graphRef knobs so the trailing object is stripped from the guest's
420    /// arguments consistently across every projection mode.
421    pub const CONFIG_KEYS: &'static [&'static str] = &[
422        "nodeLabels",
423        "edgeTypes",
424        "relationshipTypes",
425        "weightProperty",
426        "includeReverse",
427        "nodeProperties",
428        "edgeProperties",
429        "projectAll",
430        "nodeQuery",
431        "edgeQuery",
432        "weightColumn",
433        "name",
434        "scopes",
435    ];
436
437    /// Keys that mark a config object as a **Cypher/Named** `graphRef` rather
438    /// than a Native label/edge-type scoping object.
439    ///
440    /// Single-sourced so the query layer's graphRef sniffing and the per-scope
441    /// parsing below cannot drift: a scope routed to the Native storage scan
442    /// when it names a Cypher query would silently project the wrong graph.
443    pub const QUERY_CONFIG_KEYS: &'static [&'static str] = &["nodeQuery", "edgeQuery", "name"];
444
445    /// Whether a config object names a Cypher/Named projection.
446    #[must_use]
447    pub fn is_query_graph_ref(cfg: &serde_json::Map<String, serde_json::Value>) -> bool {
448        Self::QUERY_CONFIG_KEYS.iter().any(|k| cfg.contains_key(*k))
449    }
450
451    /// Parses the `scopes` map into pre-declared named projections.
452    ///
453    /// A guest that needs more than one view of the store declares them at the
454    /// CALL site rather than projecting on demand, because projection is the one
455    /// thing a guest must not be able to trigger in a loop:
456    ///
457    /// ```cypher
458    /// CALL myplugin.compare([], {
459    ///   nodeLabels: ['Cell'], edgeTypes: ['ADJ'],
460    ///   scopes: {
461    ///     agg:  {nodeLabels: ['Cell'], edgeTypes: ['AGGREGATES']},
462    ///     flow: {nodeQuery: 'MATCH (c:Cell) RETURN id(c) AS id'}
463    ///   }
464    /// })
465    /// ```
466    ///
467    /// The outer object stays the *primary* projection — the one `emit` keys its
468    /// `nodeId` column to. Each scope value is parsed by the same
469    /// [`Self::from_config_object`], so every Native knob works per scope; a
470    /// scope bearing a Cypher/Named key is carried through verbatim for the
471    /// resolver instead.
472    ///
473    /// # Errors
474    /// Returns a message naming the offending scope when `scopes` is not an
475    /// object, a scope name is empty, a scope value is not an object, or a scope
476    /// is named `graph` (which would shadow the primary handle's own accessor).
477    pub fn scopes_from_config_object(
478        cfg: &serde_json::Map<String, serde_json::Value>,
479    ) -> Result<Vec<GraphScopeSpec>, String> {
480        let Some(raw) = cfg.get("scopes") else {
481            return Ok(Vec::new());
482        };
483        let map = raw
484            .as_object()
485            .ok_or_else(|| "`scopes` must be an object of {name: projection-config}".to_string())?;
486        let mut out = Vec::with_capacity(map.len());
487        for (name, value) in map {
488            if name.is_empty() {
489                return Err("a scope name must not be empty".to_string());
490            }
491            if name == "graph" {
492                return Err(
493                    "`graph` is not a valid scope name: it is the primary projection, \
494                     reached with `gc.graph()` rather than `gc.graph_named(..)`"
495                        .to_string(),
496                );
497            }
498            let obj = value.as_object().ok_or_else(|| {
499                format!("scope `{name}` must be a projection-config object, got {value}")
500            })?;
501            let graph_ref = Self::is_query_graph_ref(obj).then(|| value.clone());
502            out.push(GraphScopeSpec {
503                name: name.clone(),
504                spec: Self::from_config_object(obj),
505                graph_ref,
506            });
507        }
508        Ok(out)
509    }
510
511    /// Rejects a `scopes` map on an algorithm that does not consume one.
512    ///
513    /// `scopes` joined [`Self::CONFIG_KEYS`] so a scopes-only object is stripped
514    /// from the guest's positional arguments. That stripping applies to *every*
515    /// algorithm, but only the guest loader adapters build the declared
516    /// projections — so without this a first-party provider would accept a
517    /// `scopes` map, project nothing, and run as if it had never been asked.
518    /// Silently ignoring a projection the caller asked for is the same failure
519    /// the unscoped-projection change (G9) made loud; this keeps it loud.
520    ///
521    /// # Errors
522    /// Returns `0x86E` naming `algorithm` when `cfg` carries a non-empty `scopes`.
523    pub fn reject_scopes(
524        cfg: &serde_json::Map<String, serde_json::Value>,
525        algorithm: &str,
526    ) -> Result<(), FnError> {
527        let has_scopes = cfg
528            .get("scopes")
529            .and_then(serde_json::Value::as_object)
530            .is_some_and(|m| !m.is_empty());
531        if !has_scopes {
532            return Ok(());
533        }
534        Err(FnError::new(
535            0x86E,
536            format!(
537                "{algorithm} does not take named `scopes`: it runs a fixed algorithm over \
538                 one projection. Named scopes are a guest-authored-algorithm feature -- \
539                 the guest is what decides which scope to read."
540            ),
541        ))
542    }
543
544    /// Like [`Self::take_from_args`] but also returns the raw config object.
545    ///
546    /// [`Self::take_from_args`] discards the object after parsing, which is fine
547    /// for the Native knobs but loses `scopes` (whose values must be re-parsed
548    /// per scope, and whose Cypher entries must survive verbatim).
549    #[must_use]
550    pub fn take_config_from_args(
551        args: &mut Vec<serde_json::Value>,
552    ) -> Option<serde_json::Map<String, serde_json::Value>> {
553        let is_config = args
554            .last()
555            .and_then(serde_json::Value::as_object)
556            .is_some_and(|o| Self::CONFIG_KEYS.iter().any(|k| o.contains_key(*k)));
557        if !is_config {
558            return None;
559        }
560        match args.pop() {
561            Some(serde_json::Value::Object(cfg)) => Some(cfg),
562            _ => None,
563        }
564    }
565
566    /// If the last element of `args` is a JSON object bearing at least one
567    /// [`Self::CONFIG_KEYS`] key, removes it from `args` and returns the parsed
568    /// Native spec; otherwise leaves `args` untouched and returns `None`.
569    ///
570    /// The Native-spec half of the "trailing object is the projection config"
571    /// convention. The guest loaders now go through
572    /// [`ProjectionPlan`](../../../uni_plugin_builtin/algorithms/bridge/struct.ProjectionPlan.html),
573    /// which needs the raw object to parse `scopes`; this remains for callers
574    /// that want only the Native knobs. Both share
575    /// [`Self::take_config_from_args`], so the recognition rule cannot drift.
576    /// In P3 Cypher/Named
577    /// mode the returned Native spec is empty (the query/name keys are unknown
578    /// to [`Self::from_config_object`]) and is ignored by the bridge in favor of
579    /// the pre-built projection — but the object is still stripped here so it
580    /// never reaches the guest function.
581    #[must_use]
582    pub fn take_from_args(args: &mut Vec<serde_json::Value>) -> Option<Self> {
583        Self::take_config_from_args(args).map(|cfg| Self::from_config_object(&cfg))
584    }
585}
586
587/// One pre-declared named projection from a CALL-site `scopes` map.
588///
589/// Named scopes are how a guest algorithm reaches more than one view of the
590/// store. They are declared at the call site and built by the host *before* the
591/// guest runs, which is the point: a guest that could project on demand could
592/// project in a loop, and projection is `O(V+E)` storage work that the native
593/// work meter does not govern.
594#[derive(Clone, Debug)]
595pub struct GraphScopeSpec {
596    /// The name the guest passes to `graph_named`.
597    pub name: String,
598    /// Native knobs for this scope, parsed by [`GraphProjectionSpec::from_config_object`].
599    pub spec: GraphProjectionSpec,
600    /// The raw scope object when it names a Cypher/Named projection, to be
601    /// resolved through the host's injected resolver rather than scanned.
602    pub graph_ref: Option<serde_json::Value>,
603}
604
605/// Every graph-projection knob, enumerated so the guest/native surface contract
606/// is checked at compile time: adding a knob forces a classification in
607/// [`ProjectionKnob::reach`] and a key in [`ProjectionKnob::config_key`] (both
608/// wildcard-free `match`es), mirroring the capability
609/// `every_variant_classified_exactly_once` exhaustiveness test. This is the
610/// anti-drift guard for the guest/native gap issue #151 exposed. See
611/// `docs/proposals/graphcompute_projection_parity_2026-07-19.md` §4.
612#[derive(Clone, Copy, Debug, PartialEq, Eq)]
613pub enum ProjectionKnob {
614    /// Scope to these vertex labels (`nodeLabels`).
615    NodeLabels,
616    /// Scope to these edge types (`edgeTypes` / `relationshipTypes`).
617    EdgeTypes,
618    /// Bind an edge property as the traversal weight (`weightProperty`).
619    WeightProperty,
620    /// Also build inbound adjacency (`includeReverse`).
621    IncludeReverse,
622    /// Materialize per-vertex property tensors (`nodeProperties`).
623    NodeProperties,
624    /// Materialize per-edge property tensors (`edgeProperties`).
625    EdgeProperties,
626    /// Cypher-mode node selection query (`nodeQuery`).
627    CypherNodeQuery,
628    /// Cypher-mode edge selection query (`edgeQuery`).
629    CypherEdgeQuery,
630    /// Cypher-mode weight column (`weightColumn`).
631    CypherWeightColumn,
632    /// Named pre-registered projection (`name`).
633    NamedGraph,
634}
635
636/// How a projection knob is reachable from a guest algorithm.
637#[derive(Clone, Copy, Debug, PartialEq, Eq)]
638pub enum KnobReach {
639    /// Parsed by the shared [`GraphProjectionSpec::from_config_object`] — the
640    /// Native + property-tensor knobs.
641    GuestNative,
642    /// Resolved by the uni-query projection seam from a Cypher/Named graphRef.
643    GuestQuerySeam,
644    /// Host/native-only — not expressible by a guest (none today; reserved for
645    /// future knobs such as orientation or parallel-edge aggregation).
646    HostOnly,
647}
648
649impl ProjectionKnob {
650    /// Every knob, the single source of truth the exhaustiveness test iterates.
651    pub const ALL: &'static [ProjectionKnob] = &[
652        ProjectionKnob::NodeLabels,
653        ProjectionKnob::EdgeTypes,
654        ProjectionKnob::WeightProperty,
655        ProjectionKnob::IncludeReverse,
656        ProjectionKnob::NodeProperties,
657        ProjectionKnob::EdgeProperties,
658        ProjectionKnob::CypherNodeQuery,
659        ProjectionKnob::CypherEdgeQuery,
660        ProjectionKnob::CypherWeightColumn,
661        ProjectionKnob::NamedGraph,
662    ];
663
664    /// The graphRef config key for this knob. Wildcard-free: a new variant fails
665    /// to compile here until it is given a key.
666    #[must_use]
667    pub fn config_key(self) -> &'static str {
668        match self {
669            ProjectionKnob::NodeLabels => "nodeLabels",
670            ProjectionKnob::EdgeTypes => "edgeTypes",
671            ProjectionKnob::WeightProperty => "weightProperty",
672            ProjectionKnob::IncludeReverse => "includeReverse",
673            ProjectionKnob::NodeProperties => "nodeProperties",
674            ProjectionKnob::EdgeProperties => "edgeProperties",
675            ProjectionKnob::CypherNodeQuery => "nodeQuery",
676            ProjectionKnob::CypherEdgeQuery => "edgeQuery",
677            ProjectionKnob::CypherWeightColumn => "weightColumn",
678            ProjectionKnob::NamedGraph => "name",
679        }
680    }
681
682    /// How a guest reaches this knob. Wildcard-free: a new variant fails to
683    /// compile here until it is classified.
684    #[must_use]
685    pub fn reach(self) -> KnobReach {
686        match self {
687            ProjectionKnob::NodeLabels
688            | ProjectionKnob::EdgeTypes
689            | ProjectionKnob::WeightProperty
690            | ProjectionKnob::IncludeReverse
691            | ProjectionKnob::NodeProperties
692            | ProjectionKnob::EdgeProperties => KnobReach::GuestNative,
693            ProjectionKnob::CypherNodeQuery
694            | ProjectionKnob::CypherEdgeQuery
695            | ProjectionKnob::CypherWeightColumn
696            | ProjectionKnob::NamedGraph => KnobReach::GuestQuerySeam,
697        }
698    }
699}
700
701#[cfg(test)]
702mod projection_knob_contract {
703    use super::{GraphProjectionSpec, KnobReach, ProjectionKnob};
704
705    #[test]
706    fn every_projection_knob_is_classified_and_keyed() {
707        // A knob added to the enum makes `reach`/`config_key` non-exhaustive
708        // (compile error) and must be appended to `ALL`. Keys must be unique and
709        // recognized as graphRef markers so the adapters strip them from guest
710        // args.
711        let mut keys = std::collections::HashSet::new();
712        for knob in ProjectionKnob::ALL {
713            let key = knob.config_key();
714            assert!(keys.insert(key), "duplicate projection config key {key}");
715            assert!(
716                GraphProjectionSpec::CONFIG_KEYS.contains(&key),
717                "{key} missing from GraphProjectionSpec::CONFIG_KEYS"
718            );
719            let _ = knob.reach(); // total by construction
720        }
721    }
722
723    #[test]
724    fn guest_native_knobs_round_trip_through_the_shared_parser() {
725        // Every GuestNative knob must actually be honored by from_config_object,
726        // so a knob can't be declared guest-reachable yet silently unparsed —
727        // the precise failure that produced issue #151.
728        for &knob in ProjectionKnob::ALL {
729            if knob.reach() != KnobReach::GuestNative {
730                continue;
731            }
732            let key = knob.config_key();
733            let sample = match knob {
734                ProjectionKnob::IncludeReverse => serde_json::json!(false),
735                ProjectionKnob::WeightProperty => serde_json::json!("w"),
736                _ => serde_json::json!(["X"]),
737            };
738            let mut obj = serde_json::Map::new();
739            obj.insert(key.to_string(), sample);
740            let spec = GraphProjectionSpec::from_config_object(&obj);
741            let honored = match knob {
742                ProjectionKnob::NodeLabels => spec.node_labels == ["X"],
743                ProjectionKnob::EdgeTypes => spec.edge_types == ["X"],
744                ProjectionKnob::WeightProperty => spec.weight_property.as_deref() == Some("w"),
745                ProjectionKnob::IncludeReverse => !spec.include_reverse,
746                ProjectionKnob::NodeProperties => spec.node_properties == ["X"],
747                ProjectionKnob::EdgeProperties => spec.edge_properties == ["X"],
748                _ => unreachable!("only GuestNative knobs reach here"),
749            };
750            assert!(honored, "from_config_object did not honor `{key}`");
751        }
752    }
753}
754
755/// Stable, read-only topology view handed to a plugin algorithm.
756///
757/// Vertices are addressed by dense `u32` slots (`0..vertex_count`);
758/// [`GraphView::to_vid`] / [`GraphView::to_slot`] translate to and from
759/// external [`Vid`]s at the boundary. Neighbor accessors return neighbor
760/// *slots*, not vids. A `GraphView` reflects the subgraph named by the
761/// [`GraphProjectionSpec`] that produced it and does not observe later
762/// writes.
763///
764/// # Panics
765///
766/// [`GraphView::out_weight`] panics unless [`GraphView::has_weights`] is
767/// `true`, and [`GraphView::in_neighbors`] / [`GraphView::in_degree`]
768/// panic unless [`GraphView::has_reverse`] is `true`. Guard with those
769/// predicates before calling.
770pub trait GraphView: Send + Sync {
771    /// Number of vertices; valid slots are `0..vertex_count`.
772    fn vertex_count(&self) -> usize;
773
774    /// Total number of outbound edges.
775    fn edge_count(&self) -> usize;
776
777    /// Outbound neighbor slots of `slot`.
778    fn out_neighbors(&self, slot: u32) -> &[u32];
779
780    /// Number of outbound edges from `slot`.
781    fn out_degree(&self, slot: u32) -> u32;
782
783    /// Inbound neighbor slots of `slot`.
784    ///
785    /// # Panics
786    ///
787    /// Panics unless [`GraphView::has_reverse`] is `true`.
788    fn in_neighbors(&self, slot: u32) -> &[u32];
789
790    /// Number of inbound edges into `slot`.
791    ///
792    /// # Panics
793    ///
794    /// Panics unless [`GraphView::has_reverse`] is `true`.
795    fn in_degree(&self, slot: u32) -> u32;
796
797    /// Whether inbound adjacency is available.
798    fn has_reverse(&self) -> bool;
799
800    /// Weight of the `edge_idx`-th outbound edge of `slot`.
801    ///
802    /// `edge_idx` indexes into [`GraphView::out_neighbors`] of `slot`.
803    ///
804    /// # Panics
805    ///
806    /// Panics unless [`GraphView::has_weights`] is `true`.
807    fn out_weight(&self, slot: u32, edge_idx: usize) -> f64;
808
809    /// Whether edge weights are available.
810    fn has_weights(&self) -> bool;
811
812    /// Translate a dense slot to its external [`Vid`].
813    fn to_vid(&self, slot: u32) -> Vid;
814
815    /// Translate an external [`Vid`] to its dense slot, if present.
816    fn to_slot(&self, vid: Vid) -> Option<u32>;
817
818    /// Iterate over every `(slot, vid)` pair in the view.
819    fn vertices(&self) -> Box<dyn Iterator<Item = (u32, Vid)> + '_>;
820}
821
822/// A black-box graph algorithm.
823///
824/// The trait is intentionally minimal: a signature describing the output,
825/// plus a `run` method returning a streaming `RecordBatch` sequence. The
826/// algorithm is responsible for fetching graph data via host APIs (out of
827/// scope of this trait — `uni-algo` will provide a `GraphView` abstraction
828/// the host adapter passes via `AlgorithmContext` once those APIs are
829/// available).
830pub trait AlgorithmProvider: Send + Sync {
831    /// Static signature.
832    fn signature(&self) -> &AlgorithmSignature;
833
834    /// Execute the algorithm.
835    ///
836    /// # Errors
837    ///
838    /// Returns [`FnError`] if the algorithm cannot be started; per-batch
839    /// failures are signaled via `Err` items in the returned stream.
840    fn run(&self, ctx: AlgorithmContext<'_>) -> Result<SendableRecordBatchStream, FnError>;
841}
842
843#[cfg(test)]
844mod tests {
845    use arrow_schema::DataType;
846    use datafusion::scalar::ScalarValue;
847
848    use super::{AlgorithmSignature, HOST_CAPABILITY_SLICES, SliceReq};
849    use crate::traits::procedure::NamedArgType;
850    use crate::traits::scalar::ArgType;
851
852    fn sig_with(args: Vec<NamedArgType>, slices: Vec<SliceReq>) -> AlgorithmSignature {
853        AlgorithmSignature {
854            args,
855            slices,
856            ..Default::default()
857        }
858    }
859
860    fn arg(name: &str, ty: ArgType, default: Option<ScalarValue>) -> NamedArgType {
861        NamedArgType {
862            name: name.into(),
863            ty,
864            default,
865            doc: String::new(),
866        }
867    }
868
869    fn cfg(json: &str) -> serde_json::Map<String, serde_json::Value> {
870        match serde_json::from_str(json).expect("valid json") {
871            serde_json::Value::Object(o) => o,
872            other => panic!("expected an object, got {other}"),
873        }
874    }
875
876    /// A `scopes`-only object must still be recognised as the projection config.
877    ///
878    /// If `scopes` were missing from `CONFIG_KEYS`, this object would not be
879    /// stripped and would arrive at the guest as a positional argument — a
880    /// silent arity shift rather than an error.
881    #[test]
882    fn a_scopes_only_object_is_recognised_as_the_projection_config() {
883        let mut args: Vec<serde_json::Value> =
884            serde_json::from_str(r#"[1, {"scopes": {"agg": {"nodeLabels": ["N"]}}}]"#)
885                .expect("valid json");
886        let spec = super::GraphProjectionSpec::take_from_args(&mut args);
887        assert!(
888            spec.is_some(),
889            "the trailing object must be taken as config"
890        );
891        assert_eq!(args.len(), 1, "only the guest's own arg may remain");
892    }
893
894    /// Each scope is parsed by the same Native parser as the primary, and a
895    /// Cypher/Named scope is carried through verbatim for the resolver.
896    #[test]
897    fn scopes_parse_per_scope_and_keep_their_mode() {
898        let scopes = super::GraphProjectionSpec::scopes_from_config_object(&cfg(r#"{"scopes": {
899                 "agg": {"nodeLabels": ["Cell"], "edgeTypes": ["AGG"], "weightProperty": "w"},
900                 "flow": {"nodeQuery": "MATCH (c) RETURN id(c) AS id"}
901               }}"#))
902        .expect("well-formed scopes");
903        assert_eq!(scopes.len(), 2);
904
905        let agg = scopes.iter().find(|s| s.name == "agg").expect("agg");
906        assert_eq!(agg.spec.node_labels, vec!["Cell".to_string()]);
907        assert_eq!(agg.spec.weight_property.as_deref(), Some("w"));
908        assert!(
909            agg.graph_ref.is_none(),
910            "a Native scope must not be routed to the resolver"
911        );
912
913        let flow = scopes.iter().find(|s| s.name == "flow").expect("flow");
914        assert!(
915            flow.graph_ref.is_some(),
916            "a Cypher scope must reach the resolver verbatim"
917        );
918    }
919
920    /// No `scopes` key is not an error — it is the ordinary single-graph CALL.
921    #[test]
922    fn an_absent_scopes_key_yields_no_scopes() {
923        let scopes =
924            super::GraphProjectionSpec::scopes_from_config_object(&cfg(r#"{"nodeLabels": ["N"]}"#))
925                .expect("no scopes is fine");
926        assert!(scopes.is_empty());
927    }
928
929    /// Malformed scope maps are named, not silently dropped.
930    #[test]
931    fn a_malformed_scopes_map_is_rejected_with_the_offending_name() {
932        for (json, needle) in [
933            (r#"{"scopes": ["agg"]}"#, "must be an object"),
934            (r#"{"scopes": {"agg": 7}}"#, "agg"),
935            (r#"{"scopes": {"": {}}}"#, "must not be empty"),
936            // `graph` would shadow the primary handle's own accessor.
937            (r#"{"scopes": {"graph": {}}}"#, "primary projection"),
938        ] {
939            let err = super::GraphProjectionSpec::scopes_from_config_object(&cfg(json))
940                .expect_err("must be rejected");
941            assert!(
942                err.contains(needle),
943                "error for {json} must mention `{needle}`, got: {err}"
944            );
945        }
946    }
947
948    #[test]
949    fn check_slices_accepts_available_and_rejects_missing() {
950        // graph-compute@1 is the host surface; @1 passes, @2 and unknown fail 0x86A.
951        let ok = sig_with(
952            vec![],
953            vec![SliceReq {
954                slice: "graph-compute".into(),
955                version: 1,
956            }],
957        );
958        assert!(ok.check_slices(HOST_CAPABILITY_SLICES).is_ok());
959
960        let too_new = sig_with(
961            vec![],
962            vec![SliceReq {
963                slice: "graph-compute".into(),
964                version: 2,
965            }],
966        );
967        let err = too_new
968            .check_slices(HOST_CAPABILITY_SLICES)
969            .expect_err("graph-compute@2 must be refused");
970        assert_eq!(err.code, 0x86A, "slice mismatch is 0x86A");
971
972        let unknown = sig_with(
973            vec![],
974            vec![SliceReq {
975                slice: "tensor-compute".into(),
976                version: 1,
977            }],
978        );
979        assert_eq!(
980            unknown
981                .check_slices(HOST_CAPABILITY_SLICES)
982                .unwrap_err()
983                .code,
984            0x86A
985        );
986
987        // No declared slices is grandfathered onto the base surface.
988        assert!(
989            sig_with(vec![], vec![])
990                .check_slices(HOST_CAPABILITY_SLICES)
991                .is_ok()
992        );
993    }
994
995    #[test]
996    fn coerce_config_passes_through_when_untyped() {
997        // Empty `args` preserves the legacy raw contract byte-for-byte.
998        let s = sig_with(vec![], vec![]);
999        assert_eq!(s.coerce_config_json("[1, 2, 3]").unwrap(), "[1, 2, 3]");
1000    }
1001
1002    #[test]
1003    fn coerce_config_fills_defaults_and_validates() {
1004        let s = sig_with(
1005            vec![
1006                arg("src", ArgType::CypherValue, None),
1007                arg(
1008                    "alpha",
1009                    ArgType::Primitive(DataType::Float64),
1010                    Some(ScalarValue::Float64(Some(0.85))),
1011                ),
1012            ],
1013            vec![],
1014        );
1015
1016        // A single provided arg fills the omitted `alpha` default.
1017        let out = s.coerce_config_json("[5]").unwrap();
1018        let arr: Vec<serde_json::Value> = serde_json::from_str(&out).unwrap();
1019        assert_eq!(arr.len(), 2, "the omitted default is appended");
1020        assert_eq!(arr[0], serde_json::json!(5));
1021        assert!((arr[1].as_f64().unwrap() - 0.85).abs() < 1e-12);
1022
1023        // A missing required arg is rejected.
1024        let err = s.coerce_config_json("[]").expect_err("src is required");
1025        assert_eq!(err.code, 0x86E);
1026
1027        // A wrong-typed alpha (string, not number) is rejected.
1028        let err = s
1029            .coerce_config_json(r#"[5, "not-a-number"]"#)
1030            .expect_err("alpha must be numeric");
1031        assert_eq!(err.code, 0x86E);
1032
1033        // Too many positional args is rejected.
1034        assert_eq!(s.coerce_config_json("[5, 0.9, 1]").unwrap_err().code, 0x86E);
1035
1036        // A CypherValue arg accepts an array (the `sourceVids` shape).
1037        let arr_src = s.coerce_config_json("[[1, 2, 3], 0.9]").unwrap();
1038        let parsed: Vec<serde_json::Value> = serde_json::from_str(&arr_src).unwrap();
1039        assert!(parsed[0].is_array(), "CypherValue accepts an array");
1040    }
1041}