chore(capture): v1 compliance adapter + CI job (capture v1, 5/6)

.github/workflows/sdk-compliance.yml

@@ -13,9 +13,21 @@ on:
 
 jobs:
   compliance:
-    name: PostHog SDK compliance tests
-    uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@85e2901ea3260a28e07a83086d59b4fb4dfc814f
+    name: PostHog SDK compliance tests (capture v0)
+    # Pinned to the 0.8.0 tag (commit be8b8d5) of the reusable workflow + harness
+    # image so v0/v1 runs are reproducible and pick up the capture-v1 suites.
+    uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@be8b8d5a3f94a249659844e94832e874f049c1e4
     with:
       adapter-dockerfile: "sdk_compliance_adapter/Dockerfile"
       adapter-context: "."
-      test-harness-version: "main-85e2901@sha256:4c8eac34e7ff66554a2c6947788c0a42b82456bc949c03bd8f6b9a10bef23ef5"
+      test-harness-version: "0.8.0"
+      report-name: "sdk-compliance-report-v0"
+
+  compliance-v1:
+    name: PostHog SDK compliance tests (capture v1)
+    uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@be8b8d5a3f94a249659844e94832e874f049c1e4
+    with:
+      adapter-dockerfile: "sdk_compliance_adapter/Dockerfile.v1"
+      adapter-context: "."
+      test-harness-version: "0.8.0"
+      report-name: "sdk-compliance-report-v1"

posthog/request.py

@@ -227,6 +227,7 @@ def post(
     gzip: bool = False,
     timeout: int = 15,
     session: Optional[requests.Session] = None,
+    api_key_field: str = "api_key",
     **kwargs,
 ) -> requests.Response:
     """Post the `kwargs` to the API"""
@@ -235,7 +236,7 @@ def post(
     body["sent_at"] = datetime.now(tz=timezone.utc).isoformat()
     trimmed_host = remove_trailing_slash(normalize_host(host))
     url = trimmed_host + cast(str, path)
-    body["api_key"] = api_key
+    body[api_key_field] = api_key
     data: str | bytes = json.dumps(body, cls=DatetimeSerializer)
     log.debug("making request: %s to url: %s", data, url)
     headers = {"Content-Type": "application/json", "User-Agent": USER_AGENT}
@@ -321,6 +322,7 @@ def flags(
         gzip,
         timeout,
         session=_get_flags_session(),
+        api_key_field="token",
         **kwargs,
     )
     return _process_response(

posthog/test/test_request.py

@@ -85,6 +85,20 @@ def test_post_sends_snake_case_sent_at(key, expected_present):
     assert (key in data) is expected_present
 
 
+def test_flags_request_uses_token_field_for_project_api_key():
+    mock_response = mock.MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {"featureFlags": {}}
+
+    with mock.patch.object(
+        request_module, "post", return_value=mock_response
+    ) as mock_post:
+        flags(TEST_API_KEY, host="https://test.posthog.com", distinct_id="user_1")
+
+    mock_post.assert_called_once()
+    assert mock_post.call_args.kwargs["api_key_field"] == "token"
+
+
 def test_message_only_debug_logs_include_posthog_prefix():
     mock_response = requests.Response()
     mock_response.status_code = 200

references/public_api_snapshot.txt

@@ -1112,7 +1112,7 @@ function posthog.request.flags(api_key: str, host: Optional[str] = None, gzip: b
 function posthog.request.get(api_key: str, url: str, host: Optional[str] = None, timeout: Optional[int] = None, etag: Optional[str] = None) -> GetResponse
 function posthog.request.is_ai_event(event_name) -> bool
 function posthog.request.normalize_host(host: Optional[str]) -> str
-function posthog.request.post(api_key: str, host: Optional[str] = None, path: Optional[str] = None, gzip: bool = False, timeout: int = 15, session: Optional[requests.Session] = None, **kwargs) -> requests.Response
+function posthog.request.post(api_key: str, host: Optional[str] = None, path: Optional[str] = None, gzip: bool = False, timeout: int = 15, session: Optional[requests.Session] = None, api_key_field: str = 'api_key', **kwargs) -> requests.Response
 function posthog.request.remote_config(personal_api_key: str, project_api_key: str, host: Optional[str] = None, key: str = '', timeout: int = 15) -> Any
 function posthog.request.reset_sessions() -> None
 function posthog.request.set_socket_options(socket_options: Optional[SocketOptions]) -> None

sdk_compliance_adapter/CONTRIBUTING.md

@@ -35,7 +35,7 @@ docker run -d --name sdk-adapter --network test-network -p 8080:8080 posthog-pyt
 docker run --rm \
   --name test-harness \
   --network test-network \
-  ghcr.io/posthog/sdk-test-harness:latest \
+  ghcr.io/posthog/sdk-test-harness:0.8.0 \
   run --adapter-url http://sdk-adapter:8080 --mock-url http://test-harness:8081
 
 # Cleanup

sdk_compliance_adapter/Dockerfile.v1

@@ -0,0 +1,25 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# Copy the SDK source code
+COPY posthog/ /app/sdk/posthog/
+COPY setup.py pyproject.toml README.md LICENSE /app/sdk/
+
+# Install the SDK from source
+RUN cd /app/sdk && pip install --no-cache-dir -e .
+
+# Install adapter dependencies
+RUN pip install --no-cache-dir flask python-dateutil
+
+# Copy adapter code
+COPY sdk_compliance_adapter/adapter.py /app/adapter.py
+
+# Select the capture-v1 protocol; same adapter code, different runtime mode.
+ENV CAPTURE_MODE=v1
+
+# Expose port 8080
+EXPOSE 8080
+
+# Run the adapter
+CMD ["python", "/app/adapter.py"]

sdk_compliance_adapter/adapter.py

@@ -14,18 +14,30 @@
 from flask import Flask, jsonify, request
 
 from posthog import Client
+from posthog.capture_compression import CaptureCompression
+from posthog.capture_v1 import post_v1 as original_post_v1
 from posthog.request import EVENTS_ENDPOINT
 from posthog.request import batch_post as original_batch_post
 from posthog.version import VERSION
 
 # Configure logging
 logging.basicConfig(
-    level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    level=logging.WARNING, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 )
 logger = logging.getLogger(__name__)
 
 app = Flask(__name__)
 
+# Selects which capture protocol this adapter process speaks. Baked at build
+# time via the CAPTURE_MODE env var ("v1" => capture-v1, anything else => legacy
+# v0), mirroring the v0/v1 Dockerfile split. One process speaks one mode and
+# advertises it via /health capabilities.
+CAPTURE_MODE = os.environ.get("CAPTURE_MODE", "")
+
+
+def is_v1() -> bool:
+    return CAPTURE_MODE == "v1"
+
 
 class RequestInfo:
     """Information about an HTTP request made by the SDK"""
@@ -132,6 +144,33 @@ def record_request(self, status_code: int, batch: List[Dict], batch_id: str):
                 if retry_attempt > 0:
                     self.total_retries += 1
 
+    def record_request_v1(
+        self, status_code: int, batch: List[Dict], attempt: int, terminal_count: int
+    ):
+        """Record a capture-v1 HTTP attempt.
+
+        Unlike v0, the retry attempt is carried on the request (PostHog-Attempt,
+        1-based) rather than inferred from a batch id, and a 2xx no longer means
+        the whole batch was accepted — only events with a terminal (non-"retry")
+        per-event result count as sent.
+        """
+        with self.lock:
+            uuid_list = [event.get("uuid", "") for event in batch]
+            self.requests_made.append(
+                RequestInfo(
+                    timestamp_ms=int(time.time() * 1000),
+                    status_code=status_code,
+                    retry_attempt=attempt - 1,
+                    event_count=len(batch),
+                    uuid_list=uuid_list,
+                )
+            )
+            if attempt > 1:
+                self.total_retries += 1
+            if 200 <= status_code < 300:
+                self.total_events_sent += terminal_count
+                self.pending_events = max(0, self.pending_events - terminal_count)
+
     def record_error(self, error: str):
         """Record an error"""
         with self.lock:
@@ -188,6 +227,60 @@ def patched_batch_post(
         raise
 
 
+def patched_post_v1(
+    api_key: str,
+    host: Optional[str],
+    batch_body: Dict,
+    *,
+    attempt: int,
+    request_id: str,
+    compression: CaptureCompression = CaptureCompression.NONE,
+    timeout: int = 15,
+    session: Any = None,
+):
+    """Patched version of post_v1 that records requests for /state assertions.
+
+    Mirrors the legacy `patched_batch_post`, but reads the retry attempt from the
+    call (1-based) and counts only terminal per-event results as sent.
+    """
+    batch = batch_body.get("batch", [])
+    try:
+        response = original_post_v1(
+            api_key,
+            host,
+            batch_body,
+            attempt=attempt,
+            request_id=request_id,
+            compression=compression,
+            timeout=timeout,
+            session=session,
+        )
+    except Exception as e:
+        status_code = getattr(e, "status", 0)
+        state.record_request_v1(
+            status_code if isinstance(status_code, int) else 0, batch, attempt, 0
+        )
+        state.record_error(str(e))
+        raise
+
+    terminal = 0
+    status = response.status_code
+    if 200 <= status < 300:
+        try:
+            results = response.json().get("results", {})
+            # Mirror send_v1_batch: only a non-retry directive is terminal. A
+            # missing/null `result` is not counted as sent.
+            terminal = sum(
+                1
+                for r in results.values()
+                if (r or {}).get("result") not in (None, "retry")
+            )
+        except Exception:
+            terminal = 0
+    state.record_request_v1(status, batch, attempt, terminal)
+    return response
+
+
 # Monkey-patch the batch_post function
 import posthog.request  # noqa: E402
 
@@ -198,16 +291,26 @@ def patched_batch_post(
 
 posthog.consumer.batch_post = patched_batch_post
 
+# Patch the capture-v1 submitter. `send_v1_batch` resolves `post_v1` as a module
+# global at call time, so patching it here covers both the async consumer and the
+# sync client paths.
+import posthog.capture_v1  # noqa: E402
+
+posthog.capture_v1.post_v1 = patched_post_v1
+
 
 @app.route("/health", methods=["GET"])
 def health():
     """Health check endpoint"""
+    capabilities = (
+        ["capture_v1", "encoding_gzip"] if is_v1() else ["capture_v0", "encoding_gzip"]
+    )
     return jsonify(
         {
             "sdk_name": "posthog-python",
             "sdk_version": VERSION,
             "adapter_version": "1.0.0",
-            "capabilities": ["capture_v0", "encoding_gzip"],
+            "capabilities": capabilities,
         }
     )
 
@@ -225,9 +328,14 @@ def init():
         api_key = data.get("api_key")
         host = data.get("host")
         flush_at = data.get("flush_at", 100)
-        flush_interval_ms = data.get("flush_interval_ms", 5000)
+        flush_interval_ms = data.get("flush_interval_ms", 500)
         max_retries = data.get("max_retries", 3)
         enable_compression = data.get("enable_compression", False)
+        # Compliance tests assert the request-level default when callers omit
+        # disable_geoip, so the adapter default keeps geoip-enabled /flags
+        # requests while still allowing per-call overrides.
+        disable_geoip = data.get("disable_geoip", False)
+        historical_migration = data.get("historical_migration", False)
 
         if not api_key:
             return jsonify({"error": "api_key is required"}), 400
@@ -237,6 +345,9 @@ def init():
         # Convert flush_interval from ms to seconds
         flush_interval = flush_interval_ms / 1000.0
 
+        # One adapter process speaks one capture protocol, selected by CAPTURE_MODE.
+        capture_mode = "v1" if is_v1() else "v0"
+
         # Create client
         client = Client(
             project_api_key=api_key,
@@ -245,19 +356,20 @@ def init():
             flush_interval=flush_interval,
             gzip=enable_compression,
             max_retries=max_retries,
-            debug=True,
-            # Compliance tests assert the request-level default when callers omit
-            # disable_geoip. Configure the adapter to exercise geoip-enabled
-            # /flags requests by default while still allowing per-call overrides.
-            disable_geoip=False,
+            debug=False,
+            disable_geoip=disable_geoip,
+            historical_migration=historical_migration,
+            capture_mode=capture_mode,
         )
 
         state.client = client
 
         logger.info(
             f"Initialized SDK with api_key={api_key[:10]}..., host={host}, "
             f"flush_at={flush_at}, flush_interval={flush_interval}, "
-            f"max_retries={max_retries}, gzip={enable_compression}"
+            f"max_retries={max_retries}, gzip={enable_compression}, "
+            f"capture_mode={capture_mode}, disable_geoip={disable_geoip}, "
+            f"historical_migration={historical_migration}"
         )
 
         return jsonify({"success": True})
@@ -279,12 +391,28 @@ def capture():
         event = data.get("event")
         properties = data.get("properties")
         timestamp = data.get("timestamp")
+        options = data.get("options")
 
         if not distinct_id:
             return jsonify({"error": "distinct_id is required"}), 400
         if not event:
             return jsonify({"error": "event is required"}), 400
 
+        # Fold capture-v1 options back into the magic `$`-prefixed properties the
+        # SDK lifts onto the wire `options` object. Renamed keys mirror the SDK's
+        # sentinel table; unknown keys get a bare `$` prefix. v0 has no wire
+        # options object, so this only applies in v1 mode.
+        if options and is_v1():
+            properties = dict(properties or {})
+            option_to_property = {
+                "cookieless_mode": "$cookieless_mode",
+                "disable_skew_correction": "$ignore_sent_at",
+                "process_person_profile": "$process_person_profile",
+                "product_tour_id": "$product_tour_id",
+            }
+            for key, value in options.items():
+                properties[option_to_property.get(key, "$" + key)] = value
+
         # Capture event
         kwargs = {"distinct_id": distinct_id, "properties": properties}
         if timestamp:

sdk_compliance_adapter/docker-compose.yml

@@ -1,7 +1,7 @@
 version: "3.8"
 
 services:
-  # PostHog Python SDK adapter
+  # PostHog Python SDK adapter (capture v0)
   sdk-adapter:
     build:
       context: ..
@@ -11,9 +11,19 @@ services:
     networks:
       - test-network
 
+  # PostHog Python SDK adapter (capture v1)
+  sdk-adapter-v1:
+    build:
+      context: ..
+      dockerfile: sdk_compliance_adapter/Dockerfile.v1
+    ports:
+      - "8082:8080"
+    networks:
+      - test-network
+
   # Test harness
   test-harness:
-    image: ghcr.io/posthog/sdk-test-harness:latest
+    image: ghcr.io/posthog/sdk-test-harness:0.8.0
     command: ["run", "--adapter-url", "http://sdk-adapter:8080", "--mock-url", "http://test-harness:8081"]
     networks:
       - test-network