diff --git a/.github/ISSUE_TEMPLATE/bug_report.md b/.github/ISSUE_TEMPLATE/bug_report.md deleted file mode 100644 index dc11cea..0000000 --- a/.github/ISSUE_TEMPLATE/bug_report.md +++ /dev/null @@ -1,30 +0,0 @@ ---- -name: Bug report -about: Report a bug in flameox -title: "" -labels: ["bug", "triage"] -assignees: [] ---- - -## Description - - - -## Steps to reproduce - -1. -2. -3. - -## Environment - -- flameox version (`flameox --version`): -- Python version (`python --version`): -- OS: -- MCP client (if applicable): - -## Relevant output - -```text - -``` diff --git a/.github/ISSUE_TEMPLATE/bug_report.yml b/.github/ISSUE_TEMPLATE/bug_report.yml new file mode 100644 index 0000000..aa7d68b --- /dev/null +++ b/.github/ISSUE_TEMPLATE/bug_report.yml @@ -0,0 +1,148 @@ +name: Bug report +description: Report a reproducible problem in flameox +title: "" +labels: + - bug + - triage +body: + - type: markdown + attributes: + value: | + Thanks for helping improve flameox. Please report one primary behavior per issue. + Keep observed facts separate from suspected causes, and remove secrets from logs. + + - type: checkboxes + id: duplicate-check + attributes: + label: Before submitting + options: + - label: I searched the open and closed issues for a related report. + required: true + - label: I have removed credentials, tokens, private paths, and other sensitive data. + required: true + + - type: textarea + id: observed-behavior + attributes: + label: Observed behavior + description: What happened? Include the affected flameox command, MCP tool, adapter, or artifact kind. + placeholder: Describe the behavior and the conditions under which it occurs. + validations: + required: true + + - type: textarea + id: reproduction + attributes: + label: Reproduction + description: Give the smallest realistic commands, configuration, input, or event sequence that demonstrates the problem. + placeholder: | + 1. Initialize or configure ... + 2. Run ... + 3. Observe ... + validations: + required: true + + - type: textarea + id: expected-behavior + attributes: + label: Expected behavior + description: What contract, invariant, output, or recovery action should have occurred instead? + validations: + required: true + + - type: textarea + id: impact-scope + attributes: + label: Impact and scope + description: Who or what is affected, and what is known not to be affected? Mention data loss, incorrect evidence, blocked recovery, or compatibility impact when relevant. + validations: + required: true + + - type: input + id: flameox-version + attributes: + label: flameox version + description: Use `flameox --version`, package metadata, or the source revision. + placeholder: 0.1.0 or git revision + validations: + required: true + + - type: input + id: python-version + attributes: + label: Python version + placeholder: Python 3.12.4 + validations: + required: true + + - type: input + id: platform + attributes: + label: Platform + description: Include OS, architecture, and relevant runtime or provider versions. + placeholder: Ubuntu 24.04 x86_64; Memray 1.20.0 + validations: + required: true + + - type: dropdown + id: interface + attributes: + label: Interface + options: + - CLI + - MCP + - Python API + - Storage or artifact format + - Other + validations: + required: true + + - type: dropdown + id: failure-stage + attributes: + label: Earliest failure stage + description: Select the earliest stage where the incorrect behavior is observable. + options: + - Request parsing or validation + - Capability discovery or provider setup + - Capture or import + - Artifact registration or provenance + - Extraction or derived catalog + - Analysis, comparison, or finding + - Recovery, cancellation, or lifecycle + - CLI or MCP transport projection + - Unknown + validations: + required: true + + - type: input + id: mcp-client + attributes: + label: MCP client (if applicable) + description: Include the client and version when the issue occurs through MCP; otherwise write not applicable. + placeholder: Client name and version + + - type: textarea + id: typed-outcome + attributes: + label: Typed outcome or recovery expected + description: Optional but useful for protocol and lifecycle problems. What result, error, recovery action, or state transition should the caller have received? + + - type: textarea + id: relevant-output + attributes: + label: Relevant output or evidence + description: Paste the smallest useful CLI output, error, MCP response, trace, or evidence excerpt. Redact sensitive values. + render: text + + - type: textarea + id: analysis + attributes: + label: Analysis or suspected cause + description: Optional. Clearly label source-derived reasoning or a hypothesis; do not present an unverified cause as fact. + + - type: textarea + id: additional-context + attributes: + label: Additional context + description: Links, reduced artifacts, screenshots, or a proposed regression test. Do not attach private evidence without removing sensitive data. diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml new file mode 100644 index 0000000..3ba13e0 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/config.yml @@ -0,0 +1 @@ +blank_issues_enabled: false diff --git a/.github/ISSUE_TEMPLATE/design_discussion.yml b/.github/ISSUE_TEMPLATE/design_discussion.yml new file mode 100644 index 0000000..7a4f57c --- /dev/null +++ b/.github/ISSUE_TEMPLATE/design_discussion.yml @@ -0,0 +1,67 @@ +name: Design discussion +description: Resolve an architectural or contract question before implementation +title: "" +labels: + - enhancement + - triage +body: + - type: markdown + attributes: + value: | + Use this for a substantial integration, a new public or persisted contract, + or a design choice with meaningful tradeoffs. Keep one decision per discussion. + + - type: checkboxes + id: duplicate-check + attributes: + label: Before submitting + options: + - label: I searched the open and closed issues and pull requests for a related discussion. + required: true + - label: I have removed credentials, tokens, private paths, and other sensitive data. + required: true + + - type: textarea + id: decision + attributes: + label: Decision needed + description: State the exact question maintainers need to resolve. + placeholder: Should flameox ... or ...? + validations: + required: true + + - type: textarea + id: context + attributes: + label: Context and evidence + description: Explain why this decision is needed now. Separate observed behavior, derived evidence, and inference. + validations: + required: true + + - type: textarea + id: options + attributes: + label: Options and tradeoffs + description: Compare a small number of viable approaches, including compatibility, safety, evidence, and maintenance costs. + validations: + required: true + + - type: textarea + id: constraints + attributes: + label: Constraints and non-goals + description: Identify authority boundaries, supported platforms, migration concerns, and intentionally excluded work. + + - type: textarea + id: success-criteria + attributes: + label: Success criteria and open questions + description: What facts or outcomes would settle the decision? List only unanswered questions that could change the choice. + validations: + required: true + + - type: textarea + id: references + attributes: + label: References + description: Link relevant source, tests, documentation, upstream tools, measurements, or prior issues. diff --git a/.github/ISSUE_TEMPLATE/feature_request.md b/.github/ISSUE_TEMPLATE/feature_request.md deleted file mode 100644 index caaba67..0000000 --- a/.github/ISSUE_TEMPLATE/feature_request.md +++ /dev/null @@ -1,23 +0,0 @@ ---- -name: Feature request -about: Propose an enhancement for flameox -title: "" -labels: ["enhancement", "triage"] -assignees: [] ---- - -## Problem - - - -## Proposed solution - - - -## Alternatives considered - - - -## Impact - - diff --git a/.github/ISSUE_TEMPLATE/feature_request.yml b/.github/ISSUE_TEMPLATE/feature_request.yml new file mode 100644 index 0000000..c2a5946 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/feature_request.yml @@ -0,0 +1,86 @@ +name: Feature request +description: Propose a focused improvement to flameox +title: "" +labels: + - enhancement + - triage +body: + - type: markdown + attributes: + value: | + Thanks for proposing an improvement. For substantial integrations or public/persisted + contract changes, describe the decision that needs maintainer agreement before implementation. + + - type: checkboxes + id: duplicate-check + attributes: + label: Before submitting + options: + - label: I searched the open and closed issues and pull requests for a related proposal. + required: true + - label: I have removed credentials, tokens, private paths, and other sensitive data. + required: true + + - type: dropdown + id: area + attributes: + label: Area + options: + - Adapters and providers + - Storage and evidence + - Investigations and comparisons + - CLI + - MCP + - Runtime safety + - Documentation + - Other + validations: + required: true + + - type: textarea + id: current-limitation + attributes: + label: Current limitation + description: What user or agent workflow is blocked, ambiguous, or unnecessarily difficult today? + validations: + required: true + + - type: textarea + id: use-cases + attributes: + label: Use cases + description: Give concrete callers, workflows, workloads, or examples. Explain why the current behavior is insufficient. + validations: + required: true + + - type: textarea + id: desired-outcome + attributes: + label: Desired outcome and success criteria + description: Describe observable behavior that would make this useful. Include evidence, compatibility, or performance criteria when they matter. + validations: + required: true + + - type: textarea + id: constraints + attributes: + label: Constraints and non-goals + description: Call out authority boundaries, supported platforms, persistence or compatibility concerns, and what should remain out of scope. + + - type: textarea + id: alternatives + attributes: + label: Alternatives considered + description: What other approaches, existing tools, or workarounds did you consider? Include why they do not meet the need. + + - type: textarea + id: possible-direction + attributes: + label: Possible direction + description: Optional proposal for implementation or integration. Treat this as a starting point, not a requirement. + + - type: textarea + id: additional-context + attributes: + label: Additional context + description: Add links to relevant documentation, upstream tools, measurements, or prior discussion. diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md index 9acba7c..9a3ebe7 100644 --- a/.github/pull_request_template.md +++ b/.github/pull_request_template.md @@ -5,23 +5,86 @@ Example: fix(storage): preserve comparison evidence The squash-merge commit and generated changelog use the PR title. --> -## Description +## Summary - + + +## Problem and expected behavior + + + +## Change + + + +## Contract and boundary impact + + + +- Semantic owner and changed stage: +- Public CLI or MCP contract: +- Storage, artifact, provenance, or schema contract: +- Adapter, provider, platform, or workload compatibility: +- Recovery, cancellation, concurrency, security, or containment impact: +- Native artifact, failed-attempt, and observed/derived/inferred claim handling: + +## Evidence and regression coverage + + + +- Tests added or updated: +- Base reproduction or other evidence: +- User-visible CLI or MCP output (if applicable): +- Remaining proof gaps: + +For performance or resource-budget claims, include the workload, baseline and +candidate, metric and units, platform, and measurement method. + +For changes that affect evidence or conclusions: + +- [ ] Observed, derived, and inferred claims remain distinguishable. +- [ ] Inputs, versions, provenance, and relevant corpus or artifact identity remain bound. +- [ ] Any compatibility, limitation, incompleteness, or uncertainty is exposed to callers. + +## Validation + + -## Approach +- `command` — result - +## Compatibility and safety -## Commands run + - -```console -uv run ruff check src tests tools -uv run mypy src tests tools -uv run pytest -q -``` +- Breaking changes or migration steps: +- Supported platform/provider changes: +- Security, privacy, or containment review: +- Performance or resource-budget impact: -## Compatibility +## Review checklist - +- [ ] The PR has one focused outcome and the title follows `type(scope): outcome`. +- [ ] Related issue is linked, or the reason for not linking one is stated above. +- [ ] Tests cover the changed observable behavior and meaningful failure path. +- [ ] Documentation or the owning contract is updated when behavior changed. +- [ ] User-visible CLI or MCP changes include a representative example or output. +- [ ] I checked the final diff for secrets, unrelated cleanup, and unsupported claims. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b8ba9be..f66da13 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -12,11 +12,11 @@ modification system. ## Before you start -Use the repository's issue templates for bugs and feature requests. Small fixes -and documentation improvements can usually go straight to a pull request. For a -substantial feature, new integration, or change to a public or persisted contract, -open an issue first so the intended behavior and contribution fit can be agreed -before implementation. +Use the repository's issue templates for bugs, feature requests, and design +discussions. Small fixes and documentation improvements can usually go straight +to a pull request. For a substantial feature, new integration, or change to a +public or persisted contract, open an issue first so the intended behavior and +contribution fit can be agreed before implementation. Search existing issues and pull requests before starting. If you discover a security vulnerability, follow [SECURITY.md](SECURITY.md) and report it privately diff --git a/docs/adapters.md b/docs/adapters.md index edeb40a..8426429 100644 --- a/docs/adapters.md +++ b/docs/adapters.md @@ -137,6 +137,17 @@ reader or export. Frame and allocation rows retain source artifact and run identity. Missing native symbols, thread identity, contexts, or native frames stay unavailable. Flameox does not infer them from display text. +### Node/V8 CPU and sampling heap profiles + +The Node adapters invoke the declared Node executable with `--cpu-prof` or +`--heap-prof`, preserving the native `.cpuprofile` or `.heapprofile` output even +when the workload exits nonzero. Node 20.16+ and 22.4+ are required. Extraction +runs in the bounded profile worker and streams the native JSON rather than +loading it into the control process. CPU tree nodes are visited once; heap +inclusive values include descendant allocations. Frame identity uses normalized +source coordinates, with a script disambiguator only when no URL is available. +Unresolved or synthetic frames remain partial symbolization. + ### Perfetto and OTLP Perfetto-compatible traces are queried through Trace Processor with versioned, diff --git a/docs/interfaces.md b/docs/interfaces.md index 593707f..317cb90 100644 --- a/docs/interfaces.md +++ b/docs/interfaces.md @@ -93,6 +93,9 @@ plan_capture └─ start_detached_capture long operation └─ get_detached_capture get_run → extract/analyze → get_evidence → record_analysis/record_finding + +V8 profile imports use `extract_node_cpu_prof` or `extract_node_heap_prof` after +the corresponding native artifact has been preserved. ``` Capability responses distinguish passive discovery from execution binding. diff --git a/src/flameox/adapters/__init__.py b/src/flameox/adapters/__init__.py index a33a94c..909905a 100644 --- a/src/flameox/adapters/__init__.py +++ b/src/flameox/adapters/__init__.py @@ -27,6 +27,8 @@ from flameox.adapters.setup_runtime import * # noqa: F403 from flameox.adapters.torch_profiler import * # noqa: F403 from flameox.adapters.toxiproxy import * # noqa: F403 + from flameox.adapters.v8_cpu_prof import * # noqa: F403 + from flameox.adapters.v8_heap_prof import * # noqa: F403 _MODULES = ( @@ -50,6 +52,8 @@ "registry", "setup_runtime", "torch_profiler", + "v8_cpu_prof", + "v8_heap_prof", "toxiproxy", ) @@ -135,6 +139,10 @@ "TraceProcessorInstallation", "TraceWindowResult", "TritonCompilerOptions", + "V8CpuProfExtractionResult", + "V8CpuProfExtractor", + "V8HeapProfExtractionResult", + "V8HeapProfExtractor", "VllmAggregateMetrics", "VllmMeasurementRow", "VllmResultDocument", diff --git a/src/flameox/adapters/builtins.py b/src/flameox/adapters/builtins.py index cde209a..4e0c7f3 100644 --- a/src/flameox/adapters/builtins.py +++ b/src/flameox/adapters/builtins.py @@ -1,11 +1,14 @@ from __future__ import annotations import json +import re import sys from dataclasses import dataclass from enum import StrEnum from pathlib import Path +from packaging.version import InvalidVersion, Version + from flameox.adapters.options import ( adapter_accepts_options, compute_sanitizer_options, @@ -58,6 +61,24 @@ class CaptureInvocation: environment: dict[str, str] +def node_version_is_supported(value: str | None) -> bool: + """Return whether Node exposes the stable V8 profile flags used by these adapters.""" + if value is None: + return False + match = re.search(r"\d+(?:\.\d+)+", value) + if match is None: + return False + try: + version = Version(match.group()) + except InvalidVersion: + return False + return ( + (version.major == 20 and version >= Version("20.16")) + or (version.major == 22 and version >= Version("22.4")) + or version.major > 22 + ) + + BUILTIN_ADAPTERS = { adapter.name: adapter for adapter in ( @@ -72,6 +93,49 @@ class CaptureInvocation: expected_overhead="No profiler overhead; process output only.", capture_limitations=("No sampled stack or operator evidence is collected.",), ), + BuiltinAdapter( + name="node-cpu-prof", + dependency_kind=AdapterDependencyKind.EXECUTABLE, + dependency="node", + supported_modes=("record",), + supported_formats=("v8-cpuprofile",), + features=("sampled_stacks", "javascript_symbols"), + remediation=("Install Node.js 20.16+ or 22.4+ which expose stable --cpu-prof flags.",), + version_args=("--version",), + output_filename="cpu.cpuprofile", + artifact_kinds=(ArtifactKind.SAMPLE_PROFILE,), + expected_overhead=( + "V8 CPU sampling overhead; exact rate depends on --cpu-prof-interval." + ), + capture_limitations=( + "Only the main Node.js thread is profiled; worker threads are not sampled.", + "The CPU profile contains sampled stack locations, not wall-clock or " + "allocation evidence.", + ), + preserve_artifact_on_nonzero=True, + ), + BuiltinAdapter( + name="node-heap-prof", + dependency_kind=AdapterDependencyKind.EXECUTABLE, + dependency="node", + supported_modes=("record",), + supported_formats=("v8-sampling-heap-profile",), + features=("allocations", "sampled_allocations", "stacks"), + remediation=("Install Node.js 20.16+ or 22.4+ which expose stable --heap-prof flags.",), + version_args=("--version",), + output_filename="heap.heapprofile", + artifact_kinds=(ArtifactKind.MEMORY_PROFILE,), + expected_overhead=( + "V8 heap sampling overhead; exact rate depends on --heap-prof-interval." + ), + capture_limitations=( + "Sampled allocation bytes are an estimate, not the exact retained heap or " + "process RSS.", + "Only allocations sampled by V8 are reported; small or short-lived " + "allocations may be underrepresented.", + ), + preserve_artifact_on_nonzero=True, + ), BuiltinAdapter( name="benchmark-samples", dependency_kind=AdapterDependencyKind.INTERNAL, @@ -453,6 +517,7 @@ def build_capture_invocation( # noqa: C901 - provider routing is intentionally timeout_seconds: float = 300, options: dict[str, object] | None = None, project_root: Path | None = None, + workload_executable: str | None = None, ) -> CaptureInvocation: adapter = BUILTIN_ADAPTERS.get(adapter_name) if ( @@ -520,6 +585,16 @@ def build_capture_invocation( # noqa: C901 - provider routing is intentionally output, *target, ) + elif adapter_name in {"node-cpu-prof", "node-heap-prof"}: + return _node_v8_capture_invocation( + adapter_name, + adapter, + workload_argv, + output_root, + output, + executable=executable, + workload_executable=workload_executable, + ) elif adapter_name == "torch.profiler": return _torch_capture_invocation( adapter, @@ -742,6 +817,71 @@ def _torch_capture_invocation( ) +def _node_v8_capture_invocation( + adapter_name: str, + adapter: BuiltinAdapter, + workload_argv: tuple[str, ...], + output_root: Path, + output: str, + *, + executable: str | None, + workload_executable: str | None, +) -> CaptureInvocation: + """Inject Node.js --cpu-prof or --heap-prof flags into a declared Node workload. + + Node.js exposes stable V8 profiling through CLI flags (Node 20.16+ / 22.4+). + The declared workload argv already starts with the Node executable, so the + adapter inserts the profiling flags immediately after argv[0] and before the + user script and its arguments. The output directory and file name are bound + explicitly so Flameox owns the artifact path and can preserve it. + """ + if not workload_argv: + raise DomainError( + ErrorCode.INVALID_CAPTURE_PLAN, + "A declared Node.js workload command is required for V8 profiling.", + ) + node_executable = workload_executable or workload_argv[0] + node_name = Path(node_executable).name + if not ( + node_name == "node" or node_name.startswith("node") or node_executable.endswith("node") + ): + raise DomainError( + ErrorCode.INVALID_CAPTURE_PLAN, + f"V8 profiling requires a Node.js workload; the declared " + f"executable is {node_executable!r}.", + remediation=("Declare a Node.js command (e.g. `node script.js`) as the workload.",), + ) + if len(workload_argv) < 2: + raise DomainError( + ErrorCode.INVALID_CAPTURE_PLAN, + "A declared Node.js script or module is required after the node executable.", + ) + # Compute the directory and file name for the V8 profile output. + output_path = Path(output) + prof_dir = str(output_path.parent) + prof_name = output_path.name + if adapter_name == "node-cpu-prof": + prof_flags = ( + "--cpu-prof", + "--cpu-prof-dir=" + prof_dir, + "--cpu-prof-name=" + prof_name, + ) + else: + prof_flags = ( + "--heap-prof", + "--heap-prof-dir=" + prof_dir, + "--heap-prof-name=" + prof_name, + ) + argv = (node_executable, *prof_flags, *workload_argv[1:]) + return CaptureInvocation( + argv=argv, + artifact_kinds=adapter.artifact_kinds, + expected_overhead=adapter.expected_overhead or "", + limitations=adapter.capture_limitations, + environment={}, + ) + + def _compute_sanitizer_capture_invocation( adapter: BuiltinAdapter, workload_argv: tuple[str, ...], diff --git a/src/flameox/adapters/v8_cpu_prof.py b/src/flameox/adapters/v8_cpu_prof.py new file mode 100644 index 0000000..0991fc1 --- /dev/null +++ b/src/flameox/adapters/v8_cpu_prof.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Literal + +from flameox.adapters.artifact_workers import IsolatedWorkerHarness +from flameox.domain import ArtifactKind, ArtifactRegistration, DomainError, ErrorCode, digest_model +from flameox.evidence import GenerationPublisher +from flameox.models import ContractModel +from flameox.storage import ArtifactStore, RunStore, Workspace +from flameox.workers.v8_profiles_contract import ( + V8_PROFILE_WORKER, + V8ProfileRequest, + V8ProfileResult, +) + + +class V8CpuProfExtractionResult(ContractModel): + schema_version: int = 1 + run_id: str + artifact_id: str + node_count: int + sample_count: int + frame_count: int + corpus_commit_id: str + limitations: tuple[str, ...] = () + + +class V8CpuProfExtractor: + name = "node-cpu-prof" + version = "1" + + def __init__(self, workspace: Workspace) -> None: + self.workspace = workspace + self.publisher = GenerationPublisher(workspace) + + def extract(self, run_id: str) -> V8CpuProfExtractionResult: + registration = _registration(self.workspace, run_id, ArtifactKind.SAMPLE_PROFILE, "CPU") + artifact = ArtifactStore(self.workspace).get(registration.artifact_id) + response = _run_worker( + self.workspace, artifact.payload_path, registration.artifact_id, "cpu" + ) + measurement_rows = _frame_measurements(run_id, registration.artifact_id, response) + published = self.publisher.publish_rows_idempotent( + { + "measurements": [ + _measurement( + run_id, + registration.artifact_id, + "cpu.samples", + response.sample_count, + "count", + ), + _measurement( + run_id, registration.artifact_id, "cpu.nodes", response.node_count, "count" + ), + ], + "frames": list(response.frames), + "frame_measurements": measurement_rows, + }, + publisher=self.name, + publisher_version=self.version, + input_run_ids=(run_id,), + input_artifact_ids=(registration.artifact_id,), + operation_identity={"profile_kind": "cpu"}, + ) + return V8CpuProfExtractionResult( + run_id=run_id, + artifact_id=registration.artifact_id, + node_count=response.node_count, + sample_count=response.sample_count, + frame_count=len(response.frames), + corpus_commit_id=published.commit.commit_id, + limitations=response.limitations, + ) + + +def _registration( + workspace: Workspace, + run_id: str, + kind: ArtifactKind, + label: str, +) -> ArtifactRegistration: + run = RunStore(workspace).read(run_id) + registrations = [item for item in run.artifacts if item.kind is kind] + if len(registrations) != 1: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + f"The run must contain exactly one V8 {label} profile artifact.", + run_id=run_id, + ) + return registrations[0] + + +def _run_worker( + workspace: Workspace, + artifact_path: Path, + artifact_id: str, + profile_kind: Literal["cpu", "heap"], +) -> V8ProfileResult: + maximum = workspace.config.storage.max_rows_per_generation + response_budget = workspace.config.execution.max_output_bytes + max_rows = min(maximum - 2, max(2, (response_budget - 64 * 1024) // 2_048)) + if maximum < 4 or max_rows < 2: + raise DomainError( + ErrorCode.QUERY_BUDGET_EXCEEDED, + "V8 profile extraction requires room for bounded frame rows.", + ) + return IsolatedWorkerHarness(workspace).run_typed_sync( + V8_PROFILE_WORKER, + V8ProfileRequest( + profile_kind=profile_kind, + artifact_path=str(artifact_path), + artifact_id=artifact_id, + project_root=str(workspace.project_root), + max_nodes=min(100_000, max_rows // 2), + max_samples=min(1_000_000, maximum), + max_rows=max_rows, + ), + ) + + +def _measurement( + run_id: str, + artifact_id: str, + name: str, + value: int, + unit: str, +) -> dict[str, object]: + return { + "measurement_id": digest_model( + {"run_id": run_id, "artifact_id": artifact_id, "name": name} + ), + "run_id": run_id, + "artifact_id": artifact_id, + "name": name, + "value_int": value, + "value_float": None, + "unit": unit, + "aggregation": "total", + "scope": "process", + "trial_id": None, + "worker_id": None, + "worker_run_index": None, + "value_index": None, + "loop_count": None, + "is_warmup": False, + "block_id": None, + "variant_id": None, + "order_in_block": None, + "phase": None, + "dimensions": {}, + "evidence_level": "observed", + } + + +def _frame_measurements( + run_id: str, + artifact_id: str, + response: V8ProfileResult, +) -> list[dict[str, object]]: + return [ + { + **dict(row), + "run_id": run_id, + "artifact_id": artifact_id, + "thread_name": None, + "process_name": None, + "phase": None, + } + for row in response.frame_measurements + ] diff --git a/src/flameox/adapters/v8_heap_prof.py b/src/flameox/adapters/v8_heap_prof.py new file mode 100644 index 0000000..4f8c07d --- /dev/null +++ b/src/flameox/adapters/v8_heap_prof.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from flameox.adapters.v8_cpu_prof import ( + _frame_measurements, + _measurement, + _registration, + _run_worker, +) +from flameox.domain import ArtifactKind +from flameox.evidence import GenerationPublisher +from flameox.models import ContractModel +from flameox.storage import ArtifactStore, Workspace + + +class V8HeapProfExtractionResult(ContractModel): + schema_version: int = 1 + run_id: str + artifact_id: str + sample_count: int + total_sampled_bytes: int + frame_count: int + corpus_commit_id: str + limitations: tuple[str, ...] = () + + +class V8HeapProfExtractor: + name = "node-heap-prof" + version = "1" + + def __init__(self, workspace: Workspace) -> None: + self.workspace = workspace + self.publisher = GenerationPublisher(workspace) + + def extract(self, run_id: str) -> V8HeapProfExtractionResult: + registration = _registration(self.workspace, run_id, ArtifactKind.MEMORY_PROFILE, "heap") + artifact = ArtifactStore(self.workspace).get(registration.artifact_id) + response = _run_worker( + self.workspace, artifact.payload_path, registration.artifact_id, "heap" + ) + published = self.publisher.publish_rows_idempotent( + { + "measurements": [ + _measurement( + run_id, + registration.artifact_id, + "memory.sampled_bytes", + response.total_sampled_bytes, + "bytes", + ), + _measurement( + run_id, + registration.artifact_id, + "memory.samples", + response.sample_count, + "count", + ), + ], + "frames": list(response.frames), + "frame_measurements": _frame_measurements( + run_id, registration.artifact_id, response + ), + }, + publisher=self.name, + publisher_version=self.version, + input_run_ids=(run_id,), + input_artifact_ids=(registration.artifact_id,), + operation_identity={"profile_kind": "heap"}, + ) + return V8HeapProfExtractionResult( + run_id=run_id, + artifact_id=registration.artifact_id, + sample_count=response.sample_count, + total_sampled_bytes=response.total_sampled_bytes, + frame_count=len(response.frames), + corpus_commit_id=published.commit.commit_id, + limitations=response.limitations, + ) diff --git a/src/flameox/application/capabilities.py b/src/flameox/application/capabilities.py index f774acc..c57a5eb 100644 --- a/src/flameox/application/capabilities.py +++ b/src/flameox/application/capabilities.py @@ -30,6 +30,7 @@ AdapterDependencyKind, BuiltinAdapter, builtin_adapter, + node_version_is_supported, ) from flameox.adapters.nsight_compute import find_ncu_report_interface from flameox.adapters.registry import AdapterRegistry @@ -607,6 +608,22 @@ async def probe(self, adapter: str, *, refresh: bool = False) -> CapabilityRepor if version_report.status is not CapabilityStatus.AVAILABLE: self._active_cache[adapter] = version_report return version_report + if adapter in {"node-cpu-prof", "node-heap-prof"} and not node_version_is_supported( + version_report.version + ): + incompatible = version_report.validated_copy( + update={ + "status": CapabilityStatus.UNAVAILABLE, + "supported_modes": (), + "supported_formats": (), + "limitations": ( + "This Node.js version does not expose the stable V8 profiling flags; " + "Node.js 20.16+ or 22.4+ is required.", + ), + } + ) + self._active_cache[adapter] = incompatible + return incompatible if adapter == "perf": result = await self._probe_perf(passive) result = result.validated_copy(update={"version": version_report.version}) @@ -624,6 +641,29 @@ async def probe(self, adapter: str, *, refresh: bool = False) -> CapabilityRepor self._active_cache[adapter] = result return result + async def probe_executable_version(self, executable: str, *, cwd: Path) -> str: + """Probe the exact executable bound to a declared workload.""" + binding = ExecutableResolver().require_host_tool(executable, cwd=cwd) + outcome = await self.broker.run( + ExecutionRequest( + argv=(str(binding.invocation_path), "--version"), + executable_binding=binding, + cwd=cwd, + environment_allowlist=(), + allowed_working_roots=(cwd,), + timeout_seconds=5, + max_output_bytes=32 * 1024, + ) + ) + output = (outcome.stdout + b"\n" + outcome.stderr).decode("utf-8", errors="replace") + version = next((line.strip() for line in output.splitlines() if line.strip()), None) + if outcome.process.exit_code != 0 or version is None: + raise DomainError( + ErrorCode.CAPABILITY_UNAVAILABLE, + "The declared Node.js executable did not return a version.", + ) + return version + async def _probe_version( self, passive: CapabilityReport, diff --git a/src/flameox/application/capture.py b/src/flameox/application/capture.py index 66caeb1..56dab1e 100644 --- a/src/flameox/application/capture.py +++ b/src/flameox/application/capture.py @@ -26,6 +26,7 @@ AdapterDependencyKind, build_capture_invocation, builtin_adapter, + node_version_is_supported, replace_compute_sanitizer_suppression, ) from flameox.adapters.compute_sanitizer import ( @@ -644,6 +645,7 @@ async def plan( output_root, capability=adapter_capability, options=bound_adapter_options, + workload_executable=str(instance.executable_binding.invocation_path), ) collector_environment.update(adapter_binding.environment) collector_argv = adapter_binding.argv @@ -2692,6 +2694,7 @@ async def _adapter_command( *, capability: CapabilityReport | None = None, options: dict[str, JsonValue] | None = None, + workload_executable: str | None = None, ) -> _AdapterBinding: adapter_definition = builtin_adapter(adapter) if adapter_definition is not None: @@ -2715,6 +2718,7 @@ async def _adapter_command( timeout_seconds=workload.timeout_seconds, options=cast(dict[str, object] | None, options), project_root=self.workspace.project_root, + workload_executable=workload_executable, ) if invocation.environment: conflicts = { @@ -2736,6 +2740,23 @@ async def _adapter_command( "to the identical value required by the adapter.", ), ) + version = capability.version + if adapter in {"node-cpu-prof", "node-heap-prof"}: + if workload_executable is None: + raise DomainError( + ErrorCode.INVALID_CAPTURE_PLAN, + "The V8 capture plan is missing the declared Node executable.", + ) + version = await self.capabilities.probe_executable_version( + workload_executable, + cwd=Path(workload.cwd), + ) + if not node_version_is_supported(version): + raise DomainError( + ErrorCode.ADAPTER_INCOMPATIBLE, + "The declared Node.js executable is too old for stable V8 profiling.", + remediation=("Use Node.js 20.16+ or 22.4+ for V8 profiling.",), + ) return _AdapterBinding( argv=invocation.argv, artifact_kinds=invocation.artifact_kinds, @@ -2747,7 +2768,7 @@ async def _adapter_command( for message in invocation.limitations ), permissions=adapter_definition.permissions, - version=capability.version, + version=version, ) registry = AdapterRegistry(self.workspace) diff --git a/src/flameox/application/operations.py b/src/flameox/application/operations.py index 0a727a5..54aea69 100644 --- a/src/flameox/application/operations.py +++ b/src/flameox/application/operations.py @@ -947,6 +947,15 @@ async def cancel(self, operation_id: str) -> OperationStatus: task = self.tasks.get(operation_id) if task is not None: await task.wait() + elif event is not None: + # The task handle can disappear just before _execute commits its terminal + # record. We still own the cancellation event in that small window, so + # wait for finalization instead of projecting an active run as unmanaged. + for _ in range(500): + current = self.store.read(operation_id) + if not isinstance(current, ActiveOperationRecord): + break + await anyio.sleep(0.01) return OperationStatus.from_record( self.store.read(operation_id), adapter=self.adapter, diff --git a/src/flameox/cli.py b/src/flameox/cli.py index 27328f0..d29f0c7 100644 --- a/src/flameox/cli.py +++ b/src/flameox/cli.py @@ -33,6 +33,8 @@ PytestExtractor, PythonStartupExtractor, SetupClient, + V8CpuProfExtractor, + V8HeapProfExtractor, ) from flameox.analysis import RecipeService from flameox.application import ( @@ -2374,6 +2376,34 @@ def extract_benchmark_samples( _emit(result, as_json=json_output) +@extract_app.command("node-cpu-prof") +def extract_node_cpu_prof( + run_id: Annotated[str, typer.Argument(help="Run containing a V8 .cpuprofile artifact.")], + workspace: WorkspaceOption = None, + json_output: JsonOption = False, +) -> None: + """Extract bounded evidence from a Node/V8 CPU profile.""" + try: + result = V8CpuProfExtractor(_workspace(workspace)).extract(run_id) + except DomainError as error: + _fail(error) + _emit(result, as_json=json_output) + + +@extract_app.command("node-heap-prof") +def extract_node_heap_prof( + run_id: Annotated[str, typer.Argument(help="Run containing a V8 .heapprofile artifact.")], + workspace: WorkspaceOption = None, + json_output: JsonOption = False, +) -> None: + """Extract bounded evidence from a Node/V8 sampling heap profile.""" + try: + result = V8HeapProfExtractor(_workspace(workspace)).extract(run_id) + except DomainError as error: + _fail(error) + _emit(result, as_json=json_output) + + @extract_app.command("kernel-validation") def extract_kernel_validation( run_id: Annotated[ diff --git a/src/flameox/mcp/server.py b/src/flameox/mcp/server.py index bbf19b3..de19ec5 100644 --- a/src/flameox/mcp/server.py +++ b/src/flameox/mcp/server.py @@ -74,6 +74,10 @@ PythonStartupExtractor, TorchProfilerCaptureOptions, TraceWindowResult, + V8CpuProfExtractionResult, + V8CpuProfExtractor, + V8HeapProfExtractionResult, + V8HeapProfExtractor, ) from flameox.analysis import ( AcceleratorLaunchAnalysisResult, @@ -3365,6 +3369,44 @@ async def extract_benchmark_samples_tool( except DomainError as error: return _failure(error) + @server.tool(name="extract_node_cpu_prof", annotations=ADDITIVE) + async def extract_node_cpu_prof_tool( + run_id: Annotated[str, Field(min_length=1, max_length=200)], + ctx: Context[AppContext], + ) -> Annotated[CallToolResult, ToolPayload[V8CpuProfExtractionResult]]: + """Extract bounded evidence from a Node/V8 CPU profile.""" + try: + result = await run_atomic_thread( + lambda: V8CpuProfExtractor( + ctx.request_context.lifespan_context.require_workspace() + ).extract(run_id) + ) + return _success( + result, + f"Extracted {result.frame_count} V8 CPU profile frames.", + ) + except DomainError as error: + return _failure(error) + + @server.tool(name="extract_node_heap_prof", annotations=ADDITIVE) + async def extract_node_heap_prof_tool( + run_id: Annotated[str, Field(min_length=1, max_length=200)], + ctx: Context[AppContext], + ) -> Annotated[CallToolResult, ToolPayload[V8HeapProfExtractionResult]]: + """Extract bounded evidence from a Node/V8 sampling heap profile.""" + try: + result = await run_atomic_thread( + lambda: V8HeapProfExtractor( + ctx.request_context.lifespan_context.require_workspace() + ).extract(run_id) + ) + return _success( + result, + f"Extracted {result.frame_count} V8 heap profile frames.", + ) + except DomainError as error: + return _failure(error) + @server.tool(name="extract_kernel_validation", annotations=ADDITIVE) async def extract_kernel_validation_tool( run_id: Annotated[str, Field(min_length=1, max_length=200)], diff --git a/src/flameox/workers/protocol.py b/src/flameox/workers/protocol.py index cdb5c97..c16b108 100644 --- a/src/flameox/workers/protocol.py +++ b/src/flameox/workers/protocol.py @@ -26,6 +26,7 @@ class WorkerOperationId(StrEnum): OTLP_PARSE = "otlp.parse" PERFETTO_QUERY = "perfetto.query" REDUCTION_EXECUTE = "reduction.execute" + V8_PROFILE_PARSE = "v8_profile.parse" class WorkerFailureKind(StrEnum): diff --git a/src/flameox/workers/v8_profiles.py b/src/flameox/workers/v8_profiles.py new file mode 100644 index 0000000..eadc2a5 --- /dev/null +++ b/src/flameox/workers/v8_profiles.py @@ -0,0 +1,368 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any, NoReturn, cast +from urllib.parse import unquote, urlparse + +import ijson + +from flameox.domain import DomainError, ErrorCode, digest_model +from flameox.workers.protocol import WorkerApplication, WorkerFailureKind, run_typed_worker +from flameox.workers.v8_profiles_contract import ( + V8_PROFILE_WORKER, + V8ProfileRequest, + V8ProfileResult, +) + + +def _parse_cpu(request: V8ProfileRequest) -> V8ProfileResult: + nodes: dict[int, dict[str, Any]] = {} + top_level = _top_level_keys(Path(request.artifact_path)) + if "nodes" not in top_level or "samples" not in top_level: + _malformed("V8 CPU profile must contain nodes and samples arrays.") + with Path(request.artifact_path).open("rb") as stream: + for node in ijson.items(stream, "nodes.item"): + if len(nodes) >= request.max_nodes: + _limit("V8 CPU profile node limit exceeded.") + if not isinstance(node, dict): + _malformed("V8 CPU profile node must be an object.") + node_id = _strict_int(node.get("id"), "node id") + if node_id in nodes: + _malformed("V8 CPU profile contains duplicate node IDs.") + call_frame = _call_frame(node.get("callFrame")) + children = node.get("children", []) + if not isinstance(children, list): + _malformed("V8 CPU profile node children must be an array.") + child_ids = tuple(_strict_int(value, "child node id") for value in children) + hit_count = _strict_int(node.get("hitCount", 0), "hit count") + if hit_count < 0: + _malformed("V8 CPU profile hit count cannot be negative.") + nodes[node_id] = { + "call_frame": call_frame, + "children": child_ids, + "hit_count": hit_count, + } + if not nodes: + _malformed("V8 CPU profile nodes array cannot be empty.") + sample_count = _count_cpu_samples(Path(request.artifact_path), request.max_samples) + return _aggregate_cpu(request, nodes, sample_count) + + +def _count_cpu_samples(path: Path, limit: int) -> int: + count = 0 + with path.open("rb") as stream: + for sample in ijson.items(stream, "samples.item"): + if count >= limit: + _limit("V8 CPU profile sample limit exceeded.") + _strict_int(sample, "sample node id") + count += 1 + return count + + +def _aggregate_cpu( + request: V8ProfileRequest, + nodes: dict[int, dict[str, Any]], + sample_count: int, +) -> V8ProfileResult: + parents: dict[int, int] = {} + for node_id, node in nodes.items(): + for child_id in node["children"]: + if child_id not in nodes: + _malformed("V8 CPU profile references an unknown child node.") + if child_id in parents: + _malformed("V8 CPU profile contains a node with multiple parents.") + parents[child_id] = node_id + roots = tuple(node_id for node_id in nodes if node_id not in parents) + if not roots and nodes: + _malformed("V8 CPU profile node tree contains a cycle.") + + frame_rows: dict[str, dict[str, Any]] = {} + aggregates: dict[str, dict[str, int]] = {} + visited: set[int] = set() + subtree_totals: dict[int, int] = {} + for root in roots: + stack: list[tuple[int, bool]] = [(root, False)] + while stack: + node_id, closing = stack.pop() + if closing: + node = nodes[node_id] + identity = _frame_identity(node["call_frame"], request) + subtree = node["hit_count"] + sum( + subtree_totals[child_id] for child_id in node["children"] + ) + values = aggregates.setdefault( + identity["frame_id"], {"self": 0, "inclusive": 0, "samples": 0} + ) + values["self"] += node["hit_count"] + values["inclusive"] += subtree + values["samples"] += node["hit_count"] + subtree_totals[node_id] = subtree + frame_rows.setdefault(identity["frame_id"], identity) + continue + if node_id in visited: + _malformed("V8 CPU profile tree contains a repeated node.") + visited.add(node_id) + node = nodes[node_id] + stack.append((node_id, True)) + for child_id in reversed(node["children"]): + stack.append((child_id, False)) + + if len(visited) != len(nodes): + _malformed("V8 CPU profile contains disconnected or unreachable nodes.") + _check_rows(request, len(frame_rows), len(aggregates)) + return V8ProfileResult( + profile_kind="cpu", + node_count=len(nodes), + sample_count=sample_count, + frames=tuple(frame_rows.values()), + frame_measurements=tuple( + { + "frame_id": frame_id, + "metric": "cpu.hit_count", + "self_value": values["self"], + "inclusive_value": values["inclusive"], + "unit": "count", + "sample_count": values["samples"], + } + for frame_id, values in sorted(aggregates.items()) + ), + limitations=( + "V8 CPU samples represent execution time, not allocation or memory evidence.", + "The CPU profile contains sampled stack locations; source-map resolution is not " + "applied.", + ), + ) + + +def _parse_heap(request: V8ProfileRequest) -> V8ProfileResult: # noqa: C901 - streaming validation + path = Path(request.artifact_path) + top_level = _top_level_keys(path) + if "head" not in top_level or "samples" not in top_level: + _malformed("V8 heap profile must contain head and samples.") + frame_rows: dict[str, dict[str, Any]] = {} + aggregates: dict[str, dict[str, int]] = {} + node_count = 0 + sample_count = 0 + total_sampled_bytes = 0 + node_stack: list[dict[str, Any]] = [] + sample: dict[str, Any] | None = None + with path.open("rb") as stream: + for prefix, event, value in ijson.parse(stream): + if event == "start_map" and (prefix == "head" or prefix.endswith(".children.item")): + if len(node_stack) >= request.max_nodes: + _limit("V8 heap profile node limit exceeded.") + node_stack.append( + {"call_frame": None, "self_size": None, "children": False, "child_total": 0} + ) + continue + if event == "start_map" and prefix == "samples.item": + sample = {} + continue + if prefix == "samples.item" and event not in {"start_map", "end_map", "map_key"}: + _malformed("V8 heap sample must be an object.") + if node_stack: + current = node_stack[-1] + if event == "start_array" and prefix.endswith(".children"): + current["children"] = True + elif event in {"string", "number", "null"}: + if prefix.endswith(".selfSize"): + current["self_size"] = value + elif prefix.endswith(".callFrame.functionName"): + current.setdefault("call_frame_values", {})["functionName"] = value + elif prefix.endswith(".callFrame.url"): + current.setdefault("call_frame_values", {})["url"] = value + elif prefix.endswith(".callFrame.lineNumber"): + current.setdefault("call_frame_values", {})["lineNumber"] = value + elif prefix.endswith(".callFrame.columnNumber"): + current.setdefault("call_frame_values", {})["columnNumber"] = value + elif prefix.endswith(".callFrame.scriptId"): + current.setdefault("call_frame_values", {})["scriptId"] = value + if ( + sample is not None + and event in {"string", "number", "null"} + and prefix.endswith((".size", ".nodeId")) + ): + sample[prefix.rsplit(".", 1)[-1]] = value + if event == "end_map" and prefix == "samples.item": + if sample is None: + _malformed("V8 heap sample is malformed.") + if sample_count >= request.max_samples: + _limit("V8 heap profile sample limit exceeded.") + size = _strict_int(sample.get("size"), "sample size") + _strict_int(sample.get("nodeId"), "sample node id") + if size < 0: + _malformed("V8 heap sample size cannot be negative.") + total_sampled_bytes += size + sample_count += 1 + sample = None + continue + if ( + event == "end_map" + and node_stack + and (prefix == "head" or prefix.endswith(".children.item")) + ): + current = node_stack.pop() + call_frame = current.get("call_frame_values") + if not isinstance(call_frame, dict) or not current["children"]: + _malformed("V8 heap profile node is missing callFrame or children.") + self_size = _strict_int(current["self_size"], "self size") + if self_size < 0: + _malformed("V8 heap profile selfSize cannot be negative.") + if node_stack: + node_stack[-1]["child_total"] += self_size + current["child_total"] + identity = _frame_identity(call_frame, request) + frame_rows.setdefault(identity["frame_id"], identity) + values = aggregates.setdefault( + identity["frame_id"], {"self": 0, "inclusive": 0, "samples": 0} + ) + values["self"] += self_size + values["inclusive"] += self_size + current["child_total"] + values["samples"] += 1 + node_count += 1 + if node_stack or sample is not None: + _malformed("V8 heap profile contains an incomplete object.") + _check_rows(request, len(frame_rows), len(aggregates)) + return V8ProfileResult( + profile_kind="heap", + node_count=node_count, + sample_count=sample_count, + total_sampled_bytes=total_sampled_bytes, + frames=tuple(frame_rows.values()), + frame_measurements=tuple( + { + "frame_id": frame_id, + "metric": "memory.self_size", + "self_value": values["self"], + "inclusive_value": values["inclusive"], + "unit": "bytes", + "sample_count": values["samples"], + } + for frame_id, values in sorted(aggregates.items()) + ), + limitations=( + "Sampled allocation bytes are an estimate from V8's sampling heap profiler, not " + "the exact retained heap or process RSS.", + "Only allocations sampled by V8 are reported; small or short-lived allocations " + "may be underrepresented.", + "Source-map resolution is not applied by this extractor.", + ), + ) + + +def _top_level_keys(path: Path) -> set[str]: + keys: set[str] = set() + with path.open("rb") as stream: + for prefix, event, value in ijson.parse(stream): + if prefix == "" and event == "map_key": + keys.add(str(value)) + return keys + + +def _call_frame(value: Any) -> dict[str, Any]: + if not isinstance(value, dict): + _malformed("V8 CPU profile callFrame must be an object.") + return cast(dict[str, Any], value) + + +def _strict_int(value: Any, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + _malformed(f"V8 profile {field} must be an integer.") + return value + + +def _frame_identity(call_frame: dict[str, Any], request: V8ProfileRequest) -> dict[str, Any]: + function_value = call_frame.get("functionName") + function = ( + function_value if isinstance(function_value, str) and function_value else "(anonymous)" + ) + url_value = call_frame.get("url") + url = url_value if isinstance(url_value, str) else "" + normalized = _normalize_url(url, Path(request.project_root)) + line = _strict_int(call_frame.get("lineNumber", 0), "line number") + column = _strict_int(call_frame.get("columnNumber", 0), "column number") + script_id = call_frame.get("scriptId") + disambiguator = ( + f"v8-script:{script_id}" + if not normalized and isinstance(script_id, (str, int)) and not isinstance(script_id, bool) + else None + ) + identity_payload: dict[str, object] = { + "language": "JavaScript", + "function": function, + "file": normalized, + "line": line, + "column": column, + } + if disambiguator is not None: + identity_payload["disambiguator"] = disambiguator + frame_id = digest_model(identity_payload) + return { + "frame_id": frame_id, + "language": "JavaScript", + "function": function, + "module": disambiguator, + "file": normalized, + "line": line, + "column": column, + "address": None, + "build_id": None, + "module_relative_address": None, + "inline_chain_id": None, + "source_state_id": None, + "artifact_id": request.artifact_id, + "inlined": False, + "symbolization": ( + "complete" + if function_value + and normalized + and normalized != "internal" + and not normalized.startswith("node:") + else "partial" + ), + } + + +def _normalize_url(value: str, project_root: Path) -> str: + if not value: + return "" + if value.startswith("node:"): + return value + parsed = urlparse(value) + path = Path(unquote(parsed.path)) if parsed.scheme == "file" else Path(value) + if not path.is_absolute(): + return value + try: + return path.resolve().relative_to(project_root.resolve()).as_posix() + except (OSError, ValueError): + return value + + +def _check_rows(request: V8ProfileRequest, frame_count: int, measurement_count: int) -> None: + if frame_count + measurement_count > request.max_rows: + _limit("V8 profile normalized row limit exceeded.") + + +def _malformed(message: str) -> NoReturn: + raise DomainError(ErrorCode.ARTIFACT_PARSE_FAILED, message) + + +def _limit(message: str) -> NoReturn: + raise DomainError(ErrorCode.QUERY_BUDGET_EXCEEDED, message) + + +def main() -> int: + return run_typed_worker( + WorkerApplication( + definition=V8_PROFILE_WORKER, + handler=lambda request, _context: ( + _parse_cpu(request) if request.profile_kind == "cpu" else _parse_heap(request) + ), + invalid_failure=WorkerFailureKind.INPUT_MALFORMED, + invalid_message="V8 profile input is malformed", + caught=(OSError, ValueError, TypeError, KeyError, ijson.common.JSONError), + ) + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/flameox/workers/v8_profiles_contract.py b/src/flameox/workers/v8_profiles_contract.py new file mode 100644 index 0000000..069d3d5 --- /dev/null +++ b/src/flameox/workers/v8_profiles_contract.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from typing import Annotated, Literal + +from pydantic import Field, JsonValue, TypeAdapter + +from flameox.models import ContractModel +from flameox.workers.protocol import WorkerDefinition, WorkerOperationId + + +class V8ProfileRequest(ContractModel): + schema_version: Literal[1] = 1 + profile_kind: Literal["cpu", "heap"] + artifact_path: str = Field(min_length=1, max_length=4_096) + artifact_id: str = Field(min_length=1, max_length=200) + project_root: str = Field(min_length=1, max_length=4_096) + max_nodes: Annotated[int, Field(gt=0, le=100_000)] + max_samples: Annotated[int, Field(gt=0, le=1_000_000)] + max_rows: Annotated[int, Field(gt=0, le=100_000)] + + +class V8ProfileResult(ContractModel): + schema_version: Literal[1] = 1 + profile_kind: Literal["cpu", "heap"] + node_count: Annotated[int, Field(ge=0)] + sample_count: Annotated[int, Field(ge=0)] + total_sampled_bytes: Annotated[int, Field(ge=0)] = 0 + frames: tuple[dict[str, JsonValue], ...] + frame_measurements: tuple[dict[str, JsonValue], ...] + limitations: tuple[str, ...] = () + + +V8_PROFILE_WORKER: WorkerDefinition[V8ProfileRequest, V8ProfileResult] = WorkerDefinition( + operation=WorkerOperationId.V8_PROFILE_PARSE, + module="flameox.workers.v8_profiles", + request=TypeAdapter(V8ProfileRequest), + response=TypeAdapter(V8ProfileResult), + name="V8 profile", + implementation="flameox.workers.v8_profiles/v1", +) diff --git a/tests/adapters/test_v8_capture_invocation.py b/tests/adapters/test_v8_capture_invocation.py new file mode 100644 index 0000000..27dc806 --- /dev/null +++ b/tests/adapters/test_v8_capture_invocation.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from flameox.adapters.builtins import ( + build_capture_invocation, + builtin_adapter, + node_version_is_supported, +) +from flameox.domain import ArtifactKind, DomainError, ErrorCode + +pytestmark = pytest.mark.unit + + +def test_node_cpu_prof_capture_injects_cpu_prof_flags(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-cpu-prof", + ("node", "script.js", "--arg"), + tmp_path, + executable="/usr/bin/node", + ) + + assert invocation.artifact_kinds == (ArtifactKind.SAMPLE_PROFILE,) + assert "--cpu-prof" in invocation.argv + dir_flag = next(a for a in invocation.argv if a.startswith("--cpu-prof-dir=")) + assert dir_flag.startswith("--cpu-prof-dir=") + assert invocation.argv[0] == "node" + assert invocation.argv[-2:] == ("script.js", "--arg") + + +def test_node_cpu_prof_capture_preserves_workload_arguments(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-cpu-prof", + ("node", "index.js", "--flag", "value"), + tmp_path, + executable="/usr/bin/node", + ) + + workload_part = invocation.argv[-3:] + assert workload_part == ("index.js", "--flag", "value") + + +def test_node_cpu_prof_capture_rejects_empty_workload(tmp_path: Path) -> None: + with pytest.raises(DomainError) as failure: + build_capture_invocation( + "node-cpu-prof", + (), + tmp_path, + executable="/usr/bin/node", + ) + + assert failure.value.code is ErrorCode.INVALID_CAPTURE_PLAN + assert "Node.js" in failure.value.message + + +def test_node_cpu_prof_capture_rejects_non_node_workload(tmp_path: Path) -> None: + with pytest.raises(DomainError) as failure: + build_capture_invocation( + "node-cpu-prof", + ("python", "script.py"), + tmp_path, + executable=None, + ) + + assert failure.value.code is ErrorCode.INVALID_CAPTURE_PLAN + + +def test_node_heap_prof_capture_injects_heap_prof_flags(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-heap-prof", + ("node", "script.js"), + tmp_path, + executable="/usr/bin/node", + ) + + assert invocation.artifact_kinds == (ArtifactKind.MEMORY_PROFILE,) + assert "--heap-prof" in invocation.argv + assert invocation.argv[0] == "node" + assert invocation.argv[-1] == "script.js" + + +def test_node_heap_prof_capture_preserves_workload_arguments(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-heap-prof", + ("node", "index.js", "--flag", "value"), + tmp_path, + executable="/usr/bin/node", + ) + + workload_part = invocation.argv[-3:] + assert workload_part == ("index.js", "--flag", "value") + + +def test_node_heap_prof_capture_rejects_empty_workload(tmp_path: Path) -> None: + with pytest.raises(DomainError) as failure: + build_capture_invocation( + "node-heap-prof", + (), + tmp_path, + executable="/usr/bin/node", + ) + + assert failure.value.code is ErrorCode.INVALID_CAPTURE_PLAN + assert "Node.js" in failure.value.message + + +def test_node_v8_capture_uses_output_filename(tmp_path: Path) -> None: + invocation_cpu = build_capture_invocation( + "node-cpu-prof", + ("node", "script.js"), + tmp_path, + executable="/usr/bin/node", + ) + invocation_heap = build_capture_invocation( + "node-heap-prof", + ("node", "script.js"), + tmp_path, + executable="/usr/bin/node", + ) + + cpu_name_flag = next(a for a in invocation_cpu.argv if a.startswith("--cpu-prof-name=")) + assert "cpu.cpuprofile" in cpu_name_flag + heap_name_flag = next(a for a in invocation_heap.argv if a.startswith("--heap-prof-name=")) + assert "heap.heapprofile" in heap_name_flag + + +def test_node_v8_adapters_preserve_failed_workload_profiles() -> None: + cpu = builtin_adapter("node-cpu-prof") + heap = builtin_adapter("node-heap-prof") + assert cpu is not None and cpu.preserve_artifact_on_nonzero is True + assert heap is not None and heap.preserve_artifact_on_nonzero is True + + +def test_node_v8_capture_uses_the_bound_workload_executable(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-cpu-prof", + ("node", "script.js"), + tmp_path, + executable="/usr/bin/node", + workload_executable="/opt/node/bin/node", + ) + + assert invocation.argv[0] == "/opt/node/bin/node" + + +@pytest.mark.parametrize( + ("version", "supported"), + [("v20.15.1", False), ("v20.16.0", True), ("v21.7.0", False), ("v22.4.0", True)], +) +def test_node_v8_version_floor(version: str, supported: bool) -> None: + assert node_version_is_supported(version) is supported diff --git a/tests/adapters/test_v8_profiles.py b/tests/adapters/test_v8_profiles.py new file mode 100644 index 0000000..f27a086 --- /dev/null +++ b/tests/adapters/test_v8_profiles.py @@ -0,0 +1,279 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from flameox.adapters import V8CpuProfExtractor, V8HeapProfExtractor +from flameox.application import ImportArtifactRequest, ImportService +from flameox.catalog import Catalog +from flameox.domain import ArtifactKind, DomainError, ErrorCode +from flameox.storage import Workspace + +pytestmark = [pytest.mark.integration] + + +def _write_cpu_profile(path: Path) -> None: + index_url = (path.parent / "index.js").as_uri() + utils_url = (path.parent / "utils.js").as_uri() + profile = { + "nodes": [ + { + "id": 0, + "callFrame": { + "functionName": "(root)", + "url": "internal", + "scriptId": "0", + "lineNumber": -1, + "columnNumber": -1, + }, + "hitCount": 0, + "children": [1, 2], + }, + { + "id": 1, + "callFrame": { + "functionName": "main", + "url": index_url, + "scriptId": "1", + "lineNumber": 10, + "columnNumber": 5, + }, + "hitCount": 100, + "children": [3], + }, + { + "id": 2, + "callFrame": { + "functionName": "idle", + "url": index_url, + "scriptId": "1", + "lineNumber": 20, + "columnNumber": 0, + }, + "hitCount": 50, + "children": [], + }, + { + "id": 3, + "callFrame": { + "functionName": "helper", + "url": utils_url, + "scriptId": "2", + "lineNumber": 5, + "columnNumber": 10, + }, + "hitCount": 30, + "children": [], + }, + ], + "samples": [1, 1, 2, 3, 1], + "startTime": 0, + "endTime": 100_000, + } + path.write_text(json.dumps(profile), encoding="utf-8") + + +def _write_heap_profile(path: Path) -> None: + alloc_url = (path.parent / "alloc.js").as_uri() + proxy_url = (path.parent / "proxy.js").as_uri() + profile = { + "head": { + "callFrame": { + "functionName": "(root)", + "url": "internal", + "scriptId": "0", + "lineNumber": -1, + "columnNumber": -1, + }, + "selfSize": 0, + "id": 0, + "children": [ + { + "callFrame": { + "functionName": "allocate", + "url": alloc_url, + "scriptId": "1", + "lineNumber": 1, + "columnNumber": 0, + }, + "selfSize": 4096, + "id": 1, + "children": [], + }, + { + "callFrame": { + "functionName": "proxy", + "url": proxy_url, + "scriptId": "2", + "lineNumber": 10, + "columnNumber": 5, + }, + "selfSize": 2048, + "id": 2, + "children": [ + { + "callFrame": { + "functionName": "nested", + "url": proxy_url, + "scriptId": "2", + "lineNumber": 11, + "columnNumber": 5, + }, + "selfSize": 1024, + "id": 3, + "children": [], + } + ], + }, + ], + }, + "samples": [ + {"size": 4096, "nodeId": 1, "ordinal": 0}, + {"size": 2048, "nodeId": 2, "ordinal": 1}, + ], + } + path.write_text(json.dumps(profile), encoding="utf-8") + + +def test_v8_cpu_prof_extractor_publishes_frame_measurements(tmp_path: Path) -> None: + capture = tmp_path / "cpu.cpuprofile" + _write_cpu_profile(capture) + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=capture, + kind=ArtifactKind.SAMPLE_PROFILE, + ) + ) + result = V8CpuProfExtractor(workspace).extract(imported.run.run_id) + + assert result.node_count == 4 + assert result.sample_count == 5 + assert result.frame_count >= 3 + assert len(result.limitations) >= 2 + with Catalog(workspace).open_snapshot() as snapshot: + frames = snapshot.execute( + "SELECT language, function, file FROM frames ORDER BY function" + ).fetchall() + frame_measurements = snapshot.execute( + "SELECT function, self_value, inclusive_value FROM frame_measurements " + "JOIN frames USING (frame_id) WHERE metric = 'cpu.hit_count' ORDER BY function" + ).fetchall() + frame_tuples = [(f[0], f[1], f[2]) for f in frames] + assert ("JavaScript", "helper", "utils.js") in frame_tuples + assert ("JavaScript", "main", "index.js") in frame_tuples + assert ("JavaScript", "idle", "index.js") in frame_tuples + assert ("helper", 30, 30) in frame_measurements + assert ("main", 100, 130) in frame_measurements + + +def test_v8_cpu_prof_extractor_rejects_non_cpu_profile(tmp_path: Path) -> None: + bad = tmp_path / "bad.cpuprofile" + bad.write_text('{"not_a_profile": true}', encoding="utf-8") + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=bad, + kind=ArtifactKind.SAMPLE_PROFILE, + ) + ) + with pytest.raises(DomainError) as failure: + V8CpuProfExtractor(workspace).extract(imported.run.run_id) + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED + + +def test_v8_cpu_prof_extractor_rejects_malformed_nodes_with_domain_error(tmp_path: Path) -> None: + capture = tmp_path / "bad.cpuprofile" + capture.write_text(json.dumps({"nodes": [{"callFrame": {}}], "samples": []}), encoding="utf-8") + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest(path=capture, kind=ArtifactKind.SAMPLE_PROFILE) + ) + with pytest.raises(DomainError) as failure: + V8CpuProfExtractor(workspace).extract(imported.run.run_id) + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED + + +def test_v8_heap_prof_extractor_publishes_sampled_bytes(tmp_path: Path) -> None: + capture = tmp_path / "heap.heapprofile" + _write_heap_profile(capture) + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=capture, + kind=ArtifactKind.MEMORY_PROFILE, + ) + ) + result = V8HeapProfExtractor(workspace).extract(imported.run.run_id) + + assert result.sample_count == 2 + assert result.total_sampled_bytes == 6144 + assert result.frame_count >= 3 + assert len(result.limitations) >= 3 + with Catalog(workspace).open_snapshot() as snapshot: + frames = snapshot.execute( + "SELECT language, function, file FROM frames ORDER BY function" + ).fetchall() + measurements = snapshot.execute( + "SELECT function, self_value, inclusive_value FROM frame_measurements " + "JOIN frames USING (frame_id) WHERE metric = 'memory.self_size'" + ).fetchall() + frame_tuples = [(f[0], f[1], f[2]) for f in frames] + assert ("JavaScript", "allocate", "alloc.js") in frame_tuples + assert ("JavaScript", "proxy", "proxy.js") in frame_tuples + assert ("proxy", 2048, 3072) in measurements + + +def test_v8_heap_prof_extractor_rejects_non_heap_profile(tmp_path: Path) -> None: + bad = tmp_path / "bad.heapprofile" + bad.write_text('{"no_head": true}', encoding="utf-8") + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=bad, + kind=ArtifactKind.MEMORY_PROFILE, + ) + ) + with pytest.raises(DomainError) as failure: + V8HeapProfExtractor(workspace).extract(imported.run.run_id) + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED + + +def test_v8_heap_prof_extractor_rejects_malformed_samples_with_domain_error( + tmp_path: Path, +) -> None: + capture = tmp_path / "bad.heapprofile" + capture.write_text( + json.dumps( + { + "head": { + "callFrame": {"functionName": "root"}, + "selfSize": 0, + "children": [], + }, + "samples": [None], + } + ), + encoding="utf-8", + ) + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest(path=capture, kind=ArtifactKind.MEMORY_PROFILE) + ) + with pytest.raises(DomainError) as failure: + V8HeapProfExtractor(workspace).extract(imported.run.run_id) + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED diff --git a/uv.lock b/uv.lock index 8de58f5..4b83dc5 100644 --- a/uv.lock +++ b/uv.lock @@ -3276,11 +3276,11 @@ wheels = [ [[package]] name = "pip" -version = "26.1.2" +version = "26.2.1" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/01/91/47e7d486260f618783899587af63ccf7980fb60245c3e63dd4571c6b57ad/pip-26.1.2.tar.gz", hash = "sha256:f49cd134c61cf2fd75e0ce2676db03e4054504a5a4986d00f8299ae632dc4605", size = 1840799, upload-time = "2026-05-31T17:33:58.56Z" } +sdist = { url = "https://files.pythonhosted.org/packages/ae/15/4500e320e6b101ec3b719ae85b697d9940b6cda672bc555bd6016fc60c6f/pip-26.2.1.tar.gz", hash = "sha256:f6ad667e89a1fe78046c8f13232b247200f5258d7828f3f7883d660878e0813f", size = 1848877, upload-time = "2026-08-04T22:51:14.148Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/5d/95/6b5cb3461ea5673ba0995989746db58eb18b91b54dbf331e72f569540946/pip-26.1.2-py3-none-any.whl", hash = "sha256:382ff9f685ee3bc25864f820aa50505825f10f5458ffff07e30a6d96e5715cab", size = 1813144, upload-time = "2026-05-31T17:33:56.772Z" }, + { url = "https://files.pythonhosted.org/packages/f3/6e/1736e5b4ae2b778ef2f81c47d797de9f891d4d8acb047a24ca37a60294dd/pip-26.2.1-py3-none-any.whl", hash = "sha256:71138adf1f4ca900cdb7d289c21b7494329f2332b6d85f0e1c42108c0384ed3e", size = 1816632, upload-time = "2026-08-04T22:51:12.472Z" }, ] [[package]]