diff --git a/analyze_arrivalscale_occupancy.py b/analyze_arrivalscale_occupancy.py
index 2b79a1d..7684db1 100644
--- a/analyze_arrivalscale_occupancy.py
+++ b/analyze_arrivalscale_occupancy.py
@@ -60,7 +60,7 @@
     r"(?P<baseline_off_cost_1k>-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?"
     r"Jobs=[\d,]+\/[\d,]+\s+\((?P<completion_rate>-?[\d.]+)%\),\s*"
     r"AvgWait=(?P<avg_wait>-?[\d.]+)h,.*?"
-    r"Dropped=(?P<agent_dropped>-?[\d,]+),.*?"
+    r"(?:Dropped|Lost)=(?P<agent_dropped>-?[\d,]+),.*?"
     r"Agent Occupancy \(Nodes\)=\s*(?P<occupancy>-?[\d.]+)%,\s*"
     r"Baseline Occupancy \(Nodes\)=\s*(?P<baseline_occupancy>-?[\d.]+)%",
     re.MULTILINE,
@@ -78,11 +78,11 @@
 )
 
 DROPPED_AGENT_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
 )
 
 DROPPED_BASELINE_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
 )
 
 
@@ -200,7 +200,8 @@ def parse_episode_metrics(
     if not occupancy:
         raise RuntimeError(
             "Could not parse episode metrics from train.py output. "
-            "Expected lines like 'Episode X: ... Savings=€.../€..., Power=..., CostPer1kCompleted=..., Agent Occupancy (Nodes)=...%'."
+            "Expected lines like 'Episode X: ... Savings=€.../€..., Power=..., CostPer1kCompleted=..., "
+            "(Lost|Dropped)=..., Agent Occupancy (Nodes)=...%'."
         )
 
     return (
diff --git a/analyze_lambda_occupancy.py b/analyze_lambda_occupancy.py
index 5f9df89..61cac37 100644
--- a/analyze_lambda_occupancy.py
+++ b/analyze_lambda_occupancy.py
@@ -53,7 +53,7 @@
     r"(?P<baseline_off_cost_1k>-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?"
     r"Jobs=[\d,]+\/[\d,]+\s+\((?P<completion_rate>-?[\d.]+)%\),\s*"
     r"AvgWait=(?P<avg_wait>-?[\d.]+)h,.*?"
-    r"Dropped=(?P<agent_dropped>-?[\d,]+),.*?"
+    r"(?:Dropped|Lost)=(?P<agent_dropped>-?[\d,]+),.*?"
     r"Agent Occupancy \(Nodes\)=\s*(?P<occupancy>-?[\d.]+)%,\s*"
     r"Baseline Occupancy \(Nodes\)=\s*(?P<baseline_occupancy>-?[\d.]+)%",
     re.MULTILINE,
@@ -71,11 +71,11 @@
 )
 
 DROPPED_AGENT_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
 )
 
 DROPPED_BASELINE_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
 )
 
 
diff --git a/analyze_seed_occupancy.py b/analyze_seed_occupancy.py
index 10c4073..cfc6b84 100644
--- a/analyze_seed_occupancy.py
+++ b/analyze_seed_occupancy.py
@@ -57,7 +57,7 @@
     r"(?P<baseline_off_cost_1k>-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?"
     r"Jobs=[\d,]+\/[\d,]+\s+\((?P<completion_rate>-?[\d.]+)%\),\s*"
     r"AvgWait=(?P<avg_wait>-?[\d.]+)h,.*?"
-    r"Dropped=(?P<agent_dropped>-?[\d,]+),.*?"
+    r"(?:Dropped|Lost)=(?P<agent_dropped>-?[\d,]+),.*?"
     r"Agent Occupancy \(Nodes\)=\s*(?P<occupancy>-?[\d.]+)%,\s*"
     r"Baseline Occupancy \(Nodes\)=\s*(?P<baseline_occupancy>-?[\d.]+)%",
     re.MULTILINE,
@@ -75,11 +75,11 @@
 )
 
 DROPPED_AGENT_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
 )
 
 DROPPED_BASELINE_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
 )
 
 
diff --git a/src/callbacks.py b/src/callbacks.py
index 2675696..ae87c7d 100644
--- a/src/callbacks.py
+++ b/src/callbacks.py
@@ -1,4 +1,4 @@
-from src.config import EPISODE_HOURS, MAX_QUEUE_SIZE, MAX_NODES
+from src.config import COST_IDLE_MW, COST_USED_MW, CORES_PER_NODE, MAX_NODES
 from stable_baselines3.common.callbacks import BaseCallback
 
 
@@ -24,63 +24,64 @@ def _on_rollout_start(self) -> None:
 
     def _on_step(self) -> bool:
         env = self.training_env.envs[0].unwrapped
-        if env.metrics.current_hour == EPISODE_HOURS-1:
-            self.logger.record("metrics/total_reward", env.metrics.episode_reward)
-            self.logger.record("metrics/reward_eff", sum(env.metrics.episode_eff_rewards) / 100)
-            self.logger.record("metrics/reward_price", sum(env.metrics.episode_price_rewards) / 100)
-            self.logger.record("metrics/penalty_idle", sum(env.metrics.episode_idle_penalties) / 100)
-            self.logger.record("metrics/penalty_job_age", sum(env.metrics.episode_job_age_penalties) / 100)
-            self.logger.record("metrics/penalty_drop", sum(env.metrics.episode_drop_penalties))
+        dones = self.locals.get("dones")
+        if dones is None or not bool(dones[0]) or not env.metrics.episode_costs:
+            return True
 
-            self.logger.record("metrics/cost", env.metrics.episode_total_cost)
-            self.logger.record("metrics/savings", env.metrics.episode_baseline_cost - env.metrics.episode_total_cost)
-            savings_off = env.metrics.episode_baseline_cost_off - env.metrics.episode_total_cost
-            self.logger.record("metrics/savings_off", savings_off)
-            savings_off_clean = savings_off if env.metrics.episode_jobs_dropped == 0 else 0.0
-            self.logger.record("metrics/savings_off_clean", savings_off_clean)
-            #self.logger.record("metrics/queue_fill_pct", env.metrics.episode_max_queue_size_reached / MAX_QUEUE_SIZE * 100)
-            self.logger.record("metrics/bl_cost", env.metrics.episode_baseline_cost)
-            self.logger.record("metrics/bl_cost_off", env.metrics.episode_baseline_cost_off)
+        episode_data = env.metrics.episode_costs[-1]
 
-            # Job metrics (agent)
-            completion_rate = (env.metrics.episode_jobs_completed / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0)
-            avg_wait = (env.metrics.episode_total_job_wait_time / env.metrics.episode_jobs_completed if env.metrics.episode_jobs_completed > 0 else 0.0)
-            self.logger.record("metrics/jobs_submitted", env.metrics.episode_jobs_submitted)
-            self.logger.record("metrics/jobs_completed", env.metrics.episode_jobs_completed)
-            self.logger.record("metrics/completion_rate", completion_rate)
-            self.logger.record("metrics/avg_wait_hours", avg_wait)
-            self.logger.record("metrics/nodes_on", env.metrics.episode_on_nodes[-1])
-            self.logger.record("metrics/nodes_used", env.metrics.episode_used_nodes[-1])
-            self.logger.record("metrics/nodes_idle", env.metrics.episode_on_nodes[-1] - env.metrics.episode_used_nodes[-1])
-            self.logger.record("metrics/nodes_off", MAX_NODES - env.metrics.episode_on_nodes[-1])
-            self.logger.record("metrics/max_queue_size", env.metrics.episode_max_queue_size_reached)
-            self.logger.record("metrics/max_backlog_size", env.metrics.episode_max_backlog_size_reached)
-            self.logger.record("metrics/jobs_dropped", env.metrics.episode_jobs_dropped)
-            self.logger.record("metrics/jobs_lost_total", env.metrics.episode_jobs_dropped)
-            loss_rate = (env.metrics.episode_jobs_dropped / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0)
-            self.logger.record("metrics/loss_rate", loss_rate)
-            self.logger.record("metrics/jobs_rejected_queue_full", env.metrics.episode_jobs_rejected_queue_full)
+        self.logger.record("metrics/cost", float(episode_data["agent_cost"]))
+        self.logger.record("metrics/savings", float(episode_data["savings_vs_baseline"]))
+        self.logger.record("metrics/savings_off", float(episode_data["savings_vs_baseline_off"]))
+        self.logger.record("metrics/baseline_cost", float(episode_data["baseline_cost"]))
+        self.logger.record("metrics/baseline_cost_off", float(episode_data["baseline_cost_off"]))
 
-            # Job metrics (baseline)
-            baseline_completion_rate = (env.metrics.episode_baseline_jobs_completed / env.metrics.episode_baseline_jobs_submitted * 100 if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0)
-            baseline_avg_wait = (env.metrics.episode_baseline_total_job_wait_time / env.metrics.episode_baseline_jobs_completed if env.metrics.episode_baseline_jobs_completed > 0 else 0.0)
-            self.logger.record("metrics/bl_jobs_submitted", env.metrics.episode_baseline_jobs_submitted)
-            self.logger.record("metrics/bl_jobs_completed", env.metrics.episode_baseline_jobs_completed)
-            self.logger.record("metrics/bl_completion_rate", baseline_completion_rate)
-            self.logger.record("metrics/bl_avg_wait_hours", baseline_avg_wait)
-            self.logger.record("metrics/bl_max_queue_size", env.metrics.episode_baseline_max_queue_size_reached)
-            self.logger.record("metrics/bl_max_backlog_size", env.metrics.episode_baseline_max_backlog_size_reached)
-            self.logger.record("metrics/bl_jobs_dropped", env.metrics.episode_baseline_jobs_dropped)
-            self.logger.record("metrics/bl_jobs_lost_total", env.metrics.episode_baseline_jobs_dropped)
-            baseline_loss_rate = (env.metrics.episode_baseline_jobs_dropped / env.metrics.episode_baseline_jobs_submitted * 100 if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0)
-            self.logger.record("metrics/bl_loss_rate", baseline_loss_rate)
-            self.logger.record("metrics/bl_jobs_rejected_queue_full", env.metrics.episode_baseline_jobs_rejected_queue_full)
+        # Job metrics (agent)
+        self.logger.record("metrics/jobs_submitted", int(episode_data["jobs_submitted"]))
+        self.logger.record("metrics/jobs_launched", int(episode_data["jobs_launched"]))
+        self.logger.record("metrics/jobs_completed", int(episode_data["jobs_completed"]))
+        self.logger.record("metrics/completion_rate", float(episode_data["completion_rate"]))
+        self.logger.record("metrics/avg_wait_hours", float(episode_data["avg_wait_time"]))
+        self.logger.record("metrics/on_nodes", int(episode_data.get("on_nodes_end", 0)))
+        self.logger.record("metrics/used_nodes", int(episode_data.get("used_nodes_end", 0)))
+        self.logger.record("metrics/max_queue_size", int(episode_data["max_queue_size"]))
+        self.logger.record("metrics/max_backlog_size", int(episode_data["max_backlog_size"]))
+        self.logger.record("metrics/max_drop_streak", int(episode_data.get("max_drop_streak", 0)))
+        self.logger.record("metrics/pending_jobs_end", int(episode_data.get("pending_jobs_end", 0)))
+        self.logger.record("metrics/pending_core_hours_end", float(episode_data.get("pending_core_hours_end", 0.0)))
+        self.logger.record("metrics/overdue_jobs_end", int(episode_data.get("overdue_jobs_end", 0)))
+        self.logger.record("metrics/jobs_dropped", int(episode_data["jobs_dropped"]))
+        self.logger.record("metrics/jobs_flushed", int(episode_data.get("jobs_flushed", 0)))
+        self.logger.record("metrics/jobs_lost_total", int(episode_data["jobs_lost_total"]))
+        self.logger.record("metrics/loss_rate", float(episode_data["loss_rate"]))
+        self.logger.record("metrics/jobs_rejected_queue_full", int(episode_data["jobs_rejected_queue_full"]))
 
-            # Power metrics
-            self.logger.record("metrics/power_mwh", env.metrics.episode_total_power_consumption_mwh)
-            self.logger.record("metrics/bl_power_mwh", env.metrics.episode_baseline_power_consumption_mwh)
-            self.logger.record("metrics/bl_off_power_mwh", env.metrics.episode_baseline_power_consumption_off_mwh)
-            self.logger.record("metrics/savings_power_vs_baseline_off", env.metrics.episode_baseline_power_consumption_off_mwh - env.metrics.episode_total_power_consumption_mwh)
+        # Job metrics (baseline)
+        self.logger.record("metrics/baseline_jobs_submitted", int(episode_data["baseline_jobs_submitted"]))
+        self.logger.record("metrics/baseline_jobs_launched", int(episode_data["baseline_jobs_launched"]))
+        self.logger.record("metrics/baseline_jobs_completed", int(episode_data["baseline_jobs_completed"]))
+        self.logger.record("metrics/baseline_completion_rate", float(episode_data["baseline_completion_rate"]))
+        self.logger.record("metrics/baseline_avg_wait_hours", float(episode_data["baseline_avg_wait_time"]))
+        self.logger.record("metrics/baseline_max_queue_size", int(episode_data["baseline_max_queue_size"]))
+        self.logger.record("metrics/baseline_max_backlog_size", int(episode_data["baseline_max_backlog_size"]))
+        self.logger.record("metrics/baseline_jobs_dropped", int(episode_data["baseline_jobs_dropped"]))
+        self.logger.record("metrics/baseline_jobs_flushed", int(episode_data.get("baseline_jobs_flushed", 0)))
+        self.logger.record("metrics/baseline_jobs_lost_total", int(episode_data["baseline_jobs_lost_total"]))
+        self.logger.record("metrics/baseline_loss_rate", float(episode_data["baseline_loss_rate"]))
+        self.logger.record("metrics/baseline_jobs_rejected_queue_full", int(episode_data["baseline_jobs_rejected_queue_full"]))
+
+        # Proportional (per-core) power metrics
+        self.logger.record("metrics/prop_power_mwh", float(episode_data["agent_prop_power_mwh"]))
+        self.logger.record("metrics/baseline_prop_power_mwh", float(episode_data["baseline_prop_power_mwh"]))
+        self.logger.record("metrics/baseline_off_prop_power_mwh", float(episode_data["baseline_off_prop_power_mwh"]))
+        self.logger.record(
+            "metrics/savings_prop_power_vs_baseline_off",
+            float(episode_data["baseline_off_prop_power_mwh"]) - float(episode_data["agent_prop_power_mwh"]),
+        )
+        self.logger.record(
+            "metrics/savings_prop_cost_vs_baseline_off",
+            float(episode_data["savings_prop_cost_vs_baseline_off"]),
+        )
 
         return True
 
diff --git a/src/environment.py b/src/environment.py
index 034e3f3..a506869 100644
--- a/src/environment.py
+++ b/src/environment.py
@@ -20,7 +20,7 @@
 from src.config import (
     MAX_NODES, MAX_QUEUE_SIZE, MAX_CHANGE, MAX_JOB_DURATION,
     CORES_PER_NODE, MAX_CORES_PER_JOB, MAX_JOB_AGE_OBS,
-    MAX_NODES_PER_JOB, EPISODE_HOURS, PENALTY_DROPPED_JOB
+    MAX_NODES_PER_JOB, EPISODE_HOURS
 )
 from src.job_management import (
     process_ongoing_jobs, add_new_jobs,
@@ -46,6 +46,7 @@ class ComputeClusterEnv(gym.Env):
     """An environment for scheduling compute jobs based on electricity price predictions."""
 
     metadata = {'render.modes': ['human', 'none']}
+    DROP_STREAK_TERMINATION_PENALTY = -200.0
 
     def render(self, mode: str = 'human') -> None:
         self.render_mode = mode
@@ -75,7 +76,9 @@ def __init__(self,
                  workload_gen: WorkloadGenerator | None = None,
                  job_arrival_scale: float = 1.0,
                  jobs_exact_replay: bool = False,
-                 output_dir: str = "sessions") -> None:
+                 output_dir: str = "sessions",
+                 jobs_exact_replay_aggregate: bool = False,
+                 flush_after_drop_streak: int = 0) -> None:
         super().__init__()
 
         self.weights = weights
@@ -90,6 +93,9 @@ def __init__(self,
         self.evaluation_mode = evaluation_mode
         self.job_arrival_scale = float(job_arrival_scale)
         self.jobs_exact_replay = bool(jobs_exact_replay)
+        self.jobs_exact_replay_aggregate = bool(jobs_exact_replay_aggregate)
+        self.flush_after_drop_streak = max(0, int(flush_after_drop_streak))
+        self.consecutive_drop_steps = 0
 
         self.next_plot_save = self.steps_per_iteration
 
@@ -118,7 +124,10 @@ def __init__(self,
             print(f"Parsed aggregated jobs for {len(self.jobs_sampler.aggregated_jobs)} hours")
             if self.jobs_exact_replay:
                 max_raw_jobs = max((len(v) for v in self.jobs_sampler.jobs.values()), default=0)
-                print("Jobs replay mode: exact timeline (raw jobs per hour)")
+                if self.jobs_exact_replay_aggregate:
+                    print("Jobs replay mode: exact timeline (aggregated per step)")
+                else:
+                    print("Jobs replay mode: exact timeline (raw jobs per hour)")
                 print(f"Max raw jobs per hour: {max_raw_jobs}")
             else:
                 self.jobs_sampler.precalculate_hourly_jobs(CORES_PER_NODE, MAX_NODES_PER_JOB)
@@ -150,6 +159,9 @@ def __init__(self,
         self.metrics.reset_timeline_metrics()
         self.metrics.reset_episode_metrics()
         self._reset_timeline_state(start_index=0)
+        # A fresh env still needs one real reset() call before rollout starts. After that,
+        # episode boundaries should only roll metrics, not wipe the live simulation state.
+        self._timeline_initialized = False
 
         # actions: - change number of available nodes:
         #   action_type:      0: decrease, 1: maintain, 2: increase
@@ -179,6 +191,7 @@ def __init__(self,
 
     def _reset_timeline_state(self, start_index: int) -> None:
         self.prices.reset(start_index=start_index)
+        self.consecutive_drop_steps = 0
 
         self.state = {
             # Initialize all nodes to be 'online but free' (0)
@@ -224,58 +237,189 @@ def _mark_queue_backlog_mutation(self) -> None:
         """Invalidate pending-job stats cache after queue/backlog content changes."""
         self._queue_backlog_version += 1
 
+    def _pending_work_summary(self, job_queue_2d: np.ndarray, backlog_queue: deque | None = None) -> dict[str, float | int]:
+        """
+        Summarize currently pending work across the visible queue and the overflow backlog.
+
+        The reward only needs a few dense signals:
+        - instantaneous runnable demand (nodes * cores)
+        - longer-horizon remaining work (core-hours)
+        - overdue mass after the intentional deferral grace period
+        """
+        if backlog_queue is None:
+            backlog_queue = self.backlog_queue
+
+        active_jobs_mask = job_queue_2d[:, 0] > 0
+        queue_rows = job_queue_2d[active_jobs_mask].astype(np.float32, copy=False)
+
+        if backlog_queue:
+            backlog_rows = np.asarray(list(backlog_queue), dtype=np.float32)
+        else:
+            backlog_rows = np.empty((0, 4), dtype=np.float32)
+
+        if queue_rows.size and backlog_rows.size:
+            combined_rows = np.vstack((queue_rows, backlog_rows))
+        elif queue_rows.size:
+            combined_rows = queue_rows
+        elif backlog_rows.size:
+            combined_rows = backlog_rows
+        else:
+            combined_rows = np.empty((0, 4), dtype=np.float32)
+
+        if combined_rows.size == 0:
+            return {
+                "pending_job_count": 0,
+                "pending_core_demand": 0.0,
+                "pending_core_hours": 0.0,
+                "pending_avg_duration": 0.0,
+                "pending_max_nodes": 0,
+                "backlog_size": len(backlog_queue),
+                "oldest_age": 0.0,
+                "overdue_jobs": 0,
+                "overdue_age_core_hours": 0.0,
+            }
+
+        durations = combined_rows[:, 0]
+        ages = combined_rows[:, 1]
+        nodes = combined_rows[:, 2]
+        cores = combined_rows[:, 3]
+        core_demand = nodes * cores
+
+        grace = float(self.reward_calculator.DEFERRAL_GRACE_HOURS)
+        overdue_age = np.clip(ages - grace, a_min=0.0, a_max=None)
+        overdue_mask = overdue_age > 0.0
+
+        return {
+            "pending_job_count": int(combined_rows.shape[0]),
+            "pending_core_demand": float(np.sum(core_demand)),
+            "pending_core_hours": float(np.sum(durations * core_demand)),
+            "pending_avg_duration": float(np.mean(durations)),
+            "pending_max_nodes": int(np.max(nodes)),
+            "backlog_size": len(backlog_queue),
+            "oldest_age": float(np.max(ages)),
+            "overdue_jobs": int(np.count_nonzero(overdue_mask)),
+            "overdue_age_core_hours": float(np.sum(overdue_age * core_demand)),
+        }
+
     def _update_pending_job_stats(self, job_queue_2d: np.ndarray) -> None:
         """Update summary statistics for all outstanding jobs (queue + backlog)."""
         # Fast path: skip recalculation if queue/backlog version is unchanged.
         if self._cached_queue_backlog_version == self._queue_backlog_version:
             return  # Stats unchanged from last step
 
-        # Slow path: recalculate pending stats after queue/backlog mutations.
-        # Collect stats from the main queue
-        current_backlog_size = len(self.backlog_queue)
-        active_jobs_mask = job_queue_2d[:, 0] > 0
-        queue_durations = job_queue_2d[active_jobs_mask, 0]
-        queue_nodes = job_queue_2d[active_jobs_mask, 2]
-        queue_cores = job_queue_2d[active_jobs_mask, 3]
-        queue_count = len(queue_durations)
-
-        # Collect stats from the backlog
-        backlog_count = current_backlog_size
-        if backlog_count > 0:
-            backlog_durations = np.array([job[0] for job in self.backlog_queue], dtype=np.int32)
-            backlog_nodes = np.array([job[2] for job in self.backlog_queue], dtype=np.int32)
-            backlog_cores = np.array([job[3] for job in self.backlog_queue], dtype=np.int32)
-        else:
-            backlog_durations = np.array([], dtype=np.int32)
-            backlog_nodes = np.array([], dtype=np.int32)
-            backlog_cores = np.array([], dtype=np.int32)
-
-        # Combine stats
-        total_count = queue_count + backlog_count
-        if total_count > 0:
-            all_durations = np.concatenate([queue_durations, backlog_durations])
-            all_nodes = np.concatenate([queue_nodes, backlog_nodes])
-            all_cores = np.concatenate([queue_cores, backlog_cores])
-
-            # Core-hours = sum of (duration * nodes * cores_per_node)
-            total_core_hours = np.sum(all_durations * all_nodes * all_cores)
-            avg_duration = np.mean(all_durations)
-            max_nodes = np.max(all_nodes)
-        else:
-            total_core_hours = 0.0
-            avg_duration = 0.0
-            max_nodes = 0
-
+        summary = self._pending_work_summary(job_queue_2d)
         # Update state
-        self.state['pending_job_count'][0] = total_count
-        self.state['pending_core_hours'][0] = total_core_hours
-        self.state['pending_avg_duration'][0] = avg_duration
-        self.state['pending_max_nodes'][0] = max_nodes
-        self.state['backlog_size'][0] = backlog_count
+        self.state['pending_job_count'][0] = int(summary["pending_job_count"])
+        self.state['pending_core_hours'][0] = float(summary["pending_core_hours"])
+        self.state['pending_avg_duration'][0] = float(summary["pending_avg_duration"])
+        self.state['pending_max_nodes'][0] = int(summary["pending_max_nodes"])
+        self.state['backlog_size'][0] = int(summary["backlog_size"])
 
         # Cache the queue/backlog version for next step.
         self._cached_queue_backlog_version = self._queue_backlog_version
 
+    @staticmethod
+    def _count_queued_jobs(job_queue_2d: np.ndarray) -> int:
+        """Count active jobs represented in a dense queue array."""
+        return int(np.count_nonzero(job_queue_2d[:, 0] > 0))
+
+    def _flush_workload_side(
+        self,
+        job_queue_2d: np.ndarray,
+        nodes: np.ndarray,
+        cores_available: np.ndarray,
+        running_jobs: dict[int, dict[str, Any]],
+        backlog_queue: deque,
+    ) -> int:
+        """
+        Drop all outstanding work for one side of the simulation and reset its cluster.
+
+        This is intentionally stronger than a normal episode rollover: queued jobs,
+        backlog jobs, and still-running jobs are all considered lost so the next
+        episode can start from a clean slate.
+        """
+        flushed_jobs = (
+            self._count_queued_jobs(job_queue_2d)
+            + len(backlog_queue)
+            + len(running_jobs)
+        )
+
+        job_queue_2d.fill(0)
+        nodes.fill(0)
+        cores_available.fill(CORES_PER_NODE)
+        running_jobs.clear()
+        backlog_queue.clear()
+        return int(flushed_jobs)
+
+    def _flush_workload_state(self) -> dict[str, int | float | bool]:
+        """Flush both agent and baseline states immediately after a trigger step."""
+        agent_job_queue_2d = self.state['job_queue'].reshape(-1, 4)
+        baseline_job_queue_2d = self.baseline_state['job_queue'].reshape(-1, 4)
+
+        agent_jobs_flushed = self._flush_workload_side(
+            agent_job_queue_2d,
+            self.state['nodes'],
+            self.cores_available,
+            self.running_jobs,
+            self.backlog_queue,
+        )
+        baseline_jobs_flushed = self._flush_workload_side(
+            baseline_job_queue_2d,
+            self.baseline_state['nodes'],
+            self.baseline_cores_available,
+            self.baseline_running_jobs,
+            self.baseline_backlog_queue,
+        )
+
+        self.next_empty_slot = 0
+        self.baseline_next_empty_slot = 0
+        self.metrics.current_running_jobs = 0
+
+        if agent_jobs_flushed > 0:
+            self.metrics.jobs_flushed += agent_jobs_flushed
+            self.metrics.episode_jobs_flushed += agent_jobs_flushed
+        if baseline_jobs_flushed > 0:
+            self.metrics.baseline_jobs_flushed += baseline_jobs_flushed
+            self.metrics.episode_baseline_jobs_flushed += baseline_jobs_flushed
+
+        flush_penalty = self.reward_calculator.loss_penalty(agent_jobs_flushed)
+
+        if agent_jobs_flushed > 0 or baseline_jobs_flushed > 0:
+            self._mark_queue_backlog_mutation()
+            self._update_pending_job_stats(agent_job_queue_2d)
+        self.consecutive_drop_steps = 0
+
+        self.state['job_queue'] = agent_job_queue_2d.flatten()
+        self.baseline_state['job_queue'] = baseline_job_queue_2d.flatten()
+
+        return {
+            "flush_applied": bool(agent_jobs_flushed > 0 or baseline_jobs_flushed > 0),
+            "agent_jobs_flushed": agent_jobs_flushed,
+            "baseline_jobs_flushed": baseline_jobs_flushed,
+            "flush_penalty": float(flush_penalty),
+        }
+
+    def _resolve_reset_start_index(self, options: dict[str, Any]) -> int:
+        """
+        Choose the initial price position for a true timeline reset.
+
+        Normal episode rollovers are continuous and should not call this helper. It is
+        only used when the whole simulation timeline is intentionally restarted.
+        """
+        if "price_start_index" in options:
+            if self.prices is not None and self.prices.external_prices is not None:
+                n_prices = len(self.prices.external_prices)
+                return int(options["price_start_index"]) % n_prices
+            return int(options["price_start_index"])
+
+        if self.prices.external_prices is not None:
+            if self.evaluation_mode:
+                return (self.episode_idx * EPISODE_HOURS) % len(self.prices.external_prices)
+            return int(self.np_random.integers(0, len(self.prices.external_prices)))
+
+        # Even the synthetic logic prices benefit from varying the initial phase during training.
+        return int(self.np_random.integers(0, self.prices.PREDICTION_WINDOW))
+
     def reset(self, seed: int | None = None, options: dict[str, Any] | None = None) -> tuple[dict[str, np.ndarray], dict]:
         if options is None:
             options = {}
@@ -289,14 +433,23 @@ def reset(self, seed: int | None = None, options: dict[str, Any] | None = None)
             self.episode_idx += 1
 
         self.metrics.reset_episode_metrics()
-        if "price_start_index" in options:
-            if self.prices is not None and self.prices.external_prices is not None:
-                n_prices = len(self.prices.external_prices)
-                start_index = int(options["price_start_index"]) % n_prices
-            else:
-                start_index = int(options["price_start_index"])
-            self.prices.reset(start_index=start_index)
-            self.state["predicted_prices"] = self.prices.predicted_prices.copy()
+
+        hard_reset = bool(options.get("hard_reset", False))
+        if not self._timeline_initialized or hard_reset:
+            # Hard reset: restart prices, queues, nodes, backlog, and running jobs.
+            # This is only for the very first env reset or for explicit ablations/debug runs.
+            start_index = self._resolve_reset_start_index(options)
+            self._reset_timeline_state(start_index=start_index)
+            self._timeline_initialized = True
+        else:
+            # Soft reset: keep the ongoing simulation exactly as-is and just start a new
+            # reporting window. Any price_start_index request is ignored unless a hard reset
+            # is explicitly requested, because jumping the price stream mid-timeline would
+            # break continuity.
+            job_queue_2d = self.state['job_queue'].reshape(-1, 4)
+            self._mark_queue_backlog_mutation()
+            self._update_pending_job_stats(job_queue_2d)
+            self.state['predicted_prices'] = self.prices.predicted_prices.copy()
 
         return self.state, {}
 
@@ -374,6 +527,9 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
         self.env_print(f">>> adding {len(new_jobs)} new jobs to the queue: {' '.join(['[{}h {} {}x{}]'.format(d, a, n, c) for d, a, n, c in new_jobs])}")
         self.env_print("job_queue: ", ' '.join(['[{} {} {} {}]'.format(d, a, n, c) for d, a, n, c in job_queue_2d if d > 0]))
 
+        # Snapshot the pending queue the agent is deciding about *before* launching jobs.
+        decision_pending_summary = self._pending_work_summary(job_queue_2d)
+
         action_type, action_magnitude, do_refill = action
         action_magnitude += 1
 
@@ -412,6 +568,8 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
                 if extra_launched > 0:
                     self.env_print(f"   {extra_launched} additional jobs launched from backlog")
 
+        remaining_pending_summary = self._pending_work_summary(job_queue_2d)
+
         # Update summary statistics for all outstanding jobs (queue + backlog)
         if queue_backlog_mutated:
             self._mark_queue_backlog_mutation()
@@ -441,6 +599,11 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
         self.metrics.episode_used_cores.append(num_used_cores)
         self.metrics.episode_job_queue_sizes.append(num_unprocessed_jobs)
         self.metrics.episode_price_stats.append(current_price)
+        self.metrics.episode_pending_jobs_end = int(remaining_pending_summary["pending_job_count"])
+        self.metrics.episode_pending_core_demand_end = float(remaining_pending_summary["pending_core_demand"])
+        self.metrics.episode_pending_core_hours_end = float(remaining_pending_summary["pending_core_hours"])
+        self.metrics.episode_overdue_jobs_end = int(remaining_pending_summary["overdue_jobs"])
+        self.metrics.episode_overdue_age_core_hours_end = float(remaining_pending_summary["overdue_age_core_hours"])
 
         # Track max queue size (queue only, without backlog)
         queue_only_size = np.sum(job_queue_2d[:, 0] > 0)
@@ -482,6 +645,8 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
             num_used_nodes, num_idle_nodes, current_price, average_future_price,
             num_off_nodes, job_queue_2d, num_unprocessed_jobs, self.weights,
             num_dropped_this_step, self.env_print, num_on_nodes, num_used_cores,
+            decision_pending_core_demand=float(decision_pending_summary["pending_core_demand"]),
+            remaining_overdue_age_core_hours=float(remaining_pending_summary["overdue_age_core_hours"]),
         )
 
         self.metrics.episode_reward += step_reward
@@ -501,11 +666,56 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
         self.metrics.episode_price_rewards.append(price_reward * 100)
         self.metrics.episode_job_age_penalties.append(job_age_penalty_norm * 100)
         self.metrics.episode_idle_penalties.append(idle_penalty_norm * 100)
-        self.metrics.episode_drop_penalties.append(PENALTY_DROPPED_JOB * num_dropped_this_step)
         self.metrics.episode_rewards.append(step_reward)
         self.metrics.jobs_dropped += num_dropped_this_step
         self.metrics.episode_jobs_dropped += num_dropped_this_step
-
+        if num_dropped_this_step > 0:
+            self.consecutive_drop_steps += 1
+        else:
+            self.consecutive_drop_steps = 0
+        if self.consecutive_drop_steps > self.metrics.episode_max_drop_streak:
+            self.metrics.episode_max_drop_streak = self.consecutive_drop_steps
+        flush_applied = False
+        flush_penalty = 0.0
+        agent_jobs_flushed = 0
+        baseline_jobs_flushed = 0
+        terminal_penalty_applied = 0.0
+        drop_streak_steps = self.consecutive_drop_steps
+        flush_triggered_by_drop_streak = (
+            self.flush_after_drop_streak > 0
+            and self.consecutive_drop_steps >= self.flush_after_drop_streak
+        )
+        if flush_triggered_by_drop_streak:
+            flush_result = self._flush_workload_state()
+            flush_applied = bool(flush_result["flush_applied"])
+            flush_penalty = float(flush_result["flush_penalty"])
+            agent_jobs_flushed = int(flush_result["agent_jobs_flushed"])
+            baseline_jobs_flushed = int(flush_result["baseline_jobs_flushed"])
+            previous_step_reward = step_reward
+            terminal_penalty_applied = float(self.DROP_STREAK_TERMINATION_PENALTY)
+            step_reward = terminal_penalty_applied
+            self.metrics.episode_reward += terminal_penalty_applied - previous_step_reward
+            if self.metrics.rewards:
+                self.metrics.rewards[-1] = terminal_penalty_applied
+            if self.metrics.episode_rewards:
+                self.metrics.episode_rewards[-1] = terminal_penalty_applied
+
+            post_flush_pending_summary = self._pending_work_summary(self.state['job_queue'].reshape(-1, 4))
+            self.metrics.episode_pending_jobs_end = int(post_flush_pending_summary["pending_job_count"])
+            self.metrics.episode_pending_core_demand_end = float(post_flush_pending_summary["pending_core_demand"])
+            self.metrics.episode_pending_core_hours_end = float(post_flush_pending_summary["pending_core_hours"])
+            self.metrics.episode_overdue_jobs_end = int(post_flush_pending_summary["overdue_jobs"])
+            self.metrics.episode_overdue_age_core_hours_end = float(post_flush_pending_summary["overdue_age_core_hours"])
+
+            if flush_applied:
+                self.env_print(
+                    f"[flush] Drop streak {self.flush_after_drop_streak}+ reached "
+                    f"({drop_streak_steps} steps). "
+                    f"Dropped outstanding work immediately: "
+                    f"agent={agent_jobs_flushed}, baseline={baseline_jobs_flushed}, "
+                    f"loss_penalty={flush_penalty:.4f}, terminal_penalty={terminal_penalty_applied:.4f}"
+                )
+        
         # print stats
         self.env_print(f"[6] End of step stats...")
         self.env_print("job queue: ", ' '.join(['[{} {} {} {}]'.format(d, a, n, c) for d, a, n, c in job_queue_2d if d > 0]))
@@ -521,7 +731,10 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
 
         truncated = False
         terminated = False
-        if self.metrics.current_hour == EPISODE_HOURS:
+        if flush_triggered_by_drop_streak:
+            terminated = True
+            truncated = False
+        elif self.metrics.current_hour == EPISODE_HOURS:
             if self.render_mode == 'human':
                 plot_episode(self, EPISODE_HOURS, MAX_NODES, False, True, self.current_step)
                 if self.plot_config.plot_once:
@@ -535,7 +748,8 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
             truncated = True
             terminated = False
 
-            # Record episode costs for long-term analysis
+        if terminated or truncated:
+            # Record episode costs before reset so callbacks/evaluation can read the finished episode.
             self.metrics.record_episode_completion(self.current_episode)
 
         # flatten job_queue again
@@ -553,7 +767,16 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool,
             "num_unprocessed_jobs": num_unprocessed_jobs,
             "num_on_nodes": num_on_nodes,
             "episode_jobs_dropped": self.metrics.episode_jobs_dropped,
-            "episode_jobs_lost_total": self.metrics.episode_jobs_dropped,
+            "episode_jobs_flushed": self.metrics.episode_jobs_flushed,
+            "episode_jobs_lost_total": self.metrics.episode_jobs_dropped + self.metrics.episode_jobs_flushed,
+            "episode_flush_applied": flush_applied,
+            "episode_flush_triggered_by_drop_streak": flush_triggered_by_drop_streak,
+            "drop_streak_steps": drop_streak_steps,
+            "drop_streak_flush_armed": flush_triggered_by_drop_streak,
+            "step_flush_penalty": flush_penalty,
+            "step_terminal_penalty": terminal_penalty_applied,
+            "step_jobs_flushed": agent_jobs_flushed,
+            "step_baseline_jobs_flushed": baseline_jobs_flushed,
         }
 
         return self.state, step_reward, terminated, truncated, info
diff --git a/src/evaluation_summary.py b/src/evaluation_summary.py
index f9eb575..2572094 100644
--- a/src/evaluation_summary.py
+++ b/src/evaluation_summary.py
@@ -57,7 +57,10 @@ def build_episode_summary_line(
         f"Jobs={int(episode_data['jobs_completed'])}/{int(episode_data['jobs_submitted'])} "
         f"({float(episode_data['completion_rate']):.0f}%), "
         f"AvgWait={float(episode_data['avg_wait_time']):.1f}h, "
+        f"PendingEnd={int(episode_data.get('pending_jobs_end', 0))}, "
+        f"OverdueEnd={int(episode_data.get('overdue_jobs_end', 0))}, "
         f"EpisodeMaxQueue={int(episode_data['max_queue_size'])}, "
+        f"MaxDropStreak={int(episode_data.get('max_drop_streak', 0))}, "
         f"Lost={int(episode_data.get('jobs_lost_total', episode_data['jobs_dropped']))}, "
         f"TimelineMaxQueue={timeline_max_queue}, "
         f"Agent Occupancy (Cores)={agent_occupancy_cores_pct:.2f}%, "
diff --git a/src/job_management.py b/src/job_management.py
index 994de55..2f5e25a 100644
--- a/src/job_management.py
+++ b/src/job_management.py
@@ -269,6 +269,18 @@ def assign_jobs_to_available_nodes(
                 "allocation": job_allocation,
                 "wait_time": int(job_age),
             }
+            # Record scheduling delay when the job starts.
+            # Completion metrics are tracked separately when the job actually finishes.
+            if is_baseline:
+                metrics.baseline_jobs_launched += 1
+                metrics.baseline_total_job_wait_time_launch += int(job_age)
+                metrics.episode_baseline_jobs_launched += 1
+                metrics.episode_baseline_total_job_wait_time_launch += int(job_age)
+            else:
+                metrics.jobs_launched += 1
+                metrics.total_job_wait_time_launch += int(job_age)
+                metrics.episode_jobs_launched += 1
+                metrics.episode_total_job_wait_time_launch += int(job_age)
             next_job_id += 1
 
             # Clear job from queue
diff --git a/src/metrics_tracker.py b/src/metrics_tracker.py
index ef2dc71..580bf8b 100644
--- a/src/metrics_tracker.py
+++ b/src/metrics_tracker.py
@@ -1,5 +1,7 @@
 """Metrics tracking and episode recording for the PowerSched environment."""
 
+from src.config import COST_IDLE_MW, COST_USED_MW, CORES_PER_NODE, MAX_NODES
+
 
 class MetricsTracker:
     """Tracks metrics throughout training episodes."""
@@ -14,6 +16,11 @@ def _safe_ratio(numerator: float, denominator: float) -> float:
         """Safe division; returns NaN when denominator is not positive."""
         return (numerator / denominator) if denominator > 0.0 else float("nan")
 
+    @staticmethod
+    def _jobs_lost_total(dropped: int, flushed: int) -> int:
+        """Total lost jobs, including regular drops and streak-triggered flushes."""
+        return int(dropped) + int(flushed)
+
     def __init__(self) -> None:
         """Initialize all metric counters."""
         self.reset_timeline_metrics()
@@ -37,20 +44,26 @@ def reset_timeline_metrics(self) -> None:
 
         # Agent job metrics (cumulative across episodes)
         self.jobs_submitted: int = 0
+        self.jobs_launched: int = 0
         self.jobs_completed: int = 0
         self.total_job_wait_time: int = 0
+        self.total_job_wait_time_launch: int = 0
         self.max_queue_size_reached: int = 0
         self.max_backlog_size_reached: int = 0
         self.jobs_dropped: int = 0
+        self.jobs_flushed: int = 0
         self.jobs_rejected_queue_full: int = 0
 
         # Baseline job metrics (cumulative across episodes)
         self.baseline_jobs_submitted: int = 0
+        self.baseline_jobs_launched: int = 0
         self.baseline_jobs_completed: int = 0
         self.baseline_total_job_wait_time: int = 0
+        self.baseline_total_job_wait_time_launch: int = 0
         self.baseline_max_queue_size_reached: int = 0
         self.baseline_max_backlog_size_reached: int = 0
         self.baseline_jobs_dropped: int = 0
+        self.baseline_jobs_flushed: int = 0
         self.baseline_jobs_rejected_queue_full: int = 0
 
         # Time series data for plotting (cumulative)
@@ -78,22 +91,38 @@ def reset_episode_metrics(self) -> None:
 
         # Agent job metrics (episode)
         self.episode_jobs_submitted: int = 0
+        self.episode_jobs_launched: int = 0
         self.episode_jobs_completed: int = 0
         self.episode_total_job_wait_time: int = 0
+        self.episode_total_job_wait_time_launch: int = 0
         self.episode_max_queue_size_reached: int = 0
         self.episode_max_backlog_size_reached: int = 0
         self.episode_jobs_dropped: int = 0
+        self.episode_jobs_flushed: int = 0
+        self.episode_max_drop_streak: int = 0
         self.episode_jobs_rejected_queue_full: int = 0
 
         # Baseline job metrics (episode)
         self.episode_baseline_jobs_submitted: int = 0
+        self.episode_baseline_jobs_launched: int = 0
         self.episode_baseline_jobs_completed: int = 0
         self.episode_baseline_total_job_wait_time: int = 0
+        self.episode_baseline_total_job_wait_time_launch: int = 0
         self.episode_baseline_max_queue_size_reached: int = 0
         self.episode_baseline_max_backlog_size_reached: int = 0
         self.episode_baseline_jobs_dropped: int = 0
+        self.episode_baseline_jobs_flushed: int = 0
         self.episode_baseline_jobs_rejected_queue_full: int = 0
 
+        # End-of-episode pending-work snapshot.
+        # These are updated every step so record_episode_completion() can report what
+        # backlog remains when the episode terminates.
+        self.episode_pending_jobs_end: int = 0
+        self.episode_pending_core_demand_end: float = 0.0
+        self.episode_pending_core_hours_end: float = 0.0
+        self.episode_overdue_jobs_end: int = 0
+        self.episode_overdue_age_core_hours_end: float = 0.0
+
         # Time series data for plotting (episode)
         self.episode_on_nodes: list[int] = []
         self.episode_used_nodes: list[int] = []
@@ -107,7 +136,6 @@ def reset_episode_metrics(self) -> None:
         self.episode_price_rewards: list[float] = []
         self.episode_idle_penalties: list[float] = []
         self.episode_job_age_penalties: list[float] = []
-        self.episode_drop_penalties: list[float] = []
         self.episode_rewards: list[float] = []
         self.episode_running_jobs_counts: list[int] = []
 
@@ -121,15 +149,16 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i
         Returns:
             Dictionary with episode data
         """
-        # Calculate average wait times
+        # Scheduling wait is measured when work launches, not when it eventually finishes.
+        # That makes "defer into cheap hours" visible immediately instead of only after completion.
         avg_wait_time: float = (
-            self.episode_total_job_wait_time / self.episode_jobs_completed
-            if self.episode_jobs_completed > 0
+            self.episode_total_job_wait_time_launch / self.episode_jobs_launched
+            if self.episode_jobs_launched > 0
             else 0.0
         )
         baseline_avg_wait_time: float = (
-            self.episode_baseline_total_job_wait_time / self.episode_baseline_jobs_completed
-            if self.episode_baseline_jobs_completed > 0
+            self.episode_baseline_total_job_wait_time_launch / self.episode_baseline_jobs_launched
+            if self.episode_baseline_jobs_launched > 0
             else 0.0
         )
 
@@ -150,13 +179,36 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i
             if self.episode_jobs_submitted
             else 0.0
         )
+        jobs_lost_total = self._jobs_lost_total(self.episode_jobs_dropped, self.episode_jobs_flushed)
+        flush_rate: float = (
+            (self.episode_jobs_flushed / self.episode_jobs_submitted * 100)
+            if self.episode_jobs_submitted
+            else 0.0
+        )
+        loss_rate: float = (
+            (jobs_lost_total / self.episode_jobs_submitted * 100)
+            if self.episode_jobs_submitted
+            else 0.0
+        )
         baseline_drop_rate: float = (
             (self.episode_baseline_jobs_dropped / self.episode_baseline_jobs_submitted * 100)
             if self.episode_baseline_jobs_submitted
             else 0.0
         )
-        loss_rate = drop_rate
-        baseline_loss_rate = baseline_drop_rate
+        baseline_jobs_lost_total = self._jobs_lost_total(
+            self.episode_baseline_jobs_dropped,
+            self.episode_baseline_jobs_flushed,
+        )
+        baseline_flush_rate: float = (
+            (self.episode_baseline_jobs_flushed / self.episode_baseline_jobs_submitted * 100)
+            if self.episode_baseline_jobs_submitted
+            else 0.0
+        )
+        baseline_loss_rate: float = (
+            (baseline_jobs_lost_total / self.episode_baseline_jobs_submitted * 100)
+            if self.episode_baseline_jobs_submitted
+            else 0.0
+        )
         agent_mean_price: float = self._effective_mean_price(
             self.episode_total_cost, self.episode_total_power_consumption_mwh
         )
@@ -179,6 +231,38 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i
         savings_vs_baseline: float = self.episode_baseline_cost - self.episode_total_cost
         savings_vs_baseline_off: float = self.episode_baseline_cost_off - self.episode_total_cost
 
+        # Proportional (per-core) power: idle_base for all on-nodes + compute delta scaled by core utilization.
+        # Formula per step: COST_IDLE_MW * num_on + (COST_USED_MW - COST_IDLE_MW) * (cores_used / CORES_PER_NODE)
+        _compute_delta_mw = COST_USED_MW - COST_IDLE_MW
+        agent_prop_power_mwh: float = sum(
+            COST_IDLE_MW * on + _compute_delta_mw * (cores / CORES_PER_NODE)
+            for on, cores in zip(self.episode_on_nodes, self.episode_used_cores)
+        )
+        # Baseline always has all MAX_NODES on
+        baseline_prop_power_mwh: float = sum(
+            COST_IDLE_MW * MAX_NODES + _compute_delta_mw * (cores / CORES_PER_NODE)
+            for cores in self.episode_baseline_used_cores
+        )
+        # Baseline_off: only used nodes are on (no idle nodes)
+        baseline_off_prop_power_mwh: float = sum(
+            COST_IDLE_MW * used + _compute_delta_mw * (cores / CORES_PER_NODE)
+            for used, cores in zip(self.episode_baseline_used_nodes, self.episode_baseline_used_cores)
+        )
+        # Proportional cost: same as prop power but multiplied by price at each step
+        agent_prop_cost: float = sum(
+            (COST_IDLE_MW * on + _compute_delta_mw * (cores / CORES_PER_NODE)) * price
+            for on, cores, price in zip(self.episode_on_nodes, self.episode_used_cores, self.episode_price_stats)
+        )
+        baseline_prop_cost: float = sum(
+            (COST_IDLE_MW * MAX_NODES + _compute_delta_mw * (cores / CORES_PER_NODE)) * price
+            for cores, price in zip(self.episode_baseline_used_cores, self.episode_price_stats)
+        )
+        baseline_off_prop_cost: float = sum(
+            (COST_IDLE_MW * used + _compute_delta_mw * (cores / CORES_PER_NODE)) * price
+            for used, cores, price in zip(self.episode_baseline_used_nodes, self.episode_baseline_used_cores, self.episode_price_stats)
+        )
+        savings_prop_cost_vs_baseline: float = baseline_prop_cost - agent_prop_cost
+        savings_prop_cost_vs_baseline_off: float = baseline_off_prop_cost - agent_prop_cost
         dropped_jobs_per_saved_euro: float = self._safe_ratio(
             float(self.episode_jobs_dropped), savings_vs_baseline
         ) if savings_vs_baseline > 0.0 else float("nan")
@@ -194,6 +278,14 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i
             'agent_power_consumption_mwh': self.episode_total_power_consumption_mwh,
             'baseline_power_consumption_mwh': self.episode_baseline_power_consumption_mwh,
             'baseline_power_consumption_off_mwh': self.episode_baseline_power_consumption_off_mwh,
+            'agent_prop_power_mwh': agent_prop_power_mwh,
+            'baseline_prop_power_mwh': baseline_prop_power_mwh,
+            'baseline_off_prop_power_mwh': baseline_off_prop_power_mwh,
+            'agent_prop_cost': agent_prop_cost,
+            'baseline_prop_cost': baseline_prop_cost,
+            'baseline_off_prop_cost': baseline_off_prop_cost,
+            'savings_prop_cost_vs_baseline': savings_prop_cost_vs_baseline,
+            'savings_prop_cost_vs_baseline_off': savings_prop_cost_vs_baseline_off,
             'agent_mean_price': agent_mean_price,
             'baseline_mean_price': baseline_mean_price,
             'baseline_off_mean_price': baseline_off_mean_price,
@@ -209,27 +301,41 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i
             'total_reward': self.episode_reward,
             # Agent job metrics
             'jobs_submitted': self.episode_jobs_submitted,
+            'jobs_launched': self.episode_jobs_launched,
             'jobs_completed': self.episode_jobs_completed,
             'avg_wait_time': avg_wait_time,
             'completion_rate': completion_rate,
+            'on_nodes_end': int(self.episode_on_nodes[-1]) if self.episode_on_nodes else 0,
+            'used_nodes_end': int(self.episode_used_nodes[-1]) if self.episode_used_nodes else 0,
             'max_queue_size': self.episode_max_queue_size_reached,
             'max_backlog_size': self.episode_max_backlog_size_reached,
+            'max_drop_streak': self.episode_max_drop_streak,
+            'pending_jobs_end': self.episode_pending_jobs_end,
+            'pending_core_demand_end': self.episode_pending_core_demand_end,
+            'pending_core_hours_end': self.episode_pending_core_hours_end,
+            'overdue_jobs_end': self.episode_overdue_jobs_end,
+            'overdue_age_core_hours_end': self.episode_overdue_age_core_hours_end,
             # Baseline job metrics
             'baseline_jobs_submitted': self.episode_baseline_jobs_submitted,
+            'baseline_jobs_launched': self.episode_baseline_jobs_launched,
             'baseline_jobs_completed': self.episode_baseline_jobs_completed,
             'baseline_avg_wait_time': baseline_avg_wait_time,
             'baseline_completion_rate': baseline_completion_rate,
             'baseline_max_queue_size': self.episode_baseline_max_queue_size_reached,
             'baseline_max_backlog_size': self.episode_baseline_max_backlog_size_reached,
-            # Loss metrics: includes age expirations and queue-full rejections.
+            # Loss metrics: includes age expirations, queue-full rejections, and streak-triggered flushes.
             "jobs_dropped": self.episode_jobs_dropped,
-            "jobs_lost_total": self.episode_jobs_dropped,
+            "jobs_flushed": self.episode_jobs_flushed,
+            "jobs_lost_total": jobs_lost_total,
             "drop_rate": drop_rate,
+            "flush_rate": flush_rate,
             "loss_rate": loss_rate,
             "jobs_rejected_queue_full": self.episode_jobs_rejected_queue_full,
             "baseline_jobs_dropped": self.episode_baseline_jobs_dropped,
-            "baseline_jobs_lost_total": self.episode_baseline_jobs_dropped,
+            "baseline_jobs_flushed": self.episode_baseline_jobs_flushed,
+            "baseline_jobs_lost_total": baseline_jobs_lost_total,
             "baseline_drop_rate": baseline_drop_rate,
+            "baseline_flush_rate": baseline_flush_rate,
             "baseline_loss_rate": baseline_loss_rate,
             "baseline_jobs_rejected_queue_full": self.episode_baseline_jobs_rejected_queue_full,
         }
diff --git a/src/plotter.py b/src/plotter.py
index db4980a..d4309df 100644
--- a/src/plotter.py
+++ b/src/plotter.py
@@ -23,6 +23,30 @@ def _as_series(x: np.ndarray | list[float] | None, n: int) -> np.ndarray | None:
     return out
 
 
+def _episode_launch_wait(metrics, baseline: bool = False) -> float:
+    """
+    Return average queueing delay measured at launch.
+
+    Old mock/test objects may still only expose completion-based wait fields, so this
+    helper keeps a compatible fallback for plotting utilities.
+    """
+    if baseline:
+        launched = getattr(metrics, "episode_baseline_jobs_launched", metrics.episode_baseline_jobs_completed)
+        total_wait = getattr(
+            metrics,
+            "episode_baseline_total_job_wait_time_launch",
+            metrics.episode_baseline_total_job_wait_time,
+        )
+    else:
+        launched = getattr(metrics, "episode_jobs_launched", metrics.episode_jobs_completed)
+        total_wait = getattr(
+            metrics,
+            "episode_total_job_wait_time_launch",
+            metrics.episode_total_job_wait_time,
+        )
+    return (total_wait / launched) if launched > 0 else 0.0
+
+
 def _compute_cumulative_savings(episode_costs: list[dict[str, float | int]]) -> dict[str, np.ndarray] | None:
     """
     episode_costs: list of dicts with keys:
@@ -113,7 +137,7 @@ def plot_episode(env: ComputeClusterEnv, num_hours: int, max_nodes: int, save: b
     if env.plot_config.plot_idle_penalty:
         ax2.plot(hours, env.metrics.episode_idle_penalties, color='green', linestyle='--', label='Idle Penalties')
     if env.plot_config.plot_job_age_penalty:
-        ax2.plot(hours, env.metrics.episode_job_age_penalties, color='yellow', linestyle='--', label='Job Age Penalties')
+        ax2.plot(hours, env.metrics.episode_job_age_penalties, color='yellow', linestyle='--', label='Backlog Pressure')
 
     ax2.tick_params(axis='y')
     if env.plot_config.plot_idle_penalty or env.plot_config.plot_job_age_penalty:
@@ -131,16 +155,8 @@ def plot_episode(env: ComputeClusterEnv, num_hours: int, max_nodes: int, save: b
         if env.metrics.episode_baseline_jobs_submitted > 0
         else 0
     )
-    avg_wait = (
-        env.metrics.episode_total_job_wait_time / env.metrics.episode_jobs_completed
-        if env.metrics.episode_jobs_completed > 0
-        else 0
-    )
-    baseline_avg_wait = (
-        env.metrics.episode_baseline_total_job_wait_time / env.metrics.episode_baseline_jobs_completed
-        if env.metrics.episode_baseline_jobs_completed > 0
-        else 0
-    )
+    avg_wait = _episode_launch_wait(env.metrics, baseline=False)
+    baseline_avg_wait = _episode_launch_wait(env.metrics, baseline=True)
     baseline_savings_pct = (
         ((env.metrics.episode_baseline_cost - env.metrics.episode_total_cost) / env.metrics.episode_baseline_cost) * 100
         if env.metrics.episode_baseline_cost > 0
@@ -197,16 +213,8 @@ def plot_dashboard(env: ComputeClusterEnv, num_hours: int, max_nodes: int, save:
         if env.metrics.episode_baseline_jobs_submitted > 0
         else 0.0
     )
-    avg_wait = (
-        (env.metrics.episode_total_job_wait_time / env.metrics.episode_jobs_completed)
-        if env.metrics.episode_jobs_completed > 0
-        else 0.0
-    )
-    baseline_avg_wait = (
-        (env.metrics.episode_baseline_total_job_wait_time / env.metrics.episode_baseline_jobs_completed)
-        if env.metrics.episode_baseline_jobs_completed > 0
-        else 0.0
-    )
+    avg_wait = _episode_launch_wait(env.metrics, baseline=False)
+    baseline_avg_wait = _episode_launch_wait(env.metrics, baseline=True)
 
     base_cost = float(env.metrics.episode_baseline_cost)
     base_cost_off = float(env.metrics.episode_baseline_cost_off)
@@ -268,7 +276,7 @@ def add_panel(title: str, series: np.ndarray | list[float] | None, ylabel: str,
     if env.plot_config.plot_idle_penalty:
         add_panel("Idle penalty (%)", env.metrics.episode_idle_penalties, "score", None)
     if env.plot_config.plot_job_age_penalty:
-        add_panel("Job-age penalty (%)", env.metrics.episode_job_age_penalties, "score", None)
+        add_panel("Backlog pressure (%)", env.metrics.episode_job_age_penalties, "score", None)
     if env.plot_config.plot_total_reward:
         add_panel("Total reward", getattr(env.metrics, "episode_rewards", None), "reward", None)
 
diff --git a/src/reward_calculation.py b/src/reward_calculation.py
index 2a36188..56b6891 100644
--- a/src/reward_calculation.py
+++ b/src/reward_calculation.py
@@ -54,16 +54,12 @@ class RewardCalculator:
     # Price scaling uses active used nodes as work proxy, matching efficiency semantics.
     PRICE_ADVANTAGE_GAIN = 4.0
     # Asymmetric node scaling: high-price execution ramps faster than low-price reward.
-    PRICE_NODE_TAU_POS = 70.0
+    PRICE_NODE_TAU_POS = 40.0
     PRICE_NODE_TAU_NEG = 40.0
-    NEGATIVE_PRICE_NODE_TAU = 14.0  # fast node saturation only for negative-price overdrive
+    PRICE_QUANTILE_LOW = 0.10
+    PRICE_QUANTILE_HIGH = 0.90
+    NEGATIVE_PRICE_NODE_TAU = 30.0  # fast node saturation only for negative-price overdrive
     NEGATIVE_PRICE_TAU = 8.0
-    # Overdrive terms for negative prices:
-    # - gain controls overdrive strength during negative-price windows
-    # - floor guarantees a minimum positive drive proportional to negative-price strength and used work
-    # Toggle behavior:
-    # - capped mode (default): overdrive is folded into tanh, so reward stays <= 1
-    # - uncapped mode: overdrive is added after tanh and can exceed 1 up to NEGATIVE_PRICE_OVERDRIVE_MAX_REWARD
     NEGATIVE_PRICE_OVERDRIVE_GAIN = 2.5
     NEGATIVE_PRICE_OVERDRIVE_FLOOR = 0.35
     NEGATIVE_PRICE_OVERDRIVE_ALLOW_ABOVE_ONE = True
@@ -71,6 +67,21 @@ class RewardCalculator:
     # Drop penalty: tanh saturation curve. TAU=20: 1 drop≈-0.05, 10 drops≈-0.46, 50 drops≈-1.0.
     DROP_PENALTY_TAU = 20.0
 
+
+    # The first 24h are treated as deliberate deferral room; after that, starvation should ramp up quickly.
+    DEFERRAL_GRACE_HOURS = 24
+    CHEAP_SERVICE_GAIN = 0.75
+    OVERDUE_BACKLOG_GAIN = 1.25
+    OVERDUE_AGE_CORE_HOUR_TAU = 0.5 * MAX_NODES * CORES_PER_NODE
+    # Separate intrinsic service-continuity signal:
+    # once work is overdue, staying near full blackout should hurt more than doing
+    # at least some useful work. This is intentionally unweighted so it remains a
+    # structural part of the objective rather than a tunable backlog coefficient.
+    INTRINSIC_STARVATION_GAIN = 1.5
+    INTRINSIC_STARVATION_OVERDUE_TAU = 0.25 * MAX_NODES * CORES_PER_NODE
+    INTRINSIC_STARVATION_SERVICE_CORE_TAU = 12.0
+    ALLOW_DROP_PENALTY = True  # whether to include penalties for dropped jobs in the reward calculation
+
     def __init__(self, prices: Prices) -> None:
         """
         Initialize reward calculator with normalization bounds.
@@ -132,6 +143,37 @@ def _price_context_average(self, average_future_price: float) -> float:
             return (history_avg + future_avg) / 2
         return average_future_price
 
+    def _price_phase_strengths(self, current_price: float) -> tuple[float, float]:
+        """
+        Map the current price into a cheap-vs-expensive phase inside the visible forecast window.
+
+        The quantile band is used as the "decision zone":
+        - at/below q_low  -> fully cheap
+        - at/above q_high -> fully expensive
+        - inside the band -> smooth linear interpolation
+
+        This is intentionally more explicit than the old sigmoid-based score. For the
+        synthetic 12h logic benchmark, it makes the cheap/expensive phase separation
+        obvious to the reward, which is exactly what we want to teach first.
+        """
+        prediction_window = np.asarray(self.prices.predicted_prices, dtype=np.float32)
+        future_reference = prediction_window[1:] if prediction_window.size > 1 else prediction_window
+        if future_reference.size < 2:
+            return 0.0, 0.0
+
+        q_low, q_high = np.quantile(
+            future_reference,
+            [self.PRICE_QUANTILE_LOW, self.PRICE_QUANTILE_HIGH],
+        )
+        price_band = float(q_high - q_low)
+        if price_band <= 1e-6:
+            return 0.0, 0.0
+
+        normalized = (current_price - float(q_low)) / price_band
+        cheap_strength = float(np.clip(1.0 - normalized, 0.0, 1.0))
+        expensive_strength = float(np.clip(normalized, 0.0, 1.0))
+        return cheap_strength, expensive_strength
+
     def _reward_price_legacy(self, current_price: float, average_future_price: float, num_processed_jobs: int) -> float:
         """Legacy linear reward: preserved for comparison/ablation."""
         context_avg = self._price_context_average(average_future_price)
@@ -203,32 +245,6 @@ def _penalty_idle_normalized(self, num_idle_nodes: int) -> float:
         normalized_penalty = -self._normalize(current_penalty, self._min_idle_penalty, self._max_idle_penalty)
         return float(np.clip(normalized_penalty, -1, 0))
 
-    @staticmethod
-    def _penalty_job_age(num_off_nodes: int, job_queue_2d: np.ndarray) -> float:
-        """Calculate saturated penalty for jobs waiting in queue when nodes are off."""
-        job_age_penalty = 0.0
-        if num_off_nodes > 0:
-            # Vectorized max age calculation (much faster than Python loop)
-            # [:, 0] selects column 0 (duration) for all rows; > 0 creates boolean mask
-            valid_mask = job_queue_2d[:, 0] > 0
-            # [valid_mask, 1] selects column 1 (age) only for rows where mask is True
-            max_age = job_queue_2d[valid_mask, 1].max() if valid_mask.any() else 0
-            if max_age > 0:
-                tau_hours = WEEK_HOURS / 2.0
-                max_factor = 1.0 - np.exp(-WEEK_HOURS / tau_hours)
-                factor = 1.0 - np.exp(-max_age / tau_hours)
-                factor = min(factor / max_factor, 1.0)
-                job_age_penalty = factor
-        return job_age_penalty
-
-    def _penalty_job_age_normalized(self, num_off_nodes: int, job_queue_2d: np.ndarray) -> float:
-        """Calculate normalized job age penalty [-1, 0]."""
-        current_penalty = self._penalty_job_age(num_off_nodes, job_queue_2d)
-        # _penalty_job_age already returns [0, 1]; negate to get [-1, 0]
-        # normalized_penalty = self._normalize(current_penalty, 0, -1)
-        normalized_penalty = -current_penalty
-        return float(np.clip(normalized_penalty, -1, 0))
-
     def _reward_energy_efficiency_normalized(self, num_used_nodes: int, num_idle_nodes: int) -> float:
         '''Redefine meaning of "efficiency". Use purely as "energy efficiency", aka: How much of the energy (in MW) which is currently needed, gets used for work.
         NOTE: Original efficiency function was doing 3 things at once. 1. Handled Blackout logic, with (2.) penalty-ish reward delay for unprocessed jobs, while blackout.
@@ -300,32 +316,118 @@ def _reward_price_utilization(self, current_price: float, average_future_price:
 
     def _blackout_term(self, num_used_nodes: int, num_idle_nodes: int, num_unprocessed_jobs: int) -> float:
         """
-        Reward/penalty for full blackout (all nodes off).
-        If queue is empty, reward the blackout. If jobs are waiting, apply a smooth penalty in [-1, 0].
+        Reward a full blackout only when there truly is no work to do.
+
+        The old implementation punished a queue during blackout immediately, which
+        collided with the benchmark objective of deferring expensive-hour work.
+        Deferral pressure now lives in the backlog term below instead of here.
         """
-        BLACKOUT_QUEUE_THRESHOLD = 10  # jobs waiting until penalty saturates to -1
-        SATURATION_FACTOR = 2
         on_nodes = num_used_nodes + num_idle_nodes
 
         if on_nodes != 0:
-            return 0.0  # only care about full blackout
+            return 0.0
+
+        return 1.0 if num_unprocessed_jobs <= 0 else 0.0
+
+    def _overdue_pressure(self, remaining_overdue_age_core_hours: float, tau: float) -> float:
+        """Smoothly map post-grace overdue mass into [0, 1]."""
+        if remaining_overdue_age_core_hours <= 0.0:
+            return 0.0
+        return float(
+            1.0 - np.exp(
+                -float(remaining_overdue_age_core_hours) / max(float(tau), 1e-6)
+            )
+        )
 
-        if num_unprocessed_jobs <= 0:
-            return 1.0  # correct blackout
+    def _penalty_job_age(
+        self,
+        current_price: float,
+        decision_pending_core_demand: float,
+        total_used_cores: int,
+    ) -> float:
+        """
+        Weighted cheap-hour service pressure used in the legacy "job age" slot.
 
-        ratio = num_unprocessed_jobs / max(BLACKOUT_QUEUE_THRESHOLD, 1)
-        penalty = np.exp(-ratio * SATURATION_FACTOR) - 1.0
-        return float(np.clip(penalty, -1.0, 0.0))
+        This term now only teaches *when* pending work should be served:
+        if the current hour is cheap and runnable demand exists, under-serving that
+        hour is penalized. Post-grace starvation is handled separately by an
+        intrinsic reward so the "do nothing" failure mode is structural rather than
+        weight-dependent.
+        """
+        cheap_strength, _ = self._price_phase_strengths(current_price)
+
+        cheap_service_shortfall = 0.0
+        if cheap_strength > 0.0 and decision_pending_core_demand > 0.0:
+            step_capacity_cores = float(MAX_NODES * CORES_PER_NODE)
+            target_service = min(float(decision_pending_core_demand), step_capacity_cores)
+            achieved_service = min(float(total_used_cores), target_service)
+            cheap_service_shortfall = cheap_strength * (1.0 - achieved_service / max(target_service, 1e-6))
+
+        return float(np.clip(self.CHEAP_SERVICE_GAIN * cheap_service_shortfall, 0.0, 1.0))
+
+    def _reward_intrinsic_starvation(
+        self,
+        remaining_overdue_age_core_hours: float,
+        total_used_cores: int,
+    ) -> float:
+        """
+        Intrinsic post-grace service-continuity reward.
+
+        Once pending work has moved past the 24h grace window, staying near zero
+        service becomes increasingly negative. Any useful work reduces this penalty
+        smoothly, which makes a full blackout with overdue backlog worse than at
+        least some utilization.
+        """
+        overdue_pressure = self._overdue_pressure(
+            remaining_overdue_age_core_hours,
+            self.INTRINSIC_STARVATION_OVERDUE_TAU,
+        )
+        if overdue_pressure <= 0.0:
+            return 0.0
+
+        zero_service_pressure = float(
+            np.exp(
+                -max(float(total_used_cores), 0.0)
+                / max(self.INTRINSIC_STARVATION_SERVICE_CORE_TAU, 1e-6)
+            )
+        )
+        return float(-self.INTRINSIC_STARVATION_GAIN * overdue_pressure * zero_service_pressure)
+
+    def _penalty_job_age_normalized(
+        self,
+        current_price: float,
+        decision_pending_core_demand: float,
+        total_used_cores: int,
+    ) -> float:
+        """Cheap-hour service pressure, normalized to [-1, 0]."""
+        current_penalty = self._penalty_job_age(
+            current_price,
+            decision_pending_core_demand,
+            total_used_cores,
+        )
+        return float(np.clip(-current_penalty, -1.0, 0.0))
 
     def _penalty_drop(self, num_dropped: int) -> float:
+        """Heavy penalty for jobs dropped this step."""
+        if num_dropped <= 0:
+            return 0.0
+        return float(-1.0 - 0.25 * min(num_dropped - 1, 1000))
+
+    def _penalty_drop_alt(self, num_dropped: int) -> float:
         """Drop penalty: tanh saturation curve bounded in [-1, 0]."""
         return -float(np.tanh(num_dropped / self.DROP_PENALTY_TAU))
+    def loss_penalty(self, num_lost: int) -> float:
+        """Weighted penalty for jobs lost this step, including streak-triggered flushes."""
+        if not self.ALLOW_DROP_PENALTY or num_lost <= 0:
+            return 0.0
+        return 0.3 * self._penalty_drop(num_lost)
 
     def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: float, average_future_price: float,
                   num_off_nodes: int, job_queue_2d: np.ndarray,
                   num_unprocessed_jobs: int, weights: Weights, num_dropped_this_step: int,
                   env_print: Callable[..., None], num_on_nodes: int,
-                  total_used_cores: int) -> tuple[float, float, float, float, float, float]:
+                  total_used_cores: int, decision_pending_core_demand: float = 0.0,
+                  remaining_overdue_age_core_hours: float = 0.0) -> tuple[float, float, float, float, float, float]:
         """
         Calculate total reward by aggregating weighted components.
 
@@ -343,6 +445,8 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo
             env_print: Print function for logging
             num_on_nodes: Number of powered-on nodes
             total_used_cores: Total cores in use across all powered nodes
+            decision_pending_core_demand: Total pending node-core demand before scheduling this step
+            remaining_overdue_age_core_hours: Post-scheduling overdue-age mass of still-pending jobs
 
         Returns:
             Tuple of (total reward, total cost, eff_reward_norm, price_reward, idle_penalty_norm, job_age_penalty_norm)
@@ -354,22 +458,36 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo
         efficiency_reward_norm += self._blackout_term(num_used_nodes, num_idle_nodes, num_unprocessed_jobs)
         efficiency_reward_weighted = weights.efficiency_weight * efficiency_reward_norm
 
-        # 2. Increase reward if current price is favorable and currently useful work is high.
-        price_reward = self._reward_price_utilization(current_price, average_future_price, total_used_cores)
         # legacy: price_reward = self._reward_price_normalized_legacy(current_price, average_future_price, total_used_cores)
+         # 2. Increase reward if current price is favorable and currently useful work is high.
+        price_reward = self._reward_price_utilization(current_price, average_future_price, total_used_cores)
         price_reward_weighted = weights.price_weight * price_reward
 
-        # 3. penalize delayed jobs, more if they are older. but only if there are turned off nodes
-        job_age_penalty_norm = self._penalty_job_age_normalized(num_off_nodes, job_queue_2d)
-        job_age_penalty_weighted = weights.job_age_weight * job_age_penalty_norm
+        # 3. Push pending work into cheap hours. The method name is kept for
+        # compatibility with existing plots/logs, but the semantics now describe
+        # cheap-hour service pressure rather than a simple "oldest queue age" penalty.
+        job_age_penalty_norm = self._penalty_job_age_normalized(
+            current_price,
+            decision_pending_core_demand,
+            total_used_cores,
+        )
+        
+        # 4. Intrinsic anti-starvation signal: once work is overdue, staying near
+        # zero throughput should be structurally worse than doing at least some work.
+        intrinsic_starvation_reward = self._reward_intrinsic_starvation(
+            remaining_overdue_age_core_hours,
+            total_used_cores,
+        )
+        
+        job_age_penalty_weighted = weights.job_age_weight * job_age_penalty_norm + intrinsic_starvation_reward
 
-        # 4. penalty for idling nodes
+        # 5. penalty for idling nodes
         idle_penalty_norm = self._penalty_idle_normalized(num_idle_nodes)
         idle_penalty_weighted = weights.idle_weight * idle_penalty_norm
 
-        # 5. penalty for lost jobs (aged out or rejected because queue/backlog was full)
-        drop_penalty = self._penalty_drop(num_dropped_this_step)
-        drop_penalty_weighted = weights.drop_weight * drop_penalty
+        # 6. penalty for lost jobs (aged out or rejected because queue/backlog was full)
+        # Streak-triggered flushes reuse the same penalty path via loss_penalty().
+        drop_penalty_weighted = self.loss_penalty(num_dropped_this_step)
 
         reward = (
             efficiency_reward_weighted
@@ -379,7 +497,12 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo
             + drop_penalty_weighted
         )
 
-        env_print(f"    > $$$TOTAL: {reward:.4f} = {efficiency_reward_weighted:.4f} + {price_reward_weighted:.4f} + {idle_penalty_weighted:.4f} + {job_age_penalty_weighted:.4f} + {drop_penalty_weighted:.4f}")
+        env_print(
+            f"    > $$$TOTAL: {reward:.4f} = "
+            f"{efficiency_reward_weighted:.4f} + {price_reward_weighted:.4f} + "
+            f"{idle_penalty_weighted:.4f} + {job_age_penalty_weighted:.4f} + "
+            f"{drop_penalty_weighted:.4f}"
+        )
         env_print(f"    > step cost: €{total_cost:.4f}")
 
         return reward, total_cost, efficiency_reward_norm, price_reward, idle_penalty_norm, job_age_penalty_norm
diff --git a/src/workloadgen.py b/src/workloadgen.py
index 57e4c70..364f86d 100644
--- a/src/workloadgen.py
+++ b/src/workloadgen.py
@@ -53,13 +53,13 @@ class WorkloadGenConfig:
     # Burst 1: many small-ish jobs at once
     burst_small_prob: float = 0.0
     burst_small_jobs_min: int = 50
-    burst_small_jobs_max: int = 750
+    burst_small_jobs_max: int = 1500
     burst_small_duration_min: int = 1
-    burst_small_duration_max: int = 8
+    burst_small_duration_max: int = 2
     burst_small_nodes_min: int = 1
-    burst_small_nodes_max: int = 2
+    burst_small_nodes_max: int = 1
     burst_small_cores_min: int = 1
-    burst_small_cores_max: int = 16
+    burst_small_cores_max: int = 4
 
     # Burst 2: heavy jobs (high duration + high resource demand)
     burst_heavy_prob: float = 0.0
diff --git a/train.py b/train.py
index e6f30c8..6ef5a0a 100644
--- a/train.py
+++ b/train.py
@@ -52,6 +52,7 @@ def main():
     parser.add_argument('--hourly-jobs', type=str, nargs='?', const="", default="", help='Path to Slurm log file for hourly statistical sampling (for use with hourly_sampler)')
     parser.add_argument('--job-arrival-scale', type=float, default=1.0, help='Scale sampled arrivals per step (1.0 = unchanged).')
     parser.add_argument('--jobs-exact-replay', action='store_true', help='For --jobs mode, replay raw jobs in timeline order (no template aggregation).')
+    parser.add_argument('--jobs-exact-replay-aggregate', action='store_true', help='With --jobs-exact-replay, aggregate each sampled raw time-bin before enqueueing.')
     parser.add_argument('--plot-rewards', action='store_true', help='Per step, plot rewards for all possible num_idle_nodes & num_used_nodes (default: False).')
     parser.add_argument('--plot-eff-reward', action=argparse.BooleanOptionalAction, default=False, help='Include efficiency reward in the plot (dashed line).')
     parser.add_argument('--plot-price-reward', action=argparse.BooleanOptionalAction, default=False, help='Include price reward in the plot (dashed line).')
@@ -76,6 +77,7 @@ def main():
     add_workloadgen_args(parser)
     parser.add_argument("--plot-dashboard", action="store_true", help="Generate dashboard plot (per-hour panels + cumulative savings).")
     parser.add_argument("--dashboard-hours", type=int, default=24*14, help="Hours to show in dashboard time-series panels (default: 336).")
+    parser.add_argument("--dashboard-interval", type=int, default=10000, help="Hours between dashboard plots (default: 10000).")
     parser.add_argument("--model", type=int, default=None, help="Load a specific model by timestep number (e.g. 5000000 loads 5000000.zip).")
     parser.add_argument("--net-arch", type=str, default="64,64", help="Hidden layer sizes for policy and value networks (comma-separated, e.g., '256,128' or '512,256,128')")
     parser.add_argument("--device", type=str, default="auto", help="Device for training: 'auto' (default, uses CUDA if available), 'cuda', 'cpu'")
@@ -83,6 +85,12 @@ def main():
     parser.add_argument("--seed-sweep", action="store_true", help="Treat this run as part of a --seeds sweep and isolate outputs under a seed-specific session subdirectory.")
     parser.add_argument("--print-policy", action="store_true", help="Print structure of the policy network.")
     parser.add_argument("--seed-path", default="", help="Path if models are saved by seed (forwarded to train.py) - only used by analyze_seed_occupancy.py, ignored otherwise.")
+    parser.add_argument(
+        "--flush-after-drop-streak",
+        type=int,
+        default=0,
+        help="Immediately flush and terminate the episode after this many consecutive dropped-job steps (0 disables).",
+    )
 
     args = parser.parse_args()
     try:
@@ -91,6 +99,8 @@ def main():
         parser.error(str(exc))
     if args.jobs_exact_replay and not norm_path(args.jobs):
         parser.error("--jobs-exact-replay requires --jobs")
+    if args.jobs_exact_replay_aggregate and not args.jobs_exact_replay:
+        parser.error("--jobs-exact-replay-aggregate requires --jobs-exact-replay")
     if args.workload_gen and args.job_arrival_scale != 1.0:
         print(
             "Warning: --job-arrival-scale is not allowed with --workload-gen; "
@@ -171,7 +181,11 @@ def main():
                             workload_gen=workload_gen,
                             job_arrival_scale=args.job_arrival_scale,
                             jobs_exact_replay=args.jobs_exact_replay,
-                            output_dir=args.output_dir)
+                            output_dir=args.output_dir,
+                            jobs_exact_replay_aggregate=args.jobs_exact_replay_aggregate,
+                            flush_after_drop_streak=args.flush_after_drop_streak)
+    env.session_dir = session_root
+    env.plots_dir = plots_dir
     env.reset(seed=args.seed)
 
     # Check if there are any saved models in models_dir
@@ -210,7 +224,7 @@ def main():
         )
         model = PPO.load(latest_model_file, env=env, tensorboard_log=log_dir, n_steps=64, batch_size=64, device=args.device)
     else:
-        print("Starting a new model training...")
+        print(f"Starting a new model training...")
         # Parse network architecture from comma-separated string (e.g., "256,128" -> [256, 128])
         net_arch_layers = [int(x) for x in args.net_arch.split(',')]
         policy_kwargs = dict(
@@ -245,7 +259,7 @@ def main():
             print("Train a model first, then run evaluation mode.")
             return
 
-        print("=== EVALUATION MODE ===")
+        print(f"=== EVALUATION MODE ===")
         print(f"Evaluation period: {args.eval_months} months ({args.eval_months * 2} episodes, Each episode = 2 weeks)")
         if evaluation_plots_dir is None:
             raise RuntimeError("Evaluation plots directory could not be determined for the selected model.")
@@ -297,26 +311,30 @@ def main():
             results = plot_cumulative_savings(env, env.metrics.episode_costs, session_dir, save=True, show=args.render == 'human')
             plot_episode_summary(env, env.metrics.episode_costs, session_dir, save=True, show=args.render == 'human', suffix=f"eval_{args.eval_months}m")
             if results:
-                print("\n=== CUMULATIVE SAVINGS ANALYSIS ===")
-                print("\nVs Baseline (with idle nodes):")
+                print(f"\n=== CUMULATIVE SAVINGS ANALYSIS ===")
+                print(f"\nVs Baseline (with idle nodes):")
                 print(f"  Total Savings: €{results['total_savings']:,.0f}")
                 print(f"  Average Monthly Reduction: {results['avg_monthly_savings_pct']:.1f}%")
                 print(f"  Annual Savings Rate: €{results['total_savings'] * 12 / args.eval_months:,.0f}/year")
 
-                print("\nVs Baseline_off (no idle nodes):")
+                print(f"\nVs Baseline_off (no idle nodes):")
                 print(f"  Total Savings: €{results['total_savings_off']:,.0f}")
                 print(f"  Average Monthly Reduction: {results['avg_monthly_savings_pct_off']:.1f}%")
                 print(f"  Annual Savings Rate: €{results['total_savings_off'] * 12 / args.eval_months:,.0f}/year")
 
                 # Calculate job metrics across all episodes
                 total_jobs_submitted = sum(ep['jobs_submitted'] for ep in env.metrics.episode_costs)
+                total_jobs_launched = sum(int(ep.get('jobs_launched', ep['jobs_completed'])) for ep in env.metrics.episode_costs)
                 total_jobs_completed = sum(ep['jobs_completed'] for ep in env.metrics.episode_costs)
                 total_baseline_submitted = sum(ep['baseline_jobs_submitted'] for ep in env.metrics.episode_costs)
+                total_baseline_launched = sum(int(ep.get('baseline_jobs_launched', ep['baseline_jobs_completed'])) for ep in env.metrics.episode_costs)
                 total_baseline_completed = sum(ep['baseline_jobs_completed'] for ep in env.metrics.episode_costs)
-                avg_wait_time = sum(ep['avg_wait_time'] * ep['jobs_completed'] for ep in env.metrics.episode_costs) / total_jobs_completed if total_jobs_completed > 0 else 0
-                avg_baseline_wait_time = sum(ep['baseline_avg_wait_time'] * ep['baseline_jobs_completed'] for ep in env.metrics.episode_costs) / total_baseline_completed if total_baseline_completed > 0 else 0
+                avg_wait_time = sum(ep['avg_wait_time'] * int(ep.get('jobs_launched', ep['jobs_completed'])) for ep in env.metrics.episode_costs) / total_jobs_launched if total_jobs_launched > 0 else 0
+                avg_baseline_wait_time = sum(ep['baseline_avg_wait_time'] * int(ep.get('baseline_jobs_launched', ep['baseline_jobs_completed'])) for ep in env.metrics.episode_costs) / total_baseline_launched if total_baseline_launched > 0 else 0
                 avg_max_queue = sum(ep['max_queue_size'] for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs)
                 avg_baseline_max_queue = sum(ep['baseline_max_queue_size'] for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs)
+                avg_pending_jobs_end = sum(int(ep.get('pending_jobs_end', 0)) for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs)
+                avg_overdue_jobs_end = sum(int(ep.get('overdue_jobs_end', 0)) for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs)
                 total_agent_cost = sum(float(ep['agent_cost']) for ep in env.metrics.episode_costs)
                 total_baseline_cost = sum(float(ep['baseline_cost']) for ep in env.metrics.episode_costs)
                 total_baseline_off_cost = sum(float(ep['baseline_cost_off']) for ep in env.metrics.episode_costs)
@@ -325,9 +343,22 @@ def main():
                 total_agent_power_mwh = sum(float(ep.get('agent_power_consumption_mwh', 0.0)) for ep in env.metrics.episode_costs)
                 total_baseline_power_mwh = sum(float(ep.get('baseline_power_consumption_mwh', 0.0)) for ep in env.metrics.episode_costs)
                 total_baseline_off_power_mwh = sum(float(ep.get('baseline_power_consumption_off_mwh', 0.0)) for ep in env.metrics.episode_costs)
+                total_agent_prop_power_mwh = sum(float(ep.get('agent_prop_power_mwh', 0.0)) for ep in env.metrics.episode_costs)
+                total_baseline_prop_power_mwh = sum(float(ep.get('baseline_prop_power_mwh', 0.0)) for ep in env.metrics.episode_costs)
+                total_baseline_off_prop_power_mwh = sum(float(ep.get('baseline_off_prop_power_mwh', 0.0)) for ep in env.metrics.episode_costs)
+                total_agent_prop_cost = sum(float(ep.get('agent_prop_cost', 0.0)) for ep in env.metrics.episode_costs)
+                total_baseline_prop_cost = sum(float(ep.get('baseline_prop_cost', 0.0)) for ep in env.metrics.episode_costs)
+                total_baseline_off_prop_cost = sum(float(ep.get('baseline_off_prop_cost', 0.0)) for ep in env.metrics.episode_costs)
+                total_savings_prop_cost_vs_baseline = total_baseline_prop_cost - total_agent_prop_cost
+                total_savings_prop_cost_vs_baseline_off = total_baseline_off_prop_cost - total_agent_prop_cost
                 total_agent_mean_price = (total_agent_cost / total_agent_power_mwh) if total_agent_power_mwh > 0 else 0.0
                 total_baseline_mean_price = (total_baseline_cost / total_baseline_power_mwh) if total_baseline_power_mwh > 0 else 0.0
                 total_baseline_off_mean_price = (total_baseline_off_cost / total_baseline_off_power_mwh) if total_baseline_off_power_mwh > 0 else 0.0
+                total_agent_prop_mean_price = (total_agent_prop_cost / total_agent_prop_power_mwh) if total_agent_prop_power_mwh > 0 else 0.0
+                total_baseline_prop_mean_price = (total_baseline_prop_cost / total_baseline_prop_power_mwh) if total_baseline_prop_power_mwh > 0 else 0.0
+                total_baseline_off_prop_mean_price = (total_baseline_off_prop_cost / total_baseline_off_prop_power_mwh) if total_baseline_off_prop_power_mwh > 0 else 0.0
+                prop_savings_pct_vs_baseline = safe_ratio(total_savings_prop_cost_vs_baseline * 100.0, total_baseline_prop_cost)
+                prop_savings_pct_vs_baseline_off = safe_ratio(total_savings_prop_cost_vs_baseline_off * 100.0, total_baseline_off_prop_cost)
                 total_agent_completion_rate = (total_jobs_completed / total_jobs_submitted * 100) if total_jobs_submitted > 0 else 0.0
                 total_baseline_completion_rate = (total_baseline_completed / total_baseline_submitted * 100) if total_baseline_submitted > 0 else 0.0
                 total_savings_vs_baseline = total_baseline_cost - total_agent_cost
@@ -345,40 +376,44 @@ def main():
                 ) if arrivals_per_hour_by_episode else 0.0
                 std_arrivals_per_hour = arrivals_variance ** 0.5
 
-                print("\n=== JOB PROCESSING METRICS ===")
-                print("\nAgent:")
+                print(f"\n=== JOB PROCESSING METRICS ===")
+                print(f"\nAgent:")
+                print(f"  Jobs Launched: {total_jobs_launched:,} / {total_jobs_submitted:,}")
                 print(f"  Jobs Completed: {total_jobs_completed:,} / {total_jobs_submitted:,} ({total_agent_completion_rate:.1f}%)")
                 print(f"  Average Wait Time: {avg_wait_time:.1f} hours")
                 print(f"  Average Max Queue Size: {avg_max_queue:.0f}")
+                print(f"  Average Pending Jobs At Episode End: {avg_pending_jobs_end:.1f}")
+                print(f"  Average Overdue Jobs At Episode End: {avg_overdue_jobs_end:.1f}")
                 print(f"  Total Cost: €{total_agent_cost:,.0f}")
                 print(f"  Job Arrivals/Hour (mean ± std): {mean_arrivals_per_hour:.2f} ± {std_arrivals_per_hour:.2f}")
 
-                print("\nBaseline:")
+                print(f"\nBaseline:")
+                print(f"  Jobs Launched: {total_baseline_launched:,} / {total_baseline_submitted:,}")
                 print(f"  Jobs Completed: {total_baseline_completed:,} / {total_baseline_submitted:,} ({total_baseline_completion_rate:.1f}%)")
                 print(f"  Average Wait Time: {avg_baseline_wait_time:.1f} hours")
                 print(f"  Average Max Queue Size: {avg_baseline_max_queue:.0f}")
                 print(f"  Baseline Total Cost: €{total_baseline_cost:,.0f}")
                 print(f"  Baseline_off Total Cost: €{total_baseline_off_cost:,.0f}")
 
-                print("\n=== COST PER 1,000 COMPLETED JOBS ===")
+                print(f"\n=== COST PER 1,000 COMPLETED JOBS ===")
                 print(f"  Agent:        {fmt_optional(total_agent_cost_per_1000_completed, 2, thousands=True)} €/1k jobs")
                 print(f"  Baseline:     {fmt_optional(total_baseline_cost_per_1000_completed, 2, thousands=True)} €/1k jobs")
                 print(f"  Baseline_off: {fmt_optional(total_baseline_off_cost_per_1000_completed, 2, thousands=True)} €/1k jobs")
 
-                print("\n=== AGENT LOST JOBS PER SAVED EURO ===")
+                print(f"\n=== AGENT LOST JOBS PER SAVED EURO ===")
                 print(f"  Total Lost Jobs (Agent): {total_jobs_dropped:,}")
                 print(f"  Total Lost Jobs (Baseline): {total_baseline_jobs_dropped:,}")
                 print(f"  Vs Baseline:     {fmt_optional(total_dropped_jobs_per_saved_euro, 6)} jobs/€")
                 print(f"  Vs Baseline_off: {fmt_optional(total_dropped_jobs_per_saved_euro_off, 6)} jobs/€")
 
 
-                print("\n=== POWER & PRICE METRICS (TOTAL OVER EVALUATION) ===")
-                print(f"  Agent:        Power={total_agent_power_mwh:,.1f} MWh, Mean Price={total_agent_mean_price:.2f} €/MWh")
-                print(f"  Baseline:     Power={total_baseline_power_mwh:,.1f} MWh, Mean Price={total_baseline_mean_price:.2f} €/MWh")
-                print(f"  Baseline_off: Power={total_baseline_off_power_mwh:,.1f} MWh, Mean Price={total_baseline_off_mean_price:.2f} €/MWh")
-                print("\n=== COST SAVINGS (TOTAL OVER EVALUATION) ===")
-                print(f"  Vs Baseline:     €{total_savings_vs_baseline:,.0f}, {fmt_optional(safe_ratio(total_savings_vs_baseline * 100.0, total_baseline_cost), 1)}%")
-                print(f"  Vs Baseline_off: €{total_savings_vs_baseline_off:,.0f}, {fmt_optional(safe_ratio(total_savings_vs_baseline_off * 100.0, total_baseline_off_cost), 1)}%")
+                print(f"\n=== POWER & PRICE METRICS (TOTAL OVER EVALUATION) ===")
+                print(f"  Agent:        Power={total_agent_prop_power_mwh:,.1f} MWh, Mean Price={total_agent_prop_mean_price:.2f} €/MWh")
+                print(f"  Baseline:     Power={total_baseline_prop_power_mwh:,.1f} MWh, Mean Price={total_baseline_prop_mean_price:.2f} €/MWh")
+                print(f"  Baseline_off: Power={total_baseline_off_prop_power_mwh:,.1f} MWh, Mean Price={total_baseline_off_prop_mean_price:.2f} €/MWh")
+                print(f"\n=== PROPORTIONAL COST SAVINGS (TOTAL OVER EVALUATION) ===")
+                print(f"  Vs Baseline:     €{total_savings_prop_cost_vs_baseline:,.0f}, {fmt_optional(prop_savings_pct_vs_baseline, 1)}%")
+                print(f"  Vs Baseline_off: €{total_savings_prop_cost_vs_baseline_off:,.0f}, {fmt_optional(prop_savings_pct_vs_baseline_off, 1)}%")
         except Exception as e:
             print(f"Could not generate cumulative savings plot: {e}")
 
@@ -412,11 +447,7 @@ def main():
                 print(f"iterations limit ({args.iter_limit}) reached: {iters}.")
                 break
             try:
-                model.learn(total_timesteps=STEPS_PER_ITERATION, reset_num_timesteps=False, tb_log_name=f"PPO", callback=ComputeClusterCallback())
-                print(f"Iteration {iters} finished in {time.time()-t0:.2f}s")
-                model.save(f"{models_dir}/{STEPS_PER_ITERATION * iters}.zip")
-
-                if args.plot_dashboard:
+                if args.plot_dashboard and iters > 1 and (STEPS_PER_ITERATION * (iters - 1)) % args.dashboard_interval == 0:  # Only plot after the first iteration to avoid empty data
                     try:
                         plot_dashboard(
                             env,
@@ -428,6 +459,11 @@ def main():
                         )
                     except Exception as e:
                         print(f"Dashboard plot failed (non-fatal): {e}")
+                        
+                model.learn(total_timesteps=STEPS_PER_ITERATION, reset_num_timesteps=False, tb_log_name=f"PPO", callback=ComputeClusterCallback())
+                print(f"Iteration {iters} finished in {time.time()-t0:.2f}s")
+                model.save(f"{models_dir}/{STEPS_PER_ITERATION * iters}.zip")
+
 
             except PlottingComplete:
                 print("Plotting complete, terminating training...")
diff --git a/train_iter.py b/train_iter.py
index ca601aa..a758e82 100644
--- a/train_iter.py
+++ b/train_iter.py
@@ -118,6 +118,7 @@ def build_command(
     seed_sweep=False,
     evaluate_savings=False,
     eval_months=0,
+    flush_after_drop_streak=0,
     workloadgen_args=None,
     output_dir=None,
 ):
@@ -147,6 +148,8 @@ def build_command(
         command += ["--seed-sweep"]
     if evaluate_savings:
         command += ["--evaluate-savings", "--eval-months", str(eval_months)]
+    if flush_after_drop_streak > 0:
+        command += ["--flush-after-drop-streak", str(flush_after_drop_streak)]
     if workloadgen_args:
         command += workloadgen_args
     if output_dir is not None:
@@ -366,8 +369,8 @@ def _reap(proc, label, fh, elapsed):
 def run_all_parallel(combinations, max_parallel, iter_limit_per_step, session, prices,
                      job_durations, jobs, hourly_jobs, job_arrival_scale, jobs_exact_replay,
                      plot_dashboard, dashboard_hours,
-                     seeds, seed_sweep, evaluate_savings, eval_months, workloadgen_args,
-                     no_tui=False, output_dir=None):
+                     seeds, seed_sweep, evaluate_savings, eval_months, flush_after_drop_streak, workloadgen_args,
+                     no_tui=False):
     multi_seed = len(seeds) > 1
     current_env = os.environ.copy()
     log_dir = make_log_dir(session, output_dir or "sessions")
@@ -383,8 +386,7 @@ def launch(combo, seed):
             iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs,
             job_arrival_scale, jobs_exact_replay,
             plot_dashboard, dashboard_hours, seed, seed_sweep,
-            evaluate_savings, eval_months, workloadgen_args,
-            output_dir=output_dir,
+            evaluate_savings, eval_months, flush_after_drop_streak, workloadgen_args,
         )
         log_path = os.path.join(log_dir, label_to_filename(label))
         log_fh = open(log_path, "w")
@@ -443,6 +445,12 @@ def main():
     parser.add_argument("--parallel", type=int, default=1, metavar="N", help="Number of training runs to execute in parallel (default: 1, sequential)")
     parser.add_argument("--evaluate-savings", action="store_true", help="Forward to train.py to evaluate savings compared to baseline.")
     parser.add_argument("--eval-months", type=int, default=6, help="Number of months to evaluate savings over (forwarded to train.py)")
+    parser.add_argument(
+        "--flush-after-drop-streak",
+        type=int,
+        default=0,
+        help="Forward to train.py: immediately flush and terminate the episode after this many consecutive dropped-job steps (0 disables).",
+    )
     parser.add_argument("--no-tui", action="store_true", help="Disable interactive TUI; print plain progress lines instead (auto-disabled when not a TTY)")
     add_workloadgen_args(parser)
 
@@ -508,6 +516,7 @@ def main():
         evaluate_savings=args.evaluate_savings,
         eval_months=args.eval_months,
         workloadgen_args=workloadgen_args,
+        flush_after_drop_streak=args.flush_after_drop_streak,
         no_tui=args.no_tui,
         output_dir=args.output_dir,
     )