fix(runner): bound native-pane running pin with staleness

A stuck _native_pane_status=running entry could pin the idle watchdog
forever after a silent forwarder/pane death. Treat running as active only
while freshly refreshed (status edge or terminal.activity), clear the pin
on external idle/failed, and cover stale/cold-start paths with tests.

Signed-off-by: SabhyaC26 <sabhyachhabria@gmail.com>

omnigent/runner/app.py

@@ -242,6 +242,13 @@ for _builder_name in (
 # Servers before 0.3.0 cannot serialize the runner's "waiting" status.
 # Unknown versions also downgrade to "running" so old servers never return 500.
 _WAITING_STATUS_MIN_SERVER_VERSION = "0.3.0"
+# A native pane ``running`` entry only pins the idle watchdog while fresh.
+# Status edges are edge-triggered (idle→running once); continuous work instead
+# refreshes the stamp via ``session.terminal.activity`` (~1s). Without a bound,
+# a SIGKILLed forwarder / frozen pane leaves ``running`` forever and the runner
+# never shuts down. 120s is well above the activity cadence and the 1s PTY idle
+# threshold, but short enough that a dead pin cannot hold billable compute open.
+_NATIVE_PANE_RUNNING_STALE_S = 120.0
 # Cached server version from the /api/version probe; ``None`` until a probe
 # succeeds. A failed probe stays ``None`` and is retried on the next
 # session-create — the GET is cheap and self-heals a transient failure.
@@ -1978,7 +1985,10 @@ def create_runner_app(
     app.state.antigravity_terminal_ensure_locks = _antigravity_terminal_ensure_locks
     _repl_terminal_ensure_locks: dict[str, asyncio.Lock] = {}
     _active_turns: dict[str, asyncio.Task[None] | None] = {}
-    _native_pane_status: dict[str, str] = {}
+    # session_id → (status, monotonic timestamp of last liveness refresh).
+    # ``running`` only pins the idle watchdog while the stamp is fresh; see
+    # :data:`_NATIVE_PANE_RUNNING_STALE_S`.
+    _native_pane_status: dict[str, tuple[str, float]] = {}
     _session_message_buffers: dict[str, list[_JsonObject]] = {}
     _ingest_next_seq: dict[str, int] = {}
     _ingest_now_serving: dict[str, int] = {}
@@ -1994,6 +2004,47 @@ def create_runner_app(
     _session_inboxes = _session_inboxes_ref
     _session_async_tasks: dict[str, dict[str, tuple[asyncio.Task[str], asyncio.Event]]] = {}
 
+    def _set_native_pane_status(session_id: str, status: str, *, now: float | None = None) -> None:
+        """Record a native pane status edge and refresh its liveness stamp.
+
+        :param session_id: Omnigent session/conversation id.
+        :param status: Status value, e.g. ``"running"`` or ``"idle"``.
+        :param now: Optional monotonic timestamp (tests inject a clock).
+        :returns: None.
+        """
+        _native_pane_status[session_id] = (
+            status,
+            time.monotonic() if now is None else now,
+        )
+
+    def _touch_native_pane_running(session_id: str, *, now: float | None = None) -> None:
+        """Refresh the liveness stamp when a running pane still has activity.
+
+        :param session_id: Omnigent session/conversation id.
+        :param now: Optional monotonic timestamp (tests inject a clock).
+        :returns: None.
+        """
+        entry = _native_pane_status.get(session_id)
+        if entry is None or entry[0] != "running":
+            return
+        _native_pane_status[session_id] = (
+            "running",
+            time.monotonic() if now is None else now,
+        )
+
+    def _has_fresh_native_pane_running(*, now: float | None = None) -> bool:
+        """Return whether any native pane is freshly ``running``.
+
+        :param now: Optional monotonic timestamp (tests inject a clock).
+        :returns: ``True`` when at least one session is ``running`` and its
+            liveness stamp is within :data:`_NATIVE_PANE_RUNNING_STALE_S`.
+        """
+        stamp = time.monotonic() if now is None else now
+        for status, seen_at in _native_pane_status.values():
+            if status == "running" and stamp - seen_at <= _NATIVE_PANE_RUNNING_STALE_S:
+                return True
+        return False
+
     def _has_active_work() -> bool:
         if _active_turns:
             return True
@@ -2011,11 +2062,17 @@ def create_runner_app(
                 return True
         # Native PTY turns clear `_active_turns` at proxy-stream end while the
         # pane is still working; the pane reaper already treats this as busy.
-        if any(status == "running" for status in _native_pane_status.values()):
+        # Bound by freshness so a stuck ``running`` entry cannot pin the
+        # runner awake forever after a silent forwarder/pane death.
+        if _has_fresh_native_pane_running():
             return True
         return False
 
     app.state.has_active_work = _has_active_work
+    # Test seams for the freshness bound.
+    app.state._set_native_pane_status = _set_native_pane_status
+    app.state._has_fresh_native_pane_running = _has_fresh_native_pane_running
+    app.state._native_pane_running_stale_s = _NATIVE_PANE_RUNNING_STALE_S
 
     def _drain_session_streams() -> None:
         for queue in list(_session_event_queues.values()):
@@ -2030,16 +2087,24 @@ def create_runner_app(
             queue = asyncio.Queue()
             _session_event_queues[session_id] = queue
         queue.put_nowait(event_body)
-        if event_body.get("type") == "session.status":
+        event_type = event_body.get("type")
+        if event_type == "session.status":
             _status_value = event_body.get("status")
             if isinstance(_status_value, str):
-                _native_pane_status[session_id] = _status_value
+                _set_native_pane_status(session_id, _status_value)
                 # Reset the runner idle timer on native status edges so a long
                 # PTY turn does not look "already timed out" the moment it
                 # settles to idle.
                 mark_activity = getattr(app.state, "mark_activity", None)
                 if callable(mark_activity):
                     mark_activity()
+        elif event_type == "session.terminal.activity":
+            # Continuous pane work does not re-emit ``running`` (edge-only);
+            # activity pulses keep the freshness stamp alive instead.
+            _touch_native_pane_running(session_id)
+            mark_activity = getattr(app.state, "mark_activity", None)
+            if callable(mark_activity):
+                mark_activity()
         _fan_out_child_delta_to_parent(session_id, event_body)
 
     def _child_preview_from_status(
@@ -6364,6 +6429,11 @@ def create_runner_app(
             delivery_ack: _SubagentDeliveryAck | None = None
             recovered_entry: _SubagentWorkEntry | None = None
             if status in ("running", "waiting", "idle", "failed"):
+                # Keep the idle-watchdog pin aligned with forwarder-observed
+                # edges without re-publishing ``session.status`` (the server
+                # already did). Terminal ``idle``/``failed`` clear a stuck
+                # ``running`` pin even when the PTY watcher never fires again.
+                _set_native_pane_status(conversation_id, status)
                 resource_registry.note_external_session_status(conversation_id, status)
                 _fan_out_child_delta_to_parent(
                     conversation_id,
@@ -8948,7 +9018,12 @@ def create_runner_app(
                 process_manager is not None and process_manager.has_active_turn(conv_id)
             ):
                 return True
-            if _native_pane_status.get(conv_id) == "running":
+            entry = _native_pane_status.get(conv_id)
+            if (
+                entry is not None
+                and entry[0] == "running"
+                and time.monotonic() - entry[1] <= _NATIVE_PANE_RUNNING_STALE_S
+            ):
                 return True
             clients = await asyncio.to_thread(_list_tmux_clients, str(pane.socket_path), "main")
             if clients:

tests/runner/test_app_sessions_native_workflow_init.py

@@ -7,6 +7,7 @@ import contextlib
 import dataclasses
 import json
 import logging
+import time
 import uuid
 from collections.abc import AsyncIterator
 from pathlib import Path
@@ -2224,6 +2225,124 @@ async def test_has_active_work_reports_native_pane_running() -> None:
     assert app.state.has_active_work() is False
 
 
+@pytest.mark.asyncio
+async def test_has_active_work_ignores_stale_native_pane_running() -> None:
+    """A stale ``running`` entry must not pin the idle watchdog forever.
+
+    If a forwarder is SIGKILLed or the pane freezes without a final status
+    edge, ``_native_pane_status`` can linger as ``running``. Without a
+    freshness bound that turns into a whole-runner leak (the mirror of the
+    premature-idle bug).
+
+    :returns: None.
+    """
+    app, _pm, _hc = _build_lifecycle_app()
+    session_id = "b2c3d4e5f60718293a4b5c6d7e8f90a1"
+    async with _runner_client(app) as client:
+        resp = await client.post(
+            "/v1/sessions",
+            json={
+                "session_id": session_id,
+                "agent_id": "880b5afda28ad55ff74cbeb9b5fc67fb",
+            },
+        )
+
+    assert resp.status_code == 201
+    set_status = app.state._set_native_pane_status
+    has_fresh = app.state._has_fresh_native_pane_running
+    stale_s = float(app.state._native_pane_running_stale_s)
+
+    # Fresh running pin keeps the runner alive (stamp = now).
+    set_status(session_id, "running")
+    assert app.state.has_active_work() is True
+
+    # Injected clock: past the staleness window the pin expires even though
+    # the stored status string is still ``running``.
+    now = time.monotonic()
+    set_status(session_id, "running", now=now)
+    assert has_fresh(now=now) is True
+    assert has_fresh(now=now + stale_s + 1.0) is False
+
+    # Drive has_active_work through the live clock with an already-stale stamp.
+    set_status(session_id, "running", now=now - stale_s - 1.0)
+    assert app.state.has_active_work() is False
+
+
+@pytest.mark.asyncio
+async def test_terminal_activity_refreshes_native_pane_running_freshness() -> None:
+    """Pane activity pulses keep a long ``running`` turn from going stale.
+
+    Status edges are idle→running only; continuous work refreshes liveness via
+    ``session.terminal.activity`` so a healthy multi-minute turn still pins
+    the watchdog.
+
+    :returns: None.
+    """
+    app, _pm, _hc = _build_lifecycle_app()
+    session_id = "c3d4e5f60718293a4b5c6d7e8f90a1b2"
+    async with _runner_client(app) as client:
+        resp = await client.post(
+            "/v1/sessions",
+            json={
+                "session_id": session_id,
+                "agent_id": "880b5afda28ad55ff74cbeb9b5fc67fb",
+            },
+        )
+
+    assert resp.status_code == 201
+    registry = app.state.session_resource_registry
+    status_publisher = getattr(registry, "_session_status_publisher", None)
+    activity_publisher = getattr(registry, "_terminal_activity_publisher", None)
+    assert callable(status_publisher) and callable(activity_publisher)
+
+    status_publisher(session_id, "running")
+    assert app.state.has_active_work() is True
+
+    # Age the stamp past the window, then an activity pulse must revive it.
+    set_status = app.state._set_native_pane_status
+    stale_s = float(app.state._native_pane_running_stale_s)
+    set_status(session_id, "running", now=time.monotonic() - stale_s - 1.0)
+    assert app.state.has_active_work() is False
+    activity_publisher(session_id, "terminal_claude_main")
+    assert app.state.has_active_work() is True
+
+
+@pytest.mark.asyncio
+async def test_external_session_status_failed_clears_native_pane_running_pin() -> None:
+    """Forwarder ``failed`` clears the idle-watchdog pin without a PTY idle edge.
+
+    Part A posts ``external_session_status: failed`` when the forwarder gives
+    up. That must drop the ``running`` pin even if the PTY watcher never
+    observes quiescence (SIGKILL / hung TUI with a frozen pane).
+
+    :returns: None.
+    """
+    app, _pm, _hc = _build_lifecycle_app()
+    session_id = "d4e5f60718293a4b5c6d7e8f90a1b2c3"
+    async with _runner_client(app) as client:
+        resp = await client.post(
+            "/v1/sessions",
+            json={
+                "session_id": session_id,
+                "agent_id": "880b5afda28ad55ff74cbeb9b5fc67fb",
+            },
+        )
+        assert resp.status_code == 201
+
+        app.state._set_native_pane_status(session_id, "running")
+        assert app.state.has_active_work() is True
+
+        fail_resp = await client.post(
+            f"/v1/sessions/{session_id}/events",
+            json={
+                "type": "external_session_status",
+                "data": {"status": "failed", "output": "forwarder gave up"},
+            },
+        )
+        assert fail_resp.status_code == 204
+        assert app.state.has_active_work() is False
+
+
 @pytest.mark.asyncio
 async def test_create_session_missing_fields() -> None:
     """``POST /v1/sessions`` with missing fields returns 400."""

tests/runner/test_resource_registry.py

@@ -927,6 +927,35 @@ async def test_notify_pane_dead_is_idempotent_with_idle_watcher_exit(
     assert len(exits) == 1
 
 
+@pytest.mark.asyncio
+async def test_notify_pane_dead_is_noop_before_terminal_observed(tmp_path: Path) -> None:
+    """Pane-death notify before observe cannot publish a spurious failed card.
+
+    Native cold start briefly has no observed lifecycle. ``notify_pane_dead``
+    must no-op until ``observe_required_terminal`` registers the pane — the
+    attach bridge also refuses to connect until ``is_alive()`` is true, so a
+    pre-pane cold start cannot trip part B.
+
+    :param tmp_path: Temporary directory for fake terminal paths.
+    :returns: None.
+    """
+    from omnigent.entities.session_resources import terminal_resource_id
+
+    terminal_registry = TerminalRegistry()
+    registry = SessionResourceRegistry(terminal_registry=terminal_registry)
+    instance = make_test_terminal_instance("claude", "main", tmp_path)
+    terminal_registry._by_conversation.setdefault("conv_cold", {})[("claude", "main")] = instance
+    exits: list[TerminalExitEvent] = []
+
+    def _publish_exit(event: TerminalExitEvent) -> None:
+        exits.append(event)
+
+    registry.set_terminal_exit_publisher(_publish_exit)
+    registry.notify_pane_dead("conv_cold", terminal_resource_id("claude", "main"))
+    await asyncio.sleep(0)
+    assert exits == []
+
+
 @pytest.mark.asyncio
 async def test_required_terminal_exit_without_observed_status_is_failure(tmp_path: Path) -> None:
     """A required terminal that never reported a PTY status fails on exit.

tests/test_native_forwarder_liveness.py

@@ -163,6 +163,50 @@ async def test_successful_poll_resets_failure_streak() -> None:
     assert posts == []
 
 
+@pytest.mark.asyncio
+async def test_cold_start_empty_polls_do_not_escalate() -> None:
+    """Native cold-start discovery polls must not trip durable ``failed``.
+
+    A cold-starting native session commonly spends several polls discovering
+    the vendor session / waiting for the transcript file. Those iterations
+    succeed (no exception) and call :func:`note_poll_success`. Even a few
+    transient exceptions below the threshold must not POST ``failed`` — that
+    was the shape of prior spurious failed-card bugs during cold start.
+
+    :returns: None.
+    """
+    posts: list[dict[str, Any]] = []
+    tracker = PollFailureTracker()
+
+    async def _post_status(
+        _client: object,
+        *,
+        session_id: str,
+        status: str,
+        output: str | None = None,
+        **_kwargs: object,
+    ) -> None:
+        posts.append({"session_id": session_id, "status": status, "output": output})
+
+    # Simulate cold-start: a couple of transient misses, then a successful
+    # discovery poll, then more quiet successful polls — never escalate.
+    for i in range(2):
+        await handle_poll_failure(
+            client=_RecordingClient(),  # type: ignore[arg-type]
+            session_id="conv_cold",
+            tracker=tracker,
+            error=RuntimeError(f"not-ready-{i}"),
+            harness="claude-native",
+            post_status=_post_status,
+        )
+    note_poll_success(tracker)
+    for _ in range(POLL_FAILURE_THRESHOLD + 2):
+        note_poll_success(tracker)
+    assert posts == []
+    assert tracker.consecutive_failures == 0
+    assert tracker.failed_status_emitted is False
+
+
 @pytest.mark.asyncio
 async def test_supervisor_restarts_escalate_within_window() -> None:
     """M supervisor restarts inside the window POST durable ``failed``.