Fix #2245: rebaseline post-consensus-timeout budget on producer progress

orchestrator/models.py

@@ -377,6 +377,30 @@ class PipelineConfig(BaseModel):
     consensus_timeout_minutes: int = Field(
         default=30, ge=1, description="Timeout for consensus before HITL escalation"
     )
+    post_consensus_iteration_budget_seconds: int = Field(
+        default=3600,
+        ge=60,
+        description=(
+            "After consensus_timeout_minutes elapses, the post-timeout poll loop "
+            "waits up to this many seconds without producer progress before "
+            "force-killing remaining containers (issue #2245). The clock "
+            "rebaselines whenever a producer issues a new CONSENSUS_PROPOSE "
+            "(initial propose or NACK→re-propose), so a healthy multi-iteration "
+            "BRC cycle no longer counts iteration time against a single fixed "
+            "budget."
+        ),
+    )
+    post_consensus_max_total_seconds: int = Field(
+        default=14400,
+        ge=60,
+        description=(
+            "Hard ceiling on the post-consensus-timeout wait, in seconds. "
+            "Caps the total time spent in the post-timeout poll loop even when "
+            "producer progress keeps rebaselining the per-iteration budget — "
+            "prevents an unbounded loop if propose events keep arriving but "
+            "consensus never converges. Default 4 hours."
+        ),
+    )
     brc_consensus_progress_gate_seconds: int = Field(
         default=300,
         ge=0,
@@ -605,6 +629,23 @@ class PipelineConfig(BaseModel):
         ),
     )
 
+    @model_validator(mode="after")
+    def _validate_post_consensus_budgets(self) -> "PipelineConfig":
+        """Reject configs where the absolute cap is below the per-iteration budget.
+
+        Without this, a misconfigured pipeline (e.g. ``iteration_budget=7200``
+        with ``max_total=3600``) silently makes the per-iteration logic
+        unreachable — the absolute cap would always fire first. See #2245.
+        """
+        if self.post_consensus_max_total_seconds < self.post_consensus_iteration_budget_seconds:
+            raise ValueError(
+                "post_consensus_max_total_seconds "
+                f"({self.post_consensus_max_total_seconds}) must be >= "
+                "post_consensus_iteration_budget_seconds "
+                f"({self.post_consensus_iteration_budget_seconds})"
+            )
+        return self
+
     @model_validator(mode="before")
     @classmethod
     def _alias_post_propose_grace(cls, data: Any) -> Any:

orchestrator/routes/pipelines.py

@@ -10795,6 +10795,28 @@ def _stop_running_containers() -> None:
             get_peer_consensus_tracker as _get_brc_tracker,  # type: ignore[no-redef]
         )
 
+    def _latest_proposal_ts(_pid: str, _sid: str | None) -> "datetime | None":
+        """Return the latest CONSENSUS_PROPOSE timestamp from the BRC tracker.
+
+        Used by the post-consensus-timeout poll loop (#2245) to rebaseline
+        the per-iteration budget on producer progress.  Returns ``None`` if
+        the tracker is unavailable, has no proposals, or any lookup raises —
+        callers treat ``None`` as "no progress signal yet" and proceed
+        without a rebaseline.
+        """
+        if _get_brc_tracker is None:
+            return None
+        try:
+            _t = _get_brc_tracker(_pid, _sid)
+        except Exception:
+            return None
+        if _t is None:
+            return None
+        try:
+            return _t.get_latest_proposal_timestamp()
+        except Exception:
+            return None
+
     def _update_agents_complete() -> None:
         """Mark all running agents as COMPLETE in pipeline state (consensus path)."""
         if store is None:
@@ -11369,15 +11391,52 @@ def _update_agents_complete() -> None:
             # container status in short steps and re-check consensus
             # between steps, early-returning on completion before
             # force-killing anything.
+            #
+            # Issue #2245: the per-iteration budget rebaselines on
+            # producer progress.  Each new CONSENSUS_PROPOSE (initial
+            # or NACK→re-propose) resets ``last_progress_at`` so the
+            # producer's next iteration gets a clean clock instead of
+            # inheriting the prior iterations' wall-clock spend.  An
+            # absolute cap (``post_consensus_max_total_seconds``)
+            # bounds the total wait so an unbounded propose churn
+            # can't stall the pipeline indefinitely.
             remaining = [e for e in active_executions if e.container_id not in exited_containers]
             if remaining:
-                post_timeout_budget = 3600  # seconds total
+                post_timeout_iteration_budget = (
+                    pipeline.config.post_consensus_iteration_budget_seconds
+                )
+                post_timeout_max_total = pipeline.config.post_consensus_max_total_seconds
                 post_timeout_poll_interval = 30  # seconds between checks
                 post_timeout_start = time.monotonic()
+                last_progress_at = post_timeout_start
+
+                # Snapshot the latest proposal timestamp at entry so we
+                # only count *new* proposals as progress signals.  ``None``
+                # is fine: the rebaseline check at the bottom of the loop
+                # short-circuits on ``last_seen_proposal_ts is None``
+                # before any datetime comparison runs.
+                last_seen_proposal_ts = _latest_proposal_ts(pipeline_id, slice_id)
 
                 while remaining:
-                    elapsed_post_timeout = time.monotonic() - post_timeout_start
-                    if elapsed_post_timeout >= post_timeout_budget:
+                    now_monotonic = time.monotonic()
+                    total_elapsed = now_monotonic - post_timeout_start
+                    iteration_elapsed = now_monotonic - last_progress_at
+                    if total_elapsed >= post_timeout_max_total:
+                        logger.warning(
+                            "Post-consensus-timeout absolute cap reached",
+                            pipeline_id=pipeline_id,
+                            total_elapsed_seconds=round(total_elapsed, 1),
+                            max_total_seconds=post_timeout_max_total,
+                        )
+                        break
+                    if iteration_elapsed >= post_timeout_iteration_budget:
+                        logger.warning(
+                            "Post-consensus-timeout iteration budget exhausted",
+                            pipeline_id=pipeline_id,
+                            iteration_elapsed_seconds=round(iteration_elapsed, 1),
+                            iteration_budget_seconds=post_timeout_iteration_budget,
+                            total_elapsed_seconds=round(total_elapsed, 1),
+                        )
                         break
 
                     # A. Re-check consensus; if agents converged during
@@ -11409,13 +11468,33 @@ def _update_agents_complete() -> None:
                         logger.info(
                             "Consensus reached during post-timeout wait",
                             pipeline_id=pipeline_id,
-                            elapsed_post_timeout_seconds=round(elapsed_post_timeout, 1),
+                            elapsed_post_timeout_seconds=round(total_elapsed, 1),
                             total_elapsed_seconds=round(_total_elapsed, 1),
                         )
                         _update_agents_complete()
                         _stop_running_containers()
                         return 0, combined_logs
 
+                    # A'. Rebaseline the iteration clock on producer
+                    # progress (#2245).  A fresh CONSENSUS_PROPOSE
+                    # timestamp means a producer just landed work
+                    # (initial propose or NACK→re-propose) — the next
+                    # round of reviews deserves its own iteration
+                    # budget, not whatever's left of the prior round's.
+                    current_proposal_ts = _latest_proposal_ts(pipeline_id, slice_id)
+                    if current_proposal_ts is not None and (
+                        last_seen_proposal_ts is None or current_proposal_ts > last_seen_proposal_ts
+                    ):
+                        logger.info(
+                            "Post-consensus-timeout clock rebaselined on producer progress",
+                            pipeline_id=pipeline_id,
+                            iteration_elapsed_seconds=round(iteration_elapsed, 1),
+                            total_elapsed_seconds=round(total_elapsed, 1),
+                            proposal_timestamp=current_proposal_ts.isoformat(),
+                        )
+                        last_seen_proposal_ts = current_proposal_ts
+                        last_progress_at = time.monotonic()
+
                     # B. Non-blocking container status check; record
                     # any that have exited naturally.
                     still_running = []

orchestrator/tests/test_consensus_polling.py

@@ -299,17 +299,17 @@ def test_timeout_creates_hitl_decision(
         """When consensus times out, a HITL decision is created."""
         # Use a callable side_effect that (a) starts before the timeout,
         # (b) jumps past the 30-min consensus timeout, and (c) keeps
-        # advancing so the post-timeout polling budget (#1921) also
-        # exhausts within a bounded number of iterations.
+        # advancing so the post-timeout per-iteration budget (#1921 /
+        # #2245) also exhausts within a bounded number of iterations.
         _calls = [0]
 
         def _monotonic():
             _calls[0] += 1
             if _calls[0] == 1:
                 return 0.0
             # Each subsequent call jumps 2000s so both the 1800s
-            # consensus timeout and the 3600s post-timeout budget
-            # elapse quickly.
+            # consensus timeout and the default 3600s per-iteration
+            # post-timeout budget elapse quickly.
             return float(1801.0 + _calls[0] * 2000.0)
 
         mock_monotonic.side_effect = _monotonic
@@ -343,8 +343,10 @@ def _monotonic():
         # Timeout with no convergence → force-kill path → exit 1 (#1921).
         # Prior to #1921 this returned 0 because wait_for_container was
         # mocked to return a clean exit; post-#1921 the polling loop
-        # force-kills still-running containers when the 3600s budget
-        # elapses, which is the realistic outcome.
+        # force-kills still-running containers when the per-iteration
+        # post-timeout budget elapses (#2245: budget rebaselines on
+        # producer progress; with no tracker proposals here the
+        # iteration clock never resets).
         assert exit_code == 1
         # HITL decision should have been created on the pipeline
         mock_add_decision.assert_called_once()
@@ -365,7 +367,8 @@ def test_timeout_emits_event(
         from events import EventType
 
         # See test_timeout_creates_hitl_decision for why monotonic must
-        # keep advancing past the 3600s post-timeout budget (#1921).
+        # keep advancing past the post-timeout per-iteration budget
+        # (#1921 / #2245).
         _calls = [0]
 
         def _monotonic():