fix(orch): mark paused-at-gate sessions as awaiting_input (#42)

The v2.0.0-rc3 finalizer-asymmetry fix (PR #41) correctly handled
graph-completed-without-pause: every non-streaming run now calls
_finalize_session_status_async() to flip the row to a terminal status.

But the *paused* branch was left as a status-write no-op: when the
graph paused at a HITL gate, the session row stayed at its pre-pause
status ('new' / 'in_progress') instead of transitioning to
'awaiting_input'. UIs that filter by status='awaiting_input' (the
approvals queue, the sessions rail Active group, /sessions in-flight
listing) missed paused sessions entirely.

This was the sibling case of CRITICAL #1 and was filed as #42 after
the rc3 ship.

The fix adds a new lock-guarded helper that mirrors
_finalize_session_status_async on the paused side:

    Orchestrator._mark_session_paused_async(session_id) -> str | None

It loads the row, captures the from-status, flips status to
'awaiting_input', saves, and emits both status_changed (per-session
event log) and session.status_changed (cross-session SSE) events via
the existing _emit_status_changed_event helper. It is a no-op when
the row is already at 'awaiting_input', already at a terminal status
(guard against a late paused-write unwinding a finalize that landed
in between), or the row is missing.

Four call sites updated to use it: the three start_session +
stream_session + retry_session paths in orchestrator.py, and the
OrchestratorService._run() background task in service.py.

Regression coverage in tests/test_finalizer_paths.py:
  - _PausedAtGateGraph fake graph that returns from ainvoke and
    reports a non-empty `next` tuple to aget_state (mirrors langgraph
    1.x's interrupt-boundary state).
  - test_orchestrator_start_session_marks_paused_run_awaiting_input:
    direct (non-streaming) path produces status='awaiting_input' on
    the row AND emits status_changed + session.status_changed events.
  - test_service_background_start_session_marks_paused_run_awaiting_input:
    same guarantee through the OrchestratorService._run() task.
  - test_mark_session_paused_is_no_op_when_already_awaiting_input:
    no spurious event emission.
  - test_mark_session_paused_is_no_op_on_terminal_status: guard
    against a late paused-write unwinding a completed terminal flip.

Two pre-existing shim tests updated to stub the new method:
  - tests/test_retry_session_locked_post_policy.py _StubOrch
  - tests/test_triggers/test_orchestrator_trigger_kwarg.py
    (orchestrator built via __new__, bypasses __init__)

Verified:
  - ASR_WEB_DIST=/tmp/empty uv run pytest --no-cov -> 1349 passed, 8 skipped
  - uv run ruff check src/ tests/ -> clean
  - uv run pyright src/runtime -> clean
  - cd web && npm run typecheck/lint/test:unit/build/check:size -> green
  - dist/* regenerated (HARD-08)

Author: aksOps

dist/app.py

@@ -6458,7 +6458,11 @@ async def _run() -> None:
                             ),
                             config=orch._thread_config(session_id),
                         )
-                    if not await orch._is_graph_paused(session_id):
+                    if await orch._is_graph_paused(session_id):
+                        # Issue #42: mark the row awaiting_input so UIs
+                        # filtering by that status see the paused session.
+                        await orch._mark_session_paused_async(session_id)
+                    else:
                         await orch._finalize_session_status_async(session_id)
                 except asyncio.CancelledError:
                     raise
@@ -14882,6 +14886,54 @@ async def _finalize_session_status_async(
         async with self._locks.acquire(session_id):
             return self._finalize_session_status(session_id)
 
+    async def _mark_session_paused_async(
+        self, session_id: str,
+    ) -> str | None:
+        """Lock-guarded write that flips a paused session to
+        ``"awaiting_input"`` and emits the matching ``status_changed``
+        events.
+
+        Companion to :meth:`_finalize_session_status_async`. The
+        finalizer handles graph-completed runs; this handles
+        graph-paused runs (HITL gate). Without it, sessions that
+        pause at a gate keep their old status (typically ``"new"`` /
+        ``"in_progress"``) and UIs that filter by ``"awaiting_input"``
+        (the approvals queue, the rail's Active group) miss them — see
+        issue #42.
+
+        No-op (returns ``None``) when the session is already at
+        ``"awaiting_input"``, already in a terminal status (a late
+        pause-write must not unwind a finalize that landed in
+        between), or the row is missing. Otherwise transitions the
+        row to ``"awaiting_input"`` and emits both per-session
+        ``status_changed`` and cross-session
+        ``session.status_changed`` events via
+        :func:`_emit_status_changed_event`.
+        """
+        async with self._locks.acquire(session_id):
+            try:
+                inc = self.store.load(session_id)
+            except (FileNotFoundError, ValueError, KeyError, LookupError):
+                return None
+            if inc.status == "awaiting_input":
+                return None
+            statuses = getattr(getattr(self, "cfg", None), "orchestrator", None)
+            if statuses is not None:
+                current_def = statuses.statuses.get(inc.status)
+                if current_def is not None and current_def.terminal:
+                    return None
+            from_status = inc.status
+            inc.status = "awaiting_input"
+            self.store.save(inc)
+            _emit_status_changed_event(
+                orch=self,
+                inc=inc,
+                from_status=from_status,
+                to_status="awaiting_input",
+                cause="gate_paused",
+            )
+            return "awaiting_input"
+
     async def _is_graph_paused(self, session_id: str) -> bool:
         """Return True iff the compiled graph has a pending step waiting
         to resume (i.e. it's paused at an ``interrupt()`` boundary).
@@ -15082,7 +15134,9 @@ async def start_session(self, *, query: str,
                        last_agent=None, error=None),
             config=self._thread_config(inc.id),
         )
-        if not await self._is_graph_paused(inc.id):
+        if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
+        else:
             await self._finalize_session_status_async(inc.id)
         return inc.id
 
@@ -15140,7 +15194,10 @@ async def stream_session(self, *, query: str, environment: str,
         # session is waiting for operator approval, not done. Stamping
         # default_terminal_status here would orphan the pending_approval
         # ToolCall row written by the gateway just before the pause.
+        # Issue #42: still flip the session row to awaiting_input so
+        # the approvals queue + sessions rail Active group pick it up.
         if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
             yield {"event": "session_paused", "incident_id": inc.id,
                    "ts": _event_ts()}
         else:
@@ -15422,8 +15479,10 @@ async def _retry_session_locked(self, session_id: str) -> AsyncIterator[dict]:
             yield self._to_ui_event(ev, session_id)
         # See ``stream_session`` for why pause-detection guards the
         # finalize call: a HITL pause must not be coerced into a
-        # terminal status.
+        # terminal status. Issue #42: still flip to awaiting_input so
+        # UIs filtering by that status see the retried session.
         if await self._is_graph_paused(session_id):
+            await self._mark_session_paused_async(session_id)
             yield {"event": "session_paused", "incident_id": session_id,
                    "ts": _event_ts()}
         else:

dist/apps/code-review.py

@@ -6511,7 +6511,11 @@ async def _run() -> None:
                             ),
                             config=orch._thread_config(session_id),
                         )
-                    if not await orch._is_graph_paused(session_id):
+                    if await orch._is_graph_paused(session_id):
+                        # Issue #42: mark the row awaiting_input so UIs
+                        # filtering by that status see the paused session.
+                        await orch._mark_session_paused_async(session_id)
+                    else:
                         await orch._finalize_session_status_async(session_id)
                 except asyncio.CancelledError:
                     raise
@@ -14935,6 +14939,54 @@ async def _finalize_session_status_async(
         async with self._locks.acquire(session_id):
             return self._finalize_session_status(session_id)
 
+    async def _mark_session_paused_async(
+        self, session_id: str,
+    ) -> str | None:
+        """Lock-guarded write that flips a paused session to
+        ``"awaiting_input"`` and emits the matching ``status_changed``
+        events.
+
+        Companion to :meth:`_finalize_session_status_async`. The
+        finalizer handles graph-completed runs; this handles
+        graph-paused runs (HITL gate). Without it, sessions that
+        pause at a gate keep their old status (typically ``"new"`` /
+        ``"in_progress"``) and UIs that filter by ``"awaiting_input"``
+        (the approvals queue, the rail's Active group) miss them — see
+        issue #42.
+
+        No-op (returns ``None``) when the session is already at
+        ``"awaiting_input"``, already in a terminal status (a late
+        pause-write must not unwind a finalize that landed in
+        between), or the row is missing. Otherwise transitions the
+        row to ``"awaiting_input"`` and emits both per-session
+        ``status_changed`` and cross-session
+        ``session.status_changed`` events via
+        :func:`_emit_status_changed_event`.
+        """
+        async with self._locks.acquire(session_id):
+            try:
+                inc = self.store.load(session_id)
+            except (FileNotFoundError, ValueError, KeyError, LookupError):
+                return None
+            if inc.status == "awaiting_input":
+                return None
+            statuses = getattr(getattr(self, "cfg", None), "orchestrator", None)
+            if statuses is not None:
+                current_def = statuses.statuses.get(inc.status)
+                if current_def is not None and current_def.terminal:
+                    return None
+            from_status = inc.status
+            inc.status = "awaiting_input"
+            self.store.save(inc)
+            _emit_status_changed_event(
+                orch=self,
+                inc=inc,
+                from_status=from_status,
+                to_status="awaiting_input",
+                cause="gate_paused",
+            )
+            return "awaiting_input"
+
     async def _is_graph_paused(self, session_id: str) -> bool:
         """Return True iff the compiled graph has a pending step waiting
         to resume (i.e. it's paused at an ``interrupt()`` boundary).
@@ -15135,7 +15187,9 @@ async def start_session(self, *, query: str,
                        last_agent=None, error=None),
             config=self._thread_config(inc.id),
         )
-        if not await self._is_graph_paused(inc.id):
+        if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
+        else:
             await self._finalize_session_status_async(inc.id)
         return inc.id
 
@@ -15193,7 +15247,10 @@ async def stream_session(self, *, query: str, environment: str,
         # session is waiting for operator approval, not done. Stamping
         # default_terminal_status here would orphan the pending_approval
         # ToolCall row written by the gateway just before the pause.
+        # Issue #42: still flip the session row to awaiting_input so
+        # the approvals queue + sessions rail Active group pick it up.
         if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
             yield {"event": "session_paused", "incident_id": inc.id,
                    "ts": _event_ts()}
         else:
@@ -15475,8 +15532,10 @@ async def _retry_session_locked(self, session_id: str) -> AsyncIterator[dict]:
             yield self._to_ui_event(ev, session_id)
         # See ``stream_session`` for why pause-detection guards the
         # finalize call: a HITL pause must not be coerced into a
-        # terminal status.
+        # terminal status. Issue #42: still flip to awaiting_input so
+        # UIs filtering by that status see the retried session.
         if await self._is_graph_paused(session_id):
+            await self._mark_session_paused_async(session_id)
             yield {"event": "session_paused", "incident_id": session_id,
                    "ts": _event_ts()}
         else:

dist/apps/incident-management.py

@@ -6523,7 +6523,11 @@ async def _run() -> None:
                             ),
                             config=orch._thread_config(session_id),
                         )
-                    if not await orch._is_graph_paused(session_id):
+                    if await orch._is_graph_paused(session_id):
+                        # Issue #42: mark the row awaiting_input so UIs
+                        # filtering by that status see the paused session.
+                        await orch._mark_session_paused_async(session_id)
+                    else:
                         await orch._finalize_session_status_async(session_id)
                 except asyncio.CancelledError:
                     raise
@@ -14947,6 +14951,54 @@ async def _finalize_session_status_async(
         async with self._locks.acquire(session_id):
             return self._finalize_session_status(session_id)
 
+    async def _mark_session_paused_async(
+        self, session_id: str,
+    ) -> str | None:
+        """Lock-guarded write that flips a paused session to
+        ``"awaiting_input"`` and emits the matching ``status_changed``
+        events.
+
+        Companion to :meth:`_finalize_session_status_async`. The
+        finalizer handles graph-completed runs; this handles
+        graph-paused runs (HITL gate). Without it, sessions that
+        pause at a gate keep their old status (typically ``"new"`` /
+        ``"in_progress"``) and UIs that filter by ``"awaiting_input"``
+        (the approvals queue, the rail's Active group) miss them — see
+        issue #42.
+
+        No-op (returns ``None``) when the session is already at
+        ``"awaiting_input"``, already in a terminal status (a late
+        pause-write must not unwind a finalize that landed in
+        between), or the row is missing. Otherwise transitions the
+        row to ``"awaiting_input"`` and emits both per-session
+        ``status_changed`` and cross-session
+        ``session.status_changed`` events via
+        :func:`_emit_status_changed_event`.
+        """
+        async with self._locks.acquire(session_id):
+            try:
+                inc = self.store.load(session_id)
+            except (FileNotFoundError, ValueError, KeyError, LookupError):
+                return None
+            if inc.status == "awaiting_input":
+                return None
+            statuses = getattr(getattr(self, "cfg", None), "orchestrator", None)
+            if statuses is not None:
+                current_def = statuses.statuses.get(inc.status)
+                if current_def is not None and current_def.terminal:
+                    return None
+            from_status = inc.status
+            inc.status = "awaiting_input"
+            self.store.save(inc)
+            _emit_status_changed_event(
+                orch=self,
+                inc=inc,
+                from_status=from_status,
+                to_status="awaiting_input",
+                cause="gate_paused",
+            )
+            return "awaiting_input"
+
     async def _is_graph_paused(self, session_id: str) -> bool:
         """Return True iff the compiled graph has a pending step waiting
         to resume (i.e. it's paused at an ``interrupt()`` boundary).
@@ -15147,7 +15199,9 @@ async def start_session(self, *, query: str,
                        last_agent=None, error=None),
             config=self._thread_config(inc.id),
         )
-        if not await self._is_graph_paused(inc.id):
+        if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
+        else:
             await self._finalize_session_status_async(inc.id)
         return inc.id
 
@@ -15205,7 +15259,10 @@ async def stream_session(self, *, query: str, environment: str,
         # session is waiting for operator approval, not done. Stamping
         # default_terminal_status here would orphan the pending_approval
         # ToolCall row written by the gateway just before the pause.
+        # Issue #42: still flip the session row to awaiting_input so
+        # the approvals queue + sessions rail Active group pick it up.
         if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
             yield {"event": "session_paused", "incident_id": inc.id,
                    "ts": _event_ts()}
         else:
@@ -15487,8 +15544,10 @@ async def _retry_session_locked(self, session_id: str) -> AsyncIterator[dict]:
             yield self._to_ui_event(ev, session_id)
         # See ``stream_session`` for why pause-detection guards the
         # finalize call: a HITL pause must not be coerced into a
-        # terminal status.
+        # terminal status. Issue #42: still flip to awaiting_input so
+        # UIs filtering by that status see the retried session.
         if await self._is_graph_paused(session_id):
+            await self._mark_session_paused_async(session_id)
             yield {"event": "session_paused", "incident_id": session_id,
                    "ts": _event_ts()}
         else:

src/runtime/orchestrator.py

@@ -1091,6 +1091,54 @@ async def _finalize_session_status_async(
         async with self._locks.acquire(session_id):
             return self._finalize_session_status(session_id)
 
+    async def _mark_session_paused_async(
+        self, session_id: str,
+    ) -> str | None:
+        """Lock-guarded write that flips a paused session to
+        ``"awaiting_input"`` and emits the matching ``status_changed``
+        events.
+
+        Companion to :meth:`_finalize_session_status_async`. The
+        finalizer handles graph-completed runs; this handles
+        graph-paused runs (HITL gate). Without it, sessions that
+        pause at a gate keep their old status (typically ``"new"`` /
+        ``"in_progress"``) and UIs that filter by ``"awaiting_input"``
+        (the approvals queue, the rail's Active group) miss them — see
+        issue #42.
+
+        No-op (returns ``None``) when the session is already at
+        ``"awaiting_input"``, already in a terminal status (a late
+        pause-write must not unwind a finalize that landed in
+        between), or the row is missing. Otherwise transitions the
+        row to ``"awaiting_input"`` and emits both per-session
+        ``status_changed`` and cross-session
+        ``session.status_changed`` events via
+        :func:`_emit_status_changed_event`.
+        """
+        async with self._locks.acquire(session_id):
+            try:
+                inc = self.store.load(session_id)
+            except (FileNotFoundError, ValueError, KeyError, LookupError):
+                return None
+            if inc.status == "awaiting_input":
+                return None
+            statuses = getattr(getattr(self, "cfg", None), "orchestrator", None)
+            if statuses is not None:
+                current_def = statuses.statuses.get(inc.status)
+                if current_def is not None and current_def.terminal:
+                    return None
+            from_status = inc.status
+            inc.status = "awaiting_input"
+            self.store.save(inc)
+            _emit_status_changed_event(
+                orch=self,
+                inc=inc,
+                from_status=from_status,
+                to_status="awaiting_input",
+                cause="gate_paused",
+            )
+            return "awaiting_input"
+
     async def _is_graph_paused(self, session_id: str) -> bool:
         """Return True iff the compiled graph has a pending step waiting
         to resume (i.e. it's paused at an ``interrupt()`` boundary).
@@ -1291,7 +1339,9 @@ async def start_session(self, *, query: str,
                        last_agent=None, error=None),
             config=self._thread_config(inc.id),
         )
-        if not await self._is_graph_paused(inc.id):
+        if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
+        else:
             await self._finalize_session_status_async(inc.id)
         return inc.id
 
@@ -1349,7 +1399,10 @@ async def stream_session(self, *, query: str, environment: str,
         # session is waiting for operator approval, not done. Stamping
         # default_terminal_status here would orphan the pending_approval
         # ToolCall row written by the gateway just before the pause.
+        # Issue #42: still flip the session row to awaiting_input so
+        # the approvals queue + sessions rail Active group pick it up.
         if await self._is_graph_paused(inc.id):
+            await self._mark_session_paused_async(inc.id)
             yield {"event": "session_paused", "incident_id": inc.id,
                    "ts": _event_ts()}
         else:
@@ -1631,8 +1684,10 @@ async def _retry_session_locked(self, session_id: str) -> AsyncIterator[dict]:
             yield self._to_ui_event(ev, session_id)
         # See ``stream_session`` for why pause-detection guards the
         # finalize call: a HITL pause must not be coerced into a
-        # terminal status.
+        # terminal status. Issue #42: still flip to awaiting_input so
+        # UIs filtering by that status see the retried session.
         if await self._is_graph_paused(session_id):
+            await self._mark_session_paused_async(session_id)
             yield {"event": "session_paused", "incident_id": session_id,
                    "ts": _event_ts()}
         else: