diff --git a/modules/goal_state_manager.py b/modules/goal_state_manager.py index 7eb562e4..d2acad1b 100644 --- a/modules/goal_state_manager.py +++ b/modules/goal_state_manager.py @@ -1,23 +1,28 @@ -"""目标模式(Goal Mode)的工作区级状态管理。 +"""目标模式(Goal Mode)的对话级状态管理。 -每个工作区/项目最多存在一个活动目标。状态落盘到 `{data_dir}/goal_state.json`, -以便在切换对话、对话压缩(conversation_id 变化)、进程重启后仍能维持。 +每个对话各自持有独立的目标状态,互不影响。状态落盘到 +`{data_dir}/goal_states/.json`,在切换对话、对话压缩、 +进程重启后仍能维持(压缩不会改变 conversation_id,压缩 handoff 重入时 +由主循环入口按当前 conversation_id 重新加载本对话状态并续注提示词)。 设计要点: -- 不依赖 web_terminal,只接受一个 data_dir,便于单测与复用。 +- 不依赖 web_terminal,只接受 data_dir 与 conversation_id,便于单测与复用。 - token 基线对齐工作区级累计 `{data_dir}/token_totals.json`(input/output/total)。 + 注意:该累计是整个工作区所有对话共享的,多对话并发运行时 max_tokens + 边界判定是近似值(可能因其他对话的消耗而提前触发)。 - review_history 保存历轮 {main_output, review_reply},用于构建交叉结构审核文本。 """ from __future__ import annotations import json +import re import time from copy import deepcopy from pathlib import Path from typing import Any, Dict, List, Optional, Union -GOAL_STATE_FILENAME = "goal_state.json" +GOAL_STATES_DIRNAME = "goal_states" # 状态机 STATUS_RUNNING = "running" @@ -36,6 +41,16 @@ REVIEW_MODE_ACTIVE = "active" PathLike = Union[str, Path] +_SAFE_CONVERSATION_ID = re.compile(r"^[A-Za-z0-9_-]+$") + + +def _validate_conversation_id(conversation_id: str) -> str: + """conversation_id 用作状态文件名,必须是安全字符,防止路径穿越。""" + cid = str(conversation_id or "").strip() + if not cid or not _SAFE_CONVERSATION_ID.match(cid): + raise ValueError(f"非法 conversation_id: {conversation_id!r}") + return cid + def _empty_state() -> Dict[str, Any]: return { @@ -45,7 +60,6 @@ def _empty_state() -> Dict[str, Any]: "review_mode": REVIEW_MODE_READONLY, "max_turns": 5, "max_tokens": None, - "start_conversation_id": None, "started_at": None, "turn_count": 0, "token_baseline": {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}, @@ -58,21 +72,22 @@ def _empty_state() -> Dict[str, Any]: class GoalStateManager: - """工作区级目标状态。一个实例对应一个工作区的 goal_state.json。""" + """对话级目标状态。一个实例对应一个对话的 goal_states/.json。""" - def __init__(self, data_dir: PathLike): + def __init__(self, data_dir: PathLike, conversation_id: str): self.data_dir = Path(data_dir).expanduser() + self.conversation_id = _validate_conversation_id(conversation_id) self.state: Dict[str, Any] = self.load() # ------------------------------------------------------------------ 路径/持久化 def _path(self) -> Path: - return self.data_dir / GOAL_STATE_FILENAME + return self.data_dir / GOAL_STATES_DIRNAME / f"{self.conversation_id}.json" @classmethod - def load_from(cls, data_dir: PathLike) -> "GoalStateManager": - """便捷构造:等价于 GoalStateManager(data_dir)。""" - return cls(data_dir) + def load_from(cls, data_dir: PathLike, conversation_id: str) -> "GoalStateManager": + """便捷构造:等价于 GoalStateManager(data_dir, conversation_id)。""" + return cls(data_dir, conversation_id) def load(self) -> Dict[str, Any]: path = self._path() @@ -112,7 +127,6 @@ class GoalStateManager: review_mode: str, max_turns: Optional[int], max_tokens: Optional[int], - conversation_id: Optional[str], token_baseline: Optional[Dict[str, int]] = None, tool_call_baseline: int = 0, ) -> Dict[str, Any]: @@ -131,7 +145,6 @@ class GoalStateManager: "review_mode": rm, "max_turns": int(max_turns) if max_turns else None, "max_tokens": int(max_tokens) if max_tokens else None, - "start_conversation_id": conversation_id, "started_at": time.time(), "turn_count": 0, "token_baseline": baseline, @@ -194,8 +207,15 @@ class GoalStateManager: return deepcopy(self.state) def clear(self) -> None: - """完全清除目标状态(用户取消)。""" + """完全清除本对话的目标状态。""" self.state = _empty_state() + path = self._path() + try: + if path.exists(): + path.unlink() + return + except OSError: + pass self.save() # ------------------------------------------------------------------ 边界判定 @@ -267,6 +287,7 @@ class GoalStateManager: except Exception: pass return { + "conversation_id": self.conversation_id, "goal": self.get_goal(), "status": self.state.get("status"), "turn_count": self.get_turn(), diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index b67e9281..ae865227 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -1868,12 +1868,11 @@ async def handle_task_with_sender( goal_text=message, current_tool_calls=0, ) - # 本标记只表示“本次用户显式开启目标模式”。目标状态已写入 - # goal_state.json 后必须立即消费掉,否则自动深层压缩后的递归 - # handoff 会再次进入 handle_task_with_sender,把压缩引导语误当作 - # 新目标并重置目标文本、轮数、token、工具次数和开始时间。 + # 本标记只表示“本次用户显式开启目标模式”,写入对话级目标状态后 + # 必须立即消费掉:自动深层压缩会递归重入 handle_task_with_sender, + # 若不消费,压缩引导语会被误当作新目标再次启动目标模式。 setattr(web_terminal, "_goal_mode_requested", False) - if goal_is_active(workspace): + if goal_is_active(workspace, conversation_id): inject_goal_prompt( web_terminal=web_terminal, messages=messages, @@ -1884,6 +1883,7 @@ async def handle_task_with_sender( web_terminal=web_terminal, workspace=workspace, sender=sender, + conversation_id=conversation_id, total_tool_calls=0, ) except Exception as exc: @@ -1969,6 +1969,7 @@ async def handle_task_with_sender( web_terminal=web_terminal, workspace=workspace, sender=sender, + conversation_id=conversation_id, total_tool_calls=total_tool_calls, ) except Exception as exc: @@ -2227,7 +2228,7 @@ async def handle_task_with_sender( debug_log("没有工具调用,结束迭代") # === 目标模式:turn 结束拦截 === try: - if goal_is_active(workspace): + if goal_is_active(workspace, conversation_id): goal_result = await handle_goal_after_turn( web_terminal=web_terminal, workspace=workspace, @@ -2311,11 +2312,12 @@ async def handle_task_with_sender( # 更新统计 total_tool_calls += len(tool_calls) try: - if goal_is_active(workspace): + if goal_is_active(workspace, conversation_id): emit_goal_progress( web_terminal=web_terminal, workspace=workspace, sender=sender, + conversation_id=conversation_id, total_tool_calls=total_tool_calls, ) except Exception as exc: diff --git a/server/goal_flow.py b/server/goal_flow.py index d35f4a16..e98eafc6 100644 --- a/server/goal_flow.py +++ b/server/goal_flow.py @@ -1,7 +1,8 @@ """目标模式(Goal Mode)在 Web 任务主循环中的编排逻辑。 -将启动、提示词注入、turn 结束后的审核/续命/停止判定集中在此, -尽量减少对 chat_flow_task_main.py 的侵入。 +目标状态是对话级的(见 modules/goal_state_manager.py):每个对话独立持有 +自己的目标,多对话并发互不影响。本模块将启动、提示词注入、turn 结束后的 +审核/续命/停止判定集中在此,尽量减少对 chat_flow_task_main.py 的侵入。 """ from __future__ import annotations @@ -57,17 +58,18 @@ def maybe_start_goal( *, web_terminal, workspace, - conversation_id: Optional[str], + conversation_id: str, goal_text: str, current_tool_calls: int, ) -> bool: - """启动一个新目标。 + """为本对话启动一个新目标。 - 本函数语义是“本次用户显式开启目标模式”而不是“续接已有目标”。 - 因此必须覆盖工作区级 goal_state.json 中的旧目标状态(包括 running/done/stopped、 - review_history、token 基线、开始时间等),避免新对话复用上一轮目标。 + 目标状态是对话级的:启动会覆盖本对话自己的旧目标状态(running/done/stopped、 + review_history、token 基线、开始时间等),不影响其他对话的目标。 """ - gsm = GoalStateManager(workspace.data_dir) + if not conversation_id: + return False + gsm = GoalStateManager(workspace.data_dir, conversation_id) cfg = load_personalization_config(workspace.data_dir) review_mode = cfg.get("goal_review_mode") or "readonly" end_conditions = cfg.get("goal_end_conditions") or ["max_turns"] @@ -78,7 +80,6 @@ def maybe_start_goal( review_mode=review_mode, max_turns=max_turns, max_tokens=max_tokens, - conversation_id=conversation_id, token_baseline=snapshot_token_baseline(web_terminal), tool_call_baseline=current_tool_calls, ) @@ -105,8 +106,13 @@ def inject_goal_prompt( ) -def goal_is_active(workspace) -> bool: - return GoalStateManager(workspace.data_dir).is_active() +def goal_is_active(workspace, conversation_id: Optional[str]) -> bool: + if not conversation_id: + return False + try: + return GoalStateManager(workspace.data_dir, conversation_id).is_active() + except ValueError: + return False def _emit_snapshot(sender, event: str, gsm: GoalStateManager, *, total_tokens: int, tool_calls: int, extra: Optional[Dict] = None): @@ -126,11 +132,18 @@ def emit_goal_progress( web_terminal, workspace, sender: Optional[Callable[[str, Dict[str, Any]], None]], + conversation_id: Optional[str], total_tool_calls: int = 0, extra: Optional[Dict[str, Any]] = None, ) -> None: - """向前端广播当前目标模式进度,用于刷新后恢复与进度弹窗动态数字。""" - gsm = GoalStateManager(workspace.data_dir) + """广播本对话的目标模式进度(快照自带 conversation_id,前端按对话过滤), + 用于刷新后恢复与进度弹窗动态数字。""" + if not conversation_id: + return + try: + gsm = GoalStateManager(workspace.data_dir, conversation_id) + except ValueError: + return if not gsm.state: return total_tokens = _load_token_total(web_terminal) @@ -158,12 +171,17 @@ async def handle_goal_after_turn( """主模型某轮结束(无 tool_calls)后的目标处理。 返回 {'action': 'inactive'|'stop'|'done'|'continue', ...}: - - inactive:当前无活动目标,调用方照常结束本次任务。 + - inactive:本对话无活动目标,调用方照常结束本次任务。 - stop:目标因空转/边界停止,调用方照常结束(事件已发)。 - done:目标达成,调用方照常结束(事件已发)。 - continue:已注入续命消息,调用方应 continue 回主循环开下一轮。 """ - gsm = GoalStateManager(workspace.data_dir) + if not conversation_id: + return {"action": "inactive"} + try: + gsm = GoalStateManager(workspace.data_dir, conversation_id) + except ValueError: + return {"action": "inactive"} if not gsm.is_active(): return {"action": "inactive"} @@ -192,7 +210,7 @@ async def handle_goal_after_turn( payload_text = gsm.build_review_payload_text(assistant_content) if callable(sender): try: - sender("goal_review_progress", {"progress": {"stage": "start", "message": "开始审核"}}) + sender("goal_review_progress", {"conversation_id": conversation_id, "progress": {"stage": "start", "message": "开始审核"}}) except Exception: pass try: @@ -200,7 +218,7 @@ async def handle_goal_after_turn( def _progress(progress: Dict[str, Any]) -> None: if callable(sender): try: - sender("goal_review_progress", {"progress": progress}) + sender("goal_review_progress", {"conversation_id": conversation_id, "progress": progress}) except Exception: pass @@ -240,9 +258,14 @@ async def handle_goal_after_turn( return {"action": "continue", "message": message} -def stop_goal_user_cancel(*, web_terminal, workspace, sender, total_tool_calls: int = 0) -> bool: - """用户停止任务时,若有活动目标则一并停止。返回是否有活动目标被停止。""" - gsm = GoalStateManager(workspace.data_dir) +def stop_goal_user_cancel(*, web_terminal, workspace, sender, conversation_id: Optional[str], total_tool_calls: int = 0) -> bool: + """用户停止任务时,若本对话有活动目标则一并停止。返回是否有活动目标被停止。""" + if not conversation_id: + return False + try: + gsm = GoalStateManager(workspace.data_dir, conversation_id) + except ValueError: + return False if not gsm.is_active(): return False total_tokens = _load_token_total(web_terminal) diff --git a/server/tasks/api.py b/server/tasks/api.py index 9411a78e..b8118c6d 100644 --- a/server/tasks/api.py +++ b/server/tasks/api.py @@ -124,16 +124,16 @@ def create_task_api(): message_source = payload.get("message_source") max_iterations = payload.get("max_iterations") goal_mode = bool(payload.get("goal_mode")) - # 用户显式发送非目标模式消息时,若工作区仍有上一轮残留的活动目标,先清理, - # 避免新对话错误继承旧目标状态。 - if not goal_mode: + # 用户显式发送非目标模式消息时,若本对话仍有残留的活动目标,先停止它, + # 避免本对话错误延续旧目标。目标状态是对话级的,不影响其他对话。 + if not goal_mode and conversation_id: try: _terminal, workspace = get_user_resources(username, workspace_id, conversation_id=conversation_id) if workspace: - gsm = GoalStateManager(workspace.data_dir) + gsm = GoalStateManager(workspace.data_dir, conversation_id) if gsm.is_active(): - gsm.mark_stopped("new_conversation") - debug_log(f"[Goal] 新任务未开启目标模式,清理工作区残留目标状态") + gsm.mark_stopped("new_message_without_goal") + debug_log(f"[Goal] 新消息未开启目标模式,停止本对话残留目标状态") except Exception as exc: debug_log(f"[Goal] 新任务清理残留目标状态失败: {exc}") skill_context_messages: List[Dict[str, str]] = [] @@ -287,11 +287,11 @@ def cancel_task_api(task_id: str): try: if ok and rec.workspace_id: _, workspace = get_user_resources(username, rec.workspace_id, conversation_id=rec.conversation_id) - if workspace: - gsm = GoalStateManager(workspace.data_dir) + if workspace and rec.conversation_id: + gsm = GoalStateManager(workspace.data_dir, rec.conversation_id) if gsm.is_active(): gsm.mark_stopped(REASON_USER_CANCEL) - debug_log(f"[Goal] 用户取消任务 {task_id},同步停止工作区目标模式") + debug_log(f"[Goal] 用户取消任务 {task_id},同步停止本对话目标模式") except Exception as exc: debug_log(f"[Goal] 取消任务时停止目标模式失败: {exc}") return jsonify({"success": True}) diff --git a/server/tasks/helpers.py b/server/tasks/helpers.py index ef5f624c..19ea6ca2 100644 --- a/server/tasks/helpers.py +++ b/server/tasks/helpers.py @@ -21,7 +21,6 @@ from server.state import stop_flags from server.utils_common import debug_log, log_conn_diag from utils.host_workspace_debug import write_host_workspace_debug from config import DATA_DIR, WORKSPACE_SKILLS_DIRNAME -from modules.goal_state_manager import GoalStateManager, REASON_USER_CANCEL from server.tasks.models import TaskRecord diff --git a/server/tasks/media.py b/server/tasks/media.py index 8b106c7a..f2074aa9 100644 --- a/server/tasks/media.py +++ b/server/tasks/media.py @@ -21,7 +21,6 @@ from server.state import stop_flags from server.utils_common import debug_log, log_conn_diag from utils.host_workspace_debug import write_host_workspace_debug from config import DATA_DIR, WORKSPACE_SKILLS_DIRNAME -from modules.goal_state_manager import GoalStateManager, REASON_USER_CANCEL SKILL_FRONTMATTER_RE = re.compile(r"^---\s*\n(?P.*?)\n---\s*\n?", re.S) diff --git a/server/tasks/models.py b/server/tasks/models.py index a8de5788..1029c983 100644 --- a/server/tasks/models.py +++ b/server/tasks/models.py @@ -285,11 +285,11 @@ class TaskManager: if rec.workspace_id: try: _, workspace = get_user_resources(username, rec.workspace_id, conversation_id=rec.conversation_id) - if workspace: - gsm = GoalStateManager(workspace.data_dir) + if workspace and rec.conversation_id: + gsm = GoalStateManager(workspace.data_dir, rec.conversation_id) if gsm.is_active(): gsm.mark_stopped(REASON_USER_CANCEL) - debug_log(f"[Goal] 用户取消任务 {task_id},同步停止工作区目标模式") + debug_log(f"[Goal] 用户取消任务 {task_id},同步停止本对话目标模式") except Exception as exc: debug_log(f"[Goal] 取消任务时停止目标模式失败: {exc}")