[v1.0] Preserve process-control exceptions across optional integrations

[codeforester]

Goal

Allow Ctrl+C, explicit process exit, and generator shutdown to cross optional integration and extension boundaries.

Background

Rich rendering, telemetry start/finish, safe span calls, and extension loading catch BaseException:

base-cli/lib/python/base_cli/integrations.py

Lines 55 to 92 in 8a93d22

	try:
	from rich.box import SIMPLE_HEAD
	from rich.console import Console
	from rich.table import Table
	from rich.text import Text
	except BaseException: # pragma: no cover - depends on optional package
	return False
	try:
	table = Table(
	box=SIMPLE_HEAD,
	expand=False,
	show_header=True,
	header_style="bold",
	pad_edge=False,
	)
	for header in headers:
	table.add_column(header, overflow="ellipsis")
	for row in rows:
	table.add_row(*(Text(value) for value in row))
	console_kwargs: dict[str, Any] = {
	"file": stream,
	"force_terminal": True,
	"color_system": None,
	"markup": False,
	"highlight": False,
	}
	if terminal_width is not None:
	console_kwargs["width"] = max(1, terminal_width)
	console = Console(**console_kwargs)
	console.print(table)
	if footer:
	console.print()
	console.print(footer)
	return True
	except BaseException: # pragma: no cover - depends on optional package
	return False

,

base-cli/lib/python/base_cli/integrations.py

Lines 95 to 157 in 8a93d22

	def start_telemetry(options: TelemetryOptions | None, context: Any) -> TelemetrySession | None:
	"""Start a safe lifecycle span, returning ``None`` on any integration failure."""
	if options is None or not options.enabled:
	return None
	try:
	tracer = options.tracer
	if tracer is None:
	from opentelemetry import trace
	tracer = trace.get_tracer(
	options.tracer_name,
	tracer_provider=options.tracer_provider,
	)
	attributes = _start_attributes(context)
	try:
	span = tracer.start_span("base_cli.run", attributes=attributes)
	except TypeError:
	# Small test/demonstration tracers may only accept a span name.
	span = tracer.start_span("base_cli.run")
	if span is None:
	return None
	_safe_span_call(span, "add_event", "base_cli.run.started", attributes=attributes)
	return TelemetrySession(span=span, started_monotonic_ns=time.monotonic_ns())
	except BaseException as exc: # pragma: no cover - optional package/runtime dependent
	_debug_integration_failure(context, "OpenTelemetry start failed", exc)
	return None
	def finish_telemetry(
	session: TelemetrySession | None,
	context: Any,
	outcome: Any,
	*,
	ended_monotonic_ns: int | None = None,
	) -> None:
	"""Finish a lifecycle span without allowing exporters to affect teardown."""
	if session is None:
	return
	try:
	ended = ended_monotonic_ns if ended_monotonic_ns is not None else time.monotonic_ns()
	duration_ms = round(max(0, ended - session.started_monotonic_ns) / 1_000_000)
	attributes = {
	**_start_attributes(context),
	"base_cli.outcome": str(getattr(outcome, "kind", "unknown")),
	"base_cli.status": str(getattr(outcome, "status", "error")),
	"base_cli.exit_code": int(getattr(outcome, "exit_code", 1)),
	"base_cli.duration_ms": duration_ms,
	}
	for key, value in attributes.items():
	_safe_span_call(session.span, "set_attribute", key, value)
	_safe_span_call(
	session.span,
	"add_event",
	"base_cli.run.finished",
	attributes=attributes,
	)
	_safe_span_call(session.span, "end")
	except BaseException as exc: # pragma: no cover - optional exporter dependent
	_debug_integration_failure(context, "OpenTelemetry finish failed", exc)

, and

base-cli/lib/python/base_cli/extensions.py

Lines 220 to 244 in 8a93d22

	def load(self, group: str, name: str) -> Any:
	"""Load one uniquely named extension and cache the resulting object."""
	if self.disabled:
	raise ExtensionsDisabledError("Python extension discovery is disabled")
	_validate_group(group)
	matches = tuple(descriptor for descriptor in self.list(group) if descriptor.name == name)
	if not matches:
	raise ExtensionDiscoveryError(f"No extension named '{name}' exists in group '{group}'.")
	if len(matches) > 1:
	raise ExtensionCollisionError(group, name, matches)
	key = (group, name)
	with self._lock:
	if key in self._loaded_cache:
	return self._loaded_cache[key]
	descriptor = matches[0]
	if descriptor.api_version not in self.supported_api_versions:
	raise ExtensionCompatibilityError(descriptor, tuple(sorted(self.supported_api_versions)))
	try:
	value = self._load_descriptor(descriptor)
	except BaseException as exc: # isolate third-party import failures
	raise ExtensionLoadError(descriptor, exc) from exc
	with self._lock:
	self._loaded_cache[key] = value
	return value

. A tracer whose start_span() raises KeyboardInterrupt is silently converted to None; an extension doing the same is wrapped as an extension load failure.

Catching third-party failures is appropriate, but swallowing KeyboardInterrupt, SystemExit, and GeneratorExit changes process semantics and can make an SRE command appear unresponsive.

Scope

• Audit every except BaseException boundary.
• Catch ordinary integration failures with Exception.
• Preserve process-control exceptions unless a teardown site has a narrowly documented reason to retain the primary outcome.

Acceptance Criteria

• Rich import/render, telemetry startup, and extension import propagate process-control exceptions.
• Ordinary optional-dependency and exporter failures remain isolated as documented.
• Intentional teardown shielding is explicit and tested separately.
• Ctrl+C produces the framework's canonical interrupted outcome.
• Tests cover KeyboardInterrupt, SystemExit, GeneratorExit, and normal plugin exceptions at each public boundary.

Validation

Run integration, extension, signal, lifecycle, and subprocess tests.

Non-Goals

Do not make ordinary telemetry/exporter failures fail successful commands.

Project Fields

• Status: Backlog
• Priority: P1
• Area: Python
• Initiative: v1.0 Readiness
• Size: S

Ownership

• Assignee: @codeforester