diff --git a/.opencode/tools/memory_manager.ts b/.opencode/tools/memory_manager.ts index f041d4a..bbceda5 100644 --- a/.opencode/tools/memory_manager.ts +++ b/.opencode/tools/memory_manager.ts @@ -1,11 +1,8 @@ import { tool } from "@opencode-ai/plugin"; -import { MemoryStore } from "../../src/memory/store.js"; -import { readBridgedRuntimeSessionContext } from "../../src/runtime/internalSessionContextBridge.js"; -import { setRuntimeSessionContext } from "../../src/runtime/sessionContext.js"; - -const memoryStore = new MemoryStore(); -const initializePromise = memoryStore.initialize(); +const internalBaseUrl = + process.env.TJWATER_AGENT_INTERNAL_BASE_URL ?? "http://127.0.0.1:8787"; +const internalToken = process.env.TJWATER_AGENT_INTERNAL_TOKEN ?? ""; export default tool({ description: @@ -25,132 +22,31 @@ export default tool({ content: tool.schema .string() .optional() - .describe( - "The durable fact or preference to remember, written as one concise sentence.", - ), + .describe("The durable fact or preference to remember, written as one concise sentence."), target_id: tool.schema .string() .optional() .describe("Stable memory entry id used by replace/remove."), }, async execute(args, context) { - await initializePromise; - const sessionContext = await readBridgedRuntimeSessionContext( - 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 ?? "", - sessionId: sessionContext.clientSessionId, - source: "tool", - traceId: sessionContext.traceId, + const response = await fetch( + `${internalBaseUrl}/internal/tools/memory-manager`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-agent-internal-token": internalToken, }, - ); - return JSON.stringify({ - ok: true, - kind: "memory", - decision: result.changed ? "accepted" : "rejected", - detail: result.detail, - target: scope, - }); - } - - const result = await memoryStore.remove( - scope, - scopeKey, - args.target_id ?? "", + body: JSON.stringify({ + ...args, + session_id: context.sessionID, + }), + }, ); - return JSON.stringify({ - ok: true, - kind: "memory", - decision: result.changed ? "accepted" : "rejected", - detail: result.detail, - target: scope, - }); + const text = await response.text(); + if (!response.ok) { + throw new Error(text); + } + return text; }, }); diff --git a/.opencode/tools/skill_manager.ts b/.opencode/tools/skill_manager.ts index c36791e..bb21b2b 100644 --- a/.opencode/tools/skill_manager.ts +++ b/.opencode/tools/skill_manager.ts @@ -1,160 +1,67 @@ import { tool } from "@opencode-ai/plugin"; -import { SkillStore } from "../../src/skills/store.js"; -import { - readBridgedRuntimeSessionContext, -} from "../../src/runtime/internalSessionContextBridge.js"; -import { type RuntimeSessionContext } from "../../src/runtime/sessionContext.js"; +const internalBaseUrl = + process.env.TJWATER_AGENT_INTERNAL_BASE_URL ?? "http://127.0.0.1:8787"; +const internalToken = process.env.TJWATER_AGENT_INTERNAL_TOKEN ?? ""; -type ToolContextReader = { - read( - sessionId: string, - ): RuntimeSessionContext | null | Promise; -}; - -const runtimeContextReader: ToolContextReader = { - read: readBridgedRuntimeSessionContext, -}; - -export const createSkillManagerTool = ( - skillStore = new SkillStore(), - toolContextStore: ToolContextReader = runtimeContextReader, - initializePromise: Promise = Promise.resolve(), -) => - tool({ - description: - "维护已验证、可复用、非敏感的 workflow 或方法模式。支持 list、write_skill、remove_skill、append_pattern、remove_pattern、write_reference、remove_reference、write_script、remove_script。", - args: { - action: tool.schema - .enum([ - "list", - "write_skill", - "remove_skill", - "append_pattern", - "remove_pattern", - "write_reference", - "remove_reference", - "write_script", - "remove_script", - ]) - .describe("Skill maintenance operation."), - reason: tool.schema - .string() - .describe( - "Why this skill maintenance action is justified for future reuse.", - ), - skill_path: tool.schema - .string() - .describe( - "Target skill directory path relative to .opencode/skills. Use 'workflow' for the workflow index, or '__root__' for the root skills index.", - ), - pattern: tool.schema - .string() - .optional() - .describe("Pattern text used by append_pattern."), - target_id: tool.schema - .string() - .optional() - .describe("Stable learned pattern id used by remove_pattern."), - file_path: tool.schema - .string() - .optional() - .describe( - "Asset file path. For references use references/*.md; for scripts use scripts/*.py.", - ), - content: tool.schema - .string() - .optional() - .describe( - "Content used by write_skill, write_reference, or write_script.", - ), - }, - async execute(args, context) { - await initializePromise; - const sessionContext = await toolContextStore.read(context.sessionID); - if (!sessionContext) { - throw new Error(`session context not found for ${context.sessionID}`); - } - if ( - sessionContext.allowLearningWrite === false && - args.action !== "list" - ) { - return JSON.stringify({ - ok: true, - 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 ?? "", - ); - - return JSON.stringify({ - ok: true, - kind: "skill", - decision: result.changed ? "accepted" : "rejected", - detail: result.detail, - target: result.target, - }); - }, - }); - -export default createSkillManagerTool(); +export default tool({ + description: + "维护已验证、可复用、非敏感的 workflow 或方法模式。支持 list、write_skill、remove_skill、append_pattern、remove_pattern、write_reference、remove_reference、write_script、remove_script。", + args: { + action: tool.schema + .enum([ + "list", + "write_skill", + "remove_skill", + "append_pattern", + "remove_pattern", + "write_reference", + "remove_reference", + "write_script", + "remove_script", + ]) + .describe("Skill maintenance operation."), + reason: tool.schema + .string() + .describe("Why this skill maintenance action is justified for future reuse."), + skill_path: tool.schema + .string() + .describe( + "Target skill directory path relative to .opencode/skills. Use 'workflow' for the workflow index, or '__root__' for the root skills index.", + ), + pattern: tool.schema.string().optional().describe("Pattern text used by append_pattern."), + target_id: tool.schema + .string() + .optional() + .describe("Stable learned pattern id used by remove_pattern."), + file_path: tool.schema + .string() + .optional() + .describe("Asset file path. For references use references/*.md; for scripts use scripts/*.py."), + content: tool.schema + .string() + .optional() + .describe("Content used by write_skill, write_reference, or write_script."), + }, + async execute(args, context) { + const response = await fetch( + `${internalBaseUrl}/internal/tools/skill-manager`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-agent-internal-token": internalToken, + }, + body: JSON.stringify({ + ...args, + session_id: context.sessionID, + }), + }, + ); + const text = await response.text(); + if (!response.ok) { + throw new Error(text); + } + return text; + }, +}); diff --git a/README.md b/README.md index 3a264b8..453e6aa 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ ## 主要能力 - 提供 `POST /api/v1/agent/sessions/{session_id}/runs` SSE 聊天接口。 -- 支持 embedded OpenCode 运行时,也可连接外部 OpenCode server。 +- 以内嵌模式启动并预热 OpenCode 运行时。 - 管理前端 `session_id` 与 OpenCode session 的映射。 - 在服务端保存当前会话的用户 token、项目、network 和 trace 上下文。 - 通过 `.opencode/tools` 和 MCP 工具驱动地图定位、图表、SCADA、历史数据和业务 API 调用。 @@ -67,13 +67,13 @@ OPENCODE_MODE=embedded 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 -TJWATER_API_BASE_URL=http://127.0.0.1:8000 -``` +## 认证续期与学习工具 + +后端工具调用遇到即将过期的 access token 或首次 `401` 时,Agent 会通过当前 SSE 流发送 `credential_refresh_required`。前端使用服务端保存的 Keycloak refresh token 强制换取新 access token,再调用 `POST /api/v1/agent/sessions/{session_id}/credential-refreshes` 唤醒原工具调用。等待上限为 30 秒,同一会话的并发请求合并为一次续期,原调用最多重试一次;`403` 不触发续期。 + +`memory_manager` 和 `skill_manager` 在 OpenCode 侧只保留内部 HTTP 桥,读取会话上下文和持久化数据的逻辑统一在 Agent 主进程中执行。长期记忆、自动学习和显式工具写入因此共享同一组 `MemoryStore`、`SkillStore` 和运行时会话上下文。 本地可使用 `.local.env` 保存开发配置;系统环境变量优先级更高。 diff --git a/contracts/agent-v1.openapi.json b/contracts/agent-v1.openapi.json index 0d76880..f4d1ffa 100644 --- a/contracts/agent-v1.openapi.json +++ b/contracts/agent-v1.openapi.json @@ -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": { "post": { "operationId": "post_sessions_session_id_permission_responses", diff --git a/contracts/manifest.json b/contracts/manifest.json index 87fc2f0..10bde5e 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "agent": { "file": "agent-v1.openapi.json", - "sha256": "7699d0b59d2710f5179c3880fa9f7de90dee09239718c86ed9ff2ce12e6f4259" + "sha256": "d559c6e76c33e7a7451743f60d85da0630d0df14fb5215228acefcf2eaea555a" } } } diff --git a/src/auth/credentialRefresh.ts b/src/auth/credentialRefresh.ts new file mode 100644 index 0000000..abe089f --- /dev/null +++ b/src/auth/credentialRefresh.ts @@ -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; + reason: CredentialRefreshReason; + reject: (error: Error) => void; + requestId: string; + resolve: (context: RuntimeSessionContext) => void; + timer: ReturnType; +}; + +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 + >(); + private readonly pending = new Map(); + + constructor(private readonly timeoutMs = 30_000) {} + + subscribe(sessionId: string, listener: CredentialRefreshListener) { + const listeners = + this.listeners.get(sessionId) ?? new Set(); + 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((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 | 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 ( + coordinator: CredentialRefreshCoordinator, + context: RuntimeSessionContext, + execute: (context: RuntimeSessionContext) => Promise, +) => { + 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; +}; diff --git a/src/config.ts b/src/config.ts index b7d8456..182538c 100644 --- a/src/config.ts +++ b/src/config.ts @@ -41,8 +41,8 @@ const envSchema = z AGENT_INTERNAL_TOKEN: optionalString(), // Agent 前置认证调用后端 /api/v1/agent/auth/context 的超时时间(毫秒)。 AGENT_AUTH_TIMEOUT_MS: z.coerce.number().int().positive().default(5000), - // opencode 运行模式:embedded 会启动本地 CLI 子进程;client 只连接现有 server。 - OPENCODE_MODE: z.enum(["embedded", "client"]).default("embedded"), + // 当前仅支持 embedded;保留字段用于让旧 client 配置在启动时明确失败。 + OPENCODE_MODE: z.literal("embedded").default("embedded"), // embedded opencode server 的监听地址。 OPENCODE_HOSTNAME: z.string().default("127.0.0.1"), // embedded opencode server 的监听端口。 @@ -55,10 +55,6 @@ const envSchema = z OPENCODE_MODEL_OPTIONS: z.string().default(defaultAgentModelOptionsJson), // opencode skills 树目录;会在运行时解析为绝对路径,避免工具 cwd 偏移。 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_PATH: z.string().default("./cli/tjwater-cli"), // TJWater 后端 API 的基础地址。 @@ -117,13 +113,6 @@ const envSchema = z .default(3600000), }) .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; try { modelOptions = parseAgentModelOptions(env.OPENCODE_MODEL_OPTIONS); @@ -154,15 +143,4 @@ const envSchema = z export type AppConfig = z.infer; -const normalizedEnv = { - ...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); +export const config: AppConfig = envSchema.parse(process.env); diff --git a/src/contracts/openapi.ts b/src/contracts/openapi.ts index 0cde29a..f0b2aad 100644 --- a/src/contracts/openapi.ts +++ b/src/contracts/openapi.ts @@ -246,6 +246,24 @@ register("/api/v1/agent/sessions/{session_id}/runs/current", "delete", { request: { params: SessionId }, 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( "/api/v1/agent/sessions/{session_id}/permission-responses", "post", diff --git a/src/learning/orchestrator.ts b/src/learning/orchestrator.ts index 34771d0..04ac637 100644 --- a/src/learning/orchestrator.ts +++ b/src/learning/orchestrator.ts @@ -77,12 +77,12 @@ type TurnReviewInput = { export class LearningOrchestrator { private readonly activeReviews = new Set(); private readonly sessionLearningStateStore = new SessionLearningStateStore(); - private readonly skillStore = new SkillStore(); constructor( private readonly runtime: OpencodeRuntimeAdapter, private readonly memoryStore: MemoryStore, private readonly transcriptStore: SessionTranscriptStore, + private readonly skillStore: SkillStore, ) {} async initialize() { diff --git a/src/learning/toolManagers.ts b/src/learning/toolManagers.ts new file mode 100644 index 0000000..389a6a0 --- /dev/null +++ b/src/learning/toolManagers.ts @@ -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, +}); diff --git a/src/routes/chat.ts b/src/routes/chat.ts index 526d696..8b03a78 100644 --- a/src/routes/chat.ts +++ b/src/routes/chat.ts @@ -2,6 +2,7 @@ import { Router } from "express"; import { z } from "zod"; import { getAgentAuthContext } from "../auth/agentAuth.js"; +import { type CredentialRefreshCoordinator } from "../auth/credentialRefresh.js"; import { agentModelOptions, isSupportedModel, @@ -121,6 +122,7 @@ export const buildChatRouter = ( sessionTranscriptStore: SessionTranscriptStore, learningOrchestrator: LearningOrchestrator, resultReferenceResolver: ResultReferenceResolver, + credentialRefreshCoordinator: CredentialRefreshCoordinator, ) => { const chatRouter = Router(); @@ -295,6 +297,16 @@ export const buildChatRouter = ( }, }; 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 = () => { run.subscribers.delete(subscriber); @@ -390,6 +402,7 @@ export const buildChatRouter = ( registerChatInteractionRoutes(chatRouter, { activeRuns, + credentialRefreshCoordinator, runtime, sessionMetadataStore, sessionUiStateStore, @@ -803,6 +816,35 @@ export const buildChatRouter = ( 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 { const preparedMessage = await buildPromptWithLearningContext( @@ -925,6 +967,12 @@ export const buildChatRouter = ( logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); }); sessionBridge.finalizeRequest(clientSessionId); + abortController.signal.removeEventListener( + "abort", + cancelCredentialRefreshOnAbort, + ); + credentialRefreshCoordinator.cancelSession(binding.sessionId); + unsubscribeCredentialRefresh(); activeRun.status = abortController.signal.aborted ? activeRun.status === "aborted" ? "aborted" diff --git a/src/routes/chatInteractionRoutes.ts b/src/routes/chatInteractionRoutes.ts index 079c49a..53ce667 100644 --- a/src/routes/chatInteractionRoutes.ts +++ b/src/routes/chatInteractionRoutes.ts @@ -2,8 +2,13 @@ import { type Router } from "express"; import { z } from "zod"; import { getAgentAuthContext } from "../auth/agentAuth.js"; +import { type CredentialRefreshCoordinator } from "../auth/credentialRefresh.js"; import { logger } from "../logger.js"; import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; +import { + getRuntimeSessionContext, + setRuntimeSessionContext, +} from "../runtime/sessionContext.js"; import { type SessionMetadataStore } from "../sessions/metadataStore.js"; import { type SessionUiStateStore } from "../sessions/uiStateStore.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([]), }); +const credentialRefreshPayloadSchema = z.object({ + request_id: z.string().min(1).max(128), +}); + type RegisterInteractionRoutesOptions = { activeRuns: Map; + credentialRefreshCoordinator: CredentialRefreshCoordinator; runtime: OpencodeRuntimeAdapter; sessionMetadataStore: SessionMetadataStore; sessionUiStateStore: SessionUiStateStore; @@ -41,11 +51,73 @@ export const registerChatInteractionRoutes = ( chatRouter: Router, { activeRuns, + credentialRefreshCoordinator, runtime, sessionMetadataStore, sessionUiStateStore, }: 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) => { const parsed = permissionReplyPayloadSchema.safeParse(req.body); if (!parsed.success) { diff --git a/src/runtime/internalSessionContextBridge.ts b/src/runtime/internalSessionContextBridge.ts deleted file mode 100644 index c83afee..0000000 --- a/src/runtime/internalSessionContextBridge.ts +++ /dev/null @@ -1,116 +0,0 @@ -import { - getRuntimeSessionContext, - setRuntimeSessionContext, - type RuntimeSessionContext, -} from "./sessionContext.js"; - -type FetchLike = ( - input: string | URL | Request, - init?: RequestInit, -) => Promise; - -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>; - 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; - 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; -}; diff --git a/src/runtime/opencode.ts b/src/runtime/opencode.ts index c2db38e..bbd7885 100644 --- a/src/runtime/opencode.ts +++ b/src/runtime/opencode.ts @@ -1,6 +1,5 @@ import { createOpencode, - createOpencodeClient, type OpencodeClient, } from "@opencode-ai/sdk/v2"; import { existsSync, readFileSync } from "node:fs"; @@ -387,19 +386,6 @@ export class OpencodeRuntimeAdapter { } private async bootstrapClient(): Promise { - 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 进程环境里, // 这样 .opencode/tools 下的自定义工具可以回调本服务。 process.env.TJWATER_AGENT_INTERNAL_BASE_URL = `http://127.0.0.1:${config.PORT}`; @@ -430,7 +416,7 @@ export class OpencodeRuntimeAdapter { } catch (error) { if (isMissingOpencodeCli(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; diff --git a/src/server.ts b/src/server.ts index 8178090..d79980a 100644 --- a/src/server.ts +++ b/src/server.ts @@ -5,6 +5,11 @@ import express from "express"; import { requireAgentAuth } from "./auth/agentAuth.js"; import { buildBackendContextHeaders } from "./auth/backendContextHeaders.js"; +import { + CredentialRefreshError, + CredentialRefreshCoordinator, + runWithCredentialRefresh, +} from "./auth/credentialRefresh.js"; import { SessionTranscriptStore } from "./sessions/transcriptStore.js"; import { ChatSessionBridge } from "./chat/sessionBridge.js"; import { config } from "./config.js"; @@ -12,6 +17,12 @@ import { SessionUiStateStore } from "./sessions/uiStateStore.js"; import { SessionMetadataStore } from "./sessions/metadataStore.js"; import { logger } from "./logger.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 { ResultReferenceResolver } from "./results/resolver.js"; import { @@ -21,12 +32,12 @@ import { import { buildChatRouter } from "./routes/chat.js"; import { buildAgentPublicRouter } from "./routes/publicApi.js"; import { opencodeRuntime } from "./runtime/opencode.js"; -import { serializeRuntimeSessionContext } from "./runtime/internalSessionContextBridge.js"; import { getRuntimeSessionContext, markRuntimeSessionAuthExpired, type RuntimeSessionContext, } from "./runtime/sessionContext.js"; +import { SkillStore } from "./skills/store.js"; const app = express(); @@ -35,15 +46,18 @@ const sessionBridge = new ChatSessionBridge(opencodeRuntime); const sessionMetadataStore = new SessionMetadataStore(); const sessionUiStateStore = new SessionUiStateStore(); const memoryStore = new MemoryStore(); +const skillStore = new SkillStore(); const sessionTranscriptStore = new SessionTranscriptStore(); const learningOrchestrator = new LearningOrchestrator( opencodeRuntime, memoryStore, sessionTranscriptStore, + skillStore, ); const resultReferenceStore = new ResultReferenceStore(); const resultReferenceResolver = new ResultReferenceResolver(resultReferenceStore); const internalToken = config.AGENT_INTERNAL_TOKEN ?? randomUUID(); +const credentialRefreshCoordinator = new CredentialRefreshCoordinator(); // 这个 token 只用于 OpenCode 子进程回调本服务的内部工具桥。 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) { res.status(403).json({ message: "forbidden" }); return; @@ -90,8 +104,86 @@ app.post("/internal/tools/session-context", (req, res) => { }); 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) => { @@ -110,15 +202,6 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => { }); 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() : ""; if (!command) { res.status(400).json({ message: "command is required" }); @@ -136,46 +219,35 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => { return; } - const authJson = JSON.stringify({ - server: config.TJWATER_API_BASE_URL, - access_token: context.accessToken, - project_id: context.projectId, - }); - - const cliArgs = ["--auth-stdin", ...command.split(/\s+/).filter(Boolean)]; - - 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((resolve, reject) => { - const timer = setTimeout(() => { - child.kill("SIGTERM"); - resolve(-1); - }, timeoutSec * 1000); - child.on("close", (code) => { - clearTimeout(timer); - resolve(code); + let result; + try { + result = await runWithCredentialRefresh( + 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), }); - child.on("error", (err) => { - clearTimeout(timer); - reject(err); - }); - }); + return; + } - if (exitCode === -1) { + if (result.status === 504) { res.status(504).json({ ok: false, schema_version: "tjwater-cli/v1", @@ -189,29 +261,102 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => { return; } - if (exitCode !== 0) { - res.status(502).json({ - ok: false, - exit_code: exitCode, - stderr: stderr.slice(0, 2000), - stdout: stdout.slice(0, 2000), - message: `CLI exited with code ${exitCode}`, - }); + if (result.status === 401) { + 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, + exit_code: result.exitCode, + stderr: result.stderr.slice(0, 2000), + message: `CLI exited with code ${result.exitCode}`, + }), + ); return; } try { - res.json(JSON.parse(stdout)); + res.json(JSON.parse(result.stdout)); } catch { res.json({ ok: true, schema_version: "tjwater-cli/v1", - raw: stdout, - stderr: stderr || undefined, + raw: result.stdout, + 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((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) => { if (req.header("x-agent-internal-token") !== internalToken) { res.status(403).json({ message: "forbidden" }); @@ -302,39 +447,60 @@ const callBackendJson = async ( context: RuntimeSessionContext, 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"); return { ok: false, status: 401, 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) => @@ -357,19 +523,6 @@ const normalizeWebSearchFreshness = (value: unknown) => { 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( context: RuntimeSessionContext, reason: NonNullable["reason"], @@ -491,6 +644,7 @@ const chatRouter = buildChatRouter( sessionTranscriptStore, learningOrchestrator, resultReferenceResolver, + credentialRefreshCoordinator, ); const authenticatedChatRouter = express.Router(); authenticatedChatRouter.use(requireAgentAuth, chatRouter); diff --git a/tests/auth/credentialRefresh.test.ts b/tests/auth/credentialRefresh.test.ts new file mode 100644 index 0000000..fe2c441 --- /dev/null +++ b/tests/auth/credentialRefresh.test.ts @@ -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 = {}) => ({ + 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" }); + }); +}); diff --git a/tests/contracts/openapi.test.ts b/tests/contracts/openapi.test.ts index 5a828d6..bc031e5 100644 --- a/tests/contracts/openapi.test.ts +++ b/tests/contracts/openapi.test.ts @@ -47,7 +47,7 @@ describe("Agent REST OpenAPI", () => { } } - expect(operationCount).toBe(13); + expect(operationCount).toBe(14); }); test("models runs as session subresources", () => { diff --git a/tests/learning/toolManagers.test.ts b/tests/learning/toolManagers.test.ts new file mode 100644 index 0000000..aace375 --- /dev/null +++ b/tests/learning/toolManagers.test.ts @@ -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(); + }); +}); diff --git a/tests/opencode/skillManagerTool.test.ts b/tests/opencode/skillManagerTool.test.ts deleted file mode 100644 index a453f5d..0000000 --- a/tests/opencode/skillManagerTool.test.ts +++ /dev/null @@ -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", - ); - }); -}); diff --git a/tests/routes/publicApi.test.ts b/tests/routes/publicApi.test.ts index a55da7f..abf9ed6 100644 --- a/tests/routes/publicApi.test.ts +++ b/tests/routes/publicApi.test.ts @@ -55,6 +55,7 @@ describe("Agent public REST router", () => { undefined as never, undefined as never, undefined as never, + undefined as never, ); const layers = (router as unknown as { stack: RouterLayer[] }).stack; const runtimeOperations = layers diff --git a/tests/runtime/internalSessionContextBridge.test.ts b/tests/runtime/internalSessionContextBridge.test.ts deleted file mode 100644 index 9acbb96..0000000 --- a/tests/runtime/internalSessionContextBridge.test.ts +++ /dev/null @@ -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"); - }); -});