From 84003c47ac8a11a8f4c8f0f23050c4bb13ac9997 Mon Sep 17 00:00:00 2001 From: elasticdotventures Date: Sun, 23 Aug 2026 10:23:46 +0000 Subject: [PATCH 1/2] feat(ledger-core): typed Docling bridge + deterministic bank-statement classifier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two modules: - docling_bridge: typed mirror of reqif-opa-mcp's DocumentGraph JSON schema (models.py), matching the real shape returned by both `reqif_ingest_cli extract` and the new ledgrrr-docling NATS Micro Service. Supersedes rule_registry::DocumentChunk, which declared the same intent but was never deserialized anywhere and didn't match the real named-field SourceAnchor shape. - bank_statement: procedural (regex, never LLM) NodeCategory classifier over DoclingNode via ufo_types::Satisfies, plus node_to_transaction_input bridging TransactionRow-classified nodes into the existing TransactionInput type. Fixes a real bug in PdfIngestOp::execute's TransactionInput construction: it built {date: candidate.section, amount: candidate.confidence.to_string(), ..} from a ReqIfCandidate — a modal-verb-detected requirement sentence has no real transaction date or dollar amount, so this type-checked but was semantically garbage (a confidence score is not a dollar amount). The new bridge only accepts nodes procedurally classified as TransactionRow and extracts real date/description/amount via the same regex used to classify them. NodeCategory::sarif_subtypes() mirrors reqif-opa-mcp's SARIF properties.subtypes tagging convention (reqif_mcp/sarif_producer.py), so classification here is round-trippable into that pipeline without a translation layer. Wiring this into PdfIngestOp itself (replacing the NDJSON/ReqIfCandidate subprocess contract) is left as a follow-up — that changes an existing, tested subprocess CLI contract and deserves its own pass rather than being folded into this one. 201 existing ledger-core tests still pass; 12 new tests added. --- crates/ledger-core/src/bank_statement.rs | 393 +++++++++++++++++++++++ crates/ledger-core/src/docling_bridge.rs | 170 ++++++++++ crates/ledger-core/src/lib.rs | 2 + 3 files changed, 565 insertions(+) create mode 100644 crates/ledger-core/src/bank_statement.rs create mode 100644 crates/ledger-core/src/docling_bridge.rs diff --git a/crates/ledger-core/src/bank_statement.rs b/crates/ledger-core/src/bank_statement.rs new file mode 100644 index 0000000..d17fdca --- /dev/null +++ b/crates/ledger-core/src/bank_statement.rs @@ -0,0 +1,393 @@ +//! Deterministic bank-statement classification over a `DoclingDocumentGraph`. +//! +//! Fixes the previous placeholder bridge in `PdfIngestOp::execute`, which +//! built `TransactionInput { date: candidate.section, amount: +//! candidate.confidence.to_string(), .. } ` from a `ReqIfCandidate` — a +//! modal-verb-detected *normative requirement sentence*, which has no real +//! transaction date or dollar amount at all. That mapping type-checked but +//! was semantically nonsense (a requirement's confidence score is not a +//! transaction amount). +//! +//! This module classifies each `DoclingNode` procedurally — regex pattern +//! matching, never an LLM — against a fixed set of `NodeCategory` shapes, +//! and only nodes classified as `NodeCategory::TransactionRow` are ever +//! turned into a `TransactionInput`. This is the "walk the tree to +//! categorize & action it correctly" step: a deterministic constraint +//! solver over `ufo_types::Satisfies`, not a generative one. + +use regex::Regex; +use rust_decimal::Decimal; +use serde::{Deserialize, Serialize}; +use std::str::FromStr; +use std::sync::LazyLock; +use ufo_types::satisfies::{Constraint, Disposition, NodeId, Satisfies, SatisfiesResult}; + +use crate::docling_bridge::{DoclingDocumentGraph, DoclingNode}; +use crate::ingest::TransactionInput; + +/// Categories a document node can be classified into. Deliberately closed +/// (not an open string) so every category has an explicit, reviewable +/// classification rule below — an unhandled category is a compile error, +/// not a silently-skipped node. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum NodeCategory { + /// A single ledger line: date + description + signed dollar amount, + /// optionally followed by a running balance. + TransactionRow, + /// Statement metadata: "Beginning balance" / "Ending balance" and + /// similar period-summary lines. + StatementHeader, + /// Fee schedule / rate disclosure text — informational, never a + /// transaction, but still worth tagging (not `Unclassified`). + FeeSchedule, + /// Legal boilerplate / disclaimers. + Disclaimer, + /// Did not match any known shape. + Unclassified, +} + +impl NodeCategory { + /// Tag strings mirroring the `properties.subtypes` convention already + /// used by `reqif-opa-mcp`'s SARIF producer (`reqif_mcp/sarif_producer.py`: + /// `requirement.subtypes` flows into `rule.properties.subtypes` / + /// `result.properties.subtypes`). Using the same dotted-namespace + /// convention here means a future round-trip back into that SARIF + /// pipeline needs no translation layer. + pub fn sarif_subtypes(self) -> Vec { + match self { + Self::TransactionRow => vec!["bank_statement.transaction_row".to_string()], + Self::StatementHeader => vec!["bank_statement.header".to_string()], + Self::FeeSchedule => vec!["bank_statement.fee_schedule".to_string()], + Self::Disclaimer => vec!["bank_statement.disclaimer".to_string()], + Self::Unclassified => vec![], + } + } +} + +/// A `ufo_types::Satisfies` constraint: "does this node belong to category C?" +pub struct CategoryConstraint(pub NodeCategory); + +impl Constraint for CategoryConstraint {} + +// Wells Fargo-style transaction row: `MM/DD` or `MM/DD/YYYY`, then anything, +// then a signed dollar amount, optionally followed by a running balance. +// Matches the date-format conventions already detected in +// `document_shape.rs::detect_date_format` (`%m/%d/%Y`, `%m/%d`). +static TRANSACTION_ROW: LazyLock = LazyLock::new(|| { + Regex::new(r"(?x) + ^\s*(?P\d{1,2}/\d{1,2}(?:/\d{2,4})?)\s+ + (?P.+?)\s+ + (?P-?\$?\d[\d,]*\.\d{2}) + (?:\s+\$?(?P\d[\d,]*\.\d{2}))?\s*$ + ").expect("static regex is valid") +}); + +static STATEMENT_HEADER: LazyLock = LazyLock::new(|| { + Regex::new(r"(?i)(beginning|opening|ending|closing)\s+balance").expect("static regex is valid") +}); + +static FEE_SCHEDULE: LazyLock = LazyLock::new(|| { + Regex::new(r"(?i)(monthly\s+service\s+fee|overdraft\s+fee|interest\s+rate|APY)").expect("static regex is valid") +}); + +static DISCLAIMER: LazyLock = LazyLock::new(|| { + Regex::new(r"(?i)(member\s+FDIC|equal\s+housing\s+lender|see\s+reverse\s+side)").expect("static regex is valid") +}); + +impl Satisfies for DoclingNode { + fn satisfies(&self, constraint: &CategoryConstraint) -> SatisfiesResult { + let Some(text) = self.text.as_deref() else { + return SatisfiesResult::unknown(); + }; + let evidence = vec![NodeId::new(self.node_id.clone())]; + + let matched = match constraint.0 { + NodeCategory::TransactionRow => TRANSACTION_ROW.is_match(text), + NodeCategory::StatementHeader => STATEMENT_HEADER.is_match(text), + NodeCategory::FeeSchedule => FEE_SCHEDULE.is_match(text), + NodeCategory::Disclaimer => DISCLAIMER.is_match(text), + NodeCategory::Unclassified => false, + }; + + if matched { + // Deterministic regex match: full confidence, no ambiguity band. + SatisfiesResult { + disposition: Disposition::Satisfied, + confidence: 1.0, + evidence_nodes: evidence, + ufo_category: ufo_types::ufo::MomentStereotype::Mode, + } + } else { + SatisfiesResult::violated("no pattern match for this category") + } + } +} + +/// A node paired with the (single, best) category it was classified into. +#[derive(Debug, Clone)] +pub struct ClassifiedNode<'a> { + pub node: &'a DoclingNode, + pub category: NodeCategory, + pub result: SatisfiesResult, +} + +/// Category check order matters only in that it is exhaustive and each +/// check is mutually exclusive by construction (a transaction row's regex +/// cannot also match the header/fee/disclaimer keyword patterns in +/// practice); ties are not possible today, so no priority scheme is needed. +const CATEGORY_ORDER: [NodeCategory; 4] = [ + NodeCategory::TransactionRow, + NodeCategory::StatementHeader, + NodeCategory::FeeSchedule, + NodeCategory::Disclaimer, +]; + +/// Walk every text-bearing node in the graph and classify it. This is the +/// constraint-solver "walk the tree to categorize & action it correctly" +/// step: deterministic, procedural, and lint-able (every `ClassifiedNode` +/// carries the `SatisfiesResult` that justified its category, not just a +/// bare label). +pub fn classify_document(graph: &DoclingDocumentGraph) -> Vec> { + graph + .text_nodes() + .map(|node| { + for &category in &CATEGORY_ORDER { + let result = node.satisfies(&CategoryConstraint(category)); + if result.disposition.is_satisfied() { + return ClassifiedNode { node, category, result }; + } + } + ClassifiedNode { + node, + category: NodeCategory::Unclassified, + result: SatisfiesResult::unknown(), + } + }) + .collect() +} + +/// Error bridging a classified node into `TransactionInput`. +#[derive(Debug, thiserror::Error)] +pub enum BridgeError { + #[error("node {0} was not classified as a transaction row")] + NotATransactionRow(String), + #[error("node {0} matched the transaction pattern but its amount failed to parse: {1}")] + AmountParse(String, rust_decimal::Error), +} + +/// Convert a `TransactionRow`-classified node into the flat +/// `TransactionInput` shape the rest of the ingest pipeline expects, +/// re-running the same regex to extract real date/description/amount +/// fields instead of the placeholder `section`/`confidence` mapping this +/// replaces. +pub fn node_to_transaction_input( + classified: &ClassifiedNode<'_>, + account_id: &str, +) -> Result { + if classified.category != NodeCategory::TransactionRow { + return Err(BridgeError::NotATransactionRow(classified.node.node_id.clone())); + } + let text = classified.node.text.as_deref().unwrap_or_default(); + let caps = TRANSACTION_ROW + .captures(text) + .ok_or_else(|| BridgeError::NotATransactionRow(classified.node.node_id.clone()))?; + + let date = caps["date"].to_string(); + let description = caps["description"].trim().to_string(); + let raw_amount = caps["amount"].replace(['$', ','], ""); + let amount = Decimal::from_str(&raw_amount) + .map_err(|e| BridgeError::AmountParse(classified.node.node_id.clone(), e))?; + + Ok(TransactionInput { + account_id: account_id.to_string(), + date, + amount: amount.to_string(), + description, + // Traceable back to the exact source node/page, unlike the old + // filename-only source_ref — a page number is embedded when known. + source_ref: match classified.node.first_page() { + Some(page) => format!("{}#page={page}", classified.node.node_id), + None => classified.node.node_id.clone(), + }, + }) +} + +/// Statement-level period metadata, distinct from any single transaction. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct StatementPeriod { + /// Raw as extracted (e.g. `"05/01/2026"`) — parsed downstream by the + /// same date-format machinery `document_shape.rs` already uses, kept + /// as a string here for the same reason `TransactionInput::date` is a + /// string: format detection is a separate, later pipeline stage. + pub start: String, + pub end: String, +} + +/// Statement-level header metadata: opening/closing balance for the period. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct StatementHeader { + pub opening_balance: Decimal, + pub closing_balance: Decimal, +} + +/// Scan classified `StatementHeader`-category nodes for beginning/ending +/// balance amounts. Returns `None` if either amount is missing — a partial +/// header is not a usable header, per the "procedurally lint-able" mandate +/// (callers should treat this as `NeedsReview`, not silently proceed with a +/// zeroed balance). +pub fn extract_statement_header(graph: &DoclingDocumentGraph) -> Option { + static BALANCE_AMOUNT: LazyLock = + LazyLock::new(|| Regex::new(r"\$?(\d[\d,]*\.\d{2})").expect("static regex is valid")); + + let mut opening: Option = None; + let mut closing: Option = None; + + for classified in classify_document(graph) { + if classified.category != NodeCategory::StatementHeader { + continue; + } + let text = classified.node.text.as_deref().unwrap_or_default(); + let Some(amount_caps) = BALANCE_AMOUNT.captures(text) else { + continue; + }; + let Ok(amount) = Decimal::from_str(&amount_caps[1].replace(',', "")) else { + continue; + }; + let lower = text.to_lowercase(); + if lower.contains("beginning") || lower.contains("opening") { + opening = Some(amount); + } else if lower.contains("ending") || lower.contains("closing") { + closing = Some(amount); + } + } + + match (opening, closing) { + (Some(opening_balance), Some(closing_balance)) => Some(StatementHeader { + opening_balance, + closing_balance, + }), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::docling_bridge::{DoclingAnchor, DoclingArtifact}; + use std::collections::HashMap; + + fn node(node_id: &str, text: &str, page: Option) -> DoclingNode { + DoclingNode { + node_id: node_id.to_string(), + node_type: "paragraph".to_string(), + text: Some(text.to_string()), + parent_id: None, + semantic_id: format!("semantic-{node_id}"), + attributes: HashMap::new(), + anchors: vec![DoclingAnchor { + kind: "pdf_page_paragraph".to_string(), + artifact_id: "artifact-1".to_string(), + page, + ..Default::default() + }], + } + } + + fn graph(nodes: Vec) -> DoclingDocumentGraph { + DoclingDocumentGraph { + schema: "document_graph/1".to_string(), + artifact: DoclingArtifact { + artifact_id: "artifact-1".to_string(), + source_uri: None, + sha256: "deadbeef".to_string(), + document_profile: "pdf_docling_v1".to_string(), + }, + profile: "pdf_docling_v1".to_string(), + nodes, + metadata: HashMap::new(), + } + } + + #[test] + fn classifies_transaction_row() { + let g = graph(vec![node("n1", "05/01 Check 1042 -$120.00 $4,880.00", Some(2))]); + let classified = classify_document(&g); + assert_eq!(classified.len(), 1); + assert_eq!(classified[0].category, NodeCategory::TransactionRow); + assert!(classified[0].result.disposition.is_satisfied()); + } + + #[test] + fn classifies_statement_header() { + let g = graph(vec![node("n1", "Beginning balance on 5/1 $5,000.00", None)]); + let classified = classify_document(&g); + assert_eq!(classified[0].category, NodeCategory::StatementHeader); + } + + #[test] + fn classifies_fee_schedule() { + let g = graph(vec![node("n1", "Monthly service fee: $12.00 unless minimum balance met", None)]); + assert_eq!(classify_document(&g)[0].category, NodeCategory::FeeSchedule); + } + + #[test] + fn classifies_disclaimer() { + let g = graph(vec![node("n1", "Wells Fargo Bank, N.A. Member FDIC.", None)]); + assert_eq!(classify_document(&g)[0].category, NodeCategory::Disclaimer); + } + + #[test] + fn unclassified_when_no_pattern_matches() { + let g = graph(vec![node("n1", "Table of Contents", None)]); + assert_eq!(classify_document(&g)[0].category, NodeCategory::Unclassified); + } + + #[test] + fn bridges_transaction_row_to_real_transaction_input_not_placeholder_garbage() { + let g = graph(vec![node("n1", "05/01 Check 1042 -$120.00 $4,880.00", Some(2))]); + let classified = classify_document(&g); + let tx = node_to_transaction_input(&classified[0], "acct-123").unwrap(); + + // The bug this replaces: date was `candidate.section` (a heading + // path), amount was `candidate.confidence.to_string()` (e.g. + // "0.85"). Assert the real values instead. + assert_eq!(tx.date, "05/01"); + assert_eq!(tx.amount, "-120.00"); + assert_eq!(tx.description, "Check 1042"); + assert_eq!(tx.account_id, "acct-123"); + assert_eq!(tx.source_ref, "n1#page=2"); + } + + #[test] + fn rejects_bridging_a_non_transaction_node() { + let g = graph(vec![node("n1", "Member FDIC.", None)]); + let classified = classify_document(&g); + assert!(node_to_transaction_input(&classified[0], "acct-123").is_err()); + } + + #[test] + fn extracts_statement_header_when_both_balances_present() { + let g = graph(vec![ + node("n1", "Beginning balance on 5/1 $5,000.00", None), + node("n2", "Ending balance on 5/31 $4,880.00", None), + ]); + let header = extract_statement_header(&g).unwrap(); + assert_eq!(header.opening_balance, Decimal::from_str("5000.00").unwrap()); + assert_eq!(header.closing_balance, Decimal::from_str("4880.00").unwrap()); + } + + #[test] + fn no_header_when_only_one_balance_present() { + let g = graph(vec![node("n1", "Beginning balance on 5/1 $5,000.00", None)]); + assert!(extract_statement_header(&g).is_none()); + } + + #[test] + fn sarif_subtypes_match_reqif_opa_mcp_convention() { + assert_eq!( + NodeCategory::TransactionRow.sarif_subtypes(), + vec!["bank_statement.transaction_row".to_string()] + ); + assert!(NodeCategory::Unclassified.sarif_subtypes().is_empty()); + } +} diff --git a/crates/ledger-core/src/docling_bridge.rs b/crates/ledger-core/src/docling_bridge.rs new file mode 100644 index 0000000..02ee968 --- /dev/null +++ b/crates/ledger-core/src/docling_bridge.rs @@ -0,0 +1,170 @@ +//! Typed Rust mirror of `reqif-opa-mcp`'s `DocumentGraph` JSON schema +//! (`reqif_ingest_cli/models.py`), so extraction output — whether piped +//! through a subprocess or returned by the `ledgrrr-docling` NATS Micro +//! Service (`reqif_ingest_cli/nats_docling_service.py`) — can be +//! deserialized directly instead of re-parsed field-by-field ad hoc. +//! +//! Supersedes `rule_registry::DocumentChunk`, which declared a similar +//! intent ("maps to reqif-opa-mcp's DocumentNode") but was never actually +//! deserialized anywhere and doesn't match the real anchor shape (the +//! Python `SourceAnchor` dataclass has named fields — page/sheet/row/ +//! column/cell/paragraph/heading_path/semantic_id — not a `[u32; 2]` pair). + +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// Mirrors `reqif_ingest_cli.models.ArtifactRecord`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct DoclingArtifact { + pub artifact_id: String, + pub source_uri: Option, + pub sha256: String, + pub document_profile: String, +} + +/// Mirrors `reqif_ingest_cli.models.SourceAnchor`. Every field beyond `kind` +/// is format-specific (PDF pages vs. XLSX cells) so all are optional. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct DoclingAnchor { + pub kind: String, + pub artifact_id: String, + #[serde(default)] + pub page: Option, + #[serde(default)] + pub sheet: Option, + #[serde(default)] + pub row: Option, + #[serde(default)] + pub column: Option, + #[serde(default)] + pub cell: Option, + #[serde(default)] + pub paragraph: Option, + #[serde(default)] + pub heading_path: Vec, + #[serde(default)] + pub semantic_id: String, +} + +/// Mirrors `reqif_ingest_cli.models.DocumentNode`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct DoclingNode { + pub node_id: String, + pub node_type: String, + #[serde(default)] + pub text: Option, + #[serde(default)] + pub parent_id: Option, + pub semantic_id: String, + #[serde(default)] + pub attributes: HashMap, + #[serde(default)] + pub anchors: Vec, +} + +impl DoclingNode { + /// First page number among this node's anchors, if any. + pub fn first_page(&self) -> Option { + self.anchors.iter().find_map(|a| a.page) + } + + /// Heading path of this node's first anchor, if any — the closest + /// analogue to a "section" for a PDF-sourced node. + pub fn heading_path(&self) -> &[String] { + self.anchors.first().map(|a| a.heading_path.as_slice()).unwrap_or(&[]) + } +} + +/// Mirrors `reqif_ingest_cli.models.DocumentGraph` — the full JSON payload +/// returned by `reqif_ingest_cli extract` and by the NATS service's +/// `ledgrrr.extract` endpoint. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct DoclingDocumentGraph { + pub schema: String, + pub artifact: DoclingArtifact, + pub profile: String, + pub nodes: Vec, + #[serde(default)] + pub metadata: HashMap, +} + +impl DoclingDocumentGraph { + /// Iterate only nodes that carry non-empty text — the vast majority of + /// classification/extraction logic operates on these. + pub fn text_nodes(&self) -> impl Iterator { + self.nodes + .iter() + .filter(|n| n.text.as_deref().is_some_and(|t| !t.trim().is_empty())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Real shape observed from a live `reqif_ingest_cli extract --pretty` + /// run against samples/standards/upstream/owasp-asvs/OWASP_ASVS_5.0.0_en.pdf + /// (2026-08-23), trimmed to one node. + const SAMPLE_JSON: &str = r#"{ + "schema": "document_graph/1", + "artifact": { + "artifact_id": "artifact-d92ba680322b", + "source_uri": null, + "sha256": "deadbeef", + "document_profile": "pdf_docling_v1" + }, + "profile": "pdf_docling_v1", + "nodes": [ + { + "node_id": "paragraph-abc123", + "node_type": "paragraph", + "text": "05/01 Check 1042 -$120.00 $4,880.00", + "parent_id": null, + "semantic_id": "semantic-eac1ea760e01", + "attributes": {"label": "pdf_text", "extractor": "pypdf"}, + "anchors": [ + { + "kind": "pdf_page_paragraph", + "artifact_id": "artifact-d92ba680322b", + "page": 2, + "sheet": null, + "row": null, + "column": null, + "cell": null, + "paragraph": 1, + "heading_path": [], + "semantic_id": "semantic-eac1ea760e01" + } + ] + } + ], + "metadata": {"extractor": "pypdf", "fallback_reason": null} + }"#; + + #[test] + fn deserializes_real_extraction_shape() { + let graph: DoclingDocumentGraph = serde_json::from_str(SAMPLE_JSON).unwrap(); + assert_eq!(graph.profile, "pdf_docling_v1"); + assert_eq!(graph.nodes.len(), 1); + let node = &graph.nodes[0]; + assert_eq!(node.first_page(), Some(2)); + assert!(node.text.as_deref().unwrap().contains("Check 1042")); + } + + #[test] + fn text_nodes_filters_empty() { + let mut graph: DoclingDocumentGraph = serde_json::from_str(SAMPLE_JSON).unwrap(); + graph.nodes.push(DoclingNode { + node_id: "empty".into(), + node_type: "paragraph".into(), + text: Some(" ".into()), + parent_id: None, + semantic_id: "empty".into(), + attributes: HashMap::new(), + anchors: vec![], + }); + assert_eq!(graph.text_nodes().count(), 1); + } +} diff --git a/crates/ledger-core/src/lib.rs b/crates/ledger-core/src/lib.rs index 3de89c3..01170f4 100644 --- a/crates/ledger-core/src/lib.rs +++ b/crates/ledger-core/src/lib.rs @@ -1,8 +1,10 @@ pub mod attest; +pub mod bank_statement; pub mod calendar; pub mod classify; pub mod constraints; pub mod crypto; +pub mod docling_bridge; pub mod document; pub mod document_shape; pub mod filename; From 752e500eb4e3cb8781c52a7880890100d7a8a11d Mon Sep 17 00:00:00 2001 From: elasticdotventures Date: Sun, 23 Aug 2026 10:34:16 +0000 Subject: [PATCH 2/2] feat(ledger-core): expose ledgrrr's classifier as its own NATS Micro Service MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds ledgrrr-nats-service (feature-gated: nats-service), a real async-nats (0.45, service feature) Micro Service registering as 'ledgrrr' with endpoint 'classify' (subject ledgrrr.classify). This is the actual 'shared ledgrrr' — not the reqif-opa-mcp Docling tool it depends on, but ledgrrr's own deterministic bank_statement classifier made network-reachable and discoverable via 'nats service list', the Rust-side counterpart to reqif-opa-mcp's Python nats_docling_service.py (reqif-opa-mcp#22). Verified end-to-end against a real NATS test server: 'ledgrrr' shows up in 'nats service list' alongside 'ledgrrr-docling'; a request built from a real Docling extraction of the OWASP ASVS PDF plus synthetic bank-statement-shaped nodes is correctly classified (Unclassified for the real ASVS prose, TransactionRow/StatementHeader for the synthetic rows) and bridged into real TransactionInput/StatementHeader values. Intended chaining: extract via ledgrrr-docling's ledgrrr.extract, then classify+bridge via this service's ledgrrr.classify — two independently discoverable, independently deployable NATS services. --- Cargo.lock | 201 +++++++++++++++++- crates/ledger-core/Cargo.toml | 8 + .../src/bin/ledgrrr_nats_service.rs | 133 ++++++++++++ 3 files changed, 336 insertions(+), 6 deletions(-) create mode 100644 crates/ledger-core/src/bin/ledgrrr_nats_service.rs diff --git a/Cargo.lock b/Cargo.lock index 39d5e10..53d1a60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -607,6 +607,43 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-nats" +version = "0.45.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "86dde77d8a733a9dbaf865a9eb65c72e09c88f3d14d3dd0d2aecf511920ee4fe" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-util", + "memchr", + "nkeys", + "nuid", + "once_cell", + "pin-project", + "portable-atomic", + "rand 0.8.6", + "regex", + "ring", + "rustls-native-certs 0.7.3", + "rustls-pemfile", + "rustls-webpki 0.102.8", + "serde", + "serde_json", + "serde_nanos", + "serde_repr", + "thiserror 1.0.69", + "time", + "tokio", + "tokio-rustls", + "tokio-stream", + "tokio-util", + "tokio-websockets", + "tracing", + "tryhard", + "url", +] + [[package]] name = "async-process" version = "2.5.0" @@ -2198,6 +2235,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid", + "pem-rfc7468", "zeroize", ] @@ -2522,6 +2560,7 @@ dependencies = [ "rand_core 0.6.4", "serde", "sha2", + "signature", "subtle", "zeroize", ] @@ -4604,6 +4643,7 @@ dependencies = [ "anyhow", "arc-kit-au", "arrow", + "async-nats", "automod", "blake3", "calamine", @@ -4612,6 +4652,7 @@ dependencies = [ "femtovg", "filetime", "frunk", + "futures-util", "glam 0.27.0", "kasuari", "ledger-attest", @@ -5622,6 +5663,21 @@ dependencies = [ "smallvec 1.15.1", ] +[[package]] +name = "nkeys" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879011babc47a1c7fdf5a935ae3cfe94f34645ca0cac1c7f6424b36fc743d1bf" +dependencies = [ + "data-encoding", + "ed25519", + "ed25519-dalek", + "getrandom 0.2.17", + "log", + "rand 0.8.6", + "signatory", +] + [[package]] name = "no_std_io2" version = "0.9.4" @@ -5718,6 +5774,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "nuid" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc895af95856f929163a0aa20c26a78d26bfdc839f51b9d5aa7a5b79e52b7e83" +dependencies = [ + "rand 0.8.6", +] + [[package]] name = "num-bigint" version = "0.4.6" @@ -6140,6 +6205,12 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "openssl-probe" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" + [[package]] name = "openssl-probe" version = "0.2.1" @@ -6273,6 +6344,15 @@ dependencies = [ "hmac", ] +[[package]] +name = "pem-rfc7468" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412" +dependencies = [ + "base64ct", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -7501,21 +7581,43 @@ dependencies = [ "once_cell", "ring", "rustls-pki-types", - "rustls-webpki", + "rustls-webpki 0.103.13", "subtle", "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5bfb394eeed242e909609f56089eecfe5fda225042e8b171791b9c95f5931e5" +dependencies = [ + "openssl-probe 0.1.6", + "rustls-pemfile", + "rustls-pki-types", + "schannel", + "security-framework 2.11.1", +] + [[package]] name = "rustls-native-certs" version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63" dependencies = [ - "openssl-probe", + "openssl-probe 0.2.1", "rustls-pki-types", "schannel", - "security-framework", + "security-framework 3.7.0", +] + +[[package]] +name = "rustls-pemfile" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" +dependencies = [ + "rustls-pki-types", ] [[package]] @@ -7540,10 +7642,10 @@ dependencies = [ "log", "once_cell", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.3", "rustls-platform-verifier-android", - "rustls-webpki", - "security-framework", + "rustls-webpki 0.103.13", + "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", "windows-sys 0.61.2", @@ -7555,6 +7657,16 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" +[[package]] +name = "rustls-webpki" +version = "0.102.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9" +dependencies = [ + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustls-webpki" version = "0.103.13" @@ -7735,6 +7847,19 @@ version = "4.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b" +[[package]] +name = "security-framework" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" +dependencies = [ + "bitflags 2.11.1", + "core-foundation 0.9.4", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + [[package]] name = "security-framework" version = "3.7.0" @@ -7908,6 +8033,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_nanos" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a93142f0367a4cc53ae0fead1bcda39e85beccfad3dcd717656cacab94b12985" +dependencies = [ + "serde", +] + [[package]] name = "serde_path_to_error" version = "0.1.20" @@ -8121,12 +8255,25 @@ dependencies = [ "libc", ] +[[package]] +name = "signatory" +version = "0.27.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1e303f8205714074f6068773f0e29527e0453937fe837c9717d066635b65f31" +dependencies = [ + "pkcs8", + "rand_core 0.6.4", + "signature", + "zeroize", +] + [[package]] name = "signature" version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ + "digest", "rand_core 0.6.4", ] @@ -9484,6 +9631,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.23.1" @@ -9537,6 +9695,27 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-websockets" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f591660438b3038dd04d16c938271c79e7e06260ad2ea2885a4861bfb238605d" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "futures-sink", + "http", + "httparse", + "rand 0.8.6", + "ring", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tokio-util", + "webpki-roots 0.26.11", +] + [[package]] name = "toktrie" version = "1.7.5" @@ -9883,6 +10062,16 @@ dependencies = [ "toml 1.1.2+spec-1.1.0", ] +[[package]] +name = "tryhard" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fe58ebd5edd976e0fe0f8a14d2a04b7c81ef153ea9a54eebc42e67c2c23b4e5" +dependencies = [ + "pin-project-lite", + "tokio", +] + [[package]] name = "ttf-parser" version = "0.21.1" diff --git a/crates/ledger-core/Cargo.toml b/crates/ledger-core/Cargo.toml index a203436..205d571 100644 --- a/crates/ledger-core/Cargo.toml +++ b/crates/ledger-core/Cargo.toml @@ -32,6 +32,8 @@ tokio = { workspace = true } tracing = "0.1" notify = "8.2" ledger-attest = { path = "../ledger-attest" } +async-nats = { version = "0.45", features = ["service"], optional = true } +futures-util = { version = "0.3", optional = true } # filesystem metadata — xattr on Linux; sidecar fallback is pure std [target.'cfg(target_os = "linux")'.dependencies] @@ -42,11 +44,17 @@ default = ["arc-kit-au"] legal-z3 = ["dep:z3"] otel-arrow = ["dep:arrow"] cedar-policy = ["dep:msft-agent-gov-ledgrrr"] +nats-service = ["dep:async-nats", "dep:futures-util"] [dependencies.msft-agent-gov-ledgrrr] path = "../msft-agent-gov-ledgrrr" optional = true +[[bin]] +name = "ledgrrr-nats-service" +path = "src/bin/ledgrrr_nats_service.rs" +required-features = ["nats-service"] + [lints] workspace = true diff --git a/crates/ledger-core/src/bin/ledgrrr_nats_service.rs b/crates/ledger-core/src/bin/ledgrrr_nats_service.rs new file mode 100644 index 0000000..54a5b97 --- /dev/null +++ b/crates/ledger-core/src/bin/ledgrrr_nats_service.rs @@ -0,0 +1,133 @@ +//! Exposes ledgrrr's own classification logic as a NATS Micro Service — +//! this, not `reqif-opa-mcp`'s Docling wrapper, is the "shared ledgrrr": +//! ledgrrr's deterministic bank-statement classifier +//! (`ledger_core::bank_statement`) becomes network-reachable and +//! discoverable via the standard `nats service list` protocol, the same +//! way `reqif-opa-mcp`'s `nats_docling_service.py` already exposes Docling +//! extraction. The two are meant to be chained by a caller: extract via +//! `ledgrrr-docling`'s `ledgrrr.extract`, then classify+bridge via this +//! service's `ledgrrr.classify`. +//! +//! Endpoint: "classify" in group "ledgrrr" (subject `ledgrrr.classify`). +//! Request: `{"graph": , "account_id": "..."}`. +//! Reply: `ClassifyResponse` JSON, or a NATS service error on failure. +//! +//! Run: `cargo run -p ledger-core --features nats-service --bin ledgrrr-nats-service` +//! Env: NATS_URL (default nats://127.0.0.1:4222), NATS_USER, NATS_PASSWORD. + +use async_nats::service::ServiceExt; +use futures_util::StreamExt; +use ledger_core::bank_statement::{ + classify_document, extract_statement_header, node_to_transaction_input, NodeCategory, +}; +use ledger_core::docling_bridge::DoclingDocumentGraph; +use ledger_core::ingest::TransactionInput; +use serde::{Deserialize, Serialize}; + +const SERVICE_NAME: &str = "ledgrrr"; +const SERVICE_VERSION: &str = env!("CARGO_PKG_VERSION"); + +#[derive(Debug, Deserialize)] +struct ClassifyRequest { + graph: DoclingDocumentGraph, + #[serde(default = "default_account_id")] + account_id: String, +} + +fn default_account_id() -> String { + "unknown".to_string() +} + +#[derive(Debug, Serialize)] +struct ClassifiedNodeSummary { + node_id: String, + category: NodeCategory, + subtypes: Vec, + satisfied: bool, + confidence: f64, +} + +#[derive(Debug, Serialize)] +struct ClassifyResponse { + classified: Vec, + transactions: Vec, + statement_header: Option, +} + +fn handle_request(payload: &[u8]) -> Result, String> { + let request: ClassifyRequest = + serde_json::from_slice(payload).map_err(|e| format!("invalid request JSON: {e}"))?; + + let classified = classify_document(&request.graph); + + let mut transactions = Vec::new(); + let mut summaries = Vec::with_capacity(classified.len()); + for c in &classified { + summaries.push(ClassifiedNodeSummary { + node_id: c.node.node_id.clone(), + category: c.category, + subtypes: c.category.sarif_subtypes(), + satisfied: c.result.disposition.is_satisfied(), + confidence: c.result.confidence, + }); + if c.category == NodeCategory::TransactionRow { + if let Ok(tx) = node_to_transaction_input(c, &request.account_id) { + transactions.push(tx); + } + } + } + + let statement_header = extract_statement_header(&request.graph); + + let response = ClassifyResponse { + classified: summaries, + transactions, + statement_header, + }; + serde_json::to_vec(&response).map_err(|e| format!("failed to serialize response: {e}")) +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); + let mut options = async_nats::ConnectOptions::new(); + if let (Ok(user), Ok(password)) = (std::env::var("NATS_USER"), std::env::var("NATS_PASSWORD")) { + options = options.user_and_password(user, password); + } + let client = options.connect(&url).await?; + + let service = client + .service_builder() + .description("ledgrrr's deterministic (non-LLM) document classification: bank-statement node categorization + TransactionInput bridging over a DoclingDocumentGraph") + .start(SERVICE_NAME, SERVICE_VERSION) + .await?; + + let group = service.group("ledgrrr"); + let mut endpoint = group.endpoint("classify").await?; + + println!("[{SERVICE_NAME}] listening on '{url}' as subject 'ledgrrr.classify'"); + + while let Some(request) = endpoint.next().await { + let result = handle_request(&request.message.payload); + match result { + Ok(bytes) => { + if let Err(e) = request.respond(Ok(bytes.into())).await { + eprintln!("failed to send reply: {e}"); + } + } + Err(msg) => { + if let Err(e) = request + .respond(Err(async_nats::service::error::Error { + code: 400, + status: msg, + })) + .await + { + eprintln!("failed to send error reply: {e}"); + } + } + } + } + + Ok(()) +}