feat(pipeline): finalize on best iteration, not last

agents/jack_manager.py

@@ -235,21 +235,30 @@ def write_retune(state: ManagerState) -> dict:
     return {"decision_log": [f"iteration {state['iteration']}: wrote retune_request.json"]}
 
 
-def proceed(state: ManagerState) -> dict:
-    """sample_for_explanation.csv (Handoff 4) — drawn from predictions_test.csv.
-    Terminal: hands off to Freddi and waits for explanations.csv."""
+def write_sample(predictions_path: str, sample_size: int = 300) -> int:
+    """Write sample_for_explanation.csv (Handoff 4) from a predictions file.
+    Shared by the proceed node and the pipeline's best-iteration restore, so the
+    sample format has exactly one definition."""
     import pandas as pd
 
-    preds = pd.read_csv(state.get("predictions_path", "mock_data/predictions_test.csv"))
+    preds = pd.read_csv(predictions_path)
     sample = (preds[["article_id", "article_title", "predicted_label", "label",
                      "confidence", "prob_up", "prob_down", "prob_neutral"]]
               .rename(columns={"label": "actual_label"}))
-    n = min(len(sample), state.get("sample_size", 300))
+    n = min(len(sample), sample_size)
     if n < len(sample):                                   # only subsample when needed
         sample = sample.sample(n=n, random_state=42)      # representative + reproducible
-    _write_decision(state)
     os.makedirs(OUTPUT_DIR, exist_ok=True)
     sample.to_csv(os.path.join(OUTPUT_DIR, "sample_for_explanation.csv"), index=False)
+    return n
+
+
+def proceed(state: ManagerState) -> dict:
+    """sample_for_explanation.csv (Handoff 4) — drawn from predictions_test.csv.
+    Terminal: hands off to Freddi and waits for explanations.csv."""
+    _write_decision(state)
+    n = write_sample(state.get("predictions_path", "mock_data/predictions_test.csv"),
+                     state.get("sample_size", 300))
     return {"decision_log": [f"iteration {state['iteration']}: wrote sample_for_explanation.csv ({n} rows)"]}
 
 

agents/pipeline_graph.py

@@ -20,7 +20,12 @@
 ## Graph shape
 
     START → process → classify → evaluate → gate ─(retune)→ classify   [CYCLE]
-                                              └(proceed)→ explain → finalize → END
+                                              └(proceed)→ select_best → explain → finalize → END
+
+`evaluate` snapshots each new-best iteration's artifacts into `outputs/best/`;
+`select_best` restores that snapshot (and redraws the explanation sample) when the
+loop's LAST iteration wasn't its best — accuracy can regress across retunes, and
+without this the pipeline would finalize on the regressed predictions.
 
 `gate` is the Manager: calling it writes either `retune_request.json` (retune) or
 `sample_for_explanation.csv` (proceed). Its own checkpointer carries the iteration
@@ -36,7 +41,9 @@
 
 from __future__ import annotations
 
+import json
 import os
+import shutil
 from dataclasses import dataclass
 
 from typing_extensions import TypedDict
@@ -46,13 +53,13 @@
     from agents.nadi_classifier import ClassifierAgent
     from agents.sabina_evaluator import EvaluatorAgent
     from agents.freddi_explanation import ExplanationAgent
-    from agents.jack_manager import ManagerAgent, OUTPUT_DIR as OUT
+    from agents.jack_manager import ManagerAgent, write_sample, OUTPUT_DIR as OUT
 except ModuleNotFoundError:  # running as a bare script with agents/ on sys.path
     from aurora_processing import ProcessingAgent
     from nadi_classifier import ClassifierAgent
     from sabina_evaluator import EvaluatorAgent
     from freddi_explanation import ExplanationAgent
-    from jack_manager import ManagerAgent, OUTPUT_DIR as OUT
+    from jack_manager import ManagerAgent, write_sample, OUTPUT_DIR as OUT
 
 # Contract files exchanged between the nodes. All live under the Manager's
 # OUTPUT_DIR except processed_data.csv, whose path Aurora returns at runtime.
@@ -63,6 +70,10 @@
 RETUNE = os.path.join(OUT, "retune_request.json")
 EXPL = os.path.join(OUT, "explanations.csv")
 
+# Snapshot dir for the best iteration's artifacts (internal to the loop, not a
+# contract handoff). Holds copies of CODE/PREDS/EVAL from the highest-accuracy pass.
+BEST_DIR = os.path.join(OUT, "best")
+
 # A retune cycle is 3 nodes (classify → evaluate → gate); with the process/explain/
 # finalize tail, ~5 iterations stays well under this. LangGraph aborts a runaway
 # cycle at recursion_limit, so we set headroom rather than rely on the default 25.
@@ -78,6 +89,8 @@ class PipelineState(TypedDict, total=False):
     retune_request_path: str | None  # None on the first pass; RETUNE on cycle passes
     final_action: str               # the gate's verdict: "retune" | "proceed"
     iteration: int                  # Manager's loop counter (for reporting)
+    last_accuracy: float            # accuracy of the most recent evaluate pass
+    best_accuracy: float            # best accuracy seen across cycle passes
 
 
 @dataclass
@@ -136,9 +149,19 @@ def classify(state: PipelineState) -> dict:
         return {}
 
     def evaluate(state: PipelineState) -> dict:
-        """Sabina: score the predictions and write evaluation_report.json."""
+        """Sabina: score the predictions and write evaluation_report.json. Also
+        snapshots this pass's artifacts into BEST_DIR whenever accuracy sets a new
+        best, so `select_best` can restore them if later retunes regress."""
         agents.sabina.run(predictions=PREDS, classifier_code=CODE)
-        return {}
+        with open(EVAL, encoding="utf-8") as f:
+            acc = json.load(f)["accuracy"]
+        out = {"last_accuracy": acc}
+        if acc > state.get("best_accuracy", -1.0):
+            os.makedirs(BEST_DIR, exist_ok=True)
+            for path in (PREDS, EVAL, CODE):
+                shutil.copy2(path, os.path.join(BEST_DIR, os.path.basename(path)))
+            out["best_accuracy"] = acc
+        return out
 
     def gate(state: PipelineState) -> dict:
         """Manager gate. Invoking it applies the accuracy gate (with convergence +
@@ -152,6 +175,24 @@ def gate(state: PipelineState) -> dict:
             out["retune_request_path"] = RETUNE  # fed back into `classify` on the cycle
         return out
 
+    def select_best(state: PipelineState) -> dict:
+        """The gate proceeded with the LAST iteration's artifacts, which are not
+        necessarily the best ones (accuracy can regress across retunes). If an
+        earlier pass scored higher, restore its snapshot over the canonical paths
+        and redraw the explanation sample from the restored predictions, so
+        explain/finalize run on the best iteration. Gate semantics are untouched:
+        the Manager already made its verdict from the last iteration's report."""
+        best, last = state.get("best_accuracy", -1.0), state.get("last_accuracy", -1.0)
+        if best <= last:
+            return {}
+        for path in (PREDS, EVAL, CODE):
+            shutil.copy2(os.path.join(BEST_DIR, os.path.basename(path)), path)
+        sample_size = getattr(agents.manager, "_defaults", {}).get("sample_size", 300)
+        write_sample(PREDS, sample_size)
+        print(f"[pipeline] last iteration regressed ({last:.2f}) — restored best "
+              f"iteration's artifacts ({best:.2f}) for explanation + finals")
+        return {}
+
     def explain(state: PipelineState) -> dict:
         """Freddi: justify each sampled prediction into explanations.csv."""
         agents.freddi.run(sample_for_explanation=SAMPLE, output=EXPL)
@@ -170,13 +211,15 @@ def route(state: PipelineState) -> str:
 
     b = StateGraph(PipelineState)
     for name, fn in [("process", process), ("classify", classify), ("evaluate", evaluate),
-                     ("gate", gate), ("explain", explain), ("finalize", finalize)]:
+                     ("gate", gate), ("select_best", select_best), ("explain", explain),
+                     ("finalize", finalize)]:
         b.add_node(name, fn)
     b.add_edge(START, "process")
     b.add_edge("process", "classify")
     b.add_edge("classify", "evaluate")
     b.add_edge("evaluate", "gate")
-    b.add_conditional_edges("gate", route, {"retune": "classify", "proceed": "explain"})
+    b.add_conditional_edges("gate", route, {"retune": "classify", "proceed": "select_best"})
+    b.add_edge("select_best", "explain")
     b.add_edge("explain", "finalize")
     b.add_edge("finalize", END)
     return b.compile(checkpointer=checkpointer)

tests/test_pipeline_graph.py

@@ -66,6 +66,88 @@ def run(self, *, predictions, classifier_code):
         return {"output_path": "outputs/evaluation_report.json"}
 
 
+class MarkedNadi:
+    """FakeNadi variant that stamps every predictions file with the pass number,
+    so the test can tell WHICH iteration's artifacts survived to the end."""
+    def __init__(self):
+        self.calls = 0
+
+    def run(self, *, processed_data, classifier_code, predictions, retune_request=None):
+        self.calls += 1
+        Path(predictions).parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy(MOCK_PRED, predictions)
+        import pandas as pd
+        df = pd.read_csv(predictions)
+        df["fake_pass"] = self.calls
+        df.to_csv(predictions, index=False)
+        Path(classifier_code).write_text(f"THRESHOLD = 0.5  # pass {self.calls}\n")
+        return {}
+
+
+class PeakSabina:
+    """Accuracy peaks on pass 2 then regresses, so the best iteration is NOT the
+    last one the gate proceeds with — exactly the case select_best must fix."""
+    ACCS = [0.20, 0.39, 0.30, 0.30, 0.30, 0.30, 0.30, 0.30]
+
+    def __init__(self):
+        self.calls = 0
+
+    def run(self, *, predictions, classifier_code):
+        acc = self.ACCS[self.calls]
+        self.calls += 1
+        report = {
+            "accuracy": acc,
+            "below_threshold": True,
+            "class_accuracy": {"up": 0.3, "down": 0.2, "neutral": 0.5},
+            "misclassified_ids": [],
+            "proposal": {"recommended_action": "retune", "reason": "low",
+                         "focus_labels": ["down"],
+                         "suggested_params": {"threshold": 0.5, "max_length": 128},
+                         "code_notes": ""},
+        }
+        os.makedirs("outputs", exist_ok=True)
+        Path("outputs/evaluation_report.json").write_text(json.dumps(report))
+        return {"output_path": "outputs/evaluation_report.json"}
+
+
+@needs_langgraph
+def test_best_iteration_restored_when_accuracy_regresses(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "outputs").mkdir()
+
+    import pandas as pd
+    import agents.pipeline_graph as pg
+    from agents.jack_manager import ManagerAgent
+    from agents.freddi_explanation import ExplanationAgent
+    from langgraph.checkpoint.memory import MemorySaver
+
+    nadi, sabina = MarkedNadi(), PeakSabina()
+    agents = pg.Agents(
+        aurora=FakeAurora(), nadi=nadi, sabina=sabina,
+        manager=ManagerAgent(predictions_path=pg.PREDS, target_accuracy=0.60,
+                             max_iterations=6, patience=2, min_delta=0.01),
+        freddi=ExplanationAgent(use_ollama=False, output_path=pg.EXPL),
+    )
+    graph = pg.build_pipeline(agents, checkpointer=MemorySaver())
+    final = graph.invoke({"retune_request_path": None},
+                         {"configurable": {"thread_id": "t"}, "recursion_limit": pg.RECURSION_LIMIT})
+
+    assert final["final_action"] == "proceed"
+    assert nadi.calls >= 3, "loop must run past the accuracy peak for this test to bite"
+
+    # The artifacts that survive are the BEST pass's (accuracy 0.39 = pass 2),
+    # not the last pass's — the whole point of select_best.
+    preds = pd.read_csv(tmp_path / "outputs" / "predictions_test.csv")
+    assert preds["fake_pass"].iloc[0] == 2, "final predictions should come from the best pass"
+    report = json.loads((tmp_path / "outputs" / "evaluation_report.json").read_text())
+    assert report["accuracy"] == 0.39
+    final_report = json.loads((tmp_path / "outputs" / "final_report.json").read_text())
+    assert final_report["final_accuracy"] == 0.39
+
+    # The explanation sample was redrawn from the restored predictions.
+    assert (tmp_path / "outputs" / "sample_for_explanation.csv").exists()
+
+
 @needs_langgraph
 def test_graph_cycles_then_finalizes(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)