diff --git a/src/hyperloom/agents/robustness/role/envelope.py b/src/hyperloom/agents/robustness/role/envelope.py index 4df8301ae3..6a885dafa4 100644 --- a/src/hyperloom/agents/robustness/role/envelope.py +++ b/src/hyperloom/agents/robustness/role/envelope.py @@ -97,6 +97,7 @@ class IntentType(str, Enum): { "current_best", "stop_reason", + "stop_ts", "last_tick_exception", "cumulative_gain", "cumulative_gain_validated", @@ -111,6 +112,7 @@ class IntentType(str, Enum): "model_name", "model_class", "start_ts", + "resumed_ts", "max_minutes", "optimization_stack", "gain_per_stack_entry", diff --git a/src/hyperloom/common/coerce.py b/src/hyperloom/common/coerce.py index 3c1e69f7d2..7784ab3f62 100644 --- a/src/hyperloom/common/coerce.py +++ b/src/hyperloom/common/coerce.py @@ -20,7 +20,7 @@ from __future__ import annotations -from datetime import datetime +from datetime import datetime, timezone from typing import Any, TypeVar _T = TypeVar("_T") @@ -128,6 +128,11 @@ def to_unix(value: Any, default: _T | None = None) -> float | _T | None: tolerated). String parsing is ISO-first: an ISO-8601 timestamp is parsed to its epoch, falling back to a bare ``float`` cast for numeric strings. + A timestamp with no offset is read as UTC, matching + :func:`hyperloom.common.timeutil.iso_z`: every producer here writes UTC, + and letting the host's ``TZ`` decide would place the same string at + different instants for the two readers of it. + Args: value: The raw timestamp value. default: Returned when *value* cannot be interpreted as a timestamp @@ -143,12 +148,15 @@ def to_unix(value: Any, default: _T | None = None) -> float | _T | None: if isinstance(value, str): text = value.strip() try: - return datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp() + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) except ValueError: try: return float(text) except ValueError: return default + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.timestamp() return default diff --git a/src/hyperloom/inference_optimizer/breakdown/SKILL.md b/src/hyperloom/inference_optimizer/breakdown/SKILL.md index 184314808a..55673398db 100644 --- a/src/hyperloom/inference_optimizer/breakdown/SKILL.md +++ b/src/hyperloom/inference_optimizer/breakdown/SKILL.md @@ -130,7 +130,7 @@ this reference is partial — `breakdown/exporter.py` is authoritative. | Section | Reads from | |----------------------|----------------------------------------------------------------------------------------------------------------------| -| `session` | `manifest.json` + `state.{session_id, stop_reason, max_minutes, tick, start_ts}` | +| `session` | `manifest.json` + `state.{session_id, stop_reason, stop_ts, max_minutes, tick, start_ts, resumed_ts}` | | `workload` | `manifest.{framework, model_*, gpu_type, tp, workload, objective}` + `state.{model_class, framework, gpu_type}` | | `baseline` | `state.{baseline_tput, baseline_accuracy, last_baseline.workspace, baseline_attempts}` + `/benchmark_*/benchmark_report.json` | | `final` | `state.{current_best, cumulative_gain, cumulative_gain_validated_*, optimization_stack}` | diff --git a/src/hyperloom/inference_optimizer/breakdown/collectors/__init__.py b/src/hyperloom/inference_optimizer/breakdown/collectors/__init__.py index 6bc063f9b6..6b789f4d96 100644 --- a/src/hyperloom/inference_optimizer/breakdown/collectors/__init__.py +++ b/src/hyperloom/inference_optimizer/breakdown/collectors/__init__.py @@ -58,6 +58,7 @@ _collect_recovery as _collect_recovery, collect_session as collect_session, collect_session_meta as collect_session_meta, + session_elapsed_minutes as session_elapsed_minutes, collect_workload as collect_workload, collect_model_info as collect_model_info, collect_baseline as collect_baseline, @@ -211,4 +212,5 @@ "collect_model_info", "collect_optimizations", "collect_v4_optimizations", + "session_elapsed_minutes", ] diff --git a/src/hyperloom/inference_optimizer/breakdown/collectors/sessions.py b/src/hyperloom/inference_optimizer/breakdown/collectors/sessions.py index 876c443331..2f7fc74818 100644 --- a/src/hyperloom/inference_optimizer/breakdown/collectors/sessions.py +++ b/src/hyperloom/inference_optimizer/breakdown/collectors/sessions.py @@ -19,6 +19,7 @@ from pathlib import Path from typing import Any +from hyperloom.common.coerce import to_unix from hyperloom.common.timeutil import iso_z, now_iso from ._common import ( @@ -495,19 +496,60 @@ def _detect_image_for_session(manifest: dict[str, Any]) -> str | None: return None -def _close_phase_stop_reason(state: dict[str, Any]) -> tuple[str, str]: - """Recover terminal reason/time from the CLOSE phase transition (next-best when ``state.stop_reason`` wasn't mirrored). +def _leg_start_ts(state: dict[str, Any], start_ts: str) -> str: + """When the session's current run leg began. + + ``start_ts`` alone does not answer this. A resume re-anchors it only after + a crash or a stop with a reason; a resume after a clean stop deliberately + keeps it, so that ``--max-hours`` still counts from the original start. + ``state.resumed_ts`` is stamped by every resume, so the later of the two is + the boundary on both paths. + + Args: + state (dict[str, Any]): Parsed ``state.json``. + start_ts (str): The session's resolved start (see + :func:`collect_session`). + + Returns: + str: The later of the two timestamps, or whichever one is parseable. + """ + resumed_ts = str(state.get("resumed_ts") or "") + dated = [(to_unix(ts), ts) for ts in (start_ts, resumed_ts)] + parseable = [(at, ts) for at, ts in dated if at is not None] + if not parseable: + return start_ts + return max(parseable)[1] + + +def _close_phase_stop_reason(state: dict[str, Any], *, leg_start_ts: str) -> tuple[str, str]: + """Recover terminal reason/time from the current leg's CLOSE transition (next-best when ``state.stop_reason`` wasn't mirrored). + + A resume clears ``state.stop_reason`` and ``stop_ts`` but cannot clear the + previous leg's CLOSE row, and that row is not evidence about the leg + running now: honouring it reports a live session as having stopped, for + the reason it stopped last time. A row from before the leg boundary is + skipped whole -- reason and timestamp -- because the timestamp is stamped + as the session's end even when the reason itself is not adopted, and the + scan carries on so a history written out of order can still be answered + from a row that does belong to this leg. + + A row is only disqualified on comparable evidence. When either timestamp + is missing or unparseable the row stands, since the whole point of the + fallback is a session whose reason never reached the state file. Args: state (dict[str, Any]): Parsed ``state.json``. + leg_start_ts (str): Start of the current leg (see + :func:`_leg_start_ts`); ``""`` when the session recorded none. Returns: tuple[str, str]: ``(reason, ts)`` from the most recent CLOSE - transition, or ``("", "")`` when no such transition exists. + transition of the current leg, or ``("", "")`` when there is none. """ history = state.get("phase_history") or [] if not isinstance(history, list): return "", "" + leg_start = to_unix(leg_start_ts) for row in reversed(history): if not isinstance(row, dict): continue @@ -515,10 +557,97 @@ def _close_phase_stop_reason(state: dict[str, Any]) -> tuple[str, str]: continue reason = str(row.get("reason") or row.get("stop_reason") or row.get("exit_reason") or "").strip() ts = str(row.get("ts") or row.get("entered_ts") or "").strip() + closed_at = to_unix(ts) + if leg_start is not None and closed_at is not None and closed_at < leg_start: + continue return reason, ts return "", "" +def _first_recorded_end(*candidates: Any) -> str: + """The first candidate that reads as a timestamp, canonicalised to ``...Z``. + + A value that does not parse is no more an end time than a missing one: + passed through it lands in ``ended_at_utc`` verbatim and collapses the + measured duration to zero, where the next candidate (or the export clock) + still answers. + + Args: + *candidates (Any): Recorded end timestamps, best evidence first. + + Returns: + str: The first parseable candidate, or ``""`` when none is. + """ + for value in candidates: + if to_unix(value) is not None: + return iso_z(value) + return "" + + +def _session_has_ended(stop_reason: Any) -> bool: + """Whether a stop reason marks the session as no longer running. + + Args: + stop_reason (Any): Raw ``stop_reason`` from a state or session section. + + Returns: + bool: ``True`` once a non-blank stop reason has been recorded. + """ + return bool(str(stop_reason or "").strip()) + + +def _measured_duration_seconds(start_ts: Any, ended_at_utc: Any, stop_reason: Any) -> int | None: + """Seconds the session ran, or ``None`` when no window can be established. + + A finished session is measured to its recorded end; only one still running + may be measured up to now, since extrapolating a finished session grows its + duration on every re-export and reads as a plausible number rather than as + missing evidence. + + Args: + start_ts (Any): Start of the window (see :func:`collect_session` for + which start that is across a resume). + ended_at_utc (Any): Recorded end of the window, if any. + stop_reason (Any): Terminal reason; a non-blank one means the session + is no longer running. + + Returns: + int | None: Whole seconds between start and end, or ``None``. + """ + start = to_unix(start_ts) + if start is None: + return None + end = to_unix(ended_at_utc) + if end is None and not _session_has_ended(stop_reason): + end = datetime.now(timezone.utc).timestamp() + if end is None or end <= start: + return None + return int(round(end - start)) + + +def session_elapsed_minutes(session_section: dict[str, Any]) -> float: + """Wall-clock minutes of the leg described by a resolved ``session`` section. + + Derived from the section's own timestamps rather than stored, so a section + assembled from the live recorder's snapshot reports the same elapsed time + as one built by :func:`collect_session`. ``session_meta`` measures the same + window from the same fields; the two agree because both producers of the + section carry those timestamps, not because either reads the other. + + Args: + session_section (dict[str, Any]): A ``session`` section. + + Returns: + float: Minutes elapsed, or ``0.0`` when no window can be established. + """ + duration_s = _measured_duration_seconds( + session_section.get("start_ts") or session_section.get("created_at_utc"), + session_section.get("ended_at_utc"), + session_section.get("stop_reason"), + ) + return round(duration_s / 60.0, 2) if duration_s is not None else 0.0 + + def _should_use_close_stop_reason(stop_reason: str, close_stop_reason: str) -> bool: """Decide whether the CLOSE-phase stop reason should override the session's. @@ -613,10 +742,21 @@ def collect_session( """Collect the session-identification + lifecycle section. Merges identifiers and timing from ``state`` and ``manifest`` (state - taking precedence on overlapping fields), computes ``elapsed_minutes`` - from the start timestamp, resolves the container image, and stamps - ``ended_at_utc`` only once a ``stop_reason`` is present. When no image can - be detected a warning is appended. + taking precedence on overlapping fields), resolves the container image, + and stamps ``ended_at_utc`` from the recorded stop timestamp only once a + ``stop_reason`` is present -- one the state file carries, or one recovered + from a CLOSE transition belonging to the current leg (see + :func:`_close_phase_stop_reason`). When no image can be detected a warning + is appended. + + ``elapsed_minutes`` runs from ``state.start_ts``, the same anchor + ``--max-hours`` is counted against, to the recorded end (or to now while + the run is still going), so the two stay comparable. A resume re-anchors + ``start_ts`` only when the previous leg crashed or stopped for a recorded + reason; after a clean stop it keeps the original start, and the elapsed + time then spans the gap between the legs -- as the budget does. The + manifest's ``created_at_utc`` names the first launch either way, and is + the fallback start only for a session that never recorded one. Args: session_dir (Path): Absolute session root. @@ -630,31 +770,28 @@ def collect_session( """ start_ts = str(state.get("start_ts") or manifest.get("created_at_utc") or "") stop_reason = str(state.get("stop_reason") or "").strip() - close_stop_reason, close_ts = _close_phase_stop_reason(state) + close_stop_reason, close_ts = _close_phase_stop_reason(state, leg_start_ts=_leg_start_ts(state, start_ts)) if _should_use_close_stop_reason(stop_reason, close_stop_reason): stop_reason = close_stop_reason ended_at_utc = "" - if stop_reason: - ended_at_utc = iso_z(close_ts) if close_ts else now_iso(timespec="seconds") - elapsed_min: float | None = None - if start_ts: - try: - start = datetime.fromisoformat(start_ts.replace("Z", "+00:00")) - elapsed_min = (datetime.now(timezone.utc) - start).total_seconds() / 60.0 - except (ValueError, TypeError): - pass + if _session_has_ended(stop_reason): + # ``stop_ts`` is stamped once, when the reason is written, so a re-export + # of a finished session keeps reporting the same end. The CLOSE + # transition and the export clock are only next-best guesses. + ended_at_utc = _first_recorded_end(state.get("stop_ts"), close_ts) or now_iso(timespec="seconds") image = _detect_image_for_session(manifest) if image is None: warnings.append("image: not configured (set HYPERLOOM_IMAGE env var)") - return { + section = { "session_id": str(state.get("session_id") or manifest.get("session_id") or ""), "claw_session_id": manifest.get("claw_session_id") or state.get("claw_session_id"), "sandbox_user_id": manifest.get("sandbox_user_id") or state.get("sandbox_user_id"), "created_at_utc": manifest.get("created_at_utc") or start_ts, + "start_ts": start_ts, "ended_at_utc": ended_at_utc, "stop_reason": stop_reason, "max_minutes": int(state.get("max_minutes") or manifest.get("max_minutes") or 0), - "elapsed_minutes": round(elapsed_min, 2) if elapsed_min is not None else 0.0, + "elapsed_minutes": 0.0, "host": str(manifest.get("host") or ""), "image": image, "code_revision": str(manifest.get("code_revision") or ""), @@ -669,6 +806,41 @@ def collect_session( # Crash / interruption / resume history. "recovery": _collect_recovery(state), } + section["elapsed_minutes"] = session_elapsed_minutes(section) + return section + + +def _session_duration_seconds( + session_section: dict[str, Any], + manifest: dict[str, Any], +) -> int: + """How long the session ran, in whole seconds. + + Measures the same window as ``session.elapsed_minutes`` (see + :func:`collect_session`) so the machine field and the human-readable one + cannot disagree, then falls back to ``elapsed_minutes`` for callers that + supply it and no usable timestamps. + + Args: + session_section (dict[str, Any]): The resolved ``session`` dict. + manifest (dict[str, Any]): Parsed ``manifest.json``. + + Returns: + int: The duration, or ``0`` when it cannot be established. + """ + duration_s = _measured_duration_seconds( + session_section.get("start_ts") + or session_section.get("created_at_utc") + or manifest.get("created_at_utc"), + session_section.get("ended_at_utc"), + session_section.get("stop_reason"), + ) + if duration_s is not None: + return duration_s + elapsed_min = session_section.get("elapsed_minutes") + if isinstance(elapsed_min, (int, float)) and elapsed_min > 0: + return int(round(elapsed_min * 60)) + return 0 # session_meta enrichment @@ -682,6 +854,12 @@ def collect_session_meta( Emitted straight from the manifest + resolved ``session`` section; the CI step only gap-fills fields the sandbox could not know (e.g. ``category``). + The duration is measured from the session's own timestamps rather than + read from a sibling key. Two producers fill the ``session`` section -- the + live recorder's snapshot and this module's collector -- and only the + collector writes ``elapsed_minutes``, so a run recorded live reported a + session that lasted zero seconds. + Args: manifest (dict[str, Any]): Parsed ``manifest.json``. session_section (dict[str, Any]): The already-built ``session`` dict. @@ -693,8 +871,7 @@ def collect_session_meta( """ image = session_section.get("image") image_str = image if isinstance(image, str) and image.strip() else "" - elapsed_min = session_section.get("elapsed_minutes") - duration_s = int(round(elapsed_min * 60)) if isinstance(elapsed_min, (int, float)) and elapsed_min > 0 else 0 + duration_s = _session_duration_seconds(session_section, manifest) return { "code_revision": str(manifest.get("code_revision") or ""), "image": image_str or None, diff --git a/src/hyperloom/inference_optimizer/breakdown/exporter.py b/src/hyperloom/inference_optimizer/breakdown/exporter.py index f105ee9b21..18e531449c 100644 --- a/src/hyperloom/inference_optimizer/breakdown/exporter.py +++ b/src/hyperloom/inference_optimizer/breakdown/exporter.py @@ -91,6 +91,56 @@ def _merge_phase_timeline( return base +def _recorded_session_value(value: Any) -> bool: + """Whether a recorder ``session`` field carries evidence. + + The snapshot writes every key on every save, so an unset field arrives as + the type's empty value rather than as a missing key -- ``0`` for the + budget and the tick count exactly as ``""`` for the ids. Only a value that + says something may overwrite what the collector resolved. + + Args: + value (Any): A fragment field value. + + Returns: + bool: ``True`` when the field was actually recorded. + """ + if value is None or value == "": + return False + return not (isinstance(value, (int, float)) and not isinstance(value, bool) and value == 0) + + +def _merge_session(fragment: Any, collector_value: Any) -> Any: + """Overlay the recorder's live ``session`` fields on the collected section. + + The recorder snapshots what the running state knows -- ids, phase, tick, + the start and end timestamps -- while everything derived from + ``manifest.json`` (container image, host, pid) and everything derived from + the timestamps (``elapsed_minutes``) only exists on the collector side. + Replacing the section wholesale dropped those, so a live-recorded run + reported no wall-clock elapsed time and no image. An empty fragment value + is absence of evidence and never overwrites a collected one, but it still + lands on a key the collector does not fill (``phase``), which the section + carried before this merge existed. + + Args: + fragment: The recorder ``session`` fragment (may be any type). + collector_value: The collector-computed session section. + + Returns: + The merged section, or ``collector_value`` when no fragment was + recorded. + """ + if not isinstance(fragment, dict) or not fragment: + return collector_value + merged = dict(collector_value) if isinstance(collector_value, dict) else {} + for key, value in fragment.items(): + if _recorded_session_value(value) or key not in merged: + merged[key] = value + merged["elapsed_minutes"] = collectors.session_elapsed_minutes(merged) + return merged + + def _load_session_json(path: Path, label: str, warnings: list[str]) -> dict[str, Any]: """Read a session JSON file as a dict; ``{}`` + warning on failure. @@ -248,8 +298,9 @@ def _pick(section: str, collector_value: Any) -> Any: exported_at = datetime.now(timezone.utc).isoformat(timespec="seconds") # Section collectors (each catches its own errors via warnings). - session_section = _pick( - "session", _safe_collect("session", lambda: collectors.collect_session(sd, state, manifest, warnings), warnings) + session_section = _merge_session( + assembled.get("session"), + _safe_collect("session", lambda: collectors.collect_session(sd, state, manifest, warnings), warnings), ) # Author-side ``session_meta`` enrichment, emitted from the manifest + # resolved ``session`` block. diff --git a/src/hyperloom/inference_optimizer/breakdown/recorder/instrument.py b/src/hyperloom/inference_optimizer/breakdown/recorder/instrument.py index 2adb76bedf..44ef1d3e52 100644 --- a/src/hyperloom/inference_optimizer/breakdown/recorder/instrument.py +++ b/src/hyperloom/inference_optimizer/breakdown/recorder/instrument.py @@ -39,7 +39,7 @@ from hyperloom.common.coerce import to_float from hyperloom.common.jsonio import read_json -from hyperloom.common.timeutil import now_iso +from hyperloom.common.timeutil import iso_z, now_iso log = logging.getLogger(__name__) @@ -866,6 +866,15 @@ def _snapshot_v4_run(rec, st: Any) -> None: def _snapshot_session(rec, st: Any) -> None: """Snapshot the ``session`` singleton from ``st`` (no-op without a session id). + A session that has stopped carries ``ended_at_utc``, taken from the state's + own stop timestamp: without it the exporter has no end to measure against + and reports the run as still going. ``start_ts`` is what the exported + elapsed time is measured from; a resume re-anchors it on the new leg only + when the previous one crashed or stopped for a recorded reason, so after a + clean stop it still names the original start. The manifest-derived fields + the live state cannot know (image, host, pid) are filled in by the + collector at export. + Args: rec: the recorder used to write the singleton. st (Any): the live ``SharedState`` to snapshot. @@ -873,6 +882,7 @@ def _snapshot_session(rec, st: Any) -> None: session_id = str(getattr(st, "session_id", "") or "") if not session_id: return + stop_reason = str(getattr(st, "stop_reason", "") or "") rec.record_singleton( "session", { @@ -880,7 +890,10 @@ def _snapshot_session(rec, st: Any) -> None: "claw_session_id": getattr(st, "claw_session_id", "") or "", "sandbox_user_id": getattr(st, "sandbox_user_id", "") or "", "start_ts": str(getattr(st, "start_ts", "") or ""), - "stop_reason": str(getattr(st, "stop_reason", "") or ""), + # A resumed run clears its reason but not necessarily the stale + # timestamp, so the pair is only ever emitted together. + "ended_at_utc": iso_z(getattr(st, "stop_ts", "")) if stop_reason else "", + "stop_reason": stop_reason, "max_minutes": int(getattr(st, "max_minutes", 0) or 0), "tick_count": int(getattr(st, "tick", 0) or 0), "phase": str(getattr(st, "phase", "") or ""), diff --git a/src/hyperloom/inference_optimizer/breakdown/schema.py b/src/hyperloom/inference_optimizer/breakdown/schema.py index e3eed6e313..4bf8bc770e 100644 --- a/src/hyperloom/inference_optimizer/breakdown/schema.py +++ b/src/hyperloom/inference_optimizer/breakdown/schema.py @@ -83,13 +83,23 @@ class SessionMeta(TypedDict, total=False): session_id (str): Hyperloom internal id (``manifest.session_id``). claw_session_id (str | None): SaFE / Claw session id (env ``CLAW_SESSION_ID``). sandbox_user_id (str | None): Sandbox user identifier, if any. - created_at_utc (str): ISO UTC timestamp when the session started. + created_at_utc (str): ISO UTC timestamp when the session was first + created; unchanged by a resume. + start_ts (str): ISO UTC timestamp the wall-clock budget is counted + from. A resume re-anchors it on the new leg only when the previous + one crashed or stopped for a recorded reason; after a clean stop it + stays at the original start, because ``--max-hours`` keeps counting + from there. ended_at_utc (str): ISO UTC timestamp when the session ended. stop_reason (str): Why the run stopped (``target_reached`` / ``time_exhausted`` / ``global_converged`` / ``max_ticks`` / ``baseline_failed`` / ...). max_minutes (int): Configured time budget in minutes. - elapsed_minutes (float): Wall-clock minutes the session ran. + elapsed_minutes (float): Wall-clock minutes from ``start_ts`` to the + end, or to now while still running, so it stays comparable with + ``max_minutes``. When a resume kept ``start_ts`` this spans the + gap between the legs as well, which is the span the budget is + charged for too. host (str): Hostname the session executed on. code_revision (str): Source revision of the optimizer. pid (int): Process id of the optimizer. @@ -107,6 +117,7 @@ class SessionMeta(TypedDict, total=False): claw_session_id: str | None # SaFE / Claw session id (env CLAW_SESSION_ID) sandbox_user_id: str | None created_at_utc: str + start_ts: str # budget anchor; re-anchored only by a resume after a crash or a recorded stop ended_at_utc: str stop_reason: str # target_reached / time_exhausted / global_converged / max_ticks / baseline_failed / ... max_minutes: int @@ -376,8 +387,8 @@ class PhaseEvent(TypedDict, total=False): kernel_id (str | None): Kernel id for kernel_agent-owned actions, else None. status (str): Outcome (``succeeded`` / ``failed``). decision (str): Decision label (``promoted`` / ``discarded`` / - ``salvaged`` / ``no_promote`` / ``error`` / ``KEEP`` / ``PARTIAL`` / - ``REVERT``). + ``salvaged`` / ``no_promote`` / ``skipped`` / ``error`` / ``KEEP`` / + ``PARTIAL`` / ``REVERT``). key_metric (float | None): Headline metric value, or None. key_metric_kind (str | None): Type/label of the key metric, or None. workspace (str | None): Benchmark workspace path, or None. @@ -398,7 +409,7 @@ class PhaseEvent(TypedDict, total=False): task_id: str kernel_id: str | None # only for kernel_agent-owned actions status: str # succeeded / failed - decision: str # promoted / discarded / salvaged / no_promote / error / KEEP / PARTIAL / REVERT + decision: str # promoted / discarded / salvaged / no_promote / skipped / error / KEEP / PARTIAL / REVERT key_metric: float | None key_metric_kind: str | None workspace: str | None diff --git a/src/hyperloom/inference_optimizer/cli/__init__.py b/src/hyperloom/inference_optimizer/cli/__init__.py index ba53758b6c..c9a450bdfa 100644 --- a/src/hyperloom/inference_optimizer/cli/__init__.py +++ b/src/hyperloom/inference_optimizer/cli/__init__.py @@ -53,6 +53,7 @@ summarize_model_config, ) from .bootstrap import ( + _begin_resume_leg, _print_final_summary, _print_session_skeleton, _reconcile_crash_count, @@ -1757,7 +1758,6 @@ async def _run_optimize(args: argparse.Namespace) -> int: f"(cannot retroactively ungrade prior KEEPs)" ) - # CRITICAL: clear leftover stop_reason or Orchestration heartbeats forever thinking work is done. prior_crash = state.crash_count # target_reached is a terminal state requiring --force-resume to push @@ -1791,18 +1791,10 @@ async def _run_optimize(args: argparse.Namespace) -> int: ) sys.exit(2) - if prior_stop or prior_crash >= 3: - state.stop_reason = "" - state.closing_phase = False - state.closing_started_unix = 0.0 - state.closing_report_task_id = "" - # Reset persisted crash_count so a fresh resume isn't immediately tripped into "emergency". - state.crash_count = 0 - # Reset start_ts to now so resume budget isn't seen as already-over-budget by the LLM. - from datetime import datetime, timezone - - state.start_ts = datetime.now(timezone.utc).isoformat(timespec="microseconds") - state.save(session_dir) + reanchor_budget = bool(prior_stop or prior_crash >= 3) + _begin_resume_leg(state, reanchor_budget=reanchor_budget) + state.save(session_dir) + if reanchor_budget: override_note = " (--force-resume override)" if force_resume and prior_stop in gated_terminal else "" print(f" → cleared stop_reason and reset crash_count (was {prior_crash}) for fresh resume{override_note}") print(f" → reset start_ts to {state.start_ts} (resume budget)") diff --git a/src/hyperloom/inference_optimizer/cli/bootstrap.py b/src/hyperloom/inference_optimizer/cli/bootstrap.py index 962c2f417c..3584b8da7b 100644 --- a/src/hyperloom/inference_optimizer/cli/bootstrap.py +++ b/src/hyperloom/inference_optimizer/cli/bootstrap.py @@ -15,10 +15,14 @@ import logging import os import sys +import time from pathlib import Path from typing import Any +from hyperloom.common.coerce import to_unix from hyperloom.common.env import forge_explicitly_enabled +from hyperloom.common.timeutil import now_iso +from hyperloom.orchestrator.phases.machine_state import bank_phase_segment from hyperloom.orchestrator.state.shared_state import SharedState from .backends import _build_robustness_options from .parser import ( @@ -435,6 +439,74 @@ def _print_final_summary( print("===============================================") +def _bank_previous_leg_phase_segment(state: SharedState) -> None: + """Bank the phase time the stopped leg spent but never recorded. + + Per-phase totals are banked at each transition out of a phase, so a leg that + stopped mid-phase left its last segment live — and the resume boundary is + about to floor that segment away as the idle gap it mostly is. + :attr:`SharedState.stop_ts` is the only recorded evidence of when the leg + ended; a clean stop or a crash leaves none, and then the segment stays + unbanked. That under-charges the phase, which is the direction the phase + clock tolerates: over-charging ends a phase early. + + The end is clamped to the present for the same reason: no leg can have run + past the moment it is being resumed, so a ``stop_ts`` stamped ahead of now + would bank the difference as spend the phase never had. + + Must run before ``resumed_ts`` is restamped, which would floor the segment + to nothing. + + Args: + state (SharedState): The loaded session state, mutated in place. + """ + stop_unix = min(to_unix(state.stop_ts, 0.0) or 0.0, time.time()) + if stop_unix <= 0.0: + return + bank_phase_segment(state, until_unix=stop_unix) + + +def _begin_resume_leg(state: SharedState, *, reanchor_budget: bool) -> str: + """Mark the start of a resumed run leg on ``state`` (caller persists). + + Every resume stamps :attr:`SharedState.resumed_ts`. The previous leg's + CLOSE transition stays in ``phase_history`` and would otherwise keep + speaking for the resumed run — a report reads it as the session's stop + reason and end time — and this boundary is what dates it as a previous + leg's. It is also what stops the phase clock charging the gap between the + two legs to whichever phase the session stopped in. + + Only a previous leg that stopped for a recorded reason, or crashed + repeatedly, re-anchors the wall-clock budget. After a clean stop + ``start_ts`` is deliberately kept, so ``--max-hours`` still counts from the + original session start and the earlier legs' wall-clock stays spent. The + phase clock moves on either branch: the two answer different questions, and + neither answer includes time nothing was running. + + Args: + state (SharedState): The loaded session state, mutated in place. + reanchor_budget (bool): Whether the budget restarts from this leg. + + Returns: + str: The timestamp stamped as this leg's boundary. + """ + _bank_previous_leg_phase_segment(state) + state.resumed_ts = now_iso() + if reanchor_budget: + # CRITICAL: clear the leftover stop_reason or Orchestration heartbeats + # forever think the work is done. + state.stop_reason = "" + state.stop_ts = "" + state.closing_phase = False + state.closing_started_unix = 0.0 + state.closing_report_task_id = "" + # Reset persisted crash_count so a fresh resume isn't immediately tripped into "emergency". + state.crash_count = 0 + # Reset start_ts to now so resume budget isn't seen as already-over-budget by the LLM. + state.start_ts = state.resumed_ts + return state.resumed_ts + + def _reconcile_crash_count(state: SharedState, session_dir: Path) -> None: """Reconcile persisted ``crash_count`` (state.json + final.json) up to the live in-memory value. diff --git a/src/hyperloom/inference_optimizer/tests/test_agent_roles_and_policy.py b/src/hyperloom/inference_optimizer/tests/test_agent_roles_and_policy.py index 223a2dc83f..b6474f8f5c 100644 --- a/src/hyperloom/inference_optimizer/tests/test_agent_roles_and_policy.py +++ b/src/hyperloom/inference_optimizer/tests/test_agent_roles_and_policy.py @@ -696,6 +696,36 @@ def test_gate_update_state_closing_phase_and_baseline_config_rejected(gate): assert exc.value.rule == "state_field", field_name +def test_gate_update_state_cannot_move_the_resume_boundary(gate): + # resumed_ts dates the current run leg: moving it hands the previous leg's + # CLOSE transition back the right to speak for this one. + assert "resumed_ts" in CORE_STATE_FIELDS + with pytest.raises(PolicyDenied) as exc: + gate.validate_intent( + "orchestration", + Intent( + type=IntentType.UPDATE_STATE, + payload={"changes": {"resumed_ts": "2026-01-01T00:00:00+00:00"}}, + ), + ) + assert exc.value.rule == "state_field" + + +def test_gate_update_state_cannot_move_a_session_end_time(gate): + # stop_ts is the timestamp half of stop_reason, written by the same setter: + # locking only the reason lets a model post-date the session's end. + assert "stop_ts" in CORE_STATE_FIELDS + with pytest.raises(PolicyDenied) as exc: + gate.validate_intent( + "orchestration", + Intent( + type=IntentType.UPDATE_STATE, + payload={"changes": {"stop_ts": "2026-01-01T00:01:00+00:00"}}, + ), + ) + assert exc.value.rule == "state_field" + + def test_core_state_fields_synced_with_robustness_envelope(): # gate.CORE_STATE_FIELDS and the robustness # envelope copy must stay byte-identical. This direct assertion never skips diff --git a/src/hyperloom/inference_optimizer/tests/test_baseline_eval_fallback.py b/src/hyperloom/inference_optimizer/tests/test_baseline_eval_fallback.py index 7e41abd55e..d08e885261 100644 --- a/src/hyperloom/inference_optimizer/tests/test_baseline_eval_fallback.py +++ b/src/hyperloom/inference_optimizer/tests/test_baseline_eval_fallback.py @@ -7,6 +7,7 @@ import asyncio import json +import logging import subprocess from pathlib import Path from types import SimpleNamespace @@ -20,6 +21,8 @@ BaselineExecutor, ) +_BASELINE_LOGGER = "hyperloom.orchestrator.actions.executors.baseline" + @pytest.fixture(autouse=True) def _isolate_leak_root(tmp_path_factory, monkeypatch): @@ -751,6 +754,49 @@ def test_salvage_uses_a_warmup_score_when_it_is_the_only_one(tmp_path): assert reason == "" +def test_the_double_run_handoff_is_not_reported_as_a_recovery(tmp_path, caplog): + """Every healthy double-run baseline reads its accuracy from the warmup round. + + The measured round runs ``RUN_EVAL=false``, so it has no accuracy of its + own by construction. Reporting that handoff as a salvage -- in the log or + in the structured warnings the report and the specialists read -- made a + normal run look like it survived a fault. + """ + attempt = tmp_path / "runs" / "baseline" / "786a793e" + _write_gsm8k_results(attempt / "warmup_round", 0.9128) + deciding = attempt / "measure_round" + deciding.mkdir(parents=True, exist_ok=True) + + executor = BaselineExecutor() + rec = _StopRecorder() + result = {"status": "succeeded", "run_eval_disabled": False, "output_dir": str(deciding)} + with caplog.at_level(logging.INFO, logger=_BASELINE_LOGGER): + executor._maybe_stop_on_missing_baseline_accuracy(_stop_ctx("vllm", rec), result) + + assert rec.stop_reason == "" + assert result["accuracy"] == pytest.approx(0.9128) + assert not [r for r in caplog.records if r.levelno >= logging.WARNING and "salvag" in r.getMessage()] + assert any("cold-start guard" in r.getMessage() for r in caplog.records) + assert "baseline_accuracy_salvaged_from_sibling_attempt" not in result.get("nonfatal_warnings", []) + + +def test_an_unexpected_gap_is_still_reported_as_a_salvage(tmp_path, caplog): + """A retry attempt reading another attempt's score is a recovery, and says so.""" + runs_baseline = tmp_path / "runs" / "baseline" + _write_gsm8k_results(runs_baseline / "786a793e" / "measure_round", 0.9128) + deciding = runs_baseline / "retry2_bootsafe" + deciding.mkdir(parents=True, exist_ok=True) + + executor = BaselineExecutor() + rec = _StopRecorder() + result = {"status": "succeeded", "run_eval_disabled": False, "output_dir": str(deciding)} + with caplog.at_level(logging.INFO, logger=_BASELINE_LOGGER): + executor._maybe_stop_on_missing_baseline_accuracy(_stop_ctx("vllm", rec), result) + + assert any(r.levelno == logging.WARNING and "salvaged" in r.getMessage() for r in caplog.records) + assert "baseline_accuracy_salvaged_from_sibling_attempt" in result.get("nonfatal_warnings", []) + + def test_salvage_prefers_a_measured_round_over_a_warmup(tmp_path): """The warmup is a fallback, not a substitute: a real round still wins.""" runs_baseline = tmp_path / "runs" / "baseline" diff --git a/src/hyperloom/inference_optimizer/tests/test_breakdown_exporter_unit.py b/src/hyperloom/inference_optimizer/tests/test_breakdown_exporter_unit.py index 8ae748e017..547d972af5 100644 --- a/src/hyperloom/inference_optimizer/tests/test_breakdown_exporter_unit.py +++ b/src/hyperloom/inference_optimizer/tests/test_breakdown_exporter_unit.py @@ -6,11 +6,15 @@ from __future__ import annotations import json +from datetime import datetime, timedelta, timezone from pathlib import Path import pytest +from hyperloom.common.timeutil import iso_z from hyperloom.inference_optimizer.breakdown import exporter as ex +from hyperloom.inference_optimizer.breakdown.collectors import sessions +from hyperloom.inference_optimizer.breakdown.collectors.sessions import collect_session_meta # ---- _load_session_json ---- @@ -415,3 +419,369 @@ def test_orchestration_context_is_empty_without_a_census_or_db(tmp_path): assert section["compactions_per_tick"] == 0.0 assert section["context_tokens_at_compaction"] == {} assert warnings == [] + + +# ---- session_meta duration ---- + + +def _freeze_now(monkeypatch, instant: datetime) -> None: + """Pin the session collector's clock to *instant*. + + Args: + monkeypatch: The pytest monkeypatch fixture. + instant (datetime): The UTC instant every ``datetime.now`` call returns. + """ + + class _FrozenDatetime(datetime): + @classmethod + def now(cls, tz=None): + return instant.astimezone(tz) if tz else instant + + monkeypatch.setattr(sessions, "datetime", _FrozenDatetime) + + +def test_session_duration_is_measured_from_the_session_timestamps(): + """The live recorder's ``session`` snapshot carries no ``elapsed_minutes``.""" + meta = collect_session_meta( + {"code_revision": "abc1234"}, + { + "start_ts": "2026-08-08T00:37:27+00:00", + "ended_at_utc": "2026-08-08T02:55:27+00:00", + }, + [], + ) + assert meta["session_duration_seconds"] == 8280 + + +def test_a_running_session_is_measured_up_to_now(): + started = datetime.now(timezone.utc) - timedelta(minutes=10) + meta = collect_session_meta({}, {"start_ts": started.isoformat()}, []) + assert 590 <= meta["session_duration_seconds"] <= 620 + + +def test_a_session_that_has_not_stopped_yet_is_still_measured_up_to_now(monkeypatch): + _freeze_now(monkeypatch, datetime(2026, 8, 8, 1, 37, 27, tzinfo=timezone.utc)) + meta = collect_session_meta( + {}, + {"start_ts": "2026-08-08T00:37:27+00:00", "stop_reason": ""}, + [], + ) + assert meta["session_duration_seconds"] == 3600 + + +def test_a_stopped_session_without_an_end_timestamp_is_not_measured_up_to_now(monkeypatch): + """The live recorder's ``session`` snapshot of a crashed run has no end.""" + _freeze_now(monkeypatch, datetime(2026, 10, 20, 0, 37, 27, tzinfo=timezone.utc)) + meta = collect_session_meta( + {}, + {"start_ts": "2026-08-08T00:37:27+00:00", "stop_reason": "coordinator_exception"}, + [], + ) + assert meta["session_duration_seconds"] == 0 + + +def test_a_stopped_session_measures_the_same_however_late_it_is_exported(monkeypatch): + section = { + "start_ts": "2026-08-08T00:37:27+00:00", + "stop_reason": "time_exhausted", + "elapsed_minutes": 138.0, + } + _freeze_now(monkeypatch, datetime(2026, 8, 8, 3, 0, 0, tzinfo=timezone.utc)) + first = collect_session_meta({}, section, [])["session_duration_seconds"] + _freeze_now(monkeypatch, datetime(2026, 10, 20, 3, 0, 0, tzinfo=timezone.utc)) + second = collect_session_meta({}, section, [])["session_duration_seconds"] + assert first == second == 8280 + + +def test_elapsed_minutes_still_answers_when_no_timestamp_does(): + meta = collect_session_meta({}, {"elapsed_minutes": 12.5}, []) + assert meta["session_duration_seconds"] == 750 + + +def test_a_session_with_nothing_to_measure_reports_zero(): + assert collect_session_meta({}, {}, [])["session_duration_seconds"] == 0 + + +def _stopped_session(session_dir: Path, *, ran_for: timedelta, stopped_ago: timedelta = timedelta(0)): + """Write a stopped session's state so the recorder fragment is spooled. + + Args: + session_dir (Path): The session directory to write into. + ran_for (timedelta): How long the session ran before it stopped. + stopped_ago (timedelta): How long before now it stopped, so an export + measured to the recorded end can be told from one measured to now. + + Returns: + SharedState: The saved state. + """ + from hyperloom.orchestrator.state.shared_state import SharedState + + # Whole seconds: the exported end is canonicalised to second precision. + stopped_at = (datetime.now(timezone.utc) - stopped_ago).replace(microsecond=0) + state = SharedState.load_or_init(session_dir) + state.session_id = "sess-1178" + state.start_ts = (stopped_at - ran_for).isoformat(timespec="microseconds") + state.set_stop_reason("time_exhausted") + state.stop_ts = stopped_at.isoformat(timespec="microseconds") + state.save(session_dir) + return state + + +def test_a_recorded_session_exports_the_time_it_actually_ran(tmp_path): + """A session that stopped days ago still ran for two hours, however late it is exported.""" + state = _stopped_session(tmp_path, ran_for=timedelta(hours=2), stopped_ago=timedelta(days=3)) + + bd = ex.build(tmp_path) + assert bd["session"]["start_ts"] == state.start_ts + assert bd["session"]["ended_at_utc"] == iso_z(state.stop_ts) + assert bd["session_meta"]["session_duration_seconds"] == 7200 + + +def test_the_human_report_reads_the_same_elapsed_time_as_the_machine_field(tmp_path): + """``elapsed_minutes`` is the key the rendered report prints; the fragment used to drop it.""" + _stopped_session(tmp_path, ran_for=timedelta(hours=2)) + + bd = ex.build(tmp_path) + elapsed_minutes = bd["session"]["elapsed_minutes"] + assert elapsed_minutes == pytest.approx(bd["session_meta"]["session_duration_seconds"] / 60.0, abs=0.02) + assert 119.0 <= elapsed_minutes <= 121.0 + + +def test_the_recorder_path_keeps_the_fields_only_the_collector_can_resolve(tmp_path, monkeypatch): + """The image comes from the manifest / environment, which the live state never sees.""" + monkeypatch.setenv("HYPERLOOM_IMAGE", "registry.example/hyperloom:test") + _stopped_session(tmp_path, ran_for=timedelta(minutes=5)) + + bd = ex.build(tmp_path) + assert bd["session"]["image"] == "registry.example/hyperloom:test" + assert bd["session_meta"]["image"] == "registry.example/hyperloom:test" + assert bd["session"]["session_dir"] == str(tmp_path.resolve()) + + +def test_a_clean_stop_resume_keeps_measuring_from_the_original_start(tmp_path): + """--max-hours still counts from there after a clean stop, so the elapsed time does too.""" + now = datetime.now(timezone.utc) + state = { + "session_id": "sess-1178", + "start_ts": (now - timedelta(hours=5)).isoformat(), + "resumed_ts": (now - timedelta(hours=1)).isoformat(), + "max_minutes": 360, + } + + section = sessions.collect_session(tmp_path, state, {}, []) + # The four hours the session was not running are charged to the budget too. + assert 299.0 <= section["elapsed_minutes"] <= 301.0 + + +def test_elapsed_time_is_measured_from_the_resumed_start_not_the_first_launch(tmp_path): + """A resume after a stop re-anchors ``start_ts``, so elapsed restarts with the budget.""" + (tmp_path / "manifest.json").write_text( + json.dumps({"session_id": "sess-1178", "created_at_utc": "2026-08-01T00:00:00+00:00"}), + encoding="utf-8", + ) + _stopped_session(tmp_path, ran_for=timedelta(minutes=30)) + + bd = ex.build(tmp_path) + assert 29.0 <= bd["session"]["elapsed_minutes"] <= 31.0 + # The first launch is still on record, so the gap before the resume is visible. + assert bd["session"]["created_at_utc"] == "2026-08-01T00:00:00+00:00" + + +def test_a_resumed_session_is_not_reported_as_stopped_by_the_previous_legs_close(tmp_path): + """A resume clears the reason in state, but the old CLOSE row stays in ``phase_history``.""" + from hyperloom.orchestrator.state.shared_state import SharedState + + now = datetime.now(timezone.utc) + state = SharedState.load_or_init(tmp_path) + state.session_id = "sess-1178" + state.record_phase_transition( + to_phase="CLOSE", + reason="time_exhausted", + ts=(now - timedelta(days=6)).isoformat(timespec="seconds"), + ) + state.start_ts = (now - timedelta(minutes=30)).isoformat(timespec="microseconds") + state.record_phase_transition(to_phase="PRELUDE", reason="resumed", ts=state.start_ts) + state.save(tmp_path) + + bd = ex.build(tmp_path) + assert bd["session"]["stop_reason"] == "" + assert bd["session"]["ended_at_utc"] == "" + assert 29.0 <= bd["session"]["elapsed_minutes"] <= 31.0 + + +def test_a_session_resumed_after_a_clean_stop_is_still_reported_as_running(tmp_path): + """A clean stop keeps ``start_ts``, so the previous leg's CLOSE sits inside the window.""" + now = datetime.now(timezone.utc) + state = { + "session_id": "sess-1178", + "start_ts": (now - timedelta(hours=3)).isoformat(), + "stop_reason": "", + "phase": "CLOSE", + "resumed_ts": (now - timedelta(hours=2)).isoformat(), + "phase_history": [ + { + "to_phase": "CLOSE", + "reason": "time_exhausted", + "ts": (now - timedelta(hours=2, minutes=30)).isoformat(), + } + ], + } + + section = sessions.collect_session(tmp_path, state, {}, []) + assert section["stop_reason"] == "" + assert section["ended_at_utc"] == "" + assert 179.0 <= section["elapsed_minutes"] <= 181.0 + + +def test_a_close_the_state_file_never_recorded_still_supplies_the_end(tmp_path): + """The fallback's own case: the run stopped and only ``phase_history`` knows why.""" + now = datetime.now(timezone.utc) + state = { + "session_id": "sess-1178", + "start_ts": (now - timedelta(hours=2)).isoformat(), + "phase_history": [ + { + "to_phase": "CLOSE", + "reason": "target_reached", + "ts": (now - timedelta(minutes=5)).isoformat(), + } + ], + } + + section = sessions.collect_session(tmp_path, state, {}, []) + assert section["stop_reason"] == "target_reached" + assert section["ended_at_utc"] != "" + assert 114.0 <= section["elapsed_minutes"] <= 116.0 + + +@pytest.mark.parametrize( + "start_ts, close_ts", + [ + ("", "2026-08-01T00:00:00+00:00"), + ("2026-08-01T00:00:00+00:00", ""), + ("2026-08-01T00:00:00+00:00", "not-a-timestamp"), + # A CLOSE at the boundary belongs to the leg that started there. + ("2026-08-01T00:00:00+00:00", "2026-08-01T00:00:00+00:00"), + ], +) +def test_a_close_stands_when_there_is_nothing_comparable_to_disqualify_it(tmp_path, start_ts, close_ts): + """Only two parseable timestamps can place a CLOSE in a previous leg.""" + state = { + "session_id": "sess-1178", + "start_ts": start_ts, + "phase_history": [{"to_phase": "CLOSE", "reason": "target_reached", "ts": close_ts}], + } + + assert sessions.collect_session(tmp_path, state, {}, [])["stop_reason"] == "target_reached" + + +def test_the_newest_close_of_the_leg_supplies_the_end(tmp_path): + """A cyclic run reaches CLOSE more than once; the last word is the session's.""" + now = datetime.now(timezone.utc) + state = { + "session_id": "sess-1178", + "start_ts": (now - timedelta(hours=4)).isoformat(), + "phase_history": [ + {"to_phase": "CLOSE", "reason": "conc_sweep_done", "ts": (now - timedelta(hours=3)).isoformat()}, + {"to_phase": "EXPLORE", "reason": "cycle_reloop", "ts": (now - timedelta(hours=2)).isoformat()}, + {"to_phase": "CLOSE", "reason": "target_reached", "ts": (now - timedelta(minutes=10)).isoformat()}, + ], + } + + section = sessions.collect_session(tmp_path, state, {}, []) + assert section["stop_reason"] == "target_reached" + assert 229.0 <= section["elapsed_minutes"] <= 231.0 + + +def test_a_history_written_out_of_order_still_supplies_the_end(tmp_path): + """The scan walks back from the newest row, so a stale one must not end the search.""" + now = datetime.now(timezone.utc) + state = { + "session_id": "sess-1178", + "start_ts": (now - timedelta(hours=2)).isoformat(), + "resumed_ts": (now - timedelta(hours=2)).isoformat(), + "phase_history": [ + {"to_phase": "CLOSE", "reason": "target_reached", "ts": (now - timedelta(minutes=5)).isoformat()}, + {"to_phase": "CLOSE", "reason": "time_exhausted", "ts": (now - timedelta(days=3)).isoformat()}, + ], + } + + assert sessions.collect_session(tmp_path, state, {}, [])["stop_reason"] == "target_reached" + + +def test_an_unreadable_stop_time_does_not_become_the_session_end(tmp_path): + """Pre-``stop_ts`` this branch stamped the export clock; a bad value must not read as an end.""" + now = datetime.now(timezone.utc) + state = { + "session_id": "sess-1178", + "start_ts": (now - timedelta(hours=2)).isoformat(), + "stop_reason": "target_reached", + "stop_ts": "not-a-timestamp", + } + + section = sessions.collect_session(tmp_path, state, {}, []) + assert section["ended_at_utc"] != "not-a-timestamp" + assert 119.0 <= section["elapsed_minutes"] <= 121.0 + + +def test_an_unreadable_stop_time_falls_back_to_the_close_transition(tmp_path): + now = datetime.now(timezone.utc) + closed_at = (now - timedelta(minutes=30)).isoformat() + state = { + "session_id": "sess-1178", + "start_ts": (now - timedelta(hours=2)).isoformat(), + "stop_reason": "target_reached", + "stop_ts": "not-a-timestamp", + "phase_history": [{"to_phase": "CLOSE", "reason": "target_reached", "ts": closed_at}], + } + + section = sessions.collect_session(tmp_path, state, {}, []) + assert section["ended_at_utc"] == iso_z(closed_at) + assert 89.0 <= section["elapsed_minutes"] <= 91.0 + + +# ---- _merge_session ---- + + +def test_the_recorder_fragment_overlays_the_collected_section(): + merged = ex._merge_session( + {"session_id": "sess-1178", "stop_reason": "target_reached"}, + {"session_id": "", "stop_reason": "", "image": "registry.example/hyperloom:test"}, + ) + assert merged["session_id"] == "sess-1178" + assert merged["stop_reason"] == "target_reached" + assert merged["image"] == "registry.example/hyperloom:test" + + +def test_a_section_with_no_fragment_is_returned_untouched(): + section = {"session_id": "sess-1178"} + assert ex._merge_session(None, section) is section + assert ex._merge_session({}, section) is section + + +def test_an_unrecorded_budget_does_not_erase_the_collected_one(): + """The snapshot writes every key on every save, so an unset int arrives as 0.""" + merged = ex._merge_session( + {"max_minutes": 0, "tick_count": 0}, + {"max_minutes": 360, "tick_count": 12}, + ) + assert merged["max_minutes"] == 360 + assert merged["tick_count"] == 12 + + +def test_the_live_phase_stays_in_the_section_even_when_blank(): + """Only the recorder knows the phase; the key was always present before the merge.""" + merged = ex._merge_session({"phase": ""}, {"session_id": "sess-1178"}) + assert merged["phase"] == "" + + +def test_the_merged_section_measures_its_own_elapsed_time(): + merged = ex._merge_session( + { + "start_ts": "2026-08-08T00:00:00+00:00", + "ended_at_utc": "2026-08-08T02:00:00+00:00", + "stop_reason": "target_reached", + }, + {"elapsed_minutes": 0.0}, + ) + assert merged["elapsed_minutes"] == 120.0 diff --git a/src/hyperloom/inference_optimizer/tests/test_breakdown_v4_core.py b/src/hyperloom/inference_optimizer/tests/test_breakdown_v4_core.py index c9b06a06ce..0bf3d880f0 100644 --- a/src/hyperloom/inference_optimizer/tests/test_breakdown_v4_core.py +++ b/src/hyperloom/inference_optimizer/tests/test_breakdown_v4_core.py @@ -487,6 +487,42 @@ def test_v4_state_snapshot_does_not_infer_canonical_entities_from_stack(tmp_path assert len(assemble_parts(tmp_path)["optimization_stack"]) == 3 +def _session_state(*, stop_reason: str, stop_ts: str) -> SimpleNamespace: + """A minimal state carrying only what the ``session`` snapshot reads. + + Args: + stop_reason (str): The state's stop reason (empty while running). + stop_ts (str): The state's recorded stop timestamp. + + Returns: + SimpleNamespace: The state stand-in to snapshot. + """ + return SimpleNamespace( + session_id="session-5", + claw_session_id="claw-5", + sandbox_user_id="sandbox-5", + start_ts="2026-07-22T10:00:00Z", + stop_reason=stop_reason, + stop_ts=stop_ts, + max_minutes=30, + tick=5, + phase="CLOSE", + ) + + +def test_session_snapshot_carries_the_end_time_of_a_stopped_run(tmp_path): + snapshot_state_sections(tmp_path, _session_state(stop_reason="time_exhausted", stop_ts="2026-07-22T10:30:00+00:00")) + session = assemble_parts(tmp_path)["session"] + assert session["ended_at_utc"] == "2026-07-22T10:30:00Z" + + +def test_session_snapshot_leaves_a_running_run_without_an_end_time(tmp_path): + # A resume clears the reason, so a leftover stop_ts must not end the session. + snapshot_state_sections(tmp_path, _session_state(stop_reason="", stop_ts="2026-07-22T10:30:00+00:00")) + session = assemble_parts(tmp_path)["session"] + assert session["ended_at_utc"] == "" + + def test_v4_critic_hook_mirrors_structured_kb_writes(tmp_path): record_critic_iteration( tmp_path, diff --git a/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py b/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py index 142b510304..a423bab234 100644 --- a/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py +++ b/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py @@ -6,6 +6,8 @@ import argparse import json +import time +from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace from types import SimpleNamespace @@ -419,6 +421,93 @@ def test_snapshot_skeleton_and_session_dir_helpers( assert cb._resolve_session_dir_for_summary(None) is None +def test_a_clean_stop_resume_records_where_the_new_leg_began() -> None: + """start_ts stays the budget anchor, so the resume timestamp is the only leg boundary.""" + state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00", crash_count=1) + + cb._begin_resume_leg(state, reanchor_budget=False) + + assert state.start_ts == "2026-08-01T00:00:00+00:00" + assert state.resumed_ts > state.start_ts + assert state.crash_count == 1 + + +def test_a_resume_after_a_stop_re_anchors_the_budget_on_the_new_leg() -> None: + state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00", crash_count=4) + state.set_stop_reason("time_exhausted") + state.closing_phase = True + + cb._begin_resume_leg(state, reanchor_budget=True) + + assert state.start_ts == state.resumed_ts + assert state.stop_reason == "" + assert state.stop_ts == "" + assert state.closing_phase is False + assert state.crash_count == 0 + + +def test_a_resume_banks_what_the_stopped_leg_spent_in_its_phase() -> None: + """A phase segment is only durable once banked, and stopping never banks it.""" + state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00") + state.phase = "PRELUDE" + state.phase_started_ts = "2026-08-01T00:00:00+00:00" + state.phase_started_unix = 1785_542_400.0 + state.set_stop_reason("time_exhausted") + # Pin where the leg ended so the banked segment is a checkable number. + state.stop_ts = "2026-08-01T00:30:00+00:00" + + cb._begin_resume_leg(state, reanchor_budget=True) + + assert state.phase_elapsed_totals["PRELUDE"] == 1800.0 + assert state.stop_ts == "" + + +def test_a_second_resume_banks_only_the_leg_that_just_stopped() -> None: + """The first leg's segment is already durable; re-banking it would double-charge the phase.""" + state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00") + state.phase = "PRELUDE" + state.phase_started_ts = "2026-08-01T00:00:00+00:00" + state.phase_started_unix = 1785_542_400.0 + state.set_stop_reason("time_exhausted") + state.stop_ts = "2026-08-01T00:30:00+00:00" + cb._begin_resume_leg(state, reanchor_budget=True) + + # A second leg picked up a day later and ran an hour, still in PRELUDE. + state.resumed_ts = "2026-08-02T00:00:00+00:00" + state.set_stop_reason("time_exhausted") + state.stop_ts = "2026-08-02T01:00:00+00:00" + cb._begin_resume_leg(state, reanchor_budget=True) + + assert state.phase_elapsed_totals["PRELUDE"] == 1800.0 + 3600.0 + + +def test_a_resume_does_not_bank_a_stop_stamped_after_the_present() -> None: + """Banking past now charges the phase for time no leg ran, which ends it early.""" + started = time.time() - 60.0 + state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00") + state.phase = "PRELUDE" + state.phase_started_ts = datetime.fromtimestamp(started, tz=timezone.utc).isoformat() + state.phase_started_unix = started + state.set_stop_reason("time_exhausted") + state.stop_ts = datetime.fromtimestamp(started + 10 * 86400.0, tz=timezone.utc).isoformat() + + cb._begin_resume_leg(state, reanchor_budget=True) + + assert 60.0 <= state.phase_elapsed_totals["PRELUDE"] < 120.0 + + +def test_a_resume_with_no_recorded_stop_leaves_the_segment_unbanked() -> None: + """A clean stop records no end time; under-charge the phase rather than guess one.""" + state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00") + state.phase = "PRELUDE" + state.phase_started_ts = "2026-08-01T00:00:00+00:00" + state.phase_started_unix = 1785_542_400.0 + + cb._begin_resume_leg(state, reanchor_budget=False) + + assert state.phase_elapsed_totals == {} + + def test_reconcile_crash_count_updates_state_and_final_json(tmp_path: Path) -> None: state = SharedState(session_id="s", crash_count=5) SharedState(session_id="s", crash_count=1).save(tmp_path) diff --git a/src/hyperloom/inference_optimizer/tests/test_coerce.py b/src/hyperloom/inference_optimizer/tests/test_coerce.py index 180ff20101..09358f6acb 100644 --- a/src/hyperloom/inference_optimizer/tests/test_coerce.py +++ b/src/hyperloom/inference_optimizer/tests/test_coerce.py @@ -4,6 +4,7 @@ from __future__ import annotations import math +import time import pytest @@ -145,6 +146,18 @@ def test_iso_offset(self): def test_numeric_string_fallback(self): assert to_unix("1700000000") == 1_700_000_000.0 + def test_a_timestamp_without_an_offset_is_utc(self, monkeypatch): + """iso_z reads a naive timestamp as UTC; both feed the same comparisons.""" + monkeypatch.setenv("TZ", "Asia/Shanghai") + time.tzset() + try: + ts = to_unix("2021-01-01T00:00:00") + finally: + monkeypatch.delenv("TZ") + time.tzset() + assert ts is not None + assert math.isclose(ts, 1_609_459_200.0) + @pytest.mark.parametrize("value", [None, True, False, "not-a-ts", object()]) def test_reject_to_default(self, value): assert to_unix(value) is None diff --git a/src/hyperloom/inference_optimizer/tests/test_conc_sweep.py b/src/hyperloom/inference_optimizer/tests/test_conc_sweep.py index 3b1e6a4d77..1c16fac327 100644 --- a/src/hyperloom/inference_optimizer/tests/test_conc_sweep.py +++ b/src/hyperloom/inference_optimizer/tests/test_conc_sweep.py @@ -28,6 +28,7 @@ _flush_partial_conc_sweep_report, _has_optimization, _order_concs_desc, + conc_sweep_declined_to_run, run_conc_sweep, ) from hyperloom.common.gain_math import conc_pair_comparison as _build_comparison @@ -710,6 +711,8 @@ async def _fake_run_grid(*, grid: list[GridVariant], **_kw): assert payload["skip_reason"] == "budget_exhausted_no_successful_pairs" assert payload["budget_exhausted"] is True assert payload["budget_skip_reason"] == "insufficient_remaining_for_variant" + # This sweep started; only the pre-flight envelope means "declined to run". + assert conc_sweep_declined_to_run(payload) is False # ActionExecutor integration (SWEEP-phase dispatch) @@ -875,6 +878,7 @@ async def _fake_run(*_a, **_kw): assert result["status"] == "succeeded" assert result["was_skipped"] is True assert result["skip_reason"] == "no_baseline_tput" + assert conc_sweep_declined_to_run(result) is True # record_conc_sweep writes state.last_conc_sweep for SWEEP completion detection. @@ -949,6 +953,36 @@ class _State: assert evidence.get("conc_sweep_status") == "failed" +def test_the_sweep_exit_evidence_separates_a_skip_from_a_spent_budget(): + """``was_skipped`` covers both outcomes, so the row must carry what tells them apart.""" + from hyperloom.orchestrator.phases.machine_state import exit_normal_sweep + + state = SharedState( + phase="SWEEP", + phase_started_ts="2026-06-02T10:00:00+00:00", + max_minutes=360, + phase_budget_pct={"SWEEP": 0.50}, + ) + state.record_conc_sweep( + {"status": "skipped", "was_skipped": True, "skip_reason": "no_optimization_to_compare"} + ) + _, declined = exit_normal_sweep(state) + assert declined["conc_sweep_was_skipped"] is True + assert declined["conc_sweep_budget_exhausted"] is False + + state.record_conc_sweep( + { + "status": "skipped", + "was_skipped": True, + "budget_exhausted": True, + "skip_reason": "budget_exhausted_no_successful_pairs", + } + ) + _, spent = exit_normal_sweep(state) + assert spent["conc_sweep_was_skipped"] is True + assert spent["conc_sweep_budget_exhausted"] is True + + def test_on_enter_sweep_drains_pending_keep_integrates(monkeypatch): """Bug #7: KERNEL→SWEEP must drain pending KEEP integrates before closeout.""" from unittest.mock import AsyncMock, MagicMock diff --git a/src/hyperloom/inference_optimizer/tests/test_optimization_journal.py b/src/hyperloom/inference_optimizer/tests/test_optimization_journal.py index 70c0d05555..c301f2cf4d 100644 --- a/src/hyperloom/inference_optimizer/tests/test_optimization_journal.py +++ b/src/hyperloom/inference_optimizer/tests/test_optimization_journal.py @@ -22,6 +22,7 @@ OUTCOME_KEEP, OUTCOME_NO_PROMOTE, OUTCOME_REVERT, + OUTCOME_SKIP, classify_change_kind, derive_journal_outcome, summarize_change, @@ -421,6 +422,24 @@ def test_derive_journal_outcome_other_kinds_keep_binary_behaviour(): assert derive_journal_outcome("profile", {"status": "reverted"}, promotable=True) == OUTCOME_KEEP +def test_a_step_that_declined_to_run_is_neither_a_keep_nor_a_dead_end(): + """A conc_sweep with nothing to compare succeeds without doing anything.""" + out = derive_journal_outcome( + "conc_sweep", + {"status": "succeeded", "was_skipped": True, "skip_reason": "no_optimization_to_compare"}, + promotable=True, + ) + assert out == OUTCOME_SKIP + + +def test_a_stray_was_skipped_cannot_demote_a_kept_patch(): + """No integrate_patch producer sets the key; a future one must not silently rewrite the verdict.""" + assert ( + derive_journal_outcome("integrate_patch", {"status": "kept", "was_skipped": True}, promotable=True) + == OUTCOME_KEEP + ) + + def test_operation_kind_for_maps_kind_and_action(): from hyperloom.orchestrator.state.optimization_journal import ( operation_kind_for, diff --git a/src/hyperloom/inference_optimizer/tests/test_phase_cumulative_budget.py b/src/hyperloom/inference_optimizer/tests/test_phase_cumulative_budget.py index 775bd2914f..4f282e539c 100644 --- a/src/hyperloom/inference_optimizer/tests/test_phase_cumulative_budget.py +++ b/src/hyperloom/inference_optimizer/tests/test_phase_cumulative_budget.py @@ -13,10 +13,16 @@ These tests pin the fixed contract: ``phase_cumulative_seconds`` totals every entry, the cap/budget guards read that total, and ``phase_elapsed_seconds`` keeps its per-entry meaning for renderers and evidence dicts. + +They also pin the resume half of it: a phase entry is not closed by the process +exiting, so the current entry spans the idle gap between two run legs unless the +leg boundary floors it. """ from __future__ import annotations +from datetime import datetime, timezone + import pytest from hyperloom.orchestrator.phases import machine_state as ps @@ -37,6 +43,19 @@ T0 = 1_700_000_000.0 T0_ISO = "2023-11-14T22:13:20+00:00" +# A 3h session that stopped half an hour into PRELUDE and was resumed 3 days +# later: long enough that charging the gap to PRELUDE dwarfs the whole budget. +RESUMED_SESSION_MINUTES = 180 +PRELUDE_LEG_SEC = 1800.0 +IDLE_GAP_SEC = 3 * 24 * 3600.0 +RESUME_UNIX = T0 + PRELUDE_LEG_SEC + IDLE_GAP_SEC +# What the resumed leg has run by the time the guards are asked. +NEW_LEG_SEC = 120.0 + + +def _iso(at_unix: float) -> str: + return datetime.fromtimestamp(at_unix, tz=timezone.utc).isoformat() + def _kernel_state() -> SharedState: state = SharedState() @@ -213,6 +232,84 @@ def test_history_rebuild_skips_unusable_rows(): assert ps.phase_elapsed_totals_from_history("nope") == {} +def _resumed_prelude_state() -> SharedState: + """A session still in PRELUDE whose previous leg stopped 3 days ago.""" + state = SharedState() + state.max_minutes = RESUMED_SESSION_MINUTES + state.phase = ps.PHASE_PRELUDE + state.phase_started_unix = T0 + state.phase_started_ts = T0_ISO + state.resumed_ts = _iso(RESUME_UNIX) + return state + + +def test_a_resume_does_not_charge_the_phase_for_the_gap_it_was_not_running(): + # phase_started_unix is only rewritten on a phase transition, and exiting + # the process is not one, so the current entry spans both legs. + state = _resumed_prelude_state() + now = RESUME_UNIX + NEW_LEG_SEC + + assert ps.phase_elapsed_seconds(state, now_unix=now) == pytest.approx(NEW_LEG_SEC) + assert ps.phase_cumulative_seconds(state, now_unix=now) == pytest.approx(NEW_LEG_SEC) + + +def test_a_resumed_phase_is_not_capped_by_time_the_process_was_down(): + state = _resumed_prelude_state() + state.phase_budget_pct = {ps.PHASE_PRELUDE: 0.4} + now = RESUME_UNIX + NEW_LEG_SEC + + assert ps.phase_cap_seconds(state) == pytest.approx(RESUMED_SESSION_MINUTES * 60.0 * 0.4) + assert ps.phase_cap_exceeded(state, now_unix=now) is False + + +def test_the_charge_back_base_of_a_resumed_phase_stays_inside_the_session(): + # The crash / recorded-stop branch re-anchors start_ts, so the session clock + # restarts. A phase clock that still spans the gap reconstructs a base of + # "the whole budget plus three days" and hands the phase an allotment the + # session cannot pay for. + state = _resumed_prelude_state() + state.start_ts = state.resumed_ts + now = RESUME_UNIX + NEW_LEG_SEC + + budget = ps.normalize_budget_pct(None) + denom = sum(pct for pct in budget.values() if pct > 0.0) + session_sec = RESUMED_SESSION_MINUTES * 60.0 + total = ps._phase_budget_total_seconds(state, now_unix=now) + + assert total == pytest.approx(session_sec * budget[ps.PHASE_PRELUDE] / denom) + + +def test_a_kept_budget_anchor_charges_the_resumed_phase_the_smaller_base(): + # The clean-stop branch keeps start_ts, so the session stays charged for the + # idle gap while the phase clock no longer is. The base is then what the + # session had left when THIS leg began, not when the phase was entered. + idle_gap = 3600.0 + state = _resumed_prelude_state() + state.start_ts = T0_ISO + state.resumed_ts = _iso(T0 + PRELUDE_LEG_SEC + idle_gap) + now = T0 + PRELUDE_LEG_SEC + idle_gap + NEW_LEG_SEC + + budget = ps.normalize_budget_pct(None) + denom = sum(pct for pct in budget.values() if pct > 0.0) + base = RESUMED_SESSION_MINUTES * 60.0 - (PRELUDE_LEG_SEC + idle_gap) + total = ps._phase_budget_total_seconds(state, now_unix=now) + + assert total == pytest.approx(base * budget[ps.PHASE_PRELUDE] / denom) + + +def test_a_later_phase_entry_supersedes_the_resume_boundary(): + # The leg boundary only floors the entry it interrupted; once the phase is + # re-entered its own stamp is the later of the two. + state = _resumed_prelude_state() + _enter(state, ps.PHASE_FRAMEWORK_AGENT, RESUME_UNIX + NEW_LEG_SEC) + + assert state.phase_elapsed_totals[ps.PHASE_PRELUDE] == pytest.approx(NEW_LEG_SEC) + assert ps.phase_elapsed_seconds( + state, + now_unix=RESUME_UNIX + NEW_LEG_SEC + 180.0, + ) == pytest.approx(180.0) + + def test_budget_exit_evidence_reports_the_time_it_judged_on(): """A cap decided on cumulative time must not be evidenced by one entry's clock. diff --git a/src/hyperloom/inference_optimizer/tests/test_phase_state_plateau.py b/src/hyperloom/inference_optimizer/tests/test_phase_state_plateau.py index f3c352aa42..70114b558b 100644 --- a/src/hyperloom/inference_optimizer/tests/test_phase_state_plateau.py +++ b/src/hyperloom/inference_optimizer/tests/test_phase_state_plateau.py @@ -32,6 +32,7 @@ is_valid_stop_reason, kernel_work_pending, ) +from hyperloom.orchestrator.state import shared_state from hyperloom.orchestrator.state.shared_state import SharedState @@ -613,6 +614,7 @@ def test_set_stop_reason_accepts_vocab(): s = SharedState() assert s.set_stop_reason("target_reached") == "target_reached" assert s.stop_reason == "target_reached" + assert s.stop_ts != "" def test_set_stop_reason_lenient_maps_unknown_to_unknown(caplog): @@ -635,6 +637,38 @@ def test_set_stop_reason_empty_string_clears(): assert s.stop_reason == "target_reached" s.set_stop_reason("") assert s.stop_reason == "" + assert s.stop_ts == "" + + +def test_a_later_stop_reason_does_not_move_the_stop_time(monkeypatch): + """CLOSE stops the session on entry and ships the breakdown; a later write must not re-date it.""" + s = SharedState() + monkeypatch.setattr(shared_state, "_now_iso", lambda: "2026-08-08T00:00:00.000000+00:00") + s.set_stop_reason("time_exhausted") + monkeypatch.setattr(shared_state, "_now_iso", lambda: "2026-08-08T02:00:00.000000+00:00") + s.set_stop_reason("target_reached") + assert s.stop_reason == "target_reached" + assert s.stop_ts == "2026-08-08T00:00:00.000000+00:00" + + +def test_rewriting_the_same_stop_reason_does_not_move_the_stop_time(monkeypatch): + """The Coordinator's ``finally`` re-asserts the reason CLOSE already wrote.""" + s = SharedState() + monkeypatch.setattr(shared_state, "_now_iso", lambda: "2026-08-08T00:00:00.000000+00:00") + s.set_stop_reason("time_exhausted") + monkeypatch.setattr(shared_state, "_now_iso", lambda: "2026-08-08T00:04:00.000000+00:00") + s.set_stop_reason(s.stop_reason) + assert s.stop_ts == "2026-08-08T00:00:00.000000+00:00" + + +def test_saving_a_stopped_session_again_does_not_move_its_stop_time(tmp_path): + s = SharedState() + s.set_stop_reason("target_reached") + pinned = s.stop_ts + s.save(tmp_path) + s.save(tmp_path) + assert s.stop_ts == pinned + assert SharedState.load_or_init(tmp_path).stop_ts == pinned def test_stop_reason_vocab_has_v08_additions(): diff --git a/src/hyperloom/inference_optimizer/tests/test_report.py b/src/hyperloom/inference_optimizer/tests/test_report.py index 998bc16b98..0e5e3f3b24 100644 --- a/src/hyperloom/inference_optimizer/tests/test_report.py +++ b/src/hyperloom/inference_optimizer/tests/test_report.py @@ -251,6 +251,53 @@ def test_explain_stop_reason_unknown_is_empty(): assert rp._explain_stop_reason("") == "" +class _SweepState: + def __init__(self, last_conc_sweep): + self.last_conc_sweep = last_conc_sweep + + +def test_a_skipped_sweep_is_not_described_as_a_finished_one(): + """``conc_sweep_done`` is also the exit for a sweep that declined to run.""" + state = _SweepState({"status": "succeeded", "was_skipped": True, "skip_reason": "no_optimization_to_compare"}) + msg = rp._explain_stop_reason("conc_sweep_done", state) + assert "did not run" in msg + assert "no_optimization_to_compare" in msg + + +def test_a_sweep_that_ran_keeps_the_plain_explanation(): + state = _SweepState({"status": "succeeded", "was_skipped": False}) + assert rp._explain_stop_reason("conc_sweep_done", state) == rp._explain_stop_reason("conc_sweep_done") + + +def test_a_skip_with_no_recorded_reason_still_says_it_was_skipped(): + state = _SweepState({"status": "succeeded", "was_skipped": True, "skip_reason": ""}) + assert "did not run" in rp._explain_stop_reason("conc_sweep_done", state) + + +def test_a_sweep_that_spent_its_budget_is_not_reported_as_one_that_never_ran(tmp_path): + """The budget path records was_skipped for a sweep that ran its whole ladder.""" + from hyperloom.orchestrator.state.shared_state import SharedState + + live = SharedState() + live.record_conc_sweep( + { + "status": "skipped", + "was_skipped": True, + "budget_exhausted": True, + "skip_reason": "budget_exhausted_no_successful_pairs", + } + ) + live.save(tmp_path) + # The report is written from a reloaded state, so the flag that separates + # the two skips has to survive the round trip to be readable at all. + state = SharedState.load_or_init(tmp_path) + + msg = rp._explain_stop_reason("conc_sweep_done", state) + assert "did not run" not in msg + assert "budget" in msg + assert "budget_exhausted_no_successful_pairs" in msg + + def test_format_md_renders_stop_explanation(): md = rp._format_md( { diff --git a/src/hyperloom/inference_optimizer/tests/test_shared_state_units.py b/src/hyperloom/inference_optimizer/tests/test_shared_state_units.py index 490ad388d8..f6f702046c 100644 --- a/src/hyperloom/inference_optimizer/tests/test_shared_state_units.py +++ b/src/hyperloom/inference_optimizer/tests/test_shared_state_units.py @@ -314,6 +314,21 @@ def test_core_field_dropped_when_allow_core_false(self): assert s.current_action == "baseline" assert s.cumulative_gain == before # core write dropped + def test_a_stop_time_cannot_be_written_apart_from_its_reason(self): + # stop_reason is a core field, so a changes dict that carries both must + # not land the timestamp half either: the pair is what the export reads + # as "the session ended then, for this reason". + s = SharedState() + s.set_stop_reason("time_exhausted") + pinned = s.stop_ts + applied = s.apply_changes( + {"stop_reason": "target_reached", "stop_ts": "2026-01-01T00:01:00+00:00"}, + allow_core=False, + ) + assert applied == {} + assert s.stop_reason == "time_exhausted" + assert s.stop_ts == pinned + def test_core_field_written_when_allow_core_true(self): s = SharedState() applied = s.apply_changes({"cumulative_gain": 999.0}, allow_core=True) diff --git a/src/hyperloom/inference_optimizer/tests/test_trajectory_reviewer.py b/src/hyperloom/inference_optimizer/tests/test_trajectory_reviewer.py index 189c73a38d..a4b8a0ed84 100644 --- a/src/hyperloom/inference_optimizer/tests/test_trajectory_reviewer.py +++ b/src/hyperloom/inference_optimizer/tests/test_trajectory_reviewer.py @@ -9,6 +9,7 @@ _stalled_cycle_count, build_trajectory_digest, ) +from hyperloom.orchestrator.state.optimization_journal import OUTCOME_SKIP @dataclass @@ -56,6 +57,15 @@ def test_exhausted_clusters_ignores_high_gain(): assert dead == [] +def test_skips_are_not_harvested_as_a_dead_end(): + """Two benign conc_sweep skips per macro-cycle are the ordinary case.""" + entries = [ + _FakeEntry(outcome=OUTCOME_SKIP, kind="other", change="conc_sweep"), + _FakeEntry(outcome=OUTCOME_SKIP, kind="other", change="conc_sweep"), + ] + assert _exhausted_clusters(entries) == [] + + def test_exhausted_clusters_none_gain(): entries = [ _FakeEntry(outcome="REVERT", kind="kernel_agent", change="patch_a"), diff --git a/src/hyperloom/inference_optimizer/tools/backfill_langfuse.py b/src/hyperloom/inference_optimizer/tools/backfill_langfuse.py index 70df645aeb..e69c429167 100644 --- a/src/hyperloom/inference_optimizer/tools/backfill_langfuse.py +++ b/src/hyperloom/inference_optimizer/tools/backfill_langfuse.py @@ -27,7 +27,7 @@ fallback). Projected by ``langfuse_mapping.decision_to_scores``: - decision_outcome (CATEGORICAL: - KEEP/REVERT/no_promote) + KEEP/REVERT/no_promote/skipped) - gain_pct / predicted_gain_pct / proposal_score (NUMERIC) when present @@ -39,7 +39,7 @@ subset that recorded it; paired by :func:`langfuse_mapping.pair_key`. * ``reports/trace/decision_trace.jsonl`` -- per-action - KEEP/REVERT/no_promote + gain_pct. + KEEP/REVERT/no_promote/skipped + gain_pct. * ``runtime/recipe_snapshot/.audit.jsonl`` -- recipe-KB read/write audit rows (rendered as the ``agent:recipe_kb`` span subtree). * ``manifest.json`` -- trace-level metadata + @@ -81,6 +81,12 @@ from typing import Any from hyperloom.common.jsonio import read_json, read_jsonl +from hyperloom.orchestrator.state.optimization_journal import ( + OUTCOME_KEEP, + OUTCOME_NO_PROMOTE, + OUTCOME_REVERT, + OUTCOME_SKIP, +) from hyperloom.orchestrator.trace import langfuse_mapping as lfmap from hyperloom.orchestrator.trace.langfuse_emitter import ( _end_obs, @@ -235,13 +241,14 @@ def print_plan(plan: dict[str, Any]) -> None: with_text = sum(1 for g in gens if g["has_text"]) print(f" - {agent}: {len(gens)} gen(s), {with_text} with text, models={models}") outcomes = [(d.get("decision") or {}).get("outcome") for d in plan["decisions"]] - keep = outcomes.count("KEEP") - rev = outcomes.count("REVERT") - nop = outcomes.count("no_promote") + keep = outcomes.count(OUTCOME_KEEP) + rev = outcomes.count(OUTCOME_REVERT) + nop = outcomes.count(OUTCOME_NO_PROMOTE) + skipped = outcomes.count(OUTCOME_SKIP) gainful = sum(1 for d in plan["decisions"] if (d.get("decision") or {}).get("gain_pct") is not None) print( f" Scores: {len(plan['decisions'])} decisions " - f"(KEEP={keep} REVERT={rev} no_promote={nop}; gain_pct set={gainful})" + f"(KEEP={keep} REVERT={rev} no_promote={nop} skipped={skipped}; gain_pct set={gainful})" ) recipe_rows = plan.get("recipe_audit") or [] recipe_writes = sum(1 for r in recipe_rows if lfmap.recipe_audit_is_write(r)) diff --git a/src/hyperloom/orchestrator/actions/executors/baseline.py b/src/hyperloom/orchestrator/actions/executors/baseline.py index 1d9c8ad140..c41f9049b5 100644 --- a/src/hyperloom/orchestrator/actions/executors/baseline.py +++ b/src/hyperloom/orchestrator/actions/executors/baseline.py @@ -96,6 +96,13 @@ # Bounded per-file read so log scanning never slurps a multi-GB server.log. _LOG_SCAN_MAX_BYTES = 262_144 +# The cold-start guard's two round directories. The warmup round is the only one +# that measures accuracy (``RUN_EVAL=true``); the measured round is hot +# throughput alone, so it carries no accuracy by construction. +_WARMUP_ROUND_DIR = "warmup_round" +_MEASURE_ROUND_DIR = "measure_round" +_DOUBLE_RUN_ROUND_DIRS = (_WARMUP_ROUND_DIR, _MEASURE_ROUND_DIR) + # Markers identifying a MoE quant scheme with no implementation for the # ``--moe-runner-backend`` in use: ``create_moe_runner`` falls through without # building a runner and the first forward pass dies (e.g. Quark MXFP4 on @@ -391,6 +398,34 @@ def _should_establish_quality_ref(task_kind: str | None, params: dict[str, Any] return not (params or {}).get("quality_ref_exempt") +def _is_double_run_accuracy_handoff( + result: dict[str, Any], + salvaged: dict[str, Any] | None, +) -> bool: + """Whether accuracy came from the warmup round because that is the design. + + The cold-start guard splits one baseline into a warmup round that measures + accuracy and a measured round that measures hot throughput only, then + decides on the measured round -- which by construction has no accuracy of + its own. Reading the warmup round's score there is the intended handoff, + not a recovery, and logging it as one makes every healthy double-run + baseline look like it survived a fault. + + Args: + result (dict[str, Any]): The deciding round's result dict. + salvaged (dict[str, Any] | None): The salvage record, whose + ``source_file`` names the round the accuracy came from. + + Returns: + bool: ``True`` only for measured-round decision + warmup-round source. + """ + out_dir = str((result or {}).get("output_dir") or "") + if not out_dir or Path(out_dir).name != _MEASURE_ROUND_DIR: + return False + source = str((salvaged or {}).get("source_file") or "") + return _WARMUP_ROUND_DIR in Path(source).parts + + # Filesystem types that can be revoked / unmounted mid-run (e.g. a wekafs/NFS # mount flap), where a process whose cwd lives on such a mount sees relative-path # writes ENOENT. Such FS types trigger local mirroring of the InferenceX @@ -1872,7 +1907,7 @@ def _hit(text: str) -> bool: root = Path(out_dir) # Double-run: the failure markers may live in the sibling warmup round, # so climb to the shared task root to scan both rounds. - if root.name in ("warmup_round", "measure_round"): + if root.name in _DOUBLE_RUN_ROUND_DIRS: root = root.parent if not root.exists(): return False @@ -1933,7 +1968,7 @@ def _window(text: str) -> str | None: root = Path(out_dir) # Double-run: the failure markers may live in the sibling warmup round, # so climb to the shared task root to scan both rounds. - if root.name in ("warmup_round", "measure_round"): + if root.name in _DOUBLE_RUN_ROUND_DIRS: root = root.parent if not root.exists(): return False, "" @@ -2329,14 +2364,28 @@ def _maybe_stop_on_missing_baseline_accuracy( # should reach enablement. salvaged = self._salvage_sibling_baseline_accuracy(result, framework) if salvaged is not None: - acc_val = self._apply_salvaged_accuracy(result, salvaged, shared_state) - log.warning( - "baseline_executor: this attempt's RESULT_DIR had no accuracy, " - "but salvaged a measured baseline accuracy=%.4f from a sibling " - "attempt (%s)", - acc_val, - salvaged.get("source_file", ""), + expected_handoff = _is_double_run_accuracy_handoff(result, salvaged) + acc_val = self._apply_salvaged_accuracy( + result, + salvaged, + shared_state, + expected_handoff=expected_handoff, ) + if expected_handoff: + log.info( + "baseline_executor: cold-start guard — reading accuracy=%.4f from " + "the warmup round (%s), the only round that measures it", + acc_val, + salvaged.get("source_file", ""), + ) + else: + log.warning( + "baseline_executor: this attempt's RESULT_DIR had no accuracy, " + "but salvaged a measured baseline accuracy=%.4f from a sibling " + "attempt (%s)", + acc_val, + salvaged.get("source_file", ""), + ) # Floor 0.0 reproduces the non-enablement "any positive accuracy is # usable" rule; ``accuracy_meets_floor`` rejects zero either way. if accuracy_meets_floor(acc_val, floor if eval_enablement else 0.0): @@ -2383,6 +2432,8 @@ def _apply_salvaged_accuracy( result: dict[str, Any], salvaged: dict[str, Any], shared_state: Any, + *, + expected_handoff: bool = False, ) -> float: """Record a salvaged sibling accuracy, publishing it as the gate reference only when it can serve as one. @@ -2398,6 +2449,11 @@ def _apply_salvaged_accuracy( salvaged: The parsed eval dict from :meth:`_salvage_sibling_baseline_accuracy`. shared_state: The live SharedState, or ``None``. + expected_handoff: Whether this read is the double-run design + (see :func:`_is_double_run_accuracy_handoff`) rather than a + recovery. The structured warning is for the recovery only; + raising it on every healthy double-run baseline leaves the + record claiming a fault the run never hit. Returns: float: The salvaged accuracy. @@ -2409,8 +2465,9 @@ def _apply_salvaged_accuracy( result["accuracy_task"] = salvaged.get("task", "gsm8k") result["accuracy_metric"] = salvaged.get("metric", "") result["accuracy_source"] = salvaged.get("source_file", "") - result.setdefault("nonfatal_warnings", []) - result["nonfatal_warnings"].append("baseline_accuracy_salvaged_from_sibling_attempt") + if not expected_handoff: + result.setdefault("nonfatal_warnings", []) + result["nonfatal_warnings"].append("baseline_accuracy_salvaged_from_sibling_attempt") if shared_state is not None and accuracy_meets_floor(acc_val, 0.0): try: shared_state.baseline_accuracy = acc_val diff --git a/src/hyperloom/orchestrator/actions/executors/report.py b/src/hyperloom/orchestrator/actions/executors/report.py index 5591788a3c..6259fe1b3c 100644 --- a/src/hyperloom/orchestrator/actions/executors/report.py +++ b/src/hyperloom/orchestrator/actions/executors/report.py @@ -415,12 +415,53 @@ def _build_failure_summary( } -def _explain_stop_reason(stop_reason): +def _explain_stop_reason(stop_reason, state=None): """Return a human-readable explanation for a terminal ``stop_reason``. + ``conc_sweep_done`` is the SWEEP exit for a concurrency sweep that reached + a terminal result, which includes one that declined to run at all and one + that spent its budget without a comparable pair. The generic wording then + tells the reader a sweep finished when none happened, so a skip is named + when ``state`` is available to say so. + Returns ``""`` for unknown/empty reasons so callers can omit the line. """ - return _STOP_REASON_EXPLANATIONS.get(str(stop_reason or "").strip(), "") + reason = str(stop_reason or "").strip() + text = _STOP_REASON_EXPLANATIONS.get(reason, "") + if reason == "conc_sweep_done" and text: + return _explain_conc_sweep_skip(state) or text + return text + + +def _explain_conc_sweep_skip(state) -> str: + """Name a skipped concurrency sweep, or ``""`` when one ran to a result. + + A sweep that consumed its whole budget without reaching a comparable pair + is recorded as skipped too, and telling the reader it never ran is the + more expensive claim to believe in exactly the sessions where the budget + is the thing under investigation. + + Args: + state: The session's shared state, or ``None``. + + Returns: + str: The explanation line, or ``""`` when nothing was skipped. + """ + last = getattr(state, "last_conc_sweep", None) + if not isinstance(last, dict) or not last.get("was_skipped"): + return "" + # Imported here, not at module scope: ``kernel.conc_sweep`` imports the + # grid runner in this same package, so a top-level import is the edge + # CodeQL reports as a cycle. + from ...kernel.conc_sweep import conc_sweep_declined_to_run # noqa: PLC0415 + + detail = str(last.get("skip_reason") or "").strip() or "no reason recorded" + if conc_sweep_declined_to_run(last): + return f"Post-sweep concurrency sweep did not run ({detail}); the phase settled and the run closed." + return ( + f"Post-sweep concurrency sweep exhausted its budget without a comparable " + f"baseline/optimized pair ({detail}); the phase settled and the run closed." + ) def _platform_fingerprint(gpu_type: str | None = None) -> dict[str, Any]: @@ -488,7 +529,7 @@ def _build_summary_dict( "model_class": state.model_class, "framework": getattr(state, "framework", "") or "", "stop_reason": stop_reason, - "stop_reason_explanation": _explain_stop_reason(stop_reason), + "stop_reason_explanation": _explain_stop_reason(stop_reason, state), "baseline_tput": state.baseline_tput, "baseline_accuracy": state.baseline_accuracy, "current_best": state.current_best, diff --git a/src/hyperloom/orchestrator/kernel/conc_sweep.py b/src/hyperloom/orchestrator/kernel/conc_sweep.py index c5619d6df5..20cc110de3 100644 --- a/src/hyperloom/orchestrator/kernel/conc_sweep.py +++ b/src/hyperloom/orchestrator/kernel/conc_sweep.py @@ -17,7 +17,7 @@ import os import time from pathlib import Path -from typing import Any +from typing import Any, Mapping from hyperloom.common import io as _common_io from hyperloom.common.gain_math import conc_pair_comparison @@ -1135,6 +1135,28 @@ def _skip(reason: str, **extras: Any) -> dict[str, Any]: return payload +def conc_sweep_declined_to_run(record: Mapping[str, Any] | None) -> bool: + """Whether a conc-sweep record is one that never started a variant. + + ``was_skipped`` covers two different outcomes: the pre-flight envelope + from :func:`_skip`, which declines before a server boots, and a sweep that + ran its whole ladder but exhausted its budget before producing a + comparable pair (see :func:`_budget_limited_without_valid_pair`). Only the + second can set ``budget_exhausted``, which separates them without reading + ``skip_reason``. Any future skip raised after variants start must set it + too, or it will be misread as a sweep that declined. + + Args: + record (Mapping[str, Any] | None): A conc-sweep payload or the + ``last_conc_sweep`` record persisted from one. + + Returns: + bool: ``True`` when the sweep declined before running anything. + """ + rec = record or {} + return bool(rec.get("was_skipped")) and not rec.get("budget_exhausted") + + async def run_conc_sweep( state: SharedState, session_dir: Path, @@ -1418,5 +1440,6 @@ async def run_conc_sweep( "_flush_conc_sweep_report", "_flush_partial_conc_sweep_report", "_order_concs_desc", + "conc_sweep_declined_to_run", "run_conc_sweep", ] diff --git a/src/hyperloom/orchestrator/knowledge/trajectory_reviewer.py b/src/hyperloom/orchestrator/knowledge/trajectory_reviewer.py index 3d773b7544..1aa1f2c541 100644 --- a/src/hyperloom/orchestrator/knowledge/trajectory_reviewer.py +++ b/src/hyperloom/orchestrator/knowledge/trajectory_reviewer.py @@ -57,6 +57,10 @@ def _load_journal_entries(session_dir: Path, shared_state: Any) -> list[Any]: def _exhausted_clusters(entries: list[Any]) -> list[dict[str, Any]]: """Group repeated REVERT / no_promote attempts by (kind, change); dead ends first. + Only outcomes that measured something count: a step recorded as + ``OUTCOME_SKIP`` never ran, and clustering it would advise the model to + abandon a direction nothing was learned about. + Args: entries: The optimization journal entries to cluster. diff --git a/src/hyperloom/orchestrator/loop/writeback.py b/src/hyperloom/orchestrator/loop/writeback.py index db8eb7e3c9..f5290d38f2 100644 --- a/src/hyperloom/orchestrator/loop/writeback.py +++ b/src/hyperloom/orchestrator/loop/writeback.py @@ -1288,7 +1288,9 @@ def _record_fact_per_task( reason = None else: error_class = str(result_dict.get("error_class") or "") or None - reason = str(result_dict.get("reason") or "") or None + # A skip states its own cause under ``skip_reason``; without it the + # timeline shows a step that did nothing and never says why. + reason = str(result_dict.get("reason") or result_dict.get("skip_reason") or "") or None journal.append_entry( JournalEntry( phase=self._journal_entry_phase(), diff --git a/src/hyperloom/orchestrator/phases/machine_state.py b/src/hyperloom/orchestrator/phases/machine_state.py index ecc647bff2..1fba9944db 100644 --- a/src/hyperloom/orchestrator/phases/machine_state.py +++ b/src/hyperloom/orchestrator/phases/machine_state.py @@ -15,6 +15,7 @@ import math from typing import Any +from hyperloom.common.coerce import to_unix from hyperloom.inference_optimizer.protocol.action_surfaces import ( COORDINATOR_INTERNAL_ACTIONS, ROBUSTNESS_DELEGATE_ONLY_ACTIONS, @@ -847,6 +848,19 @@ def _phase_started_unix(state: Any) -> float: return 0.0 +def _resume_boundary_unix(state: Any) -> float: + """Return when the current run leg began, i.e. the most recent ``--resume``. + + Args: + state (Any): Frozen SharedState view exposing ``resumed_ts``. + + Returns: + float: Leg start in seconds since the epoch, or ``0.0`` for a session + that has only ever run once (or an unparseable stamp). + """ + return max(0.0, to_unix(getattr(state, "resumed_ts", ""), 0.0) or 0.0) + + def _kernel_idle_since_unix(state: Any) -> float: """Return when the current KERNEL idle streak opened, defensively coerced. @@ -935,6 +949,18 @@ def phase_elapsed_seconds(state: Any, *, now_unix: float | None = None) -> float Returns ``0.0`` when the phase start timestamp is unset (phase not yet entered) so callers can treat "not started" as zero elapsed. + Exiting the process is not a phase transition, so ``phase_started_unix`` + survives a ``--resume`` and the entry it stamps spans both run legs. The + current leg's boundary (:func:`_resume_boundary_unix`) therefore floors the + segment: the phase was not executing while nothing was, and a session + resumed days later would otherwise read as having overspent every phase + ceiling before it did any work. The floor only applies to the entry the + stop interrupted — a later entry stamps a newer ``phase_started_unix``. + + The previous leg's own share of that entry is not measured here; a resume + banks it into ``phase_elapsed_totals`` when the stopped leg recorded when + it ended. + Args: state (Any): Frozen SharedState view exposing ``phase_started_unix``. now_unix (float | None): Override for the current time; defaults to @@ -946,6 +972,7 @@ def phase_elapsed_seconds(state: Any, *, now_unix: float | None = None) -> float started = _phase_started_unix(state) if started <= 0: return 0.0 + started = max(started, _resume_boundary_unix(state)) now = float(now_unix if now_unix is not None else _now_unix(state)) return max(0.0, now - started) @@ -959,9 +986,14 @@ def phase_elapsed_totals_from_history(history: Any) -> dict[str, float]: is deliberately excluded: :func:`phase_cumulative_seconds` adds the live segment itself, and double-counting it would over-charge the phase. - ``phase_history`` is capped, so a very long session reconstructs a LOWER - bound. That is the safe direction for a budget guard — it can under-charge a - resumed phase, but it can never invent time the phase did not spend. + The rebuild is an estimate, not a bound in either direction. ``phase_history`` + is capped, so a very long session loses its oldest segments and is + under-charged. And no row marks the process exiting, so two rows either side + of a resume bound one "segment" spanning the idle gap between the legs, and + charge it to the phase the earlier row named — the over-charge direction + :func:`phase_elapsed_seconds` floors off the live segment. That floor reads + ``resumed_ts``, which dates the newest leg only; ``phase_history`` dates none + of them, so the rebuild cannot repeat it. Args: history (Any): The ``phase_history`` list; any other type yields ``{}``. @@ -1093,9 +1125,12 @@ def _phase_budget_total_seconds( session_remaining = session_remaining_seconds(state, now_unix=now_unix) if session_remaining is not None: - # Charge-back. remaining_at_entry reconstructs the time left when this - # phase started (session_remaining shrinks as phase_elapsed grows, so - # their sum is constant across the phase). + # Charge-back. remaining_at_entry reconstructs the time left when the + # phase's live segment opened: within one run leg session_remaining + # shrinks exactly as phase_elapsed grows, so their sum holds. A resume + # that keeps ``start_ts`` drops that sum by the idle gap between the + # legs — the session is charged for it, the phase is not — so a resumed + # phase charges back against a smaller, honest base. remaining_at_entry = max(0.0, session_remaining + phase_elapsed_seconds(state, now_unix=now_unix)) if is_long_run(state): # Long bounded run: the per-cycle window caps the base as a planning @@ -2182,7 +2217,17 @@ def exit_normal_sweep( if cs_status == "failed": return "conc_sweep_failed", {"conc_sweep_status": cs_status} if cs_status in ("succeeded", "partial", "completed", "skipped"): - return "conc_sweep_done", {"conc_sweep_status": cs_status} + evidence: dict[str, Any] = {"conc_sweep_status": cs_status} + # A sweep that declined to run is also terminal, and the exit + # reason alone cannot tell the two apart afterwards. was_skipped + # covers both declining and spending the whole budget without a + # comparable pair, so it is only carried with the flag that + # separates them (see kernel.conc_sweep.conc_sweep_declined_to_run). + if last_conc.get("was_skipped"): + evidence["conc_sweep_was_skipped"] = True + evidence["conc_sweep_budget_exhausted"] = bool(last_conc.get("budget_exhausted")) + evidence["conc_sweep_skip_reason"] = str(last_conc.get("skip_reason") or "") + return "conc_sweep_done", evidence remaining = phase_budget_remaining_seconds( state, budget_pct=budget_pct, @@ -2857,8 +2902,55 @@ def make_lifecycle_event( # Phase-transition / lifecycle write-owner functions (take ``state`` first and -# own the phase_history / lifecycle bookkeeping). ``SharedState`` exposes -# forwarding shims so existing callers reach these. +# own the phase_history / lifecycle bookkeeping). ``SharedState`` keeps +# forwarding shims for the two that were once its methods, so existing +# ``state.record_*(...)`` call sites still reach them; ``bank_phase_segment`` +# never was one and is called by name, from here and from the resume path. +def bank_phase_segment(state, *, until_unix: float) -> float: + """Bank the current phase's live segment, ending at ``until_unix``, into the durable totals. + + ``phase_started_unix`` holds the live segment and is overwritten by the next + phase entry, so a phase's spend only survives once it is banked here. Called + at every transition out of a phase, and by a resume for the segment the + stopped leg never transitioned out of. + + Args: + state: The live SharedState; ``phase_elapsed_totals`` (and the EXPLORE + accumulator) are mutated in place. + until_unix (float): When the segment ended, in seconds since the epoch. + + Returns: + float: Seconds banked. ``0.0`` when no phase is set, which is the very + first transition of a fresh session — it has no segment to bank. + """ + phase = (getattr(state, "phase", "") or "").strip().upper() + if not phase: + return 0.0 + segment = phase_elapsed_seconds(state, now_unix=until_unix) + totals = getattr(state, "phase_elapsed_totals", None) + totals = dict(totals) if isinstance(totals, dict) else {} + try: + banked = max(0.0, float(totals.get(phase, 0.0) or 0.0)) + except (TypeError, ValueError): + banked = 0.0 + totals[phase] = banked + segment + state.phase_elapsed_totals = totals + # EXPLORE keeps its own accumulator: it carries a tri-state "unknown" for + # legacy resumes that status telemetry reports as absent, whereas + # ``phase_elapsed_totals`` must never report "unknown" — a budget guard + # would read that as "no cap". The two answer different questions. + if phase == PHASE_EXPLORE: + raw_accumulated = getattr(state, "explore_elapsed_accum_s", 0.0) + if raw_accumulated is not None: + try: + accumulated = float(raw_accumulated or 0.0) + except (TypeError, ValueError): + state.explore_elapsed_accum_s = None + else: + state.explore_elapsed_accum_s = accumulated + segment + return segment + + def record_phase_transition( state, *, @@ -2889,35 +2981,9 @@ def record_phase_transition( now_ts = ts or _dt.now(_tz.utc).isoformat(timespec="seconds") now_unix = float(ts_unix if ts_unix is not None else _time.time()) from_phase = (state.phase or "").strip().upper() - if from_phase: - # Bank the finished segment for EVERY phase so the budget guards can - # charge a phase for the whole run instead of the current entry. Empty - # ``from_phase`` is the very first transition of a fresh session, which - # has no segment to bank. - totals = getattr(state, "phase_elapsed_totals", None) - totals = dict(totals) if isinstance(totals, dict) else {} - try: - banked = max(0.0, float(totals.get(from_phase, 0.0) or 0.0)) - except (TypeError, ValueError): - banked = 0.0 - totals[from_phase] = banked + phase_elapsed_seconds(state, now_unix=now_unix) - state.phase_elapsed_totals = totals - # EXPLORE keeps its own accumulator: it carries a tri-state "unknown" for - # legacy resumes that status telemetry reports as absent, whereas - # ``phase_elapsed_totals`` must never report "unknown" — a budget guard - # would read that as "no cap". The two answer different questions. - if from_phase == PHASE_EXPLORE: - raw_accumulated = getattr(state, "explore_elapsed_accum_s", 0.0) - if raw_accumulated is not None: - try: - accumulated = float(raw_accumulated or 0.0) - except (TypeError, ValueError): - state.explore_elapsed_accum_s = None - else: - state.explore_elapsed_accum_s = accumulated + phase_elapsed_seconds( - state, - now_unix=now_unix, - ) + # Bank the finished segment for EVERY phase so the budget guards can charge + # a phase for the whole run instead of the current entry. + bank_phase_segment(state, until_unix=now_unix) row = make_history_row( from_phase=from_phase, to_phase=to_phase, @@ -3084,6 +3150,7 @@ def record_lifecycle_event( "abort_prelude", "allowed_actions_for", "apply_escalate_budget_bump", + "bank_phase_segment", "compute_next_phase", "compute_plateau_explore", "compute_plateau_framework_agent", diff --git a/src/hyperloom/orchestrator/policy/gate.py b/src/hyperloom/orchestrator/policy/gate.py index bf7ac1f117..219ea3bd3e 100644 --- a/src/hyperloom/orchestrator/policy/gate.py +++ b/src/hyperloom/orchestrator/policy/gate.py @@ -532,6 +532,10 @@ def _source_file_candidates(value: str) -> tuple[str, ...]: { "current_best", "stop_reason", + # Paired with stop_reason and written by the same setter: locking one + # without the other lets an update_state move the session's end time + # away from the reason it was stamped for. + "stop_ts", "last_tick_exception", "cumulative_gain", "cumulative_gain_validated", @@ -546,6 +550,9 @@ def _source_file_candidates(value: str) -> tuple[str, ...]: "model_name", "model_class", "start_ts", + # Where the current run leg begins; a forged value hands a previous + # leg's CLOSE transition back the right to speak for this one. + "resumed_ts", "max_minutes", # fact-layer KEEP ledger; Coordinator is the sole writer. "optimization_stack", diff --git a/src/hyperloom/orchestrator/state/optimization_journal.py b/src/hyperloom/orchestrator/state/optimization_journal.py index a98a30042a..05c670bd7b 100644 --- a/src/hyperloom/orchestrator/state/optimization_journal.py +++ b/src/hyperloom/orchestrator/state/optimization_journal.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. # SPDX-License-Identifier: MIT -"""Per-session optimization journal — structured JSON record of every KEEP / REVERT / no_promote decision. +"""Per-session optimization journal — structured JSON record of every KEEP / REVERT / no_promote / skipped decision. Lives at ``/reports/optimization_journal.json``; rewritten incrementally (atomic tmp + ``os.replace``) so a mid-session crash leaves a usable artifact. @@ -31,12 +31,21 @@ OUTCOME_KEEP: str = "KEEP" OUTCOME_REVERT: str = "REVERT" OUTCOME_NO_PROMOTE: str = "no_promote" +# A step that declined to run. Distinct from ``no_promote``, which readers take +# as "we tried this and it did not pay" and harvest as a direction to stop +# exploring (see ``knowledge.trajectory_reviewer``); a step that never ran is +# evidence of nothing. +OUTCOME_SKIP: str = "skipped" # Task kinds whose result carries an authoritative per-status verdict the # journal outcome must follow rather than the coarse dispatcher ``promotable`` # flag (a ``reverted`` patch is promotable yet was rolled back). _STATUS_DRIVEN_JOURNAL_KINDS: frozenset[str] = frozenset({"integrate_patch", "framework_agent"}) +# Task kinds whose result can legitimately declare ``was_skipped``. Scoped so +# reusing that key elsewhere cannot silently demote a kept patch. +_SKIPPABLE_JOURNAL_KINDS: frozenset[str] = frozenset({"conc_sweep"}) + # The only status meaning the change was adopted into current_best. _JOURNAL_KEEP_STATUSES: frozenset[str] = frozenset({"kept"}) @@ -79,7 +88,7 @@ def _optional_int(value: Any) -> int | None: @dataclass class JournalEntry: - """One KEEP / REVERT / no_promote decision (``None`` distinguishes "not measured" from "measured zero").""" + """One KEEP / REVERT / no_promote / skipped decision (``None`` distinguishes "not measured" from "measured zero").""" phase: str iter: int @@ -365,6 +374,11 @@ def derive_journal_outcome( For every other task kind the binary behaviour applies (``promotable`` → KEEP, else REVERT). + A result from a skippable kind declaring ``was_skipped`` outranks both + rules. A skip succeeds by design -- a concurrency sweep with no + optimization to compare has nothing to do and says so -- but a success that + changed nothing is neither a KEEP nor a measured dead end. + Args: task_kind: The settled task's kind. result_dict: The task result dict (``status`` drives the patch kinds). @@ -372,9 +386,12 @@ def derive_journal_outcome( Returns: One of :data:`OUTCOME_KEEP` / :data:`OUTCOME_REVERT` / - :data:`OUTCOME_NO_PROMOTE`. + :data:`OUTCOME_NO_PROMOTE` / :data:`OUTCOME_SKIP`. """ - if (task_kind or "").lower() in _STATUS_DRIVEN_JOURNAL_KINDS: + kind = (task_kind or "").lower() + if kind in _SKIPPABLE_JOURNAL_KINDS and (result_dict or {}).get("was_skipped"): + return OUTCOME_SKIP + if kind in _STATUS_DRIVEN_JOURNAL_KINDS: status = str((result_dict or {}).get("status") or "").strip().lower() if status in _JOURNAL_KEEP_STATUSES: return OUTCOME_KEEP @@ -524,6 +541,7 @@ def summarize_change( "OUTCOME_KEEP", "OUTCOME_NO_PROMOTE", "OUTCOME_REVERT", + "OUTCOME_SKIP", "classify_change_kind", "derive_journal_outcome", "summarize_change", diff --git a/src/hyperloom/orchestrator/state/shared_state.py b/src/hyperloom/orchestrator/state/shared_state.py index f6ebc53854..d7168c4327 100644 --- a/src/hyperloom/orchestrator/state/shared_state.py +++ b/src/hyperloom/orchestrator/state/shared_state.py @@ -27,6 +27,8 @@ extra_envs, workspace, latency means) cumulative_gain float — % over baseline stop_reason str — set when graceful stop fires + stop_ts str — ISO timestamp of the first stop_reason write + resumed_ts str — ISO timestamp of the most recent --resume current_action str — what's running right now (set by Orchestration) crash_count int — incremented by the Coordinator when a tick/agent exception is recorded; also appends to @@ -610,6 +612,17 @@ class SharedState(_RenderMixin, _ExploreStateMixin): # Tput watermark for gain-driven roofline refresh; Coordinator re-enqueues at a compound 10% step. last_roofline_tput: float = 0.0 stop_reason: str = "" + # When the session first stopped, and therefore its end time for + # consumers. Stamped by the first ``set_stop_reason`` write and left alone + # by later ones, so the CLOSE sequence's own artifacts and any re-export + # quote the same end; cleared with the reason on resume. + stop_ts: str = "" + # When the current run leg began, i.e. the most recent ``--resume``; empty + # for a session that has only ever run once. ``start_ts`` cannot answer + # this: a resume after a clean stop deliberately keeps it so the wall-clock + # budget still counts from the original start, which leaves this the only + # record of where the previous leg ended. + resumed_ts: str = "" # Closing phase — set when wall-clock deadline fires; Coordinator only drains a ``report`` task. Cleared on resume. closing_phase: bool = False closing_started_unix: float = 0.0 @@ -918,8 +931,8 @@ class SharedState(_RenderMixin, _ExploreStateMixin): # re-entered once per macro-cycle. That turned "KERNEL gets 15% of the run" # into "KERNEL gets 15% of the run every time it is entered": three entries # burned 288% of the cap while the other phases starved. The guards read - # this total plus the live segment instead. ``record_phase_transition`` is - # the only writer; unlike ``explore_elapsed_accum_s`` there is no "unknown" + # this total plus the live segment instead. ``bank_phase_segment`` is the + # only writer; unlike ``explore_elapsed_accum_s`` there is no "unknown" # sentinel, because a budget guard must never read "unknown" as "no cap" # (see ``from_raw`` for how a pre-upgrade state is reconstructed). phase_elapsed_totals: dict[str, float] = field(default_factory=dict) @@ -1313,8 +1326,8 @@ def from_dict(cls, raw: dict[str, Any]) -> "SharedState": # transition in phase_history, so the completed segments are # reconstructible. Rebuilding beats defaulting to an empty dict: an empty # dict silently re-arms the per-entry bug for the rest of a resumed run. - # phase_history is capped, so the rebuild is a LOWER bound — the safe - # direction, since it can only under-charge a phase, never invent time. + # The rebuild errs in both directions — the cap drops old segments, and a + # segment straddling a resume carries the idle gap (see the helper). if not isinstance(filtered.get("phase_elapsed_totals"), dict): from ..phases.machine_state import phase_elapsed_totals_from_history @@ -1734,9 +1747,13 @@ def set_stop_reason( ) -> str: """Validated writer for :attr:`stop_reason` (Inv-8.3 closed vocab): values outside ``STOP_REASON_VOCAB`` map to ``"unknown"`` (lenient) or raise (``strict=True``, default env ``INFERENCE_OPTIMIZER_STRICT_STOP_REASON``). Returns value written. + The first write also stamps :attr:`stop_ts`, so the session's end is + recorded once by its producer instead of being guessed by whoever reads + the state later. + Args: value (str): The proposed stop reason; blank clears - :attr:`stop_reason`. + :attr:`stop_reason` and :attr:`stop_ts`. strict (bool | None): When ``True`` an out-of-vocab value raises; when ``None`` the mode is read from ``INFERENCE_OPTIMIZER_STRICT_STOP_REASON``. @@ -1754,10 +1771,10 @@ def set_stop_reason( text = str(value or "").strip() if not text: self.stop_reason = "" + self.stop_ts = "" return "" if is_valid_stop_reason(text): - self.stop_reason = text - return text + return self._commit_stop_reason(text) if strict is None: strict_env = ( os.environ.get( @@ -1779,8 +1796,28 @@ def set_stop_reason( "INFERENCE_OPTIMIZER_STRICT_STOP_REASON=1 to fail-fast.", text, ) - self.stop_reason = "unknown" - return "unknown" + return self._commit_stop_reason("unknown") + + def _commit_stop_reason(self, reason: str) -> str: + """Write a validated stop reason, stamping the end time on the first one. + + The session ends when its first terminal reason is recorded. Later + calls may refine the reason -- CLOSE stops the session on entry and the + Coordinator's ``finally`` re-asserts it -- but the CLOSE sequence writes + ``session_breakdown.json`` in between, so moving the timestamp would + leave the shipped artifact and the state disagreeing about when the run + ended. + + Args: + reason (str): The validated, non-blank reason to record. + + Returns: + str: ``reason``, so callers can return it unchanged. + """ + self.stop_reason = reason + if not self.stop_ts: + self.stop_ts = _now_iso() + return reason # escalate hint plumbing def set_pending_escalate_hint(self, hint: str) -> str: diff --git a/src/hyperloom/orchestrator/trace/langfuse_mapping.py b/src/hyperloom/orchestrator/trace/langfuse_mapping.py index bd072741d3..4db3205dc3 100644 --- a/src/hyperloom/orchestrator/trace/langfuse_mapping.py +++ b/src/hyperloom/orchestrator/trace/langfuse_mapping.py @@ -416,7 +416,7 @@ def decision_to_scores(decision_row: dict[str, Any]) -> list[dict[str, Any]]: """Project one ``decision_trace.jsonl`` row onto one or more Score dicts. Always emits a CATEGORICAL ``decision_outcome`` score (KEEP / REVERT / - no_promote); additionally NUMERIC ``gain_pct`` (measured gain), + no_promote / skipped); additionally NUMERIC ``gain_pct`` (measured gain), ``predicted_gain_pct`` (the proposer's estimate) and ``proposal_score`` (mean pre-decision rater score) when the decision carries them. Each returned dict is transport-agnostic