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..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 @@ -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,18 @@ 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 f"This is what I think the directive is:\n{directive_text}\nApply it? (y/n)" + + def _resolve_chat_key( user: dict[str, Any], chat_id: str | None, @@ -809,6 +823,98 @@ 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. No state change was applied." + del _PENDING_PROPOSALS_BY_CHAT_KEY[chat_key] + state_before = _snapshot_engine_state(engine) preprocess_error: str | None = None drafted_result, preprocess_error = await self._preprocess_user_input( @@ -845,83 +951,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..9ad3c9e 100644 --- a/python/tests/test_openwebui_pipe_with_directive_drafter.py +++ b/python/tests/test_openwebui_pipe_with_directive_drafter.py @@ -89,11 +89,63 @@ 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) + + async def fake_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", fake_draft) + + 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( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "please use docker"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __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 == ( + "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_approval_applies_stored_directive(monkeypatch) -> None: + module = _load_module("owui_with_drafter_failed_transition_followup", monkeypatch) compile_inputs: list[str] = [] real_create_engine = module.create_engine @@ -109,8 +161,11 @@ def tracked_step(user_input: str): 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" - async def fake_draft(*args, **kwargs): + async def update_draft(*args, **kwargs): return DraftResult( source="test", result=CanonicalDirective( @@ -120,13 +175,8 @@ async def fake_draft(*args, **kwargs): ), ) - monkeypatch.setattr(module.Pipe, "_draft_user_input", fake_draft) - - pipe = module.Pipe() - pipe.valves.BASE_MODEL_ID = "base-model" - pipe.valves.PREPROCESSOR_MODEL_ID = "prep-model" - - result = asyncio.run( + monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) + seed = asyncio.run( pipe.pipe( { "model": "pipe-model", @@ -134,27 +184,53 @@ async def fake_draft(*args, **kwargs): }, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-before-step", + __chat_id__="chat-failed-transition", + ) + ) + follow_up = asyncio.run( + pipe.pipe( + {"model": "pipe-model", "messages": [{"role": "user", "content": "y"}]}, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-failed-transition", + ) + ) + show_state = asyncio.run( + pipe.pipe( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "show state"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-failed-transition", ) ) - assert result == "State updated." + assert seed == ( + "This is what I think the directive is:\nuse docker\nApply it? (y/n)" + ) + assert follow_up == "State updated." + 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", + ) + ) -def test_failed_transition_is_rejected_and_follow_up_is_a_new_request( - monkeypatch, -) -> None: - module = _load_module("owui_with_drafter_failed_transition_followup", monkeypatch) - forwarded: list[dict[str, object]] = [] + assert second_follow_up != "State updated." + assert compile_inputs == ["use docker"] - async def forward( - _: object, payload: dict[str, object], __: object - ) -> dict[str, object]: - forwarded.append(payload) - return {"choices": [{"message": {"content": "downstream"}}]} - module.generate_chat_completion = forward +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" @@ -170,7 +246,7 @@ async def update_draft(*args, **kwargs): ) monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) - seed = asyncio.run( + asyncio.run( pipe.pipe( { "model": "pipe-model", @@ -178,61 +254,192 @@ async def update_draft(*args, **kwargs): }, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-failed-transition", + __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", ) ) - async def no_draft(*args, **kwargs): + show_state = asyncio.run( + pipe.pipe( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "show state"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-state-preserved", + ) + ) + + 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) + 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="prohibit docker", - kind=DirectiveKind.PROHIBIT_ITEM, + text="use docker", + kind=DirectiveKind.USE_ITEM, operands=MappingProxyType({"item": "docker"}), ), ) - monkeypatch.setattr(module.Pipe, "_draft_user_input", no_draft) - rejected = asyncio.run( + monkeypatch.setattr(module.Pipe, "_draft_user_input", update_draft) + asyncio.run( pipe.pipe( { "model": "pipe-model", - "messages": [{"role": "user", "content": "prohibit docker"}], + "messages": [{"role": "user", "content": "please use docker"}], }, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-failed-transition", + __chat_id__="chat-pending-show-state", ) ) - async def follow_up_no_directive(*args, **kwargs): + show_state = asyncio.run( + pipe.pipe( + { + "model": "pipe-model", + "messages": [{"role": "user", "content": "show state"}], + }, + __user__={"id": "u1"}, + __request__=object(), + __chat_id__="chat-pending-show-state", + ) + ) + + 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", follow_up_no_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": "yes"}]}, + {"model": "pipe-model", "messages": [{"role": "user", "content": "hello"}]}, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-failed-transition", + __chat_id__="chat-pending-unrelated-followup", ) ) - assert seed == "State updated." - assert rejected == ( - '"docker" is currently in use.\nRemove or replace it before prohibiting it.' + 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_failed_transition_does_not_change_existing_engine_state(monkeypatch) -> None: - module = _load_module( - "owui_with_drafter_failed_transition_state_preserved", monkeypatch - ) +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" @@ -262,36 +469,31 @@ async def no_draft(*args, **kwargs): }, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-state-preserved", + __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": "prohibit docker"}], - }, + {"model": "pipe-model", "messages": [{"role": "user", "content": "hello"}]}, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-state-preserved", + __chat_id__="chat-no-stale-pending", ) ) - - show_state = asyncio.run( + later_yes = asyncio.run( pipe.pipe( - { - "model": "pipe-model", - "messages": [{"role": "user", "content": "show state"}], - }, + {"model": "pipe-model", "messages": [{"role": "user", "content": "yes"}]}, __user__={"id": "u1"}, __request__=object(), - __chat_id__="chat-state-preserved", + __chat_id__="chat-no-stale-pending", ) ) - assert show_state == "Premise: none\nUse: docker\nProhibit: none" + 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: @@ -359,7 +561,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 +594,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 +749,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(