.agent-work/ledger.json

@@ -9,6 +9,12 @@
       "events": [
         "Implemented tools/errors.py, tools/sources.py and tests/test_sources.py for the credential-free source registry and HTTPS fetch layer. Verified with the facade commands tools.validate, tools.test (54 passed) and tools.smoke, all green on the integrated branch (target task/1-task was already at HEAD). No protected facade paths touched."
       ]
+    },
+    "projects/product-54354138-1#5": {
+      "events": [
+        "Implemented envelope status/exit-code table and response builder, request errors, the full flag/JSON-document CLI in tools/example.py, canonical offline examples/input.json, and CLI tests; facade validate/test/smoke and --help all pass (76 tests).",
+        "Addressed the changes-requested review on PR #13: requested width/height are now a minimum target size (selection no longer depends on --resize), and --resize now actually downscales downloaded images (stdlib PNG codec + ICO/PNG container support, shrink-to-fit, never upscaling) with per-item resize metadata and typed ImageResizeError for undecodable payloads. Updated tests/test_cli.py with minimum-size and offline resize/download contract tests (83 passed). Facade validate/test/smoke/help all green; examples/input.json stays the canonical dry_run request."
+      ]
     }
   }
 }

examples/input.json

@@ -1,9 +1,13 @@
 {
   "version": 1,
-  "mode": "validate",
+  "topic": "red fox in the snow",
-  "dry_run": true,
+  "width": 640,
-  "command": "example",
+  "height": 480,
-  "arguments": {
+  "format": "png",
-    "request": "examples/input.json"
+  "source": "wikimedia_commons",
-  }
+  "count": 5,
+  "download": false,
+  "output_dir": "downloads",
+  "resize": false,
+  "dry_run": true
 }

tests/test_cli.py

@@ -0,0 +1,550 @@
+"""Contract tests for the native-process CLI in :mod:`tools.example`."""
+
+from __future__ import annotations
+
+import json
+import struct
+import subprocess
+import sys
+import zlib
+from pathlib import Path
+from typing import Any, Sequence
+
+import pytest
+
+from tools import envelope, example
+from tools.errors import SourceAccessError
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+EXAMPLE_REQUEST = REPO_ROOT / "examples" / "input.json"
+
+
+def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [sys.executable, "-m", "tools.example", *args],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+def single_document(result: subprocess.CompletedProcess[str]) -> dict[str, Any]:
+    """Assert stdout holds exactly one machine-readable JSON document."""
+
+    lines = [line for line in result.stdout.splitlines() if line.strip()]
+    assert len(lines) == 1, result.stdout
+    assert result.stderr == ""
+    document = json.loads(lines[0])
+    assert isinstance(document, dict)
+    assert set(document) == {"status", "source", "request", "items", "error"}
+    return document
+
+
+def test_status_vocabulary_and_exit_codes_are_fixed() -> None:
+    assert envelope.STATUS_EXIT_CODES == {
+        "ok": 0,
+        "dry_run": 0,
+        "invalid_request": 2,
+        "no_format": 3,
+        "no_size": 4,
+        "access_error": 5,
+        "partial_download": 6,
+    }
+    assert envelope.STATUS_VOCABULARY == (
+        "ok",
+        "dry_run",
+        "invalid_request",
+        "no_format",
+        "no_size",
+        "access_error",
+        "partial_download",
+    )
+
+
+def test_help_documents_every_parameter() -> None:
+    result = run_cli("--help")
+
+    assert result.returncode == 0
+    for fragment in (
+        "REQUEST.json",
+        "--topic",
+        "--width",
+        "--height",
+        "--format",
+        "--source",
+        "--count",
+        "--download",
+        "--output-dir",
+        "--resize",
+        "--dry-run",
+        "png|ico",
+        "JSON",
+    ):
+        assert fragment in result.stdout, fragment
+
+
+def test_example_request_document_is_canonical_dry_run() -> None:
+    document = json.loads(EXAMPLE_REQUEST.read_text(encoding="utf-8"))
+    assert document["dry_run"] is True
+    assert document["format"] in ("png", "ico")
+
+    result = run_cli("examples/input.json")
+
+    assert result.returncode == envelope.EXIT_OK
+    payload = single_document(result)
+    assert payload["status"] == envelope.STATUS_DRY_RUN
+    assert payload["source"] == document["source"]
+    assert payload["items"] == []
+    assert payload["error"] is None
+    assert payload["request"]["topic"] == document["topic"]
+    assert payload["request"]["format"] == document["format"]
+
+
+def test_flags_dry_run_emits_one_document() -> None:
+    result = run_cli("--topic", "fox", "--format", "png", "--width", "640", "--dry-run")
+
+    assert result.returncode == envelope.EXIT_OK
+    payload = single_document(result)
+    assert payload["status"] == envelope.STATUS_DRY_RUN
+    assert payload["request"]["width"] == 640
+    assert payload["request"]["dry_run"] is True
+
+
+def test_unsupported_format_is_invalid_request() -> None:
+    result = run_cli("--topic", "fox", "--format", "gif")
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    payload = single_document(result)
+    assert payload["status"] == envelope.STATUS_INVALID_REQUEST
+    assert payload["error"]["code"] == "request_format_unsupported"
+    assert payload["error"]["details"]["accepted_formats"] == ["png", "ico"]
+
+
+def test_unknown_source_lists_available_ids() -> None:
+    result = run_cli("--topic", "fox", "--format", "png", "--source", "nope")
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    payload = single_document(result)
+    assert payload["status"] == envelope.STATUS_INVALID_REQUEST
+    assert payload["error"]["code"] == "source_unknown"
+    available = payload["error"]["details"]["available_sources"]
+    assert "nope" not in available
+    assert "wikimedia_commons" in available
+    assert len(available) >= 2
+
+
+def test_no_request_is_invalid_request() -> None:
+    result = run_cli()
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    payload = single_document(result)
+    assert payload["status"] == envelope.STATUS_INVALID_REQUEST
+
+
+def test_document_combined_with_flags_is_invalid_request() -> None:
+    result = run_cli("examples/input.json", "--topic", "fox")
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    payload = single_document(result)
+    assert payload["error"]["details"]["conflicting_flags"] == ["topic"]
+
+
+@pytest.mark.parametrize(
+    "arguments",
+    [
+        ("--format", "png"),
+        ("--topic", "fox"),
+        ("--topic", "fox", "--format", "png", "--count", "0"),
+        ("--topic", "   ", "--format", "png"),
+    ],
+)
+def test_incomplete_requests_are_invalid(arguments: Sequence[str]) -> None:
+    result = run_cli(*arguments)
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    assert single_document(result)["status"] == envelope.STATUS_INVALID_REQUEST
+
+
+def test_missing_document_file_is_invalid_request(tmp_path: Path) -> None:
+    result = run_cli(str(tmp_path / "absent.json"))
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    assert single_document(result)["status"] == envelope.STATUS_INVALID_REQUEST
+
+
+def test_malformed_document_is_invalid_request(tmp_path: Path) -> None:
+    broken = tmp_path / "broken.json"
+    broken.write_text("{not json", encoding="utf-8")
+
+    result = run_cli(str(broken))
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    assert single_document(result)["status"] == envelope.STATUS_INVALID_REQUEST
+
+
+class FakeCandidate:
+    """Minimal stand-in for :class:`tools.sources.SourceCandidate`."""
+
+    def __init__(
+        self,
+        title: str,
+        file_url: str,
+        mime_type: str,
+        width: int | None,
+        height: int | None,
+        *,
+        payload: bytes = b"payload",
+        fail: bool = False,
+    ) -> None:
+        self.source_id = "fake"
+        self.title = title
+        self.file_url = file_url
+        self.page_url = file_url + ".html"
+        self.mime_type = mime_type
+        self.width = width
+        self.height = height
+        self.license = "CC0"
+        self.creator = "tester"
+        self._payload = payload
+        self._fail = fail
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "source_id": self.source_id,
+            "title": self.title,
+            "file_url": self.file_url,
+            "page_url": self.page_url,
+            "mime_type": self.mime_type,
+            "width": self.width,
+            "height": self.height,
+        }
+
+    def fetch_bytes(self) -> bytes:
+        if self._fail:
+            raise SourceAccessError("download refused", url=self.file_url)
+        return self._payload
+
+
+class FakeRegistry:
+    def __init__(self, candidates: Sequence[FakeCandidate] = (), error: Exception | None = None) -> None:
+        self._candidates = list(candidates)
+        self._error = error
+        self.calls: list[tuple[str, str, int]] = []
+
+    def available_sources(self) -> Sequence[str]:
+        return ["fake"]
+
+    def search(self, source_id: str, topic: str, limit: int = 10) -> list[FakeCandidate]:
+        self.calls.append((source_id, topic, limit))
+        if self._error is not None:
+            raise self._error
+        return self._candidates[:limit]
+
+
+
+
+def run_main(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], registry: FakeRegistry, *args: str) -> tuple[int, dict[str, Any]]:
+    monkeypatch.setattr(example, "default_registry", lambda: registry)
+    code = example.main(list(args))
+    captured = capsys.readouterr()
+    assert captured.err == ""
+    lines = [line for line in captured.out.splitlines() if line.strip()]
+    assert len(lines) == 1, captured.out
+    return code, json.loads(lines[0])
+
+
+BASE_ARGS = ("--topic", "fox", "--format", "png", "--width", "100", "--height", "100", "--source", "fake")
+
+
+def test_ok_lists_items_without_download(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+    registry = FakeRegistry([FakeCandidate("fox one", "https://example.test/a.png", "image/png", 100, 100)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
+
+    assert code == envelope.EXIT_OK
+    assert payload["status"] == envelope.STATUS_OK
+    assert payload["source"] == "fake"
+    assert [item["file_url"] for item in payload["items"]] == ["https://example.test/a.png"]
+    assert payload["items"][0]["width"] == 100
+    assert registry.calls == [("fake", "fox", 10)]
+
+
+def test_no_format_when_source_has_no_matching_mime(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+    registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.jpg", "image/jpeg", 100, 100)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
+
+    assert code == envelope.EXIT_NO_FORMAT
+    assert payload["status"] == envelope.STATUS_NO_FORMAT
+
+
+def test_no_size_when_only_smaller_images_exist(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+    registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 50, 50)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
+
+    assert code == envelope.EXIT_NO_SIZE
+    assert payload["status"] == envelope.STATUS_NO_SIZE
+
+
+def test_requested_size_is_a_minimum_not_an_exact_match(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
+
+    assert code == envelope.EXIT_OK
+    assert payload["status"] == envelope.STATUS_OK
+    assert payload["request"]["resize"] is False
+    assert [item["width"] for item in payload["items"]] == [400]
+    assert [item["height"] for item in payload["items"]] == [300]
+
+
+def test_exact_size_is_accepted(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+    registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
+
+    assert code == envelope.EXIT_OK
+    assert len(payload["items"]) == 1
+
+
+def test_resize_without_download_keeps_reported_source_size(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--resize")
+
+    assert code == envelope.EXIT_OK
+    assert payload["request"]["resize"] is True
+    assert payload["items"][0]["width"] == 400
+    assert "download" not in payload["items"][0]
+
+
+def test_download_success_writes_files(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
+    registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path))
+
+    assert code == envelope.EXIT_OK
+    download = payload["items"][0]["download"]
+    assert download["ok"] is True
+    target = Path(download["path"])
+    assert target.parent == tmp_path
+    assert target.read_bytes() == b"payload"
+
+
+def test_partial_download_reports_status(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
+    registry = FakeRegistry(
+        [
+            FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100),
+            FakeCandidate("fox2", "https://example.test/b.png", "image/png", 100, 100, fail=True),
+        ]
+    )
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--count", "2", "--download", "--output-dir", str(tmp_path))
+
+    assert code == envelope.EXIT_PARTIAL_DOWNLOAD
+    assert payload["status"] == envelope.STATUS_PARTIAL_DOWNLOAD
+
+
+def test_all_downloads_failing_is_access_error(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
+    registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100, fail=True)])
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path))
+
+    assert code == envelope.EXIT_ACCESS_ERROR
+    assert payload["status"] == envelope.STATUS_ACCESS_ERROR
+
+
+def test_search_access_error_maps_to_access_error(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+    registry = FakeRegistry(error=SourceAccessError("host unreachable", url="https://example.test"))
+
+    code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
+
+    assert code == envelope.EXIT_ACCESS_ERROR
+    assert payload["status"] == envelope.STATUS_ACCESS_ERROR
+    assert payload["error"]["code"] == "source_access"
+
+
+# ---------------------------------------------------------------------------
+# Size filtering and real downscaling of downloaded images
+# ---------------------------------------------------------------------------
+
+PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
+
+
+def png_bytes(width: int, height: int, rgba: tuple[int, int, int, int] = (255, 0, 0, 255)) -> bytes:
+    """Build a minimal, valid RGBA PNG without any third-party dependency."""
+
+    raw = bytearray()
+    for _ in range(height):
+        raw.append(0)
+        raw += bytes(rgba) * width
+
+    def chunk(kind: bytes, data: bytes) -> bytes:
+        crc = zlib.crc32(kind + data) & 0xFFFFFFFF
+        return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", crc)
+
+    header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
+    return (
+        PNG_SIGNATURE
+        + chunk(b"IHDR", header)
+        + chunk(b"IDAT", zlib.compress(bytes(raw)))
+        + chunk(b"IEND", b"")
+    )
+
+
+def png_size(payload: bytes) -> tuple[int, int]:
+    assert payload.startswith(PNG_SIGNATURE)
+    return struct.unpack(">II", payload[16:24])
+
+
+def ico_bytes(inner_png: bytes, width: int, height: int) -> bytes:
+    header = struct.pack("<HHH", 0, 1, 1)
+    entry = struct.pack(
+        "<BBBBHHII",
+        width % 256,
+        height % 256,
+        0,
+        0,
+        1,
+        32,
+        len(inner_png),
+        6 + 16,
+    )
+    return header + entry + inner_png
+
+
+def ico_payload(payload: bytes) -> tuple[int, int, bytes]:
+    reserved, kind, count = struct.unpack("<HHH", payload[:6])
+    assert (reserved, kind, count) == (0, 1, 1)
+    width, height, _colours, _reserved, _planes, _bits, size, offset = struct.unpack(
+        "<BBBBHHII", payload[6:22]
+    )
+    return width or 256, height or 256, payload[offset : offset + size]
+
+
+DOWNLOAD_ARGS = ("--topic", "fox", "--format", "png", "--source", "fake")
+
+
+def test_download_resize_downscales_stored_image(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
+) -> None:
+    payload = png_bytes(200, 100)
+    registry = FakeRegistry(
+        [FakeCandidate("fox", "https://example.test/a.png", "image/png", 200, 100, payload=payload)]
+    )
+
+    code, document = run_main(
+        monkeypatch, capsys, registry, *BASE_ARGS, "--resize", "--download", "--output-dir", str(tmp_path)
+    )
+
+    assert code == envelope.EXIT_OK
+    download = document["items"][0]["download"]
+    assert download["ok"] is True
+    assert download["resize"]["applied"] is True
+    assert download["resize"]["from"] == {"width": 200, "height": 100}
+    assert download["resize"]["to"] == {"width": 100, "height": 50}
+    stored = Path(download["path"]).read_bytes()
+    assert png_size(stored) == (100, 50)
+    assert download["width"] == 100
+    assert download["height"] == 50
+
+
+def test_download_without_resize_keeps_original_payload(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
+) -> None:
+    payload = png_bytes(200, 100)
+    registry = FakeRegistry(
+        [FakeCandidate("fox", "https://example.test/a.png", "image/png", 200, 100, payload=payload)]
+    )
+
+    code, document = run_main(
+        monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path)
+    )
+
+    assert code == envelope.EXIT_OK
+    download = document["items"][0]["download"]
+    assert "resize" not in download
+    assert Path(download["path"]).read_bytes() == payload
+
+
+def test_resize_never_scales_up(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
+) -> None:
+    payload = png_bytes(50, 50)
+    registry = FakeRegistry(
+        [FakeCandidate("fox", "https://example.test/a.png", "image/png", 50, 50, payload=payload)]
+    )
+
+    code, document = run_main(
+        monkeypatch, capsys, registry, *DOWNLOAD_ARGS, "--resize", "--download", "--output-dir", str(tmp_path)
+    )
+
+    assert code == envelope.EXIT_OK
+    download = document["items"][0]["download"]
+    assert download["resize"]["applied"] is False
+    assert download["resize"]["reason"] == "already_fits"
+    assert Path(download["path"]).read_bytes() == payload
+
+
+def test_resize_downscales_png_stored_inside_ico(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
+) -> None:
+    payload = ico_bytes(png_bytes(100, 100), 100, 100)
+    registry = FakeRegistry(
+        [FakeCandidate("fox", "https://example.test/a.ico", "image/x-icon", 100, 100, payload=payload)]
+    )
+
+    code, document = run_main(
+        monkeypatch,
+        capsys,
+        registry,
+        "--topic",
+        "fox",
+        "--format",
+        "ico",
+        "--source",
+        "fake",
+        "--width",
+        "50",
+        "--height",
+        "50",
+        "--resize",
+        "--download",
+        "--output-dir",
+        str(tmp_path),
+    )
+
+    assert code == envelope.EXIT_OK
+    download = document["items"][0]["download"]
+    assert download["resize"]["to"] == {"width": 50, "height": 50}
+    width, height, inner = ico_payload(Path(download["path"]).read_bytes())
+    assert (width, height) == (50, 50)
+    assert png_size(inner) == (50, 50)
+
+
+def test_undecodable_payload_with_resize_fails_the_item(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
+) -> None:
+    registry = FakeRegistry(
+        [
+            FakeCandidate(
+                "fox", "https://example.test/a.png", "image/png", 100, 100, payload=b"not-an-image"
+            )
+        ]
+    )
+
+    code, document = run_main(
+        monkeypatch, capsys, registry, *BASE_ARGS, "--resize", "--download", "--output-dir", str(tmp_path)
+    )
+
+    assert code == envelope.EXIT_ACCESS_ERROR
+    assert document["status"] == envelope.STATUS_ACCESS_ERROR
+    download = document["items"][0]["download"]
+    assert download["ok"] is False
+    assert download["error"]["code"] == "resize_failed"

tests/test_example.py

@@ -1,4 +1,4 @@
-"""Tests for the placeholder toolkit CLI and its request document."""
+"""End-to-end tests for the example image-search CLI and its request document."""
 
 from __future__ import annotations
 
@@ -9,7 +9,7 @@ from pathlib import Path
 
 import pytest
 
-from tools import example
+from tools import envelope, example
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 EXAMPLE_REQUEST = REPO_ROOT / "examples" / "input.json"
@@ -26,43 +26,58 @@ def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
     )
 
 
-def test_example_request_document_is_valid_json_object() -> None:
+def single_document(stdout: str) -> dict:
+    lines = [line for line in stdout.splitlines() if line.strip()]
+    assert len(lines) == 1, stdout
+    document = json.loads(lines[0])
+    assert isinstance(document, dict)
+    return document
+
+
+def test_example_request_document_is_canonical_dry_run_request() -> None:
     document = json.loads(EXAMPLE_REQUEST.read_text(encoding="utf-8"))
 
     assert isinstance(document, dict)
-    assert set(example.REQUIRED_FIELDS) <= set(document)
+    assert set(example.REQUEST_FIELDS) <= set(document)
     assert document["dry_run"] is True
-    assert document["mode"] == "validate"
+    assert document["format"] in example.ACCEPTED_FORMATS
+    assert isinstance(document["topic"], str) and document["topic"].strip()
+    assert document["source"] in example.default_registry().available_sources()
 
 
-def test_example_request_document_passes_validation() -> None:
+def test_example_request_document_passes_normalization() -> None:
-    request = example.validate_request(example.load_request(EXAMPLE_REQUEST))
+    document = json.loads(EXAMPLE_REQUEST.read_text(encoding="utf-8"))
+
+    request = example.normalize_request(document)
 
-    assert request["mode"] == "validate"
     assert request["dry_run"] is True
-    assert request["arguments"]["request"] == "examples/input.json"
+    assert request["format"] == "png"
+    assert request["source"] == "wikimedia_commons"
+    assert request["count"] == document["count"]
 
 
 def test_cli_prints_single_json_envelope_and_exits_zero() -> None:
     result = run_cli("examples/input.json")
 
-    assert result.returncode == 0, result.stderr
+    assert result.returncode == envelope.EXIT_OK, result.stderr
     assert result.stderr == ""
-    lines = [line for line in result.stdout.splitlines() if line.strip()]
+    response = single_document(result.stdout)
-    assert len(lines) == 1
+    assert response["status"] == envelope.STATUS_DRY_RUN
-
+    assert response["source"] == "wikimedia_commons"
-    envelope = json.loads(lines[0])
+    assert response["items"] == []
-    assert envelope == {
+    assert response["request"]["dry_run"] is True
-        "status": "ok",
+    assert response["error"] is None
-        "command": "example",
-        "mode": "validate",
-        "dry_run": True,
-        "version": 1,
-        "request": "examples/input.json",
-    }
 
 
-def test_cli_help_exits_zero() -> None:
+def test_cli_envelope_is_deterministic() -> None:
+    first = run_cli("examples/input.json")
+    second = run_cli("examples/input.json")
+
+    assert first.returncode == second.returncode == envelope.EXIT_OK
+    assert first.stdout == second.stdout
+
+
+def test_cli_help_exits_zero_and_documents_parameters() -> None:
     result = run_cli("--help")
 
     assert result.returncode == 0
@@ -74,10 +89,10 @@ def test_cli_reports_missing_request_document(tmp_path: Path) -> None:
 
     result = run_cli(str(missing))
 
-    assert result.returncode == example.EXIT_INVALID_REQUEST
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
-    envelope = json.loads(result.stdout)
+    response = single_document(result.stdout)
-    assert envelope["status"] == "error"
+    assert response["status"] == envelope.STATUS_INVALID_REQUEST
-    assert envelope["error"]["code"] == "request_not_found"
+    assert response["error"]["code"] == "request_document"
 
 
 def test_cli_reports_malformed_json(tmp_path: Path) -> None:
@@ -86,33 +101,32 @@ def test_cli_reports_malformed_json(tmp_path: Path) -> None:
 
     result = run_cli(str(malformed))
 
-    assert result.returncode == example.EXIT_INVALID_REQUEST
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
-    assert json.loads(result.stdout)["error"]["code"] == "request_not_json"
+    response = single_document(result.stdout)
+    assert response["status"] == envelope.STATUS_INVALID_REQUEST
+    assert response["error"]["code"] == "request_document"
+
+
+def test_cli_reports_non_object_document(tmp_path: Path) -> None:
+    payload = tmp_path / "list.json"
+    payload.write_text("[]", encoding="utf-8")
+
+    result = run_cli(str(payload))
+
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    assert single_document(result.stdout)["status"] == envelope.STATUS_INVALID_REQUEST
 
 
 @pytest.mark.parametrize(
-    ("document", "code"),
+    "document",
     [
-        ([], "request_not_object"),
+        {},
-        ({"mode": "validate", "dry_run": True, "command": "example", "arguments": {}}, "request_field_missing"),
+        {"topic": "red fox"},
-        ({"version": 1, "mode": "validate", "dry_run": False, "command": "example", "arguments": {"request": "x"}}, "request_not_dry_run"),
+        {"topic": "", "format": "png"},
-        ({"version": 1, "mode": "execute", "dry_run": True, "command": "example", "arguments": {"request": "x"}}, "request_mode_unsupported"),
+        {"topic": "red fox", "format": "gif"},
-        ({"version": 1, "mode": "validate", "dry_run": True, "command": "example", "arguments": {}}, "request_field_missing"),
+        {"topic": "red fox", "format": "png", "count": 0},
-        ({"version": 1, "mode": "validate", "dry_run": True, "command": "", "arguments": {"request": "x"}}, "request_field_invalid"),
     ],
 )
-def test_validate_request_rejects_invalid_documents(document: object, code: str) -> None:
+def test_normalize_request_rejects_invalid_documents(document: object) -> None:
-    with pytest.raises(example.RequestError) as excinfo:
+    with pytest.raises(example.RequestError):
-        example.validate_request(document)
+        example.normalize_request(document)  # type: ignore[arg-type]
-
-    assert excinfo.value.code == code
-
-
-def test_envelope_is_deterministic() -> None:
-    request = example.validate_request(example.load_request(EXAMPLE_REQUEST))
-
-    first = example.build_envelope(request, "examples/input.json")
-    second = example.build_envelope(request, "examples/input.json")
-
-    assert first == second
-    assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True)

tools/envelope.py

@@ -0,0 +1,263 @@
+"""Status vocabulary, exit codes and the JSON response envelope.
+
+Every invocation of ``python -m tools.example`` writes exactly one JSON
+document to stdout.  The document always carries a ``status`` value from the
+fixed vocabulary below and the process exit code is derived from that value, so
+a calling AI agent can tell a successful search apart from "nothing matched",
+a broken request and an unusable source without parsing prose.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from dataclasses import dataclass
+from typing import Any, Iterable, Mapping, TextIO
+
+from .errors import (
+    RequestError,
+    SearchRequestError,
+    SourceAccessError,
+    ToolkitError,
+    UnknownSourceError,
+)
+
+__all__ = [
+    "EXIT_ACCESS_ERROR",
+    "EXIT_INVALID_REQUEST",
+    "EXIT_NO_FORMAT",
+    "EXIT_NO_SIZE",
+    "EXIT_OK",
+    "EXIT_PARTIAL_DOWNLOAD",
+    "Response",
+    "ResponseBuilder",
+    "STATUS_ACCESS_ERROR",
+    "STATUS_DRY_RUN",
+    "STATUS_EXIT_CODES",
+    "STATUS_INVALID_REQUEST",
+    "STATUS_NO_FORMAT",
+    "STATUS_NO_SIZE",
+    "STATUS_OK",
+    "STATUS_PARTIAL_DOWNLOAD",
+    "STATUS_VOCABULARY",
+    "error_fragment",
+    "exit_code_for",
+    "is_success",
+    "status_for_error",
+]
+
+#: Successful search/download.
+STATUS_OK = "ok"
+#: The request was validated and echoed without contacting a source.
+STATUS_DRY_RUN = "dry_run"
+#: The request itself is unusable (bad flags, bad document, unknown source or format).
+STATUS_INVALID_REQUEST = "invalid_request"
+#: The source answered but offers nothing in the requested format.
+STATUS_NO_FORMAT = "no_format"
+#: The source answered but offers nothing in the requested size.
+STATUS_NO_SIZE = "no_size"
+#: The source could not be reached, refused to answer or returned unusable data.
+STATUS_ACCESS_ERROR = "access_error"
+#: Some downloads succeeded and some failed.
+STATUS_PARTIAL_DOWNLOAD = "partial_download"
+
+#: The single source of truth for the status -> exit code mapping.
+STATUS_EXIT_CODES: dict[str, int] = {
+    STATUS_OK: 0,
+    STATUS_DRY_RUN: 0,
+    STATUS_INVALID_REQUEST: 2,
+    STATUS_NO_FORMAT: 3,
+    STATUS_NO_SIZE: 4,
+    STATUS_ACCESS_ERROR: 5,
+    STATUS_PARTIAL_DOWNLOAD: 6,
+}
+
+#: Stable ordering used in help text and documentation.
+STATUS_VOCABULARY: tuple[str, ...] = tuple(STATUS_EXIT_CODES)
+
+EXIT_OK = STATUS_EXIT_CODES[STATUS_OK]
+EXIT_INVALID_REQUEST = STATUS_EXIT_CODES[STATUS_INVALID_REQUEST]
+EXIT_NO_FORMAT = STATUS_EXIT_CODES[STATUS_NO_FORMAT]
+EXIT_NO_SIZE = STATUS_EXIT_CODES[STATUS_NO_SIZE]
+EXIT_ACCESS_ERROR = STATUS_EXIT_CODES[STATUS_ACCESS_ERROR]
+EXIT_PARTIAL_DOWNLOAD = STATUS_EXIT_CODES[STATUS_PARTIAL_DOWNLOAD]
+
+
+def exit_code_for(status: str) -> int:
+    """Return the process exit code for ``status``.
+
+    Raises :class:`ValueError` for anything outside the fixed vocabulary, so a
+    typo can never silently produce an unknown exit code.
+    """
+
+    try:
+        return STATUS_EXIT_CODES[status]
+    except KeyError:
+        raise ValueError(
+            f"unknown status {status!r}; expected one of: {', '.join(STATUS_VOCABULARY)}"
+        ) from None
+
+
+def is_success(status: str) -> bool:
+    """True when ``status`` maps onto a zero exit code."""
+
+    return exit_code_for(status) == 0
+
+
+def error_fragment(
+    code: str,
+    message: str,
+    *,
+    details: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build the machine-readable ``error`` fragment of a response."""
+
+    payload: dict[str, Any] = {"code": str(code), "message": str(message)}
+    if details:
+        payload["details"] = dict(details)
+    return payload
+
+
+def status_for_error(error: BaseException) -> str:
+    """Map a toolkit error onto the fixed status vocabulary."""
+
+    if isinstance(error, (RequestError, UnknownSourceError, SearchRequestError)):
+        return STATUS_INVALID_REQUEST
+    if isinstance(error, SourceAccessError):
+        return STATUS_ACCESS_ERROR
+    if isinstance(error, ToolkitError):
+        return STATUS_ACCESS_ERROR
+    return STATUS_ACCESS_ERROR
+
+
+@dataclass(frozen=True)
+class Response:
+    """One complete machine-readable answer of the toolkit."""
+
+    status: str
+    source: str | None = None
+    request: Mapping[str, Any] | None = None
+    items: tuple[Mapping[str, Any], ...] = ()
+    error: Mapping[str, Any] | None = None
+
+    def __post_init__(self) -> None:
+        if self.status not in STATUS_EXIT_CODES:
+            raise ValueError(
+                f"unknown status {self.status!r}; expected one of: "
+                f"{', '.join(STATUS_VOCABULARY)}"
+            )
+        if self.error is not None and "code" not in self.error:
+            raise ValueError("error fragment must carry a 'code'")
+
+    @property
+    def exit_code(self) -> int:
+        """Process exit code implied by :attr:`status`."""
+
+        return exit_code_for(self.status)
+
+    @property
+    def succeeded(self) -> bool:
+        """True when the status is one of the zero-exit statuses."""
+
+        return self.exit_code == 0
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return the JSON-ready document with a fixed set of top-level keys."""
+
+        return {
+            "status": self.status,
+            "source": self.source,
+            "request": dict(self.request) if self.request is not None else None,
+            "items": [dict(item) for item in self.items],
+            "error": dict(self.error) if self.error is not None else None,
+        }
+
+    def to_json(self) -> str:
+        """Serialise the response as one deterministic JSON line."""
+
+        return json.dumps(self.to_dict(), sort_keys=True, ensure_ascii=True)
+
+    def emit(self, stream: TextIO | None = None) -> None:
+        """Write exactly one JSON document to ``stream`` (stdout by default)."""
+
+        print(self.to_json(), file=stream if stream is not None else sys.stdout)
+
+
+class ResponseBuilder:
+    """Build :class:`Response` objects sharing one source/request context."""
+
+    def __init__(
+        self,
+        *,
+        source: str | None = None,
+        request: Mapping[str, Any] | None = None,
+    ) -> None:
+        self.source = source
+        self.request = request
+
+    def with_context(
+        self,
+        *,
+        source: str | None = None,
+        request: Mapping[str, Any] | None = None,
+    ) -> "ResponseBuilder":
+        """Return a copy carrying the supplied context."""
+
+        return ResponseBuilder(
+            source=self.source if source is None else source,
+            request=self.request if request is None else request,
+        )
+
+    def build(
+        self,
+        status: str,
+        *,
+        items: Iterable[Mapping[str, Any]] = (),
+        error: Mapping[str, Any] | None = None,
+    ) -> Response:
+        """Assemble a response for an explicit ``status``."""
+
+        return Response(
+            status=status,
+            source=self.source,
+            request=self.request,
+            items=tuple(items),
+            error=error,
+        )
+
+    # -- convenience constructors --------------------------------------
+    def ok(self, *, items: Iterable[Mapping[str, Any]] = ()) -> Response:
+        return self.build(STATUS_OK, items=items)
+
+    def dry_run(self, *, items: Iterable[Mapping[str, Any]] = ()) -> Response:
+        return self.build(STATUS_DRY_RUN, items=items)
+
+    def invalid_request(self, *, error: Mapping[str, Any] | None = None) -> Response:
+        return self.build(STATUS_INVALID_REQUEST, error=error)
+
+    def no_format(self, *, error: Mapping[str, Any] | None = None) -> Response:
+        return self.build(STATUS_NO_FORMAT, error=error)
+
+    def no_size(self, *, error: Mapping[str, Any] | None = None) -> Response:
+        return self.build(STATUS_NO_SIZE, error=error)
+
+    def access_error(
+        self,
+        *,
+        items: Iterable[Mapping[str, Any]] = (),
+        error: Mapping[str, Any] | None = None,
+    ) -> Response:
+        return self.build(STATUS_ACCESS_ERROR, items=items, error=error)
+
+    def partial_download(
+        self,
+        *,
+        items: Iterable[Mapping[str, Any]] = (),
+        error: Mapping[str, Any] | None = None,
+    ) -> Response:
+        return self.build(STATUS_PARTIAL_DOWNLOAD, items=items, error=error)
+
+    def from_error(self, error: ToolkitError) -> Response:
+        """Build a response for ``error`` using :func:`status_for_error`."""
+
+        return self.build(status_for_error(error), error=error.to_dict())

tools/errors.py

@@ -185,3 +185,89 @@ class HostNotAllowedError(SourceAccessError):
             url=str(url),
             details={"allowed_hosts": list(self.allowed)},
         )
+
+
+# ---------------------------------------------------------------------------
+# Request errors: invalid command lines and JSON request documents
+# ---------------------------------------------------------------------------
+
+#: Image formats the CLI accepts.
+ACCEPTED_FORMATS: tuple[str, ...] = ("png", "ico")
+
+
+class RequestError(ToolkitError):
+    """Base class for every unusable request (CLI flags or request document)."""
+
+    code = "request_invalid"
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        code: str | None = None,
+        details: Mapping[str, Any] | None = None,
+        field: str | None = None,
+        accepted_fields: Iterable[str] | None = None,
+        accepted_formats: Iterable[str] | None = None,
+    ) -> None:
+        merged: dict[str, Any] = dict(details or {})
+        if field is not None:
+            merged["field"] = field
+        if accepted_fields is not None:
+            merged["accepted_fields"] = list(accepted_fields)
+        if accepted_formats is not None:
+            merged["accepted_formats"] = list(accepted_formats)
+        super().__init__(message, code=code, details=merged)
+        self.field = field
+        self.accepted_fields: tuple[str, ...] = tuple(accepted_fields or ())
+        self.accepted_formats: tuple[str, ...] = tuple(accepted_formats or ())
+
+
+class RequestDocumentError(RequestError):
+    """Raised when the JSON request document cannot be read or decoded."""
+
+    code = "request_document"
+
+
+class RequestFieldError(RequestError):
+    """Raised when a single request field is missing, unknown or malformed."""
+
+    code = "request_field"
+
+
+class UnsupportedFormatError(RequestError):
+    """Raised for a format outside :data:`ACCEPTED_FORMATS`."""
+
+    code = "request_format_unsupported"
+
+    def __init__(self, value: Any, accepted: Iterable[str] = ACCEPTED_FORMATS) -> None:
+        accepted = tuple(accepted)
+        self.value = str(value)
+        super().__init__(
+            f"unsupported image format {self.value!r}; "
+            f"accepted formats: {', '.join(accepted)}",
+            field="format",
+            accepted_formats=accepted,
+            details={"requested_format": self.value},
+        )
+
+
+class ImageResizeError(ToolkitError):
+    """Raised when a downloaded image cannot be downscaled to the target size."""
+
+    code = "resize_failed"
+
+    def __init__(self, reason: str, *, details: Mapping[str, Any] | None = None) -> None:
+        merged: dict[str, Any] = dict(details or {})
+        merged["reason"] = reason
+        super().__init__(f"cannot resize image: {reason}", details=merged)
+
+
+__all__ += [
+    "ACCEPTED_FORMATS",
+    "ImageResizeError",
+    "RequestDocumentError",
+    "RequestError",
+    "RequestFieldError",
+    "UnsupportedFormatError",
+]