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 <[email protected]>
2026-09-28 02:23:22 +08:00
1 parent 743fe2486d
commit fd3ffe5a8d
@@ -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<RefreshReport> {
+  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;
+}