feat(auth): sync Keycloak users for authorization

The legacy online simulation path still exported internal ACTIVE states as STATUS keywords, which EPANET 2 rejects. Preserve numeric settings and EPANET 3 output while passing the format through both exporters. Surface the original engine error inside the model transaction so failed runs roll back before result storage.

Regression guard: six failures reproduced on the deployed baseline, including native input validation. All 34 focused export, service and endpoint checks pass after the fix; the real project also returns native node and link results.

.env.example

@@ -1,90 +1,83 @@
-# TJWater Server 环境变量配置模板
-# 复制此文件为 .env 并填写实际值
-# CI/CD: 生产环境变量由 Dev 主机的受控 backend.env 注入，不要将完整 .env 保存为 Gitea 仓库密钥。
-ENVIRONMENT="production"
-NETWORK_NAME="tjwater"
-# ============================================
-# 敏感配置加密 (必填)
-# ============================================
+# TJWaterServerBinary 环境变量模板
+# 复制为 .env 后按启用的功能填写；不要提交 .env 或任何真实凭据。
+# 生产配置由部署主机的受控 backend.env 注入，不随镜像或 CI/CD 同步本地 .env。
+# 进程环境变量优先；修改 Compose env_file 后需重新创建容器。
 
-# Fernet 格式，生产环境必须替换为独立密钥
-# 生成方式: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
-# 用于项目数据库 DSN、GeoServer 管理密码等敏感配置
+# 基础运行
+PROJECT_NAME="TJWater Server"
+ENVIRONMENT="production"
+API_V1_STR="/api/v1"
+NETWORK_NAME="tjwater"
+
+# Fernet 格式；用于加密项目数据库 DSN、GeoServer 管理密码等敏感配置。
+# 生成方式：python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
 DATABASE_ENCRYPTION_KEY="replace-with-generated-fernet-key"
 
-# ============================================
-# 数据库配置 (PostgreSQL)
-# ============================================
+# 业务数据库（PostgreSQL）
 DB_NAME="tjwater"
 DB_HOST="localhost"
 DB_PORT="5432"
 DB_USER="tjwater"
 DB_PASSWORD="password"
 
-# ============================================
-# 数据库配置 (TimescaleDB)
-# ============================================
+# 时序数据库（TimescaleDB）
 TIMESCALEDB_DB_NAME="tjwater"
 TIMESCALEDB_DB_HOST="localhost"
 TIMESCALEDB_DB_PORT="5433"
 TIMESCALEDB_DB_USER="tjwater"
 TIMESCALEDB_DB_PASSWORD="password"
 
-# ============================================
-# 元数据数据库配置 (Metadata DB)
-# ============================================
+# 元数据数据库
 METADATA_DB_NAME="system_hub"
 METADATA_DB_HOST="localhost"
 METADATA_DB_PORT="5432"
 METADATA_DB_USER="tjwater"
 METADATA_DB_PASSWORD="password"
+METADATA_DB_POOL_SIZE=5
+METADATA_DB_MAX_OVERFLOW=10
 
-# Per-project synchronous connection pools
-PROJECT_PG_CACHE_SIZE="16"
-PROJECT_TS_CACHE_SIZE="16"
-PROJECT_PG_POOL_MIN_SIZE="0"
-PROJECT_PG_POOL_SIZE="4"
-PROJECT_PG_MAX_OVERFLOW="2"
-PROJECT_TS_POOL_MIN_SIZE="0"
-PROJECT_TS_POOL_MAX_SIZE="4"
+# 项目数据库连接池、缓存与模板
+PROJECT_PG_CACHE_SIZE=16
+PROJECT_TS_CACHE_SIZE=16
+PROJECT_PG_POOL_MIN_SIZE=0
+PROJECT_PG_POOL_SIZE=4
+PROJECT_PG_MAX_OVERFLOW=2
+PROJECT_TS_POOL_MIN_SIZE=0
+PROJECT_TS_POOL_MAX_SIZE=4
 WNDB_SCHEMA_TEMPLATE_DB_NAME="tjwater_v2_schema_template"
 TIMESCALEDB_SCHEMA_TEMPLATE_DB_NAME="tjwater_v2_timescale_template"
-WNDB_TEMP_DB_MAX_COUNT="8"
+WNDB_TEMP_DB_MAX_COUNT=8
 
-# ============================================
 # GeoServer 项目供应
-# ============================================
 GEOSERVER_URL="http://localhost:8080/geoserver"
 GEOSERVER_USERNAME="admin"
 GEOSERVER_PASSWORD="password"
-# 留空时分别复用 DB_HOST / DB_PORT / DB_USER / DB_PASSWORD。
+# 留空时分别复用 DB_HOST、DB_PORT、DB_USER、DB_PASSWORD。
 # GeoServer 在容器内运行时，GEOSERVER_DB_HOST 通常应填写数据库服务名。
 GEOSERVER_DB_HOST=""
 GEOSERVER_DB_PORT=""
 GEOSERVER_DB_USER=""
 GEOSERVER_DB_PASSWORD=""
-# 30 天客户端缓存，单位为秒。
-GEOSERVER_CLIENT_CACHE_SECONDS="2592000"
+# 客户端缓存时间，单位为秒；默认 30 天。
+GEOSERVER_CLIENT_CACHE_SECONDS=2592000
 
-# ============================================
-# Keycloak JWT (可选)
-# ============================================
-KEYCLOAK_PUBLIC_KEY="-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----"
-KEYCLOAK_ALGORITHM=RS256
+# Keycloak JWT 校验
+# 未启用认证时可留空；生产环境应注入完整公钥和预期 audience。
+KEYCLOAK_PUBLIC_KEY=""
+KEYCLOAK_ALGORITHM="RS256"
 KEYCLOAK_AUDIENCE="account"
 KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS=900
 
+# 浏览器跨域访问（主 API）
+# 必须填写明确来源（协议、域名或 IP、端口），英文逗号分隔；不允许 *。
+CORS_ALLOWED_ORIGINS=http://127.0.0.1:5173,http://localhost:5173,http://127.0.0.1:15173,http://localhost:15173
 
-# ============================================
-# Bocha Web Search API
-# ============================================
-BOCHA_API_KEY="sk-your-bocha-api-key"
+# Bocha 网页搜索（旧版仍保留；未启用时令牌留空）
+BOCHA_API_KEY=""
 BOCHA_WEB_SEARCH_URL="https://api.bochaai.com/v1/web-search"
 BOCHA_WEB_SEARCH_TIMEOUT_SECONDS=30
 
-# ============================================
-# Tianditu Geocoding API
-# ============================================
-TIANDITU_GEOCODER_TOKEN="your-tianditu-geocoder-token"
+# 天地图地理编码（旧版使用 TIANDITU_GEOCODER_TOKEN）
+TIANDITU_GEOCODER_TOKEN=""
 TIANDITU_GEOCODER_URL="https://api.tianditu.gov.cn/geocoder"
 TIANDITU_GEOCODER_TIMEOUT_SECONDS=30

AUTHENTICATION_AND_USER_MANAGEMENT.md

@@ -37,11 +37,17 @@ but still need a project membership for project business APIs.
 Every authenticated metadata-user resolution validates the Keycloak access token
 and reads `sub`, `preferred_username`, and `email` claims. The
 backend finds `users` by `keycloak_id = sub`, rejects inactive or missing users,
-then refreshes `username`, `email`, and `last_login_at`.
+then refreshes `username`, `email`, and `last_login_at` for active users.
 
-This keeps local display data current without changing the identity binding.
-There is no Keycloak webhook requirement; second-level user or permission sync is
-out of scope unless explicitly requested later.
+System administrators explicitly synchronize the Keycloak user directory from
+the system-management page. New identities are stored with role `user` and
+`is_active = false`; existing identities keep their TJWater role, active state,
+superuser flag, and project memberships. This makes pending accounts visible
+without requiring the user to log in and without granting access automatically.
+The administrator's Keycloak account must have the `realm-management`
+`query-users` and `view-users` client roles. The backend forwards that access
+token only to the same token issuer's Keycloak Admin API; no management
+credential is exposed to the browser.
 
 ## Admin APIs
 

README.md

@@ -43,6 +43,22 @@ conda run -n server uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
 conda activate server
 ```
 
+## CORS 配置
+
+后端通过 `.env` 或进程环境变量中的 `CORS_ALLOWED_ORIGINS` 设置允许的浏览器页面来源，
+进程环境变量优先。必须使用英文逗号分隔的明确来源，不能填写 `*`，例如：
+
+```dotenv
+CORS_ALLOWED_ORIGINS=https://demo.waternetwork.cn,http://192.168.1.133,https://192.168.1.133,http://192.168.1.110,https://192.168.1.110
+```
+
+未设置时默认只允许本机 `5173`、`15173` 开发来源。来源需包含协议、域名或 IP、端口，
+不带路径；非默认端口须逐项列出。CORS 匹配浏览器页面来源，不是 Nginx 连接 IP 或域名解析结果。
+保留跨域凭据支持及 `Authorization`、`X-Project-Id` 请求头，接口认证和权限校验继续生效。
+
+修改后需重启后端进程。容器部署需同步更新宿主机的受控 `backend.env`；
+使用 Compose `env_file` 注入时，需要重新创建容器才能读取新值，CI/CD 不会自动同步本地 `.env`。
+
 ## 常用命令
 
 ```bash

app/api/v1/endpoints/admin_metadata.py

@@ -12,11 +12,17 @@ from app.auth.metadata_dependencies import (
     get_current_metadata_admin,
     get_metadata_repository,
 )
+from app.auth.keycloak_dependencies import (
+    get_current_keycloak_payload,
+    oauth2_optional,
+)
 from app.core.audit import AuditAction, log_audit_event
 from app.domain.schemas.admin_metadata import (
     AdminProjectCreateRequest,
     AdminProjectResponse,
     AdminProjectUpdateRequest,
+    KeycloakUserSyncIssue,
+    KeycloakUserSyncResponse,
     MetadataUsersBatchSyncRequest,
     MetadataUserResponse,
     MetadataUserSyncRequest,
@@ -36,6 +42,11 @@ from app.infra.db.metadb.repositories.metadata_repository import (
     MetadataRepository,
     ProjectDbRouting,
 )
+from app.services.keycloak_user_sync import (
+    KeycloakUserSyncConfigurationError,
+    KeycloakUserSyncRequestError,
+    fetch_keycloak_users,
+)
 
 router = APIRouter()
 
@@ -229,6 +240,94 @@ async def sync_metadata_users_batch(
     return results
 
 
+@router.post(
+    "/admin/keycloak-user-syncs",
+    response_model=KeycloakUserSyncResponse,
+)
+async def sync_keycloak_user_directory(
+    current_user=Depends(get_current_metadata_admin),
+    keycloak_payload: dict = Depends(get_current_keycloak_payload),
+    keycloak_access_token: str | None = Depends(oauth2_optional),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> KeycloakUserSyncResponse:
+    try:
+        keycloak_users = await fetch_keycloak_users(
+            issuer=str(keycloak_payload.get("iss") or ""),
+            bearer_token=keycloak_access_token or "",
+        )
+    except KeycloakUserSyncConfigurationError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=str(exc),
+        ) from exc
+    except KeycloakUserSyncRequestError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=str(exc),
+        ) from exc
+
+    created = 0
+    updated = 0
+    issues: list[KeycloakUserSyncIssue] = []
+    for keycloak_user in keycloak_users:
+        if keycloak_user.service_account_client_id:
+            issues.append(
+                KeycloakUserSyncIssue(
+                    username=keycloak_user.username,
+                    reason="service_account",
+                )
+            )
+            continue
+        if not keycloak_user.email:
+            issues.append(
+                KeycloakUserSyncIssue(
+                    username=keycloak_user.username,
+                    reason="missing_email",
+                )
+            )
+            continue
+        try:
+            _, was_created = await metadata_repo.sync_user_snapshot_from_keycloak(
+                keycloak_id=keycloak_user.keycloak_id,
+                username=keycloak_user.username,
+                email=keycloak_user.email,
+            )
+        except IntegrityError:
+            await metadata_repo.session.rollback()
+            issues.append(
+                KeycloakUserSyncIssue(
+                    username=keycloak_user.username,
+                    reason="identity_conflict",
+                )
+            )
+        except SQLAlchemyError as exc:
+            await metadata_repo.session.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail="Metadata database error during Keycloak user synchronization",
+            ) from exc
+        else:
+            created += int(was_created)
+            updated += int(not was_created)
+
+    result = KeycloakUserSyncResponse(
+        scanned=len(keycloak_users),
+        created=created,
+        updated=updated,
+        skipped=len(issues),
+        issues=issues,
+    )
+    await log_audit_event(
+        action=AuditAction.UPDATE,
+        user_id=current_user.id,
+        resource_type="keycloak_user_directory",
+        request_data=result.model_dump(mode="json", exclude={"issues"}),
+        response_status=status.HTTP_200_OK,
+        session=metadata_repo.session,
+    )
+    return result
+
+
 @router.get("/admin/users", response_model=List[MetadataUserResponse])
 async def list_metadata_users(
     skip: int = Query(0, ge=0),

app/auth/metadata_dependencies.py

@@ -52,6 +52,7 @@ async def get_current_metadata_user(
 ):
     keycloak_sub = _keycloak_sub_from_payload(keycloak_payload)
     username = get_keycloak_preferred_username(keycloak_payload)
+    email = _email_from_payload(keycloak_payload)
     try:
         user = await metadata_repo.get_user_by_keycloak_id(keycloak_sub)
     except SQLAlchemyError as exc:
@@ -65,13 +66,14 @@ async def get_current_metadata_user(
         ) from exc
     if not user or not user.is_active:
         raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user"
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="User authorization pending",
         )
     try:
         user = await metadata_repo.refresh_user_keycloak_snapshot(
             user,
             username=username,
-            email=_email_from_payload(keycloak_payload),
+            email=email,
         )
     except SQLAlchemyError as exc:
         logger.error(

app/core/config.py

@@ -63,6 +63,19 @@ class Settings(BaseSettings):
     KEYCLOAK_AUDIENCE: str = ""
     KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS: int = 900
 
+    # Browser cross-origin access; process environment overrides .env.
+    CORS_ALLOWED_ORIGINS: str = (
+        "http://127.0.0.1:5173,http://localhost:5173,"
+        "http://127.0.0.1:15173,http://localhost:15173"
+    )
+
+    @property
+    def cors_allowed_origins(self) -> list[str]:
+        origins = [value.strip() for value in self.CORS_ALLOWED_ORIGINS.split(",") if value.strip()]
+        if "*" in origins:
+            raise ValueError("CORS_ALLOWED_ORIGINS must list explicit origins")
+        return origins
+
     # Bocha Web Search API
     BOCHA_API_KEY: str = ""
     BOCHA_WEB_SEARCH_URL: str = "https://api.bochaai.com/v1/web-search"

app/domain/schemas/admin_metadata.py

@@ -50,6 +50,19 @@ class MetadataUserSyncResult(BaseModel):
     error: str | None = None
 
 
+class KeycloakUserSyncIssue(BaseModel):
+    username: str
+    reason: str
+
+
+class KeycloakUserSyncResponse(BaseModel):
+    scanned: int
+    created: int
+    updated: int
+    skipped: int
+    issues: list[KeycloakUserSyncIssue] = Field(default_factory=list)
+
+
 class ProjectMemberCreateRequest(BaseModel):
     user_id: UUID
     project_role: ProjectRole = "viewer"

app/infra/db/metadb/repositories/metadata_repository.py

@@ -5,6 +5,7 @@ from uuid import UUID, uuid4
 
 from cryptography.fernet import InvalidToken
 from sqlalchemy import delete, func, select
+from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.encryption import (
@@ -158,6 +159,44 @@ class MetadataRepository:
         await self.session.refresh(user)
         return user
 
+    async def sync_user_snapshot_from_keycloak(
+        self,
+        *,
+        keycloak_id: UUID,
+        username: str,
+        email: str,
+    ) -> tuple[models.User, bool]:
+        user = await self.get_user_by_keycloak_id(keycloak_id)
+        if user is not None:
+            user.username = username
+            user.email = email
+            user.updated_at = _utcnow()
+            await self.session.commit()
+            await self.session.refresh(user)
+            return user, False
+
+        statement = (
+            insert(models.User)
+            .values(
+                id=uuid4(),
+                keycloak_id=keycloak_id,
+                username=username,
+                email=email,
+                role="user",
+                is_active=False,
+                is_superuser=False,
+            )
+            .on_conflict_do_nothing(index_elements=[models.User.keycloak_id])
+            .returning(models.User.id)
+        )
+        result = await self.session.execute(statement)
+        inserted_id = result.scalar_one_or_none()
+        await self.session.commit()
+        user = await self.get_user_by_keycloak_id(keycloak_id)
+        if user is None:
+            raise RuntimeError("Keycloak user synchronization did not persist a user")
+        return user, inserted_id is not None
+
     async def refresh_user_keycloak_snapshot(
         self,
         user: models.User,

app/main.py

@@ -66,7 +66,7 @@ app.add_middleware(AuditMiddleware)
 # 确保这是你最后一个添加的 app.add_middleware
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],  # 允许所有来源
+    allow_origins=settings.cors_allowed_origins,
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],

app/native/wndb/inp/exporter.py

@@ -120,7 +120,7 @@ def dump_inp(project: str, inp: str, version: str = '3'):
             file.write('\n'.join(inp_out_demand(project)))
 
         elif name == STATUS:
-            file.write('\n'.join(inp_out_status(project)))
+            file.write('\n'.join(inp_out_status(project, version=version)))
 
         elif name == PATTERNS:
             if version == '3':
@@ -232,7 +232,7 @@ def export_inp(project: str, version: str = '3') -> ChangeSet:
             inp += '\n'.join(inp_out_demand(project))
 
         elif name == STATUS:
-            inp += '\n'.join(inp_out_status(project))
+            inp += '\n'.join(inp_out_status(project, version=version))
 
         elif name == PATTERNS:
             if version == '3':

app/native/wndb/model/status.py

@@ -93,14 +93,16 @@ def inp_in_status(line: str) -> str:
         return str(f"insert into network.link_initial_settings (link_id, status, setting) values ({sql_literal(link)}, null, {sql_literal(float(value))});")
 
 
-def inp_out_status(name: str) -> list[str]:
+def inp_out_status(name: str, version: str = '2') -> list[str]:
     lines = []
     objs = read_all(name, 'select link_id as link, status, setting from network.link_initial_settings order by link_id')
     for obj in objs:
         link = obj['link']
         status = obj['status'] if obj['status'] != None else ''
         setting = obj['setting'] if obj['setting'] != None else ''
-        if status != '':
+        # EPANET 2 valves use their configured setting by default; ACTIVE is
+        # an internal state, not a valid keyword in its [STATUS] section.
+        if status != '' and not (version == '2' and status.upper() == LINK_STATUS_ACTIVE):
             lines.append(f'{link} {status}')
         if setting != '':
             lines.append(f'{link} {setting}')

app/services/keycloak_user_sync.py

@@ -0,0 +1,117 @@
+from dataclasses import dataclass
+from urllib.parse import quote
+from uuid import UUID
+
+import httpx
+
+
+class KeycloakUserSyncConfigurationError(ValueError):
+    pass
+
+
+class KeycloakUserSyncRequestError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class KeycloakUserSnapshot:
+    keycloak_id: UUID
+    username: str
+    email: str | None
+    service_account_client_id: str | None = None
+
+
+def _keycloak_admin_users_url(issuer: str) -> str:
+    issuer = issuer.rstrip("/")
+    marker = "/realms/"
+    if not issuer or marker not in issuer:
+        raise KeycloakUserSyncConfigurationError(
+            "Keycloak token issuer is not a complete realm URL"
+        )
+    server_url, realm = issuer.rsplit(marker, 1)
+    if not server_url or not realm:
+        raise KeycloakUserSyncConfigurationError(
+            "Keycloak token issuer is not a complete realm URL"
+        )
+    return f"{server_url}/admin/realms/{quote(realm, safe='')}/users"
+
+
+async def fetch_keycloak_users(
+    *,
+    issuer: str,
+    bearer_token: str,
+    client: httpx.AsyncClient | None = None,
+) -> list[KeycloakUserSnapshot]:
+    admin_users_url = _keycloak_admin_users_url(issuer)
+    if not bearer_token:
+        raise KeycloakUserSyncConfigurationError(
+            "Keycloak administrator access token is missing"
+        )
+    owns_client = client is None
+    if client is None:
+        client = httpx.AsyncClient(timeout=15.0)
+
+    try:
+        try:
+            snapshots: list[KeycloakUserSnapshot] = []
+            first = 0
+            page_size = 100
+            while True:
+                response = await client.get(
+                    admin_users_url,
+                    params={"first": first, "max": page_size},
+                    headers={
+                        "Accept": "application/json",
+                        "Authorization": f"Bearer {bearer_token}",
+                    },
+                )
+                response.raise_for_status()
+                payload = response.json()
+                if not isinstance(payload, list):
+                    raise KeycloakUserSyncRequestError(
+                        "Keycloak users response is not a list"
+                    )
+                for item in payload:
+                    if not isinstance(item, dict):
+                        continue
+                    try:
+                        keycloak_id = UUID(str(item.get("id", "")))
+                    except ValueError:
+                        continue
+                    username = str(item.get("username") or "").strip()
+                    if not username:
+                        continue
+                    email_value = item.get("email")
+                    email = str(email_value).strip() if email_value else None
+                    service_account_value = item.get("serviceAccountClientId")
+                    snapshots.append(
+                        KeycloakUserSnapshot(
+                            keycloak_id=keycloak_id,
+                            username=username,
+                            email=email,
+                            service_account_client_id=(
+                                str(service_account_value)
+                                if service_account_value
+                                else None
+                            ),
+                        )
+                    )
+                if len(payload) < page_size:
+                    break
+                first += page_size
+            return snapshots
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == 403:
+                raise KeycloakUserSyncRequestError(
+                    "Keycloak administrator requires query-users and view-users roles"
+                ) from exc
+            raise KeycloakUserSyncRequestError(
+                "Keycloak user directory request failed"
+            ) from exc
+        except httpx.HTTPError as exc:
+            raise KeycloakUserSyncRequestError(
+                "Keycloak user directory request failed"
+            ) from exc
+    finally:
+        if owns_client:
+            await client.aclose()

app/services/simulation.py

@@ -490,6 +490,23 @@ def run_simulation(
                     set_status(name_c, cs)
         # 运行并返回结果
         result_data = json.loads(run_project(name_c))
+        if result_data.get("simulation_result") == "failed":
+            report = result_data.get("report") or ""
+            logging.error("EPANET failed for project %s:\n%s", name_c, report)
+            lines = report.splitlines()
+            error_index = next(
+                (i for i, line in enumerate(lines) if line.strip().startswith("Error ")),
+                None,
+            )
+            detail = (
+                " ".join(line.strip() for line in lines[error_index:error_index + 2])[:600]
+                if error_index is not None
+                else "引擎未返回错误详情，请查看服务日志"
+            )
+            raise RuntimeError(f"EPANET 计算失败：{detail}")
+        output_data = result_data.get("output")
+        if not isinstance(output_data, dict):
+            raise RuntimeError("run_project did not return JSON output content")
     refresh_materialized_views_after_commit(name_c)
     time_cost_end = time.perf_counter()
     print(
@@ -498,9 +515,6 @@ def run_simulation(
             time_cost_end - time_cost_start,
         )
     )
-    output_data = result_data.get("output")
-    if not isinstance(output_data, dict):
-        raise RuntimeError("run_project did not return JSON output content")
     node_result = output_data.get("node_results")
     link_result = output_data.get("link_results")
     if node_result is None or link_result is None:

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "5b8ec0b469f1194e72214003a9cf427489f48de19ae31ee6492821cf64916b6d"
+      "sha256": "19961572fad04f8545a4f607d4bf7276de64a84b266daecbb21d74e0975f97ef"
     }
   }
 }

tests/api/test_admin_metadata_endpoints.py

@@ -20,6 +20,7 @@ from app.domain.schemas.admin_metadata import (
     ProjectMemberUpdateRequest,
 )
 from app.infra.db.metadb.repositories.metadata_repository import ProjectDbRouting
+from app.services.keycloak_user_sync import KeycloakUserSnapshot
 
 
 @pytest.fixture
@@ -176,6 +177,66 @@ async def test_batch_sync_metadata_users_returns_per_user_results(monkeypatch):
     assert admin_metadata.log_audit_event.await_count == 2
 
 
+@pytest.mark.anyio
+async def test_sync_keycloak_user_directory_preserves_authorization(monkeypatch):
+    new_id = uuid4()
+    existing_id = uuid4()
+    repo = SimpleNamespace(
+        session=SimpleNamespace(rollback=AsyncMock()),
+        sync_user_snapshot_from_keycloak=AsyncMock(
+            side_effect=[
+                (_user(keycloak_id=new_id, is_active=False), True),
+                (_user(keycloak_id=existing_id, role="admin"), False),
+            ]
+        ),
+    )
+    monkeypatch.setattr(
+        admin_metadata,
+        "fetch_keycloak_users",
+        AsyncMock(
+            return_value=[
+                KeycloakUserSnapshot(new_id, "new-user", "new@example.com"),
+                KeycloakUserSnapshot(existing_id, "admin-user", "admin@example.com"),
+                KeycloakUserSnapshot(uuid4(), "no-email", None),
+                KeycloakUserSnapshot(
+                    uuid4(),
+                    "service-account-client",
+                    "service@example.com",
+                    service_account_client_id="client",
+                ),
+            ]
+        ),
+    )
+    monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
+
+    response = await admin_metadata.sync_keycloak_user_directory(
+        current_user=_user(role="admin", is_superuser=True),
+        keycloak_payload={"iss": "https://keycloak.example.com/realms/tjwater"},
+        keycloak_access_token="admin-token",
+        metadata_repo=repo,
+    )
+
+    assert response.scanned == 4
+    assert response.created == 1
+    assert response.updated == 1
+    assert response.skipped == 2
+    assert [issue.reason for issue in response.issues] == [
+        "missing_email",
+        "service_account",
+    ]
+    assert repo.sync_user_snapshot_from_keycloak.await_count == 2
+    admin_metadata.fetch_keycloak_users.assert_awaited_once_with(
+        issuer="https://keycloak.example.com/realms/tjwater",
+        bearer_token="admin-token",
+    )
+    assert repo.sync_user_snapshot_from_keycloak.await_args_list[1].kwargs == {
+        "keycloak_id": existing_id,
+        "username": "admin-user",
+        "email": "admin@example.com",
+    }
+    admin_metadata.log_audit_event.assert_awaited_once()
+
+
 @pytest.mark.anyio
 async def test_update_metadata_user_updates_role_and_active_status(monkeypatch):
     user_id = uuid4()

tests/unit/test_analysis_simulation.py

@@ -277,3 +277,51 @@ def test_extended_simulation_marks_run_failed_when_result_storage_fails(monkeypa
         )
 
     assert [call[1]["status"] for call in lifecycle_calls] == ["failed"]
+
+
+@pytest.mark.parametrize("has_output", [False, True])
+def test_engine_failure_preserves_report_and_rolls_back_model_changes(monkeypatch, has_output):
+    from app.services import simulation
+
+    transaction_outcome = []
+
+    @contextmanager
+    def project_transaction(_name):
+        try:
+            yield
+        except Exception:
+            transaction_outcome.append("rollback")
+            raise
+        else:
+            transaction_outcome.append("commit")
+
+    monkeypatch.setattr(simulation, "project_transaction", project_transaction)
+    monkeypatch.setattr(simulation, "get_time", lambda _: {
+        "HYDRAULIC TIMESTEP": "0:15:00",
+        "DURATION": "0:00",
+        "PATTERN START": "0:00",
+    })
+    monkeypatch.setattr(simulation, "set_time", lambda *_: None)
+    result = {
+        "simulation_result": "failed",
+        "report": "EPANET report\n  Error 202: illegal numeric value ACTIVE in [STATUS] section:\n  V1 ACTIVE\n",
+    }
+    if has_output:
+        result["output"] = {"node_results": [], "link_results": [], "times": {"num_periods": 1}}
+    monkeypatch.setattr(simulation, "run_project", lambda _: json.dumps(result))
+    refresh = Mock()
+    store = Mock()
+    monkeypatch.setattr(simulation, "refresh_materialized_views_after_commit", refresh)
+    monkeypatch.setattr(simulation.TimescaleInternalStorage, "store_realtime_simulation", store)
+
+    with pytest.raises(RuntimeError, match="EPANET.*Error 202"):
+        simulation.run_simulation(
+            name="demo",
+            simulation_type="realtime",
+            modify_pattern_start_time="2026-07-30T04:30:00Z",
+            scada_mappings=_empty_scada_mappings(simulation),
+        )
+
+    assert transaction_outcome == ["rollback"]
+    refresh.assert_not_called()
+    store.assert_not_called()

tests/unit/test_cors.py

@@ -0,0 +1,73 @@
+import pytest
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.testclient import TestClient
+
+from app.core.config import Settings
+
+
+ALLOWED_ORIGINS = [
+    "https://demo.waternetwork.cn",
+    "http://192.168.1.133",
+    "https://192.168.1.133",
+    "http://192.168.1.110",
+    "https://192.168.1.110",
+]
+
+
+@pytest.mark.parametrize("origins", ["*", " *, http://localhost:5173", "http://localhost:5173,*"])
+def test_wildcard_origins_are_rejected(origins: str):
+    with pytest.raises(ValueError, match="must list explicit origins"):
+        Settings(_env_file=None, CORS_ALLOWED_ORIGINS=origins).cors_allowed_origins
+
+
+@pytest.mark.parametrize(
+    "origin",
+    ALLOWED_ORIGINS + [
+        "http://192.168.1.114",
+        "http://192.168.1.133:5173",
+        "https://next-demo.waternetwork.cn",
+        "http://192.168.1.133.evil.example",
+        "null",
+    ],
+)
+def test_explicit_origin_preflight(origin: str):
+    settings = Settings(_env_file=None, CORS_ALLOWED_ORIGINS=", ".join(ALLOWED_ORIGINS))
+    app = FastAPI()
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.cors_allowed_origins,
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+    with TestClient(app) as client:
+        response = client.options(
+            "/api/v1/projects",
+            headers={
+                "Origin": origin,
+                "Access-Control-Request-Method": "GET",
+                "Access-Control-Request-Headers": "authorization,x-project-id",
+            },
+        )
+    allowed = origin in ALLOWED_ORIGINS
+    assert response.status_code == (200 if allowed else 400)
+    assert response.headers.get("access-control-allow-origin") == (origin if allowed else None)
+    if allowed:
+        assert response.headers["access-control-allow-credentials"] == "true"
+        assert "authorization" in response.headers["access-control-allow-headers"]
+
+
+def test_default_origins_are_explicit_local_sources(monkeypatch):
+    monkeypatch.delenv("CORS_ALLOWED_ORIGINS", raising=False)
+    settings = Settings(_env_file=None)
+    assert settings.cors_allowed_origins == [
+        "http://127.0.0.1:5173", "http://localhost:5173",
+        "http://127.0.0.1:15173", "http://localhost:15173",
+    ]
+
+
+def test_environment_controls_origins(monkeypatch):
+    monkeypatch.setenv("CORS_ALLOWED_ORIGINS", "http://192.168.1.133,https://192.168.1.110")
+    settings = Settings(_env_file=None)
+    assert settings.cors_allowed_origins == ["http://192.168.1.133", "https://192.168.1.110"]

tests/unit/test_epanet_status_export.py

@@ -0,0 +1,70 @@
+import ctypes
+import platform
+from pathlib import Path
+
+import pytest
+
+from app.native.wndb.model import status
+
+
+def test_epanet2_preserves_valve_setting_without_emitting_active(monkeypatch):
+    monkeypatch.setattr(status, "read_all", lambda *_: [
+        {"link": "V1", "status": "ACTIVE", "setting": None},
+        {"link": "V2", "status": "ACTIVE", "setting": 2.5},
+        {"link": "V3", "status": "CLOSED", "setting": None},
+        {"link": "P1", "status": "OPEN", "setting": 0},
+    ])
+    assert status.inp_out_status("model") == ["V2 2.5", "V3 CLOSED", "P1 OPEN", "P1 0"]
+
+
+def test_epanet3_retains_explicit_active_status(monkeypatch):
+    monkeypatch.setattr(status, "read_all", lambda *_: [
+        {"link": "V1", "status": "ACTIVE", "setting": 100},
+    ])
+    assert status.inp_out_status("model", version="3") == ["V1 ACTIVE", "V1 100"]
+
+
+@pytest.mark.parametrize("version", ["2", "3"])
+def test_file_and_text_exporters_forward_the_selected_format(monkeypatch, tmp_path, version):
+    from app.native.wndb.inp import exporter
+
+    monkeypatch.setattr(exporter, "have_project", lambda _: True)
+    monkeypatch.setattr(exporter, "section_name", [exporter.STATUS])
+    monkeypatch.setattr(exporter, "section_names_for_epanetv2", [exporter.STATUS])
+    monkeypatch.setattr(status, "read_all", lambda *_: [
+        {"link": "V1", "status": "ACTIVE", "setting": 100},
+    ])
+    path = tmp_path / "model.inp"
+    exporter.dump_inp("model", str(path), version)
+    text = exporter.export_inp("model", version).operations[0]["inp"]
+    assert "V1 100" in path.read_text() and "V1 100" in text
+    assert ("V1 ACTIVE" in path.read_text()) == (version == "3")
+    assert ("V1 ACTIVE" in text) == (version == "3")
+
+
+def test_exported_active_valve_passes_bundled_epanet2_input_validation(monkeypatch, tmp_path):
+    library = Path(__file__).resolve().parents[2] / "app/infra/epanet/linux/libepanet2.so"
+    if platform.system() != "Linux" or not library.exists():
+        pytest.skip("Bundled Linux EPANET library is unavailable")
+    monkeypatch.setattr(status, "read_all", lambda *_: [
+        {"link": "V1", "status": "ACTIVE", "setting": None},
+    ])
+    inp = tmp_path / "valve.inp"
+    inp.write_text(
+        "[JUNCTIONS]\nJ1 0 1\nJ2 0 1\n[RESERVOIRS]\nR1 100\n"
+        "[PIPES]\nP1 R1 J1 100 300 100 0 OPEN\n"
+        "[VALVES]\nV1 J1 J2 600 TCV 100 0\n[STATUS]\n"
+        + "\n".join(status.inp_out_status("model"))
+        + "\n[OPTIONS]\nUNITS LPS\n[END]\n"
+    )
+    lib = ctypes.CDLL(str(library))
+    lib.EN_createproject.argtypes = [ctypes.POINTER(ctypes.c_void_p)]
+    lib.EN_open.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
+    lib.EN_deleteproject.argtypes = [ctypes.c_void_p]
+    handle = ctypes.c_void_p()
+    assert lib.EN_createproject(ctypes.byref(handle)) == 0
+    try:
+        code = lib.EN_open(handle, str(inp).encode(), str(tmp_path / "valve.rpt").encode(), b"")
+        assert code == 0
+    finally:
+        lib.EN_deleteproject(handle)

tests/unit/test_keycloak_user_sync.py

@@ -0,0 +1,54 @@
+from uuid import uuid4
+
+import httpx
+import pytest
+
+from app.services import keycloak_user_sync
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+async def test_fetch_keycloak_users_uses_admin_token_and_paginates():
+    user_ids = [uuid4() for _ in range(101)]
+    requested_first: list[int] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.headers["authorization"] == "Bearer admin-token"
+        first = int(request.url.params["first"])
+        requested_first.append(first)
+        page_ids = user_ids[first : first + 100]
+        return httpx.Response(
+            200,
+            json=[
+                {
+                    "id": str(user_id),
+                    "username": f"user-{first + index}",
+                    "email": f"user-{first + index}@example.com",
+                }
+                for index, user_id in enumerate(page_ids)
+            ],
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        users = await keycloak_user_sync.fetch_keycloak_users(
+            issuer="https://keycloak.example.com/realms/tjwater",
+            bearer_token="admin-token",
+            client=client,
+        )
+
+    assert len(users) == 101
+    assert requested_first == [0, 100]
+    assert users[-1].username == "user-100"
+
+
+@pytest.mark.anyio
+async def test_fetch_keycloak_users_requires_realm_issuer():
+    with pytest.raises(keycloak_user_sync.KeycloakUserSyncConfigurationError):
+        await keycloak_user_sync.fetch_keycloak_users(
+            issuer="",
+            bearer_token="admin-token",
+        )

tests/unit/test_metadata_repository_users.py

@@ -0,0 +1,44 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+
+from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+async def test_keycloak_snapshot_sync_preserves_existing_authorization():
+    keycloak_id = uuid4()
+    user = SimpleNamespace(
+        keycloak_id=keycloak_id,
+        username="old-name",
+        email="old@example.com",
+        role="admin",
+        is_active=True,
+        is_superuser=True,
+        updated_at=None,
+    )
+    session = SimpleNamespace(commit=AsyncMock(), refresh=AsyncMock())
+    repository = MetadataRepository(session)
+    repository.get_user_by_keycloak_id = AsyncMock(return_value=user)
+
+    result, created = await repository.sync_user_snapshot_from_keycloak(
+        keycloak_id=keycloak_id,
+        username="new-name",
+        email="new@example.com",
+    )
+
+    assert created is False
+    assert result.username == "new-name"
+    assert result.email == "new@example.com"
+    assert result.role == "admin"
+    assert result.is_active is True
+    assert result.is_superuser is True
+    session.commit.assert_awaited_once()
+    session.refresh.assert_awaited_once_with(user)