fix(collector): Preserve integrate patch proposal attribution

src/hyperloom/inference_optimizer/breakdown/collectors/attribution.py

@@ -148,13 +148,14 @@ def _phase_at(ts_unix: float | None, timeline: list[tuple[float, str]]) -> str:
     return current
 
 
-def _entry_family(entry: dict[str, Any], *, inferred_phase: str = "") -> str:
+def _entry_family(entry: dict[str, Any]) -> str:
     """Resolve attribution family using phase/ownership metadata when needed.
 
     ``integrate_patch`` is not intrinsically a Framework action. Framework
     authoring owns explicitly marked or FRAMEWORK_AGENT entries; EXPLORE owns
     its own patch applications; PRELUDE baseline-enablement entries are
-    configuration prerequisites and remain non-attributable.
+    configuration prerequisites and remain non-attributable. Missing ownership
+    stays unattributed instead of being inferred from execution phase.
     """
 
     action = str(entry.get("action") or "").strip().lower()
@@ -164,16 +165,16 @@ def _entry_family(entry: dict[str, Any], *, inferred_phase: str = "") -> str:
         return "unattributed"
     if entry.get("framework_agent_authoring"):
         return "framework"
-    phase = str(
-        entry.get("source_phase") or entry.get("phase") or inferred_phase or ""
-    ).strip().upper()
-    return {
-        "FRAMEWORK": "framework",
-        "FRAMEWORK_AGENT": "framework",
-        "EXPLORE": "explore",
-        "KERNEL": "kernel_agent",
-        "KERNEL_AGENT": "kernel_agent",
-    }.get(phase, "unattributed")
+    phase = str(entry.get("source_phase") or entry.get("phase") or "").strip().upper()
+    provenance = str(entry.get("provenance") or "").strip().lower()
+    specialist_owned = bool(entry.get("domain")) or provenance.startswith(
+        "specialist:"
+    )
+    if phase in {"FRAMEWORK", "FRAMEWORK_AGENT"}:
+        return "framework"
+    if phase == "EXPLORE" or specialist_owned:
+        return "explore"
+    return "unattributed"
 
 
 def _promote_legacy_gain_entries(
@@ -226,6 +227,8 @@ def _promote_legacy_gain_entries(
         for key in (
             "source_phase",
             "phase",
+            "domain",
+            "gap_layer",
             "framework_agent_authoring",
             "baseline_enablement",
             "attribution_eligible",
@@ -308,7 +311,6 @@ def collect_attribution(
         "gemm_tuning": 0.0,
         "geak": 0.0,
     }
-    timeline = _phase_timeline(state)
     unattributed_actions: set[str] = set()
     for e in entries:
         if not isinstance(e, dict):
@@ -318,10 +320,7 @@ def collect_attribution(
         delta = _to_float(e.get("delta_pct"))
         if delta is None:
             continue
-        fam = _entry_family(
-            e,
-            inferred_phase=_phase_at(_entry_ts(e), timeline),
-        )
+        fam = _entry_family(e)
         family_totals[fam] = family_totals.get(fam, 0.0) + max(delta, 0.0)
         if fam in {"other", "unattributed"} and delta > 0:
             unattributed_actions.add(str(e.get("action") or "<missing>"))
@@ -403,7 +402,8 @@ def _collect_phase_breakdown(
     """Per-phase gain attribution.
 
     Assigns each KEEP entry to the phase active at its acceptance
-    timestamp (explore further splits by domain, kernel by kernel_id).
+    timestamp (explore further splits by domain, kernel by kernel_id), except
+    ``integrate_patch`` which follows explicit proposal ownership.
     Missing phase_history → everything lands under ``unattributed``.
 
     Args:
@@ -467,7 +467,16 @@ def _collect_phase_breakdown(
         if phase == "framework_agent":
             phase = "framework"
         action = str(e.get("action") or "").lower()
-        fam = _entry_family(e, inferred_phase=phase)
+        fam = _entry_family(e)
+        if action.startswith("integrate_patch"):
+            # For this delayed application mechanism, proposal ownership is the
+            # attribution phase; the acceptance timestamp is only execution
+            # context and must not manufacture a kernel_agent/"?" row.
+            phase = (
+                fam
+                if fam in {"framework", "explore"}
+                else "unattributed"
+            )
         # gemm_tuning runs inside KERNEL but is bucketed separately.
         if fam == "gemm_tuning":
             phase = "gemm_tuning"
@@ -491,7 +500,13 @@ def _collect_phase_breakdown(
         if phase == "explore":
             by_domain = bucket.setdefault("by_domain", {})
             fp = str(e.get("fingerprint") or e.get("variant_fingerprint") or "")
-            raw_prov = provenance_by_fp.get(fp) or str(e.get("provenance") or "") or "default_grid"
+            entry_domain = str(e.get("domain") or "").strip()
+            raw_prov = (
+                provenance_by_fp.get(fp)
+                or str(e.get("provenance") or "")
+                or (f"specialist:{entry_domain}" if entry_domain else "")
+                or "default_grid"
+            )
             domain = _normalize_specialist_key(raw_prov)
             by_domain[domain] = round(
                 float(by_domain.get(domain, 0.0)) + float(delta),

src/hyperloom/inference_optimizer/breakdown/collectors/optimizations.py

@@ -149,19 +149,41 @@ def _resolve_source(
         raw.get("tuned_file"),
         raw.get("final_overlay"),
     )
-    if (
+    if not action.startswith("integrate_patch") and (
         any(value not in (None, "", [], {}) for value in kernel_markers)
         or action in {"geak_e2e", "gemm_tuning", "fusion", "integrate"}
         or action.startswith("kernel_opt")
     ):
         return "kernel_agent", "action_family"
+    if action in {"framework", "framework_agent"}:
+        return "framework_agent", "action_family"
 
     explicit = _normalized_phase(
         raw.get("source_phase")
         or gain.get("source_phase")
         or raw.get("phase")
         or gain.get("phase")
     )
+    if action.startswith("integrate_patch"):
+        if raw.get("framework_agent_authoring") or gain.get(
+            "framework_agent_authoring"
+        ):
+            return "framework_agent", "recorded"
+        provenance = str(
+            raw.get("provenance") or gain.get("provenance") or ""
+        ).strip().lower()
+        specialist_owned = bool(
+            raw.get("domain")
+            or gain.get("domain")
+            or provenance.startswith("specialist:")
+        )
+        if explicit in {"framework_agent", "explore"}:
+            return explicit, "recorded"
+        if specialist_owned:
+            return "explore", "recorded"
+        # integrate_patch ownership must be explicit. Never infer it from the
+        # phase active when the delayed application happened.
+        return "unattributed", "unknown"
     if explicit:
         return explicit, "recorded"
 
@@ -174,8 +196,6 @@ def _resolve_source(
 
     if action in {"explore", "backends", "params"}:
         return "explore", "action_family"
-    if action == "framework_agent":
-        return "framework_agent", "action_family"
     return "unattributed", "unknown"
 
 
@@ -220,6 +240,7 @@ def _artifacts(raw: dict[str, Any]) -> list[dict[str, str]]:
         ("report", "final_report_path"),
         ("report", "report_path"),
         ("overlay", "final_overlay"),
+        ("source_manifest", "source_manifest"),
     )
     out = [
         {"kind": str(item.get("kind") or ""), "path": str(item.get("path") or "")}
@@ -235,6 +256,15 @@ def _artifacts(raw: dict[str, Any]) -> list[dict[str, str]]:
             continue
         seen.add(key)
         out.append({"kind": kind, "path": path})
+    target_files = raw.get("target_files") or []
+    if isinstance(target_files, list):
+        for value in target_files:
+            path = str(value or "").strip()
+            key = ("target_file", path)
+            if not path or key in seen:
+                continue
+            seen.add(key)
+            out.append({"kind": "target_file", "path": path})
     return out
 
 

src/hyperloom/inference_optimizer/tests/test_coordinator_async_batch2_unit.py

@@ -225,6 +225,14 @@ async def test_resume_consistency_replays_orphaned_integrate_keep(coord: Coordin
                     "status": "kept",
                     "specialist_task_id": "spec-orphan",
                     "output_throughput": 123.0,
+                    "source_phase": "FRAMEWORK_AGENT",
+                    "domain": "serving_specialist",
+                    "provenance": "specialist:serving_specialist",
+                    "framework_agent_authoring": True,
+                    "source_manifest": (
+                        "/session/optimization_stack/src/spec-orphan/manifest.json"
+                    ),
+                    "target_files": ["vllm/model.py"],
                 },
             },
         )
@@ -237,6 +245,16 @@ async def test_resume_consistency_replays_orphaned_integrate_keep(coord: Coordin
     assert replay["variant"] == "spec-orphan"
     assert coord.shared_state.optimization_stack[-1]["action"] == "integrate_patch"
     assert coord.shared_state.optimization_stack[-1]["variant_name"] == "spec-orphan"
+    assert coord.shared_state.optimization_stack[-1]["source_phase"] == "FRAMEWORK_AGENT"
+    assert coord.shared_state.optimization_stack[-1]["provenance"] == (
+        "specialist:serving_specialist"
+    )
+    assert coord.shared_state.optimization_stack[-1]["source_manifest"] == (
+        "/session/optimization_stack/src/spec-orphan/manifest.json"
+    )
+    assert coord.shared_state.optimization_stack[-1]["target_files"] == [
+        "vllm/model.py"
+    ]
     assert coord.shared_state.resume_pending_revalidation is True
 
 
@@ -269,6 +287,9 @@ async def test_resume_consistency_replays_pending_integrate_keep(coord: Coordina
     assert replay["appended"] is True
     assert coord.shared_state.pending_integrate == {}
     assert coord.shared_state.optimization_stack[-1]["variant_name"] == "spec-pending"
+    assert "source_phase" not in coord.shared_state.optimization_stack[-1]
+    assert "domain" not in coord.shared_state.optimization_stack[-1]
+    assert "framework_agent_authoring" not in coord.shared_state.optimization_stack[-1]
     assert coord.shared_state.resume_pending_revalidation is True
 
 

src/hyperloom/inference_optimizer/tests/test_coordinator_async_methods_coverage_unit.py

@@ -479,7 +479,20 @@ async def test_autosubmit_creates_proposal_for_real_file(coord: Coordinator) ->
     wt = spec_root / "worktree"
     wt.mkdir(parents=True, exist_ok=True)
     (wt / "kernel.py").write_text("# patched\n", encoding="utf-8")
-    task = Task(task_id=sid, kind="specialist", state="running", params={}, idempotency_key="k3")
+    coord.shared_state.phase = "KERNEL_AGENT"
+    task = Task(
+        task_id=sid,
+        kind="specialist",
+        state="running",
+        params={
+            "domain": "serving_specialist",
+            "gap_canonical_id": "gap.framework.fp8",
+            "gap_layer": "framework",
+            "framework": "other-framework",
+            "framework_agent_authoring": True,
+        },
+        idempotency_key="k3",
+    )
     n_before = len(coord.state.pending_proposals)
     await coord._maybe_autosubmit_specialist_patches(
         task=task,
@@ -489,6 +502,13 @@ async def test_autosubmit_creates_proposal_for_real_file(coord: Coordinator) ->
         },
     )
     assert len(coord.state.pending_proposals) == n_before + 1
+    pending = list(coord.state.pending_proposals.values())[-1]
+    params = pending.payload["params"]
+    assert params["source_phase"] == "FRAMEWORK_AGENT"
+    assert params["domain"] == "serving_specialist"
+    assert params["provenance"] == "specialist:serving_specialist"
+    assert params["gap_canonical_id"] == "gap.framework.fp8"
+    assert "framework" not in params
 
 
 @pytest.mark.asyncio

src/hyperloom/inference_optimizer/tests/test_promote_shared_state_lock.py

@@ -406,6 +406,48 @@ async def test_promote_integrate_patch_kept_lifts_and_clears_pending(session_dir
     assert not hasattr(s, "last_integrate_patch")
 
 
+@pytest.mark.asyncio
+async def test_integrate_patch_preserves_proposal_owner_across_phase_change(
+    session_dir,
+):
+    coord = _coord(session_dir)
+    s = coord.shared_state
+    s.baseline_tput = 100.0
+    s.phase = "KERNEL_AGENT"
+
+    await coord._promote_to_shared_state(
+        "integrate_patch",
+        {
+            "status": "kept",
+            "output_throughput": 110.0,
+            "specialist_task_id": "spec-framework",
+            "delta_pct": 10.0,
+            "extra_server_args_applied": "--quantization fp8_per_channel",
+            "workspace": "/w",
+        },
+        task=_task(
+            "integrate_patch",
+            task_id="t-cross-phase",
+            params={
+                "specialist_task_id": "spec-framework",
+                "source_phase": "FRAMEWORK_AGENT",
+                "domain": "serving_specialist",
+                "provenance": "specialist:serving_specialist",
+                "gap_canonical_id": "gap.framework.fp8",
+                "gap_layer": "framework",
+                "framework_agent_authoring": True,
+            },
+        ),
+    )
+
+    entry = s.optimization_stack[0]
+    assert entry["source_phase"] == "FRAMEWORK_AGENT"
+    assert entry["domain"] == "serving_specialist"
+    assert entry["provenance"] == "specialist:serving_specialist"
+    assert entry["gap_canonical_id"] == "gap.framework.fp8"
+    assert entry["framework_agent_authoring"] is True
+
+
 @pytest.mark.asyncio
 async def test_prebaseline_enablement_patch_is_config_only_not_gain(session_dir):
     """A patch required to establish baseline stays reproducible but has no gain."""
@@ -473,6 +515,7 @@ async def test_promote_framework_agent_kept_lifts_and_records_progress(session_d
     coord = _coord(session_dir)
     s = coord.shared_state
     s.baseline_tput = 100.0
+    s.phase = "KERNEL_AGENT"
 
     await coord._promote_to_shared_state(
         "framework_agent",
@@ -496,6 +539,8 @@ async def test_promote_framework_agent_kept_lifts_and_records_progress(session_d
     assert row["batch_id"] == "b1"
     assert s.current_best["action"] == "framework"
     assert s.current_best["tput"] == 130.0
+    assert s.optimization_stack[-1]["source_phase"] == "FRAMEWORK_AGENT"
+    assert s.optimization_stack[-1]["provenance"] == "framework_agent"
     assert not hasattr(s, "last_framework_agent")
 
 
@@ -731,7 +776,7 @@ def test_lift_applies_unset_envs_before_new_envs(session_dir):
 
 @pytest.mark.asyncio
 async def test_lift_copies_source_snapshot_into_stack_entry(session_dir):
-    """source_snapshot/framework_root/base_sha from the lift bv reach the stack entry."""
+    """Source snapshot manifest and changed files reach the stack entry."""
     coord = _coord(session_dir)
     s = coord.shared_state
     s.baseline_tput = 1000.0
@@ -747,13 +792,21 @@ async def test_lift_copies_source_snapshot_into_stack_entry(session_dir):
             "tput": 1500.0,
             "scope": "source_patch",
             "source_snapshot": "/session/optimization_stack/src/abc123",
+            "source_manifest": "/session/optimization_stack/src/abc123/manifest.json",
+            "target_files": ["vllm/model_executor/layers/quantization/foo.py"],
             "framework_root": "/opt/vllm",
             "base_sha": "deadbeef",
         },
     )
 
     top = s.optimization_stack[-1]
     assert top.get("source_snapshot") == "/session/optimization_stack/src/abc123"
+    assert top.get("source_manifest") == (
+        "/session/optimization_stack/src/abc123/manifest.json"
+    )
+    assert top.get("target_files") == [
+        "vllm/model_executor/layers/quantization/foo.py"
+    ]
     assert top.get("framework_root") == "/opt/vllm"
     assert top.get("base_sha") == "deadbeef"