修复综合模式历史会话泄漏内部指令。

将综合模式执行文本与用户落库文本彻底解耦,确保只持久化用户真实输入;同时在消息读取接口增加旧脏数据过滤,并补充回归测试,避免内部派单指令再次作为用户消息展示。

Made-with: Cursor
This commit is contained in:
oliver 2026-04-29 00:02:30 +08:00
parent 048e8ec1c1
commit 8b8265eb46
7 changed files with 254 additions and 12 deletions

View file

@ -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],
}

View file

@ -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,
)

View file

@ -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)

View file

@ -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",

View file

@ -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:

View file

@ -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", "")) == "你好"

View file

@ -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] = []