Merge branch 'main' into fix/request-phase-policy-ask

A policy returning ASK on the REQUEST phase (e.g. the LLM prompt
classifier matching a user message) was silently denied: the input
path returned a "pending" verdict that nothing waited on, so the
/events handler collapsed it to "[Denied by policy]". Unlike
tool_call, the REQUEST phase has no runner-side approval park — the
message has not been forwarded to a runner yet.

Make _evaluate_input_policy park server-side on ASK via the existing
_hold_native_ask_gate (the same hold the native tool_call gate uses):
accept -> ALLOW (forward the message), decline/timeout -> DENY
(fail-closed). Thread the FastAPI request through from post_event for
disconnect detection, and generalize _hold_native_ask_gate's docstring
(it now serves REQUEST as well as TOOL_CALL). Add request-phase ASK
approve/decline unit tests.

omnigent/server/routes/sessions.py

@@ -3435,21 +3435,26 @@ async def _hold_native_ask_gate(
     conversation_store: ConversationStore,
 ) -> bool:
     """
-    Hold a native-harness tool call until a human resolves the ASK.
+    Hold a server-side ASK gate until a human resolves it.
 
-    URL-based elicitation for native harnesses. Publishes a
-    ``response.elicitation_request`` (the web UI / REPL render the
-    approve card) and parks a server-side Future via
+    Publishes a ``response.elicitation_request`` (the web UI / REPL
+    render the approve card) and parks a server-side Future via
     :func:`_publish_and_wait_for_harness_elicitation`, exactly as the
     ``PermissionRequest`` hook does. The human approves through the
     elicitation's resolve URL; this collapses the verdict to a single
     boolean the caller maps to ALLOW / DENY.
 
+    Used for any phase whose ASK must be resolved on the server rather
+    than by a runner-side ``wait_for_user_approval`` park:
+    :attr:`Phase.TOOL_CALL` (the native ``PreToolUse`` hook gate) and
+    :attr:`Phase.REQUEST` (the user-message input gate, which has no
+    runner in the loop yet — see :func:`_evaluate_input_policy`).
+
     Unlike the old ASK→``defer`` path, the gate lives on the server,
     so a permissive native ``permission_mode`` (``acceptEdits`` /
-    ``bypassPermissions``) cannot skip it — the tool call stays
-    blocked until a real human verdict. Timeout / disconnect fail
-    closed (return ``False`` → DENY).
+    ``bypassPermissions``) cannot skip it — the action stays blocked
+    until a real human verdict. Timeout / disconnect fail closed
+    (return ``False`` → DENY).
 
     On approve, the ASK-accumulated ``set_labels`` / ``state_updates``
     are applied (POLICIES.md §7.2: side effects land only on approve);
@@ -3458,10 +3463,12 @@ async def _hold_native_ask_gate(
     :param request: FastAPI request, for upstream-disconnect detection
         inside the parking helper.
     :param session_id: Omnigent session id, e.g. ``"conv_abc123"``.
-    :param phase: Enforcement phase — :attr:`Phase.TOOL_CALL` here
-        (the only phase a native PreToolUse hook can block).
-    :param data: The proto event ``data``; for a tool call,
-        ``{"name": "Bash", "arguments": {"command": "ls"}}``.
+    :param phase: Enforcement phase being gated, e.g.
+        :attr:`Phase.TOOL_CALL` or :attr:`Phase.REQUEST`.
+    :param data: The proto event ``data`` — for a tool call,
+        ``{"name": "Bash", "arguments": {"command": "ls"}}``; for a
+        request, the user ``message`` body
+        (``{"role": "user", "content": [...]}``).
     :param engine: The policy engine, used to resolve the per-policy
         ``ask_timeout`` and to apply approved side effects.
     :param result: The composed ASK :class:`PolicyResult` — carries
@@ -8963,6 +8970,7 @@ def _extract_user_text_from_event(body: SessionEventInput) -> str:
 
 
 async def _evaluate_input_policy(
+    request: Request,
     session_id: str,
     conv: Conversation,
     body: SessionEventInput,
@@ -8973,26 +8981,36 @@ async def _evaluate_input_policy(
     actor: dict[str, str] | None = None,
 ) -> dict[str, Any] | None:
     """
-    Evaluate a user message against INPUT phase policy rules.
+    Evaluate a user message against REQUEST (input) phase policy rules.
 
-    Pure evaluation — does NOT persist the event. Returns
-    ``None`` on ALLOW (caller should persist via the active
-    path). Returns a verdict dict on DENY or ASK (caller
-    should NOT forward to runner).
+    Does not persist the event. On ALLOW returns ``None`` (caller
+    forwards the message). On DENY returns a verdict dict (caller does
+    NOT forward). On ASK this function **parks for human approval**
+    before returning: unlike the ``tool_call`` phase — where the runner
+    parks via ``wait_for_user_approval`` — the REQUEST phase has no
+    runner in the loop yet (the message hasn't been forwarded), so the
+    approval gate must live here. It reuses :func:`_hold_native_ask_gate`
+    (the same server-side park the native ``tool_call`` gate uses):
+    accept collapses to ALLOW (``None``, forward the message), while
+    decline / timeout collapses to a DENY verdict (fail-closed).
 
+    :param request: The active FastAPI request, threaded to
+        :func:`_hold_native_ask_gate` for upstream-disconnect detection
+        while parked on an ASK.
     :param session_id: Session/conversation identifier,
         e.g. ``"conv_abc123"``.
     :param conv: The session's :class:`Conversation` entity.
     :param body: The validated ``message`` event.
     :param conversation_store: Store for label state.
     :param agent_store: Store for agent spec lookups.
-    :param runner_router: Unused, kept for signature
+    :param _runner_router: Unused, kept for signature
         consistency.
     :param actor: Authenticated principal, e.g.
         ``{"run_as": "alice@example.com"}``. ``None`` when
         identity is unknown.
-    :returns: ``None`` on ALLOW (fall through to persist
-        path). Verdict dict on DENY/ASK.
+    :returns: ``None`` on ALLOW or an approved ASK (fall through to the
+        forward path). A verdict dict ``{"verdict": "deny", "reason":
+        ...}`` on DENY or a declined / timed-out ASK.
     """
 
     user_text = _extract_user_text_from_event(body)
@@ -9036,18 +9054,29 @@ async def _evaluate_input_policy(
             "reason": result.reason or "Denied by policy",
         }
 
-    # ASK — publish elicitation event.
-    elicitation_id = await _register_policy_elicitation(
+    # ASK — park server-side for human approval. The REQUEST phase has no
+    # runner-side approval round-trip (the message has not been forwarded to
+    # a runner yet, so nothing would park on a "pending" verdict — it would
+    # collapse to a silent deny). Hold the gate here exactly like the native
+    # tool_call path: _hold_native_ask_gate publishes the approval card,
+    # awaits the human verdict on a server-side Future, and applies the
+    # deciding policy's writes only on accept (POLICIES.md §7.2). Accept ->
+    # ALLOW (fall through to forward the message); decline / timeout ->
+    # DENY (fail-closed).
+    approved = await _hold_native_ask_gate(
+        request,
         session_id=session_id,
+        phase=Phase.REQUEST,
+        data=body.data,
+        engine=engine,
         result=result,
-        arguments_preview=user_text[:1024],
         conversation_store=conversation_store,
     )
+    if approved:
+        return None
     return {
-        "verdict": "pending",
-        "elicitation_id": elicitation_id,
-        # Spec-resolved approval window; the runner's park honors it.
-        "ask_timeout": resolve_ask_timeout(engine, result),
+        "verdict": "deny",
+        "reason": result.reason or "Denied by policy",
     }
 
 
@@ -15459,6 +15488,7 @@ def create_sessions_router(
         ):
             try:
                 _input_verdict = await _evaluate_input_policy(
+                    request,
                     session_id,
                     conv,
                     body,
@@ -15504,6 +15534,7 @@ def create_sessions_router(
                 return {"queued": False, "denied": True, "reason": reason}
         elif body.type == _SLASH_COMMAND_TYPE and conv.agent_id is not None:
             _input_verdict = await _evaluate_input_policy(
+                request,
                 session_id,
                 conv,
                 _build_skill_slash_command_policy_body(body),

tests/server/routes/test_sessions_policy.py

@@ -25,7 +25,7 @@ from omnigent.server.routes.sessions import (
 )
 from omnigent.server.schemas import SessionEventInput
 from omnigent.spec import AgentSpec
-from omnigent.spec.types import PolicySpec
+from omnigent.spec.types import Phase, PolicySpec
 
 # ── Stub stores ──────────────────────────────────────────────
 
@@ -128,6 +128,25 @@ class _FakeBody:
     data: dict[str, Any]
 
 
+class _FakeRequest:
+    """Minimal stand-in for a FastAPI ``Request``.
+
+    ``_evaluate_input_policy`` only passes the request through to
+    ``_hold_native_ask_gate`` (for upstream-disconnect detection while
+    parked on an ASK). The ALLOW / DENY / skip tests never reach the
+    gate, and the ASK tests stub the gate out, so the request is never
+    actually introspected — this exists only to fill the positional
+    parameter with a real object rather than ``None``.
+    """
+
+    async def is_disconnected(self) -> bool:
+        """Report the client as connected.
+
+        :returns: Always ``False`` (test client never disconnects).
+        """
+        return False
+
+
 # ── Helpers ──────────────────────────────────────────────────
 
 
@@ -182,6 +201,7 @@ def _make_spec_no_guardrails() -> AgentSpec:
 
 _CACHE_PATCH = "omnigent.server.routes.sessions.get_agent_cache"
 _ENGINE_PATCH = "omnigent.server.routes.sessions.build_policy_engine"
+_HOLD_GATE_PATCH = "omnigent.server.routes.sessions._hold_native_ask_gate"
 _STREAM_PATCH = "omnigent.server.routes.sessions.session_stream"
 
 
@@ -462,6 +482,7 @@ async def test_input_allow_verdict():
         mock_engine.apply_label_writes = lambda x: None
 
         result = await _evaluate_input_policy(
+            _FakeRequest(),
             "sess_1",
             conv,
             body,
@@ -501,6 +522,7 @@ async def test_input_deny_verdict():
         mock_engine.apply_label_writes = lambda x: None
 
         result = await _evaluate_input_policy(
+            _FakeRequest(),
             "sess_1",
             conv,
             body,
@@ -564,6 +586,7 @@ async def test_skill_slash_command_policy_body_uses_typed_command_text():
         mock_engine.apply_label_writes = lambda x: None
 
         result = await _evaluate_input_policy(
+            _FakeRequest(),
             "sess_1",
             conv,
             policy_body,
@@ -595,6 +618,7 @@ async def test_input_no_guardrails_skips_policy():
     ):
         mock_cache.return_value.load.return_value = loaded
         result = await _evaluate_input_policy(
+            _FakeRequest(),
             "sess_1",
             conv,
             body,
@@ -620,6 +644,7 @@ async def test_input_empty_text_skips_policy():
     )
 
     result = await _evaluate_input_policy(
+        _FakeRequest(),
         "sess_1",
         conv,
         body,
@@ -631,6 +656,156 @@ async def test_input_empty_text_skips_policy():
     assert result is None
 
 
+@pytest.mark.asyncio
+async def test_input_ask_approved_falls_through_to_allow():
+    """A REQUEST-phase ASK the user APPROVES collapses to ALLOW.
+
+    Regression guard for the request-phase approval round-trip. The
+    REQUEST phase has no runner-side park (the message has not been
+    forwarded yet), so the input path must hold the gate server-side
+    via ``_hold_native_ask_gate`` and, on accept, return ``None`` so the
+    /events handler forwards the message. Before the fix, an input ASK
+    returned a ``pending`` verdict that the handler collapsed to
+    ``[Denied by policy]`` — the approval card was published but nothing
+    waited on it.
+    """
+    conv_store = _FakeConversationStore()
+    agent_store = _FakeAgentStore(agent=_make_agent())
+    conv = conv_store.get_conversation("sess_1")
+    body = _make_user_message_body("delete the file /tmp/policy-demo.txt")
+
+    spec = _make_spec_with_guardrails()
+    loaded = LoadedAgent(spec=spec, workdir="/tmp/fake")
+    ask_result = PolicyResult(
+        action=PolicyAction.ASK,
+        reason="Deleting files requires approval",
+        deciding_policy="llm_prompt_classifier_policy",
+    )
+
+    async def _eval(_ctx: Any) -> PolicyResult:
+        return ask_result
+
+    held_phases: list[Phase] = []
+
+    async def _fake_hold(
+        _request: Any,
+        *,
+        session_id: str,
+        phase: Phase,
+        data: dict[str, Any],
+        engine: Any,
+        result: PolicyResult,
+        conversation_store: Any,
+    ) -> bool:
+        """Stand in for the server-side approval park; simulate approve.
+
+        Records the phase so the test can assert the ASK was routed
+        through the gate at the REQUEST phase (not the old pending path).
+
+        :returns: ``True`` — the user clicked Approve.
+        """
+        held_phases.append(phase)
+        return True
+
+    with (
+        patch(_CACHE_PATCH) as mock_cache,
+        patch(_ENGINE_PATCH) as mock_build,
+        patch(_HOLD_GATE_PATCH, new=_fake_hold),
+    ):
+        mock_cache.return_value.load.return_value = loaded
+        mock_engine = mock_build.return_value
+        mock_engine.evaluate = _eval
+        mock_engine.apply_label_writes = lambda x: None
+
+        result = await _evaluate_input_policy(
+            _FakeRequest(),
+            "sess_1",
+            conv,
+            body,
+            conv_store,
+            agent_store,
+            None,
+        )
+
+    # The ASK was routed through the server-side approval park, at the
+    # REQUEST phase. If this is empty, the input path skipped the gate
+    # (the regressed "pending"/silent-deny path); a non-REQUEST phase
+    # would mean the wrong gate fired.
+    assert held_phases == [Phase.REQUEST]
+    # Approve -> None -> the /events handler forwards the message. A dict
+    # here would mean the message was wrongly blocked despite approval.
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_input_ask_declined_denies():
+    """A REQUEST-phase ASK the user DECLINES (or times out) collapses to DENY.
+
+    The fail-closed half of the request-phase round-trip: when the
+    server-side park returns ``False`` (decline / cancel / timeout),
+    ``_evaluate_input_policy`` returns a deny verdict carrying the
+    deciding policy's reason so the /events handler refuses to forward
+    the message.
+    """
+    conv_store = _FakeConversationStore()
+    agent_store = _FakeAgentStore(agent=_make_agent())
+    conv = conv_store.get_conversation("sess_1")
+    body = _make_user_message_body("delete the file /tmp/policy-demo.txt")
+
+    spec = _make_spec_with_guardrails()
+    loaded = LoadedAgent(spec=spec, workdir="/tmp/fake")
+    ask_result = PolicyResult(
+        action=PolicyAction.ASK,
+        reason="Deleting files requires approval",
+        deciding_policy="llm_prompt_classifier_policy",
+    )
+
+    async def _eval(_ctx: Any) -> PolicyResult:
+        return ask_result
+
+    async def _fake_hold(
+        _request: Any,
+        *,
+        session_id: str,
+        phase: Phase,
+        data: dict[str, Any],
+        engine: Any,
+        result: PolicyResult,
+        conversation_store: Any,
+    ) -> bool:
+        """Stand in for the server-side approval park; simulate decline.
+
+        :returns: ``False`` — the user declined / the gate timed out.
+        """
+        return False
+
+    with (
+        patch(_CACHE_PATCH) as mock_cache,
+        patch(_ENGINE_PATCH) as mock_build,
+        patch(_HOLD_GATE_PATCH, new=_fake_hold),
+    ):
+        mock_cache.return_value.load.return_value = loaded
+        mock_engine = mock_build.return_value
+        mock_engine.evaluate = _eval
+        mock_engine.apply_label_writes = lambda x: None
+
+        result = await _evaluate_input_policy(
+            _FakeRequest(),
+            "sess_1",
+            conv,
+            body,
+            conv_store,
+            agent_store,
+            None,
+        )
+
+    # Decline -> deny verdict carrying the deciding policy's reason. A
+    # ``None`` here would mean a declined ASK silently let the message
+    # through (the dangerous direction).
+    assert result["verdict"] == "deny"
+    assert result["reason"] == "Deleting files requires approval"
+
+
 # ── OUTPUT policy tests (step 5.7) ──────────────────────────