Fix #2068: BRC heartbeat refreshes gateway session liveness

gateway/gateway.py

@@ -5543,6 +5543,33 @@ def session_delete_by_container(container_id: str) -> tuple[Response, int] | Res
     return make_success("Session deleted")
 
 
+@app.route("/api/v1/sessions/by-container/<container_id>/heartbeat", methods=["POST"])
+@require_launcher_auth
+def session_heartbeat_by_container(container_id: str) -> tuple[Response, int] | Response:
+    """
+    Refresh a session's idle timer by container ID (orchestrator-only path).
+
+    Used by the orchestrator to keep agent sessions alive while their
+    container is heartbeating on the BRC bus but not making gateway
+    requests — without this, the idle pruner evicts the session after
+    EGG_SESSION_IDLE_TIMEOUT_MINUTES even though the agent is still
+    working (see #2068).
+
+    Auth: Bearer {launcher_secret}
+
+    Returns 404 if no session exists for the container.  No per-session
+    rate limit because the launcher secret already gates access — only
+    the orchestrator can call this.
+    """
+    session_manager = get_session_manager()
+    refreshed = session_manager.heartbeat_session_by_container(container_id)
+
+    if not refreshed:
+        return make_error("Session not found for container", status_code=404)
+
+    return make_success("Heartbeat recorded")
+
+
 @app.route("/api/v1/sessions/<session_token>", methods=["GET"])
 @require_launcher_auth
 def session_get(session_token: str) -> tuple[Response, int] | Response:

gateway/session_manager.py

@@ -941,6 +941,39 @@ def delete_session_by_container(self, container_id: str) -> tuple[bool, threadin
 
         return False, None
 
+    def heartbeat_session_by_container(self, container_id: str) -> bool:
+        """Refresh a session's ``last_seen`` (and TTL) by container ID.
+
+        Used by the orchestrator to keep gateway sessions alive while an
+        agent is heartbeating through the BRC bus but not making gateway
+        requests (e.g. a producer in ``WAITING_FOR_EVENT`` during a long
+        review cycle).  Without this, the idle pruner evicts the session
+        after ``DEFAULT_SESSION_IDLE_TIMEOUT_MINUTES`` even though the
+        agent is still working — see #2068.
+
+        Mirrors ``validate_session``'s in-flight pattern: the TTL extend
+        is in-memory only.  We do **not** ``_save_to_disk()`` here — at
+        ~1 fan-out per agent every 60s a per-call atomic file write
+        would dominate gateway disk I/O for no benefit (worst-case loss
+        on a gateway crash is the session ages out and is re-registered
+        on the next gateway op, which is the same recovery path
+        ``validate_session`` already relies on).  Disk persistence
+        happens at lifecycle events (registration, deletion, expiry).
+
+        Args:
+            container_id: Container ID whose session to refresh.
+
+        Returns:
+            True if a matching, non-expired session was refreshed; False
+            if no session exists for the container.
+        """
+        with self._lock:
+            for session in self._sessions.values():
+                if session.container_id == container_id and not session.is_expired():
+                    session.extend_ttl(self._ttl_hours)
+                    return True
+        return False
+
     def prune_expired_sessions(self) -> int:
         """
         Remove all expired sessions.

gateway/tests/test_gateway.py

@@ -5118,6 +5118,46 @@ def test_session_delete_by_container_not_found(self, client, launcher_auth_heade
             assert response.status_code == 404
 
 
+class TestSessionHeartbeatByContainer:
+    """Tests for POST /api/v1/sessions/by-container/<id>/heartbeat (#2068)."""
+
+    def test_refreshes_session(self, client, launcher_auth_headers):
+        mock_session_mgr = MagicMock()
+        mock_session_mgr.heartbeat_session_by_container.return_value = True
+
+        with patch.object(gateway, "get_session_manager", return_value=mock_session_mgr):
+            response = client.post(
+                "/api/v1/sessions/by-container/egg-agent-pipeline-1-coder/heartbeat",
+                headers=launcher_auth_headers,
+            )
+
+        assert response.status_code == 200
+        data = json.loads(response.data)
+        assert data["success"] is True
+        mock_session_mgr.heartbeat_session_by_container.assert_called_once_with(
+            "egg-agent-pipeline-1-coder"
+        )
+
+    def test_returns_404_when_session_missing(self, client, launcher_auth_headers):
+        mock_session_mgr = MagicMock()
+        mock_session_mgr.heartbeat_session_by_container.return_value = False
+
+        with patch.object(gateway, "get_session_manager", return_value=mock_session_mgr):
+            response = client.post(
+                "/api/v1/sessions/by-container/nonexistent/heartbeat",
+                headers=launcher_auth_headers,
+            )
+
+        assert response.status_code == 404
+
+    def test_requires_launcher_auth(self, client):
+        """Endpoint must reject requests without the launcher secret."""
+        response = client.post(
+            "/api/v1/sessions/by-container/egg-agent-pipeline-1-coder/heartbeat",
+        )
+        assert response.status_code == 401
+
+
 class TestBranchIsolation:
     """Tests for branch isolation enforcement in pipeline worktree sessions.
 

gateway/tests/test_session_manager.py

@@ -779,6 +779,84 @@ def test_delete_clears_token_cache(self, manager):
         assert not result.valid
 
 
+class TestHeartbeatByContainer:
+    """Tests for heartbeat_session_by_container (#2068)."""
+
+    @pytest.fixture
+    def manager(self, tmp_path):
+        return SessionManager(persistence_file=tmp_path / "sessions.json")
+
+    def test_heartbeat_refreshes_last_seen(self, manager):
+        """A heartbeat call advances last_seen so the idle pruner spares the session."""
+        _token, session = manager.register_session(
+            container_id="egg-agent-pipeline-1-coder",
+            container_ip="172.18.0.5",
+            mode="private",
+        )
+
+        # Backdate last_seen past the idle timeout window.
+        session.last_seen = datetime.now(UTC) - timedelta(minutes=120)
+
+        refreshed = manager.heartbeat_session_by_container("egg-agent-pipeline-1-coder")
+        assert refreshed is True
+
+        # Session is now fresh -- not pruned by a 60-minute idle sweep.
+        assert manager.prune_idle_sessions(idle_timeout_minutes=60) == 0
+        assert manager.get_session_by_container("egg-agent-pipeline-1-coder") is not None
+
+    def test_heartbeat_unknown_container(self, manager):
+        """Heartbeat for an unknown container returns False without raising."""
+        assert manager.heartbeat_session_by_container("not-a-real-container") is False
+
+    def test_heartbeat_extends_ttl(self, manager):
+        """Heartbeat advances expires_at to ``now + ttl`` (mirrors validate_session).
+
+        The 23h54m floor (rather than e.g. 23h59m) is a deliberate 6-minute
+        slack to absorb test-execution jitter between ``register_session``
+        and the post-heartbeat ``datetime.now(UTC)`` call: a stalled
+        runner could in principle take a few seconds between the two,
+        and the assertion only needs to prove ``extend_ttl`` actually
+        fired (not that the clock was perfectly still).  The strict
+        ``>`` against ``backdated_expiry`` (``now + 23.5h``) is what
+        proves the heartbeat moved the deadline forward — the 23h54m
+        floor is the *upper-bound* sanity check that it landed near
+        ``now + 24h`` rather than being clamped lower.
+        """
+        _token, session = manager.register_session(
+            container_id="egg-agent-pipeline-1-coder",
+            container_ip="172.18.0.5",
+            mode="private",
+        )
+        # Backdate both ``last_seen`` and ``expires_at`` 30 minutes into
+        # the past.  ``extend_ttl`` should pull ``expires_at`` forward to
+        # roughly ``now + 24h`` — strictly greater than the backdated
+        # value of ``now + 23.5h``.
+        backdated_last_seen = datetime.now(UTC) - timedelta(minutes=30)
+        session.last_seen = backdated_last_seen
+        session.expires_at = backdated_last_seen + timedelta(hours=24)
+        backdated_expiry = session.expires_at
+
+        manager.heartbeat_session_by_container("egg-agent-pipeline-1-coder")
+        refreshed_session = manager.get_session_by_container("egg-agent-pipeline-1-coder")
+        assert refreshed_session is not None
+        # Strict ``>`` proves the heartbeat moved ``expires_at`` forward;
+        # 23h54m floor proves it landed near ``now + 24h`` (sanity check
+        # — see docstring for the 6-minute slack rationale).
+        assert refreshed_session.expires_at > backdated_expiry
+        assert refreshed_session.expires_at > datetime.now(UTC) + timedelta(hours=23, minutes=54)
+
+    def test_heartbeat_ignores_expired_session(self, manager):
+        """Already-expired sessions are not silently revived."""
+        _token, session = manager.register_session(
+            container_id="egg-agent-pipeline-1-coder",
+            container_ip="172.18.0.5",
+            mode="private",
+        )
+        session.expires_at = datetime.now(UTC) - timedelta(minutes=1)
+
+        assert manager.heartbeat_session_by_container("egg-agent-pipeline-1-coder") is False
+
+
 class TestSessionModes:
     """Tests for session mode handling."""
 

orchestrator/gateway_client.py

@@ -573,6 +573,43 @@ def delete_session_by_container(self, container_id: str) -> bool:
             )
             return False
 
+    def heartbeat_session_by_container(self, container_id: str) -> bool:
+        """Refresh a session's idle timer by container ID.
+
+        Requires launcher secret authentication.  Used to keep gateway
+        sessions alive while an agent is heartbeating on the BRC bus but
+        not making gateway requests — see #2068.
+
+        Args:
+            container_id: Container ID whose session to refresh.
+
+        Returns:
+            True if the session was refreshed; False if there is no
+            matching session or the gateway request failed.  Best-effort
+            — callers should not fail on a False return.
+        """
+        try:
+            result = self._make_request(
+                f"/api/v1/sessions/by-container/{quote(container_id, safe='')}/heartbeat",
+                method="POST",
+                use_launcher_auth=True,
+            )
+            return result.get("success", False)
+        except GatewayError as e:
+            # Log the full container_id (not a secret — already shows up
+            # in k8s `get pods` output) so the failing pipeline+role is
+            # identifiable from #2068's exact failure mode.  The sibling
+            # ``delete_session_by_container`` truncates to 12 chars
+            # (``egg-agent-is`` for realistic ids), which loses both
+            # pipeline and role; reviewer NB4 on #2076 flagged that as
+            # un-debuggable here even if it's pre-existing there.
+            logger.warning(
+                "Failed to heartbeat session by container",
+                container_id=container_id,
+                error=str(e),
+            )
+            return False
+
     def create_worktrees(
         self,
         container_id: str,

orchestrator/heartbeat.py

@@ -42,6 +42,8 @@ def __init__(self, window_seconds: int = 60) -> None:
         self._windows: dict[tuple[str, str], deque[float]] = {}
         # (pipeline_id, role) -> last-seen (state, waiting_on)
         self._last_state: dict[tuple[str, str], tuple[str, str]] = {}
+        # (pipeline_id, role) -> last gateway-session fan-out epoch-seconds
+        self._last_fan_out: dict[tuple[str, str], float] = {}
 
     def check_rate_limit(
         self,
@@ -102,6 +104,48 @@ def record_state(
         with self._lock:
             self._last_state[key] = (state, waiting_on or "")
 
+    def should_fan_out_gateway_session(
+        self,
+        pipeline_id: str,
+        role: str,
+        min_interval_seconds: float,
+    ) -> bool:
+        """Throttle gateway-session fan-outs (issue #2076 NB2).
+
+        The HEARTBEAT route fans out a gateway-session refresh both on
+        the dedup early-return path and after the rate-limit gate.  The
+        dedup path bypasses the per-role rate limiter (by design — see
+        ``check_rate_limit``'s NB1 from #1897), so a misbehaving agent
+        hot-looping with identical state can amplify into the gateway
+        without burning rate budget.  This per-role cooldown caps that
+        amplification independently of the heartbeat-acceptance rate.
+
+        Returns ``True`` and records the new timestamp if at least
+        ``min_interval_seconds`` have passed since the previous fan-out
+        for this ``(pipeline_id, role)``; returns ``False`` (without
+        recording) if the caller should skip the fan-out this round.
+        Any non-positive ``min_interval_seconds`` (``<= 0``) disables
+        throttling — every call returns ``True`` without recording.
+
+        Callers must pass a finite real number. ``float('nan')`` falls
+        through both branches (``nan <= 0`` and ``now - last < nan`` are
+        both False), which would silently behave as "always record,
+        never suppress" — not a meaningful state.  The realistic call
+        site (``_GATEWAY_FANOUT_MIN_INTERVAL_SECONDS = 30.0``) is a
+        module constant, but if a future env-var-driven knob lands
+        (#2076 NB5) it should sanitize NaN/inf at parse time.
+        """
+        if min_interval_seconds <= 0:
+            return True
+        key = (pipeline_id, role)
+        now = time.time()
+        with self._lock:
+            last = self._last_fan_out.get(key, 0.0)
+            if now - last < min_interval_seconds:
+                return False
+            self._last_fan_out[key] = now
+        return True
+
     def clear(self, pipeline_id: str) -> None:
         """Drop all state for a pipeline (on phase transition)."""
         with self._lock:
@@ -111,6 +155,9 @@ def clear(self, pipeline_id: str) -> None:
             for key in list(self._last_state):
                 if key[0] == pipeline_id:
                     del self._last_state[key]
+            for key in list(self._last_fan_out):
+                if key[0] == pipeline_id:
+                    del self._last_fan_out[key]
 
 
 _coordinator: HeartbeatCoordinator | None = None