mirror of
https://wget.la/https://github.com/leookun/cursor-byok
synced 2026-08-17 19:47:10 +08:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 leookun
|
4864d3675b | ||
|
leokunandGitHub
|
3f95318a49 | ||
|
leookun
|
9373e57ebf | ||
|
leokunandGitHub
|
f1992b0cfe | ||
|
haoge0211
|
67a9c27931 | ||
|
leookun
|
3cf8bdbc3c | ||
|
leokunandGitHub
|
684953a80b |
@@ -0,0 +1,95 @@
|
|||||||
|
<div align="center">
|
||||||
|
|
||||||
|
# cursor-byok
|
||||||
|
|
||||||
|
[使用教程](https://docs.leokun.cn) · [下载最新版](https://github.com/leookun/cursor-byok/releases/latest) · [问题反馈](https://github.com/leookun/cursor-byok/issues) · [English](./README.md)
|
||||||
|
|
||||||
|
[](https://github.com/leookun/cursor-byok/releases/latest)
|
||||||
|
[](https://github.com/leookun/cursor-byok/releases)
|
||||||
|
[](./LICENSE)
|
||||||
|
[](https://github.com/leookun/cursor-byok/releases/latest)
|
||||||
|
|
||||||
|
</div>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
## 项目介绍
|
||||||
|
|
||||||
|
cursor-byok 是一个开源的 Cursor 本地模型接入工具。它通过运行在本机的服务连接 Cursor 与你配置的模型 API，让模型请求使用自己的渠道处理，同时保留 Cursor Agent 的工具调用、Skills 和 MCP 等能力。
|
||||||
|
|
||||||
|
你可以接入 OpenAI、Anthropic 及其兼容服务，自由配置接口地址、模型、密钥和请求参数，不再局限于平台预设的模型渠道。
|
||||||
|
|
||||||
|
> [!IMPORTANT]
|
||||||
|
> cursor-byok 本身免费开源，但你接入的模型 API 可能由对应服务商收费。本项目不是 Cursor 官方产品，与 Cursor 或其开发公司无隶属关系。
|
||||||
|
|
||||||
|
## 核心能力
|
||||||
|
|
||||||
|
- **自定义模型渠道**：配置自己的 API 地址、访问密钥和模型标识。
|
||||||
|
- **多种接口协议**：支持 OpenAI、Anthropic 兼容接口及自定义端点。
|
||||||
|
- **模型管理**：添加、复制、编辑、排序和批量测试多个模型配置。
|
||||||
|
- **连接性能测试**：查看首字延迟、生成速度与模型服务的原始响应。
|
||||||
|
- **Agent 工作流**：支持工具调用、Skills、MCP 和多轮会话。
|
||||||
|
- **会话统计**：查看 Token 消耗、缓存命中率、对话轮次和价值估算。
|
||||||
|
- **跨平台运行**：支持 macOS、Windows 和 Linux。
|
||||||
|
|
||||||
|
## 快速开始
|
||||||
|
|
||||||
|
1. 从 [GitHub Releases](https://github.com/leookun/cursor-byok/releases/latest) 下载对应平台的最新版本。
|
||||||
|
2. 启动 cursor-byok，打开“模型配置”，填写接口地址、API Key 和模型标识。
|
||||||
|
3. 测试模型配置；测试通过后返回主界面启动服务。
|
||||||
|
4. 打开 Cursor，选择已配置的模型并开始使用 Agent。
|
||||||
|
|
||||||
|
更完整的安装、系统配置和常见问题说明，请查看 [详细使用教程](https://docs.leokun.cn)。
|
||||||
|
|
||||||
|
## 模型管理
|
||||||
|
|
||||||
|
模型配置支持 OpenAI 与 Anthropic 两类接口协议。每个模型渠道可以独立设置上下文窗口、最大输出 Token、推理强度、自定义请求头和额外请求参数。
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
## 工作原理
|
||||||
|
|
||||||
|
```text
|
||||||
|
Cursor 客户端
|
||||||
|
│
|
||||||
|
│ Agent 请求与工具结果
|
||||||
|
▼
|
||||||
|
cursor-byok 本地服务
|
||||||
|
│
|
||||||
|
│ OpenAI / Anthropic 兼容请求
|
||||||
|
▼
|
||||||
|
你配置的模型 API
|
||||||
|
```
|
||||||
|
|
||||||
|
cursor-byok 在本机负责协议适配、模型请求转发、工具调用衔接与会话状态管理。模型 API Key 和应用配置保存在本机；实际请求仍会发送到你所配置的模型服务商。
|
||||||
|
|
||||||
|
## 为什么做这个项目
|
||||||
|
|
||||||
|
很多 Agent 产品会将工具能力、模型选择、订阅方案和计费方式绑定在一起，用户只能使用平台提供的模型渠道。
|
||||||
|
|
||||||
|
我希望将模型选择权交还给用户：开发者可以充分利用已有的模型 API 和额度，自由选择适合自己的模型与服务商，也可以在需要时自托管相关服务。
|
||||||
|
|
||||||
|
## 路线图
|
||||||
|
|
||||||
|
项目将继续改进模型兼容性、Agent 工具链、本地运行稳定性和自托管体验，并探索更多 IDE、Chat 与 Agent 场景。
|
||||||
|
|
||||||
|
详细计划与进展请查看 [正式版路线图](https://github.com/leookun/cursor-byok/discussions/32)。
|
||||||
|
|
||||||
|
## 社区与支持
|
||||||
|
|
||||||
|
- [使用教程](https://docs.leokun.cn)
|
||||||
|
- [GitHub Issues](https://github.com/leookun/cursor-byok/issues)
|
||||||
|
- [Telegram 交流群](https://t.me/cursor_byok)
|
||||||
|
- QQ 交流群：`1095916242`、`1094411438`、`1095918002`、`1094419321`
|
||||||
|
|
||||||
|
<a href="https://trendshift.io/repositories/39260?utm_source=repository-badge&utm_medium=badge&utm_campaign=badge-repository-39260" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/repositories/39260" alt="leookun/cursor-byok | Trendshift" width="250" height="55" /></a>
|
||||||
|
|
||||||
|
## 开发与贡献
|
||||||
|
|
||||||
|
欢迎提交 Issue 和 Pull Request。开发环境、构建命令、项目结构及提交规范请阅读 [贡献指南](./CONTRIBUTING.md)。
|
||||||
|
|
||||||
|
## 许可证
|
||||||
|
|
||||||
|
本项目基于 [MIT License](./LICENSE) 开源。
|
||||||
@@ -1,95 +0,0 @@
|
|||||||
<div align="center">
|
|
||||||
|
|
||||||
# cursor-byok
|
|
||||||
|
|
||||||
[User Guide](https://docs.leokun.cn) · [Latest Release](https://github.com/leookun/cursor-byok/releases/latest) · [Report an Issue](https://github.com/leookun/cursor-byok/issues) · [简体中文](./README.md)
|
|
||||||
|
|
||||||
[](https://github.com/leookun/cursor-byok/releases/latest)
|
|
||||||
[](https://github.com/leookun/cursor-byok/releases)
|
|
||||||
[](./LICENSE)
|
|
||||||
[](https://github.com/leookun/cursor-byok/releases/latest)
|
|
||||||
|
|
||||||
</div>
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
## About
|
|
||||||
|
|
||||||
cursor-byok is an open-source local model gateway for Cursor. It runs a service on your machine that connects Cursor to the model APIs you configure, routes model requests through your own providers, and preserves Cursor Agent capabilities such as tool calling, Skills, and MCP.
|
|
||||||
|
|
||||||
You can connect OpenAI- and Anthropic-compatible services, customize endpoints, model IDs, API keys, and request parameters, and use model channels beyond the options built into the platform.
|
|
||||||
|
|
||||||
> [!IMPORTANT]
|
|
||||||
> cursor-byok is free and open source, but the model APIs you connect may charge for usage. This is an independent project and is not affiliated with or endorsed by Cursor or its developers.
|
|
||||||
|
|
||||||
## Features
|
|
||||||
|
|
||||||
- **Bring your own model channels:** Configure your own API endpoint, credentials, and model IDs.
|
|
||||||
- **Multiple API protocols:** Use OpenAI- and Anthropic-compatible APIs or a custom endpoint.
|
|
||||||
- **Model management:** Add, duplicate, edit, reorder, and batch-test multiple model configurations.
|
|
||||||
- **Connection benchmarks:** Measure time to first token, generation speed, and inspect raw provider responses.
|
|
||||||
- **Agent workflows:** Keep tool calling, Skills, MCP, and multi-turn conversations available.
|
|
||||||
- **Session metrics:** Track token usage, cache hit rate, conversation turns, and estimated value.
|
|
||||||
- **Cross-platform:** Run on macOS, Windows, and Linux.
|
|
||||||
|
|
||||||
## Quick Start
|
|
||||||
|
|
||||||
1. Download the latest build for your platform from [GitHub Releases](https://github.com/leookun/cursor-byok/releases/latest).
|
|
||||||
2. Launch cursor-byok, open **Model Settings**, and enter the endpoint, API key, and model ID.
|
|
||||||
3. Test the model configuration. Once it passes, return to the dashboard and start the service.
|
|
||||||
4. Open Cursor, select the configured model, and start using Agent.
|
|
||||||
|
|
||||||
For complete installation steps, system configuration, and troubleshooting, see the [User Guide](https://docs.leokun.cn).
|
|
||||||
|
|
||||||
## Model Management
|
|
||||||
|
|
||||||
Model configurations support both OpenAI and Anthropic API protocols. Each model channel can independently define its context window, maximum output tokens, reasoning effort, custom headers, and additional request parameters.
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
## How It Works
|
|
||||||
|
|
||||||
```text
|
|
||||||
Cursor client
|
|
||||||
│
|
|
||||||
│ Agent requests and tool results
|
|
||||||
▼
|
|
||||||
cursor-byok local service
|
|
||||||
│
|
|
||||||
│ OpenAI- / Anthropic-compatible requests
|
|
||||||
▼
|
|
||||||
Your model API
|
|
||||||
```
|
|
||||||
|
|
||||||
cursor-byok handles protocol adaptation, model request forwarding, tool-call coordination, and conversation state on your machine. API keys and application settings are stored locally; requests are still sent to the model provider you configure.
|
|
||||||
|
|
||||||
## Why This Project
|
|
||||||
|
|
||||||
Many Agent products bundle their tool capabilities with a fixed set of models, subscriptions, and billing options, leaving users limited to the channels offered by the platform.
|
|
||||||
|
|
||||||
cursor-byok is built to return model choice to the user. Developers can make full use of the APIs and credits they already have, choose the models and providers that fit their needs, and self-host related services when required.
|
|
||||||
|
|
||||||
## Roadmap
|
|
||||||
|
|
||||||
The project will continue to improve model compatibility, Agent tooling, local runtime stability, and the self-hosting experience while exploring support for more IDE, chat, and Agent workflows.
|
|
||||||
|
|
||||||
See the [release roadmap](https://github.com/leookun/cursor-byok/discussions/32) for plans and progress.
|
|
||||||
|
|
||||||
## Community and Support
|
|
||||||
|
|
||||||
- [User Guide](https://docs.leokun.cn)
|
|
||||||
- [GitHub Issues](https://github.com/leookun/cursor-byok/issues)
|
|
||||||
- [Telegram community](https://t.me/cursor_byok)
|
|
||||||
- QQ groups: `1095916242`, `1094411438`, `1095918002`, `1094419321`
|
|
||||||
|
|
||||||
<a href="https://trendshift.io/repositories/39260?utm_source=repository-badge&utm_medium=badge&utm_campaign=badge-repository-39260" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/repositories/39260" alt="leookun/cursor-byok | Trendshift" width="250" height="55" /></a>
|
|
||||||
|
|
||||||
## Development and Contributing
|
|
||||||
|
|
||||||
Issues and pull requests are welcome. See the [Contributing Guide](./CONTRIBUTING_EN.md) for prerequisites, build commands, project structure, and contribution guidelines.
|
|
||||||
|
|
||||||
## License
|
|
||||||
|
|
||||||
This project is open source under the [MIT License](./LICENSE).
|
|
||||||
@@ -2,7 +2,7 @@
|
|||||||
|
|
||||||
# cursor-byok
|
# cursor-byok
|
||||||
|
|
||||||
[使用教程](https://docs.leokun.cn) · [下载最新版](https://github.com/leookun/cursor-byok/releases/latest) · [问题反馈](https://github.com/leookun/cursor-byok/issues) · [English](./README-EN.md)
|
[User Guide](https://docs.leokun.cn) · [Latest Release](https://github.com/leookun/cursor-byok/releases/latest) · [Report an Issue](https://github.com/leookun/cursor-byok/issues) · [简体中文](./README-CN.md)
|
||||||
|
|
||||||
[](https://github.com/leookun/cursor-byok/releases/latest)
|
[](https://github.com/leookun/cursor-byok/releases/latest)
|
||||||
[](https://github.com/leookun/cursor-byok/releases)
|
[](https://github.com/leookun/cursor-byok/releases)
|
||||||
@@ -11,85 +11,85 @@
|
|||||||
|
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
## 项目介绍
|
## About
|
||||||
|
|
||||||
cursor-byok 是一个开源的 Cursor 本地模型接入工具。它通过运行在本机的服务连接 Cursor 与你配置的模型 API，让模型请求使用自己的渠道处理，同时保留 Cursor Agent 的工具调用、Skills 和 MCP 等能力。
|
cursor-byok is an open-source local model gateway for Cursor. It runs a service on your machine that connects Cursor to the model APIs you configure, routes model requests through your own providers, and preserves Cursor Agent capabilities such as tool calling, Skills, and MCP.
|
||||||
|
|
||||||
你可以接入 OpenAI、Anthropic 及其兼容服务，自由配置接口地址、模型、密钥和请求参数，不再局限于平台预设的模型渠道。
|
You can connect OpenAI- and Anthropic-compatible services, customize endpoints, model IDs, API keys, and request parameters, and use model channels beyond the options built into the platform.
|
||||||
|
|
||||||
> [!IMPORTANT]
|
> [!IMPORTANT]
|
||||||
> cursor-byok 本身免费开源，但你接入的模型 API 可能由对应服务商收费。本项目不是 Cursor 官方产品，与 Cursor 或其开发公司无隶属关系。
|
> cursor-byok is free and open source, but the model APIs you connect may charge for usage. This is an independent project and is not affiliated with or endorsed by Cursor or its developers.
|
||||||
|
|
||||||
## 核心能力
|
## Features
|
||||||
|
|
||||||
- **自定义模型渠道**：配置自己的 API 地址、访问密钥和模型标识。
|
- **Bring your own model channels:** Configure your own API endpoint, credentials, and model IDs.
|
||||||
- **多种接口协议**：支持 OpenAI、Anthropic 兼容接口及自定义端点。
|
- **Multiple API protocols:** Use OpenAI- and Anthropic-compatible APIs or a custom endpoint.
|
||||||
- **模型管理**：添加、复制、编辑、排序和批量测试多个模型配置。
|
- **Model management:** Add, duplicate, edit, reorder, and batch-test multiple model configurations.
|
||||||
- **连接性能测试**：查看首字延迟、生成速度与模型服务的原始响应。
|
- **Connection benchmarks:** Measure time to first token, generation speed, and inspect raw provider responses.
|
||||||
- **Agent 工作流**：支持工具调用、Skills、MCP 和多轮会话。
|
- **Agent workflows:** Keep tool calling, Skills, MCP, and multi-turn conversations available.
|
||||||
- **会话统计**：查看 Token 消耗、缓存命中率、对话轮次和价值估算。
|
- **Session metrics:** Track token usage, cache hit rate, conversation turns, and estimated value.
|
||||||
- **跨平台运行**：支持 macOS、Windows 和 Linux。
|
- **Cross-platform:** Run on macOS, Windows, and Linux.
|
||||||
|
|
||||||
## 快速开始
|
## Quick Start
|
||||||
|
|
||||||
1. 从 [GitHub Releases](https://github.com/leookun/cursor-byok/releases/latest) 下载对应平台的最新版本。
|
1. Download the latest build for your platform from [GitHub Releases](https://github.com/leookun/cursor-byok/releases/latest).
|
||||||
2. 启动 cursor-byok，打开“模型配置”，填写接口地址、API Key 和模型标识。
|
2. Launch cursor-byok, open **Model Settings**, and enter the endpoint, API key, and model ID.
|
||||||
3. 测试模型配置；测试通过后返回主界面启动服务。
|
3. Test the model configuration. Once it passes, return to the dashboard and start the service.
|
||||||
4. 打开 Cursor，选择已配置的模型并开始使用 Agent。
|
4. Open Cursor, select the configured model, and start using Agent.
|
||||||
|
|
||||||
更完整的安装、系统配置和常见问题说明，请查看 [详细使用教程](https://docs.leokun.cn)。
|
For complete installation steps, system configuration, and troubleshooting, see the [User Guide](https://docs.leokun.cn).
|
||||||
|
|
||||||
## 模型管理
|
## Model Management
|
||||||
|
|
||||||
模型配置支持 OpenAI 与 Anthropic 两类接口协议。每个模型渠道可以独立设置上下文窗口、最大输出 Token、推理强度、自定义请求头和额外请求参数。
|
Model configurations support both OpenAI and Anthropic API protocols. Each model channel can independently define its context window, maximum output tokens, reasoning effort, custom headers, and additional request parameters.
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
## 工作原理
|
## How It Works
|
||||||
|
|
||||||
```text
|
```text
|
||||||
Cursor 客户端
|
Cursor client
|
||||||
│
|
│
|
||||||
│ Agent 请求与工具结果
|
│ Agent requests and tool results
|
||||||
▼
|
▼
|
||||||
cursor-byok 本地服务
|
cursor-byok local service
|
||||||
│
|
│
|
||||||
│ OpenAI / Anthropic 兼容请求
|
│ OpenAI- / Anthropic-compatible requests
|
||||||
▼
|
▼
|
||||||
你配置的模型 API
|
Your model API
|
||||||
```
|
```
|
||||||
|
|
||||||
cursor-byok 在本机负责协议适配、模型请求转发、工具调用衔接与会话状态管理。模型 API Key 和应用配置保存在本机；实际请求仍会发送到你所配置的模型服务商。
|
cursor-byok handles protocol adaptation, model request forwarding, tool-call coordination, and conversation state on your machine. API keys and application settings are stored locally; requests are still sent to the model provider you configure.
|
||||||
|
|
||||||
## 为什么做这个项目
|
## Why This Project
|
||||||
|
|
||||||
很多 Agent 产品会将工具能力、模型选择、订阅方案和计费方式绑定在一起，用户只能使用平台提供的模型渠道。
|
Many Agent products bundle their tool capabilities with a fixed set of models, subscriptions, and billing options, leaving users limited to the channels offered by the platform.
|
||||||
|
|
||||||
我希望将模型选择权交还给用户：开发者可以充分利用已有的模型 API 和额度，自由选择适合自己的模型与服务商，也可以在需要时自托管相关服务。
|
cursor-byok is built to return model choice to the user. Developers can make full use of the APIs and credits they already have, choose the models and providers that fit their needs, and self-host related services when required.
|
||||||
|
|
||||||
## 路线图
|
## Roadmap
|
||||||
|
|
||||||
项目将继续改进模型兼容性、Agent 工具链、本地运行稳定性和自托管体验，并探索更多 IDE、Chat 与 Agent 场景。
|
The project will continue to improve model compatibility, Agent tooling, local runtime stability, and the self-hosting experience while exploring support for more IDE, chat, and Agent workflows.
|
||||||
|
|
||||||
详细计划与进展请查看 [正式版路线图](https://github.com/leookun/cursor-byok/discussions/32)。
|
See the [release roadmap](https://github.com/leookun/cursor-byok/discussions/32) for plans and progress.
|
||||||
|
|
||||||
## 社区与支持
|
## Community and Support
|
||||||
|
|
||||||
- [使用教程](https://docs.leokun.cn)
|
- [User Guide](https://docs.leokun.cn)
|
||||||
- [GitHub Issues](https://github.com/leookun/cursor-byok/issues)
|
- [GitHub Issues](https://github.com/leookun/cursor-byok/issues)
|
||||||
- [Telegram 交流群](https://t.me/cursor_byok)
|
- [Telegram community](https://t.me/cursor_byok)
|
||||||
- QQ 交流群：`1095916242`、`1094411438`、`1095918002`、`1094419321`
|
- QQ groups: `1095916242`, `1094411438`, `1095918002`, `1094419321`
|
||||||
|
|
||||||
<a href="https://trendshift.io/repositories/39260?utm_source=repository-badge&utm_medium=badge&utm_campaign=badge-repository-39260" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/repositories/39260" alt="leookun/cursor-byok | Trendshift" width="250" height="55" /></a>
|
<a href="https://trendshift.io/repositories/39260?utm_source=repository-badge&utm_medium=badge&utm_campaign=badge-repository-39260" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/repositories/39260" alt="leookun/cursor-byok | Trendshift" width="250" height="55" /></a>
|
||||||
|
|
||||||
## 开发与贡献
|
## Development and Contributing
|
||||||
|
|
||||||
欢迎提交 Issue 和 Pull Request。开发环境、构建命令、项目结构及提交规范请阅读 [贡献指南](./CONTRIBUTING.md)。
|
Issues and pull requests are welcome. See the [Contributing Guide](./CONTRIBUTING_EN.md) for prerequisites, build commands, project structure, and contribution guidelines.
|
||||||
|
|
||||||
## 许可证
|
## License
|
||||||
|
|
||||||
本项目基于 [MIT License](./LICENSE) 开源。
|
This project is open source under the [MIT License](./LICENSE).
|
||||||
|
|||||||
+1
-1
@@ -8,7 +8,7 @@ info:
|
|||||||
description: "Cursor助手"
|
description: "Cursor助手"
|
||||||
copyright: "© 2026, Cursor助手"
|
copyright: "© 2026, Cursor助手"
|
||||||
comments: "Cursor助手"
|
comments: "Cursor助手"
|
||||||
version: "0.0.46"
|
version: "0.0.47"
|
||||||
|
|
||||||
dev_mode:
|
dev_mode:
|
||||||
root_path: .
|
root_path: .
|
||||||
|
|||||||
@@ -17,9 +17,9 @@
|
|||||||
<key>CFBundlePackageType</key>
|
<key>CFBundlePackageType</key>
|
||||||
<string>APPL</string>
|
<string>APPL</string>
|
||||||
<key>CFBundleShortVersionString</key>
|
<key>CFBundleShortVersionString</key>
|
||||||
<string>0.0.46</string>
|
<string>0.0.47</string>
|
||||||
<key>CFBundleVersion</key>
|
<key>CFBundleVersion</key>
|
||||||
<string>0.0.46</string>
|
<string>0.0.47</string>
|
||||||
<key>LSMinimumSystemVersion</key>
|
<key>LSMinimumSystemVersion</key>
|
||||||
<string>12.0.0</string>
|
<string>12.0.0</string>
|
||||||
<key>LSUIElement</key>
|
<key>LSUIElement</key>
|
||||||
|
|||||||
@@ -17,9 +17,9 @@
|
|||||||
<key>CFBundlePackageType</key>
|
<key>CFBundlePackageType</key>
|
||||||
<string>APPL</string>
|
<string>APPL</string>
|
||||||
<key>CFBundleShortVersionString</key>
|
<key>CFBundleShortVersionString</key>
|
||||||
<string>0.0.46</string>
|
<string>0.0.47</string>
|
||||||
<key>CFBundleVersion</key>
|
<key>CFBundleVersion</key>
|
||||||
<string>0.0.46</string>
|
<string>0.0.47</string>
|
||||||
<key>LSMinimumSystemVersion</key>
|
<key>LSMinimumSystemVersion</key>
|
||||||
<string>12.0.0</string>
|
<string>12.0.0</string>
|
||||||
<key>LSUIElement</key>
|
<key>LSUIElement</key>
|
||||||
|
|||||||
@@ -6,7 +6,7 @@
|
|||||||
name: "Cursor助手"
|
name: "Cursor助手"
|
||||||
arch: ${GOARCH}
|
arch: ${GOARCH}
|
||||||
platform: "linux"
|
platform: "linux"
|
||||||
version: "0.0.46"
|
version: "0.0.47"
|
||||||
section: "default"
|
section: "default"
|
||||||
priority: "extra"
|
priority: "extra"
|
||||||
maintainer: ${GIT_COMMITTER_NAME} <${GIT_COMMITTER_EMAIL}>
|
maintainer: ${GIT_COMMITTER_NAME} <${GIT_COMMITTER_EMAIL}>
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
{
|
{
|
||||||
"fixed": {
|
"fixed": {
|
||||||
"file_version": "0.0.46"
|
"file_version": "0.0.47"
|
||||||
},
|
},
|
||||||
"info": {
|
"info": {
|
||||||
"0000": {
|
"0000": {
|
||||||
"ProductVersion": "0.0.46",
|
"ProductVersion": "0.0.47",
|
||||||
"CompanyName": "Cursor助手",
|
"CompanyName": "Cursor助手",
|
||||||
"FileDescription": "Cursor助手",
|
"FileDescription": "Cursor助手",
|
||||||
"LegalCopyright": "© 2026, Cursor助手",
|
"LegalCopyright": "© 2026, Cursor助手",
|
||||||
|
|||||||
@@ -14,7 +14,7 @@
|
|||||||
!define INFO_PRODUCTNAME "Cursor助手"
|
!define INFO_PRODUCTNAME "Cursor助手"
|
||||||
!endif
|
!endif
|
||||||
!ifndef INFO_PRODUCTVERSION
|
!ifndef INFO_PRODUCTVERSION
|
||||||
!define INFO_PRODUCTVERSION "0.0.46"
|
!define INFO_PRODUCTVERSION "0.0.47"
|
||||||
!endif
|
!endif
|
||||||
!ifndef INFO_COPYRIGHT
|
!ifndef INFO_COPYRIGHT
|
||||||
!define INFO_COPYRIGHT "© 2026, Cursor助手"
|
!define INFO_COPYRIGHT "© 2026, Cursor助手"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
|
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
|
||||||
<assembly manifestVersion="1.0" xmlns="urn:schemas-microsoft-com:asm.v1" xmlns:asmv3="urn:schemas-microsoft-com:asm.v3">
|
<assembly manifestVersion="1.0" xmlns="urn:schemas-microsoft-com:asm.v1" xmlns:asmv3="urn:schemas-microsoft-com:asm.v3">
|
||||||
<assemblyIdentity type="win32" name="com.cursor.wuxianxubei" version="0.0.46" processorArchitecture="*"/>
|
<assemblyIdentity type="win32" name="com.cursor.wuxianxubei" version="0.0.47" processorArchitecture="*"/>
|
||||||
<dependency>
|
<dependency>
|
||||||
<dependentAssembly>
|
<dependentAssembly>
|
||||||
<assemblyIdentity type="win32" name="Microsoft.Windows.Common-Controls" version="6.0.0.0" processorArchitecture="*" publicKeyToken="6595b64144ccf1df" language="*"/>
|
<assemblyIdentity type="win32" name="Microsoft.Windows.Common-Controls" version="6.0.0.0" processorArchitecture="*" publicKeyToken="6595b64144ccf1df" language="*"/>
|
||||||
|
|||||||
@@ -50,6 +50,7 @@ type openAIResponsesRequestBody struct {
|
|||||||
|
|
||||||
type openAIResponsesReasoning struct {
|
type openAIResponsesReasoning struct {
|
||||||
Effort string `json:"effort,omitempty"`
|
Effort string `json:"effort,omitempty"`
|
||||||
|
Summary string `json:"summary,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type openAIToolAccumulator struct {
|
type openAIToolAccumulator struct {
|
||||||
@@ -944,7 +945,7 @@ func (adapter *OpenAIAdapter) streamResponses(ctx context.Context, req StreamReq
|
|||||||
requestBody.Tools = tools
|
requestBody.Tools = tools
|
||||||
}
|
}
|
||||||
if effort := strings.TrimSpace(req.ReasoningEffort); effort != "" {
|
if effort := strings.TrimSpace(req.ReasoningEffort); effort != "" {
|
||||||
requestBody.Reasoning = &openAIResponsesReasoning{Effort: effort}
|
requestBody.Reasoning = &openAIResponsesReasoning{Effort: effort, Summary: "auto"}
|
||||||
requestBody.Include = []string{"reasoning.encrypted_content"}
|
requestBody.Include = []string{"reasoning.encrypted_content"}
|
||||||
}
|
}
|
||||||
body = requestBody
|
body = requestBody
|
||||||
|
|||||||
@@ -2,12 +2,67 @@ package modeladapter
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func TestOpenAIResponsesRequestsReasoningSummary(t *testing.T) {
|
||||||
|
var requestBody map[string]any
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
||||||
|
if err := json.NewDecoder(request.Body).Decode(&requestBody); err != nil {
|
||||||
|
http.Error(writer, err.Error(), http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writer.Header().Set("Content-Type", "text/event-stream")
|
||||||
|
_, _ = fmt.Fprint(writer, "data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"check the request\"}\n\n")
|
||||||
|
_, _ = fmt.Fprint(writer, "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"model\":\"gpt-5.6\",\"status\":\"completed\",\"output_text\":\"done\"}}\n\n")
|
||||||
|
_, _ = fmt.Fprint(writer, "data: [DONE]\n\n")
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
|
||||||
|
adapter := &OpenAIAdapter{client: server.Client()}
|
||||||
|
events := make([]ModelEvent, 0, 4)
|
||||||
|
err := adapter.Stream(context.Background(), StreamRequest{
|
||||||
|
RequestID: "request-1",
|
||||||
|
RunID: "run-1",
|
||||||
|
ModelCallID: "model-call-1",
|
||||||
|
BaseURL: server.URL,
|
||||||
|
APIKey: "test-key",
|
||||||
|
ProviderModelID: "gpt-5.6",
|
||||||
|
OpenAIEndpoint: "/v1/responses",
|
||||||
|
ReasoningEffort: "high",
|
||||||
|
Messages: []Message{{Role: "user", Content: "hello"}},
|
||||||
|
MaxTokens: 128,
|
||||||
|
}, func(event ModelEvent) error {
|
||||||
|
events = append(events, event)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("stream failed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
reasoning, ok := requestBody["reasoning"].(map[string]any)
|
||||||
|
if !ok {
|
||||||
|
t.Fatalf("reasoning request body missing: %#v", requestBody)
|
||||||
|
}
|
||||||
|
if got := reasoning["effort"]; got != "high" {
|
||||||
|
t.Fatalf("reasoning.effort = %#v, want high", got)
|
||||||
|
}
|
||||||
|
if got := reasoning["summary"]; got != "auto" {
|
||||||
|
t.Fatalf("reasoning.summary = %#v, want auto", got)
|
||||||
|
}
|
||||||
|
include, ok := requestBody["include"].([]any)
|
||||||
|
if !ok || len(include) != 1 || include[0] != "reasoning.encrypted_content" {
|
||||||
|
t.Fatalf("reasoning include = %#v, want encrypted content", requestBody["include"])
|
||||||
|
}
|
||||||
|
assertOpenAIEventKindCount(t, events, ModelEventKindThinkingDelta, 1)
|
||||||
|
assertOpenAIEventKindCount(t, events, ModelEventKindThinkingCompleted, 1)
|
||||||
|
assertOpenAIEventKindCount(t, events, ModelEventKindTextDelta, 1)
|
||||||
|
}
|
||||||
|
|
||||||
func TestOpenAIChatCompletionsIgnoresBlankFinishReason(t *testing.T) {
|
func TestOpenAIChatCompletionsIgnoresBlankFinishReason(t *testing.T) {
|
||||||
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
||||||
writer.Header().Set("Content-Type", "text/event-stream")
|
writer.Header().Set("Content-Type", "text/event-stream")
|
||||||
|
|||||||
@@ -730,6 +730,9 @@ func (service *Service) handleProviderDoneEvent(stream *ActiveStream, payload *s
|
|||||||
service.setTurnPhase(stream, TurnPhaseFailed)
|
service.setTurnPhase(stream, TurnPhaseFailed)
|
||||||
return service.closeStreamWithProviderError(stream, conversationID, turnSeq, requestID, accumulatedText, accumulatedReasoning, accumulatedReasoningSignature, accumulatedReasoningSignatureSource, accumulatedReasoningItemID, accumulatedReasoningStatus, accumulatedReasoningSummary, usage, providerErr, !hadToolInvocation)
|
return service.closeStreamWithProviderError(stream, conversationID, turnSeq, requestID, accumulatedText, accumulatedReasoning, accumulatedReasoningSignature, accumulatedReasoningSignatureSource, accumulatedReasoningItemID, accumulatedReasoningStatus, accumulatedReasoningSummary, usage, providerErr, !hadToolInvocation)
|
||||||
}
|
}
|
||||||
|
if err := service.flushAssistantText(stream, conversationID, turnSeq, requestID, accumulatedText, accumulatedReasoning, accumulatedReasoningSignature, accumulatedReasoningSignatureSource, accumulatedReasoningItemID, accumulatedReasoningStatus, accumulatedReasoningSummary, !hadToolInvocation); err != nil {
|
||||||
|
return service.failStream(stream, "unknown", fmt.Errorf("flush failed provider output: %w", err))
|
||||||
|
}
|
||||||
service.setTurnPhase(stream, TurnPhaseFailed)
|
service.setTurnPhase(stream, TurnPhaseFailed)
|
||||||
return service.failStream(stream, "unknown", payload.Err)
|
return service.failStream(stream, "unknown", payload.Err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,15 +19,29 @@ type pendingCheckpointBlobWrite struct {
|
|||||||
blob CheckpointBlob
|
blob CheckpointBlob
|
||||||
}
|
}
|
||||||
|
|
||||||
func clonePendingTurnCompletion(completion *pendingTurnCompletion) *pendingTurnCompletion {
|
func successfulCheckpointTerminalAction(completion *pendingTurnCompletion) checkpointTerminalAction {
|
||||||
if completion == nil {
|
if completion == nil {
|
||||||
return nil
|
return checkpointTerminalAction{}
|
||||||
|
}
|
||||||
|
return checkpointTerminalAction{
|
||||||
|
Kind: checkpointTerminalActionComplete,
|
||||||
|
Completion: *completion,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func failedCheckpointTerminalAction(errorCode string, errorMessage string) checkpointTerminalAction {
|
||||||
|
return checkpointTerminalAction{
|
||||||
|
Kind: checkpointTerminalActionFail,
|
||||||
|
ErrorCode: strings.TrimSpace(errorCode),
|
||||||
|
ErrorMessage: strings.TrimSpace(errorMessage),
|
||||||
}
|
}
|
||||||
cloned := *completion
|
|
||||||
return &cloned
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (service *Service) queueCheckpointProjection(stream *ActiveStream, projection *CheckpointProjection, completion *pendingTurnCompletion) error {
|
func (service *Service) queueCheckpointProjection(stream *ActiveStream, projection *CheckpointProjection, completion *pendingTurnCompletion) error {
|
||||||
|
return service.queueCheckpointProjectionWithTerminal(stream, projection, successfulCheckpointTerminalAction(completion))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (service *Service) queueCheckpointProjectionWithTerminal(stream *ActiveStream, projection *CheckpointProjection, terminal checkpointTerminalAction) error {
|
||||||
if service == nil || stream == nil || projection == nil || projection.State == nil {
|
if service == nil || stream == nil || projection == nil || projection.State == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -43,8 +57,8 @@ func (service *Service) queueCheckpointProjection(stream *ActiveStream, projecti
|
|||||||
if stream.ConfirmedCheckpointBlobs == nil {
|
if stream.ConfirmedCheckpointBlobs == nil {
|
||||||
stream.ConfirmedCheckpointBlobs = make(map[string]struct{})
|
stream.ConfirmedCheckpointBlobs = make(map[string]struct{})
|
||||||
}
|
}
|
||||||
if completion == nil && stream.PendingCheckpoint != nil {
|
if terminal.Kind == checkpointTerminalActionNone && stream.PendingCheckpoint != nil {
|
||||||
completion = stream.PendingCheckpoint.Completion
|
terminal = stream.PendingCheckpoint.Terminal
|
||||||
}
|
}
|
||||||
required := make(map[string]struct{}, len(projection.Blobs))
|
required := make(map[string]struct{}, len(projection.Blobs))
|
||||||
pendingKeys := make(map[string]struct{}, len(stream.PendingCheckpointBlobWrites))
|
pendingKeys := make(map[string]struct{}, len(stream.PendingCheckpointBlobWrites))
|
||||||
@@ -76,9 +90,9 @@ func (service *Service) queueCheckpointProjection(stream *ActiveStream, projecti
|
|||||||
stream.PendingCheckpoint = &pendingCheckpointPublish{
|
stream.PendingCheckpoint = &pendingCheckpointPublish{
|
||||||
State: state,
|
State: state,
|
||||||
Required: required,
|
Required: required,
|
||||||
Completion: clonePendingTurnCompletion(completion),
|
Terminal: terminal,
|
||||||
}
|
}
|
||||||
if completion != nil {
|
if terminal.Kind != checkpointTerminalActionNone {
|
||||||
stream.Phase = TurnPhaseCheckpointing
|
stream.Phase = TurnPhaseCheckpointing
|
||||||
}
|
}
|
||||||
stream.UpdatedAt = time.Now().UTC()
|
stream.UpdatedAt = time.Now().UTC()
|
||||||
@@ -94,13 +108,8 @@ func (service *Service) queueCheckpointProjection(stream *ActiveStream, projecti
|
|||||||
if service.checkpointProjectionReady(stream) {
|
if service.checkpointProjectionReady(stream) {
|
||||||
return service.publishReadyCheckpoint(stream)
|
return service.publishReadyCheckpoint(stream)
|
||||||
}
|
}
|
||||||
// Keep the latest live UI state ahead of an immediate client abort. Blob writes are
|
// Checkpoints reference these Blob IDs, so the client must confirm every
|
||||||
// ordered before this snapshot; acknowledgements still gate terminal completion.
|
// required Blob before the checkpoint becomes visible.
|
||||||
if completion == nil {
|
|
||||||
if err := service.publishPendingCheckpoint(stream); err != nil {
|
|
||||||
return service.finishAfterCheckpointSyncFailure(stream, fmt.Errorf("publish pending checkpoint: %w", err))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
service.scheduleStreamTimer(
|
service.scheduleStreamTimer(
|
||||||
stream,
|
stream,
|
||||||
providerTimerKey(streamTimerCheckpointBlobs, ""),
|
providerTimerKey(streamTimerCheckpointBlobs, ""),
|
||||||
@@ -113,31 +122,6 @@ func (service *Service) queueCheckpointProjection(stream *ActiveStream, projecti
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (service *Service) publishPendingCheckpoint(stream *ActiveStream) error {
|
|
||||||
if service == nil || stream == nil {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
stream.mu.Lock()
|
|
||||||
pending := stream.PendingCheckpoint
|
|
||||||
if pending == nil || pending.Published {
|
|
||||||
stream.mu.Unlock()
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
pending.Published = true
|
|
||||||
state := pending.State
|
|
||||||
stream.UpdatedAt = time.Now().UTC()
|
|
||||||
stream.mu.Unlock()
|
|
||||||
if err := service.broker.Publish(stream.RequestID, StreamEvent{Message: buildCheckpointMessage(state)}); err != nil {
|
|
||||||
stream.mu.Lock()
|
|
||||||
if stream.PendingCheckpoint == pending {
|
|
||||||
pending.Published = false
|
|
||||||
}
|
|
||||||
stream.mu.Unlock()
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (service *Service) checkpointProjectionReady(stream *ActiveStream) bool {
|
func (service *Service) checkpointProjectionReady(stream *ActiveStream) bool {
|
||||||
if stream == nil {
|
if stream == nil {
|
||||||
return false
|
return false
|
||||||
@@ -207,24 +191,18 @@ func (service *Service) publishReadyCheckpoint(stream *ActiveStream) error {
|
|||||||
}
|
}
|
||||||
stream.PendingCheckpoint = nil
|
stream.PendingCheckpoint = nil
|
||||||
state := pending.State
|
state := pending.State
|
||||||
completion := clonePendingTurnCompletion(pending.Completion)
|
terminal := pending.Terminal
|
||||||
published := pending.Published
|
|
||||||
stream.UpdatedAt = time.Now().UTC()
|
stream.UpdatedAt = time.Now().UTC()
|
||||||
stream.mu.Unlock()
|
stream.mu.Unlock()
|
||||||
clearStreamTimer(stream, providerTimerKey(streamTimerCheckpointBlobs, ""))
|
clearStreamTimer(stream, providerTimerKey(streamTimerCheckpointBlobs, ""))
|
||||||
if !published {
|
|
||||||
if err := service.broker.Publish(stream.RequestID, StreamEvent{Message: buildCheckpointMessage(state)}); err != nil {
|
if err := service.broker.Publish(stream.RequestID, StreamEvent{Message: buildCheckpointMessage(state)}); err != nil {
|
||||||
if completion != nil {
|
if terminal.Kind != checkpointTerminalActionNone {
|
||||||
log.Printf("forwarder checkpoint publish skipped before successful terminal request_id=%s err=%v", stream.RequestID, err)
|
log.Printf("forwarder checkpoint publish skipped before terminal request_id=%s err=%v", stream.RequestID, err)
|
||||||
return service.finishSuccessfulTurnAfterCheckpoint(stream, *completion)
|
return service.finishCheckpointTerminalAction(stream, terminal)
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
return service.finishCheckpointTerminalAction(stream, terminal)
|
||||||
if completion != nil {
|
|
||||||
return service.finishSuccessfulTurnAfterCheckpoint(stream, *completion)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (service *Service) handleCheckpointBlobTimeout(stream *ActiveStream) error {
|
func (service *Service) handleCheckpointBlobTimeout(stream *ActiveStream) error {
|
||||||
@@ -251,12 +229,23 @@ func (service *Service) finishAfterCheckpointSyncFailure(stream *ActiveStream, c
|
|||||||
if cause != nil {
|
if cause != nil {
|
||||||
log.Printf("forwarder checkpoint blob sync skipped request_id=%s conversation_id=%s err=%v", stream.RequestID, stream.ConversationID, cause)
|
log.Printf("forwarder checkpoint blob sync skipped request_id=%s conversation_id=%s err=%v", stream.RequestID, stream.ConversationID, cause)
|
||||||
}
|
}
|
||||||
if pending != nil && pending.Completion != nil {
|
if pending != nil {
|
||||||
return service.finishSuccessfulTurnAfterCheckpoint(stream, *pending.Completion)
|
return service.finishCheckpointTerminalAction(stream, pending.Terminal)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (service *Service) finishCheckpointTerminalAction(stream *ActiveStream, terminal checkpointTerminalAction) error {
|
||||||
|
switch terminal.Kind {
|
||||||
|
case checkpointTerminalActionComplete:
|
||||||
|
return service.finishSuccessfulTurnAfterCheckpoint(stream, terminal.Completion)
|
||||||
|
case checkpointTerminalActionFail:
|
||||||
|
return service.finishFailedTurnAfterCheckpoint(stream, terminal.ErrorCode, terminal.ErrorMessage)
|
||||||
|
default:
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (service *Service) discardPendingCheckpoint(stream *ActiveStream, reason string) {
|
func (service *Service) discardPendingCheckpoint(stream *ActiveStream, reason string) {
|
||||||
if stream == nil {
|
if stream == nil {
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -8,22 +8,43 @@ import (
|
|||||||
"cursor/gen/agentv1"
|
"cursor/gen/agentv1"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestCheckpointBlobSyncPublishesNonTerminalCheckpointBeforeAcknowledgements(t *testing.T) {
|
func TestCheckpointBlobSyncWaitsForAcknowledgementsBeforePublishingNonTerminalCheckpoint(t *testing.T) {
|
||||||
service, stream, projection := testCheckpointBlobProjection(t)
|
service, stream, projection := testCheckpointBlobProjection(t)
|
||||||
if err := service.queueCheckpointProjection(stream, projection, nil); err != nil {
|
if err := service.queueCheckpointProjection(stream, projection, nil); err != nil {
|
||||||
t.Fatalf("queueCheckpointProjection() error = %v", err)
|
t.Fatalf("queueCheckpointProjection() error = %v", err)
|
||||||
}
|
}
|
||||||
events := readCheckpointTestEvents(t, service, stream)
|
events := readCheckpointTestEvents(t, service, stream)
|
||||||
if len(events) != len(projection.Blobs)+1 {
|
if len(events) != len(projection.Blobs) {
|
||||||
t.Fatalf("events before ACK = %d, want %d Blob writes and one checkpoint", len(events), len(projection.Blobs))
|
t.Fatalf("events before ACK = %d, want %d Blob writes", len(events), len(projection.Blobs))
|
||||||
}
|
}
|
||||||
for _, event := range events[:len(projection.Blobs)] {
|
for _, event := range events {
|
||||||
if event.Message.GetKvServerMessage().GetSetBlobArgs() == nil {
|
if event.Message.GetKvServerMessage().GetSetBlobArgs() == nil {
|
||||||
t.Fatalf("event before ACK = %#v, want set_blob_args", event.Message)
|
t.Fatalf("event before ACK = %#v, want set_blob_args", event.Message)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if checkpoint := events[len(events)-1].Message.GetConversationCheckpointUpdate(); checkpoint == nil || len(checkpoint.GetTurns()) != 1 {
|
|
||||||
t.Fatalf("last event before ACK = %#v, want one Blob-backed turn", events[len(events)-1].Message)
|
stream.mu.Lock()
|
||||||
|
var firstRequestID uint32
|
||||||
|
for requestID := range stream.PendingCheckpointBlobWrites {
|
||||||
|
firstRequestID = requestID
|
||||||
|
break
|
||||||
|
}
|
||||||
|
stream.mu.Unlock()
|
||||||
|
if firstRequestID == 0 {
|
||||||
|
t.Fatal("checkpoint projection has no pending Blob writes")
|
||||||
|
}
|
||||||
|
if err := service.handleCheckpointBlobResult(stream, &agentv1.KvClientMessage{
|
||||||
|
Id: firstRequestID,
|
||||||
|
Message: &agentv1.KvClientMessage_SetBlobResult{
|
||||||
|
SetBlobResult: &agentv1.SetBlobResult{},
|
||||||
|
},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("first Blob ACK error = %v", err)
|
||||||
|
}
|
||||||
|
for _, event := range readCheckpointTestEvents(t, service, stream) {
|
||||||
|
if event.Message.GetConversationCheckpointUpdate() != nil {
|
||||||
|
t.Fatal("checkpoint published after only a partial Blob acknowledgement")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
acknowledgeCheckpointBlobs(t, service, stream)
|
acknowledgeCheckpointBlobs(t, service, stream)
|
||||||
@@ -98,7 +119,123 @@ func TestCheckpointBlobTimeoutDoesNotFailSuccessfulTurn(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestCancellationKeepsPublishedCheckpointAndIgnoresLateAcknowledgements(t *testing.T) {
|
func TestCheckpointBlobSyncPublishesCheckpointBeforeFailedTerminal(t *testing.T) {
|
||||||
|
service, stream, _ := testCheckpointBlobProjection(t)
|
||||||
|
if err := service.failActiveStream(
|
||||||
|
stream,
|
||||||
|
stream.ConversationID,
|
||||||
|
stream.RequestID,
|
||||||
|
"model-call-1",
|
||||||
|
"provider_error",
|
||||||
|
"provider failed",
|
||||||
|
); err != nil {
|
||||||
|
t.Fatalf("failActiveStream() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, event := range readCheckpointTestEvents(t, service, stream) {
|
||||||
|
if event.Message.GetConversationCheckpointUpdate() != nil || event.End {
|
||||||
|
t.Fatalf("event before ACK = %#v, want only Blob writes", event)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
stream.mu.Lock()
|
||||||
|
phaseBeforeACK := stream.Phase
|
||||||
|
statusBeforeACK := stream.Status
|
||||||
|
stream.mu.Unlock()
|
||||||
|
if phaseBeforeACK != TurnPhaseCheckpointing || isTerminalStreamStatus(statusBeforeACK) {
|
||||||
|
t.Fatalf("before ACK phase=%s status=%s, want checkpointing and non-terminal", phaseBeforeACK, statusBeforeACK)
|
||||||
|
}
|
||||||
|
|
||||||
|
acknowledgeCheckpointBlobs(t, service, stream)
|
||||||
|
events := readCheckpointTestEvents(t, service, stream)
|
||||||
|
checkpointIndex, endIndex := -1, -1
|
||||||
|
for index, event := range events {
|
||||||
|
switch {
|
||||||
|
case event.Message.GetConversationCheckpointUpdate() != nil:
|
||||||
|
checkpointIndex = index
|
||||||
|
case event.End:
|
||||||
|
endIndex = index
|
||||||
|
if event.TerminalErrorCode != "provider_error" || event.TerminalErrorMessage != "provider failed" {
|
||||||
|
t.Fatalf("terminal event = %#v, want provider error", event)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if checkpointIndex < 0 || endIndex <= checkpointIndex {
|
||||||
|
t.Fatalf("terminal order checkpoint=%d end=%d", checkpointIndex, endIndex)
|
||||||
|
}
|
||||||
|
stream.mu.Lock()
|
||||||
|
phaseAfterACK := stream.Phase
|
||||||
|
statusAfterACK := stream.Status
|
||||||
|
stream.mu.Unlock()
|
||||||
|
if phaseAfterACK != TurnPhaseFailed || statusAfterACK != StreamStatusFailed {
|
||||||
|
t.Fatalf("after ACK phase=%s status=%s, want failed", phaseAfterACK, statusAfterACK)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCheckpointBlobTimeoutStillPublishesFailedTerminal(t *testing.T) {
|
||||||
|
service, stream, _ := testCheckpointBlobProjection(t)
|
||||||
|
if err := service.failActiveStream(
|
||||||
|
stream,
|
||||||
|
stream.ConversationID,
|
||||||
|
stream.RequestID,
|
||||||
|
"model-call-1",
|
||||||
|
"provider_error",
|
||||||
|
"provider failed",
|
||||||
|
); err != nil {
|
||||||
|
t.Fatalf("failActiveStream() error = %v", err)
|
||||||
|
}
|
||||||
|
if err := service.handleCheckpointBlobTimeout(stream); err != nil {
|
||||||
|
t.Fatalf("handleCheckpointBlobTimeout() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
events := readCheckpointTestEvents(t, service, stream)
|
||||||
|
var checkpoint, failedEnd bool
|
||||||
|
for _, event := range events {
|
||||||
|
checkpoint = checkpoint || event.Message.GetConversationCheckpointUpdate() != nil
|
||||||
|
failedEnd = failedEnd || event.End && event.TerminalErrorCode == "provider_error" && event.TerminalErrorMessage == "provider failed"
|
||||||
|
}
|
||||||
|
if checkpoint || !failedEnd {
|
||||||
|
t.Fatalf("timeout events checkpoint=%v failed_end=%v", checkpoint, failedEnd)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestManualCompactionNoopWaitsForCheckpointBeforeTerminal(t *testing.T) {
|
||||||
|
service, stream, _ := testCheckpointBlobProjection(t)
|
||||||
|
conversation, _, _, err := service.snapshotCheckpointConversation(stream)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("snapshotCheckpointConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
if _, err := service.store.SaveConversationWithEntries(stream.ConversationID, conversation, conversation.Entries); err != nil {
|
||||||
|
t.Fatalf("SaveConversationWithEntries() error = %v", err)
|
||||||
|
}
|
||||||
|
if err := service.finishManualCompactionNoop(stream); err != nil {
|
||||||
|
t.Fatalf("finishManualCompactionNoop() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, event := range readCheckpointTestEvents(t, service, stream) {
|
||||||
|
if event.Message.GetInteractionUpdate().GetTurnEnded() != nil || event.End {
|
||||||
|
t.Fatalf("terminal event before checkpoint Blob ACK = %#v", event)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
acknowledgeCheckpointBlobs(t, service, stream)
|
||||||
|
|
||||||
|
events := readCheckpointTestEvents(t, service, stream)
|
||||||
|
checkpointIndex, turnEndedIndex, endIndex := -1, -1, -1
|
||||||
|
for index, event := range events {
|
||||||
|
switch {
|
||||||
|
case event.Message.GetConversationCheckpointUpdate() != nil:
|
||||||
|
checkpointIndex = index
|
||||||
|
case event.Message.GetInteractionUpdate().GetTurnEnded() != nil:
|
||||||
|
turnEndedIndex = index
|
||||||
|
case event.End:
|
||||||
|
endIndex = index
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if checkpointIndex < 0 || turnEndedIndex <= checkpointIndex || endIndex <= turnEndedIndex {
|
||||||
|
t.Fatalf("terminal order checkpoint=%d turn_ended=%d end=%d", checkpointIndex, turnEndedIndex, endIndex)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCancellationDiscardsUnpublishedCheckpointAndIgnoresLateAcknowledgements(t *testing.T) {
|
||||||
service, stream, projection := testCheckpointBlobProjection(t)
|
service, stream, projection := testCheckpointBlobProjection(t)
|
||||||
if err := service.queueCheckpointProjection(stream, projection, nil); err != nil {
|
if err := service.queueCheckpointProjection(stream, projection, nil); err != nil {
|
||||||
t.Fatalf("queueCheckpointProjection() error = %v", err)
|
t.Fatalf("queueCheckpointProjection() error = %v", err)
|
||||||
@@ -110,8 +247,8 @@ func TestCancellationKeepsPublishedCheckpointAndIgnoresLateAcknowledgements(t *t
|
|||||||
checkpointBeforeCancel++
|
checkpointBeforeCancel++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if checkpointBeforeCancel != 1 {
|
if checkpointBeforeCancel != 0 {
|
||||||
t.Fatalf("checkpoints before cancel = %d, want 1", checkpointBeforeCancel)
|
t.Fatalf("checkpoints before cancel = %d, want 0", checkpointBeforeCancel)
|
||||||
}
|
}
|
||||||
stream.mu.Lock()
|
stream.mu.Lock()
|
||||||
requestIDs := make([]uint32, 0, len(stream.PendingCheckpointBlobWrites))
|
requestIDs := make([]uint32, 0, len(stream.PendingCheckpointBlobWrites))
|
||||||
@@ -149,7 +286,7 @@ func TestCancellationKeepsPublishedCheckpointAndIgnoresLateAcknowledgements(t *t
|
|||||||
stream.mu.Lock()
|
stream.mu.Lock()
|
||||||
pending := stream.PendingCheckpoint
|
pending := stream.PendingCheckpoint
|
||||||
stream.mu.Unlock()
|
stream.mu.Unlock()
|
||||||
if checkpointCount != 1 || !canceledEnd || pending != nil {
|
if checkpointCount != 0 || !canceledEnd || pending != nil {
|
||||||
t.Fatalf("cancel events checkpoints=%d canceled_end=%v pending=%v", checkpointCount, canceledEnd, pending != nil)
|
t.Fatalf("cancel events checkpoints=%d canceled_end=%v pending=%v", checkpointCount, canceledEnd, pending != nil)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -234,7 +234,7 @@ func (service *Service) buildLegacyCompactionPlan(base *compactionPlan, conversa
|
|||||||
if conversation == nil || base == nil {
|
if conversation == nil || base == nil {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
candidates := buildContextCompactionCandidates(checkpointProjectionEntries(conversation.Entries), base.CurrentTurnSeq, base.CurrentRequestID)
|
candidates := buildContextCompactionCandidates(replayablePromptProjectionEntries(conversation.Entries), base.CurrentTurnSeq, base.CurrentRequestID)
|
||||||
if len(candidates) == 0 {
|
if len(candidates) == 0 {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
@@ -260,7 +260,7 @@ func (service *Service) buildAutoCompactionPlanFromHistory(base *compactionPlan,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
currentCandidate, hasCurrentCandidate := buildCurrentTurnCompactionCandidate(checkpointProjectionEntries(conversation.Entries), base.CurrentTurnSeq, base.CurrentRequestID)
|
currentCandidate, hasCurrentCandidate := buildCurrentTurnCompactionCandidate(replayablePromptProjectionEntries(conversation.Entries), base.CurrentTurnSeq, base.CurrentRequestID)
|
||||||
if !hasCurrentCandidate {
|
if !hasCurrentCandidate {
|
||||||
return legacyPlan, nil
|
return legacyPlan, nil
|
||||||
}
|
}
|
||||||
@@ -447,16 +447,8 @@ func (service *Service) handleCompactionEvent(stream *ActiveStream, payload *str
|
|||||||
if err := service.completeManualCompactionTurn(stream); err != nil {
|
if err := service.completeManualCompactionTurn(stream); err != nil {
|
||||||
return service.failStream(stream, "unknown", err)
|
return service.failStream(stream, "unknown", err)
|
||||||
}
|
}
|
||||||
if err := service.broker.Publish(stream.RequestID, StreamEvent{
|
completion := manualCompactionTurnCompletion(stream)
|
||||||
Message: buildTurnEndedMessage(0, 0, 0, 0),
|
return service.publishCheckpointWithCompletion(stream.RequestID, stream.ConversationID, &completion)
|
||||||
}); err != nil {
|
|
||||||
return service.failStream(stream, "unknown", err)
|
|
||||||
}
|
|
||||||
if err := service.broker.Complete(stream.RequestID, "", ""); err != nil {
|
|
||||||
return service.failStream(stream, "unknown", err)
|
|
||||||
}
|
|
||||||
service.setTurnPhase(stream, TurnPhaseCompleted)
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
return service.requestProviderAction(stream, providerActionResume)
|
return service.requestProviderAction(stream, providerActionResume)
|
||||||
}
|
}
|
||||||
@@ -500,12 +492,8 @@ func (service *Service) finishManualCompactionNoop(stream *ActiveStream) error {
|
|||||||
if err := service.completeManualCompactionTurn(stream); err != nil {
|
if err := service.completeManualCompactionTurn(stream); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := service.broker.Publish(stream.RequestID, StreamEvent{
|
completion := manualCompactionTurnCompletion(stream)
|
||||||
Message: buildTurnEndedMessage(0, 0, 0, 0),
|
return service.publishCheckpointWithCompletion(stream.RequestID, stream.ConversationID, &completion)
|
||||||
}); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return service.broker.Complete(stream.RequestID, "", "")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (service *Service) completeManualCompactionTurn(stream *ActiveStream) error {
|
func (service *Service) completeManualCompactionTurn(stream *ActiveStream) error {
|
||||||
@@ -530,10 +518,21 @@ func (service *Service) completeManualCompactionTurn(stream *ActiveStream) error
|
|||||||
if err := service.syncSummaryCarryForward(conversationID, requestID, modelCallID); err != nil {
|
if err := service.syncSummaryCarryForward(conversationID, requestID, modelCallID); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
service.setTurnPhase(stream, TurnPhaseCompleted)
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func manualCompactionTurnCompletion(stream *ActiveStream) pendingTurnCompletion {
|
||||||
|
if stream == nil {
|
||||||
|
return pendingTurnCompletion{}
|
||||||
|
}
|
||||||
|
return pendingTurnCompletion{
|
||||||
|
ConversationID: strings.TrimSpace(stream.ConversationID),
|
||||||
|
RequestID: strings.TrimSpace(stream.RequestID),
|
||||||
|
TurnSeq: stream.TurnSeq,
|
||||||
|
ModelCallID: "turn:" + strings.TrimSpace(stream.RequestID),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (service *Service) publishSummaryCompleted(stream *ActiveStream, hookMessage string) error {
|
func (service *Service) publishSummaryCompleted(stream *ActiveStream, hookMessage string) error {
|
||||||
if service == nil || stream == nil {
|
if service == nil || stream == nil {
|
||||||
return nil
|
return nil
|
||||||
@@ -568,6 +567,7 @@ func (service *Service) applyCompactionPlan(stream *ActiveStream, conversationID
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
originalEntryCount := len(candidateConversation.Entries)
|
||||||
if err := applyCompactionToConversation(candidateConversation, plan, summaryText); err != nil {
|
if err := applyCompactionToConversation(candidateConversation, plan, summaryText); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -582,9 +582,9 @@ func (service *Service) applyCompactionPlan(stream *ActiveStream, conversationID
|
|||||||
if validationErr := validateCompactionCandidateBudget(recompiled, plan); validationErr != nil {
|
if validationErr := validateCompactionCandidateBudget(recompiled, plan); validationErr != nil {
|
||||||
return validationErr
|
return validationErr
|
||||||
}
|
}
|
||||||
replacementEntries := append([]HistoryEntry(nil), candidateConversation.Entries...)
|
compactionEntries := append([]HistoryEntry(nil), candidateConversation.Entries[originalEntryCount:]...)
|
||||||
if service.store != nil {
|
if service.store != nil {
|
||||||
persisted, err := service.store.ReplaceEntries(conversationID, replacementEntries, func(item *ConversationFile) error {
|
persisted, _, err := service.store.AppendEntriesWithUpdate(conversationID, resetEntrySequences(compactionEntries), func(item *ConversationFile) error {
|
||||||
if item == nil {
|
if item == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -605,10 +605,7 @@ func (service *Service) applyCompactionPlan(stream *ActiveStream, conversationID
|
|||||||
if item == nil {
|
if item == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
item.Entries = nil
|
appendEntriesInPlace(item, resetEntrySequences(compactionEntries))
|
||||||
item.NextEntrySeq = 1
|
|
||||||
item.NextTurnSeq = 1
|
|
||||||
appendEntriesInPlace(item, resetEntrySequences(replacementEntries))
|
|
||||||
item.TokenDetailsUsedTokens = 0
|
item.TokenDetailsUsedTokens = 0
|
||||||
clearConversationAutoCompactionState(item)
|
clearConversationAutoCompactionState(item)
|
||||||
return nil
|
return nil
|
||||||
@@ -643,14 +640,13 @@ func applyCompactionToConversation(conversation *ConversationFile, plan *Pending
|
|||||||
if conversation == nil || plan == nil {
|
if conversation == nil || plan == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
replacementEntries, err := buildCompactedContextEntries(conversation, plan, summaryText)
|
compactionEntries, err := buildCompactedContextEntries(conversation, plan, summaryText)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
conversation.Entries = nil
|
// Canonical history stays append-only. The prompt projector applies the
|
||||||
conversation.NextEntrySeq = 1
|
// latest summary marker when constructing model-visible replay.
|
||||||
conversation.NextTurnSeq = 1
|
appendEntriesInPlace(conversation, resetEntrySequences(compactionEntries))
|
||||||
appendEntriesInPlace(conversation, resetEntrySequences(replacementEntries))
|
|
||||||
conversation.TokenDetailsUsedTokens = 0
|
conversation.TokenDetailsUsedTokens = 0
|
||||||
clearConversationAutoCompactionState(conversation)
|
clearConversationAutoCompactionState(conversation)
|
||||||
if conversation.TokenDetailsMaxTokens == 0 {
|
if conversation.TokenDetailsMaxTokens == 0 {
|
||||||
@@ -671,39 +667,8 @@ func buildCompactedContextEntries(conversation *ConversationFile, plan *PendingC
|
|||||||
if ok {
|
if ok {
|
||||||
entries = append(entries, runtimeEntry)
|
entries = append(entries, runtimeEntry)
|
||||||
}
|
}
|
||||||
if conversation == nil || !plan.PreserveCurrentTurnInputs {
|
|
||||||
return entries, nil
|
return entries, nil
|
||||||
}
|
}
|
||||||
entries = append(entries, buildAutoCompactionPreservedCurrentTurnEntries(conversation.Entries, plan)...)
|
|
||||||
return entries, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func buildAutoCompactionPreservedCurrentTurnEntries(entries []HistoryEntry, plan *PendingCompaction) []HistoryEntry {
|
|
||||||
if len(entries) == 0 || plan == nil || !plan.PreserveCurrentTurnInputs {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
latestToolCallID := latestCompletedToolCallIDForTurn(entries, plan.CurrentTurnSeq, plan.CurrentRequestID)
|
|
||||||
preservedIndexes := autoCompactionPreservedEntryIndexes(entries, plan.CurrentTurnSeq, plan.CurrentRequestID, latestToolCallID)
|
|
||||||
if len(preservedIndexes) == 0 {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
preserved := make([]HistoryEntry, 0, len(preservedIndexes))
|
|
||||||
for index, entry := range entries {
|
|
||||||
if _, ok := preservedIndexes[index]; !ok {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
switch strings.TrimSpace(entry.Kind) {
|
|
||||||
case "compaction_summary", "compacted_summary", "compaction_request":
|
|
||||||
continue
|
|
||||||
case "tool_result":
|
|
||||||
if rewritten, ok := rewriteAutoCompactionToolResultEntry(entry, autoCompactionPreservedToolResultLimitBytes, false); ok {
|
|
||||||
entry = rewritten
|
|
||||||
}
|
|
||||||
}
|
|
||||||
preserved = append(preserved, entry)
|
|
||||||
}
|
|
||||||
return preserved
|
|
||||||
}
|
|
||||||
|
|
||||||
func newCompactionSummaryEntry(plan *PendingCompaction, summaryText string) HistoryEntry {
|
func newCompactionSummaryEntry(plan *PendingCompaction, summaryText string) HistoryEntry {
|
||||||
payload, _ := json.Marshal(compactionSummaryEntryPayload{
|
payload, _ := json.Marshal(compactionSummaryEntryPayload{
|
||||||
|
|||||||
@@ -0,0 +1,206 @@
|
|||||||
|
package forwarder
|
||||||
|
|
||||||
|
import (
|
||||||
|
"reflect"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"google.golang.org/protobuf/encoding/protojson"
|
||||||
|
|
||||||
|
"cursor/gen/agentv1"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestApplyCompactionToConversationPreservesCanonicalHistory(t *testing.T) {
|
||||||
|
conversation := compactionAppendOnlyConversation(t)
|
||||||
|
originalEntries := append([]HistoryEntry(nil), conversation.Entries...)
|
||||||
|
plan := &PendingCompaction{
|
||||||
|
Trigger: "manual",
|
||||||
|
CurrentTurnSeq: 2,
|
||||||
|
CurrentRequestID: "request-2",
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := applyCompactionToConversation(conversation, plan, "earlier context summary"); err != nil {
|
||||||
|
t.Fatalf("applyCompactionToConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(conversation.Entries) <= len(originalEntries) {
|
||||||
|
t.Fatalf("entries after compaction = %d, want the %d original entries plus a summary marker", len(conversation.Entries), len(originalEntries))
|
||||||
|
}
|
||||||
|
if !reflect.DeepEqual(conversation.Entries[:len(originalEntries)], originalEntries) {
|
||||||
|
t.Fatal("compaction changed the canonical history prefix")
|
||||||
|
}
|
||||||
|
|
||||||
|
projector := NewHistoryProjector()
|
||||||
|
projection, err := projector.ProjectCheckpointProjection(conversation)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ProjectCheckpointProjection() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(projection.State.GetTurns()) != 2 {
|
||||||
|
t.Fatalf("checkpoint turns after compaction = %d, want 2 visible turns", len(projection.State.GetTurns()))
|
||||||
|
}
|
||||||
|
replay, err := projector.ProjectPromptReplay(conversation)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ProjectPromptReplay() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(replay) != 1 || replay[0].Role != "user" || !strings.Contains(replay[0].Content, "earlier context summary") {
|
||||||
|
t.Fatalf("prompt replay after compaction = %#v, want only the compacted summary", replay)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCompactedPromptProjectionPlacesSummaryBeforePreservedCurrentTurn(t *testing.T) {
|
||||||
|
conversation := &ConversationFile{
|
||||||
|
ConversationID: "conversation-1",
|
||||||
|
RootConversationID: "conversation-1",
|
||||||
|
Mode: "agent",
|
||||||
|
NextTurnSeq: 1,
|
||||||
|
NextEntrySeq: 1,
|
||||||
|
}
|
||||||
|
appendEntriesInPlace(conversation, []HistoryEntry{
|
||||||
|
compactionTestUserEntry(t, 1, "request-1", "current question", "message-1"),
|
||||||
|
newToolCallEntry(1, "request-1", "call-1", "Read", "", "", checkpointTestReadToolCall(t, nil)),
|
||||||
|
newToolResultEntry(1, "request-1", "call-1", "Read", `{"path":"/tmp/example.txt"}`, "file contents", "", checkpointTestReadToolCall(t, nil)),
|
||||||
|
})
|
||||||
|
plan := &PendingCompaction{
|
||||||
|
Trigger: "auto",
|
||||||
|
CurrentTurnSeq: 1,
|
||||||
|
CurrentRequestID: "request-1",
|
||||||
|
PreserveCurrentTurnInputs: true,
|
||||||
|
}
|
||||||
|
if err := applyCompactionToConversation(conversation, plan, "current progress summary"); err != nil {
|
||||||
|
t.Fatalf("applyCompactionToConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
projected := compactedPromptProjectionEntries(conversation.Entries)
|
||||||
|
promptKinds := make([]string, 0, len(projected))
|
||||||
|
for _, entry := range projected {
|
||||||
|
if isPromptReplayEntryKind(entry.Kind) {
|
||||||
|
promptKinds = append(promptKinds, entry.Kind)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
want := []string{"compacted_summary", "user_message", "tool_call", "tool_result"}
|
||||||
|
if !reflect.DeepEqual(promptKinds, want) {
|
||||||
|
t.Fatalf("compacted prompt entry order = %#v, want %#v", promptKinds, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCompactionPlanningDoesNotRecompactArchivedHistory(t *testing.T) {
|
||||||
|
conversation := compactionAppendOnlyConversation(t)
|
||||||
|
if err := applyCompactionToConversation(conversation, &PendingCompaction{
|
||||||
|
Trigger: "manual",
|
||||||
|
CurrentTurnSeq: 2,
|
||||||
|
CurrentRequestID: "request-2",
|
||||||
|
}, "archived history summary"); err != nil {
|
||||||
|
t.Fatalf("applyCompactionToConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
appendEntriesInPlace(conversation, []HistoryEntry{
|
||||||
|
compactionTestUserEntry(t, 3, "request-3", "new question", "message-3"),
|
||||||
|
})
|
||||||
|
|
||||||
|
plan, err := (&Service{}).buildLegacyCompactionPlan(&compactionPlan{
|
||||||
|
CurrentTurnSeq: 3,
|
||||||
|
CurrentRequestID: "request-3",
|
||||||
|
}, conversation, false, 0)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("buildLegacyCompactionPlan() error = %v", err)
|
||||||
|
}
|
||||||
|
if plan != nil {
|
||||||
|
t.Fatalf("buildLegacyCompactionPlan() = %#v, want no already summarized candidates", plan)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestApplyCompactionPlanPersistsHistoryAppendOnly(t *testing.T) {
|
||||||
|
store := NewConversationFileStore(t.TempDir())
|
||||||
|
conversation := compactionAppendOnlyConversation(t)
|
||||||
|
if _, _, err := store.AppendEntries(conversation.ConversationID, resetEntrySequences(conversation.Entries)); err != nil {
|
||||||
|
t.Fatalf("AppendEntries() error = %v", err)
|
||||||
|
}
|
||||||
|
persisted, err := store.LoadConversation(conversation.ConversationID)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("initial LoadConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
originalEntries := append([]HistoryEntry(nil), persisted.Entries...)
|
||||||
|
projector := NewHistoryProjector()
|
||||||
|
service := &Service{
|
||||||
|
store: store,
|
||||||
|
projector: projector,
|
||||||
|
compiler: compactionProjectionCompiler{projector: projector},
|
||||||
|
}
|
||||||
|
stream := &ActiveStream{
|
||||||
|
RequestID: "request-2",
|
||||||
|
ConversationID: conversation.ConversationID,
|
||||||
|
TurnSeq: 2,
|
||||||
|
Mode: agentv1.AgentMode_AGENT_MODE_AGENT,
|
||||||
|
CheckpointConversation: persisted,
|
||||||
|
}
|
||||||
|
plan := &PendingCompaction{
|
||||||
|
Trigger: "manual",
|
||||||
|
CurrentTurnSeq: 2,
|
||||||
|
CurrentRequestID: "request-2",
|
||||||
|
ContextWindowSize: 1_000_000,
|
||||||
|
}
|
||||||
|
if err := service.applyCompactionPlan(stream, conversation.ConversationID, plan, "persisted summary"); err != nil {
|
||||||
|
t.Fatalf("applyCompactionPlan() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
loaded, err := store.LoadConversation(conversation.ConversationID)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("LoadConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(loaded.Entries) <= len(originalEntries) {
|
||||||
|
t.Fatalf("persisted entries after compaction = %d, want more than %d", len(loaded.Entries), len(originalEntries))
|
||||||
|
}
|
||||||
|
for index := range originalEntries {
|
||||||
|
if !reflect.DeepEqual(loaded.Entries[index], originalEntries[index]) {
|
||||||
|
t.Fatalf("persisted history entry %d changed after compaction:\ngot %#v\nwant %#v", index, loaded.Entries[index], originalEntries[index])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type compactionProjectionCompiler struct {
|
||||||
|
projector *HistoryProjector
|
||||||
|
}
|
||||||
|
|
||||||
|
func (compiler compactionProjectionCompiler) Compile(conversation *ConversationFile, _ agentv1.AgentMode, _ string, _ string) (CompiledConversation, error) {
|
||||||
|
messages, err := compiler.projector.ProjectPromptReplay(conversation)
|
||||||
|
return CompiledConversation{Messages: messages}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (compactionProjectionCompiler) DerivePromptContexts(*ConversationFile, agentv1.AgentMode, string) ([]PromptContextMessage, error) {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func compactionAppendOnlyConversation(t *testing.T) *ConversationFile {
|
||||||
|
t.Helper()
|
||||||
|
conversation := &ConversationFile{
|
||||||
|
ConversationID: "conversation-1",
|
||||||
|
RootConversationID: "conversation-1",
|
||||||
|
Mode: "agent",
|
||||||
|
NextTurnSeq: 1,
|
||||||
|
NextEntrySeq: 1,
|
||||||
|
TokenDetailsUsedTokens: 42_000,
|
||||||
|
TokenDetailsMaxTokens: 50_000,
|
||||||
|
}
|
||||||
|
appendEntriesInPlace(conversation, []HistoryEntry{
|
||||||
|
compactionTestUserEntry(t, 1, "request-1", "first question", "message-1"),
|
||||||
|
newAssistantTextEntry(1, "request-1", "first answer", "", ""),
|
||||||
|
compactionTestUserEntry(t, 2, "request-2", "second question", "message-2"),
|
||||||
|
newAssistantTextEntry(2, "request-2", "second answer", "", ""),
|
||||||
|
})
|
||||||
|
return conversation
|
||||||
|
}
|
||||||
|
|
||||||
|
func compactionTestUserEntry(t *testing.T, turnSeq int64, requestID string, text string, messageID string) HistoryEntry {
|
||||||
|
t.Helper()
|
||||||
|
payload, err := protojson.Marshal(&agentv1.UserMessage{Text: text, MessageId: messageID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal user message: %v", err)
|
||||||
|
}
|
||||||
|
return HistoryEntry{
|
||||||
|
TurnSeq: turnSeq,
|
||||||
|
RequestID: requestID,
|
||||||
|
Role: "user",
|
||||||
|
Kind: "user_message",
|
||||||
|
Payload: payload,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var _ PromptCompiler = compactionProjectionCompiler{}
|
||||||
@@ -121,10 +121,15 @@ func (store *ConversationFileStore) LoadConversation(conversationID string) (*Co
|
|||||||
|
|
||||||
// AppendEntries 把已经发生的语义事件追加到 context.json，并同步 state.json。
|
// AppendEntries 把已经发生的语义事件追加到 context.json，并同步 state.json。
|
||||||
func (store *ConversationFileStore) AppendEntries(conversationID string, entries []HistoryEntry) (*ConversationFile, []HistoryEntry, error) {
|
func (store *ConversationFileStore) AppendEntries(conversationID string, entries []HistoryEntry) (*ConversationFile, []HistoryEntry, error) {
|
||||||
|
return store.AppendEntriesWithUpdate(conversationID, entries, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
// AppendEntriesWithUpdate 原子追加 context entries，并在同一把会话锁内更新 state metadata。
|
||||||
|
func (store *ConversationFileStore) AppendEntriesWithUpdate(conversationID string, entries []HistoryEntry, update func(*ConversationFile) error) (*ConversationFile, []HistoryEntry, error) {
|
||||||
if store == nil {
|
if store == nil {
|
||||||
return nil, nil, fmt.Errorf("conversation file store is nil")
|
return nil, nil, fmt.Errorf("conversation file store is nil")
|
||||||
}
|
}
|
||||||
if len(entries) == 0 {
|
if len(entries) == 0 && update == nil {
|
||||||
conversation, err := store.LoadConversation(conversationID)
|
conversation, err := store.LoadConversation(conversationID)
|
||||||
return conversation, nil, err
|
return conversation, nil, err
|
||||||
}
|
}
|
||||||
@@ -162,6 +167,11 @@ func (store *ConversationFileStore) AppendEntries(conversationID string, entries
|
|||||||
conversation.Mode = alias
|
conversation.Mode = alias
|
||||||
}
|
}
|
||||||
assigned := appendEntriesInPlace(conversation, entries)
|
assigned := appendEntriesInPlace(conversation, entries)
|
||||||
|
if update != nil {
|
||||||
|
if err := update(conversation); err != nil {
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
deriveConversationLoopState(conversation)
|
deriveConversationLoopState(conversation)
|
||||||
if err := store.writeConversationLocked(normalizedConversationID, conversation); err != nil {
|
if err := store.writeConversationLocked(normalizedConversationID, conversation); err != nil {
|
||||||
return nil, nil, err
|
return nil, nil, err
|
||||||
@@ -762,6 +772,7 @@ func mergeConversationMetadata(target *ConversationFile, source *ConversationFil
|
|||||||
target.CurrentPlanText = source.CurrentPlanText
|
target.CurrentPlanText = source.CurrentPlanText
|
||||||
target.CurrentPlans = clonePlanRegistryEntries(source.CurrentPlans)
|
target.CurrentPlans = clonePlanRegistryEntries(source.CurrentPlans)
|
||||||
target.CurrentTodos = cloneTodoItems(source.CurrentTodos)
|
target.CurrentTodos = cloneTodoItems(source.CurrentTodos)
|
||||||
|
target.ImportedTurnIDs = cloneByteSlices(source.ImportedTurnIDs)
|
||||||
target.LatestRequestPrefix = cloneConversationRequestPrefix(source.LatestRequestPrefix)
|
target.LatestRequestPrefix = cloneConversationRequestPrefix(source.LatestRequestPrefix)
|
||||||
target.LastProviderCall = cloneConversationProviderCall(source.LastProviderCall)
|
target.LastProviderCall = cloneConversationProviderCall(source.LastProviderCall)
|
||||||
if !source.CreatedAt.IsZero() && (target.CreatedAt.IsZero() || source.CreatedAt.Before(target.CreatedAt)) {
|
if !source.CreatedAt.IsZero() && (target.CreatedAt.IsZero() || source.CreatedAt.Before(target.CreatedAt)) {
|
||||||
@@ -894,6 +905,7 @@ func cloneConversationFile(conversation *ConversationFile) *ConversationFile {
|
|||||||
cloned := *conversation
|
cloned := *conversation
|
||||||
cloned.CurrentPlans = clonePlanRegistryEntries(conversation.CurrentPlans)
|
cloned.CurrentPlans = clonePlanRegistryEntries(conversation.CurrentPlans)
|
||||||
cloned.CurrentTodos = cloneTodoItems(conversation.CurrentTodos)
|
cloned.CurrentTodos = cloneTodoItems(conversation.CurrentTodos)
|
||||||
|
cloned.ImportedTurnIDs = cloneByteSlices(conversation.ImportedTurnIDs)
|
||||||
cloned.LatestRequestPrefix = cloneConversationRequestPrefix(conversation.LatestRequestPrefix)
|
cloned.LatestRequestPrefix = cloneConversationRequestPrefix(conversation.LatestRequestPrefix)
|
||||||
cloned.LastProviderCall = cloneConversationProviderCall(conversation.LastProviderCall)
|
cloned.LastProviderCall = cloneConversationProviderCall(conversation.LastProviderCall)
|
||||||
cloned.Entries = append([]HistoryEntry(nil), conversation.Entries...)
|
cloned.Entries = append([]HistoryEntry(nil), conversation.Entries...)
|
||||||
|
|||||||
@@ -0,0 +1,167 @@
|
|||||||
|
package forwarder
|
||||||
|
|
||||||
|
import (
|
||||||
|
"crypto/sha256"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"google.golang.org/protobuf/proto"
|
||||||
|
|
||||||
|
"cursor/gen/agentv1"
|
||||||
|
modeladapter "cursor/internal/backend/agent/model"
|
||||||
|
promptengine "cursor/internal/backend/agent/prompt"
|
||||||
|
)
|
||||||
|
|
||||||
|
type importedBlobStore map[string][]byte
|
||||||
|
|
||||||
|
func newImportedBlobStore(items []*agentv1.PreFetchedBlob) (importedBlobStore, error) {
|
||||||
|
if len(items) == 0 {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
store := make(importedBlobStore, len(items))
|
||||||
|
for _, item := range items {
|
||||||
|
if item == nil || len(item.GetId()) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if len(item.GetId()) != sha256.Size {
|
||||||
|
return nil, fmt.Errorf("prefetched blob id length %d, want %d", len(item.GetId()), sha256.Size)
|
||||||
|
}
|
||||||
|
digest := sha256.Sum256(item.GetValue())
|
||||||
|
if string(digest[:]) != string(item.GetId()) {
|
||||||
|
return nil, fmt.Errorf("prefetched blob %x failed SHA-256 validation", item.GetId())
|
||||||
|
}
|
||||||
|
store[string(item.GetId())] = append([]byte(nil), item.GetValue()...)
|
||||||
|
}
|
||||||
|
return store, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (store importedBlobStore) resolve(id []byte) ([]byte, bool) {
|
||||||
|
if len(id) == 0 || len(store) == 0 {
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
|
value, ok := store[string(id)]
|
||||||
|
return append([]byte(nil), value...), ok
|
||||||
|
}
|
||||||
|
|
||||||
|
func decodeImportedTurn(raw []byte, blobs importedBlobStore) (*agentv1.ConversationTurnStructure, []byte, error) {
|
||||||
|
if data, ok := blobs.resolve(raw); ok {
|
||||||
|
turn := &agentv1.ConversationTurnStructure{}
|
||||||
|
if err := proto.Unmarshal(data, turn); err != nil || turn.GetTurn() == nil {
|
||||||
|
return nil, nil, fmt.Errorf("decode imported turn blob %x: %w", raw, firstNonNilError(err, fmt.Errorf("turn payload is empty")))
|
||||||
|
}
|
||||||
|
return turn, append([]byte(nil), raw...), nil
|
||||||
|
}
|
||||||
|
turn := &agentv1.ConversationTurnStructure{}
|
||||||
|
if err := proto.Unmarshal(raw, turn); err == nil && turn.GetTurn() != nil {
|
||||||
|
return turn, nil, nil
|
||||||
|
}
|
||||||
|
if len(raw) == sha256.Size {
|
||||||
|
return nil, append([]byte(nil), raw...), nil
|
||||||
|
}
|
||||||
|
return nil, nil, fmt.Errorf("decode imported inline turn")
|
||||||
|
}
|
||||||
|
|
||||||
|
func decodeImportedUserMessage(raw []byte, blobs importedBlobStore) (*agentv1.UserMessage, error) {
|
||||||
|
data := raw
|
||||||
|
if resolved, ok := blobs.resolve(raw); ok {
|
||||||
|
data = resolved
|
||||||
|
} else if len(raw) == sha256.Size {
|
||||||
|
candidate := &agentv1.UserMessage{}
|
||||||
|
if err := proto.Unmarshal(raw, candidate); err != nil || !hasKnownUserMessageContent(candidate) {
|
||||||
|
return nil, fmt.Errorf("missing prefetched user message blob %x", raw)
|
||||||
|
}
|
||||||
|
return candidate, nil
|
||||||
|
}
|
||||||
|
message := &agentv1.UserMessage{}
|
||||||
|
if err := proto.Unmarshal(data, message); err != nil {
|
||||||
|
return nil, fmt.Errorf("decode imported turn user_message: %w", err)
|
||||||
|
}
|
||||||
|
return message, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func decodeImportedStep(raw []byte, blobs importedBlobStore) (*agentv1.ConversationStep, error) {
|
||||||
|
data := raw
|
||||||
|
if resolved, ok := blobs.resolve(raw); ok {
|
||||||
|
data = resolved
|
||||||
|
} else if len(raw) == sha256.Size {
|
||||||
|
candidate := &agentv1.ConversationStep{}
|
||||||
|
if err := proto.Unmarshal(raw, candidate); err != nil || candidate.GetMessage() == nil {
|
||||||
|
return nil, fmt.Errorf("missing prefetched conversation step blob %x", raw)
|
||||||
|
}
|
||||||
|
return candidate, nil
|
||||||
|
}
|
||||||
|
step := &agentv1.ConversationStep{}
|
||||||
|
if err := proto.Unmarshal(data, step); err != nil {
|
||||||
|
return nil, fmt.Errorf("decode imported turn step: %w", err)
|
||||||
|
}
|
||||||
|
if step.GetMessage() == nil {
|
||||||
|
return nil, fmt.Errorf("decode imported turn step: payload is empty")
|
||||||
|
}
|
||||||
|
return step, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func importedBlobTurnMessages(turn *agentv1.ConversationTurnStructure, blobs importedBlobStore) ([]modeladapter.Message, error) {
|
||||||
|
if turn == nil || turn.GetAgentConversationTurn() == nil {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
agentTurn := turn.GetAgentConversationTurn()
|
||||||
|
messages := make([]modeladapter.Message, 0, 1+len(agentTurn.GetSteps()))
|
||||||
|
if len(agentTurn.GetUserMessage()) > 0 {
|
||||||
|
userMessage, err := decodeImportedUserMessage(agentTurn.GetUserMessage(), blobs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if replay, ok := promptengine.BuildUserMessageReplayMessage(userMessage); ok {
|
||||||
|
messages = append(messages, toModelMessage(replay))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, rawStep := range agentTurn.GetSteps() {
|
||||||
|
if len(rawStep) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
step, err := decodeImportedStep(rawStep, blobs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
for _, replay := range promptengine.BuildLegacyMessagesFromConversationStep(step) {
|
||||||
|
messages = append(messages, toModelMessage(replay))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return messages, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func importedTurnIDs(turns [][]byte, blobs importedBlobStore) ([][]byte, error) {
|
||||||
|
ids := make([][]byte, 0, len(turns))
|
||||||
|
for _, raw := range turns {
|
||||||
|
if len(raw) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
_, id, err := decodeImportedTurn(raw, blobs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if len(id) > 0 {
|
||||||
|
ids = append(ids, id)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return ids, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func hasKnownUserMessageContent(message *agentv1.UserMessage) bool {
|
||||||
|
if message == nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return message.GetText() != "" ||
|
||||||
|
message.GetMessageId() != "" ||
|
||||||
|
message.GetSelectedContext() != nil ||
|
||||||
|
message.GetRichText() != "" ||
|
||||||
|
len(message.GetConversationStateBlobId()) > 0 ||
|
||||||
|
len(message.GetTextBlobId()) > 0 ||
|
||||||
|
len(message.GetRichTextBlobId()) > 0
|
||||||
|
}
|
||||||
|
|
||||||
|
func firstNonNilError(err error, fallback error) error {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return fallback
|
||||||
|
}
|
||||||
@@ -0,0 +1,109 @@
|
|||||||
|
package forwarder
|
||||||
|
|
||||||
|
import (
|
||||||
|
"crypto/sha256"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"google.golang.org/protobuf/proto"
|
||||||
|
|
||||||
|
"cursor/gen/agentv1"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestImportedConversationStateRestoresBlobOnlyForkAndCheckpointPrefix(t *testing.T) {
|
||||||
|
parent := compactionAppendOnlyConversation(t)
|
||||||
|
parent.Entries = parent.Entries[:2]
|
||||||
|
parent.NextEntrySeq = 3
|
||||||
|
parent.NextTurnSeq = 2
|
||||||
|
projection, err := NewHistoryProjector().ProjectCheckpointProjection(parent)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ProjectCheckpointProjection() error = %v", err)
|
||||||
|
}
|
||||||
|
prefetched := make([]*agentv1.PreFetchedBlob, 0, len(projection.Blobs))
|
||||||
|
for _, blob := range projection.Blobs {
|
||||||
|
prefetched = append(prefetched, &agentv1.PreFetchedBlob{Id: blob.ID, Value: blob.Data})
|
||||||
|
}
|
||||||
|
state := proto.Clone(projection.State).(*agentv1.ConversationStateStructure)
|
||||||
|
state.RootPromptMessagesJson = nil
|
||||||
|
conversation, err := newRuntimeConversation("fork-conversation", agentv1.AgentMode_AGENT_MODE_AGENT)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("newRuntimeConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
entries, err := (&Service{}).importConversationState(conversation, state, prefetched)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("importConversationState() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(conversation.ImportedTurnIDs) != 1 || conversation.NextTurnSeq != 2 {
|
||||||
|
t.Fatalf("imported prefix turns=%d next_turn_seq=%d, want 1 and 2", len(conversation.ImportedTurnIDs), conversation.NextTurnSeq)
|
||||||
|
}
|
||||||
|
if len(entries) != 2 {
|
||||||
|
t.Fatalf("imported model entries = %d, want parent user and assistant", len(entries))
|
||||||
|
}
|
||||||
|
appendEntriesInPlace(conversation, append(entries,
|
||||||
|
compactionTestUserEntry(t, 2, "request-2", "fork question", "message-2"),
|
||||||
|
))
|
||||||
|
forkProjection, err := NewHistoryProjector().ProjectCheckpointProjection(conversation)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("fork ProjectCheckpointProjection() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(forkProjection.State.GetTurns()) != 2 {
|
||||||
|
t.Fatalf("fork checkpoint turns = %d, want imported parent plus local fork turn", len(forkProjection.State.GetTurns()))
|
||||||
|
}
|
||||||
|
if string(forkProjection.State.GetTurns()[0]) != string(projection.State.GetTurns()[0]) {
|
||||||
|
t.Fatal("fork checkpoint did not preserve the imported parent turn ID as its prefix")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestImportedConversationStateRejectsUnresolvedBlobTurn(t *testing.T) {
|
||||||
|
turnID := sha256.Sum256([]byte("missing imported turn"))
|
||||||
|
conversation, err := newRuntimeConversation("fork-conversation", agentv1.AgentMode_AGENT_MODE_AGENT)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("newRuntimeConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
if _, err := (&Service{}).importConversationState(conversation, &agentv1.ConversationStateStructure{
|
||||||
|
Turns: [][]byte{turnID[:]},
|
||||||
|
}, nil); err == nil {
|
||||||
|
t.Fatal("importConversationState() accepted an unresolved Blob turn")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestImportedTurnIDsPersistThroughConversationStore(t *testing.T) {
|
||||||
|
store := NewConversationFileStore(t.TempDir())
|
||||||
|
turnID := sha256.Sum256([]byte("parent turn"))
|
||||||
|
conversation, err := newRuntimeConversation("fork-conversation", agentv1.AgentMode_AGENT_MODE_AGENT)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("newRuntimeConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
conversation.ImportedTurnIDs = [][]byte{turnID[:]}
|
||||||
|
persisted, err := store.SaveConversationWithEntries(conversation.ConversationID, conversation, []HistoryEntry{
|
||||||
|
compactionTestUserEntry(t, 2, "request-2", "fork question", "message-2"),
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("SaveConversationWithEntries() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(persisted.ImportedTurnIDs) != 1 || string(persisted.ImportedTurnIDs[0]) != string(turnID[:]) {
|
||||||
|
t.Fatalf("persisted ImportedTurnIDs = %x, want %x", persisted.ImportedTurnIDs, turnID)
|
||||||
|
}
|
||||||
|
loaded, err := store.LoadConversation(conversation.ConversationID)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("LoadConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(loaded.ImportedTurnIDs) != 1 || string(loaded.ImportedTurnIDs[0]) != string(turnID[:]) {
|
||||||
|
t.Fatalf("loaded ImportedTurnIDs = %x, want %x", loaded.ImportedTurnIDs, turnID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRewindImportedTurnPrefixUsesClientForkPoint(t *testing.T) {
|
||||||
|
ids := make([][]byte, 3)
|
||||||
|
for index := range ids {
|
||||||
|
digest := sha256.Sum256([]byte{byte(index + 1)})
|
||||||
|
ids[index] = digest[:]
|
||||||
|
}
|
||||||
|
trimmed := rewindImportedTurnPrefix(ids, runRewindDecision{
|
||||||
|
TargetTurnSeq: 4,
|
||||||
|
HasClientTurnCount: true,
|
||||||
|
ClientTurnCount: 1,
|
||||||
|
})
|
||||||
|
if len(trimmed) != 1 || string(trimmed[0]) != string(ids[0]) {
|
||||||
|
t.Fatalf("rewindImportedTurnPrefix() = %x, want first imported turn only", trimmed)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -2,6 +2,7 @@ package forwarder
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
@@ -125,6 +126,53 @@ func TestCancelPersistsInterruptedProviderOutputIdempotently(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestGenericProviderFailurePersistsAccumulatedOutput(t *testing.T) {
|
||||||
|
service, stream, _ := testCheckpointBlobProjection(t)
|
||||||
|
conversation, _, _, err := service.snapshotCheckpointConversation(stream)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("snapshotCheckpointConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
if _, err := service.store.SaveConversationWithEntries(stream.ConversationID, conversation, conversation.Entries); err != nil {
|
||||||
|
t.Fatalf("SaveConversationWithEntries() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
stream.mu.Lock()
|
||||||
|
stream.CurrentModelCallID = "model-call-1"
|
||||||
|
stream.ProviderActive = true
|
||||||
|
stream.ProviderAccumulatedText = "partial answer before transport failure"
|
||||||
|
stream.Status = StreamStatusStreaming
|
||||||
|
stream.Phase = TurnPhaseProviderRunning
|
||||||
|
stream.mu.Unlock()
|
||||||
|
if err := service.handleProviderDoneEvent(stream, &streamProviderEvent{
|
||||||
|
Done: true,
|
||||||
|
Err: errors.New("transport failed"),
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("handleProviderDoneEvent() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
persisted, err := service.store.LoadConversation(stream.ConversationID)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("LoadConversation() error = %v", err)
|
||||||
|
}
|
||||||
|
foundPartialOutput := false
|
||||||
|
for _, entry := range persisted.Entries {
|
||||||
|
if entry.Kind != "assistant_text" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
var payload assistantTextPayload
|
||||||
|
if err := json.Unmarshal(entry.Payload, &payload); err != nil {
|
||||||
|
t.Fatalf("decode assistant entry: %v", err)
|
||||||
|
}
|
||||||
|
if payload.Text == "partial answer before transport failure" {
|
||||||
|
foundPartialOutput = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !foundPartialOutput {
|
||||||
|
t.Fatal("generic provider failure discarded accumulated assistant output")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestCancelPreservesPersistedTurnActivityWithoutLiveAccumulator(t *testing.T) {
|
func TestCancelPreservesPersistedTurnActivityWithoutLiveAccumulator(t *testing.T) {
|
||||||
service, stream, _ := testCheckpointBlobProjection(t)
|
service, stream, _ := testCheckpointBlobProjection(t)
|
||||||
conversation, _, _, err := service.snapshotCheckpointConversation(stream)
|
conversation, _, _, err := service.snapshotCheckpointConversation(stream)
|
||||||
|
|||||||
@@ -326,20 +326,24 @@ func compactedPromptProjectionEntries(entries []HistoryEntry) []HistoryEntry {
|
|||||||
latestToolCallID := latestCompletedToolCallIDForTurn(entries, compactionPayload.CurrentTurnSeq, compactionPayload.CurrentRequestID)
|
latestToolCallID := latestCompletedToolCallIDForTurn(entries, compactionPayload.CurrentTurnSeq, compactionPayload.CurrentRequestID)
|
||||||
preservedIndexes = autoCompactionPreservedEntryIndexes(entries, compactionPayload.CurrentTurnSeq, compactionPayload.CurrentRequestID, latestToolCallID)
|
preservedIndexes = autoCompactionPreservedEntryIndexes(entries, compactionPayload.CurrentTurnSeq, compactionPayload.CurrentRequestID, latestToolCallID)
|
||||||
}
|
}
|
||||||
filtered := make([]HistoryEntry, 0, len(entries)-compactionIndex)
|
filtered := make([]HistoryEntry, 0, len(entries)-compactionIndex+len(preservedIndexes))
|
||||||
for index, entry := range entries {
|
for index := 0; index < compactionIndex; index++ {
|
||||||
if index < compactionIndex && isPromptReplayEntryKind(entry.Kind) {
|
if !isPromptReplayEntryKind(entries[index].Kind) {
|
||||||
if _, ok := preservedIndexes[index]; !ok {
|
filtered = append(filtered, entries[index])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
filtered = append(filtered, entries[compactionIndex])
|
||||||
|
for index := 0; index < compactionIndex; index++ {
|
||||||
|
if _, ok := preservedIndexes[index]; !ok || isCompactionSummaryKind(entries[index].Kind) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
}
|
entry := entries[index]
|
||||||
if index < compactionIndex {
|
|
||||||
if rewritten, ok := compactedProjectionPreservedEntry(entry); ok {
|
if rewritten, ok := compactedProjectionPreservedEntry(entry); ok {
|
||||||
entry = rewritten
|
entry = rewritten
|
||||||
}
|
}
|
||||||
}
|
|
||||||
filtered = append(filtered, entry)
|
filtered = append(filtered, entry)
|
||||||
}
|
}
|
||||||
|
filtered = append(filtered, entries[compactionIndex+1:]...)
|
||||||
return filtered
|
return filtered
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -575,7 +579,7 @@ func (projector *HistoryProjector) ProjectCheckpointProjection(conversation *Con
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
state.Turns = turnIDs
|
state.Turns = append(cloneByteSlices(conversation.ImportedTurnIDs), turnIDs...)
|
||||||
replayMessages, err := projector.ProjectPromptReplay(conversation)
|
replayMessages, err := projector.ProjectPromptReplay(conversation)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -1292,7 +1296,7 @@ func filterCheckpointPersistentToolReplay(messages []promptengine.Message) []pro
|
|||||||
return filtered
|
return filtered
|
||||||
}
|
}
|
||||||
|
|
||||||
func restoreImportedReplayUserMessages(messages []promptengine.Message, importedTurns [][]byte) []promptengine.Message {
|
func restoreImportedReplayUserMessages(messages []promptengine.Message, importedTurns [][]byte, blobs importedBlobStore) []promptengine.Message {
|
||||||
if len(messages) == 0 || len(importedTurns) == 0 {
|
if len(messages) == 0 || len(importedTurns) == 0 {
|
||||||
return messages
|
return messages
|
||||||
}
|
}
|
||||||
@@ -1301,16 +1305,16 @@ func restoreImportedReplayUserMessages(messages []promptengine.Message, imported
|
|||||||
if len(rawTurn) == 0 {
|
if len(rawTurn) == 0 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
turn := &agentv1.ConversationTurnStructure{}
|
turn, _, err := decodeImportedTurn(rawTurn, blobs)
|
||||||
if err := proto.Unmarshal(rawTurn, turn); err != nil {
|
if err != nil || turn == nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
agentTurn := turn.GetAgentConversationTurn()
|
agentTurn := turn.GetAgentConversationTurn()
|
||||||
if agentTurn == nil || len(agentTurn.GetUserMessage()) == 0 {
|
if agentTurn == nil || len(agentTurn.GetUserMessage()) == 0 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
userMessage := &agentv1.UserMessage{}
|
userMessage, err := decodeImportedUserMessage(agentTurn.GetUserMessage(), blobs)
|
||||||
if err := proto.Unmarshal(agentTurn.GetUserMessage(), userMessage); err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
replay, ok := promptengine.BuildUserMessageReplayMessage(userMessage)
|
replay, ok := promptengine.BuildUserMessageReplayMessage(userMessage)
|
||||||
|
|||||||
@@ -224,6 +224,7 @@ func (service *Service) applyRunRewindToConversation(conversation *ConversationF
|
|||||||
conversation.Entries = nil
|
conversation.Entries = nil
|
||||||
conversation.NextEntrySeq = 1
|
conversation.NextEntrySeq = 1
|
||||||
conversation.NextTurnSeq = 1
|
conversation.NextTurnSeq = 1
|
||||||
|
conversation.ImportedTurnIDs = rewindImportedTurnPrefix(conversation.ImportedTurnIDs, decision)
|
||||||
appendEntriesInPlace(conversation, appendReplacementRunEntries(decision.PrefixEntries, entries))
|
appendEntriesInPlace(conversation, appendReplacementRunEntries(decision.PrefixEntries, entries))
|
||||||
applyRunRewindConversationState(conversation, intent, turnSeq)
|
applyRunRewindConversationState(conversation, intent, turnSeq)
|
||||||
deriveConversationLoopState(conversation)
|
deriveConversationLoopState(conversation)
|
||||||
@@ -269,10 +270,30 @@ func applyRunRewindMetadata(conversation *ConversationFile, source *Conversation
|
|||||||
if source.TokenDetailsMaxTokens > 0 {
|
if source.TokenDetailsMaxTokens > 0 {
|
||||||
conversation.TokenDetailsMaxTokens = source.TokenDetailsMaxTokens
|
conversation.TokenDetailsMaxTokens = source.TokenDetailsMaxTokens
|
||||||
}
|
}
|
||||||
|
decision := runRewindDecision{TargetTurnSeq: turnSeq}
|
||||||
|
if intent.ConversationState != nil {
|
||||||
|
decision.HasClientTurnCount = true
|
||||||
|
decision.ClientTurnCount = len(intent.ConversationState.GetTurns())
|
||||||
|
}
|
||||||
|
conversation.ImportedTurnIDs = rewindImportedTurnPrefix(source.ImportedTurnIDs, decision)
|
||||||
}
|
}
|
||||||
applyRunRewindConversationState(conversation, intent, turnSeq)
|
applyRunRewindConversationState(conversation, intent, turnSeq)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func rewindImportedTurnPrefix(importedTurnIDs [][]byte, decision runRewindDecision) [][]byte {
|
||||||
|
keep := decision.TargetTurnSeq - 1
|
||||||
|
if decision.HasClientTurnCount {
|
||||||
|
keep = int64(decision.ClientTurnCount)
|
||||||
|
}
|
||||||
|
if keep <= 0 || len(importedTurnIDs) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if keep > int64(len(importedTurnIDs)) {
|
||||||
|
keep = int64(len(importedTurnIDs))
|
||||||
|
}
|
||||||
|
return cloneByteSlices(importedTurnIDs[:keep])
|
||||||
|
}
|
||||||
|
|
||||||
func (service *Service) logRunRewindDecision(requestID string, conversationID string, eventName string, decision runRewindDecision) {
|
func (service *Service) logRunRewindDecision(requestID string, conversationID string, eventName string, decision runRewindDecision) {
|
||||||
if service == nil || !decision.Evaluated {
|
if service == nil || !decision.Evaluated {
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ func (service *Service) bootstrapRuntimeConversation(intent InboundIntent) (*Con
|
|||||||
}
|
}
|
||||||
importedEntries := []HistoryEntry(nil)
|
importedEntries := []HistoryEntry(nil)
|
||||||
if len(conversation.Entries) == 0 && intent.ConversationState != nil {
|
if len(conversation.Entries) == 0 && intent.ConversationState != nil {
|
||||||
importedEntries, err = service.importConversationState(conversation, intent.ConversationState)
|
importedEntries, err = service.importConversationState(conversation, intent.ConversationState, intent.PreFetchedBlobs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, agentv1.AgentMode_AGENT_MODE_AGENT, 0, nil, err
|
return nil, agentv1.AgentMode_AGENT_MODE_AGENT, 0, nil, err
|
||||||
}
|
}
|
||||||
@@ -138,6 +138,7 @@ func (service *Service) syncConversationRecord(conversationID string, conversati
|
|||||||
item.AutoCompactionReserveTokens = conversation.AutoCompactionReserveTokens
|
item.AutoCompactionReserveTokens = conversation.AutoCompactionReserveTokens
|
||||||
item.AutoCompactionTriggeredAt = conversation.AutoCompactionTriggeredAt
|
item.AutoCompactionTriggeredAt = conversation.AutoCompactionTriggeredAt
|
||||||
item.AutoCompactionSourceModelCallID = conversation.AutoCompactionSourceModelCallID
|
item.AutoCompactionSourceModelCallID = conversation.AutoCompactionSourceModelCallID
|
||||||
|
item.ImportedTurnIDs = cloneByteSlices(conversation.ImportedTurnIDs)
|
||||||
item.LatestRequestPrefix = cloneConversationRequestPrefix(conversation.LatestRequestPrefix)
|
item.LatestRequestPrefix = cloneConversationRequestPrefix(conversation.LatestRequestPrefix)
|
||||||
item.LastProviderCall = cloneConversationProviderCall(conversation.LastProviderCall)
|
item.LastProviderCall = cloneConversationProviderCall(conversation.LastProviderCall)
|
||||||
item.CreatedAt = conversation.CreatedAt
|
item.CreatedAt = conversation.CreatedAt
|
||||||
|
|||||||
@@ -559,6 +559,7 @@ func (service *Service) decodeInboundIntent(requestID string, message *agentv1.A
|
|||||||
}
|
}
|
||||||
intent.ConversationID = conversationID
|
intent.ConversationID = conversationID
|
||||||
intent.ConversationState = runRequest.GetConversationState()
|
intent.ConversationState = runRequest.GetConversationState()
|
||||||
|
intent.PreFetchedBlobs = runRequest.GetPreFetchedBlobs()
|
||||||
intent.UserMessage = extractUserMessage(message)
|
intent.UserMessage = extractUserMessage(message)
|
||||||
intent.RequestContext = extractRequestContext(message)
|
intent.RequestContext = extractRequestContext(message)
|
||||||
if service.shouldIgnoreEmptyResumeRunRequest(requestID, runRequest, intent.UserMessage, intent.RequestContext) {
|
if service.shouldIgnoreEmptyResumeRunRequest(requestID, runRequest, intent.UserMessage, intent.RequestContext) {
|
||||||
@@ -606,6 +607,7 @@ func (service *Service) decodeInboundIntent(requestID string, message *agentv1.A
|
|||||||
intent.ConversationID = conversationID
|
intent.ConversationID = conversationID
|
||||||
intent.SubagentTypeName = strings.TrimSpace(prewarmRequest.GetSubagentTypeName())
|
intent.SubagentTypeName = strings.TrimSpace(prewarmRequest.GetSubagentTypeName())
|
||||||
intent.ConversationState = prewarmRequest.GetConversationState()
|
intent.ConversationState = prewarmRequest.GetConversationState()
|
||||||
|
intent.PreFetchedBlobs = prewarmRequest.GetPreFetchedBlobs()
|
||||||
intent.Mode, intent.ModeSource, intent.HasExplicitMode, err = extractPrewarmMode(prewarmRequest)
|
intent.Mode, intent.ModeSource, intent.HasExplicitMode, err = extractPrewarmMode(prewarmRequest)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return InboundIntent{}, err
|
return InboundIntent{}, err
|
||||||
@@ -2246,6 +2248,15 @@ func (service *Service) finishSuccessfulTurnAfterCheckpoint(stream *ActiveStream
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (service *Service) finishFailedTurnAfterCheckpoint(stream *ActiveStream, terminalCode string, terminalMessage string) error {
|
||||||
|
if stream == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
err := service.broker.Fail(stream.RequestID, terminalCode, terminalMessage)
|
||||||
|
service.setTurnPhase(stream, TurnPhaseFailed)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
func (service *Service) failStreamIfNonTerminal(stream *ActiveStream, terminalCode string, cause error) error {
|
func (service *Service) failStreamIfNonTerminal(stream *ActiveStream, terminalCode string, cause error) error {
|
||||||
if stream == nil || cause == nil {
|
if stream == nil || cause == nil {
|
||||||
return nil
|
return nil
|
||||||
@@ -2265,6 +2276,10 @@ func (service *Service) publishCheckpoint(requestID string, conversationID strin
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (service *Service) publishCheckpointWithCompletion(requestID string, _ string, completion *pendingTurnCompletion) error {
|
func (service *Service) publishCheckpointWithCompletion(requestID string, _ string, completion *pendingTurnCompletion) error {
|
||||||
|
return service.publishCheckpointWithTerminalAction(requestID, successfulCheckpointTerminalAction(completion))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (service *Service) publishCheckpointWithTerminalAction(requestID string, terminal checkpointTerminalAction) error {
|
||||||
stream, ok := service.broker.Get(requestID)
|
stream, ok := service.broker.Get(requestID)
|
||||||
if !ok || stream == nil {
|
if !ok || stream == nil {
|
||||||
return fmt.Errorf("request is not active: %s", requestID)
|
return fmt.Errorf("request is not active: %s", requestID)
|
||||||
@@ -2282,7 +2297,7 @@ func (service *Service) publishCheckpointWithCompletion(requestID string, _ stri
|
|||||||
}
|
}
|
||||||
projection.State.PendingToolCalls = buildPendingToolCalls(pendingExecs, pendingInteractions)
|
projection.State.PendingToolCalls = buildPendingToolCalls(pendingExecs, pendingInteractions)
|
||||||
service.rewriteCheckpointTokenDetailsForClient(stream, conversation, projection.State)
|
service.rewriteCheckpointTokenDetailsForClient(stream, conversation, projection.State)
|
||||||
return service.queueCheckpointProjection(stream, projection, completion)
|
return service.queueCheckpointProjectionWithTerminal(stream, projection, terminal)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (service *Service) rewriteCheckpointTokenDetailsForClient(stream *ActiveStream, conversation *ConversationFile, state *agentv1.ConversationStateStructure) {
|
func (service *Service) rewriteCheckpointTokenDetailsForClient(stream *ActiveStream, conversation *ConversationFile, state *agentv1.ConversationStateStructure) {
|
||||||
@@ -2422,18 +2437,20 @@ func (service *Service) failActiveStream(stream *ActiveStream, conversationID st
|
|||||||
if cancel != nil {
|
if cancel != nil {
|
||||||
cancel()
|
cancel()
|
||||||
}
|
}
|
||||||
service.setTurnPhase(stream, TurnPhaseFailed)
|
if err := service.syncSummaryCarryForward(conversationID, requestID, modelCallID); err != nil {
|
||||||
var firstErr error
|
log.Printf(
|
||||||
if err := service.syncSummaryCarryForward(conversationID, requestID, modelCallID); err != nil && firstErr == nil {
|
"forwarder summary sync before failed terminal skipped request_id=%s model_call_id=%s err=%v",
|
||||||
firstErr = err
|
strings.TrimSpace(requestID),
|
||||||
|
strings.TrimSpace(modelCallID),
|
||||||
|
err,
|
||||||
|
)
|
||||||
}
|
}
|
||||||
if err := service.publishCheckpoint(requestID, conversationID); err != nil && firstErr == nil {
|
terminal := failedCheckpointTerminalAction(terminalCode, terminalMessage)
|
||||||
firstErr = err
|
if err := service.publishCheckpointWithTerminalAction(requestID, terminal); err != nil {
|
||||||
|
log.Printf("forwarder checkpoint queue before failed terminal skipped request_id=%s err=%v", strings.TrimSpace(requestID), err)
|
||||||
|
return service.finishFailedTurnAfterCheckpoint(stream, terminalCode, terminalMessage)
|
||||||
}
|
}
|
||||||
if err := service.broker.Fail(requestID, terminalCode, terminalMessage); err != nil && firstErr == nil {
|
return nil
|
||||||
firstErr = err
|
|
||||||
}
|
|
||||||
return firstErr
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// buildRunEntries 构造一次 run intent 需要写入 history 的首批 entry。
|
// buildRunEntries 构造一次 run intent 需要写入 history 的首批 entry。
|
||||||
|
|||||||
@@ -45,13 +45,25 @@ func (snapshot turnUsageSnapshot) requestTokensTotal() int64 {
|
|||||||
return snapshot.promptTokensTotal() + nonNegativeInt64(snapshot.OutputTokens)
|
return snapshot.promptTokensTotal() + nonNegativeInt64(snapshot.OutputTokens)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (service *Service) importConversationState(item *ConversationFile, state *agentv1.ConversationStateStructure) ([]HistoryEntry, error) {
|
func (service *Service) importConversationState(item *ConversationFile, state *agentv1.ConversationStateStructure, prefetchedBlobs []*agentv1.PreFetchedBlob) ([]HistoryEntry, error) {
|
||||||
if item == nil || state == nil {
|
if item == nil || state == nil {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
blobs, err := newImportedBlobStore(prefetchedBlobs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
importedIDs, err := importedTurnIDs(state.GetTurns(), blobs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
item.TokenDetailsUsedTokens = state.GetTokenDetails().GetUsedTokens()
|
item.TokenDetailsUsedTokens = state.GetTokenDetails().GetUsedTokens()
|
||||||
|
item.ImportedTurnIDs = importedIDs
|
||||||
|
if minimumNextTurnSeq := int64(len(item.ImportedTurnIDs)) + 1; item.NextTurnSeq < minimumNextTurnSeq {
|
||||||
|
item.NextTurnSeq = minimumNextTurnSeq
|
||||||
|
}
|
||||||
entries := make([]HistoryEntry, 0, 2)
|
entries := make([]HistoryEntry, 0, 2)
|
||||||
if messages, err := importedConversationStateModelMessages(state); err != nil {
|
if messages, err := importedConversationStateModelMessagesWithBlobs(state, blobs); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
} else {
|
} else {
|
||||||
for _, message := range messages {
|
for _, message := range messages {
|
||||||
@@ -105,6 +117,10 @@ func (service *Service) importConversationState(item *ConversationFile, state *a
|
|||||||
}
|
}
|
||||||
|
|
||||||
func importedConversationStateModelMessages(state *agentv1.ConversationStateStructure) ([]modeladapter.Message, error) {
|
func importedConversationStateModelMessages(state *agentv1.ConversationStateStructure) ([]modeladapter.Message, error) {
|
||||||
|
return importedConversationStateModelMessagesWithBlobs(state, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func importedConversationStateModelMessagesWithBlobs(state *agentv1.ConversationStateStructure, blobs importedBlobStore) ([]modeladapter.Message, error) {
|
||||||
if state == nil {
|
if state == nil {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
@@ -113,7 +129,7 @@ func importedConversationStateModelMessages(state *agentv1.ConversationStateStru
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("decode imported replay messages: %w", err)
|
return nil, fmt.Errorf("decode imported replay messages: %w", err)
|
||||||
}
|
}
|
||||||
decoded = restoreImportedReplayUserMessages(decoded, state.GetTurns())
|
decoded = restoreImportedReplayUserMessages(decoded, state.GetTurns(), blobs)
|
||||||
decoded = filterLegacyPlainWriteReplay(decoded)
|
decoded = filterLegacyPlainWriteReplay(decoded)
|
||||||
decoded = filterInternalPromptContextReplay(decoded)
|
decoded = filterInternalPromptContextReplay(decoded)
|
||||||
messages := make([]modeladapter.Message, 0, len(decoded))
|
messages := make([]modeladapter.Message, 0, len(decoded))
|
||||||
@@ -133,35 +149,18 @@ func importedConversationStateModelMessages(state *agentv1.ConversationStateStru
|
|||||||
if len(rawTurn) == 0 {
|
if len(rawTurn) == 0 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
turn := &agentv1.ConversationTurnStructure{}
|
turn, turnID, err := decodeImportedTurn(rawTurn, blobs)
|
||||||
if err := proto.Unmarshal(rawTurn, turn); err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("decode imported turn: %w", err)
|
return nil, err
|
||||||
}
|
}
|
||||||
agentTurn := turn.GetAgentConversationTurn()
|
if turn == nil && len(turnID) > 0 {
|
||||||
if agentTurn == nil {
|
return nil, fmt.Errorf("missing prefetched turn blob %x", turnID)
|
||||||
continue
|
|
||||||
}
|
|
||||||
if rawUser := agentTurn.GetUserMessage(); len(rawUser) > 0 {
|
|
||||||
userMessage := &agentv1.UserMessage{}
|
|
||||||
if err := proto.Unmarshal(rawUser, userMessage); err != nil {
|
|
||||||
return nil, fmt.Errorf("decode imported turn user_message: %w", err)
|
|
||||||
}
|
|
||||||
if replay, ok := promptengine.BuildUserMessageReplayMessage(userMessage); ok {
|
|
||||||
messages = append(messages, toModelMessage(replay))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for _, rawStep := range agentTurn.GetSteps() {
|
|
||||||
if len(rawStep) == 0 {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
step := &agentv1.ConversationStep{}
|
|
||||||
if err := proto.Unmarshal(rawStep, step); err != nil {
|
|
||||||
return nil, fmt.Errorf("decode imported turn step: %w", err)
|
|
||||||
}
|
|
||||||
for _, replay := range promptengine.BuildLegacyMessagesFromConversationStep(step) {
|
|
||||||
messages = append(messages, toModelMessage(replay))
|
|
||||||
}
|
}
|
||||||
|
turnMessages, err := importedBlobTurnMessages(turn, blobs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
}
|
}
|
||||||
|
messages = append(messages, turnMessages...)
|
||||||
}
|
}
|
||||||
return normalizeReplayMessageSequence(messages), nil
|
return normalizeReplayMessageSequence(messages), nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -39,6 +39,7 @@ type ConversationFile struct {
|
|||||||
CurrentPlanText string `json:"current_plan_text,omitempty"`
|
CurrentPlanText string `json:"current_plan_text,omitempty"`
|
||||||
CurrentPlans map[string]*agentv1.PlanRegistryEntry `json:"current_plans,omitempty"`
|
CurrentPlans map[string]*agentv1.PlanRegistryEntry `json:"current_plans,omitempty"`
|
||||||
CurrentTodos []*agentv1.TodoItem `json:"current_todos,omitempty"`
|
CurrentTodos []*agentv1.TodoItem `json:"current_todos,omitempty"`
|
||||||
|
ImportedTurnIDs [][]byte `json:"imported_turn_ids,omitempty"`
|
||||||
LatestRequestPrefix *ConversationRequestPrefix `json:"latest_request_prefix,omitempty"`
|
LatestRequestPrefix *ConversationRequestPrefix `json:"latest_request_prefix,omitempty"`
|
||||||
LastProviderCall *ConversationProviderCall `json:"last_provider_call,omitempty"`
|
LastProviderCall *ConversationProviderCall `json:"last_provider_call,omitempty"`
|
||||||
CreatedAt time.Time `json:"created_at"`
|
CreatedAt time.Time `json:"created_at"`
|
||||||
@@ -224,11 +225,25 @@ type pendingTurnCompletion struct {
|
|||||||
Disposition pendingCompletionDisposition
|
Disposition pendingCompletionDisposition
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type checkpointTerminalActionKind uint8
|
||||||
|
|
||||||
|
const (
|
||||||
|
checkpointTerminalActionNone checkpointTerminalActionKind = iota
|
||||||
|
checkpointTerminalActionComplete
|
||||||
|
checkpointTerminalActionFail
|
||||||
|
)
|
||||||
|
|
||||||
|
type checkpointTerminalAction struct {
|
||||||
|
Kind checkpointTerminalActionKind
|
||||||
|
Completion pendingTurnCompletion
|
||||||
|
ErrorCode string
|
||||||
|
ErrorMessage string
|
||||||
|
}
|
||||||
|
|
||||||
type pendingCheckpointPublish struct {
|
type pendingCheckpointPublish struct {
|
||||||
State *agentv1.ConversationStateStructure
|
State *agentv1.ConversationStateStructure
|
||||||
Required map[string]struct{}
|
Required map[string]struct{}
|
||||||
Completion *pendingTurnCompletion
|
Terminal checkpointTerminalAction
|
||||||
Published bool
|
|
||||||
}
|
}
|
||||||
|
|
||||||
type PendingCompaction struct {
|
type PendingCompaction struct {
|
||||||
@@ -430,6 +445,7 @@ type InboundIntent struct {
|
|||||||
SubagentTypeName string
|
SubagentTypeName string
|
||||||
SubagentModelOverrides map[string]runtimecore.SubagentModelOverrideSelection
|
SubagentModelOverrides map[string]runtimecore.SubagentModelOverrideSelection
|
||||||
ConversationState *agentv1.ConversationStateStructure
|
ConversationState *agentv1.ConversationStateStructure
|
||||||
|
PreFetchedBlobs []*agentv1.PreFetchedBlob
|
||||||
UserMessage *agentv1.UserMessage
|
UserMessage *agentv1.UserMessage
|
||||||
RequestContext *agentv1.RequestContext
|
RequestContext *agentv1.RequestContext
|
||||||
ClientMessage *agentv1.AgentClientMessage
|
ClientMessage *agentv1.AgentClientMessage
|
||||||
|
|||||||
+6
-6
@@ -8,9 +8,9 @@ QQ交流群：
|
|||||||
Tg群组:
|
Tg群组:
|
||||||
https://t.me/cursor_byok
|
https://t.me/cursor_byok
|
||||||
|
|
||||||
- 修复检查点，支持Fork Chat
|
- 修复对话中断时回复内容丢失
|
||||||
- 修复打断对话的上下文丢失问题
|
- 重构检查点压缩，提升稳定性
|
||||||
- 重构UI
|
- 修复OpenAI推理摘要显示
|
||||||
- 支持拖动模型排序
|
- 修复Anthropic思考块缺失
|
||||||
- 支持一键拉模型
|
- 支持Shell工具流式输出
|
||||||
- 支持非主流chat端点
|
- 修复CLI模型名称显示
|
||||||
|
|||||||
Reference in New Issue
Block a user