diff --git a/solstone/observe/describe.py b/solstone/observe/describe.py index f148221e9..2ca4d3a6d 100644 --- a/solstone/observe/describe.py +++ b/solstone/observe/describe.py @@ -59,6 +59,7 @@ from solstone.think.journal_io import install_file from solstone.think.markdown import bound_extraction_markdown from solstone.think.prompts import load_prompt from solstone.think.providers import fanout_policy +from solstone.think.providers.shared import is_non_retryable_generate_reason from solstone.think.utils import ( day_from_path, get_config, @@ -1067,7 +1068,13 @@ class VideoProcessor: work_key=work_key, batch=batch, ) - if has_error and req.retry_count < 4: + if ( + has_error + and req.retry_count < 4 + and not is_non_retryable_generate_reason( + getattr(req, "reason_code", None) + ) + ): req.retry_count += 1 total_frames -= 1 # Don't count retries batch.add(req) @@ -1396,7 +1403,13 @@ class VideoProcessor: work_key=work_key, batch=batch, ) - if has_error and req.retry_count < 4: + if ( + has_error + and req.retry_count < 4 + and not is_non_retryable_generate_reason( + getattr(req, "reason_code", None) + ) + ): req.retry_count += 1 batch.add(req) logger.info( diff --git a/solstone/think/providers/shared.py b/solstone/think/providers/shared.py index a9aed7e88..27862b867 100644 --- a/solstone/think/providers/shared.py +++ b/solstone/think/providers/shared.py @@ -20,6 +20,7 @@ from typing import Any, Callable, Literal, Mapping, Optional, Union from typing_extensions import Required, TypedDict from solstone.think.providers import is_cloud_provider +from solstone.think.responsiveness import NON_RESPONSIVE_REASON_CODE from solstone.think.utils import now_ms # --------------------------------------------------------------------------- @@ -271,6 +272,17 @@ RUNTIME_REASON_CODES = frozenset( ) +def is_non_retryable_generate_reason(reason_code: str | None) -> bool: + """Return True when retrying the same generate request cannot change outcome. + + Members are failures deterministic for the same request, so another attempt + only burns quota. `schema_invalid` is deliberately not a member: the model + may produce valid JSON on retry, preserving that high-volume retry path. + """ + + return reason_code == NON_RESPONSIVE_REASON_CODE + + PROVIDER_ERROR_TEXT_CAP_CHARS = 4096 @@ -727,6 +739,7 @@ __all__ = [ "classify_provider_error", "exception_chain", "is_cloud_model_not_found", + "is_non_retryable_generate_reason", "mark_cloud_model_request", "safe_raw", ] diff --git a/tests/test_batch.py b/tests/test_batch.py index 9dc4296bc..9f943b544 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -4,12 +4,14 @@ """Tests for the Batch async batch processor.""" import asyncio +from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from solstone.think.batch import Batch, BatchRequest from solstone.think.models import GEMINI_FLASH, SchemaValidationError +from solstone.think.responsiveness import NON_RESPONSIVE_REASON_CODE def _result(text: str = "Response", finish_reason: str = "stop", **extra): @@ -659,3 +661,51 @@ async def test_batch_full_result_schema_invalid_stays_hard_failure(mock_agenerat assert len(results) == 1 assert results[0].response is None assert "schema validation" in results[0].error + + +@pytest.mark.asyncio +async def test_batch_preserves_non_responsive_exception_reason_code(monkeypatch): + import solstone.think.providers as providers_package + from solstone.think import batch as batch_module + from solstone.think import models + + provider_module = SimpleNamespace( + run_agenerate=AsyncMock( + return_value={ + "text": "I cannot describe this screen.", + "model": "provider-model", + "finish_reason": "stop", + } + ) + ) + monkeypatch.setattr( + models, + "resolve_provider", + lambda _interface: ("fake", "provider-model"), + ) + monkeypatch.setattr( + providers_package, + "get_provider_module", + lambda _provider: provider_module, + ) + monkeypatch.setattr( + batch_module, + "resolve_provider", + lambda _interface: ("fake", "provider-model"), + ) + + batch = Batch(max_concurrent=5) + req = batch.create( + contents="Test prompt", + context="observe.describe.frame", + json_output=True, + ) + batch.add(req) + + results = [] + async for completed_req in batch.drain_batch(): + results.append(completed_req) + + assert len(results) == 1 + assert results[0].response is None + assert results[0].reason_code == NON_RESPONSIVE_REASON_CODE diff --git a/tests/test_describe_capacity_retry.py b/tests/test_describe_capacity_retry.py index 72e493b80..a395f2f5c 100644 --- a/tests/test_describe_capacity_retry.py +++ b/tests/test_describe_capacity_retry.py @@ -7,11 +7,16 @@ import io import json from pathlib import Path from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest from PIL import Image from solstone.observe import describe as describe_module +from solstone.think.responsiveness import ( + NON_RESPONSIVE_OUTPUT_MESSAGE, + NON_RESPONSIVE_REASON_CODE, +) def _video_path(tmp_path: Path) -> Path: @@ -165,6 +170,50 @@ def _install_fakes(monkeypatch, *, mode: str) -> None: monkeypatch.setattr(describe_module, "callosum_send", lambda *args, **kwargs: True) +def _install_real_batch_provider(monkeypatch, outcomes: list[dict]) -> SimpleNamespace: + import solstone.think.providers as providers_package + from solstone.think import batch as batch_module + from solstone.think import models + + provider_module = SimpleNamespace( + run_agenerate=AsyncMock(side_effect=outcomes), + ) + monkeypatch.setattr(models, "resolve_provider", lambda _interface: ("fake", "m")) + monkeypatch.setattr( + batch_module, "resolve_provider", lambda _interface: ("fake", "m") + ) + monkeypatch.setattr( + providers_package, + "get_provider_module", + lambda _provider: provider_module, + ) + monkeypatch.setattr(describe_module, "callosum_send", lambda *args, **kwargs: True) + return provider_module + + +def _non_responsive_result() -> dict: + return { + "text": "I cannot describe this screen.", + "model": "provider-model", + "finish_reason": "stop", + } + + +def _describe_result(primary: str) -> dict: + return { + "text": json.dumps( + { + "visual_description": "A code editor is open.", + "primary": primary, + "secondary": "none", + "overlap": True, + } + ), + "model": "provider-model", + "finish_reason": "stop", + } + + @pytest.mark.asyncio @pytest.mark.parametrize("mode", ["phase1", "phase3"]) async def test_capacity_class_describe_errors_retry_and_promote_output( @@ -232,6 +281,111 @@ async def test_capacity_class_exhausted_frame_with_successful_sibling_marks_fail _assert_no_describe_temp(output_path.parent) +@pytest.mark.asyncio +async def test_describe_non_responsive_phase1_does_not_retry(tmp_path, monkeypatch): + video_path = _video_path(tmp_path) + output_path = video_path.with_suffix(".jsonl") + frame_bytes = _png_bytes() + processor = _processor( + video_path, [_frame(1, frame_bytes), _frame(2, frame_bytes)], monkeypatch + ) + provider_module = _install_real_batch_provider( + monkeypatch, + [_non_responsive_result(), _describe_result("code")], + ) + monkeypatch.setattr( + describe_module, + "select_frames_for_extraction", + lambda *_args, **_kwargs: [], + ) + + await processor.process_with_vision( + max_concurrent=1, + output_path=output_path, + work_key="20250101/143022_300/screen", + ) + + assert provider_module.run_agenerate.await_count == 2 + + +@pytest.mark.asyncio +async def test_describe_non_responsive_single_frame_decline_preserves_other_frames( + tmp_path, monkeypatch +): + from solstone.convey.provider_readiness import is_blocking_reason + + video_path = _video_path(tmp_path) + output_path = video_path.with_suffix(".jsonl") + frame_bytes = _png_bytes() + processor = _processor( + video_path, [_frame(1, frame_bytes), _frame(2, frame_bytes)], monkeypatch + ) + _install_real_batch_provider( + monkeypatch, + [_non_responsive_result(), _describe_result("code")], + ) + monkeypatch.setattr( + describe_module, + "select_frames_for_extraction", + lambda *_args, **_kwargs: [], + ) + + assert is_blocking_reason(NON_RESPONSIVE_REASON_CODE) is False + + await processor.process_with_vision( + max_concurrent=1, + output_path=output_path, + work_key="20250101/143022_300/screen", + ) + + rows = _jsonl_rows(output_path) + assert rows[0]["_solstone_processing"]["state"] == "failed" + assert rows[0]["_solstone_processing"]["reason_code"] == "analysis_failed" + declined = next(row for row in rows[1:] if row["frame_id"] == 1) + sibling = next(row for row in rows[1:] if row["frame_id"] == 2) + assert NON_RESPONSIVE_OUTPUT_MESSAGE in declined["error"] + assert sibling["analysis"]["primary"] == "code" + assert "error" not in sibling + assert output_path.exists() + _assert_no_describe_temp(output_path.parent) + + +@pytest.mark.asyncio +async def test_describe_non_responsive_phase3_does_not_retry_and_does_not_block( + tmp_path, monkeypatch +): + video_path = _video_path(tmp_path) + output_path = video_path.with_suffix(".jsonl") + frame_bytes = _png_bytes() + processor = _processor(video_path, [_frame(1, frame_bytes)], monkeypatch) + provider_module = _install_real_batch_provider( + monkeypatch, + [_describe_result("code"), _non_responsive_result()], + ) + monkeypatch.setattr( + describe_module, + "select_frames_for_extraction", + lambda *_args, **_kwargs: [1], + ) + + await processor.process_with_vision( + max_concurrent=1, + output_path=output_path, + work_key="20250101/143022_300/screen", + ) + + rows = _jsonl_rows(output_path) + assert provider_module.run_agenerate.await_count == 2 + assert rows[0]["_solstone_processing"]["state"] == "failed" + row = rows[1] + assert row["frame_id"] == 1 + assert row["enhanced"] is True + assert row["content"] == {} + assert NON_RESPONSIVE_OUTPUT_MESSAGE in row["error"] + assert "retries" not in row["requests"][-1] + _assert_no_describe_temp(output_path.parent) + + @pytest.mark.asyncio async def test_phase3_extraction_error_demotes_record_but_ships_row( tmp_path, monkeypatch