feat(adcp): support ADCP_BASE_URL env override in sync_schemas.py (#285)

* feat(adcp): support ADCP_BASE_URL env override in sync_schemas.py

Matches the JS SDK pattern so cross-SDK CI can set ADCP_BASE_URL once
to point both clients at a fixture CDN or pre-release bundle server.
Trailing slashes on the env var value are stripped to prevent a doubled
/protocol path segment. Adds a print when the override is active.

Closes #284

https://claude.ai/code/session_01KXfYPHYKr3R6EDqUbbo9ty

* fix(test): guard test_default_value against ADCP_BASE_URL in env; use ! sigil

test_default_value was relying on the already-loaded _mod which bakes in
whatever ADCP_BASE_URL was when pytest started — exactly the CI case this
feature enables. Switched to the same fresh-module-load pattern the other
two env-override tests already use, with monkeypatch.delenv to guarantee
the default code path. Also align the override print line with the
established ! sigil convention.

https://claude.ai/code/session_01KXfYPHYKr3R6EDqUbbo9ty

* feat(adcp): reject ADCP_BASE_URL values ending in /protocol

Acting on python-expert review of this PR:

The script appends '/protocol' itself, so an override like
'https://fixture.example.com/protocol' would silently produce
'/protocol/protocol' and 404 against any sensible CDN. Convert that
into a loud ValueError at module import so the typo surfaces
immediately rather than during the fetch.

Two new tests cover the failure mode (suffix + suffix-with-trailing-
slash). The existing trailing-slash-strip test still covers the
common case where users add an accidental trailing slash on the host
itself.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: bokelley

scripts/sync_schemas.py

@@ -16,6 +16,12 @@
 The target version comes from `src/adcp/ADCP_VERSION`. If that version's
 bundle is not published, sync falls back to `latest.tgz` (the dev snapshot).
 
+Environment variables:
+  ADCP_BASE_URL       Override the protocol host (default: https://adcontextprotocol.org).
+                      Set to point at a fixture CDN for cross-SDK CI or pre-release testing.
+                      Trailing slashes are stripped automatically. Do NOT include "/protocol".
+  ADCP_SKIP_SIGNATURE Set to "1" to skip Sigstore verification and trust the SHA-256 only.
+
 Usage:
     python scripts/sync_schemas.py              # sync schemas + skills
     python scripts/sync_schemas.py --no-skills  # schemas only (e.g. drift checks)
@@ -41,7 +47,17 @@
 CACHE_DIR = REPO_ROOT / "schemas" / "cache"
 SKILLS_DIR = REPO_ROOT / "skills"
 VERSION_FILE = REPO_ROOT / "src" / "adcp" / "ADCP_VERSION"
-BUNDLE_BASE_URL = "https://adcontextprotocol.org/protocol"
+_ADCP_BASE = os.environ.get("ADCP_BASE_URL", "https://adcontextprotocol.org").rstrip("/")
+# Reject overrides ending in /protocol — appending our own /protocol below
+# would silently produce //protocol and 404 against any sensible CDN. Fail
+# loud at module import so the typo surfaces immediately.
+if _ADCP_BASE.endswith("/protocol"):
+    raise ValueError(
+        f"ADCP_BASE_URL={_ADCP_BASE!r} ends with '/protocol'. The script "
+        "appends '/protocol' itself — pass only the protocol host "
+        "(e.g. https://adcontextprotocol.org)."
+    )
+BUNDLE_BASE_URL = _ADCP_BASE + "/protocol"
 USER_AGENT = "adcp-python-sdk/3.0"
 
 # Sigstore keyless verification identity. Must match the upstream release
@@ -118,9 +134,7 @@ def fetch_signature_sidecars(version: str) -> tuple[bytes | None, bytes | None]:
     return sig, crt
 
 
-def verify_cosign_signature(
-    tgz_bytes: bytes, sig_bytes: bytes, crt_bytes: bytes
-) -> None:
+def verify_cosign_signature(tgz_bytes: bytes, sig_bytes: bytes, crt_bytes: bytes) -> None:
     """Verify the bundle with `cosign verify-blob`.
 
     Raises RuntimeError if cosign is not installed or verification fails.
@@ -193,9 +207,7 @@ def replace_cache_from_bundle(bundle_root: Path) -> int:
     """
     schemas_src = bundle_root / "schemas"
     if not schemas_src.is_dir():
-        raise RuntimeError(
-            f"Bundle missing expected directory: {bundle_root.name}/schemas/"
-        )
+        raise RuntimeError(f"Bundle missing expected directory: {bundle_root.name}/schemas/")
 
     if CACHE_DIR.exists():
         shutil.rmtree(CACHE_DIR)
@@ -274,6 +286,8 @@ def main() -> None:
     args = parser.parse_args()
 
     target_version = get_target_adcp_version()
+    if "ADCP_BASE_URL" in os.environ:
+        print(f"  ! ADCP_BASE_URL override active: {_ADCP_BASE}")
     print(f"Syncing AdCP protocol bundle from {BUNDLE_BASE_URL}...")
     print(f"Target version: {target_version}")
     print(f"Schema cache: {CACHE_DIR}")
@@ -283,9 +297,7 @@ def main() -> None:
 
     try:
         print(f"Fetching {target_version}.tgz + checksum...")
-        tgz_bytes, expected_sha, effective_version = fetch_bundle_with_fallback(
-            target_version
-        )
+        tgz_bytes, expected_sha, effective_version = fetch_bundle_with_fallback(target_version)
     except (HTTPError, URLError) as exc:
         print(f"\n✗ Failed to download bundle: {exc}", file=sys.stderr)
         sys.exit(1)
@@ -311,10 +323,7 @@ def main() -> None:
             sys.exit(1)
 
         if sig_bytes is None or crt_bytes is None:
-            print(
-                f"  ! No Sigstore sidecars for adcp-{effective_version} "
-                "(checksum-only trust)"
-            )
+            print(f"  ! No Sigstore sidecars for adcp-{effective_version} " "(checksum-only trust)")
         else:
             try:
                 verify_cosign_signature(tgz_bytes, sig_bytes, crt_bytes)

tests/test_sync_schemas.py

@@ -124,9 +124,7 @@ def test_no_manifest_returns_zero(self, capsys: pytest.CaptureFixture[str]) -> N
             assert result == 0
         assert "No manifest.json" in capsys.readouterr().out
 
-    def test_empty_skills_list_returns_zero(
-        self, capsys: pytest.CaptureFixture[str]
-    ) -> None:
+    def test_empty_skills_list_returns_zero(self, capsys: pytest.CaptureFixture[str]) -> None:
         with tempfile.TemporaryDirectory() as tmp_str:
             tmp = Path(tmp_str)
             bundle_root = _make_bundle(tmp, manifest_skills=[])
@@ -179,15 +177,11 @@ def test_previous_snapshot_created_on_update(self) -> None:
             existing.mkdir()
             (existing / "SKILL.md").write_text("# Old Brand")
 
-            bundle_root = _make_bundle(
-                tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}}
-            )
+            bundle_root = _make_bundle(tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}})
             sync_skills_from_bundle(bundle_root, skills_dir)
 
             assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == "# New Brand"
-            assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == (
-                "# Old Brand"
-            )
+            assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == ("# Old Brand")
 
     def test_previous_snapshot_replaced_on_second_update(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_str:
@@ -204,14 +198,10 @@ def test_previous_snapshot_replaced_on_second_update(self) -> None:
             existing.mkdir()
             (existing / "SKILL.md").write_text("# Old Brand")
 
-            bundle_root = _make_bundle(
-                tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}}
-            )
+            bundle_root = _make_bundle(tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}})
             sync_skills_from_bundle(bundle_root, skills_dir)
 
-            assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == (
-                "# Old Brand"
-            )
+            assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == ("# Old Brand")
 
     def test_local_only_skill_untouched(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_str:
@@ -263,9 +253,7 @@ def test_path_traversal_slash_in_name_rejected(self) -> None:
             with pytest.raises(RuntimeError, match="Unsafe skill name rejected"):
                 sync_skills_from_bundle(bundle_root, skills_dir)
 
-    def test_non_string_name_skipped(
-        self, capsys: pytest.CaptureFixture[str]
-    ) -> None:
+    def test_non_string_name_skipped(self, capsys: pytest.CaptureFixture[str]) -> None:
         with tempfile.TemporaryDirectory() as tmp_str:
             tmp = Path(tmp_str)
             bundle_root = _make_bundle(
@@ -280,9 +268,7 @@ def test_non_string_name_skipped(
             assert count == 1  # only the valid string entry is synced
         assert "Skipping non-string" in capsys.readouterr().out
 
-    def test_missing_bundle_skill_dir_skipped(
-        self, capsys: pytest.CaptureFixture[str]
-    ) -> None:
+    def test_missing_bundle_skill_dir_skipped(self, capsys: pytest.CaptureFixture[str]) -> None:
         with tempfile.TemporaryDirectory() as tmp_str:
             tmp = Path(tmp_str)
             # Manifest lists a skill that has no corresponding directory in the bundle
@@ -322,9 +308,7 @@ def test_missing_bundle_skill_preserves_existing_dst(
             sync_skills_from_bundle(bundle_root, skills_dir)
 
             # dst must not be touched when src is absent
-            assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == (
-                "# Existing Brand"
-            )
+            assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == ("# Existing Brand")
         assert "missing in bundle" in capsys.readouterr().out
 
     def test_multiple_skills_synced(self) -> None:
@@ -346,3 +330,62 @@ def test_multiple_skills_synced(self) -> None:
             assert (skills_dir / "adcp-brand" / "SKILL.md").exists()
             assert (skills_dir / "adcp-creative" / "SKILL.md").exists()
             assert (skills_dir / "call-adcp-agent" / "SKILL.md").exists()
+
+
+# ---------------------------------------------------------------------------
+# BUNDLE_BASE_URL env override (ADCP_BASE_URL)
+# ---------------------------------------------------------------------------
+
+
+class TestBundleBaseUrl:
+    def test_default_value(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Fresh load with env var absent — guards against shell having ADCP_BASE_URL set.
+        monkeypatch.delenv("ADCP_BASE_URL", raising=False)
+        fresh_spec = importlib.util.spec_from_file_location("sync_schemas_default", _SCRIPT)
+        assert fresh_spec is not None and fresh_spec.loader is not None
+        fresh_mod = importlib.util.module_from_spec(fresh_spec)
+        fresh_spec.loader.exec_module(fresh_mod)  # type: ignore[union-attr]
+        assert fresh_mod.BUNDLE_BASE_URL == "https://adcontextprotocol.org/protocol"
+
+    def test_env_override(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Fresh module load with ADCP_BASE_URL set to verify override is applied.
+        monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com")
+        fresh_spec = importlib.util.spec_from_file_location("sync_schemas_fresh", _SCRIPT)
+        assert fresh_spec is not None and fresh_spec.loader is not None
+        fresh_mod = importlib.util.module_from_spec(fresh_spec)
+        fresh_spec.loader.exec_module(fresh_mod)  # type: ignore[union-attr]
+        assert fresh_mod.BUNDLE_BASE_URL == "https://fixture.example.com/protocol"
+
+    def test_env_override_strips_trailing_slash(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Trailing slash on ADCP_BASE_URL must not produce "//protocol".
+        monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/")
+        fresh_spec = importlib.util.spec_from_file_location("sync_schemas_fresh2", _SCRIPT)
+        assert fresh_spec is not None and fresh_spec.loader is not None
+        fresh_mod = importlib.util.module_from_spec(fresh_spec)
+        fresh_spec.loader.exec_module(fresh_mod)  # type: ignore[union-attr]
+        assert fresh_mod.BUNDLE_BASE_URL == "https://fixture.example.com/protocol"
+        assert "//protocol" not in fresh_mod.BUNDLE_BASE_URL
+
+    def test_env_override_rejects_protocol_suffix(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Override ending in /protocol would double-append. Fail loud at
+        # import rather than silently 404-ing later.
+        monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/protocol")
+        fresh_spec = importlib.util.spec_from_file_location("sync_schemas_protocol_suffix", _SCRIPT)
+        assert fresh_spec is not None and fresh_spec.loader is not None
+        fresh_mod = importlib.util.module_from_spec(fresh_spec)
+        with pytest.raises(ValueError, match="ends with '/protocol'"):
+            fresh_spec.loader.exec_module(fresh_mod)  # type: ignore[union-attr]
+
+    def test_env_override_rejects_protocol_suffix_with_trailing_slash(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Same guard, but with a trailing slash on the override — rstrip
+        # runs first, so the /protocol still trips the check.
+        monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/protocol/")
+        fresh_spec = importlib.util.spec_from_file_location(
+            "sync_schemas_protocol_trailing", _SCRIPT
+        )
+        assert fresh_spec is not None and fresh_spec.loader is not None
+        fresh_mod = importlib.util.module_from_spec(fresh_spec)
+        with pytest.raises(ValueError, match="ends with '/protocol'"):
+            fresh_spec.loader.exec_module(fresh_mod)  # type: ignore[union-attr]