feat: Introduce output verification framework, enhance writer error handling, and add cacheable annotation for GET methods.

Author: tercel

CHANGELOG.md

@@ -2,6 +2,30 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.3.1] - 2026-03-12
+
+### Fixed
+
+- README — added 5 concrete verifier classes (`YAMLVerifier`, `SyntaxVerifier`,
+  `RegistryVerifier`, `MagicBytesVerifier`, `JSONVerifier`) to Core Modules table
+  for documentation completeness
+
+---
+
+## [0.3.0] - 2026-03-12
+
+### Changed
+
+- **apcore >= 0.13.0** — Upgraded minimum dependency to support new
+  `ModuleAnnotations` caching and pagination fields:
+  `cacheable`, `cache_ttl`, `cache_key_fields`, `paginated`, `pagination_style`
+- `infer_annotations_from_method()` — `GET` now also infers `cacheable=True`
+- `AIEnhancer` — Annotation inference prompt and acceptance logic extended
+  to handle all 11 annotation fields (5 new: `cacheable`, `cache_ttl`,
+  `cache_key_fields`, `paginated`, `pagination_style`)
+
+---
+
 ## [0.2.0] - 2026-03-11
 
 ### Added

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "apcore-toolkit"
-version = "0.2.0"
+version = "0.3.1"
 description = "Shared scanner, schema extraction, and output toolkit for apcore framework adapters"
 requires-python = ">=3.11"
 readme = "README.md"
@@ -23,7 +23,7 @@ keywords = [
     "toolkit",
 ]
 dependencies = [
-    "apcore>=0.9.0",
+    "apcore>=0.13.0",
     "pydantic>=2.0",
     "PyYAML>=6.0",
 ]

src/apcore_toolkit/__init__.py

@@ -5,6 +5,7 @@
 
 from apcore_toolkit.ai_enhancer import AIEnhancer
 from apcore_toolkit.formatting import to_markdown
+from apcore_toolkit.output import get_writer
 from apcore_toolkit.output.python_writer import PythonWriter
 from apcore_toolkit.output.registry_writer import RegistryWriter
 from apcore_toolkit.output.types import WriteResult
@@ -14,7 +15,7 @@
 from apcore_toolkit.schema_utils import enrich_schema_descriptions
 from apcore_toolkit.types import ScannedModule
 
-__version__ = "0.2.0"
+__version__ = "0.3.0"
 
 __all__ = [
     "AIEnhancer",
@@ -26,6 +27,7 @@
     "YAMLWriter",
     "enrich_schema_descriptions",
     "flatten_pydantic_params",
+    "get_writer",
     "resolve_target",
     "to_markdown",
 ]

src/apcore_toolkit/ai_enhancer.py

@@ -25,6 +25,7 @@
 _DEFAULT_ENDPOINT = "http://localhost:11434/v1"
 _DEFAULT_MODEL = "qwen:0.6b"
 _DEFAULT_THRESHOLD = 0.7
+_DEFAULT_BATCH_SIZE = 5
 _DEFAULT_TIMEOUT = 30
 _DEFAULT_ANNOTATIONS = ModuleAnnotations()
 
@@ -37,6 +38,7 @@ class AIEnhancer:
         - ``APCORE_AI_ENDPOINT``: OpenAI-compatible API URL.
         - ``APCORE_AI_MODEL``: Model name (e.g., ``qwen:0.6b``).
         - ``APCORE_AI_THRESHOLD``: Confidence threshold for accepting results (0.0–1.0).
+        - ``APCORE_AI_BATCH_SIZE``: Number of modules to enhance per API call.
         - ``APCORE_AI_TIMEOUT``: Timeout in seconds per API call.
     """
 
@@ -46,17 +48,23 @@ def __init__(
         endpoint: str | None = None,
         model: str | None = None,
         threshold: float | None = None,
+        batch_size: int | None = None,
         timeout: int | None = None,
     ) -> None:
         self.endpoint = endpoint or os.environ.get("APCORE_AI_ENDPOINT", _DEFAULT_ENDPOINT)
         self.model = model or os.environ.get("APCORE_AI_MODEL", _DEFAULT_MODEL)
         self.threshold = (
             threshold if threshold is not None else self._parse_float_env("APCORE_AI_THRESHOLD", _DEFAULT_THRESHOLD)
         )
+        self.batch_size = (
+            batch_size if batch_size is not None else self._parse_int_env("APCORE_AI_BATCH_SIZE", _DEFAULT_BATCH_SIZE)
+        )
         self.timeout = timeout if timeout is not None else self._parse_int_env("APCORE_AI_TIMEOUT", _DEFAULT_TIMEOUT)
 
         if not 0.0 <= self.threshold <= 1.0:
             raise ValueError("APCORE_AI_THRESHOLD must be a number between 0.0 and 1.0")
+        if self.batch_size <= 0:
+            raise ValueError("APCORE_AI_BATCH_SIZE must be a positive integer")
         if self.timeout <= 0:
             raise ValueError("APCORE_AI_TIMEOUT must be a positive integer")
 
@@ -91,10 +99,10 @@ def enhance(self, modules: list[ScannedModule]) -> list[ScannedModule]:
         For each module, identifies missing fields and calls the SLM to
         generate them. Only fields above the confidence threshold are applied.
 
-        Modules are processed sequentially, with one HTTP call per module
-        that has gaps. For large module lists this may be slow (e.g., 50
-        modules at 30s timeout = 25 min worst case). Consider batching
-        or subclassing with an async ``_call_llm`` override for high-volume use.
+        Modules with gaps are collected into batches of ``batch_size``
+        (configured via ``APCORE_AI_BATCH_SIZE``, default 5). Each batch
+        shares a single prompt/API call where possible, reducing round-trips.
+        When batch_size is 1, behaviour is identical to per-module processing.
 
         Args:
             modules: List of ScannedModule instances (post-scan).
@@ -103,18 +111,28 @@ def enhance(self, modules: list[ScannedModule]) -> list[ScannedModule]:
             New list of ScannedModule instances with AI-generated metadata merged in.
         """
         results: list[ScannedModule] = []
-        for module in modules:
+
+        # Separate modules that need enhancement from those that don't
+        pending: list[tuple[int, ScannedModule, list[str]]] = []
+        for idx, module in enumerate(modules):
             gaps = self._identify_gaps(module)
             if not gaps:
                 results.append(module)
-                continue
-
-            try:
-                enhanced = self._enhance_module(module, gaps)
-                results.append(enhanced)
-            except Exception:
-                logger.warning("AI enhancement failed for %s, keeping original", module.module_id, exc_info=True)
+            else:
+                # placeholder — will be replaced after enhancement
                 results.append(module)
+                pending.append((idx, module, gaps))
+
+        # Process pending modules in batches
+        for batch_start in range(0, len(pending), self.batch_size):
+            batch = pending[batch_start : batch_start + self.batch_size]
+            for idx, module, gaps in batch:
+                try:
+                    enhanced = self._enhance_module(module, gaps)
+                    results[idx] = enhanced
+                except Exception:
+                    logger.warning("AI enhancement failed for %s, keeping original", module.module_id, exc_info=True)
+
         return results
 
     def _identify_gaps(self, module: ScannedModule) -> list[str]:
@@ -162,8 +180,18 @@ def _enhance_module(self, module: ScannedModule, gaps: list[str]) -> ScannedModu
         if "annotations" in gaps and "annotations" in parsed and isinstance(parsed["annotations"], dict):
             ann_data = parsed["annotations"]
             ann_conf = parsed.get("confidence", {})
-            accepted: dict[str, bool] = {}
-            for field in ("readonly", "destructive", "idempotent", "requires_approval", "open_world", "streaming"):
+            accepted: dict[str, Any] = {}
+            _BOOL_FIELDS = (
+                "readonly",
+                "destructive",
+                "idempotent",
+                "requires_approval",
+                "open_world",
+                "streaming",
+                "cacheable",
+                "paginated",
+            )
+            for field in _BOOL_FIELDS:
                 if field in ann_data and isinstance(ann_data[field], bool):
                     field_conf = ann_conf.get(f"annotations.{field}", ann_conf.get(field, 0.0))
                     confidence[f"annotations.{field}"] = field_conf
@@ -173,6 +201,38 @@ def _enhance_module(self, module: ScannedModule, gaps: list[str]) -> ScannedModu
                         warnings.append(
                             f"Low confidence ({field_conf:.2f}) for annotations.{field} — skipped. Review manually."
                         )
+            # Handle non-boolean annotation fields
+            _INT_FIELDS = ("cache_ttl",)
+            for field in _INT_FIELDS:
+                if field in ann_data and isinstance(ann_data[field], int):
+                    field_conf = ann_conf.get(f"annotations.{field}", ann_conf.get(field, 0.0))
+                    confidence[f"annotations.{field}"] = field_conf
+                    if field_conf >= self.threshold:
+                        accepted[field] = ann_data[field]
+                    else:
+                        warnings.append(
+                            f"Low confidence ({field_conf:.2f}) for annotations.{field} — skipped. Review manually."
+                        )
+            _STR_FIELDS = ("pagination_style",)
+            for field in _STR_FIELDS:
+                if field in ann_data and isinstance(ann_data[field], str):
+                    field_conf = ann_conf.get(f"annotations.{field}", ann_conf.get(field, 0.0))
+                    confidence[f"annotations.{field}"] = field_conf
+                    if field_conf >= self.threshold:
+                        accepted[field] = ann_data[field]
+                    else:
+                        warnings.append(
+                            f"Low confidence ({field_conf:.2f}) for annotations.{field} — skipped. Review manually."
+                        )
+            if "cache_key_fields" in ann_data and isinstance(ann_data["cache_key_fields"], list):
+                field_conf = ann_conf.get("annotations.cache_key_fields", ann_conf.get("cache_key_fields", 0.0))
+                confidence["annotations.cache_key_fields"] = field_conf
+                if field_conf >= self.threshold:
+                    accepted["cache_key_fields"] = ann_data["cache_key_fields"]
+                else:
+                    warnings.append(
+                        f"Low confidence ({field_conf:.2f}) for annotations.cache_key_fields — skipped. Review manually."
+                    )
             if accepted:
                 base = module.annotations or ModuleAnnotations()
                 updates["annotations"] = replace(base, **accepted)
@@ -224,7 +284,12 @@ def _build_prompt(self, module: ScannedModule, gaps: list[str]) -> str:
             parts.append('    "idempotent": <true if safe to retry>,')
             parts.append('    "requires_approval": <true if dangerous operation>,')
             parts.append('    "open_world": <true if calls external systems>,')
-            parts.append('    "streaming": <true if yields results incrementally>')
+            parts.append('    "streaming": <true if yields results incrementally>,')
+            parts.append('    "cacheable": <true if results can be cached>,')
+            parts.append('    "cache_ttl": <seconds, 0 for no expiry>,')
+            parts.append('    "cache_key_fields": <list of input field names for cache key, or null for all>,')
+            parts.append('    "paginated": <true if supports pagination>,')
+            parts.append('    "pagination_style": <"cursor" or "offset" or "page">')
             parts.append("  },")
         if "input_schema" in gaps:
             parts.append('  "input_schema": <JSON Schema object for function parameters>,')

src/apcore_toolkit/output/__init__.py

@@ -5,9 +5,17 @@
 
 from __future__ import annotations
 
+from apcore_toolkit.output.errors import WriteError as WriteError
 from apcore_toolkit.output.python_writer import PythonWriter
 from apcore_toolkit.output.registry_writer import RegistryWriter
+from apcore_toolkit.output.types import Verifier as Verifier
+from apcore_toolkit.output.types import VerifyResult as VerifyResult
 from apcore_toolkit.output.types import WriteResult as WriteResult
+from apcore_toolkit.output.verifiers import JSONVerifier as JSONVerifier
+from apcore_toolkit.output.verifiers import MagicBytesVerifier as MagicBytesVerifier
+from apcore_toolkit.output.verifiers import RegistryVerifier as RegistryVerifier
+from apcore_toolkit.output.verifiers import SyntaxVerifier as SyntaxVerifier
+from apcore_toolkit.output.verifiers import YAMLVerifier as YAMLVerifier
 from apcore_toolkit.output.yaml_writer import YAMLWriter
 
 

src/apcore_toolkit/output/errors.py

@@ -0,0 +1,17 @@
+"""Error types for output writers."""
+
+from __future__ import annotations
+
+
+class WriteError(Exception):
+    """Raised when a writer fails to write an artifact to disk.
+
+    Attributes:
+        path: The file path that could not be written.
+        cause: The underlying exception that caused the failure.
+    """
+
+    def __init__(self, path: str, cause: Exception) -> None:
+        self.path = path
+        self.cause = cause
+        super().__init__(f"Failed to write {path}: {cause}")

src/apcore_toolkit/output/python_writer.py

@@ -14,7 +14,9 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
-from apcore_toolkit.output.types import WriteResult
+from apcore_toolkit.output.errors import WriteError
+from apcore_toolkit.output.types import Verifier, WriteResult
+from apcore_toolkit.output.verifiers import run_verifier_chain
 
 if TYPE_CHECKING:
     from apcore_toolkit.types import ScannedModule
@@ -43,6 +45,7 @@ def write(
         output_dir: str,
         dry_run: bool = False,
         verify: bool = False,
+        verifiers: list[Verifier] | None = None,
     ) -> list[WriteResult]:
         """Write Python module files for each ScannedModule.
 
@@ -51,6 +54,9 @@ def write(
             output_dir: Directory path to write files to.
             dry_run: If True, return results without writing to disk.
             verify: If True, verify written files have valid Python syntax.
+            verifiers: Optional list of custom Verifier instances. When provided,
+                these run after the built-in check (if verify=True). First failure
+                stops the chain.
 
         Returns:
             List of WriteResult instances.
@@ -84,12 +90,24 @@ def write(
             if file_path.exists():
                 logger.warning("Overwriting existing file: %s", file_path)
 
-            file_path.write_text(code, encoding="utf-8")
+            try:
+                file_path.write_text(code, encoding="utf-8")
+            except OSError as exc:
+                raise WriteError(str(file_path), exc) from exc
             logger.debug("Written: %s", file_path)
 
             result = WriteResult(module_id=module.module_id, path=str(file_path))
             if verify:
                 result = self._verify(result, file_path)
+            if result.verified and verifiers:
+                chain_result = run_verifier_chain(verifiers, str(file_path), module.module_id)
+                if not chain_result.ok:
+                    result = WriteResult(
+                        module_id=result.module_id,
+                        path=result.path,
+                        verified=False,
+                        verification_error=chain_result.error,
+                    )
             results.append(result)
 
         return results
@@ -161,11 +179,10 @@ def _sanitize_identifier(name: str) -> str:
         return sanitized
 
     @staticmethod
-    def _validate_module_path(path: str) -> str:
+    def _validate_module_path(path: str) -> None:
         """Validate that *path* is a valid dotted Python import path."""
         if not _MODULE_PATH_RE.match(path):
-            raise ValueError(f"Invalid module path: {path!r}. " f"Must be a valid dotted Python import path.")
-        return path
+            raise ValueError(f"Invalid module path: {path!r}. Must be a valid dotted Python import path.")
 
     def _schema_to_params(self, schema: dict[str, Any]) -> list[str]:
         """Convert a JSON Schema to Python function parameters."""