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..4646fba7 --- /dev/null +++ b/tests/test_otel.py @@ -0,0 +1,477 @@ +"""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 + + +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 ------------------------------------------ + + +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()