diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 8778b58..6e96de5 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -36,6 +36,22 @@ jobs:
     - name: Test with tox
       run: tox
 
+  e2e-windows:
+    # Flow Launcher only runs on Windows; exercise the real process boundary there.
+    runs-on: windows-latest
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.11'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install tox
+    - name: Run e2e tests
+      run: tox -e py311 -- tests/e2e
+
 concurrency:
   group: tests
   cancel-in-progress: true
\ No newline at end of file
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
new file mode 100644
index 0000000..b75811d
--- /dev/null
+++ b/tests/e2e/conftest.py
@@ -0,0 +1,120 @@
+"""Helpers for spawning fixture plugins as real subprocesses.
+
+These tests exercise the exact process boundary Flow Launcher uses:
+V1 plugins get one JSON-RPC request as argv[1] and answer on stdout in a
+fresh process per request; V2 plugins are a single long-lived process
+speaking newline-delimited JSON-RPC over stdin/stdout.
+"""
+from __future__ import annotations
+
+import json
+import os
+import queue
+import subprocess
+import sys
+import threading
+from pathlib import Path
+from typing import Any, Optional
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+FIXTURES = Path(__file__).resolve().parent / 'fixtures'
+V1_PLUGIN = FIXTURES / 'v1_plugin' / 'main.py'
+V2_PLUGIN = FIXTURES / 'v2_plugin' / 'main.py'
+
+READ_TIMEOUT = 15.0
+
+
+def plugin_env() -> dict:
+    """Subprocess environment with the repo importable even without install."""
+    env = os.environ.copy()
+    env['PYTHONPATH'] = str(REPO_ROOT) + os.pathsep + env.get('PYTHONPATH', '')
+    return env
+
+
+@pytest.fixture
+def run_v1(tmp_path):
+    """Spawn the V1 fixture exactly as Flow Launcher does: fresh process,
+    request JSON in argv[1], response read from stdout."""
+    def _run(request: dict) -> str:
+        completed = subprocess.run(
+            [sys.executable, str(V1_PLUGIN), json.dumps(request)],
+            capture_output=True, text=True, encoding='utf-8',
+            cwd=tmp_path, env=plugin_env(), timeout=READ_TIMEOUT,
+        )
+        assert completed.returncode == 0, completed.stderr
+        return completed.stdout
+    return _run
+
+
+class V2PluginProcess:
+    """A persistent V2 plugin subprocess with a line-reader thread so tests
+    never block forever on a plugin that stops responding."""
+
+    def __init__(self, tmp_path: Path) -> None:
+        self.proc = subprocess.Popen(
+            [sys.executable, str(V2_PLUGIN)],
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+            text=True, encoding='utf-8', bufsize=1,
+            cwd=tmp_path, env=plugin_env(),
+        )
+        self._lines: queue.Queue = queue.Queue()
+        self._reader = threading.Thread(target=self._pump, daemon=True)
+        self._reader.start()
+
+    def _pump(self) -> None:
+        for line in self.proc.stdout:
+            if line.strip():
+                self._lines.put(line)
+
+    def send(self, message: dict) -> None:
+        self.proc.stdin.write(json.dumps(message) + '\n')
+        self.proc.stdin.flush()
+
+    def read_message(self, timeout: float = READ_TIMEOUT) -> dict:
+        try:
+            return json.loads(self._lines.get(timeout=timeout))
+        except queue.Empty:
+            raise AssertionError(
+                f"No response from V2 plugin within {timeout}s; "
+                f"stderr: {self._drain_stderr()}"
+            )
+
+    def request(self, request_id: int, method: str, params: Optional[list] = None,
+                **extra: Any) -> dict:
+        """Send a request and return the response bearing the same id."""
+        message = {'jsonrpc': '2.0', 'id': request_id, 'method': method,
+                   'params': params or [], **extra}
+        self.send(message)
+        response = self.read_message()
+        assert response.get('id') == request_id, (
+            f"Expected response to id={request_id}, got: {response}")
+        return response
+
+    def assert_no_output(self, wait: float = 0.5) -> None:
+        try:
+            line = self._lines.get(timeout=wait)
+        except queue.Empty:
+            return
+        raise AssertionError(f"Expected silence, but plugin wrote: {line!r}")
+
+    def _drain_stderr(self) -> str:
+        if self.proc.poll() is None:
+            return '<process still running>'
+        return self.proc.stderr.read()
+
+    def close(self) -> None:
+        if self.proc.poll() is None:
+            self.proc.kill()
+        self.proc.wait(timeout=5)
+        for stream in (self.proc.stdin, self.proc.stdout, self.proc.stderr):
+            if stream:
+                stream.close()
+
+
+@pytest.fixture
+def v2_plugin(tmp_path):
+    process = V2PluginProcess(tmp_path)
+    yield process
+    process.close()
diff --git a/tests/e2e/fixtures/v1_plugin/main.py b/tests/e2e/fixtures/v1_plugin/main.py
new file mode 100644
index 0000000..3087392
--- /dev/null
+++ b/tests/e2e/fixtures/v1_plugin/main.py
@@ -0,0 +1,24 @@
+"""Fixture plugin spawned as a real subprocess by the e2e tests (V1 protocol)."""
+import json
+
+from pyflowlauncher import Plugin, Result
+
+plugin = Plugin()
+
+
+@plugin.on_method
+def query(q: str):
+    yield Result(
+        title=f"echo: {q}",
+        subtitle=json.dumps(plugin.settings),
+        icon="icon.png",
+    )
+
+
+@plugin.on_method
+def context_menu(data):
+    yield Result(title=f"context: {json.dumps(data)}")
+
+
+if __name__ == '__main__':
+    plugin.run()
diff --git a/tests/e2e/fixtures/v1_plugin/plugin.json b/tests/e2e/fixtures/v1_plugin/plugin.json
new file mode 100644
index 0000000..cc313c3
--- /dev/null
+++ b/tests/e2e/fixtures/v1_plugin/plugin.json
@@ -0,0 +1,12 @@
+{
+    "ID": "e2e-fixture-v1",
+    "Name": "E2E Fixture V1",
+    "Author": "pyFlowLauncher tests",
+    "Version": "0.0.1",
+    "Language": "python",
+    "Description": "Fixture plugin for subprocess e2e tests (V1 protocol).",
+    "Website": "https://github.com/Garulf/pyFlowLauncher",
+    "ExecuteFileName": "main.py",
+    "IcoPath": "icon.png",
+    "ActionKeyword": "e2e"
+}
diff --git a/tests/e2e/fixtures/v2_plugin/main.py b/tests/e2e/fixtures/v2_plugin/main.py
new file mode 100644
index 0000000..87607d8
--- /dev/null
+++ b/tests/e2e/fixtures/v2_plugin/main.py
@@ -0,0 +1,24 @@
+"""Fixture plugin spawned as a real subprocess by the e2e tests (V2 protocol)."""
+import json
+
+from pyflowlauncher import Plugin, Result
+
+plugin = Plugin()
+
+
+@plugin.on_method
+def query(q: str):
+    yield Result(
+        title=f"echo: {q}",
+        subtitle=json.dumps(plugin.settings),
+        icon="icon.png",
+    )
+
+
+@plugin.on_method
+def context_menu(data):
+    yield Result(title=f"context: {json.dumps(data)}")
+
+
+if __name__ == '__main__':
+    plugin.run()
diff --git a/tests/e2e/fixtures/v2_plugin/plugin.json b/tests/e2e/fixtures/v2_plugin/plugin.json
new file mode 100644
index 0000000..adac9cf
--- /dev/null
+++ b/tests/e2e/fixtures/v2_plugin/plugin.json
@@ -0,0 +1,12 @@
+{
+    "ID": "e2e-fixture-v2",
+    "Name": "E2E Fixture V2",
+    "Author": "pyFlowLauncher tests",
+    "Version": "0.0.1",
+    "Language": "python_v2",
+    "Description": "Fixture plugin for subprocess e2e tests (V2 protocol).",
+    "Website": "https://github.com/Garulf/pyFlowLauncher",
+    "ExecuteFileName": "main.py",
+    "IcoPath": "icon.png",
+    "ActionKeyword": "e2e"
+}
diff --git a/tests/e2e/test_v1_subprocess.py b/tests/e2e/test_v1_subprocess.py
new file mode 100644
index 0000000..857be30
--- /dev/null
+++ b/tests/e2e/test_v1_subprocess.py
@@ -0,0 +1,45 @@
+"""End-to-end tests for the V1 protocol across a real process boundary.
+
+Flow Launcher spawns a fresh process per request with the JSON-RPC request
+serialized into argv[1] and deserializes whatever the process writes to
+stdout. These tests do exactly that — they catch buffering, encoding, and
+stream-pollution bugs that in-process tests with patched stdio cannot.
+"""
+import json
+
+
+def test_query_roundtrip(run_v1):
+    stdout = run_v1({'method': 'query', 'parameters': ['hello'], 'settings': {}})
+    response = json.loads(stdout)
+    assert response['Result'][0]['Title'] == 'echo: hello'
+
+
+def test_stdout_is_pure_json(run_v1):
+    """Nothing (logging, warnings, prints) may pollute the response stream."""
+    stdout = run_v1({'method': 'query', 'parameters': ['x'], 'settings': {}})
+    json.loads(stdout)
+
+
+def test_settings_reach_the_plugin(run_v1):
+    settings = {'api_key': 'abc123', 'max_results': 5}
+    stdout = run_v1({'method': 'query', 'parameters': ['q'], 'settings': settings})
+    response = json.loads(stdout)
+    assert json.loads(response['Result'][0]['SubTitle']) == settings
+
+
+def test_unicode_query(run_v1):
+    stdout = run_v1({'method': 'query', 'parameters': ['héllo ☃'], 'settings': {}})
+    response = json.loads(stdout)
+    assert response['Result'][0]['Title'] == 'echo: héllo ☃'
+
+
+def test_context_menu(run_v1):
+    stdout = run_v1({'method': 'context_menu', 'parameters': [['ctx-data']], 'settings': {}})
+    response = json.loads(stdout)
+    assert response['Result'][0]['Title'] == 'context: ["ctx-data"]'
+
+
+def test_empty_query(run_v1):
+    stdout = run_v1({'method': 'query', 'parameters': [''], 'settings': {}})
+    response = json.loads(stdout)
+    assert response['Result'][0]['Title'] == 'echo: '
diff --git a/tests/e2e/test_v2_subprocess.py b/tests/e2e/test_v2_subprocess.py
new file mode 100644
index 0000000..06f1b65
--- /dev/null
+++ b/tests/e2e/test_v2_subprocess.py
@@ -0,0 +1,66 @@
+"""End-to-end tests for the V2 protocol across a real process boundary.
+
+Flow Launcher keeps one plugin process alive and speaks newline-delimited
+JSON-RPC over its stdin/stdout. These tests drive a real subprocess with the
+exact message shapes the host sends (query params as objects, lifecycle
+methods, $/ notifications) and assert on the raw response stream.
+"""
+import json
+
+
+def query_titles(response: dict) -> list:
+    return [r['Title'] for r in response['result']['result']]
+
+
+def test_initialize_then_query(v2_plugin):
+    response = v2_plugin.request(1, 'initialize', [{}])
+    assert response['result'] == {}
+    assert response['error'] is None
+
+    # The host sends the query as an object: [Query, Settings.Inner]
+    response = v2_plugin.request(2, 'query', [{'search': 'hello', 'rawQuery': 'e2e hello'}, {}])
+    assert query_titles(response) == ['echo: hello']
+
+
+def test_persistent_process_serves_many_queries(v2_plugin):
+    for i, term in enumerate(('one', 'two', 'three'), start=1):
+        response = v2_plugin.request(i, 'query', [{'search': term}, {}])
+        assert query_titles(response) == [f'echo: {term}']
+    assert v2_plugin.proc.poll() is None
+
+
+def test_settings_from_query_params(v2_plugin):
+    settings = {'token': 'xyz', 'limit': 3}
+    response = v2_plugin.request(1, 'query', [{'search': 'q'}, settings])
+    subtitle = response['result']['result'][0]['SubTitle']
+    assert json.loads(subtitle) == settings
+
+
+def test_cancel_request_notification_is_ignored(v2_plugin):
+    v2_plugin.send({'jsonrpc': '2.0', 'method': '$/cancelRequest', 'params': {'id': 99}})
+    v2_plugin.assert_no_output()
+    response = v2_plugin.request(2, 'query', [{'search': 'after-cancel'}, {}])
+    assert query_titles(response) == ['echo: after-cancel']
+
+
+def test_context_menu(v2_plugin):
+    response = v2_plugin.request(1, 'context_menu', [['ctx-data']])
+    assert query_titles(response) == ['context: ["ctx-data"]']
+
+
+def test_unicode_roundtrip(v2_plugin):
+    response = v2_plugin.request(1, 'query', [{'search': 'héllo ☃'}, {}])
+    assert query_titles(response) == ['echo: héllo ☃']
+
+
+def test_close_shuts_down_cleanly(v2_plugin):
+    response = v2_plugin.request(1, 'close', [])
+    assert response['result'] == {}
+    assert v2_plugin.proc.wait(timeout=10) == 0
+
+
+def test_malformed_line_does_not_kill_the_process(v2_plugin):
+    v2_plugin.proc.stdin.write('this is not json\n')
+    v2_plugin.proc.stdin.flush()
+    response = v2_plugin.request(1, 'query', [{'search': 'still-alive'}, {}])
+    assert query_titles(response) == ['echo: still-alive']