diff --git a/capture.py b/capture.py
index 50c587e..9d38327 100755
--- a/capture.py
+++ b/capture.py
@@ -343,14 +343,31 @@ def capture_text(types, app):
     emit({"type": "text", "text": text, "bytes": len(data), "app": app})
 
 
-def main():
-    # The watcher pipes the clipboard payload to us; wl-clipboard blocks on
-    # writes if we close the pipe early, so drain it instead (we still probe
-    # types ourselves below).
+def drain_stdin(timeout=READ_TIMEOUT):
+    """Discard the watcher's payload in fixed chunks, with a total deadline.
+
+    False makes main exit without probing; process exit closes the pipe so
+    a stalled or endless producer cannot keep this helper alive indefinitely.
+    """
+    deadline = time.monotonic() + timeout
     try:
-        sys.stdin.buffer.read()
-    except Exception:
-        pass
+        fd = sys.stdin.fileno()
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return False
+            ready, _, _ = select.select([fd], [], [], remaining)
+            if not ready:
+                return False
+            if not os.read(fd, 65536):
+                return True
+    except (OSError, ValueError):
+        return False
+
+
+def main():
+    if not drain_stdin():
+        return
     types = list_types()
     if not types:
         return
diff --git a/tests/run.sh b/tests/run.sh
index ddfcd3c..82bbb97 100755
--- a/tests/run.sh
+++ b/tests/run.sh
@@ -1,5 +1,6 @@
 #!/bin/bash
-# Run all plugin logic tests. Requires node.
+# Run all plugin logic tests. Requires node and python3.
 set -euo pipefail
 cd "$(dirname "${BASH_SOURCE[0]}")/.."
+python3 -m unittest discover -s tests -p 'test_capture.py'
 exec node --test 'tests/*.mjs'
diff --git a/tests/test_capture.py b/tests/test_capture.py
new file mode 100644
index 0000000..c3e4d18
--- /dev/null
+++ b/tests/test_capture.py
@@ -0,0 +1,56 @@
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+class WatcherInputTests(unittest.TestCase):
+    def helper(self, stdin, timeout=0.2):
+        return subprocess.Popen(
+            [sys.executable, "-c", """
+import capture
+import sys
+import tracemalloc
+tracemalloc.start()
+drain = capture.drain_stdin
+capture.drain_stdin = lambda: drain(timeout=float(sys.argv[1]))
+capture.list_types = lambda: print('probed') or []
+capture.main()
+assert tracemalloc.get_traced_memory()[1] < 1024 * 1024
+""", str(timeout)], stdin=stdin, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        )
+
+    def finish(self, proc, expected, payload=None):
+        try:
+            out, err = proc.communicate(input=payload, timeout=8)
+            self.assertEqual(proc.returncode, 0, err.decode())
+            self.assertEqual(out, expected)
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait()
+
+    def test_large_payload_is_discarded_without_accumulation(self):
+        with tempfile.TemporaryFile() as source:
+            source.truncate(64 * 1024 * 1024)
+            self.finish(self.helper(source, timeout=5), b"probed\n")
+
+    def test_finite_pipe_still_probes(self):
+        self.finish(self.helper(subprocess.PIPE, timeout=5), b"probed\n",
+                    payload=b"clipboard content\n" * 16384)
+
+    def test_stalled_pipe_exits_without_probing(self):
+        reader, writer = os.pipe()
+        try:
+            with os.fdopen(reader, "rb") as source:
+                self.finish(self.helper(source), b"")
+        finally:
+            os.close(writer)
+
+    def test_endless_source_exits_without_probing(self):
+        with open('/dev/zero', 'rb') as source:
+            self.finish(self.helper(source), b"")
+
+    def test_empty_input_still_probes(self):
+        self.finish(self.helper(subprocess.DEVNULL), b"probed\n")