From 507aaf0e994dfef74edfbca2fb3bf65639e41457 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sun, 2 Aug 2026 07:32:11 +0800 Subject: [PATCH 1/4] fix(evidence): preserve adapter and setup failure diagnostics --- src/flameox/adapters/pytest.py | 127 ++++++++++++++++++++-- src/flameox/adapters/setup_runtime.py | 57 +++++++++- src/flameox/application/capabilities.py | 109 ++++++++++++++++--- src/flameox/application/capture.py | 32 +++++- src/flameox/application/operations.py | 22 +++- src/flameox/collectors/torch_launcher.py | 131 +++++++++++++++-------- src/flameox/domain/models.py | 1 + tests/adapters/test_pytest.py | 87 ++++++++++++++- tests/application/test_capabilities.py | 124 ++++++++++++++++++++- tests/collectors/test_torch_launcher.py | 72 +++++++++++++ tests/ownership.toml | 6 ++ 11 files changed, 689 insertions(+), 79 deletions(-) create mode 100644 tests/collectors/test_torch_launcher.py diff --git a/src/flameox/adapters/pytest.py b/src/flameox/adapters/pytest.py index 62bf732..d4f1ed8 100644 --- a/src/flameox/adapters/pytest.py +++ b/src/flameox/adapters/pytest.py @@ -22,6 +22,7 @@ class PytestExtractionResult(ContractModel): executed_count: int passed_count: int failed_count: int + environment_blocked_count: int skipped_count: int errored_count: int unexecuted_count: int @@ -168,6 +169,20 @@ def extract(self, run_id: str) -> PytestExtractionResult: outcome_counts = _test_outcomes(phases) executed = set(phases) unexecuted = collected - executed + compiler_diagnostic = self._external_compiler_diagnostic(run) + environment_blocked_nodes, environment_blocked_count, classified_counts = ( + self._classify_external_compiler_outcomes( + phases, + unexecuted, + compiler_diagnostic is not None, + ) + ) + self._apply_external_compiler_evidence( + rows, + limitations, + environment_blocked_nodes, + compiler_diagnostic, + ) first_observed = min(failure_observed, default=None) first_reported = min(failure_reported, default=None) collection_started = [ @@ -189,9 +204,10 @@ def extract(self, run_id: str) -> PytestExtractionResult: ("pytest.tests.collected", len(collected)), ("pytest.tests.executed", len(executed)), ("pytest.tests.passed", outcome_counts["passed"]), - ("pytest.tests.failed", outcome_counts["failed"]), - ("pytest.tests.skipped", outcome_counts["skipped"]), - ("pytest.tests.errored", outcome_counts["errored"]), + ("pytest.tests.failed", classified_counts["failed"]), + ("pytest.tests.environment_blocked", environment_blocked_count), + ("pytest.tests.skipped", classified_counts["skipped"]), + ("pytest.tests.errored", classified_counts["errored"]), ("pytest.tests.unexecuted", len(unexecuted)), ("pytest.fixture_setup.total", fixture_setup_ns), ): @@ -273,9 +289,10 @@ def extract(self, run_id: str) -> PytestExtractionResult: collected_count=len(collected), executed_count=len(executed), passed_count=outcome_counts["passed"], - failed_count=outcome_counts["failed"], - skipped_count=outcome_counts["skipped"], - errored_count=outcome_counts["errored"], + failed_count=classified_counts["failed"], + environment_blocked_count=environment_blocked_count, + skipped_count=classified_counts["skipped"], + errored_count=classified_counts["errored"], unexecuted_count=len(unexecuted), fixture_setup_count=fixture_setup_count, fixture_setup_ns=fixture_setup_ns, @@ -291,6 +308,92 @@ def extract(self, run_id: str) -> PytestExtractionResult: limitations=tuple(limitations), ) + def _external_compiler_diagnostic(self, run: RunManifest) -> str | None: + outputs: list[str] = [] + for registration in run.artifacts: + if registration.kind is not ArtifactKind.PROCESS_OUTPUT: + continue + if registration.role not in {"stdout", "stderr"}: + continue + try: + payload = ArtifactStore(self.workspace).get(registration.artifact_id) + outputs.append( + payload.payload_path.read_bytes()[: 64 * 1024].decode(errors="replace") + ) + except (DomainError, OSError, UnicodeError): + continue + combined = " ".join(outputs) + lowered = combined.casefold() + markers = ( + "cuda_runtime.h: no such file or directory", + "fatal error: cuda_runtime.h", + "nvcc fatal", + "cannot find -lcudart", + "hip_runtime.h: no such file or directory", + "clang: error: no such file or directory", + "gcc: fatal error", + ) + if not any(marker in lowered for marker in markers): + return None + return ( + " ".join(combined.split())[:500] or "External compiler returned no diagnostic detail." + ) + + @staticmethod + def _apply_external_compiler_evidence( + rows: list[dict[str, Any]], + limitations: list[str], + environment_blocked_nodes: set[str], + compiler_diagnostic: str | None, + ) -> None: + if environment_blocked_nodes: + for row in rows: + dimensions = row.get("dimensions") + nodeid = dimensions.get("nodeid") if isinstance(dimensions, dict) else None + if nodeid in environment_blocked_nodes: + assert isinstance(dimensions, dict) + dimensions["classification"] = "environment_blocked" + dimensions["original_outcome"] = dimensions.get("outcome", "unknown") + if compiler_diagnostic is not None: + limitations.extend( + ( + "An external compiler prerequisite failure was detected; affected pytest " + "outcomes are environment-blocked evidence, not a confirmed application " + "defect.", + f"External compiler diagnostic: {compiler_diagnostic}", + ) + ) + + @staticmethod + def _classify_external_compiler_outcomes( + phases: dict[str, dict[str, str]], + unexecuted: set[str], + compiler_failed: bool, + ) -> tuple[set[str], int, dict[str, int]]: + outcome_counts = _test_outcomes(phases) + environment_blocked_nodes = { + nodeid + for nodeid, reports in phases.items() + if compiler_failed and _test_outcome(reports) in {"failed", "errored"} + } + environment_blocked_count = len(environment_blocked_nodes) + if compiler_failed and not environment_blocked_nodes: + environment_blocked_count = len(unexecuted) + blocked_outcomes = { + outcome: sum( + 1 + for nodeid in environment_blocked_nodes + if _test_outcome(phases[nodeid]) == outcome + ) + for outcome in ("failed", "errored") + } + classified_counts = { + **outcome_counts, + "failed": outcome_counts["failed"] - blocked_outcomes["failed"], + "errored": outcome_counts["errored"] - blocked_outcomes["errored"], + } + return environment_blocked_nodes, environment_blocked_count, classified_counts + def _registration(self, run: RunManifest) -> Any: matches = [item for item in run.artifacts if item.kind is ArtifactKind.TEST_EXECUTION] if len(matches) != 1: @@ -391,6 +494,18 @@ def _test_outcomes(phases: dict[str, dict[str, str]]) -> dict[str, int]: return counts +def _test_outcome(reports: dict[str, str]) -> str: + if reports.get("setup") == "failed" or reports.get("teardown") == "failed": + return "errored" + if reports.get("call") == "failed": + return "failed" + if "skipped" in reports.values(): + return "skipped" + if reports.get("call") == "passed": + return "passed" + return "errored" + + def _completion( events: list[dict[str, Any]], *, diff --git a/src/flameox/adapters/setup_runtime.py b/src/flameox/adapters/setup_runtime.py index fa76a58..fa2343c 100644 --- a/src/flameox/adapters/setup_runtime.py +++ b/src/flameox/adapters/setup_runtime.py @@ -316,14 +316,21 @@ def install_trace_processor( ) atomic_write_text(workspace.paths.config, updated.to_toml()) temporary = None - except DomainError: - raise + except DomainError as exc: + raise _annotate_staging_error(exc) from exc except (OSError, urllib.error.URLError, ValueError) as exc: + category = _staging_failure_category(exc) + detail = _bounded_staging_detail(exc) raise DomainError( ErrorCode.PROCESS_FAILED, "FlameOx could not stage the managed Trace Processor.", retryable=True, - details={"next_tool": "prepare_capabilities", "adapter": "perfetto"}, + details={ + "next_tool": "prepare_capabilities", + "adapter": "perfetto", + "failure_category": category, + "failure_detail": detail, + }, remediation=( "Retry prepare_capabilities; if the download remains unavailable, install " "the official user-space binary or configure analysis.trace_processor_path.", @@ -335,6 +342,50 @@ def install_trace_processor( return TraceProcessorInstallation(TRACE_PROCESSOR_VERSION, target, True) +def _annotate_staging_error(error: DomainError) -> DomainError: + """Retain a bounded cause when a staging helper already raised a domain error.""" + details = dict(error.details) + details.setdefault("failure_category", _domain_failure_category(error)) + details.setdefault( + "failure_detail", + _bounded_staging_detail(details.get("error") or error.message), + ) + details.setdefault("phase", "staging_trace_processor") + return DomainError( + error.code, + error.message, + retryable=error.retryable, + details=details, + remediation=error.remediation, + run_id=error.run_id, + ) + + +def _domain_failure_category(error: DomainError) -> str: + if error.code is ErrorCode.PROCESS_CANCELLED: + return "cancelled" + if error.code is ErrorCode.PROCESS_TIMEOUT: + return "timeout" + if error.code is ErrorCode.ARTIFACT_TOO_LARGE: + return "download_limit" + if error.code is ErrorCode.CAPABILITY_UNAVAILABLE: + return "unsupported_platform" + return "verification" + + +def _staging_failure_category(error: BaseException) -> str: + if isinstance(error, urllib.error.URLError): + return "network" + if isinstance(error, OSError): + return "filesystem" + return "verification" + + +def _bounded_staging_detail(error: object) -> str: + detail = " ".join(str(error).split()) + return detail[:500] or "The staging operation returned no diagnostic detail." + + def _verify_trace_processor( executable: Path, *, diff --git a/src/flameox/application/capabilities.py b/src/flameox/application/capabilities.py index ae3b340..76fdafe 100644 --- a/src/flameox/application/capabilities.py +++ b/src/flameox/application/capabilities.py @@ -4,6 +4,7 @@ import json import os import platform +import re import shutil import subprocess import sys @@ -476,6 +477,8 @@ async def _probe_perf(self, passive: CapabilityReport) -> CapabilityReport: ) or diagnostic ) + if self._is_perf_permission_denial(diagnostic): + return self._perf_permission_failure(passive, diagnostic) return self._perf_failure(passive, diagnostic) except (OSError, ValueError) as error: return self._perf_failure(passive, f"Active perf probe failed: {type(error).__name__}.") @@ -495,21 +498,25 @@ async def _probe_perf(self, passive: CapabilityReport) -> CapabilityReport: } ) if self._is_perf_permission_denial(diagnostic): - return passive.model_copy( - update={ - "status": CapabilityStatus.PERMISSION_REQUIRED, - "permission_status": "denied", - "limitations": (diagnostic or "perf event access was denied.",), - "remediation": ( - "Lower the kernel perf_event_paranoid policy or grant the process " - "perf event access, then refresh capabilities.", - ), - "probe_kind": "active", - "probed_at": utc_now(), - } - ) + return self._perf_permission_failure(passive, diagnostic) return self._perf_failure(passive, diagnostic or "perf sampling probe failed.") + def _perf_permission_failure( + self, + passive: CapabilityReport, + diagnostic: str, + ) -> CapabilityReport: + return passive.model_copy( + update={ + "status": CapabilityStatus.PERMISSION_REQUIRED, + "permission_status": "denied", + "limitations": (diagnostic or "perf event access was denied.",), + "remediation": (self._perf_remediation(diagnostic),), + "probe_kind": "active", + "probed_at": utc_now(), + } + ) + def _perf_failure(self, passive: CapabilityReport, diagnostic: str) -> CapabilityReport: return passive.model_copy( update={ @@ -539,6 +546,19 @@ def _is_perf_permission_denial(diagnostic: str) -> bool: ) ) + @staticmethod + def _perf_remediation(diagnostic: str) -> str: + lowered = diagnostic.casefold() + match = re.search(r"perf_event_paranoid(?: setting is|=)\s*(\d+)", lowered) + setting = match.group(1) if match is not None else None + current = f" (observed perf_event_paranoid={setting})" if setting else "" + return ( + "perf sampling is unusable because the kernel denied perf_event_open" + f"{current}. Lower kernel.perf_event_paranoid to a policy value that permits " + "this process, or grant CAP_PERFMON/CAP_SYS_ADMIN according to local policy, " + "then call list_capabilities(mode='active_refresh') before planning." + ) + @staticmethod def _bounded_diagnostic(value: str) -> str: normalized = " ".join(value.split()) @@ -562,6 +582,7 @@ def prepare( adapters: tuple[str, ...], *, cancel_event: threading.Event | None = None, + phase_callback: Callable[[str], None] | None = None, ) -> CapabilitySetupResult: """Install only declared FlameOx-managed providers into this runtime.""" reports = {item.adapter: item for item in self.list().capabilities} @@ -686,22 +707,32 @@ def prepare( "A workspace is required to stage the managed Trace Processor.", details={"next_tool": "initialize_workspace"}, ) + self._record_setup_receipt( + requested, + completed=self._available_requested(requested), + phase="staging_trace_processor", + ) + if phase_callback is not None: + phase_callback("staging_trace_processor") install_trace_processor(self.workspace, cancel_event=cancel_event) self._check_cancelled(cancel_event) except DomainError as exc: + failure = self._annotate_setup_phase(exc, pending_trace=pending_trace) self._record_setup_receipt( requested, completed=self._available_requested(requested), phase="failed", - error=exc.message, + error=self._setup_failure_message(failure), ) - raise + if failure is exc: + raise + raise failure from exc except (OSError, subprocess.SubprocessError, portalocker.exceptions.LockException) as exc: self._record_setup_receipt( requested, completed=self._available_requested(requested), phase="failed", - error=str(exc)[:500], + error=self._bounded_setup_detail(exc), ) raise DomainError( ErrorCode.PROCESS_FAILED, @@ -753,6 +784,36 @@ def prepare( setup_verification=self._verification(requested, refreshed), ) + @staticmethod + def _bounded_setup_detail(error: object) -> str: + detail = " ".join(str(error).split()) + return detail[:500] or "Capability setup returned no diagnostic detail." + + @staticmethod + def _annotate_setup_phase(error: DomainError, *, pending_trace: bool) -> DomainError: + if not pending_trace or isinstance(error.details.get("phase"), str): + return error + details = dict(error.details) + details["phase"] = "staging_trace_processor" + return DomainError( + error.code, + error.message, + retryable=error.retryable, + details=details, + remediation=error.remediation, + run_id=error.run_id, + ) + + @classmethod + def _setup_failure_message(cls, error: DomainError) -> str: + phase = error.details.get("phase") + category = error.details.get("failure_category") + detail = error.details.get("failure_detail") or error.details.get("error") + if not isinstance(category, str) or not isinstance(detail, str): + return error.message + phase_label = f" [phase={phase}]" if isinstance(phase, str) else "" + return f"{error.message}{phase_label} [{category}] {cls._bounded_setup_detail(detail)}" + @staticmethod def _check_cancelled(cancel_event: threading.Event | None) -> None: if cancel_event is not None and cancel_event.is_set(): @@ -1095,12 +1156,26 @@ async def _run( await progress("installing_packages", 1, 3, "Installing declared optional providers.") cancel_event = threading.Event() self.runner.set_cancel_hook(operation_id, cancel_event.set) + loop = asyncio.get_running_loop() + + def report_phase(phase: str) -> None: + if phase != "staging_trace_processor": + return + + async def emit_progress() -> None: + await progress(phase, 2, 3, "Staging the managed Trace Processor.") + + future = asyncio.run_coroutine_threadsafe(emit_progress(), loop) + future.result() + try: try: + requested = self._requested(operation_id) result = await asyncio.to_thread( self.service.prepare, - self._requested(operation_id), + requested, cancel_event=cancel_event, + phase_callback=report_phase, ) except DomainError as error: receipt = self.service._read_setup_receipt() diff --git a/src/flameox/application/capture.py b/src/flameox/application/capture.py index 0795eaa..0c044b2 100644 --- a/src/flameox/application/capture.py +++ b/src/flameox/application/capture.py @@ -1122,6 +1122,34 @@ async def execute( validation_status = ValidationStatus.ERROR validation_limitations.append(error.message) await capture.report(6, "Validation complete") + torch_diagnostics = output_root / "torch-profiler-diagnostics.json" + if plan.adapter == "torch.profiler" and torch_diagnostics.is_file(): + try: + diagnostic_payload = json.loads(torch_diagnostics.read_text(encoding="utf-8")) + diagnostic_phase = diagnostic_payload.get("phase") + diagnostic_status = diagnostic_payload.get("status") + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + diagnostic_phase = None + diagnostic_status = None + if diagnostic_status == "failed" and isinstance(diagnostic_phase, str): + collector_limitation_details.append( + _limitation( + "collector", + "failure_phase", + f"Torch profiler collector failed during {diagnostic_phase}.", + ) + ) + registrations.append( + await self._register_path_async( + run_id, + torch_diagnostics, + kind=ArtifactKind.COLLECTOR_METADATA, + role="torch_profiler_diagnostics", + media_type="application/json", + producer=plan.adapter, + producer_version=plan.adapter_version, + ) + ) for name, payload, kind, role, media_type in ( ( "stdout.bin", @@ -1138,7 +1166,9 @@ async def execute( "application/octet-stream", ), ): - if not payload: + if not payload and not ( + plan.adapter == "torch.profiler" and not collector_succeeded + ): continue path = output_root / name atomic_write_bytes(path, payload) diff --git a/src/flameox/application/operations.py b/src/flameox/application/operations.py index 341b73a..0daaef8 100644 --- a/src/flameox/application/operations.py +++ b/src/flameox/application/operations.py @@ -77,6 +77,7 @@ class OperationRecord(ContractModel): item_outcomes: tuple[OperationItemOutcome, ...] = Field(default=(), max_length=64) failure_code: str | None = None failure_message: str | None = None + failure_details: dict[str, Any] | None = None cancellation_requested: bool = False cleanup_status: Literal["not_required", "pending", "complete", "incomplete"] = "not_required" terminal_receipt: dict[str, Any] | None = None @@ -104,6 +105,7 @@ class OperationStatus(ContractModel): cleanup_status: Literal["not_required", "pending", "complete", "incomplete"] failure_code: str | None failure_message: str | None + failure_details: dict[str, Any] | None terminal_receipt: dict[str, Any] | None recovery: OperationRecovery | None created_at: datetime @@ -400,13 +402,16 @@ async def _finish_domain_failure( recovery=self._retry_recovery(self.store.read(operation_id)), ) else: + failure_details = self._failure_details(error) + failure_phase = failure_details.get("phase") await self._update( operation_id, state="failed", - phase="failed", + phase=(failure_phase if isinstance(failure_phase, str) else "failed"), cleanup_status="complete", failure_code=error.code.value, failure_message=error.message, + failure_details=failure_details, item_outcomes=self._items( operation_id, "retryable" if error.retryable else "failed", @@ -423,6 +428,21 @@ async def _finish_domain_failure( ), ) + @staticmethod + def _failure_details(error: DomainError) -> dict[str, Any]: + """Persist only bounded, recovery-relevant diagnostics from a domain failure.""" + details: dict[str, Any] = {} + for key in ("phase", "failure_category", "adapter", "next_tool"): + value = error.details.get(key) + if isinstance(value, str) and value: + details[key] = value[:200] + detail = error.details.get("failure_detail") or error.details.get("error") + if detail is not None: + normalized = " ".join(str(detail).split())[:500] + if normalized: + details["failure_detail"] = normalized + return details + async def _finish_unexpected_failure(self, operation_id: str, error: Exception) -> None: await self._update( operation_id, diff --git a/src/flameox/collectors/torch_launcher.py b/src/flameox/collectors/torch_launcher.py index dcf938b..38efdb2 100644 --- a/src/flameox/collectors/torch_launcher.py +++ b/src/flameox/collectors/torch_launcher.py @@ -4,8 +4,29 @@ import json import runpy import sys +from contextlib import suppress from pathlib import Path +from flameox.atomic import atomic_write_json + + +def _write_diagnostic( + path: Path, + *, + phase: str, + status: str, + detail: str | None = None, +) -> None: + payload: dict[str, object] = { + "schema_version": "flameox.torch-profiler-diagnostics.v1", + "phase": phase, + "status": status, + } + if detail is not None: + payload["detail"] = " ".join(detail.split())[:500] + with suppress(OSError): + atomic_write_json(path, payload) + def main() -> None: parser = argparse.ArgumentParser( @@ -18,56 +39,72 @@ def main() -> None: target.add_argument("--script") parser.add_argument("arguments", nargs=argparse.REMAINDER) options = parser.parse_args() - - if options.module is not None: - sys.path.insert(0, str(Path.cwd())) - script_path = None - else: - script_path = Path(options.script).resolve() - sys.path.insert(0, str(script_path.parent)) - try: - import torch - except ImportError as exc: - parser.error(f"PyTorch is unavailable: {exc}") + diagnostic_path = Path(options.output).resolve().parent / "torch-profiler-diagnostics.json" + _write_diagnostic(diagnostic_path, phase="wrapper_startup", status="started") + phase = "wrapper_startup" try: - config = json.loads(options.config) - except json.JSONDecodeError as exc: - parser.error(f"Invalid profiler configuration: {exc}") - if not isinstance(config, dict) or config.get("mode") != "whole_entrypoint": - parser.error("Whole-entrypoint launcher requires whole_entrypoint mode") - configured_activities = config.get("activities") - if not isinstance(configured_activities, list): - parser.error("Profiler activities are missing") - activities = [] - if "cpu" in configured_activities: - activities.append(torch.profiler.ProfilerActivity.CPU) - if "cuda" in configured_activities and not torch.cuda.is_available(): - parser.error("The capture plan requires CUDA, but CUDA is unavailable") - if "cuda" in configured_activities: - activities.append(torch.profiler.ProfilerActivity.CUDA) - if "cuda_if_available" in configured_activities and torch.cuda.is_available(): - activities.append(torch.profiler.ProfilerActivity.CUDA) - if not activities: - parser.error("No requested torch.profiler activity is available") - output = Path(options.output) - output.parent.mkdir(parents=True, exist_ok=True) - target_name = options.module or options.script - assert target_name is not None - sys.argv = [target_name, *options.arguments] - with torch.profiler.profile( - activities=activities, - record_shapes=config["record_shapes"], - profile_memory=config["profile_memory"], - with_stack=config["with_stack"], - with_flops=config["with_flops"], - with_modules=config["with_modules"], - ) as profile: if options.module is not None: - runpy.run_module(options.module, run_name="__main__", alter_sys=True) + sys.path.insert(0, str(Path.cwd())) + script_path = None else: - assert script_path is not None - runpy.run_path(str(script_path), run_name="__main__") - profile.export_chrome_trace(str(output)) + script_path = Path(options.script).resolve() + sys.path.insert(0, str(script_path.parent)) + try: + import torch + except ImportError as exc: + parser.error(f"PyTorch is unavailable: {exc}") + try: + config = json.loads(options.config) + except json.JSONDecodeError as exc: + parser.error(f"Invalid profiler configuration: {exc}") + if not isinstance(config, dict) or config.get("mode") != "whole_entrypoint": + parser.error("Whole-entrypoint launcher requires whole_entrypoint mode") + configured_activities = config.get("activities") + if not isinstance(configured_activities, list): + parser.error("Profiler activities are missing") + activities = [] + if "cpu" in configured_activities: + activities.append(torch.profiler.ProfilerActivity.CPU) + if "cuda" in configured_activities and not torch.cuda.is_available(): + parser.error("The capture plan requires CUDA, but CUDA is unavailable") + if "cuda" in configured_activities: + activities.append(torch.profiler.ProfilerActivity.CUDA) + if "cuda_if_available" in configured_activities and torch.cuda.is_available(): + activities.append(torch.profiler.ProfilerActivity.CUDA) + if not activities: + parser.error("No requested torch.profiler activity is available") + output = Path(options.output) + output.parent.mkdir(parents=True, exist_ok=True) + target_name = options.module or options.script + assert target_name is not None + sys.argv = [target_name, *options.arguments] + phase = "workload_execution" + _write_diagnostic(diagnostic_path, phase=phase, status="running") + with torch.profiler.profile( + activities=activities, + record_shapes=config["record_shapes"], + profile_memory=config["profile_memory"], + with_stack=config["with_stack"], + with_flops=config["with_flops"], + with_modules=config["with_modules"], + ) as profile: + if options.module is not None: + runpy.run_module(options.module, run_name="__main__", alter_sys=True) + else: + assert script_path is not None + runpy.run_path(str(script_path), run_name="__main__") + phase = "trace_finalization" + _write_diagnostic(diagnostic_path, phase=phase, status="running") + profile.export_chrome_trace(str(output)) + _write_diagnostic(diagnostic_path, phase="completed", status="succeeded") + except BaseException as exc: + _write_diagnostic( + diagnostic_path, + phase=phase, + status="failed", + detail=f"{type(exc).__name__}: {exc}", + ) + raise if __name__ == "__main__": diff --git a/src/flameox/domain/models.py b/src/flameox/domain/models.py index 1d570cb..109d8f1 100644 --- a/src/flameox/domain/models.py +++ b/src/flameox/domain/models.py @@ -486,6 +486,7 @@ class RequirementResult(ContractModel): "available", "absent", "permission_denied", + "environment_blocked", "unsupported", "unknown", "probe_failed", diff --git a/tests/adapters/test_pytest.py b/tests/adapters/test_pytest.py index de00a59..5641108 100644 --- a/tests/adapters/test_pytest.py +++ b/tests/adapters/test_pytest.py @@ -9,8 +9,15 @@ from flameox.adapters import PytestExtractor from flameox.application import ImportArtifactRequest, ImportService from flameox.catalog import Catalog -from flameox.domain import ArtifactKind, DomainError, ErrorCode -from flameox.storage import Workspace +from flameox.domain import ( + ArtifactKind, + ArtifactRegistration, + DomainError, + ErrorCode, + Sensitivity, + new_id, +) +from flameox.storage import ArtifactStore, RunStore, Workspace def _event(event: str, **fields: Any) -> str: @@ -134,6 +141,82 @@ def test_pytest_extracts_fixture_cost_outcomes_and_failure_latency(tmp_path: Pat assert ("pytest.time_to_first_failure.reported", 1_800, None, None) in rows +def test_pytest_marks_external_cuda_compile_failure_as_environment_blocked( + tmp_path: Path, +) -> None: + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + source = tmp_path / "pytest-events.jsonl" + source.write_text( + "\n".join( + ( + _event( + "run_started", + run_started_at_ns=1_000, + pytest_version="9.0", + python_version="3.12", + platform="test", + scheduler="no", + requested_workers="0", + ), + _event("collection_started"), + _event("test_collected", nodeid="test_gpu.py::test_compile"), + _event( + "test_phase", + nodeid="test_gpu.py::test_compile", + worker_id="master", + phase="setup", + outcome="failed", + duration_ns=10, + started_at_ns=1_100, + stopped_at_ns=1_110, + controller_received_at_ns=1_120, + wasxfail=False, + ), + _event("session_finished", exit_status=1), + ) + ) + + "\n" + ) + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest(path=source, kind=ArtifactKind.TEST_EXECUTION) + ) + stderr = tmp_path / "stderr.bin" + stderr.write_text("fatal error: cuda_runtime.h: No such file or directory\n") + stored = ArtifactStore(workspace).import_path( + stderr, + allowed_roots=(tmp_path,), + max_bytes=workspace.config.capture.max_artifact_bytes, + ) + run = RunStore(workspace).read(imported.run.run_id) + stderr_registration = ArtifactRegistration( + registration_id=new_id(), + run_id=run.run_id, + artifact_id=stored.content.artifact_id, + display_name="stderr.bin", + media_type="application/octet-stream", + kind=ArtifactKind.PROCESS_OUTPUT, + role="stderr", + sensitivity=Sensitivity.NORMAL, + ) + RunStore(workspace).append( + run.model_copy(update={"revision": 2, "artifacts": (*run.artifacts, stderr_registration)}), + expected_revision=1, + ) + + result = PytestExtractor(workspace).extract(imported.run.run_id) + + assert result.failed_count == 0 + assert result.errored_count == 0 + assert result.environment_blocked_count == 1 + assert any("environment-blocked" in item for item in result.limitations) + with Catalog(workspace).open_snapshot() as snapshot: + row = snapshot.execute( + "SELECT value_int FROM measurements WHERE name = 'pytest.tests.environment_blocked'" + ).fetchone() + assert row == (1,) + + @pytest.mark.parametrize("payload", ("", "{}\n", "{broken\n")) def test_pytest_rejects_malformed_event_streams(tmp_path: Path, payload: str) -> None: workspace = Workspace.initialize(tmp_path) diff --git a/tests/application/test_capabilities.py b/tests/application/test_capabilities.py index 02fc19c..fd95466 100644 --- a/tests/application/test_capabilities.py +++ b/tests/application/test_capabilities.py @@ -127,7 +127,10 @@ async def test_perf_probe_exercises_permissions_and_cleans_staging( _probe_outcome( exit_code=1, stdout=b"", - stderr=b"Error: perf_event_open: Operation not permitted\n", + stderr=( + b"Error: perf_event_open: Operation not permitted\n" + b"perf_event_paranoid setting is 4\n" + ), ), _probe_outcome(exit_code=0), _probe_outcome(exit_code=1, stdout=b"", stderr=b"unexpected perf failure\n"), @@ -149,7 +152,8 @@ async def test_perf_probe_exercises_permissions_and_cleans_staging( assert cached == granted assert refreshed.status is CapabilityStatus.PERMISSION_REQUIRED assert refreshed.permission_status == "denied" - assert refreshed.remediation + assert "perf_event_paranoid=4" in refreshed.remediation[0] + assert "active_refresh" in refreshed.remediation[0] assert len(broker.requests) == 4 record_request = broker.requests[1] assert record_request.argv[1:8] == ( @@ -389,6 +393,59 @@ def test_prepare_capabilities_records_failure_when_uv_is_missing( assert receipt["error"] == "uv is missing from PATH." +def test_trace_processor_staging_preserves_phase_and_bounded_cause( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = Workspace.initialize(tmp_path) + service = CapabilityService( + workspace, + capability_manifest=tmp_path / "capabilities.json", + ) + report = CapabilityReport( + adapter="perfetto", + status=CapabilityStatus.UNAVAILABLE, + setup=CapabilitySetup( + extra="trace", + method="prepare_capabilities", + next_tool="prepare_capabilities", + requirement="perfetto>=0.57,<0.58", + ), + ) + monkeypatch.setattr(service, "list", lambda: CapabilityList(capabilities=(report,))) + monkeypatch.setattr("flameox.application.capabilities.shutil.which", lambda _: "/usr/bin/uv") + monkeypatch.setattr( + "flameox.application.capabilities.subprocess.run", + lambda *args, **kwargs: subprocess.CompletedProcess(args[0], 0, "", ""), + ) + + def fail_staging(*args: object, **kwargs: object) -> object: + raise DomainError( + ErrorCode.PROCESS_FAILED, + "FlameOx could not stage the managed Trace Processor.", + retryable=True, + details={ + "next_tool": "prepare_capabilities", + "adapter": "perfetto", + "failure_category": "network", + "failure_detail": "synthetic TLS failure", + }, + ) + + monkeypatch.setattr("flameox.application.capabilities.install_trace_processor", fail_staging) + phases: list[str] = [] + + with pytest.raises(DomainError): + service.prepare(("perfetto",), phase_callback=phases.append) + + receipt = json.loads((tmp_path / "capability-setup.json").read_text()) + assert phases == ["staging_trace_processor"] + assert receipt["phase"] == "failed" + assert "phase=staging_trace_processor" in receipt["error"] + assert "network" in receipt["error"] + assert "synthetic TLS failure" in receipt["error"] + + def test_prepare_capabilities_is_idempotent_when_provider_is_available( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -451,8 +508,10 @@ def prepare( adapters: tuple[str, ...], *, cancel_event: object, + phase_callback: Any, ) -> CapabilitySetupResult: del cancel_event + del phase_callback return CapabilitySetupResult( requested=adapters, installed=adapters, @@ -500,6 +559,67 @@ def _read_setup_receipt(self) -> None: await manager.shutdown() +@pytest.mark.anyio +async def test_capability_setup_failure_keeps_staging_phase_and_diagnostics( + tmp_path: Path, +) -> None: + workspace = Workspace.initialize(tmp_path) + + class FailingCapabilityService: + def prepare( + self, + adapters: tuple[str, ...], + *, + cancel_event: object, + phase_callback: Any, + ) -> CapabilitySetupResult: + del adapters, cancel_event + phase_callback("staging_trace_processor") + raise DomainError( + ErrorCode.PROCESS_FAILED, + "FlameOx could not stage the managed Trace Processor.", + retryable=True, + details={ + "adapter": "perfetto", + "phase": "staging_trace_processor", + "failure_category": "network", + "failure_detail": "synthetic TLS failure", + }, + ) + + def _read_setup_receipt(self) -> None: + return None + + manager = CapabilitySetupManager( + workspace, + cast(CapabilityService, FailingCapabilityService()), + ) + try: + started = await manager.start(("perfetto",), "staging-failure-proof") + failed = started + for _ in range(100): + failed = await manager.status(started.operation_id) + if failed.state == "failed": + break + await asyncio.sleep(0.01) + + assert failed.state == "failed" + assert failed.phase == "staging_trace_processor" + assert [item.phase for item in failed.progress] == [ + "validating_request", + "installing_packages", + "staging_trace_processor", + ] + assert failed.failure_details == { + "phase": "staging_trace_processor", + "failure_category": "network", + "adapter": "perfetto", + "failure_detail": "synthetic TLS failure", + } + finally: + await manager.shutdown() + + def test_entry_point_approval_is_revoked_when_installed_content_changes( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/collectors/test_torch_launcher.py b/tests/collectors/test_torch_launcher.py new file mode 100644 index 0000000..2e2601f --- /dev/null +++ b/tests/collectors/test_torch_launcher.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Literal + +import pytest + +from flameox.collectors import torch_launcher + + +def test_torch_launcher_records_workload_phase_on_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FakeProfile: + def __enter__(self) -> FakeProfile: + return self + + def __exit__(self, *_: object) -> Literal[False]: + return False + + def export_chrome_trace(self, path: str) -> None: + Path(path).write_text("{}") + + fake_torch = SimpleNamespace( + cuda=SimpleNamespace(is_available=lambda: False), + profiler=SimpleNamespace( + ProfilerActivity=SimpleNamespace(CPU="cpu", CUDA="cuda"), + profile=lambda **_: FakeProfile(), + ), + ) + monkeypatch.setitem(sys.modules, "torch", fake_torch) + workload = tmp_path / "failing_workload.py" + workload.write_text("raise RuntimeError('synthetic workload failure')\n") + output = tmp_path / "output" + output.mkdir() + monkeypatch.setattr( + sys, + "argv", + [ + "torch-launcher", + "--output", + str(output / "torch-trace.json"), + "--config", + json.dumps( + { + "mode": "whole_entrypoint", + "activities": ["cpu"], + "record_shapes": False, + "profile_memory": False, + "with_stack": False, + "with_flops": False, + "with_modules": False, + } + ), + "--script", + str(workload), + ], + ) + + with pytest.raises(RuntimeError, match="synthetic workload failure"): + torch_launcher.main() + + diagnostics = json.loads( + (output / "torch-profiler-diagnostics.json").read_text(encoding="utf-8") + ) + assert diagnostics["phase"] == "workload_execution" + assert diagnostics["status"] == "failed" + assert "synthetic workload failure" in diagnostics["detail"] diff --git a/tests/ownership.toml b/tests/ownership.toml index 1d572c1..a699650 100644 --- a/tests/ownership.toml +++ b/tests/ownership.toml @@ -423,6 +423,12 @@ lane = "optional" paths = ["tests/application/test_capture_torch_provider.py"] markers = ["integration", "optional", "process", "serial", "requires_torch"] +[[ownership]] +owner = "torch-launcher" +lane = "process" +paths = ["tests/collectors/test_torch_launcher.py"] +markers = ["unit"] + [[ownership]] owner = "investigation-revisions" lane = "application" From 8c1779eefc1a6157f4951977f033932ae97432d2 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sun, 2 Aug 2026 07:32:17 +0800 Subject: [PATCH 2/4] feat(preflight): classify unavailable perf and CUDA toolchains --- docs/adapters.md | 12 +- src/flameox/application/preflight.py | 185 ++++++++++++++++++++- tests/application/test_capture_planning.py | 44 +++++ 3 files changed, 239 insertions(+), 2 deletions(-) diff --git a/docs/adapters.md b/docs/adapters.md index 97099ba..c401ec5 100644 --- a/docs/adapters.md +++ b/docs/adapters.md @@ -268,7 +268,17 @@ disabled. A successful probe reports granted permission; a recognized remediation; other failures remain `degraded` with bounded diagnostics. Passive capability discovery therefore reports that an active probe is required, and a capture plan must bind and recheck the active report immediately before running -the workload. +the workload. A denied probe is reported as `permission_required`, which is +unusable for sampling until the host's `perf_event_open` policy is changed; +the report includes the observed kernel restriction and a remediation to grant +the required event access before refreshing capabilities. + +Workloads that declare `nvcc` receive a separate bounded CUDA toolkit preflight. +It compiles a tiny header probe and distinguishes an installed compiler from a +usable development toolkit. Missing `cuda_runtime.h` is recorded as +`environment_blocked` with the compiler diagnostic and a remediation to install +the CUDA development headers. Flameox does not build the workload during this +check. #### `perfetto` diff --git a/src/flameox/application/preflight.py b/src/flameox/application/preflight.py index 35f5288..0b3235d 100644 --- a/src/flameox/application/preflight.py +++ b/src/flameox/application/preflight.py @@ -1,6 +1,7 @@ from __future__ import annotations import shutil +import tempfile from importlib.metadata import PackageNotFoundError, distribution from pathlib import Path from typing import Literal @@ -17,6 +18,7 @@ RequirementResult, digest_model, ) +from flameox.execution import ExecutionRequest, SubprocessBroker from flameox.storage import Workspace @@ -26,10 +28,12 @@ def __init__( workspace: Workspace, *, capabilities: CapabilityService | None = None, + broker: SubprocessBroker | None = None, ) -> None: self.workspace = workspace self.workloads = WorkloadService(workspace) self.capabilities = capabilities or CapabilityService(workspace) + self.broker = broker or SubprocessBroker() async def inspect( self, @@ -41,7 +45,11 @@ async def inspect( requirements = config.requirements results: list[RequirementResult] = [] for name in requirements.executables: - results.append(self._executable(name, required=name not in requirements.optional)) + required = name not in requirements.optional + if name == "nvcc": + results.append(await self._cuda_toolkit(required=required, mode=mode)) + else: + results.append(self._executable(name, required=required)) for name in requirements.python_distributions: results.append(self._distribution(name, required=name not in requirements.optional)) passive = {item.adapter: item for item in self.capabilities.list().capabilities} @@ -148,6 +156,181 @@ def _executable(self, name: str, *, required: bool) -> RequirementResult: evidence=(str(path),), ) + async def _cuda_toolkit( + self, + *, + required: bool, + mode: Literal["passive", "active"], + ) -> RequirementResult: + resolved = shutil.which("nvcc") + if resolved is None: + return self._executable("nvcc", required=required) + path = Path(resolved).resolve() + try: + path.relative_to(self.workspace.project_root.resolve()) + except ValueError: + pass + else: + return RequirementResult( + requirement="nvcc", + kind="executable", + required=required, + probe_kind="active" if mode == "active" else "passive", + status="unsupported", + evidence=(str(path),), + limitations=( + "Repository-controlled nvcc is not used for CUDA toolkit readiness checks.", + ), + ) + if mode == "passive": + return RequirementResult( + requirement="nvcc", + kind="executable", + required=required, + probe_kind="active", + status="unknown", + identity=str(path), + evidence=(str(path),), + limitations=( + "nvcc is installed, but CUDA headers and host/device compilation were not " + "checked in passive preflight mode.", + ), + remediation=( + "Re-plan with preflight_mode='active' to run the bounded CUDA toolkit probe.", + ), + ) + + try: + with tempfile.TemporaryDirectory( + dir=self.workspace.paths.staging, + prefix="cuda-preflight-", + ) as temporary: + root = Path(temporary) + source = root / "header_probe.cu" + output = root / "header_probe.o" + source.write_text( + "#include \n" + "__global__ void flameox_probe_kernel() {}\n" + "int main() { return 0; }\n", + encoding="ascii", + ) + outcome = await self.broker.run( + ExecutionRequest( + argv=( + str(path), + "-x", + "cu", + "-c", + str(source), + "-o", + str(output), + ), + cwd=self.workspace.project_root, + environment_allowlist=("PATH",), + allowed_working_roots=(self.workspace.project_root, root), + timeout_seconds=30, + max_output_bytes=64 * 1024, + ) + ) + diagnostic = self._bounded_diagnostic( + (outcome.stdout + b"\n" + outcome.stderr).decode( + "utf-8", + errors="replace", + ) + ) + if outcome.process.exit_code == 0 and output.is_file() and output.stat().st_size: + return RequirementResult( + requirement="nvcc", + kind="executable", + required=required, + probe_kind="active", + status="available", + identity=str(path), + evidence=(str(path), "bounded_cuda_header_compile"), + ) + return self._cuda_compile_failure( + required=required, + path=path, + diagnostic=diagnostic, + ) + except DomainError as error: + process = error.details.get("process") + diagnostic = error.message + if isinstance(process, dict): + diagnostic = ( + " ".join( + str(value) + for value in (process.get("stdout"), process.get("stderr")) + if value + ) + or diagnostic + ) + return self._cuda_compile_failure( + required=required, + path=path, + diagnostic=self._bounded_diagnostic(diagnostic), + ) + except (OSError, ValueError) as error: + return self._cuda_compile_failure( + required=required, + path=path, + diagnostic=self._bounded_diagnostic(f"{type(error).__name__}: {error}"), + ) + + @classmethod + def _cuda_compile_failure( + cls, + *, + required: bool, + path: Path, + diagnostic: str, + ) -> RequirementResult: + lowered = diagnostic.casefold() + permission_denied = any( + marker in lowered + for marker in ("permission denied", "operation not permitted", "access denied") + ) + missing_header = "cuda_runtime.h" in lowered and any( + marker in lowered for marker in ("no such file", "not found", "cannot open") + ) + if permission_denied: + status: Literal["permission_denied", "environment_blocked"] = "permission_denied" + limitation = "The bounded CUDA toolkit compile was denied by the host environment." + remediation = ( + "Grant the configured process permission to invoke nvcc and access the CUDA " + "toolkit, then refresh preflight.", + ) + else: + status = "environment_blocked" + limitation = ( + "The CUDA toolkit is environment-blocked: the bounded nvcc compile did not " + "produce an object file." + ) + remediation = ( + "Install the CUDA development toolkit, including cuda_runtime.h, and ensure " + "nvcc can find its include roots, then refresh preflight.", + ) + if missing_header: + limitation = ( + "The CUDA toolkit is environment-blocked: cuda_runtime.h is missing from " + "nvcc's include path." + ) + return RequirementResult( + requirement="nvcc", + kind="executable", + required=required, + probe_kind="active", + status=status, + identity=str(path), + evidence=(str(path), diagnostic), + limitations=(limitation,), + remediation=remediation, + ) + + @staticmethod + def _bounded_diagnostic(value: str) -> str: + return " ".join(value.split())[:500] or "nvcc returned no diagnostic output." + def _distribution(self, name: str, *, required: bool) -> RequirementResult: try: requirement = Requirement(name) diff --git a/tests/application/test_capture_planning.py b/tests/application/test_capture_planning.py index 02204ae..3ed5bcd 100644 --- a/tests/application/test_capture_planning.py +++ b/tests/application/test_capture_planning.py @@ -1,6 +1,7 @@ from __future__ import annotations from pathlib import Path +from typing import Any import pytest @@ -17,11 +18,24 @@ CapabilityStatus, DomainError, ErrorCode, + ProcessResult, ) +from flameox.execution import ExecutionOutcome, ExecutionRequest, SubprocessBroker from flameox.storage import Workspace from tests.support.capture import write_workload +class _NvccProbeBroker(SubprocessBroker): + async def run(self, request: ExecutionRequest, **_: Any) -> ExecutionOutcome: + return ExecutionOutcome( + process=ProcessResult(exit_code=1, cleanup_complete=True), + stdout=b"", + stderr=b"fatal error: cuda_runtime.h: No such file or directory\n", + resolved_executable=Path(request.argv[0]), + containment="process_group", + ) + + def test_current_workload_definition_is_active_and_bound_to_plans(tmp_path: Path) -> None: workspace = Workspace.initialize(tmp_path) write_workload(tmp_path) @@ -205,6 +219,36 @@ async def probe(self, adapter: str, *, refresh: bool = False) -> CapabilityRepor assert result.requirements[0].probe_kind == "active" +@pytest.mark.anyio +async def test_active_nvcc_preflight_classifies_missing_cuda_headers( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = Workspace.initialize(tmp_path) + (tmp_path / "flameox.toml").write_text( + """ +schema_version = 1 +[workloads.gpu] +argv = ["python", "-c", "print('gpu')"] +[workloads.gpu.requirements] +executables = ["nvcc"] +""" + ) + monkeypatch.setattr("flameox.application.preflight.shutil.which", lambda _: "/usr/bin/nvcc") + + result = await PreflightService( + workspace, + broker=_NvccProbeBroker(), + ).inspect("gpu", mode="active") + + requirement = result.requirements[0] + assert result.disposition == "blocked" + assert requirement.status == "environment_blocked" + assert "cuda_runtime.h" in requirement.limitations[0] + assert "cuda_runtime.h" in requirement.remediation[0] + assert "cuda_runtime.h" in requirement.evidence[1] + + @pytest.mark.anyio async def test_required_preflight_failure_blocks_capture_planning(tmp_path: Path) -> None: workspace = Workspace.initialize(tmp_path) From b145dee99d432407365d421908a51ced14510f2c Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sun, 2 Aug 2026 07:32:21 +0800 Subject: [PATCH 3/4] feat(mcp): support explicit external workspaces --- docs/interfaces.md | 22 ++++++++++++++-------- docs/storage-and-evidence.md | 8 +++++--- src/flameox/cli.py | 18 ++++++++++++++++-- src/flameox/mcp/server.py | 33 ++++++++++++++++++++++++++++----- tests/mcp/test_workflows.py | 23 +++++++++++++++++++++++ 5 files changed, 86 insertions(+), 18 deletions(-) diff --git a/docs/interfaces.md b/docs/interfaces.md index 2c13462..855337c 100644 --- a/docs/interfaces.md +++ b/docs/interfaces.md @@ -33,7 +33,9 @@ the exact active executable and fixed launcher arguments. It is read-only and fails on configuration drift instead of repairing it implicitly. Client launchers contain the absolute path of a verified managed runtime and -the fixed arguments `mcp serve --project-root .`. Setup never passes `--init`. +the fixed arguments `mcp serve --project-root .`. Setup never passes `--init` or +an external workspace; hosts that need one can add `--workspace PATH` to an +explicit server invocation. Each MCP client therefore binds flameox's project root to the client's launch directory. Client setup is complete when the launcher is verified; each checkout still requires deliberate project initialization through `flameox init`, @@ -236,8 +238,8 @@ manifest. ### MCP ```text -flameox mcp serve [--init] -flameox mcp inspect +flameox mcp serve [--init] [--workspace PATH] +flameox mcp inspect [--workspace PATH] ``` `serve` uses stdio exclusively. `--init` performs the additive workspace @@ -386,11 +388,12 @@ execution explicitly. Planning still does not execute the workload; #### `initialize_workspace` -Additive and idempotent. Initializes only the MCP server's fixed project root. +Additive and idempotent. Initializes only the MCP server's fixed project root, +or the explicit workspace root selected when the server started. If the server already owns an initialized workspace, the call returns its current -status without replacing the detached-capture manager. It cannot select an -external path or configure workloads. Hosts may instead start the server using -`flameox mcp serve --init`. +status without replacing the detached-capture manager. It cannot switch to a +different path or configure workloads. Hosts may instead start the server using +`flameox mcp serve --init --workspace PATH`. #### `workload_configuration_status` @@ -459,7 +462,10 @@ request cleanup. The operation accepts only adapter names reported by `list_capabilities` as managed setup actions: `coverage`, `memray`, `perfetto`, `py-spy`, `pytest`, and `torch.profiler`. It installs the published FlameOx extra into the active managed Python runtime and stages the pinned user-space -Trace Processor under `.diagnostics/tools` for `perfetto`. Its receipt includes +Trace Processor under the active workspace's `tools/` directory for `perfetto`. +Setup progress uses `validating_request`, `installing_packages`, +`staging_trace_processor`, `verifying`, and `completed`; a staging failure keeps +the staging phase and bounded cause in its durable status. Its receipt includes the workspace identity, exact request digest, named phase, bounded progress, item outcomes, cancellation/cleanup state, and next recovery action. It never runs the declared workload, mutates source, installs arbitrary packages, diff --git a/docs/storage-and-evidence.md b/docs/storage-and-evidence.md index 794845e..e380b73 100644 --- a/docs/storage-and-evidence.md +++ b/docs/storage-and-evidence.md @@ -20,9 +20,11 @@ If none exists, commands that mutate state fail with a remediation suggesting An explicit `--workspace` overrides discovery. The resolved workspace must be inside the selected project root unless the user explicitly supplies an -external absolute path through the CLI. MCP tools cannot choose an arbitrary -external workspace, but artifact import explicitly permits the fixed project -root or the system temporary directory as bounded source roots. +external path through the CLI or MCP server startup. The project root remains +the workload and source root when an external workspace is selected. MCP tools +cannot switch workspace roots after startup, but artifact import explicitly +permits the fixed project root or the system temporary directory as bounded +source roots. In a Git repository, `flameox init` adds `.diagnostics/` to `.git/info/exclude` when it is not already ignored. It does not edit the diff --git a/src/flameox/cli.py b/src/flameox/cli.py index dacf5f6..95963f5 100644 --- a/src/flameox/cli.py +++ b/src/flameox/cli.py @@ -1921,9 +1921,16 @@ def mcp_serve( bool, typer.Option("--init", help="Initialize .diagnostics before protocol startup."), ] = False, + workspace: Annotated[ + Path | None, + typer.Option( + "--workspace", + help="Explicit workspace root; keeps the project root as the workload root.", + ), + ] = None, ) -> None: """Serve flameox over stdio; stdout is reserved for protocol messages.""" - run_server(project_root, initialize=initialize) + run_server(project_root, initialize=initialize, workspace_root=workspace) @mcp_app.command("inspect") @@ -1932,12 +1939,19 @@ def mcp_inspect( Path, typer.Option("--project-root", help="Fixed project root exposed to MCP."), ] = Path("."), + workspace: Annotated[ + Path | None, + typer.Option("--workspace", help="Explicit workspace root used by the inspected server."), + ] = None, json_output: JsonOption = False, ) -> None: """List the schemas and annotations exposed by the MCP adapter.""" async def inspect_server() -> dict[str, Any]: - async with Client(create_server(project_root), raise_exceptions=True) as client: + async with Client( + create_server(project_root, workspace_root=workspace), + raise_exceptions=True, + ) as client: tools = await client.list_tools() resources = await client.list_resource_templates() instructions = client.instructions diff --git a/src/flameox/mcp/server.py b/src/flameox/mcp/server.py index 9ddcfcd..f25c9ce 100644 --- a/src/flameox/mcp/server.py +++ b/src/flameox/mcp/server.py @@ -701,20 +701,31 @@ def create_server( project_root: Path, *, initialize: bool = False, + workspace_root: Path | None = None, ) -> StrictMCPServer[AppContext]: project_root = project_root.resolve() + if workspace_root is not None and "\x00" in str(workspace_root): + raise DomainError( + ErrorCode.WORKSPACE_INVALID, + "The MCP workspace root cannot contain NUL bytes.", + remediation=("Provide a valid local workspace directory path.",), + ) + selected_workspace_root = workspace_root.resolve() if workspace_root is not None else None lifespan_state: list[AppContext] = [] @asynccontextmanager async def lifespan(_: MCPServer[AppContext]) -> AsyncIterator[AppContext]: workspace: Workspace | None if initialize: - workspace = Workspace.initialize(project_root) + workspace = Workspace.initialize( + project_root, + workspace_root=selected_workspace_root, + ) else: try: workspace = Workspace.discover( project_root, - explicit=project_root / ".diagnostics", + explicit=selected_workspace_root, ) except DomainError: workspace = None @@ -804,7 +815,10 @@ async def initialize_workspace( result = workspace_status(state.workspace) return _success(result, f"Workspace is already initialized: {result.workspace_id}.") - workspace = Workspace.initialize(state.project_root) + workspace = Workspace.initialize( + state.project_root, + workspace_root=selected_workspace_root, + ) capture_plans = CapturePlanRegistry( max_parallel_captures=workspace.config.capture.max_parallel_captures ) @@ -2821,5 +2835,14 @@ def _active_state(states: list[AppContext]) -> AppContext: return states[0] -def run_server(project_root: Path, *, initialize: bool = False) -> None: - create_server(project_root, initialize=initialize).run() +def run_server( + project_root: Path, + *, + initialize: bool = False, + workspace_root: Path | None = None, +) -> None: + create_server( + project_root, + initialize=initialize, + workspace_root=workspace_root, + ).run() diff --git a/tests/mcp/test_workflows.py b/tests/mcp/test_workflows.py index db29360..198b134 100644 --- a/tests/mcp/test_workflows.py +++ b/tests/mcp/test_workflows.py @@ -131,6 +131,29 @@ async def test_cli_json_and_mcp_result_are_same_domain_model(tmp_path: Path) -> assert mcp.structured_content["result"] == expected +@pytest.mark.anyio +async def test_mcp_can_bind_an_explicit_external_workspace_root(tmp_path: Path) -> None: + project_root = tmp_path / "project" + workspace_root = tmp_path / "evidence" + project_root.mkdir() + + async with Client( + create_server( + project_root, + initialize=True, + workspace_root=workspace_root, + ), + raise_exceptions=True, + ) as client: + result = await client.call_tool("workspace_status", {}) + + assert result.is_error is False + assert result.structured_content is not None + assert result.structured_content["result"]["project_root"] == str(project_root.resolve()) + assert (workspace_root / "workspace.json").is_file() + assert not (project_root / ".diagnostics").exists() + + @pytest.mark.anyio async def test_mcp_inspect_instructions_match_initialize_metadata(tmp_path: Path) -> None: Workspace.initialize(tmp_path) From cc2bef185fe2a7d76cc6e94251b19a1c1ccd364a Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sun, 2 Aug 2026 12:11:31 +0800 Subject: [PATCH 4/4] test: refresh collection preservation baseline --- tests/collection-baseline.toml | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/tests/collection-baseline.toml b/tests/collection-baseline.toml index 58048cc..c98bd0a 100644 --- a/tests/collection-baseline.toml +++ b/tests/collection-baseline.toml @@ -1,4 +1,4 @@ -expected_test_count = 405 +expected_test_count = 411 # New files are normalized to their pre-split path so this receipt proves # test identity and parametrized case preservation across the decomposition. @@ -32,8 +32,8 @@ expected_test_count = 405 'tests/adapters/test_perfetto.py' = 'c4a12900de4584860227f61bc86b606fc78a76b0d533ea3aa8558dbfa422caf8' 'tests/analysis/test_recipes.py' = '87ea564890d57d16d9e6303089121662be325905ff4ee485d489c8efeb8d6d7e' 'tests/application/test_records_and_comparisons.py' = 'd052ad24e744c16d0058b1c01ab76a18d84bce0729d63cf824d5cd23046d0759' -'tests/application/test_workloads_and_capture.py' = '71dee8107660f1630ba0e35c5657be53c9f99a436edcad34f152ac7b996e04f9' -'tests/mcp/test_server.py' = '4ae328b622a03523015cf982bc0d7855c2797561871882b769f935e3d9ad3342' +'tests/application/test_workloads_and_capture.py' = 'c6abbca69ea11e525282bcd60e9121d6e037adf9f22960c4f802771a0de23778' +'tests/mcp/test_server.py' = '769fe7e151bacb8ef038d3f2ae0dbe94611e24ecdbd7f0a34f6a85d2fad21f59' 'tests/test_cli_setup.py' = '2e1c3a607b7b4cefecb44f98c88c2e65ccbc0afaae39201c2c443eef0204c561' [files] @@ -44,7 +44,7 @@ expected_test_count = 405 'tests/adapters/test_perfetto.py' = { count = 6, digest = 'f10a653fb41df80732bff21b5351b1f4eacfb30dcb358f3b690fb3dc5d3fe70f' } 'tests/adapters/test_perfetto_normalization.py' = { count = 1, digest = '1590a06a80ccf72735753b892d3d06ab654b42cf3c170630c5365493ddcccf52' } 'tests/adapters/test_pyperf.py' = { count = 4, digest = 'a0b5fdf06a752d5d375254b2f05a5f6962df9a1e917181695f13ce6f24923fb5' } -'tests/adapters/test_pytest.py' = { count = 6, digest = 'd7756e162c74611dd6c160c09586b40e4361157d3f021dc2497247a9bfed1a50' } +'tests/adapters/test_pytest.py' = { count = 7, digest = '61f4cda2ea988643405cb58f23cca3f7a08927a97bf5827a9a228196883dfbea' } 'tests/adapters/test_python_startup.py' = { count = 8, digest = '8cfb0e6e9eeb98fad169198876908f40b710c9d138284860869c9f0713fd2538' } 'tests/adapters/test_setup_runtime.py' = { count = 5, digest = '7d2b902ed64a32c461b2425a20c404583fbfd6d9996f57a1c621194620109b5b' } 'tests/analysis/test_comparison.py' = { count = 4, digest = '311406582a5fda58dbd1e74e6f7dff58cbe2bb760536031bb160f0c50ff08fa2' } @@ -53,7 +53,7 @@ expected_test_count = 405 'tests/application/test_analysis_records.py' = { count = 1, digest = '7a08286f866dbeef9c9a0500cead2b3ebf6ee6546f455ed3b5cc5407ee84b69f' } 'tests/application/test_artifact_service.py' = { count = 2, digest = '2ed111222920a42e8679d1e0281f185dc0d28406470ca888d80d0a01cd87731c' } 'tests/application/test_async_work.py' = { count = 1, digest = '5c21ab9d048e8676c0b086ff6859e87b28da82ab793c2b70de975f592355067e' } -'tests/application/test_capabilities.py' = { count = 12, digest = '15565af8e2a91a9cffabe3340007eaec10277710db69ef6f6a3a415b16555ef4' } +'tests/application/test_capabilities.py' = { count = 14, digest = '6a00720ca068f3d49560fa9b97ab2f48c85f22974ff5d1c3a68d4aa3e4d602b3' } 'tests/application/test_capture_native_outputs.py' = { count = 5, digest = '73ffb01416e94a3044feeecb6f11f1fb452f0b1595ecc8b3ac3ef4f749918f77' } 'tests/application/test_comparison_samples.py' = { count = 2, digest = 'f28179055d031cfb2d74b5f002c1c19278a64a46c5d0c02e32f91a4e9fef23df' } 'tests/application/test_detached.py' = { count = 8, digest = '25178dcfc15e9ce50eff6595814d67d223ab1550e4a0a6873862ec0e5414b0d7' } @@ -79,7 +79,7 @@ expected_test_count = 405 'tests/application/test_third_party_adapters.py' = { count = 7, digest = 'd169c20bdbb6f514887e75fe045f1a8950f946d915abc6d22d6836ac1959ab0c' } 'tests/application/test_viewers.py' = { count = 8, digest = '17e9b4ec485fff59f49c11bc9a513e28da97e4d6d0af6e945ba548ab72aef663' } 'tests/application/test_workloads.py' = { count = 4, digest = '8498d6450bad63aec49693d92b49d53555d8f134492f13f4fa8e1502adb9f758' } -'tests/application/test_workloads_and_capture.py' = { count = 46, digest = 'e46978fe067047aa9a06ce2685fbec0f84179e220ffe112f9b9dcdfb37dcc298' } +'tests/application/test_workloads_and_capture.py' = { count = 47, digest = 'c6abbca69ea11e525282bcd60e9121d6e037adf9f22960c4f802771a0de23778' } 'tests/application/test_operations.py' = { count = 5, digest = '2e31c1f386284e3c91f0dc6cbabd839c98e38aa2c46f52c175fd0629e9044c5c' } 'tests/domain/test_identity.py' = { count = 3, digest = '0dd4b644c18a6e8d07d56f9feb07c9af53da4791f465627d4c0f0fe6060a71e5' } 'tests/domain/test_models.py' = { count = 11, digest = '3c66f63c2bd90a4d7af1393bd59fe08072fe9872cf2a3c0d101095ba6280e016' } @@ -89,7 +89,8 @@ expected_test_count = 405 'tests/golden/test_memory_regression.py' = { count = 1, digest = '7898013cd64ec203e28d00b6549f002586a2d23f0c06eba6f1c979478f1aac1e' } 'tests/golden/test_reverse_scan.py' = { count = 1, digest = '11e8db62d67d86b3e2be63a8e3425dbd7927329f4d405a3d6718251286a20bea' } 'tests/mcp/test_agent_workflows.py' = { count = 6, digest = 'a2acbf5c684611b29f8b69aca1ab5f1f0743e36c32d807a7206826de1e481011' } -'tests/mcp/test_server.py' = { count = 37, digest = 'ab9f9d9f2b4fa37a7ebba360ce480f9cd81e2389e6e307ab035efe3ff9ae6e26' } +'tests/mcp/test_server.py' = { count = 38, digest = '769fe7e151bacb8ef038d3f2ae0dbe94611e24ecdbd7f0a34f6a85d2fad21f59' } +'tests/collectors/test_torch_launcher.py' = { count = 1, digest = 'b875d2f61ff086520d190513676e50fd1e6cf7152a9697e7b3c3d61d360736a3' } 'tests/performance/test_catalog_scale.py' = { count = 4, digest = '1c7587225525c2e0cc877bf62105c1d2447ec5d230e0a9d0eaef0ae7322e9214' } 'tests/security/test_offline.py' = { count = 1, digest = 'a9de75a583146be9fccb27627ae4fdd6d6f3265912dad9423fe97415ba833fd3' } 'tests/storage/test_artifacts_and_runs.py' = { count = 14, digest = '0f27557254a46c2635bf5cfd47766e05d84e3bb38becf8d8bf6271cd2ebbfc2e' }