From 8b8265eb4673dab507fadc9dafaed6eceefce4b6 Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 29 Apr 2026 00:02:30 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=BB=BC=E5=90=88=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=E5=8E=86=E5=8F=B2=E4=BC=9A=E8=AF=9D=E6=B3=84=E6=BC=8F?= =?UTF-8?q?=E5=86=85=E9=83=A8=E6=8C=87=E4=BB=A4=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将综合模式执行文本与用户落库文本彻底解耦，确保只持久化用户真实输入；同时在消息读取接口增加旧脏数据过滤，并补充回归测试，避免内部派单指令再次作为用户消息展示。 Made-with: Cursor --- interfaces/admin/chat_api.py | 37 ++++++- runtime/agent_core_attempt.py | 4 +- runtime/agent_core_run.py | 2 + runtime/direct_loop.py | 4 +- runtime/gateway.py | 45 ++++++-- tests/test_chat_api_message_filter.py | 31 ++++++ tests/test_oclaw_gateway_trace.py | 143 ++++++++++++++++++++++++++ 7 files changed, 254 insertions(+), 12 deletions(-) create mode 100644 tests/test_chat_api_message_filter.py diff --git a/interfaces/admin/chat_api.py b/interfaces/admin/chat_api.py index 29989590..9ee18e66 100644 --- a/interfaces/admin/chat_api.py +++ b/interfaces/admin/chat_api.py @@ -222,6 +222,40 @@ def _serialize_message(m: Any) -> dict[str, Any]: } +def _extract_manager_instruction_text(reasoning_content: str) -> str | None: + text = str(reasoning_content or "") + marker = "instruction:\n" + idx = text.find(marker) + if idx < 0: + return None + instr = text[idx + len(marker) :].strip() + return instr or None + + +def _filter_internal_instruction_user_messages(msgs: list[Any]) -> list[Any]: + """Hide legacy polluted user rows that equal manager dispatch instruction text.""" + instruction_texts: set[str] = set() + for m in msgs: + role = str(getattr(m, "role", "") or "") + event_type = str(getattr(m, "event_type", "") or "") + if role != "assistant" or event_type != "reasoning": + continue + instr = _extract_manager_instruction_text(str(getattr(m, "content", "") or "")) + if instr: + instruction_texts.add(instr) + if not instruction_texts: + return msgs + filtered: list[Any] = [] + for m in msgs: + role = str(getattr(m, "role", "") or "") + event_type = str(getattr(m, "event_type", "") or "") + content = str(getattr(m, "content", "") or "").strip() + if role == "user" and event_type == "user_text" and content in instruction_texts: + continue + filtered.append(m) + return filtered + + def _register_stop(session_id: str) -> threading.Event: with _CHAT_STOP_LOCK: ev = threading.Event() @@ -947,9 +981,10 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor raise HTTPException(status_code=404, detail="session_not_found") meta = store.get_session_messages_meta(session_id) msgs = store.get_messages(session_id=session_id, limit=_CHAT_MSG_LIMIT) + msgs = _filter_internal_instruction_user_messages(msgs) return { "ok": True, - "message_count": int(meta.message_count or 0), + "message_count": len(msgs), "messages": [_serialize_message(m) for m in msgs], } diff --git a/runtime/agent_core_attempt.py b/runtime/agent_core_attempt.py index 636d8699..92379768 100644 --- a/runtime/agent_core_attempt.py +++ b/runtime/agent_core_attempt.py @@ -65,6 +65,7 @@ class AttemptRunnerInput: wire_policy_role: str | None = None prompt_build_context: dict[str, Any] | None = None turn_uuid: str | None = None + persisted_user_text: str | None = None @dataclass(frozen=True) @@ -133,6 +134,7 @@ def run_attempt(*, store: Any, data: AttemptRunnerInput) -> AttemptRunnerOutput: workspace_dir=data.workspace_dir, memory_context=data.memory_context, persist_user_message=bool(data.persist_user_message), + persisted_user_text=data.persisted_user_text, skill_binding_role=data.skill_binding_role, wire_policy_role=data.wire_policy_role, prompt_build_context=data.prompt_build_context, @@ -144,7 +146,7 @@ def run_attempt(*, store: Any, data: AttemptRunnerInput) -> AttemptRunnerOutput: session_id=data.msg.session_id, tenant_id=data.msg.tenant_id, user_id=data.msg.user_id, - user_text=data.msg.text, + user_text=str(data.persisted_user_text if data.persisted_user_text is not None else data.msg.text or ""), assistant_text=outcome.final_text, turn_uuid=outcome.turn_uuid, ) diff --git a/runtime/agent_core_run.py b/runtime/agent_core_run.py index ee2ced58..64afc491 100644 --- a/runtime/agent_core_run.py +++ b/runtime/agent_core_run.py @@ -47,6 +47,7 @@ class AgentCoreRunInput: oclaw_worker_id: str | None = None skill_binding_role: str | None = None wire_policy_role: str | None = None + persisted_user_text: str | None = None @dataclass(frozen=True) @@ -218,6 +219,7 @@ def run_agent_core(*, store: Any, data: AgentCoreRunInput) -> AgentCoreRunOutput wire_policy_role=data.wire_policy_role, prompt_build_context=(dict(data.msg.metadata or {}) if isinstance(data.msg.metadata, dict) else None), turn_uuid=user_turn_uuid, + persisted_user_text=data.persisted_user_text, ), ) attempts.append(out.state) diff --git a/runtime/direct_loop.py b/runtime/direct_loop.py index ed9a3be7..bf6512ec 100644 --- a/runtime/direct_loop.py +++ b/runtime/direct_loop.py @@ -920,6 +920,7 @@ def run_oclaw_direct_loop( workspace_dir: str | None = None, memory_context: OclawMemoryContext | None = None, persist_user_message: bool = True, + persisted_user_text: str | None = None, tool_signature_budget: int = 2, skill_binding_role: str | None = None, wire_policy_role: str | None = None, @@ -929,11 +930,12 @@ def run_oclaw_direct_loop( """A minimal oclaw-style loop: model -> tool_uses -> execute -> tool_results -> continue.""" _check_stop(should_stop) turn_uuid = str(turn_uuid or "").strip() or str(uuid.uuid4()) + persisted_text = str(user_text if persisted_user_text is None else persisted_user_text or "") if persist_user_message: store.add_message( session_id=session_id, role="user", - content=str(user_text or ""), + content=persisted_text, attachments=attachments, turn_uuid=turn_uuid, event_type="user_text", diff --git a/runtime/gateway.py b/runtime/gateway.py index 0880a14c..c7a1e45b 100644 --- a/runtime/gateway.py +++ b/runtime/gateway.py @@ -105,6 +105,26 @@ class OclawGateway: def __init__(self, *, store: Any): self.store = store + @staticmethod + def _looks_like_manager_instruction(reply: str, instruction: str) -> bool: + r = str(reply or "").strip() + ins = str(instruction or "").strip() + if not r or not ins: + return False + if r == ins: + return True + if len(ins) >= 16 and ins in r: + return True + # Heuristic: prefix overlap is usually enough for instruction leakage. + a = r[:120] + b = ins[:120] + common = 0 + for x, y in zip(a, b): + if x != y: + break + common += 1 + return common >= 24 + @staticmethod def _parse_json_object(text: str) -> dict[str, Any] | None: t = str(text or "").strip() @@ -1178,6 +1198,7 @@ class OclawGateway: store=self.store, data=AgentCoreRunInput( msg=exec_msg, + persisted_user_text=str(msg.text or ""), lang=lang, system_prompt=sys_prompt, model=model, @@ -1202,15 +1223,12 @@ class OclawGateway: executed_turn_uuid = str(getattr(core_out.outcome, "turn_uuid", "") or "") specialist_reply = str(core_out.outcome.final_text or "") if manager_memory_mode: - # manager_memory: write memory silently, but keep dialog output independent. - # User-facing reply comes from manager dispatch instruction_text. - reply = str(manager_instruction_text or "").strip() - if not reply: - reply = ( - "已执行记忆写入。" - if not str(lang or "").startswith("en") - else "Memory write executed." - ) + # manager_memory: never expose manager dispatch instruction_text to end users. + reply = ( + "已执行记忆写入。" + if not str(lang or "").startswith("en") + else "Memory write executed." + ) elif interaction_mode == "comprehensive": reply = self._manager_finalize_output( msg=msg, @@ -1221,6 +1239,15 @@ class OclawGateway: memory_enabled=memory_enabled, on_token=on_token, ) + if self._looks_like_manager_instruction(reply, manager_instruction_text): + if not self._looks_like_manager_instruction(specialist_reply, manager_instruction_text): + reply = str(specialist_reply or "").strip() + else: + reply = ( + "抱歉，我暂时无法给出可展示的结果，请稍后再试。" + if not str(lang or "").startswith("en") + else "Sorry, no user-safe result is available right now. Please try again later." + ) else: reply = specialist_reply except Exception as exc: diff --git a/tests/test_chat_api_message_filter.py b/tests/test_chat_api_message_filter.py new file mode 100644 index 00000000..13b80388 --- /dev/null +++ b/tests/test_chat_api_message_filter.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from oclaw.interfaces.admin.chat_api import _filter_internal_instruction_user_messages + + +def test_filter_internal_instruction_user_messages_hides_polluted_user_row() -> None: + polluted_text = "请检查并修复网关启动失败。" + rows = [ + SimpleNamespace(role="user", event_type="user_text", content=polluted_text), + SimpleNamespace( + role="assistant", + event_type="reasoning", + content=f"任务分配\nspecialist=ops\ninstruction:\n{polluted_text}", + ), + SimpleNamespace(role="assistant", event_type="assistant_text", content="已修复"), + ] + out = _filter_internal_instruction_user_messages(rows) + assert len(out) == 2 + assert [str(getattr(x, "event_type", "")) for x in out] == ["reasoning", "assistant_text"] + + +def test_filter_internal_instruction_user_messages_keeps_normal_user_rows() -> None: + rows = [ + SimpleNamespace(role="user", event_type="user_text", content="你好"), + SimpleNamespace(role="assistant", event_type="assistant_text", content="你好，有什么我可以帮你？"), + ] + out = _filter_internal_instruction_user_messages(rows) + assert len(out) == 2 + assert str(getattr(out[0], "content", "")) == "你好" diff --git a/tests/test_oclaw_gateway_trace.py b/tests/test_oclaw_gateway_trace.py index b2707501..d359424d 100644 --- a/tests/test_oclaw_gateway_trace.py +++ b/tests/test_oclaw_gateway_trace.py @@ -248,6 +248,7 @@ def test_gateway_comprehensive_mode_manager_first_selects_specialist(monkeypatch def _run_agent_core(**kwargs): data = kwargs.get("data") captured["exec_text"] = getattr(getattr(data, "msg", None), "text", None) + captured["persisted_user_text"] = getattr(data, "persisted_user_text", None) return SimpleNamespace(outcome=SimpleNamespace(final_text="specialist_answer")) monkeypatch.setattr("oclaw.runtime.gateway.run_agent_core", _run_agent_core) @@ -274,6 +275,7 @@ def test_gateway_comprehensive_mode_manager_first_selects_specialist(monkeypatch assert out.selected_specialist == "image" assert chosen.get("sid") == "image" assert captured.get("exec_text") == "Please edit the image background." + assert captured.get("persisted_user_text") == "edit this image background" def test_gateway_comprehensive_mode_writes_task_assignment_reasoning(monkeypatch: pytest.MonkeyPatch) -> None: @@ -523,6 +525,147 @@ def test_gateway_comprehensive_mode_ignores_manager_self_and_dispatches_speciali assert out.reply_text == "manager_self_final" +def test_gateway_manager_memory_mode_does_not_leak_instruction_text(monkeypatch: pytest.MonkeyPatch) -> None: + class Store: + def get_setting(self, _k: str) -> str: + return "" + + def add_trace_event(self, **_kwargs: object) -> None: + return None + + class _ManagerModel: + def __init__(self) -> None: + self.calls = 0 + + def chat(self, _messages, _tools, *, on_token=None): + self.calls += 1 + if self.calls == 1: + return LLMResponse( + content=( + '{"route":{"kind":"manager_memory","specialist":"manager","reason":"memory write"},' + '"dispatch":{"instruction_text":"内部指令：写入记忆库，不可对用户展示。",' + '"memory_write_text":"请写入记忆：用户偏好咖啡。"}}' + ), + tool_calls=[], + ) + return LLMResponse(content="manager_memory_internal_result", tool_calls=[]) + + class _Exec: + def __init__(self, model=None): + self.model = model + self.tools = object() + self.system_prompt = "" + + monkeypatch.setattr( + "oclaw.runtime.gateway.get_manager_prompt_prebuild", + lambda **kwargs: { + "manager_context": "manager", + "allowed_fixed": ("generalist", "ops", "image", "memory"), + "allowed_fixed_quoted": '"generalist", "ops", "image", "memory"', + }, + ) + + def _run_agent_core(**kwargs): + class _Outcome: + final_text = "manager_memory_internal_result" + + class _Out: + outcome = _Outcome() + + return _Out() + + monkeypatch.setattr("oclaw.runtime.gateway.run_agent_core", _run_agent_core) + + gw = OclawGateway(store=Store()) + msg = StandardMessage( + session_id="sid-mm-1", + tenant_id="t1", + user_id="u1", + role="user", + channel="admin_chat", + text="帮我记住我喜欢咖啡", + attachments=[], + metadata={"interaction_mode": "comprehensive"}, + ) + out = gw.handle_turn(msg=msg, lang="zh", executor=_Exec(model=_ManagerModel())) + assert out.interaction_mode == "comprehensive" + assert out.reply_text == "已执行记忆写入。" + assert "内部指令" not in out.reply_text + + +def test_gateway_comprehensive_mode_suppresses_instruction_echo(monkeypatch: pytest.MonkeyPatch) -> None: + class Store: + def get_setting(self, _k: str) -> str: + return "" + + def add_trace_event(self, **_kwargs: object) -> None: + return None + + instruction = "用户说厉害；请友好回应，表达感谢并询问是否有具体问题。" + + class _ManagerModel: + def __init__(self) -> None: + self.calls = 0 + + def chat(self, _messages, _tools, *, on_token=None): + self.calls += 1 + if self.calls == 1: + return LLMResponse( + content=( + '{"route":{"specialist":"generalist","reason":"smalltalk"},' + f'"dispatch":{{"instruction_text":"{instruction}"}}' + "}" + ), + tool_calls=[], + ) + # Simulate finalize path echoing instruction. + return LLMResponse(content=instruction, tool_calls=[]) + + class _Exec: + def __init__(self, model=None): + self.model = model + self.tools = object() + self.system_prompt = "" + + monkeypatch.setattr( + "oclaw.runtime.gateway.get_manager_prompt_prebuild", + lambda **kwargs: { + "manager_context": "manager", + "allowed_fixed": ("generalist", "ops", "image", "memory"), + "allowed_fixed_quoted": '"generalist", "ops", "image", "memory"', + }, + ) + + # Specialist output also echoes instruction -> should still be suppressed. + def _run_agent_core(**kwargs): + class _Outcome: + final_text = instruction + + class _Out: + outcome = _Outcome() + + return _Out() + + monkeypatch.setattr("oclaw.runtime.gateway.run_agent_core", _run_agent_core) + + gw = OclawGateway(store=Store()) + msg = StandardMessage( + session_id="sid-echo-1", + tenant_id="t1", + user_id="u1", + role="user", + channel="admin_chat", + text="你真厉害", + attachments=[], + metadata={"interaction_mode": "comprehensive"}, + ) + out = gw.handle_turn(msg=msg, lang="zh", executor=_Exec(model=_ManagerModel())) + assert out.interaction_mode == "comprehensive" + assert "用户说厉害" not in out.reply_text + assert "请友好回应" not in out.reply_text + assert out.reply_text == "抱歉，我暂时无法给出可展示的结果，请稍后再试。" + + def test_gateway_command_hook_uses_parsed_command_and_context(monkeypatch: pytest.MonkeyPatch) -> None: calls: list[dict] = []