diff --git a/CHANGELOG.md b/CHANGELOG.md index 9df232e..abff507 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ ## Unreleased +- 对话步骤进度改为按每轮任务保存完整计划；长任务、刷新或网络重连后可恢复当前步骤，不再因历史分页或首个实时事件错过而出现进度消失、串到上一轮或无法完成。 + - 改进 Origin 多面板图排版：由 Origin 统一排列图层，共享横轴时仅在底行显示横轴标题和刻度标签，图例可自动避让数据；同时消除中文标题和坐标轴文字在导出图片中的异常横线。 - 修复 Origin 对数坐标图可能从 `1E-10` 开始、导致有效数据挤在图形右侧的问题；多面板未单独填写纵轴名称时，也会优先使用面板标题或数据列名，不再显示笼统的 `Y`。 diff --git a/DESIGN.md b/DESIGN.md index 42a8f89..21ccaf6 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -199,7 +199,7 @@ Admin GET /v1/admin/*(require_admin;overview + usage/models|users + storage/ Export GET /v1/tasks/{id}/export(docx) ``` -**SSE 事件**:`run_start / llm_start / text{delta} / reasoning{delta}(thinking 模型推理流,前端灰色折叠卡)/ tool_call / tool_result(预览,完整走 DB)/ llm_end / model_switch / warn{msg}(熔断·重复拦截·折叠失败等运行时提醒)/ context_fold{phase,...}(§8.8 Phase 2 折叠 start/done)/ cancelled / error / done`。fan-out:每订阅独立 queue;迟到订阅立收 done。事件不持久化(messages 走 PG)。 +**SSE 事件**:`run_start / llm_start / text{delta} / reasoning{delta}(thinking 模型推理流,前端灰色折叠卡)/ progress_snapshot{run_id,steps}(当前 user message 即 run 边界,从 messages 投影恢复)/ tool_call / tool_result(预览,完整走 DB)/ llm_end / model_switch / warn{msg}(熔断·重复拦截·折叠失败等运行时提醒)/ context_fold{phase,...}(§8.8 Phase 2 折叠 start/done)/ cancelled / error / done`。`task_progress` 每次提交完整步骤快照,前端整体替换；旧 `set_plan/update_step` 仅在历史投影时兼容。fan-out:每订阅独立 queue;迟到订阅先从 PG 恢复当前 run 最新进度,终态迟到订阅立收 done。普通直播事件不持久化(messages 走 PG)。 **版本化**:`/v1` minor 半年兼容,major 6 个月 deprecation。**CORS**:本地 `*`,部署收紧。 ### 7.3 认证 @@ -234,7 +234,7 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) - working_dir 存相对 ROOT posix 串,读写统一过 `core/paths.py`;入口 `validate_task_name` 拒空/`/\NUL`/`.` 起头。 - `auto_title_pending`(0023)只是一轮 UI 命名闸，不是 task 状态机；旧创建入口/存量行恒 false，快速入口首发后消费，人工改名优先清闸。 -- **0004 简化**:runs 表只写不读、run_id 单活 run 下全冗余 → 合并 `run_status/run_error` 入 tasks。**0006**:`tasks.model_profile` 为 source-of-truth(PATCH 切、下条 send 生效);usage_events 重建 v2 多态形态,统计 source-of-truth;tasks 三列保留作粗概览。run_status 终态:ok 收回 idle,error(出错)与 cancelled(用户停止)是持久终态 —— 前端 `renderPersistedRunTerminal` 据此在每次重渲后补持久卡(扛过收尾 loadMessages 整屏重建),刷新/切任务仍在;下次起新 run(post_message 写 running)覆盖清掉。 +- **0004 简化**:runs 表只写不读、独立 run 实体在单活形态下冗余 → 合并 `run_status/run_error` 入 tasks；需要前端关联本轮时直接复用该轮 user message UUID 为 `run_id`，不恢复 runs 表。**0006**:`tasks.model_profile` 为 source-of-truth(PATCH 切、下条 send 生效);usage_events 重建 v2 多态形态,统计 source-of-truth;tasks 三列保留作粗概览。run_status 终态:ok 收回 idle,error(出错)与 cancelled(用户停止)是持久终态 —— 前端 `renderPersistedRunTerminal` 据此在每次重渲后补持久卡(扛过收尾 loadMessages 整屏重建),刷新/切任务仍在;下次起新 run(post_message 写 running)覆盖清掉。 - **0029 消息序号**:`tasks.next_message_idx` 在 task 行锁下统一分配 `messages.idx`，Web、agent 与渠道追加不再各自维护序号或依赖冲突重试；分配时仍与 `max(idx)` 校准，允许蓝绿发布窗口内旧实例继续写入。清空消息与计数器在同一事务归零。 - **No-subtask**:同 user 下前缀互含即拒(归一 posix 后 Python 端比对);同 working_dir 允许。 - **文件面板先备料**:user_root 与普通目录都可显式新建直接子目录；创建成功后前端进入该目录，用户可先上传/选入资料，再把顶层空目录选作新对话 working_dir。目录 leaf 复用 `validate_task_name`，不允许借 UI 创建点目录或路径式名称。 diff --git a/RUN.md b/RUN.md index 4456ddc..0139228 100644 --- a/RUN.md +++ b/RUN.md @@ -363,7 +363,7 @@ $env:ZCBOT_EVAL_TOKEN = "" | `GET/POST /v1/admin/external-system-definitions` | 管理员列出或新增可信外部系统目录 | admin | | `PUT/DELETE /v1/admin/external-system-definitions/{id}` | 管理员编辑、停用或删除目录项；已有用户连接时拒绝删除 | admin | | `GET /v1/tasks/{id}/messages` | LiteLLM payload 透传；另带 `artifact_refs`(助手产物)与 `attachment_refs`(用户附件)。两者均以 `null` 表示旧消息、`[]` 表示新消息明确为空、非空数组表示 task-relative 结构化引用 | 必填 | -| `POST /v1/tasks/{id}/messages` | `{content, attachments?:[{path,kind,label?}], image_model?=""}` 发消息；`attachments` 路径由后端按当前 task working_dir 校验并规范化，允许纯附件消息；旧客户端省略该字段继续兼容。返 `{events_url}`；**`run_status` 是 running/cancelling → 409**；UI 应 disable send 直到 SSE `done` | 必填 | +| `POST /v1/tasks/{id}/messages` | `{content, attachments?:[{path,kind,label?}], image_model?=""}` 发消息；`attachments` 路径由后端按当前 task working_dir 校验并规范化，允许纯附件消息；旧客户端省略该字段继续兼容。返 `{events_url,run_id}`，其中 `run_id` 是本轮 user message UUID；**`run_status` 是 running/cancelling → 409**；UI 应 disable send 直到 SSE `done` | 必填 | | `GET /v1/tasks/{id}/events` | SSE 流(`event: ` + `data: `);订阅 task 当前活动 | 必填 | | `POST /v1/tasks/{id}/cancel` | 协作式 cancel;`run_status != running` → 409;LLM 走 streaming,chunk 间 poll cancel — 延迟 100ms 级,基本秒退 | 必填 | | `GET /v1/procs` | 当前用户全部后台进程(bg proc,§8.12;shell/run_python `background=true` 启动);纯文件系统读取,前端运行条 5s 轮询用 | 必填 | @@ -384,7 +384,7 @@ $env:ZCBOT_EVAL_TOKEN = "" | `GET /v1/models` | 列 chat LLM 模型清单(扫 `config/models/*.yaml`),前端顶栏切换 / 新建对话框下拉用 | 必填 | | `GET /v1/image_models` | 列图像生成 variant 清单(扫 `config/media/doubao.yaml` image 段),前端"生图"下拉用;yaml 无 image variant → 空列表 → UI 隐藏下拉 | 必填 | -**SSE 事件**(每帧 `event: ` + `data: `):`run_start{}` → `llm_start{}` → `text{delta}` / `tool_call{name,args,args_preview}` / `tool_result{name,preview,truncated}` → `llm_end{prompt_tokens,completion_tokens}` → `done{}`;cancel 走 `cancelled{}` 后随 `done{}` 收流;异常走 `error{msg}`。30s 无 event 服务端发 `: ping` 心跳。nginx 反代记得关 buffering(响应头已带 `X-Accel-Buffering: no` 默认起效)。 +**SSE 事件**(每帧 `event: ` + `data: `):建连时若当前 run 已发布计划,先补 `progress_snapshot{run_id,steps}` → `run_start{}` → `llm_start{}` → `text{delta}` / `tool_call{name,args,args_preview}` / `tool_result{name,preview,truncated}` → `llm_end{prompt_tokens,completion_tokens}` → `done{}`;cancel 走 `cancelled{}` 后随 `done{}` 收流;异常走 `error{msg}`。`task_progress` 新协议每次携带完整 `steps`,客户端整体替换；消息分页响应也附加 `progress_snapshot`,刷新不依赖当前 30 条窗口。30s 无 event 服务端发 `: ping` 心跳。nginx 反代记得关 buffering(响应头已带 `X-Accel-Buffering: no` 默认起效)。 **SSE 客户端注意**:浏览器原生 `EventSource` 不支持自定义 header,无法塞 Bearer token。要么 `fetch + ReadableStream` 自解 SSE 帧(dev.html 走的就是这条),要么后端日后加 `?token=...` query(目前不支持,避免 token 进 access log)。 diff --git a/tests/frontend_task_progress.test.mjs b/tests/frontend_task_progress.test.mjs index a4ffd24..3687341 100644 --- a/tests/frontend_task_progress.test.mjs +++ b/tests/frontend_task_progress.test.mjs @@ -51,6 +51,23 @@ test("tool calls can apply progress updates on top of previous task progress", ( ]); }); +test("a full snapshot replaces the prior run plan instead of merging it", () => { + const previous = [ + { id: "old", title: "上一轮计划", status: "completed" }, + ]; + const updated = applyProgressAction(previous, { + steps: [ + { id: "s1", title: "分析需求", status: "completed" }, + { id: "s2", title: "执行修改", status: "in_progress" }, + ], + }); + + assert.deepEqual(updated, [ + { id: "s1", title: "分析需求", status: "completed" }, + { id: "s2", title: "执行修改", status: "in_progress" }, + ]); +}); + test("a completed step force-completes earlier dangling steps (monotonic heal)", () => { const steps = [ { id: "s1", title: "摄取素材", status: "in_progress" }, diff --git a/tests/test_run_lifecycle.py b/tests/test_run_lifecycle.py index 5dae9a4..c1f59c1 100644 --- a/tests/test_run_lifecycle.py +++ b/tests/test_run_lifecycle.py @@ -43,6 +43,7 @@ class ClaimRunTests(unittest.TestCase): self.assertEqual(claim.metadata, {"marker": "ok"}) added = session.add.call_args.args[0] + self.assertEqual(claim.run_id, added.message_id) self.assertEqual(added.task_id, tid) self.assertEqual(added.idx, 7) self.assertEqual( diff --git a/tests/test_task_progress_projection.py b/tests/test_task_progress_projection.py new file mode 100644 index 0000000..ff0457f --- /dev/null +++ b/tests/test_task_progress_projection.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import json +import unittest + +from web.task_progress import project_progress_payloads + + +def _call(args: dict) -> dict: + return { + "role": "assistant", + "tool_calls": [{ + "function": { + "name": "task_progress", + "arguments": json.dumps(args, ensure_ascii=False), + }, + }], + } + + +class TaskProgressProjectionTests(unittest.TestCase): + def test_latest_full_snapshot_replaces_prior_snapshot(self) -> None: + steps, seen = project_progress_payloads([ + _call({"steps": [ + {"id": "old", "title": "旧计划", "status": "in_progress"}, + ]}), + _call({"steps": [ + {"id": "s1", "title": "分析", "status": "completed"}, + {"id": "s2", "title": "实现", "status": "in_progress"}, + ]}), + ]) + + self.assertTrue(seen) + self.assertEqual([step["id"] for step in steps], ["s1", "s2"]) + + def test_legacy_updates_remain_replayable(self) -> None: + steps, seen = project_progress_payloads([ + _call({"action": "set_plan", "steps": [ + {"id": "s1", "title": "分析", "status": "in_progress"}, + {"id": "s2", "title": "实现", "status": "pending"}, + ]}), + _call({"action": "update_step", "step": { + "id": "s2", "status": "completed", + }}), + ]) + + self.assertTrue(seen) + self.assertEqual([step["status"] for step in steps], ["completed", "completed"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_task_progress_tool.py b/tests/test_task_progress_tool.py index fa1fc09..95c2c2b 100644 --- a/tests/test_task_progress_tool.py +++ b/tests/test_task_progress_tool.py @@ -5,17 +5,18 @@ from tools.task_progress import TaskProgressTool class TaskProgressToolTests(unittest.TestCase): - def test_schema_exposes_set_update_and_clear_actions(self) -> None: + def test_schema_requires_complete_steps_snapshot(self) -> None: schema = TaskProgressTool().schema fn = schema["function"] self.assertEqual(fn["name"], "task_progress") - action_enum = fn["parameters"]["properties"]["action"]["enum"] - self.assertEqual(action_enum, ["set_plan", "update_step", "clear"]) + params = fn["parameters"] + self.assertEqual(params["required"], ["steps"]) + self.assertNotIn("action", params["properties"]) + self.assertNotIn("step", params["properties"]) def test_execute_returns_short_ui_only_result(self) -> None: out = TaskProgressTool().execute( - action="set_plan", steps=[ {"id": "s1", "title": "理解需求", "status": "completed"}, {"id": "s2", "title": "实现功能", "status": "in_progress"}, @@ -23,19 +24,9 @@ class TaskProgressToolTests(unittest.TestCase): ) data = json.loads(out) - self.assertEqual(data, {"ok": True, "action": "set_plan", "step_count": 2}) + self.assertEqual(data, {"ok": True, "step_count": 2}) self.assertLess(len(out), 80) - def test_execute_normalizes_update_step_without_echoing_title(self) -> None: - out = TaskProgressTool().execute( - action="update_step", - step={"id": "s2", "title": "实现功能", "status": "completed"}, - ) - - data = json.loads(out) - self.assertEqual(data, {"ok": True, "action": "update_step", "step_id": "s2"}) - self.assertNotIn("实现功能", out) - if __name__ == "__main__": unittest.main() diff --git a/tools/task_progress.py b/tools/task_progress.py index 0c90c75..e1350dc 100644 --- a/tools/task_progress.py +++ b/tools/task_progress.py @@ -7,7 +7,7 @@ arguments for Web rendering and is compacted out of older LLM context. from __future__ import annotations import json -from typing import Any +from typing import Any, ClassVar from .base import Tool @@ -15,24 +15,23 @@ from .base import Tool class TaskProgressTool(Tool): name = "task_progress" description = ( - "Publish or update a concise user-visible progress checklist for the current task. " - "Use only for meaningful multi-step work: set the plan once, update a step when it " - "starts or completes, and when all work is done mark the final step completed (do NOT " - "clear). Use clear only when the plan is no longer relevant. This is a UI progress " - "signal, not a work product." + "Publish the complete current user-visible progress checklist for this run. " + "Use only for meaningful multi-step work. Every call must include the full checklist, " + "including unchanged steps; never send a partial step patch. Keep stable step ids while " + "revising the plan, allow at most one in_progress step, and mark every step completed " + "before a successful final answer. This is a UI progress signal, not a work product." ) - parameters = { + parameters: ClassVar[dict[str, Any]] = { "type": "object", "additionalProperties": False, "properties": { - "action": { + "explanation": { "type": "string", - "enum": ["set_plan", "update_step", "clear"], - "description": "set_plan replaces the checklist; update_step changes one step; clear removes it.", + "description": "Optional short reason when the plan materially changes.", }, "steps": { "type": "array", - "description": "Required for set_plan. Keep to 3-7 user-meaningful steps.", + "description": "The complete current checklist. Keep to 3-7 user-meaningful steps.", "items": { "type": "object", "additionalProperties": False, @@ -47,32 +46,14 @@ class TaskProgressTool(Tool): "required": ["id", "title", "status"], }, }, - "step": { - "type": "object", - "description": "Required for update_step.", - "additionalProperties": False, - "properties": { - "id": {"type": "string"}, - "title": {"type": "string"}, - "status": { - "type": "string", - "enum": ["pending", "in_progress", "completed"], - }, - }, - "required": ["id", "status"], - }, }, - "required": ["action"], + "required": ["steps"], } def execute(self, **kwargs: Any) -> str: - action = str(kwargs.get("action") or "") - out: dict[str, Any] = {"ok": True, "action": action} - if action == "set_plan": - steps = kwargs.get("steps") - out["step_count"] = len(steps) if isinstance(steps, list) else 0 - elif action == "update_step": - step = kwargs.get("step") - if isinstance(step, dict) and step.get("id"): - out["step_id"] = str(step["id"]) + steps = kwargs.get("steps") + out: dict[str, Any] = { + "ok": True, + "step_count": len(steps) if isinstance(steps, list) else 0, + } return json.dumps(out, ensure_ascii=False, separators=(",", ":")) diff --git a/web/routers/messages.py b/web/routers/messages.py index 13c4225..0def393 100644 --- a/web/routers/messages.py +++ b/web/routers/messages.py @@ -42,6 +42,7 @@ from ..run_lifecycle import ( schedule_claimed_run, ) from ..schemas import MessageRequest, OptimizePromptRequest +from ..task_progress import latest_run_progress from ..userfiles import load_user_root @@ -137,9 +138,11 @@ def register_message_routes(app, *, require_user) -> None: .where(Message.task_id == tid, Message.idx > last_idx) .limit(1) ).first() is not None + progress_snapshot = latest_run_progress(s, tid) return { "has_more": has_more, "has_more_after": has_more_after, + "progress_snapshot": progress_snapshot, "messages": [ { "idx": r.idx, @@ -320,7 +323,10 @@ def register_message_routes(app, *, require_user) -> None: )) app.state.aux_tasks.add(title_task) title_task.add_done_callback(app.state.aux_tasks.discard) - return {"events_url": f"/v1/tasks/{tid}/events"} + return { + "events_url": f"/v1/tasks/{tid}/events", + "run_id": str(claim.run_id), + } @app.post("/v1/tasks/{task_id}/cancel", status_code=202, tags=["tasks"]) def cancel_task( @@ -638,6 +644,12 @@ def register_message_routes(app, *, require_user) -> None: return q = broker.subscribe(tid) try: + # Durable catch-up for events emitted before this subscriber existed. + # Subscribing first keeps updates emitted during the DB read queued. + with session_scope() as s: + snapshot = latest_run_progress(s, tid) + if snapshot is not None: + yield sse_event("progress_snapshot", snapshot) while True: try: ev = await asyncio.wait_for(q.get(), timeout=30.0) diff --git a/web/run_lifecycle.py b/web/run_lifecycle.py index 3f4c72d..2a60740 100644 --- a/web/run_lifecycle.py +++ b/web/run_lifecycle.py @@ -8,7 +8,7 @@ from __future__ import annotations import asyncio from dataclasses import dataclass, field from typing import Any, Callable, Optional -from uuid import UUID +from uuid import UUID, uuid4 from sqlalchemy import select, update @@ -44,6 +44,7 @@ PrepareClaim = Callable[[Any, Task], tuple[dict[str, Any], dict[str, Any]]] class RunClaim: """抢占成功后返回给调用方的领域元数据。""" + run_id: UUID metadata: dict[str, Any] = field(default_factory=dict) @@ -79,7 +80,9 @@ def claim_run_with_message( # task 行锁串行化同一 task 的 idx 分配，无需额外序列表或 advisory lock。 next_idx = allocate_message_idx(s, task_id, locked_task=task) + run_id = uuid4() s.add(Message( + message_id=run_id, task_id=task_id, idx=int(next_idx), payload={"role": "user", "content": user_message}, @@ -92,7 +95,7 @@ def claim_run_with_message( **extra_values, } s.execute(update(Task).where(Task.task_id == task_id).values(**values)) - return RunClaim(metadata=dict(metadata)) + return RunClaim(run_id=run_id, metadata=dict(metadata)) def schedule_claimed_run( diff --git a/web/static/js/chat.js b/web/static/js/chat.js index 2a66fa7..77619c1 100644 --- a/web/static/js/chat.js +++ b/web/static/js/chat.js @@ -1023,6 +1023,7 @@ function alignedEarlierLimit(firstIdx) { async function loadMessages({ render = true } = {}) { const data = await api("GET", `/v1/tasks/${state.taskId}/messages?limit=${MSG_PAGE}`); state.loadedMessages = data.messages || []; + state.taskProgressSnapshot = data.progress_snapshot || null; state.msgHasMore = !!data.has_more; state.msgHasMoreNewer = !!data.has_more_after; // 尾部窗口通常为 false state.msgLoadingEarlier = false; @@ -1454,7 +1455,7 @@ function ensureRunningTaskSubscribed(taskId, url, seed = {}) { curSeg: null, cancelling: seed.run_status === "cancelling", workingDir: seed.working_dir || "", - progressSteps: cloneProgressSteps(state.taskProgressByTask.get(taskId)), + progressSteps: [], }; state.liveRuns.set(taskId, run); state.streaming = true; @@ -1798,7 +1799,9 @@ function renderMessages(msgs, { stickBottom = true } = {}) { } if (stickBottom) wrap.scrollTop = wrap.scrollHeight; - setTaskProgress(state.taskId, currentProgressSteps); + const snapshotSteps = state.taskProgressSnapshot && Array.isArray(state.taskProgressSnapshot.steps) + ? state.taskProgressSnapshot.steps : null; + setTaskProgress(state.taskId, snapshotSteps === null ? currentProgressSteps : snapshotSteps); upgradeMediaArtifacts(wrap); renderPersistedRunTerminal(); // 上次 run error/cancelled 终态 → 末尾补持久卡(所有重渲路径统一走这) renderLiveRunIfVisible(); @@ -2741,8 +2744,10 @@ async function sendMessage(overrideText) { cancelling: false, workingDir: state.taskMeta && state.taskMeta.working_dir, autoTitleEligible: !attachmentOnly, - progressSteps: cloneProgressSteps(state.taskProgressByTask.get(taskId)), + runId: r.run_id || "", + progressSteps: [], }; + setTaskProgress(taskId, []); // 预建的空占位 .body 即首个文字段(首字到达前显示「思考中」) run.curSeg = { el: asstCard.querySelector(".body"), acc: "", pending: false }; setRunPhase(run, "llm"); // POST 返回即起跳:覆盖 build_agent + 首轮 TTFT 的空窗 @@ -3035,6 +3040,12 @@ function handleSseEvent(ev, asstCard, ctx) { if (nearBottom) stream.scrollTop = stream.scrollHeight; }); } + } else if (t === "progress_snapshot") { + const snapshot = ev.data || {}; + if (ctx.runId && snapshot.run_id && ctx.runId !== snapshot.run_id) return; + if (!ctx.runId && snapshot.run_id) ctx.runId = snapshot.run_id; + ctx.progressSteps = cloneProgressSteps(snapshot.steps); + setTaskProgress(ctx.taskId, ctx.progressSteps); } else if (t === "tool_call") { const fn = (ev.data && ev.data.name) || "?"; const args = (ev.data && ev.data.args) || ""; diff --git a/web/static/js/progress.js b/web/static/js/progress.js index 6831f17..5a7523e 100644 --- a/web/static/js/progress.js +++ b/web/static/js/progress.js @@ -39,7 +39,9 @@ export function applyProgressAction(progress, args) { if (!args || typeof args !== "object") return current; const action = args.action || ""; if (action === "clear") return []; - if (action === "set_plan") { + // Current protocol sends a complete snapshot on every call. Legacy set_plan + // has the same shape, so both formats intentionally converge here. + if (Array.isArray(args.steps)) { const planned = Array.isArray(args.steps) ? args.steps.map(normalizeProgressStep).filter(Boolean) : []; return enforceMonotonicProgress(planned); } diff --git a/web/static/js/state.js b/web/static/js/state.js index 5f0231a..d7ff9b5 100644 --- a/web/static/js/state.js +++ b/web/static/js/state.js @@ -55,6 +55,7 @@ export const state = { streaming: false, // 兼容旧判断:任一 task 是否在流式中 liveRuns: new Map(), // task_id -> 当前浏览器会话内运行中的回复卡/累计文本 taskProgressByTask: new Map(), // task_id -> 历史消息重放后的当前进度步骤 + taskProgressSnapshot: null, // 当前 task 最近一轮的分页外完整进度快照 // 消息分页(尾部窗口 + 向上滚动加载更早):切 task 默认只拉最近一批, // 顶部 sentinel 进视口自动往前补。loadedMessages 是当前已加载的升序窗口, // renderMessages 对它做纯函数渲染(时序累积逻辑无需改)。 diff --git a/web/task_progress.py b/web/task_progress.py new file mode 100644 index 0000000..181f830 --- /dev/null +++ b/web/task_progress.py @@ -0,0 +1,126 @@ +"""Project the latest run-scoped progress snapshot from append-only messages.""" +from __future__ import annotations + +import json +from collections.abc import Iterable +from typing import Any +from uuid import UUID + +from sqlalchemy import select + +from core.storage.models import Message + +_VALID_STATUSES = {"pending", "in_progress", "completed"} + + +def _normalize_step(value: Any) -> dict[str, str] | None: + if not isinstance(value, dict): + return None + step_id = str(value.get("id") or "").strip() + title = str(value.get("title") or "").strip() + status = str(value.get("status") or "pending") + if not step_id or not title: + return None + if status not in _VALID_STATUSES: + status = "pending" + return {"id": step_id, "title": title, "status": status} + + +def _heal_monotonic(steps: list[dict[str, str]]) -> list[dict[str, str]]: + last_completed = max( + (i for i, step in enumerate(steps) if step["status"] == "completed"), + default=-1, + ) + return [ + {**step, "status": "completed"} if i < last_completed else dict(step) + for i, step in enumerate(steps) + ] + + +def apply_progress_args( + current: list[dict[str, str]], args: Any, +) -> list[dict[str, str]]: + """Apply current full snapshots plus legacy set/update/clear calls.""" + if not isinstance(args, dict): + return [dict(step) for step in current] + action = args.get("action") or "" + if action == "clear": + return [] + if isinstance(args.get("steps"), list): + normalized: list[dict[str, str]] = [] + for raw in args["steps"]: + step = _normalize_step(raw) + if step is not None: + normalized.append(step) + return _heal_monotonic(normalized) + if action != "update_step" or not isinstance(args.get("step"), dict): + return [dict(step) for step in current] + + raw = args["step"] + step_id = str(raw.get("id") or "").strip() + if not step_id: + return [dict(step) for step in current] + next_steps: list[dict[str, str]] = [] + found = False + for step in current: + if step["id"] != step_id: + next_steps.append(dict(step)) + continue + found = True + status = str(raw.get("status") or step["status"]) + next_steps.append({ + "id": step_id, + "title": str(raw.get("title") or step["title"]).strip(), + "status": status if status in _VALID_STATUSES else "pending", + }) + if not found: + normalized = _normalize_step(raw) + if normalized is not None: + next_steps.append(normalized) + return _heal_monotonic(next_steps) + + +def project_progress_payloads( + payloads: Iterable[dict[str, Any]], +) -> tuple[list[dict[str, str]], bool]: + steps: list[dict[str, str]] = [] + seen = False + for payload in payloads: + if not isinstance(payload, dict) or payload.get("role") != "assistant": + continue + for call in payload.get("tool_calls") or []: + function = call.get("function") if isinstance(call, dict) else None + if not isinstance(function, dict) or function.get("name") != "task_progress": + continue + raw_args = function.get("arguments") or "{}" + try: + args = json.loads(raw_args) if isinstance(raw_args, str) else raw_args + except (TypeError, json.JSONDecodeError): + args = {} + steps = apply_progress_args(steps, args) + seen = True + return steps, seen + + +def latest_run_progress(session, task_id: UUID) -> dict[str, Any] | None: + """Return the latest user message id and that run's projected plan.""" + run = session.execute( + select(Message.message_id, Message.idx) + .where( + Message.task_id == task_id, + Message.payload["role"].astext == "user", + ) + .order_by(Message.idx.desc()) + .limit(1) + ).first() + if run is None: + return None + payloads = session.execute( + select(Message.payload) + .where(Message.task_id == task_id, Message.idx > run.idx) + .order_by(Message.idx) + ).scalars().all() + steps, seen = project_progress_payloads(payloads) + if not seen: + return None + return {"run_id": str(run.message_id), "steps": steps}