feat(agent): add credential refresh and unify learning tools

This commit is contained in:
2026-08-06 10:16:50 +08:00
parent 2dc37e3fd8
commit 258f4996eb
21 changed files with 1270 additions and 770 deletions
+19 -123
View File
@@ -1,11 +1,8 @@
import { tool } from "@opencode-ai/plugin"; import { tool } from "@opencode-ai/plugin";
import { MemoryStore } from "../../src/memory/store.js"; const internalBaseUrl =
import { readBridgedRuntimeSessionContext } from "../../src/runtime/internalSessionContextBridge.js"; process.env.TJWATER_AGENT_INTERNAL_BASE_URL ?? "http://127.0.0.1:8787";
import { setRuntimeSessionContext } from "../../src/runtime/sessionContext.js"; const internalToken = process.env.TJWATER_AGENT_INTERNAL_TOKEN ?? "";
const memoryStore = new MemoryStore();
const initializePromise = memoryStore.initialize();
export default tool({ export default tool({
description: description:
@@ -25,132 +22,31 @@ export default tool({
content: tool.schema content: tool.schema
.string() .string()
.optional() .optional()
.describe( .describe("The durable fact or preference to remember, written as one concise sentence."),
"The durable fact or preference to remember, written as one concise sentence.",
),
target_id: tool.schema target_id: tool.schema
.string() .string()
.optional() .optional()
.describe("Stable memory entry id used by replace/remove."), .describe("Stable memory entry id used by replace/remove."),
}, },
async execute(args, context) { async execute(args, context) {
await initializePromise; const response = await fetch(
const sessionContext = await readBridgedRuntimeSessionContext( `${internalBaseUrl}/internal/tools/memory-manager`,
context.sessionID,
);
if (!sessionContext) {
throw new Error(`session context not found for ${context.sessionID}`);
}
const scope =
args.scope === "user"
? "user"
: args.scope === "workspace"
? "workspace"
: null;
if (!scope) {
return JSON.stringify({
ok: true,
kind: "memory",
decision: "rejected",
detail: `unsupported scope: ${args.scope}; use exact keyword 'user' or 'workspace'`,
});
}
if (sessionContext.allowLearningWrite === false && args.action !== "list") {
return JSON.stringify({
ok: true,
kind: "memory",
decision: "rejected",
detail: "memory writes are disabled for this session",
});
}
const scopeKey =
scope === "user" ? sessionContext.actorKey : sessionContext.projectKey;
if (args.action === "list") {
const readScopes = {
...(sessionContext.memoryListReadScopes ?? {}),
[scope]: true,
};
setRuntimeSessionContext({
...sessionContext,
memoryListReadScopes: readScopes,
});
return JSON.stringify({
ok: true,
kind: "memory",
decision: "accepted",
detail: "memory listed",
items: await memoryStore.list(scope, scopeKey),
target: scope,
});
}
if (args.action === "add") {
if (sessionContext.memoryListReadScopes?.[scope] !== true) {
return JSON.stringify({
ok: true,
kind: "memory",
decision: "rejected",
detail: `must list ${scope} memory and review existing entries before add`,
target: scope,
});
}
const result = await memoryStore.upsert(scope, scopeKey, {
content: args.content ?? "",
sessionId: sessionContext.clientSessionId,
source: "tool",
traceId: sessionContext.traceId,
});
if (!result.entry) {
return JSON.stringify({
ok: true,
kind: "memory",
decision: "rejected",
detail: "content rejected by persistence policy",
});
}
return JSON.stringify({
ok: true,
kind: "memory",
decision: result.changed ? "accepted" : "deduped",
detail: result.detail,
entry: result.entry,
target: scope,
});
}
if (args.action === "replace") {
const result = await memoryStore.replace(
scope,
scopeKey,
args.target_id ?? "",
{ {
content: args.content ?? "", method: "POST",
sessionId: sessionContext.clientSessionId, headers: {
source: "tool", "Content-Type": "application/json",
traceId: sessionContext.traceId, "x-agent-internal-token": internalToken,
},
body: JSON.stringify({
...args,
session_id: context.sessionID,
}),
}, },
); );
return JSON.stringify({ const text = await response.text();
ok: true, if (!response.ok) {
kind: "memory", throw new Error(text);
decision: result.changed ? "accepted" : "rejected",
detail: result.detail,
target: scope,
});
} }
return text;
const result = await memoryStore.remove(
scope,
scopeKey,
args.target_id ?? "",
);
return JSON.stringify({
ok: true,
kind: "memory",
decision: result.changed ? "accepted" : "rejected",
detail: result.detail,
target: scope,
});
}, },
}); });
+26 -119
View File
@@ -1,27 +1,10 @@
import { tool } from "@opencode-ai/plugin"; import { tool } from "@opencode-ai/plugin";
import { SkillStore } from "../../src/skills/store.js"; const internalBaseUrl =
import { process.env.TJWATER_AGENT_INTERNAL_BASE_URL ?? "http://127.0.0.1:8787";
readBridgedRuntimeSessionContext, const internalToken = process.env.TJWATER_AGENT_INTERNAL_TOKEN ?? "";
} from "../../src/runtime/internalSessionContextBridge.js";
import { type RuntimeSessionContext } from "../../src/runtime/sessionContext.js";
type ToolContextReader = { export default tool({
read(
sessionId: string,
): RuntimeSessionContext | null | Promise<RuntimeSessionContext | null>;
};
const runtimeContextReader: ToolContextReader = {
read: readBridgedRuntimeSessionContext,
};
export const createSkillManagerTool = (
skillStore = new SkillStore(),
toolContextStore: ToolContextReader = runtimeContextReader,
initializePromise: Promise<unknown> = Promise.resolve(),
) =>
tool({
description: description:
"维护已验证、可复用、非敏感的 workflow 或方法模式。支持 list、write_skill、remove_skill、append_pattern、remove_pattern、write_reference、remove_reference、write_script、remove_script。", "维护已验证、可复用、非敏感的 workflow 或方法模式。支持 list、write_skill、remove_skill、append_pattern、remove_pattern、write_reference、remove_reference、write_script、remove_script。",
args: { args: {
@@ -40,18 +23,13 @@ export const createSkillManagerTool = (
.describe("Skill maintenance operation."), .describe("Skill maintenance operation."),
reason: tool.schema reason: tool.schema
.string() .string()
.describe( .describe("Why this skill maintenance action is justified for future reuse."),
"Why this skill maintenance action is justified for future reuse.",
),
skill_path: tool.schema skill_path: tool.schema
.string() .string()
.describe( .describe(
"Target skill directory path relative to .opencode/skills. Use 'workflow' for the workflow index, or '__root__' for the root skills index.", "Target skill directory path relative to .opencode/skills. Use 'workflow' for the workflow index, or '__root__' for the root skills index.",
), ),
pattern: tool.schema pattern: tool.schema.string().optional().describe("Pattern text used by append_pattern."),
.string()
.optional()
.describe("Pattern text used by append_pattern."),
target_id: tool.schema target_id: tool.schema
.string() .string()
.optional() .optional()
@@ -59,102 +37,31 @@ export const createSkillManagerTool = (
file_path: tool.schema file_path: tool.schema
.string() .string()
.optional() .optional()
.describe( .describe("Asset file path. For references use references/*.md; for scripts use scripts/*.py."),
"Asset file path. For references use references/*.md; for scripts use scripts/*.py.",
),
content: tool.schema content: tool.schema
.string() .string()
.optional() .optional()
.describe( .describe("Content used by write_skill, write_reference, or write_script."),
"Content used by write_skill, write_reference, or write_script.",
),
}, },
async execute(args, context) { async execute(args, context) {
await initializePromise; const response = await fetch(
const sessionContext = await toolContextStore.read(context.sessionID); `${internalBaseUrl}/internal/tools/skill-manager`,
if (!sessionContext) { {
throw new Error(`session context not found for ${context.sessionID}`); method: "POST",
} headers: {
if ( "Content-Type": "application/json",
sessionContext.allowLearningWrite === false && "x-agent-internal-token": internalToken,
args.action !== "list" },
) { body: JSON.stringify({
return JSON.stringify({ ...args,
ok: true, session_id: context.sessionID,
kind: "skill", }),
decision: "rejected", },
detail: "skill writes are disabled for this session",
});
}
if (args.action === "list") {
const result = await skillStore.list(args.skill_path);
if (!result) {
return JSON.stringify({
ok: true,
kind: "skill",
decision: "rejected",
detail:
"invalid skill_path; expected a relative path under .opencode/skills",
});
}
return JSON.stringify({
ok: true,
kind: "skill",
decision: "accepted",
detail: "skill listed",
references: result.references,
scripts: result.scripts,
skill_path: result.skillPath,
target: result.target,
patterns: result.patterns,
});
}
const result =
args.action === "write_skill"
? await skillStore.writeSkill(args.skill_path, args.content ?? "")
: args.action === "remove_skill"
? await skillStore.removeSkill(args.skill_path)
: args.action === "append_pattern"
? await skillStore.appendPattern(
args.skill_path,
args.pattern ?? "",
)
: args.action === "remove_pattern"
? await skillStore.removePattern(
args.skill_path,
args.target_id ?? "",
)
: args.action === "write_reference"
? await skillStore.writeReference(
args.skill_path,
args.file_path ?? "",
args.content ?? "",
)
: args.action === "remove_reference"
? await skillStore.removeReference(
args.skill_path,
args.file_path ?? "",
)
: args.action === "write_script"
? await skillStore.writeScript(
args.skill_path,
args.file_path ?? "",
args.content ?? "",
)
: await skillStore.removeScript(
args.skill_path,
args.file_path ?? "",
); );
const text = await response.text();
return JSON.stringify({ if (!response.ok) {
ok: true, throw new Error(text);
kind: "skill", }
decision: result.changed ? "accepted" : "rejected", return text;
detail: result.detail,
target: result.target,
});
}, },
}); });
export default createSkillManagerTool();
+7 -7
View File
@@ -5,7 +5,7 @@
## 主要能力 ## 主要能力
- 提供 `POST /api/v1/agent/sessions/{session_id}/runs` SSE 聊天接口。 - 提供 `POST /api/v1/agent/sessions/{session_id}/runs` SSE 聊天接口。
- 支持 embedded OpenCode 运行时,也可连接外部 OpenCode server - 以内嵌模式启动并预热 OpenCode 运行时。
- 管理前端 `session_id` 与 OpenCode session 的映射。 - 管理前端 `session_id` 与 OpenCode session 的映射。
- 在服务端保存当前会话的用户 token、项目、network 和 trace 上下文。 - 在服务端保存当前会话的用户 token、项目、network 和 trace 上下文。
- 通过 `.opencode/tools` 和 MCP 工具驱动地图定位、图表、SCADA、历史数据和业务 API 调用。 - 通过 `.opencode/tools` 和 MCP 工具驱动地图定位、图表、SCADA、历史数据和业务 API 调用。
@@ -67,13 +67,13 @@ OPENCODE_MODE=embedded
TJWATER_API_BASE_URL=http://127.0.0.1:8000 TJWATER_API_BASE_URL=http://127.0.0.1:8000
``` ```
Client 模式连接外部 OpenCode server 当前仅支持 Embedded 模式,不支持连接外部 OpenCode server
```bash ## 认证续期与学习工具
OPENCODE_MODE=client
OPENCODE_CLIENT_BASE_URL=http://127.0.0.1:4096 后端工具调用遇到即将过期的 access token 或首次 `401` 时,Agent 会通过当前 SSE 流发送 `credential_refresh_required`。前端使用服务端保存的 Keycloak refresh token 强制换取新 access token,再调用 `POST /api/v1/agent/sessions/{session_id}/credential-refreshes` 唤醒原工具调用。等待上限为 30 秒,同一会话的并发请求合并为一次续期,原调用最多重试一次;`403` 不触发续期。
TJWATER_API_BASE_URL=http://127.0.0.1:8000
``` `memory_manager``skill_manager` 在 OpenCode 侧只保留内部 HTTP 桥,读取会话上下文和持久化数据的逻辑统一在 Agent 主进程中执行。长期记忆、自动学习和显式工具写入因此共享同一组 `MemoryStore``SkillStore` 和运行时会话上下文。
本地可使用 `.local.env` 保存开发配置;系统环境变量优先级更高。 本地可使用 `.local.env` 保存开发配置;系统环境变量优先级更高。
+149
View File
@@ -1386,6 +1386,155 @@
} }
} }
}, },
"/api/v1/agent/sessions/{session_id}/credential-refreshes": {
"post": {
"operationId": "post_sessions_session_id_credential_refreshes",
"tags": [
"Agent"
],
"security": [
{
"bearerAuth": []
}
],
"summary": "Resume a waiting agent tool call with refreshed credentials",
"parameters": [
{
"schema": {
"type": "string",
"maxLength": 128
},
"required": true,
"name": "session_id",
"in": "path"
}
],
"requestBody": {
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"request_id": {
"type": "string",
"minLength": 1,
"maxLength": 128
}
},
"required": [
"request_id"
]
}
}
}
},
"responses": {
"202": {
"description": "Successful response",
"content": {
"application/json": {
"schema": {
"type": "object",
"additionalProperties": {
"nullable": true
}
}
}
}
},
"400": {
"description": "Invalid request",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"401": {
"description": "Authentication required",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"403": {
"description": "Insufficient permission",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"404": {
"description": "Resource not found",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"409": {
"description": "Resource conflict",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"422": {
"description": "Validation error",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"500": {
"description": "Internal server error",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"502": {
"description": "Upstream dependency error",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
},
"503": {
"description": "Dependency unavailable",
"content": {
"application/problem+json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
}
}
}
}
},
"/api/v1/agent/sessions/{session_id}/permission-responses": { "/api/v1/agent/sessions/{session_id}/permission-responses": {
"post": { "post": {
"operationId": "post_sessions_session_id_permission_responses", "operationId": "post_sessions_session_id_permission_responses",
+1 -1
View File
@@ -3,7 +3,7 @@
"contracts": { "contracts": {
"agent": { "agent": {
"file": "agent-v1.openapi.json", "file": "agent-v1.openapi.json",
"sha256": "7699d0b59d2710f5179c3880fa9f7de90dee09239718c86ed9ff2ce12e6f4259" "sha256": "d559c6e76c33e7a7451743f60d85da0630d0df14fb5215228acefcf2eaea555a"
} }
} }
} }
+232
View File
@@ -0,0 +1,232 @@
import { randomUUID } from "node:crypto";
import { type RuntimeSessionContext } from "../runtime/sessionContext.js";
export type CredentialRefreshReason =
| "access_token_expired"
| "access_token_rejected";
export type CredentialRefreshEvent =
| {
type: "credential_refresh_required";
requestId: string;
reason: CredentialRefreshReason;
timeoutMs: number;
}
| {
type: "credential_refreshed";
requestId: string;
}
| {
type: "credential_refresh_failed";
requestId: string;
message: string;
};
type PendingRefresh = {
deadlineAt: number;
promise: Promise<RuntimeSessionContext>;
reason: CredentialRefreshReason;
reject: (error: Error) => void;
requestId: string;
resolve: (context: RuntimeSessionContext) => void;
timer: ReturnType<typeof setTimeout>;
};
type CredentialRefreshListener = (event: CredentialRefreshEvent) => void;
export class CredentialRefreshError extends Error {
override readonly name = "CredentialRefreshError";
constructor(
message: string,
readonly code: "cancelled" | "failed" | "timeout" | "unavailable" = "failed",
) {
super(message);
}
}
const AUTH_EXPIRY_SKEW_MS = 30_000;
export const isRuntimeCredentialExpired = (
context: RuntimeSessionContext,
now = Date.now(),
) => {
if (!context.tokenExpiresAt) {
return false;
}
const expiresAt = Date.parse(context.tokenExpiresAt);
return Number.isFinite(expiresAt) && now >= expiresAt - AUTH_EXPIRY_SKEW_MS;
};
export class CredentialRefreshCoordinator {
private readonly listeners = new Map<
string,
Set<CredentialRefreshListener>
>();
private readonly pending = new Map<string, PendingRefresh>();
constructor(private readonly timeoutMs = 30_000) {}
subscribe(sessionId: string, listener: CredentialRefreshListener) {
const listeners =
this.listeners.get(sessionId) ?? new Set<CredentialRefreshListener>();
listeners.add(listener);
this.listeners.set(sessionId, listeners);
return () => {
listeners.delete(listener);
if (listeners.size === 0) {
this.listeners.delete(sessionId);
}
};
}
request(sessionId: string, reason: CredentialRefreshReason) {
const existing = this.pending.get(sessionId);
if (existing) {
return existing.promise;
}
if (!this.listeners.get(sessionId)?.size) {
return Promise.reject(
new CredentialRefreshError(
"credential refresh channel is unavailable",
"unavailable",
),
);
}
const requestId = `credential-${randomUUID()}`;
let resolvePromise!: (context: RuntimeSessionContext) => void;
let rejectPromise!: (error: Error) => void;
const promise = new Promise<RuntimeSessionContext>((resolve, reject) => {
resolvePromise = resolve;
rejectPromise = reject;
});
const timer = setTimeout(() => {
this.fail(sessionId, requestId, "credential refresh timed out", "timeout");
}, this.timeoutMs);
this.pending.set(sessionId, {
deadlineAt: Date.now() + this.timeoutMs,
promise,
reason,
reject: rejectPromise,
requestId,
resolve: resolvePromise,
timer,
});
this.emit(sessionId, {
type: "credential_refresh_required",
requestId,
reason,
timeoutMs: this.timeoutMs,
});
return promise;
}
resolve(
sessionId: string,
requestId: string,
context: RuntimeSessionContext,
) {
const pending = this.pending.get(sessionId);
if (!pending || pending.requestId !== requestId) {
return false;
}
clearTimeout(pending.timer);
this.pending.delete(sessionId);
pending.resolve(context);
this.emit(sessionId, {
type: "credential_refreshed",
requestId,
});
return true;
}
fail(
sessionId: string,
requestId: string,
message: string,
code: CredentialRefreshError["code"] = "failed",
emitFailureEvent = true,
) {
const pending = this.pending.get(sessionId);
if (!pending || pending.requestId !== requestId) {
return false;
}
clearTimeout(pending.timer);
this.pending.delete(sessionId);
pending.reject(new CredentialRefreshError(message, code));
if (emitFailureEvent) {
this.emit(sessionId, {
type: "credential_refresh_failed",
requestId,
message,
});
}
return true;
}
cancelSession(sessionId: string, message = "credential refresh cancelled") {
const pending = this.pending.get(sessionId);
if (!pending) {
return false;
}
return this.fail(
sessionId,
pending.requestId,
message,
"cancelled",
false,
);
}
getPendingRequestId(sessionId: string) {
return this.pending.get(sessionId)?.requestId;
}
getPendingEvent(
sessionId: string,
): Extract<CredentialRefreshEvent, { type: "credential_refresh_required" }> | null {
const pending = this.pending.get(sessionId);
if (!pending) return null;
return {
type: "credential_refresh_required",
requestId: pending.requestId,
reason: pending.reason,
timeoutMs: Math.max(0, pending.deadlineAt - Date.now()),
};
}
private emit(sessionId: string, event: CredentialRefreshEvent) {
for (const listener of this.listeners.get(sessionId) ?? []) {
listener(event);
}
}
}
export const runWithCredentialRefresh = async <T extends { status: number }>(
coordinator: CredentialRefreshCoordinator,
context: RuntimeSessionContext,
execute: (context: RuntimeSessionContext) => Promise<T>,
) => {
let activeContext = context;
let refreshed = false;
if (isRuntimeCredentialExpired(activeContext)) {
activeContext = await coordinator.request(
activeContext.sessionId,
"access_token_expired",
);
refreshed = true;
}
let result = await execute(activeContext);
if (result.status !== 401 || refreshed) {
return result;
}
activeContext = await coordinator.request(
activeContext.sessionId,
"access_token_rejected",
);
result = await execute(activeContext);
return result;
};
+3 -25
View File
@@ -41,8 +41,8 @@ const envSchema = z
AGENT_INTERNAL_TOKEN: optionalString(), AGENT_INTERNAL_TOKEN: optionalString(),
// Agent 前置认证调用后端 /api/v1/agent/auth/context 的超时时间(毫秒)。 // Agent 前置认证调用后端 /api/v1/agent/auth/context 的超时时间(毫秒)。
AGENT_AUTH_TIMEOUT_MS: z.coerce.number().int().positive().default(5000), AGENT_AUTH_TIMEOUT_MS: z.coerce.number().int().positive().default(5000),
// opencode 运行模式:embedded 会启动本地 CLI 子进程;client 只连接现有 server // 当前仅支持 embedded;保留字段用于让旧 client 配置在启动时明确失败
OPENCODE_MODE: z.enum(["embedded", "client"]).default("embedded"), OPENCODE_MODE: z.literal("embedded").default("embedded"),
// embedded opencode server 的监听地址。 // embedded opencode server 的监听地址。
OPENCODE_HOSTNAME: z.string().default("127.0.0.1"), OPENCODE_HOSTNAME: z.string().default("127.0.0.1"),
// embedded opencode server 的监听端口。 // embedded opencode server 的监听端口。
@@ -55,10 +55,6 @@ const envSchema = z
OPENCODE_MODEL_OPTIONS: z.string().default(defaultAgentModelOptionsJson), OPENCODE_MODEL_OPTIONS: z.string().default(defaultAgentModelOptionsJson),
// opencode skills 树目录;会在运行时解析为绝对路径,避免工具 cwd 偏移。 // opencode skills 树目录;会在运行时解析为绝对路径,避免工具 cwd 偏移。
OPENCODE_SKILLS_ROOT_DIR: z.string().default("./.opencode/skills"), OPENCODE_SKILLS_ROOT_DIR: z.string().default("./.opencode/skills"),
// client 模式下,目标 opencode server 的基础地址。
OPENCODE_CLIENT_BASE_URL: z.string().url().optional(),
// 旧版 client 模式环境变量名,保留兼容,解析时会映射到 OPENCODE_CLIENT_BASE_URL。
OPENCODE_BASE_URL: z.string().url().optional(),
// tjwater-cli 可执行文件路径。 // tjwater-cli 可执行文件路径。
TJWATER_CLI_PATH: z.string().default("./cli/tjwater-cli"), TJWATER_CLI_PATH: z.string().default("./cli/tjwater-cli"),
// TJWater 后端 API 的基础地址。 // TJWater 后端 API 的基础地址。
@@ -117,13 +113,6 @@ const envSchema = z
.default(3600000), .default(3600000),
}) })
.superRefine((env, ctx) => { .superRefine((env, ctx) => {
if (env.OPENCODE_MODE === "client" && !env.OPENCODE_CLIENT_BASE_URL) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
path: ["OPENCODE_CLIENT_BASE_URL"],
message: "OPENCODE_CLIENT_BASE_URL is required when OPENCODE_MODE=client",
});
}
let modelOptions; let modelOptions;
try { try {
modelOptions = parseAgentModelOptions(env.OPENCODE_MODEL_OPTIONS); modelOptions = parseAgentModelOptions(env.OPENCODE_MODEL_OPTIONS);
@@ -154,15 +143,4 @@ const envSchema = z
export type AppConfig = z.infer<typeof envSchema>; export type AppConfig = z.infer<typeof envSchema>;
const normalizedEnv = { export const config: AppConfig = envSchema.parse(process.env);
...process.env,
OPENCODE_MODE:
process.env.OPENCODE_MODE ??
(process.env.OPENCODE_CLIENT_BASE_URL || process.env.OPENCODE_BASE_URL
? "client"
: "embedded"),
OPENCODE_CLIENT_BASE_URL:
process.env.OPENCODE_CLIENT_BASE_URL ?? process.env.OPENCODE_BASE_URL,
};
export const config: AppConfig = envSchema.parse(normalizedEnv);
+18
View File
@@ -246,6 +246,24 @@ register("/api/v1/agent/sessions/{session_id}/runs/current", "delete", {
request: { params: SessionId }, request: { params: SessionId },
responses: { 202: jsonResponse(JsonObject), 204: { description: "No active run" } }, responses: { 202: jsonResponse(JsonObject), 204: { description: "No active run" } },
}); });
register(
"/api/v1/agent/sessions/{session_id}/credential-refreshes",
"post",
{
summary: "Resume a waiting agent tool call with refreshed credentials",
request: {
params: SessionId,
body: {
content: {
"application/json": {
schema: z.object({ request_id: z.string().min(1).max(128) }),
},
},
},
},
responses: { 202: jsonResponse(JsonObject) },
},
);
register( register(
"/api/v1/agent/sessions/{session_id}/permission-responses", "/api/v1/agent/sessions/{session_id}/permission-responses",
"post", "post",
+1 -1
View File
@@ -77,12 +77,12 @@ type TurnReviewInput = {
export class LearningOrchestrator { export class LearningOrchestrator {
private readonly activeReviews = new Set<string>(); private readonly activeReviews = new Set<string>();
private readonly sessionLearningStateStore = new SessionLearningStateStore(); private readonly sessionLearningStateStore = new SessionLearningStateStore();
private readonly skillStore = new SkillStore();
constructor( constructor(
private readonly runtime: OpencodeRuntimeAdapter, private readonly runtime: OpencodeRuntimeAdapter,
private readonly memoryStore: MemoryStore, private readonly memoryStore: MemoryStore,
private readonly transcriptStore: SessionTranscriptStore, private readonly transcriptStore: SessionTranscriptStore,
private readonly skillStore: SkillStore,
) {} ) {}
async initialize() { async initialize() {
+194
View File
@@ -0,0 +1,194 @@
import { type MemoryScope, MemoryStore } from "../memory/store.js";
import {
setRuntimeSessionContext,
type RuntimeSessionContext,
} from "../runtime/sessionContext.js";
import { SkillStore } from "../skills/store.js";
export type MemoryManagerInput = {
action: "add" | "list" | "replace" | "remove";
content?: string;
scope: string;
target_id?: string;
};
export type SkillManagerInput = {
action:
| "list"
| "write_skill"
| "remove_skill"
| "append_pattern"
| "remove_pattern"
| "write_reference"
| "remove_reference"
| "write_script"
| "remove_script";
content?: string;
file_path?: string;
pattern?: string;
skill_path: string;
target_id?: string;
};
export const executeMemoryManager = async (
memoryStore: MemoryStore,
sessionContext: RuntimeSessionContext,
input: MemoryManagerInput,
) => {
const scope: MemoryScope | null =
input.scope === "user"
? "user"
: input.scope === "workspace"
? "workspace"
: null;
if (!scope) {
return rejected(
"memory",
`unsupported scope: ${input.scope}; use exact keyword 'user' or 'workspace'`,
);
}
if (sessionContext.allowLearningWrite === false && input.action !== "list") {
return rejected("memory", "memory writes are disabled for this session");
}
const scopeKey =
scope === "user" ? sessionContext.actorKey : sessionContext.projectKey;
if (input.action === "list") {
setRuntimeSessionContext({
...sessionContext,
memoryListReadScopes: {
...(sessionContext.memoryListReadScopes ?? {}),
[scope]: true,
},
});
return {
ok: true,
kind: "memory",
decision: "accepted",
detail: "memory listed",
items: await memoryStore.list(scope, scopeKey),
target: scope,
};
}
if (input.action === "add") {
if (sessionContext.memoryListReadScopes?.[scope] !== true) {
return {
...rejected(
"memory",
`must list ${scope} memory and review existing entries before add`,
),
target: scope,
};
}
const result = await memoryStore.upsert(scope, scopeKey, {
content: input.content ?? "",
sessionId: sessionContext.clientSessionId,
source: "tool",
traceId: sessionContext.traceId,
});
if (!result.entry) {
return rejected("memory", "content rejected by persistence policy");
}
return {
ok: true,
kind: "memory",
decision: result.changed ? "accepted" : "deduped",
detail: result.detail,
entry: result.entry,
target: scope,
};
}
const result =
input.action === "replace"
? await memoryStore.replace(scope, scopeKey, input.target_id ?? "", {
content: input.content ?? "",
sessionId: sessionContext.clientSessionId,
source: "tool",
traceId: sessionContext.traceId,
})
: await memoryStore.remove(scope, scopeKey, input.target_id ?? "");
return {
ok: true,
kind: "memory",
decision: result.changed ? "accepted" : "rejected",
detail: result.detail,
target: scope,
};
};
export const executeSkillManager = async (
skillStore: SkillStore,
sessionContext: RuntimeSessionContext,
input: SkillManagerInput,
) => {
if (sessionContext.allowLearningWrite === false && input.action !== "list") {
return rejected("skill", "skill writes are disabled for this session");
}
if (input.action === "list") {
const result = await skillStore.list(input.skill_path);
if (!result) {
return rejected(
"skill",
"invalid skill_path; expected a relative path under .opencode/skills",
);
}
return {
ok: true,
kind: "skill",
decision: "accepted",
detail: "skill listed",
references: result.references,
scripts: result.scripts,
skill_path: result.skillPath,
target: result.target,
patterns: result.patterns,
};
}
const result =
input.action === "write_skill"
? await skillStore.writeSkill(input.skill_path, input.content ?? "")
: input.action === "remove_skill"
? await skillStore.removeSkill(input.skill_path)
: input.action === "append_pattern"
? await skillStore.appendPattern(input.skill_path, input.pattern ?? "")
: input.action === "remove_pattern"
? await skillStore.removePattern(input.skill_path, input.target_id ?? "")
: input.action === "write_reference"
? await skillStore.writeReference(
input.skill_path,
input.file_path ?? "",
input.content ?? "",
)
: input.action === "remove_reference"
? await skillStore.removeReference(
input.skill_path,
input.file_path ?? "",
)
: input.action === "write_script"
? await skillStore.writeScript(
input.skill_path,
input.file_path ?? "",
input.content ?? "",
)
: await skillStore.removeScript(
input.skill_path,
input.file_path ?? "",
);
return {
ok: true,
kind: "skill",
decision: result.changed ? "accepted" : "rejected",
detail: result.detail,
target: result.target,
};
};
const rejected = (kind: "memory" | "skill", detail: string) => ({
ok: true,
kind,
decision: "rejected",
detail,
});
+48
View File
@@ -2,6 +2,7 @@ import { Router } from "express";
import { z } from "zod"; import { z } from "zod";
import { getAgentAuthContext } from "../auth/agentAuth.js"; import { getAgentAuthContext } from "../auth/agentAuth.js";
import { type CredentialRefreshCoordinator } from "../auth/credentialRefresh.js";
import { import {
agentModelOptions, agentModelOptions,
isSupportedModel, isSupportedModel,
@@ -121,6 +122,7 @@ export const buildChatRouter = (
sessionTranscriptStore: SessionTranscriptStore, sessionTranscriptStore: SessionTranscriptStore,
learningOrchestrator: LearningOrchestrator, learningOrchestrator: LearningOrchestrator,
resultReferenceResolver: ResultReferenceResolver, resultReferenceResolver: ResultReferenceResolver,
credentialRefreshCoordinator: CredentialRefreshCoordinator,
) => { ) => {
const chatRouter = Router(); const chatRouter = Router();
@@ -295,6 +297,16 @@ export const buildChatRouter = (
}, },
}; };
run.subscribers.add(subscriber); run.subscribers.add(subscriber);
const pendingCredentialRefresh =
credentialRefreshCoordinator.getPendingEvent(sessionRecord.sessionId);
if (pendingCredentialRefresh) {
subscriber.write(pendingCredentialRefresh.type, {
session_id: sessionRecord.sessionId,
request_id: pendingCredentialRefresh.requestId,
reason: pendingCredentialRefresh.reason,
timeout_ms: pendingCredentialRefresh.timeoutMs,
});
}
const cleanup = () => { const cleanup = () => {
run.subscribers.delete(subscriber); run.subscribers.delete(subscriber);
@@ -390,6 +402,7 @@ export const buildChatRouter = (
registerChatInteractionRoutes(chatRouter, { registerChatInteractionRoutes(chatRouter, {
activeRuns, activeRuns,
credentialRefreshCoordinator,
runtime, runtime,
sessionMetadataStore, sessionMetadataStore,
sessionUiStateStore, sessionUiStateStore,
@@ -803,6 +816,35 @@ export const buildChatRouter = (
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
}); });
}; };
const unsubscribeCredentialRefresh = credentialRefreshCoordinator.subscribe(
binding.sessionId,
(event) => {
publish(event.type, {
session_id: clientSessionId,
request_id: event.requestId,
...(event.type === "credential_refresh_required"
? {
reason: event.reason,
timeout_ms: event.timeoutMs,
}
: {}),
...(event.type === "credential_refresh_failed"
? { message: event.message }
: {}),
});
},
);
const cancelCredentialRefreshOnAbort = () => {
credentialRefreshCoordinator.cancelSession(
binding.sessionId,
"credential refresh cancelled because the agent run was aborted",
);
};
abortController.signal.addEventListener(
"abort",
cancelCredentialRefreshOnAbort,
{ once: true },
);
try { try {
const preparedMessage = await buildPromptWithLearningContext( const preparedMessage = await buildPromptWithLearningContext(
@@ -925,6 +967,12 @@ export const buildChatRouter = (
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
}); });
sessionBridge.finalizeRequest(clientSessionId); sessionBridge.finalizeRequest(clientSessionId);
abortController.signal.removeEventListener(
"abort",
cancelCredentialRefreshOnAbort,
);
credentialRefreshCoordinator.cancelSession(binding.sessionId);
unsubscribeCredentialRefresh();
activeRun.status = abortController.signal.aborted activeRun.status = abortController.signal.aborted
? activeRun.status === "aborted" ? activeRun.status === "aborted"
? "aborted" ? "aborted"
+72
View File
@@ -2,8 +2,13 @@ import { type Router } from "express";
import { z } from "zod"; import { z } from "zod";
import { getAgentAuthContext } from "../auth/agentAuth.js"; import { getAgentAuthContext } from "../auth/agentAuth.js";
import { type CredentialRefreshCoordinator } from "../auth/credentialRefresh.js";
import { logger } from "../logger.js"; import { logger } from "../logger.js";
import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
import {
getRuntimeSessionContext,
setRuntimeSessionContext,
} from "../runtime/sessionContext.js";
import { type SessionMetadataStore } from "../sessions/metadataStore.js"; import { type SessionMetadataStore } from "../sessions/metadataStore.js";
import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; import { type SessionUiStateStore } from "../sessions/uiStateStore.js";
import { toActorKey, toProjectKey } from "../utils/fileStore.js"; import { toActorKey, toProjectKey } from "../utils/fileStore.js";
@@ -26,8 +31,13 @@ const questionReplyPayloadSchema = z.object({
answers: z.array(z.array(z.string().max(2000))).default([]), answers: z.array(z.array(z.string().max(2000))).default([]),
}); });
const credentialRefreshPayloadSchema = z.object({
request_id: z.string().min(1).max(128),
});
type RegisterInteractionRoutesOptions = { type RegisterInteractionRoutesOptions = {
activeRuns: Map<string, ActiveRun>; activeRuns: Map<string, ActiveRun>;
credentialRefreshCoordinator: CredentialRefreshCoordinator;
runtime: OpencodeRuntimeAdapter; runtime: OpencodeRuntimeAdapter;
sessionMetadataStore: SessionMetadataStore; sessionMetadataStore: SessionMetadataStore;
sessionUiStateStore: SessionUiStateStore; sessionUiStateStore: SessionUiStateStore;
@@ -41,11 +51,73 @@ export const registerChatInteractionRoutes = (
chatRouter: Router, chatRouter: Router,
{ {
activeRuns, activeRuns,
credentialRefreshCoordinator,
runtime, runtime,
sessionMetadataStore, sessionMetadataStore,
sessionUiStateStore, sessionUiStateStore,
}: RegisterInteractionRoutesOptions, }: RegisterInteractionRoutesOptions,
) => { ) => {
chatRouter.post("/sessions/:session_id/credential-refreshes", async (req, res) => {
const parsed = credentialRefreshPayloadSchema.safeParse(req.body);
if (!parsed.success) {
res.status(400).json({
message: "invalid request payload",
detail: parsed.error.flatten(),
});
return;
}
const authContext = getAgentAuthContext(req);
const actorKey = toActorKey(authContext.userId);
const projectKey = toProjectKey(authContext.projectId);
const sessionRecord = await sessionMetadataStore.get(
{
actorKey,
projectId: authContext.projectId,
projectKey,
userId: authContext.userId,
},
req.params.session_id,
);
if (!sessionRecord) {
res.status(404).json({ message: "session not found" });
return;
}
const current = getRuntimeSessionContext(sessionRecord.sessionId);
if (!current || current.actorKey !== actorKey || current.projectKey !== projectKey) {
res.status(409).json({ message: "runtime session context unavailable" });
return;
}
if (
credentialRefreshCoordinator.getPendingRequestId(sessionRecord.sessionId) !==
parsed.data.request_id
) {
res.status(409).json({ message: "credential refresh request is no longer pending" });
return;
}
const refreshedContext = {
...current,
accessToken: authContext.accessToken,
authExpired: undefined,
network: authContext.network,
projectId: authContext.projectId,
tokenExpiresAt: authContext.tokenExpiresAt,
traceId: req.header("x-trace-id")?.trim() || current.traceId,
};
setRuntimeSessionContext(refreshedContext);
credentialRefreshCoordinator.resolve(
sessionRecord.sessionId,
parsed.data.request_id,
refreshedContext,
);
res.status(202).json({
session_id: sessionRecord.sessionId,
request_id: parsed.data.request_id,
status: "accepted",
});
});
chatRouter.post("/sessions/:session_id/permission-responses", async (req, res) => { chatRouter.post("/sessions/:session_id/permission-responses", async (req, res) => {
const parsed = permissionReplyPayloadSchema.safeParse(req.body); const parsed = permissionReplyPayloadSchema.safeParse(req.body);
if (!parsed.success) { if (!parsed.success) {
-116
View File
@@ -1,116 +0,0 @@
import {
getRuntimeSessionContext,
setRuntimeSessionContext,
type RuntimeSessionContext,
} from "./sessionContext.js";
type FetchLike = (
input: string | URL | Request,
init?: RequestInit,
) => Promise<Response>;
type InternalSessionContextClientOptions = {
baseUrl?: string;
internalToken?: string;
fetchImpl?: FetchLike;
};
type InternalSessionContextPayload = {
actor_key: string;
allow_learning_write?: boolean;
client_session_id: string;
memory_list_read_scopes?: Partial<Record<"user" | "workspace", boolean>>;
project_key: string;
session_id: string;
trace_id: string;
};
export const serializeRuntimeSessionContext = (
context: RuntimeSessionContext,
): InternalSessionContextPayload => ({
actor_key: context.actorKey,
allow_learning_write: context.allowLearningWrite,
client_session_id: context.clientSessionId,
memory_list_read_scopes: context.memoryListReadScopes,
project_key: context.projectKey,
session_id: context.sessionId,
trace_id: context.traceId,
});
const requireString = (
value: unknown,
field: keyof InternalSessionContextPayload,
) => {
if (typeof value !== "string" || value.length === 0) {
throw new Error(`invalid internal session context field: ${field}`);
}
return value;
};
const parseRuntimeSessionContext = (
value: unknown,
): RuntimeSessionContext => {
if (!value || typeof value !== "object" || Array.isArray(value)) {
throw new Error("invalid internal session context response");
}
const payload = value as Record<string, unknown>;
const readScopes = payload.memory_list_read_scopes;
return {
actorKey: requireString(payload.actor_key, "actor_key"),
allowLearningWrite:
typeof payload.allow_learning_write === "boolean"
? payload.allow_learning_write
: undefined,
clientSessionId: requireString(
payload.client_session_id,
"client_session_id",
),
memoryListReadScopes:
readScopes && typeof readScopes === "object" && !Array.isArray(readScopes)
? (readScopes as RuntimeSessionContext["memoryListReadScopes"])
: undefined,
projectKey: requireString(payload.project_key, "project_key"),
sessionId: requireString(payload.session_id, "session_id"),
traceId: requireString(payload.trace_id, "trace_id"),
};
};
export const readBridgedRuntimeSessionContext = async (
sessionId: string,
options: InternalSessionContextClientOptions = {},
) => {
const localContext = getRuntimeSessionContext(sessionId);
if (localContext) {
return localContext;
}
const baseUrl = (
options.baseUrl ??
process.env.TJWATER_AGENT_INTERNAL_BASE_URL ??
"http://127.0.0.1:8787"
).replace(/\/+$/, "");
const internalToken =
options.internalToken ?? process.env.TJWATER_AGENT_INTERNAL_TOKEN ?? "";
const response = await (options.fetchImpl ?? fetch)(
`${baseUrl}/internal/tools/session-context`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
"x-agent-internal-token": internalToken,
},
body: JSON.stringify({ session_id: sessionId }),
},
);
const text = await response.text();
if (!response.ok) {
throw new Error(text || `session context bridge failed: ${response.status}`);
}
const context = parseRuntimeSessionContext(JSON.parse(text));
if (context.sessionId !== sessionId) {
throw new Error("internal session context id mismatch");
}
setRuntimeSessionContext(context);
return context;
};
+1 -15
View File
@@ -1,6 +1,5 @@
import { import {
createOpencode, createOpencode,
createOpencodeClient,
type OpencodeClient, type OpencodeClient,
} from "@opencode-ai/sdk/v2"; } from "@opencode-ai/sdk/v2";
import { existsSync, readFileSync } from "node:fs"; import { existsSync, readFileSync } from "node:fs";
@@ -387,19 +386,6 @@ export class OpencodeRuntimeAdapter {
} }
private async bootstrapClient(): Promise<OpencodeClient> { private async bootstrapClient(): Promise<OpencodeClient> {
if (config.OPENCODE_MODE === "client") {
logger.info(
{
baseUrl: config.OPENCODE_CLIENT_BASE_URL,
mode: config.OPENCODE_MODE,
},
"connecting to opencode server in client mode",
);
return createOpencodeClient({
baseUrl: config.OPENCODE_CLIENT_BASE_URL,
});
}
// embedded 模式下,把服务内工具桥地址注入到 opencode 进程环境里, // embedded 模式下,把服务内工具桥地址注入到 opencode 进程环境里,
// 这样 .opencode/tools 下的自定义工具可以回调本服务。 // 这样 .opencode/tools 下的自定义工具可以回调本服务。
process.env.TJWATER_AGENT_INTERNAL_BASE_URL = `http://127.0.0.1:${config.PORT}`; process.env.TJWATER_AGENT_INTERNAL_BASE_URL = `http://127.0.0.1:${config.PORT}`;
@@ -430,7 +416,7 @@ export class OpencodeRuntimeAdapter {
} catch (error) { } catch (error) {
if (isMissingOpencodeCli(error)) { if (isMissingOpencodeCli(error)) {
throw new Error( throw new Error(
"embedded mode requires the opencode CLI to be installed and available in PATH; otherwise set OPENCODE_MODE=client and provide OPENCODE_CLIENT_BASE_URL", "embedded mode requires the opencode CLI to be installed and available in PATH",
); );
} }
throw error; throw error;
+252 -98
View File
@@ -5,6 +5,11 @@ import express from "express";
import { requireAgentAuth } from "./auth/agentAuth.js"; import { requireAgentAuth } from "./auth/agentAuth.js";
import { buildBackendContextHeaders } from "./auth/backendContextHeaders.js"; import { buildBackendContextHeaders } from "./auth/backendContextHeaders.js";
import {
CredentialRefreshError,
CredentialRefreshCoordinator,
runWithCredentialRefresh,
} from "./auth/credentialRefresh.js";
import { SessionTranscriptStore } from "./sessions/transcriptStore.js"; import { SessionTranscriptStore } from "./sessions/transcriptStore.js";
import { ChatSessionBridge } from "./chat/sessionBridge.js"; import { ChatSessionBridge } from "./chat/sessionBridge.js";
import { config } from "./config.js"; import { config } from "./config.js";
@@ -12,6 +17,12 @@ import { SessionUiStateStore } from "./sessions/uiStateStore.js";
import { SessionMetadataStore } from "./sessions/metadataStore.js"; import { SessionMetadataStore } from "./sessions/metadataStore.js";
import { logger } from "./logger.js"; import { logger } from "./logger.js";
import { LearningOrchestrator } from "./learning/orchestrator.js"; import { LearningOrchestrator } from "./learning/orchestrator.js";
import {
executeMemoryManager,
executeSkillManager,
type MemoryManagerInput,
type SkillManagerInput,
} from "./learning/toolManagers.js";
import { MemoryStore } from "./memory/store.js"; import { MemoryStore } from "./memory/store.js";
import { ResultReferenceResolver } from "./results/resolver.js"; import { ResultReferenceResolver } from "./results/resolver.js";
import { import {
@@ -21,12 +32,12 @@ import {
import { buildChatRouter } from "./routes/chat.js"; import { buildChatRouter } from "./routes/chat.js";
import { buildAgentPublicRouter } from "./routes/publicApi.js"; import { buildAgentPublicRouter } from "./routes/publicApi.js";
import { opencodeRuntime } from "./runtime/opencode.js"; import { opencodeRuntime } from "./runtime/opencode.js";
import { serializeRuntimeSessionContext } from "./runtime/internalSessionContextBridge.js";
import { import {
getRuntimeSessionContext, getRuntimeSessionContext,
markRuntimeSessionAuthExpired, markRuntimeSessionAuthExpired,
type RuntimeSessionContext, type RuntimeSessionContext,
} from "./runtime/sessionContext.js"; } from "./runtime/sessionContext.js";
import { SkillStore } from "./skills/store.js";
const app = express(); const app = express();
@@ -35,15 +46,18 @@ const sessionBridge = new ChatSessionBridge(opencodeRuntime);
const sessionMetadataStore = new SessionMetadataStore(); const sessionMetadataStore = new SessionMetadataStore();
const sessionUiStateStore = new SessionUiStateStore(); const sessionUiStateStore = new SessionUiStateStore();
const memoryStore = new MemoryStore(); const memoryStore = new MemoryStore();
const skillStore = new SkillStore();
const sessionTranscriptStore = new SessionTranscriptStore(); const sessionTranscriptStore = new SessionTranscriptStore();
const learningOrchestrator = new LearningOrchestrator( const learningOrchestrator = new LearningOrchestrator(
opencodeRuntime, opencodeRuntime,
memoryStore, memoryStore,
sessionTranscriptStore, sessionTranscriptStore,
skillStore,
); );
const resultReferenceStore = new ResultReferenceStore(); const resultReferenceStore = new ResultReferenceStore();
const resultReferenceResolver = new ResultReferenceResolver(resultReferenceStore); const resultReferenceResolver = new ResultReferenceResolver(resultReferenceStore);
const internalToken = config.AGENT_INTERNAL_TOKEN ?? randomUUID(); const internalToken = config.AGENT_INTERNAL_TOKEN ?? randomUUID();
const credentialRefreshCoordinator = new CredentialRefreshCoordinator();
// 这个 token 只用于 OpenCode 子进程回调本服务的内部工具桥。 // 这个 token 只用于 OpenCode 子进程回调本服务的内部工具桥。
process.env.TJWATER_AGENT_INTERNAL_TOKEN = internalToken; process.env.TJWATER_AGENT_INTERNAL_TOKEN = internalToken;
@@ -74,7 +88,7 @@ app.get("/health", async (_req, res) => {
} }
}); });
app.post("/internal/tools/session-context", (req, res) => { app.post("/internal/tools/memory-manager", async (req, res) => {
if (req.header("x-agent-internal-token") !== internalToken) { if (req.header("x-agent-internal-token") !== internalToken) {
res.status(403).json({ message: "forbidden" }); res.status(403).json({ message: "forbidden" });
return; return;
@@ -90,8 +104,86 @@ app.post("/internal/tools/session-context", (req, res) => {
}); });
return; return;
} }
const action = req.body?.action;
if (
typeof action !== "string" ||
!["add", "list", "replace", "remove"].includes(action) ||
typeof req.body?.scope !== "string"
) {
res.status(400).json({ message: "invalid memory manager request" });
return;
}
try {
res.json(
await executeMemoryManager(memoryStore, context, {
action: action as MemoryManagerInput["action"],
content:
typeof req.body?.content === "string" ? req.body.content : undefined,
scope: req.body.scope,
target_id:
typeof req.body?.target_id === "string" ? req.body.target_id : undefined,
}),
);
} catch (error) {
res.status(500).json({
message: "memory manager failed",
detail: error instanceof Error ? error.message : String(error),
});
}
});
res.json(serializeRuntimeSessionContext(context)); app.post("/internal/tools/skill-manager", async (req, res) => {
if (req.header("x-agent-internal-token") !== internalToken) {
res.status(403).json({ message: "forbidden" });
return;
}
const sessionId =
typeof req.body?.session_id === "string" ? req.body.session_id.trim() : "";
const context = sessionId ? getRuntimeSessionContext(sessionId) : null;
if (!context) {
res.status(404).json({ message: "session context not found", detail: sessionId });
return;
}
const action = req.body?.action;
if (
typeof action !== "string" ||
![
"list",
"write_skill",
"remove_skill",
"append_pattern",
"remove_pattern",
"write_reference",
"remove_reference",
"write_script",
"remove_script",
].includes(action) ||
typeof req.body?.skill_path !== "string"
) {
res.status(400).json({ message: "invalid skill manager request" });
return;
}
try {
res.json(
await executeSkillManager(skillStore, context, {
action: action as SkillManagerInput["action"],
content:
typeof req.body?.content === "string" ? req.body.content : undefined,
file_path:
typeof req.body?.file_path === "string" ? req.body.file_path : undefined,
pattern:
typeof req.body?.pattern === "string" ? req.body.pattern : undefined,
skill_path: req.body.skill_path,
target_id:
typeof req.body?.target_id === "string" ? req.body.target_id : undefined,
}),
);
} catch (error) {
res.status(500).json({
message: "skill manager failed",
detail: error instanceof Error ? error.message : String(error),
});
}
}); });
app.post("/internal/tools/tjwater-cli-call", async (req, res) => { app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
@@ -110,15 +202,6 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
}); });
return; return;
} }
if (isRuntimeAuthExpired(context)) {
markAuthExpired(context, "access_token_expired");
res.status(401).json({
message: "access token expired; refresh chat context",
detail: sessionId,
});
return;
}
const command = typeof req.body?.command === "string" ? req.body.command.trim() : ""; const command = typeof req.body?.command === "string" ? req.body.command.trim() : "";
if (!command) { if (!command) {
res.status(400).json({ message: "command is required" }); res.status(400).json({ message: "command is required" });
@@ -136,46 +219,35 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
return; return;
} }
const authJson = JSON.stringify({ let result;
server: config.TJWATER_API_BASE_URL, try {
access_token: context.accessToken, result = await runWithCredentialRefresh(
project_id: context.projectId, credentialRefreshCoordinator,
context,
(activeContext) => executeCliCommand(activeContext, command, timeoutSec),
);
} catch (error) {
if (!(error instanceof CredentialRefreshError)) {
const detail = error instanceof Error ? error.message : String(error);
res.status(502).json({
message: "CLI execution failed",
detail,
}); });
return;
}
if (error.code === "cancelled") {
res.status(409).json({ message: "agent run was aborted" });
return;
}
markAuthExpired(context, "access_token_expired");
res.status(401).json({
message: "credential refresh failed",
detail: error instanceof Error ? error.message : String(error),
});
return;
}
const cliArgs = ["--auth-stdin", ...command.split(/\s+/).filter(Boolean)]; if (result.status === 504) {
const child = spawn(config.TJWATER_CLI_PATH, cliArgs, {
stdio: ["pipe", "pipe", "pipe"],
});
let stdout = "";
let stderr = "";
child.stdout.on("data", (data: Buffer) => {
stdout += data.toString("utf-8");
});
child.stderr.on("data", (data: Buffer) => {
stderr += data.toString("utf-8");
});
child.stdin.write(authJson);
child.stdin.end();
const exitCode = await new Promise<number | null>((resolve, reject) => {
const timer = setTimeout(() => {
child.kill("SIGTERM");
resolve(-1);
}, timeoutSec * 1000);
child.on("close", (code) => {
clearTimeout(timer);
resolve(code);
});
child.on("error", (err) => {
clearTimeout(timer);
reject(err);
});
});
if (exitCode === -1) {
res.status(504).json({ res.status(504).json({
ok: false, ok: false,
schema_version: "tjwater-cli/v1", schema_version: "tjwater-cli/v1",
@@ -189,29 +261,102 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
return; return;
} }
if (exitCode !== 0) { if (result.status === 401) {
res.status(502).json({ markAuthExpired(
getRuntimeSessionContext(sessionId) ?? context,
"access_token_rejected",
);
}
if (result.exitCode !== 0) {
res
.status(result.status)
.type("application/json")
.send(
result.stdout ||
JSON.stringify({
ok: false, ok: false,
exit_code: exitCode, exit_code: result.exitCode,
stderr: stderr.slice(0, 2000), stderr: result.stderr.slice(0, 2000),
stdout: stdout.slice(0, 2000), message: `CLI exited with code ${result.exitCode}`,
message: `CLI exited with code ${exitCode}`, }),
}); );
return; return;
} }
try { try {
res.json(JSON.parse(stdout)); res.json(JSON.parse(result.stdout));
} catch { } catch {
res.json({ res.json({
ok: true, ok: true,
schema_version: "tjwater-cli/v1", schema_version: "tjwater-cli/v1",
raw: stdout, raw: result.stdout,
stderr: stderr || undefined, stderr: result.stderr || undefined,
}); });
} }
}); });
const executeCliCommand = async (
context: RuntimeSessionContext,
command: string,
timeoutSec: number,
) => {
const child = spawn(
config.TJWATER_CLI_PATH,
["--auth-stdin", ...command.split(/\s+/).filter(Boolean)],
{ stdio: ["pipe", "pipe", "pipe"] },
);
let stdout = "";
let stderr = "";
child.stdout.on("data", (data: Buffer) => {
stdout += data.toString("utf-8");
});
child.stderr.on("data", (data: Buffer) => {
stderr += data.toString("utf-8");
});
child.stdin.write(
JSON.stringify({
server: config.TJWATER_API_BASE_URL,
access_token: context.accessToken,
project_id: context.projectId,
}),
);
child.stdin.end();
const exitCode = await new Promise<number | null>((resolve, reject) => {
const timer = setTimeout(() => {
child.kill("SIGTERM");
resolve(-1);
}, timeoutSec * 1000);
child.on("close", (code) => {
clearTimeout(timer);
resolve(code);
});
child.on("error", (error) => {
clearTimeout(timer);
reject(error);
});
});
let errorCode = "";
try {
const payload = JSON.parse(stdout) as { error?: { code?: unknown } };
errorCode =
typeof payload.error?.code === "string" ? payload.error.code : "";
} catch {
errorCode = "";
}
const status =
exitCode === -1
? 504
: errorCode === "HTTP_401" || errorCode === "UNAUTHENTICATED"
? 401
: errorCode === "HTTP_403"
? 403
: exitCode === 0
? 200
: 502;
return { exitCode, status, stderr, stdout };
};
app.post("/internal/tools/store-render-ref", async (req, res) => { app.post("/internal/tools/store-render-ref", async (req, res) => {
if (req.header("x-agent-internal-token") !== internalToken) { if (req.header("x-agent-internal-token") !== internalToken) {
res.status(403).json({ message: "forbidden" }); res.status(403).json({ message: "forbidden" });
@@ -302,39 +447,60 @@ const callBackendJson = async (
context: RuntimeSessionContext, context: RuntimeSessionContext,
payload: unknown, payload: unknown,
) => { ) => {
if (isRuntimeAuthExpired(context)) { try {
const result = await runWithCredentialRefresh(
credentialRefreshCoordinator,
context,
async (activeContext) => {
const controller = new AbortController();
const timer = setTimeout(
() => controller.abort(),
config.TJWATER_API_TIMEOUT_MS,
);
try {
const response = await fetch(
new URL(path, config.TJWATER_API_BASE_URL),
{
method: "POST",
headers: buildBackendContextHeaders(activeContext),
body: JSON.stringify(payload),
signal: controller.signal,
},
);
return {
ok: response.ok,
status: response.status,
text: await response.text(),
};
} finally {
clearTimeout(timer);
}
},
);
if (result.status === 401) {
markAuthExpired(
getRuntimeSessionContext(context.sessionId) ?? context,
"access_token_rejected",
);
}
return result;
} catch (error) {
if (!(error instanceof CredentialRefreshError)) {
throw error;
}
if (error.code === "cancelled") {
throw error;
}
markAuthExpired(context, "access_token_expired"); markAuthExpired(context, "access_token_expired");
return { return {
ok: false, ok: false,
status: 401, status: 401,
text: JSON.stringify({ text: JSON.stringify({
message: "access token expired; refresh chat context", message: "credential refresh failed",
detail: error instanceof Error ? error.message : String(error),
}), }),
}; };
} }
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), config.TJWATER_API_TIMEOUT_MS);
try {
const headers = buildBackendContextHeaders(context);
const response = await fetch(new URL(path, config.TJWATER_API_BASE_URL), {
method: "POST",
headers,
body: JSON.stringify(payload),
signal: controller.signal,
});
const text = await response.text();
if (response.status === 401) {
markAuthExpired(context, "access_token_rejected");
}
return {
ok: response.ok,
status: response.status,
text,
};
} finally {
clearTimeout(timer);
}
}; };
const parseStringArray = (value: unknown) => const parseStringArray = (value: unknown) =>
@@ -357,19 +523,6 @@ const normalizeWebSearchFreshness = (value: unknown) => {
return webSearchFreshnessMap[value] ?? value; return webSearchFreshnessMap[value] ?? value;
}; };
const AUTH_EXPIRY_SKEW_MS = 30_000;
function isRuntimeAuthExpired(context: RuntimeSessionContext) {
if (!context.tokenExpiresAt) {
return false;
}
const expiresAt = Date.parse(context.tokenExpiresAt);
if (!Number.isFinite(expiresAt)) {
return false;
}
return Date.now() >= expiresAt - AUTH_EXPIRY_SKEW_MS;
}
function markAuthExpired( function markAuthExpired(
context: RuntimeSessionContext, context: RuntimeSessionContext,
reason: NonNullable<RuntimeSessionContext["authExpired"]>["reason"], reason: NonNullable<RuntimeSessionContext["authExpired"]>["reason"],
@@ -491,6 +644,7 @@ const chatRouter = buildChatRouter(
sessionTranscriptStore, sessionTranscriptStore,
learningOrchestrator, learningOrchestrator,
resultReferenceResolver, resultReferenceResolver,
credentialRefreshCoordinator,
); );
const authenticatedChatRouter = express.Router(); const authenticatedChatRouter = express.Router();
authenticatedChatRouter.use(requireAgentAuth, chatRouter); authenticatedChatRouter.use(requireAgentAuth, chatRouter);
+100
View File
@@ -0,0 +1,100 @@
import { describe, expect, test } from "bun:test";
import {
CredentialRefreshCoordinator,
CredentialRefreshError,
runWithCredentialRefresh,
} from "../../src/auth/credentialRefresh.js";
import { type RuntimeSessionContext } from "../../src/runtime/sessionContext.js";
const context = (overrides: Partial<RuntimeSessionContext> = {}) => ({
accessToken: "old-token",
actorKey: "user-1",
clientSessionId: "client-1",
projectId: "project-1",
projectKey: "project-1",
sessionId: "session-1",
traceId: "trace-1",
...overrides,
});
describe("CredentialRefreshCoordinator", () => {
test("deduplicates concurrent refreshes for one session", async () => {
const coordinator = new CredentialRefreshCoordinator();
const requestIds: string[] = [];
coordinator.subscribe("session-1", (event) => {
if (event.type === "credential_refresh_required") {
requestIds.push(event.requestId);
}
});
const expired = context({ tokenExpiresAt: new Date(0).toISOString() });
const execute = async (active: RuntimeSessionContext) => ({
status: 200,
token: active.accessToken,
});
const first = runWithCredentialRefresh(coordinator, expired, execute);
const second = runWithCredentialRefresh(coordinator, expired, execute);
await Promise.resolve();
expect(requestIds).toHaveLength(1);
expect(coordinator.getPendingEvent("session-1")).toMatchObject({
type: "credential_refresh_required",
requestId: requestIds[0],
reason: "access_token_expired",
});
coordinator.resolve(
"session-1",
requestIds[0]!,
context({ accessToken: "fresh-token" }),
);
expect(await first).toEqual({ status: 200, token: "fresh-token" });
expect(await second).toEqual({ status: 200, token: "fresh-token" });
expect(coordinator.getPendingEvent("session-1")).toBeNull();
});
test("retries one time on 401 and never refreshes a 403", async () => {
const coordinator = new CredentialRefreshCoordinator();
let requestId = "";
coordinator.subscribe("session-1", (event) => {
if (event.type === "credential_refresh_required") {
requestId = event.requestId;
}
});
let attempts = 0;
const resultPromise = runWithCredentialRefresh(
coordinator,
context(),
async () => ({ status: ++attempts === 1 ? 401 : 401 }),
);
await Promise.resolve();
coordinator.resolve("session-1", requestId, context({ accessToken: "fresh-token" }));
expect((await resultPromise).status).toBe(401);
expect(attempts).toBe(2);
requestId = "";
expect(
(await runWithCredentialRefresh(coordinator, context(), async () => ({ status: 403 })))
.status,
).toBe(403);
expect(requestId).toBe("");
});
test("fails explicitly when no event stream can refresh credentials", async () => {
await expect(
runWithCredentialRefresh(
new CredentialRefreshCoordinator(),
context({ tokenExpiresAt: new Date(0).toISOString() }),
async () => ({ status: 200 }),
),
).rejects.toBeInstanceOf(CredentialRefreshError);
});
test("reports run cancellation separately from authentication failure", async () => {
const coordinator = new CredentialRefreshCoordinator();
coordinator.subscribe("session-1", () => undefined);
const pending = coordinator.request("session-1", "access_token_rejected");
coordinator.cancelSession("session-1");
await expect(pending).rejects.toMatchObject({ code: "cancelled" });
});
});
+1 -1
View File
@@ -47,7 +47,7 @@ describe("Agent REST OpenAPI", () => {
} }
} }
expect(operationCount).toBe(13); expect(operationCount).toBe(14);
}); });
test("models runs as session subresources", () => { test("models runs as session subresources", () => {
+104
View File
@@ -0,0 +1,104 @@
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import {
executeMemoryManager,
executeSkillManager,
} from "../../src/learning/toolManagers.js";
import { MemoryStore } from "../../src/memory/store.js";
import {
getRuntimeSessionContext,
removeRuntimeSessionContext,
setRuntimeSessionContext,
type RuntimeSessionContext,
} from "../../src/runtime/sessionContext.js";
import { SkillStore } from "../../src/skills/store.js";
describe("main-process learning tool managers", () => {
let tempDir: string;
let memoryStore: MemoryStore;
let skillStore: SkillStore;
let context: RuntimeSessionContext;
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), "tjwater-learning-tools-"));
memoryStore = new MemoryStore(
join(tempDir, "memory"),
join(tempDir, "backup", "memory"),
);
skillStore = new SkillStore(
join(tempDir, "skills"),
join(tempDir, "backup", "skills"),
);
await memoryStore.initialize();
context = {
actorKey: "actor-1",
allowLearningWrite: true,
clientSessionId: "client-session-1",
projectKey: "project-1",
sessionId: "session-1",
traceId: "trace-1",
};
setRuntimeSessionContext(context);
});
afterEach(async () => {
removeRuntimeSessionContext(context.sessionId);
await rm(tempDir, { force: true, recursive: true });
});
it("enforces list-before-add using the canonical runtime context", async () => {
const rejected = await executeMemoryManager(memoryStore, context, {
action: "add",
content: "用户偏好查看压力单位为 MPa",
scope: "user",
});
expect(rejected.decision).toBe("rejected");
await executeMemoryManager(memoryStore, context, {
action: "list",
scope: "user",
});
const refreshedContext = getRuntimeSessionContext(context.sessionId)!;
const accepted = await executeMemoryManager(memoryStore, refreshedContext, {
action: "add",
content: "用户偏好查看压力单位为 MPa",
scope: "user",
});
expect(accepted.decision).toBe("accepted");
expect(await memoryStore.list("user", context.actorKey)).toHaveLength(1);
});
it("writes and removes skills through the shared store", async () => {
const content = [
"---",
"name: pressure-review",
"description: Pressure review workflow.",
"---",
"",
"# Pressure Review",
].join("\n");
const written = await executeSkillManager(skillStore, context, {
action: "write_skill",
content,
skill_path: "workflow/pressure-review",
});
expect(written.decision).toBe("accepted");
expect("target" in written).toBe(true);
if (!("target" in written)) throw new Error("write returned no target");
await expect(readFile(written.target, "utf8")).resolves.toContain(
"# Pressure Review\n",
);
const removed = await executeSkillManager(skillStore, context, {
action: "remove_skill",
skill_path: "workflow/pressure-review",
});
expect(removed.decision).toBe("accepted");
expect("target" in removed).toBe(true);
if (!("target" in removed)) throw new Error("remove returned no target");
await expect(readFile(removed.target, "utf8")).rejects.toThrow();
});
});
-140
View File
@@ -1,140 +0,0 @@
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createSkillManagerTool } from "../../.opencode/tools/skill_manager.js";
import { type RuntimeSessionContext } from "../../src/runtime/sessionContext.js";
import { SkillStore } from "../../src/skills/store.js";
describe("skill_manager tool", () => {
let tempDir: string;
let skillStore: SkillStore;
let context: RuntimeSessionContext;
const toolContext = {
abort: new AbortController().signal,
agent: "test",
ask: (() => undefined) as never,
directory: "",
messageID: "message-1",
metadata: () => undefined,
sessionID: "session-1",
worktree: "",
};
const skillDocument = (body: string) =>
[
"---",
"name: pressure-review",
"description: Pressure review workflow.",
"---",
"",
body,
].join("\n");
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), "tjwater-skill-tool-"));
skillStore = new SkillStore(
join(tempDir, "skills"),
join(tempDir, "backup", "skills"),
);
context = {
actorKey: "actor-1",
allowLearningWrite: true,
clientSessionId: "client-session-1",
projectKey: "project-1",
sessionId: "session-1",
traceId: "trace-1",
};
});
afterEach(async () => {
await rm(tempDir, { force: true, recursive: true });
});
it("dispatches skill-level write, overwrite, and remove actions", async () => {
const tool = createSkillManagerTool(
skillStore,
{ read: () => context },
Promise.resolve(),
);
const writeResult = JSON.parse(
await tool.execute(
{
action: "write_skill",
content: skillDocument("# Pressure Review"),
reason: "verified reusable workflow",
skill_path: "workflow/pressure-review",
},
toolContext,
) as string,
);
expect(writeResult.decision).toBe("accepted");
await expect(readFile(writeResult.target, "utf8")).resolves.toContain(
"# Pressure Review\n",
);
const updateResult = JSON.parse(
await tool.execute(
{
action: "write_skill",
content: skillDocument("# Updated Pressure Review"),
reason: "verified reusable workflow overwrite",
skill_path: "workflow/pressure-review",
},
toolContext,
) as string,
);
expect(updateResult.decision).toBe("accepted");
await expect(readFile(updateResult.target, "utf8")).resolves.toContain(
"# Updated Pressure Review\n",
);
const removeResult = JSON.parse(
await tool.execute(
{
action: "remove_skill",
reason: "workflow is obsolete",
skill_path: "workflow/pressure-review",
},
toolContext,
) as string,
);
expect(removeResult.decision).toBe("accepted");
await expect(readFile(removeResult.target, "utf8")).rejects.toThrow();
});
it("writes the root skills index through the reserved alias", async () => {
const tool = createSkillManagerTool(
skillStore,
{ read: () => context },
Promise.resolve(),
);
const writeResult = JSON.parse(
await tool.execute(
{
action: "write_skill",
content: [
"---",
"name: skills",
"description: TJWater Skills root index.",
"---",
"",
"# TJWater Skills",
].join("\n"),
reason: "refresh root skills index",
skill_path: "__root__",
},
toolContext,
) as string,
);
expect(writeResult.decision).toBe("accepted");
await expect(readFile(writeResult.target, "utf8")).resolves.toContain(
"# TJWater Skills\n",
);
});
});
+1
View File
@@ -55,6 +55,7 @@ describe("Agent public REST router", () => {
undefined as never, undefined as never,
undefined as never, undefined as never,
undefined as never, undefined as never,
undefined as never,
); );
const layers = (router as unknown as { stack: RouterLayer[] }).stack; const layers = (router as unknown as { stack: RouterLayer[] }).stack;
const runtimeOperations = layers const runtimeOperations = layers
@@ -1,83 +0,0 @@
import { afterEach, describe, expect, it } from "bun:test";
import {
readBridgedRuntimeSessionContext,
serializeRuntimeSessionContext,
} from "../../src/runtime/internalSessionContextBridge.js";
import { removeRuntimeSessionContext } from "../../src/runtime/sessionContext.js";
describe("readBridgedRuntimeSessionContext", () => {
const sessionId = "remote-opencode-session";
afterEach(() => {
removeRuntimeSessionContext(sessionId);
});
it("hydrates a child-process context through the authenticated internal bridge", async () => {
const calls: Array<{ input: string; init?: RequestInit }> = [];
const context = await readBridgedRuntimeSessionContext(sessionId, {
baseUrl: "http://127.0.0.1:8787",
internalToken: "internal-secret",
fetchImpl: async (input, init) => {
calls.push({ input: String(input), init });
return Response.json({
actor_key: "actor-1",
allow_learning_write: true,
client_session_id: "client-session-1",
project_key: "project-1",
session_id: sessionId,
trace_id: "trace-1",
});
},
});
expect(context).toMatchObject({
actorKey: "actor-1",
allowLearningWrite: true,
clientSessionId: "client-session-1",
projectKey: "project-1",
sessionId,
traceId: "trace-1",
});
expect(calls).toHaveLength(1);
expect(calls[0]?.input).toBe(
"http://127.0.0.1:8787/internal/tools/session-context",
);
expect(calls[0]?.init?.headers).toEqual({
"Content-Type": "application/json",
"x-agent-internal-token": "internal-secret",
});
expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({
session_id: sessionId,
});
});
it("does not expose backend credentials to the opencode child process", () => {
const context = {
accessToken: "backend-secret",
actorKey: "actor-1",
clientSessionId: "client-session-1",
network: "network-1",
projectId: "project-id-1",
projectKey: "project-key-1",
sessionId,
traceId: "trace-1",
};
expect(context).toHaveProperty("accessToken", "backend-secret");
const payload = serializeRuntimeSessionContext(context);
expect(payload).toEqual({
actor_key: "actor-1",
allow_learning_write: undefined,
client_session_id: "client-session-1",
memory_list_read_scopes: undefined,
project_key: "project-key-1",
session_id: sessionId,
trace_id: "trace-1",
});
expect(payload).not.toHaveProperty("access_token");
expect(payload).not.toHaveProperty("project_id");
expect(payload).not.toHaveProperty("network");
});
});