diff --git a/project/TICKETS.md b/project/TICKETS.md index f3fc36d..d4ca1b0 100644 --- a/project/TICKETS.md +++ b/project/TICKETS.md @@ -15,4 +15,6 @@ analysis artifacts. | **ticket-007** | [`README.md`](./ticket-007/README.md) | [`preprompt.md`](./ticket-007/preprompt.md) | - | [`ai-codex.md`](./ticket-007/ai-codex.md) | [`ai-codex-logs.txt`](./ticket-007/ai-codex-logs.txt) | [`changelog.md`](./ticket-007/changelog.md) | | **ticket-008** | [`README.md`](./ticket-008/README.md) | [`preprompt.md`](./ticket-008/preprompt.md) | - | [`ai-codex.md`](./ticket-008/ai-codex.md) | [`ai-codex-logs.txt`](./ticket-008/ai-codex-logs.txt) | [`changelog.md`](./ticket-008/changelog.md) | | **ticket-009** | [`README.md`](./ticket-009/README.md) | [`preprompt.md`](./ticket-009/preprompt.md) | - | [`ai-codex.md`](./ticket-009/ai-codex.md) | [`ai-codex-logs.txt`](./ticket-009/ai-codex-logs.txt) | [`changelog.md`](./ticket-009/changelog.md) | +| **ticket-010** | [`README.md`](./ticket-010/README.md) | [`preprompt.md`](./ticket-010/preprompt.md) | - | [`ai-codex.md`](./ticket-010/ai-codex.md) | [`ai-codex-logs.txt`](./ticket-010/ai-codex-logs.txt) | [`changelog.md`](./ticket-010/changelog.md) | +| **ticket-011** | [`README.md`](./ticket-011/README.md) | [`preprompt.md`](./ticket-011/preprompt.md) | - | [`ai-codex.md`](./ticket-011/ai-codex.md) | [`ai-codex-logs.txt`](./ticket-011/ai-codex-logs.txt) | [`changelog.md`](./ticket-011/changelog.md) | diff --git a/project/ticket-011/README.md b/project/ticket-011/README.md new file mode 100644 index 0000000..c99f9c4 --- /dev/null +++ b/project/ticket-011/README.md @@ -0,0 +1,40 @@ +# Ticket 011: Implement energy telemetry and URI Process opportunity ranking + +- **ID**: ticket-011 +- **Owner**: unresolved:human +- **Status**: IN_PROGRESS +- **Workflow state**: VALIDATION +- **Created**: 2026-08-30 + +## Goal and scope + +Implement the v2 measurement contract declared by ticket-010. Observe optional +RAPL, cgroup v2 and PSI counters without controlling the process. Rank recurring +Process URIs independently by CPU, wall time, energy or I/O savings potential. +Accept an explicit process revision for version-over-version regression analysis. + +## Acceptance criteria + +- [x] AC-01: Scope is approved by the originating user request. +- [x] AC-02: Existing v1 samples remain readable and v2 telemetry discloses + method, attribution and confidence. +- [x] AC-03: Ranking requires twelve comparable samples by default and never + combines unlike resource units into an implicit score. +- [x] AC-04: CLI tests cover the opportunity query. +- [x] AC-05: Functional tests, a v2 schema instance check, a live + smoke test and governance pass. + +## Validation evidence + +- `/usr/bin/python3 -m pytest -q`: 14 passed. +- A generated v2 sample validates with Draft 2020-12 JSON Schema. +- A live `sleep` measurement and CPU opportunity query completed successfully. +- Runtime and CLI versions are regression-tested against package metadata. +- The live host exposed shared cgroup/PSI evidence but no readable RAPL domain; + energy was correctly marked unavailable. + +## Participants + +- Human participant: user:tom; authorization is recorded by the originating + request and no synthesized user file was created. +- Agent participant: [ai-codex.md](ai-codex.md) diff --git a/project/ticket-011/ai-codex-logs.txt b/project/ticket-011/ai-codex-logs.txt new file mode 100644 index 0000000..e69de29 diff --git a/project/ticket-011/ai-codex.md b/project/ticket-011/ai-codex.md new file mode 100644 index 0000000..866dacc --- /dev/null +++ b/project/ticket-011/ai-codex.md @@ -0,0 +1,45 @@ +--- +participant-id: agent:codex +participant: codex +role: agent +ticket: ticket-011 +--- +# Participant: codex (AI agent) + +## Understanding + +Implement the application half of ticket-010 without claiming exclusive energy +or cgroup attribution on a shared host. Keep measurement advisory and private. + +## Execution plan + +1. Add backward-compatible v2 sample objects. +2. Observe optional RAPL/cgroup/PSI counters without privilege. +3. Add objective-specific ranking and CLI exposure. +4. Verify behavior and governance. + +## Actual changes + +- Corrected the runtime and CLI version projection to match package version + `0.2.0`, with a regression test covering both interfaces. + +- Initialized the bounded ticket and recorded SESSION_EXECUTION_AUTHORIZATION + from the request to execute this work. +- Added backward-compatible sample v2 serialization with explicit unavailable + fallbacks for optional telemetry. +- Added read-only Linux RAPL, host CPU, cgroup v2 and PSI observation. Raw + cgroup paths are replaced by bounded hashes and shared attribution is named. +- Added CPU, wall, energy and I/O opportunity ranking with a twelve-sample + default gate, explicit units and an advisory-only result. +- Added CLI coverage and passed fourteen tests, schema validation, a live smoke + test and repository governance. +- Added a quiet measurement mode for runtimes that must preserve the wrapped + process output channel while appending resource evidence. + +## Blockers + +- None inside the recorded intent; proceed without a second confirmation. +- New authority remains required for destructive action, secret access, new + external coordination or material objective expansion. Protected delivery + may be invoked without another prompt when publication is in scope; its + exact-head trusted approval remains independent evidence. diff --git a/project/ticket-011/changelog.md b/project/ticket-011/changelog.md new file mode 100644 index 0000000..f9afbf2 --- /dev/null +++ b/project/ticket-011/changelog.md @@ -0,0 +1,6 @@ +# Ticket Changelog (ticket-011) + +## [0.1.0] - 2026-08-30 + +- Initial governance scaffold created. +- No human participant identity or content was generated. diff --git a/project/ticket-011/intent.json b/project/ticket-011/intent.json new file mode 100644 index 0000000..6765dba --- /dev/null +++ b/project/ticket-011/intent.json @@ -0,0 +1,62 @@ +{ + "schema": "new-project.intent/v3", + "ticket": "ticket-011", + "summary": "Implement energy telemetry and URI Process opportunity ranking", + "workstream": "application", + "classification": { + "kind": "FEATURE", + "priority": "P1", + "origin": "requested" + }, + "placement": { + "home": "semcod", + "runtimeOwner": "semcod", + "shape": "runtime_service", + "adopt": ["wellmanifest/new-project", "wellmanifest/logs", "wellmanifest/performance"] + }, + "allowedPaths": ["project/ticket-011/**", "TODO.md", "project/TICKETS.md", "src/**", "tests/**"], + "forbiddenPaths": ["project/ticket-*/user-*.md"], + "stacks": [], + "dependsOn": [], + "conflictsWith": [], + "integrationTicket": null, + "delivery": { + "acceptedBaseSha": "4fbc1477fee93d72f14b13164b5f85a308a359b2", + "targetBranch": "main", + "outcome": "Collect optional Linux energy and pressure evidence and rank recurring URI Process savings under an explicit objective.", + "nonGoals": [ + "Create or mutate cgroups, scheduling policy or process authorization.", + "Claim process-exclusive attribution for shared counters.", + "Persist arguments, output, environment values, raw cgroup paths or secrets.", + "Apply optimizations automatically or rank from fewer than twelve comparable samples by default." + ], + "complexity": "L", + "estimatedMinutes": 30, + "budgets": { + "maxImplementationFiles": 15, + "maxAffectedComponents": 5, + "maxPublicInterfaceChanges": 3, + "maxRuntimeDependencies": 0 + }, + "architecture": { + "status": "accepted", + "decision": "Observe existing counters without privilege, retain v1 reads, disclose attribution confidence and rank CPU, wall, energy or I/O independently.", + "components": [ + {"name": "sample-and-observer", "paths": ["src/estimation/model.py", "src/estimation/monitor.py"]}, + {"name": "analysis-interface", "paths": ["src/estimation/__init__.py", "src/estimation/stats.py", "src/estimation/cli.py"]}, + {"name": "verification", "paths": ["tests/**"]} + ], + "responsibilityChanges": false, + "interfaceChanges": ["Emit v2 samples with optional telemetry.", "Add the opportunities CLI query.", "Accept an explicit bounded process revision."], + "dataChanges": [], + "ui": {"impact": "none", "states": [], "evidence": []}, + "rollback": "Stop emitting v2 and remove the query; existing append-only evidence remains readable." + }, + "runtimeDependencies": [], + "validation": [ + {"criterion": "AC-02", "commands": ["PYTHONPATH=src python -m pytest -q"], "evidence": "bounded optional telemetry and v1 compatibility"}, + {"criterion": "AC-03", "commands": ["PYTHONPATH=src python -m pytest -q"], "evidence": "objective-specific rankings and minimum sample gate"}, + {"criterion": "AC-05", "commands": ["PYTHONPATH=src ./project/governance-check.sh"], "evidence": "repository governance passes"} + ] + } +} diff --git a/project/ticket-011/preprompt.md b/project/ticket-011/preprompt.md new file mode 100644 index 0000000..26d9e24 --- /dev/null +++ b/project/ticket-011/preprompt.md @@ -0,0 +1,15 @@ +# Ticket preprompt + +- **Task ID**: ticket-011 +- **Task title**: Implement energy telemetry and URI Process opportunity ranking +- **Created**: 2026-08-30T10:39:06Z + +Keep executable implementation outside this governance/evidence directory. +Read a human-owned user-*.md file only when one exists. +The request to execute this work creates SESSION_EXECUTION_AUTHORIZATION; +proceed within the recorded intent without a redundant confirmation prompt. +Require new authority for destructive action, secrets, external coordination, +or material objective expansion. When publication is in scope, session +authorization permits the declared protected delivery process and its merge +after exact-head trusted approval without another prompt. Session prose is +never approval evidence and the agent must not merge directly. diff --git a/src/estimation/__init__.py b/src/estimation/__init__.py index b092348..f1f970f 100644 --- a/src/estimation/__init__.py +++ b/src/estimation/__init__.py @@ -1,5 +1,7 @@ """Resource estimation for ticket-bound URI processes.""" +__version__ = "0.2.0" + from estimation.model import Sample, canonical_process_uri from estimation.monitor import measure_command, observe_pid from estimation.stats import aggregate_samples, estimate_workload @@ -15,5 +17,3 @@ "measure_command", "observe_pid", ] - -__version__ = "0.1.0" diff --git a/src/estimation/cli.py b/src/estimation/cli.py index 9851461..7554406 100644 --- a/src/estimation/cli.py +++ b/src/estimation/cli.py @@ -7,9 +7,10 @@ import sys from typing import Sequence +from estimation import __version__ from estimation.model import Sample from estimation.monitor import measure_command, observe_pid -from estimation.stats import aggregate_samples, estimate_workload +from estimation.stats import aggregate_samples, estimate_workload, rank_opportunities from estimation.store import append_sample, load_samples @@ -25,15 +26,17 @@ def _context(parser: argparse.ArgumentParser) -> None: parser.add_argument("--store", default=DEFAULT_STORE) parser.add_argument("--events", default=DEFAULT_EVENTS) parser.add_argument("--interval", type=float, default=1.0) + parser.add_argument("--process-revision", default=os.getenv("SUBACTOR_PROCESS_REVISION")) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="estimation") - parser.add_argument("--version", action="version", version="estimation 0.1.0") + parser.add_argument("--version", action="version", version=f"estimation {__version__}") commands = parser.add_subparsers(dest="command_name", required=True) run = commands.add_parser("run", help="measure a caller-authorized command") _context(run) + run.add_argument("--quiet", action="store_true", help="persist the sample without printing it") run.add_argument("command", nargs=argparse.REMAINDER) observe = commands.add_parser("observe", help="observe an existing PID tree") @@ -50,6 +53,12 @@ def _parser() -> argparse.ArgumentParser: estimate.add_argument("--quantity", type=int, default=1) estimate.add_argument("--parallelism", type=int, default=1) + opportunities = commands.add_parser("opportunities", help="rank recurring optimization opportunities") + opportunities.add_argument("--store", default=DEFAULT_STORE) + opportunities.add_argument("--objective", choices=("cpu", "wall", "energy", "io"), default="cpu") + opportunities.add_argument("--minimum-samples", type=int, default=12) + opportunities.add_argument("--reduction-fraction", type=float, default=0.30) + validate = commands.add_parser("validate", help="validate all stored samples") validate.add_argument("--store", default=DEFAULT_STORE) return parser @@ -63,7 +72,8 @@ def _require_process_uri(value: str | None) -> str: def _write_sample(sample: Sample, args: argparse.Namespace) -> None: append_sample(sample, Path(args.store), Path(args.events)) - print(json.dumps(sample.to_dict(), ensure_ascii=False, sort_keys=True)) + if not getattr(args, "quiet", False): + print(json.dumps(sample.to_dict(), ensure_ascii=False, sort_keys=True)) def main(argv: Sequence[str] | None = None) -> int: @@ -80,6 +90,7 @@ def main(argv: Sequence[str] | None = None) -> int: ticket_id=args.ticket, correlation_id=args.correlation_id, interval_seconds=args.interval, + process_revision=args.process_revision, ) _write_sample(sample, args) return int(sample.exit_code or 0) @@ -92,6 +103,7 @@ def main(argv: Sequence[str] | None = None) -> int: correlation_id=args.correlation_id, interval_seconds=args.interval, duration_seconds=args.duration, + process_revision=args.process_revision, ) _write_sample(sample, args) return 0 @@ -108,6 +120,15 @@ def main(argv: Sequence[str] | None = None) -> int: ) print(json.dumps(result, indent=2, sort_keys=True)) return 0 + if args.command_name == "opportunities": + result = rank_opportunities( + samples, + objective=args.objective, + minimum_samples=args.minimum_samples, + reduction_fraction=args.reduction_fraction, + ) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if args.command_name == "validate": print(json.dumps({"ok": True, "samples": len(samples), "store": str(args.store)})) return 0 diff --git a/src/estimation/model.py b/src/estimation/model.py index ec32dad..ac41d32 100644 --- a/src/estimation/model.py +++ b/src/estimation/model.py @@ -6,6 +6,7 @@ import json import os import platform +import re from typing import Any, Sequence from urllib.parse import urlsplit, urlunsplit import uuid @@ -13,7 +14,35 @@ import psutil -SAMPLE_SCHEMA = "semcod.estimation.sample/v1" +SAMPLE_SCHEMA = "semcod.estimation.sample/v2" +SUPPORTED_SAMPLE_SCHEMAS = {"semcod.estimation.sample/v1", SAMPLE_SCHEMA} +PROCESS_REVISION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/+\-]{0,127}$") + + +def unavailable_energy() -> dict[str, Any]: + return { + "joules": None, + "method": "unavailable", + "confidence": "none", + "domains": 0, + } + + +def unavailable_kernel_observation() -> dict[str, Any]: + return { + "cgroup_id": None, + "attribution": "unavailable", + "cpu_seconds": None, + "memory_peak_bytes": None, + "read_bytes": None, + "write_bytes": None, + "pids_peak": None, + "pressure": { + "cpu_some_seconds": None, + "io_some_seconds": None, + "memory_some_seconds": None, + }, + } def utc_now() -> str: @@ -31,6 +60,15 @@ def canonical_process_uri(value: str) -> str: return canonical[:320] +def canonical_process_revision(value: str | None) -> str | None: + revision = str(value or "").strip() + if not revision: + return None + if not PROCESS_REVISION.fullmatch(revision): + raise ValueError("process revision contains unsupported characters") + return revision + + def argv_sha256(argv: Sequence[str]) -> str: encoded = json.dumps(list(argv), ensure_ascii=False, separators=(",", ":")) return hashlib.sha256(encoded.encode("utf-8")).hexdigest() @@ -74,15 +112,21 @@ class Sample: host_profile: dict[str, Any] raw_output_included: bool = False secret_material_included: bool = False + energy: dict[str, Any] | None = None + kernel: dict[str, Any] | None = None + process_revision: str | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, payload: dict[str, Any]) -> "Sample": - if payload.get("schema") != SAMPLE_SCHEMA: + if payload.get("schema") not in SUPPORTED_SAMPLE_SCHEMAS: raise ValueError("unsupported estimation sample schema") - return cls(**payload) + normalized = dict(payload) + normalized.setdefault("energy", unavailable_energy()) + normalized.setdefault("kernel", unavailable_kernel_observation()) + return cls(**normalized) def build_sample( @@ -105,6 +149,9 @@ def build_sample( outcome: str, program: str, argv: Sequence[str], + energy: dict[str, Any] | None = None, + kernel: dict[str, Any] | None = None, + process_revision: str | None = None, ) -> Sample: canonical = canonical_process_uri(process_uri) duration = max(0.0, float(duration_seconds)) @@ -133,4 +180,7 @@ def build_sample( program=os.path.basename(program)[:128] or "unknown", argv_sha256=argv_sha256(argv), host_profile=host_profile(), + energy=energy or unavailable_energy(), + kernel=kernel or unavailable_kernel_observation(), + process_revision=canonical_process_revision(process_revision), ) diff --git a/src/estimation/monitor.py b/src/estimation/monitor.py index 6867f98..375fa42 100644 --- a/src/estimation/monitor.py +++ b/src/estimation/monitor.py @@ -1,7 +1,9 @@ from __future__ import annotations from dataclasses import dataclass, field +import hashlib import os +from pathlib import Path import signal import subprocess import time @@ -12,6 +14,169 @@ from estimation.model import Sample, build_sample, utc_now +POWER_CAP_ROOT = Path("/sys/class/powercap") +CGROUP_ROOT = Path("/sys/fs/cgroup") + + +def _read_number(path: Path) -> int | None: + try: + return int(path.read_text(encoding="utf-8").strip()) + except (OSError, ValueError): + return None + + +def _rapl_snapshot(root: Path = POWER_CAP_ROOT) -> dict[str, tuple[int, int | None]]: + result: dict[str, tuple[int, int | None]] = {} + try: + domains = [item for item in root.iterdir() if (item / "energy_uj").is_file()] + except OSError: + return result + for domain in domains: + energy = _read_number(domain / "energy_uj") + if energy is not None: + key = hashlib.sha256(domain.name.encode("utf-8")).hexdigest()[:16] + result[key] = (energy, _read_number(domain / "max_energy_range_uj")) + return result + + +def _rapl_delta(start: dict[str, tuple[int, int | None]], end: dict[str, tuple[int, int | None]]) -> tuple[float | None, int]: + delta = 0 + domains = 0 + for key, (before, maximum) in start.items(): + if key not in end: + continue + after = end[key][0] + change = after - before + if change < 0 and maximum: + change += maximum + if change >= 0: + delta += change + domains += 1 + return (delta / 1_000_000.0, domains) if domains else (None, 0) + + +def _host_cpu_seconds(path: Path = Path("/proc/stat")) -> float | None: + try: + fields = path.read_text(encoding="utf-8").splitlines()[0].split()[1:] + return sum(int(value) for value in fields) / float(os.sysconf("SC_CLK_TCK")) + except (OSError, ValueError, IndexError): + return None + + +def _cgroup_directory(pid: int, root: Path = CGROUP_ROOT) -> Path | None: + try: + rows = Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8").splitlines() + relative = next(row.split("::", 1)[1] for row in rows if "::" in row) + candidate = (root / relative.lstrip("/")).resolve() + candidate.relative_to(root.resolve()) + return candidate + except (OSError, StopIteration, ValueError): + return None + + +def _key_values(path: Path) -> dict[str, int]: + try: + return { + fields[0]: int(fields[1]) + for line in path.read_text(encoding="utf-8").splitlines() + if len(fields := line.split()) >= 2 and fields[1].isdigit() + } + except OSError: + return {} + + +def _io_values(path: Path) -> tuple[int, int] | None: + try: + reads = writes = 0 + for line in path.read_text(encoding="utf-8").splitlines(): + values = {key: int(value) for token in line.split()[1:] for key, value in [token.split("=", 1)]} + reads += values.get("rbytes", 0) + writes += values.get("wbytes", 0) + return reads, writes + except (OSError, ValueError): + return None + + +def _pressure_total(path: Path) -> int | None: + try: + line = next(row for row in path.read_text(encoding="utf-8").splitlines() if row.startswith("some ")) + return int(next(token.split("=", 1)[1] for token in line.split() if token.startswith("total="))) + except (OSError, StopIteration, ValueError): + return None + + +def _cgroup_snapshot(directory: Path | None) -> dict[str, int | str | None]: + if directory is None: + return {} + cpu = _key_values(directory / "cpu.stat") + io = _io_values(directory / "io.stat") + return { + "cgroup_id": "cgroup:" + hashlib.sha256(str(directory).encode("utf-8")).hexdigest()[:16], + "cpu_usec": cpu.get("usage_usec"), + "memory_peak": _read_number(directory / "memory.peak"), + "read_bytes": io[0] if io else None, + "write_bytes": io[1] if io else None, + "pids_peak": _read_number(directory / "pids.peak"), + "psi_cpu": _pressure_total(directory / "cpu.pressure"), + "psi_io": _pressure_total(directory / "io.pressure"), + "psi_memory": _pressure_total(directory / "memory.pressure"), + } + + +def _counter_delta(before: int | str | None, after: int | str | None, scale: float = 1.0) -> float | int | None: + if not isinstance(before, int) or not isinstance(after, int): + return None + return max(0, after - before) / scale + + +@dataclass +class LinuxObservation: + pid: int + rapl: dict[str, tuple[int, int | None]] = field(default_factory=_rapl_snapshot) + host_cpu_seconds: float | None = field(default_factory=_host_cpu_seconds) + cgroup_directory: Path | None = None + cgroup: dict[str, int | str | None] = field(default_factory=dict) + + @classmethod + def start(cls, pid: int) -> "LinuxObservation": + directory = _cgroup_directory(pid) + return cls(pid=pid, cgroup_directory=directory, cgroup=_cgroup_snapshot(directory)) + + def finish(self, process_cpu_seconds: float) -> tuple[dict[str, object], dict[str, object]]: + package_joules, domains = _rapl_delta(self.rapl, _rapl_snapshot()) + host_end = _host_cpu_seconds() + host_delta = None if self.host_cpu_seconds is None or host_end is None else max(0.0, host_end - self.host_cpu_seconds) + if package_joules is not None and host_delta and process_cpu_seconds >= 0: + energy = { + "joules": round(package_joules * min(1.0, process_cpu_seconds / host_delta), 6), + "method": "rapl_cpu_share", + "confidence": "low", + "domains": domains, + } + else: + energy = {"joules": None, "method": "unavailable", "confidence": "none", "domains": domains} + + end = _cgroup_snapshot(self.cgroup_directory) + same_cgroup = bool(self.cgroup) and self.cgroup.get("cgroup_id") == end.get("cgroup_id") + def delta(name: str, scale: float = 1.0): + return _counter_delta(self.cgroup.get(name), end.get(name), scale) if same_cgroup else None + kernel = { + "cgroup_id": end.get("cgroup_id") if same_cgroup else None, + "attribution": "shared" if same_cgroup else "unavailable", + "cpu_seconds": delta("cpu_usec", 1_000_000.0), + "memory_peak_bytes": end.get("memory_peak") if same_cgroup else None, + "read_bytes": delta("read_bytes"), + "write_bytes": delta("write_bytes"), + "pids_peak": end.get("pids_peak") if same_cgroup else None, + "pressure": { + "cpu_some_seconds": delta("psi_cpu", 1_000_000.0), + "io_some_seconds": delta("psi_io", 1_000_000.0), + "memory_some_seconds": delta("psi_memory", 1_000_000.0), + }, + } + return energy, kernel + + @dataclass class ResourceAccumulator: cpu_by_pid: dict[int, tuple[float, float]] = field(default_factory=dict) @@ -76,7 +241,11 @@ def _finish_sample( outcome: str, program: str, argv: Sequence[str], + observation: LinuxObservation, + process_revision: str | None, ) -> Sample: + process_cpu_seconds = accumulator.cpu_user_seconds + accumulator.cpu_system_seconds + energy, kernel = observation.finish(process_cpu_seconds) return build_sample( process_uri=process_uri, process_key=process_key, @@ -96,6 +265,9 @@ def _finish_sample( outcome=outcome, program=program, argv=argv, + energy=energy, + kernel=kernel, + process_revision=process_revision, ) @@ -107,6 +279,7 @@ def measure_command( ticket_id: str | None = None, correlation_id: str | None = None, interval_seconds: float = 1.0, + process_revision: str | None = None, ) -> Sample: argv = [str(item) for item in command] if not argv: @@ -117,6 +290,7 @@ def measure_command( child = subprocess.Popen(argv) root = psutil.Process(child.pid) accumulator = ResourceAccumulator() + observation = LinuxObservation.start(child.pid) interrupted = False try: @@ -147,6 +321,8 @@ def measure_command( outcome=outcome, program=os.path.basename(argv[0]), argv=argv, + observation=observation, + process_revision=process_revision, ) @@ -159,6 +335,7 @@ def observe_pid( correlation_id: str | None = None, interval_seconds: float = 1.0, duration_seconds: float = 10.0, + process_revision: str | None = None, ) -> Sample: root = psutil.Process(int(pid)) interval = max(0.01, float(interval_seconds)) @@ -166,6 +343,7 @@ def observe_pid( started_at = utc_now() started_monotonic = time.monotonic() accumulator = ResourceAccumulator() + observation = LinuxObservation.start(root.pid) while time.monotonic() - started_monotonic < duration_limit: try: @@ -192,4 +370,6 @@ def observe_pid( outcome="observed", program=program, argv=[program, f"pid:{int(pid)}"], + observation=observation, + process_revision=process_revision, ) diff --git a/src/estimation/stats.py b/src/estimation/stats.py index 4f1185b..f1ea645 100644 --- a/src/estimation/stats.py +++ b/src/estimation/stats.py @@ -19,6 +19,13 @@ "max_processes", ) +OPPORTUNITY_OBJECTIVES = { + "cpu": "CPU seconds", + "wall": "wall seconds", + "energy": "joules", + "io": "I/O bytes", +} + def _percentile(values: list[float], quantile: float) -> float: ordered = sorted(float(value) for value in values) @@ -68,6 +75,11 @@ def aggregate_samples(samples: Iterable[Sample]) -> dict[str, Any]: name: _metric_summary([float(getattr(item, name)) for item in successful]) for name in METRICS } if successful else {} + energy_values = [ + float(item.energy["joules"]) + for item in successful + if item.energy and isinstance(item.energy.get("joules"), (int, float)) + ] processes[key] = { "samples": len(group), "successful_samples": len(successful), @@ -75,14 +87,88 @@ def aggregate_samples(samples: Iterable[Sample]) -> dict[str, Any]: "confidence": _confidence(len(successful)), "hosts": len({item.host_profile["host_id"] for item in group}), "metrics": metrics, + "energy": { + "samples": len(energy_values), + "coverage": len(energy_values) / len(successful) if successful else 0.0, + "joules": _metric_summary(energy_values) if energy_values else {}, + }, } return { - "schema": "semcod.estimation.report/v1", + "schema": "semcod.estimation.report/v2", "processes": processes, "total_samples": sum(len(group) for group in groups.values()), } +def _objective_value(sample: Sample, objective: str) -> float | None: + if objective == "cpu": + return float(sample.cpu_user_seconds + sample.cpu_system_seconds) + if objective == "wall": + return float(sample.duration_seconds) + if objective == "io": + return float(sample.read_bytes + sample.write_bytes) + if objective == "energy": + value = sample.energy.get("joules") if sample.energy else None + return float(value) if isinstance(value, (int, float)) else None + raise ValueError(f"unsupported objective {objective!r}; choose from {', '.join(OPPORTUNITY_OBJECTIVES)}") + + +def rank_opportunities( + samples: Iterable[Sample], + *, + objective: str = "cpu", + minimum_samples: int = 12, + reduction_fraction: float = 0.30, +) -> dict[str, Any]: + if objective not in OPPORTUNITY_OBJECTIVES: + raise ValueError(f"unsupported objective {objective!r}; choose from {', '.join(OPPORTUNITY_OBJECTIVES)}") + if minimum_samples < 1: + raise ValueError("minimum_samples must be positive") + if not 0 < reduction_fraction <= 1: + raise ValueError("reduction_fraction must be greater than zero and at most one") + + groups: dict[str, list[Sample]] = defaultdict(list) + for sample in samples: + if sample.outcome in {"succeeded", "observed"}: + groups[sample.process_key].append(sample) + + ranked: list[dict[str, Any]] = [] + excluded: list[dict[str, Any]] = [] + for key, group in sorted(groups.items()): + values = [value for item in group if (value := _objective_value(item, objective)) is not None] + record = { + "process_key": key, + "successful_samples": len(group), + "objective_samples": len(values), + "coverage": len(values) / len(group), + "confidence": _confidence(len(values)), + } + if len(values) < minimum_samples: + excluded.append({**record, "reason": "insufficient_objective_samples"}) + continue + total = sum(values) + ranked.append({ + **record, + "observed_total": round(total, 6), + "p90_per_execution": round(_percentile(values, 0.90), 6), + "potential_saving": round(total * reduction_fraction, 6), + }) + + ranked.sort(key=lambda item: (-item["potential_saving"], item["process_key"])) + for position, item in enumerate(ranked, 1): + item["rank"] = position + return { + "schema": "semcod.estimation.opportunities/v1", + "objective": objective, + "unit": OPPORTUNITY_OBJECTIVES[objective], + "minimum_samples": minimum_samples, + "reduction_fraction": reduction_fraction, + "ranking": ranked, + "excluded": excluded, + "advisory_only": True, + } + + def estimate_workload( samples: Iterable[Sample], process_uri: str, diff --git a/tests/test_estimation.py b/tests/test_estimation.py index c03e604..13aa775 100644 --- a/tests/test_estimation.py +++ b/tests/test_estimation.py @@ -6,17 +6,19 @@ from pathlib import Path import sys +import pytest + from estimation.cli import main -from estimation.model import build_sample, canonical_process_uri, utc_now +from estimation.model import Sample, build_sample, canonical_process_revision, canonical_process_uri, utc_now from estimation.monitor import measure_command, observe_pid -from estimation.stats import aggregate_samples, estimate_workload +from estimation.stats import aggregate_samples, estimate_workload, rank_opportunities from estimation.store import append_sample, load_samples import estimation.store as store_module -def _sample(duration: float, memory: int, *, outcome: str = "succeeded"): +def _sample(duration: float, memory: int, *, outcome: str = "succeeded", process_uri: str = "diagit://fleet/worktrees", energy_joules: float | None = None): return build_sample( - process_uri="diagit://fleet/worktrees?root=/private/path", + process_uri=process_uri + "?root=/private/path", process_key=None, ticket_id="PLF-TEST", correlation_id=None, @@ -34,6 +36,12 @@ def _sample(duration: float, memory: int, *, outcome: str = "succeeded"): outcome=outcome, program="diagit", argv=["diagit", "--secret-value", "never-store-me"], + energy={ + "joules": energy_joules, + "method": "rapl_cpu_share" if energy_joules is not None else "unavailable", + "confidence": "low" if energy_joules is not None else "none", + "domains": 1 if energy_joules is not None else 0, + }, ) @@ -41,6 +49,13 @@ def test_canonical_uri_removes_query_and_fragment() -> None: assert canonical_process_uri("Diagit://fleet/worktrees?root=/tmp#x") == "diagit://fleet/worktrees" +def test_process_revision_is_explicit_and_bounded() -> None: + assert canonical_process_revision("git:abc123") == "git:abc123" + assert canonical_process_revision("") is None + with pytest.raises(ValueError, match="unsupported characters"): + canonical_process_revision("revision with spaces") + + def test_measure_command_and_store_do_not_persist_arguments(tmp_path: Path) -> None: sample = measure_command( [sys.executable, "-c", "sum(i*i for i in range(10000))", "secret-marker"], @@ -59,6 +74,21 @@ def test_measure_command_and_store_do_not_persist_arguments(tmp_path: Path) -> N assert "token=hidden" not in sample.process_key assert "token=hidden" not in store.read_text(encoding="utf-8") assert load_samples(store)[0].argv_sha256 == sample.argv_sha256 + assert sample.schema == "semcod.estimation.sample/v2" + assert sample.energy is not None + assert sample.kernel is not None + assert sample.kernel["attribution"] in {"shared", "unavailable"} + + +def test_v1_sample_remains_readable() -> None: + payload = _sample(1.0, 100).to_dict() + payload["schema"] = "semcod.estimation.sample/v1" + payload.pop("energy") + payload.pop("kernel") + restored = Sample.from_dict(payload) + assert restored.schema == "semcod.estimation.sample/v1" + assert restored.energy["method"] == "unavailable" + assert restored.kernel["attribution"] == "unavailable" def test_event_store_builds_a_hash_chain(tmp_path: Path) -> None: @@ -113,6 +143,53 @@ def test_report_and_workload_estimate_use_successful_p90() -> None: assert estimate["low_confidence"] is False +def test_opportunities_rank_absolute_savings_for_explicit_objective() -> None: + samples = [ + *[_sample(10.0, 100, process_uri="task://slow/work") for _ in range(12)], + *[_sample(2.0, 100, process_uri="task://fast/work") for _ in range(12)], + *[_sample(100.0, 100, process_uri="task://one-off/work") for _ in range(2)], + ] + result = rank_opportunities(samples, objective="cpu", reduction_fraction=0.25) + assert [item["process_key"] for item in result["ranking"]] == ["task://slow/work", "task://fast/work"] + assert result["ranking"][0]["potential_saving"] == 18.0 + assert result["excluded"][0]["process_key"] == "task://one-off/work" + assert result["advisory_only"] is True + + +def test_energy_opportunities_disclose_coverage_and_ignore_missing_values() -> None: + samples = [ + *[_sample(1.0, 100, process_uri="task://energy/a", energy_joules=5.0) for _ in range(12)], + *[_sample(1.0, 100, process_uri="task://energy/b") for _ in range(12)], + ] + result = rank_opportunities(samples, objective="energy") + assert result["ranking"][0]["potential_saving"] == 18.0 + assert result["ranking"][0]["coverage"] == 1.0 + assert result["excluded"][0]["objective_samples"] == 0 + + +def test_cli_prints_opportunity_ranking(tmp_path: Path, capsys) -> None: + store = tmp_path / "samples.jsonl" + events = tmp_path / "events.jsonl" + for _ in range(12): + append_sample(_sample(1.0, 100), store, events) + assert main(["opportunities", "--store", str(store), "--objective", "wall"]) == 0 + result = json.loads(capsys.readouterr().out) + assert result["ranking"][0]["process_key"] == "diagit://fleet/worktrees" + + +def test_cli_quiet_run_persists_without_mixing_sample_into_stdout(tmp_path: Path, capfd) -> None: + store = tmp_path / "samples.jsonl" + events = tmp_path / "events.jsonl" + assert main([ + "run", "--process-uri", "test://quiet/run", "--store", str(store), + "--events", str(events), "--interval", "0.01", "--process-revision", "git:abc123", "--quiet", "--", + sys.executable, "-c", "print('handler-output')", + ]) == 0 + assert capfd.readouterr().out.strip() == "handler-output" + assert load_samples(store)[0].process_key == "test://quiet/run" + assert load_samples(store)[0].process_revision == "git:abc123" + + def test_observe_current_process_is_bounded() -> None: sample = observe_pid( os.getpid(), diff --git a/tests/test_version.py b/tests/test_version.py new file mode 100644 index 0000000..d0eb9ab --- /dev/null +++ b/tests/test_version.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from pathlib import Path +import subprocess +import sys +import tomllib + +import estimation + + +def test_runtime_and_cli_versions_match_package_metadata() -> None: + metadata = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8")) + expected = metadata["project"]["version"] + + assert estimation.__version__ == expected + result = subprocess.run( + [ + sys.executable, + "-c", + "from estimation.cli import main; main(['--version'])", + ], + check=True, + capture_output=True, + text=True, + ) + assert result.stdout.strip() == f"estimation {expected}"