chat: 持久化继续消息，chat-history 交替历史
基线 main 811912c（PR #125 Muse 风格之后）：服务端与单测与 9538825 相同；React 侧在 Muse 风格 ContinuePanel 上重放 chat-history 集成（新版输入框占位文案保持 Muse 风格的“消息”）， console.html 由新版 src 重新构建。 - 新表 task_continue_messages：POST /continue 的用户消息与状态改回 queued 同一事务落盘，409 被拒不留 phantom 消息 - GET /tasks/{id}/chat-history：prompt + 继续 + 后续消息 + 链路按 时间交错拼好返回；部署前继续以 legacy_continue_count 如实标注 - React ContinuePanel 优先用 chat-history，旧服务端自动回退
This commit is contained in:
host committed
1 parent
811912c841
commit
f02c513729
9 files changed
+524
-40
No files matched your search
+232
-1
@@ -1215,6 +1215,97 @@ def _followup_to_dict(row):
|
||||
}
|
||||
|
||||
|
||||
def _ensure_continue_messages_table(conn):
|
||||
"""幂等建继续消息表：POST /continue 接受的每条用户消息都在
|
||||
与状态改回 queued 的同一事务里落盘（见 _continue_task），
|
||||
否则继续消息只活在 resume_message 列里、worker 跑完即清，
|
||||
聊天历史永远拼不出用户那一半。老库第一次跑到这里时自动
|
||||
补表，不丢数据；此前已发生的继续（audit 里 task_continued
|
||||
有、表里没有）无法追回，chat-history 会如实标注缺口数，
|
||||
不编造。列写法与 followups 表一致（DOUBLE PRECISION 在
|
||||
SQLite 与 Postgres 下都可用）。"""
|
||||
conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS task_continue_messages("
|
||||
"id TEXT PRIMARY KEY, "
|
||||
"task_id TEXT, "
|
||||
"project_id TEXT, "
|
||||
"body TEXT, "
|
||||
"created_by TEXT, "
|
||||
"created_at DOUBLE PRECISION)"
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_continue_messages_task"
|
||||
" ON task_continue_messages(task_id, created_at, id)")
|
||||
|
||||
|
||||
def _continue_message_to_dict(row):
|
||||
return {
|
||||
"id": row["id"],
|
||||
"task_id": row["task_id"],
|
||||
"project_id": row["project_id"],
|
||||
"body": row["body"],
|
||||
"created_by": row["created_by"],
|
||||
"created_at": row["created_at"],
|
||||
}
|
||||
|
||||
|
||||
def _chat_message_from_trace(entry):
|
||||
"""链路条目 -> 聊天消息（与 React 的 mapTrace.ts 同款映射，
|
||||
两边改动要同步）：message 是助手回复，exec/file/tool/call
|
||||
是工具气泡，plan/reasoning/agent-start/agent-done 是居中
|
||||
状态提示。链路条目本身没有发送方字段，这里不编造角色，
|
||||
只按 kind 分流。"""
|
||||
kind = entry.get("kind") or ""
|
||||
eid = entry.get("id") or ""
|
||||
if kind == "message":
|
||||
text = ((("[" + entry["phase"] + "] ") if entry.get("phase")
|
||||
else "") + (entry.get("text") or ""))
|
||||
return {"id": eid, "role": "assistant", "source": "trace",
|
||||
"kind": kind, "label": "助手", "text": text}
|
||||
if kind in ("exec", "file", "tool", "call"):
|
||||
label = {"exec": "命令", "file": "文件",
|
||||
"tool": "工具"}.get(kind, "调用")
|
||||
if kind == "exec":
|
||||
# 与 React traceLineText 同款：缺失字段为空，不渲染 None。
|
||||
def _s(v):
|
||||
return "" if v is None else str(v)
|
||||
text = "%s（exit=%s，%sms，cwd=%s）" % (
|
||||
entry.get("command") or "", _s(entry.get("exitCode")),
|
||||
_s(entry.get("durationMs")), entry.get("cwd") or "")
|
||||
elif kind == "file":
|
||||
text = "%s %s %s" % (entry.get("path") or "",
|
||||
entry.get("change") or "",
|
||||
entry.get("status") or "")
|
||||
elif kind == "tool":
|
||||
text = "%s/%s %s" % (entry.get("server") or "",
|
||||
entry.get("tool") or "",
|
||||
entry.get("status") or "")
|
||||
else:
|
||||
text = entry.get("name") or ""
|
||||
return {"id": eid, "role": "tool", "source": "trace",
|
||||
"kind": kind, "label": label, "text": text}
|
||||
if kind == "plan":
|
||||
return {"id": eid, "role": "status", "source": "trace",
|
||||
"kind": kind, "text": entry.get("text") or ""}
|
||||
if kind == "reasoning":
|
||||
return {"id": eid, "role": "status", "source": "trace",
|
||||
"kind": kind,
|
||||
"text": "思考：" + (entry.get("summary") or "")}
|
||||
if kind == "agent-start":
|
||||
return {"id": eid, "role": "status", "source": "trace",
|
||||
"kind": kind,
|
||||
"text": "子代理开始：" + (entry.get("name") or "")
|
||||
+ (("（" + entry["prompt"] + "）")
|
||||
if entry.get("prompt") else "")}
|
||||
if kind == "agent-done":
|
||||
return {"id": eid, "role": "status", "source": "trace",
|
||||
"kind": kind,
|
||||
"text": "子代理结束：" + (entry.get("name") or "")
|
||||
+ " " + (entry.get("status") or "")}
|
||||
return {"id": eid, "role": "status", "source": "trace",
|
||||
"kind": kind, "text": json.dumps(entry)[:300]}
|
||||
|
||||
|
||||
def _ensure_environments_table(conn):
|
||||
"""幂等建环境表：server init_db 与各环境接口都可调它，老库
|
||||
第一次跑到这里时自动补表，不丢数据。类型写法与 followups
|
||||
@@ -1420,6 +1511,7 @@ def init_db(db_path):
|
||||
migrate_tasks_table(conn)
|
||||
migrate_projects_table(conn)
|
||||
_ensure_followups_table(conn)
|
||||
_ensure_continue_messages_table(conn)
|
||||
_ensure_environments_table(conn)
|
||||
_ensure_quotas_table(conn)
|
||||
_backfill_default_quotas(conn)
|
||||
@@ -1454,6 +1546,7 @@ def init_db(db_path):
|
||||
migrate_tasks_table(conn)
|
||||
migrate_projects_table(conn)
|
||||
_ensure_followups_table(conn)
|
||||
_ensure_continue_messages_table(conn)
|
||||
_ensure_environments_table(conn)
|
||||
_ensure_quotas_table(conn)
|
||||
_backfill_default_quotas(conn)
|
||||
@@ -1983,6 +2076,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
self._create_followup(user_id, rest[: -len("/followups")])
|
||||
elif rest.endswith("/followups") and method == "GET":
|
||||
self._list_followups(user_id, rest[: -len("/followups")])
|
||||
elif rest.endswith("/chat-history") and method == "GET":
|
||||
self._task_chat_history(
|
||||
user_id, rest[: -len("/chat-history")])
|
||||
elif rest.endswith("/messages") and method == "POST":
|
||||
self._create_peer_message(user_id, rest[: -len("/messages")])
|
||||
elif rest.endswith("/messages") and method == "GET":
|
||||
@@ -4476,6 +4572,124 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
self._send_json(200, {
|
||||
"followups": [_followup_to_dict(r) for r in rows]})
|
||||
|
||||
def _task_chat_history(self, user_id, task_id):
|
||||
"""GET /tasks/{id}/chat-history：真正的交替历史，一次性
|
||||
按时间交错拼好返回，前端不再猜顺序。
|
||||
|
||||
来源（全部是后端已有记录，不编造）：
|
||||
- 首条 prompt：任务行 prompt + created_at（用户）
|
||||
- 继续消息：task_continue_messages（用户，部署后才入库）
|
||||
- 后续消息：task_followups（用户，含任务间互发）
|
||||
- 助手/工具/状态：链路 message/exec/file/tool/call/
|
||||
plan/reasoning/agent-* 条目（trace 的 ts 是毫秒时间戳，
|
||||
无 ts 的条目按链路顺序贴在上一条有 ts 的条目之后，
|
||||
不给它伪造时间戳——输出里 ts_ms 为 null）
|
||||
|
||||
排序：按 ts_ms 升序，同一毫秒用户消息在链路条目之前
|
||||
（用户消息是触发该轮的原因），其余按原顺序稳定排序。
|
||||
旧任务缺口：部署前发生的继续只在 audit 里留了
|
||||
task_continued（只有任务 id，没有正文），无法追回；
|
||||
legacy_continue_count = audit 次数 - 已入库条数，>0 时
|
||||
前端必须在聊天窗上标“此前继续消息未入库，历史不完整”，
|
||||
不能假装这就是全部对话。异机 worktree 时链路读不到，
|
||||
messages 里只有用户侧 + note，不谎报无过程。"""
|
||||
row = self._fetch_task(task_id)
|
||||
if row is None:
|
||||
self._send_json(404, {"error": "task not found"})
|
||||
return
|
||||
task = _task_to_dict(row)
|
||||
with _auth_store(self.server.db_path) as store:
|
||||
if not store.check_permission(
|
||||
user_id, task["project_id"], "read"):
|
||||
self._send_json(403, {"error": "forbidden"})
|
||||
return
|
||||
entries, trace_note = self._trace_snapshot(task)
|
||||
conn = self._db()
|
||||
try:
|
||||
_ensure_continue_messages_table(conn)
|
||||
_ensure_followups_table(conn)
|
||||
cont_rows = conn.execute(
|
||||
"SELECT * FROM task_continue_messages WHERE task_id = ?"
|
||||
" ORDER BY created_at, id",
|
||||
(task_id,),
|
||||
).fetchall()
|
||||
fu_rows = conn.execute(
|
||||
"SELECT * FROM task_followups WHERE task_id = ?"
|
||||
" ORDER BY created_at, id",
|
||||
(task_id,),
|
||||
).fetchall()
|
||||
legacy = None
|
||||
try:
|
||||
aud = conn.execute(
|
||||
"SELECT COUNT(*) FROM audit_log"
|
||||
" WHERE action = 'task_continued' AND detail = ?",
|
||||
(task_id,),
|
||||
).fetchone()
|
||||
legacy = max(0, int(aud[0]) - len(cont_rows))
|
||||
except Exception:
|
||||
legacy = None
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
merged = []
|
||||
|
||||
def _ms(seconds):
|
||||
try:
|
||||
return int(float(seconds) * 1000)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
if task.get("prompt"):
|
||||
merged.append((0, _ms(task.get("created_at")), {
|
||||
"id": "task-prompt", "role": "user", "source": "prompt",
|
||||
"label": "用户", "text": task["prompt"],
|
||||
"ts_ms": _ms(task.get("created_at"))}))
|
||||
for r in cont_rows:
|
||||
d = _continue_message_to_dict(r)
|
||||
merged.append((0, _ms(d["created_at"]), {
|
||||
"id": "continue-" + d["id"], "role": "user",
|
||||
"source": "continue", "label": "用户",
|
||||
"text": d["body"], "ts_ms": _ms(d["created_at"])}))
|
||||
for r in fu_rows:
|
||||
d = _followup_to_dict(r)
|
||||
merged.append((0, _ms(d["created_at"]), {
|
||||
"id": "followup-" + d["id"], "role": "user",
|
||||
"source": "followup", "label": "后续消息",
|
||||
"text": d["body"], "ts_ms": _ms(d["created_at"])}))
|
||||
last_trace_ms = _ms(task.get("created_at"))
|
||||
for e in entries:
|
||||
if not isinstance(e, dict):
|
||||
continue
|
||||
if str(e.get("id") or "").startswith("fu-"):
|
||||
# 直接发送的后续消息已在链路里留了一条回声
|
||||
# （_append_direct_trace 写 kind=message），
|
||||
# followups 表里同一条已作用户消息返回；这里
|
||||
# 跳过回声，否则同一条消息会先以用户、再以
|
||||
# 助手各出现一次。
|
||||
continue
|
||||
msg = _chat_message_from_trace(e)
|
||||
ts = e.get("ts")
|
||||
if isinstance(ts, (int, float)):
|
||||
last_trace_ms = int(ts)
|
||||
sort_ms = int(ts)
|
||||
else:
|
||||
sort_ms = last_trace_ms
|
||||
msg["ts_ms"] = int(ts) if isinstance(ts, (int, float)) else None
|
||||
merged.append((1, sort_ms, msg))
|
||||
# 排序键：(时间, 用户优先, 原顺序)。merged 里用户项的
|
||||
# 第一元组是 0、链路项是 1；Python 排序稳定，同一
|
||||
# (时间, 优先级) 内保持插入顺序。
|
||||
merged.sort(key=lambda item: (
|
||||
item[1] if item[1] is not None else 0, item[0]))
|
||||
self._send_json(200, {
|
||||
"task_id": task_id,
|
||||
"status": task["status"],
|
||||
"messages": [m for _p, _t, m in merged],
|
||||
"continue_messages_stored": len(cont_rows),
|
||||
"legacy_continue_count": legacy,
|
||||
"trace_note": trace_note,
|
||||
})
|
||||
|
||||
def _create_peer_message(self, user_id, task_id):
|
||||
"""任务间互发消息（API 发送方）：以 from_task_id 任务的名义
|
||||
给目标任务发消息。投递语义与 worker 路径共用 deliver 层，
|
||||
@@ -6623,13 +6837,30 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
% ", ".join(sets),
|
||||
tuple(vals),
|
||||
)
|
||||
conn.commit()
|
||||
if cur.rowcount != 1:
|
||||
# 状态已变化：更新 0 行，同一事务里什么都不留——
|
||||
# 尤其不能先把继续消息落盘再报 409，那会凭空多
|
||||
# 一条从未被执行的“用户消息”。
|
||||
conn.rollback()
|
||||
self._send_json(409, {
|
||||
"error": "任务状态已变化，请刷新后重试",
|
||||
"status": task["status"],
|
||||
})
|
||||
return
|
||||
# 继续消息与状态改回 queued 同一事务：要么一起生效，
|
||||
# 要么一起不生效，绝不出现“任务已重排队、消息却丢了”
|
||||
# 的半截状态。created_at 用本请求的 updated_at 时间戳
|
||||
# （vals[1]），与任务行的时间源一致，便于按时间交错
|
||||
# 拼聊天历史。
|
||||
_ensure_continue_messages_table(conn)
|
||||
conn.execute(
|
||||
"INSERT INTO task_continue_messages(id, task_id,"
|
||||
" project_id, body, created_by, created_at)"
|
||||
" VALUES(?, ?, ?, ?, ?, ?)",
|
||||
(uuid.uuid4().hex, task_id, task["project_id"],
|
||||
message, user_id, vals[1]),
|
||||
)
|
||||
conn.commit()
|
||||
new_row = conn.execute(
|
||||
"SELECT * FROM tasks WHERE id = ?", (task_id,)
|
||||
).fetchone()
|
||||
|
||||
Reference in new issue
Block a user