refactor: unify schema and io boundary fixtures

docs/openapi.yaml

@@ -362,6 +362,115 @@ paths:
       security:
         - OpenClawAdminToken:
             []
+  /models/search:
+    get:
+      operationId: "get_models_search"
+      summary: "Search normalized model entries across managed installs and catalog sources."
+      responses:
+        200:
+          description: "OK"
+      x-openclaw-auth: "Admin"
+      x-openclaw-section: "1.3C Model Management & Installations"
+      x-openclaw-legacy-path: "/moltbot/models/search"
+      x-openclaw-auth-tier: "admin"
+      security:
+        - OpenClawAdminToken:
+            []
+  /models/downloads:
+    post:
+      operationId: "post_models_downloads"
+      summary: "Create a managed model download task with progress/cancel lifecycle."
+      responses:
+        200:
+          description: "OK"
+      x-openclaw-auth: "Admin"
+      x-openclaw-section: "1.3C Model Management & Installations"
+      x-openclaw-legacy-path: "/moltbot/models/downloads"
+      x-openclaw-auth-tier: "admin"
+      security:
+        - OpenClawAdminToken:
+            []
+    get:
+      operationId: "get_models_downloads"
+      summary: "List model download tasks with snapshot or delta cursor semantics (`since_seq`)."
+      responses:
+        200:
+          description: "OK"
+      x-openclaw-auth: "Admin"
+      x-openclaw-section: "1.3C Model Management & Installations"
+      x-openclaw-legacy-path: "/moltbot/models/downloads"
+      x-openclaw-auth-tier: "admin"
+      security:
+        - OpenClawAdminToken:
+            []
+  /models/downloads/{task_id}:
+    get:
+      operationId: "get_models_downloads_task_id"
+      summary: "Get one model download task by id."
+      responses:
+        200:
+          description: "OK"
+      x-openclaw-auth: "Admin"
+      x-openclaw-section: "1.3C Model Management & Installations"
+      x-openclaw-legacy-path: "/moltbot/models/downloads/{task_id}"
+      x-openclaw-auth-tier: "admin"
+      security:
+        - OpenClawAdminToken:
+            []
+      parameters:
+        - name: "task_id"
+          in: "path"
+          required: true
+          schema:
+            type: "string"
+  /models/downloads/{task_id}/cancel:
+    post:
+      operationId: "post_models_downloads_task_id_cancel"
+      summary: "Cancel a queued or running model download task."
+      responses:
+        200:
+          description: "OK"
+      x-openclaw-auth: "Admin"
+      x-openclaw-section: "1.3C Model Management & Installations"
+      x-openclaw-legacy-path: "/moltbot/models/downloads/{task_id}/cancel"
+      x-openclaw-auth-tier: "admin"
+      security:
+        - OpenClawAdminToken:
+            []
+      parameters:
+        - name: "task_id"
+          in: "path"
+          required: true
+          schema:
+            type: "string"
+  /models/import:
+    post:
+      operationId: "post_models_import"
+      summary: "Import a completed managed download into the bounded install root after provenance and hash verification."
+      responses:
+        200:
+          description: "OK"
+      x-openclaw-auth: "Admin"
+      x-openclaw-section: "1.3C Model Management & Installations"
+      x-openclaw-legacy-path: "/moltbot/models/import"
+      x-openclaw-auth-tier: "admin"
+      security:
+        - OpenClawAdminToken:
+            []
+  /models/installations:
+    get:
+      operationId: "get_models_installations"
+      summary: "List managed model installations."
+      responses:
+        200:
+          description: "OK"
+      x-openclaw-auth: "Admin"
+      x-openclaw-section: "1.3C Model Management & Installations"
+      x-openclaw-legacy-path: "/moltbot/models/installations"
+      x-openclaw-auth-tier: "admin"
+      security:
+        - OpenClawAdminToken:
+            []
   /chat:
     post:
       operationId: "post_chat"
@@ -370,7 +479,7 @@ paths:
         200:
           description: "OK"
       x-openclaw-auth: "Admin/Local"
-      x-openclaw-section: "1.3C LLM Management & Chat"
+      x-openclaw-section: "1.3D LLM Management & Chat"
       x-openclaw-legacy-path: "/moltbot/llm/chat"
       x-openclaw-auth-tier: "admin"
       security:
@@ -384,7 +493,7 @@ paths:
         200:
           description: "OK"
       x-openclaw-auth: "Admin"
-      x-openclaw-section: "1.3C LLM Management & Chat"
+      x-openclaw-section: "1.3D LLM Management & Chat"
       x-openclaw-legacy-path: "/moltbot/llm/test"
       x-openclaw-auth-tier: "admin"
       security:
@@ -398,7 +507,7 @@ paths:
         200:
           description: "OK"
       x-openclaw-auth: "Admin"
-      x-openclaw-section: "1.3C LLM Management & Chat"
+      x-openclaw-section: "1.3D LLM Management & Chat"
       x-openclaw-legacy-path: "/moltbot/llm/models"
       x-openclaw-auth-tier: "admin"
       security:

docs/release/api_contract.md

@@ -104,7 +104,27 @@ Connector diagnostics contract notes:
 - `/connector/installations` diagnostics may include aggregate `health_counts` in addition to lifecycle `status_counts`
 - `/connector/installations/resolve` may expose a stable `health_code` alongside the legacy `reject_reason` so clients can distinguish `workspace_unbound` vs token-health failures without parsing status text
 
-### 1.3C LLM Management & Chat
+### 1.3C Model Management & Installations
+
+**Base Path**: `/openclaw/`
+**Auth**: Admin Token Required
+
+| Method | Path | Legacy Path | Auth | Description |
+| :--- | :--- | :--- | :--- | :--- |
+| `GET` | `/models/search` | `/moltbot/models/search` | Admin | Search normalized model entries across managed installs and catalog sources. |
+| `POST` | `/models/downloads` | `/moltbot/models/downloads` | Admin | Create a managed model download task with progress/cancel lifecycle. |
+| `GET` | `/models/downloads` | `/moltbot/models/downloads` | Admin | List model download tasks with snapshot or delta cursor semantics (`since_seq`). |
+| `GET` | `/models/downloads/{task_id}` | `/moltbot/models/downloads/{task_id}` | Admin | Get one model download task by id. |
+| `POST` | `/models/downloads/{task_id}/cancel` | `/moltbot/models/downloads/{task_id}/cancel` | Admin | Cancel a queued or running model download task. |
+| `POST` | `/models/import` | `/moltbot/models/import` | Admin | Import a completed managed download into the bounded install root after provenance and hash verification. |
+| `GET` | `/models/installations` | `/moltbot/models/installations` | Admin | List managed model installations. |
+
+Model-manager contract notes:
+- `/models/downloads` supports `since_seq` cursor polling and may return deterministic delta metadata (`requested_since_seq`, `effective_since_seq`, `next_since_seq`, truncation/reset hints) alongside the task list
+- download creation requires structured provenance metadata (`publisher`, `license`, `source_url`) and a 64-char `expected_sha256`
+- import keeps fail-closed destination/filename validation and re-checks the staged file hash before activation
+
+### 1.3D LLM Management & Chat
 
 **LLM Base Path**: `/openclaw/llm/`
 

models/schemas.py

@@ -3,7 +3,28 @@ import re
 from dataclasses import asdict, dataclass, field
 from typing import Any, Dict, List, Optional, Union
 
-SCHEMA_VERSION = "260127"
+if __package__ and "." in __package__:
+    from ..services.request_contracts import (
+        MAX_BODY_SIZE,
+        MAX_INPUT_STRING_LENGTH,
+        MAX_JOB_ID_LENGTH,
+        MAX_PROFILE_ID_LENGTH,
+        MAX_TEMPLATE_ID_LENGTH,
+        MAX_TRACE_ID_LENGTH,
+        SCHEMA_VERSION,
+        WEBHOOK_JOB_REQUEST_CONTRACT,
+    )
+else:  # pragma: no cover - top-level test import mode
+    from services.request_contracts import (  # type: ignore
+        MAX_BODY_SIZE,
+        MAX_INPUT_STRING_LENGTH,
+        MAX_JOB_ID_LENGTH,
+        MAX_PROFILE_ID_LENGTH,
+        MAX_TEMPLATE_ID_LENGTH,
+        MAX_TRACE_ID_LENGTH,
+        SCHEMA_VERSION,
+        WEBHOOK_JOB_REQUEST_CONTRACT,
+    )
 
 
 @dataclass
@@ -93,16 +114,6 @@ class ParamPatch:
     reason: Optional[str] = None
 
 
-# S2: Webhook input schema
-# Maximum field lengths for security
-MAX_JOB_ID_LENGTH = 64
-MAX_TEMPLATE_ID_LENGTH = 64
-MAX_PROFILE_ID_LENGTH = 64
-MAX_INPUT_STRING_LENGTH = 2048
-MAX_BODY_SIZE = 65536  # 64KB
-MAX_TRACE_ID_LENGTH = 64
-
-
 @dataclass
 class WebhookJobRequest:
     """
@@ -138,13 +149,7 @@ class WebhookJobRequest:
             raise ValueError(f"profile_id exceeds max length ({MAX_PROFILE_ID_LENGTH})")
 
         # Validate inputs (only allowed keys, string length limits)
-        allowed_input_keys = {
+        allowed_input_keys = set(WEBHOOK_JOB_REQUEST_CONTRACT["allowed_input_keys"])
-            "requirements",
-            "goal",
-            "seed",
-            "positive_prompt",
-            "negative_prompt",
-        }
         for key, value in self.inputs.items():
             if key not in allowed_input_keys:
                 raise ValueError(f"Unknown input key: {key}")
@@ -157,21 +162,13 @@ class WebhookJobRequest:
     def from_dict(cls, data: Dict[str, Any]) -> "WebhookJobRequest":
         """Parse and validate from dict."""
         # 1. Check for unknown keys (Strict validation)
-        allowed_top_level = {
+        allowed_top_level = set(WEBHOOK_JOB_REQUEST_CONTRACT["allowed_top_level"])
-            "version",
-            "template_id",
-            "profile_id",
-            "inputs",
-            "job_id",
-            "trace_id",
-            "callback",
-        }
         unknown = set(data.keys()) - allowed_top_level
         if unknown:
             raise ValueError(f"Unknown fields: {unknown}")
 
         # 2. Check required fields
-        required = {"version", "template_id", "profile_id"}
+        required = set(WEBHOOK_JOB_REQUEST_CONTRACT["required_top_level"])
         missing = required - set(data.keys())
         if missing:
             raise ValueError(f"Missing required fields: {missing}")

services/model_manager.py

@@ -252,11 +252,14 @@ def _is_sha256(value: str) -> bool:
 
 
 def _sanitize_subdir(text: str) -> str:
-    parts = [
+    raw_parts = str(text or "").replace("\\", "/").split("/")
-        p
+    if any(part in {".", ".."} for part in raw_parts if part):
-        for p in str(text or "").replace("\\", "/").split("/")
+        # CRITICAL: reject traversal markers instead of stripping them; silent cleanup weakens import-boundary guarantees.
-        if p not in {"", ".", ".."}
+        raise ModelManagerError(
-    ]
+            "invalid_destination",
+            "destination_subdir must not contain traversal segments",
+        )
+    parts = [p for p in raw_parts if p]
     if not parts:
         raise ModelManagerError("invalid_destination", "destination_subdir is required")
     cleaned = []

services/model_manager_transfer.py

@@ -14,6 +14,11 @@ import uuid
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from .request_contracts import (
+    MODEL_MANAGER_IMPORT_CONTRACT,
+    MODEL_MANAGER_PROVENANCE_CONTRACT,
+)
+
 
 def validate_url_policy(*, manager: Any, url: str) -> None:
     if not str(url or "").strip():
@@ -37,13 +42,15 @@ def validate_url_policy(*, manager: Any, url: str) -> None:
 def validate_provenance(*, manager: Any, provenance: Dict[str, Any]) -> Dict[str, Any]:
     if not isinstance(provenance, dict):
         raise manager._error("invalid_provenance", "provenance must be an object")
+    note_max_chars = int(MODEL_MANAGER_PROVENANCE_CONTRACT["note_max_chars"])
     out = {
         "publisher": str(provenance.get("publisher") or "").strip(),
         "license": str(provenance.get("license") or "").strip(),
         "source_url": str(provenance.get("source_url") or "").strip(),
-        "note": str(provenance.get("note") or "").strip()[:500],
+        "note": str(provenance.get("note") or "").strip()[:note_max_chars],
     }
-    if not out["publisher"] or not out["license"] or not out["source_url"]:
+    required_fields = list(MODEL_MANAGER_PROVENANCE_CONTRACT["required_fields"])
+    if any(not out[field] for field in required_fields):
         raise manager._error(
             "invalid_provenance",
             "provenance.publisher, provenance.license, provenance.source_url are required",
@@ -450,6 +457,7 @@ def import_downloaded_model(
             pass
         raise
     safe_tags: List[str] = []
+    max_tags = int(MODEL_MANAGER_IMPORT_CONTRACT["tags"]["max_items"])
     for item in tags or []:
         if not isinstance(item, str):
             continue
@@ -457,7 +465,7 @@ def import_downloaded_model(
         if not clean or clean in safe_tags:
             continue
         safe_tags.append(clean)
-        if len(safe_tags) >= 24:
+        if len(safe_tags) >= max_tags:
             break
     rec = {
         "id": str(uuid.uuid4()),

services/request_contracts.py

@@ -0,0 +1,188 @@
+"""
+R144 shared JSON-serializable request/schema fixtures.
+
+These fixtures are consumed by runtime validators and contract tests so route
+behavior, schema limits, and public docs do not drift independently.
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any, Dict
+
+SCHEMA_VERSION = "260127"
+
+MAX_JOB_ID_LENGTH = 64
+MAX_TEMPLATE_ID_LENGTH = 64
+MAX_PROFILE_ID_LENGTH = 64
+MAX_INPUT_STRING_LENGTH = 2048
+MAX_BODY_SIZE = 65536
+MAX_TRACE_ID_LENGTH = 64
+
+WEBHOOK_JOB_REQUEST_CONTRACT: Dict[str, Any] = {
+    "schema_id": "webhook_job_request_v1",
+    "version": 1,
+    "required_top_level": ["version", "template_id", "profile_id"],
+    "allowed_top_level": [
+        "version",
+        "template_id",
+        "profile_id",
+        "inputs",
+        "job_id",
+        "trace_id",
+        "callback",
+    ],
+    "allowed_input_keys": [
+        "requirements",
+        "goal",
+        "seed",
+        "positive_prompt",
+        "negative_prompt",
+    ],
+    "limits": {
+        "job_id_max_length": MAX_JOB_ID_LENGTH,
+        "template_id_max_length": MAX_TEMPLATE_ID_LENGTH,
+        "profile_id_max_length": MAX_PROFILE_ID_LENGTH,
+        "input_string_max_length": MAX_INPUT_STRING_LENGTH,
+        "body_max_bytes": MAX_BODY_SIZE,
+        "trace_id_max_length": MAX_TRACE_ID_LENGTH,
+    },
+    "trace_id_pattern": "^[a-zA-Z0-9_-]+$",
+}
+
+MODEL_MANAGER_PROVENANCE_CONTRACT: Dict[str, Any] = {
+    "schema_id": "model_manager_provenance_v1",
+    "required_fields": ["publisher", "license", "source_url"],
+    "optional_fields": ["note"],
+    "note_max_chars": 500,
+}
+
+MODEL_MANAGER_IMPORT_CONTRACT: Dict[str, Any] = {
+    "schema_id": "model_manager_import_v1",
+    "required_fields": ["task_id"],
+    "optional_fields": ["destination_subdir", "filename", "tags"],
+    "tags": {
+        "max_items": 24,
+        "normalize": "strip_lower_dedupe",
+    },
+}
+
+R144_ROUTE_FIXTURES: Dict[str, Any] = {
+    "webhook": [
+        {
+            "method": "POST",
+            "path": "/webhook",
+            "legacy_path": "/moltbot/webhook",
+            "auth": "Webhook Secret",
+        },
+        {
+            "method": "POST",
+            "path": "/webhook/submit",
+            "legacy_path": "/moltbot/webhook/submit",
+            "auth": "Webhook Secret",
+        },
+        {
+            "method": "POST",
+            "path": "/webhook/validate",
+            "legacy_path": "/moltbot/webhook/validate",
+            "auth": "Webhook Secret",
+        },
+    ],
+    "model_manager": [
+        {
+            "method": "GET",
+            "path": "/models/search",
+            "legacy_path": "/moltbot/models/search",
+            "auth": "Admin",
+        },
+        {
+            "method": "POST",
+            "path": "/models/downloads",
+            "legacy_path": "/moltbot/models/downloads",
+            "auth": "Admin",
+        },
+        {
+            "method": "GET",
+            "path": "/models/downloads",
+            "legacy_path": "/moltbot/models/downloads",
+            "auth": "Admin",
+        },
+        {
+            "method": "GET",
+            "path": "/models/downloads/{task_id}",
+            "legacy_path": "/moltbot/models/downloads/{task_id}",
+            "auth": "Admin",
+        },
+        {
+            "method": "POST",
+            "path": "/models/downloads/{task_id}/cancel",
+            "legacy_path": "/moltbot/models/downloads/{task_id}/cancel",
+            "auth": "Admin",
+        },
+        {
+            "method": "POST",
+            "path": "/models/import",
+            "legacy_path": "/moltbot/models/import",
+            "auth": "Admin",
+        },
+        {
+            "method": "GET",
+            "path": "/models/installations",
+            "legacy_path": "/moltbot/models/installations",
+            "auth": "Admin",
+        },
+    ],
+}
+
+R144_IO_BOUNDARY_MATRIX: Dict[str, Any] = {
+    "webhook_validate": [
+        {
+            "case_id": "body_gt_limit_rejected",
+            "limit_bytes": MAX_BODY_SIZE,
+            "expected_status": 413,
+            "expected_error": "payload_too_large",
+        },
+        {
+            "case_id": "body_eq_limit_not_rejected_by_size_gate",
+            "limit_bytes": MAX_BODY_SIZE,
+            "expected_status": 400,
+            "expected_error": "validation_error",
+        },
+        {
+            "case_id": "malformed_json_rejected",
+            "expected_status": 400,
+            "expected_error": "invalid_json",
+        },
+    ],
+    "model_manager_download_create": [
+        {
+            "case_id": "provenance_must_be_object",
+            "expected_status": 400,
+            "expected_error": "invalid_provenance",
+        }
+    ],
+    "model_manager_import": [
+        {
+            "case_id": "invalid_filename_extension_rejected",
+            "expected_status": 400,
+            "expected_error": "invalid_filename",
+        },
+        {
+            "case_id": "invalid_destination_rejected",
+            "expected_status": 400,
+            "expected_error": "invalid_destination",
+        },
+    ],
+}
+
+
+def get_serializable_contract_bundle() -> Dict[str, Any]:
+    return deepcopy(
+        {
+            "webhook_job_request": WEBHOOK_JOB_REQUEST_CONTRACT,
+            "model_manager_provenance": MODEL_MANAGER_PROVENANCE_CONTRACT,
+            "model_manager_import": MODEL_MANAGER_IMPORT_CONTRACT,
+            "route_fixtures": R144_ROUTE_FIXTURES,
+            "io_boundary_matrix": R144_IO_BOUNDARY_MATRIX,
+        }
+    )

tests/test_model_manager_api.py

@@ -18,6 +18,7 @@ except Exception:  # pragma: no cover
 
 from api import model_manager as mm_api
 from services.model_manager import ModelManager
+from services.request_contracts import R144_IO_BOUNDARY_MATRIX
 
 
 @unittest.skipIf(web is None, "aiohttp not installed")
@@ -70,14 +71,8 @@ class TestModelManagerAPI(AioHTTPTestCase):
             time.sleep(0.02)
         self.fail(f"Task {task_id} did not finish")
 
-    @patch("api.model_manager.require_admin_token", return_value=(True, None))
+    async def _create_completed_task(self, *, model_id: str = "api-model") -> str:
-    @patch(
+        payload = f"{model_id}-bytes".encode("utf-8")
-        "services.model_manager.validate_outbound_url",
-        return_value=("https", "example.com", 443, ["1.1.1.1"]),
-    )
-    @unittest_run_loop
-    async def test_download_and_import_contract(self, _mock_validate, _mock_admin):
-        payload = b"api-model-bytes"
         digest = hashlib.sha256(payload).hexdigest()
 
         def fake_download(task, _cancel_event):
@@ -92,26 +87,35 @@ class TestModelManagerAPI(AioHTTPTestCase):
         create_resp = await self.client.post(
             "/openclaw/models/downloads",
             json={
-                "model_id": "api-model",
+                "model_id": model_id,
                 "name": "API Model",
                 "model_type": "checkpoint",
                 "source": "catalog",
                 "source_label": "Catalog",
-                "download_url": "https://example.com/api-model.safetensors",
+                "download_url": f"https://example.com/{model_id}.safetensors",
                 "expected_sha256": digest,
                 "provenance": {
                     "publisher": "OpenClaw",
                     "license": "OpenRAIL",
-                    "source_url": "https://example.com/api-model",
+                    "source_url": f"https://example.com/{model_id}",
                 },
             },
         )
         self.assertEqual(create_resp.status, 201)
         created = await create_resp.json()
         task_id = created["task"]["task_id"]
-
         done = await self._wait_task_terminal(task_id)
         self.assertEqual(done["state"], "completed")
+        return task_id
+
+    @patch("api.model_manager.require_admin_token", return_value=(True, None))
+    @patch(
+        "services.model_manager.validate_outbound_url",
+        return_value=("https", "example.com", 443, ["1.1.1.1"]),
+    )
+    @unittest_run_loop
+    async def test_download_and_import_contract(self, _mock_validate, _mock_admin):
+        task_id = await self._create_completed_task(model_id="api-model")
 
         import_resp = await self.client.post(
             "/openclaw/models/import", json={"task_id": task_id}
@@ -142,6 +146,62 @@ class TestModelManagerAPI(AioHTTPTestCase):
         resp = await self.client.get("/openclaw/models/search")
         self.assertEqual(resp.status, 403)
 
+    @patch("api.model_manager.require_admin_token", return_value=(True, None))
+    @patch(
+        "services.model_manager.validate_outbound_url",
+        return_value=("https", "example.com", 443, ["1.1.1.1"]),
+    )
+    @unittest_run_loop
+    async def test_r144_download_create_boundary_matrix(
+        self, _mock_validate, _mock_admin
+    ):
+        case = R144_IO_BOUNDARY_MATRIX["model_manager_download_create"][0]
+        resp = await self.client.post(
+            "/openclaw/models/downloads",
+            json={
+                "model_id": "bad-provenance",
+                "name": "Bad Provenance",
+                "model_type": "checkpoint",
+                "source": "catalog",
+                "source_label": "Catalog",
+                "download_url": "https://example.com/bad-provenance.safetensors",
+                "expected_sha256": "a" * 64,
+                "provenance": "bad",
+            },
+        )
+        self.assertEqual(resp.status, case["expected_status"])
+        body = await resp.json()
+        self.assertEqual(body["error"], case["expected_error"])
+
+    @patch("api.model_manager.require_admin_token", return_value=(True, None))
+    @patch(
+        "services.model_manager.validate_outbound_url",
+        return_value=("https", "example.com", 443, ["1.1.1.1"]),
+    )
+    @unittest_run_loop
+    async def test_r144_import_boundary_matrix(self, _mock_validate, _mock_admin):
+        task_id = await self._create_completed_task(model_id="boundary-model")
+        cases = R144_IO_BOUNDARY_MATRIX["model_manager_import"]
+        payloads = {
+            "invalid_filename_extension_rejected": {
+                "task_id": task_id,
+                "filename": "bad.txt",
+            },
+            "invalid_destination_rejected": {
+                "task_id": task_id,
+                "destination_subdir": "../escape",
+            },
+        }
+        for case in cases:
+            with self.subTest(case=case["case_id"]):
+                resp = await self.client.post(
+                    "/openclaw/models/import",
+                    json=payloads[case["case_id"]],
+                )
+                self.assertEqual(resp.status, case["expected_status"])
+                body = await resp.json()
+                self.assertEqual(body["error"], case["expected_error"])
+
 
 if __name__ == "__main__":  # pragma: no cover
     unittest.main()

tests/test_r144_contract_matrix.py

@@ -0,0 +1,31 @@
+import json
+import unittest
+
+from services.openapi_generation import parse_api_contract_markdown
+from services.request_contracts import get_serializable_contract_bundle
+
+
+class TestR144ContractMatrix(unittest.TestCase):
+    def test_contract_bundle_is_json_serializable(self):
+        bundle = get_serializable_contract_bundle()
+        encoded = json.dumps(bundle, sort_keys=True)
+        self.assertIn("webhook_job_request_v1", encoded)
+        self.assertIn("model_manager_import_v1", encoded)
+
+    def test_documented_routes_cover_r144_fixture_paths(self):
+        documented = {
+            (route.method, route.path)
+            for route in parse_api_contract_markdown()
+        }
+        bundle = get_serializable_contract_bundle()
+        missing = []
+        for family in bundle["route_fixtures"].values():
+            for route in family:
+                key = (route["method"], route["path"])
+                if key not in documented:
+                    missing.append(key)
+        self.assertEqual(missing, [], f"Undocumented R144 route fixtures: {missing}")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webhook_validate.py

@@ -37,6 +37,26 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 from api.webhook_validate import webhook_validate_handler
 from models.schemas import WebhookJobRequest
 from services.execution_budgets import BudgetExceededError
+from services.request_contracts import MAX_BODY_SIZE, R144_IO_BOUNDARY_MATRIX
+
+
+def _json_payload_with_exact_size(target_bytes: int) -> str:
+    base = {"version": 1, "template_id": "t", "profile_id": "p", "pad": ""}
+    text = json.dumps(base, separators=(",", ":"))
+    overhead = len(text.encode("utf-8"))
+    filler = max(0, target_bytes - overhead)
+    base["pad"] = "x" * filler
+    text = json.dumps(base, separators=(",", ":"))
+    delta = target_bytes - len(text.encode("utf-8"))
+    if delta > 0:
+        base["pad"] += "x" * delta
+        text = json.dumps(base, separators=(",", ":"))
+    elif delta < 0:
+        base["pad"] = base["pad"][:delta]
+        text = json.dumps(base, separators=(",", ":"))
+    if len(text.encode("utf-8")) != target_bytes:
+        raise AssertionError("failed to build exact-size JSON payload")
+    return text
 
 
 @unittest.skipUnless(_AIOHTTP_AVAILABLE, "aiohttp not installed")
@@ -157,9 +177,10 @@ class TestWebhookValidateContract(AioHTTPTestCase):
     @unittest_run_loop
     async def test_payload_too_large_413_body(self):
         """Should return 413 for large body."""
+        case = R144_IO_BOUNDARY_MATRIX["webhook_validate"][0]
         with patch("api.webhook_validate.require_auth", return_value=(True, None)):
             with patch("api.webhook_validate.check_rate_limit", return_value=True):
-                large_payload = {"data": "x" * (2 * 1024 * 1024)}  # > MAX_BODY_SIZE
+                large_payload = {"data": "x" * (2 * MAX_BODY_SIZE)}
                 resp = await self.client.post(
                     "/validate",
                     json=large_payload,
@@ -169,10 +190,30 @@ class TestWebhookValidateContract(AioHTTPTestCase):
                     },
                 )
 
-                self.assertEqual(resp.status, 413)
+                self.assertEqual(resp.status, case["expected_status"])
                 data = await resp.json()
                 self.assertFalse(data["ok"])
-                self.assertEqual(data["error"], "payload_too_large")
+                self.assertEqual(data["error"], case["expected_error"])
+
+    @unittest_run_loop
+    async def test_r144_boundary_equal_body_is_not_rejected_by_size_gate(self):
+        """R144: payload exactly at the byte cap should pass the size gate."""
+        case = R144_IO_BOUNDARY_MATRIX["webhook_validate"][1]
+        with patch("api.webhook_validate.require_auth", return_value=(True, None)):
+            with patch("api.webhook_validate.check_rate_limit", return_value=True):
+                payload = _json_payload_with_exact_size(case["limit_bytes"])
+                resp = await self.client.post(
+                    "/validate",
+                    data=payload.encode("utf-8"),
+                    headers={
+                        "Authorization": "Bearer token",
+                        "Content-Type": "application/json",
+                    },
+                )
+
+                self.assertEqual(resp.status, case["expected_status"])
+                data = await resp.json()
+                self.assertEqual(data["error"], case["expected_error"])
 
     @unittest_run_loop
     async def test_payload_too_large_413_render(self):
@@ -219,6 +260,7 @@ class TestWebhookValidateContract(AioHTTPTestCase):
     @unittest_run_loop
     async def test_invalid_json_400(self):
         """Should return 400 for malformed JSON."""
+        case = R144_IO_BOUNDARY_MATRIX["webhook_validate"][2]
         with patch("api.webhook_validate.require_auth", return_value=(True, None)):
             with patch("api.webhook_validate.check_rate_limit", return_value=True):
                 resp = await self.client.post(
@@ -230,10 +272,10 @@ class TestWebhookValidateContract(AioHTTPTestCase):
                     },
                 )
 
-                self.assertEqual(resp.status, 400)
+                self.assertEqual(resp.status, case["expected_status"])
                 data = await resp.json()
                 self.assertFalse(data["ok"])
-                self.assertEqual(data["error"], "invalid_json")
+                self.assertEqual(data["error"], case["expected_error"])
 
     @unittest_run_loop
     async def test_validation_error_400(self):