Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
88b41ba04c | ||
|
|
b33d950d2b | ||
|
|
f0bf667d02 | ||
|
|
e8a335d5ba | ||
|
|
45687bcf67 | ||
|
|
178e74b403 | ||
|
|
a37f06352c | ||
|
|
9286ca67e9 | ||
|
|
7b23c61723 |
@@ -1,7 +1,15 @@
|
||||
.git
|
||||
.gitea
|
||||
node_modules
|
||||
.opencode/node_modules
|
||||
.env
|
||||
.env.*
|
||||
.local.env
|
||||
dist
|
||||
.vscode
|
||||
data
|
||||
logs
|
||||
source
|
||||
coverage
|
||||
test-results
|
||||
*.log
|
||||
|
||||
+1
-1
@@ -5,7 +5,7 @@ LOG_LEVEL=info
|
||||
OPENCODE_MODE=embedded
|
||||
OPENCODE_MODEL=deepseek/deepseek-v4-flash
|
||||
OPENCODE_HOSTNAME=127.0.0.1
|
||||
OPENCODE_PORT=4096
|
||||
OPENCODE_PORT=0
|
||||
|
||||
AGENT_LOCAL_USER_ID=local-user
|
||||
# 必须填写 Server /api/v1/projects 返回的项目 UUID。
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
name: Agent CI/CD
|
||||
|
||||
on:
|
||||
push:
|
||||
tags:
|
||||
- "v*"
|
||||
workflow_dispatch: {}
|
||||
|
||||
jobs:
|
||||
build-test-publish-and-deploy:
|
||||
uses: OrgTJWater/ci-templates/.gitea/workflows/container-cd.yml@main
|
||||
with:
|
||||
image_name: gitea.waternetwork.cn/orgtjwater/next-tjwater-agent
|
||||
dockerfile: Dockerfile
|
||||
build_context: .
|
||||
test_target: build
|
||||
deploy_service: next-agent
|
||||
deploy_host: 192.168.1.114
|
||||
secrets:
|
||||
REGISTRY_USERNAME: ${{ secrets.REGISTRY_USERNAME }}
|
||||
REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }}
|
||||
DEV_DEPLOY_SSH_KEY: ${{ secrets.DEV_DEPLOY_SSH_KEY }}
|
||||
@@ -32,8 +32,8 @@ Agent 负责:
|
||||
1. `render_scada_analysis` 只能在分析结论已经完成后调用,不得让前端计算、猜测或修改高中低等级。
|
||||
2. 每次普通工具调用必须填写具体 `reason`。
|
||||
3. 不生成 JS、JSX、HTML、CSS 或可执行前端代码。
|
||||
4. SCADA 地图结果只允许使用可信 `sensor_id` 和 `high | medium | low | unrated` 等级,不得传入自定义标签、颜色或 HTML。
|
||||
5. 同一批结果不得包含重复 `sensor_id`,每次最多 100 个点位。
|
||||
4. SCADA 地图结果只允许使用后端 API 返回的可信 `device_id` 和 `high | medium | low | unrated` 等级,不得传入自定义标签、颜色或 HTML。
|
||||
5. 同一批结果不得包含重复 `device_id`,每次最多 100 个点位。
|
||||
6. 只有 SCADA 地图工具返回浏览器的成功结果后,才能声称已渲染或已清除;工具报错、超时或无活跃浏览器连接时,必须明确告知用户动作失败,不得伪造已渲染的点位、编号或等级。
|
||||
|
||||
## 执行约束
|
||||
|
||||
@@ -2,7 +2,7 @@ import { tool } from "@opencode-ai/plugin";
|
||||
|
||||
import { executeFrontendAction } from "./frontend_action.js";
|
||||
|
||||
const sensorId = tool.schema.string().trim().min(1).max(128);
|
||||
const deviceId = tool.schema.string().trim().min(1).max(128);
|
||||
|
||||
export default tool({
|
||||
description:
|
||||
@@ -16,7 +16,7 @@ export default tool({
|
||||
items: tool.schema
|
||||
.array(
|
||||
tool.schema.object({
|
||||
sensor_id: sensorId.describe("Trusted SCADA sensor identifier."),
|
||||
device_id: deviceId.describe("Trusted SCADA device identifier from the backend API."),
|
||||
level: tool.schema
|
||||
.enum(["high", "medium", "low", "unrated"])
|
||||
.describe("Completed analysis level; use unrated only when no level was concluded."),
|
||||
@@ -25,8 +25,8 @@ export default tool({
|
||||
.min(1)
|
||||
.max(100)
|
||||
.refine(
|
||||
(items) => new Set(items.map((item) => item.sensor_id)).size === items.length,
|
||||
"sensor_id values must be unique",
|
||||
(items) => new Set(items.map((item) => item.device_id)).size === items.length,
|
||||
"device_id values must be unique",
|
||||
)
|
||||
.describe("One to one hundred unique SCADA analysis results."),
|
||||
},
|
||||
|
||||
@@ -45,4 +45,13 @@ and UIEnvelope mapping. Tests must not depend on mutable files under `data/`.
|
||||
|
||||
This experimental service has no user authentication. Do not add secrets,
|
||||
tokens, passwords, API keys, session dumps, or generated result references to
|
||||
the repository.
|
||||
the repository. Keep `.env`, `data/`, and `logs/` statically denied in
|
||||
`opencode.json`; do not replace those denials with prompt-only guidance. Keep
|
||||
arbitrary Bash and edit operations approval-gated and subagents disabled.
|
||||
|
||||
## Runtime Readiness
|
||||
|
||||
Complete the OpenCode health check, warmup session, and model tool discovery
|
||||
before the HTTP server starts listening. A warmup failure must fail startup.
|
||||
Once listening, `/health` must expose the `ready` and `warmed_up` readiness
|
||||
fields used by deployment checks.
|
||||
|
||||
+14
-15
@@ -1,15 +1,17 @@
|
||||
FROM oven/bun:canary-slim AS bun-bin
|
||||
FROM oven/bun:canary-slim@sha256:24a8816a7388dd900e851b61b6d44a4a1908f464661a9ffe393e13e661f0ae3e AS bun-bin
|
||||
|
||||
FROM smanx/opencode:latest AS base
|
||||
FROM smanx/opencode:1.18.13@sha256:b976acda21efffacd44abd7847dac7d646910dbaa477d1877e2881b39cf22a91 AS base
|
||||
USER root
|
||||
ARG UBUNTU_APT_MIRROR=mirrors.aliyun.com
|
||||
ARG PYPI_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple
|
||||
ARG PYPI_TRUSTED_HOST=pypi.tuna.tsinghua.edu.cn
|
||||
ARG NPM_REGISTRY=https://registry.npmmirror.com
|
||||
ENV VIRTUAL_ENV=/opt/venv
|
||||
ENV PATH="$VIRTUAL_ENV/bin:/root/.local/bin:$PATH"
|
||||
ENV PIP_INDEX_URL=${PYPI_INDEX_URL}
|
||||
ENV PIP_TRUSTED_HOST=${PYPI_TRUSTED_HOST}
|
||||
ENV UV_INDEX_URL=${PYPI_INDEX_URL}
|
||||
ENV BUN_CONFIG_REGISTRY=${NPM_REGISTRY}
|
||||
|
||||
RUN sed -i "s|http://archive.ubuntu.com|https://${UBUNTU_APT_MIRROR}|g; s|http://security.ubuntu.com|https://${UBUNTU_APT_MIRROR}|g" /etc/apt/sources.list 2>/dev/null || true && \
|
||||
sed -i "s|http://archive.ubuntu.com|https://${UBUNTU_APT_MIRROR}|g; s|http://security.ubuntu.com|https://${UBUNTU_APT_MIRROR}|g" /etc/apt/sources.list.d/*.sources 2>/dev/null || true && \
|
||||
@@ -19,14 +21,14 @@ RUN sed -i "s|http://archive.ubuntu.com|https://${UBUNTU_APT_MIRROR}|g; s|http:/
|
||||
unzip \
|
||||
python3 \
|
||||
python3-venv && \
|
||||
curl -LsSf https://astral.sh/uv/install.sh | sh && \
|
||||
ln -s /root/.local/bin/uv /usr/local/bin/uv && \
|
||||
ln -sf /usr/bin/python3 /usr/local/bin/python && \
|
||||
mkdir -p /root/.config/pip && \
|
||||
printf "[global]\nindex-url = %s\ntrusted-host = %s\n" "$PIP_INDEX_URL" "$PIP_TRUSTED_HOST" > /root/.config/pip/pip.conf && \
|
||||
uv venv "$VIRTUAL_ENV" && \
|
||||
uv pip install --python "$VIRTUAL_ENV/bin/python" \
|
||||
--index-url "$UV_INDEX_URL" \
|
||||
python3 -m venv "$VIRTUAL_ENV" && \
|
||||
pip install \
|
||||
--index-url "$PIP_INDEX_URL" \
|
||||
--trusted-host "$PIP_TRUSTED_HOST" \
|
||||
uv \
|
||||
pip \
|
||||
setuptools \
|
||||
wheel \
|
||||
@@ -37,6 +39,7 @@ RUN sed -i "s|http://archive.ubuntu.com|https://${UBUNTU_APT_MIRROR}|g; s|http:/
|
||||
rich \
|
||||
ipython \
|
||||
pytest && \
|
||||
ln -s "$VIRTUAL_ENV/bin/uv" /usr/local/bin/uv && \
|
||||
rm -rf /var/lib/apt/lists/*
|
||||
|
||||
COPY --from=bun-bin /usr/local/bin/bun /usr/local/bin/bun
|
||||
@@ -54,29 +57,25 @@ WORKDIR /app
|
||||
|
||||
COPY --from=deps /app/node_modules ./node_modules
|
||||
COPY --from=deps /app/.opencode/node_modules ./.opencode/node_modules
|
||||
COPY package.json bun.lock ./
|
||||
COPY tsconfig.json opencode.json README.md .gitignore ./
|
||||
COPY src ./src
|
||||
COPY .opencode ./.opencode
|
||||
RUN bun run check
|
||||
|
||||
FROM base AS runner
|
||||
FROM build AS runner
|
||||
WORKDIR /app
|
||||
|
||||
ENV NODE_ENV=production
|
||||
ENV HOST=0.0.0.0
|
||||
ENV PORT=8787
|
||||
|
||||
COPY --from=deps /app/node_modules ./node_modules
|
||||
COPY --from=deps /app/.opencode/node_modules ./.opencode/node_modules
|
||||
COPY package.json bun.lock ./
|
||||
COPY tsconfig.json opencode.json .gitignore ./
|
||||
COPY src ./src
|
||||
COPY .opencode ./.opencode
|
||||
|
||||
COPY entrypoint.sh /entrypoint.sh
|
||||
RUN chmod +x /entrypoint.sh
|
||||
|
||||
ENTRYPOINT ["/entrypoint.sh"]
|
||||
|
||||
EXPOSE 8787
|
||||
HEALTHCHECK --interval=30s --timeout=5s --start-period=60s --retries=3 \
|
||||
CMD curl --fail --silent --show-error "http://127.0.0.1:${PORT}/health" >/dev/null || exit 1
|
||||
CMD ["bun", "src/server.ts"]
|
||||
|
||||
@@ -1,53 +1,52 @@
|
||||
# TJWaterAgentExperimental
|
||||
# TJWater Agent 实验服务
|
||||
|
||||
Experimental opencode-based Agent service for TJWater UI orchestration.
|
||||
基于 OpenCode 的实验性 Agent 服务,用于编排 TJWater 前端交互。
|
||||
|
||||
This repository is intentionally separate from the production `TJWaterAgent`
|
||||
history. It does not include user authentication or the old `tjwater-cli`
|
||||
bridge. Runtime requests use a local default context and can later be wired to a
|
||||
real auth provider through `src/context/`.
|
||||
本仓库与生产环境中的 `TJWaterAgent` 历史代码相互独立,不包含用户认证,也不包含旧版 `tjwater-cli` 桥接层。运行时请求默认使用本地上下文,后续可通过 `src/context/` 接入实际的认证提供方。
|
||||
|
||||
## Current Boundaries
|
||||
## 当前边界
|
||||
|
||||
- No Keycloak / metadata auth context.
|
||||
- No required `Authorization` or `X-Project-Id` headers.
|
||||
- No `cli/`, no `tjwater_cli` opencode tool, no free-form CLI command bridge.
|
||||
- Agent visual output is expressed through `agent-ui@1` UIEnvelope events.
|
||||
- Tools that depend on the TJWater Server are disabled. Arbitrary URLs, SQL,
|
||||
shell commands, and filesystem paths remain unavailable.
|
||||
- 不包含 Keycloak 或元数据认证上下文。
|
||||
- 不强制要求 `Authorization` 或 `X-Project-Id` 请求头。
|
||||
- 不包含 `cli/`、`tjwater_cli` OpenCode 工具或自由形式的 CLI 命令桥接。
|
||||
- Agent 的结构化界面输出使用 `agent-ui@1` UIEnvelope 事件;受控浏览器操作使用 `frontend-action@1` 前端动作。
|
||||
- 依赖 TJWater Server 的工具处于禁用状态,Agent 不能访问任意 URL、SQL、Shell 命令或文件系统路径。
|
||||
|
||||
## Main API
|
||||
## 主要 API
|
||||
|
||||
```text
|
||||
GET /health
|
||||
GET /api/v1/agent/chat/models
|
||||
GET /api/v1/agent/chat/ui-registry
|
||||
POST /api/v1/agent/chat/session
|
||||
GET /api/v1/agent/chat/sessions
|
||||
GET /api/v1/agent/chat/session/:session_id
|
||||
GET /api/v1/agent/chat/session/:session_id/stream
|
||||
POST /api/v1/agent/chat/stream
|
||||
POST /api/v1/agent/chat/abort
|
||||
GET /health
|
||||
GET /api/v1/agent/chat/models
|
||||
GET /api/v1/agent/chat/ui-registry
|
||||
GET /api/v1/agent/chat/frontend-action-registry
|
||||
POST /api/v1/agent/chat/frontend-actions/:action_id/result
|
||||
POST /api/v1/agent/chat/session
|
||||
GET /api/v1/agent/chat/sessions
|
||||
GET /api/v1/agent/chat/session/:session_id
|
||||
GET /api/v1/agent/chat/session/:session_id/stream
|
||||
POST /api/v1/agent/chat/stream
|
||||
POST /api/v1/agent/chat/abort
|
||||
```
|
||||
|
||||
`POST /api/v1/agent/chat/stream` returns SSE. Important events:
|
||||
`POST /api/v1/agent/chat/stream` 通过 SSE 返回事件流。主要事件如下:
|
||||
|
||||
| event | Purpose |
|
||||
| 事件 | 用途 |
|
||||
| --- | --- |
|
||||
| `progress` | Agent planning/tool progress |
|
||||
| `token` | Assistant text delta |
|
||||
| `ui_envelope` | Structured UI description for trusted frontend rendering |
|
||||
| `tool_call` | Deprecated compatibility/debug event |
|
||||
| `permission_request` | Runtime permission request |
|
||||
| `question_request` | Runtime/user clarification request |
|
||||
| `done` | Run completed |
|
||||
| `error` | Run failed or aborted |
|
||||
| `progress` | Agent 规划和工具执行进度 |
|
||||
| `token` | 助手回复的增量文本 |
|
||||
| `ui_envelope` | 供可信前端渲染的结构化界面描述 |
|
||||
| `frontend_action` | 请求可信前端执行受控动作 |
|
||||
| `tool_call` | 用于兼容和调试的旧版工具调用事件 |
|
||||
| `permission_request` | 运行时权限请求 |
|
||||
| `question_request` | 运行时向用户发起的澄清请求 |
|
||||
| `todo_update` | Agent 待办事项更新 |
|
||||
| `session_title` | 会话标题更新 |
|
||||
| `done` | 本次运行完成 |
|
||||
| `error` | 本次运行失败或被中止 |
|
||||
|
||||
## Local Context
|
||||
## 本地上下文
|
||||
|
||||
The service uses `src/context/localContext.ts`.
|
||||
|
||||
Defaults:
|
||||
服务通过 `src/context/localContext.ts` 提供本地上下文,默认值如下:
|
||||
|
||||
```text
|
||||
AGENT_LOCAL_USER_ID=local-user
|
||||
@@ -55,12 +54,9 @@ AGENT_LOCAL_PROJECT_ID=local-project
|
||||
AGENT_LOCAL_NETWORK=local-network
|
||||
```
|
||||
|
||||
Monitoring time-series data is read from the agent project's `source/`
|
||||
directory and analyzed through validated local skills. `get_project_context`,
|
||||
`list_scada_assets`, `query_scada_readings`, `web_search`, and `geocode` are not
|
||||
exposed because they depend on the TJWater Server.
|
||||
监测时序数据从 Agent 项目的 `source/` 目录读取,并通过经过校验的本地技能进行分析。`get_project_context`、`list_scada_assets`、`query_scada_readings`、`web_search` 和 `geocode` 依赖 TJWater Server,因此当前不对外开放。
|
||||
|
||||
Optional request headers can override the local context during experiments:
|
||||
实验期间可以通过以下请求头覆盖本地上下文:
|
||||
|
||||
```text
|
||||
X-Agent-Local-User
|
||||
@@ -69,21 +65,31 @@ X-Agent-Local-Network
|
||||
X-Trace-Id
|
||||
```
|
||||
|
||||
## UIEnvelope
|
||||
## UIEnvelope 与前端动作
|
||||
|
||||
The registry is exposed at:
|
||||
UIEnvelope 注册表通过以下接口提供:
|
||||
|
||||
```text
|
||||
GET /api/v1/agent/chat/ui-registry
|
||||
```
|
||||
|
||||
No visual or frontend-action tools are currently exposed to the Agent. The
|
||||
registry endpoints remain available for protocol compatibility and return no
|
||||
registered charts, components, or actions.
|
||||
该接口用于兼容 `agent-ui@1` 协议。目前没有注册图表语法、界面组件或 UIEnvelope 动作。
|
||||
|
||||
See [docs/agent-ui-protocol.md](docs/agent-ui-protocol.md).
|
||||
前端动作使用独立的注册表:
|
||||
|
||||
## Development
|
||||
```text
|
||||
GET /api/v1/agent/chat/frontend-action-registry
|
||||
```
|
||||
|
||||
当前提供 `render_scada_analysis` 和 `clear_scada_analysis` 两个地图叠加层动作。浏览器需要在流式请求中声明 `frontend-action@1` 能力,执行动作后再通过以下接口回传结果:
|
||||
|
||||
```text
|
||||
POST /api/v1/agent/chat/frontend-actions/:action_id/result
|
||||
```
|
||||
|
||||
协议详情参见 [Agent UI 协议](docs/agent-ui-protocol.md)。
|
||||
|
||||
## 本地开发
|
||||
|
||||
```bash
|
||||
bun install
|
||||
@@ -92,4 +98,15 @@ bun run check
|
||||
bun test tests/**/*.test.ts
|
||||
```
|
||||
|
||||
`bun run check` typechecks the service and `.opencode` tools.
|
||||
`bun run check` 会检查服务代码和 `.opencode` 工具的类型。
|
||||
|
||||
## CI/CD
|
||||
|
||||
Gitea Actions 通过 `.gitea/workflows/package.yml` 复用
|
||||
`OrgTJWater/ci-templates` 的容器发布流程。手动触发只构建镜像并在 Dev
|
||||
服务器运行候选验证;从 `main` 创建并推送 `v*` tag 后,候选验证通过才会
|
||||
提升为正式 `agent` 服务。
|
||||
|
||||
仓库需要配置 `REGISTRY_USERNAME`、`REGISTRY_PASSWORD` 和
|
||||
`DEV_DEPLOY_SSH_KEY` 三个 Gitea Secrets。`.env`、持久化数据、日志和
|
||||
`source/` 业务数据不会写入镜像;它们由 Dev 服务器通过环境文件和卷挂载。
|
||||
|
||||
+45
-6
@@ -8,15 +8,44 @@
|
||||
}
|
||||
},
|
||||
"model": "deepseek/deepseek-v4-flash",
|
||||
"experimental": {
|
||||
"continue_loop_on_deny": true
|
||||
},
|
||||
"server": {
|
||||
"hostname": "127.0.0.1",
|
||||
"port": 4096
|
||||
},
|
||||
"permission": {
|
||||
"*": "allow",
|
||||
"external_directory": "ask",
|
||||
"bash": {
|
||||
"*": "ask",
|
||||
"external_directory": "deny",
|
||||
"read": {
|
||||
"*": "allow",
|
||||
".env": "deny",
|
||||
".env.*": "deny",
|
||||
"*.env": "deny",
|
||||
"**/.env": "deny",
|
||||
"**/.env.*": "deny",
|
||||
"**/*.env": "deny",
|
||||
"data/**": "deny",
|
||||
"**/data/**": "deny",
|
||||
"logs/**": "deny",
|
||||
"**/logs/**": "deny"
|
||||
},
|
||||
"edit": {
|
||||
"*": "ask",
|
||||
".env": "deny",
|
||||
".env.*": "deny",
|
||||
"*.env": "deny",
|
||||
"**/.env": "deny",
|
||||
"**/.env.*": "deny",
|
||||
"**/*.env": "deny",
|
||||
"data/**": "deny",
|
||||
"**/data/**": "deny",
|
||||
"logs/**": "deny",
|
||||
"**/logs/**": "deny"
|
||||
},
|
||||
"bash": {
|
||||
"*": "ask",
|
||||
"rm *": "ask",
|
||||
"rmdir *": "ask",
|
||||
"mv *": "ask",
|
||||
@@ -24,10 +53,20 @@
|
||||
"chown *": "ask",
|
||||
"sudo *": "ask",
|
||||
"curl *": "ask",
|
||||
"wget *": "ask"
|
||||
"wget *": "ask",
|
||||
"*.env*": "deny",
|
||||
"*data/*": "deny",
|
||||
"* data": "deny",
|
||||
"*/data": "deny",
|
||||
"*logs/*": "deny",
|
||||
"* logs": "deny",
|
||||
"*/logs": "deny"
|
||||
},
|
||||
"edit": "ask",
|
||||
"task": "deny"
|
||||
"question": "allow",
|
||||
"task": "deny",
|
||||
"todo": "allow",
|
||||
"todoread": "allow",
|
||||
"todowrite": "allow"
|
||||
},
|
||||
"default_agent": "instruction"
|
||||
}
|
||||
|
||||
+2
-2
@@ -58,8 +58,8 @@ const envSchema = z
|
||||
OPENCODE_MODE: z.enum(["embedded", "client"]).default("embedded"),
|
||||
// embedded opencode server 的监听地址。
|
||||
OPENCODE_HOSTNAME: z.string().default("127.0.0.1"),
|
||||
// embedded opencode server 的监听端口。
|
||||
OPENCODE_PORT: z.coerce.number().int().positive().default(4096),
|
||||
// embedded opencode server 的监听端口;0 表示由系统动态分配。
|
||||
OPENCODE_PORT: z.coerce.number().int().nonnegative().default(0),
|
||||
// opencode SDK 启动或连接运行时时的超时时间(毫秒)。
|
||||
OPENCODE_TIMEOUT_MS: z.coerce.number().int().positive().default(5000),
|
||||
// 默认使用的 opencode 模型标识。
|
||||
|
||||
@@ -2,7 +2,7 @@ import { z } from "zod";
|
||||
import type { FrontendActionManifest } from "./types.js";
|
||||
|
||||
type Definition = { manifest: FrontendActionManifest; inputSchema: z.ZodTypeAny; outputSchema: z.ZodTypeAny };
|
||||
const sensorId = z.string().trim().min(1).max(128);
|
||||
const deviceId = z.string().trim().min(1).max(128);
|
||||
const scadaLevel = z.enum(["high", "medium", "low", "unrated"]);
|
||||
const levelCounts = z.object({
|
||||
high: z.number().int().nonnegative(),
|
||||
@@ -11,18 +11,18 @@ const levelCounts = z.object({
|
||||
unrated: z.number().int().nonnegative(),
|
||||
}).strict();
|
||||
const renderScadaAnalysisInput = z.object({
|
||||
items: z.array(z.object({ sensor_id: sensorId, level: scadaLevel }).strict()).min(1).max(100),
|
||||
items: z.array(z.object({ device_id: deviceId, level: scadaLevel }).strict()).min(1).max(100),
|
||||
}).strict().superRefine((value, context) => {
|
||||
const seen = new Set<string>();
|
||||
value.items.forEach((item, index) => {
|
||||
if (seen.has(item.sensor_id)) {
|
||||
if (seen.has(item.device_id)) {
|
||||
context.addIssue({
|
||||
code: z.ZodIssueCode.custom,
|
||||
message: "sensor_id values must be unique",
|
||||
path: ["items", index, "sensor_id"],
|
||||
message: "device_id values must be unique",
|
||||
path: ["items", index, "device_id"],
|
||||
});
|
||||
}
|
||||
seen.add(item.sensor_id);
|
||||
seen.add(item.device_id);
|
||||
});
|
||||
});
|
||||
const definitions: Definition[] = [
|
||||
@@ -36,8 +36,8 @@ const definitions: Definition[] = [
|
||||
},
|
||||
inputSchema: renderScadaAnalysisInput,
|
||||
outputSchema: z.object({
|
||||
rendered_ids: z.array(sensorId),
|
||||
missing_ids: z.array(sensorId),
|
||||
rendered_ids: z.array(deviceId),
|
||||
missing_ids: z.array(deviceId),
|
||||
level_counts: levelCounts,
|
||||
fitted: z.literal(true),
|
||||
}).strict(),
|
||||
|
||||
+72
-1170
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,22 @@
|
||||
import {
|
||||
type ActiveRun,
|
||||
type RunStatus,
|
||||
} from "./chatUiState.js";
|
||||
|
||||
export type ChatRunState = {
|
||||
activeRuns: Map<string, ActiveRun>;
|
||||
lastRunStatuses: Map<string, RunStatus>;
|
||||
getStatus: (sessionId: string) => RunStatus | undefined;
|
||||
};
|
||||
|
||||
export const createChatRunState = (): ChatRunState => {
|
||||
const activeRuns = new Map<string, ActiveRun>();
|
||||
const lastRunStatuses = new Map<string, RunStatus>();
|
||||
|
||||
return {
|
||||
activeRuns,
|
||||
lastRunStatuses,
|
||||
getStatus: (sessionId) =>
|
||||
activeRuns.get(sessionId)?.status ?? lastRunStatuses.get(sessionId),
|
||||
};
|
||||
};
|
||||
@@ -0,0 +1,426 @@
|
||||
import { type Router } from "express";
|
||||
import { z } from "zod";
|
||||
|
||||
import { type ChatSessionBridge } from "../chat/sessionBridge.js";
|
||||
import { getLocalAgentContext } from "../context/localContext.js";
|
||||
import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js";
|
||||
import { logger } from "../logger.js";
|
||||
import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
|
||||
import {
|
||||
type SessionMetadataStore,
|
||||
type SessionRecord,
|
||||
} from "../sessions/metadataStore.js";
|
||||
import { type SessionTranscriptStore } from "../sessions/transcriptStore.js";
|
||||
import { type SessionUiStateStore } from "../sessions/uiStateStore.js";
|
||||
import { toActorKey, toProjectKey } from "../utils/fileStore.js";
|
||||
import { type ChatRunState } from "./chatRunState.js";
|
||||
import { buildForkedSessionUiState } from "./chatSession.js";
|
||||
import { type StreamSubscriber } from "./chatUiState.js";
|
||||
|
||||
const createSessionPayloadSchema = z.object({
|
||||
session_id: z.string().max(128).optional(),
|
||||
parent_session_id: z.string().max(128).optional(),
|
||||
});
|
||||
|
||||
const forkPayloadSchema = z.object({
|
||||
session_id: z.string().max(128).optional(),
|
||||
keep_message_count: z.coerce.number().int().min(0),
|
||||
});
|
||||
|
||||
type RegisterChatSessionRoutesOptions = {
|
||||
frontendActionCoordinator: FrontendActionCoordinator;
|
||||
runState: ChatRunState;
|
||||
runtime: OpencodeRuntimeAdapter;
|
||||
sessionBridge: ChatSessionBridge;
|
||||
sessionMetadataStore: SessionMetadataStore;
|
||||
sessionTranscriptStore: SessionTranscriptStore;
|
||||
sessionUiStateStore: SessionUiStateStore;
|
||||
};
|
||||
|
||||
const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
|
||||
sessionId: sessionRecord.sessionId,
|
||||
});
|
||||
|
||||
const toSse = (event: string, data: Record<string, unknown>) =>
|
||||
`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
|
||||
|
||||
export const registerChatSessionRoutes = (
|
||||
chatRouter: Router,
|
||||
options: RegisterChatSessionRoutesOptions,
|
||||
) => {
|
||||
const {
|
||||
frontendActionCoordinator,
|
||||
runState,
|
||||
runtime,
|
||||
sessionBridge,
|
||||
sessionMetadataStore,
|
||||
sessionTranscriptStore,
|
||||
sessionUiStateStore,
|
||||
} = options;
|
||||
const { activeRuns, lastRunStatuses } = runState;
|
||||
const getSessionRunStatus = runState.getStatus;
|
||||
|
||||
chatRouter.post("/session", async (req, res) => {
|
||||
const parsed = createSessionPayloadSchema.safeParse(req.body ?? {});
|
||||
if (!parsed.success) {
|
||||
res.status(400).json({
|
||||
message: "invalid request payload",
|
||||
detail: parsed.error.flatten(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const userId = authContext.userId;
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
const requestedSessionId = parsed.data.session_id?.trim();
|
||||
let sessionId = requestedSessionId;
|
||||
if (!sessionId) {
|
||||
try {
|
||||
sessionId = (await runtime.createSession()).id;
|
||||
} catch (error) {
|
||||
sessionId = sessionBridge.createClientSessionId();
|
||||
logger.warn(
|
||||
{ err: error, sessionId },
|
||||
"runtime unavailable while creating chat session; using local degraded session",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const { record, created } = await sessionMetadataStore.ensure({
|
||||
actorKey,
|
||||
parentSessionId: parsed.data.parent_session_id,
|
||||
projectId,
|
||||
projectKey,
|
||||
sessionId,
|
||||
userId,
|
||||
});
|
||||
|
||||
res.status(created ? 201 : 200).json({
|
||||
session_id: record.sessionId,
|
||||
created_at: record.createdAt,
|
||||
updated_at: record.updatedAt,
|
||||
status: record.status,
|
||||
title: record.title,
|
||||
parent_session_id: record.parentSessionId,
|
||||
});
|
||||
});
|
||||
|
||||
chatRouter.get("/sessions", async (req, res) => {
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const userId = authContext.userId;
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
const records = await sessionMetadataStore.list({
|
||||
actorKey,
|
||||
projectId,
|
||||
projectKey,
|
||||
userId,
|
||||
});
|
||||
res.json({
|
||||
sessions: records.map((record) => ({
|
||||
id: record.sessionId,
|
||||
title: record.title ?? "新对话",
|
||||
created_at: record.createdAt,
|
||||
updated_at: record.updatedAt,
|
||||
status: record.status,
|
||||
parent_session_id: record.parentSessionId,
|
||||
is_streaming: activeRuns.get(record.sessionId)?.status === "running",
|
||||
run_status: getSessionRunStatus(record.sessionId),
|
||||
})),
|
||||
});
|
||||
});
|
||||
|
||||
chatRouter.get("/session/:session_id", async (req, res) => {
|
||||
const sessionId = req.params.session_id?.trim();
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const userId = authContext.userId;
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
if (!sessionId) {
|
||||
res.status(400).json({ message: "session_id is required" });
|
||||
return;
|
||||
}
|
||||
|
||||
const sessionRecord = await sessionMetadataStore.get(
|
||||
{
|
||||
actorKey,
|
||||
projectId,
|
||||
projectKey,
|
||||
userId,
|
||||
},
|
||||
sessionId,
|
||||
);
|
||||
if (!sessionRecord) {
|
||||
res.status(404).json({ message: "session not found" });
|
||||
return;
|
||||
}
|
||||
|
||||
const state = await sessionUiStateStore.read(
|
||||
toSessionUiStateContext(sessionRecord),
|
||||
);
|
||||
res.json({
|
||||
id: sessionRecord.sessionId,
|
||||
title: sessionRecord.title ?? "新对话",
|
||||
is_title_manually_edited: state?.isTitleManuallyEdited ?? false,
|
||||
created_at: sessionRecord.createdAt,
|
||||
updated_at: sessionRecord.updatedAt,
|
||||
status: sessionRecord.status,
|
||||
session_id: sessionRecord.sessionId,
|
||||
messages: state?.messages ?? [],
|
||||
parent_session_id: sessionRecord.parentSessionId,
|
||||
is_streaming: activeRuns.get(sessionRecord.sessionId)?.status === "running",
|
||||
run_status: getSessionRunStatus(sessionRecord.sessionId),
|
||||
});
|
||||
});
|
||||
|
||||
chatRouter.get("/session/:session_id/stream", async (req, res) => {
|
||||
const sessionId = req.params.session_id?.trim();
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const userId = authContext.userId;
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
if (!sessionId) {
|
||||
res.status(400).json({ message: "session_id is required" });
|
||||
return;
|
||||
}
|
||||
|
||||
const sessionRecord = await sessionMetadataStore.get(
|
||||
{ actorKey, projectId, projectKey, userId },
|
||||
sessionId,
|
||||
);
|
||||
if (!sessionRecord) {
|
||||
res.status(404).json({ message: "session not found" });
|
||||
return;
|
||||
}
|
||||
|
||||
res.status(200);
|
||||
res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
|
||||
res.setHeader("Cache-Control", "no-cache");
|
||||
res.setHeader("Connection", "keep-alive");
|
||||
res.setHeader("X-Accel-Buffering", "no");
|
||||
res.flushHeaders?.();
|
||||
|
||||
const run = activeRuns.get(sessionRecord.sessionId);
|
||||
const state = await sessionUiStateStore.read(toSessionUiStateContext(sessionRecord));
|
||||
res.write(
|
||||
toSse("state", {
|
||||
session_id: sessionRecord.sessionId,
|
||||
messages: state?.messages ?? run?.messages ?? [],
|
||||
is_streaming: run?.status === "running",
|
||||
run_status: getSessionRunStatus(sessionRecord.sessionId) ?? "completed",
|
||||
}),
|
||||
);
|
||||
|
||||
if (!run || run.status !== "running") {
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
|
||||
const subscriber: StreamSubscriber = {
|
||||
write: (event, data) => {
|
||||
if (!res.writableEnded && !res.destroyed) {
|
||||
res.write(toSse(event, data));
|
||||
}
|
||||
},
|
||||
close: () => {
|
||||
if (!res.writableEnded && !res.destroyed) {
|
||||
res.end();
|
||||
}
|
||||
},
|
||||
};
|
||||
run.subscribers.add(subscriber);
|
||||
|
||||
const cleanup = () => {
|
||||
run.subscribers.delete(subscriber);
|
||||
};
|
||||
req.on("close", cleanup);
|
||||
res.on("close", cleanup);
|
||||
});
|
||||
|
||||
chatRouter.patch("/session/:session_id/title", async (req, res) => {
|
||||
const sessionId = req.params.session_id?.trim();
|
||||
const title =
|
||||
typeof req.body?.title === "string" ? req.body.title.trim() : "";
|
||||
const isTitleManuallyEdited =
|
||||
typeof req.body?.is_title_manually_edited === "boolean"
|
||||
? req.body.is_title_manually_edited
|
||||
: undefined;
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const userId = authContext.userId;
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
if (!sessionId || !title) {
|
||||
res.status(400).json({ message: "session_id and title are required" });
|
||||
return;
|
||||
}
|
||||
const sessionRecord = await sessionMetadataStore.get(
|
||||
{ actorKey, projectId, projectKey, userId },
|
||||
sessionId,
|
||||
);
|
||||
if (!sessionRecord) {
|
||||
res.status(404).json({ message: "session not found" });
|
||||
return;
|
||||
}
|
||||
const nextSessionRecord = await sessionMetadataStore.touch(sessionRecord, {
|
||||
title,
|
||||
});
|
||||
const state = await sessionUiStateStore.read(
|
||||
toSessionUiStateContext(nextSessionRecord),
|
||||
);
|
||||
if (state) {
|
||||
await sessionUiStateStore.write(
|
||||
toSessionUiStateContext(nextSessionRecord),
|
||||
{
|
||||
...state,
|
||||
isTitleManuallyEdited:
|
||||
isTitleManuallyEdited ?? state.isTitleManuallyEdited,
|
||||
},
|
||||
);
|
||||
}
|
||||
res.json({
|
||||
id: nextSessionRecord.sessionId,
|
||||
title: nextSessionRecord.title,
|
||||
updated_at: nextSessionRecord.updatedAt,
|
||||
});
|
||||
});
|
||||
|
||||
chatRouter.delete("/session/:session_id", async (req, res) => {
|
||||
const sessionId = req.params.session_id?.trim();
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const userId = authContext.userId;
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
if (!sessionId) {
|
||||
res.status(400).json({ message: "session_id is required" });
|
||||
return;
|
||||
}
|
||||
const sessionRecord = await sessionMetadataStore.get(
|
||||
{ actorKey, projectId, projectKey, userId },
|
||||
sessionId,
|
||||
);
|
||||
if (!sessionRecord) {
|
||||
res.status(204).end();
|
||||
return;
|
||||
}
|
||||
await sessionUiStateStore.remove(toSessionUiStateContext(sessionRecord));
|
||||
await sessionBridge.deleteSession({
|
||||
clientSessionId: sessionRecord.sessionId,
|
||||
sessionId: sessionRecord.sessionId,
|
||||
});
|
||||
frontendActionCoordinator.cancelSession(sessionRecord.sessionId);
|
||||
activeRuns.delete(sessionRecord.sessionId);
|
||||
lastRunStatuses.delete(sessionRecord.sessionId);
|
||||
await sessionMetadataStore.remove(sessionRecord);
|
||||
res.status(204).end();
|
||||
});
|
||||
|
||||
chatRouter.post("/fork", async (req, res) => {
|
||||
const parsed = forkPayloadSchema.safeParse(req.body);
|
||||
if (!parsed.success) {
|
||||
res.status(400).json({
|
||||
message: "invalid request payload",
|
||||
detail: parsed.error.flatten(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const traceId = req.header("x-trace-id") ?? undefined;
|
||||
const userId = authContext.userId;
|
||||
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
const sourceSessionId = parsed.data.session_id?.trim();
|
||||
const sourceSessionRecord = sourceSessionId
|
||||
? await sessionMetadataStore.get(
|
||||
{
|
||||
actorKey,
|
||||
projectId,
|
||||
projectKey,
|
||||
userId,
|
||||
},
|
||||
sourceSessionId,
|
||||
)
|
||||
: null;
|
||||
const forkSession = await runtime.createSession();
|
||||
const { record: targetSessionRecord } = await sessionMetadataStore.ensure({
|
||||
actorKey,
|
||||
parentSessionId: sourceSessionId,
|
||||
projectId,
|
||||
projectKey,
|
||||
sessionId: forkSession.id,
|
||||
userId,
|
||||
});
|
||||
const nextSessionId = targetSessionRecord.sessionId;
|
||||
|
||||
if (sourceSessionId) {
|
||||
await sessionTranscriptStore.cloneThread(
|
||||
{
|
||||
actorKey,
|
||||
clientSessionId: sourceSessionId,
|
||||
projectKey,
|
||||
sessionId: sourceSessionId,
|
||||
},
|
||||
{
|
||||
actorKey,
|
||||
clientSessionId: nextSessionId,
|
||||
projectKey,
|
||||
sessionId: nextSessionId,
|
||||
},
|
||||
parsed.data.keep_message_count,
|
||||
);
|
||||
}
|
||||
const sourceState = sourceSessionRecord
|
||||
? await sessionUiStateStore.read(
|
||||
toSessionUiStateContext(sourceSessionRecord),
|
||||
)
|
||||
: null;
|
||||
const forkTitle = sourceSessionRecord?.title
|
||||
? `${sourceSessionRecord.title} 副本`
|
||||
: "新对话副本";
|
||||
const titledTargetSessionRecord = await sessionMetadataStore.touch(
|
||||
targetSessionRecord,
|
||||
{ title: forkTitle },
|
||||
);
|
||||
await sessionUiStateStore.write(
|
||||
toSessionUiStateContext(titledTargetSessionRecord),
|
||||
buildForkedSessionUiState(sourceState, {
|
||||
keepMessageCount: parsed.data.keep_message_count,
|
||||
targetSessionId: nextSessionId,
|
||||
}),
|
||||
);
|
||||
|
||||
logger.info(
|
||||
{
|
||||
sourceSessionId: parsed.data.session_id,
|
||||
sessionId: nextSessionId,
|
||||
traceId,
|
||||
projectId,
|
||||
keepMessageCount: parsed.data.keep_message_count,
|
||||
},
|
||||
"forked chat session",
|
||||
);
|
||||
|
||||
res.status(200).json({
|
||||
session_id: nextSessionId,
|
||||
});
|
||||
} catch (error) {
|
||||
const detail = error instanceof Error ? error.message : String(error);
|
||||
logger.error({ err: error }, "chat fork failed");
|
||||
res.status(500).json({
|
||||
message: "chat fork failed",
|
||||
detail,
|
||||
});
|
||||
}
|
||||
});
|
||||
};
|
||||
@@ -0,0 +1,811 @@
|
||||
import { type Router } from "express";
|
||||
import {
|
||||
UI_MESSAGE_STREAM_HEADERS,
|
||||
type UIMessage,
|
||||
type UIMessageChunk,
|
||||
} from "ai";
|
||||
import { z } from "zod";
|
||||
|
||||
import { isSupportedModel } from "../chat/models.js";
|
||||
import { type ChatSessionBridge } from "../chat/sessionBridge.js";
|
||||
import { getLocalAgentContext } from "../context/localContext.js";
|
||||
import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js";
|
||||
import { type LearningOrchestrator } from "../learning/orchestrator.js";
|
||||
import { logger } from "../logger.js";
|
||||
import { type MemoryStore } from "../memory/store.js";
|
||||
import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
|
||||
import {
|
||||
getRuntimeSessionContext,
|
||||
setRuntimeSessionContext,
|
||||
} from "../runtime/sessionContext.js";
|
||||
import {
|
||||
type SessionMetadataStore,
|
||||
type SessionRecord,
|
||||
} from "../sessions/metadataStore.js";
|
||||
import { type SessionTranscriptStore } from "../sessions/transcriptStore.js";
|
||||
import { type SessionUiStateStore } from "../sessions/uiStateStore.js";
|
||||
import type { UIEnvelopePayload } from "../uiEnvelope/types.js";
|
||||
import { toActorKey, toProjectKey } from "../utils/fileStore.js";
|
||||
import {
|
||||
buildPromptWithLearningContext,
|
||||
extractLatestFrontendTurn,
|
||||
generateSessionTitle,
|
||||
resolveRequestedStreamSessionId,
|
||||
shouldGenerateSessionTitle,
|
||||
shouldRestoreConversationForRuntime,
|
||||
} from "./chatSession.js";
|
||||
import {
|
||||
collectTextContent,
|
||||
type PermissionRequestPayload,
|
||||
type QuestionRequestPayload,
|
||||
streamPromptResponse,
|
||||
type TodoUpdatePayload,
|
||||
} from "./chatStream.js";
|
||||
import { type ChatRunState } from "./chatRunState.js";
|
||||
import {
|
||||
type ActiveRun,
|
||||
type StreamSubscriber,
|
||||
appendBackendToolArtifact,
|
||||
appendBackendUiEnvelope,
|
||||
cancelBackendTodos,
|
||||
completeBackendProgress,
|
||||
createInitialStreamingMessages,
|
||||
isObjectRecord,
|
||||
toFrontendPermission,
|
||||
toPermissionStatus,
|
||||
updateLastAssistantMessage,
|
||||
updateLastAssistantPermission,
|
||||
updateLastAssistantQuestion,
|
||||
upsertBackendProgress,
|
||||
upsertBackendQuestion,
|
||||
upsertBackendTodoUpdate,
|
||||
} from "./chatUiState.js";
|
||||
|
||||
const payloadSchema = z.object({
|
||||
id: z.string().max(128).optional(),
|
||||
messages: z.array(z.unknown()).max(200).optional(),
|
||||
message: z.string().min(1).max(10000).optional(),
|
||||
session_id: z.string().max(128).optional(),
|
||||
model: z.string().refine(isSupportedModel, {
|
||||
message: "unsupported model",
|
||||
}).optional(),
|
||||
approval_mode: z.enum(["request", "always"]).optional().default("request"),
|
||||
capabilities: z.array(z.string().max(64)).max(20).optional().default([]),
|
||||
});
|
||||
|
||||
const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
|
||||
sessionId: sessionRecord.sessionId,
|
||||
});
|
||||
|
||||
const runtimeHasConversation = async (
|
||||
runtime: OpencodeRuntimeAdapter,
|
||||
sessionId: string,
|
||||
) => {
|
||||
const messages = await runtime.messages(sessionId, 1);
|
||||
return messages.some(
|
||||
(message) =>
|
||||
message.info.role === "user" || message.info.role === "assistant",
|
||||
);
|
||||
};
|
||||
|
||||
const isUiMessage = (value: unknown): value is UIMessage =>
|
||||
isObjectRecord(value) &&
|
||||
typeof value.id === "string" &&
|
||||
(value.role === "system" || value.role === "user" || value.role === "assistant") &&
|
||||
Array.isArray(value.parts);
|
||||
|
||||
const readUiMessageText = (message: UIMessage | undefined) =>
|
||||
message?.parts
|
||||
.flatMap((part) =>
|
||||
isObjectRecord(part) && part.type === "text" && typeof part.text === "string"
|
||||
? [part.text]
|
||||
: [],
|
||||
)
|
||||
.join("")
|
||||
.trim() ?? "";
|
||||
|
||||
const readLatestUserText = (messages: UIMessage[]) => {
|
||||
for (let index = messages.length - 1; index >= 0; index -= 1) {
|
||||
const message = messages[index];
|
||||
if (message.role !== "user") {
|
||||
continue;
|
||||
}
|
||||
const text = readUiMessageText(message);
|
||||
if (text) {
|
||||
return text;
|
||||
}
|
||||
}
|
||||
return "";
|
||||
};
|
||||
|
||||
const toUiMessages = (messages: unknown[] | undefined) =>
|
||||
Array.isArray(messages) ? messages.filter(isUiMessage) : [];
|
||||
|
||||
const toLegacyMessages = (messages: UIMessage[]) =>
|
||||
messages.flatMap((message) => {
|
||||
const content = readUiMessageText(message);
|
||||
if (!content || (message.role !== "user" && message.role !== "assistant")) {
|
||||
return [];
|
||||
}
|
||||
return [
|
||||
{
|
||||
id: message.id,
|
||||
role: message.role,
|
||||
content,
|
||||
},
|
||||
];
|
||||
});
|
||||
|
||||
const createFallbackUserUiMessage = (content: string): UIMessage => ({
|
||||
id: `user-${Date.now().toString(36)}`,
|
||||
role: "user",
|
||||
parts: [{ type: "text", text: content }],
|
||||
});
|
||||
|
||||
const pipeUiChunk = (
|
||||
res: {
|
||||
write: (chunk: string) => void;
|
||||
writableEnded?: boolean;
|
||||
destroyed?: boolean;
|
||||
},
|
||||
chunk: UIMessageChunk,
|
||||
) => {
|
||||
if (!res.writableEnded && !res.destroyed) {
|
||||
res.write(`data: ${JSON.stringify(chunk)}\n\n`);
|
||||
}
|
||||
};
|
||||
|
||||
type RegisterChatStreamRouteOptions = {
|
||||
frontendActionCoordinator: FrontendActionCoordinator;
|
||||
learningOrchestrator: LearningOrchestrator;
|
||||
memoryStore: MemoryStore;
|
||||
runState: ChatRunState;
|
||||
runtime: OpencodeRuntimeAdapter;
|
||||
sessionBridge: ChatSessionBridge;
|
||||
sessionMetadataStore: SessionMetadataStore;
|
||||
sessionTranscriptStore: SessionTranscriptStore;
|
||||
sessionUiStateStore: SessionUiStateStore;
|
||||
};
|
||||
|
||||
export const registerChatStreamRoute = (
|
||||
chatRouter: Router,
|
||||
{
|
||||
frontendActionCoordinator,
|
||||
learningOrchestrator,
|
||||
memoryStore,
|
||||
runState,
|
||||
runtime,
|
||||
sessionBridge,
|
||||
sessionMetadataStore,
|
||||
sessionTranscriptStore,
|
||||
sessionUiStateStore,
|
||||
}: RegisterChatStreamRouteOptions,
|
||||
) => {
|
||||
const { activeRuns, lastRunStatuses } = runState;
|
||||
chatRouter.post("/stream", async (req, res) => {
|
||||
const parsed = payloadSchema.safeParse(req.body);
|
||||
if (!parsed.success) {
|
||||
res.status(400).json({
|
||||
message: "invalid request payload",
|
||||
detail: parsed.error.flatten(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const authContext = getLocalAgentContext(req);
|
||||
const projectId = authContext.projectId;
|
||||
const traceId = req.header("x-trace-id") ?? undefined;
|
||||
const userId = authContext.userId;
|
||||
const actorKey = toActorKey(userId);
|
||||
const projectKey = toProjectKey(projectId);
|
||||
const requestUiMessages = toUiMessages(parsed.data.messages);
|
||||
const promptText =
|
||||
parsed.data.message?.trim() || readLatestUserText(requestUiMessages);
|
||||
if (!promptText) {
|
||||
res.status(400).json({
|
||||
message: "message or messages with latest user text is required",
|
||||
});
|
||||
return;
|
||||
}
|
||||
const requestedSessionId = resolveRequestedStreamSessionId({
|
||||
sessionId: parsed.data.session_id,
|
||||
promptText,
|
||||
createClientSessionId: () => sessionBridge.createClientSessionId(),
|
||||
});
|
||||
const existingSessionRecord = requestedSessionId
|
||||
? await sessionMetadataStore.get(
|
||||
{ actorKey, projectId, projectKey, userId },
|
||||
requestedSessionId,
|
||||
)
|
||||
: null;
|
||||
const hadExistingRuntimeSession = Boolean(existingSessionRecord);
|
||||
|
||||
const { binding, requestContext, created } = await sessionBridge.resolve({
|
||||
sessionId: requestedSessionId,
|
||||
network: authContext.network,
|
||||
projectId,
|
||||
traceId,
|
||||
userId,
|
||||
});
|
||||
setRuntimeSessionContext({
|
||||
...(getRuntimeSessionContext(binding.sessionId)!),
|
||||
capabilities: parsed.data.capabilities,
|
||||
});
|
||||
const { record: ensuredSessionRecord, created: sessionCreated } =
|
||||
await sessionMetadataStore.ensure({
|
||||
actorKey,
|
||||
projectId,
|
||||
projectKey,
|
||||
sessionId: binding.sessionId,
|
||||
userId,
|
||||
});
|
||||
const activeSessionRecord = await sessionMetadataStore.touch(ensuredSessionRecord);
|
||||
const hasRuntimeConversation = hadExistingRuntimeSession
|
||||
? await runtimeHasConversation(runtime, binding.sessionId)
|
||||
: false;
|
||||
const shouldRestoreConversation = shouldRestoreConversationForRuntime({
|
||||
hadExistingSessionRecord: hadExistingRuntimeSession,
|
||||
runtimeHasConversation: hasRuntimeConversation,
|
||||
});
|
||||
const historyContext = {
|
||||
actorKey: requestContext.actorKey,
|
||||
clientSessionId: requestContext.clientSessionId,
|
||||
projectKey: requestContext.projectKey,
|
||||
sessionId: requestContext.clientSessionId,
|
||||
};
|
||||
const initialSessionState = await sessionUiStateStore.read(
|
||||
toSessionUiStateContext(activeSessionRecord),
|
||||
);
|
||||
const persistedUiMessages = toUiMessages(initialSessionState?.messages);
|
||||
const requestMessages =
|
||||
requestUiMessages.length > 0
|
||||
? requestUiMessages
|
||||
: [
|
||||
...persistedUiMessages,
|
||||
createFallbackUserUiMessage(promptText),
|
||||
];
|
||||
let latestUiMessages = requestMessages;
|
||||
const baseMessages = toLegacyMessages(requestMessages);
|
||||
if (activeRuns.get(activeSessionRecord.sessionId)?.status === "running") {
|
||||
res.status(409).json({
|
||||
message: "session is already streaming",
|
||||
session_id: activeSessionRecord.sessionId,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8);
|
||||
|
||||
logger.info(
|
||||
{
|
||||
clientSessionId: requestContext.clientSessionId,
|
||||
sessionId: binding.sessionId,
|
||||
created: created || sessionCreated,
|
||||
model: parsed.data.model,
|
||||
approvalMode: parsed.data.approval_mode,
|
||||
traceId: requestContext.traceId,
|
||||
projectId: requestContext.projectId,
|
||||
},
|
||||
"processing chat request",
|
||||
);
|
||||
|
||||
res.status(200);
|
||||
for (const [header, value] of Object.entries(UI_MESSAGE_STREAM_HEADERS)) {
|
||||
res.setHeader(header, value);
|
||||
}
|
||||
res.setHeader("X-Accel-Buffering", "no");
|
||||
res.flushHeaders?.();
|
||||
|
||||
const clientSessionId = requestContext.clientSessionId;
|
||||
let streamClosed = false;
|
||||
const abortController = new AbortController();
|
||||
sessionBridge.registerAbortController(clientSessionId, abortController);
|
||||
const initialMessages = createInitialStreamingMessages(
|
||||
baseMessages,
|
||||
promptText,
|
||||
);
|
||||
const activeRun: ActiveRun = {
|
||||
clientSessionId,
|
||||
controller: abortController,
|
||||
messages: initialMessages,
|
||||
pendingPermissions: new Map(),
|
||||
pendingQuestions: new Map(),
|
||||
status: "running",
|
||||
subscribers: new Set(),
|
||||
};
|
||||
activeRuns.set(clientSessionId, activeRun);
|
||||
lastRunStatuses.set(clientSessionId, "running");
|
||||
const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord);
|
||||
const assistantUiMessageId = `assistant-${Date.now().toString(36)}`;
|
||||
const textPartId = "answer";
|
||||
const assistantParts: Record<string, unknown>[] = [];
|
||||
let responseText = "";
|
||||
let textStarted = false;
|
||||
let textEnded = false;
|
||||
let uiFinished = false;
|
||||
const syncLatestUiMessages = () => {
|
||||
latestUiMessages = [
|
||||
...requestMessages,
|
||||
{
|
||||
id: assistantUiMessageId,
|
||||
role: "assistant",
|
||||
parts: assistantParts.map((part) => ({ ...part })),
|
||||
} as UIMessage,
|
||||
];
|
||||
};
|
||||
const writeUiChunk = (chunk: UIMessageChunk) => {
|
||||
pipeUiChunk(res, chunk);
|
||||
};
|
||||
const writeTextDelta = (delta: string) => {
|
||||
if (!delta || uiFinished) {
|
||||
return;
|
||||
}
|
||||
if (!textStarted) {
|
||||
textStarted = true;
|
||||
assistantParts.push({
|
||||
type: "text",
|
||||
text: "",
|
||||
state: "streaming",
|
||||
});
|
||||
writeUiChunk({ type: "text-start", id: textPartId });
|
||||
}
|
||||
responseText += delta;
|
||||
const textPart = assistantParts.find(
|
||||
(part) => part.type === "text",
|
||||
);
|
||||
if (textPart) {
|
||||
textPart.text = responseText;
|
||||
textPart.state = "streaming";
|
||||
}
|
||||
writeUiChunk({ type: "text-delta", id: textPartId, delta });
|
||||
writeUiChunk({
|
||||
type: "data-stream_token",
|
||||
data: {
|
||||
session_id: clientSessionId,
|
||||
message_id: assistantUiMessageId,
|
||||
content: delta,
|
||||
},
|
||||
transient: true,
|
||||
} as UIMessageChunk);
|
||||
syncLatestUiMessages();
|
||||
};
|
||||
const writeDataPart = (event: string, data: Record<string, unknown>) => {
|
||||
if (uiFinished) {
|
||||
return;
|
||||
}
|
||||
const type = `data-${event}`;
|
||||
const id =
|
||||
typeof data.id === "string"
|
||||
? data.id
|
||||
: typeof data.request_id === "string"
|
||||
? data.request_id
|
||||
: typeof data.envelope_id === "string"
|
||||
? data.envelope_id
|
||||
: undefined;
|
||||
const existingIndex =
|
||||
id === undefined
|
||||
? -1
|
||||
: assistantParts.findIndex(
|
||||
(part) => part.type === type && part.id === id,
|
||||
);
|
||||
const part = id ? { type, id, data } : { type, data };
|
||||
if (existingIndex >= 0) {
|
||||
assistantParts[existingIndex] = part;
|
||||
} else {
|
||||
assistantParts.push(part);
|
||||
}
|
||||
writeUiChunk(part as UIMessageChunk);
|
||||
syncLatestUiMessages();
|
||||
};
|
||||
const finishUiMessage = (finishReason: "stop" | "error" = "stop") => {
|
||||
if (uiFinished) {
|
||||
return;
|
||||
}
|
||||
if (!textStarted && assistantParts.length === 0) {
|
||||
writeTextDelta("Agent 已完成处理。");
|
||||
}
|
||||
if (textStarted && !textEnded) {
|
||||
textEnded = true;
|
||||
const textPart = assistantParts.find(
|
||||
(part) => part.type === "text",
|
||||
);
|
||||
if (textPart) {
|
||||
textPart.state = "done";
|
||||
}
|
||||
writeUiChunk({ type: "text-end", id: textPartId });
|
||||
}
|
||||
syncLatestUiMessages();
|
||||
writeUiChunk({ type: "finish", finishReason });
|
||||
uiFinished = true;
|
||||
};
|
||||
writeUiChunk({ type: "start", messageId: assistantUiMessageId });
|
||||
let persistQueue = sessionUiStateStore.write(sessionUiStateContext, {
|
||||
sessionId: activeSessionRecord.sessionId,
|
||||
isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
|
||||
messages: latestUiMessages,
|
||||
});
|
||||
const queueSessionUiStatePersist = () => {
|
||||
const snapshot = {
|
||||
sessionId: activeSessionRecord.sessionId,
|
||||
isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
|
||||
messages: latestUiMessages,
|
||||
};
|
||||
persistQueue = persistQueue
|
||||
.catch((error) => {
|
||||
logger.warn(
|
||||
{ err: error, sessionId: clientSessionId },
|
||||
"failed to persist previous chat stream state",
|
||||
);
|
||||
})
|
||||
.then(() => sessionUiStateStore.write(sessionUiStateContext, snapshot));
|
||||
return persistQueue;
|
||||
};
|
||||
const primarySubscriber: StreamSubscriber = {
|
||||
write: (event, data) => {
|
||||
if (streamClosed || res.writableEnded || res.destroyed) {
|
||||
return;
|
||||
}
|
||||
if (event === "token") {
|
||||
writeTextDelta(typeof data.content === "string" ? data.content : "");
|
||||
return;
|
||||
}
|
||||
if (event === "done") {
|
||||
finishUiMessage("stop");
|
||||
return;
|
||||
}
|
||||
if (event === "error") {
|
||||
writeUiChunk({
|
||||
type: "error",
|
||||
errorText:
|
||||
typeof data.message === "string"
|
||||
? data.message
|
||||
: "Agent stream failed",
|
||||
});
|
||||
finishUiMessage("error");
|
||||
return;
|
||||
}
|
||||
writeDataPart(event, data);
|
||||
},
|
||||
close: () => {
|
||||
if (!res.writableEnded && !res.destroyed) {
|
||||
res.end();
|
||||
}
|
||||
},
|
||||
};
|
||||
activeRun.subscribers.add(primarySubscriber);
|
||||
const handleClientClose = () => {
|
||||
streamClosed = true;
|
||||
activeRun.subscribers.delete(primarySubscriber);
|
||||
};
|
||||
|
||||
req.on("close", handleClientClose);
|
||||
res.on("close", handleClientClose);
|
||||
|
||||
const publish = (event: string, data: Record<string, unknown>) => {
|
||||
const subscriberData =
|
||||
event === "token"
|
||||
? {
|
||||
...data,
|
||||
message_id:
|
||||
typeof data.message_id === "string"
|
||||
? data.message_id
|
||||
: assistantUiMessageId,
|
||||
}
|
||||
: data;
|
||||
if (event === "token") {
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
content: `${typeof message.content === "string" ? message.content : ""}${typeof data.content === "string" ? data.content : ""}`,
|
||||
isError: false,
|
||||
}));
|
||||
} else if (event === "progress") {
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
progress: upsertBackendProgress(message.progress, data),
|
||||
}));
|
||||
} else if (event === "done") {
|
||||
activeRun.status = "completed";
|
||||
lastRunStatuses.set(clientSessionId, "completed");
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
content:
|
||||
typeof message.content === "string" && message.content.trim()
|
||||
? message.content
|
||||
: "Agent 已完成处理,但没有生成文本回答。请查看过程记录,或换个更具体的问题重试。",
|
||||
progress: completeBackendProgress(message.progress),
|
||||
}));
|
||||
} else if (event === "error") {
|
||||
activeRun.status = activeRun.status === "aborted" ? "aborted" : "error";
|
||||
lastRunStatuses.set(clientSessionId, activeRun.status);
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
content:
|
||||
typeof message.content === "string" && message.content.trim()
|
||||
? message.content
|
||||
: `⚠️ **错误:** ${typeof data.message === "string" ? data.message : "unknown error"}`,
|
||||
isError: true,
|
||||
progress: completeBackendProgress(message.progress),
|
||||
todos: cancelBackendTodos(message.todos),
|
||||
}));
|
||||
} else if (event === "permission_request") {
|
||||
const payload = data as PermissionRequestPayload;
|
||||
activeRun.pendingPermissions.set(payload.request_id, payload);
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
permissions: [
|
||||
...(Array.isArray(message.permissions) ? message.permissions : []),
|
||||
toFrontendPermission(payload),
|
||||
],
|
||||
}));
|
||||
} else if (event === "permission_response") {
|
||||
const requestId =
|
||||
typeof data.request_id === "string" ? data.request_id : undefined;
|
||||
const reply =
|
||||
data.reply === "once" || data.reply === "always" || data.reply === "reject"
|
||||
? data.reply
|
||||
: undefined;
|
||||
if (requestId && reply) {
|
||||
activeRun.pendingPermissions.delete(requestId);
|
||||
activeRun.messages = updateLastAssistantPermission(
|
||||
activeRun.messages,
|
||||
requestId,
|
||||
(permission) => ({
|
||||
...permission,
|
||||
status: toPermissionStatus(reply),
|
||||
repliedAt: Date.now(),
|
||||
}),
|
||||
);
|
||||
}
|
||||
} else if (event === "question_request") {
|
||||
const payload = data as QuestionRequestPayload;
|
||||
let shouldTrackQuestion = true;
|
||||
if (payload.tool?.callID) {
|
||||
if (payload.request_id !== payload.tool.callID) {
|
||||
activeRun.pendingQuestions.delete(payload.tool.callID);
|
||||
} else {
|
||||
const hasActionableQuestion = [...activeRun.pendingQuestions.values()].some(
|
||||
(question) =>
|
||||
question.tool?.callID === payload.tool?.callID &&
|
||||
question.request_id !== payload.tool?.callID,
|
||||
);
|
||||
if (hasActionableQuestion) {
|
||||
activeRun.messages = updateLastAssistantMessage(
|
||||
activeRun.messages,
|
||||
(message) => ({
|
||||
...message,
|
||||
questions: upsertBackendQuestion(message.questions, payload),
|
||||
}),
|
||||
);
|
||||
shouldTrackQuestion = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (shouldTrackQuestion) {
|
||||
activeRun.pendingQuestions.set(payload.request_id, payload);
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
questions: upsertBackendQuestion(message.questions, payload),
|
||||
}));
|
||||
}
|
||||
} else if (event === "question_response") {
|
||||
const requestId =
|
||||
typeof data.request_id === "string" ? data.request_id : undefined;
|
||||
if (requestId) {
|
||||
activeRun.pendingQuestions.delete(requestId);
|
||||
activeRun.messages = updateLastAssistantQuestion(
|
||||
activeRun.messages,
|
||||
requestId,
|
||||
(question) => ({
|
||||
...question,
|
||||
status: data.rejected === true ? "rejected" : "answered",
|
||||
repliedAt: Date.now(),
|
||||
answers: Array.isArray(data.answers) ? data.answers : question.answers,
|
||||
error: undefined,
|
||||
}),
|
||||
);
|
||||
}
|
||||
} else if (event === "todo_update") {
|
||||
const payload = data as TodoUpdatePayload;
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
todos: upsertBackendTodoUpdate(message.todos, payload),
|
||||
}));
|
||||
} else if (event === "tool_call") {
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
artifacts: appendBackendToolArtifact(message.artifacts, data),
|
||||
}));
|
||||
} else if (event === "ui_envelope") {
|
||||
const payload = data as UIEnvelopePayload;
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
uiEnvelopes: appendBackendUiEnvelope(message.uiEnvelopes, payload),
|
||||
}));
|
||||
}
|
||||
|
||||
for (const subscriber of activeRun.subscribers) {
|
||||
subscriber.write(event, subscriberData);
|
||||
}
|
||||
void queueSessionUiStatePersist().catch((error) => {
|
||||
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
|
||||
});
|
||||
};
|
||||
|
||||
const unregisterFrontendActionSink = parsed.data.capabilities.includes("frontend-action@1")
|
||||
? frontendActionCoordinator.registerSink(clientSessionId, (request) => publish("frontend_action", request as unknown as Record<string, unknown>))
|
||||
: undefined;
|
||||
|
||||
try {
|
||||
const preparedMessage = await buildPromptWithLearningContext(
|
||||
memoryStore,
|
||||
requestContext.actorKey,
|
||||
requestContext.projectKey,
|
||||
{
|
||||
recentTurns,
|
||||
persistedMessages: baseMessages,
|
||||
message: promptText,
|
||||
restoreConversation: shouldRestoreConversation,
|
||||
},
|
||||
);
|
||||
const streamResult = await streamPromptResponse({
|
||||
runtime,
|
||||
sessionId: binding.sessionId,
|
||||
clientSessionId,
|
||||
message: preparedMessage,
|
||||
model: parsed.data.model,
|
||||
approvalMode: parsed.data.approval_mode,
|
||||
traceId: requestContext.traceId,
|
||||
projectId: requestContext.projectId,
|
||||
signal: abortController.signal,
|
||||
suppressLegacyFrontendActions: parsed.data.capabilities.includes("frontend-action@1"),
|
||||
write: (event, data) => {
|
||||
publish(event, data);
|
||||
},
|
||||
});
|
||||
await persistQueue.catch((error) => {
|
||||
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
|
||||
});
|
||||
|
||||
if (!streamResult.aborted && !streamResult.failed) {
|
||||
const messages = await runtime.messages(binding.sessionId, 60);
|
||||
const assistantMessage = [...messages]
|
||||
.reverse()
|
||||
.find((message) => message.info.role === "assistant");
|
||||
const assistantText = collectTextContent(assistantMessage?.parts ?? []);
|
||||
const latestSessionRecord =
|
||||
(await sessionMetadataStore.get(
|
||||
{ actorKey, projectId, projectKey, userId },
|
||||
activeSessionRecord.sessionId,
|
||||
)) ?? activeSessionRecord;
|
||||
const latestSessionState = await sessionUiStateStore.read(
|
||||
toSessionUiStateContext(latestSessionRecord),
|
||||
);
|
||||
const existingSessionTitle = latestSessionRecord.title;
|
||||
let sessionTitle = existingSessionTitle;
|
||||
const shouldGenerateTitle = shouldGenerateSessionTitle({
|
||||
recentTurnCount: recentTurns.length,
|
||||
isTitleManuallyEdited:
|
||||
latestSessionState?.isTitleManuallyEdited ?? false,
|
||||
});
|
||||
if (shouldGenerateTitle) {
|
||||
sessionTitle = await generateSessionTitle(runtime, {
|
||||
sessionId: binding.sessionId,
|
||||
latestAssistantMessage: assistantText,
|
||||
latestUserMessage: promptText,
|
||||
fallbackTitle: existingSessionTitle,
|
||||
});
|
||||
}
|
||||
await sessionMetadataStore.touch(latestSessionRecord, {
|
||||
...(sessionTitle && sessionTitle !== existingSessionTitle
|
||||
? { title: sessionTitle }
|
||||
: {}),
|
||||
});
|
||||
if (
|
||||
shouldGenerateTitle &&
|
||||
sessionTitle &&
|
||||
sessionTitle !== existingSessionTitle
|
||||
) {
|
||||
publish("session_title", {
|
||||
session_id: clientSessionId,
|
||||
title: sessionTitle,
|
||||
});
|
||||
await persistQueue.catch((error) => {
|
||||
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
|
||||
});
|
||||
}
|
||||
const latestTurn = extractLatestFrontendTurn(activeRun.messages);
|
||||
if (latestTurn) {
|
||||
void learningOrchestrator.onTurnCompleted({
|
||||
...latestTurn,
|
||||
requestContext,
|
||||
sessionId: clientSessionId,
|
||||
}).catch((error) => {
|
||||
logger.warn(
|
||||
{ err: error, sessionId: clientSessionId },
|
||||
"stream-completed learning failed",
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
unregisterFrontendActionSink?.();
|
||||
if (abortController.signal.aborted) {
|
||||
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
|
||||
...message,
|
||||
content:
|
||||
typeof message.content === "string" && message.content.trim()
|
||||
? message.content
|
||||
: "⚠️ **请求已中断**",
|
||||
isError: true,
|
||||
progress: completeBackendProgress(message.progress),
|
||||
todos: cancelBackendTodos(message.todos),
|
||||
}));
|
||||
if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) {
|
||||
writeUiChunk({
|
||||
type: "error",
|
||||
errorText: "请求已中断",
|
||||
});
|
||||
finishUiMessage("error");
|
||||
}
|
||||
void queueSessionUiStatePersist().catch((error) => {
|
||||
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state");
|
||||
});
|
||||
}
|
||||
if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) {
|
||||
finishUiMessage(activeRun.status === "error" ? "error" : "stop");
|
||||
}
|
||||
await persistQueue.catch((error) => {
|
||||
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
|
||||
});
|
||||
sessionBridge.finalizeRequest(clientSessionId);
|
||||
activeRun.status = abortController.signal.aborted
|
||||
? activeRun.status === "aborted"
|
||||
? "aborted"
|
||||
: "aborted"
|
||||
: activeRun.status === "running"
|
||||
? "completed"
|
||||
: activeRun.status;
|
||||
lastRunStatuses.set(clientSessionId, activeRun.status);
|
||||
for (const subscriber of activeRun.subscribers) {
|
||||
subscriber.close();
|
||||
}
|
||||
activeRun.subscribers.clear();
|
||||
if (
|
||||
activeRun.pendingPermissions.size === 0 &&
|
||||
activeRun.pendingQuestions.size === 0
|
||||
) {
|
||||
activeRuns.delete(clientSessionId);
|
||||
}
|
||||
streamClosed = true;
|
||||
req.off("close", handleClientClose);
|
||||
res.off("close", handleClientClose);
|
||||
}
|
||||
|
||||
if (!res.writableEnded && !res.destroyed) {
|
||||
res.end();
|
||||
}
|
||||
} catch (error) {
|
||||
const detail = error instanceof Error ? error.message : String(error);
|
||||
logger.error({ err: error }, "chat stream failed");
|
||||
if (res.headersSent) {
|
||||
if (!res.writableEnded && !res.destroyed) {
|
||||
pipeUiChunk(res, {
|
||||
type: "error",
|
||||
errorText: detail,
|
||||
});
|
||||
pipeUiChunk(res, {
|
||||
type: "finish",
|
||||
finishReason: "error",
|
||||
});
|
||||
res.end();
|
||||
}
|
||||
return;
|
||||
}
|
||||
res.status(500).json({
|
||||
message: "chat stream failed",
|
||||
detail,
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
};
|
||||
+104
-3
@@ -4,6 +4,7 @@ import {
|
||||
type OpencodeClient,
|
||||
} from "@opencode-ai/sdk/v2";
|
||||
import { existsSync, readFileSync } from "node:fs";
|
||||
import { createServer } from "node:net";
|
||||
import { resolve } from "node:path";
|
||||
|
||||
import { config } from "../config.js";
|
||||
@@ -44,6 +45,34 @@ type RuntimeMessage = {
|
||||
const getRuntimeMessageRole = (message: RuntimeMessage) => message.info.role;
|
||||
const getRuntimeMessageId = (message: RuntimeMessage) => message.info.id;
|
||||
|
||||
const isTcpPortAvailable = (hostname: string, port: number) =>
|
||||
new Promise<boolean>((resolve, reject) => {
|
||||
const server = createServer();
|
||||
server.unref();
|
||||
server.once("error", (error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "EADDRINUSE") {
|
||||
resolve(false);
|
||||
return;
|
||||
}
|
||||
reject(error);
|
||||
});
|
||||
server.listen({ host: hostname, port, exclusive: true }, () => {
|
||||
server.close((error) => (error ? reject(error) : resolve(true)));
|
||||
});
|
||||
});
|
||||
|
||||
export const resolveEmbeddedOpencodePort = async (
|
||||
hostname: string,
|
||||
configuredPort: number,
|
||||
) => {
|
||||
if (configuredPort === 0) {
|
||||
return 0;
|
||||
}
|
||||
return (await isTcpPortAvailable(hostname, configuredPort))
|
||||
? configuredPort
|
||||
: 0;
|
||||
};
|
||||
|
||||
export class OpencodeRuntimeAdapter {
|
||||
private clientPromise: Promise<OpencodeClient> | null = null;
|
||||
private closeServer: (() => void) | null = null;
|
||||
@@ -64,6 +93,64 @@ export class OpencodeRuntimeAdapter {
|
||||
return requireData(response.data, "global.health");
|
||||
}
|
||||
|
||||
async warmup(): Promise<void> {
|
||||
const client = await this.ensureClient();
|
||||
const healthStartedAt = Date.now();
|
||||
const healthResponse = await client.global.health();
|
||||
const health = requireData(healthResponse.data, "global.health");
|
||||
logDevelopmentDebug("opencode warmup health check completed", {
|
||||
elapsedMs: Math.max(0, Date.now() - healthStartedAt),
|
||||
healthy: health.healthy,
|
||||
version: health.version,
|
||||
});
|
||||
|
||||
const sessionStartedAt = Date.now();
|
||||
const sessionResponse = await client.session.create({
|
||||
title: "tjwater-agent-warmup",
|
||||
});
|
||||
const session = requireData(sessionResponse.data, "session.create");
|
||||
logDevelopmentDebug("opencode warmup session created", {
|
||||
elapsedMs: Math.max(0, Date.now() - sessionStartedAt),
|
||||
sessionId: session.id,
|
||||
});
|
||||
|
||||
try {
|
||||
const [provider, model] = config.OPENCODE_MODEL.split("/");
|
||||
if (!provider || !model) {
|
||||
throw new Error(
|
||||
`invalid OPENCODE_MODEL; expected provider/model, received ${config.OPENCODE_MODEL}`,
|
||||
);
|
||||
}
|
||||
const toolsStartedAt = Date.now();
|
||||
const toolsResponse = await client.tool.list({ provider, model });
|
||||
const tools = requireData(toolsResponse.data, "tool.list");
|
||||
logDevelopmentDebug("opencode warmup tools loaded", {
|
||||
elapsedMs: Math.max(0, Date.now() - toolsStartedAt),
|
||||
model: config.OPENCODE_MODEL,
|
||||
sessionId: session.id,
|
||||
toolCount: tools.length,
|
||||
});
|
||||
} finally {
|
||||
const cleanupStartedAt = Date.now();
|
||||
let cleanupSucceeded = true;
|
||||
await client.session.delete(
|
||||
{ sessionID: session.id },
|
||||
{ throwOnError: true },
|
||||
).catch((error) => {
|
||||
cleanupSucceeded = false;
|
||||
logger.warn(
|
||||
{ err: error, sessionId: session.id },
|
||||
"failed to remove opencode warmup session",
|
||||
);
|
||||
});
|
||||
logDevelopmentDebug("opencode warmup session cleanup completed", {
|
||||
elapsedMs: Math.max(0, Date.now() - cleanupStartedAt),
|
||||
sessionId: session.id,
|
||||
succeeded: cleanupSucceeded,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async createSession(title?: string) {
|
||||
const client = await this.ensureClient();
|
||||
const response = await client.session.create({
|
||||
@@ -350,10 +437,24 @@ export class OpencodeRuntimeAdapter {
|
||||
process.env.TJWATER_AGENT_INTERNAL_TOKEN ??
|
||||
"";
|
||||
|
||||
const port = await resolveEmbeddedOpencodePort(
|
||||
config.OPENCODE_HOSTNAME,
|
||||
config.OPENCODE_PORT,
|
||||
);
|
||||
if (port !== config.OPENCODE_PORT) {
|
||||
logger.warn(
|
||||
{
|
||||
hostname: config.OPENCODE_HOSTNAME,
|
||||
configuredPort: config.OPENCODE_PORT,
|
||||
},
|
||||
"configured opencode port is occupied; falling back to a dynamic port",
|
||||
);
|
||||
}
|
||||
|
||||
logger.info(
|
||||
{
|
||||
hostname: config.OPENCODE_HOSTNAME,
|
||||
port: config.OPENCODE_PORT,
|
||||
port,
|
||||
model: config.OPENCODE_MODEL,
|
||||
mode: config.OPENCODE_MODE,
|
||||
},
|
||||
@@ -365,7 +466,7 @@ export class OpencodeRuntimeAdapter {
|
||||
try {
|
||||
runtime = await createOpencode({
|
||||
hostname: config.OPENCODE_HOSTNAME,
|
||||
port: config.OPENCODE_PORT,
|
||||
port,
|
||||
timeout: config.OPENCODE_TIMEOUT_MS,
|
||||
config: buildOpencodeConfig(),
|
||||
});
|
||||
@@ -382,7 +483,7 @@ export class OpencodeRuntimeAdapter {
|
||||
{
|
||||
elapsedMs: Math.max(0, Date.now() - startedAt),
|
||||
hostname: config.OPENCODE_HOSTNAME,
|
||||
port: config.OPENCODE_PORT,
|
||||
baseUrl: runtime.server.url,
|
||||
mode: config.OPENCODE_MODE,
|
||||
},
|
||||
"opencode server started in embedded mode",
|
||||
|
||||
+32
-22
@@ -75,6 +75,8 @@ app.get("/health", async (_req, res) => {
|
||||
const runtime = await opencodeRuntime.health();
|
||||
res.json({
|
||||
ok: true,
|
||||
ready: true,
|
||||
warmed_up: true,
|
||||
runtime,
|
||||
sessions: sessionBridge.count(),
|
||||
});
|
||||
@@ -82,6 +84,8 @@ app.get("/health", async (_req, res) => {
|
||||
const detail = error instanceof Error ? error.message : String(error);
|
||||
res.status(503).json({
|
||||
ok: false,
|
||||
ready: false,
|
||||
warmed_up: true,
|
||||
message: "opencode runtime unavailable",
|
||||
detail,
|
||||
sessions: sessionBridge.count(),
|
||||
@@ -126,17 +130,17 @@ app.post("/internal/tools/session-search", async (req, res) => {
|
||||
|
||||
app.use(
|
||||
"/api/v1/agent/chat",
|
||||
buildChatRouter(
|
||||
sessionBridge,
|
||||
opencodeRuntime,
|
||||
sessionMetadataStore,
|
||||
sessionUiStateStore,
|
||||
memoryStore,
|
||||
sessionTranscriptStore,
|
||||
learningOrchestrator,
|
||||
resultReferenceResolver,
|
||||
buildChatRouter({
|
||||
frontendActionCoordinator,
|
||||
),
|
||||
learningOrchestrator,
|
||||
memoryStore,
|
||||
resultReferenceResolver,
|
||||
runtime: opencodeRuntime,
|
||||
sessionBridge,
|
||||
sessionMetadataStore,
|
||||
sessionTranscriptStore,
|
||||
sessionUiStateStore,
|
||||
}),
|
||||
);
|
||||
|
||||
const bootstrap = async () => {
|
||||
@@ -148,23 +152,12 @@ const bootstrap = async () => {
|
||||
resultReferenceStore.initialize(),
|
||||
sessionTranscriptStore.initialize(),
|
||||
]);
|
||||
resultReferenceStore.startCleanupLoop();
|
||||
};
|
||||
|
||||
await bootstrap();
|
||||
|
||||
const server = app.listen(config.PORT, config.HOST, () => {
|
||||
logger.info(
|
||||
{ host: config.HOST, port: config.PORT },
|
||||
"TJWaterAgent listening",
|
||||
);
|
||||
void warmupOpencodeRuntime();
|
||||
});
|
||||
|
||||
const warmupOpencodeRuntime = async () => {
|
||||
const startedAt = Date.now();
|
||||
try {
|
||||
await opencodeRuntime.ensureClient();
|
||||
await opencodeRuntime.warmup();
|
||||
logger.info(
|
||||
{
|
||||
elapsedMs: Math.max(0, Date.now() - startedAt),
|
||||
@@ -181,9 +174,26 @@ const warmupOpencodeRuntime = async () => {
|
||||
},
|
||||
"failed to warm up opencode runtime",
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
await bootstrap();
|
||||
await warmupOpencodeRuntime();
|
||||
resultReferenceStore.startCleanupLoop();
|
||||
|
||||
const server = app.listen(config.PORT, config.HOST, () => {
|
||||
logger.info(
|
||||
{
|
||||
host: config.HOST,
|
||||
port: config.PORT,
|
||||
ready: true,
|
||||
warmedUp: true,
|
||||
},
|
||||
"TJWaterAgent listening",
|
||||
);
|
||||
});
|
||||
|
||||
const shutdown = async () => {
|
||||
logger.info("shutting down TJWaterAgent");
|
||||
server.close();
|
||||
|
||||
@@ -8,8 +8,8 @@ import { frontendActionRegistry } from "../../src/frontendAction/registry.js";
|
||||
import type { FrontendActionRequest } from "../../src/frontendAction/types.js";
|
||||
|
||||
const items = [
|
||||
{ sensor_id: "MP01", level: "high" },
|
||||
{ sensor_id: "MP02", level: "medium" },
|
||||
{ device_id: "MP01", level: "high" },
|
||||
{ device_id: "MP02", level: "medium" },
|
||||
] as const;
|
||||
|
||||
const start = (
|
||||
@@ -119,10 +119,10 @@ describe("FrontendActionCoordinator", () => {
|
||||
params: value,
|
||||
});
|
||||
expect(() => request({ items: [] })).toThrow(FrontendActionError);
|
||||
expect(() => request({ items: Array.from({ length: 101 }, (_, index) => ({ sensor_id: `MP${index}`, level: "low" })) })).toThrow("invalid frontend action params");
|
||||
expect(() => request({ items: [{ sensor_id: "MP01", level: "high" }, { sensor_id: "MP01", level: "low" }] })).toThrow("invalid frontend action params");
|
||||
expect(() => request({ items: [{ sensor_id: "x".repeat(129), level: "low" }] })).toThrow("invalid frontend action params");
|
||||
expect(() => request({ items: [{ sensor_id: "MP01", level: "critical" }] })).toThrow("invalid frontend action params");
|
||||
expect(() => request({ items: Array.from({ length: 101 }, (_, index) => ({ device_id: `MP${index}`, level: "low" })) })).toThrow("invalid frontend action params");
|
||||
expect(() => request({ items: [{ device_id: "MP01", level: "high" }, { device_id: "MP01", level: "low" }] })).toThrow("invalid frontend action params");
|
||||
expect(() => request({ items: [{ device_id: "x".repeat(129), level: "low" }] })).toThrow("invalid frontend action params");
|
||||
expect(() => request({ items: [{ device_id: "MP01", level: "critical" }] })).toThrow("invalid frontend action params");
|
||||
});
|
||||
|
||||
it("keeps terminal results idempotent and rejects conflicts", async () => {
|
||||
|
||||
@@ -12,7 +12,7 @@ describe("appendBackendToolArtifact", () => {
|
||||
const artifacts = appendBackendToolArtifact([], {
|
||||
tool: "render_scada_analysis",
|
||||
reason: "显示已完成的分级结论",
|
||||
params: { items: [{ sensor_id: "MP01", level: "high" }] },
|
||||
params: { items: [{ device_id: "MP01", level: "high" }] },
|
||||
}) as Array<Record<string, unknown>>;
|
||||
|
||||
expect(artifacts[0]).toMatchObject({
|
||||
|
||||
@@ -1,7 +1,12 @@
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { type OpencodeClient } from "@opencode-ai/sdk/v2";
|
||||
import { createServer } from "node:net";
|
||||
|
||||
import { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js";
|
||||
import { config } from "../../src/config.js";
|
||||
import {
|
||||
OpencodeRuntimeAdapter,
|
||||
resolveEmbeddedOpencodePort,
|
||||
} from "../../src/runtime/opencode.js";
|
||||
|
||||
const createRuntimeAdapter = (
|
||||
messages: unknown[],
|
||||
@@ -85,3 +90,73 @@ describe("OpencodeRuntimeAdapter.ensureClient", () => {
|
||||
expect(attempts).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("resolveEmbeddedOpencodePort", () => {
|
||||
it("falls back to a dynamic port when the configured port is occupied", async () => {
|
||||
const server = createServer();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") {
|
||||
server.close();
|
||||
throw new Error("expected a TCP server address");
|
||||
}
|
||||
|
||||
try {
|
||||
await expect(
|
||||
resolveEmbeddedOpencodePort("127.0.0.1", address.port),
|
||||
).resolves.toBe(0);
|
||||
} finally {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("OpencodeRuntimeAdapter.warmup", () => {
|
||||
it("initializes the project session and model tools before reporting ready", async () => {
|
||||
const calls: string[] = [];
|
||||
const client = {
|
||||
global: {
|
||||
health: async () => {
|
||||
calls.push("health");
|
||||
return { data: { healthy: true, version: "test" } };
|
||||
},
|
||||
},
|
||||
session: {
|
||||
create: async () => {
|
||||
calls.push("session.create");
|
||||
return { data: { id: "warmup-session" } };
|
||||
},
|
||||
delete: async ({ sessionID }: { sessionID: string }) => {
|
||||
calls.push(`session.delete:${sessionID}`);
|
||||
return { data: true };
|
||||
},
|
||||
},
|
||||
tool: {
|
||||
list: async (model: { provider: string; model: string }) => {
|
||||
calls.push(`tool.list:${model.provider}/${model.model}`);
|
||||
return { data: [] };
|
||||
},
|
||||
},
|
||||
} as unknown as OpencodeClient;
|
||||
const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), {
|
||||
clientPromise: null,
|
||||
closeServer: null,
|
||||
ensureClient: async () => client,
|
||||
}) as OpencodeRuntimeAdapter;
|
||||
|
||||
await runtime.warmup();
|
||||
|
||||
expect(calls).toEqual([
|
||||
"health",
|
||||
"session.create",
|
||||
`tool.list:${config.OPENCODE_MODEL}`,
|
||||
"session.delete:warmup-session",
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
import { describe, expect, it, test } from "bun:test";
|
||||
import { readFile } from "node:fs/promises";
|
||||
|
||||
describe("internal OpenCode permissions", () => {
|
||||
it("keeps protected paths denied in every approval mode", async () => {
|
||||
const config = JSON.parse(await readFile("opencode.json", "utf8")) as {
|
||||
permission?: Record<string, string | Record<string, string>>;
|
||||
};
|
||||
const permission = config.permission ?? {};
|
||||
const bash = permission.bash as Record<string, string> | undefined;
|
||||
const edit = permission.edit as Record<string, string> | undefined;
|
||||
const read = permission.read as Record<string, string> | undefined;
|
||||
|
||||
expect(permission["*"]).toBe("ask");
|
||||
expect(permission.external_directory).toBe("deny");
|
||||
expect(permission.task).toBe("deny");
|
||||
expect(permission.question).toBe("allow");
|
||||
expect(permission.todowrite).toBe("allow");
|
||||
expect(read?.["*"]).toBe("allow");
|
||||
expect(read?.["data/**"]).toBe("deny");
|
||||
expect(read?.["**/logs/**"]).toBe("deny");
|
||||
expect(edit?.["*"]).toBe("ask");
|
||||
expect(edit?.["data/**"]).toBe("deny");
|
||||
expect(edit?.["**/logs/**"]).toBe("deny");
|
||||
expect(bash?.["*"]).toBe("ask");
|
||||
expect(bash?.["*.env*"]).toBe("deny");
|
||||
expect(bash?.["*data/*"]).toBe("deny");
|
||||
expect(bash?.["*logs/*"]).toBe("deny");
|
||||
});
|
||||
});
|
||||
|
||||
describe("OpenCode runtime config", () => {
|
||||
test("continues the agent loop after a question or permission is denied", async () => {
|
||||
const config = await Bun.file(new URL("../../opencode.json", import.meta.url)).json();
|
||||
|
||||
expect(config.experimental?.continue_loop_on_deny).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -12,7 +12,7 @@ describe("toUiEnvelopeFromToolCall", () => {
|
||||
it("does not generate duplicate UIEnvelopes for SCADA frontend actions", () => {
|
||||
expect(toUiEnvelopeFromToolCall({
|
||||
tool: "render_scada_analysis",
|
||||
params: { items: [{ sensor_id: "MP01", level: "high" }] },
|
||||
params: { items: [{ device_id: "MP01", level: "high" }] },
|
||||
})).toBeNull();
|
||||
expect(toUiEnvelopeFromToolCall({
|
||||
tool: "clear_scada_analysis",
|
||||
|
||||
Reference in New Issue
Block a user