diff --git a/configs/watchdog_listener_config.toml b/configs/watchdog_listener_config.toml
new file mode 100644
index 00000000..e16f049e
--- /dev/null
+++ b/configs/watchdog_listener_config.toml
@@ -0,0 +1,62 @@
+# =======================================================
+# Config for the Watchdog Listener
+# =======================================================
+
+[parser]
+
+[parser.class]
+type = "dynamic"
+name = "LocalToS3BatchListener"
+parts = [
+  ["listeners.watchdog_listener", "WatchdogListenerMixin"],
+  ["uploaders.local_to_s3", "S3UploaderMixin"]
+]
+
+
+[parser.init]
+# The directory where the state file ('upload_state.json') will be stored.
+state_path = "path/to/state/file"
+
+[parser.init.source]
+# The local directory that the listener will monitor for new files.
+path = "path/to/local/dir"
+#File name patterns to include/exclude.
+include_patterns = [".*\\.csv$", ".*\\.txt$"]
+exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"]
+#Batch configuration
+# Trigger the pipeline after 5 files have accumulated...
+batch_max_size = 5
+# ...or after 60 seconds have passed since the first file arrived,
+# whichever comes first.
+batch_max_latency_seconds = 60
+
+[parser.init.target]
+# Settings for the S3 uploader.
+bucket = "bucket-name" # Your S3 bucket name
+prefix = "raw_data/"
+# Set to 'true' to upload files that have been modified.
+allow-overwrite = true
+
+
+
+# Enable logging to a local file using the python's built-in logging.
+[logging.file]
+filepath = "path/to/log/file.txt"
+level = 0             # level of events to include in this logger
+max_size = 21048576   # Max size, in bytes, of a log file before it is rotated out
+max_files = 5         # Max number of files to keep before deleting the oldest
+
+# Enable logging to Sentry using the Sentry SDK API
+# [logging.sentry]
+# dsn = "placeholder"
+# level = 10
+# event_level = 30
+
+[dependencies]
+# Required dependencies for this specific parser configuration.
+# The `prepare.py` script will install these.
+pip = [
+  "boto3",
+  "watchdog"
+]
+
diff --git a/listen.py b/listen.py
new file mode 100644
index 00000000..e07eb990
--- /dev/null
+++ b/listen.py
@@ -0,0 +1,49 @@
+"""
+Entry-point script to run a parser in continuous listener mode.
+This is equivalent to: python run.py config.toml --mode listen
+"""
+
+import argparse
+
+# Reuse the existing functions from run.py to load the config and build the parser.
+from run import load_config, load_parser
+
+
+if __name__ == '__main__':
+    # 1. Set up argument parsing to accept the path to a config file.
+    arg_parser = argparse.ArgumentParser(
+        description="Run a data-net-source parser in continuous listener mode."
+    )
+    arg_parser.add_argument(
+        'config_file',
+        type=str,
+        help='Path to the .toml config file that specifies the parser to run.'
+    )
+    args = arg_parser.parse_args()
+
+    # 2. Load the specified configuration file using the helper from run.py.
+    config = load_config(args.config_file)
+
+    # 3. Dynamically build the parser object from the mixins defined in the config.
+    parser = load_parser(config['parser'])
+
+    # 4. Set up logging as defined in the config, if present.
+    if 'logging' in config:
+        parser.make_loggers(config['logging'])
+
+    # 5. Call the .listen() method to start the long-running, event-driven process.
+    try:
+        parser.listen()
+    except AttributeError as e:
+        if 'listen' in str(e):
+            # Provide a helpful error if the user tries to run a checker-based config
+            # with this script, as it will not have a .listen() method.
+            parser.error(
+                f"The configured parser '{parser.__class__.__name__}' does not have listener capabilities. "
+                f"This config appears to use checker mixins instead of listener mixins. "
+                f"Try using 'python run.py {args.config_file}' for checker mode."
+            )
+        else:
+            parser.error(f"Parser setup failed: {e}", exc_info=True)
+    except Exception as e:
+        parser.error(f"The listener exited with an unexpected error: {e}", exc_info=True)
\ No newline at end of file
diff --git a/run.py b/run.py
index 3b314e19..0411a0a4 100644
--- a/run.py
+++ b/run.py
@@ -12,6 +12,7 @@
 
 
 def load_config(config_fp: Union[str, PathLike]) -> dict:
+
     with open(config_fp, 'r') as f:
         toml_config = toml.load(f)
     return toml_config
@@ -106,7 +107,9 @@ def apply_overrides(config: dict, overrides: List[str], allow_overwrite: bool =
 
 
 if __name__ == '__main__':
-    arg_parser = argparse.ArgumentParser()
+    arg_parser = argparse.ArgumentParser(
+        description="Run a data-net-source parser in batch checker mode."
+    )
     arg_parser.add_argument('config_file', type=str,
                             help='Path to the config file that specifies the parser to run')
     arg_parser.add_argument(
@@ -138,4 +141,14 @@ def apply_overrides(config: dict, overrides: List[str], allow_overwrite: bool =
     if 'logging' in config:
         parser.make_loggers(config['logging'])
 
-    parser.process()
+    # Checker mode (batch processing)
+    try:
+        parser.process()
+    except AttributeError as e:
+        parser.error(
+            f"Parser '{parser.__class__.__name__}' does not support checker mode. "
+            f"This config appears to use listener mixins instead of checker mixins. "
+            f"Try using 'python listen.py {args.config_file}' for continuous listener mode."
+        )
+    except Exception as e:
+        parser.error(f"Checker mode failed: {e}", exc_info=True)
diff --git a/source/common.py b/source/common.py
index 31d3dc56..4d653a2c 100644
--- a/source/common.py
+++ b/source/common.py
@@ -64,6 +64,7 @@ def __init__(self, state_path, source=None, middle=None, target=None):
             self.target_location = target
 
     def process(self):
+
         self.start_notify()
         try:
             to_do = self.check()
@@ -87,6 +88,41 @@ def process(self):
         else:
             self.end_notify(0)
 
+    def listen(self):
+        """
+        Execute the parser in continuous listener mode.
+        This is for event-driven, continuous operation.
+        
+        This orchestrates the listener lifecycle.
+        """
+        self.start_notify()
+        try:
+            # Setup phase - prepare batching and observer
+            self.setup_batching()
+            self.setup_observer()
+            self.start_observer()
+            
+            # Main listening loop - runs continuously
+            self.info(f"Listener is now monitoring for events...")
+            self.run_observer_loop()
+            
+        except KeyboardInterrupt:
+            self.warning("Listener stopped by user. Processing any remaining files...")
+            # Process any remaining files in the batch
+            if hasattr(self, '_process_batch'):
+                self.process_batch()
+            self.end_notify(0)
+        except Exception as e:
+            import sys, traceback
+            self.error(e)
+            self.error(f'Encountered {str(e)}')
+            self.debug(traceback.format_exception(*sys.exc_info()))
+            self.end_notify(1)
+        finally:
+            # Cleanup phase - stop observer
+            if hasattr(self, 'stop_observer'):
+                self.stop_observer()
+
     def check(self):
         """
         Check should look for new data that needs to be uploaded
diff --git a/source/listeners/base.py b/source/listeners/base.py
new file mode 100644
index 00000000..3bde6032
--- /dev/null
+++ b/source/listeners/base.py
@@ -0,0 +1,254 @@
+import time
+import threading
+import copy
+import os
+import json
+from datetime import datetime
+from abc import ABC, abstractmethod
+
+from watchdog.observers import Observer
+from source.common import EMPTY_LOG
+
+
+class BaseListener(ABC):
+    """
+    Abstract base class for all listener mixins.
+
+    This class provides event-driven data processing capabilities with batching,
+    threading, and state management. Listener mixins should be combined with
+    ParserCommon through dynamic class composition, similar to BaseChecker.
+    """
+    state_filename = 'upload_state.json'
+
+    # -------------------------------------------------------------------------
+    # --- Abstract methods to be implemented by concrete listeners ---
+    # -------------------------------------------------------------------------
+
+    @property
+    @abstractmethod
+    def listener_name(self) -> str:
+        """A simple attribute naming the mixin class for later reference."""
+        pass
+
+    @abstractmethod
+    def parse_event(self, raw_event: any) -> tuple[str | None, str | None]:
+        """
+        Parse a raw event from the source to extract file paths.
+
+        This method translates a source-specific event object into file paths
+        that can be processed by the batch system.
+
+        :param raw_event: The native event object from the source (e.g., a
+                          watchdog event, a message from a queue).
+        :return: A tuple of (final_path, old_path). For simple events, old_path 
+                 should be None. For move events, both paths should be provided.
+                 Return (None, None) to indicate the event should be skipped.
+        """
+        pass
+
+    @abstractmethod
+    def save(self, completed: dict):
+        """
+        Save the state of processed files.
+
+        :param completed: A dict with 'success', 'failure', and 'skipped' lists.
+        """
+        pass
+
+    @abstractmethod
+    def clean(self):
+        """Perform cleanup actions, like removing old files or consolidating state entries."""
+        pass
+
+    @abstractmethod
+    def create_event_handler(self):
+        """
+        Create the specific event handler for this listener type.
+        
+        :return: Event handler instance compatible with watchdog Observer
+        """
+        pass
+
+    # -------------------------------------------------------------------------
+    # --- Concrete methods providing Observer management ---
+    # -------------------------------------------------------------------------
+
+    def setup_observer(self):
+        """Initialize the watchdog observer and common configuration."""
+        self.observer = Observer()
+        source_config = self.source_location
+        self.watch_path = source_config.get('path')
+        if not self.watch_path:
+            raise ValueError("[parser.init.source] must contain a 'path' key.")
+        
+        self.recursive = source_config.get('recursive', True)
+        self.info(f"Observer setup for path: {self.watch_path}, recursive: {self.recursive}")
+
+    def start_observer(self):
+        """Start the watchdog observer with the event handler."""
+        if not hasattr(self, 'observer') or not self.observer:
+            self.setup_observer()
+
+        event_handler = self.create_event_handler()
+        self.observer.schedule(event_handler, self.watch_path, recursive=self.recursive)
+        self.observer.start()
+        self.info(f"Started {self.listener_name} observer on {self.watch_path}")
+
+    def stop_observer(self):
+        """Stop and join the watchdog observer."""
+        if hasattr(self, 'observer') and self.observer:
+            self.observer.stop()
+            self.observer.join()
+            self.info(f"Stopped {self.listener_name} observer")
+
+    def run_observer_loop(self):
+        """Run the main observer loop. Cleanup is handled by ParserCommon.listen()."""
+        while True:
+            time.sleep(1)
+
+    # -------------------------------------------------------------------------
+    # --- Concrete methods providing the batching functionality ---
+    # -------------------------------------------------------------------------
+
+    def setup_batching(self):
+        """Initializes all common batching attributes from the configuration."""
+        source_config = self.source_location
+        self._batch_max_size = source_config.get('batch_max_size', 10)
+        self._batch_max_latency_seconds = source_config.get('batch_max_latency_seconds', 60)
+        self._file_buffer = []
+        self._in_flight = set()
+        self._buffer_lock = threading.Lock()
+        self._batch_timer = None
+        self.info(f"Batching configured with max size: {self._batch_max_size} and max latency: {self._batch_max_latency_seconds}s")
+
+    def add_to_batch(self, raw_event: any):
+        """
+        Adds a file from an event to the batch buffer, handles renames,
+        and checks batch triggers using a unified logic.
+        """
+        final_path, old_path = self.parse_event(raw_event)
+        if not final_path:
+            return  # The event was ignored by the parser.
+
+        # Before locking, check if this file is already being processed.
+        if final_path in self._in_flight:
+            self.debug(f"Ignoring event for in-flight file: {final_path}")
+            return
+
+        process_now = False
+        with self._buffer_lock:
+            # If the event was a move, try to remove the old path from the buffer.
+            if old_path and old_path in self._file_buffer:
+                self._file_buffer.remove(old_path)
+
+            # Add the new/final path, but only if it's not already present.
+            if final_path not in self._file_buffer:
+                self._file_buffer.append(final_path)
+                self.info(f"Added to buffer: {final_path}. Current size: {len(self._file_buffer)}")
+
+            if not self._file_buffer:
+                return
+
+            # Start the timer only if this is the first item in a new batch.
+            if len(self._file_buffer) == 1:
+                self._batch_timer = threading.Timer(
+                    self._batch_max_latency_seconds, self.process_batch
+                )
+                self._batch_timer.start()
+
+            # If the buffer is full, set a flag to process the batch.
+            if len(self._file_buffer) >= self._batch_max_size:
+                process_now = True
+        
+        if process_now:
+            self.process_batch()
+
+    def process_batch(self):
+        """Processes all files currently in the buffer."""
+        files_to_process = []
+        with self._buffer_lock:
+            if not self._file_buffer:
+                return
+
+            if self._batch_timer and self._batch_timer.is_alive():
+                self._batch_timer.cancel()
+            
+            files_to_process = self._file_buffer.copy()
+            self._file_buffer.clear()
+            # Add the files to the in-flight set to avoid duplicates.
+            self._in_flight.update(files_to_process)
+
+        self.info(f"Processing batch of {len(files_to_process)} files...")
+        try:
+            tasks = {'to do': files_to_process, 'failure': []}
+            self.info(f"Batch of {len(files_to_process)} files processed successfully to the next stage.")
+            ready = self.transform(tasks)
+            completed = self.upload(ready)
+            self.save(completed)
+            self.clean()
+        except Exception as e:
+            self.error(f"Failed to process batch: {e}", exc_info=True)
+        finally:
+            # Always remove the files from the in-flight set after processing.
+            self._in_flight.difference_update(files_to_process)
+
+    # -------------------------------------------------------------------------
+    # --- State Management Helper Methods (from BaseChecker) ---
+    # -------------------------------------------------------------------------
+
+    def load_state(self) -> dict:
+        """Load the saved upload state from a previously saved file."""
+        with open(os.path.join(self.state_path, self.state_filename)) as f:
+            log = json.load(f)
+        return log
+
+    def write_state(self, state_data: dict):
+        """Write the passed state of successes and failures to file."""
+        with open(os.path.join(self.state_path, self.state_filename), 'w') as log:
+            json.dump(state_data, log, indent=2)
+
+    def clean_outdated(self, full_state: dict) -> dict:
+        """Remove all entries for a particular source file except for the most recent one."""
+        def iter_state(key, state_dict):
+            for obj in state_dict.get(key, []):
+                if 'filename' in obj:
+                    yield obj['filename'], obj
+                elif 'uploaded' in obj:
+                    yield obj['uploaded'], obj
+
+        re_organized = {}
+        for category in full_state:
+            for filename, event in iter_state(category, full_state):
+                if filename not in re_organized:
+                    re_organized[filename] = []
+                re_organized[filename].append((category, event))
+
+        reduced = copy.deepcopy(EMPTY_LOG)
+        for filename, entries in re_organized.items():
+            if len(entries) > 1:
+                entries = sorted(entries, key=lambda x: x[1]['timestamp'])
+            cat, event = entries[-1]
+            reduced[cat].append(event)
+
+        return reduced
+
+    def clean_old_success(self, state: dict) -> dict:
+        """Delete local files that have been successfully uploaded long enough ago."""
+        successes = state.get('success', [])
+        kept_success = []
+        now = datetime.now().timestamp()
+        has_delete = hasattr(self, 'delete_age_hours')
+        for uploaded in successes:
+            age = (now - uploaded['timestamp']) / (60 * 60)
+            if has_delete and self.delete_age_hours >= 0 and age > self.delete_age_hours:
+                self.info(f'Deleting {uploaded.get("uploaded")}')
+                try:
+                    os.remove(uploaded.get("uploaded"))
+                except (FileNotFoundError, TypeError):
+                    self.warning(f'File was already deleted or path was invalid!')
+            else:
+                kept_success.append(uploaded)
+
+        new_log = copy.deepcopy(state)
+        new_log['success'] = kept_success
+        return new_log
\ No newline at end of file
diff --git a/source/listeners/watchdog_listener.py b/source/listeners/watchdog_listener.py
new file mode 100644
index 00000000..0c124645
--- /dev/null
+++ b/source/listeners/watchdog_listener.py
@@ -0,0 +1,104 @@
+import os
+import time
+import threading
+from datetime import datetime
+from watchdog.events import (
+    FileSystemEventHandler,
+    FileClosedEvent,
+    FileMovedEvent,
+    FileCreatedEvent,
+    FileModifiedEvent,
+    RegexMatchingEventHandler
+)
+
+from source.listeners.base import BaseListener
+
+
+class PipelineEventHandler(RegexMatchingEventHandler):
+    """
+    A custom event handler that bridges watchdog and the data-net-source pipeline.
+    """
+    def __init__(self, parser_instance, **kwargs):
+        # Pass all regex arguments to the superclass for filtering.
+        super().__init__(**kwargs)
+        self.parser = parser_instance
+        self.parser.info("RegexMatchingEventHandler initialized. Ready for matched events.")
+
+    def on_created(self, event: FileCreatedEvent):
+        """Called when a file or directory is created."""
+        self.parser.add_to_batch(event)
+
+    def on_modified(self, event: FileModifiedEvent):
+        """Called when a file is modified."""
+        self.parser.add_to_batch(event)
+
+    def on_closed(self, event: FileClosedEvent):
+        """Called when a file opened for writing is closed."""
+        self.parser.add_to_batch(event)
+
+    def on_moved(self, event: FileMovedEvent):
+        """Called when a file is moved or renamed."""
+        self.parser.add_to_batch(event)
+
+
+class WatchdogListenerMixin(BaseListener):
+    """
+    A concrete listener that uses 'watchdog' to monitor a directory and processes
+    new files.
+    """
+    @property
+    def listener_name(self) -> str:
+        return "WatchdogBatchListener"
+
+    def create_event_handler(self):
+        """Create the RegexMatchingEventHandler for this listener."""
+        source_config = self.source_location
+        
+        # Preparing arguments for the RegexMatchingEventHandler from the config file.
+        handler_kwargs = {
+            'regexes': source_config.get('include_patterns', ['.*']), # Default to match everything
+            'ignore_regexes': source_config.get('exclude_patterns', []),
+            'ignore_directories': True, # Using the handler's built-in directory ignoring - helps empty directories
+            'case_sensitive': False
+        }
+        
+        return PipelineEventHandler(parser_instance=self, **handler_kwargs)
+
+    def parse_event(self, raw_event: FileSystemEventHandler) -> tuple[str | None, str | None]:
+        """
+        Parses a watchdog event to extract file paths and filter unwanted files,
+        returning a (final_path, old_path) tuple. Path-based filtering is
+        handled by the RegexMatchingEventHandler.
+        """
+        # Ignore all events that are for directories, double checking here.
+        if raw_event.is_directory:
+            return None, None
+
+        final_path, old_path = None, None
+        if isinstance(raw_event, FileMovedEvent):
+            final_path = raw_event.dest_path
+            old_path = raw_event.src_path
+        else:
+            final_path = raw_event.src_path
+
+        return final_path, old_path
+
+    def save(self, completed: dict):
+        """Saves the state for a completed batch of files."""
+        now = datetime.now().timestamp()
+        for status, events in completed.items():
+            for event in events:
+                event['timestamp'] = now
+
+        current_state = self.load_state()
+        for key, events in completed.items():
+            if events:
+                current_state[key].extend(events)
+        self.write_state(current_state)
+
+    def clean(self):
+        """Performs state file cleanup after a batch is processed."""
+        current_state = self.load_state()
+        deduplicated_state = self.clean_outdated(current_state)
+        final_state = self.clean_old_success(deduplicated_state)
+        self.write_state(final_state)
\ No newline at end of file
diff --git a/source/uploaders/local_to_s3.py b/source/uploaders/local_to_s3.py
index 13a71be8..aa4d74e4 100644
--- a/source/uploaders/local_to_s3.py
+++ b/source/uploaders/local_to_s3.py
@@ -1,4 +1,4 @@
-# source/uploaders/s3.py
+# source/uploaders/local_to_s3.py
 
 import boto3
 import os