diff --git a/scripts/code_metrics_baseline.txt b/scripts/code_metrics_baseline.txt index c0eb7f530f..b3030fa464 100644 --- a/scripts/code_metrics_baseline.txt +++ b/scripts/code_metrics_baseline.txt @@ -30,12 +30,6 @@ cognitive-complexity src/hyperloom/orchestrator/kernel/conc_sweep.py::_sweep_one cognitive-complexity src/hyperloom/orchestrator/kernel/conc_sweep.py::run_conc_sweep 60 cognitive-complexity src/hyperloom/orchestrator/kernel/forge_fusion.py::_normalize_manifest 44 cognitive-complexity src/hyperloom/orchestrator/kernel/request_handlers.py::_resolve_forge_server_log 49 -cognitive-complexity src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator._build_recipe_attrs_from_state 41 -cognitive-complexity src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator._collect_attempt_provenance 34 -cognitive-complexity src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator._collect_workload_tags 37 -cognitive-complexity src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator.finalize_recipe_and_journal 48 -cognitive-complexity src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator.kb_amend_recipe 49 -cognitive-complexity src/hyperloom/orchestrator/knowledge/recipe_journal.py::_record_config_attempts 36 cognitive-complexity src/hyperloom/orchestrator/loop/writeback.py::WritebackCollaborator._drain_agent_keep_outbox 34 cognitive-complexity src/hyperloom/orchestrator/loop/writeback.py::WritebackCollaborator._keep_patch_sources 42 cognitive-complexity src/hyperloom/orchestrator/loop/writeback.py::WritebackCollaborator._promote_baseline 96 @@ -348,11 +342,6 @@ function-lines src/hyperloom/orchestrator/kernel/request_handlers.py::_run_forge function-lines src/hyperloom/orchestrator/kernel/request_handlers.py::_run_forge_gemm_tuning 117 function-lines src/hyperloom/orchestrator/kernel/request_handlers.py::_run_integrate_rebaseline_with_lock_retry 106 function-lines src/hyperloom/orchestrator/kernel/request_handlers.py::integrate_handler 157 -function-lines src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator._build_recipe_attrs_from_state 99 -function-lines src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator._record_fact_per_variant 114 -function-lines src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator.finalize_recipe_and_journal 211 -function-lines src/hyperloom/orchestrator/knowledge/recipe_journal.py::RecipeJournalCollaborator.kb_amend_recipe 127 -function-lines src/hyperloom/orchestrator/knowledge/recipe_journal.py::_record_config_attempts 136 function-lines src/hyperloom/orchestrator/knowledge/recipe_kb/local_store.py::LocalRecipeStore.put_recipe 102 function-lines src/hyperloom/orchestrator/knowledge/remote_recipe/client.py::RemoteRecipeClient.read 95 function-lines src/hyperloom/orchestrator/knowledge/remote_recipe/client.py::RemoteRecipeClient.write_if_better 82 @@ -491,7 +480,7 @@ module-lines src/hyperloom/orchestrator/actions/executors/roofline.py 1303 module-lines src/hyperloom/orchestrator/kernel/apply_kernel_patch.py 2656 module-lines src/hyperloom/orchestrator/kernel/conc_sweep.py 1429 module-lines src/hyperloom/orchestrator/kernel/request_handlers.py 4690 -module-lines src/hyperloom/orchestrator/knowledge/recipe_journal.py 1671 +module-lines src/hyperloom/orchestrator/knowledge/recipe_journal.py 1269 module-lines src/hyperloom/orchestrator/loop/coordinator.py 1337 module-lines src/hyperloom/orchestrator/loop/dispatcher.py 1453 module-lines src/hyperloom/orchestrator/loop/intent_router.py 1409 diff --git a/src/hyperloom/common/coerce.py b/src/hyperloom/common/coerce.py index ec270b3d5b..34c4446276 100644 --- a/src/hyperloom/common/coerce.py +++ b/src/hyperloom/common/coerce.py @@ -24,6 +24,18 @@ def to_float(value: Any, default: _T | None = None) -> float | _T | None: return result if math.isfinite(result) else default +def recorded_float(value: Any, default: float = 0.0) -> float: + """Coerce a recorded value to ``float``, keeping ``inf``, ``nan`` and ``bool`` as they were recorded. + + :func:`to_float` returns *default* for those three instead, which would + render a degenerate figure as one nobody can tell from a real measurement. + """ + try: + return float(value or default) + except (TypeError, ValueError): + return default + + def to_int(value: Any, default: _T | None = None) -> int | _T | None: """Coerce *value* to ``int``, rejecting ``bool``, ``None``, and non-finite floats.""" if value is None or isinstance(value, bool): @@ -116,6 +128,7 @@ def to_list(value: Any) -> list[Any]: __all__ = [ "to_float", + "recorded_float", "to_int", "first_float", "first_int", diff --git a/src/hyperloom/inference_optimizer/tests/test_coerce.py b/src/hyperloom/inference_optimizer/tests/test_coerce.py index d4658f5054..ac0c338c40 100644 --- a/src/hyperloom/inference_optimizer/tests/test_coerce.py +++ b/src/hyperloom/inference_optimizer/tests/test_coerce.py @@ -13,6 +13,7 @@ first_int, first_str, optional_positive_int, + recorded_float, to_float, to_int, to_list, @@ -69,6 +70,36 @@ def test_custom_default(self): assert to_float(None, default=-1.0) == -1.0 +class TestRecordedFloat: + @pytest.mark.parametrize( + "value,expected", + [ + (1, 1.0), + (1.5, 1.5), + ("2.5", 2.5), + (-4, -4.0), + ], + ) + def test_parses(self, value, expected): + assert recorded_float(value) == expected + + @pytest.mark.parametrize("value", [None, "", "abc", object(), [1]]) + def test_absent_or_unparseable_reads_as_the_default(self, value): + assert recorded_float(value) == 0.0 + assert recorded_float(value, -1.0) == -1.0 + + def test_infinity_survives_so_a_degenerate_measurement_stays_visible(self): + assert recorded_float(float("inf")) == math.inf + assert recorded_float("Infinity") == math.inf + assert math.isnan(recorded_float(float("nan"))) + + @pytest.mark.parametrize("value", [float("inf"), float("nan"), True]) + def test_keeps_what_to_float_rejects(self, value): + """The three inputs on which the two coercions disagree, so a swap is never silent.""" + assert to_float(value, 0.0) == 0.0 + assert recorded_float(value) != 0.0 or math.isnan(recorded_float(value)) + + class TestToInt: @pytest.mark.parametrize( "value,expected", diff --git a/src/hyperloom/inference_optimizer/tests/test_coordinator_kb_writes.py b/src/hyperloom/inference_optimizer/tests/test_coordinator_kb_writes.py index 47936e3097..7401a3d881 100644 --- a/src/hyperloom/inference_optimizer/tests/test_coordinator_kb_writes.py +++ b/src/hyperloom/inference_optimizer/tests/test_coordinator_kb_writes.py @@ -25,6 +25,9 @@ recipe_canonical_id, ) from hyperloom.orchestrator.knowledge.config import KnowledgeConfig +from hyperloom.orchestrator.knowledge.recipe_journal import ( + recipe_attrs_from_state, +) from hyperloom.inference_optimizer.breakdown.recorder import close_out as _close_out @@ -433,7 +436,7 @@ def test_best_config_reads_stack_args_from_canonical_server_key( } ] ss.cumulative_gain_validated = 10.0 - attrs = coord.recipe_journal._build_recipe_attrs_from_state() + attrs = recipe_attrs_from_state(coord.shared_state, coord.session_dir.name) assert attrs["best_config"]["extra_server_args"] == ("--page-size 32 --schedule-policy lpm"), ( "stack-layer launch args must be read from the canonical " "extra_server_args key; got " @@ -557,7 +560,7 @@ def test_session_entry_carries_throughput_date_and_actions( ] ss.cumulative_gain_validated = 7.5 ss.cumulative_gain_validated_stack_len = 2 - attrs = coord.recipe_journal._build_recipe_attrs_from_state() + attrs = recipe_attrs_from_state(coord.shared_state, coord.session_dir.name) sessions = attrs.get("sessions") or [] assert sessions, "no session entry emitted" s = sessions[0] diff --git a/src/hyperloom/inference_optimizer/tests/test_coordinator_sync_helpers_coverage_unit.py b/src/hyperloom/inference_optimizer/tests/test_coordinator_sync_helpers_coverage_unit.py index ae6df93e10..79eb807ee7 100644 --- a/src/hyperloom/inference_optimizer/tests/test_coordinator_sync_helpers_coverage_unit.py +++ b/src/hyperloom/inference_optimizer/tests/test_coordinator_sync_helpers_coverage_unit.py @@ -15,6 +15,9 @@ from hyperloom.orchestrator.loop.coordinator import Coordinator from hyperloom.orchestrator.state.objective import TargetGainObjective from hyperloom.inference_optimizer.protocol.intent import Intent, IntentType +from hyperloom.orchestrator.knowledge.recipe_journal import ( + workload_kb_tags, +) from hyperloom.orchestrator.loop.dispatcher import HeldLeases @@ -560,7 +563,7 @@ def test_collect_workload_tags(coord: Coordinator, monkeypatch) -> None: "workload_mode": "serving", } - assert coord.recipe_journal._collect_workload_tags() == { + assert workload_kb_tags(coord.shared_state) == { "framework": "sglang", "model_class": "moe", "model_name": "Qwen3-32B", @@ -591,19 +594,19 @@ def test_workload_tags_read_ep_from_the_env_only_without_a_shared_state_value(co ss = coord.shared_state ss.ep = 8 - tags = coord.recipe_journal._collect_workload_tags() + tags = workload_kb_tags(coord.shared_state) assert tags["ep"] == 8 assert tags["pp"] == 3 ss.ep = 0 - assert coord.recipe_journal._collect_workload_tags()["ep"] == 2 + assert workload_kb_tags(coord.shared_state)["ep"] == 2 def test_an_unparseable_or_non_positive_parallelism_env_is_not_tagged(coord: Coordinator, monkeypatch) -> None: monkeypatch.setenv("EP", "two") monkeypatch.setenv("PP", "0") - tags = coord.recipe_journal._collect_workload_tags() + tags = workload_kb_tags(coord.shared_state) assert "ep" not in tags assert "pp" not in tags @@ -618,7 +621,7 @@ def test_workload_tags_survive_a_resumed_null_fingerprint_meta(coord: Coordinato ss.stack_fingerprint_meta = None ss.baseline_workload_extra = None - assert coord.recipe_journal._collect_workload_tags() == {"framework": "vllm"} + assert workload_kb_tags(coord.shared_state) == {"framework": "vllm"} def test_unknown_and_empty_fingerprint_entries_are_not_tagged(coord: Coordinator, monkeypatch) -> None: @@ -634,7 +637,7 @@ def test_unknown_and_empty_fingerprint_entries_are_not_tagged(coord: Coordinator "image_digest": "sha256:abc", } - tags = coord.recipe_journal._collect_workload_tags() + tags = workload_kb_tags(coord.shared_state) assert tags == {"framework": "vllm", "image_digest": "sha256:abc"} diff --git a/src/hyperloom/inference_optimizer/tests/test_recipe_sedimentation.py b/src/hyperloom/inference_optimizer/tests/test_recipe_sedimentation.py index 82aa52bc4b..b27df6da9d 100644 --- a/src/hyperloom/inference_optimizer/tests/test_recipe_sedimentation.py +++ b/src/hyperloom/inference_optimizer/tests/test_recipe_sedimentation.py @@ -17,6 +17,10 @@ ) from hyperloom.orchestrator.loop.coordinator import AdvisoryServices, Coordinator from hyperloom.orchestrator.knowledge.recipe_kb import LocalRecipeStore, RecipeKB +from hyperloom.orchestrator.knowledge.recipe_journal import ( + attempt_provenance, + recipe_attrs_from_state, +) def _make_coordinator(tmp_path: Path) -> Coordinator: @@ -67,7 +71,7 @@ def test_collect_attempt_provenance_maps_keep_and_revert(tmp_path): }, ) - kept, kept_by_gap, reverted = coord.recipe_journal._collect_attempt_provenance() + kept, kept_by_gap, reverted = attempt_provenance(coord.shared_state) assert kept == {"mtp_on": "https://pr/123"} assert kept_by_gap == {"gap.research_hint.0": "https://pr/123"} assert len(reverted) == 1 @@ -106,7 +110,7 @@ def test_provenance_resolves_by_gap_id_when_name_mismatches(tmp_path): }, ) - attrs = coord.recipe_journal._build_recipe_attrs_from_state() + attrs = recipe_attrs_from_state(coord.shared_state, coord.session_dir.name) row = next(x for x in attrs["what_worked"] if x["name"] == "k007") assert row["source"] == "https://pr/777" @@ -142,7 +146,7 @@ def test_build_recipe_attrs_sediments_source_and_revert(tmp_path): }, ) - attrs = coord.recipe_journal._build_recipe_attrs_from_state() + attrs = recipe_attrs_from_state(coord.shared_state, coord.session_dir.name) ww = attrs["what_worked"] mtp = next(x for x in ww if x["name"] == "mtp_on") assert mtp["source"] == "https://pr/123" @@ -174,7 +178,7 @@ def test_sediment_toggle_off_keeps_recipe_ephemeral(tmp_path): }, ) - attrs = coord.recipe_journal._build_recipe_attrs_from_state() + attrs = recipe_attrs_from_state(coord.shared_state, coord.session_dir.name) mtp = next(x for x in attrs["what_worked"] if x["name"] == "mtp_on") assert "source" not in mtp @@ -219,7 +223,7 @@ def test_experience_rows_survive_the_kb_round_trip(tmp_path): model="m", hardware="h", framework_name="f", - # Exactly what recipe_journal._build_recipe_attrs_from_state emits. + # Exactly what recipe_journal.recipe_attrs_from_state emits. what_worked=[ { "name": "mtp_on", diff --git a/src/hyperloom/inference_optimizer/tests/test_remote_recipe_v2.py b/src/hyperloom/inference_optimizer/tests/test_remote_recipe_v2.py index 6a1925ddb9..2bd70405c3 100644 --- a/src/hyperloom/inference_optimizer/tests/test_remote_recipe_v2.py +++ b/src/hyperloom/inference_optimizer/tests/test_remote_recipe_v2.py @@ -3126,7 +3126,7 @@ def test_remote_read_is_wired_into_cli_t0_and_close_write_remains() -> None: fromlist=["_bootstrap_recipe_kb"], )._bootstrap_recipe_kb ) - close_source = inspect.getsource(RecipeJournalCollaborator.finalize_recipe_and_journal) + close_source = inspect.getsource(RecipeJournalCollaborator._publish_remote_recipe) assert "RemoteWarmRecipeAdapter" in bootstrap_source assert "HyperloomRemoteKB.from_env().write" in close_source assert "write_final_remote_recipe" not in close_source diff --git a/src/hyperloom/orchestrator/knowledge/_config_timeline.py b/src/hyperloom/orchestrator/knowledge/_config_timeline.py new file mode 100644 index 0000000000..6fb6f6564c --- /dev/null +++ b/src/hyperloom/orchestrator/knowledge/_config_timeline.py @@ -0,0 +1,169 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""The configuration arm's attempt rows on the FRAMEWORK_AGENT event.""" + +from __future__ import annotations + +from typing import Any, Collection, Mapping, Sequence + +from hyperloom.inference_optimizer.breakdown.recorder.framework_event import ( + ARM_CONFIG, + DISPOSITION_ATTEMPTED, + STEP_ATTEMPTED, + GateVerdict, +) + +#: Explore outcomes that are not an attempt at anything: a variant skipped as a +#: duplicate was never measured, so no measurement would back its funnel row. +_NON_ATTEMPT_OUTCOMES: frozenset[str] = frozenset({"SKIPPED_DEDUP"}) + + +def _attempt_failure(row: Mapping[str, Any]) -> dict[str, Any]: + return { + "error_class": str(row.get("error_class") or ""), + "error_excerpt": str(row.get("error_excerpt") or ""), + # Explore classified this variant's failure once; the verdict it stored is the one verdict. + "attribution": str(row.get("failure_attribution") or ""), + } + + +def _attempt_fields( + row: Mapping[str, Any], + gates: Sequence[Mapping[str, Any]], + shared: Mapping[str, str], + adopted: bool, +) -> dict[str, Any]: + variant = row.get("variant") if isinstance(row.get("variant"), dict) else {} + metrics = row.get("metrics") if isinstance(row.get("metrics"), dict) else {} + outcome = str(row.get("outcome") or "") + # This arm gates accuracy rather than reporting it, so the block both arms + # carry is projected from the gate that ruled. No gate row means nothing + # gated the variant, which is not a gate that refused it. + ruling = next((gate for gate in gates if str(gate.get("gate") or "") == "accuracy"), {}) + return { + "arm": ARM_CONFIG, + **shared, + "provenance": row.get("provenance"), + "outcome": outcome, + "reason": row.get("reason"), + "reasoning": variant.get("note"), + "reasoning_origin": variant.get("reasoning_origin"), + "experience_citations": variant.get("experience_citations"), + "stage": row.get("stage"), + "fingerprint": row.get("fingerprint"), + # The fingerprint is the join key; the name is what a reader + # recognises the variant by. + "variant_name": row.get("variant_name"), + "measurement": { + # Both ends of the pair: the anchor advances on every KEEP, so a + # percentage without its denominator adds to nothing. + "before_tput": metrics.get("base_tput"), + "after_tput": metrics.get("tput"), + "gain_pct": metrics.get("gain_pct"), + "runtime_sec": metrics.get("runtime_sec"), + "estimated_output_throughput": metrics.get("estimated_output_throughput"), + }, + "config_delta": { + "extra_server_args": variant.get("extra_server_args"), + "extra_envs": variant.get("extra_envs"), + "remove_args": variant.get("remove_args"), + "unset_envs": variant.get("unset_envs"), + "args_mode": variant.get("args_mode"), + }, + "accuracy": { + "required": True if ruling else None, + "value": ruling.get("observed"), + "reference": ruling.get("threshold"), + "passed": ruling.get("passed"), + }, + "failure": _attempt_failure(row), + "artifacts": { + "workspace": row.get("workspace"), + "server_log_path": row.get("server_log_path"), + "raw_result_path": row.get("raw_result_path"), + }, + "decision": outcome, + "adopted": adopted, + # Every variant recorded here ran: the deduped one is skipped above. + "benched": True, + # Recorded rather than referenced: every KEEP advances the stack, so + # the session's current config is not what this variant was measured + # on top of. + "measured_against": row.get("measured_against"), + # What stood behind the verdict. Absent when nothing ruled. + "validation_basis": row.get("validation_basis"), + # A pair is what makes a gain addable, so eligibility follows the pair + # being present rather than the outcome being a KEEP. + "attribution_eligible": (adopted and metrics.get("base_tput") is not None and metrics.get("tput") is not None), + } + + +def _record_variant_attempt( + recorder: Any, + row: Any, + shared: Mapping[str, str], + adopted_variants: Collection[str], +) -> bool: + if not isinstance(row, dict): + return False + if str(row.get("outcome") or "") in _NON_ATTEMPT_OUTCOMES: + return False + # The fingerprint identifies the variant within the round and the round + # within the task, so the three together identify the attempt. + fingerprint = str(row.get("fingerprint") or "") + parts = (shared["task_id"], shared["round_id"], fingerprint) + attempt_id = ":".join(part for part in parts if part) or shared["task_id"] + if not attempt_id: + return False + gates = [gate for gate in (row.get("gates") or []) if isinstance(gate, dict)] + recorder.record_attempt(attempt_id, **_attempt_fields(row, gates, shared, fingerprint in adopted_variants)) + for gate in gates: + recorder.record_attempt_gate( + attempt_id, + GateVerdict( + gate=str(gate.get("gate") or ""), + passed=gate.get("passed"), + reason=str(gate.get("reason") or ""), + observed=gate.get("observed"), + threshold=gate.get("threshold"), + ), + ) + return True + + +def _record_config_attempts( + coord: Any, + *, + task: Any, + per_variant: list[dict[str, Any]], + result_dict: Mapping[str, Any], + adopted_variants: Collection[str], +) -> None: + """Record the configuration arm's measured attempts on the framework event. + + Timeline only: the ledger row is written by ``fact_write_hook``, which does + not depend on a recorder being open. The outcome is recorded verbatim here, + unlike the journal beside it, which collapses ``KEEP_UNSTABLE`` and + ``KILLED_OVERTIME`` into a plain revert. + """ + recorder = coord._coord.phase_framework.timeline() + if recorder is None: + return + params = getattr(task, "params", None) or {} + shared = { + "task_id": str(getattr(task, "task_id", "") or ""), + "round_id": str(result_dict.get("round_id") or ""), + "proposal_ref": str(params.get("proposal_msg_id") or ""), + } + recorded = 0 + for row in per_variant: + if _record_variant_attempt(recorder, row, shared, adopted_variants): + recorded += 1 + if not (recorded and shared["proposal_ref"]): + return + # Settled here rather than at materialization: a grid whose task dies before + # any variant runs has no attempt to stand behind an ``attempted`` reading, + # and is more honestly left unsettled. + recorder.record_proposal_step(shared["proposal_ref"], step=STEP_ATTEMPTED, outcome=str(recorded)) + recorder.settle_proposal(shared["proposal_ref"], disposition=DISPOSITION_ATTEMPTED) diff --git a/src/hyperloom/orchestrator/knowledge/_recipe_attrs.py b/src/hyperloom/orchestrator/knowledge/_recipe_attrs.py new file mode 100644 index 0000000000..18171e3e84 --- /dev/null +++ b/src/hyperloom/orchestrator/knowledge/_recipe_attrs.py @@ -0,0 +1,242 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""What a session's recipe attributes say about where its optimizations came from.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Collection, Mapping, NamedTuple, Sequence + +from hyperloom.inference_optimizer.session.optimization_journal import OUTCOME_KEEP, OUTCOME_REVERT + +from .workload_tags import workload_shape_tags + +#: The probe's own word for a version it could not read. +_UNIDENTIFIED_VERSION: str = "unknown" + + +class AttemptProvenance(NamedTuple): + """Where a session's proven optimizations came from, and what it reverted. + + Keepers are indexed twice because a stack layer may identify itself either + by name or by the research gap it closed. + """ + + kept_by_name: dict[str, str] + kept_by_gap: dict[str, str] + reverted: list[dict[str, Any]] + + +def _known_version(raw: Any) -> str: + version = str(raw or "").strip() + return "" if version == _UNIDENTIFIED_VERSION else version + + +def _runtime_version_tags(fingerprint_meta: Mapping[str, Any], framework: str) -> dict[str, str]: + tags: dict[str, str] = {} + framework_key = framework.lower() + if framework_key in {"sglang", "vllm"}: + active = _known_version(fingerprint_meta.get(framework_key)) + if active: + tags["framework_version"] = active + for source_key, tag in (("rocm", "rocm_version"), ("aiter", "aiter_version"), ("image_digest", "image_digest")): + version = _known_version(fingerprint_meta.get(source_key)) + if version: + tags[tag] = version + return tags + + +def _baseline_workload_tags(workload_extra: Mapping[str, Any]) -> dict[str, Any]: + tags: dict[str, Any] = {} + for key in ("max_running_requests", "max_num_seqs"): + value = workload_extra.get(key) + if isinstance(value, int) and value > 0: + tags[key] = value + for key in ("chunked_prefill_enabled", "enable_torch_compile"): + value = workload_extra.get(key) + if isinstance(value, bool): + tags[key] = value + for key in ("quant_scheme", "workload_mode"): + value = workload_extra.get(key) + if isinstance(value, str) and value.strip(): + tags[key] = value.strip() + return tags + + +def workload_kb_tags(state: Any) -> dict[str, Any]: + """The workload-shape tags describing the session this recipe came from.""" + tags: dict[str, Any] = {} + # ``model_family`` is deliberately absent, being no part of the + # seven-dimension Recipe identity. + for attr in ("framework", "model_class", "model_name"): + text = str(getattr(state, attr, "") or "").strip() + if text: + tags[attr] = text + precision = getattr(state, "precision", None) + if precision not in (None, "", 0): + tags["precision"] = precision + tags.update(workload_shape_tags(state)) + fingerprint_meta = state.stack_fingerprint_meta or {} + if isinstance(fingerprint_meta, dict): + tags.update(_runtime_version_tags(fingerprint_meta, str(tags.get("framework") or ""))) + workload_extra = state.baseline_workload_extra or {} + if isinstance(workload_extra, dict): + tags.update(_baseline_workload_tags(workload_extra)) + return tags + + +def _reverted_attempt_row(name: str, attempt: Mapping[str, Any], source: str) -> dict[str, Any]: + row: dict[str, Any] = { + "name": name, + "reason": "reverted", + "gain_pct": attempt.get("gain_pct"), + } + if source: + row["source"] = source + return row + + +def _gap_attempt_provenance(gap: Mapping[str, Any]) -> AttemptProvenance: + source = str(gap.get("provenance") or "").strip() + canonical_id = str(gap.get("canonical_id") or "").strip() + kept_by_name: dict[str, str] = {} + kept_by_gap: dict[str, str] = {} + reverted: list[dict[str, Any]] = [] + for attempt in gap.get("attempts") or []: + if not isinstance(attempt, dict): + continue + variant = str(attempt.get("variant_name") or "").strip() + kernel = str(attempt.get("kernel_id") or "").strip() + outcome = str(attempt.get("outcome") or "").strip().upper() + if outcome == OUTCOME_KEEP and source: + for name in (variant, kernel): + if name: + kept_by_name.setdefault(name, source) + if canonical_id: + kept_by_gap.setdefault(canonical_id, source) + elif outcome == OUTCOME_REVERT and (variant or kernel): + reverted.append(_reverted_attempt_row(variant or kernel, attempt, source)) + return AttemptProvenance(kept_by_name=kept_by_name, kept_by_gap=kept_by_gap, reverted=reverted) + + +def attempt_provenance(state: Any) -> AttemptProvenance: + """Trace each proven optimization back to the research hint it came from.""" + kept_by_name: dict[str, str] = {} + kept_by_gap: dict[str, str] = {} + reverted: list[dict[str, Any]] = [] + for gap in state.gaps or []: + if not isinstance(gap, dict): + continue + from_gap = _gap_attempt_provenance(gap) + for name, source in from_gap.kept_by_name.items(): + kept_by_name.setdefault(name, source) + for canonical_id, source in from_gap.kept_by_gap.items(): + kept_by_gap.setdefault(canonical_id, source) + reverted.extend(from_gap.reverted) + return AttemptProvenance(kept_by_name=kept_by_name, kept_by_gap=kept_by_gap, reverted=reverted) + + +def _best_config(current_best: Mapping[str, Any], optimization_stack: Sequence[Any]) -> dict[str, Any]: + config: dict[str, Any] = {} + current_args = current_best.get("extra_server_args") + if current_args: + config["extra_server_args"] = str(current_args) + for key in ("extra_envs", "name", "tput", "accuracy"): + if key in current_best: + config[key] = current_best[key] + last = optimization_stack[-1] if optimization_stack else None + stack_args = ( + str(last.get("candidate_extra_server_args") or last.get("extra_server_args") or "").strip() + if isinstance(last, dict) + else "" + ) + if stack_args: + config["extra_server_args"] = stack_args + return config + + +def _what_worked( + optimization_stack: Sequence[Any], + gain_per_entry: Sequence[Any], + provenance: AttemptProvenance, +) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for index, entry in enumerate(optimization_stack): + if not isinstance(entry, dict): + continue + name = str(entry.get("variant_name") or entry.get("name") or entry.get("kernel_id") or "") + row: dict[str, Any] = { + "name": name, + "extra_server_args": str(entry.get("extra_server_args") or ""), + "extra_envs": dict(entry.get("extra_envs") or {}), + "gain_pct": gain_per_entry[index] if index < len(gain_per_entry) else None, + } + # The layer's gap id wins; the variant or kernel name answers for + # layers recorded before gap ids were stamped. + gap_id = str(entry.get("gap_canonical_id") or "").strip() + by_name = provenance.kept_by_name + source = (provenance.kept_by_gap.get(gap_id) if gap_id else "") or by_name.get(name) + source = source or by_name.get(str(entry.get("kernel_id") or "")) or "" + if source: + row["source"] = source + rows.append(row) + return rows + + +def _what_failed(recent_failures: Sequence[Any], reverted: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for failure in recent_failures[-10:]: + if isinstance(failure, dict): + rows.append( + { + "name": str(failure.get("name") or failure.get("action") or ""), + "reason": str(failure.get("reason") or failure.get("error_class") or ""), + } + ) + rows.extend(reverted) + return rows + + +def _session_row( + state: Any, + session_name: str, + optimization_stack: Collection[Any], + final_throughput: float, +) -> dict[str, Any]: + validated_stack_len = int(state.cumulative_gain_validated_stack_len or 0) + actions = [ + str(entry.get("variant_name") or entry.get("name") or entry.get("action") or "").strip() + for entry in optimization_stack + if isinstance(entry, dict) + ] + return { + "session_id": str(state.recipe_kb_session_id or session_name), + "gain_pct": float(state.cumulative_gain_validated or 0.0), + "stack_len": validated_stack_len or len(optimization_stack), + "throughput_before": float(state.baseline_tput or 0.0), + "throughput_after": final_throughput, + "date": datetime.now(timezone.utc).isoformat(), + "actions_taken": [name for name in actions if name], + } + + +def recipe_attrs_from_state(state: Any, session_name: str) -> dict[str, Any]: + """Materialise the recipe-shaped view of one session's settled state. + + ``framework_version`` is left unset here because it is manifest-derived; the + T0 backfill writes it. + """ + optimization_stack = state.optimization_stack or [] + # A session that does not sediment its attempts reports no provenance. + provenance = attempt_provenance(state) if state.recipe_sediment_enabled else AttemptProvenance({}, {}, []) + final_throughput = float(state.current_best.get("tput", 0.0)) + return { + "best_config": _best_config(state.current_best, optimization_stack), + "best_throughput": final_throughput, + "what_worked": _what_worked(optimization_stack, state.gain_per_stack_entry or [], provenance), + "what_failed": _what_failed(state.last_action_failures or [], provenance.reverted), + "last_profiled": str(state.cumulative_gain_validated_ts or ""), + "workload": workload_kb_tags(state), + "sessions": [_session_row(state, session_name, optimization_stack, final_throughput)], + } diff --git a/src/hyperloom/orchestrator/knowledge/recipe_journal.py b/src/hyperloom/orchestrator/knowledge/recipe_journal.py index 1c5d117e1c..0d433a6689 100644 --- a/src/hyperloom/orchestrator/knowledge/recipe_journal.py +++ b/src/hyperloom/orchestrator/knowledge/recipe_journal.py @@ -6,13 +6,18 @@ from __future__ import annotations import logging as _logging +from copy import copy from dataclasses import dataclass from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, Collection, Mapping +from types import MappingProxyType +from typing import TYPE_CHECKING, Any, Collection, Mapping, NamedTuple -from hyperloom.common.coerce import to_float, to_int +from hyperloom.common.coerce import first_float, recorded_float, to_float, to_int, to_mapping, to_str_list from hyperloom.common.io import JsonLineFormat, append_jsonl from hyperloom.inference_optimizer.breakdown.recorder import close_out as _close_out +from hyperloom.inference_optimizer.breakdown.recorder.framework_event import ( + ARM_CONFIG, +) from hyperloom.inference_optimizer.recipe_snapshot_constants import detect_framework_version from hyperloom.inference_optimizer.session.optimization_journal import ( OUTCOME_KEEP, @@ -25,11 +30,13 @@ summarize_change, ) from hyperloom.orchestrator.knowledge.recipe_kb import LocalRecipeStoreError, recipe_canonical_id -from hyperloom.orchestrator.knowledge.workload_tags import workload_shape_tags from hyperloom.orchestrator.lever import LEVER_CONFIG, lever_kind_for_task from ..collaborator import CoordinatorCollaborator from ..state.attempt_ledger import AttemptRow, record_attempt_row +from .config import KnowledgeStoreMode +from ._config_timeline import _NON_ATTEMPT_OUTCOMES, _record_config_attempts +from ._recipe_attrs import attempt_provenance, recipe_attrs_from_state, workload_kb_tags if TYPE_CHECKING: from ..loop.coordinator import Coordinator @@ -63,6 +70,103 @@ _REMOTE_VALIDATION_ERROR = "RemoteRecipeValidationError" +#: Explore outcomes the journal collapses into a plain revert. +_MEASURED_LOSS_OUTCOMES: frozenset[str] = frozenset({OUTCOME_REVERT, "FAILED"}) + + +#: Recipe fields an amend carries forward when the caller does not override +#: them, each with the empty value ``put_recipe`` falls back to. +PRESERVED_RECIPE_FIELDS: Mapping[str, Any] = MappingProxyType( + { + "best_config": {}, + "best_throughput": 0.0, + "what_worked": [], + "what_failed": [], + "remaining_gaps": [], + "last_profiled": "", + "stack_fingerprint": {}, + "sessions": [], + "authority": "EXPERIENTIAL", + "confidence": 0.85, + "evidence_refs": [], + } +) + +#: Row keys ``put_recipe`` owns; everything else on a live row is a session +#: extra and survives the amend untouched. +_RECIPE_OWNED_KEYS: frozenset[str] = frozenset(PRESERVED_RECIPE_FIELDS) | frozenset( + { + "canonical_id", + "version", + "created_at", + "updated_at", + "model", + "hardware", + "framework", + "framework_name", + "framework_version", + "precision", + "pitfalls", + "lessons", + "provenance", + } +) + +#: Per-variant measurement detail the journal row carries beyond the headline +#: gain and throughput. +_VARIANT_DETAIL_METRICS: tuple[str, ...] = ( + "runtime_sec", + "wall_clock_ratio_vs_baseline", + "estimated_output_throughput", +) + + +@dataclass(frozen=True) +class FactRecord: + """One settled outcome to write to the Recipe KB as a lesson or a pitfall. + + ``pitfall_severity_dict`` is the result dict per-task, and the variant's merged + metrics + outcome per-variant. ``variant_name`` is set only by the per-variant + caller, and reaching ``provenance_details`` is what it is for. + """ + + task: "Task" + source_session_id: str + is_keep: bool + change: str + gain_pct: float | None + throughput_after: float | None + best_config_candidate: dict[str, Any] | None + evidence_refs: list[str] + pitfall_severity_dict: dict[str, Any] + variant_name: str | None = None + + +@dataclass(frozen=True) +class RemoteRecipeAudit: + """The outcome of one remote Recipe KB publish attempt, as an audit row records it.""" + + status: str + canonical_id: str + session_id: str + primary_metric: str = "" + primary_value: float = 0.0 + reason: str = "" + error: Exception | None = None + + +class _FrameworkIdentity(NamedTuple): + name: str + version: str + precision: str + + +class _VariantMeasurement(NamedTuple): + gain_pct: float | None + throughput_after: float | None + detail: dict[str, Any] + + def _remote_result_type(status: str, reason: str) -> str: """The ``close_out.RESULT_*`` code for a remote KB Store write result. @@ -81,11 +185,10 @@ def _remote_result_type(status: str, reason: str) -> str: def _predicted_gain(*sources: dict[str, Any] | None) -> float | None: - """First non-zero ``predicted_gain_pct`` (``to_float``-parsed) across ordered sources. + """First non-zero ``predicted_gain_pct`` across ordered sources. - Sources are checked in order; a non-zero prediction wins. Returns ``None`` - when none carry a usable value so the journal row stays ``predicted``-free - for unpredicted (default-grid) changes rather than recording a fake 0. + ``None`` when none carry one, so the journal row for an unpredicted + (default-grid) change stays ``predicted``-free rather than recording a fake 0. """ for src in sources: if not isinstance(src, dict): @@ -96,9 +199,8 @@ def _predicted_gain(*sources: dict[str, Any] | None) -> float | None: return None -#: Explore outcomes that are not an attempt at anything: a variant skipped as a -#: duplicate was never measured, so no measurement would back its funnel row. -_NON_ATTEMPT_OUTCOMES = frozenset({"SKIPPED_DEDUP"}) +def _extra_server_args(payload: Mapping[str, Any]) -> str: + return " ".join(to_str_list(payload.get("extra_server_args"))) def _record_config_run(coord: Any, *, task: Any, result_dict: Mapping[str, Any]) -> None: @@ -110,13 +212,13 @@ def _record_config_run(coord: Any, *, task: Any, result_dict: Mapping[str, Any]) read as an arm that was never tried. Recorded at completion, where the outcome is known, with the dispatch time taken off the task row; a grid that measured nothing still lands, which is the case - :func:`_record_config_attempts` never sees. + :func:`record_config_attempts` never sees. """ recorder = coord._coord.phase_framework.timeline() if recorder is None: return from hyperloom.common.timeutil import now_iso - from hyperloom.inference_optimizer.breakdown.recorder.framework_event import ARM_CONFIG, ROLE_CONFIG + from hyperloom.inference_optimizer.breakdown.recorder.framework_event import ROLE_CONFIG task_id = str(getattr(task, "task_id", "") or "") if not task_id: @@ -139,201 +241,127 @@ def _record_config_run(coord: Any, *, task: Any, result_dict: Mapping[str, Any]) ) -def _record_config_attempts( - coord: Any, - *, - task: Any, - per_variant: list[dict[str, Any]], - result_dict: Mapping[str, Any], - adopted_variants: Collection[str], -) -> None: - """Record the configuration arm's measured attempts on the framework event. - - Timeline only: the ledger row is written by ``fact_write_hook``, which does - not depend on a recorder being open. The outcome is recorded verbatim here, - unlike the journal beside it, which collapses ``KEEP_UNSTABLE`` and - ``KILLED_OVERTIME`` into a plain revert. - """ - recorder = coord._coord.phase_framework.timeline() - if recorder is None: - return - from hyperloom.inference_optimizer.breakdown.recorder.framework_event import ARM_CONFIG, GateVerdict - - task_id = str(getattr(task, "task_id", "") or "") - params = getattr(task, "params", None) or {} - round_id = str(result_dict.get("round_id") or "") - recorded = 0 - for row in per_variant: +def rows_not_superseded(live_rows: Any, superseding: Any, key: str) -> list[dict[str, Any]]: + claimed = {str(row.get(key) or "") for row in superseding or [] if isinstance(row, dict)} + kept: list[dict[str, Any]] = [] + for row in live_rows or []: if not isinstance(row, dict): continue - outcome = str(row.get("outcome") or "") - if outcome in _NON_ATTEMPT_OUTCOMES: + if str(row.get(key) or "") in claimed: continue - metrics = row.get("metrics") if isinstance(row.get("metrics"), dict) else {} - variant = row.get("variant") if isinstance(row.get("variant"), dict) else {} - fingerprint = str(row.get("fingerprint") or "") - adopted = fingerprint in adopted_variants - # The fingerprint identifies the variant within the round and the round - # within the task, so the three together identify the attempt. - attempt_id = ":".join(part for part in (task_id, round_id, fingerprint) if part) or task_id - if not attempt_id: + kept.append(dict(row)) + return kept + + +def appended(live_facts: Any, addition: Mapping[str, Any] | None) -> list[Any]: + facts = list(live_facts or []) + if addition is not None: + facts.append(addition) + return facts + + +def amended_recipe_fields(live: Mapping[str, Any], overrides: Mapping[str, Any]) -> dict[str, Any]: + # An absent or empty live value falls back to the field's own default, so + # ``put_recipe`` never sees ``None`` where it declares a container. + merged: dict[str, Any] = {} + for field, default in PRESERVED_RECIPE_FIELDS.items(): + if field in overrides: + merged[field] = overrides[field] continue - gates = [gate for gate in (row.get("gates") or []) if isinstance(gate, dict)] - # This arm gates accuracy rather than reporting it, so the block both - # arms carry is projected from the gate that ruled. No gate row means - # nothing gated the variant, which is not a gate that refused it. - accuracy_gate = next((gate for gate in gates if str(gate.get("gate") or "") == "accuracy"), {}) - error_class = str(row.get("error_class") or "") - error_excerpt = str(row.get("error_excerpt") or "") - recorder.record_attempt( - attempt_id, - arm=ARM_CONFIG, - round_id=round_id, - task_id=task_id, - proposal_ref=str(params.get("proposal_msg_id") or ""), - provenance=str(row.get("provenance") or ""), - outcome=outcome, - reason=str(row.get("reason") or ""), - reasoning=str(variant.get("note") or ""), - reasoning_origin=str(variant.get("reasoning_origin") or ""), - experience_citations=variant.get("experience_citations") or [], - stage=str(row.get("stage") or ""), - fingerprint=fingerprint, - # The fingerprint is the join key; the name is what a reader - # recognises the variant by. - variant_name=str(row.get("variant_name") or ""), - measurement={ - # Both ends of the pair: the anchor advances on every KEEP, - # so a percentage without its denominator adds to nothing. - "before_tput": metrics.get("base_tput"), - "after_tput": metrics.get("tput"), - "gain_pct": metrics.get("gain_pct"), - "runtime_sec": metrics.get("runtime_sec"), - "estimated_output_throughput": metrics.get("estimated_output_throughput"), - }, - config_delta={ - "extra_server_args": variant.get("extra_server_args"), - "extra_envs": variant.get("extra_envs"), - "remove_args": variant.get("remove_args"), - "unset_envs": variant.get("unset_envs"), - "args_mode": variant.get("args_mode"), - }, - accuracy={ - "required": True if accuracy_gate else None, - "value": accuracy_gate.get("observed"), - "reference": accuracy_gate.get("threshold"), - "passed": accuracy_gate.get("passed"), - }, - failure={ - "error_class": error_class, - "error_excerpt": error_excerpt, - "attribution": str(row.get("failure_attribution") or ""), - }, - artifacts={ - "workspace": str(row.get("workspace") or ""), - "server_log_path": str(row.get("server_log_path") or ""), - "raw_result_path": str(row.get("raw_result_path") or ""), - }, - decision=outcome, - adopted=adopted, - # Every variant recorded here ran: the deduped one is skipped above. - benched=True, - # Recorded rather than referenced: every KEEP advances the - # stack, so the session's current config is not what this - # variant was measured on top of. - measured_against=row.get("measured_against") or {}, - # What stood behind the verdict. Absent when nothing ruled. - validation_basis=str(row.get("validation_basis") or ""), - # A pair is what makes a gain addable, so eligibility follows - # the pair being present rather than the outcome being a KEEP. - attribution_eligible=(adopted and metrics.get("base_tput") is not None and metrics.get("tput") is not None), - ) - for gate in gates: - if not str(gate.get("gate") or ""): - continue - recorder.record_attempt_gate( - attempt_id, - GateVerdict( - gate=str(gate.get("gate")), - passed=gate.get("passed"), - reason=str(gate.get("reason") or ""), - observed=gate.get("observed"), - threshold=gate.get("threshold"), - ), - ) - recorded += 1 - proposal_ref = str(params.get("proposal_msg_id") or "") - if not (recorded and proposal_ref): - return - # Settled here rather than at materialization: a grid whose task dies before - # any variant runs has no attempt to stand behind an ``attempted`` reading, - # and is more honestly left unsettled. - from hyperloom.inference_optimizer.breakdown.recorder.framework_event import ( - DISPOSITION_ATTEMPTED, - STEP_ATTEMPTED, + live_value = live.get(field) + merged[field] = type(default)(live_value) if live_value else copy(default) + return merged + + +def amended_recipe_extras(live: Mapping[str, Any], overrides: Mapping[str, Any], state: Any) -> dict[str, Any]: + # The architecture tags are re-stamped from config.json each time because a + # row written before the architecture was read carries none. + extras = {key: value for key, value in live.items() if key not in _RECIPE_OWNED_KEYS} + extras.update(overrides.get("extras") or {}) + architectures = state.model_architectures or [] + if isinstance(architectures, list): + named = [str(name).strip() for name in architectures if str(name or "").strip()] + if named: + extras["architectures"] = named + model_type = str(state.model_type or "").strip() + if model_type: + extras["model_type"] = model_type + return extras + + +def _framework_identity(state: Any) -> _FrameworkIdentity: + # Probed from the installed framework when the session never recorded a + # version, so a recipe is never filed under a blank one. + name = str(state.framework or "") + version = str(state.framework_version or "") + if not version and name: + version = detect_framework_version(name) + return _FrameworkIdentity(name=name, version=version, precision=str(state.precision or "")) + + +def _result_payload(result: Any) -> dict[str, Any]: + payload = result.result if hasattr(result, "result") else (result or {}) + return payload if isinstance(payload, dict) else {} + + +def _grid_attempt_row( + variant_outcome: Any, + *, + task_id: str, + round_id: str, + adopted: bool, +) -> AttemptRow | None: + outcome = str(variant_outcome.get("outcome") or "") if isinstance(variant_outcome, dict) else "" + if not outcome or outcome in _NON_ATTEMPT_OUTCOMES: + return None + metrics = variant_outcome.get("metrics") if isinstance(variant_outcome.get("metrics"), dict) else {} + return AttemptRow( + lever_kind=LEVER_CONFIG, + task_id=task_id, + round_id=round_id, + fingerprint=str(variant_outcome.get("fingerprint") or ""), + variant_name=str(variant_outcome.get("variant_name") or ""), + outcome=outcome, + adopted=adopted, + gain_pct=metrics.get("gain_pct"), + before_tput=metrics.get("base_tput"), + after_tput=metrics.get("tput"), + error_class=str(variant_outcome.get("error_class") or ""), + provenance=str(variant_outcome.get("provenance") or ""), ) - recorder.record_proposal_step(proposal_ref, step=STEP_ATTEMPTED, outcome=str(recorded)) - recorder.settle_proposal(proposal_ref, disposition=DISPOSITION_ATTEMPTED) +def _variant_verdict(outcome: str, adopted: bool) -> Verdict: + if outcome == OUTCOME_KEEP: + return Verdict.ADOPTED if adopted else Verdict.REFUSED + if outcome in _MEASURED_LOSS_OUTCOMES: + return Verdict.REVERTED + return Verdict.REFUSED -def _extra_server_args(payload: Mapping[str, Any]) -> str: - """Read canonical ``extra_server_args`` from a payload.""" - value = payload.get("extra_server_args") - if value is None: - return "" - if isinstance(value, str): - return value - if isinstance(value, (list, tuple)): - return " ".join(str(v).strip() for v in value if str(v).strip()) - return str(value) +def _variant_measurement(metrics: Mapping[str, Any]) -> _VariantMeasurement: + return _VariantMeasurement( + gain_pct=to_float(metrics.get("gain_pct")), + throughput_after=to_float(metrics.get("output_throughput")), + detail={key: metrics[key] for key in _VARIANT_DETAIL_METRICS if metrics.get(key) is not None}, + ) -@dataclass(frozen=True) -class FactRecord: - """One settled outcome to write to the Recipe KB as a lesson or a pitfall. - Attributes: - task: The completed task (provides task_id). - source_session_id: Hyperloom-local session id stamped on provenance. - is_keep: True when the outcome is a validated KEEP. - change: Summarized change string (used in the statement). - gain_pct: Measured gain percentage, or ``None``. - throughput_after: Measured throughput after the change, or ``None``. - best_config_candidate: Pre-extracted best-config dict (differs between - per-task and per-variant callers). - evidence_refs: Evidence reference strings stamped on the provenance - (task-only or task+variant refs). - pitfall_severity_dict: The dict passed to ``_pitfall_severity_for`` - (per-task passes the result dict; per-variant passes a merged - metrics + outcome dict). - variant_name: Variant name, present only for per-variant calls; added to - ``provenance_details`` when non-None. - """ +def _variant_change_attrs(variant_attrs: Any, variant_name: str) -> dict[str, Any]: + attrs = dict(variant_attrs) if isinstance(variant_attrs, dict) else {} + named_knob = attrs.get("extra_server_args") or attrs.get("extra_envs") or attrs.get("name") + if not named_knob and variant_name: + attrs["name"] = variant_name + return attrs - task: "Task" - source_session_id: str - is_keep: bool - change: str - gain_pct: float | None - throughput_after: float | None - best_config_candidate: dict[str, Any] | None - evidence_refs: list[str] - pitfall_severity_dict: dict[str, Any] - variant_name: str | None = None +def _finalize_outcome(*, status: str, reason: str, backend: str, result_type: str) -> dict[str, Any]: + return {"status": status, "reason": reason, "backend": backend, "result_type": result_type} -@dataclass(frozen=True) -class RemoteRecipeAudit: - """The outcome of one remote Recipe KB publish attempt, as an audit row records it.""" - status: str - canonical_id: str - session_id: str - primary_metric: str = "" - primary_value: float = 0.0 - reason: str = "" - error: Exception | None = None +def _finalize_error(exc: BaseException, *, reason: str, backend: str, result_type: str) -> dict[str, Any]: + failed = _finalize_outcome(status="error", reason=reason, backend=backend, result_type=result_type) + return {**failed, "error_class": type(exc).__name__} class RecipeJournalCollaborator(CoordinatorCollaborator): @@ -366,7 +394,7 @@ def journal_integrate_keep(self, result: dict[str, Any], *, lift_kind: str, new_ journal.append_entry( JournalEntry( phase=self.journal_entry_phase(), - lever_kind=lever_kind_for_task(lift_kind, result if isinstance(result, dict) else None), + lever_kind=lever_kind_for_task(lift_kind, result), iter=int(self.shared_state.tick or 0), kind=kind, change=change, @@ -383,14 +411,7 @@ def journal_integrate_keep(self, result: dict[str, Any], *, lift_kind: str, new_ log.exception("integrate journal append failed") def _source_session_id(self) -> str: - """Return the hyperloom-local session id used as source_session_id on KB fact writes. - - NOT a KB-side session id; prefers recipe_kb_session_id, falls back to session_dir.name. - - Returns: - The hyperloom-local session id (recipe_kb_session_id when set, else - ``session_dir.name``). - """ + """The hyperloom-local session id stamped on KB fact writes, never a KB-side one.""" return str(self.shared_state.recipe_kb_session_id or "") or self.session_dir.name async def fact_write_hook( @@ -403,16 +424,10 @@ async def fact_write_hook( ) -> None: """Per-task fact-write entry point (per_variant for explore grids, else per-task). - Args: - task: The completed task being recorded. - result: The task's :class:`SubAgentResult` (or result dict). - verdict: The settlement verdict. - adopted_variants: Fingerprints of the explore winners whose lift - landed; a per-variant KEEP outside it did not adopt anything. + ``adopted_variants`` holds the fingerprints of the explore winners whose lift + landed; a per-variant KEEP outside it did not adopt anything. """ - result_dict = result.result if hasattr(result, "result") else (result or {}) - if not isinstance(result_dict, dict): - result_dict = {} + result_dict = _result_payload(result) source_session_id = self._source_session_id() per_variant = result_dict.get("per_variant_outcomes") if task.kind == "explore": @@ -426,25 +441,11 @@ async def fact_write_hook( adopted_variants=adopted_variants, ) round_id = str(result_dict.get("round_id") or "") + task_id = str(task.task_id or "") for vo in per_variant: adopted = str(vo.get("fingerprint") or "") in adopted_variants - outcome = str(vo.get("outcome") or "") if isinstance(vo, dict) else "" - if outcome and outcome not in _NON_ATTEMPT_OUTCOMES: - metrics = vo.get("metrics") if isinstance(vo.get("metrics"), dict) else {} - row = AttemptRow( - lever_kind=LEVER_CONFIG, - task_id=str(task.task_id or ""), - round_id=round_id, - fingerprint=str(vo.get("fingerprint") or ""), - variant_name=str(vo.get("variant_name") or ""), - outcome=outcome, - adopted=adopted, - gain_pct=metrics.get("gain_pct"), - before_tput=metrics.get("base_tput"), - after_tput=metrics.get("tput"), - error_class=str(vo.get("error_class") or ""), - provenance=str(vo.get("provenance") or ""), - ) + row = _grid_attempt_row(vo, task_id=task_id, round_id=round_id, adopted=adopted) + if row is not None: record_attempt_row(self.shared_state, row) self._record_fact_per_variant( task=task, source_session_id=source_session_id, variant_outcome=vo, adopted=adopted @@ -459,12 +460,7 @@ async def fact_write_hook( self.shared_state.save(self.session_dir) def ensure_journal(self) -> Journal: - """Lazy-instantiate the per-session :class:`Journal` (load_or_create reads an existing file on resume). - - Returns: - The per-session :class:`Journal` instance (created on first call, - with the baseline backfilled on subsequent calls). - """ + """Lazy-instantiate the per-session :class:`Journal` (load_or_create reads an existing file on resume).""" existing = self._journal if existing is None: ss = self.shared_state @@ -485,22 +481,8 @@ def _pitfall_severity_for( self, result_dict: dict[str, Any] | None, ) -> str | None: - """Decide whether a failed result warrants a pitfall row. - - ``crash`` / ``oom`` / ``hang`` / ``detokenizer_stall`` on ``error_class``, - or ``crash`` / ``oom`` / ``hang`` on ``status``, yield - ``SEVERITY_CRASH``; a ``gain_pct`` at or below - ``PITFALL_REGRESS_THRESHOLD_PCT`` (-5.0) yields ``SEVERITY_REGRESS``; - otherwise ``None``. - - Args: - result_dict: The failed task's result dict; non-dict yields ``None``. - - Returns: - The pitfall severity (``SEVERITY_CRASH`` / ``SEVERITY_REGRESS``), or - ``None`` when no pitfall is warranted. - """ - if not isinstance(result_dict, dict): + """The severity of the pitfall a failed result warrants, or ``None`` for no pitfall.""" + if not result_dict: return None error_class = str(result_dict.get("error_class") or "").lower() # ``detokenizer_stall`` is a hang in all but name; record it as a @@ -510,48 +492,30 @@ def _pitfall_severity_for( status = str(result_dict.get("status") or "").lower() if status in ("crash", "oom", "hang"): return _SEVERITY_CRASH - gain = result_dict.get("gain_pct") - try: - gain_pct = float(gain) if gain is not None else None - except (TypeError, ValueError): - gain_pct = None + gain_pct = to_float(result_dict.get("gain_pct")) if gain_pct is not None and gain_pct <= self.PITFALL_REGRESS_THRESHOLD_PCT: return _SEVERITY_REGRESS return None def journal_entry_phase(self) -> str: - """Return the current phase label for journal entries. - - Returns: - str: The uppercased phase name, or ``"UNKNOWN"`` when unset. - """ + """The uppercased phase name journal entries are stamped with, ``"UNKNOWN"`` when unset.""" return str(self.shared_state.phase or "").strip().upper() or "UNKNOWN" def _record_fact_impl(self, fact: FactRecord) -> None: - """Shared KB write for _record_fact_per_task and _record_fact_per_variant. - - Writes one KB lesson (on KEEP with positive gain) or one KB pitfall - (on REVERT/failure) to the recipe row, then returns. Call only after - the journal entry has been appended and ``recipe_kb`` is confirmed - non-None by the caller. + """Write one KB lesson (KEEP with positive gain) or one KB pitfall (REVERT/failure). - Args: - fact: The completed task, its outcome and the evidence to stamp on - the provenance. + Call only after the journal entry has been appended and ``recipe_kb`` is + confirmed non-None by the caller. """ - models = [str(self.shared_state.model_name or "")] if self.shared_state.model_name else [] - hardware = [str(self.shared_state.gpu_type or "")] if self.shared_state.gpu_type else [] - workload_tags = self._collect_workload_tags() - extra = workload_tags if workload_tags else None + state = self.shared_state now_iso = datetime.now(timezone.utc).isoformat(timespec="seconds") - provenance_base: dict[str, Any] = { "source_session_id": fact.source_session_id, "source_task_id": fact.task.task_id, "evidence": list(fact.evidence_refs or []), - "applicable_models": list(models or []), - "applicable_hardware": list(hardware or []), - "extra": dict(extra or {}), + "applicable_models": [str(state.model_name)] if state.model_name else [], + "applicable_hardware": [str(state.gpu_type)] if state.gpu_type else [], + "extra": dict(workload_kb_tags(state)), "now": now_iso, } if fact.variant_name is not None: @@ -609,23 +573,16 @@ def _record_fact_per_task( result_dict: dict[str, Any], verdict: Verdict, ) -> None: - """Per-task fact write — one journal row + maybe one KB fact (source_session_id is hyperloom-local). - - Args: - task: The completed task being recorded. - source_session_id: The hyperloom-local session id stamped on the - fact provenance. - result_dict: The task result dict. - verdict: The settlement verdict (ADOPTED → KEEP and a lesson, else - a non-KEEP row and maybe a pitfall). + """Per-task fact write — one journal row + maybe one KB fact. + + ``ADOPTED`` journals a KEEP and may write a lesson; any other verdict + journals a non-KEEP row and may write a pitfall. """ journal = self.ensure_journal() # integrate_patch reports its delta under ``delta_pct``; # fall back to it so a reverted/kept patch shows its REAL measured delta # in the journal instead of a null gain. - gain_pct = to_float(result_dict.get("gain_pct")) - if gain_pct is None: - gain_pct = to_float(result_dict.get("delta_pct")) + gain_pct = first_float(result_dict.get("gain_pct"), result_dict.get("delta_pct")) throughput_after = to_float(result_dict.get("output_throughput")) kind = classify_change_kind(task.kind, None) change = summarize_change(task.kind, None, result_dict) @@ -642,7 +599,7 @@ def _record_fact_per_task( journal.append_entry( JournalEntry( phase=self.journal_entry_phase(), - lever_kind=lever_kind_for_task(kind, result_dict if isinstance(result_dict, dict) else None), + lever_kind=lever_kind_for_task(kind, result_dict), iter=int(self.shared_state.tick or 0), kind=kind, change=change, @@ -659,10 +616,8 @@ def _record_fact_per_task( ), ) ) - if self.recipe_kb is None: return - self._record_fact_impl( FactRecord( task=task, @@ -688,16 +643,10 @@ def _build_statement( kind: str, severity: str | None = None, ) -> str: - """Build the lesson statement / pitfall description hashed into the KB canonical_id; MUST exclude volatile fields (e.g. gain_pct) so N sessions merge instead of producing N rows. Identity = framework + change + model/hw. + """Build the lesson statement / pitfall description hashed into the KB canonical_id. - Args: - change: The summarized change description. - kind: ``"lesson"`` or ``"pitfall"`` — selects the rendered form. - severity: The pitfall severity, rendered only when ``kind`` is - ``"pitfall"``. - - Returns: - The identity-stable statement / description string. + Identity is framework + change + model/hw. It MUST exclude volatile fields + (``gain_pct`` and the like) so N sessions merge instead of producing N rows. """ framework = str(self.shared_state.framework or "").strip() fw_tag = f"[{framework or '?'}] " @@ -716,16 +665,9 @@ def _build_measured_impact( stack_depth: int, measured_at: str, ) -> dict[str, Any]: - """Structured ``measured_impact`` payload (dict not legacy string so consumers parse without regex); stack_depth = stack length before this lesson lands. - - Args: - gain_pct: The measured gain percent, or ``None``. - throughput_after: Throughput after the change, or ``None``. - stack_depth: Optimization-stack length before this lesson lands. - measured_at: ISO timestamp of the measurement. + """Structured ``measured_impact`` payload, a dict rather than a legacy string so consumers parse without regex. - Returns: - A compact ``measured_impact`` dict with ``None`` fields stripped. + ``stack_depth`` is the optimization-stack length before this lesson lands. """ out: dict[str, Any] = { "gain_pct": float(gain_pct) if gain_pct is not None else None, @@ -747,84 +689,47 @@ def _record_fact_per_variant( ) -> None: """Per-variant fact write — mirror of _record_fact_per_task for explore per-variant decisions. - Args: - task: The completed explore task. - source_session_id: The hyperloom-local session id stamped on the - fact provenance. - variant_outcome: One per-variant outcome row (name, outcome, - metrics). - adopted: Whether this variant's lift landed; an executor KEEP that - was not adopted journals as ``no_promote``. + An executor KEEP that was not adopted journals as ``no_promote``. """ journal = self.ensure_journal() outcome_raw = str(variant_outcome.get("outcome") or "") - if outcome_raw == "SKIPPED_DEDUP": + if outcome_raw in _NON_ATTEMPT_OUTCOMES: return - if outcome_raw == "KEEP": - verdict = Verdict.ADOPTED if adopted else Verdict.REFUSED - elif outcome_raw in ("REVERT", "FAILED"): - verdict = Verdict.REVERTED - else: - verdict = Verdict.REFUSED - outcome = derive_journal_outcome(verdict, variant_outcome) + outcome = derive_journal_outcome(_variant_verdict(outcome_raw, adopted), variant_outcome) variant_name = str(variant_outcome.get("variant_name") or "") - metrics = variant_outcome.get("metrics") or {} - gain_pct = to_float(metrics.get("gain_pct") if isinstance(metrics, dict) else None) - throughput_after = to_float(metrics.get("output_throughput") if isinstance(metrics, dict) else None) + metrics = to_mapping(variant_outcome.get("metrics")) + measurement = _variant_measurement(metrics) variant_attrs = variant_outcome.get("variant") or {} - kind = classify_change_kind( - task.kind, - variant_attrs if isinstance(variant_attrs, dict) else None, - ) - # Ensure the change summary is variant-specific (else every explore variant writes an identical row). - change_attrs = dict(variant_attrs) if isinstance(variant_attrs, dict) else {} - if ( - not (change_attrs.get("extra_server_args") or change_attrs.get("extra_envs") or change_attrs.get("name")) - and variant_name - ): - change_attrs["name"] = variant_name + attrs = variant_attrs if isinstance(variant_attrs, dict) else None + kind = classify_change_kind(task.kind, attrs) + change_attrs = _variant_change_attrs(variant_attrs, variant_name) change = summarize_change(task.kind, change_attrs, None) - error_class = None - reason = None - if outcome == OUTCOME_REVERT: - error_class = str(variant_outcome.get("error_class") or "") or None - reason = str(variant_outcome.get("reason") or "") or None - # Proposer attribution + per-variant measurement detail, carried from the - # explore executor's per_variant_outcomes so the decision row records who - # proposed the change and how it measured (beyond headline gain/tput). - detail_metrics = { - k: metrics[k] - for k in ( - "runtime_sec", - "wall_clock_ratio_vs_baseline", - "estimated_output_throughput", - ) - if isinstance(metrics, dict) and metrics.get(k) is not None - } + # Only a measured loss carries a failure; a refusal has nothing to report. + reverted = outcome == OUTCOME_REVERT + tick = int(self.shared_state.tick or 0) journal.append_entry( JournalEntry( phase=self.journal_entry_phase(), - lever_kind=lever_kind_for_task(kind, variant_outcome if isinstance(variant_outcome, dict) else None), - iter=int(self.shared_state.tick or 0), + lever_kind=lever_kind_for_task(kind, variant_outcome), + iter=tick, kind=kind, change=change, outcome=outcome, - gain_pct=gain_pct, - throughput_after=throughput_after, - error_class=error_class, - reason=reason, + gain_pct=measurement.gain_pct, + throughput_after=measurement.throughput_after, + error_class=(str(variant_outcome.get("error_class") or "") or None) if reverted else None, + reason=(str(variant_outcome.get("reason") or "") or None) if reverted else None, task_id=task.task_id, variant_name=variant_name, provenance=str(variant_outcome.get("provenance") or ""), scope=str(variant_outcome.get("scope") or ""), fingerprint=str(variant_outcome.get("fingerprint") or ""), - metrics=detail_metrics, - tick=int(self.shared_state.tick or 0), - predicted_gain_pct=_predicted_gain( - variant_outcome, - variant_attrs if isinstance(variant_attrs, dict) else None, - getattr(task, "params", None), - ), + # Proposer attribution and the measurement detail beyond the + # headline figures, so the row records how the change measured + # and who proposed it. + metrics=measurement.detail, + tick=tick, + predicted_gain_pct=_predicted_gain(variant_outcome, attrs, getattr(task, "params", None)), ) ) @@ -837,8 +742,8 @@ def _record_fact_per_variant( source_session_id=source_session_id, is_keep=outcome == OUTCOME_KEEP, change=change, - gain_pct=gain_pct, - throughput_after=throughput_after, + gain_pct=measurement.gain_pct, + throughput_after=measurement.throughput_after, best_config_candidate=self.extract_kept_best_config( task=task, variant_attrs=change_attrs, @@ -846,7 +751,7 @@ def _record_fact_per_variant( # Workload-shape tags — see _record_fact_per_task. evidence_refs=[f"log:task-{task.task_id}", f"variant:{variant_name}"], pitfall_severity_dict={ - **(metrics if isinstance(metrics, dict) else {}), + **metrics, "error_class": variant_outcome.get("error_class"), "status": variant_outcome.get("outcome"), }, @@ -854,202 +759,6 @@ def _record_fact_per_variant( ) ) - def _collect_workload_tags(self) -> dict[str, Any]: - """Return the workload-shape KB tag dict for the current session; shared by recipe attrs + lesson/pitfall writes so the warm-start reader filters symmetrically. - - Returns: - A dict of workload-shape KB tags (framework, model, parallelism, - runtime versions, baseline workload extras) with empty values - omitted. - """ - ss = self.shared_state - out: dict[str, Any] = {} - # model_family is not part of the seven-dimension Recipe identity. - for attr in ("framework", "model_class", "model_name"): - text = str(getattr(ss, attr, "") or "").strip() - if text: - out[attr] = text - precision = getattr(ss, "precision", None) - if precision not in (None, "", 0): - out["precision"] = precision - out.update(workload_shape_tags(ss)) - # runtime version tags from stack_fingerprint_meta (cli writes at boot, resume reads verbatim). - fp_meta = ss.stack_fingerprint_meta or {} - if isinstance(fp_meta, dict): - # framework_version is whichever of sglang/vllm is active. - fw_lc = out.get("framework", "").lower() - if fw_lc in ("sglang", "vllm"): - version = str(fp_meta.get(fw_lc) or "").strip() - if version and version != "unknown": - out["framework_version"] = version - for src_key, dst_key in ( - ("rocm", "rocm_version"), - ("aiter", "aiter_version"), - ("image_digest", "image_digest"), - ): - version = str(fp_meta.get(src_key) or "").strip() - if version and version != "unknown": - out[dst_key] = version - # per-baseline workload extras from materialized YAML; keep bool False (don't drop an "explicitly disabled" signal). - wl_extra = ss.baseline_workload_extra or {} - if isinstance(wl_extra, dict): - for k in ("max_running_requests", "max_num_seqs"): - v = wl_extra.get(k) - if isinstance(v, int) and v > 0: - out[k] = v - for k in ("chunked_prefill_enabled", "enable_torch_compile"): - v = wl_extra.get(k) - if isinstance(v, bool): - out[k] = v - for k in ("quant_scheme", "workload_mode"): - v = wl_extra.get(k) - if isinstance(v, str) and v.strip(): - out[k] = v.strip() - return out - - def _collect_attempt_provenance( - self, - ) -> tuple[dict[str, str], dict[str, str], list[dict[str, Any]]]: - """Map proven optimizations to their research-hint origin from the gaps[] attempts ledger; returns (kept_sources by name/kernel, kept_by_gap by canonical_id, reverted_rows). Fail-soft. - - Returns: - A ``(kept_sources, kept_by_gap, reverted_rows)`` tuple: KEEP'd - provenance keyed by variant/kernel name, KEEP'd provenance keyed by - gap canonical_id, and reverted-attempt rows. - """ - kept_sources: dict[str, str] = {} - kept_by_gap: dict[str, str] = {} - reverted_rows: list[dict[str, Any]] = [] - gaps = self.shared_state.gaps or [] - for gap in gaps: - if not isinstance(gap, dict): - continue - provenance = str(gap.get("provenance") or "").strip() - canonical = str(gap.get("canonical_id") or "").strip() - for attempt in gap.get("attempts") or []: - if not isinstance(attempt, dict): - continue - variant = str(attempt.get("variant_name") or "").strip() - kernel = str(attempt.get("kernel_id") or "").strip() - outcome = str(attempt.get("outcome") or "").strip().upper() - if outcome == "KEEP" and provenance: - if variant: - kept_sources.setdefault(variant, provenance) - if kernel: - kept_sources.setdefault(kernel, provenance) - if canonical: - kept_by_gap.setdefault(canonical, provenance) - elif outcome == "REVERT" and (variant or kernel): - row: dict[str, Any] = { - "name": variant or kernel, - "reason": "reverted", - "gain_pct": attempt.get("gain_pct"), - } - if provenance: - row["source"] = provenance - reverted_rows.append(row) - return kept_sources, kept_by_gap, reverted_rows - - def _build_recipe_attrs_from_state(self) -> dict[str, Any]: - """Materialise the recipe-shaped view of :class:`SharedState` (defensive getattr). - - Returns: - A recipe-shaped attrs dict (best_config, what_worked, what_failed, - workload tags, session row) for KB recipe writes. - """ - ss = self.shared_state - current_best = ss.current_best - opt_stack = ss.optimization_stack or [] - gain_per_stack = ss.gain_per_stack_entry or [] - last_failures = ss.last_action_failures or [] - # RecipeKB best_config keys on the canonical extra_server_args field. - best_config: dict[str, Any] = {} - cb_args = current_best.get("extra_server_args") - if cb_args: - best_config["extra_server_args"] = str(cb_args) - for key in ("extra_envs", "name", "tput", "accuracy"): - if key in current_best: - best_config[key] = current_best[key] - # Prefer the last validated stack layer for launch args (current_best may carry a corrupted string). - if opt_stack: - last_entry = opt_stack[-1] - if isinstance(last_entry, dict): - stack_args = str( - last_entry.get("candidate_extra_server_args") or last_entry.get("extra_server_args") or "", - ).strip() - if stack_args: - best_config["extra_server_args"] = stack_args - sediment_on = bool(ss.recipe_sediment_enabled) - kept_sources, kept_by_gap, reverted_rows = self._collect_attempt_provenance() if sediment_on else ({}, {}, []) - what_worked: list[dict[str, Any]] = [] - for idx, entry in enumerate(opt_stack): - if not isinstance(entry, dict): - continue - gain_per: float | None = None - if idx < len(gain_per_stack): - gain_per = gain_per_stack[idx] - name = str(entry.get("variant_name") or entry.get("name") or entry.get("kernel_id") or "") - row: dict[str, Any] = { - "name": name, - "extra_server_args": str(entry.get("extra_server_args") or ""), - "extra_envs": dict(entry.get("extra_envs") or {}), - "gain_pct": gain_per, - } - # Prefer the entry's gap-id provenance (naming-independent); fall back to name/kernel_id match. - entry_gap = str(entry.get("gap_canonical_id") or "").strip() - src = ( - (kept_by_gap.get(entry_gap) if entry_gap else None) - or kept_sources.get(name) - or kept_sources.get(str(entry.get("kernel_id") or "")) - ) - if src: - row["source"] = src - what_worked.append(row) - what_failed: list[dict[str, Any]] = [] - for failure in last_failures[-10:]: - if isinstance(failure, dict): - what_failed.append( - { - "name": str(failure.get("name") or failure.get("action") or ""), - "reason": str(failure.get("reason") or failure.get("error_class") or ""), - } - ) - for rev in reverted_rows: - what_failed.append(rev) - cumulative_validated = float(ss.cumulative_gain_validated or 0.0) - validated_stack_len = int(ss.cumulative_gain_validated_stack_len or 0) - # Workload-shape tags for shape-filtered warm-start queries (shared via _collect_workload_tags). - workload_tags = self._collect_workload_tags() - # framework_version left unset here (manifest-derived); the T0 backfill writes it. - return { - "best_config": best_config, - "best_throughput": float(current_best.get("tput", 0.0)), - "what_worked": what_worked, - "what_failed": what_failed, - "last_profiled": str(ss.cumulative_gain_validated_ts or ""), - "workload": workload_tags, - "sessions": [ - { - "session_id": str(ss.recipe_kb_session_id or self.session_dir.name), - "gain_pct": cumulative_validated, - "stack_len": validated_stack_len or len(opt_stack), - # arbor-shape provenance so the session row is self-describing (before/after tput + knobs). - "throughput_before": float(ss.baseline_tput or 0.0), - "throughput_after": float(current_best.get("tput", 0.0)), - "date": datetime.now(timezone.utc).isoformat(), - "actions_taken": [ - nm - for nm in ( - str(e.get("variant_name") or e.get("name") or e.get("action") or "").strip() - for e in opt_stack - if isinstance(e, dict) - ) - if nm - ], - } - ], - } - def _record_remote_recipe_audit(self, source: str, audit: RemoteRecipeAudit) -> None: """Append one best-effort, secret-free KB Store publish audit row.""" try: @@ -1198,10 +907,7 @@ def _record_write_back_settled( ) def _write_back_scope(self) -> dict[str, Any]: - """The workload dimensions the published recipe is partitioned by. - - Matches the ``RecipeScope`` shape the section already published. - """ + """The workload dimensions the published recipe is partitioned by, in ``RecipeScope`` shape.""" state = self.shared_state return { "kernel_optimizer": str(state.kernel_optimizer or ""), @@ -1211,217 +917,209 @@ def _write_back_scope(self) -> dict[str, Any]: "osl": to_int(state.osl, None), } + def _unvalidated_stack_outcome(self) -> dict[str, Any] | None: + # ``current_best`` and ``cumulative_gain_validated`` are only one + # Recipe's figures when nothing was lifted since the last validation; + # otherwise every sink below would pair the newer config with the + # older gain. + state = self.shared_state + if not state.optimization_stack_has_unvalidated_keeps(): + return None + log.warning( + "Recipe finalize skipped: working recipe is not validated (stack=%s/%s generation=%s/%s)", + len(state.optimization_stack or []), + state.cumulative_gain_validated_stack_len, + state.working_recipe_generation, + state.validated_recipe_generation, + ) + return _finalize_outcome( + status="skipped", + reason="unvalidated_recipe_stack", + backend="none", + result_type=_close_out.RESULT_UNVALIDATED_RECIPE, + ) + def finalize_recipe_and_journal( self, *, source: str = "close", ) -> dict[str, Any]: """Finalize Recipe state and return a secret-free observable outcome.""" - # ``current_best`` and ``cumulative_gain_validated`` are only one Recipe's - # figures when nothing was lifted since the last validation. Otherwise - # every sink below -- journal, local KB, remote KB -- would pair the - # newer config with the older gain, so none of them runs. - if self.shared_state.optimization_stack_has_unvalidated_keeps(): - ss = self.shared_state - log.warning( - "Recipe finalize skipped: working recipe is not validated (stack=%s/%s generation=%s/%s)", - len(ss.optimization_stack or []), - ss.cumulative_gain_validated_stack_len, - ss.working_recipe_generation, - ss.validated_recipe_generation, - ) - return { - "status": "skipped", - "reason": "unvalidated_recipe_stack", - "backend": "none", - "result_type": _close_out.RESULT_UNVALIDATED_RECIPE, - } - journal = self.ensure_journal() - ss = self.shared_state - final_tput = float(ss.current_best.get("tput", 0.0)) - total_gain = float(ss.cumulative_gain_validated or 0.0) - journal.finalize( + unvalidated = self._unvalidated_stack_outcome() + if unvalidated is not None: + return unvalidated + state = self.shared_state + final_tput = float(state.current_best.get("tput", 0.0)) + self.ensure_journal().finalize( final_throughput=final_tput if final_tput > 0 else None, - total_gain_pct=total_gain, + total_gain_pct=float(state.cumulative_gain_validated or 0.0), ) - kp = self.knowledge_plane - if bool(getattr(kp, "kb_disabled", False)): + knowledge_plane = self.knowledge_plane + if bool(getattr(knowledge_plane, "kb_disabled", False)): log.info("Recipe KB finalize skipped (--degraded-kb)") - return { - "status": "skipped", - "reason": "degraded_kb", - "backend": "disabled", - "result_type": _close_out.RESULT_KB_DISABLED, - } + return _finalize_outcome( + status="skipped", + reason="degraded_kb", + backend="disabled", + result_type=_close_out.RESULT_KB_DISABLED, + ) - from ..knowledge.config import KnowledgeConfig, KnowledgeStoreMode + from ..knowledge.config import KnowledgeConfig try: - config = getattr(kp, "config", None) or KnowledgeConfig.from_env() + config = getattr(knowledge_plane, "config", None) or KnowledgeConfig.from_env() except ValueError as exc: log.exception("Recipe KB finalize configuration failed (non-fatal)") - return { - "status": "error", - "reason": f"configuration:{type(exc).__name__}", - "backend": "unknown", - "result_type": _close_out.RESULT_CONFIGURATION_FAILED, - "error_class": type(exc).__name__, - } + return _finalize_error( + exc, + reason=f"configuration:{type(exc).__name__}", + backend="unknown", + result_type=_close_out.RESULT_CONFIGURATION_FAILED, + ) from hyperloom.common.perf_metric import agentx_active - if ( - agentx_active(benchmark_mode=self.shared_state.benchmark_mode) - and config.mode is not KnowledgeStoreMode.REMOTE - ): - return { - "status": "skipped", - "reason": "agentx_local_store_unsupported", - "backend": str(getattr(config, "mode", "") or "unknown"), - "result_type": _close_out.RESULT_AGENTX_BLOCKED, - } - if config.mode is KnowledgeStoreMode.REMOTE: - # Remote mode has one Recipe sink: the KB Store final session - # writer. T0 and runtime amendment are intentionally absent. - remote_cid = "" - remote_sid = str( - self.shared_state.recipe_kb_session_id or self.shared_state.session_id or self.session_dir.name + remote = config.mode is KnowledgeStoreMode.REMOTE + if agentx_active(benchmark_mode=self.shared_state.benchmark_mode) and not remote: + return _finalize_outcome( + status="skipped", + reason="agentx_local_store_unsupported", + backend=str(getattr(config, "mode", "") or "unknown"), + result_type=_close_out.RESULT_AGENTX_BLOCKED, ) - from ..knowledge.remote_recipe import HyperloomRemoteKB, KBStoreError - - try: - remote_cid = self.workload_canonical_id() - remote_result = HyperloomRemoteKB.from_env().write( - remote_cid, - self.shared_state, - session_id=remote_sid, - ) - log.info( - "Remote Recipe KB finalize: status=%s reason=%s cid=%s sid=%s", - remote_result.status, - remote_result.reason, - remote_cid, - remote_result.session_id, - ) - self._record_remote_recipe_audit( - source, - RemoteRecipeAudit( - status=remote_result.status, - canonical_id=remote_cid, - session_id=remote_result.session_id, - primary_metric=remote_result.primary_metric, - primary_value=remote_result.primary_value, - reason=remote_result.reason, - ), - ) - return { - "status": remote_result.status, - "reason": remote_result.reason, - "backend": "kb-store", - "canonical_id": str(getattr(remote_result, "canonical_id", "") or remote_cid), - "session_id": str(getattr(remote_result, "session_id", "") or remote_sid), - "result_type": _remote_result_type(remote_result.status, remote_result.reason), - } - except (KBStoreError, OSError, ValueError) as exc: - self._record_remote_recipe_audit( - source, RemoteRecipeAudit(status="error", canonical_id=remote_cid, session_id=remote_sid, error=exc) - ) - log.exception("Remote Recipe KB finalize failed (non-fatal)") - return { - "status": "error", - "reason": type(exc).__name__, - "backend": "kb-store", - "canonical_id": remote_cid, - "session_id": remote_sid, - "result_type": ( + if remote: + return self._publish_remote_recipe(source) + return self._write_local_recipe() + + def _publish_remote_recipe(self, source: str) -> dict[str, Any]: + # Remote mode has one Recipe sink: the KB Store final session writer. + # T0 and runtime amendment are intentionally absent. + state = self.shared_state + session_id = str(state.recipe_kb_session_id or state.session_id or self.session_dir.name) + canonical_id = "" + from ..knowledge.remote_recipe import HyperloomRemoteKB, KBStoreError + + try: + canonical_id = self.workload_canonical_id() + written = HyperloomRemoteKB.from_env().write(canonical_id, state, session_id=session_id) + except (KBStoreError, OSError, ValueError) as exc: + self._record_remote_recipe_audit( + source, + RemoteRecipeAudit( + status="error", + canonical_id=canonical_id, + session_id=session_id, + error=exc, + ), + ) + log.exception("Remote Recipe KB finalize failed (non-fatal)") + return { + **_finalize_error( + exc, + reason=type(exc).__name__, + backend="kb-store", + result_type=( _close_out.RESULT_BUNDLE_BUILD_FAILED if type(exc).__name__ == _REMOTE_VALIDATION_ERROR else _close_out.RESULT_TRANSPORT_FAILED ), - "error_class": type(exc).__name__, - } - - # Local mode never consults ambient KB_STORE_* credentials. - if self.recipe_kb is None: - return { - "status": "skipped", - "reason": "no_recipe_backend", - "backend": "local", - "result_type": _close_out.RESULT_KB_DISABLED, - } - ss = self.shared_state - model_name = ss.model_name or "" - gpu_type = ss.gpu_type or "" - if not model_name or not gpu_type: - log.info( - "recipe KB finalize_recipe: missing model/hardware (model=%r hardware=%r); skipping update_recipe", - model_name, - gpu_type, - ) - return { - "status": "skipped", - "reason": "missing_model_or_hardware", - "backend": "local", - "result_type": _close_out.RESULT_INVALID_SCOPE, + ), + "canonical_id": canonical_id, + "session_id": session_id, } - attrs = self._build_recipe_attrs_from_state() - # Hoist workload tags flat into top-level recipe attrs (shallow-merged) for warm-start filters. - workload_tags = attrs.get("workload") or {} + log.info( + "Remote Recipe KB finalize: status=%s reason=%s cid=%s sid=%s", + written.status, + written.reason, + canonical_id, + written.session_id, + ) + self._record_remote_recipe_audit( + source, + RemoteRecipeAudit( + status=written.status, + canonical_id=canonical_id, + session_id=written.session_id, + primary_metric=written.primary_metric, + primary_value=written.primary_value, + reason=written.reason, + ), + ) + return { + **_finalize_outcome( + status=written.status, + reason=written.reason, + backend="kb-store", + result_type=_remote_result_type(written.status, written.reason), + ), + "canonical_id": str(getattr(written, "canonical_id", "") or canonical_id), + "session_id": str(getattr(written, "session_id", "") or session_id), + } - # sessions[] read-modify-write: read anchor, drop prior entry with our session_id (resume safety), append ours, write back. + def _final_recipe_overrides(self) -> dict[str, Any]: + # Read-modify-write under the cid flock, so two finalises of the same + # session leave one row rather than tearing the lists they both append to. + state = self.shared_state + attrs = recipe_attrs_from_state(state, self.session_dir.name) + live = self.recipe_kb.get_authoritative_recipe(canonical_id=self.workload_canonical_id()) or {} my_sessions = list(attrs["sessions"] or []) - my_session_ids = {str((s or {}).get("session_id") or "") for s in my_sessions if isinstance(s, dict)} - # v2: read-modify-write the recipe row; sessions[] merged in-process under the cid flock so concurrent finalises don't tear. - merged_sessions: list[dict[str, Any]] = list(my_sessions) - existing_row: dict[str, Any] = {} - if self.recipe_kb is not None: - cid = self.workload_canonical_id() - # Read exactly the local store's authority row. - existing_row = self.recipe_kb.get_authoritative_recipe(canonical_id=cid) or {} - existing_sessions: list[dict[str, Any]] = [] - for row in existing_row.get("sessions") or []: - if not isinstance(row, dict): - continue - if str(row.get("session_id") or "") in my_session_ids: - # Resume/retry of the same session — our new entry supersedes the prior one. - continue - existing_sessions.append(dict(row)) - merged_sessions = existing_sessions + my_sessions - + extras = dict(attrs.get("workload") or {}) overrides: dict[str, Any] = { "what_worked": attrs["what_worked"], "what_failed": attrs["what_failed"], "last_profiled": attrs["last_profiled"], - "sessions": merged_sessions, - "extras": dict(workload_tags or {}), + "sessions": rows_not_superseded(live.get("sessions"), my_sessions, "session_id") + my_sessions, + "extras": extras, } - # Overwrite best_config/best_throughput only on a real improvement: requires has_validated_win AND my_tput > live_tput. - my_tput = float(attrs.get("best_throughput") or 0.0) - cb_args_now = str(ss.current_best.get("extra_server_args") or "").strip() - validated_gain = float(ss.cumulative_gain_validated or 0.0) - has_validated_win = bool((ss.optimization_stack or []) or validated_gain > 0.0 or cb_args_now) - try: - live_tput = float(existing_row.get("best_throughput") or 0.0) - except (TypeError, ValueError): - live_tput = 0.0 - if has_validated_win and my_tput > live_tput: + # The row's own figures survive a session that never validated a win, + # or whose win does not beat what the row already advertises. + my_throughput = float(attrs.get("best_throughput") or 0.0) + won = bool(state.optimization_stack or float(state.cumulative_gain_validated or 0.0) > 0.0) + won = won or bool(str(state.current_best.get("extra_server_args") or "").strip()) + if won and my_throughput > recorded_float(live.get("best_throughput")): overrides["best_config"] = attrs["best_config"] - overrides["best_throughput"] = my_tput + overrides["best_throughput"] = my_throughput + return overrides + + def _write_local_recipe(self) -> dict[str, Any]: + # Local mode never consults ambient KB_STORE_* credentials. + if self.recipe_kb is None: + return _finalize_outcome( + status="skipped", + reason="no_recipe_backend", + backend="local", + result_type=_close_out.RESULT_KB_DISABLED, + ) + state = self.shared_state + model_name = state.model_name or "" + gpu_type = state.gpu_type or "" + if not model_name or not gpu_type: + log.info( + "recipe KB finalize_recipe: missing model/hardware (model=%r hardware=%r); skipping update_recipe", + model_name, + gpu_type, + ) + return _finalize_outcome( + status="skipped", + reason="missing_model_or_hardware", + backend="local", + result_type=_close_out.RESULT_INVALID_SCOPE, + ) self.kb_amend_recipe( - recipe_overrides=overrides, + recipe_overrides=self._final_recipe_overrides(), provenance_details={ "phase": "close_finalize", "evidence": [ - f"log:session-{ss.recipe_kb_session_id or self.session_dir.name}", + f"log:session-{state.recipe_kb_session_id or self.session_dir.name}", ], }, ) - return { - "status": "written", - "reason": "", - "backend": "local", - "result_type": _close_out.RESULT_WRITTEN, - } + return _finalize_outcome( + status="written", + reason="", + backend="local", + result_type=_close_out.RESULT_WRITTEN, + ) def _kb_hardware_slug(self) -> str: """Topology-aware hardware dimension for the recipe ``canonical_id``.""" @@ -1433,27 +1131,19 @@ def _kb_hardware_slug(self) -> str: def workload_canonical_id(self) -> str: """Return the workload's canonical seven-dimension Recipe identity.""" - ss = self.shared_state - workload = ss.model_name or "unknown_model" - hw = self._kb_hardware_slug() - framework = str(ss.framework or "") - framework_version = str(ss.framework_version or "") - if not framework_version and framework: - framework_version = detect_framework_version(framework) - precision = str(ss.precision or "") - model_type = str(ss.model_type or "") - architectures = ss.model_architectures or [] + state = self.shared_state + identity = _framework_identity(state) from hyperloom.common.perf_metric import agentx_active return recipe_canonical_id( - model=workload, - hardware=hw, - framework_name=framework, - framework_version=framework_version, - precision=precision, - model_type=model_type, - architectures=architectures, - scheme=("agentx" if agentx_active(benchmark_mode=ss.benchmark_mode) else "inference"), + model=state.model_name or "unknown_model", + hardware=self._kb_hardware_slug(), + framework_name=identity.name, + framework_version=identity.version, + precision=identity.precision, + model_type=str(state.model_type or ""), + architectures=state.model_architectures or [], + scheme=("agentx" if agentx_active(benchmark_mode=state.benchmark_mode) else "inference"), ) def read_local_recipe_row(self) -> dict[str, Any]: @@ -1462,14 +1152,9 @@ def read_local_recipe_row(self) -> dict[str, Any]: return {} tick = int(self.shared_state.tick or 0) cache = self._local_recipe_cache - if isinstance(cache, tuple) and len(cache) == 2 and cache[0] == tick: + if cache is not None and cache[0] == tick: return cache[1] - row = ( - self.recipe_kb.get_authoritative_recipe( - canonical_id=self.workload_canonical_id(), - ) - or {} - ) + row = self.recipe_kb.get_authoritative_recipe(canonical_id=self.workload_canonical_id()) or {} self._local_recipe_cache = (tick, row) return row @@ -1521,15 +1206,8 @@ def kb_best_config_overrides_for_keep( _extra_server_args(live_bc).strip() or (isinstance(live_bc.get("extra_envs"), Mapping) and live_bc.get("extra_envs")) ) - try: - live_tput = float(live.get("best_throughput") or 0.0) - except (TypeError, ValueError): - live_tput = 0.0 - try: - new_tput = float(throughput_after or 0.0) - except (TypeError, ValueError): - new_tput = 0.0 - + live_tput = recorded_float(live.get("best_throughput")) + new_tput = recorded_float(throughput_after) if not live_has_config or (new_tput > 0.0 and new_tput >= live_tput): overrides: dict[str, Any] = { "best_config": dict(best_config_candidate), @@ -1547,125 +1225,45 @@ def kb_amend_recipe( recipe_overrides: dict[str, Any] | None = None, provenance_details: dict[str, Any] | None = None, ) -> None: - """Read-modify-write helper for the recipe-snapshot KB: load live row, append lesson/pitfall, merge recipe_overrides (unset fields preserved), write back. Read or write failures propagate; lesson/pitfall appended without dedup.""" + """Read-modify-write the local recipe row: append the fact, merge the overrides, write it back. + + Read or write failures propagate. A lesson or pitfall is appended + without dedup, and a field the caller did not override keeps the value + the live row carries. + """ config = getattr(self.knowledge_plane, "config", None) - if getattr(getattr(config, "mode", None), "value", None) == "remote" or self.recipe_kb is None: + mode = getattr(getattr(config, "mode", None), "value", None) + if mode == KnowledgeStoreMode.REMOTE.value or self.recipe_kb is None: return from hyperloom.common.perf_metric import agentx_active - if agentx_active(benchmark_mode=self.shared_state.benchmark_mode): + state = self.shared_state + if agentx_active(benchmark_mode=state.benchmark_mode): return - cid = self.workload_canonical_id() - - ss = self.shared_state - framework = str(ss.framework or "") - framework_version = str(ss.framework_version or "") - if not framework_version and framework: - framework_version = detect_framework_version(framework) - precision = str(ss.precision or "") - - # Local mode reads the exact authority row before amending it. - live = self.recipe_kb.get_authoritative_recipe(canonical_id=cid) or {} - - lessons = list(live.get("lessons") or []) - if append_lesson is not None: - lessons.append(append_lesson) - pitfalls = list(live.get("pitfalls") or []) - if append_pitfall is not None: - pitfalls.append(append_pitfall) - - # Build put_recipe kwargs, preserving live fields the caller didn't override. + canonical_id = self.workload_canonical_id() + identity = _framework_identity(state) + live = self.recipe_kb.get_authoritative_recipe(canonical_id=canonical_id) or {} overrides = dict(recipe_overrides or {}) - _reserved = { - "canonical_id", - "version", - "created_at", - "updated_at", - "model", - "hardware", - "framework", - "framework_name", - "framework_version", - "precision", - "best_config", - "best_throughput", - "what_worked", - "what_failed", - "remaining_gaps", - "pitfalls", - "lessons", - "last_profiled", - "stack_fingerprint", - "sessions", - "authority", - "confidence", - "evidence_refs", - "provenance", - } - prior_extras = {k: v for k, v in live.items() if k not in _reserved} - merged_extras = {**prior_extras, **(overrides.get("extras") or {})} - # Re-stamp config.json architecture-identity tags; skipped when unset. - _arch = ss.model_architectures or [] - if isinstance(_arch, list): - _arch_list = [str(a).strip() for a in _arch if str(a or "").strip()] - if _arch_list: - merged_extras["architectures"] = _arch_list - _mtype = str(ss.model_type or "").strip() - if _mtype: - merged_extras["model_type"] = _mtype put_kwargs: dict[str, Any] = { - "canonical_id": cid, - "model": ss.model_name or "unknown_model", + "canonical_id": canonical_id, + "model": state.model_name or "unknown_model", "hardware": self._kb_hardware_slug(), - "framework_name": framework, - "framework_version": framework_version, - "precision": precision, - "best_config": overrides.get("best_config") - if "best_config" in overrides - else dict(live.get("best_config") or {}), - "best_throughput": overrides.get("best_throughput") - if "best_throughput" in overrides - else float(live.get("best_throughput") or 0.0), - "what_worked": overrides.get("what_worked") - if "what_worked" in overrides - else list(live.get("what_worked") or []), - "what_failed": overrides.get("what_failed") - if "what_failed" in overrides - else list(live.get("what_failed") or []), - "remaining_gaps": overrides.get("remaining_gaps") - if "remaining_gaps" in overrides - else list(live.get("remaining_gaps") or []), - "pitfalls": pitfalls, - "lessons": lessons, - "last_profiled": overrides.get("last_profiled") - if "last_profiled" in overrides - else str(live.get("last_profiled") or ""), - "stack_fingerprint": overrides.get("stack_fingerprint") - if "stack_fingerprint" in overrides - else dict(live.get("stack_fingerprint") or {}), - "sessions": overrides.get("sessions") if "sessions" in overrides else list(live.get("sessions") or []), - "extras": merged_extras, - # Preserve audit fields across the amend (else put_recipe resets them to defaults). - "authority": overrides.get("authority") - if "authority" in overrides - else str(live.get("authority") or "EXPERIENTIAL"), - "confidence": overrides.get("confidence") - if "confidence" in overrides - else float(live.get("confidence") or 0.85), - "evidence_refs": overrides.get("evidence_refs") - if "evidence_refs" in overrides - else list(live.get("evidence_refs") or []), + "framework_name": identity.name, + "framework_version": identity.version, + "precision": identity.precision, + "pitfalls": appended(live.get("pitfalls"), append_pitfall), + "lessons": appended(live.get("lessons"), append_lesson), + "extras": amended_recipe_extras(live, overrides, state), "provenance": { "source": "hyperloom-inference-optimizer", "generator": "coordinator", - "generated_at": datetime.now(timezone.utc).isoformat( - timespec="microseconds", - ), + "generated_at": datetime.now(timezone.utc).isoformat(timespec="microseconds"), "details": dict(provenance_details or {}), }, + **amended_recipe_fields(live, overrides), } self.recipe_kb.put_recipe(**put_kwargs) self._local_recipe_cache = None -__all__ = ["RecipeJournalCollaborator"] +__all__ = ["RecipeJournalCollaborator", "attempt_provenance", "recipe_attrs_from_state", "workload_kb_tags"]