diff --git a/README.md b/README.md
index 01c6554..67558e3 100644
--- a/README.md
+++ b/README.md
@@ -108,6 +108,42 @@ following behavior applies to all of them:
   Tasks that don't finish in the window are left in the stream for Redis reclaim.
   Set the deployment's `terminationGracePeriodSeconds` comfortably above
   `WORKER_DRAIN_TIMEOUT_SEC`.
+- **Whole-query timeout budget (`QUERY_TIMEOUT_SEC`)** — see below.
+
+##### Query timeout budget
+
+The ARS and the other external callers stop waiting for a Shepherd query after
+about five minutes, and the synchronous `/query` endpoint gives up around the
+same point. Work done past that is work nobody receives — it only takes worker
+slots (and process-pool children) away from queries that can still be answered.
+
+So the server stamps each query with an absolute deadline at intake, and that
+deadline travels with the task from operation to operation. Every worker checks
+it as it picks a task up (in `shepherd_utils.shared.get_tasks`, on both freshly
+delivered and reclaimed messages). If the budget is spent the worker does *not*
+run the operation: it drops the rest of the workflow and routes the query
+straight to `finish_query`, which ends it the way any other query ends — state
+`COMPLETED` in Postgres with a `TIMEOUT` status, callback rows reaped, logs
+saved (including the line explaining why the response is partial), and whatever
+was gathered POSTed to the callback URL. A synchronous caller therefore gets a
+partial response instead of waiting out its own timeout for nothing.
+
+| Setting | Meaning |
+| --- | --- |
+| `QUERY_TIMEOUT_SEC` | The budget, in seconds (default 300). `0` disables deadlines entirely — tasks then run however old they are, as they did before. |
+
+A client that explicitly asks to wait longer (TRAPI `parameters.timeout`) is not
+cut short: the larger of the two wins. Two streams are exempt — `finish_query`,
+which *is* the wrap-up, and `merge_message`, which folds in callbacks an
+upstream service has already done the work for. Tasks with no deadline (a
+payload enqueued before this shipped) are never expired, so a rollout is safe
+mid-flight.
+
+This is a fast path that settles a query at the moment it goes over. The
+monitor's abandoned-query reaper (`MONITOR_ABANDONED_QUERY_SEC`, default 600s)
+remains the backstop for queries that go over *without* any worker picking a
+task up for them — e.g. one whose driving worker died with nothing left in a
+stream.
 
 ##### Kubernetes sizing (Helm)
 
diff --git a/shepherd_server/base_routes.py b/shepherd_server/base_routes.py
index 04041a9..a424038 100644
--- a/shepherd_server/base_routes.py
+++ b/shepherd_server/base_routes.py
@@ -35,6 +35,7 @@
     setup_logging,
 )
 from shepherd_utils.otel import setup_tracer
+from shepherd_utils.task_deadline import deadline_field, query_deadline
 
 setup_logging()
 
@@ -147,6 +148,15 @@ async def run_query(
             logger,
             target=target_name,
         )
+        # Stamp the whole-query deadline once, here, and let it ride along with
+        # the task from operation to operation. Workers check it as they pick a
+        # task up and wrap the query up instead of running work whose answer
+        # would land after the caller has stopped waiting.
+        deadline = query_deadline(query)
+        if deadline is not None:
+            logger.debug(
+                f"Query {query_id} has {deadline - time.time():.0f}s to finish."
+            )
         await add_task(
             target,
             {
@@ -156,6 +166,7 @@ async def run_query(
                 "log_level": level_number,
                 "otel": json.dumps(span_carrier),
                 "metadata": json.dumps({}),
+                **deadline_field(deadline),
             },
             logger,
         )
diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index d6243f4..0c6b07a 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -166,6 +166,22 @@ class Settings(BaseSettings):
     reclaim_interval_sec: int = 10
     reclaim_max_batch: int = 50
 
+    # Whole-query wall-clock budget, in seconds. The ARS and the other external
+    # callers stop waiting for a Shepherd query after ~5 minutes, and the
+    # synchronous /query endpoint gives up around the same point, so work done
+    # past this is work nobody receives -- it only takes worker slots (and
+    # process-pool children) away from queries that can still be answered. Each
+    # query is stamped with an absolute deadline at intake and it travels with
+    # the task; a worker that picks up a task whose query is past it skips the
+    # operation and routes the query straight to finish_query, which settles
+    # its state in Postgres, reaps its callbacks, saves its logs and delivers
+    # whatever was gathered. A client asking to wait longer (TRAPI
+    # parameters.timeout) is not cut short -- the larger of the two wins. Set to
+    # 0 to disable the deadline entirely (the previous behavior: tasks run
+    # however old they are, and only the monitor's abandoned-query reaper --
+    # monitor_abandoned_query_sec -- eventually settles the row).
+    query_timeout_sec: float = 300.0
+
     # Poison-pill circuit breaker. A task that keeps killing its worker before it
     # can ack (e.g. a payload so large that decoding/sorting it trips the cgroup
     # OOM killer -- an uncatchable SIGKILL that no in-process try/except can turn
diff --git a/shepherd_utils/shared.py b/shepherd_utils/shared.py
index c0879a0..d9675d8 100644
--- a/shepherd_utils/shared.py
+++ b/shepherd_utils/shared.py
@@ -27,6 +27,11 @@
 from .heartbeat import Heartbeat
 from .logger import attach_query_handler, resolve_log_level, setup_logging
 from .reclaim import reclaim_orphaned
+from .task_deadline import (
+    TIMEOUT_STATUS,
+    carry_deadline,
+    seconds_overdue,
+)
 
 # Cap each per-stream duration queue so a stopped monitor can't OOM the broker.
 # 10k entries per stream is well above what we'd accumulate in a 30s drain
@@ -351,15 +356,17 @@ async def _terminate_task(
     ara_task: Tuple[str, dict],
     logger: logging.Logger,
     reason: str,
+    status: str = "ERROR",
 ) -> None:
-    """Terminally clear a message: ack+delete it and end its query with ERROR.
+    """Terminally clear a message: ack+delete it and end its query.
 
     Making a message terminal: ack + delete it (``mark_task_as_complete`` does
     ``XACK`` then ``XDEL``) so it leaves the PEL and the stream, and -- when we
-    can still identify the parent query -- route it to ``finish_query`` with an
-    ERROR status so the query ends cleanly instead of hanging, mirroring
-    ``handle_task_failure``. Shared by the "unprocessable message" and
-    "poison-pill / over-delivered" paths.
+    can still identify the parent query -- route it to ``finish_query`` so the
+    query ends cleanly instead of hanging, mirroring ``handle_task_failure``.
+    Shared by the "unprocessable message", "poison-pill / over-delivered" and
+    "past its deadline" paths; ``status`` is what ``finish_query`` records in
+    ``shepherd_brain`` and is what tells those cases apart afterwards.
     """
     msg_id = ara_task[0]
     fields = ara_task[1] if len(ara_task) > 1 and isinstance(ara_task[1], dict) else {}
@@ -376,8 +383,9 @@ async def _terminate_task(
                     "workflow": "[]",
                     "log_level": fields.get("log_level", 20),
                     "otel": fields.get("otel", "{}"),
-                    "status": "ERROR",
+                    "status": status,
                     "metadata": fields.get("metadata", "{}"),
+                    **carry_deadline(fields),
                 },
                 logger,
             )
@@ -417,6 +425,85 @@ async def _discard_unprocessable_task(
     )
 
 
+# Streams whose tasks are never expired, however old their query is.
+#
+# ``finish_query`` *is* the wrap-up: expiring it would leave unset exactly the
+# state expiring is meant to settle, and a query would never end.
+# ``merge_message`` sits off the workflow chain -- its tasks are enqueued by the
+# /callback endpoint for work an upstream service has already done and paid for,
+# and dropping one would strand that callback in the ready index rather than
+# saving anything.
+_DEADLINE_EXEMPT_STREAMS = frozenset({"finish_query", "merge_message"})
+
+
+async def _expire_task(
+    stream: str,
+    group: str,
+    ara_task: Tuple[str, dict],
+    logger: logging.Logger,
+    overdue: float,
+) -> None:
+    """Wrap a query up instead of running an operation nobody is waiting for.
+
+    The caller stopped waiting when the query passed its budget (see
+    ``shepherd_utils.task_deadline``), so running this operation would spend a
+    worker slot on an answer that can't be delivered -- and every operation
+    after it would do the same. Instead the query finishes the ordinary way:
+    ``finish_query`` sets its terminal state in Postgres, reaps its callback
+    rows, saves its logs and POSTs whatever was gathered to the callback URL, so
+    the databases end up exactly as they do for any completed query, with a
+    ``TIMEOUT`` status recording why the response is partial.
+
+    The explanation is logged and flushed to the query's own log list *before*
+    the wrap-up is enqueued: ``finish_query`` reads those logs into the response
+    it delivers, and it may well pick the task up before this coroutine returns.
+    """
+    fields = ara_task[1] if len(ara_task) > 1 and isinstance(ara_task[1], dict) else {}
+    response_id = fields.get("response_id")
+    reason = (
+        f"query exceeded its time budget {overdue:.1f}s ago; skipping the "
+        f"{stream} operation and returning what has been gathered so far"
+    )
+
+    async def _flush_logs() -> None:
+        if not response_id:
+            return
+        try:
+            await save_logs(response_id, logger)
+        except Exception as e:
+            logger.error(f"Failed to save logs for timed-out query: {e}")
+
+    logger.warning(f"Query timed out: {reason}.")
+    await _flush_logs()
+    await _terminate_task(stream, group, ara_task, logger, reason, TIMEOUT_STATUS)
+    # ``logger`` is this query's own logger, and what _terminate_task logged is
+    # still sitting in its handler. Nothing else will run for this query, so
+    # flush again rather than leaving the entry queued for a flush that never
+    # comes.
+    await _flush_logs()
+
+
+async def _handled_as_expired(
+    stream: str,
+    group: str,
+    ara_task: Tuple[str, dict],
+    logger: logging.Logger,
+) -> bool:
+    """Whether this task's query is past its deadline (and has been wrapped up).
+
+    Tasks with no deadline -- an exempt stream, a payload from a server that
+    predates the field, or a deployment with the budget disabled -- are never
+    expired, so this is a no-op unless a deadline says otherwise.
+    """
+    if stream in _DEADLINE_EXEMPT_STREAMS:
+        return False
+    overdue = seconds_overdue(ara_task[1] if len(ara_task) > 1 else None)
+    if overdue <= 0:
+        return False
+    await _expire_task(stream, group, ara_task, logger, overdue)
+    return True
+
+
 class TaskSlots:
     """Concurrency limiter that also tracks how many tasks are *actively running*.
 
@@ -575,6 +662,12 @@ async def get_tasks(
                     )
                     task_limiter.release_slot()
                     continue
+                # Reclaim can hand back a message that has been sitting in a
+                # dead consumer's PEL for a while, so this is exactly where a
+                # query is most likely to have outlived its budget.
+                if await _handled_as_expired(stream, group, ara_task, task_logger):
+                    task_limiter.release_slot()
+                    continue
                 # Dispatching real work: count it as in-flight until the worker
                 # releases the slot in its finally.
                 task_limiter.dispatch()
@@ -609,6 +702,12 @@ async def get_tasks(
                 )
                 task_limiter.release_slot()
                 continue
+            # The query may have run out of budget while this task waited its
+            # turn in the stream (or while an earlier operation ran long). Wrap
+            # it up rather than starting work whose answer arrives too late.
+            if await _handled_as_expired(stream, group, ara_task, task_logger):
+                task_limiter.release_slot()
+                continue
             # send the task to a async background task
             # this could be async, multi-threaded, etc.
             task_limiter.dispatch()
@@ -647,6 +746,9 @@ async def wrap_up_task(
             "log_level": task[1].get("log_level", 20),
             "otel": task[1]["otel"],
             "metadata": task[1]["metadata"],
+            # The budget is measured from intake, so it travels with the query
+            # rather than restarting at each operation.
+            **carry_deadline(task[1]),
         },
         logger,
     )
@@ -677,6 +779,7 @@ async def handle_task_failure(
             "otel": task[1]["otel"],
             "status": "ERROR",
             "metadata": task[1]["metadata"],
+            **carry_deadline(task[1]),
         },
         logger,
     )
diff --git a/shepherd_utils/task_deadline.py b/shepherd_utils/task_deadline.py
new file mode 100644
index 0000000..a5b1278
--- /dev/null
+++ b/shepherd_utils/task_deadline.py
@@ -0,0 +1,128 @@
+"""Whole-query timeout budget carried alongside every task.
+
+The ARS and the other external callers give up on a Shepherd query after about
+five minutes, and the synchronous ``/query`` endpoint stops holding its
+connection open around the same point. Work done after that is work nobody is
+waiting for -- but the pipeline had no notion of it: a task handed from worker
+to worker kept going indefinitely, each hop taking a concurrency slot (and, for
+the CPU-bound workers, a process-pool child) away from queries whose answer can
+still be delivered, while the query's row sat in a non-terminal state until the
+monitor's abandoned-query reaper eventually swept it.
+
+So each query now carries an absolute deadline, stamped once at intake and
+passed along with the task from operation to operation. A worker checks it as
+it picks the task up; if the budget is spent it hands the query to
+``finish_query`` instead of running the operation (see
+``shepherd_utils.shared``), so the query ends the ordinary way -- terminal state
+in Postgres, callback rows reaped, logs saved, whatever was gathered POSTed to
+the callback URL -- rather than being dropped on the floor.
+
+The deadline is stored as absolute epoch seconds rather than a start time plus
+a budget: one field per task, and every stage agrees on the same instant without
+needing to know how the budget was chosen. Comparing it against the local clock
+assumes containers agree on the time to within a second or so, which is true of
+NTP-synced hosts and, for the common case, is the same clock anyway.
+"""
+
+import time
+from typing import Any, Dict, Mapping, Optional
+
+from .config import settings
+
+# Task payload field holding the query's absolute deadline (epoch seconds).
+DEADLINE_FIELD = "query_deadline"
+
+# ``status`` recorded in ``shepherd_brain`` for a query cut short by its budget.
+# Deliberately distinct from the ERROR a failed operation records: nothing went
+# wrong here, there just wasn't time left to keep going.
+TIMEOUT_STATUS = "TIMEOUT"
+
+
+def query_budget(query: Optional[Mapping[str, Any]] = None) -> float:
+    """How many seconds of work a query gets before it stops being useful.
+
+    ``settings.query_timeout_sec`` is the fleet-wide budget, matching what
+    upstream callers wait. A client that explicitly asks to wait *longer* (TRAPI
+    ``parameters.timeout``, which the sync endpoint also polls against) is not
+    cut short: its own number wins when it is the larger of the two. An
+    unparseable value is ignored rather than failing the query. Returns 0 when
+    the budget is disabled, meaning "no deadline".
+    """
+    budget = float(settings.query_timeout_sec)
+    if budget <= 0:
+        return 0.0
+    parameters = (query or {}).get("parameters") or {}
+    requested = parameters.get("timeout") if isinstance(parameters, Mapping) else None
+    if requested is not None:
+        try:
+            budget = max(budget, float(requested))
+        except (TypeError, ValueError):
+            pass
+    return budget
+
+
+def query_deadline(
+    query: Optional[Mapping[str, Any]] = None,
+    start: Optional[float] = None,
+) -> Optional[float]:
+    """Absolute epoch time by which this query's work must be finished.
+
+    ``None`` when the budget is disabled, which leaves the task unstamped and
+    therefore never expired.
+    """
+    budget = query_budget(query)
+    if budget <= 0:
+        return None
+    return (time.time() if start is None else start) + budget
+
+
+def deadline_field(deadline: Optional[float]) -> Dict[str, str]:
+    """Payload fragment carrying ``deadline``, or nothing when there isn't one.
+
+    Millisecond precision is far finer than anything that depends on it, and
+    keeps the stream entry short.
+    """
+    if deadline is None:
+        return {}
+    return {DEADLINE_FIELD: f"{float(deadline):.3f}"}
+
+
+def carry_deadline(fields: Any) -> Dict[str, str]:
+    """Payload fragment propagating a task's deadline to its follow-on task.
+
+    Every hop that enqueues the next operation passes this through, so the
+    budget is measured from the query's intake rather than restarting at each
+    stage.
+    """
+    return deadline_field(get_deadline(fields))
+
+
+def get_deadline(fields: Any) -> Optional[float]:
+    """Read the deadline off a task payload, or ``None`` if it hasn't got one."""
+    if not hasattr(fields, "get"):
+        return None
+    raw = fields.get(DEADLINE_FIELD)
+    if raw is None or raw == "":
+        return None
+    try:
+        return float(raw)
+    except (TypeError, ValueError):
+        return None
+
+
+def seconds_overdue(fields: Any, now: Optional[float] = None) -> float:
+    """How far past its deadline this task's query is; 0.0 if it isn't.
+
+    Fails open: a task with no deadline -- one enqueued by an older server, or
+    by a path that doesn't stamp one -- is never overdue and runs exactly as it
+    did before.
+    """
+    deadline = get_deadline(fields)
+    if deadline is None:
+        return 0.0
+    return max(0.0, (time.time() if now is None else now) - deadline)
+
+
+def is_expired(fields: Any, now: Optional[float] = None) -> bool:
+    """Whether this task's query has outlived its budget."""
+    return seconds_overdue(fields, now) > 0.0
diff --git a/tests/unit/test_task_deadline.py b/tests/unit/test_task_deadline.py
new file mode 100644
index 0000000..1919eae
--- /dev/null
+++ b/tests/unit/test_task_deadline.py
@@ -0,0 +1,541 @@
+"""Tests for the whole-query timeout budget.
+
+Queries used to run to completion however long ago they were submitted, even
+though the ARS and the sync endpoint stop waiting after ~5 minutes. These cover
+the deadline stamped at intake, its propagation from operation to operation, and
+the wrap-up a worker does instead of running an operation whose query has
+outlived its budget.
+"""
+
+import asyncio
+import logging
+import time
+
+import pytest
+
+from shepherd_server import base_routes
+from shepherd_utils import shared
+from shepherd_utils.config import settings
+from shepherd_utils.logger import attach_query_handler
+from shepherd_utils.task_deadline import (
+    DEADLINE_FIELD,
+    TIMEOUT_STATUS,
+    carry_deadline,
+    deadline_field,
+    get_deadline,
+    is_expired,
+    query_budget,
+    query_deadline,
+    seconds_overdue,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# --- budget / deadline arithmetic -------------------------------------------
+
+
+def test_query_budget_defaults_to_configured_value(monkeypatch):
+    monkeypatch.setattr(settings, "query_timeout_sec", 300.0)
+    assert query_budget() == 300.0
+    assert query_budget({}) == 300.0
+    assert query_budget({"parameters": None}) == 300.0
+
+
+def test_query_budget_honors_a_client_asking_to_wait_longer(monkeypatch):
+    """A caller that explicitly waits longer than the fleet budget isn't cut
+    short partway through its own timeout."""
+    monkeypatch.setattr(settings, "query_timeout_sec", 300.0)
+    assert query_budget({"parameters": {"timeout": 600}}) == 600.0
+
+
+def test_query_budget_ignores_a_shorter_or_unusable_client_timeout(monkeypatch):
+    monkeypatch.setattr(settings, "query_timeout_sec", 300.0)
+    # aragorn's lookup timeout rides on the same parameter and is shorter than
+    # the whole-query budget; it must not shrink it.
+    assert query_budget({"parameters": {"timeout": 210}}) == 300.0
+    assert query_budget({"parameters": {"timeout": "soon"}}) == 300.0
+    assert query_budget({"parameters": {"timeout": None}}) == 300.0
+
+
+def test_query_budget_disabled_returns_zero(monkeypatch):
+    monkeypatch.setattr(settings, "query_timeout_sec", 0.0)
+    # Even a client-supplied timeout can't re-enable a disabled budget.
+    assert query_budget({"parameters": {"timeout": 600}}) == 0.0
+
+
+def test_query_deadline_is_start_plus_budget(monkeypatch):
+    monkeypatch.setattr(settings, "query_timeout_sec", 300.0)
+    assert query_deadline({}, start=1000.0) == 1300.0
+
+
+def test_query_deadline_is_none_when_disabled(monkeypatch):
+    monkeypatch.setattr(settings, "query_timeout_sec", 0.0)
+    assert query_deadline({}, start=1000.0) is None
+
+
+# --- payload field round trip -----------------------------------------------
+
+
+def test_deadline_field_round_trips_through_a_payload():
+    fields = deadline_field(1234.5678)
+    # Redis stream values are strings; millisecond precision is plenty.
+    assert fields == {DEADLINE_FIELD: "1234.568"}
+    assert get_deadline(fields) == pytest.approx(1234.568)
+
+
+def test_deadline_field_is_empty_without_a_deadline():
+    assert deadline_field(None) == {}
+    assert carry_deadline({"query_id": "q1"}) == {}
+
+
+def test_carry_deadline_propagates_an_existing_deadline():
+    assert carry_deadline({DEADLINE_FIELD: "999.5"}) == {DEADLINE_FIELD: "999.500"}
+
+
+def test_get_deadline_tolerates_junk():
+    assert get_deadline({DEADLINE_FIELD: ""}) is None
+    assert get_deadline({DEADLINE_FIELD: "never"}) is None
+    assert get_deadline(None) is None
+    assert get_deadline("not-a-mapping") is None
+
+
+def test_overdue_only_counts_time_past_the_deadline():
+    now = 1000.0
+    assert seconds_overdue({DEADLINE_FIELD: "1100"}, now=now) == 0.0
+    assert seconds_overdue({DEADLINE_FIELD: "1000"}, now=now) == 0.0
+    assert seconds_overdue({DEADLINE_FIELD: "940"}, now=now) == 60.0
+
+
+def test_a_task_without_a_deadline_never_expires():
+    """Fail open: payloads from an older server keep running as before."""
+    assert is_expired({"query_id": "q1"}) is False
+    assert is_expired({DEADLINE_FIELD: "junk"}) is False
+    assert is_expired({DEADLINE_FIELD: str(time.time() + 60)}) is False
+    assert is_expired({DEADLINE_FIELD: str(time.time() - 1)}) is True
+
+
+# --- the wrap-up an expired task triggers -----------------------------------
+
+
+def _expired_task(msg_id="42-0"):
+    return (
+        msg_id,
+        {
+            "query_id": "q1",
+            "response_id": "r1",
+            "workflow": '[{"id": "aragorn.score"}]',
+            "log_level": 20,
+            "otel": "{}",
+            "metadata": "{}",
+            DEADLINE_FIELD: f"{time.time() - 30:.3f}",
+        },
+    )
+
+
+@pytest.fixture
+def wrap_up_spy(monkeypatch):
+    """Capture the broker/db calls the expiry path makes, in order."""
+    calls = {"added": [], "acked": [], "saved_logs": [], "order": []}
+
+    async def _fake_add_task(queue, payload, _logger):
+        calls["added"].append((queue, payload))
+        calls["order"].append("add_task")
+
+    async def _fake_mark_complete(stream, group, msg_id, _logger, retries=0):
+        calls["acked"].append((stream, group, msg_id))
+        calls["order"].append("ack")
+
+    async def _fake_save_logs(response_id, _logger):
+        calls["saved_logs"].append(response_id)
+        calls["order"].append("save_logs")
+
+    monkeypatch.setattr(shared, "add_task", _fake_add_task)
+    monkeypatch.setattr(shared, "mark_task_as_complete", _fake_mark_complete)
+    monkeypatch.setattr(shared, "save_logs", _fake_save_logs)
+    return calls
+
+
+@pytest.mark.asyncio
+async def test_expire_task_finishes_the_query_instead_of_running_it(wrap_up_spy):
+    """The operation is skipped, but the query still ends the ordinary way:
+    routed to finish_query (which settles Postgres, reaps callbacks and delivers
+    what was gathered) and its message cleared from the stream."""
+    task = _expired_task()
+    await shared._expire_task("aragorn.score", "consumer", task, logger, 30.0)
+
+    assert wrap_up_spy["acked"] == [("aragorn.score", "consumer", "42-0")]
+    assert len(wrap_up_spy["added"]) == 1
+    queue, payload = wrap_up_spy["added"][0]
+    assert queue == "finish_query"
+    assert payload["status"] == TIMEOUT_STATUS
+    assert payload["query_id"] == "q1"
+    assert payload["response_id"] == "r1"
+    # No further operations: the remaining workflow is dropped.
+    assert payload["workflow"] == "[]"
+    # The deadline rides along so finish_query's own bookkeeping sees it.
+    assert payload[DEADLINE_FIELD] == task[1][DEADLINE_FIELD]
+
+
+@pytest.mark.asyncio
+async def test_expire_task_saves_its_explanation_before_handing_off(wrap_up_spy):
+    """finish_query reads the query's logs into the response it delivers, and
+    can pick the task up immediately -- so the timeout note has to be persisted
+    before the hand-off, not after."""
+    task_logger = logging.getLogger("shepherd.test.expiry.q1")
+    attach_query_handler(task_logger)
+
+    await shared._expire_task(
+        "aragorn.score", "consumer", _expired_task(), task_logger, 30.0
+    )
+
+    # Flushed before the hand-off, then again to clear what the hand-off itself
+    # logged (nothing else runs for this query to drain it later).
+    assert wrap_up_spy["saved_logs"] == ["r1", "r1"]
+    assert wrap_up_spy["order"] == ["save_logs", "add_task", "ack", "save_logs"]
+
+
+@pytest.mark.asyncio
+async def test_expire_task_still_wraps_up_when_logs_cannot_be_saved(
+    monkeypatch, wrap_up_spy
+):
+    """A Redis hiccup while flushing logs must not strand the query."""
+
+    async def _boom(response_id, _logger):
+        raise RuntimeError("redis down")
+
+    monkeypatch.setattr(shared, "save_logs", _boom)
+    await shared._expire_task("aragorn.score", "consumer", _expired_task(), logger, 1.0)
+
+    assert wrap_up_spy["acked"] == [("aragorn.score", "consumer", "42-0")]
+    assert wrap_up_spy["added"][0][0] == "finish_query"
+
+
+# --- which tasks the check applies to ---------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_handled_as_expired_skips_a_task_within_its_budget(wrap_up_spy):
+    task = (
+        "1-0",
+        {"query_id": "q1", "response_id": "r1", DEADLINE_FIELD: str(time.time() + 60)},
+    )
+    assert (
+        await shared._handled_as_expired("aragorn.score", "consumer", task, logger)
+        is False
+    )
+    assert wrap_up_spy["added"] == []
+    assert wrap_up_spy["acked"] == []
+
+
+@pytest.mark.asyncio
+async def test_handled_as_expired_skips_a_task_without_a_deadline(wrap_up_spy):
+    task = ("1-0", {"query_id": "q1", "response_id": "r1"})
+    assert (
+        await shared._handled_as_expired("aragorn.score", "consumer", task, logger)
+        is False
+    )
+    assert wrap_up_spy["added"] == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stream", ["finish_query", "merge_message"])
+async def test_exempt_streams_are_never_expired(stream, wrap_up_spy):
+    """finish_query IS the wrap-up -- expiring it would leave the query's state
+    unset forever -- and merge_message folds callbacks upstream already paid for.
+    """
+    assert (
+        await shared._handled_as_expired(stream, "consumer", _expired_task(), logger)
+        is False
+    )
+    assert wrap_up_spy["added"] == []
+    assert wrap_up_spy["acked"] == []
+
+
+@pytest.mark.asyncio
+async def test_handled_as_expired_wraps_up_an_overdue_task(wrap_up_spy):
+    assert (
+        await shared._handled_as_expired(
+            "aragorn.score", "consumer", _expired_task(), logger
+        )
+        is True
+    )
+    assert wrap_up_spy["added"][0][1]["status"] == TIMEOUT_STATUS
+
+
+# --- propagation from operation to operation --------------------------------
+
+
+@pytest.mark.asyncio
+async def test_wrap_up_task_passes_the_deadline_to_the_next_operation(wrap_up_spy):
+    """The budget is measured from intake, so it must not restart each hop."""
+    task = (
+        "7-0",
+        {
+            "query_id": "q1",
+            "response_id": "r1",
+            "workflow": '[{"id": "aragorn.score"}, {"id": "sort_results_score"}]',
+            "otel": "{}",
+            "metadata": "{}",
+            DEADLINE_FIELD: "1700.000",
+        },
+    )
+    await shared.wrap_up_task("aragorn.score", "consumer", task, logger)
+
+    queue, payload = wrap_up_spy["added"][0]
+    assert queue == "sort_results_score"
+    assert payload[DEADLINE_FIELD] == "1700.000"
+
+
+@pytest.mark.asyncio
+async def test_handle_task_failure_passes_the_deadline_along(wrap_up_spy):
+    task = (
+        "8-0",
+        {
+            "query_id": "q1",
+            "response_id": "r1",
+            "otel": "{}",
+            "metadata": "{}",
+            DEADLINE_FIELD: "1700.000",
+        },
+    )
+    await shared.handle_task_failure("aragorn.score", "consumer", task, logger)
+
+    queue, payload = wrap_up_spy["added"][0]
+    assert queue == "finish_query"
+    assert payload["status"] == "ERROR"
+    assert payload[DEADLINE_FIELD] == "1700.000"
+
+
+@pytest.mark.asyncio
+async def test_terminate_task_still_defaults_to_an_error_status(wrap_up_spy):
+    """The poison-pill/unprocessable paths keep their ERROR status."""
+    await shared._terminate_task(
+        "aragorn.score", "consumer", _expired_task(), logger, "poison pill"
+    )
+    assert wrap_up_spy["added"][0][1]["status"] == "ERROR"
+
+
+# --- intake stamps the deadline ---------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_query_stamps_the_deadline_on_the_first_task(monkeypatch):
+    monkeypatch.setattr(settings, "query_timeout_sec", 300.0)
+    added = []
+
+    async def _fake_add_query(*args, **kwargs):
+        return None
+
+    async def _fake_add_task(queue, payload, _logger):
+        added.append((queue, payload))
+
+    monkeypatch.setattr(base_routes, "add_query", _fake_add_query)
+    monkeypatch.setattr(base_routes, "add_task", _fake_add_task)
+
+    before = time.time()
+    await base_routes.run_query("aragorn", {"message": {}})
+
+    _, payload = added[0]
+    assert get_deadline(payload) == pytest.approx(before + 300.0, abs=5)
+
+
+@pytest.mark.asyncio
+async def test_run_query_leaves_the_task_unstamped_when_disabled(monkeypatch):
+    """With the budget off, tasks look exactly as they did before -- and so
+    never expire."""
+    monkeypatch.setattr(settings, "query_timeout_sec", 0.0)
+    added = []
+
+    async def _fake_add_query(*args, **kwargs):
+        return None
+
+    async def _fake_add_task(queue, payload, _logger):
+        added.append((queue, payload))
+
+    monkeypatch.setattr(base_routes, "add_query", _fake_add_query)
+    monkeypatch.setattr(base_routes, "add_task", _fake_add_task)
+
+    await base_routes.run_query("aragorn", {"message": {}})
+    assert DEADLINE_FIELD not in added[0][1]
+
+
+# --- get_tasks integration --------------------------------------------------
+
+
+async def _async_noop(*args, **kwargs):
+    return None
+
+
+async def _no_reclaim(*args, **kwargs):
+    return []
+
+
+class _FakeHeartbeat:
+    def __init__(self):
+        self.marked = False
+        self.stopped = False
+
+    async def mark_clean_shutdown(self):
+        self.marked = True
+
+    async def stop(self):
+        self.stopped = True
+
+
+@pytest.fixture(autouse=True)
+def _reset_shutdown_state():
+    shared._shutdown = asyncio.Event()
+    shared._signal_handlers_installed = False
+    shared._active_heartbeat = None
+    yield
+    shared._shutdown = asyncio.Event()
+    shared._signal_handlers_installed = False
+    shared._active_heartbeat = None
+
+
+def _stub_worker_startup(monkeypatch):
+    fake_hb = _FakeHeartbeat()
+
+    class _HBFactory:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def start(self):
+            return fake_hb
+
+    monkeypatch.setattr(shared, "initialize_db", _async_noop)
+    monkeypatch.setattr(shared, "Heartbeat", _HBFactory)
+    monkeypatch.setattr(shared, "install_shutdown_handlers", lambda hb=None: None)
+    monkeypatch.setattr(settings, "broker_unhealthy_exit_sec", 0.0)
+    shared._active_heartbeat = fake_hb
+    return fake_hb
+
+
+@pytest.mark.asyncio
+async def test_get_tasks_does_not_hand_an_expired_task_to_the_worker(
+    monkeypatch, wrap_up_spy
+):
+    """The whole point: a stale task is wrapped up rather than worked on."""
+    _stub_worker_startup(monkeypatch)
+    task = _expired_task("500-0")
+
+    async def _fake_get_task(*args, **kwargs):
+        # One delivery, then ask the loop to drain and exit.
+        shared._request_shutdown()
+        return task
+
+    monkeypatch.setattr(shared, "get_task", _fake_get_task)
+    monkeypatch.setattr(shared, "reclaim_orphaned", _no_reclaim)
+
+    yielded = []
+    with pytest.raises(SystemExit) as exc:
+        async for item in shared.get_tasks("aragorn.score", "consumer", "cid", 8):
+            yielded.append(item)
+
+    assert exc.value.code == 0
+    assert yielded == []
+    assert wrap_up_spy["acked"] == [("aragorn.score", "consumer", "500-0")]
+    assert wrap_up_spy["added"][0][0] == "finish_query"
+    assert wrap_up_spy["added"][0][1]["status"] == TIMEOUT_STATUS
+
+
+@pytest.mark.asyncio
+async def test_get_tasks_still_hands_over_a_task_within_budget(
+    monkeypatch, wrap_up_spy
+):
+    _stub_worker_startup(monkeypatch)
+    task = (
+        "501-0",
+        {
+            "query_id": "q1",
+            "response_id": "r1",
+            "workflow": "[]",
+            "otel": "{}",
+            "metadata": "{}",
+            DEADLINE_FIELD: f"{time.time() + 120:.3f}",
+        },
+    )
+
+    async def _fake_get_task(*args, **kwargs):
+        shared._request_shutdown()
+        return task
+
+    monkeypatch.setattr(shared, "get_task", _fake_get_task)
+    monkeypatch.setattr(shared, "reclaim_orphaned", _no_reclaim)
+
+    yielded = []
+    with pytest.raises(SystemExit):
+        async for item in shared.get_tasks("aragorn.score", "consumer", "cid", 8):
+            yielded.append(item)
+            item[3].release()
+
+    assert [t[0][0] for t in yielded] == ["501-0"]
+    assert wrap_up_spy["added"] == []
+
+
+@pytest.mark.asyncio
+async def test_get_tasks_expires_a_reclaimed_task(monkeypatch, wrap_up_spy):
+    """A message stranded in a dead consumer's PEL is the likeliest way a task
+    outlives its query's budget."""
+    _stub_worker_startup(monkeypatch)
+    monkeypatch.setattr(settings, "max_task_deliveries", 0)  # breaker off
+    stale = _expired_task("502-0")
+
+    async def _fake_reclaim(stream, group, consumer, _logger, **kwargs):
+        shared._request_shutdown()
+        return [stale]
+
+    monkeypatch.setattr(shared, "reclaim_orphaned", _fake_reclaim)
+    monkeypatch.setattr(shared, "get_task", _async_noop)
+
+    yielded = []
+    with pytest.raises(SystemExit):
+        async for item in shared.get_tasks("aragorn.score", "consumer", "cid", 8):
+            yielded.append(item)
+
+    assert yielded == []
+    assert wrap_up_spy["acked"] == [("aragorn.score", "consumer", "502-0")]
+    assert wrap_up_spy["added"][0][1]["status"] == TIMEOUT_STATUS
+
+
+class _NonBlockingSlots(shared.TaskSlots):
+    """TaskSlots that refuses to wait for a permit.
+
+    A leaked permit makes ``get_tasks`` block forever on its next poll, which as
+    a test failure mode is a hung suite. Turning the wait into an immediate
+    error keeps the regression loud and fast.
+    """
+
+    async def acquire(self):
+        assert (
+            not self._sem.locked()
+        ), "get_tasks blocked waiting for a permit that was never released"
+        await super().acquire()
+
+
+@pytest.mark.asyncio
+async def test_expired_task_does_not_leak_a_concurrency_slot(monkeypatch, wrap_up_spy):
+    """Expiring must free the slot the poll reserved, or a worker fed stale
+    tasks would wedge with every permit held."""
+    _stub_worker_startup(monkeypatch)
+    monkeypatch.setattr(shared, "TaskSlots", _NonBlockingSlots)
+    seen = []
+
+    async def _fake_get_task(*args, **kwargs):
+        seen.append(1)
+        if len(seen) >= 3:
+            shared._request_shutdown()
+        return _expired_task(f"60{len(seen)}-0")
+
+    monkeypatch.setattr(shared, "get_task", _fake_get_task)
+    monkeypatch.setattr(shared, "reclaim_orphaned", _no_reclaim)
+
+    # A task limit of 1: every poll after the first needs the previous task's
+    # permit back.
+    with pytest.raises(SystemExit):
+        async for _ in shared.get_tasks("aragorn.score", "consumer", "cid", 1):
+            pass
+
+    assert len(seen) == 3
+    assert len(wrap_up_spy["added"]) == 3