From 88cdf4159c0641ec02416acb622b4607a19925b2 Mon Sep 17 00:00:00 2001 From: Derek Mitchell Date: Tue, 22 Sep 2026 10:44:42 -0700 Subject: [PATCH 1/3] updated agent control config for o11y cloud --- .../8-apply-guardrails/1-add-agent-control.md | 4 +- .../4-app-with-controls/README.md | 9 ++- .../4-app-with-controls/agent.py | 30 ++++----- .../4-app-with-controls/app.py | 8 ++- .../helpers/agent_control_helpers.py | 67 ++++++++++++------- .../4-app-with-controls/k8s-o11y.yaml | 2 +- 6 files changed, 71 insertions(+), 49 deletions(-) diff --git a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md index a7e6730a99..5f657661a9 100644 --- a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md +++ b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md @@ -41,10 +41,10 @@ please use the following command instead: ```bash kubectl create configmap splunk-agent-control-config \ - --from-literal=AGENT_CONTROL_URL="https://app.$REALM.observability.splunkcloud.com/api/agent-control" \ + --from-literal=AGENT_CONTROL_URL="https://app.$REALM.observability.splunkcloud.com/ao/agent-control" \ --from-literal=AGENT_CONTROL_AGENT_NAME="agent-control-example" \ --from-literal=AGENT_CONTROL_API_KEY_HEADER="Splunk-AO-API-Key" \ - --from-literal=AGENT_CONTROL_RUNTIME_AUTH_MODE="jwt" \ + --from-literal=AGENT_CONTROL_RUNTIME_TOKEN_HEADER="X-Agent-Control-Runtime-Token" \ --from-literal=AGENT_CONTROL_TARGET_TYPE="agent_stream" ``` diff --git a/workshop/healthcare-assistant/4-app-with-controls/README.md b/workshop/healthcare-assistant/4-app-with-controls/README.md index 038835f48c..4640bfd672 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/README.md +++ b/workshop/healthcare-assistant/4-app-with-controls/README.md @@ -9,11 +9,14 @@ Edit the `.streamlit/secrets.toml` file and add the following environment variab ```yaml # Agent Controls # ----------------------------------------------------------------------------- -agent_control_url = "https://console.multitenant.galileocloud.io/api/agent-control" +# Standalone: +agent_control_url = "https://console.multitenant.sao.splunkcloud.com/api/agent-control" +# Observability Cloud: +# agent_control_url = "https://app..observability.splunkcloud.com/ao/agent-control" agent_control_agent_name = "agent-control-example" agent_control_runtime_auth_mode="jwt" -agent_control_api_key_header = "Splunk-AO-Key" -agent_control_target_type="log_stream" +agent_control_api_key_header = "Splunk-AO-API-Key" +agent_control_target_type="agent_stream" ``` ## Read New Config diff --git a/workshop/healthcare-assistant/4-app-with-controls/agent.py b/workshop/healthcare-assistant/4-app-with-controls/agent.py index 13434b6c70..2a956a7b74 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/agent.py +++ b/workshop/healthcare-assistant/4-app-with-controls/agent.py @@ -16,6 +16,17 @@ from langgraph.prebuilt import ToolNode, tools_condition from config import TOOLS_DIR, load_config, load_system_prompt +from helpers.agent_control_helpers import ( + build_agent_control_steps, + ensure_trace_started, + finalize_trace, + format_blocked_message, + init_agent_control, + infer_control_step_name, + make_controlled_tool, + notify_control_block, + uses_internal_sql_control, +) from rag import create_rag_tool from tools import logic as tools_logic @@ -28,17 +39,6 @@ enable_console_logging(level=logging.INFO) from agent_control import ControlSteerError, ControlViolationError, control -from helpers.agent_control_helpers import ( - build_agent_control_steps, - ensure_trace_started, - finalize_trace, - format_blocked_message, - init_agent_control, - infer_control_step_name, - make_controlled_tool, - notify_control_block, - uses_internal_sql_control, -) LLM_STEP_NAME = "Healthcare Assistant" MAX_STEER_RETRIES = 3 @@ -256,7 +256,7 @@ async def invoke_chatbot(state, config: RunnableConfig): message = AIMessage( content=( "I'm sorry, I'm unable to process your request right now due to " - "a control evaluation error. Please try again or contact support." + f"a control evaluation error: {e}" ) ) break @@ -279,9 +279,6 @@ async def _invoke_graph( *, splunk_ao_logger, ): - if splunk_ao_logger.experiment_id is None: - splunk_ao_context.start_session(external_id=self.session_id) - # Nest LangGraph spans under the trace started by ensure_trace_started(). callback = SplunkAOAsyncCallback( splunk_ao_logger, @@ -332,6 +329,9 @@ async def _process_query_async(self, messages: List[Dict[str, str]]) -> str: agent_stream=os.getenv("SPLUNK_AO_AGENT_STREAM"), ): splunk_ao_logger = splunk_ao_context.get_logger_instance() + # Observability Cloud does not resolve agent_stream_id until a + # session starts. Agent Control jwt evaluation requires that ID. + splunk_ao_context.start_session(external_id=self.session_id) self._init_agent_control(splunk_ao_logger) ensure_trace_started(splunk_ao_logger, langchain_messages, trace_name="Run Agent") try: diff --git a/workshop/healthcare-assistant/4-app-with-controls/app.py b/workshop/healthcare-assistant/4-app-with-controls/app.py index 1d6ee1211b..1ec0b7eb8f 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/app.py +++ b/workshop/healthcare-assistant/4-app-with-controls/app.py @@ -2,8 +2,12 @@ import os import uuid -import streamlit as st from dotenv import load_dotenv + +# Load .env before importing the agent so Agent Control SDK settings see local URLs. +load_dotenv() + +import streamlit as st from langchain_core.messages import AIMessage, HumanMessage from agent import HealthcareAgent @@ -15,8 +19,6 @@ from rag import get_rag_system from setup_env import setup_environment -load_dotenv() - if not os.getenv("_ENV_LOADED"): setup_environment() os.environ["_ENV_LOADED"] = "true" diff --git a/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py b/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py index ad76d2b02a..e2c6c8468a 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py +++ b/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py @@ -9,7 +9,6 @@ import agent_control from agent_control import ControlSteerError, ControlViolationError, control -from splunk_ao.agent_streams import get_agent_stream _initialized = False @@ -349,7 +348,17 @@ def init_agent_control( steps: Optional[list] = None, force: bool = False, ) -> bool: - """Initialize or refresh Agent Control for the current Splunk AO logger/session.""" + """Initialize Agent Control for the current Splunk AO logger/session. + + Configured per the Agent Control docs, identically for standalone and + Observability Cloud: + https://agent-observability-docs.splunk.com/how-to-guides/agent-control/initialize-and-configure-agent-control + + The deployment-specific server URL and API-key header come straight from the + AGENT_CONTROL_* environment variables, and the control target is the agent + stream ID the logger resolved when its session started (agent.py starts the + session before calling this). + """ global _initialized if splunk_ao_logger is None: @@ -357,48 +366,56 @@ def init_agent_control( return False server_url = os.environ.get("AGENT_CONTROL_URL") - agent_name = os.environ.get("AGENT_CONTROL_AGENT_NAME") - api_key = os.environ.get("SPLUNK_AO_API_KEY") - api_key_header = os.environ.get("AGENT_CONTROL_API_KEY_HEADER", "Splunk-AO-Key") + agent_name = os.environ.get("AGENT_CONTROL_AGENT_NAME", "default") + api_key = os.environ.get("SPLUNK_AO_API_KEY") or os.environ.get("SPLUNK_AO_O11Y_TOKEN") + api_key_header = os.environ.get("AGENT_CONTROL_API_KEY_HEADER", "Splunk-AO-API-Key") + target_type = os.environ.get("AGENT_CONTROL_TARGET_TYPE", "agent_stream") + target_id = getattr(splunk_ao_logger, "agent_stream_id", None) - if not all([server_url, agent_name, api_key]): + if not server_url or not api_key or not target_id: print( - "⚠️ Agent Control not configured " - "(set AGENT_CONTROL_URL, AGENT_CONTROL_AGENT_NAME, and SPLUNK_AO_API_KEY)" + "⚠️ Agent Control not configured (need AGENT_CONTROL_URL, a credential in " + "SPLUNK_AO_API_KEY/SPLUNK_AO_O11Y_TOKEN, and a started session with a " + "resolved agent_stream_id)" ) return False + # Publish resolved IDs (matches the docs' os.environ pattern). + os.environ["SPLUNK_AO_AGENT_STREAM_ID"] = str(target_id) + if getattr(splunk_ao_logger, "project_id", None): + os.environ["SPLUNK_AO_PROJECT_ID"] = str(splunk_ao_logger.project_id) + session_key = (agent_name, project_name, agent_stream, server_url) if not force and getattr(init_agent_control, "_last_session_key", None) == session_key: return True # Control spans require splunk_ao_logger.enable_agent_control() (done in agent.py). + control_steps = steps or STANDARD_AGENT_CONTROL_STEPS try: - agent_stream_data = get_agent_stream(name=agent_stream, project_name=project_name) + agent_control.init( + agent_name=agent_name, + agent_description=agent_description, + server_url=server_url, + api_key=api_key, + api_key_header=api_key_header, + observability_enabled=True, + observability_sink_name="registered", + target_type=target_type, + target_id=str(target_id), + steps=control_steps, + ) except Exception as e: - print(f"⚠️ Agent Control: failed to resolve agent stream '{agent_stream}': {e}") + print(f"⚠️ Agent Control init failed: {e}") return False - control_steps = steps or STANDARD_AGENT_CONTROL_STEPS - - agent_control.init( - agent_name=agent_name, - agent_description=agent_description, - server_url=server_url, - api_key=api_key, - api_key_header=api_key_header, - observability_enabled=True, - observability_sink_name="registered", - target_type="log_stream", - target_id=agent_stream_data.id, - steps=control_steps, - ) _initialized = True init_agent_control._last_session_key = session_key step_names = ", ".join(s["name"] for s in control_steps) print( f"✅ Agent Control initialized for agent '{agent_name}' " - f"(project={project_name}, agent_stream={agent_stream}, steps={step_names})" + f"(project={project_name}, agent_stream={agent_stream}, " + f"server_url={server_url}, api_key_header={api_key_header}, " + f"target_type={target_type}, target_id={target_id}, steps={step_names})" ) return True diff --git a/workshop/healthcare-assistant/4-app-with-controls/k8s-o11y.yaml b/workshop/healthcare-assistant/4-app-with-controls/k8s-o11y.yaml index 2f3fdf4a92..355186b1b7 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/k8s-o11y.yaml +++ b/workshop/healthcare-assistant/4-app-with-controls/k8s-o11y.yaml @@ -16,7 +16,7 @@ spec: spec: containers: - name: healthcare-assistant - image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v2 + image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v3 imagePullPolicy: Always ports: - containerPort: 8501 From c66339c6687de8654d6e2267dbe0a1f56afb50ef Mon Sep 17 00:00:00 2001 From: Derek Mitchell Date: Tue, 22 Sep 2026 19:04:49 -0700 Subject: [PATCH 2/3] multiple changes for agent control to work with o11y cloud SAO --- .../3-generate-traffic.md | 6 +- .../8-apply-guardrails/1-add-agent-control.md | 5 +- .../8-apply-guardrails/3-test-controls.md | 53 +++--- .../k8s-demo-o11y-eu-splunk-show.yaml | 2 +- .../k8s-demo-o11y-us-splunk-show.yaml | 2 +- .../k8s-demo-standalone.yaml | 2 +- workshop/healthcare-assistant/.dockerignore | 10 +- .../healthcare-assistant/1-base-app/agent.py | 43 ++++- .../2-app-with-instrumentation/agent.py | 43 ++++- .../2-app-with-instrumentation/k8s-o11y.yaml | 2 +- .../2-app-with-instrumentation/k8s.yaml | 2 +- .../3-app-with-experiments/agent.py | 43 ++++- .../4-app-with-controls/agent.py | 43 ++++- .../helpers/agent_control_helpers.py | 176 ++++++++++++++++-- 14 files changed, 340 insertions(+), 92 deletions(-) diff --git a/content/en/ninja-workshops/ai/3-splunk-agent-observability/4-instrument-the-application/3-generate-traffic.md b/content/en/ninja-workshops/ai/3-splunk-agent-observability/4-instrument-the-application/3-generate-traffic.md index f5d647c991..b8700425cc 100644 --- a/content/en/ninja-workshops/ai/3-splunk-agent-observability/4-instrument-the-application/3-generate-traffic.md +++ b/content/en/ninja-workshops/ai/3-splunk-agent-observability/4-instrument-the-application/3-generate-traffic.md @@ -132,14 +132,14 @@ Then rebuild the Docker image: ```bash cd ~/workshop/healthcare-assistant -docker build -f 2-app-with-instrumentation/Dockerfile -t localhost:9999/healthcare-assistant:app-with-instrumentation-v2 . -docker push localhost:9999/healthcare-assistant:app-with-instrumentation-v2 +docker build -f 2-app-with-instrumentation/Dockerfile -t localhost:9999/healthcare-assistant:app-with-instrumentation-v3 . +docker push localhost:9999/healthcare-assistant:app-with-instrumentation-v3 ``` Update the `~/workshop/healthcare-assistant/2-app-with-instrumentation/k8s.yaml` file to reference the local image instead: ```` -image: localhost:9999/healthcare-assistant:app-with-instrumentation-v2 +image: localhost:9999/healthcare-assistant:app-with-instrumentation-v3 ```` And redeploy the application: diff --git a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md index 5f657661a9..c4af069005 100644 --- a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md +++ b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md @@ -43,9 +43,10 @@ please use the following command instead: kubectl create configmap splunk-agent-control-config \ --from-literal=AGENT_CONTROL_URL="https://app.$REALM.observability.splunkcloud.com/ao/agent-control" \ --from-literal=AGENT_CONTROL_AGENT_NAME="agent-control-example" \ - --from-literal=AGENT_CONTROL_API_KEY_HEADER="Splunk-AO-API-Key" \ + --from-literal=AGENT_CONTROL_API_KEY_HEADER="X-SF-Token" \ --from-literal=AGENT_CONTROL_RUNTIME_TOKEN_HEADER="X-Agent-Control-Runtime-Token" \ - --from-literal=AGENT_CONTROL_TARGET_TYPE="agent_stream" + --from-literal=AGENT_CONTROL_RUNTIME_AUTH_MODE="jwt" \ + --from-literal=AGENT_CONTROL_TARGET_TYPE="log_stream" ``` {{< /step >}} diff --git a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/3-test-controls.md b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/3-test-controls.md index a8f26dafc6..aaf5198baf 100644 --- a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/3-test-controls.md +++ b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/3-test-controls.md @@ -55,36 +55,8 @@ section of Splunk Show event: {{% notice title="Troubleshooting" style="tip" icon="exclamation-triangle" %}} -To see what Agent Control is doing, enable console logging in `~/workshop/healthcare-assistant/4-app-with-controls/agent.py`: - -```python -from splunk_ao.utils.log_config import enable_console_logging - -enable_console_logging() -``` - -Then rebuild the Docker image: - -```bash -cd ~/workshop/healthcare-assistant -docker build -f 4-app-with-controls/Dockerfile -t localhost:9999/healthcare-assistant:app-with-controls-v2 . -docker push localhost:9999/healthcare-assistant:app-with-controls-v2 -``` - -Update the `~/workshop/healthcare-assistant/4-app-with-controls/k8s.yaml` file to reference the local image instead: - -```` -image: localhost:9999/healthcare-assistant:app-with-controls-v2 -```` - -And redeploy the application: - -```bash -cd ~/workshop/healthcare-assistant/4-app-with-controls -kubectl apply -f k8s.yaml -``` - -Use the following command to view the application logs: +The app already logs to stdout, so you can see what Agent Control is doing directly with +`kubectl logs` — no code change or image rebuild required. View the application logs: {{< tabs id="healthcare-app-logs" >}} {{% tab title="Script" %}} @@ -115,6 +87,27 @@ INFO - splunk_ao.logger - Session started with ID: ec03c538-cf9e-4bed-b97e-4b3c2 Watch the terminal for `Agent Control initialized` on startup and `BLOCKED` / `STEERED` messages when a control fires. +If you need to dig deeper — for example, to see exactly which environment variables the app +reads and the full HTTP request/response for every Agent Control call (agent registration, +runtime token exchange, and control evaluation) — set `AGENT_CONTROL_DEBUG=true` in the +`splunk-agent-control-config` ConfigMap and restart the app: + +```bash +kubectl patch configmap splunk-agent-control-config --type merge -p '{"data":{"AGENT_CONTROL_DEBUG":"true"}}' +kubectl rollout restart deployment/healthcare-assistant +``` + +Credential headers (such as `X-SF-Token`) are masked in these logs by default. If you need to +verify the exact token value, add `AGENT_CONTROL_DEBUG_UNMASK=true` the same way: + +```bash +kubectl patch configmap splunk-agent-control-config --type merge -p '{"data":{"AGENT_CONTROL_DEBUG_UNMASK":"true"}}' +kubectl rollout restart deployment/healthcare-assistant +``` + +Leave both unset for normal use — the app only logs a concise `Agent Control initialized` +line and any configuration warnings. + {{% /notice %}} {{< /step >}} diff --git a/workshop/demos/healthcare-assistant/k8s-demo-o11y-eu-splunk-show.yaml b/workshop/demos/healthcare-assistant/k8s-demo-o11y-eu-splunk-show.yaml index cfa41ccb0c..3ab41b8957 100644 --- a/workshop/demos/healthcare-assistant/k8s-demo-o11y-eu-splunk-show.yaml +++ b/workshop/demos/healthcare-assistant/k8s-demo-o11y-eu-splunk-show.yaml @@ -16,7 +16,7 @@ spec: spec: containers: - name: healthcare-assistant-eu - image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v2 + image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v3 imagePullPolicy: Always ports: - containerPort: 8501 diff --git a/workshop/demos/healthcare-assistant/k8s-demo-o11y-us-splunk-show.yaml b/workshop/demos/healthcare-assistant/k8s-demo-o11y-us-splunk-show.yaml index 7823fede91..87a3cbae07 100644 --- a/workshop/demos/healthcare-assistant/k8s-demo-o11y-us-splunk-show.yaml +++ b/workshop/demos/healthcare-assistant/k8s-demo-o11y-us-splunk-show.yaml @@ -16,7 +16,7 @@ spec: spec: containers: - name: healthcare-assistant-us - image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v2 + image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v3 imagePullPolicy: Always ports: - containerPort: 8501 diff --git a/workshop/demos/healthcare-assistant/k8s-demo-standalone.yaml b/workshop/demos/healthcare-assistant/k8s-demo-standalone.yaml index a5fd09725e..34ff0728f9 100644 --- a/workshop/demos/healthcare-assistant/k8s-demo-standalone.yaml +++ b/workshop/demos/healthcare-assistant/k8s-demo-standalone.yaml @@ -16,7 +16,7 @@ spec: spec: containers: - name: healthcare-assistant - image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v2 + image: ghcr.io/splunk/healthcare-assistant:app-with-controls-v3 imagePullPolicy: Always ports: - containerPort: 8501 diff --git a/workshop/healthcare-assistant/.dockerignore b/workshop/healthcare-assistant/.dockerignore index 75f886e232..68db7d227b 100644 --- a/workshop/healthcare-assistant/.dockerignore +++ b/workshop/healthcare-assistant/.dockerignore @@ -1,8 +1,8 @@ **/__pycache__ **/*.pyc **/*.pyo -1-base-app/.streamlit/secrets.toml -1-base-app/k8s.yaml -1-base-app/postgres.yaml -1-base-app/setup-job.yaml -1-base-app/start_vectordb.sh +**/.streamlit/secrets.toml +**/k8s*.yaml +**/postgres.yaml +**/setup-job.yaml +**/start_vectordb.sh diff --git a/workshop/healthcare-assistant/1-base-app/agent.py b/workshop/healthcare-assistant/1-base-app/agent.py index fd64dc5597..392de84a62 100644 --- a/workshop/healthcare-assistant/1-base-app/agent.py +++ b/workshop/healthcare-assistant/1-base-app/agent.py @@ -2,8 +2,8 @@ import asyncio import inspect import json +import threading import uuid -from concurrent.futures import ThreadPoolExecutor from typing import Annotated, List, Dict, Optional, TypedDict from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -23,15 +23,40 @@ class State(TypedDict): messages: Annotated[list, add_messages] +_bg_loop: Optional[asyncio.AbstractEventLoop] = None +_bg_loop_lock = threading.Lock() + + +def _get_background_loop() -> asyncio.AbstractEventLoop: + """Return a process-wide event loop running on a dedicated daemon thread. + + Streamlit calls process_query() synchronously on every interaction. Using + asyncio.run() per call creates and then *closes* a new loop each time, but + modern langchain-openai/openai reuse a cached async HTTP client across calls, + so its connection pool outlives the loop that created it. When a later request + reaps a keep-alive connection bound to that closed loop, httpx raises + "RuntimeError: Event loop is closed". Running every coroutine on one stable, + long-lived loop keeps the shared client and its pool on a single loop. + """ + global _bg_loop + if _bg_loop is not None and not _bg_loop.is_closed(): + return _bg_loop + with _bg_loop_lock: + if _bg_loop is None or _bg_loop.is_closed(): + loop = asyncio.new_event_loop() + threading.Thread( + target=loop.run_forever, + name="agent-async-loop", + daemon=True, + ).start() + _bg_loop = loop + return _bg_loop + + def _run_async(coro): - """Run an async coroutine from sync code (e.g. Streamlit).""" - try: - asyncio.get_running_loop() - except RuntimeError: - return asyncio.run(coro) - - with ThreadPoolExecutor(max_workers=1) as executor: - return executor.submit(asyncio.run, coro).result() + """Run an async coroutine from sync code (e.g. Streamlit) on the shared loop.""" + loop = _get_background_loop() + return asyncio.run_coroutine_threadsafe(coro, loop).result() class HealthcareAgent: diff --git a/workshop/healthcare-assistant/2-app-with-instrumentation/agent.py b/workshop/healthcare-assistant/2-app-with-instrumentation/agent.py index d458f01262..d5643cc541 100644 --- a/workshop/healthcare-assistant/2-app-with-instrumentation/agent.py +++ b/workshop/healthcare-assistant/2-app-with-instrumentation/agent.py @@ -2,8 +2,8 @@ import asyncio import inspect import json +import threading import uuid -from concurrent.futures import ThreadPoolExecutor from typing import Annotated, List, Dict, Optional, TypedDict from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -31,15 +31,40 @@ class State(TypedDict): messages: Annotated[list, add_messages] +_bg_loop: Optional[asyncio.AbstractEventLoop] = None +_bg_loop_lock = threading.Lock() + + +def _get_background_loop() -> asyncio.AbstractEventLoop: + """Return a process-wide event loop running on a dedicated daemon thread. + + Streamlit calls process_query() synchronously on every interaction. Using + asyncio.run() per call creates and then *closes* a new loop each time, but + modern langchain-openai/openai reuse a cached async HTTP client across calls, + so its connection pool outlives the loop that created it. When a later request + reaps a keep-alive connection bound to that closed loop, httpx raises + "RuntimeError: Event loop is closed". Running every coroutine on one stable, + long-lived loop keeps the shared client and its pool on a single loop. + """ + global _bg_loop + if _bg_loop is not None and not _bg_loop.is_closed(): + return _bg_loop + with _bg_loop_lock: + if _bg_loop is None or _bg_loop.is_closed(): + loop = asyncio.new_event_loop() + threading.Thread( + target=loop.run_forever, + name="agent-async-loop", + daemon=True, + ).start() + _bg_loop = loop + return _bg_loop + + def _run_async(coro): - """Run an async coroutine from sync code (e.g. Streamlit).""" - try: - asyncio.get_running_loop() - except RuntimeError: - return asyncio.run(coro) - - with ThreadPoolExecutor(max_workers=1) as executor: - return executor.submit(asyncio.run, coro).result() + """Run an async coroutine from sync code (e.g. Streamlit) on the shared loop.""" + loop = _get_background_loop() + return asyncio.run_coroutine_threadsafe(coro, loop).result() class HealthcareAgent: diff --git a/workshop/healthcare-assistant/2-app-with-instrumentation/k8s-o11y.yaml b/workshop/healthcare-assistant/2-app-with-instrumentation/k8s-o11y.yaml index 97a4f47416..450ff23c4b 100644 --- a/workshop/healthcare-assistant/2-app-with-instrumentation/k8s-o11y.yaml +++ b/workshop/healthcare-assistant/2-app-with-instrumentation/k8s-o11y.yaml @@ -16,7 +16,7 @@ spec: spec: containers: - name: healthcare-assistant - image: ghcr.io/splunk/healthcare-assistant:app-with-instrumentation-v2 + image: ghcr.io/splunk/healthcare-assistant:app-with-instrumentation-v3 imagePullPolicy: Always ports: - containerPort: 8501 diff --git a/workshop/healthcare-assistant/2-app-with-instrumentation/k8s.yaml b/workshop/healthcare-assistant/2-app-with-instrumentation/k8s.yaml index ffea16e539..2367c7008d 100644 --- a/workshop/healthcare-assistant/2-app-with-instrumentation/k8s.yaml +++ b/workshop/healthcare-assistant/2-app-with-instrumentation/k8s.yaml @@ -16,7 +16,7 @@ spec: spec: containers: - name: healthcare-assistant - image: ghcr.io/splunk/healthcare-assistant:app-with-instrumentation-v2 + image: ghcr.io/splunk/healthcare-assistant:app-with-instrumentation-v3 imagePullPolicy: Always ports: - containerPort: 8501 diff --git a/workshop/healthcare-assistant/3-app-with-experiments/agent.py b/workshop/healthcare-assistant/3-app-with-experiments/agent.py index 71815961c5..f93291be48 100644 --- a/workshop/healthcare-assistant/3-app-with-experiments/agent.py +++ b/workshop/healthcare-assistant/3-app-with-experiments/agent.py @@ -2,8 +2,8 @@ import asyncio import inspect import json +import threading import uuid -from concurrent.futures import ThreadPoolExecutor from typing import Annotated, List, Dict, Optional, TypedDict from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -30,15 +30,40 @@ class State(TypedDict): messages: Annotated[list, add_messages] +_bg_loop: Optional[asyncio.AbstractEventLoop] = None +_bg_loop_lock = threading.Lock() + + +def _get_background_loop() -> asyncio.AbstractEventLoop: + """Return a process-wide event loop running on a dedicated daemon thread. + + Streamlit calls process_query() synchronously on every interaction. Using + asyncio.run() per call creates and then *closes* a new loop each time, but + modern langchain-openai/openai reuse a cached async HTTP client across calls, + so its connection pool outlives the loop that created it. When a later request + reaps a keep-alive connection bound to that closed loop, httpx raises + "RuntimeError: Event loop is closed". Running every coroutine on one stable, + long-lived loop keeps the shared client and its pool on a single loop. + """ + global _bg_loop + if _bg_loop is not None and not _bg_loop.is_closed(): + return _bg_loop + with _bg_loop_lock: + if _bg_loop is None or _bg_loop.is_closed(): + loop = asyncio.new_event_loop() + threading.Thread( + target=loop.run_forever, + name="agent-async-loop", + daemon=True, + ).start() + _bg_loop = loop + return _bg_loop + + def _run_async(coro): - """Run an async coroutine from sync code (e.g. Streamlit).""" - try: - asyncio.get_running_loop() - except RuntimeError: - return asyncio.run(coro) - - with ThreadPoolExecutor(max_workers=1) as executor: - return executor.submit(asyncio.run, coro).result() + """Run an async coroutine from sync code (e.g. Streamlit) on the shared loop.""" + loop = _get_background_loop() + return asyncio.run_coroutine_threadsafe(coro, loop).result() class HealthcareAgent: diff --git a/workshop/healthcare-assistant/4-app-with-controls/agent.py b/workshop/healthcare-assistant/4-app-with-controls/agent.py index 2a956a7b74..1e7cb4dfa3 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/agent.py +++ b/workshop/healthcare-assistant/4-app-with-controls/agent.py @@ -2,8 +2,8 @@ import asyncio import inspect import json +import threading import uuid -from concurrent.futures import ThreadPoolExecutor from typing import Annotated, Any, Dict, List, Optional, TypedDict from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -48,15 +48,40 @@ class State(TypedDict): messages: Annotated[list, add_messages] +_bg_loop: Optional[asyncio.AbstractEventLoop] = None +_bg_loop_lock = threading.Lock() + + +def _get_background_loop() -> asyncio.AbstractEventLoop: + """Return a process-wide event loop running on a dedicated daemon thread. + + Streamlit calls process_query() synchronously on every interaction. Using + asyncio.run() per call creates and then *closes* a new loop each time, but + modern langchain-openai/openai reuse a cached async HTTP client across calls, + so its connection pool outlives the loop that created it. When a later request + reaps a keep-alive connection bound to that closed loop, httpx raises + "RuntimeError: Event loop is closed". Running every coroutine on one stable, + long-lived loop keeps the shared client and its pool on a single loop. + """ + global _bg_loop + if _bg_loop is not None and not _bg_loop.is_closed(): + return _bg_loop + with _bg_loop_lock: + if _bg_loop is None or _bg_loop.is_closed(): + loop = asyncio.new_event_loop() + threading.Thread( + target=loop.run_forever, + name="agent-async-loop", + daemon=True, + ).start() + _bg_loop = loop + return _bg_loop + + def _run_async(coro): - """Run an async coroutine from sync code (e.g. Streamlit).""" - try: - asyncio.get_running_loop() - except RuntimeError: - return asyncio.run(coro) - - with ThreadPoolExecutor(max_workers=1) as executor: - return executor.submit(asyncio.run, coro).result() + """Run an async coroutine from sync code (e.g. Streamlit) on the shared loop.""" + loop = _get_background_loop() + return asyncio.run_coroutine_threadsafe(coro, loop).result() def _message_content_text(message: BaseMessage) -> str: diff --git a/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py b/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py index e2c6c8468a..9990904761 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py +++ b/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py @@ -4,13 +4,146 @@ import functools import inspect import json +import logging import os +import sys from typing import Any, Callable, Optional import agent_control from agent_control import ControlSteerError, ControlViolationError, control +logger = logging.getLogger(__name__) + + +def _ensure_logger_visible() -> None: + """Attach a stdout handler so these logs show up in ``kubectl logs``. + + splunk_ao's ``enable_console_logging()`` only wires the ``splunk_ao`` logger + (with propagate disabled), and the app never configures the root logger, so + without this our records would be dropped. Runs once and stays out of the way + if logging has already been configured for this module. + """ + if logger.handlers or logger.level != logging.NOTSET: + return + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(logging.Formatter("%(levelname)s - %(name)s - %(message)s")) + logger.addHandler(handler) + logger.setLevel(logging.INFO) + logger.propagate = False # our own handler; avoid duplicates if root is later configured + + +_ensure_logger_visible() + _initialized = False +_env_logged = False +_http_logging_installed = False + +# Header/env values that must be masked in logs (credentials / tokens). Header +# *names* ending in _HEADER are safe; only their values would be secret. +_SENSITIVE_HEADERS = frozenset({ + "x-sf-token", "authorization", "splunk-ao-api-key", "galileo-api-key", + "x-api-key", "x-agent-control-runtime-token", +}) +_SENSITIVE_ENV_SUBSTR = ("TOKEN", "API_KEY", "SECRET", "PASSWORD") + + +def _env_flag(name: str, default: bool = False) -> bool: + """Parse a boolean-ish environment variable.""" + raw = (os.environ.get(name) or "").strip().lower() + if not raw: + return default + return raw in ("1", "true", "yes", "on") + + +def _mask_secret(value: Optional[str]) -> str: + """Mask a credential for logging while leaving it recognizable.""" + if not value: + return "" + if len(value) <= 8: + return f"*** (len={len(value)})" + return f"{value[:4]}…{value[-4:]} (len={len(value)})" + + +def _format_headers_for_log(headers, *, unmask: bool = False) -> str: + rendered = [] + for name, value in headers.items(): + if not unmask and name.lower() in _SENSITIVE_HEADERS: + value = _mask_secret(value) + rendered.append(f"{name}: {value}") + return " | ".join(rendered) + + +def _log_agent_control_env() -> None: + """Log (once) the AGENT_CONTROL_*/SPLUNK_AO_* env vars the app sees, secrets masked.""" + global _env_logged + if _env_logged: + return + _env_logged = True + logger.info("Agent Control init — environment variables visible to the app:") + for name in sorted(os.environ): + if not name.startswith(("AGENT_CONTROL_", "SPLUNK_AO_")): + continue + value = os.environ[name] + if not name.endswith("_HEADER") and any(s in name for s in _SENSITIVE_ENV_SUBSTR): + value = _mask_secret(value) + logger.info(" %s=%s", name, value) + + +def _install_agent_control_http_logging() -> None: + """Log every Agent Control HTTP request/response (URL, headers, body) via httpx hooks. + + Patches ``httpx.AsyncClient`` once to attach event hooks. The hooks only act on + agent-control URLs, so other httpx traffic (OpenAI, Splunk AO) is left alone. + Credential headers are masked unless AGENT_CONTROL_DEBUG_HTTP_UNMASK is truthy. + """ + global _http_logging_installed + if _http_logging_installed: + return + try: + import httpx + except Exception as e: # pragma: no cover - httpx is a hard dep of the SDK + logger.warning("Agent Control HTTP logging unavailable (httpx import failed: %s)", e) + return + + unmask = _env_flag("AGENT_CONTROL_DEBUG_UNMASK") + + async def _log_request(request) -> None: + if "agent-control" not in str(request.url): + return + body = request.content.decode("utf-8", "replace") if request.content else "" + logger.info( + "→ Agent Control request: %s %s\n headers: %s\n body: %s", + request.method, request.url, + _format_headers_for_log(request.headers, unmask=unmask), body[:2000], + ) + + async def _log_response(response) -> None: + if "agent-control" not in str(response.request.url): + return + await response.aread() + logger.info( + "← Agent Control response: HTTP %s %s\n headers: %s\n body: %s", + response.status_code, response.request.url, + _format_headers_for_log(response.headers, unmask=unmask), response.text[:2000], + ) + + original_init = httpx.AsyncClient.__init__ + + @functools.wraps(original_init) + def _patched_init(self, *args, **kwargs): + hooks = dict(kwargs.get("event_hooks") or {}) + hooks["request"] = list(hooks.get("request", [])) + [_log_request] + hooks["response"] = list(hooks.get("response", [])) + [_log_response] + kwargs["event_hooks"] = hooks + original_init(self, *args, **kwargs) + + httpx.AsyncClient.__init__ = _patched_init + _http_logging_installed = True + logger.info( + "Agent Control HTTP logging enabled (httpx request/response hooks; " + "credential headers %s)", + "UNMASKED" if unmask else "masked (set AGENT_CONTROL_DEBUG_UNMASK=true to reveal)", + ) # Hardcoded fallback when no domain tool list is available. STANDARD_AGENT_CONTROL_STEPS = [ @@ -362,7 +495,7 @@ def init_agent_control( global _initialized if splunk_ao_logger is None: - print("⚠️ Agent Control not initialized (no Splunk AO logger)") + logger.warning("Agent Control not initialized (no Splunk AO logger)") return False server_url = os.environ.get("AGENT_CONTROL_URL") @@ -370,13 +503,21 @@ def init_agent_control( api_key = os.environ.get("SPLUNK_AO_API_KEY") or os.environ.get("SPLUNK_AO_O11Y_TOKEN") api_key_header = os.environ.get("AGENT_CONTROL_API_KEY_HEADER", "Splunk-AO-API-Key") target_type = os.environ.get("AGENT_CONTROL_TARGET_TYPE", "agent_stream") + observability_sink_name = os.environ.get("AGENT_CONTROL_OBSERVABILITY_SINK_NAME", "registered") target_id = getattr(splunk_ao_logger, "agent_stream_id", None) + # Deep debugging (env dump + raw HTTP request/response logging) is opt-in and + # off by default. Set AGENT_CONTROL_DEBUG=true to enable it when investigating. + debug = _env_flag("AGENT_CONTROL_DEBUG") + if debug: + _log_agent_control_env() + if not server_url or not api_key or not target_id: - print( - "⚠️ Agent Control not configured (need AGENT_CONTROL_URL, a credential in " + logger.warning( + "Agent Control not configured (need AGENT_CONTROL_URL, a credential in " "SPLUNK_AO_API_KEY/SPLUNK_AO_O11Y_TOKEN, and a started session with a " - "resolved agent_stream_id)" + "resolved agent_stream_id). server_url=%s api_key_set=%s target_id=%s", + server_url, bool(api_key), target_id, ) return False @@ -387,11 +528,24 @@ def init_agent_control( session_key = (agent_name, project_name, agent_stream, server_url) if not force and getattr(init_agent_control, "_last_session_key", None) == session_key: + logger.debug("Agent Control already initialized for this session; skipping re-init") return True # Control spans require splunk_ao_logger.enable_agent_control() (done in agent.py). control_steps = steps or STANDARD_AGENT_CONTROL_STEPS + if debug: + # Log the exact init parameters, then turn on wire logging so init()'s own + # register/health calls (and later evaluation calls) are captured. + logger.info( + "Calling agent_control.init(): agent_name=%s server_url=%s api_key_header=%s " + "api_key=%s target_type=%s target_id=%s observability_sink_name=%s steps=[%s]", + agent_name, server_url, api_key_header, _mask_secret(api_key), + target_type, target_id, observability_sink_name, + ", ".join(s["name"] for s in control_steps), + ) + _install_agent_control_http_logging() + try: agent_control.init( agent_name=agent_name, @@ -400,22 +554,22 @@ def init_agent_control( api_key=api_key, api_key_header=api_key_header, observability_enabled=True, - observability_sink_name="registered", + observability_sink_name=observability_sink_name, target_type=target_type, target_id=str(target_id), steps=control_steps, ) except Exception as e: - print(f"⚠️ Agent Control init failed: {e}") + logger.warning("Agent Control init failed: %s", e, exc_info=True) return False _initialized = True init_agent_control._last_session_key = session_key step_names = ", ".join(s["name"] for s in control_steps) - print( - f"✅ Agent Control initialized for agent '{agent_name}' " - f"(project={project_name}, agent_stream={agent_stream}, " - f"server_url={server_url}, api_key_header={api_key_header}, " - f"target_type={target_type}, target_id={target_id}, steps={step_names})" + logger.info( + "✅ Agent Control initialized for agent '%s' (project=%s agent_stream=%s " + "server_url=%s api_key_header=%s target_type=%s target_id=%s steps=[%s])", + agent_name, project_name, agent_stream, server_url, api_key_header, + target_type, target_id, step_names, ) return True From f590a1d117af9f5e00236c65aaa095ef2a81e2f6 Mon Sep 17 00:00:00 2001 From: Derek Mitchell Date: Wed, 23 Sep 2026 10:10:37 -0700 Subject: [PATCH 3/3] addressing issues found in testing with agent control --- .../8-apply-guardrails/1-add-agent-control.md | 2 +- .../helpers/agent_control_helpers.py | 17 +++++++++++------ 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md index c4af069005..376dcd2546 100644 --- a/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md +++ b/content/en/ninja-workshops/ai/3-splunk-agent-observability/8-apply-guardrails/1-add-agent-control.md @@ -33,7 +33,7 @@ kubectl create configmap splunk-agent-control-config \ --from-literal=AGENT_CONTROL_AGENT_NAME="agent-control-example" \ --from-literal=AGENT_CONTROL_API_KEY_HEADER="Splunk-AO-API-Key" \ --from-literal=AGENT_CONTROL_RUNTIME_AUTH_MODE="jwt" \ - --from-literal=AGENT_CONTROL_TARGET_TYPE="agent_stream" + --from-literal=AGENT_CONTROL_TARGET_TYPE="log_stream" ``` Alternatively, if you're using Splunk Agent Observability within Splunk Observability Cloud for this workshop, diff --git a/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py b/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py index 9990904761..182d956f1c 100644 --- a/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py +++ b/workshop/healthcare-assistant/4-app-with-controls/helpers/agent_control_helpers.py @@ -502,7 +502,12 @@ def init_agent_control( agent_name = os.environ.get("AGENT_CONTROL_AGENT_NAME", "default") api_key = os.environ.get("SPLUNK_AO_API_KEY") or os.environ.get("SPLUNK_AO_O11Y_TOKEN") api_key_header = os.environ.get("AGENT_CONTROL_API_KEY_HEADER", "Splunk-AO-API-Key") - target_type = os.environ.get("AGENT_CONTROL_TARGET_TYPE", "agent_stream") + # SAO authorizes the control target as a log_stream (an agent stream *is* a + # log stream, and its ID is the target_id). target_type="agent_stream" is + # rejected upstream with AUTH_UPSTREAM_REJECTED. The Splunk docs show + # "agent_stream", but that value does not work against SAO — default to + # log_stream, matching splunk_ao.get_agent_control_target. + target_type = os.environ.get("AGENT_CONTROL_TARGET_TYPE", "log_stream") observability_sink_name = os.environ.get("AGENT_CONTROL_OBSERVABILITY_SINK_NAME", "registered") target_id = getattr(splunk_ao_logger, "agent_stream_id", None) @@ -521,11 +526,11 @@ def init_agent_control( ) return False - # Publish resolved IDs (matches the docs' os.environ pattern). - os.environ["SPLUNK_AO_AGENT_STREAM_ID"] = str(target_id) - if getattr(splunk_ao_logger, "project_id", None): - os.environ["SPLUNK_AO_PROJECT_ID"] = str(splunk_ao_logger.project_id) - + # NOTE: do not publish SPLUNK_AO_PROJECT_ID / SPLUNK_AO_AGENT_STREAM_ID into + # os.environ. The project/agent-stream *names* are already set from the + # ConfigMap, and splunk_ao.resolve_routing rejects having both a name and an + # ID in the environment ("Cannot configure both project name and ID"). We + # pass target_id to agent_control.init() directly from the logger instead. session_key = (agent_name, project_name, agent_stream, server_url) if not force and getattr(init_agent_control, "_last_session_key", None) == session_key: logger.debug("Agent Control already initialized for this session; skipping re-init")