fix(canonical-formats): address Argus review follow-ups + nits on #841

Argus APPROVED #841 and noted five non-blocking follow-ups + four nits.
This commit addresses the ones that sharpen wire correctness or close
real attack surfaces.

## agent_url canonicalisation in v1 inbound lookup

``find_declaration_by_v1_format_id`` now canonicalises both buyer and
seller ``agent_url`` per RFC 3986 §6 before comparing: scheme
lowercased, host lowercased, default port (443/80) stripped. Without
this a seller publishing ``https://Creative.AdContextProtocol.org``
would silently miss-match a buyer's
``https://creative.adcontextprotocol.org`` and the SDK would return a
wrongful ``UNSUPPORTED_FEATURE``. Argus flagged this as the only
documented MUST divergence — ``core/format-id.json:11`` is normative.

## ANSI / control-char scrub in advisory identifier echo

``_echo_identifier`` now escapes every C0 control char (incl. ``\\n``
``\\r`` ``\\x1b``), the C1 range, and Unicode LS/PS line separators
to a literal ``\\u<hex>`` form before applying the 128-char cap. A
seller publishing ``product_id`` with ``\\n`` or ``\\x1b[`` previously
round-tripped that string into ``errors[].details.product_id`` and
could forge log lines or manipulate ANSI-aware operator tooling.

## sdk_id alignment between installed + dev

Hardcodes ``_SDK_DIST_NAME = "adcontextprotocol-adcp-python"`` (the
spec example form). Installed wheels publish the dist name as plain
``adcp``; the prior code emitted ``adcp@<version>`` from a wheel install
vs ``adcontextprotocol-adcp-python@0.0.0-dev`` from a checkout,
breaking the multi-hop ``(code, field, sdk_id)`` dedup contract in
``core/error.json`` for the same SDK.

## Experimental markers + nits

- Added ``.. note:: Experimental`` to ``glob_match``,
  ``structural_match``, ``load_default_registry``.
- Removed duplicate ``NotificationConfig`` in ``adcp.types/__all__``.
- Removed ``_echo_identifier`` from ``advisory.py``'s ``__all__``.
- Two new tests pull ``SDK_ID`` through the documented public path
  (``from adcp.canonical_formats import SDK_ID``).

## Tests

5117 passed locally. New coverage: agent_url host-casefolding +
default-port stripping; newline / ANSI / Unicode line-separator
scrubbing; ``SDK_ID`` public-import + canonical-prefix invariants.

## Deferred per Argus's own suggestion

- ``_versions_overlap`` fail-loud on ``~`` / ``^`` — latent until
  #741 half 2 actually consumes the registry on the inbound path.
- ``_walk_for_credential_keys`` cycle guard — not wire-reachable.

Refs: #741, #841

Author: bokelley

MIGRATION_v5_to_v6.md

@@ -189,10 +189,13 @@ if decl.format_kind is CanonicalFormatKind.image:
 **SDK-source advisory provenance:**
 
 All advisories emitted by the projection layer carry `source="sdk"` and
-`sdk_id="<dist_name>@<version>"` where `<dist_name>` is read from the
-installed distribution metadata (`importlib.metadata.metadata("adcp")["Name"]`).
-Adopters relying on a particular `sdk_id` for multi-hop dedup should
-pin to a specific SDK release rather than parsing the string.
+`sdk_id="adcontextprotocol-adcp-python@<version>"`. The distribution-name
+prefix is fixed (independent of `pyproject.toml`'s `[project].name`)
+so wheel installs and dev installs emit the same attribution string —
+the multi-hop `(code, field, sdk_id)` dedup contract in `core/error.json`
+keys on this and would corrupt under drift. Adopters relying on a
+particular `sdk_id` for multi-hop dedup should pin to a specific SDK
+release rather than parsing the string.
 
 **Not yet shipped (later beta increments):** v1 → v2 reverse projection,
 `pixel_tracker` bidirectional contract, the 14 reference fixtures and

src/adcp/canonical_formats/advisory.py

@@ -18,7 +18,6 @@
 
 from functools import lru_cache
 from importlib.metadata import PackageNotFoundError
-from importlib.metadata import metadata as _pkg_metadata
 from importlib.metadata import version as _pkg_version
 from typing import Any, TypeAlias
 
@@ -37,38 +36,75 @@
 SdkAdvisory: TypeAlias = Error
 
 
+# Canonical distribution name for the wire ``sdk_id``. Hardcoded (rather
+# than read from ``pyproject.toml``'s ``[project].name``) so installed
+# and dev builds emit the same audit-trail attribution. The
+# ``core/error.json`` dedup contract keys on ``(code, field, sdk_id)``;
+# drift here corrupts multi-hop deduplication. Installed wheels publish
+# the PyPI distribution as ``adcp``; the fully-qualified
+# ``adcontextprotocol-adcp-python`` form is what the spec example uses
+# and what cross-SDK consumers expect.
+_SDK_DIST_NAME: str = "adcontextprotocol-adcp-python"
+
+
 @lru_cache(maxsize=1)
 def _resolve_sdk_id() -> str:
     """Return the wire-format ``sdk_id`` for this SDK build.
 
     Format per ``core/error.json``: ``<sdk_package_name>@<version>``.
-    Reads the installed distribution's ``Name`` from the package metadata
-    so the prefix never drifts from the PyPI distribution name — the
-    audit-trail dedup key depends on it.
+    The package-name prefix is fixed (``_SDK_DIST_NAME``) so installed
+    and dev builds emit the same attribution; only the version
+    component varies between them. Without this, dev installs would
+    emit a different ``sdk_id`` from wheel installs and break the
+    multi-hop dedup contract for the same SDK.
 
     Cached because the resolution is process-stable: the package
     metadata doesn't change at runtime. Lazy (computed on first call)
     so setuptools-scm-style late version resolution still works.
-    Falls back to a development marker when the package isn't
-    installed (e.g., running directly out of a checkout without
-    ``pip install -e``).
+    Falls back to a ``0.0.0-dev`` version marker when the package
+    isn't installed.
     """
     try:
-        dist_name = _pkg_metadata("adcp")["Name"]
         v = _pkg_version("adcp")
     except PackageNotFoundError:
-        dist_name = "adcontextprotocol-adcp-python"
         v = "0.0.0-dev"
-    return f"{dist_name}@{v}"
+    return f"{_SDK_DIST_NAME}@{v}"
 
 
 def _echo_identifier(value: str | None) -> str | None:
-    """Cap seller-controlled identifiers before echoing into advisory details."""
+    """Cap + scrub seller-controlled identifiers before echoing into advisory details.
+
+    Two defenses applied in order:
+
+    1. **Control-character scrub** — replaces every C0 control char
+       (``\\x00``-``\\x1f``), the C1 range (``\\x7f``-``\\x9f``), and
+       all Unicode line separators with a literal ``"\\u<hex>"``
+       escape. A seller publishing a ``product_id`` containing ``\\n``
+       or ``\\x1b[`` would otherwise round-trip into
+       ``errors[].details.product_id``, forging log lines or
+       triggering ANSI escape sequences in operator tooling that
+       prints one advisory per line.
+
+    2. **Length cap** — at 128 chars (after escaping), so a malformed
+       seller identifier cannot grow the multi-hop ``errors[]`` array
+       unbounded into the idempotency replay cache.
+
+    Returns ``None`` for ``None`` input (the explicit absent-product case).
+    """
     if value is None:
         return None
-    if len(value) <= _MAX_ECHOED_IDENTIFIER_LEN:
-        return value
-    return value[:_MAX_ECHOED_IDENTIFIER_LEN] + "…[truncated]"
+    scrubbed_chars: list[str] = []
+    for ch in value:
+        cp = ord(ch)
+        # C0 (incl. \t, \n, \r) + DEL + C1 + LS/PS line separators.
+        if cp < 0x20 or 0x7F <= cp <= 0x9F or ch in ("
", "
"):
+            scrubbed_chars.append(f"\\u{cp:04x}")
+        else:
+            scrubbed_chars.append(ch)
+    scrubbed = "".join(scrubbed_chars)
+    if len(scrubbed) <= _MAX_ECHOED_IDENTIFIER_LEN:
+        return scrubbed
+    return scrubbed[:_MAX_ECHOED_IDENTIFIER_LEN] + "…[truncated]"
 
 
 def __getattr__(name: str) -> Any:
@@ -124,9 +160,10 @@ def make_sdk_advisory(
 
 # ``SDK_ID`` is resolved lazily via module ``__getattr__`` (above) — listed
 # here so the public surface is documented + introspectable via ``dir()``.
+# ``_echo_identifier`` and ``_resolve_sdk_id`` are private helpers; not
+# part of ``__all__``.
 __all__ = [  # noqa: F822 — SDK_ID provided via module __getattr__
     "SDK_ID",
     "SdkAdvisory",
-    "_echo_identifier",
     "make_sdk_advisory",
 ]

src/adcp/canonical_formats/format_options.py

@@ -24,9 +24,44 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
+from urllib.parse import urlsplit, urlunsplit
 
 from adcp.types import CanonicalFormatKind, Error, FormatId, ProductFormatDeclaration
 
+# Default ports per RFC 3986 §3.2.3 — stripped during canonicalization
+# so ``https://x.example:443`` matches ``https://x.example``.
+_DEFAULT_PORTS: dict[str, int] = {"http": 80, "https": 443}
+
+
+def _canonicalize_agent_url(raw: str) -> str:
+    """Return ``raw`` with scheme + host lowercased and default port stripped.
+
+    Per ``core/format-id.json`` (normative): callers MUST canonicalize
+    ``agent_url`` before comparing two ``FormatId`` values for identity.
+    Pydantic's ``AnyUrl`` does trailing-slash normalization but not
+    RFC 3986 §6 host-casefolding or default-port stripping — a seller
+    publishing ``"https://Creative.AdContextProtocol.org"`` would
+    silently miss-match a buyer's ``"https://creative.adcontextprotocol.org"``
+    without this step.
+
+    Non-throwing: malformed inputs round-trip as-is. The lookup is a
+    closed-set match, not a security check; we don't want to reject
+    here, just normalize what we can.
+    """
+    try:
+        parts = urlsplit(raw)
+    except ValueError:
+        return raw
+    if not parts.scheme or not parts.hostname:
+        return raw
+    scheme = parts.scheme.lower()
+    host = parts.hostname.lower()
+    port = parts.port
+    if port is not None and port == _DEFAULT_PORTS.get(scheme):
+        port = None
+    netloc = host if port is None else f"{host}:{port}"
+    return urlunsplit((scheme, netloc, parts.path, parts.query, ""))
+
 
 class FormatKindNotInClosedSetError(ValueError):
     """Raised when a ``format_kind`` is not in the product's ``format_options[]``.
@@ -172,12 +207,13 @@ def find_declaration_by_v1_format_id(
         should be rejected with ``UNSUPPORTED_FEATURE`` — the v1
         ``format_id`` is not a recognised entry for this product.
     """
-    target_url = str(format_id.agent_url)
+    target_url = _canonicalize_agent_url(str(format_id.agent_url))
     target_id = format_id.id
     for decl in format_options:
         refs = decl.v1_format_ref or []
         for ref in refs:
-            if str(ref.agent_url) == target_url and ref.id == target_id:
+            ref_url = _canonicalize_agent_url(str(ref.agent_url))
+            if ref_url == target_url and ref.id == target_id:
                 return decl
     return None
 

src/adcp/canonical_formats/registry.py

@@ -125,6 +125,13 @@ def load_default_registry() -> V1V2CanonicalFormatMappingRegistry:
     cached per process (the registry is immutable for a given SDK
     build, keyed by ``ADCP_VERSION``).
 
+    .. note::
+       **Experimental.** The registry is consulted only on the v1 → v2
+       inbound path (lands in #741 part 2). Adopters using this helper
+       to inspect the bundled mappings SHOULD pin to the SDK release
+       they integrate against; the return shape may sharpen when the
+       inbound consumer lands.
+
     Raises:
         RegistryLoadError: when the bundle is missing, malformed JSON,
             or fails schema validation.
@@ -143,6 +150,13 @@ def glob_match(value: str, pattern: str) -> bool:
     Treats ``*`` as a permissive wildcard (any chars including ``_``).
     Other regex metacharacters are escaped — the pattern language is
     glob, not regex.
+
+    .. note::
+       **Experimental.** This helper is unused on the v2 → v1 path (the
+       registry is consulted only on the v1 → v2 inbound path, which
+       lands in #741's second PR). The signature may shift when that
+       path consumes it; adopters SHOULD pin to the SDK release they
+       integrate against.
     """
     if pattern == "*":
         return True
@@ -283,6 +297,10 @@ def structural_match(
     Returns:
         ``True`` iff every constraint declared on ``pattern`` is satisfied
         by the v1 format's structural shape.
+
+    .. note::
+       **Experimental.** Same caveat as :func:`glob_match` — the v1 → v2
+       inbound consumer lands in the second half of #741.
     """
     if hasattr(pattern, "model_dump"):
         p = pattern.model_dump(exclude_none=True)

src/adcp/types/__init__.py

@@ -1197,7 +1197,6 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "AvailableMetric",
     "NotificationConfig",
     "PushNotificationConfig",
-    "NotificationConfig",
     "ReportingCapabilities",
     "WholesaleFeedEvent",
     "WholesaleFeedWebhook",

tests/fixtures/public_api_snapshot.json

@@ -776,7 +776,6 @@
     "MetricType",
     "ModuleType",
     "NotificationConfig",
-    "NotificationConfig",
     "NotificationType",
     "OfferPrice",
     "Offering",

tests/test_canonical_formats_options.py

@@ -202,6 +202,58 @@ def test_v1_inbound_lookup_distinguishes_by_agent_url() -> None:
     assert find_declaration_by_v1_format_id(other_seller, [decl]) is None
 
 
+def test_v1_inbound_lookup_canonicalises_agent_url_host_case() -> None:
+    """RFC 3986 §6 host-casefolding: ``Creative.X`` must match ``creative.x``.
+
+    Without canonicalization a seller publishing
+    ``https://Creative.AdContextProtocol.org`` would silently miss-match
+    a buyer's ``https://creative.adcontextprotocol.org`` and the SDK
+    would return a wrongful ``UNSUPPORTED_FEATURE``.
+    """
+    from adcp.canonical_formats import find_declaration_by_v1_format_id
+    from adcp.types import FormatId
+
+    seller_decl = ProductFormatDeclaration(
+        format_kind=CanonicalFormatKind.image,
+        params={},
+        v1_format_ref=[
+            FormatId(
+                agent_url="https://Creative.AdContextProtocol.org",
+                id="display_300x250_image",
+            ),
+        ],
+    )
+    buyer_ref = FormatId(
+        agent_url="https://creative.adcontextprotocol.org",
+        id="display_300x250_image",
+    )
+
+    assert find_declaration_by_v1_format_id(buyer_ref, [seller_decl]) is seller_decl
+
+
+def test_v1_inbound_lookup_canonicalises_default_port() -> None:
+    """Default-port stripping: ``https://x.example:443`` matches ``https://x.example``."""
+    from adcp.canonical_formats import find_declaration_by_v1_format_id
+    from adcp.types import FormatId
+
+    seller_decl = ProductFormatDeclaration(
+        format_kind=CanonicalFormatKind.image,
+        params={},
+        v1_format_ref=[
+            FormatId(
+                agent_url="https://creative.adcontextprotocol.org:443",
+                id="display_300x250_image",
+            ),
+        ],
+    )
+    buyer_ref = FormatId(
+        agent_url="https://creative.adcontextprotocol.org",
+        id="display_300x250_image",
+    )
+
+    assert find_declaration_by_v1_format_id(buyer_ref, [seller_decl]) is seller_decl
+
+
 def test_v1_inbound_lookup_with_no_refs_returns_none() -> None:
     from adcp.canonical_formats import find_declaration_by_v1_format_id
     from adcp.types import FormatId

tests/test_canonical_formats_projection.py

@@ -302,3 +302,36 @@ def test_advisory_product_id_is_truncated() -> None:
     assert echoed.endswith("…[truncated]")
     # The literal cap (128) plus the truncation marker.
     assert echoed.startswith("x" * 128)
+
+
+def test_advisory_product_id_scrubs_newlines() -> None:
+    """Newline injection attempts are escaped so operator log emitters
+    don't see forged lines from seller-controlled identifiers."""
+    decl = ProductFormatDeclaration(format_kind=CanonicalFormatKind.video_vast, params={})
+
+    result = project_declaration_to_v1(decl, product_id="prod_alpha\nFAKE LOG LINE\nprod_omega")
+
+    echoed = result.advisories[0].details["product_id"]
+    assert "\n" not in echoed
+    assert "\\u000a" in echoed
+
+
+def test_advisory_product_id_scrubs_ansi_escape() -> None:
+    """ANSI ``\\x1b[`` escape sequences are neutralised before echoing."""
+    decl = ProductFormatDeclaration(format_kind=CanonicalFormatKind.video_vast, params={})
+
+    result = project_declaration_to_v1(decl, product_id="prod\x1b[31mRED\x1b[0m")
+
+    echoed = result.advisories[0].details["product_id"]
+    assert "\x1b" not in echoed
+    assert "\\u001b" in echoed
+
+
+def test_advisory_product_id_scrubs_unicode_line_separator() -> None:
+    """Unicode LS/PS line separators escape too — they break naive line splitters."""
+    decl = ProductFormatDeclaration(format_kind=CanonicalFormatKind.video_vast, params={})
+
+    result = project_declaration_to_v1(decl, product_id="prod
injected")
+
+    echoed = result.advisories[0].details["product_id"]
+    assert "
" not in echoed

tests/test_canonical_formats_public_api.py

@@ -72,6 +72,28 @@ def test_canonical_format_kind_has_13_values() -> None:
     assert len(CanonicalFormatKind) == 13
 
 
+def test_sdk_id_reachable_via_public_package_path() -> None:
+    """``SDK_ID`` is documented as importable from :mod:`adcp.canonical_formats`.
+
+    It's resolved lazily via module ``__getattr__`` so this also
+    exercises that dispatch path against the documented import.
+    """
+    from adcp.canonical_formats import SDK_ID
+
+    assert isinstance(SDK_ID, str)
+    assert SDK_ID.startswith("adcontextprotocol-adcp-python@")
+
+
+def test_sdk_id_uses_canonical_distribution_prefix() -> None:
+    """Dev installs and wheel installs MUST emit the same ``sdk_id`` prefix
+    so the multi-hop ``(code, field, sdk_id)`` dedup contract holds across
+    deployment shapes."""
+    from adcp.canonical_formats import SDK_ID
+
+    prefix, _, _ = SDK_ID.partition("@")
+    assert prefix == "adcontextprotocol-adcp-python"
+
+
 def test_canonical_kind_values_match_spec() -> None:
     """The 13 wire values are the canonical-formats vocabulary; lock them in."""
     from adcp.types import CanonicalFormatKind