From 33834f3c22cb6aabe04fa47b481f4441bb826664 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 02:44:20 +0530 Subject: [PATCH 1/2] item 8: the OpenTelemetry sink MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SPEC-v0.2 §8, acceptance test T29, and T30 now exists as a named test. One span per action, opened on ACTION_PROPOSED with one span event per Event. A retry is a new action and a new span; the two are correlated by ctrlrun.effect_key, which is the identity that actually spans attempts. Two things §8 leaves to the implementation, both resolved in the module's own words rather than silently: The span ends on the receipt, one call after the terminal event §8 names. ctrlrun.receipt_id, ctrlrun.result and the final attempt live on the receipt, and a receipt always follows a terminal event in the same Control call. Ending a step earlier would export a span that cannot say how the action came out, which is the only thing a reader wants from one. An action awaiting a human gets no receipt, so its span stays open — §8 already accepts that, and the open-span map is bounded so a long-running gateway does not leak one per abandoned approval. Evicted spans are dropped rather than ended: ending one would assert an outcome that never happened. Argument values are withheld by default, and ctrlrun.effect_key is exported always — but an effect key is built from arguments, so refund:{payment_id} puts a payment id in every backend receiving these spans. That is intended: §8 names the effect key as what correlates a retry, and a correlation key nobody can see correlates nothing. It is now asserted by a test rather than discovered, because it makes the choice of effect template a choice about what leaves the building. Span status is OK for committed, ERROR for failed and ambiguous, and unset for denied and blocked. A refusal is the system working as designed; marking it an error teaches an on-call rotation to ignore the signal that matters. The gateway grows --otel and --otel-arguments, and Control exposes its sinks so one more can be composed onto it. CI installs the extra. --- .github/workflows/ci.yml | 4 +- pyproject.toml | 13 +- src/ctrlrun/cli/main.py | 11 + src/ctrlrun/control.py | 5 + src/ctrlrun/gateway/__init__.py | 13 ++ src/ctrlrun/otel.py | 222 +++++++++++++++++++ tests/test_otel.py | 366 ++++++++++++++++++++++++++++++++ tests/test_packaging.py | 64 ++++++ 8 files changed, 694 insertions(+), 4 deletions(-) create mode 100644 src/ctrlrun/otel.py create mode 100644 tests/test_otel.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1379f1d4..c4f87f20 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,7 +24,7 @@ jobs: python -m pip install --upgrade pip # The gateway's tests need its extra; T30 still proves `import ctrlrun` does not # touch it, in a subprocess, whether or not it happens to be installed here. - pip install -e ".[dev,gateway]" + pip install -e ".[dev,gateway,otel]" - name: pytest run: pytest @@ -64,7 +64,7 @@ jobs: run: | mkdir -p /tmp/sdist && tar xzf dist/*.tar.gz -C /tmp/sdist --strip-components=1 cd /tmp/sdist - pip install ".[dev,gateway]" + pip install ".[dev,gateway,otel]" pytest -q - name: The quick start works from the wheel diff --git a/pyproject.toml b/pyproject.toml index 1c3dca81..cbd4cb6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,8 +48,14 @@ dev = [ # first row — a gateway that could not tell "never connected" from "no reply" would have to # call everything AMBIGUOUS, which is safe and useless. gateway = ["httpx>=0.27"] -# Item 8 adds the OpenTelemetry API, SDK and OTLP/HTTP exporter. -otel = [] +# §8's sink hands spans to whatever processor the application configured. The API alone +# would be enough for a library, but an operator installing this extra wants to be able to +# export without assembling a stack, so the SDK and the OTLP/HTTP exporter come with it. +otel = [ + "opentelemetry-api>=1.20", + "opentelemetry-sdk>=1.20", + "opentelemetry-exporter-otlp-proto-http>=1.20", +] [tool.setuptools.packages.find] where = ["src"] @@ -95,6 +101,9 @@ select = ["E", "F", "W", "I", "N", "UP", "B", "ANN", "SIM", "RUF"] # Same reasoning for the ACS adapter: its subject matter is an untyped JSON-RPC envelope off # somebody else's wire, and `_Unknown` follows this codebase's error naming. "src/ctrlrun/acs.py" = ["ANN401", "N818"] +# The OTel sink holds objects from a lazily-imported SDK: annotating them would mean +# importing `opentelemetry` at module scope, which is the one thing §1.1 forbids. +"src/ctrlrun/otel.py" = ["ANN401"] # Type hints are a src/ constraint; tests stay cheap to write. N802 lets acceptance # tests carry the SPEC §7 identifier verbatim (test_T7_...). "tests/*" = ["ANN", "N802"] diff --git a/src/ctrlrun/cli/main.py b/src/ctrlrun/cli/main.py index df2c7dc5..ca565bfe 100644 --- a/src/ctrlrun/cli/main.py +++ b/src/ctrlrun/cli/main.py @@ -473,6 +473,13 @@ def _approval_dict(record: ApprovalRecord) -> dict[str, Any]: is_flag=True, help="Permit an http:// webhook url, loopback only.", ) +@click.option("--otel", is_flag=True, help="Export one span per action (ctrlrun[otel]).") +@click.option( + "--otel-arguments", + is_flag=True, + help="Include argument values as span attributes. Off by default: arguments carry " + "customer identifiers and amounts, and a trace backend is not the receipt store.", +) def gateway( upstream: str, alias: str, @@ -491,6 +498,8 @@ def gateway( webhook_url: str | None, webhook_secret_file: str | None, allow_insecure_webhook: bool, + otel: bool, + otel_arguments: bool, ) -> None: """Front an MCP server, applying this directory's policy to every tools/call.""" host, _, port = listen.rpartition(":") @@ -517,6 +526,8 @@ def gateway( webhook_url=webhook_url, webhook_secret_file=webhook_secret_file, allow_insecure_webhook=allow_insecure_webhook, + otel=otel, + otel_arguments=otel_arguments, ) except CTRLRunError as exc: raise _fail(exc) from exc diff --git a/src/ctrlrun/control.py b/src/ctrlrun/control.py index e935c74c..47ed9c62 100644 --- a/src/ctrlrun/control.py +++ b/src/ctrlrun/control.py @@ -260,6 +260,11 @@ def store(self) -> StateStore: def approvals(self) -> ApprovalProvider: return self._approvals + @property + def sinks(self) -> tuple[EventSink, ...]: + """The sinks this Control fans out to, in registration order (SPEC-v0.2 §4.1).""" + return self._sinks + @property def lease(self) -> timedelta: """How long a reservation this Control takes is held for (SPEC-v0.1 §5.3 E3).""" diff --git a/src/ctrlrun/gateway/__init__.py b/src/ctrlrun/gateway/__init__.py index 36d1aba3..e17b9d0c 100644 --- a/src/ctrlrun/gateway/__init__.py +++ b/src/ctrlrun/gateway/__init__.py @@ -50,6 +50,8 @@ def serve(*, upstream: str, alias: str, **options: Any) -> None: from ..webhook import WebhookApprovalProvider, webhook_secret from .server import Gateway, GatewayConfig, httpx_forwarder, serve_forever + otel = bool(options.pop("otel", False)) + otel_arguments = bool(options.pop("otel_arguments", False)) public_url = options.pop("public_url", None) webhook_url = options.pop("webhook_url", None) secret_file = options.pop("webhook_secret_file", None) @@ -58,6 +60,17 @@ def serve(*, upstream: str, alias: str, **options: Any) -> None: config = GatewayConfig(upstream=upstream, alias=alias, webhook_secret=secret, **options) control = Control.from_file() + if otel: + # §8 — one more sink beside the JSONL one `from_file` installed. It never blocks and + # never raises into the kernel (§4.2), so adding it changes no decision. + from ..otel import OTelEventSink + + control = Control( + control.policy, + control.store, + control.approvals, + sinks=[*control.sinks, OTelEventSink(arguments=otel_arguments)], + ) if webhook_url: # §7 — the provider notifies; the gateway's endpoint receives. Both need the same # secret, and neither takes it from a command-line argument (§7.3). diff --git a/src/ctrlrun/otel.py b/src/ctrlrun/otel.py new file mode 100644 index 00000000..f45756cf --- /dev/null +++ b/src/ctrlrun/otel.py @@ -0,0 +1,222 @@ +"""The OpenTelemetry sink. Build-list item 8; SPEC-v0.2 §8. Ships in `ctrlrun[otel]`. + +An `EventSink` (§4) and nothing more: it never decides, never blocks, and never raises. What +it exports is a copy of what the store already holds, so a failing exporter is a lost export +and never a failed action — §4.2's rule, and the reason this is a sink rather than a hook. + +**One span per action**, opened on `ACTION_PROPOSED` and carrying one span event per `Event`. +A retry is a new action and therefore a new span; the two are correlated by +`ctrlrun.effect_key`, which is the identity that actually spans attempts (v0.1 §5). +""" + +from __future__ import annotations + +import importlib +import logging +import threading +from collections import OrderedDict +from typing import Any, Final + +from .errors import MissingDependency +from .receipt import Event, Receipt + +_LOG = logging.getLogger(__name__) + +#: Imported by name so a missing extra is a `MissingDependency` rather than a +#: `ModuleNotFoundError` from halfway down an import chain (SPEC-v0.2 §1.1). +_OTEL_MODULE: Final = "opentelemetry.trace" +_EXTRA: Final = "otel" + +#: SPEC-v0.2 §8 — the attribute prefix, and the one exception to it being free. +PREFIX: Final = "ctrlrun." +ARGUMENT_PREFIX: Final = "ctrlrun.argument." + +#: An action's span ends when its receipt arrives. A suspended one never gets a receipt, so +#: its span stays open; this bounds how many can accumulate in a long-running process. +DEFAULT_MAX_OPEN_SPANS: Final = 1024 + +#: `ReceiptResult` values that mark a span ERROR. A refusal is the system working as +#: designed, and marking it an error teaches an on-call rotation to ignore the signal that +#: matters — so `denied` and `blocked` leave the status unset. +_ERROR_RESULTS: Final = frozenset({"failed", "ambiguous"}) +_OK_RESULTS: Final = frozenset({"committed"}) + + +class OTelEventSink: + """Export every `Event` and `Receipt` as OpenTelemetry spans (SPEC-v0.2 §8). + + **Never blocks.** Spans are handed to whatever span processor the application configured + and nothing is flushed synchronously; with no configured tracer provider the API's no-op + provider makes the whole thing free. A process that dies mid-action leaves that span + unended — stated, not solved. + """ + + def __init__( + self, + *, + tracer_provider: Any = None, + arguments: bool = False, + max_open_spans: int = DEFAULT_MAX_OPEN_SPANS, + ) -> None: + trace = _trace() + self._trace = trace + self._tracer = ( + tracer_provider.get_tracer("ctrlrun") + if tracer_provider is not None + else trace.get_tracer("ctrlrun") + ) + # SPEC-v0.2 §8 — argument *values* are not attributes unless asked for. Arguments + # carry customer identifiers and amounts, and a trace backend is not the receipt + # store. The default is the fail-closed one. + self._arguments = arguments + self._max_open = max(1, max_open_spans) + self._open: OrderedDict[str, Any] = OrderedDict() + self._lock = threading.Lock() + + @property + def open_spans(self) -> int: + """How many actions have a span open. Bounded; see `max_open_spans`.""" + with self._lock: + return len(self._open) + + # --- EventSink ---------------------------------------------------------------------- + + def on_event(self, event: Event) -> None: + span = self._span_for(event) + if span is None: + return + span.add_event(str(event.type), attributes=self._event_attributes(event)) + + def on_receipt(self, receipt: Receipt) -> None: + """End the action's span, carrying everything only the receipt knows. + + SPEC: §8 says the span ends on the action's *terminal event*. It ends here instead, + one call later, because `ctrlrun.receipt_id`, `ctrlrun.result` and the final + `ctrlrun.attempt` live on the receipt and a receipt always follows a terminal event + in the same `Control` call. Ending a step earlier would mean exporting a span that + cannot say how the action came out, which is the only thing a reader wants from it. + """ + with self._lock: + span = self._open.pop(receipt.action_id, None) + if span is None: + return + # §8 — the span name is the action name, and only the receipt carries it: an `Event` + # has no action name (v0.1 §6.2), so the span opens under a placeholder and is + # renamed here. Renaming before `end()` is what `update_name` is for. + span.update_name(receipt.action) + for name, value in self._receipt_attributes(receipt).items(): + span.set_attribute(name, value) + span.set_status(self._status(str(receipt.result), receipt.error)) + span.end() + + # --- internals ---------------------------------------------------------------------- + + def _span_for(self, event: Event) -> Any: + from .receipt import EventType + + with self._lock: + span = self._open.get(event.action_id) + if span is not None: + return span + if event.type is not EventType.ACTION_PROPOSED: + # Every action opens with ACTION_PROPOSED (v0.1 §6.2). An event for an action + # this sink never saw start belongs to a span somebody else owns — a process + # that restarted, or a `ctrlrun resolve` run days later — and inventing one + # would export a span with no beginning. + return None + span = self._tracer.start_span( + PENDING_SPAN_NAME, + attributes={f"{PREFIX}action_id": event.action_id}, + ) + self._open[event.action_id] = span + self._evict_if_needed() + return span + + def _evict_if_needed(self) -> None: + """Bound the open-span map. Called under the lock. + + An action awaiting a human never reaches a receipt, so its span never ends. Without a + bound a long-running gateway accumulates one per abandoned approval. Evicted spans are + dropped rather than ended: ending one would export a span asserting an outcome that + never happened. + """ + while len(self._open) > self._max_open: + action_id, _ = self._open.popitem(last=False) + _LOG.warning( + "dropped the open OTel span for %s: more than %d actions are awaiting a " + "terminal state at once", + action_id, + self._max_open, + ) + + def _event_attributes(self, event: Event) -> dict[str, Any]: + attributes: dict[str, Any] = {} + if event.effect_key is not None: + attributes[f"{PREFIX}effect_key"] = event.effect_key + if event.approval_id is not None: + attributes[f"{PREFIX}approval_id"] = event.approval_id + for key, value in event.data.items(): + if _exportable(value): + attributes[f"{PREFIX}{key}"] = value + return attributes + + def _receipt_attributes(self, receipt: Receipt) -> dict[str, Any]: + attributes: dict[str, Any] = { + f"{PREFIX}action_id": receipt.action_id, + f"{PREFIX}action.name": receipt.action, + f"{PREFIX}action_hash": receipt.action_hash, + f"{PREFIX}principal.agent": receipt.principal.agent, + f"{PREFIX}environment": receipt.environment, + f"{PREFIX}decision": str(receipt.decision), + f"{PREFIX}decision_reason": receipt.decision_reason, + f"{PREFIX}attempt": receipt.attempt, + f"{PREFIX}result": str(receipt.result), + f"{PREFIX}receipt_id": receipt.receipt_id, + } + for name, value in ( + (f"{PREFIX}principal.user", receipt.principal.user), + (f"{PREFIX}resource", receipt.resource), + (f"{PREFIX}effect_key", receipt.effect_key), + (f"{PREFIX}approval_id", receipt.approval_id), + (f"{PREFIX}approver", receipt.approver), + ): + if value is not None: + attributes[name] = value + if self._arguments: + for key, value in receipt.arguments.items(): + if _exportable(value): + attributes[f"{ARGUMENT_PREFIX}{key}"] = value + return attributes + + def _status(self, result: str, error: str | None) -> Any: + status = self._trace.Status + code = self._trace.StatusCode + if result in _ERROR_RESULTS: + return status(code.ERROR, error) + if result in _OK_RESULTS: + return status(code.OK) + # `denied` and `blocked`: unset, deliberately. + return status(code.UNSET) + + +#: What a span is called between `ACTION_PROPOSED` and its receipt. An `Event` carries no +#: action name (v0.1 §6.2), so a span that never reaches a receipt — one awaiting a human — +#: keeps this name. It is deliberately not the tool's name: guessing one would be worse. +PENDING_SPAN_NAME: Final = "ctrlrun.action" + + +def _exportable(value: object) -> bool: + """Whether OpenTelemetry can carry this value as an attribute without a conversion. + + Anything else is dropped rather than stringified: an attribute whose type changed on the + way out is worse than one that is absent, because a query written against it silently + stops matching. + """ + return isinstance(value, str | bool | int | float) + + +def _trace() -> Any: + try: + return importlib.import_module(_OTEL_MODULE) + except ImportError as exc: + raise MissingDependency(_OTEL_MODULE, _EXTRA) from exc diff --git a/tests/test_otel.py b/tests/test_otel.py new file mode 100644 index 00000000..b49da89b --- /dev/null +++ b/tests/test_otel.py @@ -0,0 +1,366 @@ +"""The OpenTelemetry sink. Build-list item 8; SPEC-v0.2 §8, acceptance test T29. + +An `EventSink` (§4) and nothing more: it never decides, never blocks, and never raises. What +it exports is a copy of what the store already holds — so a failing exporter is a lost export, +never a failed action. +""" + +from __future__ import annotations + +import pytest + +from ctrlrun import ( + ApprovalRequired, + Control, + InMemoryStateStore, + MissingDependency, + NotExecuted, + Policy, + context, + protect, +) + +pytest.importorskip("opentelemetry.sdk", reason="the otel extra is not installed") + +from opentelemetry import trace +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 ctrlrun.otel import OTelEventSink + +POLICY = """ +schema: ctrlrun.policy/v1 +actions: + stripe.refund: + rules: + - when: { amount_lte: 500 } + decision: allow + - when: { amount_lte: 5000 } + decision: approve + - decision: deny +""" + + +@pytest.fixture +def exporter(): + """A real SDK tracer provider with an in-memory exporter, as §8's test asks for.""" + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + return exporter, provider + + +@pytest.fixture +def store(): + store = InMemoryStateStore() + yield store + store.close() + + +def _control(store, provider, **options): + return Control( + Policy.from_yaml(POLICY), + store, + sinks=[OTelEventSink(tracer_provider=provider, **options)], + ) + + +def _refund(control, executor=None): + @protect("stripe.refund", effect="refund:{payment_id}", control=control) + def refund(payment_id: str, amount: int) -> str: + return executor() if executor else "re_1" + + return refund + + +def _spans(exporter): + return exporter.get_finished_spans() + + +def _attrs(span): + return dict(span.attributes or {}) + + +# --- T29: one span per action ---------------------------------------------------------- + + +def test_T29_one_action_produces_one_span_named_for_the_action(store, exporter): + export, provider = exporter + with context(agent="refund-agent", user="alice"): + _refund(_control(store, provider))(payment_id="txn_1", amount=200) + + spans = _spans(export) + assert len(spans) == 1 + assert spans[0].name == "stripe.refund" + + +def test_T29_the_span_carries_every_attribute_the_spec_names(store, exporter): + export, provider = exporter + with context(agent="refund-agent", user="alice"): + _refund(_control(store, provider))(payment_id="txn_1", amount=200) + + attributes = _attrs(_spans(export)[0]) + for name in ( + "ctrlrun.action_id", + "ctrlrun.action.name", + "ctrlrun.action_hash", + "ctrlrun.principal.agent", + "ctrlrun.principal.user", + "ctrlrun.environment", + "ctrlrun.decision", + "ctrlrun.decision_reason", + "ctrlrun.effect_key", + "ctrlrun.attempt", + "ctrlrun.result", + "ctrlrun.receipt_id", + ): + assert name in attributes, name + assert attributes["ctrlrun.action.name"] == "stripe.refund" + assert attributes["ctrlrun.principal.agent"] == "refund-agent" + assert attributes["ctrlrun.principal.user"] == "alice" + assert attributes["ctrlrun.effect_key"] == "refund:txn_1" + assert attributes["ctrlrun.result"] == "committed" + assert attributes["ctrlrun.attempt"] == 1 + + +def test_T29_every_event_becomes_a_span_event_named_by_its_type(store, exporter): + export, provider = exporter + with context(agent="refund-agent"): + _refund(_control(store, provider))(payment_id="txn_1", amount=200) + + names = [event.name for event in _spans(export)[0].events] + assert names == [str(event.type) for event in store.events()] + + +def test_T29_a_span_events_data_is_flattened_to_ctrlrun_attributes(store, exporter): + export, provider = exporter + with context(agent="refund-agent"): + _refund(_control(store, provider))(payment_id="txn_1", amount=200) + + policy_evaluated = next( + event for event in _spans(export)[0].events if event.name == "POLICY_EVALUATED" + ) + assert dict(policy_evaluated.attributes)["ctrlrun.decision"] == "allow" + + +def test_T29_the_approval_attributes_appear_when_there_was_one(store, exporter): + export, provider = exporter + control = _control(store, provider) + refund = _refund(control) + with context(agent="refund-agent"), pytest.raises(ApprovalRequired) as pending: + refund(payment_id="txn_2", amount=2000) + store.grant_approval(pending.value.request_id, "cli:local") + from ctrlrun import with_approval + + with context(agent="refund-agent"), with_approval(pending.value.request_id): + refund(payment_id="txn_2", amount=2000) + + attributes = _attrs(_spans(export)[-1]) + assert attributes["ctrlrun.approval_id"] == pending.value.request_id + assert attributes["ctrlrun.approver"] == "cli:local" + + +# --- T29: argument values are not exported unless asked for ---------------------------- + + +def test_T29_argument_values_are_not_attributes_by_default(store, exporter): + """Arguments carry customer identifiers and amounts, and a trace backend is not the + receipt store. The default is the fail-closed one.""" + export, provider = exporter + with context(agent="refund-agent"): + _refund(_control(store, provider))(payment_id="txn_secret", amount=200) + + attributes = _attrs(_spans(export)[0]) + assert not [name for name in attributes if name.startswith("ctrlrun.argument.")] + assert attributes.get("ctrlrun.argument.payment_id") is None + + +def test_T29_argument_values_appear_only_when_the_option_is_set(store, exporter): + export, provider = exporter + with context(agent="refund-agent"): + _refund(_control(store, provider, arguments=True))(payment_id="txn_secret", amount=200) + + attributes = _attrs(_spans(export)[0]) + assert attributes["ctrlrun.argument.payment_id"] == "txn_secret" + assert attributes["ctrlrun.argument.amount"] == 200 + + +def test_the_effect_key_carries_argument_text_even_when_arguments_are_withheld(store, exporter): + """The one tension in §8, asserted rather than left to be discovered. + + Argument values are withheld by default, and `ctrlrun.effect_key` is exported always — + but an effect key is *built* from arguments, so `refund:{payment_id}` puts a payment id + in every backend receiving these spans. That is intended: §8 names the effect key as what + correlates a retry, and a correlation key nobody can see correlates nothing. It makes the + choice of effect template a choice about what leaves the building. + """ + export, provider = exporter + with context(agent="refund-agent"): + _refund(_control(store, provider))(payment_id="txn_1", amount=200) + + assert _attrs(_spans(export)[0])["ctrlrun.effect_key"] == "refund:txn_1" + + +# --- T29: span status ------------------------------------------------------------------ + + +def test_T29_status_is_ok_for_committed(store, exporter): + export, provider = exporter + with context(agent="refund-agent"): + _refund(_control(store, provider))(payment_id="txn_1", amount=200) + + assert _spans(export)[0].status.status_code is trace.StatusCode.OK + + +def test_T29_status_is_error_for_ambiguous(store, exporter): + export, provider = exporter + control = _control(store, provider) + + def lose_the_response(): + raise TimeoutError("response lost") + + with context(agent="refund-agent"), pytest.raises(TimeoutError): + _refund(control, lose_the_response)(payment_id="txn_1", amount=200) + + assert _spans(export)[0].status.status_code is trace.StatusCode.ERROR + + +def test_T29_status_is_error_for_failed(store, exporter): + export, provider = exporter + control = _control(store, provider) + + def nothing_happened(): + raise NotExecuted("rejected before any side effect") + + with context(agent="refund-agent"), pytest.raises(NotExecuted): + _refund(control, nothing_happened)(payment_id="txn_1", amount=200) + + assert _spans(export)[0].status.status_code is trace.StatusCode.ERROR + + +def test_T29_status_is_unset_for_denied(store, exporter): + """A refusal is the system working as designed. Marking it an error teaches an on-call + rotation to ignore the signal that matters.""" + export, provider = exporter + from ctrlrun import ActionDenied + + with context(agent="refund-agent"), pytest.raises(ActionDenied): + _refund(_control(store, provider))(payment_id="txn_1", amount=999999) + + assert _spans(export)[0].status.status_code is trace.StatusCode.UNSET + + +def test_T29_status_is_unset_for_blocked(store, exporter): + export, provider = exporter + from ctrlrun import AmbiguousEffect + + control = _control(store, provider) + + def lose_the_response(): + raise TimeoutError("response lost") + + with context(agent="refund-agent"), pytest.raises(TimeoutError): + _refund(control, lose_the_response)(payment_id="txn_1", amount=200) + with context(agent="refund-agent"), pytest.raises(AmbiguousEffect): + _refund(control)(payment_id="txn_1", amount=200) + + blocked = _spans(export)[-1] + assert _attrs(blocked)["ctrlrun.result"] == "blocked" + assert blocked.status.status_code is trace.StatusCode.UNSET + + +# --- T29: a failing exporter cannot affect an action ----------------------------------- + + +def test_T29_an_exporter_that_raises_does_not_affect_the_action(store, caplog): + """T17's rule applied to this sink (§4.2).""" + + class Exploding(InMemorySpanExporter): + def export(self, spans): + raise RuntimeError("the collector is down") + + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(Exploding())) + control = _control(store, provider) + + with caplog.at_level("WARNING", logger="ctrlrun"), context(agent="refund-agent"): + assert _refund(control)(payment_id="txn_1", amount=200) == "re_1" + + assert store.receipts()[-1].result.value == "committed" + + +def test_a_sink_with_no_configured_provider_is_free(store): + """With no tracer provider the API's no-op one makes this cost nothing (§8).""" + control = Control(Policy.from_yaml(POLICY), store, sinks=[OTelEventSink()]) + + with context(agent="refund-agent"): + assert _refund(control)(payment_id="txn_1", amount=200) == "re_1" + + +# --- §8: a retry is a new span, correlated by the effect key --------------------------- + + +def test_a_retry_is_a_new_span_correlated_by_the_effect_key(store, exporter): + export, provider = exporter + control = _control(store, provider) + attempts = {"n": 0} + + def fails_then_works(): + attempts["n"] += 1 + if attempts["n"] == 1: + raise NotExecuted("nothing ran") + return "re_1" + + refund = _refund(control, fails_then_works) + with context(agent="refund-agent"), pytest.raises(NotExecuted): + refund(payment_id="txn_1", amount=200) + with context(agent="refund-agent"): + refund(payment_id="txn_1", amount=200) + + spans = _spans(export) + assert len(spans) == 2 + assert {_attrs(span)["ctrlrun.effect_key"] for span in spans} == {"refund:txn_1"} + assert _attrs(spans[0])["ctrlrun.action_id"] != _attrs(spans[1])["ctrlrun.action_id"] + assert [_attrs(span)["ctrlrun.attempt"] for span in spans] == [1, 2] + + +def test_an_action_awaiting_a_human_leaves_its_span_open(store, exporter): + """Stated, not solved (§8): the action has not reached a terminal state, so there is no + receipt and nothing to end the span with.""" + export, provider = exporter + with context(agent="refund-agent"), pytest.raises(ApprovalRequired): + _refund(_control(store, provider))(payment_id="txn_2", amount=2000) + + assert _spans(export) == () + + +def test_the_open_span_map_is_bounded(store, exporter): + """A long-running gateway must not leak a span per abandoned approval.""" + _, provider = exporter + sink = OTelEventSink(tracer_provider=provider, max_open_spans=4) + control = Control(Policy.from_yaml(POLICY), store, sinks=[sink]) + refund = _refund(control) + + for index in range(10): + with context(agent="refund-agent"), pytest.raises(ApprovalRequired): + refund(payment_id=f"txn_{index}", amount=2000) + + assert sink.open_spans <= 4 + + +# --- T30: the extra says so when it is absent ------------------------------------------ + + +def test_T30_a_missing_otel_extra_raises_MissingDependency(monkeypatch): + import ctrlrun.otel as otel + + monkeypatch.setattr(otel, "_OTEL_MODULE", "no_such_module_at_all") + + with pytest.raises(MissingDependency) as raised: + OTelEventSink() + + assert "pip install 'ctrlrun[otel]'" in str(raised.value) + assert not isinstance(raised.value, ImportError) diff --git a/tests/test_packaging.py b/tests/test_packaging.py index d621c4ed..e8473f77 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -33,6 +33,70 @@ def _pyproject() -> dict: return tomllib.loads(path.read_text(encoding="utf-8")) +def test_T30_a_subprocess_importing_ctrlrun_pulls_in_no_module_from_an_extra(): + """SPEC-v0.2 §10's T30, in one place. A **subprocess**, because in-process this would + pass or fail on whatever pytest happened to import first.""" + finished = subprocess.run( + [ + sys.executable, + "-c", + "import ctrlrun, sys;" + "print(sorted(n for n in sys.modules" + " if n.split('.')[0] in ('httpx', 'opentelemetry')" + " or n in ('ctrlrun.gateway', 'ctrlrun.otel', 'ctrlrun.acs')))", + ], + capture_output=True, + text=True, + check=True, + ) + + assert finished.stdout.strip() == "[]" + + +@pytest.mark.parametrize( + ("factory", "extra"), + [ + ("from ctrlrun.gateway import serve; serve(upstream='http://x', alias='a')", "gateway"), + ("from ctrlrun.otel import OTelEventSink; OTelEventSink()", "otel"), + ], +) +def test_T30_a_missing_extra_says_which_one_to_install(factory, extra, tmp_path): + """Never `ModuleNotFoundError`: an operator reads that as a broken package rather than + as an option they did not select. Run in a subprocess with the extra's module hidden.""" + script = ( + "import sys, importlib.abc\n" + "class Block(importlib.abc.MetaPathFinder):\n" + " def find_spec(self, name, path=None, target=None):\n" + f" if name.split('.')[0] in ('httpx', 'opentelemetry'):\n" + " raise ImportError(name)\n" + " return None\n" + "sys.meta_path.insert(0, Block())\n" + "from ctrlrun import MissingDependency\n" + "try:\n" + f" {factory}\n" + "except MissingDependency as exc:\n" + " print(str(exc))\n" + "except BaseException as exc:\n" + " print('WRONG:', type(exc).__name__, exc)\n" + ) + finished = subprocess.run( + [sys.executable, "-c", script], capture_output=True, text=True, check=True + ) + + assert f"pip install 'ctrlrun[{extra}]'" in finished.stdout, finished.stdout + assert "WRONG" not in finished.stdout + + +def test_the_otel_extra_declares_the_api_sdk_and_exporter(): + """§11 — the API alone would do for a library, but an operator installing this wants to + export without assembling a stack.""" + declared = " ".join(_pyproject()["project"]["optional-dependencies"]["otel"]) + + assert "opentelemetry-api" in declared + assert "opentelemetry-sdk" in declared + assert "otlp" in declared + + def test_core_declares_only_pyyaml_and_click(): declared = { name.split("[")[0].split(">")[0].split("=")[0].strip().lower() From 920e4f3307070e35726a9fb82231e4d8cc19c111 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 02:56:16 +0530 Subject: [PATCH 2/2] test: five OTel paths no test reached, one of them a wrong-trace bug MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mutation-testing §8 found five gaps, and the last one was worth the whole exercise. A receipt whose span had been evicted could have ended somebody else's span — closing a live action and stamping it with a different action's outcome. The window is narrow and real: a suspended action is the only way a receipt arrives long after its ACTION_PROPOSED under the same action_id (§6.9), so if its span is evicted meanwhile, that receipt has no span to end. A wrong trace is worse than a missing one, and nothing tested it. My first attempt at that test was wrong in an instructive way: I reached for a re-presented approval, which is a *new* action_id with its own span, so the window never opened. Suspended/resume is the only path that reopens it. The other four: an event for an action the sink never saw start (a restarted process, a `ctrlrun resolve` days later) must open no span; an evicted span is dropped rather than ended, because ending one asserts an outcome that never happened; a receipt ends only its own action's span; and an argument value OpenTelemetry cannot carry is dropped rather than stringified, since an attribute whose type changed on the way out is worse than one that is absent. M20 is an equivalent mutant against the import test and was retargeted: a module-scope opentelemetry import is invisible to `import ctrlrun`, because __init__ does not import ctrlrun.otel. It is caught by T30's MissingDependency test, which is where the guarantee actually lives. --- tests/test_otel.py | 111 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/tests/test_otel.py b/tests/test_otel.py index b49da89b..4646fba7 100644 --- a/tests/test_otel.py +++ b/tests/test_otel.py @@ -351,6 +351,117 @@ def test_the_open_span_map_is_bounded(store, exporter): assert sink.open_spans <= 4 +def test_an_event_for_an_action_this_sink_never_saw_start_opens_no_span(store, exporter): + """Every action opens with ACTION_PROPOSED (v0.1 §6.2). An event for one this sink never + saw belongs to a span somebody else owns — a process that restarted, or a `ctrlrun + resolve` run days later — and inventing one would export a span with no beginning.""" + from datetime import UTC, datetime + + from ctrlrun.receipt import Event, EventType + + export, provider = exporter + sink = OTelEventSink(tracer_provider=provider) + + sink.on_event( + Event( + type=EventType.EFFECT_RESOLVED, + action_id="act_from_another_process", + ts=datetime.now(UTC), + data={"state": "failed", "resolved_by": "human"}, + ) + ) + + assert sink.open_spans == 0 + assert _spans(export) == () + + +def test_an_evicted_span_is_dropped_rather_than_ended(store, exporter): + """Ending one would export a span asserting an outcome that never happened.""" + export, provider = exporter + sink = OTelEventSink(tracer_provider=provider, max_open_spans=2) + control = Control(Policy.from_yaml(POLICY), store, sinks=[sink]) + refund = _refund(control) + + for index in range(6): + with context(agent="refund-agent"), pytest.raises(ApprovalRequired): + refund(payment_id=f"txn_{index}", amount=2000) + + assert sink.open_spans <= 2 + assert _spans(export) == (), "an evicted span must never reach the exporter" + + +def test_a_receipt_ends_only_its_own_action_span(store, exporter): + """One action commits while another is suspended awaiting a human. The receipt must end + the span it names and leave the other open, not the nearest one to hand.""" + export, provider = exporter + control = _control(store, provider) + refund = _refund(control) + + with context(agent="refund-agent"), pytest.raises(ApprovalRequired): + refund(payment_id="txn_waiting", amount=2000) + with context(agent="refund-agent"): + refund(payment_id="txn_done", amount=200) + + spans = _spans(export) + assert len(spans) == 1 + assert _attrs(spans[0])["ctrlrun.effect_key"] == "refund:txn_done" + + +def test_a_receipt_whose_span_was_evicted_ends_nobody_elses(store, exporter): + """The window eviction opens, and the one that matters. + + A suspended action is the only way a receipt arrives long after its `ACTION_PROPOSED` + under the *same* action_id (§6.9): `Control.resume` writes one for the action that + suspended. If its span was evicted meanwhile, that receipt has no span to end — and + ending "the nearest span to hand" would close a live action's span and stamp it with a + different action's outcome. A wrong trace is worse than a missing one. + """ + from ctrlrun import Suspended + + export, provider = exporter + sink = OTelEventSink(tracer_provider=provider, max_open_spans=1) + control = Control(Policy.from_yaml(POLICY), store, sinks=[sink]) + + @protect("stripe.refund", effect="refund:{payment_id}", control=control) + def suspending(payment_id: str, amount: int) -> str: + raise Suspended("held-continuation") + + with context(agent="refund-agent"), pytest.raises(Suspended): + suspending(payment_id="txn_suspended", amount=200) + + # A second action's ACTION_PROPOSED evicts the suspended action's span. + with context(agent="refund-agent"), pytest.raises(ApprovalRequired): + _refund(control)(payment_id="txn_holds_the_slot", amount=2000) + + # The suspended action now finishes, and its receipt carries the original action_id. + control.resume("held-continuation", lambda: "re_1") + + assert _spans(export) == (), "the evicted action had no span, and took nobody else's" + assert sink.open_spans == 1 + + +def test_an_argument_value_otel_cannot_carry_is_dropped_not_stringified(store, exporter): + """An attribute whose type changed on the way out is worse than one that is absent: a + query written against it silently stops matching.""" + export, provider = exporter + control = Control( + Policy.from_yaml(POLICY), + store, + sinks=[OTelEventSink(tracer_provider=provider, arguments=True)], + ) + + @protect("stripe.refund", effect="refund:{payment_id}", control=control) + def refund(payment_id: str, amount: int, tags: list) -> str: + return "re_1" + + with context(agent="refund-agent"): + refund(payment_id="txn_1", amount=200, tags=["a", "b"]) + + attributes = _attrs(_spans(export)[0]) + assert attributes["ctrlrun.argument.payment_id"] == "txn_1" + assert "ctrlrun.argument.tags" not in attributes + + # --- T30: the extra says so when it is absent ------------------------------------------