From ab0f2321eb9101849c4706b39ab79978f6ab8b2d Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Mon, 10 Aug 2026 03:13:22 -0400 Subject: [PATCH 1/2] feat: add HITL approval to OpenWebUI directive drafter --- .../open_webui_pipe_with_directive_drafter.py | 192 ++++++++++-------- ...t_openwebui_pipe_with_directive_drafter.py | 185 ++++++++++------- 2 files changed, 218 insertions(+), 159 deletions(-) diff --git a/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py b/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py index d9a13aa..cc2a11a 100644 --- a/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py +++ b/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py @@ -11,7 +11,8 @@ 1. Run heuristic directive drafter (fast, high-precision cases) 2. Fall back to Open WebUI-native model completion when needed -3. Pass resulting directive (or original input) to `engine.step(...)` +3. Present drafted canonical directives for host-side approval before + calling `engine.step(...)` Core decision handling remains the same as the base integration. Failed transitions are rejected for the current request and do not leave @@ -69,6 +70,7 @@ def Field(*, default: Any, description: str = "") -> Any: # type: ignore[no-red _CC_MARKER = "[[cc_state]]" _ENGINES_BY_CHAT_KEY: dict[str, Engine] = {} +_PENDING_PROPOSALS_BY_CHAT_KEY: dict[str, str] = {} class _EngineSnapshot(TypedDict): @@ -76,6 +78,22 @@ class _EngineSnapshot(TypedDict): policies: dict[str, PolicyValue] +def _is_explicit_approval(message: str) -> bool: + return message.strip().lower() in {"y", "yes"} + + +def _is_explicit_rejection(message: str) -> bool: + return message.strip().lower() in {"n", "no"} + + +def _render_proposal_prompt(directive_text: str) -> str: + return ( + "This is what I think the directive is:\n" + f"{directive_text}\n" + "Apply it? (y/n)" + ) + + def _resolve_chat_key( user: dict[str, Any], chat_id: str | None, @@ -809,6 +827,97 @@ async def pipe( if latest_user_text.strip().lower() == "show state": return _render_show_state_summary(engine) + pending_proposal = _PENDING_PROPOSALS_BY_CHAT_KEY.get(chat_key) + if pending_proposal is not None: + if _is_explicit_approval(latest_user_text): + del _PENDING_PROPOSALS_BY_CHAT_KEY[chat_key] + state_before = _snapshot_engine_state(engine) + engine_snapshot_json = engine.export_json() + compile_input = pending_proposal + logger.debug("preprocessor: approved_pending_input=%r", compile_input) + decision = engine.step(compile_input) + if decision["kind"] == DecisionKind.ERROR: + kind = DecisionKind.ERROR.value + elif is_update(decision): + kind = DECISION_UPDATE + else: + kind = DecisionKind.NO_DIRECTIVE.value + logger.debug("preprocessor: decision=%s", kind) + state_after = _snapshot_engine_state(engine) + + if decision["kind"] == DecisionKind.ERROR: + _ENGINES_BY_CHAT_KEY[chat_key] = _restore_engine_from_snapshot( + engine_snapshot_json + ) + return self._with_trace( + decision["message"] or "", + original_input=latest_user_text, + compiler_input=compile_input, + decision=decision, + state_before=state_before, + state_after=state_after, + preprocessor_output=compile_input, + llm_called=False, + ) + if decision["kind"] == DecisionKind.NO_DIRECTIVE: + state_injected = ( + "yes" if _has_non_empty_authoritative_state(engine) else "no" + ) + response = await self._forward_passthrough( + body, + __user__, + __request__, + base_model_id=base_model_id, + engine=engine, + ) + return self._with_trace( + response, + original_input=latest_user_text, + compiler_input=compile_input, + decision=decision, + state_before=state_before, + state_after=state_after, + preprocessor_output=compile_input, + llm_called=base_model_id is not None, + state_injected=state_injected, + ) + if is_update(decision): + return self._with_trace( + "State updated.", + original_input=latest_user_text, + compiler_input=compile_input, + decision=decision, + state_before=state_before, + state_after=state_after, + preprocessor_output=compile_input, + llm_called=False, + ) + + state_injected = ( + "yes" if _has_non_empty_authoritative_state(engine) else "no" + ) + response = await self._forward_passthrough( + body, + __user__, + __request__, + base_model_id=base_model_id, + engine=engine, + ) + return self._with_trace( + response, + original_input=latest_user_text, + compiler_input=compile_input, + decision=decision, + state_before=state_before, + state_after=state_after, + preprocessor_output=compile_input, + llm_called=base_model_id is not None, + state_injected=state_injected, + ) + if _is_explicit_rejection(latest_user_text): + del _PENDING_PROPOSALS_BY_CHAT_KEY[chat_key] + return "Directive discarded." + state_before = _snapshot_engine_state(engine) preprocess_error: str | None = None drafted_result, preprocess_error = await self._preprocess_user_input( @@ -845,83 +954,6 @@ async def pipe( state_injected=state_injected, ) - engine_snapshot_json = engine.export_json() compile_input = drafted_result.result.text - logger.debug("preprocessor: engine_input=%r", compile_input) - decision = engine.step(compile_input) - if decision["kind"] == DecisionKind.ERROR: - kind = DecisionKind.ERROR.value - elif is_update(decision): - kind = DECISION_UPDATE - else: - kind = DecisionKind.NO_DIRECTIVE.value - logger.debug("preprocessor: decision=%s", kind) - state_after = _snapshot_engine_state(engine) - - if decision["kind"] == DecisionKind.ERROR: - _ENGINES_BY_CHAT_KEY[chat_key] = _restore_engine_from_snapshot( - engine_snapshot_json - ) - return self._with_trace( - decision["message"] or "", - original_input=latest_user_text, - compiler_input=compile_input, - decision=decision, - state_before=state_before, - state_after=state_after, - preprocessor_output=compile_input, - llm_called=False, - ) - if decision["kind"] == DecisionKind.NO_DIRECTIVE: - state_injected = ( - "yes" if _has_non_empty_authoritative_state(engine) else "no" - ) - response = await self._forward_passthrough( - body, - __user__, - __request__, - base_model_id=base_model_id, - engine=engine, - ) - return self._with_trace( - response, - original_input=latest_user_text, - compiler_input=compile_input, - decision=decision, - state_before=state_before, - state_after=state_after, - preprocessor_output=compile_input, - llm_called=base_model_id is not None, - state_injected=state_injected, - ) - if is_update(decision): - return self._with_trace( - "State updated.", - original_input=latest_user_text, - compiler_input=compile_input, - decision=decision, - state_before=state_before, - state_after=state_after, - preprocessor_output=compile_input, - llm_called=False, - ) - - state_injected = "yes" if _has_non_empty_authoritative_state(engine) else "no" - response = await self._forward_passthrough( - body, - __user__, - __request__, - base_model_id=base_model_id, - engine=engine, - ) - return self._with_trace( - response, - original_input=latest_user_text, - compiler_input=compile_input, - decision=decision, - state_before=state_before, - state_after=state_after, - preprocessor_output=compile_input, - llm_called=base_model_id is not None, - state_injected=state_injected, - ) + _PENDING_PROPOSALS_BY_CHAT_KEY[chat_key] = compile_input + return _render_proposal_prompt(compile_input) diff --git a/python/tests/test_openwebui_pipe_with_directive_drafter.py b/python/tests/test_openwebui_pipe_with_directive_drafter.py index 6b0b14c..94371d1 100644 --- a/python/tests/test_openwebui_pipe_with_directive_drafter.py +++ b/python/tests/test_openwebui_pipe_with_directive_drafter.py @@ -89,27 +89,14 @@ def _guarded_import( module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) module._ENGINES_BY_CHAT_KEY.clear() + module._PENDING_PROPOSALS_BY_CHAT_KEY.clear() return module -def test_directive_drafting_runs_before_compiler_step(monkeypatch) -> None: +def test_canonical_draft_creates_approval_prompt_and_does_not_mutate_state( + monkeypatch, +) -> None: module = _load_module("owui_with_drafter_before_step", monkeypatch) - compile_inputs: list[str] = [] - real_create_engine = module.create_engine - - def create_engine_with_tracking(): - engine = real_create_engine() - original_step = engine.step - - def tracked_step(user_input: str): - compile_inputs.append(user_input) - return original_step(user_input) - - engine.step = tracked_step - return engine - - monkeypatch.setattr(module, "create_engine", create_engine_with_tracking) - async def fake_draft(*args, **kwargs): return DraftResult( source="test", @@ -125,6 +112,7 @@ async def fake_draft(*args, **kwargs): pipe = module.Pipe() pipe.valves.BASE_MODEL_ID = "base-model" pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" + chat_id = "chat-before-step" result = asyncio.run( pipe.pipe( @@ -134,27 +122,44 @@ async def fake_draft(*args, **kwargs): }, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-before-step", + __chat_id__=chat_id, + ) + ) + show_state = asyncio.run( + pipe.pipe( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "show state"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__=chat_id, ) ) - assert result == "State updated." - assert compile_inputs == ["use docker"] + assert result == ( + "This is what I think the directive is:\nuse docker\nApply it? (y/n)" + ) + assert show_state == "Premise: none\nUse: none\nProhibit: none" -def test_failed_transition_is_rejected_and_follow_up_is_a_new_request( - monkeypatch, -) -> None: +def test_approval_applies_stored_directive(monkeypatch) -> None: module = _load_module("owui_with_drafter_failed_transition_followup", monkeypatch) - forwarded: list[dict[str, object]] = [] + compile_inputs: list[str] = [] + real_create_engine = module.create_engine - async def forward( - _: object, payload: dict[str, object], __: object - ) -> dict[str, object]: - forwarded.append(payload) - return {"choices": [{"message": {"content": "downstream"}}]} + def create_engine_with_tracking(): + engine = real_create_engine() + original_step = engine.step - module.generate_chat_completion = forward + def tracked_step(user_input: str): + compile_inputs.append(user_input) + return original_step(user_input) + + engine.step = tracked_step + return engine + + monkeypatch.setattr(module, "create_engine", create_engine_with_tracking) pipe = module.Pipe() pipe.valves.BASE_MODEL_ID = "base-model" pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" @@ -181,58 +186,36 @@ async def update_draft(*args, **kwargs): __chat_id__="chat-failed-transition", ) ) - - async def no_draft(*args, **kwargs): - return DraftResult( - source="test", - result=CanonicalDirective( - text="prohibit docker", - kind=DirectiveKind.PROHIBIT_ITEM, - operands=MappingProxyType({"item": "docker"}), - ), - ) - - monkeypatch.setattr(module.Pipe, "_draft_user_input", no_draft) - rejected = asyncio.run( + follow_up = asyncio.run( pipe.pipe( - { - "model": "pipe-model", - "messages": [{"role": "user", "content": "prohibit docker"}], - }, + {"model": "pipe-model", "messages": [{"role": "user", "content": "y"}]}, __user__={"id": "u1"}, __request__=object(), __chat_id__="chat-failed-transition", ) ) - - async def follow_up_no_directive(*args, **kwargs): - return DraftResult( - source="test", - result=NoDirective(reason="reject.confident_non_directive"), - ) - - monkeypatch.setattr(module.Pipe, "_draft_user_input", follow_up_no_directive) - follow_up = asyncio.run( + show_state = asyncio.run( pipe.pipe( - {"model": "pipe-model", "messages": [{"role": "user", "content": "yes"}]}, + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "show state"}], + }, __user__={"id": "u1"}, __request__=object(), __chat_id__="chat-failed-transition", ) ) - assert seed == "State updated." - assert rejected == ( - '"docker" is currently in use.\nRemove or replace it before prohibiting it.' + assert seed == ( + "This is what I think the directive is:\nuse docker\nApply it? (y/n)" ) - assert follow_up == {"choices": [{"message": {"content": "downstream"}}]} - assert len(forwarded) == 1 + assert follow_up == "State updated." + assert show_state == "Premise: none\nUse: docker\nProhibit: none" + assert compile_inputs == ["use docker"] -def test_failed_transition_does_not_change_existing_engine_state(monkeypatch) -> None: - module = _load_module( - "owui_with_drafter_failed_transition_state_preserved", monkeypatch - ) +def test_rejection_does_not_mutate_state(monkeypatch) -> None: + module = _load_module("owui_with_drafter_failed_transition_state_preserved", monkeypatch) pipe = module.Pipe() pipe.valves.BASE_MODEL_ID = "base-model" pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" @@ -247,12 +230,6 @@ async def update_draft(*args, **kwargs): ), ) - async def no_draft(*args, **kwargs): - return DraftResult( - source="test", - result=NoDirective(reason="reject.confident_non_directive"), - ) - monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) asyncio.run( pipe.pipe( @@ -265,13 +242,20 @@ async def no_draft(*args, **kwargs): __chat_id__="chat-state-preserved", ) ) + rejected = asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "n"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-state-preserved", + ) + ) - monkeypatch.setattr(module.Pipe, "_draft_user_input", no_draft) - asyncio.run( + show_state = asyncio.run( pipe.pipe( { "model": "pipe-model", - "messages": [{"role": "user", "content": "prohibit docker"}], + "messages": [{"role": "user", "content": "show state"}], }, __user__={"id": "u1"}, __request__=object(), @@ -279,6 +263,39 @@ async def no_draft(*args, **kwargs): ) ) + assert rejected == "Directive discarded." + assert show_state == "Premise: none\nUse: none\nProhibit: none" + + +def test_pending_approval_does_not_affect_show_state(monkeypatch) -> None: + module = _load_module("owui_with_drafter_pending_show_state", monkeypatch) + pipe = module.Pipe() + pipe.valves.BASE_MODEL_ID = "base-model" + pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" + + async def update_draft(*args, **kwargs): + return DraftResult( + source="test", + result=CanonicalDirective( + text="use docker", + kind=DirectiveKind.USE_ITEM, + operands=MappingProxyType({"item": "docker"}), + ), + ) + + monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) + asyncio.run( + pipe.pipe( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "please use docker"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-pending-show-state", + ) + ) + show_state = asyncio.run( pipe.pipe( { @@ -287,11 +304,11 @@ async def no_draft(*args, **kwargs): }, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-state-preserved", + __chat_id__="chat-pending-show-state", ) ) - assert show_state == "Premise: none\nUse: docker\nProhibit: none" + assert show_state == "Premise: none\nUse: none\nProhibit: none" def test_fallback_to_raw_input_path_preserves_host_behavior(monkeypatch) -> None: @@ -359,7 +376,7 @@ async def update_draft(*args, **kwargs): ) monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) - update = asyncio.run( + proposal = asyncio.run( pipe.pipe( { "model": "pipe-model", @@ -392,7 +409,9 @@ async def no_draft(*args, **kwargs): ) ) - assert update == "State updated." + assert proposal == ( + "This is what I think the directive is:\nuse docker\nApply it? (y/n)" + ) assert passthrough == {"choices": [{"message": {"content": "downstream"}}]} assert len(forwarded) == 1 @@ -545,6 +564,14 @@ async def update_draft(*args, **kwargs): __chat_id__=chat_id, ) ) + asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "y"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__=chat_id, + ) + ) async def no_draft(*args, **kwargs): return DraftResult( From 87132430901489c954bfe925f843ea194b07dd11 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Mon, 10 Aug 2026 03:28:05 -0400 Subject: [PATCH 2/2] fix: make OpenWebUI directive approval one-turn --- .../open_webui_pipe_with_directive_drafter.py | 9 +- ...t_openwebui_pipe_with_directive_drafter.py | 189 +++++++++++++++++- 2 files changed, 190 insertions(+), 8 deletions(-) diff --git a/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py b/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py index cc2a11a..25a924d 100644 --- a/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py +++ b/python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py @@ -87,11 +87,7 @@ def _is_explicit_rejection(message: str) -> bool: def _render_proposal_prompt(directive_text: str) -> str: - return ( - "This is what I think the directive is:\n" - f"{directive_text}\n" - "Apply it? (y/n)" - ) + return f"This is what I think the directive is:\n{directive_text}\nApply it? (y/n)" def _resolve_chat_key( @@ -916,7 +912,8 @@ async def pipe( ) if _is_explicit_rejection(latest_user_text): del _PENDING_PROPOSALS_BY_CHAT_KEY[chat_key] - return "Directive discarded." + return "Directive discarded. No state change was applied." + del _PENDING_PROPOSALS_BY_CHAT_KEY[chat_key] state_before = _snapshot_engine_state(engine) preprocess_error: str | None = None diff --git a/python/tests/test_openwebui_pipe_with_directive_drafter.py b/python/tests/test_openwebui_pipe_with_directive_drafter.py index 94371d1..9ad3c9e 100644 --- a/python/tests/test_openwebui_pipe_with_directive_drafter.py +++ b/python/tests/test_openwebui_pipe_with_directive_drafter.py @@ -97,6 +97,7 @@ def test_canonical_draft_creates_approval_prompt_and_does_not_mutate_state( monkeypatch, ) -> None: module = _load_module("owui_with_drafter_before_step", monkeypatch) + async def fake_draft(*args, **kwargs): return DraftResult( source="test", @@ -213,9 +214,23 @@ async def update_draft(*args, **kwargs): assert show_state == "Premise: none\nUse: docker\nProhibit: none" assert compile_inputs == ["use docker"] + second_follow_up = asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "yes"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-failed-transition", + ) + ) + + assert second_follow_up != "State updated." + assert compile_inputs == ["use docker"] + def test_rejection_does_not_mutate_state(monkeypatch) -> None: - module = _load_module("owui_with_drafter_failed_transition_state_preserved", monkeypatch) + module = _load_module( + "owui_with_drafter_failed_transition_state_preserved", monkeypatch + ) pipe = module.Pipe() pipe.valves.BASE_MODEL_ID = "base-model" pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" @@ -263,9 +278,20 @@ async def update_draft(*args, **kwargs): ) ) - assert rejected == "Directive discarded." + assert rejected == "Directive discarded. No state change was applied." assert show_state == "Premise: none\nUse: none\nProhibit: none" + after_rejection = asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "yes"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-state-preserved", + ) + ) + + assert after_rejection != "State updated." + def test_pending_approval_does_not_affect_show_state(monkeypatch) -> None: module = _load_module("owui_with_drafter_pending_show_state", monkeypatch) @@ -311,6 +337,165 @@ async def update_draft(*args, **kwargs): assert show_state == "Premise: none\nUse: none\nProhibit: none" +def test_unrelated_follow_up_while_pending_does_not_apply_proposal(monkeypatch) -> None: + module = _load_module("owui_with_drafter_pending_unrelated_followup", monkeypatch) + compile_inputs: list[str] = [] + forwarded: list[dict[str, object]] = [] + real_create_engine = module.create_engine + + def create_engine_with_tracking(): + engine = real_create_engine() + original_step = engine.step + + def tracked_step(user_input: str): + compile_inputs.append(user_input) + return original_step(user_input) + + engine.step = tracked_step + return engine + + async def forward( + _: object, payload: dict[str, object], __: object + ) -> dict[str, object]: + forwarded.append(payload) + return {"choices": [{"message": {"content": "downstream"}}]} + + monkeypatch.setattr(module, "create_engine", create_engine_with_tracking) + module.generate_chat_completion = forward + pipe = module.Pipe() + pipe.valves.BASE_MODEL_ID = "base-model" + pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" + + async def update_draft(*args, **kwargs): + return DraftResult( + source="test", + result=CanonicalDirective( + text="use docker", + kind=DirectiveKind.USE_ITEM, + operands=MappingProxyType({"item": "docker"}), + ), + ) + + async def no_draft(*args, **kwargs): + return DraftResult( + source="test", + result=NoDirective(reason="reject.confident_non_directive"), + ) + + monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) + proposal = asyncio.run( + pipe.pipe( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "please use docker"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-pending-unrelated-followup", + ) + ) + + monkeypatch.setattr(module.Pipe, "_draft_user_input", no_draft) + follow_up = asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "hello"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-pending-unrelated-followup", + ) + ) + + assert proposal == ( + "This is what I think the directive is:\nuse docker\nApply it? (y/n)" + ) + assert follow_up == {"choices": [{"message": {"content": "downstream"}}]} + assert compile_inputs == [] + assert len(forwarded) == 1 + + +def test_no_stale_pending_proposal_remains_after_non_approval_response( + monkeypatch, +) -> None: + module = _load_module("owui_with_drafter_no_stale_pending", monkeypatch) + compile_inputs: list[str] = [] + forwarded: list[dict[str, object]] = [] + real_create_engine = module.create_engine + + def create_engine_with_tracking(): + engine = real_create_engine() + original_step = engine.step + + def tracked_step(user_input: str): + compile_inputs.append(user_input) + return original_step(user_input) + + engine.step = tracked_step + return engine + + async def forward( + _: object, payload: dict[str, object], __: object + ) -> dict[str, object]: + forwarded.append(payload) + return {"choices": [{"message": {"content": "downstream"}}]} + + monkeypatch.setattr(module, "create_engine", create_engine_with_tracking) + module.generate_chat_completion = forward + pipe = module.Pipe() + pipe.valves.BASE_MODEL_ID = "base-model" + pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" + + async def update_draft(*args, **kwargs): + return DraftResult( + source="test", + result=CanonicalDirective( + text="use docker", + kind=DirectiveKind.USE_ITEM, + operands=MappingProxyType({"item": "docker"}), + ), + ) + + async def no_draft(*args, **kwargs): + return DraftResult( + source="test", + result=NoDirective(reason="reject.confident_non_directive"), + ) + + monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) + asyncio.run( + pipe.pipe( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "please use docker"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-no-stale-pending", + ) + ) + + monkeypatch.setattr(module.Pipe, "_draft_user_input", no_draft) + asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "hello"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-no-stale-pending", + ) + ) + later_yes = asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "yes"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-no-stale-pending", + ) + ) + + assert later_yes == {"choices": [{"message": {"content": "downstream"}}]} + assert compile_inputs == [] + assert len(forwarded) == 2 + + def test_fallback_to_raw_input_path_preserves_host_behavior(monkeypatch) -> None: module = _load_module("owui_with_drafter_raw", monkeypatch) forwarded: list[dict[str, object]] = []