From fd3ffe5a8d3e5a630ca3af47eab3877803c00784 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 02:23:22 +0800 Subject: [PATCH] feat(server): pi module with upstream CRUD, server-side OAuth and token refresh - /pi/providers, /pi/upstreams CRUD, test (auth-only by default, no cost), models - subscription login via pi models.login with a suspended interaction: browser mode waits for the pasted redirect URL, device mode polls server-side; per-provider login lock, readable callback-port-in-use errors - credential store locks the pi_upstream row (select for update) so api and worker refreshes never double-write a rotated refresh token - refreshPiTokens task: refresh before expiry; failure marks expired and opens a relogin decision, closed again after a successful login - runAgent picks upstreamId > agent_run.upstream_id > default upstream > legacy settings; credentials injected by the server, never logged Co-Authored-By: Claude Opus 5.5 --- apps/server/src/modules/agent/runner.ts | 46 ++- apps/server/src/modules/agent/tasks.ts | 11 +- apps/server/src/modules/index.ts | 4 + apps/server/src/modules/pi/index.ts | 94 ++++++ apps/server/src/modules/pi/oauth.ts | 273 +++++++++++++++++ apps/server/src/modules/pi/refresh.ts | 109 +++++++ apps/server/src/modules/pi/routes.ts | 163 ++++++++++ apps/server/src/modules/pi/service.ts | 222 ++++++++++++++ apps/server/src/modules/pi/store.ts | 219 ++++++++++++++ apps/server/test/api-registry.test.ts | 28 +- apps/server/test/modules.test.ts | 3 +- apps/server/test/pi/migration.test.ts | 84 +++++ apps/server/test/pi/upstreams.test.ts | 387 ++++++++++++++++++++++++ 13 files changed, 1617 insertions(+), 26 deletions(-) create mode 100644 apps/server/src/modules/pi/index.ts create mode 100644 apps/server/src/modules/pi/oauth.ts create mode 100644 apps/server/src/modules/pi/refresh.ts create mode 100644 apps/server/src/modules/pi/routes.ts create mode 100644 apps/server/src/modules/pi/service.ts create mode 100644 apps/server/src/modules/pi/store.ts create mode 100644 apps/server/test/pi/migration.test.ts create mode 100644 apps/server/test/pi/upstreams.test.ts diff --git a/apps/server/src/modules/agent/runner.ts b/apps/server/src/modules/agent/runner.ts index 82864773..4e913918 100644 --- a/apps/server/src/modules/agent/runner.ts +++ b/apps/server/src/modules/agent/runner.ts @@ -23,6 +23,7 @@ import { KnownCapabilityId, } from "@umt/contracts"; import { type AgentRunRow, type DownloaderInstanceRow, sql } from "@umt/db"; +import { type RunUpstream, resolveRunUpstream } from "../pi/index.ts"; import type { ServerContext } from "../types.ts"; import { DEMO_SCRIPTS, parseFakeScripts } from "./fake-scripts.ts"; import { buildGoalReport } from "./report.ts"; @@ -113,28 +114,33 @@ export interface DriveResult { finalText: string; } -/** 设置 → @umt/agent 的运行参数(apiKey 现场经 secrets 解密,不落日志) */ +/** + * 设置 → @umt/agent 的运行参数。按上游运行时(W7-A)凭据由上游的 CredentialStore 在请求时解密注入,这里不取明文; + * 没有任何上游时回退 v0.1 旧字段(pi.provider / pi.apiKeySecretRef,apiKey 现场经 secrets 解密,不落日志)。 + */ async function agentSettings( env: AgentEnv, s: GlobalSettings, budget: AgentBudget, fake: boolean, + upstream: RunUpstream | null, ): Promise { const pi = s.pi; - let apiKey = "fake-key-not-used"; - if (!fake) { - if (!pi.apiKeySecretRef) throw new Error("未配置 Agent 上游密钥(设置 → pi.apiKey)"); + let apiKey: string | undefined; + if (!fake && !upstream) { + if (!pi.apiKeySecretRef) throw new Error("未配置 Pi 上游(设置 → 模型上游)"); apiKey = await env.ctx.secrets.resolve(pi.apiKeySecretRef); if (!pi.model) throw new Error("未配置模型(设置 → pi.model)"); } const maxTokens = budget.maxTokens ?? s.agent.maxTokensPerRun ?? undefined; const maxCost = budget.maxCostUsd ?? s.agent.maxCostUsdPerRun ?? undefined; + const up = upstream?.row; return { - provider: pi.provider === "openai-compatible" ? "openai-compatible" : "anthropic", - ...(pi.baseUrl && !fake ? { baseUrl: pi.baseUrl } : {}), - apiKey, - model: pi.model ?? "fake-model", - thinkingLevel: pi.thinkingLevel, + provider: up ? up.provider : pi.provider === "openai-compatible" ? "openai-compatible" : "anthropic", + ...(!up && pi.baseUrl && !fake ? { baseUrl: pi.baseUrl } : {}), + ...(apiKey ? { apiKey } : fake && !up ? { apiKey: "fake-key-not-used" } : {}), + model: (up ? up.default_model : pi.model) ?? (fake ? "fake-model" : "default"), + thinkingLevel: up ? up.thinking_level : pi.thinkingLevel, contextBudgetTokens: pi.contextBudget, compaction: { mode: pi.compaction.enabled ? "threshold" : "manual", @@ -311,11 +317,15 @@ export async function drive( const blocked = await dailyBudgetBlock(env, s, run); if (blocked) return { status: "failed", reason: "budget", budget: "daily", message: blocked, finalText: "" }; - const fake = s.pi.provider === "fake"; const scope = (run.scope ?? {}) as unknown as RunScope; + // 上游:run.upstream_id(runAgent 的 upstreamId 覆盖)> settings.pi.defaultUpstreamId > 旧字段兼容 + let upstream: RunUpstream | null; + let fake: boolean; let settings: AgentSettings; try { - settings = await agentSettings(env, s, (run.budget ?? {}) as AgentBudget, fake); + upstream = await resolveRunUpstream(ctx, run.upstream_id); + fake = upstream ? upstream.row.provider === "fake" : s.pi.provider === "fake"; + settings = await agentSettings(env, s, (run.budget ?? {}) as AgentBudget, fake, upstream); } catch (err) { return { status: "failed", reason: "error", budget: null, message: (err as Error).message, finalText: "" }; } @@ -324,9 +334,10 @@ export async function drive( .set({ status: "running", started_at: run.started_at ?? new Date(), - provider: fake ? "fake" : s.pi.provider, - model: fake ? "fake-model" : s.pi.model, - thinking_level: s.pi.thinkingLevel, + provider: fake ? "fake" : settings.provider, + model: upstream ? (upstream.row.default_model ?? (fake ? "fake-model" : null)) : fake ? "fake-model" : s.pi.model, + thinking_level: settings.thinkingLevel ?? null, + ...(upstream ? { upstream_id: upstream.row.id } : {}), }) .where("id", "=", runId) .execute(); @@ -351,7 +362,12 @@ export async function drive( } const runtime = fake ? createFakeRuntime({ registry, scripts, onEvent, onEventError }).runtime - : new PiAgentRuntime({ registry, onEvent, onEventError }); + : new PiAgentRuntime({ + registry, + onEvent, + onEventError, + ...(upstream ? { upstream: { spec: upstream.spec, credentials: upstream.credentials } } : {}), + }); const maxSeq = await ctx.db .selectFrom("agent_run_event") .select(sql`max(seq)`.as("m")) diff --git a/apps/server/src/modules/agent/tasks.ts b/apps/server/src/modules/agent/tasks.ts index ebc3a58f..969f22e9 100644 --- a/apps/server/src/modules/agent/tasks.ts +++ b/apps/server/src/modules/agent/tasks.ts @@ -29,12 +29,21 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) defineTask({ name: "runAgent", version: 1, - inputSchema: z.object({ agentRunId: id }), + /** upstreamId:覆盖默认上游(W7-A);缺省用 agent_run.upstream_id,再缺省 settings.pi.defaultUpstreamId */ + inputSchema: z.object({ agentRunId: id, upstreamId: id.optional() }), retryPolicy: DB_RETRY, handler: async (t) => { const e = env(); const runId = t.input.agentRunId; + const upstreamId = t.input.upstreamId; await t.step.run("start", async () => { + if (upstreamId) + await e.ctx.db + .updateTable("agent_run") + .set({ upstream_id: upstreamId }) + .where("id", "=", runId) + .where("started_at", "is", null) + .execute(); await e.ctx.db .updateTable("task") .set({ status: "running", started_at: new Date(), attempts: sql`attempts + 1` }) diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts index bc20367d..56542a07 100644 --- a/apps/server/src/modules/index.ts +++ b/apps/server/src/modules/index.ts @@ -2,6 +2,8 @@ import { acquisitionOrganizeModule } from "./acquisition-organize/index.ts"; import { agentModule } from "./agent/index.ts"; import { authSettingsModule } from "./auth-settings/index.ts"; import { fileManagerModule } from "./file-manager/index.ts"; +import { piModule } from "./pi/index.ts"; +import { schedulesModule } from "./schedules/index.ts"; import { sourcesIdentityModule } from "./sources-identity/index.ts"; import type { ServerModule } from "./types.ts"; @@ -17,4 +19,6 @@ export const SERVER_MODULES: readonly ServerModule[] = [ acquisitionOrganizeModule, agentModule, fileManagerModule, + piModule, + schedulesModule, ]; diff --git a/apps/server/src/modules/pi/index.ts b/apps/server/src/modules/pi/index.ts new file mode 100644 index 00000000..db1f55db --- /dev/null +++ b/apps/server/src/modules/pi/index.ts @@ -0,0 +1,94 @@ +import { randomUUID } from "node:crypto"; +import type { FakeOAuthControl } from "@umt/agent"; +import { defineTask } from "@umt/tasks"; +import { z } from "zod"; +import type { ServerContext, ServerModule } from "../types.ts"; +import { type OAuthOptions, OAuthSessions } from "./oauth.ts"; +import { refreshPiTokens } from "./refresh.ts"; +import { registerPiRoutes } from "./routes.ts"; + +export { type RunUpstream, resolveRunUpstream } from "./service.ts"; + +/** + * W7-A Pi 模型上游:pi_upstream CRUD、能力表、测试、模型表、服务端订阅登录(挂起式 interaction)、 + * 定时任务 refreshPiTokens(OAuth 过期前刷新,行锁内写回)。runAgent 经 ./service.ts 的 resolveRunUpstream 取上游。 + * 用法见 docs/runbook/agent.md「上游与登录」。 + */ +export interface PiModuleOptions { + /** provider = fake 的模拟 OAuth(测试注入) */ + fakeOAuth?: FakeOAuthControl; + oauth?: Omit; + /** 刷新任务入队间隔(默认 10 分钟;<=0 关闭) */ + refreshIntervalMs?: number; + /** 剩余有效期少于它就刷新(默认 30 分钟) */ + refreshWindowMs?: number; +} + +export function createPiModule( + opts: PiModuleOptions = {}, +): ServerModule & { sessions(ctx: ServerContext): OAuthSessions } { + const client = opts.fakeOAuth ? { fakeOAuth: opts.fakeOAuth } : undefined; + const sessions = new WeakMap(); + const sessionsFor = (ctx: ServerContext) => { + let s = sessions.get(ctx); + if (!s) { + s = new OAuthSessions(ctx, { ...opts.oauth, ...(client ? { client } : {}) }); + sessions.set(ctx, s); + } + return s; + }; + let timer: ReturnType | null = null; + const live: OAuthSessions[] = []; + const interval = opts.refreshIntervalMs ?? 10 * 60_000; + const windowMs = opts.refreshWindowMs ?? 30 * 60_000; + + return { + name: "pi", + sessions: sessionsFor, + registerRoutes(app, ctx) { + registerPiRoutes(app, ctx, { + oauth: () => { + const s = sessionsFor(ctx); + if (!live.includes(s)) live.push(s); + return s; + }, + ...(client ? { client } : {}), + }); + }, + registerTasks(registry, ctx) { + registry.register( + defineTask({ + name: "refreshPiTokens", + version: 1, + inputSchema: z.object({}).passthrough(), + retryPolicy: { maximumAttempts: 2, initialInterval: "10s", backoffCoefficient: 2, maximumInterval: "60s" }, + handler: async (t) => + t.step.run("refresh", () => refreshPiTokens(ctx, { windowMs, ...(client ? { client } : {}) })), + }), + ); + }, + start(ctx) { + if (interval <= 0) return; + const enqueue = async () => { + try { + await ctx.engine.enqueue( + "refreshPiTokens", + {}, + { idempotencyKey: `refreshPiTokens:${Math.floor(Date.now() / interval)}`, traceId: randomUUID() }, + ); + } catch (err) { + ctx.logger.warn({ err: { message: (err as Error).message } }, "refreshPiTokens 入队失败"); + } + }; + void enqueue(); + timer = setInterval(() => void enqueue(), interval); + }, + async stop() { + if (timer) clearInterval(timer); + timer = null; + await Promise.all(live.splice(0).map((s) => s.shutdown())); + }, + }; +} + +export const piModule = createPiModule(); diff --git a/apps/server/src/modules/pi/oauth.ts b/apps/server/src/modules/pi/oauth.ts new file mode 100644 index 00000000..3e396fc3 --- /dev/null +++ b/apps/server/src/modules/pi/oauth.ts @@ -0,0 +1,273 @@ +import { randomUUID } from "node:crypto"; +import { + classifyUpstreamError, + OAUTH_CALLBACK_PORTS, + PiUpstreamClient, + providerInfo, + type UpstreamAuthEvent, + type UpstreamAuthPrompt, + type UpstreamClientOptions, + type UpstreamLoginInteraction, +} from "@umt/agent"; +import type { PiOAuthMode, PiOAuthStartResult, PiOAuthState } from "@umt/contracts"; +import type { PiUpstreamRow } from "@umt/db"; +import { ApiException } from "../../http/errors.ts"; +import type { ServerContext } from "../types.ts"; +import { dbCredentialStore, getRow, setStatus, specOf } from "./store.ts"; + +// 服务端订阅登录(研究文档 §3):调用 pi 的 models.login,注入“挂起式” interaction —— +// select 用 start 请求里的 mode 作答、text 用 enterpriseDomain 作答、manual_code 挂起等 /oauth/complete 粘贴; +// notify(auth_url | device_code) 写进会话记录,start 拿到第一个事件就返回。 +// 会话只活在当前 api 进程内存(interaction 是闭包):多实例部署时登录请求要粘到同一实例。 +// 同一 provider 的登录串行(pi 的浏览器流程会在服务端监听固定回调端口,并发会撞端口)。 + +export interface OAuthOptions { + /** 会话有效期(默认 15 分钟) */ + sessionTtlMs?: number; + /** start 等第一个事件的上限(默认 20s) */ + startWaitMs?: number; + /** complete 缺省等待(默认 25s) */ + completeWaitMs?: number; + client?: UpstreamClientOptions; +} + +interface Session { + id: string; + upstreamId: string; + provider: string; + mode: PiOAuthMode; + state: PiOAuthState; + message: string | null; + progress: string[]; + authorizeUrl: string | null; + userCode: string | null; + verificationUrl: string | null; + instructions: string | null; + expiresAt: number; + abort: AbortController; + /** manual_code 挂起的解析函数 */ + paste: ((v: string) => void) | null; + pasteWaiters: (() => void)[]; + firstEvent: PromiseWithResolvers; + done: Promise; + timer: ReturnType; +} + +const MODE_RE: Record = { browser: /browser/i, device: /device/i }; + +/** 登录失败信息转成可读提示(端口占用、取消、超时) */ +export function readableLoginError(provider: string, err: unknown): string { + const raw = err instanceof Error ? err.message : String(err); + if (/EADDRINUSE|address already in use/i.test(raw)) { + const port = OAUTH_CALLBACK_PORTS[provider] ?? raw.match(/:(\d{2,5})\b/)?.[1]; + return `服务端回调端口${port ? ` ${port}` : ""} 被占用(可能另有登录进行中,或容器内有 Claude Code / Codex CLI 占用该端口),请释放端口或稍后重试`; + } + if (/cancel|abort/i.test(raw)) return "登录已取消"; + if (/state mismatch/i.test(raw)) return "粘贴的地址与本次登录不匹配(state 不一致),请用本次打开的授权链接重新登录"; + if (/\b404\b/.test(raw) && provider === "openai-codex") + return "该账号未开启设备码登录(Codex 返回 404),请改用浏览器登录"; + return classifyUpstreamError(err).message; +} + +export class OAuthSessions { + private readonly sessions = new Map(); + constructor( + private readonly ctx: ServerContext, + private readonly opts: OAuthOptions = {}, + ) {} + + private active(provider: string): Session | undefined { + for (const s of this.sessions.values()) if (s.provider === provider && s.state === "pending") return s; + return undefined; + } + + private finish(s: Session, state: PiOAuthState, message: string | null) { + if (s.state !== "pending") return; + s.state = state; + s.message = message; + s.firstEvent.resolve(); + for (const w of s.pasteWaiters.splice(0)) w(); + clearTimeout(s.timer); + // 结束后保留一会儿供 complete 查询结果 + s.timer = setTimeout(() => this.sessions.delete(s.id), 5 * 60_000); + (s.timer as { unref?: () => void }).unref?.(); + } + + cancel(s: Session, reason: PiOAuthState = "cancelled", message = "登录已取消") { + this.finish(s, reason, message); + s.abort.abort(new Error(message)); + } + + cancelUpstream(upstreamId: string) { + for (const s of this.sessions.values()) if (s.upstreamId === upstreamId && s.state === "pending") this.cancel(s); + } + + async start(row: PiUpstreamRow, req: { mode?: PiOAuthMode; enterpriseDomain?: string }): Promise { + const info = providerInfo(row.provider); + if (row.auth_kind !== "oauth" || !info?.oauth) + throw new ApiException( + "precondition_failed", + "该上游不是订阅登录型(authKind 需为 oauth,且 provider 支持订阅登录)", + ); + if (!row.enabled) throw new ApiException("precondition_failed", "上游已停用"); + const mode = req.mode ?? info.oauth.modes[0]!; + if (!info.oauth.modes.includes(mode)) + throw new ApiException( + "validation_failed", + `${info.name} 不支持 ${mode} 登录,可用:${info.oauth.modes.join(" / ")}`, + ); + + const busy = this.active(row.provider); + if (busy) { + if (busy.upstreamId !== row.id) + throw new ApiException( + "conflict", + `${info.name} 已有登录进行中(同一 provider 的登录需串行),请先完成或等待其过期(${new Date(busy.expiresAt).toISOString()})`, + ); + this.cancel(busy, "cancelled", "已被新的登录请求取代"); + await busy.done.catch(() => undefined); + } + + const ttl = this.opts.sessionTtlMs ?? 15 * 60_000; + const s: Session = { + id: randomUUID(), + upstreamId: row.id, + provider: row.provider, + mode, + state: "pending", + message: null, + progress: [], + authorizeUrl: null, + userCode: null, + verificationUrl: null, + instructions: null, + expiresAt: Date.now() + ttl, + abort: new AbortController(), + paste: null, + pasteWaiters: [], + firstEvent: Promise.withResolvers(), + done: Promise.resolve(), + timer: setTimeout(() => this.cancel(s, "expired", "登录会话已超时,请重新开始"), ttl), + }; + (s.timer as { unref?: () => void }).unref?.(); + this.sessions.set(s.id, s); + + const interaction: UpstreamLoginInteraction = { + signal: s.abort.signal, + prompt: (p: UpstreamAuthPrompt) => this.prompt(s, p, req), + notify: (e: UpstreamAuthEvent) => this.notify(s, e), + }; + const client = new PiUpstreamClient(specOf(row), dbCredentialStore(this.ctx, row.id), this.opts.client); + s.done = client.login(interaction).then( + async () => { + await setStatus(this.ctx, row.id, { status: "ok", verified: true }); + this.finish(s, "succeeded", null); + }, + async (err) => { + if (s.state !== "pending") return; // 已取消/超时 + const message = readableLoginError(row.provider, err); + this.ctx.logger.warn({ upstreamId: row.id, provider: row.provider, message }, "订阅登录失败"); + this.finish(s, "failed", message); + }, + ); + + const waitMs = this.opts.startWaitMs ?? 20_000; + await Promise.race([s.firstEvent.promise, new Promise((r) => setTimeout(r, waitMs).unref?.())]); + return this.startView(s); + } + + private async prompt(s: Session, p: UpstreamAuthPrompt, req: { mode?: PiOAuthMode; enterpriseDomain?: string }) { + if (p.type === "select") { + const hit = p.options.find((o) => MODE_RE[s.mode].test(`${o.id} ${o.label}`)); + if (!hit) throw new Error(`provider 不提供 ${s.mode} 登录方式`); + return hit.id; + } + if (p.type === "text") return req.enterpriseDomain ?? ""; + if (p.type === "manual_code") { + return await new Promise((resolve, reject) => { + const onAbort = () => reject(new Error("Login cancelled")); + p.signal?.addEventListener("abort", onAbort, { once: true }); + s.abort.signal.addEventListener("abort", onAbort, { once: true }); + s.paste = (v) => { + s.paste = null; + resolve(v); + }; + for (const w of s.pasteWaiters.splice(0)) w(); + }); + } + throw new Error("该登录流程需要额外输入(secret),服务端登录不支持"); + } + + private notify(s: Session, e: UpstreamAuthEvent) { + if (e.type === "auth_url") { + s.authorizeUrl = e.url; + s.instructions = e.instructions ?? null; + s.mode = "browser"; + s.firstEvent.resolve(); + } else if (e.type === "device_code") { + s.userCode = e.userCode; + s.verificationUrl = e.verificationUri; + s.mode = "device"; + if (e.expiresInSeconds) s.expiresAt = Math.min(s.expiresAt, Date.now() + e.expiresInSeconds * 1000); + s.firstEvent.resolve(); + } else { + s.progress.push(e.message.slice(0, 300)); + if (s.progress.length > 20) s.progress.shift(); + if (e.type === "info" && !s.instructions) s.instructions = e.message.slice(0, 500); + } + } + + private startView(s: Session): PiOAuthStartResult { + return { + sessionId: s.id, + mode: s.mode, + state: s.state, + authorizeUrl: s.authorizeUrl, + userCode: s.userCode, + verificationUrl: s.verificationUrl, + instructions: s.instructions, + expiresAt: new Date(s.expiresAt).toISOString(), + message: s.message, + }; + } + + /** browser:把粘贴内容交给挂起的 manual_code;然后(两种模式)等结果到 waitMs,仍未完成返回 pending */ + async complete( + upstreamId: string, + req: { sessionId: string; pastedUrlOrCode?: string; waitMs?: number }, + ): Promise<{ sessionId: string; state: PiOAuthState; message: string | null; progress: string[] }> { + const s = this.sessions.get(req.sessionId); + if (!s || s.upstreamId !== upstreamId) + return { sessionId: req.sessionId, state: "expired", message: "登录会话不存在或已过期,请重新开始", progress: [] }; + if (req.pastedUrlOrCode && s.state === "pending") { + if (s.mode !== "browser") throw new ApiException("bad_request", "设备码登录不需要粘贴,等待授权完成即可"); + if (!s.paste) { + // manual_code 提示可能还没发出:短暂等待 + await Promise.race([ + new Promise((r) => s.pasteWaiters.push(r)), + new Promise((r) => setTimeout(r, 2000).unref?.()), + ]); + } + if (!s.paste) throw new ApiException("conflict", "当前登录不在等待粘贴的阶段"); + s.paste(req.pastedUrlOrCode); + } + const waitMs = req.waitMs ?? this.opts.completeWaitMs ?? 25_000; + if (s.state === "pending") + await Promise.race([s.done, new Promise((r) => setTimeout(r, waitMs).unref?.())]).catch(() => undefined); + return { sessionId: s.id, state: s.state, message: s.message, progress: [...s.progress] }; + } + + async shutdown() { + for (const s of this.sessions.values()) { + if (s.state === "pending") this.cancel(s, "cancelled", "服务正在停止"); + clearTimeout(s.timer); + } + this.sessions.clear(); + } +} + +export async function requireRow(ctx: ServerContext, id: string): Promise { + const row = await getRow(ctx, id); + if (!row) throw new ApiException("not_found", "上游不存在"); + return row; +} diff --git a/apps/server/src/modules/pi/refresh.ts b/apps/server/src/modules/pi/refresh.ts new file mode 100644 index 00000000..f79d59bd --- /dev/null +++ b/apps/server/src/modules/pi/refresh.ts @@ -0,0 +1,109 @@ +import { PiUpstreamClient, type UpstreamClientOptions } from "@umt/agent"; +import { type PiUpstreamRow, sql } from "@umt/db"; +import type { ServerContext } from "../types.ts"; +import { dbCredentialStore, setStatus, specOf } from "./store.ts"; + +// refreshPiTokens:对即将过期的 OAuth 上游提前刷新。刷新本身走 pi 的 getAuth(minOAuthValidityMs): +// pi 在我们的 CredentialStore.modify(pi_upstream 行锁)里双重检查过期时间,再调 provider 的 refresh 并写回轮换后的 +// token —— Agent 运行时(worker)与测试端点(api)的按需刷新走的是同一条路径,不会双写。 +// 刷新失败(refresh token 失效)→ status = expired + 待处理事项“上游 X 需要重新登录”;旧凭据保留。 + +export interface RefreshOptions { + /** 剩余有效期少于它就刷新(默认 30 分钟) */ + windowMs?: number; + client?: UpstreamClientOptions; +} + +export interface RefreshReport { + checked: number; + refreshed: number; + failed: { upstreamId: string; message: string }[]; +} + +const DECISION_KEY = (id: string) => `pi_upstream_expired:${id}`; + +/** 与其它模块的 upsertDecision 同一写法(跨模块不 import):同键未关闭时只累加 attempts */ +async function upsertExpiredDecision(ctx: ServerContext, row: PiUpstreamRow, message: string) { + const title = `模型上游「${row.name}」登录已失效,需要重新登录`; + await ctx.db.transaction().execute(async (trx) => { + const r = await sql<{ id: string; inserted: boolean }>` + insert into decision (kind, severity, title, summary, subject_type, subject_id, dedupe_key, evidence, options) + values ('other', 'warning', ${title}, ${message}, 'pi_upstream', ${row.id}, ${DECISION_KEY(row.id)}, + ${JSON.stringify({ provider: row.provider, upstreamId: row.id, error: message })}::jsonb, + ${JSON.stringify([{ key: "relogin", label: "去设置页重新登录", description: "设置 → 模型上游 → 登录" }])}::jsonb) + on conflict (dedupe_key) where status in ('open', 'investigating', 'awaiting_user') + do update set attempts = decision.attempts + 1, evidence = excluded.evidence, summary = excluded.summary, + updated_at = now() + returning id, (xmax = 0) as inserted`.execute(trx); + const d = r.rows[0]!; + if (d.inserted) + await ctx.outbox.emit( + trx, + "decision.created", + { decisionId: d.id, kind: "other", severity: "warning", title }, + { idempotencyKey: `decision.created:${d.id}` }, + ); + }); +} + +/** 重新登录成功后关闭“需要重新登录”事项 */ +export async function resolveExpiredDecision(ctx: ServerContext, upstreamId: string) { + await ctx.db + .updateTable("decision") + .set({ + status: "resolved", + resolved_at: new Date(), + resolved_by: "system:pi", + resolution: JSON.stringify({ key: "relogin" }), + }) + .where("dedupe_key", "=", DECISION_KEY(upstreamId)) + .where("status", "in", ["open", "investigating", "awaiting_user"]) + .execute(); +} + +/** 刷新单个上游(api 测试端点与定时任务共用) */ +export async function refreshUpstream( + ctx: ServerContext, + row: PiUpstreamRow, + opts: RefreshOptions = {}, +): Promise<{ ok: true } | { ok: false; message: string; kind: string }> { + const client = new PiUpstreamClient(specOf(row), dbCredentialStore(ctx, row.id), opts.client); + try { + await client.resolveAuth(opts.windowMs ?? 30 * 60_000); + return { ok: true }; + } catch (err) { + const e = err as { kind?: string; message?: string }; + const message = e.message ?? String(err); + if (e.kind === "oauth") { + await setStatus(ctx, row.id, { status: "expired", lastError: message }); + await upsertExpiredDecision(ctx, row, message); + } else if (e.kind === "auth") { + await setStatus(ctx, row.id, { status: "auth_required", lastError: message }); + } + return { ok: false, message, kind: e.kind ?? "other" }; + } +} + +export async function refreshPiTokens(ctx: ServerContext, opts: RefreshOptions = {}): Promise { + const windowMs = opts.windowMs ?? 30 * 60_000; + const rows = await ctx.db + .selectFrom("pi_upstream") + .selectAll() + .where("auth_kind", "=", "oauth") + .where("enabled", "=", true) + .where("secret_ref", "is not", null) + .where("status", "!=", "expired") + .where("oauth_expires_at", "is not", null) + .where("oauth_expires_at", "<", new Date(Date.now() + windowMs)) + .execute(); + const report: RefreshReport = { checked: rows.length, refreshed: 0, failed: [] }; + for (const row of rows) { + const r = await refreshUpstream(ctx, row, { ...opts, windowMs }); + if (r.ok) report.refreshed += 1; + else { + report.failed.push({ upstreamId: row.id, message: r.message }); + ctx.logger.warn({ upstreamId: row.id, provider: row.provider, message: r.message }, "Pi 上游 token 刷新失败"); + } + } + return report; +} diff --git a/apps/server/src/modules/pi/routes.ts b/apps/server/src/modules/pi/routes.ts new file mode 100644 index 00000000..2acf8ad9 --- /dev/null +++ b/apps/server/src/modules/pi/routes.ts @@ -0,0 +1,163 @@ +import { createRoute, type RouteConfig } from "@hono/zod-openapi"; +import { providerCatalog, type UpstreamClientOptions, UpstreamError } from "@umt/agent"; +import { + type Actor, + Api, + apiResponse, + type EndpointDef, + type PiOAuthCompleteRequest, + type PiOAuthStartRequest, + type PiUpstreamCreate, + type PiUpstreamTestRequest, + type PiUpstreamUpdate, +} from "@umt/contracts"; +import type { z } from "zod"; +import { requireAdminSession, requireScope } from "../../http/auth.ts"; +import { ApiException } from "../../http/errors.ts"; +import type { ModuleApp, ServerContext } from "../types.ts"; +import type { OAuthSessions } from "./oauth.ts"; +import { requireRow } from "./oauth.ts"; +import { refreshUpstream, resolveExpiredDecision } from "./refresh.ts"; +import { auditPi, clientFor, createUpstream, deleteUpstream, listUpstreams, updateUpstream } from "./service.ts"; +import { getRow, setStatus, toView } from "./store.ts"; + +type Infer = T extends z.ZodType ? z.infer : Record; +interface Input { + params: Infer; + query: Infer; + body: Infer; + actor: Actor | undefined; +} + +/** 薄封装:合同 EndpointDef → OpenAPI 路由 + requireScope(+ session)+ 出参按合同 parse */ +function route(app: ModuleApp, def: D, handler: (i: Input) => Promise) { + const request: RouteConfig["request"] = {}; + if (def.params) request.params = def.params as never; + if (def.query) request.query = def.query as never; + if (def.body) request.body = { content: { "application/json": { schema: def.body as never } }, required: true }; + const r = createRoute({ + method: def.method, + path: def.path, + summary: def.summary, + tags: ["pi"], + middleware: [requireScope(def.scope), ...(def.auth === "session" ? [requireAdminSession()] : [])] as never, + request, + responses: { + 200: { description: def.summary, content: { "application/json": { schema: apiResponse(def.response) } } }, + }, + }); + // biome-ignore lint/suspicious/noExplicitAny: 路由泛型在封装里擦除,入参已由 zod-openapi 校验 + app.openapi(r, (async (c: any) => { + const input = { + params: def.params ? c.req.valid("param") : {}, + query: def.query ? c.req.valid("query") : {}, + body: def.body ? c.req.valid("json") : {}, + actor: c.get("actor"), + } as Input; + const data = def.response.parse(await handler(input)); + return c.json({ ok: true, data }, 200); + // biome-ignore lint/suspicious/noExplicitAny: 同上 + }) as any); +} + +export function registerPiRoutes( + app: ModuleApp, + ctx: ServerContext, + deps: { oauth: () => OAuthSessions; client?: UpstreamClientOptions }, +) { + route(app, Api.pi.providers, async ({ query }) => providerCatalog({ models: query.models !== "false" })); + + route(app, Api.pi.list, async () => await listUpstreams(ctx)); + + route(app, Api.pi.get, async ({ params }) => await toView(ctx, await requireRow(ctx, params.id))); + + route(app, Api.pi.create, async ({ body, actor }) => await createUpstream(ctx, actor, body as PiUpstreamCreate)); + + route(app, Api.pi.update, async ({ params, body, actor }) => { + const row = await requireRow(ctx, params.id); + return await updateUpstream(ctx, actor, row, body as PiUpstreamUpdate, () => deps.oauth().cancelUpstream(row.id)); + }); + + route(app, Api.pi.remove, async ({ params, actor }) => { + const row = await requireRow(ctx, params.id); + deps.oauth().cancelUpstream(row.id); + await deleteUpstream(ctx, actor, row); + return { ok: true as const }; + }); + + route(app, Api.pi.test, async ({ params, body }) => { + const row = await requireRow(ctx, params.id); + const req = body as PiUpstreamTestRequest; + const mode = req.mode ?? "auth"; + const started = Date.now(); + let out: { + ok: boolean; + source: string | null; + model: string | null; + errorKind: UpstreamError["kind"] | null; + message: string | null; + }; + try { + // OAuth 型先走与定时任务同一条刷新路径(失败会标 expired 并建待处理事项) + if (row.auth_kind === "oauth" && row.secret_ref) { + const r = await refreshUpstream(ctx, row, { windowMs: 0, ...(deps.client ? { client: deps.client } : {}) }); + if (!r.ok) throw new UpstreamError(r.kind as UpstreamError["kind"], r.message); + } + const t = await clientFor(ctx, row, deps.client).test({ mode, ...(req.model ? { model: req.model } : {}) }); + out = { + ok: true, + source: t.source, + model: t.model, + errorKind: null, + message: t.inCatalog ? null : `模型 ${t.model} 不在随包目录中,将按手填 id 调用`, + }; + await setStatus(ctx, row.id, { status: "ok", verified: true }); + } catch (err) { + const e = err instanceof UpstreamError ? err : new UpstreamError("other", (err as Error).message); + out = { ok: false, source: null, model: req.model ?? row.default_model, errorKind: e.kind, message: e.message }; + const fresh = await getRow(ctx, row.id); + if (fresh && fresh.status !== "expired") + await setStatus(ctx, row.id, { + status: e.kind === "auth" ? "auth_required" : "error", + lastError: e.message, + }); + } + return { ...out, mode, latencyMs: Date.now() - started, upstream: await toView(ctx, (await getRow(ctx, row.id))!) }; + }); + + route(app, Api.pi.models, async ({ params }) => { + const row = await requireRow(ctx, params.id); + try { + return await clientFor(ctx, row, deps.client).listModels(); + } catch (err) { + if (err instanceof UpstreamError) throw new ApiException("validation_failed", err.message); + throw err; + } + }); + + route(app, Api.pi.oauthStart, async ({ params, body, actor }) => { + const row = await requireRow(ctx, params.id); + const r = await deps.oauth().start(row, body as PiOAuthStartRequest); + await auditPi(ctx, actor, "pi_upstream.oauth_start", row.id, { mode: r.mode, sessionId: r.sessionId }); + return r; + }); + + route(app, Api.pi.oauthComplete, async ({ params, body, actor }) => { + const row = await requireRow(ctx, params.id); + const req = body as PiOAuthCompleteRequest; + const r = await deps.oauth().complete(row.id, req); + if (r.state === "succeeded") { + await resolveExpiredDecision(ctx, row.id); + await auditPi(ctx, actor, "pi_upstream.oauth_login", row.id, { sessionId: r.sessionId }); + } + return { ...r, upstream: await toView(ctx, (await getRow(ctx, row.id))!) }; + }); + + route(app, Api.pi.oauthLogout, async ({ params, actor }) => { + const row = await requireRow(ctx, params.id); + deps.oauth().cancelUpstream(row.id); + await clientFor(ctx, row, deps.client).logout(); + await auditPi(ctx, actor, "pi_upstream.oauth_logout", row.id); + return await toView(ctx, (await getRow(ctx, row.id))!); + }); +} diff --git a/apps/server/src/modules/pi/service.ts b/apps/server/src/modules/pi/service.ts new file mode 100644 index 00000000..a75b12a9 --- /dev/null +++ b/apps/server/src/modules/pi/service.ts @@ -0,0 +1,222 @@ +import { randomUUID } from "node:crypto"; +import { + type PiUpstreamClient as Client, + PiUpstreamClient, + providerInfo, + type UpstreamClientOptions, + type UpstreamCredentialStore, + type UpstreamSpec, +} from "@umt/agent"; +import type { Actor, PiUpstream, PiUpstreamCreate, PiUpstreamUpdate } from "@umt/contracts"; +import type { PiUpstreamRow } from "@umt/db"; +import { ApiException } from "../../http/errors.ts"; +import type { ServerContext } from "../types.ts"; +import { + dbCredentialStore, + defaultUpstreamId, + getRow, + parseExtra, + setDefaultUpstream, + specOf, + toView, +} from "./store.ts"; + +// 上游 CRUD 与运行时取用。本文件的导出也供 agent 模块(runAgent)使用:按上游 id 取 spec + 凭据存储。 + +export function clientFor(ctx: ServerContext, row: PiUpstreamRow, opts?: UpstreamClientOptions): Client { + return new PiUpstreamClient(specOf(row), dbCredentialStore(ctx, row.id), opts); +} + +/** 审计(action 为自由文本;失败只记日志) */ +export async function auditPi( + ctx: ServerContext, + actor: Actor | undefined, + action: string, + id: string | null, + payload: Record = {}, +) { + try { + await ctx.db + .insertInto("audit_event") + .values({ + actor_type: actor?.type === "api_key" ? "api_key" : actor ? "user" : "system", + actor_id: actor?.id ?? null, + action, + target_type: "pi_upstream", + target_id: id, + payload: JSON.stringify(payload), + }) + .execute(); + } catch (err) { + ctx.logger.warn({ err, action }, "审计写入失败"); + } +} + +function validateShape(input: { + provider: string; + authKind: string; + baseUrl?: string | null | undefined; + defaultModel?: string | null | undefined; +}) { + const info = providerInfo(input.provider); + if (!info) throw new ApiException("validation_failed", `未知 provider:${input.provider}`); + if (!info.authKinds.includes(input.authKind as never)) + throw new ApiException( + "validation_failed", + `${info.name} 不支持认证方式 ${input.authKind},可用:${info.authKinds.join(" / ")}`, + ); + if (input.baseUrl && (!info.baseUrlOverridable || input.authKind === "oauth")) + throw new ApiException( + "validation_failed", + `${info.name}${input.authKind === "oauth" ? "(订阅登录)" : ""} 不能覆盖 baseUrl`, + ); + if (input.provider === "openai-compatible" && (!input.baseUrl || !input.defaultModel)) + throw new ApiException("validation_failed", "自定义兼容端点必须填 baseUrl 与模型 id"); + return info; +} + +const statusFor = (authKind: string, hasKey: boolean) => + authKind === "none" || (authKind === "api_key" && hasKey) ? ("ok" as const) : ("auth_required" as const); + +const isUnique = (err: unknown) => (err as { code?: string })?.code === "23505"; + +export async function listUpstreams( + ctx: ServerContext, +): Promise<{ items: PiUpstream[]; defaultUpstreamId: string | null }> { + const rows = await ctx.db.selectFrom("pi_upstream").selectAll().orderBy("created_at").orderBy("id").execute(); + const def = await defaultUpstreamId(ctx); + return { items: await Promise.all(rows.map((r) => toView(ctx, r, def))), defaultUpstreamId: def }; +} + +export async function createUpstream( + ctx: ServerContext, + actor: Actor | undefined, + body: PiUpstreamCreate, +): Promise { + validateShape(body); + const id = randomUUID(); + let secretRef: string | null = null; + if (body.apiKey) secretRef = await ctx.secrets.create({ purpose: "llm_api_key", plaintext: body.apiKey }); + try { + await ctx.db + .insertInto("pi_upstream") + .values({ + id, + name: body.name, + provider: body.provider, + auth_kind: body.authKind, + secret_ref: secretRef, + base_url: body.baseUrl ?? null, + default_model: body.defaultModel ?? null, + thinking_level: body.thinkingLevel ?? "medium", + extra: JSON.stringify(body.extra ?? {}), + enabled: body.enabled ?? true, + status: statusFor(body.authKind, !!secretRef), + }) + .execute(); + } catch (err) { + if (secretRef) await ctx.secrets.remove(secretRef).catch(() => {}); + if (isUnique(err)) throw new ApiException("conflict", `上游名称「${body.name}」已存在`); + throw err; + } + if (body.makeDefault || !(await defaultUpstreamId(ctx))) await setDefaultUpstream(ctx, id); + await auditPi(ctx, actor, "pi_upstream.create", id, { + provider: body.provider, + authKind: body.authKind, + apiKey: body.apiKey ? "(set)" : undefined, + }); + return await toView(ctx, (await getRow(ctx, id))!); +} + +export async function updateUpstream( + ctx: ServerContext, + actor: Actor | undefined, + row: PiUpstreamRow, + body: PiUpstreamUpdate, + onCredentialDropped: () => void, +): Promise { + const authKind = body.authKind ?? row.auth_kind; + const baseUrl = body.baseUrl !== undefined ? body.baseUrl : authKind === "oauth" ? null : row.base_url; + const defaultModel = body.defaultModel !== undefined ? body.defaultModel : row.default_model; + validateShape({ provider: row.provider, authKind, baseUrl, defaultModel }); + if (body.apiKey && authKind !== "api_key") + throw new ApiException("validation_failed", "只有 api_key 型上游可以填 apiKey"); + + const kindChanged = authKind !== row.auth_kind; + const store = dbCredentialStore(ctx, row.id); + // 认证方式变了或清除 key:旧凭据作废(连同 OAuth 登录会话) + if (kindChanged || body.apiKey === null) { + await store.delete(); + onCredentialDropped(); + } + await ctx.db + .updateTable("pi_upstream") + .set({ + ...(body.name !== undefined ? { name: body.name } : {}), + auth_kind: authKind, + base_url: baseUrl, + default_model: defaultModel, + ...(body.thinkingLevel !== undefined ? { thinking_level: body.thinkingLevel } : {}), + ...(body.extra !== undefined ? { extra: JSON.stringify(body.extra) } : {}), + ...(body.enabled !== undefined ? { enabled: body.enabled } : {}), + }) + .where("id", "=", row.id) + .execute() + .catch((err) => { + if (isUnique(err)) throw new ApiException("conflict", `上游名称「${body.name}」已存在`); + throw err; + }); + if (typeof body.apiKey === "string") { + const key = body.apiKey; + await store.modify(async () => ({ type: "api_key", key })); + } + // 状态按新配置重算(OAuth 已登录的保持) + const fresh = (await getRow(ctx, row.id))!; + if (fresh.auth_kind !== "oauth" && (kindChanged || body.apiKey !== undefined || body.extra !== undefined)) { + await ctx.db + .updateTable("pi_upstream") + .set({ status: statusFor(fresh.auth_kind, !!fresh.secret_ref), last_error: null }) + .where("id", "=", row.id) + .execute(); + } + const def = await defaultUpstreamId(ctx); + if (body.makeDefault === true && def !== row.id) await setDefaultUpstream(ctx, row.id); + if (body.makeDefault === false && def === row.id) await setDefaultUpstream(ctx, null); + await auditPi(ctx, actor, "pi_upstream.update", row.id, { + ...body, + ...(body.apiKey !== undefined ? { apiKey: body.apiKey === null ? "(cleared)" : "(set)" } : {}), + }); + return await toView(ctx, (await getRow(ctx, row.id))!); +} + +export async function deleteUpstream(ctx: ServerContext, actor: Actor | undefined, row: PiUpstreamRow) { + await ctx.db.deleteFrom("pi_upstream").where("id", "=", row.id).execute(); + if (row.secret_ref) await ctx.secrets.remove(row.secret_ref).catch(() => {}); + if ((await defaultUpstreamId(ctx)) === row.id) await setDefaultUpstream(ctx, null); + await auditPi(ctx, actor, "pi_upstream.delete", row.id, { provider: row.provider, name: row.name }); +} + +// ── 运行时(agent 模块 runAgent 使用)── + +export interface RunUpstream { + row: PiUpstreamRow; + spec: UpstreamSpec; + credentials: UpstreamCredentialStore; +} + +/** + * 取运行用的上游:显式 upstreamId 优先,否则 settings.pi.defaultUpstreamId;都没有返回 null(调用方走旧字段兼容)。 + * 上游停用 / 未登录 / 登录失效时抛出可读错误。 + */ +export async function resolveRunUpstream(ctx: ServerContext, upstreamId?: string | null): Promise { + // 直接读库(不走 30s 设置缓存),删除/切换默认上游后立即生效 + const id = upstreamId ?? (await defaultUpstreamId(ctx)); + if (!id) return null; + const row = await getRow(ctx, id); + if (!row) throw new Error(upstreamId ? `指定的上游 ${upstreamId} 不存在` : "默认上游已不存在,请在设置里重新选择"); + if (!row.enabled) throw new Error(`上游「${row.name}」已停用`); + if (row.status === "expired") throw new Error(`上游「${row.name}」登录已失效,请在设置里重新登录`); + if (row.status === "auth_required") + throw new Error(`上游「${row.name}」${row.auth_kind === "oauth" ? "尚未登录" : "未配置 API Key"}`); + return { row, spec: { ...specOf(row), extra: parseExtra(row.extra) }, credentials: dbCredentialStore(ctx, row.id) }; +} diff --git a/apps/server/src/modules/pi/store.ts b/apps/server/src/modules/pi/store.ts new file mode 100644 index 00000000..62dbe060 --- /dev/null +++ b/apps/server/src/modules/pi/store.ts @@ -0,0 +1,219 @@ +import type { UpstreamCredential, UpstreamCredentialStore, UpstreamSpec } from "@umt/agent"; +import { + DEFAULT_SETTINGS, + type PiUpstream, + type PiUpstreamRow as PiUpstreamApiRow, + PiUpstreamExtra, +} from "@umt/contracts"; +import { type PiUpstreamRow, sql } from "@umt/db"; +import type { ServerContext } from "../types.ts"; + +// pi_upstream 的读写与凭据存储。凭据(API Key 明文或 OAuth token 包 JSON)在 secret 表,本表只存 secret_ref。 +// 写凭据一律经 dbCredentialStore.modify:事务内 `select … for update` 锁住上游行 → 读当前凭据 → fn → 写回, +// api 与 worker 两个进程的刷新都走这里,refresh token 轮换不会被并发刷新打断。 + +/** 设置变更事件名(与 auth-settings 的 SETTINGS_CHANGED_EVENT 一致;跨模块不 import) */ +const SETTINGS_CHANGED = "settings.changed"; + +/** 超过这个值视为不过期(OpenRouter 换到的是永久 key,expires = MAX_SAFE_INTEGER) */ +const NEVER_EXPIRES_MS = 8.64e15; + +type Ctx = Pick; + +export async function getRow(ctx: Pick, id: string): Promise { + return await ctx.db.selectFrom("pi_upstream").selectAll().where("id", "=", id).executeTakeFirst(); +} + +export function specOf(row: PiUpstreamRow): UpstreamSpec { + const extra = parseExtra(row.extra); + return { + id: row.id, + provider: row.provider, + authKind: row.auth_kind, + baseUrl: row.base_url, + model: row.default_model, + thinkingLevel: row.thinking_level, + extra, + }; +} + +export function parseExtra(v: unknown): PiUpstreamExtra { + const r = PiUpstreamExtra.safeParse(v ?? {}); + return r.success ? r.data : {}; +} + +function decode(authKind: string, plaintext: string): UpstreamCredential | undefined { + if (authKind === "oauth") { + try { + const v = JSON.parse(plaintext) as UpstreamCredential; + return v && typeof v === "object" && v.type === "oauth" ? v : undefined; + } catch { + return undefined; + } + } + return plaintext ? { type: "api_key", key: plaintext } : undefined; +} + +async function readSecret(ctx: Ctx, row: Pick) { + if (!row.secret_ref) return undefined; + try { + return decode(row.auth_kind, await ctx.secrets.resolve(row.secret_ref)); + } catch (err) { + ctx.logger.warn({ secretRef: row.secret_ref, err: { message: (err as Error).message } }, "上游凭据读取失败"); + return undefined; + } +} + +/** 该上游的凭据存储(读写行锁;明文只在返回值里) */ +export function dbCredentialStore(ctx: Ctx, upstreamId: string): UpstreamCredentialStore { + return { + async read() { + const row = await ctx.db + .selectFrom("pi_upstream") + .select(["auth_kind", "secret_ref"]) + .where("id", "=", upstreamId) + .executeTakeFirst(); + return row ? await readSecret(ctx, row) : undefined; + }, + async modify(fn) { + let created: string | null = null; + try { + return await ctx.db.transaction().execute(async (trx) => { + const row = await trx + .selectFrom("pi_upstream") + .select(["auth_kind", "secret_ref"]) + .where("id", "=", upstreamId) + .forUpdate() + .executeTakeFirst(); + if (!row) throw new Error("上游不存在或已删除"); + const cur = await readSecret(ctx, row); + const next = await fn(cur); + if (next === undefined) return cur; + const plaintext = next.type === "oauth" ? JSON.stringify(next) : (next.key ?? ""); + let ref = row.secret_ref; + if (ref) await ctx.secrets.replace(ref, plaintext); + else { + ref = await ctx.secrets.create({ purpose: "llm_api_key", plaintext }); + created = ref; + } + const expires = + next.type === "oauth" && Number.isFinite(next.expires) && next.expires < NEVER_EXPIRES_MS + ? new Date(next.expires) + : null; + await trx + .updateTable("pi_upstream") + .set({ secret_ref: ref, status: "ok", last_error: null, oauth_expires_at: expires }) + .where("id", "=", upstreamId) + .execute(); + return next; + }); + } catch (err) { + if (created) await ctx.secrets.remove(created).catch(() => {}); + throw err; + } + }, + async delete() { + const ref = await ctx.db.transaction().execute(async (trx) => { + const row = await trx + .selectFrom("pi_upstream") + .select(["secret_ref"]) + .where("id", "=", upstreamId) + .forUpdate() + .executeTakeFirst(); + if (!row) return null; + await trx + .updateTable("pi_upstream") + .set({ secret_ref: null, status: "auth_required", oauth_expires_at: null }) + .where("id", "=", upstreamId) + .execute(); + return row.secret_ref; + }); + if (ref) await ctx.secrets.remove(ref).catch(() => {}); + }, + }; +} + +const mask = (key: string) => (key.length >= 12 ? `${key.slice(0, 4)}…${key.slice(-4)}` : "****"); + +function accountOf(c: UpstreamCredential | undefined): string | null { + if (c?.type !== "oauth") return null; + for (const k of ["email", "accountId", "account", "enterpriseUrl", "login"]) { + const v = c[k]; + if (typeof v === "string" && v) return v.slice(0, 200); + } + return null; +} + +const iso = (d: Date | null) => (d ? d.toISOString() : null); + +export function rowToApi(row: PiUpstreamRow): PiUpstreamApiRow { + return { + id: row.id, + name: row.name, + provider: row.provider as PiUpstreamApiRow["provider"], + authKind: row.auth_kind, + baseUrl: row.base_url, + defaultModel: row.default_model, + thinkingLevel: row.thinking_level, + extra: parseExtra(row.extra), + enabled: row.enabled, + status: row.status, + lastError: row.last_error, + lastVerifiedAt: iso(row.last_verified_at), + oauthExpiresAt: iso(row.oauth_expires_at), + createdAt: row.created_at.toISOString(), + updatedAt: row.updated_at.toISOString(), + }; +} + +/** API 视图:掩码 / 账号信息现场解密得到,明文不出本函数 */ +export async function toView(ctx: Ctx, row: PiUpstreamRow, defaultId?: string | null): Promise { + const def = defaultId === undefined ? await defaultUpstreamId(ctx) : defaultId; + const cred = await readSecret(ctx, row); + return { + ...rowToApi(row), + hasCredential: !!cred, + credentialMask: cred?.type === "api_key" && cred.key ? mask(cred.key) : null, + account: accountOf(cred), + isDefault: def === row.id, + }; +} + +// ── 默认上游(settings.pi.defaultUpstreamId)── + +export async function defaultUpstreamId(ctx: Pick): Promise { + const r = await sql<{ + id: string | null; + }>`select value->>'defaultUpstreamId' as id from settings where key = 'pi'`.execute(ctx.db); + return r.rows[0]?.id ?? null; +} + +/** 改默认上游:只动 settings.pi 的 defaultUpstreamId 一个字段;广播 settings.changed 让各进程缓存失效 */ +export async function setDefaultUpstream(ctx: Ctx, id: string | null): Promise { + const patch = JSON.stringify({ defaultUpstreamId: id }); + await sql`insert into settings (key, value, updated_at) + values ('pi', ${JSON.stringify({ ...DEFAULT_SETTINGS.pi, defaultUpstreamId: id })}::jsonb, now()) + on conflict (key) do update set value = settings.value || ${patch}::jsonb, updated_at = now()`.execute( + ctx.db, + ); + const s = ctx.settings as { invalidate?: () => void }; + s.invalidate?.(); + void ctx.events.publish(SETTINGS_CHANGED, { keys: ["pi"] }); +} + +/** 状态更新(测试/调用结果);不碰凭据 */ +export async function setStatus( + ctx: Pick, + id: string, + patch: { status: PiUpstreamRow["status"]; lastError?: string | null; verified?: boolean }, +): Promise { + await ctx.db + .updateTable("pi_upstream") + .set({ + status: patch.status, + last_error: patch.lastError ?? null, + ...(patch.verified ? { last_verified_at: new Date() } : {}), + }) + .where("id", "=", id) + .execute(); +} diff --git a/apps/server/test/api-registry.test.ts b/apps/server/test/api-registry.test.ts index 539cced9..6fbdbe58 100644 --- a/apps/server/test/api-registry.test.ts +++ b/apps/server/test/api-registry.test.ts @@ -58,15 +58,25 @@ describe("Api 注册表鉴权标注", () => { .filter((e) => e.auth === "session") .map((e) => `${e.method.toUpperCase()} ${e.path}`) .sort(); - expect(session).toEqual([ - "GET /auth/totp", - "GET /files/download", - "PATCH /approvals/{id}", - "POST /approvals", - "POST /auth/password", - "POST /auth/totp", - "POST /files/trash/purge", - ]); + expect(session).toEqual( + [ + "GET /auth/totp", + "GET /files/download", + "PATCH /approvals/{id}", + "POST /approvals", + "POST /auth/password", + "POST /auth/totp", + "POST /files/trash/purge", + // W7-A Pi 上游:写、测试、订阅登录只接受控制台会话 + "POST /pi/upstreams", + "PATCH /pi/upstreams/{id}", + "DELETE /pi/upstreams/{id}", + "POST /pi/upstreams/{id}/test", + "POST /pi/upstreams/{id}/oauth/start", + "POST /pi/upstreams/{id}/oauth/complete", + "POST /pi/upstreams/{id}/oauth/logout", + ].sort(), + ); for (const e of listEndpoints().filter((x) => x.auth === "session")) expect(e.scope).toBe("admin"); }); diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts index ad60a43e..806da2f3 100644 --- a/apps/server/test/modules.test.ts +++ b/apps/server/test/modules.test.ts @@ -43,13 +43,14 @@ const pingModule: ServerModule = { }; describe("ServerModule 骨架", () => { - it("清单里是四个 W2 模块 + 文件管理,名字唯一", () => { + it("清单里是四个 W2 模块 + 文件管理 + Pi 上游,名字唯一", () => { expect(SERVER_MODULES.map((m) => m.name)).toEqual([ "auth-settings", "sources-identity", "acquisition-organize", "agent", "file-manager", + "pi", ]); }); diff --git a/apps/server/test/pi/migration.test.ts b/apps/server/test/pi/migration.test.ts new file mode 100644 index 00000000..b1bfe830 --- /dev/null +++ b/apps/server/test/pi/migration.test.ts @@ -0,0 +1,84 @@ +import { createMigrator, type DB, migrateToLatest, sql } from "@umt/db"; +import { Kysely, PostgresDialect } from "kysely"; +import pg from "pg"; +import { afterAll, describe, expect, it } from "vitest"; +import { createSecretStore } from "../../src/modules/auth-settings/provided.ts"; +import { MASTER_KEY, PG_URL, pgAvailable, withSchemaLock } from "../auth-settings/harness.ts"; + +// 0027:已保存的旧 settings.pi(v0.1 单上游)一次性转成一条上游并设为默认;密钥所有权移交给上游。 + +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("0027 迁移旧 settings.pi → 上游", () => { + const schema = `w7a_mig_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 6)}`; + const admin = async (q: string) => { + const c = new pg.Client({ connectionString: PG_URL }); + await c.connect(); + await c.query(q).finally(() => c.end()); + }; + let db: Kysely; + + afterAll(async () => { + await db?.destroy(); + await withSchemaLock(() => admin(`drop schema if exists ${schema} cascade`)); + }); + + it("anthropic + baseUrl + apiKeySecretRef → 默认上游(api_key, ok);重复迁移不再生成", async () => { + await admin(`create schema ${schema}`); + db = new Kysely({ + dialect: new PostgresDialect({ + pool: new pg.Pool({ connectionString: PG_URL, max: 2, options: `-c search_path=${schema},public` }), + }), + }); + await withSchemaLock(async () => { + const r = await createMigrator(db, { migrationTableSchema: schema }).migrateTo("0026_organization_file_actions"); + if (r.error) throw r.error; + }); + const secrets = createSecretStore({ db, masterKey: MASTER_KEY }); + const ref = await secrets.create({ purpose: "llm_api_key", plaintext: "sk-ant-legacy-key" }); + const legacy = { + provider: "anthropic", + baseUrl: "https://proxy.local/v1", + model: "claude-sonnet-4-5", + thinkingLevel: "high", + contextBudget: 100_000, + compaction: { enabled: true, thresholdRatio: 0.8 }, + subagentsEnabled: false, + apiKeySecretRef: ref, + }; + await sql`insert into settings (key, value) values ('pi', ${JSON.stringify(legacy)}::jsonb)`.execute(db); + + await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema })); + + const rows = await db.selectFrom("pi_upstream").selectAll().execute(); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + provider: "anthropic", + auth_kind: "api_key", + secret_ref: ref, + base_url: "https://proxy.local/v1", + default_model: "claude-sonnet-4-5", + thinking_level: "high", + status: "ok", + enabled: true, + }); + const s = await sql<{ value: Record }>`select value from settings where key = 'pi'`.execute(db); + expect(s.rows[0]!.value).toMatchObject({ + defaultUpstreamId: rows[0]!.id, + apiKeySecretRef: null, + contextBudget: 100_000, + }); + expect(await secrets.resolve(ref)).toBe("sk-ant-legacy-key"); + + // 回滚再迁移:密钥交还旧字段,再次迁移得到一条新上游(不会重复) + await withSchemaLock(async () => { + const r = await createMigrator(db, { migrationTableSchema: schema }).migrateTo("0026_organization_file_actions"); + if (r.error) throw r.error; + }); + const back = await sql<{ value: Record }>`select value from settings where key = 'pi'`.execute(db); + expect(back.rows[0]!.value.apiKeySecretRef).toBe(ref); + expect(back.rows[0]!.value.defaultUpstreamId).toBeUndefined(); + await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema })); + expect(await db.selectFrom("pi_upstream").select("id").execute()).toHaveLength(1); + }, 120_000); +}); diff --git a/apps/server/test/pi/upstreams.test.ts b/apps/server/test/pi/upstreams.test.ts new file mode 100644 index 00000000..a9b69033 --- /dev/null +++ b/apps/server/test/pi/upstreams.test.ts @@ -0,0 +1,387 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { createServer, type Server } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { FakeOAuthControl } from "@umt/agent"; +import { type DB, sql } from "@umt/db"; +import { Kysely, PostgresDialect } from "kysely"; +import pg from "pg"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createAgentModule } from "../../src/modules/agent/index.ts"; +import { drive } from "../../src/modules/agent/runner.ts"; +import { ensureSetupToken } from "../../src/modules/auth-settings/auth-service.ts"; +import { createSecretStore, createSettingsReader } from "../../src/modules/auth-settings/provided.ts"; +import { createPiModule } from "../../src/modules/pi/index.ts"; +import { refreshPiTokens } from "../../src/modules/pi/refresh.ts"; +import { dbCredentialStore } from "../../src/modules/pi/store.ts"; +import type { ServerContext } from "../../src/modules/types.ts"; +import { + Browser, + bearer, + createHarness, + type Harness, + MASTER_KEY, + PG_URL, + pgAvailable, +} from "../auth-settings/harness.ts"; +import { DEV_KEY, json } from "../helpers.ts"; + +// W7-A Pi 上游:端点鉴权、CRUD 与凭据不出 API、OAuth 状态机(fake provider 模拟 browser / device,不联网)、 +// 同 provider 登录串行、端口占用提示、刷新任务的轮换与并发写(两个进程同时刷新只刷新一次)、runAgent 用指定上游。 + +const hasPg = await pgAvailable(); +const PASSWORD = "correct horse battery"; + +describe.skipIf(!hasPg)("Pi 上游(真实 PG)", () => { + let h: Harness; + let admin: Browser; + const control = new FakeOAuthControl(); + const agent = createAgentModule({ + fakeScripts: [{ match: /.*/, steps: [], finalText: "fake 完成" }], + }); + const sessionDir = mkdtempSync(join(tmpdir(), "umt-pi-")); + let mock: Server; + let mockUrl = ""; + const seenAuth: string[] = []; + + beforeAll(async () => { + const pi = createPiModule({ fakeOAuth: control, refreshIntervalMs: 0, oauth: { completeWaitMs: 2000 } }); + h = await createHarness({ extraModules: [pi, agent] }); + agent.registerCapabilities?.(h.ctx.capabilities, h.ctx); + const token = await ensureSetupToken(h.ctx); + admin = new Browser(h.app); + const r = await admin.req("POST", "/auth/setup", { setupToken: token, username: "admin", password: PASSWORD }); + expect(r.status).toBe(200); + // OpenAI 兼容的最小流式端点:记录 Authorization,回一句话 + mock = createServer((req, res) => { + seenAuth.push(String(req.headers.authorization ?? "")); + req.resume(); + res.writeHead(200, { "content-type": "text/event-stream" }); + const chunk = (o: unknown) => res.write(`data: ${JSON.stringify(o)}\n\n`); + chunk({ + id: "c1", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { role: "assistant", content: "上游回复" }, finish_reason: null }], + }); + chunk({ + id: "c1", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + usage: { prompt_tokens: 3, completion_tokens: 2, total_tokens: 5 }, + }); + res.end("data: [DONE]\n\n"); + }); + await new Promise((r) => mock.listen(0, "127.0.0.1", r)); + mockUrl = `http://127.0.0.1:${(mock.address() as { port: number }).port}/v1`; + }, 120_000); + + afterAll(async () => { + await new Promise((r) => mock?.close(r)); + for (const m of [agent]) await m.stop?.(); + await h?.close(); + rmSync(sessionDir, { recursive: true, force: true }); + }); + + const req = async (method: string, path: string, body?: unknown) => { + const res = await admin.req(method, path, body); + return { status: res.status, body: await json(res) }; + }; + const create = async (body: Record) => { + const r = await req("POST", "/pi/upstreams", body); + expect(r.status, JSON.stringify(r.body)).toBe(200); + return r.body.data; + }; + + it("鉴权:未登录 401;Bearer admin 只能读(写/登录要控制台会话);read key 403", async () => { + expect((await h.app.request("/api/v1/pi/providers")).status).toBe(401); + const key = bearer(h.app, DEV_KEY); + expect((await key("GET", "/pi/providers?models=false")).status).toBe(200); + expect((await key("GET", "/pi/upstreams")).status).toBe(200); + expect((await key("POST", "/pi/upstreams", { name: "x", provider: "fake", authKind: "none" })).status).toBe(403); + const readKey = (await json(await admin.req("POST", "/api-keys", { name: "reader", scopes: ["read"] }))).data + .plaintextKey as string; + expect((await bearer(h.app, readKey)("GET", "/pi/upstreams")).status).toBe(403); + const cat = await json(await key("GET", "/pi/providers?models=false")); + expect(cat.data.providers).toHaveLength(43); + const anthropic = cat.data.providers.find((p: { id: string }) => p.id === "anthropic"); + expect(anthropic.termsNote).toMatch(/封禁/); + }); + + it("CRUD:密钥只写、只回掩码;第一条自动成为默认;校验 provider/authKind/baseUrl;重名 409", async () => { + const a = await create({ + name: "OpenAI 主号", + provider: "openai", + authKind: "api_key", + apiKey: "sk-live-AAAA-secret-1234", + }); + expect(a).toMatchObject({ status: "ok", hasCredential: true, credentialMask: "sk-l…1234", isDefault: true }); + expect(JSON.stringify(a)).not.toContain("secret"); + const b = await create({ + name: "OpenAI 副号", + provider: "openai", + authKind: "api_key", + apiKey: "sk-live-BBBB-secret-5678", + baseUrl: "https://proxy.local/v1", + makeDefault: true, + }); + expect(b.isDefault).toBe(true); + const list = (await req("GET", "/pi/upstreams")).body.data; + expect(list.defaultUpstreamId).toBe(b.id); + expect(list.items.filter((u: { isDefault: boolean }) => u.isDefault)).toHaveLength(1); + expect(JSON.stringify(list)).not.toMatch(/secret-\d/); + expect((await h.ctx.settings.get("pi")).defaultUpstreamId).toBe(b.id); + + expect( + (await req("POST", "/pi/upstreams", { name: "c", provider: "openai-codex", authKind: "api_key" })).status, + ).toBe(422); + expect( + ( + await req("POST", "/pi/upstreams", { + name: "c", + provider: "github-copilot", + authKind: "api_key", + baseUrl: "https://x.local", + }) + ).status, + ).toBe(422); + expect( + (await req("POST", "/pi/upstreams", { name: "OpenAI 主号", provider: "openai", authKind: "api_key" })).status, + ).toBe(409); + + // 清 key → auth_required;测试报 auth + const cleared = (await req("PATCH", `/pi/upstreams/${a.id}`, { apiKey: null })).body.data; + expect(cleared).toMatchObject({ status: "auth_required", hasCredential: false }); + const t = (await req("POST", `/pi/upstreams/${a.id}/test`, {})).body.data; + expect(t).toMatchObject({ ok: false, errorKind: "auth" }); + // 重新填 key → 测试(只解析凭据 + 校验模型,不联网)通过 + await req("PATCH", `/pi/upstreams/${a.id}`, { apiKey: "sk-live-CCCC-secret-9999", defaultModel: "gpt-5" }); + const t2 = (await req("POST", `/pi/upstreams/${a.id}/test`, {})).body.data; + expect(t2).toMatchObject({ ok: true, mode: "auth", source: "stored credential", model: "gpt-5" }); + expect(t2.upstream.lastVerifiedAt).toBeTruthy(); + const models = (await req("GET", `/pi/upstreams/${a.id}/models`)).body.data; + expect(models.items.length).toBeGreaterThan(10); + expect(models.refreshed).toBe(false); + + // 删除默认上游 → 默认清空,secret 一并删除 + const secretRef = ( + await h.db.selectFrom("pi_upstream").select("secret_ref").where("id", "=", b.id).executeTakeFirstOrThrow() + ).secret_ref!; + expect((await req("DELETE", `/pi/upstreams/${b.id}`)).status).toBe(200); + expect((await req("GET", "/pi/upstreams")).body.data.defaultUpstreamId).toBeNull(); + expect(await h.db.selectFrom("secret").select("id").where("id", "=", secretRef).executeTakeFirst()).toBeUndefined(); + await req("DELETE", `/pi/upstreams/${a.id}`); + expect(h.lines.join("\n")).not.toMatch(/sk-live-\w+-secret/); + }); + + it("OAuth browser:粘错 state 失败可读;粘对 URL 成功,token 包加密入库", async () => { + const u = await create({ name: "fake 订阅 A", provider: "fake", authKind: "oauth" }); + expect(u.status).toBe("auth_required"); + const s1 = (await req("POST", `/pi/upstreams/${u.id}/oauth/start`, { mode: "browser" })).body.data; + expect(s1).toMatchObject({ mode: "browser", state: "pending", userCode: null }); + expect(s1.authorizeUrl).toMatch(/^https:\/\/fake-oauth\.invalid\/authorize\?state=/); + const bad = ( + await req("POST", `/pi/upstreams/${u.id}/oauth/complete`, { + sessionId: s1.sessionId, + pastedUrlOrCode: "http://localhost:1455/auth/callback?code=abc&state=nope", + }) + ).body.data; + expect(bad).toMatchObject({ state: "failed" }); + expect(bad.message).toMatch(/state/); + + const s2 = (await req("POST", `/pi/upstreams/${u.id}/oauth/start`, {})).body.data; + const state = new URL(s2.authorizeUrl).searchParams.get("state"); + const done = ( + await req("POST", `/pi/upstreams/${u.id}/oauth/complete`, { + sessionId: s2.sessionId, + pastedUrlOrCode: `http://localhost:1455/auth/callback?code=xyz&state=${state}`, + }) + ).body.data; + expect(done).toMatchObject({ + state: "succeeded", + upstream: { status: "ok", hasCredential: true, account: "fake-browser-user" }, + }); + expect(done.upstream.oauthExpiresAt).toBeTruthy(); + const row = await h.db + .selectFrom("pi_upstream") + .select("secret_ref") + .where("id", "=", u.id) + .executeTakeFirstOrThrow(); + const sec = await h.db + .selectFrom("secret") + .select("ciphertext") + .where("id", "=", row.secret_ref!) + .executeTakeFirstOrThrow(); + expect(Buffer.from(sec.ciphertext).toString("utf8")).not.toContain("fake-access"); + expect(JSON.stringify(done)).not.toContain("fake-access"); + // 过期会话 + const gone = (await req("POST", `/pi/upstreams/${u.id}/oauth/complete`, { sessionId: "nope" })).body.data; + expect(gone.state).toBe("expired"); + // 测试端点对 OAuth 上游可用 + expect((await req("POST", `/pi/upstreams/${u.id}/test`, {})).body.data).toMatchObject({ + ok: true, + source: "OAuth", + }); + // 登出 → 凭据删除 + const out = (await req("POST", `/pi/upstreams/${u.id}/oauth/logout`)).body.data; + expect(out).toMatchObject({ status: "auth_required", hasCredential: false }); + await req("DELETE", `/pi/upstreams/${u.id}`); + }); + + it("OAuth device:服务端轮询,未批准返回 pending,批准后 succeeded;同 provider 登录串行;端口占用提示", async () => { + control.deviceAutoApproveMs = null; + const a = await create({ name: "fake 订阅 D1", provider: "fake", authKind: "oauth" }); + const b = await create({ name: "fake 订阅 D2", provider: "fake", authKind: "oauth" }); + const s = (await req("POST", `/pi/upstreams/${a.id}/oauth/start`, { mode: "device" })).body.data; + expect(s).toMatchObject({ mode: "device", state: "pending", verificationUrl: "https://fake-oauth.invalid/device" }); + expect(s.userCode).toMatch(/^FAKE-/); + + // 同 provider 另一个上游的登录要等这一个结束 + expect((await req("POST", `/pi/upstreams/${b.id}/oauth/start`, { mode: "device" })).status).toBe(409); + + const p = (await req("POST", `/pi/upstreams/${a.id}/oauth/complete`, { sessionId: s.sessionId, waitMs: 100 })).body + .data; + expect(p.state).toBe("pending"); + control.approveDevice(); + const ok = (await req("POST", `/pi/upstreams/${a.id}/oauth/complete`, { sessionId: s.sessionId, waitMs: 5000 })) + .body.data; + expect(ok).toMatchObject({ state: "succeeded", upstream: { account: "fake-device-user", status: "ok" } }); + + // 端口占用 → 可读提示 + control.browserListenError = "listen EADDRINUSE: address already in use 127.0.0.1:1455"; + const f = (await req("POST", `/pi/upstreams/${b.id}/oauth/start`, { mode: "browser" })).body.data; + expect(f.state).toBe("failed"); + expect(f.message).toMatch(/端口 1455 被占用/); + control.browserListenError = null; + control.deviceAutoApproveMs = 3000; + await req("DELETE", `/pi/upstreams/${a.id}`); + await req("DELETE", `/pi/upstreams/${b.id}`); + }); + + /** 第二个“进程”:独立连接池 + 独立 secrets / settings */ + function secondProcess(): { ctx: ServerContext; close(): Promise } { + const db = new Kysely({ + dialect: new PostgresDialect({ + pool: new pg.Pool({ connectionString: PG_URL, max: 3, options: `-c search_path=${h.schema},public` }), + }), + }); + const ctx = { + ...h.ctx, + db, + secrets: createSecretStore({ db, masterKey: MASTER_KEY }), + settings: createSettingsReader({ db, logger: h.ctx.logger }), + } as ServerContext; + return { ctx, close: () => db.destroy() }; + } + + it("刷新任务:两个进程同时刷新只刷新一次,轮换后的 refresh token 写回;失败 → expired + 待处理事项,重新登录后关闭", async () => { + control.deviceAutoApproveMs = 0; + const u = await create({ name: "fake 订阅 R", provider: "fake", authKind: "oauth" }); + const s = (await req("POST", `/pi/upstreams/${u.id}/oauth/start`, { mode: "device" })).body.data; + await req("POST", `/pi/upstreams/${u.id}/oauth/complete`, { sessionId: s.sessionId, waitMs: 5000 }); + // 让 token 快过期 + const store = dbCredentialStore(h.ctx, u.id); + await store.modify(async (cur) => ({ ...(cur as object), expires: Date.now() + 60_000 }) as never); + const before = (await store.read()) as { refresh: string }; + + control.refreshCalls = 0; + control.refreshDelayMs = 150; + const p2 = secondProcess(); + try { + const [r1, r2] = await Promise.all([ + refreshPiTokens(h.ctx, { client: { fakeOAuth: control } }), + refreshPiTokens(p2.ctx, { client: { fakeOAuth: control } }), + ]); + expect(control.refreshCalls).toBe(1); + expect(r1.failed).toEqual([]); + expect(r2.failed).toEqual([]); + } finally { + await p2.close(); + control.refreshDelayMs = 0; + } + const after = (await store.read()) as { refresh: string; expires: number }; + expect(after.refresh).not.toBe(before.refresh); + expect(after.expires).toBeGreaterThan(Date.now() + 30 * 60_000); + // 已不在刷新窗口内:再跑一次不刷新 + await refreshPiTokens(h.ctx, { client: { fakeOAuth: control } }); + expect(control.refreshCalls).toBe(1); + + // refresh token 失效 + await store.modify(async (cur) => ({ ...(cur as object), expires: Date.now() + 60_000 }) as never); + control.failRefresh = true; + const r = await refreshPiTokens(h.ctx, { client: { fakeOAuth: control } }); + control.failRefresh = false; + expect(r.failed).toHaveLength(1); + const view = (await req("GET", `/pi/upstreams/${u.id}`)).body.data; + expect(view).toMatchObject({ status: "expired", hasCredential: true }); + expect(view.lastError).toMatch(/invalid_grant/); + const d = await h.db + .selectFrom("decision") + .selectAll() + .where("dedupe_key", "=", `pi_upstream_expired:${u.id}`) + .executeTakeFirstOrThrow(); + expect(d).toMatchObject({ status: "open", subject_type: "pi_upstream", subject_id: u.id }); + // 失效后不再自动刷新 + expect((await refreshPiTokens(h.ctx, { client: { fakeOAuth: control } })).checked).toBe(0); + + const s2 = (await req("POST", `/pi/upstreams/${u.id}/oauth/start`, { mode: "device" })).body.data; + const ok = (await req("POST", `/pi/upstreams/${u.id}/oauth/complete`, { sessionId: s2.sessionId, waitMs: 5000 })) + .body.data; + expect(ok.upstream.status).toBe("ok"); + const d2 = await h.db.selectFrom("decision").select("status").where("id", "=", d.id).executeTakeFirstOrThrow(); + expect(d2.status).toBe("resolved"); + control.deviceAutoApproveMs = 3000; + await req("DELETE", `/pi/upstreams/${u.id}`); + }); + + async function newRun(upstreamId: string | null) { + const r = await h.db + .insertInto("agent_run") + .values({ + trigger: "user", + goal: "测试上游选择", + scope: JSON.stringify({ capabilities: ["task.inspect"], allowedKinds: null }), + upstream_id: upstreamId, + }) + .returning("id") + .executeTakeFirstOrThrow(); + return r.id; + } + + it("runAgent:缺省用默认上游,指定 upstreamId 覆盖;key 由服务端解密注入请求;停用上游报可读错误", async () => { + const env = { ...agent.env(h.ctx), opts: { ...agent.env(h.ctx).opts, sessionDir } }; + const fake = await create({ name: "默认 fake", provider: "fake", authKind: "none", makeDefault: true }); + const custom = await create({ + name: "本地兼容端点", + provider: "openai-compatible", + authKind: "api_key", + apiKey: "sk-local-upstream-key", + baseUrl: mockUrl, + defaultModel: "local-model", + thinkingLevel: "off", + }); + + const r1 = await newRun(null); + const d1 = await drive(env, h.ctx.capabilities, r1, new AbortController().signal); + expect(d1).toMatchObject({ status: "completed", finalText: "fake 完成" }); + const row1 = await h.db.selectFrom("agent_run").selectAll().where("id", "=", r1).executeTakeFirstOrThrow(); + expect(row1).toMatchObject({ provider: "fake", upstream_id: fake.id }); + + const r2 = await newRun(custom.id); + const d2 = await drive(env, h.ctx.capabilities, r2, new AbortController().signal); + expect(d2, JSON.stringify(d2)).toMatchObject({ status: "completed", finalText: "上游回复" }); + expect(seenAuth.at(-1)).toBe("Bearer sk-local-upstream-key"); + const row2 = await h.db.selectFrom("agent_run").selectAll().where("id", "=", r2).executeTakeFirstOrThrow(); + expect(row2).toMatchObject({ provider: "openai-compatible", model: "local-model", upstream_id: custom.id }); + const events = await sql<{ payload: unknown }>`select payload from agent_run_event where run_id = ${r2}`.execute( + h.db, + ); + expect(JSON.stringify(events.rows)).not.toContain("sk-local-upstream-key"); + expect(h.lines.join("\n")).not.toContain("sk-local-upstream-key"); + + await req("PATCH", `/pi/upstreams/${custom.id}`, { enabled: false }); + const r3 = await newRun(custom.id); + const d3 = await drive(env, h.ctx.capabilities, r3, new AbortController().signal); + expect(d3).toMatchObject({ status: "failed" }); + expect(d3.message).toMatch(/已停用/); + }); +});