[Phase 6.9] Directory Watchers

Mark commented

2026-04-02 20:55:14 +00:00

Owner

Overview

Directory watchers for continuous file-based document pipelines.

Problem Statement

Currently files must be manually submitted for processing. Cannot:

Monitor directories for new files
Process documents as they arrive
Build continuous ingestion pipelines
Automate document workflows

Proposed Solution

6.9.1: Watcher Configuration

# config/watchers.yaml
watchers:
  - path: ./inbox/articles
    intent_type: process_article
    recursive: true
    file_patterns: ["*.md", "*.txt", "*.pdf"]
    on_event: created  # created, modified, both
    
  - path: ./inbox/reports
    intent_type: analyze_report
    recursive: false
    file_patterns: ["*.xlsx", "*.csv"]

6.9.2: Directory Watcher Service

class DirectoryWatcher:
    """Monitor directories and emit intents for new files."""
    
    def watch(self, path: str, config: WatchConfig):
        """Start watching directory"""
        
    def stop(self, path: str):
        """Stop watching directory"""
        
    def on_file_event(self, event: FileSystemEvent):
        """Handle file event, emit intent"""
        
    def emit_intent(self, file_path: str, config: WatchConfig):
        """Create and emit intent for file"""

6.9.3: Event Handling

# File created in watched directory
File: ./inbox/articles/new_article.md
    ↓
Watcher detects creation
    ↓
FileIngestor.ingest_file(path)
    ↓
Intent emitted:
{
    "intent_type": "process_article",
    "payload": {
        "file_path": "./inbox/articles/new_article.md",
        "document_id": "doc_abc123",
        "content": "...",
        "metadata": {...}
    }
}
    ↓
Intent processed, artifact saved
    ↓
Optional: Move to processed/ or delete

6.9.4: Post-Processing Actions

watchers:
  - path: ./inbox/articles
    intent_type: process_article
    on_complete: move  # move, delete, archive
    completed_path: ./processed/articles

Deliverables

runtime/directory_watcher.py - Watcher service
config/watchers.yaml - Watcher configuration
Integration with FileIngestor (#71)
File pattern matching
Post-processing actions (move, delete)
API endpoints: /v1/watchers/*
Unit tests
Integration tests

Dependencies

#71 (File Input Pipeline) - for file ingestion

Estimated Duration

1 week

Success Criteria

Directories monitored for new files
File events trigger intents
File patterns respected
Processed files moved/archived
Multiple watchers coexist

Parent: #69 (Phase 6 Meta)
Depends on: #71 (File Input)

## Overview Directory watchers for continuous file-based document pipelines. ## Problem Statement Currently files must be manually submitted for processing. Cannot: - Monitor directories for new files - Process documents as they arrive - Build continuous ingestion pipelines - Automate document workflows ## Proposed Solution ### 6.9.1: Watcher Configuration ```yaml # config/watchers.yaml watchers: - path: ./inbox/articles intent_type: process_article recursive: true file_patterns: ["*.md", "*.txt", "*.pdf"] on_event: created # created, modified, both - path: ./inbox/reports intent_type: analyze_report recursive: false file_patterns: ["*.xlsx", "*.csv"] ``` ### 6.9.2: Directory Watcher Service ```python class DirectoryWatcher: """Monitor directories and emit intents for new files.""" def watch(self, path: str, config: WatchConfig): """Start watching directory""" def stop(self, path: str): """Stop watching directory""" def on_file_event(self, event: FileSystemEvent): """Handle file event, emit intent""" def emit_intent(self, file_path: str, config: WatchConfig): """Create and emit intent for file""" ``` ### 6.9.3: Event Handling ```python # File created in watched directory File: ./inbox/articles/new_article.md ↓ Watcher detects creation ↓ FileIngestor.ingest_file(path) ↓ Intent emitted: { "intent_type": "process_article", "payload": { "file_path": "./inbox/articles/new_article.md", "document_id": "doc_abc123", "content": "...", "metadata": {...} } } ↓ Intent processed, artifact saved ↓ Optional: Move to processed/ or delete ``` ### 6.9.4: Post-Processing Actions ```yaml watchers: - path: ./inbox/articles intent_type: process_article on_complete: move # move, delete, archive completed_path: ./processed/articles ``` ## Deliverables - [ ] `runtime/directory_watcher.py` - Watcher service - [ ] `config/watchers.yaml` - Watcher configuration - [ ] Integration with FileIngestor (#71) - [ ] File pattern matching - [ ] Post-processing actions (move, delete) - [ ] API endpoints: `/v1/watchers/*` - [ ] Unit tests - [ ] Integration tests ## Dependencies - #71 (File Input Pipeline) - for file ingestion ## Estimated Duration 1 week ## Success Criteria - [ ] Directories monitored for new files - [ ] File events trigger intents - [ ] File patterns respected - [ ] Processed files moved/archived - [ ] Multiple watchers coexist ## Related - Parent: #69 (Phase 6 Meta) - Depends on: #71 (File Input)

Mark referenced this issue

2026-04-02 20:56:09 +00:00

[Phase 6] Operational Utility - Meta Issue #69

Mark added the

phase-5

label

2026-04-02 20:56:14 +00:00

Mark added

feature-request

and removed

phase-5

labels

2026-05-12 23:27:00 +00:00

Mark referenced this issue

2026-05-18 10:33:49 +00:00

[Phase 6] Operational Utility - Meta Issue #69

Mark added

blocked:needs-decision

and removed

feature-request

labels

2026-05-21 03:11:23 +00:00

Mark removed the

blocked:needs-decision

label

2026-05-28 07:19:15 +00:00

Mark added the

blocked:needs-decision

label

2026-06-07 06:15:41 +00:00

Mark commented

2026-06-07 06:15:42 +00:00

Author

Owner

⏸ Needs Decision

What needs to be decided?

Is this issue still relevant? If yes, remove the blocked:needs-decision label to route back to triage.

Blocking factors:

Didn't meet auto-approve criteria — scope=small, risk=low, spec=partial

Scope: small Risk: low Spec: partial Type: feature
Summary: Add a directory watcher service that monitors filesystem paths for new/modified documents and automatically emits processing intents into the mesh.
Suggestion: Don't reimplement watchdog integration — the heavy lifting already exists in substrate/document_scanner.py's DirectoryWatcher class (polling + watchdog-optimized backends, file metadata, skip patterns, and change detection). Create runtime/directory_watcher_service.py that loads config/watchers.yaml, wraps DirectoryWatcher instances, and pipes file events into the backplane as intent.discovered signals with properly shaped intent payloads. Post-processing (move/delete/archive) is a simple callback. This is an orchestration layer on top of existing infrastructure — about 3-4 files total (service + config + tests).
Rationale: Scope is small because the filesystem watching primitives already exist in substrate/document_scanner.py (DirectoryWatcher class with both polling and watchdog backends). The issue mostly requires an orchestration layer that loads YAML config, wires watchers to the backplane, and handles post-processing. Risk is low because it's additive (no changes to existing code paths) and follows established patterns (watchdog Observer, backplane intent emission). Spec alignment is partial — the intent-ingestion-protocol.md already defines File Watcher as an entrypoint, so it aligns with the architecture, but it's not listed in any current milestone (MILESTONES.md has no Phase 6.9). It's a clean additive feature with no breaking changes.

To unblock: Remove the blocked:needs-decision label. The triage will apply the appropriate outcome based on your choice above.

## ⏸ Needs Decision **What needs to be decided?** - **Is this issue still relevant?** If yes, remove the `blocked:needs-decision` label to route back to triage. **Blocking factors:** - **Didn't meet auto-approve criteria** — scope=small, risk=low, spec=partial --- **Scope:** small **Risk:** low **Spec:** partial **Type:** feature **Summary:** Add a directory watcher service that monitors filesystem paths for new/modified documents and automatically emits processing intents into the mesh. **Suggestion:** Don't reimplement watchdog integration — the heavy lifting already exists in substrate/document_scanner.py's DirectoryWatcher class (polling + watchdog-optimized backends, file metadata, skip patterns, and change detection). Create runtime/directory_watcher_service.py that loads config/watchers.yaml, wraps DirectoryWatcher instances, and pipes file events into the backplane as intent.discovered signals with properly shaped intent payloads. Post-processing (move/delete/archive) is a simple callback. This is an orchestration layer on top of existing infrastructure — about 3-4 files total (service + config + tests). **Rationale:** Scope is small because the filesystem watching primitives already exist in substrate/document_scanner.py (DirectoryWatcher class with both polling and watchdog backends). The issue mostly requires an orchestration layer that loads YAML config, wires watchers to the backplane, and handles post-processing. Risk is low because it's additive (no changes to existing code paths) and follows established patterns (watchdog Observer, backplane intent emission). Spec alignment is partial — the intent-ingestion-protocol.md already defines File Watcher as an entrypoint, so it aligns with the architecture, but it's not listed in any current milestone (MILESTONES.md has no Phase 6.9). It's a clean additive feature with no breaking changes. **To unblock:** Remove the `blocked:needs-decision` label. The triage will apply the appropriate outcome based on your choice above.

Mark added the

blocked:preflight

label

2026-07-31 04:26:46 +00:00

Mark commented

2026-08-11 04:32:36 +00:00

Author

Owner

Approved — removing blocked:needs-decision. Triage failed to auto-approve, but this is low/medium risk, aligned.

Mark removed the

blocked:needs-decision

blocked:preflight