diff --git a/README.md b/README.md index 0b28b7f..0a1d8f1 100644 --- a/README.md +++ b/README.md @@ -353,6 +353,9 @@ ENABLE_OCI_STREAMING=false OCI_STREAM_ENDPOINT= OCI_STREAM_OCID= OCI_STREAM_PARTITION_KEY=agent-events +# Lista opcional, separada por vírgulas, de eventos não publicados no Pub/Sub. +# Os eventos continuam disponíveis para os demais destinos de observabilidade. +PUBSUB_EXCLUDED_EVENT_TYPES=GRL.NATIVE_OUTPUT_GUARDRAILS ENABLE_INPUT_GUARDRAILS=true ENABLE_OUTPUT_GUARDRAILS=true @@ -469,6 +472,20 @@ Autentica utilizando OCI Instance Principals. Autentica utilizando OCI Resource Principals. +**OCI_AUTH_MODE**=oke_workload_identity + +Autentica workloads executando em OKE com OCI OKE Workload Identity, usando o +signer `get_oke_workload_identity_resource_principal_signer()` do SDK OCI. +Esse modo é específico para Pods no OKE e não deve ser confundido com +`resource_principal`, destinado a OCI Functions e outros contextos de Resource +Principal. + +O Pod deve executar com uma `ServiceAccount` configurada para OKE Workload +Identity e o dynamic group associado deve possuir as políticas IAM necessárias +para o compartment do Generative AI. O SDK usa automaticamente o token padrão +da ServiceAccount em +`/var/run/secrets/kubernetes.io/serviceaccount/token`. + ### OCI_GENAI_API_KEY API Key utilizada pelo provider `oci_openai`. @@ -482,6 +499,7 @@ API Key utilizada pelo provider `oci_openai`. | oci_sdk | config_file | Não | OCI Config File | | oci_sdk | instance_principal | Não | Instance Principal | | oci_sdk | resource_principal | Não | Resource Principal | +| oci_sdk | oke_workload_identity | Não | OKE Workload Identity | | openai_compatible | Ignorado | Não | API Key do endpoint | diff --git a/README_en.md b/README_en.md index a456722..c25cc4f 100644 --- a/README_en.md +++ b/README_en.md @@ -351,6 +351,9 @@ ENABLE_OCI_STREAMING=false OCI_STREAM_ENDPOINT= OCI_STREAM_OCID= OCI_STREAM_PARTITION_KEY=agent-events +# Optional comma-separated list of events not published to Pub/Sub. +# Events remain available to other observability destinations. +PUBSUB_EXCLUDED_EVENT_TYPES=GRL.NATIVE_OUTPUT_GUARDRAILS ENABLE_INPUT_GUARDRAILS=true ENABLE_OUTPUT_GUARDRAILS=true @@ -467,6 +470,19 @@ Authenticates using OCI Instance Principals. Authenticates using OCI Resource Principals. +**OCI_AUTH_MODE**=oke_workload_identity + +Authenticates workloads running on OKE using OCI OKE Workload Identity and the +OCI SDK `get_oke_workload_identity_resource_principal_signer()` signer. This +mode is specific to Pods running on OKE and must not be confused with +`resource_principal`, which is intended for OCI Functions and other Resource +Principal contexts. + +The Pod must run with a `ServiceAccount` configured for OKE Workload Identity, +and the associated dynamic group must have the IAM policies required for the +Generative AI compartment. The SDK automatically uses the default ServiceAccount +token at `/var/run/secrets/kubernetes.io/serviceaccount/token`. + ### OCI_GENAI_API_KEY API Key used by the `oci_openai` provider. @@ -480,6 +496,7 @@ API Key used by the `oci_openai` provider. | oci_sdk | config_file | No | OCI Config File | | oci_sdk | instance_principal | No | Instance Principal | | oci_sdk | resource_principal | No | Resource Principal | +| oci_sdk | oke_workload_identity | No | OKE Workload Identity | | openai_compatible | Ignored | No | Endpoint API Key | --- diff --git a/libs/agent_framework/docs/IC_NOC_GRL_LANGFUSE_NATIVE.md b/libs/agent_framework/docs/IC_NOC_GRL_LANGFUSE_NATIVE.md index 7facb07..1cab309 100644 --- a/libs/agent_framework/docs/IC_NOC_GRL_LANGFUSE_NATIVE.md +++ b/libs/agent_framework/docs/IC_NOC_GRL_LANGFUSE_NATIVE.md @@ -50,6 +50,10 @@ Cada evento vira uma observation/span com `name` igual ao código: - `NOC.001` - `GRL.004` +No modo `LANGFUSE_TRACE_MODE=compact`, eventos `IC.*`, `AGA.*` e `NOC.*` +continuam visíveis como spans filhos do span raiz. Eles não são substituídos por +tags da trace. Eventos técnicos de baixo nível continuam sujeitos à compactação. + A metadata recebe automaticamente: - `tag` diff --git a/libs/agent_framework/docs/README_TIM_OBSERVER_PAYLOAD_FIX.md b/libs/agent_framework/docs/README_TIM_OBSERVER_PAYLOAD_FIX.md index f5a0c9d..2e19618 100644 --- a/libs/agent_framework/docs/README_TIM_OBSERVER_PAYLOAD_FIX.md +++ b/libs/agent_framework/docs/README_TIM_OBSERVER_PAYLOAD_FIX.md @@ -17,6 +17,7 @@ Esta versão corrige dois gaps da migração do `agent_framework_oci`: - Publica flat por padrão. - Mantém modo legado por configuração. - Exclui `NOC.*` do Pub/Sub por padrão, seguindo a lib antiga. + - Permite excluir tipos de evento específicos do Pub/Sub por configuração. - Injeta `sequence` automaticamente no payload flat antes do publish. - `src/agent_framework/analytics/tim_sequence.py` @@ -45,6 +46,10 @@ Esta versão corrige dois gaps da migração do `agent_framework_oci`: PUBSUB_PAYLOAD_MODE=flat PUBSUB_EXCLUDE_NOC=true +# Lista opcional, separada por vírgulas, de tipos de evento não publicados no Pub/Sub. +# Os eventos continuam disponíveis para os demais destinos de observabilidade. +PUBSUB_EXCLUDED_EVENT_TYPES=GRL.NATIVE_OUTPUT_GUARDRAILS + # Sequence automático por sessão no payload Pub/Sub flat PUBSUB_SEQUENCE_ENABLED=true diff --git a/libs/agent_framework/src/agent_framework.egg-info/SOURCES.txt b/libs/agent_framework/src/agent_framework.egg-info/SOURCES.txt index 15b07a8..50d3a83 100644 --- a/libs/agent_framework/src/agent_framework.egg-info/SOURCES.txt +++ b/libs/agent_framework/src/agent_framework.egg-info/SOURCES.txt @@ -1,6 +1,8 @@ pyproject.toml src/agent_framework/__init__.py +src/agent_framework/gateway_policy_context.py src/agent_framework/observer.py +src/agent_framework/runtime_mcp_gateway_adapter.py src/agent_framework.egg-info/PKG-INFO src/agent_framework.egg-info/SOURCES.txt src/agent_framework.egg-info/dependency_links.txt @@ -11,6 +13,8 @@ src/agent_framework/analytics/composite_publisher.py src/agent_framework/analytics/event_builder.py src/agent_framework/analytics/factory.py src/agent_framework/analytics/publisher.py +src/agent_framework/analytics/tim_payload_mapper.py +src/agent_framework/analytics/tim_sequence.py src/agent_framework/analytics/providers/__init__.py src/agent_framework/analytics/providers/kafka.py src/agent_framework/analytics/providers/langfuse.py @@ -32,6 +36,8 @@ src/agent_framework/config/agent_registry.py src/agent_framework/config/settings.py src/agent_framework/events/__init__.py src/agent_framework/events/oci_streaming.py +src/agent_framework/gateways/__init__.py +src/agent_framework/gateways/mcp_gateway_client.py src/agent_framework/global_supervisor/__init__.py src/agent_framework/global_supervisor/client.py src/agent_framework/global_supervisor/config.py @@ -148,6 +154,7 @@ src/agent_framework/observability/langgraph_telemetry.py src/agent_framework/observability/llm_advisors.py src/agent_framework/observability/noc_contract.py src/agent_framework/observability/noc_events.py +src/agent_framework/observability/noc_otel.py src/agent_framework/observability/observer.py src/agent_framework/observability/otel.py src/agent_framework/observability/streaming_events.py diff --git a/libs/agent_framework/src/agent_framework/analytics/providers/langfuse.py b/libs/agent_framework/src/agent_framework/analytics/providers/langfuse.py index f999f11..7e521ba 100644 --- a/libs/agent_framework/src/agent_framework/analytics/providers/langfuse.py +++ b/libs/agent_framework/src/agent_framework/analytics/providers/langfuse.py @@ -328,6 +328,10 @@ class LangfuseAnalyticsPublisher(AnalyticsPublisher): "parent_observation_id": body.get("parent_observation_id") or metadata.get("parent_observation_id") or _current_parent_observation_id(), }) + # Keep correlation metadata on the trace, but do not turn every control + # event code into a trace tag. IC/NOC/GRL are represented by the child + # observation below; tags are not a substitute for the event span and + # high-cardinality event-code tags make the trace harder to inspect. self._update_current_trace(langfuse_metadata) # Prefer current/correlated observation API. For internal/technical events, @@ -349,7 +353,8 @@ class LangfuseAnalyticsPublisher(AnalyticsPublisher): _update_observation(observation, output={"published": True}) return except Exception: - logger.debug("Falha ao publicar Langfuse observation para %s", effective_event_type, exc_info=True) + log = logger.warning if is_internal else logger.debug + log("Falha ao publicar Langfuse observation para %s", effective_event_type, exc_info=True) if is_internal or is_technical: return @@ -388,12 +393,6 @@ class LangfuseAnalyticsPublisher(AnalyticsPublisher): try: kwargs: dict[str, Any] = { "metadata": {k: v for k, v in metadata.items() if v is not None}, - "tags": [tag for tag, enabled in ( - ("ic", metadata.get("ic")), - ("noc", metadata.get("noc")), - ("grl", metadata.get("grl")), - (str(metadata.get("tag")), metadata.get("tag")), - ) if enabled], } session_id = metadata.get("sessionId") or metadata.get("session_id") if session_id: diff --git a/libs/agent_framework/src/agent_framework/analytics/providers/pubsub.py b/libs/agent_framework/src/agent_framework/analytics/providers/pubsub.py index 2a98637..92efb24 100644 --- a/libs/agent_framework/src/agent_framework/analytics/providers/pubsub.py +++ b/libs/agent_framework/src/agent_framework/analytics/providers/pubsub.py @@ -41,6 +41,11 @@ class PubSubAnalyticsPublisher(AnalyticsPublisher): self.timeout_seconds = float(timeout_seconds or os.getenv("GCP_PUBSUB_TIMEOUT_SECONDS") or 30) self.payload_mode = (os.getenv("PUBSUB_PAYLOAD_MODE") or os.getenv("ANALYTICS_PUBSUB_PAYLOAD_MODE") or "flat").strip().lower() self.exclude_noc = (os.getenv("PUBSUB_EXCLUDE_NOC") or "true").strip().lower() in {"1", "true", "yes", "y", "on"} + self.excluded_event_types = { + item.strip().upper() + for item in os.getenv("PUBSUB_EXCLUDED_EVENT_TYPES", "").split(",") + if item.strip() + } from google.cloud import pubsub_v1 # type: ignore @@ -72,6 +77,11 @@ class PubSubAnalyticsPublisher(AnalyticsPublisher): raise ValueError("Configure GCP_PUBSUB_TOPIC_PATH, AGENT_PUBSUB_TOPIC ou GCP_PROJECT_ID + GCP_PUBSUB_TOPIC") async def publish(self, event_type: str, payload: dict[str, Any]) -> None: + event_key = str(event_type).upper() + if event_key in self.excluded_event_types: + logger.debug("analytics.pubsub.skipped_event event_type=%s", event_type) + return + metadata = payload.get("metadata") if isinstance(payload, dict) else None is_noc = str(event_type).startswith("NOC.") or (isinstance(metadata, dict) and metadata.get("noc") is True) if is_noc and self.exclude_noc: diff --git a/libs/agent_framework/src/agent_framework/config/settings.py b/libs/agent_framework/src/agent_framework/config/settings.py index 247381c..455b976 100644 --- a/libs/agent_framework/src/agent_framework/config/settings.py +++ b/libs/agent_framework/src/agent_framework/config/settings.py @@ -35,7 +35,7 @@ class Settings(BaseSettings): # config_file = ~/.oci/config profile (default/local development) # instance_principal = OCI Instance Principal signer (Compute/OKE without API key) # resource_principal = OCI Resource Principal signer (Functions/resource principal contexts) - OCI_AUTH_MODE: Literal['config_file','instance_principal','resource_principal'] = 'config_file' + OCI_AUTH_MODE: Literal['config_file','instance_principal','resource_principal', 'oke_workload_identity'] = 'config_file' OCI_CONFIG_FILE: str = '~/.oci/config' OCI_PROFILE: str = 'DEFAULT' OCI_COMPARTMENT_ID: str | None = None diff --git a/libs/agent_framework/src/agent_framework/observability/telemetry.py b/libs/agent_framework/src/agent_framework/observability/telemetry.py index 7d56239..6605d1f 100644 --- a/libs/agent_framework/src/agent_framework/observability/telemetry.py +++ b/libs/agent_framework/src/agent_framework/observability/telemetry.py @@ -77,7 +77,10 @@ _COMPACT_SUPPRESSED_SPAN_PREFIXES = ( "workflow.routing_decision", "workflow.supervisor_review", ) -_COMPACT_VISIBLE_EVENT_PREFIXES = ("AGA.", "NOC.") +# Control events remain first-class observations even in compact mode. Compact +# mode suppresses low-level workflow noise, but IC/NOC payloads are operational +# evidence and must stay inspectable as child spans in Langfuse. +_COMPACT_VISIBLE_EVENT_PREFIXES = ("IC.", "AGA.", "NOC.") def _raw_correlation_id(attrs: dict[str, Any] | None = None) -> str | None: diff --git a/libs/agent_framework/src/agent_framework/oci/auth.py b/libs/agent_framework/src/agent_framework/oci/auth.py index 20a5656..4cd763e 100644 --- a/libs/agent_framework/src/agent_framework/oci/auth.py +++ b/libs/agent_framework/src/agent_framework/oci/auth.py @@ -41,6 +41,13 @@ def get_oci_config_and_signer(settings: Any) -> tuple[dict[str, Any], Any | None logger.info("OCI auth resolved with resource principal region=%s", config.get("region")) return config, signer + if mode in {"oke_workload_identity", "oke_workload_identity"}: + signer = oci.auth.signers.get_oke_workload_identity_resource_principal_signer() + config = {"region": region or getattr(signer, "region", None)} + logger.info("OCI auth resolved with OKE workload identity region=%s", config.get("region")) + return config, signer + + raise ValueError( - "Unsupported OCI_AUTH_MODE=%r. Use config_file, instance_principal or resource_principal." % mode + "Unsupported OCI_AUTH_MODE=%r. Use config_file, instance_principal, resource_principal or oke_workload_identity." % mode ) diff --git a/templates/agent_template_backend/data/agent_framework.db b/templates/agent_template_backend/data/agent_framework.db index d1d18fd..99e9a60 100644 Binary files a/templates/agent_template_backend/data/agent_framework.db and b/templates/agent_template_backend/data/agent_framework.db differ diff --git a/tests/unit/test_pubsub_analytics_publisher.py b/tests/unit/test_pubsub_analytics_publisher.py new file mode 100644 index 0000000..d0046f9 --- /dev/null +++ b/tests/unit/test_pubsub_analytics_publisher.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import sys +import types + +import pytest + +from agent_framework.analytics.providers.pubsub import PubSubAnalyticsPublisher + + +class _FakeFuture: + def result(self, timeout: float | None = None) -> str: + return "fake-message-id" + + +class _FakePublisherClient: + def __init__(self) -> None: + self.calls: list[tuple[str, bytes, dict[str, str]]] = [] + + def publish(self, topic_path: str, *, data: bytes, **kwargs: str) -> _FakeFuture: + self.calls.append((topic_path, data, kwargs)) + return _FakeFuture() + + +@pytest.fixture +def pubsub_publisher(monkeypatch: pytest.MonkeyPatch) -> PubSubAnalyticsPublisher: + client = _FakePublisherClient() + pubsub_v1 = types.ModuleType("google.cloud.pubsub_v1") + pubsub_v1.PublisherClient = lambda: client # type: ignore[attr-defined] + google_cloud = types.ModuleType("google.cloud") + google_cloud.pubsub_v1 = pubsub_v1 # type: ignore[attr-defined] + google = types.ModuleType("google") + google.cloud = google_cloud # type: ignore[attr-defined] + + monkeypatch.setitem(sys.modules, "google", google) + monkeypatch.setitem(sys.modules, "google.cloud", google_cloud) + monkeypatch.setitem(sys.modules, "google.cloud.pubsub_v1", pubsub_v1) + monkeypatch.setenv("PUBSUB_EXCLUDED_EVENT_TYPES", "GRL.NATIVE_OUTPUT_GUARDRAILS") + monkeypatch.setenv("PUBSUB_PAYLOAD_MODE", "legacy") + + publisher = PubSubAnalyticsPublisher(topic_path="projects/test/topics/analytics") + publisher.client = client + return publisher + + +@pytest.mark.asyncio +async def test_excluded_event_is_not_sent_to_pubsub(pubsub_publisher: PubSubAnalyticsPublisher) -> None: + await pubsub_publisher.publish("GRL.NATIVE_OUTPUT_GUARDRAILS", {"session_id": "session-1"}) + + assert pubsub_publisher.client.calls == [] + + +@pytest.mark.asyncio +async def test_non_excluded_event_is_sent_to_pubsub(pubsub_publisher: PubSubAnalyticsPublisher) -> None: + await pubsub_publisher.publish("GRL.002", {"session_id": "session-1"}) + + assert len(pubsub_publisher.client.calls) == 1 + topic_path, _, attributes = pubsub_publisher.client.calls[0] + assert topic_path == "projects/test/topics/analytics" + assert attributes["event_type"] == "GRL.002" diff --git a/tests/unit/test_telemetry_langfuse_compact.py b/tests/unit/test_telemetry_langfuse_compact.py index 29b7634..785c4f6 100644 --- a/tests/unit/test_telemetry_langfuse_compact.py +++ b/tests/unit/test_telemetry_langfuse_compact.py @@ -2,7 +2,8 @@ from __future__ import annotations import pytest -from agent_framework.observability.context import clear_observability_context +from agent_framework.analytics.providers.langfuse import LangfuseAnalyticsPublisher +from agent_framework.observability.context import clear_observability_context, set_observability_context from agent_framework.observability.telemetry import Telemetry @@ -61,6 +62,7 @@ class FakeLangfuse: def __init__(self, *, legacy_api: bool = False): self.observations = [] self.propagations = [] + self.trace_updates = [] self.flush_count = 0 self.api = FakeApi() if legacy_api else None @@ -72,6 +74,9 @@ class FakeLangfuse: def propagate_attributes(self, **kwargs): return FakePropagationContext(self, kwargs) + def update_current_trace(self, **kwargs): + self.trace_updates.append(kwargs) + def flush(self): self.flush_count += 1 @@ -104,31 +109,61 @@ def telemetry_with_fake_langfuse(*, legacy_api: bool = False): @pytest.mark.asyncio -async def test_compact_keeps_root_output_and_shows_only_aga_noc_events(): +async def test_compact_keeps_root_output_and_shows_ic_aga_noc_as_spans(): clear_observability_context() telemetry = telemetry_with_fake_langfuse() async with telemetry.span("agent.gateway_message", session_id="s1", input={"request": "cms"}, _root_span=True) as span: - await telemetry.event("IC.INTERNAL", {"step": "hidden"}, kind="ic") + await telemetry.event("IC.INTERNAL", {"step": "visible"}, kind="ic") await telemetry.event("NOC.001", {"step": "visible"}, kind="noc") await telemetry.event("AGA.010", {"step": "visible"}, kind="ic") span.set_output({"answer": "ok"}) names = [obs.kwargs["name"] for obs in telemetry.langfuse.observations] - assert names == ["agent.gateway_message", "NOC.001", "AGA.010"] + assert names == ["agent.gateway_message", "IC.INTERNAL", "NOC.001", "AGA.010"] root = telemetry.langfuse.observations[0] assert root.updates[-1]["input"] == {"request": "cms"} assert root.updates[-1]["output"] == {"answer": "ok"} assert root.trace_io_updates[-1] == {"input": {"request": "cms"}, "output": {"answer": "ok"}} - assert telemetry.langfuse.observations[1].kwargs.get("trace_context") is None - assert telemetry.langfuse.observations[2].kwargs.get("trace_context") is None + for observation in telemetry.langfuse.observations[1:]: + assert observation.kwargs.get("trace_context") is None + assert observation.kwargs["as_type"] == "span" + + ic = telemetry.langfuse.observations[1] + assert ic.kwargs["input"]["step"] == "visible" + assert ic.updates[-1]["input"]["step"] == "visible" + assert ic.updates[-1]["output"] == {"status": "ok"} assert telemetry.langfuse.propagations[-1]["trace_name"] == "agent.gateway_message" aggregated = root.updates[-1]["metadata"]["aggregated_events"] assert [event["name"] for event in aggregated] == ["IC.INTERNAL", "NOC.001", "AGA.010"] +@pytest.mark.asyncio +async def test_analytics_control_event_is_a_span_and_not_a_trace_tag(): + clear_observability_context() + set_observability_context(request_id="req-1", trace_id="req-1", session_id="s1") + langfuse = FakeLangfuse() + publisher = LangfuseAnalyticsPublisher(langfuse=langfuse) + envelope = { + "eventType": "IC.ORDER_CONFIRMED", + "source": "agent_framework", + "payload": {"tag": "IC.ORDER_CONFIRMED", "order_id": "order-1"}, + "metadata": {"ic": True}, + } + + await publisher.publish("IC.ORDER_CONFIRMED", envelope) + + assert [obs.kwargs["name"] for obs in langfuse.observations] == ["IC.ORDER_CONFIRMED"] + observation = langfuse.observations[0] + assert observation.kwargs["as_type"] == "span" + assert observation.kwargs["input"] == envelope + assert observation.updates[-1]["output"] == {"published": True} + assert len(langfuse.trace_updates) == 1 + assert "tags" not in langfuse.trace_updates[0] + + @pytest.mark.asyncio async def test_compact_generation_records_io_model_parameters_and_usage_details(): clear_observability_context()