diff --git a/pyproject.toml b/pyproject.toml
index 3896eb58..83158dcc 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -48,6 +48,7 @@ dependencies = [
     "PyYAML",
     "s3fs",
     "shapely",
+   "watchdog",
     "xarray",
 ]
 dynamic = ["version"]
diff --git a/src/echodataflow/deployment/core.py b/src/echodataflow/deployment/core.py
index 8fe9301f..b435e97f 100644
--- a/src/echodataflow/deployment/core.py
+++ b/src/echodataflow/deployment/core.py
@@ -27,6 +27,10 @@
     "processes",
     "threads_per_worker",
 }
-ALLOWED_TRIGGER_KEYS = {"expect", "resource_name"}
+ALLOWED_TRIGGER_KEYS = {
+    "expect",
+    "resource_name",
+    "resource_scope",
+}
 ALLOWED_SOURCE_KEYS = {"mode", "git"}
 ALLOWED_GIT_SOURCE_KEYS = {"url", "branch"}
diff --git a/src/echodataflow/deployment/deployment_engine.py b/src/echodataflow/deployment/deployment_engine.py
index 40b3b662..2eddfd3b 100644
--- a/src/echodataflow/deployment/deployment_engine.py
+++ b/src/echodataflow/deployment/deployment_engine.py
@@ -227,18 +227,31 @@ def build_cron(interval: int | None, cron_offset: int = 0) -> str | None:
     return f"*/{interval} * * * *"
 
 
-def build_triggers(trigger_items: list[dict[str, Any]]) -> list[Any]:
-    return [
-        DeploymentEventTrigger(
-            expect={item["expect"]},
-            match_related={
+def build_triggers(
+    trigger_items: list[dict[str, Any]],
+) -> list[Any]:
+    triggers = []
+
+    for item in trigger_items:
+        kwargs = {
+            "expect": {item["expect"]},
+        }
+
+        if item["resource_scope"] == "primary":
+            kwargs["match"] = {
+                "prefect.resource.name": item["resource_name"],
+            }
+        else:
+            kwargs["match_related"] = {
                 "prefect.resource.name": item["resource_name"],
                 "prefect.resource.role": "deployment",
-            },
+            }
+
+        triggers.append(
+            DeploymentEventTrigger(**kwargs)
         )
-        for item in trigger_items
-    ]
 
+    return triggers
 
 def _reject_unknown_keys(
     value: dict[str, Any],
@@ -397,34 +410,55 @@ def validate_triggers(
     When configured, triggers must be a non-empty list of mappings with
     non-empty string values for `expect` and `resource_name`.
     """
+
     triggers = validate_optional_non_empty_list(
         triggers,
         field_name=f"deploy_cfg.flows.{flow_key}.triggers",
         item_label="trigger",
     )
+
     if triggers is None:
         return None
 
     validated_triggers: list[dict[str, Any]] = []
+
     for trigger_item in triggers:
         if not isinstance(trigger_item, dict):
             raise ValueError(f"deploy_cfg.flows.{flow_key}.triggers entries must be mappings")
 
         expect = trigger_item.get("expect")
         resource_name = trigger_item.get("resource_name")
+        resource_scope = trigger_item.get(
+            "resource_scope",
+            "related",
+        )
+
         if not isinstance(expect, str) or not expect.strip():
             raise ValueError(
-                f"deploy_cfg.flows.{flow_key}.triggers entries must define a non-empty 'expect'"
+                f"deploy_cfg.flows.{flow_key}.triggers entries "
+                "must define a non-empty 'expect'"
             )
+
         if not isinstance(resource_name, str) or not resource_name.strip():
             raise ValueError(
-                f"deploy_cfg.flows.{flow_key}.triggers entries must define a non-empty 'resource_name'"
+                f"deploy_cfg.flows.{flow_key}.triggers entries "
+                "must define a non-empty 'resource_name'"
+            )
+
+        if resource_scope not in {
+            "primary",
+            "related",
+        }:
+            raise ValueError(
+                f"deploy_cfg.flows.{flow_key}.triggers "
+                "resource_scope must be 'primary' or 'related'"
             )
 
         validated_triggers.append(
             {
                 "expect": expect.strip(),
                 "resource_name": resource_name.strip(),
+                "resource_scope": resource_scope,
             }
         )
 
diff --git a/src/echodataflow/deployment/flow_registry.py b/src/echodataflow/deployment/flow_registry.py
index d349348e..57aeb6e8 100644
--- a/src/echodataflow/deployment/flow_registry.py
+++ b/src/echodataflow/deployment/flow_registry.py
@@ -66,4 +66,8 @@ class FlowRegistration:
     "update_cache_MVBS": FlowRegistration(
         entrypoint="echodataflow/flows/flows_viz_cloud.py:flow_update_cache_MVBS",
     ),
+    "transect_update": FlowRegistration(
+        entrypoint="echodataflow/flows/flows_transect.py:flow_transect_update",
+        description="Process updates to transect start/end information.",
+    ),
 }
diff --git a/src/echodataflow/flows/flows_transect.py b/src/echodataflow/flows/flows_transect.py
new file mode 100644
index 00000000..5df85f3d
--- /dev/null
+++ b/src/echodataflow/flows/flows_transect.py
@@ -0,0 +1,141 @@
+from pathlib import Path
+
+import pandas as pd
+from prefect import flow
+
+
+def get_changed_transects(
+    current: pd.DataFrame,
+    previous: pd.DataFrame,
+) -> pd.DataFrame:
+    """Return transect segments that are new or have been updated."""
+
+    key_columns = [
+        "transectPart",
+        "transectNumber",
+        "transectStart",
+        "transectEnd",
+    ]
+
+    return (
+        current.merge(
+            previous[key_columns],
+            on=key_columns,
+            how="left",
+            indicator=True,
+        )
+        .query("_merge == 'left_only'")
+        .drop(columns="_merge")
+        .drop_duplicates(subset=key_columns)
+    )
+
+
+def find_overlapping_sv_files(
+    df_sv: pd.DataFrame,
+    start_time: pd.Timestamp,
+    end_time: pd.Timestamp,
+) -> pd.DataFrame:
+    """Return Sv registry rows overlapping a transect time interval."""
+
+    return df_sv[
+        (df_sv["last_ping_time"] >= start_time)
+        & (df_sv["first_ping_time"] <= end_time)
+    ].copy()
+
+
+@flow(log_prints=True)
+def flow_transect_update(
+    path_transect_csv: str,
+    path_snapshot_csv: str,
+    path_main: str,
+    file_Sv_csv: str = "Sv_files.csv",
+):
+    """Identify updated transects and find overlapping Sv files."""
+
+    path_transect = Path(path_transect_csv)
+    path_snapshot = Path(path_snapshot_csv)
+    path_sv_csv = Path(path_main) / file_Sv_csv
+
+    # Read the current transect information, preserving transect identifiers
+    # as strings so values with leading zeros (e.g., "002") are not converted
+    # to integers by pandas
+    current = pd.read_csv(
+        path_transect,
+        dtype={
+            "transectPart": "string",
+            "transectNumber": "string",
+            "transectStart": "string",
+            "transectEnd": "string",
+        },
+    )
+
+    if not path_snapshot.exists():
+        print("No previous transect snapshot found. Initializing snapshot.")
+        current.to_csv(path_snapshot, index=False)
+        return
+
+    previous = pd.read_csv(
+        path_snapshot,
+        dtype={
+            "transectPart": "string",
+            "transectNumber": "string",
+            "transectStart": "string",
+            "transectEnd": "string",
+        },
+    )
+
+    changed = get_changed_transects(current, previous)
+
+    # Only process completed transects
+    changed = changed.dropna(subset=["transectEnd"])
+
+    if changed.empty:
+        print("No new or updated transect segments.")
+        current.to_csv(path_snapshot, index=False)
+        return
+
+    print(f"Found {len(changed)} new or updated transect segment(s):")
+    print(changed)
+
+    # Load Sv tracking information created by raw2Sv
+    if not path_sv_csv.exists():
+        print(f"Sv tracking file does not exist yet: {path_sv_csv}")
+        return
+
+    df_sv = pd.read_csv(
+        path_sv_csv,
+        index_col=0,
+        date_format="ISO8601",
+        parse_dates=["first_ping_time", "last_ping_time"],
+    )
+
+    if df_sv["first_ping_time"].dt.tz is None:
+        df_sv["first_ping_time"] = df_sv["first_ping_time"].dt.tz_localize("UTC")
+
+    if df_sv["last_ping_time"].dt.tz is None:
+        df_sv["last_ping_time"] = df_sv["last_ping_time"].dt.tz_localize("UTC")
+
+    # Find Sv files overlapping each changed transect
+    for _, transect in changed.iterrows():
+        start_time = pd.to_datetime(transect["transectStart"], utc=True)
+        end_time = pd.to_datetime(transect["transectEnd"], utc=True)
+
+        overlapping_sv = find_overlapping_sv_files(
+            df_sv,
+            start_time,
+            end_time,
+        )
+
+        sv_filenames = sorted(overlapping_sv["Sv_filename"].tolist())
+
+        print(
+            f"\nTransect {transect['transectPart']}: "
+            f"{start_time} to {end_time}"
+        )
+        print(f"Found {len(sv_filenames)} overlapping Sv file(s):")
+
+        for filename in sv_filenames:
+            print(f"- {filename}")
+
+    # Save snapshot only after processing the current CSV
+    current.to_csv(path_snapshot, index=False)
\ No newline at end of file
diff --git a/src/echodataflow/utils/file_watcher.py b/src/echodataflow/utils/file_watcher.py
new file mode 100644
index 00000000..cc7f3add
--- /dev/null
+++ b/src/echodataflow/utils/file_watcher.py
@@ -0,0 +1,107 @@
+from collections.abc import Callable
+from pathlib import Path
+
+from watchdog.events import FileSystemEvent, FileSystemEventHandler
+from watchdog.observers import Observer
+
+
+class FileUpdateHandler(FileSystemEventHandler):
+    """Run a callback when the target file is updated."""
+
+    def __init__(
+        self,
+        target_file: str | Path,
+        callback: Callable[[Path], None],
+    ):
+        self.target_file = Path(target_file).resolve()
+        self.callback = callback
+
+    def _handle_path(self, path: str | Path) -> None:
+        event_path = Path(path).resolve()
+
+        if event_path == self.target_file:
+            self.callback(event_path)
+
+    def on_modified(self, event: FileSystemEvent) -> None:
+        if not event.is_directory:
+            self._handle_path(event.src_path)
+
+    def on_created(self, event: FileSystemEvent) -> None:
+        if not event.is_directory:
+            self._handle_path(event.src_path)
+
+    def on_moved(self, event: FileSystemEvent) -> None:
+        if not event.is_directory:
+            self._handle_path(event.dest_path)
+
+
+class FileCreatedHandler(FileSystemEventHandler):
+    """Run a callback when a matching file is created or modified."""
+
+    def __init__(
+        self,
+        callback: Callable[[Path], None],
+        pattern: str,
+    ):
+        self.callback = callback
+        self.pattern = pattern
+
+    def _handle(self, event: FileSystemEvent) -> None:
+        if event.is_directory:
+            return
+
+        event_path = Path(event.src_path).resolve()
+
+        if event_path.match(self.pattern):
+            self.callback(event_path)
+
+    def on_created(self, event: FileSystemEvent) -> None:
+        self._handle(event)
+
+    def on_modified(self, event: FileSystemEvent) -> None:
+        self._handle(event)
+
+
+def watch_file(
+    target_file: str | Path,
+    callback: Callable[[Path], None],
+) -> Observer:
+    """Start watching a file for modifications."""
+
+    target_file = Path(target_file).resolve()
+
+    observer = Observer()
+    observer.schedule(
+        FileUpdateHandler(
+            target_file=target_file,
+            callback=callback,
+        ),
+        str(target_file.parent),
+        recursive=False,
+    )
+    observer.start()
+
+    return observer
+
+
+def watch_directory(
+    directory: str | Path,
+    callback: Callable[[Path], None],
+    pattern: str,
+) -> Observer:
+    """Start watching a directory for matching file changes."""
+
+    directory = Path(directory).resolve()
+
+    observer = Observer()
+    observer.schedule(
+        FileCreatedHandler(
+            callback=callback,
+            pattern=pattern,
+        ),
+        str(directory),
+        recursive=False,
+    )
+    observer.start()
+
+    return observer
\ No newline at end of file
diff --git a/src/echodataflow/utils/raw_monitor.py b/src/echodataflow/utils/raw_monitor.py
new file mode 100644
index 00000000..82488d28
--- /dev/null
+++ b/src/echodataflow/utils/raw_monitor.py
@@ -0,0 +1,32 @@
+from pathlib import Path
+
+from prefect.events import emit_event
+
+from echodataflow.utils.file_watcher import watch_directory
+
+
+RAW_UPDATE_EVENT = "echodataflow.raw.updated"
+RAW_RESOURCE_ID = "raw-monitor"
+
+
+def emit_raw_update_event(path: Path) -> None:
+    """Emit a Prefect event when a RAW file arrives."""
+
+    emit_event(
+        event=RAW_UPDATE_EVENT,
+        resource={
+            "prefect.resource.id": RAW_RESOURCE_ID,
+            "prefect.resource.name": RAW_RESOURCE_ID,
+            "path": str(path),
+        },
+    )
+
+
+def watch_raw_directory(path: str | Path):
+    """Watch a directory for new RAW files."""
+
+    return watch_directory(
+        directory=path,
+        callback=emit_raw_update_event,
+        pattern="*.raw",
+    )
\ No newline at end of file
diff --git a/src/echodataflow/utils/transect_monitor.py b/src/echodataflow/utils/transect_monitor.py
new file mode 100644
index 00000000..fe0ffc12
--- /dev/null
+++ b/src/echodataflow/utils/transect_monitor.py
@@ -0,0 +1,39 @@
+from pathlib import Path
+
+from prefect.events import emit_event
+
+from echodataflow.utils.file_watcher import watch_file
+
+
+TRANSECT_UPDATE_EVENT = "echodataflow.transect.updated"
+TRANSECT_RESOURCE_ID = "transect-start-end-time"
+TRANSECT_RELATED_RESOURCE_ID = "transect-monitor"
+
+
+def emit_transect_update_event(path: Path) -> None:
+    """Emit a Prefect event when the transect CSV is updated."""
+
+    emit_event(
+        event=TRANSECT_UPDATE_EVENT,
+        resource={
+            "prefect.resource.id": TRANSECT_RESOURCE_ID,
+            "path": str(path),
+        },
+        related=[
+            {
+                "prefect.resource.id": TRANSECT_RELATED_RESOURCE_ID,
+                "prefect.resource.name": TRANSECT_RELATED_RESOURCE_ID,
+                "prefect.resource.role": "deployment",
+            }
+        ],
+    )
+
+
+def watch_transect_file(path: str | Path):
+    """Watch the transect start/end CSV and emit a Prefect event on update."""
+
+    return watch_file(
+        target_file=path,
+        callback=emit_transect_update_event,
+    )
+    
\ No newline at end of file
diff --git a/tests/deployment/test_deploy_engine.py b/tests/deployment/test_deploy_engine.py
index 1c1f02e7..cdff2bd4 100644
--- a/tests/deployment/test_deploy_engine.py
+++ b/tests/deployment/test_deploy_engine.py
@@ -327,7 +327,11 @@ def test_validate_deploy_config_accepts_every_allowed_key(install_prefect_stubs)
         "processes",
         "threads_per_worker",
     }
-    assert core.ALLOWED_TRIGGER_KEYS == {"expect", "resource_name"}
+    assert core.ALLOWED_TRIGGER_KEYS == {
+        "expect",
+        "resource_name",
+        "resource_scope",
+    }
     assert core.ALLOWED_SOURCE_KEYS == {"mode", "git"}
     assert core.ALLOWED_GIT_SOURCE_KEYS == {"url", "branch"}
 
@@ -550,6 +554,59 @@ def test_build_deploy_specs_rejects_trigger_missing_resource_name(install_prefec
         )
 
 
+def test_validate_triggers_defaults_resource_scope_to_related(install_prefect_stubs):
+    install_prefect_stubs()
+    engine = importlib.import_module("echodataflow.deployment.deployment_engine")
+
+    triggers = engine.validate_triggers(
+        [{"expect": "test.event", "resource_name": "test-resource"}],
+        flow_key="test_flow",
+    )
+
+    assert triggers == [
+        {
+            "expect": "test.event",
+            "resource_name": "test-resource",
+            "resource_scope": "related",
+        }
+    ]
+
+
+def test_validate_triggers_accepts_primary_resource_scope(install_prefect_stubs):
+    install_prefect_stubs()
+    engine = importlib.import_module("echodataflow.deployment.deployment_engine")
+
+    triggers = engine.validate_triggers(
+        [
+            {
+                "expect": "test.event",
+                "resource_name": "test-resource",
+                "resource_scope": "primary",
+            }
+        ],
+        flow_key="test_flow",
+    )
+
+    assert triggers[0]["resource_scope"] == "primary"
+
+
+def test_validate_triggers_rejects_invalid_resource_scope(install_prefect_stubs):
+    install_prefect_stubs()
+    engine = importlib.import_module("echodataflow.deployment.deployment_engine")
+
+    with pytest.raises(ValueError, match="resource_scope must be 'primary' or 'related'"):
+        engine.validate_triggers(
+            [
+                {
+                    "expect": "test.event",
+                    "resource_name": "test-resource",
+                    "resource_scope": "invalid",
+                }
+            ],
+            flow_key="test_flow",
+        )
+
+
 def test_build_deploy_specs_rejects_inject_time_offset_for_incompatible_flow(
     install_prefect_stubs,
 ):
diff --git a/tests/test_file_watcher.py b/tests/test_file_watcher.py
new file mode 100644
index 00000000..e25df720
--- /dev/null
+++ b/tests/test_file_watcher.py
@@ -0,0 +1,114 @@
+from pathlib import Path
+
+from echodataflow.utils.file_watcher import (
+    FileCreatedHandler,
+    FileUpdateHandler,
+)
+
+class FakeEvent:
+    is_directory = False
+
+    def __init__(
+        self,
+        src_path: Path,
+        dest_path: Path | None = None,
+    ):
+        self.src_path = str(src_path)
+        self.dest_path = str(dest_path) if dest_path else str(src_path)
+
+
+def test_file_update_handler_calls_callback_for_target(tmp_path):
+    target = tmp_path / "transect_start_end_time.csv"
+    target.touch()
+
+    detected = []
+
+    handler = FileUpdateHandler(
+        target_file=target,
+        callback=detected.append,
+    )
+
+    handler.on_modified(FakeEvent(target))
+
+    assert detected == [target.resolve()]
+
+
+def test_file_update_handler_ignores_other_files(tmp_path):
+    target = tmp_path / "transect_start_end_time.csv"
+    other = tmp_path / "other.csv"
+
+    target.touch()
+    other.touch()
+
+    detected = []
+
+    handler = FileUpdateHandler(
+        target_file=target,
+        callback=detected.append,
+    )
+
+    handler.on_modified(FakeEvent(other))
+
+    assert detected == []
+    
+    
+def test_file_update_handler_calls_callback_for_created_target(tmp_path):
+    target = tmp_path / "transect_start_end_time.csv"
+    detected = []
+
+    handler = FileUpdateHandler(
+        target_file=target,
+        callback=detected.append,
+    )
+
+    handler.on_created(FakeEvent(target))
+
+    assert detected == [target.resolve()]
+
+
+def test_file_update_handler_calls_callback_for_moved_target(tmp_path):
+    target = tmp_path / "transect_start_end_time.csv"
+    temporary = tmp_path / "temporary.csv"
+    detected = []
+
+    handler = FileUpdateHandler(
+        target_file=target,
+        callback=detected.append,
+    )
+
+    handler.on_moved(
+        FakeEvent(
+            src_path=temporary,
+            dest_path=target,
+        )
+    )
+
+    assert detected == [target.resolve()]
+
+
+def test_file_created_handler_calls_callback_for_matching_file(tmp_path):
+    raw_file = tmp_path / "example.raw"
+    detected = []
+
+    handler = FileCreatedHandler(
+        callback=detected.append,
+        pattern="*.raw",
+    )
+
+    handler.on_created(FakeEvent(raw_file))
+
+    assert detected == [raw_file.resolve()]
+
+
+def test_file_created_handler_ignores_nonmatching_file(tmp_path):
+    other_file = tmp_path / "example.txt"
+    detected = []
+
+    handler = FileCreatedHandler(
+        callback=detected.append,
+        pattern="*.raw",
+    )
+
+    handler.on_created(FakeEvent(other_file))
+
+    assert detected == []
\ No newline at end of file
diff --git a/tests/test_flow_transect.py b/tests/test_flow_transect.py
new file mode 100644
index 00000000..fa65cfaf
--- /dev/null
+++ b/tests/test_flow_transect.py
@@ -0,0 +1,219 @@
+import pandas as pd
+
+from echodataflow.flows.flows_transect import (
+    find_overlapping_sv_files,
+    flow_transect_update,
+    get_changed_transects,
+)
+
+
+def test_flow_transect_update_initializes_snapshot(tmp_path):
+    transect_csv = tmp_path / "transects.csv"
+    snapshot_csv = tmp_path / "snapshot.csv"
+    path_main = tmp_path / "output"
+    path_main.mkdir()
+
+    pd.DataFrame(
+        {
+            "transectPart": ["001"],
+            "transectNumber": ["001"],
+            "transectStart": ["2024-07-07T00:00:00Z"],
+            "transectEnd": ["2024-07-07T00:10:00Z"],
+        }
+    ).to_csv(transect_csv, index=False)
+
+    flow_transect_update.fn(
+        path_transect_csv=str(transect_csv),
+        path_snapshot_csv=str(snapshot_csv),
+        path_main=str(path_main),
+    )
+
+    assert snapshot_csv.exists()
+
+    snapshot = pd.read_csv(
+        snapshot_csv,
+        dtype={
+            "transectPart": str,
+            "transectNumber": str,
+        },
+    )
+
+    assert len(snapshot) == 1
+    assert snapshot.loc[0, "transectPart"] == "001"
+
+
+def test_flow_transect_update_finds_overlapping_sv(tmp_path, capsys):
+    transect_csv = tmp_path / "transects.csv"
+    snapshot_csv = tmp_path / "snapshot.csv"
+    path_main = tmp_path / "output"
+    path_main.mkdir()
+
+    previous = pd.DataFrame(
+        {
+            "transectPart": ["001"],
+            "transectNumber": ["001"],
+            "transectStart": ["2024-07-07T00:00:00Z"],
+            "transectEnd": ["2024-07-07T00:10:00Z"],
+        }
+    )
+
+    current = pd.concat(
+        [
+            previous,
+            pd.DataFrame(
+                {
+                    "transectPart": ["002"],
+                    "transectNumber": ["002"],
+                    "transectStart": ["2024-07-07T00:20:00Z"],
+                    "transectEnd": ["2024-07-07T00:30:00Z"],
+                }
+            ),
+        ],
+        ignore_index=True,
+    )
+
+    previous.to_csv(snapshot_csv, index=False)
+    current.to_csv(transect_csv, index=False)
+
+    pd.DataFrame(
+        {
+            "Sv_filename": [
+                "before_Sv.zarr",
+                "overlap_Sv.zarr",
+                "after_Sv.zarr",
+            ],
+            "first_ping_time": [
+                "2024-07-07T00:00:00Z",
+                "2024-07-07T00:15:00Z",
+                "2024-07-07T00:40:00Z",
+            ],
+            "last_ping_time": [
+                "2024-07-07T00:05:00Z",
+                "2024-07-07T00:25:00Z",
+                "2024-07-07T00:50:00Z",
+            ],
+        }
+    ).to_csv(path_main / "Sv_files.csv")
+
+    flow_transect_update.fn(
+        path_transect_csv=str(transect_csv),
+        path_snapshot_csv=str(snapshot_csv),
+        path_main=str(path_main),
+    )
+
+    output = capsys.readouterr().out
+
+    assert "Found 1 new or updated transect segment(s)" in output
+    assert "Transect 002" in output
+    assert "Found 1 overlapping Sv file(s)" in output
+    assert "overlap_Sv.zarr" in output
+
+
+def test_get_changed_transects():
+    previous = pd.DataFrame(
+        {
+            "transectPart": ["001"],
+            "transectNumber": ["001"],
+            "transectStart": ["2024-07-07T00:00:00Z"],
+            "transectEnd": ["2024-07-07T00:10:00Z"],
+        }
+    )
+
+    current = pd.concat(
+        [
+            previous,
+            pd.DataFrame(
+                {
+                    "transectPart": ["002"],
+                    "transectNumber": ["002"],
+                    "transectStart": ["2024-07-07T00:20:00Z"],
+                    "transectEnd": ["2024-07-07T00:30:00Z"],
+                }
+            ),
+        ],
+        ignore_index=True,
+    )
+
+    changed = get_changed_transects(current, previous)
+
+    assert len(changed) == 1
+    assert changed.iloc[0]["transectPart"] == "002"
+
+
+def test_find_overlapping_sv_files():
+    df_sv = pd.DataFrame(
+        {
+            "Sv_filename": [
+                "before_Sv.zarr",
+                "overlap_Sv.zarr",
+                "after_Sv.zarr",
+            ],
+            "first_ping_time": pd.to_datetime(
+                [
+                    "2024-07-07T00:00:00Z",
+                    "2024-07-07T00:15:00Z",
+                    "2024-07-07T00:40:00Z",
+                ],
+                utc=True,
+            ),
+            "last_ping_time": pd.to_datetime(
+                [
+                    "2024-07-07T00:05:00Z",
+                    "2024-07-07T00:25:00Z",
+                    "2024-07-07T00:50:00Z",
+                ],
+                utc=True,
+            ),
+        }
+    )
+
+    overlapping = find_overlapping_sv_files(
+        df_sv,
+        pd.Timestamp("2024-07-07T00:20:00Z"),
+        pd.Timestamp("2024-07-07T00:30:00Z"),
+    )
+
+    assert overlapping["Sv_filename"].tolist() == ["overlap_Sv.zarr"]
+    
+def test_flow_transect_update_ignores_open_transect(tmp_path, capsys):
+    transect_csv = tmp_path / "transects.csv"
+    snapshot_csv = tmp_path / "snapshot.csv"
+    path_main = tmp_path / "output"
+    path_main.mkdir()
+
+    previous = pd.DataFrame(
+        {
+            "transectPart": ["001"],
+            "transectNumber": ["001"],
+            "transectStart": ["2024-07-07T00:00:00Z"],
+            "transectEnd": ["2024-07-07T00:10:00Z"],
+        }
+    )
+
+    current = pd.concat(
+        [
+            previous,
+            pd.DataFrame(
+                {
+                    "transectPart": ["002"],
+                    "transectNumber": ["002"],
+                    "transectStart": ["2024-07-07T00:20:00Z"],
+                    "transectEnd": [pd.NA],
+                }
+            ),
+        ],
+        ignore_index=True,
+    )
+
+    previous.to_csv(snapshot_csv, index=False)
+    current.to_csv(transect_csv, index=False)
+
+    flow_transect_update.fn(
+        path_transect_csv=str(transect_csv),
+        path_snapshot_csv=str(snapshot_csv),
+        path_main=str(path_main),
+    )
+
+    output = capsys.readouterr().out
+
+    assert "No new or updated transect segments." in output
\ No newline at end of file
diff --git a/tests/test_raw_monitor.py b/tests/test_raw_monitor.py
new file mode 100644
index 00000000..b90fc98b
--- /dev/null
+++ b/tests/test_raw_monitor.py
@@ -0,0 +1,53 @@
+from pathlib import Path
+
+from echodataflow.utils import raw_monitor
+
+
+def test_emit_raw_update_event(monkeypatch, tmp_path):
+    raw_file = tmp_path / "example.raw"
+    raw_file.touch()
+
+    emitted = {}
+
+    def fake_emit_event(**kwargs):
+        emitted.update(kwargs)
+
+    monkeypatch.setattr(
+        raw_monitor,
+        "emit_event",
+        fake_emit_event,
+    )
+
+    raw_monitor.emit_raw_update_event(raw_file)
+
+    assert emitted["event"] == "echodataflow.raw.updated"
+    assert emitted["resource"]["prefect.resource.id"] == "raw-monitor"
+    assert emitted["resource"]["prefect.resource.name"] == "raw-monitor"
+    assert emitted["resource"]["path"] == str(raw_file)
+
+
+def test_watch_raw_directory(monkeypatch, tmp_path):
+    called = {}
+
+    def fake_watch_directory(
+        directory,
+        callback,
+        pattern,
+    ):
+        called["directory"] = directory
+        called["callback"] = callback
+        called["pattern"] = pattern
+        return "observer"
+
+    monkeypatch.setattr(
+        raw_monitor,
+        "watch_directory",
+        fake_watch_directory,
+    )
+
+    result = raw_monitor.watch_raw_directory(tmp_path)
+
+    assert result == "observer"
+    assert called["directory"] == tmp_path
+    assert called["callback"] is raw_monitor.emit_raw_update_event
+    assert called["pattern"] == "*.raw"
\ No newline at end of file
diff --git a/tests/test_transect_monitor.py b/tests/test_transect_monitor.py
new file mode 100644
index 00000000..6aacd0fc
--- /dev/null
+++ b/tests/test_transect_monitor.py
@@ -0,0 +1,25 @@
+from pathlib import Path
+from echodataflow.utils import transect_monitor
+
+
+def test_emit_transect_update_event(monkeypatch, tmp_path):
+    target = tmp_path / "transect_start_end_time.csv"
+    target.touch()
+
+    emitted = {}
+
+    def fake_emit_event(**kwargs):
+        emitted.update(kwargs)
+
+    monkeypatch.setattr(
+        transect_monitor,
+        "emit_event",
+        fake_emit_event,
+    )
+
+    transect_monitor.emit_transect_update_event(target)
+
+    assert emitted["event"] == "echodataflow.transect.updated"
+    assert emitted["resource"]["prefect.resource.id"] == "transect-start-end-time"
+    assert emitted["resource"]["path"] == str(target)
+    
\ No newline at end of file