feat(license): 授权码即用户身份（codes 模式）与超管发码面板

- 新增 CodeLicenseProvider：每个授权码恒定派生唯一 user_id，会话/项目/主数据按码隔离，
  同一授权码跨浏览器/设备激活找回同一数据；内置超管码 aps@Z@Agent@Admin（永久+全角色）
- APS_LICENSE_CLIENTS 控制作用端（web/desktop），JMS 配置存在时 JMS 优先
- 设置面板新增「用户与授权」（仅 admin 可见）：登记姓名/手机号/公司，预设时长或
  自定义到期时间，自动生成授权码，停用即时生效，可删除并查看激活状态
- Web 端未激活弹「请提供授权码」激活卡片；Web 会话使用独立 aps_license_session Cookie
- /api/health 授权配置缺失时兜底返回 invalid 标记，不再 500
- aps_license_codes 表入库管理，SQLite 老库自动补列迁移
- 新增 21 项端到端用例；golden 回归 2398 通过

.env.example

@@ -82,6 +82,16 @@ APS_LICENSE_PROVIDER=unconfigured
 # mock 模式必须显式配置随机签名密钥和租户标识；npm run dev:desktop 会为本地开发自动生成。
 # APS_MOCK_LICENSE_SECRET=
 # APS_MOCK_LICENSE_TENANT_UUID=
+#
+# ---- 授权码模式（codes：授权码即用户身份，Web/桌面均可激活）----
+# 每个授权码唯一对应一个用户身份，智能体会话、项目、主数据全部绑定到该授权码并互相隔离；
+# 同一授权码可在任意浏览器/设备激活并找回同一份数据。
+# 内置超级管理员授权码 aps@Z@Agent@Admin（永久授权 + 全功能角色），其余授权码经 APS_LICENSE_CODES 发放。
+# APS_LICENSE_PROVIDER=codes
+# APS_LICENSE_SECRET=至少32个随机字符（HMAC 签名密钥，生产必须独立随机生成）
+# APS_LICENSE_TENANT_UUID=platform
+# APS_LICENSE_CLIENTS=web,desktop
+# APS_LICENSE_CODES={"APS-USER-001":"year","APS-USER-002":{"type":"month","name":"张三"}}
 
 # ---- 现场数据目录（可选；未配置时不绑定任何机器目录）----
 # KANGNI_DATA_DIR=

README.md

@@ -72,6 +72,25 @@ Web 端会从 JMS 读取启用企业，用户选择所属企业后使用真实
 `APS-HOUR-DEMO`、`APS-DAY-DEMO`、`APS-WEEK-DEMO`、`APS-MONTH-DEMO`、
 `APS-YEAR-DEMO`、`APS-PERP-DEMO`。
 
+### 授权码部署（codes 模式：授权码即用户身份）
+
+独立 Web 交付（不走 JMS）时可改用授权码认证：用户打开系统即看到"请提供授权码"，
+激活后智能体会话、项目、主数据全部绑定到该授权码；每个授权码只能看到自己的数据，
+同一授权码换浏览器/设备激活可找回同一份数据。配置示例：
+
+```bash
+APS_LICENSE_PROVIDER=codes
+APS_LICENSE_SECRET=至少32个随机字符
+APS_LICENSE_CLIENTS=web,desktop
+# 发放给用户的一一对应的授权码（类型：hour/day/week/month/year/permanent）
+APS_LICENSE_CODES={"APS-USER-001":"year","APS-USER-002":{"type":"month","name":"张三"}}
+```
+
+内置超级管理员授权码 `aps@Z@Agent@Admin`（永久授权 + 全功能角色），无需配置即可使用；
+用管理员码登录后，可在「设置 → 用户与授权」面板直接发放/停用/删除用户授权码
+（登记姓名、手机号、公司、有效时间，系统自动生成授权码），无需重启服务；
+`APS_AUTH_PROVIDER=jms` 同时存在时 JMS 优先，授权码模式不生效。
+
 ```bash
 # Web（浏览器 http://localhost:5173 ，API :8000）
 npm run dev:web

apps/web/src/api/client.ts

@@ -1735,3 +1735,67 @@ export async function postMeshWatchdog(body: { stallTimeoutSec?: number; maxRetr
   const resp = await apiFetch('/api/mesh/watchdog', { method: 'POST', body: JSON.stringify(body) });
   return (await resp.json()) as { error?: string };
 }
+
+// ---------------- 超管 · 用户与授权码管理（codes 模式） ----------------
+export type LicenseCodeActivation = {
+  status: string;
+  activatedAt: number;
+  lastSeenAt: number;
+  expiresAt: number | null;
+};
+
+export type LicenseCodeItem = {
+  id: string;
+  code: string;
+  name: string;
+  mobile: string;
+  company: string;
+  licenseType: string;
+  licenseLabel: string;
+  expiresAt?: number | null;
+  status: string;
+  createdAt: string;
+  activation: LicenseCodeActivation | null;
+};
+
+async function readApiError(resp: Response): Promise<Error> {
+  try {
+    const body = await resp.json() as { detail?: { message?: string } | string; error?: { message?: string } };
+    if (typeof body.detail === 'object' && body.detail?.message) return new Error(body.detail.message);
+    if (typeof body.detail === 'string') return new Error(body.detail);
+    if (body.error?.message) return new Error(body.error.message);
+  } catch { /* fall through */ }
+  return new Error(`请求失败（${resp.status}）`);
+}
+
+export async function fetchLicenseCodes(): Promise<LicenseCodeItem[]> {
+  const resp = await apiFetch('/api/admin/license/codes');
+  if (!resp.ok) throw await readApiError(resp);
+  return ((await resp.json()) as { codes: LicenseCodeItem[] }).codes;
+}
+
+export async function createLicenseCode(payload: {
+  name: string; mobile: string; company: string; licenseType: string; code?: string; expiresAt?: number;
+}): Promise<LicenseCodeItem> {
+  const resp = await apiFetch('/api/admin/license/codes', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify(payload),
+  });
+  if (!resp.ok) throw await readApiError(resp);
+  return ((await resp.json()) as { code: LicenseCodeItem }).code;
+}
+
+export async function setLicenseCodeStatus(id: string, status: 'active' | 'disabled'): Promise<void> {
+  const resp = await apiFetch(`/api/admin/license/codes/${encodeURIComponent(id)}/status`, {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ status }),
+  });
+  if (!resp.ok) throw await readApiError(resp);
+}
+
+export async function deleteLicenseCode(id: string): Promise<void> {
+  const resp = await apiFetch(`/api/admin/license/codes/${encodeURIComponent(id)}`, { method: 'DELETE' });
+  if (!resp.ok) throw await readApiError(resp);
+}

apps/web/src/auth/AuthEntry.tsx

@@ -48,7 +48,7 @@ export default function AuthEntry(props: {
         settings: 'Settings', result: props.hasVisualResult ? 'Open session result' : 'No session result', logout: 'Sign out',
       };
   const displayName = user?.fullname || user?.username || '';
-  const desktop = auth.clientMode === 'desktop';
+  const licenseMode = auth.clientMode === 'desktop' || auth.licenseMode;
   const licenseMeta = auth.license
     ? (auth.license.permanent
         ? auth.license.label
@@ -65,12 +65,12 @@ export default function AuthEntry(props: {
             className="account-trigger guest-login-trigger"
             disabled={checking}
             onClick={props.onLogin}
-            title={checking ? '正在验证访问状态' : (desktop ? '客户端授权' : '登录')}
+            title={checking ? '正在验证访问状态' : (licenseMode ? '授权激活' : '登录')}
           >
-            <span className="account-avatar">{desktop ? <SafetyCertificateOutlined /> : <UserOutlined />}</span>
+            <span className="account-avatar">{licenseMode ? <SafetyCertificateOutlined /> : <UserOutlined />}</span>
             <span className="account-trigger-copy">
-              <b>{checking ? '正在验证' : (desktop ? '授权进入' : '登录')}</b>
+              <b>{checking ? '正在验证' : (licenseMode ? '授权进入' : '登录')}</b>
-              <small>{checking ? '请稍候' : (desktop ? '激活后使用本机项目与会话' : '登录后同步项目与会话')}</small>
+              <small>{checking ? '请稍候' : (licenseMode ? '激活后使用项目与会话' : '登录后同步项目与会话')}</small>
             </span>
             {!checking && <LoginOutlined className="guest-login-arrow" />}
           </button>
@@ -111,7 +111,7 @@ export default function AuthEntry(props: {
           </button>
           <div className="account-menu-separator" />
           <button type="button" role="menuitem" className="account-logout" onClick={() => void auth.logout()}>
-            <LogoutOutlined /><span>{desktop ? '退出授权' : labels.logout}</span><RightOutlined />
+            <LogoutOutlined /><span>{licenseMode ? '退出授权' : labels.logout}</span><RightOutlined />
           </button>
           {menuNote && <div className="account-menu-note" role="status">{menuNote}</div>}
         </div>

apps/web/src/auth/AuthGate.tsx

@@ -27,7 +27,7 @@ function GuestSidebar(props: {
   onToggleCollapsed: () => void;
 }) {
   const auth = useAuth();
-  const accessWord = auth.clientMode === 'desktop' ? '授权' : '登录';
+  const accessWord = (auth.clientMode === 'desktop' || auth.licenseMode) ? '授权' : '登录';
   const action = props.checking ? undefined : props.onLogin;
   return (
     <aside className={`workspace-sidebar guest-sidebar${props.collapsed ? ' collapsed' : ''}`}>

apps/web/src/auth/AuthProvider.tsx

@@ -51,6 +51,7 @@ type AuthContextValue = {
   user: AuthUser | null;
   license: LicenseInfo | null;
   clientMode: ClientMode;
+  licenseMode: boolean;
   provider: string;
   error: string;
   login: (payload: LoginPayload) => Promise<void>;
@@ -84,6 +85,7 @@ export function AuthProvider(props: { children: ReactNode }) {
   const [user, setUser] = useState<AuthUser | null>(null);
   const [license, setLicense] = useState<LicenseInfo | null>(null);
   const clientMode = getClientMode();
+  const [licenseMode, setLicenseMode] = useState(clientMode === 'desktop');
   const [provider, setProvider] = useState('unconfigured');
   const [error, setError] = useState('');
   const clearError = useCallback(() => setError(''), []);
@@ -94,8 +96,10 @@ export function AuthProvider(props: { children: ReactNode }) {
     try {
       const health = await clientFetch('/api/health');
       if (health.ok) {
-        const info = await health.json() as { authProvider?: string; licenseProvider?: string };
+        const info = await health.json() as { authProvider?: string; licenseProvider?: string; licenseAuth?: boolean };
-        setProvider((clientMode === 'desktop' ? info.licenseProvider : info.authProvider) || 'unconfigured');
+        const useLicense = clientMode === 'desktop' || info.licenseAuth === true;
+        setLicenseMode(useLicense);
+        setProvider((useLicense ? info.licenseProvider : info.authProvider) || 'unconfigured');
       }
       const response = await clientFetch('/api/auth/me');
       if (!response.ok) {
@@ -155,7 +159,7 @@ export function AuthProvider(props: { children: ReactNode }) {
   }, [clientMode, provider]);
 
   const activateLicense = useCallback(async (code: string) => {
-    if (clientMode !== 'desktop') throw new Error('授权码仅可用于桌面客户端');
+    if (clientMode !== 'desktop' && !licenseMode) throw new Error('授权码在当前部署不可用');
     setError('');
     const response = await clientFetch('/api/auth/license/activate', {
       method: 'POST',
@@ -173,7 +177,7 @@ export function AuthProvider(props: { children: ReactNode }) {
     setLicense(body.license || null);
     setProvider(body.provider || provider);
     setStatus('authenticated');
-  }, [clientMode, provider]);
+  }, [clientMode, licenseMode, provider]);
 
   const logout = useCallback(async () => {
     try {
@@ -185,8 +189,8 @@ export function AuthProvider(props: { children: ReactNode }) {
   }, [refresh]);
 
   const value = useMemo<AuthContextValue>(
-    () => ({ status, user, license, clientMode, provider, error, login, clearError, activateLicense, logout, refresh }),
+    () => ({ status, user, license, clientMode, licenseMode, provider, error, login, clearError, activateLicense, logout, refresh }),
-    [status, user, license, clientMode, provider, error, login, clearError, activateLicense, logout, refresh],
+    [status, user, license, clientMode, licenseMode, provider, error, login, clearError, activateLicense, logout, refresh],
   );
   return <AuthContext.Provider value={value}>{props.children}</AuthContext.Provider>;
 }
@@ -399,16 +403,16 @@ export function LicenseDialog(props: { open: boolean; onClose: () => void }) {
           disabled={busy} onClick={props.onClose}><CloseOutlined /></button>
         <header className="login-brand">
           <span className="login-brand-mark license-brand-mark" aria-hidden="true"><SafetyCertificateOutlined /></span>
-          <div><h1 id="license-title">激活工业智核</h1><p>APS Desktop</p></div>
+          <div><h1 id="license-title">激活工业智核</h1><p>{auth.clientMode === 'desktop' ? 'APS Desktop' : 'APS Agent'}</p></div>
         </header>
         <div className="license-summary">
-          <strong>客户端授权</strong>
+          <strong>授权码激活</strong>
-          <span>授权时长由授权码决定</span>
+          <span>数据将绑定到该授权码，授权时长由授权码决定</span>
         </div>
         <form className="login-form" onSubmit={submit}>
           <label><span>授权码</span><div><KeyOutlined /><input autoFocus autoComplete="off" spellCheck={false}
-            value={code} onChange={event => setCode(event.target.value.toUpperCase())}
+            value={code} onChange={event => setCode(auth.provider === 'mock' ? event.target.value.toUpperCase() : event.target.value)}
-            placeholder="请输入客户端授权码" /></div></label>
+            placeholder="请提供授权码" /></div></label>
           <button className="login-submit" type="submit" disabled={busy || !code.trim()}>
             {busy ? '验证中...' : '授权进入'}
           </button>
@@ -429,7 +433,7 @@ export function LicenseDialog(props: { open: boolean; onClose: () => void }) {
 
 export function AccessDialog(props: { open: boolean; onClose: () => void }) {
   const auth = useAuth();
-  return auth.clientMode === 'desktop'
+  return auth.clientMode === 'desktop' || auth.licenseMode
     ? <LicenseDialog {...props} />
     : <LoginDialog {...props} />;
 }

apps/web/src/gov/GovConsole.tsx

@@ -23,6 +23,8 @@ import {
   type ConstraintItem, type ConstraintLibrary, type ConstraintProfile, type ScheduleParams, type SopCompilableItem,
 } from '../api/client';
 import SkillConsole from '../skills/SkillConsole';
+import LicenseAdminView from './LicenseAdminView';
+import { useAuth } from '../auth/AuthProvider';
 import KnowledgePanel from '../knowledge/KnowledgePanel';
 import { devToolsEnabled } from '../features/devtools';
 
@@ -31,10 +33,10 @@ export type GovSection =
   | 'gate.pending' | 'gate.history' | 'gate.audit' | 'gate.stats' | 'gate.tools' | 'gate.policy'
   | 'regen.modules' | 'regen.tests'
   | 'knowledge.assets'
-  | 'settings.personal' | 'settings.schedule' | 'settings.constraints' | 'settings.integration' | 'settings.skills';
+  | 'settings.personal' | 'settings.schedule' | 'settings.constraints' | 'settings.integration' | 'settings.skills' | 'settings.license';
 
 // dev: true 的分组属开发/运维工具，计划员界面不出现（需 ?devtools=1 开启）
-const NAV: { group: string; dev?: boolean; items: { id: GovSection; label: string }[] }[] = [
+const NAV: { group: string; dev?: boolean; items: { id: GovSection; label: string; admin?: boolean }[] }[] = [
   {
     group: '通用',
     items: [
@@ -43,6 +45,7 @@ const NAV: { group: string; dev?: boolean; items: { id: GovSection; label: strin
       { id: 'settings.constraints', label: '约束配置' },
       { id: 'settings.integration', label: '模型与集成' },
       { id: 'settings.skills', label: '算法 Skill' },
+      { id: 'settings.license', label: '用户与授权', admin: true },
       { id: 'knowledge.assets', label: '知识平台' },
     ],
   },
@@ -88,7 +91,12 @@ export default function GovConsole(props: {
     return () => document.removeEventListener('keydown', onKey);
   }, [props]);
 
-  const nav = useMemo(() => NAV.filter(g => !g.dev || devToolsEnabled()), []);
+  const auth = useAuth();
+  const isAdmin = Boolean(auth.user?.roles?.includes('admin'));
+  const nav = useMemo(() => NAV
+    .filter(g => !g.dev || devToolsEnabled())
+    .map(g => ({ ...g, items: g.items.filter(it => !it.admin || isAdmin) }))
+    .filter(g => g.items.length > 0), [isAdmin]);
 
   if (!props.open) return null;
 
@@ -145,6 +153,7 @@ export default function GovConsole(props: {
                 onSendCommand={props.onSendCommand}
               />
             )}
+            {section === 'settings.license' && <LicenseAdminView />}
             {section === 'knowledge.assets' && (
               <KnowledgePanel sessionId={props.sessionId ?? 'web'} />
             )}

apps/web/src/gov/LicenseAdminView.tsx

@@ -0,0 +1,216 @@
+// ============================================================
+// 用户与授权（moduleId: web-gov-license-admin, 可重生 ✅）
+// 仅超级管理员可见：发放授权码（姓名/手机号/公司/有效期，自动生成码）、
+// 停用/启用/删除，并查看每个码的激活状态。授权码即用户身份，数据按码隔离。
+// ============================================================
+import { useCallback, useEffect, useState } from 'react';
+
+import {
+  createLicenseCode,
+  deleteLicenseCode,
+  fetchLicenseCodes,
+  setLicenseCodeStatus,
+  type LicenseCodeItem,
+} from '../api/client';
+
+const LICENSE_TYPE_OPTIONS: { value: string; label: string }[] = [
+  { value: 'hour', label: '小时授权' },
+  { value: 'day', label: '天授权' },
+  { value: 'week', label: '周授权' },
+  { value: 'month', label: '月授权' },
+  { value: 'year', label: '年授权' },
+  { value: 'permanent', label: '永久授权' },
+  { value: 'custom', label: '自定义时间（选择到期时间）' },
+];
+
+function fmtTime(epoch?: number | null): string {
+  if (!epoch) return '—';
+  return new Date(epoch * 1000).toLocaleString('zh-CN', { hour12: false });
+}
+
+function activationText(item: LicenseCodeItem): string {
+  if (!item.activation) return '未激活';
+  const act = item.activation;
+  if (act.status === 'expired') return `已到期（${fmtTime(act.expiresAt)}）`;
+  return `已激活 · 最近使用 ${fmtTime(act.lastSeenAt)}`;
+}
+
+export default function LicenseAdminView() {
+  const [items, setItems] = useState<LicenseCodeItem[] | null>(null);
+  const [notice, setNotice] = useState<string | null>(null);
+  const [busy, setBusy] = useState(false);
+
+  const [name, setName] = useState('');
+  const [mobile, setMobile] = useState('');
+  const [company, setCompany] = useState('');
+  const [licenseType, setLicenseType] = useState('year');
+  const [customExpiry, setCustomExpiry] = useState('');
+
+  const load = useCallback(() => {
+    void fetchLicenseCodes()
+      .then(setItems)
+      .catch(err => {
+        setItems([]);
+        setNotice(err instanceof Error ? err.message : '加载授权码列表失败');
+      });
+  }, []);
+  useEffect(() => { load(); }, [load]);
+
+  const create = async () => {
+    if (!name.trim()) { setNotice('请填写用户名称'); return; }
+    if (licenseType === 'custom' && !customExpiry) { setNotice('请选择到期时间'); return; }
+    setBusy(true);
+    setNotice(null);
+    try {
+      const payload: Parameters<typeof createLicenseCode>[0] = {
+        name: name.trim(), mobile: mobile.trim(), company: company.trim(), licenseType,
+      };
+      if (licenseType === 'custom') {
+        payload.expiresAt = Math.floor(new Date(customExpiry).getTime() / 1000);
+      }
+      const created = await createLicenseCode(payload);
+      setNotice(`已为「${created.name}」生成授权码：${created.code}（请复制下发给用户）`);
+      setName(''); setMobile(''); setCompany('');
+      load();
+    } catch (err) {
+      setNotice(err instanceof Error ? err.message : '创建失败');
+    } finally {
+      setBusy(false);
+    }
+  };
+
+  const toggle = async (item: LicenseCodeItem) => {
+    setBusy(true);
+    setNotice(null);
+    try {
+      const next = item.status === 'active' ? 'disabled' : 'active';
+      await setLicenseCodeStatus(item.id, next);
+      setNotice(next === 'disabled' ? `已停用「${item.name}」的授权码` : `已启用「${item.name}」的授权码`);
+      load();
+    } catch (err) {
+      setNotice(err instanceof Error ? err.message : '操作失败');
+    } finally {
+      setBusy(false);
+    }
+  };
+
+  const remove = async (item: LicenseCodeItem) => {
+    if (!window.confirm(`确认删除「${item.name}」的授权码？删除后该码将无法激活。`)) return;
+    setBusy(true);
+    setNotice(null);
+    try {
+      await deleteLicenseCode(item.id);
+      setNotice(`已删除「${item.name}」的授权码`);
+      load();
+    } catch (err) {
+      setNotice(err instanceof Error ? err.message : '删除失败');
+    } finally {
+      setBusy(false);
+    }
+  };
+
+  const copy = async (code: string) => {
+    try {
+      await navigator.clipboard.writeText(code);
+      setNotice(`已复制授权码：${code}`);
+    } catch {
+      setNotice(`复制失败，请手动复制：${code}`);
+    }
+  };
+
+  return (
+    <>
+      <div className="gov-card">
+        <div className="gov-card-title">发放授权码</div>
+        <p className="gov-kv" style={{ marginBottom: 10 }}>
+          授权码即用户身份：用户激活后，其会话、项目与数据全部绑定到该授权码，互相隔离。
+          有效期从用户首次激活起算。
+        </p>
+        <label className="params-slider-row">
+          <span className="params-slider-label">用户名称</span>
+          <input type="text" value={name} onChange={e => setName(e.target.value)}
+            placeholder="必填，如：张三" style={{ flex: 1, minWidth: 140 }} />
+        </label>
+        <label className="params-slider-row">
+          <span className="params-slider-label">手机号</span>
+          <input type="text" value={mobile} onChange={e => setMobile(e.target.value)}
+            placeholder="选填" style={{ flex: 1, minWidth: 140 }} />
+        </label>
+        <label className="params-slider-row">
+          <span className="params-slider-label">公司名称</span>
+          <input type="text" value={company} onChange={e => setCompany(e.target.value)}
+            placeholder="选填" style={{ flex: 1, minWidth: 140 }} />
+        </label>
+        <label className="params-slider-row">
+          <span className="params-slider-label">有效时间</span>
+          <select value={licenseType} onChange={e => setLicenseType(e.target.value)}
+            style={{ flex: 1, minWidth: 140 }}>
+            {LICENSE_TYPE_OPTIONS.map(opt => (
+              <option key={opt.value} value={opt.value}>{opt.label}</option>
+            ))}
+          </select>
+        </label>
+        {licenseType === 'custom' && (
+          <label className="params-slider-row">
+            <span className="params-slider-label">到期时间</span>
+            <input type="datetime-local" value={customExpiry}
+              onChange={e => setCustomExpiry(e.target.value)} style={{ flex: 1, minWidth: 140 }} />
+          </label>
+        )}
+        <button type="button" className="gov-action" disabled={busy || !name.trim()} onClick={() => void create()}>
+          {busy ? '处理中…' : '生成授权码'}
+        </button>
+      </div>
+
+      {notice && <div className="gov-notice">{notice}</div>}
+
+      <div className="gov-card">
+        <div className="gov-card-title">用户与授权码</div>
+        {items === null && <div className="gov-empty">加载中…</div>}
+        {items !== null && items.length === 0 && (
+          <div className="gov-empty">尚未发放任何授权码。使用上方表单为用户生成。</div>
+        )}
+        {items !== null && items.length > 0 && (
+          <table className="grid gov-table">
+            <thead>
+              <tr>
+                <th style={{ textAlign: 'left' }}>用户</th><th>手机号</th><th>公司</th>
+                <th style={{ textAlign: 'left' }}>授权码</th><th>类型</th><th>状态</th>
+                <th style={{ textAlign: 'left' }}>激活情况</th><th>操作</th>
+              </tr>
+            </thead>
+            <tbody>
+              {items.map(item => (
+                <tr key={item.id}>
+                  <td style={{ textAlign: 'left' }}>{item.name}</td>
+                  <td>{item.mobile || '—'}</td>
+                  <td>{item.company || '—'}</td>
+                  <td style={{ textAlign: 'left', fontFamily: 'monospace' }}>
+                    {item.code}
+                    <button type="button" className="sidebar-mini-btn" title="复制授权码"
+                      style={{ marginLeft: 6 }} onClick={() => void copy(item.code)}>复制</button>
+                  </td>
+                  <td>{item.licenseLabel}{item.expiresAt ? `（至 ${fmtTime(item.expiresAt)}）` : ''}</td>
+                  <td>
+                    <span className={`risk ${item.status === 'active' ? 'r-low' : 'r-high'}`}>
+                      {item.status === 'active' ? '有效' : '已停用'}
+                    </span>
+                  </td>
+                  <td style={{ textAlign: 'left' }}>{activationText(item)}</td>
+                  <td>
+                    <button type="button" className="btn-ghost" disabled={busy}
+                      onClick={() => void toggle(item)}>
+                      {item.status === 'active' ? '停用' : '启用'}
+                    </button>
+                    <button type="button" className="btn-ghost" disabled={busy}
+                      onClick={() => void remove(item)}>删除</button>
+                  </td>
+                </tr>
+              ))}
+            </tbody>
+          </table>
+        )}
+      </div>
+    </>
+  );
+}

server/auth/licenses.py

@@ -17,7 +17,7 @@ from sqlalchemy import select
 from server.auth.context import IdentityContext
 from server.auth.providers import AuthError
 from server.db.database import get_session
-from server.db.models import LicenseActivation
+from server.db.models import LicenseActivation, LicenseCode
 
 
 LICENSE_DURATIONS: dict[str, int | None] = {
@@ -36,6 +36,7 @@ LICENSE_LABELS = {
     "month": "月授权",
     "year": "年授权",
     "permanent": "永久授权",
+    "custom": "自定义授权",
 }
 
 DEFAULT_MOCK_CODES: dict[str, str] = {
@@ -48,12 +49,37 @@ DEFAULT_MOCK_CODES: dict[str, str] = {
 }
 
 DESKTOP_COOKIE = "aps_desktop_session"
+WEB_LICENSE_COOKIE = "aps_license_session"
 
 
 def is_desktop_request(request: Request) -> bool:
     return (request.headers.get("x-aps-client") or "").strip().lower() == "desktop"
 
 
+def license_clients() -> frozenset[str]:
+    """授权码机制作用的客户端集合（APS_LICENSE_CLIENTS，默认仅 desktop，兼容既有部署）。"""
+    raw = (os.environ.get("APS_LICENSE_CLIENTS") or "desktop").strip().lower()
+    return frozenset(item.strip() for item in raw.split(",") if item.strip())
+
+
+def use_license_auth(request: Request) -> bool:
+    """判定当前请求是否应走授权码认证。
+
+    - 桌面客户端：始终走授权（与既有行为一致）。
+    - Web 客户端：仅当 APS_LICENSE_CLIENTS 包含 web、授权提供方为 codes（正式授权码模式）、
+      且未配置 JMS 认证（JMS 环境优先，避免与 MES 集成冲突）时走授权码。
+    """
+    if is_desktop_request(request):
+        return True
+    if "web" not in license_clients():
+        return False
+    if (os.environ.get("APS_LICENSE_PROVIDER") or "").strip().lower() != "codes":
+        return False
+    from server.auth.providers import get_auth_provider
+
+    return get_auth_provider().mode != "jms"
+
+
 def _b64encode(raw: bytes) -> str:
     return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
 
@@ -294,6 +320,253 @@ class MockLicenseProvider(LicenseProvider):
         return identity, self._issue(identity, device_hash)
 
 
+ADMIN_LICENSE_CODE = "aps@Z@Agent@Admin"
+ADMIN_LICENSE_ROLES = ("system", "admin", "planner", "approver", "auditor", "scheduler", "license")
+DEFAULT_CODE_ROLES = ("license", "planner")
+
+
+class CodeLicenseProvider(LicenseProvider):
+    """正式授权码提供方：授权码即用户身份。
+
+    每个授权码唯一对应一个稳定的 user_id（由授权码哈希派生），智能体会话、项目、
+    主数据等全部按 tenant_uuid + user_id 归属到该授权码；不同授权码之间数据天然隔离，
+    同一授权码可在任意浏览器/设备激活并找回同一份数据。
+    内置超级管理员授权码 aps@Z@Agent@Admin（永久授权 + 全功能角色），
+    其余授权码经 APS_LICENSE_CODES（JSON）发放。
+    """
+
+    mode = "codes"
+
+    def __init__(self) -> None:
+        secret = (os.environ.get("APS_LICENSE_SECRET") or "").strip()
+        if len(secret) < 32:
+            raise AuthError(
+                "LICENSE_CONFIG_INVALID",
+                "授权配置缺失：APS_LICENSE_SECRET（至少 32 个字符）",
+                503,
+            )
+        self.secret = secret.encode()
+        tenant = (os.environ.get("APS_LICENSE_TENANT_UUID") or "platform").strip()
+        self.tenant_uuid = tenant or "platform"
+
+    def _catalog(self) -> dict[str, dict[str, Any]]:
+        catalog: dict[str, dict[str, Any]] = {
+            ADMIN_LICENSE_CODE: {
+                "type": "permanent",
+                "roles": ADMIN_LICENSE_ROLES,
+                "name": "aps-admin",
+            },
+        }
+        configured = (os.environ.get("APS_LICENSE_CODES") or "").strip()
+        if not configured:
+            return catalog
+        try:
+            data = json.loads(configured)
+        except json.JSONDecodeError as exc:
+            raise AuthError("LICENSE_CONFIG_INVALID", "授权码配置无效（APS_LICENSE_CODES 须为 JSON）", 503) from exc
+        if isinstance(data, dict):
+            for code, grant in data.items():
+                if not isinstance(code, str) or not code.strip():
+                    continue
+                if isinstance(grant, dict):
+                    license_type = str(grant.get("type") or "")
+                    roles = grant.get("roles")
+                    name = grant.get("name")
+                else:
+                    license_type, roles, name = str(grant), None, None
+                if license_type not in LICENSE_DURATIONS:
+                    continue
+                entry: dict[str, Any] = {"type": license_type}
+                if isinstance(roles, (list, tuple)) and roles:
+                    entry["roles"] = tuple(str(role) for role in roles)
+                else:
+                    entry["roles"] = DEFAULT_CODE_ROLES
+                if isinstance(name, str) and name.strip():
+                    entry["name"] = name.strip()[:64]
+                catalog[code.strip()] = entry
+        # 超管在设置面板发放的授权码（DB 登记，覆盖同名环境配置）
+        catalog.update(self._db_codes())
+        return catalog
+
+    def _db_codes(self) -> dict[str, dict[str, Any]]:
+        with get_session() as session:
+            rows = session.scalars(select(LicenseCode).where(
+                LicenseCode.tenant_uuid == self.tenant_uuid,
+                LicenseCode.status == "active",
+                LicenseCode.deleted == 0,
+            )).all()
+        result: dict[str, dict[str, Any]] = {}
+        for row in rows:
+            if row.license_type not in LICENSE_DURATIONS and not row.expires_at:
+                continue
+            db_entry: dict[str, Any] = {"type": row.license_type, "roles": DEFAULT_CODE_ROLES}
+            if row.expires_at:
+                db_entry["expiresAt"] = int(row.expires_at)
+            if row.name:
+                db_entry["name"] = row.name
+            result[row.code] = db_entry
+        return result
+
+    @staticmethod
+    def _scope(request: Request) -> str:
+        scope = "desktop" if is_desktop_request(request) else "web"
+        if scope not in license_clients():
+            raise AuthError("LICENSE_CLIENT_DISABLED", "授权码在当前部署的该客户端不可用", 403)
+        return scope
+
+    @staticmethod
+    def _code_user_id(tenant_uuid: str, code: str) -> int:
+        # 授权码即身份：同一授权码恒定派生同一 user_id（保持在有符号 BIGINT 范围内）。
+        digest = _sha256(f"aps-license:{tenant_uuid}:{code}")
+        return int(digest[:15], 16)
+
+    def _issue(self, identity: IdentityContext) -> str:
+        payload = {**identity.to_dict(), "issuer": "codes"}
+        raw = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+        body = _b64encode(raw)
+        signature = _b64encode(hmac.new(self.secret, body.encode("ascii"), hashlib.sha256).digest())
+        return f"{body}.{signature}"
+
+    def _verify_token(self, token: str) -> IdentityContext:
+        try:
+            body, signature = token.split(".", 1)
+            expected = _b64encode(hmac.new(self.secret, body.encode("ascii"), hashlib.sha256).digest())
+            if not hmac.compare_digest(signature, expected):
+                raise ValueError("signature")
+            payload = json.loads(_b64decode(body))
+            activation_id = str(payload.get("activation_id") or "")
+            if not activation_id:
+                raise ValueError("activation")
+        except Exception as exc:
+            raise AuthError("LICENSE_INVALID", "授权凭证无效", 401) from exc
+
+        now = int(time.time())
+        with get_session() as session:
+            row = session.get(LicenseActivation, activation_id)
+            if row is None or row.deleted != 0:
+                raise AuthError("LICENSE_INVALID", "授权凭证无效", 401)
+            if row.status != "active" or row.revoked_at is not None:
+                raise AuthError("LICENSE_REVOKED", "授权已被停用", 401)
+            if row.expires_at is not None and row.expires_at <= now:
+                row.status = "expired"
+                session.commit()
+                raise AuthError("LICENSE_EXPIRED", "授权码已到期，请联系发放方续期", 401)
+            code_row = session.scalar(select(LicenseCode).where(
+                LicenseCode.code_hash == row.code_hash,
+                LicenseCode.deleted == 0,
+            ))
+            if code_row is not None and code_row.status != "active":
+                raise AuthError("LICENSE_REVOKED", "授权码已被停用，请联系管理员", 401)
+            if now - row.last_seen_at >= 3600:
+                row.last_seen_at = now
+                session.commit()
+        roles = payload.get("roles")
+        return IdentityContext(
+            user_id=row.user_id,
+            username=str(payload.get("username") or "license-user"),
+            fullname=str(payload.get("fullname") or "授权用户"),
+            tenant_uuid=row.tenant_uuid,
+            roles=tuple(str(role) for role in roles) if isinstance(roles, list) and roles else DEFAULT_CODE_ROLES,
+            expires_at=row.expires_at or 4_102_444_800,  # 2100-01-01；DB 校验仍支持停用/到期
+            auth_kind="license",
+            license_type=row.license_type,
+            license_activated_at=row.activated_at,
+            license_expires_at=row.expires_at,
+            activation_id=row.id,
+        )
+
+    async def activate(self, request: Request, payload: dict[str, Any]) -> tuple[IdentityContext, str]:
+        self._scope(request)
+        code = str(payload.get("code") or "").strip()
+        if not code:
+            raise AuthError("LICENSE_CODE_REQUIRED", "请输入授权码", 400)
+        entry = self._catalog().get(code)
+        if entry is None:
+            raise AuthError("LICENSE_CODE_INVALID", "授权码无效", 401)
+
+        now = int(time.time())
+        code_hash = _sha256(f"codes:{self.tenant_uuid}:{code}")
+        with get_session() as session:
+            row = session.scalar(select(LicenseActivation).where(
+                LicenseActivation.code_hash == code_hash,
+                LicenseActivation.deleted == 0,
+            ))
+            if row is not None:
+                if row.revoked_at is not None:
+                    raise AuthError("LICENSE_REVOKED", "授权已被停用", 401)
+                if row.expires_at is not None and row.expires_at <= now:
+                    if row.status != "expired":
+                        row.status = "expired"
+                        session.commit()
+                    raise AuthError("LICENSE_EXPIRED", "授权码已到期，请联系发放方续期", 401)
+                if row.status != "active":
+                    raise AuthError("LICENSE_REVOKED", "授权已被停用", 401)
+                row.last_seen_at = now
+                session.commit()
+                session.refresh(row)
+            else:
+                seconds = LICENSE_DURATIONS.get(entry["type"])
+                absolute_expiry = entry.get("expiresAt")
+                if absolute_expiry:
+                    expires_at = int(absolute_expiry)
+                elif seconds is not None:
+                    expires_at = now + seconds
+                else:
+                    expires_at = None
+                if expires_at is not None and expires_at <= now:
+                    raise AuthError("LICENSE_EXPIRED", "授权码已到期，请联系发放方续期", 401)
+                row = LicenseActivation(
+                    id=uuid.uuid4().hex,
+                    code_hash=code_hash,
+                    code_hint=f"{code[:2]}***{code[-2:]}" if len(code) > 4 else "***",
+                    device_hash=_sha256(f"codes-shared:{self.tenant_uuid}"),
+                    user_id=self._code_user_id(self.tenant_uuid, code),
+                    license_type=entry["type"],
+                    status="active",
+                    activated_at=now,
+                    expires_at=expires_at,
+                    last_seen_at=now,
+                    revoked_at=None,
+                    tenant_uuid=self.tenant_uuid,
+                    creator_id=None,
+                    updater_id=None,
+                    updated_at=_utc_text(now),
+                    deleted=0,
+                    created_at=_utc_text(now),
+                )
+                session.add(row)
+                session.commit()
+                session.refresh(row)
+
+        identity = IdentityContext(
+            user_id=row.user_id,
+            username=str(entry.get("name") or f"lic-{row.code_hint}"),
+            fullname="超级管理员" if code == ADMIN_LICENSE_CODE else str(entry.get("name") or "授权用户"),
+            tenant_uuid=row.tenant_uuid,
+            roles=tuple(entry.get("roles") or DEFAULT_CODE_ROLES),
+            expires_at=row.expires_at or 4_102_444_800,
+            auth_kind="license",
+            license_type=row.license_type,
+            license_activated_at=row.activated_at,
+            license_expires_at=row.expires_at,
+            activation_id=row.id,
+        )
+        return identity, self._issue(identity)
+
+    async def authenticate(self, request: Request) -> IdentityContext:
+        scope = self._scope(request)
+        cookie = DESKTOP_COOKIE if scope == "desktop" else WEB_LICENSE_COOKIE
+        token = request.cookies.get(cookie) or ""
+        if not token:
+            message = "请先使用授权码激活客户端" if scope == "desktop" else "请提供授权码激活后使用"
+            raise AuthError("LICENSE_REQUIRED", message, 401)
+        return self._verify_token(token)
+
+    async def refresh(self, request: Request) -> tuple[IdentityContext, str]:
+        identity = await self.authenticate(request)
+        return identity, self._issue(identity)
+
+
 _provider: LicenseProvider | None = None
 _provider_mode: str | None = None
 
@@ -302,6 +575,11 @@ def get_license_provider() -> LicenseProvider:
     global _provider, _provider_mode
     mode = (os.environ.get("APS_LICENSE_PROVIDER") or "unconfigured").strip().lower()
     if _provider is None or _provider_mode != mode:
-        _provider = MockLicenseProvider() if mode == "mock" else UnconfiguredLicenseProvider()
+        if mode == "mock":
+            _provider = MockLicenseProvider()
+        elif mode == "codes":
+            _provider = CodeLicenseProvider()
+        else:
+            _provider = UnconfiguredLicenseProvider()
         _provider_mode = mode
     return _provider

server/auth/middleware.py

@@ -10,7 +10,7 @@ from starlette.datastructures import MutableHeaders
 
 from server.auth import agent_tokens
 from server.auth.context import IdentityContext, bind_identity, reset_identity
-from server.auth.licenses import get_license_provider, is_desktop_request
+from server.auth.licenses import get_license_provider, use_license_auth
 from server.auth.providers import AuthError, get_auth_provider
 
 PUBLIC_API_PATHS = {
@@ -160,7 +160,7 @@ class AuthenticationMiddleware:
             try:
                 provider = (
                     get_license_provider()
-                    if is_desktop_request(request)
+                    if use_license_auth(request)
                     else get_auth_provider()
                 )
                 identity = await provider.authenticate(request)

server/db/database.py

@@ -147,6 +147,9 @@ def _migrate_legacy_sqlite(engine: Engine) -> None:
             "updated_at": "VARCHAR(32) NOT NULL DEFAULT ''",
             "deleted": "BIGINT NOT NULL DEFAULT 0",
         },
+        "aps_license_codes": {
+            "expires_at": "BIGINT",
+        },
     }
     tables = set(inspector.get_table_names())
     with engine.begin() as connection:

server/db/models.py

@@ -363,3 +363,27 @@ class LicenseActivation(TenantAuditMixin, Base):
         Index("ix_license_device_status", "device_hash", "status", "deleted"),
         Index("ix_license_tenant_user", "tenant_uuid", "user_id", "deleted"),
     )
+
+
+class LicenseCode(TenantAuditMixin, Base):
+    """授权码发放登记（codes 模式）：超级管理员在设置面板维护。
+
+    code 明文存储——授权码本身就是发给用户的凭证，超管需要随时查看/复制下发；
+    激活侧一律按 code_hash 匹配（与 CodeLicenseProvider 的派生口径一致）。
+    """
+
+    __tablename__ = "aps_license_codes"
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    code: Mapped[str] = mapped_column(String(128), nullable=False)
+    code_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    name: Mapped[str] = mapped_column(String(64), nullable=False, default="")
+    mobile: Mapped[str] = mapped_column(String(32), default="")
+    company: Mapped[str] = mapped_column(String(128), default="")
+    license_type: Mapped[str] = mapped_column(String(16), nullable=False)
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
+    expires_at: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    created_at: Mapped[str] = mapped_column(String(32), default="")
+
+    __table_args__ = (
+        UniqueConstraint("code_hash", "deleted", name="uk_license_code_hash"),
+    )

server/gateway/app.py

@@ -74,13 +74,20 @@ from server.state.checkpoints import get_checkpoints   # 成对快照仓（时
 from server.state.store import get_store               # 世界状态单例
 from server.auth.context import get_identity
 from server.auth.middleware import AuthenticationMiddleware
-from server.auth.licenses import DESKTOP_COOKIE, get_license_provider, is_desktop_request
+from server.auth.licenses import (
+    DESKTOP_COOKIE,
+    WEB_LICENSE_COOKIE,
+    get_license_provider,
+    is_desktop_request,
+    use_license_auth,
+)
 from server.auth.providers import AuthError, get_auth_provider
 from server.gateway.plan_api import router as plan_router
 from server.gateway.masterdata_api import router as masterdata_router
 from server.gateway.governance_api import router as governance_router, gov_router as governance_audit_router
 from server.gateway.mps_api import router as mps_router
 from server.gateway.agent_api import router as agent_router
+from server.gateway.license_admin_api import router as license_admin_router
 
 
 logger = logging.getLogger(__name__)
@@ -1022,6 +1029,7 @@ def create_app() -> FastAPI:
     app.include_router(governance_audit_router)
     app.include_router(mps_router)
     app.include_router(agent_router)
+    app.include_router(license_admin_router)
 
     @app.exception_handler(PermissionError)
     async def permission_error_handler(_request: Request, exc: PermissionError) -> Response:
@@ -1047,11 +1055,12 @@ def create_app() -> FastAPI:
             path="/",
         )
 
-    def set_license_cookie(response: Response, token: str, expires_at: int | None) -> None:
+    def set_license_cookie(response: Response, token: str, expires_at: int | None, request: Request) -> None:
         import time
         remaining = max(1, (expires_at - int(time.time()))) if expires_at else 10 * 365 * 24 * 60 * 60
+        cookie_name = DESKTOP_COOKIE if is_desktop_request(request) else WEB_LICENSE_COOKIE
         response.set_cookie(
-            DESKTOP_COOKIE,
+            cookie_name,
             token,
             httponly=True,
             secure=(os.environ.get("APS_AUTH_COOKIE_SECURE") or "0") == "1",
@@ -1076,13 +1085,23 @@ def create_app() -> FastAPI:
 
     @app.get("/api/health")
     async def health(request: Request) -> dict:
+        try:
+            license_mode = get_license_provider().mode
+        except AuthError as exc:
+            # 授权配置错误不应打挂探活：明确标示，由部署方修复环境变量
+            license_mode = f"invalid:{exc.code}"
+        try:
+            license_auth = use_license_auth(request)
+        except AuthError:
+            license_auth = False
         return {
             "ok": True,
             "version": _app_version(),
             "interfaceVersion": INTERFACE_VERSION,
             "clientMode": "desktop" if is_desktop_request(request) else "web",
             "authProvider": get_auth_provider().mode,
-            "licenseProvider": get_license_provider().mode,
+            "licenseProvider": license_mode,
+            "licenseAuth": license_auth,
         }
 
     @app.get("/api/features")
@@ -1097,9 +1116,9 @@ def create_app() -> FastAPI:
 
     @app.post("/api/auth/login")
     async def auth_login(req: AuthLoginRequest, request: Request, response: Response):
-        if is_desktop_request(request):
+        if use_license_auth(request):
             return auth_error_response(AuthError(
-                "DESKTOP_USES_LICENSE", "桌面客户端请使用授权码激活", 400,
+                "LICENSE_ACTIVATION_REQUIRED", "当前部署请使用授权码激活", 400,
             ))
         try:
             identity, token = await get_auth_provider().login(req.model_dump())
@@ -1111,9 +1130,9 @@ def create_app() -> FastAPI:
     @app.get("/api/auth/tenants")
     async def auth_tenants(request: Request):
         """公开租户列表已下线（避免暴露企业信息）；请手填企业名称登录。"""
-        if is_desktop_request(request):
+        if use_license_auth(request):
             return auth_error_response(AuthError(
-                "DESKTOP_USES_LICENSE", "桌面客户端请使用授权码激活", 400,
+                "LICENSE_ACTIVATION_REQUIRED", "当前部署请使用授权码激活", 400,
             ))
         return auth_error_response(AuthError(
             "AUTH_TENANT_LIST_REMOVED",
@@ -1128,9 +1147,9 @@ def create_app() -> FastAPI:
         tenantCode: str = "",
         tenantName: str = "",
     ):
-        if is_desktop_request(request):
+        if use_license_auth(request):
             return auth_error_response(AuthError(
-                "DESKTOP_USES_LICENSE", "桌面客户端请使用授权码激活", 400,
+                "LICENSE_ACTIVATION_REQUIRED", "当前部署请使用授权码激活", 400,
             ))
         try:
             return await get_auth_provider().precheck(
@@ -1141,9 +1160,9 @@ def create_app() -> FastAPI:
 
     @app.get("/api/auth/captcha")
     async def auth_captcha(request: Request):
-        if is_desktop_request(request):
+        if use_license_auth(request):
             return auth_error_response(AuthError(
-                "DESKTOP_USES_LICENSE", "桌面客户端请使用授权码激活", 400,
+                "LICENSE_ACTIVATION_REQUIRED", "当前部署请使用授权码激活", 400,
             ))
         try:
             return await get_auth_provider().captcha()
@@ -1157,7 +1176,7 @@ def create_app() -> FastAPI:
             identity, token = await provider.activate(request, req.model_dump())
         except AuthError as exc:
             return auth_error_response(exc)
-        set_license_cookie(response, token, identity.license_expires_at)
+        set_license_cookie(response, token, identity.license_expires_at, request)
         return {
             "user": identity.to_dict(),
             "provider": provider.mode,
@@ -1167,7 +1186,7 @@ def create_app() -> FastAPI:
     @app.get("/api/auth/me")
     async def auth_me(request: Request) -> dict:
         identity = get_identity(required=True)
-        if is_desktop_request(request):
+        if use_license_auth(request):
             provider = get_license_provider()
             return {
                 "user": identity.to_dict(),
@@ -1178,27 +1197,28 @@ def create_app() -> FastAPI:
 
     @app.post("/api/auth/refresh")
     async def auth_refresh(request: Request, response: Response):
-        provider = get_license_provider() if is_desktop_request(request) else get_auth_provider()
+        provider = get_license_provider() if use_license_auth(request) else get_auth_provider()
         try:
             identity, token = await provider.refresh(request)
         except AuthError as exc:
             return auth_error_response(exc)
-        if is_desktop_request(request):
+        if use_license_auth(request):
-            set_license_cookie(response, token, identity.license_expires_at)
+            set_license_cookie(response, token, identity.license_expires_at, request)
         else:
             set_session_cookie(response, token)
         return {
             "user": identity.to_dict(),
             "provider": provider.mode,
-            "license": provider.describe(identity) if is_desktop_request(request) else None,
+            "license": provider.describe(identity) if use_license_auth(request) else None,
         }
 
     @app.post("/api/auth/logout")
     async def auth_logout(request: Request, response: Response) -> dict:
-        provider = get_license_provider() if is_desktop_request(request) else get_auth_provider()
+        provider = get_license_provider() if use_license_auth(request) else get_auth_provider()
         await provider.logout(request)
         response.delete_cookie("aps_session", path="/")
         response.delete_cookie(DESKTOP_COOKIE, path="/")
+        response.delete_cookie(WEB_LICENSE_COOKIE, path="/")
         return {"ok": True}
 
     @app.get("/api/auth/users")
@@ -4371,7 +4391,7 @@ def create_app() -> FastAPI:
             ingest_actor = actor or "desktop-updater"       # 本机可信通道（updater 进程）
         else:                                               # web 模式：要求登录身份
             try:
-                if is_desktop_request(request):
+                if use_license_auth(request):
                     identity = await get_license_provider().authenticate(request)
                 else:
                     identity = await get_auth_provider().authenticate(request)

server/gateway/license_admin_api.py

@@ -0,0 +1,218 @@
+# ============================================================
+# 超级管理员授权码管理 API（moduleId: gateway-license-admin-api, 可重生 ✅）
+# codes 模式（APS_LICENSE_PROVIDER=codes）下，超管在设置面板「用户与授权」
+# 发放/停用/删除授权码；每个授权码即一个用户身份，数据按码隔离。
+# ============================================================
+from __future__ import annotations
+
+import hashlib
+import os
+import secrets
+import time
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+
+from server.auth.context import IdentityContext, get_identity
+from server.auth.licenses import ADMIN_LICENSE_CODE, LICENSE_DURATIONS, LICENSE_LABELS
+from server.db.database import get_session
+from server.db.models import LicenseActivation, LicenseCode
+
+router = APIRouter(prefix="/api/admin/license", tags=["license-admin"])
+
+# 去掉 0/O/1/I/L 等易混淆字符，便于人工抄录下发
+_CODE_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
+
+
+def _raise(status_code: int, code: str, message: str) -> None:
+    raise HTTPException(status_code=status_code, detail={"code": code, "message": message})
+
+
+def _require_admin() -> IdentityContext:
+    identity = get_identity(required=True)
+    if "admin" not in identity.roles:
+        _raise(403, "LICENSE_ADMIN_REQUIRED", "仅超级管理员可管理用户与授权码")
+    return identity
+
+
+def _codes_enabled() -> None:
+    if (os.environ.get("APS_LICENSE_PROVIDER") or "").strip().lower() != "codes":
+        _raise(409, "LICENSE_MODE_DISABLED", "当前部署未启用授权码模式（APS_LICENSE_PROVIDER=codes）")
+
+
+def _tenant() -> str:
+    tenant = (os.environ.get("APS_LICENSE_TENANT_UUID") or "platform").strip()
+    return tenant or "platform"
+
+
+def _code_hash(tenant: str, code: str) -> str:
+    # 与 CodeLicenseProvider 的派生口径一致：codes:{tenant}:{code}
+    return hashlib.sha256(f"codes:{tenant}:{code}".encode()).hexdigest()
+
+
+def _generate_code() -> str:
+    return "-".join("".join(secrets.choice(_CODE_ALPHABET) for _ in range(4)) for _ in range(4))
+
+
+def _utc_text(epoch: int) -> str:
+    return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat()
+
+
+class LicenseCodeCreateRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=64)
+    mobile: str = Field(default="", max_length=32)
+    company: str = Field(default="", max_length=128)
+    licenseType: str = Field(min_length=1, max_length=16)
+    code: str = Field(default="", max_length=128)
+    expiresAt: int | None = Field(default=None)
+
+
+class LicenseCodeStatusRequest(BaseModel):
+    status: str = Field(min_length=1, max_length=16)
+
+
+def _serialize(row: LicenseCode, activation: LicenseActivation | None) -> dict:
+    return {
+        "id": row.id,
+        "code": row.code,
+        "name": row.name,
+        "mobile": row.mobile,
+        "company": row.company,
+        "licenseType": row.license_type,
+        "licenseLabel": LICENSE_LABELS.get(row.license_type, row.license_type),
+        "expiresAt": row.expires_at,
+        "status": row.status,
+        "createdAt": row.created_at,
+        "activation": None if activation is None else {
+            "status": activation.status,
+            "activatedAt": activation.activated_at,
+            "lastSeenAt": activation.last_seen_at,
+            "expiresAt": activation.expires_at,
+        },
+    }
+
+
+@router.get("/codes")
+async def list_license_codes() -> dict:
+    _codes_enabled()
+    _require_admin()
+    tenant = _tenant()
+    with get_session() as session:
+        rows = list(session.scalars(select(LicenseCode).where(
+            LicenseCode.tenant_uuid == tenant,
+            LicenseCode.deleted == 0,
+        ).order_by(LicenseCode.created_at.desc())).all())
+        activations: dict[str, LicenseActivation] = {}
+        hashes = [row.code_hash for row in rows]
+        if hashes:
+            for act in session.scalars(select(LicenseActivation).where(
+                LicenseActivation.code_hash.in_(hashes),
+                LicenseActivation.deleted == 0,
+            )).all():
+                activations[act.code_hash] = act
+    return {"codes": [_serialize(row, activations.get(row.code_hash)) for row in rows]}
+
+
+@router.post("/codes")
+async def create_license_code(req: LicenseCodeCreateRequest) -> dict:
+    _codes_enabled()
+    identity = _require_admin()
+    license_type = req.licenseType.strip().lower()
+    now = int(time.time())
+    expires_at = req.expiresAt
+    if license_type == "custom":
+        if not expires_at:
+            _raise(400, "LICENSE_EXPIRY_REQUIRED", "自定义授权必须选择到期时间")
+    elif license_type not in LICENSE_DURATIONS:
+        _raise(400, "LICENSE_TYPE_INVALID", "授权时长类型无效")
+    if expires_at is not None and expires_at <= now:
+        _raise(400, "LICENSE_EXPIRY_PAST", "到期时间必须晚于当前时间")
+    tenant = _tenant()
+    code = req.code.strip()
+    with get_session() as session:
+        if code:
+            if code == ADMIN_LICENSE_CODE:
+                _raise(409, "LICENSE_CODE_DUPLICATED", "该授权码为内置超级管理员码")
+            code_hash = _code_hash(tenant, code)
+            clash = session.scalar(select(LicenseCode).where(
+                LicenseCode.code_hash == code_hash,
+                LicenseCode.deleted == 0,
+            ))
+            if clash is not None:
+                _raise(409, "LICENSE_CODE_DUPLICATED", "授权码已存在")
+        else:
+            code_hash = ""
+            for _ in range(20):
+                candidate = _generate_code()
+                if candidate == ADMIN_LICENSE_CODE:
+                    continue
+                candidate_hash = _code_hash(tenant, candidate)
+                clash = session.scalar(select(LicenseCode).where(
+                    LicenseCode.code_hash == candidate_hash,
+                    LicenseCode.deleted == 0,
+                ))
+                if clash is None:
+                    code, code_hash = candidate, candidate_hash
+                    break
+            if not code:
+                _raise(500, "LICENSE_CODE_GEN_FAILED", "授权码生成失败，请重试")
+        now = int(time.time())
+        row = LicenseCode(
+            id=uuid.uuid4().hex,
+            code=code,
+            code_hash=code_hash,
+            name=req.name.strip(),
+            mobile=req.mobile.strip(),
+            company=req.company.strip(),
+            license_type=license_type,
+            status="active",
+            expires_at=expires_at,
+            tenant_uuid=tenant,
+            creator_id=identity.user_id or None,
+            updater_id=identity.user_id or None,
+            updated_at=_utc_text(now),
+            deleted=0,
+            created_at=_utc_text(now),
+        )
+        session.add(row)
+        session.commit()
+        session.refresh(row)
+    return {"code": _serialize(row, None)}
+
+
+@router.post("/codes/{code_id}/status")
+async def set_license_code_status(code_id: str, req: LicenseCodeStatusRequest) -> dict:
+    _codes_enabled()
+    identity = _require_admin()
+    status = req.status.strip().lower()
+    if status not in {"active", "disabled"}:
+        _raise(400, "LICENSE_STATUS_INVALID", "状态无效（active/disabled）")
+    now = int(time.time())
+    with get_session() as session:
+        row = session.get(LicenseCode, code_id)
+        if row is None or row.deleted != 0:
+            _raise(404, "LICENSE_CODE_NOT_FOUND", "授权码不存在")
+        row.status = status
+        row.updater_id = identity.user_id or None
+        row.updated_at = _utc_text(now)
+        session.commit()
+    return {"ok": True, "status": status}
+
+
+@router.delete("/codes/{code_id}")
+async def delete_license_code(code_id: str) -> dict:
+    _codes_enabled()
+    identity = _require_admin()
+    now = int(time.time())
+    with get_session() as session:
+        row = session.get(LicenseCode, code_id)
+        if row is None or row.deleted != 0:
+            _raise(404, "LICENSE_CODE_NOT_FOUND", "授权码不存在")
+        row.deleted = 1
+        row.updater_id = identity.user_id or None
+        row.updated_at = _utc_text(now)
+        session.commit()
+    return {"ok": True}

tests/golden/test_web_license_codes_auth.py

@@ -0,0 +1,432 @@
+"""正式授权码（codes）模式：Web 端授权码即身份的端到端用例。"""
+
+from __future__ import annotations
+
+import json
+import time
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from tests.auth_provider import install_test_auth
+
+TENANT = "tenant-codes-000000000000000001"
+CODES = {
+    "USER-CODE-ALPHA": "year",
+    "USER-CODE-BETA": {"type": "month", "name": "beta-user"},
+    "USER-CODE-HOUR": "hour",
+}
+
+
+@pytest.fixture()
+def codes_app(tmp_path, monkeypatch):
+    monkeypatch.setenv("APS_LICENSE_PROVIDER", "codes")
+    monkeypatch.setenv("APS_LICENSE_SECRET", "codes-test-secret-0123456789abcdef")
+    monkeypatch.setenv("APS_LICENSE_TENANT_UUID", TENANT)
+    monkeypatch.setenv("APS_LICENSE_CLIENTS", "web,desktop")
+    monkeypatch.setenv("APS_LICENSE_CODES", json.dumps(CODES))
+    # 钉死认证提供方：套件中途 server.main 的 load_dotenv 可能把开发机 .env 的 jms 重新注入
+    monkeypatch.setenv("APS_AUTH_PROVIDER", "unconfigured")
+    monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "codes.db"))
+    monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
+
+    from server.auth import licenses
+    from server.auth import providers as auth_providers
+    from server.db.database import reset_engine
+    from server.state import store as world_store
+
+    install_test_auth(monkeypatch, TENANT)
+    licenses._provider = None
+    licenses._provider_mode = None
+    auth_providers._provider = None
+    auth_providers._provider_mode = None
+    world_store._stores.clear()
+    reset_engine()
+
+    from server.gateway.app import create_app
+
+    app = create_app()
+    yield app
+
+    reset_engine()
+    world_store._stores.clear()
+    licenses._provider = None
+    licenses._provider_mode = None
+    auth_providers._provider = None
+    auth_providers._provider_mode = None
+
+
+def test_health_reports_license_auth(codes_app):
+    client = TestClient(codes_app)
+    body = client.get("/api/health").json()
+    assert body["ok"] is True
+    assert body["licenseProvider"] == "codes"
+    assert body["licenseAuth"] is True
+
+
+def test_health_survives_missing_secret(tmp_path, monkeypatch):
+    monkeypatch.setenv("APS_LICENSE_PROVIDER", "codes")
+    monkeypatch.delenv("APS_LICENSE_SECRET", raising=False)
+    monkeypatch.setenv("APS_LICENSE_CLIENTS", "web")
+    monkeypatch.setenv("APS_AUTH_PROVIDER", "unconfigured")
+    monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "bad.db"))
+    monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
+
+    from server.auth import licenses
+    from server.auth import providers as auth_providers
+    from server.db.database import reset_engine
+    from server.state import store as world_store
+
+    install_test_auth(monkeypatch, TENANT)
+    licenses._provider = None
+    licenses._provider_mode = None
+    auth_providers._provider = None
+    auth_providers._provider_mode = None
+    world_store._stores.clear()
+    reset_engine()
+
+    from server.gateway.app import create_app
+
+    client = TestClient(create_app())
+    response = client.get("/api/health")
+    assert response.status_code == 200
+    assert response.json()["licenseProvider"] == "invalid:LICENSE_CONFIG_INVALID"
+
+    reset_engine()
+    world_store._stores.clear()
+    licenses._provider = None
+    licenses._provider_mode = None
+    auth_providers._provider = None
+    auth_providers._provider_mode = None
+
+
+def test_provider_requires_secret(monkeypatch):
+    from server.auth.licenses import CodeLicenseProvider
+    from server.auth.providers import AuthError
+
+    monkeypatch.delenv("APS_LICENSE_SECRET", raising=False)
+    with pytest.raises(AuthError, match="授权配置缺失") as exc:
+        CodeLicenseProvider()
+    assert exc.value.code == "LICENSE_CONFIG_INVALID"
+
+
+def test_web_activation_binds_identity_and_isolates_data(codes_app):
+    alpha = TestClient(codes_app)
+    beta = TestClient(codes_app)
+
+    response = alpha.post("/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"})
+    assert response.status_code == 200
+    body = response.json()
+    assert body["user"]["auth_kind"] == "license"
+    assert body["user"]["tenant_uuid"] == TENANT
+    assert body["license"]["type"] == "year"
+    alpha_id = body["user"]["user_id"]
+
+    assert beta.post("/api/auth/license/activate", json={"code": "USER-CODE-BETA"}).status_code == 200
+    beta_me = beta.get("/api/auth/me").json()
+    assert beta_me["user"]["username"] == "beta-user"
+    assert beta_me["user"]["user_id"] != alpha_id
+
+    assert alpha.post(
+        "/api/projects", json={"id": "alpha_private", "name": "授权A的私有项目"},
+    ).status_code == 200
+    workspace = beta.get("/api/workspace")
+    assert workspace.status_code == 200
+    assert workspace.json()["projects"] == []
+    assert beta.get("/api/projects/alpha_private/members").status_code == 404
+
+    # 同一授权码换浏览器激活：找回同一身份与同一份数据
+    alpha_second_browser = TestClient(codes_app)
+    assert alpha_second_browser.post(
+        "/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"},
+    ).status_code == 200
+    me_again = alpha_second_browser.get("/api/auth/me").json()
+    assert me_again["user"]["user_id"] == alpha_id
+    projects = alpha_second_browser.get("/api/workspace").json()["projects"]
+    assert any(project["id"] == "alpha_private" for project in projects)
+
+
+def test_admin_license_code_is_permanent_with_admin_roles(codes_app):
+    client = TestClient(codes_app)
+    response = client.post("/api/auth/license/activate", json={"code": "aps@Z@Agent@Admin"})
+    assert response.status_code == 200
+    body = response.json()
+    assert body["user"]["fullname"] == "超级管理员"
+    assert "admin" in body["user"]["roles"]
+    assert body["license"]["type"] == "permanent"
+    assert body["license"]["expiresAt"] is None
+    assert body["license"]["permanent"] is True
+
+
+def test_codes_are_case_sensitive(codes_app):
+    client = TestClient(codes_app)
+    response = client.post("/api/auth/license/activate", json={"code": "user-code-alpha"})
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "LICENSE_CODE_INVALID"
+
+
+def test_invalid_and_blank_code_rejected(codes_app):
+    client = TestClient(codes_app)
+    response = client.post("/api/auth/license/activate", json={"code": "NO-SUCH-CODE"})
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "LICENSE_CODE_INVALID"
+    blank = client.post("/api/auth/license/activate", json={"code": "   "})
+    assert blank.status_code == 400
+    assert blank.json()["error"]["code"] == "LICENSE_CODE_REQUIRED"
+
+
+def test_unactivated_web_request_requires_license(codes_app):
+    client = TestClient(codes_app)
+    response = client.get("/api/auth/me")
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "LICENSE_REQUIRED"
+    assert "授权码" in response.json()["error"]["message"]
+
+
+def test_login_endpoint_redirects_to_license_under_codes(codes_app):
+    client = TestClient(codes_app)
+    response = client.post("/api/auth/login", json={
+        "method": "password", "username": "planner", "password": "test",
+    })
+    assert response.status_code == 400
+    assert response.json()["error"]["code"] == "LICENSE_ACTIVATION_REQUIRED"
+
+
+def test_expired_code_is_rejected_on_next_request(codes_app):
+    client = TestClient(codes_app)
+    assert client.post(
+        "/api/auth/license/activate", json={"code": "USER-CODE-HOUR"},
+    ).status_code == 200
+
+    from server.db.database import get_session
+    from server.db.models import LicenseActivation
+
+    with get_session() as session:
+        row = session.scalar(select(LicenseActivation))
+        assert row is not None
+        row.expires_at = int(time.time()) - 1
+        session.commit()
+
+    response = client.get("/api/auth/me")
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "LICENSE_EXPIRED"
+
+    again = client.post("/api/auth/license/activate", json={"code": "USER-CODE-HOUR"})
+    assert again.status_code == 401
+    assert again.json()["error"]["code"] == "LICENSE_EXPIRED"
+
+
+def test_desktop_client_also_activates_with_codes(codes_app):
+    client = TestClient(codes_app)
+    headers = {"X-APS-Client": "desktop", "X-APS-Device-ID": "device-codes-0001"}
+    response = client.post("/api/auth/license/activate", headers=headers, json={"code": "USER-CODE-ALPHA"})
+    assert response.status_code == 200
+    me = client.get("/api/auth/me", headers=headers)
+    assert me.status_code == 200
+    assert me.json()["user"]["auth_kind"] == "license"
+
+
+def test_logout_clears_web_license_session(codes_app):
+    client = TestClient(codes_app)
+    assert client.post("/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"}).status_code == 200
+    assert client.get("/api/auth/me").status_code == 200
+    assert client.post("/api/auth/logout").status_code == 200
+    assert client.get("/api/auth/me").status_code == 401
+
+
+# ---------------- 超管授权码管理 API ----------------
+
+def _admin_client(app) -> TestClient:
+    client = TestClient(app)
+    assert client.post("/api/auth/license/activate", json={"code": "aps@Z@Agent@Admin"}).status_code == 200
+    return client
+
+
+def test_admin_issues_code_and_user_activates_with_name(codes_app):
+    admin = _admin_client(codes_app)
+    created = admin.post("/api/admin/license/codes", json={
+        "name": "张三", "mobile": "13800001111", "company": "锐扬", "licenseType": "month",
+    })
+    assert created.status_code == 200
+    issued = created.json()["code"]
+    assert issued["name"] == "张三"
+    assert len(issued["code"]) == 19  # XXXX-XXXX-XXXX-XXXX
+
+    listed = admin.get("/api/admin/license/codes").json()["codes"]
+    assert any(item["id"] == issued["id"] for item in listed)
+
+    user = TestClient(codes_app)
+    response = user.post("/api/auth/license/activate", json={"code": issued["code"]})
+    assert response.status_code == 200
+    body = response.json()
+    assert body["user"]["username"] == "张三"
+    assert body["user"]["fullname"] == "张三"
+    assert body["license"]["type"] == "month"
+
+    listed_after = admin.get("/api/admin/license/codes").json()["codes"]
+    entry = next(item for item in listed_after if item["id"] == issued["id"])
+    assert entry["activation"] is not None
+    assert entry["activation"]["status"] == "active"
+
+
+def test_non_admin_cannot_manage_codes(codes_app):
+    user = TestClient(codes_app)
+    assert user.post("/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"}).status_code == 200
+    response = user.get("/api/admin/license/codes")
+    assert response.status_code == 403
+    assert response.json()["detail"]["code"] == "LICENSE_ADMIN_REQUIRED"
+    denied = user.post("/api/admin/license/codes", json={"name": "越权", "licenseType": "year"})
+    assert denied.status_code == 403
+
+
+def test_disable_code_blocks_existing_session_and_new_activation(codes_app):
+    admin = _admin_client(codes_app)
+    issued = admin.post("/api/admin/license/codes", json={
+        "name": "李四", "licenseType": "year",
+    }).json()["code"]
+
+    user = TestClient(codes_app)
+    assert user.post("/api/auth/license/activate", json={"code": issued["code"]}).status_code == 200
+    assert user.get("/api/auth/me").status_code == 200
+
+    assert admin.post(
+        f"/api/admin/license/codes/{issued['id']}/status", json={"status": "disabled"},
+    ).status_code == 200
+
+    denied = user.get("/api/auth/me")
+    assert denied.status_code == 401
+    assert denied.json()["error"]["code"] == "LICENSE_REVOKED"
+    reactivate = user.post("/api/auth/license/activate", json={"code": issued["code"]})
+    assert reactivate.status_code == 401
+
+
+def test_delete_code_removes_it_from_catalog(codes_app):
+    admin = _admin_client(codes_app)
+    issued = admin.post("/api/admin/license/codes", json={
+        "name": "王五", "licenseType": "day",
+    }).json()["code"]
+    assert admin.delete(f"/api/admin/license/codes/{issued['id']}").status_code == 200
+
+    user = TestClient(codes_app)
+    response = user.post("/api/auth/license/activate", json={"code": issued["code"]})
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "LICENSE_CODE_INVALID"
+
+
+def test_duplicate_and_admin_code_rejected(codes_app):
+    admin = _admin_client(codes_app)
+    first = admin.post("/api/admin/license/codes", json={
+        "name": "甲", "licenseType": "year", "code": "CUSTOM-CODE-1",
+    })
+    assert first.status_code == 200
+    dup = admin.post("/api/admin/license/codes", json={
+        "name": "乙", "licenseType": "year", "code": "CUSTOM-CODE-1",
+    })
+    assert dup.status_code == 409
+    admin_code = admin.post("/api/admin/license/codes", json={
+        "name": "丙", "licenseType": "year", "code": "aps@Z@Agent@Admin",
+    })
+    assert admin_code.status_code == 409
+
+
+@pytest.fixture()
+def mock_license_app(tmp_path, monkeypatch):
+    monkeypatch.setenv("APS_LICENSE_PROVIDER", "mock")
+    monkeypatch.setenv("APS_MOCK_LICENSE_SECRET", "license-test-secret-0123456789abcdef")
+    monkeypatch.setenv("APS_MOCK_TENANT_UUID", TENANT)
+    monkeypatch.setenv("APS_AUTH_PROVIDER", "unconfigured")
+    monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "mock.db"))
+    monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
+
+    from server.auth import licenses
+    from server.auth import providers as auth_providers
+    from server.db.database import reset_engine
+    from server.state import store as world_store
+
+    install_test_auth(monkeypatch, TENANT)
+    licenses._provider = None
+    licenses._provider_mode = None
+    auth_providers._provider = None
+    auth_providers._provider_mode = None
+    world_store._stores.clear()
+    reset_engine()
+
+    from server.gateway.app import create_app
+
+    app = create_app()
+    yield app
+
+    reset_engine()
+    world_store._stores.clear()
+    licenses._provider = None
+    licenses._provider_mode = None
+    auth_providers._provider = None
+    auth_providers._provider_mode = None
+
+
+def test_admin_api_requires_codes_mode(mock_license_app):
+    # mock 模式下管理接口明确拒绝，避免误以为配置生效
+    client = TestClient(mock_license_app)
+    headers = {"X-APS-Client": "desktop", "X-APS-Device-ID": "device-admin-0001"}
+    assert client.post(
+        "/api/auth/license/activate", headers=headers, json={"code": "APS-PERP-DEMO"},
+    ).status_code == 200
+    response = client.get("/api/admin/license/codes", headers=headers)
+    assert response.status_code == 409
+    assert response.json()["detail"]["code"] == "LICENSE_MODE_DISABLED"
+
+
+def test_custom_expiry_code_uses_absolute_deadline(codes_app):
+    admin = _admin_client(codes_app)
+    deadline = int(time.time()) + 3 * 24 * 3600
+    created = admin.post("/api/admin/license/codes", json={
+        "name": "自定义用户", "licenseType": "custom", "expiresAt": deadline,
+    })
+    assert created.status_code == 200
+    issued = created.json()["code"]
+    assert issued["licenseType"] == "custom"
+    assert issued["expiresAt"] == deadline
+
+    user = TestClient(codes_app)
+    body = user.post("/api/auth/license/activate", json={"code": issued["code"]}).json()
+    assert body["license"]["type"] == "custom"
+    assert body["license"]["label"] == "自定义授权"
+    assert body["license"]["expiresAt"] == deadline
+
+
+def test_custom_expiry_validation(codes_app):
+    admin = _admin_client(codes_app)
+    missing = admin.post("/api/admin/license/codes", json={
+        "name": "缺时间", "licenseType": "custom",
+    })
+    assert missing.status_code == 400
+    assert missing.json()["detail"]["code"] == "LICENSE_EXPIRY_REQUIRED"
+
+    past = admin.post("/api/admin/license/codes", json={
+        "name": "过去时间", "licenseType": "custom", "expiresAt": int(time.time()) - 60,
+    })
+    assert past.status_code == 400
+    assert past.json()["detail"]["code"] == "LICENSE_EXPIRY_PAST"
+
+
+def test_expired_custom_code_cannot_activate(codes_app):
+    admin = _admin_client(codes_app)
+    deadline = int(time.time()) + 3600
+    issued = admin.post("/api/admin/license/codes", json={
+        "name": "将到期", "licenseType": "custom", "expiresAt": deadline,
+    }).json()["code"]
+
+    # 直接把码的截止时间改到过去，模拟到期后激活
+    from server.db.database import get_session
+    from server.db.models import LicenseCode
+
+    with get_session() as session:
+        row = session.get(LicenseCode, issued["id"])
+        row.expires_at = int(time.time()) - 1
+        session.commit()
+
+    user = TestClient(codes_app)
+    response = user.post("/api/auth/license/activate", json={"code": issued["code"]})
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "LICENSE_EXPIRED"