diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py
index cc6082ec66b0..c913c3d8e428 100644
--- a/tests/tools/test_async_delegation.py
+++ b/tests/tools/test_async_delegation.py
@@ -229,6 +229,183 @@ def test_interrupt_all_signals_running_children():
     assert evt["status"] == "interrupted"
 
 
+def test_async_delegation_timeout_finalizes_stuck_runner():
+    gate = threading.Event()
+    interrupted = {"count": 0}
+
+    def stuck_runner():
+        gate.wait(timeout=5)
+        return {"status": "completed", "summary": "too late"}
+
+    def interrupt_fn():
+        interrupted["count"] += 1
+
+    res = ad.dispatch_async_delegation(
+        goal="stuck child", context=None, toolsets=None, role="leaf",
+        model="m", session_key="", runner=stuck_runner,
+        interrupt_fn=interrupt_fn, max_async_children=1, timeout_seconds=0.1,
+    )
+    assert res["status"] == "dispatched"
+
+    evt = _drain_one(timeout=2.0)
+    try:
+        assert evt is not None
+        assert evt["type"] == "async_delegation"
+        assert evt["status"] == "timeout"
+        assert evt["delegation_id"] == res["delegation_id"]
+        assert evt["api_calls"] == 0
+        assert "timed out after 0.1s" in evt["error"]
+        assert interrupted["count"] == 1
+        assert ad.active_count() == 0
+    finally:
+        gate.set()
+
+    # If the ignored runner eventually returns, it must not enqueue a second
+    # completion for a delegation the watchdog already finalized.
+    assert _drain_one(timeout=0.5) is None
+
+
+def test_async_delegation_batch_timeout_finalizes_stuck_runner():
+    gate = threading.Event()
+    interrupted = {"count": 0}
+
+    def stuck_batch():
+        gate.wait(timeout=5)
+        return {"results": [{"status": "completed", "summary": "too late"}]}
+
+    def interrupt_fn():
+        interrupted["count"] += 1
+
+    res = ad.dispatch_async_delegation_batch(
+        goals=["a", "b"], context="ctx", toolsets=None, role="leaf",
+        model="m", session_key="", runner=stuck_batch,
+        interrupt_fn=interrupt_fn, max_async_children=1, timeout_seconds=0.1,
+    )
+    assert res["status"] == "dispatched"
+
+    evt = _drain_one(timeout=2.0)
+    try:
+        assert evt is not None
+        assert evt["type"] == "async_delegation"
+        assert evt["status"] == "timeout"
+        assert evt["is_batch"] is True
+        assert evt["goals"] == ["a", "b"]
+        assert evt["results"] == []
+        assert "timed out after 0.1s" in evt["error"]
+        assert interrupted["count"] == 1
+        assert ad.active_count() == 0
+    finally:
+        gate.set()
+
+    assert _drain_one(timeout=0.5) is None
+
+
+def test_timeout_stays_finalizing_until_durable_persistence(tmp_path, monkeypatch):
+    gate = threading.Event()
+    persist_entered = threading.Event()
+    allow_persist = threading.Event()
+    real_persist = ad._persist_completion
+
+    def blocking_persist(event, result):
+        persist_entered.set()
+        allow_persist.wait(timeout=5)
+        real_persist(event, result)
+
+    def stuck_runner():
+        gate.wait(timeout=5)
+        return {"status": "completed", "summary": "too late"}
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(ad, "_persist_completion", blocking_persist)
+    dispatched = ad.dispatch_async_delegation(
+        goal="durable timeout", context=None, toolsets=None, role="leaf",
+        model="m", session_key="owner", runner=stuck_runner,
+        max_async_children=1, timeout_seconds=0.05,
+    )
+
+    try:
+        assert persist_entered.wait(timeout=2)
+        assert ad.active_count() == 1
+        record = next(
+            item for item in ad.list_async_delegations()
+            if item["delegation_id"] == dispatched["delegation_id"]
+        )
+        assert record["status"] == "finalizing"
+        assert process_registry.completion_queue.empty()
+
+        allow_persist.set()
+        evt = _drain_for(dispatched["delegation_id"])
+        assert evt is not None
+        assert evt["status"] == "timeout"
+        assert ad.active_count() == 0
+        durable = ad.get_durable_delegation(dispatched["delegation_id"])
+        assert durable["state"] == "timeout"
+        assert durable["delivery_state"] == "pending"
+    finally:
+        allow_persist.set()
+        gate.set()
+
+
+def test_timeout_completion_restores_once_after_process_restart(tmp_path):
+    repo = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+    env = {**os.environ, "HERMES_HOME": str(tmp_path), "PYTHONPATH": repo}
+    producer = r'''
+import json
+import threading
+import time
+from tools import async_delegation as ad
+gate = threading.Event()
+r = ad.dispatch_async_delegation(
+    goal="restart timeout", context=None, toolsets=None, role="leaf", model="m",
+    session_key="owner-session", parent_session_id="durable-parent",
+    runner=lambda: gate.wait(timeout=60), timeout_seconds=.05,
+)
+deadline = time.time() + 5
+while ad.active_count() and time.time() < deadline:
+    time.sleep(.01)
+row = ad.get_durable_delegation(r["delegation_id"])
+print(json.dumps({"delegation_id": r["delegation_id"], "row": row}, sort_keys=True))
+'''
+    first = subprocess.run(
+        [sys.executable, "-c", producer], cwd=repo, env=env,
+        text=True, capture_output=True, timeout=15, check=True,
+    )
+    produced = json.loads(first.stdout.strip().splitlines()[-1])
+    delegation_id = produced["delegation_id"]
+    assert produced["row"]["state"] == "timeout"
+    assert produced["row"]["delivery_state"] == "pending"
+
+    consumer = r'''
+import json
+from tools.process_registry import process_registry
+evt = process_registry.completion_queue.get_nowait()
+print(json.dumps({"event": evt, "remaining": process_registry.completion_queue.qsize()}, sort_keys=True))
+'''
+    second = subprocess.run(
+        [sys.executable, "-c", consumer], cwd=repo, env=env,
+        text=True, capture_output=True, timeout=15, check=True,
+    )
+    restored = json.loads(second.stdout.strip().splitlines()[-1])
+    assert restored["remaining"] == 0
+    assert restored["event"]["delegation_id"] == delegation_id
+    assert restored["event"]["status"] == "timeout"
+    assert restored["event"]["restored"] is True
+
+    acker = f'''
+from tools import async_delegation as ad
+assert ad.mark_completion_delivered({delegation_id!r})
+'''
+    subprocess.run(
+        [sys.executable, "-c", acker], cwd=repo, env=env,
+        text=True, capture_output=True, timeout=15, check=True,
+    )
+    probe = subprocess.run(
+        [sys.executable, "-c", "from tools.process_registry import process_registry; print(process_registry.completion_queue.qsize())"],
+        cwd=repo, env=env, text=True, capture_output=True, timeout=15, check=True,
+    )
+    assert probe.stdout.strip().splitlines()[-1] == "0"
+
+
 def test_completed_records_pruned_to_cap():
     # Run more than the retention cap quickly; ensure list doesn't grow forever.
     for i in range(ad._MAX_RETAINED_COMPLETED + 10):
@@ -767,6 +944,45 @@ def test_delegate_task_background_batch_runs_as_one_unit(monkeypatch):
     assert _drain_one() is None
 
 
+def test_delegate_task_background_passes_child_timeout_to_async_registry(monkeypatch):
+    import json
+    from unittest.mock import MagicMock
+    import tools.delegate_tool as dt
+
+    parent = MagicMock()
+    parent._delegate_depth = 0
+    parent.session_id = "sess"
+    parent._interrupt_requested = False
+    parent._active_children = []
+    parent._active_children_lock = None
+
+    fake_child = MagicMock()
+    fake_child._delegate_role = "leaf"
+    fake_child._subagent_id = "s1"
+
+    creds = {
+        "model": "m", "provider": None, "base_url": None, "api_key": None,
+        "api_mode": None, "command": None, "args": None,
+    }
+    captured = {}
+
+    def fake_dispatch(**kwargs):
+        captured.update(kwargs)
+        return {"status": "dispatched", "delegation_id": "deleg_timeout"}
+
+    monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
+    monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds)
+    monkeypatch.setattr(dt, "_get_child_timeout", lambda: 600.0)
+    monkeypatch.setattr(ad, "dispatch_async_delegation_batch", fake_dispatch)
+
+    out = dt.delegate_task(goal="background timeout", background=True, parent_agent=parent)
+
+    parsed = json.loads(out)
+    assert parsed["status"] == "dispatched"
+    assert parsed["delegation_id"] == "deleg_timeout"
+    assert captured["timeout_seconds"] == 600.0
+
+
 def test_model_dispatch_forces_background():
     """The MODEL-facing dispatch path forces background=True for any top-level
     delegation (single task OR batch), and keeps it off for an orchestrator
diff --git a/tools/async_delegation.py b/tools/async_delegation.py
index 7e88899e5851..c2ed82c99baa 100644
--- a/tools/async_delegation.py
+++ b/tools/async_delegation.py
@@ -483,6 +483,10 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]:
     }
 
 
+def _format_timeout_seconds(timeout_seconds: float) -> str:
+    return f"{timeout_seconds:g}"
+
+
 def _get_executor(max_workers: int) -> ThreadPoolExecutor:
     """Lazily create (or grow) the shared daemon executor.
 
@@ -572,6 +576,7 @@ def dispatch_async_delegation(
     origin_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
+    timeout_seconds: Optional[float] = None,
 ) -> Dict[str, Any]:
     """Spawn ``runner`` on the daemon executor and return a handle immediately.
 
@@ -624,6 +629,7 @@ def dispatch_async_delegation(
         "dispatched_at": dispatched_at,
         "completed_at": None,
         "interrupt_fn": interrupt_fn,
+        "timeout_seconds": timeout_seconds,
     }
     # Capacity check and record insert under ONE lock hold — checking
     # active_count() separately would let two concurrent dispatches (e.g.
@@ -679,6 +685,7 @@ def dispatch_async_delegation(
             "status": "rejected",
             "error": f"Failed to schedule async delegation: {exc}",
         }
+    _start_timeout_watchdog(delegation_id, timeout_seconds, is_batch=False)
 
     logger.info(
         "Dispatched async delegation %s (session_key=%s): %s",
@@ -689,19 +696,39 @@ def dispatch_async_delegation(
 
 def _finalize(delegation_id: str, result: Dict[str, Any], status: str) -> None:
     """Mark a record complete and push the completion event onto the queue."""
+    claimed = _begin_finalization(delegation_id)
+    if claimed is None:
+        return
+    event_record, _interrupt_fn = claimed
+
+    _push_completion_event(event_record, result, status)
+    _finish_finalization(delegation_id, status)
+
+
+def _begin_finalization(
+    delegation_id: str,
+) -> Optional[tuple[Dict[str, Any], Optional[Callable[[], None]]]]:
+    """Atomically claim terminal delivery while keeping the record active."""
     with _records_lock:
         record = _records.get(delegation_id)
-        if record is None:
+        if record is None or record.get("status") != "running":
             return
         # Stay active until durable persistence and queue publication finish;
         # otherwise process shutdown can kill this daemon worker in the narrow
         # gap after status flips but before SQLite is committed.
         record["status"] = "finalizing"
         record["completed_at"] = time.time()
+        interrupt_fn = record.get("interrupt_fn")
         record["interrupt_fn"] = None  # drop the closure; child is done
+        timer = record.pop("_timeout_timer", None)
         event_record = dict(record)
 
-    _push_completion_event(event_record, result, status)
+    if timer is not None:
+        timer.cancel()
+    return event_record, interrupt_fn
+
+
+def _finish_finalization(delegation_id: str, status: str) -> None:
     with _records_lock:
         record = _records.get(delegation_id)
         if record is not None:
@@ -783,6 +810,7 @@ def dispatch_async_delegation_batch(
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
     delegation_id: Optional[str] = None,
+    timeout_seconds: Optional[float] = None,
 ) -> Dict[str, Any]:
     """Dispatch a WHOLE fan-out batch as ONE background unit.
 
@@ -828,6 +856,7 @@ def dispatch_async_delegation_batch(
         "completed_at": None,
         "interrupt_fn": interrupt_fn,
         "is_batch": True,
+        "timeout_seconds": timeout_seconds,
     }
     with _records_lock:
         running = sum(
@@ -884,6 +913,7 @@ def dispatch_async_delegation_batch(
             "status": "rejected",
             "error": f"Failed to schedule async delegation batch: {exc}",
         }
+    _start_timeout_watchdog(delegation_id, timeout_seconds, is_batch=True)
 
     logger.info(
         "Dispatched async delegation batch %s (%d task(s), session_key=%s)",
@@ -896,22 +926,26 @@ def _finalize_batch(
     delegation_id: str, combined: Dict[str, Any], status: str
 ) -> None:
     """Mark a batch record complete and push ONE combined completion event."""
-    with _records_lock:
-        record = _records.get(delegation_id)
-        if record is None:
-            return
-        record["status"] = "finalizing"
-        record["completed_at"] = time.time()
-        record["interrupt_fn"] = None
-        event_record = dict(record)
+    claimed = _begin_finalization(delegation_id)
+    if claimed is None:
+        return
+    event_record, _interrupt_fn = claimed
 
+    _push_batch_completion_event(event_record, combined, status)
+    _finish_finalization(delegation_id, status)
+
+
+def _push_batch_completion_event(
+    event_record: Dict[str, Any], combined: Dict[str, Any], status: str
+) -> None:
+    """Push a combined async-delegation batch completion event."""
     try:
         from tools.process_registry import process_registry
     except Exception as exc:  # pragma: no cover
         logger.error(
             "Async delegation batch %s finished but process_registry import "
             "failed; result lost: %s",
-            delegation_id, exc,
+            event_record.get("delegation_id"), exc,
         )
         return
 
@@ -919,7 +953,7 @@ def _finalize_batch(
     completed_at = event_record.get("completed_at") or time.time()
     evt = {
         "type": "async_delegation",
-        "delegation_id": delegation_id,
+        "delegation_id": event_record.get("delegation_id"),
         "session_key": event_record.get("session_key", ""),
         "origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
         "origin_session_id": event_record.get("origin_session_id", ""),
@@ -951,14 +985,102 @@ def _finalize_batch(
         logger.error(
             "Async delegation batch %s: failed to enqueue completion event; "
             "result lost: %s",
-            delegation_id, exc,
+            event_record.get("delegation_id"), exc,
         )
-    finally:
-        with _records_lock:
-            record = _records.get(delegation_id)
-            if record is not None:
-                record["status"] = status
-            _prune_completed_locked()
+
+
+def _start_timeout_watchdog(
+    delegation_id: str,
+    timeout_seconds: Optional[float],
+    *,
+    is_batch: bool,
+) -> None:
+    """Finalize a detached delegation if its worker never returns.
+
+    ``delegate_task(background=true)`` returns a handle immediately; the only
+    user-visible result is the completion event this module enqueues later. If
+    the worker thread wedges before returning, the normal ``finally`` block is
+    never reached and users only see a permanent "dispatched" state. A
+    configured timeout must therefore be enforced by the async registry itself,
+    not only inside ``delegate_tool._run_single_child``.
+    """
+    if timeout_seconds is None or timeout_seconds <= 0:
+        return
+
+    timer = threading.Timer(
+        timeout_seconds,
+        _expire_delegation,
+        args=(delegation_id, float(timeout_seconds)),
+        kwargs={"is_batch": is_batch},
+    )
+    timer.daemon = True
+    with _records_lock:
+        record = _records.get(delegation_id)
+        if record is None or record.get("status") != "running":
+            return
+        record["timeout_seconds"] = float(timeout_seconds)
+        record["_timeout_timer"] = timer
+    timer.start()
+
+
+def _expire_delegation(
+    delegation_id: str,
+    timeout_seconds: float,
+    *,
+    is_batch: bool,
+) -> None:
+    """Timeout a still-running async delegation and emit one completion."""
+    claimed = _begin_finalization(delegation_id)
+    if claimed is None:
+        return
+    event_record, interrupt_fn = claimed
+
+    completed_at = event_record.get("completed_at") or time.time()
+    duration = round(
+        completed_at - (event_record.get("dispatched_at") or completed_at),
+        2,
+    )
+    timeout_text = _format_timeout_seconds(timeout_seconds)
+    error = (
+        f"Async delegation {delegation_id} timed out after {timeout_text}s "
+        "without producing a completion event. Hermes requested interruption; "
+        "the detached worker may be stuck before or inside the first model API "
+        "call."
+    )
+    if is_batch:
+        _push_batch_completion_event(
+            event_record,
+            {
+                "results": [],
+                "error": error,
+                "total_duration_seconds": duration,
+            },
+            "timeout",
+        )
+    else:
+        _push_completion_event(
+            event_record,
+            {
+                "status": "timeout",
+                "summary": None,
+                "error": error,
+                "api_calls": 0,
+                "duration_seconds": duration,
+                "exit_reason": "timeout",
+            },
+            "timeout",
+        )
+    _finish_finalization(delegation_id, "timeout")
+
+    if callable(interrupt_fn):
+        try:
+            interrupt_fn()
+        except Exception as exc:
+            logger.debug(
+                "Async delegation %s timeout interrupt failed: %s",
+                delegation_id,
+                exc,
+            )
 
 
 def list_async_delegations() -> List[Dict[str, Any]]:
@@ -968,7 +1090,11 @@ def list_async_delegations() -> List[Dict[str, Any]]:
     """
     with _records_lock:
         return [
-            {k: v for k, v in r.items() if k != "interrupt_fn"}
+            {
+                k: v
+                for k, v in r.items()
+                if k not in {"interrupt_fn", "_timeout_timer"}
+            }
             for r in _records.values()
         ]
 
@@ -1066,4 +1192,11 @@ def _reset_for_tests() -> None:
         _executor = None
         _executor_max_workers = 0
     with _records_lock:
+        timers = [
+            timer
+            for record in _records.values()
+            if (timer := record.get("_timeout_timer")) is not None
+        ]
         _records.clear()
+    for timer in timers:
+        timer.cancel()
diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py
index 761785d7aae2..2b5b31dc914d 100644
--- a/tools/delegate_tool.py
+++ b/tools/delegate_tool.py
@@ -3057,6 +3057,7 @@ def delegate_task(
             # Reuse the live-transcript directory's id (when created) so the
             # returned delegation_id matches cache/delegation/live/<id>/.
             delegation_id=live_deleg_id,
+            timeout_seconds=_get_child_timeout(),
         )
 
         if dispatch.get("status") == "dispatched":