diff --git a/.github/workflows/sdk-smoke.yml b/.github/workflows/sdk-smoke.yml index 8589e0e0d..661e86ade 100644 --- a/.github/workflows/sdk-smoke.yml +++ b/.github/workflows/sdk-smoke.yml @@ -89,6 +89,7 @@ jobs: tests/harness/test_simulation_mark_bootstrap.py tests/harness/test_process_preflight.py tests/providers + tests/harness/test_observability.py tests/harness/test_scenario_source_provision_agent.py tests/harness/test_framework_detection.py tests/harness/test_hosted_entrypoint.py diff --git a/src/fi/alk/harness/hosted_authoring_entrypoint.py b/src/fi/alk/harness/hosted_authoring_entrypoint.py index 4ca795fe2..20762d291 100644 --- a/src/fi/alk/harness/hosted_authoring_entrypoint.py +++ b/src/fi/alk/harness/hosted_authoring_entrypoint.py @@ -71,6 +71,27 @@ def _platform_simulator_values(values: dict[str, str]) -> dict[str, str]: } +# In secrets.json only names a customer can't declare may be kept: SIMULATOR_* and FI_* are the +# platform's. A customer may declare GOOGLE_CLOUD_PROJECT, so under that name it is masked there. +_PLATFORM_ONLY_SETTINGS = frozenset( + name + for name in observability.NON_SECRET_SETTINGS + if name.startswith(("SIMULATOR_", "FI_")) +) + + +def _span_secret_values( + all_values: dict[str, str], gateway_values: dict[str, str] +) -> tuple[str, ...]: + """The values spans replace whole: everything in both files, except the platform settings a + span keeps. Each file is read on its own and the tuples joined, so a name in both masks both + values.""" + return ( + *observability.declared_values(all_values, _PLATFORM_ONLY_SETTINGS), + *observability.declared_values(gateway_values), + ) + + def _configure_generation_environment(values: dict[str, str]) -> None: for name in _PASSTHROUGH: if values.get(name): @@ -194,7 +215,13 @@ def main(argv: list[str] | None = None) -> int: os.environ["ALK_FOREGROUND_COORDINATOR"] = "1" job_id, run_id, telemetry = _authoring_job_context(forwarded) if job_id: - observability.begin(job_id, run_id, telemetry) + # Spans carry tool output, and a tool can print a key. + observability.begin( + job_id, + run_id, + telemetry, + secret_values=_span_secret_values(all_values, gateway_values), + ) observability.stage("authoring") try: from .authoring_runtime_validation import RuntimeValidationError diff --git a/src/fi/alk/harness/hosted_entrypoint.py b/src/fi/alk/harness/hosted_entrypoint.py index 6ccbc8a23..0e23afbe6 100644 --- a/src/fi/alk/harness/hosted_entrypoint.py +++ b/src/fi/alk/harness/hosted_entrypoint.py @@ -2272,6 +2272,19 @@ async def run_job( ) scenarios_client.configure_offline_control(events_spool.root / "control") + # Read once, at boot: the runtime deletes secrets.json on its first provision(). + customer_secret_values = deps.peek_secret_values() + declared_secret_values = ( + *customer_secret_values, + *tuple(simulator_secret_values.values()), + ) + # Spans: the customer's values go whole (no names come with them here). The simulator channel + # goes whole too, apart from the platform settings named in NON_SECRET_SETTINGS, which a span + # keeps. + span_secret_values = ( + *customer_secret_values, + *observability.declared_values(simulator_secret_values), + ) adapter = OutboundAdapter( capabilities, events_spool=events_spool, @@ -2279,10 +2292,7 @@ async def run_job( results_client=results_client, artifacts_client=artifacts_client, channel_state=channel_state, - extra_secret_values=( - *deps.peek_secret_values(), - *tuple(simulator_secret_values.values()), - ), + extra_secret_values=declared_secret_values, clock=deps.clock, flush_window_seconds=deps.flush_window_seconds, ) @@ -2491,8 +2501,12 @@ async def _fail( if broker_socket is not None: os.environ["ALK_COMPOSE_BROKER_SOCKET"] = str(broker_socket) + # Spans are a channel out too (span_secret_values says which values they replace). observability.begin( - job.job_id, job.run_id, (job.metadata or {}).get("telemetry") + job.job_id, + job.run_id, + (job.metadata or {}).get("telemetry"), + secret_values=span_secret_values, ) if job.seed is None: diff --git a/src/fi/alk/harness/observability.py b/src/fi/alk/harness/observability.py index a8de48ed3..34d225963 100644 --- a/src/fi/alk/harness/observability.py +++ b/src/fi/alk/harness/observability.py @@ -9,9 +9,16 @@ from __future__ import annotations import contextlib +import json +import logging import os +import re from typing import Any, Iterator, Mapping +from opentelemetry.sdk.trace.export import SpanExporter + +logger = logging.getLogger(__name__) + _OFF = {"0", "off", "false", "no"} _provider: Any = None @@ -28,7 +35,12 @@ def enabled() -> bool: return bool(os.getenv("FI_API_KEY") and os.getenv("FI_SECRET_KEY")) -def begin(job_id: str, run_id: str, context: Mapping[str, Any] | None = None) -> None: +def begin( + job_id: str, + run_id: str, + context: Mapping[str, Any] | None = None, + secret_values: tuple[str, ...] = (), +) -> None: """Open the job's session, once, as early as the job id is known.""" global _provider, _tracer, _context if not enabled() or _provider is not None: @@ -46,6 +58,7 @@ def begin(job_id: str, run_id: str, context: Mapping[str, Any] | None = None) -> using_user, ) from fi_instrumentation.fi_types import ProjectType + from fi_instrumentation.otel import BatchSpanProcessor _provider = register( project_name=os.getenv("FI_HARNESS_PROJECT") or "hosted-harness", @@ -54,7 +67,12 @@ def begin(job_id: str, run_id: str, context: Mapping[str, Any] | None = None) -> set_global_tracer_provider=False, verbose=False, ) + cleaner = _CleaningExporter(_exporter(), secret_values) + _swap(_provider, BatchSpanProcessor(span_exporter=cleaner)) _instrument(_provider) + # After the instrumentors, which are handed the provider and may add a processor of their + # own: any second processor would send spans to Observe uncleaned. + _only_the_cleaner(_provider, cleaner) # Only FITracer writes the baggage-carried session, user, metadata and tags onto a span. _tracer = FITracer(_provider.get_tracer(__name__)) _session.enter_context(using_session(job_id)) @@ -63,11 +81,465 @@ def begin(job_id: str, run_id: str, context: Mapping[str, Any] | None = None) -> _session.enter_context(using_user(organization)) _session.enter_context(using_metadata(dict(_context))) _session.enter_context(using_tags(_tags())) - except Exception: + except Exception as error: + # Untraced must mean nothing can still export. On fi 1.0.0 the provider's shutdown leaves + # a SimpleSpanProcessor holding fi's HTTPSpanExporter posting, so every processor is + # taken off the provider first, then shut down. + removed: Any = () + with contextlib.suppress(Exception): + removed = _provider._active_span_processor._span_processors + _provider._active_span_processor._span_processors = () + for processor in removed: + with contextlib.suppress(Exception): + processor.shutdown() + with contextlib.suppress(Exception): + if _provider is not None: + _provider.shutdown() + with contextlib.suppress(Exception): + # The class only: an error's text can repeat what it failed on. + logger.warning("tracing off: %s", type(error).__name__) _provider = None _tracer = None +def _exporter() -> Any: + from fi_instrumentation.otel import HTTPSpanExporter + + # Endpoint from FI_BASE_URL, headers from FI_API_KEY and FI_SECRET_KEY, as register() does. + return HTTPSpanExporter() + + +def _swap(provider: Any, processor: Any) -> None: + """Make processor the provider's only one, in place of whatever register() installed. + + Private surface of fi-instrumentation's TracerProvider and the SDK's: _active_span_processor, + its _span_processors tuple, and _default_processor. Clearing the flag stops fi 1.0.0's + add_span_processor from shutting the cleaner down and dropping it. No tracer from this + provider exists yet, so the processors replaced hold no spans. + """ + old = provider._active_span_processor._span_processors + provider._active_span_processor._span_processors = (processor,) + provider._default_processor = False + for each in old: + each.shutdown() + + +class _TracingNotClean(Exception): + pass + + +def _only_the_cleaner(provider: Any, cleaner: Any) -> None: + processors = provider._active_span_processor._span_processors + if ( + len(processors) != 1 + or getattr(processors[0], "span_exporter", None) is not cleaner + ): + raise _TracingNotClean() + + +# The lengths traceai cuts span text to. A cut between ":" and "@" leaves a numeric password +# looking like a port, which the URL cleaner keeps. +_CUTS = (5000, 2000, 1000, 500) +_SCHEME_CHARACTERS = frozenset( + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+.-" +) + + +def _cut(value: str) -> str: + """A value of a cut length that ends inside an address loses it, from its scheme on. + + Run on the text as traceai recorded it, before anything else shortens it. A 5,000-character + value ends in traceai's "...", which goes with the address. + """ + if len(value) not in _CUTS: + return value + return _cut_address(value) + + +def _cut_address(value: str) -> str: + at = value.rfind("://") + if at < 0 or any(character.isspace() for character in value[at + 3 :]): + return value + start = at + while start > 0 and value[start - 1] in _SCHEME_CHARACTERS: + start -= 1 + return value[:start] + "[redacted]" + + +class _CleaningExporter(SpanExporter): + """Hand the real exporter copies of each span with every string in it cleaned. + + A span that can't be gathered or copied is dropped, and a batch that can't be cleaned is + dropped whole: nothing raw reaches the exporter, and no exception reaches the SDK, whose own + log line would carry the error's text. + """ + + def __init__(self, inner: Any, secret_values: tuple[str, ...] = ()) -> None: + self._inner = inner + # The job's own secret values, replaced whole as the other outbound channels replace them + # (outbound.redact_outbound_text): a key a tool printed has no shape to recognise. + self._secrets = _secret_patterns(secret_values) + # One cleaned copy per resource, made the first time it is seen. + self._resources: dict[int, tuple[Any, Any]] = {} + + def export(self, spans: Any) -> Any: + from opentelemetry.sdk.trace.export import SpanExportResult + + error: Exception | None = None + try: + spans = list(spans) + flat: list[str] = [] + plans: list[Any] = [] + for span in spans: + try: + plans.append(_gather(span, flat, self._secrets)) + except Exception as dropped: + # Strings this span already put in flat stay there, unused. + error = error or dropped + plans.append(None) + from fi.alk.providers import agent + + # One call for the whole batch, so the cleaner logs at most once per batch. Each + # attribute's key has already been applied in _take: a value whose key names a + # secret never reaches this list. + cleaned = agent._redact(flat) + copies = [] + for span, plan in zip(spans, plans): + if plan is None: + continue + try: + copies.append(_rebuild(span, plan, cleaned, self._resource(span.resource))) + except Exception as dropped: + error = error or dropped + if not copies: + # Every span was dropped: send nothing, rather than an empty request. + return SpanExportResult.FAILURE + return self._inner.export(copies) + except Exception as failed: + error = failed + return SpanExportResult.FAILURE + finally: + if error is not None: + with contextlib.suppress(Exception): + logger.warning("trace spans dropped: %s", type(error).__name__) + + def _resource(self, resource: Any) -> Any: + known = self._resources.get(id(resource)) + if known is not None and known[0] is resource: + return known[1] + from opentelemetry.sdk.resources import Resource + + from fi.alk.providers import agent + + flat: list[str] = [] + # No declared values here: they would also hit platform configuration (the project name), + # which the platform routes the job's spans by. + plan = { + key: _take(value, flat, key, ()) + for key, value in (resource.attributes or {}).items() + } + cleaned = agent._redact(flat) + copy = Resource( + {key: _give(where, cleaned) for key, where in plan.items()}, resource.schema_url + ) + self._resources[id(resource)] = (resource, copy) + return copy + + def shutdown(self) -> None: + self._inner.shutdown() + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return self._inner.force_flush(timeout_millis) + + +# outbound.redact_outbound_text's floor: a shorter declared value would mask ordinary words. +_SHORTEST_SECRET = 8 +_SHORTEST_PIN = 4 + + +# Platform settings the simulator and gateway channels carry beside their credentials. A span +# keeps these: masking them hid every LLM span's model and sent a job's traces to a project +# named "***". Any name not listed is treated as a secret, so a new setting fails closed (masked) +# until it is added here. +NON_SECRET_SETTINGS = frozenset( + { + "AGENTCC_BASE_URL", + "ALK_BACKGROUND_NOISE", + "ALK_BACKGROUND_NOISE_CATALOG", + "ALK_GOOGLE_ADK_EXPLICITLY_APPROVED_BY_USER", + "ALK_HARNESS", + "ALK_HARNESS_MODEL", + "ALK_HARNESS_THINKING", + "ALK_JUDGE_MODE", + "ALK_VERTEX_LOCATION", + "ALK_VOICEMAIL_SCENARIOS", + "FI_BASE_URL", + "FI_HARNESS_PROJECT", + "GOOGLE_CLOUD_LOCATION", + "GOOGLE_CLOUD_PROJECT", + "GOOGLE_GENAI_USE_VERTEXAI", + "HARNESS_BACKGROUND_NOISE_VOLUME", + "HARNESS_CALLER_BARGE_IN_RATE", + "HARNESS_OBSERVABILITY", + "LIVEKIT_URL", + "SIMULATOR_LLM_FALLBACK_MODELS", + "SIMULATOR_LLM_MODEL", + "SIMULATOR_LLM_PROVIDER", + "SIMULATOR_LLM_THINKING", + "SIMULATOR_STT_MODEL", + "SIMULATOR_STT_PROVIDER", + "SIMULATOR_TTS_MODEL", + "SIMULATOR_TTS_PROVIDER", + } +) + + +def declared_values( + values: Mapping[str, str], settings: frozenset[str] = NON_SECRET_SETTINGS +) -> tuple[str, ...]: + """Every value in one alias -> value source, except those under a name in ``settings``. + + Call it once per source and join the tuples. Merging the sources first would let one value + overwrite another under the same name, and the overwritten one would never be masked. + """ + return tuple( + str(value) for alias, value in values.items() if value and alias not in settings + ) + + +def _secret_patterns(secret_values: tuple[str, ...]) -> tuple[Any, ...]: + """What _mask replaces: each value of 8 or more characters, also as JSON escapes it (text that + isn't parsed as JSON carries it escaped), longest first; then one pattern for short all-digit + values such as a 4-digit PIN, matched only between non-digits so it can't hit inside a longer + number. Other short values would hit ordinary words, so they are left.""" + long = {value for value in secret_values if len(value) >= _SHORTEST_SECRET} + forms = set(long) + for value in long: + for ascii_only in (True, False): + forms.add(json.dumps(value, ensure_ascii=ascii_only)[1:-1]) + short = { + value + for value in secret_values + if _SHORTEST_PIN <= len(value) < _SHORTEST_SECRET + and value.isascii() + and value.isdigit() + } + patterns: list[Any] = sorted(forms, key=len, reverse=True) + if short: + alternatives = "|".join(re.escape(value) for value in sorted(short, key=len, reverse=True)) + patterns.append(re.compile(rf"(? str: + for secret in secrets: + if isinstance(secret, str): + text = text.replace(secret, "***") + else: + text = secret.sub("***", text) + return text + + +def _named_secret(key: str) -> bool: + """The cleaner's own key rule (providers/agent.py _redact_in), applied to one string.""" + from fi.alk.providers.agent import _REDACTION_MARKERS + + lowered = key.lower() + return lowered == "headers" or any(marker in lowered for marker in _REDACTION_MARKERS) + + +def _take(value: Any, flat: list[str], key: str, secrets: tuple[Any, ...]) -> Any: + """Put each string of an attribute value in flat; return where to find it again. + + Only strings are cleaned: a number or a bool under a key such as gen_ai.usage.input_tokens + passes as it is. + """ + if isinstance(value, str): + return _take_text(value, flat, key, secrets) + if isinstance(value, (tuple, list)): + return ("sequence", [_take(item, flat, key, secrets) for item in value]) + return ("as is", value) + + +def _take_text(value: str, flat: list[str], key: str, secrets: tuple[Any, ...]) -> Any: + if _named_secret(key): + return ("as is", "[redacted]") + # _cut and _as_json read the value as recorded: replacing a secret first changes its length + # (so a value traceai cut is no longer of a cut length) and, inside JSON, its escaping. + cut = _cut(value) + if cut == value: + # traceai records tool input, messages and metadata as JSON. Cleaned as prose, a URL with + # a query takes the rest of its whitespace-free run, the JSON's closing quote and comma + # with it; cleaned as a structure, each string is cleaned on its own and the JSON stays + # JSON. A value cut short isn't JSON any more, so it is cleaned as text. + tree = _as_json(value) + if tree is not None: + return ("json", _take_json(tree, flat, key, secrets)) + # The cut is decided on the value as recorded, but made on the masked text: cutting first can + # take a secret's tail with the address and leave its head, which no longer matches. + flat.append(_cut_address(_mask(value, secrets)) if cut != value else _mask(value, secrets)) + return ("text", len(flat) - 1) + + +def _as_json(value: str) -> Any: + opening = value.lstrip()[:1] + if opening not in ("{", "["): + return None + try: + tree = json.loads(value) + except (ValueError, RecursionError): + return None + if not isinstance(tree, (dict, list)) or _too_deep(tree): + return None + return tree + + +# Deeper than this is cleaned as text: the walks below recurse, and the interpreter's limit is +# not ours to hit inside a batch. +_JSON_DEPTH = 64 + + +def _too_deep(tree: Any) -> bool: + """Whether tree nests past _JSON_DEPTH, measured without recursion.""" + pending = [(tree, 1)] + while pending: + node, depth = pending.pop() + if depth > _JSON_DEPTH: + return True + children = node.values() if isinstance(node, dict) else node + pending.extend( + (child, depth + 1) for child in children if isinstance(child, (dict, list)) + ) + return False + + +def _take_json(node: Any, flat: list[str], key: str, secrets: tuple[Any, ...]) -> Any: + if isinstance(node, dict): + pairs = [] + for name, item in node.items(): + # The key is a string that goes out too. The key rule reads it raw. + flat.append(_mask(str(name), secrets)) + pairs.append((len(flat) - 1, _take_json(item, flat, str(name), secrets))) + return ("dict", pairs) + if isinstance(node, list): + return ("list", [_take_json(item, flat, key, secrets) for item in node]) + if isinstance(node, str): + if _named_secret(key): + return ("as is", "[redacted]") + # A string inside JSON can itself be JSON (a tool's input dumped into a message). + tree = _as_json(node) + if tree is not None: + return ("json text", _take_json(tree, flat, key, secrets)) + flat.append(_mask(node, secrets)) + return ("text", len(flat) - 1) + return ("as is", node) + + +def _give(where: Any, cleaned: list[Any]) -> Any: + kind, body = where + if kind == "text": + return cleaned[body] + if kind == "sequence": + return tuple(_give(item, cleaned) for item in body) + if kind == "json": + return json.dumps(_give_json(body, cleaned)) + return body + + +def _give_json(where: Any, cleaned: list[Any]) -> Any: + kind, body = where + if kind == "dict": + rebuilt: dict[str, Any] = {} + for where_name, item in body: + name = cleaned[where_name] + # Two keys can clean to the same text: keep both values. + unique, count = name, 1 + while unique in rebuilt: + count += 1 + unique = f"{name} ({count})" + rebuilt[unique] = _give_json(item, cleaned) + return rebuilt + if kind == "list": + return [_give_json(item, cleaned) for item in body] + if kind == "text": + return cleaned[body] + if kind == "json text": + return json.dumps(_give_json(body, cleaned)) + return body + + +def _gather(span: Any, flat: list[str], secrets: tuple[Any, ...]) -> Any: + def mapping(attributes: Any) -> Any: + return { + key: _take(value, flat, key, secrets) for key, value in (attributes or {}).items() + } + + description = span.status.description + return ( + _take(span.name, flat, "", secrets), + mapping(span.attributes), + [ + (_take(event.name, flat, "", secrets), mapping(event.attributes)) + for event in span.events + ], + [mapping(link.attributes) for link in span.links], + _take(description, flat, "", secrets) + if isinstance(description, str) + else ("as is", description), + ) + + +def _rebuild(span: Any, plan: Any, cleaned: list[Any], resource: Any) -> Any: + from opentelemetry.attributes import BoundedAttributes + from opentelemetry.sdk import trace as sdk_trace + from opentelemetry.sdk.util import BoundedList + from opentelemetry.trace import Link, Status + + name, attributes, events, links, description = plan + + # The SDK reports a dropped count only off these two classes' .dropped, so the copy uses + # them and carries each count over. + def bounded(where: Any, dropped: int) -> Any: + rebuilt = BoundedAttributes( + maxlen=None, + attributes={key: _give(each, cleaned) for key, each in where.items()}, + immutable=True, + ) + rebuilt.dropped = dropped + return rebuilt + + new_events = BoundedList(None) + new_events.extend( + sdk_trace.Event( + _give(event_name, cleaned), + bounded(where, event.dropped_attributes), + event.timestamp, + ) + for event, (event_name, where) in zip(span.events, events) + ) + new_events.dropped = span.dropped_events + new_links = BoundedList(None) + new_links.extend( + Link(link.context, bounded(where, link.dropped_attributes)) + for link, where in zip(span.links, links) + ) + new_links.dropped = span.dropped_links + return sdk_trace.ReadableSpan( + name=_give(name, cleaned), + context=span.context, + parent=span.parent, + resource=resource, + attributes=bounded(attributes, span.dropped_attributes), + events=new_events, + links=new_links, + kind=span.kind, + status=Status(span.status.status_code, _give(description, cleaned)), + start_time=span.start_time, + end_time=span.end_time, + instrumentation_scope=span.instrumentation_scope, + ) + + def stage_event(event_type: str, name: str, payload: Mapping[str, Any] | None = None) -> None: """Drive stage spans from the harness's own stage announcements.""" if _tracer is None: diff --git a/tests/harness/test_hosted_authoring_entrypoint.py b/tests/harness/test_hosted_authoring_entrypoint.py index f89a12245..821c721cd 100644 --- a/tests/harness/test_hosted_authoring_entrypoint.py +++ b/tests/harness/test_hosted_authoring_entrypoint.py @@ -338,3 +338,103 @@ def test_good_subfolder_continues(tmp_path, monkeypatch): assert calls == ["compose", "auto"] assert not (tmp_path / "authoring" / "failure.json").exists() + + +def test_authoring_masks_every_value_in_spans_but_the_platform_settings( + tmp_path, monkeypatch +) -> None: + from unittest import mock + + from opentelemetry.sdk.resources import Resource + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + + from fi.alk.harness import observability + + secrets = tmp_path / "secrets.json" + gateway = tmp_path / "simulator-secrets.json" + job = tmp_path / "job.json" + # secrets.json: the customer's declared values, some under names the platform also uses, and + # an Observe export secret. A SIMULATOR_* model name is the platform's and must survive. + secrets.write_text( + json.dumps( + { + "RETELL_API_KEY": "retell-customer-key-1234567890", + "CUSTOMER_DSN": "customer-dsn-value-1234567890", + "OPENAI_API_KEY": "sk-customer-FAKE-AAAA2222", + "AGENTCC_API_KEY": "cust-own-agentcc-key-fake-123", + "GOOGLE_CLOUD_PROJECT": "cust-gcp-proj-fake-778899", + "FI_API_KEY": "fi-observe-export-key-1234567890", + "SIMULATOR_LLM_MODEL": "gemini-2.5-flash-lite", + } + ), + encoding="utf-8", + ) + # The gateway's default values: credentials among models and the Observe project, some under + # the same names as the customer's, and the guest-booking PIN and target number. + gateway.write_text( + json.dumps( + { + "OPENAI_API_KEY": "sk-platform-FAKE-BBBB2222", + "AGENTCC_API_KEY": "agentcc-key-1234567890", + "AGENTCC_BASE_URL": "https://gateway.example.test", + "ALK_HARNESS_MODEL": "vertex_ai/claude-sonnet-4-5", + "FI_HARNESS_PROJECT": "hosted-harness-prod", + "GOOGLE_CLOUD_PROJECT": "platform-project-1", + "ALK_CAB_GUEST_POC_PIN": "4821", + "ALK_CAB_GUEST_POC_TARGET_PHONE_NUMBER": "+15550104477", + "SOME_NEW_GATEWAY_VALUE": "unlisted-gateway-value-42", + } + ), + encoding="utf-8", + ) + job.write_text(json.dumps({"job_id": "job-1", "run_id": "run-1"}), encoding="utf-8") + monkeypatch.setattr(entrypoint, "_SECRETS_PATH", secrets) + monkeypatch.setattr(entrypoint, "_SIMULATOR_SECRETS_PATH", gateway) + monkeypatch.setattr(entrypoint, "_ADC_PATH", tmp_path / "adc.json") + monkeypatch.setattr(entrypoint, "_TARGET_SECRETS_PATH", tmp_path / "target.json") + monkeypatch.setattr( + entrypoint, "authoring_main", lambda argv, *, validate_runtime: 0 + ) + handed: list[tuple[str, ...]] = [] + + def begin(job_id, run_id, context=None, secret_values=()): + handed.append(tuple(secret_values)) + + monkeypatch.setattr(observability, "begin", begin) + + with mock.patch.dict(entrypoint.os.environ): + assert entrypoint.main([str(job)]) == 0 + + assert len(handed) == 1 + memory = InMemorySpanExporter() + provider = TracerProvider( + resource=Resource({"project_name": "hosted-harness-prod"}) + ) + provider.add_span_processor(SimpleSpanProcessor(memory)) + with provider.get_tracer("test").start_as_current_span( + "llm", + attributes={ + "gen_ai.request.model": "vertex_ai/claude-sonnet-4-5", + "llm.model_name": "gemini-2.5-flash-lite", + "input.value": "a agentcc-key-1234567890 b retell-customer-key-1234567890 " + "c customer-dsn-value-1234567890 d sk-customer-FAKE-AAAA2222 " + "e cust-own-agentcc-key-fake-123 f cust-gcp-proj-fake-778899 " + "g fi-observe-export-key-1234567890 h sk-platform-FAKE-BBBB2222 " + "i pin 4821. j +15550104477 k 14821 l unlisted-gateway-value-42", + }, + ): + pass + sink = InMemorySpanExporter() + observability._CleaningExporter(sink, handed[0]).export(memory.get_finished_spans()) + + (copy,) = sink.get_finished_spans() + assert copy.attributes["gen_ai.request.model"] == "vertex_ai/claude-sonnet-4-5" + assert copy.attributes["llm.model_name"] == "gemini-2.5-flash-lite" + assert copy.resource.attributes["project_name"] == "hosted-harness-prod" + assert copy.attributes["input.value"] == ( + "a *** b *** c *** d *** e *** f *** g *** h *** i pin ***. j *** k 14821 l ***" + ) diff --git a/tests/harness/test_hosted_entrypoint.py b/tests/harness/test_hosted_entrypoint.py index 7af13a83e..8923f5976 100644 --- a/tests/harness/test_hosted_entrypoint.py +++ b/tests/harness/test_hosted_entrypoint.py @@ -5515,3 +5515,84 @@ def test_the_activity_printed_to_stdout_loses_url_credentials(capsys, monkeypatc assert " [grep https://hooks.example/v1" in printed assert " > agent.py:3: hook = https://hooks.example/v1\n > agent.py:4: ok" in printed assert "The agent posts to https://hooks.example/v1 after." in printed + + +def test_the_job_masks_every_value_in_spans_but_the_platform_settings( + monkeypatch, +) -> None: + from opentelemetry.sdk.resources import Resource + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + + from fi.alk.harness import observability + + harness = _build_harness( + scenarios=[ + FakeScenario("passing", "platform-passing", [FakeSubGoal("holds", True)]) + ] + ) + harness.deps.secrets_path.write_text( + json.dumps( + { + "CUSTOMER_DSN": "customer-dsn-value-1234567890", + "OPENAI_API_KEY": "sk-customer-FAKE-AAAA2222", + "GOOGLE_CLOUD_PROJECT": "cust-gcp-proj-fake-778899", + } + ), + encoding="utf-8", + ) + harness.deps.simulator_secrets_path = harness.tmp / "simulator-secrets.json" + # The gateway's values: credentials among models and the Observe project, one name the customer + # also declared, and a value under a name not on the settings list. + harness.deps.simulator_secrets_path.write_text( + json.dumps( + { + "AGENTCC_API_KEY": "agentcc-key-1234567890", + "OPENAI_API_KEY": "sk-platform-FAKE-BBBB2222", + "SIP_OUTBOUND_FROM_NUMBER": "+15550109988", + "ALK_HARNESS_MODEL": "vertex_ai/claude-sonnet-4-5", + "SIMULATOR_LLM_MODEL": "gemini-2.5-flash-lite", + "FI_HARNESS_PROJECT": "hosted-harness-prod", + "GOOGLE_CLOUD_PROJECT": "platform-project-1", + } + ), + encoding="utf-8", + ) + handed: list[tuple[str, ...]] = [] + + def begin(job_id, run_id, context=None, secret_values=()): + handed.append(tuple(secret_values)) + + monkeypatch.setattr(observability, "begin", begin) + + with mock.patch.dict(os.environ): + assert _run(harness) == he.EXIT_OK + + assert len(handed) == 1 + memory = InMemorySpanExporter() + provider = TracerProvider( + resource=Resource({"project_name": "hosted-harness-prod"}) + ) + provider.add_span_processor(SimpleSpanProcessor(memory)) + with provider.get_tracer("test").start_as_current_span( + "llm", + attributes={ + "gen_ai.request.model": "vertex_ai/claude-sonnet-4-5", + "llm.model_name": "gemini-2.5-flash-lite", + "input.value": "a agentcc-key-1234567890 b customer-dsn-value-1234567890 " + "c sk-customer-FAKE-AAAA2222 d cust-gcp-proj-fake-778899 " + "e sk-platform-FAKE-BBBB2222 f +15550109988", + }, + ): + pass + sink = InMemorySpanExporter() + observability._CleaningExporter(sink, handed[0]).export(memory.get_finished_spans()) + + (copy,) = sink.get_finished_spans() + assert copy.attributes["gen_ai.request.model"] == "vertex_ai/claude-sonnet-4-5" + assert copy.attributes["llm.model_name"] == "gemini-2.5-flash-lite" + assert copy.resource.attributes["project_name"] == "hosted-harness-prod" + assert copy.attributes["input.value"] == "a *** b *** c *** d *** e *** f ***" diff --git a/tests/harness/test_observability.py b/tests/harness/test_observability.py index a83021eeb..080e24b88 100644 --- a/tests/harness/test_observability.py +++ b/tests/harness/test_observability.py @@ -1,14 +1,27 @@ -"""The harness's span shape: stages and scenarios must parent the calls made inside them.""" +"""The harness's span shape: stages and scenarios must parent the calls made inside them. + +And what reaches Observe: begin() installs a cleaning exporter as the provider's only processor, +so every string a span carries is cleaned of credentials first (TH-9504). +""" from __future__ import annotations +import asyncio +import json +import logging + import pytest -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import SimpleSpanProcessor +import requests +from opentelemetry.sdk.trace import SpanLimits, TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import Link, SpanContext, Status, StatusCode, TraceFlags from fi.alk.harness import observability +# The real one, kept before any fixture swaps it out. +_REAL_EXPORTER = observability._exporter + @pytest.fixture def exporter(monkeypatch): @@ -72,3 +85,1124 @@ def test_tracing_off_is_silent_rather_than_fatal(monkeypatch): observability.stage_event("harness.stage.started", "environment") with observability.scenario("k", 0) as span: assert span is None + + +URL = "https://u:pw@h/x?key=v" +CLEAN = "https://h/x" + + +def _leaks(text): + return "pw@" in text or "key=v" in text + + +class _Run: + """The real begin() and register(), with the cleaner's exporter in memory.""" + + def __init__(self, memory, sent, posts, restore_export): + self.memory = memory + self.sent = sent + self.posts = posts + self.provider = None + self._restore_export = restore_export + + def begin(self, secret_values=()): + observability.begin("job", "run", secret_values=secret_values) + self.provider = observability._provider + return self + + def live_transport(self): + """Let fi's HTTPSpanExporter run for real, so a span it sends shows in posts.""" + self._restore_export() + + @property + def tracer(self): + return self.provider.get_tracer("test") + + @property + def cleaner(self): + return self.provider._active_span_processor._span_processors[0].span_exporter + + def spans(self): + self.provider.force_flush() + return self.memory.get_finished_spans() + + def one(self): + spans = self.spans() + assert len(spans) == 1 + return spans[0] + + +@pytest.fixture +def _real_begin(monkeypatch, caplog): + from fi_instrumentation import otel as fi_otel + + monkeypatch.setenv("FI_API_KEY", "fake-api-key") + monkeypatch.setenv("FI_SECRET_KEY", "fake-secret-key") + # A closed port: nothing is listening, and the recorders below stop any send first. + monkeypatch.setenv("FI_BASE_URL", "http://127.0.0.1:9") + monkeypatch.delenv("HARNESS_OBSERVABILITY", raising=False) + monkeypatch.setattr(observability, "_provider", None) + monkeypatch.setattr(observability, "_tracer", None) + monkeypatch.setattr(observability, "_context", {}) + memory = InMemorySpanExporter() + monkeypatch.setattr(observability, "_exporter", lambda: memory) + + sent = [] + + def recorder(self, spans, *args, **kwargs): + sent.append(list(spans)) + return SpanExportResult.SUCCESS + + original_export = fi_otel.HTTPSpanExporter.export + # On the class, so any exporter register() built records what it was sent. + monkeypatch.setattr(fi_otel.HTTPSpanExporter, "export", recorder) + + posts = [] + + def post(self, url, *args, **kwargs): + posts.append(url) + # Accepted, so a send let through doesn't retry with backoff. + response = requests.Response() + response.status_code = 200 + return response + + monkeypatch.setattr(requests.Session, "post", post) + caplog.set_level(logging.WARNING) + run = _Run( + memory, + sent, + posts, + lambda: monkeypatch.setattr( + fi_otel.HTTPSpanExporter, "export", original_export + ), + ) + yield run + observability.end() + if run.provider is not None: + # Stops the batch processor's worker thread; end() only flushes. + run.provider.shutdown() + + +def _warnings(caplog): + return [ + record.getMessage() + for record in caplog.records + if record.name == observability.logger.name + and record.levelno >= logging.WARNING + ] + + +def _readable(name="span", limits=None, links=(), **attributes): + """A finished span, built on a provider of its own.""" + memory = InMemorySpanExporter() + provider = TracerProvider(span_limits=limits) if limits else TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(memory)) + with provider.get_tracer("test").start_as_current_span( + name, attributes=attributes, links=list(links) + ): + pass + return memory.get_finished_spans()[0] + + +def _context(): + return SpanContext(1, 2, is_remote=False, trace_flags=TraceFlags(1)) + + +def test_t1_a_span_is_exported_and_cleaned(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + span.set_attribute("tool.input", URL) + + assert run.one().attributes["tool.input"] == CLEAN + + +def test_t2_the_cleaner_is_the_only_processor_and_registers_exporter_gets_nothing( + _real_begin, +): + run = _real_begin.begin() + assert observability._tracer is not None + processors = run.provider._active_span_processor._span_processors + assert len(processors) == 1 + cleaner = processors[0].span_exporter + assert isinstance(cleaner, observability._CleaningExporter) + assert cleaner._inner is run.memory + assert run.provider._default_processor is False + with run.tracer.start_as_current_span("tool") as span: + span.set_attribute("tool.input", URL) + + run.one() + assert run.sent == [] + assert run.posts == [] + + +def test_t3_a_raw_processor_left_after_the_swap_turns_tracing_off( + _real_begin, monkeypatch, caplog +): + kept = [] + + def append(provider, processor): + kept.append(provider) + multi = provider._active_span_processor + multi._span_processors = multi._span_processors + (processor,) + + monkeypatch.setattr(observability, "_swap", append) + run = _real_begin.begin() + + assert observability._tracer is None + with kept[0].get_tracer("test").start_as_current_span("tool") as span: + span.set_attribute("tool.input", URL) + kept[0].force_flush() + assert run.memory.get_finished_spans() == () + assert run.sent == [] + assert _warnings(caplog) == ["tracing off: _TracingNotClean"] + + +def test_t3b_an_instrumentor_that_adds_a_processor_turns_tracing_off_and_stops_export( + _real_begin, monkeypatch, caplog +): + from fi_instrumentation import otel as fi_otel + + held = [] + kept = [] + + def adds(provider): + # As an instrumentor would: a tracer kept from the provider, and an exporter of its own. + kept.append(provider) + held.append(provider.get_tracer("instrumentor")) + provider.add_span_processor(SimpleSpanProcessor(fi_otel.HTTPSpanExporter())) + + monkeypatch.setattr(observability, "_instrument", adds) + _real_begin.live_transport() + run = _real_begin.begin() + + assert observability._tracer is None + assert _warnings(caplog) == ["tracing off: _TracingNotClean"] + span = held[0].start_span("tool") + span.set_attribute("claude_agent.tool.input", URL) + span.end() + assert run.posts == [] + assert run.memory.get_finished_spans() == () + # On any fi version, nothing is left on the provider to export through. + assert kept[0]._active_span_processor._span_processors == () + + +def test_t3c_turning_tracing_off_shuts_down_every_processor_it_takes_off( + _real_begin, monkeypatch +): + # Taking a processor off the provider is what stops it exporting; shutting it down is what + # stops its worker thread and closes its exporter, so a job that ran untraced leaves neither. + closed = [] + + class _Recording(SimpleSpanProcessor): + def shutdown(self): + closed.append(self) + super().shutdown() + + added = _Recording(InMemorySpanExporter()) + kept = [] + + def adds(provider): + kept.append(provider) + provider.add_span_processor(added) + + monkeypatch.setattr(observability, "_instrument", adds) + _real_begin.begin() + + assert observability._tracer is None + assert added in closed + assert kept[0]._active_span_processor._span_processors == () + + +def test_t3y_a_processor_added_after_begin_leaves_the_cleaner_in_place(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + run.provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter())) + with run.tracer.start_as_current_span("after"): + pass + + assert [span.name for span in run.spans()] == ["after"] + + +def test_t4_a_list_valued_attribute_is_cleaned(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + span.set_attribute("tool.args", ("a", URL)) + + assert run.one().attributes["tool.args"] == ("a", CLEAN) + + +def test_t5_events_exceptions_and_the_status_are_cleaned(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span( + "tool", record_exception=False, set_status_on_exception=False + ) as span: + span.add_event("note", {"text": "fetched " + URL}) + span.add_event("saw " + URL) + try: + raise ValueError("bad " + URL) + except ValueError as error: + span.record_exception(error) + span.set_status(Status(StatusCode.ERROR, "failed " + URL)) + + out = run.one() + note, named, exception = out.events + assert note.attributes["text"] == "fetched " + CLEAN + assert named.name == "saw " + CLEAN + assert exception.attributes["exception.message"] == "bad " + CLEAN + stacktrace = exception.attributes["exception.stacktrace"] + assert "ValueError: bad " + CLEAN in stacktrace + assert not _leaks(stacktrace) + assert out.status.status_code is StatusCode.ERROR + assert out.status.description == "failed " + CLEAN + + +def test_t6_the_span_name_is_cleaned(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("GET " + URL): + pass + + assert run.one().name == "GET " + CLEAN + + +def test_t16_link_attributes_are_cleaned(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + link = Link(_context(), {"why": URL, "many": ("a", URL)}) + with run.tracer.start_as_current_span("tool", links=[link]): + pass + + out = run.one().links[0] + assert out.attributes["why"] == CLEAN + assert out.attributes["many"] == ("a", CLEAN) + + +def test_t7_numbers_and_bools_pass_unchanged_with_no_key_to_mask_them(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("llm") as span: + span.set_attribute("gen_ai.usage.input_tokens", 12) + span.set_attribute("tool.is_error", True) + + out = run.one() + assert out.attributes["gen_ai.usage.input_tokens"] == 12 + assert out.attributes["tool.is_error"] is True + + +def test_t8_dropped_counts_are_carried_over(): + limits = SpanLimits( + max_span_attributes=1, + max_events=1, + max_links=1, + max_event_attributes=1, + max_link_attributes=1, + ) + memory = InMemorySpanExporter() + provider = TracerProvider(span_limits=limits) + provider.add_span_processor(SimpleSpanProcessor(memory)) + # Each bounded list keeps its latest items. + links = [Link(_context(), {"d": 1}), Link(_context(), {"a": 1, "b": 2, "c": 3})] + with provider.get_tracer("test").start_as_current_span("span", links=links) as span: + span.set_attributes({"x": 1, "y": 2, "z": 3}) + span.add_event("first") + span.add_event("second") + span.add_event("kept", {"a": 1, "b": 2}) + original = memory.get_finished_spans()[0] + sink = InMemorySpanExporter() + + observability._CleaningExporter(sink).export([original]) + + copy = sink.get_finished_spans()[0] + assert (copy.dropped_attributes, copy.dropped_events, copy.dropped_links) == ( + 2, + 2, + 1, + ) + assert ( + copy.events[0].dropped_attributes == original.events[0].dropped_attributes == 1 + ) + assert copy.links[0].dropped_attributes == original.links[0].dropped_attributes == 2 + assert (copy.context, copy.parent) == (original.context, original.parent) + assert copy.kind == original.kind + assert (copy.start_time, copy.end_time) == (original.start_time, original.end_time) + assert copy.instrumentation_scope == original.instrumentation_scope + # A cleaned copy of the resource; with no secret in it, the same attributes. + assert copy.resource.attributes == original.resource.attributes + + +EARLIER = "see https://a.example/p?q=1 then " +CUT_TAIL = "https://deploy:8374" + + +def _ending(length, tail=CUT_TAIL, after=""): + """A value of length characters: a URL with a query, padding, then tail last.""" + end = tail + after + value = EARLIER + "x" * (length - len(EARLIER) - len(end) - 1) + " " + end + assert len(value) == length + return value + + +@pytest.mark.parametrize("length", [5000, 2000, 1000, 500]) +def test_t9_an_address_ending_a_value_cut_by_traceai_goes_whole(_real_begin, length): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + # traceai cuts at 5,000 as result[:4997] + "...". + span.set_attribute( + "tool.input", _ending(length, after="..." if length == 5000 else "") + ) + + value = run.one().attributes["tool.input"] + assert "8374" not in value + assert "q=1" not in value + assert value.endswith(" [redacted]") + + +def test_t9_a_value_not_of_a_cut_length_keeps_its_port(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + span.set_attribute("tool.input", _ending(4999)) + + value = run.one().attributes["tool.input"] + assert "q=1" not in value + assert value.endswith(" " + CUT_TAIL) + + +def test_t9c_a_cut_value_whose_last_address_is_followed_by_text_keeps_it(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + span.set_attribute("bash.command", _ending(500, after=" done")) + + assert run.one().attributes["bash.command"].endswith(" " + CUT_TAIL + " done") + + +def test_t10_an_empty_scheme_is_replaced_from_the_separator(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + span.set_attribute( + "bash.command", _ending(500, tail="${DB_SCHEME}://admin:8374") + ) + + value = run.one().attributes["bash.command"] + assert "8374" not in value + assert value.endswith(" ${DB_SCHEME}[redacted]") + + +def test_t11_traceais_embedded_tool_result_cut_is_cleaned(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("conversation") as span: + # _client_wrapper.py writes f"[Tool Result: {result[:100]}...]" mid-value. + span.set_attribute("output.full", "[Tool Result: " + URL + "...]") + span.set_attribute("output.cut", "[Tool Result: https://user:8374...]") + + out = run.one().attributes + assert not _leaks(out["output.full"]) + assert "8374" not in out["output.cut"] + assert "[redacted]" in out["output.cut"] + + +def test_t12_a_raise_while_cleaning_drops_the_batch_and_logs_only_the_class( + _real_begin, monkeypatch, caplog +): + from fi.alk.providers import agent + + run = _real_begin.begin() + assert observability._tracer is not None + span = _readable("tool", **{"tool.input": URL}) + + def raises(value, key=""): + raise ValueError("cannot parse " + URL) + + monkeypatch.setattr(agent, "_redact", raises) + caplog.clear() + + assert run.cleaner.export([span]) is SpanExportResult.FAILURE + assert run.spans() == () + for record in caplog.records: + assert not _leaks(record.getMessage()) + assert not _leaks(str(record.exc_text or "")) + assert _warnings(caplog) == ["trace spans dropped: ValueError"] + + +def test_t13_a_span_that_cannot_be_rebuilt_is_dropped_alone( + _real_begin, monkeypatch, caplog +): + from opentelemetry.sdk import trace as sdk_trace + + run = _real_begin.begin() + assert observability._tracer is not None + # Built before the patch, which would otherwise break ending them. + spans = [_readable("boom", **{"tool.input": URL}), _readable("fine")] + real = sdk_trace.ReadableSpan + + def picky(*args, **kwargs): + if kwargs.get("name") == "boom": + raise KeyError("planted") + return real(*args, **kwargs) + + monkeypatch.setattr(sdk_trace, "ReadableSpan", picky) + caplog.clear() + + run.cleaner.export(spans) + + assert [span.name for span in run.memory.get_finished_spans()] == ["fine"] + assert _warnings(caplog) == ["trace spans dropped: KeyError"] + + +def test_t14_end_flushes_through_the_cleaner(_real_begin): + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("last") as span: + span.set_attribute("tool.input", URL) + + observability.end() + + spans = run.memory.get_finished_spans() + assert [span.name for span in spans] == ["last"] + assert spans[0].attributes["tool.input"] == CLEAN + + +def test_t14b_the_cleaners_force_flush_reaches_the_inner_exporter(): + calls = [] + + class Inner(InMemorySpanExporter): + def force_flush(self, timeout_millis=30000): + calls.append(timeout_millis) + return "inner's answer" + + assert observability._CleaningExporter(Inner()).force_flush(7) == "inner's answer" + assert calls == [7] + + +def test_t15_traceais_recorded_tool_input_is_cleaned(_real_begin): + pytest.importorskip("traceai_claude_agent_sdk") + from traceai_claude_agent_sdk import ClaudeAgentInstrumentor + from traceai_claude_agent_sdk._hooks import ( + create_post_tool_use_hook, + create_pre_tool_use_hook, + ) + + # A singleton: undone first, so begin()'s _instrument hands it this provider. + instrumentor = ClaudeAgentInstrumentor() + instrumentor.uninstrument() + try: + run = _real_begin.begin() + assert observability._tracer is not None + tracer = instrumentor._tracer + pre = create_pre_tool_use_hook(tracer) + post = create_post_tool_use_hook(tracer) + asyncio.run( + pre({"tool_name": "WebFetch", "tool_input": {"url": URL}}, "tool-1", None) + ) + asyncio.run( + post({"tool_name": "WebFetch", "tool_response": "ok"}, "tool-1", None) + ) + + (span,) = [span for span in run.spans() if span.name == "tool.WebFetch"] + value = span.attributes["claude_agent.tool.input"] + assert "https://h/x" in value + assert not _leaks(value) + finally: + instrumentor.uninstrument() + + +# -- the copy's structure, the check's identity half, the inner exporter, the real exporter ----- + + +def test_r1_the_copy_keeps_parent_kind_link_context_and_event_time(): + from opentelemetry.trace import SpanKind + + memory = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(memory)) + tracer = provider.get_tracer("test") + with tracer.start_as_current_span("root"): + with tracer.start_as_current_span( + "child", kind=SpanKind.CLIENT, links=[Link(_context(), {"a": 1})] + ) as span: + span.add_event("e", timestamp=123) + original = [s for s in memory.get_finished_spans() if s.name == "child"][0] + sink = InMemorySpanExporter() + + observability._CleaningExporter(sink).export([original]) + + copy = sink.get_finished_spans()[0] + # A lost parent would reach Observe as a root, flattening the stage -> call nesting. + assert original.parent is not None + assert copy.parent == original.parent + assert copy.kind is SpanKind.CLIENT + assert copy.links[0].context == original.links[0].context + assert copy.events[0].timestamp == 123 + + +def test_r2_a_lone_processor_that_is_not_the_cleaner_turns_tracing_off( + _real_begin, monkeypatch, caplog +): + def replaces(provider): + provider._active_span_processor._span_processors = ( + SimpleSpanProcessor(InMemorySpanExporter()), + ) + + monkeypatch.setattr(observability, "_instrument", replaces) + _real_begin.begin() + + assert observability._tracer is None + assert _warnings(caplog) == ["tracing off: _TracingNotClean"] + + +def test_r3_two_spans_dropped_in_one_batch_log_one_line( + _real_begin, monkeypatch, caplog +): + from opentelemetry.sdk import trace as sdk_trace + + run = _real_begin.begin() + assert observability._tracer is not None + spans = [_readable("boom"), _readable("boom"), _readable("fine")] + real = sdk_trace.ReadableSpan + + def picky(*args, **kwargs): + if kwargs.get("name") == "boom": + raise KeyError("planted") + return real(*args, **kwargs) + + monkeypatch.setattr(sdk_trace, "ReadableSpan", picky) + caplog.clear() + run.cleaner.export(spans) + + assert _warnings(caplog) == ["trace spans dropped: KeyError"] + + +class _Inner(InMemorySpanExporter): + def __init__(self, result=SpanExportResult.FAILURE, error=None): + super().__init__() + self.result, self.error, self.closed = result, error, False + + def export(self, spans): + if self.error is not None: + raise self.error + return self.result + + def shutdown(self): + self.closed = True + + +def test_r4_the_inner_exporters_raise_is_kept_and_logged_by_class(caplog): + # A transport error's text can carry the request it failed on; the SDK would log it whole. + caplog.set_level(logging.WARNING) + inner = _Inner(error=ConnectionError("POST " + URL)) + + result = observability._CleaningExporter(inner).export([_readable()]) + + assert result is SpanExportResult.FAILURE + assert _warnings(caplog) == ["trace spans dropped: ConnectionError"] + assert not any(URL in record.getMessage() for record in caplog.records) + + +def test_r5_the_inner_exporters_result_and_shutdown_pass_through(): + inner = _Inner(result=SpanExportResult.FAILURE) + cleaner = observability._CleaningExporter(inner) + + assert cleaner.export([_readable()]) is SpanExportResult.FAILURE + cleaner.shutdown() + assert inner.closed + + +def test_r6_the_swap_shuts_down_what_register_installed(_real_begin): + from fi_instrumentation import register + from fi_instrumentation.fi_types import ProjectType + + provider = register( + project_name="p", + project_type=ProjectType.OBSERVE, + set_global_tracer_provider=False, + verbose=False, + ) + (old,) = provider._active_span_processor._span_processors + closed = [] + old.shutdown = lambda *a, **k: closed.append(True) + + observability._swap(provider, SimpleSpanProcessor(InMemorySpanExporter())) + + assert closed == [True] + provider.shutdown() + + +def _bodies(monkeypatch): + bodies = [] + + def post(self, url, *args, **kwargs): + bodies.append((url, kwargs.get("data") or b"")) + response = requests.Response() + response.status_code = 200 + return response + + monkeypatch.setattr(requests.Session, "post", post) + return bodies + + +def test_t2b_the_real_exporter_posts_clean_spans_to_registers_endpoint( + _real_begin, monkeypatch +): + # Every other test swaps _exporter out; this is the one path that reaches Observe. + monkeypatch.setattr(observability, "_exporter", _REAL_EXPORTER) + _real_begin.live_transport() + bodies = _bodies(monkeypatch) + run = _real_begin.begin() + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + span.set_attribute("tool.input", URL) + run.provider.force_flush() + + assert [url for url, _ in bodies] == ["http://127.0.0.1:9/tracer/v1/traces"] + assert b"https://h/x" in bodies[0][1] and b"pw@" not in bodies[0][1] + + +def test_a_batch_whose_every_span_is_dropped_sends_nothing(_real_begin, monkeypatch): + from fi_instrumentation import otel as fi_otel + from opentelemetry.sdk import trace as sdk_trace + + _real_begin.live_transport() + bodies = _bodies(monkeypatch) + span = _readable("boom", **{"tool.input": URL}) + + def refuses(*args, **kwargs): + raise KeyError("planted") + + monkeypatch.setattr(sdk_trace, "ReadableSpan", refuses) + + result = observability._CleaningExporter(fi_otel.HTTPSpanExporter()).export([span]) + + assert result is SpanExportResult.FAILURE + assert bodies == [] + + +# -- JSON values, declared secrets, the key rule, the resource --------------------------------- + + +def _export_one(span, secret_values=()): + sink = InMemorySpanExporter() + observability._CleaningExporter(sink, secret_values).export([span]) + return sink.get_finished_spans()[0] + + +@pytest.mark.parametrize( + "value", + [ + # traceai's own json.dumps of a WebFetch input, and the same compact. + '{"url": "https://u:pw@docs.example.com/guide?section=auth#setup", ' + '"prompt": "Summarise the auth section"}', + '{"url":"https://u:pw@h/a?b=1","next":"keep me"}', + '[{"role": "user", "content": "fetch https://h/x?key=v please"}, {"role": "x"}]', + ], +) +def test_a_json_value_is_cleaned_as_a_structure_and_stays_json(value): + copy = _export_one(_readable(**{"input.value": value})) + + out = copy.attributes["input.value"] + tree = json.loads(out) + assert "pw@" not in out and "key=v" not in out and "b=1" not in out + if isinstance(tree, dict) and "prompt" in tree: + assert tree == { + "url": "https://docs.example.com/guide", + "prompt": "Summarise the auth section", + } + if isinstance(tree, dict) and "next" in tree: + assert tree["next"] == "keep me" + if isinstance(tree, list): + assert tree[1] == {"role": "x"} + + +def test_a_json_string_inside_json_is_cleaned_as_a_structure_too(): + inner = json.dumps({"url": "https://h/a?key=v", "next": "keep me"}) + value = json.dumps({"role": "tool", "content": inner}) + + out = _export_one(_readable(**{"input.value": value})).attributes["input.value"] + + tree = json.loads(out) + assert json.loads(tree["content"]) == {"url": "https://h/a", "next": "keep me"} + + +def test_a_json_value_cut_by_traceai_is_cleaned_as_text(): + value = ('{"note": "' + "x" * 4900 + " https://deploy:8374")[:4997] + "..." + out = _export_one(_readable(**{"tool.input": value})).attributes["tool.input"] + assert "8374" not in out + + +def test_a_declared_secret_never_reaches_the_inner_exporter(): + secret = "sk-JOBSECRET-1234567890" + span = _readable( + **{ + "output.value": "OPENAI_API_KEY=" + secret, + "tool.input": '{"k": "' + secret + '"}', + } + ) + + copy = _export_one(span, (secret, "short")) + + assert secret not in copy.attributes["output.value"] + assert copy.attributes["output.value"] == "OPENAI_API_KEY=***" + assert secret not in copy.attributes["tool.input"] + + +def test_a_declared_secret_in_an_exception_is_replaced(_real_begin): + secret = "sk-JOBSECRET-1234567890" + run = _real_begin.begin(secret_values=(secret,)) + assert observability._tracer is not None + with run.tracer.start_as_current_span("tool") as span: + try: + raise RuntimeError("login failed key=" + secret) + except RuntimeError as error: + span.record_exception(error) + + (exception,) = run.one().events + assert secret not in exception.attributes["exception.message"] + assert secret not in exception.attributes["exception.stacktrace"] + + +def test_a_string_whose_key_names_a_secret_is_masked_and_a_count_is_kept(): + span = _readable( + **{ + "http.request.header.authorization": ("Bearer sk-abc-123456",), + "api_key": "plain-text-value", + "gen_ai.usage.input_tokens": 12, + "input.value": '{"usage": {"input_tokens": 12}, "token": "abc123def"}', + } + ) + + attributes = _export_one(span).attributes + + assert attributes["http.request.header.authorization"] == ("[redacted]",) + assert attributes["api_key"] == "[redacted]" + assert attributes["gen_ai.usage.input_tokens"] == 12 + assert json.loads(attributes["input.value"]) == { + "usage": {"input_tokens": 12}, + "token": "[redacted]", + } + + +def test_the_resource_is_cleaned_once_and_reused(): + from opentelemetry.sdk.resources import Resource + + memory = InMemorySpanExporter() + provider = TracerProvider(resource=Resource({"note": "postgres://u:pw@db/x?k=v"})) + provider.add_span_processor(SimpleSpanProcessor(memory)) + for name in ("one", "two"): + with provider.get_tracer("test").start_as_current_span(name): + pass + sink = InMemorySpanExporter() + + observability._CleaningExporter(sink).export(memory.get_finished_spans()) + + first, second = sink.get_finished_spans() + assert first.resource.attributes["note"] == "postgres://db/x" + assert first.resource is second.resource + + +# -- which declared values, the order, keys, depth ---------------------------------------------- + +# What the gateway sends on the simulator channel: credentials among platform configuration. +GATEWAY_CHANNEL = { + "ALK_HARNESS_MODEL": "vertex_ai/claude-sonnet-4-5", + "SIMULATOR_LLM_MODEL": "gemini-2.5-flash-lite", + "FI_HARNESS_PROJECT": "hosted-harness-prod", + "FI_BASE_URL": "https://api.futureagi.example", + "GOOGLE_CLOUD_PROJECT": "platform-project-1", + "AGENTCC_API_KEY": "agentcc-key-1234567890", + "OPENAI_API_KEY": "sk-OPENAI-1234567890", + "FI_SECRET_KEY": "fi-secret-1234567890", + "LIVEKIT_API_SECRET": "livekit-secret-1234567890", +} +GATEWAY_CREDENTIALS = { + "agentcc-key-1234567890", + "sk-OPENAI-1234567890", + "fi-secret-1234567890", + "livekit-secret-1234567890", +} + + +def test_declared_values_keeps_every_value_but_the_listed_settings(): + channel = { + **GATEWAY_CHANNEL, + "SOME_NEW_SETTING": "unlisted-value-1234", + "EMPTY": "", + } + + assert set(observability.declared_values(channel)) == GATEWAY_CREDENTIALS | { + "unlisted-value-1234" + } + assert observability.declared_values({"A": "x-value-123"}, frozenset({"A"})) == () + + +def test_the_non_secret_settings_are_simulator_aliases_and_none_names_a_credential(): + from fi.alk.harness.hosted_entrypoint import _SIMULATOR_SECRET_ALIASES + + assert observability.NON_SECRET_SETTINGS <= _SIMULATOR_SECRET_ALIASES + assert not [ + name + for name in observability.NON_SECRET_SETTINGS + if observability._named_secret(name) + ] + + +def _platform_span(): + """An LLM span on a resource that names the Observe project, carrying a real credential.""" + from opentelemetry.sdk.resources import Resource + + memory = InMemorySpanExporter() + provider = TracerProvider( + resource=Resource({"project_name": "hosted-harness-prod"}) + ) + provider.add_span_processor(SimpleSpanProcessor(memory)) + with provider.get_tracer("test").start_as_current_span( + "llm", + attributes={ + "gen_ai.request.model": "vertex_ai/claude-sonnet-4-5", + "llm.model_name": "gemini-2.5-flash-lite", + "input.value": "the key is sk-OPENAI-1234567890", + }, + ): + pass + return memory.get_finished_spans()[0] + + +def _assert_platform_config_survives(secret_values): + copy = _export_one(_platform_span(), secret_values) + + assert copy.attributes["gen_ai.request.model"] == "vertex_ai/claude-sonnet-4-5" + assert copy.attributes["llm.model_name"] == "gemini-2.5-flash-lite" + assert copy.resource.attributes["project_name"] == "hosted-harness-prod" + assert copy.attributes["input.value"] == "the key is ***" + + +def test_the_gateways_default_values_leave_models_and_project_unchanged(): + _assert_platform_config_survives(observability.declared_values(GATEWAY_CHANNEL)) + + +def test_declared_values_never_touch_the_resource(): + # Even a value that is the project's name: the platform routes the job's spans by it. + span = _platform_span() + + copy = _export_one(span, ("hosted-harness-prod",)) + + assert copy.resource.attributes["project_name"] == "hosted-harness-prod" + + +def test_a_declared_value_under_eight_characters_is_not_replaced(): + copy = _export_one(_readable(**{"output.value": "word abc1234 ok"}), ("abc1234",)) + + assert copy.attributes["output.value"] == "word abc1234 ok" + + +def test_a_short_all_digit_value_is_replaced_only_as_a_whole_number(): + copy = _export_one( + _readable(**{"output.value": "pin 4821, again 4821. not 14821 or 48210"}), + ("4821",), + ) + + assert copy.attributes["output.value"] == "pin ***, again ***. not 14821 or 48210" + + +def test_a_declared_value_under_four_digits_is_not_replaced(): + copy = _export_one(_readable(**{"output.value": "code 482 ok"}), ("482",)) + + assert copy.attributes["output.value"] == "code 482 ok" + + +def test_an_array_value_is_read_as_json(): + # As prose, the URL would take the closing quote and bracket with it and break the JSON. + out = _export_one(_readable(**{"input.value": '["https://h/x?key=v", "keep"]'})) + + assert json.loads(out.attributes["input.value"]) == ["https://h/x", "keep"] + + +def test_a_declared_secret_does_not_disarm_the_cut_rule(): + secret = "sk-JOBSECRET-1234567890" + value = _ending(2000) + value = secret + value[len(secret) :] + assert len(value) == 2000 and secret in value + + out = _export_one(_readable(**{"tool.input": value}), (secret,)) + + assert out.attributes["tool.input"].endswith(" [redacted]") + assert ( + "8374" not in out.attributes["tool.input"] + and secret not in out.attributes["tool.input"] + ) + + +def test_a_cut_json_value_is_not_read_as_json_even_when_it_is_complete_json(): + # The guard is cut == value: a value of a cut length that ends in an address goes as text, + # whatever else it parses as. + tail = " https://deploy:8374" + complete = '["' + "x" * (2000 - 4 - len(tail)) + tail + '"]' + assert len(complete) == 2000 and json.loads(complete) + + out = _export_one(_readable(**{"tool.input": complete})).attributes["tool.input"] + + assert "8374" not in out + with pytest.raises(ValueError): + json.loads(out) + + +@pytest.mark.parametrize("secret", ['pa"ss\\word-12345', "pässword-1234567"]) +def test_a_declared_secret_with_json_escapes_is_replaced_inside_json(secret): + value = json.dumps({"note": "login with " + secret, "list": [secret]}) + assert secret not in value + + out = _export_one(_readable(**{"input.value": value}), (secret,)).attributes[ + "input.value" + ] + + assert json.loads(out) == {"note": "login with ***", "list": ["***"]} + + +def test_a_declared_secret_in_a_json_key_is_replaced(): + secret = "sk-JOBVALUE-1234567890" + value = json.dumps({secret: "ok"}) + + out = _export_one(_readable(**{"input.value": value}), (secret,)).attributes[ + "input.value" + ] + + assert json.loads(out) == {"***": "ok"} + + +def test_a_json_key_is_cleaned_as_a_string(): + value = json.dumps({"https://u:pw@h/x?key=v": "ok"}) + + out = _export_one(_readable(**{"input.value": value})).attributes["input.value"] + + assert "pw" not in out and "key=v" not in out + assert list(json.loads(out).values()) == ["ok"] + + +def test_two_keys_that_clean_to_the_same_text_both_keep_their_values(): + value = json.dumps({"https://a:1@h/x": "first", "https://b:2@h/x": "second"}) + + out = json.loads( + _export_one(_readable(**{"input.value": value})).attributes["input.value"] + ) + + assert sorted(out.values()) == ["first", "second"] + assert len(out) == 2 + + +def test_a_key_that_names_a_secret_still_masks_its_value_when_the_key_is_cleaned(): + value = json.dumps({"token": "abc123def", "n": 1}) + + out = _export_one(_readable(**{"input.value": value})).attributes["input.value"] + + assert json.loads(out) == {"token": "[redacted]", "n": 1} + + +def test_json_nested_past_the_cap_is_cleaned_as_text_and_the_batch_goes_on(): + deep = "[" * 1500 + "]" * 1500 + spans = [ + _readable("deep", **{"input.value": deep, "tool.input": URL}), + _readable("normal", **{"tool.input": URL}), + ] + sink = InMemorySpanExporter() + + result = observability._CleaningExporter(sink).export(spans) + + sent = {span.name: span for span in sink.get_finished_spans()} + assert set(sent) == {"deep", "normal"} + assert sent["normal"].attributes["tool.input"] == CLEAN + # Sent as text, unchanged: there is nothing in it to clean. + assert sent["deep"].attributes["input.value"] == deep + assert not _leaks(sent["deep"].attributes["tool.input"]) + assert result is not SpanExportResult.FAILURE + + +def test_json_at_the_depth_cap_is_still_read_as_json_and_one_past_it_is_not(): + depth = observability._JSON_DEPTH + inside = "[" * depth + '"https://h/x?key=v"' + "]" * depth + past = "[" * (depth + 1) + '"https://h/x?key=v"' + "]" * (depth + 1) + + assert observability._as_json(inside) is not None + assert observability._as_json(past) is None + assert observability._as_json("[" * 100000 + "]" * 100000) is None + + +def test_a_span_that_cannot_be_gathered_is_dropped_alone(monkeypatch, caplog): + caplog.set_level(logging.WARNING) + real = observability._gather + + def picky(span, flat, secrets): + if span.name == "boom": + raise KeyError("planted") + return real(span, flat, secrets) + + monkeypatch.setattr(observability, "_gather", picky) + sink = InMemorySpanExporter() + + observability._CleaningExporter(sink).export( + [_readable("boom"), _readable("fine", **{"tool.input": URL})] + ) + + assert [span.name for span in sink.get_finished_spans()] == ["fine"] + assert sink.get_finished_spans()[0].attributes["tool.input"] == CLEAN + assert _warnings(caplog) == ["trace spans dropped: KeyError"] + + +def test_a_declared_secret_running_into_a_cut_address_goes_whole(): + # The cut is decided on the recorded length but made on the masked text, so it can't take the + # secret's tail with the address and leave its head. + secret = "A1B2C3_D4E5F6G7" + tail = " key=" + secret + "https://deploy:8374" + value = "x" * (2000 - len(tail)) + tail + + out = _export_one(_readable(**{"tool.input": value}), (secret,)).attributes[ + "tool.input" + ] + + assert "A1B2C3" not in out and "8374" not in out + assert out.endswith(" key=***[redacted]") + + +@pytest.mark.parametrize("secret", ["pässword-1234567", 'pa"ss\\word-12345']) +def test_a_secret_json_escapes_is_replaced_in_text_that_is_not_parsed(secret): + node = {"pw": "x", "note": "login " + secret} + for _ in range(65): + node = [node] + deep = json.dumps(node) + short = json.dumps({"note": "login " + secret + " " + "z" * 1200})[:1000] + + for value in (deep, short): + out = _export_one(_readable(**{"input.value": value}), (secret,)).attributes[ + "input.value" + ] + assert json.dumps(secret)[1:-1] not in out and secret not in out + + +def test_a_secret_in_json_text_inside_json_text_is_replaced(): + secret = "sk-NESTED-FAKE-1234567890" + inner = json.dumps({"k": "token " + secret}) + value = json.dumps({"role": "tool", "content": json.dumps({"message": inner})}) + + out = _export_one(_readable(**{"input.value": value}), (secret,)).attributes[ + "input.value" + ] + + assert secret not in out + level1 = json.loads(json.loads(out)["content"]) + assert json.loads(level1["message"]) == {"k": "token ***"} + + +def test_three_json_keys_that_clean_alike_keep_all_three_values(): + value = json.dumps( + {"https://a:1@h/x": "1", "https://b:2@h/x": "2", "https://c:3@h/x": "3"} + ) + + out = json.loads( + _export_one(_readable(**{"input.value": value})).attributes["input.value"] + ) + + assert sorted(out.values()) == ["1", "2", "3"]