diff --git a/docs/adapters.md b/docs/adapters.md index 519fe20..4526e7a 100644 --- a/docs/adapters.md +++ b/docs/adapters.md @@ -146,9 +146,12 @@ Nsight Compute capture requires both the `ncu` executable and NVIDIA's vendor-sh `extras/python/ncu_report.py` reader. Flameox resolves both before workload execution and returns external setup guidance if the installation is incomplete. -Provider ownership is explicit during capture. External collectors installed -with Flameox, such as py-spy, execute from the launched uvx environment rather -than ambient request `PATH`. In-process collectors, including coverage.py and +Provider ownership is explicit during capture. py-spy executes from a verified session binding +prepared independently with `uvx --from py-spy==0.4.2`, or from beside the server interpreter. +Preparation checks the collector version and binds its executable identity; later capture rechecks +that identity. A removed or changed uv cache entry requires preparation again. Capture never +silently installs dependencies or resolves managed collectors from request `PATH`. +In-process collectors, including coverage.py, PyTorch, and Memray, must be installed in the declared workload interpreter; capture probes that interpreter before execution and never substitutes Flameox's interpreter. diff --git a/docs/architecture.md b/docs/architecture.md index 5c23b1c..55fab72 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -64,9 +64,13 @@ capability selector or a free-form analysis argument object. Capture-provider contracts supply the discriminated provider variants for each compatible capture tool. Missing packages, executables, permissions, versions, or platforms do not change the catalog; the attempted tool returns typed remediation. The separately invoked CLI setup command or MCP -`prepare_providers` tool may prepare the exact version-pinned uvx environment named by an explicit -Python provider set; neither creates project state nor owns a durable operation or provider -inventory. Host profilers, drivers, and permissions remain external and receive guidance only. +`prepare_providers` tool resolves dependencies according to where they execute. CLI setup prepares +the complete version-pinned server environment. MCP preparation can activate a pinned standalone +py-spy collector in the existing session; server-import dependencies are checked against the active +release's complete requested dependency contract. Neither creates project state nor owns a durable +operation or provider inventory. Host profilers, drivers, and permissions remain external. +`providers/preparation.py` owns bounded preparation and verified session executable bindings; +transports do not infer readiness from installation receipts. Request validation is also a normalization boundary. Runtime admission constructs the typed capability and provider models once; execution, provenance, limitations, and result descriptions diff --git a/docs/interfaces.md b/docs/interfaces.md index 41d9dab..16ae29e 100644 --- a/docs/interfaces.md +++ b/docs/interfaces.md @@ -93,23 +93,29 @@ but never serialize arbitrary exception text: filesystem exceptions may contain dependency exceptions may contain argv or environment-derived values. Unexpected failures use a stable operation-specific summary; cancellation remains a control path and is re-raised. -`prepare_providers` and the capture tools are open-world. Preparation resolves the exact package -requirement through `uvx`, verifies that environment by running Flameox's version command, and -returns the same requirement in a global, version-pinned MCP launcher. Capture tools execute an -explicit caller-supplied target that may itself access external services. The request names the -complete managed -provider set; Flameox keeps no installed-provider inventory or setup receipt. System profilers, -drivers, device access, and OS permissions remain external requirements. Flameox returns guidance -for them but does not invoke a system package manager or elevate privileges. Preparation creates no -project state, -durable job, or plan, and it cannot add packages to the currently running server. A result with -a non-null `next_action` identifies the typed `reconnect_mcp` handoff, explains that the current -server is unchanged, and directs the agent to reconnect with the returned launcher before retrying -the capture. Host-only preparation returns `next_action: null`. Managed provider IDs are `aiperf`, -`memray`, `otlp`, `perfetto`, `py-spy`, and `torch`; every call declares the complete desired set. A -provider such as Perfetto may be both prepared Python support and an external host Trace Processor -requirement. Preparation waits up to 1,800 seconds by default; callers may set `timeout_seconds` -from 1 through 3,600. A uvx failure returns its complete stderr in `SETUP_FAILURE`. +`prepare_providers` and capture tools are open-world. Preparation uses request-owned bounded +subprocesses and an overall deadline; cancellation settles the installer and its descendants. +py-spy is prepared as a pinned standalone collector and becomes available in the same live session. +Bindings are activated only after the complete request succeeds, including any server preparation; +failure, timeout, or cancellation leaves prior session bindings unchanged. Preparation forwards +the safe uv controls `UV_OFFLINE`, `UV_CACHE_DIR`, `UV_PYTHON_DOWNLOADS`, and `UV_NO_CONFIG`. +Repeating preparation reuses its verified binding without installation. The returned launcher still +names the complete requested server provider set; preparing another provider does not remove +existing session bindings. No durable provider inventory, project state, or job is created. + +Server-import requirements are compared with the active release and installed dependency versions. +`activation_status` is `ready`, `restart_required`, `unknown`, or `not_applicable`. `next_action` is +null for ready or host-only requests. Otherwise `reconnect_mcp` explicitly says whether reconnection +is required or conditional and tells the caller to preserve needed session analyses first. An +unknown identity does not establish a required restart. A prepared server environment is verified +with its version command; preparation does not replace active imports. + +Managed IDs remain `aiperf`, `memray`, `otlp`, `perfetto`, `py-spy`, and `torch`. External host tools, +drivers, permissions, and workload-interpreter requirements are reported separately and remain +unverified by preparation. Both structured results and text summaries carry those handoffs. No +system package manager or privilege elevation is invoked. Perfetto still requires an externally +installed Trace Processor. The deadline defaults to 1,800 seconds and accepts 1 through 3,600; +MCP preparation failures use bounded path-free diagnostics, while CLI setup retains local stderr. ## Sources and limits @@ -117,15 +123,20 @@ The strict source union is: ```text PathSource {kind: "path", path, format?, producer?, expected_sha256?} -EvidenceSource {kind: "evidence", evidence_id, artifact_role?} +EvidenceSource {kind: "evidence", evidence_id, artifact_role? OR artifact_selector?} ``` Continuations are opaque integrity cursors bound to the request and exact input digests. They contain no authority, credentials, or artifact data and are not an authentication boundary: a caller already authorized to submit the analysis can choose which of its rows to request. They can cross process boundaries, so -a CLI invocation can resume a previous page. A changed input cannot reuse a -continuation. +a CLI invocation can resume a previous page. Tokens bind ordered content digests, formats, +producer identities, arguments, and limits, independently of storage paths and publication roles. +After preserving a capture, repeat the original options and limits with the evidence resource's +ordered `analysis_sources` and the returned continuation. Scratch can be released immediately. +A changed input cannot reuse a continuation. Tokens issued by older path-bound implementations +must be restarted with a fresh analysis. Preview `offset` counts logical rows: text lines, JSONL +records, CSV data records, Parquet records, and projected JSON entries. Decoded offsets must be integers within the available bounded population. Negative offsets and offsets at or beyond the end fail with `INVALID_INPUT`; they never use Python slicing semantics or @@ -174,6 +185,17 @@ eligible blocks and a deterministic percentile interval when at least three blocks survive capture/oracle validation, and classifies the effect against the declared threshold. Work is not detached; the request receives progress and owns cancellation. +The experiment's `point_estimate_classification` is descriptive; its `decision_basis` is explicit +on the metrics block. It does not claim confidence-qualified improvement or equivalence. + +Capture `outcome` is computed from every execution before diagnostic compaction and retains exact +success/failure counts. MCP error classification consumes that outcome even when no execution +diagnostics fit inline. Each execution identifies whether `returncode` belongs to the workload or +collector, retains the invoked executable SHA-256, and leaves `workload_returncode` null for wrapped +captures. Exit ownership is declared by each invocation builder: self-reporting workloads retain +their observed exit even when they use a provider other than `direct`. A usable profile does not +prove workload success. Preserved stdout, stderr, and profiles +are individually selectable from the evidence resource. The first declared case is the baseline. A case inherits the target argv when it omits `argv`, and its environment overrides the target environment. Each block randomizes case order from the declared seed. The semantic oracle runs after every successful capture in that case environment; @@ -205,7 +227,7 @@ flameox setup flameox mcp serve|inspect flameox analyze [--continuation TOKEN] [--preserve] flameox capture [--experiment JSON] [--preserve] -- -flameox evidence query|show +flameox evidence query|show|location ``` `setup` detects supported coding agents and uses one multi-select prompt to choose which global MCP diff --git a/docs/investigations.md b/docs/investigations.md index eac9d96..62575c2 100644 --- a/docs/investigations.md +++ b/docs/investigations.md @@ -52,8 +52,12 @@ The 0.2 runtime accepts `wall_time_ns` and paired `median_difference` or `mean_difference`. Each non-baseline case is compared with the first declared case within the same blocks. Failed or oracle-invalid pairs are excluded and reported as limitations; fewer than three eligible pairs produce a descriptive -estimate without a confidence interval. The practical-threshold decision is -returned as typed comparison evidence, not retained only as request metadata. +estimate without a confidence interval. `point_estimate_classification` describes only the +observed estimate against the practical margin, with `decision_basis=descriptive_point_estimate` +on the experiment metrics block. It replaces the ambiguous `decision` field. `within_threshold` +does not establish equivalence; a wide interval may still span material improvement and regression. +The deterministic percentile interval is reported separately and is not a calibrated equivalence +test. Semantic correctness still requires the declared oracle. Artifact comparison is separate from that experiment result. Capture the representative baseline and candidate summaries independently, preserve them if they must survive the session, then submit diff --git a/docs/storage-and-evidence.md b/docs/storage-and-evidence.md index 702031c..c8a001a 100644 --- a/docs/storage-and-evidence.md +++ b/docs/storage-and-evidence.md @@ -95,6 +95,25 @@ Missing or corrupt resources are MCP resource errors. Native payload bytes are not exposed through resources; their digest and role remain visible in the manifest. +Agent projections expose an opaque `source` accepted unchanged by analysis tools for each artifact, +plus `logical_sources` for directory bundles and ordered `analysis_sources` for the original +analysis. Selectors address immutable manifest positions, not hashes of private roles that could +be checked against guessed filenames. File selectors always select exact members, even when a +filename contains the directory-role delimiter. New manifests include `source_layout`: each source +declares its file/directory kind, exact artifact indices, and identity, with an ordered mapping for +the original analysis inputs. Empty directories retain their metadata without inventing a native +payload, and re-preserving a selected member retains its file identity. Readers validate membership, +digests, sizes, and analysis mappings. Existing manifests without this optional field remain +readable using their original role-based bundle convention. Missing or +ambiguous selectors point back to the evidence resource for enumeration. + +Corruption remains fail-closed. Runtime errors include the selected configuration source, a +path-free store identifier, and recovery instructions. `flameox evidence location` prints the +resolved directory locally without reading or initializing the repository. Restore the original +store from a known-good backup, or select a distinct empty store with `FLAMEOX_DATA_DIR` and +restart/reconnect. Switching stores does not recover old evidence. Do not delete existing data or +synthesize replacement metadata. + ## Format evolution This is repository format `1`. Unsupported repository or manifest versions diff --git a/src/flameox/cli.py b/src/flameox/cli.py index b581466..4ee9cd6 100644 --- a/src/flameox/cli.py +++ b/src/flameox/cli.py @@ -426,6 +426,18 @@ def evidence_query( runtime.close() +@evidence_app.command("location") +def evidence_location() -> None: + """Show the selected local evidence directory without opening or modifying it.""" + runtime = _runtime() + try: + _write( + {"directory": str(runtime.repository.root), "environment_variable": "FLAMEOX_DATA_DIR"} + ) + finally: + runtime.close() + + @evidence_app.command("show") def evidence_show( evidence_id: Annotated[str, typer.Argument()], diff --git a/src/flameox/execution.py b/src/flameox/execution.py index f0fd131..c4f9472 100644 --- a/src/flameox/execution.py +++ b/src/flameox/execution.py @@ -48,6 +48,10 @@ "UV_EXTRA_INDEX_URL", "UV_INDEX", "UV_NATIVE_TLS", + "UV_OFFLINE", + "UV_CACHE_DIR", + "UV_PYTHON_DOWNLOADS", + "UV_NO_CONFIG", "PIP_INDEX_URL", "PIP_EXTRA_INDEX_URL", ) diff --git a/src/flameox/mcp/server.py b/src/flameox/mcp/server.py index 5b0e75e..a184f9c 100644 --- a/src/flameox/mcp/server.py +++ b/src/flameox/mcp/server.py @@ -11,7 +11,6 @@ from pathlib import Path from typing import Annotated, Any, Literal, cast -import anyio from mcp.server import MCPServer from mcp.server.mcpserver import Context from mcp_types import CallToolResult, ContentBlock, ResourceLink, TextContent, ToolAnnotations @@ -25,7 +24,7 @@ ) from pydantic.json_schema import JsonSchemaValue -import flameox.setup as provider_setup +import flameox.providers.environment as provider_setup from flameox import __version__ from flameox.mcp.capability_tools import ( Execution, @@ -119,7 +118,10 @@ class LauncherEnvelope(BaseModel): class ReconnectActionEnvelope(BaseModel): kind: Literal["reconnect_mcp"] = Field(description="Reconnect the MCP server process.") - message: str = Field(description="Required handoff before retrying with the prepared provider.") + message: str = Field( + description="Handoff conditions and session-evidence preservation guidance." + ) + necessity: Literal["required", "conditional"] class PreparationEnvelope(_Envelope): @@ -139,7 +141,13 @@ class PreparationEnvelope(_Envelope): description="Version-pinned launcher for the requested provider set." ) next_action: ReconnectActionEnvelope | None = Field( - description="Required reconnection action, or null when the current server can continue." + description=( + "Reconnection handoff with explicit necessity; null for ready or host-only preparation." + ) + ) + activation_status: Literal["ready", "restart_required", "unknown", "not_applicable"] + workload_requirements: list[ExternalRequirementEnvelope] = Field( + description="Requirements to verify in the exact workload environment before capture." ) @@ -223,6 +231,21 @@ def _success_summary(value: dict[str, Any], *, resource: ResourceLink | None) -> action = ( "reconnect using the returned launcher" if preparation_action else "continue capture" ) + if ( + isinstance(preparation_action, dict) + and preparation_action.get("necessity") == "conditional" + ): + action = ( + "verify the active environment; reconnect only if the requested support is absent" + ) + if value.get("external_requirements"): + action = ( + (action + "; " if preparation_action else "") + + "verify or satisfy the listed external requirements before capture; " + "host readiness has not been verified" + ) + if value.get("workload_requirements"): + action += "; verify the listed requirements in the exact workload interpreter" return ( f"Provider preparation completed; next: {action}. " "Full details are in structuredContent." @@ -297,6 +320,7 @@ async def lifespan(_: MCPServer[AnalysisRuntime]) -> AsyncIterator[AnalysisRunti @server.tool(annotations=PREPARE, structured_output=True) async def prepare_providers( + ctx: Context[AnalysisRuntime], provider_ids: Annotated[ list[str], Field( @@ -317,21 +341,25 @@ async def prepare_providers( """Prepare managed providers and return any required MCP reconnection action.""" try: - preparation = await anyio.to_thread.run_sync( - provider_setup.prepare_providers, provider_ids, timeout_seconds - ) + preparation = await runtime(ctx).dependencies.prepare(provider_ids, timeout_seconds) except provider_setup.ProviderSelectionFailure as error: return _failure(RuntimeFailure("INVALID_INPUT", str(error))) except provider_setup.SetupFailure as error: return _failure(RuntimeFailure("SETUP_FAILURE", str(error))) next_action = None - if preparation.restart_required: + if preparation.restart_required is not False: next_action = { "kind": "reconnect_mcp", + "necessity": "required" if preparation.restart_required else "conditional", "message": ( - "Reconnect Flameox with the returned launcher before retrying the capture; " - "the current server process is unchanged." + "The active server does not satisfy the requested dependency contract. " + if preparation.restart_required + else "The active server dependency identity could not be verified. " + ) + + ( + "Preserve needed session analyses before reconnecting with the complete " + "returned launcher. Reconnection ends session-local evidence access." ), } return _success( @@ -351,6 +379,11 @@ async def prepare_providers( "args": preparation.launcher_args, }, "next_action": next_action, + "activation_status": preparation.activation_status, + "workload_requirements": [ + {"provider_id": item.provider_id, "guidance": item.guidance} + for item in preparation.workload_requirements + ], } ) @@ -485,11 +518,12 @@ async def progress(current: int, total: int, message: str) -> None: failed = [ item for item in value["capture"]["executions"] if item["status"] != "succeeded" ] - if failed: + if value["capture"]["outcome"]["status"] != "succeeded": return _failure( RuntimeFailure( "EXECUTION_FAILURE", - "One or more captured targets exited unsuccessfully.", + "One or more capture executions failed; consult exit attribution " + "and preserved diagnostics before inferring workload failure.", details={"partial_evidence": value, "failed_executions": failed}, ) ) diff --git a/src/flameox/providers/availability.py b/src/flameox/providers/availability.py index 3d497d3..a27bbff 100644 --- a/src/flameox/providers/availability.py +++ b/src/flameox/providers/availability.py @@ -25,4 +25,12 @@ WORKLOAD_PYTHON_REQUIREMENTS = { "coverage": ("coverage", "coverage", ">=7.14,<8"), "memray": ("memray", "memray", ">=1.17"), + "torch-profiler": ("torch", "torch", ">=2.7"), +} + +WORKLOAD_PROVIDER_GUIDANCE = { + "memray": "Install compatible Memray in the exact workload Python interpreter; " + "server preparation only supplies the analysis reader.", + "torch": "Install compatible PyTorch in the exact workload Python interpreter and " + "verify the requested CPU or accelerator activity is supported.", } diff --git a/src/flameox/providers/capture.py b/src/flameox/providers/capture.py index 8767a37..19d2e33 100644 --- a/src/flameox/providers/capture.py +++ b/src/flameox/providers/capture.py @@ -8,7 +8,7 @@ from collections.abc import Callable from dataclasses import dataclass from pathlib import Path -from typing import cast +from typing import Literal, cast from flameox.runtime_contracts import ( BenchmarkSamplesCaptureArguments, @@ -34,6 +34,8 @@ class CaptureInvocation: argv: tuple[str, ...] environment: dict[str, str] artifacts: tuple[tuple[Path, str, str], ...] + # Wrapper processes cannot prove the workload's exit without a separate receipt. + returncode_scope: Literal["workload", "collector"] = "collector" @dataclass(frozen=True, slots=True) @@ -56,7 +58,7 @@ def _arguments(request: CaptureBuildRequest, expected: type[CaptureArguments]) - def _direct(request: CaptureBuildRequest, _: ManagedExecutable) -> CaptureInvocation: _arguments(request, EmptyArguments) - return CaptureInvocation(tuple(request.target_argv), request.environment, ()) + return CaptureInvocation(tuple(request.target_argv), request.environment, (), "workload") def _pyperf(request: CaptureBuildRequest, _: ManagedExecutable) -> CaptureInvocation: @@ -142,7 +144,9 @@ def _node_cpu(request: CaptureBuildRequest, _: ManagedExecutable) -> CaptureInvo f"--cpu-prof-name={output.name}", *request.target_argv[1:], ) - return CaptureInvocation(argv, request.environment, ((output, "cpuprofile", "cpu-profile"),)) + return CaptureInvocation( + argv, request.environment, ((output, "cpuprofile", "cpu-profile"),), "workload" + ) def _node_heap(request: CaptureBuildRequest, _: ManagedExecutable) -> CaptureInvocation: @@ -156,7 +160,7 @@ def _node_heap(request: CaptureBuildRequest, _: ManagedExecutable) -> CaptureInv *request.target_argv[1:], ) return CaptureInvocation( - argv, request.environment, ((output, "heapprofile", "memory-profile"),) + argv, request.environment, ((output, "heapprofile", "memory-profile"),), "workload" ) @@ -174,7 +178,7 @@ def _benchmark_samples(request: CaptureBuildRequest, _: ManagedExecutable) -> Ca sort_keys=True, ) return CaptureInvocation( - tuple(request.target_argv), environment, ((output, "samples", "benchmark"),) + tuple(request.target_argv), environment, ((output, "samples", "benchmark"),), "workload" ) @@ -186,6 +190,7 @@ def _nvbench(request: CaptureBuildRequest, _: ManagedExecutable) -> CaptureInvoc (*request.target_argv, "--jsonbin", str(output)), request.environment, ((output_root, "nvbench", "benchmark"),), + "workload", ) @@ -217,7 +222,7 @@ def _torch_profiler(request: CaptureBuildRequest, _: ManagedExecutable) -> Captu "FLAMEOX_TORCH_PROFILER_OUTPUT_ROOT": str(output_root), } return CaptureInvocation( - tuple(request.target_argv), environment, ((output, "pytorch", "trace"),) + tuple(request.target_argv), environment, ((output, "pytorch", "trace"),), "workload" ) @@ -390,7 +395,10 @@ def _observations(request: CaptureBuildRequest, _: ManagedExecutable) -> Capture output = request.directory / "observations.jsonl" environment = {**request.environment, "FLAMEOX_OBSERVATIONS_PATH": str(output)} return CaptureInvocation( - tuple(request.target_argv), environment, ((output, "observations", "observations"),) + tuple(request.target_argv), + environment, + ((output, "observations", "observations"),), + "workload", ) diff --git a/src/flameox/providers/environment.py b/src/flameox/providers/environment.py new file mode 100644 index 0000000..0e41d8f --- /dev/null +++ b/src/flameox/providers/environment.py @@ -0,0 +1,128 @@ +"""Provider environment contracts shared by CLI setup and live preparation.""" + +from __future__ import annotations + +import importlib.metadata +from dataclasses import dataclass +from typing import Literal + +from packaging.requirements import Requirement + +from flameox import __version__ +from flameox.providers.availability import ( + MANAGED_PROVIDER_EXTRAS, + SYSTEM_PROVIDER_GUIDANCE, + WORKLOAD_PROVIDER_GUIDANCE, +) + +DEFAULT_PREPARATION_TIMEOUT_SECONDS = 1_800 +MAX_PREPARATION_TIMEOUT_SECONDS = 3_600 + + +class SetupFailure(RuntimeError): + pass + + +class ProviderSelectionFailure(SetupFailure): + pass + + +@dataclass(frozen=True, slots=True) +class ExternalRequirement: + provider_id: str + guidance: str + + +@dataclass(frozen=True, slots=True) +class ProviderPreparation: + requested_providers: list[str] + prepared_managed_providers: list[str] + external_requirements: list[ExternalRequirement] + preparation_command: list[str] + launcher_command: str + launcher_args: list[str] + activation_status: Literal["ready", "restart_required", "unknown", "not_applicable"] = "unknown" + + @property + def preparation_status(self) -> Literal["prepared", "not_applicable"]: + return "prepared" if self.prepared_managed_providers else "not_applicable" + + @property + def restart_required(self) -> bool | None: + if not self.prepared_managed_providers or self.activation_status in { + "ready", + "not_applicable", + }: + return False + return True if self.activation_status == "restart_required" else None + + @property + def workload_requirements(self) -> list[ExternalRequirement]: + return [ + ExternalRequirement(provider, WORKLOAD_PROVIDER_GUIDANCE[provider]) + for provider in self.requested_providers + if provider in WORKLOAD_PROVIDER_GUIDANCE + ] + + +def active_provider_status(providers: list[str]) -> Literal["ready", "restart_required", "unknown"]: + """Compare the active distribution against the complete requested release contract.""" + try: + distribution = importlib.metadata.distribution("flameox") + if distribution.version != __version__: + return "unknown" + extras = {MANAGED_PROVIDER_EXTRAS[item] for item in providers} + requirements = [Requirement(item) for item in distribution.requires or []] + for requirement in requirements: + if requirement.marker is not None and not any( + requirement.marker.evaluate({"extra": extra}) for extra in extras | {""} + ): + continue + try: + version = importlib.metadata.version(requirement.name) + except importlib.metadata.PackageNotFoundError: + return "restart_required" + if not requirement.specifier.contains(version, prereleases=True): + return "restart_required" + return "ready" if requirements else "unknown" + except (importlib.metadata.PackageNotFoundError, ValueError): + return "unknown" + + +def _validate_providers(providers: list[str]) -> None: + unknown = sorted( + set(providers).difference(MANAGED_PROVIDER_EXTRAS).difference(SYSTEM_PROVIDER_GUIDANCE) + ) + if unknown: + supported = ", ".join(sorted(MANAGED_PROVIDER_EXTRAS | SYSTEM_PROVIDER_GUIDANCE)) + raise ProviderSelectionFailure( + f"Unknown provider {unknown[0]!r}; choose one of: {supported}" + ) + + +def mcp_launcher(providers: list[str]) -> tuple[str, list[str]]: + """Return a version-bound MCP launcher for client configuration.""" + + _validate_providers(providers) + extras = sorted( + { + MANAGED_PROVIDER_EXTRAS[provider] + for provider in providers + if provider in MANAGED_PROVIDER_EXTRAS + } + ) + extras_suffix = f"[{','.join(extras)}]" if extras else "" + requirement = f"flameox{extras_suffix}=={__version__}" + return ( + "uvx", + ["--python", "3.12", "--from", requirement, "flameox"], + ) + + +def external_provider_requirements(providers: list[str]) -> list[ExternalRequirement]: + _validate_providers(providers) + return [ + ExternalRequirement(provider, SYSTEM_PROVIDER_GUIDANCE[provider]) + for provider in dict.fromkeys(providers) + if provider in SYSTEM_PROVIDER_GUIDANCE + ] diff --git a/src/flameox/providers/preparation.py b/src/flameox/providers/preparation.py new file mode 100644 index 0000000..038e323 --- /dev/null +++ b/src/flameox/providers/preparation.py @@ -0,0 +1,205 @@ +"""Request-owned dependency preparation and session-local collector activation.""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys +from contextlib import suppress +from pathlib import Path + +import anyio + +from flameox import __version__ +from flameox.command_binding import ExecutableResolver +from flameox.executable_models import ResolvedExecutable +from flameox.execution import ( + INSTALLER_ENVIRONMENT_ALLOWLIST, + ExecutionRequest, + ProcessExecutionError, + SubprocessBroker, +) +from flameox.providers.availability import MANAGED_PROVIDER_EXTRAS +from flameox.providers.environment import ( + DEFAULT_PREPARATION_TIMEOUT_SECONDS, + MAX_PREPARATION_TIMEOUT_SECONDS, + ProviderPreparation, + SetupFailure, + active_provider_status, + external_provider_requirements, + mcp_launcher, +) +from flameox.runtime_contracts import RuntimeFailure +from flameox.runtime_errors import DomainError + +# Keep the standalone collector aligned with the tested version in uv.lock. +PY_SPY_VERSION = "0.4.2" + + +class ProviderDependencies: + """Own verified executable bindings, never an installed-provider inventory.""" + + def __init__(self, broker: SubprocessBroker, scratch: Path) -> None: + self.broker = broker + self.scratch = scratch + self._py_spy: ResolvedExecutable | None = None + + def py_spy_executable(self) -> str | None: + if self._py_spy is None: + return None + try: + current = self._bind(str(self._py_spy.invocation_path)) + except (DomainError, OSError) as error: + raise self._expired_collector() from error + if current.identity != self._py_spy.identity: + raise self._expired_collector() + return str(current.invocation_path) + + @staticmethod + def _expired_collector() -> RuntimeFailure: + return RuntimeFailure( + "UNAVAILABLE_CAPABILITY", + "The prepared collector is missing or changed; prepare py-spy again before capture.", + details={"preparation_tool": "prepare_providers", "provider_ids": ["py-spy"]}, + ) + + def verify_capture_binding(self, provider_id: str, binding: ResolvedExecutable) -> None: + if ( + provider_id == "py-spy" + and self._py_spy is not None + and ( + binding.identity != self._py_spy.identity + or binding.invocation_path != self._py_spy.invocation_path + ) + ): + raise self._expired_collector() + + def _bind(self, executable: str) -> ResolvedExecutable: + return ExecutableResolver().require_host_tool( + executable, cwd=self.scratch, environment=dict(os.environ) + ) + + async def _run(self, argv: list[str], timeout: int) -> bytes: + try: + request = ExecutionRequest( + argv=tuple(argv), + executable_binding=self._bind(argv[0]), + cwd=self.scratch, + allowed_working_roots=(self.scratch,), + environment_allowlist=INSTALLER_ENVIRONMENT_ALLOWLIST, + timeout_seconds=timeout, + max_output_bytes=256 * 1024, + ) + task = asyncio.create_task(self.broker.run(request)) + try: + result = await asyncio.shield(task) + except asyncio.CancelledError: + task.cancel() + # AnyIO cancellation is level-triggered. Let the broker settle its + # readers and descendants under one cancellation before unwinding. + with anyio.CancelScope(shield=True), suppress(asyncio.CancelledError): + await task + raise + except (DomainError, ProcessExecutionError, OSError) as error: + raise SetupFailure( + "Provider preparation could not complete; verify uvx availability, package " + "index access, platform support, and the preparation timeout." + ) from error + if getattr(result.process.termination, "exit_code", None) != 0: + raise SetupFailure( + "Provider preparation failed; verify package availability and platform support." + ) + return result.stdout + + async def _prepare_py_spy(self, timeout: int) -> tuple[ResolvedExecutable, list[str]]: + try: + if self.py_spy_executable() is not None: + assert self._py_spy is not None + return self._py_spy, [] + except RuntimeFailure: + pass # This explicit preparation request authorizes resolving a new binding. + adjacent = Path(sys.executable).with_name("py-spy.exe" if os.name == "nt" else "py-spy") + if adjacent.is_file(): + output = await self._run([str(adjacent), "--version"], min(timeout, 10)) + if output.decode().strip() == f"py-spy {PY_SPY_VERSION}": + return self._bind(str(adjacent)), [] + probe = ( + "import importlib.metadata,json,os,sysconfig; " + "from pathlib import Path; " + "print(json.dumps({'version':importlib.metadata.version('py-spy')," + "'executable':str(Path(sysconfig.get_path('scripts')) / " + "('py-spy.exe' if os.name == 'nt' else 'py-spy'))}))" + ) + command = [ + "uvx", + "--isolated", + "--python", + "3.12", + "--from", + f"py-spy=={PY_SPY_VERSION}", + "python", + "-c", + probe, + ] + output = await self._run(command, timeout) + try: + receipt = json.loads(output) + if receipt["version"] != PY_SPY_VERSION: + raise ValueError("version mismatch") + path = Path(receipt["executable"]) + if not path.is_absolute(): + raise ValueError("non-absolute collector") + binding = self._bind(str(path)) + version = await self._run([str(path), "--version"], min(timeout, 10)) + if version.decode().strip() != f"py-spy {PY_SPY_VERSION}": + raise ValueError("collector version mismatch") + except (KeyError, TypeError, ValueError, DomainError, OSError) as error: + raise SetupFailure("Prepared collector identity could not be verified.") from error + return binding, command + + async def prepare( + self, + providers: list[str], + timeout_seconds: int = DEFAULT_PREPARATION_TIMEOUT_SECONDS, + ) -> ProviderPreparation: + if not 1 <= timeout_seconds <= MAX_PREPARATION_TIMEOUT_SECONDS: + raise SetupFailure("Invalid provider preparation timeout.") + try: + with anyio.fail_after(timeout_seconds): + return await self._prepare(providers, timeout_seconds) + except TimeoutError as error: + raise SetupFailure("Provider preparation exceeded its overall timeout.") from error + + async def _prepare(self, providers: list[str], timeout_seconds: int) -> ProviderPreparation: + requested = list(dict.fromkeys(providers)) + launcher_command, launcher_args = mcp_launcher(requested) + managed = [item for item in requested if item in MANAGED_PROVIDER_EXTRAS] + command: list[str] = [] + collector: ResolvedExecutable | None = None + if "py-spy" in managed: + collector, command = await self._prepare_py_spy(timeout_seconds) + server_providers = [item for item in managed if item != "py-spy"] + activation = active_provider_status(server_providers) if server_providers else "ready" + if server_providers and activation != "ready": + # The launcher still names the complete desired server environment. Preparing + # it neither changes active imports nor discards unrelated collector bindings. + command = [launcher_command, *launcher_args, "--version"] + output = await self._run(command, timeout_seconds) + if output.decode().strip() != __version__: + raise SetupFailure("Prepared Flameox release does not match the running release.") + preparation = ProviderPreparation( + requested, + managed, + external_provider_requirements(requested), + command, + launcher_command, + [*launcher_args, "mcp", "serve"], + activation if managed else "not_applicable", + ) + # Publish session state only after every requested preparation succeeds. + # No await follows this commit, and failed requests never roll back another + # concurrent request's successful binding. + if collector is not None: + self._py_spy = collector + return preparation diff --git a/src/flameox/repository.py b/src/flameox/repository.py index 5dd6e5e..2ba2ad2 100644 --- a/src/flameox/repository.py +++ b/src/flameox/repository.py @@ -34,6 +34,19 @@ def __init__(self, code: str, message: str) -> None: self.message = message +def artifact_selector(evidence_id: str, collection: str, index: int) -> str: + """Address an immutable public position, never a guessable private filename.""" + return hashlib.sha256(canonical_bytes([evidence_id, collection, index])).hexdigest() + + +def evidence_source(evidence_id: str, collection: str, index: int) -> dict[str, str]: + return { + "kind": "evidence", + "evidence_id": evidence_id, + "artifact_selector": artifact_selector(evidence_id, collection, index), + } + + @dataclass(frozen=True, slots=True) class NativeArtifact: path: Path @@ -197,6 +210,7 @@ def read_agent_projection(self, evidence_id: str) -> dict[str, Any]: manifest = self.read(evidence_id) body = manifest["body"] + logical_artifacts = self.logical_artifacts(body) capture_request = body["capture_request"] safe_capture: dict[str, Any] | None = None if capture_request is not None: @@ -232,6 +246,12 @@ def read_agent_projection(self, evidence_id: str) -> dict[str, Any]: return { "format_version": manifest["format_version"], "evidence_id": manifest["evidence_id"], + "analysis_sources": self._analysis_sources(manifest, logical_artifacts), + "logical_sources": [ + {key: item[key] for key in ("sha256", "size_bytes", "format")} + | {"source": evidence_source(evidence_id, "logical", index)} + for index, item in enumerate(logical_artifacts) + ], "body": { "evidence_kind": body["evidence_kind"], "capability_id": body["capability_id"], @@ -250,7 +270,8 @@ def read_agent_projection(self, evidence_id: str) -> dict[str, Any]: }, "artifacts": [ {key: item[key] for key in ("sha256", "size_bytes", "format")} - for item in body["artifacts"] + | {"source": evidence_source(evidence_id, "artifact", index)} + for index, item in enumerate(body["artifacts"]) ], "data_files": [ {key: item[key] for key in ("sha256", "size_bytes", "media_type")} @@ -259,6 +280,66 @@ def read_agent_projection(self, evidence_id: str) -> dict[str, Any]: }, } + @staticmethod + def logical_artifacts(body: Mapping[str, Any]) -> list[dict[str, Any]]: + if "source_layout" in body: + return cast(list[dict[str, Any]], body["source_layout"]["sources"]) + artifacts = body["artifacts"] + logical = [item | {"is_directory": False} for item in artifacts if ":" not in item["role"]] + bundles: dict[str, list[dict[str, Any]]] = {} + for artifact in artifacts: + if ":" in artifact["role"]: + bundles.setdefault(artifact["role"].split(":", 1)[0], []).append(artifact) + for role, members in bundles.items(): + digest = hashlib.sha256() + for member in sorted(members, key=lambda item: Path(item["role"].split(":", 1)[1])): + relative = member["role"].split(":", 1)[1] + digest.update(relative.encode() + bytes.fromhex(member["sha256"])) + logical.append( + { + "role": role, + "sha256": digest.hexdigest(), + "size_bytes": sum(item["size_bytes"] for item in members), + "format": members[0]["format"], + "producer": members[0].get("producer"), + "is_directory": True, + "artifact_indices": [ + index for index, item in enumerate(artifacts) if item in members + ], + } + ) + return logical + + @staticmethod + def _analysis_sources( + manifest: Mapping[str, Any], + logical_artifacts: list[dict[str, Any]], + ) -> list[dict[str, str]]: + body = manifest["body"] + if "source_layout" in body: + return [ + evidence_source(manifest["evidence_id"], "logical", index) + for index in body["source_layout"]["analysis_sources"] + ] + sources = [] + for item in body["analysis_request"].get("inputs", []): + matching = [ + (index, artifact) + for index, artifact in enumerate(logical_artifacts) + if artifact["sha256"] == item["sha256"] + and artifact["format"] == item["format"] + and artifact.get("producer") == item.get("producer") + ] + if not matching: + return [] + # Preserve identity when distinct sources happen to have identical bytes. + index, _ = next( + (match for match in matching if match[1]["role"] == item.get("role")), + matching[0], + ) + sources.append(evidence_source(manifest["evidence_id"], "logical", index)) + return sources + @staticmethod def _safe_execution_projection(execution: Mapping[str, Any]) -> dict[str, Any]: oracle = execution.get("semantic_oracle") @@ -270,6 +351,9 @@ def _safe_execution_projection(execution: Mapping[str, Any]) -> dict[str, Any]: for key in ( "block", "returncode", + "returncode_scope", + "executable_sha256", + "workload_returncode", "status", "failure_code", "missing_artifact_roles", @@ -641,7 +725,7 @@ def _validate_manifest_body(body: Mapping[str, Any]) -> None: "artifacts", "data_files", } - if set(body) != expected_keys: + if set(body) not in (expected_keys, expected_keys | {"source_layout"}): raise RepositoryError( "REPOSITORY_CORRUPTION", "Evidence manifest body has an invalid shape." ) @@ -721,6 +805,121 @@ def _validate_manifest_body(body: Mapping[str, Any]) -> None: artifacts = body["artifacts"] if not isinstance(artifacts, list): raise RepositoryError("REPOSITORY_CORRUPTION", "Evidence artifacts are invalid.") + if "source_layout" in body: + EvidenceRepository._validate_source_layout(body) + + @staticmethod + def _validate_source_layout(body: Mapping[str, Any]) -> None: + layout = body["source_layout"] + if ( + not isinstance(layout, dict) + or set(layout) != {"sources", "analysis_sources"} + or not isinstance(layout["sources"], list) + or not isinstance(layout["analysis_sources"], list) + ): + raise RepositoryError("REPOSITORY_CORRUPTION", "Evidence source layout is invalid.") + artifacts = body["artifacts"] + used: set[int] = set() + roles: set[str] = set() + for item in layout["sources"]: + if ( + not isinstance(item, dict) + or set(item) + != { + "role", + "sha256", + "size_bytes", + "format", + "producer", + "is_directory", + "artifact_indices", + } + or not isinstance(item["role"], str) + or not item["role"] + or item["role"] in roles + or not _is_digest(item["sha256"]) + or type(item["size_bytes"]) is not int + or item["size_bytes"] < 0 + or not isinstance(item["format"], str) + or not item["format"] + or (item["producer"] is not None and not isinstance(item["producer"], str)) + or type(item["is_directory"]) is not bool + or not isinstance(item["artifact_indices"], list) + ): + raise RepositoryError("REPOSITORY_CORRUPTION", "Evidence source layout is invalid.") + roles.add(item["role"]) + members = [] + for index in item["artifact_indices"]: + if type(index) is not int or not 0 <= index < len(artifacts) or index in used: + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence membership is invalid." + ) + used.add(index) + member = artifacts[index] + if ( + not isinstance(member, dict) + or not isinstance(member.get("role"), str) + or not _is_digest(member.get("sha256")) + or type(member.get("size_bytes")) is not int + or member["size_bytes"] < 0 + or member.get("format") != item["format"] + or member.get("producer") != item["producer"] + ): + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence membership is invalid." + ) + members.append(member) + if item["is_directory"]: + prefix = item["role"] + ":" + digest = hashlib.sha256() + for member in sorted( + members, key=lambda member: Path(member["role"][len(prefix) :]) + ): + relative = member["role"][len(prefix) :] + path = Path(relative) + if ( + not member["role"].startswith(prefix) + or path.is_absolute() + or ".." in path.parts + or not path.parts + ): + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence membership is invalid." + ) + digest.update(relative.encode() + bytes.fromhex(member["sha256"])) + expected_digest = digest.hexdigest() + else: + if len(members) != 1 or members[0]["role"] != item["role"]: + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence file membership is invalid." + ) + expected_digest = members[0]["sha256"] + if item["sha256"] != expected_digest or item["size_bytes"] != sum( + member["size_bytes"] for member in members + ): + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence source identity is invalid." + ) + if used != set(range(len(artifacts))): + raise RepositoryError("REPOSITORY_CORRUPTION", "Evidence membership is incomplete.") + inputs = body["analysis_request"].get("inputs", []) + indices = layout["analysis_sources"] + if indices and len(indices) != len(inputs): + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence analysis source layout is invalid." + ) + for index, input_item in zip(indices, inputs, strict=False): + if type(index) is not int or not 0 <= index < len(layout["sources"]): + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence analysis source layout is invalid." + ) + source = layout["sources"][index] + if any( + source.get(key) != input_item.get(key) for key in ("sha256", "format", "producer") + ): + raise RepositoryError( + "REPOSITORY_CORRUPTION", "Evidence analysis source identity is invalid." + ) @staticmethod def _validate_capture_request(value: Any) -> None: @@ -763,8 +962,27 @@ def _validate_capture_execution(value: Any) -> None: "containment", "limit", } - if not isinstance(value, dict) or set(value) != fields: + attribution_fields = {"returncode_scope", "workload_returncode", "executable_sha256"} + if not isinstance(value, dict) or set(value) not in (fields, fields | attribution_fields): raise RepositoryError("REPOSITORY_CORRUPTION", "Evidence capture request is invalid.") + if attribution_fields <= set(value) and ( + value["returncode_scope"] not in {"workload", "collector"} + or not isinstance(value["executable_sha256"], str) + or re.fullmatch(LOWERCASE_SHA256_PATTERN, value["executable_sha256"]) is None + or ( + value["workload_returncode"] is not None + and type(value["workload_returncode"]) is not int + ) + or ( + value["returncode_scope"] == "collector" + and value["workload_returncode"] is not None + ) + or ( + value["returncode_scope"] == "workload" + and value["workload_returncode"] != value["returncode"] + ) + ): + raise RepositoryError("REPOSITORY_CORRUPTION", "Capture exit attribution is invalid.") argv_fields = (value.get("argv"), value.get("capture_argv")) missing_roles = value.get("missing_artifact_roles") scalar_valid = ( diff --git a/src/flameox/runtime_contracts.py b/src/flameox/runtime_contracts.py index b60863a..c77b3a5 100644 --- a/src/flameox/runtime_contracts.py +++ b/src/flameox/runtime_contracts.py @@ -76,6 +76,17 @@ class EvidenceSource(StrictModel): min_length=1, max_length=80, ) + artifact_selector: str | None = Field( + default=None, + description="Opaque artifact selector returned by the evidence resource; pass unchanged.", + pattern=LOWERCASE_SHA256_PATTERN, + ) + + @model_validator(mode="after") + def one_selector(self) -> EvidenceSource: + if self.artifact_role is not None and self.artifact_selector is not None: + raise ValueError("Use artifact_role or artifact_selector, not both") + return self Source = Annotated[PathSource | EvidenceSource, Field(discriminator="kind")] @@ -87,7 +98,12 @@ class EmptyArguments(StrictModel): class PreviewArguments(StrictModel): offset: int = Field( - default=0, description="Zero-based byte offset at which to begin the preview.", ge=0 + default=0, + description=( + "Zero-based logical row offset: text lines, JSONL records, CSV data records, " + "Parquet records, or projected JSON entries. Continuations advance in the same unit." + ), + ge=0, ) @@ -722,7 +738,8 @@ class ExperimentDesign(StrictModel): ) practical_threshold: float = Field( description=( - "Absolute difference considered practically equivalent, in metric units (nanoseconds)." + "Absolute margin for describing the observed point estimate, in nanoseconds. " + "Being within this margin does not establish statistical or semantic equivalence." ), ge=0, ) diff --git a/src/flameox/setup.py b/src/flameox/setup.py index fec91a6..04ae1dd 100644 --- a/src/flameox/setup.py +++ b/src/flameox/setup.py @@ -20,20 +20,40 @@ MANAGED_PROVIDER_EXTRAS, SYSTEM_PROVIDER_GUIDANCE, ) +from flameox.providers.environment import ( + DEFAULT_PREPARATION_TIMEOUT_SECONDS as DEFAULT_PREPARATION_TIMEOUT_SECONDS, +) +from flameox.providers.environment import ( + MAX_PREPARATION_TIMEOUT_SECONDS as MAX_PREPARATION_TIMEOUT_SECONDS, +) +from flameox.providers.environment import ( + ExternalRequirement as ExternalRequirement, +) +from flameox.providers.environment import ( + ProviderPreparation as ProviderPreparation, +) +from flameox.providers.environment import ( + ProviderSelectionFailure as ProviderSelectionFailure, +) +from flameox.providers.environment import ( + SetupFailure as SetupFailure, +) +from flameox.providers.environment import ( + _validate_providers, +) +from flameox.providers.environment import ( + active_provider_status as active_provider_status, +) +from flameox.providers.environment import ( + external_provider_requirements as external_provider_requirements, +) +from flameox.providers.environment import ( + mcp_launcher as mcp_launcher, +) -DEFAULT_PREPARATION_TIMEOUT_SECONDS = 1_800 -MAX_PREPARATION_TIMEOUT_SECONDS = 3_600 PATH_CLI_PROBE_TIMEOUT_SECONDS = 5 -class SetupFailure(RuntimeError): - pass - - -class ProviderSelectionFailure(SetupFailure): - pass - - class SetupClient(StrEnum): CLAUDE = "claude" CURSOR = "cursor" @@ -312,17 +332,13 @@ def _jsonc_update_mcp_entry(source: str, section_name: str, entry: object) -> st root_start = _jsonc_skip_trivia(source, 0) root_properties, root_end = _jsonc_object_properties(source, root_start) root_indent = ( - _jsonc_line_indent(source, root_properties[0].key_start, " ") - if root_properties - else " " + _jsonc_line_indent(source, root_properties[0].key_start, " ") if root_properties else " " ) section = next((item for item in root_properties if item.key == section_name), None) if section is None: entry_indent = f"{root_indent} " section_value = ( - "{\n" - f"{_jsonc_property_text('flameox', entry, entry_indent)}\n" - f"{root_indent}}}" + f"{{\n{_jsonc_property_text('flameox', entry, entry_indent)}\n{root_indent}}}" ) return _jsonc_insert_property( source, @@ -343,7 +359,7 @@ def _jsonc_update_mcp_entry(source: str, section_name: str, entry: object) -> st if existing is not None: rendered_entry = json.dumps(entry, ensure_ascii=False, indent=2) rendered_entry = rendered_entry.replace(chr(10), chr(10) + entry_indent) - return f"{source[:existing.value_start]}{rendered_entry}{source[existing.value_end:]}" + return f"{source[: existing.value_start]}{rendered_entry}{source[existing.value_end :]}" return _jsonc_insert_property( source, section_end, @@ -499,69 +515,6 @@ def apply_client_setup(plans: list[ClientSetupPlan]) -> list[ClientSetupResult]: return results -@dataclass(frozen=True, slots=True) -class ExternalRequirement: - provider_id: str - guidance: str - - -@dataclass(frozen=True, slots=True) -class ProviderPreparation: - requested_providers: list[str] - prepared_managed_providers: list[str] - external_requirements: list[ExternalRequirement] - preparation_command: list[str] - launcher_command: str - launcher_args: list[str] - - @property - def preparation_status(self) -> Literal["prepared", "not_applicable"]: - return "prepared" if self.prepared_managed_providers else "not_applicable" - - @property - def restart_required(self) -> bool: - return bool(self.prepared_managed_providers) - - -def _validate_providers(providers: list[str]) -> None: - unknown = sorted( - set(providers).difference(MANAGED_PROVIDER_EXTRAS).difference(SYSTEM_PROVIDER_GUIDANCE) - ) - if unknown: - supported = ", ".join(sorted(MANAGED_PROVIDER_EXTRAS | SYSTEM_PROVIDER_GUIDANCE)) - raise ProviderSelectionFailure( - f"Unknown provider {unknown[0]!r}; choose one of: {supported}" - ) - - -def mcp_launcher(providers: list[str]) -> tuple[str, list[str]]: - """Return a version-bound MCP launcher for client configuration.""" - - _validate_providers(providers) - extras = sorted( - { - MANAGED_PROVIDER_EXTRAS[provider] - for provider in providers - if provider in MANAGED_PROVIDER_EXTRAS - } - ) - extras_suffix = f"[{','.join(extras)}]" if extras else "" - requirement = f"flameox{extras_suffix}=={__version__}" - return ( - "uvx", - ["--python", "3.12", "--from", requirement, "flameox"], - ) - - -def external_provider_requirements(providers: list[str]) -> list[ExternalRequirement]: - _validate_providers(providers) - return [ - ExternalRequirement(provider, SYSTEM_PROVIDER_GUIDANCE[provider]) - for provider in dict.fromkeys(providers) - if provider in SYSTEM_PROVIDER_GUIDANCE - ] - - def _decode_stderr(stderr: bytes | str | None) -> str: if not stderr: return "" @@ -629,4 +582,5 @@ def prepare_providers( preparation_command, launcher_command, server_args, + active_provider_status(managed) if managed else "not_applicable", ) diff --git a/src/flameox/stateless.py b/src/flameox/stateless.py index 19f4782..ff4dc33 100644 --- a/src/flameox/stateless.py +++ b/src/flameox/stateless.py @@ -63,6 +63,7 @@ from flameox.providers.nvbench import NvbenchProvider from flameox.providers.otlp import OtlpProvider from flameox.providers.perfetto import PerfettoProvider +from flameox.providers.preparation import ProviderDependencies from flameox.providers.reliability import ReliabilityProvider from flameox.providers.source_evidence import SourceEvidenceProvider from flameox.providers.structured_workers import StructuredWorkerProviders @@ -71,6 +72,7 @@ EvidenceRepository, NativeArtifact, RepositoryError, + artifact_selector, sha256_file, ) from flameox.runtime_contracts import ( @@ -195,8 +197,16 @@ def __init__( self.repository = EvidenceRepository( evidence_directory or default_data_directory(), self.session_id ) + self._repository_configuration = ( + "explicit_directory" + if evidence_directory is not None + else "environment_override" + if os.environ.get("FLAMEOX_DATA_DIR") + else "platform_default" + ) self.analyses: OrderedDict[str, CachedAnalysis] = OrderedDict() self.conversions: OrderedDict[tuple[str, str], Path] = OrderedDict() + self.dependencies = ProviderDependencies(self.broker, self.scratch) def close(self) -> None: self._temporary.cleanup() @@ -369,7 +379,14 @@ def analyze( f"No typed {capability_id} provider accepts the supplied inputs", ) if provider_analysis is None: - rows, observed, complete = self._read_rows(resolved, offset, selected_limits.max_rows) + try: + rows, observed, complete = self._read_rows( + resolved, offset, selected_limits.max_rows + ) + except (ijson.JSONError, json.JSONDecodeError, UnicodeDecodeError) as error: + raise RuntimeFailure( + "DECODE_FAILURE", "Artifact preview could not decode the input." + ) from error if continuation is not None and offset >= observed: raise RuntimeFailure( "INVALID_INPUT", "Continuation offset is beyond the available evidence" @@ -599,6 +616,7 @@ async def capture_and_analyze( # noqa: C901 - capture lifecycle keeps failure a "external_setup_guidance": SYSTEM_PROVIDER_GUIDANCE[target.provider_id], }, ) + self.dependencies.verify_capture_binding(target.provider_id, binding) if experiment is not None and experiment.semantic_oracle is not None: self._require_host_tool( experiment.semantic_oracle[0], @@ -619,6 +637,11 @@ async def capture_and_analyze( # noqa: C901 - capture lifecycle keeps failure a ) pending_executions.append( self._pending_capture_execution(case, block, argv, invocation.argv, cwd) + | { + "executable_sha256": binding.identity.sha256.removeprefix("sha256:"), + "returncode_scope": invocation.returncode_scope, + "workload_returncode": None, + } ) self._check_capture_provenance_capacity( target=target, @@ -796,6 +819,11 @@ async def capture_and_analyze( # noqa: C901 - capture lifecycle keeps failure a "capture_argv": list(invocation.argv), "cwd": str(cwd), "returncode": exit_code, + "executable_sha256": binding.identity.sha256.removeprefix("sha256:"), + "returncode_scope": invocation.returncode_scope, + "workload_returncode": exit_code + if invocation.returncode_scope == "workload" + else None, "status": status, "failure_code": failure_code, "missing_artifact_roles": missing_artifact_roles, @@ -932,6 +960,7 @@ def _capture_failure_result( "mode": mode, "requested_capability_id": capability_id, "executions": executions, + "outcome": self._capture_outcome(executions), }, "analysis_failure": failure_body, } @@ -975,6 +1004,7 @@ def _finalize_capture_result( "mode": mode, "requested_capability_id": capability_id, "executions": executions, + "outcome": self._capture_outcome(executions), } if experiment is not None: experiment_blocks, experiment_limitations = self._experiment_blocks( @@ -1104,16 +1134,17 @@ def _managed_executable(name: str) -> str | None: return str(candidate) return None - @staticmethod - def _require_managed_executable(provider_id: str, name: str) -> str: - executable = AnalysisRuntime._managed_executable(name) + def _require_managed_executable(self, provider_id: str, name: str) -> str: + executable = ( + self.dependencies.py_spy_executable() if provider_id == "py-spy" else None + ) or self._managed_executable(name) if executable is None: raise RuntimeFailure( "UNAVAILABLE_CAPABILITY", ( f"Managed provider executable is unavailable: {name}. Call " - f"prepare_providers with provider_ids=[{provider_id!r}], reconnect with " - "the returned launcher, then retry." + f"prepare_providers with provider_ids=[{provider_id!r}], then follow its " + "activation guidance and retry." ), details={ "provider_id": provider_id, @@ -1263,7 +1294,7 @@ def preserve_evidence(self, analysis_id: str) -> dict[str, Any]: self.repository.read(str(cached.preserved["evidence_id"])) except RepositoryError as exc: if exc.code != "MISSING_EVIDENCE": - raise RuntimeFailure(exc.code, exc.message) from exc + raise self._repository_failure(exc) from exc cached.preserved = None except OSError as exc: raise RuntimeFailure( @@ -1274,6 +1305,7 @@ def preserve_evidence(self, analysis_id: str) -> dict[str, Any]: return dict(cached.preserved) if cached.preserved is None: artifacts: list[NativeArtifact] = [] + source_layout: list[dict[str, Any]] = [] role_counts: dict[str, int] = {} for source in cached.sources: role_counts[source.role] = role_counts.get(source.role, 0) + 1 @@ -1283,7 +1315,18 @@ def preserve_evidence(self, analysis_id: str) -> dict[str, Any]: if role_counts[source.role] == 1 else f"source-{source_index:04d}/{source.role}" ) + layout = { + "role": publication_role, + "sha256": source.sha256, + "size_bytes": source.size_bytes, + "format": source.format, + "producer": source.producer, + "is_directory": source.path.is_dir(), + "artifact_indices": [], + } + source_layout.append(layout) if source.path.is_file(): + layout["artifact_indices"] = [len(artifacts)] artifacts.append( NativeArtifact( source.path, @@ -1301,7 +1344,11 @@ def preserve_evidence(self, analysis_id: str) -> dict[str, Any]: "MISSING_OR_CHANGED_INPUT", f"Input changed before preservation: {source.path}", ) - for path in self._directory_files(source.path): + files = self._directory_files(source.path) + layout["artifact_indices"] = list( + range(len(artifacts), len(artifacts) + len(files)) + ) + for path in files: digest, size = sha256_file(path) relative = path.relative_to(source.path).as_posix() artifacts.append( @@ -1314,15 +1361,36 @@ def preserve_evidence(self, analysis_id: str) -> dict[str, Any]: source.producer, ) ) + analysis_source_indices: list[int] = [] + for item in cached.manifest_body["analysis_request"].get("inputs", []): + index = next( + ( + index + for index, source in enumerate(cached.sources) + if (str(source.path), source.sha256, source.format, source.producer) + == (item["path"], item["sha256"], item["format"], item.get("producer")) + ), + None, + ) + if index is None: + analysis_source_indices = [] + break + analysis_source_indices.append(index) try: cached.preserved = self.repository.preserve( - manifest_body=cached.manifest_body, + manifest_body=cached.manifest_body + | { + "source_layout": { + "sources": source_layout, + "analysis_sources": analysis_source_indices, + } + }, artifacts=artifacts, analysis=self._durable_analysis(cached.result), ) self._release_analysis_scratch(cached) except RepositoryError as exc: - raise RuntimeFailure(exc.code, exc.message) from exc + raise self._repository_failure(exc) from exc except OSError as exc: raise RuntimeFailure( "REPOSITORY_IO_FAILURE", "Evidence could not be preserved." @@ -1416,7 +1484,7 @@ def query_evidence( cursor=cursor, ) except RepositoryError as exc: - raise RuntimeFailure(exc.code, exc.message) from exc + raise self._repository_failure(exc) from exc except OSError as exc: raise RuntimeFailure( "REPOSITORY_IO_FAILURE", "The evidence repository could not be queried." @@ -1426,7 +1494,7 @@ def read_evidence(self, evidence_id: str) -> dict[str, Any]: try: return self.repository.read(evidence_id) except RepositoryError as exc: - raise RuntimeFailure(exc.code, exc.message) from exc + raise self._repository_failure(exc) from exc except OSError as exc: raise RuntimeFailure( "REPOSITORY_IO_FAILURE", "The requested evidence could not be read." @@ -1436,12 +1504,30 @@ def read_evidence_agent_projection(self, evidence_id: str) -> dict[str, Any]: try: return self.repository.read_agent_projection(evidence_id) except RepositoryError as exc: - raise RuntimeFailure(exc.code, exc.message) from exc + raise self._repository_failure(exc) from exc except OSError as exc: raise RuntimeFailure( "REPOSITORY_IO_FAILURE", "The requested evidence projection could not be read." ) from exc + def _repository_failure(self, error: RepositoryError) -> RuntimeFailure: + details: dict[str, Any] = {} + if error.code in {"REPOSITORY_CORRUPTION", "UNSUPPORTED_REPOSITORY_FORMAT"}: + details = { + "configuration_source": self._repository_configuration, + "configuration_variable": "FLAMEOX_DATA_DIR", + "store_identifier": hashlib.sha256(str(self.repository.root).encode()).hexdigest(), + "recovery": [ + "Restore the original repository from a known-good backup; do not delete " + "existing data or synthesize repository.json.", + "Alternatively set FLAMEOX_DATA_DIR to a distinct empty directory and " + "restart or reconnect Flameox. Switching stores does not recover old evidence; " + "preserve any recoverable session evidence before ending the session.", + ], + "local_diagnostic": "flameox evidence location", + } + return RuntimeFailure(error.code, error.message, details=details) + def _resolve_sources( self, sources: Sequence[Source], limits: RequestLimits ) -> list[ResolvedSource]: @@ -1538,22 +1624,60 @@ def _resolve_evidence_source(self, source: EvidenceSource) -> ResolvedSource: artifacts = manifest["body"].get("artifacts", []) if not isinstance(artifacts, list): raise RuntimeFailure("REPOSITORY_CORRUPTION", "Evidence artifacts are invalid") - role = source.artifact_role or self._default_evidence_role(artifacts) - selected = [ - item - for item in artifacts - if isinstance(item, dict) - and (item.get("role") == role or str(item.get("role", "")).startswith(role + ":")) - ] - if not selected: + role = source.artifact_role + logical = self.repository.logical_artifacts(manifest["body"]) + selected_source: dict[str, Any] | None = None + if source.artifact_selector is not None: + selected_source = next( + ( + item + | {"is_directory": collection == "logical" and item.get("is_directory", False)} + for collection, items in (("artifact", artifacts), ("logical", logical)) + for index, item in enumerate(items) + if artifact_selector(source.evidence_id, collection, index) + == source.artifact_selector + ), + None, + ) + if selected_source is None: + raise RuntimeFailure( + "MISSING_EVIDENCE", + "The requested evidence artifact selector is absent", + details={"resource_uri": f"flameox://evidence/{source.evidence_id}"}, + ) + role = selected_source["role"] + if role is None: + try: + role = self._default_evidence_role(logical) + except RuntimeFailure as error: + raise RuntimeFailure( + error.code, + error.message, + details={"resource_uri": f"flameox://evidence/{source.evidence_id}"}, + ) from error + if selected_source is None: + # Legacy role selection also prefers an exact file over a bundle prefix. + selected_source = next( + ( + item + for item in [ + *(artifact | {"is_directory": False} for artifact in artifacts), + *logical, + ] + if item["role"] == role + ), + None, + ) + if selected_source is None: raise RuntimeFailure( "MISSING_EVIDENCE", "The requested evidence artifact role is absent" ) - if len(selected) > 1 or any( - str(item.get("role", "")).startswith(role + ":") for item in selected - ): - return self._materialize_evidence_bundle(source.evidence_id, role, selected) - artifact = selected[0] + if selected_source.get("is_directory", False): + selected = [artifacts[index] for index in selected_source["artifact_indices"]] + return self._materialize_evidence_bundle( + source.evidence_id, role, selected, selected_source + ) + artifact = selected_source digest = str(artifact["sha256"]) path = self.repository.root / "artifacts" / "sha256" / digest[:2] / digest / "payload" return ResolvedSource( @@ -1568,7 +1692,7 @@ def _resolve_evidence_source(self, source: EvidenceSource) -> ResolvedSource: @staticmethod def _default_evidence_role(artifacts: list[Any]) -> str: roots = { - str(item["role"]).split(":", 1)[0] + str(item["role"]) for item in artifacts if isinstance(item, dict) and isinstance(item.get("role"), str) @@ -1584,7 +1708,11 @@ def _default_evidence_role(artifacts: list[Any]) -> str: return roots.pop() def _materialize_evidence_bundle( - self, evidence_id: str, role: str, artifacts: list[dict[str, Any]] + self, + evidence_id: str, + role: str, + artifacts: list[dict[str, Any]], + metadata: dict[str, Any], ) -> ResolvedSource: bundle_key = hashlib.sha256(f"{evidence_id}:{role}".encode()).hexdigest() destination = self.scratch / "evidence-sources" / bundle_key @@ -1626,12 +1754,16 @@ def _materialize_evidence_bundle( if stage.exists(): shutil.rmtree(stage) digest, size, _file_count = self._hash_path(destination) - formats = {str(item["format"]) for item in artifacts} + formats = {str(item["format"]) for item in [*artifacts, metadata]} producers = { str(item["producer"]) if item.get("producer") is not None else None - for item in artifacts + for item in [*artifacts, metadata] } - if len(formats) != 1 or len(producers) != 1: + if ( + len(formats) != 1 + or len(producers) != 1 + or (digest, size) != (metadata["sha256"], metadata["size_bytes"]) + ): raise RuntimeFailure( "REPOSITORY_CORRUPTION", "Evidence bundle metadata is inconsistent" ) @@ -2099,7 +2231,12 @@ def _iter_rows(self, path: Path, format_name: str) -> Iterator[dict[str, Any]]: "coverage", }: with path.open("rb") as stream: - prefix = stream.read(1) + prefix = b"" + while chunk := stream.read(4096): + significant = chunk.lstrip(b" \t\r\n") + if significant: + prefix = significant[:1] + break stream.seek(0) if prefix == b"[": for value in ijson.items(stream, "item", use_float=True): @@ -2167,15 +2304,18 @@ def _shrink_result( def _bound_capture_result( self, result: dict[str, Any], limit: int, identity: Mapping[str, Any], offset: int ) -> None: - if len(canonical_bytes(result)) <= limit: - return capture = cast(dict[str, Any], result["capture"]) executions = cast(list[dict[str, Any]], capture["executions"]) + if len(canonical_bytes(result)) <= limit: + return compact = [ { "case": item["case"], "block": item["block"], "returncode": item["returncode"], + "executable_sha256": item.get("executable_sha256"), + "returncode_scope": item.get("returncode_scope", "unknown"), + "workload_returncode": item.get("workload_returncode"), "status": item["status"], "failure_code": item["failure_code"], "wall_time_ns": item["wall_time_ns"], @@ -2207,6 +2347,17 @@ def _bound_capture_result( if len(canonical_bytes(result)) > limit: self._shrink_result(result, limit, identity, offset) + @staticmethod + def _capture_outcome(executions: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + succeeded = sum(item["status"] == "succeeded" for item in executions) + failed = len(executions) - succeeded + return { + "status": "failed" if failed else "succeeded", + "execution_count": len(executions), + "succeeded_count": succeeded, + "failed_count": failed, + } + @staticmethod def _experiment_blocks( experiment: ExperimentDesign, executions: Sequence[Mapping[str, Any]] @@ -2268,7 +2419,7 @@ def _experiment_blocks( "confidence_level": 0.95 if confidence_low is not None else None, "method": method, "practical_threshold": experiment.practical_threshold, - "decision": decision, + "point_estimate_classification": decision, "paired_blocks": len(differences), "declared_blocks": experiment.blocks, } @@ -2280,6 +2431,7 @@ def _experiment_blocks( "values": { "experiment_metric": experiment.metric, "experiment_estimand": experiment.estimand, + "decision_basis": "descriptive_point_estimate", "baseline_case": baseline, "comparison_count": len(rows), }, @@ -2323,7 +2475,7 @@ def _experiment_interval( def _encode_continuation(self, identity: Mapping[str, Any], offset: int) -> str: payload = { - "request": hashlib.sha256(canonical_bytes(identity)).hexdigest(), + "request": self._continuation_digest(identity), "offset": offset, } checksum = hashlib.sha256(canonical_bytes(payload)).hexdigest() @@ -2342,10 +2494,9 @@ def _decode_continuation( value = json.loads(base64.urlsafe_b64decode(token + "=" * (-len(token) % 4))) payload = value["payload"] expected = hashlib.sha256(canonical_bytes(payload)).hexdigest() - if ( - not secrets.compare_digest(value["checksum"], expected) - or payload["request"] != hashlib.sha256(canonical_bytes(identity)).hexdigest() - ): + if not secrets.compare_digest(value["checksum"], expected) or payload[ + "request" + ] != self._continuation_digest(identity): raise ValueError offset = payload["offset"] if type(offset) is not int or offset < 0: @@ -2356,6 +2507,16 @@ def _decode_continuation( "INVALID_INPUT", "Continuation does not match this request and its inputs" ) from exc + @staticmethod + def _continuation_digest(identity: Mapping[str, Any]) -> str: + semantic = dict(identity) + if "inputs" in semantic: + semantic["inputs"] = [ + {key: value for key, value in item.items() if key not in {"path", "role"}} + for item in semantic["inputs"] + ] + return hashlib.sha256(canonical_bytes(semantic)).hexdigest() + @staticmethod def _resolve_capture_cwd(value: str) -> Path: candidate = Path(value) diff --git a/tests/test_evidence_handoffs.py b/tests/test_evidence_handoffs.py new file mode 100644 index 0000000..0fe943d --- /dev/null +++ b/tests/test_evidence_handoffs.py @@ -0,0 +1,521 @@ +from __future__ import annotations + +import hashlib +import json +import os +import sys +from pathlib import Path +from typing import Any + +import anyio +import pytest +from mcp import Client +from mcp_types import TextResourceContents + +from flameox.canonical import canonical_bytes +from flameox.mcp import create_server +from flameox.runtime_contracts import ( + CaptureTarget, + EvidenceSource, + ExperimentCase, + ExperimentDesign, + PathSource, + PreviewArguments, + RequestLimits, + RuntimeFailure, +) +from flameox.stateless import AnalysisRuntime + + +@pytest.mark.integration +@pytest.mark.parametrize("members", [[], ["foo", "foo:bar"], ["credentials.json"]]) +def test_directory_handoff_preserves_empty_bundles_and_exact_members( + tmp_path: Path, members: list[str] +) -> None: + bundle = tmp_path / "bundle" + bundle.mkdir() + for name in members: + (bundle / name).write_text(name + "\n") + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + result = runtime.analyze("artifact.preview", [PathSource(path=str(bundle))], {}) + ref = runtime.preserve_evidence(result["analysis_id"]) + runtime.close() + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + projection = runtime.read_evidence_agent_projection(ref["evidence_id"]) + assert len(projection["analysis_sources"]) == 1 + sources = [EvidenceSource.model_validate(item) for item in projection["analysis_sources"]] + restored = runtime.analyze("artifact.preview", sources, {}) + assert restored["inputs"][0]["sha256"] == result["inputs"][0]["sha256"] + for artifact in projection["body"]["artifacts"]: + selected = runtime.analyze( + "artifact.preview", [EvidenceSource.model_validate(artifact["source"])], {} + ) + assert selected["inputs"][0]["sha256"] == artifact["sha256"] + selected_ref = runtime.preserve_evidence(selected["analysis_id"]) + selected_projection = runtime.read_evidence_agent_projection( + selected_ref["evidence_id"] + ) + assert len(selected_projection["analysis_sources"]) == 1 + again = runtime.analyze( + "artifact.preview", + [EvidenceSource.model_validate(selected_projection["analysis_sources"][0])], + {}, + ) + assert again["inputs"][0]["sha256"] == artifact["sha256"] + assert artifact["source"]["artifact_selector"] not in { + hashlib.sha256(f"input:{name}".encode()).hexdigest() for name in members + } + if members: + legacy = runtime.analyze( + "artifact.preview", + [ + EvidenceSource( + kind="evidence", + evidence_id=ref["evidence_id"], + artifact_role=f"input:{members[0]}", + ) + ], + {}, + ) + assert legacy["blocks"][1]["rows"][0]["text"] == members[0] + else: + assert projection["body"]["artifacts"] == [] + default = runtime.analyze( + "artifact.preview", + [EvidenceSource(kind="evidence", evidence_id=ref["evidence_id"])], + {}, + ) + assert default["inputs"][0]["sha256"] == result["inputs"][0]["sha256"] + finally: + runtime.close() + + +@pytest.mark.process +@pytest.mark.parametrize("provider", ["benchmark-samples", "observations", "torch-profiler"]) +def test_self_reporting_capture_retains_workload_exit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str +) -> None: + def workload_available( + self: AnalysisRuntime, + provider_id: str, + target_argv: list[str], + environment: dict[str, str], + *, + cwd: Path, + ) -> None: + pass # The fixture exits without importing optional workload SDKs. + + monkeypatch.setattr(AnalysisRuntime, "_require_workload_python_provider", workload_available) + + async def exercise() -> None: + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + result = await runtime.capture_and_analyze( + CaptureTarget( + argv=[sys.executable, "-c", "raise SystemExit(7)"], + cwd=str(tmp_path), + provider_id=provider, + ), + { + "benchmark-samples": "benchmark.summary", + "observations": "failures.summary", + "torch-profiler": "trace.summary", + }[provider], + preserve=True, + ) + execution = result["capture"]["executions"][0] + assert execution["returncode_scope"] == "workload" + assert execution["workload_returncode"] == 7 + manifest = runtime.read_evidence(result["preserved"]["evidence_id"]) + assert manifest["body"]["capture_request"]["executions"][0]["workload_returncode"] == 7 + finally: + runtime.close() + + anyio.run(exercise) + + +@pytest.mark.integration +@pytest.mark.parametrize( + "corruption", ["role_type", "index_type", "duplicate", "digest", "kind", "analysis_index"] +) +def test_source_layout_corruption_is_a_typed_repository_failure( + tmp_path: Path, corruption: str +) -> None: + bundle = tmp_path / "bundle" + bundle.mkdir() + (bundle / "member").write_text("contents") + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + result = runtime.analyze("artifact.preview", [PathSource(path=str(bundle))], {}) + ref = runtime.preserve_evidence(result["analysis_id"]) + evidence_id = ref["evidence_id"] + path = tmp_path / "store" / "evidence" / "sha256" / evidence_id[:2] / evidence_id + manifest = json.loads((path / "manifest.json").read_text()) + layout = manifest["body"]["source_layout"] + match corruption: + case "role_type": + manifest["body"]["artifacts"][0]["role"] = 1 + case "index_type": + layout["sources"][0]["artifact_indices"] = [True] + case "duplicate": + layout["sources"][0]["artifact_indices"] = [0, 0] + case "digest": + layout["sources"][0]["sha256"] = "0" * 64 + case "kind": + layout["sources"][0]["is_directory"] = False + case "analysis_index": + layout["analysis_sources"] = [1] + malformed_id = hashlib.sha256(canonical_bytes(manifest["body"])).hexdigest() + manifest["evidence_id"] = malformed_id + destination = path.parent.parent / malformed_id[:2] / malformed_id + destination.parent.mkdir(exist_ok=True) + path.rename(destination) + (destination / "manifest.json").write_bytes(canonical_bytes(manifest)) + with pytest.raises(RuntimeFailure) as failure: + runtime.read_evidence_agent_projection(malformed_id) + assert failure.value.code == "REPOSITORY_CORRUPTION" + assert "recovery" in failure.value.details + finally: + runtime.close() + + +@pytest.mark.integration +def test_artifact_extra_fields_cannot_override_file_identity(tmp_path: Path) -> None: + artifact = tmp_path / "input.txt" + artifact.write_text("contents") + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + result = runtime.analyze("artifact.preview", [PathSource(path=str(artifact))], {}) + ref = runtime.preserve_evidence(result["analysis_id"]) + evidence_id = ref["evidence_id"] + path = tmp_path / "store" / "evidence" / "sha256" / evidence_id[:2] / evidence_id + manifest = json.loads((path / "manifest.json").read_text()) + manifest["body"]["artifacts"][0]["is_directory"] = True + updated_id = hashlib.sha256(canonical_bytes(manifest["body"])).hexdigest() + manifest["evidence_id"] = updated_id + destination = path.parent.parent / updated_id[:2] / updated_id + destination.parent.mkdir(exist_ok=True) + path.rename(destination) + (destination / "manifest.json").write_bytes(canonical_bytes(manifest)) + projection = runtime.read_evidence_agent_projection(updated_id) + for source in ( + EvidenceSource.model_validate(projection["body"]["artifacts"][0]["source"]), + EvidenceSource(kind="evidence", evidence_id=updated_id, artifact_role="input"), + ): + restored = runtime.analyze("artifact.preview", [source], {}) + assert restored["inputs"][0]["sha256"] == result["inputs"][0]["sha256"] + finally: + runtime.close() + + +@pytest.mark.integration +@pytest.mark.parametrize("whitespace", ["", "\n", " \t\r\n" * 2000], ids=["compact", "lf", "long"]) +def test_preview_json_whitespace_preserves_rows(tmp_path: Path, whitespace: str) -> None: + artifact = tmp_path / "array.json" + artifact.write_text(whitespace + '[{"value":1},{"value":2}]') + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + result = runtime.analyze("artifact.preview", [PathSource(path=str(artifact))], {}) + assert [row["value"] for row in result["blocks"][1]["rows"]] == [1, 2] + assert result["coverage"]["complete"] is True + finally: + runtime.close() + + +@pytest.mark.integration +def test_preview_offset_is_a_logical_row(tmp_path: Path) -> None: + assert "row" in PreviewArguments.model_json_schema()["properties"]["offset"]["description"] + artifact = tmp_path / "lines.txt" + artifact.write_text("a\nlong second line\nthird\n") + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + result = runtime.analyze( + "artifact.preview", [PathSource(path=str(artifact))], {"offset": 1} + ) + assert [row["text"] for row in result["blocks"][1]["rows"]] == ["long second line", "third"] + finally: + runtime.close() + + +@pytest.mark.process +@pytest.mark.parametrize("inline", [False, True]) +def test_preserved_capture_continuation_uses_discovered_sources( + tmp_path: Path, inline: bool +) -> None: + async def exercise() -> None: + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + limits = RequestLimits(max_rows=2) + first = await runtime.capture_and_analyze( + CaptureTarget( + argv=[sys.executable, "-c", "print('one\\ntwo\\nthree\\nfour\\nfive')"], + cwd=str(tmp_path), + provider_id="direct", + ), + "artifact.preview", + limits=limits, + preserve=inline, + ) + preserved = first.get("preserved") or runtime.preserve_evidence(first["analysis_id"]) + execution = first["capture"]["executions"][0] + assert execution["returncode_scope"] == "workload" + assert execution["workload_returncode"] == 0 + projection = runtime.read_evidence_agent_projection(preserved["evidence_id"]) + sources = [ + EvidenceSource.model_validate(item) for item in projection["analysis_sources"] + ] + rows = list(first["blocks"][1]["rows"]) + token = first["continuation"] + while token: + page = runtime.analyze( + "artifact.preview", sources, {}, limits=limits, continuation=token + ) + rows.extend(page["blocks"][1]["rows"]) + token = page["continuation"] + assert [row["text"] for row in rows] == ["one", "two", "three", "four", "five"] + assert not list(runtime.scratch.glob("capture-*")) + artifacts = projection["body"]["artifacts"] + for artifact in artifacts: + selected = runtime.analyze( + "artifact.preview", [EvidenceSource.model_validate(artifact["source"])], {} + ) + assert selected["inputs"][0]["sha256"] == artifact["sha256"] + with pytest.raises(RuntimeFailure, match="Continuation"): + runtime.analyze( + "artifact.preview", + sources, + {"offset": 1}, + limits=limits, + continuation=first["continuation"], + ) + finally: + runtime.close() + + anyio.run(exercise) + + +@pytest.mark.unit +@pytest.mark.parametrize("failed_index", [0, 15]) +@pytest.mark.parametrize("text_width", [140, 170]) +def test_capture_summary_survives_diagnostic_truncation(failed_index: int, text_width: int) -> None: + executions = [ + dict( + case=f"case-{i}", + block=1, + returncode=int(i == failed_index), + status="failed" if i == failed_index else "succeeded", + failure_code=None, + wall_time_ns=100, + containment="process_group", + limit=None, + semantic_oracle=None, + ) + for i in range(16) + ] + result: dict[str, Any] = dict( + analysis_id="a" * 64, + capability_id="artifact.preview", + coverage=dict(complete=True, rows_returned=100, rows_observed=100), + blocks=[ + dict(type="metrics", values={}), + dict(type="table", rows=[dict(text="x" * text_width) for _ in range(100)]), + ], + limitations=[], + continuation=None, + capture=dict(executions=executions, outcome=AnalysisRuntime._capture_outcome(executions)), + ) + runtime = object.__new__(AnalysisRuntime) + runtime._bound_capture_result(result, 16384, {}, 0) + assert result["capture"]["outcome"] == { + "status": "failed", + "execution_count": 16, + "succeeded_count": 15, + "failed_count": 1, + } + assert len(canonical_bytes(result)) <= 16384 + if text_width == 170: + assert result["capture"]["executions"] == [] + + +@pytest.mark.integration +@pytest.mark.parametrize("failure_code", [None, "SEMANTIC_ORACLE_FAILED"]) +def test_mcp_uses_capture_outcome_when_execution_diagnostics_are_empty( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_code: str | None, +) -> None: + executions = [{"status": "failed", "failure_code": failure_code} for _ in range(16)] + + async def capture(self: AnalysisRuntime, *args: Any, **kwargs: Any) -> dict[str, Any]: + return { + "capture": { + "outcome": self._capture_outcome(executions), + "executions": [], + "executions_truncated": 16, + } + } + + monkeypatch.setattr(AnalysisRuntime, "capture_and_analyze", capture) + + async def exercise() -> None: + async with Client(create_server(evidence_directory=tmp_path / "store")) as client: + result = await client.call_tool( + "capture_process_output", + { + "target": {"argv": [sys.executable, "-c", "pass"], "cwd": str(tmp_path)}, + "provider": {"kind": "direct"}, + "execution": {"kind": "single"}, + }, + ) + assert result.is_error + assert result.structured_content["code"] == "EXECUTION_FAILURE" + outcome = result.structured_content["details"]["partial_evidence"]["capture"]["outcome"] + assert outcome["failed_count"] == 16 + + anyio.run(exercise) + + +@pytest.mark.unit +def test_experiment_classification_explicitly_describes_point_estimate() -> None: + design = ExperimentDesign( + cases=[ExperimentCase(name="base"), ExperimentCase(name="candidate")], + blocks=3, + seed=7, + metric="wall_time_ns", + estimand="mean_difference", + practical_threshold=10, + ) + executions = [ + dict(case=case, block=block, status="succeeded", wall_time_ns=value) + for block, candidate in enumerate([1000, 2000, 3000], 1) + for case, value in [("base", 2000), ("candidate", candidate)] + ] + blocks, _ = AnalysisRuntime._experiment_blocks(design, executions) + row = blocks[-1]["rows"][0] + assert row["estimate"] == 0 + assert row["confidence_low"] < -10 < 10 < row["confidence_high"] + assert row["point_estimate_classification"] == "within_threshold" + assert blocks[0]["values"]["decision_basis"] == "descriptive_point_estimate" + + +@pytest.mark.integration +def test_repository_corruption_has_path_free_recovery(tmp_path: Path) -> None: + store = tmp_path / "private-store" + store.mkdir() + (store / "existing-evidence").write_text("retain") + runtime = AnalysisRuntime(evidence_directory=store) + try: + with pytest.raises(RuntimeFailure) as failure: + runtime.query_evidence() + assert failure.value.code == "REPOSITORY_CORRUPTION" + details = failure.value.details + assert details["configuration_variable"] == "FLAMEOX_DATA_DIR" + assert "restore" in str(details).lower() + assert str(store) not in str(details) + assert (store / "existing-evidence").read_text() == "retain" + finally: + runtime.close() + + +@pytest.mark.process +@pytest.mark.skipif(os.name == "nt", reason="POSIX collector fixture") +def test_mcp_collector_failure_retains_profile_and_unknown_workload_status( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + collector = tmp_path / "collector" + collector.write_text( + f"#!{sys.executable}\n" + "import json,sys\n" + "from pathlib import Path\n" + "document = {'shared': {'frames': [{'name': 'work'}]}, " + "'profiles': [{'type': 'sampled', 'samples': [[0]], 'weights': [1.0]}]}\n" + "Path(sys.argv[sys.argv.index('--output')+1]).write_text(json.dumps(document))\n" + "print('collector could not reap child', file=sys.stderr)\n" + "sys.exit(1)\n" + ) + collector.chmod(0o755) + monkeypatch.setattr( + AnalysisRuntime, "_managed_executable", staticmethod(lambda _: str(collector)) + ) + + async def exercise() -> None: + async with Client(create_server(evidence_directory=tmp_path / "store")) as client: + result = await client.call_tool( + "capture_cpu_hotspots", + { + "target": {"argv": [sys.executable, "-c", "pass"], "cwd": str(tmp_path)}, + "provider": {"kind": "py-spy"}, + "execution": {"kind": "single"}, + "preserve": True, + }, + ) + assert result.is_error + partial = result.structured_content["details"]["partial_evidence"] + assert partial["capture"]["outcome"]["failed_count"] == 1 + execution = partial["capture"]["executions"][0] + assert execution["returncode"] == 1 + assert execution["returncode_scope"] == "collector" + assert execution["workload_returncode"] is None + resource = await client.read_resource(partial["preserved"]["uri"]) + assert isinstance(resource.contents[0], TextResourceContents) + projection = json.loads(resource.contents[0].text) + texts: list[str] = [] + for artifact in projection["body"]["artifacts"]: + if artifact["format"] == "text": + preview = await client.call_tool( + "preview_artifact", {"sources": [artifact["source"]]} + ) + texts.extend( + row["text"] for row in preview.structured_content["blocks"][1]["rows"] + ) + assert "collector could not reap child" in texts + assert partial["blocks"][1]["rows"] + + anyio.run(exercise) + + +@pytest.mark.integration +def test_directory_sources_round_trip_without_revealing_member_names(tmp_path: Path) -> None: + bundle = tmp_path / "private-bundle" + bundle.mkdir() + (bundle / "private-member.txt").write_text("one") + (bundle / "other.txt").write_text("two") + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + first = runtime.analyze( + "artifact.preview", [PathSource(path=str(bundle), format="text")], {} + ) + preserved = runtime.preserve_evidence(first["analysis_id"]) + resource = runtime.read_evidence_agent_projection(preserved["evidence_id"]) + assert "private-member" not in json.dumps(resource) + assert "private-bundle" not in json.dumps(resource) + assert len(resource["analysis_sources"]) == 1 + page = runtime.analyze( + "artifact.preview", + [EvidenceSource.model_validate(item) for item in resource["analysis_sources"]], + {}, + ) + assert page["blocks"][1]["rows"] == first["blocks"][1]["rows"] + finally: + runtime.close() + + +@pytest.mark.integration +@pytest.mark.parametrize("content", ["[]", "\n[]", "{}", '\n{"key": 1}', "[", "\n[{]"]) +def test_preview_json_empty_object_and_malformed_boundaries(tmp_path: Path, content: str) -> None: + artifact = tmp_path / "input.json" + artifact.write_text(content) + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + if content in {"[", "\n[{]"}: + with pytest.raises(RuntimeFailure) as failure: + runtime.analyze("artifact.preview", [PathSource(path=str(artifact))], {}) + assert failure.value.code == "DECODE_FAILURE" + else: + result = runtime.analyze("artifact.preview", [PathSource(path=str(artifact))], {}) + assert len(result["blocks"][1]["rows"]) == (1 if "key" in content else 0) + assert result["coverage"]["complete"] + finally: + runtime.close() diff --git a/tests/test_provider_preparation.py b/tests/test_provider_preparation.py new file mode 100644 index 0000000..1b24f52 --- /dev/null +++ b/tests/test_provider_preparation.py @@ -0,0 +1,276 @@ +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path +from types import SimpleNamespace + +import anyio +import pytest +from mcp import Client +from mcp_types import TextContent + +from flameox import __version__ +from flameox.mcp import create_server +from flameox.providers.preparation import PY_SPY_VERSION, ProviderDependencies +from flameox.runtime_contracts import PathSource, RuntimeFailure +from flameox.setup import SetupFailure, active_provider_status +from flameox.stateless import AnalysisRuntime + + +@pytest.mark.unit +@pytest.mark.parametrize( + ("release", "installed", "expected"), + [ + (__version__, "2.0", "ready"), + (__version__, "1.0", "restart_required"), + ("0.0.0", "2.0", "unknown"), + ], +) +def test_active_provider_contract_checks_release_and_requirements( + monkeypatch: pytest.MonkeyPatch, + release: str, + installed: str, + expected: str, +) -> None: + distribution = SimpleNamespace(version=release, requires=['memray>=2; extra == "memory"']) + monkeypatch.setattr( + "flameox.providers.environment.importlib.metadata.distribution", lambda _: distribution + ) + monkeypatch.setattr( + "flameox.providers.environment.importlib.metadata.version", lambda _: installed + ) + assert active_provider_status(["memray"]) == expected + + +@pytest.mark.integration +@pytest.mark.parametrize("status", ["ready", "restart_required", "unknown"]) +def test_mcp_preparation_reports_verified_or_conditional_handoff( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + status: str, +) -> None: + monkeypatch.setattr("flameox.providers.preparation.active_provider_status", lambda _: status) + + async def run(self: ProviderDependencies, argv: list[str], timeout: int) -> bytes: + return __version__.encode() + + monkeypatch.setattr(ProviderDependencies, "_run", run) + + async def exercise() -> None: + async with Client(create_server(evidence_directory=tmp_path / "store")) as client: + result = await client.call_tool( + "prepare_providers", {"provider_ids": ["memray", "perf"]} + ) + assert not result.is_error + value = result.structured_content + assert value["activation_status"] == status + if status == "ready": + assert value["next_action"] is None + else: + assert value["next_action"]["necessity"] == ( + "required" if status == "restart_required" else "conditional" + ) + assert "Preserve" in value["next_action"]["message"] + assert isinstance(result.content[0], TextContent) + assert "host readiness has not been verified" in result.content[0].text + + anyio.run(exercise) + + +def fake_collector_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + interpreter = sys.executable + server = tmp_path / "server" + server.mkdir() + server_python = server / "python" + server_python.symlink_to(interpreter) + monkeypatch.setattr("flameox.providers.preparation.sys.executable", str(server_python)) + collector = tmp_path / "collector" + collector.write_text(f"#!{interpreter}\nprint('py-spy {PY_SPY_VERSION}')\n") + collector.chmod(0o755) + uvx = tmp_path / "uvx" + receipt = json.dumps({"version": PY_SPY_VERSION, "executable": str(collector)}) + uvx.write_text(f"#!{interpreter}\nprint({receipt!r})\n") + uvx.chmod(0o755) + monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ.get("PATH", "")) + return collector + + +@pytest.mark.process +@pytest.mark.skipif(os.name == "nt", reason="POSIX executable fixtures") +def test_preparation_preserves_safe_uv_controls( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + controls = { + "UV_OFFLINE": "1", + "UV_CACHE_DIR": str(tmp_path / "cache"), + "UV_PYTHON_DOWNLOADS": "never", + "UV_NO_CONFIG": "1", + } + for key, value in controls.items(): + monkeypatch.setenv(key, value) + monkeypatch.setenv("PYTHONPATH", "/untrusted-import-path") + monkeypatch.setenv("UNRELATED_SECRET", "must-not-forward") + + async def exercise() -> None: + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + output = await runtime.dependencies._run( + [sys.executable, "-c", "import os,json; print(json.dumps(dict(os.environ)))"], 10 + ) + environment = json.loads(output) + assert {key: environment.get(key) for key in controls} == controls + assert "PYTHONPATH" not in environment + assert "UNRELATED_SECRET" not in environment + finally: + runtime.close() + + anyio.run(exercise) + + +@pytest.mark.process +@pytest.mark.skipif(os.name == "nt", reason="POSIX executable fixtures") +@pytest.mark.parametrize("failure", ["exit", "timeout", "cancel"]) +def test_mixed_preparation_commits_no_collector_on_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str +) -> None: + fake_collector_environment(tmp_path, monkeypatch) + monkeypatch.setattr("flameox.providers.preparation.active_provider_status", lambda _: "unknown") + original = ProviderDependencies._run + + async def run(self: ProviderDependencies, argv: list[str], timeout: int) -> bytes: + if argv[-1] == "--version" and argv[0] == "uvx": + if failure == "exit": + raise SetupFailure("server preparation failed") + await anyio.sleep(60) + return await original(self, argv, timeout) + + monkeypatch.setattr(ProviderDependencies, "_run", run) + + async def exercise() -> None: + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + if failure == "cancel": + with anyio.move_on_after(0.5) as scope: + await runtime.dependencies.prepare(["py-spy", "memray"]) + assert scope.cancel_called + else: + with pytest.raises(SetupFailure): + await runtime.dependencies.prepare(["py-spy", "memray"], timeout_seconds=1) + assert runtime.dependencies.py_spy_executable() is None + finally: + runtime.close() + + anyio.run(exercise) + + +@pytest.mark.process +@pytest.mark.skipif(os.name == "nt", reason="POSIX executable fixtures") +def test_prepare_collector_keeps_session_and_reuses_verified_binding( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + collector = fake_collector_environment(tmp_path, monkeypatch) + + async def exercise() -> None: + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + artifact = tmp_path / "input.txt" + artifact.write_text("retain session") + try: + analysis = runtime.analyze("artifact.preview", [PathSource(path=str(artifact))], {}) + first = await runtime.dependencies.prepare(["py-spy", "perf"]) + assert first.activation_status == "ready" + assert first.restart_required is False + assert first.preparation_command[5] == f"py-spy=={PY_SPY_VERSION}" + assert runtime._require_managed_executable("py-spy", "py-spy") == str(collector) + second = await runtime.dependencies.prepare(["py-spy"]) + assert second.preparation_command == [] + host = await runtime.dependencies.prepare(["perf"]) + assert host.activation_status == "not_applicable" + assert runtime.dependencies.py_spy_executable() == str(collector) + assert runtime.preserve_evidence(analysis["analysis_id"])["artifact_count"] == 1 + collector.write_text(collector.read_text() + "# changed\n") + with pytest.raises(RuntimeFailure, match="missing or changed"): + runtime.dependencies.py_spy_executable() + finally: + runtime.close() + + anyio.run(exercise) + + +@pytest.mark.process +@pytest.mark.skipif(os.name == "nt", reason="POSIX executable fixtures") +@pytest.mark.parametrize("server_succeeds", [False, True]) +def test_mixed_preparation_does_not_rollback_concurrent_success( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, server_succeeds: bool +) -> None: + collector = fake_collector_environment(tmp_path, monkeypatch) + monkeypatch.setattr("flameox.providers.preparation.active_provider_status", lambda _: "unknown") + original = ProviderDependencies._run + + async def exercise() -> None: + reached_server = anyio.Event() + release_server = anyio.Event() + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + + async def run(self: ProviderDependencies, argv: list[str], timeout: int) -> bytes: + if argv[-1] == "--version" and argv[0] == "uvx": + reached_server.set() + await release_server.wait() + if not server_succeeds: + raise SetupFailure("server preparation failed") + return __version__.encode() + return await original(self, argv, timeout) + + monkeypatch.setattr(ProviderDependencies, "_run", run) + + async def mixed_request() -> None: + if server_succeeds: + result = await runtime.dependencies.prepare(["py-spy", "memray"]) + assert result.prepared_managed_providers == ["py-spy", "memray"] + else: + with pytest.raises(SetupFailure): + await runtime.dependencies.prepare(["py-spy", "memray"]) + + try: + with anyio.fail_after(10): + async with anyio.create_task_group() as group: + group.start_soon(mixed_request) + await reached_server.wait() + assert runtime.dependencies.py_spy_executable() is None + await runtime.dependencies.prepare(["py-spy"]) + assert runtime.dependencies.py_spy_executable() == str(collector) + release_server.set() + assert runtime.dependencies.py_spy_executable() == str(collector) + finally: + runtime.close() + + anyio.run(exercise) + + +@pytest.mark.process +@pytest.mark.skipif(os.name == "nt", reason="POSIX executable fixtures") +def test_preparation_cancellation_cleans_up_and_does_not_activate( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + interpreter = sys.executable + fake_collector_environment(tmp_path, monkeypatch) + uvx = tmp_path / "uvx" + uvx.write_text(f"#!{interpreter}\nimport time\ntime.sleep(60)\n") + + async def exercise() -> None: + runtime = AnalysisRuntime(evidence_directory=tmp_path / "store") + try: + with anyio.move_on_after(0.2) as scope: + await runtime.dependencies.prepare(["py-spy"]) + assert scope.cancel_called + assert runtime.dependencies.py_spy_executable() is None + with pytest.raises(SetupFailure, match=r"timeout|complete"): + await runtime.dependencies.prepare(["py-spy"], timeout_seconds=1) + assert runtime.dependencies.py_spy_executable() is None + finally: + runtime.close() + + anyio.run(exercise) diff --git a/tests/test_setup_stateless.py b/tests/test_setup_stateless.py index 8a54089..323a033 100644 --- a/tests/test_setup_stateless.py +++ b/tests/test_setup_stateless.py @@ -102,7 +102,7 @@ def run(command: list[str], **kwargs: object) -> CompletedProcess[bytes]: assert prepared.requested_providers == ["memray", "py-spy"] assert prepared.prepared_managed_providers == ["memray", "py-spy"] assert prepared.preparation_status == "prepared" - assert prepared.restart_required is True + assert prepared.restart_required is (prepared.activation_status == "restart_required") assert calls == [ [ "/usr/bin/uvx", @@ -333,7 +333,7 @@ def test_opencode_jsonc_replaces_its_flameox_entry_in_place(tmp_path: Path) -> N jsonc.write_text( '{\n "mcp": {\n // Flameox notes\n' ' "flameox": {"type": "local", "command": ["custom"], "cwd": "/work"},\n' - ' },\n}\n' + " },\n}\n" ) plan = plan_client_setup([SetupClient.OPENCODE], [], home=tmp_path)[0] diff --git a/tests/test_stateless.py b/tests/test_stateless.py index 0731e68..4647161 100644 --- a/tests/test_stateless.py +++ b/tests/test_stateless.py @@ -3072,7 +3072,9 @@ def test_mcp_prepares_managed_providers_and_only_guides_host_tools( ) -> None: preparation_calls: list[list[str]] = [] - def prepare(provider_ids: list[str], timeout_seconds: int) -> ProviderPreparation: + async def prepare( + self: Any, provider_ids: list[str], timeout_seconds: int + ) -> ProviderPreparation: assert timeout_seconds == 1_800 if provider_ids == ["unknown-provider"]: raise ProviderSelectionFailure("Unknown provider 'unknown-provider'") @@ -3121,7 +3123,7 @@ def prepare(provider_ids: list[str], timeout_seconds: int) -> ProviderPreparatio ], ) - monkeypatch.setattr("flameox.mcp.server.provider_setup.prepare_providers", prepare) + monkeypatch.setattr("flameox.providers.preparation.ProviderDependencies.prepare", prepare) async def exercise() -> None: async with Client( @@ -3157,13 +3159,10 @@ async def exercise() -> None: } ] assert result.structured_content["preparation"]["status"] == "prepared" - assert result.structured_content["next_action"] == { - "kind": "reconnect_mcp", - "message": ( - "Reconnect Flameox with the returned launcher before retrying the capture; " - "the current server process is unchanged." - ), - } + assert result.structured_content["next_action"]["kind"] == "reconnect_mcp" + assert result.structured_content["next_action"]["necessity"] == "conditional" + assert "Preserve" in result.structured_content["next_action"]["message"] + assert "external requirements" in result.content[0].text assert result.structured_content["launcher"]["args"][3] == ( f"flameox[memory]=={__version__}" ) @@ -3621,7 +3620,7 @@ async def exercise() -> None: assert comparison["metric"] == "wall_time_ns" assert comparison["estimand"] == "median_difference" assert comparison["paired_blocks"] == 1 - assert comparison["decision"] in { + assert comparison["point_estimate_classification"] in { "practically_improved", "practically_regressed", "within_threshold", @@ -3751,7 +3750,7 @@ def test_experiment_zero_effect_is_within_zero_threshold() -> None: ) assert blocks[-1]["rows"][0]["estimate"] == 0 - assert blocks[-1]["rows"][0]["decision"] == "within_threshold" + assert blocks[-1]["rows"][0]["point_estimate_classification"] == "within_threshold" @pytest.mark.integration