diff --git a/statek/executors/job.py b/statek/executors/job.py index 9d78929..1ffaf23 100644 --- a/statek/executors/job.py +++ b/statek/executors/job.py @@ -34,7 +34,7 @@ ) import dbzero as db0 from dbzero import memo, enum -from statek.pyenv import PyEnv +from statek.pyenv import Error, ErrorKind, PyEnv from statek.executors.llm_usage import LLM_Usage from statek.executors.chat_log_item import ( ChatLogItem, @@ -894,21 +894,22 @@ def _log_tool_call_result(self, call_spec, result: str): else: self._log(call_line) - def console_append(self, output: str, error_message: str = None): + def console_append( + self, output: str, error: Optional[Error] = None, + ) -> None: """ Append output to the console and optionally log it. Args: output: The output string to append - error_message: optional error message (if execution resulted in an exception) + error: optional classified error associated with the console entry """ self.py_env.console_append(output) - if error_message is not None: - last_item = self.chat_log[-1] if self.chat_log else None - console_pos = last_item.console_pos if isinstance(last_item, ChatLogItem) else 0 + if error is not None: + console_pos = len(self.py_env.console) - 1 if self.py_env.exceptions is None: self.py_env.exceptions = {} - self.py_env.exceptions[console_pos] = error_message + self.py_env.exceptions[console_pos] = error # Console is logged in batches at block/turn boundaries (see _log_console_batch) @property @@ -2115,30 +2116,29 @@ def _warmup_console_positions(self) -> set: """Returns console_pos values that correspond to warmup blocks.""" return {item.console_pos for item in self.chat_log if isinstance(item, WarmupLogItem)} - @staticmethod - def _tool_error_count(item) -> int: - """Return the number of ToolError entries in item.tool_log.""" - if item.tool_log is None: - return 0 - if isinstance(item.tool_log, ToolError): - return 1 - if isinstance(item.tool_log, str): - return 0 - return sum(1 for entry in item.tool_log if isinstance(entry, ToolError)) + def _exception_messages_by_turn(self) -> Iterable[Tuple[ChatLogItem, List[str]]]: + """Resolve console events once; latest equal execution boundary wins. - def _has_exception(self, item) -> bool: - """Return True if *item* has a code exception or a tool exception.""" - if self.py_env.exceptions and item.console_pos in self.py_env.exceptions: - return True - return self._tool_error_count(item) > 0 - - def _exception_count_for_item(self, item) -> int: - """Return the number of exceptions for a single chat log item.""" - count = 0 - if self.py_env.exceptions and item.console_pos in self.py_env.exceptions: - count += 1 - count += self._tool_error_count(item) - return count + Error console entries belong to half-open execution ranges. + """ + items = [item for item in self.chat_log + if isinstance(item, (LLM_LogItem, WarmupLogItem))] + errors = sorted((self.py_env.exceptions or {}).items()) + position = 0 + for index, item in enumerate(items): + end = items[index + 1].console_pos if index + 1 < len(items) else None + messages = [] + while position < len(errors) and (end is None or errors[position][0] < end): + key, error = errors[position] + position += 1 + if key >= item.console_pos and error.kind == ErrorKind.EXECUTION: + messages.append(error.message) + tool_log = item.tool_log + if isinstance(tool_log, ToolError): + messages.append(tool_log.err_message) + elif tool_log is not None and not isinstance(tool_log, str): + messages.extend(entry.err_message for entry in tool_log if isinstance(entry, ToolError)) + yield item, messages @property def exception_count(self) -> int: @@ -2147,9 +2147,7 @@ def exception_count(self) -> int: Counts both code-execution exceptions (py_env.exceptions) and tool-call exceptions (ToolError entries in tool_log). """ - if not self.chat_log: - return 0 - return sum(self._exception_count_for_item(item) for item in self.chat_log + return sum(len(messages) for item, messages in self._exception_messages_by_turn() if isinstance(item, LLM_LogItem)) @property @@ -2162,10 +2160,10 @@ def max_consecutive_exceptions(self) -> int: return 0 max_streak = 0 streak = 0 - for item in self.chat_log: + for item, messages in self._exception_messages_by_turn(): if not isinstance(item, LLM_LogItem): continue - if self._has_exception(item): + if messages: streak += 1 max_streak = max(max_streak, streak) else: diff --git a/statek/executors/utils.py b/statek/executors/utils.py index 7f5f38d..5d7a2c3 100644 --- a/statek/executors/utils.py +++ b/statek/executors/utils.py @@ -51,6 +51,7 @@ from statek.statek_push_queue import StatekPushQueue from statek.llm_api import LLM_API, LLM_Response from statek.llm_harness import get_llm_harness +from statek.pyenv import Error, ErrorKind from statek.model_name import ensure_model_name, format_model_for_provider, select_model_provider from statek.settings import ChatStyle, full_agent_trace, get_statek_settings, statek_log from statek.system import inject_context @@ -617,7 +618,7 @@ async def exec_step(code_str: str, job: Job, instr_num: Optional[int] = None, _exec_code_body( code_str, job, global_context, local_context, output_fn=lambda s: job.console_append(s), - error_fn=lambda msg: job.console_append(msg, error_message=msg), + error_fn=lambda msg: job.console_append(msg, error=Error(ErrorKind.EXECUTION, msg)), instr_num=instr_num, allowed_tools=allowed_tools, ) @@ -1073,7 +1074,7 @@ async def run_job_step(job: Job, provider: str = None) -> bool: and job.job_def.chat_style != ChatStyle.DIRECT # pylint: disable=no-member and _is_empty_code(code_str)): error_msg = "Error: no code submitted." - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error=Error(ErrorKind.EXECUTION, error_msg)) # Step 5: Execute regular tool calls (not python_cli) if present and not a continuation last_chat_log_item = job.chat_log[-1] if job.chat_log else None @@ -1238,7 +1239,7 @@ def _cli_error_append(cli_idx, text): await handle_dialog(processed_step.text, _local_context=local_context) except Exception as e: error_msg = f"{type(e).__name__}: {e}" - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error=Error(ErrorKind.EXECUTION, error_msg)) dialog_error = True # Step 16: MD_DIALOG/DIRECT text-only responses can finish only after @@ -1425,7 +1426,7 @@ async def job_worker(semaphore, job: Job, provider: str = None): error_msg = f"LLM_HarnessError: {e}" statek_log(error_msg, level='error') job.py_env.exit_status = f"Error: {e}" - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error=Error(ErrorKind.HARNESS, error_msg)) job.error = JobDefError(e) job.set_status(JobStatus.DONE) handle_critical_error(e) @@ -1434,7 +1435,7 @@ async def job_worker(semaphore, job: Job, provider: str = None): import traceback error_msg = f"Job {db0.uuid(job)} failed with error: {e}\n{traceback.format_exc()}" statek_log(error_msg, level='error') - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error=Error(ErrorKind.EXECUTION, error_msg)) job.error = JobDefError(e) job.set_status(JobStatus.DONE) handle_critical_error(e) diff --git a/statek/pyenv.py b/statek/pyenv.py index 6ad2f56..57fa811 100644 --- a/statek/pyenv.py +++ b/statek/pyenv.py @@ -13,11 +13,26 @@ # limitations under the License. from dataclasses import dataclass, field -from typing import List, Dict, Union +from typing import List, Dict, Optional, Union import dbzero as db0 from statek.future import FutureResult + +@db0.enum(values=["EXECUTION", "HARNESS"]) +class ErrorKind: + """Origin of a recorded console error.""" + + +@db0.memo(no_default_tags=True) +@dataclass +class Error: + """Persistent error message classified at its point of origin.""" + + kind: ErrorKind + message: str + + @db0.memo @dataclass class PyEnv: @@ -32,8 +47,9 @@ class PyEnv: local_state: Dict = field(default_factory=dict) # Console outputs of the LLM's program console: List[str] = None - # Optional error message by chat_log item ID - exceptions: Dict[int, str] = None + # Diagnostics by console-entry index, including + # harness-limit messages retained for display but excluded from execution counters. + exceptions: Optional[Dict[int, Error]] = None # Messages pushed into the console of an active job push_log: Dict[int, Union[str, List[str]]] = None # The next instruction ID for continuation diff --git a/tests/test_handle_critical_error.py b/tests/test_handle_critical_error.py index 3d31a85..20941bb 100644 --- a/tests/test_handle_critical_error.py +++ b/tests/test_handle_critical_error.py @@ -3,12 +3,15 @@ # pylint: disable=too-few-public-methods,no-member import asyncio +from typing import Callable from unittest.mock import AsyncMock, patch import pytest from statek.exceptions import LLM_HarnessError -from statek.executors.job import JobDefError, JobStatus +from statek.pyenv import Error, ErrorKind +from statek.executors.job import Job, JobDefError, JobStatus +from statek.executors.chat_log_item import LLM_LogItem from statek.executors.utils import handle_critical_error, job_worker from statek.system import error_handler from statek.utils import _statek_ctx_scope @@ -46,28 +49,60 @@ def test_no_job_in_context_does_not_raise(self): handle_critical_error(RuntimeError("ignored")) @pytest.mark.asyncio - async def test_job_worker_notifies_handlers_on_harness_error(self, job_factory): + @pytest.mark.parametrize("message", [ + "Maximum token usage exceeded: 55320/50002.0", + "Maximum number of exceptions exceeded: 8/6.0", + "Maximum consecutive exceptions exceeded: 8/6.0", + "Maximum number of turns exceeded: 101/100.0", + "Token budget exhausted", + "", + ]) + @pytest.mark.parametrize("with_execution_errors", [False, True]) + async def test_job_worker_notifies_handlers_on_harness_error( + self, job_factory: Callable[..., Job], message: str, with_execution_errors: bool, + ) -> None: """job_worker calls handle_critical_error when LLM_HarnessError is raised.""" _call_log.clear() job = job_factory() + for index in range(3 if with_execution_errors else 1): + job.chat_log.append(LLM_LogItem(console_pos=len(job.py_env.console or []))) + if with_execution_errors: + execution_error = f"ValueError: failure {index}" + job.console_append( + execution_error, error=Error(ErrorKind.EXECUTION, execution_error), + ) + previous_errors = { + key: error.message for key, error in (job.py_env.exceptions or {}).items() + } job.add_error_handler(_capture, "ctx") semaphore = asyncio.Semaphore(1) + exc = LLM_HarnessError(message) with patch('statek.executors.utils.run_job_step', new_callable=AsyncMock) as mock_step: - mock_step.side_effect = LLM_HarnessError("too many turns") + mock_step.side_effect = exc await job_worker(semaphore, job) assert len(_call_log) == 1 - assert isinstance(_call_log[0][1], LLM_HarnessError) + assert _call_log[0][1] is exc assert job.status == JobStatus.DONE assert isinstance(job.error, JobDefError) - assert job.error.error_message == "too many turns" + assert job.error.error_message == message + assert job.py_env.exit_status == f"Error: {message}" + error_msg = f"LLM_HarnessError: {message}" + assert job.py_env.console[-1] == error_msg + assert {key: error.message for key, error in job.py_env.exceptions.items()} == { + **previous_errors, len(job.py_env.console) - 1: error_msg, + } + assert job.exception_count == (3 if with_execution_errors else 0) + assert job.max_consecutive_exceptions == (3 if with_execution_errors else 0) + assert job.py_env.exceptions[len(job.py_env.console) - 1].kind == ErrorKind.HARNESS @pytest.mark.asyncio async def test_job_worker_notifies_handlers_on_generic_exception(self, job_factory): """job_worker calls handle_critical_error for any other critical exception.""" _call_log.clear() job = job_factory() + job.chat_log.append(LLM_LogItem(console_pos=0)) job.add_error_handler(_capture, "ctx") semaphore = asyncio.Semaphore(1) @@ -81,3 +116,5 @@ async def test_job_worker_notifies_handlers_on_generic_exception(self, job_facto assert job.status == JobStatus.DONE assert isinstance(job.error, JobDefError) assert job.error.error_message == "unexpected failure" + assert job.exception_count == 1 + assert job.py_env.exceptions[0].kind == ErrorKind.EXECUTION diff --git a/tests/test_job.py b/tests/test_job.py index dab2dc1..565c23e 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -21,6 +21,7 @@ ) from statek.llm_api import LLM_Response, LLM_StepData, LLM_Stats, OpenRouter_API from statek.model_pricing import set_model_pricing +from statek.pyenv import Error, ErrorKind from statek.model_name import ( ModelName, ensure_model_name, @@ -396,9 +397,11 @@ def test_console_append_error_accepts_initial_legacy_message(self, job_factory): job = job_factory() job.chat_log.append("Initial user message") - job.console_append("Failure", error_message="error") + job.console_append("Failure", error=Error(ErrorKind.EXECUTION, "error")) - assert job.py_env.exceptions == {0: "error"} + assert list(job.py_env.exceptions) == [0] + assert job.py_env.exceptions[0].message == "error" + assert job.py_env.exceptions[0].kind == ErrorKind.EXECUTION def test_contains_ext_ref_false_before_add(self, job_factory): """Job does not report unrelated memo objects as external refs.""" diff --git a/tests/test_warmup_harness_counting.py b/tests/test_warmup_harness_counting.py index 319b369..02e17d2 100644 --- a/tests/test_warmup_harness_counting.py +++ b/tests/test_warmup_harness_counting.py @@ -1,10 +1,16 @@ """Tests for harness counting: warmup exclusions and tool-exception inclusion.""" # pylint: disable=no-member,redefined-outer-name +from typing import Callable + +import dbzero as db0 import pytest from statek.executors.job import Job, JobStatus from statek.executors.chat_log_item import LLM_LogItem, ToolError, WarmupLogItem +from statek.llm_harness import LLM_Harness +from statek.pyenv import Error, ErrorKind +from tests.conftest import DB0_DIR @pytest.fixture @@ -21,7 +27,9 @@ def _make(chat_log_items, exceptions=None): for item in chat_log_items: job.chat_log.append(item) if exceptions: - job.py_env.exceptions = exceptions + job.py_env.exceptions = { + key: Error(ErrorKind.EXECUTION, message) for key, message in exceptions.items() + } return job return _make @@ -65,6 +73,134 @@ def test_mixed_warmup_and_llm(self, db0_fixture, make_job): # pylint: disable=u class TestExceptionCountExcludesWarmup: """exception_count should not include exceptions from warmup blocks.""" + @pytest.mark.parametrize("message", [ + "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", + "LLM_HarnessError: Maximum number of exceptions exceeded: 8/6.0", + "LLM_HarnessError: Maximum consecutive exceptions exceeded: 8/6.0", + "LLM_HarnessError: Maximum number of turns exceeded: 101/100.0", + "Dowolny komunikat zatrzymania", + ]) + def test_terminal_diagnostic_preserves_shared_position_accounting( + self, make_job: Callable[..., Job], message: str, + ) -> None: + """A terminal diagnostic must not duplicate events or join real-error streaks.""" + items = [LLM_LogItem(console_pos=0), LLM_LogItem(console_pos=1), + LLM_LogItem(console_pos=1), LLM_LogItem(console_pos=2)] + items[0].push_tool_result(ToolError(err_message="tool error")) + job = make_job(items) + job.console_append("output") + job.console_append("code error", error=Error(ErrorKind.EXECUTION, "code error")) + job.console_append("another error", error=Error(ErrorKind.EXECUTION, "another error")) + assert job.exception_count == 3 + assert job.max_consecutive_exceptions == 2 + + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) + + assert job.py_env.exceptions[3].message == message + assert job.py_env.exceptions[3].kind == ErrorKind.HARNESS + assert job.exception_count == 3 + assert job.max_consecutive_exceptions == 2 + + @pytest.mark.parametrize("with_execution_error", [False, True]) + def test_distinct_diagnostics_retain_unique_console_keys( + self, make_job: Callable[..., Job], with_execution_error: bool, + ) -> None: + """Successive diagnostics cannot replace an earlier real error or each other.""" + job = make_job([LLM_LogItem(console_pos=0)]) + expected = {} + if with_execution_error: + job.console_append( + "ValueError: bad code", error=Error(ErrorKind.EXECUTION, "ValueError: bad code"), + ) + expected[0] = "ValueError: bad code" + for message in ( + "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", + "LLM_HarnessError: Maximum number of turns exceeded: 101/100.0", + ): + expected[len(expected)] = message + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) + + assert {key: error.message for key, error in job.py_env.exceptions.items()} == expected + assert list(job.py_env.console) == list(expected.values()) + assert job.exception_count == int(with_execution_error) + assert job.max_consecutive_exceptions == int(with_execution_error) + + @pytest.mark.parametrize("message", [ + "ValueError: Maximum token usage exceeded: application quota", + "LLM_HarnessError: application failure", + "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", + ]) + def test_application_errors_are_not_filtered( + self, make_job: Callable[..., Job], message: str, + ) -> None: + """Message wording cannot exclude an application error from accounting.""" + job = make_job([LLM_LogItem(console_pos=0)]) + job.console_append(message, error=Error(ErrorKind.EXECUTION, message)) + assert job.exception_count == 1 + assert job.max_consecutive_exceptions == 1 + + def test_shared_position_counts_one_event(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=1) for _ in range(8)], + exceptions={1: "ValueError: bad code"}) + assert job.exception_count == 1 + assert job.max_consecutive_exceptions == 1 + + def test_new_errors_have_unique_console_keys(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=0)]) + job.console_append("output") + job.console_append("first", error=Error(ErrorKind.EXECUTION, "same error")) + job.console_append("second", error=Error(ErrorKind.EXECUTION, "same error")) + assert {key: error.message for key, error in job.py_env.exceptions.items()} == { + 1: "same error", 2: "same error", + } + assert job.exception_count == 2 + assert job.max_consecutive_exceptions == 1 + + def test_terminal_diagnostics_do_not_poison_resume(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=0) for _ in range(8)]) + for message in ("Token budget exhausted", "Execution stopped"): + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) + assert job.exception_count == 0 + assert job.max_consecutive_exceptions == 0 + job.set_status(JobStatus.DONE) + job.push_user_message("continue") + LLM_Harness(None, 6, 6, None).check_before_step(job) + + def test_shared_warmup_boundary_has_single_owner(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=0), + WarmupLogItem(console_pos=0, warmup_block_num=0)], + exceptions={0: "warmup error"}) + assert job.exception_count == 0 + assert job.max_consecutive_exceptions == 0 + + def test_diagnostic_classification_survives_reopen(self, make_job: Callable[..., Job]) -> None: + """Persist provenance independently of identical diagnostic and error text.""" + job = make_job([LLM_LogItem(console_pos=0)]) + message = "LLM_HarnessError: Maximum token usage exceeded" + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) + job.console_append(message, error=Error(ErrorKind.EXECUTION, message)) + 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 list(restored.py_env.console) == [message, message] + assert restored.py_env.exceptions[0].kind == ErrorKind.HARNESS + assert restored.py_env.exceptions[1].kind == ErrorKind.EXECUTION + assert [error.message for error in restored.py_env.exceptions.values()] == [ + message, message, + ] + assert restored.exception_count == 1 + assert restored.max_consecutive_exceptions == 1 + + def test_success_breaks_shared_position_streak(self, db0_fixture, make_job): # pylint: disable=unused-argument + items = [LLM_LogItem(console_pos=0), LLM_LogItem(console_pos=1), + LLM_LogItem(console_pos=1), LLM_LogItem(console_pos=2)] + items[0].push_tool_result(ToolError(err_message="tool error")) + job = make_job(items, exceptions={1: "code error"}) + assert job.exception_count == 2 + assert job.max_consecutive_exceptions == 1 + def test_no_exceptions(self, db0_fixture, make_job): # pylint: disable=unused-argument job = make_job([LLM_LogItem(console_pos=0)]) assert job.exception_count == 0