JOJO/agent-Specialization
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(goal): 目标状态改为对话级存储，修复多对话目标串扰与进度事件串台

Browse Source 
- goal_state.json(工作区级单实例) → goal_states/<conversation_id>.json(对话级)
- 移除 start_conversation_id 冗余字段与“工作区唯一目标”“压缩改id”等过时注释
- goal 进度/审核事件快照补 conversation_id，前端现有过滤逻辑自动生效
- 清理 tasks/media.py、tasks/helpers.py 未使用的残留 import
This commit is contained in:
JOJO 2026-08-19 13:53:51 +08:00
parent 4b381f3206
commit 17c49f6258
7 changed files with 100 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
modules/goal_state_manager.py
View File

@ -1,23 +1,28 @@
"""目标模式（Goal Mode）的工作区级状态管理。
"""目标模式（Goal Mode）的对话级状态管理。
每个工作区/项目最多存在一个活动目标。状态落盘到 `{data_dir}/goal_state.json`，
以便在切换对话、对话压缩（conversation_id 变化）、进程重启后仍能维持。
每个对话各自持有独立的目标状态，互不影响。状态落盘到
`{data_dir}/goal_states/<conversation_id>.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/<conversation_id>.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(),

16
server/chat_flow_task_main.py
View File

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

63
server/goal_flow.py
View File

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

18
server/tasks/api.py
View File

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

1
server/tasks/helpers.py
View File

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

1
server/tasks/media.py
View File

@ -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<body>.*?)\n---\s*\n?", re.S)

6
server/tasks/models.py
View File

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