fix(examples-py): validate Pinecone, Pipecat, and Portkey OTel 2.x spans

python/tracing/pinecone/01_upsert_and_query.py

@@ -4,17 +4,19 @@
 import time
 from typing import Any
 
-from respan import Respan, workflow
-
 from _shared import (
     create_pinecone_index,
     create_respan,
     execution_id,
     finish_respan,
+    live_configured,
+    marker,
     print_result,
     response_field,
+    to_jsonable,
     workflow_attributes,
 )
+from respan import Respan, workflow
 
 WORKFLOW_NAME = "pinecone_upsert_and_query_workflow"
 
@@ -33,29 +35,32 @@ def wait_until_fetchable(index: Any, namespace: str, vector_id: str) -> Any:
         if vector_id in (response_field(fetched, "vectors", {}) or {}):
             return fetched
         time.sleep(1)
-    raise TimeoutError(f"Pinecone did not expose {vector_id!r} within {timeout:g}s")
+    raise TimeoutError(
+        f"Pinecone did not expose the example vector within {timeout:g}s"
+    )
 
 
 @workflow(name=WORKFLOW_NAME)
-def run_upsert_and_query(run_id: str) -> dict:
+def run_upsert_and_query(topic: str, top_k: int) -> dict[str, Any]:
     index = create_pinecone_index()
-    namespace = f"respan-example-{run_id}"
+    execution = execution_id()
+    namespace = f"respan-example-{execution}"
     stats = index.describe_index_stats()
     dimension = int(response_field(stats, "dimension", 0) or 0)
     if dimension < 1:
         raise RuntimeError("PINECONE_INDEX_NAME must reference a dense-vector index")
 
-    vector_ids = [f"{run_id}-python", f"{run_id}-rust", f"{run_id}-pasta"]
+    vector_ids = [f"{execution}-python", f"{execution}-rust", f"{execution}-pasta"]
     vectors = [
         {
             "id": vector_ids[0],
             "values": basis_vector(dimension, 0),
-            "metadata": {"topic": "programming", "text": "Python is approachable."},
+            "metadata": {"topic": topic, "text": "Python is approachable."},
         },
         {
             "id": vector_ids[1],
             "values": basis_vector(dimension, 1),
-            "metadata": {"topic": "programming", "text": "Rust emphasizes safety."},
+            "metadata": {"topic": topic, "text": "Rust emphasizes safety."},
         },
         {
             "id": vector_ids[2],
@@ -70,29 +75,35 @@ def run_upsert_and_query(run_id: str) -> dict:
         queried = index.query(
             namespace=namespace,
             vector=basis_vector(dimension, 0),
-            top_k=2,
+            top_k=top_k,
             include_metadata=True,
+            include_values=True,
+        )
+        return to_jsonable(
+            {
+                "index": os.getenv("PINECONE_INDEX_NAME", "loopback-index"),
+                "mode": "live" if live_configured() else "deterministic",
+                "namespace": namespace,
+                "dimension": dimension,
+                "upsert": upserted,
+                "fetch": fetched,
+                "query": queried,
+            }
         )
-        return {
-            "index": os.environ["PINECONE_INDEX_NAME"],
-            "namespace": namespace,
-            "dimension": dimension,
-            "upsert": upserted,
-            "fetch": fetched,
-            "query": queried,
-        }
     finally:
-        # Only remove IDs created by this run; the existing index is never modified.
         index.delete(ids=vector_ids, namespace=namespace)
 
 
 def main() -> None:
-    run_id = execution_id()
-    respan = create_respan(WORKFLOW_NAME)
+    run_marker = marker()
+    execution = execution_id()
+    respan = create_respan(WORKFLOW_NAME, run_marker)
     try:
-        with Respan.propagate_attributes(**workflow_attributes(WORKFLOW_NAME, run_id)):
-            result = run_upsert_and_query(run_id)
-        print_result(WORKFLOW_NAME, result)
+        with Respan.propagate_attributes(
+            **workflow_attributes(WORKFLOW_NAME, run_marker, execution)
+        ):
+            result = run_upsert_and_query("programming", 2)
+        print_result(WORKFLOW_NAME, result, run_marker)
     finally:
         finish_respan(respan)
 

python/tracing/pinecone/02_async_fetch.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from _shared import (
+    create_async_pinecone_index,
+    create_respan,
+    execution_id,
+    finish_respan,
+    marker,
+    print_result,
+    to_jsonable,
+    workflow_attributes,
+)
+from respan import Respan, workflow
+
+WORKFLOW_NAME = "pinecone_async_fetch_workflow"
+
+
+@workflow(name=WORKFLOW_NAME)
+async def run_async_fetch(vector_ids: list[str], namespace: str) -> dict[str, Any]:
+    index = create_async_pinecone_index()
+    try:
+        result = await index.fetch(ids=vector_ids, namespace=namespace)
+        return to_jsonable({"ids": vector_ids, "namespace": namespace, "fetch": result})
+    finally:
+        await index.close()
+
+
+async def main() -> None:
+    run_marker = marker()
+    execution = execution_id()
+    respan = create_respan(WORKFLOW_NAME, run_marker)
+    try:
+        with Respan.propagate_attributes(
+            **workflow_attributes(WORKFLOW_NAME, run_marker, execution)
+        ):
+            result = await run_async_fetch(["trace-doc"], "respan-example")
+        print_result(WORKFLOW_NAME, result, run_marker)
+    finally:
+        finish_respan(respan)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/tracing/pinecone/03_expected_error.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from _shared import (
+    create_pinecone_index,
+    create_respan,
+    execution_id,
+    finish_respan,
+    marker,
+    print_result,
+    workflow_attributes,
+)
+from pinecone.exceptions import PineconeApiException
+from respan import Respan, workflow
+
+WORKFLOW_NAME = "pinecone_expected_error_workflow"
+
+
+@workflow(name=WORKFLOW_NAME)
+def run_expected_error(namespace: str) -> None:
+    create_pinecone_index().delete(ids=["missing-vector"], namespace=namespace)
+
+
+def main() -> None:
+    run_marker = marker()
+    execution = execution_id()
+    respan = create_respan(WORKFLOW_NAME, run_marker)
+    try:
+        try:
+            with Respan.propagate_attributes(
+                **workflow_attributes(WORKFLOW_NAME, run_marker, execution)
+            ):
+                run_expected_error("error")
+        except PineconeApiException as exc:
+            result = {
+                "expected_error": type(exc).__name__,
+                "message": "deterministic Pinecone outage",
+            }
+        else:
+            raise AssertionError("the deterministic Pinecone failure did not occur")
+        print_result(WORKFLOW_NAME, result, run_marker)
+    finally:
+        finish_respan(respan)
+
+
+if __name__ == "__main__":
+    main()

python/tracing/pinecone/README.md

@@ -1,30 +1,48 @@
-# Pinecone Tracing Example
+# Pinecone tracing examples
 
-This example traces a concise `describe_index_stats` -> `upsert` -> `fetch` ->
-`query` flow against an existing dense-vector Pinecone index. It never creates,
-configures, or deletes an index, and it cleans up only the unique IDs written by
-the current run.
+These examples use the real Pinecone Python SDK and local editable Respan
+instrumentation. Without Pinecone credentials they run against a bounded local
+protocol fixture, so sync, async, success, and service-error paths remain
+repeatable. When both `PINECONE_API_KEY` and `PINECONE_INDEX_NAME` are set, the
+round-trip example uses that existing dense-vector index and deletes only its
+own unique IDs.
 
-Add these values to the repo-root `.env`:
+## Setup
+
+```bash
+cd python/tracing/pinecone
+pip install -r requirements.txt
+```
+
+For local instrumentation development, install the package from the sibling
+checkout before running the examples:
+
+```bash
+pip install -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-pinecone
+```
+
+Required in the repository-root `.env`:
 
 ```dotenv
 RESPAN_API_KEY=...
+```
+
+Optional live Pinecone settings:
+
+```dotenv
 PINECONE_API_KEY=...
 PINECONE_INDEX_NAME=your-existing-index
-# Recommended when known; avoids resolving the host by index name.
 PINECONE_INDEX_HOST=your-index-host
 ```
 
-The emitted workflow name is `pinecone_upsert_and_query_workflow`.
+`PINECONE_INDEX_HOST` is required for the async live example. The expected-error
+example always uses the deterministic fixture and never mutates a live index.
 
 ## Run
 
 ```bash
-cd python/tracing/pinecone
-pip install -r requirements.txt
-python run_all.py
+RESPAN_EXAMPLE_RUN_ID=my-exact-marker python run_all.py
 ```
 
-Set `PINECONE_INGEST_TIMEOUT_SECONDS` to change the default 30-second fetch
-polling window. When running before the instrumentor is published, include its
-local `src` directory and the local Respan packages on `PYTHONPATH`.
+The runner preserves the exact marker for all three processes, applies a
+per-process timeout, continues after failures, and reports them together.

python/tracing/pinecone/_loopback.py

@@ -0,0 +1,121 @@
+"""Deterministic protocol fixture used by the real Pinecone SDK examples."""
+
+from __future__ import annotations
+
+import json
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from threading import RLock, Thread
+from typing import Any
+from urllib.parse import parse_qs, urlsplit
+
+_lock = RLock()
+_server: ThreadingHTTPServer | None = None
+_thread: Thread | None = None
+
+
+class _Handler(BaseHTTPRequestHandler):
+    server_version = "PineconeExampleFixture/1.0"
+
+    def log_message(self, *_args: Any) -> None:
+        return
+
+    def _reply(self, status: int, value: object) -> None:
+        payload = json.dumps(value, allow_nan=False).encode("utf-8")
+        self.send_response(status)
+        self.send_header("content-type", "application/json")
+        self.send_header("content-length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def _body(self) -> dict[str, Any]:
+        length = int(self.headers.get("content-length", "0"))
+        raw = self.rfile.read(length) if length else b"{}"
+        try:
+            value = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            return {}
+        return value if isinstance(value, dict) else {}
+
+    def do_GET(self) -> None:
+        if self.path.startswith("/vectors/fetch"):
+            query = parse_qs(urlsplit(self.path).query)
+            vector_id = (query.get("ids") or ["trace-doc"])[0]
+            namespace = (query.get("namespace") or ["respan-example"])[0]
+            self._reply(
+                200,
+                {
+                    "namespace": namespace,
+                    "vectors": {
+                        vector_id: {
+                            "id": vector_id,
+                            "values": [1.0, 0.0, 0.0, 0.0],
+                            "metadata": {"topic": "tracing"},
+                        }
+                    },
+                },
+            )
+            return
+        self._reply(404, {"message": "not found"})
+
+    def do_POST(self) -> None:
+        body = self._body()
+        if self.path == "/describe_index_stats":
+            self._reply(
+                200,
+                {
+                    "dimension": 4,
+                    "indexFullness": 0.0,
+                    "namespaces": {"respan-example": {"vectorCount": 3}},
+                    "totalVectorCount": 3,
+                },
+            )
+        elif self.path == "/vectors/upsert":
+            self._reply(200, {"upsertedCount": len(body.get("vectors", []))})
+        elif self.path == "/query":
+            self._reply(
+                200,
+                {
+                    "namespace": body.get("namespace", "respan-example"),
+                    "matches": [
+                        {
+                            "id": "trace-doc",
+                            "score": 0.99,
+                            "values": [1.0, 0.0, 0.0, 0.0],
+                            "metadata": {
+                                "topic": "tracing",
+                                "text": "Pinecone instrumentation is active.",
+                            },
+                        }
+                    ],
+                },
+            )
+        elif self.path == "/vectors/delete" and body.get("namespace") == "error":
+            self._reply(503, {"message": "deterministic Pinecone outage"})
+        elif self.path == "/vectors/delete":
+            self._reply(200, {})
+        else:
+            self._reply(404, {"message": "not found"})
+
+
+def loopback_host() -> str:
+    global _server, _thread
+    with _lock:
+        if _server is None:
+            _server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+            _thread = Thread(target=_server.serve_forever, daemon=True)
+            _thread.start()
+        host, port = _server.server_address
+        return f"http://{host}:{port}"
+
+
+def shutdown_loopback() -> None:
+    global _server, _thread
+    with _lock:
+        server, thread = _server, _thread
+        _server = None
+        _thread = None
+    if server is not None:
+        server.shutdown()
+        server.server_close()
+    if thread is not None:
+        thread.join(timeout=2)