feat(shapes): emit JSON Schema if/then for Pose coco-17 arity (FILTER-454)

RELEASE.md

@@ -15,7 +15,7 @@ OpenFilter Library release notes
 ### Added
 
 - **OpenTelemetry MQ hop spans + per-frame trace propagation ([PLAT-866](https://plainsight-ai.atlassian.net/browse/PLAT-866), #99)**: Every filter-to-filter hop now produces `mq.send` / `mq.recv` outer hop spans (attributes: `topic`, `payload_bytes`, `frame.id`, `frame.format`, `mq.id`), with nested `frame.serialize` / `frame.deserialize` CPU sub-spans and `zmq.send_multipart` / `zmq.recv_multipart` kernel-syscall sub-spans. `frame.encode_jpg` / `frame.decode_jpg` codec sub-spans fire on the jpg transport path only. Per-frame W3C trace context (`traceparent` / `tracestate`) now travels inside the ZMQ envelope via the standard `opentelemetry.propagate.inject` / `extract` (`TraceContextTextMapPropagator`); the consumer-side `{FilterClass}.process` span (PLAT-848) parents on the per-frame context extracted from the envelope, so filter A's frame-17 spans and filter B's frame-17 spans share a single trace ID and nest correctly. The `TRACEPARENT` env-var fallback (PLAT-848 behavior) is retained for source filters that never call `mq.recv` (e.g. `VideoIn`). When tracing is disabled (the default, `TELEMETRY_EXPORTER_TYPE=silent`), the hot path short-circuits cleanly: codec / kernel sub-spans bail on a single `is_recording()` check before any tracer lookup or `Span` allocation, and the per-recv `time_ns()` brackets + per-part byte tally in `ZMQReceiver.recv_once` / `ZMQSender.send_maybe` are gated on the module-level hop tracer being set (so they don't run either). No new environment variables.
-
+- **Emit JSON Schema if/then for Pose coco-17 arity ([FILTER-454](https://plainsight-ai.atlassian.net/browse/FILTER-454))**: The `Pose` output shape now emits standard JSON Schema (draft 2020-12) `if`/`then` constraints for the coco-17 keypoint-arity invariant. When `skeleton == "coco-17"`, `keypoints` must contain exactly 17 entries. This closes the validation asymmetry between Pydantic producers and raw JSON consumers.
 - **Windowed async batch dispatch for `process_batch()`** ([FILTER-457](https://plainsight-ai.atlassian.net/browse/FILTER-457), [FILTER-458](https://plainsight-ai.atlassian.net/browse/FILTER-458), [FILTER-459](https://plainsight-ai.atlassian.net/browse/FILTER-459), [FILTER-460](https://plainsight-ai.atlassian.net/browse/FILTER-460), #95): four composable, opt-in additions to the runtime batch path. `process_batch()` result slots may now be deferred `Callable`s that `mq.send` resolves lazily, so a filter can hand batch work to a worker thread instead of blocking the loop thread on the batch's wall-clock duration. A new `accumulate_window` config decouples the runtime ring buffer from `batch_size`, with a public `Filter.select_batch()` hook to pick which frames are dispatched. `Filter.flush_batch()` plus `batch_trigger: "auto" | "manual"` give filters explicit control over when a batch drains. A `batch_workers` pool runs `process_batch()` on a `ThreadPoolExecutor` with semaphore backpressure, so the loop accumulates the next batch while previous batches are still in flight. All four preserve existing behavior when unset (`batch_workers=1`, `batch_trigger="auto"`, `accumulate_window` unset).
 
 ## v1.0.0 - 2026-05-18

openfilter/filter_runtime/shapes.py

@@ -28,7 +28,7 @@
 
 from typing import ClassVar, Literal
 
-from pydantic import Field, model_validator
+from pydantic import ConfigDict, Field, model_validator
 
 from .output import FilterOutputSchema
 
@@ -47,12 +47,15 @@
     "OCRSpan",
     "OCRSpanSet",
     "ClassificationResult",
+    "SKELETON_ARITY",
     "SHAPE_ID_BASE",
 ]
 
 
 SHAPE_ID_BASE = "https://schemas.plainsight.ai/shapes"
 
+SKELETON_ARITY = {"coco-17": 17}
+
 
 def _shape_id(slug: str) -> str:
     return f"{SHAPE_ID_BASE}/{slug}/v1"
@@ -274,13 +277,36 @@ class Pose(FilterOutputSchema):
         description="Keypoint-order convention; consumers infer edges from this.",
     )
 
+    # Standard JSON Schema (draft 2020-12) mirror of `_validate_skeleton_arity`
+    # so stock validators (Ajv, santhosh-tekuri) enforce the coco-17 arity from
+    # `emit_schema()` alone — no openfilter custom-keyword vocabulary needed
+    # (FILTER-454 slice A). `__init_subclass__` merges `$id` into this dict.
+    model_config = ConfigDict(
+        json_schema_extra={
+            "if": {
+                "properties": {"skeleton": {"const": next(iter(SKELETON_ARITY))}},
+                "required": ["skeleton"],
+            },
+            "then": {
+                "properties": {
+                    "keypoints": {
+                        "minItems": next(iter(SKELETON_ARITY.values())),
+                        "maxItems": next(iter(SKELETON_ARITY.values())),
+                    }
+                },
+            },
+        }
+    )
+
     @model_validator(mode="after")
     def _validate_skeleton_arity(self) -> "Pose":
-        if self.skeleton == "coco-17" and len(self.keypoints) != 17:
-            raise ValueError(
-                f"Pose.skeleton='coco-17' requires 17 keypoints, "
-                f"got {len(self.keypoints)}"
-            )
+        if self.skeleton in SKELETON_ARITY:
+            expected = SKELETON_ARITY[self.skeleton]
+            if len(self.keypoints) != expected:
+                raise ValueError(
+                    f"Pose.skeleton={self.skeleton!r} requires {expected} keypoints, "
+                    f"got {len(self.keypoints)}"
+                )
         return self
 
 

tests/test_output_schema.py

@@ -360,6 +360,32 @@ def test_pose_coco17_rejects_wrong_arity() -> None:
         )
 
 
+def test_pose_emits_coco17_arity_if_then() -> None:
+    """Slice A (FILTER-454): the coco-17 keypoint-arity invariant — Pydantic-only
+    via `_validate_skeleton_arity` — is mirrored into standard JSON Schema
+    `if`/`then` so stock validators (Ajv, santhosh-tekuri) enforce it from
+    `emit_schema()` alone, without the openfilter custom-keyword vocabulary."""
+    schema = Pose.emit_schema()
+    assert schema["if"] == {
+        "properties": {"skeleton": {"const": "coco-17"}},
+        "required": ["skeleton"],
+    }
+    assert schema["then"] == {
+        "properties": {"keypoints": {"minItems": 17, "maxItems": 17}},
+    }
+    # $id still merges in alongside the if/then (json_schema_extra co-exists).
+    assert schema["$id"] == f"{SHAPE_ID_BASE}/pose/v1"
+
+
+def test_pose_arity_if_then_travels_into_defs() -> None:
+    """The constraint rides with the Pose `$defs` entry wherever Pose is
+    `$ref`'d, so consumers validating a PoseSet get it too."""
+    schema = Pose.emit_schema()
+    pose_def = PoseSet.emit_schema()["$defs"]["Pose"]
+    assert pose_def["if"] == schema["if"]
+    assert pose_def["then"] == schema["then"]
+
+
 def test_classification_result_supports_multilabel() -> None:
     single = ClassificationResult(classes=["dog"], confidences=[0.9])
     assert single.multilabel is False