From 3102d907a543ff0596cbc4b00973bebbc07587fb Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Wed, 11 Nov 2026 03:52:51 +0530 Subject: [PATCH 1/6] fix(harness): clean every trace span before it leaves for Observe Tool input, prompts, errors and recorded exceptions reach the platform's Observe project in span text when a hosted job carries FI keys, and no cleaner saw them. A cleaning exporter now copies each span with every string put through the kit's cleaner, and sits in front of fi's HTTP exporter as the provider's only processor. add_span_processor isn't the install step: on fi 1.0.0 it turns export off, and on 1.2.x it runs beside the raw export. begin() checks that one processor, the cleaner, remains after the instrumentors, and otherwise turns tracing off, taking every processor off the provider first. A value cut at one of traceai's lengths loses a trailing address whole, so a numeric password split from its "@" isn't kept as a port. A span that can't be copied is dropped, never sent raw. CI now runs these tests. TH-9504 Signed-off-by: khushalsonawat --- .github/workflows/sdk-smoke.yml | 1 + src/fi/alk/harness/observability.py | 224 ++++++++- tests/harness/test_observability.py | 714 +++++++++++++++++++++++++++- 3 files changed, 935 insertions(+), 4 deletions(-) diff --git a/.github/workflows/sdk-smoke.yml b/.github/workflows/sdk-smoke.yml index 728d41b49..2e005f29c 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/observability.py b/src/fi/alk/harness/observability.py index a8de48ed3..554b228c0 100644 --- a/src/fi/alk/harness/observability.py +++ b/src/fi/alk/harness/observability.py @@ -9,9 +9,14 @@ from __future__ import annotations import contextlib +import logging import os 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 @@ -46,6 +51,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 +60,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()) + _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 +74,222 @@ 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 + 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 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) -> None: + self._inner = inner + + 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 = [_gather(span, flat) for span in spans] + from fi.alk.providers import agent + + # One call for the whole batch, with no key: the cleaner logs at most once per call, + # and a key such as gen_ai.usage.input_tokens would mask a count. + cleaned = agent._redact(flat) + copies = [] + for span, plan in zip(spans, plans): + try: + copies.append(_rebuild(span, plan, cleaned)) + 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 shutdown(self) -> None: + self._inner.shutdown() + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return self._inner.force_flush(timeout_millis) + + +def _take(value: Any, flat: list[str]) -> Any: + """Put each string of an attribute value in flat; return where to find it again.""" + if isinstance(value, str): + flat.append(_cut(value)) + return ("text", len(flat) - 1) + if isinstance(value, (tuple, list)): + return ("sequence", [_take(item, flat) for item in value]) + return ("as is", value) + + +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) + return body + + +def _gather(span: Any, flat: list[str]) -> Any: + def mapping(attributes: Any) -> Any: + return {key: _take(value, flat) for key, value in (attributes or {}).items()} + + description = span.status.description + return ( + _take(span.name, flat), + mapping(span.attributes), + [(_take(event.name, flat), mapping(event.attributes)) for event in span.events], + [mapping(link.attributes) for link in span.links], + _take(description, flat) + if isinstance(description, str) + else ("as is", description), + ) + + +def _rebuild(span: Any, plan: Any, cleaned: list[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=span.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_observability.py b/tests/harness/test_observability.py index a83021eeb..19a46dea8 100644 --- a/tests/harness/test_observability.py +++ b/tests/harness/test_observability.py @@ -1,14 +1,26 @@ -"""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 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 +84,699 @@ 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): + observability.begin("job", "run") + 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 + assert copy.resource is original.resource + + +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() + + +# -- change review round 1: the copy's structure, the check's identity half, the inner exporter, +# -- and 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 == [] From 086f84c742bd2d2b5cd649714528da1dc7ac5d13 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Wed, 11 Nov 2026 11:31:05 +0530 Subject: [PATCH 2/6] fix(harness): keep JSON span values JSON, replace the job's declared secrets, and clean by key and resource Review on #280 (Dileep). A JSON span value holding a URL with a query lost its closing quote and comma to the URL cleaner, so Observe showed broken text: JSON values are now cleaned as a structure and dumped back. The job's declared secret values, which the other outbound channels already replace, are now replaced in spans too; both entrypoints pass them to begin(). A string whose attribute key names a secret is masked, as the cleaner's key rule does elsewhere, while numbers stay untouched. The resource is cleaned once per resource and reused. TH-9504 Signed-off-by: khushalsonawat --- .../harness/hosted_authoring_entrypoint.py | 9 +- src/fi/alk/harness/hosted_entrypoint.py | 16 +- src/fi/alk/harness/observability.py | 151 +++++++++++++++--- tests/harness/test_observability.py | 120 +++++++++++++- 4 files changed, 268 insertions(+), 28 deletions(-) diff --git a/src/fi/alk/harness/hosted_authoring_entrypoint.py b/src/fi/alk/harness/hosted_authoring_entrypoint.py index 4ca795fe2..16c4afeae 100644 --- a/src/fi/alk/harness/hosted_authoring_entrypoint.py +++ b/src/fi/alk/harness/hosted_authoring_entrypoint.py @@ -194,7 +194,14 @@ 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) + # Every value the job declared, as the outbound channels replace them: spans carry tool + # output, and a tool can print a key. + observability.begin( + job_id, + run_id, + telemetry, + secret_values=(*all_values.values(), *gateway_values.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..0bbbc572c 100644 --- a/src/fi/alk/harness/hosted_entrypoint.py +++ b/src/fi/alk/harness/hosted_entrypoint.py @@ -2272,6 +2272,11 @@ 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(). + declared_secret_values = ( + *deps.peek_secret_values(), + *tuple(simulator_secret_values.values()), + ) adapter = OutboundAdapter( capabilities, events_spool=events_spool, @@ -2279,10 +2284,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 +2493,12 @@ async def _fail( if broker_socket is not None: os.environ["ALK_COMPOSE_BROKER_SOCKET"] = str(broker_socket) + # The same declared values the outbound adapter replaces: spans are a channel out too. 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=declared_secret_values, ) if job.seed is None: diff --git a/src/fi/alk/harness/observability.py b/src/fi/alk/harness/observability.py index 554b228c0..e1c59b7a2 100644 --- a/src/fi/alk/harness/observability.py +++ b/src/fi/alk/harness/observability.py @@ -9,6 +9,7 @@ from __future__ import annotations import contextlib +import json import logging import os from typing import Any, Iterator, Mapping @@ -33,7 +34,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: @@ -60,7 +66,7 @@ def begin(job_id: str, run_id: str, context: Mapping[str, Any] | None = None) -> set_global_tracer_provider=False, verbose=False, ) - cleaner = _CleaningExporter(_exporter()) + 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 @@ -163,8 +169,19 @@ class _CleaningExporter(SpanExporter): carry the error's text. """ - def __init__(self, inner: Any) -> None: + 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 = tuple( + sorted( + {value for value in secret_values if len(value) >= _SHORTEST_SECRET}, + key=len, + reverse=True, + ) + ) + # 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 @@ -173,16 +190,17 @@ def export(self, spans: Any) -> Any: try: spans = list(spans) flat: list[str] = [] - plans = [_gather(span, flat) for span in spans] + plans = [_gather(span, flat, self._secrets) for span in spans] from fi.alk.providers import agent - # One call for the whole batch, with no key: the cleaner logs at most once per call, - # and a key such as gen_ai.usage.input_tokens would mask a count. + # 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): try: - copies.append(_rebuild(span, plan, cleaned)) + copies.append(_rebuild(span, plan, cleaned, self._resource(span.resource))) except Exception as dropped: error = error or dropped if not copies: @@ -197,6 +215,26 @@ def export(self, spans: Any) -> Any: 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] = [] + plan = { + key: _take(value, flat, key, self._secrets) + 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() @@ -204,42 +242,117 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: return self._inner.force_flush(timeout_millis) -def _take(value: Any, flat: list[str]) -> Any: - """Put each string of an attribute value in flat; return where to find it again.""" +# outbound.redact_outbound_text's floor: a shorter declared value would mask ordinary words. +_SHORTEST_SECRET = 8 + + +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[str, ...]) -> 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): - flat.append(_cut(value)) - return ("text", len(flat) - 1) + return _take_text(value, flat, key, secrets) if isinstance(value, (tuple, list)): - return ("sequence", [_take(item, flat) for item in value]) + 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[str, ...]) -> Any: + for secret in secrets: + value = value.replace(secret, "***") + if _named_secret(key): + return ("as is", "[redacted]") + 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)) + flat.append(cut) + 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: + return None + return tree if isinstance(tree, (dict, list)) else None + + +def _take_json(node: Any, flat: list[str], key: str) -> Any: + if isinstance(node, dict): + return ("dict", [(name, _take_json(item, flat, str(name))) for name, item in node.items()]) + if isinstance(node, list): + return ("list", [_take_json(item, flat, key) for item in node]) + if isinstance(node, str): + if _named_secret(key): + return ("as is", "[redacted]") + flat.append(node) + 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), ensure_ascii=False) + return body + + +def _give_json(where: Any, cleaned: list[Any]) -> Any: + kind, body = where + if kind == "dict": + return {name: _give_json(item, cleaned) for name, item in body} + if kind == "list": + return [_give_json(item, cleaned) for item in body] + if kind == "text": + return cleaned[body] return body -def _gather(span: Any, flat: list[str]) -> Any: +def _gather(span: Any, flat: list[str], secrets: tuple[str, ...]) -> Any: def mapping(attributes: Any) -> Any: - return {key: _take(value, flat) for key, value in (attributes or {}).items()} + return { + key: _take(value, flat, key, secrets) for key, value in (attributes or {}).items() + } description = span.status.description return ( - _take(span.name, flat), + _take(span.name, flat, "", secrets), mapping(span.attributes), - [(_take(event.name, flat), mapping(event.attributes)) for event in span.events], + [ + (_take(event.name, flat, "", secrets), mapping(event.attributes)) + for event in span.events + ], [mapping(link.attributes) for link in span.links], - _take(description, flat) + _take(description, flat, "", secrets) if isinstance(description, str) else ("as is", description), ) -def _rebuild(span: Any, plan: Any, cleaned: list[Any]) -> Any: +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 @@ -278,7 +391,7 @@ def bounded(where: Any, dropped: int) -> Any: name=_give(name, cleaned), context=span.context, parent=span.parent, - resource=span.resource, + resource=resource, attributes=bounded(attributes, span.dropped_attributes), events=new_events, links=new_links, diff --git a/tests/harness/test_observability.py b/tests/harness/test_observability.py index 19a46dea8..9917db3d2 100644 --- a/tests/harness/test_observability.py +++ b/tests/harness/test_observability.py @@ -7,6 +7,7 @@ from __future__ import annotations import asyncio +import json import logging import pytest @@ -104,8 +105,8 @@ def __init__(self, memory, sent, posts, restore_export): self.provider = None self._restore_export = restore_export - def begin(self): - observability.begin("job", "run") + def begin(self, secret_values=()): + observability.begin("job", "run", secret_values=secret_values) self.provider = observability._provider return self @@ -428,7 +429,8 @@ def test_t8_dropped_counts_are_carried_over(): 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 - assert copy.resource is original.resource + # 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 " @@ -780,3 +782,115 @@ def refuses(*args, **kwargs): assert result is SpanExportResult.FAILURE assert bodies == [] + + +# -- review (Dileep, #280): 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_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 From 3b833a2e402f2cccd934d01e8a19d9bddc41b368 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Wed, 11 Nov 2026 12:01:06 +0530 Subject: [PATCH 3/6] fix(harness): clean a JSON string inside a JSON span value as a structure too A message can carry a tool's input dumped as a JSON string inside JSON; cleaned as text, a URL with a query in it would take the rest of the string. Raised by Ash from #278's review. TH-9504 Signed-off-by: khushalsonawat --- src/fi/alk/harness/observability.py | 6 ++++++ tests/harness/test_observability.py | 10 ++++++++++ 2 files changed, 16 insertions(+) diff --git a/src/fi/alk/harness/observability.py b/src/fi/alk/harness/observability.py index e1c59b7a2..bbcaed0eb 100644 --- a/src/fi/alk/harness/observability.py +++ b/src/fi/alk/harness/observability.py @@ -304,6 +304,10 @@ def _take_json(node: Any, flat: list[str], key: str) -> Any: 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)) flat.append(node) return ("text", len(flat) - 1) return ("as is", node) @@ -328,6 +332,8 @@ def _give_json(where: Any, cleaned: list[Any]) -> Any: 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), ensure_ascii=False) return body diff --git a/tests/harness/test_observability.py b/tests/harness/test_observability.py index 9917db3d2..8dc3946a1 100644 --- a/tests/harness/test_observability.py +++ b/tests/harness/test_observability.py @@ -820,6 +820,16 @@ def test_a_json_value_is_cleaned_as_a_structure_and_stays_json(value): 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"] From 37af856a8b538b955b0fa9b7aa5041f2d9a6107e Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Thu, 12 Nov 2026 11:37:46 +0530 Subject: [PATCH 4/6] fix(harness): spans replace only declared credentials, and clean after the cut Platform settings in the simulator channel and the authoring secrets file (model names, the Observe project) are no longer replaced in spans; only values whose alias names a credential are, plus the guest-booking PIN and target number. The resource is never given declared values. Cut and JSON reading run on the value as recorded and declared values are replaced after; JSON keys are cleaned; JSON deeper than 64 levels is cleaned as text, and one span that fails to gather is dropped alone. Refs: TH-9504 Signed-off-by: khushalsonawat --- .../harness/hosted_authoring_entrypoint.py | 29 ++- src/fi/alk/harness/hosted_entrypoint.py | 13 +- src/fi/alk/harness/observability.py | 102 ++++++-- .../test_hosted_authoring_entrypoint.py | 91 ++++++++ tests/harness/test_hosted_entrypoint.py | 69 ++++++ tests/harness/test_observability.py | 220 +++++++++++++++++- 6 files changed, 498 insertions(+), 26 deletions(-) diff --git a/src/fi/alk/harness/hosted_authoring_entrypoint.py b/src/fi/alk/harness/hosted_authoring_entrypoint.py index 16c4afeae..c21f351e1 100644 --- a/src/fi/alk/harness/hosted_authoring_entrypoint.py +++ b/src/fi/alk/harness/hosted_authoring_entrypoint.py @@ -71,6 +71,30 @@ def _platform_simulator_values(values: dict[str, str]) -> dict[str, str]: } +# Not named as credentials, but secret: the guest-booking PIN, and its private target number. +_UNNAMED_SECRETS = frozenset({PIN_ENV, TARGET_PHONE_ENV}) + + +def _span_secret_values( + all_values: dict[str, str], gateway_values: dict[str, str] +) -> tuple[str, ...]: + """The values spans replace whole. The customer's go whole. The platform's (the SIMULATOR_* + aliases, the passthrough and Observe settings, the gateway file) go only where they are + credentials, since they also hold model names and the Observe project, which a span keeps.""" + platform_names = _PASSTHROUGH | _OBSERVABILITY_PASSTHROUGH + + def platform(name: str) -> bool: + return name.startswith("SIMULATOR_") or name in platform_names + + customer = tuple(value for name, value in all_values.items() if value and not platform(name)) + platform_values = {name: value for name, value in all_values.items() if platform(name)} + platform_values.update(gateway_values) + return ( + *customer, + *observability.credential_values(platform_values, also=_UNNAMED_SECRETS), + ) + + def _configure_generation_environment(values: dict[str, str]) -> None: for name in _PASSTHROUGH: if values.get(name): @@ -194,13 +218,12 @@ 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: - # Every value the job declared, as the outbound channels replace them: spans carry tool - # output, and a tool can print a key. + # Spans carry tool output, and a tool can print a key. observability.begin( job_id, run_id, telemetry, - secret_values=(*all_values.values(), *gateway_values.values()), + secret_values=_span_secret_values(all_values, gateway_values), ) observability.stage("authoring") try: diff --git a/src/fi/alk/harness/hosted_entrypoint.py b/src/fi/alk/harness/hosted_entrypoint.py index 0bbbc572c..ce532b38c 100644 --- a/src/fi/alk/harness/hosted_entrypoint.py +++ b/src/fi/alk/harness/hosted_entrypoint.py @@ -2273,10 +2273,15 @@ 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 = ( - *deps.peek_secret_values(), + *customer_secret_values, *tuple(simulator_secret_values.values()), ) + span_secret_values = ( + *customer_secret_values, + *observability.credential_values(simulator_secret_values), + ) adapter = OutboundAdapter( capabilities, events_spool=events_spool, @@ -2493,12 +2498,14 @@ async def _fail( if broker_socket is not None: os.environ["ALK_COMPOSE_BROKER_SOCKET"] = str(broker_socket) - # The same declared values the outbound adapter replaces: spans are a channel out too. + # Spans are a channel out too. The customer's declared values go whole; the simulator + # channel only where its alias names a credential, as it also holds model names and the + # Observe project, which a span must keep. observability.begin( job.job_id, job.run_id, (job.metadata or {}).get("telemetry"), - secret_values=declared_secret_values, + 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 bbcaed0eb..6a0cc39c9 100644 --- a/src/fi/alk/harness/observability.py +++ b/src/fi/alk/harness/observability.py @@ -164,9 +164,9 @@ def _cut(value: str) -> str: class _CleaningExporter(SpanExporter): """Hand the real exporter copies of each span with every string in it cleaned. - A span that can't be 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. + 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: @@ -190,7 +190,14 @@ def export(self, spans: Any) -> Any: try: spans = list(spans) flat: list[str] = [] - plans = [_gather(span, flat, self._secrets) for span in spans] + 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 @@ -199,6 +206,8 @@ def export(self, spans: Any) -> Any: 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: @@ -224,8 +233,10 @@ def _resource(self, resource: Any) -> Any: 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, self._secrets) + key: _take(value, flat, key, ()) for key, value in (resource.attributes or {}).items() } cleaned = agent._redact(flat) @@ -246,6 +257,28 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: _SHORTEST_SECRET = 8 +def credential_values( + values: Mapping[str, str], also: frozenset[str] = frozenset() +) -> tuple[str, ...]: + """The values in an alias -> value channel whose alias names a credential. + + The simulator and gateway channels hold platform configuration too (model names, the Observe + project): replacing those in spans would mask every LLM span's model and send the job's traces + to a project named "***". ``also`` names secrets whose alias doesn't say so, such as a PIN. + """ + return tuple( + str(value) + for alias, value in values.items() + if value and (_named_secret(str(alias)) or alias in also) + ) + + +def _mask(text: str, secrets: tuple[str, ...]) -> str: + for secret in secrets: + text = text.replace(secret, "***") + 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 @@ -268,10 +301,10 @@ def _take(value: Any, flat: list[str], key: str, secrets: tuple[str, ...]) -> An def _take_text(value: str, flat: list[str], key: str, secrets: tuple[str, ...]) -> Any: - for secret in secrets: - value = value.replace(secret, "***") 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 @@ -280,8 +313,8 @@ def _take_text(value: str, flat: list[str], key: str, secrets: tuple[str, ...]) # 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)) - flat.append(cut) + return ("json", _take_json(tree, flat, key, secrets)) + flat.append(_mask(cut, secrets)) return ("text", len(flat) - 1) @@ -291,24 +324,50 @@ def _as_json(value: str) -> Any: return None try: tree = json.loads(value) - except ValueError: + except (ValueError, RecursionError): + return None + if not isinstance(tree, (dict, list)) or _too_deep(tree): return None - return tree if isinstance(tree, (dict, list)) else 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 _take_json(node: Any, flat: list[str], key: str) -> Any: +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[str, ...]) -> Any: if isinstance(node, dict): - return ("dict", [(name, _take_json(item, flat, str(name))) for name, item in node.items()]) + 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) for item in node]) + 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)) - flat.append(node) + return ("json text", _take_json(tree, flat, key, secrets)) + flat.append(_mask(node, secrets)) return ("text", len(flat) - 1) return ("as is", node) @@ -327,7 +386,16 @@ def _give(where: Any, cleaned: list[Any]) -> Any: def _give_json(where: Any, cleaned: list[Any]) -> Any: kind, body = where if kind == "dict": - return {name: _give_json(item, cleaned) for name, item in body} + 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": diff --git a/tests/harness/test_hosted_authoring_entrypoint.py b/tests/harness/test_hosted_authoring_entrypoint.py index f89a12245..cb516f5b6 100644 --- a/tests/harness/test_hosted_authoring_entrypoint.py +++ b/tests/harness/test_hosted_authoring_entrypoint.py @@ -338,3 +338,94 @@ def test_good_subfolder_continues(tmp_path, monkeypatch): assert calls == ["compose", "auto"] assert not (tmp_path / "authoring" / "failure.json").exists() + + +def test_authoring_hands_spans_the_customers_values_whole_and_only_the_gateways_credentials( + 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.write_text( + json.dumps( + { + "RETELL_API_KEY": "retell-customer-key-1234567890", + "CUSTOMER_DSN": "customer-dsn-value-1234567890", + # This file also carries platform settings beside the customer's values. + "SIMULATOR_LLM_MODEL": "gemini-2.5-flash-lite", + "FI_HARNESS_PROJECT": "hosted-harness-prod", + "FI_API_KEY": "fi-platform-key-1234567890", + } + ), + encoding="utf-8", + ) + # The gateway's default values: credentials among models and the Observe project. + gateway.write_text( + json.dumps( + { + "AGENTCC_API_KEY": "agentcc-key-1234567890", + "AGENTCC_BASE_URL": "https://gateway.example.test", + "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", + "ALK_CAB_GUEST_POC_PIN": "48213977", + "ALK_CAB_GUEST_POC_TARGET_PHONE_NUMBER": "+15550104477", + } + ), + 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 fi-platform-key-1234567890 " + "e 48213977 f +15550104477", + }, + ): + 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_hosted_entrypoint.py b/tests/harness/test_hosted_entrypoint.py index 7af13a83e..1dd85b6b1 100644 --- a/tests/harness/test_hosted_entrypoint.py +++ b/tests/harness/test_hosted_entrypoint.py @@ -5515,3 +5515,72 @@ 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_hands_spans_the_customers_values_whole_and_only_the_gateways_credentials( + 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"}), encoding="utf-8" + ) + harness.deps.simulator_secrets_path = harness.tmp / "simulator-secrets.json" + # The gateway's default values: credentials among models and the Observe project. + harness.deps.simulator_secrets_path.write_text( + json.dumps( + { + "AGENTCC_API_KEY": "agentcc-key-1234567890", + "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", + }, + ): + 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 ***" diff --git a/tests/harness/test_observability.py b/tests/harness/test_observability.py index 8dc3946a1..80d06bf77 100644 --- a/tests/harness/test_observability.py +++ b/tests/harness/test_observability.py @@ -613,8 +613,7 @@ def test_t15_traceais_recorded_tool_input_is_cleaned(_real_begin): instrumentor.uninstrument() -# -- change review round 1: the copy's structure, the check's identity half, the inner exporter, -# -- and the real exporter -------------------------------------------------------------------- +# -- 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(): @@ -784,7 +783,7 @@ def refuses(*args, **kwargs): assert bodies == [] -# -- review (Dileep, #280): JSON values, declared secrets, the key rule, the resource ---------- +# -- JSON values, declared secrets, the key rule, the resource --------------------------------- def _export_one(span, secret_values=()): @@ -904,3 +903,218 @@ def test_the_resource_is_cleaned_once_and_reused(): 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_credential_values_keeps_the_aliases_that_name_a_credential(): + assert set(observability.credential_values(GATEWAY_CHANNEL)) == GATEWAY_CREDENTIALS + assert observability.credential_values( + {"A_TOKEN": "", "B_SECRET": "kept-value"} + ) == ("kept-value",) + + +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.credential_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": "pin 1234567 ok"}), ("1234567",)) + + assert copy.attributes["output.value"] == "pin 1234567 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"] From 7fc070d504a7532a801af2a265912cfbd302327c Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Thu, 12 Nov 2026 12:02:37 +0530 Subject: [PATCH 5/6] fix(harness): authoring spans take every customer value whole, and mask a PIN The authoring secrets file is split by where a value comes from, not by its name: every customer value goes whole, and the SIMULATOR_* values and the gateway file each give only their credentials, never merged. A short all-digit secret such as a 4-digit PIN is masked as a whole number. A secret is also masked in its JSON-escaped forms, and a traceai cut is made on the masked text so no part of an adjacent secret stays. Cleaned JSON is dumped as ASCII. Refs: TH-9504 Signed-off-by: khushalsonawat --- .../harness/hosted_authoring_entrypoint.py | 27 ++++--- src/fi/alk/harness/hosted_entrypoint.py | 8 +- src/fi/alk/harness/observability.py | 61 +++++++++++---- .../test_hosted_authoring_entrypoint.py | 26 ++++--- tests/harness/test_observability.py | 75 ++++++++++++++++++- 5 files changed, 156 insertions(+), 41 deletions(-) diff --git a/src/fi/alk/harness/hosted_authoring_entrypoint.py b/src/fi/alk/harness/hosted_authoring_entrypoint.py index c21f351e1..f1cb1773a 100644 --- a/src/fi/alk/harness/hosted_authoring_entrypoint.py +++ b/src/fi/alk/harness/hosted_authoring_entrypoint.py @@ -78,20 +78,25 @@ def _platform_simulator_values(values: dict[str, str]) -> dict[str, str]: def _span_secret_values( all_values: dict[str, str], gateway_values: dict[str, str] ) -> tuple[str, ...]: - """The values spans replace whole. The customer's go whole. The platform's (the SIMULATOR_* - aliases, the passthrough and Observe settings, the gateway file) go only where they are - credentials, since they also hold model names and the Observe project, which a span keeps.""" - platform_names = _PASSTHROUGH | _OBSERVABILITY_PASSTHROUGH + """The values spans replace whole, chosen by where each comes from, never by its name: a + customer may declare OPENAI_API_KEY or GOOGLE_CLOUD_PROJECT as well as the platform. - def platform(name: str) -> bool: - return name.startswith("SIMULATOR_") or name in platform_names - - customer = tuple(value for name, value in all_values.items() if value and not platform(name)) - platform_values = {name: value for name, value in all_values.items() if platform(name)} - platform_values.update(gateway_values) + - secrets.json holds the customer's declared values and the Observe export secrets; they go + whole, apart from the platform's SIMULATOR_* aliases, which customers can't use; + - those, and the gateway file, also hold model names and the Observe project, which a span + keeps, so only their credentials go. The two are filtered apart, never merged, so a customer + value can't be lost under a name the gateway also sends. + """ + customer = tuple( + value + for name, value in all_values.items() + if value and not name.startswith("SIMULATOR_") + ) + simulator = {name: value for name, value in all_values.items() if name.startswith("SIMULATOR_")} return ( *customer, - *observability.credential_values(platform_values, also=_UNNAMED_SECRETS), + *observability.credential_values(simulator, also=_UNNAMED_SECRETS), + *observability.credential_values(gateway_values, also=_UNNAMED_SECRETS), ) diff --git a/src/fi/alk/harness/hosted_entrypoint.py b/src/fi/alk/harness/hosted_entrypoint.py index ce532b38c..7a51bf340 100644 --- a/src/fi/alk/harness/hosted_entrypoint.py +++ b/src/fi/alk/harness/hosted_entrypoint.py @@ -2278,6 +2278,10 @@ async def run_job( *customer_secret_values, *tuple(simulator_secret_values.values()), ) + # Spans: secrets.json goes whole, since the platform writes only the customer's declared + # values and the Observe export secrets there. The simulator channel goes only where an alias + # names a credential. Its other values (models, the Observe project, the SIP trunk and caller + # number, base URLs) are platform settings, not secrets, and a span keeps them. span_secret_values = ( *customer_secret_values, *observability.credential_values(simulator_secret_values), @@ -2498,9 +2502,7 @@ async def _fail( if broker_socket is not None: os.environ["ALK_COMPOSE_BROKER_SOCKET"] = str(broker_socket) - # Spans are a channel out too. The customer's declared values go whole; the simulator - # channel only where its alias names a credential, as it also holds model names and the - # Observe project, which a span must keep. + # Spans are a channel out too (span_secret_values says which values they replace). observability.begin( job.job_id, job.run_id, diff --git a/src/fi/alk/harness/observability.py b/src/fi/alk/harness/observability.py index 6a0cc39c9..c2e440920 100644 --- a/src/fi/alk/harness/observability.py +++ b/src/fi/alk/harness/observability.py @@ -12,6 +12,7 @@ import json import logging import os +import re from typing import Any, Iterator, Mapping from opentelemetry.sdk.trace.export import SpanExporter @@ -152,6 +153,10 @@ def _cut(value: str) -> str: """ 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 @@ -173,13 +178,7 @@ 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 = tuple( - sorted( - {value for value in secret_values if len(value) >= _SHORTEST_SECRET}, - key=len, - reverse=True, - ) - ) + 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]] = {} @@ -255,6 +254,7 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: # outbound.redact_outbound_text's floor: a shorter declared value would mask ordinary words. _SHORTEST_SECRET = 8 +_SHORTEST_PIN = 4 def credential_values( @@ -273,9 +273,36 @@ def credential_values( ) -def _mask(text: str, secrets: tuple[str, ...]) -> str: +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: - text = text.replace(secret, "***") + if isinstance(secret, str): + text = text.replace(secret, "***") + else: + text = secret.sub("***", text) return text @@ -287,7 +314,7 @@ def _named_secret(key: str) -> bool: return lowered == "headers" or any(marker in lowered for marker in _REDACTION_MARKERS) -def _take(value: Any, flat: list[str], key: str, secrets: tuple[str, ...]) -> Any: +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 @@ -300,7 +327,7 @@ def _take(value: Any, flat: list[str], key: str, secrets: tuple[str, ...]) -> An return ("as is", value) -def _take_text(value: str, flat: list[str], key: str, secrets: tuple[str, ...]) -> Any: +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 @@ -314,7 +341,9 @@ def _take_text(value: str, flat: list[str], key: str, secrets: tuple[str, ...]) tree = _as_json(value) if tree is not None: return ("json", _take_json(tree, flat, key, secrets)) - flat.append(_mask(cut, 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) @@ -350,7 +379,7 @@ def _too_deep(tree: Any) -> bool: return False -def _take_json(node: Any, flat: list[str], key: str, secrets: tuple[str, ...]) -> Any: +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(): @@ -379,7 +408,7 @@ def _give(where: Any, cleaned: list[Any]) -> Any: if kind == "sequence": return tuple(_give(item, cleaned) for item in body) if kind == "json": - return json.dumps(_give_json(body, cleaned), ensure_ascii=False) + return json.dumps(_give_json(body, cleaned)) return body @@ -401,11 +430,11 @@ def _give_json(where: Any, cleaned: list[Any]) -> Any: if kind == "text": return cleaned[body] if kind == "json text": - return json.dumps(_give_json(body, cleaned), ensure_ascii=False) + return json.dumps(_give_json(body, cleaned)) return body -def _gather(span: Any, flat: list[str], secrets: tuple[str, ...]) -> Any: +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() diff --git a/tests/harness/test_hosted_authoring_entrypoint.py b/tests/harness/test_hosted_authoring_entrypoint.py index cb516f5b6..1e629d780 100644 --- a/tests/harness/test_hosted_authoring_entrypoint.py +++ b/tests/harness/test_hosted_authoring_entrypoint.py @@ -357,30 +357,34 @@ def test_authoring_hands_spans_the_customers_values_whole_and_only_the_gateways_ 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", - # This file also carries platform settings beside the customer's values. + "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", - "FI_HARNESS_PROJECT": "hosted-harness-prod", - "FI_API_KEY": "fi-platform-key-1234567890", } ), encoding="utf-8", ) - # The gateway's default values: credentials among models and the Observe project. + # 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", - "SIMULATOR_LLM_MODEL": "gemini-2.5-flash-lite", "FI_HARNESS_PROJECT": "hosted-harness-prod", "GOOGLE_CLOUD_PROJECT": "platform-project-1", - "ALK_CAB_GUEST_POC_PIN": "48213977", + "ALK_CAB_GUEST_POC_PIN": "4821", "ALK_CAB_GUEST_POC_TARGET_PHONE_NUMBER": "+15550104477", } ), @@ -416,8 +420,10 @@ def begin(job_id, run_id, context=None, secret_values=()): "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 fi-platform-key-1234567890 " - "e 48213977 f +15550104477", + "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", }, ): pass @@ -428,4 +434,6 @@ def begin(job_id, run_id, context=None, 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"] == "a *** b *** c *** d *** e *** f ***" + assert copy.attributes["input.value"] == ( + "a *** b *** c *** d *** e *** f *** g *** h *** i pin ***. j *** k 14821" + ) diff --git a/tests/harness/test_observability.py b/tests/harness/test_observability.py index 80d06bf77..6631004ee 100644 --- a/tests/harness/test_observability.py +++ b/tests/harness/test_observability.py @@ -978,9 +978,24 @@ def test_declared_values_never_touch_the_resource(): def test_a_declared_value_under_eight_characters_is_not_replaced(): - copy = _export_one(_readable(**{"output.value": "pin 1234567 ok"}), ("1234567",)) + copy = _export_one(_readable(**{"output.value": "word abc1234 ok"}), ("abc1234",)) - assert copy.attributes["output.value"] == "pin 1234567 ok" + 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(): @@ -1118,3 +1133,59 @@ def picky(span, flat, secrets): 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"] From b6ac5f1b1be3561029756a135bb5cff194b301d9 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Thu, 12 Nov 2026 12:08:30 +0530 Subject: [PATCH 6/6] fix(harness): spans mask every declared value but a listed set of platform settings Picking secrets by name failed both ways, so the rule now fails closed: every value in secrets.json, the simulator channel and the gateway file is masked, except names on NON_SECRET_SETTINGS (model names, providers, the Observe project, locations, base URLs). In secrets.json only the platform's own SIMULATOR_* and FI_* names may be kept. Each source is read on its own, so a name in two files masks both values. Refs: TH-9504 Signed-off-by: khushalsonawat --- .../harness/hosted_authoring_entrypoint.py | 32 +++++------- src/fi/alk/harness/hosted_entrypoint.py | 9 ++-- src/fi/alk/harness/observability.py | 52 +++++++++++++++---- .../test_hosted_authoring_entrypoint.py | 7 +-- tests/harness/test_hosted_entrypoint.py | 22 ++++++-- tests/harness/test_observability.py | 29 ++++++++--- 6 files changed, 103 insertions(+), 48 deletions(-) diff --git a/src/fi/alk/harness/hosted_authoring_entrypoint.py b/src/fi/alk/harness/hosted_authoring_entrypoint.py index f1cb1773a..20762d291 100644 --- a/src/fi/alk/harness/hosted_authoring_entrypoint.py +++ b/src/fi/alk/harness/hosted_authoring_entrypoint.py @@ -71,32 +71,24 @@ def _platform_simulator_values(values: dict[str, str]) -> dict[str, str]: } -# Not named as credentials, but secret: the guest-booking PIN, and its private target number. -_UNNAMED_SECRETS = frozenset({PIN_ENV, TARGET_PHONE_ENV}) +# 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, chosen by where each comes from, never by its name: a - customer may declare OPENAI_API_KEY or GOOGLE_CLOUD_PROJECT as well as the platform. - - - secrets.json holds the customer's declared values and the Observe export secrets; they go - whole, apart from the platform's SIMULATOR_* aliases, which customers can't use; - - those, and the gateway file, also hold model names and the Observe project, which a span - keeps, so only their credentials go. The two are filtered apart, never merged, so a customer - value can't be lost under a name the gateway also sends. - """ - customer = tuple( - value - for name, value in all_values.items() - if value and not name.startswith("SIMULATOR_") - ) - simulator = {name: value for name, value in all_values.items() if name.startswith("SIMULATOR_")} + """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 ( - *customer, - *observability.credential_values(simulator, also=_UNNAMED_SECRETS), - *observability.credential_values(gateway_values, also=_UNNAMED_SECRETS), + *observability.declared_values(all_values, _PLATFORM_ONLY_SETTINGS), + *observability.declared_values(gateway_values), ) diff --git a/src/fi/alk/harness/hosted_entrypoint.py b/src/fi/alk/harness/hosted_entrypoint.py index 7a51bf340..0e23afbe6 100644 --- a/src/fi/alk/harness/hosted_entrypoint.py +++ b/src/fi/alk/harness/hosted_entrypoint.py @@ -2278,13 +2278,12 @@ async def run_job( *customer_secret_values, *tuple(simulator_secret_values.values()), ) - # Spans: secrets.json goes whole, since the platform writes only the customer's declared - # values and the Observe export secrets there. The simulator channel goes only where an alias - # names a credential. Its other values (models, the Observe project, the SIP trunk and caller - # number, base URLs) are platform settings, not secrets, and a span keeps them. + # 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.credential_values(simulator_secret_values), + *observability.declared_values(simulator_secret_values), ) adapter = OutboundAdapter( capabilities, diff --git a/src/fi/alk/harness/observability.py b/src/fi/alk/harness/observability.py index c2e440920..34d225963 100644 --- a/src/fi/alk/harness/observability.py +++ b/src/fi/alk/harness/observability.py @@ -257,19 +257,53 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: _SHORTEST_PIN = 4 -def credential_values( - values: Mapping[str, str], also: frozenset[str] = frozenset() +# 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, ...]: - """The values in an alias -> value channel whose alias names a credential. + """Every value in one alias -> value source, except those under a name in ``settings``. - The simulator and gateway channels hold platform configuration too (model names, the Observe - project): replacing those in spans would mask every LLM span's model and send the job's traces - to a project named "***". ``also`` names secrets whose alias doesn't say so, such as a PIN. + 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 (_named_secret(str(alias)) or alias in also) + str(value) for alias, value in values.items() if value and alias not in settings ) diff --git a/tests/harness/test_hosted_authoring_entrypoint.py b/tests/harness/test_hosted_authoring_entrypoint.py index 1e629d780..821c721cd 100644 --- a/tests/harness/test_hosted_authoring_entrypoint.py +++ b/tests/harness/test_hosted_authoring_entrypoint.py @@ -340,7 +340,7 @@ def test_good_subfolder_continues(tmp_path, monkeypatch): assert not (tmp_path / "authoring" / "failure.json").exists() -def test_authoring_hands_spans_the_customers_values_whole_and_only_the_gateways_credentials( +def test_authoring_masks_every_value_in_spans_but_the_platform_settings( tmp_path, monkeypatch ) -> None: from unittest import mock @@ -386,6 +386,7 @@ def test_authoring_hands_spans_the_customers_values_whole_and_only_the_gateways_ "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", @@ -423,7 +424,7 @@ def begin(job_id, run_id, context=None, secret_values=()): "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", + "i pin 4821. j +15550104477 k 14821 l unlisted-gateway-value-42", }, ): pass @@ -435,5 +436,5 @@ def begin(job_id, run_id, context=None, secret_values=()): 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" + "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 1dd85b6b1..8923f5976 100644 --- a/tests/harness/test_hosted_entrypoint.py +++ b/tests/harness/test_hosted_entrypoint.py @@ -5517,7 +5517,7 @@ def test_the_activity_printed_to_stdout_loses_url_credentials(capsys, monkeypatc assert "The agent posts to https://hooks.example/v1 after." in printed -def test_the_job_hands_spans_the_customers_values_whole_and_only_the_gateways_credentials( +def test_the_job_masks_every_value_in_spans_but_the_platform_settings( monkeypatch, ) -> None: from opentelemetry.sdk.resources import Resource @@ -5535,14 +5535,24 @@ def test_the_job_hands_spans_the_customers_values_whole_and_only_the_gateways_cr ] ) harness.deps.secrets_path.write_text( - json.dumps({"CUSTOMER_DSN": "customer-dsn-value-1234567890"}), encoding="utf-8" + 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 default values: credentials among models and the Observe project. + # 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", @@ -5572,7 +5582,9 @@ def begin(job_id, run_id, context=None, secret_values=()): 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", + "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 @@ -5583,4 +5595,4 @@ def begin(job_id, run_id, context=None, 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"] == "a *** b ***" + 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 6631004ee..080e24b88 100644 --- a/tests/harness/test_observability.py +++ b/tests/harness/test_observability.py @@ -927,11 +927,28 @@ def test_the_resource_is_cleaned_once_and_reused(): } -def test_credential_values_keeps_the_aliases_that_name_a_credential(): - assert set(observability.credential_values(GATEWAY_CHANNEL)) == GATEWAY_CREDENTIALS - assert observability.credential_values( - {"A_TOKEN": "", "B_SECRET": "kept-value"} - ) == ("kept-value",) +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(): @@ -965,7 +982,7 @@ def _assert_platform_config_survives(secret_values): def test_the_gateways_default_values_leave_models_and_project_unchanged(): - _assert_platform_config_survives(observability.credential_values(GATEWAY_CHANNEL)) + _assert_platform_config_survives(observability.declared_values(GATEWAY_CHANNEL)) def test_declared_values_never_touch_the_resource():