Harden loop self-observability

src/hyrule_engineering_loop/backend.py

@@ -580,7 +580,96 @@ def _parse_harness_output(self, stdout: str) -> dict[str, Any]:
             decoded = json.loads(stdout)
         except (json.JSONDecodeError, ValueError):
             return {}
-        return decoded if isinstance(decoded, dict) else {}
+        if not isinstance(decoded, dict):
+            return {}
+
+        parsed = dict(decoded)
+
+        def number_from(mapping: Mapping[str, Any], *keys: str) -> float | None:
+            for key in keys:
+                value = mapping.get(key)
+                if isinstance(value, (int, float)) and not isinstance(value, bool):
+                    return float(value)
+                if isinstance(value, str):
+                    try:
+                        return float(value)
+                    except ValueError:
+                        continue
+            return None
+
+        raw_usage = parsed.get("usage")
+        normalized_usage = dict(raw_usage) if isinstance(raw_usage, dict) else {}
+        input_tokens = number_from(
+            normalized_usage,
+            "input_tokens",
+            "inputTokens",
+            "prompt_tokens",
+            "promptTokens",
+        )
+        if input_tokens is None:
+            input_tokens = number_from(
+                parsed, "input_tokens", "inputTokens", "prompt_tokens", "promptTokens"
+            )
+        output_tokens = number_from(
+            normalized_usage,
+            "output_tokens",
+            "outputTokens",
+            "completion_tokens",
+            "completionTokens",
+        )
+        if output_tokens is None:
+            output_tokens = number_from(
+                parsed,
+                "output_tokens",
+                "outputTokens",
+                "completion_tokens",
+                "completionTokens",
+            )
+        if input_tokens is not None:
+            normalized_usage["input_tokens"] = int(input_tokens)
+        if output_tokens is not None:
+            normalized_usage["output_tokens"] = int(output_tokens)
+        if normalized_usage:
+            parsed["usage"] = normalized_usage
+
+        cost = number_from(parsed, "total_cost_usd", "totalCostUsd", "cost_usd", "costUsd")
+        raw_cost = parsed.get("cost")
+        if cost is None and isinstance(raw_cost, dict):
+            cost = number_from(raw_cost, "usd", "total_usd", "totalCostUsd", "cost_usd")
+        if cost is not None:
+            parsed["total_cost_usd"] = float(cost)
+
+        turns = number_from(parsed, "num_turns", "numTurns", "turns")
+        if turns is not None:
+            parsed["num_turns"] = int(turns)
+
+        if "is_error" not in parsed:
+            parsed["is_error"] = parsed.get("type") == "error" or bool(parsed.get("error"))
+
+        result = parsed.get("result")
+        if isinstance(result, dict):
+            text = next(
+                (
+                    str(result[key])
+                    for key in ("text", "content", "message", "summary")
+                    if isinstance(result.get(key), str)
+                ),
+                "",
+            )
+            parsed["result"] = text
+        elif not isinstance(result, str):
+            text = next(
+                (
+                    str(parsed[key])
+                    for key in ("output", "text", "content", "message")
+                    if isinstance(parsed.get(key), str)
+                ),
+                "",
+            )
+            if text:
+                parsed["result"] = text
+
+        return parsed
 
     def execute(
         self,
@@ -722,7 +811,7 @@ class PiBackend(SubprocessBackend):
     """
 
     name = "pi"
-    default_command = ("pi", "--print", "{prompt}")
+    default_command = ("pi", "--print", "--mode", "json", "{prompt}")
     extra_env_names = PI_PROVIDER_ENV_NAMES
 
 

src/hyrule_engineering_loop/daemon.py

@@ -18,6 +18,7 @@
 
 import json
 import os
+import re
 import ssl
 import time
 import urllib.request
@@ -93,6 +94,31 @@
 }
 HIGH_RISK_LABELS = frozenset({"critical", "security"})
 
+BUDGET_LABEL_TIERS: dict[str, str] = {
+    "loop:budget-s": "s",
+    "loop:budget-small": "s",
+    "loop:budget-m": "m",
+    "loop:budget-medium": "m",
+    "loop:budget-l": "l",
+    "loop:budget-large": "l",
+    "loop:budget-xl": "xl",
+    "loop:budget-extra-large": "xl",
+}
+BUDGET_BODY_RE = re.compile(
+    r"(?im)^\s*(?:loop[-_ ]?budget|budget)\s*:\s*"
+    r"(?P<tier>s|small|m|medium|l|large|xl|extra[-_ ]?large)\s*$"
+)
+BUDGET_TIER_MULTIPLIERS: dict[str, float] = {
+    "s": 1.0,
+    "m": 1.0,
+    "l": 2.0,
+    "xl": 4.0,
+}
+BUDGET_TIER_FLOORS: dict[str, tuple[int, int, float]] = {
+    "l": (40, 90, 10.0),
+    "xl": (80, 180, 20.0),
+}
+
 
 class DaemonError(RuntimeError):
     """Raised when a daemon cycle cannot run at all."""
@@ -599,6 +625,68 @@ def _run_cost(final_state: dict[str, Any]) -> float:
     )
 
 
+def _normalize_budget_tier(value: str) -> str | None:
+    normalized = value.strip().lower().replace("_", "-").replace(" ", "-")
+    if normalized in {"s", "small"}:
+        return "s"
+    if normalized in {"m", "medium"}:
+        return "m"
+    if normalized in {"l", "large"}:
+        return "l"
+    if normalized in {"xl", "extra-large"}:
+        return "xl"
+    return None
+
+
+def issue_budget_tier(item: IntakeItem, *, body: str) -> str | None:
+    """Resolve the requested per-issue budget tier from labels or body fields."""
+    for label in item.labels:
+        tier = BUDGET_LABEL_TIERS.get(label.strip().lower())
+        if tier is not None:
+            return tier
+    match = BUDGET_BODY_RE.search(body)
+    if match is None:
+        return None
+    return _normalize_budget_tier(match.group("tier"))
+
+
+def backend_budget_for_issue(
+    config: DaemonConfig,
+    item: IntakeItem,
+    *,
+    body: str,
+) -> dict[str, Any]:
+    """Build the backend budget for one issue, including explicit sizing signals."""
+    budget: dict[str, Any] = {
+        "max_iterations": config.max_iterations_per_run,
+        "max_wall_clock_minutes": config.max_wall_clock_minutes_per_run,
+        "max_cost_usd": config.max_cost_usd_per_run,
+        "tier": "default",
+    }
+    tier = issue_budget_tier(item, body=body)
+    if tier is None:
+        return budget
+
+    budget["tier"] = tier
+    multiplier = BUDGET_TIER_MULTIPLIERS[tier]
+    budget["max_iterations"] = max(1, int(config.max_iterations_per_run * multiplier))
+    budget["max_wall_clock_minutes"] = max(
+        1,
+        int(config.max_wall_clock_minutes_per_run * multiplier),
+    )
+    budget["max_cost_usd"] = round(config.max_cost_usd_per_run * multiplier, 4)
+
+    floor = BUDGET_TIER_FLOORS.get(tier)
+    if floor is not None:
+        min_iterations, min_minutes, min_cost = floor
+        budget["max_iterations"] = max(budget["max_iterations"], min_iterations)
+        budget["max_wall_clock_minutes"] = max(
+            budget["max_wall_clock_minutes"], min_minutes
+        )
+        budget["max_cost_usd"] = max(float(budget["max_cost_usd"]), min_cost)
+    return budget
+
+
 # --- the cycle --------------------------------------------------------------
 
 
@@ -767,11 +855,7 @@ def daemon_once(
             allowed_paths=effective_allowed_paths,
             source_files=["README.md"],
             memory_dir=config.memory_dir,
-            backend_budget={
-                "max_iterations": config.max_iterations_per_run,
-                "max_wall_clock_minutes": config.max_wall_clock_minutes_per_run,
-                "max_cost_usd": config.max_cost_usd_per_run,
-            },
+            backend_budget=backend_budget_for_issue(config, item, body=body),
             knowledge_context=config.knowledge_context,
             knowledge_learning_dir=config.knowledge_learning_dir,
         )

src/hyrule_engineering_loop/gate_runner.py

@@ -5,6 +5,7 @@
 import json
 import subprocess
 import sys
+import tomllib
 from pathlib import Path
 from typing import Any, Iterable, Sequence
 
@@ -45,6 +46,8 @@ def run_gate_commands(
         if not argv:
             raise ValueError("gate command cannot be empty")
 
+        startup_error = False
+        result: dict[str, Any]
         try:
             completed = subprocess.run(
                 argv,
@@ -67,14 +70,28 @@ def run_gate_commands(
                 "stdout": _clip(_as_text(exc.stdout)),
                 "stderr": _clip(_as_text(exc.stderr) or f"timed out after {timeout_seconds}s"),
             }
+        except OSError as exc:
+            startup_error = True
+            result = {
+                "command": argv,
+                "returncode": 127,
+                "stdout": "",
+                "stderr": _clip(f"could not start command: {exc}"),
+            }
+
+        result["status"] = "passed" if result["returncode"] == 0 else "failed"
 
         results.append(result)
         if result["returncode"] != 0:
             errors.append(
                 {
                     "node": "gate_execution",
                     "domain": "ci",
-                    "message": f"command failed: {' '.join(argv)}",
+                    "message": (
+                        f"command could not start: {' '.join(argv)}"
+                        if startup_error
+                        else f"command failed: {' '.join(argv)}"
+                    ),
                     "returncode": result["returncode"],
                     "stderr": result["stderr"],
                 }
@@ -83,12 +100,43 @@ def run_gate_commands(
     return results, errors
 
 
-def select_gate_commands_for_mutations(paths: Iterable[str]) -> list[list[str]]:
+def _python_project_gate_prefix(cwd: Path | str | None) -> list[str] | None:
+    if cwd is None:
+        return None
+    pyproject = Path(cwd) / "pyproject.toml"
+    if not pyproject.is_file():
+        return None
+    try:
+        data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
+    except tomllib.TOMLDecodeError:
+        return None
+    project = data.get("project")
+    optional = project.get("optional-dependencies") if isinstance(project, dict) else None
+    if isinstance(optional, dict) and "dev" in optional:
+        return ["uv", "run", "--extra", "dev"]
+    groups = data.get("dependency-groups")
+    if isinstance(groups, dict) and "dev" in groups:
+        return ["uv", "run", "--group", "dev"]
+    return None
+
+
+def select_gate_commands_for_mutations(
+    paths: Iterable[str],
+    *,
+    cwd: Path | str | None = None,
+) -> list[list[str]]:
     """Select local, workspace-safe gates from proposed mutation paths."""
     normalized = [path.split(":", 1)[1] if ":" in path else path for path in paths]
     if not normalized:
         return []
     if any(path.endswith(".py") for path in normalized):
+        prefix = _python_project_gate_prefix(cwd)
+        if prefix is not None:
+            return [
+                [*prefix, "ruff", "check", "."],
+                [*prefix, "mypy", "."],
+                [*prefix, "python", "-m", "pytest", "-q"],
+            ]
         return [[sys.executable, "-m", "compileall", "-q", "."]]
     if all(path.startswith("docs/") or path.endswith((".md", ".txt", ".rst")) for path in normalized):
         paths_literal = repr(json.dumps(normalized))