diff --git a/statek/__init__.py b/statek/__init__.py
index e824b47..e0f04bf 100644
--- a/statek/__init__.py
+++ b/statek/__init__.py
@@ -52,7 +52,7 @@
 from .executors.post_processor import FinalCheck, PostProcessor
 from .system import (tool, subtask, docstr, get_any, get_all, error_handler,
                      is_valid_error_handler, docs_style, find_sub_task_handler,
-                     find_tools)
+                     get_pending_notifications, find_tools)
 from .shared_context import (
     ContextCategory,
     ContextCategoryDict,
@@ -70,6 +70,7 @@
     complete_sub_task,
     create_new_job,
     create_sub_task,
+    send_notification_to,
 )
 from .runner import start_statek, start_statek_async
 from . import task
@@ -187,6 +188,8 @@ def init(settings: Optional[StatekSettings] = None, restricted: bool = True) ->
     "create_new_job",
     "create_sub_task",
     "find_sub_task_handler",
+    "get_pending_notifications",
+    "send_notification_to",
     "ContextCategory",
     "ContextCategoryDict",
     "ContextVar",
diff --git a/statek/executors/chat_log_item.py b/statek/executors/chat_log_item.py
index 7f4fc36..b313b16 100644
--- a/statek/executors/chat_log_item.py
+++ b/statek/executors/chat_log_item.py
@@ -124,6 +124,18 @@ class SubTaskLogItem(ChatLogItem):
     handler: "SubTaskHandler"
 
 
+@db0.memo(no_default_tags=True)
+@dataclass(kw_only=True)
+class NotificationLogItem(ChatLogItem):
+    """An arbitrary plain-text message delivered between jobs."""
+
+    message: str
+
+    def __llm_repr__(self) -> str:
+        """Show the message without recursively formatting its saved tool result."""
+        return f"NotificationLogItem(message={self.message!r})"
+
+
 @db0.memo(no_default_tags=True)
 @dataclass
 class UserLogItem:
diff --git a/statek/executors/job.py b/statek/executors/job.py
index 2d4b2c4..fc1da78 100644
--- a/statek/executors/job.py
+++ b/statek/executors/job.py
@@ -42,6 +42,7 @@
 from statek.executors.chat_log_item import (
     ChatLogItem,
     LLM_LogItem,
+    NotificationLogItem,
     PostProcessedItem,
     ReminderLogItem,
     SubTaskLogItem,
@@ -64,6 +65,7 @@
     build_warmup_code,
     extract_dialog,
     get_current_job,
+    format_default_llm_repr,
     parse_tool_log,
     parse_warmup_block,
     perm_ctx_get,
@@ -693,8 +695,8 @@ def add_locals(self, **kwargs) -> None:
         """Inject additional values into this job's Python local state."""
         self.py_env.update_locals(**kwargs)
 
-    def _pending_notifications(self) -> List[SubTaskLogItem]:
-        """Return pending subtask notifications, initializing old jobs lazily."""
+    def _pending_notifications(self) -> List[Union[SubTaskLogItem, NotificationLogItem]]:
+        """Return pending chat-log notifications, initializing old jobs lazily."""
         if not hasattr(self, "_Job__pending_chat_log") or self.__pending_chat_log is None:
             self.__pending_chat_log = []
         return self.__pending_chat_log
@@ -702,7 +704,7 @@ def _pending_notifications(self) -> List[SubTaskLogItem]:
     def find_sub_task_handler(self, id: Optional[Any] = None) -> Optional["SubTaskHandler"]:  # pylint: disable=redefined-builtin
         """Find a pending or logged subtask handler by id, or the most recent one."""
         for item in reversed(self._pending_notifications()):
-            if id is None or item.handler.id == id:
+            if isinstance(item, SubTaskLogItem) and (id is None or item.handler.id == id):
                 return item.handler
         for item in reversed(self.chat_log):
             if isinstance(item, SubTaskLogItem) and (id is None or item.handler.id == id):
@@ -730,10 +732,15 @@ def _chat_log_finalized_for_notifications(self) -> bool:
         return True
 
     def _process_pending_notifications(self) -> bool:
-        """Move pending subtask notifications to chat_log at safe boundaries."""
+        """Move finalized pending notifications into the ordered chat history."""
         pending = self._pending_notifications()
         if not pending or not self._chat_log_finalized_for_notifications():
             return False
+        console_pos = len(self.py_env.console) if self.py_env.console else 0
+        for item in pending:
+            item.console_pos = console_pos
+            if isinstance(item, NotificationLogItem) and item.tool_log is None:
+                item.tool_log = format_default_llm_repr([item])
         self.chat_log.extend(pending)
         self.__pending_chat_log = []
         return True
@@ -757,6 +764,24 @@ def push_notification(self, handler: "SubTaskHandler") -> None:
             return
         self._pending_notifications().append(item)
 
+    def send_notification(self, message: str) -> None:
+        """Record a plain-text notification for this job's next safe boundary.
+
+        A completed job resumes with the notification in its history. Other
+        jobs queue it without disturbing their pending execution or future.
+        """
+        item = NotificationLogItem(
+            console_pos=len(self.py_env.console) if self.py_env.console else 0,
+            message=message,
+        )
+        item.tool_log = format_default_llm_repr([item])
+        if self.status == JobStatus.DONE:  # pylint: disable=no-member
+            self.chat_log.append(item)
+            self.set_status(JobStatus.STARTED)  # pylint: disable=no-member
+            self.py_env.exit_status = None
+            return
+        self._pending_notifications().append(item)
+
     def _current_model_pricing(self):
         """Return pricing for the concrete model used by the current LLM request."""
         metadata = self.job_def.metadata or {}
@@ -1062,6 +1087,16 @@ def _subtask_tool_call(idx: int, handler: "SubTaskHandler") -> CallSpecWrapper:
             kwargs={"code": Job._subtask_lookup_code(handler)},
         )
 
+    @staticmethod
+    def _notification_tool_call(idx: int) -> CallSpecWrapper:
+        """Return the synthetic python_cli call for one delivered text message."""
+        return CallSpecWrapper(
+            id=f"STATEK-NOTIFICATION-{idx:03d}",
+            func_name="python_cli",
+            args=None,
+            kwargs={"code": "print(get_pending_notifications())"},
+        )
+
     def get_next_prompt(self) -> str:
         """
         Generate the next prompt to be included in the LLM chat.
@@ -1237,6 +1272,7 @@ def _build_chat_history(
                     PostProcessedItem,
                     ReminderLogItem,
                     SubTaskLogItem,
+                    NotificationLogItem,
                 )):
                     next_pos = items[j].console_pos
                     break
@@ -1319,6 +1355,25 @@ def _yield_pushes(lo: int, hi: int) -> Iterable[ChatHistoryItem]:
                     )
                 continue
 
+            if isinstance(item, NotificationLogItem):
+                call = self._notification_tool_call(idx)
+                yield ChatHistoryItem(
+                    role=ChatRole.ASSISTANT,
+                    content=None,
+                    content_src=ContentSource.SYSTEM,
+                    tool_calls=[call],
+                )
+                result = item.tool_log if item.tool_log is not None else format_default_llm_repr([item])
+                if isinstance(result, ToolError):
+                    result = result.err_message
+                yield ChatHistoryItem(
+                    role=ChatRole.TOOL,
+                    content=result,
+                    content_src=ContentSource.CONSOLE,
+                    tool_calls=[call],
+                )
+                continue
+
             # Determine the source content of the assistant turn.
             if isinstance(item, WarmupLogItem):
                 block_num = item.warmup_block_num
diff --git a/statek/executors/utils.py b/statek/executors/utils.py
index 865ed7f..309f53f 100644
--- a/statek/executors/utils.py
+++ b/statek/executors/utils.py
@@ -938,7 +938,7 @@ def _log_pending_console(job: Job):
 
 
 def _process_pending_notification_follow_up(job: Job, harness) -> bool:
-    """Process pending subtask notifications before a job is finalized."""
+    """Process pending notifications before a job is finalized."""
     if not job._process_pending_notifications():  # pylint: disable=protected-access
         return False
     job.py_env.exit_status = None
@@ -1211,6 +1211,7 @@ def _cli_error_append(cli_idx, text):
 
     # Step 12: Get next request parameters — log pending console batch first
     _log_pending_console(job)
+    job._process_pending_notifications()  # pylint: disable=protected-access
     request_difficulty = job.get_current_difficulty()
     request = job.get_next_request(difficulty=request_difficulty)
     request["metadata"] = dict(request["metadata"] or {})
diff --git a/statek/system.py b/statek/system.py
index c777a80..cccf8ba 100644
--- a/statek/system.py
+++ b/statek/system.py
@@ -26,6 +26,7 @@
 from .dbzero_restricted import as_unrestricted
 from .future import get_any_future, get_all_future, FutureResult
 from .docstring import parse_tool_docstring, format_docstring
+from .executors.chat_log_item import ChatLogItem
 from .utils import find_locals, get_current_agent_name, get_current_job, _statek_ctx_scope
 
 
@@ -451,6 +452,22 @@ def wrapper(*args, **kwargs):
     return _decorate(f)
 
 
+@tool(system=True)
+def get_pending_notifications(**kwargs) -> Iterable[ChatLogItem]:  # pylint: disable=unused-argument
+    """Return an ordered snapshot of the current job's unprocessed notifications.
+
+    The snapshot contains pending chat log items, including subtask results,
+    without moving them into the job's history or consuming the queue.
+
+    Raises:
+        RuntimeError: When called without a current job.
+    """
+    job = get_current_job()
+    if job is None:
+        raise RuntimeError("get_pending_notifications requires a current job")
+    return list(job._pending_notifications())  # pylint: disable=protected-access
+
+
 @tool(system=True)
 def find_sub_task_handler(id: Optional[Any] = None, **kwargs) -> Any:  # pylint: disable=redefined-builtin,unused-argument
     """Return a subtask handler from the current job notification history.
diff --git a/statek/task.py b/statek/task.py
index ec2c184..3590f15 100644
--- a/statek/task.py
+++ b/statek/task.py
@@ -410,6 +410,30 @@ def complete(self, result: Optional[Any] = None, error: Optional[str] = None) ->
             parent_job.push_notification(self)
 
 
+@tool(system=True)
+def send_notification_to(target: SubTaskHandler | Job, message: str, **kwargs) -> None:  # pylint: disable=unused-argument
+    """Send plain text to an existing job or subtask handler's job.
+
+    The message is recorded now but delivered at the recipient's next safe
+    execution boundary, not by interrupting its current code or awaited future.
+    A completed job is restarted to handle the new notification.
+
+    Args:
+        target: An existing Job or SubTaskHandler for the recipient job.
+        message: Plain-text message to deliver without executing it as code.
+
+    Raises:
+        TypeError: If target is not a Job or SubTaskHandler.
+    """
+    if isinstance(target, SubTaskHandler):
+        job = target.job
+    elif isinstance(target, Job):
+        job = target
+    else:
+        raise TypeError("target must be a Job or SubTaskHandler")
+    job.send_notification(message)
+
+
 def complete_sub_task(result: Optional[Any] = None, error: Optional[str] = None) -> None:
     """Complete the subtask handler registered in the current child job locals."""
     job = get_current_job()
diff --git a/statek/utils.py b/statek/utils.py
index 177a7e6..61ee003 100644
--- a/statek/utils.py
+++ b/statek/utils.py
@@ -24,6 +24,7 @@
 from dataclasses import dataclass, is_dataclass, fields as dataclass_fields
 from datetime import datetime
 from decimal import Decimal
+from types import UnionType
 from typing import (
     Any,
     Callable,
@@ -639,8 +640,8 @@ def _type_to_json_schema(type_hint) -> str:
     origin = get_origin(type_hint)
     args = get_args(type_hint)
 
-    # Optional[X] = Union[X, None] and other Union types
-    if origin is Union:
+    # Optional[X] = Union[X, None]; support both typing.Union and X | Y.
+    if origin in (Union, UnionType):
         non_none = [a for a in args if a is not _NONE_TYPE]
         return _type_to_json_schema(non_none[0]) if non_none else "null"
 
diff --git a/tests/test_job.py b/tests/test_job.py
index 5ab46dc..59a4a3e 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -9,7 +9,7 @@
 from unittest.mock import patch, MagicMock
 import dbzero as db0
 import pytest
-from tests.conftest import create_chat_log_item, set_warmup_positions
+from tests.conftest import DB0_DIR, create_chat_log_item, set_warmup_positions
 from statek.executors.job import (
     DialogItem,
     Job,
@@ -33,6 +33,7 @@
 from statek.agents.dialog_agent import RecurringReminder
 from statek.executors.chat_log_item import (
     LLM_LogItem,
+    NotificationLogItem,
     PostProcessedItem,
     ReminderLogItem,
     SubTaskLogItem,
@@ -41,10 +42,11 @@
 )
 from statek.executors.post_processor import PostProcessor
 from statek.executors.utils import _log_pending_console
+from statek.future import FutureResult
 from statek.settings import ChatStyle, LLM_API_Settings
 from statek.locale import StatekLocale, StatekLangCode, StatekCountryCode
 from statek.prompt_config import make_system_prompt, parse_system_prompt
-from statek.utils import CodeBlock, CallSpec, _statek_ctx_scope
+from statek.utils import CodeBlock, CallSpec, _statek_ctx_scope, format_default_llm_repr
 from statek.system import find_sub_task_handler
 from statek.task import SubTaskHandler, TaskError
 
@@ -3056,6 +3058,216 @@ def test_push_notification_active_buffers_until_last_item_finalized(self, job_fa
         assert job._process_pending_notifications() is False  # pylint: disable=protected-access
 
 
+# pylint: disable=protected-access
+class TestTextNotificationHistory:
+    """A plain-text notification is a durable, synthetic tool-result event."""
+
+    def test_notification_is_persisted_and_formatted_without_recursive_tool_log(self, job_factory):
+        """The message is required and the saved result does not expand itself."""
+        job = job_factory()
+        item = NotificationLogItem(console_pos=0, message="Żółw\n<script>")
+        item.tool_log = format_default_llm_repr([item])
+        job._pending_notifications().append(item)
+
+        assert job._pending_notifications()[0].message == "Żółw\n<script>"
+        assert "Żółw" in item.tool_log
+        assert "tool_log" not in item.tool_log
+        assert job._process_pending_notifications() is True
+        assert job.chat_log[-1] is item
+
+    def test_mixed_queue_keeps_order_and_lookup_ignores_text_items(self, job_factory):
+        """Text messages never masquerade as completed subtask handlers."""
+        job = job_factory()
+        handler = _completed_subtask_handler(job_factory(), subtask_id="child", result="done")
+        subtask = SubTaskLogItem(console_pos=0, handler=handler, tool_log="done")
+        notification = NotificationLogItem(console_pos=0, message="hello", tool_log="hello")
+        job._pending_notifications().extend([subtask, notification])
+
+        assert job.find_sub_task_handler(id="child") is handler
+        assert job.find_sub_task_handler() is handler
+        assert job._process_pending_notifications() is True
+        assert job.chat_log == [subtask, notification]
+        assert job._process_pending_notifications() is False
+        history = list(job.get_chat_history())
+        codes = [
+            entry.tool_calls[0].kwargs["code"]
+            for entry in history if entry.role == ChatRole.ASSISTANT
+        ]
+        assert codes == [
+            "print(find_sub_task_handler(id='child'))",
+            "print(get_pending_notifications())",
+        ]
+
+    def test_multiple_messages_get_distinct_saved_results_and_call_ids(self, job_factory):
+        """Each queued message retains order and has an independent synthetic pair."""
+        job = job_factory()
+        first = NotificationLogItem(console_pos=0, message="same")
+        second = NotificationLogItem(console_pos=0, message="same")
+        job._pending_notifications().extend([first, second])
+
+        assert job._process_pending_notifications() is True
+        assert first.tool_log == second.tool_log
+        assert "same" in first.tool_log
+        assert "tool_log" not in first.tool_log
+        history = list(job.get_chat_history())
+        assert [entry.role for entry in history] == [
+            ChatRole.ASSISTANT, ChatRole.TOOL, ChatRole.ASSISTANT, ChatRole.TOOL,
+        ]
+        ids = [history[0].tool_calls[0].id, history[2].tool_calls[0].id]
+        assert ids == ["STATEK-NOTIFICATION-000", "STATEK-NOTIFICATION-001"]
+        assert history[1].tool_calls[0].id == ids[0]
+        assert history[3].tool_calls[0].id == ids[1]
+
+    def test_synthetic_history_has_stable_pair_and_no_extra_turn(self, job_factory):
+        """A text notification yields one paired synthetic call, not an LLM action."""
+        job = job_factory()
+        job.chat_log.append(NotificationLogItem(
+            console_pos=0, message="hello", tool_log="message: hello",
+        ))
+
+        history = list(job.get_chat_history())
+        formatted = [format_chat_history_item(entry, ChatStyle.DIRECT) for entry in history]
+
+        assert [entry.role for entry in history] == [ChatRole.ASSISTANT, ChatRole.TOOL]
+        assert history[0].content_src == ContentSource.SYSTEM
+        assert history[0].tool_calls[0].kwargs == {"code": "print(get_pending_notifications())"}
+        assert formatted[0]["tool_calls"][0]["id"] == "STATEK-NOTIFICATION-000"
+        assert formatted[1]["tool_call_id"] == "STATEK-NOTIFICATION-000"
+        assert formatted[1]["content"] == "message: hello"
+        assert job.count_llm_actions() == 0
+        assert job.num_turns == 0
+
+    def test_delivery_preserves_preceding_console_and_historical_requests(self, job_factory):
+        """Send-time console positions cannot truncate output printed before drain."""
+        job = job_factory()
+        job.chat_log.append(LLM_LogItem(console_pos=0, llm_resp="print('after send')"))
+        notification = NotificationLogItem(console_pos=0, message="hello", tool_log="hello")
+        job._pending_notifications().append(notification)
+        job.py_env.console = ["after send"]
+
+        assert job._process_pending_notifications() is True
+        assert notification.console_pos == 1
+        assert [item.content for item in job.get_chat_history()] == [
+            "print('after send')", "after send", None, "hello",
+        ]
+
+        job.chat_log.append(LLM_LogItem(console_pos=1, llm_resp="print('next')"))
+        history = list(job.get_request_data(0)["chat_history"])
+        assert not history
+        next_history = list(job.get_request_data(1)["chat_history"])
+        assert [entry.content for entry in next_history] == [
+            "print('after send')", "after send", None, "hello",
+        ]
+
+# pylint: enable=protected-access
+
+
+# pylint: disable=protected-access
+class TestSendNotification:
+    """Sending text preserves recipient execution state until a safe boundary."""
+
+    @pytest.mark.parametrize("status", [
+        JobStatus.READY, JobStatus.WARMING_UP, JobStatus.STARTED, JobStatus.SUSPENDED,
+    ])
+    def test_active_statuses_queue_inert_text_without_changing_continuation(
+        self, job_factory, status,
+    ):
+        """Active and suspended recipients retain their current work and future."""
+        job = job_factory()
+        job.set_status(status)
+        job.chat_log.append(LLM_LogItem(console_pos=0, llm_resp=None))
+        job.py_env.console = ["previous output"]
+        job.py_env.exit_status = None
+        job.next_instr_num = 3
+        future = FutureResult(deps=job, state_num=0)
+        job.awaited_result = future
+
+        job.send_notification("Żółw\nexit('not code')")
+
+        assert job.status == status
+        assert job.awaited_result is future
+        assert job.next_instr_num == 3
+        assert job.py_env.exit_status is None
+        assert len(job.chat_log) == 1
+        assert list(job.py_env.console) == ["previous output"]
+        item = job._pending_notifications()[0]
+        assert isinstance(item, NotificationLogItem)
+        assert item.message == "Żółw\nexit('not code')"
+        assert item.console_pos == 1
+        assert "Żółw" in item.tool_log
+        assert "tool_log" not in item.tool_log
+        assert job._process_pending_notifications() is False
+
+    def test_done_job_restarts_with_direct_saved_notification(self, job_factory):
+        """A completed job resumes without stale exit status or an extra user turn."""
+        job = job_factory()
+        job.chat_log.append(LLM_LogItem(console_pos=0, llm_resp="pass"))
+        job.py_env.console = ["old output"]
+        job.set_status(JobStatus.DONE)
+        job.py_env.exit_status = "finished"
+
+        job.send_notification("please review")
+
+        assert job.status == JobStatus.STARTED
+        assert job.py_env.exit_status is None
+        assert not job._pending_notifications()
+        assert len(job.chat_log) == 2
+        item = job.chat_log[-1]
+        assert isinstance(item, NotificationLogItem)
+        assert item.message == "please review"
+        assert item.console_pos == 1
+        assert "please review" in item.tool_log
+        assert job.num_turns == 1
+
+    def test_repeated_sends_after_done_keep_each_message_in_order(self, job_factory):
+        """Only the first send reopens DONE; later sends wait in the shared queue."""
+        job = job_factory()
+        job.set_status(JobStatus.DONE)
+        job.py_env.exit_status = "finished"
+
+        job.send_notification("first")
+        first = job.chat_log[-1]
+        job.send_notification("second")
+        job.send_notification("third")
+
+        assert job.status == JobStatus.STARTED
+        assert job.chat_log == [first]
+        assert first.message == "first"
+        assert [item.message for item in job._pending_notifications()] == ["second", "third"]
+        assert job._process_pending_notifications() is True
+        assert [item.message for item in job.chat_log] == ["first", "second", "third"]
+        assert [item.tool_log for item in job.chat_log] == [
+            format_default_llm_repr([item]) for item in job.chat_log
+        ]
+
+    def test_saved_message_survives_dbzero_reopen(self, job_factory):
+        """Both pending messages and their stored results survive reconnecting."""
+        job = job_factory()
+        job.send_notification("persistent")
+        job_id = db0.uuid(job)
+        db0.close()
+        db0.init(DB0_DIR, read_write=True)
+        db0.open("test_prefix", "rw")
+
+        restored = db0.fetch(job_id)
+        assert restored._pending_notifications()[0].message == "persistent"
+        assert "persistent" in restored._pending_notifications()[0].tool_log
+
+    def test_sending_does_not_reopen_a_completed_subtask_handler(self, job_factory):
+        """A second job completion is not manufactured for an existing handler."""
+        job = job_factory()
+        handler = _completed_subtask_handler(job, subtask_id="child", result="finished")
+        job.chat_log.append(SubTaskLogItem(console_pos=0, handler=handler, tool_log="finished"))
+        job.set_status(JobStatus.DONE)
+
+        job.send_notification("extra information")
+
+        assert handler.is_completed
+        assert handler.result == "finished"
+        assert job.find_sub_task_handler(id="child") is handler
+
+
+# pylint: enable=protected-access
 # ---------------------------------------------------------------------------
 # get_next_request with user messages (str / UserLogItem)
 # ---------------------------------------------------------------------------
diff --git a/tests/test_notification_tools.py b/tests/test_notification_tools.py
new file mode 100644
index 0000000..00cfddf
--- /dev/null
+++ b/tests/test_notification_tools.py
@@ -0,0 +1,196 @@
+"""Behavioral tests for job notification system tools."""
+
+# pylint: disable=no-member,protected-access
+
+from unittest.mock import patch
+
+import pytest
+
+from statek import get_pending_notifications, send_notification_to
+from statek.executors.chat_log_item import NotificationLogItem, SubTaskLogItem
+from statek.executors.job import Job
+from statek.executors.utils import _setup_execution_context, exec_tool
+from statek.llm_api import select_request_tools
+from statek.system import (
+    brief,
+    docstr,
+    find_tools,
+    get_pending_notifications as system_get_pending_notifications,
+)
+from statek.task import SubTaskHandler
+from statek.utils import CallSpec, _statek_ctx_scope, format_tool_spec
+
+
+def test_pending_notifications_empty_and_repeatable(job_factory):
+    """Reading an empty pending queue never changes the job."""
+    job = job_factory()
+    with _statek_ctx_scope({"job": job}):
+        assert not get_pending_notifications()
+        assert not get_pending_notifications()
+    assert job.chat_log == []
+
+
+def test_pending_notifications_are_ordered_detached_and_non_consuming(job_factory):
+    """Returned collection is independent while its entries retain identity."""
+    job = job_factory()
+    first = SubTaskLogItem(console_pos=0, handler=SubTaskHandler(job=job, id="first"))
+    second = SubTaskLogItem(console_pos=0, handler=SubTaskHandler(job=job, id="second"))
+    job._pending_notifications().extend([first, second])
+    initial_status = job.status
+
+    with _statek_ctx_scope({"job": job}):
+        snapshot = get_pending_notifications()
+        assert snapshot == [first, second]
+        snapshot.clear()
+        assert get_pending_notifications() == [first, second]
+
+    assert job.status == initial_status
+    assert job.chat_log == []
+    assert job._pending_notifications() == [first, second]
+    job._process_pending_notifications()
+    with _statek_ctx_scope({"job": job}):
+        assert not get_pending_notifications()
+
+
+def test_pending_notifications_initializes_old_job_queue(job_factory):
+    """Previously stored jobs without an initialized queue can be inspected."""
+    job = job_factory()
+    job._Job__pending_chat_log = None
+    with _statek_ctx_scope({"job": job}):
+        assert not get_pending_notifications()
+
+
+def test_pending_notifications_includes_mixed_items(job_factory):
+    """The getter retains text and subtask notifications in arrival order."""
+    job = job_factory()
+    subtask = SubTaskLogItem(console_pos=0, handler=SubTaskHandler(job=job))
+    message = NotificationLogItem(console_pos=0, message="hello")
+    job._pending_notifications().extend([message, subtask])
+
+    with _statek_ctx_scope({"job": job}):
+        assert get_pending_notifications() == [message, subtask]
+        assert get_pending_notifications() == [message, subtask]
+
+    assert not job.chat_log
+
+
+def test_pending_notifications_requires_current_job():
+    """Missing execution context is reported rather than treated as empty."""
+    with _statek_ctx_scope({}):
+        with pytest.raises(RuntimeError, match="current job"):
+            get_pending_notifications()
+
+
+def test_pending_notifications_system_discovery_and_scope():
+    """Registered tool is selectable in system scopes but not application-only."""
+    assert get_pending_notifications is system_get_pending_notifications
+    assert get_pending_notifications in find_tools("SYSTEM")
+    assert get_pending_notifications in select_request_tools(
+        metadata={"LLM_TOOLS_SCOPE": "SYSTEM"}, available_tools=[]
+    )
+    assert get_pending_notifications in select_request_tools(
+        metadata={"LLM_TOOLS_SCOPE": "+get_pending_notifications"}, available_tools=[]
+    )
+    assert get_pending_notifications not in select_request_tools(
+        metadata={"LLM_TOOLS_SCOPE": "APPLICATION"}, available_tools=[]
+    )
+
+
+def test_pending_notifications_is_injected_into_python_execution(job_factory):
+    """A job's Python environment exposes the registered system tool."""
+    job = job_factory()
+    global_context = {}
+    with _setup_execution_context(job, global_context, {}):
+        assert global_context["get_pending_notifications"]() == []
+
+
+def test_send_notification_to_job_calls_delivery_once(job_factory):
+    """Forwarding a direct Job target preserves the exact message."""
+    target = job_factory()
+    with patch.object(Job, "send_notification", autospec=True) as send:
+        assert send_notification_to(target, "Żółw\nexit('not code')") is None
+    send.assert_called_once_with(target, "Żółw\nexit('not code')")
+
+
+def test_send_notification_to_unfinished_handler_uses_its_job(job_factory):
+    """A handler can be used while its child job is still running."""
+    target = job_factory()
+    handler = SubTaskHandler(job=target, id="child")
+    send_notification_to(handler, "hello")
+
+    assert handler.is_completed is False
+    assert target._pending_notifications()[0].message == "hello"
+    assert not target.chat_log
+
+
+@pytest.mark.parametrize("target", ["unknown_job", None, 17])
+def test_send_notification_to_rejects_invalid_external_targets(job_factory, target):
+    """Unsupported tool input does not partly deliver a notification."""
+    sender = job_factory()
+    with pytest.raises(TypeError, match="target"):
+        send_notification_to(target, "hello")
+    assert not sender.chat_log
+    assert not sender._pending_notifications()
+
+
+def test_send_notification_to_is_registered_and_documented(capsys):
+    """Agents can discover the tool, its signature, and brief/full help."""
+    assert send_notification_to in find_tools("SYSTEM")
+    assert send_notification_to not in find_tools("APPLICATION")
+    assert send_notification_to.__annotations__["target"] == SubTaskHandler | Job
+    assert send_notification_to in select_request_tools(
+        metadata={"LLM_TOOLS_SCOPE": "SYSTEM"}, available_tools=[]
+    )
+    assert send_notification_to in select_request_tools(
+        metadata={"LLM_TOOLS_SCOPE": "+send_notification_to"}, available_tools=[]
+    )
+    assert send_notification_to not in select_request_tools(
+        metadata={"LLM_TOOLS_SCOPE": "APPLICATION"}, available_tools=[]
+    )
+    brief(send_notification_to)
+    brief_output = capsys.readouterr().out
+    docstr(send_notification_to)
+    doc_output = capsys.readouterr().out
+    assert "send_notification_to" in brief_output
+    assert "send_notification_to" in doc_output
+    assert len(doc_output) > len(brief_output)
+
+
+def test_send_notification_to_formal_schema_accepts_local_target_name():
+    """The provider must request a name resolvable to a Job or handler, not an array."""
+    spec = format_tool_spec(send_notification_to)["function"]["parameters"]
+
+    assert spec["properties"]["target"]["type"] == "string"
+    assert spec["properties"]["message"]["type"] == "string"
+    assert spec["required"] == ["target", "message"]
+
+
+def test_send_notification_to_injected_into_python_context(job_factory):
+    """A Python-CLI job can call the system tool with an existing child."""
+    sender = job_factory()
+    target = job_factory()
+    global_context = {}
+    with _setup_execution_context(sender, global_context, {}):
+        global_context["send_notification_to"](target, "from python")
+    assert target._pending_notifications()[0].message == "from python"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("target_kind", ["job", "handler"])
+async def test_send_notification_to_formal_call_binds_local_target(job_factory, target_kind):
+    """Formal tool calls resolve local Job and handler names through the union type."""
+    sender = job_factory()
+    target = job_factory()
+    local_target = target if target_kind == "job" else SubTaskHandler(job=target)
+    sender.add_locals(child=local_target)
+    call = CallSpec(
+        id="notification", func_name="send_notification_to",
+        kwargs={"target": "child", "message": "from formal tool"},
+    )
+
+    result, error = await exec_tool(call, sender)
+
+    assert error is None
+    assert result == ""
+    assert len(target._pending_notifications()) == 1
+    assert target._pending_notifications()[0].message == "from formal tool"
diff --git a/tests/test_run_job_step.py b/tests/test_run_job_step.py
index 248abe9..2654b01 100644
--- a/tests/test_run_job_step.py
+++ b/tests/test_run_job_step.py
@@ -9,10 +9,13 @@
 import dbzero as db0
 
 from statek.agents.agent import Agent
+import statek.settings as notification_settings
+from statek.chat_history import ChatRole
 from statek.prompt_config import make_system_prompt
 from statek.agents.dialog_agent import DialogAgent
 from statek.executors.chat_log_item import (
     LLM_LogItem,
+    NotificationLogItem,
     PostProcessedItem,
     ReminderLogItem,
     SubTaskLogItem,
@@ -35,6 +38,165 @@
 _critical_error_calls = []
 
 
+# pylint: disable=protected-access
+class TestTextNotificationDrain:
+    """Queued text notifications reach the next safe provider request."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("status", [JobStatus.READY, JobStatus.STARTED])
+    async def test_queued_message_is_visible_before_request(self, job_factory, status):
+        """Plain code without exit and fresh jobs both drain before the LLM call."""
+        job = job_factory()
+        job.set_status(status)
+        if status == JobStatus.STARTED:
+            job.chat_log.append(LLM_LogItem(console_pos=0, llm_resp="print('output')"))
+        notification = NotificationLogItem(console_pos=0, message="hello", tool_log="hello")
+        job._pending_notifications().append(notification)
+        mock_api = MagicMock()
+        mock_api.process_request = AsyncMock(return_value=_llm_response("pass"))
+
+        with patch("statek.executors.utils.LLM_API") as mock_llm_api, \
+             patch("statek.executors.utils.get_llm_harness", return_value=MagicMock()):
+            mock_llm_api.get.return_value = mock_api
+            await run_job_step(job)
+
+        sent_history = mock_api.process_request.await_args.kwargs["chat_history"]
+        assert any(item.role == ChatRole.TOOL and item.content == "hello" for item in sent_history)
+        if status == JobStatus.STARTED:
+            assert notification.console_pos == 1
+            assert any(item.content == "output" for item in sent_history)
+        assert not job._pending_notifications()
+
+    @pytest.mark.asyncio
+    async def test_unresolved_future_keeps_notification_pending(self, job_factory):
+        """A message must not wake a job waiting for its own future."""
+        job = job_factory()
+        job.set_status(JobStatus.STARTED)
+        job.set_status(JobStatus.SUSPENDED)
+        job.awaited_result = create_future_not_ready()
+        job._pending_notifications().append(NotificationLogItem(console_pos=0, message="hello"))
+        with patch("statek.executors.utils.get_llm_harness", return_value=MagicMock()), \
+             patch("statek.executors.utils.LLM_API") as mock_llm_api:
+            assert await run_job_step(job) is False
+        mock_llm_api.get.assert_not_called()
+        assert len(job._pending_notifications()) == 1
+        assert not job.chat_log
+
+
+# pylint: enable=protected-access
+
+
+class TestSendNotificationExecution:
+    """Sending a notification does not bypass existing job execution boundaries."""
+
+    @pytest.mark.asyncio
+    async def test_reopened_done_job_sees_saved_message_without_replaying_code(self, job_factory):
+        """A restarted job has no stale exit and asks the provider about the message."""
+        job = job_factory()
+        job.chat_log.append(LLM_LogItem(console_pos=0, llm_resp="exit('old')"))
+        job.set_status(JobStatus.DONE)
+        job.py_env.exit_status = "old"
+        job.send_notification("new information")
+        mock_api = MagicMock(process_request=AsyncMock(return_value=_llm_response("pass")))
+
+        with patch("statek.executors.utils.LLM_API") as mock_llm_api, \
+             patch("statek.executors.utils.get_llm_harness", return_value=MagicMock()):
+            mock_llm_api.get.return_value = mock_api
+            await run_job_step(job)
+
+        history = mock_api.process_request.await_args.kwargs["chat_history"]
+        assert any(
+            item.role == ChatRole.TOOL and "new information" in item.content
+            for item in history
+        )
+        assert job.status == JobStatus.STARTED
+        assert job.py_env.exit_status is None
+
+    @pytest.mark.asyncio
+    async def test_ready_warmup_keeps_pending_message_until_warmup_finishes(self, job_def_factory):
+        """A notification sent during warmup waits for its safe request boundary."""
+        job = Job(job_def=job_def_factory(warmup_code="print('warmup')"))
+        job.send_notification("warmup message")
+        mock_api = MagicMock(process_request=AsyncMock(return_value=_llm_response("pass")))
+
+        with patch("statek.executors.utils.LLM_API") as mock_llm_api, \
+             patch("statek.executors.utils.get_llm_harness", return_value=MagicMock()):
+            mock_llm_api.get.return_value = mock_api
+            await run_job_step(job)
+
+        history = mock_api.process_request.await_args.kwargs["chat_history"]
+        assert any(
+            item.role == ChatRole.TOOL and "warmup message" in item.content
+            for item in history
+        )
+        assert job.status == JobStatus.STARTED
+
+    @pytest.mark.asyncio
+    async def test_send_during_suspended_future_does_not_release_it(self, job_factory):
+        """No provider call occurs while the recipient's future is unresolved."""
+        job = job_factory()
+        job.set_status(JobStatus.SUSPENDED)
+        future = create_future_not_ready()
+        job.awaited_result = future
+        job.send_notification("wait for me")
+
+        with patch("statek.executors.utils.get_llm_harness", return_value=MagicMock()), \
+             patch("statek.executors.utils.LLM_API") as mock_llm_api:
+            assert await run_job_step(job) is False
+
+        mock_llm_api.get.assert_not_called()
+        assert job.awaited_result is future
+        assert job.status == JobStatus.SUSPENDED
+        assert not job.chat_log
+
+    @pytest.mark.asyncio
+    async def test_code_exit_waits_for_notification_follow_up(self, job_factory):
+        """A pending message gets a request before code-triggered DONE."""
+        job = job_factory()
+        job.set_status(JobStatus.STARTED)
+        job.chat_log.append(LLM_LogItem(console_pos=0, llm_resp="exit('done')"))
+        job.send_notification("read this first")
+        mock_api = MagicMock(process_request=AsyncMock(return_value=_llm_response("pass")))
+
+        with patch("statek.executors.utils.LLM_API") as mock_llm_api, \
+             patch("statek.executors.utils.get_llm_harness", return_value=MagicMock()):
+            mock_llm_api.get.return_value = mock_api
+            assert await run_job_step(job) is False
+            assert job.status == JobStatus.STARTED
+            assert await run_job_step(job) is False
+
+        history = mock_api.process_request.await_args.kwargs["chat_history"]
+        assert any(
+            item.role == ChatRole.TOOL and "read this first" in item.content
+            for item in history
+        )
+        assert job.py_env.exit_status is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("chat_style", [
+        notification_settings.ChatStyle.DIRECT,
+        notification_settings.ChatStyle.MD_DIALOG,
+    ])
+    async def test_dialog_autofinalization_includes_notification(self, job_def_factory, chat_style):
+        """A dialog reply can finish after it has received the queued message."""
+        job_def = job_def_factory()
+        job_def.set_chat_style(chat_style)
+        job = Job(job_def=job_def, job_status=JobStatus.STARTED)
+        job.send_notification("dialog note")
+        mock_api = MagicMock(process_request=AsyncMock(return_value=_llm_response("reply")))
+
+        with patch("statek.executors.utils.LLM_API") as mock_llm_api, \
+             patch("statek.executors.utils.get_llm_harness", return_value=MagicMock()), \
+             patch("statek.executors.utils.handle_dialog", new_callable=AsyncMock):
+            mock_llm_api.get.return_value = mock_api
+            result = await run_job_step(job)
+
+        history = mock_api.process_request.await_args.kwargs["chat_history"]
+        assert any(item.role == ChatRole.TOOL and "dialog note" in item.content for item in history)
+        assert result is True
+        assert job.status == JobStatus.DONE
+
+
 @error_handler
 def _capture_critical_error(context, error=None):
     """Capture application critical-error notifications from warmup failures."""
@@ -1915,10 +2077,10 @@ async def test_direct_text_only_response_with_reminder_continues(
 
 
     @pytest.mark.asyncio
-    async def test_direct_text_only_response_with_pending_subtask_continues(
+    async def test_direct_text_only_response_with_pending_subtask_in_request(
         self, job_def_factory, db0_fixture  # pylint: disable=unused-argument
     ):
-        """Pending subtask notifications are processed before DIRECT auto-exit."""
+        """A pending subtask is visible in the request before DIRECT auto-exit."""
         job = self._make_job(job_def_factory)
         handler = _completed_subtask_handler(
             Job(job_def=job.job_def, job_status=JobStatus.READY),
@@ -1941,11 +2103,12 @@ async def test_direct_text_only_response_with_pending_subtask_continues(
             mock_llm_api_cls.get.return_value = mock_api
             result = await run_job_step(job)
 
-        assert result is False
-        assert job.status == JobStatus.STARTED
-        assert isinstance(job.chat_log[-1], SubTaskLogItem)
-        assert job.chat_log[-1].handler is handler
-        assert job.py_env.exit_status is None
+        assert result is True
+        assert job.status == JobStatus.DONE
+        assert isinstance(job.chat_log[-2], SubTaskLogItem)
+        assert job.chat_log[-2].handler is handler
+        sent_history = mock_api.process_request.await_args.kwargs["chat_history"]
+        assert any(item.role == ChatRole.TOOL and item.content == "done" for item in sent_history)
 
 class TestHandleDialogMarkdownMedia:
     """Tests markdown media-link preservation during dialog delivery."""