Test LangSmith built-in query filtering

tests/contrib/langsmith/test_integration.py

@@ -1271,8 +1271,6 @@ async def test_temporal_prefixed_query_not_traced(
 
             # Built-in queries — should NOT be traced
             await handle.query("__temporal_workflow_metadata")
-            await handle.query("__stack_trace")
-            await handle.query("__enhanced_stack_trace")
 
             # User query — should be traced
             await handle.query(QueryFilteringWorkflow.my_query)

tests/contrib/langsmith/test_interceptor.py

@@ -16,9 +16,11 @@
     HEADER_KEY,
     _extract_context,
     _inject_context,
+    _LangSmithWorkflowInboundInterceptor,
     _maybe_run,
     _ReplaySafeRunTree,
 )
+from temporalio.worker import HandleQueryInput
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -88,6 +90,48 @@ def _get_runtree_metadata(MockRunTree: MagicMock) -> dict[str, Any]:
     return kwargs.get("metadata", {})
 
 
+# ===================================================================
+# TestBuiltinQueryFiltering
+# ===================================================================
+
+
+class _RecordingWorkflowInboundInterceptor:
+    def __init__(self) -> None:
+        self.queries: list[HandleQueryInput] = []
+
+    async def handle_query(self, input: HandleQueryInput) -> str:
+        self.queries.append(input)
+        return "forwarded"
+
+
+class TestBuiltinQueryFiltering:
+    async def test_builtin_queries_bypass_tracing(self) -> None:
+        next_interceptor = _RecordingWorkflowInboundInterceptor()
+        interceptor = _LangSmithWorkflowInboundInterceptor(
+            next_interceptor  # type: ignore[arg-type]
+        )
+
+        with patch.object(interceptor, "_workflow_maybe_run") as maybe_run:
+            for query in (
+                "__temporal_workflow_metadata",
+                "__stack_trace",
+                "__enhanced_stack_trace",
+            ):
+                assert (
+                    await interceptor.handle_query(
+                        HandleQueryInput(id="id", query=query, args=[], headers={})
+                    )
+                    == "forwarded"
+                )
+
+        assert [input.query for input in next_interceptor.queries] == [
+            "__temporal_workflow_metadata",
+            "__stack_trace",
+            "__enhanced_stack_trace",
+        ]
+        maybe_run.assert_not_called()
+
+
 # ===================================================================
 # TestContextPropagation
 # ===================================================================