diff --git a/docs/validation.md b/docs/validation.md
index 8203f2e..0bd6758 100644
--- a/docs/validation.md
+++ b/docs/validation.md
@@ -104,6 +104,15 @@ the worst case at one timeout per engine rather than one per query:
  "queriesRun": 1, "queriesPlanned": 27}
 ```
 
+A timeout is recorded as one (`EXECUTION_FAILED`, exit code 124) however it
+arrives. The Comunica-backed engines (`comunica`, `hdt`, `cottas`) answer a
+timeout with HTTP 200 and a response cut short by `!TIMEDOUT!`, so that marker
+is read as a timeout, and no query passes unless its result is well-formed
+SPARQL Results JSON. Comunica replaces its worker after a timeout, so the
+endpoint is restarted before the next query; the restart time is reported as
+`endpointRestartSeconds` on that query and is not counted in its
+`wallSeconds`.
+
 Options for a graph that is genuinely too large:
 
 ```bash
diff --git a/src/validation/validation_runner.py b/src/validation/validation_runner.py
index 3b87fc8..9df2933 100644
--- a/src/validation/validation_runner.py
+++ b/src/validation/validation_runner.py
@@ -2609,8 +2609,11 @@ def _envelope(
         time_path: Path | None = None,
         error: str | None = None,
     ) -> dict[str, Any]:
+        problem = result_problem(raw_path) if returncode == 0 else None
+        if problem is not None and error is None:
+            error = problem
         envelope = {
-            "status": "PASS" if returncode == 0 else "EXECUTION_FAILED",
+            "status": "PASS" if returncode == 0 and problem is None else "EXECUTION_FAILED",
             "engine": self.name,
             "exitCode": returncode,
             "wallSeconds": time.monotonic() - started,
@@ -2683,6 +2686,45 @@ def run_query_process(
             return 124, f"query exceeded {timeout}s"
 
 
+#: What a Comunica endpoint appends to a response when its own per-query
+#: timeout fires. By then the 200 status and the start of the result have been
+#: sent, so the timeout reaches the client as a successful HTTP exchange whose
+#: body is cut short by this marker -- not as an error or a closed socket.
+COMUNICA_TIMEOUT_MARKER = b"!TIMEDOUT!"
+
+
+def ends_with_timeout_marker(payload: bytes) -> bool:
+    return payload.rstrip().endswith(COMUNICA_TIMEOUT_MARKER)
+
+
+def result_problem(raw_path: Path) -> str | None:
+    """Why a result an engine reported as successful is not one, or None.
+
+    An engine's exit status says it stopped, not that it answered. A Comunica
+    endpoint whose own timeout fires still returns HTTP 200 with a truncated
+    body, so trusting the status recorded an hour-long timeout as a passing
+    query. A passing execution must have produced SPARQL Results JSON: a
+    ``results.bindings`` list, or an ASK ``boolean``.
+    """
+    try:
+        payload = raw_path.read_bytes()
+    except OSError as error:
+        return f"engine reported success but its result is unreadable: {error}"
+    if ends_with_timeout_marker(payload):
+        return "engine ended its response with its timeout marker (!TIMEDOUT!)"
+    try:
+        document = json.loads(payload)
+    except ValueError as error:
+        return f"engine reported success but its result is not JSON: {error}"
+    if isinstance(document, dict):
+        if isinstance(document.get("boolean"), bool):
+            return None
+        results = document.get("results")
+        if isinstance(results, dict) and isinstance(results.get("bindings"), list):
+            return None
+    return "engine reported success but its result is not SPARQL Results JSON"
+
+
 def is_timeout_error(error: BaseException) -> bool:
     """Whether an exception from an HTTP query is a timeout rather than a fault.
 
@@ -2778,6 +2820,8 @@ def _init_endpoint(self, options: dict[str, Any]) -> None:
         self.endpoint: str | None = None
         self.warmup_seconds: float | None = None
         self.command: list[str] = []
+        #: Set by a query that timed out; the next query restarts the endpoint.
+        self.restart_pending = False
 
     def _endpoint_source_argument(self) -> str:
         """The source argv entry, in whatever form this engine's binary needs."""
@@ -2956,7 +3000,44 @@ def endpoint_describe(self) -> dict[str, Any]:
             "command": list(self.command),
         }
 
+    def _restart_after_timeout(self) -> tuple[float, str | None]:
+        """Replace the endpoint a timed-out query left behind.
+
+        When Comunica's timeout fires it kills the worker and its master forks
+        a new one, which refuses connections until it has bound. Queried at
+        once, every remaining query failed in under a millisecond with
+        "Connection refused" -- eight of thirteen lost to one timeout. When the
+        client gives up first, the worker is still busy with the old query.
+        Either way the endpoint cannot be trusted with the next query, so it is
+        restarted, at the cost of one bind and warm-up, which is charged to
+        neither query. Done lazily, so ``--stop-after-query-timeout`` pays
+        nothing for it.
+        """
+        self.restart_pending = False
+        warmup_seconds = self.warmup_seconds
+        started = time.monotonic()
+        self.stop()
+        try:
+            self._start_endpoint()
+        except RuntimeError as error:
+            self.stop()
+            return time.monotonic() - started, str(error)
+        finally:
+            self.warmup_seconds = warmup_seconds
+        return time.monotonic() - started, None
+
     def execute(self, query_id: str, query_path: Path) -> dict[str, Any]:
+        restart: dict[str, Any] = {}
+        if self.restart_pending:
+            seconds, failure = self._restart_after_timeout()
+            restart["endpointRestartSeconds"] = seconds
+            if failure is not None:
+                restart["endpointRestartError"] = failure
+        envelope = self._execute(query_id, query_path)
+        envelope.update(restart)
+        return envelope
+
+    def _execute(self, query_id: str, query_path: Path) -> dict[str, Any]:
         raw_path = self.raw_dir / f"{query_id}.sparql.json"
         stderr_path = self.raw_dir / f"{query_id}.stderr.txt"
         started = time.monotonic()
@@ -2979,6 +3060,7 @@ def execute(self, query_id: str, query_path: Path) -> dict[str, Any]:
             # 124 is the timeout convention the query loop keys on; anything
             # else here really is an execution failure.
             timed_out = is_timeout_error(error)
+            self.restart_pending = timed_out
             message = (
                 f"query exceeded {self.query_timeout}s" if timed_out else str(error)
             )
@@ -2987,7 +3069,19 @@ def execute(self, query_id: str, query_path: Path) -> dict[str, Any]:
                 started=started, raw_path=raw_path, stderr_path=stderr_path,
                 error=message,
             )
+        # The truncated body is kept as the evidence of what was received.
         raw_path.write_bytes(payload)
+        if ends_with_timeout_marker(payload):
+            self.restart_pending = True
+            message = (
+                f"query exceeded {self.query_timeout}s (the endpoint ended its "
+                f"response with {COMUNICA_TIMEOUT_MARKER.decode()})"
+            )
+            stderr_path.write_text(message, encoding="utf-8")
+            return self._envelope(
+                query_id, query_path, returncode=124, started=started,
+                raw_path=raw_path, stderr_path=stderr_path, error=message,
+            )
         stderr_path.write_bytes(b"")
         return self._envelope(
             query_id, query_path, returncode=0, started=started,
diff --git a/test/test_query_timeout_marker_unit.py b/test/test_query_timeout_marker_unit.py
new file mode 100644
index 0000000..d989bfb
--- /dev/null
+++ b/test/test_query_timeout_marker_unit.py
@@ -0,0 +1,388 @@
+"""A query that times out must never be recorded as passing.
+
+When a Comunica endpoint's own per-query timeout fires, it has already sent
+HTTP 200 and the start of the result, so it ends the response with
+``!TIMEDOUT!`` and kills its worker. The validator read the 200 as success:
+in the vcf-bench-3 scale run (HG005, 1,000,000 records, HDT), q05 ran for
+3600.06s, returned the body below, and was recorded as ``PASS``. The worker's
+replacement was not yet listening, so q06-q13 were then refused in under a
+millisecond each.
+
+These tests pin both halves: the marker is a timeout (exit 124, not a pass),
+any exit-0 result that is not SPARQL Results JSON fails, and the endpoint is
+restarted before the next query so one timeout costs one query, not the rest
+of the suite.
+"""
+
+import argparse
+import copy
+import importlib.util
+import json
+import tempfile
+import unittest
+import urllib.error
+from pathlib import Path
+from unittest import mock
+
+from test import validation_fixtures as fixtures
+from test.helpers import VerboseTestCase
+
+RUNNER_PATH = Path(__file__).resolve().parents[1] / "src" / "validation" / "validation_runner.py"
+
+
+def load_runner():
+    spec = importlib.util.spec_from_file_location("validation_runner_timeout_marker", RUNNER_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+V = load_runner()
+
+#: The q05 response recorded on vcf-bench-3, byte for byte (raw/hdt/
+#: q05_sample_genotype_counts.sparql.json in the v3.3.1 provenance rerun).
+RECORDED_TIMEOUT_BODY = (
+    b'{"head": {"vars":["sampleId","genotypeClass","callCount"]},\n'
+    b'"results": { "bindings": [\n'
+    b"!TIMEDOUT!"
+)
+EMPTY_RESULT = b'{"results": {"bindings": []}}'
+
+
+class FakeComunica:
+    """A Comunica endpoint whose worker dies when its timeout fires.
+
+    ``Popen`` starts it, a query in ``timeouts`` gets the recorded truncated
+    body, and from then on every connection is refused until it is started
+    again -- what the real master does while it forks a replacement worker.
+    """
+
+    def __init__(self, *, timeouts=(), answers=None, restartable=True):
+        self.timeouts = set(timeouts)
+        self.answers = answers or {}
+        self.restartable = restartable
+        self.alive = False
+        self.starts = 0
+        self.queries = []
+
+    def popen(self, *_args, **_kwargs):
+        self.starts += 1
+        self.alive = self.starts == 1 or self.restartable
+        process = mock.MagicMock()
+        process.poll.return_value = None
+        return process
+
+    def post(self, _engine, _endpoint, query, *, timeout):
+        if not self.alive:
+            raise urllib.error.URLError(ConnectionRefusedError(111, "Connection refused"))
+        query_id = query.splitlines()[0].lstrip("# ").strip()
+        if query_id not in self.answers and query_id not in self.timeouts:
+            return EMPTY_RESULT  # bind and warm-up probes
+        self.queries.append(query_id)
+        if query_id in self.timeouts:
+            self.alive = False
+            return RECORDED_TIMEOUT_BODY
+        return json.dumps(self.answers[query_id]).encode("utf-8")
+
+    def patches(self):
+        return [
+            mock.patch.object(V.shutil, "which", return_value="/usr/bin/comunica-sparql-file-http"),
+            mock.patch.object(V, "tool_version", return_value=None),
+            mock.patch.object(V.subprocess, "Popen", side_effect=self.popen),
+            mock.patch.object(V, "_terminate_process_group"),
+            mock.patch.object(V.time, "sleep"),
+            mock.patch.object(V.ComunicaHttpEndpointMixin, "_post", self.post_method()),
+        ]
+
+    def post_method(self):
+        fake = self
+
+        def _post(engine, endpoint, query, *, timeout):
+            return fake.post(engine, endpoint, query, timeout=timeout)
+
+        return _post
+
+
+def query_file(directory, query_id):
+    path = Path(directory) / f"{query_id}.rq"
+    path.write_text(f"# {query_id}\nSELECT * WHERE {{ ?s ?p ?o }}\n", encoding="utf-8")
+    return path
+
+
+class EngineTimeoutMarkerTests(VerboseTestCase):
+    def _run(self, fake, query_ids, *, options=None):
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            source = tmp_path / "cohort.nt"
+            source.write_bytes(b"<s> <p> <o> .\n")
+            engine = V.build_engine(
+                "comunica", source, raw_dir=tmp_path, scratch=tmp_path,
+                options={"query_timeout": 3600, **(options or {})},
+            )
+            patches = fake.patches()
+            for patch in patches:
+                patch.start()
+            try:
+                engine.start()
+                warmup = engine.warmup_seconds
+                results = [engine.execute(qid, query_file(tmp_path, qid)) for qid in query_ids]
+                raws = [Path(r["rawResult"]).read_bytes() for r in results]
+                engine.stop()
+            finally:
+                for patch in reversed(patches):
+                    patch.stop()
+        return engine, warmup, results, raws
+
+    def test_the_recorded_timeout_response_is_a_timeout_not_a_pass(self):
+        fake = FakeComunica(timeouts={"q05"})
+        _engine, _warmup, (result,), (raw,) = self._run(fake, ["q05"])
+        self.assertEqual(result["status"], "EXECUTION_FAILED")
+        # 124 is the timeout convention --stop-after-query-timeout keys on.
+        self.assertEqual(result["exitCode"], 124)
+        self.assertIn("exceeded 3600s", result["error"])
+        self.assertIn("!TIMEDOUT!", result["error"])
+        # What the endpoint sent is kept as the evidence, not blanked.
+        self.assertEqual(raw, RECORDED_TIMEOUT_BODY)
+
+    def test_the_query_after_a_timeout_runs_on_a_restarted_endpoint(self):
+        answer = {"results": {"bindings": [{"n": {"value": "1"}}]}}
+        fake = FakeComunica(timeouts={"q05"}, answers={"q06": answer, "q07": answer})
+        engine, warmup, results, raws = self._run(fake, ["q05", "q06", "q07"])
+        timed_out, after, later = results
+
+        self.assertEqual(timed_out["exitCode"], 124)
+        # Before the fix q06 and q07 were refused: the worker was gone.
+        self.assertEqual(after["status"], "PASS")
+        self.assertEqual(later["status"], "PASS")
+        self.assertEqual(json.loads(raws[1]), answer)
+        self.assertEqual(fake.queries, ["q05", "q06", "q07"])
+        # Restarted exactly once, by the query that followed the timeout, and
+        # the restart is reported on that query rather than hidden.
+        self.assertEqual(fake.starts, 2)
+        self.assertIn("endpointRestartSeconds", after)
+        self.assertNotIn("endpointRestartError", after)
+        self.assertNotIn("endpointRestartSeconds", later)
+        self.assertNotIn("endpointRestartSeconds", timed_out)
+        # The restart's warm-up is not passed off as the original setup's.
+        self.assertEqual(engine.warmup_seconds, warmup)
+
+    def test_a_client_side_timeout_also_restarts_the_endpoint(self):
+        """If urlopen gives up first, the worker is still busy with the old query."""
+        import socket
+
+        fake = FakeComunica(answers={"q06": {"results": {"bindings": []}}})
+        original = fake.post
+
+        def post(engine, endpoint, query, *, timeout):
+            if query.startswith("# q05"):
+                raise urllib.error.URLError(socket.timeout("timed out"))
+            return original(engine, endpoint, query, timeout=timeout)
+
+        fake.post = post
+        _engine, _warmup, (timed_out, after), _raws = self._run(fake, ["q05", "q06"])
+        self.assertEqual(timed_out["exitCode"], 124)
+        self.assertEqual(after["status"], "PASS")
+        self.assertEqual(fake.starts, 2)
+
+    def test_an_ordinary_failure_does_not_restart_the_endpoint(self):
+        """Only a timeout leaves the worker in doubt; a refused query does not."""
+        fake = FakeComunica(answers={"q06": {"results": {"bindings": []}}})
+        original = fake.post
+
+        def post(engine, endpoint, query, *, timeout):
+            if query.startswith("# q05"):
+                raise urllib.error.URLError("query rejected")
+            return original(engine, endpoint, query, timeout=timeout)
+
+        fake.post = post
+        _engine, _warmup, (failed, after), _raws = self._run(fake, ["q05", "q06"])
+        self.assertEqual(failed["exitCode"], 1)
+        self.assertEqual(after["status"], "PASS")
+        self.assertEqual(fake.starts, 1)
+
+    def test_an_endpoint_that_will_not_restart_fails_the_rest_quickly(self):
+        fake = FakeComunica(timeouts={"q05"}, answers={"q06": {}, "q07": {}}, restartable=False)
+        _engine, _warmup, results, _raws = self._run(
+            fake, ["q05", "q06", "q07"], options={"comunica_bind_timeout": 1},
+        )
+        _timed_out, after, later = results
+        self.assertEqual(after["status"], "EXECUTION_FAILED")
+        self.assertIn("did not bind", after["endpointRestartError"])
+        # One restart attempt, not one per remaining query.
+        self.assertEqual(fake.starts, 2)
+        self.assertEqual(later["status"], "EXECUTION_FAILED")
+        self.assertIn("not running", later["error"])
+        self.assertNotIn("endpointRestartSeconds", later)
+
+    def test_a_marker_mid_stream_is_still_a_timeout(self):
+        """Comunica appends the marker to whatever it had already streamed."""
+        self.assertTrue(V.ends_with_timeout_marker(b'{"results":{"bindings":[{"a":1},!TIMEDOUT!\n'))
+        self.assertTrue(V.ends_with_timeout_marker(b"!TIMEDOUT!"))
+        self.assertFalse(V.ends_with_timeout_marker(EMPTY_RESULT))
+        # A literal that happens to contain the word is not a timeout.
+        self.assertFalse(V.ends_with_timeout_marker(
+            b'{"results":{"bindings":[{"v":{"value":"!TIMEDOUT!"}}]}}'
+        ))
+
+
+class ResultProblemTests(VerboseTestCase):
+    """Exit 0 is not a pass unless the engine produced a SPARQL result."""
+
+    def _problem(self, payload):
+        with tempfile.TemporaryDirectory() as td:
+            raw = Path(td) / "q.sparql.json"
+            if payload is not None:
+                raw.write_bytes(payload)
+            return V.result_problem(raw)
+
+    def test_well_formed_results_pass(self):
+        self.assertIsNone(self._problem(EMPTY_RESULT))
+        self.assertIsNone(self._problem(b'{"head":{},"results":{"bindings":[{"x":{"value":"1"}}]}}'))
+        self.assertIsNone(self._problem(b'{"head":{},"boolean":false}'))
+
+    def test_the_recorded_timeout_body_is_named_as_a_timeout(self):
+        self.assertIn("timeout marker", self._problem(RECORDED_TIMEOUT_BODY))
+
+    def test_malformed_results_are_problems(self):
+        for payload in (
+            b"",
+            b'{"results": {"bindings": [',
+            b"<html>502 Bad Gateway</html>",
+            b"[]",
+            b'{"results": {}}',
+            b'{"results": {"bindings": {}}}',
+            b'{"boolean": "true"}',
+            b'{"error": "out of memory"}',
+        ):
+            with self.subTest(payload=payload):
+                self.assertIsNotNone(self._problem(payload))
+
+    def test_a_missing_result_file_is_a_problem(self):
+        self.assertIn("unreadable", self._problem(None))
+
+    def test_every_engine_envelope_applies_the_check(self):
+        """QLever and the file engines share _envelope, so none can pass garbage."""
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            engine = V.build_engine(
+                "qlever", tmp_path / "a.nt", raw_dir=tmp_path, scratch=tmp_path, options={},
+            )
+            raw = tmp_path / "q.sparql.json"
+            raw.write_bytes(b'{"results": {"bindings": [')
+            envelope = engine._envelope(
+                "q", tmp_path / "q.rq", returncode=0, started=V.time.monotonic(),
+                raw_path=raw, stderr_path=tmp_path / "q.stderr.txt",
+            )
+            self.assertEqual(envelope["status"], "EXECUTION_FAILED")
+            self.assertEqual(envelope["exitCode"], 0)
+            self.assertIn("not JSON", envelope["error"])
+
+            raw.write_bytes(EMPTY_RESULT)
+            envelope = engine._envelope(
+                "q", tmp_path / "q.rq", returncode=0, started=V.time.monotonic(),
+                raw_path=raw, stderr_path=tmp_path / "q.stderr.txt",
+            )
+            self.assertEqual(envelope["status"], "PASS")
+            self.assertNotIn("error", envelope)
+
+
+def bindings_for(value):
+    def cell(item):
+        return {key: {"value": str(val)} for key, val in item.items()}
+
+    if isinstance(value, dict):
+        return {"results": {"bindings": [cell(value)]}}
+    return {"results": {"bindings": [cell(row) for row in value]}}
+
+
+class RunValidationRecordsTheTimeoutTests(VerboseTestCase):
+    """The incident end to end: run_validation, a real engine, a fake Comunica."""
+
+    QUERIES = ("q04_filter_distribution", "q05_sample_genotype_counts", "q06_ac_an_distribution")
+
+    def _args(self, tmp_path):
+        return argparse.Namespace(
+            results_dir=tmp_path / "results",
+            representation="expanded",
+            info_representation="structured",
+            header_representation="structured",
+            progress_path=tmp_path / ".progress" / "validation.jsonl",
+            quiet=True,
+            scratch_dir=tmp_path / "scratch",
+            rdf=tmp_path / "sample.nt",
+            rdf_format="nt",
+            engine="comunica",
+            engines=["comunica"],
+            mapping_policy="strict",
+            strict_conformance=False,
+            shacl_shapes=None,
+            query_timeout=3600,
+            validation_time_budget=0,
+            stop_after_query_timeout=False,
+            qlever_memory_gb=4,
+            qlever_port=7019,
+            comunica_port=7020,
+            hdt_port=7021,
+            comunica_bind_timeout=60,
+            comunica_warmup_timeout=60,
+            qlever_startup_timeout=60,
+            qlever_index_arg=[],
+            qlever_server_arg=[],
+            vcf=tmp_path / "sample.vcf",
+            filter_oracle="cyvcf2",
+            dataset_id="sample",
+            queries=self.QUERIES,
+            shacl_max_triples=V.DEFAULT_SHACL_MAX_TRIPLES,
+            node_heap_mb=V.DEFAULT_NODE_HEAP_MB,
+        )
+
+    def test_a_timed_out_query_is_not_a_pass_and_the_next_query_still_runs(self):
+        parser_summary = copy.deepcopy(fixtures.parser_summary("expanded"))
+        parser_summary.setdefault("sourceSha256", "0" * 64)
+        answers = {qid: bindings_for(parser_summary[qid]) for qid in self.QUERIES}
+        fake = FakeComunica(timeouts={"q05_sample_genotype_counts"}, answers=answers)
+
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            (tmp_path / "scratch").mkdir()
+            (tmp_path / "sample.vcf").write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n")
+            (tmp_path / "sample.nt").write_text("<s> <p> <o> .\n")
+            queries = tmp_path / "queries"
+            queries.mkdir()
+            args = self._args(tmp_path)
+            patches = fake.patches() + [
+                mock.patch.object(V, "query_path",
+                                  side_effect=lambda _dir, qid: query_file(queries, qid)),
+                mock.patch.object(V, "parse_vcf", return_value=parser_summary),
+                mock.patch.object(V, "attach_census_expectations",
+                                  side_effect=lambda p, *a, **k: p),
+                mock.patch.object(V, "validate_ntriples",
+                                  return_value={"status": "PASS", "tripleCount": 1}),
+                mock.patch.object(V, "materialize_ntriples",
+                                  return_value=(tmp_path / "sample.nt", {})),
+            ]
+            for patch in patches:
+                patch.start()
+            try:
+                rc = V.run_validation(args)
+            finally:
+                for patch in reversed(patches):
+                    patch.stop()
+            benchmark = json.loads((args.results_dir / "benchmark.json").read_text())
+            executions = json.loads(
+                (args.results_dir / "engines" / "comunica" / "query-executions.json").read_text()
+            )
+
+        per_query = benchmark["engines"]["comunica"]["queries"]
+        self.assertEqual(per_query["q05_sample_genotype_counts"]["status"], "EXECUTION_FAILED")
+        self.assertEqual(executions["q05_sample_genotype_counts"]["exitCode"], 124)
+        self.assertEqual(per_query["q04_filter_distribution"]["status"], "PASS")
+        self.assertEqual(per_query["q06_ac_an_distribution"]["status"], "PASS")
+        self.assertIn("endpointRestartSeconds", executions["q06_ac_an_distribution"])
+        self.assertEqual(fake.queries, list(self.QUERIES))
+        self.assertNotEqual(rc, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()