{"slug":"business/INGESTION_AND_AGENT_PIPELINE","title":"Ingestion, Pipeline, and Agent Processing Specification","section":"business","sectionLabel":"Business","status":"unclassified","type":"document","path":"business/INGESTION_AND_AGENT_PIPELINE.mdx","bytes":8275,"description":null,"updated":null,"source":"integratewise/Integratewise-docs-wiki","content":"\nThis document details the canonical architecture of the IntegrateWise data ingestion pipeline and AI agent processing core. It maps exactly how files are captured, when sync triggers fire, how data connects to the Spine database (Core Memory Substrate), and how specialized agents orchestrate reasoning.\n\n---\n\n## 1. File Ingestion Mechanics & Triggers\n\nTo maintain a zero-overhead server footprint, data capture operates on a decentralized, event-driven model.\n\n```\nLocal Folder Monitor (chokidar) ──[HTTP POST]──▶ Webhook Ingress Worker\n                                                        │\n                                                        ▼\n                                                Loader (provenance)\n                                                        │\n                                                        ▼\n                                            Cloudflare Ingestion Queue\n                                                        │\n                                                        ▼\n                                            Normalizer (8-Stage Pipeline)\n```\n\n### Ingestion Triggers (The Instants of Capture)\n\nData ingestion and run flow updates are initiated by four distinct trigger types:\n\n1.  **Event-Driven Folder Monitor (Instantaneous)**:\n    - **The Actor**: The local `Folder Monitor` (built using `chokidar` in Node.js) runs as a background process on the user's operational Mac/ops machine.\n    - **The Instant**: Triggered instantaneously when a file is **created**, **modified**, or **saved** inside monitored filesystem directories (such as the local markdown memory vault).\n    - **The Flow**: The monitor instantly reads the file chunk, extracts basic file-system metadata (path, size, modification timestamp), and streams the payload via an HTTP POST request to the `webhook-ingress` worker, which passes it to the `loader`.\n2.  **API Connector Delta Sync (Scheduled Intervals)**:\n    - **The Actor**: The `connector-sync` scheduler.\n    - **The Instant**: Triggered at intervals determined by the tenant's sync entitlement tier (Free: every 24h, Starter: 4h, Professional: 1h, Enterprise: 15min).\n    - **The Flow**: The loader retrieves delta updates (records modified since `last_sync_at`) directly from third-party tools (HubSpot, Salesforce, Jira) and queues them for processing.\n3.  **Manual User Upload (On-Demand)**:\n    - **The Actor**: The frontend Workspace Shell.\n    - **The Instant**: Triggered instantly when a user drags-and-drops a document or file into the loader interface.\n    - **The Flow**: The client streams the file upload to the Hono Gateway, which routes it directly to the `loader` service.\n4.  **User Workbench Interactive Changes & Run-Flow Updates (Real-Time State Ingestion)**:\n    - **The Actor**: The frontend User Workbench (interactive pages, settings panels, curated memory boards, governance panels, and playbook trigger dashboards).\n    - **The Instant**: Triggered in real-time when the user makes any interactive update in their workbench—such as toggling alert settings, modifying account health thresholds, approving pending queue curation tasks, editing memory text blocks, or initiating a new run flow.\n    - **The Flow**: The frontend client captures all changes and updates in the user workbench and immediately executes a write/update through the `spineClient` layer back to the **Spine DB / Core Memory Substrate**. This instantly emits a state change signal, triggering the **Twin Reasoning Engine** to re-ground its context and the **Signal Engine** to recalculate company KPIs, while automatically initiating any approved playbook execution flows.\n\n---\n\n## 2. Ingestion to Spine Database Connectivity\n\nEvery ingestion path flows through the `loader` into the Cloudflare Ingestion Queue (`pipeline-process`) before hitting the **8-Stage Normalizer**.\n\n### The 8-Stage Normalizer Pipeline\n\nThe `normalizer` processes incoming records sequentially to build the Spine context substrate:\n\n```\n[S1: Intake] ──▶ [S2: Classify] ──▶ [S3: Filter] ──▶ [S4: Refine]\n                                                         │\n                                                         ▼\n[S8: Resolve] ◀── [S7.5: Identity] ◀── [S7: Sanity] ◀── [S6: Validate] ◀── [S5: Extract]\n```\n\n1.  **S1: Intake**: Receives raw payloads, assigns a provenance hash (`source_id`), and logs metadata.\n2.  **S2: Classify**: Classifies the incoming record into one of our 7 canonical resource types (Person, Account, Task, Note, Alert, Conversation, Signal) using light AI-in-the-loop if the type is ambiguous.\n3.  **S3: Filter**: Filters out spam, template code, and redundant noise.\n4.  **S4: Refine**: Sanitizes values and normalizes schema mapping to our strict TypeScript definitions.\n5.  **S5: Extract**: Extracts entities, values, or semantic topics.\n6.  **S6: Validate**: Enforces structural constraints and data invariants.\n7.  **S7: Sanity**: Verifies logic consistency, flagging anomalous data points.\n8.  **S7.5: Entity Resolution (Identity Matching)**:\n    - _The Core Challenge_: Mapping scattered identities across platforms (e.g., Salesforce contact `nirmal@org.com` = Jira assignee `Nirmal Prince` = HubSpot deal champion).\n    - _The Resolution_: Normalizer queries the edge `link_data` table to resolve entities to a single, unified canonical ID. If no match exists, a new identity is provisioned.\n9.  **S8: Resolve / Write**: Writes the clean, normalized entity to the **Spine DB / Core Memory Substrate** through a secure service binding.\n\n---\n\n## 3. AI Agent Processing Flow\n\nOnce data is securely resolved inside the Spine DB, the **Intelligence Layer (Agents)** activates to orchestrate reasoning, governance, and updates:\n\n```\nSpine DB Entity Write\n      │\n      ├─▶ Triage Bot (Governance Scoring: auto-approves or stages in Queue)\n      │\n      └─▶ Signal Engine (Always active: compares state and detects deviations)\n                │\n                ▼\n           Twin Interface (Assembles 360-degree context and reasoning)\n                │\n                ▼\n           Operator Handoff (Hands playbooks to local client for execution)\n```\n\n### The Ingestion-to-Agent Lifecycles\n\n#### A. The Triage Bot (Governance Gate)\n\n- **Trigger**: Fires on _every_ proposed write or memory promotion.\n- **Operation**: Evaluates the proposal against our 5-dimension scoring model (Evidence, Category Fit, Novelty, Specificity, Linkage).\n- **Outcome**:\n  - `Score >= 0.85`: Auto-approves the proposal, committing the change immediately to the Spine DB.\n  - `Score 0.60 - 0.84`: Stages the proposal in the Governance Queue (`pending_review`) for human HITL approval.\n  - `Score < 0.60`: Auto-rejects the change, logging the decision logic securely.\n\n#### B. The Signal Engine (Continuous Awareness)\n\n- **Trigger**: Always active, continuously scanning the edge databases.\n- **Operation**: Compares current Spine entity states against baseline goals. It checks for slippage, commitment breaches, or critical anomalies.\n- **Outcome**: Generates an actionable `signal` record in the database if an anomaly is detected, and flags it in the user's Twin brief.\n\n#### C. The Twin (Context Assembly & Conversation)\n\n- **Trigger**: Activated by user session loading, alert triage, or conversation.\n- **Operation**: Assembles a complete 360-degree context by pulling operational signals from edge cache D1 tables and historical decisions from the Spine DB. It ranks and compresses the dataset before reasoning.\n- **Outcome**: Presents the user with clear proposals, reasoning chains, and playbooks.\n\n#### D. The Operator & Handoff Boundary\n\n- **Trigger**: User approves a staged proposal.\n- **Operation**: The Operator parses the approved proposal, compiles the step-by-step instructions into a functional **Playbook JSON**, and dispatches it.\n- **Boundary Handoff**: The server-side Operator hands off the **Playbook JSON** directly to the customer's local client/agent running on their own machine. The local client executes the actual write operations locally (e.g., calling HubSpot APIs), keeping our server runtime footprint at zero.\n"}