diff --git a/statek/executors/job.py b/statek/executors/job.py index de6698b..2d4b2c4 100644 --- a/statek/executors/job.py +++ b/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 diff --git a/statek/executors/utils.py b/statek/executors/utils.py index 7644d3d..1147cbf 100644 --- a/statek/executors/utils.py +++ b/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 diff --git a/tests/test_handle_critical_error.py b/tests/test_handle_critical_error.py index 20941bb..7a66b81 100644 --- a/tests/test_handle_critical_error.py +++ b/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)] diff --git a/tests/test_process_agent_events.py b/tests/test_process_agent_events.py index 4eacc2a..5a01a6b 100644 --- a/tests/test_process_agent_events.py +++ b/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() diff --git a/tests/test_process_push_notifications.py b/tests/test_process_push_notifications.py index 9835da9..d867f08 100644 --- a/tests/test_process_push_notifications.py +++ b/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() diff --git a/tests/test_run_job_step.py b/tests/test_run_job_step.py index cc3333b..7d889e3 100644 --- a/tests/test_run_job_step.py +++ b/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.""" diff --git a/tests/test_task.py b/tests/test_task.py index 29a220b..d5d4a3f 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -21,8 +21,18 @@ from statek.locale import StatekLocale, StatekLangCode, StatekCountryCode from statek.prompt_config import make_system_prompt from statek.settings import StatekSettings +from statek.system import error_handler from statek.utils import CodeBlock, _statek_ctx_scope +_critical_error_calls = [] + + +@error_handler +def _capture_critical_error(context, error=None): + """Capture application critical-error notifications for delegation tests.""" + _critical_error_calls.append((context, error)) + + def _noop_error_handler(context, error=None): """Minimal error handler for tests.""" @@ -394,6 +404,24 @@ def test_delegate_task_with_parent_job_copies_error_handlers( assert len(child_result.job.error_handlers) == 1 assert child_result.job.error_handlers[0].error_handler is _noop_error_handler + def test_delegate_task_snapshots_parent_critical_error_context( + self, db0_fixture, supervised_agent, mock_settings + ): + """A delayed child failure retains the parent's context at delegation time.""" + _critical_error_calls.clear() + supervised_agent.context["critical_error_handler"] = _capture_critical_error + parent_job = delegate_task(supervised_agent).job + parent_job.bind_critical_error_context("delegated-message") + + child_job = delegate_task( + supervised_agent, parent_job=parent_job + ).job + parent_job.push_user_message("new-parent-message") + failure = RuntimeError("child failed") + child_job.notify_handlers(error=failure) + + assert _critical_error_calls == [("delegated-message", failure)] + def test_delegate_task_with_parent_job_stores_parent_job( self, db0_fixture, supervised_agent, mock_settings ): @@ -1308,6 +1336,25 @@ def test_start_dialog_with_parent_job_copies_error_handlers( assert child_job.error_handlers[0].error_handler is _noop_error_handler assert child_job.parent_job is parent_job + def test_start_dialog_rebinds_inherited_context_to_initial_message( + self, db0_fixture, mock_settings + ): + """A child dialog failure targets its own last accepted user message.""" + _critical_error_calls.clear() + agent = DialogAgent( + send_message=_make_send_message, _metadata={"MODEL": "test-model"} + ) + agent.context["critical_error_handler"] = _capture_critical_error + parent_job = start_dialog(agent, message="parent-message") + + child_job = start_dialog( + agent, message="child-message", parent_job=parent_job + ) + failure = RuntimeError("child failed") + child_job.notify_handlers(error=failure) + + assert _critical_error_calls == [("child-message", failure)] + def test_start_dialog_dict_shared_vars_populates_local_state( self, db0_fixture, mock_settings ):