Compounding Intelligence

RGI Tech-Process Fusion

Three enterprise layers — process intelligence, ERP applications, and EDW & data catalogs — hold the signals. Today they're siloed. Tech-Process Fusion creates causal connectivity across all three, so the platform can reason about process changes, not just observe them.

pip install graph-attention-engine # Apache 2.0

The Gap

Three enterprise layers, no causal connectivity

The Architecture

Situation Analyzer — the differentiator

Not a connector. An orchestrator that assembles cross-system context so JM can reason about process state.

class SituationAnalyzer:
    """Orchestrates cross-system context assembly.
    This is the component that makes Tech-Process Fusion work."""

    def __init__(self, sap, celonis, graph_store, factor_extractor):
        self.sap = sap                  # SAPConnector
        self.celonis = celonis          # CelonisConnector
        self.graph = graph_store        # GraphStore (AGE or SQLite)
        self.extractor = factor_extractor

    async def assemble(self, alert_id: str) -> SituationContext:
        """Pull context from 3 systems, merge into one view."""
        # 1. Graph context — what's connected to this alert
        graph_ctx = await self.graph.query(
            "MATCH (a:Alert {id: $id})-[:AFFECTS]->(s:System)"
            "-[:DEPENDS_ON]->(d:System) RETURN a, s, d",
            params={"id": alert_id}
        )

        # 2. ERP context — supplier, PO, invoice data
        supplier_id = graph_ctx.get("supplier_id")
        erp_ctx = await self.sap.get_context(supplier_id)

        # 3. Process context — bottleneck, variant, timing
        process_id = graph_ctx.get("process_id")
        process_ctx = await self.celonis.get_bottleneck(process_id)

        # 4. Merge into unified SituationContext
        context = SituationContext(
            alert=graph_ctx,
            erp=erp_ctx,           # POs, invoices, amounts
            process=process_ctx,    # timing, variants, conformance
            downstream=graph_ctx.downstream_systems,
            schema_changes=graph_ctx.recent_changes,
        )

        # 5. Extract factor vector for JM scoring
        context.factors = self.extractor.extract(context)
        return context
        # → factors ready for CompoundingScorer.score()

class SAPConnector:
    """OData V2 connector to SAP S/4HANA sandbox.
    Falls back to cached fixtures when sandbox unavailable."""

    BASE = "https://sandbox.api.sap.com/s4hanacloud/sap/opu/odata/sap"

    async def get_purchase_orders(self, supplier_id: str):
        """GET /API_PURCHASEORDER_PROCESS_SRV/A_PurchaseOrder"""
        resp = await self._get(
            f"{self.BASE}/API_PURCHASEORDER_PROCESS_SRV/A_PurchaseOrder",
            params={"$filter": f"Supplier eq '{supplier_id}'",
                    "$select": "PurchaseOrder,PurchasingOrganization,"
                               "PurchaseOrderNetAmount,Supplier"},
            headers={"APIKey": self.api_key}
        )
        return resp["d"]["results"]

    async def get_invoices(self, supplier_id: str):
        """GET /API_SUPPLIERINVOICE_PROCESS_SRV/A_SupplierInvoice"""
        return await self._get(
            f"{self.BASE}/API_SUPPLIERINVOICE_PROCESS_SRV/A_SupplierInvoice",
            params={"$filter": f"InvoicingParty eq '{supplier_id}'"}
        )

    async def get_context(self, supplier_id: str) -> dict:
        """Unified ERP context for SituationAnalyzer."""
        pos = await self.get_purchase_orders(supplier_id)
        invoices = await self.get_invoices(supplier_id)
        return {
            "purchase_orders": pos,
            "total_po_value": sum(float(p["PurchaseOrderNetAmount"]) for p in pos),
            "invoice_count": len(invoices),
            "supplier_id": supplier_id,
        }


class CelonisConnector:
    """REST connector to Celonis Developer Portal.
    Zero signup — any Bearer token works for demo data."""

    BASE = "https://16abf815-remockly.com/intelligence/api"

    async def get_bottleneck(self, process_id: str) -> dict:
        """Knowledge model query — find the slowest activity."""
        km = await self._get(
            f"{self.BASE}/knowledge-models/{process_id}/data",
            headers={"Authorization": f"Bearer {self.token}"}
        )
        activities = km.get("activities", [])
        bottleneck = max(activities, key=lambda a: a["duration_seconds"])
        return {
            "bottleneck_activity": bottleneck["name"],
            "duration_seconds": bottleneck["duration_seconds"],
            "pct_of_total": bottleneck["pct_of_cycle_time"],
            "variant_count": km.get("variant_count", 0),
        }

# Cross-graph query: the computation Celonis cannot do.
# Connects process bottleneck → schema change → downstream impact.

async def diagnose_bottleneck(graph_store, bottleneck_activity: str):
    """Traverse the graph to find root cause of a process bottleneck.
    This crosses three data domains in one query."""

    # 1. Find schema changes that affect the bottleneck activity
    result = await graph_store.query("""
        MATCH (a:Activity {name: $activity})
              -[:BOTTLENECK_AT]->(sys:PipelineSystem)
              <-[:AFFECTS]-(sc:SchemaChange)
        WHERE sc.timestamp > now() - interval '90 days'
        RETURN sc.name AS change, sc.impact AS impact,
               sys.name AS system, sc.code_count AS codes
        ORDER BY sc.impact DESC
    """, params={"activity": bottleneck_activity})

    # 2. Find downstream systems affected
    downstream = await graph_store.query("""
        MATCH (sys:PipelineSystem {name: $system})
              -[:FEEDS*1..3]->(d:PipelineSystem)
        RETURN d.name AS downstream, length(p) AS hops
    """, params={"system": result[0]["system"]})

    # 3. Find supplier correlation
    supplier_corr = await graph_store.query("""
        MATCH (s:Supplier)-[:INVOICE_PATTERN]->(a:Activity {name: $activity})
        RETURN s.name AS supplier,
               s.avg_processing_time / a.avg_time AS slowdown_factor
        ORDER BY slowdown_factor DESC LIMIT 5
    """, params={"activity": bottleneck_activity})

    return {
        "root_cause": result[0]["change"],    # MATKL_V2
        "code_count": result[0]["codes"],      # 340,000
        "join_fanout": f"{result[0]['impact']}x",  # 9x
        "downstream_count": len(downstream),   # 7 systems
        "slowest_supplier": supplier_corr[0],  # Aster, 3.1x
    }

# Score + Learn: JM reasons about the assembled context.
# The Situation Analyzer already extracted the factor vector.

async def score_and_learn(scorer, context: SituationContext):
    """Score a fix recommendation using learned judgment."""

    # Score — O(1), zero LLM calls
    result = scorer.score(
        factors=context.factors,        # 8 factors from SA
        category=context.category,      # e.g. "schema_remediation"
    )
    # → action="pre_join_filter", confidence=0.87

    # Find similar past fixes for basis
    similar = await context.graph.get_similar_decisions(
        category=context.category,
        factors=context.factors,
        limit=3
    )
    # → 3 similar fixes, all successful

    # After human confirms the fix...
    outcome = await wait_for_confirmation()

    # Learn — centroid updates, conservation gate checks
    learn_result = scorer.learn(
        factors=context.factors,
        action=result.action_index,
        category=context.category,
        confirmed=outcome.confirmed,     # η=0.05 if confirmed
    )
    # → centroid shifts toward this fix
    # → conservation gate: PASS (disp=0.034 < 0.4)

    # Check for pattern transfer
    if learn_result["promoted"]:
        # AgentEvolver: 7 tested, 1 promoted, 6 rejected
        await transfer_pattern(
            source_system=context.system,
            target_systems=context.downstream,
            pattern=result.action,
        )
    # → billing_api auto-resolves next occurrence

The Situation Analyzer is the differentiator. Not the Celonis connection — anyone can connect to Celonis. The differentiator is computation on process state inside the live context graph:

Assembles context from SAP + Celonis + Graph in one call
Extracts factor vectors for JM scoring — no LLM needed
SOC copilot uses it in triage.py (the decision pipeline)
DataOps uses it via context_router.py (13 context endpoints, same pattern)

The Computation Chain

Five acts — with real data from the live stack

Tuesday, 8:14 AM. Priya's P2P dashboard lights up — "Match Invoice to GR" just hit 42 minutes, 3x the normal cycle. 70% of total cycle time. Her team has nine pipeline systems and 400 invoices a day. The question isn't what's slow — Celonis already shows that. The question is why, and what to do about it.

Live Data

The intelligence map — from the running stack

What the Situation Analyzer computed from SAP + Celonis + 9 pipeline systems. Oct 2026.

Gold lines = computed cross-system signals. Google Trends → sentiment. MarineTraffic → logistics. Census data → demand. Not configured connections — the graph discovered them from decision patterns.

Experiment

With and without Tech-Process Fusion

Same bottleneck stream, 4 quarters. Without: manual diagnosis, no learning. With: SA + JM + conservation gate.

Exceptions per quarter — manual vs Tech-Process Fusion

Time to root cause — same bottleneck, two approaches

Fix recommendation confidence — JM learns from verified outcomes

# WITHOUT Tech-Process Fusion
# Every bottleneck is a fresh investigation

def handle_bottleneck_manual(alert):
    """4-6 hours per incident. No learning. No transfer."""

    # 1. Analyst opens Celonis dashboard
    # Sees: "Match Invoice to GR" at 42 min
    # Knows WHERE — but not WHY

    # 2. Analyst opens SAP, searches manually
    # Checks PO history, supplier records
    # No cross-reference to schema changes

    # 3. Analyst asks colleagues (tribal knowledge)
    # "Has anyone seen this before?"
    # If the expert left → start from zero

    # 4. Fix is ad-hoc
    fix = guess_fix(alert)  # no confidence score
    apply(fix)              # no basis, no similar-fix lookup
    # No outcome tracking
    # No verification
    # No centroid update

    # 5. Next occurrence → repeat from step 1
    # Q1: 847 exceptions
    # Q2: 831 exceptions  (no improvement)
    # Q3: 856 exceptions  (regression — expert left)
    # Q4: 847 exceptions  (new hire, back to baseline)

# WITH Tech-Process Fusion
# Every verified fix sharpens the centroid geometry

async def handle_bottleneck_tpf(alert, sa, scorer):
    """<1 min root cause. JM compounds. Patterns transfer."""

    # 1. Situation Analyzer assembles cross-system context
    ctx = await sa.assemble(alert.id)
    # → process timing (Celonis) + ERP data (SAP)
    #   + schema changes (EDW) + graph traversal

    # 2. Cross-graph query finds root cause
    diagnosis = await diagnose_bottleneck(
        ctx.graph, ctx.process.bottleneck_activity
    )
    # → MATKL_V2 schema change, 340K codes, 9x fanout
    # → Aster invoices 3.1x slower
    # → 7 downstream systems affected

    # 3. JM scores fix recommendation — O(1), no LLM
    result = scorer.score(ctx.factors, ctx.category)
    # → action="pre_join_filter", confidence=0.87
    # → basis: 3 similar past fixes

    # 4. After confirmation → centroid updates
    scorer.learn(ctx.factors, result.action_index,
                 ctx.category, confirmed=True)
    # → conservation gate: PASS (disp=0.034)

    # 5. AgentEvolver promotes pattern
    await transfer_pattern(ctx.system, ctx.downstream)
    # → billing_api auto-resolves next occurrence

    # Q1: 847 exceptions (same start)
    # Q2: 623 exceptions (JM learning)
    # Q3: 445 exceptions (patterns transferring)
    # Q4: 312 exceptions (63% reduction, compounding)

Manual (Q4)

847

No improvement over 4 quarters

TPF (Q4)

312

JM + conservation + transfer

Reduction

63%

Modeled, D-CEL stack

Root Cause

<1 min

vs 4–6 hours manual

Annual Value

$1.62M

DataOps modeled

Why the gap widens: without TPF, every bottleneck is a fresh investigation — the manual line stays flat because there's no learning mechanism. With TPF, every verified fix sharpens the centroid geometry. The 100th similar bottleneck resolves faster than the 10th. Patterns transfer to downstream systems. JM compounds. Tribal knowledge doesn't.

Market signal: funded startups tackle individual slices — Resolve AI (~$190M, SRE), Ardent AI ($2.15M, data engineering), DataBahn + Pantomath ($47M, pipelines), Dodge AI ($2.65M, ERP maintenance), Evos ($9M, autonomous ops). Each solves one piece. None creates causal connectivity across process intelligence + ERP + EDW with learned judgment and conservation-gated operations.

Validation

Tested across the platform

Test cases

5,702

Across 5 copilots

Enterprise-cut

8/8

D-CEL tests passing

DataOps IKS

31.0

Conservation GREEN