[Phase 6.2] File Input Pipeline

Mark commented

2026-04-02 20:52:54 +00:00

Owner

Overview

Enable mesh to ingest files, URLs, and documents as input for intents.

Problem Statement

Current mesh can only process intents with text payloads. Cannot:

Read local files (markdown, code, configs)
Fetch URLs (web pages, APIs)
Parse documents (PDF, DOCX, XLSX)
Ingest git repos (project context)
Watch directories for new files

This limits mesh to generating content, not processing existing content.

Proposed Solution

6.2.1: File Ingestor

class FileIngestor:
    """Ingest files, URLs, documents into processable format."""
    
    def ingest_file(self, path: str) -> Document:
        """Read local file, detect format, extract text"""
        
    def ingest_url(self, url: str) -> Document:
        """Fetch URL, parse HTML, extract content"""
        
    def ingest_git_repo(self, repo_url: str, paths: list[str]) -> list[Document]:
        """Clone repo, read specified files"""
        
    def parse_pdf(self, path: str) -> Document:
        """Extract text from PDF"""
        
    def parse_docx(self, path: str) -> Document:
        """Extract text from Word document"""
        
    def parse_xlsx(self, path: str) -> Document:
        """Extract data from spreadsheet"""

6.2.2: Document Schema

document:
  id: str  # UUID
  source: str  # file path, URL, repo
  format: str  # txt, md, pdf, docx, xlsx, html
  content: str  # Extracted text
  metadata:
    title: str
    author: str
    created: datetime
    size_bytes: int
    page_count: int  # For PDFs
    chunk_count: int  # After chunking

6.2.3: Document Chunking

class DocumentChunker:
    """Split large documents for LLM processing."""
    
    def chunk(self, doc: Document, max_tokens: int = 4000) -> list[Chunk]:
        """Split into token-limited chunks"""
        
    def chunk_by_section(self, doc: Document) -> list[Chunk]:
        """Split by markdown headings, paragraphs"""
        
    def overlap_chunks(self, doc: Document, overlap: int = 200) -> list[Chunk]:
        """Chunks with overlap for context preservation"""

6.2.4: Directory Watcher

class DirectoryWatcher:
    """Monitor directory for new files, trigger intents."""
    
    def watch(self, path: str, intent_type: str, recursive: bool = True):
        """Watch path, emit intent when file added"""
        
    def stop(self):
        """Stop watching"""

6.2.5: Intent Integration

# Intent with file input
intent_type: analyze_document
payload:
  input_type: file  # file, url, repo
  source: /path/to/document.pdf
  task: summarize
  output_format: markdown

Deliverables

substrate/file_ingestor.py - FileIngestor class
substrate/document_chunker.py - Chunking logic
PDF parsing (pdfplumber or PyPDF2)
DOCX parsing (python-docx)
URL fetching (requests + BeautifulSoup)
Git repo ingestion (gitpython)
Directory watcher (watchdog)
API endpoints: /v1/ingest/*
Unit tests
Integration tests

Dependencies

#70 (Artifact System) - for storing processed outputs

Estimated Duration

1 week

Success Criteria

Can read local files (txt, md, py, yaml, json)
Can fetch URLs and extract content
Can parse PDFs (text extraction)
Can parse DOCX files
Large documents chunked for LLM limits
Directory watcher triggers intents on new files

Parent: #69 (Phase 6 Meta)
Depends on: #70 (Artifacts)

## Overview Enable mesh to ingest files, URLs, and documents as input for intents. ## Problem Statement Current mesh can only process intents with text payloads. Cannot: - Read local files (markdown, code, configs) - Fetch URLs (web pages, APIs) - Parse documents (PDF, DOCX, XLSX) - Ingest git repos (project context) - Watch directories for new files This limits mesh to generating content, not processing existing content. ## Proposed Solution ### 6.2.1: File Ingestor ```python class FileIngestor: """Ingest files, URLs, documents into processable format.""" def ingest_file(self, path: str) -> Document: """Read local file, detect format, extract text""" def ingest_url(self, url: str) -> Document: """Fetch URL, parse HTML, extract content""" def ingest_git_repo(self, repo_url: str, paths: list[str]) -> list[Document]: """Clone repo, read specified files""" def parse_pdf(self, path: str) -> Document: """Extract text from PDF""" def parse_docx(self, path: str) -> Document: """Extract text from Word document""" def parse_xlsx(self, path: str) -> Document: """Extract data from spreadsheet""" ``` ### 6.2.2: Document Schema ```yaml document: id: str # UUID source: str # file path, URL, repo format: str # txt, md, pdf, docx, xlsx, html content: str # Extracted text metadata: title: str author: str created: datetime size_bytes: int page_count: int # For PDFs chunk_count: int # After chunking ``` ### 6.2.3: Document Chunking ```python class DocumentChunker: """Split large documents for LLM processing.""" def chunk(self, doc: Document, max_tokens: int = 4000) -> list[Chunk]: """Split into token-limited chunks""" def chunk_by_section(self, doc: Document) -> list[Chunk]: """Split by markdown headings, paragraphs""" def overlap_chunks(self, doc: Document, overlap: int = 200) -> list[Chunk]: """Chunks with overlap for context preservation""" ``` ### 6.2.4: Directory Watcher ```python class DirectoryWatcher: """Monitor directory for new files, trigger intents.""" def watch(self, path: str, intent_type: str, recursive: bool = True): """Watch path, emit intent when file added""" def stop(self): """Stop watching""" ``` ### 6.2.5: Intent Integration ```yaml # Intent with file input intent_type: analyze_document payload: input_type: file # file, url, repo source: /path/to/document.pdf task: summarize output_format: markdown ``` ## Deliverables - [ ] `substrate/file_ingestor.py` - FileIngestor class - [ ] `substrate/document_chunker.py` - Chunking logic - [ ] PDF parsing (pdfplumber or PyPDF2) - [ ] DOCX parsing (python-docx) - [ ] URL fetching (requests + BeautifulSoup) - [ ] Git repo ingestion (gitpython) - [ ] Directory watcher (watchdog) - [ ] API endpoints: `/v1/ingest/*` - [ ] Unit tests - [ ] Integration tests ## Dependencies - #70 (Artifact System) - for storing processed outputs ## Estimated Duration 1 week ## Success Criteria - [ ] Can read local files (txt, md, py, yaml, json) - [ ] Can fetch URLs and extract content - [ ] Can parse PDFs (text extraction) - [ ] Can parse DOCX files - [ ] Large documents chunked for LLM limits - [ ] Directory watcher triggers intents on new files ## Related - Parent: #69 (Phase 6 Meta) - Depends on: #70 (Artifacts)