9 Commits
Author SHA1 Message Date
jiang 88b41ba04c refactor(scada): use canonical device identifiers
Generic Container CI/CD / test-build-publish (push) Successful in 32s
Agent CI/CD / build-test-publish-and-deploy (push) Successful in 32s
2026-09-08 18:18:51 +08:00
jiang b33d950d2b fix: avoid embedded opencode port conflicts 2026-08-19 16:56:25 +08:00
jiang f0bf667d02 fix(agent): harden permissions and startup readiness 2026-08-18 17:10:48 +08:00
jiang e8a335d5ba fix: isolate Next agent deployment
Generic Container CI/CD / test-build-publish (push) Successful in 33s
Agent CI/CD / build-test-publish-and-deploy (push) Successful in 33s
2026-08-17 19:43:50 +08:00
jiang 45687bcf67 build: use mainland package mirrors
Generic Container CI/CD / test-build-publish (push) Successful in 31s
Agent CI/CD / build-test-publish-and-deploy (push) Successful in 31s
2026-08-17 19:25:38 +08:00
jiang 178e74b403 feat: add container CI/CD workflow 2026-08-17 19:02:02 +08:00
jiang a37f06352c fix: continue agent after denied interactions 2026-07-28 18:08:08 +08:00
jiang 9286ca67e9 refactor: split chat route responsibilities 2026-07-22 18:16:43 +08:00
jiang 7b23c61723 docs: translate agent readme to Chinese 2026-07-22 15:20:21 +08:00
22 changed files with 1772 additions and 1293 deletions
+8
View File
@@ -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
View File
@@ -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。
+22
View File
@@ -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 }}
+2 -2
View File
@@ -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 地图工具返回浏览器的成功结果后,才能声称已渲染或已清除;工具报错、超时或无活跃浏览器连接时,必须明确告知用户动作失败,不得伪造已渲染的点位、编号或等级。
## 执行约束
+4 -4
View File
@@ -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."),
},
+10 -1
View File
@@ -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
View File
@@ -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"]
+67 -50
View File
@@ -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
View File
@@ -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
View File
@@ -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 模型标识。
+8 -8
View File
@@ -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
View File
File diff suppressed because it is too large Load Diff
+22
View File
@@ -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),
};
};
+426
View File
@@ -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,
});
}
});
};
+811
View File
@@ -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
View File
@@ -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
View File
@@ -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();
+6 -6
View File
@@ -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 () => {
+1 -1
View File
@@ -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({
+76 -1
View File
@@ -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",
]);
});
});
+38
View File
@@ -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);
});
});
+1 -1
View File
@@ -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",