fix(web): make task progress run-scoped and recoverable

CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ## Unreleased
 
+- 对话步骤进度改为按每轮任务保存完整计划；长任务、刷新或网络重连后可恢复当前步骤，不再因历史分页或首个实时事件错过而出现进度消失、串到上一轮或无法完成。
+
 - 改进 Origin 多面板图排版：由 Origin 统一排列图层，共享横轴时仅在底行显示横轴标题和刻度标签，图例可自动避让数据；同时消除中文标题和坐标轴文字在导出图片中的异常横线。
 
 - 修复 Origin 对数坐标图可能从 `1E-10` 开始、导致有效数据挤在图形右侧的问题；多面板未单独填写纵轴名称时，也会优先使用面板标题或数据列名，不再显示笼统的 `Y`。

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 创建点目录或路径式名称。

RUN.md

@@ -363,7 +363,7 @@ $env:ZCBOT_EVAL_TOKEN = "<dedicated-eval-user-jwt>"
 | `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: <type>` + `data: <json>`);订阅 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 = "<dedicated-eval-user-jwt>"
 | `GET /v1/models` | 列 chat LLM 模型清单(扫 `config/models/*.yaml`),前端顶栏切换 / 新建对话框下拉用 | 必填 |
 | `GET /v1/image_models` | 列图像生成 variant 清单(扫 `config/media/doubao.yaml` image 段),前端"生图"下拉用;yaml 无 image variant → 空列表 → UI 隐藏下拉 | 必填 |
 
-**SSE 事件**(每帧 `event: <type>` + `data: <JSON>`):`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: <type>` + `data: <JSON>`):建连时若当前 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)。
 

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

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(

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

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"]
+        params = fn["parameters"]
-        self.assertEqual(action_enum, ["set_plan", "update_step", "clear"])
+        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()

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. "
+        "Publish the complete current user-visible progress checklist for this run. "
-        "Use only for meaningful multi-step work: set the plan once, update a step when it "
+        "Use only for meaningful multi-step work. Every call must include the full checklist, "
-        "starts or completes, and when all work is done mark the final step completed (do NOT "
+        "including unchanged steps; never send a partial step patch. Keep stable step ids while "
-        "clear). Use clear only when the plan is no longer relevant. This is a UI progress "
+        "revising the plan, allow at most one in_progress step, and mark every step completed "
-        "signal, not a work product."
+        "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": "Optional short reason when the plan materially changes.",
-                "description": "set_plan replaces the checklist; update_step changes one step; clear removes it.",
             },
             "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": ["steps"],
-                "required": ["id", "status"],
-            },
-        },
-        "required": ["action"],
     }
 
     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
+        out: dict[str, Any] = {
-        elif action == "update_step":
+            "ok": True,
-            step = kwargs.get("step")
+            "step_count": len(steps) if isinstance(steps, list) else 0,
-            if isinstance(step, dict) and step.get("id"):
+        }
-                out["step_id"] = str(step["id"])
         return json.dumps(out, ensure_ascii=False, separators=(",", ":"))

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)

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(

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) || "";

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

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 对它做纯函数渲染(时序累积逻辑无需改)。

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}