[BUG] No error indication is displayed for a message after a critical…

statek/executors/job.py

@@ -585,6 +585,17 @@ def __call__(self, error: Optional[Exception] = None):
             pass
 
 
+def _notify_bound_critical_error(
+    binding: Tuple[Callable, Any], error: Optional[Exception] = None
+) -> None:
+    """Invoke an application handler stored as a replaceable error-handler entry."""
+    error_handler, context = binding
+    error_handler(context, error=error)
+
+
+_notify_bound_critical_error.is_error_handler = True
+
+
 @memo
 @db0.tag_fields("agent", "job_def", "job_status")
 class Job:
@@ -823,6 +834,38 @@ def add_error_handlers_from(self, parent_job: 'Job') -> None:
         """Copy all registered error handlers from *parent_job* into this job."""
         self.error_handlers.extend(parent_job.error_handlers)
 
+    def bind_critical_error_context(self, context: Any) -> None:
+        """Bind the latest accepted application input to critical job failure.
+
+        The job's agent may expose a ``critical_error_handler`` adapter. An
+        inherited binding is retained when the child agent does not expose its
+        own adapter. Only the previous latest-input binding is replaced; other
+        registered error handlers remain unchanged.
+
+        Args:
+            context: Application-specific input to report if the job fails.
+        """
+        current = next(
+            (
+                entry
+                for entry in self.error_handlers
+                if entry.error_handler is _notify_bound_critical_error
+            ),
+            None,
+        )
+        handler = self.agent.get_adapter("critical_error_handler")
+        if handler is None and current is not None:
+            handler = current.context[0]
+        self.error_handlers = [
+            entry
+            for entry in self.error_handlers
+            if entry.error_handler is not _notify_bound_critical_error
+        ]
+        if handler is not None:
+            self.add_error_handler(
+                _notify_bound_critical_error, (handler, context)
+            )
+
     def notify_handlers(self, error: Optional[Exception] = None) -> None:
         """Invoke all registered error handlers and clear the handler list.
 
@@ -2370,6 +2413,7 @@ def push_user_message(self, message: Any) -> bool:
         Returns:
             True if the job was transitioned DONE → STARTED, False otherwise.
         """
+        source_message = message
         message = self._resolve_user_message(message)
 
         if self.job_def.chat_style in (ChatStyle.MD_DIALOG, ChatStyle.DIRECT):  # pylint: disable=no-member
@@ -2389,6 +2433,7 @@ def push_user_message(self, message: Any) -> bool:
             else:
                 self.py_env.push_log[key] = [existing, message]
 
+        reactivated = False
         if self.status == JobStatus.DONE:  # pylint: disable=no-member
             self.num_completions = 1 if self.num_completions is None else self.num_completions + 1
             self.set_status(JobStatus.STARTED)  # pylint: disable=no-member
@@ -2397,5 +2442,7 @@ def push_user_message(self, message: Any) -> bool:
                 console_pos=len(self.py_env.console) if self.py_env.console else 0,
                 llm_resp=None
             ))
-            return True
-        return False  # pylint: disable=no-member
+            reactivated = True
+
+        self.bind_critical_error_context(source_message)
+        return reactivated

statek/executors/utils.py

@@ -1371,7 +1371,8 @@ def process_agent_events(
                 break
             events = queue.pop_from_agent_queue(agent, max_count - processed)
             for event in events:
-                create_new_job(agent, shared_vars={local_name: event})
+                job = create_new_job(agent, shared_vars={local_name: event})
+                job.bind_critical_error_context(event)
                 processed += 1
 
 

tests/test_handle_critical_error.py

@@ -6,6 +6,7 @@
 from typing import Callable
 from unittest.mock import AsyncMock, patch
 
+import dbzero as db0
 import pytest
 
 from statek.exceptions import LLM_HarnessError
@@ -15,10 +16,12 @@
 from statek.executors.utils import handle_critical_error, job_worker
 from statek.system import error_handler
 from statek.utils import _statek_ctx_scope
+from tests.conftest import DB0_DIR
 
 
 # Module-level capture list — mutations visible across calls
 _call_log = []
+_application_call_log = []
 
 
 @error_handler
@@ -27,6 +30,12 @@ def _capture(context, error=None):  # pylint: disable=unused-argument
     _call_log.append((context, error))
 
 
+@error_handler
+def _capture_application_error(context, error=None):
+    """Capture the job's application-level critical-error notification."""
+    _application_call_log.append((context, error))
+
+
 class TestHandleCriticalError:
     """Unit tests for handle_critical_error."""
 
@@ -43,6 +52,51 @@ def test_notifies_job_handlers_with_error(self, job_factory):
         assert len(_call_log) == 1
         assert _call_log[0][1] is exc
 
+    def test_bound_application_handler_uses_existing_handler_collection(
+        self, job_factory
+    ):
+        """Latest-input binding replaces itself without adding state to Job."""
+        _call_log.clear()
+        _application_call_log.clear()
+        job = job_factory()
+        job.job_def.agent.context["critical_error_handler"] = (
+            _capture_application_error
+        )
+        job.add_error_handler(_capture, "ordinary-context")
+        job.bind_critical_error_context("first-message")
+        job.bind_critical_error_context("latest-message")
+
+        assert not hasattr(job, "_Job__critical_error_handler")
+        assert len(job.error_handlers) == 2
+
+        exc = RuntimeError("boom")
+        with _statek_ctx_scope({'job': job}):
+            handle_critical_error(exc)
+            handle_critical_error(exc)
+
+        assert _call_log == [("ordinary-context", exc)]
+        assert _application_call_log == [("latest-message", exc)]
+        assert job.error_handlers == []
+
+    def test_bound_application_handler_survives_reopen(self, job_factory):
+        """The standard persisted handler collection retains the active binding."""
+        _application_call_log.clear()
+        job = job_factory()
+        job.job_def.agent.context["critical_error_handler"] = (
+            _capture_application_error
+        )
+        job.bind_critical_error_context("latest-message")
+        identifier = db0.uuid(job)
+
+        db0.close()
+        db0.init(DB0_DIR, read_write=True)
+        db0.open("test_prefix", "rw")
+        restored = db0.fetch(identifier)
+        failure = RuntimeError("after reopen")
+        restored.notify_handlers(error=failure)
+
+        assert _application_call_log == [("latest-message", failure)]
+
     def test_no_job_in_context_does_not_raise(self):
         """handle_critical_error is a no-op when no job is found in context."""
         # No _STATEK_CTX in scope — must not raise
@@ -118,3 +172,21 @@ async def test_job_worker_notifies_handlers_on_generic_exception(self, job_facto
         assert job.error.error_message == "unexpected failure"
         assert job.exception_count == 1
         assert job.py_env.exceptions[0].kind == ErrorKind.EXECUTION
+
+    @pytest.mark.asyncio
+    async def test_job_worker_notifies_bound_application_handler(self, job_factory):
+        """A worker-level critical exception reports the captured application context."""
+        _application_call_log.clear()
+        job = job_factory()
+        job.job_def.agent.context["critical_error_handler"] = (
+            _capture_application_error
+        )
+        job.bind_critical_error_context("latest-message")
+        semaphore = asyncio.Semaphore(1)
+
+        exc = ValueError("unexpected failure")
+        with patch('statek.executors.utils.run_job_step', new_callable=AsyncMock) as mock_step:
+            mock_step.side_effect = exc
+            await job_worker(semaphore, job)
+
+        assert _application_call_log == [("latest-message", exc)]

tests/test_process_agent_events.py

@@ -8,6 +8,16 @@
 from statek.executors.utils import process_agent_events
 from statek.prompt_config import make_system_prompt
 from statek.statek_push_queue import StatekPushQueue
+from statek.system import error_handler
+
+
+_critical_error_calls = []
+
+
+@error_handler
+def _capture_critical_error(context, error=None):
+    """Capture application critical-error notifications for event tests."""
+    _critical_error_calls.append((context, error))
 
 
 def _current_queue_prefixes():
@@ -46,6 +56,26 @@ def test_process_agent_events_creates_job_with_event_shared_var(db0_fixture):
     assert job.py_env.local_state["event"] is event
 
 
+def test_process_agent_events_binds_event_as_critical_error_context(db0_fixture):
+    """A queued event is bound before its newly created job can execute."""
+    _critical_error_calls.clear()
+    agent = _make_agent()
+    agent.context["critical_error_handler"] = _capture_critical_error
+    event = _QueuedEvent("hello")
+    queue = StatekPushQueue()
+    queue.push_to_agent_queue(agent, event)
+
+    process_agent_events(
+        agents={agent}, max_count=10, queue_prefixes=_current_queue_prefixes()
+    )
+
+    job = next(iter(db0.find(Job, db0.as_tag(agent))))
+    failure = RuntimeError("event failed")
+    job.notify_handlers(error=failure)
+
+    assert _critical_error_calls == [(event, failure)]
+
+
 def test_process_agent_events_reuses_matching_job_def(db0_fixture):
     agent = _make_agent()
     queue = StatekPushQueue()

tests/test_process_push_notifications.py

@@ -7,6 +7,16 @@
 from statek.locale import StatekCountryCode, StatekLangCode, StatekLocale
 from statek.statek_push_queue import StatekPushQueue
 from statek.executors.utils import process_push_notifications
+from statek.system import error_handler
+
+
+_critical_error_calls = []
+
+
+@error_handler
+def _capture_critical_error(context, error=None):
+    """Capture application critical-error notifications for push tests."""
+    _critical_error_calls.append((context, error))
 
 
 def _current_queue_prefixes():
@@ -25,6 +35,13 @@ def _make_started_job():
                job_status=JobStatus.STARTED)  # pylint: disable=no-member
 
 
+def _make_started_job_with_critical_error_handler():
+    """Create a started job configured with an application critical handler."""
+    job = _make_started_job()
+    job.job_def.agent.context["critical_error_handler"] = _capture_critical_error
+    return job
+
+
 @db0.memo
 class _QueuedMessage:
     def __init__(self, text):
@@ -123,6 +140,41 @@ def test_processes_multiple_notifications_for_same_job(self, db0_fixture):
         assert "msg1" in val
         assert "msg2" in val
 
+    def test_critical_error_targets_last_accepted_notification(self, db0_fixture):
+        """Only the latest message accepted into active processing is reported."""
+        _critical_error_calls.clear()
+        job = _make_started_job_with_critical_error_handler()
+        queue = StatekPushQueue()
+        first = _QueuedMessage("first")
+        second = _QueuedMessage("second")
+        queue.push_to_job_console(db0.uuid(job), first)
+        queue.push_to_job_console(db0.uuid(job), second)
+
+        process_push_notifications(queue_prefixes=_current_queue_prefixes())
+        failure = RuntimeError("processing failed")
+        job.notify_handlers(error=failure)
+
+        assert _critical_error_calls == [(second, failure)]
+
+    def test_critical_error_does_not_target_unaccepted_notification(self, db0_fixture):
+        """A message left in the queue is not selected by an earlier failure."""
+        _critical_error_calls.clear()
+        job = _make_started_job_with_critical_error_handler()
+        queue = StatekPushQueue()
+        accepted = _QueuedMessage("accepted")
+        queued = _QueuedMessage("queued")
+        queue.push_to_job_console(db0.uuid(job), accepted)
+        queue.push_to_job_console(db0.uuid(job), queued)
+
+        process_push_notifications(
+            max_count=1, queue_prefixes=_current_queue_prefixes()
+        )
+        failure = RuntimeError("processing failed")
+        job.notify_handlers(error=failure)
+
+        assert _critical_error_calls == [(accepted, failure)]
+        assert queue.pop_from_job_console(10) == [(db0.uuid(job), queued)]
+
     def test_notification_keeps_job_definition_locale(self, db0_fixture):
         """An English message does not change a Polish job's response locale."""
         job = _make_started_job()

tests/test_run_job_step.py

@@ -26,11 +26,21 @@
 from statek.llm_harness import LLM_Harness
 from statek.llm_api import LLM_Response, LLM_StepData, LLM_Stats, CallParams, OpenRouter_API
 from statek.settings import LLM_API_Settings
+from statek.system import error_handler
 from statek.task import SubTaskHandler
 from statek.task_difficulty import TaskDifficulty
 from statek.utils import CodeBlock, CallSpec
 
 
+_critical_error_calls = []
+
+
+@error_handler
+def _capture_critical_error(context, error=None):
+    """Capture application critical-error notifications from warmup failures."""
+    _critical_error_calls.append((context, error))
+
+
 @db0.memo
 @dataclass
 class MemoObject:
@@ -842,6 +852,33 @@ async def test_warmup_exception_sets_job_done(self, db0_fixture):  # pylint: dis
 
         assert job.status == JobStatus.DONE
 
+    @pytest.mark.asyncio
+    async def test_warmup_exception_notifies_bound_application_handler(
+        self, db0_fixture  # pylint: disable=unused-argument
+    ):
+        """A critical warmup failure reports the context bound before execution."""
+        _critical_error_calls.clear()
+        agent = Agent(
+            role="warmup_exc_handler",
+            _system_prompt=make_system_prompt("Test"),
+            _metadata={"MODEL": "test-model"},
+            _tools=[],
+        )
+        agent.context["critical_error_handler"] = _capture_critical_error
+        job_def = JobDef(
+            agent=agent,
+            job_params=None,
+            warmup_code="raise ValueError('boom')",
+        )
+        job = Job(job_def=job_def, job_status=JobStatus.READY)
+        job.bind_critical_error_context("event-message")
+
+        await run_job_step(job)
+
+        assert len(_critical_error_calls) == 1
+        assert _critical_error_calls[0][0] == "event-message"
+        assert isinstance(_critical_error_calls[0][1], ValueError)
+
     @pytest.mark.asyncio
     async def test_warmup_exception_calls_set_error_on_job_def(self, db0_fixture):  # pylint: disable=unused-argument
         """A non-FutureError exception in warmup calls job_def.set_error with the exception."""