代码提交

offline-backend-20260512/backend/app/gateway/routers/threads.py

@@ -1064,6 +1064,19 @@ class AppendMessageRequest(BaseModel):
     type: str = Field(default="human", description="Message role: human/ai/system")
 
 
+class EditMessageRequest(BaseModel):
+    """Request body for editing one persisted chat message."""
+
+    content: str = Field(..., description="Replacement message text")
+
+    @field_validator("content")
+    @classmethod
+    def _content_not_blank(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("Message content cannot be empty")
+        return value
+
+
 def _flatten_msg_content(content: Any) -> str:
     """把 message.content(str / 结构化 block 列表)拍平成纯文本。"""
     if isinstance(content, str):
@@ -1079,6 +1092,17 @@ def _flatten_msg_content(content: Any) -> str:
     return str(content) if content else ""
 
 
+def _set_message_content(message: Any, content: str) -> None:
+    """Replace a message's visible content while preserving id/metadata/files."""
+    if isinstance(message, dict):
+        message["content"] = content
+        return
+    try:
+        setattr(message, "content", content)
+    except Exception as exc:
+        raise TypeError("Message content cannot be edited") from exc
+
+
 def _extract_last_ai_text(messages: list[Any]) -> str:
     """返回最后一条 AI 消息的纯文本;没有则空串。兼容 message 对象与序列化 dict 两种存储形态。"""
     for msg in reversed(messages or []):
@@ -1176,6 +1200,64 @@ async def get_thread_last_ai_message(thread_id: str, request: Request, max_chars
     return {"delivered": full_len > 0, "content": text, "length": full_len}
 
 
+@router.delete("/{thread_id}/messages")
+@require_permission("threads", "write", owner_check=True, require_existing=True)
+async def clear_thread_messages(thread_id: str, request: Request) -> dict[str, Any]:
+    """Clear all messages from a thread's latest checkpoint."""
+    checkpointer = get_checkpointer(request)
+    config: dict[str, Any] = {
+        "configurable": {
+            "thread_id": thread_id,
+            "checkpoint_ns": "",
+        }
+    }
+    try:
+        checkpoint_tuple = await checkpointer.aget_tuple(config)
+    except Exception:
+        logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id))
+        raise HTTPException(status_code=500, detail="Failed to get thread state")
+
+    if checkpoint_tuple is None:
+        raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
+
+    checkpoint: dict[str, Any] = dict(getattr(checkpoint_tuple, "checkpoint", {}) or {})
+    metadata: dict[str, Any] = dict(getattr(checkpoint_tuple, "metadata", {}) or {})
+    channel_values: dict[str, Any] = dict(checkpoint.get("channel_values", {}))
+
+    messages = channel_values.get("messages")
+    if not isinstance(messages, list):
+        messages = []
+
+    deleted_ids = [mid for m in messages if (mid := _message_id(m))]
+    channel_values["messages"] = []
+    checkpoint["channel_values"] = channel_values
+    checkpoint["id"] = str(uuid6(clock_seq=-2))
+    checkpoint["ts"] = now_iso()
+    versions = dict(checkpoint.get("channel_versions", {}))
+    new_ver = checkpointer.get_next_version(versions.get("messages"), None)
+    versions["messages"] = new_ver
+    checkpoint["channel_versions"] = versions
+    metadata["updated_at"] = now_iso()
+
+    try:
+        await checkpointer.aput(config, checkpoint, metadata, {"messages": new_ver})
+    except Exception:
+        logger.exception("Failed to clear messages from thread %s", sanitize_log_param(thread_id))
+        raise HTTPException(status_code=500, detail="Failed to clear thread messages")
+
+    logger.info(
+        "Cleared %d message(s) from thread %s",
+        len(messages),
+        sanitize_log_param(thread_id),
+    )
+    return {
+        "ok": True,
+        "deleted_ids": deleted_ids,
+        "deleted_count": len(messages),
+        "remaining": 0,
+    }
+
+
 @router.delete("/{thread_id}/messages/{message_id}")
 @require_permission("threads", "write", owner_check=True, require_existing=True)
 async def delete_thread_message(thread_id: str, message_id: str, request: Request) -> dict[str, Any]:
@@ -1266,6 +1348,88 @@ async def delete_thread_message(thread_id: str, message_id: str, request: Reques
     return {"ok": True, "deleted_ids": deleted_ids, "remaining": len(filtered)}
 
 
+@router.patch("/{thread_id}/messages/{message_id}")
+@require_permission("threads", "write", owner_check=True, require_existing=True)
+async def edit_thread_message(
+    thread_id: str,
+    message_id: str,
+    body: EditMessageRequest,
+    request: Request,
+) -> dict[str, Any]:
+    """Edit one message in a thread's latest checkpoint.
+
+    This is intentionally narrower than ``POST /state``: callers send only the
+    target id and replacement text, while the server preserves all other
+    message fields (attachments, timestamps, tool metadata, ids) and bumps the
+    ``messages`` channel version so the edited checkpoint is durable.
+    """
+    checkpointer = get_checkpointer(request)
+    config: dict[str, Any] = {
+        "configurable": {
+            "thread_id": thread_id,
+            "checkpoint_ns": "",
+        }
+    }
+    try:
+        checkpoint_tuple = await checkpointer.aget_tuple(config)
+    except Exception:
+        logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id))
+        raise HTTPException(status_code=500, detail="Failed to get thread state")
+
+    if checkpoint_tuple is None:
+        raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
+
+    checkpoint: dict[str, Any] = dict(getattr(checkpoint_tuple, "checkpoint", {}) or {})
+    metadata: dict[str, Any] = dict(getattr(checkpoint_tuple, "metadata", {}) or {})
+    channel_values: dict[str, Any] = dict(checkpoint.get("channel_values", {}))
+    messages = list(channel_values.get("messages") or [])
+    if not messages:
+        raise HTTPException(status_code=404, detail="Thread has no messages")
+
+    target_index = next(
+        (i for i, m in enumerate(messages) if _message_id(m) == message_id),
+        None,
+    )
+    if target_index is None:
+        raise HTTPException(status_code=404, detail=f"Message {message_id} not found")
+
+    try:
+        _set_message_content(messages[target_index], body.content)
+    except TypeError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+    channel_values["messages"] = messages
+    checkpoint["channel_values"] = channel_values
+    checkpoint["id"] = str(uuid6(clock_seq=-2))
+    checkpoint["ts"] = now_iso()
+    versions = dict(checkpoint.get("channel_versions", {}))
+    new_ver = checkpointer.get_next_version(versions.get("messages"), None)
+    versions["messages"] = new_ver
+    checkpoint["channel_versions"] = versions
+    metadata["updated_at"] = now_iso()
+
+    try:
+        await checkpointer.aput(config, checkpoint, metadata, {"messages": new_ver})
+    except Exception:
+        logger.exception(
+            "Failed to edit message %s from thread %s",
+            sanitize_log_param(message_id),
+            sanitize_log_param(thread_id),
+        )
+        raise HTTPException(status_code=500, detail="Failed to edit message")
+
+    serialized_messages = serialize_channel_values({"messages": [messages[target_index]]}).get("messages", [])
+    logger.info(
+        "Edited message %s in thread %s",
+        sanitize_log_param(message_id),
+        sanitize_log_param(thread_id),
+    )
+    return {
+        "ok": True,
+        "message": serialized_messages[0] if serialized_messages else {"id": message_id, "content": body.content},
+    }
+
+
 @router.post("/{thread_id}/history", response_model=list[HistoryEntry])
 @require_permission("threads", "read", owner_check=True)
 async def get_thread_history(thread_id: str, body: ThreadHistoryRequest, request: Request) -> list[HistoryEntry]: