Enforce 95% Python SDK coverage in CI

.github/workflows/ci.yml

@@ -21,5 +21,8 @@ jobs:
         run: |
           pip install -e ".[dev]"
 
-      - name: Run tests
+      - name: Run tests with coverage
         run: pytest -q
+
+      - name: Enforce coverage threshold
+        run: bash ./scripts/check-coverage.sh 95

.gitignore

@@ -3,6 +3,8 @@ __pycache__/
 *.py[cod]
 *$py.class
 .pytest_cache/
+.coverage
+htmlcov/
 
 # Virtual environments
 .venv/

README.md

@@ -8,12 +8,15 @@ Early scaffolding repo for IntentProof's Python SDK. Tracks the
 Node SDK's wrap()/exporter/outbox contract so a Python application
 can emit and verify the same signed execution events.
 
-## Planned scope
+## Development
 
-- Event wrapping helpers
-- Correlation and chain management
-- Canonical serialization and signing
-- Hosted ingest transport
+```bash
+pip install -e ".[dev]"
+pytest
+```
+
+CI enforces at least 95% line coverage on `src/intentproof/` (see
+`pyproject.toml` and `scripts/check-coverage.sh`).
 
 ## License
 

pyproject.toml

@@ -17,11 +17,26 @@ dependencies = [
 [project.optional-dependencies]
 dev = [
     "pytest>=8.0.0",
+    "pytest-cov>=5.0.0",
 ]
 
 [tool.setuptools.packages.find]
 where = ["src"]
 
+[tool.coverage.run]
+source = ["intentproof"]
+omit = []
+
+[tool.coverage.report]
+fail_under = 95
+show_missing = true
+skip_empty = true
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 pythonpath = ["src"]
+addopts = [
+    "--cov=intentproof",
+    "--cov-report=term-missing",
+    "--cov-fail-under=95",
+]

scripts/check-coverage.sh

@@ -0,0 +1,41 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+MIN_COVERAGE="${1:-95}"
+
+COVERAGE=()
+if command -v coverage >/dev/null 2>&1; then
+  COVERAGE=(coverage)
+else
+  for py in python python3; do
+    if command -v "$py" >/dev/null 2>&1 && "$py" -m coverage --version >/dev/null 2>&1; then
+      COVERAGE=("$py" -m coverage)
+      break
+    fi
+  done
+fi
+
+if [[ ${#COVERAGE[@]} -eq 0 ]]; then
+  echo "coverage CLI not found; install dev deps with: pip install -e \".[dev]\"" >&2
+  exit 2
+fi
+
+TOTAL_LINE="$("${COVERAGE[@]}" report --include='src/intentproof/*' | awk '/^TOTAL/{print; exit}')"
+if [[ -z "$TOTAL_LINE" ]]; then
+  echo "unable to read total coverage; run pytest with --cov first" >&2
+  exit 2
+fi
+
+TOTAL_PERCENT="$(printf '%s' "$TOTAL_LINE" | awk '{print $NF}' | tr -d '%')"
+
+echo "Total coverage: ${TOTAL_PERCENT}%"
+echo "Minimum required: ${MIN_COVERAGE}%"
+
+if awk -v got="$TOTAL_PERCENT" -v min="$MIN_COVERAGE" 'BEGIN { exit !(got + 0 >= min + 0) }'; then
+  echo "PASS: coverage threshold met"
+  exit 0
+fi
+
+echo "FAIL: coverage threshold not met" >&2
+exit 1

tests/test_async_wrap.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from intentproof import client, configure, wrap
+from intentproof import client, configure, run_with_correlation_id, wrap
 
 
 def test_cancelled_error_not_recorded(tmp_path) -> None:
@@ -25,3 +25,94 @@ async def cancelled() -> None:
         asyncio.run(fn())
 
     assert client.get_outbox().get_events() == []
+
+
+def test_async_wrap_records_success(tmp_path) -> None:
+    configure(
+        db_path=str(tmp_path / "outbox.db"),
+        data_dir=str(tmp_path / "data"),
+        tenant_id="tnt_async",
+    )
+
+    async def add(a: int, b: int) -> int:
+        return a + b
+
+    fn = wrap(intent="Async", action="async.add", fn=add)
+    result = run_with_correlation_id("corr-async-ok", lambda: asyncio.run(fn(2, 3)))
+
+    assert result == 5
+    events = client.get_outbox().get_events()
+    assert len(events) == 1
+    assert events[0]["status"] == "ok"
+    assert events[0]["output"] == 5
+
+
+def test_async_wrap_records_error(tmp_path) -> None:
+    configure(
+        db_path=str(tmp_path / "outbox.db"),
+        data_dir=str(tmp_path / "data"),
+        tenant_id="tnt_async",
+    )
+
+    async def boom() -> None:
+        raise ValueError("async boom")
+
+    fn = wrap(intent="Async", action="async.boom", fn=boom)
+    with pytest.raises(ValueError, match="async boom"):
+        asyncio.run(fn())
+
+    events = client.get_outbox().get_events()
+    assert events[-1]["status"] == "error"
+    assert events[-1]["error"] == {"message": "async boom"}
+
+
+def test_async_wrap_preserves_app_error_when_record_fails(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    configure(
+        db_path=str(tmp_path / "outbox.db"),
+        data_dir=str(tmp_path / "data"),
+        tenant_id="tnt_async",
+    )
+
+    async def boom() -> None:
+        raise ValueError("async boom")
+
+    fn = wrap(intent="Async", action="async.boom", fn=boom)
+
+    def fail_record(**_kwargs: object) -> None:
+        raise RuntimeError("outbox unavailable")
+
+    monkeypatch.setattr(
+        "intentproof.instrumentation._record_execution", fail_record
+    )
+
+    with pytest.raises(ValueError, match="async boom") as exc_info:
+        asyncio.run(fn())
+
+    assert isinstance(exc_info.value.__cause__, RuntimeError)
+
+
+def test_async_wrap_record_failure_without_app_error_raises(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    configure(
+        db_path=str(tmp_path / "outbox.db"),
+        data_dir=str(tmp_path / "data"),
+        tenant_id="tnt_async",
+    )
+
+    async def ok() -> int:
+        return 1
+
+    fn = wrap(intent="Async", action="async.ok", fn=ok)
+
+    def fail_record(**_kwargs: object) -> None:
+        raise RuntimeError("outbox unavailable")
+
+    monkeypatch.setattr(
+        "intentproof.instrumentation._record_execution", fail_record
+    )
+
+    with pytest.raises(RuntimeError, match="outbox unavailable"):
+        asyncio.run(fn())

tests/test_canon.py

@@ -2,8 +2,17 @@
 
 import os
 import unittest
+from unittest.mock import patch
 
-from intentproof.canon import canonicalize
+from intentproof.canon import (
+    _CanonObject,
+    _encode_int,
+    _encode_object,
+    _format_es6,
+    _integer_to_scientific,
+    _to_shortest_scientific,
+    canonicalize,
+)
 
 
 # Ported from intentproof-spec/conformance/jcs_vectors.ts
@@ -191,6 +200,74 @@ def test_rejects_unclosed_object(self):
             canonicalize('{')
 
 
+class TestLargeIntegers(unittest.TestCase):
+    def test_safe_integer_uses_decimal(self):
+        self.assertEqual(canonicalize(9007199254740992), "9007199254740992")
+
+    def test_out_of_range_integer_raises(self):
+        with self.assertRaises(ValueError):
+            canonicalize(10**400)
+
+    def test_integer_overflow_on_float_conversion_raises(self):
+        with patch("intentproof.canon.float", side_effect=OverflowError):
+            with self.assertRaises(ValueError):
+                canonicalize(10**400)
+
+
+class TestUnsupportedTypes(unittest.TestCase):
+    def test_rejects_non_string_object_keys(self):
+        with self.assertRaises(TypeError):
+            canonicalize({1: "x"})
+
+    def test_rejects_unsupported_values(self):
+        with self.assertRaises(TypeError):
+            canonicalize(object())
+
+    def test_rejects_unsupported_nested_values(self):
+        with self.assertRaises(TypeError):
+            canonicalize([object()])
+
+
+class TestFloatEdgeBranches(unittest.TestCase):
+    def test_zero_float(self):
+        self.assertEqual(canonicalize(0.0), "0")
+
+    def test_negative_zero_string(self):
+        self.assertEqual(canonicalize("-0"), "0")
+
+    def test_trailing_zero_decimal_string(self):
+        self.assertEqual(canonicalize("5.0"), "5")
+
+    def test_integer_to_scientific_zero_strings(self):
+        self.assertEqual(_integer_to_scientific("0"), ("0", 0))
+        self.assertEqual(_integer_to_scientific("000"), ("0", 0))
+
+    def test_to_shortest_scientific_without_dot_in_repr(self):
+        with patch("intentproof.canon.repr", return_value="42"):
+            self.assertEqual(_to_shortest_scientific(42.0), ("42", 1))
+
+    def test_format_es6_when_digits_round_to_zero(self):
+        with patch(
+            "intentproof.canon._to_shortest_scientific", return_value=("0", 0)
+        ):
+            self.assertEqual(_format_es6(1.0), "0")
+
+    def test_large_integer_beyond_safe_range_uses_float_formatting(self):
+        value = 2**53 + 1
+        self.assertEqual(canonicalize(value), _encode_int(value))
+
+    def test_encode_int_rejects_infinite_float_conversion(self):
+        with patch("intentproof.canon.float", return_value=float("inf")):
+            with self.assertRaises(ValueError):
+                _encode_int(2**60)
+
+    def test_encode_object_rejects_unsupported_nested_type(self):
+        obj = _CanonObject([("key", 1)])
+        obj.values["key"] = frozenset({1})
+        with self.assertRaises(TypeError):
+            _encode_object(obj)
+
+
 class TestPolicyBodyCrossCheck(unittest.TestCase):
     def test_byte_equality_with_go_fixture(self):
         body = {